[release-v1.43] Manual fix for clone without source pvc but target pvc already populated

pkg/apiserver/webhooks/dataimportcron-validate.go

@@ -109,7 +109,7 @@ func (wh *dataImportCronValidatingWebhook) validateDataImportCronSpec(request *a
 		return causes
 	}
 
-	causes = wh.validateDataVolumeSpec(request, k8sfield.NewPath("Template"), &spec.Template.Spec, nil)
+	causes = wh.validateDataVolumeSpec(request, k8sfield.NewPath("Template"), &spec.Template.Spec, nil, "")
 	if len(causes) > 0 {
 		return causes
 	}

pkg/apiserver/webhooks/datavolume-validate.go

@@ -81,7 +81,7 @@ func validateContentTypes(sourcePVC *v1.PersistentVolumeClaim, spec *cdiv1.DataV
 	return sourceContentType == targetContentType, sourceContentType, targetContentType
 }
 
-func (wh *dataVolumeValidatingWebhook) validateDataVolumeSpec(request *admissionv1.AdmissionRequest, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec, namespace *string) []metav1.StatusCause {
+func (wh *dataVolumeValidatingWebhook) validateDataVolumeSpec(request *admissionv1.AdmissionRequest, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec, namespace *string, name string) []metav1.StatusCause {
 	var causes []metav1.StatusCause
 	var url string
 	var sourceType string
@@ -163,7 +163,7 @@ func (wh *dataVolumeValidatingWebhook) validateDataVolumeSpec(request *admission
 		return causes
 	}
 	if spec.SourceRef != nil {
-		cause := wh.validateSourceRef(request, spec, field, namespace)
+		cause := wh.validateSourceRef(request, spec, field, namespace, name)
 		if cause != nil {
 			causes = append(causes, *cause)
 		}
@@ -290,7 +290,7 @@ func (wh *dataVolumeValidatingWebhook) validateDataVolumeSpec(request *admission
 			return causes
 		}
 		if request.Operation == admissionv1.Create {
-			cause := wh.validateDataVolumeSourcePVC(spec.Source.PVC, field.Child("source", "PVC"), spec)
+			cause := wh.validateDataVolumeSourcePVC(spec.Source.PVC, field.Child("source", "PVC"), spec, namespace, name)
 			if cause != nil {
 				causes = append(causes, *cause)
 			}
@@ -363,7 +363,7 @@ func validateDataVolumeSourceRegistry(sourceRegistry *cdiv1.DataVolumeSourceRegi
 	return causes
 }
 
-func (wh *dataVolumeValidatingWebhook) validateSourceRef(request *admissionv1.AdmissionRequest, spec *cdiv1.DataVolumeSpec, field *k8sfield.Path, namespace *string) *metav1.StatusCause {
+func (wh *dataVolumeValidatingWebhook) validateSourceRef(request *admissionv1.AdmissionRequest, spec *cdiv1.DataVolumeSpec, field *k8sfield.Path, namespace *string, name string) *metav1.StatusCause {
 	if spec.SourceRef.Kind == "" {
 		return &metav1.StatusCause{
 			Type:    metav1.CauseTypeFieldValueInvalid,
@@ -406,13 +406,29 @@ func (wh *dataVolumeValidatingWebhook) validateSourceRef(request *admissionv1.Ad
 			Field:   field.Child("sourceRef").String(),
 		}
 	}
-	return wh.validateDataVolumeSourcePVC(dataSource.Spec.Source.PVC, field.Child("sourceRef"), spec)
+	return wh.validateDataVolumeSourcePVC(dataSource.Spec.Source.PVC, field.Child("sourceRef"), spec, namespace, name)
 }
 
-func (wh *dataVolumeValidatingWebhook) validateDataVolumeSourcePVC(PVC *cdiv1.DataVolumeSourcePVC, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec) *metav1.StatusCause {
+func (wh *dataVolumeValidatingWebhook) validateDataVolumeSourcePVC(PVC *cdiv1.DataVolumeSourcePVC, field *k8sfield.Path, spec *cdiv1.DataVolumeSpec, namespace *string, name string) *metav1.StatusCause {
 	sourcePVC, err := wh.k8sClient.CoreV1().PersistentVolumeClaims(PVC.Namespace).Get(context.TODO(), PVC.Name, metav1.GetOptions{})
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
+			if namespace != nil {
+				// If the target PVC is already populated we don't need the source PVC to exists
+				targetPVC, err := wh.k8sClient.CoreV1().PersistentVolumeClaims(*namespace).Get(context.TODO(), name, metav1.GetOptions{})
+				if err == nil {
+					populatedFor, ok := targetPVC.Annotations[controller.AnnPopulatedFor]
+					if ok && populatedFor == name {
+						return nil
+					}
+				} else if !k8serrors.IsNotFound(err) {
+					return &metav1.StatusCause{
+						Message: err.Error(),
+						Field:   field.String(),
+					}
+				}
+			}
+
 			return &metav1.StatusCause{
 				Type:    metav1.CauseTypeFieldValueNotFound,
 				Message: fmt.Sprintf("Source PVC %s/%s not found", PVC.Namespace, PVC.Name),
@@ -541,7 +557,7 @@ func (wh *dataVolumeValidatingWebhook) Admit(ar admissionv1.AdmissionReview) *ad
 		}
 	}
 
-	causes = wh.validateDataVolumeSpec(ar.Request, k8sfield.NewPath("spec"), &dv.Spec, &dv.Namespace)
+	causes = wh.validateDataVolumeSpec(ar.Request, k8sfield.NewPath("spec"), &dv.Spec, &dv.Namespace, dv.Name)
 	if len(causes) > 0 {
 		klog.Infof("rejected DataVolume admission %s", causes)
 		return toRejectedAdmissionResponse(causes)

pkg/apiserver/webhooks/datavolume-validate_test.go

@@ -188,6 +188,22 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(Equal(false))
 		})
 
+		It("should accept DataVolume with PVC source on create if source PVC does not exist, but target pvc exists and populated", func() {
+			dataVolume := newPVCDataVolume("testDV", "testNamespace", "test")
+			targetPVC := &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      dataVolume.Name,
+					Namespace: dataVolume.Namespace,
+					Annotations: map[string]string{
+						"cdi.kubevirt.io/storage.populatedFor": dataVolume.Name,
+					},
+				},
+				Spec: *dataVolume.Spec.PVC,
+			}
+			resp := validateDataVolumeCreate(dataVolume, targetPVC)
+			Expect(resp.Allowed).To(Equal(true))
+		})
+
 		It("should reject invalid DataVolume source PVC namespace on create", func() {
 			dataVolume := newPVCDataVolume("testDV", "", "test")
 			resp := validateDataVolumeCreate(dataVolume)
@@ -505,6 +521,36 @@ var _ = Describe("Validating Webhook", func() {
 			Expect(resp.Allowed).To(Equal(false))
 		})
 
+		It("should accept DataVolume with SourceRef on create if DataSource exists, source PVC does not exist, but target pvc exists and populated", func() {
+			dataVolume := newDataSourceDataVolume("testDV", &testNamespace, "test")
+			dataSource := &cdiv1.DataSource{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      dataVolume.Spec.SourceRef.Name,
+					Namespace: testNamespace,
+				},
+				Spec: cdiv1.DataSourceSpec{
+					Source: cdiv1.DataSourceSource{
+						PVC: &cdiv1.DataVolumeSourcePVC{
+							Name:      "testPVC",
+							Namespace: testNamespace,
+						},
+					},
+				},
+			}
+			targetPVC := &corev1.PersistentVolumeClaim{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      dataVolume.Name,
+					Namespace: dataVolume.Namespace,
+					Annotations: map[string]string{
+						"cdi.kubevirt.io/storage.populatedFor": dataVolume.Name,
+					},
+				},
+				Spec: *dataVolume.Spec.PVC,
+			}
+			resp := validateDataVolumeCreateEx(dataVolume, []runtime.Object{targetPVC}, []runtime.Object{dataSource})
+			Expect(resp.Allowed).To(Equal(true))
+		})
+
 		It("should reject DataVolume with empty SourceRef name on create", func() {
 			dataVolume := newDataSourceDataVolume("testDV", &testNamespace, "")
 			resp := validateDataVolumeCreate(dataVolume)

pkg/controller/datavolume-controller.go

@@ -473,26 +473,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 +517,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,10 +526,20 @@ 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) {
 
+	var err error
+	selectedCloneStrategy := NoClone
 	if !prePopulated && !pvcPopulated {
+		selectedCloneStrategy, err = r.selectCloneStrategy(datavolume, pvcSpec)
+		if err != nil {
+			return reconcile.Result{}, err
+		}
+
+		if selectedCloneStrategy == SmartClone {
+			r.sccs.StartController()
+		}
+
 		if pvc == nil {
 			if selectedCloneStrategy == SmartClone {
 				snapshotClassName, _ := r.getSnapshotClassForSmartClone(datavolume, pvcSpec)
@@ -646,10 +648,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

pkg/controller/datavolume-controller_test.go

@@ -1552,6 +1552,38 @@ var _ = Describe("All DataVolume Tests", func() {
 			Entry("should default to snapshot", nil, nil, cdiv1.CloneStrategySnapshot),
 		)
 	})
+
+	var _ = Describe("Clone without source", func() {
+		scName := "testsc"
+		sc := createStorageClassWithProvisioner(scName, map[string]string{
+			AnnDefaultStorageClass: "true",
+		}, "csi-plugin")
+
+		It("Validate clone already populated without source completes", func() {
+			dv := newCloneDataVolume("test-dv")
+			storageProfile := createStorageProfile(scName, nil, filesystemMode)
+			pvcAnnotations := make(map[string]string)
+			pvcAnnotations[AnnPopulatedFor] = "test-dv"
+			pvc := createPvcInStorageClass("test-dv", metav1.NamespaceDefault, &scName, nil, nil, corev1.ClaimBound)
+			pvc.SetAnnotations(make(map[string]string))
+			pvc.GetAnnotations()[AnnPopulatedFor] = "test-dv"
+			reconciler = createDatavolumeReconciler(dv, pvc, storageProfile, sc)
+
+			prePopulated := false
+			pvcPopulated := true
+			result, err := reconciler.reconcileClone(reconciler.log, dv, pvc, dv.Spec.PVC.DeepCopy(), "", prePopulated, pvcPopulated)
+			Expect(err).ToNot(HaveOccurred())
+			err = reconciler.client.Get(context.TODO(), types.NamespacedName{Name: "test-dv", Namespace: metav1.NamespaceDefault}, dv)
+			Expect(err).ToNot(HaveOccurred())
+			Expect(dv.Status.ClaimName).To(Equal("test-dv"))
+			Expect(dv.Status.Phase).To(Equal(cdiv1.Succeeded))
+			Expect(dv.Annotations[AnnPrePopulated]).To(Equal("test-dv"))
+			Expect(dv.Annotations[annCloneType]).To(BeEmpty())
+			Expect(result).To(Equal(reconcile.Result{}))
+		})
+
+	})
+
 	var _ = Describe("Get Pod from PVC", func() {
 		var (
 			pvc *corev1.PersistentVolumeClaim

tests/BUILD.bazel

@@ -106,6 +106,7 @@ go_test(
         "//vendor/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//vendor/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//vendor/k8s.io/apimachinery/pkg/util/rand:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes:go_default_library",
         "//vendor/k8s.io/client-go/kubernetes/scheme:go_default_library",
         "//vendor/k8s.io/client-go/tools/clientcmd:go_default_library",

tests/cloner_test.go

@@ -18,6 +18,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/rand"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
@@ -776,6 +777,30 @@ var _ = Describe("all clone tests", func() {
 				Expect(utils.GetCloneType(f.CdiClient, dataVolume)).To(Equal("csivolumeclone"))
 			})
 		})
+
+		Context("Clone without a source PVC", func() {
+			It("Should not clone when PopulatedFor annotation exists", func() {
+				fsVM := v1.PersistentVolumeMode(v1.PersistentVolumeFilesystem)
+				targetName := "target" + rand.String(12)
+
+				By(fmt.Sprintf("Creating target pvc: %s/%s", f.Namespace.Name, targetName))
+				targetPvc, err := utils.CreatePVCFromDefinition(f.K8sClient, f.Namespace.Name,
+					utils.NewPVCDefinition(targetName, "1Gi", map[string]string{controller.AnnPopulatedFor: targetName}, nil))
+				Expect(err).ToNot(HaveOccurred())
+				f.ForceBindIfWaitForFirstConsumer(targetPvc)
+				cloneDV := utils.NewDataVolumeForImageCloningAndStorageSpec(targetName, "1Gi", f.Namespace.Name, "non-existing-source", nil, &fsVM)
+				_, err = utils.CreateDataVolumeFromDefinition(f.CdiClient, f.Namespace.Name, cloneDV)
+				Expect(err).ToNot(HaveOccurred())
+				By("Wait for clone DV Succeeded phase")
+				err = utils.WaitForDataVolumePhaseWithTimeout(f.CdiClient, f.Namespace.Name, cdiv1.Succeeded, targetName, cloneCompleteTimeout)
+				Expect(err).ToNot(HaveOccurred())
+				dv, err := f.CdiClient.CdiV1beta1().DataVolumes(f.Namespace.Name).Get(context.TODO(), targetName, metav1.GetOptions{})
+				Expect(err).ToNot(HaveOccurred())
+				_, ok := dv.Annotations["cdi.kubevirt.io/cloneType"]
+				Expect(ok).To(BeFalse())
+			})
+		})
+
 	})
 
 	var _ = Describe("Validate creating multiple clones of same source Data Volume", func() {