diff --git a/.agents/skills/trust-boundary/SKILL.md b/.agents/skills/trust-boundary/SKILL.md index 9cac854..af382ab 100644 --- a/.agents/skills/trust-boundary/SKILL.md +++ b/.agents/skills/trust-boundary/SKILL.md @@ -4,8 +4,8 @@ description: > Checklist for code that decides which devices a container may open, or what the daemon believes about its own configuration. Apply before editing internal/config/**, internal/policy/**, the label handling and rule collection in internal/processor/**, - the SIGHUP reload path in cmd/swarm-device-access/config.go, deployments/** (including - the Dockerfile) or examples/**. Read it before writing the change, not after review + the SIGHUP reload path in cmd/swarm-device-access/config.go, internal/launcher/**, + deployments/** (including the Dockerfile) or examples/**. Read it before writing the change, not after review finds the hole. --- @@ -161,18 +161,28 @@ file are logged and ignored. Covered by `TestReload_*` and `TestMergeSettings_*` ## 6. Least privilege in the image and the deployment The daemon has to be privileged: it attaches BPF programs to other containers' cgroups and stats -host device nodes. So the rule is not "unprivileged" but "no more than the documented set". The -README's Docker Compose for Swarm snippet and `deployments/docker/docker-compose.yaml` run it with -`--privileged`, `--cgroupns=host`, `--pid=host`, `--userns=host` and exactly three bind mounts: -`/sys:/host/sys`, `/var/run/docker.sock` and `/dev:/dev`. The DBus socket -(`/run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket`) stays **optional and commented -out** — the daemon degrades to a startup warning without it. `deployments/docker/Dockerfile` builds -a static binary onto `dhi.io/static` with `USER 0` and nothing else in the runtime stage. - -- [ ] No new capability, namespace flag or host bind mount beyond the documented set, in the - Dockerfile, `deployments/**`, `examples/**` or the README — and if one is truly needed, the +host device nodes. So the rule is not "unprivileged" but "no more than the documented set". + +The single source of truth for that set is `daemonSpec` (`internal/launcher/spec.go`): the Swarm +service in the README's Docker Compose for Swarm snippet, `deployments/docker/docker-compose.yaml` +and `examples/**` runs the image as an unprivileged `launch` task with only the Docker socket, and +the launcher creates the daemon with `Privileged`, host cgroup, PID and user namespaces, network +`none`, `AutoRemove`, and exactly three bind `Mounts`: the Docker socket at `/var/run/docker.sock`, +`/sys:/host/sys` and `/dev:/dev`. `Mounts`, never `Binds`: a missing source must fail the create, +not make Docker create an empty directory on the host. The optional extras are opt-in launch flags: +`-dbus` (`/run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket` — the daemon degrades to a +startup warning without it), `-config-dir` (read-only at `/etc/swarm-device-access`) and +`-host-network`. `TestDaemonSpec` pins the whole host config, so a new mount or flag shows up as a +test change. The launcher's own inputs are checked too: `-config-dir` and `-host-docker-socket` +must be absolute, clean paths, and it removes a `swarm-device-access` container only when its +owning launcher is confirmed gone — never on a transient error, and never a foreign container. +`deployments/docker/Dockerfile` builds a static binary onto `dhi.io/static` with `USER 0` and +nothing else in the runtime stage. + +- [ ] No new capability, namespace flag or host bind mount beyond the documented set, in + `daemonSpec`, the Dockerfile, `deployments/**`, `examples/**` or the README — and if one is truly needed, the README and `AGENTS.md` runtime requirements change in the same commit. -- [ ] The DBus mount is never made mandatory. +- [ ] The DBus mount is never made mandatory (`-dbus` stays off by default). - [ ] A Dockerfile change is checked with `make docker-build`. - [ ] Nothing secret is baked into a layer: build args and `COPY` sources are not a secret store. diff --git a/.mk/integration-test.mk b/.mk/integration-test.mk index 828e8d1..de64525 100644 --- a/.mk/integration-test.mk +++ b/.mk/integration-test.mk @@ -23,12 +23,17 @@ # make go-test-integration SDA_IT_ENFORCE=1 SDA_IT_REQUIRE_RELOAD=1 # # Every container the suite creates carries the swarm-device-access-it.envid -# label. After a killed run, remove the leftovers of every run with: +# label. The daemon a test launcher creates does not; it is swept once its +# launcher no longer exists. After a killed run, remove the leftovers of every +# run with: # make sweep-test-leaks INTEGRATION_TIMEOUT ?= 6m INTEGRATION_PKG ?= ./test/integration/... INTEGRATION_ENV_LABEL := swarm-device-access-it.envid +# The labels internal/launcher puts on the daemon containers it creates. +LAUNCHER_ROLE_LABEL := io.github.leinardi.swarm-device-access.role +LAUNCHER_OWNER_LABEL := io.github.leinardi.swarm-device-access.launcher # go test changes CWD to the package dir, so the binary path must be absolute. export SDA_TEST_BINARY ?= $(REPO_ROOT)/$(DIST_DIR)/$(BIN_NAME) @@ -60,4 +65,16 @@ integration-image: ## Build the daemon image the enforcement test runs, as $(INT .PHONY: sweep-test-leaks sweep-test-leaks: ## Remove every container left behind by integration test runs - docker ps -aq --filter "label=$(INTEGRATION_ENV_LABEL)" | xargs -r docker rm -f + @set -eu; \ + ids=$$(docker ps -aq --filter "label=$(INTEGRATION_ENV_LABEL)"); \ + if [ -n "$$ids" ]; then docker rm -f $$ids; fi; \ + daemons=$$(docker ps -a --filter "label=$(LAUNCHER_ROLE_LABEL)=daemon" \ + --format '{{.ID}} {{.Label "$(LAUNCHER_OWNER_LABEL)"}}'); \ + printf '%s\n' "$$daemons" | while read -r id owner; do \ + [ -n "$$id" ] && [ -n "$$owner" ] || continue; \ + if err=$$(docker container inspect "$$owner" 2>&1 >/dev/null); then continue; fi; \ + case "$$err" in \ + *"No such"*) docker rm -f "$$id" ;; \ + *) echo "inspect launcher $$owner of daemon $$id: $$err" >&2; exit 1 ;; \ + esac; \ + done diff --git a/.serena/memories/core.md b/.serena/memories/core.md index c354c0c..72b61fa 100644 --- a/.serena/memories/core.md +++ b/.serena/memories/core.md @@ -11,11 +11,13 @@ cmd/swarm-device-access/ # main binary (//go:build linux) main.go # run(): flags, settings, openat2 probe, store, processor, daemon.Run flags.go # CLI flag definitions config.go # settings merge/validate, SIGHUP reloader + launch.go # `launch` subcommand: launcher flags, runLaunch version.go # version/commit/date (filled by ldflags) internal/ config/ # strict YAML loader; runtime Store/Publisher with generations policy/ # mode, label parsing, globs: Enabled, Denied, Authorized (cross-platform) + launcher/ # launch mode: daemonSpec (privileged set), self ID, stale cleanup, supervise daemon/ # event loop and coordinator daemon.go # Run: subscribe, coordinator, startup pass, reload watcher, listenEvents events.go # listenEvents/consumeEvents (reconnect backoff), processOne @@ -38,7 +40,7 @@ internal/ systemd/ # DBus Reloading signal watcher test/integration/ # daemon integration tests (build tag integration) -deployments/docker/ # Dockerfile, compose file with the sh wrapper, example config +deployments/docker/ # Dockerfile, compose file running the launcher, example config docs/ # architecture.md, testing.md, release.md ``` diff --git a/AGENTS.md b/AGENTS.md index ce62a8c..2576d9e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -80,6 +80,14 @@ and aggregates per device), then apply it (`applyPinned`: pidfd pin, `/proc/ /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. - stop_grace_period: 30s + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host directory read-only at /etc/swarm-device-access. Mount the + # directory, not the file: an editor that replaces the file would leave a + # file mount on the old content, and a reload would silently apply it. + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + - -log-level=info + - -log-format=text volumes: - /var/run/docker.sock:/var/run/docker.sock deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first + # The launcher's shutdown takes at most 20s; see below. + stop_grace_period: 30s # Example consumer service: a Swarm task that bind-mounts a GPU device. cuda-worker: @@ -237,25 +214,51 @@ services: The daemon compose file is also available at [`deployments/docker/docker-compose.yaml`](deployments/docker/docker-compose.yaml). -The wrapper is a small `sh` script rather than a bare `docker run`, so that -there is exactly one daemon container per node, always named -`swarm-device-access`: - -- It builds the `docker run` arguments with one `set -- "$$@" ...` line each: - options before the image, daemon flags after it. To enable an optional mount - or flag, uncomment its line; it cannot end up on the wrong side of the image. -- It sets a `SIGTERM`/`SIGINT` trap, then removes any `swarm-device-access` - container a killed wrapper left behind, so the replacement task does not - start a second daemon next to an orphan or, normally, hit a name conflict (if - it does, Swarm restarts the task and the next attempt clears it). -- It starts `docker run` in the background and waits on it, because a shell - blocked on a foreground child runs its traps only after the child exits. -- On `SIGTERM` from Swarm the trap runs `docker stop -t 10 - swarm-device-access`, waits for the daemon to exit, and exits with its - status. `stop_grace_period: 30s` gives it time to do so. - -Compose interpolates `$` in the file, so the script writes `$$` for a literal -`$`. +`swarm-device-access launch [launch flags] -- [daemon flags]` finds its own +container, reads the ID of the image it runs, and creates one container named +`swarm-device-access` from that image ID, with the arguments after `--` as the +daemon's flags. The daemon container gets exactly the documented privileged +set: `--privileged`, the host cgroup, PID and user namespaces, no network, and +bind mounts of the host Docker socket at `/var/run/docker.sock`, `/sys` at +`/host/sys` and `/dev` at `/dev`, plus the optional mounts below. The launcher +streams the daemon's output as its own, so `docker service logs` shows the +daemon, and exits with the daemon's status. + +| Launch flag | Default | Description | +| --- | --- | --- | +| `-dbus` | `false` | Bind the host's `/run/dbus/system_bus_socket` at `/var/run/dbus/system_bus_socket` in the daemon, so it reapplies rules after `systemctl daemon-reload`. | +| `-config-dir` | `""` | Host directory bound read-only at `/etc/swarm-device-access`, for the daemon's `-config`. Must be an absolute, clean path. | +| `-host-network` | `false` | Run the daemon in the host network instead of none; only `-metrics-addr` and `-debug-addr` need it. | +| `-host-docker-socket` | `/var/run/docker.sock` | Host path of the Docker socket bound into the daemon. Must be an absolute, clean path. The launcher itself always uses its own `/var/run/docker.sock` mount. | +| `-log-level` | `info` | The launcher's own log level: `debug`, `info`, `warn`, `error`. | +| `-log-format` | `text` | The launcher's own log format: `text`, `json`, `plain`. | +| `-help` | | Print the launch flags and exit | + +Running the daemon from the launcher's own image ID, not a tag, means the +service image is the daemon image: + +- An image watcher such as gantry, or `docker service update --image`, updates + the service, and the new launcher starts the new daemon. Every node runs the + same digest. +- A Swarm rollback (`docker service rollback`) rolls the daemon back with it. +- `update_config.order: stop-first` stops the old launcher, which stops and + removes its daemon, before the new one starts. The launcher's shutdown (a 10s + stop, the wait for the removal and the drain of the last log lines) shares a + single 20s budget, inside `stop_grace_period: 30s`. + +Each daemon container is labelled with the ID of the launcher that owns it. +Before it creates a daemon, a launcher removes a `swarm-device-access` +container only when it is a daemon whose launcher is confirmed gone or +stopped, or a daemon left by the `sh` wrapper of earlier releases. A daemon +whose launcher still runs, a launcher whose state cannot be read, or any other +container with that name makes the launcher exit with an error, and Swarm +retries it. + +Releases before 1.0.0 documented a `docker:29` service that ran an `sh` wrapper +around `docker run`. It still works, but it is no longer documented: the +service image never changed, so image watchers and rollbacks did not follow +daemon releases. To migrate, replace the service with the one above; the first +launcher removes the daemon container the wrapper left running. ## ⚙️ Configuration diff --git a/SECURITY.md b/SECURITY.md index 21d62fe..a2ace2a 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -62,7 +62,7 @@ Expected response time: acknowledgment within 7 days, patch or mitigation plan w ## Known design constraints - The daemon must run as root. Dropping to a minimal capability set (`CAP_BPF`, `CAP_PERFMON`, `CAP_SYS_ADMIN`, `CAP_SYS_RESOURCE`) is theoretically possible but has not been tested across kernel versions and is not supported at this time. -- The daemon uses `privileged: true` in the reference Swarm deployment, started by a wrapper service with `docker run`. Swarm services do support `cap_add` (Docker Engine 20.10 and newer), but not the host cgroup, PID and user namespaces the daemon also needs, which is why the wrapper exists. With the wrapper, or outside Swarm, explicit capabilities could be used instead of `privileged`; that is not tested or supported (see above). +- The daemon uses `privileged: true` in the reference Swarm deployment, where an unprivileged launcher service (`swarm-device-access launch`) creates it through the Docker socket from the launcher's own image. Swarm services do support `cap_add` (Docker Engine 20.10 and newer), but not the host cgroup, PID and user namespaces the daemon also needs, which is why the launcher exists. The launcher itself holds the Docker socket, so it is host-root equivalent too. The privileged set the daemon gets is fixed in `daemonSpec` (`internal/launcher/spec.go`). Outside Swarm, explicit capabilities could be used instead of `privileged`; that is not tested or supported (see above). - **cgroup v1: grants made by a previous daemon instance are not revoked.** On cgroup v1 the daemon can only tell its own `devices.allow` exceptions from the runtime's through an in-memory ledger. A restarted daemon starts with an empty ledger, so exceptions its predecessor added look like the diff --git a/cmd/swarm-device-access/launch.go b/cmd/swarm-device-access/launch.go new file mode 100644 index 0000000..026b28a --- /dev/null +++ b/cmd/swarm-device-access/launch.go @@ -0,0 +1,205 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "os" + "os/signal" + "path/filepath" + "slices" + "syscall" + + "github.com/moby/moby/client" + + "github.com/leinardi/swarm-device-access/internal/config" + "github.com/leinardi/swarm-device-access/internal/launcher" + "github.com/leinardi/swarm-device-access/internal/logger" +) + +const ( + // launchCommand is the first argument that selects launcher mode. + launchCommand = "launch" + // launcherComponent tags the launcher's own log records. + launcherComponent = "launcher" +) + +var ( + errPathNotAbsolute = errors.New("must be an absolute, clean path") + errUnexpectedArg = errors.New("unexpected argument; daemon flags go after --") +) + +// launchSettings is everything the launch flags control. +type launchSettings struct { + LogFormat string + LogLevel string + Opts launcher.Options +} + +// parseLaunchArgs parses the launcher's flags. Everything after the first +// "--" is passed to the daemon verbatim. flag.ErrHelp is returned for -help. +func parseLaunchArgs(args []string, output io.Writer) (launchSettings, error) { + daemonArgs := []string{} + + sep := slices.Index(args, "--") + if sep >= 0 { + daemonArgs = slices.Clone(args[sep+1:]) + args = args[:sep] + } + + flags := flag.NewFlagSet(launchCommand, flag.ContinueOnError) + flags.SetOutput(output) + flags.Usage = func() { + fmt.Fprintln(output, "Usage: swarm-device-access launch [flags] -- [daemon flags]") + flags.PrintDefaults() + } + + var parsed launchSettings + + flags.StringVar( + &parsed.LogFormat, + "log-format", + "text", + "Launcher log format: json, text or plain", + ) + flags.StringVar( + &parsed.LogLevel, + "log-level", + "info", + "Launcher log level: debug, info, warn or error", + ) + flags.StringVar( + &parsed.Opts.HostDockerSocket, + "host-docker-socket", + launcher.DefaultHostDockerSocket, + "Host path of the Docker socket bound into the daemon", + ) + flags.BoolVar(&parsed.Opts.DBus, "dbus", false, + "Bind the host's systemd DBus socket into the daemon, enabling daemon-reload handling") + flags.StringVar( + &parsed.Opts.ConfigDir, + "config-dir", + "", + "Host directory bound read-only at "+launcher.ContainerConfigDir+" in the daemon. Empty = none.", + ) + flags.BoolVar( + &parsed.Opts.HostNetwork, + "host-network", + false, + "Run the daemon in the host network (needed for -metrics-addr/-debug-addr); default is no network", + ) + + err := flags.Parse(args) + if err != nil { + return launchSettings{}, fmt.Errorf("launch: %w", err) + } + + if flags.NArg() > 0 { + return launchSettings{}, fmt.Errorf("launch: %q: %w", flags.Arg(0), errUnexpectedArg) + } + + parsed.Opts.DaemonArgs = daemonArgs + + err = parsed.validate() + if err != nil { + return launchSettings{}, err + } + + return parsed, nil +} + +func (s *launchSettings) validate() error { + err := config.ValidateEnums(s.LogFormat, s.LogLevel, "", "flag") + if err != nil { + return fmt.Errorf("launch: %w", err) + } + + err = checkHostPath("host-docker-socket", s.Opts.HostDockerSocket) + if err != nil { + return err + } + + if s.Opts.ConfigDir != "" { + return checkHostPath("config-dir", s.Opts.ConfigDir) + } + + return nil +} + +// checkHostPath rejects a host path Docker would resolve differently from +// what the operator reads: a relative one, or one with . or .. elements. +func checkHostPath(name, path string) error { + if !filepath.IsAbs(path) || filepath.Clean(path) != path { + return fmt.Errorf("launch: flag %q: %q %w", "-"+name, path, errPathNotAbsolute) + } + + return nil +} + +// runLaunch runs launcher mode and returns the daemon's exit status, or 1 +// when the launcher itself fails. It skips everything only the daemon +// needs: the config file, the openat2 probe, RLIMIT_MEMLOCK, SIGHUP and the +// observability servers. +func runLaunch(args []string) int { + parsed, err := parseLaunchArgs(args, os.Stderr) + if errors.Is(err, flag.ErrHelp) { + return 0 + } + + if err != nil { + fmt.Fprintln(os.Stderr, err) + + return 1 + } + + logger.Configure(parsed.LogFormat, parsed.LogLevel, false) + + logger.L().Info("swarm-device-access launcher starting", + "component", launcherComponent, + "version", version, + "commit", commit, + "date", date, + "daemon_args", parsed.Opts.DaemonArgs, + ) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + // API version negotiation is the client default; it runs lazily on the first request. + cli, err := client.New(client.WithHost(launcher.DockerHost)) + if err != nil { + logger.L().Error("docker client init failed", "component", launcherComponent, "err", err) + + return 1 + } + defer cli.Close() + + status, err := launcher.Run(ctx, cli, &parsed.Opts) + if err != nil { + logger.L().Error("launcher failed", "component", launcherComponent, "err", err) + + return 1 + } + + return status +} diff --git a/cmd/swarm-device-access/launch_test.go b/cmd/swarm-device-access/launch_test.go new file mode 100644 index 0000000..2aa35c0 --- /dev/null +++ b/cmd/swarm-device-access/launch_test.go @@ -0,0 +1,177 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package main + +import ( + "bytes" + "errors" + "flag" + "io" + "os" + "slices" + "strings" + "testing" + + "github.com/leinardi/swarm-device-access/internal/launcher" +) + +func TestParseLaunchArgs_SplitsAtDoubleDash(t *testing.T) { + t.Parallel() + + parsed, err := parseLaunchArgs([]string{ + "-dbus", "-host-network", "-config-dir", "/srv/sda", "-log-format", "json", + "--", "-dry-run", "-device-allow", "/dev/sd*", "--", "-x", + }, io.Discard) + if err != nil { + t.Fatal(err) + } + + want := launcher.Options{ + HostDockerSocket: "/var/run/docker.sock", + DBus: true, + ConfigDir: "/srv/sda", + HostNetwork: true, + DaemonArgs: []string{"-dry-run", "-device-allow", "/dev/sd*", "--", "-x"}, + } + + got := parsed.Opts + if got.HostDockerSocket != want.HostDockerSocket || got.DBus != want.DBus || + got.ConfigDir != want.ConfigDir || got.HostNetwork != want.HostNetwork || + !slices.Equal(got.DaemonArgs, want.DaemonArgs) { + t.Errorf("options = %+v, want %+v", got, want) + } + + if parsed.LogFormat != "json" || parsed.LogLevel != "info" { + t.Errorf("log settings = %q, %q", parsed.LogFormat, parsed.LogLevel) + } +} + +func TestParseLaunchArgs_NoDaemonArgs(t *testing.T) { + t.Parallel() + + parsed, err := parseLaunchArgs(nil, io.Discard) + if err != nil { + t.Fatal(err) + } + + if len(parsed.Opts.DaemonArgs) != 0 || parsed.Opts.DBus || parsed.Opts.ConfigDir != "" { + t.Errorf("defaults = %+v", parsed.Opts) + } +} + +func TestParseLaunchArgs_Rejects(t *testing.T) { + t.Parallel() + + cases := map[string]struct { + args []string + wantErr error + wantMsg string + }{ + "relative config dir": { + []string{"-config-dir", "etc/sda"}, + errPathNotAbsolute, + `"-config-dir"`, + }, + "unclean config dir": { + []string{"-config-dir", "/srv/../etc"}, + errPathNotAbsolute, + `"-config-dir"`, + }, + "trailing slash": { + []string{"-config-dir", "/srv/sda/"}, + errPathNotAbsolute, + `"-config-dir"`, + }, + "relative socket": { + []string{"-host-docker-socket", "docker.sock"}, + errPathNotAbsolute, + `"-host-docker-socket"`, + }, + "empty socket": { + []string{"-host-docker-socket", ""}, + errPathNotAbsolute, + `"-host-docker-socket"`, + }, + "daemon flag before --": { + []string{"-dbus", "-dry-run"}, + nil, + "flag provided but not defined", + }, + "stray argument": {[]string{"extra", "--", "-dry-run"}, errUnexpectedArg, `"extra"`}, + "unknown log format": {[]string{"-log-format", "xml"}, nil, `flag "log-format"`}, + "unknown log level": {[]string{"-log-level", "trace"}, nil, `flag "log-level"`}, + } + + for name, tc := range cases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + _, err := parseLaunchArgs(tc.args, io.Discard) + if err == nil { + t.Fatal("accepted") + } + + if tc.wantErr != nil && !errors.Is(err, tc.wantErr) { + t.Errorf("err = %v, want %v", err, tc.wantErr) + } + + if !strings.Contains(err.Error(), tc.wantMsg) { + t.Errorf("err = %v, want it to mention %s", err, tc.wantMsg) + } + }) + } +} + +func TestParseLaunchArgs_Help(t *testing.T) { + t.Parallel() + + var out bytes.Buffer + + _, err := parseLaunchArgs([]string{"-help"}, &out) + if !errors.Is(err, flag.ErrHelp) { + t.Fatalf("err = %v, want flag.ErrHelp", err) + } + + for _, want := range []string{"launch [flags] -- [daemon flags]", "-host-docker-socket", "-config-dir"} { + if !strings.Contains(out.String(), want) { + t.Errorf("help does not mention %q:\n%s", want, out.String()) + } + } +} + +// run dispatches to launcher mode on a leading "launch", before the daemon's +// flags are parsed: -help and a bad -config-dir are the launcher's answers. +// Not parallel: it sets os.Args. +func TestRun_DispatchesLaunch(t *testing.T) { + saved := os.Args + + t.Cleanup(func() { os.Args = saved }) + + os.Args = []string{"swarm-device-access", "launch", "-help"} + + if status := run(); status != 0 { + t.Errorf("launch -help: status %d, want 0", status) + } + + os.Args = []string{"swarm-device-access", "launch", "-config-dir", "relative"} + + if status := run(); status != 1 { + t.Errorf("launch -config-dir relative: status %d, want 1", status) + } +} diff --git a/cmd/swarm-device-access/main.go b/cmd/swarm-device-access/main.go index a9c47b5..64e0d07 100644 --- a/cmd/swarm-device-access/main.go +++ b/cmd/swarm-device-access/main.go @@ -104,9 +104,16 @@ func startServers(ctx context.Context, effective *settings) (stop func(), err er } func run() int { + if len(os.Args) > 1 && os.Args[1] == launchCommand { + return runLaunch(os.Args[2:]) + } + flag.Parse() if *help { + fmt.Fprintln(flag.CommandLine.Output(), + "Usage: swarm-device-access [flags]\n"+ + " swarm-device-access launch [launch flags] -- [daemon flags] (see launch -help)") flag.PrintDefaults() return 0 diff --git a/deployments/docker/docker-compose.yaml b/deployments/docker/docker-compose.yaml index d45d0cf..6ba900e 100644 --- a/deployments/docker/docker-compose.yaml +++ b/deployments/docker/docker-compose.yaml @@ -1,56 +1,35 @@ services: swarm-device-access: - image: docker:29 # dclint disable-line service-image-require-explicit-tag + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - # The wrapper only needs the host Docker socket to dispatch `docker run`. + # The launcher only needs the host Docker socket: it creates the daemon + # container through the Docker API. - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the documented privileged set: --privileged, host cgroup, pid and user + # namespaces, no network, and binds for the Docker socket, /sys and /dev. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - set -- "$$@" -log-level info -log-format text - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - - entrypoint: [ "sh", "-c" ] + - launch + # Launcher flags. Bind the host's systemd DBus socket, so the daemon + # reapplies device rules after systemctl daemon-reload, which wipes cgroup + # BPF programs (without it the daemon warns and skips reload handling): + # - -dbus + # Bind a host directory read-only at /etc/swarm-device-access, for -config + # below. Mount the directory, not the file: an editor that replaces the + # file would leave a file mount on the old content, and a reload would + # silently apply it. + # - -config-dir=/etc/swarm-device-access + # Run the daemon in the host network, needed for -metrics-addr/-debug-addr: + # - -host-network + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + - -log-level=info + - -log-format=text restart: unless-stopped @@ -58,6 +37,11 @@ services: mode: global restart_policy: condition: any + # One daemon per node: the new launcher must not start while the old one + # still owns the node's daemon. + update_config: + order: stop-first - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + # The launcher's shutdown (stop the daemon with a 10s grace period, wait + # for its removal, drain its last log lines) takes at most 20s. stop_grace_period: 30s diff --git a/docs/architecture.md b/docs/architecture.md index 3f778e8..049ce41 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -135,6 +135,41 @@ daemon restarting, for one), is retried with a backoff of 1 s doubling to 5 min, first failure is logged at `Warn`, later ones at `Debug`. Every successful subscription requests a pass, because a reload that completed while the watcher was not subscribed, including one during the startup pass, sent a signal nobody received. +## Launcher + +Swarm rejects `privileged`, `cgroup: host`, `pid: host` and `userns_mode: host` on services, so the +Swarm service runs the image as `swarm-device-access launch [launch flags] -- [daemon flags]`, +unprivileged and with only the Docker socket (`internal/launcher`): + +1. **Self-identification.** The launcher finds its container ID in `/proc/self/mountinfo`: the root of + the `/etc/hostname`, `/etc/hosts` and `/etc/resolv.conf` mounts is + `…/containers/<64 hex>/` whatever the data-root. It inspects that container and takes its + `Image`, the `sha256:` image ID. No match, two IDs, a failed inspect or a non-ID image all exit + non-zero. +2. **Stale-daemon cleanup.** Every container named `swarm-device-access` is checked. A daemon (label + `io.github.leinardi.swarm-device-access.role=daemon`) is removed only when its launcher (label + `io.github.leinardi.swarm-device-access.launcher`) is confirmed gone (`NotFound`) or not running, + or is this very container restarted; one left by the old `sh` wrapper (image title + `swarm-device-access`, no role label) is removed too. A live owner, any other inspect error or a + foreign container aborts the launcher. +3. **Create, attach, wait, start**, in the order the `docker run` CLI uses. `daemonSpec` is the one + source of the daemon's privileged set: the image ID, `Privileged`, host cgroup, PID and user + namespaces, network `none` (or `host` with `-host-network`), `AutoRemove`, a 10s stop timeout, + and bind `Mounts` (never `Binds`, so a missing source fails the create) for the Docker socket, + `/sys` at `/host/sys`, `/dev`, and optionally the DBus socket and a read-only config directory. + The attach copy goroutine starts before `ContainerStart`, and `ContainerWait(removed)` is + registered before it, so no output and no exit is missed. Any failure after the create removes + the created container. +4. **Supervision.** The attach and wait streams each run on their own context, detached from the + signal context; only their establishment is bounded (`DockerCallTimeout`), and a call that ignores + its canceled context (the hijacked attach reads the upgrade response on a raw connection) is + abandoned at the deadline, its late result released in the background. The launcher returns the + daemon's status after it is removed and its output drained. `SIGTERM`, a broken wait stream or a + lost log stream stop the daemon **by ID**, and every shutdown shares one 20s budget (a 10s stop + plus `DockerCallTimeout`) for the stop, the wait for the removal and the drain, inside the compose + files' `stop_grace_period: 30s`. With the wait stream broken, the removal is confirmed by polling + inspect until the daemon is gone. + ## BPF program structure For cgroup v2 hosts, device control is implemented via `BPF_CGROUP_DEVICE` programs. The code in `internal/cgroup/ebpf.go` (ported from @@ -224,7 +259,8 @@ symlinks to directories and non-device entries, and `errors` counts per-device f | Package | Responsibility | | --- | --- | -| `cmd/swarm-device-access` | Entrypoint: flags, settings merge and validation, config file reload on `SIGHUP` (`config.go`), wiring | +| `cmd/swarm-device-access` | Entrypoint: flags, settings merge and validation, config file reload on `SIGHUP` (`config.go`), wiring; `launch` subcommand (`launch.go`) | +| `internal/launcher` | Launcher mode: self-identification, the daemon container spec (`daemonSpec`, the privileged set), stale-daemon cleanup, attach/wait/start and supervision | | `internal/config` | Strict YAML loader, runtime config `Store` with generations and its `Publisher` | | `internal/policy` | Cross-platform policy evaluation: mode (opt-in/all), label parsing, glob allow/deny (`Denied`, `Authorized`) | | `internal/daemon` | Event loop (`Run`, `listenEvents`, `consumeEvents`, `processOne`) and the coordinator that owns passes, retries, the `processed` dedup and container reservations | diff --git a/docs/testing.md b/docs/testing.md index 0f93c2a..737b878 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -121,11 +121,13 @@ or deployment behavior: 7. If `/run/dbus/system_bus_socket` is mounted and systemd is available, run `systemctl daemon-reload` and verify the daemon logs that it re-applied rules. -8. On a Swarm node, `kill -9` the wrapper task's container and confirm that - Swarm's replacement task starts without a name conflict and that exactly one - `swarm-device-access` daemon container is running afterwards - (`docker ps --filter 'name=^swarm-device-access$'`; without the anchors the - filter also matches the wrapper task's container). +8. On a Swarm node, `kill -9` the launcher task's container and confirm that + Swarm's replacement task removes the orphaned daemon and that exactly one + `swarm-device-access` daemon container is running afterwards, labelled with + the new launcher's ID + (`docker ps --filter 'name=^swarm-device-access$' --format '{{.ID}} {{.Label "io.github.leinardi.swarm-device-access.launcher"}}'`; + without the anchors the filter also matches the launcher task's container). + `TestLauncher_RunsDaemonFromItsOwnImage` covers the same case outside Swarm. 9. Verify observability by starting with `-metrics-addr 127.0.0.1:9090` and `-debug-addr 127.0.0.1:6060`, then checking `/healthz`, `/readyz`, `/metrics`, and diff --git a/examples/audio-passthrough/docker-compose.yaml b/examples/audio-passthrough/docker-compose.yaml index 77c5a23..a084b63 100644 --- a/examples/audio-passthrough/docker-compose.yaml +++ b/examples/audio-passthrough/docker-compose.yaml @@ -23,57 +23,29 @@ services: mode: replicated replicas: 1 swarm-device-access: - image: docker:29 + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. - entrypoint: ["sh", "-c"] + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the privileged set the daemon needs. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - # Restrict to ALSA sound devices only: - # set -- "$$@" -device-allow '/dev/snd/*' - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host config directory read-only at /etc/swarm-device-access: + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + # Restrict to ALSA sound devices only: + # - -device-allow=/dev/snd/* + # Leaves time for the launcher to stop the daemon and drain its logs. stop_grace_period: 30s deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first diff --git a/examples/gpu-passthrough/docker-compose.yaml b/examples/gpu-passthrough/docker-compose.yaml index 339607c..05a3797 100644 --- a/examples/gpu-passthrough/docker-compose.yaml +++ b/examples/gpu-passthrough/docker-compose.yaml @@ -28,60 +28,32 @@ services: mode: replicated replicas: 1 swarm-device-access: - image: docker:29 + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. - entrypoint: ["sh", "-c"] + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the privileged set the daemon needs. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - # Policy mode: opt-in (the default) processes only labelled containers; - # all processes every container on the node: - # set -- "$$@" -policy-mode opt-in - # Limit scope to NVIDIA devices only: - # set -- "$$@" -device-allow '/dev/nvidia*' - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host config directory read-only at /etc/swarm-device-access: + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + # Policy mode: opt-in (the default) processes only labelled containers; + # all processes every container on the node: + # - -policy-mode=opt-in + # Limit scope to NVIDIA devices only: + # - -device-allow=/dev/nvidia* + # Leaves time for the launcher to stop the daemon and drain its logs. stop_grace_period: 30s deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first diff --git a/examples/smartctl-exporter/docker-compose.yaml b/examples/smartctl-exporter/docker-compose.yaml index 5c83608..e805f58 100644 --- a/examples/smartctl-exporter/docker-compose.yaml +++ b/examples/smartctl-exporter/docker-compose.yaml @@ -39,61 +39,33 @@ services: mode: global swarm-device-access: - image: docker:29 + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. - entrypoint: ["sh", "-c"] + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the privileged set the daemon needs. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - # Restrict to disk devices only, one glob per flag: flags are not split - # on commas (only the labels are): - # set -- "$$@" -device-allow '/dev/sd*' - # set -- "$$@" -device-allow '/dev/nvme*' - # set -- "$$@" -device-allow '/dev/sg*' - # set -- "$$@" -device-allow '/dev/disk/*' - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host config directory read-only at /etc/swarm-device-access: + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + # Restrict to disk devices only, one glob per flag: flags are not split + # on commas (only the labels are): + # - -device-allow=/dev/sd* + # - -device-allow=/dev/nvme* + # - -device-allow=/dev/sg* + # - -device-allow=/dev/disk/* + # Leaves time for the launcher to stop the daemon and drain its logs. stop_grace_period: 30s deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first diff --git a/examples/usb-passthrough/docker-compose.yaml b/examples/usb-passthrough/docker-compose.yaml index e78d26f..12f134f 100644 --- a/examples/usb-passthrough/docker-compose.yaml +++ b/examples/usb-passthrough/docker-compose.yaml @@ -11,61 +11,33 @@ services: swarm-device-access: - image: docker:29 + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. - entrypoint: ["sh", "-c"] + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the privileged set the daemon needs. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - # Restrict to USB bus devices only. Nodes are /dev/bus/usb//, - # and * does not match across "/", so the glob needs both levels: - # set -- "$$@" -device-allow '/dev/bus/usb/*/*' - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host config directory read-only at /etc/swarm-device-access: + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + # Restrict to USB bus devices only. Nodes are /dev/bus/usb//, + # and * does not match across "/", so the glob needs both levels: + # - -device-allow=/dev/bus/usb/*/* + # Leaves time for the launcher to stop the daemon and drain its logs. stop_grace_period: 30s deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first usb-worker: image: ubuntu:24.04 diff --git a/examples/v4l2-passthrough/docker-compose.yaml b/examples/v4l2-passthrough/docker-compose.yaml index 16b6ac6..cb6974f 100644 --- a/examples/v4l2-passthrough/docker-compose.yaml +++ b/examples/v4l2-passthrough/docker-compose.yaml @@ -11,60 +11,32 @@ services: swarm-device-access: - image: docker:29 + image: ghcr.io/leinardi/swarm-device-access:1 volumes: - /var/run/docker.sock:/var/run/docker.sock - # Swarm rejects privileged/cgroup/pid/userns on services. This wrapper - # launches the actual daemon with `docker run`, where those flags are valid. - entrypoint: ["sh", "-c"] + # Swarm rejects privileged/cgroup/pid/userns on services. The launcher runs + # unprivileged and creates the daemon container from this same image, with + # the privileged set the daemon needs. command: - - | - # docker run options, before the image. Uncomment a line to add it. - set -- --rm --name=swarm-device-access \ - --privileged --cgroupns=host --pid=host --userns=host \ - -v /sys:/host/sys \ - -v /var/run/docker.sock:/var/run/docker.sock \ - -v /dev:/dev - # Reapply device rules after systemctl daemon-reload, which wipes cgroup - # BPF programs (without it the daemon warns and skips reload handling). - # The container-side path must be under /var/run: dhi.io/static has no - # /var/run -> /run symlink. - # set -- "$$@" -v /run/dbus/system_bus_socket:/var/run/dbus/system_bus_socket - # A config file for -config below. Mount the directory, not the file: - # an editor that replaces the file would leave a file mount on the old - # content, and a reload would silently apply it. - # set -- "$$@" -v /etc/swarm-device-access:/etc/swarm-device-access:ro - set -- "$$@" ghcr.io/leinardi/swarm-device-access:latest - # Daemon flags, after the image; before it they are docker run flags. - # Reload the config file with: docker kill -s HUP swarm-device-access - # set -- "$$@" -config /etc/swarm-device-access/config.yaml - # Restrict to V4L2 devices only: - # set -- "$$@" -device-allow '/dev/video*' - - # On SIGTERM or SIGINT from Swarm, stop the daemon and exit with its - # status. The trap is set before the daemon starts, and the daemon runs - # in the background: a shell waiting on a foreground child only runs - # its traps after the child exits. - pid= - stop_daemon() { - docker stop -t 10 swarm-device-access >/dev/null 2>&1 - wait $$pid - exit $$? - } - trap stop_daemon TERM INT - - # Clear a daemon container left behind by a wrapper that was killed. - docker rm -f swarm-device-access >/dev/null 2>&1 || true - docker run "$$@" & - pid=$$! - wait "$$pid" - exit $$? - # Leaves time for docker stop -t 10 before Swarm kills the wrapper. + - launch + # Launcher flags. Reapply device rules after systemctl daemon-reload: + # - -dbus + # Bind a host config directory read-only at /etc/swarm-device-access: + # - -config-dir=/etc/swarm-device-access + - -- + # Daemon flags, passed to the daemon verbatim. + # Reload the config file with: docker kill -s HUP swarm-device-access + # - -config=/etc/swarm-device-access/config.yaml + # Restrict to V4L2 devices only: + # - -device-allow=/dev/video* + # Leaves time for the launcher to stop the daemon and drain its logs. stop_grace_period: 30s deploy: mode: global restart_policy: condition: any + update_config: + order: stop-first v4l2-worker: image: linuxserver/ffmpeg:latest # dclint disable-line service-image-require-explicit-tag diff --git a/internal/launcher/fake_test.go b/internal/launcher/fake_test.go new file mode 100644 index 0000000..387ff71 --- /dev/null +++ b/internal/launcher/fake_test.go @@ -0,0 +1,483 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "bufio" + "bytes" + "context" + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "strings" + "sync" + "testing" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" +) + +const ( + testLauncherID = "1111111111111111111111111111111111111111111111111111111111111111" + testImageID = "sha256:2222222222222222222222222222222222222222222222222222222222222222" + testDaemonID = "3333333333333333333333333333333333333333333333333333333333333333" + + // stdout and stderr stream types of the multiplexed attach stream. + streamStdout = 1 + streamStderr = 2 +) + +var ( + errInjected = errors.New("injected failure") + errDenied = errors.New("permission denied") +) + +// hijackedConn is the launcher's end of a fake attach connection: closing it +// ends the reader, as closing a real hijacked connection does. +type hijackedConn struct { + net.Conn + + reader *io.PipeReader + closeOnce sync.Once + closed chan struct{} +} + +func newHijackedConn(reader *io.PipeReader) *hijackedConn { + return &hijackedConn{reader: reader, closed: make(chan struct{})} +} + +func (c *hijackedConn) Close() error { + c.closeOnce.Do(func() { close(c.closed) }) + + return c.reader.Close() //nolint:wrapcheck // fake connection +} + +type stopCall struct { + id string + timeout int +} + +// fakeDocker is an in-memory Docker engine for one launcher run. +type fakeDocker struct { + mu sync.Mutex + calls []string + + inspect map[string]client.ContainerInspectResult + inspectEr map[string]error + list []container.Summary + removed []string + created client.ContainerCreateOptions + stops []stopCall + + createErr error + attachErr error + startErr error + // attachHang and waitHang block the call until its context ends. + attachHang bool + waitHang bool + // attachStuck blocks the attach, whatever its context, until it is + // closed, like a dockerd that accepts the upgrade and never answers. + attachStuck chan struct{} + // stuckConn is the connection the stuck attach returns once released. + stuckConn *hijackedConn + // daemonGone reports the created daemon as removed to inspect. + daemonGone bool + // stop replaces the default stop, which ends the daemon. + stop func(ctx context.Context) + + // The daemon's side of the attach stream and of the wait. + output *io.PipeWriter + outputOnce sync.Once + waitResult chan container.WaitResponse + waitErr chan error + started chan struct{} +} + +func newFakeDocker() *fakeDocker { + return &fakeDocker{ + inspect: map[string]client.ContainerInspectResult{ + testLauncherID: {Container: container.InspectResponse{ + ID: testLauncherID, + Image: testImageID, + State: &container.State{Running: true}, + }}, + }, + inspectEr: map[string]error{}, + waitResult: make(chan container.WaitResponse, 1), + waitErr: make(chan error, 1), + started: make(chan struct{}), + } +} + +func (f *fakeDocker) ContainerInspect( + _ context.Context, + containerID string, + _ client.ContainerInspectOptions, +) (client.ContainerInspectResult, error) { + f.record("inspect " + containerID) + + f.mu.Lock() + defer f.mu.Unlock() + + err, failed := f.inspectEr[containerID] + if failed { + return client.ContainerInspectResult{}, err + } + + if containerID == testDaemonID && !f.daemonGone { + return client.ContainerInspectResult{Container: container.InspectResponse{ + ID: testDaemonID, + State: &container.State{Running: true}, + }}, nil + } + + res, found := f.inspect[containerID] + if !found { + return client.ContainerInspectResult{}, fmt.Errorf( + "no such container %s: %w", + containerID, + cerrdefs.ErrNotFound, + ) + } + + return res, nil +} + +func (f *fakeDocker) ContainerList( + _ context.Context, + _ client.ContainerListOptions, +) (client.ContainerListResult, error) { + f.record("list") + + return client.ContainerListResult{Items: f.list}, nil +} + +func (f *fakeDocker) ContainerRemove( + _ context.Context, + containerID string, + _ client.ContainerRemoveOptions, +) (client.ContainerRemoveResult, error) { + f.record("remove " + containerID) + + f.mu.Lock() + f.removed = append(f.removed, containerID) + f.mu.Unlock() + + return client.ContainerRemoveResult{}, nil +} + +func (f *fakeDocker) ContainerCreate( + _ context.Context, + options client.ContainerCreateOptions, +) (client.ContainerCreateResult, error) { + f.record("create") + + if f.createErr != nil { + return client.ContainerCreateResult{}, f.createErr + } + + f.mu.Lock() + f.created = options + f.mu.Unlock() + + return client.ContainerCreateResult{ID: testDaemonID}, nil +} + +func (f *fakeDocker) ContainerAttach( + ctx context.Context, + _ string, + _ client.ContainerAttachOptions, +) (client.ContainerAttachResult, error) { + f.record("attach") + + if f.attachHang { + <-ctx.Done() + + return client.ContainerAttachResult{}, fmt.Errorf("attach: %w", ctx.Err()) + } + + if f.attachStuck != nil { + <-f.attachStuck + + reader, _ := io.Pipe() + conn := newHijackedConn(reader) + + f.mu.Lock() + f.stuckConn = conn + f.mu.Unlock() + + return client.ContainerAttachResult{HijackedResponse: client.HijackedResponse{ + Conn: conn, + Reader: bufio.NewReader(reader), + }}, nil + } + + if f.attachErr != nil { + return client.ContainerAttachResult{}, f.attachErr + } + + reader, writer := io.Pipe() + f.output = writer + + return client.ContainerAttachResult{HijackedResponse: client.HijackedResponse{ + Conn: newHijackedConn(reader), + Reader: bufio.NewReader(reader), + }}, nil +} + +func (f *fakeDocker) ContainerWait( + ctx context.Context, + _ string, + _ client.ContainerWaitOptions, +) client.ContainerWaitResult { + f.record("wait") + + if f.waitHang { + <-ctx.Done() + + errs := make(chan error, 1) + errs <- ctx.Err() + + return client.ContainerWaitResult{Result: make(chan container.WaitResponse), Error: errs} + } + + // The stream only fails on its own when a test injects an error, or + // when its context ends: a deadline on it would show up here. + errs := make(chan error, 1) + + go func() { + select { + case err := <-f.waitErr: + errs <- err + case <-ctx.Done(): + errs <- ctx.Err() + } + }() + + return client.ContainerWaitResult{Result: f.waitResult, Error: errs} +} + +func (f *fakeDocker) ContainerStart( + _ context.Context, + _ string, + _ client.ContainerStartOptions, +) (client.ContainerStartResult, error) { + f.record("start") + + if f.startErr != nil { + return client.ContainerStartResult{}, f.startErr + } + + close(f.started) + + return client.ContainerStartResult{}, nil +} + +func (f *fakeDocker) ContainerStop( + ctx context.Context, + containerID string, + options client.ContainerStopOptions, +) (client.ContainerStopResult, error) { + f.record("stop " + containerID) + + timeout := -1 + if options.Timeout != nil { + timeout = *options.Timeout + } + + f.mu.Lock() + f.stops = append(f.stops, stopCall{id: containerID, timeout: timeout}) + f.mu.Unlock() + + if f.stop != nil { + f.stop(ctx) + } else { + f.emit(streamStdout, "daemon stopping") + f.exit(0) + } + + return client.ContainerStopResult{}, nil +} + +func (f *fakeDocker) record(call string) { + f.mu.Lock() + defer f.mu.Unlock() + + f.calls = append(f.calls, call) +} + +func (f *fakeDocker) recorded() []string { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]string(nil), f.calls...) +} + +func (f *fakeDocker) stopCalls() []stopCall { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]stopCall(nil), f.stops...) +} + +func (f *fakeDocker) removedIDs() []string { + f.mu.Lock() + defer f.mu.Unlock() + + return append([]string(nil), f.removed...) +} + +// emit writes one multiplexed line of daemon output. It returns once the +// launcher has read it. +func (f *fakeDocker) emit(stream byte, line string) { + payload := []byte(line + "\n") + size := uint32(len(payload)) //nolint:gosec // test lines are a few bytes + + var frame bytes.Buffer + + frame.Write([]byte{stream, 0, 0, 0}) + _ = binary.Write(&frame, binary.BigEndian, size) + frame.Write(payload) + + _, _ = f.output.Write(frame.Bytes()) +} + +// closeOutput ends the attach stream, with an error when err is set. +func (f *fakeDocker) closeOutput(err error) { + f.outputOnce.Do(func() { _ = f.output.CloseWithError(err) }) +} + +// exit ends the daemon's output and reports its removal with status. +func (f *fakeDocker) exit(status int64) { + f.closeOutput(nil) + f.markGone() + + f.waitResult <- container.WaitResponse{StatusCode: status} +} + +// markGone makes inspect report the daemon removed. +func (f *fakeDocker) markGone() { + f.mu.Lock() + defer f.mu.Unlock() + + f.daemonGone = true +} + +// syncBuffer is a bytes.Buffer safe for the copy goroutine and the test. +type syncBuffer struct { + mu sync.Mutex + buf bytes.Buffer +} + +func (b *syncBuffer) Write(data []byte) (int, error) { + b.mu.Lock() + defer b.mu.Unlock() + + return b.buf.Write(data) //nolint:wrapcheck // test buffer +} + +func (b *syncBuffer) String() string { + b.mu.Lock() + defer b.mu.Unlock() + + return b.buf.String() +} + +type runResult struct { + status int + err error +} + +type testRun struct { + docker *fakeDocker + runner *runner + stdout *syncBuffer + stderr *syncBuffer +} + +func newTestRun(docker *fakeDocker) *testRun { + stdout, stderr := &syncBuffer{}, &syncBuffer{} + + return &testRun{ + docker: docker, + stdout: stdout, + stderr: stderr, + runner: &runner{ + docker: docker, + opts: Options{HostDockerSocket: DefaultHostDockerSocket}, + stdout: stdout, + stderr: stderr, + selfID: func() (string, error) { return testLauncherID, nil }, + callTimeout: 2 * time.Second, + shutdownBudget: 5 * time.Second, + pollInterval: 10 * time.Millisecond, + }, + } +} + +// start runs the launcher in the background. +func (tr *testRun) start(ctx context.Context) <-chan runResult { + done := make(chan runResult, 1) + + go func() { + status, err := tr.runner.run(ctx) + done <- runResult{status: status, err: err} + }() + + return done +} + +// waitStarted blocks until the daemon container was started. +func (tr *testRun) waitStarted(t *testing.T) { + t.Helper() + + select { + case <-tr.docker.started: + case <-time.After(2 * time.Second): + t.Fatalf("daemon not started; calls: %v", tr.docker.recorded()) + } +} + +// result waits for the launcher to return. +func result(t *testing.T, done <-chan runResult) runResult { + t.Helper() + + select { + case res := <-done: + return res + case <-time.After(10 * time.Second): + t.Fatal("launcher did not return") + + return runResult{} + } +} + +func contains(calls []string, want string) bool { + for _, call := range calls { + if strings.HasPrefix(call, want) { + return true + } + } + + return false +} diff --git a/internal/launcher/launcher.go b/internal/launcher/launcher.go new file mode 100644 index 0000000..856f7d2 --- /dev/null +++ b/internal/launcher/launcher.go @@ -0,0 +1,539 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +// Package launcher runs the privileged daemon container from the image the +// launcher itself runs, and supervises it: the launcher is an unprivileged +// Swarm service task, so Swarm's rolling updates, rollbacks and image +// watchers act on the daemon too. +package launcher + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "os" + "strings" + "time" + + cerrdefs "github.com/containerd/errdefs" + "github.com/moby/moby/api/pkg/stdcopy" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" + + "github.com/leinardi/swarm-device-access/internal/daemon" + "github.com/leinardi/swarm-device-access/internal/logger" +) + +const ( + // DockerHost is the Docker endpoint the launcher uses: its own socket + // mount, never a configurable path, so it cannot disagree with the + // socket bound into the daemon. + DockerHost = "unix://" + containerDockerSocket + + component = "launcher" + + // imageIDPrefix is the prefix of a content-addressed image ID. A tag + // would let the daemon run a different image than the launcher. + imageIDPrefix = "sha256:" + + // shutdownBudget bounds everything a shutdown does, the stop, the wait + // for the removal and the log drain together. It stays inside the + // compose files' stop_grace_period of 30s, so Swarm never kills a + // launcher that is still stopping its daemon. + shutdownBudget = stopTimeout + daemon.DockerCallTimeout + + removalPollInterval = 250 * time.Millisecond +) + +var ( + errEstablishTimeout = errors.New("not established before timeout") + errNoImageID = errors.New("launcher container has no content-addressed image ID") + errWaitStream = errors.New("daemon wait stream failed") + errLogStream = errors.New("daemon log stream lost while the daemon was running") + errShutdownBudget = errors.New("daemon not removed within the shutdown budget") +) + +// dockerAPI is the subset of *client.Client the launcher uses. +type dockerAPI interface { + ContainerInspect( + ctx context.Context, + containerID string, + options client.ContainerInspectOptions, + ) (client.ContainerInspectResult, error) + ContainerList( + ctx context.Context, + options client.ContainerListOptions, + ) (client.ContainerListResult, error) + ContainerRemove( + ctx context.Context, + containerID string, + options client.ContainerRemoveOptions, + ) (client.ContainerRemoveResult, error) + ContainerCreate( + ctx context.Context, + options client.ContainerCreateOptions, + ) (client.ContainerCreateResult, error) + ContainerAttach( + ctx context.Context, + containerID string, + options client.ContainerAttachOptions, + ) (client.ContainerAttachResult, error) + ContainerWait( + ctx context.Context, + containerID string, + options client.ContainerWaitOptions, + ) client.ContainerWaitResult + ContainerStart( + ctx context.Context, + containerID string, + options client.ContainerStartOptions, + ) (client.ContainerStartResult, error) + ContainerStop( + ctx context.Context, + containerID string, + options client.ContainerStopOptions, + ) (client.ContainerStopResult, error) +} + +type runner struct { + docker dockerAPI + opts Options + stdout io.Writer + stderr io.Writer + selfID func() (string, error) + + callTimeout time.Duration + shutdownBudget time.Duration + // pollInterval spaces the inspects that confirm a removal the wait + // stream can no longer report. + pollInterval time.Duration +} + +// session is a created daemon container with its log and wait streams. +type session struct { + id string + hijacked client.HijackedResponse + wait client.ContainerWaitResult + + // copyDone reports the log copy's result once; copyFinished and copyErr + // keep it, so later drains read them instead of the channel. + copyDone chan error + copyFinished bool + copyErr error + + cancelAttach context.CancelFunc + cancelWait context.CancelFunc +} + +// Run creates the daemon container from the launcher's own image, streams +// its output to the launcher's stdout and stderr, and returns its exit +// status once it is removed. ctx ends the daemon: it is stopped and Run +// still returns its status. An error means the launcher failed; the status +// is then 1. +func Run(ctx context.Context, docker dockerAPI, opts *Options) (int, error) { + run := &runner{ + docker: docker, + opts: *opts, + stdout: os.Stdout, + stderr: os.Stderr, + selfID: selfIDFromMountinfo, + callTimeout: daemon.DockerCallTimeout, + shutdownBudget: shutdownBudget, + pollInterval: removalPollInterval, + } + + return run.run(ctx) +} + +func launcherLog() *slog.Logger { + return logger.L().With("component", component) +} + +func (r *runner) run(ctx context.Context) (int, error) { + launcherID, imageID, err := r.identify(ctx) + if err != nil { + return 1, err + } + + err = r.removeStale(ctx, launcherID) + if err != nil { + return 1, err + } + + createCtx, cancelCreate := context.WithTimeout(ctx, r.callTimeout) + created, err := r.docker.ContainerCreate(createCtx, daemonSpec(&r.opts, imageID, launcherID)) + + cancelCreate() + + if err != nil { + return 1, fmt.Errorf("create daemon container: %w", err) + } + + for _, warning := range created.Warnings { + launcherLog().Warn("docker warning on daemon create", "warning", warning) + } + + sess, err := r.prepare(ctx, created.ID) + if err != nil { + return 1, r.abandon(ctx, created.ID, err) + } + defer sess.close() + + startCtx, cancelStart := context.WithTimeout(ctx, r.callTimeout) + _, err = r.docker.ContainerStart(startCtx, created.ID, client.ContainerStartOptions{}) + + cancelStart() + + if err != nil { + return 1, r.abandon(ctx, created.ID, fmt.Errorf("start daemon container: %w", err)) + } + + launcherLog().Info("daemon started", + "id", shortID(created.ID), "image", imageID, "launcher", shortID(launcherID)) + + return r.supervise(ctx, sess) +} + +// identify returns the launcher's container ID and the ID of the image it +// runs. Swarm already pulled the task's digest, so no pull is needed. +func (r *runner) identify(ctx context.Context) (launcherID, imageID string, err error) { + launcherID, err = r.selfID() + if err != nil { + return "", "", fmt.Errorf("identify launcher container: %w", err) + } + + inspectCtx, cancel := context.WithTimeout(ctx, r.callTimeout) + defer cancel() + + self, err := r.docker.ContainerInspect(inspectCtx, launcherID, client.ContainerInspectOptions{}) + if err != nil { + return "", "", fmt.Errorf("inspect launcher container %s: %w", shortID(launcherID), err) + } + + imageID = self.Container.Image + if !strings.HasPrefix(imageID, imageIDPrefix) { + return "", "", fmt.Errorf("%w: %q", errNoImageID, imageID) + } + + return launcherID, imageID, nil +} + +// abandon removes a daemon container whose setup failed, so a failed +// launcher never leaves a daemon behind. +func (r *runner) abandon(ctx context.Context, id string, cause error) error { + return errors.Join(cause, r.removeContainer(ctx, id)) +} + +// prepare attaches to the daemon's output, starts copying it and registers +// the wait for its removal, all before the daemon starts, in the order the +// docker run CLI uses: nothing the daemon prints or its exit is missed, and +// the name is free when the wait returns. +func (r *runner) prepare(ctx context.Context, containerID string) (*session, error) { + attached, cancelAttach, err := establish(ctx, r.callTimeout, + func(streamCtx context.Context) (client.ContainerAttachResult, error) { + return r.docker.ContainerAttach(streamCtx, containerID, client.ContainerAttachOptions{ + Stream: true, + Stdout: true, + Stderr: true, + }) + }, + func(res client.ContainerAttachResult) { res.Close() }, + ) + if err != nil { + return nil, fmt.Errorf("attach to daemon container: %w", err) + } + + sess := &session{ + id: containerID, + hijacked: attached.HijackedResponse, + copyDone: make(chan error, 1), + cancelAttach: cancelAttach, + cancelWait: func() {}, + } + + // Started before the daemon: copying after ContainerStart returns could + // block the daemon on a full attach buffer and deadlock the start. + go func() { + _, copyErr := stdcopy.StdCopy(r.stdout, r.stderr, sess.hijacked.Reader) + sess.copyDone <- copyErr + }() + + sess.wait, sess.cancelWait, err = establish(ctx, r.callTimeout, + func(streamCtx context.Context) (client.ContainerWaitResult, error) { + res := r.docker.ContainerWait(streamCtx, containerID, client.ContainerWaitOptions{ + Condition: container.WaitConditionRemoved, + }) + + // A request that failed outright reports it before returning. + select { + case waitErr := <-res.Error: + return res, waitErr + default: + return res, nil + } + }, + func(client.ContainerWaitResult) {}, + ) + if err != nil { + sess.close() + + return nil, fmt.Errorf("wait for daemon container: %w", err) + } + + return sess, nil +} + +// establish runs a streaming call with a bounded establishment. The stream +// gets its own context, detached from ctx: a signal must not cut the +// stream before the daemon's status and last lines arrive, and a deadline +// would kill it once established. The call is raced against a timer +// instead; on timeout (or ctx ending first) its context is canceled and +// establish returns at once. It does not wait for the call: the client's +// hijacked attach reads the upgrade response on a raw connection that no +// context reaches, so a dockerd that accepts and never answers would block +// the launcher forever. A result the call still returns later is released +// by a background goroutine. On success the returned cancel func owns the +// stream context. +// +//nolint:ireturn // T is the concrete result of the one streaming call; generic over the two stream kinds +func establish[T any]( + ctx context.Context, + timeout time.Duration, + call func(context.Context) (T, error), + release func(T), +) (T, context.CancelFunc, error) { + type outcome struct { + val T + err error + } + + streamCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) + done := make(chan outcome, 1) + + go func() { + val, err := call(streamCtx) + done <- outcome{val: val, err: err} + }() + + timer := time.NewTimer(timeout) + defer timer.Stop() + + var ( + zero T + cause error + ) + + select { + case out := <-done: + if out.err != nil { + cancel() + + return zero, func() {}, out.err + } + + return out.val, cancel, nil + case <-timer.C: + cause = errEstablishTimeout + case <-ctx.Done(): + cause = fmt.Errorf("canceled: %w", ctx.Err()) + } + + cancel() + + go func() { + out := <-done + if out.err == nil { + release(out.val) + } + }() + + return zero, func() {}, cause +} + +// supervise waits for the first of: the daemon's removal, a broken wait +// stream, a lost log stream, or ctx ending. +func (r *runner) supervise(ctx context.Context, sess *session) (int, error) { + for { + select { + case res := <-sess.wait.Result: + drainCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), r.callTimeout) + status := sess.finish(drainCtx, &res) + + cancel() + + return status, nil + + case waitErr := <-sess.wait.Error: + launcherLog().Error("daemon wait stream failed; stopping the daemon", "err", waitErr) + + return r.shutdown(ctx, sess, waitErr) + + case copyErr := <-sess.pendingCopy(): + sess.recordCopy(copyErr) + + if copyErr == nil { + // End of stream: the daemon is exiting; its status follows. + continue + } + + launcherLog().Error("daemon log stream lost; stopping the daemon", "err", copyErr) + + _, shutdownErr := r.shutdown(ctx, sess, nil) + + return 1, errors.Join(fmt.Errorf("%w: %w", errLogStream, copyErr), shutdownErr) + + case <-ctx.Done(): + launcherLog().Info("stopping daemon", "id", shortID(sess.id)) + + return r.shutdown(ctx, sess, nil) + } + } +} + +// shutdown stops the daemon and waits for its removal and its last log +// lines, all within one shutdownBudget. waitErr is set when the wait stream +// has already failed, or it fails here: the removal is then confirmed by +// polling inspect instead, the output is still drained, and the status, +// which nothing reports any more, is 1. +func (r *runner) shutdown(ctx context.Context, sess *session, waitErr error) (int, error) { + budgetCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), r.shutdownBudget) + defer cancel() + + r.stop(budgetCtx, sess.id) + + if waitErr == nil { + select { + case res := <-sess.wait.Result: + return sess.finish(budgetCtx, &res), nil + case waitErr = <-sess.wait.Error: + case <-budgetCtx.Done(): + return 1, r.budgetExceeded() + } + } + + removeErr := r.awaitRemoval(budgetCtx, sess.id) + sess.drain(budgetCtx) + + return 1, errors.Join(fmt.Errorf("%w: %w", errWaitStream, waitErr), removeErr) +} + +// awaitRemoval polls until inspect reports containerID gone, or ctx ends. +// At the budget, the latest inspect error, if the last inspect failed, is +// returned with the timeout: a permission or connection error is the +// diagnosis, not the budget. +func (r *runner) awaitRemoval(ctx context.Context, containerID string) error { + ticker := time.NewTicker(r.pollInterval) + defer ticker.Stop() + + for { + inspectCtx, cancel := context.WithTimeout(ctx, r.callTimeout) + _, err := r.docker.ContainerInspect( + inspectCtx, + containerID, + client.ContainerInspectOptions{}, + ) + + cancel() + + if cerrdefs.IsNotFound(err) { + return nil + } + + select { + case <-ticker.C: + case <-ctx.Done(): + if err != nil { + return errors.Join( + r.budgetExceeded(), + fmt.Errorf("last inspect of daemon: %w", err), + ) + } + + return r.budgetExceeded() + } + } +} + +func (r *runner) budgetExceeded() error { + return fmt.Errorf("%w (%s)", errShutdownBudget, r.shutdownBudget) +} + +// stop stops the daemon by ID, never by name, so a launcher cannot stop a +// daemon that a newer launcher created. +func (r *runner) stop(ctx context.Context, id string) { + seconds := int(stopTimeout / time.Second) + + _, err := r.docker.ContainerStop(ctx, id, client.ContainerStopOptions{Timeout: &seconds}) + if err != nil && !cerrdefs.IsNotFound(err) { + launcherLog().Warn("stop daemon failed", "id", shortID(id), "err", err) + } +} + +// finish drains the daemon's remaining output within ctx and returns its +// exit status. +func (s *session) finish(ctx context.Context, res *container.WaitResponse) int { + s.drain(ctx) + + if res.Error != nil && res.Error.Message != "" { + launcherLog().Warn("daemon wait reported an error", "err", res.Error.Message) + } + + launcherLog().Info("daemon exited", "id", shortID(s.id), "status", res.StatusCode) + + return int(res.StatusCode) +} + +// drain waits, within ctx, for the daemon's output to end. +func (s *session) drain(ctx context.Context) { + if !s.copyFinished { + select { + case copyErr := <-s.copyDone: + s.recordCopy(copyErr) + case <-ctx.Done(): + launcherLog().Warn("daemon log stream not drained before timeout", "id", shortID(s.id)) + } + } + + if s.copyErr != nil { + launcherLog().Warn("daemon log stream ended with an error", "err", s.copyErr) + } +} + +// pendingCopy returns copyDone until its one result is recorded, then nil, +// which a select never receives from. +func (s *session) pendingCopy() <-chan error { + if s.copyFinished { + return nil + } + + return s.copyDone +} + +func (s *session) recordCopy(err error) { + s.copyFinished = true + s.copyErr = err +} + +// close releases both streams. +func (s *session) close() { + s.hijacked.Close() + s.cancelAttach() + s.cancelWait() +} diff --git a/internal/launcher/launcher_test.go b/internal/launcher/launcher_test.go new file mode 100644 index 0000000..597a80c --- /dev/null +++ b/internal/launcher/launcher_test.go @@ -0,0 +1,557 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "context" + "errors" + "slices" + "strings" + "testing" + "time" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" +) + +func TestRun_CallOrder(t *testing.T) { + t.Parallel() + + run := newTestRun(newFakeDocker()) + done := run.start(t.Context()) + run.waitStarted(t) + run.docker.exit(0) + + res := result(t, done) + if res.err != nil || res.status != 0 { + t.Fatalf("run = %d, %v; want 0, nil", res.status, res.err) + } + + want := []string{"inspect " + testLauncherID, "list", "create", "attach", "wait", "start"} + if got := run.docker.recorded(); !slices.Equal(got, want) { + t.Errorf("calls = %v, want %v", got, want) + } + + created := run.docker.created + if created.Name != DaemonName || created.Config.Image != testImageID || + created.Config.Labels[LabelLauncher] != testLauncherID { + t.Errorf("created %+v, want the daemon from the launcher's image and ID", created) + } +} + +// The wait and attach streams have no deadline: a daemon that runs far +// longer than the call timeout keeps running, and its status is returned. +func TestRun_LongLivedDaemon(t *testing.T) { + t.Parallel() + + const callTimeout = 20 * time.Millisecond + + run := newTestRun(newFakeDocker()) + run.runner.callTimeout = callTimeout + + done := run.start(t.Context()) + run.waitStarted(t) + + // Real elapsed window: well past the call timeout, which a deadline on + // either stream would have hit. + time.Sleep(10 * callTimeout) + + select { + case res := <-done: + t.Fatalf("launcher returned while the daemon runs: %d, %v", res.status, res.err) + default: + } + + run.docker.emit(streamStdout, "still here") + run.docker.exit(3) + + res := result(t, done) + if res.err != nil || res.status != 3 { + t.Fatalf("run = %d, %v; want 3, nil", res.status, res.err) + } + + if !strings.Contains(run.stdout.String(), "still here") { + t.Errorf("stdout = %q, want the daemon's output", run.stdout.String()) + } +} + +func TestRun_EstablishmentBounded(t *testing.T) { + t.Parallel() + + for _, stream := range []string{"attach", "wait"} { + t.Run(stream, func(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + docker.attachHang = stream == "attach" + docker.waitHang = stream == "wait" + + run := newTestRun(docker) + run.runner.callTimeout = 50 * time.Millisecond + + res := result(t, run.start(t.Context())) + if !errors.Is(res.err, errEstablishTimeout) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errEstablishTimeout) + } + + calls := docker.recorded() + if contains(calls, "start") { + t.Errorf("daemon started after a failed %s: %v", stream, calls) + } + + if !slices.Equal(docker.removedIDs(), []string{testDaemonID}) { + t.Errorf("removed %v, want the created daemon", docker.removedIDs()) + } + }) + } +} + +// A Docker call that ignores its canceled context (the hijacked attach reads +// the upgrade response on a raw connection) must not hold the launcher past +// the timeout; the result it returns later is released. +func TestRun_EstablishmentIgnoringCancel(t *testing.T) { + t.Parallel() + + const callTimeout = 50 * time.Millisecond + + docker := newFakeDocker() + docker.attachStuck = make(chan struct{}) + + run := newTestRun(docker) + run.runner.callTimeout = callTimeout + + began := time.Now() + + res := result(t, run.start(t.Context())) + if !errors.Is(res.err, errEstablishTimeout) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errEstablishTimeout) + } + + if elapsed := time.Since(began); elapsed > 20*callTimeout { + t.Errorf("returned after %s, want soon after the %s timeout", elapsed, callTimeout) + } + + if !slices.Equal(docker.removedIDs(), []string{testDaemonID}) { + t.Errorf("removed %v, want the created daemon", docker.removedIDs()) + } + + close(docker.attachStuck) + + // Positive eventual: the late connection is closed by the background + // release; closed is the observable. + select { + case <-waitStuckConn(t, docker).closed: + case <-time.After(2 * time.Second): + t.Error("late attach result was not released") + } +} + +// waitStuckConn polls until the stuck attach has returned its connection. +func waitStuckConn(t *testing.T, docker *fakeDocker) *hijackedConn { + t.Helper() + + deadline := time.Now().Add(2 * time.Second) + + for time.Now().Before(deadline) { + docker.mu.Lock() + conn := docker.stuckConn + docker.mu.Unlock() + + if conn != nil { + return conn + } + + time.Sleep(5 * time.Millisecond) + } + + t.Fatal("stuck attach never returned") + + return nil +} + +func TestRun_AttachErrorRemovesCreated(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + docker.attachErr = errInjected + + res := result(t, newTestRun(docker).start(t.Context())) + if !errors.Is(res.err, errInjected) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errInjected) + } + + if contains(docker.recorded(), "start") { + t.Errorf("daemon started after a failed attach: %v", docker.recorded()) + } + + if !slices.Equal(docker.removedIDs(), []string{testDaemonID}) { + t.Errorf("removed %v, want the created daemon", docker.removedIDs()) + } +} + +func TestRun_StartErrorRemovesCreated(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + docker.startErr = errInjected + + res := result(t, newTestRun(docker).start(t.Context())) + if !errors.Is(res.err, errInjected) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errInjected) + } + + if !slices.Equal(docker.removedIDs(), []string{testDaemonID}) { + t.Errorf("removed %v, want the created daemon", docker.removedIDs()) + } +} + +// The last lines the daemon prints, after its removal is already reported, +// still reach the launcher's output before Run returns. +func TestRun_FinalLogsDrained(t *testing.T) { + t.Parallel() + + run := newTestRun(newFakeDocker()) + done := run.start(t.Context()) + run.waitStarted(t) + + run.docker.emit(streamStdout, "out line") + run.docker.emit(streamStderr, "err line") + + run.docker.waitResult <- container.WaitResponse{StatusCode: 2} + + run.docker.emit(streamStdout, "last out") + run.docker.emit(streamStderr, "last err") + run.docker.closeOutput(nil) + + res := result(t, done) + if res.err != nil || res.status != 2 { + t.Fatalf("run = %d, %v; want 2, nil", res.status, res.err) + } + + if got := run.stdout.String(); got != "out line\nlast out\n" { + t.Errorf("stdout = %q", got) + } + + if got := run.stderr.String(); got != "err line\nlast err\n" { + t.Errorf("stderr = %q", got) + } +} + +// A log stream that fails while the daemon runs would lose its logs +// silently: the launcher stops the daemon by ID and fails. +func TestRun_LogStreamLost(t *testing.T) { + t.Parallel() + + run := newTestRun(newFakeDocker()) + run.docker.stop = func(context.Context) { run.docker.waitResult <- container.WaitResponse{} } + + done := run.start(t.Context()) + run.waitStarted(t) + run.docker.closeOutput(errInjected) + + res := result(t, done) + if !errors.Is(res.err, errLogStream) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errLogStream) + } + + requireStopped(t, run.docker) +} + +// An inspect that keeps failing while the removal is polled is reported +// with the budget, not replaced by it. +func TestRun_WaitStreamErrorKeepsInspectError(t *testing.T) { + t.Parallel() + + run := newTestRun(newFakeDocker()) + run.runner.shutdownBudget = 100 * time.Millisecond + run.docker.stop = func(context.Context) {} + + done := run.start(t.Context()) + run.waitStarted(t) + + run.docker.mu.Lock() + run.docker.inspectEr[testDaemonID] = errDenied + run.docker.mu.Unlock() + + run.docker.waitErr <- errInjected + + res := result(t, done) + if !errors.Is(res.err, errShutdownBudget) || !errors.Is(res.err, errDenied) || res.status != 1 { + t.Fatalf( + "run = %d, %v; want 1, %v and %v", + res.status, + res.err, + errShutdownBudget, + errDenied, + ) + } +} + +// With the wait stream broken, the launcher still stops the daemon by ID, +// confirms its removal by inspecting it, and drains its last lines. +func TestRun_WaitStreamError(t *testing.T) { + t.Parallel() + + const step = 50 * time.Millisecond + + run := newTestRun(newFakeDocker()) + run.docker.stop = func(context.Context) { + go func() { + time.Sleep(step) + run.docker.emit(streamStdout, "daemon last words") + run.docker.closeOutput(nil) + time.Sleep(step) + run.docker.markGone() + }() + } + + done := run.start(t.Context()) + run.waitStarted(t) + + run.docker.waitErr <- errInjected + + res := result(t, done) + if !errors.Is(res.err, errWaitStream) || !errors.Is(res.err, errInjected) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errWaitStream) + } + + if errors.Is(res.err, errShutdownBudget) { + t.Errorf("removal not confirmed: %v", res.err) + } + + requireStopped(t, run.docker) + + calls := run.docker.recorded() + if !contains(calls[slices.Index(calls, "stop "+testDaemonID):], "inspect "+testDaemonID) { + t.Errorf("calls %v, want the removal confirmed by inspect after the stop", calls) + } + + if !strings.Contains(run.stdout.String(), "daemon last words") { + t.Errorf("stdout = %q, want the daemon's last line", run.stdout.String()) + } +} + +// A daemon that never disappears after a broken wait stream ends the +// shutdown at the budget. +func TestRun_WaitStreamErrorBudget(t *testing.T) { + t.Parallel() + + const budget = 200 * time.Millisecond + + run := newTestRun(newFakeDocker()) + run.runner.shutdownBudget = budget + run.docker.stop = func(context.Context) {} + + done := run.start(t.Context()) + run.waitStarted(t) + + began := time.Now() + + run.docker.waitErr <- errInjected + + res := result(t, done) + if !errors.Is(res.err, errWaitStream) || !errors.Is(res.err, errShutdownBudget) || + res.status != 1 { + t.Fatalf( + "run = %d, %v; want 1, %v and %v", + res.status, + res.err, + errWaitStream, + errShutdownBudget, + ) + } + + if elapsed := time.Since(began); elapsed > 3*budget { + t.Errorf("shutdown took %s, want within the %s budget", elapsed, budget) + } +} + +// A signal stops the daemon by ID; the launcher still waits for its +// removal on the stream context, drains its output and returns its status. +func TestRun_CancelStopsDaemon(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + run := newTestRun(newFakeDocker()) + done := run.start(ctx) + run.waitStarted(t) + cancel() + + res := result(t, done) + if res.err != nil || res.status != 0 { + t.Fatalf("run = %d, %v; want 0, nil", res.status, res.err) + } + + requireStopped(t, run.docker) + + if !strings.Contains(run.stdout.String(), "daemon stopping") { + t.Errorf("stdout = %q, want the daemon's last line", run.stdout.String()) + } +} + +// The stop, the wait for removal and the drain share one budget: each is +// slow, and together they still end within one budget, not the sum. +func TestRun_ShutdownBudgetShared(t *testing.T) { + t.Parallel() + + const ( + budget = 500 * time.Millisecond + slowStop = 200 * time.Millisecond + slowRemove = 150 * time.Millisecond + ) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + run := newTestRun(newFakeDocker()) + run.runner.shutdownBudget = budget + // Longer than the budget, so a drain bounded by it would show. + run.runner.callTimeout = 10 * budget + run.docker.stop = func(context.Context) { + time.Sleep(slowStop) + + go func() { + time.Sleep(slowRemove) + + run.docker.waitResult <- container.WaitResponse{StatusCode: 0} + }() + // The output is never closed: the drain runs out of budget. + } + + done := run.start(ctx) + run.waitStarted(t) + + began := time.Now() + + cancel() + + res := result(t, done) + elapsed := time.Since(began) + + if res.err != nil || res.status != 0 { + t.Fatalf("run = %d, %v; want 0, nil", res.status, res.err) + } + + if elapsed > 2*budget { + t.Errorf("shutdown took %s, want within one budget of %s", elapsed, budget) + } +} + +func TestRun_ShutdownBudgetExhausted(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + run := newTestRun(newFakeDocker()) + run.runner.shutdownBudget = 100 * time.Millisecond + run.docker.stop = func(context.Context) {} + + done := run.start(ctx) + run.waitStarted(t) + cancel() + + res := result(t, done) + if !errors.Is(res.err, errShutdownBudget) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errShutdownBudget) + } + + run.docker.closeOutput(nil) +} + +// An output stream that ends before the removal is reported leaves nothing +// to drain: the status is returned at once, not after a drain timeout. +func TestRun_EOFBeforeExit(t *testing.T) { + t.Parallel() + + run := newTestRun(newFakeDocker()) + run.runner.callTimeout = 10 * time.Second + + done := run.start(t.Context()) + run.waitStarted(t) + run.docker.closeOutput(nil) + + began := time.Now() + + run.docker.waitResult <- container.WaitResponse{StatusCode: 5} + + res := result(t, done) + if res.err != nil || res.status != 5 { + t.Fatalf("run = %d, %v; want 5, nil", res.status, res.err) + } + + if elapsed := time.Since(began); elapsed > 2*time.Second { + t.Errorf("returned after %s, want at once", elapsed) + } +} + +func TestRun_SetupErrorsStartNothing(t *testing.T) { + t.Parallel() + + cases := map[string]func(*fakeDocker, *runner){ + "self id": func(_ *fakeDocker, r *runner) { + r.selfID = func() (string, error) { return "", errInjected } + }, + "self inspect": func(f *fakeDocker, _ *runner) { + f.inspectEr[testLauncherID] = errInjected + }, + "image tag instead of id": func(f *fakeDocker, _ *runner) { + f.inspect[testLauncherID] = client.ContainerInspectResult{ + Container: container.InspectResponse{ + Image: "ghcr.io/leinardi/swarm-device-access:1", + }, + } + }, + "create": func(f *fakeDocker, _ *runner) { + f.createErr = errInjected + }, + } + + for name, setup := range cases { + t.Run(name, func(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + run := newTestRun(docker) + setup(docker, run.runner) + + res := result(t, run.start(t.Context())) + if res.err == nil || res.status != 1 { + t.Fatalf("run = %d, %v; want 1 and an error", res.status, res.err) + } + + for _, call := range []string{"attach", "start"} { + if contains(docker.recorded(), call) { + t.Errorf("%s called: %v", call, docker.recorded()) + } + } + }) + } +} + +func requireStopped(t *testing.T, docker *fakeDocker) { + t.Helper() + + stops := docker.stopCalls() + if len(stops) != 1 || stops[0].id != testDaemonID || stops[0].timeout != 10 { + t.Errorf("stops = %+v, want one of %s with timeout 10", stops, testDaemonID) + } +} diff --git a/internal/launcher/self.go b/internal/launcher/self.go new file mode 100644 index 0000000..3130320 --- /dev/null +++ b/internal/launcher/self.go @@ -0,0 +1,106 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "bufio" + "errors" + "fmt" + "io" + "os" + "regexp" + "slices" + "strings" +) + +const selfMountinfo = "/proc/self/mountinfo" + +var ( + errSelfNotFound = errors.New( + "no Docker container ID in the mount table; is the launcher running in a container?", + ) + errSelfAmbiguous = errors.New("more than one Docker container ID in the mount table") + + // containerFileRoot matches the source of the per-container files Docker + // bind-mounts into every container, whatever its data-root + // (/var/lib/docker, /mnt/.ix-apps/docker, ...) or hostname. + containerFileRoot = regexp.MustCompile( + `/containers/([0-9a-f]{64})/(?:hostname|hosts|resolv\.conf)$`, + ) + + // containerFileTargets are where those files are mounted. + containerFileTargets = []string{ + "/etc/hostname", + "/etc/hosts", + "/etc/resolv.conf", + } +) + +// mountinfoFields are the fields of a /proc//mountinfo line up to the +// mount point: mount ID, parent ID, major:minor, root, mount point. +const mountinfoFields = 5 + +// selfIDFromMountinfo returns the ID of the container whose mount table +// this is. +func selfIDFromMountinfo() (string, error) { + file, err := os.Open(selfMountinfo) + if err != nil { + return "", fmt.Errorf("open %s: %w", selfMountinfo, err) + } + defer file.Close() + + return parseSelfID(file) +} + +// parseSelfID finds the container ID in the root of the mounts at /etc/hostname, +// /etc/hosts and /etc/resolv.conf. No match, or two different IDs, is an +// error: the launcher must never guess which image it runs. +func parseSelfID(reader io.Reader) (string, error) { + var found string + + scanner := bufio.NewScanner(reader) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < mountinfoFields || !slices.Contains(containerFileTargets, fields[4]) { + continue + } + + match := containerFileRoot.FindStringSubmatch(fields[3]) + if match == nil { + continue + } + + if found != "" && found != match[1] { + return "", errSelfAmbiguous + } + + found = match[1] + } + + err := scanner.Err() + if err != nil { + return "", fmt.Errorf("read mount table: %w", err) + } + + if found == "" { + return "", errSelfNotFound + } + + return found, nil +} diff --git a/internal/launcher/spec.go b/internal/launcher/spec.go new file mode 100644 index 0000000..4b0ec51 --- /dev/null +++ b/internal/launcher/spec.go @@ -0,0 +1,141 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "slices" + "time" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/mount" + "github.com/moby/moby/client" +) + +const ( + // DaemonName is the name of the daemon container. One name per node is + // what lets the next launcher find, and replace, a daemon its + // predecessor left behind. + DaemonName = "swarm-device-access" + + // LabelRole marks the containers a launcher creates. Both labels sit + // outside the swarm-device-access. prefix, where the daemon would report + // them as unknown policy labels. + LabelRole = "io.github.leinardi.swarm-device-access.role" + // LabelLauncher holds the ID of the launcher container that owns the + // daemon. + LabelLauncher = "io.github.leinardi.swarm-device-access.launcher" + // RoleDaemon is the LabelRole value of a daemon container. + RoleDaemon = "daemon" + + // DefaultHostDockerSocket is the host path of the Docker socket bound + // into the daemon. + DefaultHostDockerSocket = "/var/run/docker.sock" + + // containerDockerSocket is where both the launcher and the daemon find + // the Docker socket: the daemon's -docker-socket default. + containerDockerSocket = "/var/run/docker.sock" + hostSysfs = "/sys" + containerSysfs = "/host/sys" + devPath = "/dev" + hostDBusSocket = "/run/dbus/system_bus_socket" + // dhi.io/static has no /var/run -> /run symlink, so the socket must be + // mounted under /var/run inside the container. + containerDBusSocket = "/var/run/dbus/system_bus_socket" + // ContainerConfigDir is where -config-dir is mounted in the daemon. + ContainerConfigDir = "/etc/swarm-device-access" + + // hostMode shares a namespace (pid, user, network) with the host. + hostMode = "host" + networkNone = "none" + + // stopTimeout is how long Docker waits after SIGTERM before it kills + // the daemon. + stopTimeout = 10 * time.Second +) + +// Options describe the daemon container a launcher creates. +type Options struct { + // HostDockerSocket is the host path of the Docker socket bound into the + // daemon. The launcher itself always uses its own /var/run/docker.sock. + HostDockerSocket string + // DBus binds the host's system bus socket, which enables the systemd + // reload watcher. + DBus bool + // ConfigDir, when set, is a host directory bound read-only at + // ContainerConfigDir. + ConfigDir string + // HostNetwork runs the daemon in the host network namespace instead of + // none; only the metrics and debug servers need a network. + HostNetwork bool + // DaemonArgs are passed to the daemon verbatim; it validates them. + DaemonArgs []string +} + +// daemonSpec is the complete daemon container: the privileged set the +// daemon needs and nothing else. Bind sources are given as Mounts, not +// Binds, so a missing host path fails the create instead of Docker creating +// an empty directory in its place. +func daemonSpec(opts *Options, imageID, launcherID string) client.ContainerCreateOptions { + stopSeconds := int(stopTimeout / time.Second) + + mounts := []mount.Mount{ + bind(opts.HostDockerSocket, containerDockerSocket, false), + bind(hostSysfs, containerSysfs, false), + bind(devPath, devPath, false), + } + + if opts.DBus { + mounts = append(mounts, bind(hostDBusSocket, containerDBusSocket, false)) + } + + if opts.ConfigDir != "" { + mounts = append(mounts, bind(opts.ConfigDir, ContainerConfigDir, true)) + } + + network := container.NetworkMode(networkNone) + if opts.HostNetwork { + network = hostMode + } + + return client.ContainerCreateOptions{ + Name: DaemonName, + Config: &container.Config{ + Image: imageID, + Cmd: slices.Clone(opts.DaemonArgs), + Labels: map[string]string{ + LabelRole: RoleDaemon, + LabelLauncher: launcherID, + }, + StopTimeout: &stopSeconds, + }, + HostConfig: &container.HostConfig{ + Privileged: true, + CgroupnsMode: container.CgroupnsModeHost, + PidMode: hostMode, + UsernsMode: hostMode, + NetworkMode: network, + AutoRemove: true, + Mounts: mounts, + }, + } +} + +func bind(source, target string, readOnly bool) mount.Mount { + return mount.Mount{Type: mount.TypeBind, Source: source, Target: target, ReadOnly: readOnly} +} diff --git a/internal/launcher/spec_test.go b/internal/launcher/spec_test.go new file mode 100644 index 0000000..5ea9488 --- /dev/null +++ b/internal/launcher/spec_test.go @@ -0,0 +1,225 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "errors" + "maps" + "reflect" + "slices" + "strings" + "testing" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/mount" +) + +func TestDaemonSpec(t *testing.T) { + t.Parallel() + + socket := mount.Mount{ + Type: mount.TypeBind, + Source: "/var/run/docker.sock", + Target: "/var/run/docker.sock", + } + sysfs := mount.Mount{Type: mount.TypeBind, Source: "/sys", Target: "/host/sys"} + dev := mount.Mount{Type: mount.TypeBind, Source: "/dev", Target: "/dev"} + dbus := mount.Mount{ + Type: mount.TypeBind, + Source: "/run/dbus/system_bus_socket", + Target: "/var/run/dbus/system_bus_socket", + } + + cases := []struct { + name string + opts Options + wantMounts []mount.Mount + wantNetwork container.NetworkMode + }{ + { + name: "base", + opts: Options{HostDockerSocket: DefaultHostDockerSocket}, + wantMounts: []mount.Mount{socket, sysfs, dev}, + wantNetwork: "none", + }, + { + name: "dbus", + opts: Options{HostDockerSocket: DefaultHostDockerSocket, DBus: true}, + wantMounts: []mount.Mount{socket, sysfs, dev, dbus}, + wantNetwork: "none", + }, + { + name: "config dir", + opts: Options{HostDockerSocket: DefaultHostDockerSocket, ConfigDir: "/srv/sda"}, + wantMounts: []mount.Mount{socket, sysfs, dev, { + Type: mount.TypeBind, + Source: "/srv/sda", + Target: "/etc/swarm-device-access", + ReadOnly: true, + }}, + wantNetwork: "none", + }, + { + name: "host network", + opts: Options{HostDockerSocket: DefaultHostDockerSocket, HostNetwork: true}, + wantMounts: []mount.Mount{socket, sysfs, dev}, + wantNetwork: "host", + }, + { + name: "custom host docker socket", + opts: Options{HostDockerSocket: "/run/user/0/docker.sock"}, + wantMounts: []mount.Mount{ + { + Type: mount.TypeBind, + Source: "/run/user/0/docker.sock", + Target: "/var/run/docker.sock", + }, + sysfs, + dev, + }, + wantNetwork: "none", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + spec := daemonSpec(&tc.opts, testImageID, testLauncherID) + + if spec.Name != "swarm-device-access" || spec.Image != "" { + t.Errorf("name %q, image shortcut %q", spec.Name, spec.Image) + } + + stopSeconds := 10 + wantHost := &container.HostConfig{ + Privileged: true, + CgroupnsMode: container.CgroupnsModeHost, + PidMode: "host", + UsernsMode: "host", + NetworkMode: tc.wantNetwork, + AutoRemove: true, + Mounts: tc.wantMounts, + } + + if !reflect.DeepEqual(spec.HostConfig, wantHost) { + t.Errorf("host config\n got %+v\nwant %+v", spec.HostConfig, wantHost) + } + + wantConfig := &container.Config{ + Image: testImageID, + Cmd: nil, + Labels: map[string]string{ + "io.github.leinardi.swarm-device-access.role": "daemon", + "io.github.leinardi.swarm-device-access.launcher": testLauncherID, + }, + StopTimeout: &stopSeconds, + } + + if !reflect.DeepEqual(spec.Config, wantConfig) { + t.Errorf("config\n got %+v\nwant %+v", spec.Config, wantConfig) + } + }) + } +} + +func TestDaemonSpec_ArgsPassedVerbatim(t *testing.T) { + t.Parallel() + + args := []string{"-dry-run", "-device-allow", "/dev/sd*", "--", "-weird"} + opts := Options{HostDockerSocket: DefaultHostDockerSocket, DaemonArgs: args} + + spec := daemonSpec(&opts, testImageID, testLauncherID) + if !slices.Equal(spec.Config.Cmd, args) { + t.Errorf("cmd = %v, want %v", spec.Config.Cmd, args) + } + + args[0] = "changed" + + if spec.Config.Cmd[0] != "-dry-run" { + t.Error("spec aliases the caller's argument slice") + } +} + +// The launcher's labels must not sit under the swarm-device-access. prefix, +// where the daemon reports unknown keys as bad policy labels. +func TestDaemonSpec_LabelsOutsidePolicyPrefix(t *testing.T) { + t.Parallel() + + opts := Options{HostDockerSocket: DefaultHostDockerSocket} + spec := daemonSpec(&opts, testImageID, testLauncherID) + + for _, key := range slices.Sorted(maps.Keys(spec.Config.Labels)) { + if strings.HasPrefix(key, "swarm-device-access.") { + t.Errorf("label %q is under the policy label prefix", key) + } + } +} + +func TestParseSelfID(t *testing.T) { + t.Parallel() + + const ( + dockerRoot = "1570 1511 0:420 / / rw,relatime - overlay root rw\n" + + "1581 1570 8:2 /var/lib/docker/containers/" + testLauncherID + "/resolv.conf " + + "/etc/resolv.conf rw,relatime - ext4 /dev/sda2 rw\n" + + "1582 1570 8:2 /var/lib/docker/containers/" + testLauncherID + "/hostname " + + "/etc/hostname rw,relatime - ext4 /dev/sda2 rw\n" + + "1583 1570 8:2 /var/lib/docker/containers/" + testLauncherID + "/hosts " + + "/etc/hosts rw,relatime - ext4 /dev/sda2 rw\n" + trueNAS = "901 800 0:61 /ix-apps/docker/containers/" + testLauncherID + "/hostname " + + "/etc/hostname rw,relatime - zfs boot-pool/.ix-apps rw\n" + separateFS = "901 800 0:61 /docker/containers/" + testLauncherID + "/hosts " + + "/etc/hosts rw - ext4 /dev/sdb1 rw\n" + noMatch = "1570 1511 0:420 / / rw,relatime - overlay root rw\n" + + "1581 1570 8:2 /etc/resolv.conf /etc/resolv.conf rw - ext4 /dev/sda2 rw\n" + // Another container's file mounted elsewhere says nothing about us. + otherTarget = "1581 1570 8:2 /var/lib/docker/containers/" + testDaemonID + "/hostname " + + "/mnt/hostname rw - ext4 /dev/sda2 rw\n" + ambiguous = "1582 1570 8:2 /var/lib/docker/containers/" + testLauncherID + "/hostname " + + "/etc/hostname rw - ext4 /dev/sda2 rw\n" + + "1583 1570 8:2 /var/lib/docker/containers/" + testDaemonID + "/hosts " + + "/etc/hosts rw - ext4 /dev/sda2 rw\n" + ) + + cases := []struct { + name string + table string + want string + wantErr error + }{ + {name: "docker data-root", table: dockerRoot, want: testLauncherID}, + {name: "truenas data-root", table: trueNAS, want: testLauncherID}, + {name: "data-root on its own filesystem", table: separateFS, want: testLauncherID}, + {name: "no match", table: noMatch, wantErr: errSelfNotFound}, + {name: "other target", table: otherTarget, wantErr: errSelfNotFound}, + {name: "ambiguous", table: ambiguous, wantErr: errSelfAmbiguous}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + got, err := parseSelfID(strings.NewReader(tc.table)) + if !errors.Is(err, tc.wantErr) || got != tc.want { + t.Errorf("parseSelfID = %q, %v; want %q, %v", got, err, tc.want, tc.wantErr) + } + }) + } +} diff --git a/internal/launcher/stale.go b/internal/launcher/stale.go new file mode 100644 index 0000000..11dc0d9 --- /dev/null +++ b/internal/launcher/stale.go @@ -0,0 +1,163 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "context" + "errors" + "fmt" + + cerrdefs "github.com/containerd/errdefs" + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" +) + +const ( + // labelImageTitle identifies a daemon started by the sh wrapper of + // earlier releases, which recorded no owner. + labelImageTitle = "org.opencontainers.image.title" + imageTitle = "swarm-device-access" +) + +var ( + errOwnedByRunningLauncher = errors.New("daemon owned by running launcher") + errNoOwner = errors.New("daemon container has no launcher label") + errOwnerState = errors.New("inspect reported no state") + errForeignContainer = errors.New("container is not a swarm-device-access daemon") +) + +// removeStale removes the daemon containers a previous launcher left behind, +// so the name is free for this one. A daemon is removed only when it +// provably has no live owner; any doubt is an error and leaves it alone. +func (r *runner) removeStale(ctx context.Context, selfID string) error { + listCtx, cancel := context.WithTimeout(ctx, r.callTimeout) + defer cancel() + + list, err := r.docker.ContainerList(listCtx, client.ContainerListOptions{ + All: true, + Filters: make(client.Filters).Add("name", "^/"+DaemonName+"$"), + }) + if err != nil { + return fmt.Errorf("list %s containers: %w", DaemonName, err) + } + + for i := range list.Items { + stale := &list.Items[i] + + err = r.checkStale(ctx, stale, selfID) + if err != nil { + return err + } + + err = r.removeContainer(ctx, stale.ID) + if err != nil { + return fmt.Errorf("remove stale daemon %s: %w", shortID(stale.ID), err) + } + } + + return nil +} + +// checkStale returns nil when stale may be removed: a daemon whose launcher +// is confirmed gone or stopped, one owned by this very launcher container +// (a restart of it), or one started by the old sh wrapper. +func (r *runner) checkStale(ctx context.Context, stale *container.Summary, selfID string) error { + if stale.Labels[LabelRole] != RoleDaemon { + if stale.Labels[labelImageTitle] == imageTitle { + launcherLog().Info("removing daemon left by the sh wrapper", + "id", shortID(stale.ID)) + + return nil + } + + return fmt.Errorf("%s (%s): %w", DaemonName, shortID(stale.ID), errForeignContainer) + } + + owner := stale.Labels[LabelLauncher] + if owner == "" { + return fmt.Errorf("%s (%s): %w", DaemonName, shortID(stale.ID), errNoOwner) + } + + if owner == selfID { + launcherLog().Info("removing daemon left by an earlier run of this launcher", + "id", shortID(stale.ID)) + + return nil + } + + inspectCtx, cancel := context.WithTimeout(ctx, r.callTimeout) + defer cancel() + + inspected, err := r.docker.ContainerInspect(inspectCtx, owner, client.ContainerInspectOptions{}) + + switch { + case cerrdefs.IsNotFound(err): + case err != nil: + // A timeout or an engine error says nothing about whether the owner + // still runs; only a confirmed absence makes the daemon an orphan. + return fmt.Errorf( + "inspect launcher %s of daemon %s: %w", + shortID(owner), + shortID(stale.ID), + err, + ) + case inspected.Container.State == nil: + return fmt.Errorf( + "launcher %s of daemon %s: %w", + shortID(owner), + shortID(stale.ID), + errOwnerState, + ) + case inspected.Container.State.Running: + return fmt.Errorf("%w %s", errOwnedByRunningLauncher, shortID(owner)) + } + + launcherLog().Info("removing orphaned daemon", + "id", shortID(stale.ID), "launcher", shortID(owner)) + + return nil +} + +// removeContainer force-removes containerID. It does not depend on ctx being live, +// so a failed setup can still clean up after a signal. +func (r *runner) removeContainer(ctx context.Context, containerID string) error { + removeCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), r.callTimeout) + defer cancel() + + _, err := r.docker.ContainerRemove( + removeCtx, + containerID, + client.ContainerRemoveOptions{Force: true}, + ) + if err != nil && !cerrdefs.IsNotFound(err) { + return fmt.Errorf("remove container %s: %w", shortID(containerID), err) + } + + return nil +} + +func shortID(fullID string) string { + const shortLen = 12 + + if len(fullID) > shortLen { + return fullID[:shortLen] + } + + return fullID +} diff --git a/internal/launcher/stale_test.go b/internal/launcher/stale_test.go new file mode 100644 index 0000000..d81f36f --- /dev/null +++ b/internal/launcher/stale_test.go @@ -0,0 +1,163 @@ +//go:build linux + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package launcher + +import ( + "context" + "errors" + "slices" + "testing" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/client" +) + +const ( + staleID = "4444444444444444444444444444444444444444444444444444444444444444" + ownerID = "5555555555555555555555555555555555555555555555555555555555555555" +) + +func TestRemoveStale(t *testing.T) { + t.Parallel() + + daemonLabels := func(owner string) map[string]string { + return map[string]string{LabelRole: RoleDaemon, LabelLauncher: owner} + } + + owner := func(running bool) client.ContainerInspectResult { + return client.ContainerInspectResult{Container: container.InspectResponse{ + ID: ownerID, + State: &container.State{Running: running}, + }} + } + + cases := []struct { + name string + labels map[string]string + owner *client.ContainerInspectResult + ownerErr error + wantErr error + wantRemoved bool + }{ + {name: "launcher gone", labels: daemonLabels(ownerID), wantRemoved: true}, + { + name: "launcher exited", + labels: daemonLabels(ownerID), + owner: new(owner(false)), + wantRemoved: true, + }, + { + name: "launcher running", + labels: daemonLabels(ownerID), + owner: new(owner(true)), + wantErr: errOwnedByRunningLauncher, + }, + { + name: "launcher inspect times out", + labels: daemonLabels(ownerID), + ownerErr: context.DeadlineExceeded, + wantErr: context.DeadlineExceeded, + }, + { + name: "launcher state unknown", + labels: daemonLabels(ownerID), + owner: &client.ContainerInspectResult{}, + wantErr: errOwnerState, + }, + { + name: "earlier run of this launcher", + labels: daemonLabels(testLauncherID), + wantRemoved: true, + }, + { + name: "no owner label", + labels: map[string]string{LabelRole: RoleDaemon}, + wantErr: errNoOwner, + }, + { + name: "sh wrapper daemon", + labels: map[string]string{"org.opencontainers.image.title": "swarm-device-access"}, + wantRemoved: true, + }, + { + name: "foreign container", + labels: map[string]string{"org.opencontainers.image.title": "something-else"}, + wantErr: errForeignContainer, + }, + {name: "unlabelled container", wantErr: errForeignContainer}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + docker.list = []container.Summary{{ID: staleID, Labels: tc.labels}} + + if tc.owner != nil { + docker.inspect[ownerID] = *tc.owner + } + + if tc.ownerErr != nil { + docker.inspectEr[ownerID] = tc.ownerErr + } + + run := newTestRun(docker) + + err := run.runner.removeStale(t.Context(), testLauncherID) + if !errors.Is(err, tc.wantErr) { + t.Fatalf("removeStale = %v, want %v", err, tc.wantErr) + } + + var wantRemoved []string + if tc.wantRemoved { + wantRemoved = []string{staleID} + } + + if got := docker.removedIDs(); !slices.Equal(got, wantRemoved) { + t.Errorf("removed %v, want %v", got, wantRemoved) + } + }) + } +} + +// A daemon owned by a live launcher stops the whole run before anything is +// created. +func TestRun_RefusesLiveDaemon(t *testing.T) { + t.Parallel() + + docker := newFakeDocker() + docker.list = []container.Summary{{ + ID: staleID, + Labels: map[string]string{LabelRole: RoleDaemon, LabelLauncher: ownerID}, + }} + docker.inspect[ownerID] = client.ContainerInspectResult{Container: container.InspectResponse{ + ID: ownerID, + State: &container.State{Running: true}, + }} + + res := result(t, newTestRun(docker).start(t.Context())) + if !errors.Is(res.err, errOwnedByRunningLauncher) || res.status != 1 { + t.Fatalf("run = %d, %v; want 1, %v", res.status, res.err, errOwnedByRunningLauncher) + } + + if contains(docker.recorded(), "create") || len(docker.removedIDs()) > 0 { + t.Errorf("calls %v, want no create and no remove", docker.recorded()) + } +} diff --git a/test/integration/enforce_test.go b/test/integration/enforce_test.go index a5b80c4..5f8f345 100644 --- a/test/integration/enforce_test.go +++ b/test/integration/enforce_test.go @@ -454,13 +454,37 @@ func startDaemonContainer( }, }) - logs, err := cli.ContainerLogs(ctx, daemonID, dockerclient.ContainerLogsOptions{ + proc := followLogs(ctx, t, cli, daemonID) + + // Registered after startContainer's removal, so it runs first: the log + // stream ends once the container is gone, and collect must be finished + // before the test completes. + t.Cleanup(func() { + stopDaemonContainer(context.WithoutCancel(ctx), t, cli, daemonID, proc) + }) + + proc.waitReady(ctx, t) + + return daemonID, proc +} + +// followLogs collects the JSON log records of containerID until its log +// stream ends. +func followLogs( + ctx context.Context, + t *testing.T, + cli *dockerclient.Client, + containerID string, +) *daemonProc { + t.Helper() + + logs, err := cli.ContainerLogs(ctx, containerID, dockerclient.ContainerLogsOptions{ ShowStdout: true, ShowStderr: true, Follow: true, }) if err != nil { - t.Fatalf("follow daemon container logs: %v", err) + t.Fatalf("follow container %s logs: %v", shortID(containerID), err) } proc := newDaemonProc() @@ -474,16 +498,7 @@ func startDaemonContainer( go proc.collect(t.Log, reader) - // Registered after startContainer's removal, so it runs first: the log - // stream ends once the container is gone, and collect must be finished - // before the test completes. - t.Cleanup(func() { - stopDaemonContainer(context.WithoutCancel(ctx), t, cli, daemonID, proc) - }) - - proc.waitReady(ctx, t) - - return daemonID, proc + return proc } // stopDaemonContainer stops the daemon and waits until its log stream ends. diff --git a/test/integration/launcher_test.go b/test/integration/launcher_test.go new file mode 100644 index 0000000..27de030 --- /dev/null +++ b/test/integration/launcher_test.go @@ -0,0 +1,278 @@ +//go:build integration + +/* + * Copyright 2026 Roberto Leinardi. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package integration + +import ( + "context" + "slices" + "testing" + "time" + + "github.com/moby/moby/api/types/container" + "github.com/moby/moby/api/types/mount" + dockerclient "github.com/moby/moby/client" + + "github.com/leinardi/swarm-device-access/internal/launcher" +) + +// launcherStopTimeoutSec matches the compose files' stop_grace_period: the +// launcher needs up to its shutdown budget to stop the daemon. +const launcherStopTimeoutSec = 30 + +// TestLauncher_RunsDaemonFromItsOwnImage starts the image in launcher mode +// with only the Docker socket, and checks the daemon container it creates, +// a graceful stop, and the replacement of a daemon orphaned by a killed +// launcher. +// +// It needs the daemon image and creates a privileged container named +// swarm-device-access, so it runs with the enforcement tests. +func TestLauncher_RunsDaemonFromItsOwnImage(t *testing.T) { + if !enforceRequested() { + t.Skipf( + "set %s=1 to run: it creates a privileged swarm-device-access container on this host", + envEnforce, + ) + } + + cli := requireDocker(t) + ctx := testCtx(t) + image := requireDaemonImage(ctx, t, cli) + + if daemons := listDaemons(ctx, t, cli); len(daemons) > 0 { + t.Fatalf( + "a %s container already exists (%s); remove it first", + launcher.DaemonName, + shortID(daemons[0].ID), + ) + } + + t.Cleanup(func() { removeTestDaemons(ctx, t, cli) }) + + t.Run("spec and graceful stop", func(t *testing.T) { + launcherID := startLauncher(ctx, t, cli, image) + requireDaemonSpec(ctx, t, cli, launcherID) + + timeout := launcherStopTimeoutSec + + _, err := cli.ContainerStop( + ctx, + launcherID, + dockerclient.ContainerStopOptions{Timeout: &timeout}, + ) + if err != nil { + t.Fatalf("stop launcher: %v", err) + } + + status := waitExit(ctx, t, cli, launcherID) + if status != 0 { + t.Errorf("launcher exit status %d, want 0", status) + } + + if daemons := listDaemons(ctx, t, cli); len(daemons) > 0 { + t.Errorf("daemon %s still exists after its launcher stopped", shortID(daemons[0].ID)) + } + }) + + t.Run("orphan replaced", func(t *testing.T) { + first := startLauncher(ctx, t, cli, image) + + _, err := cli.ContainerKill(ctx, first, dockerclient.ContainerKillOptions{Signal: "KILL"}) + if err != nil { + t.Fatalf("kill launcher: %v", err) + } + + waitExit(ctx, t, cli, first) + + second := startLauncher(ctx, t, cli, image) + + daemons := listDaemons(ctx, t, cli) + if len(daemons) != 1 || daemons[0].Labels[launcher.LabelLauncher] != second { + t.Fatalf( + "daemons %+v, want exactly one, owned by launcher %s", + daemons, + shortID(second), + ) + } + }) +} + +// startLauncher runs the image in launcher mode with only the Docker socket +// mounted and a dry-run daemon behind it, and waits until the daemon's +// startup, as seen in the launcher's own output, is complete. +func startLauncher( + ctx context.Context, + t *testing.T, + cli *dockerclient.Client, + image string, +) string { + t.Helper() + + launcherID := startContainer(ctx, t, cli, + &container.Config{ + Image: image, + Cmd: []string{ + "launch", "-log-format=json", + "--", "-dry-run", "-log-format=json", "-log-level=debug", + }, + }, + &container.HostConfig{Binds: []string{hostDockerSocket + ":/var/run/docker.sock"}}, + ) + + followLogs(ctx, t, cli, launcherID).waitReady(ctx, t) + + return launcherID +} + +// requireDaemonSpec checks the daemon container against the documented +// privileged set: the launcher's image, host namespaces, no network, and +// exactly the three bind mounts, given as Mounts. +func requireDaemonSpec( + ctx context.Context, + t *testing.T, + cli *dockerclient.Client, + launcherID string, +) { + t.Helper() + + self, err := cli.ContainerInspect(ctx, launcherID, dockerclient.ContainerInspectOptions{}) + if err != nil { + t.Fatalf("inspect launcher: %v", err) + } + + res, err := cli.ContainerInspect( + ctx, + launcher.DaemonName, + dockerclient.ContainerInspectOptions{}, + ) + if err != nil { + t.Fatalf("inspect daemon: %v", err) + } + + daemon := res.Container + host := daemon.HostConfig + + if daemon.Image != self.Container.Image { + t.Errorf("daemon image %s, want the launcher's %s", daemon.Image, self.Container.Image) + } + + if daemon.Config.Labels[launcher.LabelLauncher] != launcherID { + t.Errorf( + "daemon launcher label %q, want %s", + daemon.Config.Labels[launcher.LabelLauncher], + launcherID, + ) + } + + type hostSpec struct { + privileged bool + cgroupns, pid, userns, net string + autoRemove bool + } + + gotHost := hostSpec{ + privileged: host.Privileged, + cgroupns: string(host.CgroupnsMode), + pid: string(host.PidMode), + userns: string(host.UsernsMode), + net: string(host.NetworkMode), + autoRemove: host.AutoRemove, + } + + wantHost := hostSpec{true, "host", "host", "host", "none", true} + if gotHost != wantHost { + t.Errorf("host config %+v, want %+v", gotHost, wantHost) + } + + if len(host.Binds) > 0 { + t.Errorf("binds %v, want Mounts only", host.Binds) + } + + want := []string{"/var/run/docker.sock:/var/run/docker.sock", "/sys:/host/sys", "/dev:/dev"} + + got := make([]string, 0, len(host.Mounts)) + + for _, mnt := range host.Mounts { + if mnt.Type != mount.TypeBind { + t.Errorf("mount %s is %s, want bind", mnt.Target, mnt.Type) + } + + got = append(got, mnt.Source+":"+mnt.Target) + } + + if !slices.Equal(got, want) { + t.Errorf("mounts %v, want %v", got, want) + } +} + +// listDaemons returns every container named swarm-device-access. +func listDaemons(ctx context.Context, t *testing.T, cli *dockerclient.Client) []container.Summary { + t.Helper() + + res, err := cli.ContainerList(ctx, dockerclient.ContainerListOptions{ + All: true, + Filters: make(dockerclient.Filters).Add("name", "^/"+launcher.DaemonName+"$"), + }) + if err != nil { + t.Fatalf("list daemon containers: %v", err) + } + + return res.Items +} + +// waitExit waits until containerID has stopped and returns its exit status. +func waitExit( + ctx context.Context, + t *testing.T, + cli *dockerclient.Client, + containerID string, +) int64 { + t.Helper() + + waitCtx, cancel := context.WithTimeout(ctx, launcherStopTimeoutSec*time.Second+cleanupTimeout) + defer cancel() + + res := cli.ContainerWait(waitCtx, containerID, dockerclient.ContainerWaitOptions{ + Condition: container.WaitConditionNotRunning, + }) + + select { + case exit := <-res.Result: + return exit.StatusCode + case err := <-res.Error: + t.Fatalf("wait for %s: %v", shortID(containerID), err) + } + + return 0 +} + +// removeTestDaemons removes a daemon a failed test left behind. Its +// launchers are this run's containers, removed by their own cleanups. +func removeTestDaemons(ctx context.Context, t *testing.T, cli *dockerclient.Client) { + t.Helper() + + listCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cleanupTimeout) + defer cancel() + + daemons := listDaemons(listCtx, t, cli) + for i := range daemons { + if daemons[i].Labels[launcher.LabelRole] == launcher.RoleDaemon { + removeContainer(ctx, t, cli, daemons[i].ID) + } + } +}