follow.go
1package runner
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "encoding/json"
8 "fmt"
9 "io"
10 "os"
11 "os/exec"
12 "time"
13
14 "git.kilimanjaro.io/ci/internal/engine"
15 "git.kilimanjaro.io/ci/internal/lex"
16)
17
18// follow tails a run's events while its segment runs, publishing records,
19// then settles the run once the segment's process is gone.
20func (r *Runner) follow(ctx context.Context, run, handle string) {
21 r.mu.Lock()
22 if r.tailing[run] {
23 r.mu.Unlock()
24 return
25 }
26 r.tailing[run] = true
27 r.mu.Unlock()
28 defer func() {
29 r.mu.Lock()
30 delete(r.tailing, run)
31 r.mu.Unlock()
32 }()
33
34 x, err := r.GetRun(run)
35 if err != nil {
36 r.Logger.Error("follow", "run", run, "err", err)
37 return
38 }
39 for {
40 alive := handle != "" && r.Launcher.Alive(handle)
41 if err := r.translate(&x); err != nil {
42 r.Logger.Error("translate events", "run", run, "err", err)
43 }
44 if !alive {
45 break
46 }
47 select {
48 case <-ctx.Done():
49 return
50 case <-time.After(250 * time.Millisecond):
51 }
52 }
53 r.settle(x)
54}
55
56// translate publishes the run's new events (from its stored offset).
57func (r *Runner) translate(x *Run) error {
58 f, err := os.Open(engine.Dir(x.Dir).Events())
59 if err != nil {
60 return err
61 }
62 defer f.Close()
63 if _, err := f.Seek(x.offset, io.SeekStart); err != nil {
64 return err
65 }
66 b, err := io.ReadAll(f)
67 if err != nil {
68 return err
69 }
70 end := bytes.LastIndexByte(b, '\n')
71 if end < 0 {
72 return nil
73 }
74 sc := bufio.NewScanner(bytes.NewReader(b[:end+1]))
75 sc.Buffer(make([]byte, 64*1024), 16*1024*1024)
76 for sc.Scan() {
77 var ev lex.RunEvent
78 if err := json.Unmarshal(sc.Bytes(), &ev); err != nil {
79 continue
80 }
81 switch {
82 case ev.Kind == lex.KindGate && ev.Gate != nil:
83 if _, err := r.Log.Append(lex.GateNSID, "", ev.Gate); err != nil {
84 return err
85 }
86 case ev.Kind == lex.KindTransition && ev.Transition != nil:
87 if _, err := r.Log.Append(lex.TransitionNSID, "", ev.Transition); err != nil {
88 return err
89 }
90 }
91 }
92 x.offset += int64(end + 1)
93 return r.updateRun(x.ID, `offset = ?`, x.offset)
94}
95
96// settle records how a segment left its run: paused at a gate, finished, or
97// lost (the process died without saying).
98func (r *Runner) settle(x Run) {
99 defer func() { <-r.slots }()
100 st, err := engine.Load(engine.Dir(x.Dir))
101 if err != nil {
102 r.Logger.Error("settle", "run", x.ID, "err", err)
103 return
104 }
105 if st.Status == engine.StatusRunning {
106 if err := engine.MarkLost(engine.Dir(x.Dir), "the run's process ended without finishing it (see segment logs)"); err != nil {
107 r.Logger.Error("mark lost", "run", x.ID, "err", err)
108 }
109 if err := r.translate(&x); err != nil {
110 r.Logger.Error("translate", "run", x.ID, "err", err)
111 }
112 if st, err = engine.Load(engine.Dir(x.Dir)); err != nil {
113 return
114 }
115 }
116 if err := r.updateRun(x.ID, `status = ?`, st.Status); err != nil {
117 r.Logger.Error("update run", "run", x.ID, "err", err)
118 }
119 r.Logger.Info("segment ended", "run", x.ID, "status", st.Status)
120 switch {
121 case st.Status == engine.StatusPaused && st.Gate != nil:
122 r.notify(Notification{Kind: "gate", Run: x, Status: st.Status,
123 Title: fmt.Sprintf("%s/%s: waiting for you", x.Repo, x.Workflow),
124 Message: orDefault(st.Gate.Prompt, "Choose the next step") + summaryLine(st.Gate.Summary)})
125 case engine.Terminal(st.Status):
126 if x.Parent == "" {
127 status, msg := tangledStatus(st)
128 r.status(x.Pipeline, x.Workflow, status, msg)
129 }
130 detail := ""
131 if st.End != nil {
132 detail = st.End.Detail
133 }
134 r.notify(Notification{Kind: "finished", Run: x, Status: st.Status,
135 Title: fmt.Sprintf("%s/%s: %s", x.Repo, x.Workflow, st.Status), Message: detail})
136 }
137}
138
139func summaryLine(s string) string {
140 if s == "" {
141 return ""
142 }
143 line, _, _ := bytes.Cut([]byte(s), []byte("\n"))
144 return "\n" + string(line)
145}
146
147// tangledStatus maps a finished run onto sh.tangled.pipeline.status.
148func tangledStatus(st *engine.State) (string, string) {
149 detail := ""
150 if st.End != nil {
151 detail = st.End.Detail
152 }
153 switch st.Status {
154 case engine.StatusSuccess:
155 return lex.StatusSuccess, ""
156 case engine.StatusCancelled:
157 if st.End != nil && st.End.Reason == lex.ReasonExpired {
158 return lex.StatusTimeout, detail
159 }
160 return lex.StatusCancelled, detail
161 }
162 return lex.StatusFailed, detail
163}
164
165func (r *Runner) notify(n Notification) {
166 if r.Cfg.Notify != nil {
167 r.Cfg.Notify(n)
168 }
169}
170
171// Reconcile picks up runs after a restart: it follows live segments again,
172// settles finished ones, starts runs that never started, and expires gates.
173func (r *Runner) Reconcile(ctx context.Context) error {
174 rows, err := r.db().Query(`select ` + runCols + ` from runs where status in ('pending', 'running')`)
175 if err != nil {
176 return err
177 }
178 var runs []Run
179 for rows.Next() {
180 x, err := scanRun(rows)
181 if err != nil {
182 rows.Close()
183 return err
184 }
185 runs = append(runs, x)
186 }
187 rows.Close()
188 for _, x := range runs {
189 if x.handle == "" {
190 if err := r.launch(ctx, x.ID, Segment{}); err != nil {
191 r.Logger.Error("start pending run", "run", x.ID, "err", err)
192 }
193 continue
194 }
195 select {
196 case r.slots <- struct{}{}: // follow releases it in settle
197 case <-ctx.Done():
198 return ctx.Err()
199 }
200 go r.follow(context.WithoutCancel(ctx), x.ID, x.handle)
201 }
202 return nil
203}
204
205// ExpireGates answers expired gates of paused runs (on a timer).
206func (r *Runner) ExpireGates(ctx context.Context) {
207 rows, err := r.db().Query(`select ` + runCols + ` from runs where status = 'paused'`)
208 if err != nil {
209 return
210 }
211 var due []string
212 for rows.Next() {
213 x, err := scanRun(rows)
214 if err != nil {
215 break
216 }
217 if st, err := engine.Load(engine.Dir(x.Dir)); err == nil && st.Gate != nil && st.Gate.Expired(time.Now()) {
218 due = append(due, x.ID)
219 }
220 }
221 rows.Close()
222 for _, id := range due {
223 if err := r.launch(ctx, id, Segment{Expire: true, Actor: "runner"}); err != nil {
224 r.Logger.Error("expire gate", "run", id, "err", err)
225 }
226 }
227}
228
229// Loop runs the gate expiry timer and the retention of finished runs until ctx ends.
230func (r *Runner) Loop(ctx context.Context) {
231 go r.pruneLoop(ctx)
232 for {
233 select {
234 case <-ctx.Done():
235 return
236 case <-time.After(30 * time.Second):
237 r.ExpireGates(ctx)
238 }
239 }
240}
241
242func gitOut(ctx context.Context, gitDir string, args ...string) (string, error) {
243 cmd := exec.CommandContext(ctx, "git", append([]string{"--git-dir", gitDir}, args...)...)
244 var stderr bytes.Buffer
245 cmd.Stderr = &stderr
246 out, err := cmd.Output()
247 if err != nil {
248 return "", fmt.Errorf("git %v: %v: %s", args, err, bytes.TrimSpace(stderr.Bytes()))
249 }
250 return string(out), nil
251}
