Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1435 (2022-11-17 16:41:38)

Add multi-cluster kind setup for testing
# Description

This is the initial setup script for running e2e tests across multiple kind clusters.

## Type of change

*Please select one of the options below.*

- New feature (non-breaking change which adds functionality)

## Discussion

*Are there any design details that you would like to discuss further?*

## Testing

Locally created the setup: 

```bash
$ kind get clusters
cluster1
cluster2
cluster3
cluster4
```

Tested with the multi-dc setup:

```bash
$ kubectl --kubeconfig $KUBECONFIG1 exec -ti test-cluster-dc1-storage-3  -- bash
[root@test-cluster-dc1-storage-3 /]# fdbcli
Using cluster file `/var/dynamic-conf/fdb.cluster'.

The database is available.

Welcome to the fdbcli. For help, type `help'.
fdb> status

Using cluster file `/var/dynamic-conf/fdb.cluster'.

Configuration:
  Redundancy mode        - double
  Storage engine         - ssd-2
  Coordinators           - 9
  Desired Commit Proxies - 2
  Desired GRV Proxies    - 1
  Desired Resolvers      - 1
  Desired Logs           - 3
  Desired Remote Logs    - 3
  Desired Log Routers    - 3
  Usable Regions         - 2
  Regions:
    Primary -
        Datacenter                    - dc1
        Satellite datacenters         - dc3, dc2
        Satellite Logs                - 3
    Remote -
        Datacenter                    - dc2
        Satellite datacenters         - dc3, dc1
        Satellite Logs                - 3

Cluster:
  FoundationDB processes - 10
  Zones                  - 10
  Machines               - 10
  Memory availability    - 8.0 GB per process on machine with least available
  Fault Tolerance        - 1 machines
  Server time            - 11/21/22 14:10:01
