Delete old version DV's with DIC GC

Signed-off-by: Ido Aharon <iaharon@redhat.com>

pkg/controller/dataimportcron-controller.go

@@ -606,6 +606,29 @@ func (r *DataImportCronReconciler) garbageCollectOldImports(ctx context.Context,
 			}
 		}
 	}
+
+	// search dv's
+
+	dvList := &cdiv1.DataVolumeList{}
+	if err := r.client.List(ctx, dvList, &client.ListOptions{Namespace: cron.Namespace, LabelSelector: selector}); err != nil {
+		return err
+	}
+
+	for _, dv := range dvList.Items {
+		pvc := &corev1.PersistentVolumeClaim{}
+		if err = r.client.Get(ctx, types.NamespacedName{Namespace: cron.Namespace, Name: dv.Name}, pvc); err != nil {
+			return err
+		}
+
+		if pvc.Labels[common.DataImportCronLabel] != cron.Name {
+			if err := r.client.Delete(ctx, &dv); err != nil {
+				if !k8serrors.IsNotFound(err) {
+					return err
+				}
+			}
+		}
+	}
+
 	return nil
 }
 

pkg/controller/dataimportcron-controller_test.go

@@ -36,6 +36,7 @@ import (
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/tools/record"
@@ -692,6 +693,60 @@ var _ = Describe("All DataImportCron Tests", func() {
 			Entry("has no tag", imageStreamName, 1),
 		)
 
+		It("XXXXXX Should succeed garbage collecting old version DVs", func() {
+			cron = newDataImportCron(cronName)
+			cron.Annotations[AnnSourceDesiredDigest] = testDigest
+			reconciler = createDataImportCronReconciler(cron)
+
+			dv := cc.NewImportDataVolume("test-dv")
+			dv.Labels = map[string]string{common.DataImportCronLabel: cronName}
+			err := reconciler.client.Create(context.TODO(), dv)
+			Expect(err).ToNot(HaveOccurred())
+
+			pvc := cc.CreatePvc(dv.Name, dv.Namespace, nil, nil)
+			pvcOwnerReferences := []metav1.OwnerReference{
+				*metav1.NewControllerRef(dv, schema.GroupVersionKind{
+					Group:   cdiv1.SchemeGroupVersion.Group,
+					Version: cdiv1.SchemeGroupVersion.Version,
+					Kind:    "DataVolume",
+				}),
+			}
+			pvc.SetOwnerReferences(pvcOwnerReferences)
+			err = reconciler.client.Create(context.TODO(), pvc)
+			Expect(err).ToNot(HaveOccurred())
+
+			// unlabeled dv
+			unlabeledDv := cc.NewImportDataVolume("test-dv2")
+			err = reconciler.client.Create(context.TODO(), unlabeledDv)
+			Expect(err).ToNot(HaveOccurred())
+
+			unlabeledPvc := cc.CreatePvc(unlabeledDv.Name, unlabeledDv.Namespace, nil, nil)
+			pvcOwnerReferences = []metav1.OwnerReference{
+				*metav1.NewControllerRef(unlabeledDv, schema.GroupVersionKind{
+					Group:   cdiv1.SchemeGroupVersion.Group,
+					Version: cdiv1.SchemeGroupVersion.Version,
+					Kind:    "DataVolume",
+				}),
+			}
+			unlabeledPvc.SetOwnerReferences(pvcOwnerReferences)
+			err = reconciler.client.Create(context.TODO(), unlabeledPvc)
+			Expect(err).ToNot(HaveOccurred())
+
+			reconciler.garbageCollectOldImports(context.TODO(), cron)
+
+			// Ensure the labeled DV is deleted
+			err = reconciler.client.Get(context.TODO(), dvKey(dv.Name), dv)
+			Expect(k8serrors.IsNotFound(err)).To(BeTrue())
+			err = reconciler.client.Get(context.TODO(), dvKey(dv.Name), pvc)
+			Expect(k8serrors.IsNotFound(err)).To(BeTrue())
+
+			// Ensure unlabeled DV and PVC are not deleted
+			err = reconciler.client.Get(context.TODO(), dvKey(unlabeledDv.Name), unlabeledDv)
+			Expect(k8serrors.IsNotFound(err)).To(BeFalse())
+			err = reconciler.client.Get(context.TODO(), dvKey(unlabeledDv.Name), unlabeledPvc)
+			Expect(k8serrors.IsNotFound(err)).To(BeFalse())
+		})
+
 		It("should pass through defaultInstancetype and defaultPreference metadata to DataVolume and DataSource", func() {
 			cron = newDataImportCron(cronName)
 			cron.Annotations[AnnSourceDesiredDigest] = testDigest

tests/dataimportcron_test.go

@@ -22,6 +22,8 @@ import (
 	"kubevirt.io/containerized-data-importer/pkg/controller"
 	"kubevirt.io/containerized-data-importer/tests/framework"
 	"kubevirt.io/containerized-data-importer/tests/utils"
+
+	cc "kubevirt.io/containerized-data-importer/pkg/controller/common"
 )
 
 const (
@@ -42,6 +44,7 @@ var _ = Describe("DataImportCron", func() {
 		reg            *cdiv1.DataVolumeSourceRegistry
 		err            error
 		ns             string
+		dvName         = "dv-garbage"
 	)
 
 	BeforeEach(func() {
@@ -246,6 +249,89 @@ var _ = Describe("DataImportCron", func() {
 		table.Entry("[test_id:8266] succeed deleting error DVs when importing new ones", false, true, 2),
 	)
 
+	FIt("[test_id:XXXX] succeed garbage collecting old version dv's", func() {
+		reg, err := getDataVolumeSourceRegistry(f)
+		//Expect(err).To(BeNil())
+		Expect(err).ToNot(HaveOccurred())
+		//defer utils.RemoveInsecureRegistry(f.CrClient, *reg.URL)
+
+		By(fmt.Sprintf("Create labeled DataVolume %s for garbage collection test", dvName))
+		dv := utils.NewDataVolumeWithRegistryImport(dvName, "5Gi", "")
+		dv.Spec.Source.Registry = reg
+		dv.Labels = map[string]string{common.DataImportCronLabel: cronName}
+		cc.AddAnnotation(dv, cc.AnnDeleteAfterCompletion, "false")
+		dv, err = utils.CreateDataVolumeFromDefinition(f.CdiClient, ns, dv)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Wait for import completion")
+		err = utils.WaitForDataVolumePhase(f, ns, cdiv1.Succeeded, dv.Name)
+		Expect(err).ToNot(HaveOccurred(), "Datavolume not in phase succeeded in time")
+
+		By(fmt.Sprintf("Verify pvc was created %s", dv.Name))
+		pvc, err := utils.WaitForPVC(f.K8sClient, ns, dv.Name)
+		Expect(err).ToNot(HaveOccurred())
+		By(fmt.Sprintf("Verify DataImportCronLabel is passed to the pvc: %s", pvc.Labels[common.DataImportCronLabel]))
+		Expect(pvc.Labels[common.DataImportCronLabel]).To(Equal(cronName))
+
+		pvc.Labels[common.DataImportCronLabel] = ""
+		By("Update DataImportCron label to be empty in the pvc")
+		_, err = f.K8sClient.CoreV1().PersistentVolumeClaims(pvc.Namespace).Update(context.TODO(), pvc, metav1.UpdateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		By(fmt.Sprintf("Create new DataImportCron %s, url %s", cronName, *reg.URL))
+		cron = utils.NewDataImportCron(cronName, "5Gi", scheduleEveryMinute, dataSourceName, importsToKeep, *reg)
+		retentionPolicy := cdiv1.DataImportCronRetainNone
+		cron.Spec.RetentionPolicy = &retentionPolicy
+		cron, err = f.CdiClient.CdiV1beta1().DataImportCrons(ns).Create(context.TODO(), cron, metav1.CreateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Wait for conditions")
+		waitForConditions(corev1.ConditionFalse, corev1.ConditionTrue)
+
+		By("Verify CurrentImports update")
+		currentImportDvName := cron.Status.CurrentImports[0].DataVolumeName
+		Expect(currentImportDvName).ToNot(BeEmpty())
+
+		By(fmt.Sprintf("Verify pvc was created %s", currentImportDvName))
+		_, err = utils.WaitForPVC(f.K8sClient, ns, currentImportDvName)
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Wait for import completion")
+		err = utils.WaitForDataVolumePhase(f, ns, cdiv1.Succeeded, currentImportDvName)
+		Expect(err).ToNot(HaveOccurred(), "Datavolume not in phase succeeded in time")
+
+		By("Check garbage collection")
+		Eventually(func() int {
+			dvList, err := f.CdiClient.CdiV1beta1().DataVolumes(ns).List(context.TODO(), metav1.ListOptions{})
+			Expect(err).ToNot(HaveOccurred())
+			return len(dvList.Items)
+		}, dataImportCronTimeout, pollingInterval).Should(Equal(0), "Garbage collection failed cleaning old imports")
+
+		By("Check that the PVC is timestamped and not garbage collected")
+		pvc, err = f.K8sClient.CoreV1().PersistentVolumeClaims(ns).Get(context.TODO(), currentImportDvName, metav1.GetOptions{})
+		Expect(err).ToNot(HaveOccurred())
+		// timestamp
+		lastUse := pvc.Annotations[controller.AnnLastUseTime]
+		Expect(lastUse).ToNot(BeEmpty())
+		ts, err := time.Parse(time.RFC3339Nano, lastUse)
+		Expect(err).ToNot(HaveOccurred())
+		Expect(ts).To(BeTemporally("<", time.Now()))
+
+		By("Delete cron")
+		err = f.CdiClient.CdiV1beta1().DataImportCrons(ns).Delete(context.TODO(), cronName, metav1.DeleteOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		By("Verify DataSource deletion")
+		Eventually(func() bool {
+			_, err := f.CdiClient.CdiV1beta1().DataSources(ns).Get(context.TODO(), dataSourceName, metav1.GetOptions{})
+			return errors.IsNotFound(err)
+		}, dataImportCronTimeout, pollingInterval).Should(BeTrue(), "DataSource was not deleted")
+
+		By("Verify original PVC deletion")
+		_, err = f.K8sClient.CoreV1().PersistentVolumeClaims(ns).Get(context.TODO(), dv.Name, metav1.GetOptions{})
+		Expect(err).To(HaveOccurred())
+	})
+
 	It("[test_id:10040] Should get digest updated by external poller", func() {
 		By("Create DataImportCron with only initial poller job")
 		cron = utils.NewDataImportCron(cronName, "5Gi", scheduleOnceAYear, dataSourceName, importsToKeep, *reg)