fix(scheduler): NodeAffinity PreFilter + per-node gpuFragmentation repack cooldown

[devin-ai-integration]

What type of PR is this?

/kind bug

What this PR does / why we need it:

Two independent scheduler fixes from the 2026-09-02/03 delphi-production incident, both in files only our fork touches.

1. predicates: register NodeAffinity as a PreFilter plugin (7198ebc)

PredicatesPlugin.InitPlugin registered NodeAffinity in filterPlugins and stableFilterPlugins but not in prefilterPlugins. The k8s NodeAffinity plugin computes GetRequiredNodeAffinity(pod) (nodeSelector + required affinity terms → label selectors, with regex-validated labels.NewRequirement) in PreFilter and caches it in cycle state; when Filter finds no state it falls back to recomputing it for every node:

s, err := getPreFilterState(state)
if err != nil {
    s = &preFilterState{requiredNodeSelectorAndAffinity: nodeaffinity.GetRequiredNodeAffinity(pod)}
}

So on delphi-production every allocate/preempt/reclaim/backfill session re-parsed each pending task's affinity once per node (≈2,600 pending tasks × 484 nodes × 3 actions). The scheduler CPU profile (pyroscope, 2026-09-02 22:27–22:57) had 30% of samples under NodeAffinity.Filter → GetRequiredNodeAffinity → labels.NewRequirement → regexp.tryBacktrack, the single largest leaf.

 nodeAffinityFilter := plugin.(*nodeaffinity.NodeAffinity)
 filterPlugins[nodeaffinity.Name] = nodeAffinityFilter
 stableFilterPlugins[nodeaffinity.Name] = nodeAffinityFilter
+prefilterPlugins[nodeaffinity.Name] = nodeAffinityFilter

With the registration, PrePredicate runs NodeAffinity.PreFilter once per task per session (every action already calls ssn.PrePredicateFn before PredicateNodes), the Filter reuses the cached selector, and pods with no nodeSelector/required affinity return Skip so handleSkipPrePredicatePlugin drops the Filter for them entirely. TestInitPlugin expectations updated accordingly.

This is the NodeAffinity half of upstream volcano-sh/volcano@03f2a0b ("fix(predicates): register more PreFilter plugins"); the VolumeZone half is left out because it is not on delphi's hot path and the fork's InitPlugin has diverged from the upstream helper it was written against.

2. rescheduling/gpuFragmentation: per-node cooldown, several drains per pass (ccfab4e)

The repack cooldown was a pool-wide lock — any node in the pool stamped with exa.ai/repack-last-eviction within cooldownSeconds (default 1800) held the whole pool — and planGpuFragmentationDrains stopped after one drain per pool. delphi runs crossPool: true, which makes the whole cluster one pool, so production repacked at most one node per 30 minutes (metrics: ~12 passes/h, 1–2 drains/h, 42 stamps total) against ~300 Consolidatable spot GPU nodes. That is why the 1-GPU flyte pods on ip-10-119-52-166 / ip-10-119-57-66 were never repacked onto one node.

Changes in planGpuFragmentationDrains / simulateDrain:

// before: pool-wide
cooled := true
for _, node := range members { if recentStamp(node) { cooled = false; break } }
if !cooled { continue }            // skips the entire pool
...
for _, cand := range sources { ...; drains = append(...); planned += len(moves); break }  // one drain per pool

// after: per node, budget-bounded
for _, member := range members {
    if !nodeCooled(member, conf, now) { continue }   // only the stamped node is held (as a source)
    ...
}
ledger := newDrainLedger(members)                    // idle capacity + drained/filled sets shared by the pass
for _, cand := range sources {
    if planned+len(victims) > conf.MaxVictims { break }   // unchanged: emptiest-first, move set atomic
    if ledger.touched(source.Name) { continue }          // already drained or filled this pass
    moves := simulateDrain(members, source, victims, gpu, ledger, predicate)  // commits to ledger on success
    ...
    planned += len(moves)                                 // no break: keep draining until the budget is spent
}

Semantics now:

