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