diff --git a/cmd/server/main.go b/cmd/server/main.go deleted file mode 100644 index 27c1288..0000000 --- a/cmd/server/main.go +++ /dev/null @@ -1,144 +0,0 @@ -package main - -import ( - "context" - "fmt" - "net/http" - "os" - "os/signal" - "syscall" - "time" - - "github.com/docker/docker/client" - "github.com/prometheus/client_golang/prometheus" - "github.com/shcizo/package-updater/internal/api" - "github.com/shcizo/package-updater/internal/config" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/logging" - "github.com/shcizo/package-updater/internal/metrics" - "github.com/shcizo/package-updater/internal/updater" -) - -// Build info is injected at link time via -ldflags="-X main.version=..." -var ( - version = "dev" - commit = "unknown" - buildTime = "unknown" -) - -func main() { - if err := run(); err != nil { - fmt.Fprintln(os.Stderr, "fatal:", err) - os.Exit(1) - } -} - -func run() error { - cfg, err := config.Load() - if err != nil { - return err - } - - logger := logging.New(cfg.LogLevel) - logger.Info("starting", - "version", version, "commit", commit, "port", cfg.Port, - "mode", cfg.Mode, "stacks_root", cfg.StacksRoot, "opt_in_label", cfg.OptInLabel, - ) - - dockerCli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation()) - if err != nil { - return fmt.Errorf("docker client: %w", err) - } - defer dockerCli.Close() - - var finder api.Finder - var exec updater.Executor - switch cfg.Mode { - case "swarm": - finder = discovery.NewSwarm(dockerCli, cfg.OptInLabel) - exec = updater.NewSwarmExecutor(dockerCli) - case "compose": - finder = discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel) - exec = updater.NewComposeExecutor() - default: - // Unreachable: config.Load already validates Mode, but guard - // here too rather than silently falling through to a nil - // finder/exec pair. - return fmt.Errorf("unknown MODE %q", cfg.Mode) - } - - reg := prometheus.NewRegistry() - m := metrics.New(reg) - m.BuildInfo.WithLabelValues(version, commit).Set(1) - - queue := updater.NewQueue(exec, m) - queue.Start(context.Background()) - defer queue.Stop() - - handlers := api.NewHandlers(finder, &submitterAdapter{queue: queue, timeout: cfg.UpdateTimeout}, dockerCli, version, commit, buildTime, m) - - mux := http.NewServeMux() - mux.HandleFunc("POST /update", handlers.Update) - mux.HandleFunc("GET /healthz", handlers.Healthz) - mux.HandleFunc("GET /version", handlers.Version) - mux.Handle("GET /metrics", metrics.Handler(reg)) - - authed := api.Auth(cfg.APIKey) - handler := api.RequestID(api.RequestLogger(logger, m)(routeAuth(mux, authed))) - - srv := &http.Server{ - Addr: ":" + cfg.Port, - Handler: handler, - ReadHeaderTimeout: 10 * time.Second, - } - - errCh := make(chan error, 1) - go func() { - logger.Info("listening", "addr", srv.Addr) - errCh <- srv.ListenAndServe() - }() - - sig := make(chan os.Signal, 1) - signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) - - select { - case s := <-sig: - logger.Info("shutdown_signal", "signal", s.String()) - case err := <-errCh: - if err != nil && err != http.ErrServerClosed { - return err - } - } - - shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - _ = srv.Shutdown(shutdownCtx) - return nil -} - -// routeAuth applies the bearer-token middleware to /update only. -// /healthz, /metrics, /version are unauthenticated by design (internal -// network only; healthcheck and Prometheus scraper need to reach them -// without secrets). -func routeAuth(next *http.ServeMux, mw func(http.Handler) http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - if r.URL.Path == "/update" { - mw(next).ServeHTTP(w, r) - return - } - next.ServeHTTP(w, r) - }) -} - -// submitterAdapter wraps Queue.Submit to inject the per-request timeout -// from config, satisfying the api.Submitter interface. -type submitterAdapter struct { - queue *updater.Queue - timeout time.Duration -} - -func (s *submitterAdapter) Submit(ctx context.Context, jobs []discovery.Job) []updater.Result { - ctx, cancel := context.WithTimeout(ctx, s.timeout) - defer cancel() - return s.queue.Submit(ctx, jobs) -} diff --git a/internal/agent/config/config.go b/internal/agent/config/config.go index 1b918d9..2b2620a 100644 --- a/internal/agent/config/config.go +++ b/internal/agent/config/config.go @@ -2,10 +2,14 @@ package config import ( + "encoding/json" "errors" + "fmt" "os" + "path/filepath" "regexp" "strings" + "time" ) var masterKeyPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) @@ -17,8 +21,17 @@ type Config struct { Token string // AGENT_TOKEN, shown once when the agent was added SelfID string // SELF_ID, defaults to the hostname (= container ID prefix) LogLevel string // LOG_LEVEL, default info + + // UpdateTimeout bounds one update job (UPDATE_TIMEOUT, default 5m). + UpdateTimeout time.Duration + // DockerConfig is the directory holding config.json with registry + // credentials (DOCKER_CONFIG). Empty means anonymous registry access. + DockerConfig string } +// defaultUpdateTimeout is used when UPDATE_TIMEOUT is unset. +const defaultUpdateTimeout = 5 * time.Minute + // Load reads and validates the agent configuration. It fails fast so a // misconfigured agent never starts dialling. func Load() (*Config, error) { @@ -28,6 +41,9 @@ func Load() (*Config, error) { Token: os.Getenv("AGENT_TOKEN"), SelfID: os.Getenv("SELF_ID"), LogLevel: os.Getenv("LOG_LEVEL"), + + UpdateTimeout: defaultUpdateTimeout, + DockerConfig: os.Getenv("DOCKER_CONFIG"), } switch { case cfg.MasterURL == "": @@ -41,6 +57,18 @@ func Load() (*Config, error) { case cfg.Token == "": return nil, errors.New("AGENT_TOKEN is required") } + if v := os.Getenv("UPDATE_TIMEOUT"); v != "" { + d, err := time.ParseDuration(v) + if err != nil || d <= 0 { + return nil, fmt.Errorf("UPDATE_TIMEOUT must be a positive duration such as 5m, got %q", v) + } + cfg.UpdateTimeout = d + } + if cfg.DockerConfig != "" { + if err := checkDockerConfig(cfg.DockerConfig); err != nil { + return nil, err + } + } if cfg.SelfID == "" { cfg.SelfID, _ = os.Hostname() } @@ -49,3 +77,17 @@ func Load() (*Config, error) { } return cfg, nil } + +// checkDockerConfig fails when DOCKER_CONFIG is set but its config.json is +// missing or not JSON, so a typo does not silently turn into anonymous pulls. +func checkDockerConfig(dir string) error { + path := filepath.Join(dir, "config.json") + raw, err := os.ReadFile(path) + if err != nil { + return fmt.Errorf("DOCKER_CONFIG: read %s: %w", path, err) + } + if !json.Valid(raw) { + return fmt.Errorf("DOCKER_CONFIG: %s is not valid JSON", path) + } + return nil +} diff --git a/internal/agent/config/config_test.go b/internal/agent/config/config_test.go index 606018e..6aac5b2 100644 --- a/internal/agent/config/config_test.go +++ b/internal/agent/config/config_test.go @@ -2,7 +2,9 @@ package config import ( "os" + "path/filepath" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -16,6 +18,8 @@ func setValid(t *testing.T) { t.Setenv("AGENT_TOKEN", "pua_abc") t.Setenv("SELF_ID", "") t.Setenv("LOG_LEVEL", "") + t.Setenv("UPDATE_TIMEOUT", "") + t.Setenv("DOCKER_CONFIG", "") } func TestLoad_Valid(t *testing.T) { @@ -25,7 +29,7 @@ func TestLoad_Valid(t *testing.T) { host, _ := os.Hostname() assert.Equal(t, &Config{ MasterURL: "wss://10.0.0.5:8090/agent", MasterKey: validKey, Token: "pua_abc", - SelfID: host, LogLevel: "info", + SelfID: host, LogLevel: "info", UpdateTimeout: 5 * time.Minute, }, cfg) } @@ -48,3 +52,45 @@ func TestLoad_Invalid(t *testing.T) { }) } } + +func TestLoad_JobSettings(t *testing.T) { + withConfig := func(t *testing.T, body string) string { + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(body), 0o600)) + return dir + } + cases := []struct { + name string + timeout string + dockerCfg func(t *testing.T) string + wantTime time.Duration + wantErr string + }{ + {name: "defaults", wantTime: 5 * time.Minute}, + {name: "custom timeout", timeout: "90s", wantTime: 90 * time.Second}, + {name: "zero timeout", timeout: "0", wantErr: "UPDATE_TIMEOUT"}, + {name: "garbage timeout", timeout: "abc", wantErr: "UPDATE_TIMEOUT"}, + {name: "docker config without file", dockerCfg: func(t *testing.T) string { return t.TempDir() }, wantErr: "DOCKER_CONFIG"}, + {name: "docker config not json", dockerCfg: func(t *testing.T) string { return withConfig(t, "nope") }, wantErr: "not valid JSON"}, + {name: "docker config ok", dockerCfg: func(t *testing.T) string { return withConfig(t, `{"auths":{}}`) }, wantTime: 5 * time.Minute}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + setValid(t) + t.Setenv("UPDATE_TIMEOUT", c.timeout) + dir := "" + if c.dockerCfg != nil { + dir = c.dockerCfg(t) + t.Setenv("DOCKER_CONFIG", dir) + } + cfg, err := Load() + if c.wantErr != "" { + assert.ErrorContains(t, err, c.wantErr) + return + } + require.NoError(t, err) + assert.Equal(t, c.wantTime, cfg.UpdateTimeout) + assert.Equal(t, dir, cfg.DockerConfig) + }) + } +} diff --git a/internal/agent/inventory/collector.go b/internal/agent/inventory/collector.go index 2ee81d1..fbe36bf 100644 --- a/internal/agent/inventory/collector.go +++ b/internal/agent/inventory/collector.go @@ -18,10 +18,11 @@ import ( ) const ( - labelProject = "com.docker.compose.project" - labelService = "com.docker.compose.service" - labelWorkingDir = "com.docker.compose.project.working_dir" - labelSwarmTask = "com.docker.swarm.service.id" + labelProject = "com.docker.compose.project" + labelService = "com.docker.compose.service" + labelWorkingDir = "com.docker.compose.project.working_dir" + labelConfigFiles = "com.docker.compose.project.config_files" + labelSwarmTask = "com.docker.swarm.service.id" ) // DockerClient is the subset of the Docker SDK the collector needs. @@ -148,6 +149,7 @@ func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests svc.ComposeProject = project svc.ComposeService = service svc.WorkingDir = ct.Labels[labelWorkingDir] + svc.ConfigFiles = ct.Labels[labelConfigFiles] return svc } svc.Kind = protocol.KindContainer diff --git a/internal/agent/inventory/collector_test.go b/internal/agent/inventory/collector_test.go index b300de5..ff464ff 100644 --- a/internal/agent/inventory/collector_test.go +++ b/internal/agent/inventory/collector_test.go @@ -200,3 +200,12 @@ func TestHello(t *testing.T) { require.NoError(t, err) assert.Equal(t, protocol.Hello{Hostname: "swarm-mgr", SwarmManager: true, SwarmClusterID: "cluster-1"}, got) } + +func TestSnapshot_ComposeConfigFilesIsRawLabel(t *testing.T) { + ct := composeContainer("abc123", "slusk", "web", "slusk:1", "sha256:img1", "running") + ct.Labels[labelConfigFiles] = "/s/a/compose.yaml,/s/a/compose.override.yaml" + got, err := New(&fakeDocker{containers: []container.Summary{ct}}, "").Snapshot(context.Background()) + require.NoError(t, err) + require.Len(t, got, 1) + assert.Equal(t, "/s/a/compose.yaml,/s/a/compose.override.yaml", got[0].ConfigFiles) +} diff --git a/internal/updater/compose_executor.go b/internal/agent/updater/compose_executor.go similarity index 100% rename from internal/updater/compose_executor.go rename to internal/agent/updater/compose_executor.go diff --git a/internal/discovery/pathcheck.go b/internal/agent/updater/pathcheck.go similarity index 100% rename from internal/discovery/pathcheck.go rename to internal/agent/updater/pathcheck.go diff --git a/internal/discovery/pathcheck_test.go b/internal/agent/updater/pathcheck_test.go similarity index 100% rename from internal/discovery/pathcheck_test.go rename to internal/agent/updater/pathcheck_test.go diff --git a/internal/updater/swarm_executor.go b/internal/agent/updater/swarm_executor.go similarity index 100% rename from internal/updater/swarm_executor.go rename to internal/agent/updater/swarm_executor.go diff --git a/internal/updater/swarm_executor_test.go b/internal/agent/updater/swarm_executor_test.go similarity index 100% rename from internal/updater/swarm_executor_test.go rename to internal/agent/updater/swarm_executor_test.go diff --git a/internal/api/handlers.go b/internal/api/handlers.go deleted file mode 100644 index 420ff35..0000000 --- a/internal/api/handlers.go +++ /dev/null @@ -1,148 +0,0 @@ -package api - -import ( - "context" - "encoding/json" - "net/http" - - "github.com/docker/docker/api/types" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/logging" - "github.com/shcizo/package-updater/internal/metrics" - "github.com/shcizo/package-updater/internal/updater" -) - -// Finder is the discovery interface used by the update handler. -type Finder interface { - FindJobs(ctx context.Context, image string) ([]discovery.Job, error) -} - -// Submitter is the queue interface used by the update handler. -type Submitter interface { - Submit(ctx context.Context, jobs []discovery.Job) []updater.Result -} - -// Pinger pings the Docker daemon for the healthcheck. -type Pinger interface { - Ping(ctx context.Context) (types.Ping, error) -} - -// Handlers wires the HTTP endpoints to the rest of the service. -type Handlers struct { - finder Finder - submitter Submitter - pinger Pinger - version string - commit string - buildTime string - metrics *metrics.Metrics -} - -// NewHandlers constructs a Handlers value with all dependencies injected. -// m may be nil, in which case metrics recording is silently skipped. -func NewHandlers(f Finder, s Submitter, p Pinger, version, commit, buildTime string, m *metrics.Metrics) *Handlers { - return &Handlers{finder: f, submitter: s, pinger: p, version: version, commit: commit, buildTime: buildTime, metrics: m} -} - -// Update implements POST /update. -func (h *Handlers) Update(w http.ResponseWriter, r *http.Request) { - var req UpdateRequest - if err := json.NewDecoder(r.Body).Decode(&req); err != nil { - writeJSONError(w, http.StatusBadRequest, "invalid JSON body") - return - } - if req.Image == "" { - writeJSONError(w, http.StatusBadRequest, "image is required") - return - } - - requestedImage := req.Image - if req.Tag != "" { - requestedImage = req.Image + ":" + req.Tag - } - - jobs, err := h.finder.FindJobs(r.Context(), requestedImage) - if err != nil { - writeJSONError(w, http.StatusInternalServerError, "discovery failed: "+err.Error()) - return - } - - resp := UpdateResponse{ - RequestID: logging.RequestIDFrom(r.Context()), - Image: req.Image, - Tag: req.Tag, - Matched: len(jobs), - Results: []ResultDTO{}, - } - - if len(jobs) == 0 { - writeJSON(w, http.StatusOK, resp) - return - } - - results := h.submitter.Submit(r.Context(), jobs) - updated, failed := 0, 0 - for _, res := range results { - composeFile := "" - if len(res.Job.ConfigFiles) > 0 { - composeFile = res.Job.ConfigFiles[0] - } - resp.Results = append(resp.Results, ResultDTO{ - Project: res.Job.Project, - Service: res.Job.Service, - ComposeFile: composeFile, - ServiceID: res.Job.ServiceID, - Status: string(res.Status), - Error: res.Error, - DurationMs: res.DurationMs, - }) - if res.Status == updater.StatusUpdated { - updated++ - } else { - failed++ - } - } - - switch { - case failed == 0: - writeJSON(w, http.StatusOK, resp) - case updated == 0: - writeJSON(w, http.StatusInternalServerError, resp) - default: - writeJSON(w, http.StatusMultiStatus, resp) - } -} - -// Healthz implements GET /healthz. -func (h *Handlers) Healthz(w http.ResponseWriter, r *http.Request) { - if _, err := h.pinger.Ping(r.Context()); err != nil { - if h.metrics != nil { - h.metrics.DockerPingUp.Set(0) - } - writeJSON(w, http.StatusServiceUnavailable, HealthResponse{Status: "unhealthy", Docker: "unreachable"}) - return - } - if h.metrics != nil { - h.metrics.DockerPingUp.Set(1) - } - writeJSON(w, http.StatusOK, HealthResponse{Status: "ok", Docker: "ok"}) -} - -// Version implements GET /version. -func (h *Handlers) Version(w http.ResponseWriter, _ *http.Request) { - writeJSON(w, http.StatusOK, VersionResponse{ - Version: h.version, - Commit: h.commit, - BuildTime: h.buildTime, - }) -} - -func writeJSON(w http.ResponseWriter, code int, body any) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(code) - _ = json.NewEncoder(w).Encode(body) -} - -func writeJSONError(w http.ResponseWriter, code int, msg string) { - writeJSON(w, code, map[string]string{"error": msg}) -} diff --git a/internal/api/handlers_test.go b/internal/api/handlers_test.go deleted file mode 100644 index 462540a..0000000 --- a/internal/api/handlers_test.go +++ /dev/null @@ -1,216 +0,0 @@ -package api_test - -import ( - "bytes" - "context" - "encoding/json" - "errors" - "io" - "net/http" - "net/http/httptest" - "strings" - "testing" - - "github.com/docker/docker/api/types" - "github.com/shcizo/package-updater/internal/api" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/updater" - "github.com/stretchr/testify/require" -) - -type fakeFinder struct { - jobs []discovery.Job - err error - - // gotImage records the image argument passed to FindJobs, for - // assertions on the exact reference the handler built. - gotImage string - // swarmShaped, when true, makes FindJobs return a Job whose Image - // field carries the received image argument, simulating - // discovery.SwarmDiscovery.FindJobs. - swarmShaped bool -} - -func (f *fakeFinder) FindJobs(_ context.Context, image string) ([]discovery.Job, error) { - f.gotImage = image - if f.swarmShaped { - return []discovery.Job{{Service: "myapp_web", ServiceID: "svc1", Image: image}}, f.err - } - return f.jobs, f.err -} - -type fakeSubmitter struct { - results []updater.Result -} - -func (f *fakeSubmitter) Submit(_ context.Context, jobs []discovery.Job) []updater.Result { - if f.results != nil { - return f.results - } - out := make([]updater.Result, len(jobs)) - for i, j := range jobs { - out[i] = updater.Result{Job: j, Status: updater.StatusUpdated} - } - return out -} - -type fakePinger struct{ err error } - -func (f *fakePinger) Ping(_ context.Context) (types.Ping, error) { - return types.Ping{}, f.err -} - -func decode[T any](t *testing.T, body io.Reader) T { - t.Helper() - var v T - require.NoError(t, json.NewDecoder(body).Decode(&v)) - return v -} - -func TestUpdate_ValidationError(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - req := httptest.NewRequest(http.MethodPost, "/update", - strings.NewReader(`{"tag":"v1.2.3"}`)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusBadRequest, w.Code) -} - -func TestUpdate_BadJSON(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`not json`)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusBadRequest, w.Code) -} - -func TestUpdate_DiscoveryFailureReturns500(t *testing.T) { - finder := &fakeFinder{err: errors.New("daemon unreachable")} - h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusInternalServerError, w.Code) -} - -func TestUpdate_ZeroMatchesReturns200(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusOK, w.Code) - resp := decode[api.UpdateResponse](t, w.Body) - require.Equal(t, 0, resp.Matched) -} - -func TestUpdate_AllSucceeded200(t *testing.T) { - finder := &fakeFinder{jobs: []discovery.Job{ - {Project: "p", Service: "s", WorkingDir: "/x", ConfigFiles: []string{"/x/c.yml"}}, - }} - h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "r/x", Tag: "v1"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusOK, w.Code) - resp := decode[api.UpdateResponse](t, w.Body) - require.Equal(t, 1, resp.Matched) - require.Equal(t, "updated", resp.Results[0].Status) - require.Equal(t, "/x/c.yml", resp.Results[0].ComposeFile) -} - -func TestUpdate_MixedReturns207(t *testing.T) { - jobs := []discovery.Job{ - {Project: "p1", Service: "s", ConfigFiles: []string{"/x/c.yml"}}, - {Project: "p2", Service: "s", ConfigFiles: []string{"/y/c.yml"}}, - } - finder := &fakeFinder{jobs: jobs} - submitter := &fakeSubmitter{results: []updater.Result{ - {Job: jobs[0], Status: updater.StatusUpdated}, - {Job: jobs[1], Status: updater.StatusFailed, Error: "boom"}, - }} - h := api.NewHandlers(finder, submitter, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusMultiStatus, w.Code) -} - -func TestUpdate_AllFailedReturns500(t *testing.T) { - jobs := []discovery.Job{{Project: "p", Service: "s", ConfigFiles: []string{"/x/c.yml"}}} - finder := &fakeFinder{jobs: jobs} - submitter := &fakeSubmitter{results: []updater.Result{ - {Job: jobs[0], Status: updater.StatusFailed, Error: "boom"}, - }} - h := api.NewHandlers(finder, submitter, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "r/x"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusInternalServerError, w.Code) -} - -func TestHealthz_OKWhenDockerUp(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - req := httptest.NewRequest(http.MethodGet, "/healthz", nil) - w := httptest.NewRecorder() - h.Healthz(w, req) - require.Equal(t, http.StatusOK, w.Code) -} - -func TestHealthz_503WhenDockerDown(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{err: errors.New("ping fail")}, "v0.0.0", "abc", "now", nil) - req := httptest.NewRequest(http.MethodGet, "/healthz", nil) - w := httptest.NewRecorder() - h.Healthz(w, req) - require.Equal(t, http.StatusServiceUnavailable, w.Code) -} - -func TestUpdate_FindJobsCalledWithTaggedImage(t *testing.T) { - finder := &fakeFinder{} - h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage) -} - -func TestUpdate_FindJobsCalledWithBareImageWhenNoTag(t *testing.T) { - finder := &fakeFinder{} - h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, "registry.example.com/myapp", finder.gotImage) -} - -func TestUpdate_SwarmModeThreadsTagThroughDiscovery(t *testing.T) { - finder := &fakeFinder{swarmShaped: true} - h := api.NewHandlers(finder, &fakeSubmitter{}, &fakePinger{}, "v0.0.0", "abc", "now", nil) - body, _ := json.Marshal(api.UpdateRequest{Image: "registry.example.com/myapp", Tag: "v2"}) - req := httptest.NewRequest(http.MethodPost, "/update", bytes.NewReader(body)) - w := httptest.NewRecorder() - h.Update(w, req) - require.Equal(t, http.StatusOK, w.Code) - // The fake Finder's returned Job.Image (as SwarmDiscovery would build - // it) reflects the image argument it received from the handler. - // Asserting on the captured argument proves the full request -> - // discovery flow carries the tag through to what would reach the - // SwarmExecutor. - require.Equal(t, "registry.example.com/myapp:v2", finder.gotImage) -} - -func TestVersion(t *testing.T) { - h := api.NewHandlers(&fakeFinder{}, &fakeSubmitter{}, &fakePinger{}, "v1.2.3", "abcdef", "2026-05-22T00:00:00Z", nil) - req := httptest.NewRequest(http.MethodGet, "/version", nil) - w := httptest.NewRecorder() - h.Version(w, req) - require.Equal(t, http.StatusOK, w.Code) - resp := decode[api.VersionResponse](t, w.Body) - require.Equal(t, "v1.2.3", resp.Version) -} diff --git a/internal/api/middleware.go b/internal/api/middleware.go deleted file mode 100644 index 8564890..0000000 --- a/internal/api/middleware.go +++ /dev/null @@ -1,112 +0,0 @@ -// Package api contains HTTP handlers, middleware, and request/response DTOs. -package api - -import ( - "crypto/rand" - "crypto/subtle" - "fmt" - "log/slog" - "net/http" - "strconv" - "strings" - "time" - - "github.com/shcizo/package-updater/internal/logging" - "github.com/shcizo/package-updater/internal/metrics" -) - -// Auth returns middleware that requires a matching bearer token. -// Compares with constant-time to defeat timing attacks. -func Auth(token string) func(http.Handler) http.Handler { - tokenBytes := []byte(token) - return func(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - h := r.Header.Get("Authorization") - const prefix = "Bearer " - if !strings.HasPrefix(h, prefix) { - writeAuthError(w) - return - } - provided := []byte(strings.TrimPrefix(h, prefix)) - if len(provided) == 0 || - subtle.ConstantTimeCompare(provided, tokenBytes) != 1 { - writeAuthError(w) - return - } - next.ServeHTTP(w, r) - }) - } -} - -func writeAuthError(w http.ResponseWriter) { - w.Header().Set("Content-Type", "application/json") - w.WriteHeader(http.StatusUnauthorized) - _, _ = w.Write([]byte(`{"error":"unauthorized"}`)) -} - -// newUUID generates a random UUID v4 string using crypto/rand. -func newUUID() string { - var b [16]byte - _, _ = rand.Read(b[:]) - b[6] = (b[6] & 0x0f) | 0x40 // version 4 - b[8] = (b[8] & 0x3f) | 0x80 // variant bits - return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", - b[0:4], b[4:6], b[6:8], b[8:10], b[10:16]) -} - -// RequestID middleware ensures every request has an X-Request-ID -// header (generated if absent) and stores it in the request context. -func RequestID(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - id := r.Header.Get("X-Request-ID") - if id == "" { - id = newUUID() - } - w.Header().Set("X-Request-ID", id) - ctx := logging.WithRequestID(r.Context(), id) - next.ServeHTTP(w, r.WithContext(ctx)) - }) -} - -// RequestLogger emits a structured access-log line per request. m may be nil, -// in which case metrics recording is silently skipped. -func RequestLogger(base *slog.Logger, m *metrics.Metrics) func(http.Handler) http.Handler { - return func(next http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - start := time.Now() - sw := &statusWriter{ResponseWriter: w, status: http.StatusOK} - next.ServeHTTP(sw, r) - logger := logging.FromContext(r.Context(), base) - logger.Info("http_request", - "method", r.Method, - "path", r.URL.Path, - "status", sw.status, - "duration_ms", time.Since(start).Milliseconds(), - "client_ip", clientIP(r), - ) - if m != nil { - m.HTTPRequests.WithLabelValues(r.URL.Path, strconv.Itoa(sw.status)).Inc() - } - }) - } -} - -type statusWriter struct { - http.ResponseWriter - status int -} - -func (s *statusWriter) WriteHeader(code int) { - s.status = code - s.ResponseWriter.WriteHeader(code) -} - -func clientIP(r *http.Request) string { - if xff := r.Header.Get("X-Forwarded-For"); xff != "" { - if comma := strings.Index(xff, ","); comma >= 0 { - return strings.TrimSpace(xff[:comma]) - } - return xff - } - return r.RemoteAddr -} diff --git a/internal/api/middleware_test.go b/internal/api/middleware_test.go deleted file mode 100644 index e973aad..0000000 --- a/internal/api/middleware_test.go +++ /dev/null @@ -1,92 +0,0 @@ -package api_test - -import ( - "io" - "log/slog" - "net/http" - "net/http/httptest" - "testing" - - "github.com/shcizo/package-updater/internal/api" - "github.com/shcizo/package-updater/internal/logging" - "github.com/stretchr/testify/require" -) - -func newTestLogger() *slog.Logger { - return slog.New(slog.NewJSONHandler(io.Discard, nil)) -} - -func TestAuth_AllowsMatchingToken(t *testing.T) { - called := false - h := api.Auth("secret")(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - called = true - w.WriteHeader(http.StatusOK) - })) - req := httptest.NewRequest(http.MethodPost, "/update", nil) - req.Header.Set("Authorization", "Bearer secret") - w := httptest.NewRecorder() - h.ServeHTTP(w, req) - require.True(t, called) - require.Equal(t, http.StatusOK, w.Code) -} - -func TestAuth_Rejects(t *testing.T) { - cases := []struct{ name, header string }{ - {"missing", ""}, - {"wrong scheme", "Token secret"}, - {"wrong value", "Bearer nope"}, - {"empty bearer", "Bearer "}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - h := api.Auth("secret")(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - t.Fatal("handler must not be called") - })) - req := httptest.NewRequest(http.MethodPost, "/update", nil) - if c.header != "" { - req.Header.Set("Authorization", c.header) - } - w := httptest.NewRecorder() - h.ServeHTTP(w, req) - require.Equal(t, http.StatusUnauthorized, w.Code) - }) - } -} - -func TestRequestID_GeneratesIfMissing(t *testing.T) { - var seenID string - h := api.RequestID(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { - seenID = logging.RequestIDFrom(r.Context()) - })) - req := httptest.NewRequest(http.MethodPost, "/update", nil) - w := httptest.NewRecorder() - h.ServeHTTP(w, req) - require.NotEmpty(t, seenID) - require.Equal(t, seenID, w.Header().Get("X-Request-ID")) -} - -func TestRequestID_UsesIncoming(t *testing.T) { - var seenID string - h := api.RequestID(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { - seenID = logging.RequestIDFrom(r.Context()) - })) - req := httptest.NewRequest(http.MethodPost, "/update", nil) - req.Header.Set("X-Request-ID", "given-id") - w := httptest.NewRecorder() - h.ServeHTTP(w, req) - require.Equal(t, "given-id", seenID) -} - -func TestRequestLogger_LogsAndDelegates(t *testing.T) { - logger := newTestLogger() - called := false - h := api.RequestLogger(logger, nil)(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { - called = true - w.WriteHeader(http.StatusTeapot) - })) - req := httptest.NewRequest(http.MethodPost, "/update", nil) - w := httptest.NewRecorder() - h.ServeHTTP(w, req) - require.True(t, called) - require.Equal(t, http.StatusTeapot, w.Code) -} diff --git a/internal/api/swarm_result_test.go b/internal/api/swarm_result_test.go deleted file mode 100644 index e6d978b..0000000 --- a/internal/api/swarm_result_test.go +++ /dev/null @@ -1,48 +0,0 @@ -package api_test - -import ( - "context" - "net/http" - "net/http/httptest" - "strings" - "testing" - - "github.com/docker/docker/api/types" - "github.com/shcizo/package-updater/internal/api" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/updater" - "github.com/stretchr/testify/require" -) - -type swarmResultFinder struct{ jobs []discovery.Job } - -func (f *swarmResultFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) { - return f.jobs, nil -} - -type swarmResultSubmitter struct{ results []updater.Result } - -func (s *swarmResultSubmitter) Submit(_ context.Context, _ []discovery.Job) []updater.Result { - return s.results -} - -type swarmResultPinger struct{} - -func (swarmResultPinger) Ping(_ context.Context) (types.Ping, error) { return types.Ping{}, nil } - -func TestUpdate_SwarmJobPopulatesServiceID(t *testing.T) { - job := discovery.Job{Service: "myapp_web", ServiceID: "svc123", Image: "registry.example.com/myapp:v2"} - finder := &swarmResultFinder{jobs: []discovery.Job{job}} - submitter := &swarmResultSubmitter{results: []updater.Result{ - {Job: job, Status: updater.StatusUpdated, DurationMs: 42}, - }} - h := api.NewHandlers(finder, submitter, swarmResultPinger{}, "v", "c", "b", nil) - - req := httptest.NewRequest(http.MethodPost, "/update", strings.NewReader(`{"image":"registry.example.com/myapp"}`)) - rec := httptest.NewRecorder() - h.Update(rec, req) - - require.Equal(t, http.StatusOK, rec.Code) - require.Contains(t, rec.Body.String(), `"service_id":"svc123"`) - require.Contains(t, rec.Body.String(), `"service":"myapp_web"`) -} diff --git a/internal/api/types.go b/internal/api/types.go deleted file mode 100644 index 02f1f63..0000000 --- a/internal/api/types.go +++ /dev/null @@ -1,40 +0,0 @@ -package api - -// UpdateRequest is the body of POST /update. -type UpdateRequest struct { - Image string `json:"image"` - Tag string `json:"tag,omitempty"` -} - -// UpdateResponse is the body of POST /update. -type UpdateResponse struct { - RequestID string `json:"request_id"` - Image string `json:"image"` - Tag string `json:"tag,omitempty"` - Matched int `json:"matched"` - Results []ResultDTO `json:"results"` -} - -// ResultDTO is one row in UpdateResponse.Results. -type ResultDTO struct { - Project string `json:"project"` - Service string `json:"service"` - ComposeFile string `json:"compose_file"` - ServiceID string `json:"service_id,omitempty"` - Status string `json:"status"` - Error string `json:"error,omitempty"` - DurationMs int64 `json:"duration_ms"` -} - -// HealthResponse is the body of GET /healthz. -type HealthResponse struct { - Status string `json:"status"` - Docker string `json:"docker"` -} - -// VersionResponse is the body of GET /version. -type VersionResponse struct { - Version string `json:"version"` - Commit string `json:"commit"` - BuildTime string `json:"build_time"` -} diff --git a/internal/config/config.go b/internal/config/config.go deleted file mode 100644 index b143434..0000000 --- a/internal/config/config.go +++ /dev/null @@ -1,58 +0,0 @@ -// Package config loads service configuration from environment variables. -package config - -import ( - "errors" - "fmt" - "os" - "time" -) - -// Config holds all runtime configuration for the service. -type Config struct { - APIKey string - StacksRoot string - Port string - LogLevel string - UpdateTimeout time.Duration - OptInLabel string - Mode string -} - -// Load reads configuration from environment variables, applies defaults, -// and validates required fields. Returns an error if validation fails so -// the service can fail-fast at startup. -func Load() (*Config, error) { - cfg := &Config{ - APIKey: os.Getenv("UPDATER_API_KEY"), - StacksRoot: getenvDefault("STACKS_ROOT", "/home/shcizo/self-hosted"), - Port: getenvDefault("PORT", "8080"), - LogLevel: getenvDefault("LOG_LEVEL", "info"), - OptInLabel: getenvDefault("OPT_IN_LABEL", "se.shcizo.auto-update"), - Mode: getenvDefault("MODE", "compose"), - } - - if cfg.APIKey == "" { - return nil, errors.New("UPDATER_API_KEY is required") - } - - if cfg.Mode != "compose" && cfg.Mode != "swarm" { - return nil, fmt.Errorf("MODE %q is invalid: must be %q or %q", cfg.Mode, "compose", "swarm") - } - - timeoutStr := getenvDefault("UPDATE_TIMEOUT", "5m") - d, err := time.ParseDuration(timeoutStr) - if err != nil { - return nil, fmt.Errorf("UPDATE_TIMEOUT %q is not a valid duration: %w", timeoutStr, err) - } - cfg.UpdateTimeout = d - - return cfg, nil -} - -func getenvDefault(key, fallback string) string { - if v := os.Getenv(key); v != "" { - return v - } - return fallback -} diff --git a/internal/config/config_test.go b/internal/config/config_test.go deleted file mode 100644 index 6e3fe88..0000000 --- a/internal/config/config_test.go +++ /dev/null @@ -1,82 +0,0 @@ -package config_test - -import ( - "testing" - "time" - - "github.com/shcizo/package-updater/internal/config" - "github.com/stretchr/testify/require" -) - -func TestLoad_RequiresAPIKey(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "") - _, err := config.Load() - require.Error(t, err) - require.Contains(t, err.Error(), "UPDATER_API_KEY") -} - -func TestLoad_AppliesDefaults(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - t.Setenv("STACKS_ROOT", "") - t.Setenv("PORT", "") - t.Setenv("LOG_LEVEL", "") - t.Setenv("UPDATE_TIMEOUT", "") - t.Setenv("OPT_IN_LABEL", "") - - cfg, err := config.Load() - require.NoError(t, err) - require.Equal(t, "secret", cfg.APIKey) - require.Equal(t, "/home/shcizo/self-hosted", cfg.StacksRoot) - require.Equal(t, "8080", cfg.Port) - require.Equal(t, "info", cfg.LogLevel) - require.Equal(t, 5*time.Minute, cfg.UpdateTimeout) - require.Equal(t, "se.shcizo.auto-update", cfg.OptInLabel) -} - -func TestLoad_OverridesViaEnv(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - t.Setenv("STACKS_ROOT", "/srv/stacks") - t.Setenv("PORT", "9090") - t.Setenv("LOG_LEVEL", "debug") - t.Setenv("UPDATE_TIMEOUT", "30s") - t.Setenv("OPT_IN_LABEL", "io.example.update") - - cfg, err := config.Load() - require.NoError(t, err) - require.Equal(t, "/srv/stacks", cfg.StacksRoot) - require.Equal(t, "9090", cfg.Port) - require.Equal(t, "debug", cfg.LogLevel) - require.Equal(t, 30*time.Second, cfg.UpdateTimeout) - require.Equal(t, "io.example.update", cfg.OptInLabel) -} - -func TestLoad_InvalidTimeoutErrors(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - t.Setenv("UPDATE_TIMEOUT", "not-a-duration") - _, err := config.Load() - require.Error(t, err) - require.Contains(t, err.Error(), "UPDATE_TIMEOUT") -} - -func TestLoad_DefaultsToComposeMode(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - cfg, err := config.Load() - require.NoError(t, err) - require.Equal(t, "compose", cfg.Mode) -} - -func TestLoad_AcceptsSwarmMode(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - t.Setenv("MODE", "swarm") - cfg, err := config.Load() - require.NoError(t, err) - require.Equal(t, "swarm", cfg.Mode) -} - -func TestLoad_RejectsInvalidMode(t *testing.T) { - t.Setenv("UPDATER_API_KEY", "secret") - t.Setenv("MODE", "kubernetes") - _, err := config.Load() - require.Error(t, err) - require.Contains(t, err.Error(), "MODE") -} diff --git a/internal/discovery/discovery.go b/internal/discovery/discovery.go deleted file mode 100644 index 6269169..0000000 --- a/internal/discovery/discovery.go +++ /dev/null @@ -1,96 +0,0 @@ -package discovery - -import ( - "context" - "fmt" - "sort" - "strings" - - "github.com/docker/docker/api/types/container" -) - -// Job describes a single (project, service, config_files) update to execute. -type Job struct { - Project string - Service string - WorkingDir string - ConfigFiles []string - // Image is the full image reference the update request asked for. - // Compose jobs ignore it (compose.yml already pins the reference to - // pull); Swarm jobs need it to know what to set on the service spec. - Image string - // ServiceID is the Swarm service ID. Empty for Compose jobs. - ServiceID string - // Refused is true when the WorkingDir falls outside STACKS_ROOT. - // The job is returned so the caller can surface a per-job "refused" - // result, but it MUST NOT be executed. - Refused bool - RefusedReason string -} - -// Discovery orchestrates "given an image, what jobs should we enqueue?". -type Discovery struct { - cli DockerClient - stacksRoot string - optInLabel string -} - -// New returns a Discovery bound to the given Docker client and config. -func New(cli DockerClient, stacksRoot, optInLabel string) *Discovery { - return &Discovery{cli: cli, stacksRoot: stacksRoot, optInLabel: optInLabel} -} - -// FindJobs lists running containers, filters by image match + opt-in label, -// extracts Compose info, applies the path safety check, and deduplicates. -func (d *Discovery) FindJobs(ctx context.Context, image string) ([]Job, error) { - all, err := d.cli.ContainerList(ctx, container.ListOptions{All: true}) - if err != nil { - return nil, fmt.Errorf("docker container list: %w", err) - } - - seen := make(map[string]struct{}) - var jobs []Job - - for _, c := range all { - if !ImagesMatch(image, c.Image) { - continue - } - if !HasOptIn(c.Labels, d.optInLabel) { - continue - } - cl, err := ParseComposeLabels(c.Labels) - if err != nil { - continue - } - - refused := !IsInsideRoot(d.stacksRoot, cl.WorkingDir) - reason := "" - if refused { - reason = fmt.Sprintf("working_dir %q outside STACKS_ROOT %q", cl.WorkingDir, d.stacksRoot) - } - - key := dedupKey(cl) - if _, dup := seen[key]; dup { - continue - } - seen[key] = struct{}{} - - jobs = append(jobs, Job{ - Project: cl.Project, - Service: cl.Service, - WorkingDir: cl.WorkingDir, - ConfigFiles: cl.ConfigFiles, - Image: image, - Refused: refused, - RefusedReason: reason, - }) - } - - return jobs, nil -} - -func dedupKey(cl ComposeLabels) string { - files := append([]string(nil), cl.ConfigFiles...) - sort.Strings(files) - return cl.Project + "|" + cl.Service + "|" + strings.Join(files, ",") -} diff --git a/internal/discovery/discovery_test.go b/internal/discovery/discovery_test.go deleted file mode 100644 index c1f5b0a..0000000 --- a/internal/discovery/discovery_test.go +++ /dev/null @@ -1,172 +0,0 @@ -package discovery_test - -import ( - "context" - "errors" - "testing" - - "github.com/docker/docker/api/types" - "github.com/docker/docker/api/types/container" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/stretchr/testify/require" -) - -type fakeDockerClient struct { - containers []types.Container - err error -} - -func (f *fakeDockerClient) ContainerList(_ context.Context, _ container.ListOptions) ([]types.Container, error) { - return f.containers, f.err -} - -func (f *fakeDockerClient) Ping(_ context.Context) (types.Ping, error) { - return types.Ping{}, nil -} - -func mkContainer(image string, labels map[string]string) types.Container { - return types.Container{Image: image, Labels: labels} -} - -func mkComposeLabels(project, service, workingDir, configFile string, optIn bool) map[string]string { - m := map[string]string{ - "com.docker.compose.project": project, - "com.docker.compose.service": service, - "com.docker.compose.project.working_dir": workingDir, - "com.docker.compose.project.config_files": configFile, - } - if optIn { - m["se.shcizo.auto-update"] = "true" - } - return m -} - -func TestFindJobs_MatchAndOptIn(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp-prod", "web", - "/home/shcizo/self-hosted/myapp-prod", - "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml", - true, - )), - mkContainer("registry.example.com/other:v1", mkComposeLabels( - "other", "web", - "/home/shcizo/self-hosted/other", - "/home/shcizo/self-hosted/other/docker-compose.yml", - true, - )), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 1) - require.Equal(t, "myapp-prod", jobs[0].Project) - require.Equal(t, "web", jobs[0].Service) - require.Equal(t, "/home/shcizo/self-hosted/myapp-prod", jobs[0].WorkingDir) -} - -func TestFindJobs_SkipsWithoutOptIn(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp-prod", "web", - "/home/shcizo/self-hosted/myapp-prod", - "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml", - false, - )), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Empty(t, jobs) -} - -func TestFindJobs_DedupReplicas(t *testing.T) { - labels := mkComposeLabels( - "myapp", "web", - "/home/shcizo/self-hosted/myapp", - "/home/shcizo/self-hosted/myapp/docker-compose.yml", - true, - ) - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", labels), - mkContainer("registry.example.com/myapp:v1", labels), - mkContainer("registry.example.com/myapp:v1", labels), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 1) -} - -func TestFindJobs_OutsideRootProducesRefusedJob(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp", "web", - "/opt/elsewhere/myapp", - "/opt/elsewhere/myapp/docker-compose.yml", - true, - )), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 1) - require.True(t, jobs[0].Refused) -} - -func TestFindJobs_MultipleStacksSameImage(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp-prod", "web", - "/home/shcizo/self-hosted/myapp-prod", - "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml", - true, - )), - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp-staging", "web", - "/home/shcizo/self-hosted/myapp-staging", - "/home/shcizo/self-hosted/myapp-staging/docker-compose.yml", - true, - )), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 2) -} - -func TestFindJobs_DockerError(t *testing.T) { - fake := &fakeDockerClient{err: errors.New("connection refused")} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - _, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.Error(t, err) -} - -func TestFindJobs_NonComposeContainerIsSkipped(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", map[string]string{ - "se.shcizo.auto-update": "true", - }), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Empty(t, jobs) -} - -func TestFindJobs_PopulatesImageField(t *testing.T) { - fake := &fakeDockerClient{containers: []types.Container{ - mkContainer("registry.example.com/myapp:v1", mkComposeLabels( - "myapp-prod", "web", - "/home/shcizo/self-hosted/myapp-prod", - "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml", - true, - )), - }} - d := discovery.New(fake, "/home/shcizo/self-hosted", "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 1) - require.Equal(t, "registry.example.com/myapp", jobs[0].Image) - require.Empty(t, jobs[0].ServiceID) -} diff --git a/internal/discovery/docker_client.go b/internal/discovery/docker_client.go deleted file mode 100644 index 1c1aed5..0000000 --- a/internal/discovery/docker_client.go +++ /dev/null @@ -1,15 +0,0 @@ -package discovery - -import ( - "context" - - "github.com/docker/docker/api/types" - "github.com/docker/docker/api/types/container" -) - -// DockerClient is the subset of the Docker SDK we depend on. -// Defined as an interface so tests can supply a fake. -type DockerClient interface { - ContainerList(ctx context.Context, opts container.ListOptions) ([]types.Container, error) - Ping(ctx context.Context) (types.Ping, error) -} diff --git a/internal/discovery/labels.go b/internal/discovery/labels.go deleted file mode 100644 index aecde76..0000000 --- a/internal/discovery/labels.go +++ /dev/null @@ -1,71 +0,0 @@ -package discovery - -import ( - "fmt" - "strings" -) - -// ComposeLabels captures the four Compose-managed labels we need -// to drive a `docker compose pull`/`up -d` against the right stack. -type ComposeLabels struct { - Project string - Service string - WorkingDir string - ConfigFiles []string -} - -const ( - labelProject = "com.docker.compose.project" - labelService = "com.docker.compose.service" - labelWorkingDir = "com.docker.compose.project.working_dir" - labelConfigFiles = "com.docker.compose.project.config_files" -) - -// ParseComposeLabels extracts the Compose labels we need. Returns an -// error naming the missing label if any required field is absent — -// in practice this should only happen if the container was not -// started by Compose. -func ParseComposeLabels(labels map[string]string) (ComposeLabels, error) { - get := func(key string) (string, error) { - v, ok := labels[key] - if !ok || v == "" { - return "", fmt.Errorf("missing required label: %s", key) - } - return v, nil - } - - project, err := get(labelProject) - if err != nil { - return ComposeLabels{}, err - } - service, err := get(labelService) - if err != nil { - return ComposeLabels{}, err - } - workingDir, err := get(labelWorkingDir) - if err != nil { - return ComposeLabels{}, err - } - configFilesRaw, err := get(labelConfigFiles) - if err != nil { - return ComposeLabels{}, err - } - - files := strings.Split(configFilesRaw, ",") - for i, f := range files { - files[i] = strings.TrimSpace(f) - } - - return ComposeLabels{ - Project: project, - Service: service, - WorkingDir: workingDir, - ConfigFiles: files, - }, nil -} - -// HasOptIn reports whether the labels include the opt-in marker with -// value "true" (exact, case-sensitive — anything else is excluded). -func HasOptIn(labels map[string]string, key string) bool { - return labels[key] == "true" -} diff --git a/internal/discovery/labels_test.go b/internal/discovery/labels_test.go deleted file mode 100644 index 179f5dd..0000000 --- a/internal/discovery/labels_test.go +++ /dev/null @@ -1,88 +0,0 @@ -package discovery_test - -import ( - "testing" - - "github.com/shcizo/package-updater/internal/discovery" - "github.com/stretchr/testify/require" -) - -func TestParseComposeLabels_Success(t *testing.T) { - labels := map[string]string{ - "com.docker.compose.project": "myapp-prod", - "com.docker.compose.service": "web", - "com.docker.compose.project.working_dir": "/home/shcizo/self-hosted/myapp-prod", - "com.docker.compose.project.config_files": "/home/shcizo/self-hosted/myapp-prod/docker-compose.yml", - "se.shcizo.auto-update": "true", - } - got, err := discovery.ParseComposeLabels(labels) - require.NoError(t, err) - require.Equal(t, "myapp-prod", got.Project) - require.Equal(t, "web", got.Service) - require.Equal(t, "/home/shcizo/self-hosted/myapp-prod", got.WorkingDir) - require.Equal(t, []string{"/home/shcizo/self-hosted/myapp-prod/docker-compose.yml"}, got.ConfigFiles) -} - -func TestParseComposeLabels_MultipleConfigFiles(t *testing.T) { - labels := map[string]string{ - "com.docker.compose.project": "myapp", - "com.docker.compose.service": "web", - "com.docker.compose.project.working_dir": "/srv/myapp", - "com.docker.compose.project.config_files": "/srv/myapp/docker-compose.yml,/srv/myapp/docker-compose.prod.yml", - } - got, err := discovery.ParseComposeLabels(labels) - require.NoError(t, err) - require.Equal(t, []string{ - "/srv/myapp/docker-compose.yml", - "/srv/myapp/docker-compose.prod.yml", - }, got.ConfigFiles) -} - -func TestParseComposeLabels_MissingFieldsError(t *testing.T) { - cases := []struct { - name string - missing string - labels map[string]string - }{ - {"project", "com.docker.compose.project", map[string]string{ - "com.docker.compose.service": "web", - "com.docker.compose.project.working_dir": "/x", - "com.docker.compose.project.config_files": "/x/y.yml", - }}, - {"service", "com.docker.compose.service", map[string]string{ - "com.docker.compose.project": "p", - "com.docker.compose.project.working_dir": "/x", - "com.docker.compose.project.config_files": "/x/y.yml", - }}, - {"working_dir", "com.docker.compose.project.working_dir", map[string]string{ - "com.docker.compose.project": "p", - "com.docker.compose.service": "web", - "com.docker.compose.project.config_files": "/x/y.yml", - }}, - {"config_files", "com.docker.compose.project.config_files", map[string]string{ - "com.docker.compose.project": "p", - "com.docker.compose.service": "web", - "com.docker.compose.project.working_dir": "/x", - }}, - } - for _, c := range cases { - t.Run(c.name, func(t *testing.T) { - _, err := discovery.ParseComposeLabels(c.labels) - require.Error(t, err) - require.Contains(t, err.Error(), c.missing) - }) - } -} - -func TestHasOptIn(t *testing.T) { - require.True(t, discovery.HasOptIn(map[string]string{ - "se.shcizo.auto-update": "true", - }, "se.shcizo.auto-update")) - require.False(t, discovery.HasOptIn(map[string]string{ - "se.shcizo.auto-update": "false", - }, "se.shcizo.auto-update")) - require.False(t, discovery.HasOptIn(map[string]string{ - "se.shcizo.auto-update": "TRUE", - }, "se.shcizo.auto-update")) - require.False(t, discovery.HasOptIn(map[string]string{}, "se.shcizo.auto-update")) -} diff --git a/internal/discovery/matching.go b/internal/discovery/matching.go deleted file mode 100644 index c872009..0000000 --- a/internal/discovery/matching.go +++ /dev/null @@ -1,33 +0,0 @@ -package discovery - -import "strings" - -// NormaliseImage strips the tag and digest from an image reference, -// returning the bare repository name. -// -// Tricky case: "localhost:5000/foo:v1" — the first colon is a port, -// not a tag. We disambiguate by splitting on "/" first and only -// treating colons in the last segment as tag separators. -func NormaliseImage(ref string) string { - if at := strings.Index(ref, "@"); at >= 0 { - ref = ref[:at] - } - slash := strings.LastIndex(ref, "/") - if slash < 0 { - if colon := strings.Index(ref, ":"); colon >= 0 { - return ref[:colon] - } - return ref - } - prefix, last := ref[:slash], ref[slash+1:] - if colon := strings.Index(last, ":"); colon >= 0 { - last = last[:colon] - } - return prefix + "/" + last -} - -// ImagesMatch reports whether two image references resolve to the same -// repository, ignoring tag and digest. Case-sensitive per spec section 5.2. -func ImagesMatch(a, b string) bool { - return NormaliseImage(a) == NormaliseImage(b) -} diff --git a/internal/discovery/matching_test.go b/internal/discovery/matching_test.go deleted file mode 100644 index 5f20fb1..0000000 --- a/internal/discovery/matching_test.go +++ /dev/null @@ -1,51 +0,0 @@ -package discovery_test - -import ( - "testing" - - "github.com/shcizo/package-updater/internal/discovery" - "github.com/stretchr/testify/require" -) - -func TestNormaliseImage(t *testing.T) { - cases := []struct { - in string - want string - }{ - {"registry.example.com/myapp", "registry.example.com/myapp"}, - {"registry.example.com/myapp:v1.2.3", "registry.example.com/myapp"}, - {"registry.example.com/myapp:latest", "registry.example.com/myapp"}, - {"registry.example.com/myapp@sha256:abc123", "registry.example.com/myapp"}, - {"registry.example.com/myapp:v1.2.3@sha256:abc123", "registry.example.com/myapp"}, - {"nginx", "nginx"}, - {"nginx:1.25-alpine", "nginx"}, - {"library/nginx:latest", "library/nginx"}, - {"gcr.io/proj/svc:tag", "gcr.io/proj/svc"}, - {"localhost:5000/myimg:v1", "localhost:5000/myimg"}, - } - for _, c := range cases { - t.Run(c.in, func(t *testing.T) { - require.Equal(t, c.want, discovery.NormaliseImage(c.in)) - }) - } -} - -func TestImagesMatch(t *testing.T) { - require.True(t, discovery.ImagesMatch( - "registry.example.com/myapp", - "registry.example.com/myapp:v1.2.3", - )) - require.True(t, discovery.ImagesMatch( - "registry.example.com/myapp:v1.0.0", - "registry.example.com/myapp:v9.9.9", - )) - require.False(t, discovery.ImagesMatch( - "registry.example.com/myapp", - "registry.example.com/otherapp", - )) - // Case-sensitive per spec section 5.2 - require.False(t, discovery.ImagesMatch( - "registry.example.com/MyApp", - "registry.example.com/myapp", - )) -} diff --git a/internal/discovery/swarm.go b/internal/discovery/swarm.go deleted file mode 100644 index 108fad5..0000000 --- a/internal/discovery/swarm.go +++ /dev/null @@ -1,60 +0,0 @@ -package discovery - -import ( - "context" - "fmt" - - "github.com/docker/docker/api/types/swarm" -) - -// SwarmDockerClient is the subset of the Docker SDK SwarmDiscovery depends -// on. Defined as an interface so tests can supply a fake. -type SwarmDockerClient interface { - ServiceList(ctx context.Context, opts swarm.ServiceListOptions) ([]swarm.Service, error) -} - -// SwarmDiscovery orchestrates "given an image, which Swarm services should -// we update?". Unlike Discovery, it has no filesystem path to check — the -// opt-in label on the service itself is the only gate, since Swarm services -// have no local compose file to anchor a STACKS_ROOT check against. -type SwarmDiscovery struct { - cli SwarmDockerClient - optInLabel string -} - -// NewSwarm returns a SwarmDiscovery bound to the given Docker client and -// opt-in label. -func NewSwarm(cli SwarmDockerClient, optInLabel string) *SwarmDiscovery { - return &SwarmDiscovery{cli: cli, optInLabel: optInLabel} -} - -// FindJobs lists Swarm services, filters by image match + opt-in label, and -// returns one Job per matching service. Signature matches Discovery.FindJobs -// so both satisfy api.Finder. -func (d *SwarmDiscovery) FindJobs(ctx context.Context, image string) ([]Job, error) { - services, err := d.cli.ServiceList(ctx, swarm.ServiceListOptions{}) - if err != nil { - return nil, fmt.Errorf("docker service list: %w", err) - } - - var jobs []Job - for _, svc := range services { - if svc.Spec.TaskTemplate.ContainerSpec == nil { - continue - } - if !ImagesMatch(image, svc.Spec.TaskTemplate.ContainerSpec.Image) { - continue - } - if !HasOptIn(svc.Spec.Labels, d.optInLabel) { - continue - } - - jobs = append(jobs, Job{ - Service: svc.Spec.Name, - ServiceID: svc.ID, - Image: image, - }) - } - - return jobs, nil -} diff --git a/internal/discovery/swarm_test.go b/internal/discovery/swarm_test.go deleted file mode 100644 index 5c3f048..0000000 --- a/internal/discovery/swarm_test.go +++ /dev/null @@ -1,86 +0,0 @@ -package discovery_test - -import ( - "context" - "errors" - "testing" - - "github.com/docker/docker/api/types/swarm" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/stretchr/testify/require" -) - -type fakeSwarmDockerClient struct { - services []swarm.Service - err error -} - -func (f *fakeSwarmDockerClient) ServiceList(_ context.Context, _ swarm.ServiceListOptions) ([]swarm.Service, error) { - return f.services, f.err -} - -func mkService(id, name, image string, labels map[string]string) swarm.Service { - return swarm.Service{ - ID: id, - Spec: swarm.ServiceSpec{ - Annotations: swarm.Annotations{Name: name, Labels: labels}, - TaskTemplate: swarm.TaskSpec{ - ContainerSpec: &swarm.ContainerSpec{Image: image}, - }, - }, - } -} - -func TestSwarmFindJobs_MatchAndOptIn(t *testing.T) { - fake := &fakeSwarmDockerClient{services: []swarm.Service{ - mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", map[string]string{ - "se.shcizo.auto-update": "true", - }), - mkService("svc-other", "other_web", "registry.example.com/other:v1", map[string]string{ - "se.shcizo.auto-update": "true", - }), - }} - d := discovery.NewSwarm(fake, "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Len(t, jobs, 1) - require.Equal(t, "myapp_web", jobs[0].Service) - require.Equal(t, "svc-myapp", jobs[0].ServiceID) - require.Equal(t, "registry.example.com/myapp", jobs[0].Image) - require.False(t, jobs[0].Refused) - require.Empty(t, jobs[0].WorkingDir) - require.Empty(t, jobs[0].ConfigFiles) -} - -func TestSwarmFindJobs_SkipsWithoutOptIn(t *testing.T) { - fake := &fakeSwarmDockerClient{services: []swarm.Service{ - mkService("svc-myapp", "myapp_web", "registry.example.com/myapp:v1", nil), - }} - d := discovery.NewSwarm(fake, "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Empty(t, jobs) -} - -func TestSwarmFindJobs_DockerError(t *testing.T) { - fake := &fakeSwarmDockerClient{err: errors.New("connection refused")} - d := discovery.NewSwarm(fake, "se.shcizo.auto-update") - _, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.Error(t, err) -} - -func TestSwarmFindJobs_NoContainerSpecIsSkipped(t *testing.T) { - fake := &fakeSwarmDockerClient{services: []swarm.Service{ - { - ID: "svc-weird", - Spec: swarm.ServiceSpec{ - Annotations: swarm.Annotations{Name: "weird", Labels: map[string]string{"se.shcizo.auto-update": "true"}}, - TaskTemplate: swarm.TaskSpec{ContainerSpec: nil}, - }, - }, - }} - d := discovery.NewSwarm(fake, "se.shcizo.auto-update") - jobs, err := d.FindJobs(context.Background(), "registry.example.com/myapp") - require.NoError(t, err) - require.Empty(t, jobs) -} diff --git a/internal/metrics/metrics.go b/internal/metrics/metrics.go deleted file mode 100644 index 119d819..0000000 --- a/internal/metrics/metrics.go +++ /dev/null @@ -1,69 +0,0 @@ -// Package metrics defines and registers the Prometheus collectors -// exported on /metrics. -package metrics - -import ( - "net/http" - - "github.com/prometheus/client_golang/prometheus" - "github.com/prometheus/client_golang/prometheus/promhttp" -) - -// Metrics bundles all collectors. Each field is the public handle the -// rest of the service uses to record observations. -type Metrics struct { - BuildInfo *prometheus.GaugeVec - HTTPRequests *prometheus.CounterVec - UpdateJobs *prometheus.CounterVec - UpdateDuration *prometheus.HistogramVec - QueueDepth prometheus.Gauge - LastUpdateTime *prometheus.GaugeVec - DockerPingUp prometheus.Gauge -} - -// New constructs Metrics and registers them with the given registry. -// Use prometheus.NewRegistry() in tests so collectors don't leak between -// runs; production code uses prometheus.DefaultRegisterer. -func New(reg prometheus.Registerer) *Metrics { - m := &Metrics{ - BuildInfo: prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Name: "package_updater_build_info", - Help: "Always 1. Labels carry version/commit for dashboards.", - }, []string{"version", "commit"}), - HTTPRequests: prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "package_updater_http_requests_total", - Help: "HTTP requests handled, labelled by endpoint and status.", - }, []string{"endpoint", "status_code"}), - UpdateJobs: prometheus.NewCounterVec(prometheus.CounterOpts{ - Name: "package_updater_update_jobs_total", - Help: "Update jobs executed, labelled by project/service/status.", - }, []string{"project", "service", "status"}), - UpdateDuration: prometheus.NewHistogramVec(prometheus.HistogramOpts{ - Name: "package_updater_update_duration_seconds", - Help: "Time taken to pull + up a single service.", - Buckets: []float64{0.5, 1, 2, 5, 10, 30, 60, 120, 300}, - }, []string{"project", "service"}), - QueueDepth: prometheus.NewGauge(prometheus.GaugeOpts{ - Name: "package_updater_queue_depth", - Help: "Current number of submissions waiting in the queue.", - }), - LastUpdateTime: prometheus.NewGaugeVec(prometheus.GaugeOpts{ - Name: "package_updater_last_update_timestamp", - Help: "Unix timestamp of the most recent successful update per service.", - }, []string{"project", "service"}), - DockerPingUp: prometheus.NewGauge(prometheus.GaugeOpts{ - Name: "package_updater_docker_ping_up", - Help: "1 if the Docker socket responded to ping, 0 otherwise.", - }), - } - reg.MustRegister( - m.BuildInfo, m.HTTPRequests, m.UpdateJobs, m.UpdateDuration, - m.QueueDepth, m.LastUpdateTime, m.DockerPingUp, - ) - return m -} - -// Handler returns the /metrics HTTP handler bound to the given registry. -func Handler(gatherer prometheus.Gatherer) http.Handler { - return promhttp.HandlerFor(gatherer, promhttp.HandlerOpts{}) -} diff --git a/internal/selfupdate/selfupdate.go b/internal/selfupdate/selfupdate.go deleted file mode 100644 index e901d03..0000000 --- a/internal/selfupdate/selfupdate.go +++ /dev/null @@ -1,56 +0,0 @@ -// Package selfupdate handles the special case where the update target -// is the running service's own container. We must finish writing the -// HTTP response (and flush + close the connection) before exec'ing -// `docker compose up -d` against ourselves, otherwise the response is -// lost when the container is replaced. -package selfupdate - -import ( - "context" - "time" - - "github.com/shcizo/package-updater/internal/discovery" -) - -// IsSelf reports whether job targets the running service. -// Matches on Compose project name (which is also typically the -// service name for single-service stacks). -func IsSelf(job discovery.Job, selfProject string) bool { - return job.Project == selfProject -} - -// innerExec is the executor abstraction we wrap. -type innerExec interface { - Execute(ctx context.Context, job discovery.Job) error -} - -// Wrapped wraps an Executor with self-update-aware deferred execution. -type Wrapped struct { - inner innerExec - selfProject string - delay time.Duration -} - -// Wrap returns a Wrapped that defers exec until after flush() for -// self-updates. The delay is added after flush before exec, so the -// kernel TCP buffer has time to drain. -func Wrap(inner innerExec, selfProject string, delay time.Duration) *Wrapped { - return &Wrapped{inner: inner, selfProject: selfProject, delay: delay} -} - -// ExecuteWithFlush runs job, invoking flush() before exec for self-updates -// and waiting `delay` after flush. For non-self jobs, exec happens -// normally and flush is not invoked at all (the HTTP layer decides -// when to flush in that case). -func (w *Wrapped) ExecuteWithFlush(ctx context.Context, job discovery.Job, flush func()) error { - if !IsSelf(job, w.selfProject) { - return w.inner.Execute(ctx, job) - } - flush() - select { - case <-time.After(w.delay): - case <-ctx.Done(): - return ctx.Err() - } - return w.inner.Execute(ctx, job) -} diff --git a/internal/selfupdate/selfupdate_test.go b/internal/selfupdate/selfupdate_test.go deleted file mode 100644 index 121181c..0000000 --- a/internal/selfupdate/selfupdate_test.go +++ /dev/null @@ -1,87 +0,0 @@ -package selfupdate_test - -import ( - "context" - "errors" - "sync" - "testing" - "time" - - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/selfupdate" - "github.com/stretchr/testify/require" -) - -type recExec struct { - mu sync.Mutex - called bool - at time.Time -} - -func (r *recExec) Execute(_ context.Context, _ discovery.Job) error { - r.mu.Lock() - r.called = true - r.at = time.Now() - r.mu.Unlock() - return nil -} - -func TestIsSelf(t *testing.T) { - require.True(t, selfupdate.IsSelf( - discovery.Job{Project: "package-updater", Service: "package-updater"}, - "package-updater", - )) - require.False(t, selfupdate.IsSelf( - discovery.Job{Project: "other", Service: "web"}, - "package-updater", - )) -} - -func TestWrap_DefersSelf(t *testing.T) { - inner := &recExec{} - wrapped := selfupdate.Wrap(inner, "package-updater", 30*time.Millisecond) - - flushed := make(chan time.Time, 1) - flush := func() { flushed <- time.Now() } - - job := discovery.Job{Project: "package-updater", Service: "package-updater"} - require.NoError(t, wrapped.ExecuteWithFlush(context.Background(), job, flush)) - - flushAt := <-flushed - inner.mu.Lock() - require.True(t, inner.called) - require.True(t, inner.at.After(flushAt)) - inner.mu.Unlock() -} - -func TestWrap_NormalJobIsImmediate(t *testing.T) { - inner := &recExec{} - wrapped := selfupdate.Wrap(inner, "package-updater", 30*time.Millisecond) - - flushed := make(chan time.Time, 1) - flush := func() { flushed <- time.Now() } - - job := discovery.Job{Project: "other", Service: "web"} - require.NoError(t, wrapped.ExecuteWithFlush(context.Background(), job, flush)) - - inner.mu.Lock() - require.True(t, inner.called) - inner.mu.Unlock() - select { - case <-flushed: - t.Fatal("flush should not be called for non-self jobs") - default: - } -} - -func TestWrap_PropagatesError(t *testing.T) { - wrapped := selfupdate.Wrap(failExec{}, "x", 1*time.Millisecond) - err := wrapped.ExecuteWithFlush(context.Background(), discovery.Job{Project: "other"}, func() {}) - require.Error(t, err) -} - -type failExec struct{} - -func (failExec) Execute(_ context.Context, _ discovery.Job) error { - return errors.New("boom") -} diff --git a/internal/updater/executor.go b/internal/updater/executor.go deleted file mode 100644 index e767282..0000000 --- a/internal/updater/executor.go +++ /dev/null @@ -1,15 +0,0 @@ -// Package updater contains the FIFO job queue, the single worker that -// drains it, and the abstraction over `docker compose` invocation. -package updater - -import ( - "context" - - "github.com/shcizo/package-updater/internal/discovery" -) - -// Executor runs `docker compose pull` then `up -d` for a single job. -// Defined as an interface so the worker can be tested with a fake. -type Executor interface { - Execute(ctx context.Context, job discovery.Job) error -} diff --git a/internal/updater/queue.go b/internal/updater/queue.go deleted file mode 100644 index c7b3a38..0000000 --- a/internal/updater/queue.go +++ /dev/null @@ -1,144 +0,0 @@ -package updater - -import ( - "context" - "time" - - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/metrics" -) - -// Status describes the outcome of executing a single Job. -type Status string - -const ( - StatusUpdated Status = "updated" - StatusFailed Status = "failed" - StatusRefused Status = "refused" - StatusTimeout Status = "timeout" -) - -// Result captures a per-Job outcome to be surfaced in the HTTP response. -type Result struct { - Job discovery.Job - Status Status - Error string - DurationMs int64 -} - -// Queue serialises Job execution through a single worker so we never -// run two `docker compose` commands against the same stack concurrently. -// It also satisfies the "single global FIFO worker" design choice in -// spec section 5.7. -type Queue struct { - exec Executor - metrics *metrics.Metrics - ch chan submission - stop chan struct{} - done chan struct{} -} - -type submission struct { - ctx context.Context - jobs []discovery.Job - results chan []Result -} - -// NewQueue constructs a queue bound to the given executor. m may be nil, -// in which case metrics recording is silently skipped. -func NewQueue(exec Executor, m *metrics.Metrics) *Queue { - return &Queue{ - exec: exec, - metrics: m, - ch: make(chan submission, 16), - stop: make(chan struct{}), - done: make(chan struct{}), - } -} - -// Start launches the single background worker. Call Stop to terminate. -func (q *Queue) Start(ctx context.Context) { - go q.run(ctx) -} - -// Stop signals the worker to exit and waits for it to finish. -func (q *Queue) Stop() { - close(q.stop) - <-q.done -} - -// Submit blocks until all jobs in the batch have been processed by the -// worker, then returns per-job results in the same order. The ctx -// timeout (if any) applies to each individual Job's execution and is -// surfaced as StatusTimeout. -func (q *Queue) Submit(ctx context.Context, jobs []discovery.Job) []Result { - resCh := make(chan []Result, 1) - q.ch <- submission{ctx: ctx, jobs: jobs, results: resCh} - if q.metrics != nil { - q.metrics.QueueDepth.Set(float64(len(q.ch))) - } - return <-resCh -} - -func (q *Queue) run(_ context.Context) { - defer close(q.done) - for { - select { - case <-q.stop: - return - case s := <-q.ch: - results := make([]Result, len(s.jobs)) - for i, j := range s.jobs { - results[i] = q.runOne(s.ctx, j) - } - s.results <- results - } - } -} - -func (q *Queue) runOne(ctx context.Context, job discovery.Job) Result { - start := time.Now() - r := Result{Job: job} - - if job.Refused { - r.Status = StatusRefused - r.Error = job.RefusedReason - r.DurationMs = time.Since(start).Milliseconds() - if q.metrics != nil { - q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(StatusRefused)).Inc() - q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(time.Since(start).Seconds()) - } - return r - } - - err := q.exec.Execute(ctx, job) - r.DurationMs = time.Since(start).Milliseconds() - switch { - case err == nil: - r.Status = StatusUpdated - case errorsIsContextDeadline(err) || ctxDeadlineExceeded(ctx): - r.Status = StatusTimeout - r.Error = err.Error() - default: - r.Status = StatusFailed - r.Error = err.Error() - } - - if q.metrics != nil { - elapsed := time.Since(start).Seconds() - q.metrics.UpdateJobs.WithLabelValues(job.Project, job.Service, string(r.Status)).Inc() - q.metrics.UpdateDuration.WithLabelValues(job.Project, job.Service).Observe(elapsed) - if r.Status == StatusUpdated { - q.metrics.LastUpdateTime.WithLabelValues(job.Project, job.Service).SetToCurrentTime() - } - } - return r -} - -func errorsIsContextDeadline(err error) bool { - return err == context.DeadlineExceeded -} - -func ctxDeadlineExceeded(ctx context.Context) bool { - return ctx.Err() == context.DeadlineExceeded -} diff --git a/internal/updater/queue_test.go b/internal/updater/queue_test.go deleted file mode 100644 index c8f99d8..0000000 --- a/internal/updater/queue_test.go +++ /dev/null @@ -1,129 +0,0 @@ -package updater_test - -import ( - "context" - "errors" - "sync" - "testing" - "time" - - dto "github.com/prometheus/client_model/go" - "github.com/prometheus/client_golang/prometheus" - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/metrics" - "github.com/shcizo/package-updater/internal/updater" - "github.com/stretchr/testify/require" -) - -type fakeExec struct { - mu sync.Mutex - calls []discovery.Job - delay time.Duration - errForSvc map[string]error -} - -func (f *fakeExec) Execute(ctx context.Context, job discovery.Job) error { - f.mu.Lock() - f.calls = append(f.calls, job) - f.mu.Unlock() - if f.delay > 0 { - select { - case <-time.After(f.delay): - case <-ctx.Done(): - return ctx.Err() - } - } - if err, ok := f.errForSvc[job.Service]; ok { - return err - } - return nil -} - -func (f *fakeExec) callsCopy() []discovery.Job { - f.mu.Lock() - defer f.mu.Unlock() - return append([]discovery.Job(nil), f.calls...) -} - -func TestQueue_ProcessesFIFO(t *testing.T) { - exec := &fakeExec{delay: 20 * time.Millisecond} - q := updater.NewQueue(exec, nil) - q.Start(context.Background()) - defer q.Stop() - - job1 := discovery.Job{Project: "a", Service: "svc"} - job2 := discovery.Job{Project: "b", Service: "svc"} - job3 := discovery.Job{Project: "c", Service: "svc"} - - results := make(chan []updater.Result, 3) - go func() { results <- q.Submit(context.Background(), []discovery.Job{job1}) }() - time.Sleep(5 * time.Millisecond) - go func() { results <- q.Submit(context.Background(), []discovery.Job{job2}) }() - time.Sleep(5 * time.Millisecond) - go func() { results <- q.Submit(context.Background(), []discovery.Job{job3}) }() - - for i := 0; i < 3; i++ { - <-results - } - - calls := exec.callsCopy() - require.Equal(t, []string{"a", "b", "c"}, []string{calls[0].Project, calls[1].Project, calls[2].Project}) -} - -func TestQueue_ReturnsPerJobResults(t *testing.T) { - exec := &fakeExec{errForSvc: map[string]error{"failing": errors.New("boom")}} - q := updater.NewQueue(exec, nil) - q.Start(context.Background()) - defer q.Stop() - - jobs := []discovery.Job{ - {Project: "p1", Service: "ok"}, - {Project: "p2", Service: "failing"}, - {Project: "p3", Service: "refused-svc", Refused: true, RefusedReason: "outside root"}, - } - results := q.Submit(context.Background(), jobs) - - require.Len(t, results, 3) - require.Equal(t, updater.StatusUpdated, results[0].Status) - require.Equal(t, updater.StatusFailed, results[1].Status) - require.Contains(t, results[1].Error, "boom") - require.Equal(t, updater.StatusRefused, results[2].Status) - require.Contains(t, results[2].Error, "outside root") -} - -func TestQueue_Timeout(t *testing.T) { - exec := &fakeExec{delay: 200 * time.Millisecond} - q := updater.NewQueue(exec, nil) - q.Start(context.Background()) - defer q.Stop() - - ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond) - defer cancel() - - jobs := []discovery.Job{{Project: "slow", Service: "svc"}} - results := q.Submit(ctx, jobs) - - require.Len(t, results, 1) - require.Equal(t, updater.StatusTimeout, results[0].Status) -} - -func TestQueue_RecordsMetrics(t *testing.T) { - reg := prometheus.NewRegistry() - m := metrics.New(reg) - exec := &fakeExec{} - q := updater.NewQueue(exec, m) - q.Start(context.Background()) - defer q.Stop() - - q.Submit(context.Background(), []discovery.Job{{Project: "p", Service: "s"}}) - - // Verify UpdateJobs counter was incremented for the successful job. - metric := &dto.Metric{} - require.NoError(t, m.UpdateJobs.WithLabelValues("p", "s", "updated").Write(metric)) - require.Equal(t, 1.0, metric.GetCounter().GetValue()) - - // Verify LastUpdateTime was set (non-zero). - tsMetric := &dto.Metric{} - require.NoError(t, m.LastUpdateTime.WithLabelValues("p", "s").Write(tsMetric)) - require.Greater(t, tsMetric.GetGauge().GetValue(), 0.0) -} diff --git a/internal/updater/worker.go b/internal/updater/worker.go deleted file mode 100644 index 7705bae..0000000 --- a/internal/updater/worker.go +++ /dev/null @@ -1,6 +0,0 @@ -package updater - -// Worker logic currently lives in queue.go (single in-process worker -// goroutine). This file is reserved for the eventual per-stack-mutex -// implementation called out in spec section 15 ("Multi-worker concurrency -// with per-stack mutex"). Intentionally empty for v1. diff --git a/internal/updater/worker_test.go b/internal/updater/worker_test.go deleted file mode 100644 index 9522b8f..0000000 --- a/internal/updater/worker_test.go +++ /dev/null @@ -1,38 +0,0 @@ -package updater_test - -import ( - "context" - "sync/atomic" - "testing" - "time" - - "github.com/shcizo/package-updater/internal/discovery" - "github.com/shcizo/package-updater/internal/updater" - "github.com/stretchr/testify/require" -) - -type counterExec struct{ n atomic.Int32 } - -func (c *counterExec) Execute(_ context.Context, _ discovery.Job) error { - c.n.Add(1) - time.Sleep(10 * time.Millisecond) - return nil -} - -func TestWorker_RunsExactlyOneAtATime(t *testing.T) { - exec := &counterExec{} - q := updater.NewQueue(exec, nil) - q.Start(context.Background()) - defer q.Stop() - - jobs := make([]discovery.Job, 10) - for i := range jobs { - jobs[i] = discovery.Job{Project: "p", Service: "svc"} - } - start := time.Now() - q.Submit(context.Background(), jobs) - elapsed := time.Since(start) - - require.GreaterOrEqual(t, elapsed, 90*time.Millisecond) - require.Equal(t, int32(10), exec.n.Load()) -}