From e313365686c36140109f39984546f3a60f94cf18 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Mon, 12 Oct 2026 18:45:44 +0200 Subject: [PATCH] feat(agent): job runner, compose override executor and swarm executor (#7) --- go.mod | 14 +- go.sum | 20 -- internal/agent/updater/compose.go | 145 ++++++++--- internal/agent/updater/compose_test.go | 132 ++++++++++ internal/agent/updater/dispatch.go | 28 ++ internal/agent/updater/dispatch_test.go | 31 +++ internal/agent/updater/override.go | 192 ++++++++++++++ internal/agent/updater/override_test.go | 199 ++++++++++++++ internal/agent/updater/pathcheck.go | 2 +- internal/agent/updater/pathcheck_test.go | 6 +- internal/agent/updater/registryauth.go | 106 ++++++++ internal/agent/updater/registryauth_test.go | 72 ++++++ internal/agent/updater/resolve.go | 49 ++++ internal/agent/updater/resolve_test.go | 38 +++ internal/agent/updater/runner.go | 272 ++++++++++++++++++++ internal/agent/updater/runner_test.go | 200 ++++++++++++++ internal/agent/updater/swarm.go | 87 +++++-- internal/agent/updater/swarm_test.go | 189 ++++++++------ 18 files changed, 1617 insertions(+), 165 deletions(-) create mode 100644 internal/agent/updater/compose_test.go create mode 100644 internal/agent/updater/dispatch.go create mode 100644 internal/agent/updater/dispatch_test.go create mode 100644 internal/agent/updater/override.go create mode 100644 internal/agent/updater/override_test.go create mode 100644 internal/agent/updater/registryauth.go create mode 100644 internal/agent/updater/registryauth_test.go create mode 100644 internal/agent/updater/resolve.go create mode 100644 internal/agent/updater/resolve_test.go create mode 100644 internal/agent/updater/runner.go create mode 100644 internal/agent/updater/runner_test.go diff --git a/go.mod b/go.mod index 8ac9846..74e3aa7 100644 --- a/go.mod +++ b/go.mod @@ -5,23 +5,22 @@ go 1.26.3 require ( github.com/coder/websocket v1.8.15 github.com/coreos/go-oidc/v3 v3.21.0 + github.com/distribution/reference v0.6.0 github.com/docker/docker v28.5.2+incompatible github.com/jackc/pgx/v5 v5.10.0 - github.com/prometheus/client_golang v1.23.2 - github.com/prometheus/client_model v0.6.2 + github.com/opencontainers/go-digest v1.0.0 github.com/stretchr/testify v1.11.1 golang.org/x/oauth2 v0.36.0 + gopkg.in/yaml.v3 v3.0.1 ) require ( github.com/Microsoft/go-winio v0.6.2 // indirect - github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/log v0.1.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect - github.com/distribution/reference v0.6.0 // indirect github.com/docker/go-connections v0.7.0 // indirect github.com/docker/go-units v0.5.0 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect @@ -35,25 +34,18 @@ require ( github.com/moby/sys/atomicwriter v0.1.0 // indirect github.com/moby/term v0.5.2 // indirect github.com/morikuni/aec v1.1.0 // indirect - github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect - github.com/opencontainers/go-digest v1.0.0 // indirect github.com/opencontainers/image-spec v1.1.1 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect - github.com/prometheus/common v0.66.1 // indirect - github.com/prometheus/procfs v0.16.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect go.opentelemetry.io/otel v1.43.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect go.opentelemetry.io/otel/metric v1.43.0 // indirect go.opentelemetry.io/otel/trace v1.43.0 // indirect - go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/sync v0.20.0 // indirect golang.org/x/sys v0.42.0 // indirect golang.org/x/text v0.35.0 // indirect golang.org/x/time v0.15.0 // indirect - google.golang.org/protobuf v1.36.11 // indirect - gopkg.in/yaml.v3 v3.0.1 // indirect gotest.tools/v3 v3.5.2 // indirect ) diff --git a/go.sum b/go.sum index f933059..7d4ff22 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,6 @@ github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEK github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E= github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY= github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU= -github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= -github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= @@ -52,14 +50,10 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0= github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4= github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo= github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4= -github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= -github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= -github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw= @@ -70,8 +64,6 @@ github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ= github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc= github.com/morikuni/aec v1.1.0 h1:vBBl0pUnvi/Je71dsRrhMBtreIqNMYErSAbEeb8jrXQ= github.com/morikuni/aec v1.1.0/go.mod h1:xDRgiq/iw5l+zkao76YTKzKttOp2cwPEne25HDkJnBw= -github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= -github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM= github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040= @@ -80,14 +72,6 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= -github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= -github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= -github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= -github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= -github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= -github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= -github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= -github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= @@ -117,10 +101,6 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= -go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= -go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= -go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= -go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= diff --git a/internal/agent/updater/compose.go b/internal/agent/updater/compose.go index c5ed2f1..b3251b5 100644 --- a/internal/agent/updater/compose.go +++ b/internal/agent/updater/compose.go @@ -3,52 +3,129 @@ package updater import ( "context" "fmt" + "io" "os/exec" "strings" + "time" - "github.com/shcizo/package-updater/internal/discovery" + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" + + "github.com/shcizo/package-updater/internal/protocol" ) -// ComposeExecutor invokes `docker compose` as a subprocess. -type ComposeExecutor struct{} +const ( + labelComposeProject = "com.docker.compose.project" + labelComposeService = "com.docker.compose.service" + labelWorkingDir = "com.docker.compose.project.working_dir" + labelConfigFiles = "com.docker.compose.project.config_files" +) -// NewComposeExecutor returns an executor that shells out to docker compose. -func NewComposeExecutor() *ComposeExecutor { - return &ComposeExecutor{} +// errorOutputTail is how much of a failed command's output goes into the +// job's error message; the full output is in the job log. +const errorOutputTail = 2 << 10 + +// tailWriter adapts a tailBuffer to io.Writer. +type tailWriter struct{ b *tailBuffer } + +func (w tailWriter) Write(p []byte) (int, error) { + w.b.Write(p) + return len(p), nil } -// Execute runs `docker compose -f ... -p pull ` -// followed by `... up -d `. Working directory is set to -// job.WorkingDir so any relative paths in the compose file resolve correctly. -func (e *ComposeExecutor) Execute(ctx context.Context, job discovery.Job) error { - if job.Refused { - return fmt.Errorf("refused: %s", job.RefusedReason) - } - if err := e.run(ctx, job, "pull"); err != nil { - return fmt.Errorf("pull: %w", err) - } - if err := e.run(ctx, job, "up", "-d"); err != nil { - return fmt.Errorf("up: %w", err) - } - return nil +// CommandRunner runs "docker " in dir with stdout and stderr to out. +type CommandRunner func(ctx context.Context, dir string, out io.Writer, args ...string) error + +// ExecDocker is the real CommandRunner. When ctx ends the process is killed +// and, after a grace period, its output pipes are closed. +func ExecDocker(ctx context.Context, dir string, out io.Writer, args ...string) error { + cmd := exec.CommandContext(ctx, "docker", args...) + cmd.Dir = dir + cmd.Stdout = out + cmd.Stderr = out + cmd.WaitDelay = 10 * time.Second + return cmd.Run() } -func (e *ComposeExecutor) run(ctx context.Context, job discovery.Job, args ...string) error { - cliArgs := []string{"compose"} - for _, f := range job.ConfigFiles { - cliArgs = append(cliArgs, "-f", f) - } - cliArgs = append(cliArgs, "-p", job.Project) - cliArgs = append(cliArgs, args...) - cliArgs = append(cliArgs, job.Service) +// ComposeDocker is the subset of the Docker SDK ComposeExecutor needs. +type ComposeDocker interface { + SelfInspector + ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error) +} - cmd := exec.CommandContext(ctx, "docker", cliArgs...) - cmd.Dir = job.WorkingDir +// ComposeExecutor updates a Compose service by writing the agent-owned +// override file and running `docker compose pull` and `up -d`. +type ComposeExecutor struct { + cli ComposeDocker + resolve *Resolver + selfID string + run CommandRunner +} - out, err := cmd.CombinedOutput() +// NewComposeExecutor returns an executor for Compose jobs. selfID names the +// agent's own container, whose mounts decide which directories may be +// touched; run is ExecDocker outside tests. +func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run CommandRunner) *ComposeExecutor { + return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run} +} + +// Execute pins job.Ref to its current digest, writes it to the stack's +// override file and recreates the service. Working directory and config +// files come from the service's container labels on this machine, never from +// the job. It returns the pinned ref. +func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) { + project, service := job.ComposeProject, job.ComposeService + cts, err := e.cli.ContainerList(ctx, container.ListOptions{ + All: true, + Filters: filters.NewArgs( + filters.Arg("label", labelComposeProject+"="+project), + filters.Arg("label", labelComposeService+"="+service), + ), + }) if err != nil { - return fmt.Errorf("docker %s: %w (output: %s)", - strings.Join(args, " "), err, strings.TrimSpace(string(out))) + return "", fmt.Errorf("list containers: %w", err) } - return nil + if len(cts) == 0 { + return "", fmt.Errorf("no container for compose service %s/%s on this machine", project, service) + } + workingDir := cts[0].Labels[labelWorkingDir] + configFiles := cts[0].Labels[labelConfigFiles] + + roots, err := MountRoots(ctx, e.cli, e.selfID) + if err != nil { + return "", err + } + files, dir, err := composeFiles(roots, configFiles) + if err != nil { + return "", err + } + if workingDir == "" { + workingDir = dir + } else if !insideAny(roots, workingDir) { + return "", notMounted(workingDir) + } + + pinned, err := e.resolve.Resolve(ctx, job.Ref) + if err != nil { + return "", err + } + fmt.Fprintf(out, "resolved %s\n", pinned) + + if _, err := WriteOverride(dir, service, pinned); err != nil { + return "", err + } + + base := []string{"compose", "-p", project} + for _, f := range files { + base = append(base, "-f", f) + } + for _, step := range [][]string{{"pull", service}, {"up", "-d", service}} { + args := append(append([]string(nil), base...), step...) + fmt.Fprintf(out, "$ docker %s\n", strings.Join(args, " ")) + tail := &tailBuffer{max: errorOutputTail} + if err := e.run(ctx, workingDir, io.MultiWriter(out, tailWriter{tail}), args...); err != nil { + return "", fmt.Errorf("docker compose %s: %w (output: %s)", step[0], err, strings.TrimSpace(tail.String())) + } + } + return pinned, nil } diff --git a/internal/agent/updater/compose_test.go b/internal/agent/updater/compose_test.go new file mode 100644 index 0000000..fae49ad --- /dev/null +++ b/internal/agent/updater/compose_test.go @@ -0,0 +1,132 @@ +package updater + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/mount" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/protocol" +) + +type fakeCompose struct { + fakeSelf + containers []container.Summary +} + +func (f fakeCompose) ContainerList(context.Context, container.ListOptions) ([]container.Summary, error) { + return f.containers, nil +} + +type ranCmd struct { + dir string + args []string +} + +type recorder struct { + cmds []ranCmd + failOn string // second arg after -f list, e.g. "pull" + failOutput string + onFirst func() +} + +func (r *recorder) run(_ context.Context, dir string, out io.Writer, args ...string) error { + if len(r.cmds) == 0 && r.onFirst != nil { + r.onFirst() + } + r.cmds = append(r.cmds, ranCmd{dir, args}) + if r.failOn != "" && strings.Contains(strings.Join(args, " "), " "+r.failOn+" ") { + io.WriteString(out, r.failOutput) + return errors.New("exit status 1") + } + return nil +} + +func TestComposeExecutor(t *testing.T) { + const d = "sha256:d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1" + root, _ := filepath.EvalSymlinks(t.TempDir()) + stack := filepath.Join(root, "slusk") + require.NoError(t, os.MkdirAll(stack, 0o755)) + job := protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4"} + pinned := job.Ref + "@" + d + + ct := container.Summary{Labels: map[string]string{ + labelWorkingDir: stack, + labelConfigFiles: filepath.Join(stack, "compose.yaml"), + }} + rw := fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeBind, Source: root, Destination: root, RW: true}}} + ro := fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeBind, Source: root, Destination: root, RW: false}}} + wantFiles := []string{"-f", filepath.Join(stack, "compose.yaml"), "-f", filepath.Join(stack, OverrideName)} + cmd := func(rest ...string) []string { + return append(append([]string{"compose", "-p", "slusk"}, wantFiles...), rest...) + } + overridePath := filepath.Join(stack, OverrideName) + cleanOverride := func() { os.Remove(overridePath) } + + setup := func(cli ComposeDocker, dist *fakeDist, rec *recorder) *ComposeExecutor { + return NewComposeExecutor(cli, NewResolver(dist, nil), "me", rec.run) + } + + t.Run("happy path", func(t *testing.T) { + t.Cleanup(cleanOverride) + rec := &recorder{onFirst: func() { + _, err := os.Stat(overridePath) + assert.NoError(t, err, "override must exist before the first command") + }} + var out strings.Builder + got, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, &out) + require.NoError(t, err) + assert.Equal(t, pinned, got) + require.Len(t, rec.cmds, 2) + assert.Equal(t, ranCmd{stack, cmd("pull", "web")}, rec.cmds[0]) + assert.Equal(t, ranCmd{stack, cmd("up", "-d", "web")}, rec.cmds[1]) + _, doc := readOverride(t, stack) + assert.Equal(t, pinned, serviceImage(doc, "web")) + assert.Contains(t, out.String(), "resolved "+pinned) + }) + t.Run("no matching container", func(t *testing.T) { + rec := &recorder{} + _, err := setup(fakeCompose{rw, nil}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard) + assert.ErrorContains(t, err, "no container for compose service slusk/web on this machine") + assert.Empty(t, rec.cmds) + }) + t.Run("resolve fails", func(t *testing.T) { + t.Cleanup(cleanOverride) + rec := &recorder{} + _, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{err: errors.New("denied")}, rec).Execute(context.Background(), job, io.Discard) + assert.ErrorContains(t, err, "denied") + assert.Empty(t, rec.cmds) + assert.NoFileExists(t, overridePath) + }) + t.Run("pull fails", func(t *testing.T) { + t.Cleanup(cleanOverride) + rec := &recorder{failOn: "pull", failOutput: "pull access denied\n"} + _, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard) + assert.ErrorContains(t, err, "pull access denied") + require.Len(t, rec.cmds, 1, "up must not run") + }) + t.Run("override refused", func(t *testing.T) { + other := filepath.Join(stack, "docker-compose.override.yml") + require.NoError(t, os.WriteFile(other, []byte("services: {}\n"), 0o644)) + t.Cleanup(func() { os.Remove(other) }) + rec := &recorder{} + _, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard) + assert.ErrorContains(t, err, "refused") + assert.Empty(t, rec.cmds) + }) + t.Run("read-only mount", func(t *testing.T) { + rec := &recorder{} + _, err := setup(fakeCompose{ro, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard) + assert.ErrorContains(t, err, "no stack directory is mounted") + assert.Empty(t, rec.cmds) + assert.NoFileExists(t, overridePath) + }) +} diff --git a/internal/agent/updater/dispatch.go b/internal/agent/updater/dispatch.go new file mode 100644 index 0000000..3792c8a --- /dev/null +++ b/internal/agent/updater/dispatch.go @@ -0,0 +1,28 @@ +package updater + +import ( + "context" + "fmt" + "io" + + "github.com/shcizo/package-updater/internal/protocol" +) + +type dispatch struct{ compose, swarm Executor } + +// NewDispatch returns an Executor that picks compose or swarm by job.Kind. +// KindContainer and unknown kinds fail. +func NewDispatch(compose, swarm Executor) Executor { + return dispatch{compose: compose, swarm: swarm} +} + +func (d dispatch) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) { + switch job.Kind { + case protocol.KindCompose: + return d.compose.Execute(ctx, job, out) + case protocol.KindSwarm: + return d.swarm.Execute(ctx, job, out) + default: + return "", fmt.Errorf("cannot update a service of kind %q", job.Kind) + } +} diff --git a/internal/agent/updater/dispatch_test.go b/internal/agent/updater/dispatch_test.go new file mode 100644 index 0000000..bfbd7cb --- /dev/null +++ b/internal/agent/updater/dispatch_test.go @@ -0,0 +1,31 @@ +package updater + +import ( + "context" + "io" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/protocol" +) + +func TestDispatch(t *testing.T) { + compose := execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "compose", nil }) + swarm := execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "swarm", nil }) + d := NewDispatch(compose, swarm) + + got, err := d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindCompose}, io.Discard) + require.NoError(t, err) + assert.Equal(t, "compose", got) + + got, err = d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindSwarm}, io.Discard) + require.NoError(t, err) + assert.Equal(t, "swarm", got) + + _, err = d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindContainer}, io.Discard) + assert.ErrorContains(t, err, "cannot update") + _, err = d.Execute(context.Background(), protocol.JobStart{Kind: "weird"}, io.Discard) + assert.ErrorContains(t, err, "cannot update") +} diff --git a/internal/agent/updater/override.go b/internal/agent/updater/override.go new file mode 100644 index 0000000..7d67ccd --- /dev/null +++ b/internal/agent/updater/override.go @@ -0,0 +1,192 @@ +package updater + +import ( + "bytes" + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/mount" + "gopkg.in/yaml.v3" +) + +// OverrideName is the override file the agent owns. docker compose loads it +// automatically next to compose.yaml or docker-compose.yml. +const OverrideName = "compose.override.yaml" + +// overrideHeader marks a file the agent may rewrite. A file without it +// belongs to the operator and is never touched. +const overrideHeader = "# managed by package-updater — do not edit\n" + +// overrideNames are the override files docker compose auto-loads. It loads +// only the first one present (in this order), so any other one next to ours +// would either hide ours or be hidden by it. +var overrideNames = []string{ + "compose.override.yml", "compose.override.yaml", + "docker-compose.override.yml", "docker-compose.override.yaml", +} + +// WriteOverride sets services..image to ref in dir/OverrideName, +// keeping every other entry in the file, and returns the file's path. It +// refuses when the file exists without the header, or when another +// auto-loaded override file exists in dir. The write is atomic. +func WriteOverride(dir, service, ref string) (string, error) { + for _, name := range overrideNames { + if name == OverrideName { + continue + } + if _, err := os.Lstat(filepath.Join(dir, name)); err == nil { + return "", fmt.Errorf("refused: %s exists; the agent only manages %s and never edits operator files", name, OverrideName) + } else if !errors.Is(err, os.ErrNotExist) { + return "", fmt.Errorf("check %s: %w", name, err) + } + } + path := filepath.Join(dir, OverrideName) + doc := map[string]any{} + raw, err := os.ReadFile(path) + switch { + case errors.Is(err, os.ErrNotExist): + case err != nil: + return "", fmt.Errorf("read override: %w", err) + case !bytes.HasPrefix(raw, []byte(overrideHeader)): + return "", fmt.Errorf("refused: %s exists and was not written by package-updater", path) + default: + if err := yaml.Unmarshal(raw, &doc); err != nil { + return "", fmt.Errorf("parse override: %w", err) + } + if doc == nil { + doc = map[string]any{} + } + } + services, _ := doc["services"].(map[string]any) + if services == nil { + services = map[string]any{} + } + entry, _ := services[service].(map[string]any) + if entry == nil { + entry = map[string]any{} + } + entry["image"] = ref + services[service] = entry + doc["services"] = services + body, err := yaml.Marshal(doc) + if err != nil { + return "", fmt.Errorf("encode override: %w", err) + } + // Temp file in the same directory + rename: a crash never leaves a + // half-written override that compose would then load. + tmp, err := os.CreateTemp(dir, "."+OverrideName+".*") + if err != nil { + return "", fmt.Errorf("write override: %w", err) + } + defer os.Remove(tmp.Name()) // no-op after a successful rename + if _, err := tmp.WriteString(overrideHeader + string(body)); err != nil { + tmp.Close() + return "", fmt.Errorf("write override: %w", err) + } + if err := tmp.Chmod(0o644); err != nil { + tmp.Close() + return "", fmt.Errorf("write override: %w", err) + } + if err := tmp.Close(); err != nil { + return "", fmt.Errorf("write override: %w", err) + } + if err := os.Rename(tmp.Name(), path); err != nil { + return "", fmt.Errorf("write override: %w", err) + } + return path, nil +} + +// SelfInspector is the subset of the Docker SDK MountRoots needs. +type SelfInspector interface { + ContainerInspect(ctx context.Context, containerID string) (container.InspectResponse, error) +} + +// errNoRoots is returned when the agent has no directory it may write stacks in. +var errNoRoots = errors.New("refused: no stack directory is mounted into the agent at the same path (read-write); " + + "mount each stacks directory as /path:/path and check that SELF_ID (default: the hostname) names the agent's container") + +// MountRoots returns the directories the agent may update Compose stacks in: +// its own container's read-write bind mounts whose host path equals the +// path inside the container. Compose labels hold host paths, so only those +// mounts make a label path usable as is. +func MountRoots(ctx context.Context, cli SelfInspector, selfID string) ([]string, error) { + if selfID == "" { + return nil, errNoRoots + } + insp, err := cli.ContainerInspect(ctx, selfID) + if err != nil { + return nil, fmt.Errorf("inspect own container %s: %w", selfID, err) + } + var roots []string + for _, m := range insp.Mounts { + if m.Type == mount.TypeBind && m.RW && filepath.Clean(m.Source) == filepath.Clean(m.Destination) { + roots = append(roots, filepath.Clean(m.Destination)) + } + } + return roots, nil +} + +// composeFiles returns the -f arguments for a job: the stack's config files +// from the container label with any copy of the override removed, then the +// override last so it wins. Every file, and the real location of the stack +// directory after resolving symlinks, must be inside one of roots. +func composeFiles(roots []string, configFilesLabel string) (files []string, dir string, err error) { + if len(roots) == 0 { + return nil, "", errNoRoots + } + for _, f := range strings.Split(configFilesLabel, ",") { + if f = strings.TrimSpace(f); f != "" { + files = append(files, f) + } + } + if len(files) == 0 { + return nil, "", errors.New("container has no com.docker.compose.project.config_files label") + } + dir = filepath.Dir(files[0]) + override := filepath.Join(dir, OverrideName) + out := files[:0] + for _, f := range files { + if !filepath.IsAbs(f) || !insideAny(roots, f) { + return nil, "", notMounted(filepath.Dir(f)) + } + if filepath.Clean(f) != override { + out = append(out, f) + } + } + real, err := filepath.EvalSymlinks(dir) + if err != nil { + return nil, "", fmt.Errorf("resolve %s: %w", dir, err) + } + // Only directories count once resolved: this drops a same-path mount of + // a single file such as /var/run/docker.sock. + var realRoots []string + for _, r := range roots { + if rr, err := filepath.EvalSymlinks(r); err == nil { + if fi, err := os.Stat(rr); err == nil && fi.IsDir() { + realRoots = append(realRoots, rr) + } + } + } + if !insideAny(realRoots, real) { + return nil, "", notMounted(dir) + } + return append(out, override), dir, nil +} + +func insideAny(roots []string, path string) bool { + for _, r := range roots { + if IsInsideRoot(r, path) { + return true + } + } + return false +} + +func notMounted(dir string) error { + return fmt.Errorf("stack directory %s is not mounted into the agent at the same path (read-write)", dir) +} diff --git a/internal/agent/updater/override_test.go b/internal/agent/updater/override_test.go new file mode 100644 index 0000000..31a2724 --- /dev/null +++ b/internal/agent/updater/override_test.go @@ -0,0 +1,199 @@ +package updater + +import ( + "context" + "errors" + "os" + "path/filepath" + "testing" + + "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/mount" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" +) + +func readOverride(t *testing.T, dir string) (string, map[string]any) { + t.Helper() + raw, err := os.ReadFile(filepath.Join(dir, OverrideName)) + require.NoError(t, err) + var doc map[string]any + require.NoError(t, yaml.Unmarshal(raw, &doc)) + return string(raw), doc +} + +func serviceImage(doc map[string]any, name string) any { + svcs, _ := doc["services"].(map[string]any) + entry, _ := svcs[name].(map[string]any) + return entry["image"] +} + +func TestWriteOverride(t *testing.T) { + t.Run("new file", func(t *testing.T) { + dir := t.TempDir() + path, err := WriteOverride(dir, "web", "a:1@sha256:aa") + require.NoError(t, err) + assert.Equal(t, filepath.Join(dir, OverrideName), path) + raw, doc := readOverride(t, dir) + assert.True(t, len(raw) > len(overrideHeader) && raw[:len(overrideHeader)] == overrideHeader) + assert.Equal(t, "a:1@sha256:aa", serviceImage(doc, "web")) + }) + t.Run("keeps other services", func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(overrideHeader+"services:\n db:\n image: db:1\n"), 0o644)) + _, err := WriteOverride(dir, "web", "a:1@sha256:aa") + require.NoError(t, err) + _, doc := readOverride(t, dir) + assert.Equal(t, "db:1", serviceImage(doc, "db")) + assert.Equal(t, "a:1@sha256:aa", serviceImage(doc, "web")) + }) + t.Run("replaces own entry", func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(overrideHeader+"services:\n web:\n image: old\n"), 0o644)) + _, err := WriteOverride(dir, "web", "a:2@sha256:bb") + require.NoError(t, err) + _, doc := readOverride(t, dir) + assert.Equal(t, "a:2@sha256:bb", serviceImage(doc, "web")) + }) + t.Run("operator file without header is never touched", func(t *testing.T) { + dir := t.TempDir() + const body = "services:\n web:\n environment: [A=1]\n" + require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(body), 0o644)) + _, err := WriteOverride(dir, "web", "a:1@sha256:aa") + assert.ErrorContains(t, err, "refused") + got, _ := os.ReadFile(filepath.Join(dir, OverrideName)) + assert.Equal(t, body, string(got)) + }) + for _, other := range []string{"compose.override.yml", "docker-compose.override.yml", "docker-compose.override.yaml"} { + t.Run("refuses next to "+other, func(t *testing.T) { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, other), []byte("services: {}\n"), 0o644)) + _, err := WriteOverride(dir, "web", "a:1@sha256:aa") + assert.ErrorContains(t, err, other) + _, statErr := os.Stat(filepath.Join(dir, OverrideName)) + assert.True(t, errors.Is(statErr, os.ErrNotExist)) + }) + } + t.Run("leaves no temp files", func(t *testing.T) { + dir := t.TempDir() + _, err := WriteOverride(dir, "web", "a:1@sha256:aa") + require.NoError(t, err) + entries, err := os.ReadDir(dir) + require.NoError(t, err) + require.Len(t, entries, 1) + assert.Equal(t, OverrideName, entries[0].Name()) + }) +} + +type fakeSelf struct { + mounts []container.MountPoint + err error +} + +func (f fakeSelf) ContainerInspect(context.Context, string) (container.InspectResponse, error) { + return container.InspectResponse{Mounts: f.mounts}, f.err +} + +func TestMountRoots(t *testing.T) { + bind := func(src, dst string, rw bool) container.MountPoint { + return container.MountPoint{Type: mount.TypeBind, Source: src, Destination: dst, RW: rw} + } + cases := []struct { + name string + self fakeSelf + selfID string + want []string + wantErr string + }{ + {"rw same path", fakeSelf{mounts: []container.MountPoint{bind("/srv/stacks", "/srv/stacks", true)}}, "me", []string{"/srv/stacks"}, ""}, + {"read-only", fakeSelf{mounts: []container.MountPoint{bind("/srv/stacks", "/srv/stacks", false)}}, "me", nil, ""}, + {"path translated", fakeSelf{mounts: []container.MountPoint{bind("/home/x/stacks", "/stacks", true)}}, "me", nil, ""}, + {"named volume", fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeVolume, Source: "/v", Destination: "/v", RW: true}}}, "me", nil, ""}, + {"inspect error", fakeSelf{err: errors.New("no such container")}, "me", nil, "inspect own container"}, + {"no self id", fakeSelf{}, "", nil, "no stack directory is mounted"}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, err := MountRoots(context.Background(), c.self, c.selfID) + if c.wantErr != "" { + assert.ErrorContains(t, err, c.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, c.want, got) + }) + } +} + +func TestComposeFiles(t *testing.T) { + root := t.TempDir() + root, err := filepath.EvalSymlinks(root) // macOS temp dirs are symlinked + require.NoError(t, err) + mk := func(p ...string) string { + dir := filepath.Join(append([]string{root}, p...)...) + require.NoError(t, os.MkdirAll(dir, 0o755)) + return dir + } + a := mk("a") + nested := mk("apps", "x", "y") + second := t.TempDir() + second, _ = filepath.EvalSymlinks(second) + secondDir := filepath.Join(second, "s") + require.NoError(t, os.MkdirAll(secondDir, 0o755)) + + outside, _ := filepath.EvalSymlinks(t.TempDir()) + require.NoError(t, os.Symlink(outside, filepath.Join(root, "evil"))) + + sock := filepath.Join(root, "docker.sock") + require.NoError(t, os.WriteFile(sock, nil, 0o644)) + + t.Run("adds override last", func(t *testing.T) { + files, dir, err := composeFiles([]string{root}, filepath.Join(a, "compose.yaml")) + require.NoError(t, err) + assert.Equal(t, []string{filepath.Join(a, "compose.yaml"), filepath.Join(a, OverrideName)}, files) + assert.Equal(t, a, dir) + }) + t.Run("override in label listed once, last", func(t *testing.T) { + label := filepath.Join(a, OverrideName) + "," + filepath.Join(a, "compose.yaml") + "," + filepath.Join(a, "extra.yaml") + files, _, err := composeFiles([]string{root}, label) + require.NoError(t, err) + assert.Equal(t, []string{filepath.Join(a, "compose.yaml"), filepath.Join(a, "extra.yaml"), filepath.Join(a, OverrideName)}, files) + }) + t.Run("nested", func(t *testing.T) { + _, dir, err := composeFiles([]string{root}, filepath.Join(nested, "compose.yaml")) + require.NoError(t, err) + assert.Equal(t, nested, dir) + }) + t.Run("second root", func(t *testing.T) { + _, _, err := composeFiles([]string{root, second}, filepath.Join(secondDir, "compose.yaml")) + require.NoError(t, err) + }) + t.Run("empty label", func(t *testing.T) { + _, _, err := composeFiles([]string{root}, "") + assert.ErrorContains(t, err, "config_files") + }) + t.Run("no roots", func(t *testing.T) { + _, _, err := composeFiles(nil, filepath.Join(a, "compose.yaml")) + assert.ErrorContains(t, err, "no stack directory is mounted") + }) + t.Run("outside every root", func(t *testing.T) { + _, _, err := composeFiles([]string{root}, filepath.Join(second, "s", "compose.yaml")) + assert.EqualError(t, err, "stack directory "+filepath.Join(second, "s")+" is not mounted into the agent at the same path (read-write)") + }) + t.Run("symlink escaping the root", func(t *testing.T) { + _, _, err := composeFiles([]string{root}, filepath.Join(root, "evil", "compose.yaml")) + assert.EqualError(t, err, "stack directory "+filepath.Join(root, "evil")+" is not mounted into the agent at the same path (read-write)") + }) + t.Run("prefix trick", func(t *testing.T) { + evil := root + "-evil" + require.NoError(t, os.MkdirAll(evil, 0o755)) + t.Cleanup(func() { os.RemoveAll(evil) }) + _, _, err := composeFiles([]string{root}, filepath.Join(evil, "compose.yaml")) + assert.ErrorContains(t, err, "not mounted") + }) + t.Run("a file root allows nothing", func(t *testing.T) { + _, _, err := composeFiles([]string{sock}, filepath.Join(sock, "compose.yaml")) + assert.Error(t, err) + }) +} diff --git a/internal/agent/updater/pathcheck.go b/internal/agent/updater/pathcheck.go index 04e8f21..29254a1 100644 --- a/internal/agent/updater/pathcheck.go +++ b/internal/agent/updater/pathcheck.go @@ -1,4 +1,4 @@ -package discovery +package updater import ( "path/filepath" diff --git a/internal/agent/updater/pathcheck_test.go b/internal/agent/updater/pathcheck_test.go index 5e08697..562d690 100644 --- a/internal/agent/updater/pathcheck_test.go +++ b/internal/agent/updater/pathcheck_test.go @@ -1,9 +1,9 @@ -package discovery_test +package updater_test import ( "testing" - "github.com/shcizo/package-updater/internal/discovery" + "github.com/shcizo/package-updater/internal/agent/updater" "github.com/stretchr/testify/require" ) @@ -26,7 +26,7 @@ func TestIsInsideRoot(t *testing.T) { } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - require.Equal(t, c.want, discovery.IsInsideRoot(c.root, c.path)) + require.Equal(t, c.want, updater.IsInsideRoot(c.root, c.path)) }) } } diff --git a/internal/agent/updater/registryauth.go b/internal/agent/updater/registryauth.go new file mode 100644 index 0000000..a19e8b4 --- /dev/null +++ b/internal/agent/updater/registryauth.go @@ -0,0 +1,106 @@ +package updater + +import ( + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "strings" + + "github.com/distribution/reference" + "github.com/docker/docker/api/types/registry" +) + +// RegistryAuth holds inline credentials from a docker config.json, already +// encoded as X-Registry-Auth values. A nil *RegistryAuth means anonymous +// access. +type RegistryAuth struct { + byHost map[string]string +} + +type dockerConfigFile struct { + Auths map[string]struct { + Auth string `json:"auth"` + IdentityToken string `json:"identitytoken"` + } `json:"auths"` + CredsStore string `json:"credsStore"` + CredHelpers map[string]string `json:"credHelpers"` +} + +// LoadRegistryAuth reads dockerConfigDir/config.json. An empty dir returns +// nil, nil. Only inline `auth` and `identitytoken` entries are supported; +// credential helpers are not available inside the agent container, which is +// logged once. +func LoadRegistryAuth(dockerConfigDir string, log *slog.Logger) (*RegistryAuth, error) { + if dockerConfigDir == "" { + return nil, nil + } + path := filepath.Join(dockerConfigDir, "config.json") + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read %s: %w", path, err) + } + var cfg dockerConfigFile + if err := json.Unmarshal(raw, &cfg); err != nil { + return nil, fmt.Errorf("parse %s: %w", path, err) + } + if (cfg.CredsStore != "" || len(cfg.CredHelpers) > 0) && log != nil { + log.Warn("registry_cred_helpers_unsupported", "path", path) + } + a := &RegistryAuth{byHost: make(map[string]string)} + for key, entry := range cfg.Auths { + host := normaliseRegistryKey(key) + ac := registry.AuthConfig{ServerAddress: host} + switch { + case entry.IdentityToken != "": + ac.IdentityToken = entry.IdentityToken + case entry.Auth != "": + dec, err := base64.StdEncoding.DecodeString(entry.Auth) + if err != nil { + return nil, fmt.Errorf("%s: auth for %s is not base64", path, key) + } + user, pass, ok := strings.Cut(string(dec), ":") + if !ok { + return nil, errors.New(path + ": auth for " + key + " is not user:password") + } + ac.Username, ac.Password = user, pass + default: + continue + } + enc, err := registry.EncodeAuthConfig(ac) + if err != nil { + return nil, fmt.Errorf("encode auth for %s: %w", key, err) + } + a.byHost[host] = enc + } + return a, nil +} + +// normaliseRegistryKey reduces a config.json key to a bare registry host, +// mapping every Docker Hub spelling to "docker.io". +func normaliseRegistryKey(key string) string { + k := strings.TrimPrefix(strings.TrimPrefix(key, "https://"), "http://") + if i := strings.Index(k, "/"); i >= 0 { + k = k[:i] + } + switch k { + case "index.docker.io", "docker.io", "registry-1.docker.io": + return "docker.io" + } + return k +} + +// For returns the X-Registry-Auth value for repo's registry, or "". +func (a *RegistryAuth) For(repo string) string { + if a == nil { + return "" + } + named, err := reference.ParseNormalizedNamed(repo) + if err != nil { + return "" + } + return a.byHost[reference.Domain(named)] +} diff --git a/internal/agent/updater/registryauth_test.go b/internal/agent/updater/registryauth_test.go new file mode 100644 index 0000000..693e814 --- /dev/null +++ b/internal/agent/updater/registryauth_test.go @@ -0,0 +1,72 @@ +package updater + +import ( + "encoding/base64" + "encoding/json" + "io" + "log/slog" + "os" + "path/filepath" + "testing" + + "github.com/docker/docker/api/types/registry" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func writeDockerConfig(t *testing.T, body string) string { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(body), 0o600)) + return dir +} + +func decodeAuth(t *testing.T, enc string) registry.AuthConfig { + t.Helper() + raw, err := base64.URLEncoding.DecodeString(enc) + require.NoError(t, err) + var ac registry.AuthConfig + require.NoError(t, json.Unmarshal(raw, &ac)) + return ac +} + +func TestRegistryAuth(t *testing.T) { + user := base64.StdEncoding.EncodeToString([]byte("bob:s3cret")) + log := slog.New(slog.NewTextHandler(io.Discard, nil)) + + t.Run("docker hub spellings", func(t *testing.T) { + for _, key := range []string{"https://index.docker.io/v1/", "index.docker.io", "docker.io", "registry-1.docker.io"} { + dir := writeDockerConfig(t, `{"auths":{"`+key+`":{"auth":"`+user+`"}}}`) + a, err := LoadRegistryAuth(dir, log) + require.NoError(t, err) + for _, repo := range []string{"gitea/gitea", "docker.io/library/nginx", "nginx"} { + ac := decodeAuth(t, a.For(repo)) + assert.Equal(t, "bob", ac.Username, "%s for %s", key, repo) + assert.Equal(t, "s3cret", ac.Password) + } + assert.Empty(t, a.For("gitea.shcizo.se/shcizo/slusk")) + } + }) + t.Run("custom registry with scheme", func(t *testing.T) { + dir := writeDockerConfig(t, `{"auths":{"https://gitea.shcizo.se":{"auth":"`+user+`"},"ghcr.io":{"identitytoken":"tok"}}}`) + a, err := LoadRegistryAuth(dir, log) + require.NoError(t, err) + assert.Equal(t, "bob", decodeAuth(t, a.For("gitea.shcizo.se/shcizo/slusk")).Username) + assert.Equal(t, "tok", decodeAuth(t, a.For("ghcr.io/x/y")).IdentityToken) + assert.Empty(t, a.For("registry.example.com/x")) + }) + t.Run("empty dir is anonymous", func(t *testing.T) { + a, err := LoadRegistryAuth("", log) + require.NoError(t, err) + assert.Nil(t, a) + assert.Empty(t, a.For("gitea/gitea")) + }) + t.Run("malformed json", func(t *testing.T) { + _, err := LoadRegistryAuth(writeDockerConfig(t, "{"), log) + assert.Error(t, err) + }) + t.Run("missing file", func(t *testing.T) { + _, err := LoadRegistryAuth(t.TempDir(), log) + assert.Error(t, err) + }) +} diff --git a/internal/agent/updater/resolve.go b/internal/agent/updater/resolve.go new file mode 100644 index 0000000..9923cbf --- /dev/null +++ b/internal/agent/updater/resolve.go @@ -0,0 +1,49 @@ +package updater + +import ( + "context" + "errors" + "fmt" + + "github.com/docker/docker/api/types/registry" + + "github.com/shcizo/package-updater/internal/imageref" +) + +// DistributionClient is the subset of the Docker SDK the Resolver needs. +// *client.Client satisfies it. +type DistributionClient interface { + DistributionInspect(ctx context.Context, imageRef, encodedRegistryAuth string) (registry.DistributionInspect, error) +} + +// Resolver turns "repo:tag" into "repo:tag@digest" by asking the local +// daemon, which contacts the registry. +type Resolver struct { + cli DistributionClient + auth *RegistryAuth +} + +// NewResolver returns a Resolver. auth may be nil (anonymous access). +func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver { + return &Resolver{cli: cli, auth: auth} +} + +// Resolve returns ref pinned to the digest its tag currently points at. It +// fails for a ref without a tag or with a digest already. +func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) { + p := imageref.Parse(ref) + switch { + case p.Digest != "": + return "", fmt.Errorf("image %q already has a digest", ref) + case p.Tag == "": + return "", fmt.Errorf("image %q has no tag to resolve", ref) + } + info, err := r.cli.DistributionInspect(ctx, ref, r.auth.For(p.Repo)) + if err != nil { + return "", fmt.Errorf("resolve %s: %w", ref, err) + } + if info.Descriptor.Digest == "" { + return "", errors.New("resolve " + ref + ": registry returned no digest") + } + return ref + "@" + info.Descriptor.Digest.String(), nil +} diff --git a/internal/agent/updater/resolve_test.go b/internal/agent/updater/resolve_test.go new file mode 100644 index 0000000..a6eab28 --- /dev/null +++ b/internal/agent/updater/resolve_test.go @@ -0,0 +1,38 @@ +package updater + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestResolver(t *testing.T) { + const d = "sha256:d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1" + auth := &RegistryAuth{byHost: map[string]string{"gitea.shcizo.se": "enc-gitea"}} + + t.Run("resolves tag with registry auth", func(t *testing.T) { + dist := &fakeDist{digest: d} + got, err := NewResolver(dist, auth).Resolve(context.Background(), "gitea.shcizo.se/shcizo/slusk:1.4") + require.NoError(t, err) + assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:1.4@"+d, got) + assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs) + assert.Equal(t, []string{"enc-gitea"}, dist.auths) + }) + t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) { + for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} { + dist := &fakeDist{digest: d} + _, err := NewResolver(dist, nil).Resolve(context.Background(), ref) + assert.Error(t, err, ref) + assert.Empty(t, dist.refs) + } + }) + t.Run("daemon error is wrapped", func(t *testing.T) { + cause := errors.New("unauthorized") + _, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1") + assert.ErrorIs(t, err, cause) + assert.ErrorContains(t, err, "repo:1") + }) +} diff --git a/internal/agent/updater/runner.go b/internal/agent/updater/runner.go new file mode 100644 index 0000000..ad8e6ea --- /dev/null +++ b/internal/agent/updater/runner.go @@ -0,0 +1,272 @@ +package updater + +import ( + "context" + "errors" + "fmt" + "io" + "sort" + "sync" + "time" + + "github.com/shcizo/package-updater/internal/protocol" +) + +// maxProgress caps the output buffered between two TakeProgress calls, so a +// chatty job cannot grow a single job.progress message without bound. +const maxProgress = 32 << 10 + +// Executor deploys one job and returns the "repo:tag@digest" it deployed. +// Everything it writes to out is streamed to the master as job output. +type Executor interface { + Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) +} + +// Progress is output a running job produced since the previous TakeProgress. +type Progress struct{ ID, Output string } + +// Finished is a job's outcome, kept until the master acknowledges it. +type Finished struct { + ID string + Result protocol.JobResult +} + +// tailBuffer keeps the last max bytes written to it. +type tailBuffer struct { + max int + buf []byte + truncated bool +} + +func (b *tailBuffer) Write(p []byte) { + b.buf = append(b.buf, p...) + if over := len(b.buf) - b.max; over > 0 { + b.buf = append(b.buf[:0], b.buf[over:]...) + b.truncated = true + } +} + +func (b *tailBuffer) String() string { return string(b.buf) } +func (b *tailBuffer) Truncated() bool { return b.truncated } +func (b *tailBuffer) Reset() { b.buf, b.truncated = nil, false } + +type jobState struct { + start protocol.JobStart + running bool + result *protocol.JobResult + output tailBuffer // everything, capped at protocol.MaxJobOutput + unsent tailBuffer // since the last TakeProgress, capped at maxProgress + reported bool // a Progress has been taken since the job started + finished time.Time // orders Pending +} + +// Runner executes jobs one at a time in the order they arrive and keeps each +// job's result until Ack. Two docker compose calls against the same stack +// would race, so there is deliberately exactly one worker. +// +// Runner satisfies conn.Jobs. +type Runner struct { + exec Executor + timeout time.Duration + now func() time.Time + + mu sync.Mutex // guards jobs and queue + jobs map[string]*jobState + queue []string + + wake chan struct{} // wakes the worker + changed chan struct{} // tells the connection there is something to send +} + +// NewRunner returns a Runner that gives every job at most timeout. Call Run +// to start the worker. +func NewRunner(exec Executor, timeout time.Duration) *Runner { + return &Runner{ + exec: exec, + timeout: timeout, + now: time.Now, + jobs: make(map[string]*jobState), + wake: make(chan struct{}, 1), + changed: make(chan struct{}, 1), + } +} + +// signal does a non-blocking send so neither side ever waits on the other. +func signal(ch chan struct{}) { + select { + case ch <- struct{}{}: + default: + } +} + +// Submit queues a job. A job whose ID the runner already holds (queued, +// running, or finished and not yet acknowledged) is ignored. +func (r *Runner) Submit(id string, start protocol.JobStart) { + r.mu.Lock() + if _, ok := r.jobs[id]; ok { + r.mu.Unlock() + return + } + r.jobs[id] = &jobState{ + start: start, + output: tailBuffer{max: protocol.MaxJobOutput}, + unsent: tailBuffer{max: maxProgress}, + } + r.queue = append(r.queue, id) + r.mu.Unlock() + signal(r.wake) +} + +// Ack forgets a finished job. Acks for unknown, queued or running jobs are +// ignored. +func (r *Runner) Ack(id string) { + r.mu.Lock() + defer r.mu.Unlock() + if j, ok := r.jobs[id]; ok && j.result != nil { + delete(r.jobs, id) + } +} + +// Known returns the sorted IDs of every job the runner holds. +func (r *Runner) Known() []string { + r.mu.Lock() + defer r.mu.Unlock() + ids := make([]string, 0, len(r.jobs)) + for id := range r.jobs { + ids = append(ids, id) + } + sort.Strings(ids) + return ids +} + +// TakeProgress returns the output running jobs produced since the previous +// call. A job's first entry (possibly with empty Output) tells the master it +// is running. +func (r *Runner) TakeProgress() []Progress { + r.mu.Lock() + defer r.mu.Unlock() + var out []Progress + for id, j := range r.jobs { + if !j.running { + continue + } + s := j.unsent.String() + if j.reported && s == "" { + continue + } + j.reported = true + j.unsent.Reset() + out = append(out, Progress{ID: id, Output: s}) + } + sort.Slice(out, func(a, b int) bool { return out[a].ID < out[b].ID }) + return out +} + +// Pending returns the results not yet acknowledged, oldest first. +func (r *Runner) Pending() []Finished { + r.mu.Lock() + defer r.mu.Unlock() + var out []Finished + var at []time.Time + for id, j := range r.jobs { + if j.result != nil { + out = append(out, Finished{ID: id, Result: *j.result}) + at = append(at, j.finished) + } + } + idx := make([]int, len(out)) + for i := range idx { + idx[i] = i + } + sort.Slice(idx, func(a, b int) bool { + if !at[idx[a]].Equal(at[idx[b]]) { + return at[idx[a]].Before(at[idx[b]]) + } + return out[idx[a]].ID < out[idx[b]].ID + }) + sorted := make([]Finished, len(out)) + for i, k := range idx { + sorted[i] = out[k] + } + return sorted +} + +// Changed returns a channel that receives when there is new progress or a +// new result to send. +func (r *Runner) Changed() <-chan struct{} { return r.changed } + +// Run works through the queue until ctx ends. Jobs still queued then are +// dropped; the master marks them lost on the agent's next hello. +func (r *Runner) Run(ctx context.Context) { + for { + id, start, ok := r.next() + if !ok { + select { + case <-ctx.Done(): + return + case <-r.wake: + continue + } + } + if ctx.Err() != nil { + return + } + r.runOne(ctx, id, start) + } +} + +// next pops the next queued job and marks it running. +func (r *Runner) next() (string, protocol.JobStart, bool) { + r.mu.Lock() + defer r.mu.Unlock() + if len(r.queue) == 0 { + return "", protocol.JobStart{}, false + } + id := r.queue[0] + r.queue = r.queue[1:] + j := r.jobs[id] + j.running = true + signal(r.changed) + return id, j.start, true +} + +func (r *Runner) runOne(ctx context.Context, id string, start protocol.JobStart) { + started := r.now() + jobCtx, cancel := context.WithTimeout(ctx, r.timeout) + ref, err := r.exec.Execute(jobCtx, start, &jobWriter{r: r, id: id}) + timedOut := errors.Is(jobCtx.Err(), context.DeadlineExceeded) + cancel() + + res := protocol.JobResult{Status: protocol.ResultSucceeded, Ref: ref, StartedAt: started, FinishedAt: r.now()} + switch { + case timedOut: + res = protocol.JobResult{Status: protocol.ResultFailed, Error: fmt.Sprintf("timed out after %s", r.timeout), StartedAt: started, FinishedAt: res.FinishedAt} + case err != nil: + res = protocol.JobResult{Status: protocol.ResultFailed, Error: err.Error(), StartedAt: started, FinishedAt: res.FinishedAt} + } + + r.mu.Lock() + j := r.jobs[id] + res.Output, res.OutputTruncated = j.output.String(), j.output.Truncated() + j.running, j.result, j.finished = false, &res, res.FinishedAt + j.unsent.Reset() + r.mu.Unlock() + signal(r.changed) +} + +// jobWriter feeds an executor's output into the job's buffers. +type jobWriter struct { + r *Runner + id string +} + +func (w *jobWriter) Write(p []byte) (int, error) { + w.r.mu.Lock() + if j, ok := w.r.jobs[w.id]; ok { + j.output.Write(p) + j.unsent.Write(p) + } + w.r.mu.Unlock() + signal(w.r.changed) + return len(p), nil +} diff --git a/internal/agent/updater/runner_test.go b/internal/agent/updater/runner_test.go new file mode 100644 index 0000000..5b6a138 --- /dev/null +++ b/internal/agent/updater/runner_test.go @@ -0,0 +1,200 @@ +package updater + +import ( + "context" + "errors" + "io" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/protocol" +) + +// fakeExec writes "pulling " and then waits on gate (if set). Refs +// starting with "bad" fail. It records the highest concurrency it saw. +type fakeExec struct { + gate chan struct{} + cur atomic.Int32 + maxSeen atomic.Int32 + started chan string +} + +func newFakeExec() *fakeExec { + return &fakeExec{gate: make(chan struct{}), started: make(chan string, 10)} +} + +func (f *fakeExec) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) { + n := f.cur.Add(1) + defer f.cur.Add(-1) + for { + m := f.maxSeen.Load() + if n <= m || f.maxSeen.CompareAndSwap(m, n) { + break + } + } + f.started <- job.Ref + io.WriteString(out, "pulling "+job.Ref+"\n") + select { + case <-f.gate: + case <-ctx.Done(): + return "", ctx.Err() + } + if strings.HasPrefix(job.Ref, "bad") { + return "", errors.New("boom") + } + return job.Ref + "@sha256:aa", nil +} + +func startRunner(t *testing.T, exec Executor, timeout time.Duration) *Runner { + t.Helper() + r := NewRunner(exec, timeout) + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + wg.Add(1) + go func() { defer wg.Done(); r.Run(ctx) }() + t.Cleanup(func() { cancel(); wg.Wait() }) + return r +} + +func waitPending(t *testing.T, r *Runner, n int) []Finished { + t.Helper() + var got []Finished + require.Eventually(t, func() bool { got = r.Pending(); return len(got) >= n }, 2*time.Second, 5*time.Millisecond) + return got +} + +func TestRunner_FIFOSingleWorker(t *testing.T) { + fe := newFakeExec() + r := startRunner(t, fe, time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + r.Submit("2", protocol.JobStart{Ref: "bad:2"}) + assert.Equal(t, "a:1", <-fe.started) + fe.gate <- struct{}{} + assert.Equal(t, "bad:2", <-fe.started) + fe.gate <- struct{}{} + + got := waitPending(t, r, 2) + assert.Equal(t, int32(1), fe.maxSeen.Load()) + require.Len(t, got, 2) + assert.Equal(t, "1", got[0].ID) + assert.Equal(t, protocol.ResultSucceeded, got[0].Result.Status) + assert.Equal(t, "a:1@sha256:aa", got[0].Result.Ref) + assert.Equal(t, "pulling a:1\n", got[0].Result.Output) + assert.Equal(t, "2", got[1].ID) + assert.Equal(t, protocol.ResultFailed, got[1].Result.Status) + assert.Equal(t, "boom", got[1].Result.Error) + assert.False(t, got[0].Result.StartedAt.IsZero()) +} + +func TestRunner_DuplicateSubmitIgnored(t *testing.T) { + fe := newFakeExec() + r := startRunner(t, fe, time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + assert.Equal(t, "a:1", <-fe.started) + r.Submit("1", protocol.JobStart{Ref: "other:9"}) + fe.gate <- struct{}{} + got := waitPending(t, r, 1) + assert.Equal(t, "a:1@sha256:aa", got[0].Result.Ref) + select { + case ref := <-fe.started: + t.Fatalf("duplicate job ran: %s", ref) + case <-time.After(50 * time.Millisecond): + } +} + +func TestRunner_KnownAndAck(t *testing.T) { + fe := newFakeExec() + r := startRunner(t, fe, time.Minute) + r.Submit("2", protocol.JobStart{Ref: "a:2"}) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + assert.Equal(t, "a:2", <-fe.started) + assert.Equal(t, []string{"1", "2"}, r.Known()) + + r.Ack("2") // running: ignored + assert.Equal(t, []string{"1", "2"}, r.Known()) + + fe.gate <- struct{}{} + waitPending(t, r, 1) + r.Ack("2") + assert.Equal(t, []string{"1"}, r.Known()) + <-fe.started + fe.gate <- struct{}{} + waitPending(t, r, 1) +} + +func TestRunner_TakeProgress(t *testing.T) { + fe := newFakeExec() + r := startRunner(t, fe, time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + <-fe.started + require.Eventually(t, func() bool { return len(r.Pending()) == 0 && len(r.Known()) == 1 }, time.Second, 5*time.Millisecond) + + var first []Progress + require.Eventually(t, func() bool { first = r.TakeProgress(); return len(first) > 0 }, time.Second, 5*time.Millisecond) + assert.Equal(t, []Progress{{ID: "1", Output: "pulling a:1\n"}}, first) + assert.Empty(t, r.TakeProgress(), "nothing new since the last call") + + fe.gate <- struct{}{} + waitPending(t, r, 1) + assert.Empty(t, r.TakeProgress(), "a finished job no longer reports progress") +} + +func TestRunner_TakeProgressReportsStartedJobOnce(t *testing.T) { + gate := make(chan struct{}) + started := make(chan struct{}) + r := startRunner(t, execFunc(func(ctx context.Context, _ protocol.JobStart, _ io.Writer) (string, error) { + close(started) + <-gate + return "x", nil + }), time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + <-started + assert.Equal(t, []Progress{{ID: "1", Output: ""}}, r.TakeProgress()) + assert.Empty(t, r.TakeProgress()) + close(gate) +} + +type execFunc func(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) + +func (f execFunc) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) { + return f(ctx, job, out) +} + +func TestRunner_Timeout(t *testing.T) { + r := startRunner(t, execFunc(func(ctx context.Context, _ protocol.JobStart, _ io.Writer) (string, error) { + <-ctx.Done() + return "", ctx.Err() + }), 20*time.Millisecond) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + got := waitPending(t, r, 1) + assert.Equal(t, protocol.ResultFailed, got[0].Result.Status) + assert.Equal(t, "timed out after 20ms", got[0].Result.Error) +} + +func TestRunner_OutputCapped(t *testing.T) { + r := startRunner(t, execFunc(func(_ context.Context, _ protocol.JobStart, out io.Writer) (string, error) { + io.WriteString(out, strings.Repeat("a", 10)) + io.WriteString(out, strings.Repeat("b", protocol.MaxJobOutput)) + return "x", nil + }), time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + got := waitPending(t, r, 1) + assert.Equal(t, strings.Repeat("b", protocol.MaxJobOutput), got[0].Result.Output) + assert.True(t, got[0].Result.OutputTruncated) +} + +func TestRunner_ChangedSignals(t *testing.T) { + r := startRunner(t, execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "x", nil }), time.Minute) + r.Submit("1", protocol.JobStart{Ref: "a:1"}) + select { + case <-r.Changed(): + case <-time.After(time.Second): + t.Fatal("no change signal") + } +} diff --git a/internal/agent/updater/swarm.go b/internal/agent/updater/swarm.go index a007792..63e2ffd 100644 --- a/internal/agent/updater/swarm.go +++ b/internal/agent/updater/swarm.go @@ -3,14 +3,19 @@ package updater import ( "context" "fmt" + "io" "github.com/docker/docker/api/types/swarm" - "github.com/shcizo/package-updater/internal/discovery" + "github.com/docker/docker/api/types/system" + + "github.com/shcizo/package-updater/internal/imageref" + "github.com/shcizo/package-updater/internal/protocol" ) -// SwarmDockerClient is the subset of the Docker SDK SwarmExecutor depends -// on. Defined as an interface so tests can supply a fake. -type SwarmDockerClient interface { +// SwarmClient is the subset of the Docker SDK SwarmExecutor depends on. +// Defined as an interface so tests can supply a fake. +type SwarmClient interface { + Info(ctx context.Context) (system.Info, error) ServiceInspectWithRaw(ctx context.Context, serviceID string, opts swarm.ServiceInspectOptions) (swarm.Service, []byte, error) ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) } @@ -18,35 +23,71 @@ type SwarmDockerClient interface { // SwarmExecutor updates a Swarm service's image via the Docker API, // equivalent to `docker service update --image`. type SwarmExecutor struct { - cli SwarmDockerClient + cli SwarmClient + resolve *Resolver + auth *RegistryAuth } // NewSwarmExecutor returns an executor that drives Swarm service updates -// through the Docker API. -func NewSwarmExecutor(cli SwarmDockerClient) *SwarmExecutor { - return &SwarmExecutor{cli: cli} +// through the Docker API. auth may be nil (anonymous registry access). +func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *SwarmExecutor { + return &SwarmExecutor{cli: cli, resolve: resolve, auth: auth} } -// Execute inspects the service to get its current spec and version (required -// by the Docker API as an optimistic-concurrency token), sets the new image -// on the container spec, and calls ServiceUpdate. QueryRegistry is set so a -// floating tag (e.g. ":latest") resolves to a fresh digest and actually -// triggers a rolling update instead of being treated as unchanged. -func (e *SwarmExecutor) Execute(ctx context.Context, job discovery.Job) error { - if job.Refused { - return fmt.Errorf("refused: %s", job.RefusedReason) +// Execute checks that the local daemon is a manager of the job's cluster, +// pins job.Ref to its current digest and updates the service to it. The +// update is addressed by the ID and version of the service that was +// inspected, so it changes exactly that object. Success means the daemon +// accepted the new spec, not that the tasks have converged. It returns the +// pinned ref. +func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) { + info, err := e.cli.Info(ctx) + if err != nil { + return "", fmt.Errorf("docker info: %w", err) + } + if !info.Swarm.ControlAvailable { + return "", fmt.Errorf("refused: this agent's daemon is not a swarm manager") + } + actual := "" + if info.Swarm.Cluster != nil { + actual = info.Swarm.Cluster.ID + } + if actual != job.SwarmClusterID { + return "", fmt.Errorf("refused: job is for swarm cluster %s, this daemon is in %s", job.SwarmClusterID, actual) } - svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.ServiceID, swarm.ServiceInspectOptions{}) + svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.SwarmService, swarm.ServiceInspectOptions{}) if err != nil { - return fmt.Errorf("inspect service %s: %w", job.ServiceID, err) + return "", fmt.Errorf("inspect service %s: %w", job.SwarmService, err) } + if svc.Spec.Name != job.SwarmService { + return "", fmt.Errorf("inspect service %s: daemon returned service %q", job.SwarmService, svc.Spec.Name) + } + if svc.Spec.TaskTemplate.ContainerSpec == nil { + return "", fmt.Errorf("service %s has no container spec", job.SwarmService) + } + + pinned, err := e.resolve.Resolve(ctx, job.Ref) + if err != nil { + return "", err + } + fmt.Fprintf(out, "resolved %s\n", pinned) spec := svc.Spec - spec.TaskTemplate.ContainerSpec.Image = job.Image - - if _, err := e.cli.ServiceUpdate(ctx, job.ServiceID, svc.Version, spec, swarm.ServiceUpdateOptions{QueryRegistry: true}); err != nil { - return fmt.Errorf("update service %s: %w", job.ServiceID, err) + spec.TaskTemplate.ContainerSpec.Image = pinned + opts := swarm.ServiceUpdateOptions{QueryRegistry: true} + if enc := e.auth.For(imageref.Parse(job.Ref).Repo); enc != "" { + opts.EncodedRegistryAuth = enc + } else { + opts.RegistryAuthFrom = swarm.RegistryAuthFromSpec } - return nil + resp, err := e.cli.ServiceUpdate(ctx, svc.ID, svc.Version, spec, opts) + if err != nil { + return "", fmt.Errorf("update service %s: %w", job.SwarmService, err) + } + for _, w := range resp.Warnings { + fmt.Fprintf(out, "warning: %s\n", w) + } + fmt.Fprintf(out, "service %s updated\n", job.SwarmService) + return pinned, nil } diff --git a/internal/agent/updater/swarm_test.go b/internal/agent/updater/swarm_test.go index 253d139..d44ddfa 100644 --- a/internal/agent/updater/swarm_test.go +++ b/internal/agent/updater/swarm_test.go @@ -1,99 +1,142 @@ -package updater_test +package updater import ( + "bytes" "context" "errors" "testing" + "github.com/docker/docker/api/types/registry" "github.com/docker/docker/api/types/swarm" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/updater" + "github.com/docker/docker/api/types/system" + "github.com/opencontainers/go-digest" + "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/shcizo/package-updater/internal/protocol" ) -type fakeSwarmDockerClient struct { - inspectService swarm.Service - inspectErr error - updateErr error +type fakeSwarm struct { + info system.Info + svc swarm.Service + inspectErr error + updateErr error + warnings []string - inspectCalled bool - updateCalled bool - updateCalledServiceID string - updateCalledVersion swarm.Version - updateCalledSpec swarm.ServiceSpec - updateCalledOpts swarm.ServiceUpdateOptions + inspectCalls int + updateCalls int + updID string + updVersion swarm.Version + updSpec swarm.ServiceSpec + updOpts swarm.ServiceUpdateOptions } -func (f *fakeSwarmDockerClient) ServiceInspectWithRaw(_ context.Context, _ string, _ swarm.ServiceInspectOptions) (swarm.Service, []byte, error) { - f.inspectCalled = true - return f.inspectService, nil, f.inspectErr +func (f *fakeSwarm) Info(context.Context) (system.Info, error) { return f.info, nil } + +func (f *fakeSwarm) ServiceInspectWithRaw(context.Context, string, swarm.ServiceInspectOptions) (swarm.Service, []byte, error) { + f.inspectCalls++ + return f.svc, nil, f.inspectErr } -func (f *fakeSwarmDockerClient) ServiceUpdate(_ context.Context, serviceID string, version swarm.Version, spec swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) { - f.updateCalled = true - f.updateCalledServiceID = serviceID - f.updateCalledVersion = version - f.updateCalledSpec = spec - f.updateCalledOpts = opts - return swarm.ServiceUpdateResponse{}, f.updateErr +func (f *fakeSwarm) ServiceUpdate(_ context.Context, id string, v swarm.Version, spec swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) { + f.updateCalls++ + f.updID, f.updVersion, f.updSpec, f.updOpts = id, v, spec, opts + return swarm.ServiceUpdateResponse{Warnings: f.warnings}, f.updateErr } -func mkInspectService(version uint64, image string) swarm.Service { +type fakeDist struct { + digest string + err error + refs []string + auths []string +} + +func (f *fakeDist) DistributionInspect(_ context.Context, ref, auth string) (registry.DistributionInspect, error) { + f.refs = append(f.refs, ref) + f.auths = append(f.auths, auth) + var d registry.DistributionInspect + d.Descriptor.Digest = digest.Digest(f.digest) + return d, f.err +} + +func managerOf(cluster string) system.Info { + var info system.Info + info.Swarm.ControlAvailable = true + info.Swarm.Cluster = &swarm.ClusterInfo{ID: cluster} + return info +} + +func swarmSvc(name string) swarm.Service { return swarm.Service{ - ID: "svc-myapp", - Meta: swarm.Meta{Version: swarm.Version{Index: version}}, + ID: "svc-id", + Meta: swarm.Meta{Version: swarm.Version{Index: 7}}, Spec: swarm.ServiceSpec{ - Annotations: swarm.Annotations{Name: "myapp_web"}, - TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: image}}, + Annotations: swarm.Annotations{Name: name}, + TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "gitea/gitea:1.26"}}, }, } } -func TestSwarmExecutor_UpdatesImageAndVersion(t *testing.T) { - fake := &fakeSwarmDockerClient{inspectService: mkInspectService(7, "registry.example.com/myapp:v1")} - e := updater.NewSwarmExecutor(fake) +func TestSwarmExecutor(t *testing.T) { + job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"} + workerInfo := managerOf("c1") + workerInfo.Swarm.ControlAvailable = false + noContainer := swarmSvc("gitea_gitea") + noContainer.Spec.TaskTemplate.ContainerSpec = nil - job := discovery.Job{Service: "myapp_web", ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"} - err := e.Execute(context.Background(), job) - - require.NoError(t, err) - require.True(t, fake.updateCalled) - require.Equal(t, "svc-myapp", fake.updateCalledServiceID) - require.Equal(t, uint64(7), fake.updateCalledVersion.Index) - require.Equal(t, "registry.example.com/myapp:v2", fake.updateCalledSpec.TaskTemplate.ContainerSpec.Image) - require.True(t, fake.updateCalledOpts.QueryRegistry) -} - -func TestSwarmExecutor_InspectError(t *testing.T) { - fake := &fakeSwarmDockerClient{inspectErr: errors.New("service not found")} - e := updater.NewSwarmExecutor(fake) - - err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-missing", Image: "x:v2"}) - - require.Error(t, err) - require.False(t, fake.updateCalled) -} - -func TestSwarmExecutor_UpdateError(t *testing.T) { - fake := &fakeSwarmDockerClient{ - inspectService: mkInspectService(1, "registry.example.com/myapp:v1"), - updateErr: errors.New("update rejected"), + cases := []struct { + name string + fake *fakeSwarm + auth *RegistryAuth + wantErr string + wantUpdate bool + wantEncAuth bool + }{ + {name: "happy path, no auth", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")}, wantUpdate: true}, + {name: "happy path, auth known", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea"), warnings: []string{"w1"}}, + auth: &RegistryAuth{byHost: map[string]string{"docker.io": "enc"}}, wantUpdate: true, wantEncAuth: true}, + {name: "cluster mismatch", fake: &fakeSwarm{info: managerOf("other"), svc: swarmSvc("gitea_gitea")}, + wantErr: "refused: job is for swarm cluster c1, this daemon is in other"}, + {name: "worker node", fake: &fakeSwarm{info: workerInfo, svc: swarmSvc("gitea_gitea")}, wantErr: "not a swarm manager"}, + {name: "other service returned", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_other")}, wantErr: "daemon returned service"}, + {name: "no container spec", fake: &fakeSwarm{info: managerOf("c1"), svc: noContainer}, wantErr: "no container spec"}, + {name: "inspect fails", fake: &fakeSwarm{info: managerOf("c1"), inspectErr: errors.New("gone")}, wantErr: "gone"}, + {name: "update fails", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea"), updateErr: errors.New("rejected")}, + wantErr: "rejected", wantUpdate: true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + dist := &fakeDist{digest: "sha256:" + string(bytes.Repeat([]byte("a"), 64))} + e := NewSwarmExecutor(c.fake, NewResolver(dist, c.auth), c.auth) + var out bytes.Buffer + ref, err := e.Execute(context.Background(), job, &out) + if c.wantErr != "" { + assert.ErrorContains(t, err, c.wantErr) + } else { + require.NoError(t, err) + assert.Equal(t, "gitea/gitea:1.27@"+dist.digest, ref) + assert.Contains(t, out.String(), "resolved "+ref) + } + if c.wantErr == "refused: job is for swarm cluster c1, this daemon is in other" || c.wantErr == "not a swarm manager" { + assert.Zero(t, c.fake.inspectCalls) + } + if !c.wantUpdate { + assert.Zero(t, c.fake.updateCalls) + return + } + assert.Equal(t, 1, c.fake.updateCalls) + assert.Equal(t, "svc-id", c.fake.updID) + assert.Equal(t, uint64(7), c.fake.updVersion.Index) + assert.Equal(t, "gitea/gitea:1.27@"+dist.digest, c.fake.updSpec.TaskTemplate.ContainerSpec.Image) + assert.True(t, c.fake.updOpts.QueryRegistry) + if c.wantEncAuth { + assert.Equal(t, "enc", c.fake.updOpts.EncodedRegistryAuth) + assert.Empty(t, c.fake.updOpts.RegistryAuthFrom) + assert.Contains(t, out.String(), "warning: w1") + } else { + assert.Empty(t, c.fake.updOpts.EncodedRegistryAuth) + assert.Equal(t, swarm.RegistryAuthFromSpec, c.fake.updOpts.RegistryAuthFrom) + } + }) } - e := updater.NewSwarmExecutor(fake) - - err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"}) - - require.Error(t, err) -} - -func TestSwarmExecutor_RefusedJobIsNotExecuted(t *testing.T) { - fake := &fakeSwarmDockerClient{} - e := updater.NewSwarmExecutor(fake) - job := discovery.Job{ServiceID: "svc-myapp", Image: "x:v2", Refused: true, RefusedReason: "not opted in"} - - err := e.Execute(context.Background(), job) - - require.Error(t, err) - require.False(t, fake.inspectCalled) - require.False(t, fake.updateCalled) }