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