api.go
1package runner
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "io"
10 "maps"
11 "net/http"
12 "os"
13 "path/filepath"
14 "slices"
15 "strings"
16 "time"
17
18 "github.com/gorilla/websocket"
19
20 "git.kilimanjaro.io/ci/internal/engine"
21 "git.kilimanjaro.io/ci/internal/eventlog"
22 "git.kilimanjaro.io/ci/internal/lex"
23 "git.kilimanjaro.io/ci/internal/manifest"
24 "git.kilimanjaro.io/ci/internal/runlog"
25 "git.kilimanjaro.io/ci/internal/tid"
26)
27
28// Handler serves the runner's API. Everything but /healthz needs token
29// (when set); the appview holds it and decides who may do what.
30//
31// GET /events records stream (spindle-compatible)
32// GET /logs/{knot}/{rkey}/{name} a workflow's log, upstream LogLine messages
33// POST /xrpc/sh.tangled.ci.triggerPipeline manual pipeline
34// GET /api/runs[?repo=&limit=] runs, newest first
35// GET /api/runs/{id} run row + state.json
36// GET /api/runs/{id}/events[?after=N] events.ndjson lines after seq N
37// GET /api/runs/{id}/log.txt plain-text log
38// GET /api/runs/{id}/files/{path...} results/ and artifacts/ files
39// POST /api/runs/{id}/gate {"option":N|"action":"x"|"dismiss":true, "actor":"..."}
40// POST /api/runs/{id}/actions/{action} start a manual action run {"actor":"..."}
41// POST /api/runs/{id}/cancel stop a running segment / dismiss a gate
42// PUT /api/runs/{id}/import store a reported local run (tar.gz)
43// POST|DELETE /api/locks/{group}?holder= lease a concurrency group (local runs)
44// GET /healthz
45func (r *Runner) Handler(token string) http.Handler {
46 mux := http.NewServeMux()
47 mux.Handle("GET /events", r.Log)
48 mux.HandleFunc("GET /logs/{knot}/{rkey}/{name}", r.serveLogs)
49 mux.HandleFunc("POST /xrpc/sh.tangled.ci.triggerPipeline", r.serveTrigger)
50 mux.HandleFunc("GET /api/runs", r.serveRuns)
51 mux.HandleFunc("GET /api/runs/{id}", r.serveRun)
52 mux.HandleFunc("GET /api/runs/{id}/events", r.serveEvents)
53 mux.HandleFunc("GET /api/runs/{id}/log.txt", r.serveText)
54 mux.HandleFunc("GET /api/runs/{id}/files/{path...}", r.serveFile)
55 mux.HandleFunc("POST /api/runs/{id}/gate", r.serveGate)
56 mux.HandleFunc("POST /api/runs/{id}/actions/{action}", r.serveAction)
57 mux.HandleFunc("POST /api/runs/{id}/cancel", r.serveCancel)
58 mux.HandleFunc("PUT /api/runs/{id}/import", r.serveImport)
59 mux.HandleFunc("POST /api/locks/{group}", r.serveLock)
60 mux.HandleFunc("DELETE /api/locks/{group}", r.serveLock)
61 authed := eventlog.RequireToken(token, mux)
62 top := http.NewServeMux()
63 top.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) {
64 if _, err := r.Log.List(0, 1); err != nil {
65 http.Error(w, err.Error(), http.StatusServiceUnavailable)
66 return
67 }
68 fmt.Fprintln(w, "ok")
69 })
70 top.Handle("/", authed)
71 return top
72}
73
74func writeJSON(w http.ResponseWriter, code int, v any) {
75 w.Header().Set("Content-Type", "application/json")
76 w.WriteHeader(code)
77 _ = json.NewEncoder(w).Encode(v)
78}
79
80func fail(w http.ResponseWriter, code int, err error) {
81 writeJSON(w, code, map[string]string{"error": err.Error()})
82}
83
84// RunView is a run as the API returns it.
85type RunView struct {
86 Run
87 State *engine.State `json:"state,omitempty"`
88 // Manual are the workflow's manual actions (buttons on the run page).
89 Manual []ManualAction `json:"manual,omitempty"`
90 // Danger lists every action with style danger (gate options too).
91 Danger []string `json:"danger,omitempty"`
92}
93
94// ManualAction is a manual action of a run's workflow.
95type ManualAction struct {
96 Name string `json:"name"`
97 Label string `json:"label"`
98 Style string `json:"style,omitempty"`
99 Gated bool `json:"gated,omitempty"`
100}
101
102func (r *Runner) view(x Run) RunView {
103 v := RunView{Run: x}
104 if st, err := engine.Load(engine.Dir(x.Dir)); err == nil {
105 v.State = st
106 if raw, err := os.ReadFile(engine.Dir(x.Dir).ManifestFile()); err == nil {
107 if m, err := manifest.Parse(st.Meta.Workflow, raw); err == nil {
108 for _, name := range slices.Sorted(maps.Keys(m.Actions)) {
109 if m.Actions[name].Style == "danger" {
110 v.Danger = append(v.Danger, name)
111 }
112 if a := m.Actions[name]; a.Manual {
113 v.Manual = append(v.Manual, ManualAction{Name: name, Label: m.Label(name), Style: a.Style, Gated: a.Gate != nil})
114 }
115 }
116 }
117 }
118 }
119 return v
120}
121
122func (r *Runner) serveRuns(w http.ResponseWriter, req *http.Request) {
123 limit := 50
124 if n, err := fmt.Sscan(req.URL.Query().Get("limit"), &limit); n != 1 || err != nil || limit <= 0 || limit > 500 {
125 limit = 50
126 }
127 runs, err := r.ListRuns(req.URL.Query().Get("repo"), limit)
128 if err != nil {
129 fail(w, 500, err)
130 return
131 }
132 out := make([]RunView, 0, len(runs))
133 for _, x := range runs {
134 out = append(out, r.view(x))
135 }
136 writeJSON(w, 200, out)
137}
138
139func (r *Runner) run(w http.ResponseWriter, req *http.Request) (Run, bool) {
140 x, err := r.GetRun(req.PathValue("id"))
141 if err != nil {
142 fail(w, 404, fmt.Errorf("no run %s", req.PathValue("id")))
143 return Run{}, false
144 }
145 return x, true
146}
147
148func (r *Runner) serveRun(w http.ResponseWriter, req *http.Request) {
149 if x, ok := r.run(w, req); ok {
150 writeJSON(w, 200, r.view(x))
151 }
152}
153
154func (r *Runner) serveEvents(w http.ResponseWriter, req *http.Request) {
155 x, ok := r.run(w, req)
156 if !ok {
157 return
158 }
159 after := 0
160 _, _ = fmt.Sscan(req.URL.Query().Get("after"), &after)
161 f, err := os.Open(engine.Dir(x.Dir).Events())
162 if err != nil {
163 fail(w, 500, err)
164 return
165 }
166 defer f.Close()
167 w.Header().Set("Content-Type", "application/x-ndjson")
168 _ = runlog.Scan(f, func(ev lex.RunEvent) error {
169 if ev.Seq > after {
170 b, _ := json.Marshal(ev)
171 _, err := w.Write(append(b, '\n'))
172 return err
173 }
174 return nil
175 })
176}
177
178// serveText renders the log as plain text: what an agent reads (ci wait).
179func (r *Runner) serveText(w http.ResponseWriter, req *http.Request) {
180 x, ok := r.run(w, req)
181 if !ok {
182 return
183 }
184 f, err := os.Open(engine.Dir(x.Dir).Events())
185 if err != nil {
186 fail(w, 500, err)
187 return
188 }
189 defer f.Close()
190 w.Header().Set("Content-Type", "text/plain; charset=utf-8")
191 _ = runlog.Scan(f, func(ev lex.RunEvent) error {
192 _, err := io.WriteString(w, TextLine(ev))
193 return err
194 })
195}
196
197// TextLine renders an event as plain text ("" for events not worth a line).
198func TextLine(ev lex.RunEvent) string {
199 switch ev.Kind {
200 case lex.KindStart:
201 return fmt.Sprintf("== %s (#%d): %s\n", ev.Action, ev.Attempt, firstLine(ev.Command))
202 case lex.KindLog:
203 return ev.Content + "\n"
204 case lex.KindEnd:
205 code := 0
206 if ev.ExitCode != nil {
207 code = *ev.ExitCode
208 }
209 return fmt.Sprintf("== %s: %s (exit %d)\n", ev.Action, ev.Status, code)
210 case lex.KindResult:
211 if ev.Result != nil && ev.Result.Summary != "" {
212 return " " + strings.ReplaceAll(strings.TrimRight(ev.Result.Summary, "\n"), "\n", "\n ") + "\n"
213 }
214 case lex.KindGate:
215 if g := ev.Gate; g != nil {
216 return fmt.Sprintf("== gate %s: %s\n", g.State, orDefault(g.Prompt, g.Choice))
217 }
218 case lex.KindRun:
219 if ev.Status != engine.StatusRunning {
220 return fmt.Sprintf("== run %s %s\n", ev.Status, ev.Content)
221 }
222 }
223 return ""
224}
225
226func firstLine(s string) string {
227 s = strings.TrimSpace(s)
228 if i := strings.IndexByte(s, '\n'); i >= 0 {
229 return s[:i] + " …"
230 }
231 return s
232}
233
234func (r *Runner) serveFile(w http.ResponseWriter, req *http.Request) {
235 x, ok := r.run(w, req)
236 if !ok {
237 return
238 }
239 p := filepath.Clean(req.PathValue("path"))
240 if (!strings.HasPrefix(p, "results/") && !strings.HasPrefix(p, "artifacts/")) || strings.Contains(p, "..") {
241 fail(w, 404, errors.New("not found"))
242 return
243 }
244 full := filepath.Join(x.Dir, p)
245 real, err := filepath.EvalSymlinks(full)
246 if err != nil || !strings.HasPrefix(real, filepath.Join(x.Dir, filepath.Dir(p))) {
247 fail(w, 404, errors.New("not found"))
248 return
249 }
250 w.Header().Set("Content-Type", "text/plain; charset=utf-8")
251 http.ServeFile(w, req, real)
252}
253
254type gateAnswer struct {
255 Option int `json:"option"` // 1-based
256 Action string `json:"action"`
257 Dismiss bool `json:"dismiss"`
258 Actor string `json:"actor"`
259}
260
261func (r *Runner) serveGate(w http.ResponseWriter, req *http.Request) {
262 x, ok := r.run(w, req)
263 if !ok {
264 return
265 }
266 var a gateAnswer
267 if err := json.NewDecoder(io.LimitReader(req.Body, 1<<16)).Decode(&a); err != nil {
268 fail(w, 400, err)
269 return
270 }
271 if err := r.AnswerGate(req.Context(), x, a); err != nil {
272 fail(w, 409, err)
273 return
274 }
275 writeJSON(w, 202, map[string]string{"run": x.ID})
276}
277
278// AnswerGate starts the segment that answers a paused run's gate.
279func (r *Runner) AnswerGate(ctx context.Context, x Run, a gateAnswer) error {
280 st, err := engine.Load(engine.Dir(x.Dir))
281 if err != nil {
282 return err
283 }
284 if x.Status != engine.StatusPaused || st.Gate == nil {
285 return fmt.Errorf("run %s is %s, not waiting at a gate", x.ID, x.Status)
286 }
287 seg := Segment{Actor: a.Actor, Dismiss: a.Dismiss}
288 if !a.Dismiss {
289 switch {
290 case a.Option >= 1 && a.Option <= len(st.Gate.Options):
291 seg.Choose = a.Option
292 case a.Action != "":
293 for i, o := range st.Gate.Options {
294 if o.Action == a.Action {
295 seg.Choose = i + 1
296 break
297 }
298 }
299 }
300 if seg.Choose == 0 {
301 return errors.New("no such option")
302 }
303 }
304 if err := r.updateRun(x.ID, `status = ?`, engine.StatusRunning); err != nil {
305 return err
306 }
307 return r.launch(context.WithoutCancel(ctx), x.ID, seg)
308}
309
310func (r *Runner) serveAction(w http.ResponseWriter, req *http.Request) {
311 x, ok := r.run(w, req)
312 if !ok {
313 return
314 }
315 var body struct {
316 Actor string `json:"actor"`
317 }
318 _ = json.NewDecoder(io.LimitReader(req.Body, 1<<16)).Decode(&body)
319 id, err := r.ManualAction(req.Context(), x, req.PathValue("action"), body.Actor)
320 if err != nil {
321 fail(w, 409, err)
322 return
323 }
324 writeJSON(w, 202, map[string]string{"run": id})
325}
326
327// ManualAction starts a child run of x at a manual action (DESIGN.md §9):
328// it shares x's checkouts and $CI_STATE and uses the manifest x ran with.
329func (r *Runner) ManualAction(ctx context.Context, x Run, action, actor string) (string, error) {
330 parent := x
331 if x.Parent != "" {
332 if p, err := r.GetRun(x.Parent); err == nil {
333 parent = p
334 }
335 }
336 pst, err := engine.Load(engine.Dir(parent.Dir))
337 if err != nil {
338 return "", err
339 }
340 raw, err := os.ReadFile(engine.Dir(parent.Dir).ManifestFile())
341 if err != nil {
342 return "", err
343 }
344 m, err := manifest.Parse(pst.Meta.Workflow, raw)
345 if err != nil {
346 return "", err
347 }
348 a, ok := m.Actions[action]
349 if !ok || !a.Manual {
350 return "", fmt.Errorf("%s is not a manual action of %s", action, m.Name)
351 }
352 id := parent.ID + "-" + action + "-" + tid.New()
353 meta := pst.Meta
354 meta.Run, meta.Parent = id, parent.Dir
355 dir := engine.Dir(filepath.Join(r.Cfg.RunsDir, id))
356 if meta.Host == "local" { // a reported local run: check its commits out here
357 if err := r.relayout(ctx, &meta, m.Repos, dir); err != nil {
358 _ = os.RemoveAll(dir.Path())
359 return "", err
360 }
361 }
362 e, err := engine.Create(m, dir, meta, nil, engine.Options{})
363 if err != nil {
364 return "", err
365 }
366 e.Log.Close()
367 t := now()
368 if err := r.insertRun(Run{ID: id, Pipeline: parent.Pipeline, Rkey: parent.Rkey, Workflow: parent.Workflow, Repo: parent.Repo,
369 Dir: dir.Path(), Parent: parent.ID, Status: lex.StatusPending, Created: t, Updated: t}); err != nil {
370 return "", err
371 }
372 return id, r.launch(context.WithoutCancel(ctx), id, Segment{From: action, Actor: actor})
373}
374
375func (r *Runner) serveCancel(w http.ResponseWriter, req *http.Request) {
376 x, ok := r.run(w, req)
377 if !ok {
378 return
379 }
380 switch x.Status {
381 case engine.StatusPaused:
382 if err := r.AnswerGate(req.Context(), x, gateAnswer{Dismiss: true, Actor: "cancel"}); err != nil {
383 fail(w, 409, err)
384 return
385 }
386 case engine.StatusRunning, lex.StatusPending:
387 if x.handle == "" || r.Launcher.Stop(x.handle) != nil {
388 fail(w, 409, errors.New("nothing to stop"))
389 return
390 }
391 default:
392 fail(w, 409, fmt.Errorf("run %s is %s", x.ID, x.Status))
393 return
394 }
395 writeJSON(w, 202, map[string]string{"run": x.ID})
396}
397
398func (r *Runner) serveTrigger(w http.ResponseWriter, req *http.Request) {
399 var in lex.TriggerPipelineInput
400 if err := json.NewDecoder(io.LimitReader(req.Body, 1<<20)).Decode(&in); err != nil || in.Trigger == nil {
401 fail(w, 400, fmt.Errorf("InvalidRequest: %v", err))
402 return
403 }
404 ref := ""
405 if in.Trigger.Ref != nil {
406 ref = *in.Trigger.Ref
407 }
408 uri, err := r.StartPipeline(req.Context(), Trigger{Kind: "manual", RepoDid: in.Repo, Sha: in.Trigger.Sha, Ref: ref,
409 Inputs: in.Trigger.Inputs, Workflows: in.Workflows})
410 if err != nil {
411 fail(w, 400, err)
412 return
413 }
414 if uri == "" {
415 fail(w, 400, errors.New("no matching workflows"))
416 return
417 }
418 writeJSON(w, 200, lex.TriggerPipelineOutput{Pipeline: uri})
419}
420
421var upgrader = websocket.Upgrader{ReadBufferSize: 1024, WriteBufferSize: 1024}
422
423// serveLogs streams a workflow's log as upstream's /logs does: LogLine
424// messages (control start/end per action execution, data per line), replay
425// then tail, closed when the run ends.
426func (r *Runner) serveLogs(w http.ResponseWriter, req *http.Request) {
427 x, err := r.RunByWorkflow(req.PathValue("rkey"), req.PathValue("name"))
428 if err != nil {
429 http.Error(w, "no such workflow", http.StatusNotFound)
430 return
431 }
432 conn, err := upgrader.Upgrade(w, req, nil)
433 if err != nil {
434 return
435 }
436 defer conn.Close()
437 ctx, cancel := context.WithCancel(req.Context())
438 defer cancel()
439 go func() {
440 for {
441 if _, _, err := conn.NextReader(); err != nil {
442 cancel()
443 return
444 }
445 }
446 }()
447 dir := engine.Dir(x.Dir)
448 var off int64
449 lastPing := time.Now()
450 for {
451 n, err := sendLogLines(conn, dir.Events(), off)
452 if err != nil {
453 return
454 }
455 off += n
456 if st, err := engine.Load(dir); err == nil && engine.Terminal(st.Status) && n == 0 {
457 _ = conn.WriteControl(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "log stream complete"), time.Now().Add(time.Second))
458 return
459 }
460 if time.Since(lastPing) > 30*time.Second {
461 if conn.WriteControl(websocket.PingMessage, nil, time.Now().Add(time.Second)) != nil {
462 return
463 }
464 lastPing = time.Now()
465 }
466 select {
467 case <-ctx.Done():
468 return
469 case <-time.After(300 * time.Millisecond):
470 }
471 }
472}
473
474func sendLogLines(conn *websocket.Conn, path string, off int64) (int64, error) {
475 f, err := os.Open(path)
476 if err != nil {
477 return 0, err
478 }
479 defer f.Close()
480 if _, err := f.Seek(off, io.SeekStart); err != nil {
481 return 0, err
482 }
483 b, err := io.ReadAll(f)
484 if err != nil {
485 return 0, err
486 }
487 end := bytes.LastIndexByte(b, '\n')
488 if end < 0 {
489 return 0, nil
490 }
491 err = runlog.Scan(bytes.NewReader(b[:end+1]), func(ev lex.RunEvent) error {
492 if l, ok := LogLineOf(ev); ok {
493 msg, _ := json.Marshal(l)
494 return conn.WriteMessage(websocket.TextMessage, msg)
495 }
496 return nil
497 })
498 return int64(end + 1), err
499}
500
501// LogLineOf translates a run event into upstream's log message, if it is one.
502func LogLineOf(ev lex.RunEvent) (lex.LogLine, bool) {
503 switch ev.Kind {
504 case lex.KindStart:
505 return lex.LogLine{Kind: "control", Content: ev.Action, Time: ev.Time, StepId: ev.Attempt,
506 StepStatus: "start", StepKind: "user", StepCommand: ev.Command}, true
507 case lex.KindEnd:
508 return lex.LogLine{Kind: "control", Content: ev.Action, Time: ev.Time, StepId: ev.Attempt,
509 StepStatus: "end", StepKind: "user"}, true
510 case lex.KindLog:
511 if ev.Attempt == 0 {
512 return lex.LogLine{}, false // engine messages outside an action
513 }
514 return lex.LogLine{Kind: "data", Content: ev.Content, Time: ev.Time, StepId: ev.Attempt, Stream: ev.Stream}, true
515 }
516 return lex.LogLine{}, false
517}
