fix(drain): don't hold eviction slots across PDB backoff sleeps

A PDB-blocked pod used to keep its concurrency slot while sleeping in
the 429 retry loop. Five blocked pods therefore occupied every slot
until the shared drain deadline, and pods with no budget at all timed
out without a single eviction attempt. The slot is now held only around
each Eviction call, acquisition is context-aware, and a pod that never
got an attempt says so instead of claiming a PDB timeout.

Reproduced live on a kind cluster (10 budget-blocked pods + 1 free pod,
20s deadline): before, zero evictions and the free pod failed with
"evict: context deadline exceeded"; after, the free pod evicts at once
and only the blocked pods report PDB timeouts.

Claude-Session: https://claude.ai/code/session_01NE55VsYGkb5tV3neKDmtZw

Author: hisco

pkg/k8score/drain_plan_test.go

@@ -2,6 +2,9 @@ package k8score
 
 import (
 	"context"
+	"fmt"
+	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -283,3 +286,78 @@ func TestPlanNodeDrainListsPDBsOnlyWhereADecisionNeedsThem(t *testing.T) {
 		}
 	}
 }
+
+// Six PDB-blocked pods sleep in their 429 retry loops; with only five concurrency
+// slots, a pod with no budget at all must still get its eviction attempted promptly.
+// The free pod's reactor is gated on five distinct blocked pods having been attempted
+// first, so this test cannot pass by racing ahead of the blocked retry loops.
+func TestDrainNodeEvictsFreePodWhilePDBBlockedPodsRetry(t *testing.T) {
+	objs := []runtime.Object{&corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "worker-1"}}}
+	for i := 0; i < 6; i++ {
+		p := drainTestPod(fmt.Sprintf("blocked-%d", i))
+		objs = append(objs, &p)
+	}
+	free := drainTestPod("free")
+	objs = append(objs, &free)
+	client := fake.NewSimpleClientset(objs...)
+
+	// Reactors run under the fake clientset's lock, so the gate must never block:
+	// the free pod is refused with the same 429 until five distinct blocked pods
+	// have been attempted, then its next retry is allowed through.
+	var mu sync.Mutex
+	blockedSeen := map[string]bool{}
+	gateOpen := false
+	var freeEvictedAfter time.Duration
+	start := time.Now()
+	pdbRefused := apierrors.NewTooManyRequests("Cannot evict pod as it would violate the pod's disruption budget.", 0)
+
+	client.PrependReactor("create", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
+		if action.GetSubresource() != "eviction" {
+			return false, nil, nil
+		}
+		ev, ok := action.(k8stesting.CreateAction).GetObject().(*policyv1.Eviction)
+		if !ok {
+			t.Errorf("eviction create carried unexpected object")
+			return true, nil, nil
+		}
+		mu.Lock()
+		defer mu.Unlock()
+		if ev.Name == "free" {
+			if !gateOpen {
+				return true, nil, pdbRefused
+			}
+			if freeEvictedAfter == 0 {
+				freeEvictedAfter = time.Since(start)
+			}
+			return true, nil, nil
+		}
+		blockedSeen[ev.Name] = true
+		if len(blockedSeen) >= 5 {
+			gateOpen = true
+		}
+		return true, nil, pdbRefused
+	})
+
+	const timeout = 3 * time.Second
+	res, err := DrainNode(context.Background(), client, "worker-1", DrainOptions{IgnoreDaemonSets: true, Timeout: timeout})
+	if err != nil {
+		t.Fatalf("DrainNode: %v", err)
+	}
+
+	if len(res.EvictedPods) != 1 || res.EvictedPods[0] != "shop/free" {
+		t.Fatalf("free pod must be evicted, got evicted=%v errors=%v", res.EvictedPods, res.Errors)
+	}
+	// Half the deadline separates "evicted while the blocked pods retried" from
+	// "evicted only after the deadline freed their slots".
+	if freeEvictedAfter <= 0 || freeEvictedAfter > timeout/2 {
+		t.Fatalf("free pod was evicted after %v; slots are being held across PDB backoff sleeps", freeEvictedAfter)
+	}
+	if len(res.Errors) != 6 {
+		t.Fatalf("all six blocked pods must report errors, got %v", res.Errors)
+	}
+	for _, e := range res.Errors {
+		if !strings.Contains(e, "timed out waiting for PDB to allow eviction") {
+			t.Fatalf("blocked pod error must name the PDB timeout, got %q", e)
+		}
+	}
+}

pkg/k8score/node_ops.go

@@ -111,10 +111,8 @@ func DrainNode(ctx context.Context, client kubernetes.Interface, nodeName string
 		wg.Add(1)
 		go func() {
 			defer wg.Done()
-			sem <- struct{}{}
-			defer func() { <-sem }()
 
-			err := evictPod(drainCtx, client, pod, opts.GracePeriodSeconds)
+			err := evictPod(drainCtx, client, pod, opts.GracePeriodSeconds, sem)
 			mu.Lock()
 			defer mu.Unlock()
 			if err != nil {
@@ -139,7 +137,11 @@ func hasLocalStorage(pod corev1.Pod) bool {
 }
 
 // evictPod evicts a single pod, retrying on PDB conflicts until the context deadline.
-func evictPod(ctx context.Context, client kubernetes.Interface, pod corev1.Pod, gracePeriod *int64) error {
+// The concurrency slot is held only around each Eviction call, never across the
+// backoff sleep: a handful of PDB-blocked pods sleeping in their retry loops would
+// otherwise occupy every slot until the shared deadline, and pods with no budget at
+// all would time out without a single eviction attempt.
+func evictPod(ctx context.Context, client kubernetes.Interface, pod corev1.Pod, gracePeriod *int64, sem chan struct{}) error {
 	eviction := &policyv1.Eviction{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      pod.Name,
@@ -154,9 +156,30 @@ func evictPod(ctx context.Context, client kubernetes.Interface, pod corev1.Pod,
 
 	backoff := 500 * time.Millisecond
 	maxBackoff := 5 * time.Second
+	attempted := false
+
+	deadlineErr := func() error {
+		if attempted {
+			return fmt.Errorf("timed out waiting for PDB to allow eviction: %w", ctx.Err())
+		}
+		return fmt.Errorf("drain deadline passed before this pod could be attempted: %w", ctx.Err())
+	}
 
 	for {
+		select {
+		case sem <- struct{}{}:
+		case <-ctx.Done():
+			return deadlineErr()
+		}
+		// The select above can pick the slot even when cancellation is also ready.
+		if ctx.Err() != nil {
+			<-sem
+			return deadlineErr()
+		}
 		err := client.PolicyV1().Evictions(pod.Namespace).Evict(ctx, eviction)
+		<-sem
+		attempted = true
+
 		if err == nil {
 			return nil
 		}