shcizo/package-updater
1
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Merge pull request 'Initial implementation: package-updater v1' (#1) from feat/initial-implementation into main

Browse Source
This commit was merged in pull request #1.
This commit is contained in:
Samuel Enocsson
2026-05-22 13:59:37 +02:00
parent 3e600aeb88 e43868bf41
commit 46ef00c742
40 changed files with 2497 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dockerignore
+10
View File
@@ -0,0 +1,10 @@
.git
.gitignore
.github
docs/
README.md
*.md
.env
.env.*
gitea-action/
docker-compose.example.yml
.gitignore
+26
View File
@@ -0,0 +1,26 @@
# Binaries
/bin/
/server
*.exe
*.dll
*.so
*.dylib
# Test binary, output of `go test -c`
*.test
# Coverage
*.out
coverage.html
# IDE
.idea/
.vscode/
*.swp
# Local env files
.env
.env.local
# Serena MCP workspace
.serena/
Dockerfile
+22
View File
@@ -0,0 +1,22 @@
# syntax=docker/dockerfile:1.7
FROM golang:1.26-alpine AS build
WORKDIR /src
COPY go.mod go.sum ./
RUN go mod download
COPY . .
ARG VERSION=dev
ARG COMMIT=unknown
ARG BUILD_TIME=unknown
RUN CGO_ENABLED=0 go build \
-ldflags="-s -w \
-X main.version=${VERSION} \
-X main.commit=${COMMIT} \
-X main.buildTime=${BUILD_TIME}" \
-o /out/package-updater ./cmd/server
FROM alpine:3.20
RUN apk add --no-cache docker-cli docker-cli-compose ca-certificates wget
COPY --from=build /out/package-updater /usr/local/bin/
EXPOSE 8080
ENTRYPOINT ["/usr/local/bin/package-updater"]
README.md
+84
View File
@@ -0,0 +1,84 @@
# package-updater
Webhook-driven Docker Compose service updater. Fills the gap between Watchtower (polling, no CI integration) and full GitOps (Argo CD, Flux) for a self-hosted, single-host environment.
**Trigger flow:**
1. Gitea workflow builds and pushes a new image to your registry.
2. Workflow calls `POST /update` on this service with the image name.
3. Service finds the matching Compose-managed container(s) on the host via Docker labels.
4. Runs `docker compose pull` + `up -d` for the relevant service(s).
See [design spec](docs/superpowers/specs/2026-05-22-package-updater-design.md) and [implementation plan](docs/superpowers/plans/2026-05-22-package-updater-implementation.md) for full design and rationale.
## How it finds the right stack
The service queries the Docker socket and reads the labels Compose itself attaches to every container:
- `com.docker.compose.project`
- `com.docker.compose.service`
- `com.docker.compose.project.working_dir`
- `com.docker.compose.project.config_files`
A container is eligible for update only if it has **both**:
- An image name matching the request (tag-agnostic), AND
- The opt-in label `se.shcizo.auto-update=true`.
Defense in depth: a valid bearer token AND the opt-in label must both be present before any container is touched.
## Quick start
1. Build and push the image (e.g. via your own CI).
2. Copy `docker-compose.example.yml` to `/home/shcizo/self-hosted/package-updater/docker-compose.yml`.
3. Create `.env` next to it: `UPDATER_API_KEY=$(openssl rand -hex 32)`.
4. Point your reverse proxy (NPM/Traefik/Caddy) at `package-updater:8080`. NPM should handle TLS.
5. `docker compose up -d`.
6. Add the opt-in label `se.shcizo.auto-update: "true"` to each service you want auto-updated.
7. Use the [Gitea composite action](gitea-action/README.md) in your repos to call `/update` after a build.
## Configuration
All via environment variables.
| Variable | Required | Default | Purpose |
|---|---|---|---|
| `UPDATER_API_KEY` | **yes** | — | Bearer token. Service refuses to start without it. |
| `STACKS_ROOT` | no | `/home/shcizo/self-hosted` | Required parent for any stack eligible to update. |
| `PORT` | no | `8080` | HTTP listen port. |
| `LOG_LEVEL` | no | `info` | `debug` / `info` / `warn` / `error`. |
| `UPDATE_TIMEOUT` | no | `5m` | Per-job timeout (Go duration). |
| `OPT_IN_LABEL` | no | `se.shcizo.auto-update` | Label name to check; value must equal `"true"`. |
## Endpoints
| Endpoint | Auth | Purpose |
|---|---|---|
| `POST /update` | Bearer token | Trigger pull + restart for matching services |
| `GET /healthz` | none | Liveness + Docker socket reachability |
| `GET /version` | none | Build info |
| `GET /metrics` | none | Prometheus exposition |
`/healthz`, `/version`, and `/metrics` are intentionally unauthenticated — they're internal-network only behind the reverse proxy.
## Observability
- **Logs**: JSON to stdout, picked up by Promtail/Alloy → Loki.
- **Metrics**: Prometheus exposition on `/metrics`. Notable: `package_updater_update_jobs_total{project,service,status}`, `package_updater_last_update_timestamp{project,service}`, `package_updater_docker_ping_up`.
## Development
```bash
go test ./...
go build ./cmd/server
docker build -t package-updater:dev .
```
## Known v1 gaps
These are tracked in the spec's section 2 and section 15 as deliberate out-of-scope:
- **Self-update wiring**: `internal/selfupdate.Wrapped` exists and is unit-tested but is not wired into the live queue. The HTTP response flush ordering for self-replacement is a future enhancement; for now, expect to manually rerun `docker compose up -d` on the host if pushing a new image of `package-updater` itself causes a mid-response interruption.
- **No rollback**: Compose's "keep old container if new fails to start" is the only safety net.
- **Single host only**.
- **No per-repo API keys**: a single shared bearer token is used.
cmd/server/main.go
+131
View File
@@ -0,0 +1,131 @@
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,
"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()
disc := discovery.New(dockerCli, cfg.StacksRoot, cfg.OptInLabel)
exec := updater.NewComposeExecutor()
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(disc, &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)
}
docker-compose.example.yml
+32
View File
@@ -0,0 +1,32 @@
# Example deployment of package-updater itself.
# Copy to /home/shcizo/self-hosted/package-updater/docker-compose.yml
# and create a .env file alongside containing:
# UPDATER_API_KEY=<openssl rand -hex 32>
services:
package-updater:
image: registry.example.com/package-updater:latest
container_name: package-updater
restart: unless-stopped
environment:
- UPDATER_API_KEY=${UPDATER_API_KEY}
- STACKS_ROOT=/home/shcizo/self-hosted
- LOG_LEVEL=info
- PORT=8080
- UPDATE_TIMEOUT=5m
volumes:
- /var/run/docker.sock:/var/run/docker.sock
- /home/shcizo/self-hosted:/home/shcizo/self-hosted:ro
labels:
- "se.shcizo.auto-update=true"
networks:
- proxy
healthcheck:
test: ["CMD", "wget", "-q", "-O-", "http://localhost:8080/healthz"]
interval: 30s
timeout: 5s
retries: 3
networks:
proxy:
external: true
docker-compose.local.yml
+40
View File
@@ -0,0 +1,40 @@
# Local development / smoke-test compose file.
# Unlike docker-compose.example.yml (which pulls a pre-built image from a registry),
# this file BUILDS the image from the local Dockerfile so you can iterate without
# pushing anywhere.
#
# Usage:
# 1. Copy .env.example to .env and fill in UPDATER_API_KEY (e.g. `openssl rand -hex 32`)
# 2. Optionally adjust STACKS_ROOT below to point at a directory you actually have
# compose stacks in (default points at this repo dir, which is fine for smoke tests)
# 3. docker compose -f docker-compose.local.yml up --build
# 4. curl -s http://localhost:8080/healthz | jq
# 5. curl -sH "Authorization: Bearer $UPDATER_API_KEY" -d '{"image":"test"}' \
# http://localhost:8080/update | jq
services:
package-updater:
build:
context: .
args:
VERSION: dev-local
COMMIT: ${COMMIT:-unknown}
BUILD_TIME: ${BUILD_TIME:-unknown}
container_name: package-updater-local
restart: unless-stopped
environment:
- UPDATER_API_KEY=${UPDATER_API_KEY}
- STACKS_ROOT=${STACKS_ROOT:-/tmp}
- LOG_LEVEL=debug
- PORT=8080
- UPDATE_TIMEOUT=5m
ports:
- "8080:8080"
volumes:
- /var/run/docker.sock:/var/run/docker.sock
- ${STACKS_ROOT:-/tmp}:${STACKS_ROOT:-/tmp}:ro
healthcheck:
test: ["CMD", "wget", "-q", "-O-", "http://localhost:8080/healthz"]
interval: 30s
timeout: 5s
retries: 3
docs/superpowers/plans/2026-05-22-package-updater-implementation.md
-2
View File
@@ -2587,7 +2587,6 @@ import (
"net/http" "net/http"
"os" "os"
"os/signal" "os/signal"
"strconv"
"syscall" "syscall"
"time" "time"
@@ -2708,7 +2707,6 @@ type submitterAdapter struct {
func (s *submitterAdapter) Submit(ctx context.Context, jobs []discovery.Job) []updater.Result { func (s *submitterAdapter) Submit(ctx context.Context, jobs []discovery.Job) []updater.Result {
ctx, cancel := context.WithTimeout(ctx, s.timeout) ctx, cancel := context.WithTimeout(ctx, s.timeout)
defer cancel() defer cancel()
_ = strconv.Itoa // silence unused-import linter for strconv in case go.sum drops it
return s.queue.Submit(ctx, jobs) return s.queue.Submit(ctx, jobs)
} }
``` ```
docs/superpowers/specs/2026-05-22-package-updater-design.md
+5 -5
View File
@@ -50,7 +50,7 @@ Watchtower remains in place for third-party images that the user does not build
3. Service validates token (constant-time compare). 3. Service validates token (constant-time compare).
4. Service lists Docker containers (running + stopped), filters by: 4. Service lists Docker containers (running + stopped), filters by:
- Image name match (tag-agnostic) - Image name match (tag-agnostic)
- Opt-in label `se.enocsson.auto-update=true` - Opt-in label `se.shcizo.auto-update=true`
5. For each match, reads Compose's built-in labels to find `working_dir`, `config_files`, `service`, `project`. 5. For each match, reads Compose's built-in labels to find `working_dir`, `config_files`, `service`, `project`.
6. Path safety check: refuse jobs whose `working_dir` is not inside `STACKS_ROOT`. 6. Path safety check: refuse jobs whose `working_dir` is not inside `STACKS_ROOT`.
7. Deduplicates `(project, service, config_files)` and enqueues one job per unique tuple. 7. Deduplicates `(project, service, config_files)` and enqueues one job per unique tuple.
@@ -161,7 +161,7 @@ Case-sensitive. No wildcards or regex (YAGNI).
### 5.3 Opt-in filter ### 5.3 Opt-in filter
Container must have label `se.enocsson.auto-update=true`. Anything else (`false`, missing, other value) is silently excluded. Container must have label `se.shcizo.auto-update=true`. Anything else (`false`, missing, other value) is silently excluded.
### 5.4 Compose label extraction ### 5.4 Compose label extraction
@@ -210,7 +210,7 @@ When the service receives an update whose image matches its own running containe
A container is eligible for update only if it has both: A container is eligible for update only if it has both:
- An image name matching the request, AND - An image name matching the request, AND
- The opt-in label `se.enocsson.auto-update=true`. - The opt-in label `se.shcizo.auto-update=true`.
Both gates are independent. Compromising either alone does not allow an attacker to trigger an update. Both gates are independent. Compromising either alone does not allow an attacker to trigger an update.
@@ -244,7 +244,7 @@ services:
- /var/run/docker.sock:/var/run/docker.sock - /var/run/docker.sock:/var/run/docker.sock
- /home/shcizo/self-hosted:/home/shcizo/self-hosted:ro - /home/shcizo/self-hosted:/home/shcizo/self-hosted:ro
labels: labels:
- "se.enocsson.auto-update=true" - "se.shcizo.auto-update=true"
networks: networks:
- proxy - proxy
healthcheck: healthcheck:
@@ -426,7 +426,7 @@ All configuration is via environment variables.
| `PORT` | no | `8080` | HTTP listen port. | | `PORT` | no | `8080` | HTTP listen port. |
| `LOG_LEVEL` | no | `info` | `debug`/`info`/`warn`/`error`. | | `LOG_LEVEL` | no | `info` | `debug`/`info`/`warn`/`error`. |
| `UPDATE_TIMEOUT` | no | `5m` | Per-job timeout. Go duration string. | | `UPDATE_TIMEOUT` | no | `5m` | Per-job timeout. Go duration string. |
| `OPT_IN_LABEL` | no | `se.enocsson.auto-update` | Label name to check (allows renaming without rebuild). Value must equal `"true"`. | | `OPT_IN_LABEL` | no | `se.shcizo.auto-update` | Label name to check (allows renaming without rebuild). Value must equal `"true"`. |
## 13. Repository Layout (planned) ## 13. Repository Layout (planned)
env.sample
+15
View File
@@ -0,0 +1,15 @@
# Copy to .env (which is gitignored) and fill in real values:
# cp env.sample .env && $EDITOR .env
# Required: bearer token clients must send as `Authorization: Bearer <value>`.
# Generate with: openssl rand -hex 32
UPDATER_API_KEY=
# Optional: root directory the service is allowed to manage stacks under.
# Defaults to /tmp for local smoke testing. For real use, set to your stacks dir,
# e.g. /home/shcizo/self-hosted
# STACKS_ROOT=/home/shcizo/self-hosted
# Optional: build info baked into the binary via -ldflags. Not required.
# COMMIT=$(git rev-parse --short HEAD)
# BUILD_TIME=$(date -u +%Y-%m-%dT%H:%M:%SZ)
gitea-action/README.md
+36
View File
@@ -0,0 +1,36 @@
# Deploy via package-updater (composite action)
Notifies `package-updater` to `docker compose pull` + `up -d` for the matching service(s) after a CI build.
## Usage
In a consumer repo's `.gitea/workflows/deploy.yml`:
```yaml
jobs:
deploy:
runs-on: ubuntu-latest
needs: [build-and-push]
steps:
- uses: gitea.example.com/shcizo/package-updater/gitea-action@v1
with:
endpoint: https://updater.example.com/update
image: registry.example.com/${{ gitea.repository }}
tag: ${{ gitea.sha }}
token: ${{ secrets.UPDATER_TOKEN }}
```
`UPDATER_TOKEN` should be set as an organisation-level secret in Gitea so all repos share it.
## Inputs
| Name | Required | Default | Description |
|---|---|---|---|
| `endpoint` | yes | — | Full URL to `/update` |
| `image` | yes | — | Image reference without tag |
| `tag` | no | `""` | Tag that was just pushed (logged for audit) |
| `token` | yes | — | Bearer token configured in package-updater |
## Failure modes
The step exits non-zero if `package-updater` returns HTTP 4xx or 5xx. This is intentional — the workflow surfaces the deploy failure to whoever pushed.
gitea-action/action.yml
+37
View File
@@ -0,0 +1,37 @@
name: "Deploy via package-updater"
description: "Notifies package-updater to pull & restart a Docker Compose service"
inputs:
endpoint:
description: "Full URL to /update (e.g. https://updater.example.com/update)"
required: true
image:
description: "Image reference without tag (e.g. registry.example.com/myapp)"
required: true
tag:
description: "Tag that was just pushed (for logging)"
required: false
default: ""
token:
description: "Bearer token for package-updater"
required: true
runs:
using: "composite"
steps:
- name: Trigger update
shell: bash
env:
TOKEN: ${{ inputs.token }}
run: |
set -euo pipefail
response=$(curl -sS -w "\n%{http_code}" \
-X POST "${{ inputs.endpoint }}" \
-H "Authorization: Bearer $TOKEN" \
-H "Content-Type: application/json" \
-d "{\"image\":\"${{ inputs.image }}\",\"tag\":\"${{ inputs.tag }}\"}")
body=$(echo "$response" | head -n -1)
code=$(echo "$response" | tail -n 1)
echo "HTTP $code"
echo "$body" | jq .
if [ "$code" -ge 400 ]; then
exit 1
fi
go.mod
+49
View File
@@ -0,0 +1,49 @@
module github.com/shcizo/package-updater
go 1.26.3
require (
github.com/docker/docker v28.5.2+incompatible
github.com/prometheus/client_golang v1.23.2
github.com/stretchr/testify v1.11.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
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/moby/docker-image-spec v1.3.1 // indirect
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/client_model v0.6.2 // 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/sys v0.42.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
)
go.sum
+128
View File
@@ -0,0 +1,128 @@
github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEKWjV8V+WSxDXJ4NFATAsZjh8iIbsQIg=
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=
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
github.com/containerd/errdefs v1.0.0 h1:tg5yIfIlQIrxYtu9ajqY42W3lpS19XqdxRQeEwYG8PI=
github.com/containerd/errdefs v1.0.0/go.mod h1:+YBYIdtsnF4Iw6nWZhJcqGSg/dwvV7tyJ/kCkyJ2k+M=
github.com/containerd/errdefs/pkg v0.3.0 h1:9IKJ06FvyNlexW690DXuQNx2KA2cUJXx151Xdx3ZPPE=
github.com/containerd/errdefs/pkg v0.3.0/go.mod h1:NJw6s9HwNuRhnjJhM7pylWwMyAkmCQvQ4GpJHEqRLVk=
github.com/containerd/log v0.1.0 h1:TCJt7ioM2cr/tfR8GPbGf9/VRAX8D2B4PjzCpfX540I=
github.com/containerd/log v0.1.0/go.mod h1:VRRf09a7mHDIRezVKTRCrOq78v577GXq3bSa3EhrzVo=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
github.com/distribution/reference v0.6.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
github.com/docker/docker v28.5.2+incompatible h1:DBX0Y0zAjZbSrm1uzOkdr1onVghKaftjlSWt4AFexzM=
github.com/docker/docker v28.5.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
github.com/docker/go-connections v0.7.0 h1:6SsRfJddP22WMrCkj19x9WKjEDTB+ahsdiGYf0mN39c=
github.com/docker/go-connections v0.7.0/go.mod h1:no1qkHdjq7kLMGUXYAduOhYPSJxxvgWBh7ogVvptn3Q=
github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
github.com/docker/go-units v0.5.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
github.com/felixge/httpsnoop v1.0.4 h1:NFTV2Zj1bL4mc9sqWACXbQFVBBg2W3GPvqp8/ESS2Wg=
github.com/felixge/httpsnoop v1.0.4/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSwevea8zH2U=
github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs=
github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c=
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=
github.com/moby/sys/atomicwriter v0.1.0/go.mod h1:Ul8oqv2ZMNHOceF643P6FKPXeCmYtlQMvpizfsSoaWs=
github.com/moby/sys/sequential v0.6.0 h1:qrx7XFUd/5DxtqcoH1h438hF5TmOvzC/lspjy7zgvCU=
github.com/moby/sys/sequential v0.6.0/go.mod h1:uyv8EUTrca5PnDsdMGXhZe6CCe8U/UiTWd+lL+7b/Ko=
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=
github.com/opencontainers/image-spec v1.1.1/go.mod h1:qpqAh3Dmcf36wStyyWU+kCeDgrGnAve2nCC8+7h8Q0M=
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=
github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ=
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 h1:CqXxU8VOmDefoh0+ztfGaymYbhdB/tT3zs79QaZTNGY=
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0/go.mod h1:BuhAPThV8PBHBvg8ZzZ/Ok3idOdhWIodywz2xEcRbJo=
go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I=
go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0=
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 h1:88Y4s2C8oTui1LGM6bTWkw0ICGcOLCAI5l6zsD1j20k=
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0/go.mod h1:Vl1/iaggsuRlrHf/hfPJPvVag77kKyvrLeD10kpMl+A=
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 h1:3iZJKlCZufyRzPzlQhUIWVmfltrXuGyfjREgGP3UUjc=
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0/go.mod h1:/G+nUPfhq2e+qiXMGxMwumDrP5jtzU+mWN7/sjT2rak=
go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM=
go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY=
go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg=
go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg=
go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw=
go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A=
go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A=
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/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8=
golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA=
golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=
golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno=
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA=
google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM=
google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
gotest.tools/v3 v3.5.2 h1:7koQfIKdy+I8UTetycgUqXWSDwpgv193Ka+qRsmBY8Q=
gotest.tools/v3 v3.5.2/go.mod h1:LtdLGcnqToBH83WByAAi/wiwSFCArdFIUV/xxN4pcjA=
internal/api/handlers.go
+142
View File
@@ -0,0 +1,142 @@
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
}
jobs, err := h.finder.FindJobs(r.Context(), req.Image)
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,
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})
}
internal/api/handlers_test.go
+168
View File
@@ -0,0 +1,168 @@
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
}
func (f *fakeFinder) FindJobs(_ context.Context, _ string) ([]discovery.Job, error) {
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 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)
}
internal/api/middleware.go
+112
View File
@@ -0,0 +1,112 @@
// 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
}
internal/api/middleware_test.go
+92
View File
@@ -0,0 +1,92 @@
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)
}
internal/api/types.go
+39
View File
@@ -0,0 +1,39 @@
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"`
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"`
}
internal/config/config.go
+52
View File
@@ -0,0 +1,52 @@
// 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
}
// 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"),
}
if cfg.APIKey == "" {
return nil, errors.New("UPDATER_API_KEY is required")
}
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
}
internal/config/config_test.go
+59
View File
@@ -0,0 +1,59 @@
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")
}
internal/discovery/discovery.go
+89
View File
@@ -0,0 +1,89 @@
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
// 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,
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, ",")
}
internal/discovery/discovery_test.go
+155
View File
@@ -0,0 +1,155 @@
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)
}
internal/discovery/docker_client.go
+15
View File
@@ -0,0 +1,15 @@
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)
}
internal/discovery/labels.go
+71
View File
@@ -0,0 +1,71 @@
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"
}
internal/discovery/labels_test.go
+88
View File
@@ -0,0 +1,88 @@
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"))
}
internal/discovery/matching.go
+33
View File
@@ -0,0 +1,33 @@
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)
}
internal/discovery/matching_test.go
+51
View File
@@ -0,0 +1,51 @@
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",
))
}
internal/discovery/pathcheck.go
+23
View File
@@ -0,0 +1,23 @@
package discovery
import (
"path/filepath"
"strings"
)
// IsInsideRoot reports whether path is the same as, or nested inside,
// root. Both are cleaned before comparison so trailing slashes, "."
// segments, and ".." escapes are handled. Prefix tricks like
// "/foo" vs "/foo-evil" are NOT considered inside.
func IsInsideRoot(root, path string) bool {
r := filepath.Clean(root)
p := filepath.Clean(path)
if p == r {
return true
}
rel, err := filepath.Rel(r, p)
if err != nil {
return false
}
return !strings.HasPrefix(rel, "..")
}
internal/discovery/pathcheck_test.go
+32
View File
@@ -0,0 +1,32 @@
package discovery_test
import (
"testing"
"github.com/shcizo/package-updater/internal/discovery"
"github.com/stretchr/testify/require"
)
func TestIsInsideRoot(t *testing.T) {
cases := []struct {
name string
root string
path string
want bool
}{
{"direct child", "/home/shcizo/self-hosted", "/home/shcizo/self-hosted/myapp", true},
{"nested", "/home/shcizo/self-hosted", "/home/shcizo/self-hosted/a/b/c", true},
{"root itself", "/home/shcizo/self-hosted", "/home/shcizo/self-hosted", true},
{"sibling", "/home/shcizo/self-hosted", "/home/shcizo/other", false},
{"parent", "/home/shcizo/self-hosted", "/home/shcizo", false},
{"unrelated", "/home/shcizo/self-hosted", "/etc/passwd", false},
{"prefix-trick", "/home/shcizo/self-hosted", "/home/shcizo/self-hosted-evil", false},
{"dotdot escape", "/home/shcizo/self-hosted", "/home/shcizo/self-hosted/../etc", false},
{"trailing slash root", "/home/shcizo/self-hosted/", "/home/shcizo/self-hosted/x", true},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
require.Equal(t, c.want, discovery.IsInsideRoot(c.root, c.path))
})
}
}
internal/logging/logger.go
+63
View File
@@ -0,0 +1,63 @@
// Package logging configures the structured JSON logger used across the service.
package logging
import (
"context"
"log/slog"
"os"
)
type ctxKey int
const requestIDKey ctxKey = iota
// New returns a slog.Logger that writes JSON to stdout at the given level.
// Valid levels: "debug", "info", "warn", "error". Unknown levels default to info.
func New(level string) *slog.Logger {
var lvl slog.Level
switch level {
case "debug":
lvl = slog.LevelDebug
case "warn":
lvl = slog.LevelWarn
case "error":
lvl = slog.LevelError
default:
lvl = slog.LevelInfo
}
handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
Level: lvl,
ReplaceAttr: func(_ []string, a slog.Attr) slog.Attr {
if a.Key == slog.TimeKey {
return slog.Attr{Key: "time", Value: a.Value}
}
if a.Key == slog.MessageKey {
return slog.Attr{Key: "event", Value: a.Value}
}
return a
},
})
return slog.New(handler)
}
// WithRequestID stores the request ID in the context for downstream loggers.
func WithRequestID(ctx context.Context, id string) context.Context {
return context.WithValue(ctx, requestIDKey, id)
}
// RequestIDFrom returns the request ID stored in ctx, or empty string if absent.
func RequestIDFrom(ctx context.Context) string {
if v, ok := ctx.Value(requestIDKey).(string); ok {
return v
}
return ""
}
// FromContext returns a logger pre-bound with the request_id from ctx (if any).
func FromContext(ctx context.Context, base *slog.Logger) *slog.Logger {
if id := RequestIDFrom(ctx); id != "" {
return base.With("request_id", id)
}
return base
}
internal/metrics/metrics.go
+69
View File
@@ -0,0 +1,69 @@
// 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{})
}
internal/selfupdate/selfupdate.go
+56
View File
@@ -0,0 +1,56 @@
// 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)
}
internal/selfupdate/selfupdate_test.go
+87
View File
@@ -0,0 +1,87 @@
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")
}
internal/updater/compose_executor.go
+54
View File
@@ -0,0 +1,54 @@
package updater
import (
"context"
"fmt"
"os/exec"
"strings"
"github.com/shcizo/package-updater/internal/discovery"
)
// ComposeExecutor invokes `docker compose` as a subprocess.
type ComposeExecutor struct{}
// NewComposeExecutor returns an executor that shells out to docker compose.
func NewComposeExecutor() *ComposeExecutor {
return &ComposeExecutor{}
}
// Execute runs `docker compose -f <file>... -p <project> pull <service>`
// followed by `... up -d <service>`. 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
}
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)
cmd := exec.CommandContext(ctx, "docker", cliArgs...)
cmd.Dir = job.WorkingDir
out, err := cmd.CombinedOutput()
if err != nil {
return fmt.Errorf("docker %s: %w (output: %s)",
strings.Join(args, " "), err, strings.TrimSpace(string(out)))
}
return nil
}
internal/updater/executor.go
+15
View File
@@ -0,0 +1,15 @@
// 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
}
internal/updater/queue.go
+144
View File
@@ -0,0 +1,144 @@
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
}
internal/updater/queue_test.go
+129
View File
@@ -0,0 +1,129 @@
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)
}
internal/updater/worker.go
+6
View File
@@ -0,0 +1,6 @@
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.
internal/updater/worker_test.go
+38
View File
@@ -0,0 +1,38 @@
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())
}