Allow creating clones without source PVC (#2306)

* Modify datavolume admission webhook to enable creating clones without source PVC

This commit modifies the datavolume admission webhook to follow a more descriptive approach, enabling the creation of clones without a source PVC.

This clone will later be handled by the datavolume-controller until the source PVC is created.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Modify the datavolume-controller to improve the handling of clones without source

Since we are allowing the creation of clones without source PVC in the admission webhook, we need to improve the handling of these clones once in the datavolume-controller.

This commit modifies said controller, so we do proper error handling and validation until the source PVC is created.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Add unit tests to check the creation of clones without source PVC

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Include a mechanism to reconcile clones without source once the source PVC is created

This commit introduces a new datavolume-controller watch so, if a clone without source is created, we make sure to reconcile it once a proper PVC is created.

It also updates/includes unit tests for proper coverage.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Include functional tests to cover the creation of clones without source PVC

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Minor refactoring of clone-related code in DataVolume reconciler to improve readability

After enabling the creation of clones without source PVC in the datavolume controller, the clone-related logic outside its reconciler has increased in size and become sparse.

This commit rearranges all this code and condenses it into the clone reconciler, without changing the loop behavior.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Modify datavolume-mutate webhook to reject the creation of clones if the source PVC's namespace doesn't exist

In previous commits, a mechanism to allow the creation of clones without source PVC was added, without ever checking if the source PVC's namespace exists or not.

This behavior could lead to permission conflicts between the user and the source's namespace since the webhook skipped all the related validation.

This commit modifies the datavolume-mutate admission webhook to reject the clone if the source PVC's namespace doesn't exist.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

* Update unit tests for proper coverage of clone-validation functions

This commit adds and updates several unit tests to improve the coverage of the clone-validation mechanism after several functions were moved to the controller.

It also introduces minor changes on related code in the datavolume-controller and functional tests following PR review.

Signed-off-by: Alvaro Romero <alromero@redhat.com>

pkg/apiserver/webhooks/datavolume-mutate.go

@@ -107,6 +107,11 @@ func (wh *dataVolumeMutatingWebhook) Admit(ar admissionv1.AdmissionReview) *admi
 		sourceNamespace = targetNamespace
 	}
 
