properly synchronize access to unhealthyNodes map (#202)

Author: dgrove-oss

internal/controller/appwrapper/appwrapper_controller.go

@@ -510,6 +510,7 @@ func (r *AppWrapperReconciler) getPodStatus(ctx context.Context, aw *workloadv1b
 		return nil, err
 	}
 	summary := &podStatusSummary{expected: pc}
+	checkUnhealthyNodes := r.Config.Autopilot != nil && r.Config.Autopilot.EvacuateWorkloads
 
 	for _, pod := range pods.Items {
 		switch pod.Status.Phase {
@@ -518,29 +519,33 @@ func (r *AppWrapperReconciler) getPodStatus(ctx context.Context, aw *workloadv1b
 		case v1.PodRunning:
 			if pod.DeletionTimestamp.IsZero() {
 				summary.running += 1
-				if len(unhealthyNodes) > 0 {
-					if resources, ok := unhealthyNodes[pod.Spec.NodeName]; ok {
-						for badResource := range resources {
-							for _, container := range pod.Spec.Containers {
-								if limit, ok := container.Resources.Limits[v1.ResourceName(badResource)]; ok {
-									if !limit.IsZero() {
-										if summary.unhealthyNodes == nil {
-											summary.unhealthyNodes = make(sets.Set[string])
+				if checkUnhealthyNodes {
+					unhealthyNodesMutex.RLock() // BEGIN CRITICAL SECTION
+					if len(unhealthyNodes) > 0 {
+						if resources, ok := unhealthyNodes[pod.Spec.NodeName]; ok {
+							for badResource := range resources {
+								for _, container := range pod.Spec.Containers {
+									if limit, ok := container.Resources.Limits[v1.ResourceName(badResource)]; ok {
+										if !limit.IsZero() {
+											if summary.unhealthyNodes == nil {
+												summary.unhealthyNodes = make(sets.Set[string])
+											}
+											summary.unhealthyNodes.Insert(pod.Spec.NodeName)
 										}
-										summary.unhealthyNodes.Insert(pod.Spec.NodeName)
 									}
-								}
-								if request, ok := container.Resources.Requests[v1.ResourceName(badResource)]; ok {
-									if !request.IsZero() {
-										if summary.unhealthyNodes == nil {
-											summary.unhealthyNodes = make(sets.Set[string])
+									if request, ok := container.Resources.Requests[v1.ResourceName(badResource)]; ok {
+										if !request.IsZero() {
+											if summary.unhealthyNodes == nil {
+												summary.unhealthyNodes = make(sets.Set[string])
+											}
+											summary.unhealthyNodes.Insert(pod.Spec.NodeName)
 										}
-										summary.unhealthyNodes.Insert(pod.Spec.NodeName)
 									}
 								}
 							}
 						}
 					}
+					unhealthyNodesMutex.RUnlock() // END CRITICAL SECTION
 				}
 			}
 		case v1.PodSucceeded:

internal/controller/appwrapper/node_health_monitor.go

@@ -18,6 +18,7 @@ package appwrapper
 
 import (
 	"context"
+	"sync"
 
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -38,8 +39,11 @@ type NodeHealthMonitor struct {
 	Config *config.AppWrapperConfig
 }
 
-// unhealthyNodes is a mapping from Node names to a set of resources that Autopilot has labeled as unhealthy on that Node
-var unhealthyNodes = make(map[string]sets.Set[string])
+var (
+	// unhealthyNodes is a mapping from Node names to a set of resources that Autopilot has labeled as unhealthy on that Node
+	unhealthyNodes      = make(map[string]sets.Set[string])
+	unhealthyNodesMutex sync.RWMutex
+)
 
 // permission to watch nodes
 //+kubebuilder:rbac:groups="",resources=nodes,verbs=get;list;watch
@@ -55,8 +59,6 @@ func (r *NodeHealthMonitor) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		return ctrl.Result{}, nil
 	}
 
-	log.FromContext(ctx).V(2).Info("Reconcilling", "node", req.NamespacedName)
-
 	flaggedResources := make(sets.Set[string])
 	for key, value := range node.GetLabels() {
 		for resource, apLabels := range r.Config.Autopilot.ResourceUnhealthyConfig {
@@ -66,22 +68,27 @@ func (r *NodeHealthMonitor) Reconcile(ctx context.Context, req ctrl.Request) (ct
 		}
 	}
 
-	hadEntries := len(unhealthyNodes) > 0
-
-	if len(flaggedResources) == 0 {
-		delete(unhealthyNodes, node.GetName())
-	} else {
+	nodeChanged := false
+	unhealthyNodesMutex.Lock() // BEGIN CRITICAL SECTION
+	if priorEntry, ok := unhealthyNodes[node.GetName()]; ok {
+		if len(flaggedResources) == 0 {
+			delete(unhealthyNodes, node.GetName())
+			nodeChanged = true
+		} else if !priorEntry.Equal(flaggedResources) {
+			unhealthyNodes[node.GetName()] = flaggedResources
+			nodeChanged = true
+		}
+	} else if len(flaggedResources) > 0 {
 		unhealthyNodes[node.GetName()] = flaggedResources
+		nodeChanged = true
 	}
+	numUnhealthy := len(unhealthyNodes)
+	unhealthyNodesMutex.Unlock() // END CRITICAL SECTION
 
-	if len(unhealthyNodes) == 0 {
-		if hadEntries {
-			log.FromContext(ctx).Info("All nodes now healthy")
-		} else {
-			log.FromContext(ctx).V(2).Info("All nodes now healthy")
-		}
-	} else {
-		log.FromContext(ctx).Info("Some nodes unhealthy", "number", len(unhealthyNodes), "details", unhealthyNodes)
+	if nodeChanged {
+		// This unsynchronized read of unhealthyNodes for logging purposes is safe because this method
+		// is the only writer to the map and the controller runtime is configured to not allow concurrent execution of this method.
+		log.FromContext(ctx).Info("Updated node health information", "Number Unhealthy Nodes", numUnhealthy, "Unhealthy Resource Details", unhealthyNodes)
 	}
 
 	return ctrl.Result{}, nil