Add volume policy tier to VolumeSnapshotClass selection

Add GetVolumeSnapshotClassFromVolumePolicy helper and extend
GetVolumeSnapshotClass with a policySnapshotClass parameter. The new
tier sits between PVC annotation and backup annotation in the priority
chain: PVC annotation > volume policy > backup annotation > VSC label.

Ref: #8807

Signed-off-by: Shubham Pampattiwar <spampatt@redhat.com>

pkg/util/csi/volume_snapshot.go

@@ -314,6 +314,7 @@ func GetVolumeSnapshotClass(
 	pvc *corev1api.PersistentVolumeClaim,
 	log logrus.FieldLogger,
 	crClient crclient.Client,
+	policySnapshotClass string,
 ) (*snapshotv1api.VolumeSnapshotClass, error) {
 	snapshotClasses := new(snapshotv1api.VolumeSnapshotClassList)
 	err := crClient.List(context.TODO(), snapshotClasses)
@@ -331,6 +332,16 @@ func GetVolumeSnapshotClass(
 		return snapshotClass, nil
 	}
 
+	// If a snapshot class is specified by volume policy, use that
+	snapshotClass, err = GetVolumeSnapshotClassFromVolumePolicy(
+		policySnapshotClass, provisioner, snapshotClasses)
+	if err != nil {
+		log.Debugf("Didn't find VolumeSnapshotClass from volume policy: %v", err)
+	}
+	if snapshotClass != nil {
+		return snapshotClass, nil
+	}
+
 	// If there is no annotation in PVC, attempt to fetch it from backup annotations
 	snapshotClass, err = GetVolumeSnapshotClassFromBackupAnnotationsForDriver(
 		backup, provisioner, snapshotClasses)
@@ -412,6 +423,34 @@ func GetVolumeSnapshotClassFromBackupAnnotationsForDriver(
 	)
 }
 
+// GetVolumeSnapshotClassFromVolumePolicy returns a VolumeSnapshotClass
+// specified by a volume policy's snapshotClass parameter. If
+// policySnapshotClass is empty, it returns nil (no match).
+func GetVolumeSnapshotClassFromVolumePolicy(
+	policySnapshotClass string,
+	provisioner string,
+	snapshotClasses *snapshotv1api.VolumeSnapshotClassList,
+) (*snapshotv1api.VolumeSnapshotClass, error) {
+	if policySnapshotClass == "" {
+		return nil, nil
+	}
+	for _, sc := range snapshotClasses.Items {
+		if strings.EqualFold(policySnapshotClass, sc.ObjectMeta.Name) {
+			if !strings.EqualFold(sc.Driver, provisioner) {
+				return nil, errors.Errorf(
+					"VolumeSnapshotClass %s specified by volume policy is not for driver %s",
+					sc.ObjectMeta.Name, provisioner,
+				)
+			}
+			return &sc, nil
+		}
+	}
+	return nil, errors.Errorf(
+		"No CSI VolumeSnapshotClass found with name %s specified by volume policy for driver %s",
+		policySnapshotClass, provisioner,
+	)
+}
+
 // GetVolumeSnapshotClassForStorageClass returns a VolumeSnapshotClass
 // for the supplied volume provisioner/ driver name.
 func GetVolumeSnapshotClassForStorageClass(

pkg/util/csi/volume_snapshot_test.go

@@ -1032,7 +1032,7 @@ func TestGetVolumeSnapshotClass(t *testing.T) {
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			actualSnapshotClass, actualError := GetVolumeSnapshotClass(
-				tc.driverName, tc.backup, tc.pvc, logrus.New(), fakeClient)
+				tc.driverName, tc.backup, tc.pvc, logrus.New(), fakeClient, "")
 			if tc.expectError {
 				require.Error(t, actualError)
 				assert.Nil(t, actualSnapshotClass)
@@ -1043,6 +1043,93 @@ func TestGetVolumeSnapshotClass(t *testing.T) {
 	}
 }
 
+func TestGetVolumeSnapshotClassFromVolumePolicy(t *testing.T) {
+	vscArray1 := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-array-1"},
+		Driver:     "infinibox-csi-driver",
+	}
+	vscArray2 := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-array-2"},
+		Driver:     "infinibox-csi-driver",
+	}
+	vscOther := &snapshotv1api.VolumeSnapshotClass{
+		ObjectMeta: metav1.ObjectMeta{Name: "vsc-other"},
+		Driver:     "other-csi-driver",
+	}
+
+	snapshotClasses := &snapshotv1api.VolumeSnapshotClassList{
+		Items: []snapshotv1api.VolumeSnapshotClass{*vscArray1, *vscArray2, *vscOther},
+	}
+
+	testCases := []struct {
+		name                string
+		policySnapshotClass string
+		provisioner         string
+		expectedVSC         *snapshotv1api.VolumeSnapshotClass
+		expectError         bool
+	}{
+		{
+			name:                "empty policy returns nil",
+			policySnapshotClass: "",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         nil,
+			expectError:         false,
+		},
+		{
+			name:                "matching VSC with correct driver",
+			policySnapshotClass: "vsc-array-1",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray1,
+			expectError:         false,
+		},
+		{
+			name:                "matching VSC with correct driver second array",
+			policySnapshotClass: "vsc-array-2",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray2,
+			expectError:         false,
+		},
+		{
+			name:                "VSC exists but wrong driver",
+			policySnapshotClass: "vsc-other",
+			provisioner:         "infinibox-csi-driver",
+			expectError:         true,
+		},
+		{
+			name:                "VSC does not exist",
+			policySnapshotClass: "non-existent",
+			provisioner:         "infinibox-csi-driver",
+			expectError:         true,
+		},
+		{
+			name:                "case-insensitive name matching",
+			policySnapshotClass: "VSC-ARRAY-1",
+			provisioner:         "infinibox-csi-driver",
+			expectedVSC:         vscArray1,
+			expectError:         false,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			actualVSC, actualError := GetVolumeSnapshotClassFromVolumePolicy(
+				tc.policySnapshotClass, tc.provisioner, snapshotClasses)
+			if tc.expectError {
+				require.Error(t, actualError)
+				assert.Nil(t, actualVSC)
+				return
+			}
+			if tc.expectedVSC == nil {
+				assert.Nil(t, actualVSC)
+			} else {
+				require.NotNil(t, actualVSC)
+				assert.Equal(t, tc.expectedVSC.Name, actualVSC.Name)
+				assert.Equal(t, tc.expectedVSC.Driver, actualVSC.Driver)
+			}
+		})
+	}
+}
+
 func TestGetVolumeSnapshotClassForStorageClass(t *testing.T) {
 	hostpathClass := &snapshotv1api.VolumeSnapshotClass{
 		ObjectMeta: metav1.ObjectMeta{