From 1d1a9f111ae375090c22d640d78164ee008ff847 Mon Sep 17 00:00:00 2001 From: Lalit Maganti Date: Sun, 1 Nov 2020 00:07:18 +0000 Subject: [PATCH] ceph: associate newly created pools to existing filesystem When new pools are added to an already existing filesystem, we would create them but not actually associate them to the filesystem. Ensure that we do this. Also, while we're here, we refactor the CreateFilesystem test: before, we were not properly testing creating the filesystem from scratch because we would return valid JSON for *every* ceph command. Change this by making the first run return an error so that we get some better coverage. After this, we also add a check for the correctness of adding new pools to the CreateFilesystem unittest. Fixes #5876 Signed-off-by: Lalit Maganti --- pkg/daemon/ceph/client/filesystem.go | 16 +++- pkg/operator/ceph/file/filesystem.go | 40 ++++++---- pkg/operator/ceph/file/filesystem_test.go | 91 ++++++++++++++++++++++- 3 files changed, 128 insertions(+), 19 deletions(-) diff --git a/pkg/daemon/ceph/client/filesystem.go b/pkg/daemon/ceph/client/filesystem.go index 6bfe375c6..e77487bc8 100644 --- a/pkg/daemon/ceph/client/filesystem.go +++ b/pkg/daemon/ceph/client/filesystem.go @@ -154,17 +154,25 @@ func CreateFilesystem(context *clusterd.Context, clusterInfo *ClusterInfo, name, // add each additional pool for i := 1; i < len(dataPools); i++ { - poolName := dataPools[i] - args = []string{"fs", "add_data_pool", name, poolName} - _, err = NewCephCommand(context, clusterInfo, args).Run() + err = AddDataPoolToFilesystem(context, clusterInfo, name, dataPools[i]) if err != nil { - logger.Errorf("failed to add pool %q to file system %q. %v", poolName, name, err) + logger.Errorf("%v", err) } } return nil } +// AddDataPoolToFilesystem associates the provided data pool with the filesystem. +func AddDataPoolToFilesystem(context *clusterd.Context, clusterInfo *ClusterInfo, name, poolName string) error { + args := []string{"fs", "add_data_pool", name, poolName} + _, err := NewCephCommand(context, clusterInfo, args).Run() + if err != nil { + return errors.Wrapf(err, "failed to add pool %q to file system %q. (%v)", poolName, name, err) + } + return nil +} + // IsMultiFSEnabled returns true if ROOK_ALLOW_MULTIPLE_FILESYSTEMS is set to "true", allowing // Rook to create multiple Ceph filesystems. False if Rook is not allowed to do so. func IsMultiFSEnabled() bool { diff --git a/pkg/operator/ceph/file/filesystem.go b/pkg/operator/ceph/file/filesystem.go index 6db562663..c9a4dd546 100644 --- a/pkg/operator/ceph/file/filesystem.go +++ b/pkg/operator/ceph/file/filesystem.go @@ -192,26 +192,38 @@ func SetPoolSize(f *Filesystem, context *clusterd.Context, clusterInfo *client.C return nil } +// updateFilesystem ensures that a filesystem which already exists matches the provided spec. +func (f *Filesystem) updateFilesystem(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error { + // Even if the fs already exists, the num active mdses may have changed + if err := client.SetNumMDSRanks(context, clusterInfo, f.Name, spec.MetadataServer.ActiveCount); err != nil { + logger.Errorf( + fmt.Sprintf("failed to set num mds ranks (max_mds) to %d for filesystem %s, still continuing. ", spec.MetadataServer.ActiveCount, f.Name) + + "this error is not critical, but mdses may not be as failure tolerant as desired. " + + fmt.Sprintf("USER should verify that the number of active mdses is %d with 'ceph fs get %s'", spec.MetadataServer.ActiveCount, f.Name) + + fmt.Sprintf(". %v", err), + ) + } + + if err := SetPoolSize(f, context, clusterInfo, clusterSpec, spec); err != nil { + return errors.Wrap(err, "failed to set pools size") + } + + dataPoolNames := generateDataPoolNames(f, spec) + for i := range spec.DataPools { + if err := client.AddDataPoolToFilesystem(context, clusterInfo, f.Name, dataPoolNames[i]); err != nil { + return err + } + } + return nil +} + // doFilesystemCreate starts the Ceph file daemons and creates the filesystem in Ceph. func (f *Filesystem) doFilesystemCreate(context *clusterd.Context, clusterInfo *client.ClusterInfo, clusterSpec *cephv1.ClusterSpec, spec cephv1.FilesystemSpec) error { _, err := client.GetFilesystem(context, clusterInfo, f.Name) if err == nil { logger.Infof("filesystem %s already exists", f.Name) - // Even if the fs already exists, the num active mdses may have changed - - if err := client.SetNumMDSRanks(context, clusterInfo, f.Name, spec.MetadataServer.ActiveCount); err != nil { - logger.Errorf( - fmt.Sprintf("failed to set num mds ranks (max_mds) to %d for filesystem %s, still continuing. ", spec.MetadataServer.ActiveCount, f.Name) + - "this error is not critical, but mdses may not be as failure tolerant as desired. " + - fmt.Sprintf("USER should verify that the number of active mdses is %d with 'ceph fs get %s'", spec.MetadataServer.ActiveCount, f.Name) + - fmt.Sprintf(". %v", err), - ) - } - if err := SetPoolSize(f, context, clusterInfo, clusterSpec, spec); err != nil { - return errors.Wrap(err, "failed to set pools size") - } - return nil + return f.updateFilesystem(context, clusterInfo, clusterSpec, spec) } if len(spec.DataPools) == 0 { return errors.New("at least one data pool must be specified") diff --git a/pkg/operator/ceph/file/filesystem_test.go b/pkg/operator/ceph/file/filesystem_test.go index 2fd8c2c41..e64710472 100644 --- a/pkg/operator/ceph/file/filesystem_test.go +++ b/pkg/operator/ceph/file/filesystem_test.go @@ -21,6 +21,7 @@ import ( "io/ioutil" "os" "path" + "reflect" "strings" "testing" @@ -76,12 +77,57 @@ func TestCreateFilesystem(t *testing.T) { mds.UpdateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub() configDir, _ := ioutil.TempDir("", "") + // Output to check multiple filesystem creation fses := `[{"name":"myfs","metadata_pool":"myfs-metadata","metadata_pool_id":1,"data_pool_ids":[2],"data_pools":["myfs-data0"]}]` + createdFsResponse := `{"fs_name": "myfs", "metadata_pool": 2, "data_pools":[3]}` + + isBasePoolOperation := func(command string, args []string) bool { + if reflect.DeepEqual(args[0:7], []string{"osd", "pool", "create", "myfs-metadata", "0", "replicated", "myfs-metadata"}) { + return true + } else if reflect.DeepEqual(args[0:5], []string{"osd", "crush", "rule", "create-replicated", "myfs-metadata"}) { + return true + } else if reflect.DeepEqual(args[0:6], []string{"osd", "pool", "set", "myfs-metadata", "size", "1"}) { + return true + } else if reflect.DeepEqual(args[0:7], []string{"osd", "pool", "create", "myfs-data0", "0", "replicated", "myfs-data0"}) { + return true + } else if reflect.DeepEqual(args[0:5], []string{"osd", "crush", "rule", "create-replicated", "myfs-data0"}) { + return true + } else if reflect.DeepEqual(args[0:6], []string{"osd", "pool", "set", "myfs-data0", "size", "1"}) { + return true + } else if reflect.DeepEqual(args[0:4], []string{"fs", "add_data_pool", "myfs", "myfs-data0"}) { + return true + } + return false + } + + firstGet := true executor := &exectest.MockExecutor{ MockExecuteCommandWithOutputFile: func(command string, outFileArg string, args ...string) (string, error) { - return "{\"key\":\"mysecurekey\"}", nil + if contains(args, "fs") && contains(args, "get") { + if firstGet { + firstGet = false + return "", errors.New("fs doesn't exist") + } + return createdFsResponse, nil + } else if contains(args, "fs") && contains(args, "ls") { + return "[]", nil + } else if contains(args, "osd") && contains(args, "lspools") { + return "[]", nil + } else if isBasePoolOperation(command, args) { + return "", nil + } else if reflect.DeepEqual(args[0:5], []string{"fs", "new", "myfs", "myfs-metadata", "myfs-data0"}) { + return "", nil + } else if contains(args, "auth") && contains(args, "get-or-create-key") { + return "{\"key\":\"mysecurekey\"}", nil + } else if contains(args, "config") && contains(args, "mds_cache_memory_limit") { + return "", nil + } else if contains(args, "set") && contains(args, "max_mds") { + return "", nil + } + assert.Fail(t, "Unexpected command") + return "", nil }, MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) { if strings.Contains(command, "ceph-authtool") { @@ -132,6 +178,48 @@ func TestCreateFilesystem(t *testing.T) { assert.ElementsMatch(t, []string{"rook-ceph-mds-myfs-a", "rook-ceph-mds-myfs-b"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated)) testopk8s.ClearDeploymentsUpdated(deploymentsUpdated) + // Increasing the number of data pools should be successful. + createDataOnePoolCount := 0 + addDataOnePoolCount := 0 + executor = &exectest.MockExecutor{ + MockExecuteCommandWithOutputFile: func(command string, outFileArg string, args ...string) (string, error) { + if contains(args, "fs") && contains(args, "get") { + return createdFsResponse, nil + } else if isBasePoolOperation(command, args) { + return "", nil + } else if reflect.DeepEqual(args[0:4], []string{"osd", "pool", "create", "myfs-data1"}) { + createDataOnePoolCount++ + return "", nil + } else if reflect.DeepEqual(args[0:4], []string{"fs", "add_data_pool", "myfs", "myfs-data1"}) { + addDataOnePoolCount++ + return "", nil + } else if contains(args, "set") && contains(args, "max_mds") { + return "", nil + } else if contains(args, "auth") && contains(args, "get-or-create-key") { + return "{\"key\":\"mysecurekey\"}", nil + } else if reflect.DeepEqual(args[0:5], []string{"osd", "crush", "rule", "create-replicated", "myfs-data1"}) { + return "", nil + } else if reflect.DeepEqual(args[0:6], []string{"osd", "pool", "set", "myfs-data1", "size", "1"}) { + return "", nil + } + assert.Fail(t, "Unexpected command") + return "", nil + }, + } + context = &clusterd.Context{ + Executor: executor, + ConfigDir: configDir, + Clientset: clientset} + fs.Spec.DataPools = append(fs.Spec.DataPools, cephv1.PoolSpec{Replicated: cephv1.ReplicatedSpec{Size: 1, RequireSafeReplicaSize: false}}) + + err = createFilesystem(context, clusterInfo, fs, &cephv1.ClusterSpec{}, metav1.OwnerReference{}, "/var/lib/rook/", scheme.Scheme) + assert.Nil(t, err) + validateStart(t, context, fs) + assert.ElementsMatch(t, []string{"rook-ceph-mds-myfs-a", "rook-ceph-mds-myfs-b"}, testopk8s.DeploymentNamesUpdated(deploymentsUpdated)) + assert.Equal(t, 1, createDataOnePoolCount) + assert.Equal(t, 1, addDataOnePoolCount) + testopk8s.ClearDeploymentsUpdated(deploymentsUpdated) + // Test multiple filesystem creation executor = &exectest.MockExecutor{ MockExecuteCommandWithOutputFile: func(command string, outFileArg string, args ...string) (string, error) { @@ -192,6 +280,7 @@ func TestCreateNopoolFilesystem(t *testing.T) { assert.Nil(t, err) validateStart(t, context, fs) } + func contains(arr []string, str string) bool { for _, a := range arr { if a == str {