[release-v1.23] Add library function to determine if a PVC has been p…

…opulated fully. (#1401)

* Add library function to determine if a PVC has been populated fully.

The logic is as following:
If PVC has no ownerRef, then we assume something else fully populated it and
will return true
If PVC has an ownerRef and its a DataVolume, then look up the DataVolume
If DV.status.Phase == succeeded, return true, return false otherwise.

Signed-off-by: Alexander Wels <awels@redhat.com>

* Renamed functions to better indicate its purpose.

Signed-off-by: Alexander Wels <awels@redhat.com>

Co-authored-by: Alexander Wels <awels@redhat.com>

pkg/apis/core/v1alpha1/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "register.go",
         "types.go",
         "types_swagger_generated.go",
+        "utils.go",
         "zz_generated.deepcopy.go",
     ],
     importpath = "kubevirt.io/containerized-data-importer/pkg/apis/core/v1alpha1",

pkg/apis/core/v1alpha1/utils.go

@@ -0,0 +1,41 @@
+/*
+Copyright 2020 The CDI Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// IsPopulated indicates if the persistent volume passed in has been fully populated. It follow the following logic
+// 1. If the PVC is not owned by a DataVolume, return true, we assume someone else has properly populated the image
+// 2. If the PVC is owned by a DataVolume, look up the DV and check the phase, if phase succeeded return true
+// 3. If the PVC is owned by a DataVolume, look up the DV and check the phase, if phase !succeeded return false
+func IsPopulated(pvc *corev1.PersistentVolumeClaim, getDvFunc func(name, namespace string) (*DataVolume, error)) (bool, error) {
+	pvcOwner := metav1.GetControllerOf(pvc)
+	if pvcOwner != nil && pvcOwner.Kind == "DataVolume" {
+		// Find the data volume:
+		dv, err := getDvFunc(pvcOwner.Name, pvc.Namespace)
+		if err != nil {
+			return false, err
+		}
+		if dv.Status.Phase != Succeeded {
+			return false, nil
+		}
+	}
+	return true, nil
+}

pkg/apis/core/v1beta1/BUILD.bazel

@@ -8,6 +8,7 @@ go_library(
         "register.go",
         "types.go",
         "types_swagger_generated.go",
+        "utils.go",
         "zz_generated.deepcopy.go",
     ],
     importpath = "kubevirt.io/containerized-data-importer/pkg/apis/core/v1beta1",

pkg/apis/core/v1beta1/utils.go

@@ -0,0 +1,41 @@
+/*
+Copyright 2020 The CDI Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1beta1
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// IsPopulated indicates if the persistent volume passed in has been fully populated. It follow the following logic
+// 1. If the PVC is not owned by a DataVolume, return true, we assume someone else has properly populated the image
+// 2. If the PVC is owned by a DataVolume, look up the DV and check the phase, if phase succeeded return true
+// 3. If the PVC is owned by a DataVolume, look up the DV and check the phase, if phase !succeeded return false
+func IsPopulated(pvc *corev1.PersistentVolumeClaim, getDvFunc func(name, namespace string) (*DataVolume, error)) (bool, error) {
+	pvcOwner := metav1.GetControllerOf(pvc)
+	if pvcOwner != nil && pvcOwner.Kind == "DataVolume" {
+		// Find the data volume:
+		dv, err := getDvFunc(pvcOwner.Name, pvc.Namespace)
+		if err != nil {
+			return false, err
+		}
+		if dv.Status.Phase != Succeeded {
+			return false, nil
+		}
+	}
+	return true, nil
+}

pkg/controller/BUILD.bazel

@@ -88,7 +88,6 @@ go_test(
         "//pkg/util/naming:go_default_library",
         "//tests/reporters:go_default_library",
         "//vendor/github.com/kubernetes-csi/external-snapshotter/v2/pkg/apis/volumesnapshot/v1beta1:go_default_library",
-        "//vendor/github.com/kubevirt/controller-lifecycle-operator-sdk/pkg/sdk/api:go_default_library",
         "//vendor/github.com/onsi/ginkgo:go_default_library",
         "//vendor/github.com/onsi/ginkgo/extensions/table:go_default_library",
         "//vendor/github.com/onsi/gomega:go_default_library",

pkg/controller/clone-controller.go

@@ -212,6 +212,14 @@ func (r *CloneReconciler) reconcileSourcePod(sourcePod *corev1.Pod, targetPvc *c
 			return false, err
 		}
 
+		sourcePopulated, err := IsPopulated(sourcePvc, r.client)
+		if err != nil {
+			return false, err
+		}
+		if !sourcePopulated {
+			return true, nil
+		}
+
 		if err := r.validateSourceAndTarget(sourcePvc, targetPvc); err != nil {
 			return false, err
 		}

pkg/controller/clone-controller_test.go

@@ -30,7 +30,6 @@ import (
 	. "github.com/onsi/gomega"
 
 	corev1 "k8s.io/api/core/v1"
-	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -404,7 +403,7 @@ var _ = Describe("Clone controller reconcile loop", func() {
 		Expect(err.Error()).To(ContainSubstring("source volumeMode (Filesystem) and target volumeMode (Block) do not match"))
 	})
 
-	It("Should error when source and target volume modes do not match (fs->block)", func() {
+	It("Should error when source and target volume modes do not match (block->fs)", func() {
 		testPvc := createPvc("testPvc1", "default", map[string]string{
 			AnnCloneRequest: "default/source", AnnPodReady: "true", AnnCloneToken: "foobaz", AnnUploadClientName: "uploadclient", AnnCloneSourcePod: "default-testPvc1-source-pod"}, nil)
 		reconciler = createCloneReconciler(testPvc, createBlockPvc("source", "default", map[string]string{}, nil))
@@ -421,6 +420,7 @@ var _ = Describe("Clone controller reconcile loop", func() {
 		Expect(err).To(HaveOccurred())
 		Expect(err.Error()).To(ContainSubstring("source volumeMode (Block) and target volumeMode (Filesystem) do not match"))
 	})
+
 })
 
 var _ = Describe("ParseCloneRequestAnnotation", func() {
@@ -567,7 +567,7 @@ var _ = Describe("TokenValidation", func() {
 		}
 	}
 
-	source := &v1.PersistentVolumeClaim{
+	source := &corev1.PersistentVolumeClaim{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "source",
 			Namespace: "sourcens",
@@ -601,7 +601,7 @@ var _ = Describe("TokenValidation", func() {
 			panic("error generating token")
 		}
 
-		target := &v1.PersistentVolumeClaim{
+		target := &corev1.PersistentVolumeClaim{
 			ObjectMeta: metav1.ObjectMeta{
 				Name:      "target",
 				Namespace: "targetns",

pkg/controller/datavolume-controller.go

@@ -264,6 +264,9 @@ func (r *DatavolumeReconciler) Reconcile(req reconcile.Request) (reconcile.Resul
 			if requeue, err := r.sourceInUse(datavolume); requeue || err != nil {
 				return reconcile.Result{Requeue: requeue}, err
 			}
+			if populated, err := r.isSourcePVCPopulated(datavolume); !populated || err != nil {
+				return reconcile.Result{Requeue: !populated}, err
+			}
 			newSnapshot := newSnapshot(datavolume, snapshotClassName)
 			if err := r.client.Create(context.TODO(), newSnapshot); err != nil {
 				if k8serrors.IsAlreadyExists(err) {
@@ -289,6 +292,15 @@ func (r *DatavolumeReconciler) Reconcile(req reconcile.Request) (reconcile.Resul
 	return r.reconcileDataVolumeStatus(datavolume, pvc)
 }
 
+// Verify that the source PVC has been completely populated.
+func (r *DatavolumeReconciler) isSourcePVCPopulated(dv *cdiv1.DataVolume) (bool, error) {
+	sourcePvc := &corev1.PersistentVolumeClaim{}
+	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: dv.Spec.Source.PVC.Name, Namespace: dv.Spec.Source.PVC.Namespace}, sourcePvc); err != nil {
+		return false, err
+	}
+	return IsPopulated(sourcePvc, r.client)
+}
+
 func (r *DatavolumeReconciler) sourceInUse(dv *cdiv1.DataVolume) (bool, error) {
 	pods, err := getPodsUsingPVCs(r.client, dv.Spec.Source.PVC.Namespace, sets.NewString(dv.Spec.Source.PVC.Name), false)
 	if err != nil {

pkg/controller/datavolume-controller_test.go

@@ -554,6 +554,68 @@ var _ = Describe("Reconcile Datavolume status", func() {
 	)
 })
 
+var _ = Describe("sourcePVCPopulated", func() {
+	var (
+		reconciler *DatavolumeReconciler
+	)
+
+	It("Should return true if source has no ownerRef", func() {
+		sourcePvc := createPvc("test", "default", nil, nil)
+		targetDv := newCloneDataVolume("test-dv")
+		reconciler = createDatavolumeReconciler(sourcePvc)
+		res, err := reconciler.isSourcePVCPopulated(targetDv)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(res).To(BeTrue())
+	})
+
+	It("Should return false and error if source has an ownerRef, but it doesn't exist", func() {
+		controller := true
+		sourcePvc := createPvc("test", "default", nil, nil)
+		targetDv := newCloneDataVolume("test-dv")
+		sourcePvc.OwnerReferences = append(sourcePvc.OwnerReferences, metav1.OwnerReference{
+			Kind:       "DataVolume",
+			Controller: &controller,
+		})
+		reconciler = createDatavolumeReconciler(sourcePvc)
+		res, err := reconciler.isSourcePVCPopulated(targetDv)
+		Expect(err).To(HaveOccurred())
+		Expect(res).To(BeFalse())
+	})
+
+	It("Should return false if source has an ownerRef, but it is not succeeded", func() {
+		controller := true
+		sourcePvc := createPvc("test", "default", nil, nil)
+		targetDv := newCloneDataVolume("test-dv")
+		sourceDv := newImportDataVolume("source-dv")
+		sourcePvc.OwnerReferences = append(sourcePvc.OwnerReferences, metav1.OwnerReference{
+			Kind:       "DataVolume",
+			Controller: &controller,
+			Name:       "source-dv",
+		})
+		reconciler = createDatavolumeReconciler(sourcePvc, sourceDv)
+		res, err := reconciler.isSourcePVCPopulated(targetDv)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(res).To(BeFalse())
+	})
+
+	It("Should return true if source has an ownerRef, but it is succeeded", func() {
+		controller := true
+		sourcePvc := createPvc("test", "default", nil, nil)
+		targetDv := newCloneDataVolume("test-dv")
+		sourceDv := newImportDataVolume("source-dv")
+		sourceDv.Status.Phase = cdiv1.Succeeded
+		sourcePvc.OwnerReferences = append(sourcePvc.OwnerReferences, metav1.OwnerReference{
+			Kind:       "DataVolume",
+			Controller: &controller,
+			Name:       "source-dv",
+		})
+		reconciler = createDatavolumeReconciler(sourcePvc, sourceDv)
+		res, err := reconciler.isSourcePVCPopulated(targetDv)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(res).To(BeTrue())
+	})
+})
+
 func podUsingCloneSource(dv *cdiv1.DataVolume, readOnly bool) *corev1.Pod {
 	return &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{

pkg/controller/util.go

@@ -470,3 +470,12 @@ func GetWorkloadNodePlacement(c client.Client) (*cdiv1.NodePlacement, error) {
 	}
 	return &crList.Items[0].Spec.Workloads, nil
 }
+
+// IsPopulated returns if the passed in PVC has been populated according to the rules outlined in pkg/apis/core/<version>/utils.go
+func IsPopulated(pvc *v1.PersistentVolumeClaim, c client.Client) (bool, error) {
+	return cdiv1.IsPopulated(pvc, func(name, namespace string) (*cdiv1.DataVolume, error) {
+		dv := &cdiv1.DataVolume{}
+		err := c.Get(context.TODO(), types.NamespacedName{Name: name, Namespace: namespace}, dv)
+		return dv, err
+	})
+}

tests/cloner_test.go

@@ -82,28 +82,7 @@ var _ = Describe("[rfe_id:1277][crit:high][vendor:cnv-qe@redhat.com][level:compo
 
 		f.ForceBindPvcIfDvIsWaitForFirstConsumer(dataVolume)
 
-		By(fmt.Sprintf("waiting for source datavolume to match phase %s", string(cdiv1.Succeeded)))
-		err = utils.WaitForDataVolumePhase(f.CdiClient, f.Namespace.Name, cdiv1.Succeeded, dataVolume.Name)
-		if err != nil {
-			dv, dverr := f.CdiClient.CdiV1beta1().DataVolumes(f.Namespace.Name).Get(context.TODO(), dataVolume.Name, metav1.GetOptions{})
-			if dverr != nil {
-				Fail(fmt.Sprintf("datavolume %s phase %s", dv.Name, dv.Status.Phase))
-			}
-		}
-		Expect(err).ToNot(HaveOccurred())
-
-		By("verifying pvc content")
 		pvc, err := f.K8sClient.CoreV1().PersistentVolumeClaims(dataVolume.Namespace).Get(context.TODO(), dataVolume.Name, metav1.GetOptions{})
-		sourceMD5, err := f.GetMD5(f.Namespace, pvc, diskImagePath, 0)
-		Expect(err).ToNot(HaveOccurred())
-
-		By("Deleting verifier pod")
-		err = f.K8sClient.CoreV1().Pods(f.Namespace.Name).Delete(context.TODO(), utils.VerifierPodName, metav1.DeleteOptions{})
-		Expect(err).ToNot(HaveOccurred())
-		Eventually(func() bool {
-			_, err := f.K8sClient.CoreV1().Pods(f.Namespace.Name).Get(context.TODO(), utils.VerifierPodName, metav1.GetOptions{})
-			return k8serrors.IsNotFound(err)
-		}, 60, 1).Should(BeTrue())
 
 		// Create targetPvc in new NS.
 		targetDV := utils.NewCloningDataVolume("target-dv", "1G", pvc)
@@ -119,6 +98,17 @@ var _ = Describe("[rfe_id:1277][crit:high][vendor:cnv-qe@redhat.com][level:compo
 		fmt.Fprintf(GinkgoWriter, "INFO: %s\n", sourcePvcDiskGroup)
 		Expect(err).ToNot(HaveOccurred())
 
+		By("verifying pvc content")
+		sourceMD5, err := f.GetMD5(f.Namespace, pvc, diskImagePath, 0)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Deleting verifier pod")
+		err = f.K8sClient.CoreV1().Pods(f.Namespace.Name).Delete(context.TODO(), utils.VerifierPodName, metav1.DeleteOptions{})
+		Expect(err).ToNot(HaveOccurred())
+		Eventually(func() bool {
+			_, err := f.K8sClient.CoreV1().Pods(f.Namespace.Name).Get(context.TODO(), utils.VerifierPodName, metav1.GetOptions{})
+			return k8serrors.IsNotFound(err)
+		}, 60, 1).Should(BeTrue())
 		completeClone(f, f.Namespace, targetPvc, diskImagePath, sourceMD5, sourcePvcDiskGroup)
 	})