Merge pull request #41 from ceph/devel

Sync the devel branch with current upstream

docs/deploy-rbd.md

@@ -64,7 +64,7 @@ make image-cephcsi
 | `mounter`                                                                                           | no                   | if set to `rbd-nbd`, use `rbd-nbd` on nodes that have `rbd-nbd` and `nbd` kernel modules to map rbd images                                                                                                                                                                                                                                                                                                                                                                                             |
 | `encrypted`                                                                                         | no                   | disabled by default, use `"true"` to enable LUKS encryption on PVC and `"false"` to disable it. **Do not change for existing storageclasses**                                                                                                                                                                                                                                                                                                                                                          |
 | `encryptionKMSID`                                                                                   | no                   | required if encryption is enabled and a kms is used to store passphrases                                                                                                                                                                                                                                                                                                                                                                                                                               |
-| `thickProvision`                                                                                    | no                   | if set to `"true"`, newly created RBD images will be completely allocated by writing zeros to it                                                                                                                                                                                                                                                                                                                                                                                                       |
+| `thickProvision`                                                                                    | no                   | if set to `"true"`, newly created RBD images will be completely allocated by writing zeros to it (**DEPRECATED**: recommended alternative solution is to use accounting/quotas for created volumes)                                                                                                                                                                                                                                                                                                    |
 
 **NOTE:** An accompanying CSI configuration file, needs to be provided to the
 running pods. Refer to [Creating CSI configuration](../examples/README.md#creating-csi-configuration)

e2e/ceph_user.go

@@ -21,6 +21,8 @@ const (
 	rbdProvisionerSecretName          = "cephcsi-rbd-provisioner"
 	rbdNamespaceNodePluginSecretName  = "cephcsi-rbd-ns-node"
 	rbdNamespaceProvisionerSecretName = "cephcsi-rbd-ns-provisioner"
+	rbdMigrationNodePluginSecretName  = "cephcsi-rbd-mig-node"
+	rbdMigrationProvisionerSecretName = "cephcsi-rbd-mig-provisioner"
 	cephFSNodePluginSecretName        = "cephcsi-cephfs-node"
 	cephFSProvisionerSecretName       = "cephcsi-cephfs-provisioner"
 )

e2e/migration.go

@@ -96,7 +96,7 @@ func validateRBDStaticMigrationPVDeletion(f *framework.Framework, appPath, scNam
 
 	err = deletePVCAndApp("", f, pvc, app)
 	if err != nil {
-		return fmt.Errorf("failed to delete PVC and application with error %w", err)
+		return fmt.Errorf("failed to delete PVC and application: %w", err)
 	}
 
 	return err
@@ -134,12 +134,142 @@ func generateClusterIDConfigMapForMigration(f *framework.Framework, c kubernetes
 	// create custom configmap
 	err = createCustomConfigMap(f.ClientSet, rbdDirPath, clusterInfo)
 	if err != nil {
-		return fmt.Errorf("failed to create configmap with error %w", err)
+		return fmt.Errorf("failed to create configmap: %w", err)
 	}
 	// restart csi pods for the configmap to take effect.
 	err = recreateCSIRBDPods(f)
 	if err != nil {
-		return fmt.Errorf("failed to recreate rbd csi pods with error %w", err)
+		return fmt.Errorf("failed to recreate rbd csi pods: %w", err)
+	}
+
+	return nil
+}
+
+// createRBDMigrationSecret creates a migration secret with the passed in user name.
+// this secret differs from csi secret data on below aspects.
+// equivalent to the `UserKey` field, migration secret has `key` field.
+// if 'userName' has passed and if it is not admin, the passed in userName will be
+// set as the `adminId` field in the secret.
+func createRBDMigrationSecret(f *framework.Framework, secretName, userName, userKey string) error {
+	secPath := fmt.Sprintf("%s/%s", rbdExamplePath, "secret.yaml")
+	sec, err := getSecret(secPath)
+	if err != nil {
+		return err
+	}
+	if secretName != "" {
+		sec.Name = secretName
+	}
+	// if its admin, we dont need to change anything in the migration secret, the CSI driver
+	// will use the key from existing secret and continue.
+	if userName != "admin" {
+		sec.StringData["adminId"] = userName
+	}
+	sec.StringData["key"] = userKey
+	sec.Namespace = cephCSINamespace
+	_, err = f.ClientSet.CoreV1().Secrets(cephCSINamespace).Create(context.TODO(), &sec, metav1.CreateOptions{})
+
+	return err
+}
+
+// createMigrationUserSecretAndSC creates migration user and a secret associated with this user first,
+// then create SC based on the same.
+func createMigrationUserSecretAndSC(f *framework.Framework, scName string) error {
+	if scName == "" {
+		scName = defaultSCName
+	}
+	err := createProvNodeCephUserAndSecret(f, true, true)
+	if err != nil {
+		return err
+	}
+
+	err = createMigrationSC(f, scName)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func createMigrationSC(f *framework.Framework, scName string) error {
+	err := deleteResource(rbdExamplePath + "storageclass.yaml")
+	if err != nil {
+		return fmt.Errorf("failed to delete storageclass: %w", err)
+	}
+	param := make(map[string]string)
+	// add new secrets to the SC parameters
+	param["csi.storage.k8s.io/provisioner-secret-namespace"] = cephCSINamespace
+	param["csi.storage.k8s.io/provisioner-secret-name"] = rbdMigrationProvisionerSecretName
+	param["csi.storage.k8s.io/controller-expand-secret-namespace"] = cephCSINamespace
+	param["csi.storage.k8s.io/controller-expand-secret-name"] = rbdMigrationProvisionerSecretName
+	param["csi.storage.k8s.io/node-stage-secret-namespace"] = cephCSINamespace
+	param["csi.storage.k8s.io/node-stage-secret-name"] = rbdMigrationNodePluginSecretName
+	err = createRBDStorageClass(f.ClientSet, f, scName, nil, param, deletePolicy)
+	if err != nil {
+		return fmt.Errorf("failed to create storageclass: %w", err)
+	}
+
+	return nil
+}
+
+// createProvNodeCephUserAndSecret fetches the ceph migration user's key and create migration secret
+// with it based on the arg values of 'provSecret' and 'nodeSecret'.
+func createProvNodeCephUserAndSecret(f *framework.Framework, provisionerSecret, nodeSecret bool) error {
+	if provisionerSecret {
+		// Fetch the key.
+		key, err := createCephUser(
+			f,
+			keyringRBDProvisionerUsername,
+			rbdProvisionerCaps(defaultRBDPool, radosNamespace),
+		)
+		if err != nil {
+			return fmt.Errorf("failed to create user %q: %w", keyringRBDProvisionerUsername, err)
+		}
+		err = createRBDMigrationSecret(f, rbdMigrationProvisionerSecretName, keyringRBDProvisionerUsername, key)
+		if err != nil {
+			return fmt.Errorf("failed to create provisioner secret: %w", err)
+		}
+	}
+
+	if nodeSecret {
+		// Fetch the key.
+		key, err := createCephUser(
+			f,
+			keyringRBDNodePluginUsername,
+			rbdNodePluginCaps(defaultRBDPool, radosNamespace))
+		if err != nil {
+			return fmt.Errorf("failed to create user %q: %w", keyringRBDNodePluginUsername, err)
+		}
+		err = createRBDMigrationSecret(f, rbdMigrationNodePluginSecretName, keyringRBDNodePluginUsername, key)
+		if err != nil {
+			return fmt.Errorf("failed to create node secret: %w", err)
+		}
+	}
+
+	return nil
+}
+
+// deleteProvNodeMigrationSecret deletes ceph migration secrets based on the
+// arg values of 'provisionerSecret' and 'nodeSecret'.
+func deleteProvNodeMigrationSecret(f *framework.Framework, provisionerSecret, nodeSecret bool) error {
+	c := f.ClientSet
+	if provisionerSecret {
+		// delete RBD provisioner secret.
+		err := c.CoreV1().
+			Secrets(cephCSINamespace).
+			Delete(context.TODO(), rbdMigrationProvisionerSecretName, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to delete provisioner secret: %w", err)
+		}
+	}
+
+	if nodeSecret {
+		// delete RBD node secret.
+		err := c.CoreV1().
+			Secrets(cephCSINamespace).
+			Delete(context.TODO(), rbdMigrationNodePluginSecretName, metav1.DeleteOptions{})
+		if err != nil {
+			return fmt.Errorf("failed to delete node secret: %w", err)
+		}
 	}
 
 	return nil

e2e/rbd.go

@@ -374,7 +374,9 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to generate clusterID configmap with error %v", err)
 				}
-				err = createRBDStorageClass(f.ClientSet, f, "migrationsc", nil, nil, deletePolicy)
+
+				// create a sc with different migration secret
+				err = createMigrationUserSecretAndSC(f, "migrationsc")
 				if err != nil {
 					e2elog.Failf("failed to create storageclass with error %v", err)
 				}
@@ -392,6 +394,15 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to create configmap with error %v", err)
 				}
+
+				err = deleteProvNodeMigrationSecret(f, true, true)
+				if err != nil {
+					e2elog.Failf("failed to delete migration users and Secrets associated with error %v", err)
+				}
+				err = createRBDStorageClass(f.ClientSet, f, defaultSCName, nil, nil, deletePolicy)
+				if err != nil {
+					e2elog.Failf("failed to create storageclass with error %v", err)
+				}
 			})
 
 			By("create a PVC and validate owner", func() {
@@ -1606,12 +1617,24 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to generate clusterID configmap with error %v", err)
 				}
-				err = validateRBDStaticMigrationPV(f, appPath, false)
+				// create node user and migration secret.
+				err = createProvNodeCephUserAndSecret(f, false, true)
+				if err != nil {
+					e2elog.Failf("failed to create users and secret with error %v", err)
+				}
+
+				err = validateRBDStaticMigrationPV(f, appPath, rbdMigrationNodePluginSecretName, false)
 				if err != nil {
 					e2elog.Failf("failed to validate rbd migrated static pv with error %v", err)
 				}
 				// validate created backend rbd images
 				validateRBDImageCount(f, 0, defaultRBDPool)
+
+				err = deleteProvNodeMigrationSecret(f, false, true)
+				if err != nil {
+					e2elog.Failf("failed to delete users and secret with error %v", err)
+				}
+
 				err = deleteConfigMap(rbdDirPath)
 				if err != nil {
 					e2elog.Failf("failed to delete configmap with error %v", err)
@@ -1627,12 +1650,24 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to generate clusterID configmap with error %v", err)
 				}
-				err = validateRBDStaticMigrationPV(f, rawAppPath, true)
+				// create node user and migration secret.
+				err = createProvNodeCephUserAndSecret(f, false, true)
+				if err != nil {
+					e2elog.Failf("failed to create users and secret with error %v", err)
+				}
+
+				err = validateRBDStaticMigrationPV(f, rawAppPath, rbdMigrationNodePluginSecretName, true)
 				if err != nil {
 					e2elog.Failf("failed to validate rbd migrated static block pv with error %v", err)
 				}
 				// validate created backend rbd images
 				validateRBDImageCount(f, 0, defaultRBDPool)
+
+				err = deleteProvNodeMigrationSecret(f, false, true)
+				if err != nil {
+					e2elog.Failf("failed to delete users and secret with error %v", err)
+				}
+
 				err = deleteConfigMap(rbdDirPath)
 				if err != nil {
 					e2elog.Failf("failed to delete configmap with error %v", err)

e2e/staticpvc.go

@@ -221,7 +221,7 @@ func validateRBDStaticPV(f *framework.Framework, appPath string, isBlock, checkI
 	return err
 }
 
-func validateRBDStaticMigrationPV(f *framework.Framework, appPath string, isBlock bool) error {
+func validateRBDStaticMigrationPV(f *framework.Framework, appPath, nodeSecretName string, isBlock bool) error {
 	opt := make(map[string]string)
 	var (
 		rbdImageName = "test-static-pv"
@@ -254,6 +254,9 @@ func validateRBDStaticMigrationPV(f *framework.Framework, appPath string, isBloc
 	if e != "" {
 		return fmt.Errorf("failed to create rbd image %s", e)
 	}
+	if nodeSecretName == "" {
+		nodeSecretName = rbdNodePluginSecretName
+	}
 
 	opt["migration"] = "true"
 	opt["clusterID"] = getMonsHash(mon)
@@ -265,7 +268,7 @@ func validateRBDStaticMigrationPV(f *framework.Framework, appPath string, isBloc
 		pvName,
 		rbdImageName,
 		size,
-		rbdNodePluginSecretName,
+		nodeSecretName,
 		cephCSINamespace,
 		sc,
 		"rbd.csi.ceph.com",

examples/rbd/storageclass.yaml

@@ -29,9 +29,10 @@ parameters:
    # eg: pool: rbdpool
    pool: <rbd-pool-name>
 
-   # Set thickProvision to true if you want RBD images to be fully allocated on
+   # Deprecated: Set thickProvision to true if you want RBD images to be fully
-   # creation (thin provisioning is the default).
+   # allocated on creation (thin provisioning is the default).
-   thickProvision: "false"
+   # thickProvision: "false"
+
    # (required) RBD image features, CSI creates image with image-format 2
    # CSI RBD currently supports `layering`, `journaling`, `exclusive-lock`
    # features. If `journaling` is enabled, must enable `exclusive-lock` too.

internal/kms/vault.go

@@ -43,7 +43,7 @@ const (
 	vaultDefaultRole           = "csi-kubernetes"
 	vaultDefaultNamespace      = ""
 	vaultDefaultPassphrasePath = ""
-	vaultDefaultCAVerify       = "true"
+	vaultDefaultCAVerify       = true
 	vaultDefaultDestroyKeys    = "true"
 )
 
@@ -208,7 +208,7 @@ func (vc *vaultConnection) initConnection(config map[string]interface{}) error {
 		keyContext[loss.KeyVaultNamespace] = vaultNamespace
 	}
 
-	verifyCA := vaultDefaultCAVerify // optional
+	verifyCA := strconv.FormatBool(vaultDefaultCAVerify) // optional
 	err = setConfigString(&verifyCA, config, "vaultCAVerify")
 	if errors.Is(err, errConfigOptionInvalid) {
 		return err

internal/kms/vault_tokens.go

@@ -101,7 +101,6 @@ func (v *vaultTokenConf) convertStdVaultToCSIConfig(s *standardVault) {
 
 	// by default the CA should get verified, only when VaultSkipVerify is
 	// set, verification should be disabled
-	v.VaultCAVerify = vaultDefaultCAVerify
 	verify, err := strconv.ParseBool(s.VaultSkipVerify)
 	if err == nil {
 		v.VaultCAVerify = strconv.FormatBool(!verify)
@@ -124,8 +123,14 @@ func transformConfig(svMap map[string]interface{}) (map[string]interface{}, erro
 		return nil, fmt.Errorf("failed to convert config %T to JSON: %w", svMap, err)
 	}
 
-	// convert the JSON back to a standardVault struct
+	// convert the JSON back to a standardVault struct, default values are
-	sv := &standardVault{}
+	// set in case the configuration does not provide all options
+	sv := &standardVault{
+		VaultDestroyKeys: vaultDefaultDestroyKeys,
+		VaultNamespace:   vaultDefaultNamespace,
+		VaultSkipVerify:  strconv.FormatBool(!vaultDefaultCAVerify),
+	}
+
 	err = json.Unmarshal(data, sv)
 	if err != nil {
 		return nil, fmt.Errorf("failed to Unmarshal the vault configuration: %w", err)

internal/kms/vault_tokens_test.go

@@ -19,6 +19,7 @@ package kms
 import (
 	"encoding/json"
 	"errors"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -208,6 +209,18 @@ func TestTransformConfig(t *testing.T) {
 	assert.Equal(t, config["vaultCAVerify"], "false")
 }
 
+func TestTransformConfigDefaults(t *testing.T) {
+	t.Parallel()
+	cm := make(map[string]interface{})
+	cm["KMS_PROVIDER"] = kmsTypeVaultTokens
+
+	config, err := transformConfig(cm)
+	require.NoError(t, err)
+	assert.Equal(t, config["encryptionKMSType"], cm["KMS_PROVIDER"])
+	assert.Equal(t, config["vaultDestroyKeys"], vaultDefaultDestroyKeys)
+	assert.Equal(t, config["vaultCAVerify"], strconv.FormatBool(vaultDefaultCAVerify))
+}
+
 func TestVaultTokensKMSRegistered(t *testing.T) {
 	t.Parallel()
 	_, ok := kmsManager.providers[kmsTypeVaultTokens]

internal/rbd/controllerserver.go

@@ -808,25 +808,34 @@ func (cs *ControllerServer) checkErrAndUndoReserve(
 func (cs *ControllerServer) DeleteVolume(
 	ctx context.Context,
 	req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
-	if err := cs.Driver.ValidateControllerServiceRequest(
+	var err error
+	if err = cs.Driver.ValidateControllerServiceRequest(
 		csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME); err != nil {
 		log.ErrorLog(ctx, "invalid delete volume req: %v", protosanitizer.StripSecrets(req))
 
 		return nil, err
 	}
 
-	cr, err := util.NewUserCredentials(req.GetSecrets())
-	if err != nil {
-		return nil, status.Error(codes.Internal, err.Error())
-	}
-	defer cr.DeleteCredentials()
-
 	// For now the image get unconditionally deleted, but here retention policy can be checked
 	volumeID := req.GetVolumeId()
 	if volumeID == "" {
 		return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
 	}
 
+	secrets := req.GetSecrets()
+	if util.IsMigrationSecret(secrets) {
+		secrets, err = util.ParseAndSetSecretMapFromMigSecret(secrets)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+	}
+
+	cr, err := util.NewUserCredentials(secrets)
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	defer cr.DeleteCredentials()
+
 	if acquired := cs.VolumeLocks.TryAcquire(volumeID); !acquired {
 		log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
 
@@ -842,17 +851,21 @@ func (cs *ControllerServer) DeleteVolume(
 	}
 	defer cs.OperationLocks.ReleaseDeleteLock(volumeID)
 
+	// if this is a migration request volID, delete the volume in backend
 	if isMigrationVolID(volumeID) {
-		log.DebugLog(ctx, "migration volume ID : %s", volumeID)
+		pmVolID, pErr := parseMigrationVolID(volumeID)
-		err = parseAndDeleteMigratedVolume(ctx, volumeID, cr)
+		if pErr != nil {
-		if err != nil && !errors.Is(err, ErrImageNotFound) {
+			return nil, status.Error(codes.InvalidArgument, pErr.Error())
-			return nil, status.Error(codes.Internal, err.Error())
+		}
+		pErr = deleteMigratedVolume(ctx, pmVolID, cr)
+		if pErr != nil && !errors.Is(pErr, ErrImageNotFound) {
+			return nil, status.Error(codes.Internal, pErr.Error())
 		}
 
 		return &csi.DeleteVolumeResponse{}, nil
 	}
 
-	rbdVol, err := genVolFromVolID(ctx, volumeID, cr, req.GetSecrets())
+	rbdVol, err := genVolFromVolID(ctx, volumeID, cr, secrets)
 	defer rbdVol.Destroy()
 	if err != nil {
 		return cs.checkErrAndUndoReserve(ctx, err, volumeID, rbdVol, cr)
@@ -1502,6 +1515,10 @@ func (cs *ControllerServer) ControllerExpandVolume(
 	}, nil
 }
 
+// logThickProvisioningDeprecation makes sure the deprecation warning about
+// thick-provisining is logged only once.
+var logThickProvisioningDeprecation = true
+
 // isThickProvisionRequest returns true in case the request contains the
 // `thickProvision` option set to `true`.
 func isThickProvisionRequest(parameters map[string]string) bool {
@@ -1517,5 +1534,11 @@ func isThickProvisionRequest(parameters map[string]string) bool {
 		return false
 	}
 
+	if logThickProvisioningDeprecation {
+		log.WarningLogMsg("thick-provisioning is deprecated and will " +
+			"be removed in a future release")
+		logThickProvisioningDeprecation = false
+	}
+
 	return thickBool
 }

internal/rbd/migration.go

@@ -74,15 +74,10 @@ func parseMigrationVolID(vh string) (*migrationVolID, error) {
 	return mh, nil
 }
 
-// parseAndDeleteMigratedVolume get rbd volume details from the migration volID
+// deleteMigratedVolume get rbd volume details from the migration volID
 // and delete the volume from the cluster, return err if there was an error on the process.
-func parseAndDeleteMigratedVolume(ctx context.Context, volumeID string, cr *util.Credentials) error {
+func deleteMigratedVolume(ctx context.Context, parsedMigHandle *migrationVolID, cr *util.Credentials) error {
-	parsedMigHandle, err := parseMigrationVolID(volumeID)
+	var err error
-	if err != nil {
-		log.ErrorLog(ctx, "failed to parse migration volumeID: %s , err: %v", volumeID, err)
-
-		return err
-	}
 	rv := &rbdVolume{}
 
 	// fill details to rv struct from parsed migration handle

internal/rbd/nodeserver.go

@@ -151,10 +151,14 @@ func healerStageTransaction(ctx context.Context, cr *util.Credentials, volOps *r
 }
 
 // populateRbdVol update the fields in rbdVolume struct based on the request it received.
+// this function also receive the credentials and secrets args as it differs in its data.
+// The credentials are used directly by functions like voljournal.Connect() and other functions
+// like genVolFromVolumeOptions() make use of secrets.
 func populateRbdVol(
 	ctx context.Context,
 	req *csi.NodeStageVolumeRequest,
-	cr *util.Credentials) (*rbdVolume, error) {
+	cr *util.Credentials,
+	secrets map[string]string) (*rbdVolume, error) {
 	var err error
 	var j *journal.Connection
 	volID := req.GetVolumeId()
@@ -179,7 +183,7 @@ func populateRbdVol(
 		disableInUseChecks = true
 	}
 
-	rv, err := genVolFromVolumeOptions(ctx, req.GetVolumeContext(), req.GetSecrets(), disableInUseChecks, true)
+	rv, err := genVolFromVolumeOptions(ctx, req.GetVolumeContext(), secrets, disableInUseChecks, true)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -254,13 +258,20 @@ func populateRbdVol(
 func (ns *NodeServer) NodeStageVolume(
 	ctx context.Context,
 	req *csi.NodeStageVolumeRequest) (*csi.NodeStageVolumeResponse, error) {
-	if err := util.ValidateNodeStageVolumeRequest(req); err != nil {
+	var err error
+	if err = util.ValidateNodeStageVolumeRequest(req); err != nil {
 		return nil, err
 	}
 
 	volID := req.GetVolumeId()
-
+	secrets := req.GetSecrets()
-	cr, err := util.NewUserCredentials(req.GetSecrets())
+	if util.IsMigrationSecret(secrets) {
+		secrets, err = util.ParseAndSetSecretMapFromMigSecret(secrets)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+	}
+	cr, err := util.NewUserCredentials(secrets)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -298,7 +309,7 @@ func (ns *NodeServer) NodeStageVolume(
 		return nil, status.Error(codes.InvalidArgument, "missing required parameter imageFeatures")
 	}
 
-	rv, err := populateRbdVol(ctx, req, cr)
+	rv, err := populateRbdVol(ctx, req, cr, secrets)
 	if err != nil {
 		return nil, err
 	}
@@ -335,7 +346,7 @@ func (ns *NodeServer) NodeStageVolume(
 
 	// perform the actual staging and if this fails, have undoStagingTransaction
 	// cleans up for us
-	transaction, err = ns.stageTransaction(ctx, req, rv, isStaticVol)
+	transaction, err = ns.stageTransaction(ctx, req, cr, rv, isStaticVol)
 	if err != nil {
 		return nil, status.Error(codes.Internal, err.Error())
 	}
@@ -352,6 +363,7 @@ func (ns *NodeServer) NodeStageVolume(
 func (ns *NodeServer) stageTransaction(
 	ctx context.Context,
 	req *csi.NodeStageVolumeRequest,
+	cr *util.Credentials,
 	volOptions *rbdVolume,
 	staticVol bool) (stageTransaction, error) {
 	transaction := stageTransaction{}
@@ -359,13 +371,6 @@ func (ns *NodeServer) stageTransaction(
 	var err error
 	var readOnly bool
 
-	var cr *util.Credentials
-	cr, err = util.NewUserCredentials(req.GetSecrets())
-	if err != nil {
-		return transaction, err
-	}
-	defer cr.DeleteCredentials()
-
 	// Allow image to be mounted on multiple nodes if it is ROX
 	if req.VolumeCapability.AccessMode.Mode == csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY {
 		log.ExtendedLog(ctx, "setting disableInUseChecks on rbd volume to: %v", req.GetVolumeId)

internal/util/credentials.go

@@ -31,6 +31,9 @@ const (
 	credMonitors         = "monitors"
 	tmpKeyFileLocation   = "/tmp/csi/keys"
 	tmpKeyFileNamePrefix = "keyfile-"
+	migUserName          = "admin"
+	migUserID            = "adminId"
+	migUserKey           = "key"
 )
 
 // Credentials struct represents credentials to access the ceph cluster.
@@ -119,3 +122,34 @@ func GetMonValFromSecret(secrets map[string]string) (string, error) {
 
 	return "", fmt.Errorf("missing %q", credMonitors)
 }
+
+// ParseAndSetSecretMapFromMigSecret parse the secretmap from the migration request and return
+// newsecretmap with the userID and userKey fields set.
+func ParseAndSetSecretMapFromMigSecret(secretmap map[string]string) (map[string]string, error) {
+	newSecretMap := make(map[string]string)
+	// parse and set userKey
+	if !IsMigrationSecret(secretmap) {
+		return nil, errors.New("passed secret map does not contain user key or it is nil")
+	}
+	newSecretMap[credUserKey] = secretmap[migUserKey]
+	// parse and set the userID
+	newSecretMap[credUserID] = migUserName
+	if secretmap[migUserID] != "" {
+		newSecretMap[credUserID] = secretmap[migUserID]
+	}
+
+	return newSecretMap, nil
+}
+
+// IsMigrationSecret validates if the passed in secretmap is a secret
+// of a migration volume request. The migration secret carry a field
+// called `key` which is the equivalent of `userKey` which is what we
+// check here for identifying the secret.
+func IsMigrationSecret(passedSecretMap map[string]string) bool {
+	// the below 'nil' check is an extra measure as the request validators like
+	// ValidateNodeStageVolumeRequest() already does the nil check, however considering
+	// this function can be called independently with a map of secret values
+	// it is good to have this check in place, also it gives clear error about this
+	// was hit on migration request compared to general one.
+	return len(passedSecretMap) != 0 && passedSecretMap[migUserKey] != ""
+}

internal/util/credentials_test.go

@@ -0,0 +1,100 @@
+/*
+Copyright 2021 The Ceph-CSI Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package util
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestIsMigrationSecret(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name string
+		vc   map[string]string
+		want bool
+	}{
+		{
+			"proper migration secret key set",
+			map[string]string{"key": "QVFBOFF2SlZheUJQRVJBQWgvS2cwT1laQUhPQno3akZwekxxdGc9PQ=="},
+			true,
+		},
+		{
+			"no key set",
+			map[string]string{"": ""},
+			false,
+		},
+	}
+	for _, tt := range tests {
+		newtt := tt
+		t.Run(newtt.name, func(t *testing.T) {
+			t.Parallel()
+			if got := IsMigrationSecret(newtt.vc); got != newtt.want {
+				t.Errorf("isMigrationSecret() = %v, want %v", got, newtt.want)
+			}
+		})
+	}
+}
+
+func TestParseAndSetSecretMapFromMigSecret(t *testing.T) {
+	t.Parallel()
+	tests := []struct {
+		name      string
+		secretmap map[string]string
+		want      map[string]string
+		wantErr   bool
+	}{
+		{
+			"valid migration secret key set",
+			map[string]string{"key": "QVFBOFF2SlZheUJQRVJBQWgvS2cwT1laQUhPQno3akZwekxxdGc9PQ=="},
+			map[string]string{"userKey": "QVFBOFF2SlZheUJQRVJBQWgvS2cwT1laQUhPQno3akZwekxxdGc9PQ==", "userID": "admin"},
+			false,
+		},
+		{
+			"migration secret key value nil",
+			map[string]string{"key": ""},
+			nil,
+			true,
+		},
+		{
+			"migration secret key field nil",
+			map[string]string{"": ""},
+			nil,
+			true,
+		},
+		{
+			"valid migration secret key and userID set",
+			map[string]string{"key": "QVFBOFF2SlZheUJQRVJBQWgvS2cwT1laQUhPQno3akZwekxxdGc9PQ==", "adminId": "pooladmin"},
+			map[string]string{"userKey": "QVFBOFF2SlZheUJQRVJBQWgvS2cwT1laQUhPQno3akZwekxxdGc9PQ==", "userID": "pooladmin"},
+			false,
+		},
+	}
+	for _, tt := range tests {
+		newtt := tt
+		t.Run(newtt.name, func(t *testing.T) {
+			t.Parallel()
+			got, err := ParseAndSetSecretMapFromMigSecret(newtt.secretmap)
+			if (err != nil) != newtt.wantErr {
+				t.Errorf("ParseAndSetSecretMapFromMigSecret() error = %v, wantErr %v", err, newtt.wantErr)
+
+				return
+			}
+			if !reflect.DeepEqual(got, newtt.want) {
+				t.Errorf("ParseAndSetSecretMapFromMigSecret() got = %v, want %v", got, newtt.want)
+			}
+		})
+	}
+}