From cd1d9eaabd60b5b0da95122b22d7dae32e56066e Mon Sep 17 00:00:00 2001 From: acmore Date: Sun, 5 Jul 2026 13:21:08 +0800 Subject: [PATCH 1/3] feat(jobs): process-group kill for detached jobs; truncate list COMMAND column MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 ("\t\t", 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 --- docs/command-reference.md | 11 ++- internal/cli/exec_jobs.go | 102 +++++++++++++++++++--- internal/cli/exec_jobs_test.go | 61 ++++++++++++- internal/cli/jobs.go | 79 ++++++++++++++--- internal/cli/jobs_logs_wait_test.go | 15 ++++ internal/cli/jobs_stop_test.go | 116 +++++++++++++++++++++++++ internal/cli/pdsh.go | 54 +++++++++--- internal/cli/pdsh_group_linux_test.go | 119 ++++++++++++++++++++++++++ internal/cli/pdsh_test.go | 15 +++- 9 files changed, 527 insertions(+), 45 deletions(-) create mode 100644 internal/cli/pdsh_group_linux_test.go diff --git a/docs/command-reference.md b/docs/command-reference.md index a8a1ae4f..ac6b4daa 100644 --- a/docs/command-reference.md +++ b/docs/command-reference.md @@ -108,7 +108,7 @@ agents can react without launching a diagnostic chain on every blip: - `--container`: override which container to exec into (default: session target container). Works in both modes. - `--script` executes the uploaded file directly when it has a shebang; otherwise it falls back to `sh ...`. - `--shell` cannot be combined with `--script`. -- `--detach`: launches the command in the background and returns immediately. Prints per-pod launch details including job id, pid, log path, and metadata path. +- `--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. - 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). - 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. - 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: - 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. - 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. - 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. -- JSON output includes both the logical job summary and the per-pod `podStates` records. +- 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. +- 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). - 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)`. - 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. - 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: ### `okdev jobs stop [session]` - Stops every still-running pod in the logical job by sending `SIGTERM`, waiting 10 seconds, then sending `SIGKILL` to survivors. -- Prints which pods were signaled during the stop flow. -- Returns non-zero when any pod could not be queried or signaled, or if pods are still running after the stop attempt. +- 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. +- 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. +- Prints which pods were signaled (`pgid=N` for group signals, `pid=N` for the fallback) during the stop flow. +- Returns non-zero when any pod could not be queried or signaled, or if processes are still alive after the stop attempt. ### `okdev exec-jobs [session]` diff --git a/internal/cli/exec_jobs.go b/internal/cli/exec_jobs.go index 3f9f6b3d..53bd21fc 100644 --- a/internal/cli/exec_jobs.go +++ b/internal/cli/exec_jobs.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "sort" + "strconv" "strings" "time" @@ -22,6 +23,8 @@ type execJobView struct { Container string `json:"container"` JobID string `json:"jobId"` PID int `json:"pid"` + PGID int `json:"pgid,omitempty"` + GroupLive int `json:"groupLive,omitempty"` State string `json:"state"` ExitCode *int `json:"exitCode,omitempty"` LogPath string `json:"logPath"` @@ -74,6 +77,7 @@ func runExecJobs(ctx context.Context, client connect.ExecClient, namespace strin fmt.Fprintln(out, "No detached exec jobs found") } if len(jobs) > 0 { + commandLimit := jobsListCommandLimit(out, jobs) table := make([][]string, 0, len(jobs)) for _, job := range jobs { table = append(table, []string{ @@ -81,7 +85,7 @@ func runExecJobs(ctx context.Context, client connect.ExecClient, namespace strin job.SummaryState, fmt.Sprintf("%d", job.Pods), job.StartedAt, - job.Command, + truncateTableCell(job.Command, commandLimit), }) } output.PrintTable(out, []string{"JOB ID", "STATE", "PODS", "STARTED", "COMMAND"}, table) @@ -223,6 +227,8 @@ func appendDetachJobRows(rows []execJobView, jobs []detachMetadata, jobID, conta Container: containerLabel, JobID: job.JobID, PID: job.PID, + PGID: job.PGID, + GroupLive: job.GroupLive, State: job.State, ExitCode: job.ExitCode, LogPath: job.StdoutPath, @@ -322,15 +328,71 @@ func earliestStartedAt(rows []execJobView) string { return earliest } +// jobsListCommandLimit bounds the COMMAND column so long training commands +// don't wrap table rows on a terminal. Returns 0 (no limit) for non-TTY +// output — piped consumers get the full command; --json is the structured +// path either way. +func jobsListCommandLimit(out io.Writer, jobs []logicalExecJobView) int { + if !isTerminalFD(out) { + return 0 + } + return jobsListCommandLimitForWidth(terminalWidth(), jobs) +} + +func jobsListCommandLimitForWidth(termWidth int, jobs []logicalExecJobView) int { + if termWidth <= 0 { + return 0 + } + widths := []int{len("JOB ID"), len("STATE"), len("PODS"), len("STARTED")} + for _, job := range jobs { + for i, s := range []string{job.JobID, job.SummaryState, fmt.Sprintf("%d", job.Pods), job.StartedAt} { + if len(s) > widths[i] { + widths[i] = len(s) + } + } + } + fixed := 0 + for _, w := range widths { + fixed += w + } + // PrintTable joins columns with two spaces; four separators precede the + // COMMAND column. + limit := termWidth - fixed - 4*2 + // Keep enough of the command to recognize the job even on narrow + // terminals; a slightly wrapped row beats an unreadable stub. + if limit < 16 { + limit = 16 + } + return limit +} + +func truncateTableCell(s string, limit int) string { + if limit <= 0 { + return s + } + r := []rune(s) + if len(r) <= limit { + return s + } + return string(r[:limit-1]) + "…" +} + func detachJobsCommand() []string { - // Each output line is "\t" where is 1 if the job's - // pid is still running and its /proc//environ contains the + // Each output line is "\t\t". is 1 if the + // job's pid is still running and its /proc//environ contains the // OKDEV_JOB_ID we set in the wrapper. Matching on an environment // variable (rather than cmdline) is robust against PID reuse even when - // the job's cmdline is the arbitrary user command. The find(1) call - // also reaps stale .tmp files left behind by killed launchers/wrappers. - // Both the current metadata dir and the legacy /tmp location are scanned - // so jobs launched by an older okdev remain visible after an upgrade. + // the job's cmdline is the arbitrary user command. counts + // live (non-zombie) processes still in the job's process group — the + // leader may be gone while its children (e.g. torchrun workers) linger, + // which is exactly what `jobs stop` must clean up. The group is + // identity-checked via OKDEV_JOB_ID in a member's environ (inherited by + // all descendants) before counting: a PGID number can only be recycled + // once the whole group is dead, so a mismatch means an unrelated group. + // The find(1) call also reaps stale .tmp files left behind by killed + // launchers/wrappers. Both the current metadata dir and the legacy /tmp + // location are scanned so jobs launched by an older okdev remain + // visible after an upgrade. script := `for dir in '` + detachMetadataDir + `' '` + legacyDetachMetadataDir + `'; do [ -d "$dir" ] || continue find "$dir" -maxdepth 1 -name '*.tmp' -type f -mmin +1 -delete 2>/dev/null || true @@ -338,6 +400,7 @@ func detachJobsCommand() []string { [ -e "$f" ] || continue contents=$(cat "$f") pid=$(printf '%s' "$contents" | sed -n 's/.*"pid":\([0-9][0-9]*\).*/\1/p' | head -n1) + pgid=$(printf '%s' "$contents" | sed -n 's/.*"pgid":\([0-9][0-9]*\).*/\1/p' | head -n1) job_id=$(basename "$f" .json) alive=0 if [ -n "$pid" ] && [ -r "/proc/$pid/environ" ]; then @@ -345,7 +408,17 @@ func detachJobsCommand() []string { alive=1 fi fi - printf '%s\t%s\n' "$alive" "$contents" + glive=0 + if [ -n "$pgid" ] && [ "$pgid" -gt 0 ] 2>/dev/null; then + 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) + for m in $members; do + if tr '\0' '\n' < "/proc/$m/environ" 2>/dev/null | grep -Fqx "OKDEV_JOB_ID=$job_id"; then + glive=$(set -- $members; echo $#) + break + fi + done + fi + printf '%s\t%s\t%s\n' "$alive" "$glive" "$contents" done done ` @@ -407,12 +480,20 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) { if strings.TrimSpace(line) == "" { continue } - // Lines from detachJobsCommand are "\t"; tolerate - // raw JSON for older containers that ran a previous cli version. + // Lines from detachJobsCommand are "\t\t"; + // tolerate the older "\t" shape and raw JSON so mixed + // cli versions against the same pod keep working. alive := true + groupLive := 0 if tab := strings.IndexByte(line, '\t'); tab >= 0 { alive = line[:tab] != "0" line = line[tab+1:] + if tab2 := strings.IndexByte(line, '\t'); tab2 >= 0 { + if n, err := strconv.Atoi(strings.TrimSpace(line[:tab2])); err == nil && n >= 0 { + groupLive = n + line = line[tab2+1:] + } + } } line = strings.TrimSpace(line) if line == "" { @@ -422,6 +503,7 @@ func parseDetachMetadataLines(raw string) ([]detachMetadata, error) { if err := json.Unmarshal([]byte(line), &job); err != nil { return nil, fmt.Errorf("parse detach job metadata: %w", err) } + job.GroupLive = groupLive if job.JobID == "" || job.Pod == "" || job.PID <= 0 || job.MetaPath == "" { return nil, errors.New("missing detach job metadata") } diff --git a/internal/cli/exec_jobs_test.go b/internal/cli/exec_jobs_test.go index fcb33fa9..37869e9a 100644 --- a/internal/cli/exec_jobs_test.go +++ b/internal/cli/exec_jobs_test.go @@ -221,8 +221,12 @@ func TestDetachJobsCommandReconcilesAndReapsTmpFiles(t *testing.T) { "alive=1", "alive=0", "grep -Fqx \"OKDEV_JOB_ID=$job_id\"", - // Per-line output is "\t". - "printf '%s\\t%s\\n' \"$alive\" \"$contents\"", + // Per-line output is "\t\t". + "printf '%s\\t%s\\t%s\\n' \"$alive\" \"$glive\" \"$contents\"", + // Group liveness: count non-zombie members of the job's process + // group, identity-checked via the inherited OKDEV_JOB_ID env var. + `"pgid":\([0-9][0-9]*\)`, + "/proc/[0-9]*/stat", // Best-effort cleanup of stale temp files left by killed wrappers. "-name '*.tmp'", "-mmin +1", @@ -388,6 +392,59 @@ func TestIsContainerUnavailableExecError(t *testing.T) { } } +func TestParseDetachMetadataLinesParsesGroupLive(t *testing.T) { + 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" + + // Older two-field lines and raw JSON must keep parsing (mixed cli + // versions against the same pod). + "0\t{\"jobId\":\"job-b\",\"pod\":\"p\",\"container\":\"dev\",\"pid\":101,\"metaPath\":\"/tmp/okdev-exec/job-b.json\",\"state\":\"running\"}\n" + jobs, err := parseDetachMetadataLines(raw) + if err != nil { + t.Fatalf("parse: %v", err) + } + if len(jobs) != 2 { + t.Fatalf("expected 2 jobs, got %+v", jobs) + } + if jobs[0].GroupLive != 3 || jobs[0].PGID != 100 { + t.Fatalf("expected groupLive=3 pgid=100, got %+v", jobs[0]) + } + if jobs[1].GroupLive != 0 || jobs[1].State != "orphaned" { + t.Fatalf("expected legacy line orphaned with groupLive=0, got %+v", jobs[1]) + } +} + +func TestJobsListCommandLimitForWidth(t *testing.T) { + jobs := []logicalExecJobView{{ + JobID: "20260705T063211Z-deadbeef", // 25 chars + SummaryState: "running(4/4)", // 12 chars + Pods: 4, + StartedAt: "2026-07-05T06:32:11Z", // 20 chars + }} + // fixed = 25 + 12 + len("PODS")=4 + 20 + 4*2 = 69 + if got := jobsListCommandLimitForWidth(120, jobs); got != 51 { + t.Fatalf("expected limit 51, got %d", got) + } + // Narrow terminals keep a recognizable floor instead of a useless stub. + if got := jobsListCommandLimitForWidth(60, jobs); got != 16 { + t.Fatalf("expected floor 16, got %d", got) + } + // Unknown width disables truncation. + if got := jobsListCommandLimitForWidth(0, jobs); got != 0 { + t.Fatalf("expected no limit, got %d", got) + } +} + +func TestTruncateTableCell(t *testing.T) { + if got := truncateTableCell("short", 10); got != "short" { + t.Fatalf("unexpected: %q", got) + } + if got := truncateTableCell("torchrun --nnodes 4 --nproc-per-node 8 pretrain.py", 20); got != "torchrun --nnodes 4…" { + t.Fatalf("unexpected truncation: %q", got) + } + if got := truncateTableCell("anything", 0); got != "anything" { + t.Fatalf("limit 0 must disable truncation: %q", got) + } +} + func TestCollectDetachJobsFiltersByJobID(t *testing.T) { client := &fakePdshExecClient{ outputs: map[string]string{ diff --git a/internal/cli/jobs.go b/internal/cli/jobs.go index dc0d4df4..3f58b02f 100644 --- a/internal/cli/jobs.go +++ b/internal/cli/jobs.go @@ -388,7 +388,7 @@ func runJobsStop(ctx context.Context, client detachJobClient, namespace string, signalErrors := make([]execJobsPodError, 0) for _, row := range job.PodStates { - if row.State != "running" { + if !stopRowNeedsSignal(row) { continue } result, signalErr := signalDetachJob(ctx, client, namespace, row, "TERM") @@ -397,7 +397,7 @@ func runJobsStop(ctx context.Context, client detachJobClient, namespace string, fmt.Fprintf(out, "%s error: %v\n", prefixByPod[row.Pod], signalErr) continue } - fmt.Fprintf(out, "%s sent SIGTERM pid=%d (%s)\n", prefixByPod[row.Pod], row.PID, result) + fmt.Fprintf(out, "%s sent SIGTERM %s (%s)\n", prefixByPod[row.Pod], stopSignalTargetLabel(row), result) } for _, e := range signalErrors { aggregated[e.Pod] = e.Error @@ -412,7 +412,7 @@ func runJobsStop(ctx context.Context, client detachJobClient, namespace string, jobGone := false waitLoop: for { - if allLogicalJobTerminal(current) { + if stopJobSettled(current) { break } select { @@ -447,9 +447,9 @@ waitLoop: } killErrors := make([]execJobsPodError, 0) - if !jobGone && !allLogicalJobTerminal(current) { + if !jobGone && !stopJobSettled(current) { for _, row := range current.PodStates { - if row.State != "running" { + if !stopRowNeedsSignal(row) { continue } result, signalErr := signalDetachJob(ctx, client, namespace, row, "KILL") @@ -458,7 +458,7 @@ waitLoop: fmt.Fprintf(out, "%s error: %v\n", prefixByPod[row.Pod], signalErr) continue } - fmt.Fprintf(out, "%s sent SIGKILL pid=%d (%s)\n", prefixByPod[row.Pod], row.PID, result) + fmt.Fprintf(out, "%s sent SIGKILL %s (%s)\n", prefixByPod[row.Pod], stopSignalTargetLabel(row), result) } } for _, e := range killErrors { @@ -484,7 +484,7 @@ waitLoop: for _, e := range finalPodErrors { aggregated[e.Pod] = e.Error } - terminalOk = allLogicalJobTerminal(final) + terminalOk = stopJobSettled(final) current = final } } @@ -538,6 +538,36 @@ func jobPodNames(job logicalExecJobView) []string { return names } +// stopRowNeedsSignal reports whether `jobs stop` should signal this pod's +// record: the job still runs, or the leader exited but verified group +// members linger (torchrun children holding GPU memory). +func stopRowNeedsSignal(row execJobView) bool { + return row.State == "running" || (row.PGID > 0 && row.GroupLive > 0) +} + +// stopJobSettled is `jobs stop`'s terminal criterion: every pod record is +// terminal AND no live group members remain. Stricter than +// allLogicalJobTerminal (used by logs/wait), which only tracks the leader's +// metadata state. +func stopJobSettled(job logicalExecJobView) bool { + if len(job.PodStates) == 0 { + return false + } + for _, row := range job.PodStates { + if stopRowNeedsSignal(row) { + return false + } + } + return true +} + +func stopSignalTargetLabel(row execJobView) string { + if row.PGID > 0 { + return fmt.Sprintf("pgid=%d", row.PGID) + } + return fmt.Sprintf("pid=%d", row.PID) +} + func allLogicalJobTerminal(job logicalExecJobView) bool { if len(job.PodStates) == 0 { return false @@ -605,15 +635,42 @@ func followDetachJobLogScript(logPath string) string { } func signalDetachJob(ctx context.Context, client detachJobClient, namespace string, row execJobView, signalName string) (string, error) { - out, err := client.ExecShInContainer(ctx, namespace, row.Pod, row.Container, signalDetachJobScript(row.JobID, row.PID, signalName)) + out, err := client.ExecShInContainer(ctx, namespace, row.Pod, row.Container, signalDetachJobScript(row.JobID, row.PID, row.PGID, signalName)) if err != nil { return "", err } return strings.TrimSpace(string(out)), nil } -func signalDetachJobScript(jobID string, pid int, signalName string) string { - return fmt.Sprintf( +// signalDetachJobScript signals a detached job. With a process group (pgid > +// 0) it signals the whole group — covering children like torchrun workers +// whose leader may already be dead — after verifying via OKDEV_JOB_ID in a +// member's environ that the group still belongs to this job (a PGID number +// cannot be recycled while any member lives). Without a group, or when the +// group is already empty, it falls through to the legacy leader-only path. +func signalDetachJobScript(jobID string, pid, pgid int, signalName string) string { + quotedJobEnv := shellQuote("OKDEV_JOB_ID=" + jobID) + groupPrelude := "" + if pgid > 0 { + groupPrelude = fmt.Sprintf( + "pgid=%d\n"+ + "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)\n"+ + "for m in $members; do\n"+ + " if tr '\\000' '\\n' < \"/proc/$m/environ\" 2>/dev/null | grep -Fqx %s; then\n"+ + " if kill -%s -- \"-$pgid\" 2>/dev/null; then\n"+ + " echo \"signaled group ($(set -- $members; echo $#) members)\"\n"+ + " exit 0\n"+ + " fi\n"+ + " echo not-running\n"+ + " exit 0\n"+ + " fi\n"+ + "done\n", + pgid, + quotedJobEnv, + signalName, + ) + } + return groupPrelude + fmt.Sprintf( "pid=%d\n"+ "env_path=\"/proc/$pid/environ\"\n"+ "if [ ! -r \"$env_path\" ]; then\n"+ @@ -630,7 +687,7 @@ func signalDetachJobScript(jobID string, pid int, signalName string) string { "fi\n"+ "echo not-running\n", pid, - shellQuote("OKDEV_JOB_ID="+jobID), + quotedJobEnv, signalName, ) } diff --git a/internal/cli/jobs_logs_wait_test.go b/internal/cli/jobs_logs_wait_test.go index bf5f1842..923fd9c4 100644 --- a/internal/cli/jobs_logs_wait_test.go +++ b/internal/cli/jobs_logs_wait_test.go @@ -263,3 +263,18 @@ func detachMetadataJSON(jobID, pod, container string, pid int, state string, exi return fmt.Sprintf("{\"jobId\":\"%s\",\"pod\":\"%s\",\"container\":\"%s\",\"pid\":%d,\"command\":\"python train.py\",\"startedAt\":\"2026-05-09T10:00:00Z\",\"stdoutPath\":\"/tmp/okdev-exec/%s.log\",\"stderrPath\":\"/tmp/okdev-exec/%s.log\",\"metaPath\":\"/tmp/okdev-exec/%s.json\",\"state\":\"%s\"%s}\n", jobID, pod, container, pid, jobID, jobID, jobID, state, exitPart) } + +// detachMetadataLineGroup renders the full "\t\t" +// scan line for a job launched with a process group. +func detachMetadataLineGroup(jobID, pod, container string, pid, pgid int, state string, exitCode *int, alive bool, groupLive int) string { + exitPart := "" + if exitCode != nil { + exitPart = fmt.Sprintf(",\"exitCode\":%d", *exitCode) + } + aliveFlag := "0" + if alive { + aliveFlag = "1" + } + return fmt.Sprintf("%s\t%d\t{\"jobId\":\"%s\",\"pod\":\"%s\",\"container\":\"%s\",\"pid\":%d,\"pgid\":%d,\"command\":\"python train.py\",\"startedAt\":\"2026-05-09T10:00:00Z\",\"stdoutPath\":\"/var/okdev/exec/%s.log\",\"stderrPath\":\"/var/okdev/exec/%s.log\",\"metaPath\":\"/var/okdev/exec/%s.json\",\"state\":\"%s\"%s}\n", + aliveFlag, groupLive, jobID, pod, container, pid, pgid, jobID, jobID, jobID, state, exitPart) +} diff --git a/internal/cli/jobs_stop_test.go b/internal/cli/jobs_stop_test.go index 16712fa3..d01e0aa2 100644 --- a/internal/cli/jobs_stop_test.go +++ b/internal/cli/jobs_stop_test.go @@ -56,6 +56,122 @@ func TestRunJobsStopEscalatesToSIGKILLAfterGrace(t *testing.T) { } } +func TestRunJobsStopSignalsLingeringGroupAfterLeaderExit(t *testing.T) { + // The leader (torchrun) already exited but two group members + // (pretrain.py workers) linger holding GPU memory. Stop must signal the + // group even though the metadata state is terminal, and only report + // success once the group is empty. + oldPollEvery := detachJobPollEvery + oldStopGrace := detachJobStopGrace + detachJobPollEvery = 5 * time.Millisecond + detachJobStopGrace = 15 * time.Millisecond + defer func() { + detachJobPollEvery = oldPollEvery + detachJobStopGrace = oldStopGrace + }() + + exit1 := 1 + client := &fakeJobsClient{ + listOutputs: map[string][]string{ + "okdev-sess-worker-0": { + detachMetadataLineGroup("job-shared", "okdev-sess-worker-0", "pytorch", 100, 100, "exited", &exit1, false, 2), + detachMetadataLineGroup("job-shared", "okdev-sess-worker-0", "pytorch", 100, 100, "exited", &exit1, false, 0), + }, + }, + execShResponses: map[string]fakeJobsExecResponse{ + "okdev-sess-worker-0|TERM": {stdout: "signaled group (2 members)\n"}, + }, + } + pods := []kube.PodSummary{{Name: "okdev-sess-worker-0", Phase: "Running"}} + var out bytes.Buffer + if err := runJobsStop(context.Background(), client, "default", pods, "pytorch", "job-shared", pdshDefaultFanout, &out); err != nil { + t.Fatalf("runJobsStop: %v", err) + } + got := out.String() + if !strings.Contains(got, "sent SIGTERM pgid=100 (signaled group (2 members))") { + t.Fatalf("expected group TERM line, got %q", got) + } + if len(client.execScripts) != 1 || !strings.Contains(client.execScripts[0], `kill -TERM -- "-$pgid"`) { + t.Fatalf("expected a single group TERM script, got %+v", client.execScripts) + } +} + +func TestRunJobsStopFailsWhileGroupMembersRemain(t *testing.T) { + // If the group never drains (e.g. processes stuck in D-state), stop must + // escalate to SIGKILL and still exit non-zero rather than reporting a + // clean stop. + oldPollEvery := detachJobPollEvery + oldStopGrace := detachJobStopGrace + detachJobPollEvery = 5 * time.Millisecond + detachJobStopGrace = 15 * time.Millisecond + defer func() { + detachJobPollEvery = oldPollEvery + detachJobStopGrace = oldStopGrace + }() + + exit1 := 1 + client := &fakeJobsClient{ + listOutputs: map[string][]string{ + "okdev-sess-worker-0": { + detachMetadataLineGroup("job-shared", "okdev-sess-worker-0", "pytorch", 100, 100, "exited", &exit1, false, 2), + }, + }, + execShResponses: map[string]fakeJobsExecResponse{ + "okdev-sess-worker-0|TERM": {stdout: "signaled group (2 members)\n"}, + "okdev-sess-worker-0|KILL": {stdout: "signaled group (2 members)\n"}, + }, + } + pods := []kube.PodSummary{{Name: "okdev-sess-worker-0", Phase: "Running"}} + var out bytes.Buffer + err := runJobsStop(context.Background(), client, "default", pods, "pytorch", "job-shared", pdshDefaultFanout, &out) + if err == nil || !strings.Contains(err.Error(), "still has running pods") { + t.Fatalf("expected still-running error, got %v", err) + } + got := out.String() + if !strings.Contains(got, "sent SIGKILL pgid=100") { + t.Fatalf("expected group KILL escalation, got %q", got) + } +} + +func TestStopRowNeedsSignalAndSettled(t *testing.T) { + running := execJobView{State: "running"} + exitedClean := execJobView{State: "exited"} + exitedWithGroup := execJobView{State: "exited", PGID: 100, GroupLive: 3} + exitedGroupDrained := execJobView{State: "exited", PGID: 100, GroupLive: 0} + if !stopRowNeedsSignal(running) || !stopRowNeedsSignal(exitedWithGroup) { + t.Fatal("running rows and lingering groups must be signaled") + } + if stopRowNeedsSignal(exitedClean) || stopRowNeedsSignal(exitedGroupDrained) { + t.Fatal("terminal rows without live group members must not be signaled") + } + if stopJobSettled(logicalExecJobView{PodStates: []execJobView{exitedClean, exitedWithGroup}}) { + t.Fatal("job with lingering group members is not settled") + } + if !stopJobSettled(logicalExecJobView{PodStates: []execJobView{exitedClean, exitedGroupDrained}}) { + t.Fatal("job with drained groups is settled") + } +} + +func TestSignalDetachJobScriptGroupAndFallback(t *testing.T) { + group := signalDetachJobScript("job-1", 100, 100, "TERM") + for _, want := range []string{ + `kill -TERM -- "-$pgid"`, + "/proc/[0-9]*/stat", + "OKDEV_JOB_ID=job-1", + // Legacy leader path stays as the fallback tail when the group is + // already empty or belongs to someone else. + "kill -TERM \"$pid\"", + } { + if !strings.Contains(group, want) { + t.Fatalf("expected group script to contain %q, got %q", want, group) + } + } + legacy := signalDetachJobScript("job-1", 100, 0, "TERM") + if strings.Contains(legacy, "pgid") { + t.Fatalf("legacy script must not reference pgid, got %q", legacy) + } +} + func TestRunJobsStopReturnsPartialFailure(t *testing.T) { oldPollEvery := detachJobPollEvery oldStopGrace := detachJobStopGrace diff --git a/internal/cli/pdsh.go b/internal/cli/pdsh.go index 1a436cc5..2b892f6f 100644 --- a/internal/cli/pdsh.go +++ b/internal/cli/pdsh.go @@ -38,6 +38,7 @@ const detachMetadataDir = "/var/okdev/exec" const legacyDetachMetadataDir = "/tmp/okdev-exec" const detachPIDPlaceholder = "__OKDEV_DETACH_PID__" +const detachPGIDPlaceholder = "__OKDEV_DETACH_PGID__" const detachExitPlaceholder = "__OKDEV_DETACH_EXIT__" // detachDirPlaceholder stands in for the metadata directory in job paths and @@ -1118,10 +1119,14 @@ type detachJobSpec struct { } type detachMetadata struct { - JobID string `json:"jobId"` - Pod string `json:"pod"` - Container string `json:"container"` - PID int `json:"pid"` + JobID string `json:"jobId"` + Pod string `json:"pod"` + Container string `json:"container"` + PID int `json:"pid"` + // PGID is the job's process group id (== PID when the wrapper launched + // the command via setsid); 0 for jobs from containers without setsid or + // launched by older okdev versions — group-wide signaling is skipped. + PGID int `json:"pgid"` Command string `json:"command"` StartedAt string `json:"startedAt"` StdoutPath string `json:"stdoutPath"` @@ -1129,6 +1134,11 @@ type detachMetadata struct { MetaPath string `json:"metaPath"` State string `json:"state"` ExitCode *int `json:"exitCode,omitempty"` + // GroupLive is the count of live (non-zombie) processes still in the + // job's process group at scan time, verified to belong to this job via + // OKDEV_JOB_ID in a member's environment. Populated from the scan line + // prefix, not from the metadata file itself. + GroupLive int `json:"-"` } type detachLaunchInfo struct { @@ -1279,14 +1289,29 @@ func detachWrapperScript(jobID string, argv []string, cleanupPaths []string, met "export OKDEV_JOB_ID=%s\n"+ "set -- %s\n"+ "cleanup() {\n%s}\n"+ - "(exec \"$@\") &\n"+ - "user_pid=$!\n"+ + // setsid puts the user command in its own process group (PGID == + // its PID) so `okdev jobs stop` can signal the whole tree — + // torchrun's children included. The backgrounded child is not a + // group leader here, so setsid(2) succeeds without forking and + // $! is the user command's own pid after exec. Without setsid we + // keep the old single-process behavior and record pgid=0 so stop + // falls back to leader-only signaling. + "if command -v setsid >/dev/null 2>&1; then\n"+ + " setsid \"$@\" &\n"+ + " user_pid=$!\n"+ + " user_pgid=$user_pid\n"+ + "else\n"+ + " echo 'okdev: setsid unavailable; job has no process group (stop signals the leader only)' >&2\n"+ + " (exec \"$@\") &\n"+ + " user_pid=$!\n"+ + " user_pgid=0\n"+ + "fi\n"+ "if [ -z \"$user_pid\" ] || [ \"$user_pid\" -le 0 ]; then\n"+ " exit 1\n"+ "fi\n"+ - "printf '%%s\\n' \"$running_json\" | sed \"s/%s/$user_pid/g\" >\"$meta_tmp\" || exit 1\n"+ + "printf '%%s\\n' \"$running_json\" | sed \"s/%s/$user_pid/g; s/%s/$user_pgid/g\" >\"$meta_tmp\" || exit 1\n"+ "mv \"$meta_tmp\" \"$meta_path\" || exit 1\n"+ - "trap 'rc=$?; printf \"%%s\\n\" \"$exit_json\" | sed \"s/%s/$user_pid/g; s/%s/$rc/g\" >\"$meta_tmp\" 2>/dev/null && mv \"$meta_tmp\" \"$meta_path\" 2>/dev/null; cleanup; exit $rc' EXIT\n"+ + "trap 'rc=$?; printf \"%%s\\n\" \"$exit_json\" | sed \"s/%s/$user_pid/g; s/%s/$user_pgid/g; s/%s/$rc/g\" >\"$meta_tmp\" 2>/dev/null && mv \"$meta_tmp\" \"$meta_path\" 2>/dev/null; cleanup; exit $rc' EXIT\n"+ "wait \"$user_pid\"\n", shellutil.Quote(metaPath), shellutil.Quote(runningTemplate), @@ -1295,7 +1320,9 @@ func detachWrapperScript(jobID string, argv []string, cleanupPaths []string, met quotedArgs, cleanupLines, detachPIDPlaceholder, + detachPGIDPlaceholder, detachPIDPlaceholder, + detachPGIDPlaceholder, detachExitPlaceholder, ) } @@ -1305,16 +1332,15 @@ func mustDetachJSONTemplate(v any) string { if err != nil { panic(err) } - out := strings.Replace(string(data), `"pid":0`, `"pid":`+detachPIDPlaceholder, 1) + // Replace "pgid" before "pid": the two keys share a suffix and the pgid + // replacement must not consume the pid occurrence. + out := strings.Replace(string(data), `"pgid":0`, `"pgid":`+detachPGIDPlaceholder, 1) + out = strings.Replace(out, `"pid":0`, `"pid":`+detachPIDPlaceholder, 1) return out } func mustDetachJSONTemplateWithExit(v detachMetadata) string { - data, err := json.Marshal(v) - if err != nil { - panic(err) - } - out := strings.Replace(string(data), `"pid":0`, `"pid":`+detachPIDPlaceholder, 1) + out := mustDetachJSONTemplate(v) out = strings.Replace(out, `"exitCode":0`, `"exitCode":`+detachExitPlaceholder, 1) return out } diff --git a/internal/cli/pdsh_group_linux_test.go b/internal/cli/pdsh_group_linux_test.go new file mode 100644 index 00000000..4e130f24 --- /dev/null +++ b/internal/cli/pdsh_group_linux_test.go @@ -0,0 +1,119 @@ +package cli + +import ( + "encoding/json" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "syscall" + "testing" + "time" +) + +// Verifies the whole group-kill chain against real processes: the detach +// launcher puts the user command in its own process group via setsid, the +// metadata records the pgid, and the signal script kills the entire tree — +// including children the leader forked (the torchrun/pretrain.py scenario). +// Linux-only: the scripts inspect /proc. +func TestDetachGroupSignalKillsProcessTreeLinux(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("requires /proc and process groups") + } + jobID := "job-group-e2e" + spec := newDetachJobSpec(jobID, "pod-x", "dev", "tree", []string{"sh", "-c", "sleep 300 & sleep 300 & wait"}, nil) + cmd := detachCommand(spec) + launcher := exec.Command(cmd[0], cmd[1], cmd[2]) + stdout, err := launcher.Output() + if err != nil { + t.Fatalf("launcher failed: %v", err) + } + var info detachLaunchInfo + lines := strings.Split(strings.TrimSpace(string(stdout)), "\n") + if err := json.Unmarshal([]byte(lines[len(lines)-1]), &info); err != nil { + t.Fatalf("parse launch record %q: %v", stdout, err) + } + metaRaw, err := os.ReadFile(info.MetaPath) + if err != nil { + t.Fatalf("read metadata: %v", err) + } + var meta detachMetadata + if err := json.Unmarshal(metaRaw, &meta); err != nil { + t.Fatalf("parse metadata %q: %v", metaRaw, err) + } + if meta.PGID <= 0 || meta.PGID != meta.PID { + t.Fatalf("expected setsid leader with pgid==pid, got %+v", meta) + } + defer func() { + _ = syscall.Kill(-meta.PGID, syscall.SIGKILL) + dir := filepath.Dir(info.MetaPath) + for _, f := range []string{info.MetaPath, info.LogPath, filepath.Join(dir, jobID+".sh")} { + _ = os.Remove(f) + } + }() + + // Leader sh + two sleeps: wait for the full tree before killing so the + // assertion below proves the *children* died, not just the leader. + waitForGroupSize(t, meta.PGID, 3, 5*time.Second) + + script := signalDetachJobScript(jobID, meta.PID, meta.PGID, "KILL") + out, err := exec.Command("sh", "-c", script).CombinedOutput() + if err != nil { + t.Fatalf("signal script failed: %v\noutput: %s", err, out) + } + if !strings.Contains(string(out), "signaled group (3 members)") { + t.Fatalf("expected group signal confirmation, got %q", out) + } + + waitForGroupSize(t, meta.PGID, 0, 5*time.Second) +} + +func waitForGroupSize(t *testing.T, pgid, want int, timeout time.Duration) { + t.Helper() + deadline := time.Now().Add(timeout) + for { + got := countGroupMembers(pgid) + if got == want { + return + } + if time.Now().After(deadline) { + t.Fatalf("process group %d has %d members, want %d", pgid, got, want) + } + time.Sleep(20 * time.Millisecond) + } +} + +// countGroupMembers counts non-zombie processes in the group by parsing +// /proc//stat, mirroring the shell scripts' technique (strip through +// the last ')' to survive spaces/parens in comm). +func countGroupMembers(pgid int) int { + entries, err := os.ReadDir("/proc") + if err != nil { + return 0 + } + count := 0 + for _, e := range entries { + if _, err := strconv.Atoi(e.Name()); err != nil { + continue + } + data, err := os.ReadFile(filepath.Join("/proc", e.Name(), "stat")) + if err != nil { + continue + } + s := string(data) + idx := strings.LastIndexByte(s, ')') + if idx < 0 { + continue + } + fields := strings.Fields(s[idx+1:]) + if len(fields) < 3 || fields[0] == "Z" { + continue + } + if g, err := strconv.Atoi(fields[2]); err == nil && g == pgid { + count++ + } + } + return count +} diff --git a/internal/cli/pdsh_test.go b/internal/cli/pdsh_test.go index 9779168d..7afb83ac 100644 --- a/internal/cli/pdsh_test.go +++ b/internal/cli/pdsh_test.go @@ -124,13 +124,20 @@ func TestDetachCommand(t *testing.T) { // clobbered by a later 'running' write from the launcher. "trap 'rc=$?", "' EXIT", - // The user command must run via `(exec CMD) &` so $! captures the - // pid of the user command itself (via execve), not the wrapper - // shell. This is the pid we report to the caller and the one that - // responds to `kill `. + // The user command runs via setsid so it leads its own process + // group (pgid == pid) and `jobs stop` can signal the whole tree; + // `(exec CMD) &` remains the fallback for images without setsid, + // recording pgid=0 so stop degrades to leader-only signaling. + "if command -v setsid >/dev/null 2>&1; then", + "setsid \"$@\" &", + "user_pgid=$user_pid", "(exec \"$@\") &", + "user_pgid=0", "user_pid=$!", "wait \"$user_pid\"", + // Both metadata writes resolve the pgid placeholder. + "s/__OKDEV_DETACH_PGID__/$user_pgid/g", + "\"pgid\":__OKDEV_DETACH_PGID__", // OKDEV_JOB_ID is exported so exec-jobs can reconcile liveness via // /proc//environ without relying on unique cmdlines. "export OKDEV_JOB_ID='job-123'", From a11055543630c665fa3880d0b3df7822ecf31871 Mon Sep 17 00:00:00 2001 From: acmore Date: Sun, 5 Jul 2026 13:26:27 +0800 Subject: [PATCH 2/3] 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 "-" 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 --- internal/cli/jobs.go | 6 +++++- internal/cli/jobs_stop_test.go | 4 ++-- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/internal/cli/jobs.go b/internal/cli/jobs.go index 3f58b02f..ae199419 100644 --- a/internal/cli/jobs.go +++ b/internal/cli/jobs.go @@ -657,7 +657,11 @@ func signalDetachJobScript(jobID string, pid, pgid int, signalName string) strin "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)\n"+ "for m in $members; do\n"+ " if tr '\\000' '\\n' < \"/proc/$m/environ\" 2>/dev/null | grep -Fqx %s; then\n"+ - " if kill -%s -- \"-$pgid\" 2>/dev/null; then\n"+ + // No `--` before the negative pgid: dash's kill builtin + // rejects it ("Illegal number: -"); the bare "-" + // operand after the signal option works in dash, bash, + // zsh, and busybox ash alike. + " if kill -%s \"-$pgid\" 2>/dev/null; then\n"+ " echo \"signaled group ($(set -- $members; echo $#) members)\"\n"+ " exit 0\n"+ " fi\n"+ diff --git a/internal/cli/jobs_stop_test.go b/internal/cli/jobs_stop_test.go index d01e0aa2..6bb26a2d 100644 --- a/internal/cli/jobs_stop_test.go +++ b/internal/cli/jobs_stop_test.go @@ -91,7 +91,7 @@ func TestRunJobsStopSignalsLingeringGroupAfterLeaderExit(t *testing.T) { if !strings.Contains(got, "sent SIGTERM pgid=100 (signaled group (2 members))") { t.Fatalf("expected group TERM line, got %q", got) } - if len(client.execScripts) != 1 || !strings.Contains(client.execScripts[0], `kill -TERM -- "-$pgid"`) { + if len(client.execScripts) != 1 || !strings.Contains(client.execScripts[0], `kill -TERM "-$pgid"`) { t.Fatalf("expected a single group TERM script, got %+v", client.execScripts) } } @@ -155,7 +155,7 @@ func TestStopRowNeedsSignalAndSettled(t *testing.T) { func TestSignalDetachJobScriptGroupAndFallback(t *testing.T) { group := signalDetachJobScript("job-1", 100, 100, "TERM") for _, want := range []string{ - `kill -TERM -- "-$pgid"`, + `kill -TERM "-$pgid"`, "/proc/[0-9]*/stat", "OKDEV_JOB_ID=job-1", // Legacy leader path stays as the fallback tail when the group is From 7cac7be9b11e3816c66c0481465744127014c7bf Mon Sep 17 00:00:00 2001 From: acmore Date: Sun, 5 Jul 2026 14:24:28 +0800 Subject: [PATCH 3/3] test(e2e): cover detach runtime-volume logs, group stop, sidecar fallback in kind smoke MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- scripts/e2e_kind_smoke.sh | 106 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 106 insertions(+) diff --git a/scripts/e2e_kind_smoke.sh b/scripts/e2e_kind_smoke.sh index 205fce4b..bb428fd2 100644 --- a/scripts/e2e_kind_smoke.sh +++ b/scripts/e2e_kind_smoke.sh @@ -496,6 +496,112 @@ if [[ "$RECONCILED_IMAGE" != "ubuntu:24.04" ]]; then fi echo "Pod reconcile verified" +# --- Detached jobs: runtime-volume logs, process-group stop, sidecar ------- +# fallback. Covers the detach chain end-to-end on a real pod: logs land on +# the shared /var/okdev volume, `jobs stop` kills the whole process tree +# (not just the leader), and logs stay readable through the sidecar after +# the dev container dies. The container-crash step is deliberately last +# before teardown: it leaves the dev container in CrashLoopBackOff. + +echo "Testing exec --detach places logs on the runtime volume" +DETACH_OUTPUT=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --detach -- sh -c 'sleep 7301 & sleep 7302 & wait') +echo "$DETACH_OUTPUT" +if [[ "$DETACH_OUTPUT" != *"log=/var/okdev/exec/"* ]]; then + echo "ERROR: expected detach log under /var/okdev/exec, got: $DETACH_OUTPUT" >&2 + exit 1 +fi +TREE_JOB_ID=$(printf '%s\n' "$DETACH_OUTPUT" | sed -n 's/.*job_id=\([^ ]*\).*/\1/p' | head -n1) +if [[ -z "$TREE_JOB_ID" ]]; then + echo "ERROR: could not extract detach job id from: $DETACH_OUTPUT" >&2 + exit 1 +fi +echo "detach on runtime volume verified (job $TREE_JOB_ID)" + +echo "Testing jobs list reports the live process group" +JOBS_JSON=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" --output json jobs list) +OKDEV_JOBS_JSON="$JOBS_JSON" OKDEV_TREE_JOB_ID="$TREE_JOB_ID" python3 - <<'PY' +import json, os, sys + +doc = json.loads(os.environ["OKDEV_JOBS_JSON"]) +job_id = os.environ["OKDEV_TREE_JOB_ID"] +jobs = [j for j in doc.get("jobs", []) if j.get("jobId") == job_id] +if not jobs: + print(f"ERROR: job {job_id} missing from jobs list: {doc}", file=sys.stderr) + sys.exit(1) +job = jobs[0] +if not job.get("state", "").startswith("running"): + print(f"ERROR: expected running state, got {job}", file=sys.stderr) + sys.exit(1) +row = job["podStates"][0] +if row.get("pgid", 0) <= 0: + print(f"ERROR: expected pgid > 0 (setsid group), got {row}", file=sys.stderr) + sys.exit(1) +# Leader sh plus the two background sleeps. +if row.get("groupLive", 0) < 3: + print(f"ERROR: expected >=3 live group members, got {row}", file=sys.stderr) + sys.exit(1) +PY +echo "jobs list process group verified" + +echo "Testing jobs stop kills the whole process tree" +STOP_OUTPUT=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" jobs stop "$TREE_JOB_ID") +echo "$STOP_OUTPUT" +if [[ "$STOP_OUTPUT" != *"pgid="* ]]; then + echo "ERROR: expected group signal (pgid=...) in stop output, got: $STOP_OUTPUT" >&2 + exit 1 +fi +LEFTOVER=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'n=0; for d in /proc/[0-9]*/cmdline; do c=$(tr "\0" " " < "$d" 2>/dev/null || true); case "$c" in "sleep 7301 "|"sleep 7302 ") n=$((n+1)) ;; esac; done; echo "leftover=$n"') +if [[ "$LEFTOVER" != *"leftover=0"* ]]; then + echo "ERROR: expected no surviving group processes after stop, got: $LEFTOVER" >&2 + exit 1 +fi +echo "jobs group stop verified" + +echo "Testing detached job logs survive dev-container death via sidecar" +MARKER_OUTPUT=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --detach -- sh -c 'echo sidecar-fallback-marker; exec sleep 7303') +MARKER_JOB_ID=$(printf '%s\n' "$MARKER_OUTPUT" | sed -n 's/.*job_id=\([^ ]*\).*/\1/p' | head -n1) +if [[ -z "$MARKER_JOB_ID" ]]; then + echo "ERROR: could not extract marker job id from: $MARKER_OUTPUT" >&2 + exit 1 +fi +LOGS_HEALTHY=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" jobs logs "$MARKER_JOB_ID") +if [[ "$LOGS_HEALTHY" != *"sidecar-fallback-marker"* ]]; then + echo "ERROR: expected marker in healthy-container job logs, got: $LOGS_HEALTHY" >&2 + exit 1 +fi + +echo "Crashing dev container into CrashLoopBackOff" +CRASH_POD_NAME=$(session_attachable_pod_name "$NAMESPACE" "$SESSION_NAME") +DEV_WAITING_JSONPATH='{.status.containerStatuses[?(@.name=="dev")].state.waiting.reason}' +CRASHED=0 +for attempt in $(seq 1 30); do + REASON=$(kubectl -n "$NAMESPACE" get pod "$CRASH_POD_NAME" -o jsonpath="$DEV_WAITING_JSONPATH" 2>/dev/null || true) + if [[ "$REASON" == "CrashLoopBackOff" ]]; then + CRASHED=1 + break + fi + # Kill the dev container's main process (sleep infinity, from the config + # template). The exec session dies with the container, so tolerate + # errors; kubelet escalates to CrashLoopBackOff after repeated crashes. + "$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" exec --no-tty --no-prefix -- sh -lc 'for d in /proc/[0-9]*/cmdline; do if [ "$(tr "\0" " " < "$d" 2>/dev/null)" = "sleep infinity " ]; then p=${d%/cmdline}; kill -9 "${p#/proc/}" 2>/dev/null || true; fi; done' >/dev/null 2>&1 || true + sleep 2 +done +if [[ "$CRASHED" -ne 1 ]]; then + echo "ERROR: dev container did not enter CrashLoopBackOff" >&2 + kubectl -n "$NAMESPACE" get pod "$CRASH_POD_NAME" -o jsonpath='{.status.containerStatuses}' >&2 || true + exit 1 +fi +echo "dev container crashed (CrashLoopBackOff)" + +LOGS_FALLBACK=$("$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" jobs logs "$MARKER_JOB_ID") +if [[ "$LOGS_FALLBACK" != *"sidecar-fallback-marker"* ]]; then + echo "ERROR: expected marker via sidecar log fallback, got: $LOGS_FALLBACK" >&2 + exit 1 +fi +echo "sidecar log fallback verified" + +echo "Detached job tests completed" + echo "Testing explicit okdev down" "$OKDEV_BIN" --config "$CFG_PATH" --session "$SESSION_NAME" down --yes assert_no_local_sync_processes "$SESSION_NAME" "$SYNC_HOME"