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