Allow StorageProfile to use a specific VolumeSnapshotClass

Signed-off-by: Arnon Gilboa <agilboa@redhat.com>

pkg/controller/clone/common.go

@@ -3,14 +3,11 @@ package clone
 import (
 	"context"
 	"fmt"
-	"sort"
 
 	"github.com/go-logr/logr"
-	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/tools/record"
@@ -222,27 +219,8 @@ func GetCompatibleVolumeSnapshotClass(ctx context.Context, c client.Client, pvcs
 		return nil, nil
 	}
 
-	volumeSnapshotClasses := &snapshotv1.VolumeSnapshotClassList{}
-	if err := c.List(ctx, volumeSnapshotClasses); err != nil {
-		if meta.IsNoMatchError(err) {
-			return nil, nil
-		}
-		return nil, err
-	}
-
-	var candidates []string
-	for _, vcs := range volumeSnapshotClasses.Items {
-		if *driver == vcs.Driver {
-			candidates = append(candidates, vcs.Name)
-		}
-	}
-
-	if len(candidates) > 0 {
-		sort.Strings(candidates)
-		return &candidates[0], nil
-	}
-
-	return nil, nil
+	//FIXME: may pass snapshotClassName if common to all pvcs?
+	return cc.GetVolumeSnapshotClass(context.TODO(), c, *driver, nil)
 }
 
 // SameVolumeMode returns true if all pvcs have the same volume mode

pkg/controller/common/util.go

