From 12689bd119f6d308fb4e312307322976233674da Mon Sep 17 00:00:00 2001 From: Travis Nielsen Date: Thu, 14 Oct 2021 19:13:36 -0600 Subject: [PATCH] ceph: enable mon failover for the arbiter in stretch mode Prior to ceph v16.2.7 the failover of the arbiter mon was not supported. Now the new tiebreaker mon can be set during the failover event and provide more dynamic stability to the mon quorum if another node is available in the arbiter zone. Signed-off-by: Travis Nielsen --- pkg/daemon/ceph/client/mon.go | 11 +++ pkg/daemon/ceph/client/mon_test.go | 15 ++++ pkg/operator/ceph/cluster/cluster.go | 3 +- pkg/operator/ceph/cluster/mon/health.go | 74 +++++++++++++------- pkg/operator/ceph/cluster/mon/health_test.go | 35 +++++++++ pkg/operator/ceph/cluster/mon/mon.go | 12 +++- 6 files changed, 120 insertions(+), 30 deletions(-) diff --git a/pkg/daemon/ceph/client/mon.go b/pkg/daemon/ceph/client/mon.go index dec00f07b..08cfc92a8 100644 --- a/pkg/daemon/ceph/client/mon.go +++ b/pkg/daemon/ceph/client/mon.go @@ -152,3 +152,14 @@ func SetMonStretchTiebreaker(context *clusterd.Context, clusterInfo *ClusterInfo logger.Infof("successfully set mon tiebreaker %q in failure domain %q", monName, bucketType) return nil } + +// SetNewTiebreaker sets the new tiebreaker mon in the stretch cluster during a failover +func SetNewTiebreaker(context *clusterd.Context, clusterInfo *ClusterInfo, monName string) error { + logger.Infof("setting new mon tiebreaker %q in arbiter zone", monName) + args := []string{"mon", "set_new_tiebreaker", monName} + if _, err := NewCephCommand(context, clusterInfo, args).Run(); err != nil { + return errors.Wrapf(err, "failed to set new mon tiebreaker %q", monName) + } + logger.Infof("successfully set new mon tiebreaker %q in arbiter zone", monName) + return nil +} diff --git a/pkg/daemon/ceph/client/mon_test.go b/pkg/daemon/ceph/client/mon_test.go index 83ef7fe54..9c3a8c441 100644 --- a/pkg/daemon/ceph/client/mon_test.go +++ b/pkg/daemon/ceph/client/mon_test.go @@ -88,15 +88,22 @@ func TestStretchElectionStrategy(t *testing.T) { func TestStretchClusterMonTiebreaker(t *testing.T) { monName := "a" failureDomain := "rack" + setTiebreaker := false + enabledStretch := false executor := &exectest.MockExecutor{} executor.MockExecuteCommandWithOutput = func(command string, args ...string) (string, error) { logger.Infof("Command: %s %v", command, args) switch { case args[0] == "mon" && args[1] == "enable_stretch_mode": + enabledStretch = true assert.Equal(t, monName, args[2]) assert.Equal(t, defaultStretchCrushRuleName, args[3]) assert.Equal(t, failureDomain, args[4]) return "", nil + case args[0] == "mon" && args[1] == "set_new_tiebreaker": + setTiebreaker = true + assert.Equal(t, monName, args[2]) + return "", nil } return "", errors.Errorf("unexpected ceph command %q", args) } @@ -105,6 +112,14 @@ func TestStretchClusterMonTiebreaker(t *testing.T) { err := SetMonStretchTiebreaker(context, clusterInfo, monName, failureDomain) assert.NoError(t, err) + assert.True(t, enabledStretch) + assert.False(t, setTiebreaker) + enabledStretch = false + + err = SetNewTiebreaker(context, clusterInfo, monName) + assert.NoError(t, err) + assert.True(t, setTiebreaker) + assert.False(t, enabledStretch) } func TestMonDump(t *testing.T) { diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index c14751d3c..c6482b19f 100755 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/pkg/operator/ceph/cluster/cluster.go @@ -137,7 +137,8 @@ func (c *cluster) reconcileCephDaemons(rookImage string, cephVersion cephver.Cep // If a stretch cluster, enable the arbiter after the OSDs are created with the CRUSH map if c.Spec.IsStretchCluster() { - if err := c.mons.ConfigureArbiter(); err != nil { + failingOver := false + if err := c.mons.ConfigureArbiter(failingOver); err != nil { return errors.Wrap(err, "failed to configure stretch arbiter") } } diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index 4cbb3510e..2cf00189c 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/pkg/operator/ceph/cluster/mon/health.go @@ -28,6 +28,7 @@ import ( cephclient "github.com/rook/rook/pkg/daemon/ceph/client" cephutil "github.com/rook/rook/pkg/daemon/ceph/util" "github.com/rook/rook/pkg/operator/ceph/controller" + "github.com/rook/rook/pkg/operator/ceph/version" "github.com/rook/rook/pkg/operator/k8sutil" kerrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -44,6 +45,8 @@ var ( timeZero = time.Duration(0) // Check whether mons are on the same node once per operator restart since it's a rare scheduling condition needToCheckMonsOnSameNode = true + // Version of Ceph where the arbiter failover is supported + arbiterFailoverSupportedCephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 7} ) // HealthChecker aggregates the mon/cluster info needed to check the health of the monitors @@ -322,34 +325,49 @@ func (c *Cluster) failMon(monCount, desiredMonCount int, name string) bool { if err := c.removeMon(name); err != nil { logger.Errorf("failed to remove mon %q. %v", name, err) } - } else { - if c.spec.IsStretchCluster() && name == c.arbiterMon { - // Ceph does not currently support updating the arbiter mon - // or else the mons in the two datacenters will not be aware anymore - // of the arbiter mon. Thus, disabling failover until the arbiter - // mon can be updated in ceph. - logger.Warningf("refusing to failover arbiter mon %q on a stretched cluster", name) - return false - } + return true + } - // prevent any voluntary mon drain while failing over - if err := c.blockMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil { - logger.Errorf("failed to block mon drain. %v", err) - } + if err := c.allowFailover(name); err != nil { + logger.Warningf("aborting mon %q failover. %v", name, err) + return false + } - // bring up a new mon to replace the unhealthy mon - if err := c.failoverMon(name); err != nil { - logger.Errorf("failed to failover mon %q. %v", name, err) - } + // prevent any voluntary mon drain while failing over + if err := c.blockMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil { + logger.Errorf("failed to block mon drain. %v", err) + } - // allow any voluntary mon drain after failover - if err := c.allowMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil { - logger.Errorf("failed to allow mon drain. %v", err) - } + // bring up a new mon to replace the unhealthy mon + if err := c.failoverMon(name); err != nil { + logger.Errorf("failed to failover mon %q. %v", name, err) + } + + // allow any voluntary mon drain after failover + if err := c.allowMonDrain(types.NamespacedName{Name: monPDBName, Namespace: c.Namespace}); err != nil { + logger.Errorf("failed to allow mon drain. %v", err) } return true } +func (c *Cluster) allowFailover(name string) error { + if !c.spec.IsStretchCluster() { + // always failover if not a stretch cluster + return nil + } + if name != c.arbiterMon { + // failover if it's a non-arbiter + return nil + } + if c.ClusterInfo.CephVersion.IsAtLeast(arbiterFailoverSupportedCephVersion) { + // failover the arbiter if at least v16.2.7 + return nil + } + + // Ceph does not support updating the arbiter mon in older versions + return errors.Errorf("refusing to failover arbiter mon %q on a stretched cluster until upgrading to ceph version %s", name, arbiterFailoverSupportedCephVersion.String()) +} + func (c *Cluster) removeOrphanMonResources() { if c.spec.Mon.VolumeClaimTemplate == nil { logger.Debug("skipping check for orphaned mon pvcs since using the host path") @@ -435,6 +453,9 @@ func (c *Cluster) failoverMon(name string) error { // remove the failed mon from a local list of the existing mons for finding a stretch zone existingMons := c.clusterInfoToMonConfig(name) + // Cache the name of the current arbiter in case it is updated during the failover + // This allows a simple check for updating the arbiter later in this method + currentArbiter := c.arbiterMon zone, err := c.findAvailableZoneIfStretched(existingMons) if err != nil { return errors.Wrap(err, "failed to find available stretch zone") @@ -473,12 +494,11 @@ func (c *Cluster) failoverMon(name string) error { } // Assign to a zone if a stretch cluster - if c.spec.IsStretchCluster() { - if name == c.arbiterMon { - // Update the arbiter mon for the stretch cluster if it changed - if err := c.ConfigureArbiter(); err != nil { - return errors.Wrap(err, "failed to configure stretch arbiter") - } + if c.spec.IsStretchCluster() && name == currentArbiter { + // Update the arbiter mon for the stretch cluster if it changed + failingOver := true + if err := c.ConfigureArbiter(failingOver); err != nil { + return errors.Wrap(err, "failed to configure stretch arbiter") } } diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index bbec6a522..f01d06c1b 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/pkg/operator/ceph/cluster/mon/health_test.go @@ -31,6 +31,7 @@ import ( cephclient "github.com/rook/rook/pkg/daemon/ceph/client" clienttest "github.com/rook/rook/pkg/daemon/ceph/client/test" "github.com/rook/rook/pkg/operator/ceph/config" + "github.com/rook/rook/pkg/operator/ceph/version" testopk8s "github.com/rook/rook/pkg/operator/k8sutil/test" "github.com/rook/rook/pkg/operator/test" exectest "github.com/rook/rook/pkg/util/exec/test" @@ -153,6 +154,40 @@ func TestCheckHealth(t *testing.T) { } } +func TestSkipMonFailover(t *testing.T) { + c := New(&clusterd.Context{}, "ns", cephv1.ClusterSpec{}, nil, nil) + c.ClusterInfo = clienttest.CreateTestClusterInfo(1) + monName := "arb" + + t.Run("don't skip failover for non-stretch", func(t *testing.T) { + assert.NoError(t, c.allowFailover(monName)) + }) + + t.Run("don't skip failover for non-arbiter", func(t *testing.T) { + c.spec.Mon.Count = 5 + c.spec.Mon.StretchCluster = &cephv1.StretchClusterSpec{ + Zones: []cephv1.StretchClusterZoneSpec{ + {Name: "a"}, + {Name: "b"}, + {Name: "c", Arbiter: true}, + }, + } + + assert.NoError(t, c.allowFailover(monName)) + }) + + t.Run("skip failover for arbiter if an older version of ceph", func(t *testing.T) { + c.arbiterMon = monName + c.ClusterInfo.CephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 6} + assert.Error(t, c.allowFailover(monName)) + }) + + t.Run("don't skip failover for arbiter if a newer version of ceph", func(t *testing.T) { + c.ClusterInfo.CephVersion = version.CephVersion{Major: 16, Minor: 2, Extra: 7} + assert.NoError(t, c.allowFailover(monName)) + }) +} + func TestEvictMonOnSameNode(t *testing.T) { ctx := context.TODO() clientset := test.New(t, 1) diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index d987ef7fb..a5e793404 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/pkg/operator/ceph/cluster/mon/mon.go @@ -339,11 +339,20 @@ func (c *Cluster) isArbiterZone(zone string) bool { return c.getArbiterZone() == zone } -func (c *Cluster) ConfigureArbiter() error { +func (c *Cluster) ConfigureArbiter(failingOver bool) error { if c.arbiterMon == "" { return errors.New("arbiter not specified for the stretch cluster") } + failureDomain := c.stretchFailureDomainName() + if failingOver { + // Set the new mon tiebreaker + if err := cephclient.SetNewTiebreaker(c.context, c.ClusterInfo, c.arbiterMon); err != nil { + return errors.Wrap(err, "failed to set new mon tiebreaker") + } + return nil + } + monDump, err := cephclient.GetMonDump(c.context, c.ClusterInfo) if err != nil { logger.Warningf("attempting to enable arbiter after failed to detect if already enabled. %v", err) @@ -355,7 +364,6 @@ func (c *Cluster) ConfigureArbiter() error { // Wait for the CRUSH map to have at least two zones // The timeout is relatively short since the operator will requeue the reconcile // and try again at a higher level if not yet found - failureDomain := c.stretchFailureDomainName() logger.Infof("enabling stretch mode... waiting for two failure domains of type %q to be found in the CRUSH map after OSD initialization", failureDomain) pollInterval := 5 * time.Second totalWaitTime := 2 * time.Minute