+	_, err := wh.k8sClient.CoreV1().Namespaces().Get(context.TODO(), sourceNamespace, metav1.GetOptions{})
+	if err != nil {
+		return toAdmissionResponseError(err)
+	}
+
 	if ar.Request.Operation == admissionv1.Update {
 		if err := json.Unmarshal(ar.Request.OldObject.Raw, &oldDataVolume); err != nil {
 			return toAdmissionResponseError(err)

pkg/apiserver/webhooks/datavolume-mutate_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/appscode/jsonpatch"
 	admissionv1 "k8s.io/api/admission/v1"
 	authorization "k8s.io/api/authorization/v1"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	fakeclient "k8s.io/client-go/kubernetes/fake"
@@ -196,6 +197,28 @@ var _ = Describe("Mutating DataVolume Webhook", func() {
 			Expect(resp.Patch).To(BeNil())
 		})
 
+		It("should reject a clone if the source PVC's namespace doesn't exist", func() {
+			dataVolume := newPVCDataVolume("testDV", "noNamespace", "test")
+			dvBytes, _ := json.Marshal(&dataVolume)
+
+			ar := &admissionv1.AdmissionReview{
+				Request: &admissionv1.AdmissionRequest{
+					Resource: metav1.GroupVersionResource{
+						Group:    cdicorev1.SchemeGroupVersion.Group,
+						Version:  cdicorev1.SchemeGroupVersion.Version,
+						Resource: "datavolumes",
+					},
+					Object: runtime.RawExtension{
+						Raw: dvBytes,
+					},
+				},
+			}
+
+			resp := mutateDVs(key, ar, false)
+			Expect(resp.Allowed).To(BeFalse())
+			Expect(resp.Patch).To(BeNil())
+		})
+
 		DescribeTable("should", func(srcNamespace string) {
 			dataVolume := newPVCDataVolume("testDV", srcNamespace, "test")
 			dvBytes, _ := json.Marshal(&dataVolume)
@@ -237,7 +260,9 @@ func mutateDVs(key *rsa.PrivateKey, ar *admissionv1.AdmissionReview, isAuthorize
 }
 
 func mutateDVsEx(key *rsa.PrivateKey, ar *admissionv1.AdmissionReview, isAuthorized bool, cdiObjects []runtime.Object) *admissionv1.AdmissionResponse {
-	client := fakeclient.NewSimpleClientset()
+	defaultNs := corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "default"}}
+	testNs := corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: "testNamespace"}}
+	client := fakeclient.NewSimpleClientset(&defaultNs, &testNs)
 	client.PrependReactor("create", "subjectaccessreviews", func(action k8stesting.Action) (bool, runtime.Object, error) {
 		if action.GetResource().Resource != "subjectaccessreviews" {
 			return false, nil, nil

pkg/apiserver/webhooks/datavolume-validate.go

@@ -72,15 +72,6 @@ func validateNameLength(name string, maxLen int) []metav1.StatusCause {
 	return causes
 }
 
-func validateContentTypes(sourcePVC *v1.PersistentVolumeClaim, spec *cdiv1.DataVolumeSpec) (bool, cdiv1.DataVolumeContentType, cdiv1.DataVolumeContentType) {
-	sourceContentType := cdiv1.DataVolumeContentType(controller.GetContentType(sourcePVC))
-	targetContentType := spec.ContentType
-	if targetContentType == "" {
-		targetContentType = cdiv1.DataVolumeKubeVirt
-	}
-	return sourceContentType == targetContentType, sourceContentType, targetContentType
-}
-
 func (wh *dataVolumeValidatingWebhook) validateDataVolumeSpec(request *admissionv1.AdmissionRequest, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec, namespace *string) []metav1.StatusCause {
 	var causes []metav1.StatusCause
 	var url string
@@ -412,12 +403,10 @@ func (wh *dataVolumeValidatingWebhook) validateSourceRef(request *admissionv1.Ad
 func (wh *dataVolumeValidatingWebhook) validateDataVolumeSourcePVC(PVC *cdiv1.DataVolumeSourcePVC, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec) *metav1.StatusCause {
 	sourcePVC, err := wh.k8sClient.CoreV1().PersistentVolumeClaims(PVC.Namespace).Get(context.TODO(), PVC.Name, metav1.GetOptions{})
 	if err != nil {
+		// We allow the creation of a clone DV even if the source PVC doesn't exist.
+		// The validation will be completed once the source PVC is created.
 		if k8serrors.IsNotFound(err) {
-			return &metav1.StatusCause{
-				Type:    metav1.CauseTypeFieldValueNotFound,
-				Message: fmt.Sprintf("Source PVC %s/%s not found", PVC.Namespace, PVC.Name),
-				Field:   field.String(),
-			}
+			return nil
 		}
 		return &metav1.StatusCause{
 			Message: err.Error(),

pkg/apiserver/webhooks/datavolume-validate_test.go

@@ -182,10 +182,10 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(Equal(true))
 		})
 
-		It("should reject DataVolume with PVC source on create if PVC does not exist", func() {
+		It("should accept DataVolume with PVC source on create if PVC does not exist", func() {
 			dataVolume := newPVCDataVolume("testDV", "testNamespace", "test")
 			resp := validateDataVolumeCreate(dataVolume)
-			Expect(resp.Allowed).To(Equal(false))
+			Expect(resp.Allowed).To(Equal(true))
 		})
 
 		It("should reject invalid DataVolume source PVC namespace on create", func() {
@@ -485,7 +485,7 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(Equal(false))
 		})
 
-		It("should reject DataVolume with SourceRef on create if DataSource exists but PVC does not exist", func() {
+		It("should accept DataVolume with SourceRef on create if DataSource exists but PVC does not exist", func() {
 			dataVolume := newDataSourceDataVolume("testDV", &testNamespace, "test")
 			dataSource := &cdiv1.DataSource{
 				ObjectMeta: metav1.ObjectMeta{
@@ -502,7 +502,7 @@ var _ = Describe("Validating Webhook", func() {
 				},
 			}
 			resp := validateDataVolumeCreateEx(dataVolume, nil, []runtime.Object{dataSource})
-			Expect(resp.Allowed).To(Equal(false))
+			Expect(resp.Allowed).To(Equal(true))
 		})
 
 		It("should reject DataVolume with empty SourceRef name on create", func() {

pkg/controller/datavolume-controller.go

@@ -47,8 +47,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
+	"sigs.k8s.io/controller-runtime/pkg/event"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
@@ -166,6 +168,14 @@ const (
 	NamespaceTransferInProgress = "NamespaceTransferInProgress"
 	// MessageNamespaceTransferInProgress is a const for reporting target transfer
 	MessageNamespaceTransferInProgress = "Transferring PersistentVolumeClaim for DataVolume %s/%s"
+	// CloneValidationFailed reports that a clone wasn't admitted by our validation mechanism (reason)
+	CloneValidationFailed = "CloneValidationFailed"
+	// MessageCloneValidationFailed reports that a clone wasn't admitted by our validation mechanism (message)
+	MessageCloneValidationFailed = "The clone doesn't meet the validation requirements"
+	// CloneWithoutSource reports that the source PVC of a clone doesn't exists (reason)
+	CloneWithoutSource = "CloneWithoutSource"
+	// MessageCloneWithoutSource reports that the source PVC of a clone doesn't exists (message)
+	MessageCloneWithoutSource = "The source PVC doesn't exist"
 
 	// AnnCSICloneRequest annotation associates object with CSI Clone Request
 	AnnCSICloneRequest = "cdi.kubevirt.io/CSICloneRequest"
@@ -404,6 +414,58 @@ func addDatavolumeControllerWatches(mgr manager.Manager, datavolumeController co
 		return err
 	}
 
+	// Watch to reconcile clones created without source
+	if err := addCloneWithoutSourceWatch(mgr, datavolumeController); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// addCloneWithoutSourceWatch reconciles clones created without source once the matching PVC is created
+func addCloneWithoutSourceWatch(mgr manager.Manager, datavolumeController controller.Controller) error {
+	const sourcePvcField = "spec.source.pvc"
+
+	getKey := func(namespace, name string) string {
+		return namespace + "/" + name
+	}
+
+	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataVolume{}, sourcePvcField, func(obj client.Object) []string {
+		if obj.(*cdiv1.DataVolume).Spec.Source != nil {
+			if pvc := obj.(*cdiv1.DataVolume).Spec.Source.PVC; pvc != nil {
+				ns := getNamespace(pvc.Namespace, obj.GetNamespace())
+				return []string{getKey(ns, pvc.Name)}
+			}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	// Function to reconcile DVs that match the selected fields
+	dataVolumeMapper := func(obj client.Object) (reqs []reconcile.Request) {
+		dvList := &cdiv1.DataVolumeList{}
+		namespacedName := types.NamespacedName{Namespace: obj.GetNamespace(), Name: obj.GetName()}
+		matchingFields := client.MatchingFields{sourcePvcField: namespacedName.String()}
+		if err := mgr.GetClient().List(context.TODO(), dvList, matchingFields); err != nil {
+			return
+		}
+		for _, dv := range dvList.Items {
+			reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: dv.Namespace, Name: dv.Name}})
+		}
+		return
+	}
+
+	if err := datavolumeController.Watch(&source.Kind{Type: &corev1.PersistentVolumeClaim{}},
+		handler.EnqueueRequestsFromMapFunc(dataVolumeMapper),
+		predicate.Funcs{
+			CreateFunc: func(e event.CreateEvent) bool { return true },
+			DeleteFunc: func(e event.DeleteEvent) bool { return false },
+			UpdateFunc: func(e event.UpdateEvent) bool { return false },
+		}); err != nil {
+		return err
+	}
+
 	return nil
 }
 
@@ -473,26 +535,18 @@ func (r *DatavolumeReconciler) Reconcile(_ context.Context, req reconcile.Reques
 		return reconcile.Result{}, err
 	}
 
-	selectedCloneStrategy, err := r.selectCloneStrategy(datavolume, pvcSpec)
-	if err != nil {
-		return reconcile.Result{}, err
-	}
-	if selectedCloneStrategy == SmartClone {
-		r.sccs.StartController()
-	}
-
 	_, dvPrePopulated := datavolume.Annotations[AnnPrePopulated]
 
-	if selectedCloneStrategy != NoClone {
-		return r.reconcileClone(log, datavolume, pvc, pvcSpec, transferName, dvPrePopulated, pvcPopulated, selectedCloneStrategy)
+	if isClone := datavolume.Spec.Source.PVC != nil; isClone {
+		return r.reconcileClone(log, datavolume, pvc, pvcSpec, transferName, dvPrePopulated, pvcPopulated)
 	}
 
 	if !dvPrePopulated {
 		if pvc == nil {
 			newPvc, err := r.createPvcForDatavolume(log, datavolume, pvcSpec)
 			if err != nil {
 				if errQuotaExceeded(err) {
-					r.updateDataVolumeStatusPhaseWithEvent(cdiv1.Pending, datavolume, nil, selectedCloneStrategy,
+					r.updateDataVolumeStatusPhaseWithEvent(cdiv1.Pending, datavolume, nil, NoClone,
 						DataVolumeEvent{
 							eventType: corev1.EventTypeWarning,
 							reason:    ErrExceededQuota,
@@ -525,7 +579,7 @@ func (r *DatavolumeReconciler) Reconcile(_ context.Context, req reconcile.Reques
 
 	// Finally, we update the status block of the DataVolume resource to reflect the
 	// current state of the world
-	return r.reconcileDataVolumeStatus(datavolume, pvc, selectedCloneStrategy)
+	return r.reconcileDataVolumeStatus(datavolume, pvc, NoClone)
 }
 
 func (r *DatavolumeReconciler) reconcileClone(log logr.Logger,
@@ -534,8 +588,24 @@ func (r *DatavolumeReconciler) reconcileClone(log logr.Logger,
 	pvcSpec *corev1.PersistentVolumeClaimSpec,
 	transferName string,
 	prePopulated bool,
-	pvcPopulated bool,
-	selectedCloneStrategy cloneStrategy) (reconcile.Result, error) {
+	pvcPopulated bool) (reconcile.Result, error) {
+
+	// Check if source PVC exists and do proper validation before attempting to clone
+	if done, err := r.validateCloneAndSourcePVC(datavolume); err != nil {
+		return reconcile.Result{}, err
+	} else if !done {
+		return reconcile.Result{}, nil
+	}
+
+	// Get the most appropiate clone strategy
+	selectedCloneStrategy, err := r.selectCloneStrategy(datavolume, pvcSpec)
+	if err != nil {
+		return reconcile.Result{}, err
+	}
+
+	if selectedCloneStrategy == SmartClone {
+		r.sccs.StartController()
+	}
 
 	if !prePopulated && !pvcPopulated {
 		if pvc == nil {
@@ -646,10 +716,6 @@ func (r *DatavolumeReconciler) ensureExtendedToken(pvc *corev1.PersistentVolumeC
 }
 
 func (r *DatavolumeReconciler) selectCloneStrategy(datavolume *cdiv1.DataVolume, pvcSpec *corev1.PersistentVolumeClaimSpec) (cloneStrategy, error) {
-	if datavolume.Spec.Source.PVC == nil {
-		return NoClone, nil
-	}
-
 	preferredCloneStrategy, err := r.getCloneStrategy(datavolume)
 	if err != nil {
 		return NoClone, err
@@ -2616,3 +2682,29 @@ func GetRequiredSpace(filesystemOverhead float64, requestedSpace int64) int64 {
 func newLongTermCloneTokenGenerator(key *rsa.PrivateKey) token.Generator {
 	return token.NewGenerator(common.ExtendedCloneTokenIssuer, key, 10*365*24*time.Hour)
 }
+
+// validateCloneAndSourcePVC checks if the source PVC of a clone exists and does proper validation
+func (r *DatavolumeReconciler) validateCloneAndSourcePVC(datavolume *cdiv1.DataVolume) (bool, error) {
+	sourcePvc, err := r.findSourcePvc(datavolume)
+	if err != nil {
+		// Clone without source
+		if k8serrors.IsNotFound(err) {
+			r.updateDataVolumeStatusPhaseWithEvent(datavolume.Status.Phase, datavolume, nil, NoClone,
+				DataVolumeEvent{
+					eventType: corev1.EventTypeWarning,
+					reason:    CloneWithoutSource,
+					message:   MessageCloneWithoutSource,
+				})
+			return false, nil
+		}
+		return false, err
+	}
+
+	err = ValidateClone(sourcePvc, &datavolume.Spec)
+	if err != nil {
+		r.recorder.Event(datavolume, corev1.EventTypeWarning, CloneValidationFailed, MessageCloneValidationFailed)
+		return false, err
+	}
+
+	return true, nil
+}

pkg/controller/datavolume-controller_test.go

@@ -1501,6 +1501,78 @@ var _ = Describe("All DataVolume Tests", func() {
 
 	})
 
+	var _ = Describe("Clone without source", func() {
+		scName := "testsc"
+		sc := createStorageClassWithProvisioner(scName, map[string]string{
+			AnnDefaultStorageClass: "true",
+		}, "csi-plugin")
+
+		It("Validate clone without source as feasible, but not done", func() {
+			dv := newCloneDataVolume("test-dv")
+			storageProfile := createStorageProfile(scName, nil, filesystemMode)
+			reconciler = createDatavolumeReconciler(dv, storageProfile, sc)
+
+			done, err := reconciler.validateCloneAndSourcePVC(dv)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(done).To(BeFalse())
+		})
+
+		It("Validate that clone without source completes after PVC is created", func() {
+			dv := newCloneDataVolume("test-dv")
+			storageProfile := createStorageProfile(scName, nil, filesystemMode)
+			reconciler = createDatavolumeReconciler(dv, storageProfile, sc)
+
+			done, err := reconciler.validateCloneAndSourcePVC(dv)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(done).To(BeFalse())
+
+			// We create the source PVC after creating the clone
+			pvc := createPvcInStorageClass("test", metav1.NamespaceDefault, &scName, nil, nil, corev1.ClaimBound)
+			err = reconciler.client.Create(context.TODO(), pvc)
+			Expect(err).ToNot(HaveOccurred())
+
+			done, err = reconciler.validateCloneAndSourcePVC(dv)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(done).To(BeTrue())
+		})
+
+		DescribeTable("Validation mechanism rejects or accepts the clone depending on the contentType combination",
+			func(sourceContentType, targetContentType string, expectedResult bool) {
+				dv := newCloneDataVolume("test-dv")
+				dv.Spec.ContentType = cdiv1.DataVolumeContentType(targetContentType)
+				storageProfile := createStorageProfile(scName, nil, filesystemMode)
+				reconciler = createDatavolumeReconciler(dv, storageProfile, sc)
+
+				done, err := reconciler.validateCloneAndSourcePVC(dv)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(done).To(BeFalse())
+
+				// We create the source PVC after creating the clone
+				pvc := createPvcInStorageClass("test", metav1.NamespaceDefault, &scName, map[string]string{
+					AnnContentType: sourceContentType}, nil, corev1.ClaimBound)
+				err = reconciler.client.Create(context.TODO(), pvc)
+				Expect(err).ToNot(HaveOccurred())
+
+				done, err = reconciler.validateCloneAndSourcePVC(dv)
+				Expect(done).To(Equal(expectedResult))
+				if expectedResult == false {
+					Expect(err).To(HaveOccurred())
+				} else {
+					Expect(err).ToNot(HaveOccurred())
+				}
+			},
+			Entry("Archive in source and target", string(cdiv1.DataVolumeArchive), string(cdiv1.DataVolumeArchive), true),
+			Entry("Archive in source and KubeVirt in target", string(cdiv1.DataVolumeArchive), string(cdiv1.DataVolumeKubeVirt), false),
+			Entry("KubeVirt in source and archive in target", string(cdiv1.DataVolumeKubeVirt), string(cdiv1.DataVolumeArchive), false),
+			Entry("KubeVirt in source and target", string(cdiv1.DataVolumeKubeVirt), string(cdiv1.DataVolumeKubeVirt), true),
+			Entry("Empty (KubeVirt by default) in source and target", "", "", true),
+			Entry("Empty (KubeVirt by default) in source and KubeVirt (explicit) in target", "", string(cdiv1.DataVolumeKubeVirt), true),
+			Entry("KubeVirt (explicit) in source and empty (KubeVirt by default) in target", string(cdiv1.DataVolumeKubeVirt), "", true),
+			Entry("Empty (kubeVirt by default) in source and archive in target", "", string(cdiv1.DataVolumeArchive), false),
+			Entry("Archive in source and empty (KubeVirt by default) in target", string(cdiv1.DataVolumeArchive), "", false),
+		)
+	})
+
 	var _ = Describe("Clone strategy", func() {
 		var (
 			hostAssisted = cdiv1.CloneStrategyHostAssisted