Merge pull request #2211 from travisn/initialize-mons-on-restart

Allow mons to scale up or down

Documentation/ceph-cluster-crd.md

@@ -67,6 +67,13 @@ This feature is only available when `useAllNodes` has been set to `false`.
 - `count`: set the number of mons to be started. The number should be odd and between `1` and `9`. If not specified the default is set to `3` and `allowMultiplePerNode` is also set to `true`.
 - `allowMultiplePerNode`: enable (`true`) or disable (`false`) the placement of multiple mons on one node. Default is `false`.
 
+If these settings are changed in the CRD the operator will update the number of mons during a periodic check of the mon health, which by default is every 45 seconds.
+
+To change the defaults that the operator uses to determine the mon health and whether to failover a mon, the following environment variables can be changed in [operator.yaml](https://github.com/rook/rook/blob/master/cluster/examples/kubernetes/ceph/operator.yaml). The intervals should be small enough that you have confidence the mons will maintain quorum, while also being
+log enough to ignore network blips where mons are failed over too often.
+- ROOK_MON_HEALTHCHECK_INTERVAL: The frequency with which to check if mons are in quorum (default is 45 seconds)
+- ROOK_MON_OUT_TIMEOUT: The interval to wait before marking a mon as "out" and starting a new mon to replace it in the quroum (default is 5 minutes)
+
 ### Node Settings
 In addition to the cluster level settings specified above, each individual node can also specify configuration to override the cluster level settings and defaults.
 If a node does not specify any configuration then it will inherit the cluster level settings.

PendingReleaseNotes.md

@@ -14,6 +14,7 @@
 - K8s client-go updated from version 1.8.2 to 1.11.3
 - The toolbox manifest now creates a deployment based on the `rook/ceph` image instead of creating a pod on a specialized `rook/ceph-toolbox` image.
 - The frequency of discovering devices on a node is reduced to 60 minutes by default, and is configurable with the setting `ROOK_DISCOVER_DEVICES_INTERVAL` in operator.yaml.
+- The number of mons can be changed by updating the `mon.count` in the cluster CRD.
 
 ## Breaking Changes
 - Ceph mons are [named consistently](https://github.com/rook/rook/issues/1751) with other daemons with the letters a, b, c, etc.

pkg/daemon/ceph/client/test/mon.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	"github.com/rook/rook/pkg/daemon/ceph/client"
+	"github.com/rook/rook/pkg/daemon/ceph/config"
 )
 
 func MonInQuorumResponse() string {
@@ -35,6 +36,22 @@ func MonInQuorumResponse() string {
 	return string(serialized)
 }
 
+func MonInQuorumResponseFromMons(mons map[string]*config.MonInfo) string {
+	resp := client.MonStatusResponse{Quorum: []int{}}
+	i := 0
+	for name := range mons {
+		resp.MonMap.Mons = append(resp.MonMap.Mons, client.MonMapEntry{
+			Name:    name,
+			Rank:    i,
+			Address: fmt.Sprintf("1.2.3.%d", i),
+		})
+		resp.Quorum = append(resp.Quorum, i)
+		i++
+	}
+	serialized, _ := json.Marshal(resp)
+	return string(serialized)
+}
+
 func MonInQuorumResponseMany(count int) string {
 	resp := client.MonStatusResponse{Quorum: []int{0}}
 	resp.MonMap.Mons = []client.MonMapEntry{}

pkg/operator/ceph/cluster/cluster.go

@@ -163,7 +163,7 @@ func (c *cluster) createInitialCrushMap() error {
 	return nil
 }
 
-func clusterChanged(oldCluster, newCluster cephv1beta1.ClusterSpec) bool {
+func clusterChanged(oldCluster, newCluster cephv1beta1.ClusterSpec, clusterRef *cluster) bool {
 	changeFound := false
 	oldStorage := oldCluster.Storage
 	newStorage := newCluster.Storage
@@ -181,5 +181,19 @@ func clusterChanged(oldCluster, newCluster cephv1beta1.ClusterSpec) bool {
 		changeFound = true
 	}
 
+	if oldCluster.Mon.Count != newCluster.Mon.Count {
+		logger.Infof("number of mons have changed from %d to %d. The health check will update the mons...", oldCluster.Mon.Count, newCluster.Mon.Count)
+		clusterRef.mons.MonCountMutex.Lock()
+		clusterRef.mons.Count = newCluster.Mon.Count
+		clusterRef.mons.MonCountMutex.Unlock()
+	}
+
+	if oldCluster.Mon.AllowMultiplePerNode != newCluster.Mon.AllowMultiplePerNode {
+		logger.Infof("allow multiple mons per node changed from %t to %t. The health check will update the mons...", oldCluster.Mon.AllowMultiplePerNode, newCluster.Mon.AllowMultiplePerNode)
+		clusterRef.mons.MonCountMutex.Lock()
+		clusterRef.mons.AllowMultiplePerNode = newCluster.Mon.AllowMultiplePerNode
+		clusterRef.mons.MonCountMutex.Unlock()
+	}
+
 	return changeFound
 }

pkg/operator/ceph/cluster/controller.go

@@ -289,8 +289,13 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
 		c.removeFinalizer(newClust)
 		return
 	}
+	cluster, ok := c.clusterMap[newClust.Namespace]
+	if !ok {
+		logger.Errorf("Cannot update cluster %s that does not exist", newClust.Namespace)
+		return
+	}
 
-	if !clusterChanged(oldClust.Spec, newClust.Spec) {
+	if !clusterChanged(oldClust.Spec, newClust.Spec, cluster) {
 		logger.Infof("update event for cluster %s is not supported", newClust.Namespace)
 		return
 	}
@@ -299,11 +304,6 @@ func (c *ClusterController) onUpdate(oldObj, newObj interface{}) {
 	logger.Debugf("old cluster: %+v", oldClust.Spec)
 	logger.Debugf("new cluster: %+v", newClust.Spec)
 
-	cluster, ok := c.clusterMap[newClust.Namespace]
-	if !ok {
-		logger.Errorf("Cannot update cluster %s that does not exist", newClust.Namespace)
-		return
-	}
 	cluster.Spec = &newClust.Spec
 
 	// attempt to update the cluster.  note this is done outside of wait.Poll because that function

pkg/operator/ceph/cluster/controller_test.go

@@ -28,6 +28,7 @@ import (
 	rookfake "github.com/rook/rook/pkg/client/clientset/versioned/fake"
 	"github.com/rook/rook/pkg/clusterd"
 	"github.com/rook/rook/pkg/daemon/ceph/agent/flexvolume/attachment"
+	"github.com/rook/rook/pkg/operator/ceph/cluster/mon"
 	"github.com/rook/rook/pkg/operator/k8sutil"
 	testop "github.com/rook/rook/pkg/operator/test"
 	exectest "github.com/rook/rook/pkg/util/exec/test"
@@ -130,7 +131,9 @@ func TestClusterChanged(t *testing.T) {
 			},
 		},
 	}
-	assert.True(t, clusterChanged(old, new))
+	c := &cluster{Spec: &cephv1beta1.ClusterSpec{}, mons: &mon.Cluster{}}
+	assert.True(t, clusterChanged(old, new, c))
+	assert.Equal(t, 0, c.Spec.Mon.Count)
 
 	// a node was removed, should be a change
 	old.Storage.Nodes = []rookalpha.Node{
@@ -140,7 +143,7 @@ func TestClusterChanged(t *testing.T) {
 	new.Storage.Nodes = []rookalpha.Node{
 		{Name: "node1", Selection: rookalpha.Selection{Devices: []rookalpha.Device{{Name: "sda"}}}},
 	}
-	assert.True(t, clusterChanged(old, new))
+	assert.True(t, clusterChanged(old, new, c))
 
 	// the nodes being in a different order should not be a change
 	old.Storage.Nodes = []rookalpha.Node{
@@ -151,7 +154,16 @@ func TestClusterChanged(t *testing.T) {
 		{Name: "node2", Selection: rookalpha.Selection{Devices: []rookalpha.Device{{Name: "sda"}}}},
 		{Name: "node1", Selection: rookalpha.Selection{Devices: []rookalpha.Device{{Name: "sda"}}}},
 	}
-	assert.False(t, clusterChanged(old, new))
+	assert.False(t, clusterChanged(old, new, c))
+	assert.Equal(t, 0, c.Spec.Mon.Count)
+
+	// If the number of mons changes, the mon count on the cluster should be updated so the health check can adjust the mons
+	new.Mon.Count = 3
+	new.Mon.AllowMultiplePerNode = true
+	assert.False(t, c.mons.AllowMultiplePerNode)
+	assert.False(t, clusterChanged(old, new, c))
+	assert.Equal(t, 3, c.mons.Count)
+	assert.True(t, c.mons.AllowMultiplePerNode)
 }
 
 func TestRemoveFinalizer(t *testing.T) {

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

@@ -67,7 +67,14 @@ func (hc *HealthChecker) Check(stopCh chan struct{}) {
 }
 
 func (c *Cluster) checkHealth() error {
-	logger.Debugf("Checking health for mons. %+v", c.clusterInfo)
+	logger.Debugf("Checking health for mons (desired=%d). %+v", c.Count, c.clusterInfo)
+
+	// Use a local mon count in case the user updates the crd in another goroutine.
+	// We need to complete a health check with a consistent value.
+	c.MonCountMutex.Lock()
+	desiredMonCount := c.Count
+	allowMultiplePerNode := c.AllowMultiplePerNode
+	c.MonCountMutex.Unlock()
 
 	// connect to the mons
 	// get the status and check for quorum
@@ -85,6 +92,7 @@ func (c *Cluster) checkHealth() error {
 
 	// first handle mons that are not in quorum but in the ceph mon map
 	// failover the unhealthy mons
+	allMonsInQuorum := true
 	for _, mon := range status.MonMap.Mons {
 		inQuorum := monInQuorum(mon, status.Quorum)
 		// if the mon is in quorum remove it from our check for "existence"
@@ -93,15 +101,15 @@ func (c *Cluster) checkHealth() error {
 			delete(monsNotFound, mon.Name)
 		} else {
 			// when the mon isn't in the clusterInfo, but is in quorum and there are
-			//enough mons, remove it else remove it on the next run
-			if inQuorum && len(status.MonMap.Mons) > c.Size {
+			// enough mons, remove it else remove it on the next run
+			if inQuorum && len(status.MonMap.Mons) > desiredMonCount {
 				logger.Warningf("mon %s not in source of truth but in quorum, removing", mon.Name)
 				c.removeMon(mon.Name)
 			} else {
 				logger.Warningf(
 					"mon %s not in source of truth and not in quorum, not enough mons to remove now (wanted: %d, current: %d)",
 					mon.Name,
-					c.Size,
+					desiredMonCount,
 					len(status.MonMap.Mons),
 				)
 			}
@@ -116,6 +124,7 @@ func (c *Cluster) checkHealth() error {
 			}
 		} else {
 			logger.Debugf("mon %s NOT found in quorum. Mon status: %+v", mon.Name, status)
+			allMonsInQuorum = false
 
 			// If not yet set, add the current time, for the timeout
 			// calculation, to the list
@@ -131,7 +140,7 @@ func (c *Cluster) checkHealth() error {
 			}
 
 			logger.Warningf("mon %s NOT found in quorum and timeout exceeded, mon will be failed over", mon.Name)
-			c.failMon(len(status.MonMap.Mons), mon.Name)
+			c.failMon(len(status.MonMap.Mons), desiredMonCount, mon.Name)
 			// only deal with one unhealthy mon per health check
 			return nil
 		}
@@ -141,32 +150,44 @@ func (c *Cluster) checkHealth() error {
 	// handle all mons that haven't been in the Ceph mon map
 	for mon := range monsNotFound {
 		logger.Warningf("mon %s NOT found in ceph mon map, failover", mon)
-		c.failMon(len(c.clusterInfo.Monitors), mon)
+		c.failMon(len(c.clusterInfo.Monitors), desiredMonCount, mon)
 		// only deal with one "not found in ceph mon map" mon per health check
 		return nil
 	}
 
-	// check if there are more than two mons running on the same node, failover one mon in that case
-	done, err := c.checkMonsOnSameNode()
+	if !allowMultiplePerNode {
+		// check if there are more than two mons running on the same node, failover one mon in that case
+		done, err := c.checkMonsOnSameNode(desiredMonCount)
+		if done || err != nil {
+			return err
+		}
+	}
+
+	done, err := c.checkMonsOnValidNodes()
 	if done || err != nil {
 		return err
 	}
 
-	done, err = c.checkMonsOnValidNodes()
-	if done || err != nil {
-		return err
-	}
-
-	// create/start new mons when there are less mons
-	if len(status.MonMap.Mons) < c.Size {
-		logger.Infof("found only %d mons less than given mon.count %d, starting more mons", len(status.MonMap.Mons), c.Size)
+	// create/start new mons when there are fewer mons than the desired count in the CRD
+	if len(status.MonMap.Mons) < desiredMonCount {
+		logger.Infof("adding mons. currently %d mons are in quorum and the desired count is %d.", len(status.MonMap.Mons), desiredMonCount)
 		return c.startMons()
 	}
 
+	// remove extra mons if the desired count has decreased in the CRD and all the mons are currently healthy
+	if allMonsInQuorum && len(status.MonMap.Mons) > desiredMonCount {
+		if desiredMonCount < 2 && len(status.MonMap.Mons) == 2 {
+			logger.Warningf("cannot reduce mon quorum size from 2 to 1")
+		} else {
+			logger.Infof("removing an extra mon. currently %d are in quorum and only %d are desired", len(status.MonMap.Mons), desiredMonCount)
+			return c.removeMon(status.MonMap.Mons[0].Name)
+		}
+	}
+
 	return nil
 }
 
-func (c *Cluster) checkMonsOnSameNode() (bool, error) {
+func (c *Cluster) checkMonsOnSameNode(desiredMonCount int) (bool, error) {
 	nodesUsed := map[string]struct{}{}
 	for name, node := range c.mapping.Node {
 		// when the node is already in the list we have more than one mon on that node
@@ -180,7 +201,7 @@ func (c *Cluster) checkMonsOnSameNode() (bool, error) {
 			// fail it over to an other node
 			if len(availableNodes) > 0 {
 				logger.Infof("rebalance: enough nodes available %d to failover mon %s", len(availableNodes), name)
-				c.failMon(len(c.clusterInfo.Monitors), name)
+				c.failMon(len(c.clusterInfo.Monitors), desiredMonCount, name)
 			} else {
 				logger.Debugf("rebalance: not enough nodes available to failover mon %s", name)
 			}
@@ -214,9 +235,9 @@ func (c *Cluster) checkMonsOnValidNodes() (bool, error) {
 	return false, nil
 }
 
-// failMon compares the monCount against c.Size (wanted mon count)
-func (c *Cluster) failMon(monCount int, name string) {
-	if monCount > c.Size {
+// failMon compares the monCount against desiredMonCount
+func (c *Cluster) failMon(monCount, desiredMonCount int, name string) {
+	if monCount > desiredMonCount {
 		// no need to create a new mon since we have an extra
 		if err := c.removeMon(name); err != nil {
 			logger.Errorf("failed to remove mon %s. %+v", name, err)

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

@@ -20,9 +20,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
-	"testing"
-
 	"os"
+	"testing"
 
 	cephv1beta1 "github.com/rook/rook/pkg/apis/ceph.rook.io/v1beta1"
 	rookalpha "github.com/rook/rook/pkg/apis/rook.io/v1alpha2"
@@ -199,7 +198,7 @@ func TestCheckHealthTwoMonsOneNode(t *testing.T) {
 	}
 
 	// initial health check should already see that there is more than one mon on one node (node0)
-	_, err := c.checkMonsOnSameNode()
+	_, err := c.checkMonsOnSameNode(3)
 	assert.Nil(t, err)
 	assert.Equal(t, "node0", c.mapping.Node["a"].Name)
 	assert.Equal(t, "node0", c.mapping.Node["b"].Name)
@@ -225,7 +224,7 @@ func TestCheckHealthTwoMonsOneNode(t *testing.T) {
 	n.Name = "node2"
 	clientset.CoreV1().Nodes().Create(n)
 
-	_, err = c.checkMonsOnSameNode()
+	_, err = c.checkMonsOnSameNode(3)
 	assert.Nil(t, err)
 
 	// check that mon c exists
@@ -247,7 +246,7 @@ func TestCheckHealthTwoMonsOneNode(t *testing.T) {
 
 	// enable different ceph mon map output
 	executorNextMons = true
-	_, err = c.checkMonsOnSameNode()
+	_, err = c.checkMonsOnSameNode(3)
 	assert.Nil(t, err)
 
 	// check that nothing has changed
@@ -347,10 +346,11 @@ func TestCheckMonsValid(t *testing.T) {
 	assert.Equal(t, "node1", c.mapping.Node["b"].Name)
 }
 
-func TestCheckLessMonsStartNewMons(t *testing.T) {
+func TestAddRemoveMons(t *testing.T) {
+	monQuorumResponse := clienttest.MonInQuorumResponse()
 	executor := &exectest.MockExecutor{
 		MockExecuteCommandWithOutputFile: func(debug bool, actionName string, command string, outFileArg string, args ...string) (string, error) {
-			return clienttest.MonInQuorumResponse(), nil
+			return monQuorumResponse, nil
 		},
 	}
 	clientset := test.New(1)
@@ -368,7 +368,34 @@ func TestCheckLessMonsStartNewMons(t *testing.T) {
 	c.waitForStart = false
 	defer os.RemoveAll(c.context.ConfigDir)
 
+	// checking the health will increase the mons as desired all in one go
 	err := c.checkHealth()
 	assert.Nil(t, err)
 	assert.Equal(t, 5, len(c.clusterInfo.Monitors), fmt.Sprintf("mons: %v", c.clusterInfo.Monitors))
+
+	// reducing the mon count to 3 will reduce the mon count once each time we call checkHealth
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
+	c.Count = 3
+	err = c.checkHealth()
+	assert.Nil(t, err)
+	assert.Equal(t, 4, len(c.clusterInfo.Monitors))
+
+	// after the second call we will be down to the expected count of 3
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
+	err = c.checkHealth()
+	assert.Nil(t, err)
+	assert.Equal(t, 3, len(c.clusterInfo.Monitors))
+
+	// now attempt to reduce the mons down to quorum size 1
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
+	c.Count = 1
+	err = c.checkHealth()
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(c.clusterInfo.Monitors))
+
+	// cannot reduce from quorum size of 2 to 1
+	monQuorumResponse = clienttest.MonInQuorumResponseFromMons(c.clusterInfo.Monitors)
+	err = c.checkHealth()
+	assert.Nil(t, err)
+	assert.Equal(t, 2, len(c.clusterInfo.Monitors))
 }

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

@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/coreos/pkg/capnslog"
@@ -75,8 +76,9 @@ type Cluster struct {
 	Namespace            string
 	Keyring              string
 	Version              string
-	Size                 int
+	Count                int
 	AllowMultiplePerNode bool
+	MonCountMutex        sync.Mutex
 	Port                 int32
 	clusterInfo          *cephconfig.ClusterInfo
 	placement            rookalpha.Placement
@@ -126,7 +128,7 @@ func New(context *clusterd.Context, namespace, dataDirHostPath, version string,
 		dataDirHostPath:      dataDirHostPath,
 		Namespace:            namespace,
 		Version:              version,
-		Size:                 mon.Count,
+		Count:                mon.Count,
 		AllowMultiplePerNode: mon.AllowMultiplePerNode,
 		maxMonID:             -1,
 		waitForStart:         true,
@@ -157,7 +159,7 @@ func (c *Cluster) Start() error {
 
 func (c *Cluster) startMons() error {
 	// init the mons config
-	mons := c.initMonConfig(c.Size)
+	mons := c.initMonConfig(c.Count)
 
 	// Assign the pods to nodes
 	if err := c.assignMons(mons); err != nil {
@@ -165,12 +167,13 @@ func (c *Cluster) startMons() error {
 	}
 
 	// Start one monitor at a time
-	for i := 0; i < c.Size; i++ {
+	for i := 0; i < c.Count; i++ {
 		logger.Infof("ensuring mon %s (%s) is started", mons[i].ResourceName, mons[i].DaemonName)
 		endIndex := len(c.clusterInfo.Monitors)
-		if endIndex < c.Size {
+		if endIndex < c.Count {
 			endIndex++
 		}
+		logger.Infof("looping to start mons. i=%d, endIndex=%d, c.Size=%d", i, endIndex, c.Count)
 
 		// Init the mon IPs
 		if err := c.initMonIPs(mons[0:endIndex]); err != nil {
@@ -319,11 +322,6 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
 		return fmt.Errorf("failed to get available nodes for mons. %+v", err)
 	}
 
-	// if all nodes already have mons return error as we don't place two mons on one node
-	if len(availableNodes) == 0 {
-		return fmt.Errorf("no nodes available for mon placement")
-	}
-
 	nodeIndex := 0
 	for _, m := range mons {
 		if _, ok := c.mapping.Node[m.DaemonName]; ok {
@@ -331,6 +329,11 @@ func (c *Cluster) assignMons(mons []*monConfig) error {
 			continue
 		}
 
+		// if we need to place a new mon and don't have any more nodes available, we fail to add the mon
+		if len(availableNodes) == 0 {
+			return fmt.Errorf("no nodes available for mon placement")
+		}
+
 		// pick one of the available nodes where the mon will be assigned
 		node := availableNodes[nodeIndex%len(availableNodes)]
 		logger.Debugf("mon %s assigned to node %s", m.DaemonName, node.Name)
@@ -379,12 +382,13 @@ func (c *Cluster) startDeployments(mons []*monConfig, index int) error {
 		return fmt.Errorf("cannot start 0 mons")
 	}
 
-	// Start the last mon in the list. The others have already been started by a previous call.
-	m := mons[index]
-	node, _ := c.mapping.Node[m.DaemonName]
-	err := c.startMon(m, node.Hostname)
-	if err != nil {
-		return fmt.Errorf("failed to create mon %s. %+v", m.DaemonName, err)
+	// Ensure each of the mons have been created. If already created, it will be a no-op.
+	for i := 0; i < len(mons); i++ {
+		node, _ := c.mapping.Node[mons[i].DaemonName]
+		err := c.startMon(mons[i], node.Hostname)
+		if err != nil {
+			return fmt.Errorf("failed to create mon %s. %+v", mons[i].DaemonName, err)
+		}
 	}
 
 	logger.Infof("mons created: %d", len(mons))
@@ -554,7 +558,7 @@ func (c *Cluster) startMon(m *monConfig, hostname string) error {
 		if !errors.IsAlreadyExists(err) {
 			return fmt.Errorf("failed to create mon %s. %+v", m.ResourceName, err)
 		}
-		logger.Infof("mon %s already exists", m.ResourceName)
+		logger.Debugf("mon %s already exists", m.ResourceName)
 	}
 
 	return nil

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

@@ -69,7 +69,7 @@ func newCluster(context *clusterd.Context, namespace string, hostNetwork bool, r
 		context:              context,
 		Namespace:            namespace,
 		Version:              "myversion",
-		Size:                 3,
+		Count:                3,
 		AllowMultiplePerNode: true,
 		maxMonID:             -1,
 		waitForStart:         false,