Skip to content

Commit 439129b

Browse files
vishrvishr
authored
feat(jobs): unify execution supervision and history (#191)
* feat(jobs): unify execution supervision and history * fix(jobs): preserve durable operator execution
1 parent 45d7614 commit 439129b

69 files changed

Lines changed: 1067 additions & 566 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎cmd/ob/job.go‎

Lines changed: 119 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,11 @@
11
package main
22

33
import (
4+
"bytes"
45
"errors"
56
"fmt"
7+
"strconv"
8+
"text/tabwriter"
69
"time"
710

811
"github.com/spf13/cobra"
@@ -14,31 +17,38 @@ import (
1417
func addJobCommand(root *cobra.Command, g *globalFlags) {
1518
group := &cobra.Command{
1619
Use: "job",
17-
Short: "plan and run a sealed one-shot manual job",
18-
Long: "Plan and run one declared `when: manual` job against the current serving release.\n\nThe job remains in the release runtime but never runs during deploy. Saved plans\nbind its release, runtime digest, immutable image and data effect so agents can\nobtain a separate approval before execution.",
20+
Short: "plan and run a sealed one-shot operator job",
21+
Long: "Plan and run one declared `operator_run: allowed` job against the current serving release.\n\nDeployment participation is independent: `deployment_phase` may be none, pre_release,\nor post_release. Saved plans bind the release, runtime digest, immutable image,\ndata effect and inputs so agents can obtain separate approval before execution.",
1922
Args: cobra.NoArgs,
2023
RunE: showCommandHelp,
2124
}
2225

2326
var planOut, backupReportOut string
27+
var planInputs []string
2428
plan := &cobra.Command{
2529
Use: "plan <id>",
2630
Short: "seal a current-release-bound one-shot job plan",
2731
Long: "Observe the current serving release and write a short-lived executable job plan.\n\nThe plan binds the exact runtime digest, digest-pinned image, job data effect,\ntarget and expiry. It reads the target and writes only the local plan artifact.",
2832
Args: cobra.ExactArgs(1),
2933
RunE: func(cmd *cobra.Command, args []string) error {
30-
return runJobPlan(cmd, g, args[0], planOut, backupReportOut)
34+
inputs, err := parseScheduleInputs(planInputs)
35+
if err != nil {
36+
return writeEarlyOperationFailure(cmd, g, codedError("job_input_invalid", "%v", err))
37+
}
38+
return runJobPlan(cmd, g, args[0], inputs, planOut, backupReportOut)
3139
},
3240
}
3341
plan.Flags().StringVarP(&planOut, "out", "o", "ob-job-plan.json", "job plan artifact path")
3442
plan.Flags().StringVar(&backupReportOut, "backup-report-out", "", "write a plan-bound backup report template when migration backup is required")
43+
plan.Flags().StringArrayVar(&planInputs, "input", nil, "input override as NAME=VALUE; repeatable and sealed into the plan")
3544

3645
var planPath, approvalPath, backupReportPath, overrideReason string
46+
var runInputs []string
3747
var breakLock, detach bool
3848
run := &cobra.Command{
3949
Use: "run [id]",
40-
Short: "run one manual job from an inline or saved sealed plan",
41-
Long: "Run one manual job through the canonical lock, fence, local-confirmation and journal boundary.\n\n" +
50+
Short: "run one operator job from an inline or saved sealed plan",
51+
Long: "Run one operator job through the canonical lock, fence, local-confirmation and journal boundary.\n\n" +
4252
"A job with schedule configured runs under its installed systemd unit and is\nfollowed by default; Ctrl-C stops following, not the host job. --detach returns\nafter that unit accepts the run. Unscheduled and migration jobs stay attached.\n\n" +
4353
"Humans may pass an id and confirm interactively. Automation should supply a\nsaved --plan and its separately created local-confirmation artifact through\n--approval; migration plans may also require the exact plan-bound --backup-report.",
4454
Args: cobra.MaximumNArgs(1),
@@ -47,7 +57,11 @@ func addJobCommand(root *cobra.Command, g *globalFlags) {
4757
if len(args) == 1 {
4858
jobID = args[0]
4959
}
50-
return runJob(cmd, g, jobID, planPath, approvalPath, backupReportPath, overrideReason, breakLock, detach)
60+
inputs, err := parseScheduleInputs(runInputs)
61+
if err != nil {
62+
return writeEarlyOperationFailure(cmd, g, codedError("job_input_invalid", "%v", err))
63+
}
64+
return runJob(cmd, g, jobID, inputs, planPath, approvalPath, backupReportPath, overrideReason, breakLock, detach)
5165
},
5266
}
5367
run.Flags().StringVar(&planPath, "plan", "", "apply a saved job plan artifact")
@@ -56,13 +70,15 @@ func addJobCommand(root *cobra.Command, g *globalFlags) {
5670
run.Flags().StringVar(&overrideReason, "override-migration-backup", "", "audited break-glass reason (requires --approval)")
5771
run.Flags().BoolVar(&breakLock, "break-lock", false, "break a stale operation lock after inspecting its holder")
5872
run.Flags().BoolVar(&detach, "detach", false, "return after the installed host unit accepts the job")
73+
run.Flags().StringArrayVar(&runInputs, "input", nil, "input override as NAME=VALUE; repeatable and sealed into the inline plan")
5974

75+
addJobReadCommands(group, g)
6076
group.AddCommand(plan, run)
6177
root.AddCommand(group)
6278
}
6379

64-
func runJobPlan(cmd *cobra.Command, g *globalFlags, jobID, outPath, backupReportOut string) error {
65-
plan, err := operationsService(cmd, g).PlanJob(cmd.Context(), onebox.PlanJobRequest{Job: jobID})
80+
func runJobPlan(cmd *cobra.Command, g *globalFlags, jobID string, inputs map[string]string, outPath, backupReportOut string) error {
81+
plan, err := operationsService(cmd, g).PlanJob(cmd.Context(), onebox.PlanJobRequest{Job: jobID, Inputs: inputs})
6682
if err != nil {
6783
return writeStructuredCommandFailure(cmd, g, "job_plan_failed", "job planning failed; inspect stderr for local diagnostics", err)
6884
}
@@ -166,10 +182,13 @@ func renderJobPlan(cmd *cobra.Command, plan *onebox.JobPlan) {
166182
}
167183
}
168184

169-
func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, backupReportPath, overrideReason string, breakLock, detach bool) error {
185+
func runJob(cmd *cobra.Command, g *globalFlags, jobID string, inputs map[string]string, planPath, approvalPath, backupReportPath, overrideReason string, breakLock, detach bool) error {
170186
if planPath != "" && jobID != "" {
171187
return writeEarlyOperationFailure(cmd, g, errors.New("supply either a job id or --plan, not both"))
172188
}
189+
if planPath != "" && len(inputs) > 0 {
190+
return writeEarlyOperationFailure(cmd, g, errors.New("--input is sealed into a plan; supply it to ob job plan, not ob job run --plan"))
191+
}
173192
if planPath == "" && jobID == "" {
174193
return writeEarlyOperationFailure(cmd, g, errors.New("job run requires an id or --plan"))
175194
}
@@ -224,7 +243,7 @@ func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, b
224243
}, "job run")
225244
}
226245

227-
plan, err := operationsService(cmd, g).PlanJob(cmd.Context(), onebox.PlanJobRequest{Job: jobID})
246+
plan, err := operationsService(cmd, g).PlanJob(cmd.Context(), onebox.PlanJobRequest{Job: jobID, Inputs: inputs})
228247
if err != nil {
229248
return err
230249
}
@@ -251,6 +270,96 @@ func runJob(cmd *cobra.Command, g *globalFlags, jobID, planPath, approvalPath, b
251270
return runMutation(cmd, g, onebox.ExecuteRequest{Kind: onebox.KindJobRun, JobPlan: &plan, Approval: approval, BreakLock: breakLock, Detach: detach}, "job run")
252271
}
253272

273+
func addJobReadCommands(group *cobra.Command, g *globalFlags) {
274+
var historyCount int
275+
history := &cobra.Command{
276+
Use: "history <job>", Short: "execution records of one job, newest first",
277+
Long: "Read retained execution records for one job across timer and operator triggers. Host-supervised records come from journald; sealed attached runs come from the operation journal. The result is retention-bounded evidence, so an absent success means no retained success was observed, not that the job never succeeded.",
278+
Args: cobra.ExactArgs(1),
279+
RunE: func(cmd *cobra.Command, args []string) error {
280+
cfg, p, err := loadAllLenient(cmd.Context(), g)
281+
if err != nil {
282+
return writeStructuredReadFailure(cmd, g, err)
283+
}
284+
e, cleanup, err := connect(cmd, g, cfg, p, newUI(cmd, g))
285+
if err != nil {
286+
return writeStructuredReadFailure(cmd, g, err)
287+
}
288+
defer cleanup()
289+
records, err := e.JobHistory(cmd.Context(), args[0], historyCount)
290+
if err != nil {
291+
return writeStructuredCommandFailure(cmd, g, "job_history_failed", "job history could not be read", err)
292+
}
293+
if isStructuredOutput(g) {
294+
return writeFiniteSuccess(cmd, g, map[string]any{"job": args[0], "runs": records})
295+
}
296+
if len(records) == 0 {
297+
fmt.Fprintf(cmd.OutOrStdout(), "no retained executions for %s\n", args[0])
298+
return nil
299+
}
300+
w := tabwriter.NewWriter(cmd.OutOrStdout(), 0, 0, 2, ' ', 0)
301+
fmt.Fprintln(w, "STARTED\tOUTCOME\tDURATION\tATTEMPTS\tEXIT\tTRIGGER\tRELEASE\tOPERATOR\tRUN")
302+
for _, r := range records {
303+
exit := "-"
304+
if r.ExitStatus != nil {
305+
exit = strconv.Itoa(*r.ExitStatus)
306+
}
307+
fmt.Fprintf(w, "%s\t%s\t%ds\t%d\t%s\t%s\t%s\t%s\t%s\n", r.StartedAt, r.Outcome, r.DurationSeconds, r.Attempts, exit, r.Trigger, orDash(r.Release), orDash(r.Operator), r.ID)
308+
}
309+
return w.Flush()
310+
},
311+
}
312+
history.Flags().IntVarP(&historyCount, "count", "n", 20, "number of newest executions to show")
313+
314+
var logsRun string
315+
logs := &cobra.Command{
316+
Use: "logs <job>", Short: "journal of one host-supervised job execution",
317+
Long: "Stream the exact systemd journal of a host-supervised job execution. By default the newest retained run is selected; --run accepts the run id printed by ob job history. Attached sealed executions retain outcome evidence but do not have a separate host log stream.",
318+
Args: cobra.ExactArgs(1),
319+
RunE: func(cmd *cobra.Command, args []string) error {
320+
cfg, p, err := loadAllLenient(cmd.Context(), g)
321+
if err != nil {
322+
return writeStructuredReadFailure(cmd, g, err)
323+
}
324+
e, cleanup, err := connect(cmd, g, cfg, p, newUI(cmd, g))
325+
if err != nil {
326+
return writeStructuredReadFailure(cmd, g, err)
327+
}
328+
defer cleanup()
329+
if g.Output == "json" {
330+
var stdout, stderr bytes.Buffer
331+
run, err := e.JobLogs(cmd.Context(), args[0], logsRun, &stdout, &stderr)
332+
data := map[string]any{"job": args[0], "run": run, "stdout": stdout.String(), "stderr": stderr.String(), "passthrough_unredacted": true}
333+
if err != nil {
334+
publicErr := publicError(err, "job_logs_failed", "job logs could not be read")
335+
publicErr.Details = data
336+
if writeErr := writeFiniteOutcome(cmd, g, cliOutcomeError, nil, publicErr); writeErr != nil {
337+
return writeErr
338+
}
339+
return withExitCode(err, 1)
340+
}
341+
return writeFiniteSuccess(cmd, g, data)
342+
}
343+
if g.Output == "ndjson" {
344+
stream := newCLIRecordStream(cmd.OutOrStdout(), commandName(cmd))
345+
run, err := e.JobLogs(cmd.Context(), args[0], logsRun, stream.channelWriter("stdout"), stream.channelWriter("stderr"))
346+
data := map[string]any{"job": args[0], "run": run, "passthrough_unredacted": true}
347+
if err != nil {
348+
if writeErr := stream.terminal(cliOutcomeError, nil, publicError(err, "job_logs_failed", "job logs could not be read")); writeErr != nil {
349+
return writeErr
350+
}
351+
return withExitCode(err, 1)
352+
}
353+
return stream.terminal(cliOutcomeSuccess, data, nil)
354+
}
355+
_, err = e.JobLogs(cmd.Context(), args[0], logsRun, cmd.OutOrStdout(), cmd.ErrOrStderr())
356+
return err
357+
},
358+
}
359+
logs.Flags().StringVar(&logsRun, "run", "", "run id from ob job history; defaults to newest host-supervised run")
360+
group.AddCommand(history, logs)
361+
}
362+
254363
func loadJobMigrationOverride(
255364
plan *onebox.JobPlan,
256365
approval *onebox.ApprovalGrant,

‎cmd/ob/output.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -96,6 +96,8 @@ var cliOutputMatrix = map[string]cliOutputClass{
9696
"ob backup verify": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
9797
"ob backup status": {Class: cliClassFiniteEnvelope, JSON: true},
9898
"ob job plan": {Class: cliClassFiniteEnvelope, JSON: true},
99+
"ob job history": {Class: cliClassFiniteEnvelope, JSON: true},
100+
"ob job logs": {Class: cliClassOperatorPassthrough, JSON: true, NDJSON: true},
99101
"ob job run": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
100102
"ob logs": {Class: cliClassOperatorPassthrough, JSON: true, NDJSON: true},
101103
"ob plan": {Class: cliClassFiniteEnvelope, JSON: true},
@@ -105,12 +107,9 @@ var cliOutputMatrix = map[string]cliOutputClass{
105107
"ob resume": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
106108
"ob rollback": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
107109
"ob schedule apply": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
108-
"ob schedule history": {Class: cliClassFiniteEnvelope, JSON: true},
109110
"ob schedule list": {Class: cliClassFiniteEnvelope, JSON: true},
110-
"ob schedule logs": {Class: cliClassOperatorPassthrough, JSON: true, NDJSON: true},
111111
"ob schedule pause": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
112112
"ob schedule resume": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
113-
"ob schedule run": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},
114113
"ob execution list": {Class: cliClassFiniteEnvelope, JSON: true},
115114
"ob execution inspect": {Class: cliClassFiniteEnvelope, JSON: true},
116115
"ob execution resume": {Class: cliClassFiniteStream, JSON: true, NDJSON: true},

‎cmd/ob/output_test.go‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -502,6 +502,8 @@ func TestLeafOutputMatrixIsClosedAndHasNoAliases(t *testing.T) {
502502
"ob exec": {Class: "operator_passthrough", NDJSON: true},
503503
"ob init": {Class: "finite_envelope", JSON: true},
504504
"ob job plan": {Class: "finite_envelope", JSON: true},
505+
"ob job history": {Class: "finite_envelope", JSON: true},
506+
"ob job logs": {Class: "operator_passthrough", JSON: true, NDJSON: true},
505507
"ob backup create": {Class: "finite_stream", JSON: true, NDJSON: true},
506508
"ob backup enable": {Class: "finite_stream", JSON: true, NDJSON: true},
507509
"ob backup disable": {Class: "finite_stream", JSON: true, NDJSON: true},
@@ -519,12 +521,9 @@ func TestLeafOutputMatrixIsClosedAndHasNoAliases(t *testing.T) {
519521
"ob resume": {Class: "finite_stream", JSON: true, NDJSON: true},
520522
"ob rollback": {Class: "finite_stream", JSON: true, NDJSON: true},
521523
"ob schedule apply": {Class: "finite_stream", JSON: true, NDJSON: true},
522-
"ob schedule history": {Class: "finite_envelope", JSON: true},
523524
"ob schedule list": {Class: "finite_envelope", JSON: true},
524-
"ob schedule logs": {Class: "operator_passthrough", JSON: true, NDJSON: true},
525525
"ob schedule pause": {Class: "finite_stream", JSON: true, NDJSON: true},
526526
"ob schedule resume": {Class: "finite_stream", JSON: true, NDJSON: true},
527-
"ob schedule run": {Class: "finite_stream", JSON: true, NDJSON: true},
528527
"ob execution list": {Class: "finite_envelope", JSON: true},
529528
"ob execution inspect": {Class: "finite_envelope", JSON: true},
530529
"ob execution resume": {Class: "finite_stream", JSON: true, NDJSON: true},

0 commit comments

Comments
 (0)