rbd: migration secret support

Signed-off-by: Humble Chirammal <[email protected]>

e2e/ceph_user.go

@@ -14,13 +14,17 @@ const (
 	keyringRBDNodePluginUsername           = "cephcsi-rbd-node"
 	keyringRBDNamespaceProvisionerUsername = "cephcsi-rbd-ns-provisioner"
 	keyringRBDNamespaceNodePluginUsername  = "cephcsi-rbd-ns-node"
+	keyringRBDMigrationProvisionerUsername = "cephcsi-rbd-mig-provisioner"
+	keyringRBDMigrationNodePluginUsername  = "cephcsi-rbd-mig-node"
 	keyringCephFSProvisionerUsername       = "cephcsi-cephfs-provisioner"
 	keyringCephFSNodePluginUsername        = "cephcsi-cephfs-node"
 	// secret names.
 	rbdNodePluginSecretName           = "cephcsi-rbd-node"
 	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

@@ -144,3 +144,71 @@ func generateClusterIDConfigMapForMigration(f *framework.Framework, c kubernetes
 
 	return nil
 }
+
+func createRBDMigrationSecret(f *framework.Framework, secretName, userName, userKey string) error {
+	scPath := fmt.Sprintf("%s/%s", rbdExamplePath, "secret.yaml")
+	sc, err := getSecret(scPath)
+	if err != nil {
+		return err
+	}
+	if secretName != "" {
+		sc.Name = secretName
+	}
+	if userName != "admin" {
+		sc.StringData["adminId"] = userName
+	}
+	sc.StringData["key"] = userKey
+	sc.Namespace = cephCSINamespace
+	_, err = f.ClientSet.CoreV1().Secrets(cephCSINamespace).Create(context.TODO(), &sc, metav1.CreateOptions{})
+
+	return err
+}
+
+func createMigrationUserSecretAndSC(f *framework.Framework, scName string) error {
+	if scName == "" {
+		scName = defaultSCName
+	}
+	key, err := createCephUser(
+		f,
+		keyringRBDMigrationProvisionerUsername,
+		rbdProvisionerCaps(defaultRBDPool, radosNamespace),
+	)
+	if err != nil {
+		return fmt.Errorf("failed to create user %s with error %w", keyringRBDMigrationProvisionerUsername, err)
+	}
+	err = createRBDMigrationSecret(f, rbdMigrationProvisionerSecretName, keyringRBDMigrationProvisionerUsername, key)
+	if err != nil {
+		return fmt.Errorf("failed to create provisioner secret with error %w", err)
+	}
+	// create rbd plugin secret
+	key, err = createCephUser(
+		f,
+		keyringRBDMigrationNodePluginUsername,
+		rbdNodePluginCaps(defaultRBDPool, radosNamespace))
+	if err != nil {
+		return fmt.Errorf("failed to create user %s with error %w", keyringRBDMigrationNodePluginUsername, err)
+	}
+	err = createRBDMigrationSecret(f, rbdMigrationNodePluginSecretName, keyringRBDMigrationNodePluginUsername, key)
+	if err != nil {
+		return fmt.Errorf("failed to create node secret with error %w", err)
+	}
+
+	err = deleteResource(rbdExamplePath + "storageclass.yaml")
+	if err != nil {
+		return fmt.Errorf("failed to delete storageclass with error %w", err)
+	}
+	param := make(map[string]string)
+	// override existing secrets
+	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 with error %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 secret
+				err = createMigrationUserSecretAndSC(f, "migrationsc")
 				if err != nil {
 					e2elog.Failf("failed to create storageclass with error %v", err)
 				}
@@ -392,6 +394,37 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to create configmap with error %v", err)
 				}
+
+				// delete RBD provisioner secret
+				err = deleteCephUser(f, keyringRBDMigrationProvisionerUsername)
+				if err != nil {
+					e2elog.Failf("failed to delete user %s with error %v", keyringRBDMigrationProvisionerUsername, err)
+				}
+				err = c.CoreV1().
+					Secrets(cephCSINamespace).
+					Delete(context.TODO(), rbdMigrationProvisionerSecretName, metav1.DeleteOptions{})
+				if err != nil {
+					e2elog.Failf("failed to delete provisioner secret with error %v", err)
+				}
+				// delete RBD plugin secret
+				err = deleteCephUser(f, keyringRBDMigrationNodePluginUsername)
+				if err != nil {
+					e2elog.Failf("failed to delete user %s with error %v", keyringRBDMigrationNodePluginUsername, err)
+				}
+				err = c.CoreV1().
+					Secrets(cephCSINamespace).
+					Delete(context.TODO(), rbdMigrationNodePluginSecretName, metav1.DeleteOptions{})
+				if err != nil {
+					e2elog.Failf("failed to delete node secret with error %v", err)
+				}
+				err = deleteResource(rbdExamplePath + "storageclass.yaml")
+				if err != nil {
+					e2elog.Failf("failed to delete storageclass 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,7 +1639,7 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to generate clusterID configmap with error %v", err)
 				}
-				err = validateRBDStaticMigrationPV(f, appPath, false)
+				err = validateRBDStaticMigrationPV(f, appPath, false, true)
 				if err != nil {
 					e2elog.Failf("failed to validate rbd migrated static pv with error %v", err)
 				}
@@ -1627,7 +1660,7 @@ var _ = Describe("RBD", func() {
 				if err != nil {
 					e2elog.Failf("failed to generate clusterID configmap with error %v", err)
 				}
-				err = validateRBDStaticMigrationPV(f, rawAppPath, true)
+				err = validateRBDStaticMigrationPV(f, rawAppPath, true, true)
 				if err != nil {
 					e2elog.Failf("failed to validate rbd migrated static block pv 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 string, isBlock, useMigrationSecret bool) error {
 	opt := make(map[string]string)
 	var (
 		rbdImageName = "test-static-pv"
@@ -230,7 +230,8 @@ func validateRBDStaticMigrationPV(f *framework.Framework, appPath string, isBloc
 		namespace    = f.UniqueName
 		// minikube creates default class in cluster, we need to set dummy
 		// storageclass on PV and PVC to avoid storageclass name mismatch
-		sc = "storage-class"
+		sc                      = "storage-class"
+		rbdNodePluginSecretName = rbdNodePluginSecretName
 	)
 
 	c := f.ClientSet
@@ -255,6 +256,22 @@ func validateRBDStaticMigrationPV(f *framework.Framework, appPath string, isBloc
 		return fmt.Errorf("failed to create rbd image %s", e)
 	}
 
+	if useMigrationSecret {
+		// create rbd plugin secret
+		key, err := createCephUser(
+			f,
+			keyringRBDMigrationNodePluginUsername,
+			rbdNodePluginCaps(defaultRBDPool, radosNamespace))
+		if err != nil {
+			return fmt.Errorf("failed to create user %s with error %w", keyringRBDMigrationNodePluginUsername, err)
+		}
+		err = createRBDMigrationSecret(f, rbdMigrationNodePluginSecretName, keyringRBDMigrationNodePluginUsername, key)
+		if err != nil {
+			return fmt.Errorf("failed to create node secret with error %w", err)
+		}
+		rbdNodePluginSecretName = rbdMigrationNodePluginSecretName
+	}
+
 	opt["migration"] = "true"
 	opt["clusterID"] = getMonsHash(mon)
 	opt["imageFeatures"] = staticPVImageFeature

internal/rbd/controllerserver.go

@@ -810,18 +810,30 @@ func (cs *ControllerServer) DeleteVolume(
 		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")
 	}
 
+	if isMigrationSecret(req.GetSecrets()) {
+		log.DebugLog(ctx, "passed secret is a migration secret : %+v", req.GetSecrets())
+		passedSecretMap, err := parseAndSetSecretMapFromMigSecret(req.GetSecrets())
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		log.DebugLog(ctx, "new secretmap : %+v", passedSecretMap)
+		// set req.Secrets here or create credentials here using the passedSecretMap
+		req.Secrets = passedSecretMap
+	}
+
+	cr, err := util.NewUserCredentials(req.GetSecrets())
+	if err != nil {
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+	defer cr.DeleteCredentials()
+
+	log.DebugLog(ctx, "credentials: %+v", cr)
 	if acquired := cs.VolumeLocks.TryAcquire(volumeID); !acquired {
 		log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, volumeID)
 
@@ -837,16 +849,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)
-		err = parseAndDeleteMigratedVolume(ctx, volumeID, cr)
-		if err != nil && !errors.Is(err, ErrImageNotFound) {
-			return nil, status.Error(codes.Internal, err.Error())
+		log.DebugLog(ctx, "migration volume ID detected")
+		pmVolID, err := parseMigrationVolID(volumeID)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		log.DebugLog(ctx, "Deleting volume with cr:%+v pmVolID: %+v ", cr, pmVolID)
+		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())
 	defer rbdVol.Destroy()
 	if err != nil {

internal/rbd/errors.go

@@ -42,4 +42,8 @@ var (
 	ErrMissingImageNameInVolID = errors.New("rbd image name information can not be empty in volID")
 	// ErrDecodeClusterIDFromMonsInVolID is returned when mons hash decoding on migration volID.
 	ErrDecodeClusterIDFromMonsInVolID = errors.New("failed to get clusterID from monitors hash in volID")
+	// ErrEmptyMigrationSecret is returned when passed secret map does not contain any data.
+	ErrEmptyMigrationSecret = errors.New("passed secret is empty")
+	// ErrEmptyUserKeyInMigrationSecret is returned when passed secret map does not contain any key for the user.
+	ErrEmptyUserKeyInMigrationSecret = errors.New("passed secret map does not contain user key")
 )

internal/rbd/migration.go

@@ -19,6 +19,7 @@ package rbd
 import (
 	"context"
 	"encoding/hex"
+	"fmt"
 	"strings"
 
 	"github.com/ceph/ceph-csi/internal/util"
@@ -32,6 +33,12 @@ func isMigrationVolID(volHash string) bool {
 		strings.Contains(volHash, migImageNamePrefix) && strings.Contains(volHash, migMonPrefix)
 }
 
+// isMigrationSecret validates if the passed in secretmap is a secret
+// of a migration volume request.
+func isMigrationSecret(passedSecretMap map[string]string) bool {
+	return passedSecretMap[migUserKey] != ""
+}
+
 // parseMigrationVolID decodes the volume ID and generates a migrationVolID
 // struct which consists of mon, image name, pool and clusterID information.
 func parseMigrationVolID(vh string) (*migrationVolID, error) {
@@ -41,13 +48,15 @@ func parseMigrationVolID(vh string) (*migrationVolID, error) {
 		// its short of length in this case, so return error
 		return nil, ErrInvalidVolID
 	}
+
 	// Store pool
 	poolHash := strings.Join(handSlice[migVolIDSplitLength:], migVolIDFieldSep)
 	poolByte, dErr := hex.DecodeString(poolHash)
 	if dErr != nil {
 		return nil, ErrMissingPoolNameInVolID
 	}
 	mh.poolName = string(poolByte)
+
 	// Parse migration mons( for clusterID) and image
 	for _, field := range handSlice[:migVolIDSplitLength] {
 		switch {
@@ -74,15 +83,16 @@ 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 {
-	parsedMigHandle, err := parseMigrationVolID(volumeID)
-	if err != nil {
-		log.ErrorLog(ctx, "failed to parse migration volumeID: %s , err: %v", volumeID, err)
+func DeleteMigratedVolume(ctx context.Context, parsedMigHandle *migrationVolID, cr *util.Credentials) error {
+	var err error
+	if parsedMigHandle == nil {
+		log.ErrorLog(ctx, "migration volume details are nil")
 
-		return err
+		return fmt.Errorf("migration volume details are nil")
 	}
+
 	rv := &rbdVolume{}
 
 	// fill details to rv struct from parsed migration handle
@@ -114,3 +124,30 @@ func parseAndDeleteMigratedVolume(ctx context.Context, volumeID string, cr *util
 
 	return nil
 }
+
+func parseAndSetSecretMapFromMigSecret(passedSecretMap map[string]string) (map[string]string, error) {
+	newSecretMap := make(map[string]string)
+	var err error
+	// 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.
+	if len(passedSecretMap) != 0 {
+		// parse input secret map and set the userID
+		if newSecretMap["adminId"] != "" {
+			newSecretMap[util.CredUserID] = newSecretMap["adminId"]
+		} else {
+			newSecretMap[util.CredUserID] = migUserName
+		}
+
+		// parse and set userKey
+		if passedSecretMap[migUserKey] == "" {
+			return nil, ErrEmptyUserKeyInMigrationSecret
+		}
+		newSecretMap[util.CredUserKey] = passedSecretMap[migUserKey]
+		return newSecretMap, err
+	}
+
+	return nil, ErrEmptyMigrationSecret
+}

internal/rbd/migration_test.go

@@ -171,3 +171,31 @@ func TestParseMigrationVolID(t *testing.T) {
 		})
 	}
 }
+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)
+			}
+		})
+	}
+}