Commit 55df8b9
Eric Bower
·
2026-05-06 23:44:47 -0400 EDT
parent 642d214
feat(runner): --wait command for debugging
2 files changed,
+219,
-19
M
main.go
+1,
-0
M
main.go
+218,
-19
| ... | ... | @@ -12,8 +12,10 @@ import ( | |
| 12 | 12 | "log/slog" | |
| 13 | 13 | "os" | |
| 14 | 14 | "os/exec" | |
| 15 | + | "os/signal" | |
| 15 | 16 | "path/filepath" | |
| 16 | 17 | "strings" | |
| 18 | + | "syscall" | |
| 17 | 19 | "time" | |
| 18 | 20 | ||
| 19 | 21 | "github.com/golang-cz/devslog" |
| ... | ... | @@ -43,6 +45,7 @@ type Cfg struct { | |
| 43 | 45 | StatusOutput io.Writer // where status JSONL is written (default: os.Stdout) | |
| 44 | 46 | StatusFilter string // "terminal" (default) or "all" | |
| 45 | 47 | HumanOutput bool // human-readable output instead of JSONL / slog | |
| 48 | + | Wait bool // block until job completes, print history and summary | |
| 46 | 49 | } | |
| 47 | 50 | ||
| 48 | 51 | type Event struct { |
| ... | ... | @@ -53,7 +56,7 @@ type Event struct { | |
| 53 | 56 | ArtifactURL string `json:"artifact_url"` // public URL for artifacts (e.g. https://{user}-artifacts.pgs.sh/{repo}) | |
| 54 | 57 | } | |
| 55 | 58 | ||
| 56 | - | func NewCfg() *Cfg { | |
| 59 | + | func NewCfg() (*Cfg, string, bool) { | |
| 57 | 60 | var keyLoc, certLoc, artifactDir, event string | |
| 58 | 61 | var monitorInterval time.Duration | |
| 59 | 62 | var logLevel string |
| ... | ... | @@ -67,9 +70,15 @@ func NewCfg() *Cfg { | |
| 67 | 70 | flag.BoolVar(&structured, "structured", false, "use structured key=value log output") | |
| 68 | 71 | var statusFilter string | |
| 69 | 72 | var human bool | |
| 73 | + | var wait bool | |
| 70 | 74 | flag.StringVar(&statusFilter, "status-filter", "terminal", "status output filter: terminal (default) or all") | |
| 71 | 75 | flag.BoolVar(&human, "human", false, "human-readable output (default: JSONL / slog)") | |
| 72 | - | flag.Parse() | |
| 76 | + | flag.BoolVar(&wait, "wait", false, "block until job completes, printing session history and summary") | |
| 77 | + | ||
| 78 | + | // Split args so the subcommand (first non-flag arg) doesn't block | |
| 79 | + | // flags that appear after it: "pici runner --wait" works. | |
| 80 | + | flags, cmd, wantHelp := splitCommand(os.Args[1:]) | |
| 81 | + | flag.CommandLine.Parse(flags) | |
| 73 | 82 | ||
| 74 | 83 | logger := newLogger("ci", logLevel, structured) | |
| 75 | 84 | ctx, cancel := context.WithCancel(context.Background()) |
| ... | ... | @@ -85,7 +94,27 @@ func NewCfg() *Cfg { | |
| 85 | 94 | MonitorInterval: monitorInterval, | |
| 86 | 95 | StatusFilter: statusFilter, | |
| 87 | 96 | HumanOutput: human, | |
| 97 | + | Wait: wait, | |
| 98 | + | }, cmd, wantHelp | |
| 99 | + | } | |
| 100 | + | ||
| 101 | + | // splitCommand separates the first non-flag argument (the subcommand) from | |
| 102 | + | // the rest of the flags, so "runner --wait" becomes flags=["--wait"], cmd="runner". | |
| 103 | + | // It strips --help/help so we can print custom help per subcommand. | |
| 104 | + | func splitCommand(args []string) (flags []string, cmd string, wantHelp bool) { | |
| 105 | + | flags = make([]string, 0, len(args)) | |
| 106 | + | for _, arg := range args { | |
| 107 | + | if arg == "--help" || arg == "help" { | |
| 108 | + | wantHelp = true | |
| 109 | + | continue | |
| 110 | + | } | |
| 111 | + | if cmd == "" && arg != "" && !strings.HasPrefix(arg, "-") { | |
| 112 | + | cmd = arg | |
| 113 | + | } else { | |
| 114 | + | flags = append(flags, arg) | |
| 115 | + | } | |
| 88 | 116 | } | |
| 117 | + | return flags, cmd, wantHelp | |
| 89 | 118 | } | |
| 90 | 119 | ||
| 91 | 120 | func parseLogLevel(s string) slog.Level { |
| ... | ... | @@ -116,18 +145,22 @@ func newLogger(space string, levelStr string, structured bool) *slog.Logger { | |
| 116 | 145 | } | |
| 117 | 146 | ||
| 118 | 147 | func main() { | |
| 119 | - | cfg := NewCfg() | |
| 120 | - | cmd := flag.Arg(0) | |
| 148 | + | cfg, cmd, wantHelp := NewCfg() | |
| 121 | 149 | ||
| 122 | 150 | cfg.Logger.Debug("setting up ci", "cfg", cfg) | |
| 123 | 151 | cfg.Logger.Debug("running cmd", "cmd", cmd) | |
| 124 | 152 | ||
| 125 | - | switch cmd { | |
| 126 | - | case "runner": | |
| 127 | - | if flag.Arg(1) == "--help" || flag.Arg(1) == "help" { | |
| 153 | + | if wantHelp && (cmd == "runner" || cmd == "monitor") { | |
| 154 | + | if cmd == "runner" { | |
| 128 | 155 | printRunnerHelp() | |
| 129 | - | return | |
| 156 | + | } else { | |
| 157 | + | printMonitorHelp() | |
| 130 | 158 | } | |
| 159 | + | return | |
| 160 | + | } | |
| 161 | + | ||
| 162 | + | switch cmd { | |
| 163 | + | case "runner": | |
| 131 | 164 | cfg.Logger.Debug("starting runner") | |
| 132 | 165 | if err := RunRunner(cfg); err != nil { | |
| 133 | 166 | cfg.Logger.Error("runner failed", "err", err) |
| ... | ... | @@ -146,10 +179,6 @@ func main() { | |
| 146 | 179 | os.Exit(1) | |
| 147 | 180 | } | |
| 148 | 181 | case "monitor": | |
| 149 | - | if flag.Arg(1) == "--help" || flag.Arg(1) == "help" { | |
| 150 | - | printMonitorHelp() | |
| 151 | - | return | |
| 152 | - | } | |
| 153 | 182 | cfg.Logger.Debug("starting monitor") | |
| 154 | 183 | if err := runMonitor(cfg); err != nil { | |
| 155 | 184 | cfg.Logger.Error("monitor failed", "err", err) |
| ... | ... | @@ -198,6 +227,7 @@ func printRunnerHelp() { | |
| 198 | 227 | USAGE | |
| 199 | 228 | echo '<event-json>' | pici runner [flags] | |
| 200 | 229 | pici runner --event '<event-json>' [flags] | |
| 230 | + | echo '<event-json>' | pici runner --wait [flags] # block until done | |
| 201 | 231 | ||
| 202 | 232 | EVENT JSON FIELDS | |
| 203 | 233 | type (required) Event type, e.g. "push", "merge_request" |
| ... | ... | @@ -215,7 +245,8 @@ FLAGS | |
| 215 | 245 | -event <json> Event JSON string (alternative to reading from stdin) | |
| 216 | 246 | -artifact-dir <path> Local directory to stage artifacts (default: /tmp/pici-artifacts) | |
| 217 | 247 | -log-level <level> Log level: debug, info, warn, error (default: info) | |
| 218 | - | -human Human-readable output (default: enabled for runner)`) | |
| 248 | + | -human Human-readable output (default: enabled for runner) | |
| 249 | + | -wait Block until job completes, printing session history and summary`) | |
| 219 | 250 | ||
| 220 | 251 | } | |
| 221 | 252 |
| ... | ... | @@ -354,21 +385,22 @@ func (eng *JobEngine) Run() error { | |
| 354 | 385 | log.Debug("starting runner session", "session", prefix+"runner") | |
| 355 | 386 | ||
| 356 | 387 | evStr := fmt.Sprintf("PICO_CI_EVENT=%s", eng.Ev.Type) | |
| 357 | - | jobStr := fmt.Sprintf("ZMX_SESSION_PREFIX=%s", childPrefix) | |
| 358 | 388 | ||
| 359 | 389 | runnerName := prefix + "runner" | |
| 360 | - | // Name the runner explicitly and don't set ZMX_SESSION_PREFIX for this | |
| 361 | - | // outer zmx run call. The prefix is only for child sessions spawned | |
| 362 | - | // inside pico.sh (via the exported env var). | |
| 390 | + | // Name the runner explicitly. Do NOT set ZMX_SESSION_PREFIX for this | |
| 391 | + | // outer zmx run call ā it would be prepended to the runner's own name. | |
| 392 | + | // The prefix is only for child sessions spawned inside pico.sh (via the | |
| 393 | + | // exported env var). | |
| 363 | 394 | cmdEnv := make([]string, 0, len(os.Environ())+2) | |
| 364 | 395 | for _, e := range os.Environ() { | |
| 365 | 396 | if !strings.HasPrefix(e, "ZMX_SESSION_PREFIX=") { | |
| 366 | 397 | cmdEnv = append(cmdEnv, e) | |
| 367 | 398 | } | |
| 368 | 399 | } | |
| 369 | - | cmdEnv = append(cmdEnv, evStr, jobStr) | |
| 400 | + | cmdEnv = append(cmdEnv, evStr) | |
| 370 | 401 | ||
| 371 | - | cmd := exec.Command("zmx", "run", runnerName, "-d", "bash", manifest) | |
| 402 | + | zmxPrefixStr := fmt.Sprintf("ZMX_SESSION_PREFIX=%s", childPrefix) | |
| 403 | + | cmd := exec.Command("zmx", "run", runnerName, "-d", zmxPrefixStr, "bash", manifest) | |
| 372 | 404 | cmd.Env = cmdEnv | |
| 373 | 405 | cmd.Dir = eng.Wk.GetDir() | |
| 374 | 406 |
| ... | ... | @@ -441,10 +473,177 @@ func eventHandler(cfg *Cfg, eventData *Event) error { | |
| 441 | 473 | } | |
| 442 | 474 | ||
| 443 | 475 | fmt.Fprintln(os.Stdout, "ā job launched") | |
| 476 | + | ||
| 477 | + | if cfg.Wait { | |
| 478 | + | if err := waitAndReport(cfg, log, eventData.Name, jobID); err != nil { | |
| 479 | + | return fmt.Errorf("wait: %w", err) | |
| 480 | + | } | |
| 481 | + | return nil | |
| 482 | + | } | |
| 483 | + | ||
| 444 | 484 | fmt.Fprintf(os.Stdout, " follow: zmx tail ci.%s.%s.runner\n", eventData.Name, jobID) | |
| 445 | 485 | return nil | |
| 446 | 486 | } | |
| 447 | 487 | ||
| 488 | + | // waitAndReport polls the job's sessions until all complete, prints live | |
| 489 | + | // progress to stdout, then dumps session history and a final summary. | |
| 490 | + | func waitAndReport(cfg *Cfg, log *slog.Logger, name, jobID string) error { | |
| 491 | + | prefix := "ci." + name + "." + jobID + "." | |
| 492 | + | ticker := time.NewTicker(cfg.MonitorInterval) | |
| 493 | + | defer ticker.Stop() | |
| 494 | + | ||
| 495 | + | // Handle ^C gracefully | |
| 496 | + | sigCh := make(chan os.Signal, 1) | |
| 497 | + | signal.Notify(sigCh, syscall.SIGINT) | |
| 498 | + | defer signal.Stop(sigCh) | |
| 499 | + | ||
| 500 | + | fmt.Fprintln(os.Stdout) | |
| 501 | + | fmt.Fprint(os.Stdout, "ā³ waiting for completion...\n") | |
| 502 | + | ||
| 503 | + | // Track what we've already printed to avoid duplicates | |
| 504 | + | type sessionState struct { | |
| 505 | + | status string // "running", "success", "failed" | |
| 506 | + | exitCode string | |
| 507 | + | duration string | |
| 508 | + | printed bool | |
| 509 | + | } | |
| 510 | + | known := make(map[string]*sessionState) | |
| 511 | + | var jobSessions []SessionInfo | |
| 512 | + | var sessionOrder []string // track insertion order for deterministic output | |
| 513 | + | ||
| 514 | + | for { | |
| 515 | + | select { | |
| 516 | + | case <-cfg.Ctx.Done(): | |
| 517 | + | return cfg.Ctx.Err() | |
| 518 | + | case <-sigCh: | |
| 519 | + | fmt.Fprintln(os.Stdout, "\nā¹ cancelled") | |
| 520 | + | return nil | |
| 521 | + | case <-ticker.C: | |
| 522 | + | } | |
| 523 | + | ||
| 524 | + | // Fetch current session list | |
| 525 | + | listOutput, err := exec.Command("zmx", "list").CombinedOutput() | |
| 526 | + | if err != nil { | |
| 527 | + | log.Error("zmx list", "err", err) | |
| 528 | + | continue | |
| 529 | + | } | |
| 530 | + | ||
| 531 | + | sessions := parseZMXList(string(listOutput)) | |
| 532 | + | jobSessions = nil | |
| 533 | + | for _, s := range sessions { | |
| 534 | + | if strings.HasPrefix(s.Name, prefix) { | |
| 535 | + | s.Short = cleanSessionShort(s.Name, prefix, name, jobID) | |
| 536 | + | jobSessions = append(jobSessions, s) | |
| 537 | + | } | |
| 538 | + | } | |
| 539 | + | ||
| 540 | + | if len(jobSessions) == 0 { | |
| 541 | + | continue // runner session may not have appeared yet | |
| 542 | + | } | |
| 543 | + | ||
| 544 | + | // Update state for each session | |
| 545 | + | for _, s := range jobSessions { | |
| 546 | + | state, ok := known[s.Short] | |
| 547 | + | if !ok { | |
| 548 | + | state = &sessionState{} | |
| 549 | + | known[s.Short] = state | |
| 550 | + | sessionOrder = append(sessionOrder, s.Short) | |
| 551 | + | } | |
| 552 | + | ||
| 553 | + | if s.Ended == "" { | |
| 554 | + | state.status = "running" | |
| 555 | + | } else { | |
| 556 | + | if s.ExitCode == "0" { | |
| 557 | + | state.status = "success" | |
| 558 | + | } else { | |
| 559 | + | state.status = "failed" | |
| 560 | + | state.exitCode = s.ExitCode | |
| 561 | + | } | |
| 562 | + | state.duration = fmtDuration(s.Created, s.Ended) | |
| 563 | + | } | |
| 564 | + | ||
| 565 | + | // Print when state changes | |
| 566 | + | if !state.printed { | |
| 567 | + | icon := map[string]string{"running": "š", "success": "ā ", "failed": "ā"}[state.status] | |
| 568 | + | detail := "" | |
| 569 | + | if state.status == "failed" && state.exitCode != "" { | |
| 570 | + | detail = fmt.Sprintf(", exit %s", state.exitCode) | |
| 571 | + | } | |
| 572 | + | if state.duration != "" && state.duration != "ā" { | |
| 573 | + | detail += fmt.Sprintf(" (%s)", state.duration) | |
| 574 | + | } | |
| 575 | + | fmt.Fprintf(os.Stdout, " %-12s %s %s%s\n", s.Short, icon, state.status, detail) | |
| 576 | + | state.printed = true | |
| 577 | + | } | |
| 578 | + | } | |
| 579 | + | ||
| 580 | + | // Check if all sessions are done | |
| 581 | + | if allCompleted(jobSessions) { | |
| 582 | + | break | |
| 583 | + | } | |
| 584 | + | } | |
| 585 | + | ||
| 586 | + | // Print last 25 lines of history for failed sessions only | |
| 587 | + | fmt.Fprintln(os.Stdout) | |
| 588 | + | for _, short := range sessionOrder { | |
| 589 | + | state := known[short] | |
| 590 | + | if state.status != "failed" { | |
| 591 | + | continue | |
| 592 | + | } | |
| 593 | + | ||
| 594 | + | full := prefix + short | |
| 595 | + | ||
| 596 | + | separator := strings.Repeat("\u2500", 50) | |
| 597 | + | fmt.Fprintln(os.Stdout, separator) | |
| 598 | + | fmt.Fprintf(os.Stdout, "Session: %s (exit %s)\n", short, state.exitCode) | |
| 599 | + | fmt.Fprintln(os.Stdout, separator) | |
| 600 | + | fmt.Fprintln(os.Stdout) | |
| 601 | + | ||
| 602 | + | history, err := fetchHistoryPlain(full) | |
| 603 | + | if err != nil { | |
| 604 | + | fmt.Fprintf(os.Stdout, " (history unavailable: %v)\n", err) | |
| 605 | + | } else { | |
| 606 | + | lines := strings.Split(history, "\n") | |
| 607 | + | // Show last 25 lines | |
| 608 | + | if len(lines) > 25 { | |
| 609 | + | fmt.Fprintf(os.Stdout, " ... (%d lines omitted)\n", len(lines)-25) | |
| 610 | + | lines = lines[len(lines)-25:] | |
| 611 | + | } | |
| 612 | + | for _, line := range lines { | |
| 613 | + | if line != "" { | |
| 614 | + | fmt.Fprintf(os.Stdout, " %s\n", line) | |
| 615 | + | } | |
| 616 | + | } | |
| 617 | + | } | |
| 618 | + | fmt.Fprintln(os.Stdout) | |
| 619 | + | } | |
| 620 | + | ||
| 621 | + | // Final summary | |
| 622 | + | exitCode, status := resolveJobExitCode(jobSessions) | |
| 623 | + | _, _, duration := computeJobTiming(jobSessions) | |
| 624 | + | icon := map[string]string{"success": "ā ", "failed": "ā"}[status] | |
| 625 | + | if exitCode == 0 { | |
| 626 | + | fmt.Fprintf(os.Stdout, "%s job finished: %s (%s)\n", icon, status, duration) | |
| 627 | + | } else { | |
| 628 | + | fmt.Fprintf(os.Stdout, "%s job failed: exit %d (%s)\n", icon, exitCode, duration) | |
| 629 | + | } | |
| 630 | + | ||
| 631 | + | return nil | |
| 632 | + | } | |
| 633 | + | ||
| 634 | + | // cleanSessionShort produces a readable short name from a full session name. | |
| 635 | + | // ci.name.jobID.step.lint ā lint | |
| 636 | + | // ci.name.jobID.step.ci.name.jobID.runner ā runner | |
| 637 | + | func cleanSessionShort(name, prefix, repoName, jobID string) string { | |
| 638 | + | short := strings.TrimPrefix(name, prefix) | |
| 639 | + | // Strip "step." prefix added by child sessions | |
| 640 | + | short = strings.TrimPrefix(short, "step.") | |
| 641 | + | // Strip nested full prefix (e.g. runner named ci.name.jobID.runner) | |
| 642 | + | nested := "ci." + repoName + "." + jobID + "." | |
| 643 | + | short = strings.TrimPrefix(short, nested) | |
| 644 | + | return short | |
| 645 | + | } | |
| 646 | + | ||
| 448 | 647 | // runMonitor is a long-lived daemon that polls all ci.* zmx sessions, | |
| 449 | 648 | // writes status as JSONL to stdout, stages artifacts, and syncs to destination. | |
| 450 | 649 | // Logs go to stderr. Compose with shell tools to route status: |