diff --git a/Documentation/ceph-cluster-crd.md b/Documentation/ceph-cluster-crd.md index 98fe17929..a0fe81eb6 100644 --- a/Documentation/ceph-cluster-crd.md +++ b/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. diff --git a/PendingReleaseNotes.md b/PendingReleaseNotes.md index 61bb1a3a1..f55921eab 100644 --- a/PendingReleaseNotes.md +++ b/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. diff --git a/pkg/daemon/ceph/client/test/mon.go b/pkg/daemon/ceph/client/test/mon.go index b355f18da..c72ad4073 100644 --- a/pkg/daemon/ceph/client/test/mon.go +++ b/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{} diff --git a/pkg/operator/ceph/cluster/cluster.go b/pkg/operator/ceph/cluster/cluster.go index 271f98686..94a009941 100644 --- a/pkg/operator/ceph/cluster/cluster.go +++ b/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 } diff --git a/pkg/operator/ceph/cluster/controller.go b/pkg/operator/ceph/cluster/controller.go index 6c10cd419..92ce6c80c 100644 --- a/pkg/operator/ceph/cluster/controller.go +++ b/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 diff --git a/pkg/operator/ceph/cluster/controller_test.go b/pkg/operator/ceph/cluster/controller_test.go index e6b48e67e..3b67c2d2a 100644 --- a/pkg/operator/ceph/cluster/controller_test.go +++ b/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) { diff --git a/pkg/operator/ceph/cluster/mon/health.go b/pkg/operator/ceph/cluster/mon/health.go index 0ecdda5fe..ecedce9eb 100644 --- a/pkg/operator/ceph/cluster/mon/health.go +++ b/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) diff --git a/pkg/operator/ceph/cluster/mon/health_test.go b/pkg/operator/ceph/cluster/mon/health_test.go index b082b9145..faa65b663 100644 --- a/pkg/operator/ceph/cluster/mon/health_test.go +++ b/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)) } diff --git a/pkg/operator/ceph/cluster/mon/mon.go b/pkg/operator/ceph/cluster/mon/mon.go index 2e556063c..a0512419f 100644 --- a/pkg/operator/ceph/cluster/mon/mon.go +++ b/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 diff --git a/pkg/operator/ceph/cluster/mon/mon_test.go b/pkg/operator/ceph/cluster/mon/mon_test.go index fccd56fc5..9b3075255 100644 --- a/pkg/operator/ceph/cluster/mon/mon_test.go +++ b/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,