diff --git a/.yamllint.yml b/.yamllint.yml index 0052824fb2d..ab0f311fe8e 100644 --- a/.yamllint.yml +++ b/.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' diff --git a/mgmt-agent/cmd/cmd.go b/mgmt-agent/cmd/cmd.go index 776c82ad9ab..94d747e8477 100644 --- a/mgmt-agent/cmd/cmd.go +++ b/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 labels only SWIFT-v2 nodes, scoped by + the AKS node label, 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.`, } diff --git a/mgmt-agent/cmd/options.go b/mgmt-agent/cmd/options.go index 5d07bf032c0..ad27d5ca77b 100644 --- a/mgmt-agent/cmd/options.go +++ b/mgmt-agent/cmd/options.go @@ -29,6 +29,7 @@ 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" @@ -36,10 +37,14 @@ import ( 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,6 +112,7 @@ type ValidatedControllerOptions struct { type completedControllerOptions struct { ctrl *controller.SwiftNICController ksmCtrl *ksmhcp.KSMHCPController + nodeHealth *nodehealth.Controller resourceWatcher *controller.ResourceWatcher podWatcher *controller.PodWatcher configMapWatcher *controller.ConfigMapWatcher @@ -104,6 +120,8 @@ type completedControllerOptions struct { 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,6 +318,7 @@ func (o *ValidatedControllerOptions) Complete(ctx context.Context) (*ControllerO completedControllerOptions: &completedControllerOptions{ ctrl: ctrl, ksmCtrl: ksmCtrl, + nodeHealth: nodeHealth, resourceWatcher: resourceWatcher, podWatcher: podWatcher, configMapWatcher: configMapWatcher, @@ -245,6 +326,8 @@ func (o *ValidatedControllerOptions) Complete(ctx context.Context) (*ControllerO 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 { diff --git a/mgmt-agent/deploy/templates/clusterrole.yaml b/mgmt-agent/deploy/templates/clusterrole.yaml index 47ad1ba5cb4..bed016c5543 100644 --- a/mgmt-agent/deploy/templates/clusterrole.yaml +++ b/mgmt-agent/deploy/templates/clusterrole.yaml @@ -13,6 +13,7 @@ rules: - get - list - watch + - patch - apiGroups: - "" resources: @@ -20,6 +21,23 @@ rules: verbs: - patch - update +- apiGroups: + - "" + resources: + - pods + verbs: + - list + - watch +- apiGroups: + - "" + resources: + - events + verbs: + - list + - watch + - create + - patch + - update - apiGroups: - "*" resources: diff --git a/mgmt-agent/deploy/templates/deployment.yaml b/mgmt-agent/deploy/templates/deployment.yaml index ce69d684107..916fb8b1d1b 100644 --- a/mgmt-agent/deploy/templates/deployment.yaml +++ b/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: diff --git a/mgmt-agent/deploy/templates/node-health-config.yaml b/mgmt-agent/deploy/templates/node-health-config.yaml new file mode 100644 index 00000000000..85a44bae950 --- /dev/null +++ b/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 }} diff --git a/mgmt-agent/go.mod b/mgmt-agent/go.mod index 258b7fd1770..89819b46ee7 100644 --- a/mgmt-agent/go.mod +++ b/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 diff --git a/mgmt-agent/pkg/controller/nodehealth/config.go b/mgmt-agent/pkg/controller/nodehealth/config.go new file mode 100644 index 00000000000..68845d2bdac --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/config.go @@ -0,0 +1,75 @@ +// 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 labeled 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 keeps its + // informers running but records no state, enqueues nothing, reconciles + // nothing, and takes no action on any node. + 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. +// Parsing is strict: an unknown key is an error rather than a silent no-op, so a +// typo in the live-edited ConfigMap (for example "enabeld: true") surfaces as a +// logged parse failure instead of leaving the controller quietly disabled. +func Parse(data []byte) (Config, error) { + cfg := Default() + if len(data) > 0 { + if err := yaml.UnmarshalStrict(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 +} diff --git a/mgmt-agent/pkg/controller/nodehealth/config_test.go b/mgmt-agent/pkg/controller/nodehealth/config_test.go new file mode 100644 index 00000000000..27bf86a24bd --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/config_test.go @@ -0,0 +1,56 @@ +// 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 + +import "testing" + +func TestDefaultIsDisabled(t *testing.T) { + cfg := Default() + if cfg.Enabled { + t.Error("default config should be disabled") + } +} + +func TestParse(t *testing.T) { + tests := []struct { + name string + data string + wantEnabled bool + wantErr bool + }{ + {name: "empty falls back to default", data: "", wantEnabled: false}, + {name: "enable", data: "enabled: true\n", wantEnabled: true}, + {name: "invalid yaml", data: "enabled: [oops", wantErr: true}, + {name: "unknown key is rejected", data: "enabeld: true\n", wantErr: true}, + {name: "unknown key alongside a known one is rejected", data: "enabled: true\nextra: 1\n", wantErr: true}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + cfg, err := Parse([]byte(tc.data)) + if tc.wantErr { + if err == nil { + t.Fatal("expected an error, got nil") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.Enabled != tc.wantEnabled { + t.Errorf("Enabled = %v, want %v", cfg.Enabled, tc.wantEnabled) + } + }) + } +} diff --git a/mgmt-agent/pkg/controller/nodehealth/consts.go b/mgmt-agent/pkg/controller/nodehealth/consts.go new file mode 100644 index 00000000000..d62affdfdef --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/consts.go @@ -0,0 +1,39 @@ +// 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 + +const ( + // ControllerName is the single source of truth for this controller's name. + // It feeds the workqueue name, which surfaces as a Prometheus label on the + // workqueue metrics, so that label never drifts from the controller. + ControllerName = "node-health" + + // fieldManager identifies this controller in server-side patch operations. + fieldManager = "mgmt-agent-node-health" + + // labelKey is the node label whose presence marks a node as wedged. A + // separate mitigation controller selects on this label. + labelKey = "node-health.aro-hcp.azure.com/status" + // labelValue is the value applied to labelKey on a wedged node. + labelValue = "wedged" + + // annotationDetector records the name of the detector that fired. + annotationDetector = "node-health.aro-hcp.azure.com/detector" + // annotationReason records a short human-readable explanation for the label. + annotationReason = "node-health.aro-hcp.azure.com/reason" + // annotationObservedAt records the RFC3339 timestamp the node was first + // labeled in the current wedge episode. + annotationObservedAt = "node-health.aro-hcp.azure.com/observed-at" +) diff --git a/mgmt-agent/pkg/controller/nodehealth/controller.go b/mgmt-agent/pkg/controller/nodehealth/controller.go new file mode 100644 index 00000000000..1a749c0264b --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/controller.go @@ -0,0 +1,550 @@ +// 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 + +import ( + "context" + "fmt" + "sync" + "sync/atomic" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/labels" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/apimachinery/pkg/util/wait" + coreinformers "k8s.io/client-go/informers/core/v1" + "k8s.io/client-go/kubernetes" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/workqueue" + "k8s.io/klog/v2" + + "github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/nodehealth/detectors" +) + +const ( + // resyncInterval is how often every node is re-enqueued for evaluation even + // without new watch events, so a firing that becomes true purely because the + // dwell elapsed (or a recovery that clears) is reconciled, and the + // wedged-nodes gauge is refreshed. + resyncInterval = 30 * time.Second + + // podByNodeIndex indexes Pods by the node they are scheduled to. + podByNodeIndex = "nodehealth-pod-by-node" + // eventBySourceHostIndex indexes Events by Event.Source.Host, the node whose + // kubelet emitted them. Keying on the emitting node instead of the involved + // Pod is deletion-safe: a node's failure Events are one lookup and the index + // does not resolve through a Pod that may already be gone. + eventBySourceHostIndex = "nodehealth-event-by-source-host" +) + +// Controller is the level-driven controller that ties detection (the pure +// detectors.Decide function, fed from shared informers) to a label/unlabel +// reconcile of wedged nodes. Its two operational switches are hot-reloadable +// via SetConfig. +type Controller struct { + nodeLister corelisters.NodeLister + podIndexer cache.Indexer + eventIndexer cache.Indexer + hasSynced []cache.InformerSynced + + // obsMu guards observedSince and successAt, which are read by the workers and + // written by the pod handlers and the config-informer thread (on a + // disabled->enabled transition). + obsMu sync.Mutex + // observedSince is the clock time the controller began observing: cache sync + // on start, or a disabled->enabled transition. Detection treats the success + // signal as indeterminate until a full window has elapsed since then, so a + // cold view after a restart or a hot enable is never misread as no-success. + observedSince time.Time + // successAt records, per node name, the most recent time the controller saw a + // non-host-network PodReadyToStartContainers=True transition. It is advanced + // from pod add/update/delete so a success survives its short-lived pod being + // deleted, and is pruned by window when read. It is cleared whenever + // observation (re)starts. + successAt map[string]time.Time + + labeler *labeler + clock func() time.Time + + workqueue workqueue.TypedRateLimitingInterface[string] + config atomic.Pointer[Config] +} + +// NewController constructs a Controller and wires the Node, Pod, and Event +// informers with the indexers the pure detectors.Decide function reads. If clock is nil, +// time.Now is used. +func NewController( + kubeClientset kubernetes.Interface, + nodeInformer coreinformers.NodeInformer, + podInformer coreinformers.PodInformer, + eventInformer coreinformers.EventInformer, + recorder record.EventRecorder, + clock func() time.Time, + initial Config, +) (*Controller, error) { + if clock == nil { + clock = time.Now + } + if err := initial.Validate(); err != nil { + return nil, fmt.Errorf("invalid initial node-health config: %w", err) + } + + if err := podInformer.Informer().AddIndexers(cache.Indexers{podByNodeIndex: podByNodeIndexFunc}); err != nil { + return nil, fmt.Errorf("failed to add pod-by-node indexer: %w", err) + } + if err := eventInformer.Informer().AddIndexers(cache.Indexers{eventBySourceHostIndex: eventBySourceHostIndexFunc}); err != nil { + return nil, fmt.Errorf("failed to add event-by-source-host indexer: %w", err) + } + + c := &Controller{ + nodeLister: nodeInformer.Lister(), + podIndexer: podInformer.Informer().GetIndexer(), + eventIndexer: eventInformer.Informer().GetIndexer(), + hasSynced: []cache.InformerSynced{ + nodeInformer.Informer().HasSynced, + podInformer.Informer().HasSynced, + eventInformer.Informer().HasSynced, + }, + labeler: newLabeler(kubeClientset, recorder, clock), + clock: clock, + successAt: make(map[string]time.Time), + workqueue: workqueue.NewTypedRateLimitingQueueWithConfig( + workqueue.DefaultTypedControllerRateLimiter[string](), + workqueue.TypedRateLimitingQueueConfig[string]{Name: ControllerName}, + ), + } + c.SetConfig(initial) + + if _, err := nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { c.enqueueNodeObject(obj) }, + UpdateFunc: func(_, newObj interface{}) { c.enqueueNodeObject(newObj) }, + }); err != nil { + return nil, fmt.Errorf("failed to add node event handler: %w", err) + } + if _, err := podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { c.recordPodSuccess(obj); c.enqueuePodObject(obj) }, + UpdateFunc: func(_, newObj interface{}) { c.recordPodSuccess(newObj); c.enqueuePodObject(newObj) }, + DeleteFunc: func(obj interface{}) { c.recordPodSuccess(obj); c.enqueuePodObject(obj) }, + }); err != nil { + return nil, fmt.Errorf("failed to add pod event handler: %w", err) + } + if _, err := eventInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ + AddFunc: func(obj interface{}) { c.enqueueEventObject(obj) }, + UpdateFunc: func(_, newObj interface{}) { c.enqueueEventObject(newObj) }, + }); err != nil { + return nil, fmt.Errorf("failed to add kubelet-event handler: %w", err) + } + + return c, nil +} + +// Config returns a copy of the current configuration snapshot. +func (c *Controller) Config() Config { return *c.config.Load() } + +// SetConfig atomically replaces the configuration snapshot. +func (c *Controller) SetConfig(cfg Config) { + c.config.Store(&cfg) +} + +// OnConfigMap parses config from the named key of a ConfigMap and applies it. +// A parse error is logged and the previous config is retained. +func (c *Controller) OnConfigMap(cm *corev1.ConfigMap, key string) { + data, ok := cm.Data[key] + if !ok { + klog.InfoS("node-health config key not found in ConfigMap, keeping current config", + "configmap", cm.Name, "key", key) + return + } + cfg, err := Parse([]byte(data)) + if err != nil { + klog.ErrorS(err, "invalid node-health config, keeping current config", "configmap", cm.Name) + return + } + klog.InfoS("node-health config reloaded", "configmap", cm.Name, "enabled", cfg.Enabled) + wasEnabled := c.config.Load().Enabled + c.SetConfig(cfg) + // A disabled->enabled transition restarts observation: the success map is + // discarded and the observedSince warm-up begins now, so the success signal + // is treated as indeterminate until a full window has been watched under the + // enabled controller. + if cfg.Enabled && !wasEnabled { + c.beginObserving(c.clock()) + } + // A hot enabled->disabled flip stops all reconciles, so the gauge would + // otherwise freeze at its last value. Zero it immediately so a disabled + // controller reads zero wedged nodes rather than a stale count. + if !cfg.Enabled && wasEnabled { + wedgedNodes.Set(0) + } +} + +// OnConfigMapDeleted reverts to the safe built-in default (disabled) when the +// config ConfigMap is removed, so rollback via ConfigMap deletion is not +// surprising for operators. +func (c *Controller) OnConfigMapDeleted() { + klog.InfoS("node-health ConfigMap deleted, reverting to disabled default config") + c.SetConfig(Default()) + // Reverting to the disabled default stops all reconciles, so zero the gauge + // rather than leave it frozen at its last count. + wedgedNodes.Set(0) +} + +func (c *Controller) enqueue(node string) { + if node == "" || !c.config.Load().Enabled { + return + } + c.workqueue.Add(node) +} + +func (c *Controller) enqueueNodeObject(obj interface{}) { + if name, ok := nodeName(obj); ok { + c.enqueue(name) + } +} + +func (c *Controller) enqueuePodObject(obj interface{}) { + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } + if pod, ok := obj.(*corev1.Pod); ok { + c.enqueue(pod.Spec.NodeName) + } +} + +func (c *Controller) enqueueEventObject(obj interface{}) { + ev, ok := obj.(*corev1.Event) + if !ok { + return + } + // Enqueue the node whose kubelet emitted the Event. Reading Source.Host + // directly is deletion-safe: it does not depend on the involved Pod still + // being cached. + if ev.Source.Host != "" { + c.enqueue(ev.Source.Host) + } +} + +// beginObserving (re)starts the observation warm-up: it stamps observedSince to +// now and clears any recorded success history so a fresh full window must be +// watched before the success signal is trusted. Called after cache sync in Run +// and on a disabled->enabled transition. +func (c *Controller) beginObserving(now time.Time) { + c.obsMu.Lock() + defer c.obsMu.Unlock() + c.observedSince = now + c.successAt = make(map[string]time.Time) +} + +// recordPodSuccess advances the per-node success timestamp when a pod shows a +// fresh sandbox (non-host-network PodReadyToStartContainers=True). It is called +// from the pod add/update/delete handlers, so a success is captured from the +// pod's last-seen state even as the short-lived pod is being deleted. A disabled +// controller records nothing: beginObserving discards the map on startup and on +// every disabled->enabled transition, so anything recorded while off could never +// be read, and keeping it would grow the map for every node name ever seen. +func (c *Controller) recordPodSuccess(obj interface{}) { + if !c.config.Load().Enabled { + return + } + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } + pod, ok := obj.(*corev1.Pod) + if !ok || pod.Spec.NodeName == "" { + return + } + at, ok := detectors.SuccessAt(pod) + if !ok { + return + } + c.obsMu.Lock() + defer c.obsMu.Unlock() + if prev, exists := c.successAt[pod.Spec.NodeName]; !exists || at.After(prev) { + c.successAt[pod.Spec.NodeName] = at + } +} + +// observation returns the node's warm-up start and its most recent recorded +// success, pruning any success older than the widest detector window so the map +// does not grow without bound. +func (c *Controller) observation(node string) (observedSince, lastSuccessAt time.Time) { + c.obsMu.Lock() + defer c.obsMu.Unlock() + observedSince = c.observedSince + at, ok := c.successAt[node] + if !ok { + return observedSince, time.Time{} + } + // Measured in absolute terms so a future-dated timestamp (kubelet clock skew) + // is pruned rather than retained forever by a negative age. + if c.clock().Sub(at).Abs() >= detectors.MaxWindow() { + delete(c.successAt, node) + return observedSince, time.Time{} + } + return observedSince, at +} + +// Run starts the controller workers and blocks until the context is cancelled. +func (c *Controller) Run(ctx context.Context, workers int) error { + defer utilruntime.HandleCrash() + defer c.workqueue.ShutDown() + + logger := klog.FromContext(ctx) + logger.Info("Starting node-health controller") + + logger.Info("Waiting for informer caches to sync") + if ok := cache.WaitForCacheSync(ctx.Done(), c.hasSynced...); !ok { + return fmt.Errorf("failed to wait for caches to sync") + } + // Record when the controller begins observing. Until a full detector window + // has elapsed past this point, detection treats the success signal as + // indeterminate so a cold view is never misread as no-success. This also + // clears any success history carried across the cache resync. + c.beginObserving(c.clock()) + + logger.Info("Starting workers", "count", workers) + for i := 0; i < workers; i++ { + go wait.UntilWithContext(ctx, c.runWorker, time.Second) + } + go wait.UntilWithContext(ctx, c.resyncAll, resyncInterval) + + logger.Info("Node-health controller started") + <-ctx.Done() + logger.Info("Shutting down node-health controller") + return nil +} + +// resyncAll re-enqueues every candidate node so time-based firings and +// recoveries are reconciled without new Events, and refreshes the wedged-nodes +// gauge. +// +// The sweep is the union of two cheap lister selectors rather than every node in +// the cluster. SWIFT-v2 nodes are the only nodes a detector can apply to, so they +// are the detection candidates; wedged-labeled nodes are added so a label left +// on a node that has since stopped being a detection candidate is still swept and +// retired. Relying on the node watch alone for that cleanup would miss any +// transition that happened while the controller was down. +func (c *Controller) resyncAll(ctx context.Context) { + if !c.config.Load().Enabled { + wedgedNodes.Set(0) + return + } + logger := klog.FromContext(ctx) + + swiftSelector := labels.SelectorFromSet(labels.Set{detectors.SwiftV2LabelKey: detectors.SwiftV2LabelValue}) + swiftNodes, err := c.nodeLister.List(swiftSelector) + if err != nil { + logger.Error(err, "resync: failed to list SWIFT nodes") + return + } + + wedgedNodeList, err := c.listWedgedNodes() + if err != nil { + logger.Error(err, "resync: failed to list wedged nodes") + return + } + wedgedNodes.Set(float64(len(wedgedNodeList))) + + // The two selectors overlap on the common case (a wedged SWIFT node), so + // dedupe before enqueuing. The workqueue would coalesce anyway; deduping here + // keeps the sweep's cost proportional to distinct nodes. + seen := make(map[string]struct{}, len(swiftNodes)+len(wedgedNodeList)) + for _, n := range swiftNodes { + seen[n.Name] = struct{}{} + } + for _, n := range wedgedNodeList { + seen[n.Name] = struct{}{} + } + for name := range seen { + c.enqueue(name) + } +} + +// listWedgedNodes returns the nodes currently carrying the wedged health label. +// It selects on the wedged label alone rather than intersecting with the SWIFT-v2 +// selector, so a node that still carries the label after losing its SWIFT label +// is still found rather than silently dropped from the gauge and the sweep. +func (c *Controller) listWedgedNodes() ([]*corev1.Node, error) { + return c.nodeLister.List(labels.SelectorFromSet(labels.Set{labelKey: labelValue})) +} + +// countWedgedNodes returns how many nodes currently carry the wedged health +// label. +func (c *Controller) countWedgedNodes() (int, error) { + nodes, err := c.listWedgedNodes() + if err != nil { + return 0, err + } + return len(nodes), nil +} + +func (c *Controller) runWorker(ctx context.Context) { + for c.processNextWorkItem(ctx) { + } +} + +func (c *Controller) processNextWorkItem(ctx context.Context) bool { + key, shutdown := c.workqueue.Get() + if shutdown { + return false + } + defer c.workqueue.Done(key) + + if err := c.syncHandler(ctx, key); err != nil { + utilruntime.HandleError(fmt.Errorf("error reconciling node %q: %w", key, err)) + c.workqueue.AddRateLimited(key) + return true + } + c.workqueue.Forget(key) + return true +} + +// syncHandler reconciles a single node's health label to match the pure +// detectors.Decide verdict computed from the informers' current view of that +// node. +func (c *Controller) syncHandler(ctx context.Context, name string) error { + logger := klog.FromContext(ctx).WithValues("node", name) + cfg := c.config.Load() + if !cfg.Enabled { + return nil + } + + node, err := c.nodeLister.Get(name) + if err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("get node: %w", err) + } + + // Answer the ownership question before paying for the node's Pods and Events. + // Every Pod and kubelet Event enqueues the node it belongs to, so on a + // management cluster most reconciles are for nodes no detector owns. Decide + // applies the same gate, so this reaches the same verdict without the scan. + if !detectors.AnyApplies(node) { + return c.retireStaleLabel(ctx, node, logger) + } + + events, pods, err := c.gather(name) + if err != nil { + return err + } + + observedSince, lastSuccessAt := c.observation(name) + decision, snap := detectors.Decide(node, events, pods, c.clock(), observedSince, lastSuccessAt) + switch decision { + case detectors.DecisionWedged: + changed, err := c.labeler.label(ctx, node, snap.DetectorName, snap) + if err != nil { + return err + } + if changed { + detectionsTotal.WithLabelValues(snap.DetectorName).Inc() + logger.Info("detector fired; node labeled wedged", + "detector", snap.DetectorName, + "failures", snap.FailureCount, + "recentSuccess", snap.RecentSuccess) + } + // Mitigation of a labeled node (cordon, taint, evict, delete) is owned + // by a separate controller; this controller stops at labeling. + case detectors.DecisionHealthy: + if _, err := c.labeler.unlabel(ctx, node); err != nil { + return err + } + case detectors.DecisionNotApplicable: + // Reached only if Decide's ownership gate disagrees with the AnyApplies + // short-circuit above, which it cannot for the same node. Kept so the + // switch covers Decide's full contract rather than relying on the caller. + return c.retireStaleLabel(ctx, node, logger) + case detectors.DecisionUnknown: + logger.V(5).Info("insufficient evidence; leaving label unchanged") + } + return nil +} + +// retireStaleLabel drops the wedged label from a node no detector owns. Such a +// label can only be one this controller left behind, for example when the node +// stopped being a SWIFT-v2 node while labeled. unlabel is a no-op when the label +// is absent or carries a foreign value. +func (c *Controller) retireStaleLabel(ctx context.Context, node *corev1.Node, logger klog.Logger) error { + changed, err := c.labeler.unlabel(ctx, node) + if err != nil { + return err + } + if changed { + logger.Info("retired stale wedged label; no detector applies to this node") + } + return nil +} + +// gather returns the Events and Pods the informers currently hold for a node: +// the Pods scheduled to it, and the Events its kubelet emitted (keyed by +// Event.Source.Host). Reading Events by emitting node rather than by involved Pod +// keeps the failure evidence available even after the Pod is deleted. +func (c *Controller) gather(nodeName string) ([]*corev1.Event, []*corev1.Pod, error) { + podObjs, err := c.podIndexer.ByIndex(podByNodeIndex, nodeName) + if err != nil { + return nil, nil, fmt.Errorf("list pods for node: %w", err) + } + pods := make([]*corev1.Pod, 0, len(podObjs)) + for _, o := range podObjs { + if pod, ok := o.(*corev1.Pod); ok { + pods = append(pods, pod) + } + } + + evObjs, err := c.eventIndexer.ByIndex(eventBySourceHostIndex, nodeName) + if err != nil { + return nil, nil, fmt.Errorf("list events for node: %w", err) + } + events := make([]*corev1.Event, 0, len(evObjs)) + for _, eo := range evObjs { + if ev, ok := eo.(*corev1.Event); ok { + events = append(events, ev) + } + } + return events, pods, nil +} + +func podByNodeIndexFunc(obj interface{}) ([]string, error) { + pod, ok := obj.(*corev1.Pod) + if !ok || pod.Spec.NodeName == "" { + return nil, nil + } + return []string{pod.Spec.NodeName}, nil +} + +func eventBySourceHostIndexFunc(obj interface{}) ([]string, error) { + ev, ok := obj.(*corev1.Event) + if !ok || ev.Source.Host == "" { + return nil, nil + } + return []string{ev.Source.Host}, nil +} + +func nodeName(obj interface{}) (string, bool) { + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + obj = tombstone.Obj + } + node, ok := obj.(*corev1.Node) + if !ok { + return "", false + } + return node.Name, true +} diff --git a/mgmt-agent/pkg/controller/nodehealth/controller_test.go b/mgmt-agent/pkg/controller/nodehealth/controller_test.go new file mode 100644 index 00000000000..c6b04146864 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/controller_test.go @@ -0,0 +1,590 @@ +// 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 + +import ( + "context" + "fmt" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/informers" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + + "github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/nodehealth/detectors" +) + +const ( + testHost = "aks-userswft1-0" + sig = "route ip+net: no such network interface" + reason = "FailedCreatePodSandBox" +) + +// newTestController builds a controller wired to fake informers, with a fixed +// clock, and returns it plus the fake clientset and the three informers so a test +// can push objects straight into their stores and call syncHandler without +// running the workqueue. Seeded nodes are added so nodeLister can resolve them. +func newTestController(t *testing.T, enabled bool, nodes ...*corev1.Node) (*Controller, *fake.Clientset, informers.SharedInformerFactory) { + t.Helper() + rt := make([]runtime.Object, 0, len(nodes)) + for _, n := range nodes { + rt = append(rt, n) + } + client := fake.NewSimpleClientset(rt...) + factory := informers.NewSharedInformerFactory(client, 0) + c, err := NewController( + client, + factory.Core().V1().Nodes(), + factory.Core().V1().Pods(), + factory.Core().V1().Events(), + record.NewFakeRecorder(64), + func() time.Time { return testNow }, + Config{Enabled: enabled}, + ) + if err != nil { + t.Fatalf("NewController: %v", err) + } + for _, n := range nodes { + if err := factory.Core().V1().Nodes().Informer().GetStore().Add(n); err != nil { + t.Fatalf("seed node: %v", err) + } + } + return c, client, factory +} + +func swiftReadyNode(name string) *corev1.Node { + return &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{detectors.SwiftV2LabelKey: detectors.SwiftV2LabelValue}, + }, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}, + }, + } +} + +func stuckPodOn(host, name string, since time.Time) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name, UID: types.UID("uid-" + name)}, + Spec: corev1.PodSpec{NodeName: host}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionFalse, + LastTransitionTime: metav1.NewTime(since), + }}, + }, + } +} + +func startedPodOn(host, name string, at time.Time, hostNet bool) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: name, UID: types.UID("uid-" + name)}, + Spec: corev1.PodSpec{NodeName: host, HostNetwork: hostNet}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionTrue, + LastTransitionTime: metav1.NewTime(at), + }}, + }, + } +} + +func failEventOn(host, pod string, last time.Time) *corev1.Event { + return &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: pod + ".evt"}, + InvolvedObject: corev1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: pod, UID: types.UID("uid-" + pod)}, + Source: corev1.EventSource{Host: host, Component: "kubelet"}, + Reason: reason, + Message: "failed to setup network for sandbox: " + sig, + LastTimestamp: metav1.NewTime(last), + FirstTimestamp: metav1.NewTime(last), + } +} + +// seedStuckStorm pushes n stuck pods and their matching failure Events into the +// informer stores for host, all stuck since the given time. +func seedStuckStorm(t *testing.T, c *Controller, host string, n int, since time.Time) { + t.Helper() + for i := 0; i < n; i++ { + name := fmt.Sprintf("p%d", i) + if err := c.podIndexer.Add(stuckPodOn(host, name, since)); err != nil { + t.Fatalf("add pod: %v", err) + } + if err := c.eventIndexer.Add(failEventOn(host, name, testNow.Add(-30*time.Second))); err != nil { + t.Fatalf("add event: %v", err) + } + } +} + +func isWedged(t *testing.T, client *fake.Clientset, name string) bool { + t.Helper() + return getNode(t, client, name).Labels[labelKey] == labelValue +} + +// warmObserved is far enough in the past that a full detector window has elapsed. +var warmObserved = testNow.Add(-30 * time.Minute) + +func TestSyncHandlerWedgesOnSustainedStorm(t *testing.T) { + node := swiftReadyNode(testHost) + c, client, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + seedStuckStorm(t, c, testHost, 3, testNow.Add(-15*time.Minute)) + + if err := c.syncHandler(context.Background(), testHost); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if !isWedged(t, client, testHost) { + t.Fatal("node should be labeled wedged after a sustained storm") + } +} + +func TestSyncHandlerDisabledIsNoop(t *testing.T) { + node := swiftReadyNode(testHost) + c, client, _ := newTestController(t, false, node) + c.beginObserving(warmObserved) + seedStuckStorm(t, c, testHost, 3, testNow.Add(-15*time.Minute)) + + if err := c.syncHandler(context.Background(), testHost); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if isWedged(t, client, testHost) { + t.Fatal("a disabled controller must never label a node") + } +} + +func TestSyncHandlerColdWarmupDoesNotWedge(t *testing.T) { + node := swiftReadyNode(testHost) + c, client, _ := newTestController(t, true, node) + // Observation began just now: a full window has not been watched yet. + c.beginObserving(testNow.Add(-1 * time.Minute)) + seedStuckStorm(t, c, testHost, 3, testNow.Add(-15*time.Minute)) + + if err := c.syncHandler(context.Background(), testHost); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if isWedged(t, client, testHost) { + t.Fatal("a cold controller must not invent a fresh wedge before a full window") + } +} + +func TestSyncHandlerRecoveryUnlabels(t *testing.T) { + // Node already carries the wedged label; a recorded success and no stuck pods + // must clear it. + node := swiftReadyNode(testHost) + node.Labels[labelKey] = labelValue + c, client, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + c.recordPodSuccess(startedPodOn(testHost, "ok", testNow.Add(-1*time.Minute), false)) + + if err := c.syncHandler(context.Background(), testHost); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if isWedged(t, client, testHost) { + t.Fatal("a recorded success with no stuck pods should unlabel the node") + } +} + +func TestShortLivedSuccessSurvivesPodDeletion(t *testing.T) { + // The key regression: a pod's sandbox came up (success) and the pod was then + // deleted, so it is no longer in the informer store, while a storm of stuck + // pods remains. The recorded success must still block a false wedge. + node := swiftReadyNode(testHost) + c, client, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + + // A short-lived success, observed via the delete handler (the pod is never + // added to the store, mirroring a pod already gone by reconcile time). + success := startedPodOn(testHost, "shortlived", testNow.Add(-2*time.Minute), false) + c.recordPodSuccess(cache.DeletedFinalStateUnknown{Key: "ns/shortlived", Obj: success}) + + seedStuckStorm(t, c, testHost, 3, testNow.Add(-15*time.Minute)) + + if err := c.syncHandler(context.Background(), testHost); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if isWedged(t, client, testHost) { + t.Fatal("a recorded success from a since-deleted pod must prevent a false wedge") + } +} + +func TestHotEnableResetsObservation(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, false, node) + + // A disabled controller records nothing, so there is no state to carry into + // the enabled controller. + c.recordPodSuccess(startedPodOn(testHost, "ok", testNow.Add(-2*time.Minute), false)) + if _, at := c.observation(testHost); !at.IsZero() { + t.Fatalf("a disabled controller must not record successes, got %v", at) + } + + // A disabled->enabled transition must reset observedSince to now and clear the + // success history, so the controller re-earns a full window before it can wedge. + cm := &corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "node-health-config"}, + Data: map[string]string{"config.yaml": "enabled: true"}, + } + c.OnConfigMap(cm, "config.yaml") + + obs, at := c.observation(testHost) + if !obs.Equal(testNow) { + t.Errorf("observedSince = %v, want %v (reset on enable)", obs, testNow) + } + if !at.IsZero() { + t.Error("success history should be cleared on a disabled->enabled transition") + } +} + +func TestObservationPrunesSuccessOlderThanMaxWindow(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + + // A success older than the widest detector window can never matter, so it is + // pruned on read. + old := testNow.Add(-detectors.MaxWindow() - time.Minute) + c.recordPodSuccess(startedPodOn(testHost, "stale", old, false)) + if _, at := c.observation(testHost); !at.IsZero() { + t.Errorf("stale success should be pruned, got %v", at) + } +} + +func TestRecordPodSuccessExcludesHostNetworkAndFailures(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + + // Host-network pods reach the condition without a delegated NIC. + c.recordPodSuccess(startedPodOn(testHost, "hostnet", testNow.Add(-1*time.Minute), true)) + // A stuck (False) pod is not a success. + c.recordPodSuccess(stuckPodOn(testHost, "stuck", testNow.Add(-1*time.Minute))) + if _, at := c.observation(testHost); !at.IsZero() { + t.Errorf("neither host-network nor stuck pods should record a success, got %v", at) + } +} + +func TestRecordPodSuccessKeepsMostRecent(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + + c.recordPodSuccess(startedPodOn(testHost, "a", testNow.Add(-5*time.Minute), false)) + c.recordPodSuccess(startedPodOn(testHost, "b", testNow.Add(-1*time.Minute), false)) + // An older success arriving later must not move the timestamp backwards. + c.recordPodSuccess(startedPodOn(testHost, "c", testNow.Add(-8*time.Minute), false)) + + if _, at := c.observation(testHost); !at.Equal(testNow.Add(-1 * time.Minute)) { + t.Errorf("lastSuccessAt = %v, want the most recent (%v)", at, testNow.Add(-1*time.Minute)) + } +} + +func TestPodByNodeIndexFunc(t *testing.T) { + keys, err := podByNodeIndexFunc(stuckPodOn(testHost, "p0", testNow)) + if err != nil { + t.Fatalf("index func: %v", err) + } + if len(keys) != 1 || keys[0] != testHost { + t.Errorf("podByNodeIndexFunc = %v, want [%q]", keys, testHost) + } + // A pod not yet scheduled has no node key. + unscheduled := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Name: "x"}} + if keys, _ := podByNodeIndexFunc(unscheduled); len(keys) != 0 { + t.Errorf("unscheduled pod should index to no node, got %v", keys) + } +} + +func TestEventBySourceHostIndexFunc(t *testing.T) { + keys, err := eventBySourceHostIndexFunc(failEventOn(testHost, "p0", testNow)) + if err != nil { + t.Fatalf("index func: %v", err) + } + if len(keys) != 1 || keys[0] != testHost { + t.Errorf("eventBySourceHostIndexFunc = %v, want [%q]", keys, testHost) + } + // An Event with no Source.Host indexes to nothing (keeps the lookup deletion-safe). + noHost := &corev1.Event{ObjectMeta: metav1.ObjectMeta{Name: "e"}} + if keys, _ := eventBySourceHostIndexFunc(noHost); len(keys) != 0 { + t.Errorf("hostless event should index to no node, got %v", keys) + } +} + +func TestEnqueueGatedByEnabled(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, false, node) + c.enqueue(testHost) + if c.workqueue.Len() != 0 { + t.Fatal("a disabled controller must not enqueue work") + } + c.SetConfig(Config{Enabled: true}) + c.enqueue(testHost) + if c.workqueue.Len() != 1 { + t.Fatalf("an enabled controller should enqueue, got queue len %d", c.workqueue.Len()) + } +} + +func TestGatherReadsPodsAndEventsForNode(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, true, node) + seedStuckStorm(t, c, testHost, 2, testNow.Add(-15*time.Minute)) + // Noise on another node must not leak into this node's gather. + if err := c.podIndexer.Add(stuckPodOn("other-node", "q0", testNow)); err != nil { + t.Fatalf("add pod: %v", err) + } + + events, pods, err := c.gather(testHost) + if err != nil { + t.Fatalf("gather: %v", err) + } + if len(pods) != 2 { + t.Errorf("gather pods = %d, want 2 (other node excluded)", len(pods)) + } + if len(events) != 2 { + t.Errorf("gather events = %d, want 2", len(events)) + } +} + +func TestCountWedgedNodesIncludesNonSwiftNodes(t *testing.T) { + // A node that carries the wedged label but has lost (or never had) the SWIFT + // label must still be counted. Scoping the count to the SWIFT selector used + // to hide exactly the stale labels an operator most needs to see. + wedgedSwift := swiftReadyNode("swift-wedged") + wedgedSwift.Labels[labelKey] = labelValue + + healthySwift := swiftReadyNode("swift-healthy") + + wedgedNonSwift := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "plain-wedged", + Labels: map[string]string{labelKey: labelValue}, + }} + + foreignValue := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "plain-foreign", + Labels: map[string]string{labelKey: "somethingelse"}, + }} + + c, _, _ := newTestController(t, true, wedgedSwift, healthySwift, wedgedNonSwift, foreignValue) + + got, err := c.countWedgedNodes() + if err != nil { + t.Fatalf("countWedgedNodes: %v", err) + } + if want := 2; got != want { + t.Errorf("countWedgedNodes() = %d, want %d (both wedged nodes, SWIFT-labeled or not)", got, want) + } +} + +func TestSyncHandlerRetiresLabelWhenNoDetectorApplies(t *testing.T) { + // A node that carries our label but is no longer a detection candidate (it + // lost the SWIFT label) must have the label retired, not held forever. + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + Annotations: map[string]string{ + annotationDetector: "swift-vf-teardown", + annotationReason: "x", + }, + }, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}, + }, + } + c, client, _ := newTestController(t, true, node) + + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler: %v", err) + } + + got := getNode(t, client, "n1") + if _, ok := got.Labels[labelKey]; ok { + t.Error("stale wedged label should be retired when no detector applies") + } + if _, ok := got.Annotations[annotationDetector]; ok { + t.Error("detection record should be cleared with the label") + } +} + +func TestSyncHandlerRetiresLabelOnNotReadyNonCandidate(t *testing.T) { + // Ownership is decided before readiness: a NotReady node that no detector + // applies to is still not ours to hold a label on. + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + }, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionFalse}}, + }, + } + c, client, _ := newTestController(t, true, node) + + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if _, ok := getNode(t, client, "n1").Labels[labelKey]; ok { + t.Error("stale label on a non-candidate node should be retired regardless of readiness") + } +} + +func TestSyncHandlerNotApplicablePreservesForeignLabelValue(t *testing.T) { + // Our key with someone else's value is not ours to remove. + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: "somethingelse"}, + }, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}, + }, + } + c, client, _ := newTestController(t, true, node) + + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if got := getNode(t, client, "n1").Labels[labelKey]; got != "somethingelse" { + t.Errorf("foreign label value = %q, want it untouched", got) + } +} + +func TestSyncHandlerNotReadySwiftNodeRetainsLabel(t *testing.T) { + // A NotReady node that IS a detection candidate is deferred to node + // lifecycle: the label must be retained, not retired. + node := swiftReadyNode("n1") + node.Labels[labelKey] = labelValue + node.Status.Conditions = []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionFalse}} + c, client, _ := newTestController(t, true, node) + + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if getNode(t, client, "n1").Labels[labelKey] != labelValue { + t.Error("a NotReady detection candidate should retain its label") + } +} + +func TestRestartRetainsExistingWedgedLabel(t *testing.T) { + // The restart view is asymmetric on purpose: a node that still carries the + // label is only unlabeled on positive evidence of recovery, never because + // the Event view is briefly empty just after startup. + node := swiftReadyNode("n1") + node.Labels[labelKey] = labelValue + node.Annotations = map[string]string{annotationDetector: "swift-vf-teardown"} + + c, client, _ := newTestController(t, true, node) + // A freshly started controller: observation begins now, no success history, + // and no Events or Pods in cache. + c.beginObserving(testNow) + + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler: %v", err) + } + if getNode(t, client, "n1").Labels[labelKey] != labelValue { + t.Error("a cold restart must retain an existing wedged label, not clear it") + } +} + +func TestOnConfigMapDeletedRevertsToDisabled(t *testing.T) { + c, _, _ := newTestController(t, true) + if !c.Config().Enabled { + t.Fatal("precondition: controller should start enabled") + } + + c.OnConfigMapDeleted() + + if c.Config().Enabled { + t.Error("deleting the ConfigMap must revert to the disabled built-in default") + } +} + +func TestOnConfigMapInvalidYAMLKeepsPreviousConfig(t *testing.T) { + c, _, _ := newTestController(t, true) + + c.OnConfigMap(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "cm"}, + Data: map[string]string{"config.yaml": "enabled: [not a bool"}, + }, "config.yaml") + + if !c.Config().Enabled { + t.Error("an unparseable config must be rejected and the previous config kept") + } +} + +func TestOnConfigMapMissingKeyKeepsPreviousConfig(t *testing.T) { + c, _, _ := newTestController(t, true) + + c.OnConfigMap(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "cm"}, + Data: map[string]string{"other.yaml": "enabled: false"}, + }, "config.yaml") + + if !c.Config().Enabled { + t.Error("a ConfigMap without the expected key must keep the previous config") + } +} + +func TestOnConfigMapHotDisableTakesEffect(t *testing.T) { + c, _, _ := newTestController(t, true) + + c.OnConfigMap(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Name: "cm"}, + Data: map[string]string{"config.yaml": "enabled: false"}, + }, "config.yaml") + + if c.Config().Enabled { + t.Fatal("hot disable should take effect") + } + // A disabled controller must take no action even if a node is enqueued. + if err := c.syncHandler(context.Background(), "n1"); err != nil { + t.Fatalf("syncHandler while disabled: %v", err) + } +} + +// TestObservationPrunesFutureDatedSuccess covers kubelet clock skew. The success +// timestamp is a Pod condition stamped by the kubelet's own clock, so it can +// arrive dated ahead of the controller's clock. recordPodSuccess keeps the +// latest timestamp, so a far-future one would otherwise win permanently and +// discard every genuine success that followed it, silently suppressing +// detection until the wall clock caught up. It is not usable evidence, so it is +// pruned on read like any other out-of-window entry. +func TestObservationPrunesFutureDatedSuccess(t *testing.T) { + node := swiftReadyNode(testHost) + c, _, _ := newTestController(t, true, node) + c.beginObserving(warmObserved) + + future := testNow.Add(detectors.MaxWindow() + time.Hour) + c.recordPodSuccess(startedPodOn(testHost, "skewed", future, false)) + if _, at := c.observation(testHost); !at.IsZero() { + t.Errorf("future-dated success should be pruned, got %v", at) + } + + // Skew inside the window is ordinary and stays usable. + near := testNow.Add(1 * time.Second) + c.recordPodSuccess(startedPodOn(testHost, "near", near, false)) + if _, at := c.observation(testHost); at.IsZero() { + t.Error("a success within the window must be retained despite small skew") + } +} diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/README.md b/mgmt-agent/pkg/controller/nodehealth/detectors/README.md new file mode 100644 index 00000000000..ff831285656 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/README.md @@ -0,0 +1,143 @@ +# node-health detectors + +This package is the pure detection core of the node-health controller. It answers +one question, with no Kubernetes I/O: given a node and the Events and Pods +currently held for it, is the node **wedged**, **healthy** (recovered), or is the +evidence insufficient to say? The controller package (`..`) consumes the verdict +and is the only thing that talks to the API server (labeling, events, metrics). + +Keeping detection isolated and side-effect free is what makes it exhaustively +table-testable: every rule below is exercised by `decide_test.go` against +synthetic Events and Pods, no cluster required. + +## The decision + +`Decide(node, events, pods, now, observedSince, lastSuccessAt) (Decision, Snapshot)` +returns one of: + +- `DecisionWedged`: a detector fired. The controller ensures the node carries the + wedged label. +- `DecisionHealthy`: recovery is positively observed (a recorded per-node success + falls within the window). The controller ensures the node is not labeled. +- `DecisionUnknown`: the evidence is insufficient, so leave the label exactly as it + is. This covers a `NotReady` node (deferred to node lifecycle) and the cold view + before a full window has been observed since observation began + (`observedSince`). Because `Unknown` never removes a label, an existing wedged + label survives a restart until recovery is actually seen, and a cold view never + invents a fresh wedge. +- `DecisionNotApplicable`: no detector applies to the node, so none can ever mark + it wedged. The controller ensures the node is not labeled. This is a definitive + statement about ownership, unlike `Unknown` which is a statement about evidence, + and it is what retires a label left on a node that has since stopped being a + detection candidate. Applicability is evaluated ahead of the `NotReady` check, + because a node no detector owns is not ours to hold a label on whether it is + Ready or not. + +`Decide` short-circuits on a wedge, only reports `Healthy` on positive success +evidence, and otherwise stays `Unknown`. It is a pure function of the node's +observed state and the controller's recorded per-node success history: same +inputs, same output. + +## How a detector is structured + +A **detector** is one fault family. Detection is modular so that adding a family +is adding code in its own file, never editing the engine: + +- **`decide.go`**: the engine. It defines the `Detector` interface + (`Applies`, `Evaluate`, `MeetsThreshold`, plus `Name`/`Reason`), the `Snapshot` of + evaluated evidence, the `Decision` type, the `registry` of detectors, and + `Decide`, which iterates the registry without knowing any concrete type. +- **`signature_detector.go`**: the shared base. `signatureDetector` implements + the `Detector` interface once for the common shape and carries the reusable + toolkit (windowed correlation of failure Events to stuck pods, condition-based + dwell math, node-Ready and success helpers). Families that fit this shape reuse + all of it. +- **`swift_vf.go`**: one fault family's specifics only, the `swift-vf-teardown` + detector value plus its applicability predicate. No evaluation logic lives here. + +`Decide` walks `registry`. For each detector it calls `Applies(node)` (skip if the +node can't exhibit the fault), then `Evaluate(...)` to gather the evidence +`Snapshot`, then `MeetsThreshold(snap, now, observedSince)`. The first detector that fires +wins and the node is `Wedged`; if none fires but some success was seen, the node is +`Healthy`; otherwise `Unknown`. + +## The `signatureDetector` shape + +Most kubelet-visible wedges look the same: pods that cannot get a sandbox, marked +by a particular Event `reason` whose message matches a signature, with no fresh +sandbox coming up. The base captures that shape with these knobs (all code +constants, never config): + +| field | meaning | +|----------------------|---------| +| `appliesTo` | predicate limiting the detector to nodes that can physically exhibit the fault (nil = all nodes) | +| `eventReason` | the kubelet Event `reason` that marks a pod as failing sandbox creation | +| `signatures` | regexes matched against the failure Event message; any match is a hit | +| `failuresFloor` | minimum number of stuck pods that have each individually been stuck past `dwell` (a floor so the storm is sustained, not the trigger) | +| `window` | rolling window over which failures and successes are counted | +| `dwell` | how long each pod must have been stuck before it counts toward the floor, filtering transient bursts | +| `requireZeroSuccess` | when true, firing requires zero fresh sandbox successes in the window | + +`MeetsThreshold` is: `SustainedCount >= failuresFloor` (stuck pods that have each been stuck +at least `dwell`) **and** (`RecentSuccess == false` when `requireZeroSuccess`, and +only once a full `window` has elapsed since `observedSince`). + +Three details that matter, all of them chosen so no signal depends on Events or +Pods that the apiserver garbage-collects: + +- **The floor and dwell are read from live Pod state, not from Event counts.** A + "stuck pod" is a pod whose `PodReadyToStartContainers` condition is `False` and + which is the subject of a matching failure Event in the window. Dwell is measured + per pod, from each pod's own `PodReadyToStartContainers=False` `lastTransitionTime`, + and `SustainedCount` counts the stuck pods whose own dwell has elapsed, so the + floor cannot be met by one long-stuck pod plus a burst of brand-new ones. Both are + durable across Event GC and a controller restart. `FailureCount` (all stuck pods) + and `StuckSince` (the oldest) are retained for reporting only. Events are used + only to classify which pods are failing (looked up per node by `Event.Source.Host`, + which is deletion-safe, and correlated to a specific Pod by `InvolvedObject.UID` so + a recycled name cannot inherit a deleted Pod's Events); they are never counted. +- **Success presence is load-bearing, and it is a recorded history, not a scan.** + `requireZeroSuccess` is what separates a hard wedge (VF gone, no fresh sandboxes) + from a flap (some pods still starting). A success is a non-host-network pod whose + `PodReadyToStartContainers` condition transitioned to `True`; host-network pods + are excluded, since they reach the condition without a delegated NIC. The + controller records the most recent such transition per node (`SuccessAt` feeds a + bounded `lastSuccessAt`), advanced from pod add/update **and** the last-seen state + of a deleted pod, so a short-lived success survives its Pod being garbage + collected. `Decide` sets `RecentSuccess` when `lastSuccessAt` falls inside the + window; it never scans the passed-in Pods for success. +- **No recorded success is `Unknown`, never a wedge, until the controller has + watched a full window.** Just after a restart or a disabled→enabled transition + the recorded history starts empty, so a detector does not fire until + `now - observedSince >= window`; before that, an empty success history is treated + as indeterminate, not as proof of a wedge. + +## The one detector today: `swift-vf-teardown` + +Applies only to SWIFT-v2 delegated-NIC nodes (label +`kubernetes.azure.com/podnetwork-swiftv2-enabled=true`, exported as +`SwiftV2LabelKey`/`SwiftV2LabelValue`): a node with no delegated secondary NIC +cannot suffer a VF teardown, so it is never a candidate. On those nodes it matches +`FailedCreatePodSandBox` Events in the `route ip+net: no such network interface` +family (plus the `network is unreachable`, `mtpnc is not ready`, and +`dhcp discover ... timed out` variants), with `failuresFloor: 3`, `window: 10m`, +`dwell: 10m`, `requireZeroSuccess: true`. + +`SwiftV2LabelKey`/`Value` are exported because the controller uses them to scope +its periodic sweep to the nodes a detector can actually fire on, rather than +sweeping every node in the cluster. The sweep is the union of that selector and +the wedged-label selector, so a node that stops being a detection candidate while +labeled is still swept and has its stale label retired +(`DecisionNotApplicable`), including when that transition happened while the +controller was down. + +## Adding a fault family + +1. If it fits the signature shape, add a new `signatureDetector` value in its own + file (e.g. `myfault.go`) with its constants, plus its `appliesTo` predicate. +2. If it needs different evidence (for example reading a CRD, not Events), add a + new type in its own file implementing the `Detector` interface directly. +3. Register it in `registry` in `decide.go`. +4. Add table cases to `decide_test.go`. + +`Decide` needs no changes in any case: it only knows the interface. diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/decide.go b/mgmt-agent/pkg/controller/nodehealth/detectors/decide.go new file mode 100644 index 00000000000..bc400a51cd6 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/decide.go @@ -0,0 +1,231 @@ +// 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 detectors holds the pure detection core of the node-health controller. +// It decides, from a node plus the Events and Pods currently held for it, +// whether the node is wedged, and is free of any Kubernetes I/O so it can be +// exhaustively table-tested. The controller package consumes Decide and acts on +// the returned Decision by labeling or unlabeling the node. +package detectors + +import ( + "fmt" + "sync" + "time" + + corev1 "k8s.io/api/core/v1" +) + +// Decision is the desired health state a reconcile derives for a node. It is the +// output of the pure Decide function and drives the label/unlabel action. +type Decision int + +const ( + // DecisionUnknown means the evidence is insufficient to make a call: leave + // the node's label exactly as it is. This covers a NotReady node (deferred to + // node lifecycle) and the cold view just after a controller restart, so an + // existing wedged label is retained until recovery is positively observed. + DecisionUnknown Decision = iota + // DecisionHealthy means recovery is confirmed (pods starting again): ensure + // the node is not labeled. + DecisionHealthy + // DecisionWedged means a detector fired: ensure the node is labeled. + DecisionWedged + // DecisionNotApplicable means no detector applies to the node at all, so no + // detector can ever mark it wedged: ensure the node is not labeled. This is + // a definitive statement about ownership, unlike DecisionUnknown which is a + // statement about evidence, and it is what retires a label left behind on a + // node that has since stopped being a detector's concern. + DecisionNotApplicable +) + +func (d Decision) String() string { + switch d { + case DecisionHealthy: + return "Healthy" + case DecisionWedged: + return "Wedged" + case DecisionNotApplicable: + return "NotApplicable" + default: + return "Unknown" + } +} + +// Detector is one fault family. Each concrete detector lives in its own file and +// contributes only its specifics (applicability, signals, thresholds); the +// evaluation primitives are shared through a common base (see signatureDetector). +// Decide iterates the registry and never depends on a concrete type, so adding a +// family is adding a Detector, not editing Decide. +type Detector interface { + // Name is a stable identifier used in logs, events, metrics, and the + // detector annotation. + Name() string + // Reason is a short human-readable explanation recorded on the node when the + // detector fires. + Reason() string + // Applies reports whether the detector is a candidate for the node. A node it + // does not apply to is never evaluated. + Applies(node *corev1.Node) bool + // Window is the detector's evaluation window. It is a fixed property of the + // detector, independent of any node, so callers that only need the window do + // not have to evaluate the detector to learn it. + Window() time.Duration + // Evaluate reads the detector's failure signals for the node from the Events + // and Pods currently held for it. The success signal is not read here: it is + // a recorded per-node history the controller passes into Decide, since a + // point-in-time scan cannot see a success whose Pod was already deleted. + Evaluate(events []*corev1.Event, pods []*corev1.Pod, now time.Time) Snapshot + // MeetsThreshold reports whether the evidence already gathered from the + // informers meets this detector's threshold. It is a pure predicate over that + // evidence, not an edge trigger: the reconcile is level-driven, so this is + // asked again on every pass from the current state. observedSince is when the + // controller began observing (caches synced, or the last disabled->enabled + // transition); the threshold is not met until a full window has been watched, + // so a cold view is never misread as zero successes. + MeetsThreshold(snap Snapshot, now, observedSince time.Time) bool +} + +// registry is the hard-coded set of detectors. A new fault family is a new +// Detector added here, reusing the shared primitives, shipped and tested as code. +var registry = []Detector{swiftVFTeardown} + +// AnyApplies reports whether any detector owns this node. It reads only the +// node, so a caller can answer the ownership question before doing the work of +// gathering the node's Pods and Events. Decide applies the same gate, so a node +// this rejects can only ever produce DecisionNotApplicable. +func AnyApplies(node *corev1.Node) bool { + if node == nil { + return false + } + for _, d := range registry { + if d.Applies(node) { + return true + } + } + return false +} + +// MaxWindow returns the longest evaluation window across all detectors. The +// controller uses it to bound its recorded per-node success history: a success +// older than the widest window can never affect any detector, so it can be +// pruned. The registry is hard-coded and the windows are fixed, so the result is +// computed once: the controller calls this on every reconcile while holding the +// observation lock. +var MaxWindow = sync.OnceValue(func() time.Duration { + var max time.Duration + for _, d := range registry { + if w := d.Window(); w > max { + max = w + } + } + return max +}) + +// Snapshot is the read-only evidence a detector evaluated for a node, retained +// for logging, the reason annotation, and the firing decision. Its fields are +// exported so the controller and labeler can render them. +type Snapshot struct { + // DetectorName is the name of the detector this snapshot belongs to. + DetectorName string + // Reason is the detector's human-readable explanation (set when it fires). + Reason string + // Window is the detector's evaluation window, for the reason string. + Window time.Duration + // FailureCount is the number of distinct pods on the node currently stuck + // without a sandbox (PodReadyToStartContainers=False) that are the subject of + // a matching failure Event in the window, regardless of how long each has been + // stuck. It is a live count of Pod state (not a windowed tally of Events), + // reported for observability. + FailureCount int + // SustainedCount is the subset of FailureCount whose pods have each + // individually been stuck for at least the dwell. It is the count the floor is + // tested against, so the floor means "this many pods each stuck past the + // dwell", not "this many stuck right now with only the oldest one aged". + SustainedCount int + // RecentSuccess reports whether the controller has a recorded per-node success + // (a non-host-network PodReadyToStartContainers=True transition) inside the + // window. It is set by Decide from the recorded lastSuccessAt, not from a scan + // of the pods passed in, so a success whose Pod was already deleted still + // counts. + RecentSuccess bool + // StuckSince is the oldest PodReadyToStartContainers=False lastTransitionTime + // among the node's currently-stuck failing pods, a GC-independent dwell signal + // read from durable Pod state, reported for observability. + StuckSince time.Time +} + +// ReasonString renders a short human-readable summary for the reason annotation. +func (s Snapshot) ReasonString() string { + success := "no recent success" + if s.RecentSuccess { + success = "a recent success" + } + return fmt.Sprintf("%d pods stuck past dwell (%d stuck total), %s in %s window", + s.SustainedCount, s.FailureCount, success, s.Window) +} + +// Decide is the pure core of the controller: given a node, the Events and Pods +// currently held for it, a clock, the time the controller began observing, and +// the node's recorded last success time, it returns the desired health state. It +// performs no I/O and is exhaustively table-tested. +func Decide(node *corev1.Node, events []*corev1.Event, pods []*corev1.Pod, now, observedSince, lastSuccessAt time.Time) (Decision, Snapshot) { + if node == nil { + return DecisionUnknown, Snapshot{} + } + // Ownership precondition, checked before readiness: if no detector applies, + // none can ever fire for this node, so any label we left on it is stale and + // must be retired. This is deliberately evaluated ahead of the Ready gate, + // because a node that is not a detector's concern is not ours to hold a label + // on whether it is Ready or not. + if !AnyApplies(node) { + return DecisionNotApplicable, Snapshot{} + } + // Node-Ready precondition: a NotReady node is left to node lifecycle. + if !isNodeReady(node) { + return DecisionUnknown, Snapshot{} + } + + sawSuccess := false + for _, d := range registry { + if !d.Applies(node) { + continue + } + snap := d.Evaluate(events, pods, now) + // A recorded success counts when it falls inside this detector's window. + // The distance is measured in absolute terms: the timestamp comes from a + // Pod condition stamped by the kubelet's clock, so skew can place it in the + // future, and a signed comparison would read any future timestamp as an + // arbitrarily fresh success and suppress detection until the wall clock + // caught up. Skew inside the window is benign, beyond it the timestamp is + // not usable evidence either way. + if !lastSuccessAt.IsZero() && now.Sub(lastSuccessAt).Abs() < snap.Window { + snap.RecentSuccess = true + sawSuccess = true + } + if d.MeetsThreshold(snap, now, observedSince) { + snap.Reason = d.Reason() + return DecisionWedged, snap + } + } + + // No detector fired. Only declare recovery on positive evidence (a recorded + // success in the window). An empty/insufficient view stays Unknown so an + // existing wedged label is retained across a restart until recovery is + // observed. + if sawSuccess { + return DecisionHealthy, Snapshot{} + } + return DecisionUnknown, Snapshot{} +} diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/decide_test.go b/mgmt-agent/pkg/controller/nodehealth/detectors/decide_test.go new file mode 100644 index 00000000000..ec9ce481dcb --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/decide_test.go @@ -0,0 +1,482 @@ +// 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 detectors + +import ( + "fmt" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +var testNow = time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + +func ago(d time.Duration) time.Time { return testNow.Add(-d) } + +// podUID is a deterministic UID for a pod name, so a failure Event and its stuck +// Pod correlate by UID the way they do in a live cluster. +func podUID(name string) types.UID { return types.UID("uid-" + name) } + +// warm is an observedSince far enough in the past that a full detector window has +// elapsed, so the success column is trusted. cold is recent, so the controller +// has not yet watched a full window and a would-be wedge stays indeterminate. +var ( + warm = ago(30 * time.Minute) + cold = ago(1 * time.Minute) +) + +const sig = "route ip+net: no such network interface" + +const nodeName = "aks-userswft2-0" + +func testNode(swift, ready bool) *corev1.Node { + labels := map[string]string{} + if swift { + labels[SwiftV2LabelKey] = SwiftV2LabelValue + } + status := corev1.ConditionFalse + if ready { + status = corev1.ConditionTrue + } + return &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{Name: nodeName, Labels: labels}, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: status}}, + }, + } +} + +// failEventFor is a FailedCreatePodSandBox Event whose kubelet Source.Host is the +// node and whose involved object is the named pod (correlated by UID), matching +// the SWIFT signature. +func failEventFor(pod string, last time.Time, msg string) *corev1.Event { + return &corev1.Event{ + InvolvedObject: corev1.ObjectReference{Kind: "Pod", Namespace: "ns", Name: pod, UID: podUID(pod)}, + Source: corev1.EventSource{Host: nodeName, Component: "kubelet"}, + Reason: reasonFailedCreatePodSandBox, + Message: msg, + LastTimestamp: metav1.NewTime(last), + FirstTimestamp: metav1.NewTime(last), + } +} + +// stuckPod is a pending pod whose PodReadyToStartContainers condition is False, +// the durable per-pod symptom of a sandbox that never came up. since is the +// condition's lastTransitionTime, which supplies the dwell. +func stuckPod(pod string, since time.Time) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: pod, UID: podUID(pod)}, + Spec: corev1.PodSpec{NodeName: nodeName}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionFalse, + LastTransitionTime: metav1.NewTime(since), + }}, + }, + } +} + +// startedPod is a pod whose PodReadyToStartContainers condition transitioned to +// True at ts: a fresh sandbox, the success signal SuccessAt records. hostNet +// marks it host-network, which must be excluded since it reaches the condition +// without a delegated NIC. +func startedPod(pod string, ts time.Time, hostNet bool) *corev1.Pod { + return &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: pod}, + Spec: corev1.PodSpec{NodeName: nodeName, HostNetwork: hostNet}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionTrue, + LastTransitionTime: metav1.NewTime(ts), + }}, + }, + } +} + +// stuckFailing builds n distinct stuck pods, each with a matching in-window +// failure Event, all entering the stuck state at since. +func stuckFailing(n int, since time.Time) ([]*corev1.Event, []*corev1.Pod) { + var events []*corev1.Event + var pods []*corev1.Pod + for i := 0; i < n; i++ { + name := fmt.Sprintf("p%d", i) + pods = append(pods, stuckPod(name, since)) + events = append(events, failEventFor(name, ago(20*time.Second), sig)) + } + return events, pods +} + +func TestDecide(t *testing.T) { + tests := []struct { + name string + node *corev1.Node + events []*corev1.Event + pods []*corev1.Pod + observedSince time.Time + lastSuccessAt time.Time + want Decision + }{ + { + name: "wedged: floor of stuck failing pods, zero success, dwell met", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(3, ago(15*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(3, ago(15*time.Minute)) + return p + }(), + want: DecisionWedged, + }, + { + name: "flap: a recorded success in the window is not a wedge", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(3, ago(15*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(3, ago(15*time.Minute)) + return p + }(), + lastSuccessAt: ago(2 * time.Minute), + want: DecisionHealthy, + }, + { + name: "below the stuck-pod floor is not wedged", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(2, ago(15*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(2, ago(15*time.Minute)) + return p + }(), + want: DecisionUnknown, + }, + { + name: "non-SWIFT node is never a candidate, so any label on it is stale", + node: testNode(false, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(5, ago(30*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(5, ago(30*time.Minute)) + return p + }(), + want: DecisionNotApplicable, + }, + { + name: "non-SWIFT NotReady node is still not a candidate", + node: testNode(false, false), + want: DecisionNotApplicable, + }, + { + name: "NotReady node defers to node lifecycle", + node: testNode(true, false), + events: func() []*corev1.Event { + e, _ := stuckFailing(5, ago(30*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(5, ago(30*time.Minute)) + return p + }(), + want: DecisionUnknown, + }, + { + name: "dwell not yet met: pods stuck too recently", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(3, ago(2*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(3, ago(2*time.Minute)) + return p + }(), + want: DecisionUnknown, + }, + { + name: "floor is per-pod dwell: one old plus two brand-new stuck pods does not fire", + node: testNode(true, true), + events: []*corev1.Event{ + failEventFor("old", ago(20*time.Second), sig), + failEventFor("new1", ago(20*time.Second), sig), + failEventFor("new2", ago(20*time.Second), sig), + }, + pods: []*corev1.Pod{ + stuckPod("old", ago(15*time.Minute)), + stuckPod("new1", ago(20*time.Second)), + stuckPod("new2", ago(20*time.Second)), + }, + want: DecisionUnknown, + }, + { + name: "recovery: a recorded success, no stuck pods", + node: testNode(true, true), + lastSuccessAt: ago(1 * time.Minute), + want: DecisionHealthy, + }, + { + name: "cold view: no signals leaves the label unchanged", + node: testNode(true, true), + want: DecisionUnknown, + }, + { + name: "non-matching signature is not a failing pod", + node: testNode(true, true), + events: []*corev1.Event{ + failEventFor("p0", ago(20*time.Second), "image pull backoff"), + failEventFor("p1", ago(20*time.Second), "image pull backoff"), + failEventFor("p2", ago(20*time.Second), "image pull backoff"), + }, + pods: []*corev1.Pod{ + stuckPod("p0", ago(15*time.Minute)), + stuckPod("p1", ago(15*time.Minute)), + stuckPod("p2", ago(15*time.Minute)), + }, + want: DecisionUnknown, + }, + { + name: "stuck pod without a matching Event is not counted", + node: testNode(true, true), + pods: []*corev1.Pod{ + stuckPod("p0", ago(15*time.Minute)), + stuckPod("p1", ago(15*time.Minute)), + stuckPod("p2", ago(15*time.Minute)), + }, + want: DecisionUnknown, + }, + { + name: "stale storm: failure Events aged out of the window", + node: testNode(true, true), + events: []*corev1.Event{ + failEventFor("p0", ago(20*time.Minute), sig), + failEventFor("p1", ago(20*time.Minute), sig), + failEventFor("p2", ago(20*time.Minute), sig), + }, + pods: []*corev1.Pod{ + stuckPod("p0", ago(20*time.Minute)), + stuckPod("p1", ago(20*time.Minute)), + stuckPod("p2", ago(20*time.Minute)), + }, + want: DecisionUnknown, + }, + { + name: "cold controller: full window not yet observed does not wedge", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(3, ago(15*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(3, ago(15*time.Minute)) + return p + }(), + observedSince: cold, + want: DecisionUnknown, + }, + { + name: "recorded success not set: a host-network pod in the slice is ignored, node stays wedged", + node: testNode(true, true), + events: func() []*corev1.Event { + e, _ := stuckFailing(3, ago(15*time.Minute)) + return e + }(), + pods: func() []*corev1.Pod { + _, p := stuckFailing(3, ago(15*time.Minute)) + return append(p, startedPod("hostnet", ago(2*time.Minute), true)) + }(), + want: DecisionWedged, + }, + { + name: "gate off: PodReadyToStartContainers absent yields Unknown", + node: testNode(true, true), + events: []*corev1.Event{ + failEventFor("p0", ago(20*time.Second), sig), + failEventFor("p1", ago(20*time.Second), sig), + failEventFor("p2", ago(20*time.Second), sig), + }, + pods: []*corev1.Pod{ + {ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "p0"}, Spec: corev1.PodSpec{NodeName: nodeName}, Status: corev1.PodStatus{Phase: corev1.PodPending}}, + {ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "p1"}, Spec: corev1.PodSpec{NodeName: nodeName}, Status: corev1.PodStatus{Phase: corev1.PodPending}}, + {ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "p2"}, Spec: corev1.PodSpec{NodeName: nodeName}, Status: corev1.PodStatus{Phase: corev1.PodPending}}, + }, + want: DecisionUnknown, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + obs := tc.observedSince + if obs.IsZero() { + obs = warm + } + got, _ := Decide(tc.node, tc.events, tc.pods, testNow, obs, tc.lastSuccessAt) + if got != tc.want { + t.Errorf("Decide() = %v, want %v", got, tc.want) + } + }) + } +} + +func TestAnyApplies(t *testing.T) { + // AnyApplies is the gate the controller uses to skip gathering a node's Pods + // and Events, so it must agree with Decide's own ownership gate. + tests := []struct { + name string + node *corev1.Node + want bool + }{ + {name: "nil node", node: nil, want: false}, + {name: "swift node", node: testNode(true, true), want: true}, + {name: "swift node that is not ready", node: testNode(true, false), want: true}, + {name: "non-swift node", node: testNode(false, true), want: false}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := AnyApplies(tc.node); got != tc.want { + t.Errorf("AnyApplies() = %v, want %v", got, tc.want) + } + if !tc.want && tc.node != nil { + if got, _ := Decide(tc.node, nil, nil, testNow, warm, time.Time{}); got != DecisionNotApplicable { + t.Errorf("Decide() = %v, want NotApplicable to match AnyApplies", got) + } + } + }) + } +} + +func TestDecideNilNode(t *testing.T) { + if got, _ := Decide(nil, nil, nil, testNow, warm, time.Time{}); got != DecisionUnknown { + t.Errorf("Decide(nil) = %v, want Unknown", got) + } +} + +func TestDecideSnapshotOnWedge(t *testing.T) { + node := testNode(true, true) + events, pods := stuckFailing(3, ago(15*time.Minute)) + + got, snap := Decide(node, events, pods, testNow, warm, time.Time{}) + if got != DecisionWedged { + t.Fatalf("Decide() = %v, want Wedged", got) + } + if snap.DetectorName != swiftVFTeardown.name { + t.Errorf("snapshot detector = %q, want %q", snap.DetectorName, swiftVFTeardown.name) + } + if snap.FailureCount != 3 { + t.Errorf("snapshot FailureCount = %d, want 3", snap.FailureCount) + } + if snap.SustainedCount != 3 { + t.Errorf("snapshot SustainedCount = %d, want 3", snap.SustainedCount) + } + if snap.RecentSuccess { + t.Errorf("snapshot RecentSuccess = %v, want false", snap.RecentSuccess) + } + if !snap.StuckSince.Equal(ago(15 * time.Minute)) { + t.Errorf("snapshot StuckSince = %v, want %v", snap.StuckSince, ago(15*time.Minute)) + } + if snap.Reason == "" { + t.Error("snapshot reason should be set on a wedge") + } +} + +func TestDwellReadFromOldestStuckPod(t *testing.T) { + // Failure Events are fresh, but the oldest stuck pod condition is well past the + // dwell, so the wedge fires on the durable Pod timestamp, not the Event age. + node := testNode(true, true) + events, pods := stuckFailing(3, ago(20*time.Minute)) + got, _ := Decide(node, events, pods, testNow, warm, time.Time{}) + if got != DecisionWedged { + t.Errorf("Decide() = %v, want Wedged", got) + } +} + +func TestEventsCorrelateByUIDNotName(t *testing.T) { + // A wedge's failure Events belong to Pods that were later deleted; new Pods + // reuse the same namespace/name but are distinct objects with fresh UIDs. The + // old Events must not be attributed to the new Pods, so name reuse cannot + // manufacture a wedge. + node := testNode(true, true) + events, pods := stuckFailing(3, ago(20*time.Minute)) + for _, p := range pods { + p.UID = types.UID(string(p.UID) + "-reused") + } + got, _ := Decide(node, events, pods, testNow, warm, time.Time{}) + if got != DecisionUnknown { + t.Errorf("Decide() = %v, want Unknown (name reuse must not inherit old Events)", got) + } +} + +func TestSignatureVariantsMatch(t *testing.T) { + msgs := []string{ + "failed to setup network for sandbox: route ip+net: no such network interface", + "plugin failed: network is unreachable", + "mtpnc is not ready for pod", + "dhcp discover for eth0 timed out after 15s", + } + for _, m := range msgs { + if !swiftVFTeardown.matches(m) { + t.Errorf("expected signature match for %q", m) + } + } + if swiftVFTeardown.matches("image pull backoff") { + t.Error("unrelated message should not match") + } +} + +func TestSuccessAt(t *testing.T) { + ts := ago(2 * time.Minute) + + // A non-host-network pod with PodReadyToStartContainers=True yields its + // transition time: the durable per-node success timestamp the controller + // records. + if at, ok := SuccessAt(startedPod("ok", ts, false)); !ok || !at.Equal(ts) { + t.Errorf("SuccessAt(started) = (%v, %v), want (%v, true)", at, ok, ts) + } + + // A host-network pod reaches the condition without a delegated NIC, so it is + // never a SWIFT success. + if _, ok := SuccessAt(startedPod("hostnet", ts, true)); ok { + t.Error("SuccessAt(host-network) = true, want false") + } + + // A stuck pod (condition False) is not a success. + if _, ok := SuccessAt(stuckPod("p0", ts)); ok { + t.Error("SuccessAt(stuck) = true, want false") + } + + // A pod without the condition (feature gate off) is not a success. + bare := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "bare"}, Spec: corev1.PodSpec{NodeName: nodeName}} + if _, ok := SuccessAt(bare); ok { + t.Error("SuccessAt(no condition) = true, want false") + } + + // A nil pod is safe. + if _, ok := SuccessAt(nil); ok { + t.Error("SuccessAt(nil) = true, want false") + } +} diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/production_evidence_test.go b/mgmt-agent/pkg/controller/nodehealth/detectors/production_evidence_test.go new file mode 100644 index 00000000000..bb7fe8e45ea --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/production_evidence_test.go @@ -0,0 +1,258 @@ +// 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 detectors + +import ( + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" +) + +// This file pins the detector against evidence captured from a real wedged +// production node during the uksouth SWIFT incident (AROSLSRE-1585). The failure +// strings and the node's label and readiness shape are taken verbatim from that +// capture, so a future edit to the signatures or the applicability label that +// would have missed the real incident fails here. Node names, namespaces and +// sandbox IDs are redacted to example values, since no detector logic depends on +// them. + +// productionSandboxMessages are verbatim kubelet FailedCreatePodSandBox messages +// from the captured incident, one per signature family the detector matches. +var productionSandboxMessages = []struct { + family string + message string +}{ + { + family: "route/no-such-network-interface", + message: `Failed to create pod sandbox: rpc error: code = Unknown desc = failed to setup network for sandbox "0000000000000000000000000000000000000000000000000000000000000001": plugin type="azure-vnet" failed (add): failed to create endpoint: SecondaryEndpointClient Error: route ip+net: no such network interface`, + }, + { + family: "network-unreachable", + message: `Failed to create pod sandbox: rpc error: code = Unknown desc = failed to setup network for sandbox "0000000000000000000000000000000000000000000000000000000000000002": plugin type="azure-vnet" failed (add): failed to create endpoint: network is unreachable`, + }, + { + family: "mtpnc-not-ready", + message: `Failed to create pod sandbox: rpc error: code = Unknown desc = failed to setup network for sandbox "0000000000000000000000000000000000000000000000000000000000000004": plugin type="azure-vnet" failed (add): failed to add ipam invoker: Failed to get IP address from CNS: network is not ready - mtpnc is not ready`, + }, + { + family: "dhcp-discover-timeout", + message: `Failed to create pod sandbox: rpc error: code = Unknown desc = failed to setup network for sandbox "0000000000000000000000000000000000000000000000000000000000000003": plugin type="azure-vnet" failed (add): failed to create endpoint: network is not ready - failed to issue dhcp discover packet to create mapping in host: timed out waiting for replies`, + }, +} + +func TestProductionSandboxMessagesMatchSignatures(t *testing.T) { + for _, tc := range productionSandboxMessages { + t.Run(tc.family, func(t *testing.T) { + if !swiftVFTeardown.matches(tc.message) { + t.Errorf("captured production message was not matched by any signature:\n%s", tc.message) + } + }) + } +} + +// productionNodeLabels are the SWIFT-related labels carried by the captured +// wedged node. The node was Ready throughout, which is the premise the whole +// controller rests on: node lifecycle never sees these nodes as broken. +func productionWedgedNode() *corev1.Node { + return &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "aks-userswift0-00000000-vmss000000", + Labels: map[string]string{ + "kubernetes.azure.com/podnetwork-swiftv2-enabled": "true", + "kubernetes.azure.com/podnetwork-multi-tenancy-enabled": "true", + "kubernetes.azure.com/podnetwork-type": "vnet", + "kubernetes.azure.com/podnetwork-name": "aks-net", + }, + }, + Status: corev1.NodeStatus{ + Conditions: []corev1.NodeCondition{{Type: corev1.NodeReady, Status: corev1.ConditionTrue}}, + }, + } +} + +func TestProductionNodeIsADetectionCandidate(t *testing.T) { + node := productionWedgedNode() + if !swiftVFTeardown.Applies(node) { + t.Fatal("the captured wedged production node must be a detection candidate") + } + if !isNodeReady(node) { + t.Fatal("the captured node was Ready while wedged; the Ready premise must hold") + } +} + +// TestProductionCompletedPodsDoNotCount replays the pod population found on the +// captured node. Alongside the wedge, the node carried four completed Job and +// CronJob pods (featuregate-generator and olm-collect-profiles), all in phase +// Succeeded with PodReadyToStartContainers=False because their sandboxes were +// torn down when they finished, days earlier. +// +// Those pods are ordinary cluster turnover, not evidence of a wedge. If they +// counted toward the floor, routine CronJob churn on any SWIFT node would drift +// the detector toward firing on its own. +func TestProductionCompletedPodsDoNotCount(t *testing.T) { + now := time.Date(2026, 7, 27, 9, 0, 43, 0, time.UTC) + ns := "ocm-example-hosted-cluster" + + completed := []struct { + name string + at time.Time + }{ + {"featuregate-generator-655jv", time.Date(2026, 7, 22, 11, 27, 34, 0, time.UTC)}, + {"olm-collect-profiles-29749045-588fv", time.Date(2026, 7, 25, 1, 25, 4, 0, time.UTC)}, + {"olm-collect-profiles-29750485-zkpm5", time.Date(2026, 7, 26, 1, 25, 4, 0, time.UTC)}, + {"olm-collect-profiles-29751925-8hhjp", time.Date(2026, 7, 27, 1, 25, 36, 0, time.UTC)}, + } + + var pods []*corev1.Pod + var events []*corev1.Event + for _, c := range completed { + uid := types.UID("uid-" + c.name) + pods = append(pods, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: c.name, UID: uid}, + Status: corev1.PodStatus{ + Phase: corev1.PodSucceeded, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionFalse, + LastTransitionTime: metav1.NewTime(c.at), + }}, + }, + }) + // Worst case: each completed pod also has a correlating in-window sandbox + // failure Event, so only the terminal phase can rule it out. + events = append(events, &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: c.name + ".evt"}, + Reason: reasonFailedCreatePodSandBox, + Message: productionSandboxMessages[0].message, + InvolvedObject: corev1.ObjectReference{Kind: "Pod", Namespace: ns, Name: c.name, UID: uid}, + Source: corev1.EventSource{Host: "aks-userswift0-00000000-vmss000000"}, + LastTimestamp: metav1.NewTime(now.Add(-1 * time.Minute)), + }) + } + + snap := swiftVFTeardown.Evaluate(events, pods, now) + if snap.SustainedCount != 0 { + t.Errorf("completed pods counted toward the floor: SustainedCount = %d, want 0", snap.SustainedCount) + } + if snap.FailureCount != 0 { + t.Errorf("completed pods counted as failures: FailureCount = %d, want 0", snap.FailureCount) + } + + // The node must not be declared wedged on completed-pod turnover alone. + observedSince := now.Add(-1 * time.Hour) + if got, _ := Decide(productionWedgedNode(), events, pods, now, observedSince, time.Time{}); got == DecisionWedged { + t.Error("completed Job and CronJob turnover must not produce a wedge") + } +} + +// productionWedgeEvidence builds the failing side of the captured incident: +// pending pods whose sandbox never came up, each stuck well past the dwell, with +// a correlating in-window kubelet Event. +func productionWedgeEvidence(now time.Time) ([]*corev1.Event, []*corev1.Pod) { + const ns = "openshift-monitoring" + const host = "aks-userswift0-00000000-vmss000000" + + var pods []*corev1.Pod + var events []*corev1.Event + for i, name := range []string{"ovnkube-node-x1", "prometheus-k8s-0", "router-default-abc"} { + uid := types.UID("uid-" + name) + pods = append(pods, &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name, UID: uid}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + Conditions: []corev1.PodCondition{{ + Type: corev1.PodReadyToStartContainers, + Status: corev1.ConditionFalse, + LastTransitionTime: metav1.NewTime(now.Add(-30 * time.Minute)), + }}, + }, + }) + events = append(events, &corev1.Event{ + ObjectMeta: metav1.ObjectMeta{Namespace: ns, Name: name + ".evt"}, + Reason: reasonFailedCreatePodSandBox, + Message: productionSandboxMessages[i%len(productionSandboxMessages)].message, + InvolvedObject: corev1.ObjectReference{Kind: "Pod", Namespace: ns, Name: name, UID: uid}, + Source: corev1.EventSource{Host: host}, + LastTimestamp: metav1.NewTime(now.Add(-1 * time.Minute)), + }) + } + return events, pods +} + +// TestProductionStaleSuccessesDoNotSuppressDetection guards the success signal's +// window scoping, which the captured incident shows is load-bearing. +// +// The wedged node was still running 46 non-host-network pods whose +// PodReadyToStartContainers condition read True: they were started long before +// the VF disappeared and kept running afterwards, because an established sandbox +// survives the teardown. The freshest of those transitions was over 8 hours old. +// +// So "has this node ever started a pod successfully" is always true on a +// long-lived node and would suppress detection forever. Only a success inside +// the detector window is evidence the node can still attach a NIC right now. +func TestProductionStaleSuccessesDoNotSuppressDetection(t *testing.T) { + now := time.Date(2026, 7, 27, 9, 0, 43, 0, time.UTC) + node := productionWedgedNode() + events, pods := productionWedgeEvidence(now) + observedSince := now.Add(-1 * time.Hour) + + // Freshest success actually observed on the captured node: 500.6 minutes old, + // far outside the 10 minute window. + staleSuccess := now.Add(-500*time.Minute - 36*time.Second) + got, snap := Decide(node, events, pods, now, observedSince, staleSuccess) + if got != DecisionWedged { + t.Fatalf("stale success suppressed detection of the real wedge: Decide = %v (%s)", got, snap.ReasonString()) + } + if snap.RecentSuccess { + t.Error("an 8 hour old success must not count as recent") + } + + // A success inside the window is real proof the NIC still attaches, so the + // same evidence must no longer read as a hard wedge. Decide reports recovery + // as Healthy and returns an empty snapshot on that path, so the decision is + // the assertion here. + freshSuccess := now.Add(-2 * time.Minute) + if got, _ = Decide(node, events, pods, now, observedSince, freshSuccess); got != DecisionHealthy { + t.Errorf("a success inside the window must rule out a hard wedge: Decide = %v, want %v", got, DecisionHealthy) + } +} + +// TestFutureDatedSuccessDoesNotSuppressDetection covers clock skew between the +// kubelet that stamps a pod condition and the controller that reads it. +// +// The success timestamp comes from a Pod condition's lastTransitionTime, written +// by the kubelet using its own node's clock. If that clock runs ahead, the +// controller records a success dated in the future. A naive "how long ago was +// it" comparison reads a future timestamp as an arbitrarily small age, so it +// always looks like a fresh success and permanently suppresses wedge detection. +// +// This is the same failure mode as a stale success suppressing detection, but it +// does not age out on its own, so it is worth pinning separately. +func TestFutureDatedSuccessDoesNotSuppressDetection(t *testing.T) { + now := time.Date(2026, 7, 27, 9, 0, 43, 0, time.UTC) + node := productionWedgedNode() + events, pods := productionWedgeEvidence(now) + observedSince := now.Add(-1 * time.Hour) + + // A success stamped an hour into the future by a skewed kubelet clock. It is + // not evidence the node can attach a NIC now, so it must not rule out a wedge. + future := now.Add(1 * time.Hour) + if got, _ := Decide(node, events, pods, now, observedSince, future); got != DecisionWedged { + t.Errorf("a future-dated success suppressed detection of a real wedge: Decide = %v, want %v", got, DecisionWedged) + } +} diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/signature_detector.go b/mgmt-agent/pkg/controller/nodehealth/detectors/signature_detector.go new file mode 100644 index 00000000000..71c6c87e56f --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/signature_detector.go @@ -0,0 +1,247 @@ +// 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 detectors + +import ( + "regexp" + "time" + + corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" +) + +// signatureDetector is the shared Detector base for fault families whose signal +// is an Event-reason storm matched by regex signatures, gated by a sustained +// floor, a dwell, and the load-bearing zero-successful-start check. A concrete +// family (see swift_vf.go) is a signatureDetector value with its own constants; +// it reuses every method below rather than duplicating the evaluation logic. +type signatureDetector struct { + // name is a stable identifier used in logs, events, metrics, and the + // detector annotation. + name string + // reason is a short human-readable explanation recorded on the node when the + // detector fires. + reason string + // appliesTo limits the detector to nodes that can physically exhibit the + // fault. A node it does not apply to is never a candidate. A nil predicate + // means the detector applies to every node. + appliesTo func(*corev1.Node) bool + // eventReason is the kubelet Event reason that carries the failure signal. + eventReason string + // signatures match the failure Event message; any match is a hit. + signatures []*regexp.Regexp + // failuresFloor is the minimum number of distinct stuck failing pods on the + // node required to fire. It is a floor to require the storm to be sustained, + // not the trigger; the wedge/flap call is made by the zero-success rule. + failuresFloor int + // window is the rolling window over which failures and successes are counted. + window time.Duration + // dwell is how long the failure must have been sustained before the detector + // fires, filtering transient bursts. + dwell time.Duration + // requireZeroSuccess makes firing require zero fresh pod-sandbox successes in + // the window (no PodReadyToStartContainers=True transition). This is the + // load-bearing discriminator between a hard-wedge (zero successes, VF gone) + // and a flap (some successes, VF present). + requireZeroSuccess bool +} + +// Name returns the detector's stable identifier. +func (d signatureDetector) Name() string { return d.name } + +// Reason returns the detector's human-readable explanation. +func (d signatureDetector) Reason() string { return d.reason } + +// Window is the detector's fixed evaluation window. +func (d signatureDetector) Window() time.Duration { return d.window } + +// Applies reports whether the detector is a candidate for the node. A nil +// predicate applies everywhere. +func (d signatureDetector) Applies(node *corev1.Node) bool { + return d.appliesTo == nil || d.appliesTo(node) +} + +// Evaluate reads the failure signal for this detector from the node's Events and +// Pods within the detector window. Every number it returns comes from live Pod +// state; Events only classify which pods are failing. The success signal is not +// read here (see Decide and the controller's recorded success history). +func (d signatureDetector) Evaluate(events []*corev1.Event, pods []*corev1.Pod, now time.Time) Snapshot { + windowStart := now.Add(-d.window) + snap := Snapshot{DetectorName: d.name, Window: d.window} + + // A pod is "failing" for this detector when it is the subject of a matching + // failure Event seen within the window. Correlation is by the Event's + // InvolvedObject.UID, not namespace/name: a UID is unique to a single Pod + // object, so a recycled name (a new Pod reusing a deleted Pod's name) cannot + // inherit the old Pod's failure Events. Events whose involved object carries + // no UID are skipped rather than matched loosely. The Events are used only to + // identify failing pods; they are never counted. + failing := make(map[types.UID]bool) + for _, ev := range events { + if ev == nil || ev.InvolvedObject.Kind != "Pod" || ev.InvolvedObject.UID == "" { + continue + } + if ev.Reason != d.eventReason || !d.matches(ev.Message) { + continue + } + if eventLastTime(ev).Before(windowStart) { + continue + } + failing[ev.InvolvedObject.UID] = true + } + + for _, p := range pods { + if p == nil || p.DeletionTimestamp != nil { + continue + } + // Floor and dwell are per pod. A pod counts toward the floor only once it + // has individually been stuck without a sandbox + // (PodReadyToStartContainers=False) for at least the dwell, so the floor is + // a count of pods each sustained past the dwell, not a snapshot of pods + // stuck right now with only the oldest one aged. This is what stops a + // single old failure alongside brand-new ones from firing, and what stops + // one long-stuck pod from firing alone. + if failing[p.UID] { + if since, stuck := stuckSince(p); stuck { + snap.FailureCount++ + if snap.StuckSince.IsZero() || since.Before(snap.StuckSince) { + snap.StuckSince = since + } + if now.Sub(since) >= d.dwell { + snap.SustainedCount++ + } + } + } + } + + return snap +} + +// MeetsThreshold reports whether the threshold is met: the floor of pods that +// have each individually been stuck for at least the dwell, with no recorded +// success in the window (when required and observable). A cold view that has not +// yet watched a full window can never meet the threshold, so an unobserved +// success signal is treated as indeterminate, never as zero. +func (d signatureDetector) MeetsThreshold(snap Snapshot, now, observedSince time.Time) bool { + // The floor counts pods each sustained past the dwell (computed in Evaluate), + // so meeting it already proves the storm held continuously; there is no + // separate oldest-pod dwell check. + if snap.SustainedCount < d.failuresFloor { + return false + } + if d.requireZeroSuccess { + // The success signal is only trustworthy once a full window has been + // observed since the controller began observing; before that, treat it as + // indeterminate and do not fire. + if observedSince.IsZero() || now.Sub(observedSince) < d.window { + return false + } + if snap.RecentSuccess { + return false + } + } + return true +} + +// matches reports whether the Event message matches any of the detector's +// signature regexes. +func (d signatureDetector) matches(message string) bool { + for _, re := range d.signatures { + if re.MatchString(message) { + return true + } + } + return false +} + +func isNodeReady(node *corev1.Node) bool { + for _, c := range node.Status.Conditions { + if c.Type == corev1.NodeReady { + return c.Status == corev1.ConditionTrue + } + } + return false +} + +// podReadyToStartCondition returns the pod's PodReadyToStartContainers condition, +// or nil if it is absent (for example when the feature gate is off). +func podReadyToStartCondition(p *corev1.Pod) *corev1.PodCondition { + for i := range p.Status.Conditions { + if p.Status.Conditions[i].Type == corev1.PodReadyToStartContainers { + return &p.Status.Conditions[i] + } + } + return nil +} + +// stuckSince reports whether a pod is currently stuck without a sandbox +// (PodReadyToStartContainers=False) and, if so, when it entered that state. The +// timestamp is the condition's lastTransitionTime, durable across Event GC and a +// controller restart. A pod whose condition is absent is not counted as stuck, so +// when the feature gate is off the detector reads Unknown rather than wedging. +func stuckSince(p *corev1.Pod) (time.Time, bool) { + // A pod that reached a terminal phase already ran, so its sandbox was torn + // down on completion and PodReadyToStartContainers goes False as a matter of + // course. That is a finished pod, not a pod that cannot start, and counting + // it would inflate the stuck count with ordinary Job and CronJob turnover. + // Real wedged nodes carry several of these at any time. + if p.Status.Phase == corev1.PodSucceeded || p.Status.Phase == corev1.PodFailed { + return time.Time{}, false + } + cond := podReadyToStartCondition(p) + if cond == nil || cond.Status != corev1.ConditionFalse { + return time.Time{}, false + } + return cond.LastTransitionTime.Time, true +} + +// SuccessAt reports whether a non-host-network pod currently shows a fresh +// sandbox (PodReadyToStartContainers=True) and, if so, when that condition last +// transitioned to True: proof the pod's sandbox and its SWIFT network were +// created, which a hard-wedge cannot do. Host-network pods are excluded, since +// they reach the condition without a delegated NIC. Keying on the transition, not +// pod age, means a container restart inside an existing sandbox is not counted as +// success. The controller uses this to record a durable per-node lastSuccessAt, +// so a success survives even after its short-lived pod is deleted. +func SuccessAt(p *corev1.Pod) (time.Time, bool) { + if p == nil || p.Spec.HostNetwork { + return time.Time{}, false + } + cond := podReadyToStartCondition(p) + if cond == nil || cond.Status != corev1.ConditionTrue { + return time.Time{}, false + } + return cond.LastTransitionTime.Time, true +} + +// eventLastTime returns the latest activity time of a (possibly aggregated) +// Event: its lastTimestamp, falling back to eventTime then firstTimestamp. +func eventLastTime(ev *corev1.Event) time.Time { + if !ev.LastTimestamp.IsZero() { + return ev.LastTimestamp.Time + } + if ev.EventTime.Time != (time.Time{}) && !ev.EventTime.IsZero() { + return ev.EventTime.Time + } + return ev.FirstTimestamp.Time +} + +func mustCompileSignatures(patterns ...string) []*regexp.Regexp { + out := make([]*regexp.Regexp, 0, len(patterns)) + for _, p := range patterns { + out = append(out, regexp.MustCompile(p)) + } + return out +} diff --git a/mgmt-agent/pkg/controller/nodehealth/detectors/swift_vf.go b/mgmt-agent/pkg/controller/nodehealth/detectors/swift_vf.go new file mode 100644 index 00000000000..7c2a175aac6 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/detectors/swift_vf.go @@ -0,0 +1,63 @@ +// 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 detectors + +import ( + "time" + + corev1 "k8s.io/api/core/v1" +) + +const ( + // SwiftV2LabelKey is the AKS-managed label marking a node as SWIFT-v2 + // delegated-NIC capable. A node without a delegated secondary NIC cannot + // suffer a VF teardown, so this label scopes the swift-vf-teardown detector. + // It is exported because the controller uses it to scope its node informer to + // the only nodes any detector can apply to today. + SwiftV2LabelKey = "kubernetes.azure.com/podnetwork-swiftv2-enabled" + // SwiftV2LabelValue is the value SwiftV2LabelKey carries on a SWIFT-v2 node. + SwiftV2LabelValue = "true" + + // reasonFailedCreatePodSandBox is the kubelet Event reason emitted when a pod + // sandbox cannot be created. It is the failure signal for the SWIFT wedge. + reasonFailedCreatePodSandBox = "FailedCreatePodSandBox" +) + +// swiftVFTeardown detects the SWIFT v2 delegated-NIC teardown wedge: on a +// SWIFT-v2 node, a sustained FailedCreatePodSandBox storm in the route / +// network-unreachable / mtpnc / dhcp-timeout family with zero successful pod +// starts across the window. Signatures, thresholds, and the applicability label +// are constants, not config. It carries only the specifics; every evaluation +// primitive comes from signatureDetector (see signature_detector.go). +var swiftVFTeardown = signatureDetector{ + name: "swift-vf-teardown", + reason: "SWIFT delegated-NIC teardown: sustained FailedCreatePodSandBox storm with zero successful pod starts", + appliesTo: isSwiftV2Node, + eventReason: reasonFailedCreatePodSandBox, + signatures: mustCompileSignatures( + `no such network interface`, + `network is unreachable`, + `mtpnc is not ready`, + `dhcp discover.*timed out`, + ), + failuresFloor: 3, + window: 10 * time.Minute, + dwell: 10 * time.Minute, + requireZeroSuccess: true, +} + +func isSwiftV2Node(node *corev1.Node) bool { + return node != nil && node.Labels[SwiftV2LabelKey] == SwiftV2LabelValue +} diff --git a/mgmt-agent/pkg/controller/nodehealth/labeler.go b/mgmt-agent/pkg/controller/nodehealth/labeler.go new file mode 100644 index 00000000000..90702948754 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/labeler.go @@ -0,0 +1,209 @@ +// 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 + +import ( + "context" + "encoding/json" + "fmt" + "time" + + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/tools/record" + "k8s.io/klog/v2" + + "github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/nodehealth/detectors" +) + +// labeler applies and removes the node-health wedged label (plus explanatory +// annotations) on nodes, emitting Kubernetes Events and metrics. All operations +// are idempotent and non-disruptive; mitigation of a labeled node is owned by a +// separate controller. +type labeler struct { + client kubernetes.Interface + recorder record.EventRecorder + clock func() time.Time +} + +// newLabeler constructs a labeler. If clock is nil, time.Now is used. +func newLabeler(client kubernetes.Interface, recorder record.EventRecorder, clock func() time.Time) *labeler { + if clock == nil { + clock = time.Now + } + return &labeler{client: client, recorder: recorder, clock: clock} +} + +// label marks a node as wedged: it sets labelKey=labelValue plus the explanatory +// annotations recording which detector fired, why, and when. +// +// It returns changed=true when it mutated the node's health record, which is +// either a fresh label transition or a refresh of a detection record that no +// longer matches the firing detector (including one whose annotations were +// stripped). A node already labeled by the same detector is left completely +// untouched, so the steady state costs no writes and no API reads. +// +// The reason annotation is deliberately not reconciled on every pass. It embeds +// live evidence counts that change constantly, so continuously reconciling it +// would rewrite every wedged node on every resync tick for no operator benefit. +// The record is a snapshot of the detection, not a live readout. +func (l *labeler) label(ctx context.Context, node *corev1.Node, detector string, snap detectors.Snapshot) (bool, error) { + logger := klog.FromContext(ctx).WithValues("node", node.Name) + + // Steady state, off the informer cache: correct label and a matching + // detection record. Nothing to do, and no need to hit the apiserver. + if node.Labels[labelKey] == labelValue && node.Annotations[annotationDetector] == detector { + logger.V(4).Info("node already labeled wedged by this detector (cache)") + return false, nil + } + // Something looks like it needs writing. The informer cache lags the + // apiserver, so a resync tick and a watch event can both reconcile off the + // same stale view. Confirm against live state before mutating, so the counter + // and the NodeHealthLabeled Event fire once per real change, not once per + // racing reconcile. + live, err := l.client.CoreV1().Nodes().Get(ctx, node.Name, metav1.GetOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, fmt.Errorf("get node %s: %w", node.Name, err) + } + if live.Labels[labelKey] == labelValue && live.Annotations[annotationDetector] == detector { + logger.V(4).Info("node already labeled wedged by this detector (live)") + return false, nil + } + + reason := snap.ReasonString() + patch, err := metadataPatch( + map[string]*string{labelKey: ptr(labelValue)}, + map[string]*string{ + annotationDetector: ptr(detector), + annotationReason: ptr(reason), + annotationObservedAt: ptr(l.clock().UTC().Format(time.RFC3339)), + }, + ) + if err != nil { + labelActionsTotal.WithLabelValues("label", "error").Inc() + return false, err + } + if err := l.patch(ctx, node.Name, patch); err != nil { + labelActionsTotal.WithLabelValues("label", "error").Inc() + return false, err + } + labelActionsTotal.WithLabelValues("label", "success").Inc() + logger.Info("labeled node wedged", "label", labelKey+"="+labelValue, "detector", detector, "reason", reason) + l.eventf(node, corev1.EventTypeWarning, "NodeHealthLabeled", + "node-health marked node wedged (detector %q; %s)", detector, reason) + return true, nil +} + +// unlabel removes the node-health wedged label and its annotations from a node. +// It is a no-op when the label is absent or set to a different value (so it will +// not clobber a same-key label owned by another actor). It returns changed=true +// when it applied a mutation. +func (l *labeler) unlabel(ctx context.Context, node *corev1.Node) (bool, error) { + logger := klog.FromContext(ctx).WithValues("node", node.Name) + + if node.Labels[labelKey] != labelValue { + return false, nil + } + // The informer cache lags the apiserver, so confirm the label is still + // present at live state before mutating, so the counter and the + // NodeHealthUnlabeled Event fire once per real transition rather than once + // per racing reconcile off a stale still-labeled node. + live, err := l.client.CoreV1().Nodes().Get(ctx, node.Name, metav1.GetOptions{}) + if err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, fmt.Errorf("get node %s: %w", node.Name, err) + } + if live.Labels[labelKey] != labelValue { + return false, nil + } + + patch, err := metadataPatch( + map[string]*string{labelKey: nil}, + map[string]*string{ + annotationDetector: nil, + annotationReason: nil, + annotationObservedAt: nil, + }, + ) + if err != nil { + labelActionsTotal.WithLabelValues("unlabel", "error").Inc() + return false, err + } + if err := l.patch(ctx, node.Name, patch); err != nil { + labelActionsTotal.WithLabelValues("unlabel", "error").Inc() + return false, err + } + labelActionsTotal.WithLabelValues("unlabel", "success").Inc() + logger.Info("removed wedged label from node", "label", labelKey) + l.eventf(node, corev1.EventTypeNormal, "NodeHealthUnlabeled", + "node-health cleared wedged label (node recovered)") + return true, nil +} + +func (l *labeler) patch(ctx context.Context, name string, patch []byte) error { + if _, err := l.client.CoreV1().Nodes().Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{FieldManager: fieldManager}); err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("patch node %s: %w", name, err) + } + return nil +} + +func (l *labeler) eventf(node *corev1.Node, eventType, reason, format string, args ...interface{}) { + if l.recorder == nil { + return + } + l.recorder.Eventf(node, eventType, reason, format, args...) +} + +// metadataPatch builds a JSON merge patch (RFC 7386) for node metadata. A nil +// value removes the key; a non-nil value sets it. +func metadataPatch(labels, annotations map[string]*string) ([]byte, error) { + meta := map[string]interface{}{} + if len(labels) > 0 { + meta["labels"] = nullableMap(labels) + } + if len(annotations) > 0 { + meta["annotations"] = nullableMap(annotations) + } + b, err := json.Marshal(map[string]interface{}{"metadata": meta}) + if err != nil { + return nil, fmt.Errorf("failed to marshal metadata patch: %w", err) + } + return b, nil +} + +func nullableMap(m map[string]*string) map[string]interface{} { + out := make(map[string]interface{}, len(m)) + for k, v := range m { + if v == nil { + out[k] = nil + } else { + out[k] = *v + } + } + return out +} + +func ptr(s string) *string { return &s } diff --git a/mgmt-agent/pkg/controller/nodehealth/labeler_test.go b/mgmt-agent/pkg/controller/nodehealth/labeler_test.go new file mode 100644 index 00000000000..8f331eb5fe8 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/labeler_test.go @@ -0,0 +1,286 @@ +// 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 + +import ( + "context" + "testing" + "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/kubernetes/fake" + "k8s.io/client-go/tools/record" + + "github.com/Azure/ARO-HCP/mgmt-agent/pkg/controller/nodehealth/detectors" +) + +// testNow is a fixed clock for labeler tests. +var testNow = time.Date(2026, 7, 29, 12, 0, 0, 0, time.UTC) + +func newTestLabeler(objs ...*corev1.Node) (*labeler, *fake.Clientset) { + rt := make([]runtime.Object, 0, len(objs)) + for _, o := range objs { + rt = append(rt, o) + } + client := fake.NewSimpleClientset(rt...) + l := newLabeler(client, record.NewFakeRecorder(16), func() time.Time { return testNow }) + return l, client +} + +func getNode(t *testing.T, client *fake.Clientset, name string) *corev1.Node { + t.Helper() + n, err := client.CoreV1().Nodes().Get(context.Background(), name, metav1.GetOptions{}) + if err != nil { + t.Fatalf("get node: %v", err) + } + return n +} + +func snap() detectors.Snapshot { + return detectors.Snapshot{DetectorName: "swift-vf-teardown", Window: 10 * time.Minute, FailureCount: 30} +} + +func TestLabelAppliesLabelAndAnnotations(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}} + l, client := newTestLabeler(node) + + changed, err := l.label(context.Background(), node, "swift-vf-teardown", snap()) + if err != nil { + t.Fatalf("label: %v", err) + } + if !changed { + t.Fatal("expected changed=true") + } + + got := getNode(t, client, "n1") + if got.Labels[labelKey] != labelValue { + t.Errorf("label = %q, want %q", got.Labels[labelKey], labelValue) + } + if got.Annotations[annotationDetector] != "swift-vf-teardown" { + t.Errorf("detector annotation = %q", got.Annotations[annotationDetector]) + } + if got.Annotations[annotationReason] == "" { + t.Error("reason annotation should be set") + } + if got.Annotations[annotationObservedAt] == "" { + t.Error("observed-at annotation should be set") + } +} + +func TestLabelIsIdempotent(t *testing.T) { + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + Annotations: map[string]string{annotationDetector: "swift-vf-teardown"}, + }, + } + l, _ := newTestLabeler(node) + + changed, err := l.label(context.Background(), node, "swift-vf-teardown", snap()) + if err != nil { + t.Fatalf("label: %v", err) + } + if changed { + t.Error("re-labeling an already-labeled node should be a no-op") + } +} + +func TestUnlabelRemovesLabelPreservingOthers(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{ + labelKey: labelValue, + "kubernetes.io/os": "linux", + detectors.SwiftV2LabelKey: detectors.SwiftV2LabelValue, + }, + Annotations: map[string]string{ + annotationDetector: "swift-vf-teardown", + annotationReason: "x", + annotationObservedAt: "t", + "unrelated": "keep", + }, + }} + l, client := newTestLabeler(node) + + changed, err := l.unlabel(context.Background(), node) + if err != nil { + t.Fatalf("unlabel: %v", err) + } + if !changed { + t.Fatal("expected changed=true") + } + + got := getNode(t, client, "n1") + if _, ok := got.Labels[labelKey]; ok { + t.Error("wedged label should be removed") + } + if got.Labels["kubernetes.io/os"] != "linux" { + t.Error("unrelated label should be preserved") + } + if got.Labels[detectors.SwiftV2LabelKey] != detectors.SwiftV2LabelValue { + t.Error("SWIFT label should be preserved") + } + if _, ok := got.Annotations[annotationDetector]; ok { + t.Error("detector annotation should be removed") + } + if got.Annotations["unrelated"] != "keep" { + t.Error("unrelated annotation should be preserved") + } +} + +func TestUnlabelNoopWhenAbsent(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}} + l, _ := newTestLabeler(node) + + changed, err := l.unlabel(context.Background(), node) + if err != nil { + t.Fatalf("unlabel: %v", err) + } + if changed { + t.Error("unlabeling a node without the label should be a no-op") + } +} + +func TestUnlabelNoopOnForeignValue(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: "somethingelse"}, + }} + l, client := newTestLabeler(node) + + changed, err := l.unlabel(context.Background(), node) + if err != nil { + t.Fatalf("unlabel: %v", err) + } + if changed { + t.Error("a same-key label with a foreign value must not be clobbered") + } + if getNode(t, client, "n1").Labels[labelKey] != "somethingelse" { + t.Error("foreign label value should be untouched") + } +} + +func TestLabelSkipsDuplicateOnStaleCache(t *testing.T) { + // The server already has the node labeled wedged, but the informer cache the + // caller reconciles off is stale and still shows it unlabeled. The live + // re-check must catch this so no duplicate counter or Event fires for what is + // not a real transition. + served := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + Annotations: map[string]string{annotationDetector: "swift-vf-teardown"}, + }} + client := fake.NewSimpleClientset(served) + rec := record.NewFakeRecorder(16) + l := newLabeler(client, rec, func() time.Time { return testNow }) + + stale := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}} + changed, err := l.label(context.Background(), stale, "swift-vf-teardown", snap()) + if err != nil { + t.Fatalf("label: %v", err) + } + if changed { + t.Error("stale-cache label of an already-labeled node must report changed=false") + } + select { + case ev := <-rec.Events: + t.Errorf("no Event expected on a duplicate transition, got %q", ev) + default: + } +} + +func TestUnlabelSkipsDuplicateOnStaleCache(t *testing.T) { + // The server has already cleared the label, but the informer cache the caller + // reconciles off is stale and still shows it present. The live re-check must + // catch this so no duplicate counter or Event fires. + served := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "n1"}} + client := fake.NewSimpleClientset(served) + rec := record.NewFakeRecorder(16) + l := newLabeler(client, rec, func() time.Time { return testNow }) + + stale := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + }} + changed, err := l.unlabel(context.Background(), stale) + if err != nil { + t.Fatalf("unlabel: %v", err) + } + if changed { + t.Error("stale-cache unlabel of an already-cleared node must report changed=false") + } + select { + case ev := <-rec.Events: + t.Errorf("no Event expected on a duplicate transition, got %q", ev) + default: + } +} + +func TestLabelRefreshesStrippedDetectionRecord(t *testing.T) { + // The node carries our label but its detection record is gone (stripped, or + // written by an older version). The record must self-heal rather than stay + // blank forever behind the already-labeled short-circuit. + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + }} + l, client := newTestLabeler(node) + + changed, err := l.label(context.Background(), node, "swift-vf-teardown", snap()) + if err != nil { + t.Fatalf("label: %v", err) + } + if !changed { + t.Fatal("a missing detection record should be refreshed") + } + + got := getNode(t, client, "n1") + if got.Annotations[annotationDetector] != "swift-vf-teardown" { + t.Errorf("detector annotation = %q, want it restored", got.Annotations[annotationDetector]) + } + if got.Annotations[annotationReason] == "" { + t.Error("reason annotation should be restored") + } + if got.Labels[labelKey] != labelValue { + t.Error("label should still be present") + } +} + +func TestLabelSteadyStateMakesNoAPICall(t *testing.T) { + // A node already labeled by the same detector is the steady state: it must + // cost neither a write nor a read, so a wedged node does not generate an + // apiserver GET on every resync tick. + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{ + Name: "n1", + Labels: map[string]string{labelKey: labelValue}, + Annotations: map[string]string{annotationDetector: "swift-vf-teardown"}, + }} + l, client := newTestLabeler(node) + client.ClearActions() + + changed, err := l.label(context.Background(), node, "swift-vf-teardown", snap()) + if err != nil { + t.Fatalf("label: %v", err) + } + if changed { + t.Error("steady state must report changed=false") + } + if acts := client.Actions(); len(acts) != 0 { + t.Errorf("steady state must not call the apiserver, got %d actions: %v", len(acts), acts) + } +} diff --git a/mgmt-agent/pkg/controller/nodehealth/metrics.go b/mgmt-agent/pkg/controller/nodehealth/metrics.go new file mode 100644 index 00000000000..4747ea76f81 --- /dev/null +++ b/mgmt-agent/pkg/controller/nodehealth/metrics.go @@ -0,0 +1,67 @@ +// 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 + +import ( + "sync" + + "k8s.io/component-base/metrics" + "k8s.io/component-base/metrics/legacyregistry" +) + +const metricsSubsystem = "nodehealth" + +var ( + detectionsTotal = metrics.NewCounterVec( + &metrics.CounterOpts{ + Subsystem: metricsSubsystem, + Name: "detections_total", + Help: "Number of times a detector caused a node to be marked wedged, counted when the node's detection record changes rather than on every re-evaluation.", + StabilityLevel: metrics.ALPHA, + }, + []string{"detector"}, + ) + + labelActionsTotal = metrics.NewCounterVec( + &metrics.CounterOpts{ + Subsystem: metricsSubsystem, + Name: "label_actions_total", + Help: "Number of label/unlabel actions, by action and result.", + StabilityLevel: metrics.ALPHA, + }, + []string{"action", "result"}, + ) + + wedgedNodes = metrics.NewGauge( + &metrics.GaugeOpts{ + Subsystem: metricsSubsystem, + Name: "wedged_nodes", + Help: "Number of nodes currently carrying the wedged health label, as observed by the node-health controller. Reported as 0 while the controller is disabled.", + StabilityLevel: metrics.ALPHA, + }, + ) +) + +var registerOnce sync.Once + +// RegisterMetrics registers the node-health metrics with the shared legacy +// registry used by the mgmt-agent /metrics endpoint. Safe to call multiple times. +func RegisterMetrics() { + registerOnce.Do(func() { + legacyregistry.MustRegister(detectionsTotal) + legacyregistry.MustRegister(labelActionsTotal) + legacyregistry.MustRegister(wedgedNodes) + }) +} diff --git a/mgmt-agent/values.yaml b/mgmt-agent/values.yaml index cf6a742634b..f22a5397c0b 100644 --- a/mgmt-agent/values.yaml +++ b/mgmt-agent/values.yaml @@ -14,6 +14,20 @@ deployment: logVerbosity: 2 health: port: 8080 +nodeHealth: + configMapName: mgmt-agent-node-health + configKey: config.yaml + # Rendered into the ConfigMap the controller watches. Detection logic, + # thresholds, and the SWIFT-v2 node scoping are hard-coded; the only runtime + # config is this switch. + # + # This file is the authoritative source for the ConfigMap's contents. The + # controller hot-reloads a live edit to the ConfigMap within one resync, which + # makes `kubectl edit` a valid break-glass to switch the controller off during + # an incident, but that edit is not durable: the next mgmt-agent rollout runs + # helm and restores the value from here. A change meant to stick goes in git. + config: + enabled: false nodeSelector: {} tolerations: [] affinity: {} diff --git a/mgmt-agent/zz_fixture_TestHelmTemplate_dev_westus3_mgmt_1_mgmt_agent.yaml b/mgmt-agent/zz_fixture_TestHelmTemplate_dev_westus3_mgmt_1_mgmt_agent.yaml index 814ea7d1e74..ec6d594fabf 100644 --- a/mgmt-agent/zz_fixture_TestHelmTemplate_dev_westus3_mgmt_1_mgmt_agent.yaml +++ b/mgmt-agent/zz_fixture_TestHelmTemplate_dev_westus3_mgmt_1_mgmt_agent.yaml @@ -24,6 +24,18 @@ metadata: name: mgmt-agent namespace: mgmt-agent --- +# Source: mgmt-agent/templates/node-health-config.yaml +apiVersion: v1 +kind: ConfigMap +metadata: + name: mgmt-agent-node-health + namespace: mgmt-agent + labels: + app.kubernetes.io/name: mgmt-agent +data: + config.yaml: | + enabled: false +--- # Source: mgmt-agent/templates/clusterrole.yaml apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRole @@ -40,6 +52,7 @@ rules: - get - list - watch + - patch - apiGroups: - "" resources: @@ -47,6 +60,23 @@ rules: verbs: - patch - update +- apiGroups: + - "" + resources: + - pods + verbs: + - list + - watch +- apiGroups: + - "" + resources: + - events + verbs: + - list + - watch + - create + - patch + - update - apiGroups: - "*" resources: @@ -212,6 +242,8 @@ spec: - "--health-address=:8080" - "--ksm-image=mcr.microsoft.com/oss/v2/kubernetes/kube-state-metrics@sha256:1234567890" - "--log-verbosity=2" + - "--node-health-configmap=mgmt-agent-node-health" + - "--node-health-config-key=config.yaml" image: "arohcpsvcdev.azurecr.io/mgmt-agent@sha256:1234567890" name: mgmt-agent-controller env: