1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "encoding/json"
8 "errors"
9 "io"
10 "net/http"
11 "os"
12 "path/filepath"
13 "regexp"
14 "strconv"
15 "strings"
16 "testing"
17 "time"
18 "unicode/utf8"
19
20 "google.golang.org/protobuf/types/known/timestamppb"
21
22 "connectrpc.com/connect"
23
24 ermite "bnema.dev/ermite-api/gen/ermite"
25 "bnema.dev/ermite-api/gen/ermite/ermiteconnect"
26
27 "bnema.dev/ermite-runner/internal/execstream"
28 "bnema.dev/ermite-runner/internal/execstream/execpb"
29 "bnema.dev/ermite-runner/internal/sandbox"
30)
31
32func TestJobRunsAndReports(t *testing.T) {
33 h := newHarness(t, simpleScript("hello", "world"))
34 h.srv.jobs = []*ermite.Job{testJob("j1")}
35 stop := h.run(t)
36
37 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
38 if err := stop(); err != nil {
39 t.Fatal(err)
40 }
41 s := h.srv
42 s.mu.Lock()
43 defer s.mu.Unlock()
44 c := s.completes[0]
45 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || c.Outputs["out"] != "v" || c.RunnerShutdown {
46 t.Fatalf("%+v", c)
47 }
48 if len(s.beats) == 0 {
49 t.Error("no heartbeat: the first one is the acknowledgement")
50 }
51 if strings.Join(s.lines, "|") != "hello|world" {
52 t.Errorf("lines %q", s.lines)
53 }
54 for n, name := range map[int32]string{1: "Set up job", 2: "Run it", 3: "Complete job"} {
55 u := s.steps[n]
56 if u == nil || u.Name != name || u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || u.StartedAt == nil || u.CompletedAt == nil {
57 t.Errorf("step %d: %v", n, u)
58 }
59 }
60 if len(c.Steps) != 3 {
61 t.Errorf("CompleteJob carries %d final steps", len(c.Steps))
62 }
63 if h.driver.stops != 1 || h.driver.reaps < 1 {
64 t.Errorf("stops=%d reaps=%d: the sandbox must be stopped and the start must reap", h.driver.stops, h.driver.reaps)
65 }
66 if strings.Contains(h.logs.String(), "ermitejob_") || strings.Contains(h.logs.String(), "ermiterun_") {
67 t.Errorf("a token reached the log: %s", h.logs.String())
68 }
69}
70
71func TestTheJobTokenAndTheRunnerTokenStayApart(t *testing.T) {
72 h := newHarness(t, simpleScript("x"))
73 h.srv.jobs = []*ermite.Job{testJob("j1")}
74 var sawJobSpec *execpb.JobSpec
75 inner := h.driver.script
76 h.driver.script = func(t *testing.T, c *execstream.Conn, run *execpb.Run, cn <-chan *execpb.Cancel) {
77 sawJobSpec = run.Job
78 inner(t, c, run, cn)
79 }
80 h.run(t)
81 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
82 // The sandbox gets the workflow and contexts, never a token.
83 if sawJobSpec == nil || sawJobSpec.JobKey != "j" || sawJobSpec.RunnerName != "r1" || sawJobSpec.RunnerEnvironment != "github-hosted" || sawJobSpec.TimeoutMinutes != 5 {
84 t.Fatalf("%v", sawJobSpec)
85 }
86 if strings.Contains(sawJobSpec.String(), "ermitejob_") || strings.Contains(sawJobSpec.String(), "ermiterun_") {
87 t.Fatal("a token reached the sandbox")
88 }
89}
90
91func TestPollRetryUsesTheSamePollIDAfterAnError(t *testing.T) {
92 h := newHarness(t, simpleScript("x"))
93 h.srv.pollHook = func(n int, req *ermite.FetchJobRequest) (*ermite.FetchJobResponse, error) {
94 if n <= 2 { // the response is lost twice
95 return nil, connect.NewError(connect.CodeUnavailable, io.EOF)
96 }
97 if n == 3 {
98 return &ermite.FetchJobResponse{Job: testJob("j1")}, nil
99 }
100 return nil, nil
101 }
102 h.run(t)
103 eventually(t, "CompleteJob and the next poll", func() bool {
104 h.srv.mu.Lock()
105 defer h.srv.mu.Unlock()
106 return len(h.srv.completes) == 1 && len(h.srv.polls) >= 4
107 })
108 h.srv.mu.Lock()
109 defer h.srv.mu.Unlock()
110 if len(h.srv.polls) < 4 {
111 t.Fatalf("polls %v", h.srv.polls)
112 }
113 if h.srv.polls[0] == "" || h.srv.polls[0] != h.srv.polls[1] || h.srv.polls[1] != h.srv.polls[2] {
114 t.Errorf("a retried poll must keep its poll_id: %v", h.srv.polls)
115 }
116 if h.srv.polls[3] == h.srv.polls[2] {
117 t.Errorf("an answered poll is followed by a new id: %v", h.srv.polls)
118 }
119}
120
121func TestPollRespectsRetryAfter(t *testing.T) {
122 h := newHarness(t, simpleScript("x"))
123 h.srv.retryAfter = 300 * time.Millisecond
124 h.run(t)
125 time.Sleep(1000 * time.Millisecond)
126 h.srv.mu.Lock()
127 n := h.srv.pollCalls
128 h.srv.mu.Unlock()
129 if n < 2 || n > 5 {
130 t.Fatalf("%d polls in 1 s with retry_after=300ms", n)
131 }
132}
133
134func TestPollRemovedRunnerStops(t *testing.T) {
135 h := newHarness(t, simpleScript("x"))
136 h.runner.Poll = NewClient(h.url, "ermiterun_wrong", http.DefaultTransport)
137 err := h.runner.Run(h.logs.ctx())
138 if !errors.Is(err, ErrRemoved) {
139 t.Fatalf("%v", err)
140 }
141}
142
143func TestNoPollWithoutFreeCapacity(t *testing.T) {
144 release := make(chan struct{})
145 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
146 stepStart(c, 1, "Set up job")
147 <-release
148 stepDone(c, 1, "Set up job", cOK)
149 jobDone(c, cOK, nil, false)
150 })
151 h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
152 h.run(t)
153 eventually(t, "first job started", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 1 })
154 time.Sleep(300 * time.Millisecond)
155 h.srv.mu.Lock()
156 calls := h.srv.pollCalls
157 h.srv.mu.Unlock()
158 if calls != 1 {
159 t.Fatalf("polled %d times with the only slot busy", calls)
160 }
161 h.driver.mu.Lock()
162 if h.driver.started != 1 {
163 t.Errorf("started=%d", h.driver.started)
164 }
165 h.driver.mu.Unlock()
166 close(release)
167 eventually(t, "both jobs done", func() bool { return h.srv.completeCount() == 2 })
168}
169
170func TestCapacityTwoRunsTwoJobsAtOnce(t *testing.T) {
171 release := make(chan struct{})
172 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
173 stepStart(c, 1, "Set up job")
174 <-release
175 stepDone(c, 1, "Set up job", cOK)
176 jobDone(c, cOK, nil, false)
177 })
178 h.runner.Cfg.Capacity = 2
179 h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2"), testJob("j3")}
180 h.run(t)
181 eventually(t, "two sandboxes", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.started == 2 })
182 time.Sleep(200 * time.Millisecond)
183 h.driver.mu.Lock()
184 if h.driver.started != 2 {
185 t.Errorf("started=%d with capacity 2", h.driver.started)
186 }
187 h.driver.mu.Unlock()
188 close(release)
189 eventually(t, "three jobs done", func() bool { return h.srv.completeCount() == 3 })
190}
191
192func TestFirstHeartbeatRetriesEvery5sStyle(t *testing.T) {
193 h := newHarness(t, simpleScript("x"))
194 h.srv.jobs = []*ermite.Job{testJob("j1")}
195 h.srv.beatHook = func(n int) error {
196 if n <= 2 {
197 return connect.NewError(connect.CodeUnavailable, io.EOF)
198 }
199 return nil
200 }
201 h.run(t)
202 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
203 h.srv.mu.Lock()
204 defer h.srv.mu.Unlock()
205 if len(h.srv.beats) < 3 {
206 t.Fatalf("beats %d", len(h.srv.beats))
207 }
208 // The job starts only after the acknowledgement: no log before the third beat.
209 if gap := h.srv.beats[1].Sub(h.srv.beats[0]); gap > 90*time.Millisecond {
210 t.Errorf("retry after %v, want about HeartbeatRetry (30ms), not the 100ms period", gap)
211 }
212}
213
214func TestHeartbeatTicksAndRetriesQuickly(t *testing.T) {
215 release := make(chan struct{})
216 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
217 stepStart(c, 1, "Set up job")
218 <-release
219 stepDone(c, 1, "Set up job", cOK)
220 jobDone(c, cOK, nil, false)
221 })
222 h.srv.jobs = []*ermite.Job{testJob("j1")}
223 h.srv.beatHook = func(n int) error {
224 if n == 3 { // one failure in the middle of the job
225 return connect.NewError(connect.CodeUnavailable, io.EOF)
226 }
227 return nil
228 }
229 h.run(t)
230 eventually(t, "five heartbeats", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 5 })
231 h.srv.mu.Lock()
232 b := append([]time.Time(nil), h.srv.beats...)
233 h.srv.mu.Unlock()
234 if d := b[2].Sub(b[1]); d < 80*time.Millisecond {
235 t.Errorf("period %v, want about 100ms", d)
236 }
237 if d := b[3].Sub(b[2]); d > 80*time.Millisecond {
238 t.Errorf("the retry after a failure came after %v, want about 30ms", d)
239 }
240 close(release)
241 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
242}
243
244func TestCancelFromWatchJobReachesTheExecutor(t *testing.T) {
245 got := make(chan *execpb.Cancel, 1)
246 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
247 stepStart(c, 1, "Set up job")
248 stepStart(c, 2, "Run sleep")
249 logLineFrame(c, 2, "working", false)
250 cn := <-cancels
251 got <- cn
252 stepDone(c, 2, "Run sleep", cCancel)
253 jobDone(c, cCancel, nil, false)
254 })
255 h.srv.jobs = []*ermite.Job{testJob("j1")}
256 h.run(t)
257 eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
258 h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Keepalive{Keepalive: &ermite.Keepalive{}}}
259 h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_USER, Message: "Canceled by the owner"}}}
260 select {
261 case cn := <-got:
262 if cn.Reason != "Canceled by the owner" || cn.Kind != execpb.CancelKind_CANCEL_KIND_USER {
263 t.Fatalf("%v", cn)
264 }
265 case <-time.After(5 * time.Second):
266 t.Fatal("the cancel did not reach the executor")
267 }
268 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
269 c := h.srv.completes[0]
270 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || c.RunnerShutdown {
271 t.Fatalf("%v", c)
272 }
273}
274
275func TestTimeoutCancelKind(t *testing.T) {
276 got := make(chan *execpb.Cancel, 1)
277 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
278 stepStart(c, 1, "Set up job")
279 logLineFrame(c, 1, "x", false)
280 got <- <-cancels
281 jobDone(c, execpb.StepConclusion_STEP_CONCLUSION_TIMED_OUT, nil, false)
282 })
283 h.srv.jobs = []*ermite.Job{testJob("j1")}
284 h.run(t)
285 eventually(t, "log", func() bool { return len(h.srv.snapshotLines()) == 1 })
286 h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Cancel{Cancel: &ermite.CancelRequest{Reason: ermite.CancelReason_CANCEL_REASON_TIMEOUT}}}
287 if cn := <-got; cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
288 t.Fatalf("%v", cn)
289 }
290 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
291 if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT {
292 t.Fatalf("%v", c.Conclusion)
293 }
294}
295
296func TestWatchJobIsReopenedAfterAnError(t *testing.T) {
297 release := make(chan struct{})
298 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
299 stepStart(c, 1, "Set up job")
300 <-release
301 jobDone(c, cOK, nil, false)
302 })
303 h.srv.jobs = []*ermite.Job{testJob("j1")}
304 h.run(t)
305 eventually(t, "the first watch", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return h.srv.watches >= 1 })
306 // A message that ends the stream cleanly without Ended: the runner reopens.
307 // The fake closes the stream only on Ended or context end, so cut the
308 // transport instead: close idle connections of the test server.
309 http.DefaultTransport.(*http.Transport).CloseIdleConnections()
310 close(release)
311 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
312}
313
314func TestLogBatchesRespectTheLimitsAndKeepOrder(t *testing.T) {
315 var lines []string
316 big := strings.Repeat("x", 1000)
317 for i := 0; i < 2500; i++ {
318 lines = append(lines, big[:1+i%1000])
319 }
320 h := newHarness(t, simpleScript(lines...))
321 h.runner.Timing.LogFlush = 50 * time.Millisecond
322 h.srv.jobs = []*ermite.Job{testJob("j1")}
323 h.run(t)
324 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
325 h.srv.mu.Lock()
326 defer h.srv.mu.Unlock()
327 if len(h.srv.lines) != 2500 {
328 t.Fatalf("the server holds %d lines", len(h.srv.lines))
329 }
330 for i, l := range h.srv.lines {
331 if len(l) != 1+(i)%1000 {
332 t.Fatalf("line %d out of order or wrong", i)
333 }
334 }
335 var next int64
336 for _, r := range h.srv.appendReq {
337 if len(r.Lines) > 1000 {
338 t.Fatalf("batch of %d lines", len(r.Lines))
339 }
340 n := 0
341 for _, l := range r.Lines {
342 n += len(l.Text)
343 }
344 if n > 64*1024 {
345 t.Fatalf("batch of %d bytes", n)
346 }
347 if r.Offset != next {
348 t.Fatalf("offset %d, want %d", r.Offset, next)
349 }
350 next += int64(len(r.Lines))
351 }
352 if next != 2500 {
353 t.Fatalf("offsets end at %d", next)
354 }
355 // Batches never mix steps.
356 for _, r := range h.srv.appendReq {
357 if r.Step < 1 {
358 t.Fatalf("step %d", r.Step)
359 }
360 }
361}
362
363func TestLogStepsAreSeparateRequests(t *testing.T) {
364 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
365 for n := 1; n <= 3; n++ {
366 stepStart(c, n, "s")
367 logLineFrame(c, n, "a", false)
368 logLineFrame(c, n, "b", false)
369 stepDone(c, n, "s", cOK)
370 }
371 jobDone(c, cOK, nil, false)
372 })
373 h.runner.Timing.LogFlush = 50 * time.Millisecond
374 h.srv.jobs = []*ermite.Job{testJob("j1")}
375 h.run(t)
376 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
377 h.srv.mu.Lock()
378 defer h.srv.mu.Unlock()
379 if len(h.srv.appendReq) != 3 {
380 t.Fatalf("%d requests, want one per step", len(h.srv.appendReq))
381 }
382 for i, r := range h.srv.appendReq {
383 if r.Step != int32(i+1) || r.Offset != int64(2*i) || len(r.Lines) != 2 {
384 t.Errorf("request %d: step %d offset %d lines %d", i, r.Step, r.Offset, len(r.Lines))
385 }
386 }
387}
388
389func TestContinuedLinesAreForwarded(t *testing.T) {
390 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
391 stepStart(c, 1, "s")
392 logLineFrame(c, 1, "first", false)
393 logLineFrame(c, 1, "piece1", false)
394 logLineFrame(c, 1, "piece2", true)
395 stepDone(c, 1, "s", cOK)
396 jobDone(c, cOK, nil, false)
397 })
398 h.srv.jobs = []*ermite.Job{testJob("j1")}
399 h.run(t)
400 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
401 h.srv.mu.Lock()
402 defer h.srv.mu.Unlock()
403 if len(h.srv.cont) != 3 || h.srv.cont[0] || h.srv.cont[1] || !h.srv.cont[2] {
404 t.Fatalf("%v", h.srv.cont)
405 }
406}
407
408func TestAppendLogRetriesAreIdempotent(t *testing.T) {
409 h := newHarness(t, simpleScript("l1", "l2", "l3"))
410 h.srv.jobs = []*ermite.Job{testJob("j1")}
411 // The first AppendLog is applied but its response is lost; the second fails
412 // before being applied. The runner resends the same lines at the same offset.
413 h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
414 switch n {
415 case 1:
416 return true, connect.NewError(connect.CodeUnavailable, io.EOF)
417 case 2:
418 return false, connect.NewError(connect.CodeInternal, io.EOF)
419 }
420 return false, nil
421 }
422 h.run(t)
423 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
424 h.srv.mu.Lock()
425 defer h.srv.mu.Unlock()
426 if strings.Join(h.srv.lines, "|") != "l1|l2|l3" {
427 t.Fatalf("lines %q: a retry must neither lose nor duplicate", h.srv.lines)
428 }
429 if h.srv.appendN < 3 {
430 t.Fatalf("only %d calls", h.srv.appendN)
431 }
432 first := h.srv.appendReq[0]
433 for _, r := range h.srv.appendReq[1:3] {
434 if r.Offset != first.Offset || len(r.Lines) != len(first.Lines) {
435 t.Errorf("a retry changed the request: offset %d/%d lines %d/%d", r.Offset, first.Offset, len(r.Lines), len(first.Lines))
436 }
437 }
438}
439
440func TestStepUpdatesAreRetried(t *testing.T) {
441 h := newHarness(t, simpleScript("x"))
442 h.srv.jobs = []*ermite.Job{testJob("j1")}
443 h.run(t)
444 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
445 // UpdateJob snapshots carry the state at each step boundary.
446 h.srv.mu.Lock()
447 defer h.srv.mu.Unlock()
448 if u := h.srv.steps[2]; u == nil || u.State != ermite.StepState_STEP_STATE_COMPLETED {
449 t.Fatalf("%v", u)
450 }
451}
452
453func TestCompleteJobIsRetriedAndIdempotent(t *testing.T) {
454 h := newHarness(t, simpleScript("x"))
455 h.srv.jobs = []*ermite.Job{testJob("j1")}
456 h.srv.completeHook = func(n int) (bool, error) {
457 switch n {
458 case 1: // applied, response lost
459 return true, connect.NewError(connect.CodeUnavailable, io.EOF)
460 case 2: // not even reached
461 return false, connect.NewError(connect.CodeInternal, io.EOF)
462 }
463 return false, nil
464 }
465 h.run(t)
466 eventually(t, "three CompleteJob calls", func() bool { return h.srv.completeCount() == 3 })
467 time.Sleep(100 * time.Millisecond)
468 if n := h.srv.completeCount(); n != 3 {
469 t.Fatalf("%d calls", n)
470 }
471 h.srv.mu.Lock()
472 defer h.srv.mu.Unlock()
473 a, b := h.srv.completes[0], h.srv.completes[2]
474 if a.Conclusion != b.Conclusion || a.Outputs["out"] != b.Outputs["out"] || len(a.Steps) != len(b.Steps) {
475 t.Error("a retry must repeat the same result")
476 }
477}
478
479func TestJobEndedStopsEverything(t *testing.T) {
480 release := make(chan struct{})
481 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
482 stepStart(c, 1, "Set up job")
483 logLineFrame(c, 1, "before", false)
484 <-release
485 logLineFrame(c, 1, "after", false)
486 stepDone(c, 1, "Set up job", cOK)
487 jobDone(c, cOK, nil, false)
488 })
489 h.srv.jobs = []*ermite.Job{testJob("j1")}
490 h.run(t)
491 eventually(t, "first line", func() bool { return len(h.srv.snapshotLines()) == 1 })
492 // The server ends the job (lost runner, log limit, ...): every call now says so.
493 h.srv.mu.Lock()
494 h.srv.ended = &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED, Error: "ended by the server"}
495 h.srv.mu.Unlock()
496 h.srv.watchMsgs <- &ermite.WatchJobResponse{Msg: &ermite.WatchJobResponse_Ended{Ended: &ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_CANCELLED}}}
497 eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
498 close(release)
499 time.Sleep(300 * time.Millisecond)
500 if n := h.srv.completeCount(); n != 0 {
501 t.Fatalf("CompleteJob was called %d times for a job the server ended", n)
502 }
503 if got := h.srv.snapshotLines(); len(got) != 1 {
504 t.Fatalf("lines after the end: %q", got)
505 }
506}
507
508func TestJobEndedFromAppendLogAbandonsTheJob(t *testing.T) {
509 release := make(chan struct{})
510 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
511 stepStart(c, 1, "Set up job")
512 logLineFrame(c, 1, "a", false)
513 <-release
514 jobDone(c, cOK, nil, false)
515 })
516 h.srv.jobs = []*ermite.Job{testJob("j1")}
517 h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
518 return false, endedErr(&ermite.JobEnded{Conclusion: ermite.CiConclusion_CI_CONCLUSION_FAILURE, Error: "log limit reached"})
519 }
520 h.run(t)
521 eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
522 close(release)
523 time.Sleep(200 * time.Millisecond)
524 if n := h.srv.completeCount(); n != 0 {
525 t.Fatalf("CompleteJob was called %d times", n)
526 }
527}
528
529func TestUnauthenticatedJobCallAbandonsAtOnce(t *testing.T) {
530 release := make(chan struct{})
531 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
532 stepStart(c, 1, "Set up job")
533 <-release
534 jobDone(c, cOK, nil, false)
535 })
536 h.srv.jobs = []*ermite.Job{testJob("j1")}
537 h.srv.beatHook = func(n int) error {
538 if n >= 2 { // the assignment was replaced
539 return connect.NewError(connect.CodeUnauthenticated, io.EOF)
540 }
541 return nil
542 }
543 h.run(t)
544 eventually(t, "the sandbox to be stopped", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.stops == 1 })
545 close(release)
546 time.Sleep(200 * time.Millisecond)
547 if n := h.srv.completeCount(); n != 0 {
548 t.Fatalf("CompleteJob was called %d times", n)
549 }
550}
551
552func TestGracefulShutdownReportsRunnerShutdown(t *testing.T) {
553 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
554 stepStart(c, 1, "Set up job")
555 stepStart(c, 2, "Run sleep")
556 logLineFrame(c, 2, "working", false)
557 cn := <-cancels
558 if cn.Kind != execpb.CancelKind_CANCEL_KIND_SHUTDOWN {
559 t.Errorf("kind %v", cn.Kind)
560 }
561 stepDone(c, 2, "Run sleep", cCancel)
562 jobDone(c, cCancel, nil, true)
563 })
564 h.srv.jobs = []*ermite.Job{testJob("j1")}
565 stop := h.run(t)
566 eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
567 if err := stop(); err != nil {
568 t.Fatal(err)
569 }
570 h.srv.mu.Lock()
571 defer h.srv.mu.Unlock()
572 if len(h.srv.completes) != 1 {
573 t.Fatalf("%d CompleteJob calls", len(h.srv.completes))
574 }
575 if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !c.RunnerShutdown {
576 t.Fatalf("%+v", c)
577 }
578 if h.driver.stops != 1 {
579 t.Errorf("stops=%d", h.driver.stops)
580 }
581}
582
583func TestShutdownHardStopsAnExecutorThatIgnoresIt(t *testing.T) {
584 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
585 stepStart(c, 1, "Set up job")
586 stepStart(c, 2, "Run hang")
587 logLineFrame(c, 2, "hung", false)
588 <-cancels // received, ignored: keep running until the stream is cut
589 select {}
590 })
591 h.runner.Timing.ShutdownGrace = 300 * time.Millisecond
592 h.srv.jobs = []*ermite.Job{testJob("j1")}
593 stop := h.run(t)
594 eventually(t, "the job to run", func() bool { return len(h.srv.snapshotLines()) == 1 })
595 start := time.Now()
596 _ = stop()
597 if time.Since(start) > 5*time.Second {
598 t.Fatal("shutdown waited for an executor that does not answer")
599 }
600 h.srv.mu.Lock()
601 defer h.srv.mu.Unlock()
602 if len(h.srv.completes) != 1 || h.srv.completes[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED || !h.srv.completes[0].RunnerShutdown {
603 t.Fatalf("%+v", h.srv.completes)
604 }
605 // Steps that never completed are closed by the runner.
606 for _, u := range h.srv.completes[0].Steps {
607 if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_CANCELLED {
608 t.Errorf("step %d: %v/%v", u.Number, u.State, u.Conclusion)
609 }
610 }
611}
612
613func TestExecutorDyingFailsTheJob(t *testing.T) {
614 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
615 stepStart(c, 1, "Set up job")
616 logLineFrame(c, 1, "partial", false)
617 // the stream closes without a result
618 })
619 h.srv.jobs = []*ermite.Job{testJob("j1")}
620 h.run(t)
621 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
622 h.srv.mu.Lock()
623 defer h.srv.mu.Unlock()
624 c := h.srv.completes[0]
625 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || c.RunnerShutdown {
626 t.Fatalf("%+v", c)
627 }
628 if len(h.srv.lines) != 2 || !strings.HasPrefix(h.srv.lines[1], "##[error]The job's sandbox ended unexpectedly") {
629 t.Fatalf("lines %q", h.srv.lines)
630 }
631 if len(c.Steps) != 1 || c.Steps[0].State != ermite.StepState_STEP_STATE_COMPLETED || c.Steps[0].Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
632 t.Fatalf("%v", c.Steps)
633 }
634}
635
636func TestSandboxStartFailureFailsTheJobInStepOne(t *testing.T) {
637 h := newHarness(t, simpleScript("x"))
638 h.driver.startErr = errors.New("no kvm today")
639 h.srv.jobs = []*ermite.Job{testJob("j1")}
640 h.run(t)
641 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
642 h.srv.mu.Lock()
643 defer h.srv.mu.Unlock()
644 if c := h.srv.completes[0]; c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE {
645 t.Fatalf("%v", c.Conclusion)
646 }
647 if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "The sandbox did not start: no kvm today") {
648 t.Fatalf("%q", h.srv.lines)
649 }
650 if h.driver.stops != 1 {
651 t.Errorf("a sandbox that failed to start is still stopped: stops=%d", h.driver.stops)
652 }
653}
654
655func TestOutputsOverTheLimitFailTheJob(t *testing.T) {
656 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
657 stepStart(c, 1, "Set up job")
658 stepDone(c, 1, "Set up job", cOK)
659 // Names that ermite would refuse: the job fails without outputs.
660 jobDone(c, cOK, map[string]string{"1bad": "x", "fine": "y"}, false)
661 })
662 h.srv.jobs = []*ermite.Job{testJob("j1")}
663 h.run(t)
664 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
665 h.srv.mu.Lock()
666 defer h.srv.mu.Unlock()
667 c := h.srv.completes[0]
668 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || len(c.Outputs) != 0 {
669 t.Fatalf("%v %v", c.Conclusion, len(c.Outputs))
670 }
671 if len(h.srv.lines) != 1 || !strings.Contains(h.srv.lines[0], "is not valid") {
672 t.Fatalf("%q", h.srv.lines)
673 }
674}
675
676func TestCheckOutputs(t *testing.T) {
677 for _, tc := range []struct {
678 in map[string]string
679 want string
680 }{
681 {nil, ""},
682 {map[string]string{"a-b_c1": "v", "_x": ""}, ""},
683 {map[string]string{"1a": "v"}, "not valid"},
684 {map[string]string{"a b": "v"}, "not valid"},
685 {map[string]string{"": "v"}, "not valid"},
686 {map[string]string{strings.Repeat("a", 256): "v"}, "not valid"},
687 {map[string]string{"Foo": "1", "foo": "2"}, "differ only by case"},
688 {map[string]string{"a": strings.Repeat("x", 1<<20+1)}, "larger than 1 MiB"},
689 {map[string]string{"a": strings.Repeat("x", 600000), "b": strings.Repeat("x", 600000)}, "in total"},
690 } {
691 got := checkOutputs(tc.in)
692 if (tc.want == "") != (got == "") || !strings.Contains(got, tc.want) {
693 t.Errorf("%v: %q, want %q", tc.in, got, tc.want)
694 }
695 }
696}
697
698func TestLeakedSandboxStopsTakingJobsUntilReaped(t *testing.T) {
699 h := newHarness(t, simpleScript("x"))
700 h.driver.stopErr = sandbox.ErrLeaked
701 h.driver.reapErr = errors.New("still leaked")
702 h.runner.Timing.ReapRetry = 50 * time.Millisecond
703 h.srv.jobs = []*ermite.Job{testJob("j1"), testJob("j2")}
704 h.run(t)
705 eventually(t, "first job done", func() bool { return h.srv.completeCount() == 1 })
706 time.Sleep(400 * time.Millisecond)
707 if h.srv.completeCount() != 1 {
708 t.Fatal("a second job started while a sandbox is leaked and cannot be reaped")
709 }
710 // Once the reap works, jobs are taken again.
711 h.driver.mu.Lock()
712 h.driver.reapErr = nil
713 h.driver.stopErr = nil
714 h.driver.mu.Unlock()
715 eventually(t, "the second job", func() bool { return h.srv.completeCount() == 2 })
716}
717
718func TestReapAtStartAndIsolation(t *testing.T) {
719 h := newHarness(t, simpleScript("x"))
720 h.run(t)
721 eventually(t, "reap", func() bool { h.driver.mu.Lock(); defer h.driver.mu.Unlock(); return h.driver.reaps >= 1 })
722 for in, want := range map[sandbox.Isolation]ermite.RunnerIsolation{
723 sandbox.IsolationVM: ermite.RunnerIsolation_RUNNER_ISOLATION_VM,
724 sandbox.IsolationContainerSysbox: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX,
725 sandbox.IsolationContainerPrivileged: ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_PRIVILEGED,
726 "other": ermite.RunnerIsolation_RUNNER_ISOLATION_UNSPECIFIED,
727 } {
728 if got := Isolation(in); got != want {
729 t.Errorf("%s: %v", in, got)
730 }
731 }
732}
733
734func TestRegisterStoresTheToken(t *testing.T) {
735 srv := newFakeServer()
736 url := srv.start(t)
737 dir := t.TempDir() + "/state"
738 cfg := Config{URL: url, Name: "r1", Labels: []string{"ubuntu-latest"}, StateDir: dir, Version: "v1"}
739 st, err := Register(context.Background(), cfg, NewClient(url, "reg-token", http.DefaultTransport), sandbox.IsolationContainerSysbox, true)
740 if err != nil {
741 t.Fatal(err)
742 }
743 if st.RunnerToken != "ermiterun_new" || st.RunnerID != "rid1" {
744 t.Fatalf("%+v", st)
745 }
746 req := srv.registers[0]
747 if req.Name != "r1" || req.Isolation != ermite.RunnerIsolation_RUNNER_ISOLATION_CONTAINER_SYSBOX || req.Os != "linux" || (req.Arch != "x64" && req.Arch != "arm64") || !req.Replace || req.Version != "v1" || len(req.Labels) != 1 {
748 t.Fatalf("%v", req)
749 }
750 fi, err := os.Stat(StateFile(dir))
751 if err != nil || fi.Mode().Perm() != 0o600 {
752 t.Fatalf("%v %v", fi, err)
753 }
754 got, err := LoadState(dir)
755 if err != nil || got != st {
756 t.Fatalf("%+v %v", got, err)
757 }
758 // Anyone can read a state file with a loose mode: refuse it.
759 if err := os.Chmod(StateFile(dir), 0o644); err != nil {
760 t.Fatal(err)
761 }
762 if _, err := LoadState(dir); err == nil || !strings.Contains(err.Error(), "accessible by other users") {
763 t.Fatalf("%v", err)
764 }
765 if _, err := LoadState(t.TempDir()); !errors.Is(err, ErrNotRegistered) {
766 t.Fatalf("%v", err)
767 }
768 // A refused registration stores nothing.
769 dir2 := t.TempDir()
770 cfg.StateDir = dir2
771 if _, err := Register(context.Background(), cfg, NewClient(url, "bad", http.DefaultTransport), sandbox.IsolationVM, false); err == nil {
772 t.Fatal("registration with a bad token succeeded")
773 }
774 if _, err := os.Stat(StateFile(dir2)); err == nil {
775 t.Fatal("a state file was written")
776 }
777}
778
779func TestLoadConfig(t *testing.T) {
780 env := map[string]string{EnvURL: " https://ermite.example ", EnvLabels: "ubuntu-latest, Ubuntu-24.04 ,ubuntu-latest,"}
781 c, err := LoadConfig(func(k string) string { return env[k] }, "my host.local", "/state", 2, "v1")
782 if err != nil {
783 t.Fatal(err)
784 }
785 if c.URL != "https://ermite.example" || c.Name != "my-host.local" || strings.Join(c.Labels, ",") != "ubuntu-latest,Ubuntu-24.04" || c.Capacity != 2 {
786 t.Fatalf("%+v", c)
787 }
788 if c.Environment != "github-hosted" {
789 t.Errorf("default environment %q", c.Environment)
790 }
791 env[EnvEnvironment] = " Self-Hosted "
792 if c, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err != nil || c.Environment != "self-hosted" {
793 t.Errorf("self-hosted: %q, %v", c.Environment, err)
794 }
795 env[EnvEnvironment] = "cloud"
796 if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil || !strings.Contains(err.Error(), EnvEnvironment) {
797 t.Errorf("an invalid environment was accepted: %v", err)
798 }
799 delete(env, EnvEnvironment)
800 env[EnvName] = "bad name"
801 if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
802 t.Error("an invalid name was accepted")
803 }
804 env[EnvName] = "ok"
805 env[EnvLabels] = "good,bad label"
806 if _, err := LoadConfig(func(k string) string { return env[k] }, "h", "/s", 1, ""); err == nil {
807 t.Error("an invalid label was accepted")
808 }
809}
810
811var _ = ermiteconnect.RunnerServiceClient(nil)
812
813// The job timeout counts from the receipt of the job, and the host enforces it
814// even when the executor does not: the sandbox is stopped and the job ends
815// timed_out.
816func TestHostEnforcesTheJobTimeout(t *testing.T) {
817 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, cancels <-chan *execpb.Cancel) {
818 stepStart(c, 1, "Set up job")
819 logLineFrame(c, 1, "stuck", false)
820 cn := <-cancels
821 if cn.Kind != execpb.CancelKind_CANCEL_KIND_TIMEOUT {
822 t.Errorf("kind %v", cn.Kind)
823 }
824 select {} // the executor ignores the cancel
825 })
826 h.runner.Timing.TimeoutGrace = 200 * time.Millisecond
827 job := testJob("j1")
828 job.TimeoutMinutes = 1
829 h.runner.Job = func(tok string) ermiteconnect.RunnerServiceClient {
830 return NewClient(h.url, tok, http.DefaultTransport)
831 }
832 h.runner.now = func() time.Time { return time.Now().Add(-59*time.Second - 700*time.Millisecond) } // received 59.7 s ago
833 h.srv.jobs = []*ermite.Job{job}
834 start := time.Now()
835 h.run(t)
836 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
837 if d := time.Since(start); d > 5*time.Second {
838 t.Fatalf("took %v", d)
839 }
840 h.srv.mu.Lock()
841 defer h.srv.mu.Unlock()
842 c := h.srv.completes[0]
843 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_TIMED_OUT || c.RunnerShutdown {
844 t.Fatalf("%v", c.Conclusion)
845 }
846 for _, u := range c.Steps {
847 if u.State != ermite.StepState_STEP_STATE_COMPLETED {
848 t.Errorf("step %d not completed", u.Number)
849 }
850 }
851}
852
853// The remaining time, not the whole timeout, goes to the executor.
854func TestRunCarriesTheRemainingDeadline(t *testing.T) {
855 var spec *execpb.JobSpec
856 inner := simpleScript("x")
857 h := newHarness(t, func(t *testing.T, c *execstream.Conn, run *execpb.Run, cn <-chan *execpb.Cancel) {
858 spec = run.Job
859 inner(t, c, run, cn)
860 })
861 h.runner.now = func() time.Time { return time.Now().Add(-3 * time.Minute) }
862 h.srv.jobs = []*ermite.Job{testJob("j1")} // 5 minutes
863 h.run(t)
864 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
865 if spec == nil || spec.TimeoutMinutes != 5 || spec.DeadlineMs < 110_000 || spec.DeadlineMs > 120_000 {
866 t.Fatalf("%v", spec)
867 }
868}
869
870// ---- the guest is untrusted: step numbers and lines are checked (M3)
871
872// hostile runs a job whose executor sends the frames of body, and checks that
873// the job fails, the sandbox is destroyed, the server never saw a violation of
874// its contract, and a line in the log names the reason.
875func hostile(t *testing.T, wantReason string, body script) *harness {
876 t.Helper()
877 h := newHarness(t, func(t *testing.T, c *execstream.Conn, run *execpb.Run, cn <-chan *execpb.Cancel) {
878 body(t, c, run, cn)
879 // A real executor would go on; this one waits for the host to cut it.
880 <-cn
881 })
882 h.srv.jobs = []*ermite.Job{testJob("j1")}
883 h.run(t)
884 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
885 h.srv.mu.Lock()
886 defer h.srv.mu.Unlock()
887 c := h.srv.completes[0]
888 if c.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || len(c.Outputs) != 0 {
889 t.Fatalf("conclusion %v outputs %v", c.Conclusion, c.Outputs)
890 }
891 found := false
892 for _, l := range h.srv.lines {
893 if strings.Contains(l, "broke the protocol") && strings.Contains(l, wantReason) {
894 found = true
895 }
896 }
897 if !found {
898 t.Fatalf("no error line mentioning %q in %q", wantReason, h.srv.lines)
899 }
900 for _, u := range c.Steps {
901 if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.Number < 1 || u.Number > 1000 {
902 t.Errorf("invalid final step %v", u)
903 }
904 }
905 if h.driver.stops != 1 {
906 t.Errorf("the sandbox was not destroyed: stops=%d", h.driver.stops)
907 }
908 return h
909}
910
911func TestGuestStepNumberZeroFailsTheJob(t *testing.T) {
912 hostile(t, "outside 1..1000", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
913 stepStart(c, 1, "Set up job")
914 stepStart(c, 0, "zero")
915 })
916}
917
918func TestGuestStepNumberOverTheLimitFailsTheJob(t *testing.T) {
919 hostile(t, "outside 1..1000", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
920 stepStart(c, 1, "Set up job")
921 stepStart(c, 1001, "far")
922 })
923}
924
925func TestGuestStepGoingBackFailsTheJob(t *testing.T) {
926 hostile(t, "started after step 3", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
927 stepStart(c, 1, "a")
928 stepStart(c, 3, "c")
929 stepStart(c, 2, "b")
930 })
931}
932
933func TestGuestStepRestartedFailsTheJob(t *testing.T) {
934 hostile(t, "started after step 1", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
935 stepStart(c, 1, "a")
936 stepDone(c, 1, "a", cOK)
937 stepStart(c, 1, "again")
938 })
939}
940
941func TestGuestCompletedStepIsNeverTouchedAgain(t *testing.T) {
942 h := hostile(t, "completed twice", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
943 stepStart(c, 1, "a")
944 stepDone(c, 1, "a", cOK)
945 stepDone(c, 1, "a", cFail)
946 })
947 if u := h.srv.Result("j1").Steps[1]; u == nil || u.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS {
948 t.Errorf("a completed step was changed: %v", u.Conclusion)
949 }
950}
951
952func TestGuestCompletingAnUnknownStepFailsTheJob(t *testing.T) {
953 hostile(t, "without having started", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
954 stepStart(c, 1, "a")
955 stepDone(c, 7, "ghost", cOK)
956 })
957}
958
959func TestGuestLogForAnUnstartedOrCompletedStepFailsTheJob(t *testing.T) {
960 hostile(t, "has not started", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
961 stepStart(c, 1, "a")
962 logLineFrame(c, 5, "x", false)
963 })
964 hostile(t, "which is completed", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
965 stepStart(c, 1, "a")
966 stepDone(c, 1, "a", cOK)
967 logLineFrame(c, 1, "late", false)
968 })
969}
970
971func TestGuestLogGoingBackAStepFailsTheJob(t *testing.T) {
972 hostile(t, "after one for step 2", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
973 stepStart(c, 1, "a")
974 stepStart(c, 2, "b")
975 logLineFrame(c, 2, "x", false)
976 logLineFrame(c, 1, "y", false)
977 })
978}
979
980func TestGuestOversizedLineFailsTheJob(t *testing.T) {
981 hostile(t, "bytes (the limit is 65536)", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
982 stepStart(c, 1, "a")
983 logLineFrame(c, 1, strings.Repeat("x", 64*1024+1), false)
984 })
985}
986
987// Text that the server would refuse is repaired, not forwarded: a line feed
988// makes more lines and long names are cut. (Invalid UTF-8 cannot reach the host:
989// proto3 strings are validated when the frame is decoded, which fails the
990// session; the executor replaces invalid bytes before it sends.)
991func TestGuestTextIsMadeValidForTheServer(t *testing.T) {
992 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
993 stepStart(c, 1, strings.Repeat("é", 800)) // 1,600 bytes
994 logLineFrame(c, 1, "two\nlines", false)
995 stepDone(c, 1, "", cOK)
996 jobDone(c, cOK, nil, false)
997 })
998 h.srv.jobs = []*ermite.Job{testJob("j1")}
999 h.run(t)
1000 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
1001 h.srv.mu.Lock()
1002 defer h.srv.mu.Unlock()
1003 if strings.Join(h.srv.lines, "|") != "two|lines" {
1004 t.Fatalf("%q", h.srv.lines)
1005 }
1006 if n := h.srv.steps[1].Name; len(n) > 1024 || !utf8.ValidString(n) {
1007 t.Fatalf("name of %d bytes", len(n))
1008 }
1009}
1010
1011// A guest that sends steps without end cannot grow the memory of the host.
1012func TestGuestFloodIsBounded(t *testing.T) {
1013 h := hostile(t, "outside 1..1000", func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
1014 for n := 1; n <= 1100; n++ {
1015 stepStart(c, n, "s")
1016 stepDone(c, n, "s", cOK)
1017 }
1018 })
1019 _ = h
1020}
1021
1022// The server's line count is the truth (m6): when the answer to an AppendLog
1023// says the server holds more lines than the request accounts for, the next
1024// offset starts from the server's count and no gap or repeat is created.
1025func TestAppendLogResyncsToTheServersCount(t *testing.T) {
1026 h := newHarness(t, func(_ *testing.T, c *execstream.Conn, _ *execpb.Run, _ <-chan *execpb.Cancel) {
1027 stepStart(c, 1, "a")
1028 logLineFrame(c, 1, "l1", false)
1029 time.Sleep(150 * time.Millisecond) // a first request of its own
1030 logLineFrame(c, 1, "l2", false)
1031 logLineFrame(c, 1, "l3", false)
1032 stepDone(c, 1, "a", cOK)
1033 jobDone(c, cOK, nil, false)
1034 })
1035 h.runner.Timing.LogFlush = 30 * time.Millisecond
1036 h.srv.jobs = []*ermite.Job{testJob("j1")}
1037 // A line the runner never sent appears in the server's log (a retry whose
1038 // answer was lost, in a real server): the runner must continue after it.
1039 h.srv.mu.Lock()
1040 h.srv.appendHook = func(n int, r *ermite.AppendLogRequest) (bool, error) {
1041 if n == 1 {
1042 _, _ = h.srv.store.Apply(&ermite.AppendLogRequest{JobId: r.JobId, Step: 1, Offset: 0, Lines: []*ermite.LogLine{
1043 {Time: timestamppb.Now(), Text: "planted"}, {Time: timestamppb.Now(), Text: "planted2"}}})
1044 return false, nil
1045 }
1046 return false, nil
1047 }
1048 h.srv.mu.Unlock()
1049 h.run(t)
1050 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
1051 h.srv.mu.Lock()
1052 defer h.srv.mu.Unlock()
1053 // The first request (offset 0, "l1") was skipped by the server because the
1054 // planted lines already took offsets 0 and 1; the runner learns it holds 2
1055 // lines (it expected 1), and goes on at offset 2: no gap, no repeat.
1056 if len(h.srv.lines) < 3 || h.srv.lines[0] != "planted" || h.srv.lines[len(h.srv.lines)-1] != "l3" {
1057 t.Fatalf("%q", h.srv.lines)
1058 }
1059 if !strings.Contains(h.logs.String(), "continuing from its count") {
1060 t.Fatalf("no warning about the resync:\n%s", h.logs.String())
1061 }
1062}
1063
1064// ---- M4: the server refuses the result itself
1065
1066func TestCompleteJobRefusedIsRetriedAsAFailureWithoutOutputs(t *testing.T) {
1067 h := newHarness(t, simpleScript("hello"))
1068 h.srv.jobs = []*ermite.Job{testJob("j1")}
1069 h.srv.completeHook = func(n int) (bool, error) {
1070 if n == 1 { // e.g. an output the server's rules refuse
1071 return false, connect.NewError(connect.CodeInvalidArgument, io.EOF)
1072 }
1073 return false, nil
1074 }
1075 h.run(t)
1076 eventually(t, "two CompleteJob calls", func() bool { return h.srv.completeCount() == 2 })
1077 time.Sleep(100 * time.Millisecond)
1078 h.srv.mu.Lock()
1079 defer h.srv.mu.Unlock()
1080 if len(h.srv.completes) != 2 {
1081 t.Fatalf("%d calls", len(h.srv.completes))
1082 }
1083 first, second := h.srv.completes[0], h.srv.completes[1]
1084 if first.Conclusion != ermite.CiConclusion_CI_CONCLUSION_SUCCESS || first.Outputs["out"] != "v" {
1085 t.Fatalf("first: %v %v", first.Conclusion, first.Outputs)
1086 }
1087 if second.Conclusion != ermite.CiConclusion_CI_CONCLUSION_FAILURE || len(second.Outputs) != 0 || second.RunnerShutdown {
1088 t.Fatalf("second: %v %v", second.Conclusion, second.Outputs)
1089 }
1090 if len(second.Steps) != 3 {
1091 t.Fatalf("steps %v", second.Steps)
1092 }
1093 for _, u := range second.Steps {
1094 if u.State != ermite.StepState_STEP_STATE_COMPLETED || u.StartedAt == nil || u.CompletedAt == nil || u.CompletedAt.AsTime().Before(u.StartedAt.AsTime()) {
1095 t.Errorf("invalid step %v", u)
1096 }
1097 }
1098 last := h.srv.lines[len(h.srv.lines)-1]
1099 if !strings.HasPrefix(last, "##[error]The runner could not report the result of the job: invalid_argument") {
1100 t.Fatalf("last line %q", last)
1101 }
1102}
1103
1104func TestCompleteJobRefusedTwiceGivesUp(t *testing.T) {
1105 h := newHarness(t, simpleScript("x"))
1106 h.srv.jobs = []*ermite.Job{testJob("j1")}
1107 h.srv.completeHook = func(int) (bool, error) {
1108 return false, connect.NewError(connect.CodeFailedPrecondition, io.EOF) // not a JobEnded detail
1109 }
1110 h.run(t)
1111 eventually(t, "two calls", func() bool { return h.srv.completeCount() == 2 })
1112 time.Sleep(200 * time.Millisecond)
1113 if n := h.srv.completeCount(); n != 2 {
1114 t.Fatalf("%d calls: one retry only", n)
1115 }
1116 if !strings.Contains(h.logs.String(), `"level":"error"`) || !strings.Contains(h.logs.String(), "refused again") {
1117 t.Fatalf("no error log:\n%s", h.logs.String())
1118 }
1119}
1120
1121// ---- G1: structured logs with the job's fields, warn for retries, error for refusals
1122
1123func TestLogsCarryTheJobFieldsAndLevels(t *testing.T) {
1124 h := newHarness(t, simpleScript("x"))
1125 h.srv.jobs = []*ermite.Job{testJob("j1")}
1126 h.srv.appendHook = func(n int, _ *ermite.AppendLogRequest) (bool, error) {
1127 if n == 1 {
1128 return false, connect.NewError(connect.CodeUnavailable, io.EOF)
1129 }
1130 return false, nil
1131 }
1132 h.srv.beatHook = func(n int) error { return nil }
1133 h.run(t)
1134 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
1135 var retry, assigned map[string]any
1136 for _, l := range strings.Split(h.logs.String(), "\n") {
1137 if l == "" {
1138 continue
1139 }
1140 var m map[string]any
1141 if err := json.Unmarshal([]byte(l), &m); err != nil {
1142 t.Fatalf("not JSON: %q", l)
1143 }
1144 switch m["message"] {
1145 case "call failed, retrying":
1146 retry = m
1147 case "job assigned":
1148 assigned = m
1149 }
1150 }
1151 for name, m := range map[string]map[string]any{"retry": retry, "assigned": assigned} {
1152 if m == nil {
1153 t.Fatalf("no %s event in\n%s", name, h.logs.String())
1154 }
1155 if m["job_id"] != "j1" || m["run_id"] != float64(7) || m["attempt"] != float64(1) {
1156 t.Errorf("%s event lacks the job fields: %v", name, m)
1157 }
1158 }
1159 if retry["level"] != "warn" || retry["call"] != "AppendLog" {
1160 t.Errorf("a retry is a warning: %v", retry)
1161 }
1162}
1163
1164// The sizes of a sandbox come from ERMITE_RUNNER_JOB_CPUS, _JOB_MEMORY and
1165// _JOB_DISK (sandbox.Config.Resolve, applied by the driver at Start): the host
1166// asks for none, so that nothing in a job can change them.
1167func TestSandboxSizesAreLeftToTheDriversConfiguration(t *testing.T) {
1168 h := newHarness(t, simpleScript("x"))
1169 h.srv.jobs = []*ermite.Job{testJob("j1")}
1170 h.run(t)
1171 eventually(t, "CompleteJob", func() bool { return h.srv.completeCount() == 1 })
1172 h.driver.mu.Lock()
1173 defer h.driver.mu.Unlock()
1174 if len(h.driver.specs) != 1 {
1175 t.Fatalf("%v", h.driver.specs)
1176 }
1177 if s := h.driver.specs[0]; s.JobID != "j1" || s.CPUs != 0 || s.MemoryMiB != 0 || s.DiskMiB != 0 {
1178 t.Fatalf("%+v", s)
1179 }
1180 // And the configuration fills them (the real rule, in package sandbox).
1181 cfg, err := sandbox.LoadConfig(func(k string) string {
1182 return map[string]string{sandbox.EnvJobCPUs: "3", sandbox.EnvJobMemory: "2G", sandbox.EnvJobDisk: "10G"}[k]
1183 })
1184 if err != nil {
1185 t.Fatal(err)
1186 }
1187 if got := cfg.Resolve(h.driver.specs[0]); got.CPUs != 3 || got.MemoryMiB != 2048 || got.DiskMiB != 10240 {
1188 t.Fatalf("%+v", got)
1189 }
1190}
1191
1192// ---- M5: shutdown
1193
1194// A job whose acknowledgement is pending when the runner shuts down is left to
1195// the server (it requeues it after 30 s): the sandbox is never started, no
1196// result is reported, and Run returns promptly.
1197func TestShutdownBeforeTheAcknowledgementLeavesTheJobToTheServer(t *testing.T) {
1198 h := newHarness(t, simpleScript("x"))
1199 h.srv.jobs = []*ermite.Job{testJob("j1")}
1200 h.srv.beatHook = func(int) error { return connect.NewError(connect.CodeUnavailable, io.EOF) } // never acknowledged
1201 stop := h.run(t)
1202 eventually(t, "a failed acknowledgement", func() bool { h.srv.mu.Lock(); defer h.srv.mu.Unlock(); return len(h.srv.beats) >= 2 })
1203 start := time.Now()
1204 if err := stop(); err != nil {
1205 t.Fatal(err)
1206 }
1207 if time.Since(start) > 3*time.Second {
1208 t.Fatalf("Run took %v to return", time.Since(start))
1209 }
1210 h.driver.mu.Lock()
1211 defer h.driver.mu.Unlock()
1212 if h.driver.news != 0 || h.driver.started != 0 {
1213 t.Fatalf("a sandbox was created for a job that was never acknowledged: news=%d started=%d", h.driver.news, h.driver.started)
1214 }
1215 if n := h.srv.completeCount(); n != 0 {
1216 t.Fatalf("CompleteJob was called %d times", n)
1217 }
1218 if !strings.Contains(h.logs.String(), "before the job was acknowledged") {
1219 t.Fatalf("no explanation in the log:\n%s", h.logs.String())
1220 }
1221}
1222
1223// The graceful shutdown of the defaults fits the time the unit gives.
1224func TestShutdownBudgetFitsTheUnit(t *testing.T) {
1225 budget := Timing{}.ShutdownBudget()
1226 if budget != 90*time.Second {
1227 t.Fatalf("the budget is %v, README and contrib/ermite-runner.service say 90 s", budget)
1228 }
1229 unit, err := os.ReadFile("../../contrib/ermite-runner.service")
1230 if err != nil {
1231 t.Skip(err)
1232 }
1233 m := regexp.MustCompile(`(?m)^TimeoutStopSec=(\d+)$`).FindStringSubmatch(string(unit))
1234 if m == nil {
1235 t.Fatal("no TimeoutStopSec in the unit")
1236 }
1237 secs, _ := strconv.Atoi(m[1])
1238 if time.Duration(secs)*time.Second < budget+15*time.Second {
1239 t.Fatalf("TimeoutStopSec=%d does not leave 15 s of margin over the %v budget", secs, budget)
1240 }
1241}
1242
1243// ---- m7: the state directory is private
1244
1245func TestSaveStateMakesAnExistingDirectoryPrivate(t *testing.T) {
1246 dir := filepath.Join(t.TempDir(), "state")
1247 if err := os.Mkdir(dir, 0o755); err != nil {
1248 t.Fatal(err)
1249 }
1250 if err := SaveState(dir, State{URL: "https://x", RunnerID: "r", Name: "n", RunnerToken: "ermiterun_t"}); err != nil {
1251 t.Fatal(err)
1252 }
1253 fi, err := os.Stat(dir)
1254 if err != nil || fi.Mode().Perm() != 0o700 {
1255 t.Fatalf("%v %v", fi, err)
1256 }
1257 if fi, _ := os.Stat(StateFile(dir)); fi.Mode().Perm() != 0o600 {
1258 t.Fatalf("%v", fi.Mode())
1259 }
1260}
1261
1262// mountNamespaceOptions are the [Service] options that give a unit a mount
1263// namespace of its own. The unit must set none of them: in container mode the
1264// runner loop-mounts the scratch disk of a job; the Docker daemon, in the host's
1265// namespace, binds it into the container, and a mount made inside a private mount
1266// namespace of the unit is invisible to it (GO-001). Besides the options that say
1267// so (PrivateTmp, ProtectHome, ProtectSystem...), systemd builds a mount
1268// namespace for ProtectKernelModules and ProtectKernelLogs too.
1269var mountNamespaceOptions = []string{
1270 "PrivateTmp", "PrivateDevices", "PrivateMounts", "PrivateUsers", "ProtectHome", "ProtectSystem",
1271 "ProtectKernelModules", "ProtectKernelLogs", "ProtectKernelTunables", "ProtectControlGroups",
1272 "ProtectProc", "ProcSubset", "ProtectHostname", "ReadOnlyPaths", "ReadWritePaths", "InaccessiblePaths",
1273 "ExecPaths", "NoExecPaths", "TemporaryFileSystem", "BindPaths", "BindReadOnlyPaths", "MountFlags",
1274 "MountAPIVFS", "RootDirectory", "RootImage", "RootEphemeral", "BindLogSockets", "LogNamespace",
1275 "PrivateIPC", "PrivatePIDs", "MountImages", "ExtensionImages", "ExtensionDirectories",
1276}
1277
1278// unitMountNamespaceOptions returns the lines of the [Service] section of a unit
1279// that set one of mountNamespaceOptions. Comments, which name the options on
1280// purpose, and other sections are ignored.
1281func unitMountNamespaceOptions(unit string) []string {
1282 var found []string
1283 inService := false
1284 for _, line := range strings.Split(unit, "\n") {
1285 line = strings.TrimSpace(line)
1286 if strings.HasPrefix(line, "[") {
1287 inService = line == "[Service]"
1288 continue
1289 }
1290 if !inService || line == "" || strings.HasPrefix(line, "#") {
1291 continue
1292 }
1293 key, _, _ := strings.Cut(line, "=")
1294 for _, f := range mountNamespaceOptions {
1295 if strings.EqualFold(strings.TrimSpace(key), f) {
1296 found = append(found, line)
1297 }
1298 }
1299 }
1300 return found
1301}
1302
1303func TestUnitHasNoMountNamespaceOptions(t *testing.T) {
1304 unit, err := os.ReadFile("../../contrib/ermite-runner.service")
1305 if err != nil {
1306 t.Skip(err)
1307 }
1308 if found := unitMountNamespaceOptions(string(unit)); len(found) > 0 {
1309 t.Fatalf("the unit sets options that give it a mount namespace of its own, which hides the loop mounts from Docker: %q", found)
1310 }
1311 // The hardening that stays must still be there.
1312 for _, want := range []string{"NoNewPrivileges=yes", "RestrictSUIDSGID=yes", "ProtectClock=yes", "RestrictAddressFamilies="} {
1313 if !strings.Contains(string(unit), "\n"+want) {
1314 t.Errorf("the unit lost %s", want)
1315 }
1316 }
1317}
1318
1319// The check sees the options in [Service] (also with spaces and another case),
1320// and ignores comments and other sections.
1321func TestUnitMountNamespaceCheckSeesOptions(t *testing.T) {
1322 unit := "[Unit]\nPrivateTmp=yes\n[Service]\nPrivateTmp=yes\n protecthome = read-only\nProtectKernelModules=yes\n# ProtectSystem=strict\nNoNewPrivileges=yes\n[Install]\nProtectSystem=yes\n"
1323 got := unitMountNamespaceOptions(unit)
1324 want := []string{"PrivateTmp=yes", "protecthome = read-only", "ProtectKernelModules=yes"}
1325 if strings.Join(got, "|") != strings.Join(want, "|") {
1326 t.Fatalf("%q", got)
1327 }
1328}