Watch DataSources in DV clone controller (kubevirt#2571)

DV with sourceRef DataSource pointing to non-existing PVC, is to be
reconciled when the PVC is created and populated and the DataSource
is Ready.

Signed-off-by: Arnon Gilboa <agilboa@redhat.com>

pkg/controller/datavolume/pvc-clone-controller.go

@@ -46,8 +46,10 @@ import (
 	"kubevirt.io/containerized-data-importer/pkg/util"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"sigs.k8s.io/controller-runtime/pkg/source"
 )
 
 type cloneStrategy int
@@ -170,6 +172,52 @@ func addDataVolumeCloneControllerWatches(mgr manager.Manager, datavolumeControll
 		return err
 	}
 
+	if err := addDataSourceWatch(mgr, datavolumeController); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func addDataSourceWatch(mgr manager.Manager, c controller.Controller) error {
+	const dvDataSourceField = "datasource"
+
+	getKey := func(namespace, name string) string {
+		return namespace + "/" + name
+	}
+
+	if err := mgr.GetFieldIndexer().IndexField(context.TODO(), &cdiv1.DataVolume{}, dvDataSourceField, func(obj client.Object) []string {
+		if sourceRef := obj.(*cdiv1.DataVolume).Spec.SourceRef; sourceRef != nil && sourceRef.Kind == cdiv1.DataVolumeDataSource {
+			ns := obj.GetNamespace()
+			if sourceRef.Namespace != nil && *sourceRef.Namespace != "" {
+				ns = *sourceRef.Namespace
+			}
+			return []string{getKey(ns, sourceRef.Name)}
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+
+	mapToDataVolume := func(obj client.Object) (reqs []reconcile.Request) {
+		var dvs cdiv1.DataVolumeList
+		matchingFields := client.MatchingFields{dvDataSourceField: getKey(obj.GetNamespace(), obj.GetName())}
+		if err := mgr.GetClient().List(context.TODO(), &dvs, matchingFields); err != nil {
+			c.GetLogger().Error(err, "Unable to list DataVolumes", "matchingFields", matchingFields)
+			return
+		}
+		for _, dv := range dvs.Items {
+			reqs = append(reqs, reconcile.Request{NamespacedName: types.NamespacedName{Namespace: dv.Namespace, Name: dv.Name}})
+		}
+		return
+	}
+
+	if err := c.Watch(&source.Kind{Type: &cdiv1.DataSource{}},
+		handler.EnqueueRequestsFromMapFunc(mapToDataVolume),
+	); err != nil {
+		return err
+	}
+
 	return nil
 }
 

tests/datasource_test.go

@@ -9,10 +9,12 @@ import (
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
 	"kubevirt.io/containerized-data-importer/pkg/controller"
+	dvc "kubevirt.io/containerized-data-importer/pkg/controller/datavolume"
 	"kubevirt.io/containerized-data-importer/tests/framework"
 	"kubevirt.io/containerized-data-importer/tests/utils"
 )
@@ -65,20 +67,40 @@ var _ = Describe("DataSource", func() {
 	}
 
 	It("[test_id:8041]status conditions should be updated on pvc create/update/delete", func() {
-		By("creating datasource")
+		By("Create DataSource with no source PVC")
 		ds := newDataSource(ds1Name)
 		ds, err := f.CdiClient.CdiV1beta1().DataSources(ds.Namespace).Create(context.TODO(), ds, metav1.CreateOptions{})
 		Expect(err).ToNot(HaveOccurred())
 		ds = waitForReadyCondition(ds, corev1.ConditionFalse, "NoPvc")
 
+		By("Update DataSource source PVC to nonexisting one")
 		ds.Spec.Source.PVC = &cdiv1.DataVolumeSourcePVC{Namespace: f.Namespace.Name, Name: pvc1Name}
 		ds, err = f.CdiClient.CdiV1beta1().DataSources(ds.Namespace).Update(context.TODO(), ds, metav1.UpdateOptions{})
 		Expect(err).ToNot(HaveOccurred())
 		ds = waitForReadyCondition(ds, corev1.ConditionFalse, "NotFound")
 
+		By("Create clone DV with SourceRef pointing the DataSource")
+		dv := utils.NewDataVolumeWithSourceRef("clone-dv", "1Gi", ds.Namespace, ds.Name)
+		dv.Annotations[controller.AnnImmediateBinding] = "true"
+		Expect(dv).ToNot(BeNil())
+		dv, err = utils.CreateDataVolumeFromDefinition(f.CdiClient, f.Namespace.Name, dv)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify DV conditions")
+		utils.WaitForConditions(f, dv.Name, dv.Namespace, time.Minute, pollingInterval,
+			&cdiv1.DataVolumeCondition{Type: cdiv1.DataVolumeBound, Status: v1.ConditionUnknown, Message: "No PVC found", Reason: dvc.CloneWithoutSource},
+			&cdiv1.DataVolumeCondition{Type: cdiv1.DataVolumeReady, Status: v1.ConditionFalse, Reason: dvc.MessageCloneWithoutSource},
+			&cdiv1.DataVolumeCondition{Type: cdiv1.DataVolumeRunning, Status: v1.ConditionFalse})
+		f.ExpectEvent(dv.Namespace).Should(ContainSubstring(dvc.CloneWithoutSource))
+
+		By("Create import DV so the missing DataSource source PVC will be ready")
 		createDv(pvc1Name, testUrl())
 		ds = waitForReadyCondition(ds, corev1.ConditionTrue, "Ready")
 
+		By("Wait for the clone DV success")
+		err = utils.WaitForDataVolumePhase(f, dv.Namespace, cdiv1.Succeeded, dv.Name)
+		Expect(err).ToNot(HaveOccurred())
+
 		deleteDvPvc(f, pvc1Name)
 		ds = waitForReadyCondition(ds, corev1.ConditionFalse, "NotFound")