cephfs: apply MutableParameters from CreateVolume and add e2e test

Address review feedback on PR #6390: the mutable VolumeAttributesClass
parameters must also be applied during CreateVolume, not only through
ControllerModifyVolume. Following the pattern in the NFS driver, the pin
logic is moved into an applyMutableParameters helper so that locking and
volume resolution happen only once, and both CreateVolume and
ControllerModifyVolume reuse it.

Also add an e2e test under e2e/cephfs.go that creates a PVC referencing a
VolumeAttributesClass with an mds-pin-export rank, verifies the class is
applied at creation time, and then updates the PVC to a second class to
exercise the ControllerModifyVolume path.

Assisted-by: goose <noreply@block.xyz>
Signed-off-by: Ismael Puerto Freire <ismaelpf@inditex.com>

Author: ismaelpuerto

e2e/cephfs.go

@@ -2732,6 +2732,99 @@ var _ = Describe(cephfsType, func() {
 			validateOmapCount(f, 0, cephfsType, metadataPool, snapsType)
 		})
 
+		It("create a PVC with a VolumeAttributesClass for MDS pinning and modify it", func() {
+			// VolumeAttributesClass support requires Kubernetes 1.34+.
+			if !k8sVersionGreaterEquals(f.ClientSet, 1, 34) {
+				framework.Logf("skipping VolumeAttributesClass test, needs Kubernetes >= 1.34")
+
+				return
+			}
+
+			// This test validates VolumeAttributesClass (VAC) support for
+			// CephFS subvolumes via MDS pinning, exercising both the
+			// CreateVolume and ControllerModifyVolume code paths:
+			//   1. Create a StorageClass.
+			//   2. Create two VACs with different mds-pin-export ranks.
+			//   3. Create a PVC referencing the first VAC and verify the PV
+			//      gets the VAC applied at creation time.
+			//   4. Update the PVC to reference the second VAC and verify the
+			//      change is applied via ControllerModifyVolume.
+			//   5. Clean up.
+			err := createCephfsStorageClass(f.ClientSet, f, true, nil)
+			if err != nil {
+				logAndFail("failed to create CephFS storageclass: %v", err)
+			}
+			defer func() {
+				err = deleteResource(cephFSExamplePath + "storageclass.yaml")
+				if err != nil {
+					logAndFail("failed to delete CephFS storageclass: %v", err)
+				}
+			}()
+
+			vacName1 := "e2e-" + f.UniqueName + "-cephfs-vac1"
+			vacName2 := "e2e-" + f.UniqueName + "-cephfs-vac2"
+
+			err = createCephFSVolumeAttributesClass(f.ClientSet, vacName1,
+				map[string]string{"mds-pin-export": "0"})
+			if err != nil {
+				logAndFail("failed to create VolumeAttributesClass: %v", err)
+			}
+			defer func() {
+				if err = deleteCephFSVolumeAttributesClass(f.ClientSet, vacName1); err != nil {
+					logAndFail("failed to delete VolumeAttributesClass: %v", err)
+				}
+			}()
+
+			err = createCephFSVolumeAttributesClass(f.ClientSet, vacName2,
+				map[string]string{"mds-pin-export": "1"})
+			if err != nil {
+				logAndFail("failed to create VolumeAttributesClass: %v", err)
+			}
+			defer func() {
+				if err = deleteCephFSVolumeAttributesClass(f.ClientSet, vacName2); err != nil {
+					logAndFail("failed to delete VolumeAttributesClass: %v", err)
+				}
+			}()
+
+			// create a PVC referencing the first VAC; this exercises the
+			// CreateVolume path applying the mutable parameters.
+			pvc, err := loadPVC(pvcPath)
+			if err != nil {
+				logAndFail("failed to load PVC: %v", err)
+			}
+			pvc.Namespace = f.UniqueName
+			pvc.Spec.VolumeAttributesClassName = &vacName1
+
+			err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+			if err != nil {
+				logAndFail("failed to create PVC: %v", err)
+			}
+
+			pv, err := getBoundPV(f.ClientSet, pvc)
+			if err != nil {
+				logAndFail("failed to get bound PV: %v", err)
+			}
+			if pv.Spec.VolumeAttributesClassName == nil ||
+				*pv.Spec.VolumeAttributesClassName != vacName1 {
+				logAndFail("PV %s does not reference the expected VolumeAttributesClass %q",
+					pv.Name, vacName1)
+			}
+			validateSubvolumeCount(f, 1, fileSystemName, subvolumegroup)
+
+			// update the PVC to reference the second VAC; this exercises the
+			// ControllerModifyVolume path.
+			err = modifyPVCVolumeAttributesClass(f.ClientSet, pvc, vacName2)
+			if err != nil {
+				logAndFail("failed to modify VolumeAttributesClass: %v", err)
+			}
+
+			err = deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+			if err != nil {
+				logAndFail("failed to delete PVC: %v", err)
+			}
+			validateSubvolumeCount(f, 0, fileSystemName, subvolumegroup)
+		})
+
 		// Make sure this should be last testcase in
 		// this file, because it deletes pool
 		It("Create a PVC and delete PVC when backend pool deleted", func() {

e2e/cephfs_helper.go

@@ -27,6 +27,7 @@ import (
 	snapapi "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"
 	v1 "k8s.io/api/core/v1"
 	scv1 "k8s.io/api/storage/v1"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
@@ -986,3 +987,65 @@ func verifyUserIdMappingMetadataSnapshotBacked(
 
 	return nil
 }
+
+// createCephFSVolumeAttributesClass creates a VolumeAttributesClass for CephFS
+// from the example manifest, optionally overriding its name and parameters.
+func createCephFSVolumeAttributesClass(
+	c kubernetes.Interface,
+	name string,
+	params map[string]string,
+) error {
+	vac, err := getVolumeAttributesClass(cephFSExamplePath + "volumeattributesclass.yaml")
+	if err != nil {
+		return fmt.Errorf("failed to get vac: %w", err)
+	}
+	if name != "" {
+		vac.Name = name
+	}
+
+	// override with the parameters passed by the caller, so that a test can
+	// exercise a specific MDS pin type.
+	if params != nil {
+		vac.Parameters = params
+	}
+
+	timeout := time.Duration(deployTimeout) * time.Minute
+
+	return wait.PollUntilContextTimeout(context.TODO(), poll, timeout, true, func(ctx context.Context) (bool, error) {
+		_, err = c.StorageV1().VolumeAttributesClasses().Create(ctx, &vac, metav1.CreateOptions{})
+		if err != nil {
+			if apierrs.IsAlreadyExists(err) {
+				return true, nil
+			}
+			if isRetryableAPIError(err) {
+				return false, nil
+			}
+
+			return false, fmt.Errorf("failed to create VolumeAttributesClass %q: %w", vac.Name, err)
+		}
+
+		return true, nil
+	})
+}
+
+// deleteCephFSVolumeAttributesClass deletes the named CephFS
+// VolumeAttributesClass.
+func deleteCephFSVolumeAttributesClass(c kubernetes.Interface, name string) error {
+	timeout := time.Duration(deployTimeout) * time.Minute
+
+	return wait.PollUntilContextTimeout(context.TODO(), poll, timeout, true, func(ctx context.Context) (bool, error) {
+		err := c.StorageV1().VolumeAttributesClasses().Delete(ctx, name, metav1.DeleteOptions{})
+		if err != nil {
+			if apierrs.IsNotFound(err) {
+				return true, nil
+			}
+			if isRetryableAPIError(err) {
+				return false, nil
+			}
+
+			return false, fmt.Errorf("failed to delete VolumeAttributesClass %q: %w", name, err)
+		}
+
+		return true, nil
+	})
+}

internal/cephfs/controllerserver.go

@@ -412,6 +412,15 @@ func (cs *cephfsControllerServer) CreateVolume(
 			}
 		}
 
+		// apply MutableParameters (e.g. MDS pinning) from a
+		// VolumeAttributesClass, if any were requested.
+		if len(req.GetMutableParameters()) != 0 {
+			if err = applyMutableParameters(ctx, volClient, vID.FsSubvolName,
+				req.GetMutableParameters()); err != nil {
+				return nil, err
+			}
+		}
+
 		return buildCreateVolumeResponse(req, volOptions, vID), nil
 	}
 
@@ -486,6 +495,15 @@ func (cs *cephfsControllerServer) CreateVolume(
 	log.DebugLog(ctx, "cephfs: successfully created backing volume named %s for request name %s",
 		vID.FsSubvolName, requestName)
 
+	// apply MutableParameters (e.g. MDS pinning) from a VolumeAttributesClass,
+	// if any were requested.
+	if len(req.GetMutableParameters()) != 0 {
+		if err = applyMutableParameters(ctx, volClient, vID.FsSubvolName,
+			req.GetMutableParameters()); err != nil {
+			return nil, err
+		}
+	}
+
 	return buildCreateVolumeResponse(req, volOptions, vID), nil
 }
 
@@ -1559,8 +1577,12 @@ func validateMDSPinSetting(key, setting string) error {
 	return nil
 }
 
-// ControllerModifyVolume modifies mutable attributes of a CephFS subvolume
-// based on parameters from a VolumeAttributesClass.
+// applyMutableParameters applies the mutable VolumeAttributesClass parameters
+// to an already resolved subvolume. It performs no locking and does not resolve
+// the volume; the caller is expected to have taken the required locks and to
+// provide a ready-to-use SubVolumeClient. This lets both CreateVolume and
+// ControllerModifyVolume share the same logic without duplicating locking or
+// volume resolution steps.
 //
 // Currently supported parameters (mutually exclusive, at most one may be set):
 //   - mds-pin-export:      pin to a specific MDS rank (e.g. "2")
@@ -1570,6 +1592,34 @@ func validateMDSPinSetting(key, setting string) error {
 // Similar To:
 //
 //	ceph fs subvolume pin <vol_name> <sub_name> <pin_type> <pin_setting>
+func applyMutableParameters(
+	ctx context.Context,
+	volClient core.SubVolumeClient,
+	volID string,
+	mutableParams map[string]string,
+) error {
+	pinType, pinSetting, err := validateMDSPinParameters(mutableParams)
+	if err != nil {
+		return status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	if pinType != "" {
+		if err = volClient.PinVolume(ctx, pinType, pinSetting); err != nil {
+			log.ErrorLog(ctx, "failed to pin subvolume %s: %v", volID, err)
+
+			return status.Error(codes.Internal, err.Error())
+		}
+		log.DebugLog(ctx, "cephfs: subvolume %s pinned with type=%s setting=%s",
+			volID, pinType, pinSetting)
+	}
+
+	return nil
+}
+
+// ControllerModifyVolume modifies mutable attributes of a CephFS subvolume
+// based on parameters from a VolumeAttributesClass. It acquires the required
+// locks, resolves the volume and delegates the actual work to
+// applyMutableParameters.
 func (cs *cephfsControllerServer) ControllerModifyVolume(
 	ctx context.Context,
 	req *csi.ControllerModifyVolumeRequest,
@@ -1617,21 +1667,10 @@ func (cs *cephfsControllerServer) ControllerModifyVolume(
 	}
 	defer volOptions.Destroy()
 
-	pinType, pinSetting, err := validateMDSPinParameters(mutableParams)
-	if err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
-	}
-
-	if pinType != "" {
-		volClient := core.NewSubVolume(volOptions.GetConnection(),
-			&volOptions.SubVolume, volOptions.ClusterID, cs.ClusterName)
-		if err = volClient.PinVolume(ctx, pinType, pinSetting); err != nil {
-			log.ErrorLog(ctx, "failed to pin subvolume %s: %v", volID, err)
-
-			return nil, status.Error(codes.Internal, err.Error())
-		}
-		log.DebugLog(ctx, "cephfs: subvolume %s pinned with type=%s setting=%s",
-			volID, pinType, pinSetting)
+	volClient := core.NewSubVolume(volOptions.GetConnection(),
+		&volOptions.SubVolume, volOptions.ClusterID, cs.ClusterName)
+	if err = applyMutableParameters(ctx, volClient, volID, mutableParams); err != nil {
+		return nil, err
 	}
 
 	return &csi.ControllerModifyVolumeResponse{}, nil