diff --git a/cmd/controller-manager/main.go b/cmd/controller-manager/main.go index 51b64d95445..8cbd0142965 100644 --- a/cmd/controller-manager/main.go +++ b/cmd/controller-manager/main.go @@ -43,6 +43,7 @@ import ( _ "volcano.sh/volcano/pkg/controllers/jobflow" _ "volcano.sh/volcano/pkg/controllers/jobtemplate" _ "volcano.sh/volcano/pkg/controllers/podgroup" + _ "volcano.sh/volcano/pkg/controllers/podstartwatchdog" _ "volcano.sh/volcano/pkg/controllers/queue" _ "volcano.sh/volcano/pkg/controllers/sharding" commonutil "volcano.sh/volcano/pkg/util" diff --git a/farai/docs/changes/README.md b/farai/docs/changes/README.md new file mode 100644 index 00000000000..a4797cee6e1 --- /dev/null +++ b/farai/docs/changes/README.md @@ -0,0 +1,21 @@ +# Changes from vanilla Volcano + +This fork (cut from upstream v1.15.0) is vanilla Volcano plus the +changes documented here — one file per changed component, each with the +motivation, the files touched, and how to use the feature: + +- [fifopriority.md](fifopriority.md) — in-tree scheduler plugin: FIFO-within- + PriorityClass ordering and preemption, driven by the submission-time stamp +- [podstartwatchdog.md](podstartwatchdog.md) — controller in + `vc-controller-manager`: marks bound-but-unstarted pods Failed past the + `volcano.sh/pod-start-timeout` deadline (job label, copied onto pods); the + job's own `PodFailed → RestartJob` policy and `maxRetry` do the rest +- [job-controller.md](job-controller.md) — eviction-triggered restarts don't + consume the retry budget; PodGroups stamped with the job's true submission + time (`volcano.sh/submitted-at`) +- [helm-chart.md](helm-chart.md) — chart knobs for the podstartwatchdog + +Consumer context (the multi-tenant platform whose admission webhook defaults the +preemptable annotation, eviction/restart policies, and the pod-start-timeout +label onto tenant jobs, plus the design rationale for all of it) lives in the +`flamingo_tenants` repo under `docs/agent/scheduler/`. diff --git a/farai/docs/changes/fifopriority.md b/farai/docs/changes/fifopriority.md new file mode 100644 index 00000000000..05bfe0a6c2b --- /dev/null +++ b/farai/docs/changes/fifopriority.md @@ -0,0 +1,73 @@ +# fifopriority — FIFO-within-priority scheduler plugin + +## Motivation + +Stock Volcano orders and preempts by PriorityClass alone, so on a busy cluster a +large job can be leapfrogged indefinitely by newer small jobs of the same class — +starvation the `priority` plugin cannot express. fifopriority orders jobs by +PriorityClass first and **submission time** second, and drives *both* allocation +order and preempt-victim selection from that one comparator (a pending job may +evict strictly-lower-class work, or equal-class work submitted in a strictly later +second). The clock is the `volcano.sh/submitted-at` PodGroup label (epoch ms — the +owning VCJob's immutable creationTimestamp, stamped by the job controller onto +every (re)created PodGroup, so it survives RestartJob and cannot be forged), with +PodGroup creationTimestamp as the fallback; comparison is at second granularity so +same-second submissions never evict each other. + +Beyond ordering, the plugin reinstates the **queue-Open check at enqueue** (jobs +submitted to Closed/Closing or missing queues do not enqueue and stay podless). +Stock proportion only provides that check inside its enqueue-stage capacity vote, +which deployments running open enqueue gates (`enableJobEnqueued: false`) must +disable — losing the state check as collateral. State only; capacity is never +gated at enqueue. + +## Changed component / files + +Scheduler (`vc-scheduler`): + +- `pkg/scheduler/plugins/fifopriority/fifopriority.go` — new plugin (comparator, + task/sub-job order, Preemptable + UnifiedEvictable victim rules, queue-Open + enqueue gate, starving test; stock `priority` parity everywhere else) +- `pkg/scheduler/plugins/fifopriority/fifopriority_test.go` — unit tests +- `pkg/scheduler/plugins/factory.go` — registration (config name `fifopriority`, + used in the scheduler config exactly where `priority` was) + +## Usage + +Name `fifopriority` in the scheduler configuration where `priority` would go +(tier 1), with gang's and drf's victim vetoes off so it alone picks preempt +victims: + +```yaml +actions: "enqueue, allocate, reclaim, preempt, backfill" +tiers: +- plugins: + - name: fifopriority + - name: gang + enablePreemptable: false + - name: conformance +- plugins: + - name: drf + enablePreemptable: false + - name: predicates + - name: proportion + enableJobEnqueued: false + - name: nodeorder + - name: binpack +``` + +`enableJobEnqueued: false` on proportion is part of the design, not an +optimization: it disables proportion's enqueue-stage capacity vote so gates stay +open. Do not rely on omitting it — that happens to behave the same only because a +tier-1 Permit short-circuits later tiers' enqueue votes, and silently stops +working if fifopriority moves out of tier 1. + +Nothing else to configure: PodGroups created by the job controller carry the +`volcano.sh/submitted-at` stamp automatically (see job-controller.md); podgroups +without one fall back to their own creation time. Two operational notes: eviction +candidates must carry `volcano.sh/preemptable: "true"` on their pods (upstream +drops unmarked candidates before any plugin runs), and a scheduler image WITHOUT +this plugin does not crash on this config — an unknown plugin name is only an +error log (`Failed to get plugin`) and tier 1 silently degrades to +conformance-only preemption — so deployments should assert the running image and +scan fresh scheduler logs after applying the config. diff --git a/farai/docs/changes/helm-chart.md b/farai/docs/changes/helm-chart.md new file mode 100644 index 00000000000..6275b5ec0c1 --- /dev/null +++ b/farai/docs/changes/helm-chart.md @@ -0,0 +1,30 @@ +# Helm chart — podstartwatchdog knob + +## Motivation + +The chart has no generic extra-args passthrough for the controllers deployment; +per-flag `custom.*` values are its native pattern (qps, worker threads, …). The +podstartwatchdog's poll-cadence flag gets the same treatment so deployments can +tune it through values instead of patching the deployment. Unset (the default) +omits the arg and the compiled default (1m) applies. There is deliberately no +timeout knob: deadlines come exclusively from the per-job +`volcano.sh/pod-start-timeout` label, copied onto the job's pods (see +podstartwatchdog.md). + +## Changed component / files + +Helm chart (`installer/helm/chart/volcano/`): + +- `values.yaml` — `custom.controller_pod_start_interval` (default `~`) +- `templates/controllers.yaml` — conditional `--pod-start-interval` arg on the + controllers container + +## Usage + +Install (or upgrade) from the in-tree chart with the knob in values: + +```sh +helm upgrade --install volcano installer/helm/chart/volcano \ + --namespace volcano-system --create-namespace \ + --set custom.controller_pod_start_interval=30s +``` diff --git a/farai/docs/changes/job-controller.md b/farai/docs/changes/job-controller.md new file mode 100644 index 00000000000..cff9162c29a --- /dev/null +++ b/farai/docs/changes/job-controller.md @@ -0,0 +1,89 @@ +# Job controller — retry accounting + PodGroup submission stamp + +Two changes to `vc-controller-manager`'s job controller. + +## 1. Evictions don't consume the retry budget + +### Motivation + +`RetryCount` counts every trip through a restart, whatever caused it, and a job +whose count reaches `maxRetry` is terminally `Failed`. With a +`PodEvicted → RestartJob` policy (whole-gang teardown on eviction), scheduler +preemption/reclaim — which works by *deleting* pods, surfacing to the controller +as `PodEvicted` — would bill every disruption against the budget, letting repeated +preemption permanently kill an innocent job. Eviction means disrupted, not failed: +eviction-triggered restarts are now free, while `PodFailed`-triggered and +directly-issued restarts count exactly as before. (Requested upstream in +volcano-sh/volcano#2560; a fix was invited but never landed.) + +### Changed files + +- `pkg/controllers/job/state/factory.go` — `state.Action` gains the triggering + `Event`; new `Action.CountsTowardRetry()` (false only for `PodEvicted`) +- `pkg/controllers/job/job_controller_util.go` — `GetStateAction` forwards the + event from the delayed action +- `pkg/controllers/job/state/running.go`, `state/pending.go` — the four + `RetryCount++` sites on `RestartJob`/`RestartTask|Pod|Partition` gated on + `CountsTowardRetry()` (the `ResumeJob` increments in aborting/aborted states are + untouched — resume is never eviction-triggered) +- `pkg/controllers/job/state/retry_test.go` — unit tests (evicted vs failed vs + direct, both states, both action kinds) + +### Usage + +Nothing to configure — evictions never consume the budget. To get whole-gang +teardown-and-requeue on eviction at no retry cost, give the job an eviction +policy: + +```yaml +spec: + maxRetry: 5 # bounds FAILURES only (CRD schema default: 3) + policies: + - {event: PodFailed, action: RestartJob} + - {event: PodEvicted, action: RestartJob} +``` + +`PodFailed`-triggered and directly-issued restarts (`vcctl`, Commands) count +exactly as before. + +## 2. PodGroups carry the job's true submission time + +### Motivation + +FIFO scheduling needs a restart-proof submission clock, but the scheduler only +sees PodGroups, and a PodGroup's `creationTimestamp` resets on every `RestartJob` +recreation — a restarted job would silently go to the back of its class's line. +The owning VCJob's `metadata.creationTimestamp` is the true submission time: +API-server-managed and immutable, so it also cannot be forged (previously an +out-of-tree admission webhook stamped a label, which a user could edit +post-create and launder onto the next recreated PodGroup). The fifopriority +plugin orders by this stamp. + +### Changed files + +- `pkg/util/util.go` — `SubmittedAtPodGroupLabel = "volcano.sh/submitted-at"` + (shared by controller and scheduler plugin) +- `pkg/controllers/job/job_controller_actions.go` — `podGroupLabels` stamps the + epoch-ms creation time onto every created PodGroup (force-set after the job + label copy, never inherited) +- `pkg/controllers/job/podgroup_stamp_test.go` — unit tests (forgery override, + second-aligned epoch-ms format) + +### Usage + +Automatic — every PodGroup the job controller (re)creates carries the label. +Inspect with: + +```sh +kubectl get podgroup -o jsonpath='{.metadata.labels.volcano\.sh/submitted-at}' +``` + +The value is the owning VCJob's `metadata.creationTimestamp` as epoch +milliseconds; it cannot be set or forged from the job side (job labels are copied +to the PodGroup, but this key is force-set after the copy). The stamp is written +only at PodGroup creation — the controller does not re-reconcile it afterwards. +Editing it directly on the PodGroup requires PodGroup write access, which tenants +must not have: RBAC is the boundary there, not this controller. Two consequences: +PodGroups that predate this feature stay unstamped (they order by their own +creation time — the pre-feature status quo) until their next recreation stamps +them, and the scheduler plugin's fallback covers them meanwhile. diff --git a/farai/docs/changes/podstartwatchdog.md b/farai/docs/changes/podstartwatchdog.md new file mode 100644 index 00000000000..df7342a3c0d --- /dev/null +++ b/farai/docs/changes/podstartwatchdog.md @@ -0,0 +1,129 @@ +# podstartwatchdog — bound pods must start + +## Motivation + +A placed pod that cannot start — unpullable image, unmountable volume, +container-create config error — never *fails* on its own: it sits Pending on its +node, charging its full resource request against the cluster forever, and no +`PodFailed` lifecycle policy ever fires. The stock `PodPending`-timeout policy +cannot be used instead because its clock starts at pod creation — on a +pass-through-enqueue cluster pods legitimately queue for hours, so it would kill +innocent queued jobs. The watchdog anchors on **binding**: a pod counts as stuck +only when bound (`spec.nodeName` set — it already holds capacity) and still +Pending, whatever the cause. Stuck past its deadline → the pod is marked +**Failed** (`status.phase`, reason `PodStartTimeout`) with a Warning Event on the +pod. + +That is the watchdog's whole job. Everything downstream belongs to the owning +job's own lifecycle policies, exactly as if a container had died: `PodFailed → +RestartJob` (the platform default) tears down and requeues the gang, **consuming +one retry per lap** — so transient causes (a registry blip, a slow cold pull that +finishes next time) self-heal on the next attempt, while persistent causes burn +`maxRetry` laps and park the job in terminal `Failed`: bounded, inspectable +(`kubectl describe vcjob` shows the restart history and final state), and +resubmittable. + +**Pod-scoped and stateless by design.** The watchdog never resolves, verifies, or +trusts a pod→job mapping — no `volcano.sh/job-name` lookup, no ownerReference +check, no job phase gate. Attribution of a failed pod to its job happens where it +always has: in the job controller's own event handlers, with their ownership and +version safeguards. There is no targeting decision to spoof — a hand-made pod +carrying the label only ever condemns itself. Statelessness comes from two +anchors that live on the pod: + +- **The clock**: the API server's binding subresource sets the + `PodScheduled=True` condition in the same write that sets `spec.nodeName`, so + its `lastTransitionTime` is the bind moment; pod phase never regresses, so + bound-and-Pending now means continuously unstarted since then. (A bound pod + without the condition can only be one created with `nodeName` preset — it never + queued, so its creationTimestamp is its bind time.) Controller restarts forget + nothing. +- **The deadline**: the `volcano.sh/pod-start-timeout` label **on the pod**, a Go + duration — the only deadline source. The job controller copies it from the + owning VCJob's label onto every pod it creates (the platform's admission + webhook defaults it onto every job), so it is stamped by a trusted component; + tenants cannot edit controller-created pods. A pod without the label, or with + an unparseable or non-positive value (`"0"` is an explicit opt-out), is not + policed. Editing the job's label reaches the next pod incarnation (e.g. the + next restart lap). + +Marking a pod Failed from outside the kubelet is a one-way door the stack honors +end to end: the API server does not restrict phase transitions in status updates; +the kubelet begins teardown when it observes an API-terminal pod +(`pod_workers.go`: "Pod is in a terminal phase (success/failed), begin teardown" — +the same contract PodGC relies on); and the Volcano job controller classifies the +transition as an ordinary `PodFailed` event. A racing kubelet status sync can at +worst delay the mark by one interval — the watchdog re-asserts. + +**Cause-agnostic cuts both ways**: time spent in *legitimate* initialization — a +cold pull of a huge image, a long-running init container (the pod stays Pending +until init containers finish) — counts against the deadline too, because from the +outside it is indistinguishable from an init that will never finish. The retry +laps soften this (a slow pull that completes within a later lap survives), but +jobs whose normal startup can approach the deadline should still set a larger +label value, and the deployment default must be sized against worst-case +cold-start time. + +## Changed component / files + +Controller manager (`vc-controller-manager`): + +- `pkg/controllers/podstartwatchdog/podstartwatchdog.go` — the whole controller: + informer plumbing, status-subresource Failed patch, and pod event, plus the + pure decision logic (bound+Pending predicate, bind-time clock anchor, + pod-label deadline) at the bottom of the file; flag `--pod-start-interval` + (poll cadence, default 1m); registered as `podstartwatchdog-controller`, + enabled by default +- `pkg/controllers/podstartwatchdog/podstartwatchdog_test.go` — unit tests for + the decision logic +- `pkg/controllers/job/job_controller_util.go` — the job controller copies the + `volcano.sh/pod-start-timeout` label from the job onto every pod it creates +- `pkg/util/util.go` — `PodStartTimeoutLabel` (`volcano.sh/pod-start-timeout`) +- `cmd/controller-manager/main.go` — package import for registration +- `installer/helm/chart/volcano/templates/controllers.yaml`, + `installer/volcano-development.yaml` — controllers RBAC gains `patch` on + `pods/status` (the Failed mark) + +## Usage + +Runs by default in `vc-controller-manager` (disable with +`--controllers=*,-podstartwatchdog-controller`). The deadline is set per job, via +a label on the VCJob (a Go duration; `"0"` is an explicit opt-out), and rides +onto the pods automatically: + +```yaml +metadata: + labels: + volcano.sh/pod-start-timeout: "2h" +``` + +A deployment ensures coverage by defaulting the label at admission (the platform +webhook injects `30m` onto jobs that don't set one). For the full +restart-then-fail semantics, jobs need the restart policy and a retry budget +(both platform webhook defaults): + +```yaml +spec: + maxRetry: 5 + policies: + - {event: PodFailed, action: RestartJob} + - {event: PodEvicted, action: RestartJob} +``` + +The only chart knob is the poll cadence: + +```yaml +custom: + controller_pod_start_interval: "30s" # default 1m +``` + +On trigger the pod gets a Warning Event (reason `PodStartTimeout`) and is marked +Failed; the job's `PodFailed → RestartJob` policy restarts the gang at the cost +of one retry, and `maxRetry` exhaustion fails the job terminally. Jobs with +legitimately slow startup (large cold image pulls, long init containers) should +carry a label larger than their worst-case initialization time — the watchdog +cannot tell slow from stuck, though a slow start that completes within a later +retry lap survives on its own. Pair with `restartPolicy: Never` on task templates +so mid-run container failures route through the same `PodFailed → RestartJob` +path, and do not use `PodPending` lifecycle policies for the same purpose — their +clock starts at pod creation, not binding, so they fire on healthy queued jobs. diff --git a/farai/docs/issues/churn.md b/farai/docs/issues/churn.md new file mode 100644 index 00000000000..07469ec56dd --- /dev/null +++ b/farai/docs/issues/churn.md @@ -0,0 +1,128 @@ +# Churn — unthrottled restart loops + +**Status: open. Not yet addressed.** This document enumerates the failure +modes; it deliberately proposes no design. + +## The underlying gap + +Volcano has no time-based brake anywhere in the restart path (verified +against the v1.15.0 tree this fork is cut from): + +- The job controller applies **no delay between restart attempts** — a + `RestartJob` kills and recreates pods as fast as the controller can cycle. + `maxRetry` is a count, not a clock: it bounds total attempts, never the + rate. (The batch/v1 Job controller it parallels re-creates failed pods + with an escalating 10s→~6m delay; Volcano's controller never implemented + an equivalent.) +- The scheduler has **no backoff queue** — every pending job is + re-evaluated every session (~1s), forever. (kube-scheduler places + unschedulable pods in a per-pod exponential backoff queue; Volcano has no + equivalent.) + +Two platform choices, each correct on its own terms, remove the remaining +external brakes and widen the gap: + +- `restartPolicy: Never` (webhook default) takes the kubelet's + CrashLoopBackOff out of the picture — deliberately, so container deaths + surface as `PodFailed` and restart whole gangs instead of stalling + invisibly in place. +- Evictions no longer consume `maxRetry` (see + `../changes/job-controller.md`) — deliberately, so victims of preemption + don't burn their failure budget — which leaves eviction loops bounded by + **neither count nor time**. + +## Failure mode 1: eviction churn (rotation starvation of large gangs) + +Preemption frees capacity asynchronously: victims drain over their +termination grace periods, and nothing reserves the freed capacity for the +job the eviction was for (allocate only *prefers* `NominatedNodeName`; +pipelined claims do not persist across sessions). In the gap, smaller +pending jobs bind into the partial holes. The evicted juniors themselves +restart immediately, keep their seniority (the `volcano.sh/submitted-at` +stamp survives restarts by design), re-enter the very next session as +pod-less pending jobs, and re-bind into freed capacity — phase-offset from +the wide senior gang whose eviction freed it. + +Consequences: + +- **A finite junior population can starve a wide senior gang indefinitely.** + Evicted juniors return with no progress made and full seniority; they + rebind staggered onto the capacity the senior needs whole; the senior + re-evicts; the rotation repeats. No fresh-arrival stream is required, and + nobody in the loop makes progress. Convergence is probabilistic with no + bound — governed by termination latency, not by any swap count. The + fifopriority plugin doc is explicit that its improvement guarantee is + per-session only and cross-session slippage is real. +- **Every lap does negative work**: whole-gang teardowns, re-pulls, + restart-from-scratch for the juniors, zero progress for the senior, and a + full preempt/drain cycle of scheduler and API load. +- **Post-retry-exemption, the loop is fully unbounded**: evicted jobs never + exhaust anything; the rotation runs until cluster load happens to break + it. + +## Failure mode 2: crash-restart churn + +With CrashLoopBackOff opted out and no controller-side delay, a job that +crashes on startup restarts as fast as controller + scheduler can cycle — +measured in our KWOK runs, a crash-looping job burns five retries in well +under a minute. + +Consequences: + +- **The retry budget measures attempts, not time**, so it protects against + the wrong thing during transient external failures: a job whose image + registry, NFS mount, or credential backend blips for two minutes can + exhaust its entire `maxRetry` budget *inside* the outage window and die + permanently — where time-spaced retries would have sailed through. + Conversely, raising `maxRetry` to survive outages just extends the + full-speed hammering. +- **Per-task restart policies are equally unthrottled**: `RestartTask`-style + policies recreate pods without any pause either; the only bound anywhere + is the retry count. +- **Load amplification**: each lap is a full kill/recreate/schedule/bind + cycle of controller, scheduler, and API server work; N crash-looping jobs + generate sustained platform load with zero useful output. + +## What existing mechanisms do and do not cover + +| Mechanism | Covers | Does not cover | +|---|---|---| +| `maxRetry` (CRD default 3) | total crash-attempt count | rate of attempts; evictions (exempt) | +| Retry exemption for evictions (`../changes/job-controller.md`) | victims' budgets | adds the unbounded-eviction-loop dimension | +| podstartwatchdog (`../changes/podstartwatchdog.md`) | bound pods that never start (fails the pod; each `PodFailed → RestartJob` lap bills a retry, so stuck-start loops are count-bounded and paced by the timeout) | anything that restarts promptly — churn is fast, not stuck | +| fifopriority ordering (`../changes/fifopriority.md`) | within-session refill seniority | cross-session slippage — the rotation lives between sessions | +| kubelet CrashLoopBackOff | (opted out via `restartPolicy: Never`) | — | +| batch/v1 Job recreate backoff | (inapplicable — VCJobs create pods directly) | — | + +Elsewhere in the ecosystem this gap is closed as a matter of course: +kube-scheduler backs off scheduling attempts per pod; Kueue applies +exponential requeue backoff to its pods-ready evictions; the batch Job +controller and the kubelet both time-space crash recovery. Volcano is the +outlier, and the platform's (individually sound) choices remove the two +brakes that would otherwise partially apply. + +## Status + +Unaddressed. One verified implementation constraint is worth preserving for +any future attempt: the scheduler session's `DeallocateFunc` event callbacks +fire during speculative, routinely-discarded operations (preempt/reclaim +evict via `Statement.Evict`; allocation rollbacks fire the same callback), +so they cannot be used as a "real eviction" signal — any future detector +must be commit-observable (e.g. task-status changes that survive to session +close). + +## Upstream references + +- [volcano-sh/volcano#13](https://github.com/volcano-sh/volcano/issues/13) — + reservation to prevent exactly this big-job starvation, requested 2019. +- [#4323](https://github.com/volcano-sh/volcano/pull/4323) / + [#4887](https://github.com/volcano-sh/volcano/pull/4887) — `Reservation` + CR in active development; unmerged. +- [#5045](https://github.com/volcano-sh/volcano/pull/5045) — WIP: persists + pipelined claims/`NominatedNodeName` across sessions (nomination-aware + placement — the actual cure for the eviction rotation); unmerged. +- [#4607](https://github.com/volcano-sh/volcano/issues/4607) — gang-atomic + preempt/reclaim, open at priority/high; would make frees chunky and + shrink the rotation's fuel. + +None of these ship in the pinned v1.15.0. diff --git a/installer/helm/chart/volcano/templates/controllers.yaml b/installer/helm/chart/volcano/templates/controllers.yaml index b2ea33b7b4b..3ac89ca765b 100644 --- a/installer/helm/chart/volcano/templates/controllers.yaml +++ b/installer/helm/chart/volcano/templates/controllers.yaml @@ -49,6 +49,10 @@ rules: - apiGroups: [""] resources: ["pods"] verbs: ["create", "get", "list", "watch", "delete", "patch", "update"] + - apiGroups: [""] + resources: ["pods/status"] + # patch: the podstartwatchdog marks bound-but-unstarted pods Failed. + verbs: ["patch"] - apiGroups: [""] resources: ["pods/finalizers"] verbs: ["update", "patch"] @@ -207,6 +211,9 @@ spec: {{- if .Values.custom.controller_enabled_controllers }} - --controllers={{.Values.custom.controller_enabled_controllers}} {{- end }} + {{- if .Values.custom.controller_pod_start_interval }} + - --pod-start-interval={{.Values.custom.controller_pod_start_interval}} + {{- end }} - -v={{.Values.custom.controller_log_level}} - 2>&1 imagePullPolicy: {{ .Values.basic.image_pull_policy }} diff --git a/installer/helm/chart/volcano/values.yaml b/installer/helm/chart/volcano/values.yaml index 1fa676ce08e..78698503254 100644 --- a/installer/helm/chart/volcano/values.yaml +++ b/installer/helm/chart/volcano/values.yaml @@ -49,6 +49,11 @@ custom: controller_worker_threads_for_podgroup: 5 # Default: "*,-sharding-controller" (sharding-controller disabled by default) controller_enabled_controllers: ~ + # Pod-start watchdog (podstartwatchdog-controller) poll cadence: how often bound + # pods that have not started are evaluated against their own + # volcano.sh/pod-start-timeout label (copied from the owning job's label — the + # only deadline source). Empty uses the compiled default (1m). + controller_pod_start_interval: ~ scheduler_kube_api_qps: 2000 scheduler_kube_api_burst: 2000 scheduler_schedule_period: 1s @@ -283,13 +288,13 @@ custom: # sharding controller watches it for live-reloadable scheduler shard # specifications. Changes take effect without a controller restart. # -# schedulerConfigs defines one entry per scheduler that participates in -# node sharding. Each entry specifies: -# name - must match the scheduler's name in NodeShard CRDs -# type - scheduler type, e.g. "volcano" or "agent" -# policies - ordered policy chain. Built-in policies include: -# allocation-rate (CPU-utilization filter/scorer), -# warmup (warmup-node scorer), and node-limit (min/max selector). +# schedulerConfigs defines one entry per scheduler that participates in +# node sharding. Each entry specifies: +# name - must match the scheduler's name in NodeShard CRDs +# type - scheduler type, e.g. "volcano" or "agent" +# policies - ordered policy chain. Built-in policies include: +# allocation-rate (CPU-utilization filter/scorer), +# warmup (warmup-node scorer), and node-limit (min/max selector). # # shardSyncPeriod sets the interval between periodic full shard reconciliations. # Accepts Go duration strings ("30s", "2m", "1h"). @@ -298,36 +303,36 @@ custom: # shard reconciliation in addition to the periodic sync. # # Example: -# sharding_configmap_data: | -# schedulerConfigs: -# - name: agent-scheduler -# type: agent -# policies: -# - name: allocation-rate -# weight: 1 -# arguments: -# minCPUUtil: 0.7 -# maxCPUUtil: 1.0 -# - name: warmup -# weight: 2 -# - name: node-limit -# arguments: -# minNodes: 1 -# maxNodes: 100 -# - name: volcano -# type: volcano -# policies: -# - name: allocation-rate -# weight: 1 -# arguments: -# minCPUUtil: 0.0 -# maxCPUUtil: 0.69 -# - name: node-limit -# arguments: -# minNodes: 1 -# maxNodes: 100 -# shardSyncPeriod: "60s" -# enableNodeEventTrigger: true +# sharding_configmap_data: | +# schedulerConfigs: +# - name: agent-scheduler +# type: agent +# policies: +# - name: allocation-rate +# weight: 1 +# arguments: +# minCPUUtil: 0.7 +# maxCPUUtil: 1.0 +# - name: warmup +# weight: 2 +# - name: node-limit +# arguments: +# minNodes: 1 +# maxNodes: 100 +# - name: volcano +# type: volcano +# policies: +# - name: allocation-rate +# weight: 1 +# arguments: +# minCPUUtil: 0.0 +# maxCPUUtil: 0.69 +# - name: node-limit +# arguments: +# minNodes: 1 +# maxNodes: 100 +# shardSyncPeriod: "60s" +# enableNodeEventTrigger: true sharding_configmap_enable: true sharding_configmap_data: ~ diff --git a/installer/volcano-development.yaml b/installer/volcano-development.yaml index 4c4148c1528..d5e43fad330 100644 --- a/installer/volcano-development.yaml +++ b/installer/volcano-development.yaml @@ -8954,6 +8954,10 @@ rules: - apiGroups: [""] resources: ["pods"] verbs: ["create", "get", "list", "watch", "delete", "patch", "update"] + - apiGroups: [""] + resources: ["pods/status"] + # patch: the podstartwatchdog marks bound-but-unstarted pods Failed. + verbs: ["patch"] - apiGroups: [""] resources: ["pods/finalizers"] verbs: ["update", "patch"] diff --git a/pkg/controllers/job/job_controller_actions.go b/pkg/controllers/job/job_controller_actions.go index 6478596f70c..3a74ca32d03 100644 --- a/pkg/controllers/job/job_controller_actions.go +++ b/pkg/controllers/job/job_controller_actions.go @@ -20,6 +20,7 @@ import ( "context" "fmt" "reflect" + "strconv" "strings" "sync" "sync/atomic" @@ -38,6 +39,7 @@ import ( jobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers" "volcano.sh/volcano/pkg/controllers/job/state" "volcano.sh/volcano/pkg/controllers/metrics" + commonutil "volcano.sh/volcano/pkg/util" ) var calMutex sync.Mutex @@ -784,6 +786,28 @@ func (cc *jobcontroller) createPVC(job *batch.Job, vcName string, volumeClaim *v return nil } +// jobSubmittedAt is the job's true submission time as the epoch-millisecond string +// stamped onto its PodGroups: the job's API-server-managed, immutable +// creationTimestamp — unlike the PodGroup's own creationTimestamp, which resets on +// every RestartJob recreation. +func jobSubmittedAt(job *batch.Job) string { + return strconv.FormatInt(job.CreationTimestamp.UnixMilli(), 10) +} + +// podGroupLabels are the job's labels plus the submission-time stamp. The stamp is +// force-set AFTER the copy — it must never be inherited from a (user-editable) job +// label, or editing the job could forge an earlier place in a FIFO queue. It is set +// only here, at PodGroup (re)creation; guarding it after that is RBAC's job (who +// can write PodGroups), not this controller's. +func podGroupLabels(job *batch.Job) map[string]string { + labels := make(map[string]string, len(job.Labels)+1) + for key, value := range job.Labels { + labels[key] = value + } + labels[commonutil.SubmittedAtPodGroupLabel] = jobSubmittedAt(job) + return labels +} + func (cc *jobcontroller) createOrUpdatePodGroup(job *batch.Job) error { // If PodGroup does not exist, create one for Job. pg, err := cc.getPodGroupByJob(job) @@ -804,7 +828,7 @@ func (cc *jobcontroller) createOrUpdatePodGroup(job *batch.Job) error { // add job.UID into its name when create new PodGroup Name: cc.generateRelatedPodGroupName(job), Annotations: job.Annotations, - Labels: job.Labels, + Labels: podGroupLabels(job), OwnerReferences: []metav1.OwnerReference{ *metav1.NewControllerRef(job, helpers.JobKind), }, diff --git a/pkg/controllers/job/job_controller_util.go b/pkg/controllers/job/job_controller_util.go index 66fae5dc75a..4f263815d01 100644 --- a/pkg/controllers/job/job_controller_util.go +++ b/pkg/controllers/job/job_controller_util.go @@ -36,6 +36,7 @@ import ( jobhelpers "volcano.sh/volcano/pkg/controllers/job/helpers" "volcano.sh/volcano/pkg/controllers/job/state" "volcano.sh/volcano/pkg/controllers/util" + commonutil "volcano.sh/volcano/pkg/util" ) // MakePodName append podname,jobname,taskName and index and returns the string. @@ -157,6 +158,12 @@ func createJobPod(job *batch.Job, template *v1.PodTemplateSpec, ix int, jobForwa if value, found := job.Labels[schedulingv2.CooldownTime]; found { pod.Labels[schedulingv2.CooldownTime] = value } + // The pod-start watchdog is pod-scoped: it reads the deadline from the pod + // itself, so the job's label rides onto every pod at creation. A label edit + // on the job reaches the next pod incarnation (e.g. after a restart lap). + if value, found := job.Labels[commonutil.PodStartTimeoutLabel]; found { + pod.Labels[commonutil.PodStartTimeoutLabel] = value + } } if jobForwarding { @@ -430,7 +437,7 @@ func isInternalAction(action v1alpha1.Action) bool { } func GetStateAction(delayAct *delayAction) state.Action { - action := state.Action{Action: delayAct.action} + action := state.Action{Action: delayAct.action, Event: delayAct.event} if delayAct.action == v1alpha1.RestartTaskAction { action.Target = state.Target{TaskName: delayAct.taskName, Type: state.TargetTypeTask} diff --git a/pkg/controllers/job/podgroup_stamp_test.go b/pkg/controllers/job/podgroup_stamp_test.go new file mode 100644 index 00000000000..6b3ac230b7c --- /dev/null +++ b/pkg/controllers/job/podgroup_stamp_test.go @@ -0,0 +1,74 @@ +/* +Copyright 2026 The Volcano Authors. + +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. +*/ + +// Tests for the PodGroup submission-time stamp: derived from the job's immutable +// creationTimestamp, force-set over anything the (user-editable) job labels claim. +package job + +import ( + "strconv" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + batch "volcano.sh/apis/pkg/apis/batch/v1alpha1" + + commonutil "volcano.sh/volcano/pkg/util" +) + +func TestPodGroupLabelsCarryTheJobCreationTimeAndOverrideForgery(t *testing.T) { + created := metav1.NewTime(time.Date(2026, 1, 1, 0, 0, 42, 0, time.UTC)) + job := &batch.Job{ObjectMeta: metav1.ObjectMeta{ + Name: "sim-a", + CreationTimestamp: created, + Labels: map[string]string{ + "app": "worker", + // A submitter editing their job cannot claim an earlier place in line: + // the stamp must come from creationTimestamp, never from a job label. + commonutil.SubmittedAtPodGroupLabel: "1", + }, + }} + + labels := podGroupLabels(job) + + want := strconv.FormatInt(created.UnixMilli(), 10) + if labels[commonutil.SubmittedAtPodGroupLabel] != want { + t.Fatalf("stamp = %q, want the job creationTimestamp %q", + labels[commonutil.SubmittedAtPodGroupLabel], want) + } + if labels["app"] != "worker" { + t.Fatalf("other job labels must be preserved; got %v", labels) + } + if job.Labels[commonutil.SubmittedAtPodGroupLabel] != "1" { + t.Fatalf("the job's own labels must not be mutated; got %v", job.Labels) + } +} + +func TestJobSubmittedAtIsSecondAlignedEpochMillis(t *testing.T) { + created := metav1.NewTime(time.Date(2026, 1, 1, 0, 0, 42, 0, time.UTC)) + job := &batch.Job{ObjectMeta: metav1.ObjectMeta{CreationTimestamp: created}} + stamp := jobSubmittedAt(job) + millis, err := strconv.ParseInt(stamp, 10, 64) + if err != nil { + t.Fatalf("stamp %q must be an integer: %v", stamp, err) + } + // creationTimestamp has second granularity, matching the scheduler plugin's + // same-second-simultaneous comparison. + if millis%1000 != 0 || millis/1000 != created.Unix() { + t.Fatalf("stamp %d must be the creation second in milliseconds (%d)", millis, created.Unix()*1000) + } +} diff --git a/pkg/controllers/job/state/factory.go b/pkg/controllers/job/state/factory.go index 2b5f3ba6f47..3b181024431 100644 --- a/pkg/controllers/job/state/factory.go +++ b/pkg/controllers/job/state/factory.go @@ -74,9 +74,22 @@ type Target struct { type Action struct { Action v1alpha1.Action + // Event is the lifecycle event that triggered the action (empty for actions + // issued directly, e.g. via vcctl or Commands). + Event v1alpha1.Event Target Target } +// CountsTowardRetry reports whether executing a restart for this action should +// consume the job's retry budget (status.RetryCount vs spec.maxRetry). Restarts +// triggered by PodEvicted do not: an evicted pod means the job was disrupted +// (scheduler preemption/reclaim deletes pods), not that it failed — burning +// retries for disruptions lets repeated preemption kill an innocent job +// permanently. Everything else (PodFailed, direct restarts) counts as before. +func (a Action) CountsTowardRetry() bool { + return a.Event != v1alpha1.PodEvictedEvent +} + // State interface. type State interface { // Execute executes the actions based on current state. diff --git a/pkg/controllers/job/state/pending.go b/pkg/controllers/job/state/pending.go index 19eb2bd8dfc..fc591600eb6 100644 --- a/pkg/controllers/job/state/pending.go +++ b/pkg/controllers/job/state/pending.go @@ -30,13 +30,17 @@ func (ps *pendingState) Execute(action Action) error { switch action.Action { case v1alpha1.RestartJobAction: return KillJob(ps.job, PodRetainPhaseNone, func(status *vcbatch.JobStatus) bool { - status.RetryCount++ + if action.CountsTowardRetry() { + status.RetryCount++ + } status.State.Phase = vcbatch.Restarting return true }) case v1alpha1.RestartTaskAction, v1alpha1.RestartPodAction, v1alpha1.RestartPartitionAction: return KillTarget(ps.job, action.Target, func(status *vcbatch.JobStatus) bool { - status.RetryCount++ + if action.CountsTowardRetry() { + status.RetryCount++ + } status.State.Phase = vcbatch.Restarting return true }) diff --git a/pkg/controllers/job/state/retry_test.go b/pkg/controllers/job/state/retry_test.go new file mode 100644 index 00000000000..af3bccb6ef6 --- /dev/null +++ b/pkg/controllers/job/state/retry_test.go @@ -0,0 +1,90 @@ +/* +Copyright 2026 The Volcano Authors. + +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. +*/ + +// Tests for the retry-budget accounting of restart actions: restarts triggered by +// PodEvicted (disruptions — the scheduler's preempt/reclaim delete pods) must not +// consume status.RetryCount, while PodFailed-triggered and directly-issued +// restarts must keep counting exactly as before. +package state + +import ( + "testing" + + vcbatch "volcano.sh/apis/pkg/apis/batch/v1alpha1" + "volcano.sh/apis/pkg/apis/bus/v1alpha1" + + "volcano.sh/volcano/pkg/controllers/apis" +) + +// stubKills replaces the KillJob/KillTarget hooks with stubs that run the +// UpdateStatusFn against status and restores them on cleanup. +func stubKills(t *testing.T, status *vcbatch.JobStatus) { + savedKillJob, savedKillTarget := KillJob, KillTarget + t.Cleanup(func() { KillJob, KillTarget = savedKillJob, savedKillTarget }) + KillJob = func(job *apis.JobInfo, podRetainPhase PhaseMap, fn UpdateStatusFn) error { + if fn != nil { + fn(status) + } + return nil + } + KillTarget = func(job *apis.JobInfo, target Target, fn UpdateStatusFn) error { + if fn != nil { + fn(status) + } + return nil + } +} + +func TestEvictionTriggeredRestartsDoNotCountTowardRetry(t *testing.T) { + states := map[string]func(job *apis.JobInfo) State{ + "running": func(job *apis.JobInfo) State { return &runningState{job: job} }, + "pending": func(job *apis.JobInfo) State { return &pendingState{job: job} }, + } + actions := []v1alpha1.Action{v1alpha1.RestartJobAction, v1alpha1.RestartTaskAction} + + for stateName, newState := range states { + for _, action := range actions { + var status vcbatch.JobStatus + stubKills(t, &status) + st := newState(&apis.JobInfo{}) + + if err := st.Execute(Action{Action: action, Event: v1alpha1.PodEvictedEvent}); err != nil { + t.Fatalf("%s/%s evicted: %v", stateName, action, err) + } + if status.RetryCount != 0 { + t.Fatalf("%s/%s: a PodEvicted-triggered restart must not consume a retry (got %d)", + stateName, action, status.RetryCount) + } + + if err := st.Execute(Action{Action: action, Event: v1alpha1.PodFailedEvent}); err != nil { + t.Fatalf("%s/%s failed: %v", stateName, action, err) + } + if status.RetryCount != 1 { + t.Fatalf("%s/%s: a PodFailed-triggered restart must consume a retry (got %d)", + stateName, action, status.RetryCount) + } + + // Directly-issued restarts (vcctl, Commands) carry no event and keep counting. + if err := st.Execute(Action{Action: action}); err != nil { + t.Fatalf("%s/%s direct: %v", stateName, action, err) + } + if status.RetryCount != 2 { + t.Fatalf("%s/%s: a direct restart must consume a retry (got %d)", + stateName, action, status.RetryCount) + } + } + } +} diff --git a/pkg/controllers/job/state/running.go b/pkg/controllers/job/state/running.go index 7a85d909d42..b93bc7a3f48 100644 --- a/pkg/controllers/job/state/running.go +++ b/pkg/controllers/job/state/running.go @@ -35,13 +35,17 @@ func (ps *runningState) Execute(action Action) error { case v1alpha1.RestartJobAction: return KillJob(ps.job, PodRetainPhaseNone, func(status *vcbatch.JobStatus) bool { status.State.Phase = vcbatch.Restarting - status.RetryCount++ + if action.CountsTowardRetry() { + status.RetryCount++ + } return true }) case v1alpha1.RestartTaskAction, v1alpha1.RestartPodAction, v1alpha1.RestartPartitionAction: return KillTarget(ps.job, action.Target, func(status *vcbatch.JobStatus) bool { status.State.Phase = vcbatch.Restarting - status.RetryCount++ + if action.CountsTowardRetry() { + status.RetryCount++ + } return true }) case v1alpha1.AbortJobAction: diff --git a/pkg/controllers/podstartwatchdog/podstartwatchdog.go b/pkg/controllers/podstartwatchdog/podstartwatchdog.go new file mode 100644 index 00000000000..09329439ae7 --- /dev/null +++ b/pkg/controllers/podstartwatchdog/podstartwatchdog.go @@ -0,0 +1,303 @@ +/* +Copyright 2026 The Volcano Authors. + +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 podstartwatchdog marks pods Failed when they have been bound to a node +// but never start, releasing the capacity the stuck pods hold hostage. +// +// A placed pod that cannot start — unpullable image, unmountable volume, +// container-create config error — never *fails* on its own: it sits Pending on +// its node, charging its full resource request against the cluster forever, and +// no `PodFailed` lifecycle policy ever fires. The stock `PodPending`-timeout +// policy cannot be used instead because its clock starts at pod creation — on a +// pass-through-enqueue cluster pods legitimately queue for hours, so it would +// kill innocent queued jobs. This watchdog anchors on **binding**: a pod counts +// as stuck only when bound (`spec.nodeName` set — it already holds capacity) and +// still Pending, whatever the cause. The clock is the pod's own bind time — the +// PodScheduled=True condition the API server writes together with spec.nodeName +// (see boundAt below) — so it survives watchdog restarts and never includes +// time spent queued. The deadline is the pod's own volcano.sh/pod-start-timeout +// label (see deadline below), copied onto every pod by the job controller +// from the owning VCJob's label, which the platform's admission webhook defaults +// onto every job. +// +// Stuck past the deadline, the pod is marked **Failed** (status.phase, reason +// PodStartTimeout) with a Warning Event. Writing a terminal phase from outside +// the kubelet is a one-way door the platform honors end to end: the API server +// does not restrict phase transitions in status updates, the kubelet begins +// teardown when it observes an API-terminal pod (pod_workers.go: "Pod is in a +// terminal phase (success/failed), begin teardown" — the same contract PodGC +// relies on), and the Volcano job controller classifies the transition as an +// ordinary `PodFailed` event. Everything downstream is the job's own lifecycle +// policy: `PodFailed → RestartJob` (the platform default) tears down and requeues +// the gang, billing one retry per lap, and `maxRetry` exhaustion parks the job in +// terminal Failed — bounded, inspectable, resubmittable. +// +// The watchdog is deliberately **pod-scoped and stateless**: it never resolves, +// verifies, or trusts a pod→job mapping — attribution is the job controller's +// business, done with its own ownership and version safeguards. There is no +// targeting decision to spoof: a hand-made pod carrying the label only ever +// condemns itself. And a Failed pod is terminal, so nothing is ever re-examined — +// no tracking maps, no dedup, nothing to lose on restart. +// +// Cause-agnostic cuts both ways: time spent in LEGITIMATE initialization — a cold +// pull of a huge image, a long-running init container (the pod stays Pending +// until init containers finish) — also counts against the deadline, because from +// the outside it is indistinguishable from an init that will never finish. Jobs +// whose normal startup can approach the deadline must carry a larger label +// value, and the deployment default must be sized against worst-case cold-start +// time, not just against obviously-broken pods. +package podstartwatchdog + +import ( + "context" + "encoding/json" + "fmt" + "time" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/selection" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/wait" + "k8s.io/client-go/kubernetes" + corelisters "k8s.io/client-go/listers/core/v1" + "k8s.io/client-go/tools/cache" + "k8s.io/client-go/tools/record" + "k8s.io/klog/v2" + + "github.com/spf13/pflag" + + "k8s.io/client-go/informers" + clientgoscheme "k8s.io/client-go/kubernetes/scheme" + corev1 "k8s.io/client-go/kubernetes/typed/core/v1" + + "volcano.sh/volcano/pkg/controllers/framework" + commonutil "volcano.sh/volcano/pkg/util" +) + +func init() { + framework.RegisterController(&podStartWatchdogController{}) +} + +type podStartWatchdogController struct { + kubeClient kubernetes.Interface + + informerFactory informers.SharedInformerFactory + + podLister corelisters.PodLister + podSynced cache.InformerSynced + + recorder record.EventRecorder + + // timeoutLabelExists selects only pods carrying the volcano.sh/pod-start-timeout + // label — the job controller stamps it (from the job's label) on every pod it + // creates; unlabeled pods are not policed and not even listed. + timeoutLabelExists labels.Selector + + interval time.Duration +} + +func (pw *podStartWatchdogController) Name() string { + return "podstartwatchdog-controller" +} + +// AddFlags implements framework.FlagProvider. There is deliberately no timeout +// flag: deadlines come exclusively from the per-pod label. +func (pw *podStartWatchdogController) AddFlags(fs *pflag.FlagSet) { + fs.DurationVar(&pw.interval, "pod-start-interval", time.Minute, + "How often the pod-start watchdog evaluates bound pods that have not started") +} + +func (pw *podStartWatchdogController) Initialize(opt *framework.ControllerOption) error { + pw.kubeClient = opt.KubeClient + + pw.informerFactory = opt.SharedInformerFactory + podInformer := opt.SharedInformerFactory.Core().V1().Pods() + pw.podLister = podInformer.Lister() + pw.podSynced = podInformer.Informer().HasSynced + + eventBroadcaster := record.NewBroadcaster() + eventBroadcaster.StartLogging(klog.Infof) + eventBroadcaster.StartRecordingToSink(&corev1.EventSinkImpl{Interface: pw.kubeClient.CoreV1().Events("")}) + // The client-go scheme, not the volcano one: events target core-v1 Pods, and a + // recorder can only reference kinds its scheme registers. + pw.recorder = eventBroadcaster.NewRecorder(clientgoscheme.Scheme, v1.EventSource{Component: pw.Name()}) + + requirement, err := labels.NewRequirement(commonutil.PodStartTimeoutLabel, selection.Exists, nil) + if err != nil { + return err + } + pw.timeoutLabelExists = labels.NewSelector().Add(*requirement) + return nil +} + +func (pw *podStartWatchdogController) Run(stopCh <-chan struct{}) { + pw.informerFactory.Start(stopCh) + if !cache.WaitForCacheSync(stopCh, pw.podSynced) { + klog.Errorf("podstartwatchdog: caches failed to sync") + return + } + + klog.Infof("Pod-start watchdog started: failing pods bound-but-unstarted past their %s label (interval %s)", + commonutil.PodStartTimeoutLabel, pw.interval) + go wait.Until(pw.evaluate, pw.interval, stopCh) + <-stopCh + klog.Infof("Shutting down pod-start watchdog") +} + +// evaluate is one stateless pass: gather labeled pods from the informer cache, +// decide the kills (the pure decision logic at the bottom of this file), and +// mark each condemned pod Failed. +func (pw *podStartWatchdogController) evaluate() { + pods, err := pw.podLister.List(pw.timeoutLabelExists) + if err != nil { + klog.Errorf("podstartwatchdog: listing pods failed: %v", err) + return + } + for _, k := range kills(pods, time.Now()) { + pw.fail(k) + } +} + +// fail marks the pod Failed via the status subresource and records why on the pod. +// The write is a one-way door (phase never regresses; the kubelet tears down on +// observing it), so a pass that races anything — the pod starting at the last +// instant, a concurrent kubelet status sync — either loses harmlessly and retries +// next interval, or lands and is final. Everything job-level (gang restart, retry +// billing, terminal failure) follows from the job's own lifecycle policies. +func (pw *podStartWatchdogController) fail(k kill) { + patch, err := json.Marshal(map[string]interface{}{ + "status": map[string]interface{}{ + "phase": v1.PodFailed, + "reason": "PodStartTimeout", + "message": fmt.Sprintf( + "bound to node %s but unable to start for %s (limit %s); marked Failed by the pod-start watchdog", + k.pod.Spec.NodeName, k.stuckFor.Round(time.Second), k.limit), + }, + }) + if err != nil { + klog.Errorf("podstartwatchdog: marshaling status patch for pod %s/%s failed: %v", k.pod.Namespace, k.pod.Name, err) + return + } + _, err = pw.kubeClient.CoreV1().Pods(k.pod.Namespace).Patch( + context.TODO(), k.pod.Name, types.MergePatchType, patch, metav1.PatchOptions{}, "status") + if err != nil { + klog.Errorf("podstartwatchdog: failed to mark pod %s/%s Failed: %v", k.pod.Namespace, k.pod.Name, err) + return + } + pw.recorder.Eventf(k.pod, v1.EventTypeWarning, "PodStartTimeout", + "Marked Failed by the pod-start watchdog: bound to a node but unable to start for %ds (limit %ds); the pod was holding capacity it can never use. The owning job's lifecycle policy decides what happens next (PodFailed -> RestartJob consumes one retry per attempt).", + int(k.stuckFor.Seconds()), int(k.limit.Seconds())) + klog.Warningf("podstartwatchdog: marked pod %s/%s Failed (bound to %s but unstarted for %s, limit %s)", + k.pod.Namespace, k.pod.Name, k.pod.Spec.NodeName, k.stuckFor.Round(time.Second), k.limit) +} + +// ---- Decision logic — pure, stateless, and clock-free below this line ----------- + +// stuckPending reports whether the pod is bound to a node yet has not started — +// cause-agnostic: unpullable (or merely slow-pulling) images, unmountable volumes, +// container-create config errors, init containers still running — anything that +// keeps a placed pod from reaching Running counts, because the pod charges its +// full resource request either way. +// +// An unbound Pending pod is a *queued* pod — it holds no capacity and must never +// count, however long it waits; binding is what anchors the clock. Running pods are +// out of scope by design: with restartPolicy Never (the platform default), any +// container death fails the pod on its own. Terminating pods (deletionTimestamp +// set) are already on their way out and are not the watchdog's to judge. Failed is +// terminal, so a pod this watchdog has marked is never re-examined — idempotency +// falls out of the phase itself, with no bookkeeping. +func stuckPending(pod *v1.Pod) bool { + if pod.Spec.NodeName == "" { + return false + } + if pod.DeletionTimestamp != nil { + return false + } + return pod.Status.Phase == v1.PodPending +} + +// boundAt is the time the pod was bound to its node — the durable clock anchor. +// The API server's binding subresource sets the PodScheduled=True condition in the +// SAME write that sets spec.nodeName (BindingREST.setPodNodeAndMetadata), so every +// scheduler-bound pod carries the bind moment in that condition's +// lastTransitionTime from the instant it holds capacity; nodeName is immutable and +// pod phase never regresses, so the condition never flaps and "bound and Pending +// now" means "continuously unstarted since boundAt". The fallback — a bound pod +// WITHOUT the condition — can only be one created with nodeName preset (it never +// passed through binding), which also never queued, so its creationTimestamp IS +// its bind time. Either way the clock excludes time spent queued and lives on the +// pod itself, so a watchdog restart forgets nothing. +func boundAt(pod *v1.Pod) time.Time { + for _, cond := range pod.Status.Conditions { + if cond.Type == v1.PodScheduled && cond.Status == v1.ConditionTrue { + return cond.LastTransitionTime.Time + } + } + return pod.CreationTimestamp.Time +} + +// deadline is the pod's own start deadline: the volcano.sh/pod-start-timeout label +// ON THE POD — the ONLY deadline source. The job controller copies the label from +// the owning VCJob onto every pod it creates (and the platform's admission webhook +// defaults it onto every job), so the deadline is stamped by a trusted component; +// a pod without the label, or with an unparseable or non-positive value ("0" is an +// explicit opt-out), is not policed. Deliberately pod-scoped: the watchdog never +// resolves — or trusts — a pod→job mapping; a hand-made pod carrying the label +// only ever condemns itself. +func deadline(pod *v1.Pod) (time.Duration, bool) { + value, found := pod.Labels[commonutil.PodStartTimeoutLabel] + if !found { + return 0, false + } + timeout, err := time.ParseDuration(value) + if err != nil || timeout <= 0 { + return 0, false + } + return timeout, true +} + +// kill names one pod to mark Failed and the numbers that condemned it. +type kill struct { + pod *v1.Pod + stuckFor time.Duration + limit time.Duration +} + +// kills is one evaluation pass: every pod that has been bound yet not started for +// at least its own deadline label is condemned — individually; what a failed pod +// means for its job (gang restart, retry billing, terminal failure) is entirely +// the job's own lifecycle policies' business, decided in the job controller. Pure +// and stateless: the clock is the pod's own bind time (boundAt), the deadline the +// pod's own label, so a watchdog restart forgets nothing and there is nothing to +// deduplicate — marking a pod Failed is terminal and self-idempotent. +func kills(pods []*v1.Pod, now time.Time) []kill { + var out []kill + for _, pod := range pods { + if !stuckPending(pod) { + continue + } + limit, policed := deadline(pod) + if !policed { + continue + } + if stuckFor := now.Sub(boundAt(pod)); stuckFor >= limit { + out = append(out, kill{pod: pod, stuckFor: stuckFor, limit: limit}) + } + } + return out +} diff --git a/pkg/controllers/podstartwatchdog/podstartwatchdog_test.go b/pkg/controllers/podstartwatchdog/podstartwatchdog_test.go new file mode 100644 index 00000000000..783d6e3dc7d --- /dev/null +++ b/pkg/controllers/podstartwatchdog/podstartwatchdog_test.go @@ -0,0 +1,242 @@ +/* +Copyright 2026 The Volcano Authors. + +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. +*/ + +// Unit tests for the pod-start watchdog's decision logic. +// +// The property that matters most is the negative: a pod waiting in the *scheduling +// queue* (Pending but unbound) must never count as stuck, however long it waits — +// on a pass-through-enqueue cluster pods legitimately queue for hours, which is +// exactly why the stock VCJob pending-timeout policy could not be used. Only +// bound-yet-unstarted pods, unstarted past their own deadline label since their +// BIND time (the PodScheduled=True condition, never the creation time of a +// scheduler-bound pod), are marked Failed — whatever kept them from starting. +package podstartwatchdog + +import ( + "testing" + "time" + + v1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + + commonutil "volcano.sh/volcano/pkg/util" +) + +var testEpoch = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + +func at(seconds float64) time.Time { + return testEpoch.Add(time.Duration(seconds * float64(time.Second))) +} + +type podSpec struct { + uid string + timeout string // the volcano.sh/pod-start-timeout label; "" = unlabeled + node string + phase v1.PodPhase + waitingReason string + terminating bool + createdAt float64 // seconds after testEpoch + boundAtSec float64 // seconds after testEpoch (PodScheduled=True lastTransitionTime) + // noBindCondition models a pod created with nodeName preset: bound, but it + // never passed through the binding subresource, so no PodScheduled condition. + noBindCondition bool +} + +func pod(spec podSpec) *v1.Pod { + if spec.uid == "" { + spec.uid = "pod-1" + } + p := &v1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + UID: types.UID(spec.uid), + Name: spec.uid + "-worker-0", + Namespace: "alice", + Labels: map[string]string{}, + CreationTimestamp: metav1.NewTime(at(spec.createdAt)), + }, + Spec: v1.PodSpec{NodeName: spec.node}, + Status: v1.PodStatus{Phase: spec.phase}, + } + if spec.timeout != "" { + p.Labels[commonutil.PodStartTimeoutLabel] = spec.timeout + } + if spec.node != "" && !spec.noBindCondition { + p.Status.Conditions = []v1.PodCondition{{ + Type: v1.PodScheduled, + Status: v1.ConditionTrue, + LastTransitionTime: metav1.NewTime(at(spec.boundAtSec)), + }} + } + if spec.waitingReason != "" { + p.Status.ContainerStatuses = []v1.ContainerStatus{ + {State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{Reason: spec.waitingReason}}}, + } + } + if spec.terminating { + now := metav1.NewTime(testEpoch) + p.DeletionTimestamp = &now + } + return p +} + +// stuckPod is the canonical bound-but-unstarted pod: bound at the test epoch, +// policed with a 30m deadline. +func stuckPod() *v1.Pod { + return pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending}) +} + +// --- stuckPending: what counts as "bound but never started" ---------------------- + +func TestQueuedPodIsNeverStuckNoMatterHowLongItWaits(t *testing.T) { + // Unbound + Pending is the queue — the case the watchdog exists to NOT touch. + if stuckPending(pod(podSpec{timeout: "30m", phase: v1.PodPending, waitingReason: "ImagePullBackOff"})) { + t.Fatal("an unbound pod must never count as stuck") + } +} + +func TestBoundPendingPodIsStuckWhateverTheCause(t *testing.T) { + // Cause-agnostic: unpullable image, unmountable volume, config error, running + // init containers, or no container status at all — bound and Pending is the + // whole predicate. Legitimate-but-slow initialization counts too, by design. + for _, reason := range []string{"", "ImagePullBackOff", "CreateContainerConfigError", "ContainerCreating", "PodInitializing"} { + p := pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending, waitingReason: reason}) + if !stuckPending(p) { + t.Fatalf("bound Pending pod (reason %q) must count as stuck", reason) + } + } +} + +func TestStartedAndTerminalAndTerminatingPodsAreNotStuck(t *testing.T) { + // Failed being terminal is also the watchdog's idempotency: a pod it marked + // Failed can never be condemned again. + cases := map[string]*v1.Pod{ + "running": pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodRunning}), + "succeeded": pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodSucceeded}), + "failed": pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodFailed}), + "terminating": pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending, terminating: true}), + } + for name, p := range cases { + if stuckPending(p) { + t.Fatalf("%s pod must not count as stuck", name) + } + } +} + +// --- boundAt: the clock anchor --------------------------------------------------- + +func TestBoundAtIsTheBindConditionNeverTheCreationTime(t *testing.T) { + // A scheduler-bound pod that queued for an hour: the clock must anchor on the + // PodScheduled=True transition (bind), not on creation — queue time is free. + p := pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending, createdAt: 0, boundAtSec: 3600}) + if !boundAt(p).Equal(at(3600)) { + t.Fatalf("boundAt = %v, want the bind condition time %v", boundAt(p), at(3600)) + } +} + +func TestBoundAtFallsBackToCreationForPresetNodePods(t *testing.T) { + // A pod created with nodeName preset never passed through binding — and never + // queued — so its creationTimestamp IS its bind time. + p := pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending, createdAt: 500, noBindCondition: true}) + if !boundAt(p).Equal(at(500)) { + t.Fatalf("boundAt = %v, want the creation time %v", boundAt(p), at(500)) + } +} + +// --- deadline: the pod's own label is the only deadline source ------------------- + +func TestDeadlineComesFromThePodLabelAlone(t *testing.T) { + limit, policed := deadline(pod(podSpec{timeout: "45m", node: "node-0", phase: v1.PodPending})) + if !policed || limit != 45*time.Minute { + t.Fatalf("deadline = (%v, %v), want (45m, true)", limit, policed) + } + + unpoliced := map[string]podSpec{ + "unlabeled": {node: "node-0", phase: v1.PodPending}, + "malformed": {timeout: "soon", node: "node-0", phase: v1.PodPending}, + "zero": {timeout: "0", node: "node-0", phase: v1.PodPending}, + "negative": {timeout: "-5m", node: "node-0", phase: v1.PodPending}, + } + for name, spec := range unpoliced { + if _, policed := deadline(pod(spec)); policed { + t.Fatalf("%s pod must not be policed", name) + } + } +} + +// --- kills: the per-pass decision ------------------------------------------------ + +func TestStuckPodIsCondemnedOnlyPastItsDeadline(t *testing.T) { + if got := kills([]*v1.Pod{stuckPod()}, at(1799)); len(got) != 0 { + t.Fatalf("still under the limit; kills=%v", got) + } + + // A single stateless pass decides: the clock is the pod's own bind time, so a + // watchdog restarted five seconds ago condemns an hour-stuck pod immediately. + got := kills([]*v1.Pod{stuckPod()}, at(1800)) + if len(got) != 1 || got[0].pod.Name != "pod-1-worker-0" || + got[0].stuckFor != 30*time.Minute || got[0].limit != 30*time.Minute { + t.Fatalf("past the limit the pod must be condemned with the bind-anchored stuck time; kills=%+v", got) + } +} + +func TestQueueTimeIsNeverCounted(t *testing.T) { + // Bound at t=3600 after an hour in the queue: the deadline runs from the bind, + // so at 3600+1799 the pod is safe and at 3600+1800 it is not. + p := pod(podSpec{timeout: "30m", node: "node-0", phase: v1.PodPending, createdAt: 0, boundAtSec: 3600}) + if got := kills([]*v1.Pod{p}, at(3600+1799)); len(got) != 0 { + t.Fatalf("queue time must not count toward the deadline; kills=%v", got) + } + if got := kills([]*v1.Pod{p}, at(3600+1800)); len(got) != 1 { + t.Fatalf("the deadline must run from the bind time; kills=%v", got) + } +} + +func TestEachStuckPodIsCondemnedIndividually(t *testing.T) { + // Pod-scoped by design: the watchdog knows nothing of jobs or gangs. Two stuck + // pods are two kills — collapsing a gang's failures into one restart is the + // job controller's business (the job state machine bills at most one retry per + // restart lap). The queued pod must survive. + pods := []*v1.Pod{ + pod(podSpec{uid: "pod-a", timeout: "30m", node: "node-0", phase: v1.PodPending}), + pod(podSpec{uid: "pod-b", timeout: "30m", node: "node-1", phase: v1.PodPending}), + pod(podSpec{uid: "pod-c", timeout: "30m", phase: v1.PodPending}), // queued + } + got := kills(pods, at(1800)) + if len(got) != 2 { + t.Fatalf("both bound stuck pods and only they must be condemned; kills=%+v", got) + } +} + +func TestPerPodDeadlinesAreHonoredInOnePass(t *testing.T) { + pods := []*v1.Pod{ + pod(podSpec{uid: "pod-fast", timeout: "10m", node: "node-0", phase: v1.PodPending}), + pod(podSpec{uid: "pod-slow", timeout: "30m", node: "node-0", phase: v1.PodPending}), + } + got := kills(pods, at(600)) + if len(got) != 1 || got[0].pod.Name != "pod-fast-worker-0" || got[0].limit != 10*time.Minute { + t.Fatalf("only the short-deadline pod may be condemned at t=600, with its limit recorded; got %+v", got) + } +} + +func TestUnpolicedPodsAreNeverCondemned(t *testing.T) { + // The pod label is the ONLY deadline source: unlabeled (or opted-out) pods are + // ignored however stuck they are. + p := pod(podSpec{node: "node-0", phase: v1.PodPending}) + if got := kills([]*v1.Pod{p}, at(999999)); len(got) != 0 { + t.Fatalf("an unpoliced pod must be ignored entirely; kills=%v", got) + } +} diff --git a/pkg/scheduler/plugins/factory.go b/pkg/scheduler/plugins/factory.go index 78ace4b7243..aa713055b6c 100644 --- a/pkg/scheduler/plugins/factory.go +++ b/pkg/scheduler/plugins/factory.go @@ -29,6 +29,7 @@ import ( "volcano.sh/volcano/pkg/scheduler/plugins/deviceshare" "volcano.sh/volcano/pkg/scheduler/plugins/drf" "volcano.sh/volcano/pkg/scheduler/plugins/extender" + "volcano.sh/volcano/pkg/scheduler/plugins/fifopriority" "volcano.sh/volcano/pkg/scheduler/plugins/gang" networktopologyaware "volcano.sh/volcano/pkg/scheduler/plugins/network-topology-aware" "volcano.sh/volcano/pkg/scheduler/plugins/nodegroup" @@ -55,6 +56,7 @@ func init() { framework.RegisterPluginBuilder(deviceshare.PluginName, deviceshare.New) framework.RegisterPluginBuilder(predicates.PluginName, predicates.New) framework.RegisterPluginBuilder(priority.PluginName, priority.New) + framework.RegisterPluginBuilder(fifopriority.PluginName, fifopriority.New) framework.RegisterPluginBuilder(nodeorder.PluginName, nodeorder.New) framework.RegisterPluginBuilder(conformance.PluginName, conformance.New) framework.RegisterPluginBuilder(binpack.PluginName, binpack.New) diff --git a/pkg/scheduler/plugins/fifopriority/fifopriority.go b/pkg/scheduler/plugins/fifopriority/fifopriority.go new file mode 100644 index 00000000000..9958dc9dedf --- /dev/null +++ b/pkg/scheduler/plugins/fifopriority/fifopriority.go @@ -0,0 +1,256 @@ +/* +Copyright 2026 The Volcano Authors. + +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 fifopriority is a replacement for the in-tree `priority` plugin that orders +jobs by PriorityClass first and submission time second. + +The one comparator (jobOrder below) drives both of the decisions the stock plugin makes +from PriorityClass alone: + + - job order: allocate processes (and refills freed capacity for) higher-class jobs + first, and within a class older jobs first; + - preempt victims: a pending job may evict strictly-lower-class work, or equal-class + work whose job was created strictly later. + +Sharing the comparator between ordering and eviction is the point, not a convenience. +If eviction used a finer or different order than allocation (as an out-of-process +victim filter must, having no ordering hook), allocate could refill an evicted slot +with a job the eviction rule still considers junior, and the next session would evict +again — an observed livelock, not a hypothetical. With one order, every refill is at +least as senior as the job the eviction was for, so each eviction strictly improves +the running set. That guarantee is per-session: victims drain asynchronously, so +junior work can still slip into a freed slot between sessions and be re-evicted — +churn bounded by the pending-junior population, not a livelock, but not zero either. + +The clock: the volcano.sh/submitted-at PodGroup label (epoch milliseconds), stamped +by the job controller from the owning VCJob's metadata.creationTimestamp — an +API-server-managed, immutable field, so a place in line cannot be forged — on every +PodGroup it creates and RE-creates. That restart-proofness is why the label exists +at all: the fallback clock, the PodGroup's own creationTimestamp, resets on +RestartJob (the controller deletes and recreates the PodGroup), which would silently +send a restarted job — even one restarted by infrastructure noise like a kubelet +OutOfcpu rejection — to the back of its class's line. PodGroups without a parseable +stamp (not created by the job controller, e.g. for bare pods) fall back to the +PodGroup creation time. + +Granularity: stamps are compared at one-second precision, matching the fallback +clock's. Jobs submitted in the same second are simultaneous — they never evict each +other, and their relative order falls through to later tiers (drf share fairness, +then the session default). + +Everything else replicates the stock priority plugin: task order and sub-job order by +priority value, the same starving-job test, and Permit/tier semantics — so this +plugin slots into the scheduler config exactly where `priority` was. +*/ +package fifopriority + +import ( + "strconv" + + "k8s.io/klog/v2" + + "volcano.sh/apis/pkg/apis/scheduling" + + "volcano.sh/volcano/pkg/scheduler/api" + "volcano.sh/volcano/pkg/scheduler/framework" + "volcano.sh/volcano/pkg/scheduler/plugins/util" + commonutil "volcano.sh/volcano/pkg/util" +) + +// PluginName is the name the scheduler configuration refers to this plugin by. +const PluginName = "fifopriority" + +// SubmittedAtLabel is the submission time (the owning VCJob's immutable +// creationTimestamp, epoch milliseconds) the job controller stamps onto every +// PodGroup it (re)creates. +const SubmittedAtLabel = commonutil.SubmittedAtPodGroupLabel + +type fifoPriorityPlugin struct{} + +// New returns a fifopriority plugin instance (framework.PluginBuilder). +func New(arguments framework.Arguments) framework.Plugin { + return &fifoPriorityPlugin{} +} + +func (pp *fifoPriorityPlugin) Name() string { + return PluginName +} + +// submitSecond is the job's submission time at second granularity: the job +// controller's stamp (the owning VCJob's creationTimestamp) when present and +// parseable, else the PodGroup's creation second. The stamp survives RestartJob +// (the controller re-stamps every PodGroup it recreates); the fallback does not — +// see the package doc. +func submitSecond(ji *api.JobInfo) int64 { + if ji.PodGroup != nil { + if value, found := ji.PodGroup.Labels[SubmittedAtLabel]; found { + if millis, err := strconv.ParseInt(value, 10, 64); err == nil { + return millis / 1000 + } + } + } + return ji.CreationTimestamp.Unix() +} + +// jobOrder is the single comparator everything in this plugin derives from: +// higher PriorityClass first, then earlier submission second; 0 only when both tie +// (same class, same second), leaving that tie to later tiers. +func jobOrder(l, r *api.JobInfo) int { + if l.Priority != r.Priority { + if l.Priority > r.Priority { + return -1 + } + return 1 + } + lSecond, rSecond := submitSecond(l), submitSecond(r) + if lSecond < rSecond { + return -1 + } + if lSecond > rSecond { + return 1 + } + return 0 +} + +// enqueueableQueueState reports whether a job may enqueue into its queue: the queue +// must exist in the session and be Open — the same state check stock proportion +// carries inside its JobEnqueueableFn (with a nil-guard stock lacks), reinstated +// here because disabling proportion's enqueue-stage capacity vote +// (enableJobEnqueued: false, required for the open-gates design) also lost the +// state check, letting jobs enqueue into Closed/Closing queues and materialize pod +// sets that can never allocate. State only — capacity is deliberately never gated +// at enqueue. +func enqueueableQueueState(queue *api.QueueInfo) bool { + return queue != nil && queue.Queue.Status.State == scheduling.QueueStateOpen +} + +// preemptVictims applies the eviction rule to candidate preemptees (already filtered by +// the preempt action to same-queue, preemptable, running tasks). Between distinct jobs a +// candidate is a victim iff the preemptor's job strictly precedes its job in jobOrder — +// strictly lower class, or equal class and submitted in a strictly later second. Within +// one job the stock rule stands: strictly lower task (pod) priority. +func preemptVictims(jobs map[api.JobID]*api.JobInfo, preemptor *api.TaskInfo, preemptees []*api.TaskInfo) []*api.TaskInfo { + preemptorJob, found := jobs[preemptor.Job] + if !found { + // Orphaned task from a deleted PodGroup; nothing sane to decide (stock behavior). + return nil + } + var victims []*api.TaskInfo + for _, preemptee := range preemptees { + preempteeJob, found := jobs[preemptee.Job] + if !found { + continue + } + if preempteeJob.UID == preemptorJob.UID { + if preemptee.Priority < preemptor.Priority { + victims = append(victims, preemptee) + } + continue + } + if jobOrder(preemptorJob, preempteeJob) < 0 { + victims = append(victims, preemptee) + } + } + return victims +} + +func (pp *fifoPriorityPlugin) OnSessionOpen(ssn *framework.Session) { + // Task order: by pod priority value, exactly like the stock plugin. The FIFO + // submission-time tiebreak is deliberately NOT repeated at this level. Within a + // single job it would be meaningless — all of a job's tasks were submitted + // together. The one place tasks of DIFFERENT jobs are ordered together is the + // preempt action's victim queue, and there a tiebreak among equal-priority + // victims buys nothing: eviction stops as soon as the preemptor fits, and any + // subset of equal-priority victims that frees enough room is as good as any + // other. + ssn.AddTaskOrderFn(pp.Name(), func(l, r interface{}) int { + lv := l.(*api.TaskInfo) + rv := r.(*api.TaskInfo) + if lv.Priority == rv.Priority { + return 0 + } + if lv.Priority > rv.Priority { + return -1 + } + return 1 + }) + + ssn.AddJobOrderFn(pp.Name(), func(l, r interface{}) int { + return jobOrder(l.(*api.JobInfo), r.(*api.JobInfo)) + }) + + // Stock parity: sub-jobs carry only a priority; order by it. + ssn.AddSubJobOrderFn(pp.Name(), func(l, r interface{}) int { + lv := l.(*api.SubJobInfo) + rv := r.(*api.SubJobInfo) + if lv.Priority > rv.Priority { + return -1 + } + if lv.Priority < rv.Priority { + return 1 + } + return 0 + }) + + ssn.AddPreemptableFn(pp.Name(), func(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) ([]*api.TaskInfo, int) { + victims := preemptVictims(ssn.Jobs, preemptor, preemptees) + klog.V(4).Infof("[%s] preemptor <%s/%s>: %d victim(s) among %d candidates", + PluginName, preemptor.Namespace, preemptor.Name, len(victims), len(preemptees)) + return victims, util.Permit + }) + + // The unified eviction path (gang-aware preemption/reclaim) applies the same rule; + // gang reclaim permits all candidates, exactly like the stock plugin. + ssn.AddUnifiedEvictableFn(pp.Name(), func(evictCtx *api.EvictionContext, candidates []*api.TaskInfo) ([]*api.TaskInfo, int) { + if evictCtx.Kind == api.EvictionKindGangReclaim { + return candidates, util.Permit + } + var victims []*api.TaskInfo + for _, candidate := range candidates { + candidateJob, found := ssn.Jobs[candidate.Job] + if !found { + continue + } + if jobOrder(evictCtx.Job, candidateJob) < 0 { + victims = append(victims, candidate) + } + } + return victims, util.Permit + }) + + // Stock starving test: a job with tasks neither ready nor waiting is starving, which + // is what makes it a preemptor candidate in the preempt action. + ssn.AddJobStarvingFns(pp.Name(), func(obj interface{}) bool { + ji := obj.(*api.JobInfo) + return ji.ReadyTaskNum()+ji.WaitingTaskNum() < int32(len(ji.Tasks)) + }) + + // The queue-Open gate at enqueue (see enqueueableQueueState): jobs submitted to + // a queue that is not Open do not enqueue and stay podless. + ssn.AddJobEnqueueableFn(pp.Name(), func(obj interface{}) int { + job := obj.(*api.JobInfo) + queue := ssn.Queues[job.Queue] + if !enqueueableQueueState(queue) { + klog.V(3).Infof("[%s] queue of job <%s/%s> is missing or not open; rejecting enqueue", + PluginName, job.Namespace, job.Name) + return util.Reject + } + return util.Permit + }) +} + +func (pp *fifoPriorityPlugin) OnSessionClose(ssn *framework.Session) {} diff --git a/pkg/scheduler/plugins/fifopriority/fifopriority_test.go b/pkg/scheduler/plugins/fifopriority/fifopriority_test.go new file mode 100644 index 00000000000..2a714e224cf --- /dev/null +++ b/pkg/scheduler/plugins/fifopriority/fifopriority_test.go @@ -0,0 +1,191 @@ +/* +Copyright 2026 The Volcano Authors. + +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. +*/ + +// Unit tests for the fifopriority comparator and victim rule. +package fifopriority + +import ( + "strconv" + "testing" + "time" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "volcano.sh/apis/pkg/apis/scheduling" + "volcano.sh/volcano/pkg/scheduler/api" +) + +var epoch = time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + +// job builds a JobInfo whose PodGroup carries the controller-stamped submit time +// (epoch ms, relative to the test epoch). Pass stampMillis < 0 for an unstamped job +// (fallback to the PodGroup creation time). +func job(uid string, priority int32, createdSecondsAfterEpoch int, stampMillis int64) *api.JobInfo { + labels := map[string]string{} + if stampMillis >= 0 { + labels[SubmittedAtLabel] = strconv.FormatInt(epoch.UnixMilli()+stampMillis, 10) + } + return &api.JobInfo{ + UID: api.JobID(uid), + Priority: priority, + CreationTimestamp: metav1.NewTime(epoch.Add(time.Duration(createdSecondsAfterEpoch) * time.Second)), + PodGroup: &api.PodGroup{ + PodGroup: scheduling.PodGroup{ + ObjectMeta: metav1.ObjectMeta{Labels: labels}, + }, + }, + } +} + +func task(uid, jobUID string, priority int32) *api.TaskInfo { + return &api.TaskInfo{ + UID: api.TaskID(uid), + Job: api.JobID(jobUID), + Priority: priority, + } +} + +func TestJobOrderClassDominatesAge(t *testing.T) { + highYoung := job("high-young", 1000, 100, 100_000) + lowOld := job("low-old", 100, 0, 0) + if jobOrder(highYoung, lowOld) != -1 { + t.Fatalf("higher class must precede regardless of age") + } + if jobOrder(lowOld, highYoung) != 1 { + t.Fatalf("lower class must follow regardless of age") + } +} + +func TestJobOrderFIFOWithinClassByStamp(t *testing.T) { + older := job("older", 100, 0, 0) + younger := job("younger", 100, 30, 30_000) + sameSecond := job("same-second", 100, 0, 500) // 500ms later: same second, must tie + if jobOrder(older, younger) != -1 || jobOrder(younger, older) != 1 { + t.Fatalf("within a class, the earlier stamp must precede") + } + if jobOrder(older, sameSecond) != 0 { + t.Fatalf("same class + same submit second must tie (0), deferring to later tiers") + } +} + +func TestStampSurvivesRestartAndBeatsCreationTime(t *testing.T) { + // The reason the stamp exists: RestartJob recreates the PodGroup, resetting its + // creationTimestamp. A restarted old job (original stamp, fresh PodGroup) must + // still precede a younger job — the fresh creation time must be ignored. + restartedOld := job("restarted-old", 100, 500, 0) // stamped at t=0, PodGroup recreated at t=500s + younger := job("younger", 100, 100, 100_000) // stamped and created at t=100s + if jobOrder(restartedOld, younger) != -1 { + t.Fatalf("a restarted job must keep its original (stamped) place in line") + } +} + +func TestUnstampedAndMalformedFallBackToCreationTime(t *testing.T) { + unstamped := job("unstamped", 100, 10, -1) + stamped := job("stamped", 100, 60, 60_000) + if jobOrder(unstamped, stamped) != -1 { + t.Fatalf("an unstamped job compares by PodGroup creation time on the same timeline") + } + + malformed := job("malformed", 100, 10, -1) + malformed.PodGroup.Labels[SubmittedAtLabel] = "not-a-number" + if jobOrder(malformed, stamped) != -1 { + t.Fatalf("a malformed stamp must fall back to creation time, not break ordering") + } + + nilPodGroup := job("nil-pg", 100, 10, -1) + nilPodGroup.PodGroup = nil + if jobOrder(nilPodGroup, stamped) != -1 { + t.Fatalf("a job with no PodGroup must fall back to creation time") + } +} + +func TestPreemptVictimsBetweenJobs(t *testing.T) { + jobs := map[api.JobID]*api.JobInfo{ + "preemptor": job("preemptor", 100, 60, 60_000), + "lower-class": job("lower-class", 10, 0, 0), // older but lower class: victim + "younger": job("younger", 100, 120, 120_000), // same class, submitted later: victim + "older": job("older", 100, 0, 0), // same class, submitted earlier: protected + "same-second": job("same-second", 100, 60, 60_400), // same class, same second: protected + "higher": job("higher", 1000, 120, 120_000), // higher class: protected + // Restarted younger job: original stamp 120s, PodGroup recreated at 10s — the + // fresh creation time must not launder it into seniority. + "restarted-younger": job("restarted-younger", 100, 10, 120_000), + } + preemptor := task("t-preemptor", "preemptor", 100) + preemptees := []*api.TaskInfo{ + task("t-lower", "lower-class", 10), + task("t-younger", "younger", 100), + task("t-older", "older", 100), + task("t-same-second", "same-second", 100), + task("t-higher", "higher", 1000), + task("t-restarted-younger", "restarted-younger", 100), + task("t-orphan", "no-such-job", 100), // job missing from session: skipped + } + + victims := preemptVictims(jobs, preemptor, preemptees) + + got := map[api.TaskID]bool{} + for _, victim := range victims { + got[victim.UID] = true + } + want := map[api.TaskID]bool{"t-lower": true, "t-younger": true, "t-restarted-younger": true} + if len(got) != len(want) { + t.Fatalf("victims = %v, want exactly %v", got, want) + } + for uid := range want { + if !got[uid] { + t.Fatalf("expected victim %s missing; victims = %v", uid, got) + } + } +} + +func TestPreemptVictimsWithinOneJob(t *testing.T) { + jobs := map[api.JobID]*api.JobInfo{"gang": job("gang", 100, 0, 0)} + preemptor := task("t-high", "gang", 500) + preemptees := []*api.TaskInfo{ + task("t-low", "gang", 100), // lower task priority in the same job: victim (stock rule) + task("t-peer", "gang", 500), // equal task priority in the same job: protected + } + + victims := preemptVictims(jobs, preemptor, preemptees) + + if len(victims) != 1 || victims[0].UID != "t-low" { + t.Fatalf("within one job only strictly-lower task priority is evictable; got %v", victims) + } +} + +func TestPreemptVictimsOrphanedPreemptor(t *testing.T) { + jobs := map[api.JobID]*api.JobInfo{"other": job("other", 100, 120, 120_000)} + preemptor := task("t-orphan", "gone", 100) + if victims := preemptVictims(jobs, preemptor, []*api.TaskInfo{task("t", "other", 100)}); victims != nil { + t.Fatalf("an orphaned preemptor must select no victims; got %v", victims) + } +} + +func TestEnqueueableQueueState(t *testing.T) { + open := &api.QueueInfo{Queue: &scheduling.Queue{Status: scheduling.QueueStatus{State: scheduling.QueueStateOpen}}} + closed := &api.QueueInfo{Queue: &scheduling.Queue{Status: scheduling.QueueStatus{State: scheduling.QueueStateClosed}}} + closing := &api.QueueInfo{Queue: &scheduling.Queue{Status: scheduling.QueueStatus{State: scheduling.QueueStateClosing}}} + + if !enqueueableQueueState(open) { + t.Fatal("an Open queue must admit enqueue") + } + for name, queue := range map[string]*api.QueueInfo{"closed": closed, "closing": closing, "missing": nil} { + if enqueueableQueueState(queue) { + t.Fatalf("a %s queue must reject enqueue (jobs stay podless)", name) + } + } +} diff --git a/pkg/util/util.go b/pkg/util/util.go index d1f4030319b..eab8c4fbadb 100644 --- a/pkg/util/util.go +++ b/pkg/util/util.go @@ -43,6 +43,29 @@ const ( NoneShardingMode = "none" ) +// PodStartTimeoutLabel is the pod-start watchdog's deadline — its ONLY deadline +// source: a Go duration string (e.g. "45m"). Set on the VCJob (the platform's +// admission webhook defaults it onto jobs that don't set one) and copied by the +// job controller onto every pod the job creates; the watchdog reads it from the +// POD and marks a pod Failed once it has been bound to a node but unstarted for +// this long since binding. What that means for the job is its own lifecycle +// policy's business (PodFailed -> RestartJob bills a retry per lap; maxRetry +// exhaustion fails the job terminally). A pod without the label, or with an +// unparseable or non-positive value ("0" is an explicit opt-out), is not +// policed. +const PodStartTimeoutLabel = "volcano.sh/pod-start-timeout" + +// SubmittedAtPodGroupLabel carries a VCJob's true submission time (its immutable +// metadata.creationTimestamp, as epoch milliseconds) on every PodGroup the job +// controller (re)creates for it. The PodGroup's own creationTimestamp resets on +// RestartJob — the controller deletes and recreates the PodGroup — which would +// silently send a restarted job to the back of a FIFO queue; the fifopriority +// scheduler plugin orders same-PriorityClass jobs by this label instead, falling +// back to PodGroup creation time for podgroups without it. Stamped controller-side +// from an API-server-managed immutable field and never inherited from job labels, +// so a submitter cannot forge an earlier place in line. +const SubmittedAtPodGroupLabel = "volcano.sh/submitted-at" + var ( defaultElectionLeaseDuration = metav1.Duration{Duration: 15 * time.Second} defaultElectionRenewDeadline = metav1.Duration{Duration: 10 * time.Second}