1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "sort"
10 "strings"
11 "sync"
12 "time"
13 "unicode/utf8"
14
15 "connectrpc.com/connect"
16 "github.com/bnema/zerowrap"
17 "google.golang.org/protobuf/proto"
18 "google.golang.org/protobuf/types/known/timestamppb"
19
20 ermite "bnema.dev/ermite-api/gen/ermite"
21 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
22
23 "bnema.dev/ermite-runner/internal/execstream"
24 "bnema.dev/ermite-runner/internal/execstream/execpb"
25 "bnema.dev/ermite-runner/internal/sandbox"
26)
27
28// jobClient is what a job needs from the RunnerService client (job token).
29type jobClient interface {
30 callClient
31 Heartbeat(context.Context, *connect.Request[ermite.HeartbeatRequest]) (*connect.Response[ermite.HeartbeatResponse], error)
32 WatchJob(context.Context, *connect.Request[ermite.WatchJobRequest]) (*connect.ServerStreamForClient[ermite.WatchJobResponse], error)
33 CompleteJob(context.Context, *connect.Request[ermite.CompleteJobRequest]) (*connect.Response[ermite.CompleteJobResponse], error)
34 ResolveActions(context.Context, *connect.Request[ermite.ResolveActionsRequest]) (*connect.Response[ermite.ResolveActionsResponse], error)
35}
36
37var _ jobClient = ermiteconnect.RunnerServiceClient(nil)
38
39// jobRun is the host's side of one job: the assignment, the sandbox, and the
40// calls to ermite with the job token.
41type jobRun struct {
42 r *Runner
43 job *ermite.Job
44 client jobClient
45 t Timing
46 log zerowrap.Logger // with job_id, run_id and attempt
47
48 // ctx lives until the job is finished or abandoned. Shutdown does not end
49 // it: a shutdown is a cancellation that the job reports.
50 ctx context.Context
51 cancel context.CancelCauseFunc
52
53 // receivedAt is when the host got the job: the job timeout counts from it
54 // (runner.proto: Job.timeout_minutes).
55 receivedAt time.Time
56
57 cancelCh chan *execpb.Cancel // the first value is sent to the executor
58 cancelMu sync.Mutex
59 cancelAt string // reason of the first cancel request, "" if none
60 cancelKind execpb.CancelKind // kind of the first cancel request
61 stopping bool // the runner is shutting down
62
63 mu sync.Mutex
64 steps map[int32]*ermite.StepUpdate
65 // What the guest has told us, for the validation of the next frame: the
66 // guest is untrusted code, its step numbers and lines are checked before
67 // they reach the server (see violate).
68 maxStarted int32
69 lastLogStep int32
70 violation string
71 // contexts is Job.Contexts as handed to the executor: with the server URLs
72 // pointed at the runner's URL when guest forge access is on.
73 contexts string
74 // Origins that are ermite for the archive tokens (see resolveResponse), and
75 // the budget of resolution requests of the job (see resolveActions).
76 publicOrigin, serverOrigin, guestForge string
77 resolveRounds, resolveRefs int
78 resolved map[string]*execpb.ResolvedAction // by "uses", for the job
79 // features is what the server accepts; annLists are the annotations the
80 // executor sent for steps that have not completed yet (see results.go).
81 features jobFeatures
82 // results says the job gets the results service (see resultsOn).
83 results bool
84 annLists map[int32][]*ermite.Annotation
85 // annSeen and sumSeen are the steps that sent their annotations and their
86 // summary: each is sent once.
87 annSeen, sumSeen map[int32]bool
88 annDropped int // annotations dropped for lack of room, see reportDroppedAnnotations
89 // hard stops the sandbox at once; set before the session starts.
90 hard func()
91 result *execpb.JobResult
92 outbox *outbox
93 endedBy error // the server ended the job, or the assignment is gone
94}
95
96var errAbandoned = errors.New("the server ended the job or revoked the assignment")
97
98func (j *jobRun) abandon(err error) {
99 j.mu.Lock()
100 if j.endedBy == nil {
101 j.endedBy = err
102 }
103 j.mu.Unlock()
104 j.cancel(errAbandoned)
105}
106
107func (j *jobRun) abandoned() error {
108 j.mu.Lock()
109 defer j.mu.Unlock()
110 return j.endedBy
111}
112
113// requestCancel asks the executor to stop; only the first request counts.
114func (j *jobRun) requestCancel(reason string, kind execpb.CancelKind) {
115 j.cancelMu.Lock()
116 defer j.cancelMu.Unlock()
117 if j.cancelAt != "" {
118 return
119 }
120 j.cancelAt = reason
121 j.cancelKind = kind
122 j.cancelCh <- &execpb.Cancel{Reason: reason, Kind: kind}
123}
124
125// firstCancelKind is the kind of the first cancel request.
126func (j *jobRun) firstCancelKind() execpb.CancelKind {
127 j.cancelMu.Lock()
128 defer j.cancelMu.Unlock()
129 return j.cancelKind
130}
131
132func (j *jobRun) cancelRequested() bool {
133 j.cancelMu.Lock()
134 defer j.cancelMu.Unlock()
135 return j.cancelAt != ""
136}
137
138// heartbeat sends the acknowledgement. It retries every HeartbeatRetry until
139// it succeeds, so that the 30 s of the server are used well; it returns false
140// when the assignment is gone.
141func (j *jobRun) ack(ctx context.Context) bool {
142 for {
143 err := j.beat(ctx)
144 if err == nil {
145 return true
146 }
147 if ctx.Err() != nil {
148 return false
149 }
150 switch classify(err) {
151 case kindEnded, kindAbandoned:
152 j.abandon(err)
153 return false
154 }
155 j.log.Warn().Err(err).Msg("first heartbeat failed, retrying")
156 if !sleep(ctx, j.t.HeartbeatRetry) {
157 return false
158 }
159 }
160}
161
162func (j *jobRun) beat(ctx context.Context) error {
163 cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
164 defer cancel()
165 _, err := j.client.Heartbeat(cctx, connect.NewRequest(&ermite.HeartbeatRequest{JobId: j.job.JobId}))
166 return err
167}
168
169// heartbeats keeps the job alive: every HeartbeatEvery, and after a failure
170// every HeartbeatRetry until one succeeds (runner.proto: Heartbeat).
171func (j *jobRun) heartbeats(ctx context.Context) {
172 wait := j.t.HeartbeatEvery
173 for {
174 if !sleep(ctx, wait) {
175 return
176 }
177 if err := j.beat(ctx); err != nil {
178 if ctx.Err() != nil {
179 return
180 }
181 switch classify(err) {
182 case kindEnded, kindAbandoned:
183 j.abandon(err)
184 return
185 }
186 j.log.Warn().Err(err).Dur("retry_in", j.t.HeartbeatRetry).Msg("heartbeat failed")
187 wait = j.t.HeartbeatRetry
188 continue
189 }
190 wait = j.t.HeartbeatEvery
191 }
192}
193
194// watch follows WatchJob: a cancel request stops the job, an Ended message or
195// the end-of-job error abandons it. The stream is reopened after any other
196// error and after a clean close without Ended.
197func (j *jobRun) watch(ctx context.Context) {
198 for ctx.Err() == nil {
199 stream, err := j.client.WatchJob(ctx, connect.NewRequest(&ermite.WatchJobRequest{JobId: j.job.JobId}))
200 if err == nil {
201 for stream.Receive() {
202 switch m := stream.Msg().Msg.(type) {
203 case *ermite.WatchJobResponse_Cancel:
204 j.requestCancel(cancelText(m.Cancel), kindOf(m.Cancel.GetReason()))
205 case *ermite.WatchJobResponse_Ended:
206 _ = stream.Close()
207 j.abandon(fmt.Errorf("job ended: %s", m.Ended.GetConclusion()))
208 return
209 }
210 }
211 err = stream.Err()
212 _ = stream.Close()
213 }
214 if ctx.Err() != nil {
215 return
216 }
217 if err != nil {
218 switch classify(err) {
219 case kindEnded, kindAbandoned:
220 j.abandon(err)
221 return
222 }
223 j.log.Warn().Err(err).Msg("WatchJob failed, reopening")
224 }
225 if !sleep(ctx, j.t.WatchRetry) {
226 return
227 }
228 }
229}
230
231func cancelText(c *ermite.CancelRequest) string {
232 if c.GetMessage() != "" {
233 return c.GetMessage()
234 }
235 return c.GetReason().String()
236}
237
238func kindOf(r ermite.CancelReason) execpb.CancelKind {
239 if r == ermite.CancelReason_CANCEL_REASON_TIMEOUT {
240 return execpb.CancelKind_CANCEL_KIND_TIMEOUT
241 }
242 // User, fail-fast, and any reason this runner does not know: cancelled.
243 return execpb.CancelKind_CANCEL_KIND_USER
244}
245
246// ---- steps
247
248// Limits on what the guest may send, which keep the memory of the host bounded
249// whatever the guest does: runner.proto allows 1,000 steps per job, names of
250// 1,024 characters, and lines of 64 KiB.
251const (
252 maxSteps = 1000
253 maxNameLen = 1024
254 maxLineText = 64 * 1024
255)
256
257// violate fails the job because the guest broke the protocol: it writes the
258// reason in the log, asks the executor to stop and destroys the sandbox. Only
259// the first violation counts; the frames that follow are dropped.
260func (j *jobRun) violate(format string, a ...any) {
261 j.mu.Lock()
262 if j.violation != "" {
263 j.mu.Unlock()
264 return
265 }
266 j.violation = fmt.Sprintf(format, a...)
267 msg := j.violation
268 j.mu.Unlock()
269 j.log.Error().Str("violation", msg).Msg("the executor broke the protocol")
270 j.sandboxError("The runner stopped the job: the executor broke the protocol: " + msg)
271 if j.hard != nil {
272 j.hard()
273 }
274}
275
276func (j *jobRun) violated() bool {
277 j.mu.Lock()
278 defer j.mu.Unlock()
279 return j.violation != ""
280}
281
282func (j *jobRun) stepStarted(s *execpb.StepStarted) {
283 j.mu.Lock()
284 switch {
285 case j.violation != "":
286 j.mu.Unlock()
287 return
288 case s.Number < 1 || s.Number > maxSteps:
289 j.mu.Unlock()
290 j.violate("step number %d is outside 1..%d", s.Number, maxSteps)
291 return
292 case s.Number <= j.maxStarted:
293 j.mu.Unlock()
294 j.violate("step %d started after step %d", s.Number, j.maxStarted)
295 return
296 }
297 j.maxStarted = s.Number
298 u := &ermite.StepUpdate{Number: s.Number, Name: cutName(s.Name), State: ermite.StepState_STEP_STATE_IN_PROGRESS, StartedAt: tsOf(s.UnixNano)}
299 j.steps[s.Number] = u
300 j.mu.Unlock()
301 j.outbox.Step(u, nil)
302}
303
304func (j *jobRun) stepCompleted(s *execpb.StepCompleted) {
305 j.mu.Lock()
306 if j.violation != "" {
307 j.mu.Unlock()
308 return
309 }
310 old := j.steps[s.Number]
311 switch {
312 case old == nil:
313 j.mu.Unlock()
314 j.violate("step %d completed without having started", s.Number)
315 return
316 case old.State == ermite.StepState_STEP_STATE_COMPLETED:
317 j.mu.Unlock()
318 j.violate("step %d completed twice", s.Number)
319 return
320 }
321 u := cloneStep(old)
322 if s.Name != "" {
323 u.Name = cutName(s.Name)
324 }
325 u.State = ermite.StepState_STEP_STATE_COMPLETED
326 u.Conclusion = conclusionOf(s.Conclusion)
327 if u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || u.Conclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
328 u.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
329 }
330 u.CompletedAt = tsOf(s.UnixNano)
331 // The outcome is how the step ended before continue-on-error; a runner that
332 // does not say leaves it unspecified, which means the conclusion.
333 if o := conclusionOf(s.Outcome); o != ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED {
334 if o == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
335 o = ermite.CiConclusion_CI_CONCLUSION_FAILURE
336 }
337 u.Outcome = o
338 }
339 j.steps[s.Number] = u
340 anns := j.takeAnnotations(s.Number)
341 j.mu.Unlock()
342 j.outbox.Step(u, anns)
343}
344
345// logLine checks and queues a line of the log. The step must be one that was
346// started and not completed, and not lower than the step of the previous line
347// (the server refuses a log that goes back); the text is made valid for the
348// server: UTF-8, no line feed (a line feed in a line makes more lines).
349func (j *jobRun) logLine(l *execpb.Log) {
350 j.mu.Lock()
351 if j.violation != "" {
352 j.mu.Unlock()
353 return
354 }
355 st := j.steps[l.Step]
356 switch {
357 case l.Step < 1 || l.Step > maxSteps || st == nil:
358 j.mu.Unlock()
359 j.violate("a log line for step %d, which has not started", l.Step)
360 return
361 case st.State == ermite.StepState_STEP_STATE_COMPLETED:
362 j.mu.Unlock()
363 j.violate("a log line for step %d, which is completed", l.Step)
364 return
365 case l.Step < j.lastLogStep:
366 j.mu.Unlock()
367 j.violate("a log line for step %d after one for step %d", l.Step, j.lastLogStep)
368 return
369 case len(l.Text) > maxLineText:
370 j.mu.Unlock()
371 j.violate("a log line of %d bytes (the limit is %d)", len(l.Text), maxLineText)
372 return
373 }
374 j.lastLogStep = l.Step
375 j.mu.Unlock()
376 text := l.Text
377 if !utf8.ValidString(text) {
378 text = strings.ToValidUTF8(text, "\ufffd")
379 }
380 for i, part := range strings.Split(text, "\n") {
381 j.outbox.Log(logLine{step: l.Step, unixNano: l.UnixNano, text: part, continued: l.Continued && i == 0})
382 }
383}
384
385// cutName bounds a step name to what the server keeps, on a character boundary.
386// The result is a copy: a substring would keep the whole name the guest sent
387// alive (see cutRunes).
388func cutName(s string) string {
389 if len(s) <= maxNameLen {
390 return strings.ToValidUTF8(s, "\ufffd")
391 }
392 s = s[:maxNameLen]
393 for !utf8.ValidString(s) && len(s) > 0 {
394 s = s[:len(s)-1]
395 }
396 return strings.Clone(s)
397}
398
399func cloneStep(u *ermite.StepUpdate) *ermite.StepUpdate {
400 return proto.Clone(u).(*ermite.StepUpdate)
401}
402
403func tsOf(unixNano int64) *timestamppb.Timestamp {
404 if unixNano == 0 {
405 return timestamppb.Now()
406 }
407 return timestamppb.New(time.Unix(0, unixNano))
408}
409
410func conclusionOf(c execpb.StepConclusion) ermite.CiConclusion {
411 switch c {
412 case execpb.StepConclusion_STEP_CONCLUSION_SUCCESS:
413 return ermite.CiConclusion_CI_CONCLUSION_SUCCESS
414 case execpb.StepConclusion_STEP_CONCLUSION_FAILURE:
415 return ermite.CiConclusion_CI_CONCLUSION_FAILURE
416 case execpb.StepConclusion_STEP_CONCLUSION_CANCELLED:
417 return ermite.CiConclusion_CI_CONCLUSION_CANCELLED
418 case execpb.StepConclusion_STEP_CONCLUSION_SKIPPED:
419 return ermite.CiConclusion_CI_CONCLUSION_SKIPPED
420 case execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT:
421 return ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT
422 }
423 return ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED
424}
425
426// finalSteps closes the steps that never completed (the executor died or was
427// stopped) with the conclusion of the job, and returns all snapshots in order.
428func (j *jobRun) finalSteps(jobConclusion ermite.CiConclusion) []*ermite.StepUpdate {
429 j.mu.Lock()
430 defer j.mu.Unlock()
431 end := ermite.CiConclusion_CI_CONCLUSION_FAILURE
432 if jobConclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED || jobConclusion == ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
433 end = ermite.CiConclusion_CI_CONCLUSION_CANCELLED
434 }
435 var out []*ermite.StepUpdate
436 for _, u := range j.steps {
437 if u.State != ermite.StepState_STEP_STATE_COMPLETED {
438 u = cloneStep(u)
439 u.State, u.Conclusion, u.CompletedAt = ermite.StepState_STEP_STATE_COMPLETED, end, timestamppb.Now()
440 }
441 out = append(out, u)
442 }
443 sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
444 return out
445}
446
447// ---- the job
448
449// run runs the whole job: acknowledgement, sandbox, executor session, final
450// report. It returns when everything is over and the sandbox is gone.
451func (j *jobRun) run(rt *Runner, shutdown <-chan struct{}) {
452 defer j.cancel(nil)
453 ctx := j.ctx
454
455 if !j.acknowledge(ctx, shutdown) {
456 return
457 }
458 var bg sync.WaitGroup
459 bg.Add(2)
460 go func() { defer bg.Done(); j.heartbeats(ctx) }()
461 go func() { defer bg.Done(); j.watch(ctx) }()
462 // The context is cancelled first, so that the heartbeat and watch loops end.
463 defer func() { j.cancel(nil); bg.Wait() }()
464
465 j.outbox = newOutbox(j.client, j.job.JobId, j.t, func(err error) { j.abandon(err) }, j.log)
466 j.outbox.start(ctx)
467
468 stopHard := j.startStopTimers(ctx, shutdown)
469 // The hard stop frees the reader of the stream if it waits for room in the
470 // outbox (the server keeps failing): nothing else would wake it.
471 go func() {
472 select {
473 case <-stopHard:
474 j.outbox.releaseProducers()
475 case <-ctx.Done():
476 }
477 }()
478 out := j.execute(ctx, rt, stopHard)
479 j.report(ctx, out, shutdown)
480}
481
482// acknowledge sends the first heartbeat, before anything else. A shutdown ends
483// the wait: the job is not started and not acknowledged, and the server gives it
484// to another runner when no heartbeat comes within 30 s (runner.proto, FetchJob:
485// this is not a failure of the job). It reports whether the job goes on.
486func (j *jobRun) acknowledge(ctx context.Context, shutdown <-chan struct{}) bool {
487 ackCtx, stopAck := context.WithCancel(ctx)
488 defer stopAck()
489 go func() {
490 select {
491 case <-shutdown:
492 stopAck()
493 case <-ackCtx.Done():
494 }
495 }()
496 if j.ack(ackCtx) {
497 return true
498 }
499 select {
500 case <-shutdown:
501 j.log.Warn().Msg("the runner is shutting down before the job was acknowledged: the server will give it to another runner")
502 default:
503 j.log.Warn().Msg("the assignment was lost before the job started")
504 }
505 return false
506}
507
508// startStopTimers arms the two reasons to stop the job from the host, each a
509// cancellation followed by a hard stop of the sandbox (the returned channel is
510// closed) if the executor does not answer: a shutdown of the runner
511// (ShutdownGrace), and the job timeout, which the executor enforces itself but
512// a stuck executor would not (TimeoutGrace after the deadline: the escalation
513// SIGINT, SIGTERM, SIGKILL takes 10 s). The goroutines end with ctx.
514func (j *jobRun) startStopTimers(ctx context.Context, shutdown <-chan struct{}) <-chan struct{} {
515 stopHard := make(chan struct{})
516 var hardOnce sync.Once
517 hard := func() { hardOnce.Do(func() { close(stopHard) }) }
518 j.hard = hard
519 // after waits grace, then stops the sandbox; it returns early with ctx.
520 after := func(grace time.Duration) {
521 timer := time.NewTimer(grace)
522 defer timer.Stop()
523 select {
524 case <-timer.C:
525 hard()
526 case <-ctx.Done():
527 }
528 }
529 go func() {
530 select {
531 case <-shutdown:
532 case <-ctx.Done():
533 return
534 }
535 j.cancelMu.Lock()
536 j.stopping = true
537 j.cancelMu.Unlock()
538 j.requestCancel("the runner is shutting down", execpb.CancelKind_CANCEL_KIND_SHUTDOWN)
539 after(j.t.ShutdownGrace)
540 }()
541 go func() {
542 wait := time.NewTimer(time.Until(j.deadline()))
543 defer wait.Stop()
544 select {
545 case <-wait.C:
546 case <-ctx.Done():
547 return
548 }
549 j.log.Warn().Int("timeout_minutes", j.timeoutMinutes()).Msg("the job timeout has passed")
550 j.requestCancel("the job timeout has passed", execpb.CancelKind_CANCEL_KIND_TIMEOUT)
551 after(j.t.TimeoutGrace)
552 }()
553 return stopHard
554}
555
556// report ends the job towards the server: the log is flushed, then CompleteJob.
557// Nothing is sent to a job that the server has ended.
558func (j *jobRun) report(ctx context.Context, out outcome, shutdown <-chan struct{}) {
559 if err := j.abandoned(); err != nil {
560 j.log.Warn().Err(err).Msg("the server ended the job: no result is reported")
561 return
562 }
563 // The report is made even when the runner is shutting down: it uses its own
564 // deadline.
565 budget := j.t.CompleteDeadline
566 select {
567 case <-shutdown:
568 // The service manager waits only so long (see Timing): once the runner's
569 // shutdown channel is closed, whatever the job's outcome, the report gets
570 // what is left of the budget. The channel is read here, not the flag set
571 // by the stop timers, which may not have run yet.
572 budget = min(budget, j.t.ShutdownReport)
573 default:
574 }
575 rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), budget)
576 defer cancel()
577 // Output limits are checked before the log is flushed: a job whose outputs
578 // the server would refuse fails without outputs, with a line in the log
579 // (runner.proto: CompleteJobRequest.outputs), and lines sent after
580 // CompleteJob are rejected.
581 if msg := checkOutputs(out.outputs); msg != "" {
582 j.sandboxError(msg)
583 out = failure()
584 }
585 // The flush gets half of the budget: when the server keeps failing the
586 // uploads, the other half is still there for CompleteJob, which is what
587 // ends the job for the server.
588 fctx, fcancel := context.WithTimeout(rctx, budget/2)
589 j.outbox.Close(fctx)
590 fcancel()
591 if err := j.abandoned(); err != nil {
592 return
593 }
594 j.reportDroppedAnnotations(rctx, budget/4)
595 j.complete(rctx, out.conclusion, out.outputs, out.shutdown)
596}
597
598// outcome is how a job ended, as far as the host can tell.
599type outcome struct {
600 conclusion ermite.CiConclusion
601 outputs map[string]string
602 // shutdown is true when the runner's shutdown cut the job short.
603 shutdown bool
604}
605
606func failure() outcome { return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE} }
607
608// execute starts the sandbox, runs the job in it and returns how it ended.
609func (j *jobRun) execute(ctx context.Context, rt *Runner, stopHard <-chan struct{}) outcome {
610 var err error
611 j.setOrigins(j.job.Contexts, rt.Cfg.URL, rt.Cfg.GuestForgeOrigin)
612 if j.contexts, err = RewriteServerURLs(j.job.Contexts, rt.Cfg.GuestForgeOrigin); err != nil {
613 j.failEarly(fmt.Sprintf("The job's contexts cannot be prepared: %v", err))
614 return failure()
615 }
616 if j.contexts, err = SetGitHubToken(j.contexts, j.job.GithubToken); err != nil {
617 j.failEarly(fmt.Sprintf("The job's contexts cannot be prepared: %v", err))
618 return failure()
619 }
620 sb, err := rt.Driver.New()
621 if err != nil {
622 j.failEarly(fmt.Sprintf("The runner has no free sandbox: %v", err))
623 return failure()
624 }
625 // Stop runs to completion even when the job context is done.
626 defer j.stopSandbox(ctx, rt, sb)
627 j.features = j.loadFeatures(ctx, rt)
628 j.results = j.resultsOn(rt)
629
630 conn, err := j.startSandbox(ctx, rt, sb, stopHard)
631 if err != nil {
632 if ctx.Err() != nil || j.cancelRequested() {
633 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: j.isStopping()}
634 }
635 j.failEarly(fmt.Sprintf("The sandbox did not start: %v", err))
636 return failure()
637 }
638 defer func() { _ = conn.Close() }()
639
640 // Hard stops (grace elapsed, or the job was abandoned) close the stream. The
641 // goroutine ends when this function returns, whatever happens to ctx.
642 sessionDone := make(chan struct{})
643 defer close(sessionDone)
644 go func() {
645 select {
646 case <-stopHard:
647 _ = conn.Close()
648 case <-ctx.Done():
649 _ = conn.Close()
650 case <-sessionDone:
651 }
652 }()
653
654 exit, err := j.drive(conn, rt)
655 return j.outcomeOf(ctx, exit, err)
656}
657
658// stopSandbox destroys the sandbox; a cleanup that left something behind stops
659// the runner from taking jobs until it is reaped.
660func (j *jobRun) stopSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox) {
661 sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), j.t.StopDeadline)
662 defer cancel()
663 if err := sb.Stop(sctx); err != nil {
664 j.log.Error().Err(err).Msg("stopping the sandbox")
665 if errors.Is(err, sandbox.ErrLeaked) {
666 rt.markLeaked()
667 }
668 }
669}
670
671// startSandbox starts the sandbox; the start is bounded by the job context and
672// by the hard stop.
673func (j *jobRun) startSandbox(ctx context.Context, rt *Runner, sb sandbox.Sandbox, stopHard <-chan struct{}) (sandbox.Conn, error) {
674 startCtx, cancelStart := context.WithCancel(ctx)
675 defer cancelStart()
676 stopWatch := make(chan struct{})
677 defer close(stopWatch)
678 go func() {
679 select {
680 case <-stopHard:
681 cancelStart()
682 case <-stopWatch:
683 }
684 }()
685 return sb.Start(startCtx, rt.spec(j.job, j.resultsHandler(rt)))
686}
687
688// drive sends the job to the executor and handles its frames until it exits.
689func (j *jobRun) drive(conn sandbox.Conn, rt *Runner) (*execpb.Exit, error) {
690 run := &execpb.Run{Job: &execpb.JobSpec{
691 WorkflowPath: j.job.WorkflowPath, WorkflowYaml: j.job.WorkflowYaml, JobKey: j.job.JobKey,
692 Matrix: j.job.Matrix, Contexts: j.contexts, Repo: j.job.Repo, TimeoutMinutes: int32(j.timeoutMinutes()),
693 DeadlineMs: max(1, time.Until(j.deadline()).Milliseconds()),
694 RunnerName: rt.Cfg.Name, RunnerVersion: rt.Cfg.Version, RunnerEnvironment: rt.Cfg.environment(),
695 SendAnnotations: j.features.annotations, SendSummaries: j.features.summaries,
696 // Job.secrets and Job.vars are not in this frame (it is limited to
697 // 2 MiB): DriveJob sends them in NamedValue frames before it. The host
698 // never logs them nor stores them. An old server leaves trusted false
699 // because it does not send it, so the executor says "not available"
700 // only when the server has ci.secrets.
701 SecretsWithheld: j.features.secrets && !j.job.Trusted,
702 }}
703 run.Job.IdTokenUrl, run.Job.IdTokenRequestToken = j.idTokenRequest()
704 if p := j.job.Permissions; p != nil && j.features.secrets {
705 run.Job.TokenPermissions = &execpb.TokenPermissions{Levels: p.GetLevels()}
706 }
707 run.Job.CachePort, run.Job.CacheToken, run.Job.CacheMode, run.Job.Results = j.runtimeFor(rt)
708 _, exit, err := execstream.DriveJob(conn, run, j.cancelCh, execstream.JobHandler{
709 StepStarted: j.stepStarted, StepCompleted: j.stepCompleted,
710 Log: j.logLine,
711 JobResult: func(r *execpb.JobResult) { j.mu.Lock(); j.result = r; j.mu.Unlock() },
712 Resolve: j.resolveActions,
713 StepAnnotations: j.stepAnnotations, StepSummary: j.stepSummary,
714 Secrets: j.job.Secrets, Vars: j.job.Vars,
715 })
716 return exit, err
717}
718
719// idTokenRequest returns Job.id_token_url, its origin rewritten to the guest
720// forge as for github.server_url (RewriteServerURLs), and
721// Job.id_token_request_token: the ACTIONS_ID_TOKEN_REQUEST_URL and
722// ACTIONS_ID_TOKEN_REQUEST_TOKEN of the steps. Both are empty unless the
723// server sent both. The token is never logged.
724func (j *jobRun) idTokenRequest() (url, token string) {
725 url, token = j.job.GetIdTokenUrl(), j.job.GetIdTokenRequestToken()
726 if url == "" || token == "" {
727 return "", ""
728 }
729 if j.guestForge != "" {
730 url, _ = rewriteOrigin(url, j.publicOrigin, j.guestForge)
731 }
732 return url, token
733}
734
735// outcomeOf decides how the job ended from what the executor reported and what
736// the host asked of it.
737func (j *jobRun) outcomeOf(ctx context.Context, exit *execpb.Exit, driveErr error) outcome {
738 j.mu.Lock()
739 res := j.result
740 j.mu.Unlock()
741
742 switch {
743 case j.violated():
744 // The step numbers or lines of the executor were wrong: whatever it
745 // reported is not trusted, the job fails (the reason is in the log).
746 return failure()
747 case res != nil:
748 c := conclusionOf(res.Conclusion)
749 if c == ermite.CiConclusion_CI_CONCLUSION_UNSPECIFIED || c == ermite.CiConclusion_CI_CONCLUSION_SKIPPED {
750 c = ermite.CiConclusion_CI_CONCLUSION_FAILURE
751 }
752 if res.Error != "" {
753 j.log.Info().Str("error", res.Error).Msg("the job reported an error")
754 }
755 return outcome{conclusion: c, outputs: res.Outputs, shutdown: res.RunnerShutdown && c == ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
756 case j.cancelRequested() || ctx.Err() != nil:
757 // The executor did not report: it was stopped. A timeout wins over a
758 // shutdown or a user cancel that came later, as on the server (the first
759 // reason decides, and the timeout is the first when it fired first).
760 switch {
761 case j.firstCancelKind() == execpb.CancelKind_CANCEL_KIND_TIMEOUT:
762 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT}
763 case j.isStopping():
764 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, shutdown: true}
765 }
766 return outcome{conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}
767 }
768 why := "the executor ended without a result"
769 if driveErr != nil {
770 why = driveErr.Error()
771 } else if exit != nil && exit.Error != "" {
772 why = exit.Error
773 }
774 j.sandboxError(fmt.Sprintf("The job's sandbox ended unexpectedly: %s", why))
775 return failure()
776}
777
778// timeoutMinutes is the job timeout: at least 1, 360 when the job has none
779// (runner.proto: Job.timeout_minutes).
780func (j *jobRun) timeoutMinutes() int {
781 if m := int(j.job.TimeoutMinutes); m > 0 {
782 return m
783 }
784 return 360
785}
786
787// deadline is when the job times out, counted from its receipt.
788func (j *jobRun) deadline() time.Time {
789 return j.receivedAt.Add(time.Duration(j.timeoutMinutes()) * time.Minute)
790}
791
792func (j *jobRun) isStopping() bool {
793 j.cancelMu.Lock()
794 defer j.cancelMu.Unlock()
795 return j.stopping
796}
797
798// failEarly reports a job that could not start: step 1 fails with the reason
799// in its log.
800func (j *jobRun) failEarly(msg string) {
801 now := time.Now()
802 j.stepStarted(&execpb.StepStarted{Number: 1, Name: "Set up job", UnixNano: now.UnixNano()})
803 j.outbox.Log(logLine{step: 1, unixNano: now.UnixNano(), text: "##[error]" + msg})
804 j.stepCompleted(&execpb.StepCompleted{Number: 1, Name: "Set up job", Conclusion: execpb.StepConclusion_STEP_CONCLUSION_FAILURE, UnixNano: time.Now().UnixNano()})
805}
806
807// sandboxError writes the reason into the log of the step that was running.
808func (j *jobRun) sandboxError(msg string) {
809 j.mu.Lock()
810 var last int32 = 1
811 for n := range j.steps {
812 last = max(last, n)
813 }
814 j.mu.Unlock()
815 j.outbox.Log(logLine{step: last, unixNano: time.Now().UnixNano(), text: "##[error]" + msg})
816}
817
818// complete reports the end of the job. CompleteJob is idempotent, so a lost
819// response is retried until it is answered (applied or not).
820func (j *jobRun) complete(ctx context.Context, conclusion ermite.CiConclusion, outputs map[string]string, shutdown bool) {
821 req := &ermite.CompleteJobRequest{
822 JobId: j.job.JobId, Conclusion: conclusion, Outputs: outputs,
823 RunnerShutdown: shutdown && conclusion == ermite.CiConclusion_CI_CONCLUSION_CANCELLED,
824 Steps: j.finalSteps(conclusion),
825 // The annotations no UpdateJob has carried: all of them with the
826 // step they belong to, whole lists (replace semantics).
827 Annotations: j.allPendingAnnotations(),
828 }
829 err := j.sendComplete(ctx, req, conclusion)
830 if err == nil || ctx.Err() != nil || classify(err) != kindPermanent {
831 return
832 }
833 // The server refused the result itself (invalid_argument, or
834 // failed_precondition that is not the end of the job). CompleteJob is
835 // "applied whole or not at all", so the job is still open: say why in the
836 // log, then report a plain failure without outputs and with steps that are
837 // valid by construction. Without this the job would stay running until the
838 // heartbeat deadline of the server and then be reported as a lost runner.
839 j.errorLog(ctx, fmt.Sprintf("The runner could not report the result of the job: %s. The job is reported as failed.", connect.CodeOf(err)))
840 j.log.Error().Err(err).Msg("CompleteJob refused: retrying as a failure without outputs")
841 req = &ermite.CompleteJobRequest{
842 JobId: j.job.JobId, Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE,
843 Steps: j.rebuiltSteps(),
844 }
845 if err := j.sendComplete(ctx, req, req.Conclusion); err != nil && ctx.Err() == nil {
846 j.log.Error().Err(err).Msg("CompleteJob refused again, giving up")
847 }
848}
849
850// sendComplete sends the request until the server answers: success, the end of
851// the job, a lost assignment, or a refusal (returned).
852func (j *jobRun) sendComplete(ctx context.Context, req *ermite.CompleteJobRequest, conclusion ermite.CiConclusion) error {
853 b := backoff{min: j.t.BackoffMin, max: j.t.BackoffMax}
854 for {
855 cctx, cancel := context.WithTimeout(ctx, j.t.CallTimeout)
856 resp, err := j.client.CompleteJob(cctx, connect.NewRequest(req))
857 cancel()
858 if err == nil {
859 if resp.Msg.Conclusion != conclusion {
860 j.log.Warn().Str("reported", conclusion.String()).Str("stored", resp.Msg.Conclusion.String()).Msg("the server holds another conclusion")
861 }
862 return nil
863 }
864 if ctx.Err() != nil {
865 j.log.Error().Err(ctx.Err()).Msg("could not report the result")
866 return err
867 }
868 switch classify(err) {
869 case kindEnded, kindAbandoned:
870 return nil
871 case kindPermanent:
872 return err
873 }
874 j.log.Warn().Err(err).Msg("CompleteJob failed, retrying")
875 if !sleep(ctx, b.next()) {
876 return err
877 }
878 }
879}
880
881// errorLog appends a line to the log after the outbox is closed.
882func (j *jobRun) errorLog(ctx context.Context, msg string) {
883 j.appendLogLine(ctx, "##[error]"+msg)
884}
885
886// appendLogLine appends one line, in the last step, after the outbox is closed.
887func (j *jobRun) appendLogLine(ctx context.Context, text string) {
888 j.mu.Lock()
889 var last int32 = 1
890 for n := range j.steps {
891 last = max(last, n)
892 }
893 j.mu.Unlock()
894 j.outbox.AppendNow(ctx, logLine{step: last, unixNano: time.Now().UnixNano(), text: text})
895}
896
897// rebuiltSteps returns every known step as a valid, completed snapshot: number
898// in 1..maxSteps, a name of at most 1,024 bytes, a conclusion the server
899// accepts (success, failure, cancelled, skipped), both times set, and no step
900// that finished before it began. A step that was running is failed.
901func (j *jobRun) rebuiltSteps() []*ermite.StepUpdate {
902 j.mu.Lock()
903 defer j.mu.Unlock()
904 now := timestamppb.Now()
905 var out []*ermite.StepUpdate
906 for n, u := range j.steps {
907 if n < 1 || n > maxSteps {
908 continue
909 }
910 c := proto.Clone(u).(*ermite.StepUpdate)
911 c.Name = cutName(c.Name)
912 if c.StartedAt == nil {
913 c.StartedAt = now
914 }
915 if c.State != ermite.StepState_STEP_STATE_COMPLETED {
916 c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
917 c.CompletedAt = now
918 }
919 switch c.Conclusion {
920 case ermite.CiConclusion_CI_CONCLUSION_SUCCESS, ermite.CiConclusion_CI_CONCLUSION_FAILURE,
921 ermite.CiConclusion_CI_CONCLUSION_CANCELLED, ermite.CiConclusion_CI_CONCLUSION_SKIPPED:
922 default:
923 c.Conclusion = ermite.CiConclusion_CI_CONCLUSION_FAILURE
924 }
925 c.State = ermite.StepState_STEP_STATE_COMPLETED
926 if c.CompletedAt == nil || c.CompletedAt.AsTime().Before(c.StartedAt.AsTime()) {
927 c.CompletedAt = c.StartedAt
928 }
929 out = append(out, c)
930 }
931 sort.Slice(out, func(a, b int) bool { return out[a].Number < out[b].Number })
932 return out
933}
934
935// Limits of CompleteJobRequest.outputs.
936const (
937 maxOutputBytes = 1 << 20
938 maxOutputName = 255
939)
940
941// checkOutputs returns a message when the server would refuse the outputs: a
942// name of 1 to 255 letters, digits, "_" and "-" starting with a letter or "_",
943// no two names equal ignoring case, each value and the sum of names and values
944// at most 1 MiB.
945func checkOutputs(outputs map[string]string) string {
946 total := 0
947 seen := map[string]string{}
948 names := make([]string, 0, len(outputs))
949 for n := range outputs {
950 names = append(names, n)
951 }
952 sort.Strings(names)
953 for _, n := range names {
954 v := outputs[n]
955 if !validOutputName(n) {
956 return fmt.Sprintf("Job output name %q is not valid: use 1 to 255 letters, digits, '_' and '-', starting with a letter or '_'.", n)
957 }
958 if prev, dup := seen[strings.ToLower(n)]; dup {
959 return fmt.Sprintf("Job outputs %q and %q differ only by case.", prev, n)
960 }
961 seen[strings.ToLower(n)] = n
962 if len(v) > maxOutputBytes {
963 return fmt.Sprintf("Job output %q is larger than 1 MiB.", n)
964 }
965 total += len(n) + len(v)
966 }
967 if total > maxOutputBytes {
968 return "The job outputs are larger than 1 MiB in total."
969 }
970 return ""
971}
972
973func validOutputName(n string) bool {
974 if n == "" || len(n) > maxOutputName {
975 return false
976 }
977 for i, r := range n {
978 letter := r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r == '_'
979 if ok := letter || (i > 0 && (r >= '0' && r <= '9' || r == '-')); !ok {
980 return false
981 }
982 }
983 return true
984}