CLOUDP-71334: Fix MongoDB 4.4 Scaling Up (#186)

Author: chatton

pkg/apis/mongodb/v1/mongodb_types.go

@@ -5,6 +5,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/mongodb/mongodb-kubernetes-operator/pkg/util/scale"
+
 	appsv1 "k8s.io/api/apps/v1"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -218,6 +220,18 @@ type MongoDB struct {
 	Status MongoDBStatus `json:"status,omitempty"`
 }
 
+func (m MongoDB) DesiredReplicas() int {
+	return m.Spec.Members
+}
+
+func (m MongoDB) CurrentReplicas() int {
+	return m.Status.Members
+}
+
+func (m *MongoDB) ReplicasThisReconciliation() int {
+	return scale.ReplicasThisReconciliation(m)
+}
+
 // MongoURI returns a mongo uri which can be used to connect to this deployment
 func (m MongoDB) MongoURI() string {
 	members := make([]string, m.Spec.Members)

pkg/controller/mongodb/replica_set_controller.go

@@ -9,6 +9,8 @@ import (
 	"os"
 	"strings"
 
+	"github.com/mongodb/mongodb-kubernetes-operator/pkg/util/scale"
+
 	"github.com/mongodb/mongodb-kubernetes-operator/pkg/util/envvar"
 	"github.com/mongodb/mongodb-kubernetes-operator/pkg/util/status"
 
@@ -174,7 +176,10 @@ func (r *ReplicaSetReconciler) Reconcile(request reconcile.Request) (reconcile.R
 	}
 
 	r.log = zap.S().With("ReplicaSet", request.NamespacedName)
-	r.log.Infow("Reconciling MongoDB", "MongoDB.Spec", mdb.Spec, "MongoDB.Status", mdb.Status)
+	r.log.Infow("Reconciling MongoDB", "MongoDB.Spec", mdb.Spec, "MongoDB.Status", mdb.Status,
+		"desiredMembers", mdb.DesiredReplicas(),
+		"currentMembers", mdb.CurrentReplicas(),
+	)
 
 	if err := r.ensureAutomationConfig(mdb); err != nil {
 		return status.Update(r.client.Status(), &mdb,
@@ -287,9 +292,18 @@ func (r *ReplicaSetReconciler) Reconcile(request reconcile.Request) (reconcile.R
 		)
 	}
 
+	if scale.IsStillScaling(mdb) {
+		return status.Update(r.client.Status(), &mdb, statusOptions().
+			withMessage(Info, fmt.Sprintf("Performing scaling operation, currentMembers=%d, desiredMembers=%d",
+				mdb.CurrentReplicas(), mdb.DesiredReplicas())).
+			withMembers(mdb.ReplicasThisReconciliation()).
+			withPendingPhase(0),
+		)
+	}
+
 	res, err := status.Update(r.client.Status(), &mdb, statusOptions().
 		withMongoURI(mdb.MongoURI()).
-		withMembers(mdb.Spec.Members).
+		withMembers(mdb.ReplicasThisReconciliation()).
 		withMessage(None, "").
 		withRunningPhase(),
 	)

pkg/controller/mongodb/replicaset_controller_test.go

@@ -393,23 +393,106 @@ func TestExistingPasswordAndKeyfile_AreUsedWhenTheSecretExists(t *testing.T) {
 }
 
 func TestScramIsConfigured(t *testing.T) {
-	AssertReplicaSetIsConfiguredWithScram(t, newScramReplicaSet())
+	assertReplicaSetIsConfiguredWithScram(t, newScramReplicaSet())
 }
 
 func TestScramIsConfiguredWhenNotSpecified(t *testing.T) {
-	AssertReplicaSetIsConfiguredWithScram(t, newTestReplicaSet())
+	assertReplicaSetIsConfiguredWithScram(t, newTestReplicaSet())
 }
 
-func AssertReplicaSetIsConfiguredWithScram(t *testing.T, mdb mdbv1.MongoDB) {
+func TestReplicaSet_IsScaledDown_OneMember_AtATime_WhenItAlreadyExists(t *testing.T) {
+	mdb := newTestReplicaSet()
+	mdb.Spec.Members = 5
+
+	mgr := client.NewManager(&mdb)
+	r := newReconciler(mgr, mockManifestProvider(mdb.Spec.Version))
+	res, err := r.Reconcile(reconcile.Request{NamespacedName: types.NamespacedName{Namespace: mdb.Namespace, Name: mdb.Name}})
+	assertReconciliationSuccessful(t, res, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+
+	assert.NoError(t, err)
+	assert.Equal(t, 5, mdb.Status.Members)
+
+	// scale members from five to three
+	mdb.Spec.Members = 3
+
+	err = mgr.GetClient().Update(context.TODO(), &mdb)
+	assert.NoError(t, err)
+
+	makeStatefulSetReady(t, mgr.GetClient(), mdb)
+
+	res, err = r.Reconcile(reconcile.Request{NamespacedName: mdb.NamespacedName()})
+
+	assert.NoError(t, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+	assert.NoError(t, err)
+
+	assert.Equal(t, true, res.Requeue)
+	assert.Equal(t, 4, mdb.Status.Members)
+
+	res, err = r.Reconcile(reconcile.Request{NamespacedName: mdb.NamespacedName()})
+
+	assert.NoError(t, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+	assert.NoError(t, err)
+	assert.Equal(t, false, res.Requeue)
+	assert.Equal(t, 3, mdb.Status.Members)
+}
+
+func TestReplicaSet_IsScaledUp_OneMember_AtATime_WhenItAlreadyExists(t *testing.T) {
+	mdb := newTestReplicaSet()
+
+	mgr := client.NewManager(&mdb)
+	r := newReconciler(mgr, mockManifestProvider(mdb.Spec.Version))
+	res, err := r.Reconcile(reconcile.Request{NamespacedName: types.NamespacedName{Namespace: mdb.Namespace, Name: mdb.Name}})
+	assertReconciliationSuccessful(t, res, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+	assert.NoError(t, err)
+	assert.Equal(t, 3, mdb.Status.Members)
+
+	// scale members from three to five
+	mdb.Spec.Members = 5
+
+	err = mgr.GetClient().Update(context.TODO(), &mdb)
+	assert.NoError(t, err)
+
+	makeStatefulSetReady(t, mgr.GetClient(), mdb)
+
+	res, err = r.Reconcile(reconcile.Request{NamespacedName: mdb.NamespacedName()})
+
+	assert.NoError(t, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+
+	assert.NoError(t, err)
+	assert.Equal(t, true, res.Requeue)
+	assert.Equal(t, 4, mdb.Status.Members)
+
+	res, err = r.Reconcile(reconcile.Request{NamespacedName: mdb.NamespacedName()})
+
+	assert.NoError(t, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+	assert.NoError(t, err)
+
+	assert.Equal(t, false, res.Requeue)
+	assert.Equal(t, 5, mdb.Status.Members)
+}
+
+func assertReplicaSetIsConfiguredWithScram(t *testing.T, mdb mdbv1.MongoDB) {
 	mgr := client.NewManager(&mdb)
 	r := newReconciler(mgr, mockManifestProvider(mdb.Spec.Version))
 	res, err := r.Reconcile(reconcile.Request{NamespacedName: types.NamespacedName{Namespace: mdb.Namespace, Name: mdb.Name}})
 	assertReconciliationSuccessful(t, res, err)
 
 	currentAc, err := getCurrentAutomationConfig(client.NewClient(mgr.GetClient()), mdb)
 	t.Run("Automation Config is configured with SCRAM", func(t *testing.T) {
-		assert.NoError(t, err)
 		assert.NotEmpty(t, currentAc.Auth.Key)
+		assert.NoError(t, err)
 		assert.NotEmpty(t, currentAc.Auth.KeyFileWindows)
 		assert.NotEmpty(t, currentAc.Auth.AutoPwd)
 		assert.False(t, currentAc.Auth.Disabled)
@@ -423,6 +506,20 @@ func AssertReplicaSetIsConfiguredWithScram(t *testing.T, mdb mdbv1.MongoDB) {
 	})
 }
 
+func TestReplicaSet_IsScaledUpToDesiredMembers_WhenFirstCreated(t *testing.T) {
+	mdb := newTestReplicaSet()
+
+	mgr := client.NewManager(&mdb)
+	r := newReconciler(mgr, mockManifestProvider(mdb.Spec.Version))
+	res, err := r.Reconcile(reconcile.Request{NamespacedName: types.NamespacedName{Namespace: mdb.Namespace, Name: mdb.Name}})
+	assertReconciliationSuccessful(t, res, err)
+
+	err = mgr.GetClient().Get(context.TODO(), mdb.NamespacedName(), &mdb)
+	assert.NoError(t, err)
+
+	assert.Equal(t, 3, mdb.Status.Members)
+}
+
 func TestOpenshift_Configuration(t *testing.T) {
 	sts := performReconciliationAndGetStatefulSet(t, "openshift_mdb.yaml")
 	assert.Equal(t, "MANAGED_SECURITY_CONTEXT", sts.Spec.Template.Spec.Containers[0].Env[1].Name)

pkg/util/scale/scale.go

@@ -0,0 +1,43 @@
+package scale
+
+// ReplicaSetScaler is an interface which is able to scale up and down a replicaset
+// a single member at a time
+type ReplicaSetScaler interface {
+	DesiredReplicas() int
+	CurrentReplicas() int
+}
+
+// ReplicasThisReconciliation returns the number of replicas that should be configured
+// for that reconciliation. As of MongoDB 4.4 we can only scale members up / down 1 at a time.
+func ReplicasThisReconciliation(replicaSetScaler ReplicaSetScaler) int {
+	// the current replica set members will be 0 when we are creating a new deployment
+	// if this is the case, we want to jump straight to the desired members and not make changes incrementally
+	if replicaSetScaler.CurrentReplicas() == 0 || replicaSetScaler.CurrentReplicas() == replicaSetScaler.DesiredReplicas() {
+		return replicaSetScaler.DesiredReplicas()
+	}
+
+	if isScalingDown(replicaSetScaler) {
+		return replicaSetScaler.CurrentReplicas() - 1
+	}
+
+	return replicaSetScaler.CurrentReplicas() + 1
+
+}
+
+func IsStillScaling(replicaSetScaler ReplicaSetScaler) bool {
+	return ReplicasThisReconciliation(replicaSetScaler) != replicaSetScaler.DesiredReplicas()
+}
+
+func isScalingDown(replicaSetScaler ReplicaSetScaler) bool {
+	return replicaSetScaler.DesiredReplicas() < replicaSetScaler.CurrentReplicas()
+}
+
+// AnyAreStillScaling reports true if any of one the provided members is still scaling
+func AnyAreStillScaling(scalers ...ReplicaSetScaler) bool {
+	for _, s := range scalers {
+		if IsStillScaling(s) {
+			return true
+		}
+	}
+	return false
+}

test/e2e/replica_set_scale/replica_set_scaling_test.go

@@ -25,7 +25,6 @@ func TestReplicaSetScale(t *testing.T) {
 	}
 
 	mdb, user := e2eutil.NewTestMongoDB("mdb0")
-	mdb.Spec.Version = "4.0.6" // TOOD: Scaling will currently not work with MongoDB 4.4
 
 	_, err := setup.GeneratePasswordForUser(user, ctx)
 	if err != nil {