Support storageProfile minimumSupportedPVCSize in clone

pkg/controller/clone/planner.go

@@ -460,8 +460,13 @@ func (p *Planner) validateSourcePVC(args *ChooseStrategyArgs, sourceClaim *corev
 		args.Log.V(3).Info("permissive clone annotation found, skipping size validation")
 		return nil
 	}
+	targetResources, err := cc.GetEffectiveStorageResources(context.TODO(), p.Client, args.TargetClaim.Spec.Resources,
+		args.TargetClaim.Spec.StorageClassName, cc.GetPVCContentType(args.TargetClaim), args.Log)
+	if err != nil {
+		return err
+	}
 
-	if err := cc.ValidateRequestedCloneSize(sourceClaim.Spec.Resources, args.TargetClaim.Spec.Resources); err != nil {
+	if err := cc.ValidateRequestedCloneSize(sourceClaim.Spec.Resources, targetResources); err != nil {
 		p.Recorder.Eventf(args.TargetClaim, corev1.EventTypeWarning, cc.ErrIncompatiblePVC, err.Error())
 		return err
 	}

pkg/controller/common/util.go

@@ -1099,6 +1099,51 @@ func AddImportVolumeMounts() []corev1.VolumeMount {
 	return volumeMounts
 }
 
+// GetEffectiveStorageResources returns the maximum of the passed storageResources and the storageProfile minimumSupportedPVCSize.
+// If the passed storageResources has no size, it is returned as-is.
+func GetEffectiveStorageResources(ctx context.Context, client client.Client, storageResources corev1.VolumeResourceRequirements,
+	storageClassName *string, contentType cdiv1.DataVolumeContentType, log logr.Logger) (corev1.VolumeResourceRequirements, error) {
+	sc, err := GetStorageClassByNameWithVirtFallback(ctx, client, storageClassName, contentType)
+	if err != nil || sc == nil {
+		return storageResources, err
+	}
+
+	requestedSize, hasSize := storageResources.Requests[corev1.ResourceStorage]
+	if !hasSize {
+		return storageResources, nil
+	}
+
+	if requestedSize, err = GetEffectiveVolumeSize(ctx, client, requestedSize, sc.Name, &log); err != nil {
+		return storageResources, err
+	}
+
+	return corev1.VolumeResourceRequirements{
+		Requests: corev1.ResourceList{
+			corev1.ResourceStorage: requestedSize,
+		},
+	}, nil
+}
+
+// GetEffectiveVolumeSize returns the maximum of the passed requestedSize and the storageProfile minimumSupportedPVCSize.
+func GetEffectiveVolumeSize(ctx context.Context, client client.Client, requestedSize resource.Quantity, storageClassName string, log *logr.Logger) (resource.Quantity, error) {
+	storageProfile := &cdiv1.StorageProfile{}
+	if err := client.Get(ctx, types.NamespacedName{Name: storageClassName}, storageProfile); err != nil {
+		return requestedSize, IgnoreNotFound(err)
+	}
+
+	if val, exists := storageProfile.Annotations[AnnMinimumSupportedPVCSize]; exists {
+		if minSize, err := resource.ParseQuantity(val); err == nil {
+			if requestedSize.Cmp(minSize) == -1 {
+				return minSize, nil
+			}
+		} else if log != nil {
+			log.V(1).Info("Invalid minimum PVC size in annotation", "value", val, "error", err)
+		}
+	}
+
+	return requestedSize, nil
+}
+
 // ValidateRequestedCloneSize validates the clone size requirements on block
 func ValidateRequestedCloneSize(sourceResources, targetResources corev1.VolumeResourceRequirements) error {
 	sourceRequest, hasSource := sourceResources.Requests[corev1.ResourceStorage]

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

@@ -445,7 +445,7 @@ func (r *PvcCloneReconciler) validateCloneAndSourcePVC(syncState *dvSyncState, l
 		return false, err
 	}
 
-	err = validateClone(sourcePvc, &datavolume.Spec)
+	err = r.validateClone(sourcePvc, &datavolume.Spec)
 	if err != nil {
 		syncErr := r.syncDataVolumeStatusPhaseWithEvent(syncState, datavolume.Status.Phase, nil,
 			Event{
@@ -463,31 +463,30 @@ func (r *PvcCloneReconciler) validateCloneAndSourcePVC(syncState *dvSyncState, l
 }
 
 // validateClone compares a clone spec against its source PVC to validate its creation
-func validateClone(sourcePVC *corev1.PersistentVolumeClaim, spec *cdiv1.DataVolumeSpec) error {
+func (r *PvcCloneReconciler) validateClone(sourcePVC *corev1.PersistentVolumeClaim, spec *cdiv1.DataVolumeSpec) error {
 	var targetResources corev1.VolumeResourceRequirements
+	var err error
 
 	valid, sourceContentType, targetContentType := validateContentTypes(sourcePVC, spec)
 	if !valid {
 		msg := fmt.Sprintf("Source contentType (%s) and target contentType (%s) do not match", sourceContentType, targetContentType)
 		return errors.New(msg)
 	}
 
-	isSizelessClone := false
+	hasSize := true
 	explicitPvcRequest := spec.PVC != nil
 	if explicitPvcRequest {
 		targetResources = spec.PVC.Resources
-	} else {
-		targetResources = spec.Storage.Resources
-		// The storage size in the target DV can be empty
-		// when cloning using the 'Storage' API
-		if _, ok := targetResources.Requests[corev1.ResourceStorage]; !ok {
-			isSizelessClone = true
+	} else if _, hasSize = spec.Storage.Resources.Requests[corev1.ResourceStorage]; hasSize {
+		targetResources, err = cc.GetEffectiveStorageResources(context.TODO(), r.client, spec.Storage.Resources, spec.Storage.StorageClassName, spec.ContentType, r.log)
+		if err != nil {
+			return err
 		}
 	}
 
 	// TODO: Spec.Storage API needs a better more complex check to validate clone size - to account for fsOverhead
 	// simple size comparison will not work here
-	if (!isSizelessClone && cc.GetVolumeMode(sourcePVC) == corev1.PersistentVolumeBlock) || explicitPvcRequest {
+	if (hasSize && cc.GetVolumeMode(sourcePVC) == corev1.PersistentVolumeBlock) || explicitPvcRequest {
 		if err := cc.ValidateRequestedCloneSize(sourcePVC.Spec.Resources, targetResources); err != nil {
 			return err
 		}

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

@@ -493,12 +493,13 @@ var _ = Describe("All DataVolume Tests", func() {
 		sourcePvc := CreatePvc("testPVC", "default", map[string]string{}, nil)
 		blockVM := corev1.PersistentVolumeBlock
 		fsVM := corev1.PersistentVolumeFilesystem
+		r := createCloneReconciler()
 
 		It("Should reject the clone if source and target have different content types", func() {
 			sourcePvc.Annotations[AnnContentType] = string(cdiv1.DataVolumeKubeVirt)
 			dvSpec := &cdiv1.DataVolumeSpec{ContentType: cdiv1.DataVolumeArchive}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring(
 				fmt.Sprintf("Source contentType (%s) and target contentType (%s) do not match", cdiv1.DataVolumeKubeVirt, cdiv1.DataVolumeArchive)))
@@ -516,7 +517,7 @@ var _ = Describe("All DataVolume Tests", func() {
 			}
 			dvSpec := &cdiv1.DataVolumeSpec{Storage: storageSpec}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("target resources requests storage size is smaller than the source"))
 		})
@@ -533,7 +534,7 @@ var _ = Describe("All DataVolume Tests", func() {
 			}
 			dvSpec := &cdiv1.DataVolumeSpec{Storage: storageSpec}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).ToNot(HaveOccurred())
 		})
 
@@ -543,7 +544,7 @@ var _ = Describe("All DataVolume Tests", func() {
 			storageSpec := &cdiv1.StorageSpec{}
 			dvSpec := &cdiv1.DataVolumeSpec{Storage: storageSpec}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).ToNot(HaveOccurred())
 		})
 
@@ -558,7 +559,7 @@ var _ = Describe("All DataVolume Tests", func() {
 			}
 			dvSpec := &cdiv1.DataVolumeSpec{PVC: pvcSpec}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).To(HaveOccurred())
 			Expect(err.Error()).To(ContainSubstring("target resources requests storage size is smaller than the source"))
 
@@ -575,7 +576,7 @@ var _ = Describe("All DataVolume Tests", func() {
 			}
 			dvSpec := &cdiv1.DataVolumeSpec{PVC: pvcSpec}
 
-			err := validateClone(sourcePvc, dvSpec)
+			err := r.validateClone(sourcePvc, dvSpec)
 			Expect(err).ToNot(HaveOccurred())
 		})
 	})

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

@@ -340,7 +340,7 @@ func (r *SnapshotCloneReconciler) validateCloneAndSourceSnapshot(syncState *dvSy
 	}
 	syncState.snapshot = snapshot
 
-	err = validateSnapshotClone(snapshot, &datavolume.Spec)
+	err = r.validateSnapshotClone(snapshot, &datavolume.Spec)
 	if err != nil {
 		syncEventErr := r.syncDataVolumeStatusPhaseWithEvent(syncState, datavolume.Status.Phase, nil,
 			Event{
@@ -358,7 +358,7 @@ func (r *SnapshotCloneReconciler) validateCloneAndSourceSnapshot(syncState *dvSy
 }
 
 // validateSnapshotClone compares a snapshot clone spec against its source snapshot to validate its creation
-func validateSnapshotClone(sourceSnapshot *snapshotv1.VolumeSnapshot, spec *cdiv1.DataVolumeSpec) error {
+func (r *SnapshotCloneReconciler) validateSnapshotClone(sourceSnapshot *snapshotv1.VolumeSnapshot, spec *cdiv1.DataVolumeSpec) error {
 	err := cc.IsSnapshotValidForClone(sourceSnapshot)
 	if err != nil {
 		return err
@@ -375,23 +375,23 @@ func validateSnapshotClone(sourceSnapshot *snapshotv1.VolumeSnapshot, spec *cdiv
 		sourceResources.Requests = corev1.ResourceList{corev1.ResourceStorage: *size}
 	}
 
-	isSizelessClone := false
+	hasSize := true
 	explicitPvcRequest := spec.PVC != nil
 	if explicitPvcRequest {
 		targetResources = spec.PVC.Resources
-	} else {
-		targetResources = spec.Storage.Resources
-		if _, ok := targetResources.Requests["storage"]; !ok {
-			isSizelessClone = true
+	} else if _, hasSize = spec.Storage.Resources.Requests[corev1.ResourceStorage]; hasSize {
+		targetResources, err = cc.GetEffectiveStorageResources(context.TODO(), r.client, spec.Storage.Resources, spec.Storage.StorageClassName, spec.ContentType, r.log)
+		if err != nil {
+			return err
 		}
 	}
 
-	if !isSizelessClone && restoreSizeAvailable {
+	if hasSize && restoreSizeAvailable {
 		// Sizes available, make sure user picked something bigger than minimal
 		if err := cc.ValidateRequestedCloneSize(sourceResources, targetResources); err != nil {
 			return err
 		}
-	} else if isSizelessClone && !restoreSizeAvailable {
+	} else if !hasSize && !restoreSizeAvailable {
 		return fmt.Errorf("size not specified by user/provisioner, can't tell how much needed for restore")
 	}
 

pkg/controller/datavolume/snapshot-clone-controller_test.go

@@ -308,6 +308,7 @@ var _ = Describe("All DataVolume Tests", func() {
 
 				var _ = Describe("validateSnapshotClone", func() {
 					type snapshotModifierFunc func(*snapshotv1.VolumeSnapshot)
+					r := createSnapshotCloneReconciler()
 
 					DescribeTable("Validation mechanism rejects or accepts the clone depending snapshot state",
 						func(expectedErr string, snapshotModifier snapshotModifierFunc) {
@@ -316,7 +317,7 @@ var _ = Describe("All DataVolume Tests", func() {
 							snapshot := createSnapshotInVolumeSnapshotClass("test-snap", dv.Namespace, &expectedSnapshotClass, nil, nil, true)
 							snapshotModifier(snapshot)
 
-							err := validateSnapshotClone(snapshot, &dv.Spec)
+							err := r.validateSnapshotClone(snapshot, &dv.Spec)
 							if expectedErr != "" {
 								Expect(err).To(HaveOccurred())
 								Expect(err.Error()).To(ContainSubstring(expectedErr))
@@ -363,7 +364,7 @@ var _ = Describe("All DataVolume Tests", func() {
 							restoreSizeParsed := resource.MustParse(restoreSize)
 							snapshot.Status.RestoreSize = &restoreSizeParsed
 
-							err := validateSnapshotClone(snapshot, &dv.Spec)
+							err := r.validateSnapshotClone(snapshot, &dv.Spec)
 							if expectedErr != "" {
 								Expect(err).To(HaveOccurred())
 								Expect(err.Error()).To(ContainSubstring(expectedErr))

pkg/controller/datavolume/util.go

@@ -299,18 +299,7 @@ func renderPvcSpecVolumeSize(client client.Client, pvcSpec *v1.PersistentVolumeC
 	}
 
 	if scName := pvcSpec.StorageClassName; scName != nil {
-		storageProfile := &cdiv1.StorageProfile{}
-		if err := client.Get(context.TODO(), types.NamespacedName{Name: *scName}, storageProfile); err == nil {
-			if val, exists := storageProfile.Annotations[cc.AnnMinimumSupportedPVCSize]; exists {
-				if minSize, err := resource.ParseQuantity(val); err == nil {
-					if requestedSize.Cmp(minSize) == -1 {
-						requestedSize = minSize
-					}
-				} else if log != nil {
-					log.V(1).Info("Invalid minimum PVC size in annotation", "value", val, "error", err)
-				}
-			}
-		} else if !k8serrors.IsNotFound(err) {
+		if requestedSize, err = cc.GetEffectiveVolumeSize(context.TODO(), client, requestedSize, *scName, log); err != nil {
 			return err
 		}
 	}

tests/cloner_test.go

@@ -2215,6 +2215,69 @@ var _ = Describe("all clone tests", func() {
 
 		})
 
+		DescribeTable("Block volumeMode clone with target smaller than the source, using storgeProfile with minPvcSize annotation", func(minSize string, shouldSucceed bool) {
+			if !f.IsBlockVolumeStorageClassAvailable() {
+				Skip("Storage Class for block volume is not available")
+			}
+
+			By(fmt.Sprintf("Create source PVC %s", sourcePVCName))
+			sc := createStorageWithMinimumSupportedPVCSize(f, minSize)
+			sourcePvc = utils.NewBlockPVCDefinition(sourcePVCName, "1Gi", nil, nil, sc)
+			sourcePvc.Namespace = f.Namespace.Name
+			sourcePvc = f.CreateAndPopulateSourcePVC(sourcePvc, "fill-source-block-pod", blockFillCommand)
+
+			targetDvName := "small-target-dv"
+			By(fmt.Sprintf("Create small target DV %s", targetDvName))
+			targetDV := utils.NewDataVolumeForImageCloningAndStorageSpec(targetDvName, "512Mi", sourcePvc.Namespace, sourcePvc.Name, sourcePvc.Spec.StorageClassName, sourcePvc.Spec.VolumeMode)
+			targetDV, err := utils.CreateDataVolumeFromDefinition(f.CdiClient, sourcePvc.Namespace, targetDV)
+			Expect(err).ToNot(HaveOccurred())
+
+			if !shouldSucceed {
+				By("The clone should fail")
+				f.ExpectEvent(targetDV.Namespace).Should(ContainSubstring(dvc.CloneValidationFailed))
+				return
+			}
+
+			By("Wait for target DV succeeded")
+			f.ForceBindPvcIfDvIsWaitForFirstConsumer(targetDV)
+			err = utils.WaitForDataVolumePhase(f, targetDV.Namespace, cdiv1.Succeeded, targetDV.Name)
+			Expect(err).ToNot(HaveOccurred())
+		},
+			Entry("[test_id:XXXX] large enough", "1Gi", true),
+			Entry("[test_id:XXXX] too small", "256Mi", false),
+			Entry("[test_id:XXXX] empty", "", false),
+		)
+
+		DescribeTable("Filesystem volumeMode clone with target smaller than the source, using storgeProfile with minPvcSize annotation", func(minSize string, shouldSucceed bool) {
+			By(fmt.Sprintf("Create source PVC %s", sourcePVCName))
+			sc := createStorageWithMinimumSupportedPVCSize(f, minSize)
+			sourcePvc = utils.NewPVCDefinition(sourcePVCName, "1Gi", nil, nil)
+			sourcePvc.Namespace = f.Namespace.Name
+			sourcePvc.Spec.StorageClassName = &sc
+			sourcePvc = f.CreateAndPopulateSourcePVC(sourcePvc, sourcePodFillerName, fillCommand+testFile+"; chmod 660 "+testBaseDir+testFile)
+
+			targetDvName := "small-target-dv"
+			By(fmt.Sprintf("Create small target DV %s", targetDvName))
+			targetDV := utils.NewDataVolumeForImageCloningAndStorageSpec(targetDvName, "512Mi", sourcePvc.Namespace, sourcePvc.Name, sourcePvc.Spec.StorageClassName, sourcePvc.Spec.VolumeMode)
+			targetDV, err := utils.CreateDataVolumeFromDefinition(f.CdiClient, sourcePvc.Namespace, targetDV)
+			Expect(err).ToNot(HaveOccurred())
+			f.ForceBindPvcIfDvIsWaitForFirstConsumer(targetDV)
+
+			if !shouldSucceed {
+				By("The clone should fail")
+				f.ExpectEvent(targetDV.Namespace).Should(ContainSubstring(controller.ErrIncompatiblePVC))
+				return
+			}
+
+			By("Wait for target DV succeeded")
+			err = utils.WaitForDataVolumePhase(f, targetDV.Namespace, cdiv1.Succeeded, targetDV.Name)
+			Expect(err).ToNot(HaveOccurred())
+		},
+			Entry("[test_id:XXXX] large enough", "1Gi", true),
+			Entry("[test_id:XXXX] too small", "256Mi", false),
+			Entry("[test_id:XXXX] empty", "", false),
+		)
+
 		It("[test_id:4276] Clone datavolume with short name", Serial, func() {
 			shortDvName := "import-long-name-dv"
 
@@ -3058,3 +3121,21 @@ func validateCloneType(f *framework.Framework, dv *cdiv1.DataVolume) {
 
 	Expect(utils.GetCloneType(f.CdiClient, dv)).To(Equal(cloneType))
 }
+
+func createStorageWithMinimumSupportedPVCSize(f *framework.Framework, minSize string) string {
+	sc, err := f.CreateNonDefaultVariationOfStorageClass(utils.DefaultStorageClass,
+		func(sc *storagev1.StorageClass) { sc.UID = "" })
+	Expect(err).ToNot(HaveOccurred())
+
+	var sp *cdiv1.StorageProfile
+	Eventually(func() error {
+		sp, err = f.CdiClient.CdiV1beta1().StorageProfiles().Get(context.TODO(), sc.Name, metav1.GetOptions{})
+		return err
+	}, time.Minute, time.Second).Should(Succeed())
+
+	sp.Annotations = map[string]string{controller.AnnMinimumSupportedPVCSize: minSize}
+	_, err = f.CdiClient.CdiV1beta1().StorageProfiles().Update(context.TODO(), sp, metav1.UpdateOptions{})
+	Expect(err).ToNot(HaveOccurred())
+
+	return sc.Name
+}