ceph mds: allow filesystem specifications that have no pools

design/external-management.md says that we should allow specifying
Filesystems that have no pools. When this is done, the CRD should
interepret that as a request to just manage the MDS daemons without
creating a new filesystem.

Don't fail the verification if there are no data pools in the spec,
and just skip creating a new filesystem if that's the case.

Also, add a testcase for this usage.

Signed-off-by: Jeff Layton <jlayton@redhat.com>

pkg/apis/ceph.rook.io/v1beta1/types.go

@@ -191,10 +191,10 @@ type FilesystemList struct {
 // FilesystemSpec represents the spec of a file system
 type FilesystemSpec struct {
 	// The metadata pool settings
-	MetadataPool PoolSpec `json:"metadataPool"`
+	MetadataPool PoolSpec `json:"metadataPool,omitempty"`
 
 	// The data pool settings
-	DataPools []PoolSpec `json:"dataPools"`
+	DataPools []PoolSpec `json:"dataPools,omitempty"`
 
 	// The mds pod info
 	MetadataServer MetadataServerSpec `json:"metadataServer"`

pkg/operator/ceph/file/filesystem.go

@@ -41,13 +41,15 @@ func createFilesystem(
 		return err
 	}
 
-	var dataPools []*model.Pool
-	for _, p := range fs.Spec.DataPools {
-		dataPools = append(dataPools, p.ToModel(""))
-	}
-	f := mdsdaemon.NewFS(fs.Name, fs.Spec.MetadataPool.ToModel(""), dataPools, fs.Spec.MetadataServer.ActiveCount)
-	if err := f.CreateFilesystem(context, fs.Namespace); err != nil {
-		return fmt.Errorf("failed to create file system %s: %+v", fs.Name, err)
+	if len(fs.Spec.DataPools) != 0 {
+		var dataPools []*model.Pool
+		for _, p := range fs.Spec.DataPools {
+			dataPools = append(dataPools, p.ToModel(""))
+		}
+		f := mdsdaemon.NewFS(fs.Name, fs.Spec.MetadataPool.ToModel(""), dataPools, fs.Spec.MetadataServer.ActiveCount)
+		if err := f.CreateFilesystem(context, fs.Namespace); err != nil {
+			return fmt.Errorf("failed to create file system %s: %+v", fs.Name, err)
+		}
 	}
 
 	filesystem, err := client.GetFilesystem(context, fs.Namespace, fs.Name)
@@ -89,8 +91,12 @@ func validateFilesystem(context *clusterd.Context, f cephv1beta1.Filesystem) err
 	if f.Namespace == "" {
 		return fmt.Errorf("missing namespace")
 	}
+	if f.Spec.MetadataServer.ActiveCount < 1 {
+		return fmt.Errorf("MetadataServer.ActiveCount must be at least 1")
+	}
+	// No data pool means that we expect the fs to exist already
 	if len(f.Spec.DataPools) == 0 {
-		return fmt.Errorf("at least one data pool required")
+		return nil
 	}
 	if err := pool.ValidatePoolSpec(context, f.Namespace, &f.Spec.MetadataPool); err != nil {
 		return fmt.Errorf("invalid metadata pool: %+v", err)
@@ -100,9 +106,6 @@ func validateFilesystem(context *clusterd.Context, f cephv1beta1.Filesystem) err
 			return fmt.Errorf("Invalid data pool: %+v", err)
 		}
 	}
-	if f.Spec.MetadataServer.ActiveCount < 1 {
-		return fmt.Errorf("MetadataServer.ActiveCount must be at least 1")
-	}
 
 	return nil
 }

pkg/operator/ceph/file/filesystem_test.go

@@ -134,6 +134,57 @@ func TestCreateFilesystem(t *testing.T) {
 	assert.Equal(t, "failed to create file system myfs: Cannot create multiple filesystems. Enable ROOK_ALLOW_MULTIPLE_FILESYSTEMS env variable to create more than one", err.Error())
 }
 
+func TestCreateNopoolFilesystem(t *testing.T) {
+	configDir, _ := ioutil.TempDir("", "")
+	// Output to check multiple file system creation
+	fses := `[{"name":"myfs"}]`
+
+	executor := &exectest.MockExecutor{
+		MockExecuteCommandWithOutputFile: func(debug bool, actionName string, command string, outFileArg string, args ...string) (string, error) {
+			return "{\"key\":\"mysecurekey\"}", nil
+		},
+		MockExecuteCommandWithOutput: func(debug bool, actionName string, command string, args ...string) (string, error) {
+			if strings.Contains(command, "ceph-authtool") {
+				cephtest.CreateConfigDir(path.Join(configDir, "ns"))
+			}
+
+			return "", nil
+		},
+	}
+	defer os.RemoveAll(configDir)
+	context := &clusterd.Context{
+		Executor:  executor,
+		ConfigDir: configDir,
+		Clientset: testop.New(3)}
+	fs := cephv1beta1.Filesystem{
+		ObjectMeta: metav1.ObjectMeta{Name: "myfs", Namespace: "ns"},
+		Spec: cephv1beta1.FilesystemSpec{
+			MetadataServer: cephv1beta1.MetadataServerSpec{
+				ActiveCount: 1,
+			},
+		},
+	}
+
+	// start a basic cluster
+	err := createFilesystem(context, fs, "v0.1", cephv1beta1.CephVersionSpec{}, false, []metav1.OwnerReference{})
+	assert.Nil(t, err)
+	validateStart(t, context, fs)
+
+	// starting again should be a no-op
+	err = createFilesystem(context, fs, "v0.1", cephv1beta1.CephVersionSpec{}, false, []metav1.OwnerReference{})
+	assert.Nil(t, err)
+	validateStart(t, context, fs)
+
+	// Test multiple filesystem creation
+	executor = &exectest.MockExecutor{
+		MockExecuteCommandWithOutputFile: func(debug bool, actionName string, command string, outFileArg string, args ...string) (string, error) {
+			if contains(args, "ls") {
+				return fses, nil
+			}
+			return "{\"key\":\"mysecurekey\"}", errors.New("multiple fs")
+		},
+	}
+}
 func contains(arr []string, str string) bool {
 	for _, a := range arr {
 		if a == str {