detached ecb0362

feat(sandbox/container): add the Docker container driver with label-based reaping

commit
ecb03620040be1053832b1536a492f50eec3f2d5
author
brice · · 2026-10-12 15:35 UTC
parent
82488a8
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
5 files · +1447 -0
Unified
UnifiedSplit
A internal/sandbox/container/container_test.go +556 -0
@@ -0,0 +1,556 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "net"
10+ "net/netip"
11+ "os"
12+ "path/filepath"
13+ "slices"
14+ "strings"
15+ "sync"
16+ "testing"
17+ "time"
18+
19+ "bnema.dev/ermite-runner/internal/execstream"
20+ "bnema.dev/ermite-runner/internal/execstream/execpb"
21+ "bnema.dev/ermite-runner/internal/sandbox"
22+ "bnema.dev/ermite-runner/internal/sandbox/sandboxtest"
23+)
24+
25+// fakeEngine plays the Docker daemon. When a container "starts", it plays the
26+// executor inside it by connecting to the bind-mounted socket.
27+type fakeEngine struct {
28+ mu sync.Mutex
29+ info daemonInfo
30+ infoErr error
31+ imageErr error
32+ networks map[string]networkSpec
33+ volumes map[string]map[string]string
34+ containers map[string]containerSpec
35+ failStart error
36+ noConnect bool
37+ exitOnRun bool
38+ removeErr error
39+ waiters map[string]chan exitResult
40+ seq int
41+ pulled []string
42+}
43+
44+func newFake() *fakeEngine {
45+ return &fakeEngine{
46+ info: daemonInfo{OSType: "linux", Runtimes: []string{"runc"}},
47+ networks: map[string]networkSpec{},
48+ volumes: map[string]map[string]string{},
49+ containers: map[string]containerSpec{},
50+ waiters: map[string]chan exitResult{},
51+ }
52+}
53+
54+func (f *fakeEngine) Info(context.Context) (daemonInfo, error) { return f.info, f.infoErr }
55+func (f *fakeEngine) EnsureImage(_ context.Context, ref string) error {
56+ f.pulled = append(f.pulled, ref)
57+ return f.imageErr
58+}
59+func (f *fakeEngine) CreateNetwork(_ context.Context, s networkSpec) (string, error) {
60+ f.mu.Lock()
61+ defer f.mu.Unlock()
62+ f.seq++
63+ id := fmt.Sprintf("net%d", f.seq)
64+ f.networks[id] = s
65+ return id, nil
66+}
67+func (f *fakeEngine) RemoveNetwork(_ context.Context, id string) error {
68+ f.mu.Lock()
69+ defer f.mu.Unlock()
70+ delete(f.networks, id)
71+ return nil
72+}
73+func (f *fakeEngine) ListNetworks(context.Context, string) ([]string, error) {
74+ f.mu.Lock()
75+ defer f.mu.Unlock()
76+ var out []string
77+ for id, n := range f.networks {
78+ if _, ok := n.Labels[sandbox.LabelJob]; ok {
79+ out = append(out, id)
80+ }
81+ }
82+ return out, nil
83+}
84+func (f *fakeEngine) CreateVolume(_ context.Context, name string, labels map[string]string) error {
85+ f.mu.Lock()
86+ defer f.mu.Unlock()
87+ f.volumes[name] = labels
88+ return nil
89+}
90+func (f *fakeEngine) RemoveVolume(_ context.Context, name string) error {
91+ f.mu.Lock()
92+ defer f.mu.Unlock()
93+ delete(f.volumes, name)
94+ return nil
95+}
96+func (f *fakeEngine) ListVolumes(context.Context, string) ([]string, error) {
97+ f.mu.Lock()
98+ defer f.mu.Unlock()
99+ var out []string
100+ for n := range f.volumes {
101+ out = append(out, n)
102+ }
103+ return out, nil
104+}
105+func (f *fakeEngine) CreateContainer(_ context.Context, s containerSpec) (string, error) {
106+ f.mu.Lock()
107+ defer f.mu.Unlock()
108+ f.seq++
109+ id := fmt.Sprintf("ctr%d", f.seq)
110+ f.containers[id] = s
111+ f.waiters[id] = make(chan exitResult, 1)
112+ return id, nil
113+}
114+func (f *fakeEngine) StartContainer(_ context.Context, id string) error {
115+ f.mu.Lock()
116+ spec := f.containers[id]
117+ fail, noConnect, exitOnRun := f.failStart, f.noConnect, f.exitOnRun
118+ w := f.waiters[id]
119+ f.mu.Unlock()
120+ if fail != nil {
121+ return fail
122+ }
123+ if exitOnRun {
124+ w <- exitResult{Code: 125}
125+ return nil
126+ }
127+ if noConnect {
128+ return nil
129+ }
130+ // The host side of the bind mount is the source of the run-dir mount.
131+ var sock string
132+ for _, b := range spec.Binds {
133+ if b.Target == sandbox.ContainerRunDir {
134+ sock = filepath.Join(b.Source, sandbox.ContainerSocket)
135+ }
136+ }
137+ go func() {
138+ c, err := net.Dial("unix", sock)
139+ if err != nil {
140+ return
141+ }
142+ defer func() { _ = c.Close() }()
143+ g := execstream.NewConn(c)
144+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Sandbox: "container"}}})
145+ if _, err := g.Recv(); err != nil {
146+ return
147+ }
148+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{Data: []byte("hello from fake container\n")}}})
149+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})
150+ _, _ = g.Recv()
151+ }()
152+ return nil
153+}
154+func (f *fakeEngine) WaitExit(_ context.Context, id string) <-chan exitResult {
155+ f.mu.Lock()
156+ defer f.mu.Unlock()
157+ return f.waiters[id]
158+}
159+func (f *fakeEngine) RemoveContainer(_ context.Context, id string) error {
160+ f.mu.Lock()
161+ defer f.mu.Unlock()
162+ if f.removeErr != nil {
163+ return f.removeErr
164+ }
165+ delete(f.containers, id)
166+ return nil
167+}
168+func (f *fakeEngine) ListContainers(context.Context, string) ([]string, error) {
169+ f.mu.Lock()
170+ defer f.mu.Unlock()
171+ var out []string
172+ for id := range f.containers {
173+ out = append(out, id)
174+ }
175+ return out, nil
176+}
177+func (f *fakeEngine) LogsTail(context.Context, string, int) (string, error) {
178+ return "fake container log line", nil
179+}
180+func (f *fakeEngine) Close() error { return nil }
181+
182+func (f *fakeEngine) objects() []string {
183+ f.mu.Lock()
184+ defer f.mu.Unlock()
185+ var out []string
186+ for id := range f.containers {
187+ out = append(out, "container "+id)
188+ }
189+ for id := range f.networks {
190+ out = append(out, "network "+id)
191+ }
192+ for n := range f.volumes {
193+ out = append(out, "volume "+n)
194+ }
195+ return out
196+}
197+
198+type cenv struct {
199+ cfg sandbox.Config
200+ eng *fakeEngine
201+ plumb *sandboxtest.Plumbing
202+}
203+
204+func newCEnv(t *testing.T) *cenv {
205+ t.Helper()
206+ dir := t.TempDir()
207+ exe := filepath.Join(dir, "runner")
208+ sandboxtest.WriteStaticELF(t, exe)
209+ return &cenv{
210+ eng: newFake(),
211+ plumb: sandboxtest.NewPlumbing(),
212+ cfg: sandbox.Config{
213+ Isolation: sandbox.Container, StateDir: filepath.Join(dir, "state"), Capacity: 2,
214+ JobCPUs: 2, JobMemoryMiB: 1024, JobDiskMiB: 1024, UIDBase: 900000,
215+ Network: netip.MustParsePrefix("10.200.0.0/16"), Executable: exe, DNSPort: 0,
216+ },
217+ }
218+}
219+
220+func (e *cenv) driver(t *testing.T) *Driver {
221+ t.Helper()
222+ d, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb, ConnectTimeout: 3 * time.Second})
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ t.Cleanup(func() { _ = d.Close() })
227+ return d
228+}
229+
230+func (e *cenv) leftovers() []string {
231+ out := e.eng.objects()
232+ if n := e.plumb.Live(); n != 0 {
233+ out = append(out, fmt.Sprintf("%d attached guests", n))
234+ }
235+ ents, _ := os.ReadDir(filepath.Join(e.cfg.StateDir, "containers"))
236+ for _, d := range ents {
237+ out = append(out, "dir "+d.Name())
238+ }
239+ return out
240+}
241+
242+func stub(t *testing.T, conn sandbox.Conn) (string, int) {
243+ t.Helper()
244+ var out strings.Builder
245+ _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })
246+ if err != nil {
247+ t.Fatal(err)
248+ }
249+ return out.String(), int(exit.Code)
250+}
251+
252+func TestRootlessDockerIsRefused(t *testing.T) {
253+ e := newCEnv(t)
254+ e.eng.info.Rootless = true
255+ _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb})
256+ if err == nil || !strings.Contains(err.Error(), "rootless") || !strings.Contains(err.Error(), "rootful") {
257+ t.Fatalf("want a clear rootless error, got %v", err)
258+ }
259+}
260+
261+func TestRuntimeSelection(t *testing.T) {
262+ e := newCEnv(t)
263+ d := e.driver(t)
264+ if !d.Privileged() || d.runtime != "" {
265+ t.Fatal("without sysbox the driver must run privileged")
266+ }
267+ e2 := newCEnv(t)
268+ e2.eng.info.Runtimes = []string{"runc", "sysbox-runc"}
269+ d2 := e2.driver(t)
270+ if d2.Privileged() || d2.runtime != SysboxRuntime {
271+ t.Fatal("sysbox listed: must use it, unprivileged")
272+ }
273+}
274+
275+func TestNewErrors(t *testing.T) {
276+ e := newCEnv(t)
277+ e.eng.infoErr = errors.New("connection refused")
278+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil || !strings.Contains(err.Error(), "DOCKER_HOST") {
279+ t.Fatalf("got %v", err)
280+ }
281+ e = newCEnv(t)
282+ e.eng.imageErr = errors.New("no such image")
283+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
284+ t.Fatal("missing image accepted")
285+ }
286+ e = newCEnv(t)
287+ e.eng.info.OSType = "windows"
288+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
289+ t.Fatal("windows daemon accepted")
290+ }
291+ e = newCEnv(t)
292+ e.cfg.Executable = "/bin/sh"
293+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
294+ t.Skip("/bin/sh is static")
295+ }
296+}
297+
298+func TestStartRunStop(t *testing.T) {
299+ e := newCEnv(t)
300+ d := e.driver(t)
301+ sb, err := d.New()
302+ if err != nil {
303+ t.Fatal(err)
304+ }
305+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "job-7", CPUs: 3, MemoryMiB: 2048})
306+ if err != nil {
307+ t.Fatal(err)
308+ }
309+ out, code := stub(t, conn)
310+ if out != "hello from fake container\n" || code != 7 {
311+ t.Fatalf("out=%q code=%d", out, code)
312+ }
313+
314+ // What was asked of Docker.
315+ e.eng.mu.Lock()
316+ var spec containerSpec
317+ for _, s := range e.eng.containers {
318+ spec = s
319+ }
320+ var netSpec networkSpec
321+ for _, n := range e.eng.networks {
322+ netSpec = n
323+ }
324+ e.eng.mu.Unlock()
325+ if !spec.Privileged || spec.Runtime != "" {
326+ t.Errorf("privileged fallback expected: %+v", spec)
327+ }
328+ if spec.Labels[sandbox.LabelJob] != "job-7" || netSpec.Labels[sandbox.LabelJob] != "job-7" {
329+ t.Errorf("every object must carry the job label: %v / %v", spec.Labels, netSpec.Labels)
330+ }
331+ if spec.MemoryMiB != 2048 || spec.CPUs != 3 {
332+ t.Errorf("resources: %+v", spec)
333+ }
334+ if !slices.Equal(spec.Entrypoint, []string{"/opt/ermite/ermite-runner", "init"}) {
335+ t.Errorf("entrypoint: %v", spec.Entrypoint)
336+ }
337+ var roExe, runDir bool
338+ for _, b := range spec.Binds {
339+ if b.Target == "/opt/ermite/ermite-runner" && b.ReadOnly && b.Source == e.cfg.Executable {
340+ roExe = true
341+ }
342+ if b.Target == sandbox.ContainerRunDir && !b.ReadOnly {
343+ runDir = true
344+ }
345+ }
346+ if !roExe || !runDir {
347+ t.Errorf("binds: %+v", spec.Binds)
348+ }
349+ if len(spec.Volumes) != 1 || spec.Volumes[0].Target != sandbox.GuestScratch {
350+ t.Errorf("scratch volume: %+v", spec.Volumes)
351+ }
352+ if spec.DNS.String() != "10.200.0.1" || spec.IPv4.String() != "10.200.0.2" || netSpec.Bridge != "erbr0" || netSpec.Subnet.String() != "10.200.0.0/30" {
353+ t.Errorf("network: %+v %+v", spec, netSpec)
354+ }
355+ if !slices.Contains(spec.Env, "ERMITE_CONNECT=unix:/run/ermite/exec.sock") {
356+ t.Errorf("env: %v", spec.Env)
357+ }
358+ if e.plumb.Live() != 1 {
359+ t.Fatal("firewall policy not attached")
360+ }
361+
362+ if err := sb.Stop(context.Background()); err != nil {
363+ t.Fatal(err)
364+ }
365+ if left := e.leftovers(); len(left) != 0 {
366+ t.Fatalf("leftovers: %v", left)
367+ }
368+ if err := sb.Stop(context.Background()); err != nil {
369+ t.Fatalf("Stop must be idempotent: %v", err)
370+ }
371+ if d.slots.InUse() != 0 {
372+ t.Fatal("slot not released")
373+ }
374+}
375+
376+func TestSysboxJobIsNotPrivileged(t *testing.T) {
377+ e := newCEnv(t)
378+ e.eng.info.Runtimes = []string{"sysbox-runc", "runc"}
379+ d := e.driver(t)
380+ sb, _ := d.New()
381+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
382+ if err != nil {
383+ t.Fatal(err)
384+ }
385+ stub(t, conn)
386+ e.eng.mu.Lock()
387+ for _, s := range e.eng.containers {
388+ if s.Privileged || s.Runtime != "sysbox-runc" {
389+ t.Errorf("sysbox job: %+v", s)
390+ }
391+ }
392+ e.eng.mu.Unlock()
393+ _ = sb.Stop(context.Background())
394+}
395+
396+func TestStartFailuresCleanEverything(t *testing.T) {
397+ cases := map[string]func(*cenv){
398+ "container cannot start": func(e *cenv) { e.eng.failStart = errors.New("oci runtime error") },
399+ "container exits early": func(e *cenv) { e.eng.exitOnRun = true },
400+ "executor never connects": func(e *cenv) {
401+ e.eng.noConnect = true
402+ },
403+ "firewall attach fails": func(e *cenv) { e.plumb.FailNext = errors.New("nft broke") },
404+ }
405+ for name, setup := range cases {
406+ t.Run(name, func(t *testing.T) {
407+ e := newCEnv(t)
408+ setup(e)
409+ d, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb, ConnectTimeout: 300 * time.Millisecond})
410+ if err != nil {
411+ t.Fatal(err)
412+ }
413+ defer func() { _ = d.Close() }()
414+ sb, _ := d.New()
415+ _, err = sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
416+ if err == nil {
417+ t.Fatal("Start succeeded")
418+ }
419+ if left := e.leftovers(); len(left) != 0 {
420+ t.Fatalf("leftovers: %v (err %v)", left, err)
421+ }
422+ if d.slots.InUse() != 0 {
423+ t.Fatal("slot leaked")
424+ }
425+ })
426+ }
427+}
428+
429+func TestFailedStartIncludesContainerLogs(t *testing.T) {
430+ e := newCEnv(t)
431+ e.eng.exitOnRun = true
432+ d := e.driver(t)
433+ sb, _ := d.New()
434+ _, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
435+ if err == nil || !strings.Contains(err.Error(), "fake container log line") {
436+ t.Fatalf("got %v", err)
437+ }
438+}
439+
440+func TestCleanupFailureKeepsSlotAndIsRetriedByReap(t *testing.T) {
441+ e := newCEnv(t)
442+ d := e.driver(t)
443+ sb, _ := d.New()
444+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
445+ if err != nil {
446+ t.Fatal(err)
447+ }
448+ stub(t, conn)
449+ e.eng.removeErr = errors.New("daemon busy")
450+ if err := sb.Stop(context.Background()); err == nil {
451+ t.Fatal("Stop must report the failed removal")
452+ }
453+ if d.slots.InUse() != 1 {
454+ t.Fatal("the slot must stay taken while its bridge and subnet still exist")
455+ }
456+ e.eng.removeErr = nil
457+ if err := d.Reap(context.Background()); err != nil {
458+ t.Fatal(err)
459+ }
460+ if left := e.leftovers(); len(left) != 0 {
461+ t.Fatalf("Reap left %v", left)
462+ }
463+}
464+
465+func TestReapRemovesLabelledObjectsOnly(t *testing.T) {
466+ e := newCEnv(t)
467+ d := e.driver(t)
468+ ctx := context.Background()
469+ // Leftovers of a crashed run, and an unrelated user's network.
470+ _, _ = e.eng.CreateContainer(ctx, containerSpec{Name: "old", Labels: map[string]string{sandbox.LabelJob: "x"}})
471+ _, _ = e.eng.CreateNetwork(ctx, networkSpec{Name: "old", Labels: map[string]string{sandbox.LabelJob: "x"}})
472+ _, _ = e.eng.CreateNetwork(ctx, networkSpec{Name: "mine", Labels: map[string]string{"app": "other"}})
473+ _ = e.eng.CreateVolume(ctx, "old", map[string]string{sandbox.LabelJob: "x"})
474+ _ = os.MkdirAll(filepath.Join(e.cfg.StateDir, "containers", "er00112233"), 0o700)
475+ _ = os.MkdirAll(filepath.Join(e.cfg.StateDir, "containers", "keep-me"), 0o700)
476+
477+ if err := d.Reap(ctx); err != nil {
478+ t.Fatal(err)
479+ }
480+ left := e.leftovers()
481+ slices.Sort(left)
482+ if len(left) != 2 || !strings.HasPrefix(left[0], "dir keep-me") || !strings.HasPrefix(left[1], "network") {
483+ t.Fatalf("only the unrelated network and directory may remain: %v", left)
484+ }
485+ if _, err := os.Stat(filepath.Join(e.cfg.StateDir, "containers", "keep-me")); err != nil {
486+ t.Fatal("Reap removed a directory that is not ours")
487+ }
488+ if e.plumb.Resets != 1 {
489+ t.Fatal("nftables table not reset")
490+ }
491+}
492+
493+func TestCapacityAndSlotIsolation(t *testing.T) {
494+ e := newCEnv(t)
495+ d := e.driver(t)
496+ a, _ := d.New()
497+ b, _ := d.New()
498+ if _, err := d.New(); !errors.Is(err, sandbox.ErrCapacity) {
499+ t.Fatalf("got %v", err)
500+ }
501+ for _, sb := range []sandbox.Sandbox{a, b} {
502+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
503+ if err != nil {
504+ t.Fatal(err)
505+ }
506+ stub(t, conn)
507+ }
508+ ifaces := map[string]bool{}
509+ for iface := range e.plumb.Attached {
510+ ifaces[iface] = true
511+ }
512+ if !ifaces["erbr0"] || !ifaces["erbr1"] {
513+ t.Fatalf("each job needs its own bridge: %v", ifaces)
514+ }
515+ _ = a.Stop(context.Background())
516+ _ = b.Stop(context.Background())
517+ if left := e.leftovers(); len(left) != 0 {
518+ t.Fatalf("leftovers: %v", left)
519+ }
520+}
521+
522+func TestStopWithCancelledContext(t *testing.T) {
523+ e := newCEnv(t)
524+ d := e.driver(t)
525+ sb, _ := d.New()
526+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
527+ if err != nil {
528+ t.Fatal(err)
529+ }
530+ stub(t, conn)
531+ ctx, cancel := context.WithCancel(context.Background())
532+ cancel()
533+ if err := sb.Stop(ctx); err != nil {
534+ t.Fatal(err)
535+ }
536+ if left := e.leftovers(); len(left) != 0 {
537+ t.Fatalf("leftovers: %v", left)
538+ }
539+}
540+
541+func TestDoubleStartRefused(t *testing.T) {
542+ e := newCEnv(t)
543+ d := e.driver(t)
544+ sb, _ := d.New()
545+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
546+ if err != nil {
547+ t.Fatal(err)
548+ }
549+ stub(t, conn)
550+ if _, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"}); err == nil {
551+ t.Fatal("second Start accepted")
552+ }
553+ _ = sb.Stop(context.Background())
554+}
555+
556+func mustPrefix(s string) netip.Prefix { return netip.MustParsePrefix(s) }
@@ -0,0 +1,556 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "net"
10+ "net/netip"
11+ "os"
12+ "path/filepath"
13+ "slices"
14+ "strings"
15+ "sync"
16+ "testing"
17+ "time"
18+
19+ "bnema.dev/ermite-runner/internal/execstream"
20+ "bnema.dev/ermite-runner/internal/execstream/execpb"
21+ "bnema.dev/ermite-runner/internal/sandbox"
22+ "bnema.dev/ermite-runner/internal/sandbox/sandboxtest"
23+)
24+
25+// fakeEngine plays the Docker daemon. When a container "starts", it plays the
26+// executor inside it by connecting to the bind-mounted socket.
27+type fakeEngine struct {
28+ mu sync.Mutex
29+ info daemonInfo
30+ infoErr error
31+ imageErr error
32+ networks map[string]networkSpec
33+ volumes map[string]map[string]string
34+ containers map[string]containerSpec
35+ failStart error
36+ noConnect bool
37+ exitOnRun bool
38+ removeErr error
39+ waiters map[string]chan exitResult
40+ seq int
41+ pulled []string
42+}
43+
44+func newFake() *fakeEngine {
45+ return &fakeEngine{
46+ info: daemonInfo{OSType: "linux", Runtimes: []string{"runc"}},
47+ networks: map[string]networkSpec{},
48+ volumes: map[string]map[string]string{},
49+ containers: map[string]containerSpec{},
50+ waiters: map[string]chan exitResult{},
51+ }
52+}
53+
54+func (f *fakeEngine) Info(context.Context) (daemonInfo, error) { return f.info, f.infoErr }
55+func (f *fakeEngine) EnsureImage(_ context.Context, ref string) error {
56+ f.pulled = append(f.pulled, ref)
57+ return f.imageErr
58+}
59+func (f *fakeEngine) CreateNetwork(_ context.Context, s networkSpec) (string, error) {
60+ f.mu.Lock()
61+ defer f.mu.Unlock()
62+ f.seq++
63+ id := fmt.Sprintf("net%d", f.seq)
64+ f.networks[id] = s
65+ return id, nil
66+}
67+func (f *fakeEngine) RemoveNetwork(_ context.Context, id string) error {
68+ f.mu.Lock()
69+ defer f.mu.Unlock()
70+ delete(f.networks, id)
71+ return nil
72+}
73+func (f *fakeEngine) ListNetworks(context.Context, string) ([]string, error) {
74+ f.mu.Lock()
75+ defer f.mu.Unlock()
76+ var out []string
77+ for id, n := range f.networks {
78+ if _, ok := n.Labels[sandbox.LabelJob]; ok {
79+ out = append(out, id)
80+ }
81+ }
82+ return out, nil
83+}
84+func (f *fakeEngine) CreateVolume(_ context.Context, name string, labels map[string]string) error {
85+ f.mu.Lock()
86+ defer f.mu.Unlock()
87+ f.volumes[name] = labels
88+ return nil
89+}
90+func (f *fakeEngine) RemoveVolume(_ context.Context, name string) error {
91+ f.mu.Lock()
92+ defer f.mu.Unlock()
93+ delete(f.volumes, name)
94+ return nil
95+}
96+func (f *fakeEngine) ListVolumes(context.Context, string) ([]string, error) {
97+ f.mu.Lock()
98+ defer f.mu.Unlock()
99+ var out []string
100+ for n := range f.volumes {
101+ out = append(out, n)
102+ }
103+ return out, nil
104+}
105+func (f *fakeEngine) CreateContainer(_ context.Context, s containerSpec) (string, error) {
106+ f.mu.Lock()
107+ defer f.mu.Unlock()
108+ f.seq++
109+ id := fmt.Sprintf("ctr%d", f.seq)
110+ f.containers[id] = s
111+ f.waiters[id] = make(chan exitResult, 1)
112+ return id, nil
113+}
114+func (f *fakeEngine) StartContainer(_ context.Context, id string) error {
115+ f.mu.Lock()
116+ spec := f.containers[id]
117+ fail, noConnect, exitOnRun := f.failStart, f.noConnect, f.exitOnRun
118+ w := f.waiters[id]
119+ f.mu.Unlock()
120+ if fail != nil {
121+ return fail
122+ }
123+ if exitOnRun {
124+ w <- exitResult{Code: 125}
125+ return nil
126+ }
127+ if noConnect {
128+ return nil
129+ }
130+ // The host side of the bind mount is the source of the run-dir mount.
131+ var sock string
132+ for _, b := range spec.Binds {
133+ if b.Target == sandbox.ContainerRunDir {
134+ sock = filepath.Join(b.Source, sandbox.ContainerSocket)
135+ }
136+ }
137+ go func() {
138+ c, err := net.Dial("unix", sock)
139+ if err != nil {
140+ return
141+ }
142+ defer func() { _ = c.Close() }()
143+ g := execstream.NewConn(c)
144+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Hello{Hello: &execpb.Hello{Sandbox: "container"}}})
145+ if _, err := g.Recv(); err != nil {
146+ return
147+ }
148+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Output{Output: &execpb.Output{Data: []byte("hello from fake container\n")}}})
149+ _ = g.Send(&execpb.Frame{Body: &execpb.Frame_Exit{Exit: &execpb.Exit{Code: 7}}})
150+ _, _ = g.Recv()
151+ }()
152+ return nil
153+}
154+func (f *fakeEngine) WaitExit(_ context.Context, id string) <-chan exitResult {
155+ f.mu.Lock()
156+ defer f.mu.Unlock()
157+ return f.waiters[id]
158+}
159+func (f *fakeEngine) RemoveContainer(_ context.Context, id string) error {
160+ f.mu.Lock()
161+ defer f.mu.Unlock()
162+ if f.removeErr != nil {
163+ return f.removeErr
164+ }
165+ delete(f.containers, id)
166+ return nil
167+}
168+func (f *fakeEngine) ListContainers(context.Context, string) ([]string, error) {
169+ f.mu.Lock()
170+ defer f.mu.Unlock()
171+ var out []string
172+ for id := range f.containers {
173+ out = append(out, id)
174+ }
175+ return out, nil
176+}
177+func (f *fakeEngine) LogsTail(context.Context, string, int) (string, error) {
178+ return "fake container log line", nil
179+}
180+func (f *fakeEngine) Close() error { return nil }
181+
182+func (f *fakeEngine) objects() []string {
183+ f.mu.Lock()
184+ defer f.mu.Unlock()
185+ var out []string
186+ for id := range f.containers {
187+ out = append(out, "container "+id)
188+ }
189+ for id := range f.networks {
190+ out = append(out, "network "+id)
191+ }
192+ for n := range f.volumes {
193+ out = append(out, "volume "+n)
194+ }
195+ return out
196+}
197+
198+type cenv struct {
199+ cfg sandbox.Config
200+ eng *fakeEngine
201+ plumb *sandboxtest.Plumbing
202+}
203+
204+func newCEnv(t *testing.T) *cenv {
205+ t.Helper()
206+ dir := t.TempDir()
207+ exe := filepath.Join(dir, "runner")
208+ sandboxtest.WriteStaticELF(t, exe)
209+ return &cenv{
210+ eng: newFake(),
211+ plumb: sandboxtest.NewPlumbing(),
212+ cfg: sandbox.Config{
213+ Isolation: sandbox.Container, StateDir: filepath.Join(dir, "state"), Capacity: 2,
214+ JobCPUs: 2, JobMemoryMiB: 1024, JobDiskMiB: 1024, UIDBase: 900000,
215+ Network: netip.MustParsePrefix("10.200.0.0/16"), Executable: exe, DNSPort: 0,
216+ },
217+ }
218+}
219+
220+func (e *cenv) driver(t *testing.T) *Driver {
221+ t.Helper()
222+ d, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb, ConnectTimeout: 3 * time.Second})
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ t.Cleanup(func() { _ = d.Close() })
227+ return d
228+}
229+
230+func (e *cenv) leftovers() []string {
231+ out := e.eng.objects()
232+ if n := e.plumb.Live(); n != 0 {
233+ out = append(out, fmt.Sprintf("%d attached guests", n))
234+ }
235+ ents, _ := os.ReadDir(filepath.Join(e.cfg.StateDir, "containers"))
236+ for _, d := range ents {
237+ out = append(out, "dir "+d.Name())
238+ }
239+ return out
240+}
241+
242+func stub(t *testing.T, conn sandbox.Conn) (string, int) {
243+ t.Helper()
244+ var out strings.Builder
245+ _, exit, err := execstream.Drive(context.Background(), conn, &execpb.Run{}, func(o *execpb.Output) { out.Write(o.Data) })
246+ if err != nil {
247+ t.Fatal(err)
248+ }
249+ return out.String(), int(exit.Code)
250+}
251+
252+func TestRootlessDockerIsRefused(t *testing.T) {
253+ e := newCEnv(t)
254+ e.eng.info.Rootless = true
255+ _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb})
256+ if err == nil || !strings.Contains(err.Error(), "rootless") || !strings.Contains(err.Error(), "rootful") {
257+ t.Fatalf("want a clear rootless error, got %v", err)
258+ }
259+}
260+
261+func TestRuntimeSelection(t *testing.T) {
262+ e := newCEnv(t)
263+ d := e.driver(t)
264+ if !d.Privileged() || d.runtime != "" {
265+ t.Fatal("without sysbox the driver must run privileged")
266+ }
267+ e2 := newCEnv(t)
268+ e2.eng.info.Runtimes = []string{"runc", "sysbox-runc"}
269+ d2 := e2.driver(t)
270+ if d2.Privileged() || d2.runtime != SysboxRuntime {
271+ t.Fatal("sysbox listed: must use it, unprivileged")
272+ }
273+}
274+
275+func TestNewErrors(t *testing.T) {
276+ e := newCEnv(t)
277+ e.eng.infoErr = errors.New("connection refused")
278+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil || !strings.Contains(err.Error(), "DOCKER_HOST") {
279+ t.Fatalf("got %v", err)
280+ }
281+ e = newCEnv(t)
282+ e.eng.imageErr = errors.New("no such image")
283+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
284+ t.Fatal("missing image accepted")
285+ }
286+ e = newCEnv(t)
287+ e.eng.info.OSType = "windows"
288+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
289+ t.Fatal("windows daemon accepted")
290+ }
291+ e = newCEnv(t)
292+ e.cfg.Executable = "/bin/sh"
293+ if _, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb}); err == nil {
294+ t.Skip("/bin/sh is static")
295+ }
296+}
297+
298+func TestStartRunStop(t *testing.T) {
299+ e := newCEnv(t)
300+ d := e.driver(t)
301+ sb, err := d.New()
302+ if err != nil {
303+ t.Fatal(err)
304+ }
305+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "job-7", CPUs: 3, MemoryMiB: 2048})
306+ if err != nil {
307+ t.Fatal(err)
308+ }
309+ out, code := stub(t, conn)
310+ if out != "hello from fake container\n" || code != 7 {
311+ t.Fatalf("out=%q code=%d", out, code)
312+ }
313+
314+ // What was asked of Docker.
315+ e.eng.mu.Lock()
316+ var spec containerSpec
317+ for _, s := range e.eng.containers {
318+ spec = s
319+ }
320+ var netSpec networkSpec
321+ for _, n := range e.eng.networks {
322+ netSpec = n
323+ }
324+ e.eng.mu.Unlock()
325+ if !spec.Privileged || spec.Runtime != "" {
326+ t.Errorf("privileged fallback expected: %+v", spec)
327+ }
328+ if spec.Labels[sandbox.LabelJob] != "job-7" || netSpec.Labels[sandbox.LabelJob] != "job-7" {
329+ t.Errorf("every object must carry the job label: %v / %v", spec.Labels, netSpec.Labels)
330+ }
331+ if spec.MemoryMiB != 2048 || spec.CPUs != 3 {
332+ t.Errorf("resources: %+v", spec)
333+ }
334+ if !slices.Equal(spec.Entrypoint, []string{"/opt/ermite/ermite-runner", "init"}) {
335+ t.Errorf("entrypoint: %v", spec.Entrypoint)
336+ }
337+ var roExe, runDir bool
338+ for _, b := range spec.Binds {
339+ if b.Target == "/opt/ermite/ermite-runner" && b.ReadOnly && b.Source == e.cfg.Executable {
340+ roExe = true
341+ }
342+ if b.Target == sandbox.ContainerRunDir && !b.ReadOnly {
343+ runDir = true
344+ }
345+ }
346+ if !roExe || !runDir {
347+ t.Errorf("binds: %+v", spec.Binds)
348+ }
349+ if len(spec.Volumes) != 1 || spec.Volumes[0].Target != sandbox.GuestScratch {
350+ t.Errorf("scratch volume: %+v", spec.Volumes)
351+ }
352+ if spec.DNS.String() != "10.200.0.1" || spec.IPv4.String() != "10.200.0.2" || netSpec.Bridge != "erbr0" || netSpec.Subnet.String() != "10.200.0.0/30" {
353+ t.Errorf("network: %+v %+v", spec, netSpec)
354+ }
355+ if !slices.Contains(spec.Env, "ERMITE_CONNECT=unix:/run/ermite/exec.sock") {
356+ t.Errorf("env: %v", spec.Env)
357+ }
358+ if e.plumb.Live() != 1 {
359+ t.Fatal("firewall policy not attached")
360+ }
361+
362+ if err := sb.Stop(context.Background()); err != nil {
363+ t.Fatal(err)
364+ }
365+ if left := e.leftovers(); len(left) != 0 {
366+ t.Fatalf("leftovers: %v", left)
367+ }
368+ if err := sb.Stop(context.Background()); err != nil {
369+ t.Fatalf("Stop must be idempotent: %v", err)
370+ }
371+ if d.slots.InUse() != 0 {
372+ t.Fatal("slot not released")
373+ }
374+}
375+
376+func TestSysboxJobIsNotPrivileged(t *testing.T) {
377+ e := newCEnv(t)
378+ e.eng.info.Runtimes = []string{"sysbox-runc", "runc"}
379+ d := e.driver(t)
380+ sb, _ := d.New()
381+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
382+ if err != nil {
383+ t.Fatal(err)
384+ }
385+ stub(t, conn)
386+ e.eng.mu.Lock()
387+ for _, s := range e.eng.containers {
388+ if s.Privileged || s.Runtime != "sysbox-runc" {
389+ t.Errorf("sysbox job: %+v", s)
390+ }
391+ }
392+ e.eng.mu.Unlock()
393+ _ = sb.Stop(context.Background())
394+}
395+
396+func TestStartFailuresCleanEverything(t *testing.T) {
397+ cases := map[string]func(*cenv){
398+ "container cannot start": func(e *cenv) { e.eng.failStart = errors.New("oci runtime error") },
399+ "container exits early": func(e *cenv) { e.eng.exitOnRun = true },
400+ "executor never connects": func(e *cenv) {
401+ e.eng.noConnect = true
402+ },
403+ "firewall attach fails": func(e *cenv) { e.plumb.FailNext = errors.New("nft broke") },
404+ }
405+ for name, setup := range cases {
406+ t.Run(name, func(t *testing.T) {
407+ e := newCEnv(t)
408+ setup(e)
409+ d, err := New(context.Background(), e.cfg, Options{engine: e.eng, Net: e.plumb, ConnectTimeout: 300 * time.Millisecond})
410+ if err != nil {
411+ t.Fatal(err)
412+ }
413+ defer func() { _ = d.Close() }()
414+ sb, _ := d.New()
415+ _, err = sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
416+ if err == nil {
417+ t.Fatal("Start succeeded")
418+ }
419+ if left := e.leftovers(); len(left) != 0 {
420+ t.Fatalf("leftovers: %v (err %v)", left, err)
421+ }
422+ if d.slots.InUse() != 0 {
423+ t.Fatal("slot leaked")
424+ }
425+ })
426+ }
427+}
428+
429+func TestFailedStartIncludesContainerLogs(t *testing.T) {
430+ e := newCEnv(t)
431+ e.eng.exitOnRun = true
432+ d := e.driver(t)
433+ sb, _ := d.New()
434+ _, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
435+ if err == nil || !strings.Contains(err.Error(), "fake container log line") {
436+ t.Fatalf("got %v", err)
437+ }
438+}
439+
440+func TestCleanupFailureKeepsSlotAndIsRetriedByReap(t *testing.T) {
441+ e := newCEnv(t)
442+ d := e.driver(t)
443+ sb, _ := d.New()
444+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
445+ if err != nil {
446+ t.Fatal(err)
447+ }
448+ stub(t, conn)
449+ e.eng.removeErr = errors.New("daemon busy")
450+ if err := sb.Stop(context.Background()); err == nil {
451+ t.Fatal("Stop must report the failed removal")
452+ }
453+ if d.slots.InUse() != 1 {
454+ t.Fatal("the slot must stay taken while its bridge and subnet still exist")
455+ }
456+ e.eng.removeErr = nil
457+ if err := d.Reap(context.Background()); err != nil {
458+ t.Fatal(err)
459+ }
460+ if left := e.leftovers(); len(left) != 0 {
461+ t.Fatalf("Reap left %v", left)
462+ }
463+}
464+
465+func TestReapRemovesLabelledObjectsOnly(t *testing.T) {
466+ e := newCEnv(t)
467+ d := e.driver(t)
468+ ctx := context.Background()
469+ // Leftovers of a crashed run, and an unrelated user's network.
470+ _, _ = e.eng.CreateContainer(ctx, containerSpec{Name: "old", Labels: map[string]string{sandbox.LabelJob: "x"}})
471+ _, _ = e.eng.CreateNetwork(ctx, networkSpec{Name: "old", Labels: map[string]string{sandbox.LabelJob: "x"}})
472+ _, _ = e.eng.CreateNetwork(ctx, networkSpec{Name: "mine", Labels: map[string]string{"app": "other"}})
473+ _ = e.eng.CreateVolume(ctx, "old", map[string]string{sandbox.LabelJob: "x"})
474+ _ = os.MkdirAll(filepath.Join(e.cfg.StateDir, "containers", "er00112233"), 0o700)
475+ _ = os.MkdirAll(filepath.Join(e.cfg.StateDir, "containers", "keep-me"), 0o700)
476+
477+ if err := d.Reap(ctx); err != nil {
478+ t.Fatal(err)
479+ }
480+ left := e.leftovers()
481+ slices.Sort(left)
482+ if len(left) != 2 || !strings.HasPrefix(left[0], "dir keep-me") || !strings.HasPrefix(left[1], "network") {
483+ t.Fatalf("only the unrelated network and directory may remain: %v", left)
484+ }
485+ if _, err := os.Stat(filepath.Join(e.cfg.StateDir, "containers", "keep-me")); err != nil {
486+ t.Fatal("Reap removed a directory that is not ours")
487+ }
488+ if e.plumb.Resets != 1 {
489+ t.Fatal("nftables table not reset")
490+ }
491+}
492+
493+func TestCapacityAndSlotIsolation(t *testing.T) {
494+ e := newCEnv(t)
495+ d := e.driver(t)
496+ a, _ := d.New()
497+ b, _ := d.New()
498+ if _, err := d.New(); !errors.Is(err, sandbox.ErrCapacity) {
499+ t.Fatalf("got %v", err)
500+ }
501+ for _, sb := range []sandbox.Sandbox{a, b} {
502+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
503+ if err != nil {
504+ t.Fatal(err)
505+ }
506+ stub(t, conn)
507+ }
508+ ifaces := map[string]bool{}
509+ for iface := range e.plumb.Attached {
510+ ifaces[iface] = true
511+ }
512+ if !ifaces["erbr0"] || !ifaces["erbr1"] {
513+ t.Fatalf("each job needs its own bridge: %v", ifaces)
514+ }
515+ _ = a.Stop(context.Background())
516+ _ = b.Stop(context.Background())
517+ if left := e.leftovers(); len(left) != 0 {
518+ t.Fatalf("leftovers: %v", left)
519+ }
520+}
521+
522+func TestStopWithCancelledContext(t *testing.T) {
523+ e := newCEnv(t)
524+ d := e.driver(t)
525+ sb, _ := d.New()
526+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
527+ if err != nil {
528+ t.Fatal(err)
529+ }
530+ stub(t, conn)
531+ ctx, cancel := context.WithCancel(context.Background())
532+ cancel()
533+ if err := sb.Stop(ctx); err != nil {
534+ t.Fatal(err)
535+ }
536+ if left := e.leftovers(); len(left) != 0 {
537+ t.Fatalf("leftovers: %v", left)
538+ }
539+}
540+
541+func TestDoubleStartRefused(t *testing.T) {
542+ e := newCEnv(t)
543+ d := e.driver(t)
544+ sb, _ := d.New()
545+ conn, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"})
546+ if err != nil {
547+ t.Fatal(err)
548+ }
549+ stub(t, conn)
550+ if _, err := sb.Start(context.Background(), sandbox.JobSpec{JobID: "j"}); err == nil {
551+ t.Fatal("second Start accepted")
552+ }
553+ _ = sb.Stop(context.Background())
554+}
555+
556+func mustPrefix(s string) netip.Prefix { return netip.MustParsePrefix(s) }
A internal/sandbox/container/docker_api_test.go +129 -0
@@ -0,0 +1,129 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "os"
8+ "os/exec"
9+ "path/filepath"
10+ "strings"
11+ "testing"
12+ "time"
13+
14+ "bnema.dev/ermite-runner/internal/execstream"
15+ "bnema.dev/ermite-runner/internal/execstream/execpb"
16+ "bnema.dev/ermite-runner/internal/sandbox"
17+ "bnema.dev/ermite-runner/internal/sandbox/sandboxtest"
18+)
19+
20+// TestDockerAPIWiring runs the real Docker Engine calls (network with a named
21+// bridge, labelled volume and container, bind mounts, wait, removal, reaping by
22+// label) against whatever daemon DOCKER_HOST points at, including a rootless
23+// one. It does not test isolation (a rootless daemon cannot run the nested
24+// dockerd): the integration test in internal/sandbox/integration does.
25+//
26+// ERMITE_RUNNER_IT_DOCKER_API=1 ERMITE_RUNNER_TEST_IMAGE=<image> go test -run DockerAPIWiring ./internal/sandbox/container
27+func TestDockerAPIWiring(t *testing.T) {
28+ if os.Getenv("ERMITE_RUNNER_IT_DOCKER_API") != "1" {
29+ t.Skip("set ERMITE_RUNNER_IT_DOCKER_API=1 (needs a reachable Docker daemon)")
30+ }
31+ image := os.Getenv("ERMITE_RUNNER_TEST_IMAGE")
32+ if image == "" {
33+ t.Fatal("set ERMITE_RUNNER_TEST_IMAGE to an image with /bin/sh (e.g. the runner image)")
34+ }
35+ if _, err := exec.LookPath("docker"); err != nil {
36+ t.Skip("docker CLI needed for the verification")
37+ }
38+ // A static test executable that plays the executor: the entrypoint of the
39+ // container is "<exe> init", and init runs "<exe> exec"... replaced here by a
40+ // tiny shell executor that speaks the stream through a Go helper binary.
41+ dir := t.TempDir()
42+ helper := filepath.Join(dir, "ermite-runner")
43+ build := exec.Command("go", "build", "-o", helper, "bnema.dev/ermite-runner")
44+ build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOWORK=off")
45+ build.Dir = repoRoot(t)
46+ if out, err := build.CombinedOutput(); err != nil {
47+ t.Fatalf("build runner: %v\n%s", err, out)
48+ }
49+
50+ plumb := sandboxtest.NewPlumbing()
51+ cfg := sandbox.Config{
52+ Isolation: sandbox.Container, StateDir: filepath.Join(dir, "state"), Capacity: 1, Image: image,
53+ JobCPUs: 1, JobMemoryMiB: 512, JobDiskMiB: 1024, UIDBase: 900000,
54+ Network: mustPrefix("10.213.0.0/16"), Executable: helper, DNSPort: 0,
55+ }
56+ ctx := context.Background()
57+ d, err := New(ctx, cfg, Options{Net: plumb, allowRootless: true, ConnectTimeout: 60 * time.Second})
58+ if err != nil {
59+ t.Fatal(err)
60+ }
61+ defer func() { _ = d.Close() }()
62+ if err := d.Reap(ctx); err != nil {
63+ t.Fatal(err)
64+ }
65+
66+ for i := range 2 {
67+ sb, err := d.New()
68+ if err != nil {
69+ t.Fatal(err)
70+ }
71+ conn, err := sb.Start(ctx, sandbox.JobSpec{JobID: "api-wiring"})
72+ if err != nil {
73+ t.Fatalf("job %d: %v", i, err)
74+ }
75+ var out strings.Builder
76+ hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, func(o *execpb.Output) { out.Write(o.Data) })
77+ if err != nil {
78+ t.Fatal(err)
79+ }
80+ if hello.Sandbox != "container" || exit.Code != 4 || !strings.HasPrefix(out.String(), "wired\n") {
81+ t.Fatalf("hello=%v exit=%v out=%q", hello, exit, out.String())
82+ }
83+ // While running, Docker knows our objects by label.
84+ if i == 0 {
85+ if ids := dockerLines(t, "ps", "-aq", "--filter", "label="+sandbox.LabelJob); len(ids) != 1 {
86+ t.Fatalf("expected one labelled container, got %v", ids)
87+ }
88+ if n := dockerLines(t, "network", "ls", "-q", "--filter", "label="+sandbox.LabelJob); len(n) != 1 {
89+ t.Fatalf("expected one labelled network, got %v", n)
90+ }
91+ if v := dockerLines(t, "volume", "ls", "-q", "--filter", "label="+sandbox.LabelJob); len(v) != 1 {
92+ t.Fatalf("expected one labelled volume, got %v", v)
93+ }
94+ }
95+ if err := sb.Stop(ctx); err != nil {
96+ t.Fatal(err)
97+ }
98+ }
99+ for _, args := range [][]string{
100+ {"ps", "-aq", "--filter", "label=" + sandbox.LabelJob},
101+ {"network", "ls", "-q", "--filter", "label=" + sandbox.LabelJob},
102+ {"volume", "ls", "-q", "--filter", "label=" + sandbox.LabelJob},
103+ } {
104+ if l := dockerLines(t, args...); len(l) != 0 {
105+ t.Fatalf("docker %v left %v", args, l)
106+ }
107+ }
108+ if plumb.Live() != 0 {
109+ t.Fatal("firewall attachments leaked")
110+ }
111+}
112+
113+func dockerLines(t *testing.T, args ...string) []string {
114+ t.Helper()
115+ out, err := exec.Command("docker", args...).Output()
116+ if err != nil {
117+ t.Fatalf("docker %v: %v", args, err)
118+ }
119+ return strings.Fields(string(out))
120+}
121+
122+func repoRoot(t *testing.T) string {
123+ t.Helper()
124+ out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}").Output()
125+ if err != nil {
126+ t.Fatal(err)
127+ }
128+ return strings.TrimSpace(string(out))
129+}
@@ -0,0 +1,129 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "os"
8+ "os/exec"
9+ "path/filepath"
10+ "strings"
11+ "testing"
12+ "time"
13+
14+ "bnema.dev/ermite-runner/internal/execstream"
15+ "bnema.dev/ermite-runner/internal/execstream/execpb"
16+ "bnema.dev/ermite-runner/internal/sandbox"
17+ "bnema.dev/ermite-runner/internal/sandbox/sandboxtest"
18+)
19+
20+// TestDockerAPIWiring runs the real Docker Engine calls (network with a named
21+// bridge, labelled volume and container, bind mounts, wait, removal, reaping by
22+// label) against whatever daemon DOCKER_HOST points at, including a rootless
23+// one. It does not test isolation (a rootless daemon cannot run the nested
24+// dockerd): the integration test in internal/sandbox/integration does.
25+//
26+// ERMITE_RUNNER_IT_DOCKER_API=1 ERMITE_RUNNER_TEST_IMAGE=<image> go test -run DockerAPIWiring ./internal/sandbox/container
27+func TestDockerAPIWiring(t *testing.T) {
28+ if os.Getenv("ERMITE_RUNNER_IT_DOCKER_API") != "1" {
29+ t.Skip("set ERMITE_RUNNER_IT_DOCKER_API=1 (needs a reachable Docker daemon)")
30+ }
31+ image := os.Getenv("ERMITE_RUNNER_TEST_IMAGE")
32+ if image == "" {
33+ t.Fatal("set ERMITE_RUNNER_TEST_IMAGE to an image with /bin/sh (e.g. the runner image)")
34+ }
35+ if _, err := exec.LookPath("docker"); err != nil {
36+ t.Skip("docker CLI needed for the verification")
37+ }
38+ // A static test executable that plays the executor: the entrypoint of the
39+ // container is "<exe> init", and init runs "<exe> exec"... replaced here by a
40+ // tiny shell executor that speaks the stream through a Go helper binary.
41+ dir := t.TempDir()
42+ helper := filepath.Join(dir, "ermite-runner")
43+ build := exec.Command("go", "build", "-o", helper, "bnema.dev/ermite-runner")
44+ build.Env = append(os.Environ(), "CGO_ENABLED=0", "GOWORK=off")
45+ build.Dir = repoRoot(t)
46+ if out, err := build.CombinedOutput(); err != nil {
47+ t.Fatalf("build runner: %v\n%s", err, out)
48+ }
49+
50+ plumb := sandboxtest.NewPlumbing()
51+ cfg := sandbox.Config{
52+ Isolation: sandbox.Container, StateDir: filepath.Join(dir, "state"), Capacity: 1, Image: image,
53+ JobCPUs: 1, JobMemoryMiB: 512, JobDiskMiB: 1024, UIDBase: 900000,
54+ Network: mustPrefix("10.213.0.0/16"), Executable: helper, DNSPort: 0,
55+ }
56+ ctx := context.Background()
57+ d, err := New(ctx, cfg, Options{Net: plumb, allowRootless: true, ConnectTimeout: 60 * time.Second})
58+ if err != nil {
59+ t.Fatal(err)
60+ }
61+ defer func() { _ = d.Close() }()
62+ if err := d.Reap(ctx); err != nil {
63+ t.Fatal(err)
64+ }
65+
66+ for i := range 2 {
67+ sb, err := d.New()
68+ if err != nil {
69+ t.Fatal(err)
70+ }
71+ conn, err := sb.Start(ctx, sandbox.JobSpec{JobID: "api-wiring"})
72+ if err != nil {
73+ t.Fatalf("job %d: %v", i, err)
74+ }
75+ var out strings.Builder
76+ hello, exit, err := execstream.Drive(ctx, conn, &execpb.Run{Script: "echo wired; id -u; exit 4"}, func(o *execpb.Output) { out.Write(o.Data) })
77+ if err != nil {
78+ t.Fatal(err)
79+ }
80+ if hello.Sandbox != "container" || exit.Code != 4 || !strings.HasPrefix(out.String(), "wired\n") {
81+ t.Fatalf("hello=%v exit=%v out=%q", hello, exit, out.String())
82+ }
83+ // While running, Docker knows our objects by label.
84+ if i == 0 {
85+ if ids := dockerLines(t, "ps", "-aq", "--filter", "label="+sandbox.LabelJob); len(ids) != 1 {
86+ t.Fatalf("expected one labelled container, got %v", ids)
87+ }
88+ if n := dockerLines(t, "network", "ls", "-q", "--filter", "label="+sandbox.LabelJob); len(n) != 1 {
89+ t.Fatalf("expected one labelled network, got %v", n)
90+ }
91+ if v := dockerLines(t, "volume", "ls", "-q", "--filter", "label="+sandbox.LabelJob); len(v) != 1 {
92+ t.Fatalf("expected one labelled volume, got %v", v)
93+ }
94+ }
95+ if err := sb.Stop(ctx); err != nil {
96+ t.Fatal(err)
97+ }
98+ }
99+ for _, args := range [][]string{
100+ {"ps", "-aq", "--filter", "label=" + sandbox.LabelJob},
101+ {"network", "ls", "-q", "--filter", "label=" + sandbox.LabelJob},
102+ {"volume", "ls", "-q", "--filter", "label=" + sandbox.LabelJob},
103+ } {
104+ if l := dockerLines(t, args...); len(l) != 0 {
105+ t.Fatalf("docker %v left %v", args, l)
106+ }
107+ }
108+ if plumb.Live() != 0 {
109+ t.Fatal("firewall attachments leaked")
110+ }
111+}
112+
113+func dockerLines(t *testing.T, args ...string) []string {
114+ t.Helper()
115+ out, err := exec.Command("docker", args...).Output()
116+ if err != nil {
117+ t.Fatalf("docker %v: %v", args, err)
118+ }
119+ return strings.Fields(string(out))
120+}
121+
122+func repoRoot(t *testing.T) string {
123+ t.Helper()
124+ out, err := exec.Command("go", "list", "-m", "-f", "{{.Dir}}").Output()
125+ if err != nil {
126+ t.Fatal(err)
127+ }
128+ return strings.TrimSpace(string(out))
129+}
A internal/sandbox/container/driver.go +447 -0
@@ -0,0 +1,447 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package container is the container sandbox, for hosts without KVM: one
4+// container per job with its own Docker daemon inside (the sysbox runtime when
5+// the daemon lists it, else a privileged container), on its own bridge network
6+// and with its own scratch volume. Everything Docker-side carries the label
7+// sandbox.LabelJob so that Reap finds it after a crash.
8+//
9+// The executor stream is a Unix socket in a per-job directory of the runner's
10+// state directory, bind-mounted into the container. The runner executable is
11+// bind-mounted read-only. Both paths are host paths given to the Docker daemon:
12+// the daemon must run on the same host (and mount namespace) as the runner.
13+package container
14+
15+import (
16+ "context"
17+ "errors"
18+ "fmt"
19+ "net"
20+ "net/netip"
21+ "os"
22+ "path/filepath"
23+ "slices"
24+ "sync"
25+ "time"
26+
27+ "github.com/bnema/zerowrap"
28+
29+ "bnema.dev/ermite-runner/internal/cmdrun"
30+ "bnema.dev/ermite-runner/internal/netpolicy"
31+ "bnema.dev/ermite-runner/internal/sandbox"
32+ "bnema.dev/ermite-runner/internal/sandbox/netsetup"
33+)
34+
35+// SysboxRuntime is the name of the sysbox runtime in `docker info`.
36+const SysboxRuntime = "sysbox-runc"
37+
38+// bridgePrefix must match the "erbr*" wildcard of the nftables policy.
39+const bridgePrefix = "erbr"
40+
41+// Options are the testable knobs of the driver.
42+type Options struct {
43+ Run cmdrun.Runner
44+ // Net overrides the network plumbing (tests).
45+ Net netsetup.Plumbing
46+ // engine overrides the Docker client (tests).
47+ engine engine
48+ // allowRootless lets tests exercise the Engine API against the rootless
49+ // daemon of a workstation. Never set outside tests: a nested dockerd cannot
50+ // work in a rootless daemon.
51+ allowRootless bool
52+ // ConnectTimeout bounds container start + init + executor connection
53+ // (default 90 s).
54+ ConnectTimeout time.Duration
55+}
56+
57+// Driver implements sandbox.Driver for containers.
58+type Driver struct {
59+ cfg sandbox.Config
60+ opt Options
61+ eng engine
62+ net netsetup.Plumbing
63+ image string
64+ runtime string // "" when privileged
65+ privileged bool
66+ exe string
67+
68+ mu sync.Mutex
69+ slots *sandbox.Slots
70+}
71+
72+// New connects to the Docker daemon (DOCKER_HOST honoured) and checks it: not
73+// rootless, the job image present (pulled if missing), the runtime decided.
74+func New(ctx context.Context, cfg sandbox.Config, opt Options) (*Driver, error) {
75+ if opt.Run == nil {
76+ opt.Run = cmdrun.Exec{}
77+ }
78+ if opt.ConnectTimeout == 0 {
79+ opt.ConnectTimeout = 90 * time.Second
80+ }
81+ eng := opt.engine
82+ if eng == nil {
83+ e, err := newDockerEngine()
84+ if err != nil {
85+ return nil, err
86+ }
87+ eng = e
88+ }
89+ d := &Driver{cfg: cfg, opt: opt, eng: eng, image: cfg.OCIImage(), slots: sandbox.NewSlots(cfg.Capacity)}
90+ fail := func(err error) (*Driver, error) { _ = eng.Close(); return nil, err }
91+
92+ exe, err := cfg.ExecutablePath()
93+ if err != nil {
94+ return fail(err)
95+ }
96+ if err := sandbox.CheckStatic(exe); err != nil {
97+ return fail(err)
98+ }
99+ d.exe = exe
100+
101+ info, err := eng.Info(ctx)
102+ if err != nil {
103+ return fail(fmt.Errorf("cannot reach the Docker daemon (DOCKER_HOST): %w", err))
104+ }
105+ if info.Rootless && !opt.allowRootless {
106+ return fail(errRootless)
107+ }
108+ if info.OSType != "" && info.OSType != "linux" {
109+ return fail(fmt.Errorf("the Docker daemon runs %s containers, need linux", info.OSType))
110+ }
111+ log := zerowrap.FromCtx(ctx)
112+ if slices.Contains(info.Runtimes, SysboxRuntime) {
113+ d.runtime = SysboxRuntime
114+ log.Info().Str("runtime", SysboxRuntime).Msg("container isolation: using the sysbox runtime")
115+ } else {
116+ d.privileged = true
117+ log.Warn().Msg("container isolation: sysbox-runc not available, jobs run in PRIVILEGED containers, which is host-root equivalent; their network policy is best effort. Prefer vm isolation or install sysbox")
118+ }
119+ if err := eng.EnsureImage(ctx, d.image); err != nil {
120+ return fail(err)
121+ }
122+ d.net = opt.Net
123+ if d.net == nil {
124+ d.net = netsetup.New(opt.Run, cfg.DNSPort, "")
125+ }
126+ if err := d.net.EnableForwarding(); err != nil {
127+ return fail(err)
128+ }
129+ return d, nil
130+}
131+
132+// Mode implements sandbox.Driver.
133+func (d *Driver) Mode() sandbox.Mode { return sandbox.Container }
134+
135+// Privileged reports whether jobs run privileged (no sysbox).
136+func (d *Driver) Privileged() bool { return d.privileged }
137+
138+// Close implements sandbox.Driver.
139+func (d *Driver) Close() error {
140+ if d.net != nil {
141+ d.net.Close()
142+ }
143+ return d.eng.Close()
144+}
145+
146+// New implements sandbox.Driver.
147+func (d *Driver) New() (sandbox.Sandbox, error) {
148+ d.mu.Lock()
149+ defer d.mu.Unlock()
150+ slot, err := d.slots.Acquire()
151+ if err != nil {
152+ return nil, err
153+ }
154+ return &containerSandbox{d: d, slot: slot}, nil
155+}
156+
157+func (d *Driver) release(slot int) {
158+ d.mu.Lock()
159+ d.slots.Release(slot)
160+ d.mu.Unlock()
161+}
162+
163+func (d *Driver) runDir(id string) string { return filepath.Join(d.cfg.StateDir, "containers", id) }
164+
165+type containerSandbox struct {
166+ d *Driver
167+ slot int
168+
169+ mu sync.Mutex
170+ started bool
171+ stopped bool
172+ id string
173+ netID string
174+ volume string
175+ container string
176+ listener net.Listener
177+ conn net.Conn
178+ iface string
179+ gateway netip.Addr
180+ attached bool
181+}
182+
183+// Start implements sandbox.Sandbox.
184+func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
185+ s.mu.Lock()
186+ defer s.mu.Unlock()
187+ if s.started || s.stopped {
188+ return nil, errors.New("container sandbox: Start called twice or after Stop")
189+ }
190+ s.started = true
191+ d := s.d
192+ log := zerowrap.FromCtx(ctx)
193+ spec = d.cfg.Resolve(spec)
194+
195+ id, err := sandbox.NewID()
196+ if err != nil {
197+ s.releaseLocked()
198+ return nil, err
199+ }
200+ s.id = id
201+ conn, err := s.run(ctx, spec)
202+ if err != nil {
203+ var tail string
204+ if s.container != "" {
205+ tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
206+ }
207+ cerr := s.cleanupLocked(context.WithoutCancel(ctx))
208+ if tail != "" {
209+ err = fmt.Errorf("%w\ncontainer log tail:\n%s", err, tail)
210+ }
211+ return nil, errors.Join(err, cerr)
212+ }
213+ log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")
214+ return conn, nil
215+}
216+
217+func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
218+ d := s.d
219+ netw, err := netpolicy.Subnet(d.cfg.Network, s.slot)
220+ if err != nil {
221+ return nil, err
222+ }
223+ labels := map[string]string{sandbox.LabelJob: spec.JobID}
224+ name := "ermite-" + s.id
225+
226+ // Per-job directory with the executor socket. It is created before anything
227+ // on the Docker side so that a crash leaves a recognisable directory.
228+ dir := d.runDir(s.id)
229+ if err := os.MkdirAll(dir, 0o711); err != nil {
230+ return nil, err
231+ }
232+ if err := os.Chmod(dir, 0o711); err != nil {
233+ return nil, err
234+ }
235+ sock := filepath.Join(dir, sandbox.ContainerSocket)
236+ ln, err := net.Listen("unix", sock)
237+ if err != nil {
238+ return nil, fmt.Errorf("listen executor socket: %w", err)
239+ }
240+ s.listener = ln
241+ // The container's root may be a mapped (sysbox) user: it must be able to
242+ // connect. The path is private to the job (random id, 0711 directory).
243+ if err := os.Chmod(sock, 0o666); err != nil {
244+ return nil, err
245+ }
246+ accepted := make(chan struct {
247+ c net.Conn
248+ err error
249+ }, 1)
250+ go func() {
251+ c, err := ln.Accept()
252+ accepted <- struct {
253+ c net.Conn
254+ err error
255+ }{c, err}
256+ }()
257+
258+ s.iface = fmt.Sprintf("%s%d", bridgePrefix, s.slot)
259+ s.gateway = netw.Gateway
260+ netID, err := d.eng.CreateNetwork(ctx, networkSpec{Name: name, Bridge: s.iface, Subnet: netw.Prefix, Gateway: netw.Gateway, Labels: labels})
261+ if err != nil {
262+ return nil, err
263+ }
264+ s.netID = netID
265+ s.attached = true
266+ if err := d.net.Attach(ctx, netpolicy.Guest{Iface: s.iface, Gateway: netw.Gateway, Addr: netw.Guest}); err != nil {
267+ return nil, err
268+ }
269+
270+ s.volume = name
271+ if err := d.eng.CreateVolume(ctx, name, labels); err != nil {
272+ s.volume = ""
273+ return nil, fmt.Errorf("create volume: %w", err)
274+ }
275+
276+ entry := sandbox.GuestExecDir + "/" + sandbox.GuestExecName
277+ cid, err := d.eng.CreateContainer(ctx, containerSpec{
278+ Name: name,
279+ Image: d.image,
280+ Entrypoint: []string{entry, "init"},
281+ Env: []string{"ERMITE_CONNECT=unix:" + sandbox.ContainerRunDir + "/" + sandbox.ContainerSocket},
282+ Labels: labels,
283+ Runtime: d.runtime,
284+ Privileged: d.privileged,
285+ Network: name,
286+ IPv4: netw.Guest,
287+ DNS: netw.Gateway,
288+ Binds: []bindMount{
289+ {Source: d.exe, Target: entry, ReadOnly: true},
290+ {Source: dir, Target: sandbox.ContainerRunDir},
291+ },
292+ Volumes: []volumeMount{{Name: name, Target: sandbox.GuestScratch}},
293+ MemoryMiB: spec.MemoryMiB,
294+ CPUs: spec.CPUs,
295+ })
296+ if err != nil {
297+ return nil, err
298+ }
299+ s.container = cid
300+ exited := d.eng.WaitExit(ctx, cid)
301+ if err := d.eng.StartContainer(ctx, cid); err != nil {
302+ return nil, fmt.Errorf("start container: %w", err)
303+ }
304+
305+ timer := time.NewTimer(d.opt.ConnectTimeout)
306+ defer timer.Stop()
307+ select {
308+ case r := <-accepted:
309+ if r.err != nil {
310+ return nil, fmt.Errorf("accept executor connection: %w", r.err)
311+ }
312+ s.conn = r.c
313+ return r.c, nil
314+ case ex := <-exited:
315+ return nil, fmt.Errorf("container exited (code %d) before the executor connected", ex.Code)
316+ case <-timer.C:
317+ return nil, fmt.Errorf("executor did not connect within %s", d.opt.ConnectTimeout)
318+ case <-ctx.Done():
319+ return nil, ctx.Err()
320+ }
321+}
322+
323+// Stop implements sandbox.Sandbox.
324+func (s *containerSandbox) Stop(ctx context.Context) error {
325+ s.mu.Lock()
326+ defer s.mu.Unlock()
327+ if s.stopped {
328+ return nil
329+ }
330+ if !s.started {
331+ s.releaseLocked()
332+ return nil
333+ }
334+ return s.cleanupLocked(context.WithoutCancel(ctx))
335+}
336+
337+func (s *containerSandbox) releaseLocked() {
338+ s.stopped = true
339+ s.d.release(s.slot)
340+}
341+
342+// cleanupLocked removes the container, then its network, volume, firewall rules
343+// and directory, continuing after errors. The slot is released only when
344+// everything is gone (so its bridge name and subnet are not reused while the
345+// old ones still exist); leftovers are reaped by label at the next start.
346+func (s *containerSandbox) cleanupLocked(ctx context.Context) error {
347+ d := s.d
348+ var errs []error
349+ if s.conn != nil {
350+ _ = s.conn.Close()
351+ s.conn = nil
352+ }
353+ if s.listener != nil {
354+ _ = s.listener.Close()
355+ s.listener = nil
356+ }
357+ if s.container != "" {
358+ if err := d.eng.RemoveContainer(ctx, s.container); err != nil {
359+ errs = append(errs, fmt.Errorf("remove container: %w", err))
360+ } else {
361+ s.container = ""
362+ }
363+ }
364+ if s.attached {
365+ if err := d.net.Detach(ctx, s.iface, s.gateway); err != nil {
366+ errs = append(errs, err)
367+ } else {
368+ s.attached = false
369+ }
370+ }
371+ if s.netID != "" && s.container == "" {
372+ if err := d.eng.RemoveNetwork(ctx, s.netID); err != nil {
373+ errs = append(errs, fmt.Errorf("remove network: %w", err))
374+ } else {
375+ s.netID = ""
376+ }
377+ }
378+ if s.volume != "" && s.container == "" {
379+ if err := d.eng.RemoveVolume(ctx, s.volume); err != nil {
380+ errs = append(errs, fmt.Errorf("remove volume: %w", err))
381+ } else {
382+ s.volume = ""
383+ }
384+ }
385+ if err := os.RemoveAll(d.runDir(s.id)); err != nil {
386+ errs = append(errs, err)
387+ }
388+ if len(errs) == 0 {
389+ s.releaseLocked()
390+ return nil
391+ }
392+ s.stopped = true
393+ err := errors.Join(errs...)
394+ log := zerowrap.FromCtx(ctx)
395+ log.Error().Err(err).Str("job", s.id).Msg("container cleanup incomplete; the next runner start will retry")
396+ return err
397+}
398+
399+// Reap implements sandbox.Driver: remove every container, network and volume
400+// carrying the job label, the nftables table, and the per-job directories.
401+func (d *Driver) Reap(ctx context.Context) error {
402+ log := zerowrap.FromCtx(ctx)
403+ var errs []error
404+ ids, err := d.eng.ListContainers(ctx, sandbox.LabelJob)
405+ if err != nil {
406+ errs = append(errs, fmt.Errorf("list containers: %w", err))
407+ }
408+ for _, id := range ids {
409+ log.Warn().Str("container", id).Msg("removing leftover job container")
410+ if err := d.eng.RemoveContainer(ctx, id); err != nil {
411+ errs = append(errs, err)
412+ }
413+ }
414+ nets, err := d.eng.ListNetworks(ctx, sandbox.LabelJob)
415+ if err != nil {
416+ errs = append(errs, fmt.Errorf("list networks: %w", err))
417+ }
418+ for _, n := range nets {
419+ log.Warn().Str("network", n).Msg("removing leftover job network")
420+ if err := d.eng.RemoveNetwork(ctx, n); err != nil {
421+ errs = append(errs, err)
422+ }
423+ }
424+ vols, err := d.eng.ListVolumes(ctx, sandbox.LabelJob)
425+ if err != nil {
426+ errs = append(errs, fmt.Errorf("list volumes: %w", err))
427+ }
428+ for _, v := range vols {
429+ log.Warn().Str("volume", v).Msg("removing leftover job volume")
430+ if err := d.eng.RemoveVolume(ctx, v); err != nil {
431+ errs = append(errs, err)
432+ }
433+ }
434+ if err := d.net.Reset(ctx); err != nil {
435+ errs = append(errs, fmt.Errorf("reset nftables table: %w", err))
436+ }
437+ root := filepath.Join(d.cfg.StateDir, "containers")
438+ ents, _ := os.ReadDir(root)
439+ for _, e := range ents {
440+ if sandbox.ValidID(e.Name()) {
441+ if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
442+ errs = append(errs, err)
443+ }
444+ }
445+ }
446+ return errors.Join(errs...)
447+}
@@ -0,0 +1,447 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package container is the container sandbox, for hosts without KVM: one
4+// container per job with its own Docker daemon inside (the sysbox runtime when
5+// the daemon lists it, else a privileged container), on its own bridge network
6+// and with its own scratch volume. Everything Docker-side carries the label
7+// sandbox.LabelJob so that Reap finds it after a crash.
8+//
9+// The executor stream is a Unix socket in a per-job directory of the runner's
10+// state directory, bind-mounted into the container. The runner executable is
11+// bind-mounted read-only. Both paths are host paths given to the Docker daemon:
12+// the daemon must run on the same host (and mount namespace) as the runner.
13+package container
14+
15+import (
16+ "context"
17+ "errors"
18+ "fmt"
19+ "net"
20+ "net/netip"
21+ "os"
22+ "path/filepath"
23+ "slices"
24+ "sync"
25+ "time"
26+
27+ "github.com/bnema/zerowrap"
28+
29+ "bnema.dev/ermite-runner/internal/cmdrun"
30+ "bnema.dev/ermite-runner/internal/netpolicy"
31+ "bnema.dev/ermite-runner/internal/sandbox"
32+ "bnema.dev/ermite-runner/internal/sandbox/netsetup"
33+)
34+
35+// SysboxRuntime is the name of the sysbox runtime in `docker info`.
36+const SysboxRuntime = "sysbox-runc"
37+
38+// bridgePrefix must match the "erbr*" wildcard of the nftables policy.
39+const bridgePrefix = "erbr"
40+
41+// Options are the testable knobs of the driver.
42+type Options struct {
43+ Run cmdrun.Runner
44+ // Net overrides the network plumbing (tests).
45+ Net netsetup.Plumbing
46+ // engine overrides the Docker client (tests).
47+ engine engine
48+ // allowRootless lets tests exercise the Engine API against the rootless
49+ // daemon of a workstation. Never set outside tests: a nested dockerd cannot
50+ // work in a rootless daemon.
51+ allowRootless bool
52+ // ConnectTimeout bounds container start + init + executor connection
53+ // (default 90 s).
54+ ConnectTimeout time.Duration
55+}
56+
57+// Driver implements sandbox.Driver for containers.
58+type Driver struct {
59+ cfg sandbox.Config
60+ opt Options
61+ eng engine
62+ net netsetup.Plumbing
63+ image string
64+ runtime string // "" when privileged
65+ privileged bool
66+ exe string
67+
68+ mu sync.Mutex
69+ slots *sandbox.Slots
70+}
71+
72+// New connects to the Docker daemon (DOCKER_HOST honoured) and checks it: not
73+// rootless, the job image present (pulled if missing), the runtime decided.
74+func New(ctx context.Context, cfg sandbox.Config, opt Options) (*Driver, error) {
75+ if opt.Run == nil {
76+ opt.Run = cmdrun.Exec{}
77+ }
78+ if opt.ConnectTimeout == 0 {
79+ opt.ConnectTimeout = 90 * time.Second
80+ }
81+ eng := opt.engine
82+ if eng == nil {
83+ e, err := newDockerEngine()
84+ if err != nil {
85+ return nil, err
86+ }
87+ eng = e
88+ }
89+ d := &Driver{cfg: cfg, opt: opt, eng: eng, image: cfg.OCIImage(), slots: sandbox.NewSlots(cfg.Capacity)}
90+ fail := func(err error) (*Driver, error) { _ = eng.Close(); return nil, err }
91+
92+ exe, err := cfg.ExecutablePath()
93+ if err != nil {
94+ return fail(err)
95+ }
96+ if err := sandbox.CheckStatic(exe); err != nil {
97+ return fail(err)
98+ }
99+ d.exe = exe
100+
101+ info, err := eng.Info(ctx)
102+ if err != nil {
103+ return fail(fmt.Errorf("cannot reach the Docker daemon (DOCKER_HOST): %w", err))
104+ }
105+ if info.Rootless && !opt.allowRootless {
106+ return fail(errRootless)
107+ }
108+ if info.OSType != "" && info.OSType != "linux" {
109+ return fail(fmt.Errorf("the Docker daemon runs %s containers, need linux", info.OSType))
110+ }
111+ log := zerowrap.FromCtx(ctx)
112+ if slices.Contains(info.Runtimes, SysboxRuntime) {
113+ d.runtime = SysboxRuntime
114+ log.Info().Str("runtime", SysboxRuntime).Msg("container isolation: using the sysbox runtime")
115+ } else {
116+ d.privileged = true
117+ log.Warn().Msg("container isolation: sysbox-runc not available, jobs run in PRIVILEGED containers, which is host-root equivalent; their network policy is best effort. Prefer vm isolation or install sysbox")
118+ }
119+ if err := eng.EnsureImage(ctx, d.image); err != nil {
120+ return fail(err)
121+ }
122+ d.net = opt.Net
123+ if d.net == nil {
124+ d.net = netsetup.New(opt.Run, cfg.DNSPort, "")
125+ }
126+ if err := d.net.EnableForwarding(); err != nil {
127+ return fail(err)
128+ }
129+ return d, nil
130+}
131+
132+// Mode implements sandbox.Driver.
133+func (d *Driver) Mode() sandbox.Mode { return sandbox.Container }
134+
135+// Privileged reports whether jobs run privileged (no sysbox).
136+func (d *Driver) Privileged() bool { return d.privileged }
137+
138+// Close implements sandbox.Driver.
139+func (d *Driver) Close() error {
140+ if d.net != nil {
141+ d.net.Close()
142+ }
143+ return d.eng.Close()
144+}
145+
146+// New implements sandbox.Driver.
147+func (d *Driver) New() (sandbox.Sandbox, error) {
148+ d.mu.Lock()
149+ defer d.mu.Unlock()
150+ slot, err := d.slots.Acquire()
151+ if err != nil {
152+ return nil, err
153+ }
154+ return &containerSandbox{d: d, slot: slot}, nil
155+}
156+
157+func (d *Driver) release(slot int) {
158+ d.mu.Lock()
159+ d.slots.Release(slot)
160+ d.mu.Unlock()
161+}
162+
163+func (d *Driver) runDir(id string) string { return filepath.Join(d.cfg.StateDir, "containers", id) }
164+
165+type containerSandbox struct {
166+ d *Driver
167+ slot int
168+
169+ mu sync.Mutex
170+ started bool
171+ stopped bool
172+ id string
173+ netID string
174+ volume string
175+ container string
176+ listener net.Listener
177+ conn net.Conn
178+ iface string
179+ gateway netip.Addr
180+ attached bool
181+}
182+
183+// Start implements sandbox.Sandbox.
184+func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
185+ s.mu.Lock()
186+ defer s.mu.Unlock()
187+ if s.started || s.stopped {
188+ return nil, errors.New("container sandbox: Start called twice or after Stop")
189+ }
190+ s.started = true
191+ d := s.d
192+ log := zerowrap.FromCtx(ctx)
193+ spec = d.cfg.Resolve(spec)
194+
195+ id, err := sandbox.NewID()
196+ if err != nil {
197+ s.releaseLocked()
198+ return nil, err
199+ }
200+ s.id = id
201+ conn, err := s.run(ctx, spec)
202+ if err != nil {
203+ var tail string
204+ if s.container != "" {
205+ tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
206+ }
207+ cerr := s.cleanupLocked(context.WithoutCancel(ctx))
208+ if tail != "" {
209+ err = fmt.Errorf("%w\ncontainer log tail:\n%s", err, tail)
210+ }
211+ return nil, errors.Join(err, cerr)
212+ }
213+ log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")
214+ return conn, nil
215+}
216+
217+func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
218+ d := s.d
219+ netw, err := netpolicy.Subnet(d.cfg.Network, s.slot)
220+ if err != nil {
221+ return nil, err
222+ }
223+ labels := map[string]string{sandbox.LabelJob: spec.JobID}
224+ name := "ermite-" + s.id
225+
226+ // Per-job directory with the executor socket. It is created before anything
227+ // on the Docker side so that a crash leaves a recognisable directory.
228+ dir := d.runDir(s.id)
229+ if err := os.MkdirAll(dir, 0o711); err != nil {
230+ return nil, err
231+ }
232+ if err := os.Chmod(dir, 0o711); err != nil {
233+ return nil, err
234+ }
235+ sock := filepath.Join(dir, sandbox.ContainerSocket)
236+ ln, err := net.Listen("unix", sock)
237+ if err != nil {
238+ return nil, fmt.Errorf("listen executor socket: %w", err)
239+ }
240+ s.listener = ln
241+ // The container's root may be a mapped (sysbox) user: it must be able to
242+ // connect. The path is private to the job (random id, 0711 directory).
243+ if err := os.Chmod(sock, 0o666); err != nil {
244+ return nil, err
245+ }
246+ accepted := make(chan struct {
247+ c net.Conn
248+ err error
249+ }, 1)
250+ go func() {
251+ c, err := ln.Accept()
252+ accepted <- struct {
253+ c net.Conn
254+ err error
255+ }{c, err}
256+ }()
257+
258+ s.iface = fmt.Sprintf("%s%d", bridgePrefix, s.slot)
259+ s.gateway = netw.Gateway
260+ netID, err := d.eng.CreateNetwork(ctx, networkSpec{Name: name, Bridge: s.iface, Subnet: netw.Prefix, Gateway: netw.Gateway, Labels: labels})
261+ if err != nil {
262+ return nil, err
263+ }
264+ s.netID = netID
265+ s.attached = true
266+ if err := d.net.Attach(ctx, netpolicy.Guest{Iface: s.iface, Gateway: netw.Gateway, Addr: netw.Guest}); err != nil {
267+ return nil, err
268+ }
269+
270+ s.volume = name
271+ if err := d.eng.CreateVolume(ctx, name, labels); err != nil {
272+ s.volume = ""
273+ return nil, fmt.Errorf("create volume: %w", err)
274+ }
275+
276+ entry := sandbox.GuestExecDir + "/" + sandbox.GuestExecName
277+ cid, err := d.eng.CreateContainer(ctx, containerSpec{
278+ Name: name,
279+ Image: d.image,
280+ Entrypoint: []string{entry, "init"},
281+ Env: []string{"ERMITE_CONNECT=unix:" + sandbox.ContainerRunDir + "/" + sandbox.ContainerSocket},
282+ Labels: labels,
283+ Runtime: d.runtime,
284+ Privileged: d.privileged,
285+ Network: name,
286+ IPv4: netw.Guest,
287+ DNS: netw.Gateway,
288+ Binds: []bindMount{
289+ {Source: d.exe, Target: entry, ReadOnly: true},
290+ {Source: dir, Target: sandbox.ContainerRunDir},
291+ },
292+ Volumes: []volumeMount{{Name: name, Target: sandbox.GuestScratch}},
293+ MemoryMiB: spec.MemoryMiB,
294+ CPUs: spec.CPUs,
295+ })
296+ if err != nil {
297+ return nil, err
298+ }
299+ s.container = cid
300+ exited := d.eng.WaitExit(ctx, cid)
301+ if err := d.eng.StartContainer(ctx, cid); err != nil {
302+ return nil, fmt.Errorf("start container: %w", err)
303+ }
304+
305+ timer := time.NewTimer(d.opt.ConnectTimeout)
306+ defer timer.Stop()
307+ select {
308+ case r := <-accepted:
309+ if r.err != nil {
310+ return nil, fmt.Errorf("accept executor connection: %w", r.err)
311+ }
312+ s.conn = r.c
313+ return r.c, nil
314+ case ex := <-exited:
315+ return nil, fmt.Errorf("container exited (code %d) before the executor connected", ex.Code)
316+ case <-timer.C:
317+ return nil, fmt.Errorf("executor did not connect within %s", d.opt.ConnectTimeout)
318+ case <-ctx.Done():
319+ return nil, ctx.Err()
320+ }
321+}
322+
323+// Stop implements sandbox.Sandbox.
324+func (s *containerSandbox) Stop(ctx context.Context) error {
325+ s.mu.Lock()
326+ defer s.mu.Unlock()
327+ if s.stopped {
328+ return nil
329+ }
330+ if !s.started {
331+ s.releaseLocked()
332+ return nil
333+ }
334+ return s.cleanupLocked(context.WithoutCancel(ctx))
335+}
336+
337+func (s *containerSandbox) releaseLocked() {
338+ s.stopped = true
339+ s.d.release(s.slot)
340+}
341+
342+// cleanupLocked removes the container, then its network, volume, firewall rules
343+// and directory, continuing after errors. The slot is released only when
344+// everything is gone (so its bridge name and subnet are not reused while the
345+// old ones still exist); leftovers are reaped by label at the next start.
346+func (s *containerSandbox) cleanupLocked(ctx context.Context) error {
347+ d := s.d
348+ var errs []error
349+ if s.conn != nil {
350+ _ = s.conn.Close()
351+ s.conn = nil
352+ }
353+ if s.listener != nil {
354+ _ = s.listener.Close()
355+ s.listener = nil
356+ }
357+ if s.container != "" {
358+ if err := d.eng.RemoveContainer(ctx, s.container); err != nil {
359+ errs = append(errs, fmt.Errorf("remove container: %w", err))
360+ } else {
361+ s.container = ""
362+ }
363+ }
364+ if s.attached {
365+ if err := d.net.Detach(ctx, s.iface, s.gateway); err != nil {
366+ errs = append(errs, err)
367+ } else {
368+ s.attached = false
369+ }
370+ }
371+ if s.netID != "" && s.container == "" {
372+ if err := d.eng.RemoveNetwork(ctx, s.netID); err != nil {
373+ errs = append(errs, fmt.Errorf("remove network: %w", err))
374+ } else {
375+ s.netID = ""
376+ }
377+ }
378+ if s.volume != "" && s.container == "" {
379+ if err := d.eng.RemoveVolume(ctx, s.volume); err != nil {
380+ errs = append(errs, fmt.Errorf("remove volume: %w", err))
381+ } else {
382+ s.volume = ""
383+ }
384+ }
385+ if err := os.RemoveAll(d.runDir(s.id)); err != nil {
386+ errs = append(errs, err)
387+ }
388+ if len(errs) == 0 {
389+ s.releaseLocked()
390+ return nil
391+ }
392+ s.stopped = true
393+ err := errors.Join(errs...)
394+ log := zerowrap.FromCtx(ctx)
395+ log.Error().Err(err).Str("job", s.id).Msg("container cleanup incomplete; the next runner start will retry")
396+ return err
397+}
398+
399+// Reap implements sandbox.Driver: remove every container, network and volume
400+// carrying the job label, the nftables table, and the per-job directories.
401+func (d *Driver) Reap(ctx context.Context) error {
402+ log := zerowrap.FromCtx(ctx)
403+ var errs []error
404+ ids, err := d.eng.ListContainers(ctx, sandbox.LabelJob)
405+ if err != nil {
406+ errs = append(errs, fmt.Errorf("list containers: %w", err))
407+ }
408+ for _, id := range ids {
409+ log.Warn().Str("container", id).Msg("removing leftover job container")
410+ if err := d.eng.RemoveContainer(ctx, id); err != nil {
411+ errs = append(errs, err)
412+ }
413+ }
414+ nets, err := d.eng.ListNetworks(ctx, sandbox.LabelJob)
415+ if err != nil {
416+ errs = append(errs, fmt.Errorf("list networks: %w", err))
417+ }
418+ for _, n := range nets {
419+ log.Warn().Str("network", n).Msg("removing leftover job network")
420+ if err := d.eng.RemoveNetwork(ctx, n); err != nil {
421+ errs = append(errs, err)
422+ }
423+ }
424+ vols, err := d.eng.ListVolumes(ctx, sandbox.LabelJob)
425+ if err != nil {
426+ errs = append(errs, fmt.Errorf("list volumes: %w", err))
427+ }
428+ for _, v := range vols {
429+ log.Warn().Str("volume", v).Msg("removing leftover job volume")
430+ if err := d.eng.RemoveVolume(ctx, v); err != nil {
431+ errs = append(errs, err)
432+ }
433+ }
434+ if err := d.net.Reset(ctx); err != nil {
435+ errs = append(errs, fmt.Errorf("reset nftables table: %w", err))
436+ }
437+ root := filepath.Join(d.cfg.StateDir, "containers")
438+ ents, _ := os.ReadDir(root)
439+ for _, e := range ents {
440+ if sandbox.ValidID(e.Name()) {
441+ if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
442+ errs = append(errs, err)
443+ }
444+ }
445+ }
446+ return errors.Join(errs...)
447+}
A internal/sandbox/container/engine.go +291 -0
@@ -0,0 +1,291 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "io"
10+ "net/netip"
11+
12+ cerrdefs "github.com/containerd/errdefs"
13+ "github.com/moby/moby/api/types/container"
14+ "github.com/moby/moby/api/types/mount"
15+ "github.com/moby/moby/api/types/network"
16+ "github.com/moby/moby/client"
17+)
18+
19+// engine is the part of the Docker Engine API the driver uses. The real
20+// implementation wraps github.com/moby/moby/client (the maintained client of
21+// the Moby project; github.com/docker/docker/client is its frozen
22+// predecessor); tests use a fake.
23+type engine interface {
24+ Info(ctx context.Context) (daemonInfo, error)
25+ // EnsureImage returns nil when ref exists locally, pulling it otherwise.
26+ EnsureImage(ctx context.Context, ref string) error
27+ CreateNetwork(ctx context.Context, spec networkSpec) (string, error)
28+ RemoveNetwork(ctx context.Context, id string) error
29+ ListNetworks(ctx context.Context, labelKey string) ([]string, error)
30+ CreateVolume(ctx context.Context, name string, labels map[string]string) error
31+ RemoveVolume(ctx context.Context, name string) error
32+ ListVolumes(ctx context.Context, labelKey string) ([]string, error)
33+ CreateContainer(ctx context.Context, spec containerSpec) (string, error)
34+ StartContainer(ctx context.Context, id string) error
35+ // WaitExit returns a channel that yields the exit code once the container
36+ // stops running (or -1 and an error when the wait itself fails).
37+ WaitExit(ctx context.Context, id string) <-chan exitResult
38+ RemoveContainer(ctx context.Context, id string) error
39+ ListContainers(ctx context.Context, labelKey string) ([]string, error)
40+ LogsTail(ctx context.Context, id string, lines int) (string, error)
41+ Close() error
42+}
43+
44+type exitResult struct {
45+ Code int
46+ Err error
47+}
48+
49+type daemonInfo struct {
50+ Rootless bool
51+ Runtimes []string
52+ OSType string
53+}
54+
55+type networkSpec struct {
56+ Name string
57+ Bridge string // host bridge device name
58+ Subnet netip.Prefix
59+ Gateway netip.Addr
60+ Labels map[string]string
61+}
62+
63+type containerSpec struct {
64+ Name string
65+ Image string
66+ Entrypoint []string
67+ Env []string
68+ Labels map[string]string
69+ Runtime string // "" = daemon default
70+ Privileged bool
71+ Network string
72+ IPv4 netip.Addr
73+ DNS netip.Addr
74+ Binds []bindMount
75+ Volumes []volumeMount
76+ MemoryMiB int
77+ CPUs int
78+}
79+
80+type bindMount struct {
81+ Source, Target string
82+ ReadOnly bool
83+}
84+
85+type volumeMount struct{ Name, Target string }
86+
87+// dockerEngine implements engine with the moby client.
88+type dockerEngine struct{ cli *client.Client }
89+
90+// newDockerEngine connects like the docker CLI: DOCKER_HOST, DOCKER_TLS_VERIFY,
91+// DOCKER_CERT_PATH, DOCKER_API_VERSION, with API version negotiation.
92+func newDockerEngine() (*dockerEngine, error) {
93+ cli, err := client.New(client.FromEnv)
94+ if err != nil {
95+ return nil, fmt.Errorf("docker client: %w", err)
96+ }
97+ return &dockerEngine{cli: cli}, nil
98+}
99+
100+func (e *dockerEngine) Close() error { return e.cli.Close() }
101+
102+func (e *dockerEngine) Info(ctx context.Context) (daemonInfo, error) {
103+ r, err := e.cli.Info(ctx, client.InfoOptions{})
104+ if err != nil {
105+ return daemonInfo{}, fmt.Errorf("docker info: %w", err)
106+ }
107+ di := daemonInfo{OSType: r.Info.OSType}
108+ for _, o := range r.Info.SecurityOptions {
109+ if o == "name=rootless" {
110+ di.Rootless = true
111+ }
112+ }
113+ for name := range r.Info.Runtimes {
114+ di.Runtimes = append(di.Runtimes, name)
115+ }
116+ return di, nil
117+}
118+
119+func (e *dockerEngine) EnsureImage(ctx context.Context, ref string) error {
120+ if _, err := e.cli.ImageInspect(ctx, ref); err == nil {
121+ return nil
122+ } else if !cerrdefs.IsNotFound(err) {
123+ return fmt.Errorf("inspect image %s: %w", ref, err)
124+ }
125+ rc, err := e.cli.ImagePull(ctx, ref, client.ImagePullOptions{})
126+ if err != nil {
127+ return fmt.Errorf("image %s is not present and cannot be pulled (build it with scripts/build-image.sh): %w", ref, err)
128+ }
129+ defer func() { _ = rc.Close() }()
130+ if err := rc.Wait(ctx); err != nil {
131+ return fmt.Errorf("pull image %s: %w", ref, err)
132+ }
133+ return nil
134+}
135+
136+func (e *dockerEngine) CreateNetwork(ctx context.Context, s networkSpec) (string, error) {
137+ f := false
138+ r, err := e.cli.NetworkCreate(ctx, s.Name, client.NetworkCreateOptions{
139+ Driver: "bridge",
140+ EnableIPv6: &f,
141+ IPAM: &network.IPAM{Config: []network.IPAMConfig{{
142+ Subnet: s.Subnet, Gateway: s.Gateway,
143+ }}},
144+ Options: map[string]string{
145+ "com.docker.network.bridge.name": s.Bridge,
146+ // The runner's nftables table does the NAT, with its own policy.
147+ "com.docker.network.bridge.enable_ip_masquerade": "false",
148+ },
149+ Labels: s.Labels,
150+ })
151+ if err != nil {
152+ return "", fmt.Errorf("create network %s: %w", s.Name, err)
153+ }
154+ return r.ID, nil
155+}
156+
157+func (e *dockerEngine) RemoveNetwork(ctx context.Context, id string) error {
158+ _, err := e.cli.NetworkRemove(ctx, id, client.NetworkRemoveOptions{})
159+ return ignoreNotFound(err)
160+}
161+
162+func (e *dockerEngine) ListNetworks(ctx context.Context, key string) ([]string, error) {
163+ r, err := e.cli.NetworkList(ctx, client.NetworkListOptions{Filters: client.Filters{}.Add("label", key)})
164+ if err != nil {
165+ return nil, err
166+ }
167+ var ids []string
168+ for _, n := range r.Items {
169+ ids = append(ids, n.ID)
170+ }
171+ return ids, nil
172+}
173+
174+func (e *dockerEngine) CreateVolume(ctx context.Context, name string, labels map[string]string) error {
175+ _, err := e.cli.VolumeCreate(ctx, client.VolumeCreateOptions{Name: name, Labels: labels})
176+ return err
177+}
178+
179+func (e *dockerEngine) RemoveVolume(ctx context.Context, name string) error {
180+ _, err := e.cli.VolumeRemove(ctx, name, client.VolumeRemoveOptions{Force: true})
181+ return ignoreNotFound(err)
182+}
183+
184+func (e *dockerEngine) ListVolumes(ctx context.Context, key string) ([]string, error) {
185+ r, err := e.cli.VolumeList(ctx, client.VolumeListOptions{Filters: client.Filters{}.Add("label", key)})
186+ if err != nil {
187+ return nil, err
188+ }
189+ var names []string
190+ for _, v := range r.Items {
191+ names = append(names, v.Name)
192+ }
193+ return names, nil
194+}
195+
196+func (e *dockerEngine) CreateContainer(ctx context.Context, s containerSpec) (string, error) {
197+ hc := &container.HostConfig{
198+ NetworkMode: container.NetworkMode(s.Network),
199+ Privileged: s.Privileged,
200+ Runtime: s.Runtime,
201+ CgroupnsMode: container.CgroupnsModePrivate,
202+ DNS: []netip.Addr{s.DNS},
203+ Sysctls: map[string]string{"net.ipv6.conf.all.disable_ipv6": "1", "net.ipv6.conf.default.disable_ipv6": "1"},
204+ }
205+ if s.MemoryMiB > 0 {
206+ hc.Memory = int64(s.MemoryMiB) << 20
207+ }
208+ if s.CPUs > 0 {
209+ hc.NanoCPUs = int64(s.CPUs) * 1_000_000_000
210+ }
211+ for _, b := range s.Binds {
212+ hc.Mounts = append(hc.Mounts, mount.Mount{Type: mount.TypeBind, Source: b.Source, Target: b.Target, ReadOnly: b.ReadOnly,
213+ BindOptions: &mount.BindOptions{CreateMountpoint: true}})
214+ }
215+ for _, v := range s.Volumes {
216+ hc.Mounts = append(hc.Mounts, mount.Mount{Type: mount.TypeVolume, Source: v.Name, Target: v.Target})
217+ }
218+ cfg := &container.Config{
219+ Image: s.Image,
220+ Entrypoint: s.Entrypoint,
221+ Cmd: []string{},
222+ Env: s.Env,
223+ Labels: s.Labels,
224+ Hostname: "ermite-job",
225+ }
226+ nc := &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{
227+ s.Network: {IPAMConfig: &network.EndpointIPAMConfig{IPv4Address: s.IPv4}},
228+ }}
229+ r, err := e.cli.ContainerCreate(ctx, client.ContainerCreateOptions{Name: s.Name, Config: cfg, HostConfig: hc, NetworkingConfig: nc})
230+ if err != nil {
231+ return "", fmt.Errorf("create container: %w", err)
232+ }
233+ return r.ID, nil
234+}
235+
236+func (e *dockerEngine) StartContainer(ctx context.Context, id string) error {
237+ _, err := e.cli.ContainerStart(ctx, id, client.ContainerStartOptions{})
238+ return err
239+}
240+
241+func (e *dockerEngine) WaitExit(ctx context.Context, id string) <-chan exitResult {
242+ out := make(chan exitResult, 1)
243+ w := e.cli.ContainerWait(ctx, id, client.ContainerWaitOptions{Condition: container.WaitConditionNextExit})
244+ go func() {
245+ select {
246+ case r := <-w.Result:
247+ out <- exitResult{Code: int(r.StatusCode)}
248+ case err := <-w.Error:
249+ out <- exitResult{Code: -1, Err: err}
250+ case <-ctx.Done():
251+ out <- exitResult{Code: -1, Err: ctx.Err()}
252+ }
253+ }()
254+ return out
255+}
256+
257+func (e *dockerEngine) RemoveContainer(ctx context.Context, id string) error {
258+ _, err := e.cli.ContainerRemove(ctx, id, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
259+ return ignoreNotFound(err)
260+}
261+
262+func (e *dockerEngine) ListContainers(ctx context.Context, key string) ([]string, error) {
263+ r, err := e.cli.ContainerList(ctx, client.ContainerListOptions{All: true, Filters: client.Filters{}.Add("label", key)})
264+ if err != nil {
265+ return nil, err
266+ }
267+ var ids []string
268+ for _, c := range r.Items {
269+ ids = append(ids, c.ID)
270+ }
271+ return ids, nil
272+}
273+
274+func (e *dockerEngine) LogsTail(ctx context.Context, id string, lines int) (string, error) {
275+ rc, err := e.cli.ContainerLogs(ctx, id, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: fmt.Sprint(lines)})
276+ if err != nil {
277+ return "", err
278+ }
279+ defer func() { _ = rc.Close() }()
280+ b, err := io.ReadAll(io.LimitReader(rc, 64<<10))
281+ return string(b), err
282+}
283+
284+func ignoreNotFound(err error) error {
285+ if err == nil || cerrdefs.IsNotFound(err) {
286+ return nil
287+ }
288+ return err
289+}
290+
291+var errRootless = errors.New("the Docker daemon is rootless: a job needs its own nested dockerd, which requires a rootful daemon (or the sysbox runtime); point DOCKER_HOST at a rootful daemon or use ERMITE_RUNNER_ISOLATION=vm")
@@ -0,0 +1,291 @@
1+// SPDX-License-Identifier: MIT
2+
3+package container
4+
5+import (
6+ "context"
7+ "errors"
8+ "fmt"
9+ "io"
10+ "net/netip"
11+
12+ cerrdefs "github.com/containerd/errdefs"
13+ "github.com/moby/moby/api/types/container"
14+ "github.com/moby/moby/api/types/mount"
15+ "github.com/moby/moby/api/types/network"
16+ "github.com/moby/moby/client"
17+)
18+
19+// engine is the part of the Docker Engine API the driver uses. The real
20+// implementation wraps github.com/moby/moby/client (the maintained client of
21+// the Moby project; github.com/docker/docker/client is its frozen
22+// predecessor); tests use a fake.
23+type engine interface {
24+ Info(ctx context.Context) (daemonInfo, error)
25+ // EnsureImage returns nil when ref exists locally, pulling it otherwise.
26+ EnsureImage(ctx context.Context, ref string) error
27+ CreateNetwork(ctx context.Context, spec networkSpec) (string, error)
28+ RemoveNetwork(ctx context.Context, id string) error
29+ ListNetworks(ctx context.Context, labelKey string) ([]string, error)
30+ CreateVolume(ctx context.Context, name string, labels map[string]string) error
31+ RemoveVolume(ctx context.Context, name string) error
32+ ListVolumes(ctx context.Context, labelKey string) ([]string, error)
33+ CreateContainer(ctx context.Context, spec containerSpec) (string, error)
34+ StartContainer(ctx context.Context, id string) error
35+ // WaitExit returns a channel that yields the exit code once the container
36+ // stops running (or -1 and an error when the wait itself fails).
37+ WaitExit(ctx context.Context, id string) <-chan exitResult
38+ RemoveContainer(ctx context.Context, id string) error
39+ ListContainers(ctx context.Context, labelKey string) ([]string, error)
40+ LogsTail(ctx context.Context, id string, lines int) (string, error)
41+ Close() error
42+}
43+
44+type exitResult struct {
45+ Code int
46+ Err error
47+}
48+
49+type daemonInfo struct {
50+ Rootless bool
51+ Runtimes []string
52+ OSType string
53+}
54+
55+type networkSpec struct {
56+ Name string
57+ Bridge string // host bridge device name
58+ Subnet netip.Prefix
59+ Gateway netip.Addr
60+ Labels map[string]string
61+}
62+
63+type containerSpec struct {
64+ Name string
65+ Image string
66+ Entrypoint []string
67+ Env []string
68+ Labels map[string]string
69+ Runtime string // "" = daemon default
70+ Privileged bool
71+ Network string
72+ IPv4 netip.Addr
73+ DNS netip.Addr
74+ Binds []bindMount
75+ Volumes []volumeMount
76+ MemoryMiB int
77+ CPUs int
78+}
79+
80+type bindMount struct {
81+ Source, Target string
82+ ReadOnly bool
83+}
84+
85+type volumeMount struct{ Name, Target string }
86+
87+// dockerEngine implements engine with the moby client.
88+type dockerEngine struct{ cli *client.Client }
89+
90+// newDockerEngine connects like the docker CLI: DOCKER_HOST, DOCKER_TLS_VERIFY,
91+// DOCKER_CERT_PATH, DOCKER_API_VERSION, with API version negotiation.
92+func newDockerEngine() (*dockerEngine, error) {
93+ cli, err := client.New(client.FromEnv)
94+ if err != nil {
95+ return nil, fmt.Errorf("docker client: %w", err)
96+ }
97+ return &dockerEngine{cli: cli}, nil
98+}
99+
100+func (e *dockerEngine) Close() error { return e.cli.Close() }
101+
102+func (e *dockerEngine) Info(ctx context.Context) (daemonInfo, error) {
103+ r, err := e.cli.Info(ctx, client.InfoOptions{})
104+ if err != nil {
105+ return daemonInfo{}, fmt.Errorf("docker info: %w", err)
106+ }
107+ di := daemonInfo{OSType: r.Info.OSType}
108+ for _, o := range r.Info.SecurityOptions {
109+ if o == "name=rootless" {
110+ di.Rootless = true
111+ }
112+ }
113+ for name := range r.Info.Runtimes {
114+ di.Runtimes = append(di.Runtimes, name)
115+ }
116+ return di, nil
117+}
118+
119+func (e *dockerEngine) EnsureImage(ctx context.Context, ref string) error {
120+ if _, err := e.cli.ImageInspect(ctx, ref); err == nil {
121+ return nil
122+ } else if !cerrdefs.IsNotFound(err) {
123+ return fmt.Errorf("inspect image %s: %w", ref, err)
124+ }
125+ rc, err := e.cli.ImagePull(ctx, ref, client.ImagePullOptions{})
126+ if err != nil {
127+ return fmt.Errorf("image %s is not present and cannot be pulled (build it with scripts/build-image.sh): %w", ref, err)
128+ }
129+ defer func() { _ = rc.Close() }()
130+ if err := rc.Wait(ctx); err != nil {
131+ return fmt.Errorf("pull image %s: %w", ref, err)
132+ }
133+ return nil
134+}
135+
136+func (e *dockerEngine) CreateNetwork(ctx context.Context, s networkSpec) (string, error) {
137+ f := false
138+ r, err := e.cli.NetworkCreate(ctx, s.Name, client.NetworkCreateOptions{
139+ Driver: "bridge",
140+ EnableIPv6: &f,
141+ IPAM: &network.IPAM{Config: []network.IPAMConfig{{
142+ Subnet: s.Subnet, Gateway: s.Gateway,
143+ }}},
144+ Options: map[string]string{
145+ "com.docker.network.bridge.name": s.Bridge,
146+ // The runner's nftables table does the NAT, with its own policy.
147+ "com.docker.network.bridge.enable_ip_masquerade": "false",
148+ },
149+ Labels: s.Labels,
150+ })
151+ if err != nil {
152+ return "", fmt.Errorf("create network %s: %w", s.Name, err)
153+ }
154+ return r.ID, nil
155+}
156+
157+func (e *dockerEngine) RemoveNetwork(ctx context.Context, id string) error {
158+ _, err := e.cli.NetworkRemove(ctx, id, client.NetworkRemoveOptions{})
159+ return ignoreNotFound(err)
160+}
161+
162+func (e *dockerEngine) ListNetworks(ctx context.Context, key string) ([]string, error) {
163+ r, err := e.cli.NetworkList(ctx, client.NetworkListOptions{Filters: client.Filters{}.Add("label", key)})
164+ if err != nil {
165+ return nil, err
166+ }
167+ var ids []string
168+ for _, n := range r.Items {
169+ ids = append(ids, n.ID)
170+ }
171+ return ids, nil
172+}
173+
174+func (e *dockerEngine) CreateVolume(ctx context.Context, name string, labels map[string]string) error {
175+ _, err := e.cli.VolumeCreate(ctx, client.VolumeCreateOptions{Name: name, Labels: labels})
176+ return err
177+}
178+
179+func (e *dockerEngine) RemoveVolume(ctx context.Context, name string) error {
180+ _, err := e.cli.VolumeRemove(ctx, name, client.VolumeRemoveOptions{Force: true})
181+ return ignoreNotFound(err)
182+}
183+
184+func (e *dockerEngine) ListVolumes(ctx context.Context, key string) ([]string, error) {
185+ r, err := e.cli.VolumeList(ctx, client.VolumeListOptions{Filters: client.Filters{}.Add("label", key)})
186+ if err != nil {
187+ return nil, err
188+ }
189+ var names []string
190+ for _, v := range r.Items {
191+ names = append(names, v.Name)
192+ }
193+ return names, nil
194+}
195+
196+func (e *dockerEngine) CreateContainer(ctx context.Context, s containerSpec) (string, error) {
197+ hc := &container.HostConfig{
198+ NetworkMode: container.NetworkMode(s.Network),
199+ Privileged: s.Privileged,
200+ Runtime: s.Runtime,
201+ CgroupnsMode: container.CgroupnsModePrivate,
202+ DNS: []netip.Addr{s.DNS},
203+ Sysctls: map[string]string{"net.ipv6.conf.all.disable_ipv6": "1", "net.ipv6.conf.default.disable_ipv6": "1"},
204+ }
205+ if s.MemoryMiB > 0 {
206+ hc.Memory = int64(s.MemoryMiB) << 20
207+ }
208+ if s.CPUs > 0 {
209+ hc.NanoCPUs = int64(s.CPUs) * 1_000_000_000
210+ }
211+ for _, b := range s.Binds {
212+ hc.Mounts = append(hc.Mounts, mount.Mount{Type: mount.TypeBind, Source: b.Source, Target: b.Target, ReadOnly: b.ReadOnly,
213+ BindOptions: &mount.BindOptions{CreateMountpoint: true}})
214+ }
215+ for _, v := range s.Volumes {
216+ hc.Mounts = append(hc.Mounts, mount.Mount{Type: mount.TypeVolume, Source: v.Name, Target: v.Target})
217+ }
218+ cfg := &container.Config{
219+ Image: s.Image,
220+ Entrypoint: s.Entrypoint,
221+ Cmd: []string{},
222+ Env: s.Env,
223+ Labels: s.Labels,
224+ Hostname: "ermite-job",
225+ }
226+ nc := &network.NetworkingConfig{EndpointsConfig: map[string]*network.EndpointSettings{
227+ s.Network: {IPAMConfig: &network.EndpointIPAMConfig{IPv4Address: s.IPv4}},
228+ }}
229+ r, err := e.cli.ContainerCreate(ctx, client.ContainerCreateOptions{Name: s.Name, Config: cfg, HostConfig: hc, NetworkingConfig: nc})
230+ if err != nil {
231+ return "", fmt.Errorf("create container: %w", err)
232+ }
233+ return r.ID, nil
234+}
235+
236+func (e *dockerEngine) StartContainer(ctx context.Context, id string) error {
237+ _, err := e.cli.ContainerStart(ctx, id, client.ContainerStartOptions{})
238+ return err
239+}
240+
241+func (e *dockerEngine) WaitExit(ctx context.Context, id string) <-chan exitResult {
242+ out := make(chan exitResult, 1)
243+ w := e.cli.ContainerWait(ctx, id, client.ContainerWaitOptions{Condition: container.WaitConditionNextExit})
244+ go func() {
245+ select {
246+ case r := <-w.Result:
247+ out <- exitResult{Code: int(r.StatusCode)}
248+ case err := <-w.Error:
249+ out <- exitResult{Code: -1, Err: err}
250+ case <-ctx.Done():
251+ out <- exitResult{Code: -1, Err: ctx.Err()}
252+ }
253+ }()
254+ return out
255+}
256+
257+func (e *dockerEngine) RemoveContainer(ctx context.Context, id string) error {
258+ _, err := e.cli.ContainerRemove(ctx, id, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true})
259+ return ignoreNotFound(err)
260+}
261+
262+func (e *dockerEngine) ListContainers(ctx context.Context, key string) ([]string, error) {
263+ r, err := e.cli.ContainerList(ctx, client.ContainerListOptions{All: true, Filters: client.Filters{}.Add("label", key)})
264+ if err != nil {
265+ return nil, err
266+ }
267+ var ids []string
268+ for _, c := range r.Items {
269+ ids = append(ids, c.ID)
270+ }
271+ return ids, nil
272+}
273+
274+func (e *dockerEngine) LogsTail(ctx context.Context, id string, lines int) (string, error) {
275+ rc, err := e.cli.ContainerLogs(ctx, id, client.ContainerLogsOptions{ShowStdout: true, ShowStderr: true, Tail: fmt.Sprint(lines)})
276+ if err != nil {
277+ return "", err
278+ }
279+ defer func() { _ = rc.Close() }()
280+ b, err := io.ReadAll(io.LimitReader(rc, 64<<10))
281+ return string(b), err
282+}
283+
284+func ignoreNotFound(err error) error {
285+ if err == nil || cerrdefs.IsNotFound(err) {
286+ return nil
287+ }
288+ return err
289+}
290+
291+var errRootless = errors.New("the Docker daemon is rootless: a job needs its own nested dockerd, which requires a rootful daemon (or the sysbox runtime); point DOCKER_HOST at a rootful daemon or use ERMITE_RUNNER_ISOLATION=vm")
A internal/sandbox/drivers/drivers.go +24 -0
@@ -0,0 +1,24 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package drivers wires the sandbox drivers into sandbox.Select.
4+package drivers
5+
6+import (
7+ "context"
8+
9+ "bnema.dev/ermite-runner/internal/sandbox"
10+ "bnema.dev/ermite-runner/internal/sandbox/container"
11+ "bnema.dev/ermite-runner/internal/sandbox/vm"
12+)
13+
14+// Factories returns the real vm and container driver factories.
15+func Factories() sandbox.Factories {
16+ return sandbox.Factories{
17+ VM: func(ctx context.Context, cfg sandbox.Config) (sandbox.Driver, error) {
18+ return vm.New(ctx, cfg, vm.Options{})
19+ },
20+ Container: func(ctx context.Context, cfg sandbox.Config) (sandbox.Driver, error) {
21+ return container.New(ctx, cfg, container.Options{})
22+ },
23+ }
24+}
@@ -0,0 +1,24 @@
1+// SPDX-License-Identifier: MIT
2+
3+// Package drivers wires the sandbox drivers into sandbox.Select.
4+package drivers
5+
6+import (
7+ "context"
8+
9+ "bnema.dev/ermite-runner/internal/sandbox"
10+ "bnema.dev/ermite-runner/internal/sandbox/container"
11+ "bnema.dev/ermite-runner/internal/sandbox/vm"
12+)
13+
14+// Factories returns the real vm and container driver factories.
15+func Factories() sandbox.Factories {
16+ return sandbox.Factories{
17+ VM: func(ctx context.Context, cfg sandbox.Config) (sandbox.Driver, error) {
18+ return vm.New(ctx, cfg, vm.Options{})
19+ },
20+ Container: func(ctx context.Context, cfg sandbox.Config) (sandbox.Driver, error) {
21+ return container.New(ctx, cfg, container.Options{})
22+ },
23+ }
24+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 4

Tags 11

↑↓ moveenter openesc close