diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index fde34f77a..f6c0deb97 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/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) } diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index 9c6734b83..298d6e32b 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/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()