Repository: FoundationDB/fdb-kubernetes-operator
Pull request #2048 (2024-06-04 15:46:28)

Add e2e migration test from split to unified image
# Description

Adding e2e test cases for converting a cluster from split image to unified and from the unified to the split image.

## Type of change

*Please select one of the options below.*

- Other

## Discussion

I adjusted the e2e testing code to make sure we can test those migrations.

## Testing

Manually tested:

```text
tail -f /Users/jscheuermann/go/src/github.com/FoundationDB/fdb-kubernetes-operator/e2e/../logs/test_operator_migrate_image_type.log

[ReportAfterSuite] Autogenerated ReportAfterSuite for --junit-report
autogenerated by Ginkgo
[ReportAfterSuite] PASSED [0.003 seconds]
------------------------------

Ran 2 of 2 Specs in 1011.427 seconds
SUCCESS! -- 2 Passed | 0 Failed | 0 Pending | 0 Skipped
--- PASS: TestOperator (1011.43s)
PASS
ok      github.com/FoundationDB/fdb-kubernetes-operator/e2e/test_operator_migrate_image_type    1011.843s
```

## Documentation

-

## Follow-up

Run those tests regularly.



Patch:
diff --git a/api/v1beta2/foundationdb_database_configuration.go b/api/v1beta2/foundationdb_database_configuration.go
index 5368a4fb..a5e651ef 100644
--- a/api/v1beta2/foundationdb_database_configuration.go
+++ b/api/v1beta2/foundationdb_database_configuration.go
@@ -24,6 +24,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"html/template"
+	"math"
 	"reflect"
 	"sort"
 	"strconv"
@@ -562,6 +563,58 @@ func (configuration DatabaseConfiguration) GetNextConfigurationChange(finalConfi
 	return finalConfiguration
 }
 
+// GetPrimaryDCID will return the DC ID of the primary data center. The primary data center is the one with the highest
+// priority in the region configuration.
+// See: https://github.com/apple/foundationdb/wiki/Multi-Region-Replication#concepts
+func (configuration DatabaseConfiguration) GetPrimaryDCID() string {
+	var candidate string
+	priority := math.MinInt
+
+	for _, region := range configuration.Regions {
+		for _, dataCenter := range region.DataCenters {
+			if dataCenter.Satellite != 0 {
+				continue
+			}
+
+			if dataCenter.Priority > priority {
+				candidate = dataCenter.ID
+				priority = dataCenter.Priority
+			}
+		}
+	}
+
+	return candidate
+}
+
+// GetMainDCsAndSatellites will return a set of main dcs and a set of satellites. If a dc is a main dc and a satellite
+// it will only be counted as a main dc.
+func (configuration DatabaseConfiguration) GetMainDCsAndSatellites() (map[string]None, map[string]None) {
+	mainDCs := map[string]None{}
+	satellites := map[string]None{}
+
+	for _, region := range configuration.Regions {
+		for _, dataCenter := range region.DataCenters {
+			if dataCenter.Satellite == 0 {
+				mainDCs[dataCenter.ID] = None{}
+				// If the dc ID is present as satellite we will remove it.
+				delete(satellites, dataCenter.ID)
+
+				// satellites
+				continue
+			}
+
+			// If the dc ID is already present in the main dcs, we will not count this as a satellite.
+			if _, ok := mainDCs[dataCenter.ID]; ok {
+				continue
+			}
+
+			satellites[dataCenter.ID] = None{}
+		}
+	}
+
+	return mainDCs, satellites
+}
+
 func (configuration DatabaseConfiguration) getRegionPriorities() map[string]int {
 	priorities := make(map[string]int, len(configuration.Regions))
 
diff --git a/controllers/change_coordinators.go b/controllers/change_coordinators.go
index 78dde269..293571f9 100644
--- a/controllers/change_coordinators.go
+++ b/controllers/change_coordinators.go
@@ -184,7 +184,8 @@ func selectCoordinators(logger logr.Logger, cluster *fdbv1beta2.FoundationDBClus
 	}
 
 	coordinators, err := locality.ChooseDistributedProcesses(cluster, candidates, coordinatorCount, locality.ProcessSelectionConstraint{
-		HardLimits: locality.GetHardLimits(cluster),
+		HardLimits:            locality.GetHardLimits(cluster),
+		SelectingCoordinators: true,
 	})
 
 	logger.Info("Current coordinators", "coordinators", coordinators, "error", err)
diff --git a/controllers/generate_initial_cluster_file.go b/controllers/generate_initial_cluster_file.go
index 1dee3415..91d9a222 100644
--- a/controllers/generate_initial_cluster_file.go
+++ b/controllers/generate_initial_cluster_file.go
@@ -124,10 +124,14 @@ func (g generateInitialClusterFile) reconcile(ctx context.Context, r *Foundation
 			logger.Info("Pod is ineligible to be a coordinator due to missing locality information", "processGroupID", processGroupID)
 			continue
 		}
+		currentLocality.Priority = cluster.GetClassCandidatePriority(currentLocality.Class)
 		processLocality = append(processLocality, currentLocality)
 	}
 
-	coordinators, err := locality.ChooseDistributedProcesses(cluster, processLocality, count, locality.ProcessSelectionConstraint{})
+	coordinators, err := locality.ChooseDistributedProcesses(cluster, processLocality, count, locality.ProcessSelectionConstraint{
+		HardLimits:            locality.GetHardLimits(cluster),
+		SelectingCoordinators: true,
+	})
 	if err != nil {
 		return &requeue{curError: err}
 	}
