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