• Cooldown is per node and only disqualifies the node as a source. Stamped destinations (which absorbed a drain) cannot be drained again for 30 min, but can keep receiving. Missing stamp = cooled; unparseable/future stamp = held (fail safe, unchanged).
• A pass drains nodes emptiest-first until maxVictims (8) is spent, so with 1-GPU pods a pass typically drains 3–8 nodes instead of 1. Total evictions per pass are still bounded by maxVictims; the "don't fall through to a fuller source under budget pressure" rule is unchanged.
• Drains within a pass share an idle-capacity ledger (no double booking of a free GPU across drains), a drained source is never a destination in the same pass (its fullness is stale), and a destination is never drained in the same pass (no bouncing a victim twice).
• Stamping (stampGpuFragmentationNodes), the drained-source node-order penalty, per-PodGroup eviction caps and the maxVictimPriority/opt-out/do-not-disrupt gates are unchanged; comments updated to drop the pool-wide-clock rationale.

Tests: TestPlanCooldownHoldsPool → TestPlanCooldownHoldsStampedSource (+ future-stamp case); new TestPlanCooldownIsPerNodeNotPerPool, TestPlanDrainsSeveralNodesPerPassWithinBudget, TestPlanLedgerPreventsDoubleBookingAcrossDrains, TestPlanDrainedSourceNeverReceivesInSamePass. All existing rescheduling tests pass unchanged (including TestPlanOneVictimAcrossPools, TestPlanNodeMoveSetIsAtomicUnderVictimBudget, TestNodeOrderPenalizesRecentlyDrainedSource).

Expected effect on delphi at the current config (interval: 5m, maxVictims: 8, cooldownSeconds: 1800): from ≤2 drained nodes/h to roughly 30–90/h while fragmentation persists, without touching the monorepo config. cooldownSeconds can be lowered later if the 30-min anti-thrash window per node turns out to be conservative.

Which issue(s) this PR fixes:

Fixes #

Special notes for your reviewer:

Verified locally: go build ./pkg/scheduler/..., go vet ./pkg/scheduler/plugins/predicates/ ./pkg/scheduler/plugins/rescheduling/, go test ./pkg/scheduler/plugins/... ./pkg/scheduler/framework/... ./pkg/scheduler/actions/... pass; TestAllocateWithPVC in actions/allocate is flaky on this branch and on exa/master alike (fails ~1 in 3 runs on both, passes on rerun).

Fix 1 is the secondary fix for the 2026-09-02 delphi "kubeflow operator hasn't updated the pytorch custom resource" incident; the primary one is running the scheduler at -v=2 (exa-labs/monorepo#138226). Fix 2 is the Volcano half of the spot-node repacking problem; the exa-scale consolidation half (candidate simulation timing out on the ~2,200 unprovisionable pending pods) is handled separately in the monorepo.

Not yet built or pushed to ECR; a monorepo training-stack.ts roll follows once this is merged and an exa-rebase-v1.14.1-<sha> image exists (staging first, as with #19).

Does this PR introduce a user-facing change?

NONE

Link to Devin session: https://app.devin.ai/sessions/9618e4fbaeb345ba9c161ec18bfdec6c
Open in Devin Desktop: https://app.devin.ai/desktop/session/9618e4fbaeb345ba9c161ec18bfdec6c?variant=devin
Requested by: @pfernandes21

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[devin-ai-integration]

CI note: E2E about Sequence failed on both runs, each on a different spec, and each spec has failed the same way on unrelated PRs in this repo:

run	failed spec	prior identical failure
33699111507	Queue Job Status Transition / Transform from running to unknown should succeed (PodGroup never reached Unknown in 5 min)	33118959397 (2026-08-27, gpuFragmentation repack PR; green on rerun)
33701348331	Job Error Handling / PodFailed→RestartPod; PodEvicted→TerminateJob, Timeout 5m (job stuck Restarting, expected Running)	32743600416 (capacitytiers), 31786673551 (repack priority gate)

Both are job-controller state transitions on pods with no nodeSelector/affinity, for which NodeAffinity.PreFilter returns Skip and Filter falls back to the same recompute path as before, so the change is a no-op for them; the 12 other e2e suites (Basic Scheduling, Scheduling Actions, Parallel Jobs, …) pass on both runs. 4 of the last 15 e2e_sequence runs in this repo failed, all on these two specs.

I can't rerun jobs here (gh run rerun → "Resource not accessible by integration"; agent-scripts/retry-ci is monorepo-only), and I shouldn't push another empty commit — could a reviewer hit "Re-run failed jobs" on 33701348331?