Merge pull request #17392 from rook/mergify/bp/release-1.19/pr-17373

mds: fix incorrect behaviour for CephFS when no active standby (backport #17373)

pkg/daemon/ceph/client/filesystem.go

@@ -119,13 +119,23 @@ func getFilesystem(context *clusterd.Context, clusterInfo *ClusterInfo, fsName s
 }
 
 // AllowStandbyReplay gets detailed status information about a Ceph filesystem.
-func AllowStandbyReplay(context *clusterd.Context, clusterInfo *ClusterInfo, fsName string, allowStandbyReplay bool) error {
+func AllowStandbyReplay(context *clusterd.Context, clusterInfo *ClusterInfo, fsName string, allowStandbyReplay bool, wantedStandbyDaemons int32) error {
 	logger.Infof("setting allow_standby_replay to %t for filesystem %q", allowStandbyReplay, fsName)
 	args := []string{"fs", "set", fsName, "allow_standby_replay", strconv.FormatBool(allowStandbyReplay)}
 	_, err := NewCephCommand(context, clusterInfo, args).Run()
 	if err != nil {
 		return errors.Wrapf(err, "failed to set allow_standby_replay to filesystem %s", fsName)
 	}
+	// if we don't need standby, set standby_count_wanted=0 to avoid
+	// getting 'insufficient standby MDS daemons' health issue
+	if !allowStandbyReplay {
+		wantedStandbyDaemons = 0
+	}
+	args = []string{"fs", "set", fsName, "standby_count_wanted", fmt.Sprint(wantedStandbyDaemons)}
+	_, err = NewCephCommand(context, clusterInfo, args).Run()
+	if err != nil {
+		return errors.Wrapf(err, "failed to set standby_count_wanted to filesystem %s", fsName)
+	}
 
 	return nil
 }

pkg/daemon/ceph/client/filesystem_test.go

@@ -757,3 +757,49 @@ func TestListSubvolumesInGroup(t *testing.T) {
 		assert.Empty(t, ret)
 	})
 }
+
+func TestAllowStandbyReplay(t *testing.T) {
+	cephFSName := "myfs"
+	newContext := func(standbyWanted bool, standbyCount int32) *clusterd.Context {
+		t.Helper()
+
+		executor := &exectest.MockExecutor{
+			MockExecuteCommandWithOutput: func(command string, args ...string) (string, error) {
+				t.Logf("Command: %s %v", command, args)
+				if args[0] == "fs" && args[1] == "set" && args[2] == cephFSName {
+					switch args[3] {
+					case "allow_standby_replay":
+						if args[4] == fmt.Sprint(standbyWanted) {
+							return "", nil
+						}
+					case "standby_count_wanted":
+						if args[4] == fmt.Sprint(standbyCount) {
+							return "", nil
+						}
+					}
+				}
+				panic(fmt.Sprintf("unhandled command %q %v", command, args))
+			},
+		}
+
+		return &clusterd.Context{Executor: executor}
+	}
+
+	t.Run("no standby mds", func(t *testing.T) {
+		ctx := newContext(false, 0)
+		err := AllowStandbyReplay(ctx, AdminTestClusterInfo("mycluster"), cephFSName, false, int32(1))
+		assert.NoError(t, err)
+	})
+
+	t.Run("standby mds", func(t *testing.T) {
+		ctx := newContext(true, 1)
+		err := AllowStandbyReplay(ctx, AdminTestClusterInfo("mycluster"), cephFSName, true, int32(1))
+		assert.NoError(t, err)
+	})
+
+	t.Run("standby mds multiple", func(t *testing.T) {
+		ctx := newContext(true, 3)
+		err := AllowStandbyReplay(ctx, AdminTestClusterInfo("mycluster"), cephFSName, true, int32(3))
+		assert.NoError(t, err)
+	})
+}

pkg/operator/ceph/file/filesystem.go

@@ -66,7 +66,7 @@ func createFilesystem(
 			return errors.Wrapf(err, "failed to create filesystem %q", fs.Name)
 		}
 	}
-	if err := cephclient.AllowStandbyReplay(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveStandby); err != nil {
+	if err := cephclient.AllowStandbyReplay(context, clusterInfo, fs.Name, fs.Spec.MetadataServer.ActiveStandby, fs.Spec.MetadataServer.ActiveCount); err != nil {
 		return errors.Wrapf(err, "failed to set allow_standby_replay to filesystem %q", fs.Name)
 	}
 

pkg/operator/ceph/file/filesystem_test.go

@@ -215,6 +215,8 @@ func fsExecutor(t *testing.T, fsName, configDir string, multiFS bool, createData
 					return "", nil
 				} else if slices.Contains(args, "set") && slices.Contains(args, "allow_standby_replay") {
 					return "", nil
+				} else if slices.Contains(args, "set") && slices.Contains(args, "standby_count_wanted") {
+					return "", nil
 				} else if slices.Contains(args, "config") && slices.Contains(args, "mds_join_fs") {
 					return "", nil
 				} else if slices.Contains(args, "flag") && slices.Contains(args, "enable_multiple") {
@@ -296,6 +298,8 @@ func fsExecutor(t *testing.T, fsName, configDir string, multiFS bool, createData
 				return "", nil
 			} else if slices.Contains(args, "set") && slices.Contains(args, "allow_standby_replay") {
 				return "", nil
+			} else if slices.Contains(args, "set") && slices.Contains(args, "standby_count_wanted") {
+				return "", nil
 			} else if slices.Contains(args, "config") && slices.Contains(args, "mds_join_fs") {
 				return "", nil
 			} else if slices.Contains(args, "flag") && slices.Contains(args, "enable_multiple") {
@@ -568,6 +572,8 @@ func TestUpgradeFilesystem(t *testing.T) {
 				return "", nil
 			} else if slices.Contains(args, "allow_standby_replay") {
 				return "", nil
+			} else if slices.Contains(args, "standby_count_wanted") {
+				return "", nil
 			}
 		}
 		if slices.Contains(args, "versions") {

pkg/operator/ceph/file/mds/mds.go

@@ -268,7 +268,7 @@ func (c *Cluster) upgradeMDS() error {
 	log.NamedInfo(nsName, logger, "upgrading MDS cluster for filesystem %q", c.fs.Name)
 
 	// 1. set allow_standby_replay to false
-	if err := cephclient.AllowStandbyReplay(c.context, c.clusterInfo, c.fs.Name, false); err != nil {
+	if err := cephclient.AllowStandbyReplay(c.context, c.clusterInfo, c.fs.Name, false, 0); err != nil {
 		return errors.Wrap(err, "failed to setting allow_standby_replay to false")
 	}
 
@@ -422,7 +422,7 @@ func finishedWithDaemonUpgrade(context *clusterd.Context, clusterInfo *cephclien
 	}
 
 	// set allow_standby_replay back
-	if err := cephclient.AllowStandbyReplay(context, clusterInfo, fsName, fs.Spec.MetadataServer.ActiveStandby); err != nil {
+	if err := cephclient.AllowStandbyReplay(context, clusterInfo, fsName, fs.Spec.MetadataServer.ActiveStandby, fs.Spec.MetadataServer.ActiveCount); err != nil {
 		return errors.Wrap(err, "failed to set allow_standby_replay to true")
 	}