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