Merge pull request #14961 from rook/mergify/bp/release-1.15/pr-14952

mds: Wait for mds standby upgrade for same fs (backport #14952)

pkg/daemon/ceph/client/filesystem.go

@@ -20,6 +20,7 @@ import (
 	ctx "context"
 	"encoding/json"
 	"fmt"
+	"regexp"
 	"strconv"
 	"syscall"
 	"time"
@@ -362,14 +363,14 @@ func deleteFSPool(context *clusterd.Context, clusterInfo *ClusterInfo, poolNames
 }
 
 // WaitForNoStandbys waits for all standbys go away
-func WaitForNoStandbys(context *clusterd.Context, clusterInfo *ClusterInfo, retryInterval, timeout time.Duration) error {
+func WaitForNoStandbys(context *clusterd.Context, clusterInfo *ClusterInfo, fsName string, retryInterval, timeout time.Duration) error {
 	err := wait.PollUntilContextTimeout(clusterInfo.Context, retryInterval, timeout, true, func(ctx ctx.Context) (bool, error) {
 		mdsDump, err := GetMDSDump(context, clusterInfo)
 		if err != nil {
 			logger.Errorf("failed to get fs dump. %v", err)
 			return false, nil
 		}
-		return len(mdsDump.Standbys) == 0, nil
+		return !filesystemHasStandby(mdsDump, fsName), nil
 	})
 
 	if err != nil {
@@ -378,6 +379,21 @@ func WaitForNoStandbys(context *clusterd.Context, clusterInfo *ClusterInfo, retr
 	return nil
 }
 
+func filesystemHasStandby(dump *MDSDump, fsName string) bool {
+	for _, standby := range dump.Standbys {
+		// The mds dump does not explicitly return the name of the filesystem that the
+		// daemon belongs to, so the matching to the filesystem name is based on the mds daemon name
+		// with a regular expression comparison with the expected suffix.
+		// For example, if the filesystem is "myfs", the standby name may be "myfs-a" or "myfs-b".
+		matchString := fmt.Sprintf("^%s-[a-z]{1}$", fsName)
+		matched, _ := regexp.MatchString(matchString, standby.Name)
+		if matched {
+			return true
+		}
+	}
+	return false
+}
+
 func GetMDSDump(context *clusterd.Context, clusterInfo *ClusterInfo) (*MDSDump, error) {
 	args := []string{"fs", "dump"}
 	cmd := NewCephCommand(context, clusterInfo, args)

pkg/daemon/ceph/client/filesystem_test.go

@@ -520,9 +520,33 @@ func TestGetMDSDump(t *testing.T) {
 	assert.Error(t, err)
 }
 
+func TestFSHasStandby(t *testing.T) {
+	// Not found in an empty list
+	fsName := "foo"
+	dump := &MDSDump{}
+	assert.False(t, filesystemHasStandby(dump, fsName))
+
+	// Not found with a non-matching name
+	dump = &MDSDump{Standbys: []MDSStandBy{{Name: "foo-cd"}}}
+	assert.False(t, filesystemHasStandby(dump, fsName))
+
+	// Found in the list with -a suffix
+	dump = &MDSDump{Standbys: []MDSStandBy{{Name: "foo-a"}}}
+	assert.True(t, filesystemHasStandby(dump, fsName))
+
+	// Found in the list with -c suffix
+	dump = &MDSDump{Standbys: []MDSStandBy{{Name: "foo-c"}}}
+	assert.True(t, filesystemHasStandby(dump, fsName))
+
+	// Does not match another filesystem
+	fsName = "bar"
+	assert.False(t, filesystemHasStandby(dump, fsName))
+}
+
 func TestWaitForNoStandbys(t *testing.T) {
 	executor := &exectest.MockExecutor{}
 	context := &clusterd.Context{Executor: executor}
+	fsName := "rook-ceph-filesystem"
 	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
 		logger.Infof("Command: %s %v", command, args)
 		if args[0] == "fs" {
@@ -542,7 +566,7 @@ func TestWaitForNoStandbys(t *testing.T) {
 		return "", errors.Errorf("unexpected ceph command %q", args)
 	}
 
-	err := WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), time.Millisecond, 5*time.Millisecond)
+	err := WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), fsName, time.Millisecond, 5*time.Millisecond)
 	assert.Error(t, err)
 
 	executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) {
@@ -555,7 +579,7 @@ func TestWaitForNoStandbys(t *testing.T) {
 		return "", errors.Errorf("unexpected ceph command %q", args)
 	}
 
-	err = WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), time.Millisecond, 5*time.Millisecond)
+	err = WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), fsName, time.Millisecond, 5*time.Millisecond)
 	assert.Error(t, err)
 
 	firstCall := true
@@ -582,7 +606,7 @@ func TestWaitForNoStandbys(t *testing.T) {
 		}
 		return "", errors.Errorf("unexpected ceph command %q", args)
 	}
-	err = WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), time.Millisecond, 5*time.Millisecond)
+	err = WaitForNoStandbys(context, AdminTestClusterInfo("mycluster"), fsName, time.Millisecond, 5*time.Millisecond)
 	assert.NoError(t, err)
 }
 

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

@@ -290,7 +290,7 @@ func (c *Cluster) upgradeMDS() error {
 		return errors.Wrap(err, "failed to scale down deployments during upgrade")
 	}
 	logger.Debugf("waiting for all standbys gone")
-	if err := cephclient.WaitForNoStandbys(c.context, c.clusterInfo, 3*time.Second, 120*time.Second); err != nil {
+	if err := cephclient.WaitForNoStandbys(c.context, c.clusterInfo, c.fs.Name, 3*time.Second, 120*time.Second); err != nil {
 		return errors.Wrap(err, "failed to wait for stopping all standbys")
 	}