[release-v1.57] Fix broken local -> rook-ceph-block clone

pkg/controller/clone/common.go

@@ -176,30 +176,50 @@ func GetDriverFromVolume(ctx context.Context, c client.Client, pvc *corev1.Persi
 	return &pv.Spec.CSI.Driver, nil
 }
 
-// GetCompatibleVolumeSnapshotClass returns a VolumeSnapshotClass name that works for all PVCs
-func GetCompatibleVolumeSnapshotClass(ctx context.Context, c client.Client, pvcs ...*corev1.PersistentVolumeClaim) (*string, error) {
-	var drivers []string
+// GetCommonDriver returns the name of the CSI driver shared by all PVCs
+func GetCommonDriver(ctx context.Context, c client.Client, pvcs ...*corev1.PersistentVolumeClaim) (*string, error) {
+	var result *string
+
 	for _, pvc := range pvcs {
 		driver, err := GetDriverFromVolume(ctx, c, pvc)
 		if err != nil {
 			return nil, err
 		}
 
-		if driver != nil {
-			drivers = append(drivers, *driver)
-			continue
+		if driver == nil {
+			sc, err := GetStorageClassForClaim(ctx, c, pvc)
+			if err != nil {
+				return nil, err
+			}
+
+			if sc == nil {
+				return nil, nil
+			}
+
+			driver = &sc.Provisioner
 		}
 
-		sc, err := GetStorageClassForClaim(ctx, c, pvc)
-		if err != nil {
-			return nil, err
+		if result == nil {
+			result = driver
 		}
 
-		if sc == nil {
+		if *result != *driver {
 			return nil, nil
 		}
+	}
 
-		drivers = append(drivers, sc.Provisioner)
+	return result, nil
+}
+
+// GetCompatibleVolumeSnapshotClass returns a VolumeSnapshotClass name that works for all PVCs
+func GetCompatibleVolumeSnapshotClass(ctx context.Context, c client.Client, pvcs ...*corev1.PersistentVolumeClaim) (*string, error) {
+	driver, err := GetCommonDriver(ctx, c, pvcs...)
+	if err != nil {
+		return nil, err
+	}
+
+	if driver == nil {
+		return nil, nil
 	}
 
 	volumeSnapshotClasses := &snapshotv1.VolumeSnapshotClassList{}
@@ -212,14 +232,7 @@ func GetCompatibleVolumeSnapshotClass(ctx context.Context, c client.Client, pvcs
 
 	var candidates []string
 	for _, vcs := range volumeSnapshotClasses.Items {
-		matches := true
-		for _, driver := range drivers {
-			if driver != vcs.Driver {
-				matches = false
-				break
-			}
-		}
-		if matches {
+		if *driver == vcs.Driver {
 			candidates = append(candidates, vcs.Name)
 		}
 	}

pkg/controller/clone/planner.go

@@ -62,6 +62,12 @@ const (
 
 	// MessageNoProvisionerMatch reports that the storageclass provisioner does not match the volumesnapshotcontent driver (message)
 	MessageNoProvisionerMatch = "The storageclass provisioner does not match the volumesnapshotcontent driver"
+
+	// IncompatibleProvisioners reports that the provisioners are incompatible (reason)
+	IncompatibleProvisioners = "IncompatibleProvisioners"
+
+	// MessageIncompatibleProvisioners reports that the provisioners are incompatible (message)
+	MessageIncompatibleProvisioners = "Provisioners are incompatible"
 )
 
 // Planner plans clone operations
@@ -447,6 +453,17 @@ func (p *Planner) validateSourcePVC(args *ChooseStrategyArgs, sourceClaim *corev
 }
 
 func (p *Planner) validateAdvancedClonePVC(ctx context.Context, args *ChooseStrategyArgs, res *ChooseStrategyResult, sourceClaim *corev1.PersistentVolumeClaim) error {
+	driver, err := GetCommonDriver(ctx, p.Client, sourceClaim, args.TargetClaim)
+	if err != nil {
+		return err
+	}
+
+	if driver == nil {
+		p.fallbackToHostAssisted(args.TargetClaim, res, IncompatibleProvisioners, MessageIncompatibleProvisioners)
+		args.Log.V(3).Info("CSIDrivers not compatible for advanced clone")
+		return nil
+	}
+
 	if !SameVolumeMode(sourceClaim, args.TargetClaim) {
 		p.fallbackToHostAssisted(args.TargetClaim, res, IncompatibleVolumeModes, MessageIncompatibleVolumeModes)
 		args.Log.V(3).Info("volume modes not compatible for advanced clone")

pkg/controller/clone/planner_test.go

@@ -484,6 +484,35 @@ var _ = Describe("Planner test", func() {
 				Expect(csr).ToNot(BeNil())
 				Expect(csr.Strategy).To(Equal(cdiv1.CloneStrategyCsiClone))
 			})
+
+			It("should return host assisted if csi-clone in storage profile is set but source is different driver", func() {
+				cs := cdiv1.CloneStrategyCsiClone
+				args := &ChooseStrategyArgs{
+					TargetClaim: createTargetClaim(),
+					DataSource:  createPVCDataSource(),
+					Log:         log,
+				}
+				sp := &cdiv1.StorageProfile{
+					ObjectMeta: metav1.ObjectMeta{
+						Name: storageClassName,
+					},
+					Status: cdiv1.StorageProfileStatus{
+						CloneStrategy: &cs,
+					},
+				}
+				sourceClaim := createSourceClaim()
+				sourceClaim.Spec.StorageClassName = pointer.String("foo")
+				sourceVolume := createSourceVolume()
+				sourceVolume.Spec.StorageClassName = "foo"
+				sourceVolume.Spec.PersistentVolumeSource.CSI.Driver = "baz"
+				planner := createPlanner(sp, createStorageClass(), sourceClaim, sourceVolume)
+				csr, err := planner.ChooseStrategy(context.Background(), args)
+				Expect(err).ToNot(HaveOccurred())
+				Expect(csr).ToNot(BeNil())
+				Expect(csr.Strategy).To(Equal(cdiv1.CloneStrategyHostAssisted))
+				Expect(csr.FallbackReason).ToNot(BeNil())
+				Expect(*csr.FallbackReason).To(Equal(MessageIncompatibleProvisioners))
+			})
 		})
 
 		Context("Snapshot source", func() {