Skip to content

Commit a217700

Browse files
dridge93dridge93
committed
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>
1 parent ee2f643 commit a217700

9 files changed

Lines changed: 273 additions & 38 deletions

File tree

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,12 @@ metadata:
1515
{{- with .Values.commonLabels }}{{ toYaml . | trim | nindent 4 }}{{- end }}
1616
provisioner: {{ .Values.driverName }}
1717
parameters:
18+
{{- if .Values.storageClass.clusterID }}
1819
clusterID: {{ .Values.storageClass.clusterID }}
20+
{{- end }}
21+
{{- if .Values.storageClass.clusterTopologyConfigMap }}
22+
clusterTopologyConfigMap: {{ .Values.storageClass.clusterTopologyConfigMap }}
23+
{{- end }}
1924
imageFeatures: {{ .Values.storageClass.imageFeatures }}
2025
{{- if .Values.storageClass.pool }}
2126
pool: {{ .Values.storageClass.pool }}

‎charts/ceph-csi-rbd/values.yaml‎

Lines changed: 11 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -365,11 +365,17 @@ storageClass:
365365
# storageclass.kubernetes.io/is-default-class: "true"
366366
annotations: {}
367367

368-
# (required) String representing a Ceph cluster to provision storage from.
369-
# Should be unique across all Ceph clusters in use for provisioning,
370-
# cannot be greater than 36 bytes in length, and should remain immutable for
371-
# the lifetime of the StorageClass in use.
372-
clusterID: <cluster-ID>
368+
# (required unless clusterTopologyConfigMap is set) String representing a Ceph
369+
# cluster to provision storage from. Should be unique across all Ceph clusters
370+
# in use for provisioning, cannot be greater than 36 bytes in length, and
371+
# should remain immutable for the lifetime of the StorageClass in use.
372+
clusterID: ""
373+
374+
# (optional) ConfigMap name containing clusterTopology entries used for
375+
# topology-based cluster selection. The ConfigMap must live in the same
376+
# namespace as the Ceph-CSI pods and provide a config.json with
377+
# `clusterTopology` entries.
378+
clusterTopologyConfigMap: ""
373379

374380
# (optional) If you want to use erasure coded pool with RBD, you need to
375381
# create two pools. one erasure coded and one replicated.

‎internal/rbd/controllerserver.go‎

