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 {