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()