CLOUDP-324440 Save rs member ids (#206)

# Summary

This PR fixes the behaviour of changing the project in a
MongoDBMultiCluster resource. Multicluster resources have a different
method of computing the member ids of each process in a replicaset
(`replicasets` field in AC). Because of that, it is possible to get
non-sequential ids by doing scaling operations. For example, we start
from this
* mongodb-0-0: id 0
* mongodb-0-1: id 1
* mongodb-1-0: id 2

If we now scale the first cluster up we end up with:
* mongodb-0-0: id 0
* mongodb-0-1: id 1
* mongodb-1-0: id 2
* mongodb-0-2: id 3

If we now change the project, the operator will find an empty AC and
will recalculate the ids it of the replicaset members.
* mongodb-0-0: id 0
* mongodb-0-1: id 1
* mongodb-0-2: id 2
* mongodb-1-0: id 3
 
But now, member `0-2` has id 2 instead of 3, and member `1-0` has id 3
instead of 2. This will instruct the agents to change the members of the
replicaset, but this change will be rejected, leaving the deployment
stuck. The operator can't retrieve the previous member ids.

In this PR we will now save the member ids we achieved in an annotation,
and whenever there is an empty AC in OM, we try to read the annotation
in case the deployment is being migrated to a new project.

### Other considerations

* The id of a replica set member has to be an integer, so we can't use
the hostnames instead.
* We can bring a future improvement to the id computation method by
using a hash between cluster index and pod index. This way, any id the
operator computes will be consistent and do not require reading the AC.
The issue is that changing the id of a replicaset member is a convoluted
process and migrating the old pseudo-sequential ids to hashed ids is
extremely complex. This means that while there still are deployments
that use these kind of id's we need to back these id's somewhere in case
the deployment is moved to a different project. This idea will be
discussed further.
* If the `MongoDBMultiCluster` is backed up with annotations and then
re-applied to a new project, the annotation will be used to compute the
member ids. This is not an issue, there is no restrictions on these ids
(apart from being an integer and unique). After the initial deployment,
any scaling operations will simply compute new sequential ids.

## Proof of Work

Added unit tests to check that annotations are created and updated
successfully, as well as using the annotation when building the
replicaset.
Also added an E2E test to ensure the resource will become ready.

## Checklist
- [x] Have you linked a jira ticket and/or is the ticket in the title?
- [x] Have you checked whether your jira ticket required DOCSP changes?
- [x] Have you checked for release_note changes?

Author: lucian-tosa

RELEASE_NOTES.md

@@ -25,6 +25,9 @@
     * [Manage Database Users using OIDC](https://www.mongodb.com/docs/kubernetes/upcoming/manage-users/) # TODO
     * [Authentication and Authorization with OIDC/OAuth 2.0](https://www.mongodb.com/docs/manual/core/oidc/security-oidc/)
 
+## Bug Fixes
+* Fixed an issue where moving a **MongoDBMultiCluster** resource to a new project (or a new OM instance) would leave the deployment in a failed state.
+
 <!-- Past Releases -->
 
 # MCK 1.1.0 Release Notes

controllers/om/automation_config_test.go

@@ -1059,7 +1059,7 @@ func TestApplyInto(t *testing.T) {
 }
 
 func changeTypes(deployment Deployment) error {
-	rs := deployment.getReplicaSets()
+	rs := deployment.GetReplicaSets()
 	deployment.setReplicaSets(rs)
 	return nil
 }

controllers/om/deployment.go

@@ -280,13 +280,13 @@ func (d Deployment) AddMonitoringAndBackup(log *zap.SugaredLogger, tls bool, caF
 	d.addBackup(log)
 }
 
-// DEPRECATED: this shouldn't be used as it may panic because of different underlying type; use getReplicaSets instead
+// DEPRECATED: this shouldn't be used as it may panic because of different underlying type; use GetReplicaSets instead
 func (d Deployment) ReplicaSets() []ReplicaSet {
 	return d["replicaSets"].([]ReplicaSet)
 }
 
 func (d Deployment) GetReplicaSetByName(name string) ReplicaSet {
-	for _, rs := range d.getReplicaSets() {
+	for _, rs := range d.GetReplicaSets() {
 		if rs.Name() == name {
 			return rs
 		}
@@ -395,7 +395,7 @@ func (d Deployment) RemoveReplicaSetByName(name string, log *zap.SugaredLogger)
 		return xerrors.New("ReplicaSet does not exist")
 	}
 
-	currentRs := d.getReplicaSets()
+	currentRs := d.GetReplicaSets()
 	toKeep := make([]ReplicaSet, len(currentRs)-1)
 	i := 0
 	for _, el := range currentRs {
@@ -685,7 +685,7 @@ func (d Deployment) ProcessesCopy() []Process {
 
 // ReplicaSetsCopy returns the COPY of replicasets in the deployment.
 func (d Deployment) ReplicaSetsCopy() []ReplicaSet {
-	return d.deepCopy().getReplicaSets()
+	return d.deepCopy().GetReplicaSets()
 }
 
 // ShardedClustersCopy returns the COPY of sharded clusters in the deployment.
@@ -958,7 +958,7 @@ func (d Deployment) getProcessByName(name string) *Process {
 }
 
 func (d Deployment) getReplicaSetByName(name string) *ReplicaSet {
-	for _, r := range d.getReplicaSets() {
+	for _, r := range d.GetReplicaSets() {
 		if r.Name() == name {
 			return &r
 		}
@@ -977,7 +977,7 @@ func (d Deployment) getShardedClusterByName(name string) *ShardedCluster {
 	return nil
 }
 
-func (d Deployment) getReplicaSets() []ReplicaSet {
+func (d Deployment) GetReplicaSets() []ReplicaSet {
 	switch v := d["replicaSets"].(type) {
 	case []ReplicaSet:
 		return v
@@ -997,7 +997,7 @@ func (d Deployment) setReplicaSets(replicaSets []ReplicaSet) {
 }
 
 func (d Deployment) addReplicaSet(rs ReplicaSet) {
-	d.setReplicaSets(append(d.getReplicaSets(), rs))
+	d.setReplicaSets(append(d.GetReplicaSets(), rs))
 }
 
 func (d Deployment) getShardedClusters() []ShardedCluster {
@@ -1052,7 +1052,7 @@ func (d Deployment) findReplicaSetsRemovedFromShardedCluster(clusterName string)
 	clusterReplicaSets := shardedCluster.getAllReplicaSets()
 	var ans []string
 
-	for _, v := range d.getReplicaSets() {
+	for _, v := range d.GetReplicaSets() {
 		if !stringutil.Contains(clusterReplicaSets, v.Name()) && isShardOfShardedCluster(clusterName, v.Name()) {
 			ans = append(ans, v.Name())
 		}

controllers/om/deployment_test.go

@@ -56,25 +56,25 @@ func TestMergeReplicaSet(t *testing.T) {
 	expectedRs := buildRsByProcesses("fooRs", createReplicaSetProcesses("fooRs"))
 
 	assert.Len(t, d.getProcesses(), 3)
-	assert.Len(t, d.getReplicaSets(), 1)
-	assert.Len(t, d.getReplicaSets()[0].Members(), 3)
-	assert.Equal(t, d.getReplicaSets()[0], expectedRs.Rs)
+	assert.Len(t, d.GetReplicaSets(), 1)
+	assert.Len(t, d.GetReplicaSets()[0].Members(), 3)
+	assert.Equal(t, d.GetReplicaSets()[0], expectedRs.Rs)
 
 	// Now the deployment "gets updated" from external - new node is added and one is removed - this should be fixed
 	// by merge
 	newProcess := NewMongodProcess("foo", "bar", "fake-mongoDBImage", false, &mdbv1.AdditionalMongodConfig{}, &mdbv1.NewStandaloneBuilder().Build().Spec, "", nil, "")
 
 	d.getProcesses()[0]["processType"] = ProcessTypeMongos                            // this will be overriden
 	d.getProcesses()[1].EnsureNetConfig()["MaxIncomingConnections"] = 20              // this will be left as-is
-	d.getReplicaSets()[0]["protocolVersion"] = 10                                     // this field will be overriden by Operator
-	d.getReplicaSets()[0].setMembers(d.getReplicaSets()[0].Members()[0:2])            // "removing" the last node in replicaset
-	d.getReplicaSets()[0].addMember(newProcess, "", automationconfig.MemberOptions{}) // "adding" some new node
-	d.getReplicaSets()[0].Members()[0]["arbiterOnly"] = true                          // changing data for first node
+	d.GetReplicaSets()[0]["protocolVersion"] = 10                                     // this field will be overriden by Operator
+	d.GetReplicaSets()[0].setMembers(d.GetReplicaSets()[0].Members()[0:2])            // "removing" the last node in replicaset
+	d.GetReplicaSets()[0].addMember(newProcess, "", automationconfig.MemberOptions{}) // "adding" some new node
+	d.GetReplicaSets()[0].Members()[0]["arbiterOnly"] = true                          // changing data for first node
 
 	mergeReplicaSet(d, "fooRs", createReplicaSetProcesses("fooRs"))
 
 	assert.Len(t, d.getProcesses(), 3)
-	assert.Len(t, d.getReplicaSets(), 1)
+	assert.Len(t, d.GetReplicaSets(), 1)
 
 	expectedRs = buildRsByProcesses("fooRs", createReplicaSetProcesses("fooRs"))
 	expectedRs.Rs.Members()[0]["arbiterOnly"] = true
@@ -89,14 +89,14 @@ func TestMergeReplica_ScaleDown(t *testing.T) {
 
 	mergeReplicaSet(d, "someRs", createReplicaSetProcesses("someRs"))
 	assert.Len(t, d.getProcesses(), 3)
-	assert.Len(t, d.getReplicaSets()[0].Members(), 3)
+	assert.Len(t, d.GetReplicaSets()[0].Members(), 3)
 
 	// "scale down"
 	scaledDownRsProcesses := createReplicaSetProcesses("someRs")[0:2]
 	mergeReplicaSet(d, "someRs", scaledDownRsProcesses)
 
 	assert.Len(t, d.getProcesses(), 2)
-	assert.Len(t, d.getReplicaSets()[0].Members(), 2)
+	assert.Len(t, d.GetReplicaSets()[0].Members(), 2)
 
 	// checking that the last member was removed
 	rsProcesses := buildRsByProcesses("someRs", createReplicaSetProcesses("someRs")).Processes
@@ -123,7 +123,7 @@ func TestMergeReplicaSet_MergeFirstProcess(t *testing.T) {
 	mergeReplicaSet(d, "fooRs", createReplicaSetProcessesCount(5, "fooRs"))
 
 	assert.Len(t, d.getProcesses(), 8)
-	assert.Len(t, d.getReplicaSets(), 2)
+	assert.Len(t, d.GetReplicaSets(), 2)
 
 	expectedRs := buildRsByProcesses("fooRs", createReplicaSetProcessesCount(5, "fooRs"))
 
@@ -177,8 +177,8 @@ func TestMergeDeployment_BigReplicaset(t *testing.T) {
 	checkNumberOfVotingMembers(t, rs, 7, 8)
 
 	// Now OM user "has changed" votes for some of the members - this must stay the same after merge
-	omDeployment.getReplicaSets()[0].Members()[2].setVotes(0).setPriority(0)
-	omDeployment.getReplicaSets()[0].Members()[4].setVotes(0).setPriority(0)
+	omDeployment.GetReplicaSets()[0].Members()[2].setVotes(0).setPriority(0)
+	omDeployment.GetReplicaSets()[0].Members()[4].setVotes(0).setPriority(0)
 
 	omDeployment.MergeReplicaSet(rs, nil, nil, zap.S())
 	checkNumberOfVotingMembers(t, rs, 5, 8)
@@ -199,10 +199,10 @@ func TestMergeDeployment_BigReplicaset(t *testing.T) {
 
 	omDeployment.MergeReplicaSet(rsToMerge, nil, nil, zap.S())
 	checkNumberOfVotingMembers(t, rs, 7, 11)
-	assert.Equal(t, 0, omDeployment.getReplicaSets()[0].Members()[2].Votes())
-	assert.Equal(t, 0, omDeployment.getReplicaSets()[0].Members()[4].Votes())
-	assert.Equal(t, float32(0), omDeployment.getReplicaSets()[0].Members()[2].Priority())
-	assert.Equal(t, float32(0), omDeployment.getReplicaSets()[0].Members()[4].Priority())
+	assert.Equal(t, 0, omDeployment.GetReplicaSets()[0].Members()[2].Votes())
+	assert.Equal(t, 0, omDeployment.GetReplicaSets()[0].Members()[4].Votes())
+	assert.Equal(t, float32(0), omDeployment.GetReplicaSets()[0].Members()[2].Priority())
+	assert.Equal(t, float32(0), omDeployment.GetReplicaSets()[0].Members()[4].Priority())
 }
 
 func TestGetAllProcessNames_MergedReplicaSetsAndShardedClusters(t *testing.T) {
@@ -360,7 +360,7 @@ func TestGetNumberOfExcessProcesses_ShardedClusterScaleDown(t *testing.T) {
 	_, err := d.MergeShardedCluster(mergeOpts)
 	assert.NoError(t, err)
 	assert.Len(t, d.getShardedClusterByName("sc001").shards(), 3)
-	assert.Len(t, d.getReplicaSets(), 4)
+	assert.Len(t, d.GetReplicaSets(), 4)
 	assert.Equal(t, 0, d.GetNumberOfExcessProcesses("sc001"))
 
 	// Now we are "scaling down" the sharded cluster - so junk replica sets will appear - this is still ok
@@ -377,7 +377,7 @@ func TestGetNumberOfExcessProcesses_ShardedClusterScaleDown(t *testing.T) {
 	_, err = d.MergeShardedCluster(mergeOpts)
 	assert.NoError(t, err)
 	assert.Len(t, d.getShardedClusterByName("sc001").shards(), 2)
-	assert.Len(t, d.getReplicaSets(), 4)
+	assert.Len(t, d.GetReplicaSets(), 4)
 
 	assert.Equal(t, 0, d.GetNumberOfExcessProcesses("sc001"))
 }
@@ -586,7 +586,7 @@ func checkShardedClusterCheckExtraReplicaSets(t *testing.T, d Deployment, expect
 		// checking that no previous replica sets are left. For this we take the name of first shard and remove the last digit
 		firstShardName := expectedReplicaSets[0].Rs.Name()
 		i := 0
-		for _, r := range d.getReplicaSets() {
+		for _, r := range d.GetReplicaSets() {
 			if strings.HasPrefix(r.Name(), firstShardName[0:len(firstShardName)-1]) {
 				i++
 			}

controllers/om/depshardedcluster_test.go

@@ -31,9 +31,9 @@ func TestMergeShardedCluster_New(t *testing.T) {
 	assert.NoError(t, err)
 
 	require.Len(t, d.getProcesses(), 15)
-	require.Len(t, d.getReplicaSets(), 4)
+	require.Len(t, d.GetReplicaSets(), 4)
 	for i := 0; i < 4; i++ {
-		require.Len(t, d.getReplicaSets()[i].Members(), 3)
+		require.Len(t, d.GetReplicaSets()[i].Members(), 3)
 	}
 	checkMongoSProcesses(t, d.getProcesses(), createMongosProcesses(3, "pretty", "cluster"))
 	checkReplicaSet(t, d, createConfigSrvRs("configSrv", true))
@@ -130,9 +130,9 @@ func TestMergeShardedCluster_ReplicaSetsModified(t *testing.T) {
 	expectedShards[0].Rs["writeConcernMajorityJournalDefault"] = true
 
 	require.Len(t, d.getProcesses(), 15)
-	require.Len(t, d.getReplicaSets(), 4)
+	require.Len(t, d.GetReplicaSets(), 4)
 	for i := 0; i < 4; i++ {
-		require.Len(t, d.getReplicaSets()[i].Members(), 3)
+		require.Len(t, d.GetReplicaSets()[i].Members(), 3)
 	}
 	checkMongoSProcesses(t, d.getProcesses(), createMongosProcesses(3, "pretty", "cluster"))
 	checkReplicaSet(t, d, createConfigSrvRs("configSrv", true))
@@ -166,7 +166,7 @@ func TestMergeShardedCluster_ShardedClusterModified(t *testing.T) {
 
 	mergeReplicaSet(d, "fakeShard", createReplicaSetProcesses("fakeShard"))
 
-	require.Len(t, d.getReplicaSets(), 5)
+	require.Len(t, d.GetReplicaSets(), 5)
 
 	// Final check - we create the expected configuration, add there correct OM changes and check for equality with merge
 	// result
@@ -188,9 +188,9 @@ func TestMergeShardedCluster_ShardedClusterModified(t *testing.T) {
 	// Note, that fake replicaset and it's processes haven't disappeared as we passed 'false' to 'MergeShardedCluster'
 	// which results in "draining" for redundant shards but not physical removal of replica sets
 	require.Len(t, d.getProcesses(), 18)
-	require.Len(t, d.getReplicaSets(), 5)
+	require.Len(t, d.GetReplicaSets(), 5)
 	for i := 0; i < 4; i++ {
-		require.Len(t, d.getReplicaSets()[i].Members(), 3)
+		require.Len(t, d.GetReplicaSets()[i].Members(), 3)
 	}
 	checkMongoSProcesses(t, d.getProcesses(), createMongosProcesses(3, "pretty", "cluster"))
 	checkReplicaSet(t, d, createConfigSrvRs("configSrv", true))

controllers/om/mockedomclient.go

@@ -734,7 +734,7 @@ func (oc *MockedOmConnection) CheckResourcesAndBackupDeleted(t *testing.T, resou
 	// This can be improved for some more complicated scenarios when we have different resources in parallel - so far
 	// just checking if deployment
 	assert.Empty(t, oc.deployment.getProcesses())
-	assert.Empty(t, oc.deployment.getReplicaSets())
+	assert.Empty(t, oc.deployment.GetReplicaSets())
 	assert.Empty(t, oc.deployment.getShardedClusters())
 	assert.Empty(t, oc.deployment.getMonitoringVersions())
 	assert.Empty(t, oc.deployment.getBackupVersions())

controllers/om/omclient.go

@@ -978,6 +978,21 @@ func (oc *HTTPOmConnection) AddPreferredHostname(agentApiKey string, value strin
 	return nil
 }
 
+func GetReplicaSetMemberIds(conn Connection) (map[string]map[string]int, error) {
+	dep, err := conn.ReadDeployment()
+	if err != nil {
+		return nil, err
+	}
+
+	finalProcessIds := make(map[string]map[string]int)
+
+	for _, replicaSet := range dep.GetReplicaSets() {
+		finalProcessIds[replicaSet.Name()] = replicaSet.MemberIds()
+	}
+
+	return finalProcessIds, nil
+}
+
 //********************************** Private methods *******************************************************************
 
 func (oc *HTTPOmConnection) get(path string) ([]byte, error) {

controllers/om/replicaset.go

@@ -146,6 +146,14 @@ func (r ReplicaSet) String() string {
 	return fmt.Sprintf("\"%s\" (members: %v)", r.Name(), r.Members())
 }
 
+func (r ReplicaSet) MemberIds() map[string]int {
+	memberIds := make(map[string]int)
+	for _, rsMember := range r.Members() {
+		memberIds[rsMember.Name()] = rsMember.Id()
+	}
+	return memberIds
+}
+
 // ***************************************** Private methods ***********************************************************
 
 func initDefaultRs(set ReplicaSet, name string, protocolVersion string) {

controllers/operator/mongodbmultireplicaset_controller.go

@@ -201,8 +201,13 @@ func (r *ReconcileMongoDbMultiReplicaSet) Reconcile(ctx context.Context, request
 		return r.updateStatus(ctx, &mrs, status, log)
 	}
 
+	finalMemberIds, err := om.GetReplicaSetMemberIds(conn)
+	if err != nil {
+		return r.updateStatus(ctx, &mrs, workflow.Failed(err), log)
+	}
+
 	mrs.Status.FeatureCompatibilityVersion = mrs.CalculateFeatureCompatibilityVersion()
-	if err := r.saveLastAchievedSpec(ctx, mrs); err != nil {
+	if err := r.saveLastAchievedSpec(ctx, mrs, finalMemberIds); err != nil {
 		return r.updateStatus(ctx, &mrs, workflow.Failed(xerrors.Errorf("Failed to set annotation: %w", err)), log)
 	}
 
@@ -627,7 +632,7 @@ func getMembersForClusterSpecItemThisReconciliation(mrs *mdbmultiv1.MongoDBMulti
 }
 
 // saveLastAchievedSpec updates the MongoDBMultiCluster resource with the spec that was just achieved.
-func (r *ReconcileMongoDbMultiReplicaSet) saveLastAchievedSpec(ctx context.Context, mrs mdbmultiv1.MongoDBMultiCluster) error {
+func (r *ReconcileMongoDbMultiReplicaSet) saveLastAchievedSpec(ctx context.Context, mrs mdbmultiv1.MongoDBMultiCluster, rsMemberIds map[string]map[string]int) error {
 	clusterSpecs, err := mrs.GetClusterSpecItems()
 	if err != nil {
 		return err
@@ -657,6 +662,16 @@ func (r *ReconcileMongoDbMultiReplicaSet) saveLastAchievedSpec(ctx context.Conte
 		annotationsToAdd[mdbmultiv1.LastClusterNumMapping] = string(clusterNumBytes)
 	}
 
+	if len(rsMemberIds) > 0 {
+		rsMemberIdsBytes, err := json.Marshal(rsMemberIds)
+		if err != nil {
+			return err
+		}
+		if len(rsMemberIdsBytes) > 0 {
+			annotationsToAdd[util.LastAchievedRsMemberIds] = string(rsMemberIdsBytes)
+		}
+	}
+
 	return annotations.SetAnnotations(ctx, &mrs, annotationsToAdd, r.client)
 }
 
@@ -699,6 +714,15 @@ func (r *ReconcileMongoDbMultiReplicaSet) updateOmDeploymentRs(ctx context.Conte
 	}
 
 	processIds := getReplicaSetProcessIdsFromReplicaSets(mrs.Name, existingDeployment)
+
+	// If there is no replicaset configuration saved in OM, it might be a new project, so we check the ids saved in annotation
+	// A project migration can happen if .spec.opsManager.configMapRef is changed, or the original configMap has been modified.
+	if len(processIds) == 0 {
+		processIds, err = getReplicaSetProcessIdsFromAnnotation(mrs)
+		if err != nil {
+			return xerrors.Errorf("failed to get member ids from annotation: %w", err)
+		}
+	}
 	log.Debugf("Existing process Ids: %+v", processIds)
 
 	certificateFileName := ""
@@ -794,6 +818,17 @@ func getReplicaSetProcessIdsFromReplicaSets(replicaSetName string, deployment om
 	return processIds
 }
 
+func getReplicaSetProcessIdsFromAnnotation(mrs mdbmultiv1.MongoDBMultiCluster) (map[string]int, error) {
+	if processIdsStr, ok := mrs.Annotations[util.LastAchievedRsMemberIds]; ok {
+		processIds := make(map[string]map[string]int)
+		if err := json.Unmarshal([]byte(processIdsStr), &processIds); err != nil {
+			return map[string]int{}, err
+		}
+		return processIds[mrs.Name], nil
+	}
+	return make(map[string]int), nil
+}
+
 func getSRVService(mrs *mdbmultiv1.MongoDBMultiCluster) corev1.Service {
 	additionalConfig := mrs.Spec.GetAdditionalMongodConfig()
 	port := additionalConfig.GetPortOrDefault()