1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "crypto/rand"
8 "encoding/base64"
9 "errors"
10 "fmt"
11 "net/http"
12 "runtime"
13 "sync"
14 "sync/atomic"
15 "time"
16
17 "connectrpc.com/connect"
18 "github.com/bnema/zerowrap"
19
20 ermite "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22
23 "bnema.dev/ermite-runner/internal/actionscache"
24 "bnema.dev/ermite-runner/internal/cachesvc"
25 "bnema.dev/ermite-runner/internal/execstream/execpb"
26 "bnema.dev/ermite-runner/internal/resultssvc"
27 "bnema.dev/ermite-runner/internal/sandbox"
28)
29
30// Runner fetches jobs from ermite and runs each one in its own sandbox.
31type Runner struct {
32 Cfg Config
33 Driver sandbox.Driver
34 // Poll is the client that carries the runner token (FetchJob).
35 Poll ermiteconnect.RunnerServiceClient
36 // Job builds the client of a job from its job token.
37 Job func(jobToken string) ermiteconnect.RunnerServiceClient
38 Timing Timing
39 // Features lists the features of the server (GetVersion). Annotations and
40 // step summaries are sent only when it says "ci.annotations" and
41 // "ci.summaries". Nil means neither is sent.
42 Features ServerFeatures
43 // Cache, when not nil, is the cache service of the runner; the sandbox
44 // drivers serve it on a channel of each sandbox (see sandbox.Config.Cache),
45 // which the guest reaches on its loopback (cachesvc.DefaultPort), and each
46 // job gets a token scoped to its repository and refs.
47 Cache *cachesvc.Service
48 // Archives, when not nil (and Cache is), is the host's cache of action
49 // archives: the host downloads them and the sandbox reads them from the cache
50 // channel (see cacheArchives). ArchiveHTTP is its HTTP client (tests); nil
51 // means the default transport.
52 Archives *actionscache.Cache
53 ArchiveHTTP *http.Client
54 // Results, when not nil, is the results service of the runner (artifacts
55 // v4). A job gets it when Cache is set too (it is served on the cache
56 // channel and checks the cache's runtime tokens), Artifacts is set and the
57 // server has the feature ci.artifacts.
58 Results *resultssvc.Service
59 // Artifacts builds the JobArtifactService client of a job from its job
60 // token.
61 Artifacts func(jobToken string) ermiteconnect.JobArtifactServiceClient
62
63 // now returns the current time; tests replace it.
64 now func() time.Time
65
66 leaked atomic.Bool
67 active sync.WaitGroup
68}
69
70func (r *Runner) clock() time.Time {
71 if r.now != nil {
72 return r.now()
73 }
74 return time.Now()
75}
76
77// spec is the sandbox request of a job. It names the job and leaves CPUs,
78// memory and disk at zero: the driver fills them from its configuration
79// (sandbox.Config.Resolve in Start), that is from ERMITE_RUNNER_JOB_CPUS,
80// _JOB_MEMORY and _JOB_DISK, so that the limits are one setting of the runner
81// and not something a job or this package can change.
82//
83// results is the job's results service, nil without one.
84func (r *Runner) spec(j *ermite.Job, results http.Handler) sandbox.JobSpec {
85 return sandbox.JobSpec{JobID: j.JobId, Results: results}
86}
87
88// markLeaked records that a sandbox could not be cleaned up: no new job is
89// taken until the running ones are over and Driver.Reap has succeeded.
90func (r *Runner) markLeaked() { r.leaked.Store(true) }
91
92// ErrRemoved is returned by Run when ermite no longer knows the runner token:
93// the runner was removed or replaced.
94var ErrRemoved = errors.New("ermite rejected the runner token: the runner was removed or replaced; register it again")
95
96// newPollID is 16 random bytes in base64url: 22 characters of the set the
97// protocol allows.
98func newPollID() string {
99 var b [16]byte
100 _, _ = rand.Read(b[:])
101 return base64.RawURLEncoding.EncodeToString(b[:])
102}
103
104// Run is the main loop. It returns nil after a graceful shutdown (ctx ended and
105// every running job reported), or an error when the runner cannot go on.
106//
107// Capacity: a poll is made only when a slot is free. A lost response is retried
108// with the same poll_id, so that the server hands back the same assignment.
109func (r *Runner) Run(ctx context.Context) error {
110 log := zerowrap.FromCtx(ctx)
111 t := r.Timing.withDefaults()
112 shutdown := make(chan struct{})
113 slots := make(chan struct{}, max(1, r.Cfg.Capacity))
114 defer func() {
115 // Graceful shutdown: running jobs are cancelled with runner_shutdown and
116 // get ShutdownGrace to stop; Run returns once they have reported.
117 close(shutdown)
118 r.active.Wait()
119 }()
120
121 if err := r.Driver.Reap(ctx); err != nil && !errors.Is(err, sandbox.ErrBusy) {
122 log.Error().Err(err).Msg("reaping what a previous run left")
123 }
124
125 var pollID string
126 bo := backoff{min: t.BackoffMin, max: t.BackoffMax}
127 for ctx.Err() == nil {
128 // Wait for a free slot, and for the end of a leak.
129 select {
130 case slots <- struct{}{}:
131 case <-ctx.Done():
132 return nil
133 }
134 if r.leaked.Load() {
135 <-slots
136 if !r.recover(ctx, log, slots, t.ReapRetry) {
137 return nil
138 }
139 continue
140 }
141 if pollID == "" {
142 pollID = newPollID()
143 }
144 pctx, cancel := context.WithTimeout(ctx, t.PollTimeout)
145 resp, err := r.Poll.FetchJob(pctx, connect.NewRequest(&ermite.FetchJobRequest{PollId: pollID, Version: r.Cfg.Version}))
146 cancel()
147 if err != nil {
148 <-slots
149 if ctx.Err() != nil {
150 return nil
151 }
152 if connect.CodeOf(err) == connect.CodeUnauthenticated {
153 return ErrRemoved
154 }
155 if connect.CodeOf(err) == connect.CodeInvalidArgument {
156 pollID = "" // our id was refused: a new one
157 }
158 d := bo.next()
159 log.Warn().Err(err).Dur("retry_in", d).Msg("FetchJob failed")
160 if !sleep(ctx, d) {
161 return nil
162 }
163 continue
164 }
165 bo.cur = 0
166 pollID = "" // answered: the next poll is a new one
167 if resp.Msg.Job == nil {
168 <-slots
169 d := t.NoJobPause
170 if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > d {
171 d = ra.AsDuration()
172 }
173 if !sleep(ctx, d) {
174 return nil
175 }
176 continue
177 }
178 r.startJob(ctx, log, resp.Msg.Job, t, shutdown, slots)
179 if ra := resp.Msg.RetryAfter; ra != nil && ra.AsDuration() > 0 {
180 if !sleep(ctx, ra.AsDuration()) {
181 return nil
182 }
183 }
184 }
185 return nil
186}
187
188// recover waits until no job runs, then reaps. It returns false when ctx ends.
189func (r *Runner) recover(ctx context.Context, log zerowrap.Logger, slots chan struct{}, retry time.Duration) bool {
190 log.Error().Msg("a sandbox could not be cleaned up: no new job until the running ones end and the leftovers are reaped")
191 for ctx.Err() == nil {
192 if len(slots) == 0 {
193 if err := r.Driver.Reap(ctx); err == nil {
194 r.leaked.Store(false)
195 log.Info().Msg("reaped: taking jobs again")
196 return true
197 } else if !errors.Is(err, sandbox.ErrBusy) {
198 log.Error().Err(err).Msg("reaping failed, retrying")
199 }
200 }
201 if !sleep(ctx, retry) {
202 return false
203 }
204 }
205 return false
206}
207
208func (r *Runner) startJob(ctx context.Context, log zerowrap.Logger, job *ermite.Job, t Timing, shutdown <-chan struct{}, slots chan struct{}) {
209 jctx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
210 jr := &jobRun{
211 r: r, job: job, client: r.Job(job.JobToken), t: t, ctx: jctx, cancel: cancel,
212 log: zerowrap.Logger{Logger: log.With().Str("job_id", job.JobId).Int64("run_id", job.RunId).Int32("attempt", job.Attempt).Logger()},
213 receivedAt: r.clock(), cancelCh: make(chan *execpb.Cancel, 1), steps: map[int32]*ermite.StepUpdate{},
214 annSeen: map[int32]bool{}, sumSeen: map[int32]bool{},
215 }
216 jr.log.Info().Str("job_key", job.JobKey).Msg("job assigned")
217 r.active.Add(1)
218 go func() {
219 defer r.active.Done()
220 defer func() { <-slots }()
221 jr.run(r, shutdown)
222 jr.log.Info().Msg("job finished")
223 }()
224}
225
226// ---- registration
227
228// Isolation maps the driver's isolation to the protocol's.
229func Isolation(i sandbox.Isolation) ermite.RunnerIsolation {
230 switch i {
231 case sandbox.IsolationVM:
232 return ermite.RunnerIsolation_RUNNER_ISOLATION_VM
233 case sandbox.IsolationContainerSysbox:
234 return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX
235 case sandbox.IsolationContainerPrivileged:
236 return ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED
237 }
238 return ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED
239}
240
241// GoArch is the architecture in GitHub's names.
242func GoArch() string {
243 if runtime.GOARCH == "arm64" {
244 return "arm64"
245 }
246 return "x64"
247}
248
249// Register registers the runner with ermite using a token that has the
250// runner:register scope, and stores the runner token (mode 0600) in the state
251// directory. replace removes a runner of the same name first.
252func Register(ctx context.Context, cfg Config, client ermiteconnect.RunnerServiceClient, isolation sandbox.Isolation, replace bool) (State, error) {
253 iso := Isolation(isolation)
254 if iso == ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED {
255 return State{}, fmt.Errorf("unknown isolation %q", isolation)
256 }
257 rctx, cancel := context.WithTimeout(ctx, 30*time.Second)
258 defer cancel()
259 resp, err := client.RegisterRunner(rctx, connect.NewRequest(&ermite.RegisterRunnerRequest{
260 Name: cfg.Name, Labels: cfg.Labels, Version: cfg.Version, Isolation: iso, Arch: GoArch(), Os: "linux", Replace: replace,
261 }))
262 if err != nil {
263 return State{}, fmt.Errorf("register: %w", err)
264 }
265 st := State{URL: cfg.URL, RunnerID: resp.Msg.RunnerId, Name: cfg.Name, RunnerToken: resp.Msg.RunnerToken}
266 if err := SaveState(cfg.StateDir, st); err != nil {
267 return State{}, fmt.Errorf("the runner is registered but its token could not be stored: %w", err)
268 }
269 return st, nil
270}