Adjust code to handle a clone of missing source PVC

In case the target pvc exists and populated we should allow the clone
to complete successfully without doing anything.
Adjust the code and add a test for that case.

Signed-off-by: Shelly Kagan <skagan@redhat.com>

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, NoClone)
+			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() {