Limit range over processes to DC (#1994)

* Limit range over processes to DC (needs removal of notes)

---------

Co-authored-by: Nicole Morales <[email protected]>

Author: nicmorales9

api/v1beta2/foundationdbcluster_types.go

@@ -2919,3 +2919,15 @@ func (cluster *FoundationDBCluster) GetProcessGroupID(processClass ProcessClass,
 func (cluster *FoundationDBCluster) IsPodIPFamily6() bool {
 	return cluster.Spec.Routing.PodIPFamily != nil && *cluster.Spec.Routing.PodIPFamily == 6
 }
+
+// ProcessSharesDC returns true if the process's locality matches the cluster's Datacenter.
+// If there is insufficient cluster information, it will return true to avoid filtering when there is insufficient data
+func (cluster *FoundationDBCluster) ProcessSharesDC(process FoundationDBStatusProcessInfo) bool {
+	if cluster == nil || cluster.Spec.DataCenter == "" {
+		return true
+	}
+	if cluster.Spec.DataCenter == process.Locality[FDBLocalityDCIDKey] {
+		return true
+	}
+	return false
+}

controllers/bounce_processes.go

@@ -323,7 +323,7 @@ func getAddressesForUpgrade(logger logr.Logger, r *FoundationDBClusterReconciler
 	}
 
 	// We don't want to check for fault tolerance here to make sure the operator is able to restart processes if some
-	// processes where restarted before the operator issued the cluster wide restart. For version incompatible upgrades
+	// processes were restarted before the operator issued the cluster wide restart. For version incompatible upgrades
 	// that would mean that the processes restarted earlier are not part of the cluster anymore leading to a fault tolerance
 	// drop.
 	if !status.Client.DatabaseStatus.Available {

controllers/choose_removals.go

@@ -70,6 +70,9 @@ func (c chooseRemovals) reconcile(ctx context.Context, r *FoundationDBClusterRec
 
 	localityMap := make(map[string]locality.Info)
 	for _, process := range status.Cluster.Processes {
+		if !cluster.ProcessSharesDC(process) {
+			continue
+		}
 		id := process.Locality[fdbv1beta2.FDBLocalityInstanceIDKey]
 		localityMap[id] = locality.Info{ID: id, Address: process.Address, LocalityData: process.Locality}
 	}
@@ -78,12 +81,12 @@ func (c chooseRemovals) reconcile(ctx context.Context, r *FoundationDBClusterRec
 
 	for _, processClass := range fdbv1beta2.ProcessClasses {
 		desiredCount := desiredCounts[processClass]
-		removedCount := currentCounts[processClass] - desiredCount
+		excessCount := currentCounts[processClass] - desiredCount
 		processClassLocality := make([]locality.Info, 0, currentCounts[processClass])
 
 		for _, processGroup := range cluster.Status.ProcessGroupsByProcessClass(processClass) {
 			if processGroup.IsMarkedForRemoval() {
-				removedCount--
+				excessCount--
 				continue
 			}
 			localityInfo, present := localityMap[string(processGroup.ProcessGroupID)]
@@ -92,8 +95,8 @@ 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))
+		if excessCount > 0 {
+			r.Recorder.Event(cluster, corev1.EventTypeNormal, "ShrinkingProcesses", fmt.Sprintf("Removing %d %s processes", excessCount, processClass))
 
 			remainingProcesses, err := locality.ChooseDistributedProcesses(cluster, processClassLocality, desiredCount, locality.ProcessSelectionConstraint{})
 			if err != nil {

controllers/remove_incompatible_processes.go

@@ -82,7 +82,7 @@ func processIncompatibleProcesses(ctx context.Context, r *FoundationDBClusterRec
 
 	logger.Info("incompatible connections", "incompatibleConnections", status.Cluster.IncompatibleConnections)
 
-	incompatibleConnections := parseIncompatibleConnections(logger, status)
+	incompatibleConnections := parseIncompatibleConnections(logger, status, cluster)
 	incompatiblePods := make([]*corev1.Pod, 0, len(incompatibleConnections))
 	for _, processGroup := range cluster.Status.ProcessGroups {
 		pod, err := r.PodLifecycleManager.GetPod(ctx, r, cluster, processGroup.GetPodName(cluster))
@@ -111,9 +111,12 @@ func processIncompatibleProcesses(ctx context.Context, r *FoundationDBClusterRec
 
 // parseIncompatibleConnections parses the incompatible connections string slice to a map and removes all false reported incompatible processes.
 // If a process is still part of the cluster status we can assume it's not an incompatible process.
-func parseIncompatibleConnections(logger logr.Logger, status *fdbv1beta2.FoundationDBStatus) map[string]fdbv1beta2.None {
+func parseIncompatibleConnections(logger logr.Logger, status *fdbv1beta2.FoundationDBStatus, cluster *fdbv1beta2.FoundationDBCluster) map[string]fdbv1beta2.None {
 	processAddressMap := map[string]fdbv1beta2.None{}
 	for _, process := range status.Cluster.Processes {
+		if !cluster.ProcessSharesDC(process) {
+			continue
+		}
 		processAddressMap[process.Address.MachineAddress()] = fdbv1beta2.None{}
 	}
 

controllers/remove_incompatible_processes_test.go

@@ -70,7 +70,7 @@ var _ = Describe("restart_incompatible_pods", func() {
 	)
 
 	DescribeTable("when parsing incompatible connections", func(status *fdbv1beta2.FoundationDBStatus, expected map[string]fdbv1beta2.None) {
-		Expect(parseIncompatibleConnections(logr.Discard(), status)).To(Equal(expected))
+		Expect(parseIncompatibleConnections(logr.Discard(), status, nil)).To(Equal(expected))
 	},
 		Entry("empty incompatible map",
 			&fdbv1beta2.FoundationDBStatus{

controllers/update_status.go

@@ -72,7 +72,7 @@ func (updateStatus) reconcile(ctx context.Context, r *FoundationDBClusterReconci
 	for _, process := range databaseStatus.Cluster.Processes {
 		versionMap[process.Version]++
 		// Ignore all processes for the process map that are for a different data center
-		if cluster.Spec.DataCenter != "" && cluster.Spec.DataCenter != process.Locality[fdbv1beta2.FDBLocalityDCIDKey] {
+		if !cluster.ProcessSharesDC(process) {
 			continue
 		}