[release-v1.43] Manual backport of 'Allow creating clones without source PVC (#2306)'

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