diff --git a/controllers/update_pod_config.go b/controllers/update_pod_config.go
index cc27583f..ed4da3d4 100644
--- a/controllers/update_pod_config.go
+++ b/controllers/update_pod_config.go
@@ -22,7 +22,7 @@ package controllers
 
 import (
 	"context"
-	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -122,7 +122,7 @@ func (updatePodConfig) reconcile(ctx context.Context, r *FoundationDBClusterReco
 				delayedRequeue = false
 			}
 
-			pod.ObjectMeta.Annotations[fdbv1beta2.OutdatedConfigMapKey] = fmt.Sprintf("%d", time.Now().Unix())
+			pod.ObjectMeta.Annotations[fdbv1beta2.OutdatedConfigMapKey] = strconv.FormatInt(time.Now().Unix(), 10)
 			err = r.PodLifecycleManager.UpdateMetadata(ctx, r, cluster, pod)
 			if err != nil {
 				allSynced = false
diff --git a/internal/locality/locality.go b/internal/locality/locality.go
index 83b1d626..18b194f7 100644
--- a/internal/locality/locality.go
+++ b/internal/locality/locality.go
@@ -21,14 +21,14 @@
 package locality
 
 import (
+	"cmp"
 	"errors"
 	"fmt"
-	"math"
-	"sort"
-
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podclient"
 	"github.com/go-logr/logr"
+	"math"
+	"slices"
 )
 
 // Info captures information about a process for the purposes of
@@ -55,16 +55,34 @@ type Info struct {
 // We have to do this to ensure we get a deterministic result for selecting the candidates
 // otherwise we get a (nearly) random result since processes are stored in a map which is by definition
 // not sorted and doesn't return values in a stable way.
-func sortLocalities(processes []Info) {
-	// Sort the processes for ID to ensure we have a stable input
-	sort.SliceStable(processes, func(i, j int) bool {
+func sortLocalities(primaryDC string, processes []Info) {
+	slices.SortStableFunc(processes, func(a, b Info) int {
+		if primaryDC != "" {
+			aDCLocality := a.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]
+			bDCLocality := b.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]
+
+			if aDCLocality != bDCLocality {
+				if aDCLocality == primaryDC {
+					return -1
+				}
+
+				if bDCLocality == primaryDC {
+					return 1
+				}
+			}
+		}
+
 		// If both have the same priority sort them by the process ID
-		if processes[i].Priority == processes[j].Priority {
-			return processes[i].ID < processes[j].ID
+		if a.Priority == b.Priority {
+			return cmp.Compare(a.ID, b.ID)
+		}
+
+		// Prefer processes with a higher priority
+		if a.Priority > b.Priority {
+			return -1
 		}
 
-		// prefer processes with a higher priority
-		return processes[i].Priority > processes[j].Priority
+		return 1
 	})
 }
 
@@ -146,13 +164,16 @@ type ProcessSelectionConstraint struct {
 	// HardLimits defines a maximum number of processes to recruit on any single
 	// value for a given locality field.
 	HardLimits map[string]int
+
+	// SelectingCoordinators must be true when the ChooseDistributedProcesses is used to select coordinators.
+	SelectingCoordinators bool
 }
 
-// ChooseDistributedProcesses recruits a maximally well-distributed set
-// of processes from a set of potential candidates.
+// ChooseDistributedProcesses recruits a maximally well-distributed set of processes from a set of potential candidates.
 func ChooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, processes []Info, count int, constraint ProcessSelectionConstraint) ([]Info, error) {
 	chosen := make([]Info, 0, count)
 	chosenIDs := make(map[string]bool, count)
+	primaryDC := cluster.DesiredDatabaseConfiguration().GetPrimaryDCID()
 
 	fields := constraint.Fields
 	if len(fields) == 0 {
@@ -180,8 +201,8 @@ func ChooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, process
 		currentLimits[field] = 1
 	}
 
-	// Sort the processes to ensure a deterministic result
-	sortLocalities(processes)
+	// Sort the processes to ensure a deterministic result.
+	sortLocalities(primaryDC, processes)
 
 	for len(chosen) < count {
 		choseAny := false
@@ -229,6 +250,7 @@ func ChooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, process
 					break
 				}
 			}
+
 			if !incrementedLimits {
 				return chosen, notEnoughProcessesError{Desired: count, Chosen: len(chosen), Options: processes}
 			}


Test patch:
diff --git a/controllers/change_coordinators_test.go b/controllers/change_coordinators_test.go
index 2c54a1ef..7a539cf7 100644
--- a/controllers/change_coordinators_test.go
+++ b/controllers/change_coordinators_test.go
@@ -83,7 +83,7 @@ var _ = Describe("Change coordinators", func() {
 			When("all processes are healthy", func() {
 				It("should only select storage processes", func() {
 					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 3))
-					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+					Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 					// Only select Storage processes since we select 3 processes and we have 4 storage processes
 					for _, candidate := range candidates {
@@ -104,7 +104,7 @@ var _ = Describe("Change coordinators", func() {
 
 				It("should only select storage processes and exclude the removed process", func() {
 					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 3))
-					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+					Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 					// Only select Storage processes since we select 3 processes and we have 4 storage processes
 					for _, candidate := range candidates {
@@ -147,7 +147,7 @@ var _ = Describe("Change coordinators", func() {
 
 				It("should select 2 storage processes and 1 TLog", func() {
 					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 3))
-					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+					Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 					// Only select Storage processes since we select 3 processes and we have 4 storage processes
 					storageCnt := 0
@@ -195,7 +195,7 @@ var _ = Describe("Change coordinators", func() {
 
 				It("should only select log processes", func() {
 					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 3))
-					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+					Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 					// Only select Storage processes since we select 3 processes and we have 4 storage processes
 					for _, candidate := range candidates {
@@ -227,9 +227,8 @@ var _ = Describe("Change coordinators", func() {
 			var candidates []locality.Info
 			var excludes []string
 			var removals []fdbv1beta2.ProcessGroupID
-			var dcCnt int
-			var satCnt int
 			var shouldFail bool
+			var primaryID string
 
 			BeforeEach(func() {
 				// ensure a clean state
@@ -241,14 +240,11 @@ var _ = Describe("Change coordinators", func() {
 
 			JustBeforeEach(func() {
 				cluster.Spec.DatabaseConfiguration.UsableRegions = 2
-				cluster.Spec.DataCenter = "dc0"
 				cluster.Spec.ProcessGroupsToRemove = removals
-				setDatabaseConfiguration(cluster, satCnt)
-
 				var err error
 				status, err = adminClient.GetStatus()
 				Expect(err).NotTo(HaveOccurred())
-				status.Cluster.Processes = generateProcessInfoForMultiRegion(dcCnt, satCnt, excludes)
+				status.Cluster.Processes = generateProcessInfoForMultiRegion(cluster.Spec.DatabaseConfiguration, excludes, cluster.GetRunningVersion())
 
 				candidates, err = selectCoordinators(testLogger, cluster, status)
 				if shouldFail {
@@ -260,23 +256,58 @@ var _ = Describe("Change coordinators", func() {
 
 			When("using 2 dcs with 1 satellite", func() {
 				BeforeEach(func() {
-					dcCnt = 2
-					satCnt = 1
+					primaryID = internal.GenerateRandomString(10)
+					remoteID := internal.GenerateRandomString(10)
+					satelliteID := internal.GenerateRandomString(10)
+					cluster.Spec.DataCenter = primaryID
+					cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       primaryID,
+									Priority: 1,
+								},
+								{
+									ID:        satelliteID,
+									Satellite: 1,
+									Priority:  1,
+								},
+								{
+									ID:        remoteID,
+									Satellite: 1,
+								},
+							},
+						},
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID: remoteID,
+								},
+								{
+									ID:        satelliteID,
+									Satellite: 1,
+									Priority:  1,
+								},
+								{
+									ID:        primaryID,
+									Satellite: 1,
+								},
+							},
+						},
+					}
 				})
 
 				When("all processes are healthy", func() {
 					It("should only select storage processes in primary and remote and Tlog in satellite", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -290,10 +321,9 @@ var _ = Describe("Change coordinators", func() {
 						Expect(storageCnt).To(BeNumerically("==", 6))
 						Expect(logCnt).To(BeNumerically("==", 3))
 						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 3))
-
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically("==", 3))
+						Expect(len(dcDistribution)).To(BeNumerically("==", 3))
+						for _, dcCount := range dcDistribution {
+							Expect(dcCount).To(BeNumerically("==", 3))
 						}
 					})
 				})
@@ -301,29 +331,27 @@ var _ = Describe("Change coordinators", func() {
 				When("some processes are excluded", func() {
 					BeforeEach(func() {
 						excludes = []string{
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
+							primaryID + "-storage-1",
+							primaryID + "-storage-2",
+							primaryID + "-storage-3",
+							primaryID + "-storage-4",
 						}
 					})
 
 					It("should only select storage processes in primary and remote and Tlog in satellites and processes that are not excluded", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
 							for _, excluded := range excludes {
 								Expect(candidate.ID).NotTo(Equal(excluded))
 							}
 
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -337,10 +365,9 @@ var _ = Describe("Change coordinators", func() {
 						Expect(storageCnt).To(BeNumerically("==", 6))
 						Expect(logCnt).To(BeNumerically("==", 3))
 						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 3))
-
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically("==", 3))
+						Expect(len(dcDistribution)).To(BeNumerically("==", 3))
+						for _, dcCount := range dcDistribution {
+							Expect(dcCount).To(BeNumerically("==", 3))
 						}
 					})
 				})
@@ -348,29 +375,27 @@ var _ = Describe("Change coordinators", func() {
 				When("some processes are removed", func() {
 					BeforeEach(func() {
 						removals = []fdbv1beta2.ProcessGroupID{
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-1"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-2"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-3"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-4"),
 						}
 					})
 
 					It("should only select storage processes in primary and remote and Tlog in satellites and processes that are not removed", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
 							for _, removed := range removals {
 								Expect(candidate.ID).NotTo(Equal(removed))
 							}
 
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -384,10 +409,10 @@ var _ = Describe("Change coordinators", func() {
 						Expect(storageCnt).To(BeNumerically("==", 6))
 						Expect(logCnt).To(BeNumerically("==", 3))
 						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 3))
+						Expect(len(dcDistribution)).To(BeNumerically("==", 3))
 
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically("==", 3))
+						for _, dcCount := range dcDistribution {
+							Expect(dcCount).To(BeNumerically("==", 3))
 						}
 					})
 				})
@@ -395,18 +420,18 @@ var _ = Describe("Change coordinators", func() {
 				When("all processes in a dc are excluded", func() {
 					BeforeEach(func() {
 						excludes = []string{
-							"dc0-storage-0",
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
-							"dc0-storage-5",
-							"dc0-storage-6",
-							"dc0-storage-7",
-							"dc0-log-0",
-							"dc0-log-1",
-							"dc0-log-2",
-							"dc0-log-3",
+							primaryID + "-storage-0",
+							primaryID + "-storage-1",
+							primaryID + "-storage-2",
+							primaryID + "-storage-3",
+							primaryID + "-storage-4",
+							primaryID + "-storage-5",
+							primaryID + "-storage-6",
+							primaryID + "-storage-7",
+							primaryID + "-log-0",
+							primaryID + "-log-1",
+							primaryID + "-log-2",
+							primaryID + "-log-3",
 						}
 
 						shouldFail = true
@@ -432,23 +457,60 @@ var _ = Describe("Change coordinators", func() {
 
 			When("using 2 dcs and 2 satellites", func() {
 				BeforeEach(func() {
-					dcCnt = 2
-					satCnt = 2
+					primaryID = internal.GenerateRandomString(10)
+					remoteID := internal.GenerateRandomString(10)
+					primarySatelliteID := internal.GenerateRandomString(10)
+					remoteSatelliteID := internal.GenerateRandomString(10)
+
+					cluster.Spec.DataCenter = primaryID
+					cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       primaryID,
+									Priority: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Satellite: 1,
+									Priority:  1,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Satellite: 1,
+								},
+							},
+						},
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID: remoteID,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Satellite: 1,
+									Priority:  1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Satellite: 1,
+								},
+							},
+						},
+					}
 				})
 
 				When("all processes are healthy", func() {
 					It("should only select storage processes in primary and remote and Tlog in satellites", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -458,13 +520,19 @@ var _ = Describe("Change coordinators", func() {
 							}
 						}
 
-						// We should have 3 SS in dc0 2 SS in dc1 and 2 Tlogs in sat0 and 2 Tlogs in sat1
+						// We should have 3 SS in the primary dc 2 SS in the remote dc and 2 Tlogs in each satellite.
 						Expect(storageCnt).To(BeNumerically("==", 5))
 						Expect(logCnt).To(BeNumerically("==", 4))
-						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 4))
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically(">=", 2))
+						Expect(primaryID).To(Equal(cluster.DesiredDatabaseConfiguration().GetPrimaryDCID()))
+						// We should have 4 different dcs,
+						Expect(dcDistribution).To(HaveLen(4))
+						for dcID, dcCount := range dcDistribution {
+							if dcID == primaryID {
+								Expect(dcCount).To(BeNumerically("==", 3))
+								continue
+							}
+
+							Expect(dcCount).To(BeNumerically("==", 2))
 						}
 					})
 				})
@@ -472,29 +540,26 @@ var _ = Describe("Change coordinators", func() {
 				When("some processes are excluded", func() {
 					BeforeEach(func() {
 						excludes = []string{
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
+							primaryID + "-storage-1",
+							primaryID + "-storage-2",
+							primaryID + "-storage-3",
+							primaryID + "-storage-4",
 						}
 					})
 
 					It("should only select storage processes in primary and remote and Tlog in satellites and processes that are not excluded", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
 							for _, excluded := range excludes {
 								Expect(candidate.ID).NotTo(Equal(excluded))
 							}
-
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -507,10 +572,15 @@ var _ = Describe("Change coordinators", func() {
 						// We should have 3 SS in dc0 2 SS in dc1 and 2 Tlogs in sat0 and 2 Tlogs in sat1
 						Expect(storageCnt).To(BeNumerically("==", 5))
 						Expect(logCnt).To(BeNumerically("==", 4))
-						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 4))
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically(">=", 2))
+						// We should have 4 different dcs,
+						Expect(dcDistribution).To(HaveLen(4))
+						for dcID, dcCount := range dcDistribution {
+							if dcID == primaryID {
+								Expect(dcCount).To(BeNumerically("==", 3))
+								continue
+							}
+
+							Expect(dcCount).To(BeNumerically("==", 2))
 						}
 					})
 				})
@@ -518,29 +588,27 @@ var _ = Describe("Change coordinators", func() {
 				When("some processes are removed", func() {
 					BeforeEach(func() {
 						removals = []fdbv1beta2.ProcessGroupID{
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-1"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-2"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-3"),
+							fdbv1beta2.ProcessGroupID(primaryID + "-storage-4"),
 						}
 					})
 
 					It("should only select storage processes in primary and remote and Tlog in satellites and processes that are not removed", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).To(HaveLen(cluster.DesiredCoordinatorCount()))
 
 						// Only select Storage processes since we select 3 processes and we have 4 storage processes
 						storageCnt := 0
 						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
 							for _, removed := range removals {
 								Expect(candidate.ID).NotTo(Equal(removed))
 							}
 
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 							if candidate.Class == fdbv1beta2.ProcessClassStorage {
 								storageCnt++
 							}
@@ -553,10 +621,15 @@ var _ = Describe("Change coordinators", func() {
 						// We should have 3 SS in dc0 2 SS in dc1 and 2 Tlogs each in sat0 and in sat1
 						Expect(storageCnt).To(BeNumerically("==", 5))
 						Expect(logCnt).To(BeNumerically("==", 4))
-						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 4))
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically(">=", 2))
+						// We should have 4 different dcs.
+						Expect(dcDistribution).To(HaveLen(4))
+						for dcID, dcCount := range dcDistribution {
+							if dcID == primaryID {
+								Expect(dcCount).To(BeNumerically("==", 3))
+								continue
+							}
+
+							Expect(dcCount).To(BeNumerically("==", 2))
 						}
 					})
 				})
@@ -564,53 +637,35 @@ var _ = Describe("Change coordinators", func() {
 				When("all processes in a dc are excluded", func() {
 					BeforeEach(func() {
 						excludes = []string{
-							"dc0-storage-0",
-							"dc0-storage-1",
-							"dc0-storage-2",
-							"dc0-storage-3",
-							"dc0-storage-4",
-							"dc0-storage-5",
-							"dc0-storage-6",
-							"dc0-storage-7",
-							"dc0-log-0",
-							"dc0-log-1",
-							"dc0-log-2",
-							"dc0-log-3",
+							primaryID + "-storage-0",
+							primaryID + "-storage-1",
+							primaryID + "-storage-2",
+							primaryID + "-storage-3",
+							primaryID + "-storage-4",
+							primaryID + "-storage-5",
+							primaryID + "-storage-6",
+							primaryID + "-storage-7",
+							primaryID + "-log-0",
+							primaryID + "-log-1",
+							primaryID + "-log-2",
+							primaryID + "-log-3",
 						}
 					})
 
 					It("should select 3 processes in each remaining dc", func() {
 						Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
-						Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
+						Expect(candidates).NotTo(BeEmpty())
 
-						// Only select Storage processes since we select 3 processes and we have 4 storage processes
-						storageCnt := 0
-						logCnt := 0
-						zoneCnt := map[string]int{}
+						dcDistribution := map[string]int{}
 						for _, candidate := range candidates {
-							for _, excluded := range excludes {
-								Expect(candidate.ID).NotTo(Equal(excluded))
-							}
-
-							zone := strings.Split(candidate.ID, "-")[0]
-							zoneCnt[zone]++
-
-							if candidate.Class == fdbv1beta2.ProcessClassStorage {
-								storageCnt++
-							}
-
-							if candidate.Class == fdbv1beta2.ProcessClassLog {
-								logCnt++
-							}
+							dcDistribution[candidate.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
 						}
 
-						// We should have 3 SS in dc1 and 3 Tlogs each in sat0 and sat1
-						Expect(storageCnt).To(BeNumerically("==", 3))
-						Expect(logCnt).To(BeNumerically("==", 6))
-						// We should have 3 different zones
-						Expect(len(zoneCnt)).To(BeNumerically("==", 3))
-						for _, zoneVal := range zoneCnt {
-							Expect(zoneVal).To(BeNumerically("==", 3))
+						// We should have 3 different dcs as the primary has no valid processes.
+						Expect(dcDistribution).To(HaveLen(3))
+						for dcID, dcCount := range dcDistribution {
+							Expect(dcID).NotTo(Equal(primaryID))
+							Expect(dcCount).To(BeNumerically("==", 3))
 						}
 					})
 				})
@@ -641,7 +696,7 @@ var _ = Describe("Change coordinators", func() {
 				status, err = adminClient.GetStatus()
 				Expect(err).NotTo(HaveOccurred())
 
-				status.Cluster.Processes = generateProcessInfoForThreeDataHall(3, nil)
+				status.Cluster.Processes = generateProcessInfoForThreeDataHall(3, nil, cluster.GetRunningVersion())
 
 				candidates, err = selectCoordinators(logr.Discard(), cluster, status)
 				Expect(err).NotTo(HaveOccurred())
@@ -652,17 +707,15 @@ var _ = Describe("Change coordinators", func() {
 					Expect(cluster.DesiredCoordinatorCount()).To(BeNumerically("==", 9))
 					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
 
-					dataHallCount := map[string]int{}
+					dataHallCounts := map[string]int{}
 					for _, candidate := range candidates {
-						Expect(candidate.ID).To(ContainSubstring("storage"))
-						dataHallCount[strings.Split(candidate.ID, "-")[0]]++
+						Expect(candidate.Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+						dataHallCounts[candidate.LocalityData[fdbv1beta2.FDBLocalityDataHallKey]]++
 					}
 
-					Expect(dataHallCount).To(Equal(map[string]int{
-						"datahall0": 3,
-						"datahall1": 3,
-						"datahall2": 3,
-					}))
+					for _, dataHallCount := range dataHallCounts {
+						Expect(dataHallCount).To(BeNumerically("==", 3))
+					}
 				})
 			})
 
@@ -681,18 +734,16 @@ var _ = Describe("Change coordinators", func() {
 					Expect(len(candidates)).To(BeNumerically("==", cluster.DesiredCoordinatorCount()))
 
 					// Only select Storage processes since we select 3 processes and we have 4 storage processes
-					dataHallCount := map[string]int{}
+					dataHallCounts := map[string]int{}
 					for _, candidate := range candidates {
 						Expect(candidate.ID).NotTo(Equal(removedProcess))
-						Expect(candidate.ID).To(ContainSubstring("storage"))
-						dataHallCount[strings.Split(candidate.ID, "-")[0]]++
+						Expect(candidate.Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+						dataHallCounts[candidate.LocalityData[fdbv1beta2.FDBLocalityDataHallKey]]++
 					}
 
-					Expect(dataHallCount).To(Equal(map[string]int{
-						"datahall0": 3,
-						"datahall1": 3,
-						"datahall2": 3,
-					}))
+					for _, dataHallCount := range dataHallCounts {
+						Expect(dataHallCount).To(BeNumerically("==", 3))
+					}
 				})
 			})
 		})
@@ -1002,38 +1053,37 @@ var _ = Describe("Change coordinators", func() {
 	)
 })
 
-func generateProcessInfoForMultiRegion(dcCount int, satCount int, excludes []string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
+func generateProcessInfoForMultiRegion(config fdbv1beta2.DatabaseConfiguration, excludes []string, version string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
 	res := map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo{}
 	logCnt := 4
 
-	for i := 0; i < dcCount; i++ {
-		dcid := fmt.Sprintf("dc%d", i)
-
-		generateProcessInfoDetails(res, dcid, "", 8, excludes, fdbv1beta2.ProcessClassStorage)
-		generateProcessInfoDetails(res, dcid, "", logCnt, excludes, fdbv1beta2.ProcessClassLog)
+	mainDCs, satelliteDCs := config.GetMainDCsAndSatellites()
+	for dcID := range mainDCs {
+		generateProcessInfoDetails(res, dcID, "", 8, excludes, fdbv1beta2.ProcessClassStorage, version)
+		generateProcessInfoDetails(res, dcID, "", logCnt, excludes, fdbv1beta2.ProcessClassLog, version)
 	}
 
-	for i := 0; i < satCount; i++ {
-		generateProcessInfoDetails(res, fmt.Sprintf("sat%d", i), "", logCnt, excludes, fdbv1beta2.ProcessClassLog)
+	for satelliteDCID := range satelliteDCs {
+		generateProcessInfoDetails(res, satelliteDCID, "", logCnt, excludes, fdbv1beta2.ProcessClassLog, version)
 	}
 
 	return res
 }
 
-func generateProcessInfoForThreeDataHall(dataHallCount int, excludes []string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
+func generateProcessInfoForThreeDataHall(dataHallCount int, excludes []string, version string) map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo {
 	res := map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo{}
 	logCnt := 4
 
 	for i := 0; i < dataHallCount; i++ {
-		dataHallID := fmt.Sprintf("datahall%d", i)
-		generateProcessInfoDetails(res, "", dataHallID, 8, excludes, fdbv1beta2.ProcessClassStorage)
-		generateProcessInfoDetails(res, "", dataHallID, logCnt, excludes, fdbv1beta2.ProcessClassLog)
+		dataHallID := internal.GenerateRandomString(10)
+		generateProcessInfoDetails(res, "", dataHallID, 8, excludes, fdbv1beta2.ProcessClassStorage, version)
+		generateProcessInfoDetails(res, "", dataHallID, logCnt, excludes, fdbv1beta2.ProcessClassLog, version)
 	}
 
 	return res
 }
 
-func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo, dcID string, dataHall string, cnt int, excludes []string, pClass fdbv1beta2.ProcessClass) {
+func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.FoundationDBStatusProcessInfo, dcID string, dataHall string, cnt int, excludes []string, pClass fdbv1beta2.ProcessClass, version string) {
 	for idx := 0; idx < cnt; idx++ {
 		excluded := false
 		var zoneID string
@@ -1058,6 +1108,7 @@ func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.Fou
 		addr := fmt.Sprintf("1.1.1.%d:4501", len(res))
 		processInfo := fdbv1beta2.FoundationDBStatusProcessInfo{
 			ProcessClass: pClass,
+			Version:      version,
 			Locality: map[string]string{
 				fdbv1beta2.FDBLocalityInstanceIDKey: zoneID,
 				fdbv1beta2.FDBLocalityZoneIDKey:     zoneID,
@@ -1081,71 +1132,3 @@ func generateProcessInfoDetails(res map[fdbv1beta2.ProcessGroupID]fdbv1beta2.Fou
 		res[fdbv1beta2.ProcessGroupID(zoneID)] = processInfo
 	}
 }
-
-func setDatabaseConfiguration(cluster *fdbv1beta2.FoundationDBCluster, satCnt int) {
-	if satCnt == 1 {
-		cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
-			{
-				DataCenters: []fdbv1beta2.DataCenter{
-					{
-						ID: "dc0",
-					},
-					{
-						ID:        "sat0",
-						Satellite: 1,
-					},
-					{
-						ID:        "dc1",
-						Satellite: 1,
-						Priority:  1,
-					},
-				},
-			},
-			{
-				DataCenters: []fdbv1beta2.DataCenter{
-					{
-						ID: "dc1",
-					},
-					{
-						ID:        "sat0",
-						Satellite: 1,
-					},
-					{
-						ID:        "dc0",
-						Satellite: 1,
-						Priority:  1,
-					},
-				},
-			},
-		}
-		return
-	}
-
-	if satCnt == 2 {
-		cluster.Spec.DatabaseConfiguration.Regions = []fdbv1beta2.Region{
-			{
-				DataCenters: []fdbv1beta2.DataCenter{
-					{
-						ID: "dc0",
-					},
-					{
-						ID:        "sat0",
-						Satellite: 1,
-					},
-				},
-			},
-			{
-				DataCenters: []fdbv1beta2.DataCenter{
-					{
-						ID:       "dc1",
-						Priority: 1,
-					},
-					{
-						ID:        "sat1",
-						Satellite: 1,
-					},
-				},
-			},
-		}
-	}
-}
diff --git a/e2e/Makefile b/e2e/Makefile
index 784467e9..718b1115 100644
--- a/e2e/Makefile
+++ b/e2e/Makefile
@@ -15,6 +15,7 @@ FDB_VERSION?=7.1.57
 NEXT_FDB_VERSION?=7.3.33
 ## Expectation is that you are running standard build image which generates both regular and debug (Symbols) images.
 FDB_IMAGE?=foundationdb/foundationdb:$(FDB_VERSION)
+UNIFIED_FDB_IMAGE?=foundationdb/fdb-kubernetes-monitor:$(FDB_VERSION)
 SIDECAR_IMAGE?=foundationdb/foundationdb-kubernetes-sidecar:$(FDB_VERSION)-1
 OPERATOR_IMAGE?=foundationdb/fdb-kubernetes-operator:latest
 REGISTRY?=docker.io
@@ -134,4 +135,5 @@ nightly-tests: run
 	  --cluster-name=$(CLUSTER_NAME) \
 	  --storage-engine=$(STORAGE_ENGINE) \
 	  --fdb-version-tag-mapping=$(FDB_VERSION_TAG_MAPPING) \
+	  --unified-fdb-image=$(UNIFIED_FDB_IMAGE) \
 	  | grep -v 'constructing many client instances from the same exec auth config can cause performance problems during cert rotation' &> $(BASE_DIR)/../logs/$<.log
diff --git a/e2e/fixtures/cluster_config.go b/e2e/fixtures/cluster_config.go
index e09e58c0..ae318b39 100644
--- a/e2e/fixtures/cluster_config.go
+++ b/e2e/fixtures/cluster_config.go
@@ -76,6 +76,8 @@ type ClusterConfig struct {
 	UseLocalityBasedExclusions bool
 	// UseDNS if enabled the FoundationDBCluster resource will enable the DNS feature.
 	UseDNS bool
+	// If enabled the cluster will be setup with the unified image.
+	UseUnifiedImage *bool
 	// CreationTracker if specified will be used to log the time between the creations steps.
 	CreationTracker CreationTrackerLogger
 	// Number of machines, this is used for calculating the number of Pods and is not correlated to the actual number
@@ -519,5 +521,6 @@ func (config *ClusterConfig) Copy() *ClusterConfig {
 		CreationCallback:           config.CreationCallback,
 		UseDNS:                     config.UseDNS,
 		UseLocalityBasedExclusions: config.UseLocalityBasedExclusions,
+		UseUnifiedImage:            config.UseUnifiedImage,
 	}
 }
diff --git a/e2e/fixtures/factory.go b/e2e/fixtures/factory.go
index 2958d422..eaf1fae1 100644
--- a/e2e/fixtures/factory.go
+++ b/e2e/fixtures/factory.go
@@ -203,15 +203,19 @@ func (factory *Factory) CreateFdbHaCluster(
 	return cluster
 }
 
-func (factory *Factory) getContainerOverrides(
-	debugSymbols bool,
-) (fdbv1beta2.ContainerOverrides, fdbv1beta2.ContainerOverrides) {
+// GetMainContainerOverrides will return the main container overrides.
+func (factory *Factory) GetMainContainerOverrides(debugSymbols bool, unifiedImage bool) fdbv1beta2.ContainerOverrides {
+	image := factory.GetFoundationDBImage()
+	if unifiedImage {
+		image = factory.GetUnifiedFoundationDBImage()
+	}
+
 	mainImage, mainTag := GetBaseImageAndTag(
-		GetDebugImage(debugSymbols, factory.GetFoundationDBImage()),
+		GetDebugImage(debugSymbols, image),
 	)
 
-	// If the version tag mapping is define make use of that.
-	imageConfigs := factory.options.getImageVersionConfig(mainImage)
+	// If the version tag mapping is defined make use of that.
+	imageConfigs := factory.options.getImageVersionConfig(mainImage, false)
 	if len(imageConfigs) == 0 {
 		// The first entry is version specific e.g. this image + tag (if specified) will be used for the provided version
 		// the second entry ensures we set the base image for e.g. upgrades independent of the version.
@@ -227,35 +231,42 @@ func (factory *Factory) getContainerOverrides(
 		}
 	}
 
-	mainOverrides := fdbv1beta2.ContainerOverrides{
+	return fdbv1beta2.ContainerOverrides{
 		EnableTLS:    false,
 		ImageConfigs: imageConfigs,
 	}
+}
 
+// GetSidecarContainerOverrides will return the sidecar container overrides. If the unified image should be used an empty
+// container override will be returned.
+func (factory *Factory) GetSidecarContainerOverrides(debugSymbols bool) fdbv1beta2.ContainerOverrides {
 	sidecarImage, sidecarTag := GetBaseImageAndTag(
 		GetDebugImage(debugSymbols, factory.GetSidecarImage()),
 	)
-	sidecarOverrides := fdbv1beta2.ContainerOverrides{
-		EnableTLS: false,
-		ImageConfigs: []fdbv1beta2.ImageConfig{
+
+	// If the version tag mapping is defined make use of that.
+	imageConfigs := factory.options.getImageVersionConfig(sidecarImage, true)
+	if len(imageConfigs) == 0 {
+		// The first entry is version specific e.g. this image + tag (if specified) will be used for the provided version
+		// the second entry ensures we set the base image for e.g. upgrades independent of the version.
+		imageConfigs = []fdbv1beta2.ImageConfig{
 			{
 				BaseImage: sidecarImage,
 				Tag:       sidecarTag,
 				Version:   factory.GetFDBVersionAsString(),
+				TagSuffix: "-1",
 			},
 			{
 				BaseImage: sidecarImage,
 				TagSuffix: "-1",
 			},
-		},
+		}
 	}
 
-	// If no tag is specified ensure we add the required tag suffix.
-	if sidecarTag == "" {
-		sidecarOverrides.ImageConfigs[0].TagSuffix = "-1"
+	return fdbv1beta2.ContainerOverrides{
+		EnableTLS:    false,
+		ImageConfigs: imageConfigs,
 	}
-
-	return mainOverrides, sidecarOverrides
 }
 
 func (factory *Factory) getClusterName() string {
@@ -776,10 +787,6 @@ func (factory *Factory) OperatorIsAtLeast(version string) bool {
 func (factory *Factory) GetClusterOptions(options ...ClusterOption) []ClusterOption {
 	options = append(options, WithTLSEnabled)
 
-	if factory.options.featureOperatorUnifiedImage {
-		options = append(options, WithUnifiedImage)
-	}
-
 	if factory.options.featureOperatorLocalities {
 		options = append(options, WithLocalitiesForExclusion)
 	}
@@ -856,6 +863,12 @@ func (factory *Factory) GetFoundationDBImage() string {
 	return prependRegistry(factory.options.registry, factory.options.fdbImage)
 }
 
+// GetUnifiedFoundationDBImage returns the unified FoundationDB image provided via command line. If a registry was defined the registry will be
+// prepended.
+func (factory *Factory) GetUnifiedFoundationDBImage() string {
+	return prependRegistry(factory.options.registry, factory.options.unifiedFDBImage)
+}
+
 // getImagePullPolicy returns the image pull policy based on the provided cloud provider. For Kind this will be Never, otherwise
 // this will Always.
 func (factory *Factory) getImagePullPolicy() corev1.PullPolicy {
diff --git a/e2e/fixtures/fdb_cluster_specs.go b/e2e/fixtures/fdb_cluster_specs.go
index fbab45ce..91cb79a0 100644
--- a/e2e/fixtures/fdb_cluster_specs.go
+++ b/e2e/fixtures/fdb_cluster_specs.go
@@ -42,9 +42,11 @@ func (factory *Factory) createFDBClusterSpec(
 	config *ClusterConfig,
 	databaseConfiguration fdbv1beta2.DatabaseConfiguration,
 ) *fdbv1beta2.FoundationDBCluster {
-	mainContainerOverrides, sidecarContainerOverrides := factory.getContainerOverrides(
-		config.DebugSymbols,
-	)
+	useUnifiedImage := pointer.BoolDeref(config.UseUnifiedImage, factory.options.featureOperatorUnifiedImage)
+	imageType := fdbv1beta2.ImageTypeSplit
+	if useUnifiedImage {
+		imageType = fdbv1beta2.ImageTypeUnified
+	}
 
 	return &fdbv1beta2.FoundationDBCluster{
 		ObjectMeta: metav1.ObjectMeta{
@@ -59,8 +61,9 @@ func (factory *Factory) createFDBClusterSpec(
 			StorageServersPerPod:          config.StorageServerPerPod,
 			LogServersPerPod:              config.LogServersPerPod,
 			LogGroup:                      config.Namespace + "-" + config.Name,
-			MainContainer:                 mainContainerOverrides,
-			SidecarContainer:              sidecarContainerOverrides,
+			MainContainer:                 factory.GetMainContainerOverrides(config.DebugSymbols, useUnifiedImage),
+			ImageType:                     &imageType,
+			SidecarContainer:              factory.GetSidecarContainerOverrides(config.DebugSymbols),
 			FaultDomain: fdbv1beta2.FoundationDBClusterFaultDomain{
 				Key: "foundationdb.org/none",
 			},
diff --git a/e2e/fixtures/fdb_data_loader.go b/e2e/fixtures/fdb_data_loader.go
index 4206c91b..eb4ebb90 100644
--- a/e2e/fixtures/fdb_data_loader.go
+++ b/e2e/fixtures/fdb_data_loader.go
@@ -101,7 +101,7 @@ spec:
       initContainers:
         {{ range $index, $version := .SidecarVersions }}
         - name: foundationdb-kubernetes-init-{{ $index }}
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: Always
           command:
             - /bin/bash
@@ -118,7 +118,7 @@ spec:
         # Install this library in a special location to force the operator to use it as the primary library.
         {{ if eq .FDBVersion.Compact "7.1" }}
         - name: foundationdb-kubernetes-init-7-1-primary
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: {{ .ImagePullPolicy }}
           args:
             # Note that we are only copying a library, rather than copying any binaries. 
@@ -242,7 +242,7 @@ spec:
             runAsGroup: 0
 {{ if .CopyAsPrimary }}
         - name: foundationdb-kubernetes-init-cluster-file
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: {{ .ImagePullPolicy }}
           args:
             - --mode
diff --git a/e2e/fixtures/fdb_operator_client.go b/e2e/fixtures/fdb_operator_client.go
index a8f4a247..2b2c4075 100644
--- a/e2e/fixtures/fdb_operator_client.go
+++ b/e2e/fixtures/fdb_operator_client.go
@@ -24,7 +24,6 @@ import (
 	"bytes"
 	ctx "context"
 	"errors"
-	"fmt"
 	"html/template"
 	"io"
 	"log"
@@ -218,7 +217,7 @@ spec:
       initContainers:
         {{ range $index, $version := .SidecarVersions }}
         - name: foundationdb-kubernetes-init-{{ $index }}
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: {{ .ImagePullPolicy }}
           command:
             - /bin/bash
@@ -236,7 +235,7 @@ spec:
         # use it as the primary library.
         {{ if .CopyAsPrimary }}
         - name: foundationdb-kubernetes-init-7-1-primary
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: {{ .ImagePullPolicy }}
           args:
             # Note that we are only copying a library, rather than copying any binaries. 
@@ -357,7 +356,7 @@ spec:
       initContainers:
         {{ range $index, $version := .SidecarVersions }}
         - name: foundationdb-kubernetes-init-{{ $index }}
-          image: {{ .BaseImage }}:{{ .SidecarTag}}
+          image: {{ .Image }}
           imagePullPolicy: {{ .ImagePullPolicy }}
           args:
             - --mode
@@ -502,10 +501,8 @@ type operatorConfig struct {
 
 // SidecarConfig represents the configuration for a sidecar. This can be used for templating.
 type SidecarConfig struct {
-	// BaseImage the image reference without a tag.
-	BaseImage string
-	// SidecarTag represents the image tag for this configuration.
-	SidecarTag string
+	// Image the image reference with the tag.
+	Image string
 	// FDBVersion represents the FoundationDB version for this config.
 	FDBVersion fdbv1beta2.Version
 	// ImagePullPolicy represents the pull policy for the sidecar.
@@ -514,22 +511,18 @@ type SidecarConfig struct {
 	CopyAsPrimary bool
 }
 
-// getSidecarConfigsSplitImage returns the sidecar config for the split image configuration.
-func (factory *Factory) getSidecarConfigsSplitImage() []SidecarConfig {
+// getSidecarConfigs returns the sidecar config based on the provided imageConfigs.
+func (factory *Factory) getSidecarConfigs(imageConfigs []fdbv1beta2.ImageConfig) []SidecarConfig {
 	var hasCopyPrimarySet bool
 	additionalSidecarVersions := factory.GetAdditionalSidecarVersions()
 	sidecarConfigs := make([]SidecarConfig, 0, len(additionalSidecarVersions)+1)
-	baseImage, tag := GetBaseImageAndTag(factory.GetSidecarImage())
-	if tag == "" {
-		tag = fmt.Sprintf("%s-1", factory.GetFDBVersion())
-	}
+	pullPolicy := factory.getImagePullPolicy()
 
-	defaultConfig := getDefaultSidecarConfig(
-		baseImage,
-		tag,
-		factory.GetFDBVersion(),
-		factory.getImagePullPolicy(),
-	)
+	defaultConfig := SidecarConfig{
+		Image:           fdbv1beta2.SelectImageConfig(imageConfigs, factory.GetFDBVersionAsString()).Image(),
+		FDBVersion:      factory.GetFDBVersion(),
+		ImagePullPolicy: pullPolicy,
+	}
 
 	if factory.GetFDBVersion().SupportsDNSInClusterFile() {
 		defaultConfig.CopyAsPrimary = true
@@ -543,72 +536,17 @@ func (factory *Factory) getSidecarConfigsSplitImage() []SidecarConfig {
 
 	// Add all other versions that are required e.g. for major or minor upgrades.
 	for _, version := range additionalSidecarVersions {
-		// Don't add the sidecar another time if we already added it
-		if version.Equal(factory.GetFDBVersion()) {
+		// Don't add the sidecar another time if we already added a protocol compatible version.
+		if version.IsProtocolCompatible(factory.GetFDBVersion()) {
 			continue
 		}
 
-		sidecarConfig := getSidecarConfig(baseImage, fmt.Sprintf("%s-1", version), version, factory.getImagePullPolicy())
-		if !hasCopyPrimarySet && version.SupportsDNSInClusterFile() {
-			sidecarConfig.CopyAsPrimary = true
-			hasCopyPrimarySet = true
-		}
-
-		sidecarConfigs = append(
-			sidecarConfigs,
-			sidecarConfig,
-		)
-	}
-
-	return sidecarConfigs
-}
-
-func getTagFromImageConfig(imageConfigs []fdbv1beta2.ImageConfig, version fdbv1beta2.Version) string {
-	for _, imageConfig := range imageConfigs {
-		if imageConfig.Version == version.String() {
-			return imageConfig.Tag
+		sidecarConfig := SidecarConfig{
+			Image:           fdbv1beta2.SelectImageConfig(imageConfigs, version.String()).Image(),
+			FDBVersion:      version,
+			ImagePullPolicy: pullPolicy,
 		}
-	}
 
-	return version.String()
-}
-
-// getSidecarConfigsSplitImage returns the sidecar config for the split image configuration.
-func (factory *Factory) getSidecarConfigsUnifiedImage() []SidecarConfig {
-	var hasCopyPrimarySet bool
-	additionalSidecarVersions := factory.GetAdditionalSidecarVersions()
-	sidecarConfigs := make([]SidecarConfig, 0, len(additionalSidecarVersions)+1)
-	baseImage, tag := GetBaseImageAndTag(factory.GetFoundationDBImage())
-	imageConfigs := factory.options.getImageVersionConfig(baseImage)
-	if tag == "" || tag == factory.GetFDBVersionAsString() {
-		tag = getTagFromImageConfig(imageConfigs, factory.GetFDBVersion())
-	}
-
-	defaultConfig := getDefaultSidecarConfig(
-		baseImage,
-		tag,
-		factory.GetFDBVersion(),
-		factory.getImagePullPolicy(),
-	)
-
-	if factory.GetFDBVersion().SupportsDNSInClusterFile() {
-		defaultConfig.CopyAsPrimary = true
-		hasCopyPrimarySet = true
-	}
-
-	sidecarConfigs = append(
-		sidecarConfigs,
-		defaultConfig,
-	)
-
-	// Add all other versions that are required e.g. for major or minor upgrades.
-	for _, version := range additionalSidecarVersions {
-		// Don't add the sidecar another time if we already added it
-		if version.Equal(factory.GetFDBVersion()) {
-			continue
-		}
-
-		sidecarConfig := getSidecarConfig(baseImage, getTagFromImageConfig(imageConfigs, version), version, factory.getImagePullPolicy())
 		if !hasCopyPrimarySet && version.SupportsDNSInClusterFile() {
 			sidecarConfig.CopyAsPrimary = true
 			hasCopyPrimarySet = true
@@ -627,23 +565,10 @@ func (factory *Factory) getSidecarConfigsUnifiedImage() []SidecarConfig {
 // all provided sidecar versions to inject FDB client libraries.
 func (factory *Factory) GetSidecarConfigs() []SidecarConfig {
 	if factory.options.featureOperatorUnifiedImage {
-		return factory.getSidecarConfigsUnifiedImage()
+		return factory.getSidecarConfigs(factory.GetMainContainerOverrides(false, true).ImageConfigs)
 	}
 
-	return factory.getSidecarConfigsSplitImage()
-}
-
-func getDefaultSidecarConfig(sidecarImage string, tag string, version fdbv1beta2.Version, imagePullPolicy corev1.PullPolicy) SidecarConfig {
-	return getSidecarConfig(sidecarImage, tag, version, imagePullPolicy)
-}
-
-func getSidecarConfig(baseImage string, tag string, version fdbv1beta2.Version, imagePullPolicy corev1.PullPolicy) SidecarConfig {
-	return SidecarConfig{
-		BaseImage:       baseImage,
-		FDBVersion:      version,
-		SidecarTag:      tag,
-		ImagePullPolicy: imagePullPolicy,
-	}
+	return factory.getSidecarConfigs(factory.GetSidecarContainerOverrides(false).ImageConfigs)
 }
 
 //nolint:revive
diff --git a/e2e/fixtures/fdb_operator_fixtures.go b/e2e/fixtures/fdb_operator_fixtures.go
index 7fd2f4b8..e4194127 100644
--- a/e2e/fixtures/fdb_operator_fixtures.go
+++ b/e2e/fixtures/fdb_operator_fixtures.go
@@ -264,9 +264,3 @@ func WithOneMinuteMinimumUptimeSecondsForBounce(
 func WithLocalitiesForExclusion(_ *Factory, cluster *fdbv1beta2.FoundationDBCluster) {
 	cluster.Spec.AutomationOptions.UseLocalitiesForExclusion = pointer.Bool(true)
 }
-
-// WithUnifiedImage is an option that enables the unified image for a cluster.
-func WithUnifiedImage(_ *Factory, cluster *fdbv1beta2.FoundationDBCluster) {
-	imageType := fdbv1beta2.ImageTypeUnified
-	cluster.Spec.ImageType = &imageType
-}
diff --git a/e2e/fixtures/options.go b/e2e/fixtures/options.go
index 475e70ec..948e2c3e 100644
--- a/e2e/fixtures/options.go
+++ b/e2e/fixtures/options.go
@@ -34,8 +34,9 @@ type FactoryOptions struct {
 	namespace                   string
 	chaosNamespace              string
 	context                     string
-	fdbImage                    string // TODO (johscheuer): Make this optional if we use the default
-	sidecarImage                string // TODO (johscheuer): Make this optional if we use the default
+	fdbImage                    string
+	unifiedFDBImage             string
+	sidecarImage                string
 	operatorImage               string
 	dataLoaderImage             string
 	registry                    string
@@ -80,19 +81,25 @@ func (options *FactoryOptions) BindFlags(fs *flag.FlagSet) {
 	fs.StringVar(
 		&options.fdbImage,
 		"fdb-image",
-		"",
+		"foundationdb/foundationdb",
 		"defines the FoundationDB image that should be used for testing",
 	)
+	fs.StringVar(
+		&options.unifiedFDBImage,
+		"unified-fdb-image",
+		"foundationdb/fdb-kubernetes-monitor",
+		"defines the unified FoundationDB image that should be used for testing",
+	)
 	fs.StringVar(
 		&options.sidecarImage,
 		"sidecar-image",
-		"",
+		"foundationdb/foundationdb-kubernetes-sidecar",
 		"defines the FoundationDB sidecar image that should be used for testing",
 	)
 	fs.StringVar(
 		&options.operatorImage,
 		"operator-image",
-		"",
+		"foundationdb/fdb-kubernetes-operator",
 		"defines the Kubernetes Operator image that should be used for testing",
 	)
 	fs.StringVar(
@@ -300,7 +307,7 @@ func (options *FactoryOptions) validateFDBVersionTagMapping() error {
 	return nil
 }
 
-func (options *FactoryOptions) getImageVersionConfig(baseImage string) []fdbv1beta2.ImageConfig {
+func (options *FactoryOptions) getImageVersionConfig(baseImage string, isSidecar bool) []fdbv1beta2.ImageConfig {
 	if options.fdbVersionTagMapping == "" {
 		return nil
 	}
@@ -310,10 +317,15 @@ func (options *FactoryOptions) getImageVersionConfig(baseImage string) []fdbv1be
 
 	for idx, mapping := range mappings {
 		versionMapping := strings.Split(mapping, ":")
+		tag := strings.TrimSpace(versionMapping[1])
+		// The sidecar requires the -1 prefix.
+		if isSidecar {
+			tag = tag + "-1"
+		}
 		imageConfig[idx] = fdbv1beta2.ImageConfig{
 			BaseImage: baseImage,
 			Version:   strings.TrimSpace(versionMapping[0]),
-			Tag:       strings.TrimSpace(versionMapping[1]),
+			Tag:       tag,
 		}
 	}
 
diff --git a/e2e/test_operator/operator_test.go b/e2e/test_operator/operator_test.go
index eec6355b..fb54b0f6 100644
--- a/e2e/test_operator/operator_test.go
+++ b/e2e/test_operator/operator_test.go
@@ -82,7 +82,7 @@ var _ = BeforeSuite(func() {
 	)
 
 	//Load some data async into the cluster. We will only block as long as the Job is created.
-	// factory.CreateDataLoaderIfAbsent(fdbCluster)
+	factory.CreateDataLoaderIfAbsent(fdbCluster)
 
 	// In order to test the robustness of the operator we try to kill the operator Pods every minute.
 	if factory.ChaosTestsEnabled() {
diff --git a/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go b/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go
new file mode 100644
index 00000000..5ecdd441
--- /dev/null
+++ b/e2e/test_operator_migrate_image_type/operator_migate_image_type_test.go
@@ -0,0 +1,161 @@
+/*
+ * operator_migrate_image_type_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2023-2024 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package operator
+
+/*
+
+This test suite includes tests for migrating between the different image types.
+*/
+
+import (
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	"k8s.io/utils/pointer"
+	"log"
+)
+
+var (
+	factory     *fixtures.Factory
+	fdbCluster  *fixtures.FdbCluster
+	testOptions *fixtures.FactoryOptions
+)
+
+func init() {
+	testOptions = fixtures.InitFlags()
+}
+
+var _ = BeforeSuite(func() {
+	factory = fixtures.CreateFactory(testOptions)
+})
+
+var _ = AfterSuite(func() {
+	if CurrentSpecReport().Failed() {
+		log.Printf("failed due to %s", CurrentSpecReport().FailureMessage())
+	}
+	factory.Shutdown()
+})
+
+var _ = PDescribe("Operator Migrate Image Type", Label("e2e"), func() {
+	When("migrating from split to unified", func() {
+		BeforeEach(func() {
+			config := fixtures.DefaultClusterConfig(false)
+			config.UseUnifiedImage = pointer.Bool(false)
+			fdbCluster = factory.CreateFdbCluster(
+				config,
+				factory.GetClusterOptions()...,
+			)
+
+			//Load some data async into the cluster. We will only block as long as the Job is created.
+			factory.CreateDataLoaderIfAbsent(fdbCluster)
+
+			// Update the cluster spec to run with the unified image.
+			spec := fdbCluster.GetCluster().Spec.DeepCopy()
+			imageType := fdbv1beta2.ImageTypeUnified
+			spec.ImageType = &imageType
+			// Generate the new config to make use of the unified images.
+			overrides := factory.GetMainContainerOverrides(false, true)
+			overrides.EnableTLS = spec.MainContainer.EnableTLS
+			spec.MainContainer = overrides
+			fdbCluster.UpdateClusterSpecWithSpec(spec)
+			Expect(fdbCluster.WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			Expect(fdbCluster.Destroy()).NotTo(HaveOccurred())
+		})
+
+		It("should convert the cluster", func() {
+			// Make sure we didn't lose data.
+			fdbCluster.EnsureTeamTrackersAreHealthy()
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
+
+			unifiedImage := factory.GetUnifiedFoundationDBImage()
+			pods := fdbCluster.GetPods()
+			for _, pod := range pods.Items {
+				// Ignore Pods that are pending the deletion.
+				if !pod.DeletionTimestamp.IsZero() {
+					continue
+				}
+
+				// With the unified image no init containers are used.
+				Expect(pod.Spec.InitContainers).To(HaveLen(0))
+				// Make sure they run the unified image.
+				Expect(pod.Spec.Containers[0].Image).To(ContainSubstring(unifiedImage))
+			}
+		})
+	})
+
+	When("migrating from unified to split", func() {
+		BeforeEach(func() {
+			config := fixtures.DefaultClusterConfig(false)
+			config.UseUnifiedImage = pointer.Bool(true)
+			fdbCluster = factory.CreateFdbCluster(
+				config,
+				factory.GetClusterOptions()...,
+			)
+
+			//Load some data async into the cluster. We will only block as long as the Job is created.
+			factory.CreateDataLoaderIfAbsent(fdbCluster)
+
+			// Update the cluster spec to run with the split image.
+			spec := fdbCluster.GetCluster().Spec.DeepCopy()
+			imageType := fdbv1beta2.ImageTypeSplit
+			spec.ImageType = &imageType
+			// Generate the new config to make use of the split images.
+			overrides := factory.GetMainContainerOverrides(false, false)
+			overrides.EnableTLS = spec.MainContainer.EnableTLS
+			spec.MainContainer = overrides
+			fdbCluster.UpdateClusterSpecWithSpec(spec)
+			Expect(fdbCluster.WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		AfterEach(func() {
+			Expect(fdbCluster.Destroy()).NotTo(HaveOccurred())
+		})
+
+		It("should convert the cluster", func() {
+			// Make sure we didn't lose data.
+			fdbCluster.EnsureTeamTrackersAreHealthy()
+			fdbCluster.EnsureTeamTrackersHaveMinReplicas()
+
+			fdbImage := factory.GetFoundationDBImage()
+			sidecarImage := factory.GetSidecarImage()
+			pods := fdbCluster.GetPods()
+			for _, pod := range pods.Items {
+				// Ignore Pods that are pending the deletion.
+				if !pod.DeletionTimestamp.IsZero() {
+					continue
+				}
+				Expect(pod.Spec.InitContainers).NotTo(HaveLen(0))
+				// Make sure they run the split image.
+				for _, container := range pod.Spec.Containers {
+					if container.Name == fdbv1beta2.MainContainerName {
+						Expect(container.Image).To(ContainSubstring(fdbImage))
+					} else {
+						Expect(container.Image).To(ContainSubstring(sidecarImage))
+					}
+				}
+			}
+		})
+	})
+})
diff --git a/e2e/test_operator_migrate_image_type/suite_test.go b/e2e/test_operator_migrate_image_type/suite_test.go
new file mode 100644
index 00000000..a4449220
--- /dev/null
+++ b/e2e/test_operator_migrate_image_type/suite_test.go
@@ -0,0 +1,34 @@
+/*
+ * suite_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package operator
+
+import (
+	"testing"
+	"time"
+
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	"github.com/onsi/gomega"
+)
+
+func TestOperator(t *testing.T) {
+	gomega.SetDefaultEventuallyTimeout(10 * time.Second)
+	fixtures.RunGinkgoTests(t, "Operator Migrate Image Type test suite")
+}
diff --git a/internal/locality/locality_test.go b/internal/locality/locality_test.go
index 046a7c45..08cb10bc 100644
--- a/internal/locality/locality_test.go
+++ b/internal/locality/locality_test.go
@@ -23,14 +23,17 @@ package locality
 import (
 	"fmt"
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podclient/mock"
-	"math"
-	"net"
-	"strings"
-
 	"github.com/go-logr/logr"
+	"github.com/onsi/gomega/gmeasure"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"math"
+	"math/rand"
+	"net"
+	"strconv"
+	"strings"
+	"time"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
@@ -116,6 +119,28 @@ func generateDefaultStatus(tls bool) *fdbv1beta2.FoundationDBStatus {
 	}
 }
 
+func generateCandidates(dcIDs []string, processesPerDc int, numberOfZones int) []Info {
+	candidates := make([]Info, processesPerDc*len(dcIDs))
+
+	idx := 0
+	for _, dcID := range dcIDs {
+		for i := 0; i < processesPerDc; i++ {
+			zoneIdx := i % numberOfZones
+			candidates[idx] = Info{
+				ID: dcID + strconv.Itoa(i),
+				LocalityData: map[string]string{
+					fdbv1beta2.FDBLocalityZoneIDKey: dcID + "-z" + strconv.Itoa(zoneIdx),
+					fdbv1beta2.FDBLocalityDCIDKey:   dcID,
+				},
+			}
+
+			idx++
+		}
+	}
+
+	return candidates
+}
+
 var _ = Describe("Localities", func() {
 	var cluster *fdbv1beta2.FoundationDBCluster
 
@@ -138,105 +163,260 @@ var _ = Describe("Localities", func() {
 	When("Sorting the localities", func() {
 		var localities []Info
 
-		BeforeEach(func() {
-			localities = []Info{
-				{
-					ID:    "storage-1",
-					Class: fdbv1beta2.ProcessClassStorage,
-				},
-				{
-					ID:    "tlog-1",
-					Class: fdbv1beta2.ProcessClassTransaction,
-				},
-				{
-					ID:    "log-1",
-					Class: fdbv1beta2.ProcessClassLog,
-				},
-				{
-					ID:    "storage-51",
-					Class: fdbv1beta2.ProcessClassStorage,
-				},
-			}
-		})
-
 		JustBeforeEach(func() {
 			for idx, cur := range localities {
+				if cur.LocalityData == nil {
+					cur.LocalityData = map[string]string{}
+				}
+
+				cur.LocalityData[fdbv1beta2.FDBLocalityZoneIDKey] = cur.ID
 				localities[idx] = Info{
-					ID:       cur.ID,
-					Class:    cur.Class,
-					Priority: cluster.GetClassCandidatePriority(cur.Class),
+					ID:           cur.ID,
+					Class:        cur.Class,
+					Priority:     cluster.GetClassCandidatePriority(cur.Class),
+					LocalityData: cur.LocalityData,
 				}
 			}
 		})
 
-		When("no other preferences are defined", func() {
+		When("only a single DC is used", func() {
 			BeforeEach(func() {
-				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{}
+				localities = []Info{
+					{
+						ID:    "storage-1",
+						Class: fdbv1beta2.ProcessClassStorage,
+					},
+					{
+						ID:    "tlog-1",
+						Class: fdbv1beta2.ProcessClassTransaction,
+					},
+					{
+						ID:    "log-1",
+						Class: fdbv1beta2.ProcessClassLog,
+					},
+					{
+						ID:    "storage-51",
+						Class: fdbv1beta2.ProcessClassStorage,
+					},
+				}
 			})
 
-			It("should sort the localities based on the IDs but prefer transaction system Pods", func() {
-				sortLocalities(localities)
-
-				Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassLog))
-				Expect(localities[0].ID).To(Equal("log-1"))
-				Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
-				Expect(localities[1].ID).To(Equal("tlog-1"))
-				Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[2].ID).To(Equal("storage-1"))
-				Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[3].ID).To(Equal("storage-51"))
+			When("no other preferences are defined", func() {
+				BeforeEach(func() {
+					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{}
+				})
+
+				It("should sort the localities based on the IDs but prefer transaction system Pods", func() {
+					sortLocalities("", localities)
+
+					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[0].ID).To(Equal("log-1"))
+					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
+					Expect(localities[1].ID).To(Equal("tlog-1"))
+					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[2].ID).To(Equal("storage-1"))
+					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[3].ID).To(Equal("storage-51"))
+				})
 			})
-		})
 
-		When("when the storage class is preferred", func() {
-			BeforeEach(func() {
-				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
-					{
-						ProcessClass: fdbv1beta2.ProcessClassStorage,
-						Priority:     10,
-					},
-				}
+			When("when the storage class is preferred", func() {
+				BeforeEach(func() {
+					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+						{
+							ProcessClass: fdbv1beta2.ProcessClassStorage,
+							Priority:     10,
+						},
+					}
+				})
+
+				It("should sort the localities based on the provided config", func() {
+					sortLocalities("", localities)
+
+					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[0].ID).To(Equal("storage-1"))
+					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[1].ID).To(Equal("storage-51"))
+					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[2].ID).To(Equal("log-1"))
+					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
+					Expect(localities[3].ID).To(Equal("tlog-1"))
+				})
 			})
 
-			It("should sort the localities based on the provided config", func() {
-				sortLocalities(localities)
-
-				Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[0].ID).To(Equal("storage-1"))
-				Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[1].ID).To(Equal("storage-51"))
-				Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassLog))
-				Expect(localities[2].ID).To(Equal("log-1"))
-				Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
-				Expect(localities[3].ID).To(Equal("tlog-1"))
+			When("when the storage class is preferred over transaction class", func() {
+				BeforeEach(func() {
+					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+						{
+							ProcessClass: fdbv1beta2.ProcessClassStorage,
+							Priority:     10,
+						},
+						{
+							ProcessClass: fdbv1beta2.ProcessClassTransaction,
+							Priority:     1,
+						},
+					}
+				})
+
+				It("should sort the localities based on the provided config", func() {
+					sortLocalities("", localities)
+
+					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[0].ID).To(Equal("storage-1"))
+					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[1].ID).To(Equal("storage-51"))
+					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[2].ID).To(Equal("log-1"))
+					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
+					Expect(localities[3].ID).To(Equal("tlog-1"))
+				})
 			})
 		})
 
-		When("when the storage class is preferred over transaction class", func() {
+		When("a multi-region setup is used", func() {
+			var primaryID, remoteID, primarySatelliteID, remoteSatelliteID string
+
 			BeforeEach(func() {
-				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
-					{
-						ProcessClass: fdbv1beta2.ProcessClassStorage,
-						Priority:     10,
-					},
-					{
-						ProcessClass: fdbv1beta2.ProcessClassTransaction,
-						Priority:     1,
+				// Generate random names to make sure we test different alphabetical orderings.
+				primaryID = internal.GenerateRandomString(10)
+				remoteID = internal.GenerateRandomString(10)
+				primarySatelliteID = internal.GenerateRandomString(10)
+				remoteSatelliteID = internal.GenerateRandomString(10)
+
+				localities = make([]Info, 12)
+				idx := 0
+				// Make sure the result is randomized
+				dcIDs := []string{primaryID, primarySatelliteID, remoteID, remoteSatelliteID}
+				rand.Shuffle(len(dcIDs), func(i, j int) {
+					dcIDs[i], dcIDs[j] = dcIDs[j], dcIDs[i]
+				})
+
+				for _, dcID := range dcIDs {
+					for i := 0; i < 3; i++ {
+						pClass := fdbv1beta2.ProcessClassStorage
+						// The first locality will be a log process
+						if i == 0 {
+							pClass = fdbv1beta2.ProcessClassLog
+						}
+
+						localities[idx] = Info{
+							ID: dcID + strconv.Itoa(i),
+							LocalityData: map[string]string{
+								fdbv1beta2.FDBLocalityZoneIDKey: dcID + "-z" + strconv.Itoa(i),
+								fdbv1beta2.FDBLocalityDCIDKey:   dcID,
+							},
+							Class: pClass,
+						}
+
+						idx++
+					}
+				}
+
+				// Randomize the order of localities.
+				rand.Shuffle(len(localities), func(i, j int) {
+					localities[i], localities[j] = localities[j], localities[i]
+				})
+
+				cluster.Spec.DatabaseConfiguration = fdbv1beta2.DatabaseConfiguration{
+					UsableRegions: 2,
+					Regions: []fdbv1beta2.Region{
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       primaryID,
+									Priority: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       remoteID,
+									Priority: 0,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
 					},
 				}
 			})
 
-			It("should sort the localities based on the provided config", func() {
-				sortLocalities(localities)
-
-				Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[0].ID).To(Equal("storage-1"))
-				Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-				Expect(localities[1].ID).To(Equal("storage-51"))
-				Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassLog))
-				Expect(localities[2].ID).To(Equal("log-1"))
-				Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
-				Expect(localities[3].ID).To(Equal("tlog-1"))
+			JustBeforeEach(func() {
+				sortLocalities(primaryID, localities)
+			})
+
+			When("no other preferences are defined", func() {
+				BeforeEach(func() {
+					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{}
+				})
+
+				It("should sort the localities based on the IDs but prefer transaction system Pods", func() {
+					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[0].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[1].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[2].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					// The rest of the clusters are ordered by priority and the process ID.
+					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[3].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[3].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+					Expect(localities[4].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[4].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[4].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+					Expect(localities[5].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[5].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[5].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+				})
+			})
+
+			When("when the storage class is preferred", func() {
+				BeforeEach(func() {
+					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+						{
+							ProcessClass: fdbv1beta2.ProcessClassStorage,
+							Priority:     10,
+						},
+					}
+				})
+
+				It("should sort the localities based on the provided config", func() {
+					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[0].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[1].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+					Expect(localities[2].LocalityData).To(HaveKeyWithValue(fdbv1beta2.FDBLocalityDCIDKey, primaryID))
+					// The rest of the clusters are ordered by priority and the process ID.
+					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[3].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[3].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+					Expect(localities[4].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[4].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[4].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+					Expect(localities[5].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+					Expect(localities[5].LocalityData).To(HaveKey(fdbv1beta2.FDBLocalityDCIDKey))
+					Expect(localities[5].LocalityData[fdbv1beta2.FDBLocalityDCIDKey]).NotTo(Equal(primaryID))
+				})
 			})
 		})
 	})
@@ -311,16 +491,16 @@ var _ = Describe("Localities", func() {
 		Context("with multiple data centers", func() {
 			BeforeEach(func() {
 				candidates = []Info{
-					{ID: "p1", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1", "dcid": "dc1"}},
-					{ID: "p2", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1", "dcid": "dc1"}},
-					{ID: "p3", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2", "dcid": "dc1"}},
-					{ID: "p4", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z3", "dcid": "dc1"}},
-					{ID: "p5", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2", "dcid": "dc1"}},
-					{ID: "p6", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z4", "dcid": "dc1"}},
-					{ID: "p7", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z5", "dcid": "dc1"}},
-					{ID: "p8", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z6", "dcid": "dc2"}},
-					{ID: "p9", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z7", "dcid": "dc2"}},
-					{ID: "p10", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z8", "dcid": "dc2"}},
+					{ID: "p1", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p2", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p3", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p4", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z3", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p5", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p6", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z4", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p7", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z5", fdbv1beta2.FDBLocalityDCIDKey: "dc1"}},
+					{ID: "p8", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z6", fdbv1beta2.FDBLocalityDCIDKey: "dc2"}},
+					{ID: "p9", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z7", fdbv1beta2.FDBLocalityDCIDKey: "dc2"}},
+					{ID: "p10", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z8", fdbv1beta2.FDBLocalityDCIDKey: "dc2"}},
 				}
 			})
 
@@ -358,6 +538,257 @@ var _ = Describe("Localities", func() {
 				})
 			})
 		})
+
+		When("a multi-region cluster is used with 4 dcs", func() {
+			var primaryID, remoteID, primarySatelliteID, remoteSatelliteID string
+			var dcIDs []string
+
+			BeforeEach(func() {
+				// Generate random names to make sure we test different alphabetical orderings.
+				primaryID = internal.GenerateRandomString(10)
+				remoteID = internal.GenerateRandomString(10)
+				primarySatelliteID = internal.GenerateRandomString(10)
+				remoteSatelliteID = internal.GenerateRandomString(10)
+
+				// Make sure the result is randomized
+				dcIDs = []string{primaryID, primarySatelliteID, remoteID, remoteSatelliteID}
+				rand.Shuffle(len(dcIDs), func(i, j int) {
+					dcIDs[i], dcIDs[j] = dcIDs[j], dcIDs[i]
+				})
+
+				cluster.Spec.DatabaseConfiguration = fdbv1beta2.DatabaseConfiguration{
+					UsableRegions: 2,
+					Regions: []fdbv1beta2.Region{
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       primaryID,
+									Priority: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       remoteID,
+									Priority: 0,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
+					},
+				}
+			})
+
+			When("testing the correct behavior with a small set of candidates", func() {
+				BeforeEach(func() {
+					candidates = generateCandidates(dcIDs, 5, 5)
+					result, err = ChooseDistributedProcesses(cluster, candidates, cluster.DesiredCoordinatorCount(), ProcessSelectionConstraint{
+						HardLimits:            GetHardLimits(cluster),
+						SelectingCoordinators: true,
+					})
+					Expect(err).NotTo(HaveOccurred())
+				})
+
+				It("should recruit the processes across multiple dcs and prefer the primary", func() {
+					Expect(len(result)).To(Equal(9))
+
+					dcCount := map[string]int{}
+					for _, process := range result {
+						dcCount[process.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
+					}
+
+					Expect(dcCount).To(Equal(map[string]int{
+						primaryID:          3,
+						remoteID:           2,
+						remoteSatelliteID:  2,
+						primarySatelliteID: 2,
+					}))
+				})
+			})
+
+			// Adding a benchmark test for the ChooseDistributedProcesses. In order to print the set use FIt.
+			When("measuring the performance for", func() {
+				It("choose distributed processes", Serial, Label("measurement"), func() {
+					// Create the new experiment.
+					experiment := gmeasure.NewExperiment("Choose Distributed Processes")
+
+					// Register the experiment as a ReportEntry - this will cause Ginkgo's reporter infrastructure
+					// to print out the experiment's report and to include the experiment in any generated reports.
+					AddReportEntry(experiment.Name, experiment)
+
+					// We sample a function repeatedly to get a statistically significant set of measurements.
+					experiment.Sample(func(_ int) {
+						candidates = generateCandidates(dcIDs, 250, 10)
+						rand.Shuffle(len(candidates), func(i, j int) {
+							candidates[i], candidates[j] = candidates[j], candidates[i]
+						})
+						// Only measure the actual execution of ChooseDistributedProcesses.
+						experiment.MeasureDuration("ChooseDistributedProcesses", func() {
+							_, _ = ChooseDistributedProcesses(cluster, candidates, cluster.DesiredCoordinatorCount(), ProcessSelectionConstraint{
+								HardLimits:            GetHardLimits(cluster),
+								SelectingCoordinators: true,
+							})
+						})
+						// We'll sample the function up to 50 times or up to a minute, whichever comes first.
+					}, gmeasure.SamplingConfig{N: 50, Duration: time.Minute})
+				})
+			})
+		})
+
+		When("a multi-region cluster is used with 4 dcs and storage processes should be preferred", func() {
+			var primaryID, remoteID, primarySatelliteID, remoteSatelliteID string
+
+			BeforeEach(func() {
+				// Generate random names to make sure we test different alphabetical orderings.
+				primaryID = internal.GenerateRandomString(10)
+				remoteID = internal.GenerateRandomString(10)
+				primarySatelliteID = internal.GenerateRandomString(10)
+				remoteSatelliteID = internal.GenerateRandomString(10)
+
+				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+					{
+						ProcessClass: fdbv1beta2.ProcessClassStorage,
+						Priority:     math.MaxInt32,
+					},
+					{
+						ProcessClass: fdbv1beta2.ProcessClassLog,
+						Priority:     0,
+					},
+				}
+				candidates = make([]Info, 18)
+				idx := 0
+				for _, dcID := range []string{primaryID, primarySatelliteID, remoteID, remoteSatelliteID} {
+					for i := 0; i < 3; i++ {
+						candidates[idx] = Info{
+							ID:    strconv.Itoa(idx),
+							Class: fdbv1beta2.ProcessClassLog,
+							LocalityData: map[string]string{
+								fdbv1beta2.FDBLocalityZoneIDKey: dcID + "-z1" + strconv.Itoa(i),
+								fdbv1beta2.FDBLocalityDCIDKey:   dcID,
+							},
+							Priority: cluster.GetClassCandidatePriority(fdbv1beta2.ProcessClassLog),
+						}
+
+						idx++
+					}
+
+					if dcID == primarySatelliteID || dcID == remoteSatelliteID {
+						continue
+					}
+
+					for i := 0; i < 3; i++ {
+						candidates[idx] = Info{
+							ID:    strconv.Itoa(idx),
+							Class: fdbv1beta2.ProcessClassStorage,
+							LocalityData: map[string]string{
+								fdbv1beta2.FDBLocalityZoneIDKey: dcID + "-z2" + strconv.Itoa(i),
+								fdbv1beta2.FDBLocalityDCIDKey:   dcID,
+							},
+							Priority: cluster.GetClassCandidatePriority(fdbv1beta2.ProcessClassStorage),
+						}
+
+						idx++
+					}
+				}
+
+				cluster.Spec.DatabaseConfiguration = fdbv1beta2.DatabaseConfiguration{
+					UsableRegions: 2,
+					Regions: []fdbv1beta2.Region{
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       primaryID,
+									Priority: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
+						{
+							DataCenters: []fdbv1beta2.DataCenter{
+								{
+									ID:       remoteID,
+									Priority: 0,
+								},
+								{
+									ID:        remoteSatelliteID,
+									Priority:  1,
+									Satellite: 1,
+								},
+								{
+									ID:        primarySatelliteID,
+									Priority:  0,
+									Satellite: 1,
+								},
+							},
+						},
+					},
+				}
+
+				result, err = ChooseDistributedProcesses(cluster, candidates, cluster.DesiredCoordinatorCount(), ProcessSelectionConstraint{
+					HardLimits:            GetHardLimits(cluster),
+					SelectingCoordinators: true,
+				})
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			It("should recruit the processes across multiple dcs and prefer the primary", func() {
+				Expect(len(result)).To(Equal(9))
+				var storageCnt int
+				var logCnt int
+
+				dcCount := map[string]int{}
+				for _, process := range result {
+					dcCount[process.LocalityData[fdbv1beta2.FDBLocalityDCIDKey]]++
+					if process.Class == fdbv1beta2.ProcessClassLog {
+						logCnt++
+						continue
+					}
+
+					storageCnt++
+				}
+
+				Expect(dcCount).To(Equal(map[string]int{
+					primaryID:          3,
+					remoteID:           2,
+					remoteSatelliteID:  2,
+					primarySatelliteID: 2,
+				}))
+
+				// 5 storage processes should be recruited in the main and remote dc
+				Expect(storageCnt).To(BeNumerically("==", 5))
+				// the satellites only have logs, so only logs are recruited
+				Expect(logCnt).To(BeNumerically("==", 4))
+			})
+		})
 	})
 
 	DescribeTable("when getting the hard limits", func(cluster *fdbv1beta2.FoundationDBCluster, expected map[string]int) {
diff --git a/internal/test_helper.go b/internal/test_helper.go
index e4df4890..077ba132 100644
--- a/internal/test_helper.go
+++ b/internal/test_helper.go
@@ -25,6 +25,8 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/pointer"
+	"math/rand"
+	"strings"
 )
 
 // CreateDefaultCluster creates a default FoundationDBCluster for testing
@@ -104,3 +106,15 @@ func GetProcessGroup(cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1
 		ProcessGroupID: processGroupID,
 	}
 }
+
+const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+// GenerateRandomString can be used to generate a random string with length n
+func GenerateRandomString(n int) string {
+	var res strings.Builder
+	for i := 0; i < n; i++ {
+		res.WriteByte(letterBytes[rand.Intn(len(letterBytes))])
+	}
+
+	return res.String()
+}