From 331bebde2ca2a7b72e14308dfa80df4853b3c8b9 Mon Sep 17 00:00:00 2001 From: Tuomas Katila Date: Tue, 8 Sep 2026 16:56:34 +0300 Subject: [PATCH 1/2] api: add XpuManagerSpec.restartOnDeviceRecovery Co-Authored-By: Claude Opus 5 Signed-off-by: Tuomas Katila --- README.md | 1 + api/v1alpha1/clusterpolicy_types.go | 23 +++++++++++++++++++ charts/gpu-base-operator-policy/README.md | 1 + .../templates/clusterpolicy.yaml | 3 +++ charts/gpu-base-operator-policy/values.yaml | 2 ++ .../crds/clusterpolicies.yaml | 10 ++++++++ .../crd/bases/intel.com_clusterpolicies.yaml | 10 ++++++++ 7 files changed, 50 insertions(+) diff --git a/README.md b/README.md index 5412b90..bf710b7 100644 --- a/README.md +++ b/README.md @@ -206,6 +206,7 @@ Applies to both DP and DRA unless noted. Thresholds that are exceeded mark the G |`spec.xpu.monitoringResource`|Set XPUMD resource for Device Plugin use.|`monitoring`| |`spec.xpu.configMapOverride`|Name of a ConfigMap in the operator namespace containing a custom OpenTelemetry Collector `config.yaml`|—| |`spec.xpu.affinity`|Optional `k8s.io/api/core/v1` `Affinity` object applied to the XPU manager daemon set|—| +|`spec.xpu.restartOnDeviceRecovery`|When to restart a node's XPU Manager pod so it can monitor a GPU its container cannot reach. `OnRecoveredDevice`, `Always` or `Disabled`.|`OnRecoveredDevice`| #### Kueue (`spec.kueue`) diff --git a/api/v1alpha1/clusterpolicy_types.go b/api/v1alpha1/clusterpolicy_types.go index 4cd4fcc..d582509 100644 --- a/api/v1alpha1/clusterpolicy_types.go +++ b/api/v1alpha1/clusterpolicy_types.go @@ -163,6 +163,13 @@ type XpuManagerSpec struct { // Set optional affinities for XPU pods // +optional Affinity *v1.Affinity `json:"affinity,omitempty"` + + // RestartOnDeviceRecovery controls whether the operator restarts a node's XPU Manager pod so + // that it can monitor a GPU its container cannot currently reach. + // +kubebuilder:validation:Enum=OnRecoveredDevice;Always;Disabled + // +kubebuilder:default:=OnRecoveredDevice + // +optional + RestartOnDeviceRecovery XpumRestartMode `json:"restartOnDeviceRecovery,omitempty"` } // RegistryTLSSpec configures TLS behavior for accessing container image registries. @@ -267,6 +274,22 @@ type BuildArg struct { Value string `json:"value"` } +// XpumRestartMode selects when the operator replaces a node's XPU Manager pod to give it a GPU its +// container cannot reach. +type XpumRestartMode string + +const ( + // XpumRestartOnRecoveredDevice restarts the pod for a usable GPU its container was never + // given a device node for, and leaves a re-enumerated one to XPU Manager's own rescan. + XpumRestartOnRecoveredDevice XpumRestartMode = "OnRecoveredDevice" + + // XpumRestartAlways also restarts when a device the container does hold re-enumerates. + XpumRestartAlways XpumRestartMode = "Always" + + // XpumRestartDisabled disables both the record and the restart. + XpumRestartDisabled XpumRestartMode = "Disabled" +) + // ClusterPolicyStatus defines the observed state of ClusterPolicy. type ClusterPolicyStatus struct { DevicePluginStatus string `json:"devicePluginStatus,omitempty"` diff --git a/charts/gpu-base-operator-policy/README.md b/charts/gpu-base-operator-policy/README.md index d8a3807..610ee69 100644 --- a/charts/gpu-base-operator-policy/README.md +++ b/charts/gpu-base-operator-policy/README.md @@ -47,6 +47,7 @@ See [Customizing the Chart Before Installing](https://helm.sh/docs/intro/using_h | xpu.logLevel | 2 | XPU manager log level. | | xpu.monitoringResource | monitoring | Monitoring resource for XPUMD with device plugin. | | xpu.configMapOverride | "" | Override the default XPUM configuration ConfigMap name. | +| xpu.restartOnDeviceRecovery | OnRecoveredDevice | When to restart a node's XPUMD pod so it can monitor a GPU its container cannot reach: `OnRecoveredDevice`, `Always` or `Disabled`. | | kueue.equalResources | [] | List of ClusterQueue configurations. | | pullSecret | null | Image pull secret. | | nodeSelector | {} | Node selector for scheduling pods. | diff --git a/charts/gpu-base-operator-policy/templates/clusterpolicy.yaml b/charts/gpu-base-operator-policy/templates/clusterpolicy.yaml index 479e328..525aa41 100644 --- a/charts/gpu-base-operator-policy/templates/clusterpolicy.yaml +++ b/charts/gpu-base-operator-policy/templates/clusterpolicy.yaml @@ -73,6 +73,9 @@ spec: {{- if .Values.xpu.configMapOverride }} configMapOverride: {{ .Values.xpu.configMapOverride | default "" }} {{- end }} +{{- if .Values.xpu.restartOnDeviceRecovery }} + restartOnDeviceRecovery: {{ .Values.xpu.restartOnDeviceRecovery | quote }} +{{- end }} {{- if .Values.kernelModule }} kernelModule: diff --git a/charts/gpu-base-operator-policy/values.yaml b/charts/gpu-base-operator-policy/values.yaml index 3fc3b75..0c655db 100644 --- a/charts/gpu-base-operator-policy/values.yaml +++ b/charts/gpu-base-operator-policy/values.yaml @@ -48,6 +48,8 @@ xpu: # - matchExpressions: # - key: gpu.intel.com/xpumd-deny-node # operator: DoesNotExist + # Restart a node's XPU Manager pod so it can monitor a GPU its container cannot reach. + restartOnDeviceRecovery: OnRecoveredDevice kueue: equalResources: diff --git a/charts/gpu-base-operator/crds/clusterpolicies.yaml b/charts/gpu-base-operator/crds/clusterpolicies.yaml index 3ed38ae..b659249 100644 --- a/charts/gpu-base-operator/crds/clusterpolicies.yaml +++ b/charts/gpu-base-operator/crds/clusterpolicies.yaml @@ -3191,6 +3191,16 @@ spec: - xe_monitoring - monitoring type: string + restartOnDeviceRecovery: + default: OnRecoveredDevice + description: |- + RestartOnDeviceRecovery controls whether the operator restarts a node's XPU Manager pod so + that it can monitor a GPU its container cannot currently reach. + enum: + - OnRecoveredDevice + - Always + - Disabled + type: string type: object required: - resourceRegistration diff --git a/config/crd/bases/intel.com_clusterpolicies.yaml b/config/crd/bases/intel.com_clusterpolicies.yaml index 3ed38ae..b659249 100644 --- a/config/crd/bases/intel.com_clusterpolicies.yaml +++ b/config/crd/bases/intel.com_clusterpolicies.yaml @@ -3191,6 +3191,16 @@ spec: - xe_monitoring - monitoring type: string + restartOnDeviceRecovery: + default: OnRecoveredDevice + description: |- + RestartOnDeviceRecovery controls whether the operator restarts a node's XPU Manager pod so + that it can monitor a GPU its container cannot currently reach. + enum: + - OnRecoveredDevice + - Always + - Disabled + type: string type: object required: - resourceRegistration From 3644ebce70c829baa1ad8f463af96de3006967ce Mon Sep 17 00:00:00 2001 From: Tuomas Katila Date: Tue, 8 Sep 2026 16:56:41 +0300 Subject: [PATCH 2/2] xpum: restart xpumd for GPUs its container cannot reach A container is given its GPU device nodes at creation. An xpumd container that started while one of its node's GPUs was in survivability mode therefore monitors every GPU on that node except that one. When the GPU is recovered from survivability mode, xpumd has to be restarted so that is can observe the newly detectable device. The restart logic is tied to the ClusterPolicy's xpu.restartOnDeviceRecovery. Similar to the recovery functionality, it is dependant on using DRA and its ResourceSlices. Co-Authored-By: Claude Opus 5 Signed-off-by: Tuomas Katila --- .../templates/namespaced_role.yaml | 8 + cmd/main.go | 7 + config/rbac/namespaced_role.yaml | 10 + internal/controller/xpum_device_refresh.go | 771 ++++++++++++++++ .../controller/xpum_device_refresh_test.go | 871 ++++++++++++++++++ 5 files changed, 1667 insertions(+) create mode 100644 internal/controller/xpum_device_refresh.go create mode 100644 internal/controller/xpum_device_refresh_test.go diff --git a/charts/gpu-base-operator/templates/namespaced_role.yaml b/charts/gpu-base-operator/templates/namespaced_role.yaml index 2729c28..fbb4bb2 100644 --- a/charts/gpu-base-operator/templates/namespaced_role.yaml +++ b/charts/gpu-base-operator/templates/namespaced_role.yaml @@ -26,6 +26,14 @@ rules: - pods/log verbs: - get +# Patching pods is how XpumDeviceRefreshReconciler records, on each XPU Manager pod, which of its +# node's GPUs its container can actually use. Kept in step with config/rbac/namespaced_role.yaml. +- apiGroups: + - "" + resources: + - pods + verbs: + - patch - apiGroups: - "" resources: diff --git a/cmd/main.go b/cmd/main.go index 54d397e..b69c965 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -375,6 +375,13 @@ func main() { setupLog.Error(err, "unable to create controller", "controller", "GPURecoveryPlan") os.Exit(1) } + if err := (&controller.XpumDeviceRefreshReconciler{ + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + }).SetupWithManager(mgr, copts); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "XpumDeviceRefresh") + os.Exit(1) + } // nolint:goconst if os.Getenv("DISABLE_WEBHOOKS") != "true" { diff --git a/config/rbac/namespaced_role.yaml b/config/rbac/namespaced_role.yaml index 4d4fce1..bbbcf0c 100644 --- a/config/rbac/namespaced_role.yaml +++ b/config/rbac/namespaced_role.yaml @@ -25,6 +25,16 @@ rules: - pods/log verbs: - get +# Patching pods is how XpumDeviceRefreshReconciler records, on each XPU Manager pod, which of its +# node's GPUs its container can actually use. Granted here rather than through a kubebuilder marker +# because those all merge into the cluster-wide role, and this write only ever targets the operator's +# own namespace. Reading and deleting pods is cluster-wide already, for the firmware update jobs. +- apiGroups: + - "" + resources: + - pods + verbs: + - patch - apiGroups: - "" resources: diff --git a/internal/controller/xpum_device_refresh.go b/internal/controller/xpum_device_refresh.go new file mode 100644 index 0000000..f3bf708 --- /dev/null +++ b/internal/controller/xpum_device_refresh.go @@ -0,0 +1,771 @@ +/* +Copyright 2026 Intel Corporation. All Rights Reserved. + +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 controller + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" + "time" + + core "k8s.io/api/core/v1" + resv1 "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/klog/v2" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + v1alpha "github.com/intel/gpu-base-operator/api/v1alpha1" +) + +// XPUMD device refresh. +// +// The xpumd device refresh controller watches the xpumd pods on each node and compares the GPUs +// they were given against the GPUs the node currently has. If a pod is missing a GPU that is +// published and usable, the pod is deleted so the DaemonSet controller replaces it with one that +// gets the node's current GPUs. +// Controller's behavior is controlled by the ClusterPolicy's spec.xpu.restartOnDeviceRecovery field, +// which can be set to Always, OnRecoveredDevice, or Disabled. OnRecoveredDevice is when a device is known +// to have been in unusable state and then comes back. Always triggers a Pod restart every time a device +// rebinds, even if it was never unusable. Disabled disables the controller entirely. + +type XpumDeviceRefreshReconciler struct { + client.Client + Scheme *runtime.Scheme + Opts ControllerOpts + + // mu guards the three per-node guard maps below. + mu sync.Mutex + + // lastRestart records when this controller last deleted the xpum pod on a node, to space out + // restarts of the same node. + lastRestart map[string]time.Time + + // restartAttempts counts consecutive restarts of a node that did not resolve its divergence. + // Reset as soon as a pass finds the node converged. + restartAttempts map[string]int + + // lostDevices tracks, per node, the devices a pod's record says it holds that the slices have + // since called unusable — tainted for recovery, bound to vfio, or with no driver bound at all. + lostDevices map[string]map[string]bool +} + +const ( + // xpumDevicesAnnotation records the GPUs whose device nodes its container was given. + // gpu.intel.com/xpum-devices: 0000-04-00-0-0xe20b,0000-05-00-0-0xe20b + xpumDevicesAnnotation = "gpu.intel.com/xpum-devices" + + // deviceAttrDriver is the ResourceSlice device attribute name for the kernel driver + deviceAttrDriver = "driver" + + // resourceSliceNodeNameIndex is the field index deviceStates selects on, so that a pass costs + // one node's slices rather than the cluster's. Slice churn is frequent and reconciles are + // per-node, so this is a hot path. + resourceSliceNodeNameIndex = "spec.nodeName" + + xpumdMonitorableDriverXe = "xe" + xpumdMonitorableDriverI915 = "i915" + + // xpumRestartCooldown is the minimum spacing between two restarts on the same Node. + xpumRestartCooldown = 2 * time.Minute + + maxConcurrentXpumRestarts = 3 + + maxXpumRestartAttempts = 3 + + restartReasonRecovered = "recovered-device" + restartReasonRebind = "rebind" +) + +// deviceState is what the ResourceSlices say about one GPU right now. Presence in the map built by +// deviceStates means the device is published for the node; absence means nothing at all is known +// about it, which is deliberately not the same as bad news. +type deviceState struct { + // recovering is set when the device carries a taint the operator recovers from, i.e. it is + // unusable now and will re-enumerate when it is fixed. + recovering bool + + // monitorable is set when the device has a driver bound whose devices xpumd can read. Clear + // for a card handed to vfio for passthrough and for one with no driver bound at all. + monitorable bool +} + +// usable reports whether a published device is one xpumd could monitor if its container had the +// device node. +func (s deviceState) usable() bool { + return s.monitorable && !s.recovering +} + +type restartOutcome int + +const ( + restartDone restartOutcome = iota + restartDeferred + restartAbandoned +) + +// The ClusterPolicy is read on every pass for spec.xpu.restartOnDeviceRecovery, and the slices are +// what say whether a GPU is usable at all. +// +kubebuilder:rbac:groups=intel.com,resources=clusterpolicies,verbs=get;list;watch +// +kubebuilder:rbac:groups=resource.k8s.io,resources=resourceslices,verbs=get;list;watch + +// Patching of Pods is limited to the operator's namespace. "patch" right is needed, but it's +// requested in the namespaced role, not here. +// +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;delete +// +kubebuilder:rbac:groups=resource.k8s.io,resources=resourceclaims,verbs=get;list;watch + +// Reconcile checks one node's xpum pods against the GPUs the node currently has, and replaces a pod +// that cannot reach one of them. +func (r *XpumDeviceRefreshReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + node := req.Name + + // Check what, if anything, the ClusterPolicy asks this controller to do. + mode, err := r.restartMode(ctx) + if err != nil { + return ctrl.Result{}, err + } + + if mode == v1alpha.XpumRestartDisabled { + r.forgetNode(node) + + return ctrl.Result{}, nil + } + + states, err := r.deviceStates(ctx, node) + if err != nil { + return ctrl.Result{}, err + } + + if len(states) == 0 { + // No Intel GPUs published for this node: nothing to compare against, and nothing to record. + r.forgetNode(node) + + return ctrl.Result{}, nil + } + + pods, err := r.xpumPodsOnNode(ctx, node) + if err != nil { + return ctrl.Result{}, err + } + + requeue := false + + for _, pod := range pods { + podRequeue, err := r.refreshPod(ctx, pod, node, states, mode) + if err != nil { + return ctrl.Result{}, err + } + + requeue = requeue || podRequeue + } + + if requeue { + return ctrl.Result{RequeueAfter: r.Opts.RequeueDelay}, nil + } + + return ctrl.Result{}, nil +} + +// refreshPod handles one xpum pod: adopt it if it has no record yet, otherwise compare the record +// against the node's current devices and restart the pod if it is missing one. Returns true if the +// pass should be retried, which happens when it is waiting for something (a claim allocation, a +// restart slot) rather than failing. +func (r *XpumDeviceRefreshReconciler) refreshPod(ctx context.Context, pod *core.Pod, node string, + states map[string]deviceState, mode v1alpha.XpumRestartMode) (bool, error) { + // Only Running pods are considered. + if pod.Status.Phase != core.PodRunning { + return false, nil + } + + // Retrieve the devices annotated on the Pod. + record, adopted := pod.Annotations[xpumDevicesAnnotation] + + // If the pod has no record yet, adopt it. + if !adopted { + return r.adoptPod(ctx, pod, states) + } + + held := parseDeviceRecord(record) + + missing := missingDevices(states, held) + + // Track rebinds whatever the mode, so that a rebind resolved under OnRecoveredDevice is consumed here + // rather than left to fire the first time somebody switches to Always. + rebound := r.trackRebinds(node, held, states) + + reason := "" + + switch { + case len(missing) > 0: + reason = restartReasonRecovered + + klog.V(2).Infof("xpum pod %s/%s was never given usable GPUs %s on node %s", + pod.Namespace, pod.Name, strings.Join(missing, ","), node) + case len(rebound) > 0 && mode == v1alpha.XpumRestartAlways: + reason = restartReasonRebind + + klog.V(2).Infof("xpum pod %s/%s holds GPUs %s that re-enumerated on node %s", + pod.Namespace, pod.Name, strings.Join(rebound, ","), node) + case len(rebound) > 0: + klog.V(2).Infof("xpum pod %s/%s holds GPUs %s that re-enumerated on node %s; leaving them to its own rescan", + pod.Namespace, pod.Name, strings.Join(rebound, ","), node) + + // Declining to act on it still consumes the edge, so it is not left waiting for whoever + // switches this node to Always later. + r.clearRebinds(node, rebound) + } + + // No reason found, do not restart the pod. + if reason == "" { + // Converged: forget any restart attempts spent getting here, so an unrelated divergence + // later starts from a full budget. + r.resetAttempts(node) + + return false, nil + } + + // Reason found, try to restart the pod. + outcome, err := r.restartPod(ctx, pod, node, reason) + if err != nil { + return false, err + } + + // Only now is a rebind consumed: the cooldown and the concurrency cap can push the restart to a + // later pass, which recomputes `rebound` and needs the edge to still be there. + if reason == restartReasonRebind && outcome != restartDeferred { + r.clearRebinds(node, rebound) + } + + return outcome == restartDeferred, nil +} + +// adoptPod writes the initial device record for a pod that does not have one. +func (r *XpumDeviceRefreshReconciler) adoptPod(ctx context.Context, pod *core.Pod, states map[string]deviceState) (bool, error) { + allocated, ready, err := r.allocatedDeviceNames(ctx, pod) + if err != nil { + return false, err + } + + if !ready { + // The claim is not allocated yet, or its status has not reached the cache. Try again later. + klog.V(2).Infof("xpum pod %s/%s has no allocated monitoring claim yet; deferring adoption", pod.Namespace, pod.Name) + + return true, nil + } + + held := make([]string, 0, len(allocated)) + + for name := range allocated { + state, published := states[name] + + if !published || state.usable() { + held = append(held, name) + } + } + + formatted := formatDeviceRecord(held) + + if err := r.patchDeviceRecord(ctx, pod, formatted); err != nil { + return false, err + } + + klog.V(2).Infof("adopted xpum pod %s/%s with devices %q", pod.Namespace, pod.Name, formatted) + + return false, nil +} + +// restartPod deletes a node's xpum pod so the DaemonSet controller replaces it with one that gets +// the node's currently usable GPUs. +func (r *XpumDeviceRefreshReconciler) restartPod(ctx context.Context, pod *core.Pod, node, reason string) (restartOutcome, error) { + r.mu.Lock() + + attempts := r.restartAttempts[node] + last := r.lastRestart[node] + + r.mu.Unlock() + + if attempts >= maxXpumRestartAttempts { + // Report and stop. Restarting again would not fix whatever is keeping the device out of + // the container, and the gauge stays up so the condition is still visible. + klog.Errorf("xpum pod on node %s still cannot use GPUs that are published and untainted after %d restarts; giving up on this node", + node, attempts) + + return restartAbandoned, nil + } + + if since := time.Since(last); !last.IsZero() && since < xpumRestartCooldown { + // Logged so that a deferred restart is distinguishable from a device this controller never + // noticed. + klog.V(2).Infof("deferring xpum restart on node %s (%s): last restart was %s ago, cooldown is %s", + node, reason, since.Truncate(time.Second), xpumRestartCooldown) + + return restartDeferred, nil + } + + restarting, err := r.restartingXpumPods(ctx) + if err != nil { + return restartDeferred, err + } + + if restarting >= maxConcurrentXpumRestarts { + klog.V(2).Infof("deferring xpum restart on node %s (%s): %d xpum pods are already restarting", + node, reason, restarting) + + return restartDeferred, nil + } + + if err := r.Delete(ctx, pod); err != nil { + return restartDeferred, fmt.Errorf("failed to delete xpum pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + + r.mu.Lock() + r.lastRestart[node] = time.Now() + r.restartAttempts[node] = attempts + 1 + r.mu.Unlock() + + klog.Infof("restarted xpum pod %s/%s: node %s has GPUs its container cannot monitor (%s)", + pod.Namespace, pod.Name, node, reason) + + return restartDone, nil +} + +// missingDevices returns the devices that are usable on the node now and were not among the ones the +// container was given, sorted so that log lines for an unchanged condition are unchanged. +func missingDevices(states map[string]deviceState, held map[string]bool) []string { + missing := make([]string, 0, len(states)) + + for name, state := range states { + if state.usable() && !held[name] { + missing = append(missing, name) + } + } + + sort.Strings(missing) + + return missing +} + +// trackRebinds notes which of the devices a container holds the slices currently call unusable, and +// reports the ones that were unusable on an earlier pass and are usable again — a card that +// re-enumerated behind a device node the container already has. +// +// The edge is any published-and-unusable state, not the recovery taint alone: a KMD unbind shows up +// as the driver attribute going "xe" → "" → "xe" with no taint anywhere, and a device handed to vfio +// and taken back has the same shape. A device vanishing from the slices is deliberately not an edge +// — see the type comment. +func (r *XpumDeviceRefreshReconciler) trackRebinds(node string, held map[string]bool, states map[string]deviceState) []string { + r.mu.Lock() + defer r.mu.Unlock() + + lost := r.lostDevices[node] + rebound := make([]string, 0, len(lost)) + + for name := range held { + state, published := states[name] + if !published { + continue + } + + switch { + case !state.usable(): + if lost == nil { + lost = map[string]bool{} + r.lostDevices[node] = lost + } + + lost[name] = true + case lost[name]: + rebound = append(rebound, name) + } + } + + sort.Strings(rebound) + + return rebound +} + +// clearRebinds forgets rebind edges that have been dealt with, either by a restart or by a mode that +// deliberately leaves them to xpumd's own rescan. Kept separate from trackRebinds so that a restart +// held back by the cooldown or the concurrency cap still has its edge on the next pass. +func (r *XpumDeviceRefreshReconciler) clearRebinds(node string, names []string) { + r.mu.Lock() + defer r.mu.Unlock() + + lost := r.lostDevices[node] + if lost == nil { + return + } + + for _, name := range names { + delete(lost, name) + } + + if len(lost) == 0 { + delete(r.lostDevices, node) + } +} + +// parseDeviceRecord reads the annotation value into the set of devices the container holds. +func parseDeviceRecord(value string) map[string]bool { + held := map[string]bool{} + + for _, entry := range strings.Split(value, ",") { + if entry == "" { + continue + } + + held[entry] = true + } + + return held +} + +// formatDeviceRecord renders a device set sorted, so that the same set always produces the same +// annotation value. +func formatDeviceRecord(names []string) string { + sorted := make([]string, len(names)) + copy(sorted, names) + sort.Strings(sorted) + + return strings.Join(sorted, ",") +} + +func (r *XpumDeviceRefreshReconciler) patchDeviceRecord(ctx context.Context, pod *core.Pod, value string) error { + patch := client.MergeFrom(pod.DeepCopy()) + + if pod.Annotations == nil { + pod.Annotations = map[string]string{} + } + + pod.Annotations[xpumDevicesAnnotation] = value + + if err := r.Patch(ctx, pod, patch); err != nil { + return fmt.Errorf("failed to record devices on xpum pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + + return nil +} + +// restartMode resolves spec.xpu.restartOnDeviceRecovery from the first ClusterPolicy, and folds in +// the two conditions that make this controller inert regardless of what the field says: +// monitoring switched off (no xpum pods to restart) and resource registration other than DRA +// (no ResourceSlices, so no way to tell whether a device is usable, and no claim for the pod either). +func (r *XpumDeviceRefreshReconciler) restartMode(ctx context.Context) (v1alpha.XpumRestartMode, error) { + cpList := &v1alpha.ClusterPolicyList{} + if err := r.List(ctx, cpList); err != nil { + return v1alpha.XpumRestartDisabled, fmt.Errorf("failed to list ClusterPolicies: %w", err) + } + + var cp *v1alpha.ClusterPolicy + + for i := range cpList.Items { + // Skip a CR that is being deleted. + if cpList.Items[i].DeletionTimestamp != nil { + continue + } + + cp = &cpList.Items[i] + + break + } + + // No active CPs found, nothing to do. + if cp == nil { + return v1alpha.XpumRestartDisabled, nil + } + + // No xpumd (monitoring) on the cluster, nothing to do. + if !cp.Spec.ResourceMonitoring { + return v1alpha.XpumRestartDisabled, nil + } + + // No DRA, nothing to do. + if !r.Opts.DRAEnable || cp.Spec.ResourceRegistration != resourceModeDRA { + return v1alpha.XpumRestartDisabled, nil + } + + switch mode := cp.Spec.XpuManagerSpec.RestartOnDeviceRecovery; mode { + case v1alpha.XpumRestartOnRecoveredDevice, v1alpha.XpumRestartAlways, v1alpha.XpumRestartDisabled: + return mode, nil + case "": + // An object that bypassed API-server defaulting, which in practice is a unit test. + return v1alpha.XpumRestartOnRecoveredDevice, nil + default: + klog.Warningf("unknown spec.xpu.restartOnDeviceRecovery %q on ClusterPolicy %s; treating it as %s", + mode, cp.Name, v1alpha.XpumRestartOnRecoveredDevice) + + return v1alpha.XpumRestartOnRecoveredDevice, nil + } +} + +// deviceStates returns every Intel GPU published for a node, with what the slices currently say +// about it. +func (r *XpumDeviceRefreshReconciler) deviceStates(ctx context.Context, node string) (map[string]deviceState, error) { + slices := &resv1.ResourceSliceList{} + if err := r.List(ctx, slices, client.MatchingFields{resourceSliceNodeNameIndex: node}); err != nil { + return nil, fmt.Errorf("failed to list ResourceSlices for node %s: %w", node, err) + } + + states := map[string]deviceState{} + + for i := range slices.Items { + slice := &slices.Items[i] + + if slice.Spec.Driver != gpuDeviceClass { + continue + } + + for j := range slice.Spec.Devices { + dev := &slice.Spec.Devices[j] + + prev, seen := states[dev.Name] + + state := deviceState{ + recovering: deviceNeedsRecovery(dev), + monitorable: xpumdMonitorableDriver(dev), + } + + // Merge with the previous state if this device has already been seen. + if seen { + state.recovering = state.recovering || prev.recovering + state.monitorable = state.monitorable && prev.monitorable + } + + states[dev.Name] = state + } + } + + return states, nil +} + +// deviceNeedsRecovery reports whether a device carries a taint the operator recovers from, i.e. it +// is unusable now and will re-enumerate once it is fixed. +func deviceNeedsRecovery(dev *resv1.Device) bool { + for i := range dev.Taints { + if _, ok := taintToDeviceNeed(dev.Taints[i].Key, v1alpha.RecoveryTypeSBR); ok { + return true + } + } + + return false +} + +// xpumdMonitorableDriver reports whether a device has a driver bound that gives xpumd something to +// monitor. +func xpumdMonitorableDriver(dev *resv1.Device) bool { + attr, published := dev.Attributes[resv1.QualifiedName(deviceAttrDriver)] + if !published || attr.StringValue == nil { + return true + } + + switch strings.ToLower(*attr.StringValue) { + case xpumdMonitorableDriverXe, xpumdMonitorableDriverI915: + return true + default: + return false + } +} + +// allocatedDeviceNames returns the device names allocated to a pod's monitoring claim, and whether +// the allocation could be read at all. +func (r *XpumDeviceRefreshReconciler) allocatedDeviceNames(ctx context.Context, pod *core.Pod) (map[string]bool, bool, error) { + names := map[string]bool{} + found := false + + for _, status := range pod.Status.ResourceClaimStatuses { + if status.ResourceClaimName == nil { + continue + } + + claim := &resv1.ResourceClaim{} + + key := client.ObjectKey{Name: *status.ResourceClaimName, Namespace: pod.Namespace} + if err := r.Get(ctx, key, claim); err != nil { + return nil, false, fmt.Errorf("failed to get ResourceClaim %s: %w", key, err) + } + + if claim.Status.Allocation == nil { + continue + } + + found = true + + for _, result := range claim.Status.Allocation.Devices.Results { + if result.Driver != gpuDeviceClass { + continue + } + + names[result.Device] = true + } + } + + return names, found, nil +} + +// xpumPodsOnNode returns the xpum pods scheduled on a node. +func (r *XpumDeviceRefreshReconciler) xpumPodsOnNode(ctx context.Context, node string) ([]*core.Pod, error) { + pods, err := r.listXpumPods(ctx) + if err != nil { + return nil, err + } + + onNode := make([]*core.Pod, 0, 1) + + for _, pod := range pods { + if pod.Spec.NodeName == node && pod.DeletionTimestamp == nil { + onNode = append(onNode, pod) + } + } + + return onNode, nil +} + +func (r *XpumDeviceRefreshReconciler) listXpumPods(ctx context.Context) ([]*core.Pod, error) { + podList := &core.PodList{} + + err := r.List(ctx, podList, client.InNamespace(r.Opts.Namespace), client.MatchingLabels{xpuLabel: xpuValue}) + if err != nil { + return nil, fmt.Errorf("failed to list xpum pods: %w", err) + } + + pods := make([]*core.Pod, 0, len(podList.Items)) + for i := range podList.Items { + pods = append(pods, &podList.Items[i]) + } + + return pods, nil +} + +// restartingXpumPods counts xpum pods that are not currently Running, which is how many nodes are +// mid-restart whether this controller caused it or not. +func (r *XpumDeviceRefreshReconciler) restartingXpumPods(ctx context.Context) (int, error) { + pods, err := r.listXpumPods(ctx) + if err != nil { + return 0, err + } + + count := 0 + + for _, pod := range pods { + if pod.Status.Phase != core.PodRunning || pod.DeletionTimestamp != nil { + count++ + } + } + + return count, nil +} + +// forgetNode drops a node's guard state and metric series. Called when the node has no GPUs to +// watch or the feature is off, so that a node leaving the cluster does not leave a gauge behind +// reading whatever it read last. +func (r *XpumDeviceRefreshReconciler) forgetNode(node string) { + r.mu.Lock() + defer r.mu.Unlock() + + delete(r.lastRestart, node) + delete(r.restartAttempts, node) + delete(r.lostDevices, node) +} + +func (r *XpumDeviceRefreshReconciler) resetAttempts(node string) { + r.mu.Lock() + defer r.mu.Unlock() + + delete(r.restartAttempts, node) +} + +// indexResourceSliceByNodeName is the index function behind resourceSliceNodeNameIndex. A slice with +// no node — one describing network-attached devices — is left out of the index entirely, so it +// cannot match a node name. +func indexResourceSliceByNodeName(obj client.Object) []string { + slice, ok := obj.(*resv1.ResourceSlice) + if !ok || slice.Spec.NodeName == nil || *slice.Spec.NodeName == "" { + return nil + } + + return []string{*slice.Spec.NodeName} +} + +// resourceSliceToNode maps a ResourceSlice event to the node it describes. +func (r *XpumDeviceRefreshReconciler) resourceSliceToNode(_ context.Context, obj client.Object) []reconcile.Request { + slice, ok := obj.(*resv1.ResourceSlice) + if !ok { + return nil + } + + if slice.Spec.Driver != gpuDeviceClass || slice.Spec.NodeName == nil { + return nil + } + + return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: *slice.Spec.NodeName}}} +} + +// xpumPodToNode maps an xpum pod event to its node. +func (r *XpumDeviceRefreshReconciler) xpumPodToNode(_ context.Context, obj client.Object) []reconcile.Request { + pod, ok := obj.(*core.Pod) + if !ok { + return nil + } + + if pod.Namespace != r.Opts.Namespace || pod.Labels[xpuLabel] != xpuValue || pod.Spec.NodeName == "" { + return nil + } + + return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: pod.Spec.NodeName}}} +} + +// SetupWithManager registers the controller with the Manager. +// +// There is no For() object: requests are node names rather than objects, and both watches map into +// that keyspace. It is a top-level controller rather than a ClusterPolicy sub-reconciler on purpose +// — ResourceSlice churn is frequent, and fanning it into ClusterPolicy would re-run the device +// plugin, DRA and misc reconcilers and diff three DaemonSets every time a GPU's taints changed. +// +// The ClusterPolicy is read on every pass but not watched, so a mode change takes effect on the next +// slice or pod event for a node rather than immediately. Nothing durable depends on the mode, so +// there is nothing for the change to act on retroactively. +func (r *XpumDeviceRefreshReconciler) SetupWithManager(mgr ctrl.Manager, opts ControllerOpts) error { + r.Opts = opts + r.lastRestart = map[string]time.Time{} + r.restartAttempts = map[string]int{} + r.lostDevices = map[string]map[string]bool{} + + // The index is only used by this controller, so it is registered here rather than alongside the + // shared drain indexes. + err := mgr.GetFieldIndexer().IndexField(context.Background(), &resv1.ResourceSlice{}, + resourceSliceNodeNameIndex, indexResourceSliceByNodeName) + if err != nil { + return fmt.Errorf("failed to register the ResourceSlice %s index: %w", resourceSliceNodeNameIndex, err) + } + + return ctrl.NewControllerManagedBy(mgr). + Watches( + &resv1.ResourceSlice{}, + handler.EnqueueRequestsFromMapFunc(r.resourceSliceToNode), + ). + Watches( + &core.Pod{}, + handler.EnqueueRequestsFromMapFunc(r.xpumPodToNode), + ). + Named("xpumdevicerefresh"). + Complete(r) +} diff --git a/internal/controller/xpum_device_refresh_test.go b/internal/controller/xpum_device_refresh_test.go new file mode 100644 index 0000000..f09d13e --- /dev/null +++ b/internal/controller/xpum_device_refresh_test.go @@ -0,0 +1,871 @@ +/* +Copyright 2026 Intel Corporation. All Rights Reserved. + +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 controller + +import ( + "context" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + core "k8s.io/api/core/v1" + resv1 "k8s.io/api/resource/v1" + "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + v1alpha "github.com/intel/gpu-base-operator/api/v1alpha1" +) + +var _ = Describe("XPU Manager device refresh", func() { + ctx := context.Background() + + const ( + refreshNode = "xpum-refresh-node" + + // A namespace of its own, labelled for admin access. The label is not test scaffolding: + // the API server refuses adminAccess requests and allocations in a namespace without it, + // so the operator's own namespace needs it for the monitoring claim to be allocated at all. + refreshNS = "xpum-refresh" + devA = "0000-04-00-0-0xe20b" + devB = "0000-05-00-0-0xe20b" + devC = "0000-06-00-0-0xe20b" + claimName = "xpum-refresh-claim" + ) + + // publishedDevice describes one device to publish: its name, the taint key it carries if any, + // and the kernel driver the DRA driver reports it bound to (defaulting to xe). + // + // unbound publishes the driver attribute as the empty string, which is how a device with no KMD + // bound appears — the state a reflash, a driver reload and a passthrough switch all pass through. + type publishedDevice struct { + name string + taint string + driver string + unbound bool + } + + BeforeEach(func() { + ns := &core.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: refreshNS, + Labels: map[string]string{"resource.kubernetes.io/admin-access": "true"}, + }, + } + + // Namespaces are never torn down: envtest has no namespace controller, so a deleted one + // stays Terminating and refuses new objects for the rest of the suite. + if err := k8sClient.Create(ctx, ns); err != nil { + Expect(errors.IsAlreadyExists(err)).To(BeTrue(), "unexpected error creating namespace: %v", err) + } + + // Sweep leftover xpum pods before every spec. Without a kubelet nothing finishes a + // graceful deletion, so a pod this suite deleted — or one the reconciler restarted — + // stays Terminating forever, and restartingXpumPods counts it against the + // concurrency cap. Three leaks and no spec could ever observe a restart again. + Expect(k8sClient.DeleteAllOf(ctx, &core.Pod{}, + client.InNamespace(refreshNS), + client.MatchingLabels{xpuLabel: xpuValue}, + client.GracePeriodSeconds(0))).To(Succeed()) + }) + + newRefreshReconciler := func() *XpumDeviceRefreshReconciler { + return &XpumDeviceRefreshReconciler{ + Client: k8sClient, + Scheme: k8sClient.Scheme(), + lastRestart: map[string]time.Time{}, + restartAttempts: map[string]int{}, + lostDevices: map[string]map[string]bool{}, + Opts: ControllerOpts{ + Namespace: refreshNS, + RequeueDelay: 2 * time.Second, + DRAEnable: true, + }, + } + } + + // makePolicy creates the single ClusterPolicy the reconciler reads its mode from. Any policy + // left over from another spec is removed first: restartMode takes the first item of the list, + // so a leak would silently decide these specs' behaviour. + makePolicy := func(mode v1alpha.XpumRestartMode, registration string) { + existing := &v1alpha.ClusterPolicyList{} + Expect(k8sClient.List(ctx, existing)).To(Succeed()) + + for i := range existing.Items { + Expect(k8sClient.Delete(ctx, &existing.Items[i])).To(Succeed()) + } + + cp := &v1alpha.ClusterPolicy{ + ObjectMeta: metav1.ObjectMeta{Name: "xpum-refresh-policy"}, + Spec: v1alpha.ClusterPolicySpec{ + ResourceRegistration: registration, + ResourceMonitoring: true, + XpuManagerSpec: v1alpha.XpuManagerSpec{ + Image: "xpumd:test", + RestartOnDeviceRecovery: mode, + }, + }, + } + Expect(k8sClient.Create(ctx, cp)).To(Succeed()) + DeferCleanup(func() { + _ = k8sClient.Delete(ctx, cp) + }) + } + + // publishSlice publishes the node's GPUs, replacing whatever was there before, so a spec can + // move a device in and out of a recovery taint the way the driver does. + publishSlice := func(devices ...publishedDevice) { + name := "slice-" + refreshNode + key := types.NamespacedName{Name: name} + + slice := &resv1.ResourceSlice{} + if err := k8sClient.Get(ctx, key, slice); err == nil { + Expect(k8sClient.Delete(ctx, slice)).To(Succeed()) + } + + devs := make([]resv1.Device, 0, len(devices)) + + for _, d := range devices { + driver := d.driver + if driver == "" && !d.unbound { + driver = "xe" + } + + dev := resv1.Device{ + Name: d.name, + Attributes: map[resv1.QualifiedName]resv1.DeviceAttribute{ + deviceAttrDeviceID: {StringValue: ptr.To("0x1234")}, + deviceAttrBDF: {StringValue: ptr.To("0000:04:00.0")}, + deviceAttrDriver: {StringValue: ptr.To(driver)}, + }, + } + + if d.taint != "" { + dev.Taints = []resv1.DeviceTaint{ + {Key: d.taint, Effect: resv1.DeviceTaintEffectNoExecute}, + } + } + + devs = append(devs, dev) + } + + fresh := &resv1.ResourceSlice{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: resv1.ResourceSliceSpec{ + Driver: gpuDeviceClass, + NodeName: ptr.To(refreshNode), + Pool: resv1.ResourcePool{Name: "pool-" + refreshNode, ResourceSliceCount: 1}, + Devices: devs, + }, + } + Expect(k8sClient.Create(ctx, fresh)).To(Succeed()) + DeferCleanup(func() { + stale := &resv1.ResourceSlice{} + if err := k8sClient.Get(ctx, key, stale); err == nil { + _ = k8sClient.Delete(ctx, stale) + } + }) + } + + // makeClaim creates the pod's monitoring claim, allocated to the named devices. + // + // Both the request and the allocation results carry adminAccess, as the real monitoring claim's + // do: counting admin-access results is the one thing this controller has to do differently from + // claimHoldsDevice, which skips them. + makeClaim := func(name string, allocated ...string) { + claim := &resv1.ResourceClaim{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: refreshNS}, + Spec: resv1.ResourceClaimSpec{ + Devices: resv1.DeviceClaim{ + Requests: []resv1.DeviceRequest{{ + Name: "gpu", + Exactly: &resv1.ExactDeviceRequest{ + DeviceClassName: gpuDeviceClass, + AdminAccess: ptr.To(true), + AllocationMode: resv1.DeviceAllocationModeAll, + }, + }}, + }, + }, + } + Expect(k8sClient.Create(ctx, claim)).To(Succeed()) + DeferCleanup(func() { + fresh := &resv1.ResourceClaim{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: refreshNS}, fresh); err == nil { + fresh.Status = resv1.ResourceClaimStatus{} + _ = k8sClient.Status().Update(ctx, fresh) + _ = k8sClient.Delete(ctx, fresh) + } + }) + + if len(allocated) == 0 { + return + } + + results := make([]resv1.DeviceRequestAllocationResult, 0, len(allocated)) + + for _, dev := range allocated { + results = append(results, resv1.DeviceRequestAllocationResult{ + Request: "gpu", + Driver: gpuDeviceClass, + Pool: "pool-" + refreshNode, + Device: dev, + AdminAccess: ptr.To(true), + }) + } + + claim.Status = resv1.ResourceClaimStatus{ + Allocation: &resv1.AllocationResult{ + Devices: resv1.DeviceAllocationResult{Results: results}, + }, + } + Expect(k8sClient.Status().Update(ctx, claim)).To(Succeed()) + } + + // makeXpumPod creates an xpum DaemonSet-style pod on the node. A nil record leaves the pod + // unadopted, which is what the adoption specs need; a pointer to the empty string is a pod that + // was adopted having been given no usable GPUs at all, which is a different thing entirely. + makeXpumPod := func(name string, phase core.PodPhase, record *string, claim string) *core.Pod { + pod := &core.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Namespace: refreshNS, + Labels: map[string]string{xpuLabel: xpuValue}, + }, + Spec: core.PodSpec{ + NodeName: refreshNode, + Containers: []core.Container{{ + Name: xpumdContainerName, + Image: "xpumd:test", + Resources: core.ResourceRequirements{ + Claims: []core.ResourceClaim{{Name: monClaim}}, + }, + }}, + ResourceClaims: []core.PodResourceClaim{{ + Name: monClaim, + ResourceClaimName: ptr.To(claim), + }}, + }, + } + + if record != nil { + pod.Annotations = map[string]string{xpumDevicesAnnotation: *record} + } + + Expect(k8sClient.Create(ctx, pod)).To(Succeed()) + DeferCleanup(func() { + fresh := &core.Pod{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: refreshNS}, fresh); err == nil { + _ = k8sClient.Delete(ctx, fresh, client.GracePeriodSeconds(0)) + } + }) + + pod.Status.Phase = phase + pod.Status.ResourceClaimStatuses = []core.PodResourceClaimStatus{{ + Name: monClaim, + ResourceClaimName: ptr.To(claim), + }} + Expect(k8sClient.Status().Update(ctx, pod)).To(Succeed()) + + return pod + } + + reconcileNode := func(r *XpumDeviceRefreshReconciler) reconcile.Result { + res, err := r.Reconcile(ctx, reconcile.Request{ + NamespacedName: types.NamespacedName{Name: refreshNode}, + }) + Expect(err).NotTo(HaveOccurred()) + + return res + } + + deviceRecord := func(name string) (string, bool) { + pod := &core.Pod{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: refreshNS}, pod); err != nil { + Expect(errors.IsNotFound(err)).To(BeTrue(), "unexpected error reading pod: %v", err) + + return "", false + } + + value, ok := pod.Annotations[xpumDevicesAnnotation] + + return value, ok + } + + podExists := func(name string) bool { + pod := &core.Pod{} + err := k8sClient.Get(ctx, types.NamespacedName{Name: name, Namespace: refreshNS}, pod) + + return err == nil && pod.DeletionTimestamp == nil + } + + Context("adoption", func() { + It("records what the container was given and does not restart it", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-adopt", core.PodRunning, nil, claimName) + + reconcileNode(newRefreshReconciler()) + + record, ok := deviceRecord("xpum-adopt") + Expect(ok).To(BeTrue()) + Expect(record).To(Equal(devA + "," + devB)) + Expect(podExists("xpum-adopt")).To(BeTrue(), "adoption must never restart a pod") + }) + + It("leaves out an allocated device that is tainted for recovery", func() { + // The monitoring claim has no selectors and tolerates every device taint, so a card in + // survivability mode is allocated to the pod. Allocation is not usability. + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice( + publishedDevice{name: devA}, + publishedDevice{name: devB, taint: "health-Survivability"}, + ) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-adopt-tainted", core.PodRunning, nil, claimName) + + reconcileNode(newRefreshReconciler()) + + record, _ := deviceRecord("xpum-adopt-tainted") + Expect(record).To(Equal(devA)) + Expect(podExists("xpum-adopt-tainted")).To(BeTrue()) + }) + + It("leaves out a device bound away from the graphics drivers", func() { + // A card handed to vfio-pci has no DRM node for anyone to monitor: it is not a device + // the container is missing, and no restart could deliver it. + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + publishSlice( + publishedDevice{name: devA}, + publishedDevice{name: devB, driver: "vfio-pci"}, + ) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-adopt-vfio", core.PodRunning, nil, claimName) + + r := newRefreshReconciler() + reconcileNode(r) + + record, _ := deviceRecord("xpum-adopt-vfio") + Expect(record).To(Equal(devA)) + + By("reconciling again now that the record is in place") + reconcileNode(r) + + Expect(podExists("xpum-adopt-vfio")).To(BeTrue(), + "a passthrough device must not restart the pod once per pass forever") + }) + + It("restarts on the next pass for a device the pod started before", func() { + // The pod won the race against the DRA driver: the device is published now but was not + // in the pod's allocation, so its container was never given a node for it. + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + makeClaim(claimName, devA) + makeXpumPod("xpum-adopt-late", core.PodRunning, nil, claimName) + + r := newRefreshReconciler() + reconcileNode(r) + + record, _ := deviceRecord("xpum-adopt-late") + Expect(record).To(Equal(devA)) + Expect(podExists("xpum-adopt-late")).To(BeTrue(), "adoption itself never restarts") + + reconcileNode(r) + + Expect(podExists("xpum-adopt-late")).To(BeFalse()) + }) + + It("records an allocated device that is absent from the slices as held", func() { + // The allocation is evidence the device was there when the container was created, which + // is when the injection happened. A publishing gap since then says nothing about what is + // in the container, and recording it as missing would restart the pod the moment the + // slice came back. + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-adopt-gap", core.PodRunning, nil, claimName) + + r := newRefreshReconciler() + reconcileNode(r) + + record, _ := deviceRecord("xpum-adopt-gap") + Expect(record).To(Equal(devA + "," + devB)) + + By("the missing slice coming back") + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + reconcileNode(r) + + Expect(podExists("xpum-adopt-gap")).To(BeTrue()) + }) + + It("waits rather than guessing when the claim is not allocated yet", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}) + makeClaim(claimName) + makeXpumPod("xpum-adopt-unallocated", core.PodRunning, nil, claimName) + + res := reconcileNode(newRefreshReconciler()) + + // A record written now would be a guess, and it would be believed for the pod's life. + _, ok := deviceRecord("xpum-adopt-unallocated") + Expect(ok).To(BeFalse()) + Expect(res.RequeueAfter).To(BeNumerically(">", 0)) + }) + + It("does not adopt a pod that is not Running", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}) + makeClaim(claimName, devA) + makeXpumPod("xpum-pending", core.PodPending, nil, claimName) + + reconcileNode(newRefreshReconciler()) + + _, ok := deviceRecord("xpum-pending") + Expect(ok).To(BeFalse()) + Expect(podExists("xpum-pending")).To(BeTrue()) + }) + }) + + Context("a GPU the container was never given", func() { + It("restarts the pod when a card that booted in survivability mode is recovered", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-boot", core.PodRunning, ptr.To(devA), claimName) + + // The reflash finished and the driver dropped the taint. + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + + reconcileNode(newRefreshReconciler()) + + Expect(podExists("xpum-boot")).To(BeFalse(), + "only a new container can be given a device node for that card") + }) + + It("restarts the pod for a GPU that appears after it started", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-new-device", core.PodRunning, ptr.To(devA), claimName) + + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devC}) + + reconcileNode(newRefreshReconciler()) + + Expect(podExists("xpum-new-device")).To(BeFalse()) + }) + + It("restarts a pod that was given nothing once a GPU becomes usable", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-empty", core.PodRunning, ptr.To(""), claimName) + + publishSlice(publishedDevice{name: devA}) + + reconcileNode(newRefreshReconciler()) + + Expect(podExists("xpum-empty")).To(BeFalse(), + "an empty record is a container that was given no GPUs, not one that was never adopted") + }) + + It("waits for a device with no driver bound rather than restarting for it", func() { + // An unbound device published as usable was the other half of the same bug: a restart + // cannot deliver a card that has no DRM device, so it fires, fails to converge and burns + // the node's whole restart budget while the reflash is still running. + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-unbound-wait", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + By("devB published with no driver bound") + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB, unbound: true}) + reconcileNode(r) + + Expect(podExists("xpum-unbound-wait")).To(BeTrue()) + + By("devB coming back on xe") + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + reconcileNode(r) + + Expect(podExists("xpum-unbound-wait")).To(BeFalse(), + "now there is a device node to hand over, and this container has none") + }) + + It("leaves a converged pod alone", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-converged", core.PodRunning, ptr.To(devA+","+devB), claimName) + + r := newRefreshReconciler() + reconcileNode(r) + reconcileNode(r) + + Expect(podExists("xpum-converged")).To(BeTrue()) + record, _ := deviceRecord("xpum-converged") + Expect(record).To(Equal(devA+","+devB), "the record is written once and never rewritten") + }) + }) + + Context("a GPU that re-enumerated behind a device node the container holds", func() { + It("leaves it to XPU Manager's own rescan under OnRecoveredDevice", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-rebind", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + By("xpumd reporting the card into survivability while the reflash runs") + publishSlice(publishedDevice{name: devA, taint: deviceTaintKeyXpumdReflash}) + reconcileNode(r) + + Expect(podExists("xpum-rebind")).To(BeTrue(), + "restarting mid-recovery would not get the device back and would interrupt monitoring") + + By("the reflash completing and the taint clearing") + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-rebind")).To(BeTrue(), + "the container still holds the node, and the card comes back on the minor it freed") + + record, _ := deviceRecord("xpum-rebind") + Expect(record).To(Equal(devA)) + }) + + It("restarts the pod under Always", func() { + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-rebind-always", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + By("the card being reset") + publishSlice(publishedDevice{name: devA, taint: deviceTaintKeyReset}) + reconcileNode(r) + + Expect(podExists("xpum-rebind-always")).To(BeTrue()) + + By("the reset completing") + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-rebind-always")).To(BeFalse()) + }) + + It("sees a KMD unbind, which carries no taint at all", func() { + // The case that exposed the taint-keyed edge: a reflash, a driver reload and a + // manageBinding switch all show up as the driver attribute going xe -> "" -> xe, with no + // device taint anywhere. Keying the edge on the taint missed every one of them, so Always + // never restarted for the most common re-enumeration there is. + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-unbind", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + By("the KMD being unbound") + publishSlice(publishedDevice{name: devA, unbound: true}) + reconcileNode(r) + + Expect(podExists("xpum-unbind")).To(BeTrue(), + "there is nothing behind an unbound device for a new container to be given either") + + By("xe binding it again") + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-unbind")).To(BeFalse()) + }) + + It("leaves an unbind to XPU Manager's rescan under OnRecoveredDevice", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-unbind-rescan", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + publishSlice(publishedDevice{name: devA, unbound: true}) + reconcileNode(r) + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-unbind-rescan")).To(BeTrue()) + }) + + It("keeps the edge when a guard defers the restart", func() { + // The edge is consumed when the restart is performed or declined, not when it is + // detected. Consuming it on detection lost the restart outright: the next pass recomputed + // the rebind set from an edge that had already been deleted and found the node converged. + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-deferred", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + r.lastRestart[refreshNode] = time.Now() + + By("a full unbind and rebind inside the cooldown window") + publishSlice(publishedDevice{name: devA, unbound: true}) + reconcileNode(r) + publishSlice(publishedDevice{name: devA}) + res := reconcileNode(r) + + Expect(podExists("xpum-deferred")).To(BeTrue(), "the cooldown holds this pass back") + Expect(res.RequeueAfter).To(BeNumerically(">", 0)) + + By("the cooldown expiring") + r.lastRestart[refreshNode] = time.Now().Add(-2 * xpumRestartCooldown) + reconcileNode(r) + + Expect(podExists("xpum-deferred")).To(BeFalse(), + "the rebind was deferred, not dropped") + }) + + It("needs the taint edge, not just a slice write, under Always", func() { + // Without a remembered taint there is no rebind: republishing an untainted device must + // not be mistaken for a card that went away and came back. + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-no-edge", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-no-edge")).To(BeTrue()) + }) + }) + + Context("changes that must not restart anything", func() { + It("ignores a device that disappears from the slices", func() { + // A slice can vanish and come back for publisher reasons — a DRA driver upgrade — with + // nothing changed inside any running container. Checked under Always, the mode that + // tracks rebinds, because that is where absence could leak in. + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA, devB) + makeXpumPod("xpum-absent", core.PodRunning, ptr.To(devA+","+devB), claimName) + + publishSlice(publishedDevice{name: devA}) + + r := newRefreshReconciler() + reconcileNode(r) + + Expect(podExists("xpum-absent")).To(BeTrue()) + + By("the slice coming back unchanged") + publishSlice(publishedDevice{name: devA}, publishedDevice{name: devB}) + reconcileNode(r) + + Expect(podExists("xpum-absent")).To(BeTrue(), + "a republished slice must not restart every xpum pod in the cluster") + }) + + It("ignores a taint the operator does not recover from", func() { + makePolicy(v1alpha.XpumRestartAlways, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-unrelated", core.PodRunning, ptr.To(devA), claimName) + + r := newRefreshReconciler() + + By("a temperature excursion tainting the device") + publishSlice(publishedDevice{name: devA, taint: "health-Temperature"}) + reconcileNode(r) + + By("the excursion clearing") + publishSlice(publishedDevice{name: devA}) + reconcileNode(r) + + Expect(podExists("xpum-unrelated")).To(BeTrue(), + "a transient fault does not re-enumerate the card, so there is nothing to pick up") + }) + }) + + Context("restartOnDeviceRecovery", func() { + It("does nothing at all when Disabled", func() { + makePolicy(v1alpha.XpumRestartDisabled, resourceModeDRA) + publishSlice(publishedDevice{name: devA}) + makeClaim(claimName, devA) + makeXpumPod("xpum-disabled", core.PodRunning, nil, claimName) + + reconcileNode(newRefreshReconciler()) + + _, ok := deviceRecord("xpum-disabled") + Expect(ok).To(BeFalse()) + Expect(podExists("xpum-disabled")).To(BeTrue()) + }) + + It("is inert when GPUs are registered through the device plugin", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, "dp") + publishSlice(publishedDevice{name: devA}) + makeClaim(claimName, devA) + makeXpumPod("xpum-dp", core.PodRunning, nil, claimName) + + reconcileNode(newRefreshReconciler()) + + _, ok := deviceRecord("xpum-dp") + Expect(ok).To(BeFalse()) + }) + }) + + Context("guards", func() { + It("gives up on a node whose divergence survives repeated restarts", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-attempts", core.PodRunning, ptr.To(""), claimName) + publishSlice(publishedDevice{name: devA}) + + r := newRefreshReconciler() + r.restartAttempts[refreshNode] = maxXpumRestartAttempts + + reconcileNode(r) + + Expect(podExists("xpum-attempts")).To(BeTrue(), + "a restart that does not help will not help the next time either") + }) + + It("spaces out restarts of the same node", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-cooldown", core.PodRunning, ptr.To(""), claimName) + publishSlice(publishedDevice{name: devA}) + + r := newRefreshReconciler() + r.lastRestart[refreshNode] = time.Now() + + res := reconcileNode(r) + + Expect(podExists("xpum-cooldown")).To(BeTrue()) + Expect(res.RequeueAfter).To(BeNumerically(">", 0)) + }) + + It("holds back when too many xpum pods are already restarting", func() { + makePolicy(v1alpha.XpumRestartOnRecoveredDevice, resourceModeDRA) + makeClaim(claimName, devA) + makeXpumPod("xpum-capped", core.PodRunning, ptr.To(""), claimName) + publishSlice(publishedDevice{name: devA}) + + for i := range maxConcurrentXpumRestarts { + // Pods on other nodes that have not reached Running: the fleet is mid-restart. + other := &core.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "xpum-other-" + string(rune('a'+i)), + Namespace: refreshNS, + Labels: map[string]string{xpuLabel: xpuValue}, + }, + Spec: core.PodSpec{ + NodeName: "other-node-" + string(rune('a'+i)), + Containers: []core.Container{{Name: xpumdContainerName, Image: "xpumd:test"}}, + }, + } + Expect(k8sClient.Create(ctx, other)).To(Succeed()) + DeferCleanup(func() { + _ = k8sClient.Delete(ctx, other, client.GracePeriodSeconds(0)) + }) + } + + res := reconcileNode(newRefreshReconciler()) + + Expect(podExists("xpum-capped")).To(BeTrue(), + "cluster monitoring must not go dark everywhere at once") + Expect(res.RequeueAfter).To(BeNumerically(">", 0)) + }) + }) +}) + +var _ = Describe("XPU Manager device record", func() { + DescribeTable("is read as the set of devices the container was given", + func(value string, expected []string) { + held := parseDeviceRecord(value) + + names := make([]string, 0, len(held)) + for name := range held { + names = append(names, name) + } + + Expect(names).To(ConsistOf(expected)) + }, + Entry("empty: a container that was given nothing", "", []string{}), + Entry("one device", "dev-a", []string{"dev-a"}), + Entry("several, in any order", "dev-b,dev-a", []string{"dev-a", "dev-b"}), + ) + + It("renders a set in a stable order", func() { + Expect(formatDeviceRecord([]string{"dev-b", "dev-a"})).To(Equal("dev-a,dev-b")) + Expect(formatDeviceRecord(nil)).To(BeEmpty()) + }) +}) + +var _ = Describe("What the slices say about a device", func() { + // The DRA driver publishes the boot-time survivability taint capitalised, while the keys xpumd + // sources are lowercase. Both spellings are written out literally here rather than referenced + // through their constants, so a change to either one has to be a deliberate edit to this table. + DescribeTable("a recovery taint is matched only in the spelling the driver publishes", + func(key string, expectRecovery bool) { + dev := &resv1.Device{ + Name: "dev-0", + Taints: []resv1.DeviceTaint{{Key: key, Effect: resv1.DeviceTaintEffectNoExecute}}, + } + + Expect(deviceNeedsRecovery(dev)).To(Equal(expectRecovery)) + }, + Entry("as the driver publishes it", "health-Survivability", true), + Entry("folded", "health-survivability", false), + Entry("shouted", "HEALTH-SURVIVABILITY", false), + Entry("sourced from xpumd", "health-xpumd-gpu.survivability", true), + Entry("a wedged card", "health-xpumd-gpu.wedged", true), + Entry("a transient condition", "health-Temperature", false), + Entry("nothing recognisable", "example.com/other", false), + ) + + DescribeTable("and the driver attribute only answers what it can", + func(driver string, monitorable bool) { + dev := &resv1.Device{ + Name: "dev-0", + Attributes: map[resv1.QualifiedName]resv1.DeviceAttribute{ + deviceAttrDriver: {StringValue: ptr.To(driver)}, + }, + } + + Expect(xpumdMonitorableDriver(dev)).To(Equal(monitorable)) + }, + // xe binds to a card in survivability mode too, which is why the taint and not this + // attribute decides whether a *bound* device is usable. + Entry("xe", "xe", true), + Entry("i915", "i915", true), + Entry("passed through", "vfio-pci", false), + Entry("passed through, underscored", "VFIO_PCI", false), + Entry("passed through via xe", "xe-vfio", false), + // Published as empty means no KMD is bound, so there is no DRM device behind it. Reading + // this as monitorable is what restarted pods for devices mid-reflash and hid every unbind + // from the rebind tracking. + Entry("nothing bound", "", false), + // An allowlist, so a KMD nobody here knows about costs a missed restart rather than a + // restart loop for a device that can never arrive. + Entry("unknown driver names", "something-new", false), + ) + + It("treats a device with no driver attribute as monitorable", func() { + Expect(xpumdMonitorableDriver(&resv1.Device{Name: "dev-0"})).To(BeTrue()) + }) +})