Merge pull request #6524 from LalitMaganti/master

ceph: associate newly created pools to existing filesystem
2020-11-11 16:15:13 -07:00
parent dd560f2305 1d1a9f111a
commit 90ec68d9ff
3 changed files with 128 additions and 19 deletions
@@ -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 {
@@ -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")
@@ -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 {