@@ -27,6 +27,7 @@ import (
 	"net/http"
 	"reflect"
 	"regexp"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
@@ -206,11 +207,13 @@ const (
 	// AnnPopulatorKind annotation is added to a PVC' to specify the population kind, so it's later
 	// checked by the common populator watches.
 	AnnPopulatorKind = AnnAPIGroup + "/storage.populator.kind"
-	//AnnUsePopulator annotation indicates if the datavolume population will use populators
+	// AnnUsePopulator annotation indicates if the datavolume population will use populators
 	AnnUsePopulator = AnnAPIGroup + "/storage.usePopulator"
 
-	//AnnDefaultStorageClass is the annotation indicating that a storage class is the default one.
+	// AnnDefaultStorageClass is the annotation indicating that a storage class is the default one
 	AnnDefaultStorageClass = "storageclass.kubernetes.io/is-default-class"
+	// AnnDefaultSnapshotClass is the annotation indicating that a snapshot class is the default one
+	AnnDefaultSnapshotClass = "snapshot.storage.kubernetes.io/is-default-class"
 
 	// AnnOpenShiftImageLookup is the annotation for OpenShift image stream lookup
 	AnnOpenShiftImageLookup = "alpha.image.policy.openshift.io/resolve-names"
@@ -1803,7 +1806,7 @@ func ValidateSnapshotCloneProvisioners(ctx context.Context, c client.Client, sna
 }
 
 // GetSnapshotClassForSmartClone looks up the snapshot class based on the storage class
-func GetSnapshotClassForSmartClone(dvName string, targetPvcStorageClassName *string, log logr.Logger, client client.Client) (string, error) {
+func GetSnapshotClassForSmartClone(dvName string, targetPvcStorageClassName, snapshotClassName *string, log logr.Logger, client client.Client) (string, error) {
 	logger := log.WithName("GetSnapshotClassForSmartClone").V(3)
 	// Check if relevant CRDs are available
 	if !isCsiCrdsDeployed(client, log) {
@@ -1820,25 +1823,59 @@ func GetSnapshotClassForSmartClone(dvName string, targetPvcStorageClassName *str
 		return "", nil
 	}
 
-	// List the snapshot classes
-	scs := &snapshotv1.VolumeSnapshotClassList{}
-	if err := client.List(context.TODO(), scs); err != nil {
-		logger.Info("Cannot list snapshot classes, falling back to host assisted clone")
+	vscName, err := GetVolumeSnapshotClass(context.TODO(), client, targetStorageClass.Provisioner, snapshotClassName)
+	if err != nil {
 		return "", err
 	}
-	for _, snapshotClass := range scs.Items {
-		// Validate association between snapshot class and storage class
-		if snapshotClass.Driver == targetStorageClass.Provisioner {
-			logger.Info("smart-clone is applicable for datavolume", "datavolume",
-				dvName, "snapshot class", snapshotClass.Name)
-			return snapshotClass.Name, nil
-		}
+	if vscName != nil {
+		logger.Info("smart-clone is applicable for datavolume", "datavolume",
+			dvName, "snapshot class", *vscName)
+		return *vscName, nil
 	}
 
 	logger.Info("Could not match snapshotter with storage class, falling back to host assisted clone")
 	return "", nil
 }
 
+// GetVolumeSnapshotClass looks up the snapshot class based on the driver and an optional specific name
+func GetVolumeSnapshotClass(ctx context.Context, c client.Client, driver string, snapshotClassName *string) (*string, error) {
+	if snapshotClassName != nil {
+		vsc := &snapshotv1.VolumeSnapshotClass{}
+		if err := c.Get(context.TODO(), types.NamespacedName{Name: *snapshotClassName}, vsc); err != nil {
+			return nil, err
+		}
+		if vsc.Driver == driver {
+			return &vsc.Name, nil
+		}
+		return nil, nil
+	}
+
+	vscList := &snapshotv1.VolumeSnapshotClassList{}
+	if err := c.List(ctx, vscList); err != nil {
+		if meta.IsNoMatchError(err) {
+			return nil, nil
+		}
+		return nil, err
+	}
+
+	var candidates []string
+	for _, vsc := range vscList.Items {
+		if vsc.Driver == driver {
+			if vsc.Annotations[AnnDefaultSnapshotClass] == "true" {
+				return &vsc.Name, nil
+			}
+			candidates = append(candidates, vsc.Name)
+		}
+	}
+
+	if len(candidates) > 0 {
+		sort.Strings(candidates)
+		return &candidates[0], nil
+	}
+
+	return nil, nil
+}
+
 // isCsiCrdsDeployed checks whether the CSI snapshotter CRD are deployed
 func isCsiCrdsDeployed(c client.Client, log logr.Logger) bool {
 	version := "v1"

pkg/controller/dataimportcron-controller.go

@@ -326,7 +326,7 @@ func (r *DataImportCronReconciler) update(ctx context.Context, dataImportCron *c
 	if desiredStorageClass != nil {
 		cc.AddAnnotation(dataImportCron, AnnStorageClass, desiredStorageClass.Name)
 	}
-	format, err := r.getSourceFormat(ctx, dataImportCron, desiredStorageClass)
+	format, err := r.getSourceFormat(ctx, desiredStorageClass)
 	if err != nil {
 		return res, err
 	}
@@ -688,8 +688,11 @@ func (r *DataImportCronReconciler) handleSnapshot(ctx context.Context, dataImpor
 		r.log.Info("Attempt to change storage class, will not try making a snapshot of the old PVC")
 		return nil
 	}
-
-	className, err := cc.GetSnapshotClassForSmartClone(pvc.Name, &desiredStorageClass.Name, r.log, r.client)
+	storageProfile := &cdiv1.StorageProfile{}
+	if err := r.client.Get(ctx, types.NamespacedName{Name: desiredStorageClass.Name}, storageProfile); err != nil {
+		return err
+	}
+	className, err := cc.GetSnapshotClassForSmartClone(pvc.Name, &desiredStorageClass.Name, storageProfile.Status.SnapshotClass, r.log, r.client)
 	if err != nil {
 		return err
 	}
@@ -713,7 +716,7 @@ func (r *DataImportCronReconciler) handleSnapshot(ctx context.Context, dataImpor
 	r.setDataImportCronResourceLabels(dataImportCron, desiredSnapshot)
 
 	currentSnapshot := &snapshotv1.VolumeSnapshot{}
-	if err := r.client.Get(context.TODO(), client.ObjectKeyFromObject(desiredSnapshot), currentSnapshot); err != nil {
+	if err := r.client.Get(ctx, client.ObjectKeyFromObject(desiredSnapshot), currentSnapshot); err != nil {
 		if !k8serrors.IsNotFound(err) {
 			return err
 		}
@@ -757,14 +760,14 @@ func (r *DataImportCronReconciler) updateDataImportCronSuccessCondition(ctx cont
 	return nil
 }
 
-func (r *DataImportCronReconciler) getSourceFormat(ctx context.Context, dataImportCron *cdiv1.DataImportCron, desiredStorageClass *storagev1.StorageClass) (cdiv1.DataImportCronSourceFormat, error) {
+func (r *DataImportCronReconciler) getSourceFormat(ctx context.Context, desiredStorageClass *storagev1.StorageClass) (cdiv1.DataImportCronSourceFormat, error) {
 	format := cdiv1.DataImportCronSourceFormatPvc
 	if desiredStorageClass == nil {
 		return format, nil
 	}
 
 	storageProfile := &cdiv1.StorageProfile{}
-	if err := r.client.Get(context.TODO(), types.NamespacedName{Name: desiredStorageClass.Name}, storageProfile); err != nil {
+	if err := r.client.Get(ctx, types.NamespacedName{Name: desiredStorageClass.Name}, storageProfile); err != nil {
 		return format, err
 	}
 	if storageProfile.Status.DataImportCronSourceFormat != nil {

pkg/controller/storageprofile-controller.go

@@ -86,10 +86,13 @@ func (r *StorageProfileReconciler) reconcileStorageProfile(sc *storagev1.Storage
 
 	storageProfile.Status.StorageClass = &sc.Name
 	storageProfile.Status.Provisioner = &sc.Provisioner
-	snapClass, err := cc.GetSnapshotClassForSmartClone("", &sc.Name, r.log, r.client)
+	snapClass, err := cc.GetSnapshotClassForSmartClone("", &sc.Name, storageProfile.Spec.SnapshotClass, r.log, r.client)
 	if err != nil {
 		return reconcile.Result{}, err
 	}
+	if snapClass != "" {
+		storageProfile.Status.SnapshotClass = &snapClass
+	}
 	storageProfile.Status.CloneStrategy = r.reconcileCloneStrategy(sc, storageProfile.Spec.CloneStrategy, snapClass)
 	storageProfile.Status.DataImportCronSourceFormat = r.reconcileDataImportCronSourceFormat(sc, storageProfile.Spec.DataImportCronSourceFormat, snapClass)
 

pkg/controller/storageprofile-controller_test.go

@@ -37,16 +37,22 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/utils/pointer"
 
 	cdiv1 "kubevirt.io/containerized-data-importer-api/pkg/apis/core/v1beta1"
 	"kubevirt.io/containerized-data-importer/pkg/common"
 	. "kubevirt.io/containerized-data-importer/pkg/controller/common"
 	"kubevirt.io/containerized-data-importer/pkg/storagecapabilities"
 )
 
+const (
+	storageClassName  = "testSC"
+	snapshotClassName = "testSnapClass"
+	cephProvisioner   = "rook-ceph.rbd.csi.ceph.com"
+)
+
 var (
 	storageProfileLog = logf.Log.WithName("storageprofile-controller-test")
-	storageClassName  = "testSC"
 	lsoLabels         = map[string]string{
 		"local.storage.openshift.io/owner-name":      "local",
 		"local.storage.openshift.io/owner-namespace": "openshift-local-storage",
@@ -313,11 +319,91 @@ var _ = Describe("Storage profile controller reconcile loop", func() {
 		Entry("Clone", cdiv1.CloneStrategyCsiClone),
 	)
 
+	It("Should succeed when updating storage profile with specific SnapshotClass", func() {
+		storageClass := CreateStorageClassWithProvisioner(storageClassName, nil, nil, cephProvisioner)
+		reconciler := createStorageProfileReconciler(storageClass, createVolumeSnapshotContentCrd(), createVolumeSnapshotClassCrd(), createVolumeSnapshotCrd())
+
+		for i := 0; i < 3; i++ {
+			snapClass := createSnapshotClass(fmt.Sprintf("snapclass-%d", i), nil, cephProvisioner)
+			err := reconciler.client.Create(context.TODO(), snapClass)
+			Expect(err).ToNot(HaveOccurred())
+		}
+
+		_, err := reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).ToNot(HaveOccurred())
+
+		sp := &cdiv1.StorageProfile{}
+		err = reconciler.client.Get(context.TODO(), types.NamespacedName{Name: storageClassName}, sp, &client.GetOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		snapName := "snapclass-1"
+		sp.Spec.SnapshotClass = &snapName
+		err = reconciler.client.Update(context.TODO(), sp, &client.UpdateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).ToNot(HaveOccurred())
+
+		err = reconciler.client.Get(context.TODO(), types.NamespacedName{Name: storageClassName}, sp, &client.GetOptions{})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(sp.Status.SnapshotClass).ToNot(BeNil())
+		Expect(*sp.Status.SnapshotClass).To(Equal(snapName))
+	})
+
+	It("Should error when updating storage profile with non-existing SnapshotClass", func() {
+		storageClass := CreateStorageClassWithProvisioner(storageClassName, nil, nil, cephProvisioner)
+		reconciler := createStorageProfileReconciler(storageClass, createVolumeSnapshotContentCrd(), createVolumeSnapshotClassCrd(), createVolumeSnapshotCrd())
+
+		snapClass := createSnapshotClass(snapshotClassName, nil, cephProvisioner)
+		err := reconciler.client.Create(context.TODO(), snapClass)
+		Expect(err).ToNot(HaveOccurred())
+
+		_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).ToNot(HaveOccurred())
+
+		sp := &cdiv1.StorageProfile{}
+		err = reconciler.client.Get(context.TODO(), types.NamespacedName{Name: storageClassName}, sp, &client.GetOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		noSuchSnapshotClassName := "no-such-snapshotclass"
+		sp.Spec.SnapshotClass = &noSuchSnapshotClassName
+		err = reconciler.client.Update(context.TODO(), sp, &client.UpdateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).To(HaveOccurred())
+		Expect(err.Error()).To(ContainSubstring(fmt.Sprintf("\"%s\" not found", noSuchSnapshotClassName)))
+	})
+
+	It("Should error when updating storage profile with existing SnapshotClass with mismatching driver", func() {
+		storageClass := CreateStorageClassWithProvisioner(storageClassName, nil, nil, cephProvisioner)
+		reconciler := createStorageProfileReconciler(storageClass, createVolumeSnapshotContentCrd(), createVolumeSnapshotClassCrd(), createVolumeSnapshotCrd())
+
+		snapClass := createSnapshotClass(snapshotClassName, nil, "no-such-provisoner")
+		err := reconciler.client.Create(context.TODO(), snapClass)
+		Expect(err).ToNot(HaveOccurred())
+
+		_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).ToNot(HaveOccurred())
+
+		sp := &cdiv1.StorageProfile{}
+		err = reconciler.client.Get(context.TODO(), types.NamespacedName{Name: storageClassName}, sp, &client.GetOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		sp.Spec.SnapshotClass = pointer.String(snapshotClassName)
+		err = reconciler.client.Update(context.TODO(), sp, &client.UpdateOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: types.NamespacedName{Name: storageClassName}})
+		Expect(err).ToNot(HaveOccurred())
+		Expect(sp.Status.SnapshotClass).To(BeNil())
+	})
+
 	DescribeTable("should set advised source format for dataimportcrons", func(provisioner string, expectedFormat cdiv1.DataImportCronSourceFormat, deploySnapClass bool) {
 		storageClass := CreateStorageClassWithProvisioner(storageClassName, map[string]string{AnnDefaultStorageClass: "true"}, map[string]string{}, provisioner)
 		reconciler := createStorageProfileReconciler(storageClass, createVolumeSnapshotContentCrd(), createVolumeSnapshotClassCrd(), createVolumeSnapshotCrd())
 		if deploySnapClass {
-			snapClass := createSnapshotClass(storageClassName+"-snapclass", nil, provisioner)
+			snapClass := createSnapshotClass(snapshotClassName, nil, provisioner)
 			err := reconciler.client.Create(context.TODO(), snapClass)
 			Expect(err).ToNot(HaveOccurred())
 		}
@@ -343,7 +429,7 @@ var _ = Describe("Storage profile controller reconcile loop", func() {
 		storageClass := CreateStorageClassWithProvisioner(storageClassName, map[string]string{AnnDefaultStorageClass: "true"}, map[string]string{}, provisioner)
 		reconciler := createStorageProfileReconciler(storageClass, createVolumeSnapshotContentCrd(), createVolumeSnapshotClassCrd(), createVolumeSnapshotCrd())
 		if deploySnapClass {
-			snapClass := createSnapshotClass(storageClassName+"-snapclass", nil, provisioner)
+			snapClass := createSnapshotClass(snapshotClassName, nil, provisioner)
 			err := reconciler.client.Create(context.TODO(), snapClass)
 			Expect(err).ToNot(HaveOccurred())
 		}

pkg/operator/resources/crds_generated.go

@@ -6907,6 +6907,11 @@ spec:
                 description: DataImportCronSourceFormat defines the format of the
                   DataImportCron-created disk image sources
                 type: string
+              snapshotClass:
+                description: SnapshotClass is optional specific VolumeSnapshotClass
+                  for CloneStrategySnapshot. If not set, a VolumeSnapshotClass is
+                  chosen according to the provisioner.
+                type: string
             type: object
           status:
             description: StorageProfileStatus provides the most recently observed
@@ -6943,6 +6948,11 @@ spec:
               provisioner:
                 description: The Storage class provisioner plugin name
                 type: string
+              snapshotClass:
+                description: SnapshotClass is optional specific VolumeSnapshotClass
+                  for CloneStrategySnapshot. If not set, a VolumeSnapshotClass is
+                  chosen according to the provisioner.
+                type: string
               storageClass:
                 description: The StorageClass name for which capabilities are defined
                 type: string