ceph: mon failover should scale down failed mon first

When a mon is failed over, the new mon is started before the old mon
is removed from quorum and purged from the cluster. If the mon has
anti-affinity, in small clusters or specific toplogies where sufficient
nodes are not available, the new mon may be prevented from starting simply
because the dead mon pod still exists. To ensure a clean failover, the
dead mon pod must first be stopped before the new mon is started.

Signed-off-by: Travis Nielsen <tnielsen@redhat.com>

pkg/operator/ceph/cluster/mon/health.go

@@ -247,9 +247,50 @@ func (c *Cluster) failMon(monCount, desiredMonCount int, name string) {
 	}
 }
 
+func (c *Cluster) updateMonDeploymentReplica(name string, enabled bool) error {
+	// get the existing deployment
+	d, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(resourceName(name), metav1.GetOptions{})
+	if err != nil {
+		return errors.Wrapf(err, "failed to get mon %q", name)
+	}
+
+	// set the desired number of replicas
+	var desiredReplicas int32
+	if enabled {
+		desiredReplicas = 1
+	}
+	originalReplicas := *d.Spec.Replicas
+	d.Spec.Replicas = &desiredReplicas
+
+	// update the deployment
+	logger.Infof("scaling the mon %q deployment to replica %d", name, desiredReplicas)
+	_, err = c.context.Clientset.AppsV1().Deployments(c.Namespace).Update(d)
+	if err != nil {
+		return errors.Wrapf(err, "failed to update mon %q replicas from %d to %d", name, originalReplicas, desiredReplicas)
+	}
+	return nil
+}
+
 func (c *Cluster) failoverMon(name string) error {
 	logger.Infof("Failing over monitor %q", name)
 
+	// Scale down the failed mon to allow a new one to start
+	if err := c.updateMonDeploymentReplica(name, false); err != nil {
+		// attempt to continue with the failover even if the bad mon could not be stopped
+		logger.Warningf("failed to stop mon %q for failover. %v", name, err)
+	}
+	newMonSucceeded := false
+	defer func() {
+		if newMonSucceeded {
+			// do nothing if the new mon was started successfully, the deployment will anyway be deleted
+			return
+		}
+		if err := c.updateMonDeploymentReplica(name, true); err != nil {
+			// attempt to continue even if the bad mon could not be restarted
+			logger.Warningf("failed to restart failed mon %q after new mon wouldn't start. %v", name, err)
+		}
+	}()
+
 	// Start a new monitor
 	m := c.newMonConfig(c.maxMonID + 1)
 	logger.Infof("starting new mon: %+v", m)
@@ -284,6 +325,7 @@ func (c *Cluster) failoverMon(name string) error {
 
 	// Only increment the max mon id if the new pod started successfully
 	c.maxMonID++
+	newMonSucceeded = true
 
 	return c.removeMon(name)
 }

pkg/operator/ceph/cluster/mon/health_test.go

@@ -29,11 +29,13 @@ import (
 	"github.com/rook/rook/pkg/clusterd"
 	"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/k8sutil"
 	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"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	apps "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	kerrors "k8s.io/apimachinery/pkg/api/errors"
@@ -106,6 +108,37 @@ func TestCheckHealth(t *testing.T) {
 	}
 }
 
+func TestScaleMonDeployment(t *testing.T) {
+	clientset := test.New(t, 1)
+	context := &clusterd.Context{Clientset: clientset}
+	c := New(context, "ns", cephv1.ClusterSpec{}, metav1.OwnerReference{}, &sync.Mutex{})
+	setCommonMonProperties(c, 1, cephv1.MonSpec{Count: 0, AllowMultiplePerNode: true}, "myversion")
+
+	name := "a"
+	c.spec.Mon.Count = 3
+	logger.Infof("initial mons: %v", c.ClusterInfo.Monitors[name])
+	monConfig := &monConfig{ResourceName: resourceName(name), DaemonName: name, DataPathMap: &config.DataPathMap{}}
+	d, err := c.makeDeployment(monConfig, false)
+	require.NoError(t, err)
+	_, err = clientset.AppsV1().Deployments(c.Namespace).Create(d)
+	require.NoError(t, err)
+
+	verifyMonReplicas(t, c, name, 1)
+	err = c.updateMonDeploymentReplica(name, false)
+	assert.NoError(t, err)
+	verifyMonReplicas(t, c, name, 0)
+
+	err = c.updateMonDeploymentReplica(name, true)
+	assert.NoError(t, err)
+	verifyMonReplicas(t, c, name, 1)
+}
+
+func verifyMonReplicas(t *testing.T, c *Cluster, name string, expected int32) {
+	d, err := c.context.Clientset.AppsV1().Deployments(c.Namespace).Get(resourceName("a"), metav1.GetOptions{})
+	require.NoError(t, err)
+	assert.Equal(t, expected, *d.Spec.Replicas)
+}
+
 func TestCheckHealthNotFound(t *testing.T) {
 	var deploymentsUpdated *[]*apps.Deployment
 	updateDeploymentAndWait, deploymentsUpdated = testopk8s.UpdateDeploymentAndWaitStub()