feat(mgmt-agent): add node-health detection and labeling controller (AROSLSRE-1588)

.yamllint.yml

@@ -34,6 +34,7 @@ ignore:
 - 'observability/arobit/deploy/templates/forwarder-secretprovider.yaml'
 - 'observability/kube-events/deploy/templates/deployment.yaml'
 - 'mgmt-agent/deploy/templates/deployment.yaml'
+- 'mgmt-agent/deploy/templates/node-health-config.yaml'
 - 'mgmt-fixes/deploy/kubelet-ds/templates/ds-kubelet-parameters.yaml'
 - 'sessiongate/deploy/templates/deployment.yaml'
 - 'sessiongate/deploy/templates/ext-authz.authorizationpolicy.yaml'

mgmt-agent/cmd/cmd.go

@@ -33,7 +33,7 @@ func NewRootCmd() (*cobra.Command, error) {
 to suit the needs of running ARO-HCP. It provides a place to put logic to bridge
 the gap between "brand new AKS cluster" and "ready to run ARO-HCP customer workloads".
 
-mgmt-agent runs two controllers under a single leader election:
+mgmt-agent runs three controllers under a single leader election:
 
 1. SWIFT NIC controller: watches Node objects on the management cluster, queries
    the Azure Compute API for each node's VM network configuration, and sets an
@@ -45,6 +45,13 @@ mgmt-agent runs two controllers under a single leader election:
    metrics (kube_node_status_condition, kube_node_info) from each HCP's API server.
    Metrics are forwarded to the HCP Azure Managed Prometheus workspace.
 
+3. Node-health controller: watches kubelet Pod Events for hard-coded failure
+   signatures (e.g. SWIFT delegated-NIC teardown that leaves a node Ready but
+   unable to start pods) and, when a SWIFT-v2 node crosses a detector threshold,
+   sets a health label on the node (and clears it when the node recovers) so an
+   out-of-band consumer can act on it. It watches only SWIFT-v2 nodes and ships
+   disabled by default.
+
 It also runs log-only watchers for Pod (when KSM is enabled) and selected CRD
 and core resources to aid operational troubleshooting.`,
 	}

mgmt-agent/cmd/options.go

@@ -29,17 +29,22 @@ import (
 
 	"github.com/spf13/cobra"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/discovery"
 	"k8s.io/client-go/dynamic"
 	dynamicinformer "k8s.io/client-go/dynamic/dynamicinformer"
 	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/cache"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
+	"k8s.io/client-go/tools/record"
 	"k8s.io/component-base/metrics/legacyregistry"
 	"k8s.io/klog/v2"
 
@@ -51,6 +56,7 @@ import (
 	sharedleaderelection "github.com/Azure/ARO-HCP/internal/leaderelection"
 	"github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller"
 	"github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/ksmhcp"
+	"github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/nodehealth"
 )
 
 const (
@@ -64,12 +70,17 @@ type RawControllerOptions struct {
 	Workers       int
 	LogVerbosity  int
 	KSMImage      string
+
+	NodeHealthConfigMapName string
+	NodeHealthConfigKey     string
 }
 
 func DefaultControllerOptions() *RawControllerOptions {
 	return &RawControllerOptions{
-		HealthAddress: ":8080",
-		Workers:       2,
+		HealthAddress:           ":8080",
+		Workers:                 2,
+		NodeHealthConfigMapName: "mgmt-agent-node-health",
+		NodeHealthConfigKey:     "config.yaml",
 	}
 }
 
@@ -82,6 +93,10 @@ func (o *RawControllerOptions) BindFlags(cmd *cobra.Command) error {
 		"Log verbosity. 0 is the default verbosity level, equivalent to INFO. "+
 			"It must be a value >= 0, where a higher value means more verbose output.")
 	cmd.Flags().StringVar(&o.KSMImage, "ksm-image", o.KSMImage, "Container image for kube-state-metrics deployed per HCP namespace")
+	cmd.Flags().StringVar(&o.NodeHealthConfigMapName, "node-health-configmap", o.NodeHealthConfigMapName,
+		"Name of the ConfigMap (in --namespace) holding the node-health configuration. The controller is disabled until this ConfigMap enables it.")
+	cmd.Flags().StringVar(&o.NodeHealthConfigKey, "node-health-config-key", o.NodeHealthConfigKey,
+		"Key within the node-health ConfigMap that holds the YAML configuration.")
 
 	return nil
 }
@@ -97,13 +112,16 @@ type ValidatedControllerOptions struct {
 type completedControllerOptions struct {
 	ctrl                     *controller.SwiftNICController
 	ksmCtrl                  *ksmhcp.KSMHCPController
+	nodeHealth               *nodehealth.Controller
 	resourceWatcher          *controller.ResourceWatcher
 	podWatcher               *controller.PodWatcher
 	configMapWatcher         *controller.ConfigMapWatcher
 	kubeInformers            kubeinformers.SharedInformerFactory
 	ksmKubeInformers         kubeinformers.SharedInformerFactory
 	clusterWideKubeInformers kubeinformers.SharedInformerFactory
 	cmWatcherInformers       kubeinformers.SharedInformerFactory
+	nodeHealthInformers      kubeinformers.SharedInformerFactory
+	nodeHealthCMInformers    kubeinformers.SharedInformerFactory
 	hypershiftInformers      hypershiftinformers.SharedInformerFactory
 	dynamicInformers         dynamicinformer.DynamicSharedInformerFactory
 	workers                  int
@@ -188,6 +206,68 @@ func (o *ValidatedControllerOptions) Complete(ctx context.Context) (*ControllerO
 		return nil, fmt.Errorf("failed to create ConfigMap watcher: %w", err)
 	}
 
+	// Node-health controller. It watches Nodes, Pods, and kubelet Pod Events
+	// through shared informers, and labels "Ready but wedged" nodes. It starts
+	// disabled and is enabled/tuned via its ConfigMap.
+	nodehealth.RegisterMetrics()
+
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: kubeClientset.CoreV1().Events("")})
+	nodeHealthRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "mgmt-agent-node-health"})
+	// Shut the broadcaster's background goroutines down when the process context
+	// is cancelled, so we do not leak them on shutdown.
+	go func() {
+		defer utilruntime.HandleCrash()
+		<-ctx.Done()
+		eventBroadcaster.Shutdown()
+	}()
+
+	// Events informer scoped to Pod-involved events to bound cache size.
+	nodeHealthInformers := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClientset, 0,
+		kubeinformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
+			opts.FieldSelector = "involvedObject.kind=Pod"
+		}),
+	)
+
+	nodeHealth, err := nodehealth.NewController(
+		kubeClientset,
+		kubeInformers.Core().V1().Nodes(),
+		clusterWideKubeInformers.Core().V1().Pods(),
+		nodeHealthInformers.Core().V1().Events(),
+		nodeHealthRecorder,
+		nil, // real clock
+		nodehealth.Default(),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create node-health controller: %w", err)
+	}
+
+	// ConfigMap informer scoped to the node-health config for hot-reload.
+	nodeHealthCMInformers := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClientset, 0,
+		kubeinformers.WithNamespace(o.Namespace),
+		kubeinformers.WithTweakListOptions(func(opts *metav1.ListOptions) {
+			opts.FieldSelector = "metadata.name=" + o.NodeHealthConfigMapName
+		}),
+	)
+	nodeHealthConfigKey := o.NodeHealthConfigKey
+	if _, err := nodeHealthCMInformers.Core().V1().ConfigMaps().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			if cm, ok := obj.(*corev1.ConfigMap); ok {
+				nodeHealth.OnConfigMap(cm, nodeHealthConfigKey)
+			}
+		},
+		UpdateFunc: func(_, newObj interface{}) {
+			if cm, ok := newObj.(*corev1.ConfigMap); ok {
+				nodeHealth.OnConfigMap(cm, nodeHealthConfigKey)
+			}
+		},
+		DeleteFunc: func(_ interface{}) {
+			nodeHealth.OnConfigMapDeleted()
+		},
+	}); err != nil {
+		return nil, fmt.Errorf("failed to add node-health ConfigMap handler: %w", err)
+	}
+
 	var ksmCtrl *ksmhcp.KSMHCPController
 	var hsInformers hypershiftinformers.SharedInformerFactory
 	var ksmKubeInformers kubeinformers.SharedInformerFactory
@@ -238,13 +318,16 @@ func (o *ValidatedControllerOptions) Complete(ctx context.Context) (*ControllerO
 		completedControllerOptions: &completedControllerOptions{
 			ctrl:                     ctrl,
 			ksmCtrl:                  ksmCtrl,
+			nodeHealth:               nodeHealth,
 			resourceWatcher:          resourceWatcher,
 			podWatcher:               podWatcher,
 			configMapWatcher:         configMapWatcher,
 			kubeInformers:            kubeInformers,
 			ksmKubeInformers:         ksmKubeInformers,
 			clusterWideKubeInformers: clusterWideKubeInformers,
 			cmWatcherInformers:       cmWatcherInformers,
+			nodeHealthInformers:      nodeHealthInformers,
+			nodeHealthCMInformers:    nodeHealthCMInformers,
 			hypershiftInformers:      hsInformers,
 			dynamicInformers:         dynInformers,
 			workers:                  o.Workers,
@@ -360,6 +443,12 @@ func (o *ControllerOptions) runControllersUnderLeaderElection(ctx context.Contex
 				if o.cmWatcherInformers != nil {
 					o.cmWatcherInformers.Start(ctx.Done())
 				}
+				if o.nodeHealthInformers != nil {
+					o.nodeHealthInformers.Start(ctx.Done())
+				}
+				if o.nodeHealthCMInformers != nil {
+					o.nodeHealthCMInformers.Start(ctx.Done())
+				}
 
 				go func() {
 					defer utilruntime.HandleCrash()
@@ -368,6 +457,13 @@ func (o *ControllerOptions) runControllersUnderLeaderElection(ctx context.Contex
 					}
 				}()
 
+				go func() {
+					defer utilruntime.HandleCrash()
+					if err := o.nodeHealth.Run(ctx, o.workers); err != nil {
+						logger.Error(err, "node-health controller failed")
+					}
+				}()
+
 				go func() {
 					defer utilruntime.HandleCrash()
 					if err := o.resourceWatcher.Run(ctx); err != nil {

mgmt-agent/deploy/templates/clusterrole.yaml

@@ -13,13 +13,31 @@ rules:
   - get
   - list
   - watch
+  - patch
 - apiGroups:
   - ""
   resources:
   - nodes/status
   verbs:
   - patch
   - update
+- apiGroups:
+  - ""
+  resources:
+  - pods
+  verbs:
+  - list
+  - watch
+- apiGroups:
+  - ""
+  resources:
+  - events
+  verbs:
+  - list
+  - watch
+  - create
+  - patch
+  - update
 - apiGroups:
   - "*"
   resources:

mgmt-agent/deploy/templates/deployment.yaml

@@ -26,6 +26,8 @@ spec:
         - "--health-address=:{{ .Values.health.port }}"
         - "--ksm-image={{ .Values.ksmImage.registry }}/{{ .Values.ksmImage.repository }}@{{ .Values.ksmImage.digest }}"
         - "--log-verbosity={{ .Values.logVerbosity }}"
+        - "--node-health-configmap={{ .Values.nodeHealth.configMapName }}"
+        - "--node-health-config-key={{ .Values.nodeHealth.configKey }}"
         image: "{{ .Values.image.registry }}/{{ .Values.image.repository }}@{{ .Values.image.digest }}"
         name: mgmt-agent-controller
         env:

mgmt-agent/deploy/templates/node-health-config.yaml

@@ -0,0 +1,10 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ .Values.nodeHealth.configMapName }}
+  namespace: {{ .Release.Namespace }}
+  labels:
+    app.kubernetes.io/name: mgmt-agent
+data:
+  {{ .Values.nodeHealth.configKey }}: |
+{{ toYaml .Values.nodeHealth.config | indent 4 }}

mgmt-agent/go.mod

@@ -20,6 +20,7 @@ require (
 	k8s.io/component-base v0.35.3
 	k8s.io/klog/v2 v2.140.0
 	k8s.io/utils v0.0.0-20260319190234-28399d86e0b5
+	sigs.k8s.io/yaml v1.6.0
 )
 
 require (
@@ -84,7 +85,6 @@ require (
 	sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
 	sigs.k8s.io/randfill v1.0.0 // indirect
 	sigs.k8s.io/structured-merge-diff/v6 v6.3.2 // indirect
-	sigs.k8s.io/yaml v1.6.0 // indirect
 )
 
 replace github.com/Azure/ARO-HCP/internal => ../internal

mgmt-agent/pkg/controller/nodehealth/config.go

@@ -0,0 +1,71 @@
+// Copyright 2025 Microsoft Corporation
+//
+// 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 nodehealth implements a management-cluster controller that detects
+// "Ready but broken" nodes from kubelet Events and labels them so they can be
+// acted on. A separate mitigation controller acts on the labelled nodes and owns
+// any disruptive action (cordon, taint, evict, delete).
+//
+// Detection is hard-coded and modular: each fault family is a Go detector that
+// reuses a shared toolkit (event-signature match, a sustained-storm floor, and
+// the load-bearing zero-successful-start check) and is a pure function of the
+// node's current state, so it is exhaustively unit-tested with no API server.
+// The only runtime configuration is the single operational switch the rollout
+// needs, enabled; there is no config-driven detection engine.
+package nodehealth
+
+import (
+	"fmt"
+
+	"sigs.k8s.io/yaml"
+)
+
+// Config holds the node-health controller's operational switch, delivered
+// via a watched ConfigMap so it can be flipped without a redeploy. Detection
+// logic, thresholds, and the SWIFT-v2 node scoping are hard-coded, not
+// configured.
+type Config struct {
+	// Enabled is a hard off switch. When false the controller still runs its
+	// informers but records no state, enqueues nothing, and takes no action.
+	Enabled bool `json:"enabled"`
+}
+
+// Default returns the built-in configuration: the controller disabled. Enabling
+// the controller is an explicit, per-environment decision.
+func Default() Config {
+	return Config{
+		Enabled: false,
+	}
+}
+
+// Parse unmarshals a YAML config document over the defaults and validates it.
+func Parse(data []byte) (Config, error) {
+	cfg := Default()
+	if len(data) > 0 {
+		if err := yaml.Unmarshal(data, &cfg); err != nil {
+			return Config{}, fmt.Errorf("failed to unmarshal node-health config: %w", err)
+		}
+	}
+	if err := cfg.Validate(); err != nil {
+		return Config{}, err
+	}
+	return cfg, nil
+}
+
+// Validate checks the configuration. The only field is a plain boolean, so the
+// only invalid state is unrepresentable; Validate exists so callers have a single,
+// stable contract and so future switches can add checks here.
+func (c *Config) Validate() error {
+	return nil
+}