Skip to content

Commit 540461a

Browse files
acmoreclaude
andauthored
feat(jobs): process-group kill for detached jobs; truncate list COMMAND column (#153)
* feat(jobs): process-group kill for detached jobs; truncate list COMMAND column Detached jobs now launch via setsid so the user command leads its own process group (pgid recorded in metadata). `jobs stop` signals the whole group — covering children like torchrun workers whose leader already exited and would previously linger holding GPU memory — and only reports success once no live group member remains. Group identity is verified via the inherited OKDEV_JOB_ID env marker before signaling, so recycled pids/pgids are never hit. Containers without setsid and jobs from older okdev versions fall back to the previous leader-only behavior. The scan line gains a groupLive count ("<alive>\t<groupLive>\t<json>", older shapes still parse) so stop's terminal criterion is "metadata terminal AND group drained"; jobs list --json exposes pgid/groupLive. jobs list also truncates the COMMAND column to the terminal width on TTY output (piped output and --json keep the full command) so long training commands no longer wrap table rows. Includes a Linux-only end-to-end test (CI) that launches a real process tree, kills the group through the actual signal script, and asserts every member is gone. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(jobs): drop -- before negative pgid in group kill for dash compatibility dash's kill builtin rejects `kill -SIG -- -pgid` with "Illegal number: -"; the bare "-<pgid>" operand after the signal option works in dash, bash, zsh, and busybox ash (verified against real process groups in all three local shells; CI's /bin/sh is dash). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * test(e2e): cover detach runtime-volume logs, group stop, sidecar fallback in kind smoke Four scenarios on a real pod: detach logs land under /var/okdev/exec; jobs list --output json reports pgid and live group members; jobs stop kills the whole process tree (verified by scanning /proc for survivors); and after crashing the dev container into CrashLoopBackOff, jobs logs still returns the job's output through the okdev-sidecar fallback. The container-crash step runs last before teardown since it leaves the dev container crashlooping. The long sleeps are nominal workloads — the group kill ends them in seconds, so wall-clock cost is the crash-loop induction (~30-40s). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
1 parent 6b44386 commit 540461a

10 files changed

Lines changed: 637 additions & 45 deletions

docs/command-reference.md

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ agents can react without launching a diagnostic chain on every blip:
108108
- `--container`: override which container to exec into (default: session target container). Works in both modes.
109109
- `--script` executes the uploaded file directly when it has a shebang; otherwise it falls back to `sh <file> ...`.
110110
- `--shell` cannot be combined with `--script`.
111-
- `--detach`: launches the command in the background and returns immediately. Prints per-pod launch details including job id, pid, log path, and metadata path.
111+
- `--detach`: launches the command in the background and returns immediately. Prints per-pod launch details including job id, pid, log path, and metadata path. The command runs in its own process group (via `setsid`) so `okdev jobs stop` can terminate the entire process tree, not just the leader.
112112
- When `--log-dir` is used with multiple declared groups, logs are written into per-group subdirectories under the requested path. `--log-dir` has no effect with `--detach` (detached jobs log inside the pod).
113113
- Detached exec preserves argv to the remote process; it is not flattened back into a single shell string. Detached `--script` uploads a per-pod temp file, launches it, and removes that temp file after completion.
114114
- Detached jobs store combined stdout/stderr and JSON metadata under `/var/okdev/exec/` on the shared okdev-runtime volume, so logs and metadata survive a target-container crash (e.g. OOMKill) and remain readable through the sidecar. If that directory cannot be created or written by the container user (e.g. a user-supplied root-owned volume with a non-root container), the launcher falls back to the legacy `/tmp/okdev-exec/` with a notice on stderr — same lifetime semantics as before. For large outputs (training logs, profiles, dumps), prefer `--log-dir` on the caller or redirect inside your command to a session volume; the runtime volume is a node-local emptyDir and heavy writers can fill it.
@@ -133,7 +133,8 @@ agents can react without launching a diagnostic chain on every blip:
133133
- If the target container is gone (e.g. OOMKilled), listing and `jobs logs` automatically retry through the `okdev-sidecar` container, which mounts the same runtime volume.
134134
- When a command fails because the container is gone, the error carries the termination cause when it can be determined, e.g. `container "pytorch" terminated: OOMKilled (exit 137, finished 2026-07-05T06:32:11Z)`. The same hint is appended to `okdev exec` FAILED lines and single-pod exec errors.
135135
- Text output is grouped by logical job id, with one row per detached launch showing job id, summarized state, pod count, earliest start time, and original command.
136-
- JSON output includes both the logical job summary and the per-pod `podStates` records.
136+
- On a terminal the COMMAND column is truncated (with ``) to the remaining terminal width so long training commands don't wrap rows; piped/redirected output and `--json` always carry the full command.
137+
- JSON output includes both the logical job summary and the per-pod `podStates` records (with `pgid` and `groupLive` — the count of live processes still in the job's process group — when available).
137138
- State values: `running` (wrapper alive and user command in flight), `exited` (user command finished; exit code is recorded in `podStates` / JSON), and `orphaned` (metadata still says `running` but the pid has exited or been recycled - typically the wrapper was `SIGKILL`ed or the container was restarted before the completion metadata could be written). Grouped text summaries render forms such as `running(1/2)`, `exited(2/2)`, and `failed(1/2)`.
138139
- When any pod fails to list its jobs, the command still prints the jobs it was able to collect from the healthy pods and reports the failures in a `FAILED:` footer (or an `errors` array in `--json` output); exit status is non-zero so scripts can detect partial failures.
139140
- With `spec.exec.fanoutMode: auto|gateway` and interpod SSH enabled, the per-pod listing routes through the in-cluster fanout gateway (one apiserver exec instead of one per pod). The listing is read-only, so any gateway problem silently falls back to direct per-pod queries.
@@ -167,8 +168,10 @@ agents can react without launching a diagnostic chain on every blip:
167168
### `okdev jobs stop <job-id> [session]`
168169

169170
- Stops every still-running pod in the logical job by sending `SIGTERM`, waiting 10 seconds, then sending `SIGKILL` to survivors.
170-
- Prints which pods were signaled during the stop flow.
171-
- Returns non-zero when any pod could not be queried or signaled, or if pods are still running after the stop attempt.
171+
- Signals the job's whole **process group** when available: detached jobs are launched via `setsid`, so children the command forked (e.g. `torchrun` workers) are signaled too — including stragglers whose leader already exited. Stop only reports success once no live group member remains, so a clean return means the process tree (and its GPU memory) is gone.
172+
- Group membership is verified via the job's `OKDEV_JOB_ID` environment marker before signaling, so recycled pids/pgids are never signaled by mistake. Jobs launched by older okdev versions (or in containers without `setsid`) fall back to leader-only signaling.
173+
- Prints which pods were signaled (`pgid=N` for group signals, `pid=N` for the fallback) during the stop flow.
174+
- Returns non-zero when any pod could not be queried or signaled, or if processes are still alive after the stop attempt.
172175

173176
### `okdev exec-jobs [session]`
174177

internal/cli/exec_jobs.go

Lines changed: 92 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import (
88
"fmt"
99
"io"
1010
"sort"
11+
"strconv"
1112
"strings"
1213
"time"
1314

@@ -22,6 +23,8 @@ type execJobView struct {
2223
Container string `json:"container"`
2324
JobID string `json:"jobId"`
2425
PID int `json:"pid"`
26+
PGID int `json:"pgid,omitempty"`
27+
GroupLive int `json:"groupLive,omitempty"`
2528
State string `json:"state"`
2629
ExitCode *int `json:"exitCode,omitempty"`
2730
LogPath string `json:"logPath"`
@@ -74,14 +77,15 @@ func runExecJobs(ctx context.Context, client connect.ExecClient, namespace strin
7477
fmt.Fprintln(out, "No detached exec jobs found")
7578
}
7679
if len(jobs) > 0 {
80+
commandLimit := jobsListCommandLimit(out, jobs)
7781
table := make([][]string, 0, len(jobs))
7882
for _, job := range jobs {
7983
table = append(table, []string{
8084
job.JobID,
8185
job.SummaryState,
8286
fmt.Sprintf("%d", job.Pods),
8387
job.StartedAt,
84-
job.Command,
88+
truncateTableCell(job.Command, commandLimit),
8589
})
8690
}
8791
output.PrintTable(out, []string{"JOB ID", "STATE", "PODS", "STARTED", "COMMAND"}, table)
@@ -223,6 +227,8 @@ func appendDetachJobRows(rows []execJobView, jobs []detachMetadata, jobID, conta
223227
Container: containerLabel,
224228
JobID: job.JobID,
225229
PID: job.PID,
230+
PGID: job.PGID,
231+
GroupLive: job.GroupLive,
226232
State: job.State,
227233
ExitCode: job.ExitCode,
228234
LogPath: job.StdoutPath,
@@ -322,30 +328,97 @@ func earliestStartedAt(rows []execJobView) string {
322328
return earliest
323329
}
324330

331+
// jobsListCommandLimit bounds the COMMAND column so long training commands
332+
// don't wrap table rows on a terminal. Returns 0 (no limit) for non-TTY
333+
// output — piped consumers get the full command; --json is the structured
334+
// path either way.
335+
func jobsListCommandLimit(out io.Writer, jobs []logicalExecJobView) int {
336+
if !isTerminalFD(out) {
337+
return 0
338+
}
339+
return jobsListCommandLimitForWidth(terminalWidth(), jobs)
340+
}
341+
342+
func jobsListCommandLimitForWidth(termWidth int, jobs []logicalExecJobView) int {
343+
if termWidth <= 0 {
344+
return 0
345+
}
346+
widths := []int{len("JOB ID"), len("STATE"), len("PODS"), len("STARTED")}
347+
for _, job := range jobs {
348+
for i, s := range []string{job.JobID, job.SummaryState, fmt.Sprintf("%d", job.Pods), job.StartedAt} {
349+
if len(s) > widths[i] {
350+
widths[i] = len(s)
351+
}
352+
}
353+
}
354+
fixed := 0
355+
for _, w := range widths {
356+
fixed += w
357+
}
358+
// PrintTable joins columns with two spaces; four separators precede the
359+
// COMMAND column.
360+
limit := termWidth - fixed - 4*2
361+
// Keep enough of the command to recognize the job even on narrow
362+
// terminals; a slightly wrapped row beats an unreadable stub.
363+
if limit < 16 {
364+
limit = 16
365+
}
366+
return limit
367+
}
368+
369+
func truncateTableCell(s string, limit int) string {
370+
if limit <= 0 {
371+
return s
372+
}
373+
r := []rune(s)
374+
if len(r) <= limit {
375+
return s
376+
}
377+
return string(r[:limit-1]) + "…"
378+
}
379+
325380
func detachJobsCommand() []string {
326-
// Each output line is "<alive>\t<json>" where <alive> is 1 if the job's
327-
// pid is still running and its /proc/<pid>/environ contains the
381+
// Each output line is "<alive>\t<groupLive>\t<json>". <alive> is 1 if the
382+
// job's pid is still running and its /proc/<pid>/environ contains the
328383
// OKDEV_JOB_ID we set in the wrapper. Matching on an environment
329384
// variable (rather than cmdline) is robust against PID reuse even when
330-
// the job's cmdline is the arbitrary user command. The find(1) call
331-
// also reaps stale .tmp files left behind by killed launchers/wrappers.
332-
// Both the current metadata dir and the legacy /tmp location are scanned
333-
// so jobs launched by an older okdev remain visible after an upgrade.
385+
// the job's cmdline is the arbitrary user command. <groupLive> counts
386+
// live (non-zombie) processes still in the job's process group — the
387+
// leader may be gone while its children (e.g. torchrun workers) linger,
388+
// which is exactly what `jobs stop` must clean up. The group is
389+
// identity-checked via OKDEV_JOB_ID in a member's environ (inherited by
390+
// all descendants) before counting: a PGID number can only be recycled
391+
// once the whole group is dead, so a mismatch means an unrelated group.
392+
// The find(1) call also reaps stale .tmp files left behind by killed
393+
// launchers/wrappers. Both the current metadata dir and the legacy /tmp
394+
// location are scanned so jobs launched by an older okdev remain
395+
// visible after an upgrade.
334396
script := `for dir in '` + detachMetadataDir + `' '` + legacyDetachMetadataDir + `'; do
335397
[ -d "$dir" ] || continue
336398
find "$dir" -maxdepth 1 -name '*.tmp' -type f -mmin +1 -delete 2>/dev/null || true
337399
for f in "$dir"/*.json; do
338400
[ -e "$f" ] || continue
339401
contents=$(cat "$f")
340402
pid=$(printf '%s' "$contents" | sed -n 's/.*"pid":\([0-9][0-9]*\).*/\1/p' | head -n1)
403+
pgid=$(printf '%s' "$contents" | sed -n 's/.*"pgid":\([0-9][0-9]*\).*/\1/p' | head -n1)
341404
job_id=$(basename "$f" .json)
342405
alive=0
343406
if [ -n "$pid" ] && [ -r "/proc/$pid/environ" ]; then
344407
if tr '\0' '\n' < "/proc/$pid/environ" 2>/dev/null | grep -Fqx "OKDEV_JOB_ID=$job_id"; then
345408
alive=1
346409
fi
347410
fi
348-
printf '%s\t%s\n' "$alive" "$contents"
411+
glive=0
412+
if [ -n "$pgid" ] && [ "$pgid" -gt 0 ] 2>/dev/null; then
413+
members=$(awk -v g="$pgid" '{ pid=$1; line=$0; sub(/^[0-9]+ \(.*\) /, "", line); split(line, f, " "); if (f[1] != "Z" && f[3] == g) print pid }' /proc/[0-9]*/stat 2>/dev/null)
414+
for m in $members; do
415+
if tr '\0' '\n' < "/proc/$m/environ" 2>/dev/null | grep -Fqx "OKDEV_JOB_ID=$job_id"; then
416+
glive=$(set -- $members; echo $#)
417+
break
418+
fi
419+
done
420+
fi
421+
printf '%s\t%s\t%s\n' "$alive" "$glive" "$contents"
349422
done
350423
done
351424
`
@@ -407,12 +480,20 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) {
407480
if strings.TrimSpace(line) == "" {
408481
continue
409482
}
410-
// Lines from detachJobsCommand are "<alive>\t<json>"; tolerate
411-
// raw JSON for older containers that ran a previous cli version.
483+
// Lines from detachJobsCommand are "<alive>\t<groupLive>\t<json>";
484+
// tolerate the older "<alive>\t<json>" shape and raw JSON so mixed
485+
// cli versions against the same pod keep working.
412486
alive := true
487+
groupLive := 0
413488
if tab := strings.IndexByte(line, '\t'); tab >= 0 {
414489
alive = line[:tab] != "0"
415490
line = line[tab+1:]
491+
if tab2 := strings.IndexByte(line, '\t'); tab2 >= 0 {
492+
if n, err := strconv.Atoi(strings.TrimSpace(line[:tab2])); err == nil && n >= 0 {
493+
groupLive = n
494+
line = line[tab2+1:]
495+
}
496+
}
416497
}
417498
line = strings.TrimSpace(line)
418499
if line == "" {
@@ -422,6 +503,7 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) {
422503
if err := json.Unmarshal([]byte(line), &job); err != nil {
423504
return nil, fmt.Errorf("parse detach job metadata: %w", err)
424505
}
506+
job.GroupLive = groupLive
425507
if job.JobID == "" || job.Pod == "" || job.PID <= 0 || job.MetaPath == "" {
426508
return nil, errors.New("missing detach job metadata")
427509
}

internal/cli/exec_jobs_test.go

Lines changed: 59 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -221,8 +221,12 @@ func TestDetachJobsCommandReconcilesAndReapsTmpFiles(t *testing.T) {
221221
"alive=1",
222222
"alive=0",
223223
"grep -Fqx \"OKDEV_JOB_ID=$job_id\"",
224-
// Per-line output is "<alive>\t<json>".
225-
"printf '%s\\t%s\\n' \"$alive\" \"$contents\"",
224+
// Per-line output is "<alive>\t<groupLive>\t<json>".
225+
"printf '%s\\t%s\\t%s\\n' \"$alive\" \"$glive\" \"$contents\"",
226+
// Group liveness: count non-zombie members of the job's process
227+
// group, identity-checked via the inherited OKDEV_JOB_ID env var.
228+
`"pgid":\([0-9][0-9]*\)`,
229+
"/proc/[0-9]*/stat",
226230
// Best-effort cleanup of stale temp files left by killed wrappers.
227231
"-name '*.tmp'",
228232
"-mmin +1",
@@ -388,6 +392,59 @@ func TestIsContainerUnavailableExecError(t *testing.T) {
388392
}
389393
}
390394

395+
func TestParseDetachMetadataLinesParsesGroupLive(t *testing.T) {
396+
raw := "1\t3\t{\"jobId\":\"job-a\",\"pod\":\"p\",\"container\":\"dev\",\"pid\":100,\"pgid\":100,\"metaPath\":\"/var/okdev/exec/job-a.json\",\"state\":\"running\"}\n" +
397+
// Older two-field lines and raw JSON must keep parsing (mixed cli
398+
// versions against the same pod).
399+
"0\t{\"jobId\":\"job-b\",\"pod\":\"p\",\"container\":\"dev\",\"pid\":101,\"metaPath\":\"/tmp/okdev-exec/job-b.json\",\"state\":\"running\"}\n"
400+
jobs, err := parseDetachMetadataLines(raw)
401+
if err != nil {
402+
t.Fatalf("parse: %v", err)
403+
}
404+
if len(jobs) != 2 {
405+
t.Fatalf("expected 2 jobs, got %+v", jobs)
406+
}
407+
if jobs[0].GroupLive != 3 || jobs[0].PGID != 100 {
408+
t.Fatalf("expected groupLive=3 pgid=100, got %+v", jobs[0])
409+
}
410+
if jobs[1].GroupLive != 0 || jobs[1].State != "orphaned" {
411+
t.Fatalf("expected legacy line orphaned with groupLive=0, got %+v", jobs[1])
412+
}
413+
}
414+
415+
func TestJobsListCommandLimitForWidth(t *testing.T) {
416+
jobs := []logicalExecJobView{{
417+
JobID: "20260705T063211Z-deadbeef", // 25 chars
418+
SummaryState: "running(4/4)", // 12 chars
419+
Pods: 4,
420+
StartedAt: "2026-07-05T06:32:11Z", // 20 chars
421+
}}
422+
// fixed = 25 + 12 + len("PODS")=4 + 20 + 4*2 = 69
423+
if got := jobsListCommandLimitForWidth(120, jobs); got != 51 {
424+
t.Fatalf("expected limit 51, got %d", got)
425+
}
426+
// Narrow terminals keep a recognizable floor instead of a useless stub.
427+
if got := jobsListCommandLimitForWidth(60, jobs); got != 16 {
428+
t.Fatalf("expected floor 16, got %d", got)
429+
}
430+
// Unknown width disables truncation.
431+
if got := jobsListCommandLimitForWidth(0, jobs); got != 0 {
432+
t.Fatalf("expected no limit, got %d", got)
433+
}
434+
}
435+
436+
func TestTruncateTableCell(t *testing.T) {
437+
if got := truncateTableCell("short", 10); got != "short" {
438+
t.Fatalf("unexpected: %q", got)
439+
}
440+
if got := truncateTableCell("torchrun --nnodes 4 --nproc-per-node 8 pretrain.py", 20); got != "torchrun --nnodes 4…" {
441+
t.Fatalf("unexpected truncation: %q", got)
442+
}
443+
if got := truncateTableCell("anything", 0); got != "anything" {
444+
t.Fatalf("limit 0 must disable truncation: %q", got)
445+
}
446+
}
447+
391448
func TestCollectDetachJobsFiltersByJobID(t *testing.T) {
392449
client := &fakePdshExecClient{
393450
outputs: map[string]string{

0 commit comments

Comments
 (0)