```

I will do some more testing and bring up an FDB cluster in that setup.

## Documentation

I will add the required documentation before merging.

## Follow-up

Create the e2e tests :)


foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: 431a37b9b710af7ae0d427b3ac88f77d2cf15dd6
* Duration 1:50:28
* Result: :white_check_mark: **SUCCEEDED**
* Error: `N/A`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/c02db6e8-087a-4897-b977-344c6f903a49.gz) (available for 30 days)
* [Build Artifact](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/c02db6e8-087a-4897-b977-344c6f903a49/workspace.zip) (available for 30 days)

foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: ee88e36ec1b759a7c512c43a5d77bf49f309192b
* Duration 1:53:54
* Result: :white_check_mark: **SUCCEEDED**
* Error: `N/A`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/cce7302c-27eb-420d-8a1b-0575a7ac8093.gz) (available for 30 days)
* [Build Artifact](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/cce7302c-27eb-420d-8a1b-0575a7ac8093/workspace.zip) (available for 30 days)

foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: ec866d48abe7d90fdc99f9ed18f4d6adfd3cf0ea
* Duration 4:06:25
* Result: :x: **FAILED**
* Error: `Error while executing command: if $(grep -q -- "--- FAIL:" logs/*.log); then echo "TESTS FAILED SEE THESE LOGS:"; echo ; grep -l -- "--- FAIL:" logs/*.log; exit 1; fi. Reason: exit status 1`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/b2292b7a-13ae-46b3-8742-6fc4e70e4dd5.gz) (available for 30 days)
* [Build Artifact](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/b2292b7a-13ae-46b3-8742-6fc4e70e4dd5/workspace.zip) (available for 30 days)

foundationdb-ci: 
### Result of fdb-kubernetes-operator-pr on Linux CentOS 7

* Commit ID: 3b4214e3b739bcfa2c1e37bd9ac5aab3a26c75d9
* Duration 4:06:13
* Result: :x: **FAILED**
* Error: `Error while executing command: if $(grep -q -- "--- FAIL:" logs/*.log); then echo "TESTS FAILED SEE THESE LOGS:"; echo ; grep -l -- "--- FAIL:" logs/*.log; exit 1; fi. Reason: exit status 1`
* [Build Logs](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator/fdb-kubernetes-operator-pr/1ea5348f-f7eb-4c39-b409-8456836abd52.gz) (available for 30 days)
* [Build Artifact](https://d1e0l78xbkh3xa.cloudfront.net/fdb-kubernetes-operator-pr/1ea5348f-f7eb-4c39-b409-8456836abd52/workspace.zip) (available for 30 days)


Patch:
diff --git a/.github/workflows/pull_request.yml b/.github/workflows/pull_request.yml
index b394fd0f..cf62e50b 100644
--- a/.github/workflows/pull_request.yml
+++ b/.github/workflows/pull_request.yml
@@ -18,7 +18,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.19.2
+          go-version: 1.19.3
       - uses: actions/cache@v2
         with:
           path: ~/go/pkg/mod
@@ -44,7 +44,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v1
       with:
-        go-version: 1.19.2
+        go-version: 1.19.3
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2
       with:
@@ -104,7 +104,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v1
       with:
-        go-version: 1.19.2
+        go-version: 1.19.3
     - name: Check out code into the Go module directory
       uses: actions/checkout@v2
       with:
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
index b1825d59..41baf6d6 100644
--- a/.github/workflows/release.yaml
+++ b/.github/workflows/release.yaml
@@ -43,7 +43,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v1
         with:
-          go-version: 1.19.2
+          go-version: 1.19.3
       #  https://github.com/goreleaser/goreleaser/issues/1311
       - name: Get current semver tag
         run: echo "::set-output name=CURRENT_TAG::$(git describe --tags --match "v*" --abbrev=0)"
diff --git a/.golangci.yml b/.golangci.yml
index 854ab805..fddafd66 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -32,7 +32,7 @@ linters:
     - depguard
     - dogsled
     - errcheck
-#    - errorlint
+    - errorlint
     - exportloopref
 #    - goconst
 #    - gocritic
diff --git a/Dockerfile b/Dockerfile
index 1ccf2acd..fe1560f8 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -1,7 +1,7 @@
 ARG BASE_IMAGE=docker.io/debian:bullseye
 
 # Build the manager binary
-FROM docker.io/library/golang:1.19.2 as builder
+FROM docker.io/library/golang:1.19.3 as builder
 
 # Install FDB this version is only required to compile the fdb operator
 ARG FDB_VERSION=6.2.29
diff --git a/Makefile b/Makefile
index fe4d475e..f1adac2a 100644
--- a/Makefile
+++ b/Makefile
@@ -39,6 +39,7 @@ GORELEASER_PKG=github.com/goreleaser/goreleaser@v1.6.3
 GORELEASER=$(GOBIN)/goreleaser
 BUILD_DEPS?=
 BUILDER?="docker"
+BUILDER_ARGS?=
 KUBECTL_ARGS?=
 
 define godep
@@ -174,7 +175,7 @@ ${GENERATED_GO}: ${GO_SRC} hack/boilerplate.go.txt ${CONTROLLER_GEN}
 
 # Build the container image
 container-build: test_if_changed
-	$(BUILDER) build --build-arg=TAG=${TAG} ${img_build_args} -t ${IMG} .
+	$(BUILDER) build --build-arg=TAG=${TAG} ${img_build_args} $(BUILDER_ARGS) -t ${IMG} .
 
 # Push the container image
 container-push:
diff --git a/api/v1beta2/foundationdb_status.go b/api/v1beta2/foundationdb_status.go
index 413334be..7bc83756 100644
--- a/api/v1beta2/foundationdb_status.go
+++ b/api/v1beta2/foundationdb_status.go
@@ -99,6 +99,9 @@ type FoundationDBStatusClusterInfo struct {
 
 	// RecoveryState represents the recovery state.
 	RecoveryState RecoveryState `json:"recovery_state,omitempty"`
+
+	// ConnectionString represents the connection string in the cluster status json output.
+	ConnectionString string `json:"connection_string,omitempty"`
 }
 
 // FaultTolerance provides information about the fault tolerance status
diff --git a/config/development/kustomization.yaml.sample b/config/development/kustomization.yaml.sample
index 5c260194..e8b0abfd 100644
--- a/config/development/kustomization.yaml.sample
+++ b/config/development/kustomization.yaml.sample
@@ -3,12 +3,14 @@ kind: Kustomization
 
 patchesStrategicMerge:
 - test_certs.yaml
-- backup_credentials.yaml
+# - backup_credentials.yaml
 - debug_logs.yaml
 
 resources:
 - ../deployment
-- ../minio
+# Disable MinIO, the old apiextensions.k8s.io/v1beta1 is not served anymore since 1.22
+# see: https://kubernetes.io/docs/reference/using-api/deprecation-guide/#customresourcedefinition-v122
+# - ../minio
 
 images:
 - name: foundationdb/fdb-kubernetes-operator
diff --git a/config/minio/kustomization.yaml b/config/minio/kustomization.yaml
index deba1049..5aa344fc 100644
--- a/config/minio/kustomization.yaml
+++ b/config/minio/kustomization.yaml
@@ -6,4 +6,4 @@ resources:
 apiVersion: kustomize.config.k8s.io/v1beta1
 kind: Kustomization
 patchesStrategicMerge:
-- instance_overrides.yaml
\ No newline at end of file
+- instance_overrides.yaml
diff --git a/controllers/admin_client_mock.go b/controllers/admin_client_mock.go
index f647a7d4..354d06f0 100644
--- a/controllers/admin_client_mock.go
+++ b/controllers/admin_client_mock.go
@@ -123,6 +123,7 @@ func (client *mockAdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, erro
 	var coordinatorAddresses []string
 	if strings.Contains(client.Cluster.Status.ConnectionString, "@") {
 		coordinatorAddresses = strings.Split(strings.Split(client.Cluster.Status.ConnectionString, "@")[1], ",")
+		status.Cluster.ConnectionString = client.Cluster.Status.ConnectionString
 	} else {
 		coordinatorAddresses = []string{}
 	}
diff --git a/controllers/bounce_processes.go b/controllers/bounce_processes.go
index e261197a..54355739 100644
--- a/controllers/bounce_processes.go
+++ b/controllers/bounce_processes.go
@@ -25,6 +25,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/restarts"
+
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/fdbadminclient"
@@ -149,12 +151,7 @@ func (bounceProcesses) reconcile(ctx context.Context, r *FoundationDBClusterReco
 // getProcessesReadyForRestart returns a slice of process addresses that can be restarted. If addresses are missing or not all processes
 // have the latest configuration this method will return a requeue struct with more details.
 func getProcessesReadyForRestart(ctx context.Context, logger logr.Logger, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster, addressMap map[string][]fdbv1beta2.ProcessAddress) ([]fdbv1beta2.ProcessAddress, *requeue) {
-	processesToBounce := fdbv1beta2.FilterByConditions(cluster.Status.ProcessGroups, map[fdbv1beta2.ProcessGroupConditionType]bool{
-		fdbv1beta2.IncorrectCommandLine: true,
-		fdbv1beta2.IncorrectPodSpec:     false,
-		fdbv1beta2.SidecarUnreachable:   false, // ignore all Process groups that are not reachable and therefore will not get any ConfigMap updates.
-	}, true)
-
+	processesToBounce := fdbv1beta2.FilterByConditions(cluster.Status.ProcessGroups, restarts.GetFilterConditions(cluster), true)
 	addresses := make([]fdbv1beta2.ProcessAddress, 0, len(processesToBounce))
 	allSynced := true
 	var missingAddress []string
@@ -170,6 +167,7 @@ func getProcessesReadyForRestart(ctx context.Context, logger logr.Logger, r *Fou
 		// missing in the status.
 		if missingTime := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses); missingTime != nil {
 			if time.Unix(*missingTime, 0).Add(cluster.GetIgnoreMissingProcessesSeconds()).Before(time.Now()) {
+				logger.Info("ignore process group with missing process", "processGroupID", processGroup.ProcessGroupID)
 				continue
 			}
 		}
@@ -205,6 +203,22 @@ func getProcessesReadyForRestart(ctx context.Context, logger logr.Logger, r *Fou
 		return nil, &requeue{message: "Waiting for config map to sync to all pods", delayedRequeue: true}
 	}
 
+	counts, err := cluster.GetProcessCountsWithDefaults()
+	if err != nil {
+		return nil, &requeue{
+			curError: err,
+		}
+	}
+
+	// If we upgrade the cluster wait until all processes are ready for the restart. In the future we can adjust this
+	// to only be a requirement for version incompatible upgrades. In addition we probably only want to block for a
+	// certain threshold either as a percentage or as a fixed number (which could also be 0).
+	if cluster.IsBeingUpgraded() && counts.Total() != len(addresses) {
+		return nil, &requeue{
+			message:        fmt.Sprintf("expected %d processes, got %d processes ready to restart", counts.Total(), len(addresses)),
+			delayedRequeue: true}
+	}
+
 	return addresses, nil
 }
 
diff --git a/controllers/change_coordinators.go b/controllers/change_coordinators.go
index e3ca5f15..850535be 100644
--- a/controllers/change_coordinators.go
+++ b/controllers/change_coordinators.go
@@ -23,6 +23,8 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/locality"
+	"github.com/go-logr/logr"
 
 	corev1 "k8s.io/api/core/v1"
 
@@ -47,15 +49,15 @@ func (c changeCoordinators) reconcile(ctx context.Context, r *FoundationDBCluste
 	}
 	defer adminClient.Close()
 
-	connectionString, err := adminClient.GetConnectionString()
+	status, err := adminClient.GetStatus()
 	if err != nil {
 		return &requeue{curError: err}
 	}
 
-	if connectionString != cluster.Status.ConnectionString {
+	if status.Cluster.ConnectionString != cluster.Status.ConnectionString {
 		logger.Info("Updating out-of-date connection string")
-		r.Recorder.Event(cluster, corev1.EventTypeNormal, "UpdatingConnectionString", fmt.Sprintf("Setting connection string to %s", connectionString))
-		cluster.Status.ConnectionString = connectionString
+		r.Recorder.Event(cluster, corev1.EventTypeNormal, "UpdatingConnectionString", fmt.Sprintf("Setting connection string to %s", status.Cluster.ConnectionString))
+		cluster.Status.ConnectionString = status.Cluster.ConnectionString
 		err = r.updateOrApply(ctx, cluster)
 
 		if err != nil {
@@ -63,17 +65,12 @@ func (c changeCoordinators) reconcile(ctx context.Context, r *FoundationDBCluste
 		}
 	}
 
-	status, err := adminClient.GetStatus()
-	if err != nil {
-		return &requeue{curError: err}
-	}
-
 	coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
 	for _, coordinator := range status.Client.Coordinators.Coordinators {
 		coordinatorStatus[coordinator.Address.String()] = false
 	}
 
-	hasValidCoordinators, allAddressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
+	hasValidCoordinators, allAddressesValid, err := locality.CheckCoordinatorValidity(logger, cluster, status, coordinatorStatus)
 	if err != nil {
 		return &requeue{curError: err}
 	}
@@ -96,7 +93,7 @@ func (c changeCoordinators) reconcile(ctx context.Context, r *FoundationDBCluste
 	logger.Info("Changing coordinators")
 	r.Recorder.Event(cluster, corev1.EventTypeNormal, "ChangingCoordinators", "Choosing new coordinators")
 
-	coordinators, err := selectCoordinators(cluster, status)
+	coordinators, err := selectCoordinators(logger, cluster, status)
 	if err != nil {
 		return &requeue{curError: err}
 	}
@@ -107,7 +104,7 @@ func (c changeCoordinators) reconcile(ctx context.Context, r *FoundationDBCluste
 	}
 
 	logger.Info("Final coordinators candidates", "coordinators", coordinatorAddresses)
-	connectionString, err = adminClient.ChangeCoordinators(coordinatorAddresses)
+	connectionString, err := adminClient.ChangeCoordinators(coordinatorAddresses)
 	if err != nil {
 		return &requeue{curError: err}
 	}
@@ -120,9 +117,10 @@ func (c changeCoordinators) reconcile(ctx context.Context, r *FoundationDBCluste
 	return nil
 }
 
+// TODO move them into separate package?
 // selectCandidates is a helper for Reconcile that picks non-excluded, not-being-removed class-matching process groups.
-func selectCandidates(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus) ([]localityInfo, error) {
-	candidates := make([]localityInfo, 0, len(status.Cluster.Processes))
+func selectCandidates(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus) ([]locality.Info, error) {
+	candidates := make([]locality.Info, 0, len(status.Cluster.Processes))
 	for _, process := range status.Cluster.Processes {
 		if process.Excluded {
 			continue
@@ -136,29 +134,28 @@ func selectCandidates(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta
 			continue
 		}
 
-		locality, err := localityInfoForProcess(process, cluster.Spec.MainContainer.EnableTLS)
+		currentLocality, err := locality.InfoForProcess(process, cluster.Spec.MainContainer.EnableTLS)
 		if err != nil {
 			return candidates, err
 		}
 
-		candidates = append(candidates, locality)
+		candidates = append(candidates, currentLocality)
 	}
 
 	return candidates, nil
 }
 
-func selectCoordinators(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus) ([]localityInfo, error) {
-	logger := log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name, "reconciler", "changeCoordinators")
+func selectCoordinators(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus) ([]locality.Info, error) {
 	var err error
 	coordinatorCount := cluster.DesiredCoordinatorCount()
 
 	candidates, err := selectCandidates(cluster, status)
 	if err != nil {
-		return []localityInfo{}, err
+		return []locality.Info{}, err
 	}
 
-	coordinators, err := chooseDistributedProcesses(cluster, candidates, coordinatorCount, processSelectionConstraint{
-		HardLimits: getHardLimits(cluster),
+	coordinators, err := locality.ChooseDistributedProcesses(cluster, candidates, coordinatorCount, locality.ProcessSelectionConstraint{
+		HardLimits: locality.GetHardLimits(cluster),
 	})
 
 	logger.Info("Current coordinators", "coordinators", coordinators)
@@ -171,7 +168,7 @@ func selectCoordinators(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1be
 		coordinatorStatus[getCoordinatorAddress(cluster, coordinator).String()] = false
 	}
 
-	hasValidCoordinators, allAddressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
+	hasValidCoordinators, allAddressesValid, err := locality.CheckCoordinatorValidity(logger, cluster, status, coordinatorStatus)
 	if err != nil {
 		return coordinators, err
 	}
@@ -187,7 +184,7 @@ func selectCoordinators(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1be
 	return coordinators, nil
 }
 
-func getCoordinatorAddress(cluster *fdbv1beta2.FoundationDBCluster, locality localityInfo) fdbv1beta2.ProcessAddress {
+func getCoordinatorAddress(cluster *fdbv1beta2.FoundationDBCluster, locality locality.Info) fdbv1beta2.ProcessAddress {
 	dnsName := locality.LocalityData[fdbv1beta2.FDBLocalityDNSNameKey]
 
 	address := locality.Address
diff --git a/controllers/choose_removals.go b/controllers/choose_removals.go
index 6d277e1e..29150c42 100644
--- a/controllers/choose_removals.go
+++ b/controllers/choose_removals.go
@@ -24,6 +24,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/locality"
+
 	corev1 "k8s.io/api/core/v1"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
@@ -60,10 +62,10 @@ func (c chooseRemovals) reconcile(ctx context.Context, r *FoundationDBClusterRec
 	if err != nil {
 		return &requeue{curError: err}
 	}
-	localityMap := make(map[string]localityInfo)
+	localityMap := make(map[string]locality.Info)
 	for _, process := range status.Cluster.Processes {
 		id := process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]
-		localityMap[id] = localityInfo{ID: id, Address: process.Address, LocalityData: process.Locality}
+		localityMap[id] = locality.Info{ID: id, Address: process.Address, LocalityData: process.Locality}
 	}
 
 	remainingProcessMap := make(map[string]bool, len(cluster.Status.ProcessGroups))
@@ -71,7 +73,7 @@ func (c chooseRemovals) reconcile(ctx context.Context, r *FoundationDBClusterRec
 	for _, processClass := range fdbv1beta2.ProcessClasses {
 		desiredCount := desiredCounts[processClass]
 		removedCount := currentCounts[processClass] - desiredCount
-		processClassLocality := make([]localityInfo, 0, currentCounts[processClass])
+		processClassLocality := make([]locality.Info, 0, currentCounts[processClass])
 
 		for _, processGroup := range cluster.Status.ProcessGroupsByProcessClass(processClass) {
 			if processGroup.IsMarkedForRemoval() {
@@ -87,7 +89,7 @@ func (c chooseRemovals) reconcile(ctx context.Context, r *FoundationDBClusterRec
 		if removedCount > 0 {
 			r.Recorder.Event(cluster, corev1.EventTypeNormal, "ShrinkingProcesses", fmt.Sprintf("Removing %d %s processes", removedCount, processClass))
 
-			remainingProcesses, err := chooseDistributedProcesses(cluster, processClassLocality, desiredCount, processSelectionConstraint{})
+			remainingProcesses, err := locality.ChooseDistributedProcesses(cluster, processClassLocality, desiredCount, locality.ProcessSelectionConstraint{})
 			if err != nil {
 				return &requeue{curError: err}
 			}
diff --git a/controllers/cluster_controller.go b/controllers/cluster_controller.go
index 5b6d8325..95d98902 100644
--- a/controllers/cluster_controller.go
+++ b/controllers/cluster_controller.go
@@ -23,11 +23,8 @@ package controllers
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
-	"math"
 	"regexp"
-	"sort"
 	"time"
 
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/fdbadminclient"
@@ -373,368 +370,6 @@ type clusterSubReconciler interface {
 	reconcile(ctx context.Context, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster) *requeue
 }
 
-// localityInfo captures information about a process for the purposes of
-// choosing diverse locality.
-type localityInfo struct {
-	// The process group ID
-	ID string
-
-	// The process's public address.
-	Address fdbv1beta2.ProcessAddress
-
-	// The locality map.
-	LocalityData map[string]string
-
-	Class fdbv1beta2.ProcessClass
-}
-
-// Sort processes by their priority and their ID.
-// 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(cluster *fdbv1beta2.FoundationDBCluster, processes []localityInfo) {
-	// Sort the processes for ID to ensure we have a stable input
-	sort.SliceStable(processes, func(i, j int) bool {
-		p1 := cluster.GetClassCandidatePriority(processes[i].Class)
-		p2 := cluster.GetClassCandidatePriority(processes[j].Class)
-
-		// If both have the same priority sort them by the process ID
-		if p1 == p2 {
-			return processes[i].ID < processes[j].ID
-		}
-
-		// prefer processes with a higher priority
-		return p1 > p2
-	})
-}
-
-// localityInfoForProcess converts the process information from the JSON status
-// into locality info for selecting processes.
-func localityInfoForProcess(process fdbv1beta2.FoundationDBStatusProcessInfo, mainContainerTLS bool) (localityInfo, error) {
-	addresses, err := fdbv1beta2.ParseProcessAddressesFromCmdline(process.CommandLine)
-	if err != nil {
-		return localityInfo{}, err
-	}
-
-	var addr fdbv1beta2.ProcessAddress
-	// Iterate the addresses and set the expected address as process address
-	// e.g. if we want to use TLS set it to the tls address otherwise use the non-TLS.
-	for _, address := range addresses {
-		if address.Flags["tls"] == mainContainerTLS {
-			addr = address
-			break
-		}
-	}
-
-	return localityInfo{
-		ID:           process.Locality["instance_id"],
-		Address:      addr,
-		LocalityData: process.Locality,
-		Class:        process.ProcessClass,
-	}, nil
-}
-
-// localityInfoForProcess converts the process information from the sidecar's
-// context into locality info for selecting processes.
-func localityInfoFromSidecar(cluster *fdbv1beta2.FoundationDBCluster, client podclient.FdbPodClient) (localityInfo, error) {
-	substitutions, err := client.GetVariableSubstitutions()
-	if err != nil {
-		return localityInfo{}, err
-	}
-
-	if substitutions == nil {
-		return localityInfo{}, nil
-	}
-
-	// This locality information is only used during the initial cluster file generation.
-	// So it should be good to only use the first process address here.
-	// This has the implication that in the initial cluster file only the first processes will be used.
-	address := cluster.GetFullAddress(substitutions["FDB_PUBLIC_IP"], 1)
-	return localityInfo{
-		ID:      substitutions["FDB_INSTANCE_ID"],
-		Address: address,
-		LocalityData: map[string]string{
-			fdbv1beta2.FDBLocalityZoneIDKey:  substitutions["FDB_ZONE_ID"],
-			fdbv1beta2.FDBLocalityDNSNameKey: substitutions["FDB_DNS_NAME"],
-		},
-	}, nil
-}
-
-// notEnoughProcessesError is returned when we cannot recruit enough processes.
-type notEnoughProcessesError struct {
-	// desired defines the number of processes we wanted to recruit.
-	Desired int
-
-	// chosen defines the number of processes we were able to recruit.
-	Chosen int
-
-	// options defines the processes that we were selecting from.
-	Options []localityInfo
-}
-
-// Error formats an error message.
-func (err notEnoughProcessesError) Error() string {
-	return fmt.Sprintf("Could only select %d processes, but %d are required", err.Chosen, err.Desired)
-}
-
-// processSelectionConstraint defines constraints on how we choose processes
-// in chooseDistributedProcesses
-type processSelectionConstraint struct {
-	// Fields defines the locality fields we should consider when selecting
-	// processes.
-	Fields []string
-
-	// HardLimits defines a maximum number of processes to recruit on any single
-	// value for a given locality field.
-	HardLimits map[string]int
-}
-
-// chooseDistributedProcesses recruits a maximally well-distributed set
-// of processes from a set of potential candidates.
-func chooseDistributedProcesses(cluster *fdbv1beta2.FoundationDBCluster, processes []localityInfo, count int, constraint processSelectionConstraint) ([]localityInfo, error) {
-	chosen := make([]localityInfo, 0, count)
-	chosenIDs := make(map[string]bool, count)
-
-	fields := constraint.Fields
-	if len(fields) == 0 {
-		fields = []string{fdbv1beta2.FDBLocalityZoneIDKey, fdbv1beta2.FDBLocalityDCIDKey}
-	}
-
-	chosenCounts := make(map[string]map[string]int, len(fields))
-	hardLimits := make(map[string]int, len(fields))
-	currentLimits := make(map[string]int, len(fields))
-
-	if constraint.HardLimits != nil {
-		for field, limit := range constraint.HardLimits {
-			hardLimits[field] = limit
-		}
-	}
-
-	for _, field := range fields {
-		chosenCounts[field] = make(map[string]int)
-		if hardLimits[field] == 0 {
-			hardLimits[field] = count
-		}
-		currentLimits[field] = 1
-	}
-
-	// Sort the processes to ensure a deterministic result
-	sortLocalities(cluster, processes)
-
-	for len(chosen) < count {
-		choseAny := false
-
-		for _, process := range processes {
-			if chosenIDs[process.ID] {
-				continue
-			}
-
-			eligible := true
-			for _, field := range fields {
-				value := process.LocalityData[field]
-				if chosenCounts[field][value] >= currentLimits[field] {
-					eligible = false
-					break
-				}
-			}
-
-			if !eligible {
-				continue
-			}
-
-			chosen = append(chosen, process)
-			chosenIDs[process.ID] = true
-
-			choseAny = true
-
-			for _, field := range fields {
-				value := process.LocalityData[field]
-				chosenCounts[field][value]++
-			}
-
-			if len(chosen) == count {
-				break
-			}
-		}
-
-		if !choseAny {
-			incrementedLimits := false
-			for indexOfField := len(fields) - 1; indexOfField >= 0; indexOfField-- {
-				field := fields[indexOfField]
-				if currentLimits[field] < hardLimits[field] {
-					currentLimits[field]++
-					incrementedLimits = true
-					break
-				}
-			}
-			if !incrementedLimits {
-				return chosen, notEnoughProcessesError{Desired: count, Chosen: len(chosen), Options: processes}
-			}
-		}
-	}
-
-	return chosen, nil
-}
-
-func getHardLimits(cluster *fdbv1beta2.FoundationDBCluster) map[string]int {
-	if cluster.Spec.DatabaseConfiguration.UsableRegions <= 1 {
-		return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1}
-	}
-
-	// TODO (johscheuer): should we calculate that based on the number of DCs?
-	maxCoordinatorsPerDC := int(math.Floor(float64(cluster.DesiredCoordinatorCount()) / 2.0))
-
-	return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1, fdbv1beta2.FDBLocalityDCIDKey: maxCoordinatorsPerDC}
-}
-
-// checkCoordinatorValidity determines if the cluster's current coordinators
-// meet the fault tolerance requirements.
-//
-// The first return value will be whether the coordinators are valid.
-// The second return value will be whether the processes have their TLS flags
-// matching the cluster spec.
-// The third return value will hold any errors encountered when checking the
-// coordinators.
-func checkCoordinatorValidity(cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus, coordinatorStatus map[string]bool) (bool, bool, error) {
-	if len(coordinatorStatus) == 0 {
-		return false, false, errors.New("unable to get coordinator status")
-	}
-
-	curLog := log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name)
-
-	allAddressesValid := true
-	allEligible := true
-	allUsingCorrectAddress := true
-
-	coordinatorZones := make(map[string]int, len(coordinatorStatus))
-	coordinatorDCs := make(map[string]int, len(coordinatorStatus))
-	processGroups := make(map[string]*fdbv1beta2.ProcessGroupStatus)
-	for _, processGroup := range cluster.Status.ProcessGroups {
-		processGroups[processGroup.ProcessGroupID] = processGroup
-	}
-
-	for _, process := range status.Cluster.Processes {
-		pLogger := curLog.WithValues("process", process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey])
-		if process.Address.IsEmpty() {
-			pLogger.Info("Skip process with empty address")
-			continue
-		}
-
-		if process.ProcessClass == fdbv1beta2.ProcessClassTest {
-			pLogger.Info("Ignoring test process")
-			continue
-		}
-
-		processGroupStatus := processGroups[process.Locality["instance_id"]]
-		pendingRemoval := processGroupStatus != nil && processGroupStatus.IsMarkedForRemoval()
-		if processGroupStatus != nil && cluster.SkipProcessGroup(processGroupStatus) {
-			log.Info("Skipping process group with pending Pod",
-				"namespace", cluster.Namespace,
-				"cluster", cluster.Name,
-				"processGroupID", process.Locality["instance_id"],
-				"class", process.ProcessClass)
-			continue
-		}
-
-		addresses, err := fdbv1beta2.ParseProcessAddressesFromCmdline(process.CommandLine)
-		if err != nil {
-			// We will end here in the error case when the address
-			// is not parsable e.g. no IP address is assigned.
-			allAddressesValid = false
-			// add: command_line
-			pLogger.Info("Could not parse address from command_line", "command_line", process.CommandLine)
-			continue
-		}
-
-		var ipAddress fdbv1beta2.ProcessAddress
-		for _, addr := range addresses {
-			if addr.Flags["tls"] == cluster.Spec.MainContainer.EnableTLS {
-				ipAddress = addr
-				break
-			}
-		}
-
-		coordinatorAddress := ""
-		_, isCoordinatorWithIP := coordinatorStatus[ipAddress.String()]
-		if isCoordinatorWithIP {
-			coordinatorAddress = ipAddress.String()
-		}
-
-		dnsName := process.Locality[fdbv1beta2.FDBLocalityDNSNameKey]
-		dnsAddress := fdbv1beta2.ProcessAddress{
-			StringAddress: dnsName,
-			Port:          ipAddress.Port,
-			Flags:         ipAddress.Flags,
-		}
-		_, isCoordinatorWithDNS := coordinatorStatus[dnsAddress.String()]
-
-		if !isCoordinatorWithDNS {
-			dnsAddress = ipAddress
-			dnsAddress.FromHostname = true
-			_, isCoordinatorWithDNS = coordinatorStatus[dnsAddress.String()]
-		}
-
-		if isCoordinatorWithDNS {
-			coordinatorAddress = dnsAddress.String()
-		}
-
-		if coordinatorAddress != "" && !process.Excluded && !pendingRemoval {
-			coordinatorStatus[coordinatorAddress] = true
-		}
-
-		if coordinatorAddress != "" {
-			coordinatorZones[process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]]++
-			coordinatorDCs[process.Locality[fdbv1beta2.FDBLocalityDCIDKey]]++
-
-			if !cluster.IsEligibleAsCandidate(process.ProcessClass) {
-				pLogger.Info("Process class of process is not eligible as coordinator", "class", process.ProcessClass, "address", coordinatorAddress)
-				allEligible = false
-			}
-
-			useDNS := cluster.UseDNSInClusterFile() && dnsName != ""
-			if (isCoordinatorWithIP && useDNS) || (isCoordinatorWithDNS && !useDNS) {
-				pLogger.Info("Coordinator is not using the correct address type", "coordinatorList", coordinatorStatus, "address", coordinatorAddress, "expectingDNS", useDNS, "usingDNS", isCoordinatorWithDNS)
-				allUsingCorrectAddress = false
-			}
-		}
-
-		if ipAddress.IPAddress == nil {
-			pLogger.Info("Process has invalid IP address", "addresses", addresses)
-			allAddressesValid = false
-		}
-	}
-
-	desiredCount := cluster.DesiredCoordinatorCount()
-	hasEnoughZones := len(coordinatorZones) == desiredCount
-	if !hasEnoughZones {
-		curLog.Info("Cluster does not have coordinators in the correct number of zones", "desiredCount", desiredCount, "coordinatorZones", coordinatorZones)
-	}
-
-	var maxCoordinatorsPerDC int
-	hasEnoughDCs := true
-	if cluster.Spec.DatabaseConfiguration.UsableRegions > 1 {
-		maxCoordinatorsPerDC = int(math.Floor(float64(desiredCount) / 2.0))
-
-		for dc, count := range coordinatorDCs {
-			if count > maxCoordinatorsPerDC {
-				curLog.Info("Cluster has too many coordinators in a single DC", "DC", dc, "count", count, "max", maxCoordinatorsPerDC)
-				hasEnoughDCs = false
-			}
-		}
-	}
-
-	allHealthy := true
-	for address, healthy := range coordinatorStatus {
-		allHealthy = allHealthy && healthy
-
-		if !healthy {
-			curLog.Info("Cluster has an unhealthy coordinator", "address", address)
-		}
-	}
-
-	return hasEnoughDCs && hasEnoughZones && allHealthy && allUsingCorrectAddress && allEligible, allAddressesValid, nil
-}
-
 // newFdbPodClient builds a client for working with an FDB Pod
 func (r *FoundationDBClusterReconciler) newFdbPodClient(cluster *fdbv1beta2.FoundationDBCluster, pod *corev1.Pod) (podclient.FdbPodClient, error) {
 	return internal.NewFdbPodClient(cluster, pod, log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name, "pod", pod.Name), r.GetTimeout, r.PostTimeout)
diff --git a/controllers/exclude_processes.go b/controllers/exclude_processes.go
index 7925094b..fb513c81 100644
--- a/controllers/exclude_processes.go
+++ b/controllers/exclude_processes.go
@@ -80,11 +80,6 @@ func (e excludeProcesses) reconcile(_ context.Context, r *FoundationDBClusterRec
 			}
 		}
 
-		hasLock, err := r.takeLock(cluster, fmt.Sprintf("excluding processes: %v", fdbProcessesToExclude))
-		if !hasLock {
-			return &requeue{curError: err}
-		}
-
 		r.Recorder.Event(cluster, corev1.EventTypeNormal, "ExcludingProcesses", fmt.Sprintf("Excluding %v", fdbProcessesToExclude))
 
 		err = adminClient.ExcludeProcesses(fdbProcessesToExclude)
diff --git a/controllers/generate_initial_cluster_file.go b/controllers/generate_initial_cluster_file.go
index b96e11c1..5b959c47 100644
--- a/controllers/generate_initial_cluster_file.go
+++ b/controllers/generate_initial_cluster_file.go
@@ -24,6 +24,8 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/locality"
+
 	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
 
 	corev1 "k8s.io/api/core/v1"
@@ -100,31 +102,31 @@ func (g generateInitialClusterFile) reconcile(ctx context.Context, r *Foundation
 		}
 	}
 
-	processLocality := make([]localityInfo, 0, len(pods))
+	processLocality := make([]locality.Info, 0, len(pods))
 	for _, pod := range pods {
 		client, message := r.getPodClient(cluster, pod)
 		if client == nil {
 			return &requeue{message: message, delay: podSchedulingDelayDuration}
 		}
-		locality, err := localityInfoFromSidecar(cluster, client)
+		currentLocality, err := locality.InfoFromSidecar(cluster, client)
 		if err != nil {
 			return &requeue{curError: err}
 		}
-		if locality.ID == "" {
+		if currentLocality.ID == "" {
 			processGroupID := internal.GetProcessGroupIDFromMeta(cluster, pod.ObjectMeta)
 			logger.Info("Pod is ineligible to be a coordinator due to missing locality information", "processGroupID", processGroupID)
 			continue
 		}
-		processLocality = append(processLocality, locality)
+		processLocality = append(processLocality, currentLocality)
 	}
 
-	coordinators, err := chooseDistributedProcesses(cluster, processLocality, count, processSelectionConstraint{})
+	coordinators, err := locality.ChooseDistributedProcesses(cluster, processLocality, count, locality.ProcessSelectionConstraint{})
 	if err != nil {
 		return &requeue{curError: err}
 	}
 
-	for _, locality := range coordinators {
-		connectionString.Coordinators = append(connectionString.Coordinators, getCoordinatorAddress(cluster, locality).String())
+	for _, currentLocality := range coordinators {
+		connectionString.Coordinators = append(connectionString.Coordinators, getCoordinatorAddress(cluster, currentLocality).String())
 	}
 
 	cluster.Status.ConnectionString = connectionString.String()
@@ -133,5 +135,6 @@ func (g generateInitialClusterFile) reconcile(ctx context.Context, r *Foundation
 	if err != nil {
 		return &requeue{curError: err}
 	}
+
 	return nil
 }
diff --git a/controllers/update_status.go b/controllers/update_status.go
index 9d9d138d..3c1d1304 100644
--- a/controllers/update_status.go
+++ b/controllers/update_status.go
@@ -26,6 +26,8 @@ import (
 	"sort"
 	"time"
 
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/locality"
+
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podmanager"
 	"github.com/go-logr/logr"
 
@@ -81,6 +83,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 
 		databaseStatus, err = adminClient.GetStatus()
 		_ = adminClient.Close()
+
 		if err != nil {
 			if cluster.Status.Configured {
 				return &requeue{curError: err, delayedRequeue: true}
@@ -220,7 +223,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 			coordinatorStatus[coordinator.Address.String()] = false
 		}
 
-		coordinatorsValid, _, err := checkCoordinatorValidity(cluster, databaseStatus, coordinatorStatus)
+		coordinatorsValid, _, err := locality.CheckCoordinatorValidity(logger, cluster, databaseStatus, coordinatorStatus)
 		if err != nil {
 			return &requeue{curError: err, delayedRequeue: true}
 		}
@@ -321,16 +324,15 @@ func tryConnectionOptions(cluster *fdbv1beta2.FoundationDBCluster, r *Foundation
 		logger.Info("Attempting to get connection string from cluster", "connectionString", connectionString)
 		cluster.Status.ConnectionString = connectionString
 		adminClient, clientErr := r.getDatabaseClientProvider().GetAdminClient(cluster, r)
-
 		if clientErr != nil {
 			return originalConnectionString, clientErr
 		}
 
 		activeConnectionString, err := adminClient.GetConnectionString()
 
-		clientErr = adminClient.Close()
-		if clientErr != nil {
-			logger.V(1).Info("Could not close admin client")
+		closeErr := adminClient.Close()
+		if closeErr != nil {
+			logger.V(1).Info("Could not close admin client", "error", closeErr)
 		}
 
 		if err == nil {
diff --git a/docs/design/implemented/automatic_replacements.md b/docs/design/implemented/automatic_replacements.md
index e7e621b1..8415659a 100644
--- a/docs/design/implemented/automatic_replacements.md
+++ b/docs/design/implemented/automatic_replacements.md
@@ -42,7 +42,7 @@ In order to make sure we consistently detect failures, we will add a new cronjob
 
 ### Making Replacements Smooth
 
-In order for replacements to succeed reliably, we need to handle the case where an instance does not have an IP address and is pending removal. The design on [improving process tracking in the cluster status](https://github.com/FoundationDB/fdb-kubernetes-operator/blob/master/docs/design/process_group_status.md) will improve this considerably by tracking the previously known IP addresses and allowing us to exclude those addresses when the pod is still pending.
+In order for replacements to succeed reliably, we need to handle the case where an instance does not have an IP address and is pending removal. The design on [improving process tracking in the cluster status](https://github.com/FoundationDB/fdb-kubernetes-operator/blob/master/docs/design/implemented/process_group_status.md) will improve this considerably by tracking the previously known IP addresses and allowing us to exclude those addresses when the pod is still pending.
 
 ### Additional Safety Measures
 
diff --git a/fdbclient/admin_client.go b/fdbclient/admin_client.go
index 0a5197e0..fcef6469 100644
--- a/fdbclient/admin_client.go
+++ b/fdbclient/admin_client.go
@@ -23,9 +23,11 @@ package fdbclient
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
+	"path"
 	"regexp"
 	"strconv"
 	"strings"
@@ -41,7 +43,9 @@ import (
 )
 
 const (
-	fdbcliStr = "fdbcli"
+	fdbcliStr     = "fdbcli"
+	fdbbackupStr  = "fdbbackup"
+	fdbrestoreStr = "fdbrestore"
 )
 
 var adminClientMutex sync.Mutex
@@ -81,6 +85,10 @@ type cliAdminClient struct {
 
 	// log implementation for logging output
 	log logr.Logger
+
+	// cmdRunner is an interface to run commands. In the real runner we use the exec package to execute binaries. In
+	// the mock runner we can define mocked output for better integration tests,.
+	cmdRunner commandRunner
 }
 
 // NewCliAdminClient generates an Admin client for a cluster
@@ -100,7 +108,14 @@ func NewCliAdminClient(cluster *fdbv1beta2.FoundationDBCluster, _ client.Client,
 		return nil, err
 	}
 
-	return &cliAdminClient{Cluster: cluster, clusterFilePath: clusterFile.Name(), useClientLibrary: true, log: log}, nil
+	logger := log.WithValues("namespace", cluster.Namespace, "cluster", cluster.Name)
+	return &cliAdminClient{
+		Cluster:          cluster,
+		clusterFilePath:  clusterFile.Name(),
+		useClientLibrary: true,
+		log:              logger,
+		cmdRunner:        &realCommandRunner{log: logger},
+	}, nil
 }
 
 // cliCommand describes a command that we are running against FDB.
@@ -123,19 +138,23 @@ type cliCommand struct {
 
 // hasTimeoutArg determines whether a command accepts a timeout argument.
 func (command cliCommand) hasTimeoutArg() bool {
-	return command.binary == "" || command.binary == fdbcliStr
+	return command.isFdbCli()
 }
 
-// hasDashInLogDir determines whether a command has a log-dir argument or a
-// logdir argument.
-func (command cliCommand) hasDashInLogDir() bool {
-	return command.binary == "" || command.binary == fdbcliStr
+// getLogDirParameter returns the log dir parameter for a command, depending on the binary the log dir parameter
+// has a dash or not.
+func (command cliCommand) getLogDirParameter() string {
+	if command.isFdbCli() {
+		return "--log-dir"
+	}
+
+	return "--logdir"
 }
 
 // getClusterFileFlag gets the flag this command uses for its cluster file
 // argument.
 func (command cliCommand) getClusterFileFlag() string {
-	if command.binary == "fdbrestore" {
+	if command.binary == fdbrestoreStr {
 		return "--dest_cluster_file"
 	}
 
@@ -151,38 +170,46 @@ func (command cliCommand) getTimeout() time.Duration {
 	return DefaultCLITimeout
 }
 
-// getBinaryPath generates the path to an FDB binary.
-func getBinaryPath(binaryName string, version string) string {
-	parsed, _ := fdbv1beta2.ParseFdbVersion(version)
-	return fmt.Sprintf("%s/%s/%s", os.Getenv("FDB_BINARY_DIR"), parsed.GetBinaryVersion(), binaryName)
+// getVersion returns the versions defined in the command or if not present returns the running version of the
+// cluster.
+func (command cliCommand) getVersion(cluster *fdbv1beta2.FoundationDBCluster) string {
+	if command.version != "" {
+		return command.version
+	}
+
+	return cluster.GetRunningVersion()
 }
 
-// runCommand executes a command in the CLI.
-func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
-	version := command.version
-	if version == "" {
-		version = client.Cluster.Status.RunningVersion
-	}
-	if version == "" {
-		version = client.Cluster.Spec.Version
+// getBinary returns the binary of the command, if unset this will default to fdbcli.
+func (command cliCommand) getBinary() string {
+	if command.binary != "" {
+		return command.binary
 	}
 
-	binaryName := command.binary
-	if binaryName == "" {
-		binaryName = fdbcliStr
-	}
+	return fdbcliStr
+}
 
-	binary := getBinaryPath(binaryName, version)
-	hardTimeout := command.getTimeout()
-	args := make([]string, 0, 9)
-	args = append(args, command.args...)
+// isFdbCli returns true if the used binary is fdbcli.
+func (command cliCommand) isFdbCli() bool {
+	return command.getBinary() == fdbcliStr
+}
+
+// getBinaryPath generates the path to an FDB binary.
+func getBinaryPath(binaryName string, version string) string {
+	parsed, _ := fdbv1beta2.ParseFdbVersion(version)
+	return path.Join(os.Getenv("FDB_BINARY_DIR"), parsed.GetBinaryVersion(), binaryName)
+}
+
+func (client *cliAdminClient) getArgsAndTimeout(command cliCommand) ([]string, time.Duration) {
+	args := make([]string, len(command.args))
+	copy(args, command.args)
 	if len(args) == 0 {
 		args = append(args, "--exec", command.command)
 	}
 
 	args = append(args, command.getClusterFileFlag(), client.clusterFilePath, "--log")
 	// We only want to pass the knobs to fdbbackup and fdbrestore
-	if binaryName != fdbcliStr {
+	if command.isFdbCli() {
 		args = append(args, client.knobs...)
 	}
 
@@ -190,7 +217,7 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 	if traceDir != "" {
 		args = append(args, "--log")
 
-		if binaryName == fdbcliStr {
+		if command.isFdbCli() {
 			format := os.Getenv("FDB_NETWORK_OPTION_TRACE_FORMAT")
 			if format == "" {
 				format = "xml"
@@ -198,28 +225,30 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 
 			args = append(args, "--trace_format", format)
 		}
-		if command.hasDashInLogDir() {
-			args = append(args, "--log-dir", traceDir)
-		} else {
-			args = append(args, "--logdir", traceDir)
-		}
+
+		args = append(args, command.getLogDirParameter(), traceDir)
 	}
 
+	hardTimeout := command.getTimeout()
 	if command.hasTimeoutArg() {
 		args = append(args, "--timeout", strconv.Itoa(int(command.getTimeout().Seconds())))
 		hardTimeout += command.getTimeout()
 	}
+
+	return args, hardTimeout
+}
+
+// runCommand executes a command in the CLI.
+func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
+	args, hardTimeout := client.getArgsAndTimeout(command)
 	timeoutContext, cancelFunction := context.WithTimeout(context.Background(), hardTimeout)
 	defer cancelFunction()
-	execCommand := exec.CommandContext(timeoutContext, binary, args...)
 
-	client.log.Info("Running command", "namespace", client.Cluster.Namespace, "cluster", client.Cluster.Name, "path", execCommand.Path, "args", execCommand.Args)
-
-	output, err := execCommand.CombinedOutput()
+	output, err := client.cmdRunner.runCommand(timeoutContext, getBinaryPath(command.getBinary(), command.getVersion(client.Cluster)), args...)
 	if err != nil {
-		exitError, canCast := err.(*exec.ExitError)
-		if canCast {
-			client.log.Error(exitError, "Error from FDB command", "namespace", client.Cluster.Namespace, "cluster", client.Cluster.Name, "code", exitError.ProcessState.ExitCode(), "stdout", string(output), "stderr", string(exitError.Stderr))
+		var exitError *exec.ExitError
+		if errors.As(err, &exitError) {
+			client.log.Error(exitError, "Error from FDB command", "code", exitError.ProcessState.ExitCode(), "stdout", string(output), "stderr", string(exitError.Stderr))
 		}
 
 		// If we hit a timeout report it as a timeout error
@@ -240,7 +269,7 @@ func (client *cliAdminClient) runCommand(command cliCommand) (string, error) {
 	} else {
 		debugOutput = outputString
 	}
-	client.log.Info("Command completed", "namespace", client.Cluster.Namespace, "cluster", client.Cluster.Name, "output", debugOutput)
+	client.log.Info("Command completed", "output", debugOutput)
 
 	return outputString, nil
 }
@@ -263,7 +292,8 @@ func (client *cliAdminClient) runCommandWithBackoff(command string) (string, err
 			break
 		}
 
-		if _, ok := err.(fdbv1beta2.TimeoutError); ok {
+		var timoutError *fdbv1beta2.TimeoutError
+		if errors.As(err, &timoutError) {
 			client.log.Info("timeout issue will retry with higher timeout")
 			currentTimeout *= 2
 			continue
@@ -521,7 +551,6 @@ func (client *cliAdminClient) CanSafelyRemove(addresses []fdbv1beta2.ProcessAddr
 
 	exclusions := getRemainingAndExcludedFromStatus(status, addresses)
 	client.log.Info("Filtering excluded processes",
-		"namespace", client.Cluster.Namespace,
 		"cluster", client.Cluster.Name,
 		"inProgress", exclusions.inProgress,
 		"fullyExcluded", exclusions.fullyExcluded,
@@ -536,7 +565,7 @@ func (client *cliAdminClient) CanSafelyRemove(addresses []fdbv1beta2.ProcessAddr
 			return nil, err
 		}
 		exclusionResults := parseExclusionOutput(output)
-		client.log.Info("Checking exclusion results", "namespace", client.Cluster.Namespace, "cluster", client.Cluster.Name, "addresses", exclusions.inProgress, "results", exclusionResults)
+		client.log.Info("Checking exclusion results", "addresses", exclusions.inProgress, "results", exclusionResults)
 		for _, address := range exclusions.inProgress {
 			if exclusionResults[address.String()] != "Success" && exclusionResults[address.String()] != "Missing" {
 				exclusions.notExcluded = append(exclusions.notExcluded, address)
@@ -676,11 +705,8 @@ func (client *cliAdminClient) VersionSupported(versionString string) (bool, erro
 		return false, nil
 	}
 
-	_, err = os.Stat(getBinaryPath("fdbcli", versionString))
+	_, err = os.Stat(getBinaryPath(fdbcliStr, versionString))
 	if err != nil {
-		if os.IsNotExist(err) {
-			return false, err
-		}
 		return false, err
 	}
 
@@ -704,10 +730,9 @@ func (client *cliAdminClient) GetProtocolVersion(version string) (string, error)
 	return protocolVersionMatch[1], nil
 }
 
-// StartBackup starts a new backup.
 func (client *cliAdminClient) StartBackup(url string, snapshotPeriodSeconds int) error {
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"start",
 			"-d",
@@ -723,7 +748,7 @@ func (client *cliAdminClient) StartBackup(url string, snapshotPeriodSeconds int)
 // StopBackup stops a backup.
 func (client *cliAdminClient) StopBackup(_ string) error {
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"discontinue",
 		},
@@ -734,7 +759,7 @@ func (client *cliAdminClient) StopBackup(_ string) error {
 // PauseBackups pauses the backups.
 func (client *cliAdminClient) PauseBackups() error {
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"pause",
 		},
@@ -745,7 +770,7 @@ func (client *cliAdminClient) PauseBackups() error {
 // ResumeBackups resumes the backups.
 func (client *cliAdminClient) ResumeBackups() error {
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"resume",
 		},
@@ -756,7 +781,7 @@ func (client *cliAdminClient) ResumeBackups() error {
 // ModifyBackup updates the backup parameters.
 func (client *cliAdminClient) ModifyBackup(snapshotPeriodSeconds int) error {
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"modify",
 			"-s",
@@ -769,7 +794,7 @@ func (client *cliAdminClient) ModifyBackup(snapshotPeriodSeconds int) error {
 // GetBackupStatus gets the status of the current backup.
 func (client *cliAdminClient) GetBackupStatus() (*fdbv1beta2.FoundationDBLiveBackupStatus, error) {
 	statusString, err := client.runCommand(cliCommand{
-		binary: "fdbbackup",
+		binary: fdbbackupStr,
 		args: []string{
 			"status",
 			"--json",
@@ -813,7 +838,7 @@ func (client *cliAdminClient) StartRestore(url string, keyRanges []fdbv1beta2.Fo
 		args = append(args, "-k", keyRangeString)
 	}
 	_, err := client.runCommand(cliCommand{
-		binary: "fdbrestore",
+		binary: fdbrestoreStr,
 		args:   args,
 	})
 	return err
@@ -822,7 +847,7 @@ func (client *cliAdminClient) StartRestore(url string, keyRanges []fdbv1beta2.Fo
 // GetRestoreStatus gets the status of the current restore.
 func (client *cliAdminClient) GetRestoreStatus() (string, error) {
 	return client.runCommand(cliCommand{
-		binary: "fdbrestore",
+		binary: fdbrestoreStr,
 		args: []string{
 			"status",
 		},
diff --git a/fdbclient/command_runner.go b/fdbclient/command_runner.go
new file mode 100644
index 00000000..472c7bc8
--- /dev/null
+++ b/fdbclient/command_runner.go
@@ -0,0 +1,64 @@
+/*
+ * command_runner.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 fdbclient
+
+import (
+	"context"
+	"github.com/go-logr/logr"
+	"os/exec"
+)
+
+// commandRunner is an interface to run commands.
+type commandRunner interface {
+	// runCommand is a method to execute a binary with the given arguments.
+	runCommand(ctx context.Context, name string, arg ...string) ([]byte, error)
+}
+
+// realCommandRunner is a struct that implements the commandRunner interface and executes commands with the exec package.
+type realCommandRunner struct {
+	log logr.Logger
+}
+
+func (runner *realCommandRunner) runCommand(ctx context.Context, name string, arg ...string) ([]byte, error) {
+	execCommand := exec.CommandContext(ctx, name, arg...)
+
+	runner.log.Info("Running command", "path", execCommand.Path, "args", execCommand.Args)
+	return execCommand.CombinedOutput()
+}
+
+// mockCommandRunner is a mock implementation of commandRunner and can be used for unit testing.
+type mockCommandRunner struct {
+	// mockedOutput is the output returned by runCommand
+	mockedOutput string
+	// mockedError is the error returned by runCommand
+	mockedError error
+	// receivedBinary will be the binary that was used to call runCommand
+	receivedBinary string
+	// receivedArgs will be the args that were used to call runCommand
+	receivedArgs []string
+}
+
+func (runner *mockCommandRunner) runCommand(_ context.Context, name string, arg ...string) ([]byte, error) {
+	runner.receivedBinary = name
+	runner.receivedArgs = arg
+
+	return []byte(runner.mockedOutput), runner.mockedError
+}
diff --git a/fdbclient/common.go b/fdbclient/common.go
index 6b7b9579..bdc13366 100644
--- a/fdbclient/common.go
+++ b/fdbclient/common.go
@@ -21,6 +21,7 @@
 package fdbclient
 
 import (
+	"errors"
 	"fmt"
 	"os"
 	"time"
@@ -74,9 +75,9 @@ func getFDBDatabase(cluster *fdbv1beta2.FoundationDBCluster) (fdb.Database, erro
 }
 
 func getValueFromDBUsingKey(cluster *fdbv1beta2.FoundationDBCluster, log logr.Logger, fdbKey string, timeout time.Duration) ([]byte, error) {
-	log.Info("Fetch values from FDB", "namespace", cluster.Namespace, "cluster", cluster.Name, "key", fdbKey)
+	log.Info("Fetch values from FDB", "key", fdbKey)
 	defer func() {
-		log.Info("Done fetching values from FDB", "namespace", cluster.Namespace, "cluster", cluster.Name, "key", fdbKey)
+		log.Info("Done fetching values from FDB", "key", fdbKey)
 	}()
 	database, err := getFDBDatabase(cluster)
 	if err != nil {
@@ -102,7 +103,8 @@ func getValueFromDBUsingKey(cluster *fdbv1beta2.FoundationDBCluster, log logr.Lo
 	})
 
 	if err != nil {
-		if fdbError, ok := err.(fdb.Error); ok {
+		var fdbError *fdb.Error
+		if errors.As(err, &fdbError) {
 			// See: https://apple.github.io/foundationdb/api-error-codes.html
 			// 1031: Operation aborted because the transaction timed out
 			if fdbError.Code == 1031 {
diff --git a/internal/error_helper.go b/internal/error_helper.go
index 5901e5b2..9ddb72cc 100644
--- a/internal/error_helper.go
+++ b/internal/error_helper.go
@@ -32,28 +32,14 @@ import (
 
 // IsNetworkError returns true if the network is a network error net.Error
 func IsNetworkError(err error) bool {
-	for err != nil {
-		if _, ok := err.(net.Error); ok {
-			return true
-		}
-
-		err = errors.Unwrap(err)
-	}
-
-	return false
+	var netError net.Error
+	return errors.As(err, &netError)
 }
 
 // IsTimeoutError returns true if the observed error was a timeout error
 func IsTimeoutError(err error) bool {
-	for err != nil {
-		if _, ok := err.(fdbv1beta2.TimeoutError); ok {
-			return true
-		}
-
-		err = errors.Unwrap(err)
-	}
-
-	return false
+	var timeoutError fdbv1beta2.TimeoutError
+	return errors.As(err, &timeoutError)
 }
 
 // IsQuotaExceeded returns true if the error returned by the Kubernetes API is a forbidden error with the error message
diff --git a/internal/locality/locality.go b/internal/locality/locality.go
new file mode 100644
index 00000000..ba56f0db
--- /dev/null
+++ b/internal/locality/locality.go
@@ -0,0 +1,391 @@
+/*
+ * locality.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 locality
+
+import (
+	"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"
+)
+
+// Info captures information about a process for the purposes of
+// choosing diverse locality.
+type Info struct {
+	// The process group ID
+	ID string
+
+	// The process's public address.
+	Address fdbv1beta2.ProcessAddress
+
+	// The locality map.
+	LocalityData map[string]string
+
+	Class fdbv1beta2.ProcessClass
+}
+
+// Sort processes by their priority and their ID.
+// 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(cluster *fdbv1beta2.FoundationDBCluster, processes []Info) {
+	// Sort the processes for ID to ensure we have a stable input
+	sort.SliceStable(processes, func(i, j int) bool {
+		p1 := cluster.GetClassCandidatePriority(processes[i].Class)
+		p2 := cluster.GetClassCandidatePriority(processes[j].Class)
+
+		// If both have the same priority sort them by the process ID
+		if p1 == p2 {
+			return processes[i].ID < processes[j].ID
+		}
+
+		// prefer processes with a higher priority
+		return p1 > p2
+	})
+}
+
+// InfoForProcess converts the process information from the JSON status
+// into locality info for selecting processes.
+func InfoForProcess(process fdbv1beta2.FoundationDBStatusProcessInfo, mainContainerTLS bool) (Info, error) {
+	addresses, err := fdbv1beta2.ParseProcessAddressesFromCmdline(process.CommandLine)
+	if err != nil {
+		return Info{}, err
+	}
+
+	var addr fdbv1beta2.ProcessAddress
+	// Iterate the addresses and set the expected address as process address
+	// e.g. if we want to use TLS set it to the tls address otherwise use the non-TLS.
+	for _, address := range addresses {
+		if address.Flags["tls"] == mainContainerTLS {
+			addr = address
+			break
+		}
+	}
+
+	return Info{
+		ID:           process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey],
+		Address:      addr,
+		LocalityData: process.Locality,
+		Class:        process.ProcessClass,
+	}, nil
+}
+
+// InfoFromSidecar converts the process information from the sidecar's
+// context into locality info for selecting processes.
+func InfoFromSidecar(cluster *fdbv1beta2.FoundationDBCluster, client podclient.FdbPodClient) (Info, error) {
+	substitutions, err := client.GetVariableSubstitutions()
+	if err != nil {
+		return Info{}, err
+	}
+
+	if substitutions == nil {
+		return Info{}, nil
+	}
+
+	// This locality information is only used during the initial cluster file generation.
+	// So it should be good to only use the first process address here.
+	// This has the implication that in the initial cluster file only the first processes will be used.
+	return Info{
+		ID:      substitutions["FDB_INSTANCE_ID"],
+		Address: cluster.GetFullAddress(substitutions["FDB_PUBLIC_IP"], 1),
+		LocalityData: map[string]string{
+			fdbv1beta2.FDBLocalityZoneIDKey:  substitutions["FDB_ZONE_ID"],
+			fdbv1beta2.FDBLocalityDNSNameKey: substitutions["FDB_DNS_NAME"],
+		},
+	}, nil
+}
+
+// notEnoughProcessesError is returned when we cannot recruit enough processes.
+type notEnoughProcessesError struct {
+	// desired defines the number of processes we wanted to recruit.
+	Desired int
+
+	// chosen defines the number of processes we were able to recruit.
+	Chosen int
+
+	// options defines the processes that we were selecting from.
+	Options []Info
+}
+
+// Error formats an error message.
+func (err notEnoughProcessesError) Error() string {
+	return fmt.Sprintf("Could only select %d processes, but %d are required", err.Chosen, err.Desired)
+}
+
+// ProcessSelectionConstraint defines constraints on how we choose processes
+// in chooseDistributedProcesses
+type ProcessSelectionConstraint struct {
+	// Fields defines the locality fields we should consider when selecting
+	// processes.
+	Fields []string
+
+	// HardLimits defines a maximum number of processes to recruit on any single
+	// value for a given locality field.
+	HardLimits map[string]int
+}
+
+// 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)
+
+	fields := constraint.Fields
+	if len(fields) == 0 {
+		fields = []string{fdbv1beta2.FDBLocalityZoneIDKey, fdbv1beta2.FDBLocalityDCIDKey}
+	}
+
+	chosenCounts := make(map[string]map[string]int, len(fields))
+	hardLimits := make(map[string]int, len(fields))
+	currentLimits := make(map[string]int, len(fields))
+
+	if constraint.HardLimits != nil {
+		for field, limit := range constraint.HardLimits {
+			hardLimits[field] = limit
+		}
+	}
+
+	for _, field := range fields {
+		chosenCounts[field] = make(map[string]int)
+		if hardLimits[field] == 0 {
+			hardLimits[field] = count
+		}
+		currentLimits[field] = 1
+	}
+
+	// Sort the processes to ensure a deterministic result
+	sortLocalities(cluster, processes)
+
+	for len(chosen) < count {
+		choseAny := false
+
+		for _, process := range processes {
+			if chosenIDs[process.ID] {
+				continue
+			}
+
+			eligible := true
+			for _, field := range fields {
+				value := process.LocalityData[field]
+				if chosenCounts[field][value] >= currentLimits[field] {
+					eligible = false
+					break
+				}
+			}
+
+			if !eligible {
+				continue
+			}
+
+			chosen = append(chosen, process)
+			chosenIDs[process.ID] = true
+
+			choseAny = true
+
+			for _, field := range fields {
+				value := process.LocalityData[field]
+				chosenCounts[field][value]++
+			}
+
+			if len(chosen) == count {
+				break
+			}
+		}
+
+		if !choseAny {
+			incrementedLimits := false
+			for indexOfField := len(fields) - 1; indexOfField >= 0; indexOfField-- {
+				field := fields[indexOfField]
+				if currentLimits[field] < hardLimits[field] {
+					currentLimits[field]++
+					incrementedLimits = true
+					break
+				}
+			}
+			if !incrementedLimits {
+				return chosen, notEnoughProcessesError{Desired: count, Chosen: len(chosen), Options: processes}
+			}
+		}
+	}
+
+	return chosen, nil
+}
+
+// GetHardLimits returns the distribution of localities.
+func GetHardLimits(cluster *fdbv1beta2.FoundationDBCluster) map[string]int {
+	if cluster.Spec.DatabaseConfiguration.UsableRegions <= 1 {
+		return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1}
+	}
+
+	// TODO (johscheuer): should we calculate that based on the number of DCs?
+	maxCoordinatorsPerDC := int(math.Floor(float64(cluster.DesiredCoordinatorCount()) / 2.0))
+
+	return map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1, fdbv1beta2.FDBLocalityDCIDKey: maxCoordinatorsPerDC}
+}
+
+// CheckCoordinatorValidity determines if the cluster's current coordinators
+// meet the fault tolerance requirements.
+//
+// The first return value will be whether the coordinators are valid.
+// The second return value will be whether the processes have their TLS flags
+// matching the cluster spec.
+// The third return value will hold any errors encountered when checking the
+// coordinators.
+func CheckCoordinatorValidity(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, status *fdbv1beta2.FoundationDBStatus, coordinatorStatus map[string]bool) (bool, bool, error) {
+	if len(coordinatorStatus) == 0 {
+		return false, false, errors.New("unable to get coordinator status")
+	}
+
+	allAddressesValid := true
+	allEligible := true
+	allUsingCorrectAddress := true
+
+	coordinatorZones := make(map[string]int, len(coordinatorStatus))
+	coordinatorDCs := make(map[string]int, len(coordinatorStatus))
+	processGroups := make(map[string]*fdbv1beta2.ProcessGroupStatus)
+	for _, processGroup := range cluster.Status.ProcessGroups {
+		processGroups[processGroup.ProcessGroupID] = processGroup
+	}
+
+	for _, process := range status.Cluster.Processes {
+		pLogger := logger.WithValues("process", process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey])
+		if process.Address.IsEmpty() {
+			pLogger.Info("Skip process with empty address")
+			continue
+		}
+
+		if process.ProcessClass == fdbv1beta2.ProcessClassTest {
+			pLogger.Info("Ignoring tester process")
+			continue
+		}
+
+		processGroupStatus := processGroups[process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]]
+		pendingRemoval := processGroupStatus != nil && processGroupStatus.IsMarkedForRemoval()
+		if processGroupStatus != nil && cluster.SkipProcessGroup(processGroupStatus) {
+			pLogger.Info("Skipping process group with pending Pod",
+				"namespace", cluster.Namespace,
+				"cluster", cluster.Name,
+				"processGroupID", process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey],
+				"class", process.ProcessClass)
+			continue
+		}
+
+		addresses, err := fdbv1beta2.ParseProcessAddressesFromCmdline(process.CommandLine)
+		if err != nil {
+			// We will end here in the error case when the address
+			// is not parsable e.g. no IP address is assigned.
+			allAddressesValid = false
+			// add: command_line
+			pLogger.Info("Could not parse address from command_line", "command_line", process.CommandLine)
+			continue
+		}
+
+		var ipAddress fdbv1beta2.ProcessAddress
+		for _, addr := range addresses {
+			if addr.Flags["tls"] == cluster.Spec.MainContainer.EnableTLS {
+				ipAddress = addr
+				break
+			}
+		}
+
+		coordinatorAddress := ""
+		_, isCoordinatorWithIP := coordinatorStatus[ipAddress.String()]
+		if isCoordinatorWithIP {
+			coordinatorAddress = ipAddress.String()
+		}
+
+		dnsName := process.Locality[fdbv1beta2.FDBLocalityDNSNameKey]
+		dnsAddress := fdbv1beta2.ProcessAddress{
+			StringAddress: dnsName,
+			Port:          ipAddress.Port,
+			Flags:         ipAddress.Flags,
+		}
+		_, isCoordinatorWithDNS := coordinatorStatus[dnsAddress.String()]
+
+		if !isCoordinatorWithDNS {
+			dnsAddress = ipAddress
+			dnsAddress.FromHostname = true
+			_, isCoordinatorWithDNS = coordinatorStatus[dnsAddress.String()]
+		}
+
+		if isCoordinatorWithDNS {
+			coordinatorAddress = dnsAddress.String()
+		}
+
+		if coordinatorAddress != "" && !process.Excluded && !pendingRemoval {
+			coordinatorStatus[coordinatorAddress] = true
+		}
+
+		if coordinatorAddress != "" {
+			coordinatorZones[process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]]++
+			coordinatorDCs[process.Locality[fdbv1beta2.FDBLocalityDCIDKey]]++
+
+			if !cluster.IsEligibleAsCandidate(process.ProcessClass) {
+				pLogger.Info("Process class of process is not eligible as coordinator", "class", process.ProcessClass, "address", coordinatorAddress)
+				allEligible = false
+			}
+
+			useDNS := cluster.UseDNSInClusterFile() && dnsName != ""
+			if (isCoordinatorWithIP && useDNS) || (isCoordinatorWithDNS && !useDNS) {
+				pLogger.Info("Coordinator is not using the correct address type", "coordinatorList", coordinatorStatus, "address", coordinatorAddress, "expectingDNS", useDNS, "usingDNS", isCoordinatorWithDNS)
+				allUsingCorrectAddress = false
+			}
+		}
+
+		if ipAddress.IPAddress == nil {
+			pLogger.Info("Process has invalid IP address", "addresses", addresses)
+			allAddressesValid = false
+		}
+	}
+
+	desiredCount := cluster.DesiredCoordinatorCount()
+	hasEnoughZones := len(coordinatorZones) == desiredCount
+	if !hasEnoughZones {
+		logger.Info("Cluster does not have coordinators in the correct number of zones", "desiredCount", desiredCount, "coordinatorZones", coordinatorZones)
+	}
+
+	var maxCoordinatorsPerDC int
+	hasEnoughDCs := true
+	if cluster.Spec.DatabaseConfiguration.UsableRegions > 1 {
+		maxCoordinatorsPerDC = int(math.Floor(float64(desiredCount) / 2.0))
+
+		for dc, count := range coordinatorDCs {
+			if count > maxCoordinatorsPerDC {
+				logger.Info("Cluster has too many coordinators in a single DC", "DC", dc, "count", count, "max", maxCoordinatorsPerDC)
+				hasEnoughDCs = false
+			}
+		}
+	}
+
+	allHealthy := true
+	for address, healthy := range coordinatorStatus {
+		if !healthy {
+			allHealthy = false
+			logger.Info("Cluster has an unhealthy coordinator", "address", address)
+		}
+	}
+
+	return hasEnoughDCs && hasEnoughZones && allHealthy && allUsingCorrectAddress && allEligible, allAddressesValid, nil
+}
diff --git a/internal/restarts/restarts.go b/internal/restarts/restarts.go
new file mode 100644
index 00000000..22b7f2c2
--- /dev/null
+++ b/internal/restarts/restarts.go
@@ -0,0 +1,44 @@
+/*
+ * restarts.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 restarts
+
+import fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+
+// GetFilterConditions returns the filter conditions to get the processes that should be restarted.
+func GetFilterConditions(cluster *fdbv1beta2.FoundationDBCluster) map[fdbv1beta2.ProcessGroupConditionType]bool {
+	if !cluster.IsBeingUpgraded() {
+		// If we don't upgrade our cluster we can ignore all process groups that are not reachable and therefore will
+		// not get any ConfigMap updates.
+		return map[fdbv1beta2.ProcessGroupConditionType]bool{
+			fdbv1beta2.IncorrectCommandLine: true,
+			fdbv1beta2.IncorrectPodSpec:     false,
+			fdbv1beta2.SidecarUnreachable:   false,
+		}
+	}
+
+	// If we perform an upgrade we have to wait until all processes are
+	// ready to be restarted. Therefore we can't ignore process groups with the SidecarUnreachable condition.
+	// We could be less restrictive here in cases where we perform a version compatible upgrade.m
+	return map[fdbv1beta2.ProcessGroupConditionType]bool{
+		fdbv1beta2.IncorrectCommandLine: true,
+		fdbv1beta2.IncorrectPodSpec:     false,
+	}
+}
diff --git a/mock-kubernetes-client/client/client.go b/mock-kubernetes-client/client/client.go
index deb49a29..aa57bcc8 100644
--- a/mock-kubernetes-client/client/client.go
+++ b/mock-kubernetes-client/client/client.go
@@ -46,13 +46,48 @@ type MockClient struct {
 
 	// scheme will be used to initialize or reset the new fake client
 	scheme *runtime.Scheme
+
+	// createHooks allow to inject custom logic to the creation of objects. See serviceCreateHook and podCreateHook as
+	// examples
+	createHooks []func(ctx context.Context, client *MockClient, object ctrlClient.Object) error
 }
 
 // NewMockClient creates a new MockClient.
-func NewMockClient(scheme *runtime.Scheme) *MockClient {
+func NewMockClient(scheme *runtime.Scheme, hooks ...func(_ context.Context, client *MockClient, object ctrlClient.Object) error) *MockClient {
+	serviceCreateHook := func(_ context.Context, client *MockClient, object ctrlClient.Object) error {
+		svc, isSvc := object.(*corev1.Service)
+		if !isSvc {
+			return nil
+		}
+
+		if svc.Spec.ClusterIP == "" {
+			svc.Spec.ClusterIP = client.generateIP()
+		}
+
+		return nil
+	}
+
+	podCreateHook := func(_ context.Context, client *MockClient, object ctrlClient.Object) error {
+		pod, isPod := object.(*corev1.Pod)
+		if !isPod {
+			return nil
+		}
+
+		v4Address := client.generatePodIPv4()
+		pod.Status.PodIP = v4Address
+		pod.Status.PodIPs = []corev1.PodIP{{IP: v4Address}, {IP: client.generatePodIPv6()}}
+
+		if pod.Status.Phase == "" {
+			pod.Status.Phase = corev1.PodRunning
+		}
+
+		return nil
+	}
+
 	return &MockClient{
-		fakeClient: fake.NewClientBuilder().WithScheme(scheme).Build(),
-		scheme:     scheme,
+		fakeClient:  fake.NewClientBuilder().WithScheme(scheme).Build(),
+		scheme:      scheme,
+		createHooks: append(hooks, serviceCreateHook, podCreateHook),
 	}
 }
 
@@ -84,25 +119,11 @@ func (client *MockClient) Create(ctx context.Context, object ctrlClient.Object,
 	object.SetGeneration(object.GetGeneration() + 1)
 	object.SetUID(uuid.NewUUID())
 
-	svc, isSvc := object.(*corev1.Service)
-	if isSvc {
-		if svc.Spec.ClusterIP == "" {
-			svc.Spec.ClusterIP = client.generateIP()
-		}
-
-		object = svc.DeepCopyObject().(ctrlClient.Object)
-	}
-
-	pod, isPod := object.(*corev1.Pod)
-	if isPod {
-		v4Address := client.generatePodIPv4()
-		pod.Status.PodIP = v4Address
-		pod.Status.PodIPs = []corev1.PodIP{{IP: v4Address}, {IP: client.generatePodIPv6()}}
-
-		if pod.Status.Phase == "" {
-			pod.Status.Phase = corev1.PodRunning
+	for _, hook := range client.createHooks {
+		err := hook(ctx, client, object)
+		if err != nil {
+			return err
 		}
-		object = pod.DeepCopyObject().(ctrlClient.Object)
 	}
 
 	return client.fakeClient.Create(ctx, object, options...)
diff --git a/scripts/.gitignore b/scripts/.gitignore
new file mode 100644
index 00000000..7cc402c7
--- /dev/null
+++ b/scripts/.gitignore
@@ -0,0 +1,1 @@
+*.kubeconfig
diff --git a/scripts/cluster1.yml b/scripts/cluster1.yml
new file mode 100644
index 00000000..8a340956
--- /dev/null
+++ b/scripts/cluster1.yml
@@ -0,0 +1,8 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+networking:
+  podSubnet: "10.10.0.0/16"
+  serviceSubnet: "10.11.0.0/16"
+nodes:
+  - role: control-plane
+    image: kindest/node:${KUBE_VERSION}
diff --git a/scripts/cluster2.yml b/scripts/cluster2.yml
new file mode 100644
index 00000000..ed5667da
--- /dev/null
+++ b/scripts/cluster2.yml
@@ -0,0 +1,8 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+networking:
+  podSubnet: "10.12.0.0/16"
+  serviceSubnet: "10.13.0.0/16"
+nodes:
+  - role: control-plane
+    image: kindest/node:${KUBE_VERSION}
diff --git a/scripts/cluster3.yml b/scripts/cluster3.yml
new file mode 100644
index 00000000..62232f8b
--- /dev/null
+++ b/scripts/cluster3.yml
@@ -0,0 +1,8 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+networking:
+  podSubnet: "10.14.0.0/16"
+  serviceSubnet: "10.15.0.0/16"
+nodes:
+  - role: control-plane
+    image: kindest/node:${KUBE_VERSION}
diff --git a/scripts/cluster4.yml b/scripts/cluster4.yml
new file mode 100644
index 00000000..299f8be9
--- /dev/null
+++ b/scripts/cluster4.yml
@@ -0,0 +1,8 @@
+kind: Cluster
+apiVersion: kind.x-k8s.io/v1alpha4
+networking:
+  podSubnet: "10.16.0.0/16"
+  serviceSubnet: "10.17.0.0/16"
+nodes:
+  - role: control-plane
+    image: kindest/node:${KUBE_VERSION}


Test patch:
diff --git a/api/v1beta2/foundationdb_status_test.go b/api/v1beta2/foundationdb_status_test.go
index 843a087f..8e566319 100644
--- a/api/v1beta2/foundationdb_status_test.go
+++ b/api/v1beta2/foundationdb_status_test.go
@@ -71,6 +71,7 @@ var _ = Describe("FoundationDBStatus", func() {
 				},
 				Cluster: FoundationDBStatusClusterInfo{
 					IncompatibleConnections: []string{},
+					ConnectionString:        "sample_cluster:JLjCjL6Vp3kWoIfHJeDZMhYqPBb1bIZr@10.1.38.94:4501,10.1.38.102:4501,10.1.38.104:4501",
 					FaultTolerance: FaultTolerance{
 						MaxZoneFailuresWithoutLosingAvailability: 1,
 						MaxZoneFailuresWithoutLosingData:         1,
@@ -464,6 +465,7 @@ var _ = Describe("FoundationDBStatus", func() {
 	When("parsing the status json with a 7.1.0-rc1 cluster", func() {
 		status := FoundationDBStatusClusterInfo{
 			IncompatibleConnections: []string{},
+			ConnectionString:        "test_cluster:aHeD9ocNXOUxi0dyzU3k7Bhg53SpyrBV@10.1.18.253:4501,10.1.18.254:4501,10.1.19.0:4501",
 			DatabaseConfiguration: DatabaseConfiguration{
 				RedundancyMode:  "double",
 				StorageEngine:   StorageEngineSSD2,
diff --git a/config/test-certs/generate_secrets.bash b/config/test-certs/generate_secrets.bash
index de6dfd07..809b36cb 100755
--- a/config/test-certs/generate_secrets.bash
+++ b/config/test-certs/generate_secrets.bash
@@ -1,4 +1,4 @@
-#! /bin/bash
+#!/usr/bin/env bash
 
 set -eu
 
@@ -23,6 +23,9 @@ set -eu
 
 # This script generates secrets with test certs for use in local testing.
 
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &> /dev/null && pwd)
+cd "${SCRIPT_DIR}"
+
 kubectl delete secrets -l app=fdb-kubernetes-operator
-kubectl create secret tls fdb-kubernetes-operator-secrets --key=config/test-certs/key.pem --cert=config/test-certs/cert.pem
+kubectl create secret tls fdb-kubernetes-operator-secrets --key=${SCRIPT_DIR}/key.pem --cert=${SCRIPT_DIR}/cert.pem
 kubectl label secret fdb-kubernetes-operator-secrets app=fdb-kubernetes-operator
diff --git a/config/tests/multi_dc/Readme.md b/config/tests/multi_dc/Readme.md
new file mode 100644
index 00000000..c68332f0
--- /dev/null
+++ b/config/tests/multi_dc/Readme.md
@@ -0,0 +1,20 @@
+# Multi DC example
+
+Ensure that you ran the `setup_e2e.sh` script and setup the environment variables for the clusters.
+This setup is currently meant to run manually. In the future this will be rewritten to be a real e2e test.
+
+## Create the multi-dc cluster
+
+This will bringup a FDB cluster running across 3 different Kubernetes clusters in a multi-region configuration.
+
+```bash
+./create.bash
+```
+
+## Delete
+
+This will remove all created resources:
+
+```bash
+./bash.bash
+```
diff --git a/config/tests/multi_dc/apply.bash b/config/tests/multi_dc/apply.bash
deleted file mode 100644
index 2ffc13a8..00000000
--- a/config/tests/multi_dc/apply.bash
+++ /dev/null
@@ -1,28 +0,0 @@
-#! /bin/bash
-
-DIR="${BASH_SOURCE%/*}"
-
-. $DIR/functions.bash
-
-
-originalConnectionString=$(getConnectionString test-cluster-dc1)
-
-function applyAll() {
-	applyFile $DIR/final.yaml dc1 $1
-	applyFile $DIR/final.yaml dc2 $1
-	applyFile $DIR/final.yaml dc3 $1
-
-	checkReconciliationLoop test-cluster-dc1
-	checkReconciliationLoop test-cluster-dc2
-	checkReconciliationLoop test-cluster-dc3
-}
-
-applyAll $originalConnectionString
-
-for dc in dc1 dc2 dc3; do
-	connectionString=$(getConnectionString test-cluster-$dc)
-	if [[ $connectionString != $originalConnectionString ]]; then
-		applyAll $connectionString
-		originalConnectionString=$connectionString
-	fi
-done
diff --git a/config/tests/multi_dc/create.bash b/config/tests/multi_dc/create.bash
index e0fae0b5..797ca777 100755
--- a/config/tests/multi_dc/create.bash
+++ b/config/tests/multi_dc/create.bash
@@ -1,4 +1,6 @@
-#! /bin/bash
+#!/usr/bin/env bash
+
+set -eu
 
 # This directory provides an example of creating a cluster using the multi-KC
 # replication topology.
@@ -15,14 +17,24 @@ DIR="${BASH_SOURCE%/*}"
 
 . $DIR/functions.bash
 
-applyFile $DIR/stage_1.yaml dc1 '""'
-checkReconciliationLoop test-cluster-dc1
-connectionString=$(getConnectionString test-cluster-dc1)
+# To test a multi-region FDB cluster setup we need to have 3 Kubernetes clusters
+cluster1=${CLUSTER1:-cluster1}
+cluster2=${CLUSTER2:-cluster2}
+cluster3=${CLUSTER3:-cluster3}
+
+# For all the clusters we have to create tha according kubeconfig
+kubeconfig1=${KUBECONFIG1:-"${cluster1}.kubeconfig"}
+kubeconfig2=${KUBECONFIG2:-"${cluster2}.kubeconfig"}
+kubeconfig3=${KUBECONFIG3:-"${cluster3}.kubeconfig"}
+
+applyFile "${DIR}/stage_1.yaml" dc1 '""' "${kubeconfig1}"
+checkReconciliationLoop test-cluster-dc1 "${kubeconfig1}"
+connectionString=$(getConnectionString test-cluster-dc1 "${kubeconfig1}")
 
-applyFile $DIR/final.yaml dc1 $connectionString
-applyFile $DIR/final.yaml dc2 $connectionString
-applyFile $DIR/final.yaml dc3 $connectionString
+applyFile "${DIR}/final.yaml" dc1 "${connectionString}" "${kubeconfig1}"
+applyFile "${DIR}/final.yaml" dc2 "${connectionString}" "${kubeconfig2}"
+applyFile "${DIR}/final.yaml" dc3 "${connectionString}" "${kubeconfig3}"
 
-checkReconciliationLoop test-cluster-dc1
-checkReconciliationLoop test-cluster-dc2
-checkReconciliationLoop test-cluster-dc3
+checkReconciliationLoop test-cluster-dc1 "${kubeconfig1}"
+checkReconciliationLoop test-cluster-dc2 "${kubeconfig2}"
+checkReconciliationLoop test-cluster-dc3 "${kubeconfig3}"
diff --git a/config/tests/multi_dc/delete.bash b/config/tests/multi_dc/delete.bash
index 0dda852b..99e53ff2 100755
--- a/config/tests/multi_dc/delete.bash
+++ b/config/tests/multi_dc/delete.bash
@@ -1,3 +1,17 @@
-#! /bin/bash
+#!/usr/bin/env bash
 
-kubectl delete fdb -l cluster-group=test-cluster
+set -eu
+
+# To test a multi-region FDB cluster setup we need to have 3 Kubernetes clusters
+cluster1=${CLUSTER1:-cluster1}
+cluster2=${CLUSTER2:-cluster2}
+cluster3=${CLUSTER3:-cluster3}
+
+# For all the clusters we have to create tha according kubeconfig
+kubeconfig1=${KUBECONFIG1:-"${cluster1}.kubeconfig"}
+kubeconfig2=${KUBECONFIG2:-"${cluster2}.kubeconfig"}
+kubeconfig3=${KUBECONFIG3:-"${cluster3}.kubeconfig"}
+
+kubectl --kubeconfig "${kubeconfig1}" delete fdb -l cluster-group=test-cluster
+kubectl --kubeconfig "${kubeconfig2}" delete fdb -l cluster-group=test-cluster
+kubectl --kubeconfig "${kubeconfig3}" delete fdb -l cluster-group=test-cluster
diff --git a/config/tests/multi_dc/final.yaml b/config/tests/multi_dc/final.yaml
index b9e361b7..39c5e980 100644
--- a/config/tests/multi_dc/final.yaml
+++ b/config/tests/multi_dc/final.yaml
@@ -7,21 +7,18 @@ apiVersion: apps.foundationdb.org/v1beta2
 kind: FoundationDBCluster
 metadata:
   labels:
-    controller-tools.k8s.io: "1.0"
     cluster-group: test-cluster
   name: test-cluster-$dc
 spec:
-  version: 6.2.30
+  version: 7.1.25
   faultDomain:
     key: foundationdb.org/none
-  instanceIDPrefix: $dc
+  processGroupIDPrefix: $dc
   dataCenter: $dc
   processCounts:
     stateless: -1
     log: $logCount
   seedConnectionString: $connectionString
-  customParameters:
-    - "knob_disable_posix_kernel_aio=1"
   databaseConfiguration:
     redundancy_mode: "double"
     usable_regions: 2
@@ -48,6 +45,8 @@ spec:
         satellite_logs: 3
   processes:
     general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
       volumeClaimTemplate:
         spec:
           resources:
diff --git a/config/tests/multi_dc/functions.bash b/config/tests/multi_dc/functions.bash
index ce9921de..15dd3071 100644
--- a/config/tests/multi_dc/functions.bash
+++ b/config/tests/multi_dc/functions.bash
@@ -1,22 +1,21 @@
 function applyFile() {
-	path=$1
-	dc=$2
-	connectionString=$3
+	dc=${2}
+
 	if [[ "${dc}" == "dc3" ]]; then
 		logCount=0
 	else
 		logCount=-1
 	fi
-	
-  sed -e "s/\$dc/$dc/" "${path}" | sed -e "s/\$logCount/$logCount/" | sed -e "s/\$connectionString/$connectionString/" | kubectl apply -f -
+
+	dc="${dc}" connectionString="${3}" logCount="${logCount}" envsubst < "${1}"| kubectl --kubeconfig "${4}" apply -f -
 }
 
 function checkReconciliation() {
 	clusterName=$1
 
-	generationsOutput=$(kubectl get fdb "${clusterName}" -o jsonpath='{.metadata.generation} {.status.generations.reconciled}')
+	generationsOutput=$(kubectl --kubeconfig "${2}" get fdb "${clusterName}" -o jsonpath='{.metadata.generation} {.status.generations.reconciled}')
 	read -ra generations <<< "${generationsOutput}"
-	if [[ ("${generations[1]}" != "")  && ("${generations[0]}" == "${generations[1]}") ]]; then
+	if [[ ("${#generations[@]}" -ge 2) && ("${generations[0]}" == "${generations[1]}") ]]; then
 		return 1
 	else
 		echo "Latest generations for $clusterName: $generationsOutput"
@@ -25,20 +24,12 @@ function checkReconciliation() {
 }
 
 function getConnectionString() {
-	clusterName=$1
-
-	kubectl get fdb "${clusterName}" -o jsonpath='{.status.connectionString}'
+	kubectl --kubeconfig "${2}" get fdb "${1}" -o jsonpath='{.status.connectionString}'
 }
 
 function checkReconciliationLoop() {
-	reconciled=0
-	name=$1
-	while [ $reconciled -ne 1 ] ; do
-		checkReconciliation "${name}"
-		reconciled=$?
-		if [[ $reconciled -ne 1 ]]; then
-			echo "Waiting for reconciliation"
-			sleep 5
-		fi
+	while checkReconciliation "${1}" "${2}"; do
+		echo "Waiting for reconciliation"
+		sleep 5
 	done
 }
diff --git a/config/tests/multi_dc/stage_1.yaml b/config/tests/multi_dc/stage_1.yaml
index 4f90a8e7..5572f842 100644
--- a/config/tests/multi_dc/stage_1.yaml
+++ b/config/tests/multi_dc/stage_1.yaml
@@ -10,17 +10,15 @@ metadata:
     cluster-group: test-cluster
   name: test-cluster-$dc
 spec:
-  version: 6.2.30
+  version: 7.1.25
   faultDomain:
     key: foundationdb.org/none
-  instanceIDPrefix: $dc
+  processGroupIDPrefix: $dc
   dataCenter: $dc
   processCounts:
     stateless: -1
     log: $logCount
   seedConnectionString: $connectionString
-  customParameters:
-    - "knob_disable_posix_kernel_aio=1"
   databaseConfiguration:
     redundancy_mode: "double"
     usable_regions: 1
@@ -30,6 +28,8 @@ spec:
             priority: 1
   processes:
     general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
       volumeClaimTemplate:
         spec:
           resources:
diff --git a/config/tests/multi_kc/apply.bash b/config/tests/multi_kc/apply.bash
deleted file mode 100644
index 75209243..00000000
--- a/config/tests/multi_kc/apply.bash
+++ /dev/null
@@ -1,27 +0,0 @@
-#! /bin/bash
-
-DIR="${BASH_SOURCE%/*}"
-
-. $DIR/functions.bash
-
-originalConnectionString=$(getConnectionString test-cluster-dc1)
-
-function applyAll() {
-	applyFile $DIR/final.yaml dc1 $1
-	applyFile $DIR/final.yaml dc2 $1
-	applyFile $DIR/final.yaml dc3 $1
-
-	checkReconciliationLoop test-cluster-dc1
-	checkReconciliationLoop test-cluster-dc2
-	checkReconciliationLoop test-cluster-dc3
-}
-
-applyAll $originalConnectionString
-
-for dc in dc1 dc2 dc3; do
-	connectionString=$(getConnectionString test-cluster-$dc)
-	if [[ $connectionString != $originalConnectionString ]]; then
-		applyAll $connectionString
-		originalConnectionString=$connectionString
-	fi
-done
diff --git a/config/tests/multi_kc/create.bash b/config/tests/multi_kc/create.bash
index e0fae0b5..cf35cb65 100755
--- a/config/tests/multi_kc/create.bash
+++ b/config/tests/multi_kc/create.bash
@@ -1,4 +1,6 @@
-#! /bin/bash
+#!/usr/bin/env bash
+
+set -eu
 
 # This directory provides an example of creating a cluster using the multi-KC
 # replication topology.
@@ -15,14 +17,24 @@ DIR="${BASH_SOURCE%/*}"
 
 . $DIR/functions.bash
 
-applyFile $DIR/stage_1.yaml dc1 '""'
-checkReconciliationLoop test-cluster-dc1
-connectionString=$(getConnectionString test-cluster-dc1)
+# To test a multi-region FDB cluster setup we need to have 3 Kubernetes clusters
+cluster1=${CLUSTER1:-cluster1}
+cluster2=${CLUSTER2:-cluster2}
+cluster3=${CLUSTER3:-cluster3}
+
+# For all the clusters we have to create tha according kubeconfig
+kubeconfig1=${KUBECONFIG1:-"${cluster1}.kubeconfig"}
+kubeconfig2=${KUBECONFIG2:-"${cluster2}.kubeconfig"}
+kubeconfig3=${KUBECONFIG3:-"${cluster3}.kubeconfig"}
+
+applyFile $DIR/stage_1.yaml dc1 '""' "${kubeconfig1}"
+checkReconciliationLoop test-cluster-dc1 "${kubeconfig1}"
+connectionString=$(getConnectionString test-cluster-dc1 "${kubeconfig1}")
 
-applyFile $DIR/final.yaml dc1 $connectionString
-applyFile $DIR/final.yaml dc2 $connectionString
-applyFile $DIR/final.yaml dc3 $connectionString
+applyFile $DIR/final.yaml dc1 $connectionString "${kubeconfig1}"
+applyFile $DIR/final.yaml dc2 $connectionString "${kubeconfig2}"
+applyFile $DIR/final.yaml dc3 $connectionString "${kubeconfig3}"
 
-checkReconciliationLoop test-cluster-dc1
-checkReconciliationLoop test-cluster-dc2
-checkReconciliationLoop test-cluster-dc3
+checkReconciliationLoop test-cluster-dc1 "${kubeconfig1}"
+checkReconciliationLoop test-cluster-dc2 "${kubeconfig2}"
+checkReconciliationLoop test-cluster-dc3 "${kubeconfig3}"
diff --git a/config/tests/multi_kc/delete.bash b/config/tests/multi_kc/delete.bash
index 0dda852b..b6683574 100644
--- a/config/tests/multi_kc/delete.bash
+++ b/config/tests/multi_kc/delete.bash
@@ -1,3 +1,17 @@
-#! /bin/bash
+#!/usr/bine/env bash
 
-kubectl delete fdb -l cluster-group=test-cluster
+set -eu
+
+# To test a multi-region FDB cluster setup we need to have 3 Kubernetes clusters
+cluster1=${CLUSTER1:-cluster1}
+cluster2=${CLUSTER2:-cluster2}
+cluster3=${CLUSTER3:-cluster3}
+
+# For all the clusters we have to create tha according kubeconfig
+kubeconfig1=${KUBECONFIG1:-"${cluster1}.kubeconfig"}
+kubeconfig2=${KUBECONFIG2:-"${cluster2}.kubeconfig"}
+kubeconfig3=${KUBECONFIG3:-"${cluster3}.kubeconfig"}
+
+kubectl --kubeconfig "${kubeconfig1}" delete fdb -l cluster-group=test-cluster
+kubectl --kubeconfig "${kubeconfig2}" delete fdb -l cluster-group=test-cluster
+kubectl --kubeconfig "${kubeconfig3}" delete fdb -l cluster-group=test-cluster
diff --git a/config/tests/multi_kc/final.yaml b/config/tests/multi_kc/final.yaml
index 291b1516..23c0cd4d 100644
--- a/config/tests/multi_kc/final.yaml
+++ b/config/tests/multi_kc/final.yaml
@@ -10,22 +10,22 @@ metadata:
     cluster-group: test-cluster
   name: test-cluster-$zone
 spec:
-  version: 6.2.30
+  version: 7.1.25
   faultDomain:
     key: foundationdb.org/kubernetes-cluster
     value: $zone
     zoneIndex: $zoneIndex
     zoneCount: 3
-  instanceIDPrefix: $zone
+  processGroupIDPrefix: $zone
   processCounts:
     stateless: -1
   seedConnectionString: $connectionString
-  customParameters:
-    - "knob_disable_posix_kernel_aio=1"
   databaseConfiguration:
     redundancy_mode: "double"
   processes:
     general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
       volumeClaimTemplate:
         spec:
           resources:
diff --git a/config/tests/multi_kc/functions.bash b/config/tests/multi_kc/functions.bash
index c7c92576..ad760ffa 100644
--- a/config/tests/multi_kc/functions.bash
+++ b/config/tests/multi_kc/functions.bash
@@ -1,18 +1,14 @@
 function applyFile() {
-	path=$1
 	zone=$2
-	connectionString=$3
-	zoneIndex=${zone:2:1}
-	
-	sed -e "s/\$zoneIndex/$zoneIndex/" $path | sed -e "s/\$zone/$zone/" | sed -e "s/\$connectionString/$connectionString/" | kubectl apply -f -
+	zone="${zone}" zoneIndex="${zone:2:1}" connectionString="${3}" envsubst < "${1}" | kubectl --kubeconfig "${4}" apply -f -
 }
 
 function checkReconciliation() {
 	clusterName=$1
 
-	generationsOutput=$(kubectl get fdb $clusterName -o jsonpath='{.metadata.generation} {.status.generations.reconciled}')
-	read -ra generations <<< $generationsOutput
-	if [[ ("${generations[1]}" != "")  && ("${generations[0]}" == "${generations[1]}") ]]; then
+	generationsOutput=$(kubectl --kubeconfig "${2}" get fdb "${clusterName}" -o jsonpath='{.metadata.generation} {.status.generations.reconciled}')
+	read -ra generations <<< "${generationsOutput}"
+	if [[ ("${#generations[@]}" -ge 2) && ("${generations[0]}" == "${generations[1]}") ]]; then
 		return 1
 	else
 		echo "Latest generations for $clusterName: $generationsOutput"
@@ -20,23 +16,13 @@ function checkReconciliation() {
 	fi
 }
 
-
 function getConnectionString() {
-	clusterName=$1
-
-	kubectl get fdb $clusterName -o jsonpath='{.status.connectionString}'
+	kubectl --kubeconfig "${2}" get fdb "${1}" -o jsonpath='{.status.connectionString}'
 }
 
-
 function checkReconciliationLoop() {
-	reconciled=0
-	name=$1
-	while [ $reconciled -ne 1 ] ; do
-		checkReconciliation $name
-		reconciled=$?
-		if [[ $reconciled -ne 1 ]]; then
-			echo "Waiting for reconcilliation"
-			sleep 5
-		fi
+	while checkReconciliation "${1}" "${2}"; do
+		echo "Waiting for reconciliation"
+		sleep 5
 	done
 }
diff --git a/config/tests/multi_kc/stage_1.yaml b/config/tests/multi_kc/stage_1.yaml
index 1aa046e0..2d323b56 100644
--- a/config/tests/multi_kc/stage_1.yaml
+++ b/config/tests/multi_kc/stage_1.yaml
@@ -10,22 +10,22 @@ metadata:
     cluster-group: test-cluster
   name: test-cluster-$zone
 spec:
-  version: 6.2.30
+  version: 7.1.25
   faultDomain:
     key: foundationdb.org/kubernetes-cluster
     value: $zone
     zoneIndex: $zoneIndex
     zoneCount: 1
-  instanceIDPrefix: $zone
+  processGroupIDPrefix: $zone
   processCounts:
     stateless: -1
   seedConnectionString: $connectionString
-  customParameters:
-    - "knob_disable_posix_kernel_aio=1"
   databaseConfiguration:
     redundancy_mode: "single"
   processes:
     general:
+      customParameters:
+      - "knob_disable_posix_kernel_aio=1"
       volumeClaimTemplate:
         spec:
           resources:
diff --git a/controllers/bounce_processes_test.go b/controllers/bounce_processes_test.go
index 9b50f89a..81fa10b1 100644
--- a/controllers/bounce_processes_test.go
+++ b/controllers/bounce_processes_test.go
@@ -25,6 +25,8 @@ import (
 	"fmt"
 	"time"
 
+	"k8s.io/utils/pointer"
+
 	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -41,21 +43,17 @@ var _ = Describe("bounceProcesses", func() {
 
 	BeforeEach(func() {
 		cluster = internal.CreateDefaultCluster()
-		disabled := false
-		cluster.Spec.LockOptions.DisableLocks = &disabled
-		err = setupClusterForTest(cluster)
-		Expect(err).NotTo(HaveOccurred())
+		cluster.Spec.LockOptions.DisableLocks = pointer.Bool(false)
+		Expect(setupClusterForTest(cluster)).NotTo(HaveOccurred())
 
 		adminClient, err = newMockAdminClientUncast(cluster, k8sClient)
 		Expect(err).NotTo(HaveOccurred())
 
 		lockClient = newMockLockClientUncast(cluster)
-
 	})
 
 	JustBeforeEach(func() {
 		requeue = bounceProcesses{}.reconcile(context.TODO(), clusterReconciler, cluster)
-		Expect(err).NotTo(HaveOccurred())
 	})
 
 	Context("with a reconciled cluster", func() {
@@ -93,8 +91,7 @@ var _ = Describe("bounceProcesses", func() {
 				}
 			}
 
-			Expect(len(adminClient.KilledAddresses)).To(Equal(len(addresses)))
-			Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+			Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
 		})
 	})
 
@@ -123,8 +120,7 @@ var _ = Describe("bounceProcesses", func() {
 				}
 			}
 
-			Expect(len(adminClient.KilledAddresses)).To(Equal(len(addresses)))
-			Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+			Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
 		})
 	})
 
@@ -212,8 +208,7 @@ var _ = Describe("bounceProcesses", func() {
 	Context("with multiple storage servers per pod", func() {
 		BeforeEach(func() {
 			cluster.Spec.StorageServersPerPod = 2
-			err = k8sClient.Update(context.TODO(), cluster)
-			Expect(err).NotTo(HaveOccurred())
+			Expect(k8sClient.Update(context.TODO(), cluster)).NotTo(HaveOccurred())
 			result, err := reconcileCluster(cluster)
 			Expect(err).NotTo(HaveOccurred())
 			Expect(result.Requeue).To(BeFalse())
@@ -241,8 +236,7 @@ var _ = Describe("bounceProcesses", func() {
 					addresses = append(addresses, fmt.Sprintf("%s:4501", address), fmt.Sprintf("%s:4503", address))
 				}
 			}
-			Expect(len(adminClient.KilledAddresses)).To(BeNumerically("==", len(addresses)))
-			Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+			Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
 		})
 	})
 
@@ -265,8 +259,7 @@ var _ = Describe("bounceProcesses", func() {
 					addresses = append(addresses, fmt.Sprintf("%s:4501", address))
 				}
 			}
-			Expect(len(adminClient.KilledAddresses)).To(BeNumerically("==", len(addresses)))
-			Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+			Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
 		})
 
 		It("should update the running version in the status", func() {
@@ -333,8 +326,7 @@ var _ = Describe("bounceProcesses", func() {
 						}
 					}
 					addresses = append(addresses, "1.2.3.4:4501")
-					Expect(len(adminClient.KilledAddresses)).To(BeNumerically("==", len(addresses)))
-					Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+					Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
 				})
 			})
 
@@ -356,8 +348,7 @@ var _ = Describe("bounceProcesses", func() {
 
 			Context("with locks disabled", func() {
 				BeforeEach(func() {
-					disabled := true
-					cluster.Spec.LockOptions.DisableLocks = &disabled
+					cluster.Spec.LockOptions.DisableLocks = pointer.Bool(true)
 				})
 
 				It("should requeue", func() {
@@ -371,8 +362,30 @@ var _ = Describe("bounceProcesses", func() {
 							addresses = append(addresses, fmt.Sprintf("%s:4501", address))
 						}
 					}
-					Expect(len(adminClient.KilledAddresses)).To(BeNumerically("==", len(addresses)))
-					Expect(adminClient.KilledAddresses).To(ContainElements(addresses))
+					Expect(adminClient.KilledAddresses).To(ConsistOf(addresses))
+				})
+
+				It("should not submit pending upgrade information", func() {
+					Expect(lockClient.pendingUpgrades).To(BeEmpty())
+				})
+			})
+
+			When("one process is missing", func() {
+				BeforeEach(func() {
+					missingProcessGroup := cluster.Status.ProcessGroups[0]
+					adminClient.MockMissingProcessGroup(missingProcessGroup.ProcessGroupID, true)
+
+					missingProcessGroup.UpdateCondition(fdbv1beta2.IncorrectCommandLine, false, cluster.Status.ProcessGroups, missingProcessGroup.ProcessGroupID)
+					missingProcessGroup.UpdateCondition(fdbv1beta2.MissingProcesses, true, cluster.Status.ProcessGroups, missingProcessGroup.ProcessGroupID)
+				})
+
+				It("should requeue", func() {
+					Expect(requeue).NotTo(BeNil())
+					Expect(requeue.message).To(Equal("expected 17 processes, got 16 processes ready to restart"))
+				})
+
+				It("shouldn't kill any processes", func() {
+					Expect(adminClient.KilledAddresses).To(BeEmpty())
 				})
 
 				It("should not submit pending upgrade information", func() {
diff --git a/controllers/change_coordinators_test.go b/controllers/change_coordinators_test.go
index 2fbf6faa..2d279ccf 100644
--- a/controllers/change_coordinators_test.go
+++ b/controllers/change_coordinators_test.go
@@ -27,6 +27,9 @@ import (
 	"net"
 	"strings"
 
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal/locality"
+	"github.com/go-logr/logr"
+
 	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/utils/pointer"
@@ -64,14 +67,14 @@ var _ = Describe("Change coordinators", func() {
 	Describe("selectCoordinators", func() {
 		Context("with a single FDB cluster", func() {
 			var status *fdbv1beta2.FoundationDBStatus
-			var candidates []localityInfo
+			var candidates []locality.Info
 
 			JustBeforeEach(func() {
 				var err error
 				status, err = adminClient.GetStatus()
 				Expect(err).NotTo(HaveOccurred())
 
-				candidates, err = selectCoordinators(cluster, status)
+				candidates, err = selectCoordinators(logr.Discard(), cluster, status)
 				Expect(err).NotTo(HaveOccurred())
 			})
 
@@ -171,7 +174,7 @@ var _ = Describe("Change coordinators", func() {
 					initialCandidates := candidates
 
 					for i := 0; i < 100; i++ {
-						newCandidates, err := selectCoordinators(cluster, status)
+						newCandidates, err := selectCoordinators(logr.Discard(), cluster, status)
 						Expect(err).NotTo(HaveOccurred())
 						Expect(newCandidates).To(Equal(initialCandidates))
 					}
@@ -202,7 +205,7 @@ var _ = Describe("Change coordinators", func() {
 
 		When("Using a HA clusters", func() {
 			var status *fdbv1beta2.FoundationDBStatus
-			var candidates []localityInfo
+			var candidates []locality.Info
 			var excludes []string
 			var removals []string
 			var dcCnt int
@@ -211,7 +214,7 @@ var _ = Describe("Change coordinators", func() {
 
 			BeforeEach(func() {
 				// ensure a clean state
-				candidates = []localityInfo{}
+				candidates = []locality.Info{}
 				excludes = []string{}
 				removals = []string{}
 				shouldFail = false
@@ -229,7 +232,7 @@ var _ = Describe("Change coordinators", func() {
 				// generate status for 2 dcs and 1 sate
 				status.Cluster.Processes = generateProcessInfo(dcCnt, satCnt, excludes)
 
-				candidates, err = selectCoordinators(cluster, status)
+				candidates, err = selectCoordinators(logr.Discard(), cluster, status)
 				if shouldFail {
 					Expect(err).To(HaveOccurred())
 				} else {
@@ -401,7 +404,7 @@ var _ = Describe("Change coordinators", func() {
 						initialCandidates := candidates
 
 						for i := 0; i < 100; i++ {
-							newCandidates, err := selectCoordinators(cluster, status)
+							newCandidates, err := selectCoordinators(logr.Discard(), cluster, status)
 							Expect(err).NotTo(HaveOccurred())
 							Expect(newCandidates).To(Equal(initialCandidates))
 						}
@@ -599,7 +602,7 @@ var _ = Describe("Change coordinators", func() {
 						initialCandidates := candidates
 
 						for i := 0; i < 100; i++ {
-							newCandidates, err := selectCoordinators(cluster, status)
+							newCandidates, err := selectCoordinators(logr.Discard(), cluster, status)
 							Expect(err).NotTo(HaveOccurred())
 							Expect(newCandidates).To(Equal(initialCandidates))
 						}
@@ -607,104 +610,6 @@ var _ = Describe("Change coordinators", func() {
 				})
 			})
 		})
-
-		// TODO add test case for multi KC
-
-		When("Sorting the localities", func() {
-			var localities []localityInfo
-
-			BeforeEach(func() {
-				localities = []localityInfo{
-					{
-						ID:    "storage-1",
-						Class: fdbv1beta2.ProcessClassStorage,
-					},
-					{
-						ID:    "tlog-1",
-						Class: fdbv1beta2.ProcessClassTransaction,
-					},
-					{
-						ID:    "log-1",
-						Class: fdbv1beta2.ProcessClassLog,
-					},
-					{
-						ID:    "storage-51",
-						Class: fdbv1beta2.ProcessClassStorage,
-					},
-				}
-			})
-
-			When("no other preferences are defined", func() {
-				BeforeEach(func() {
-					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{}
-				})
-
-				It("should sort the localities based on the IDs", func() {
-					sortLocalities(cluster, localities)
-
-					Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassLog))
-					Expect(localities[0].ID).To(Equal("log-1"))
-					Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-					Expect(localities[1].ID).To(Equal("storage-1"))
-					Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
-					Expect(localities[2].ID).To(Equal("storage-51"))
-					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
-					Expect(localities[3].ID).To(Equal("tlog-1"))
-				})
-			})
-
-			When("when the storage class is preferred", func() {
-				BeforeEach(func() {
-					cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
-						{
-							ProcessClass: fdbv1beta2.ProcessClassStorage,
-							Priority:     0,
-						},
-					}
-				})
-
-				It("should sort the localities based on the provided config", func() {
-					sortLocalities(cluster, 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:     1,
-						},
-						{
-							ProcessClass: fdbv1beta2.ProcessClassTransaction,
-							Priority:     0,
-						},
-					}
-				})
-
-				It("should sort the localities based on the provided config", func() {
-					sortLocalities(cluster, 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.ProcessClassTransaction))
-					Expect(localities[2].ID).To(Equal("tlog-1"))
-					Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassLog))
-					Expect(localities[3].ID).To(Equal("log-1"))
-				})
-			})
-		})
 	})
 
 	Describe("reconcile", func() {
diff --git a/controllers/cluster_controller_test.go b/controllers/cluster_controller_test.go
index 393df336..ab413029 100644
--- a/controllers/cluster_controller_test.go
+++ b/controllers/cluster_controller_test.go
@@ -24,13 +24,11 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	"net"
 	"regexp"
 	"sort"
 	"strings"
 	"time"
 
-	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/fdbadminclient"
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/podmanager"
 
 	"k8s.io/utils/pointer"
@@ -3311,532 +3309,6 @@ var _ = Describe("cluster_controller", func() {
 		})
 	})
 
-	Describe("chooseDistributedProcesses", func() {
-		var candidates []localityInfo
-		var result []localityInfo
-		var err error
-
-		Context("with a flat set of processes", func() {
-			BeforeEach(func() {
-				candidates = []localityInfo{
-					{ID: "p1", LocalityData: map[string]string{"zoneid": "z1"}},
-					{ID: "p2", LocalityData: map[string]string{"zoneid": "z1"}},
-					{ID: "p3", LocalityData: map[string]string{"zoneid": "z2"}},
-					{ID: "p4", LocalityData: map[string]string{"zoneid": "z3"}},
-					{ID: "p5", LocalityData: map[string]string{"zoneid": "z2"}},
-					{ID: "p6", LocalityData: map[string]string{"zoneid": "z4"}},
-					{ID: "p7", LocalityData: map[string]string{"zoneid": "z5"}},
-				}
-				result, err = chooseDistributedProcesses(cluster, candidates, 5, processSelectionConstraint{})
-				Expect(err).NotTo(HaveOccurred())
-			})
-
-			It("should recruit the processes across multiple zones", func() {
-				Expect(len(result)).To(Equal(5))
-				Expect(result[0].ID).To(Equal("p1"))
-				Expect(result[1].ID).To(Equal("p3"))
-				Expect(result[2].ID).To(Equal("p4"))
-				Expect(result[3].ID).To(Equal("p6"))
-				Expect(result[4].ID).To(Equal("p7"))
-			})
-		})
-
-		Context("with fewer zones than desired processes", func() {
-			BeforeEach(func() {
-				candidates = []localityInfo{
-					{ID: "p1", LocalityData: map[string]string{"zoneid": "z1"}},
-					{ID: "p2", LocalityData: map[string]string{"zoneid": "z1"}},
-					{ID: "p3", LocalityData: map[string]string{"zoneid": "z2"}},
-					{ID: "p4", LocalityData: map[string]string{"zoneid": "z3"}},
-					{ID: "p5", LocalityData: map[string]string{"zoneid": "z2"}},
-					{ID: "p6", LocalityData: map[string]string{"zoneid": "z4"}},
-				}
-			})
-
-			Context("with no hard limit", func() {
-				It("should only re-use zones as necessary", func() {
-					result, err = chooseDistributedProcesses(cluster, candidates, 5, processSelectionConstraint{})
-					Expect(err).NotTo(HaveOccurred())
-
-					Expect(len(result)).To(Equal(5))
-					Expect(result[0].ID).To(Equal("p1"))
-					Expect(result[1].ID).To(Equal("p3"))
-					Expect(result[2].ID).To(Equal("p4"))
-					Expect(result[3].ID).To(Equal("p6"))
-					Expect(result[4].ID).To(Equal("p2"))
-				})
-			})
-
-			Context("with a hard limit", func() {
-				It("should give an error", func() {
-					result, err = chooseDistributedProcesses(cluster, candidates, 5, processSelectionConstraint{
-						HardLimits: map[string]int{"zoneid": 1},
-					})
-					Expect(err).To(HaveOccurred())
-					Expect(err.Error()).To(Equal("Could only select 4 processes, but 5 are required"))
-				})
-			})
-		})
-
-		Context("with multiple data centers", func() {
-			BeforeEach(func() {
-				candidates = []localityInfo{
-					{ID: "p1", LocalityData: map[string]string{"zoneid": "z1", "dcid": "dc1"}},
-					{ID: "p2", LocalityData: map[string]string{"zoneid": "z1", "dcid": "dc1"}},
-					{ID: "p3", LocalityData: map[string]string{"zoneid": "z2", "dcid": "dc1"}},
-					{ID: "p4", LocalityData: map[string]string{"zoneid": "z3", "dcid": "dc1"}},
-					{ID: "p5", LocalityData: map[string]string{"zoneid": "z2", "dcid": "dc1"}},
-					{ID: "p6", LocalityData: map[string]string{"zoneid": "z4", "dcid": "dc1"}},
-					{ID: "p7", LocalityData: map[string]string{"zoneid": "z5", "dcid": "dc1"}},
-					{ID: "p8", LocalityData: map[string]string{"zoneid": "z6", "dcid": "dc2"}},
-					{ID: "p9", LocalityData: map[string]string{"zoneid": "z7", "dcid": "dc2"}},
-					{ID: "p10", LocalityData: map[string]string{"zoneid": "z8", "dcid": "dc2"}},
-				}
-			})
-
-			Context("with the default constraints", func() {
-				BeforeEach(func() {
-					result, err = chooseDistributedProcesses(cluster, candidates, 5, processSelectionConstraint{})
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				It("should recruit the processes across multiple zones and data centers", func() {
-					Expect(len(result)).To(Equal(5))
-					Expect(result[0].ID).To(Equal("p1"))
-					Expect(result[1].ID).To(Equal("p10"))
-					Expect(result[2].ID).To(Equal("p3"))
-					Expect(result[3].ID).To(Equal("p8"))
-					Expect(result[4].ID).To(Equal("p4"))
-				})
-			})
-
-			Context("when only distributing across data centers", func() {
-				BeforeEach(func() {
-					result, err = chooseDistributedProcesses(cluster, candidates, 5, processSelectionConstraint{
-						Fields: []string{"dcid"},
-					})
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				It("should recruit the processes across data centers", func() {
-					Expect(len(result)).To(Equal(5))
-					Expect(result[0].ID).To(Equal("p1"))
-					Expect(result[1].ID).To(Equal("p10"))
-					Expect(result[2].ID).To(Equal("p2"))
-					Expect(result[3].ID).To(Equal("p8"))
-					Expect(result[4].ID).To(Equal("p3"))
-				})
-			})
-		})
-	})
-
-	Describe("checkCoordinatorValidity", func() {
-		var status *fdbv1beta2.FoundationDBStatus
-		var adminClient fdbadminclient.AdminClient
-		var err error
-
-		BeforeEach(func() {
-			err = k8sClient.Create(context.TODO(), cluster)
-			Expect(err).NotTo(HaveOccurred())
-
-			result, err := reconcileCluster(cluster)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(result.Requeue).To(BeFalse())
-
-			generations, err := reloadClusterGenerations(cluster)
-			Expect(err).NotTo(HaveOccurred())
-			Expect(generations.Reconciled).To(Equal(int64(1)))
-
-			adminClient, err = newMockAdminClient(cluster, k8sClient)
-			Expect(err).NotTo(HaveOccurred())
-
-			status, err = adminClient.GetStatus()
-			Expect(err).NotTo(HaveOccurred())
-		})
-
-		Context("with the default configuration", func() {
-			It("should report the coordinators as valid", func() {
-				coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-				for _, coordinator := range status.Client.Coordinators.Coordinators {
-					coordinatorStatus[coordinator.Address.String()] = false
-				}
-
-				coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-				Expect(coordinatorsValid).To(BeTrue())
-				Expect(addressesValid).To(BeTrue())
-				Expect(err).To(BeNil())
-			})
-		})
-
-		Context("with a tester process without a command line field", func() {
-			It("should ignore the test process and report the coordinators as valid", func() {
-				testProcess := fdbv1beta2.FoundationDBStatusProcessInfo{
-					Address: fdbv1beta2.ProcessAddress{
-						IPAddress: net.ParseIP("9.9.9.9"),
-					},
-					ProcessClass: fdbv1beta2.ProcessClassTest,
-				}
-				Expect(testProcess.Address.IsEmpty()).To(BeFalse())
-				status.Cluster.Processes[string(fdbv1beta2.ProcessClassTest)] = testProcess
-
-				coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-				for _, coordinator := range status.Client.Coordinators.Coordinators {
-					coordinatorStatus[coordinator.Address.String()] = false
-				}
-
-				coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-				Expect(coordinatorsValid).To(BeTrue())
-				Expect(addressesValid).To(BeTrue())
-				Expect(err).To(BeNil())
-			})
-		})
-
-		Context("with too few coordinators", func() {
-			BeforeEach(func() {
-				status.Client.Coordinators.Coordinators = status.Client.Coordinators.Coordinators[0:2]
-			})
-
-			It("should report the coordinators as not valid", func() {
-				coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-				for _, coordinator := range status.Client.Coordinators.Coordinators {
-					coordinatorStatus[coordinator.Address.String()] = false
-				}
-
-				coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-				Expect(coordinatorsValid).To(BeFalse())
-				Expect(addressesValid).To(BeTrue())
-				Expect(err).To(BeNil())
-			})
-		})
-
-		Context("with too few zones", func() {
-			BeforeEach(func() {
-				zone := ""
-				for _, process := range status.Cluster.Processes {
-					if process.Address.Equal(status.Client.Coordinators.Coordinators[0].Address) {
-						zone = process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]
-					}
-				}
-				for _, process := range status.Cluster.Processes {
-					if process.Address.Equal(status.Client.Coordinators.Coordinators[1].Address) {
-						process.Locality["zoneid"] = zone
-					}
-				}
-			})
-
-			It("should report the coordinators as not valid", func() {
-				coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-				for _, coordinator := range status.Client.Coordinators.Coordinators {
-					coordinatorStatus[coordinator.Address.String()] = false
-				}
-
-				coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-				Expect(coordinatorsValid).To(BeFalse())
-				Expect(addressesValid).To(BeTrue())
-				Expect(err).To(BeNil())
-			})
-		})
-
-		Context("with multiple regions", func() {
-			BeforeEach(func() {
-				coordinatorCount := 9
-				cluster.Spec.ProcessCounts.Storage = coordinatorCount
-
-				err = k8sClient.Update(context.TODO(), cluster)
-				Expect(err).NotTo(HaveOccurred())
-
-				result, err := reconcileCluster(cluster)
-				Expect(err).NotTo(HaveOccurred())
-				Expect(result.Requeue).To(BeFalse())
-
-				generation, err := reloadCluster(cluster)
-				Expect(err).NotTo(HaveOccurred())
-				Expect(generation).To(Equal(int64(2)))
-
-				status, err = adminClient.GetStatus()
-				Expect(err).NotTo(HaveOccurred())
-
-				cluster.Spec.DatabaseConfiguration.UsableRegions = 2
-
-				coordinators := make([]fdbv1beta2.FoundationDBStatusCoordinator, 0, coordinatorCount)
-				dc := 0
-				for _, process := range status.Cluster.Processes {
-					if process.ProcessClass == fdbv1beta2.ProcessClassStorage {
-						coordinators = append(coordinators, fdbv1beta2.FoundationDBStatusCoordinator{
-							Address:   process.Address,
-							Reachable: true,
-						})
-						dc++
-						process.Locality["dcid"] = fmt.Sprintf("dc%d", dc)
-						dc = dc % 3
-					}
-				}
-				status.Client.Coordinators.Coordinators = coordinators
-			})
-
-			Context("with coordinators divided across three DCs", func() {
-				It("should report the coordinators as valid", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(coordinatorsValid).To(BeTrue())
-					Expect(addressesValid).To(BeTrue())
-					Expect(err).To(BeNil())
-				})
-			})
-
-			Context("with coordinators divided across two DCs", func() {
-				BeforeEach(func() {
-					for _, process := range status.Cluster.Processes {
-						if process.Locality["dcid"] == "dc3" {
-							process.Locality["dcid"] = "dc1"
-						}
-					}
-				})
-				It("should report the coordinators as not valid", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(coordinatorsValid).To(BeFalse())
-					Expect(addressesValid).To(BeTrue())
-					Expect(err).To(BeNil())
-				})
-			})
-		})
-
-		When("changing the TLS setting", func() {
-			BeforeEach(func() {
-				cluster.Status.RequiredAddresses = fdbv1beta2.RequiredAddressSet{
-					TLS:    true,
-					NonTLS: true,
-				}
-			})
-
-			When("TLS is disabled", func() {
-				BeforeEach(func() {
-					cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
-						EnableTLS: false,
-					}
-					cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
-						EnableTLS: false,
-					}
-
-					adminClient, err = newMockAdminClient(cluster, k8sClient)
-					Expect(err).NotTo(HaveOccurred())
-
-					status, err = adminClient.GetStatus()
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				It("should report the coordinators addresses as valid", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					_, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(addressesValid).To(BeTrue())
-					Expect(err).To(BeNil())
-				})
-
-				When("converting back to TLS", func() {
-					BeforeEach(func() {
-						cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
-							EnableTLS: true,
-						}
-						cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
-							EnableTLS: true,
-						}
-
-						adminClient, err = newMockAdminClient(cluster, k8sClient)
-						Expect(err).NotTo(HaveOccurred())
-
-						status, err = adminClient.GetStatus()
-						Expect(err).NotTo(HaveOccurred())
-					})
-
-					It("should report the coordinators addresses as valid", func() {
-						coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-						for _, coordinator := range status.Client.Coordinators.Coordinators {
-							coordinatorStatus[coordinator.Address.String()] = false
-						}
-
-						_, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-						Expect(addressesValid).To(BeTrue())
-						Expect(err).To(BeNil())
-					})
-				})
-			})
-
-			When("TLS is enabled", func() {
-				BeforeEach(func() {
-					cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
-						EnableTLS: true,
-					}
-					cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
-						EnableTLS: true,
-					}
-
-					adminClient, err = newMockAdminClient(cluster, k8sClient)
-					Expect(err).NotTo(HaveOccurred())
-
-					status, err = adminClient.GetStatus()
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				It("should report the coordinators addresses as valid", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					_, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(addressesValid).To(BeTrue())
-					Expect(err).To(BeNil())
-				})
-
-				When("converting back to non-TLS", func() {
-					BeforeEach(func() {
-						cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
-							EnableTLS: false,
-						}
-
-						cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
-							EnableTLS: false,
-						}
-
-						adminClient, err = newMockAdminClient(cluster, k8sClient)
-						Expect(err).NotTo(HaveOccurred())
-
-						status, err = adminClient.GetStatus()
-						Expect(err).NotTo(HaveOccurred())
-					})
-
-					It("should report the coordinators addresses as valid", func() {
-						coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-						for _, coordinator := range status.Client.Coordinators.Coordinators {
-							coordinatorStatus[coordinator.Address.String()] = false
-						}
-
-						_, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-						Expect(addressesValid).To(BeTrue())
-						Expect(err).To(BeNil())
-					})
-				})
-			})
-		})
-
-		When("Generating the locality info based on a process", func() {
-			BeforeEach(func() {
-				cluster.Status.RequiredAddresses = fdbv1beta2.RequiredAddressSet{
-					TLS:    true,
-					NonTLS: true,
-				}
-
-				cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
-					EnableTLS: true,
-				}
-				cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
-					EnableTLS: true,
-				}
-
-				adminClient, err = newMockAdminClient(cluster, k8sClient)
-				Expect(err).NotTo(HaveOccurred())
-
-				status, err = adminClient.GetStatus()
-				Expect(err).NotTo(HaveOccurred())
-			})
-
-			When("TLS is disabled", func() {
-				It("the locality address should not contain tls", func() {
-					var proc fdbv1beta2.FoundationDBStatusProcessInfo
-					for _, p := range status.Cluster.Processes {
-						proc = p
-						// Select the first process Info
-						break
-					}
-
-					info, err := localityInfoForProcess(proc, false)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(info.Address).NotTo(HaveSuffix("tls"))
-				})
-			})
-
-			When("TLS is enabled", func() {
-				It("the locality address should contain tls", func() {
-					var proc fdbv1beta2.FoundationDBStatusProcessInfo
-					for _, p := range status.Cluster.Processes {
-						proc = p
-						// Select the first process Info
-						break
-					}
-
-					info, err := localityInfoForProcess(proc, true)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(info.Address).To(HaveSuffix("tls"))
-				})
-			})
-		})
-
-		When("enabling DNS names in the cluster file", func() {
-			BeforeEach(func() {
-				cluster.Spec.Routing.UseDNSInClusterFile = pointer.Bool(true)
-			})
-
-			Context("when the pods do not have DNS names assigned", func() {
-				It("should report valid coordinators", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(coordinatorsValid).To(BeTrue())
-					Expect(addressesValid).To(BeTrue())
-				})
-			})
-
-			Context("when the pods have DNS names assigned", func() {
-				BeforeEach(func() {
-					pods := &corev1.PodList{}
-					err = k8sClient.List(context.TODO(), pods)
-					Expect(err).NotTo(HaveOccurred())
-
-					for _, pod := range pods.Items {
-						container := pod.Spec.Containers[1]
-						container.Env = append(container.Env, corev1.EnvVar{Name: "FDB_DNS_NAME", Value: internal.GetPodDNSName(cluster, pod.Name)})
-						pod.Spec.Containers[1] = container
-						err = k8sClient.Update(context.TODO(), &pod)
-						Expect(err).NotTo(HaveOccurred())
-					}
-
-					status, err = adminClient.GetStatus()
-					Expect(err).NotTo(HaveOccurred())
-				})
-
-				It("should reject coordinators based on IP addresses", func() {
-					coordinatorStatus := make(map[string]bool, len(status.Client.Coordinators.Coordinators))
-					for _, coordinator := range status.Client.Coordinators.Coordinators {
-						coordinatorStatus[coordinator.Address.String()] = false
-					}
-
-					coordinatorsValid, addressesValid, err := checkCoordinatorValidity(cluster, status, coordinatorStatus)
-					Expect(err).NotTo(HaveOccurred())
-					Expect(coordinatorsValid).To(BeFalse())
-					Expect(addressesValid).To(BeTrue())
-				})
-
-			})
-		})
-	})
-
 	Context("Setting the partial connection string", func() {
 		var partialConnectionString fdbv1beta2.ConnectionString
 
diff --git a/fdbclient/admin_client_test.go b/fdbclient/admin_client_test.go
index 38707820..03809fff 100644
--- a/fdbclient/admin_client_test.go
+++ b/fdbclient/admin_client_test.go
@@ -21,7 +21,13 @@
 package fdbclient
 
 import (
+	"errors"
 	"net"
+	"os"
+	"path"
+	"time"
+
+	"github.com/go-logr/logr"
 
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	. "github.com/onsi/ginkgo/v2"
@@ -197,4 +203,433 @@ var _ = Describe("admin_client_test", func() {
 			),
 		)
 	})
+
+	When("getting the log dir parameter", func() {
+		DescribeTable("it should return the correct format of the log dir paramater",
+			func(cmd cliCommand, expected string) {
+				Expect(cmd.getLogDirParameter()).To(Equal(expected))
+			},
+			Entry("no binary set",
+				cliCommand{
+					binary: "",
+				},
+				"--log-dir",
+			),
+			Entry("fdbcli binary set",
+				cliCommand{
+					binary: fdbcliStr,
+				},
+				"--log-dir",
+			),
+			Entry("fdbbackup binary set",
+				cliCommand{
+					binary: fdbbackupStr,
+				},
+				"--logdir",
+			),
+			Entry("fdbrestore binary set",
+				cliCommand{
+					binary: fdbrestoreStr,
+				},
+				"--logdir",
+			),
+		)
+	})
+
+	When("getting the binary name", func() {
+		DescribeTable("it should return the correct binary name",
+			func(cmd cliCommand, expected string) {
+				Expect(cmd.getBinary()).To(Equal(expected))
+			},
+			Entry("no binary set",
+				cliCommand{
+					binary: "",
+				},
+				fdbcliStr,
+			),
+			Entry("fdbcli binary set",
+				cliCommand{
+					binary: fdbcliStr,
+				},
+				fdbcliStr,
+			),
+			Entry("fdbbackup binary set",
+				cliCommand{
+					binary: fdbbackupStr,
+				},
+				fdbbackupStr,
+			),
+			Entry("fdbrestore binary set",
+				cliCommand{
+					binary: fdbrestoreStr,
+				},
+				fdbrestoreStr,
+			),
+		)
+	})
+
+	When("checking if the binary is fdbcli", func() {
+		DescribeTable("it should return the correct output",
+			func(cmd cliCommand, expected bool) {
+				Expect(cmd.isFdbCli()).To(Equal(expected))
+			},
+			Entry("no binary set",
+				cliCommand{
+					binary: "",
+				},
+				true,
+			),
+			Entry("fdbcli binary set",
+				cliCommand{
+					binary: fdbcliStr,
+				},
+				true,
+			),
+			Entry("fdbbackup binary set",
+				cliCommand{
+					binary: fdbbackupStr,
+				},
+				false,
+			),
+			Entry("fdbrestore binary set",
+				cliCommand{
+					binary: fdbrestoreStr,
+				},
+				false,
+			),
+		)
+	})
+
+	When("getting the version to run the command", func() {
+		DescribeTable("it should return the correct output",
+			func(cmd cliCommand, cluster *fdbv1beta2.FoundationDBCluster, expected string) {
+				Expect(cmd.getVersion(cluster)).To(Equal(expected))
+			},
+			Entry("version is set set",
+				cliCommand{
+					version: "7.1.15",
+				},
+				nil,
+				"7.1.15",
+			),
+			Entry("version is not set and running version is defined",
+				cliCommand{},
+				&fdbv1beta2.FoundationDBCluster{
+					Spec: fdbv1beta2.FoundationDBClusterSpec{
+						Version: "7.1.25",
+					},
+					Status: fdbv1beta2.FoundationDBClusterStatus{
+						RunningVersion: "7.1.15",
+					},
+				},
+				"7.1.15",
+			),
+			Entry("version is not set and running version is  not defined",
+				cliCommand{},
+				&fdbv1beta2.FoundationDBCluster{
+					Spec: fdbv1beta2.FoundationDBClusterSpec{
+						Version: "7.1.25",
+					},
+					Status: fdbv1beta2.FoundationDBClusterStatus{
+						RunningVersion: "",
+					},
+				},
+				"7.1.25",
+			),
+		)
+	})
+
+	When("getting the args for the command", func() {
+		var command cliCommand
+		var client *cliAdminClient
+		var args, expectedArgs []string
+		var timeout, expectedTimeout time.Duration
+
+		JustBeforeEach(func() {
+			Expect(client).NotTo(BeNil())
+			args, timeout = client.getArgsAndTimeout(command)
+			Expect(timeout).To(Equal(expectedTimeout))
+			Expect(args).To(ContainElements(expectedArgs))
+			Expect(len(args)).To(BeNumerically("==", len(expectedArgs)))
+		})
+
+		When("the used command defines args", func() {
+			BeforeEach(func() {
+				command = cliCommand{
+					args:    []string{"--version"},
+					version: "7.1.25",
+					timeout: 1 * time.Second,
+				}
+
+				client = &cliAdminClient{
+					Cluster:          nil,
+					clusterFilePath:  "test",
+					useClientLibrary: true,
+					log:              logr.Discard(),
+				}
+			})
+
+			When("trace options are disabled", func() {
+				BeforeEach(func() {
+					expectedTimeout = 2 * time.Second
+					expectedArgs = []string{
+						"--version",
+						"--timeout",
+						"1",
+					}
+				})
+			})
+		})
+
+		When("the used command is a fdbcli command", func() {
+			BeforeEach(func() {
+				command = cliCommand{
+					command: "maintenance off",
+					timeout: 1 * time.Second,
+				}
+
+				client = &cliAdminClient{
+					Cluster:          nil,
+					clusterFilePath:  "test",
+					useClientLibrary: true,
+					log:              logr.Discard(),
+				}
+			})
+
+			When("trace options are disabled", func() {
+				BeforeEach(func() {
+					expectedTimeout = 2 * time.Second
+					expectedArgs = []string{
+						"--exec",
+						"maintenance off",
+						"test",
+						"--timeout",
+						"1",
+					}
+				})
+			})
+
+			When("trace options are enabled", func() {
+				BeforeEach(func() {
+					GinkgoT().Setenv("FDB_NETWORK_OPTION_TRACE_ENABLE", "/tmp")
+					expectedTimeout = 2 * time.Second
+					expectedArgs = []string{
+						"--exec",
+						"maintenance off",
+						"test",
+						"--log",
+						"--trace_format",
+						"xml",
+						"--log-dir",
+						"/tmp",
+						"--timeout",
+						"1",
+					}
+				})
+
+				When("a different trace format is defined", func() {
+					BeforeEach(func() {
+						GinkgoT().Setenv("FDB_NETWORK_OPTION_TRACE_FORMAT", "json")
+						expectedTimeout = 2 * time.Second
+						expectedArgs = []string{
+							"--exec",
+							"maintenance off",
+							"test",
+							"--log",
+							"--trace_format",
+							"json",
+							"--log-dir",
+							"/tmp",
+							"--timeout",
+							"1",
+						}
+					})
+				})
+			})
+		})
+
+		When("the used command is a fdbrestore command", func() {
+			BeforeEach(func() {
+				command = cliCommand{
+					binary: fdbrestoreStr,
+					args: []string{
+						"status",
+					},
+					timeout: 1 * time.Second,
+				}
+
+				client = &cliAdminClient{
+					Cluster:          nil,
+					clusterFilePath:  "test",
+					useClientLibrary: true,
+					log:              logr.Discard(),
+				}
+			})
+
+			When("trace options are disabled", func() {
+				BeforeEach(func() {
+					expectedTimeout = 2 * time.Second
+					expectedArgs = []string{
+						"status",
+					}
+				})
+			})
+
+			When("trace options are enabled", func() {
+				BeforeEach(func() {
+					GinkgoT().Setenv("FDB_NETWORK_OPTION_TRACE_ENABLE", "/tmp")
+					expectedTimeout = 2 * time.Second
+					expectedArgs = []string{
+						"status",
+						"--log",
+						"--trace_format",
+						"xml",
+						"--logdir",
+						"/tmp",
+					}
+				})
+
+				When("a different trace format is defined", func() {
+					BeforeEach(func() {
+						GinkgoT().Setenv("FDB_NETWORK_OPTION_TRACE_FORMAT", "json")
+						expectedTimeout = 2 * time.Second
+						expectedArgs = []string{
+							"status",
+							"--log",
+							"--trace_format",
+							"json",
+							"--logdir",
+							"/tmp",
+						}
+					})
+				})
+			})
+		})
+	})
+
+	When("getting the protocol version from fdbcli", func() {
+		var mockRunner *mockCommandRunner
+		var protocolVersion string
+		var err error
+
+		JustBeforeEach(func() {
+			cliClient := &cliAdminClient{
+				Cluster:          nil,
+				clusterFilePath:  "test",
+				useClientLibrary: true,
+				log:              logr.Discard(),
+				cmdRunner:        mockRunner,
+			}
+
+			protocolVersion, err = cliClient.GetProtocolVersion("7.1.21")
+		})
+
+		When("the fdbcli call returns the expected output", func() {
+			BeforeEach(func() {
+				mockRunner = &mockCommandRunner{
+					mockedError: nil,
+					mockedOutput: `fdbcli --version
+FoundationDB CLI 7.1 (v7.1.21)
+source version e9f38c7169d21dde901b7b9408e1c5a8df182d64
+protocol fdb00b071010000`,
+				}
+			})
+
+			It("should report the protocol version", func() {
+				Expect(protocolVersion).To(Equal("fdb00b071010000"))
+				Expect(err).NotTo(HaveOccurred())
+				Expect(mockRunner.receivedBinary).To(Equal("7.1/" + fdbcliStr))
+				Expect(mockRunner.receivedArgs).To(ContainElements("--version"))
+			})
+		})
+
+		When("an error is returned", func() {
+			BeforeEach(func() {
+				mockRunner = &mockCommandRunner{
+					mockedError: errors.New("boom"),
+					mockedOutput: `fdbcli --version
+FoundationDB CLI 7.1 (v7.1.21)
+source version e9f38c7169d21dde901b7b9408e1c5a8df182d64
+protocol fdb00b071010000`,
+				}
+			})
+
+			It("should report the error", func() {
+				Expect(protocolVersion).To(Equal(""))
+				Expect(err).To(HaveOccurred())
+				Expect(mockRunner.receivedBinary).To(Equal("7.1/" + fdbcliStr))
+				Expect(mockRunner.receivedArgs).To(ContainElements("--version"))
+			})
+		})
+
+		When("the protocol version is missing", func() {
+			BeforeEach(func() {
+				mockRunner = &mockCommandRunner{
+					mockedError:  errors.New("boom"),
+					mockedOutput: "",
+				}
+			})
+
+			It("should report the error", func() {
+				Expect(protocolVersion).To(Equal(""))
+				Expect(err).To(HaveOccurred())
+				Expect(mockRunner.receivedBinary).To(Equal("7.1/" + fdbcliStr))
+				Expect(mockRunner.receivedArgs).To(ContainElements("--version"))
+			})
+		})
+	})
+
+	When("validating if the version is supported", func() {
+		var mockRunner *mockCommandRunner
+		var supported bool
+		var err error
+
+		JustBeforeEach(func() {
+			cliClient := &cliAdminClient{
+				Cluster:          nil,
+				clusterFilePath:  "test",
+				useClientLibrary: true,
+				log:              logr.Discard(),
+				cmdRunner:        mockRunner,
+			}
+
+			supported, err = cliClient.VersionSupported(fdbv1beta2.Versions.Default.String())
+		})
+
+		When("the binary does not exist", func() {
+			BeforeEach(func() {
+				mockRunner = &mockCommandRunner{
+					mockedError:  nil,
+					mockedOutput: ``,
+				}
+			})
+
+			It("should return an error", func() {
+				Expect(supported).To(BeFalse())
+				Expect(err).To(HaveOccurred())
+			})
+		})
+
+		When("the binary exists", func() {
+			BeforeEach(func() {
+				tmpDir := GinkgoT().TempDir()
+				GinkgoT().Setenv("FDB_BINARY_DIR", tmpDir)
+
+				binaryDir := path.Join(tmpDir, fdbv1beta2.Versions.Default.GetBinaryVersion())
+				Expect(os.MkdirAll(binaryDir, 0700)).NotTo(HaveOccurred())
+				_, err := os.Create(path.Join(binaryDir, fdbcliStr))
+				Expect(err).NotTo(HaveOccurred())
+
+				mockRunner = &mockCommandRunner{
+					mockedError:  nil,
+					mockedOutput: ``,
+				}
+			})
+
+			It("should return that the version is supported", func() {
+				Expect(supported).To(BeTrue())
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+	})
 })
diff --git a/fdbclient/command_runner_test.go b/fdbclient/command_runner_test.go
new file mode 100644
index 00000000..e1376c90
--- /dev/null
+++ b/fdbclient/command_runner_test.go
@@ -0,0 +1,48 @@
+/*
+ * command_runner_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 fdbclient
+
+import (
+	"context"
+	"github.com/go-logr/logr"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("command_runner", func() {
+	When("running a command", func() {
+		var output []byte
+		var err error
+
+		BeforeEach(func() {
+			runner := realCommandRunner{
+				log: logr.Discard(),
+			}
+
+			output, err = runner.runCommand(context.TODO(), "echo", "hello")
+		})
+
+		It("should execute the command successfully", func() {
+			Expect(err).NotTo(HaveOccurred())
+			Expect(string(output)).To(Equal("hello\n"))
+		})
+	})
+})
diff --git a/internal/locality/locality_test.go b/internal/locality/locality_test.go
new file mode 100644
index 00000000..f9741c9b
--- /dev/null
+++ b/internal/locality/locality_test.go
@@ -0,0 +1,1006 @@
+package locality
+
+/*
+ * change_coordinators_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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.
+ */
+
+import (
+	"fmt"
+	"math"
+	"net"
+	"strings"
+	"time"
+
+	"github.com/go-logr/logr"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	"github.com/FoundationDB/fdb-kubernetes-operator/internal"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func generateDummyProcessInfo(id string, dcID string, port int, tls bool) fdbv1beta2.FoundationDBStatusProcessInfo {
+	ipString := "1.1.1." + strings.Split(id, "-")[1]
+
+	var tlsSuffix string
+	var flags map[string]bool
+	if tls {
+		flags = map[string]bool{"tls": true}
+		tlsSuffix = ":tls"
+	}
+
+	info := fdbv1beta2.FoundationDBStatusProcessInfo{
+		ProcessClass: fdbv1beta2.ProcessClassStorage,
+		Address: fdbv1beta2.ProcessAddress{
+			IPAddress: net.ParseIP(ipString),
+			Port:      port,
+			Flags:     flags,
+		},
+		CommandLine: fmt.Sprintf("... --public_address=%s:%d%s ...", ipString, port, tlsSuffix),
+		Locality: map[string]string{
+			fdbv1beta2.FDBLocalityInstanceIDKey: id,
+			fdbv1beta2.FDBLocalityZoneIDKey:     id,
+			fdbv1beta2.FDBLocalityDCIDKey:       dcID,
+		},
+	}
+
+	return info
+}
+
+func generateDefaultStatus(tls bool) *fdbv1beta2.FoundationDBStatus {
+	var flags map[string]bool
+
+	port := 4501
+	if tls {
+		port = 4500
+		flags = map[string]bool{"tls": true}
+	}
+
+	return &fdbv1beta2.FoundationDBStatus{
+		Client: fdbv1beta2.FoundationDBStatusLocalClientInfo{
+			Coordinators: fdbv1beta2.FoundationDBStatusCoordinatorInfo{
+				Coordinators: []fdbv1beta2.FoundationDBStatusCoordinator{
+					{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.1"),
+							Port:      port,
+							Flags:     flags,
+						},
+						Reachable: true,
+					},
+					{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.2"),
+							Port:      port,
+							Flags:     flags,
+						},
+						Reachable: true,
+					},
+					{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.3"),
+							Port:      port,
+							Flags:     flags,
+						},
+						Reachable: true,
+					},
+				},
+			},
+		},
+		Cluster: fdbv1beta2.FoundationDBStatusClusterInfo{
+			Processes: map[string]fdbv1beta2.FoundationDBStatusProcessInfo{
+				"1": generateDummyProcessInfo("test-1", "dc1", port, tls),
+				"2": generateDummyProcessInfo("test-2", "dc1", port, tls),
+				"3": generateDummyProcessInfo("test-3", "dc1", port, tls),
+			},
+		},
+	}
+}
+
+var _ = Describe("Change coordinators", func() {
+	var cluster *fdbv1beta2.FoundationDBCluster
+
+	BeforeEach(func() {
+		cluster = internal.CreateDefaultCluster()
+		disabled := false
+		cluster.Spec.LockOptions.DisableLocks = &disabled
+		cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+			{
+				ProcessClass: fdbv1beta2.ProcessClassStorage,
+				Priority:     math.MaxInt32,
+			},
+			{
+				ProcessClass: fdbv1beta2.ProcessClassLog,
+				Priority:     0,
+			},
+		}
+	})
+
+	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,
+				},
+			}
+		})
+
+		When("no other preferences are defined", func() {
+			BeforeEach(func() {
+				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{}
+			})
+
+			It("should sort the localities based on the IDs", func() {
+				sortLocalities(cluster, localities)
+
+				Expect(localities[0].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+				Expect(localities[0].ID).To(Equal("log-1"))
+				Expect(localities[1].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+				Expect(localities[1].ID).To(Equal("storage-1"))
+				Expect(localities[2].Class).To(Equal(fdbv1beta2.ProcessClassStorage))
+				Expect(localities[2].ID).To(Equal("storage-51"))
+				Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassTransaction))
+				Expect(localities[3].ID).To(Equal("tlog-1"))
+			})
+		})
+
+		When("when the storage class is preferred", func() {
+			BeforeEach(func() {
+				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+					{
+						ProcessClass: fdbv1beta2.ProcessClassStorage,
+						Priority:     0,
+					},
+				}
+			})
+
+			It("should sort the localities based on the provided config", func() {
+				sortLocalities(cluster, 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:     1,
+					},
+					{
+						ProcessClass: fdbv1beta2.ProcessClassTransaction,
+						Priority:     0,
+					},
+				}
+			})
+
+			It("should sort the localities based on the provided config", func() {
+				sortLocalities(cluster, 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.ProcessClassTransaction))
+				Expect(localities[2].ID).To(Equal("tlog-1"))
+				Expect(localities[3].Class).To(Equal(fdbv1beta2.ProcessClassLog))
+				Expect(localities[3].ID).To(Equal("log-1"))
+			})
+		})
+	})
+
+	Describe("chooseDistributedProcesses", func() {
+		var candidates []Info
+		var result []Info
+		var err error
+
+		Context("with a flat set of processes", func() {
+			BeforeEach(func() {
+				candidates = []Info{
+					{ID: "p1", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1"}},
+					{ID: "p2", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1"}},
+					{ID: "p3", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2"}},
+					{ID: "p4", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z3"}},
+					{ID: "p5", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2"}},
+					{ID: "p6", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z4"}},
+					{ID: "p7", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z5"}},
+				}
+				result, err = ChooseDistributedProcesses(cluster, candidates, 5, ProcessSelectionConstraint{})
+				Expect(err).NotTo(HaveOccurred())
+			})
+
+			It("should recruit the processes across multiple zones", func() {
+				Expect(len(result)).To(Equal(5))
+				Expect(result[0].ID).To(Equal("p1"))
+				Expect(result[1].ID).To(Equal("p3"))
+				Expect(result[2].ID).To(Equal("p4"))
+				Expect(result[3].ID).To(Equal("p6"))
+				Expect(result[4].ID).To(Equal("p7"))
+			})
+		})
+
+		Context("with fewer zones than desired processes", func() {
+			BeforeEach(func() {
+				candidates = []Info{
+					{ID: "p1", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1"}},
+					{ID: "p2", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z1"}},
+					{ID: "p3", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2"}},
+					{ID: "p4", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z3"}},
+					{ID: "p5", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z2"}},
+					{ID: "p6", LocalityData: map[string]string{fdbv1beta2.FDBLocalityZoneIDKey: "z4"}},
+				}
+			})
+
+			Context("with no hard limit", func() {
+				It("should only re-use zones as necessary", func() {
+					result, err = ChooseDistributedProcesses(cluster, candidates, 5, ProcessSelectionConstraint{})
+					Expect(err).NotTo(HaveOccurred())
+
+					Expect(len(result)).To(Equal(5))
+					Expect(result[0].ID).To(Equal("p1"))
+					Expect(result[1].ID).To(Equal("p3"))
+					Expect(result[2].ID).To(Equal("p4"))
+					Expect(result[3].ID).To(Equal("p6"))
+					Expect(result[4].ID).To(Equal("p2"))
+				})
+			})
+
+			Context("with a hard limit", func() {
+				It("should give an error", func() {
+					result, err = ChooseDistributedProcesses(cluster, candidates, 5, ProcessSelectionConstraint{
+						HardLimits: map[string]int{fdbv1beta2.FDBLocalityZoneIDKey: 1},
+					})
+					Expect(err).To(HaveOccurred())
+					Expect(err.Error()).To(Equal("Could only select 4 processes, but 5 are required"))
+				})
+			})
+		})
+
+		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"}},
+				}
+			})
+
+			Context("with the default constraints", func() {
+				BeforeEach(func() {
+					result, err = ChooseDistributedProcesses(cluster, candidates, 5, ProcessSelectionConstraint{})
+					Expect(err).NotTo(HaveOccurred())
+				})
+
+				It("should recruit the processes across multiple zones and data centers", func() {
+					Expect(len(result)).To(Equal(5))
+					Expect(result[0].ID).To(Equal("p1"))
+					Expect(result[1].ID).To(Equal("p10"))
+					Expect(result[2].ID).To(Equal("p3"))
+					Expect(result[3].ID).To(Equal("p8"))
+					Expect(result[4].ID).To(Equal("p4"))
+				})
+			})
+
+			Context("when only distributing across data centers", func() {
+				BeforeEach(func() {
+					result, err = ChooseDistributedProcesses(cluster, candidates, 5, ProcessSelectionConstraint{
+						Fields: []string{"dcid"},
+					})
+					Expect(err).NotTo(HaveOccurred())
+				})
+
+				It("should recruit the processes across data centers", func() {
+					Expect(len(result)).To(Equal(5))
+					Expect(result[0].ID).To(Equal("p1"))
+					Expect(result[1].ID).To(Equal("p10"))
+					Expect(result[2].ID).To(Equal("p2"))
+					Expect(result[3].ID).To(Equal("p8"))
+					Expect(result[4].ID).To(Equal("p3"))
+				})
+			})
+		})
+	})
+
+	DescribeTable("when getting the hard limits", func(cluster *fdbv1beta2.FoundationDBCluster, expected map[string]int) {
+		Expect(GetHardLimits(cluster)).To(Equal(expected))
+	},
+		Entry("default cluster with one usable region",
+			&fdbv1beta2.FoundationDBCluster{},
+			map[string]int{
+				fdbv1beta2.FDBLocalityZoneIDKey: 1,
+			},
+		),
+		Entry("default cluster with two usable regiosn",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					DatabaseConfiguration: fdbv1beta2.DatabaseConfiguration{
+						UsableRegions: 2,
+					},
+				},
+			},
+			map[string]int{
+				fdbv1beta2.FDBLocalityZoneIDKey: 1,
+				fdbv1beta2.FDBLocalityDCIDKey:   4,
+			},
+		),
+	)
+
+	DescribeTable("when getting the locality info from a process", func(process fdbv1beta2.FoundationDBStatusProcessInfo, mainContainerTLS bool, expected Info, expectedError bool) {
+		info, err := InfoForProcess(process, mainContainerTLS)
+		if expectedError {
+			Expect(err).To(HaveOccurred())
+			return
+		}
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(info).To(Equal(expected))
+	},
+		Entry("a process info without tls",
+			fdbv1beta2.FoundationDBStatusProcessInfo{
+				CommandLine: "... --public_address=1.1.1.1:4501,1.1.1.1:4500:tls ...",
+				Locality: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+				ProcessClass: fdbv1beta2.ProcessClassStorage,
+			},
+			false,
+			Info{
+				ID: "test",
+				Address: fdbv1beta2.ProcessAddress{
+					IPAddress: net.ParseIP("1.1.1.1"),
+					Port:      4501,
+				},
+				Class: fdbv1beta2.ProcessClassStorage,
+				LocalityData: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+			},
+			false,
+		),
+		Entry("a process info with tls",
+			fdbv1beta2.FoundationDBStatusProcessInfo{
+				CommandLine: "... --public_address=1.1.1.1:4501,1.1.1.1:4500:tls ...",
+				Locality: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+				ProcessClass: fdbv1beta2.ProcessClassStorage,
+			},
+			true,
+			Info{
+				ID: "test",
+				Address: fdbv1beta2.ProcessAddress{
+					IPAddress: net.ParseIP("1.1.1.1"),
+					Port:      4500,
+					Flags: map[string]bool{
+						"tls": true,
+					},
+				},
+				Class: fdbv1beta2.ProcessClassStorage,
+				LocalityData: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+			},
+			false,
+		),
+		Entry("a process info with missing public address",
+			fdbv1beta2.FoundationDBStatusProcessInfo{
+				CommandLine: "",
+				Locality: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+				ProcessClass: fdbv1beta2.ProcessClassStorage,
+			},
+			true,
+			Info{
+				ID: "test",
+				Address: fdbv1beta2.ProcessAddress{
+					IPAddress: net.ParseIP("1.1.1.1"),
+					Port:      4500,
+					Flags: map[string]bool{
+						"tls": true,
+					},
+				},
+				Class: fdbv1beta2.ProcessClassStorage,
+				LocalityData: map[string]string{
+					fdbv1beta2.FDBLocalityInstanceIDKey: "test",
+				},
+			},
+			true,
+		),
+	)
+
+	DescribeTable("when getting the locality info from a sidecar", func(cluster *fdbv1beta2.FoundationDBCluster, pod *corev1.Pod, expected Info, expectedError bool) {
+		client, err := internal.NewMockFdbPodClient(cluster, pod)
+		Expect(err).NotTo(HaveOccurred())
+
+		info, err := InfoFromSidecar(cluster, client)
+		if expectedError {
+			Expect(err).To(HaveOccurred())
+			return
+		}
+
+		Expect(err).NotTo(HaveOccurred())
+		Expect(info).To(Equal(expected))
+	},
+		Entry("a Pod with all fields set",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					FaultDomain: fdbv1beta2.FoundationDBClusterFaultDomain{
+						Key: "foundationdb.org/none",
+					},
+				},
+				Status: fdbv1beta2.FoundationDBClusterStatus{
+					RequiredAddresses: fdbv1beta2.RequiredAddressSet{
+						NonTLS: true,
+					},
+				},
+			},
+			&corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test",
+					Labels: map[string]string{
+						fdbv1beta2.FDBProcessGroupIDLabel: "test",
+					},
+				},
+				Spec: corev1.PodSpec{
+					Containers: []corev1.Container{
+						{
+							Name: "foundationdb-kubernetes-sidecar",
+							Args: []string{
+								"--public-ip-family",
+								"4",
+							},
+						},
+					},
+				},
+				Status: corev1.PodStatus{
+					PodIPs: []corev1.PodIP{
+						{IP: "1.1.1.1"},
+					},
+				},
+			},
+			Info{
+				ID: "test",
+				Address: fdbv1beta2.ProcessAddress{
+					IPAddress: net.ParseIP("1.1.1.1"),
+					Port:      4501,
+				},
+				LocalityData: map[string]string{
+					fdbv1beta2.FDBLocalityZoneIDKey:  "test",
+					fdbv1beta2.FDBLocalityDNSNameKey: "",
+				},
+			},
+			false,
+		),
+		Entry("the sidecar is not reachable",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					FaultDomain: fdbv1beta2.FoundationDBClusterFaultDomain{
+						Key: "foundationdb.org/none",
+					},
+				},
+				Status: fdbv1beta2.FoundationDBClusterStatus{
+					RequiredAddresses: fdbv1beta2.RequiredAddressSet{
+						NonTLS: true,
+					},
+				},
+			},
+			&corev1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Annotations: map[string]string{
+						internal.MockUnreachableAnnotation: "true",
+					},
+				},
+			},
+			Info{},
+			true,
+		),
+	)
+
+	Describe("checkCoordinatorValidity", func() {
+		var status *fdbv1beta2.FoundationDBStatus
+		var coordinatorStatus map[string]bool
+
+		BeforeEach(func() {
+			cluster.Status.ProcessGroups = []*fdbv1beta2.ProcessGroupStatus{
+				{
+					ProcessGroupID: "test-1",
+				},
+				{
+					ProcessGroupID: "test-2",
+				},
+				{
+					ProcessGroupID: "test-3",
+				},
+			}
+
+			status = generateDefaultStatus(false)
+		})
+
+		JustBeforeEach(func() {
+			coordinatorStatus = make(map[string]bool, len(status.Client.Coordinators.Coordinators))
+			for _, coordinator := range status.Client.Coordinators.Coordinators {
+				coordinatorStatus[coordinator.Address.String()] = false
+			}
+		})
+
+		Context("with the default configuration", func() {
+			It("should report the coordinators as valid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeTrue())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+
+		When("an empty coordinator status is passed down", func() {
+			It("should report the coordinators as invalid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, nil)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeFalse())
+				Expect(err).To(HaveOccurred())
+			})
+		})
+
+		When("changing the coordinator selection setting", func() {
+			BeforeEach(func() {
+				cluster.Spec.CoordinatorSelection = []fdbv1beta2.CoordinatorSelectionSetting{
+					{
+						ProcessClass: fdbv1beta2.ProcessClassLog,
+					},
+				}
+			})
+
+			It("should report the coordinators as invalid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		When("one process is marked for exclusion", func() {
+			BeforeEach(func() {
+				process := status.Cluster.Processes["1"]
+				process.Excluded = true
+				status.Cluster.Processes["1"] = process
+			})
+
+			It("should report the coordinators as invalid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+
+		When("a process has an empty address", func() {
+			BeforeEach(func() {
+				status.Cluster.Processes["4"] = fdbv1beta2.FoundationDBStatusProcessInfo{}
+			})
+
+			It("should ignore the process", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeTrue())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		When("the coordinators are listening on TLS", func() {
+			BeforeEach(func() {
+				cluster.Spec.MainContainer.EnableTLS = true
+			})
+
+			It("should report the coordinators as invalid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeFalse())
+				Expect(err).NotTo(HaveOccurred())
+			})
+		})
+
+		When("a process has a process group status with a pending Pod", func() {
+			BeforeEach(func() {
+				status.Cluster.Processes["4"] = fdbv1beta2.FoundationDBStatusProcessInfo{
+					Address: fdbv1beta2.ProcessAddress{
+						IPAddress: net.ParseIP("1.1.1.4"),
+						Port:      4501,
+					},
+					Locality: map[string]string{
+						fdbv1beta2.FDBLocalityInstanceIDKey: "test-4",
+					},
+				}
+
+				cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups,
+					&fdbv1beta2.ProcessGroupStatus{
+						ProcessGroupID: "test-4",
+						ProcessGroupConditions: []*fdbv1beta2.ProcessGroupCondition{
+							{
+								ProcessGroupConditionType: fdbv1beta2.PodPending,
+								Timestamp:                 time.Now().Add(-45 * time.Minute).Unix(),
+							},
+						},
+					})
+			})
+
+			It("should ignore the process"+
+				"", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeTrue())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		When("a process misses the public-address flag", func() {
+			BeforeEach(func() {
+				status.Cluster.Processes["4"] = fdbv1beta2.FoundationDBStatusProcessInfo{
+					Address: fdbv1beta2.ProcessAddress{
+						IPAddress: net.ParseIP("1.1.1.4"),
+						Port:      4501,
+					},
+				}
+			})
+
+			It("should report that not all addresses are valid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeTrue())
+				Expect(addressesValid).To(BeFalse())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		Context("with a tester process without a command line field", func() {
+			BeforeEach(func() {
+				testProcess := fdbv1beta2.FoundationDBStatusProcessInfo{
+					Address: fdbv1beta2.ProcessAddress{
+						IPAddress: net.ParseIP("9.9.9.9"),
+					},
+					ProcessClass: fdbv1beta2.ProcessClassTest,
+				}
+				Expect(testProcess.Address.IsEmpty()).To(BeFalse())
+				status.Cluster.Processes[string(fdbv1beta2.ProcessClassTest)] = testProcess
+			})
+
+			It("should ignore the test process and report the coordinators as valid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeTrue())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		Context("with too few coordinators", func() {
+			BeforeEach(func() {
+				status.Client.Coordinators.Coordinators = status.Client.Coordinators.Coordinators[0:2]
+			})
+
+			It("should report the coordinators as not valid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		Context("with too few zones", func() {
+			BeforeEach(func() {
+				zone := ""
+				for _, process := range status.Cluster.Processes {
+					if process.Address.Equal(status.Client.Coordinators.Coordinators[0].Address) {
+						zone = process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]
+					}
+				}
+				for _, process := range status.Cluster.Processes {
+					if process.Address.Equal(status.Client.Coordinators.Coordinators[1].Address) {
+						process.Locality[fdbv1beta2.FDBLocalityZoneIDKey] = zone
+					}
+				}
+			})
+
+			It("should report the coordinators as not valid", func() {
+				coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+				Expect(coordinatorsValid).To(BeFalse())
+				Expect(addressesValid).To(BeTrue())
+				Expect(err).To(BeNil())
+			})
+		})
+
+		Context("with multiple regions", func() {
+			BeforeEach(func() {
+				cluster.Spec.DatabaseConfiguration.UsableRegions = 2
+
+				status.Cluster.Processes["4"] = generateDummyProcessInfo("test-4", "dc2", 4501, false)
+				status.Cluster.Processes["5"] = generateDummyProcessInfo("test-5", "dc2", 4501, false)
+				status.Cluster.Processes["6"] = generateDummyProcessInfo("test-6", "dc2", 4501, false)
+				status.Cluster.Processes["7"] = generateDummyProcessInfo("test-7", "dc3", 4501, false)
+				status.Cluster.Processes["8"] = generateDummyProcessInfo("test-8", "dc3", 4501, false)
+				status.Cluster.Processes["9"] = generateDummyProcessInfo("test-9", "dc3", 4501, false)
+
+				status.Client.Coordinators.Coordinators = append(status.Client.Coordinators.Coordinators,
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.4"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.5"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.6"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.7"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.8"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+					fdbv1beta2.FoundationDBStatusCoordinator{
+						Address: fdbv1beta2.ProcessAddress{
+							IPAddress: net.ParseIP("1.1.1.9"),
+							Port:      4501,
+						},
+						Reachable: true,
+					},
+				)
+			})
+
+			Context("with coordinators divided across three DCs", func() {
+				It("should report the coordinators as valid", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(coordinatorsValid).To(BeTrue())
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).To(BeNil())
+				})
+			})
+
+			Context("with coordinators divided across two DCs", func() {
+				BeforeEach(func() {
+					for _, process := range status.Cluster.Processes {
+						if process.Locality["dcid"] == "dc3" {
+							process.Locality["dcid"] = "dc1"
+						}
+					}
+				})
+				It("should report the coordinators as not valid", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(coordinatorsValid).To(BeFalse())
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).To(BeNil())
+				})
+			})
+		})
+
+		When("changing the TLS setting", func() {
+			BeforeEach(func() {
+				cluster.Status.RequiredAddresses = fdbv1beta2.RequiredAddressSet{
+					TLS:    true,
+					NonTLS: true,
+				}
+			})
+
+			When("TLS is disabled", func() {
+				BeforeEach(func() {
+					cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
+						EnableTLS: false,
+					}
+					cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
+						EnableTLS: false,
+					}
+				})
+
+				It("should report the coordinators addresses as valid", func() {
+					_, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).To(BeNil())
+				})
+
+				When("converting back to TLS", func() {
+					BeforeEach(func() {
+						cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
+							EnableTLS: true,
+						}
+						cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
+							EnableTLS: true,
+						}
+
+						status = generateDefaultStatus(true)
+					})
+
+					It("should report the coordinators addresses as valid", func() {
+						_, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+						Expect(addressesValid).To(BeTrue())
+						Expect(err).To(BeNil())
+					})
+				})
+			})
+
+			When("TLS is enabled", func() {
+				BeforeEach(func() {
+					cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
+						EnableTLS: true,
+					}
+					cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
+						EnableTLS: true,
+					}
+
+					status = generateDefaultStatus(true)
+				})
+
+				It("should report the coordinators addresses as valid", func() {
+					_, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).To(BeNil())
+				})
+
+				When("converting back to non-TLS", func() {
+					BeforeEach(func() {
+						cluster.Spec.MainContainer = fdbv1beta2.ContainerOverrides{
+							EnableTLS: false,
+						}
+
+						cluster.Spec.SidecarContainer = fdbv1beta2.ContainerOverrides{
+							EnableTLS: false,
+						}
+
+						status = generateDefaultStatus(false)
+					})
+
+					It("should report the coordinators addresses as valid", func() {
+						_, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+						Expect(addressesValid).To(BeTrue())
+						Expect(err).To(BeNil())
+					})
+				})
+			})
+		})
+
+		When("enabling DNS names in the cluster file", func() {
+			BeforeEach(func() {
+				cluster.Spec.Routing.UseDNSInClusterFile = pointer.Bool(true)
+			})
+
+			When("the pods do not have DNS names assigned", func() {
+				It("should report valid coordinators", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(coordinatorsValid).To(BeTrue())
+					Expect(addressesValid).To(BeTrue())
+				})
+			})
+
+			When("the pods have DNS names assigned", func() {
+				BeforeEach(func() {
+					for _, process := range status.Cluster.Processes {
+						process.Locality[fdbv1beta2.FDBLocalityDNSNameKey] = process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]
+					}
+				})
+
+				It("should reject coordinators based on IP addresses", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(coordinatorsValid).To(BeFalse())
+					Expect(addressesValid).To(BeTrue())
+				})
+			})
+
+			When("the pods have DNS names assigned and the coordinator have DNS names", func() {
+				BeforeEach(func() {
+					status.Client.Coordinators.Coordinators = nil
+
+					for _, process := range status.Cluster.Processes {
+						process.Locality[fdbv1beta2.FDBLocalityDNSNameKey] = process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]
+						status.Client.Coordinators.Coordinators = append(status.Client.Coordinators.Coordinators,
+							fdbv1beta2.FoundationDBStatusCoordinator{
+								Address: fdbv1beta2.ProcessAddress{
+									StringAddress: process.Locality[fdbv1beta2.FDBLocalityZoneIDKey],
+									Port:          4501,
+								},
+							})
+					}
+				})
+
+				It("should return that the coordinators are valid", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(coordinatorsValid).To(BeTrue())
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).NotTo(HaveOccurred())
+				})
+			})
+
+			When("the pods have DNS names assigned and the coordinator have DNS names but DNS is disabled", func() {
+				BeforeEach(func() {
+					status.Client.Coordinators.Coordinators = nil
+
+					for _, process := range status.Cluster.Processes {
+						process.Locality[fdbv1beta2.FDBLocalityDNSNameKey] = process.Locality[fdbv1beta2.FDBLocalityZoneIDKey]
+						status.Client.Coordinators.Coordinators = append(status.Client.Coordinators.Coordinators,
+							fdbv1beta2.FoundationDBStatusCoordinator{
+								Address: fdbv1beta2.ProcessAddress{
+									StringAddress: process.Locality[fdbv1beta2.FDBLocalityZoneIDKey],
+									Port:          4501,
+								},
+							})
+					}
+
+					cluster.Spec.Routing.UseDNSInClusterFile = pointer.Bool(false)
+				})
+
+				It("should return that the coordinators are invalid", func() {
+					coordinatorsValid, addressesValid, err := CheckCoordinatorValidity(logr.Discard(), cluster, status, coordinatorStatus)
+					Expect(coordinatorsValid).To(BeFalse())
+					Expect(addressesValid).To(BeTrue())
+					Expect(err).NotTo(HaveOccurred())
+				})
+			})
+		})
+	})
+})
diff --git a/internal/locality/suite_test.go b/internal/locality/suite_test.go
new file mode 100644
index 00000000..127b750e
--- /dev/null
+++ b/internal/locality/suite_test.go
@@ -0,0 +1,33 @@
+/*
+ * suite_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 locality
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestCmd(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "FDB locality")
+}
diff --git a/internal/pod_client_test.go b/internal/pod_client_test.go
index c9cf9be3..0a25cc42 100644
--- a/internal/pod_client_test.go
+++ b/internal/pod_client_test.go
@@ -21,13 +21,14 @@
 package internal
 
 import (
+	"net/http"
+	"net/url"
+	"time"
+
 	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
 	"github.com/hashicorp/go-retryablehttp"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"net/http"
-	"net/url"
-	"time"
 )
 
 var _ = Describe("pod_client", func() {
diff --git a/internal/restarts/restarts_test.go b/internal/restarts/restarts_test.go
new file mode 100644
index 00000000..90f4eafa
--- /dev/null
+++ b/internal/restarts/restarts_test.go
@@ -0,0 +1,72 @@
+/*
+ * restarts.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 restarts
+
+import (
+	fdbv1beta2 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta2"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("restarts", func() {
+	DescribeTable("when getting the filter conditions for a cluster", func(cluster *fdbv1beta2.FoundationDBCluster, expected map[fdbv1beta2.ProcessGroupConditionType]bool) {
+		Expect(GetFilterConditions(cluster)).To(Equal(expected))
+	},
+		Entry("when no upgrade is performed",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					Version: "7.1.25",
+				},
+				Status: fdbv1beta2.FoundationDBClusterStatus{
+					RunningVersion: "7.1.25",
+				},
+			},
+			map[fdbv1beta2.ProcessGroupConditionType]bool{
+				fdbv1beta2.IncorrectCommandLine: true,
+				fdbv1beta2.IncorrectPodSpec:     false,
+				fdbv1beta2.SidecarUnreachable:   false,
+			}),
+		Entry("when the running version is missing",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					Version: "7.1.25",
+				},
+			},
+			map[fdbv1beta2.ProcessGroupConditionType]bool{
+				fdbv1beta2.IncorrectCommandLine: true,
+				fdbv1beta2.IncorrectPodSpec:     false,
+				fdbv1beta2.SidecarUnreachable:   false,
+			}),
+		Entry("when an upgrade is performed",
+			&fdbv1beta2.FoundationDBCluster{
+				Spec: fdbv1beta2.FoundationDBClusterSpec{
+					Version: "7.1.25",
+				},
+				Status: fdbv1beta2.FoundationDBClusterStatus{
+					RunningVersion: "6.3.25",
+				},
+			},
+			map[fdbv1beta2.ProcessGroupConditionType]bool{
+				fdbv1beta2.IncorrectCommandLine: true,
+				fdbv1beta2.IncorrectPodSpec:     false,
+			}),
+	)
+})
diff --git a/internal/restarts/suite_test.go b/internal/restarts/suite_test.go
new file mode 100644
index 00000000..34db1667
--- /dev/null
+++ b/internal/restarts/suite_test.go
@@ -0,0 +1,33 @@
+/*
+ * suite_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2022 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 restarts
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestCmd(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "Restarts Suite")
+}
diff --git a/internal/test_helper.go b/internal/test_helper.go
index 63ae73f9..060dc308 100644
--- a/internal/test_helper.go
+++ b/internal/test_helper.go
@@ -29,9 +29,6 @@ import (
 
 // CreateDefaultCluster creates a default FoundationDBCluster for testing
 func CreateDefaultCluster() *fdbv1beta2.FoundationDBCluster {
-	trueValue := true
-	failureDetectionWindow := 1
-
 	return &fdbv1beta2.FoundationDBCluster{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      "operator-test-1",
@@ -48,8 +45,8 @@ func CreateDefaultCluster() *fdbv1beta2.FoundationDBCluster {
 			},
 			AutomationOptions: fdbv1beta2.FoundationDBClusterAutomationOptions{
 				Replacements: fdbv1beta2.AutomaticReplacementOptions{
-					Enabled:                     &trueValue,
-					FailureDetectionTimeSeconds: &failureDetectionWindow,
+					Enabled:                     pointer.Bool(true),
+					FailureDetectionTimeSeconds: pointer.Int(1),
 				},
 				WaitBetweenRemovalsSeconds: pointer.Int(0),
 			},
diff --git a/mock-kubernetes-client/client/client_test.go b/mock-kubernetes-client/client/client_test.go
index cef0ee68..4657c9da 100644
--- a/mock-kubernetes-client/client/client_test.go
+++ b/mock-kubernetes-client/client/client_test.go
@@ -22,6 +22,8 @@ package client
 
 import (
 	"context"
+	"fmt"
+	"k8s.io/utils/pointer"
 	"sort"
 	"time"
 
@@ -457,6 +459,58 @@ var _ = Describe("[mock client]", func() {
 			Expect(events.Items[0].ObjectMeta.Annotations).To(Equal(map[string]string{"anno": "value"}))
 		})
 	})
+
+	When("adding a custom create hook", func() {
+		var expectedPods = int32(4)
+
+		BeforeEach(func() {
+			client = NewMockClient(scheme.Scheme, func(ctx context.Context, client *MockClient, object ctrlClient.Object) error {
+				replicaSet, isReplicaSet := object.(*appsv1.ReplicaSet)
+				if !isReplicaSet {
+					return nil
+				}
+
+				expectedReplicas := pointer.Int32Deref(replicaSet.Spec.Replicas, 1)
+
+				for i := int32(0); i < expectedReplicas; i++ {
+					err := client.Create(ctx, &corev1.Pod{
+						ObjectMeta: metav1.ObjectMeta{
+							Namespace: replicaSet.Namespace,
+							Name:      fmt.Sprintf("%s-%d", replicaSet.Name, i),
+						},
+					})
+					_ = err
+				}
+
+				replicaSet.Labels = map[string]string{
+					"test": "success",
+				}
+
+				return nil
+			})
+
+			Expect(client.Create(context.TODO(), &appsv1.ReplicaSet{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unicorn",
+				},
+				Spec: appsv1.ReplicaSetSpec{
+					Replicas: pointer.Int32(expectedPods),
+				},
+			})).NotTo(HaveOccurred())
+		})
+
+		It("should run the webhook and create 4 Pods", func() {
+			podList := &corev1.PodList{}
+			Expect(client.List(context.TODO(), podList)).NotTo(HaveOccurred())
+			Expect(podList.Items).To(HaveLen(int(expectedPods)))
+
+			// Ensure the label is update
+			replicaSet := &appsv1.ReplicaSet{}
+			Expect(client.Get(context.TODO(), ctrlClient.ObjectKey{Name: "unicorn"}, replicaSet)).NotTo(HaveOccurred())
+			Expect(pointer.Int32Deref(replicaSet.Spec.Replicas, -1)).To(BeNumerically("==", expectedPods))
+			Expect(replicaSet.Labels).To(HaveKeyWithValue("test", "success"))
+		})
+	})
 })
 
 func sortPodsByName(pods *corev1.PodList) {
diff --git a/scripts/setup_e2e.sh b/scripts/setup_e2e.sh
new file mode 100755
index 00000000..61442037
--- /dev/null
+++ b/scripts/setup_e2e.sh
@@ -0,0 +1,148 @@
+#!/usr/bin/env bash
+
+# This source file is part of the FoundationDB open source project
+#
+# Copyright 2022 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.
+
+set -eo errexit
+
+# wait_for_node_setup is used to wait until the kubelet has started and the Pod CIDR is available.
+function wait_for_node_setup() {
+  for _ in {1..60}; do
+    if [ -n "$(kubectl --kubeconfig "${1}" get node  -o jsonpath='{range .items[*]}{.spec.podCIDR}{"\n"}')" ]; then
+      return
+    fi
+    sleep 1
+  done
+
+  exit 1
+}
+
+function preload_foundationdb_images() {
+  kind load docker-image "foundationdb/foundationdb:${2}" --name "${1}"
+  kind load docker-image "foundationdb/foundationdb-kubernetes-sidecar:${2}-1" --name "${1}"
+}
+
+# add_routes is used to allow Pods in the different clusters to communicate
+function add_routes() {
+  unset IFS
+  routes=$(kubectl --kubeconfig "${3}" get node "${2}" -o jsonpath='ip route add {.spec.podCIDR} via {.status.addresses[?(.type=="InternalIP")].address}')
+  echo "Connecting cluster ${1} to ${2}"
+
+  IFS=$'\n'
+  for n in $(kind get nodes --name "${1}"); do
+    for r in $routes; do
+      docker exec "${n}" bash -c "${r}"
+    done
+   done
+   unset IFS
+}
+
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &> /dev/null && pwd)
+cd "${SCRIPT_DIR}"
+
+# Kubernetes version for the Kind clusters
+KUBE_VERSION=${KUBE_VERSION:-"v1.24.7"}
+FDB_VERSION=${FDB_VERSION:-"7.1.25"}
+
+# To test a multi-region FDB cluster setup we need to have 4 Kubernetes clusters
+cluster1=${CLUSTER1:-cluster1}
+cluster2=${CLUSTER2:-cluster2}
+cluster3=${CLUSTER3:-cluster3}
+cluster4=${CLUSTER4:-cluster4}
+
+# For all the clusters we have to create tha according kubeconfig
+kubeconfig1=${KUBECONFIG1:-"${cluster1}.kubeconfig"}
+kubeconfig2=${KUBECONFIG2:-"${cluster2}.kubeconfig"}
+kubeconfig3=${KUBECONFIG3:-"${cluster3}.kubeconfig"}
+kubeconfig4=${KUBECONFIG4:-"${cluster4}.kubeconfig"}
+
+# We have to export them for envsubst
+export KUBE_VERSION
+export reg_name
+export reg_port
+
+echo "Creating Kind clusters"
+envsubst < "${cluster1}.yml" | kind create cluster --name "${cluster1}" --config=-
+envsubst < "${cluster2}.yml" | kind create cluster --name "${cluster2}" --config=-
+envsubst < "${cluster3}.yml" | kind create cluster --name "${cluster3}" --config=-
+envsubst < "${cluster4}.yml" | kind create cluster --name "${cluster4}" --config=-
+
+unset KUBE_VERSION reg_name reg_port
+
+kind get kubeconfig --name "${cluster1}" > "${kubeconfig1}"
+kind get kubeconfig --name "${cluster2}" > "${kubeconfig2}"
+kind get kubeconfig --name "${cluster3}" > "${kubeconfig3}"
+kind get kubeconfig --name "${cluster4}" > "${kubeconfig4}"
+
+echo "Preload FoundationDB images"
+docker pull "foundationdb/foundationdb:${FDB_VERSION}"
+docker pull "foundationdb/foundationdb-kubernetes-sidecar:${FDB_VERSION}-1"
+
+preload_foundationdb_images "${cluster1}" "${FDB_VERSION}"
+preload_foundationdb_images "${cluster2}" "${FDB_VERSION}"
+preload_foundationdb_images "${cluster3}" "${FDB_VERSION}"
+preload_foundationdb_images "${cluster4}" "${FDB_VERSION}"
+
+echo "Waiting for Kind nodes to be initialized"
+wait_for_node_setup "${kubeconfig1}"
+wait_for_node_setup "${kubeconfig2}"
+wait_for_node_setup "${kubeconfig3}"
+wait_for_node_setup "${kubeconfig4}"
+
+echo "Connect Kind clusters"
+add_routes "${cluster1}" "${cluster2}-control-plane" "${kubeconfig2}"
+add_routes "${cluster1}" "${cluster3}-control-plane" "${kubeconfig3}"
+add_routes "${cluster1}" "${cluster4}-control-plane" "${kubeconfig4}"
+
+add_routes "${cluster2}" "${cluster1}-control-plane" "${kubeconfig1}"
+add_routes "${cluster2}" "${cluster3}-control-plane" "${kubeconfig3}"
+add_routes "${cluster2}" "${cluster4}-control-plane" "${kubeconfig4}"
+
+add_routes "${cluster3}" "${cluster1}-control-plane" "${kubeconfig1}"
+add_routes "${cluster3}" "${cluster2}-control-plane" "${kubeconfig2}"
+add_routes "${cluster3}" "${cluster4}-control-plane" "${kubeconfig4}"
+
+add_routes "${cluster4}" "${cluster1}-control-plane" "${kubeconfig1}"
+add_routes "${cluster4}" "${cluster2}-control-plane" "${kubeconfig2}"
+add_routes "${cluster4}" "${cluster3}-control-plane" "${kubeconfig3}"
+
+make -C "${SCRIPT_DIR}/.." container-build
+kind load docker-image "fdb-kubernetes-operator:latest" --name "${cluster1}"
+kind load docker-image "fdb-kubernetes-operator:latest" --name "${cluster2}"
+kind load docker-image "fdb-kubernetes-operator:latest" --name "${cluster3}"
+kind load docker-image "fdb-kubernetes-operator:latest" --name "${cluster4}"
+
+echo "Installing the test certificates"
+KUBECONFIG=${SCRIPT_DIR}/${kubeconfig1} ${SCRIPT_DIR}/../config/test-certs/generate_secrets.bash
+KUBECONFIG=${SCRIPT_DIR}/${kubeconfig2} ${SCRIPT_DIR}/../config/test-certs/generate_secrets.bash
+KUBECONFIG=${SCRIPT_DIR}/${kubeconfig3} ${SCRIPT_DIR}/../config/test-certs/generate_secrets.bash
+KUBECONFIG=${SCRIPT_DIR}/${kubeconfig4} ${SCRIPT_DIR}/../config/test-certs/generate_secrets.bash
+
+echo "Install the CRDs and the operator in every Kind cluster"
+KUBECTL_ARGS="--kubeconfig ${SCRIPT_DIR}/${kubeconfig1}" make -C "${SCRIPT_DIR}/.." deploy
+KUBECTL_ARGS="--kubeconfig ${SCRIPT_DIR}/${kubeconfig2}" make -C "${SCRIPT_DIR}/.." deploy
+KUBECTL_ARGS="--kubeconfig ${SCRIPT_DIR}/${kubeconfig3}" make -C "${SCRIPT_DIR}/.." deploy
+KUBECTL_ARGS="--kubeconfig ${SCRIPT_DIR}/${kubeconfig4}" make -C "${SCRIPT_DIR}/.." deploy
+
+# TODO(johscheuer): Next step is to add e2e tests for this setup.
+
+echo "For clean up run kind delete clusters ${cluster1} ${cluster2} ${cluster3} ${cluster4}"
+
+echo "For usage run the following export commands:
+export KUBECONFIG1=${SCRIPT_DIR}/${kubeconfig1}
+export KUBECONFIG2=${SCRIPT_DIR}/${kubeconfig2}
+export KUBECONFIG3=${SCRIPT_DIR}/${kubeconfig3}
+export KUBECONFIG4=${SCRIPT_DIR}/${kubeconfig4}"