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 disk. The scratch disk is a sparse ext4 file of
7// ERMITE_RUNNER_JOB_DISK, loop-mounted on the host (so it needs root, like the
8// rest of this mode) and bind-mounted at /scratch: the nested dockerd keeps its
9// data root there and the job workspace is a directory of it, so a job cannot
10// write more than its disk size to the host (the container's own writable layer
11// stays small: nothing big is written outside /scratch and /home/runner/work).
12// Everything Docker-side carries the label sandbox.LabelJob so that Reap finds
13// it after a crash.
14//
15// The executor stream is a Unix socket in a per-job directory of the runner's
16// state directory, bind-mounted into the container. The runner executable and
17// the host's external tools (Node.js) are bind-mounted read-only. Both paths are host paths given to the Docker daemon:
18// the daemon must run on the same host (and mount namespace) as the runner.
19package container
20
21import (
22 "context"
23 "errors"
24 "fmt"
25 "net"
26 "net/netip"
27 "os"
28 "path/filepath"
29 "slices"
30 "strings"
31 "sync"
32 "time"
33
34 "github.com/bnema/zerowrap"
35
36 "bnema.dev/ermite-runner/internal/cmdrun"
37 "bnema.dev/ermite-runner/internal/netpolicy"
38 "bnema.dev/ermite-runner/internal/sandbox"
39 "bnema.dev/ermite-runner/internal/sandbox/disk"
40 "bnema.dev/ermite-runner/internal/sandbox/netsetup"
41)
42
43// SysboxRuntime is the name of the sysbox runtime in `docker info`.
44const SysboxRuntime = "sysbox-runc"
45
46// bridgePrefix must match the "erbr*" wildcard of the nftables policy.
47const bridgePrefix = "erbr"
48
49// Options are the testable knobs of the driver.
50type Options struct {
51 Run cmdrun.Runner
52 // Net overrides the network plumbing (tests).
53 Net netsetup.Plumbing
54 // engine overrides the Docker client (tests).
55 engine engine
56 // UnsafeAllowRootlessForTests lets tests exercise the Engine API and the executor against
57 // the rootless daemon of a workstation. Never set outside tests: a nested
58 // dockerd cannot work in a rootless daemon, so the jobs would run without
59 // Docker. The factories of package drivers never set it.
60 UnsafeAllowRootlessForTests bool
61 // ConnectTimeout bounds container start + init + executor connection
62 // (default 90 s).
63 ConnectTimeout time.Duration
64 // Mounter overrides the loop mounts (tests).
65 Mounter *disk.Mounter
66 // FreeSpace overrides statfs for the free-space warning (tests).
67 FreeSpace func(string) (int64, error)
68}
69
70// Driver implements sandbox.Driver for containers.
71type Driver struct {
72 cfg sandbox.Config
73 opt Options
74 eng engine
75 net netsetup.Plumbing
76 mnt disk.Mounter
77 lock *sandbox.HostLock
78 image string
79 runtime string // "" when privileged
80 privileged bool
81 exe string
82
83 mu sync.Mutex
84 slots *sandbox.Slots
85}
86
87// New takes the host-wide lock, connects to the Docker daemon (DOCKER_HOST
88// honoured) and checks it: not rootless, the job image present (pulled if
89// missing), the runtime decided (sysbox when listed; else privileged, unless
90// ERMITE_RUNNER_DENY_PRIVILEGED=1, which refuses). IPv4 forwarding is enabled
91// host-wide and stays on (see the README).
92func New(ctx context.Context, cfg sandbox.Config, opt Options) (*Driver, error) {
93 if opt.Run == nil {
94 opt.Run = cmdrun.Exec{}
95 }
96 if opt.ConnectTimeout == 0 {
97 opt.ConnectTimeout = 90 * time.Second
98 }
99 d := &Driver{cfg: cfg, opt: opt, image: cfg.OCIImage(), slots: sandbox.NewSlots(cfg.Capacity)}
100 if opt.Mounter != nil {
101 d.mnt = *opt.Mounter
102 } else {
103 d.mnt = disk.Mounter{Run: opt.Run}
104 }
105 exe, err := cfg.ExecutablePath()
106 if err != nil {
107 return nil, err
108 }
109 if err := sandbox.CheckStatic(exe); err != nil {
110 return nil, err
111 }
112 d.exe = exe
113
114 if d.lock, err = sandbox.AcquireHostLock(cfg.LockFile); err != nil {
115 return nil, err
116 }
117 eng := opt.engine
118 if eng == nil {
119 e, err := newDockerEngine()
120 if err != nil {
121 _ = d.lock.Release()
122 return nil, err
123 }
124 eng = e
125 }
126 d.eng = eng
127 fail := func(err error) (*Driver, error) {
128 _ = eng.Close()
129 _ = d.lock.Release()
130 return nil, err
131 }
132
133 info, err := eng.Info(ctx)
134 if err != nil {
135 return fail(fmt.Errorf("cannot reach the Docker daemon (DOCKER_HOST): %w", err))
136 }
137 if info.Rootless && !opt.UnsafeAllowRootlessForTests {
138 return fail(errRootless)
139 }
140 if info.OSType != "" && info.OSType != "linux" {
141 return fail(fmt.Errorf("the Docker daemon runs %s containers, need linux", info.OSType))
142 }
143 log := zerowrap.FromCtx(ctx)
144 if slices.Contains(info.Runtimes, SysboxRuntime) {
145 d.runtime = SysboxRuntime
146 log.Info().Str("runtime", SysboxRuntime).Msg("container isolation: using the sysbox runtime")
147 } else {
148 if cfg.DenyPrivileged {
149 return fail(fmt.Errorf("the sysbox runtime is not available and %s=1 forbids privileged containers (install sysbox, or unset the setting to accept host-root-equivalent jobs)", sandbox.EnvDenyPrivileged))
150 }
151 d.privileged = true
152 log.Warn().Msg(privilegedWarning)
153 }
154 if err := disk.CheckMkfs(ctx, opt.Run); err != nil {
155 return fail(err)
156 }
157 if err := os.MkdirAll(cfg.StateDir, 0o755); err != nil {
158 return fail(fmt.Errorf("state directory: %w", err))
159 }
160 sandbox.WarnIfLowSpace(ctx, cfg.StateDir, cfg.Capacity, cfg.JobDiskMiB, opt.FreeSpace)
161 if err := eng.EnsureImage(ctx, d.image); err != nil {
162 return fail(err)
163 }
164 d.net = opt.Net
165 if d.net == nil {
166 n := newNet(opt.Run, cfg.DNSPort, info.Podman, cfg.GuestForge.DNSName())
167 d.net = n
168 }
169 if err := d.net.EnableForwarding(); err != nil {
170 return fail(err)
171 }
172 if cfg.GuestForge != nil {
173 log.Warn().Msg(cfg.GuestForge.Warning())
174 }
175 return d, nil
176}
177
178// newNet builds the host network plumbing. Only podman creates the bridge
179// address late (see daemonInfo.Podman), so only there the DNS listener may come
180// before the address; on Docker a missing gateway address stays a startup error.
181func newNet(run cmdrun.Runner, dnsPort int, podman bool, forgeHost string) *netsetup.Net {
182 n := netsetup.New(run, dnsPort, "")
183 n.DNS.FreeBind = podman
184 n.DNS.Override = forgeHost
185 return n
186}
187
188// maxSockPath is the longest Unix socket path Linux accepts (108 with the NUL).
189const maxSockPath = 107
190
191// checkSockPaths verifies that the sockets of a job's socket directory fit in
192// sockaddr_un.sun_path (108 bytes): the executor socket and, when the cache is
193// enabled, the cache socket, which is one byte longer. It says so instead of the
194// "invalid argument" of bind.
195func checkSockPaths(dir string, cache bool) error {
196 socks := []struct{ what, name string }{{"executor", sandbox.ContainerSocket}}
197 if cache {
198 socks = append(socks, struct{ what, name string }{"cache", sandbox.CacheSocket})
199 }
200 for _, s := range socks {
201 p := filepath.Join(dir, s.name)
202 if len(p) > maxSockPath {
203 return fmt.Errorf("%s socket path %q is %d bytes, over the %d-byte limit of a Unix socket: use a shorter %s", s.what, p, len(p), maxSockPath, sandbox.EnvStateDir)
204 }
205 }
206 return nil
207}
208
209// privilegedWarning is logged when the driver starts and again at every job
210// start: a privileged container is host-root equivalent.
211const privilegedWarning = "privileged container: host-root equivalent, network policy best effort"
212
213// Mode implements sandbox.Driver.
214func (d *Driver) Mode() sandbox.Mode { return sandbox.Container }
215
216// Isolation implements sandbox.Driver.
217func (d *Driver) Isolation() sandbox.Isolation {
218 if d.privileged {
219 return sandbox.IsolationContainerPrivileged
220 }
221 return sandbox.IsolationContainerSysbox
222}
223
224// Privileged reports whether jobs run privileged (no sysbox).
225func (d *Driver) Privileged() bool { return d.privileged }
226
227// Close implements sandbox.Driver.
228func (d *Driver) Close() error {
229 if d.net != nil {
230 d.net.Close()
231 }
232 err := d.eng.Close()
233 return errors.Join(err, d.lock.Release())
234}
235
236// New implements sandbox.Driver.
237func (d *Driver) New() (sandbox.Sandbox, error) {
238 id, err := sandbox.NewID()
239 if err != nil {
240 return nil, err
241 }
242 d.mu.Lock()
243 defer d.mu.Unlock()
244 slot, err := d.slots.Acquire()
245 if err != nil {
246 return nil, err
247 }
248 return &containerSandbox{d: d, slot: slot, id: id}, nil
249}
250
251func (d *Driver) release(slot int) {
252 d.mu.Lock()
253 d.slots.Release(slot)
254 d.mu.Unlock()
255}
256
257func (d *Driver) markLeaked(slot int) {
258 d.mu.Lock()
259 d.slots.MarkLeaked(slot)
260 d.mu.Unlock()
261}
262
263func (d *Driver) runDir(id string) string { return filepath.Join(d.cfg.StateDir, "containers", id) }
264
265// sockDir holds only the executor socket: it is the one directory of the job
266// that is bind-mounted (as /run/ermite), so the job never sees the scratch disk
267// file, nor the other host paths of its own sandbox.
268func (d *Driver) sockDir(id string) string { return filepath.Join(d.runDir(id), "sock") }
269
270// scratchDir is where the loop-mounted scratch disk of a job is mounted.
271func (d *Driver) scratchDir(id string) string { return filepath.Join(d.runDir(id), "scratch") }
272
273// workDir is the job workspace, a directory of the scratch disk.
274func (d *Driver) workDir(id string) string { return filepath.Join(d.scratchDir(id), "work") }
275
276// Where the job workspace and its temporary files appear in the container.
277const (
278 containerWorkspace = "/home/runner/work"
279 containerTmp = "/tmp"
280)
281
282type containerSandbox struct {
283 d *Driver
284 slot int
285 id string
286 gate sandbox.BootGate
287
288 mu sync.Mutex
289 started bool
290 stopped bool
291 leakErr error
292 netID string
293 container string
294 mounted bool
295 listener net.Listener
296 cacheOn bool
297 conn net.Conn
298 iface string
299 gateway netip.Addr
300 attached bool
301}
302
303// ID implements sandbox.Sandbox.
304func (s *containerSandbox) ID() string { return s.id }
305
306// Start implements sandbox.Sandbox.
307func (s *containerSandbox) Start(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
308 // A Stop from another goroutine aborts the boot through the gate.
309 bootCtx, err := s.gate.Begin(ctx)
310 if err != nil {
311 return nil, fmt.Errorf("container sandbox: %w", err)
312 }
313 defer s.gate.End()
314
315 s.mu.Lock()
316 defer s.mu.Unlock()
317 if s.started || s.stopped {
318 return nil, errors.New("container sandbox: Start called twice or after Stop")
319 }
320 s.started = true
321 d := s.d
322 log := zerowrap.FromCtx(ctx)
323 spec = d.cfg.Resolve(spec)
324 if d.privileged {
325 // At every job start, not just at driver start: the job log and the
326 // operator's logs must say what kind of isolation this job got.
327 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(privilegedWarning)
328 }
329 if d.cfg.GuestForge != nil {
330 log.Warn().Str("job", spec.JobID).Str("container", s.id).Msg(d.cfg.GuestForge.Warning())
331 }
332 conn, err := s.run(bootCtx, spec)
333 if err != nil {
334 var tail string
335 if s.container != "" {
336 tail, _ = d.eng.LogsTail(context.WithoutCancel(ctx), s.container, 40)
337 }
338 cerr := s.cleanupLocked(context.WithoutCancel(ctx))
339 if tail != "" {
340 err = fmt.Errorf("%w\ncontainer log tail:\n%s", err, tail)
341 }
342 return nil, errors.Join(err, cerr)
343 }
344 log.Info().Str("container", s.container).Int("slot", s.slot).Str("job", spec.JobID).Bool("privileged", d.privileged).Msg("container running")
345 return conn, nil
346}
347
348// run runs the boot stages in order; each records what it created for cleanup.
349func (s *containerSandbox) run(ctx context.Context, spec sandbox.JobSpec) (sandbox.Conn, error) {
350 netw, err := netpolicy.Subnet(s.d.cfg.Network, s.slot)
351 if err != nil {
352 return nil, err
353 }
354 accepted, err := s.runSocket()
355 if err != nil {
356 return nil, err
357 }
358 if err := s.runCache(spec); err != nil {
359 return nil, err
360 }
361 if err := s.runScratch(ctx, spec); err != nil {
362 return nil, err
363 }
364 labels := map[string]string{sandbox.LabelJob: spec.JobID}
365 if err := s.runNetwork(ctx, netw, labels); err != nil {
366 return nil, err
367 }
368 exited, err := s.runContainer(ctx, spec, netw, labels)
369 if err != nil {
370 return nil, err
371 }
372 return s.waitExecutor(ctx, accepted, exited)
373}
374
375type acceptResult struct {
376 c net.Conn
377 err error
378}
379
380// runSocket creates the per-job directory and listens on the executor socket in
381// it. The directory exists before anything on the Docker side, so a crash
382// leaves a recognisable one.
383func (s *containerSandbox) runSocket() (<-chan acceptResult, error) {
384 d := s.d
385 for _, dir := range []string{d.runDir(s.id), d.sockDir(s.id)} {
386 if err := os.MkdirAll(dir, 0o711); err != nil {
387 return nil, err
388 }
389 if err := os.Chmod(dir, 0o711); err != nil {
390 return nil, err
391 }
392 }
393 dir := d.sockDir(s.id)
394 sock := filepath.Join(dir, sandbox.ContainerSocket)
395 if err := checkSockPaths(dir, d.cfg.Cache != nil); err != nil {
396 return nil, err
397 }
398 ln, err := net.Listen("unix", sock)
399 if err != nil {
400 return nil, fmt.Errorf("listen executor socket: %w", err)
401 }
402 s.listener = ln
403 // The container's root may be a mapped (sysbox) user: it must be able to
404 // connect. The path is private to the job (random id, 0711 directory).
405 if err := os.Chmod(sock, 0o666); err != nil {
406 return nil, err
407 }
408 accepted := make(chan acceptResult, 1)
409 go func() {
410 c, err := ln.Accept()
411 accepted <- acceptResult{c, err}
412 }()
413 return accepted, nil
414}
415
416// runCache serves the cache on a second socket of the job's private
417// directory, next to the executor socket: the guest relay of the container
418// connects to /run/ermite/cache.sock. Only this container has that directory
419// mounted, and the service accepts on it the tokens of this job only.
420func (s *containerSandbox) runCache(spec sandbox.JobSpec) error {
421 d := s.d
422 if d.cfg.Cache == nil {
423 return nil
424 }
425 sock := filepath.Join(d.sockDir(s.id), sandbox.CacheSocket)
426 // The container's root may be a mapped (sysbox) user: it must be able to
427 // connect, as for the executor socket.
428 // No bandwidth limit (0): a container has no per-job network rate limit
429 // (NetMbit is a micro-VM setting), and its cache traffic is a local socket.
430 if err := d.cfg.ServeCache(s.id, sock, spec.JobID, 0, func(p string) error { return os.Chmod(p, 0o666) }); err != nil {
431 return fmt.Errorf("serve the cache socket: %w", err)
432 }
433 s.cacheOn = true
434 return nil
435}
436
437// runScratch creates the sparse ext4 scratch disk (ERMITE_RUNNER_JOB_DISK) and
438// loop-mounts it; the workspace directory lives on it.
439func (s *containerSandbox) runScratch(ctx context.Context, spec sandbox.JobSpec) error {
440 d := s.d
441 file := filepath.Join(d.runDir(s.id), "scratch.ext4")
442 if err := disk.CreateScratch(ctx, d.opt.Run, file, spec.DiskMiB, -1); err != nil {
443 return err
444 }
445 mnt := d.scratchDir(s.id)
446 if err := os.MkdirAll(mnt, 0o755); err != nil {
447 return err
448 }
449 if err := d.mnt.Mount(ctx, file, mnt); err != nil {
450 return err
451 }
452 s.mounted = true
453 // Workspace, temp and the nested dockerd's data root on the scratch disk. The
454 // executor (phase G) sets the job layout and ownership inside the sandbox.
455 for _, sub := range []string{"work", "tmp", "docker"} {
456 if err := os.MkdirAll(filepath.Join(mnt, sub), 0o755); err != nil {
457 return err
458 }
459 }
460 return os.Chmod(filepath.Join(mnt, "tmp"), sandbox.TmpMode)
461}
462
463// runNetwork creates the job's bridge and installs the firewall policy.
464func (s *containerSandbox) runNetwork(ctx context.Context, netw netpolicy.Net, labels map[string]string) error {
465 d := s.d
466 s.iface = fmt.Sprintf("%s%d", bridgePrefix, s.slot)
467 s.gateway = netw.Gateway
468 netID, err := d.eng.CreateNetwork(ctx, networkSpec{Name: "ermite-" + s.id, Bridge: s.iface, Subnet: netw.Prefix, Gateway: netw.Gateway, Labels: labels})
469 if err != nil {
470 return err
471 }
472 s.netID = netID
473 s.attached = true
474 return d.net.Attach(ctx, netpolicy.Guest{Iface: s.iface, Gateway: netw.Gateway, Addr: netw.Guest, ForgePort: d.cfg.GuestForge.TCPPort()})
475}
476
477// runContainer creates and starts the container.
478func (s *containerSandbox) runContainer(ctx context.Context, spec sandbox.JobSpec, netw netpolicy.Net, labels map[string]string) (<-chan exitResult, error) {
479 d := s.d
480 entry := sandbox.GuestExecDir + "/" + sandbox.GuestExecName
481 binds := []bindMount{
482 {Source: d.exe, Target: entry, ReadOnly: true},
483 {Source: d.sockDir(s.id), Target: sandbox.ContainerRunDir},
484 {Source: d.scratchDir(s.id), Target: sandbox.GuestScratch},
485 {Source: d.workDir(s.id), Target: containerWorkspace},
486 {Source: filepath.Join(d.scratchDir(s.id), "tmp"), Target: containerTmp},
487 }
488 binds = append(binds, externalBinds(d.cfg.ExternalsOf())...)
489 cid, err := d.eng.CreateContainer(ctx, containerSpec{
490 Name: "ermite-" + s.id,
491 Image: d.image,
492 Entrypoint: []string{entry, "init"},
493 Env: []string{"ERMITE_CONNECT=unix:" + sandbox.ContainerRunDir + "/" + sandbox.ContainerSocket},
494 Labels: labels,
495 Runtime: d.runtime,
496 Privileged: d.privileged,
497 Network: "ermite-" + s.id,
498 IPv4: netw.Guest,
499 DNS: netw.Gateway,
500 Binds: binds,
501 MemoryMiB: spec.MemoryMiB,
502 CPUs: spec.CPUs,
503 Pids: d.cfg.JobPids,
504 })
505 if err != nil {
506 return nil, err
507 }
508 s.container = cid
509 exited := d.eng.WaitExit(ctx, cid)
510 if err := d.eng.StartContainer(ctx, cid); err != nil {
511 return nil, fmt.Errorf("start container: %w", err)
512 }
513 return exited, nil
514}
515
516// externalBinds are the read-only bind mounts of the host's external tools
517// (Node.js) at /home/runner/externals/<name>, GitHub's layout. A container needs
518// no more than the host path: it reads the tool directly.
519func externalBinds(set []sandbox.External) []bindMount {
520 var out []bindMount
521 for _, e := range set {
522 out = append(out, bindMount{Source: e.Dir, Target: sandbox.GuestExternalsDir + "/" + e.Name, ReadOnly: true})
523 }
524 return out
525}
526
527// waitExecutor waits for the executor to connect. Only ctx (the caller's
528// context, or Stop through the boot gate), the container exiting, and the
529// timeout end the wait.
530func (s *containerSandbox) waitExecutor(ctx context.Context, accepted <-chan acceptResult, exited <-chan exitResult) (sandbox.Conn, error) {
531 timer := time.NewTimer(s.d.opt.ConnectTimeout)
532 defer timer.Stop()
533 select {
534 case r := <-accepted:
535 if r.err != nil {
536 return nil, fmt.Errorf("accept executor connection: %w", r.err)
537 }
538 s.conn = r.c
539 return r.c, nil
540 case ex := <-exited:
541 if ex.Err != nil {
542 return nil, fmt.Errorf("waiting for the container: %w", ex.Err)
543 }
544 return nil, fmt.Errorf("container exited (code %d) before the executor connected", ex.Code)
545 case <-timer.C:
546 return nil, fmt.Errorf("executor did not connect within %s", s.d.opt.ConnectTimeout)
547 case <-ctx.Done():
548 return nil, context.Cause(ctx)
549 }
550}
551
552// Stop implements sandbox.Sandbox.
553func (s *containerSandbox) Stop(ctx context.Context) error {
554 // Abort a boot in progress first: Start holds s.mu for its whole duration.
555 s.gate.Abort()
556 s.mu.Lock()
557 defer s.mu.Unlock()
558 if s.stopped {
559 return s.leakErr
560 }
561 if !s.started {
562 s.releaseLocked()
563 return nil
564 }
565 return s.cleanupLocked(context.WithoutCancel(ctx))
566}
567
568func (s *containerSandbox) releaseLocked() {
569 s.stopped = true
570 s.d.release(s.slot)
571}
572
573// cleanupLocked removes the container first; only when it is confirmed gone are
574// the firewall policy, the network, the scratch mount and the files removed. A
575// container that may still run keeps its policy and bridge: tearing the policy
576// down under it would remove the isolation it relies on. Failures mark the
577// sandbox as leaked (sandbox.ErrLeaked): its slot stays taken until Reap.
578func (s *containerSandbox) cleanupLocked(ctx context.Context) error {
579 d := s.d
580 var errs []error
581 if s.conn != nil {
582 _ = s.conn.Close()
583 s.conn = nil
584 }
585 if s.listener != nil {
586 _ = s.listener.Close()
587 s.listener = nil
588 }
589 if s.cacheOn {
590 d.cfg.StopCache(s.id)
591 s.cacheOn = false
592 }
593 if s.container != "" {
594 if err := d.eng.RemoveContainer(ctx, s.container); err != nil {
595 errs = append(errs, fmt.Errorf("remove container: %w", err))
596 } else {
597 s.container = ""
598 }
599 }
600 if s.container == "" {
601 if s.attached {
602 if err := d.net.Detach(ctx, s.iface, s.gateway); err != nil {
603 errs = append(errs, err)
604 } else {
605 s.attached = false
606 }
607 }
608 if s.netID != "" {
609 if err := d.eng.RemoveNetwork(ctx, s.netID); err != nil {
610 errs = append(errs, fmt.Errorf("remove network: %w", err))
611 } else {
612 s.netID = ""
613 }
614 }
615 if s.mounted {
616 if err := d.mnt.Unmount(ctx, d.scratchDir(s.id)); err != nil {
617 errs = append(errs, err)
618 } else {
619 s.mounted = false
620 }
621 }
622 // Never delete files below a mount that is still there.
623 if !s.mounted {
624 if err := os.RemoveAll(d.runDir(s.id)); err != nil {
625 errs = append(errs, err)
626 }
627 }
628 }
629 if len(errs) == 0 {
630 s.releaseLocked()
631 return nil
632 }
633 s.stopped = true
634 d.markLeaked(s.slot)
635 s.leakErr = fmt.Errorf("%w: container sandbox %s: %w", sandbox.ErrLeaked, s.id, errors.Join(errs...))
636 log := zerowrap.FromCtx(ctx)
637 log.Error().Err(s.leakErr).Str("job", s.id).Msg("container cleanup incomplete; call Reap to finish it")
638 return s.leakErr
639}
640
641// Reap implements sandbox.Driver: remove every container and network carrying
642// the job label, the nftables table, the scratch mounts and the per-job
643// directories, and free the slots of leaked sandboxes. It returns ErrBusy while
644// a sandbox is running.
645func (d *Driver) Reap(ctx context.Context) error {
646 log := zerowrap.FromCtx(ctx)
647 d.mu.Lock()
648 busy := d.slots.Busy()
649 d.mu.Unlock()
650 if busy > 0 {
651 return fmt.Errorf("%w (%d running)", sandbox.ErrBusy, busy)
652 }
653 var errs []error
654 ids, err := d.eng.ListContainers(ctx, sandbox.LabelJob)
655 if err != nil {
656 errs = append(errs, fmt.Errorf("list containers: %w", err))
657 }
658 for _, id := range ids {
659 log.Warn().Str("container", id).Msg("removing leftover job container")
660 if err := d.eng.RemoveContainer(ctx, id); err != nil {
661 errs = append(errs, err)
662 }
663 }
664 nets, err := d.eng.ListNetworks(ctx, sandbox.LabelJob)
665 if err != nil {
666 errs = append(errs, fmt.Errorf("list networks: %w", err))
667 }
668 for _, n := range nets {
669 log.Warn().Str("network", n).Msg("removing leftover job network")
670 if err := d.eng.RemoveNetwork(ctx, n); err != nil {
671 errs = append(errs, err)
672 }
673 }
674 if err := d.net.Reset(ctx); err != nil {
675 errs = append(errs, fmt.Errorf("reset nftables table: %w", err))
676 }
677 // Unmount before deleting: a directory with a live mount below it must not
678 // be removed (and EBUSY would be the best case).
679 root := filepath.Join(d.cfg.StateDir, "containers")
680 mounts, merr := d.mnt.MountPoints()
681 if merr != nil {
682 errs = append(errs, fmt.Errorf("read mounts: %w", merr))
683 }
684 stuck := map[string]bool{}
685 for _, mp := range mounts {
686 rel, ok := strings.CutPrefix(mp, root+"/")
687 if !ok {
688 continue
689 }
690 id, _, _ := strings.Cut(rel, "/")
691 if !sandbox.ValidID(id) {
692 continue
693 }
694 log.Warn().Str("mount", mp).Msg("unmounting leftover scratch disk")
695 if err := d.mnt.Unmount(ctx, mp); err != nil {
696 stuck[id] = true
697 errs = append(errs, err)
698 }
699 }
700 ents, _ := os.ReadDir(root)
701 for _, e := range ents {
702 if sandbox.ValidID(e.Name()) && !stuck[e.Name()] {
703 if err := os.RemoveAll(filepath.Join(root, e.Name())); err != nil {
704 errs = append(errs, err)
705 }
706 }
707 }
708 if len(errs) == 0 {
709 d.mu.Lock()
710 d.slots.ReleaseLeaked()
711 d.mu.Unlock()
712 }
713 return errors.Join(errs...)
714}