Adding multi-cluster support via array of cluster configs based on topology Signed-off-by: Devin Ridge <dridge@globalnoc.iu.edu>

Signed-off-by: Devin Ridge <dridge@globalnoc.iu.edu>

Author: dridge93

charts/ceph-csi-rbd/templates/storageclass.yaml

@@ -15,7 +15,12 @@ metadata:
     {{- with .Values.commonLabels }}{{ toYaml . | trim | nindent 4 }}{{- end }}
 provisioner: {{ .Values.driverName }}
 parameters:
+{{- if .Values.storageClass.clusterID }}
   clusterID: {{ .Values.storageClass.clusterID }}
+{{- end }}
+{{- if .Values.storageClass.clusterTopologyConfigMap }}
+  clusterTopologyConfigMap: {{ .Values.storageClass.clusterTopologyConfigMap }}
+{{- end }}
   imageFeatures: {{ .Values.storageClass.imageFeatures }}
 {{- if .Values.storageClass.pool }}  
   pool: {{ .Values.storageClass.pool }}

charts/ceph-csi-rbd/values.yaml

@@ -365,11 +365,17 @@ storageClass:
   #   storageclass.kubernetes.io/is-default-class: "true"
   annotations: {}
 
-  # (required) String representing a Ceph cluster to provision storage from.
-  # Should be unique across all Ceph clusters in use for provisioning,
-  # cannot be greater than 36 bytes in length, and should remain immutable for
-  # the lifetime of the StorageClass in use.
-  clusterID: <cluster-ID>
+  # (required unless clusterTopologyConfigMap is set) String representing a Ceph
+  # cluster to provision storage from. Should be unique across all Ceph clusters
+  # in use for provisioning, cannot be greater than 36 bytes in length, and
+  # should remain immutable for the lifetime of the StorageClass in use.
+  clusterID: ""
+
+  # (optional) ConfigMap name containing clusterTopology entries used for
+  # topology-based cluster selection. The ConfigMap must live in the same
+  # namespace as the Ceph-CSI pods and provide a config.json with
+  # `clusterTopology` entries.
+  clusterTopologyConfigMap: ""
 
   # (optional) If you want to use erasure coded pool with RBD, you need to
   # create two pools. one erasure coded and one replicated.

internal/rbd/controllerserver.go

@@ -80,14 +80,17 @@ func (cs *ControllerServer) validateVolumeReq(ctx context.Context, req *csi.Crea
 	}
 	options := req.GetParameters()
 	if value, ok := options["clusterID"]; !ok || value == "" {
-		return status.Error(codes.InvalidArgument, "empty cluster ID to provision volume from")
+		if _, ok := options["clusterTopologyConfigMap"]; !ok {
+			return status.Error(codes.InvalidArgument, "empty cluster ID to provision volume from")
+		}
 	}
 	poolValue, poolOK := options["pool"]
 	topologyConstrainedPoolsValue, topologyOK := options["topologyConstrainedPools"]
