Save rs member ids in annotation/deployment state

Author: lucian-tosa

controllers/om/mockedomclient.go

@@ -141,6 +141,22 @@ func (oc *MockedOmConnection) ConfigureProject(project *Project) {
 	oc.context.OrgID = project.OrgID
 }
 
+func (oc *MockedOmConnection) GetReplicaSetMemberIds() (map[string]map[string]int, error) {
+	oc.addToHistory(reflect.ValueOf(oc.GetReplicaSetMemberIds))
+	dep, err := oc.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
+}
+
 var _ Connection = &MockedOmConnection{}
 
 // NewEmptyMockedOmConnection is the standard function for creating mocked connections that is usually used for testing

controllers/om/omclient.go

@@ -64,6 +64,10 @@ type Connection interface {
 	GetPreferredHostnames(agentApiKey string) ([]PreferredHostname, error)
 	AddPreferredHostname(agentApiKey string, value string, isRegexp bool) error
 
+	// GetReplicaSetMemberIds returns a map with the replicaset name as the key.
+	// The value is another map where the key is the replicaset member name and the value is its member id.
+	GetReplicaSetMemberIds() (map[string]map[string]int, error)
+
 	backup.GroupConfigReader
 	backup.GroupConfigUpdater
 
@@ -273,6 +277,21 @@ func (oc *HTTPOmConnection) GetAgentAuthMode() (string, error) {
 	return ac.Auth.AutoAuthMechanism, nil
 }
 
+func (oc *HTTPOmConnection) GetReplicaSetMemberIds() (map[string]map[string]int, error) {
+	dep, err := oc.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
+}
+
 var _ Connection = &HTTPOmConnection{}
 
 // NewOpsManagerConnection stores OpsManger api endpoint and authentication credentials.

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

@@ -198,8 +198,14 @@ func (r *ReconcileMongoDbMultiReplicaSet) Reconcile(ctx context.Context, request
 		return r.updateStatus(ctx, &mrs, status, log)
 	}
 
+	// Save replicasets member ids in annotation
+	finalMemberIds, err := conn.GetReplicaSetMemberIds()
+	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)
 	}
 
@@ -624,7 +630,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
@@ -654,6 +660,14 @@ func (r *ReconcileMongoDbMultiReplicaSet) saveLastAchievedSpec(ctx context.Conte
 		annotationsToAdd[mdbmultiv1.LastClusterNumMapping] = string(clusterNumBytes)
 	}
 
+	rsMemberIdsBytes, err := json.Marshal(rsMemberIds)
+	if err != nil {
+		return err
+	}
+	if string(rsMemberIdsBytes) != "null" {
+		annotationsToAdd[util.LastAchievedRsMemberIds] = string(rsMemberIdsBytes)
+	}
+
 	return annotations.SetAnnotations(ctx, &mrs, annotationsToAdd, r.client)
 }
 
@@ -696,6 +710,10 @@ 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
+	if len(processIds) == 0 {
+		processIds = getReplicaSetProcessIdsFromAnnotation(mrs)
+	}
 	log.Debugf("Existing process Ids: %+v", processIds)
 
 	certificateFileName := ""
@@ -791,6 +809,16 @@ func getReplicaSetProcessIdsFromReplicaSets(replicaSetName string, deployment om
 	return processIds
 }
 
