gate.go
1package engine
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "time"
9
10 "git.kilimanjaro.io/ci/internal/lex"
11 "git.kilimanjaro.io/ci/internal/manifest"
12 "git.kilimanjaro.io/ci/internal/runlog"
13)
14
15// ErrPause tells the engine to leave a gate for later: the run is saved as
16// paused and the process can exit.
17var ErrPause = errors.New("paused at gate")
18
19// Choice answers a gate.
20type Choice struct {
21 Option int // index into the gate's options, when not Dismiss/Expire
22 Dismiss bool // end the run here
23 Expire bool // the gate timed out
24 Actor string
25}
26
27// Gatekeeper decides gates as they are reached. Returning ErrPause (or
28// having no Gatekeeper) pauses the run; it continues with Resume.
29type Gatekeeper interface {
30 Decide(ctx context.Context, e *Engine, g *PendingGate) (Choice, error)
31}
32
33// GateFunc adapts a function to Gatekeeper.
34type GateFunc func(ctx context.Context, e *Engine, g *PendingGate) (Choice, error)
35
36func (f GateFunc) Decide(ctx context.Context, e *Engine, g *PendingGate) (Choice, error) {
37 return f(ctx, e, g)
38}
39
40// reachGate records g, supersedes older runs waiting at the same
41// concurrency group, and asks the Gatekeeper. It returns the next decision,
42// or paused=true when the run should stop here.
43func (e *Engine) reachGate(ctx context.Context, g *PendingGate) (Decision, bool, error) {
44 e.st.Gate = g
45 e.st.Status = StatusPaused
46 e.emit(lex.RunEvent{Kind: lex.KindGate, Gate: e.gateRecord(g, lex.GateWaiting, Choice{})})
47 if err := e.st.Save(e.Dir); err != nil {
48 return Decision{}, false, err
49 }
50 e.supersedeOlder(g)
51
52 var c Choice
53 err := ErrPause
54 if e.Opts.Gates != nil && !e.Opts.Detach {
55 c, err = e.Opts.Gates.Decide(ctx, e, g)
56 }
57 if errors.Is(err, ErrPause) {
58 e.releaseLocks()
59 e.emit(lex.RunEvent{Kind: lex.KindRun, Status: StatusPaused, Content: "waiting: " + gateTitle(g)})
60 return Decision{}, true, e.st.Save(e.Dir)
61 }
62 if err != nil {
63 return Decision{}, false, err
64 }
65 d, err := e.answer(g, c)
66 return d, false, err
67}
68
69// answer records a decision on g and returns what follows.
70func (e *Engine) answer(g *PendingGate, c Choice) (Decision, error) {
71 var d Decision
72 state := lex.GateApproved
73 switch {
74 case c.Expire:
75 state, d = lex.GateExpired, e.router.Expire(g)
76 case c.Dismiss:
77 state, d = lex.GateDismissed, e.router.Dismiss(g)
78 default:
79 var err error
80 if d, err = e.router.Choose(g, c.Option); err != nil {
81 return Decision{}, err
82 }
83 }
84 e.emit(lex.RunEvent{Kind: lex.KindGate, Gate: e.gateRecord(g, state, c)})
85 e.st.Gate = nil
86 e.st.Status = StatusRunning
87 return d, e.st.Save(e.Dir)
88}
89
90func (e *Engine) gateRecord(g *PendingGate, state string, c Choice) *lex.GateRecord {
91 r := &lex.GateRecord{
92 Pipeline: e.st.Meta.Pipeline, Run: e.st.Meta.Run, Workflow: e.st.Meta.Workflow, Seq: e.st.Seq,
93 From: g.From, Prompt: g.Prompt, Summary: g.Summary, Options: g.Options, State: state,
94 ExpiresAt: g.ExpiresAt, CreatedAt: g.CreatedAt, Actor: c.Actor,
95 }
96 if state != lex.GateWaiting {
97 r.DecidedAt = now()
98 }
99 if state == lex.GateApproved && c.Option < len(g.Options) {
100 r.Choice = g.Options[c.Option].Action
101 }
102 return r
103}
104
105func gateTitle(g *PendingGate) string {
106 if g.Prompt != "" {
107 return g.Prompt
108 }
109 return "choose: " + optionNames(g)
110}
111
112// Expired reports whether the gate's deadline has passed.
113func (g *PendingGate) Expired(at time.Time) bool {
114 t, err := time.Parse(time.RFC3339Nano, g.ExpiresAt)
115 return err == nil && at.After(t)
116}
117
118// Open reopens an existing run (to resume it).
119func Open(dir Dir, ex Executor, opts Options) (*Engine, error) {
120 st, err := Load(dir)
121 if err != nil {
122 return nil, err
123 }
124 raw, err := os.ReadFile(dir.ManifestFile())
125 if err != nil {
126 return nil, err
127 }
128 m, err := manifest.Parse(st.Meta.Workflow, raw)
129 if err != nil {
130 return nil, fmt.Errorf("%s: %w", dir.ManifestFile(), err)
131 }
132 log, err := runlog.Open(dir.Events(), st.Meta.Run)
133 if err != nil {
134 return nil, err
135 }
136 return &Engine{
137 M: m, Dir: dir, Exec: ex, Log: log, BaseEnv: osEnviron(), Opts: opts,
138 router: &Router{M: m, Detached: st.Meta.Parent != ""},
139 st: st,
140 }, nil
141}
142
143// Resume answers the gate a paused run is waiting at and continues it. A
144// gate past its deadline expires whatever the choice.
145func (e *Engine) Resume(ctx context.Context, c Choice) (*State, error) {
146 g := e.st.Gate
147 if e.st.Status != StatusPaused || g == nil {
148 return nil, fmt.Errorf("run %s is %s, not waiting at a gate", e.st.Meta.Run, e.st.Status)
149 }
150 if !c.Expire && g.Expired(time.Now()) {
151 c = Choice{Expire: true, Actor: c.Actor}
152 }
153 if c.Expire && !g.Expired(time.Now()) {
154 return nil, fmt.Errorf("the gate expires at %s, not yet", g.ExpiresAt)
155 }
156 if !c.Dismiss && !c.Expire && (c.Option < 0 || c.Option >= len(g.Options)) {
157 return nil, fmt.Errorf("no option %d", c.Option+1)
158 }
159 e.emit(lex.RunEvent{Kind: lex.KindRun, Status: StatusRunning, Content: "resumed"})
160 d, err := e.answer(g, c)
161 if err != nil {
162 return nil, err
163 }
164 return e.drive(ctx, d)
165}
166
167// MarkLost ends a run whose process died without finishing it (the runner
168// found its unit gone): it is recorded as a failure with detail.
169func MarkLost(dir Dir, detail string) error {
170 e, err := Open(dir, nil, Options{})
171 if err != nil {
172 return err
173 }
174 defer e.Log.Close()
175 if Terminal(e.st.Status) || e.st.Status == StatusPaused {
176 return nil
177 }
178 _, err = e.finish(End{StatusFailure, lex.ReasonError, detail}, e.lastAction())
179 return err
180}
