Repository: FoundationDB/fdb-kubernetes-operator
Pull request #1888 (2023-11-17 11:55:18)

Add cache to plugin
# Description

Fixes: https://github.com/FoundationDB/fdb-kubernetes-operator/issues/1675
Fixes: https://github.com/FoundationDB/fdb-kubernetes-operator/issues/1496 (fetching the same cluster multiple times is not an issue as they will be read from the cache)

## Type of change

*Please select one of the options below.*

- New feature (non-breaking change which adds functionality)

## Discussion

-

## Testing

Ran some manual tests.

## Documentation

-

## Follow-up

We should remove some of the logic in the plugin to make sure we make use of the get calls and remove any caches we put on top of the controller client.



Patch:
diff --git a/api/v1beta2/foundationdbcluster_types.go b/api/v1beta2/foundationdbcluster_types.go
index f21b9053..01619b83 100644
--- a/api/v1beta2/foundationdbcluster_types.go
+++ b/api/v1beta2/foundationdbcluster_types.go
@@ -584,7 +584,10 @@ func (processGroupStatus *ProcessGroupStatus) AllAddressesExcluded(logger logr.L
 	}
 
 	localityExclusionString := processGroupStatus.GetExclusionString()
-	if isRemaining, isPresent := remainingMap[localityExclusionString]; isPresent && !isRemaining {
+	if isRemaining, isPresent := remainingMap[localityExclusionString]; isPresent {
+		if isRemaining {
+			return false, fmt.Errorf("process has missing exclusion string in exclusion results: %s", localityExclusionString)
+		}
 		logger.V(1).Info("process group is fully excluded based on locality based exclusions", "processGroupID", processGroupStatus.ProcessGroupID, "exclusionString", localityExclusionString)
 		return true, nil
 	}
@@ -1313,6 +1316,7 @@ func (cluster *FoundationDBCluster) calculateProcessCountFromRole(count int, alt
 	if count < 0 {
 		return 0
 	}
+
 	return count
 }
 
@@ -1366,9 +1370,7 @@ func (cluster *FoundationDBCluster) GetProcessCountsWithDefaults() (ProcessCount
 	roleCounts := cluster.GetRoleCountsWithDefaults()
 	processCounts := cluster.Spec.ProcessCounts.DeepCopy()
 
-	isSatellite := false
-	isMain := false
-
+	var isSatellite, isMain bool
 	satelliteLogs := 0
 	for _, region := range cluster.Spec.DatabaseConfiguration.Regions {
 		for _, dataCenter := range region.DataCenters {
@@ -1396,12 +1398,14 @@ func (cluster *FoundationDBCluster) GetProcessCountsWithDefaults() (ProcessCount
 		processCounts.Storage = cluster.calculateProcessCount(false,
 			roleCounts.Storage)
 	}
+
 	if processCounts.Log == 0 {
 		processCounts.Log = cluster.calculateProcessCount(true,
 			cluster.calculateProcessCountFromRole(roleCounts.Logs+satelliteLogs, processCounts.Log),
 			cluster.calculateProcessCountFromRole(roleCounts.RemoteLogs, processCounts.Log),
 		)
 	}
+
 	if processCounts.Stateless == 0 {
 		primaryStatelessCount := cluster.calculateProcessCountFromRole(1, processCounts.Master) +
 			cluster.calculateProcessCountFromRole(1, processCounts.ClusterController) +
diff --git a/controllers/exclude_processes.go b/controllers/exclude_processes.go
index 4ea4074c..fff71be5 100644
--- a/controllers/exclude_processes.go
+++ b/controllers/exclude_processes.go
@@ -75,18 +75,19 @@ func (e excludeProcesses) reconcile(_ context.Context, r *FoundationDBClusterRec
 
 		desiredProcessesMap := desiredProcesses.Map()
 		for processClass := range fdbProcessesToExcludeByClass {
+			contextLogger := logger.WithValues("processClass", processClass)
 			ongoingExclusions := ongoingExclusionsByClass[processClass]
 			processesToExclude := fdbProcessesToExcludeByClass[processClass]
 
-			allowedExclusions, missingProcesses := canExcludeNewProcesses(logger, cluster, processClass, desiredProcessesMap[processClass], ongoingExclusions, r.InSimulation)
+			allowedExclusions, missingProcesses := getAllowedExclusionsAndMissingProcesses(contextLogger, cluster, processClass, desiredProcessesMap[processClass], ongoingExclusions, r.InSimulation)
 			if allowedExclusions <= 0 {
-				logger.Info("Waiting for missing processes before continuing with the exclusion", "processClass", processClass, "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
+				contextLogger.Info("Waiting for missing processes before continuing with the exclusion", "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
 				continue
 			}
 
 			// If we are not able to exclude all processes at once print a log message.
 			if len(processesToExclude) > allowedExclusions {
-				logger.Info("Some processes are still missing but continuing with the exclusion", "processClass", processClass, "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
+				contextLogger.Info("Some processes are still missing but continuing with the exclusion", "missingProcesses", missingProcesses, "addressesToExclude", processesToExclude, "allowedExclusions", allowedExclusions, "ongoingExclusions", ongoingExclusions)
 			}
 
 			if len(processesToExclude) < allowedExclusions {
@@ -184,22 +185,23 @@ func getProcessesToExclude(exclusions []fdbv1beta2.ProcessAddress, cluster *fdbv
 	return fdbProcessesToExcludeByClass, ongoingExclusionsByClass
 }
 
-// canExcludeNewProcesses will check if new processes for the specified process class can be excluded. The calculation takes
+// getAllowedExclusionsAndMissingProcesses will check if new processes for the specified process class can be excluded. The calculation takes
 // the current ongoing exclusions into account and the desired process count. If there are process groups that have
 // the MissingProcesses condition this method will forbid exclusions until all process groups with this condition have
 // this condition for longer than ignoreMissingProcessDuration. The idea behind this is to try to exclude as many processes
 // at once e.g. to reduce the number of recoveries and data movement.
-func canExcludeNewProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, desiredProcessCount int, ongoingExclusions int, inSimulation bool) (int, []fdbv1beta2.ProcessGroupID) {
+func getAllowedExclusionsAndMissingProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, processClass fdbv1beta2.ProcessClass, desiredProcessCount int, ongoingExclusions int, inSimulation bool) (int, []fdbv1beta2.ProcessGroupID) {
 	// Block excludes on missing processes not marked for removal unless they are missing for a long time and the process might be broken
 	// or the namespace quota was hit.
 	missingProcesses := make([]fdbv1beta2.ProcessGroupID, 0)
-	validProcesses := make([]fdbv1beta2.ProcessGroupID, 0)
+	var validProcesses int
 
 	exclusionsAllowed := true
 	for _, processGroup := range cluster.Status.ProcessGroups {
 		if processGroup.ProcessClass != processClass {
 			continue
 		}
+
 		// Those should already be filtered out by the previous method.
 		if processGroup.IsMarkedForRemoval() && processGroup.IsExcluded() {
 			continue
@@ -217,33 +219,27 @@ func canExcludeNewProcesses(logger logr.Logger, cluster *fdbv1beta2.FoundationDB
 			continue
 		}
 
-		validProcesses = append(validProcesses, processGroup.ProcessGroupID)
+		validProcesses++
 	}
 
 	if !exclusionsAllowed {
-		logger.Info("Found at least one missing process, that was not missing for a long time", "missingProcesses", missingProcesses)
+		logger.Info("Found at least one missing process, that was not missing for more than 5 minutes", "missingProcesses", missingProcesses)
 		return 0, missingProcesses
 	}
 
-	logger.V(1).Info("canExcludeNewProcesses", "validProcesses", len(validProcesses), "desiredProcessCount", desiredProcessCount, "ongoingExclusions", ongoingExclusions)
-
-	// The assumption here is that we will only exclude a process if there is a replacement ready for it. We could relax
-	// this requirement in the future and take the fault tolerance into account. We add the desired fault tolerance to
-	// have some buffer to prevent cases where the operator might need to exclude more processes but there are more missing
-	// processes.
-	allowedExclusions := cluster.DesiredFaultTolerance() + len(validProcesses) - desiredProcessCount - ongoingExclusions
-
-	// If automatic replacements are enabled and the allowed exclusions is less than or equal to 0, we have to check
-	// how many processes are missing and if more processes are missing than the automatic replacements is allowed to
-	// replace, we will allow exclusions for the count of automatic replacements removing the already ongoing exclusions.
-	// This code should make sure that the operator can automatically replace processes, even in the case where multiple
-	// processes are failing.
-	if cluster.GetEnableAutomaticReplacements() && allowedExclusions <= 0 {
-		automaticReplacements := cluster.GetMaxConcurrentAutomaticReplacements()
-		if len(missingProcesses) > automaticReplacements {
-			return automaticReplacements - ongoingExclusions, missingProcesses
-		}
+	return getAllowedExclusions(logger, validProcesses, desiredProcessCount, ongoingExclusions, cluster.DesiredFaultTolerance()), missingProcesses
+}
+
+// getAllowedExclusions will return the number of allowed exclusions. If no exclusions are allowed this method will return a 0.
+// The assumption here is that we will only exclude a process if there is a replacement ready for it. We add the desired fault
+// tolerance to have some buffer to prevent cases where the operator might need to exclude more processes but there are more
+// missing processes.
+func getAllowedExclusions(logger logr.Logger, validProcesses int, desiredProcessCount int, ongoingExclusions int, faultTolerance int) int {
+	logger.V(1).Info("getAllowedExclusions", "validProcesses", validProcesses, "desiredProcessCount", desiredProcessCount, "ongoingExclusions", ongoingExclusions, "faultTolerance", faultTolerance)
+	allowedExclusions := validProcesses + faultTolerance - desiredProcessCount - ongoingExclusions
+	if allowedExclusions < 0 {
+		return 0
 	}
 
-	return allowedExclusions, missingProcesses
+	return allowedExclusions
 }
diff --git a/controllers/remove_process_groups.go b/controllers/remove_process_groups.go
index 45a651c7..3b0c4bd3 100644
--- a/controllers/remove_process_groups.go
+++ b/controllers/remove_process_groups.go
@@ -134,7 +134,7 @@ func (u removeProcessGroups) reconcile(ctx context.Context, r *FoundationDBClust
 	// This will return a map of the newly removed ProcessGroups and the ProcessGroups with the ResourcesTerminating condition
 	removedProcessGroups := r.removeProcessGroups(ctx, logger, cluster, zoneRemovals, zonedRemovals[removals.TerminatingZone])
 
-	err = includeProcessGroup(ctx, r, cluster, removedProcessGroups)
+	err = includeProcessGroup(ctx, logger, r, cluster, removedProcessGroups, status)
 	if err != nil {
 		return &requeue{curError: err}
 	}
@@ -251,14 +251,18 @@ func confirmRemoval(ctx context.Context, logger logr.Logger, r *FoundationDBClus
 	return true, canBeIncluded, nil
 }
 
-func includeProcessGroup(ctx context.Context, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster, removedProcessGroups map[fdbv1beta2.ProcessGroupID]bool) error {
+func includeProcessGroup(ctx context.Context, logger logr.Logger, r *FoundationDBClusterReconciler, cluster *fdbv1beta2.FoundationDBCluster, removedProcessGroups map[fdbv1beta2.ProcessGroupID]bool, status *fdbv1beta2.FoundationDBStatus) error {
 	adminClient, err := r.getDatabaseClientProvider().GetAdminClient(cluster, r)
 	if err != nil {
 		return err
 	}
 	defer adminClient.Close()
 
-	fdbProcessesToInclude := getProcessesToInclude(cluster, removedProcessGroups)
+	fdbProcessesToInclude, err := getProcessesToInclude(logger, cluster, removedProcessGroups, status)
+	if err != nil {
+		return err
+	}
+
 	if len(fdbProcessesToInclude) > 0 {
 		r.Recorder.Event(cluster, corev1.EventTypeNormal, "IncludingProcesses", fmt.Sprintf("Including removed processes: %v", fdbProcessesToInclude))
 
@@ -276,21 +280,42 @@ func includeProcessGroup(ctx context.Context, r *FoundationDBClusterReconciler,
 	return nil
 }
 
-func getProcessesToInclude(cluster *fdbv1beta2.FoundationDBCluster, removedProcessGroups map[fdbv1beta2.ProcessGroupID]bool) []fdbv1beta2.ProcessAddress {
+func getProcessesToInclude(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, removedProcessGroups map[fdbv1beta2.ProcessGroupID]bool, status *fdbv1beta2.FoundationDBStatus) ([]fdbv1beta2.ProcessAddress, error) {
 	fdbProcessesToInclude := make([]fdbv1beta2.ProcessAddress, 0)
 
 	if len(removedProcessGroups) == 0 {
-		return fdbProcessesToInclude
+		return fdbProcessesToInclude, nil
+	}
+
+	excludedServers, err := fdbstatus.GetExclusions(status)
+	if err != nil {
+		return fdbProcessesToInclude, fmt.Errorf("unable to get excluded servers from status, %w", err)
+	}
+	excludedServersMap := make(map[string]fdbv1beta2.None, len(excludedServers))
+	for _, excludedServer := range excludedServers {
+		excludedServersMap[excludedServer.String()] = fdbv1beta2.None{}
 	}
 
 	idx := 0
 	for _, processGroup := range cluster.Status.ProcessGroups {
 		if processGroup.IsMarkedForRemoval() && removedProcessGroups[processGroup.ProcessGroupID] {
-			if cluster.UseLocalitiesForExclusion() {
-				fdbProcessesToInclude = append(fdbProcessesToInclude, fdbv1beta2.ProcessAddress{StringAddress: processGroup.GetExclusionString()})
+			foundInExcludedServerList := false
+			exclusionString := processGroup.GetExclusionString()
+			if _, ok := excludedServersMap[exclusionString]; ok {
+				fdbProcessesToInclude = append(fdbProcessesToInclude, fdbv1beta2.ProcessAddress{StringAddress: exclusionString})
+				foundInExcludedServerList = true
 			}
 			for _, pAddr := range processGroup.Addresses {
-				fdbProcessesToInclude = append(fdbProcessesToInclude, fdbv1beta2.ProcessAddress{IPAddress: net.ParseIP(pAddr)})
+				if _, ok := excludedServersMap[pAddr]; ok {
+					fdbProcessesToInclude = append(fdbProcessesToInclude, fdbv1beta2.ProcessAddress{IPAddress: net.ParseIP(pAddr)})
+					foundInExcludedServerList = true
+				}
+			}
+			if !foundInExcludedServerList {
+				// This means that the process is marked for exclusion and is also removed in the previous step but is missing
+				// its entry in the excluded servers in the status. This should not throw an error as this will block the
+				// inclusion for other processes, but we should have a record of this event happening in the logs.
+				logger.Info("processGroup is included but is missing from excluded server list", "processGroup", processGroup)
 			}
 			continue
 		}
@@ -301,7 +326,7 @@ func getProcessesToInclude(cluster *fdbv1beta2.FoundationDBCluster, removedProce
 	// Remove the trailing duplicates.
 	cluster.Status.ProcessGroups = cluster.Status.ProcessGroups[:idx]
 
-	return fdbProcessesToInclude
+	return fdbProcessesToInclude, nil
 }
 
 func (r *FoundationDBClusterReconciler) getProcessGroupsToRemove(logger logr.Logger, cluster *fdbv1beta2.FoundationDBCluster, remainingMap map[string]bool, cordSet map[string]fdbv1beta2.None) (bool, bool, []*fdbv1beta2.ProcessGroupStatus) {
diff --git a/kubectl-fdb/Readme.md b/kubectl-fdb/Readme.md
index 68cb50df..d358c76e 100644
--- a/kubectl-fdb/Readme.md
+++ b/kubectl-fdb/Readme.md
@@ -16,6 +16,9 @@ Install from release:
 
 ```bash
 pushd $TMPDIR
+if [ -f "$TMPDIR/kubectl-fdb-version.txt" ]; then
+   rm "$TMPDIR/kubectl-fdb-version.txt"
+fi
 OS=$(uname)
 ARCH=$(uname -m)
 VERSION="$(curl -s "https://api.github.com/repos/FoundationDB/fdb-kubernetes-operator/releases/latest" | jq -r '.tag_name')"
@@ -51,3 +54,11 @@ Run `kubectl fdb help` to get the latest help.
 We have a list of [planned operations](https://github.com/FoundationDB/fdb-kubernetes-operator/issues?q=is%3Aissue+is%3Aopen+label%3Aplugin)
 that we want to implement.
 Raise an issue if you miss a specific command to operate FDB on Kubernetes.
+
+## Plugin Version Check
+
+Per default the plugin will check if there is a newer version present.
+For this the plugin will contact the GitHub API and check for the latest release.
+If a newer version was found the command will exit with the request to update the plugin.
+The version check can be disabled by setting `--version-check=false` when issuing a command.
+The plugin will create a local cache in the temp directory to reduce the calls to the GitHub API, the file is located under `$TMPDIR/kubectl-fdb-version.txt`.
diff --git a/kubectl-fdb/cmd/analyze.go b/kubectl-fdb/cmd/analyze.go
index ff15ae35..c36b6cff 100644
--- a/kubectl-fdb/cmd/analyze.go
+++ b/kubectl-fdb/cmd/analyze.go
@@ -72,6 +72,11 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
+			shouldAnalyzeStatus, err := cmd.Flags().GetBool("analyze-status")
+			if err != nil {
+				return err
+			}
+
 			ignoreConditions, err := cmd.Flags().GetStringArray("ignore-condition")
 			if err != nil {
 				return err
@@ -90,7 +95,7 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return cmd.Help()
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
@@ -139,6 +144,10 @@ func newAnalyzeCmd(streams genericclioptions.IOStreams) *cobra.Command {
 					errs = append(errs, err)
 				}
 
+				if !shouldAnalyzeStatus {
+					continue
+				}
+
 				err = analyzeStatus(cmd, config, clientSet, kubeClient, cluster, autoFix)
 				if err != nil {
 					errs = append(errs, err)
@@ -182,6 +191,7 @@ kubectl fdb analyze --ignore-removals=false sample-cluster-1
 
 	cmd.Flags().Bool("auto-fix", false, "defines if the analyze tasks should try to fix the found issues (e.g. replace Pods).")
 	cmd.Flags().Bool("all-clusters", false, "defines all clusters in the given namespace should be analyzed.")
+	cmd.Flags().Bool("analyze-status", false, "defines if the command should also analyze the machine-readable status of the provided cluster(s).")
 	// We might want to move this into the root cmd if we need this in multiple places
 	cmd.Flags().Bool("no-color", false, "Disable color output.")
 	cmd.Flags().StringArray("ignore-condition", nil, "specify which process group conditions should be ignored and not be printed to stdout.")
diff --git a/kubectl-fdb/cmd/buggify_crash_loop.go b/kubectl-fdb/cmd/buggify_crash_loop.go
index 943cf7b4..23a351ad 100644
--- a/kubectl-fdb/cmd/buggify_crash_loop.go
+++ b/kubectl-fdb/cmd/buggify_crash_loop.go
@@ -61,7 +61,7 @@ func newBuggifyCrashLoop(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/buggify_empty_monitor_conf.go b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
index 9ae6baf8..5ca001f1 100644
--- a/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
+++ b/kubectl-fdb/cmd/buggify_empty_monitor_conf.go
@@ -52,7 +52,7 @@ func newBuggifyEmptyMonitorConf(streams genericclioptions.IOStreams) *cobra.Comm
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/buggify_no_schedule.go b/kubectl-fdb/cmd/buggify_no_schedule.go
index 48656521..ca178040 100644
--- a/kubectl-fdb/cmd/buggify_no_schedule.go
+++ b/kubectl-fdb/cmd/buggify_no_schedule.go
@@ -56,7 +56,7 @@ func newBuggifyNoSchedule(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/configuration.go b/kubectl-fdb/cmd/configuration.go
index e3cda9b8..7bf6cc79 100644
--- a/kubectl-fdb/cmd/configuration.go
+++ b/kubectl-fdb/cmd/configuration.go
@@ -52,7 +52,7 @@ func newConfigurationCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/cordon.go b/kubectl-fdb/cmd/cordon.go
index db063c8e..27248356 100644
--- a/kubectl-fdb/cmd/cordon.go
+++ b/kubectl-fdb/cmd/cordon.go
@@ -64,7 +64,7 @@ func newCordonCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/deprecation.go b/kubectl-fdb/cmd/deprecation.go
index fefe1920..5f45d4a3 100644
--- a/kubectl-fdb/cmd/deprecation.go
+++ b/kubectl-fdb/cmd/deprecation.go
@@ -57,7 +57,7 @@ Deprecated settings that should be replace by a newer setting (or removed) are p
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/exclusion_status.go b/kubectl-fdb/cmd/exclusion_status.go
index 54fc5813..39c60de6 100644
--- a/kubectl-fdb/cmd/exclusion_status.go
+++ b/kubectl-fdb/cmd/exclusion_status.go
@@ -41,7 +41,7 @@ func newExclusionStatusCmd(streams genericclioptions.IOStreams) *cobra.Command {
 		Use:   "exclusion-status",
 		Short: "Get the exclusion status for all excluded processes.",
 		Long:  "Get the exclusion status for all excluded processes.",
-		Args:  cobra.ExactValidArgs(1),
+		Args:  cobra.MatchAll(cobra.ExactArgs(1), cobra.OnlyValidArgs),
 		RunE: func(cmd *cobra.Command, args []string) error {
 			ignoreFullyExcluded, err := cmd.Flags().GetBool("ignore-fully-excluded")
 			if err != nil {
@@ -63,7 +63,7 @@ func newExclusionStatusCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/exec.go b/kubectl-fdb/cmd/exec.go
index dfde0c40..cba42df5 100644
--- a/kubectl-fdb/cmd/exec.go
+++ b/kubectl-fdb/cmd/exec.go
@@ -53,7 +53,7 @@ func newExecCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/fix_coordinator_ips.go b/kubectl-fdb/cmd/fix_coordinator_ips.go
index 07952c31..1991ceac 100644
--- a/kubectl-fdb/cmd/fix_coordinator_ips.go
+++ b/kubectl-fdb/cmd/fix_coordinator_ips.go
@@ -59,7 +59,7 @@ func newFixCoordinatorIPsCmd(streams genericclioptions.IOStreams) *cobra.Command
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/k8s_client.go b/kubectl-fdb/cmd/k8s_client.go
index a4ba9716..e2898c55 100644
--- a/kubectl-fdb/cmd/k8s_client.go
+++ b/kubectl-fdb/cmd/k8s_client.go
@@ -22,10 +22,13 @@ package cmd
 
 import (
 	"bytes"
-	ctx "context"
+	"context"
 	"fmt"
+	"github.com/go-logr/logr"
 	"io"
+	"k8s.io/klog/v2"
 	"math/rand"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
 	"strings"
 
 	fdbv1beta1 "github.com/FoundationDB/fdb-kubernetes-operator/api/v1beta1"
@@ -45,7 +48,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
-func getKubeClient(o *fdbBOptions) (client.Client, error) {
+func getKubeClient(ctx context.Context, o *fdbBOptions) (client.Client, error) {
 	config, err := o.configFlags.ToRESTConfig()
 	if err != nil {
 		return nil, err
@@ -56,7 +59,46 @@ func getKubeClient(o *fdbBOptions) (client.Client, error) {
 	utilruntime.Must(fdbv1beta1.AddToScheme(scheme))
 	utilruntime.Must(fdbv1beta2.AddToScheme(scheme))
 
-	return client.New(config, client.Options{Scheme: scheme})
+	// Don't printout any log messages from client-go.
+	klog.SetLogger(logr.Discard())
+
+	internalClient, err := client.NewWithWatch(config, client.Options{
+		Scheme: scheme,
+		Opts: client.WarningHandlerOptions{
+			SuppressWarnings: true,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	namespace, err := getNamespace(*o.configFlags.Namespace)
+	if err != nil {
+		return nil, err
+	}
+
+	cacheBuilder := cache.MultiNamespacedCacheBuilder([]string{namespace})
+	internalCache, err := cacheBuilder(config, cache.Options{
+		Scheme: scheme,
+		Mapper: internalClient.RESTMapper(),
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	// Make sure the internal cache is started.
+	go func() {
+		_ = internalCache.Start(ctx)
+	}()
+
+	internalCache.WaitForCacheSync(ctx)
+
+	return client.NewDelegatingClient(client.NewDelegatingClientInput{
+		CacheReader:       internalCache,
+		Client:            internalClient,
+		UncachedObjects:   nil,
+		CacheUnstructured: false,
+	})
 }
 
 func getNamespace(namespace string) (string, error) {
@@ -69,9 +111,9 @@ func getNamespace(namespace string) (string, error) {
 		return "", err
 	}
 
-	if context, ok := clientCfg.Contexts[clientCfg.CurrentContext]; ok {
-		if context.Namespace != "" {
-			return context.Namespace, nil
+	if ctx, ok := clientCfg.Contexts[clientCfg.CurrentContext]; ok {
+		if ctx.Namespace != "" {
+			return ctx.Namespace, nil
 		}
 	}
 
@@ -80,13 +122,13 @@ func getNamespace(namespace string) (string, error) {
 
 func getOperator(kubeClient client.Client, operatorName string, namespace string) (*appsv1.Deployment, error) {
 	operator := &appsv1.Deployment{}
-	err := kubeClient.Get(ctx.Background(), client.ObjectKey{Namespace: namespace, Name: operatorName}, operator)
+	err := kubeClient.Get(context.Background(), client.ObjectKey{Namespace: namespace, Name: operatorName}, operator)
 	return operator, err
 }
 
 func loadCluster(kubeClient client.Client, namespace string, clusterName string) (*fdbv1beta2.FoundationDBCluster, error) {
 	cluster := &fdbv1beta2.FoundationDBCluster{}
-	err := kubeClient.Get(ctx.Background(), types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster)
+	err := kubeClient.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: clusterName}, cluster)
 	if err != nil {
 		return nil, err
 	}
@@ -94,12 +136,13 @@ func loadCluster(kubeClient client.Client, namespace string, clusterName string)
 	if err != nil {
 		return nil, err
 	}
+
 	return cluster, err
 }
 
 func getNodes(kubeClient client.Client, nodeSelector map[string]string) ([]string, error) {
 	var nodesList corev1.NodeList
-	err := kubeClient.List(ctx.Background(), &nodesList, client.MatchingLabels(nodeSelector))
+	err := kubeClient.List(context.Background(), &nodesList, client.MatchingLabels(nodeSelector))
 	if err != nil {
 		return []string{}, err
 	}
@@ -116,7 +159,7 @@ func getNodes(kubeClient client.Client, nodeSelector map[string]string) ([]strin
 func getPodsForCluster(kubeClient client.Client, cluster *fdbv1beta2.FoundationDBCluster) (*corev1.PodList, error) {
 	var podList corev1.PodList
 	err := kubeClient.List(
-		ctx.Background(),
+		context.Background(),
 		&podList,
 		client.MatchingLabels(cluster.GetMatchLabels()),
 		client.InNamespace(cluster.GetNamespace()))
diff --git a/kubectl-fdb/cmd/plugin_version_checker.go b/kubectl-fdb/cmd/plugin_version_checker.go
new file mode 100644
index 00000000..6ab45e76
--- /dev/null
+++ b/kubectl-fdb/cmd/plugin_version_checker.go
@@ -0,0 +1,128 @@
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/hashicorp/go-retryablehttp"
+	"io"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+// VersionChecker interface to help us mock test the version checker
+type VersionChecker interface {
+	getLatestPluginVersion() (string, error)
+}
+
+// RealVersionChecker will do actual call to GitHub API to get version
+type RealVersionChecker struct{}
+
+// KubectlFbReleaseURL is the public GitHub URL we read the latest version from
+const KubectlFbReleaseURL = "https://api.github.com/repos/FoundationDB/fdb-kubernetes-operator/releases/latest"
+
+// LocalTempVersionFileName is where we cache plugin version for 24 hours, so we don't call GitHub for every single command(also rate-limit issue on GitHub api calls)
+const LocalTempVersionFileName = "kubectl-fdb-version.txt"
+
+// PluginVersionDetails Contains [partial] plugin version details from GitHub
+type PluginVersionDetails struct {
+	ID      int64  `json:"id"`
+	Version string `json:"tag_name"`
+	Name    string `json:"name"`
+}
+
+// It reads the latest fdb plugin version from local temp file,
+// if not exists, gets it from GitHub and store it locally and then returns it
+func (versionChecker *RealVersionChecker) getLatestPluginVersion() (string, error) {
+	fileName := filepath.Join(os.TempDir(), LocalTempVersionFileName)
+	_, err := os.Stat(fileName)
+	if os.IsNotExist(err) {
+		return updateLocalVersion(fileName)
+	}
+	if isVersionFileCreatedToday(fileName) {
+		return readVersionFromLocalFile(fileName)
+	}
+	return updateLocalVersion(fileName)
+}
+
+// readVersionFromLocalFile reads the version value from the local temp file
+func readVersionFromLocalFile(fileName string) (string, error) {
+	file, _ := os.OpenFile(fileName, os.O_RDONLY, 0644)
+	defer file.Close()
+	content, err := os.ReadFile(fileName)
+	if err != nil {
+		return "", err
+	}
+	return string(content), err
+}
+
+// Check if the local temp version file is created within the last 24 hours
+func isVersionFileCreatedToday(filename string) bool {
+	fileInfo, err := os.Stat(filename)
+	if err != nil {
+		return false
+	}
+	return time.Since(fileInfo.ModTime()) <= 24*time.Hour
+}
+
+// write version value in local temp file
+func writeVersionToLocalTempFile(fileName string, version string) (int, error) {
+	file, err := os.OpenFile(fileName, os.O_WRONLY|os.O_TRUNC|os.O_CREATE, 0644)
+	if err != nil {
+		// Handle the error if the file cannot be opened.
+		panic(err)
+	}
+	defer file.Close()
+	return file.WriteString(version)
+}
+
+// read version from HitHub and update local version file
+func updateLocalVersion(fileName string) (string, error) {
+	latestVersion, err := readVersionFromGitHub()
+	if err != nil {
+		return "", err
+	}
+	_, err = writeVersionToLocalTempFile(fileName, latestVersion)
+	return latestVersion, err
+}
+
+// read the latest release version number from GitHub, due to GitHub api rate limit we don't do it for every command
+func readVersionFromGitHub() (string, error) {
+	retryClient := retryablehttp.NewClient()
+	retryClient.RetryMax = 2
+	retryClient.RetryWaitMax = 1 * time.Second
+	retryClient.Logger = nil
+	retryClient.CheckRetry = retryablehttp.ErrorPropagatedRetryPolicy
+
+	retryClient.HTTPClient.Timeout = 1 * time.Second
+	req, _ := retryablehttp.NewRequest(http.MethodGet, KubectlFbReleaseURL, nil)
+	resp, err := retryClient.Do(req)
+	if err != nil || resp == nil {
+		fmt.Println("Failed to fetch kubectl-fdb version from GitHub")
+		return "", err
+	}
+	if resp == nil {
+		return "", nil
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+
+	if err != nil {
+		fmt.Println("Error in reading version from GitHub")
+		return "", err
+	}
+	var result = PluginVersionDetails{}
+	if !strings.Contains(strings.ToUpper(resp.Status), "OK") {
+		fmt.Println("Failed to get parse version info from GitHub response")
+		return "", err
+	}
+	err = json.Unmarshal(body, &result)
+	if err != nil {
+		fmt.Println("Failed to read version from GitHub response\n", err)
+		return "", err
+	}
+	//removing 'v' from beginning of release version
+	return result.Version[1:], nil
+}
diff --git a/kubectl-fdb/cmd/remove_process_group.go b/kubectl-fdb/cmd/remove_process_group.go
index fcade941..2e7919cc 100644
--- a/kubectl-fdb/cmd/remove_process_group.go
+++ b/kubectl-fdb/cmd/remove_process_group.go
@@ -60,7 +60,7 @@ func newRemoveProcessGroupCmd(streams genericclioptions.IOStreams) *cobra.Comman
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/restart.go b/kubectl-fdb/cmd/restart.go
index 82e0fa8a..457fd50c 100644
--- a/kubectl-fdb/cmd/restart.go
+++ b/kubectl-fdb/cmd/restart.go
@@ -74,7 +74,7 @@ func newRestartCmd(streams genericclioptions.IOStreams) *cobra.Command {
 				return err
 			}
 
-			kubeClient, err := getKubeClient(o)
+			kubeClient, err := getKubeClient(cmd.Context(), o)
 			if err != nil {
 				return err
 			}
diff --git a/kubectl-fdb/cmd/root.go b/kubectl-fdb/cmd/root.go
index 2aebebb5..19e0ac6e 100644
--- a/kubectl-fdb/cmd/root.go
+++ b/kubectl-fdb/cmd/root.go
@@ -54,7 +54,7 @@ func newFDBOptions(streams genericclioptions.IOStreams) *fdbBOptions {
 }
 
 // NewRootCmd provides a cobra command wrapping FDB actions
-func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
+func NewRootCmd(streams genericclioptions.IOStreams, pluginVersionChecker VersionChecker) *cobra.Command {
 	rand.Seed(time.Now().Unix())
 
 	o := newFDBOptions(streams)
@@ -67,6 +67,13 @@ func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			return cmd.Help()
 		},
+		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
+			versionCheck, _ := cmd.Flags().GetBool("version-check")
+			if versionCheck {
+				return usingLatestPluginVersion(cmd, pluginVersionChecker)
+			}
+			return nil
+		},
 	}
 	cmd.SetOut(o.Out)
 	cmd.SetErr(o.ErrOut)
@@ -74,6 +81,7 @@ func NewRootCmd(streams genericclioptions.IOStreams) *cobra.Command {
 
 	viper.SetDefault("license", "apache 2")
 	cmd.PersistentFlags().StringP("operator-name", "o", "fdb-kubernetes-operator-controller-manager", "Name of the Deployment for the operator.")
+	cmd.PersistentFlags().Bool("version-check", true, "If the plugin should check if a newer release of the plugin exists. If so the command will not be executed.")
 	cmd.PersistentFlags().BoolP("wait", "w", true, "If the plugin should wait for confirmation before executing any action")
 	cmd.PersistentFlags().Uint16P("sleep", "z", 0, "The plugin should sleep between sequential operations for the defined time in seconds (default 0)")
 	o.configFlags.AddFlags(cmd.Flags())
@@ -145,3 +153,25 @@ func printStatement(cmd *cobra.Command, line string, mesType messageType) {
 	cmd.Printf("✔ %s\n", line)
 	color.Unset()
 }
+
+// will check plugin version and won't let any interaction happen with cluster, if it's not latest release version
+func usingLatestPluginVersion(cmd *cobra.Command, pluginVersionChecker VersionChecker) error {
+	// If the user has a self build plugin we are not performing any checks.
+	if strings.ToLower(pluginVersion) == "latest" {
+		return nil
+	}
+	latestPluginVersion, err := pluginVersionChecker.getLatestPluginVersion()
+	if err != nil {
+		return err
+	}
+
+	if pluginVersion != latestPluginVersion {
+		versionMessage := "kubectl-fdb plugin is not up-to-date, please install the latest version and try again!\n" +
+			"Your version:[" + pluginVersion + "], latest release version:[" + latestPluginVersion + "].\n" +
+			"Installation instructions can be found here: https://github.com/FoundationDB/fdb-kubernetes-operator/blob/main/kubectl-fdb/Readme.md"
+		cmd.Println(versionMessage)
+		return fmt.Errorf("outdated plugin version")
+	}
+
+	return nil
+}
diff --git a/kubectl-fdb/cmd/version.go b/kubectl-fdb/cmd/version.go
index 2f83c916..fd6df61f 100644
--- a/kubectl-fdb/cmd/version.go
+++ b/kubectl-fdb/cmd/version.go
@@ -55,7 +55,7 @@ func newVersionCmd(streams genericclioptions.IOStreams) *cobra.Command {
 			}
 
 			if !clientOnly {
-				kubeClient, err := getKubeClient(o)
+				kubeClient, err := getKubeClient(cmd.Context(), o)
 				if err != nil {
 					return err
 				}
diff --git a/kubectl-fdb/main.go b/kubectl-fdb/main.go
index e4f6f338..fffa904c 100644
--- a/kubectl-fdb/main.go
+++ b/kubectl-fdb/main.go
@@ -31,8 +31,7 @@ import (
 func main() {
 	flags := pflag.NewFlagSet("kubectl-fdb", pflag.ExitOnError)
 	pflag.CommandLine = flags
-
-	root := cmd.NewRootCmd(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr})
+	root := cmd.NewRootCmd(genericclioptions.IOStreams{In: os.Stdin, Out: os.Stdout, ErrOut: os.Stderr}, &cmd.RealVersionChecker{})
 	if err := root.Execute(); err != nil {
 		os.Exit(1)
 	}
diff --git a/pkg/fdbadminclient/mock/admin_client_mock.go b/pkg/fdbadminclient/mock/admin_client_mock.go
index b2bdcfa6..b1c2d452 100644
--- a/pkg/fdbadminclient/mock/admin_client_mock.go
+++ b/pkg/fdbadminclient/mock/admin_client_mock.go
@@ -389,6 +389,17 @@ func (client *AdminClient) GetStatus() (*fdbv1beta2.FoundationDBStatus, error) {
 		status.Cluster.DatabaseConfiguration.VersionFlags.LogSpill = 2
 	}
 
+	if len(client.ExcludedAddresses) > 0 {
+		status.Cluster.DatabaseConfiguration.ExcludedServers = make([]fdbv1beta2.ExcludedServers, 0, len(client.ExcludedAddresses))
+	}
+	for excludedAddresses := range client.ExcludedAddresses {
+		if net.ParseIP(excludedAddresses) != nil {
+			status.Cluster.DatabaseConfiguration.ExcludedServers = append(status.Cluster.DatabaseConfiguration.ExcludedServers, fdbv1beta2.ExcludedServers{Address: excludedAddresses})
+		} else {
+			status.Cluster.DatabaseConfiguration.ExcludedServers = append(status.Cluster.DatabaseConfiguration.ExcludedServers, fdbv1beta2.ExcludedServers{Locality: excludedAddresses})
+		}
+	}
+
 	status.Cluster.FullReplication = true
 	status.Cluster.Data.State.Healthy = true
 	status.Cluster.Data.State.Name = "healthy"
@@ -595,15 +606,19 @@ func (client *AdminClient) GetExclusions() ([]fdbv1beta2.ProcessAddress, error)
 		return nil, client.mockError
 	}
 
-	pAddrs := make([]fdbv1beta2.ProcessAddress, len(client.ExcludedAddresses))
+	pAddrs := make([]fdbv1beta2.ProcessAddress, 0, len(client.ExcludedAddresses))
 	for addr := range client.ExcludedAddresses {
-		pAddrs = append(pAddrs, fdbv1beta2.ProcessAddress{
-			IPAddress: net.ParseIP(addr),
-			Port:      0,
-			Flags:     nil,
-		})
+		ip := net.ParseIP(addr)
+		if ip == nil {
+			pAddrs = append(pAddrs, fdbv1beta2.ProcessAddress{StringAddress: addr})
+		} else {
+			pAddrs = append(pAddrs, fdbv1beta2.ProcessAddress{
+				IPAddress: net.ParseIP(addr),
+				Port:      0,
+				Flags:     nil,
+			})
+		}
 	}
-
 	return pAddrs, nil
 }
 


Test patch:
diff --git a/api/v1beta2/foundationdbcluster_types_test.go b/api/v1beta2/foundationdbcluster_types_test.go
index aa0cac2f..9f3820ba 100644
--- a/api/v1beta2/foundationdbcluster_types_test.go
+++ b/api/v1beta2/foundationdbcluster_types_test.go
@@ -5552,6 +5552,26 @@ var _ = Describe("[api] FoundationDBCluster", func() {
 			false,
 			fmt.Errorf("process has missing address in exclusion results: 192.168.0.2"),
 		),
+		Entry("process group is excluded using locality and it is present in the remaining map as excluded",
+			&ProcessGroupStatus{
+				ProcessGroupID: "storage-1",
+			},
+			map[string]bool{
+				fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix): false,
+			},
+			true,
+			nil,
+		),
+		Entry("process group is excluded using locality and it is present in the remaining map as not excluded",
+			&ProcessGroupStatus{
+				ProcessGroupID: "storage-1",
+			},
+			map[string]bool{
+				fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix): true,
+			},
+			false,
+			fmt.Errorf("process has missing exclusion string in exclusion results: %s", fmt.Sprintf("%s:storage-1", FDBLocalityExclusionPrefix)),
+		),
 	)
 
 	DescribeTable("when getting the removal mode", func(cluster *FoundationDBCluster, expected PodUpdateMode) {
diff --git a/controllers/exclude_processes_test.go b/controllers/exclude_processes_test.go
index a9da13f2..96627fb2 100644
--- a/controllers/exclude_processes_test.go
+++ b/controllers/exclude_processes_test.go
@@ -38,6 +38,7 @@ var _ = Describe("exclude_processes", func() {
 	var allowedExclusions int
 	var ongoingExclusions int
 	var missingProcesses []fdbv1beta2.ProcessGroupID
+	var processClass fdbv1beta2.ProcessClass
 
 	When("validating if processes can be excluded", func() {
 		BeforeEach(func() {
@@ -62,220 +63,211 @@ var _ = Describe("exclude_processes", func() {
 		JustBeforeEach(func() {
 			processCounts, err := cluster.GetProcessCountsWithDefaults()
 			Expect(err).NotTo(HaveOccurred())
-			allowedExclusions, missingProcesses = canExcludeNewProcesses(globalControllerLogger, cluster, fdbv1beta2.ProcessClassStorage, processCounts.Storage, ongoingExclusions, false)
+			allowedExclusions, missingProcesses = getAllowedExclusionsAndMissingProcesses(globalControllerLogger, cluster, processClass, processCounts.Map()[processClass], ongoingExclusions, false)
 		})
 
 		When("using a small cluster", func() {
-			When("all processes are healthy", func() {
-				When("no additional processes are running", func() {
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(BeEmpty())
-					})
+			When("using the storage process class", func() {
+				BeforeEach(func() {
+					processClass = fdbv1beta2.ProcessClassStorage
 				})
 
-				When("no additional processes are running", func() {
-					When("the redundancy mode is single", func() {
+				When("all processes are healthy", func() {
+					When("no additional processes are running", func() {
 						It("should not allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(BeEmpty())
 						})
 					})
 
-					When("the redundancy mode is double", func() {
-						BeforeEach(func() {
-							cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeDouble
+					When("no additional processes are running", func() {
+						When("the redundancy mode is single", func() {
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
-							Expect(missingProcesses).To(BeEmpty())
-						})
-					})
+						When("the redundancy mode is double", func() {
+							BeforeEach(func() {
+								cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeDouble
+							})
 
-					When("the redundancy mode is triple", func() {
-						BeforeEach(func() {
-							cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeTriple
-						})
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
-							Expect(missingProcesses).To(BeEmpty())
-						})
-					})
-				})
+							When("there are failed processes", func() {
+								BeforeEach(func() {
 
-				When("one additional process is running", func() {
-					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-					})
+									for idx, processGroup := range cluster.Status.ProcessGroups {
+										if processGroup.ProcessClass != processClass {
+											continue
+										}
 
-					It("should allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 1))
-						Expect(missingProcesses).To(BeEmpty())
-					})
+										cluster.Status.ProcessGroups[idx].UpdateCondition(fdbv1beta2.MissingProcesses, true)
+										cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+										break
+									}
+								})
 
-					When("there is one ongoing exclusion", func() {
-						BeforeEach(func() {
-							ongoingExclusions = 1
+								It("should not allow the exclusion", func() {
+									Expect(allowedExclusions).To(BeZero())
+									Expect(missingProcesses).To(ConsistOf(fdbv1beta2.ProcessGroupID("storage-1")))
+								})
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 0))
-							Expect(missingProcesses).To(BeEmpty())
+						When("the redundancy mode is triple", func() {
+							BeforeEach(func() {
+								cluster.Spec.DatabaseConfiguration.RedundancyMode = fdbv1beta2.RedundancyModeTriple
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()))
+								Expect(missingProcesses).To(BeEmpty())
+							})
+
+							When("there are failed processes", func() {
+								BeforeEach(func() {
+
+									for idx, processGroup := range cluster.Status.ProcessGroups {
+										if processGroup.ProcessClass != processClass {
+											continue
+										}
+
+										cluster.Status.ProcessGroups[idx].UpdateCondition(fdbv1beta2.MissingProcesses, true)
+										cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+										break
+									}
+								})
+
+								It("should allow the exclusion", func() {
+									Expect(allowedExclusions).To(BeNumerically("==", cluster.DesiredFaultTolerance()-1))
+									Expect(missingProcesses).To(ConsistOf(fdbv1beta2.ProcessGroupID("storage-1")))
+								})
+							})
 						})
 					})
 
-					When("the additional process is marked for removal", func() {
+					When("one additional process is running", func() {
 						BeforeEach(func() {
-							// In this case the process group is marked for removal but is counted against the valid
-							// processes as the process is not yet excluded.
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
 						It("should allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 1))
 							Expect(missingProcesses).To(BeEmpty())
 						})
-					})
 
-					When("the additional process is marked for removal and is excluded", func() {
-						BeforeEach(func() {
-							// In this case we have no additional processes running, as the valid processes is the
-							// same as the desired count of processes.
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
-							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].SetExclude()
+						When("there is one ongoing exclusion", func() {
+							BeforeEach(func() {
+								ongoingExclusions = 1
+							})
+
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
 
-						It("should not allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 0))
-							Expect(missingProcesses).To(BeEmpty())
+						When("the additional process is marked for removal", func() {
+							BeforeEach(func() {
+								// In this case the process group is marked for removal but is counted against the valid
+								// processes as the process is not yet excluded.
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 1))
+								Expect(missingProcesses).To(BeEmpty())
+							})
 						})
-					})
-				})
-			})
 
-			When("one process group is missing", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, 1, fdbv1beta2.ProcessClassStorage)
-				})
+						When("the additional process is marked for removal and is excluded", func() {
+							BeforeEach(func() {
+								// In this case we have no additional processes running, as the valid processes is the
+								// same as the desired count of processes.
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].MarkForRemoval()
+								cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].SetExclude()
+							})
 
-				When("no additional processes are running", func() {
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(1))
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(BeEmpty())
+							})
+						})
 					})
 				})
 
-				When("additional processes are running", func() {
+				When("one process group is missing", func() {
 					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						// Add two process groups
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-					})
-
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(1))
+						createMissingProcesses(cluster, 1, processClass)
 					})
 
-					When("there is one ongoing exclusion", func() {
-						BeforeEach(func() {
-							ongoingExclusions = 1
-						})
-
+					When("no additional processes are running", func() {
 						It("should not allow the exclusion", func() {
 							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(1))
 						})
 					})
 
-					When("the missing timestamp is older than 5 minutes", func() {
+					When("additional processes are running", func() {
 						BeforeEach(func() {
-							for idx, processGroup := range cluster.Status.ProcessGroups {
-								timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-								if timestamp == nil {
-									continue
-								}
-
-								cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-							}
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							// Add two process groups
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
-						It("should allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 1))
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(1))
 						})
-					})
-				})
-			})
-
-			When("more process groups are missing than the operator is allowed to automatically replace", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, cluster.GetMaxConcurrentAutomaticReplacements()+1, fdbv1beta2.ProcessClassStorage)
-				})
 
-				It("should not allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-2"}))
-				})
-
-				When("the missing timestamp is older than 5 minutes", func() {
-					BeforeEach(func() {
-						for idx, processGroup := range cluster.Status.ProcessGroups {
-							timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-							if timestamp == nil {
-								continue
-							}
-
-							cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-						}
-					})
-
-					When("automatic replacements enabled", func() {
-						When("no exclusions are ongoing", func() {
-							It("should allow the exclusion to replace as many processes as automatic replacements are allowed", func() {
-								Expect(allowedExclusions).To(BeNumerically("==", cluster.GetMaxConcurrentAutomaticReplacements()))
-								Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
-							})
-						})
-
-						When("one exclusion is ongoing", func() {
+						When("there is one ongoing exclusion", func() {
 							BeforeEach(func() {
 								ongoingExclusions = 1
 							})
 
-							It("should allow the exclusion to replace as many processes as automatic replacements are allowed", func() {
-								Expect(allowedExclusions).To(BeNumerically("==", cluster.GetMaxConcurrentAutomaticReplacements()-ongoingExclusions))
-								Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
+							It("should not allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 0))
+								Expect(missingProcesses).To(HaveLen(1))
 							})
 						})
-					})
 
-					When("automatic replacements disabled", func() {
-						BeforeEach(func() {
-							cluster.Spec.AutomationOptions.Replacements.Enabled = pointer.Bool(false)
-						})
+						When("the missing timestamp is older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
 
-						It("should not allow the exclusions", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", -1*len(missingProcesses)))
-							Expect(missingProcesses).To(HaveLen(cluster.GetMaxConcurrentAutomaticReplacements() + 1))
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 1))
+								Expect(missingProcesses).To(HaveLen(1))
+							})
 						})
 					})
 				})
-			})
 
-			When("two process groups of a different type are missing", func() {
-				BeforeEach(func() {
-					createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassLog)
-				})
+				When("two process groups of a different type are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassLog)
+					})
 
-				It("should allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(BeEmpty())
+					It("should allow the exclusion", func() {
+						Expect(allowedExclusions).To(BeNumerically("==", 0))
+						Expect(missingProcesses).To(BeEmpty())
+					})
 				})
 			})
 		})
@@ -293,63 +285,124 @@ var _ = Describe("exclude_processes", func() {
 				Expect(err).NotTo(HaveOccurred())
 			})
 
-			When("two process groups are missing", func() {
+			When("the storage class is used", func() {
 				BeforeEach(func() {
-					createMissingProcesses(cluster, 2, fdbv1beta2.ProcessClassStorage)
+					processClass = fdbv1beta2.ProcessClassStorage
 				})
 
-				When("no additional processes are running", func() {
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(2))
-					})
-				})
-
-				When("additional processes are running", func() {
+				When("two process groups are missing", func() {
 					BeforeEach(func() {
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
-						cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", fdbv1beta2.ProcessClassStorage, nil))
-						cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+						createMissingProcesses(cluster, 2, processClass)
 					})
 
-					It("should not allow the exclusion", func() {
-						Expect(allowedExclusions).To(BeNumerically("==", 0))
-						Expect(missingProcesses).To(HaveLen(2))
+					When("no additional processes are running", func() {
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
 					})
 
-					When("the missing timestamps are older than 5 minutes", func() {
+					When("additional processes are running", func() {
 						BeforeEach(func() {
-							for idx, processGroup := range cluster.Status.ProcessGroups {
-								timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
-								if timestamp == nil {
-									continue
-								}
-
-								cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
-							}
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
 						})
 
-						It("should allow the exclusion", func() {
-							Expect(allowedExclusions).To(BeNumerically("==", 2))
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
 							Expect(missingProcesses).To(HaveLen(2))
 						})
+
+						When("the missing timestamps are older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
+
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 2))
+								Expect(missingProcesses).To(HaveLen(2))
+							})
+						})
+					})
+				})
+
+				When("five process groups are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 5, processClass)
+					})
+
+					It("should not allow the exclusion", func() {
+						Expect(allowedExclusions).To(BeNumerically("==", 0))
+						Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-10", "storage-11", "storage-12", "storage-13"}))
 					})
 				})
 			})
 
-			When("five process groups are missing", func() {
+			When("the log class is used", func() {
 				BeforeEach(func() {
-					createMissingProcesses(cluster, 5, fdbv1beta2.ProcessClassStorage)
+					processClass = fdbv1beta2.ProcessClassLog
 				})
 
-				It("should not allow the exclusion", func() {
-					Expect(allowedExclusions).To(BeNumerically("==", 0))
-					Expect(missingProcesses).To(Equal([]fdbv1beta2.ProcessGroupID{"storage-1", "storage-10", "storage-11", "storage-12", "storage-13"}))
+				When("two process groups are missing", func() {
+					BeforeEach(func() {
+						createMissingProcesses(cluster, 2, processClass)
+					})
+
+					When("no additional processes are running", func() {
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
+					})
+
+					When("additional processes are running", func() {
+						BeforeEach(func() {
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1337", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1338", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1339", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+							cluster.Status.ProcessGroups = append(cluster.Status.ProcessGroups, fdbv1beta2.NewProcessGroupStatus("storage-1340", processClass, nil))
+							cluster.Status.ProcessGroups[len(cluster.Status.ProcessGroups)-1].ProcessGroupConditions = nil
+						})
+
+						It("should not allow the exclusion", func() {
+							Expect(allowedExclusions).To(BeNumerically("==", 0))
+							Expect(missingProcesses).To(HaveLen(2))
+						})
+
+						When("the missing timestamps are older than 5 minutes", func() {
+							BeforeEach(func() {
+								for idx, processGroup := range cluster.Status.ProcessGroups {
+									timestamp := processGroup.GetConditionTime(fdbv1beta2.MissingProcesses)
+									if timestamp == nil {
+										continue
+									}
+
+									cluster.Status.ProcessGroups[idx].ProcessGroupConditions[0].Timestamp = time.Now().Add(-10 * time.Minute).Unix()
+								}
+							})
+
+							It("should allow the exclusion", func() {
+								Expect(allowedExclusions).To(BeNumerically("==", 2))
+								Expect(missingProcesses).To(HaveLen(2))
+							})
+						})
+					})
 				})
 			})
 		})
@@ -652,6 +705,53 @@ var _ = Describe("exclude_processes", func() {
 			})
 		})
 	})
+
+	DescribeTable("when getting the allowed exclusions", func(validProcesses int, desiredProcessCount int, ongoingExclusions int, faultTolerance int, expected int) {
+		Expect(getAllowedExclusions(GinkgoLogr, validProcesses, desiredProcessCount, ongoingExclusions, faultTolerance)).To(BeNumerically("==", expected))
+	},
+		Entry("when as many valid processes are running as desired",
+			10,
+			10,
+			0,
+			2,
+			2),
+		Entry("when as many valid processes are running as desired but one exclusion is ongoing",
+			10,
+			10,
+			1,
+			2,
+			1),
+		Entry("when as many valid processes are running as desired but two exclusions are ongoing",
+			10,
+			10,
+			2,
+			2,
+			0),
+		Entry("when more valid processes are running than desired",
+			20,
+			10,
+			0,
+			2,
+			12),
+		Entry("when more valid processes are running than desired but there are 10 exclusions ongoing",
+			20,
+			10,
+			10,
+			2,
+			2),
+		Entry("when only 8 valid processes are running",
+			8,
+			10,
+			0,
+			2,
+			0),
+		Entry("when no valid processes are running",
+			0,
+			10,
+			0,
+			2,
+			0),
+	)
 })
 
 func createMissingProcesses(cluster *fdbv1beta2.FoundationDBCluster, count int, processClass fdbv1beta2.ProcessClass) {
diff --git a/controllers/remove_process_groups_test.go b/controllers/remove_process_groups_test.go
index f1460a4d..2f3896b6 100644
--- a/controllers/remove_process_groups_test.go
+++ b/controllers/remove_process_groups_test.go
@@ -23,6 +23,7 @@ package controllers
 import (
 	"context"
 	"fmt"
+	"github.com/go-logr/logr"
 
 	"github.com/FoundationDB/fdb-kubernetes-operator/pkg/fdbadminclient/mock"
 
@@ -462,6 +463,9 @@ var _ = Describe("remove_process_groups", func() {
 
 	Context("validating getProcessesToInclude", func() {
 		var removedProcessGroups map[fdbv1beta2.ProcessGroupID]bool
+		var status *fdbv1beta2.FoundationDBStatus
+		var err error
+		var adminClient *mock.AdminClient
 
 		BeforeEach(func() {
 			cluster = &fdbv1beta2.FoundationDBCluster{
@@ -492,6 +496,13 @@ var _ = Describe("remove_process_groups", func() {
 				},
 			}
 			removedProcessGroups = make(map[fdbv1beta2.ProcessGroupID]bool)
+			adminClient, err = mock.NewMockAdminClientUncast(cluster, k8sClient)
+			Expect(err).NotTo(HaveOccurred())
+		})
+
+		JustBeforeEach(func() {
+			status, err = adminClient.GetStatus()
+			Expect(err).NotTo(HaveOccurred())
 		})
 
 		Context("cluster doesn't support inclusions using locality", func() {
@@ -501,7 +512,10 @@ var _ = Describe("remove_process_groups", func() {
 
 			When("including no process", func() {
 				It("should not include any process", func() {
-					Expect(len(getProcessesToInclude(cluster, removedProcessGroups))).To(Equal(0))
+					processesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(len(processesToInclude)).To(Equal(0))
+					Expect(len(cluster.Status.ProcessGroups)).To(Equal(16))
 				})
 			})
 
@@ -510,13 +524,15 @@ var _ = Describe("remove_process_groups", func() {
 					processGroup := cluster.Status.ProcessGroups[0]
 					Expect(processGroup.ProcessGroupID).To(Equal(fdbv1beta2.ProcessGroupID("storage-1")))
 					processGroup.MarkForRemoval()
-					cluster.Status.ProcessGroups[0] = processGroup
-
+					for _, address := range processGroup.Addresses {
+						adminClient.ExcludedAddresses[address] = fdbv1beta2.None{}
+					}
 					removedProcessGroups[processGroup.ProcessGroupID] = true
 				})
 
 				It("should include one process", func() {
-					fdbProcessesToInclude := getProcessesToInclude(cluster, removedProcessGroups)
+					fdbProcessesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
 					Expect(len(fdbProcessesToInclude)).To(Equal(1))
 					Expect(fdbv1beta2.ProcessAddressesString(fdbProcessesToInclude, " ")).To(Equal("1.1.1.1"))
 					Expect(len(cluster.Status.ProcessGroups)).To(Equal(15))
@@ -531,7 +547,8 @@ var _ = Describe("remove_process_groups", func() {
 
 			When("including no process", func() {
 				It("should not include any process", func() {
-					fdbProcessesToInclude := getProcessesToInclude(cluster, removedProcessGroups)
+					fdbProcessesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
 					Expect(len(fdbProcessesToInclude)).To(Equal(0))
 					Expect(len(cluster.Status.ProcessGroups)).To(Equal(16))
 				})
@@ -544,18 +561,66 @@ var _ = Describe("remove_process_groups", func() {
 					removedProcessGroup = cluster.Status.ProcessGroups[0]
 					Expect(removedProcessGroup.ProcessGroupID).To(Equal(fdbv1beta2.ProcessGroupID("storage-1")))
 					removedProcessGroup.MarkForRemoval()
-					cluster.Status.ProcessGroups[0] = removedProcessGroup
+					adminClient.ExcludedAddresses[removedProcessGroup.GetExclusionString()] = fdbv1beta2.None{}
+					removedProcessGroups[removedProcessGroup.ProcessGroupID] = true
+				})
 
+				It("should include one process", func() {
+					fdbProcessesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(len(fdbProcessesToInclude)).To(Equal(1))
+					Expect(fdbv1beta2.ProcessAddressesString(fdbProcessesToInclude, " ")).To(Equal(removedProcessGroup.GetExclusionString()))
+					Expect(len(cluster.Status.ProcessGroups)).To(Equal(15))
+				})
+			})
+
+			When("including a process which is excluded both by IP and locality", func() {
+				var removedProcessGroup *fdbv1beta2.ProcessGroupStatus
+
+				BeforeEach(func() {
+					removedProcessGroup = cluster.Status.ProcessGroups[0]
+					Expect(removedProcessGroup.ProcessGroupID).To(Equal(fdbv1beta2.ProcessGroupID("storage-1")))
+					removedProcessGroup.MarkForRemoval()
+
+					adminClient.ExcludedAddresses[removedProcessGroup.GetExclusionString()] = fdbv1beta2.None{}
+					adminClient.ExcludedAddresses[removedProcessGroup.Addresses[0]] = fdbv1beta2.None{}
 					removedProcessGroups[removedProcessGroup.ProcessGroupID] = true
 				})
 
 				It("should include one process", func() {
-					fdbProcessesToInclude := getProcessesToInclude(cluster, removedProcessGroups)
+					fdbProcessesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
 					Expect(len(fdbProcessesToInclude)).To(Equal(2))
 					Expect(fdbv1beta2.ProcessAddressesString(fdbProcessesToInclude, " ")).To(Equal(fmt.Sprintf("%s %s", removedProcessGroup.GetExclusionString(), removedProcessGroup.Addresses[0])))
 					Expect(len(cluster.Status.ProcessGroups)).To(Equal(15))
 				})
 			})
+
+			When("one excluded process is missing from excluded servers", func() {
+				var removedProcessGroup *fdbv1beta2.ProcessGroupStatus
+				var removedProcessGroup2 *fdbv1beta2.ProcessGroupStatus
+
+				BeforeEach(func() {
+					removedProcessGroup = cluster.Status.ProcessGroups[0]
+					Expect(removedProcessGroup.ProcessGroupID).To(Equal(fdbv1beta2.ProcessGroupID("storage-1")))
+					removedProcessGroup.MarkForRemoval()
+					removedProcessGroups[removedProcessGroup.ProcessGroupID] = true
+
+					removedProcessGroup2 = cluster.Status.ProcessGroups[1]
+					Expect(removedProcessGroup2.ProcessGroupID).To(Equal(fdbv1beta2.ProcessGroupID("storage-2")))
+					removedProcessGroup2.MarkForRemoval()
+					adminClient.ExcludedAddresses[removedProcessGroup2.GetExclusionString()] = fdbv1beta2.None{}
+					removedProcessGroups[removedProcessGroup2.ProcessGroupID] = true
+				})
+
+				It("should include one process", func() {
+					fdbProcessesToInclude, err := getProcessesToInclude(logr.Logger{}, cluster, removedProcessGroups, status)
+					Expect(err).NotTo(HaveOccurred())
+					Expect(len(fdbProcessesToInclude)).To(Equal(1))
+					Expect(fdbv1beta2.ProcessAddressesString(fdbProcessesToInclude, " ")).To(Equal(removedProcessGroup2.GetExclusionString()))
+					Expect(len(cluster.Status.ProcessGroups)).To(Equal(14))
+				})
+			})
 		})
 	})
 })
diff --git a/e2e/fixtures/factory.go b/e2e/fixtures/factory.go
index 36f91050..6a08e415 100644
--- a/e2e/fixtures/factory.go
+++ b/e2e/fixtures/factory.go
@@ -123,7 +123,7 @@ func (factory *Factory) getClient() *kubernetes.Clientset {
 
 // DeletePod deletes the provided Pod
 func (factory *Factory) DeletePod(pod *corev1.Pod) {
-	gomega.Expect(factory.GetControllerRuntimeClient().Delete(ctx.TODO(), pod)).NotTo(gomega.HaveOccurred())
+	factory.Delete(pod)
 }
 
 // GetPod returns the Pod matching the namespace and name
@@ -818,6 +818,16 @@ func (factory *Factory) CreateIfAbsent(object client.Object) error {
 	return nil
 }
 
+// Delete will delete the provided resource if it exists.
+func (factory *Factory) Delete(object client.Object) {
+	err := factory.GetControllerRuntimeClient().Delete(ctx.TODO(), object)
+	if err == nil || k8serrors.IsNotFound(err) {
+		return
+	}
+
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
+}
+
 // GetOperatorImage returns the operator image provided via command line. If a registry was defined the registry will be
 // prepended.
 func (factory *Factory) GetOperatorImage() string {
diff --git a/e2e/fixtures/fdb_cluster.go b/e2e/fixtures/fdb_cluster.go
index 8624b970..a14c06bb 100644
--- a/e2e/fixtures/fdb_cluster.go
+++ b/e2e/fixtures/fdb_cluster.go
@@ -633,7 +633,7 @@ func (fdbCluster *FdbCluster) ReplacePod(pod corev1.Pod, waitForReconcile bool)
 }
 
 // ReplacePods replaces the provided Pods in the current FoundationDBCluster.
-func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod) {
+func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod, waitForReconcile bool) {
 	for _, pod := range pods {
 		fdbCluster.cluster.Spec.ProcessGroupsToRemove = append(
 			fdbCluster.cluster.Spec.ProcessGroupsToRemove,
@@ -642,6 +642,10 @@ func (fdbCluster *FdbCluster) ReplacePods(pods []corev1.Pod) {
 	}
 	fdbCluster.UpdateClusterSpec()
 
+	if !waitForReconcile {
+		return
+	}
+
 	gomega.Expect(fdbCluster.WaitForReconciliation()).NotTo(gomega.HaveOccurred())
 }
 
diff --git a/e2e/fixtures/kubernetes_fixtures.go b/e2e/fixtures/kubernetes_fixtures.go
index 9be51da5..12c9bd73 100644
--- a/e2e/fixtures/kubernetes_fixtures.go
+++ b/e2e/fixtures/kubernetes_fixtures.go
@@ -53,7 +53,7 @@ func (factory *Factory) MultipleNamespaces(dcIDs []string) []string {
 	res := make([]string, len(dcIDs))
 	for idx, dcID := range dcIDs {
 		namespace := factory.createNamespace(dcID)
-		log.Println("Create namespace" + namespace)
+		log.Println("Create namespace", namespace)
 		res[idx] = namespace
 	}
 
diff --git a/e2e/test_operator/operator_test.go b/e2e/test_operator/operator_test.go
index 4b1f420d..95001aec 100644
--- a/e2e/test_operator/operator_test.go
+++ b/e2e/test_operator/operator_test.go
@@ -384,6 +384,9 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 			spec.AutomationOptions.UseLocalitiesForExclusion = pointer.Bool(useLocalitiesForExclusion)
 			fdbCluster.UpdateClusterSpecWithSpec(spec)
 			Expect(fdbCluster.GetCluster().UseLocalitiesForExclusion()).To(Equal(useLocalitiesForExclusion))
+
+			// Making sure we included back all the process groups after exclusion is complete.
+			Expect(fdbCluster.GetStatus().Cluster.DatabaseConfiguration.ExcludedServers).To(BeEmpty())
 		})
 
 		When("IP addresses are used for exclusion", func() {
@@ -913,7 +916,7 @@ var _ = Describe("Operator", Label("e2e", "pr"), func() {
 		var replacedPods []corev1.Pod
 
 		BeforeEach(func() {
-			fdbCluster.ReplacePods(fixtures.RandomPickPod(fdbCluster.GetStatelessPods().Items, 4))
+			fdbCluster.ReplacePods(fixtures.RandomPickPod(fdbCluster.GetStatelessPods().Items, 4), true)
 		})
 
 		AfterEach(func() {
diff --git a/e2e/test_operator/suite_test.go b/e2e/test_operator/suite_test.go
index 82b59f4a..2a5649e2 100644
--- a/e2e/test_operator/suite_test.go
+++ b/e2e/test_operator/suite_test.go
@@ -28,7 +28,7 @@ import (
 	"github.com/onsi/gomega"
 )
 
-func TestOperatorCreationVelocity(t *testing.T) {
+func TestOperator(t *testing.T) {
 	gomega.SetDefaultEventuallyTimeout(10 * time.Second)
 	fixtures.RunGinkgoTests(t, "Operator test suite")
 }
diff --git a/e2e/test_operator_ha/operator_ha_test.go b/e2e/test_operator_ha/operator_ha_test.go
new file mode 100644
index 00000000..a4b56833
--- /dev/null
+++ b/e2e/test_operator_ha/operator_ha_test.go
@@ -0,0 +1,162 @@
+/*
+ * operator_ha_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package operatorha
+
+/*
+This test suite includes functional tests to ensure normal operational tasks are working fine.
+Those tests include replacements of healthy or fault Pods and setting different configurations.
+
+The assumption is that every test case reverts the changes that were done on the cluster.
+In order to improve the test speed we only create one FoundationDB HA cluster initially.
+This cluster will be used for all tests.
+*/
+
+import (
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	chaosmesh "github.com/chaos-mesh/chaos-mesh/api/v1alpha1"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var (
+	factory     *fixtures.Factory
+	fdbCluster  *fixtures.HaFdbCluster
+	testOptions *fixtures.FactoryOptions
+)
+
+func init() {
+	testOptions = fixtures.InitFlags()
+}
+
+var _ = BeforeSuite(func() {
+	factory = fixtures.CreateFactory(testOptions)
+	fdbCluster = factory.CreateFdbHaCluster(
+		fixtures.DefaultClusterConfigWithHaMode(fixtures.HaFourZoneSingleSat, false),
+		factory.GetClusterOptions()...,
+	)
+
+	// Load some data into the cluster.
+	factory.CreateDataLoaderIfAbsent(fdbCluster.GetPrimary())
+
+	// In order to test the robustness of the operator we try to kill the operator Pods every minute.
+	if factory.ChaosTestsEnabled() {
+		for _, cluster := range fdbCluster.GetAllClusters() {
+			factory.ScheduleInjectPodKill(
+				fixtures.GetOperatorSelector(cluster.Namespace()),
+				"*/2 * * * *",
+				chaosmesh.OneMode,
+			)
+		}
+	}
+})
+
+var _ = AfterSuite(func() {
+	if CurrentSpecReport().Failed() {
+		log.Printf("failed due to %s", CurrentSpecReport().FailureMessage())
+	}
+	factory.Shutdown()
+})
+
+var _ = Describe("Operator HA tests", Label("e2e", "pr"), func() {
+	var availabilityCheck bool
+
+	AfterEach(func() {
+		// Reset availabilityCheck if a test case removes this check.
+		availabilityCheck = true
+		if CurrentSpecReport().Failed() {
+			factory.DumpStateHaCluster(fdbCluster)
+		}
+		Expect(fdbCluster.WaitForReconciliation()).ToNot(HaveOccurred())
+		factory.StopInvariantCheck()
+		// Make sure all data is present in the cluster
+		fdbCluster.GetPrimary().EnsureTeamTrackersAreHealthy()
+		fdbCluster.GetPrimary().EnsureTeamTrackersHaveMinReplicas()
+	})
+
+	JustBeforeEach(func() {
+		if availabilityCheck {
+			Expect(fdbCluster.GetPrimary().InvariantClusterStatusAvailable()).NotTo(HaveOccurred())
+		}
+	})
+
+	When("replacing satellite Pods and the new Pods are stuck in pending", func() {
+		var desiredRunningPods int
+		var quota *corev1.ResourceQuota
+
+		BeforeEach(func() {
+			satellite := fdbCluster.GetPrimarySatellite()
+			satelliteCluster := satellite.GetCluster()
+
+			processCounts, err := satelliteCluster.GetProcessCountsWithDefaults()
+			Expect(err).NotTo(HaveOccurred())
+
+			// Create Quota to limit the PVCs that can be created to 0. This will mean no new PVCs can be created.
+			quota = &corev1.ResourceQuota{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      "testing-quota",
+					Namespace: satellite.Namespace(),
+				},
+				Spec: corev1.ResourceQuotaSpec{
+					Hard: corev1.ResourceList{
+						"persistentvolumeclaims": resource.MustParse(strconv.Itoa(0)),
+					},
+				},
+			}
+			Expect(factory.CreateIfAbsent(quota)).NotTo(HaveOccurred())
+
+			desiredRunningPods = processCounts.Log - satelliteCluster.DesiredFaultTolerance()
+
+			// Replace all Pods for this cluster.
+			satellite.ReplacePods(satellite.GetAllPods().Items, false)
+		})
+
+		AfterEach(func() {
+			// Make sure that the quota is deleted and new PVCs can be created.
+			factory.Delete(quota)
+			Expect(fdbCluster.GetPrimarySatellite().WaitForReconciliation()).NotTo(HaveOccurred())
+		})
+
+		It("should not replace too many Pods and bring down the satellite", func() {
+			satellite := fdbCluster.GetPrimarySatellite()
+
+			Consistently(func() int {
+				var runningPods int
+				for _, pod := range satellite.GetAllPods().Items {
+					if pod.Status.Phase != corev1.PodRunning {
+						continue
+					}
+
+					runningPods++
+				}
+
+				// We should add here another check that the cluster stays in the primary.
+				return runningPods
+			}).WithTimeout(10 * time.Minute).WithPolling(2 * time.Second).Should(BeNumerically(">=", desiredRunningPods))
+		})
+	})
+})
diff --git a/e2e/test_operator_ha/suite_test.go b/e2e/test_operator_ha/suite_test.go
new file mode 100644
index 00000000..ece141dc
--- /dev/null
+++ b/e2e/test_operator_ha/suite_test.go
@@ -0,0 +1,34 @@
+/*
+ * suite_test.go
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2023 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package operatorha
+
+import (
+	"testing"
+	"time"
+
+	"github.com/FoundationDB/fdb-kubernetes-operator/e2e/fixtures"
+	"github.com/onsi/gomega"
+)
+
+func TestOperatorHA(t *testing.T) {
+	gomega.SetDefaultEventuallyTimeout(10 * time.Second)
+	fixtures.RunGinkgoTests(t, "Operator HA test suite")
+}
diff --git a/kubectl-fdb/cmd/analyze_test.go b/kubectl-fdb/cmd/analyze_test.go
index 7d509342..b09c2b74 100644
--- a/kubectl-fdb/cmd/analyze_test.go
+++ b/kubectl-fdb/cmd/analyze_test.go
@@ -486,7 +486,7 @@ var _ = Describe("[plugin] analyze cluster", func() {
 			errBuffer := bytes.Buffer{}
 			inBuffer := bytes.Buffer{}
 
-			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 			rootCmd.SetArgs([]string{"analyze"})
 
 			err := rootCmd.Execute()
diff --git a/kubectl-fdb/cmd/deprecation_test.go b/kubectl-fdb/cmd/deprecation_test.go
index 5c673e77..2c1f8d8c 100644
--- a/kubectl-fdb/cmd/deprecation_test.go
+++ b/kubectl-fdb/cmd/deprecation_test.go
@@ -209,7 +209,7 @@ var _ = Describe("[plugin] deprecation command", func() {
 			errBuffer := bytes.Buffer{}
 			inBuffer := bytes.Buffer{}
 
-			cmd = NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd = NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 		})
 
 		It("should print out the help information", func() {
diff --git a/kubectl-fdb/cmd/restart_test.go b/kubectl-fdb/cmd/restart_test.go
index 89f37994..86105f30 100644
--- a/kubectl-fdb/cmd/restart_test.go
+++ b/kubectl-fdb/cmd/restart_test.go
@@ -42,7 +42,7 @@ var _ = Describe("[plugin] root command", func() {
 		})
 
 		It("should not throw an error", func() {
-			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 
 			args := []string{"restart", "-c", "sample"}
 			cmd.SetArgs(args)
diff --git a/kubectl-fdb/cmd/root_test.go b/kubectl-fdb/cmd/root_test.go
index d34b0145..63b58293 100644
--- a/kubectl-fdb/cmd/root_test.go
+++ b/kubectl-fdb/cmd/root_test.go
@@ -42,7 +42,7 @@ var _ = Describe("[plugin] root command", func() {
 		})
 
 		It("should not throw an error", func() {
-			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			cmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 			cmd.SetArgs([]string{})
 			Expect(cmd.Execute()).NotTo(HaveOccurred())
 		})
diff --git a/kubectl-fdb/cmd/suite_test.go b/kubectl-fdb/cmd/suite_test.go
index ccc2cc5c..56ba5f29 100644
--- a/kubectl-fdb/cmd/suite_test.go
+++ b/kubectl-fdb/cmd/suite_test.go
@@ -19,6 +19,19 @@ var cluster *fdbv1beta2.FoundationDBCluster
 var clusterName = "test"
 var namespace = "test"
 
+// MockVersionChecker will help to mock the calls to GitHub API to get 'latest' version in tests
+type MockVersionChecker struct {
+	MockedVersion string
+}
+
+func (versionChecker *MockVersionChecker) getLatestPluginVersion() (string, error) {
+	if versionChecker.MockedVersion == "" {
+		return "latest", nil
+	}
+
+	return versionChecker.MockedVersion, nil
+}
+
 func TestCmd(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "FDB plugin")
diff --git a/kubectl-fdb/cmd/version_test.go b/kubectl-fdb/cmd/version_test.go
index 40a2b79a..97b960d7 100644
--- a/kubectl-fdb/cmd/version_test.go
+++ b/kubectl-fdb/cmd/version_test.go
@@ -46,7 +46,7 @@ var _ = Describe("[plugin] version command", func() {
 			errBuffer = bytes.Buffer{}
 			inBuffer = bytes.Buffer{}
 
-			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer})
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{})
 
 			args := []string{"version", "--client-only"}
 			rootCmd.SetArgs(args)
@@ -162,4 +162,33 @@ var _ = Describe("[plugin] version command", func() {
 				}),
 		)
 	})
+	When("running the version command with old version", func() {
+		var outBuffer bytes.Buffer
+		var errBuffer bytes.Buffer
+		var inBuffer bytes.Buffer
+
+		AfterEach(func() {
+			pluginVersion = "latest"
+		})
+		BeforeEach(func() {
+			pluginVersion = "1.0.0"
+			// We use these buffers to check the input/output
+			outBuffer = bytes.Buffer{}
+			errBuffer = bytes.Buffer{}
+			inBuffer = bytes.Buffer{}
+
+			rootCmd := NewRootCmd(genericclioptions.IOStreams{In: &inBuffer, Out: &outBuffer, ErrOut: &errBuffer}, &MockVersionChecker{MockedVersion: "2.0.0"})
+
+			args := []string{"version", "--client-only"}
+			rootCmd.SetArgs(args)
+
+			err := rootCmd.Execute()
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("should print out the client version", func() {
+			Expect(outBuffer.String()).To(ContainSubstring(
+				"kubectl-fdb plugin is not up-to-date, please install the latest version and try again!"))
+		})
+	})
 })