+func getReplicaSetProcessIdsFromAnnotation(mrs mdbmultiv1.MongoDBMultiCluster) map[string]int {
+	processIds := make(map[string]map[string]int)
+	if processIdsStr, ok := mrs.Annotations[util.LastAchievedRsMemberIds]; ok {
+		if err := json.Unmarshal([]byte(processIdsStr), &processIds); err != nil {
+			return map[string]int{}
+		}
+	}
+	return processIds[mrs.Name]
+}
+
 func getSRVService(mrs *mdbmultiv1.MongoDBMultiCluster) corev1.Service {
 	additionalConfig := mrs.Spec.GetAdditionalMongodConfig()
 	port := additionalConfig.GetPortOrDefault()

controllers/operator/mongodbshardedcluster_controller.go

@@ -98,9 +98,10 @@ func newShardedClusterReconciler(ctx context.Context, kubeClient client.Client,
 }
 
 type ShardedClusterDeploymentState struct {
-	CommonDeploymentState `json:",inline"`
-	LastAchievedSpec      *mdbv1.MongoDbSpec   `json:"lastAchievedSpec"`
-	Status                *mdbv1.MongoDbStatus `json:"status"`
+	CommonDeploymentState   `json:",inline"`
+	LastAchievedRsMemberIds map[string]map[string]int `json:"lastAchievedRsMemberIds"`
+	LastAchievedSpec        *mdbv1.MongoDbSpec        `json:"lastAchievedSpec"`
+	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.
@@ -978,6 +979,13 @@ func (r *ShardedClusterReconcileHelper) Reconcile(ctx context.Context, log *zap.
 		return r.updateStatus(ctx, sc, workflow.Failed(err), log)
 	}
 
+	// Save replicasets member ids in deployment state
+	finalProcessIds, err := conn.GetReplicaSetMemberIds()
+	if err != nil {
+		return r.updateStatus(ctx, sc, workflow.Failed(err), log)
+	}
+	r.deploymentState.LastAchievedRsMemberIds = finalProcessIds
+
 	// Save last achieved spec in state
 	r.deploymentState.LastAchievedSpec = &sc.Spec
 	log.Infof("Finished reconciliation for Sharded Cluster! %s", completionMessage(conn.BaseURL(), conn.GroupID()))
@@ -1907,7 +1915,7 @@ func (r *ShardedClusterReconcileHelper) publishDeployment(ctx context.Context, c
 	}
 
 	configSrvProcesses, configSrvMemberOptions := r.createDesiredConfigSrvProcessesAndMemberOptions(configSrvMemberCertPath)
-	configRs, _ := buildReplicaSetFromProcesses(sc.ConfigRsName(), configSrvProcesses, sc, configSrvMemberOptions, existingDeployment)
+	configRs, _ := buildReplicaSetFromProcesses(sc.ConfigRsName(), configSrvProcesses, sc, configSrvMemberOptions, existingDeployment, r.deploymentState.LastAchievedRsMemberIds[sc.ConfigRsName()])
 
 	// Shards
 	shards := make([]om.ReplicaSetWithProcesses, sc.Spec.ShardCount)
@@ -1918,7 +1926,7 @@ func (r *ShardedClusterReconcileHelper) publishDeployment(ctx context.Context, c
 		shardInternalClusterPaths = append(shardInternalClusterPaths, fmt.Sprintf("%s/%s", util.InternalClusterAuthMountPath, shardOptions.InternalClusterHash))
 		shardMemberCertPath := fmt.Sprintf("%s/%s", util.TLSCertMountPath, shardOptions.CertificateHash)
 		desiredShardProcesses, desiredShardMemberOptions := r.createDesiredShardProcessesAndMemberOptions(shardIdx, shardMemberCertPath)
-		shards[shardIdx], _ = buildReplicaSetFromProcesses(r.sc.ShardRsName(shardIdx), desiredShardProcesses, sc, desiredShardMemberOptions, existingDeployment)
+		shards[shardIdx], _ = buildReplicaSetFromProcesses(r.sc.ShardRsName(shardIdx), desiredShardProcesses, sc, desiredShardMemberOptions, existingDeployment, r.deploymentState.LastAchievedRsMemberIds[r.sc.ShardRsName(shardIdx)])
 	}
 
 	// updateOmAuthentication normally takes care of the certfile rotation code, but since sharded-cluster is special pertaining multiple clusterfiles, we code this part here for now.
@@ -2225,10 +2233,15 @@ func createMongodProcessForShardedCluster(mongoDBImage string, forceEnterprise b
 
 // buildReplicaSetFromProcesses creates the 'ReplicaSetWithProcesses' with specified processes. This is of use only
 // for sharded cluster (config server, shards)
-func buildReplicaSetFromProcesses(name string, members []om.Process, mdb *mdbv1.MongoDB, memberOptions []automationconfig.MemberOptions, deployment om.Deployment) (om.ReplicaSetWithProcesses, error) {
+func buildReplicaSetFromProcesses(name string, members []om.Process, mdb *mdbv1.MongoDB, memberOptions []automationconfig.MemberOptions, deployment om.Deployment, savedProcessIds map[string]int) (om.ReplicaSetWithProcesses, error) {
 	replicaSet := om.NewReplicaSet(name, mdb.Spec.GetMongoDBVersion())
 
 	existingProcessIds := getReplicaSetProcessIdsFromReplicaSets(replicaSet.Name(), deployment)
+	// If there is no replicaset configuration saved in OM, it might be a new project, so we check the ids saved in deployment state
+	if len(existingProcessIds) == 0 {
+		existingProcessIds = savedProcessIds
+	}
+
 	var rsWithProcesses om.ReplicaSetWithProcesses
 	if mdb.Spec.IsMultiCluster() {
 		// we're passing nil as connectivity argument as in sharded clusters horizons don't make much sense as we don't expose externally individual shards

controllers/operator/mongodbshardedcluster_controller_multi_test.go

@@ -1088,7 +1088,10 @@ func TestMigrateToNewDeploymentState(t *testing.T) {
 	err = kubeClient.Get(ctx, types.NamespacedName{Name: configMapName, Namespace: sc.Namespace}, stateConfigMap)
 	require.NoError(t, err)
 
-	expectedDeploymentState := generateExpectedDeploymentState(t, sc)
+	memberIds, err := omConnectionFactory.GetConnection().GetReplicaSetMemberIds()
+	require.NoError(t, err)
+
+	expectedDeploymentState := generateExpectedDeploymentState(t, sc, memberIds)
 	require.Contains(t, stateConfigMap.Data, stateKey)
 	require.JSONEq(t, expectedDeploymentState, stateConfigMap.Data[stateKey])
 
@@ -3587,18 +3590,19 @@ func getMultiClusterFQDN(stsName string, namespace string, clusterIdx int, podId
 	return fmt.Sprintf("%s-svc.%s.svc.%s", getPodName(stsName, clusterIdx, podIdx), namespace, clusterDomain)
 }
 
-func generateExpectedDeploymentState(t *testing.T, sc *mdbv1.MongoDB) string {
+func generateExpectedDeploymentState(t *testing.T, sc *mdbv1.MongoDB, memberIds map[string]map[string]int) string {
 	lastSpec, _ := sc.GetLastSpec()
 	expectedState := ShardedClusterDeploymentState{
 		CommonDeploymentState: CommonDeploymentState{
 			ClusterMapping: map[string]int{},
 		},
-		LastAchievedSpec: lastSpec,
-		Status:           &sc.Status,
+		LastAchievedRsMemberIds: memberIds,
+		LastAchievedSpec:        lastSpec,
+		Status:                  &sc.Status,
 	}
-	lastSpecBytes, err := json.Marshal(expectedState)
+	expectedStateBytes, err := json.Marshal(expectedState)
 	require.NoError(t, err)
-	return string(lastSpecBytes)
+	return string(expectedStateBytes)
 }
 
 func loadMongoDBResource(resourceYamlPath string) (*mdbv1.MongoDB, error) {

controllers/operator/mongodbshardedcluster_controller_test.go

@@ -1644,7 +1644,7 @@ func createDeploymentFromShardedCluster(t *testing.T, updatable v1.CustomResourc
 			construct.GetPodEnvOptions(),
 		)
 		shardSts := construct.DatabaseStatefulSet(*sh, shardOptions, zap.S())
-		shards[i], _ = buildReplicaSetFromProcesses(shardSts.Name, createShardProcesses("fake-mongoDBImage", false, shardSts, sh, ""), sh, sh.Spec.GetMemberOptions(), om.NewDeployment())
+		shards[i], _ = buildReplicaSetFromProcesses(shardSts.Name, createShardProcesses("fake-mongoDBImage", false, shardSts, sh, ""), sh, sh.Spec.GetMemberOptions(), om.NewDeployment(), make(map[string]int))
 	}
 
 	desiredMongosConfig := createMongosSpec(sh)
@@ -1665,7 +1665,7 @@ func createDeploymentFromShardedCluster(t *testing.T, updatable v1.CustomResourc
 		construct.GetPodEnvOptions(),
 	)
 	configSvrSts := construct.DatabaseStatefulSet(*sh, configServerOptions, zap.S())
-	configRs, _ := buildReplicaSetFromProcesses(configSvrSts.Name, createConfigSrvProcesses("fake-mongoDBImage", false, configSvrSts, sh, ""), sh, sh.Spec.GetMemberOptions(), om.NewDeployment())
+	configRs, _ := buildReplicaSetFromProcesses(configSvrSts.Name, createConfigSrvProcesses("fake-mongoDBImage", false, configSvrSts, sh, ""), sh, sh.Spec.GetMemberOptions(), om.NewDeployment(), make(map[string]int))
 
 	d := om.NewDeployment()
 	_, err := d.MergeShardedCluster(om.DeploymentShardedClusterMergeOptions{

pkg/util/constants.go

@@ -280,7 +280,10 @@ const (
 	// TODO: remove this from here and move it to the certs package
 	// This currently creates an import cycle
 	InternalCertAnnotationKey = "internalCertHash"
-	LastAchievedSpec          = "mongodb.com/v1.lastSuccessfulConfiguration"
+
+	// Annotation keys used by the operator
+	LastAchievedSpec        = "mongodb.com/v1.lastSuccessfulConfiguration"
+	LastAchievedRsMemberIds = "mongodb.com/v1.lastAchievedRsMemberIds"
 
 	// SecretVolumeName is the name of the volume resource.
 	SecretVolumeName = "secret-certs"