Delete old version DVs with DIC GC

pkg/controller/dataimportcron-controller.go

@@ -804,7 +804,7 @@ func (r *DataImportCronReconciler) garbageCollectOldImports(ctx context.Context,
 		maxImports = int(*cron.Spec.ImportsToKeep)
 	}
 
-	if err := r.garbageCollectPVCs(ctx, cron.Namespace, selector, maxImports); err != nil {
+	if err := r.garbageCollectPVCs(ctx, cron.Namespace, cron.Name, selector, maxImports); err != nil {
 		return err
 	}
 	if err := r.garbageCollectSnapshots(ctx, cron.Namespace, selector, maxImports); err != nil {
@@ -814,7 +814,7 @@ func (r *DataImportCronReconciler) garbageCollectOldImports(ctx context.Context,
 	return nil
 }
 
-func (r *DataImportCronReconciler) garbageCollectPVCs(ctx context.Context, namespace string, selector labels.Selector, maxImports int) error {
+func (r *DataImportCronReconciler) garbageCollectPVCs(ctx context.Context, namespace, cronName string, selector labels.Selector, maxImports int) error {
 	pvcList := &corev1.PersistentVolumeClaimList{}
 
 	if err := r.client.List(ctx, pvcList, &client.ListOptions{Namespace: namespace, LabelSelector: selector}); err != nil {
@@ -832,6 +832,27 @@ func (r *DataImportCronReconciler) garbageCollectPVCs(ctx context.Context, names
 		}
 	}
 
+	dvList := &cdiv1.DataVolumeList{}
+	if err := r.client.List(ctx, dvList, &client.ListOptions{Namespace: namespace, LabelSelector: selector}); err != nil {
+		return err
+	}
+
+	if len(dvList.Items) > maxImports {
+		for _, dv := range dvList.Items {
+			pvc := &corev1.PersistentVolumeClaim{}
+			if err := r.client.Get(ctx, types.NamespacedName{Namespace: namespace, Name: dv.Name}, pvc); err != nil {
+				return err
+			}
+
+			if pvc.Labels[common.DataImportCronLabel] != cronName {
+				r.log.Info("Deleting old version dv/pvc", "name", pvc.Name, "pvc.uid", pvc.UID)
+				if err := r.deleteDvPvc(ctx, dv.Name, dv.Namespace); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
 	return nil
 }
 

pkg/controller/dataimportcron-controller_test.go

@@ -706,6 +706,63 @@ var _ = Describe("All DataImportCron Tests", func() {
 			Entry("has no tag", imageStreamName, 1),
 		)
 
+		It("Should succeed garbage collecting old version DVs", func() {
+			cron = newDataImportCron(cronName)
+			importsToKeep := int32(1)
+			cron.Spec.ImportsToKeep = &importsToKeep
+			reconciler = createDataImportCronReconciler(cron)
+
+			// Labeled DV and unlabeled PVC
+			dv1 := cc.NewImportDataVolume("test-dv1")
+			dv1.Labels = map[string]string{common.DataImportCronLabel: cronName}
+			err := reconciler.client.Create(context.TODO(), dv1)
+			Expect(err).ToNot(HaveOccurred())
+
+			pvc1 := cc.CreatePvc(dv1.Name, dv1.Namespace, nil, nil)
+			err = reconciler.client.Create(context.TODO(), pvc1)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Labeled DV and PVC
+			dv2 := cc.NewImportDataVolume("test-dv2")
+			dv2.Labels = map[string]string{common.DataImportCronLabel: cronName}
+			err = reconciler.client.Create(context.TODO(), dv2)
+			Expect(err).ToNot(HaveOccurred())
+
+			pvc2 := cc.CreatePvc(dv2.Name, dv2.Namespace, nil, nil)
+			pvc2.Labels = map[string]string{common.DataImportCronLabel: cronName}
+			err = reconciler.client.Create(context.TODO(), pvc2)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Unlabeled DV and PVC
+			dv3 := cc.NewImportDataVolume("test-dv3")
+			err = reconciler.client.Create(context.TODO(), dv3)
+			Expect(err).ToNot(HaveOccurred())
+
+			pvc3 := cc.CreatePvc(dv3.Name, dv3.Namespace, nil, nil)
+			err = reconciler.client.Create(context.TODO(), pvc3)
+			Expect(err).ToNot(HaveOccurred())
+
+			err = reconciler.garbageCollectOldImports(context.TODO(), cron)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Ensure the old version DV is deleted (labeled DV and unlabeled PVC).
+			// The labeled PVC will not be deleted here, as there is no relevant controller.
+			err = reconciler.client.Get(context.TODO(), dvKey(dv1.Name), dv1)
+			Expect(k8serrors.IsNotFound(err)).To(BeTrue())
+
+			// Ensure the new version DV is not deleted (labeled DV and labeled PVC).
+			err = reconciler.client.Get(context.TODO(), dvKey(dv2.Name), dv2)
+			Expect(err).ToNot(HaveOccurred())
+			err = reconciler.client.Get(context.TODO(), dvKey(pvc2.Name), pvc2)
+			Expect(err).ToNot(HaveOccurred())
+
+			// Ensure unrelated DVs and PVCs are not deleted (unlabeled DV and PVC)
+			err = reconciler.client.Get(context.TODO(), dvKey(dv3.Name), dv3)
+			Expect(err).ToNot(HaveOccurred())
+			err = reconciler.client.Get(context.TODO(), dvKey(pvc3.Name), pvc3)
+			Expect(err).ToNot(HaveOccurred())
+		})
+
 		It("should pass through metadata to DataVolume and DataSource", func() {
 			cron = newDataImportCron(cronName)
 			cron.Annotations[AnnSourceDesiredDigest] = testDigest

tests/dataimportcron_test.go

@@ -443,6 +443,7 @@ var _ = Describe("DataImportCron", func() {
 		if format == cdiv1.DataImportCronSourceFormatSnapshot && !f.IsSnapshotStorageClassAvailable() {
 			Skip("Volumesnapshot support needed to test DataImportCron with Volumesnapshot sources")
 		}
+		const oldDvName = "old-version-dv"
 
 		configureStorageProfileResultingFormat(format)
 
@@ -478,9 +479,33 @@ var _ = Describe("DataImportCron", func() {
 
 		switch format {
 		case cdiv1.DataImportCronSourceFormatPvc:
+			By(fmt.Sprintf("Create labeled DataVolume %s for old DVs garbage collection test", oldDvName))
+			dv := utils.NewDataVolumeWithRegistryImport(oldDvName, "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")
+			f.ForceBindPvcIfDvIsWaitForFirstConsumer(dv)
+			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())
+
 			pvcList, err := f.K8sClient.CoreV1().PersistentVolumeClaims(ns).List(context.TODO(), metav1.ListOptions{})
 			Expect(err).ToNot(HaveOccurred())
-			Expect(pvcList.Items).To(HaveLen(garbageSources))
+			Expect(pvcList.Items).To(HaveLen(garbageSources + 1))
 		case cdiv1.DataImportCronSourceFormatSnapshot:
 			snapshots := &snapshotv1.VolumeSnapshotList{}
 			err := f.CrClient.List(context.TODO(), snapshots, &client.ListOptions{Namespace: ns})
@@ -506,6 +531,12 @@ var _ = Describe("DataImportCron", func() {
 		By("Check garbage collection")
 		switch format {
 		case cdiv1.DataImportCronSourceFormatPvc:
+			By("Check old DV garbage collection")
+			Eventually(func() error {
+				_, err := f.CdiClient.CdiV1beta1().DataVolumes(ns).Get(context.TODO(), oldDvName, metav1.GetOptions{})
+				return err
+			}, dataImportCronTimeout, pollingInterval).Should(Satisfy(errors.IsNotFound), "Garbage collection failed cleaning old DV")
+
 			pvcList := &corev1.PersistentVolumeClaimList{}
 			Eventually(func() int {
 				pvcList, err = f.K8sClient.CoreV1().PersistentVolumeClaims(ns).List(context.TODO(), metav1.ListOptions{})
@@ -551,7 +582,7 @@ var _ = Describe("DataImportCron", func() {
 			Expect(found).To(BeTrue())
 		}
 	},
-		Entry("[test_id:7406] with PVC sources", cdiv1.DataImportCronSourceFormatPvc),
+		Entry("[test_id:7406] with PVC & DV sources", cdiv1.DataImportCronSourceFormatPvc),
 		Entry("[test_id:10033] with snapshot sources", cdiv1.DataImportCronSourceFormatSnapshot),
 	)