CLOUDP-288551: Block scaling both ways at runtime (#4099)

# Summary

This PR blocks reconciliation in case we attempt to scale components up
and down at the same time (both ways).

When changing member distribution of a component, they should only scale
down, or up.
A component can be a shard, config servers or mongos
E.g (for a distribution of a component over 3 clusters):
- Scaling from 1 2 2 to 3 3 2 would be valid
- Scaling from 2 2 2 to 1 2 3 would be invalid, as the component is
scaling from 1 to 2 members on cluster 1, but from 2 to 3 on cluster 3.

Multiple cases are illustrated in unit tests.

We need to be defensive in these cases, because it can create
inconsistencies in the behaviour of the operator. For example when we
decide if we should publish the automation config first, we check at the
cluster level, so we can detect that we scale down, whereas the global
replica set (across all clusters) is scaling up.

This change will also impact single cluster sharded clusters (e.g
scaling up shards, but scaling down config servers)

**Note:** we chose to do this check at runtime rather than in webhooks.
Doing it at the webhook level would require to duplicate a lot of logic
from the reconciler, to compute the desired configuration based on shard
overrides, member config...
With the scalers available, it is much easier and makes more sense to do
it at runtime.

## List of changes

- New method in sharded reconciler to block reconciliation if needed.
- Fixed a bug which might cause losing the current sizes from the
deployment state when upgrading the operator from <=1.27
(10gen/ops-manager-kubernetes@0fa2d50)
- Unit test for this mechanism in various scaling scenarios.
- Modified the `MultiClusterReplicaSetScaler` interface, to avoid type
casting.
- Fixed E2E tests that were attempting both ways reconciliation.
- Added a constant for "slaney", the default name of a sharded cluster
in unit tests.

## Documentation

This ticket will be closed with documentation changes required to update
our public documentation.

## Proof of Work

Unit test `TestBlockReconcileScalingBothWays`
Example error message:
```
Cannot perform scale up and scale down operations at the same time. Scaling Up: [Component=shard idx 0, Cluster=__default, Current=0, Desired=2;], Scaling Down: [Component=configSrv, Cluster=__default, Current=3, Desired=2;]"
```

**Note:** The e2e test `e2e_operator_upgrade_sharded_cluster` is failing
and we are aware of that. It is related to how the state configmap is
handled during a migration. We opened this PR nonetheless as this
failure is not directly related to these changes, but we are working on
adding a fix.

---------

Co-authored-by: Łukasz Sierant <[email protected]>

Author: 2 people

RELEASE_NOTES.md

@@ -2,6 +2,9 @@
 <!-- Next Release -->
 # MongoDB Enterprise Kubernetes Operator 1.32.0
 
+## New Features
+* **MongoDB**: To ensure the correctness of scaling operations, a new validation has been added to Sharded Cluster deployments. This validation restricts scaling different components in two directions simultaneously within a single change to the YAML file. For example, it is not allowed to add more nodes (scaling up) to shards while simultaneously removing (scaling down) config servers or mongos. This restriction also applies to multi-cluster deployments. A simple change that involves "moving" one node from one cluster to another—without altering the total number of members—will now be blocked. It is necessary to perform a scale-up operation first and then execute a separate change for scaling down.
+
 ## Bug Fixes
 * Fixes the bug when status of `MongoDBUser` was being set to `Updated` prematurely. For example, new users were not immediately usable following `MongoDBUser` creation despite the operator reporting `Updated` state.
 * Fixed a bug causing cluster health check issues when ordering of users and tokens differed in Kubeconfig.

controllers/om/automation_status_test.go

@@ -39,7 +39,8 @@ func TestCheckAutomationStatusIsGoal(t *testing.T) {
 				relevantProcesses: []string{"a", "b"},
 			},
 			expectedResult: true,
-			expectedMsg:    "processes that reached goal state: [a b]",
+			// We can not check for the full message as the ordering of the processes won't be deterministic (stored in a map)
+			expectedMsg: "processes that reached goal state:",
 		}, {
 			name: "one not in goal state",
 			args: args{

controllers/operator/common_controller_test.go

@@ -480,7 +480,7 @@ func checkReconcilePending(ctx context.Context, t *testing.T, reconciler reconci
 	assert.Nil(t, e, "When pending, error should be nil")
 	assert.Equal(t, failedResult, result)
 
-	// also need to make sure the object status is updated to failed
+	// also need to make sure the object status is pending
 	assert.NoError(t, client.Get(ctx, mock.ObjectKeyFromApiObject(object), object))
 	assert.Equal(t, status.PhasePending, object.Status.Phase)
 	assert.Contains(t, object.Status.Message, expectedErrorMessage)

controllers/operator/construct/scalers/appdb_scaler.go

@@ -33,6 +33,10 @@ func (s *appDBSingleClusterScaler) DesiredReplicas() int {
 	return s.opsManager.Spec.AppDB.Members
 }
 
+func (s *appDBSingleClusterScaler) TargetReplicas() int {
+	return s.DesiredReplicas()
+}
+
 func (s *appDBSingleClusterScaler) CurrentReplicas() int {
 	return s.opsManager.Status.AppDbStatus.Members
 }
@@ -48,3 +52,5 @@ func (s *appDBSingleClusterScaler) MemberClusterName() string {
 func (s *appDBSingleClusterScaler) MemberClusterNum() int {
 	return 0
 }
+
+func (s *appDBSingleClusterScaler) ScalerDescription() string { return "AppDB" }

controllers/operator/construct/scalers/interfaces/interfaces.go

@@ -5,6 +5,9 @@ import "github.com/mongodb/mongodb-kubernetes-operator/pkg/util/scale"
 type MultiClusterReplicaSetScaler interface {
 	scale.ReplicaSetScaler
 	ScalingFirstTime() bool
+	TargetReplicas() int
 	MemberClusterName() string
 	MemberClusterNum() int
+	// ScalerDescription contains the name of the component associated to that scaler (shard, config server, AppDB...)
+	ScalerDescription() string
 }

controllers/operator/construct/scalers/replicaset_scaler.go

@@ -142,6 +142,10 @@ func (s *MultiClusterReplicaSetScaler) MemberClusterNum() int {
 	return s.memberClusterNum
 }
 
+func (s *MultiClusterReplicaSetScaler) ScalerDescription() string {
+	return s.scalerDescription
+}
+
 func (s *MultiClusterReplicaSetScaler) String() string {
 	return fmt.Sprintf("{MultiClusterReplicaSetScaler (%s): still scaling: %t (finishing this reconcile: %t), clusterName=%s, clusterIdx=%d, current/target replicas:%d/%d, "+
 		"replicas this reconciliation: %d, scaling first time: %t}", s.scalerDescription, s.CurrentReplicas() != s.TargetReplicas(), scale.ReplicasThisReconciliation(s) == s.TargetReplicas(), s.memberClusterName, s.memberClusterNum,

controllers/operator/mongodbshardedcluster_controller.go

@@ -91,6 +91,21 @@ type ShardedClusterDeploymentState struct {
 	Status                *mdbv1.MongoDbStatus `json:"status"`
 }
 
+// updateStatusFromResourceStatus updates the status in the deployment state with values from the resource status with additional ensurance that no data is accidentally lost.
+// In a rare situation when we're performing an upgrade of the operator from non-deployment state version (<=1.27) the migrateToNewDeploymentState
+// function correctly migrates the sizes of the cluster, but then, in case of an early return (in case of any error or waiting too long for the sts/agents)
+// the updateStatus might clear the migrated data.
+// This function ensures we're copying the status, but at the same time we're not losing those sizes from the deployment state.
+// The logic of updateStatus in the reconciler works on options. If the option is not passed, the value is not updated, but it's also not cleared if the option is not passed.
+// Early returns with updateStatus don't pass any options, so the calculated status shouldn't clear the sizes we've just calculated into the deployment state.
+func (s *ShardedClusterDeploymentState) updateStatusFromResourceStatus(statusFromResource mdbv1.MongoDbStatus) {
+	resultStatus := statusFromResource.DeepCopy()
+	if resultStatus.SizeStatusInClusters == nil && s.Status.SizeStatusInClusters != nil {
+		resultStatus.SizeStatusInClusters = s.Status.SizeStatusInClusters.DeepCopy()
+	}
+	s.Status = resultStatus
+}
+
 func NewShardedClusterDeploymentState() *ShardedClusterDeploymentState {
 	return &ShardedClusterDeploymentState{
 		CommonDeploymentState: CommonDeploymentState{ClusterMapping: map[string]int{}},
@@ -204,6 +219,11 @@ func (r *ShardedClusterReconcileHelper) createShardsMemberClusterLists(shardsMap
 					return count
 				}
 			}
+			// Because we store one common distribution for all shards in ShardMongodsInClusters, we need to make sure
+			// we assign a size of 0 to newly created shards, as they haven't scaled yet.
+			if shardIdx >= deploymentState.Status.ShardCount {
+				return 0
+			}
 			if count, ok := deploymentState.Status.SizeStatusInClusters.ShardMongodsInClusters[memberClusterName]; ok {
 				// Otherwise get the default one ShardMongodsInClusters
 				// ShardMongodsInClusters is not correct in the edge case where all shards are overridden
@@ -647,6 +667,41 @@ func NewShardedClusterReconcilerHelper(ctx context.Context, reconciler *Reconcil
 	return helper, nil
 }
 
+func blockScalingBothWays(desiredReplicasScalers []interfaces.MultiClusterReplicaSetScaler) error {
+	scalingUp := false
+	scalingDown := false
+	var scalingUpLogs []string
+	var scalingDownLogs []string
+
+	// We have one scaler instance per component per cluster. That means we block scaling both ways across components,
+	// but also within a single component
+	// For example, if a component (e.g the config server) tries to scale up on member cluster 1 and scale down on
+	// member cluster 2, reconciliation will be blocked, even if the total number of replicas for this component stays
+	// the same.
+	for _, mcScaler := range desiredReplicasScalers {
+		desired := mcScaler.TargetReplicas()
+		current := mcScaler.CurrentReplicas()
+		logMessage := fmt.Sprintf("Component=%s, Cluster=%s, Current=%d, Desired=%d;", mcScaler.ScalerDescription(), mcScaler.MemberClusterName(), current, desired)
+		if desired > current {
+			scalingUp = true
+			scalingUpLogs = append(scalingUpLogs, logMessage)
+		}
+		if desired < current {
+			scalingDown = true
+			scalingDownLogs = append(scalingDownLogs, logMessage)
+		}
+	}
+
+	if scalingUp && scalingDown {
+		return xerrors.Errorf(
+			"Cannot perform scale up and scale down operations at the same time. Scaling Up: %v, Scaling Down: %v",
+			scalingUpLogs, scalingDownLogs,
+		)
+	}
+
+	return nil
+}
+
 func (r *ShardedClusterReconcileHelper) initializeStateStore(ctx context.Context, reconciler *ReconcileCommonController, sc *mdbv1.MongoDB, log *zap.SugaredLogger) error {
 	r.deploymentState = NewShardedClusterDeploymentState()
 
@@ -724,17 +779,26 @@ func (r *ShardedClusterReconcileHelper) Reconcile(ctx context.Context, log *zap.
 		return r.commonController.updateStatus(ctx, sc, workflow.Invalid("%s", err.Error()), log)
 	}
 
-	if !architectures.IsRunningStaticArchitecture(sc.Annotations) {
-		agents.UpgradeAllIfNeeded(ctx, agents.ClientSecret{Client: r.commonController.client, SecretClient: r.commonController.SecretClient}, r.omConnectionFactory, GetWatchedNamespace(), false)
-	}
-
 	log.Info("-> ShardedCluster.Reconcile")
 	log.Infow("ShardedCluster.Spec", "spec", sc.Spec)
 	log.Infow("ShardedCluster.Status", "status", r.deploymentState.Status)
 	log.Infow("ShardedCluster.deploymentState", "sizeStatus", r.deploymentState.Status.MongodbShardedClusterSizeConfig, "sizeStatusInClusters", r.deploymentState.Status.SizeStatusInClusters)
 
 	r.logAllScalers(log)
 
+	// After processing normal validations, we check for conflicting scale-up and scale-down operations within the same
+	// reconciliation cycle. If both scaling directions are detected, we block the reconciliation.
+	// This is not currently possible to do it safely with the operator. We check direction of scaling to decide for
+	// global operations like publishing AC first.
+	// Therefore, we can obtain inconsistent behaviour in case scaling goes in both directions.
+	if err := blockScalingBothWays(r.getAllScalers()); err != nil {
+		return r.updateStatus(ctx, sc, workflow.Failed(err), log)
+	}
+
+	if !architectures.IsRunningStaticArchitecture(sc.Annotations) {
+		agents.UpgradeAllIfNeeded(ctx, agents.ClientSecret{Client: r.commonController.client, SecretClient: r.commonController.SecretClient}, r.omConnectionFactory, GetWatchedNamespace(), false)
+	}
+
 	projectConfig, credsConfig, err := project.ReadConfigAndCredentials(ctx, r.commonController.client, r.commonController.SecretClient, sc, log)
 	if err != nil {
 		return r.updateStatus(ctx, sc, workflow.Failed(err), log)
@@ -2238,7 +2302,7 @@ func (r *ShardedClusterReconcileHelper) updateStatus(ctx context.Context, resour
 	} else {
 		// UpdateStatus in the sharded cluster controller should be executed only once per reconcile (always with a return)
 		// We are saving the status and writing back to the state configmap at this time
-		r.deploymentState.Status = resource.Status.DeepCopy()
+		r.deploymentState.updateStatusFromResourceStatus(resource.Status)
 		if err := r.stateStore.WriteState(ctx, r.deploymentState, log); err != nil {
 			return r.commonController.updateStatus(ctx, resource, workflow.Failed(xerrors.Errorf("Failed to write deployment state after updating status: %w", err)), log, nil)
 		}
@@ -2282,11 +2346,12 @@ func (r *ShardedClusterReconcileHelper) GetShardScaler(shardIdx int, memberClust
 	return scalers.NewMultiClusterReplicaSetScaler(fmt.Sprintf("shard idx %d", shardIdx), r.desiredShardsConfiguration[shardIdx].ClusterSpecList, memberCluster.Name, memberCluster.Index, r.shardsMemberClustersMap[shardIdx])
 }
 
-func (r *ShardedClusterReconcileHelper) getAllScalers() []scale.ReplicaSetScaler {
-	var result []scale.ReplicaSetScaler
+func (r *ShardedClusterReconcileHelper) getAllScalers() []interfaces.MultiClusterReplicaSetScaler {
+	var result []interfaces.MultiClusterReplicaSetScaler
 	for shardIdx := 0; shardIdx < r.sc.Spec.ShardCount; shardIdx++ {
 		for _, memberCluster := range r.shardsMemberClustersMap[shardIdx] {
-			result = append(result, r.GetShardScaler(shardIdx, memberCluster))
+			scaler := r.GetShardScaler(shardIdx, memberCluster)
+			result = append(result, scaler)
 		}
 	}
 
@@ -2673,7 +2738,7 @@ func (r *ShardedClusterReconcileHelper) isStillScaling() bool {
 // The difference vs isStillScaling is subtle. isStillScaling tells us if we're generally in the process of scaling (current sizes != spec).
 func (r *ShardedClusterReconcileHelper) shouldContinueScalingOneByOne() bool {
 	for _, s := range r.getAllScalers() {
-		if scale.ReplicasThisReconciliation(s) != s.(*scalers.MultiClusterReplicaSetScaler).TargetReplicas() {
+		if scale.ReplicasThisReconciliation(s) != s.TargetReplicas() {
 			return true
 		}
 	}