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