From 92ada526108f8bf93ff1672035725cf23568beff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Mon, 3 Jan 2022 18:24:38 +0100 Subject: [PATCH 1/3] nfs: use hashing function from existing k8sutil package MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We have a function to hash a string so let's use it. Signed-off-by: Sébastien Han --- pkg/operator/ceph/nfs/nfs.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/pkg/operator/ceph/nfs/nfs.go b/pkg/operator/ceph/nfs/nfs.go index 236140003..2da06a4b8 100644 --- a/pkg/operator/ceph/nfs/nfs.go +++ b/pkg/operator/ceph/nfs/nfs.go @@ -18,7 +18,6 @@ limitations under the License. package nfs import ( - "crypto/sha256" "fmt" "github.com/banzaicloud/k8s-objectmatcher/patch" @@ -215,13 +214,7 @@ func (r *ReconcileCephNFS) createConfigMap(n *cephv1.CephNFS, name string) (stri } } - h := sha256.New() - if _, err := h.Write([]byte(fmt.Sprintf("%v", configMap.Data))); err != nil { - return "", "", errors.Wrapf(err, "failed to get hash of ganesha config map") - } - configHash := fmt.Sprintf("%x", h.Sum(nil)) - - return configMap.Name, configHash, nil + return configMap.Name, k8sutil.Hash(fmt.Sprintf("%v", configMap.Data)), nil } // Down scale the ganesha server From 640d30339d51fb298be998191cc84426e1520ef1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Tue, 4 Jan 2022 10:36:48 +0100 Subject: [PATCH 2/3] subvolumegroup: fix code deletion MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We want to log properly when ENOTEMPTY is caught during deletion, so fixing the condition. Signed-off-by: Sébastien Han --- pkg/operator/ceph/file/subvolumegroup/controller.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go index c65434723..09895b9e6 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller.go +++ b/pkg/operator/ceph/file/subvolumegroup/controller.go @@ -245,8 +245,8 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesys code, ok := exec.ExitStatus(err) // If the subvolumegroup has subvolumes the command will fail with: // Error ENOTEMPTY: error in rmdir /volumes/csi - if ok && code != int(syscall.ENOTEMPTY) { - return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first.", cephFilesystemSubVolumeGroup.Name) + if ok && code == int(syscall.ENOTEMPTY) { + return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name) } return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name) From 739abf66627a531dcec04924b2b221520bea5d48 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?S=C3=A9bastien=20Han?= Date: Wed, 22 Dec 2021 14:57:34 +0100 Subject: [PATCH 3/3] csi: change rook-ceph-csi-config to expose clusterID for subvolume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The rook-ceph-csi-config config map now includes CephFS subvolumegroups when a subvolumegroup CR is created. It is useful for ceph-csi to understand which subvolumegroup to use for a given cluster. Signed-off-by: Sébastien Han --- pkg/operator/ceph/cluster/cluster.go | 5 +- pkg/operator/ceph/cluster/cluster_external.go | 2 +- pkg/operator/ceph/cluster/controller.go | 11 +- pkg/operator/ceph/cluster/mon/drain_test.go | 3 +- pkg/operator/ceph/cluster/mon/health_test.go | 13 +- pkg/operator/ceph/cluster/mon/mon.go | 8 +- pkg/operator/ceph/cluster/mon/mon_test.go | 7 +- pkg/operator/ceph/cluster/mon/node_test.go | 5 +- pkg/operator/ceph/cluster/mon/service_test.go | 3 +- pkg/operator/ceph/cluster/mon/spec_test.go | 4 - pkg/operator/ceph/csi/cluster_config.go | 86 +++++--- pkg/operator/ceph/csi/cluster_config_test.go | 189 ++++++++++++------ pkg/operator/ceph/csi/predicate.go | 3 +- .../ceph/file/subvolumegroup/controller.go | 40 +++- .../file/subvolumegroup/controller_test.go | 29 ++- 15 files changed, 279 insertions(+), 129 deletions(-) diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 3a6ffdb6b..c9cf926f4 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -22,7 +22,6 @@ import ( "fmt" "os/exec" "path" - "sync" "syscall" "github.com/pkg/errors" @@ -67,7 +66,7 @@ type clusterHealth struct { internalCancel context.CancelFunc } -func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync.Mutex, ownerInfo *k8sutil.OwnerInfo) *cluster { +func newCluster(c *cephv1.CephCluster, context *clusterd.Context, ownerInfo *k8sutil.OwnerInfo) *cluster { return &cluster{ // at this phase of the cluster creation process, the identity components of the cluster are // not yet established. we reserve this struct which is filled in as soon as the cluster's @@ -79,7 +78,7 @@ func newCluster(c *cephv1.CephCluster, context *clusterd.Context, csiMutex *sync namespacedName: types.NamespacedName{Namespace: c.Namespace, Name: c.Name}, monitoringRoutines: make(map[string]*clusterHealth), ownerInfo: ownerInfo, - mons: mon.New(context, c.Namespace, c.Spec, ownerInfo, csiMutex), + mons: mon.New(context, c.Namespace, c.Spec, ownerInfo), } } diff --git a/pkg/operator/ceph/cluster/cluster_external.go b/pkg/operator/ceph/cluster/cluster_external.go index 284d34a9a..fb97a9adf 100644 --- a/pkg/operator/ceph/cluster/cluster_external.go +++ b/pkg/operator/ceph/cluster/cluster_external.go @@ -111,7 +111,7 @@ func (c *ClusterController) configureExternalCephCluster(cluster *cluster) error } // Save CSI configmap - err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, c.csiConfigMutex) + err = csi.SaveClusterConfig(c.context.Clientset, c.namespacedName.Namespace, cluster.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(cluster.ClusterInfo.Monitors)}) if err != nil { return errors.Wrap(err, "failed to update csi cluster config") } diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index d232a821f..f1fa42874 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/pkg/operator/ceph/cluster/controller.go @@ -21,7 +21,6 @@ import ( "context" "fmt" "os" - "sync" "github.com/coreos/pkg/capnslog" "github.com/pkg/errors" @@ -85,7 +84,6 @@ type ClusterController struct { context *clusterd.Context rookImage string clusterMap map[string]*cluster - csiConfigMutex *sync.Mutex osdChecker *osd.OSDHealthMonitor client client.Client namespacedName types.NamespacedName @@ -327,10 +325,9 @@ func (r *ReconcileCephCluster) reconcileDelete(cephCluster *cephv1.CephCluster) // NewClusterController create controller for watching cluster custom resources created func NewClusterController(context *clusterd.Context, rookImage string) *ClusterController { return &ClusterController{ - context: context, - rookImage: rookImage, - clusterMap: make(map[string]*cluster), - csiConfigMutex: &sync.Mutex{}, + context: context, + rookImage: rookImage, + clusterMap: make(map[string]*cluster), } } @@ -343,7 +340,7 @@ func (c *ClusterController) reconcileCephCluster(clusterObj *cephv1.CephCluster, cluster, ok := c.clusterMap[clusterObj.Namespace] if !ok { // It's a new cluster so let's populate the struct - cluster = newCluster(clusterObj, c.context, c.csiConfigMutex, ownerInfo) + cluster = newCluster(clusterObj, c.context, ownerInfo) } cluster.namespacedName = c.namespacedName diff --git a/pkg/operator/ceph/cluster/mon/drain_test.go b/pkg/operator/ceph/cluster/mon/drain_test.go index f64e1f6e7..2ff565d21 100644 --- a/pkg/operator/ceph/cluster/mon/drain_test.go +++ b/pkg/operator/ceph/cluster/mon/drain_test.go @@ -18,7 +18,6 @@ package mon import ( "context" - "sync" "testing" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -48,7 +47,7 @@ func createFakeCluster(t *testing.T, cephClusterObj *cephv1.CephCluster, k8sVers cl := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects().Build() clientset := test.New(t, 3) - c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo, &sync.Mutex{}) + c := New(&clusterd.Context{Client: cl, Clientset: clientset}, mockNamespace, cephClusterObj.Spec, ownerInfo) test.SetFakeKubernetesVersion(clientset, k8sVersion) return c } diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index f27a1a75b..7ba8724a8 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/pkg/operator/ceph/cluster/mon/health_test.go @@ -22,7 +22,6 @@ import ( "io/ioutil" "os" "reflect" - "sync" "testing" "time" @@ -66,7 +65,7 @@ func TestCheckHealth(t *testing.T) { Executor: executor, } ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo) // clusterInfo is nil so we return err err := c.checkHealth(ctx) assert.NotNil(t, err) @@ -155,7 +154,7 @@ func TestCheckHealth(t *testing.T) { } func TestSkipMonFailover(t *testing.T) { - c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil, nil) + c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil) c.ClusterInfo = clienttest.CreateTestClusterInfo(1) monName := "arb" @@ -201,7 +200,7 @@ func TestEvictMonOnSameNode(t *testing.T) { } context := &clusterd.Context{Clientset: clientset, ConfigDir: configDir, Executor: executor} ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0}, "myversion") c.maxMonID = 2 c.waitForStart = false @@ -256,7 +255,7 @@ func TestScaleMonDeployment(t *testing.T) { clientset := test.New(t, 1) context := &clusterd.Context{Clientset: clientset} ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0, AllowMultiplePerNode: true}, "myversion") name := "a" @@ -307,7 +306,7 @@ func TestCheckHealthNotFound(t *testing.T) { Executor: executor, } ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo) setCommonMonProperties(c, 2, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.waitForStart = false defer os.RemoveAll(c.context.ConfigDir) @@ -370,7 +369,7 @@ func TestAddRemoveMons(t *testing.T) { Executor: executor, } ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(context, "ns", cephv1.ClusterSpec{}, ownerInfo) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 5, AllowMultiplePerNode: true}, "myversion") c.maxMonID = 0 // "a" is max mon id c.waitForStart = false diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 070ce2edc..bf7988ba3 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -120,7 +120,6 @@ type Cluster struct { monTimeoutList map[string]time.Time mapping *Mapping ownerInfo *k8sutil.OwnerInfo - csiConfigMutex *sync.Mutex isUpgrade bool arbiterMon string } @@ -164,7 +163,7 @@ type SchedulingResult struct { } // New creates an instance of a mon cluster -func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo, csiConfigMutex *sync.Mutex) *Cluster { +func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, ownerInfo *k8sutil.OwnerInfo) *Cluster { return &Cluster{ context: context, spec: spec, @@ -175,8 +174,7 @@ func New(context *clusterd.Context, namespace string, spec cephv1.ClusterSpec, o mapping: &Mapping{ Schedule: map[string]*MonScheduleInfo{}, }, - ownerInfo: ownerInfo, - csiConfigMutex: csiConfigMutex, + ownerInfo: ownerInfo, } } @@ -1042,7 +1040,7 @@ func (c *Cluster) saveMonConfig() error { return errors.Wrap(err, "failed to write connection config for new mons") } - if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, c.csiConfigMutex); err != nil { + if err := csi.SaveClusterConfig(c.context.Clientset, c.Namespace, c.ClusterInfo, &csi.CsiClusterConfigEntry{Monitors: csi.MonEndpoints(c.ClusterInfo.Monitors)}); err != nil { return errors.Wrap(err, "failed to update csi cluster config") } diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index 14e6f37db..67fb0b4af 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/pkg/operator/ceph/cluster/mon/mon_test.go @@ -25,7 +25,6 @@ import ( "reflect" "strconv" "strings" - "sync" "testing" "time" @@ -273,7 +272,7 @@ func validateStart(t *testing.T, c *Cluster) { func TestPersistMons(t *testing.T) { clientset := test.New(t, 1) ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{Annotations: cephv1.AnnotationsSpec{cephv1.KeyClusterMetadata: cephv1.Annotations{"key": "value"}}}, ownerInfo) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") // Persist mon a @@ -302,7 +301,7 @@ func TestSaveMonEndpoints(t *testing.T) { configDir, _ := ioutil.TempDir("", "") defer os.RemoveAll(configDir) ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo) setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") // create the initial config map @@ -350,7 +349,7 @@ func TestMaxMonID(t *testing.T) { configDir, _ := ioutil.TempDir("", "") defer os.RemoveAll(configDir) ownerInfo := cephclient.NewMinimumOwnerInfoWithOwnerRef() - c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset, ConfigDir: configDir}, "ns", cephv1.ClusterSpec{}, ownerInfo) c.ClusterInfo = clienttest.CreateTestClusterInfo(1) // when the configmap is not found, the maxMonID is -1 diff --git a/pkg/operator/ceph/cluster/mon/node_test.go b/pkg/operator/ceph/cluster/mon/node_test.go index 8fa24cf59..98bf7fbe8 100644 --- a/pkg/operator/ceph/cluster/mon/node_test.go +++ b/pkg/operator/ceph/cluster/mon/node_test.go @@ -19,7 +19,6 @@ package mon import ( "context" "strings" - "sync" "testing" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -37,7 +36,7 @@ import ( func TestNodeAffinity(t *testing.T) { ctx := context.TODO() clientset := test.New(t, 4) - c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.spec.Placement = map[cephv1.KeyType]cephv1.Placement{} @@ -167,7 +166,7 @@ func TestPodMemory(t *testing.T) { func TestHostNetwork(t *testing.T) { clientset := test.New(t, 3) - c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "myversion") c.spec.Network.HostNetwork = true diff --git a/pkg/operator/ceph/cluster/mon/service_test.go b/pkg/operator/ceph/cluster/mon/service_test.go index 62b5dc574..c8cf31e36 100644 --- a/pkg/operator/ceph/cluster/mon/service_test.go +++ b/pkg/operator/ceph/cluster/mon/service_test.go @@ -18,7 +18,6 @@ package mon import ( "context" - "sync" "testing" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -33,7 +32,7 @@ import ( func TestCreateService(t *testing.T) { ctx := context.TODO() clientset := test.New(t, 1) - c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, &sync.Mutex{}) + c := New(&clusterd.Context{Clientset: clientset}, "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}) c.ClusterInfo = client.AdminTestClusterInfo("rook-ceph") m := &monConfig{ResourceName: "rook-ceph-mon-b", DaemonName: "b"} clusterIP, err := c.createService(m) diff --git a/pkg/operator/ceph/cluster/mon/spec_test.go b/pkg/operator/ceph/cluster/mon/spec_test.go index 2b213e2be..c1f58ad9f 100644 --- a/pkg/operator/ceph/cluster/mon/spec_test.go +++ b/pkg/operator/ceph/cluster/mon/spec_test.go @@ -17,7 +17,6 @@ limitations under the License. package mon import ( - "sync" "testing" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" @@ -48,7 +47,6 @@ func testPodSpec(t *testing.T, monID string, pvc bool) { "ns", cephv1.ClusterSpec{}, ownerInfo, - &sync.Mutex{}, ) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion") c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"} @@ -98,7 +96,6 @@ func TestDeploymentPVCSpec(t *testing.T) { "ns", cephv1.ClusterSpec{}, ownerInfo, - &sync.Mutex{}, ) setCommonMonProperties(c, 0, cephv1.MonSpec{Count: 3, AllowMultiplePerNode: true}, "rook/rook:myversion") c.spec.CephVersion = cephv1.CephVersionSpec{Image: "quay.io/ceph/ceph:myceph"} @@ -158,7 +155,6 @@ func testRequiredDuringScheduling(t *testing.T, hostNetwork, allowMultiplePerNod "ns", cephv1.ClusterSpec{}, &k8sutil.OwnerInfo{}, - &sync.Mutex{}, ) c.spec.Network.HostNetwork = hostNetwork diff --git a/pkg/operator/ceph/csi/cluster_config.go b/pkg/operator/ceph/csi/cluster_config.go index 03e76934f..3f4ad3b8c 100644 --- a/pkg/operator/ceph/csi/cluster_config.go +++ b/pkg/operator/ceph/csi/cluster_config.go @@ -33,15 +33,22 @@ import ( ) var ( - logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi") + logger = capnslog.NewPackageLogger("github.com/rook/rook", "ceph-csi") + configMutex sync.Mutex ) -type csiClusterConfigEntry struct { - ClusterID string `json:"clusterID"` - Monitors []string `json:"monitors"` +type CsiClusterConfigEntry struct { + ClusterID string `json:"clusterID"` + Monitors []string `json:"monitors"` + CephFS *CsiCephFSSpec `json:"cephFS,omitempty"` + RadosNamespace string `json:"radosNamespace,omitempty"` } -type csiClusterConfig []csiClusterConfigEntry +type CsiCephFSSpec struct { + SubvolumeGroup string `json:"subvolumeGroup,omitempty"` +} + +type csiClusterConfig []CsiClusterConfigEntry // FormatCsiClusterConfig returns a json-formatted string containing // the cluster-to-mon mapping required to configure ceph csi. @@ -79,7 +86,7 @@ func formatCsiClusterConfig(cc csiClusterConfig) (string, error) { return string(ccJson), nil } -func monEndpoints(mons map[string]*cephclient.MonInfo) []string { +func MonEndpoints(mons map[string]*cephclient.MonInfo) []string { endpoints := make([]string, 0) for _, m := range mons { endpoints = append(endpoints, m.Endpoint) @@ -89,32 +96,59 @@ func monEndpoints(mons map[string]*cephclient.MonInfo) []string { // updateCsiClusterConfig returns a json-formatted string containing // the cluster-to-mon mapping required to configure ceph csi. -func updateCsiClusterConfig( - curr, clusterKey string, mons map[string]*cephclient.MonInfo) (string, error) { - +func updateCsiClusterConfig(curr, clusterKey string, newCsiClusterConfigEntry *CsiClusterConfigEntry) (string, error) { var ( cc csiClusterConfig - centry csiClusterConfigEntry + centry CsiClusterConfigEntry found bool ) + cc, err := parseCsiClusterConfig(curr) if err != nil { return "", errors.Wrap(err, "failed to parse current csi cluster config") } + // Regardless of which controllers call updateCsiClusterConfig(), the values will be preserved since + // a lock is acquired for the update operation. So concurrent updates (rare event) will block and + // wait for the other update to complete. Monitors and Subvolumegroup will be updated + // independently and won't collide. for i, centry := range cc { if centry.ClusterID == clusterKey { - centry.Monitors = monEndpoints(mons) + // If the new entry is nil, this means the entry is being deleted so remove it from the list + if newCsiClusterConfigEntry == nil { + cc = append(cc[:i], cc[i+1:]...) + found = true + break + } + centry.Monitors = newCsiClusterConfigEntry.Monitors + if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" { + centry.CephFS = newCsiClusterConfigEntry.CephFS + } + if newCsiClusterConfigEntry.RadosNamespace != "" { + centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace + } found = true cc[i] = centry break } } if !found { - centry.ClusterID = clusterKey - centry.Monitors = monEndpoints(mons) - cc = append(cc, centry) + // If it's the first time we create the cluster, the entry does not exist, so the removal + // will fail with a dangling pointer + if newCsiClusterConfigEntry != nil { + centry.ClusterID = clusterKey + centry.Monitors = newCsiClusterConfigEntry.Monitors + // Add a condition not to fill with empty values + if newCsiClusterConfigEntry.CephFS != nil && newCsiClusterConfigEntry.CephFS.SubvolumeGroup != "" { + centry.CephFS = newCsiClusterConfigEntry.CephFS + } + if newCsiClusterConfigEntry.RadosNamespace != "" { + centry.RadosNamespace = newCsiClusterConfigEntry.RadosNamespace + } + cc = append(cc, centry) + } } + return formatCsiClusterConfig(cc) } @@ -155,24 +189,24 @@ func CreateCsiConfigMap(namespace string, clientset kubernetes.Interface, ownerI // value that is provided to ceph-csi uses in the storage class. // The locker l is typically a mutex and is used to prevent the config // map from being updated for multiple clusters simultaneously. -func SaveClusterConfig( - clientset kubernetes.Interface, clusterNamespace string, - clusterInfo *cephclient.ClusterInfo, l sync.Locker) error { +func SaveClusterConfig(clientset kubernetes.Interface, clusterNamespace string, clusterInfo *cephclient.ClusterInfo, newCsiClusterConfigEntry *CsiClusterConfigEntry) error { if !CSIEnabled() { + logger.Debug("skipping csi config update because csi is not enabled") return nil } - l.Lock() - defer l.Unlock() + + configMutex.Lock() + defer configMutex.Unlock() + // csi is deployed into the same namespace as the operator csiNamespace := os.Getenv(k8sutil.PodNamespaceEnvVar) if csiNamespace == "" { - return errors.Errorf("namespace value missing for %s", k8sutil.PodNamespaceEnvVar) + return errors.Errorf("namespace value missing for %q", k8sutil.PodNamespaceEnvVar) } - logger.Debugf("Using %+v for CSI ConfigMap Namespace", csiNamespace) + logger.Debugf("using %q for csi configmap namespace", csiNamespace) // fetch current ConfigMap contents - configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context, - ConfigName, metav1.GetOptions{}) + configMap, err := clientset.CoreV1().ConfigMaps(csiNamespace).Get(clusterInfo.Context, ConfigName, metav1.GetOptions{}) if err != nil { return errors.Wrap(err, "failed to fetch current csi config map") } @@ -182,8 +216,8 @@ func SaveClusterConfig( if currData == "" { currData = "[]" } - newData, err := updateCsiClusterConfig( - currData, clusterNamespace, clusterInfo.Monitors) + + newData, err := updateCsiClusterConfig(currData, clusterNamespace, newCsiClusterConfigEntry) if err != nil { return errors.Wrap(err, "failed to update csi config map data") } @@ -191,7 +225,7 @@ func SaveClusterConfig( // update ConfigMap with new contents if _, err := clientset.CoreV1().ConfigMaps(csiNamespace).Update(clusterInfo.Context, configMap, metav1.UpdateOptions{}); err != nil { - return errors.Wrapf(err, "failed to update csi config map") + return errors.Wrap(err, "failed to update csi config map") } return nil diff --git a/pkg/operator/ceph/csi/cluster_config_test.go b/pkg/operator/ceph/csi/cluster_config_test.go index afa53e074..08c9e1cad 100644 --- a/pkg/operator/ceph/csi/cluster_config_test.go +++ b/pkg/operator/ceph/csi/cluster_config_test.go @@ -19,71 +19,142 @@ package csi import ( "testing" - cephclient "github.com/rook/rook/pkg/daemon/ceph/client" "github.com/stretchr/testify/assert" ) func TestUpdateCsiClusterConfig(t *testing.T) { - // initialize an empty list & add a simple mons list - mons := map[string]*cephclient.MonInfo{ - "foo": {Name: "foo", Endpoint: "1.2.3.4:5000"}, + csiClusterConfigEntry := CsiClusterConfigEntry{ + Monitors: []string{"1.2.3.4:5000"}, } - s, err := updateCsiClusterConfig("[]", "alpha", mons) - assert.NoError(t, err) - assert.Equal(t, s, - `[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`) - - // add a 2nd mon to the current cluster - mons["bar"] = &cephclient.MonInfo{ - Name: "bar", Endpoint: "10.11.12.13:5000"} - s, err = updateCsiClusterConfig(s, "alpha", mons) - assert.NoError(t, err) - cc, err := parseCsiClusterConfig(s) - assert.NoError(t, err) - assert.Equal(t, len(cc), 1) - assert.Equal(t, cc[0].ClusterID, "alpha") - assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") - assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") - assert.Equal(t, len(cc[0].Monitors), 2) - - // add a 2nd cluster with 3 mons - mons2 := map[string]*cephclient.MonInfo{ - "flim": {Name: "flim", Endpoint: "20.1.1.1:5000"}, - "flam": {Name: "flam", Endpoint: "20.1.1.2:5000"}, - "blam": {Name: "blam", Endpoint: "20.1.1.3:5000"}, + csiClusterConfigEntry2 := CsiClusterConfigEntry{ + Monitors: []string{"20.1.1.1:5000", "20.1.1.2:5000", "20.1.1.3:5000"}, } - s, err = updateCsiClusterConfig(s, "beta", mons2) - assert.NoError(t, err) - cc, err = parseCsiClusterConfig(s) - assert.NoError(t, err) - assert.Equal(t, len(cc), 2) - assert.Equal(t, cc[0].ClusterID, "alpha") - assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") - assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") - assert.Equal(t, len(cc[0].Monitors), 2) - assert.Equal(t, cc[1].ClusterID, "beta") - assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000") - assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000") - assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000") - assert.Equal(t, len(cc[1].Monitors), 3) + csiClusterConfigEntry3 := CsiClusterConfigEntry{ + Monitors: []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"}, + CephFS: &CsiCephFSSpec{ + SubvolumeGroup: "mygroup", + }, + } + var s string + var err error - // remove a mon from the 2nd cluster - delete(mons2, "blam") - s, err = updateCsiClusterConfig(s, "beta", mons2) - assert.NoError(t, err) - cc, err = parseCsiClusterConfig(s) - assert.NoError(t, err) - assert.Equal(t, len(cc), 2) - assert.Equal(t, cc[0].ClusterID, "alpha") - assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") - assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") - assert.Equal(t, len(cc[0].Monitors), 2) - assert.Equal(t, cc[1].ClusterID, "beta") - assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000") - assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000") - assert.Equal(t, len(cc[1].Monitors), 2) + t.Run("add a simple mons list", func(t *testing.T) { + s, err = updateCsiClusterConfig("[]", "alpha", &csiClusterConfigEntry) + assert.NoError(t, err) + assert.Equal(t, s, + `[{"clusterID":"alpha","monitors":["1.2.3.4:5000"]}]`) + }) - // does it return error on garbage input? - _, err = updateCsiClusterConfig("qqq", "beta", mons2) - assert.Error(t, err) + t.Run("add a 2nd mon to the current cluster", func(t *testing.T) { + csiClusterConfigEntry.Monitors = append(csiClusterConfigEntry.Monitors, "10.11.12.13:5000") + s, err = updateCsiClusterConfig(s, "alpha", &csiClusterConfigEntry) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 1, len(cc)) + assert.Equal(t, "alpha", cc[0].ClusterID) + assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") + assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") + assert.Equal(t, 2, len(cc[0].Monitors)) + }) + + t.Run("add a 2nd cluster with 3 mons", func(t *testing.T) { + s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 2, len(cc)) + assert.Equal(t, "alpha", cc[0].ClusterID) + assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") + assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") + assert.Equal(t, 2, len(cc[0].Monitors)) + assert.Equal(t, "beta", cc[1].ClusterID) + assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000") + assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000") + assert.Contains(t, cc[1].Monitors, "20.1.1.3:5000") + assert.Equal(t, 3, len(cc[1].Monitors)) + }) + + t.Run("remove a mon from the 2nd cluster", func(t *testing.T) { + i := 2 + // Remove last element of the slice + csiClusterConfigEntry2.Monitors = append(csiClusterConfigEntry2.Monitors[:i], csiClusterConfigEntry2.Monitors[i+1:]...) + s, err = updateCsiClusterConfig(s, "beta", &csiClusterConfigEntry2) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 2, len(cc)) + assert.Equal(t, "alpha", cc[0].ClusterID) + assert.Contains(t, cc[0].Monitors, "1.2.3.4:5000") + assert.Contains(t, cc[0].Monitors, "10.11.12.13:5000") + assert.Equal(t, 2, len(cc[0].Monitors)) + assert.Equal(t, "beta", cc[1].ClusterID) + assert.Contains(t, cc[1].Monitors, "20.1.1.1:5000") + assert.Contains(t, cc[1].Monitors, "20.1.1.2:5000") + assert.Equal(t, 2, len(cc[1].Monitors)) + }) + + t.Run("add a 3rd cluster with subvolumegroup", func(t *testing.T) { + s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 3, len(cc)) + assert.Equal(t, "alpha", cc[0].ClusterID) + assert.Equal(t, 2, len(cc[0].Monitors)) + assert.Equal(t, "beta", cc[1].ClusterID) + assert.Equal(t, "baba", cc[2].ClusterID) + assert.Equal(t, "10.1.1.1:5000", cc[2].Monitors[0]) + assert.Equal(t, 3, len(cc[2].Monitors)) + assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup) + + }) + + t.Run("add a 4th mon to the 3rd cluster and subvolumegroup is preserved", func(t *testing.T) { + csiClusterConfigEntry3.Monitors = append(csiClusterConfigEntry3.Monitors, "10.11.12.13:5000") + s, err = updateCsiClusterConfig(s, "baba", &csiClusterConfigEntry3) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 3, len(cc)) + assert.Equal(t, 4, len(cc[2].Monitors)) + assert.Equal(t, "mygroup", cc[2].CephFS.SubvolumeGroup) + }) + + t.Run("remove subvolumegroup", func(t *testing.T) { + csiClusterConfigEntry3.CephFS.SubvolumeGroup = "" + s, err = updateCsiClusterConfig(s, "baba", nil) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 2, len(cc)) // we only have 2 clusters now + }) + + t.Run("add subvolumegroup and mons after", func(t *testing.T) { + csiClusterConfigEntry4 := CsiClusterConfigEntry{ + CephFS: &CsiCephFSSpec{ + SubvolumeGroup: "mygroup2", + }, + } + s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4) + assert.NoError(t, err) + cc, err := parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, 3, len(cc), cc) + assert.Equal(t, 0, len(cc[2].Monitors)) + assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup, cc) + + csiClusterConfigEntry4.Monitors = []string{"10.1.1.1:5000", "10.1.1.2:5000", "10.1.1.3:5000"} + s, err = updateCsiClusterConfig(s, "quatre", &csiClusterConfigEntry4) + assert.NoError(t, err) + cc, err = parseCsiClusterConfig(s) + assert.NoError(t, err) + assert.Equal(t, "mygroup2", cc[2].CephFS.SubvolumeGroup) + assert.Equal(t, 3, len(cc[2].Monitors)) + }) + + t.Run("does it return error on garbage input?", func(t *testing.T) { + _, err = updateCsiClusterConfig("qqq", "beta", &csiClusterConfigEntry2) + assert.Error(t, err) + }) } diff --git a/pkg/operator/ceph/csi/predicate.go b/pkg/operator/ceph/csi/predicate.go index 2742475e9..bfe4055e8 100644 --- a/pkg/operator/ceph/csi/predicate.go +++ b/pkg/operator/ceph/csi/predicate.go @@ -22,7 +22,6 @@ import ( "github.com/google/go-cmp/cmp" cephv1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1" - "github.com/rook/rook/pkg/operator/ceph/controller" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" v1 "k8s.io/api/core/v1" kerrors "k8s.io/apimachinery/pkg/api/errors" @@ -47,7 +46,7 @@ func predicateController(ctx context.Context, c client.Client, opNamespace strin // If a Ceph Cluster is created we want to reconcile the csi driver if cephCluster, ok := e.Object.(*cephv1.CephCluster); ok { // If there are more than one ceph cluster in the same namespace do not reconcile - if controller.DuplicateCephClusters(ctx, c, e.Object, false) { + if opcontroller.DuplicateCephClusters(ctx, c, e.Object, false) { return false } diff --git a/pkg/operator/ceph/file/subvolumegroup/controller.go b/pkg/operator/ceph/file/subvolumegroup/controller.go index 09895b9e6..9c9af8fa6 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller.go +++ b/pkg/operator/ceph/file/subvolumegroup/controller.go @@ -40,7 +40,9 @@ import ( "github.com/rook/rook/pkg/clusterd" "github.com/rook/rook/pkg/operator/ceph/cluster/mon" opcontroller "github.com/rook/rook/pkg/operator/ceph/controller" + "github.com/rook/rook/pkg/operator/ceph/csi" "github.com/rook/rook/pkg/operator/ceph/reporting" + "github.com/rook/rook/pkg/operator/k8sutil" kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -175,9 +177,18 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ logger.Debugf("deleting subvolume group %q", cephFilesystemSubVolumeGroup.Name) err := r.deleteSubVolumeGroup(cephFilesystemSubVolumeGroup) if err != nil { + if strings.Contains(err.Error(), opcontroller.UninitializedCephConfigError) { + logger.Info(opcontroller.OperatorNotInitializedMessage) + return opcontroller.WaitForRequeueIfOperatorNotInitialized, nil + } return reconcile.Result{}, errors.Wrapf(err, "failed to delete ceph ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name) } + err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, nil) + if err != nil { + return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config") + } + // Remove finalizer err = opcontroller.RemoveFinalizer(r.opManagerContext, r.client, cephFilesystemSubVolumeGroup) if err != nil { @@ -218,6 +229,20 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) reconcile(request reconcile.Requ return reconcile.Result{}, errors.Wrapf(err, "failed to create or update ceph filesystem subvolume group %q", cephFilesystemSubVolumeGroup.Name) } + // Update CSI config map + // If the mon endpoints change, the mon health check go routine will take care of updating the + // config map, so no special care is needed in this controller + csiClusterConfigEntry := csi.CsiClusterConfigEntry{ + Monitors: csi.MonEndpoints(r.clusterInfo.Monitors), + CephFS: &csi.CsiCephFSSpec{ + SubvolumeGroup: cephFilesystemSubVolumeGroup.Name, + }, + } + err = csi.SaveClusterConfig(r.context.Clientset, buildClusterID(cephFilesystemSubVolumeGroup), r.clusterInfo, &csiClusterConfigEntry) + if err != nil { + return reconcile.Result{}, errors.Wrap(err, "failed to save cluster config") + } + // Success! Let's update the status r.updateStatus(r.client, request.NamespacedName, cephv1.ConditionReady) @@ -241,11 +266,16 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) createOrUpdateSubVolumeGroup(cep // Delete the ceph filesystem subvolume group func (r *ReconcileCephFilesystemSubVolumeGroup) deleteSubVolumeGroup(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) error { logger.Infof("deleting ceph filesystem subvolume group object %q", cephFilesystemSubVolumeGroup.Name) - if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Name, cephFilesystemSubVolumeGroup.Spec.FilesystemName); err != nil { + if err := cephclient.DeleteCephFSSubVolumeGroup(r.context, r.clusterInfo, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name); err != nil { code, ok := exec.ExitStatus(err) + // If the subvolumegroup does not exit, we should not return an error + if ok && code == int(syscall.ENOENT) { + logger.Debugf("ceph filesystem subvolume group %q do not exist", cephFilesystemSubVolumeGroup.Name) + return nil + } // If the subvolumegroup has subvolumes the command will fail with: // Error ENOTEMPTY: error in rmdir /volumes/csi - if ok && code == int(syscall.ENOTEMPTY) { + if ok && (code == int(syscall.ENOTEMPTY)) { return errors.Wrapf(err, "failed to delete ceph filesystem subvolume group %q, remove the subvolumes first", cephFilesystemSubVolumeGroup.Name) } @@ -272,9 +302,15 @@ func (r *ReconcileCephFilesystemSubVolumeGroup) updateStatus(client client.Clien } cephFilesystemSubVolumeGroup.Status.Phase = status + cephFilesystemSubVolumeGroup.Status.Info = map[string]string{"clusterID": buildClusterID(cephFilesystemSubVolumeGroup)} if err := reporting.UpdateStatus(client, cephFilesystemSubVolumeGroup); err != nil { logger.Errorf("failed to set ceph ceph filesystem subvolume group %q status to %q. %v", name, status, err) return } logger.Debugf("ceph ceph filesystem subvolume group %q status updated to %q", name, status) } + +func buildClusterID(cephFilesystemSubVolumeGroup *cephv1.CephFilesystemSubVolumeGroup) string { + clusterID := fmt.Sprintf("%s-%s-file-%s", cephFilesystemSubVolumeGroup.Namespace, cephFilesystemSubVolumeGroup.Spec.FilesystemName, cephFilesystemSubVolumeGroup.Name) + return k8sutil.Hash(clusterID) +} diff --git a/pkg/operator/ceph/file/subvolumegroup/controller_test.go b/pkg/operator/ceph/file/subvolumegroup/controller_test.go index 6fffbdea5..1d0e5a555 100644 --- a/pkg/operator/ceph/file/subvolumegroup/controller_test.go +++ b/pkg/operator/ceph/file/subvolumegroup/controller_test.go @@ -26,6 +26,7 @@ import ( rookclient "github.com/rook/rook/pkg/client/clientset/versioned/fake" "github.com/rook/rook/pkg/client/clientset/versioned/scheme" "github.com/rook/rook/pkg/clusterd" + "github.com/rook/rook/pkg/operator/ceph/csi" "github.com/rook/rook/pkg/operator/k8sutil" testop "github.com/rook/rook/pkg/operator/test" exectest "github.com/rook/rook/pkg/util/exec/test" @@ -206,15 +207,39 @@ func TestCephClientController(t *testing.T) { client: cl, scheme: s, context: c, - opManagerContext: context.TODO(), + opManagerContext: ctx, } + // Enable CSI + csi.EnableRBD = true + os.Setenv("POD_NAMESPACE", namespace) + // Create CSI config map + ownerRef := &metav1.OwnerReference{} + ownerInfo := k8sutil.NewOwnerInfoWithOwnerRef(ownerRef, "") + err = csi.CreateCsiConfigMap(namespace, c.Clientset, ownerInfo) + assert.NoError(t, err) + res, err := r.Reconcile(ctx, req) assert.NoError(t, err) assert.False(t, res.Requeue) - err = r.client.Get(context.TODO(), req.NamespacedName, cephFilesystemSubVolumeGroup) + err = r.client.Get(ctx, req.NamespacedName, cephFilesystemSubVolumeGroup) assert.NoError(t, err) assert.Equal(t, cephv1.ConditionReady, cephFilesystemSubVolumeGroup.Status.Phase) + assert.NotEmpty(t, cephFilesystemSubVolumeGroup.Status.Info["clusterID"]) + + // test that csi configmap is created + cm, err := c.Clientset.CoreV1().ConfigMaps(namespace).Get(ctx, csi.ConfigName, metav1.GetOptions{}) + assert.NoError(t, err) + assert.NotEmpty(t, cm.Data[csi.ConfigKey]) + assert.Contains(t, cm.Data[csi.ConfigKey], "clusterID") + assert.Contains(t, cm.Data[csi.ConfigKey], "group-a") }) } + +func Test_buildClusterID(t *testing.T) { + longName := "foooooooooooooooooooooooooooooooooooooooooooo" + cephFilesystemSubVolumeGroup := &cephv1.CephFilesystemSubVolumeGroup{ObjectMeta: metav1.ObjectMeta{Namespace: "rook-ceph", Name: longName}, Spec: cephv1.CephFilesystemSubVolumeGroupSpec{FilesystemName: "myfs"}} + clusterID := buildClusterID(cephFilesystemSubVolumeGroup) + assert.Equal(t, "29e92135b7e8c014079b9f9f3566777d", clusterID) +}