From 4136f90894481c743f0c4eae0b146c7e931fff37 Mon Sep 17 00:00:00 2001 From: Tuomas Katila Date: Mon, 31 Aug 2026 09:08:52 +0300 Subject: [PATCH] controller: drain: extract node taint and pod drain primitives Upcoming GPU recovery controller needs similar functionality that exists in the GPU FW update controller now. Move drain functionality to its own file so it can be used from both controllers. This also moves the controller to use Eviction API instead of Pod Delete. Signed-off-by: Tuomas Katila --- charts/gpu-base-operator/templates/role.yaml | 6 + cmd/main.go | 8 + config/rbac/role.yaml | 6 + internal/controller/drain.go | 425 ++++++++++ internal/controller/drain_test.go | 799 ++++++++++++++++++ .../gpufirmwareupdate_controller.go | 171 +--- .../gpufirmwareupdate_controller_test.go | 14 +- 7 files changed, 1286 insertions(+), 143 deletions(-) create mode 100644 internal/controller/drain.go create mode 100644 internal/controller/drain_test.go diff --git a/charts/gpu-base-operator/templates/role.yaml b/charts/gpu-base-operator/templates/role.yaml index 51b89f7..a90e6e9 100644 --- a/charts/gpu-base-operator/templates/role.yaml +++ b/charts/gpu-base-operator/templates/role.yaml @@ -31,6 +31,12 @@ rules: - get - list - watch +- apiGroups: + - "" + resources: + - pods/eviction + verbs: + - create - apiGroups: - admissionregistration.k8s.io resources: diff --git a/cmd/main.go b/cmd/main.go index f94c9ba..5021218 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -17,6 +17,7 @@ limitations under the License. package main import ( + "context" "crypto/tls" "flag" "os" @@ -346,6 +347,13 @@ func main() { ModuleLoaderServiceAccountName: moduleLoaderSAName, } + // Registered here rather than in a controller's SetupWithManager: the index is shared by + // every controller that drains a node, and controller-runtime allows only one registration. + if err := controller.SetupDrainIndices(context.Background(), mgr); err != nil { + setupLog.Error(err, "unable to set up drain cache indexes") + os.Exit(1) + } + if err := (&controller.ClusterPolicyReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index b68ee6b..a07fcba 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -31,6 +31,12 @@ rules: - get - list - watch +- apiGroups: + - "" + resources: + - pods/eviction + verbs: + - create - apiGroups: - admissionregistration.k8s.io resources: diff --git a/internal/controller/drain.go b/internal/controller/drain.go new file mode 100644 index 0000000..3425725 --- /dev/null +++ b/internal/controller/drain.go @@ -0,0 +1,425 @@ +/* +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" + "errors" + "fmt" + "slices" + + core "k8s.io/api/core/v1" + policy "k8s.io/api/policy/v1" + resv1 "k8s.io/api/resource/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/klog/v2" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// This file holds the node taint / pod drain primitives used when a controller has to clear a +// node before touching its GPUs. They are free functions taking an explicit client rather than +// reconciler methods, so more than one controller can call them; each keeps its own CR status +// bookkeeping and ctrl.Result handling, which is the part that genuinely differs. +// +// GPUFirmwareUpdate is the only caller today. The GPU recovery controller, which lands +// separately, is the reason the primitives are free functions instead of methods: it drains for a +// PCIe reset, and that is deliberately NOT the same drain as a firmware update's. What differs is +// only *which* pods have to leave the node: +// +// - A firmware write does not reset the bus, so only GPU pods are at risk. GPUFirmwareUpdate +// selects exactly those, with gpuPodsOnNode. +// - An SBR or slot reset can disturb the host itself, which would take an unrelated pod down +// just as hard as a GPU one. A reset therefore has to empty the node of everything evictable, +// with podsBlockingDrain. +// +// How they leave is shared: both go through the Eviction API (evictPods), so a workload owner's +// PodDisruptionBudget is honoured rather than silently broken by a node the operator decided to +// clear. +// +// Sharing the primitives and not the policy is the point: neither drain imposes its rules on the +// other, and the parts that are genuinely identical — is this taint already there, which pods are +// on this node, does this pod hold an Intel GPU — exist once. + +// +kubebuilder:rbac:groups="",resources=pods/eviction,verbs=create + +// drainAction is what a full-node drain should do about a particular pod. +type drainAction int + +const ( + // mirrorPodAnnotation marks a static pod's API-server mirror. Such a pod is owned by the + // kubelet's local manifest rather than by the API server: deleting or evicting the mirror + // does not stop the container and the kubelet recreates it immediately. + mirrorPodAnnotation = "kubernetes.io/config.mirror" + + // podNodeNameIndex is the field index podsOnNode selects on. Pods are cached cluster-wide (see + // ctrl.Options.Cache.ByObject in cmd/main.go). + podNodeNameIndex = "spec.nodeName" + + // drainIgnore means the pod neither needs evicting nor blocks the drain. + drainIgnore drainAction = iota + drainEvict + drainAwait +) + +// ensureNodeTaint adds taint to the named node if an identical taint is not already present. +// Returns added=false when the taint was already there, so the caller can tell a fresh taint +// from a repeat pass without re-reading the node. +// +// "Identical" means key, value and effect all match: the value carries the owning CR's name, +// so two CRs tainting the same node do not clobber each other's entry. +func ensureNodeTaint(ctx context.Context, c client.Client, nodeName string, taint core.Taint) (added bool, err error) { + node := &core.Node{} + + if err := c.Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil { + return false, fmt.Errorf("failed to get node %s: %w", nodeName, err) + } + + for i := range node.Spec.Taints { + if taintsEqual(node.Spec.Taints[i], taint) { + return false, nil + } + } + + node.Spec.Taints = append(node.Spec.Taints, taint) + + if err := c.Update(ctx, node); err != nil { + return false, fmt.Errorf("failed to update node %s with taint %s: %w", nodeName, taint.Key, err) + } + + return true, nil +} + +// removeNodeTaint drops taint from the named node. Returns removed=false when the node did +// not carry it, which is not an error: an admin may have removed it by hand, and the caller +// is only asking for the taint to be gone. +func removeNodeTaint(ctx context.Context, c client.Client, nodeName string, taint core.Taint) (removed bool, err error) { + node := &core.Node{} + + if err := c.Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil { + return false, fmt.Errorf("failed to get node %s: %w", nodeName, err) + } + + kept := make([]core.Taint, 0, len(node.Spec.Taints)) + + for i := range node.Spec.Taints { + if taintsEqual(node.Spec.Taints[i], taint) { + removed = true + + continue + } + + kept = append(kept, node.Spec.Taints[i]) + } + + if !removed { + return false, nil + } + + node.Spec.Taints = kept + + if err := c.Update(ctx, node); err != nil { + return false, fmt.Errorf("failed to remove taint %s from node %s: %w", taint.Key, nodeName, err) + } + + return true, nil +} + +// taintsEqual compares the three fields that identify a taint. TimeAdded is deliberately +// excluded: it is set by the API server on NoExecute taints, so comparing whole structs would +// make a taint we added ourselves look like a different one on the next pass. +func taintsEqual(a, b core.Taint) bool { + return a.Key == b.Key && a.Value == b.Value && a.Effect == b.Effect +} + +// nodesWithTaint returns the names of all nodes currently carrying taint. Reconciling taints +// against the cluster rather than against a list kept in CR status is what makes a taint +// outlive a lost status write: it is still found, and still cleaned up. +func nodesWithTaint(ctx context.Context, c client.Client, taint core.Taint) ([]string, error) { + nodeList := &core.NodeList{} + + if err := c.List(ctx, nodeList); err != nil { + return nil, fmt.Errorf("failed to list nodes: %w", err) + } + + var tainted []string + + for i := range nodeList.Items { + for j := range nodeList.Items[i].Spec.Taints { + if taintsEqual(nodeList.Items[i].Spec.Taints[j], taint) { + tainted = append(tainted, nodeList.Items[i].Name) + + break + } + } + } + + return tainted, nil +} + +// indexPodByNodeName is the index function behind podNodeNameIndex. An unscheduled pod is on no +// node and is left out of the index entirely, so it cannot match a node name. +func indexPodByNodeName(obj client.Object) []string { + pod, ok := obj.(*core.Pod) + if !ok || pod.Spec.NodeName == "" { + return nil + } + + return []string{pod.Spec.NodeName} +} + +// SetupDrainIndices registers the cache indexes the drain primitives need. Call it once per +// manager, from main, before any controller that drains is set up. +func SetupDrainIndices(ctx context.Context, mgr ctrl.Manager) error { + if err := mgr.GetFieldIndexer().IndexField(ctx, &core.Pod{}, podNodeNameIndex, indexPodByNodeName); err != nil { + return fmt.Errorf("failed to register the %s pod index: %w", podNodeNameIndex, err) + } + + return nil +} + +// podsOnNode returns every cached pod assigned to the named node, across all namespaces. The +// filtering happens in the cache index, so the cost is proportional to the pods on that one node +// rather than to the size of the cluster. +func podsOnNode(ctx context.Context, c client.Client, nodeName string) ([]*core.Pod, error) { + var pods core.PodList + + listOpts := []client.ListOption{ + client.InNamespace(core.NamespaceAll), + client.MatchingFields{podNodeNameIndex: nodeName}, + } + + if err := c.List(ctx, &pods, listOpts...); err != nil { + return nil, fmt.Errorf("failed to list pods on node %s: %w", nodeName, err) + } + + onNode := make([]*core.Pod, 0, len(pods.Items)) + + for i := range pods.Items { + onNode = append(onNode, &pods.Items[i]) + } + + return onNode, nil +} + +// podClaimsIntelGPU reports whether any of a pod's resource claims (direct or via template) +// requests a device from the Intel GPU DRA device class. +func podClaimsIntelGPU(ctx context.Context, c client.Client, claims []core.PodResourceClaim, namespace string) (bool, error) { + for _, rc := range claims { + if rc.ResourceClaimName != nil { + resClaim := resv1.ResourceClaim{} + + if err := c.Get(ctx, client.ObjectKey{Name: *rc.ResourceClaimName, Namespace: namespace}, &resClaim); err != nil { + return false, fmt.Errorf("failed to get ResourceClaim %s: %w", *rc.ResourceClaimName, err) + } + + if requestsIntelGPU(resClaim.Spec.Devices.Requests) { + return true, nil + } + } + + if rc.ResourceClaimTemplateName != nil { + resClaimTmpl := resv1.ResourceClaimTemplate{} + + if err := c.Get(ctx, client.ObjectKey{Name: *rc.ResourceClaimTemplateName, Namespace: namespace}, &resClaimTmpl); err != nil { + return false, fmt.Errorf("failed to get ResourceClaimTemplate %s: %w", *rc.ResourceClaimTemplateName, err) + } + + if requestsIntelGPU(resClaimTmpl.Spec.Spec.Devices.Requests) { + return true, nil + } + } + } + + return false, nil +} + +// requestsIntelGPU reports whether any device request — including the alternatives in a +// firstAvailable list — names the Intel GPU device class. +func requestsIntelGPU(requests []resv1.DeviceRequest) bool { + for _, req := range requests { + if req.Exactly != nil && req.Exactly.DeviceClassName == gpuDraDeviceClass { + return true + } + + for _, fa := range req.FirstAvailable { + if fa.DeviceClassName == gpuDraDeviceClass { + return true + } + } + } + + return false +} + +// gpuPodsOnNode returns the pods on a node that hold an Intel GPU, either through the device +// plugin's extended resources or through a DRA claim. +func gpuPodsOnNode(ctx context.Context, c client.Client, nodeName string) ([]*core.Pod, error) { + pods, err := podsOnNode(ctx, c, nodeName) + if err != nil { + return nil, err + } + + gpuPods := []*core.Pod{} + + for _, pod := range pods { + include := false + + // Extended GPU resources requested by any container. + for _, cnt := range pod.Spec.Containers { + if _, found := cnt.Resources.Limits[xeResource]; found { + include = true + } else if _, found := cnt.Resources.Limits[i915Resource]; found { + include = true + } + } + + if !include && len(pod.Spec.ResourceClaims) > 0 { + isGPU, err := podClaimsIntelGPU(ctx, c, pod.Spec.ResourceClaims, pod.Namespace) + if err != nil { + return nil, fmt.Errorf("failed to check ResourceClaims for pod %s: %w", pod.Name, err) + } + + include = isGPU + } + + if include { + gpuPods = append(gpuPods, pod) + } + } + + return gpuPods, nil +} + +// drainNeverEvicts reports whether a full-node drain leaves this pod on the node *permanently*, +// and why. +func drainNeverEvicts(pod *core.Pod, operatorNamespace string, skipNamespaces []string) (string, bool) { + if pod.Namespace == operatorNamespace { + return "operator namespace", true + } + + if slices.Contains(skipNamespaces, pod.Namespace) { + return "namespace in namespacesToSkip", true + } + + if _, isMirror := pod.Annotations[mirrorPodAnnotation]; isMirror { + return "static pod", true + } + + for i := range pod.OwnerReferences { + if pod.OwnerReferences[i].Kind == "DaemonSet" { + return "DaemonSet pod", true + } + } + + return "", false +} + +// classifyPodForDrain decides how a full-node drain should treat one pod, returning the +// action and a short human-readable reason for it. +func classifyPodForDrain(pod *core.Pod, operatorNamespace string, skipNamespaces []string) (drainAction, string) { + if reason, never := drainNeverEvicts(pod, operatorNamespace, skipNamespaces); never { + return drainIgnore, reason + } + + if pod.Status.Phase == core.PodSucceeded || pod.Status.Phase == core.PodFailed { + return drainIgnore, fmt.Sprintf("already %s", pod.Status.Phase) + } + + // Checked after the skip rules so that a terminating DaemonSet or static pod is not + // waited on: those are recreated by design and would block forever. + if pod.DeletionTimestamp != nil { + return drainAwait, "already terminating" + } + + return drainEvict, "evictable" +} + +// podsBlockingDrain splits the pods on a node into those a drain must evict and those it +// merely has to wait for. A node is drained when both lists are empty. +func podsBlockingDrain(ctx context.Context, c client.Client, nodeName, operatorNamespace string, + skipNamespaces []string) (toEvict, toAwait []*core.Pod, err error) { + pods, err := podsOnNode(ctx, c, nodeName) + if err != nil { + return nil, nil, err + } + + for _, pod := range pods { + action, reason := classifyPodForDrain(pod, operatorNamespace, skipNamespaces) + + switch action { + case drainEvict: + toEvict = append(toEvict, pod) + case drainAwait: + toAwait = append(toAwait, pod) + case drainIgnore: + klog.V(2).Infof("drain of node %s ignoring pod %s/%s (%s)", nodeName, pod.Namespace, pod.Name, reason) + } + } + + return toEvict, toAwait, nil +} + +// evictPods evicts every pod in pods, and is the step both drains share. It is safe to call on +// every pass: a pod already on its way out is skipped rather than evicted a second time, so a +// drain that is simply taking its time costs no API calls at all. +func evictPods(ctx context.Context, c client.Client, pods []*core.Pod) error { + var errs []error + + for _, pod := range pods { + if pod.DeletionTimestamp != nil { + continue + } + + err := evictPod(ctx, c, pod) + + switch { + case err == nil: + klog.Infof("Evicted pod %s/%s from node %s", pod.Namespace, pod.Name, pod.Spec.NodeName) + case apierrors.IsNotFound(err): + // The pod went away between the List and the eviction, which is the outcome we + // wanted. Losing that race must not fail the drain. + klog.V(2).Infof("Pod %s/%s already gone", pod.Namespace, pod.Name) + case apierrors.IsTooManyRequests(err): + klog.V(2).Infof("Eviction of pod %s/%s deferred by a PodDisruptionBudget, will retry: %v", + pod.Namespace, pod.Name, err) + default: + errs = append(errs, err) + } + } + + return errors.Join(errs...) +} + +// evictPod requests eviction of a pod through the policy/v1 eviction subresource rather than +// deleting it outright, so that PodDisruptionBudgets are honoured and the workload owner's +// availability guarantees are not silently broken by a node drain. +func evictPod(ctx context.Context, c client.Client, pod *core.Pod) error { + eviction := &policy.Eviction{ + ObjectMeta: metav1.ObjectMeta{ + Name: pod.Name, + Namespace: pod.Namespace, + }, + } + + if err := c.SubResource("eviction").Create(ctx, pod, eviction); err != nil { + return fmt.Errorf("failed to evict pod %s/%s: %w", pod.Namespace, pod.Name, err) + } + + return nil +} diff --git a/internal/controller/drain_test.go b/internal/controller/drain_test.go new file mode 100644 index 0000000..a514bb8 --- /dev/null +++ b/internal/controller/drain_test.go @@ -0,0 +1,799 @@ +/* +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" + "errors" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + core "k8s.io/api/core/v1" + resv1 "k8s.io/api/resource/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +// The primitives in drain.go are exercised against a fake client rather than envtest: every one +// of them is a pure function of what the API returns, and the fake client is the only way to +// provoke the List/Get/Update failures the error paths exist for. +// +// One thing the fake client does not model is the PodDisruptionBudget check behind the eviction +// subresource — it deletes the pod unconditionally. The eviction specs below therefore pin that +// evictPod goes through the subresource at all (rather than deleting outright, which would ignore +// PDBs), not what the API server does with a budget that forbids it. + +func drainScheme() *runtime.Scheme { + s := runtime.NewScheme() + Expect(core.AddToScheme(s)).To(Succeed()) + Expect(resv1.AddToScheme(s)).To(Succeed()) + + return s +} + +// drainClientBuilder is a fake client builder carrying the same spec.nodeName pod index that +// SetupDrainIndexes registers on a real manager's cache, so podsOnNode behaves here as it does in +// the cluster — and the index function itself is covered by every spec that lists pods. +func drainClientBuilder() *fake.ClientBuilder { + return fake.NewClientBuilder().WithScheme(drainScheme()). + WithIndex(&core.Pod{}, podNodeNameIndex, indexPodByNodeName) +} + +// drainTestNode is a node carrying the given taints. +func drainTestNode(name string, taints ...core.Taint) *core.Node { + return &core.Node{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: core.NodeSpec{Taints: taints}, + } +} + +// drainTestPod is a running pod on a node, with no GPU of any kind. +func drainTestPod(name, namespace, nodeName string) *core.Pod { + return &core.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Spec: core.PodSpec{ + NodeName: nodeName, + Containers: []core.Container{{Name: "c", Image: "busybox"}}, + }, + Status: core.PodStatus{Phase: core.PodRunning}, + } +} + +// withGPUResource adds an extended-resource GPU request, the device plugin's way of holding a GPU. +func withGPUResource(pod *core.Pod, name core.ResourceName) *core.Pod { + count, err := resource.ParseQuantity("1") + Expect(err).NotTo(HaveOccurred()) + + pod.Spec.Containers[0].Resources.Limits = core.ResourceList{name: count} + + return pod +} + +// withClaim references a ResourceClaim by name, DRA's way of holding a GPU. +func withClaim(pod *core.Pod, claimName string) *core.Pod { + pod.Spec.ResourceClaims = []core.PodResourceClaim{ + {Name: "gpu", ResourceClaimName: &claimName}, + } + + return pod +} + +// withClaimTemplate references a ResourceClaimTemplate, the other DRA shape. +func withClaimTemplate(pod *core.Pod, templateName string) *core.Pod { + pod.Spec.ResourceClaims = []core.PodResourceClaim{ + {Name: "gpu", ResourceClaimTemplateName: &templateName}, + } + + return pod +} + +// gpuClaim is a ResourceClaim requesting a device from the given device class. +func gpuClaim(name, namespace, deviceClass string) *resv1.ResourceClaim { + return &resv1.ResourceClaim{ + ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: namespace}, + Spec: resv1.ResourceClaimSpec{ + Devices: resv1.DeviceClaim{ + Requests: []resv1.DeviceRequest{ + { + Name: "req", + Exactly: &resv1.ExactDeviceRequest{DeviceClassName: deviceClass}, + }, + }, + }, + }, + } +} + +var _ = Describe("Drain primitives", func() { + ctx := context.Background() + + noSchedule := func(key, value string) core.Taint { + return core.Taint{Key: key, Value: value, Effect: core.TaintEffectNoSchedule} + } + + Context("taintsEqual", func() { + // The whole reason this helper exists instead of == on the struct. + It("should ignore TimeAdded", func() { + now := metav1.NewTime(time.Now()) + + a := noSchedule("k", "v") + b := noSchedule("k", "v") + b.TimeAdded = &now + + Expect(taintsEqual(a, b)).To(BeTrue(), + "the API server sets TimeAdded itself, so comparing it would make our own taint "+ + "look foreign on the next pass and it would be added a second time") + }) + + It("should distinguish taints by value", func() { + // The value carries the owning CR's name, which is what keeps two CRs tainting the + // same node from clobbering each other's entry. + Expect(taintsEqual(noSchedule("k", "cr-a"), noSchedule("k", "cr-b"))).To(BeFalse()) + }) + + It("should distinguish taints by effect", func() { + a := noSchedule("k", "v") + b := a + b.Effect = core.TaintEffectNoExecute + + Expect(taintsEqual(a, b)).To(BeFalse()) + }) + }) + + Context("ensureNodeTaint", func() { + taint := noSchedule("gpu-update-in-progress", "fu-1") + + It("should add a taint the node does not have", func() { + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1")).Build() + + added, err := ensureNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(added).To(BeTrue()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(ConsistOf(taint)) + }) + + It("should report added=false and not duplicate an existing taint", func() { + // This is what makes the taint step idempotent across requeues: a repeat pass has to + // be distinguishable from a first one without re-reading the node. + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1", taint)).Build() + + added, err := ensureNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(added).To(BeFalse()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(HaveLen(1)) + }) + + It("should not re-add a taint that only differs in TimeAdded", func() { + now := metav1.NewTime(time.Now()) + stamped := taint + stamped.TimeAdded = &now + + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1", stamped)).Build() + + added, err := ensureNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(added).To(BeFalse()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(HaveLen(1), "a stamped taint is still our taint") + }) + + It("should keep taints set by somebody else", func() { + foreign := noSchedule("other.example.com/thing", "x") + + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1", foreign)).Build() + + _, err := ensureNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(ConsistOf(foreign, taint)) + }) + + It("should fail when the node is gone", func() { + c := drainClientBuilder().Build() + + added, err := ensureNodeTaint(ctx, c, "no-such-node", taint) + Expect(err).To(HaveOccurred()) + Expect(added).To(BeFalse()) + }) + }) + + Context("removeNodeTaint", func() { + taint := noSchedule("gpu-update-in-progress", "fu-1") + + It("should remove the taint and leave the others", func() { + foreign := noSchedule("other.example.com/thing", "x") + + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1", foreign, taint)).Build() + + removed, err := removeNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(removed).To(BeTrue()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(ConsistOf(foreign)) + }) + + It("should remove a taint the API server stamped with TimeAdded", func() { + // The bug this replaced: comparing whole structs made a stamped taint unfindable, so + // the node was reported as having lost the taint while still carrying it. + now := metav1.NewTime(time.Now()) + stamped := taint + stamped.TimeAdded = &now + + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1", stamped)).Build() + + removed, err := removeNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(removed).To(BeTrue()) + + node := &core.Node{} + Expect(c.Get(ctx, types.NamespacedName{Name: "node-1"}, node)).To(Succeed()) + Expect(node.Spec.Taints).To(BeEmpty()) + }) + + It("should report removed=false rather than an error when the taint is absent", func() { + // An admin removing it by hand is not a failure: the caller only wants it gone. + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1")).Build() + + removed, err := removeNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + Expect(removed).To(BeFalse()) + }) + + It("should not write to the node when there is nothing to remove", func() { + c := drainClientBuilder(). + WithObjects(drainTestNode("node-1")). + WithInterceptorFuncs(interceptor.Funcs{ + Update: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.UpdateOption) error { + Fail("removeNodeTaint must not update a node it changed nothing on") + + return nil + }, + }).Build() + + _, err := removeNodeTaint(ctx, c, "node-1", taint) + Expect(err).NotTo(HaveOccurred()) + }) + }) + + Context("nodesWithTaint", func() { + taint := noSchedule("gpurecovery", "plan-1") + + It("should find every node carrying the taint and no others", func() { + // Asking the cluster rather than a list in CR status is what lets a taint survive a + // lost status write and still be cleaned up. + c := drainClientBuilder().WithObjects( + drainTestNode("tainted-1", taint), + drainTestNode("clean"), + drainTestNode("tainted-2", noSchedule("unrelated", "y"), taint), + drainTestNode("other-value", noSchedule("gpurecovery", "plan-2")), + ).Build() + + names, err := nodesWithTaint(ctx, c, taint) + Expect(err).NotTo(HaveOccurred()) + Expect(names).To(ConsistOf("tainted-1", "tainted-2")) + }) + + It("should return nothing when no node carries it", func() { + c := drainClientBuilder(). + WithObjects(drainTestNode("clean")).Build() + + names, err := nodesWithTaint(ctx, c, taint) + Expect(err).NotTo(HaveOccurred()) + Expect(names).To(BeEmpty()) + }) + }) + + Context("podsOnNode", func() { + It("should return the pods on that node across all namespaces", func() { + c := drainClientBuilder().WithObjects( + drainTestPod("a", "default", "node-1"), + drainTestPod("b", "kube-system", "node-1"), + drainTestPod("c", "default", "node-2"), + ).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + + names := []string{} + for _, pod := range pods { + names = append(names, pod.Namespace+"/"+pod.Name) + } + + Expect(names).To(ConsistOf("default/a", "kube-system/b")) + }) + + It("should skip unscheduled pods", func() { + // A pod with an empty spec.nodeName is on no node, and must not match one. + c := drainClientBuilder().WithObjects( + drainTestPod("pending", "default", ""), + ).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(BeEmpty()) + }) + + It("should return an empty list, not an error, for a node with no pods", func() { + c := drainClientBuilder().Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(BeEmpty()) + }) + }) + + Context("requestsIntelGPU", func() { + It("should match an exact request for the Intel GPU device class", func() { + Expect(requestsIntelGPU([]resv1.DeviceRequest{ + {Exactly: &resv1.ExactDeviceRequest{DeviceClassName: gpuDraDeviceClass}}, + })).To(BeTrue()) + }) + + It("should match an Intel GPU among firstAvailable alternatives", func() { + // A firstAvailable list is a set of alternatives, any one of which may end up + // allocated, so a single Intel GPU entry makes the pod a GPU pod. + Expect(requestsIntelGPU([]resv1.DeviceRequest{ + {FirstAvailable: []resv1.DeviceSubRequest{ + {DeviceClassName: "other.example.com"}, + {DeviceClassName: gpuDraDeviceClass}, + }}, + })).To(BeTrue()) + }) + + It("should not match another vendor's device class", func() { + Expect(requestsIntelGPU([]resv1.DeviceRequest{ + {Exactly: &resv1.ExactDeviceRequest{DeviceClassName: "other.example.com"}}, + })).To(BeFalse()) + }) + + It("should not match an empty request list", func() { + Expect(requestsIntelGPU(nil)).To(BeFalse()) + }) + }) + + Context("gpuPodsOnNode", func() { + It("should include pods holding a GPU through either driver's extended resource", func() { + c := drainClientBuilder().WithObjects( + withGPUResource(drainTestPod("xe", "default", "node-1"), xeResource), + withGPUResource(drainTestPod("i915", "default", "node-1"), i915Resource), + drainTestPod("plain", "default", "node-1"), + ).Build() + + pods, err := gpuPodsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + + names := []string{} + for _, pod := range pods { + names = append(names, pod.Name) + } + + Expect(names).To(ConsistOf("xe", "i915")) + }) + + It("should include a pod holding a GPU through a ResourceClaim", func() { + c := drainClientBuilder().WithObjects( + withClaim(drainTestPod("dra", "default", "node-1"), "claim-gpu"), + gpuClaim("claim-gpu", "default", gpuDraDeviceClass), + ).Build() + + pods, err := gpuPodsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(HaveLen(1)) + Expect(pods[0].Name).To(Equal("dra")) + }) + + It("should include a pod holding a GPU through a ResourceClaimTemplate", func() { + c := drainClientBuilder().WithObjects( + withClaimTemplate(drainTestPod("dra-tmpl", "default", "node-1"), "tmpl-gpu"), + &resv1.ResourceClaimTemplate{ + ObjectMeta: metav1.ObjectMeta{Name: "tmpl-gpu", Namespace: "default"}, + Spec: resv1.ResourceClaimTemplateSpec{ + Spec: gpuClaim("ignored", "default", gpuDraDeviceClass).Spec, + }, + }, + ).Build() + + pods, err := gpuPodsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(HaveLen(1)) + Expect(pods[0].Name).To(Equal("dra-tmpl")) + }) + + It("should exclude a pod whose claim is for another vendor's devices", func() { + c := drainClientBuilder().WithObjects( + withClaim(drainTestPod("other-vendor", "default", "node-1"), "claim-other"), + gpuClaim("claim-other", "default", "other.example.com"), + ).Build() + + pods, err := gpuPodsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(BeEmpty()) + }) + + It("should fail when a referenced claim cannot be read", func() { + // Guessing "not a GPU pod" here would drop a GPU pod from the eviction list and let + // a firmware update start underneath a running workload. + c := drainClientBuilder().WithObjects( + withClaim(drainTestPod("dangling", "default", "node-1"), "missing-claim"), + ).Build() + + _, err := gpuPodsOnNode(ctx, c, "node-1") + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("dangling")) + }) + }) + + Context("classifyPodForDrain", func() { + const operatorNS = "operator-ns" + + podIn := func(namespace string) *core.Pod { + return drainTestPod("p", namespace, "node-1") + } + + It("should evict a pod in a namespace nobody asked to skip", func() { + action, reason := classifyPodForDrain(podIn("team-a"), operatorNS, []string{"kube-system"}) + Expect(action).To(Equal(drainEvict)) + Expect(reason).To(Equal("evictable")) + }) + + It("should ignore a pod in a listed namespace", func() { + action, reason := classifyPodForDrain(podIn("kube-system"), operatorNS, []string{"cert-manager", "kube-system"}) + Expect(action).To(Equal(drainIgnore)) + Expect(reason).To(ContainSubstring("namespacesToSkip"), + "the log line is how an admin finds out why a pod was left behind") + }) + + // The operator namespace skip is not a default an admin can replace by supplying a list of + // their own: evicting there aborts the reconcile driving the drain and kills any Job the + // operator is running on the node. + It("should ignore the operator namespace even when a skip list is set", func() { + action, reason := classifyPodForDrain(podIn(operatorNS), operatorNS, []string{"kube-system"}) + Expect(action).To(Equal(drainIgnore)) + Expect(reason).To(Equal("operator namespace")) + }) + + It("should ignore a DaemonSet pod", func() { + // It gets an automatic NoSchedule toleration, so evicting it brings it straight back + // and the drain would never converge. + pod := podIn("team-a") + pod.OwnerReferences = []metav1.OwnerReference{{Kind: "DaemonSet", Name: "ds"}} + + action, reason := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainIgnore)) + Expect(reason).To(Equal("DaemonSet pod")) + }) + + It("should still evict a pod owned by something other than a DaemonSet", func() { + pod := podIn("team-a") + pod.OwnerReferences = []metav1.OwnerReference{{Kind: "ReplicaSet", Name: "rs"}} + + action, _ := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainEvict)) + }) + + It("should ignore a static pod's mirror", func() { + // Evicting the mirror stops nothing: the kubelet owns the manifest and recreates it. + pod := podIn("kube-system") + pod.Annotations = map[string]string{mirrorPodAnnotation: "abc"} + + action, reason := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainIgnore)) + Expect(reason).To(Equal("static pod")) + }) + + DescribeTable("should ignore a pod that has already finished", + func(phase core.PodPhase) { + pod := podIn("team-a") + pod.Status.Phase = phase + + action, reason := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainIgnore)) + Expect(reason).To(ContainSubstring("already")) + }, + Entry("Succeeded", core.PodSucceeded), + Entry("Failed", core.PodFailed), + ) + + It("should await a terminating pod rather than evict it again", func() { + pod := podIn("team-a") + now := metav1.NewTime(time.Now()) + pod.DeletionTimestamp = &now + + action, reason := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainAwait)) + Expect(reason).To(Equal("already terminating")) + }) + + // Order matters: a terminating DaemonSet pod is recreated by design, so awaiting it would + // block the drain forever. The permanent-exclusion rules have to be checked first. + It("should ignore a terminating DaemonSet pod instead of awaiting it", func() { + pod := podIn("team-a") + now := metav1.NewTime(time.Now()) + pod.DeletionTimestamp = &now + pod.OwnerReferences = []metav1.OwnerReference{{Kind: "DaemonSet", Name: "ds"}} + + action, _ := classifyPodForDrain(pod, operatorNS, nil) + Expect(action).To(Equal(drainIgnore)) + }) + }) + + Context("drainNeverEvicts", func() { + const operatorNS = "operator-ns" + + // The deliberate difference from classifyPodForDrain: a Succeeded or terminating pod is + // not evicted, but it does clear by itself, so a DRA claim held by one is worth waiting + // for. Folding the two predicates together would make such a wait give up early. + It("should not report a finished pod as a permanent occupant", func() { + pod := drainTestPod("p", "team-a", "node-1") + pod.Status.Phase = core.PodSucceeded + + _, never := drainNeverEvicts(pod, operatorNS, nil) + Expect(never).To(BeFalse()) + }) + + It("should not report a terminating pod as a permanent occupant", func() { + pod := drainTestPod("p", "team-a", "node-1") + now := metav1.NewTime(time.Now()) + pod.DeletionTimestamp = &now + + _, never := drainNeverEvicts(pod, operatorNS, nil) + Expect(never).To(BeFalse()) + }) + + It("should report a DaemonSet pod as a permanent occupant", func() { + pod := drainTestPod("p", "team-a", "node-1") + pod.OwnerReferences = []metav1.OwnerReference{{Kind: "DaemonSet", Name: "ds"}} + + _, never := drainNeverEvicts(pod, operatorNS, nil) + Expect(never).To(BeTrue()) + }) + }) + + Context("podsBlockingDrain", func() { + const operatorNS = "operator-ns" + + It("should split the node's pods into evict, await and ignore", func() { + terminating := drainTestPod("terminating", "team-a", "node-1") + terminating.Finalizers = []string{"test.example.com/hold"} + + daemon := drainTestPod("daemon", "team-a", "node-1") + daemon.OwnerReferences = []metav1.OwnerReference{{Kind: "DaemonSet", Name: "ds"}} + + c := drainClientBuilder().WithObjects( + drainTestPod("workload", "team-a", "node-1"), + drainTestPod("operator", operatorNS, "node-1"), + drainTestPod("skipped", "kube-system", "node-1"), + drainTestPod("elsewhere", "team-a", "node-2"), + daemon, + terminating, + ).Build() + + // The finalizer keeps the pod around with a deletionTimestamp, which is what a pod + // mid-eviction looks like to a later pass. + Expect(c.Delete(ctx, terminating)).To(Succeed()) + + toEvict, toAwait, err := podsBlockingDrain(ctx, c, "node-1", operatorNS, []string{"kube-system"}) + Expect(err).NotTo(HaveOccurred()) + + evictNames := []string{} + for _, pod := range toEvict { + evictNames = append(evictNames, pod.Name) + } + + awaitNames := []string{} + for _, pod := range toAwait { + awaitNames = append(awaitNames, pod.Name) + } + + Expect(evictNames).To(ConsistOf("workload")) + Expect(awaitNames).To(ConsistOf("terminating")) + }) + + It("should report both lists empty for a node that is already drained", func() { + // Both empty is the drained condition, so an all-ignored node must not read as busy. + daemon := drainTestPod("daemon", "team-a", "node-1") + daemon.OwnerReferences = []metav1.OwnerReference{{Kind: "DaemonSet", Name: "ds"}} + + c := drainClientBuilder().WithObjects( + daemon, + drainTestPod("operator", operatorNS, "node-1"), + ).Build() + + toEvict, toAwait, err := podsBlockingDrain(ctx, c, "node-1", operatorNS, nil) + Expect(err).NotTo(HaveOccurred()) + Expect(toEvict).To(BeEmpty()) + Expect(toAwait).To(BeEmpty()) + }) + }) + + Context("evictPods", func() { + It("should evict every pod given to it", func() { + c := drainClientBuilder().WithObjects( + drainTestPod("a", "team-a", "node-1"), + drainTestPod("b", "team-b", "node-1"), + ).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(evictPods(ctx, c, pods)).To(Succeed()) + + remaining := &core.PodList{} + Expect(c.List(ctx, remaining)).To(Succeed()) + Expect(remaining.Items).To(BeEmpty()) + }) + + It("should not evict a pod that is already terminating", func() { + // A drain that is merely slow must cost nothing: the pod is on its way out and + // re-evicting it would be a pointless API call on every poll. + pod := drainTestPod("going", "team-a", "node-1") + pod.Finalizers = []string{"test.example.com/hold"} + + c := drainClientBuilder().WithObjects(pod). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourceCreate: func(_ context.Context, _ client.Client, _ string, + _ client.Object, _ client.Object, _ ...client.SubResourceCreateOption) error { + Fail("evictPods must skip a pod that already has a deletionTimestamp") + + return nil + }, + }).Build() + + Expect(c.Delete(ctx, pod)).To(Succeed()) + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + Expect(pods).To(HaveLen(1)) + Expect(evictPods(ctx, c, pods)).To(Succeed()) + }) + + // The fake client does not implement PodDisruptionBudgets, so the 429 the API server + // would send is injected. This is the case that made the polling retry necessary: the + // first attempt is refused and a later one succeeds. + It("should treat a PodDisruptionBudget refusal as 'not yet', not a failure", func() { + c := drainClientBuilder(). + WithObjects(drainTestPod("protected", "team-a", "node-1")). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourceCreate: func(_ context.Context, _ client.Client, _ string, + _ client.Object, _ client.Object, _ ...client.SubResourceCreateOption) error { + return apierrors.NewTooManyRequests("cannot evict, disruption budget is exhausted", 10) + }, + }).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + + Expect(evictPods(ctx, c, pods)).To(Succeed(), + "a refusal from a PDB is retried on the next pass, so failing the drain here "+ + "would abort an update that was going to succeed") + + Expect(c.Get(ctx, types.NamespacedName{Name: "protected", Namespace: "team-a"}, &core.Pod{})). + To(Succeed(), "the pod is still there, so the caller keeps the node in draining") + }) + + It("should not fail when the pod disappeared before it could be evicted", func() { + // Lost race between the List and the eviction. The pod being gone is what we wanted. + c := drainClientBuilder().Build() + + Expect(evictPods(ctx, c, []*core.Pod{drainTestPod("ghost", "team-a", "node-1")})).To(Succeed()) + }) + + It("should return a real failure so a drain that can never work surfaces", func() { + // Missing RBAC on pods/eviction looks like this. Swallowing it would leave the CR + // sitting in draining forever with nothing to point at. + c := drainClientBuilder(). + WithObjects(drainTestPod("victim", "team-a", "node-1")). + WithInterceptorFuncs(interceptor.Funcs{ + SubResourceCreate: func(_ context.Context, _ client.Client, _ string, + _ client.Object, _ client.Object, _ ...client.SubResourceCreateOption) error { + return apierrors.NewForbidden(schema.GroupResource{Resource: "pods"}, "victim", + errors.New("no permission")) + }, + }).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + + err = evictPods(ctx, c, pods) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("team-a/victim")) + }) + + It("should try every pod even when one of them fails", func() { + // One broken pod must not shield the rest: the node has to end up clear. + attempted := []string{} + + c := drainClientBuilder().WithObjects( + drainTestPod("bad", "team-a", "node-1"), + drainTestPod("good", "team-a", "node-1"), + ).WithInterceptorFuncs(interceptor.Funcs{ + SubResourceCreate: func(_ context.Context, _ client.Client, _ string, + obj client.Object, _ client.Object, _ ...client.SubResourceCreateOption) error { + attempted = append(attempted, obj.GetName()) + + if obj.GetName() == "bad" { + return apierrors.NewInternalError(errors.New("boom")) + } + + return nil + }, + }).Build() + + pods, err := podsOnNode(ctx, c, "node-1") + Expect(err).NotTo(HaveOccurred()) + + Expect(evictPods(ctx, c, pods)).To(HaveOccurred()) + Expect(attempted).To(ConsistOf("bad", "good")) + }) + }) + + Context("evictPod", func() { + It("should evict through the eviction subresource", func() { + pod := drainTestPod("victim", "team-a", "node-1") + + c := drainClientBuilder().WithObjects(pod). + WithInterceptorFuncs(interceptor.Funcs{ + Delete: func(_ context.Context, _ client.WithWatch, _ client.Object, _ ...client.DeleteOption) error { + Fail("evictPod must go through pods/eviction, not Delete, or PodDisruptionBudgets are bypassed") + + return nil + }, + }).Build() + + Expect(evictPod(ctx, c, pod)).To(Succeed()) + + err := c.Get(ctx, types.NamespacedName{Name: "victim", Namespace: "team-a"}, &core.Pod{}) + Expect(apierrors.IsNotFound(err)).To(BeTrue()) + }) + + It("should wrap the failure with the pod's identity", func() { + // The error travels into CR status, where "which pod" is the only useful part. + c := drainClientBuilder().Build() + + err := evictPod(ctx, c, drainTestPod("victim", "team-a", "node-1")) + Expect(err).To(HaveOccurred()) + Expect(err.Error()).To(ContainSubstring("team-a/victim")) + }) + }) +}) diff --git a/internal/controller/gpufirmwareupdate_controller.go b/internal/controller/gpufirmwareupdate_controller.go index 07c08f9..a03655a 100644 --- a/internal/controller/gpufirmwareupdate_controller.go +++ b/internal/controller/gpufirmwareupdate_controller.go @@ -26,7 +26,6 @@ import ( batch "k8s.io/api/batch/v1" core "k8s.io/api/core/v1" - resv1 "k8s.io/api/resource/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -207,6 +206,11 @@ func (r *GPUFirmwareUpdateReconciler) verifyGivenParameters(fu *intelcomv1alpha1 return nil } +// taintAndDrainNodes taints each node and evicts the GPU pods on it, returning the nodes +// that still had GPU pods and are therefore draining. +// +// This is the first eviction attempt, not the only one: checkForNodeDrainStatus retries on every +// poll, because a PodDisruptionBudget can refuse an eviction that later succeeds. func (r *GPUFirmwareUpdateReconciler) taintAndDrainNodes(ctx context.Context, fu *intelcomv1alpha1.GPUFirmwareUpdate, nodesToProcess []string) ([]string, error) { taintToBeAdded := taintTemplate taintToBeAdded.Key = fu.Spec.UpdateTaint @@ -218,28 +222,19 @@ func (r *GPUFirmwareUpdateReconciler) taintAndDrainNodes(ctx context.Context, fu for _, nodeName := range nodesToProcess { klog.Infof("Selected node \"%s\" for firmware update", nodeName) - node := &core.Node{} - - if err := r.Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil { - lastErr = fmt.Errorf("failed to get node %s: %v", nodeName, err) + added, err := ensureNodeTaint(ctx, r.Client, nodeName, taintToBeAdded) + if err != nil { + lastErr = err break } - if slices.Contains(node.Spec.Taints, taintToBeAdded) { + if !added { klog.Infof("Node \"%s\" already has taint %s, skipping\n", nodeName, fu.Spec.UpdateTaint) - } else { - node.Spec.Taints = append(node.Spec.Taints, taintToBeAdded) - - if err := r.Update(ctx, node); err != nil { - lastErr = fmt.Errorf("failed to update node %s with taint: %v", nodeName, err) - - break - } } - if toEvict, err := r.getGPUPodsForNode(ctx, nodeName); err != nil { - lastErr = fmt.Errorf("failed to get GPU pods for node %s: %v", nodeName, err) + if toEvict, err := gpuPodsOnNode(ctx, r.Client, nodeName); err != nil { + lastErr = fmt.Errorf("failed to get GPU pods for node %s: %w", nodeName, err) } else { klog.Infof("Evicting %d pods from node %s\n", len(toEvict), nodeName) @@ -247,16 +242,8 @@ func (r *GPUFirmwareUpdateReconciler) taintAndDrainNodes(ctx context.Context, fu draining = append(draining, nodeName) } - for _, pod := range toEvict { - if pod.DeletionTimestamp != nil { - continue - } - - if err := r.Delete(ctx, pod); err != nil { - klog.Errorf("Failed to delete pod %s from node %s: %v", pod.Name, nodeName, err) - } else { - klog.Infof("Evicted pod %s from node %s", pod.Name, nodeName) - } + if err := evictPods(ctx, r.Client, toEvict); err != nil { + lastErr = fmt.Errorf("failed to evict GPU pods from node %s: %w", nodeName, err) } } } @@ -323,95 +310,6 @@ func (r *GPUFirmwareUpdateReconciler) selectNodesToUpdate(ctx context.Context, f return selected } -func (r *GPUFirmwareUpdateReconciler) checkClaimForIntelGPURequests(ctx context.Context, claims []core.PodResourceClaim, namespace string) (bool, error) { - for _, rc := range claims { - if rc.ResourceClaimName != nil { - resClaim := resv1.ResourceClaim{} - - if err := r.Get(ctx, client.ObjectKey{Name: *rc.ResourceClaimName, Namespace: namespace}, &resClaim); err != nil { - return false, fmt.Errorf("failed to get ResourceClaim %s: %v", *rc.ResourceClaimName, err) - } - - for _, req := range resClaim.Spec.Devices.Requests { - if req.Exactly != nil && req.Exactly.DeviceClassName == gpuDraDeviceClass { - return true, nil - } - for _, fa := range req.FirstAvailable { - if fa.DeviceClassName == gpuDraDeviceClass { - return true, nil - } - } - } - } - if rc.ResourceClaimTemplateName != nil { - resClaimTmpl := resv1.ResourceClaimTemplate{} - - if err := r.Get(ctx, client.ObjectKey{Name: *rc.ResourceClaimTemplateName, Namespace: namespace}, &resClaimTmpl); err != nil { - return false, fmt.Errorf("failed to get ResourceClaimTemplate %s: %v", *rc.ResourceClaimTemplateName, err) - } - - for _, req := range resClaimTmpl.Spec.Spec.Devices.Requests { - if req.Exactly != nil && req.Exactly.DeviceClassName == gpuDraDeviceClass { - return true, nil - } - for _, fa := range req.FirstAvailable { - if fa.DeviceClassName == gpuDraDeviceClass { - return true, nil - } - } - } - } - } - - return false, nil -} - -func (r *GPUFirmwareUpdateReconciler) getGPUPodsForNode(ctx context.Context, nodeName string) ([]*core.Pod, error) { - var pods core.PodList - - listOpts := []client.ListOption{ - client.InNamespace(core.NamespaceAll), // Search across all namespaces - client.MatchingFields{"spec.nodeName": nodeName}, - } - - if err := r.List(ctx, &pods, listOpts...); err != nil { - return nil, fmt.Errorf("failed to list pods on node %s: %v", nodeName, err) - } - - gpuPods := []*core.Pod{} - for index := range pods.Items { - pod := pods.Items[index] - - include := false - - // Check for extended GPU resources. - for _, c := range pod.Spec.Containers { - if _, found := c.Resources.Limits[xeResource]; found { - include = true - } else if _, found := c.Resources.Limits[i915Resource]; found { - include = true - } - } - - // Check for DRA resource claims. - if !include { - if len(pod.Spec.ResourceClaims) > 0 { - if isGpu, err := r.checkClaimForIntelGPURequests(ctx, pod.Spec.ResourceClaims, pod.Namespace); err != nil { - return nil, fmt.Errorf("failed to check ResourceClaims for pod %s: %v", pod.Name, err) - } else if isGpu { - include = true - } - } - } - - if include { - gpuPods = append(gpuPods, &pods.Items[index]) - } - } - - return gpuPods, nil -} - func (r *GPUFirmwareUpdateReconciler) verifyContentImage(ctx context.Context, fu *intelcomv1alpha1.GPUFirmwareUpdate) error { return r.imgVerify.Verify(ctx, &fu.Spec) } @@ -508,7 +406,7 @@ func (r *GPUFirmwareUpdateReconciler) checkForNodeDrainStatus(ctx context.Contex var lastErr error for _, nodeName := range fu.Status.NodeInfos.Draining { - pods, err := r.getGPUPodsForNode(ctx, nodeName) + pods, err := gpuPodsOnNode(ctx, r.Client, nodeName) if err != nil { lastErr = err @@ -517,6 +415,16 @@ func (r *GPUFirmwareUpdateReconciler) checkForNodeDrainStatus(ctx context.Contex if len(pods) != 0 { remaining = append(remaining, nodeName) + + // Retry the eviction rather than only observing it. A PodDisruptionBudget may have + // refused the first attempt, and nothing else would ever ask again. Pods that are + // already terminating are skipped inside evictPods, so a drain that is merely slow + // makes no API calls here. + if err := evictPods(ctx, r.Client, pods); err != nil { + lastErr = fmt.Errorf("failed to evict GPU pods from node %s: %w", nodeName, err) + + break + } } } @@ -873,28 +781,23 @@ func (r *GPUFirmwareUpdateReconciler) untaintNodesAndFinalize(ctx context.Contex var lastErr error for _, nodeName := range fu.Status.NodeInfos.All { - node := &core.Node{} - if err := r.Get(ctx, client.ObjectKey{Name: nodeName}, node); err != nil { - klog.Errorf("Failed to get node %s: %v", nodeName, err) + // removeNodeTaint matches on key/value/effect rather than on the whole struct. The + // previous slices.Index compared TimeAdded too, which the API server sets itself, so a + // taint could be reported as lost while it was still on the node. + removed, err := removeNodeTaint(ctx, r.Client, nodeName, taintToBeRemoved) + if err != nil { + klog.Errorf("Failed to remove taint from node %s: %v", nodeName, err) lastErr = err + continue } - tindex := slices.Index(node.Spec.Taints, taintToBeRemoved) - - if tindex < 0 { + if !removed { klog.Warningf("Node %s does not have taint %s, skipping\n", nodeName, fu.Spec.UpdateTaint) fu.Status.Messages = append(fu.Status.Messages, fmt.Sprintf("Node %s lost taint %s, skipping", nodeName, fu.Spec.UpdateTaint)) continue } - - node.Spec.Taints = slices.Delete(node.Spec.Taints, tindex, tindex+1) - - if err := r.Update(ctx, node); err != nil { - klog.Errorf("Failed to update node %s: %v", nodeName, err) - lastErr = err - } } endResult := "completed successfully" @@ -1009,11 +912,6 @@ func (r *GPUFirmwareUpdateReconciler) Reconcile(ctx context.Context, req ctrl.Re return ctrl.Result{}, nil } -func podIndexerFunc(rawObj client.Object) []string { - pod := rawObj.(*core.Pod) - return []string{pod.Spec.NodeName} -} - // SetupWithManager sets up the controller with the Manager. func (r *GPUFirmwareUpdateReconciler) SetupWithManager(mgr ctrl.Manager, copts ControllerOpts) error { r.Opts = copts @@ -1027,11 +925,6 @@ func (r *GPUFirmwareUpdateReconciler) SetupWithManager(mgr ctrl.Manager, copts C r.logRet = newLogsRetriever(clientset) r.imgVerify = newContentImageVerifier(mgr.GetAPIReader(), copts.Namespace) - pod := &core.Pod{} - if err := mgr.GetFieldIndexer().IndexField(context.Background(), pod, "spec.nodeName", podIndexerFunc); err != nil { - return err - } - return ctrl.NewControllerManagedBy(mgr). For(&intelcomv1alpha1.GPUFirmwareUpdate{}). Named("gpufirmwareupdate"). diff --git a/internal/controller/gpufirmwareupdate_controller_test.go b/internal/controller/gpufirmwareupdate_controller_test.go index 57f4186..bbb0a72 100644 --- a/internal/controller/gpufirmwareupdate_controller_test.go +++ b/internal/controller/gpufirmwareupdate_controller_test.go @@ -306,9 +306,9 @@ var _ = Describe("GPUFirmwareUpdate Controller", func() { // Create a fake client with a mock Update function fakeClient = fake.NewClientBuilder(). WithScheme(scheme). + WithIndex(&core.Pod{}, podNodeNameIndex, indexPodByNodeName). WithObjects(objs...). WithObjects(nodes[0], nodes[1]). - WithIndex(&core.Pod{}, "spec.nodeName", podIndexerFunc). WithStatusSubresource(&intelcomv1alpha1.GPUFirmwareUpdate{}). Build() @@ -671,6 +671,12 @@ var _ = Describe("GPUFirmwareUpdate Controller", func() { err = fakeClient.Get(ctx, types.NamespacedName{Name: "pod-4", Namespace: "default"}, &core.Pod{}) Expect(apierrors.IsNotFound(err)).To(BeTrue(), "Pod-4 on the draining node should be evicted") + // A firmware update drains GPU pods only. Pod-2 requests no GPU and holds no claim, + // so it must survive: gpuPodsOnNode is shared with a full-node drain and this pins + // which of the two policies the firmware update still uses. + Expect(fakeClient.Get(ctx, types.NamespacedName{Name: "pod-2", Namespace: "default"}, &core.Pod{})). + To(Succeed(), "Pod-2 requests no GPU and must not be evicted by a firmware update") + // DRAINING IN PROGRESS RECONCILE ret, err = controllerReconciler.Reconcile(ctx, reconcile.Request{ @@ -830,7 +836,7 @@ var _ = Describe("GPUFirmwareUpdate Controller Errors", func() { // Create a fake client with a mock Update function fakeClient = fake.NewClientBuilder(). WithScheme(scheme). - WithIndex(&core.Pod{}, "spec.nodeName", podIndexerFunc). + WithIndex(&core.Pod{}, podNodeNameIndex, indexPodByNodeName). WithStatusSubresource(&intelcomv1alpha1.GPUFirmwareUpdate{}). Build() @@ -931,7 +937,7 @@ var _ = Describe("GPUFirmwareUpdate Controller Errors", func() { fakeClient = fake.NewClientBuilder(). WithScheme(oldScheme). - WithIndex(&core.Pod{}, "spec.nodeName", podIndexerFunc). + WithIndex(&core.Pod{}, podNodeNameIndex, indexPodByNodeName). WithObjects(nodes[0], nodes[1]). WithStatusSubresource(&intelcomv1alpha1.GPUFirmwareUpdate{}). WithInterceptorFuncs( @@ -1138,7 +1144,7 @@ var _ = Describe("verifyContentImage via beginUpdate", func() { fakeClient = fake.NewClientBuilder(). WithScheme(scheme). - WithIndex(&core.Pod{}, "spec.nodeName", podIndexerFunc). + WithIndex(&core.Pod{}, podNodeNameIndex, indexPodByNodeName). WithStatusSubresource(&intelcomv1alpha1.GPUFirmwareUpdate{}). Build()