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