CLOUDP-347497: Single cluster Replica Set Controller Refactoring (#486)

# CLOUDP-347497 - Single cluster Replica Set Controller Refactoring

## Why this refactoring

The single-cluster RS controller was mixing two concerns:
- **Kubernetes stuff** (StatefulSets, pods, volumes)
- **Ops Manager/MongoDB stuff** (MongoDB processes, replication config)

This worked fine for single-cluster, but it's a problem when you think
about multi-cluster:
- Multi-cluster has **multiple StatefulSets** (one per cluster) but only
**one logical ReplicaSet** in Ops Manager
- The OM automation config doesn't care about how many K8s clusters you
have or how the pods are deployed

So we need to separate these layers properly.

## Main changes

### 1. Broke down the huge Reconcile() method

Before: ~300 lines of inline logic in Reconcile()

Now:
```go
Reconcile()
  ├── reconcileMemberResources()        // Handles all K8s resource creation
  │   ├── reconcileHostnameOverrideConfigMap()
  │   ├── ensureRoles()
  │   └── reconcileStatefulSet()        // StatefulSet-specific logic isolated here
  │       └── buildStatefulSetOptions() // Builds STS configuration
  └── updateOmDeploymentRs()            // Handles Ops Manager automation config updates
```

This makes it way easier to understand what's happening and matches the
multi-cluster controller structure.

### 2. Removed StatefulSet dependency from OM operations

Created new helper functions that work directly with MongoDB resources
instead of StatefulSets:
- `CreateMongodProcessesFromMongoDB()` - was using StatefulSet before
- `BuildFromMongoDBWithReplicas()` - same
- `WaitForRsAgentsToRegisterByResource()` - same

These mirror the existing `...FromStatefulSet` functions but take
MongoDB resources instead.

**Why it matters:** The OM layer now only cares about the MongoDB
resource definition, not how it's deployed in K8s. This makes the code
work the same way for both single-cluster and multi-cluster.

### 3. Added publishAutomationConfigFirstRS checks

Dedicated function for RS instead of using the shared one. Does not rely
on a statefulset object.
## Important for review

The ideal way to review this PR is to compare the new structure to the
`mongodbmultireplicaset_controller.go`. The aim of the refactoring is to
get the single cluster controller closer to it.

Look at:
- `reconcileMemberResources()` in both controllers - similar structure
now
- `updateOmDeploymentRs()` - no more StatefulSet dependency
- New helper functions in `om/process` and `om/replicaset` - mirror
existing patterns

## Bug found along the way

The logic to handle **scale up + disable TLS at the same time** doesn't
actually work properly and should be blocked by validation (see [draft
PR #490](#490) for
more details).

## Tests added

Added tests for the new helper functions:

- `TestCreateMongodProcessesFromMongoDB` - basic scenarios, scaling,
custom domains, TLS, additional config
- `TestBuildFromMongoDBWithReplicas` - integration test checking
ReplicaSet structure and member options propagation
- `TestPublishAutomationConfigFirstRS` - automation config publish logic
with various TLS/auth scenarios

Author: Julien-Ben

api/v1/mdb/mongodb_types.go

@@ -1038,7 +1038,7 @@ func (a *Authentication) IsOIDCEnabled() bool {
 	return stringutil.Contains(a.GetModes(), util.OIDC)
 }
 
-// GetModes returns the modes of the Authentication instance of an empty
+// GetModes returns the modes of the Authentication instance, or an empty
 // list if it is nil
 func (a *Authentication) GetModes() []string {
 	if a == nil {

controllers/om/process/om_process.go

@@ -22,6 +22,18 @@ func CreateMongodProcessesWithLimit(mongoDBImage string, forceEnterprise bool, s
 	return processes
 }
 
+// CreateMongodProcessesFromMongoDB creates mongod processes directly from MongoDB resource without StatefulSet
+func CreateMongodProcessesFromMongoDB(mongoDBImage string, forceEnterprise bool, mdb *mdbv1.MongoDB, limit int, fcv string, tlsCertPath string) []om.Process {
+	hostnames, names := dns.GetDNSNames(mdb.Name, mdb.ServiceName(), mdb.Namespace, mdb.Spec.GetClusterDomain(), limit, mdb.Spec.DbCommonSpec.GetExternalDomain())
+	processes := make([]om.Process, len(hostnames))
+
+	for idx, hostname := range hostnames {
+		processes[idx] = om.NewMongodProcess(names[idx], hostname, mongoDBImage, forceEnterprise, mdb.Spec.GetAdditionalMongodConfig(), &mdb.Spec, tlsCertPath, mdb.Annotations, fcv)
+	}
+
+	return processes
+}
+
 // CreateMongodProcessesWithLimitMulti creates the process array for automationConfig based on MultiCluster CR spec
 func CreateMongodProcessesWithLimitMulti(mongoDBImage string, forceEnterprise bool, mrs mdbmultiv1.MongoDBMultiCluster, certFileName string) ([]om.Process, error) {
 	hostnames := make([]string, 0)

controllers/om/process/om_process_test.go

@@ -0,0 +1,146 @@
+package process
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	mdbv1 "github.com/mongodb/mongodb-kubernetes/api/v1/mdb"
+	"github.com/mongodb/mongodb-kubernetes/pkg/util/maputil"
+)
+
+const (
+	defaultMongoDBImage = "mongodb/mongodb-enterprise-server:7.0.0"
+	defaultFCV          = "7.0"
+	defaultNamespace    = "test-namespace"
+)
+
+func TestCreateMongodProcessesFromMongoDB(t *testing.T) {
+	t.Run("Happy path - creates processes with correct integration", func(t *testing.T) {
+		mdb := baseReplicaSet("test-rs", 3)
+		processes := CreateMongodProcessesFromMongoDB(
+			defaultMongoDBImage,
+			false,
+			mdb,
+			3,
+			defaultFCV,
+			"",
+		)
+
+		assert.Len(t, processes, 3, "Should create 3 processes")
+
+		// Verify basic integration - processes are created with correct names and FCV
+		for i, process := range processes {
+			expectedName := fmt.Sprintf("test-rs-%d", i)
+			assert.Equal(t, expectedName, process.Name(), "Process name should be generated correctly")
+			assert.Equal(t, defaultFCV, process.FeatureCompatibilityVersion(), "FCV should be set correctly")
+			assert.NotEmpty(t, process.HostName(), "Hostname should be generated")
+		}
+	})
+
+	t.Run("Limit parameter controls process count", func(t *testing.T) {
+		mdb := baseReplicaSet("scale-rs", 5)
+
+		// Test limit less than members (scale up in progress)
+		processesScaleUp := CreateMongodProcessesFromMongoDB(
+			defaultMongoDBImage,
+			false,
+			mdb,
+			3, // limit
+			defaultFCV,
+			"",
+		)
+		assert.Len(t, processesScaleUp, 3, "Limit should control process count during scale up")
+
+		// Test limit greater than members (scale down in progress)
+		processesScaleDown := CreateMongodProcessesFromMongoDB(
+			defaultMongoDBImage,
+			false,
+			mdb,
+			7, // limit
+			defaultFCV,
+			"",
+		)
+		assert.Len(t, processesScaleDown, 7, "Limit should control process count during scale down")
+
+		// Test limit zero
+		processesZero := CreateMongodProcessesFromMongoDB(
+			defaultMongoDBImage,
+			false,
+			mdb,
+			0, // limit
+			defaultFCV,
+			"",
+		)
+		assert.Empty(t, processesZero, "Zero limit should create empty process slice")
+	})
+
+	t.Run("TLS cert path flows through to processes", func(t *testing.T) {
+		mdb := baseReplicaSet("tls-rs", 2)
+		mdb.Spec.Security = &mdbv1.Security{
+			TLSConfig: &mdbv1.TLSConfig{Enabled: true},
+		}
+
+		tlsCertPath := "/custom/path/to/cert.pem"
+		processes := CreateMongodProcessesFromMongoDB(
+			defaultMongoDBImage,
+			false,
+			mdb,
+			2,
+			defaultFCV,
+			tlsCertPath,
+		)
+
+		assert.Len(t, processes, 2)
+
+		// Verify TLS configuration is properly integrated
+		for i, process := range processes {
+			tlsConfig := process.TLSConfig()
+			assert.NotNil(t, tlsConfig, "TLS config should be set when cert path provided")
+			assert.Equal(t, tlsCertPath, tlsConfig["certificateKeyFile"], "TLS cert path should match at index %d", i)
+		}
+	})
+}
+
+func TestCreateMongodProcessesFromMongoDB_AdditionalConfig(t *testing.T) {
+	config := mdbv1.NewAdditionalMongodConfig("storage.engine", "inMemory").
+		AddOption("replication.oplogSizeMB", 2048)
+
+	mdb := mdbv1.NewReplicaSetBuilder().
+		SetName("config-rs").
+		SetNamespace(defaultNamespace).
+		SetMembers(2).
+		SetVersion("7.0.0").
+		SetFCVersion(defaultFCV).
+		SetAdditionalConfig(config).
+		Build()
+
+	processes := CreateMongodProcessesFromMongoDB(
+		defaultMongoDBImage,
+		false,
+		mdb,
+		2,
+		defaultFCV,
+		"",
+	)
+
+	assert.Len(t, processes, 2)
+
+	for i, process := range processes {
+		assert.Equal(t, "inMemory", maputil.ReadMapValueAsInterface(process.Args(), "storage", "engine"),
+			"Storage engine mismatch at index %d", i)
+		assert.Equal(t, 2048, maputil.ReadMapValueAsInterface(process.Args(), "replication", "oplogSizeMB"),
+			"OplogSizeMB mismatch at index %d", i)
+	}
+}
+
+func baseReplicaSet(name string, members int) *mdbv1.MongoDB {
+	return mdbv1.NewReplicaSetBuilder().
+		SetName(name).
+		SetNamespace(defaultNamespace).
+		SetMembers(members).
+		SetVersion("7.0.0").
+		SetFCVersion(defaultFCV).
+		Build()
+}

controllers/om/replicaset/om_replicaset.go

@@ -30,6 +30,16 @@ func BuildFromStatefulSetWithReplicas(mongoDBImage string, forceEnterprise bool,
 	return rsWithProcesses
 }
 
+// BuildFromMongoDBWithReplicas returns a replica set that can be set in the Automation Config
+// based on the given MongoDB resource directly without requiring a StatefulSet.
+func BuildFromMongoDBWithReplicas(mongoDBImage string, forceEnterprise bool, mdb *mdbv1.MongoDB, replicas int, fcv string, tlsCertPath string) om.ReplicaSetWithProcesses {
+	members := process.CreateMongodProcessesFromMongoDB(mongoDBImage, forceEnterprise, mdb, replicas, fcv, tlsCertPath)
+	replicaSet := om.NewReplicaSet(mdb.Name, mdb.Spec.GetMongoDBVersion())
+	rsWithProcesses := om.NewReplicaSetWithProcesses(replicaSet, members, mdb.Spec.GetMemberOptions())
+	rsWithProcesses.SetHorizons(mdb.Spec.GetHorizonConfig())
+	return rsWithProcesses
+}
+
 // PrepareScaleDownFromMap performs additional steps necessary to make sure removed members are not primary (so no
 // election happens and replica set is available) (see
 // https://jira.mongodb.org/browse/HELP-3818?focusedCommentId=1548348 for more details)
@@ -65,30 +75,13 @@ func PrepareScaleDownFromMap(omClient om.Connection, rsMembers map[string][]stri
 		log.Debugw("Marked replica set members as non-voting", "replica set with members", rsMembers)
 	}
 
-	// TODO practice shows that automation agents can get stuck on setting db to "disabled" also it seems that this process
-	// works correctly without explicit disabling - feel free to remove this code after some time when it is clear
-	// that everything works correctly without disabling
-
-	// Stage 2. Set disabled to true
-	//err = omClient.ReadUpdateDeployment(
-	//	func(d om.Deployment) error {
-	//		d.DisableProcesses(allProcesses)
-	//		return nil
-	//	},
-	//)
-	//
-	//if err != nil {
-	//	return errors.New(fmt.Sprintf("Unable to set disabled to true, hosts: %v, err: %w", allProcesses, err))
-	//}
-	//log.Debugw("Disabled processes", "processes", allProcesses)
-
 	log.Infow("Performed some preliminary steps to support scale down", "hosts", processes)
 
 	return nil
 }
 
-func PrepareScaleDownFromStatefulSet(omClient om.Connection, statefulSet appsv1.StatefulSet, rs *mdbv1.MongoDB, log *zap.SugaredLogger) error {
-	_, podNames := dns.GetDnsForStatefulSetReplicasSpecified(statefulSet, rs.Spec.GetClusterDomain(), rs.Status.Members, nil)
+func PrepareScaleDownFromMongoDB(omClient om.Connection, rs *mdbv1.MongoDB, log *zap.SugaredLogger) error {
+	_, podNames := dns.GetDNSNames(rs.Name, rs.ServiceName(), rs.Namespace, rs.Spec.GetClusterDomain(), rs.Status.Members, rs.Spec.DbCommonSpec.GetExternalDomain())
 	podNames = podNames[scale.ReplicasThisReconciliation(rs):rs.Status.Members]
 
 	if len(podNames) != 1 {

controllers/om/replicaset/om_replicaset_test.go

@@ -0,0 +1,95 @@
+package replicaset
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"k8s.io/utils/ptr"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	mdbv1 "github.com/mongodb/mongodb-kubernetes/api/v1/mdb"
+	"github.com/mongodb/mongodb-kubernetes/controllers/om"
+	"github.com/mongodb/mongodb-kubernetes/mongodb-community-operator/pkg/automationconfig"
+)
+
+// This test focuses on the integration/glue logic, not re-testing components.
+func TestBuildFromMongoDBWithReplicas(t *testing.T) {
+	memberOptions := []automationconfig.MemberOptions{
+		{Votes: ptr.To(1), Priority: ptr.To("1.0")},
+		{Votes: ptr.To(1), Priority: ptr.To("0.5")},
+		{Votes: ptr.To(0), Priority: ptr.To("0")}, // Non-voting member
+	}
+
+	mdb := &mdbv1.MongoDB{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      "test-rs",
+			Namespace: "test-namespace",
+		},
+		Spec: mdbv1.MongoDbSpec{
+			DbCommonSpec: mdbv1.DbCommonSpec{
+				Version: "7.0.5",
+				Security: &mdbv1.Security{
+					TLSConfig:      &mdbv1.TLSConfig{},
+					Authentication: &mdbv1.Authentication{},
+				},
+				Connectivity: &mdbv1.MongoDBConnectivity{
+					ReplicaSetHorizons: []mdbv1.MongoDBHorizonConfig{},
+				},
+			},
+			Members:      5, // Spec (target) is 5 members
+			MemberConfig: memberOptions,
+		},
+	}
+
+	// 3 replicas is less than spec.Members, scale up scenario
+	replicas := 3
+	rsWithProcesses := BuildFromMongoDBWithReplicas(
+		"mongodb/mongodb-enterprise-server:7.0.5",
+		false,
+		mdb,
+		replicas,
+		"7.0",
+		"",
+	)
+
+	// Assert: ReplicaSet structure
+	assert.Equal(t, "test-rs", rsWithProcesses.Rs.Name(), "ReplicaSet ID should match MongoDB name")
+	assert.Equal(t, "1", rsWithProcesses.Rs["protocolVersion"], "Protocol version should be set to 1 for this MongoDB version")
+
+	// Assert: Member count is controlled by replicas parameter, NOT mdb.Spec.Members
+	members := rsWithProcesses.Rs["members"].([]om.ReplicaSetMember)
+	assert.Len(t, members, replicas, "Member count should match replicas parameter (3), not mdb.Spec.Members (5)")
+	assert.Equal(t, 3, len(members), "Should have exactly 3 members")
+
+	// Assert: Processes are created correctly
+	assert.Len(t, rsWithProcesses.Processes, replicas, "Process count should match replicas parameter")
+	expectedProcessNames := []string{"test-rs-0", "test-rs-1", "test-rs-2"}
+	expectedHostnames := []string{
+		"test-rs-0.test-rs-svc.test-namespace.svc.cluster.local",
+		"test-rs-1.test-rs-svc.test-namespace.svc.cluster.local",
+		"test-rs-2.test-rs-svc.test-namespace.svc.cluster.local",
+	}
+
+	for i := 0; i < replicas; i++ {
+		assert.Equal(t, expectedProcessNames[i], rsWithProcesses.Processes[i].Name(),
+			"Process name mismatch at index %d", i)
+		assert.Equal(t, expectedHostnames[i], rsWithProcesses.Processes[i].HostName(),
+			"Process hostname mismatch at index %d", i)
+	}
+
+	// Assert: Member options are propagated
+	assert.Equal(t, 1, members[0].Votes(), "Member 0 should have 1 vote")
+	assert.Equal(t, float32(1.0), members[0].Priority(), "Member 0 should have priority 1.0")
+	assert.Equal(t, 1, members[1].Votes(), "Member 1 should have 1 vote")
+	assert.Equal(t, float32(0.5), members[1].Priority(), "Member 1 should have priority 0.5")
+	assert.Equal(t, 0, members[2].Votes(), "Member 2 should have 0 votes (non-voting)")
+	assert.Equal(t, float32(0), members[2].Priority(), "Member 2 should have priority 0")
+
+	// Assert: Member host field contains process name (not full hostname)
+	// Note: ReplicaSetMember["host"] is the process name, not the full hostname
+	for i := 0; i < replicas; i++ {
+		assert.Equal(t, expectedProcessNames[i], members[i].Name(),
+			"Member host should match process name at index %d", i)
+	}
+}

controllers/operator/agents/agents.go

@@ -96,6 +96,19 @@ func WaitForRsAgentsToRegister(set appsv1.StatefulSet, members int, clusterName
 	return nil
 }
 
+// WaitForRsAgentsToRegisterByResource waits for RS agents to register using MongoDB resource directly without StatefulSet
+func WaitForRsAgentsToRegisterByResource(rs *mdbv1.MongoDB, members int, omConnection om.Connection, log *zap.SugaredLogger) error {
+	hostnames, _ := dns.GetDNSNames(rs.Name, rs.ServiceName(), rs.Namespace, rs.Spec.GetClusterDomain(), members, rs.Spec.DbCommonSpec.GetExternalDomain())
+
+	log = log.With("mongodb", rs.Name)
+
+	ok, msg := waitUntilRegistered(omConnection, log, retryParams{retrials: 5, waitSeconds: 3}, hostnames...)
+	if !ok {
+		return getAgentRegisterError(msg)
+	}
+	return nil
+}
+
 // WaitForRsAgentsToRegisterSpecifiedHostnames waits for the specified agents to registry with Ops Manager.
 func WaitForRsAgentsToRegisterSpecifiedHostnames(omConnection om.Connection, hostnames []string, log *zap.SugaredLogger) error {
 	ok, msg := waitUntilRegistered(omConnection, log, retryParams{retrials: 10, waitSeconds: 9}, hostnames...)

controllers/operator/mongodbmultireplicaset_controller.go

@@ -166,11 +166,6 @@ func (r *ReconcileMongoDbMultiReplicaSet) Reconcile(ctx context.Context, request
 
 	r.SetupCommonWatchers(&mrs, nil, nil, mrs.Name)
 
-	publishAutomationConfigFirst, err := r.publishAutomationConfigFirstMultiCluster(ctx, &mrs, log)
-	if err != nil {
-		return r.updateStatus(ctx, &mrs, workflow.Failed(err), log)
-	}
-
 	// If tls is enabled we need to configure the "processes" array in opsManager/Cloud Manager with the
 	// correct tlsCertPath, with the new tls design, this path has the certHash in it(so that cert can be rotated
 	// without pod restart).
@@ -210,6 +205,11 @@ func (r *ReconcileMongoDbMultiReplicaSet) Reconcile(ctx context.Context, request
 		}
 	}
 
+	publishAutomationConfigFirst, err := r.publishAutomationConfigFirstMultiCluster(ctx, &mrs, log)
+	if err != nil {
+		return r.updateStatus(ctx, &mrs, workflow.Failed(err), log)
+	}
+
 	status := workflow.RunInGivenOrder(publishAutomationConfigFirst,
 		func() workflow.Status {
 			if err := r.updateOmDeploymentRs(ctx, conn, mrs, agentCertPath, tlsCertPath, internalClusterCertPath, false, log); err != nil {