Lines changed: 74 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -81,14 +81,17 @@ func (cs *ControllerServer) validateVolumeReq(ctx context.Context, req *csi.Crea
8181
}
8282
options := req.GetParameters()
8383
if value, ok := options["clusterID"]; !ok || value == "" {
84-
return status.Error(codes.InvalidArgument, "empty cluster ID to provision volume from")
84+
if _, ok := options["clusterTopologyConfigMap"]; !ok {
85+
return status.Error(codes.InvalidArgument, "empty cluster ID to provision volume from")
86+
}
8587
}
8688
poolValue, poolOK := options["pool"]
8789
topologyConstrainedPoolsValue, topologyOK := options["topologyConstrainedPools"]
90+
_, clusterTopologyOK := options["clusterTopologyConfigMap"]
8891
if !poolOK {
8992
if topologyOK && topologyConstrainedPoolsValue == "" {
9093
return status.Error(codes.InvalidArgument, "empty pool name or topologyConstrainedPools to provision volume")
91-
} else if !topologyOK {
94+
} else if !topologyOK && !clusterTopologyOK {
9295
return status.Error(codes.InvalidArgument, "missing or empty pool name to provision volume from")
9396
}
9497
} else if poolValue == "" {
@@ -155,7 +158,6 @@ func validateStriping(parameters map[string]string) error {
155158
func (cs *ControllerServer) parseVolCreateRequest(
156159
ctx context.Context,
157160
req *csi.CreateVolumeRequest,
158-
cr *util.Credentials,
159161
) (*rbdVolume, error) {
160162
// TODO (sbezverk) Last check for not exceeding total storage capacity
161163

@@ -230,17 +232,26 @@ func (cs *ControllerServer) parseVolCreateRequest(
230232
return nil, status.Error(codes.InvalidArgument, err.Error())
231233
}
232234

235+
// store cluster topology information from the request if present
236+
var clusterTopologyRequirement *csi.TopologyRequirement
237+
rbdVol.ClusterTopologies, clusterTopologyRequirement, err = util.GetClusterTopologiesFromRequest(req)
238+
if err != nil {
239+
return nil, status.Error(codes.InvalidArgument, err.Error())
240+
}
241+
if rbdVol.TopologyRequirement == nil {
242+
rbdVol.TopologyRequirement = clusterTopologyRequirement
243+
}
244+
233245
// parse QOS parameters from mutable parameters
234246
err = rbdVol.SetQOS(ctx, req.GetMutableParameters())
235247
if err != nil {
236248
return nil, status.Error(codes.InvalidArgument, err.Error())
237249
}
238250

239-
err = rbdVol.Connect(cr)
251+
// Get QosParameters from SC if qos configuration existing in SC
252+
err = rbdVol.SetQOS(ctx, req.GetParameters())
240253
if err != nil {
241-
log.ErrorLog(ctx, "failed to connect to volume %v: %v", rbdVol.RbdImageName, err)
242-
243-
return nil, status.Error(codes.Internal, err.Error())
254+
return nil, status.Error(codes.InvalidArgument, err.Error())
244255
}
245256

246257
// NOTE: rbdVol does not contain VolID and RbdImageName populated, everything
@@ -278,6 +289,10 @@ func (rbdVol *rbdVolume) ToCSI(ctx context.Context) (*csi.Volume, error) {
278289
vol.VolumeContext["dataPool"] = rbdVol.DataPool
279290
}
280291

292+
if rbdVol.ClusterSecretName != "" {
293+
vol.VolumeContext["clusterSecretName"] = rbdVol.ClusterSecretName
294+
}
295+
281296
if rbdVol.Topology != nil {
282297
vol.AccessibleTopology = []*csi.Topology{
283298
{
@@ -362,19 +377,48 @@ func (cs *ControllerServer) CreateVolume(
362377
return nil, err
363378
}
364379

380+
rbdVol, err := cs.parseVolCreateRequest(ctx, req)
381+
if err != nil {
382+
return nil, err
383+
}
384+
defer rbdVol.Destroy(ctx)
385+
386+
selectedCluster := util.ClusterTopology{}
387+
if rbdVol.ClusterTopologies != nil {
388+
selectedCluster, _, err = util.FindClusterAndTopology(rbdVol.ClusterTopologies, rbdVol.TopologyRequirement)
389+
if err != nil {
390+
return nil, status.Error(codes.InvalidArgument, err.Error())
391+
}
392+
if selectedCluster.ClusterID == "" {
393+
return nil, status.Error(codes.InvalidArgument, "no matching cluster found for provided topology requirements")
394+
}
395+
// persist selected secret for volume context
396+
rbdVol.ClusterSecretName = selectedCluster.SecretName
397+
}
398+
399+
secrets := req.GetSecrets()
400+
if len(secrets) == 0 && selectedCluster.SecretName != "" {
401+
namespace, nsErr := util.GetPodNamespace()
402+
if nsErr != nil {
403+
return nil, status.Error(codes.InvalidArgument, nsErr.Error())
404+
}
405+
secrets, err = k8s.GetSecret(selectedCluster.SecretName, namespace)
406+
if err != nil {
407+
return nil, status.Error(codes.InvalidArgument, err.Error())
408+
}
409+
}
410+
if len(secrets) == 0 {
411+
return nil, status.Error(codes.InvalidArgument, "missing credentials for provisioning")
412+
}
413+
365414
// TODO: create/get a connection from the ConnPool, and do not pass the
366415
// credentials to any of the utility functions.
367416

368-
cr, err := util.NewUserCredentialsWithMigration(req.GetSecrets())
417+
cr, err := util.NewUserCredentialsWithMigration(secrets)
369418
if err != nil {
370419
return nil, status.Error(codes.InvalidArgument, err.Error())
371420
}
372421
defer cr.DeleteCredentials()
373-
rbdVol, err := cs.parseVolCreateRequest(ctx, req, cr)
374-
if err != nil {
375-
return nil, err
376-
}
377-
defer rbdVol.Destroy(ctx)
378422
// Existence and conflict checks
379423
if acquired := cs.VolumeLocks.TryAcquire(req.GetName()); !acquired {
380424
log.ErrorLog(ctx, util.VolumeOperationAlreadyExistsFmt, req.GetName())
@@ -399,6 +443,13 @@ func (cs *ControllerServer) CreateVolume(
399443
return nil, status.Error(codes.Internal, err.Error())
400444
}
401445

446+
err = rbdVol.Connect(cr)
447+
if err != nil {
448+
log.ErrorLog(ctx, "failed to connect to volume %v: %v", rbdVol.RbdImageName, err)
449+
450+
return nil, status.Error(codes.Internal, err.Error())
451+
}
452+
402453
found, err := rbdVol.Exists(ctx, parentVol)
403454
if err != nil {
404455
return nil, getGRPCErrorForCreateVolume(err)
@@ -868,8 +919,8 @@ func checkContentSource(
868919
return nil, nil, status.Error(codes.NotFound, "volume cannot be empty")
869920
}
870921
volID := vol.GetVolumeId()
871-
if err := util.ValidateVolumeID(volID, true); err != nil {
872-
return nil, nil, status.Error(codes.InvalidArgument, err.Error())
922+
if volID == "" {
923+
return nil, nil, status.Errorf(codes.NotFound, "volume ID cannot be empty")
873924
}
874925
rbdvol, err := GenVolFromVolID(ctx, volID, cr, req.GetSecrets())
875926
if err != nil {
@@ -957,8 +1008,8 @@ func (cs *ControllerServer) DeleteVolume(
9571008

9581009
// For now the image get unconditionally deleted, but here retention policy can be checked
9591010
volumeID := req.GetVolumeId()
960-
if err := util.ValidateVolumeID(volumeID, true); err != nil {
961-
return nil, status.Error(codes.InvalidArgument, err.Error())
1011+
if volumeID == "" {
1012+
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
9621013
}
9631014

9641015
cr, err := util.NewUserCredentialsWithMigration(req.GetSecrets())
@@ -1121,8 +1172,8 @@ func (cs *ControllerServer) ValidateVolumeCapabilities(
11211172
ctx context.Context,
11221173
req *csi.ValidateVolumeCapabilitiesRequest,
11231174
) (*csi.ValidateVolumeCapabilitiesResponse, error) {
1124-
if err := util.ValidateVolumeID(req.GetVolumeId(), util.IsStaticVol(req.GetVolumeContext())); err != nil {
1125-
return nil, status.Error(codes.InvalidArgument, err.Error())
1175+
if req.GetVolumeId() == "" {
1176+
return nil, status.Error(codes.InvalidArgument, "empty volume ID in request")
11261177
}
11271178

11281179
if len(req.GetVolumeCapabilities()) == 0 {
@@ -1595,8 +1646,8 @@ func (cs *ControllerServer) ControllerExpandVolume(
15951646
}
15961647

15971648
volID := req.GetVolumeId()
1598-
if err := util.ValidateVolumeID(volID, true); err != nil {
1599-
return nil, status.Error(codes.InvalidArgument, err.Error())
1649+
if volID == "" {
1650+
return nil, status.Error(codes.InvalidArgument, "volume ID cannot be empty")
16001651
}
16011652

16021653
capRange := req.GetCapacityRange()
@@ -1777,8 +1828,8 @@ func (cs *ControllerServer) ControllerUnpublishVolume(
17771828
if !k8s.RunsOnKubernetes() {
17781829
return &csi.ControllerUnpublishVolumeResponse{}, nil
17791830
}
1780-
if err := util.ValidateVolumeID(req.GetVolumeId(), true); err != nil {
1781-
return nil, status.Error(codes.InvalidArgument, err.Error())
1831+
if req.GetVolumeId() == "" {
1832+
return nil, status.Error(codes.InvalidArgument, "Volume ID cannot be empty")
17821833
}
17831834

17841835
volumeId := req.GetVolumeId()

‎internal/rbd/nodeserver.go‎

Lines changed: 16 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -352,7 +352,22 @@ func (ns *NodeServer) NodeStageVolume(
352352
}
353353

354354
volID := req.GetVolumeId()
355-
cr, err := util.NewUserCredentialsWithMigration(req.GetSecrets())
355+
secrets := req.GetSecrets()
356+
if len(secrets) == 0 {
357+
clusterSecretName := req.GetVolumeContext()["clusterSecretName"]
358+
if clusterSecretName != "" {
359+
namespace, nsErr := util.GetPodNamespace()
360+
if nsErr != nil {
361+
return nil, status.Error(codes.InvalidArgument, nsErr.Error())
362+
}
363+
secrets, err = k8s.GetSecret(clusterSecretName, namespace)
364+
if err != nil {
365+
return nil, status.Error(codes.InvalidArgument, err.Error())
366+
}
367+
}
368+
}
369+
370+
cr, err := util.NewUserCredentialsWithMigration(secrets)
356371
if err != nil {
357372
return nil, status.Error(codes.InvalidArgument, err.Error())
358373
}

‎internal/rbd/rbd_journal.go‎

Lines changed: 24 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -472,6 +472,29 @@ func updateTopologyConstraints(rbdVol *rbdVolume, rbdSnap *rbdSnapshot) error {
472472

473473
return nil
474474
}
475+
if rbdVol.ClusterTopologies != nil {
476+
cluster, topology, err := util.FindClusterAndTopology(rbdVol.ClusterTopologies, rbdVol.TopologyRequirement)
477+
if err != nil {
478+
return err
479+
}
480+
if cluster.ClusterID == "" {
481+
return fmt.Errorf("no matching cluster found for provided topology requirements")
482+
}
483+
rbdVol.ClusterID = cluster.ClusterID
484+
rbdVol.Monitors = cluster.Monitors
485+
rbdVol.Pool = cluster.Pool
486+
rbdVol.DataPool = cluster.DataPool
487+
rbdVol.JournalPool = cluster.Pool
488+
rbdVol.Topology = topology
489+
rbdVol.ClusterSecretName = cluster.SecretName
490+
rbdVol.RadosNamespace, err = util.GetRBDRadosNamespace(util.CsiConfigFile, rbdVol.ClusterID)
491+
if err != nil {
492+
return err
493+
}
494+
495+
return nil
496+
}
497+
475498
// update request based on topology constrained parameters (if present)
476499
poolName, dataPoolName, topology, err := util.FindPoolAndTopology(rbdVol.TopologyPools, rbdVol.TopologyRequirement)
477500
if err != nil {
@@ -667,7 +690,7 @@ func RegenerateJournal(
667690
}
668691
}
669692
// Update Metadata on reattach of the same old PV
670-
parameters := k8s.PrepareVolumeMetadata(claimName, owner, requestName)
693+
parameters := k8s.PrepareVolumeMetadata(claimName, owner, "")
671694
err = rbdVol.setAllMetadata(parameters)
672695
if err != nil {
673696
return "", fmt.Errorf("failed to set volume metadata: %w", err)

‎internal/rbd/rbd_util.go‎

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -183,6 +183,7 @@ type rbdVolume struct {
183183
// VolName and MonValueFromSecret are retained from older plugin versions (<= 1.0.0)
184184
// for backward compatibility reasons
185185
TopologyPools *[]util.TopologyConstrainedPool
186+
ClusterTopologies *[]util.ClusterTopology
186187
TopologyRequirement *csi.TopologyRequirement
187188
Topology map[string]string
188189
// 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 {
197198
LogStrategy string
198199
VolName string
199200
MonValueFromSecret string
201+
ClusterSecretName string
200202
// Network namespace file path to execute nsenter command
201203
NetNamespaceFilePath string
202204
// RequestedVolSize has the size of the volume requested by the user and
@@ -1159,7 +1161,7 @@ func genSnapFromSnapID(
11591161
}()
11601162

11611163
if imageAttributes.KmsID != "" && imageAttributes.EncryptionType == crypto.EncryptionTypeBlock {
1162-
err = rbdSnap.configureBlockEncryption(imageAttributes.KmsID, secrets, nil)
1164+
err = rbdSnap.configureBlockEncryption(imageAttributes.KmsID, secrets)
11631165
if err != nil {
11641166
return rbdSnap, fmt.Errorf("failed to configure block encryption for "+
11651167
"%q: %w", rbdSnap, err)
@@ -1261,7 +1263,7 @@ func generateVolumeFromVolumeID(
12611263
rbdVol.Owner = imageAttributes.Owner
12621264

12631265
if imageAttributes.KmsID != "" && imageAttributes.EncryptionType == crypto.EncryptionTypeBlock {
1264-
err = rbdVol.configureBlockEncryption(imageAttributes.KmsID, secrets, nil)
1266+
err = rbdVol.configureBlockEncryption(imageAttributes.KmsID, secrets)
12651267
if err != nil {
12661268
return rbdVol, err
12671269
}
@@ -1440,7 +1442,9 @@ func genVolFromVolumeOptions(
14401442
rbdVol.Pool, ok = volOptions["pool"]
14411443
if !ok {
14421444
if _, ok = volOptions["topologyConstrainedPools"]; !ok {
1443-
return nil, errors.New("empty pool name or topologyConstrainedPools to provision volume")
1445+
if _, ok = volOptions["clusterTopologyConfigMap"]; !ok {
1446+
return nil, errors.New("empty pool name, topologyConstrainedPools, or clusterTopologyConfigMap to provision volume")
1447+
}
14441448
}
14451449
}
14461450

@@ -1451,18 +1455,24 @@ func genVolFromVolumeOptions(
14511455

14521456
clusterID, err := util.GetClusterID(volOptions)
14531457
if err != nil {
1454-
return nil, err
1458+
if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
1459+
return nil, err
1460+
}
14551461
}
14561462
rbdVol.Monitors, rbdVol.ClusterID, err = util.GetMonsAndClusterID(ctx, clusterID, checkClusterIDMapping)
14571463
if err != nil {
1458-
log.ErrorLog(ctx, "failed getting mons (%s)", err)
1464+
if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
1465+
log.ErrorLog(ctx, "failed getting mons (%s)", err)
14591466

1460-
return nil, err
1467+
return nil, err
1468+
}
14611469
}
14621470

14631471
rbdVol.RadosNamespace, err = util.GetRBDRadosNamespace(util.CsiConfigFile, rbdVol.ClusterID)
14641472
if err != nil {
1465-
return nil, err
1473+
if _, ok := volOptions["clusterTopologyConfigMap"]; !ok {
1474+
return nil, err
1475+
}
14661476
}
14671477
if rbdVol.Mounter, ok = volOptions["mounter"]; !ok {
14681478
rbdVol.Mounter = rbdDefaultMounter

‎internal/util/podnamespace.go‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,22 @@
1+
package util
2+
3+
import (
4+
"fmt"
5+
"os"
6+
)
7+
8+
const (
9+
// podNamespaceEnv ENV should be set in the cephcsi container.
10+
podNamespaceEnv = "POD_NAMESPACE"
11+
)
12+
13+
// GetPodNamespace reads the POD_NAMESPACE environment variable to discover the
14+
// namespace the driver pod is running in.
15+
func GetPodNamespace() (string, error) {
16+
ns := os.Getenv(podNamespaceEnv)
17+
if ns == "" {
18+
return "", fmt.Errorf("%q is not set in the environment", podNamespaceEnv)
19+
}
20+
21+
return ns, nil
22+
}

0 commit comments

Comments
 (0)