Downscale one master at a time & don't remove last running master

operators/pkg/controller/elasticsearch/driver/downscale.go

@@ -5,36 +5,15 @@
 package driver
 
 import (
-	appsv1 "k8s.io/api/apps/v1"
-
-	"github.com/elastic/cloud-on-k8s/operators/pkg/apis/elasticsearch/v1alpha1"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/common/reconciler"
 	esclient "github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/client"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/label"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/migration"
-	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/observer"
-	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/reconcile"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/sset"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/version/zen1"
 	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/version/zen2"
-	"github.com/elastic/cloud-on-k8s/operators/pkg/utils/k8s"
 )
 
-// downscaleContext holds the context of this downscale, including clients and states,
-// propagated from the main driver.
-type downscaleContext struct {
-	// clients
-	k8sClient k8s.Client
-	esClient  esclient.Client
-	// driver states
-	resourcesState reconcile.ResourcesState
-	observedState  observer.State
-	reconcileState *reconcile.State
-	expectations   *reconciler.Expectations
-	// ES cluster
-	es v1alpha1.Elasticsearch
-}
-
 // HandleDownscale attempts to downscale actual StatefulSets towards expected ones.
 func HandleDownscale(
 	downscaleCtx downscaleContext,
@@ -60,9 +39,15 @@ func HandleDownscale(
 		return results.WithError(err)
 	}
 
+	// make sure we only downscale nodes we're allowed to
+	downscaleState, err := newDownscaleState(downscaleCtx.k8sClient, downscaleCtx.es)
+	if err != nil {
+		return results.WithError(err)
+	}
+
 	for _, downscale := range downscales {
 		// attempt the StatefulSet downscale (may or may not remove nodes)
-		requeue, err := attemptDownscale(downscaleCtx, downscale, leavingNodes, actualStatefulSets)
+		requeue, err := attemptDownscale(downscaleCtx, downscale, downscaleState, leavingNodes, actualStatefulSets)
 		if err != nil {
 			return results.WithError(err)
 		}
@@ -88,47 +73,6 @@ func noOnGoingDeletion(downscaleCtx downscaleContext, actualStatefulSets sset.St
 	return actualStatefulSets.PodReconciliationDone(downscaleCtx.k8sClient, downscaleCtx.es)
 }
 
-// ssetDownscale helps with the downscale of a single StatefulSet.
-// A StatefulSet removal (going from 0 to 0 replicas) is also considered as a Downscale here.
-type ssetDownscale struct {
-	statefulSet     appsv1.StatefulSet
-	initialReplicas int32
-	targetReplicas  int32
-}
-
-// leavingNodeNames returns names of the nodes that are supposed to leave the Elasticsearch cluster
-// for this StatefulSet. They are ordered by highest ordinal first;
-func (d ssetDownscale) leavingNodeNames() []string {
-	if d.targetReplicas >= d.initialReplicas {
-		return nil
-	}
-	leavingNodes := make([]string, 0, d.initialReplicas-d.targetReplicas)
-	for i := d.initialReplicas - 1; i >= d.targetReplicas; i-- {
-		leavingNodes = append(leavingNodes, sset.PodName(d.statefulSet.Name, i))
-	}
-	return leavingNodes
-}
-
-// isRemoval returns true if this downscale is a StatefulSet removal.
-func (d ssetDownscale) isRemoval() bool {
-	// StatefulSet does not have any replica, and should not have one
-	return d.initialReplicas == 0 && d.targetReplicas == 0
-}
-
-// isReplicaDecrease returns true if this downscale corresponds to decreasing replicas.
-func (d ssetDownscale) isReplicaDecrease() bool {
-	return d.targetReplicas < d.initialReplicas
-}
-
-// leavingNodeNames returns the names of all nodes that should leave the cluster (across StatefulSets).
-func leavingNodeNames(downscales []ssetDownscale) []string {
-	leavingNodes := []string{}
-	for _, d := range downscales {
-		leavingNodes = append(leavingNodes, d.leavingNodeNames()...)
-	}
-	return leavingNodes
-}
-
 // calculateDownscales compares expected and actual StatefulSets to return a list of ssetDownscale.
 // We also include StatefulSets removal (0 replicas) in those downscales.
 func calculateDownscales(expectedStatefulSets sset.StatefulSetList, actualStatefulSets sset.StatefulSetList) []ssetDownscale {
@@ -165,15 +109,21 @@ func scheduleDataMigrations(esClient esclient.Client, leavingNodes []string) err
 // or deletes the StatefulSet entirely if it should not contain any replica.
 // Nodes whose data migration is not over will not be removed.
 // A boolean is returned to indicate if a requeue should be scheduled if the entire downscale could not be performed.
-func attemptDownscale(ctx downscaleContext, downscale ssetDownscale, allLeavingNodes []string, statefulSets sset.StatefulSetList) (bool, error) {
+func attemptDownscale(
+	ctx downscaleContext,
+	downscale ssetDownscale,
+	state *downscaleState,
+	allLeavingNodes []string,
+	statefulSets sset.StatefulSetList,
+) (bool, error) {
 	switch {
 	case downscale.isRemoval():
 		ssetLogger(downscale.statefulSet).Info("Deleting statefulset")
 		return false, ctx.k8sClient.Delete(&downscale.statefulSet)
 
 	case downscale.isReplicaDecrease():
 		// adjust the theoretical downscale to one we can safely perform
-		performable := calculatePerformableDownscale(ctx, downscale, allLeavingNodes)
+		performable := calculatePerformableDownscale(ctx, state, downscale, allLeavingNodes)
 		if !performable.isReplicaDecrease() {
 			// no downscale can be performed for now, let's requeue
 			return true, nil
@@ -193,6 +143,7 @@ func attemptDownscale(ctx downscaleContext, downscale ssetDownscale, allLeavingN
 // It returns the updated downscale and a boolean indicating whether a requeue should be done.
 func calculatePerformableDownscale(
 	ctx downscaleContext,
+	state *downscaleState,
 	downscale ssetDownscale,
 	allLeavingNodes []string,
 ) ssetDownscale {
@@ -206,6 +157,10 @@ func calculatePerformableDownscale(
 	}
 	// iterate on all leaving nodes (ordered by highest ordinal first)
 	for _, node := range downscale.leavingNodeNames() {
+		if canDownscale, reason := checkDownscaleInvariants(*state, downscale.statefulSet); !canDownscale {
+			ssetLogger(downscale.statefulSet).V(1).Info("Cannot downscale StatefulSet", "node", node, "reason", reason)
+			return performableDownscale
+		}
 		if migration.IsMigratingData(ctx.observedState, node, allLeavingNodes) {
 			ssetLogger(downscale.statefulSet).V(1).Info("Data migration not over yet, skipping node deletion", "node", node)
 			ctx.reconcileState.UpdateElasticsearchMigrating(ctx.resourcesState, ctx.observedState)
@@ -214,6 +169,7 @@ func calculatePerformableDownscale(
 		}
 		// data migration over: allow pod to be removed
 		performableDownscale.targetReplicas--
+		state.recordOneRemoval(downscale.statefulSet)
 	}
 	return performableDownscale
 }

operators/pkg/controller/elasticsearch/driver/downscale_invariants.go

@@ -0,0 +1,68 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package driver
+
+import (
+	"github.com/elastic/cloud-on-k8s/operators/pkg/apis/elasticsearch/v1alpha1"
+	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/label"
+	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/reconcile"
+	"github.com/elastic/cloud-on-k8s/operators/pkg/controller/elasticsearch/sset"
+	"github.com/elastic/cloud-on-k8s/operators/pkg/utils/k8s"
+	appsv1 "k8s.io/api/apps/v1"
+)
+
+const (
+	OneMasterAtATimeInvariant        = "A master node is already in the process of being removed"
+	AtLeastOneRunningMasterInvariant = "Cannot remove the last running master node"
+)
+
+// checkDownscaleInvariants returns true if the given state state allows downscaling the given StatefulSet.
+// If not, it also returns the reason why.
+func checkDownscaleInvariants(state downscaleState, statefulSet appsv1.StatefulSet) (bool, string) {
+	if !label.IsMasterNodeSet(statefulSet) {
+		// only care about master nodes
+		return true, ""
+	}
+	if state.masterRemovalInProgress {
+		return false, OneMasterAtATimeInvariant
+	}
+	if state.runningMasters == 1 {
+		return false, AtLeastOneRunningMasterInvariant
+	}
+	return true, ""
+}
+
+// downscaleState tracks the state of a downscale to be checked against invariants
+type downscaleState struct {
+	// masterRemovalInProgress indicates whether a master node is in the process of being removed already.
+	masterRemovalInProgress bool
+	// runningMasters indicates how many masters are currently running in the cluster.
+	runningMasters int
+}
+
+// newDownscaleState creates a new downscaleState.
+func newDownscaleState(c k8s.Client, es v1alpha1.Elasticsearch) (*downscaleState, error) {
+	// retrieve the number of masters running ready
+	actualPods, err := sset.GetActualPodsForCluster(c, es)
+	if err != nil {
+		return nil, err
+	}
+	mastersReady := reconcile.AvailableElasticsearchNodes(label.FilterMasterNodePods(actualPods))
+
+	return &downscaleState{
+		masterRemovalInProgress: false,
+		runningMasters:          len(mastersReady),
+	}, nil
+}
+
+// recordOneRemoval updates the state to consider a 1-replica downscale of the given statefulSet.
+func (s *downscaleState) recordOneRemoval(statefulSet appsv1.StatefulSet) {
+	if !label.IsMasterNodeSet(statefulSet) {
+		// only care about master nodes
+		return
+	}
+	s.masterRemovalInProgress = true
+	s.runningMasters--
+}