diff --git a/pkg/daemon/ceph/client/filesystem.go b/pkg/daemon/ceph/client/filesystem.go index 048f7ce1b..1798a9411 100644 --- a/pkg/daemon/ceph/client/filesystem.go +++ b/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) diff --git a/pkg/daemon/ceph/client/filesystem_test.go b/pkg/daemon/ceph/client/filesystem_test.go index e23a84128..2fcb89992 100644 --- a/pkg/daemon/ceph/client/filesystem_test.go +++ b/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) } diff --git a/pkg/operator/ceph/file/mds/mds.go b/pkg/operator/ceph/file/mds/mds.go index a31510177..de2656f34 100644 --- a/pkg/operator/ceph/file/mds/mds.go +++ b/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") }