+	_, clusterTopologyOK := options["clusterTopologyConfigMap"]
 	if !poolOK {
 		if topologyOK && topologyConstrainedPoolsValue == "" {
 			return status.Error(codes.InvalidArgument, "empty pool name or topologyConstrainedPools to provision volume")
-		} else if !topologyOK {
+		} else if !topologyOK && !clusterTopologyOK {
 			return status.Error(codes.InvalidArgument, "missing or empty pool name to provision volume from")
 		}
 	} else if poolValue == "" {
@@ -154,7 +157,6 @@ func validateStriping(parameters map[string]string) error {
 func (cs *ControllerServer) parseVolCreateRequest(
 	ctx context.Context,
 	req *csi.CreateVolumeRequest,
-	cr *util.Credentials,
 ) (*rbdVolume, error) {
 	// TODO (sbezverk) Last check for not exceeding total storage capacity
 
@@ -229,17 +231,22 @@ func (cs *ControllerServer) parseVolCreateRequest(
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
+	// store cluster topology information from the request if present
+	rbdVol.ClusterTopologies, _, err = util.GetClusterTopologiesFromRequest(req)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
 	// parse QOS parameters from mutable parameters
 	err = rbdVol.SetQOS(ctx, req.GetMutableParameters())
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	err = rbdVol.Connect(cr)
+	// Get QosParameters from SC if qos configuration existing in SC
+	err = rbdVol.SetQOS(ctx, req.GetParameters())
 	if err != nil {
-		log.ErrorLog(ctx, "failed to connect to volume %v: %v", rbdVol.RbdImageName, err)
-
-		return nil, status.Error(codes.Internal, err.Error())
+		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
 	// NOTE: rbdVol does not contain VolID and RbdImageName populated, everything
@@ -277,6 +284,10 @@ func (rbdVol *rbdVolume) ToCSI(ctx context.Context) (*csi.Volume, error) {
 		vol.VolumeContext["dataPool"] = rbdVol.DataPool
 	}
 
+	if rbdVol.ClusterSecretName != "" {
+		vol.VolumeContext["clusterSecretName"] = rbdVol.ClusterSecretName
+	}
+
 	if rbdVol.Topology != nil {
 		vol.AccessibleTopology = []*csi.Topology{
 			{
@@ -363,19 +374,48 @@ func (cs *ControllerServer) CreateVolume(
 		return nil, err
 	}
 
+	rbdVol, err := cs.parseVolCreateRequest(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+	defer rbdVol.Destroy(ctx)
+
+	selectedCluster := util.ClusterTopology{}
+	if rbdVol.ClusterTopologies != nil {
+		selectedCluster, _, err = util.FindClusterAndTopology(rbdVol.ClusterTopologies, rbdVol.TopologyRequirement)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+		if selectedCluster.ClusterID == "" {
+			return nil, status.Error(codes.InvalidArgument, "no matching cluster found for provided topology requirements")
+		}
+		// persist selected secret for volume context
+		rbdVol.ClusterSecretName = selectedCluster.SecretName
+	}
+
+	secrets := req.GetSecrets()
+	if len(secrets) == 0 && selectedCluster.SecretName != "" {
+		namespace, nsErr := util.GetPodNamespace()
+		if nsErr != nil {
+			return nil, status.Error(codes.InvalidArgument, nsErr.Error())
+		}
+		secrets, err = k8s.GetSecret(selectedCluster.SecretName, namespace)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, err.Error())
+		}
+	}
+	if len(secrets) == 0 {
+		return nil, status.Error(codes.InvalidArgument, "missing credentials for provisioning")
+	}
+
 	// TODO: create/get a connection from the ConnPool, and do not pass the
 	// credentials to any of the utility functions.
 
-	cr, err := util.NewUserCredentialsWithMigration(req.GetSecrets())
+	cr, err := util.NewUserCredentialsWithMigration(secrets)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 	defer cr.DeleteCredentials()
-	rbdVol, err := cs.parseVolCreateRequest(ctx, req, cr)
-	if err != nil {
-		return nil, err
-	}
-	defer rbdVol.Destroy(ctx)
 	// Existence and conflict checks
 	if acquired := cs.VolumeLocks.TryAcquire(req.GetName()); !acquired {
 		log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, req.GetName())
@@ -400,6 +440,13 @@ func (cs *ControllerServer) CreateVolume(
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	err = rbdVol.Connect(cr)
+	if err != nil {
+		log.ErrorLog(ctx, "failed to connect to volume %v: %v", rbdVol.RbdImageName, err)
+
+		return nil, status.Error(codes.Internal, err.Error())
+	}
+
 	found, err := rbdVol.Exists(ctx, parentVol)
 	if err != nil {
 		return nil, getGRPCErrorForCreateVolume(err)
@@ -869,8 +916,8 @@ func checkContentSource(
 			return nil, nil, status.Error(codes.NotFound, "volume cannot be empty")
 		}
 		volID := vol.GetVolumeId()
-		if err := util.ValidateVolumeID(volID, true); err != nil {
-			return nil, nil, status.Error(codes.InvalidArgument, err.Error())
+		if volID == "" {
+			return nil, nil, status.Errorf(codes.NotFound, "volume ID cannot be empty")
 		}
 		rbdvol, err := GenVolFromVolID(ctx, volID, cr, req.GetSecrets())
 		if err != nil {
@@ -958,8 +1005,8 @@ func (cs *ControllerServer) DeleteVolume(
 
 	// For now the image get unconditionally deleted, but here retention policy can be checked
 	volumeID := req.GetVolumeId()
-	if err := util.ValidateVolumeID(volumeID, true); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
+	if volumeID == "" {
+		return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
 	}
 
 	cr, err := util.NewUserCredentialsWithMigration(req.GetSecrets())
@@ -1122,8 +1169,8 @@ func (cs *ControllerServer) ValidateVolumeCapabilities(
 	ctx context.Context,
 	req *csi.ValidateVolumeCapabilitiesRequest,
 ) (*csi.ValidateVolumeCapabilitiesResponse, error) {
-	if err := util.ValidateVolumeID(req.GetVolumeId(), util.IsStaticVol(req.GetVolumeContext())); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
+	if req.GetVolumeId() == "" {
+		return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
 	}
 
 	if len(req.GetVolumeCapabilities()) == 0 {
@@ -1596,8 +1643,8 @@ func (cs *ControllerServer) ControllerExpandVolume(
 	}
 
 	volID := req.GetVolumeId()
-	if err := util.ValidateVolumeID(volID, true); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
+	if volID == "" {
+		return nil, status.Error(codes.InvalidArgument, "volume ID cannot be empty")
 	}
 
 	capRange := req.GetCapacityRange()
@@ -1778,8 +1825,8 @@ func (cs *ControllerServer) ControllerUnpublishVolume(
 	if !k8s.RunsOnKubernetes() {
 		return &csi.ControllerUnpublishVolumeResponse{}, nil
 	}
-	if err := util.ValidateVolumeID(req.GetVolumeId(), true); err != nil {
-		return nil, status.Error(codes.InvalidArgument, err.Error())
+	if req.GetVolumeId() == "" {
+		return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
 	}
 
 	volumeId := req.GetVolumeId()

internal/rbd/rbd_journal.go

@@ -472,6 +472,29 @@ func updateTopologyConstraints(rbdVol *rbdVolume, rbdSnap *rbdSnapshot) error {
 
 		return nil
 	}
+	if rbdVol.ClusterTopologies != nil {
+		cluster, topology, err := util.FindClusterAndTopology(rbdVol.ClusterTopologies, rbdVol.TopologyRequirement)
+		if err != nil {
+			return err
+		}
+		if cluster.ClusterID == "" {
+			return fmt.Errorf("no matching cluster found for provided topology requirements")
+		}
+		rbdVol.ClusterID = cluster.ClusterID
+		rbdVol.Monitors = cluster.Monitors
+		rbdVol.Pool = cluster.Pool
+		rbdVol.DataPool = cluster.DataPool
+		rbdVol.JournalPool = cluster.Pool
+		rbdVol.Topology = topology
+		rbdVol.ClusterSecretName = cluster.SecretName
+		rbdVol.RadosNamespace, err = util.GetRBDRadosNamespace(util.CsiConfigFile, rbdVol.ClusterID)
+		if err != nil {
+			return err
+		}
+
+		return nil
+	}
+
 	// update request based on topology constrained parameters (if present)
 	poolName, dataPoolName, topology, err := util.FindPoolAndTopology(rbdVol.TopologyPools, rbdVol.TopologyRequirement)
 	if err != nil {
@@ -667,7 +690,7 @@ func RegenerateJournal(
 			}
 		}
 		// Update Metadata on reattach of the same old PV
-		parameters := k8s.PrepareVolumeMetadata(claimName, owner, requestName)
+		parameters := k8s.PrepareVolumeMetadata(claimName, owner, "")
 		err = rbdVol.setAllMetadata(parameters)
 		if err != nil {
 			return "", fmt.Errorf("failed to set volume metadata: %w", err)

internal/rbd/rbd_util.go

@@ -183,6 +183,7 @@ type rbdVolume struct {
 	// VolName and MonValueFromSecret are retained from older plugin versions (<= 1.0.0)
 	// for backward compatibility reasons
 	TopologyPools       *[]util.TopologyConstrainedPool
+	ClusterTopologies   *[]util.ClusterTopology
 	TopologyRequirement *csi.TopologyRequirement
 	Topology            map[string]string
 	// DataPool is where the data for images in `Pool` are stored, this is used as the `--data-pool`
@@ -197,6 +198,7 @@ type rbdVolume struct {
 	LogStrategy        string
 	VolName            string
 	MonValueFromSecret string
+	ClusterSecretName  string
 	// Network namespace file path to execute nsenter command
 	NetNamespaceFilePath string
 	// RequestedVolSize has the size of the volume requested by the user and
@@ -1159,7 +1161,7 @@ func genSnapFromSnapID(
 	}()
 
 	if imageAttributes.KmsID != "" && imageAttributes.EncryptionType == crypto.EncryptionTypeBlock {
-		err = rbdSnap.configureBlockEncryption(imageAttributes.KmsID, secrets, nil)
+		err = rbdSnap.configureBlockEncryption(imageAttributes.KmsID, secrets)
 		if err != nil {
 			return rbdSnap, fmt.Errorf("failed to configure block encryption for "+
 				"%q: %w", rbdSnap, err)
@@ -1261,7 +1263,7 @@ func generateVolumeFromVolumeID(
 	rbdVol.Owner = imageAttributes.Owner
 
 	if imageAttributes.KmsID != "" && imageAttributes.EncryptionType == crypto.EncryptionTypeBlock {
-		err = rbdVol.configureBlockEncryption(imageAttributes.KmsID, secrets, nil)
+		err = rbdVol.configureBlockEncryption(imageAttributes.KmsID, secrets)
 		if err != nil {
 			return rbdVol, err
 		}
@@ -1440,7 +1442,9 @@ func genVolFromVolumeOptions(
 	rbdVol.Pool, ok = volOptions["pool"]
 	if !ok {
 		if _, ok = volOptions["topologyConstrainedPools"]; !ok {
-			return nil, errors.New("empty pool name or topologyConstrainedPools to provision volume")
+			if _, ok = volOptions["clusterTopologyConfigMap"]; !ok {
+				return nil, errors.New("empty pool name, topologyConstrainedPools, or clusterTopologyConfigMap to provision volume")
+			}
 		}
 	}
 
@@ -1451,18 +1455,24 @@ func genVolFromVolumeOptions(
 
 	clusterID, err := util.GetClusterID(volOptions)
 	if err != nil {
-		return nil, err
+		if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
+			return nil, err
+		}
 	}
 	rbdVol.Monitors, rbdVol.ClusterID, err = util.GetMonsAndClusterID(ctx, clusterID, checkClusterIDMapping)
 	if err != nil {
-		log.ErrorLog(ctx, "failed getting mons (%s)", err)
+		if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
+			log.ErrorLog(ctx, "failed getting mons (%s)", err)
 
-		return nil, err
+			return nil, err
+		}
 	}
 
 	rbdVol.RadosNamespace, err = util.GetRBDRadosNamespace(util.CsiConfigFile, rbdVol.ClusterID)
 	if err != nil {
-		return nil, err
+		if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
+			return nil, err
+		}
 	}
 	if rbdVol.Mounter, ok = volOptions["mounter"]; !ok {
 		rbdVol.Mounter = rbdDefaultMounter

internal/util/podnamespace.go

@@ -0,0 +1,22 @@
+package util
+
+import (
+	"fmt"
+	"os"
+)
+
+const (
+	// podNamespaceEnv ENV should be set in the cephcsi container.
+	podNamespaceEnv = "POD_NAMESPACE"
+)
+
+// GetPodNamespace reads the POD_NAMESPACE environment variable to discover the
+// namespace the driver pod is running in.
+func GetPodNamespace() (string, error) {
+	ns := os.Getenv(podNamespaceEnv)
+	if ns == "" {
+		return "", fmt.Errorf("%q is not set in the environment", podNamespaceEnv)
+	}
+
+	return ns, nil
+}