From 5bf2f4cd526a94541da4d27777f46ea9f2f61584 Mon Sep 17 00:00:00 2001 From: Josh Sandlin Date: Tue, 28 Jul 2026 15:54:34 -0400 Subject: [PATCH] paste in sidecar code / build and account for ansible tweaks to settle it in --- .github/workflows/deploy.yml | 4 +- DEPLOYING.md | 5 +- ansible/templates/zot-config.json.j2 | 12 + build-and-push.sh | 8 +- docker-compose.yaml | 38 ++ sidecar/.dockerignore | 22 + sidecar/.golangci.yml | 13 + sidecar/Dockerfile | 80 ++++ sidecar/Makefile | 41 ++ sidecar/cmd/zot-ephemeral-ttl/main.go | 39 ++ sidecar/go.mod | 64 +++ sidecar/go.sum | 151 +++++++ sidecar/internal/app/app.go | 80 ++++ sidecar/internal/app/app_test.go | 82 ++++ sidecar/internal/config/config.go | 76 ++++ sidecar/internal/config/config_test.go | 136 ++++++ sidecar/internal/events/cloudevents.go | 69 +++ sidecar/internal/events/cloudevents_test.go | 119 +++++ sidecar/internal/reaper/reaper.go | 83 ++++ sidecar/internal/reaper/reaper_test.go | 201 +++++++++ sidecar/internal/redistest/redistest.go | 73 +++ sidecar/internal/registry/registry.go | 51 +++ sidecar/internal/registry/registry_test.go | 79 ++++ sidecar/internal/server/server.go | 104 +++++ sidecar/internal/server/server_test.go | 219 +++++++++ sidecar/internal/store/store.go | 184 ++++++++ sidecar/internal/store/store_test.go | 467 ++++++++++++++++++++ sidecar/internal/ttl/ttl.go | 62 +++ sidecar/internal/ttl/ttl_test.go | 92 ++++ 29 files changed, 2646 insertions(+), 8 deletions(-) create mode 100644 sidecar/.dockerignore create mode 100644 sidecar/.golangci.yml create mode 100644 sidecar/Dockerfile create mode 100644 sidecar/Makefile create mode 100644 sidecar/cmd/zot-ephemeral-ttl/main.go create mode 100644 sidecar/go.mod create mode 100644 sidecar/go.sum create mode 100644 sidecar/internal/app/app.go create mode 100644 sidecar/internal/app/app_test.go create mode 100644 sidecar/internal/config/config.go create mode 100644 sidecar/internal/config/config_test.go create mode 100644 sidecar/internal/events/cloudevents.go create mode 100644 sidecar/internal/events/cloudevents_test.go create mode 100644 sidecar/internal/reaper/reaper.go create mode 100644 sidecar/internal/reaper/reaper_test.go create mode 100644 sidecar/internal/redistest/redistest.go create mode 100644 sidecar/internal/registry/registry.go create mode 100644 sidecar/internal/registry/registry_test.go create mode 100644 sidecar/internal/server/server.go create mode 100644 sidecar/internal/server/server_test.go create mode 100644 sidecar/internal/store/store.go create mode 100644 sidecar/internal/store/store_test.go create mode 100644 sidecar/internal/ttl/ttl.go create mode 100644 sidecar/internal/ttl/ttl_test.go diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 5457340..6156038 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -38,8 +38,8 @@ jobs: - name: Build and push images run: | - docker compose build web - docker compose push web + docker compose build web zot-ephemeral-ttl + docker compose push web zot-ephemeral-ttl deploy: name: Deploy to Production diff --git a/DEPLOYING.md b/DEPLOYING.md index e63a813..01b288c 100644 --- a/DEPLOYING.md +++ b/DEPLOYING.md @@ -6,9 +6,10 @@ already-running Ubuntu host that you can reach as `root` over SSH, listed in `ansible/inventory/group_vars/all.yml` (that value is what the Cloudflare A record is pointed at). -## Build the ttl.sh web image +## Build the ttl.sh images -Only the `web` service is built here — zot is used off the shelf. +The `web` and `zot-ephemeral-ttl` services are built here — zot and redis are used +off the shelf. 1. Docker and Docker Compose are installed 2. Authenticated to `ghcr.io` (`docker login ghcr.io`) diff --git a/ansible/templates/zot-config.json.j2 b/ansible/templates/zot-config.json.j2 index 8823c53..3911c2b 100644 --- a/ansible/templates/zot-config.json.j2 +++ b/ansible/templates/zot-config.json.j2 @@ -20,5 +20,17 @@ }, "log": { "level": "info" + }, + "extensions": { + "events": { + "enable": true, + "sinks": [ + { + "type": "http", + "address": "http://zot-ephemeral-ttl:8080/events", + "timeout": "5s" + } + ] + } } } diff --git a/build-and-push.sh b/build-and-push.sh index 668b450..954bc53 100755 --- a/build-and-push.sh +++ b/build-and-push.sh @@ -1,7 +1,7 @@ #!/usr/bin/env bash -# builds, tags, and pushes the web image referenced in docker-compose.yaml -# (zot is used off the shelf and is never built here) +# builds, tags, and pushes the images referenced in docker-compose.yaml +# (zot and redis are used off the shelf and are never built here) -docker compose build web -docker compose push web \ No newline at end of file +docker compose build web zot-ephemeral-ttl +docker compose push web zot-ephemeral-ttl \ No newline at end of file diff --git a/docker-compose.yaml b/docker-compose.yaml index d6183f6..ed6312a 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -26,6 +26,44 @@ services: networks: - ttlsh-network + # Tag-driven TTL sidecar. Zot posts image.updated CloudEvents to it (see the + # events extension in zot-config.json) and it DELETEs manifests once expired. + zot-ephemeral-ttl: + image: ghcr.io/replicatedhq/ttlsh-ephemeral-ttl:latest + build: + context: sidecar + dockerfile: Dockerfile + platforms: + - "linux/amd64" + container_name: ttlsh-ephemeral-ttl + restart: always + pull_policy: always + networks: + - ttlsh-network + depends_on: + redis: + condition: service_healthy + + # Backing store for the sidecar's expiry index. + redis: + image: redis:8-alpine + container_name: ttlsh-redis + restart: always + command: ["redis-server", "--appendonly", "yes"] + volumes: + - redis-data:/data + networks: + - ttlsh-network + healthcheck: + test: ["CMD", "redis-cli", "ping"] + interval: 5s + timeout: 3s + retries: 30 + start_period: 2s + networks: ttlsh-network: driver: bridge + +volumes: + redis-data: diff --git a/sidecar/.dockerignore b/sidecar/.dockerignore new file mode 100644 index 0000000..8951a75 --- /dev/null +++ b/sidecar/.dockerignore @@ -0,0 +1,22 @@ +# Keep build context lean. Anything not needed by `go build` lives here. +.git +.gitignore +.github +.vscode +.idea +*.md +LICENSE + +# Local build artifacts. +bin/ +dist/ +out/ +*.test +*.out +coverage.txt +coverage.html + +# Editor / OS junk. +.DS_Store +*.swp +*~ diff --git a/sidecar/.golangci.yml b/sidecar/.golangci.yml new file mode 100644 index 0000000..ac58e26 --- /dev/null +++ b/sidecar/.golangci.yml @@ -0,0 +1,13 @@ +version: "2" + +linters: + default: standard + enable: + - bodyclose + - errorlint + - misspell + - unconvert + +formatters: + enable: + - gofmt diff --git a/sidecar/Dockerfile b/sidecar/Dockerfile new file mode 100644 index 0000000..3b2a08a --- /dev/null +++ b/sidecar/Dockerfile @@ -0,0 +1,80 @@ +# syntax=docker/dockerfile:1.7 +# +# Multi-stage build for the zot-ephemeral-ttl sidecar. This file builds; it +# does not gate — formatting, vet, lint, and tests run in CI. + +# ---- build arguments ------------------------------------------------------- +ARG GO_VERSION=1.26 +ARG ALPINE_VERSION=3.22 + +# Cross-compilation: BuildKit injects TARGETOS/TARGETARCH for the +# requested --platform; default to the build host's values when invoked +# without --platform (e.g. plain `docker compose build`). +ARG TARGETOS=linux +ARG TARGETARCH + +# Version metadata wired into the binary via -ldflags. +ARG VERSION=dev +ARG COMMIT=unknown +ARG BUILD_DATE=unknown + + +# ---- deps: download modules once and cache them --------------------------- +FROM --platform=$BUILDPLATFORM golang:${GO_VERSION}-alpine${ALPINE_VERSION} AS deps +WORKDIR /src + +# git for VCS info, ca-certs in case `go mod download` talks to a private proxy. +RUN apk add --no-cache git ca-certificates + +COPY go.mod go.sum ./ +RUN --mount=type=cache,target=/go/pkg/mod \ + go mod download -x + + +# ---- build: produce the static binary ------------------------------------- +FROM deps AS build +COPY . . +ARG TARGETOS +ARG TARGETARCH +ARG VERSION +ARG COMMIT +ARG BUILD_DATE + +# CGO_ENABLED=0 + pure-Go deps => the binary runs from `FROM scratch`. +# -buildvcs=false keeps host git state out; the metadata is set via -X below. +RUN --mount=type=cache,target=/go/pkg/mod \ + --mount=type=cache,target=/root/.cache/go-build \ + CGO_ENABLED=0 GOOS=${TARGETOS} GOARCH=${TARGETARCH} \ + go build \ + -trimpath \ + -buildvcs=false \ + -ldflags="-s -w \ + -X main.version=${VERSION} \ + -X main.commit=${COMMIT} \ + -X main.buildDate=${BUILD_DATE}" \ + -o /out/zot-ephemeral-ttl \ + ./cmd/zot-ephemeral-ttl + + +# ---- runtime: scratch image with just the binary -------------------------- +FROM scratch AS runtime + +ARG VERSION +ARG COMMIT +ARG BUILD_DATE + +LABEL org.opencontainers.image.title="zot-ephemeral-ttl" \ + org.opencontainers.image.description="Tag-driven TTL sidecar for zot. Subscribes to image.updated CloudEvents and DELETEs expired manifests." \ + org.opencontainers.image.source="https://github.com/nullbytelabs/zot-ephemeral-ttl" \ + org.opencontainers.image.licenses="Apache-2.0" \ + org.opencontainers.image.version="${VERSION}" \ + org.opencontainers.image.revision="${COMMIT}" \ + org.opencontainers.image.created="${BUILD_DATE}" + +COPY --from=build /out/zot-ephemeral-ttl /zot-ephemeral-ttl + +# scratch has no shell, so HEALTHCHECK is intentionally omitted; probe +# /healthz from outside. + +EXPOSE 8080 +ENTRYPOINT ["/zot-ephemeral-ttl"] diff --git a/sidecar/Makefile b/sidecar/Makefile new file mode 100644 index 0000000..f89b028 --- /dev/null +++ b/sidecar/Makefile @@ -0,0 +1,41 @@ +# The Go toolchain is the primary interface to this repo: `go test ./...` works +# on a clean checkout. This file covers the two things that need more than a +# one-word go command — the checks CI gates on, and a version-stamped image. + +IMAGE ?= zot-ephemeral-ttl +TAG ?= dev + +.DEFAULT_GOAL := test + +.PHONY: test +test: + go test -race ./... + +# Everything CI gates on, in the order CI runs it. +.PHONY: check +check: + @unformatted=$$(gofmt -l .); \ + if [ -n "$$unformatted" ]; then \ + echo "not gofmt-formatted:"; echo "$$unformatted"; \ + echo "run 'gofmt -w .' to fix."; exit 1; \ + fi + go vet ./... + @cp go.mod .go.mod.tidycheck && cp go.sum .go.sum.tidycheck; \ + go mod tidy; \ + rc=0; \ + if ! cmp -s go.mod .go.mod.tidycheck || ! cmp -s go.sum .go.sum.tidycheck; then \ + echo "go.mod/go.sum were not tidy; 'go mod tidy' has fixed them — commit the result."; \ + rc=1; \ + fi; \ + rm -f .go.mod.tidycheck .go.sum.tidycheck; \ + exit $$rc + go test -race ./... + +.PHONY: image +image: + docker build \ + --build-arg VERSION=$$(git describe --tags --always 2>/dev/null || echo dev) \ + --build-arg COMMIT=$$(git rev-parse --short HEAD 2>/dev/null || echo unknown) \ + --build-arg BUILD_DATE=$$(date -u +%Y-%m-%dT%H:%M:%SZ) \ + --tag $(IMAGE):$(TAG) \ + . diff --git a/sidecar/cmd/zot-ephemeral-ttl/main.go b/sidecar/cmd/zot-ephemeral-ttl/main.go new file mode 100644 index 0000000..e4b10f5 --- /dev/null +++ b/sidecar/cmd/zot-ephemeral-ttl/main.go @@ -0,0 +1,39 @@ +// Command zot-ephemeral-ttl is a sidecar for zot that deletes tags whose names +// encode a TTL (e.g. ":1h", ":30m", ":7d"). It loads config, handles signals, +// and hands off to package app; package ttl documents the expiry policy. +package main + +import ( + "context" + "log" + "os/signal" + "syscall" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/app" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/config" +) + +// Build metadata, set via -ldflags "-X main.version=..." in the Dockerfile. +var ( + version = "dev" + commit = "unknown" + buildDate = "unknown" +) + +func main() { + log.SetFlags(log.LstdFlags | log.LUTC) + cfg, err := config.Load() + if err != nil { + log.Fatalf("config: %v", err) + } + log.Printf("zot-ephemeral-ttl %s (commit=%s built=%s) starting: listen=%s redis=%s sweep=%s zot=%s default_ttl=%s max_ttl=%s", + version, commit, buildDate, + cfg.Listen, cfg.RedactedRedisURL(), cfg.SweepInterval, cfg.ZotURL, cfg.DefaultTTL, cfg.MaxTTL) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + if err := app.Run(ctx, cfg); err != nil { + log.Fatalf("run: %v", err) + } +} diff --git a/sidecar/go.mod b/sidecar/go.mod new file mode 100644 index 0000000..8c6f48b --- /dev/null +++ b/sidecar/go.mod @@ -0,0 +1,64 @@ +module github.com/nullbytelabs/zot-ephemeral-ttl + +go 1.26 + +require ( + github.com/redis/go-redis/v9 v9.21.0 + github.com/testcontainers/testcontainers-go v0.43.0 + github.com/testcontainers/testcontainers-go/modules/redis v0.43.0 +) + +require ( + dario.cat/mergo v1.0.2 // indirect + github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c // indirect + github.com/Microsoft/go-winio v0.6.2 // indirect + github.com/cenkalti/backoff/v4 v4.3.0 // 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/containerd/platforms v0.2.1 // indirect + github.com/cpuguy83/dockercfg v0.3.2 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/distribution/reference v0.6.0 // indirect + github.com/docker/go-connections v0.6.0 // indirect + github.com/docker/go-units v0.5.0 // indirect + github.com/ebitengine/purego v0.10.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/go-ole/go-ole v1.2.6 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/klauspost/compress v1.18.5 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/magiconair/properties v1.8.10 // indirect + github.com/mdelapenya/tlscert v0.2.0 // indirect + github.com/moby/docker-image-spec v1.3.1 // indirect + github.com/moby/go-archive v0.2.0 // indirect + github.com/moby/moby/api v1.54.2 // indirect + github.com/moby/moby/client v0.4.0 // indirect + github.com/moby/patternmatcher v0.6.1 // indirect + github.com/moby/sys/sequential v0.6.0 // indirect + github.com/moby/sys/user v0.4.0 // indirect + github.com/moby/sys/userns v0.1.0 // indirect + github.com/moby/term v0.5.2 // indirect + github.com/opencontainers/go-digest v1.0.0 // indirect + github.com/opencontainers/image-spec v1.1.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect + github.com/shirou/gopsutil/v4 v4.26.5 // indirect + github.com/sirupsen/logrus v1.9.4 // indirect + github.com/stretchr/testify v1.11.1 // indirect + github.com/tklauser/go-sysconf v0.3.16 // indirect + github.com/tklauser/numcpus v0.11.0 // indirect + github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0 // indirect + go.opentelemetry.io/otel v1.41.0 // indirect + go.opentelemetry.io/otel/metric v1.41.0 // indirect + go.opentelemetry.io/otel/trace v1.41.0 // indirect + go.uber.org/atomic v1.11.0 // indirect + golang.org/x/crypto v0.54.0 // indirect + golang.org/x/sys v0.47.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/sidecar/go.sum b/sidecar/go.sum new file mode 100644 index 0000000..92cf820 --- /dev/null +++ b/sidecar/go.sum @@ -0,0 +1,151 @@ +dario.cat/mergo v1.0.2 h1:85+piFYR1tMbRrLcDwR18y4UKJ3aH1Tbzi24VRW1TK8= +dario.cat/mergo v1.0.2/go.mod h1:E/hbnu0NxMFBjpMIE34DRGLWqDy0g5FuKDhCb31ngxA= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6 h1:He8afgbRMd7mFxO99hRNu+6tazq8nFF9lIwo9JFroBk= +github.com/AdaLogics/go-fuzz-headers v0.0.0-20240806141605-e8a1dd7889d6/go.mod h1:8o94RPi1/7XTJvwPpRSzSUedZrtlirdB3r9Z20bi2f8= +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/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs= +github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c= +github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA= +github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= +github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= +github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +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/containerd/platforms v0.2.1 h1:zvwtM3rz2YHPQsF2CHYM8+KtB5dvhISiXh5ZpSBQv6A= +github.com/containerd/platforms v0.2.1/go.mod h1:XHCb+2/hzowdiut9rkudds9bE5yJ7npe7dG/wG+uFPw= +github.com/cpuguy83/dockercfg v0.3.2 h1:DlJTyZGBDlXqUZ2Dk2Q3xHs/FtnooJJVaad2S9GKorA= +github.com/cpuguy83/dockercfg v0.3.2/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc= +github.com/creack/pty v1.1.24 h1:bJrF4RRfyJnbTJqzRLHzcGaZK1NeM5kTC9jGgovnR1s= +github.com/creack/pty v1.1.24/go.mod h1:08sCNb52WyoAwi2QDyzUCTgcvVFhUzewun7wtTfvcwE= +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/go-connections v0.6.0 h1:LlMG9azAe1TqfR7sO+NJttz1gy6KO7VJBh+pMmjSD94= +github.com/docker/go-connections v0.6.0/go.mod h1:AahvXYshr6JgfUJGdDCs2b5EZG/vmaMAntpSFH5BFKE= +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/ebitengine/purego v0.10.0 h1:QIw4xfpWT6GWTzaW5XEKy3HXoqrJGx1ijYHzTF0/ISU= +github.com/ebitengine/purego v0.10.0/go.mod h1:iIjxzd6CiRiOG0UyXP+V1+jWqUXVjPKLAI0mRfJZTmQ= +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/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= +github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +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/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/klauspost/cpuid/v2 v2.2.10 h1:tBs3QSyvjDyFTq3uoc/9xFpCuOsJQFNPiAhYdw2skhE= +github.com/klauspost/cpuid/v2 v2.2.10/go.mod h1:hqwkgyIinND0mEev00jJYCxPNVRVXFQeu1XKlok6oO0= +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/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= +github.com/magiconair/properties v1.8.10 h1:s31yESBquKXCV9a/ScB3ESkOjUYYv+X0rg8SYxI99mE= +github.com/magiconair/properties v1.8.10/go.mod h1:Dhd985XPs7jluiymwWYZ0G4Z61jb3vdS329zhj2hYo0= +github.com/mdelapenya/tlscert v0.2.0 h1:7H81W6Z/4weDvZBNOfQte5GpIMo0lGYEeWbkGp5LJHI= +github.com/mdelapenya/tlscert v0.2.0/go.mod h1:O4njj3ELLnJjGdkN7M/vIVCpZ+Cf0L6muqOG4tLSl8o= +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/go-archive v0.2.0 h1:zg5QDUM2mi0JIM9fdQZWC7U8+2ZfixfTYoHL7rWUcP8= +github.com/moby/go-archive v0.2.0/go.mod h1:mNeivT14o8xU+5q1YnNrkQVpK+dnNe/K6fHqnTg4qPU= +github.com/moby/moby/api v1.54.2 h1:wiat9QAhnDQjA7wk1kh/TqHz2I1uUA7M7t9SAl/JNXg= +github.com/moby/moby/api v1.54.2/go.mod h1:+RQ6wluLwtYaTd1WnPLykIDPekkuyD/ROWQClE83pzs= +github.com/moby/moby/client v0.4.0 h1:S+2XegzHQrrvTCvF6s5HFzcrywWQmuVnhOXe2kiWjIw= +github.com/moby/moby/client v0.4.0/go.mod h1:QWPbvWchQbxBNdaLSpoKpCdf5E+WxFAgNHogCWDoa7g= +github.com/moby/patternmatcher v0.6.1 h1:qlhtafmr6kgMIJjKJMDmMWq7WLkKIo23hsrpR3x084U= +github.com/moby/patternmatcher v0.6.1/go.mod h1:hDPoyOpDY7OrrMDLaYoY3hf52gNCR/YOUYxkhApJIxc= +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/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= +github.com/moby/sys/user v0.4.0/go.mod h1:bG+tYYYJgaMtRKgEmuueC0hJEAZWwtIbZTB+85uoHjs= +github.com/moby/sys/userns v0.1.0 h1:tVLXkFOxVu9A64/yh59slHVv9ahO9UIev4JZusOLG/g= +github.com/moby/sys/userns v0.1.0/go.mod h1:IHUYgu/kao6N8YZlp9Cf444ySSvCmDlmzUcYfDHOl28= +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/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/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 h1:o4JXh1EVt9k/+g42oCprj/FisM4qX9L3sZB3upGN2ZU= +github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/redis/go-redis/v9 v9.21.0 h1:FPBE4hhbAke+TLmcY3WkpbDffJEomdqPn3HYiqAtL9E= +github.com/redis/go-redis/v9 v9.21.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA= +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/shirou/gopsutil/v4 v4.26.5 h1:RPcBXkpz7kOj9PqGFQOlBPZHsyaPvPVQc098y9RmCNM= +github.com/shirou/gopsutil/v4 v4.26.5/go.mod h1:LZ6ewCSkBqUpvSOf+LsTGnRinC6iaNUNMGBtDkJBaLQ= +github.com/sirupsen/logrus v1.9.4 h1:TsZE7l11zFCLZnZ+teH4Umoq5BhEIfIzfRDZ1Uzql2w= +github.com/sirupsen/logrus v1.9.4/go.mod h1:ftWc9WdOfJ0a92nsE2jF5u5ZwH8Bv2zdeOC42RjbV2g= +github.com/stretchr/objx v0.5.3 h1:jmXUvGomnU1o3W/V5h2VEradbpJDwGrzugQQvL0POH4= +github.com/stretchr/objx v0.5.3/go.mod h1:rDQraq+vQZU7Fde9LOZLr8Tax6zZvy4kuNKF+QYS+U0= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/testcontainers/testcontainers-go v0.43.0 h1:oEQx5MW2DGd9z3AeEQfB2lPM0eLs7ztyaGRu75bFo5A= +github.com/testcontainers/testcontainers-go v0.43.0/go.mod h1:+VxkT2NQnKOZPKi6praMuMKYHYyOGXr0XSBSlSMCzFo= +github.com/testcontainers/testcontainers-go/modules/redis v0.43.0 h1:qzATMhrltLr07KcGl/d674ouqI0AFtf6wnQb3VnqP7M= +github.com/testcontainers/testcontainers-go/modules/redis v0.43.0/go.mod h1:ygEcEUIZzmIlOKpjBfnPn/lUIRNorr1kPj3XfFPTQXM= +github.com/tklauser/go-sysconf v0.3.16 h1:frioLaCQSsF5Cy1jgRBrzr6t502KIIwQ0MArYICU0nA= +github.com/tklauser/go-sysconf v0.3.16/go.mod h1:/qNL9xxDhc7tx3HSRsLWNnuzbVfh3e7gh/BmM179nYI= +github.com/tklauser/numcpus v0.11.0 h1:nSTwhKH5e1dMNsCdVBukSZrURJRoHbSEQjdEbY+9RXw= +github.com/tklauser/numcpus v0.11.0/go.mod h1:z+LwcLq54uWZTX0u/bGobaV34u6V7KNlTZejzM6/3MQ= +github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= +github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +github.com/zeebo/xxh3 v1.1.0 h1:s7DLGDK45Dyfg7++yxI0khrfwq9661w9EN78eP/UZVs= +github.com/zeebo/xxh3 v1.1.0/go.mod h1:IisAie1LELR4xhVinxWS5+zf1lA4p0MW4T+w+W07F5s= +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.60.0 h1:sbiXRNDSWJOTobXh5HyQKjq6wUC5tNybqjIqDpAY4CU= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.60.0/go.mod h1:69uWxva0WgAA/4bu2Yy70SLDBwZXuQ6PbBpbsa5iZrQ= +go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= +go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= +go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= +go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= +go.opentelemetry.io/otel/sdk v1.35.0 h1:iPctf8iprVySXSKJffSS79eOjl9pvxV9ZqOWT0QejKY= +go.opentelemetry.io/otel/sdk v1.35.0/go.mod h1:+ga1bZliga3DxJ3CQGg3updiaAJoNECOgJREo9KHGQg= +go.opentelemetry.io/otel/sdk/metric v1.35.0 h1:1RriWBmCKgkeHEhM7a2uMjMUfP7MsOF5JpUCaEqEI9o= +go.opentelemetry.io/otel/sdk/metric v1.35.0/go.mod h1:is6XYCUMpcKi+ZsOvfluY5YstFnhW0BidkR+gL+qN+w= +go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= +go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw= +golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk= +golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.45.0 h1:NwWyBmoJCbfTHpxrWoZ9C6/VxOf7ic219I8xZZFdrf0= +golang.org/x/term v0.45.0/go.mod h1:9aqxs0blBcrm/n0L9QW0aRVD+ktan8ssZromtqJC43w= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +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= +pgregory.net/rapid v1.2.0 h1:keKAYRcjm+e1F0oAuU5F5+YPAWcyxNNRK2wud503Gnk= +pgregory.net/rapid v1.2.0/go.mod h1:PY5XlDGj0+V1FCq0o192FdRhpKHGTRIWBgqjDBTrq04= diff --git a/sidecar/internal/app/app.go b/sidecar/internal/app/app.go new file mode 100644 index 0000000..0082b50 --- /dev/null +++ b/sidecar/internal/app/app.go @@ -0,0 +1,80 @@ +// Package app wires the concrete components together and runs them until the +// context is cancelled. It is separate from package main so the full +// startup/shutdown path can be tested; main only loads config and handles +// signals. +package app + +import ( + "context" + "errors" + "io" + "log" + "net" + "net/http" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/config" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/reaper" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/registry" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/server" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/store" +) + +// storeBackend is the union of persistence capabilities app injects into the +// server and reaper, so neither consumer names the concrete adapter. +type storeBackend interface { + server.Store // Upsert + reaper.Store // Expired, Delete + io.Closer +} + +// Run connects to Redis, starts the sweep loop and HTTP server, and blocks +// until ctx is cancelled (or the HTTP server fails). It returns the first +// error encountered during startup or shutdown. +func Run(ctx context.Context, cfg config.Config) error { + var st storeBackend + st, err := store.Open(cfg.RedisURL) + if err != nil { + return err + } + defer func() { + if err := st.Close(); err != nil { + log.Printf("closing store: %v", err) + } + }() + + // Bind up front so a bad address fails synchronously and a ":0" caller can + // read back the chosen port. + ln, err := net.Listen("tcp", cfg.Listen) + if err != nil { + return err + } + + srv := server.New(st, cfg.DefaultTTL, cfg.MaxTTL, time.Now) + httpSrv := &http.Server{ + Handler: srv.Routes(), + ReadHeaderTimeout: 10 * time.Second, + } + + rp := reaper.New(cfg.SweepInterval, st, registry.New(cfg.ZotURL)) + go rp.Run(ctx) + + serveErr := make(chan error, 1) + go func() { + log.Printf("listening on %s", ln.Addr()) + serveErr <- httpSrv.Serve(ln) + }() + + select { + case <-ctx.Done(): + case err := <-serveErr: + if err != nil && !errors.Is(err, http.ErrServerClosed) { + return err + } + } + + log.Printf("shutting down") + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return httpSrv.Shutdown(shutdownCtx) +} diff --git a/sidecar/internal/app/app_test.go b/sidecar/internal/app/app_test.go new file mode 100644 index 0000000..5858093 --- /dev/null +++ b/sidecar/internal/app/app_test.go @@ -0,0 +1,82 @@ +package app + +import ( + "context" + "os" + "testing" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/config" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/redistest" +) + +// The startup-path test needs a real Redis; redistest starts one per package. +func TestMain(m *testing.M) { os.Exit(redistest.Run(m)) } + +// baseConfig returns a config pointing at that server. +func baseConfig(t *testing.T) config.Config { + t.Helper() + return config.Config{ + Listen: "127.0.0.1:0", + RedisURL: redistest.URL(t), + SweepInterval: time.Hour, + ZotURL: "http://127.0.0.1:1", + DefaultTTL: 24 * time.Hour, + MaxTTL: 24 * time.Hour, + } +} + +// TestRunCleanShutdown exercises the real startup/shutdown path. SweepInterval +// is an hour so the reaper does its one immediate sweep against an empty store +// and then idles, never reaching the (unreachable) registry. +func TestRunCleanShutdown(t *testing.T) { + cfg := baseConfig(t) + + ctx, cancel := context.WithCancel(context.Background()) + + errCh := make(chan error, 1) + go func() { + errCh <- Run(ctx, cfg) + }() + + // Give Run a moment to bind the listener and start serving. + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case err := <-errCh: + if err != nil { + t.Fatalf("Run returned error on clean shutdown: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Run did not return within 2s after context cancellation") + } +} + +// TestRunStoreOpenFailure points Run at an address nothing is listening on, so +// the store's startup PING fails and Run returns before binding anything. It +// needs no server of its own. +func TestRunStoreOpenFailure(t *testing.T) { + cfg := config.Config{ + Listen: "127.0.0.1:0", + RedisURL: "redis://127.0.0.1:1", // connection refused + SweepInterval: time.Hour, + ZotURL: "http://127.0.0.1:1", + DefaultTTL: 24 * time.Hour, + MaxTTL: 24 * time.Hour, + } + + errCh := make(chan error, 1) + go func() { + errCh <- Run(context.Background(), cfg) + }() + + select { + case err := <-errCh: + if err == nil { + t.Fatal("Run returned nil; expected store-open error") + } + case <-time.After(10 * time.Second): + t.Fatal("Run did not return promptly on store-open failure") + } +} diff --git a/sidecar/internal/config/config.go b/sidecar/internal/config/config.go new file mode 100644 index 0000000..3f99d11 --- /dev/null +++ b/sidecar/internal/config/config.go @@ -0,0 +1,76 @@ +// Package config loads zot-ephemeral-ttl's runtime configuration from the +// environment. All knobs use the ZOT_EPHEMERAL_TTL_ prefix and fall back +// to values suitable for the docker-compose deployment. +package config + +import ( + "errors" + "fmt" + "net/url" + "os" + "strings" + "time" +) + +type Config struct { + Listen string + RedisURL string + SweepInterval time.Duration + ZotURL string + DefaultTTL time.Duration + MaxTTL time.Duration +} + +// Load reads configuration from the environment, applying defaults for any +// unset values. Every malformed duration is reported, not just the first. +func Load() (Config, error) { + var errs []error + dur := func(key string, def time.Duration) time.Duration { + d, err := envDur(key, def) + if err != nil { + errs = append(errs, err) + return def + } + return d + } + + cfg := Config{ + Listen: envStr("ZOT_EPHEMERAL_TTL_LISTEN", ":8080"), + RedisURL: envStr("ZOT_EPHEMERAL_TTL_REDIS_URL", "redis://redis:6379/0"), + SweepInterval: dur("ZOT_EPHEMERAL_TTL_SWEEP_INTERVAL", 30*time.Second), + ZotURL: strings.TrimRight(envStr("ZOT_EPHEMERAL_TTL_ZOT_URL", "http://zot:5000"), "/"), + DefaultTTL: dur("ZOT_EPHEMERAL_TTL_DEFAULT_TTL", 24*time.Hour), + MaxTTL: dur("ZOT_EPHEMERAL_TTL_MAX_TTL", 24*time.Hour), + } + return cfg, errors.Join(errs...) +} + +// RedactedRedisURL returns RedisURL with any password replaced by "xxxxx", +// safe to write to logs. An unparseable URL is reported as such rather than +// echoed, since it may still contain a credential. +func (c Config) RedactedRedisURL() string { + u, err := url.Parse(c.RedisURL) + if err != nil { + return "(unparseable redis url)" + } + return u.Redacted() +} + +func envStr(key, def string) string { + if v := os.Getenv(key); v != "" { + return v + } + return def +} + +func envDur(key string, def time.Duration) (time.Duration, error) { + v := os.Getenv(key) + if v == "" { + return def, nil + } + d, err := time.ParseDuration(v) + if err != nil { + return 0, fmt.Errorf("invalid duration for %s=%q: %w", key, v, err) + } + return d, nil +} diff --git a/sidecar/internal/config/config_test.go b/sidecar/internal/config/config_test.go new file mode 100644 index 0000000..91c145b --- /dev/null +++ b/sidecar/internal/config/config_test.go @@ -0,0 +1,136 @@ +package config + +import ( + "strings" + "testing" + "time" +) + +func TestEnvStr(t *testing.T) { + t.Setenv("X_ZOT_EPHEMERAL_TTL_TEST", "") + if got := envStr("X_ZOT_EPHEMERAL_TTL_TEST", "fallback"); got != "fallback" { + t.Errorf("empty env should return default, got %q", got) + } + t.Setenv("X_ZOT_EPHEMERAL_TTL_TEST", "value") + if got := envStr("X_ZOT_EPHEMERAL_TTL_TEST", "fallback"); got != "value" { + t.Errorf("set env should win, got %q", got) + } +} + +func TestLoad(t *testing.T) { + keys := []string{ + "ZOT_EPHEMERAL_TTL_LISTEN", + "ZOT_EPHEMERAL_TTL_REDIS_URL", + "ZOT_EPHEMERAL_TTL_SWEEP_INTERVAL", + "ZOT_EPHEMERAL_TTL_ZOT_URL", + "ZOT_EPHEMERAL_TTL_DEFAULT_TTL", + "ZOT_EPHEMERAL_TTL_MAX_TTL", + } + + t.Run("defaults", func(t *testing.T) { + for _, k := range keys { + t.Setenv(k, "") + } + got, err := Load() + if err != nil { + t.Fatalf("defaults: unexpected error: %v", err) + } + want := Config{ + Listen: ":8080", + RedisURL: "redis://redis:6379/0", + SweepInterval: 30 * time.Second, + ZotURL: "http://zot:5000", + DefaultTTL: 24 * time.Hour, + MaxTTL: 24 * time.Hour, + } + if got != want { + t.Fatalf("defaults: got %+v want %+v", got, want) + } + }) + + t.Run("overrides", func(t *testing.T) { + t.Setenv("ZOT_EPHEMERAL_TTL_LISTEN", "127.0.0.1:9090") + t.Setenv("ZOT_EPHEMERAL_TTL_REDIS_URL", "rediss://cache.internal:6380/3") + t.Setenv("ZOT_EPHEMERAL_TTL_SWEEP_INTERVAL", "5s") + t.Setenv("ZOT_EPHEMERAL_TTL_ZOT_URL", "http://zot:5000/") + t.Setenv("ZOT_EPHEMERAL_TTL_DEFAULT_TTL", "1h") + t.Setenv("ZOT_EPHEMERAL_TTL_MAX_TTL", "2h") + got, err := Load() + if err != nil { + t.Fatalf("overrides: unexpected error: %v", err) + } + want := Config{ + Listen: "127.0.0.1:9090", + RedisURL: "rediss://cache.internal:6380/3", + SweepInterval: 5 * time.Second, + ZotURL: "http://zot:5000", + DefaultTTL: time.Hour, + MaxTTL: 2 * time.Hour, + } + if got != want { + t.Fatalf("overrides: got %+v want %+v", got, want) + } + }) +} + +// TestRedactedRedisURL: the startup banner logs the Redis URL, which commonly +// carries a password, so it must go through redaction first. +func TestRedactedRedisURL(t *testing.T) { + tests := []struct { + name string + in string + want string + }{ + {"no credentials", "redis://redis:6379/0", "redis://redis:6379/0"}, + {"password only", "redis://:hunter2@redis:6379/0", "redis://:xxxxx@redis:6379/0"}, + {"user and password", "rediss://app:hunter2@cache:6380/1", "rediss://app:xxxxx@cache:6380/1"}, + {"unparseable", "redis://%zz@host", "(unparseable redis url)"}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if got := (Config{RedisURL: tc.in}).RedactedRedisURL(); got != tc.want { + t.Errorf("got %q want %q", got, tc.want) + } + }) + } +} + +// TestLoadReportsEveryBadDuration checks that a malformed duration surfaces as +// an error naming the offending variable, and that all of them are reported +// rather than just the first. +func TestLoadReportsEveryBadDuration(t *testing.T) { + t.Setenv("ZOT_EPHEMERAL_TTL_SWEEP_INTERVAL", "not-a-duration") + t.Setenv("ZOT_EPHEMERAL_TTL_MAX_TTL", "also-bad") + + cfg, err := Load() + if err == nil { + t.Fatal("expected an error for malformed durations, got nil") + } + for _, want := range []string{"ZOT_EPHEMERAL_TTL_SWEEP_INTERVAL", "ZOT_EPHEMERAL_TTL_MAX_TTL"} { + if !strings.Contains(err.Error(), want) { + t.Errorf("error should name %s; got %v", want, err) + } + } + // The bad values fall back to their defaults so the caller sees a usable + // Config alongside the error. + if cfg.SweepInterval != 30*time.Second || cfg.MaxTTL != 24*time.Hour { + t.Errorf("bad durations should fall back to defaults, got %+v", cfg) + } +} + +func TestEnvDur(t *testing.T) { + t.Setenv("X_ZOT_EPHEMERAL_TTL_DUR", "") + got, err := envDur("X_ZOT_EPHEMERAL_TTL_DUR", 5*time.Second) + if err != nil || got != 5*time.Second { + t.Errorf("empty env: got %v, %v", got, err) + } + t.Setenv("X_ZOT_EPHEMERAL_TTL_DUR", "2m") + got, err = envDur("X_ZOT_EPHEMERAL_TTL_DUR", time.Second) + if err != nil || got != 2*time.Minute { + t.Errorf("set env: got %v, %v", got, err) + } + t.Setenv("X_ZOT_EPHEMERAL_TTL_DUR", "nope") + if _, err := envDur("X_ZOT_EPHEMERAL_TTL_DUR", time.Second); err == nil { + t.Error("malformed duration: expected error, got nil") + } +} diff --git a/sidecar/internal/events/cloudevents.go b/sidecar/internal/events/cloudevents.go new file mode 100644 index 0000000..f193ca0 --- /dev/null +++ b/sidecar/internal/events/cloudevents.go @@ -0,0 +1,69 @@ +// Package events parses zot's events-extension CloudEvents. +// +// zot uses the cloudevents/sdk-go HTTP protocol, which defaults to binary +// content mode: event metadata travels in Ce-* headers and the JSON data field +// is the request body. This package also accepts structured mode (Content-Type +// application/cloudevents+json), where the whole envelope is the body. +package events + +import ( + "encoding/json" + "fmt" + "net/http" + "strings" +) + +const ( + ImageUpdatedType = "zotregistry.image.updated" +) + +// ImageUpdatedData is the subset of zot's event payload this service acts on. +// Fields it does not read are deliberately absent: unmarshal ignores unknown +// keys, so declaring them would only add ways for a type mismatch to reject an +// otherwise usable event. +type ImageUpdatedData struct { + Name string `json:"name"` + Reference string `json:"reference"` + Digest string `json:"digest"` +} + +type structuredEnvelope struct { + Type string `json:"type"` + Data json.RawMessage `json:"data"` +} + +// Event is the minimum we need from any incoming POST. +type Event struct { + Type string + Data ImageUpdatedData +} + +func Parse(r *http.Request, body []byte) (Event, error) { + contentType := r.Header.Get("Content-Type") + + // Structured mode: full envelope is the body. + if strings.HasPrefix(contentType, "application/cloudevents+json") { + var env structuredEnvelope + if err := json.Unmarshal(body, &env); err != nil { + return Event{}, fmt.Errorf("decode structured envelope: %w", err) + } + var data ImageUpdatedData + if len(env.Data) > 0 { + if err := json.Unmarshal(env.Data, &data); err != nil { + return Event{}, fmt.Errorf("decode structured data: %w", err) + } + } + return Event{Type: env.Type, Data: data}, nil + } + + // Binary mode: Ce-* headers + JSON body. Header.Get canonicalizes the key, + // so a lowercased ce-type from an intermediary resolves here too. + t := r.Header.Get("Ce-Type") + var data ImageUpdatedData + if len(body) > 0 { + if err := json.Unmarshal(body, &data); err != nil { + return Event{}, fmt.Errorf("decode binary data: %w", err) + } + } + return Event{Type: t, Data: data}, nil +} diff --git a/sidecar/internal/events/cloudevents_test.go b/sidecar/internal/events/cloudevents_test.go new file mode 100644 index 0000000..361c27c --- /dev/null +++ b/sidecar/internal/events/cloudevents_test.go @@ -0,0 +1,119 @@ +package events + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func TestParseBinaryMode(t *testing.T) { + body, _ := json.Marshal(ImageUpdatedData{ + Name: "foo/bar", + Reference: "1h", + Digest: "sha256:deadbeef", + }) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Ce-Type", ImageUpdatedType) + + evt, err := Parse(req, body) + if err != nil { + t.Fatalf("parse: %v", err) + } + if evt.Type != ImageUpdatedType { + t.Errorf("type = %q want %q", evt.Type, ImageUpdatedType) + } + if evt.Data.Name != "foo/bar" || evt.Data.Reference != "1h" { + t.Errorf("data = %+v", evt.Data) + } +} + +func TestParseStructuredMode(t *testing.T) { + // Includes the envelope keys this package deliberately does not model + // (source, id, datacontenttype); they must be ignored, not rejected. + body := []byte(`{ + "type": "` + ImageUpdatedType + `", + "source": "zot", + "id": "abc", + "datacontenttype": "application/json", + "data": {"name":"foo/bar","reference":"30m","digest":"sha256:abc"} + }`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/cloudevents+json; charset=utf-8") + + evt, err := Parse(req, body) + if err != nil { + t.Fatalf("parse: %v", err) + } + if evt.Type != ImageUpdatedType { + t.Errorf("type = %q", evt.Type) + } + if evt.Data.Reference != "30m" { + t.Errorf("reference = %q", evt.Data.Reference) + } +} + +func TestParseStructuredEmptyData(t *testing.T) { + env := structuredEnvelope{Type: "some.other.type"} + body, _ := json.Marshal(env) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/cloudevents+json") + evt, err := Parse(req, body) + if err != nil { + t.Fatalf("parse: %v", err) + } + if evt.Type != "some.other.type" { + t.Errorf("type = %q", evt.Type) + } + if evt.Data.Name != "" { + t.Errorf("expected zero data, got %+v", evt.Data) + } +} + +func TestParseStructuredMalformed(t *testing.T) { + body := []byte(`{not json`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/cloudevents+json") + if _, err := Parse(req, body); err == nil { + t.Fatal("expected error for malformed structured envelope") + } +} + +func TestParseBinaryMalformed(t *testing.T) { + body := []byte(`{not json`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", ImageUpdatedType) + if _, err := Parse(req, body); err == nil { + t.Fatal("expected error for malformed binary body") + } +} + +func TestParseStructuredDataDecodeError(t *testing.T) { + // Valid envelope, but `data` is a JSON array which cannot unmarshal into + // the ImageUpdatedData struct -> exercises the "decode structured data" + // error path. + body := []byte(`{"type":"zotregistry.image.updated","data":[1,2,3]}`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Content-Type", "application/cloudevents+json") + + if _, err := Parse(req, body); err == nil { + t.Fatal("expected error decoding structured data array into struct") + } +} + +func TestParseBinaryEmptyBody(t *testing.T) { + req := httptest.NewRequest(http.MethodPost, "/events", nil) + req.Header.Set("Ce-Type", ImageUpdatedType) + evt, err := Parse(req, nil) + if err != nil { + t.Fatalf("parse: %v", err) + } + if evt.Type != ImageUpdatedType { + t.Errorf("type = %q", evt.Type) + } + if evt.Data.Name != "" { + t.Errorf("expected zero data, got %+v", evt.Data) + } +} diff --git a/sidecar/internal/reaper/reaper.go b/sidecar/internal/reaper/reaper.go new file mode 100644 index 0000000..986512d --- /dev/null +++ b/sidecar/internal/reaper/reaper.go @@ -0,0 +1,83 @@ +// Package reaper implements the background sweep loop that periodically queries +// the store for expired tags and asks the registry to remove their manifests. A +// row is removed from the store only after the registry confirms the manifest +// is gone, so a failed delete is retried on the next tick. +package reaper + +import ( + "context" + "log" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/store" +) + +// Store is the subset of the persistence layer the reaper needs. +type Store interface { + Expired(now time.Time) ([]store.Row, error) + Delete(repo, tag string) error +} + +// ManifestDeleter removes a manifest by tag from the registry. +type ManifestDeleter interface { + DeleteManifest(ctx context.Context, repo, tag string) error +} + +type Reaper struct { + interval time.Duration + store Store + registry ManifestDeleter + clock func() time.Time +} + +// New constructs a Reaper with clock set to time.Now. +func New(interval time.Duration, st Store, reg ManifestDeleter) *Reaper { + return &Reaper{ + interval: interval, + store: st, + registry: reg, + clock: time.Now, + } +} + +// Run sweeps every interval until ctx is cancelled. +func (r *Reaper) Run(ctx context.Context) { + t := time.NewTicker(r.interval) + defer t.Stop() + + // Sweep once on start so a restart doesn't wait a full interval to reap + // already-expired rows. + r.sweepOnce(ctx) + + for { + select { + case <-ctx.Done(): + return + case <-t.C: + r.sweepOnce(ctx) + } + } +} + +func (r *Reaper) sweepOnce(ctx context.Context) { + now := r.clock() + rows, err := r.store.Expired(now) + if err != nil { + log.Printf("sweep: query expired: %v", err) + return + } + if len(rows) == 0 { + return + } + log.Printf("sweep: %d expired tag(s)", len(rows)) + for _, row := range rows { + if err := r.registry.DeleteManifest(ctx, row.Repository, row.Tag); err != nil { + log.Printf("sweep: delete %s:%s: %v", row.Repository, row.Tag, err) + continue + } + log.Printf("sweep: deleted %s:%s", row.Repository, row.Tag) + if err := r.store.Delete(row.Repository, row.Tag); err != nil { + log.Printf("sweep: row delete %s:%s: %v", row.Repository, row.Tag, err) + } + } +} diff --git a/sidecar/internal/reaper/reaper_test.go b/sidecar/internal/reaper/reaper_test.go new file mode 100644 index 0000000..e1c4aab --- /dev/null +++ b/sidecar/internal/reaper/reaper_test.go @@ -0,0 +1,201 @@ +package reaper + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/store" +) + +// In-memory fakes for the reaper's two dependencies. No real DB, no real HTTP. +type fakeStore struct { + mu sync.Mutex + expired []store.Row + expiredErr error + deleted [][2]string // recorded (repo,tag) + deleteErr error +} + +func (f *fakeStore) Expired(now time.Time) ([]store.Row, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.expired, f.expiredErr +} + +func (f *fakeStore) Delete(repo, tag string) error { + f.mu.Lock() + defer f.mu.Unlock() + f.deleted = append(f.deleted, [2]string{repo, tag}) + return f.deleteErr +} + +func (f *fakeStore) deletedCalls() [][2]string { + f.mu.Lock() + defer f.mu.Unlock() + out := make([][2]string, len(f.deleted)) + copy(out, f.deleted) + return out +} + +type fakeDeleter struct { + mu sync.Mutex + calls [][2]string + failOn map[string]bool // key "repo:tag" -> return error +} + +func (d *fakeDeleter) DeleteManifest(_ context.Context, repo, tag string) error { + d.mu.Lock() + defer d.mu.Unlock() + d.calls = append(d.calls, [2]string{repo, tag}) + if d.failOn[repo+":"+tag] { + return errors.New("boom") + } + return nil +} + +func (d *fakeDeleter) callList() [][2]string { + d.mu.Lock() + defer d.mu.Unlock() + out := make([][2]string, len(d.calls)) + copy(out, d.calls) + return out +} + +func row(repo, tag string) store.Row { + return store.Row{ + Repository: repo, + Tag: tag, + ManifestDigest: "sha256:" + repo + tag, + ExpiresAt: 1, + CreatedAt: 0, + } +} + +func TestSweepOnceDeletesExpiredRow(t *testing.T) { + fs := &fakeStore{expired: []store.Row{row("r", "old")}} + fd := &fakeDeleter{} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + if got := fd.callList(); len(got) != 1 || got[0] != [2]string{"r", "old"} { + t.Fatalf("DeleteManifest calls = %v, want one [r old]", got) + } + if got := fs.deletedCalls(); len(got) != 1 || got[0] != [2]string{"r", "old"} { + t.Fatalf("store.Delete calls = %v, want one [r old]", got) + } +} + +func TestSweepOnceExpiredError(t *testing.T) { + fs := &fakeStore{expiredErr: errors.New("db down")} + fd := &fakeDeleter{} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + if got := fd.callList(); len(got) != 0 { + t.Fatalf("DeleteManifest calls = %v, want none", got) + } + if got := fs.deletedCalls(); len(got) != 0 { + t.Fatalf("store.Delete calls = %v, want none", got) + } +} + +func TestSweepOnceNoExpiredRows(t *testing.T) { + fs := &fakeStore{expired: nil} + fd := &fakeDeleter{} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + if got := fd.callList(); len(got) != 0 { + t.Fatalf("DeleteManifest calls = %v, want none", got) + } + if got := fs.deletedCalls(); len(got) != 0 { + t.Fatalf("store.Delete calls = %v, want none", got) + } +} + +func TestSweepOnceDeleterErrorPreservesRow(t *testing.T) { + fs := &fakeStore{expired: []store.Row{row("r", "stuck")}} + fd := &fakeDeleter{failOn: map[string]bool{"r:stuck": true}} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + if got := fd.callList(); len(got) != 1 || got[0] != [2]string{"r", "stuck"} { + t.Fatalf("DeleteManifest calls = %v, want one [r stuck]", got) + } + if got := fs.deletedCalls(); len(got) != 0 { + t.Fatalf("store.Delete calls = %v, want none (row preserved)", got) + } +} + +func TestSweepOnceStoreDeleteError(t *testing.T) { + fs := &fakeStore{ + expired: []store.Row{row("r", "old")}, + deleteErr: errors.New("delete failed"), + } + fd := &fakeDeleter{} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + if got := fd.callList(); len(got) != 1 || got[0] != [2]string{"r", "old"} { + t.Fatalf("DeleteManifest calls = %v, want one [r old]", got) + } + if got := fs.deletedCalls(); len(got) != 1 || got[0] != [2]string{"r", "old"} { + t.Fatalf("store.Delete calls = %v, want one [r old]", got) + } +} + +// A failure on one row must not stop the sweep: both manifest deletes are +// attempted, and only the row that succeeded is removed from the store. +func TestSweepOnceMixedResults(t *testing.T) { + fs := &fakeStore{expired: []store.Row{row("r", "bad"), row("r", "good")}} + fd := &fakeDeleter{failOn: map[string]bool{"r:bad": true}} + rp := New(time.Hour, fs, fd) + + rp.sweepOnce(context.Background()) + + gotCalls := fd.callList() + if len(gotCalls) != 2 || gotCalls[0] != [2]string{"r", "bad"} || gotCalls[1] != [2]string{"r", "good"} { + t.Fatalf("DeleteManifest calls = %v, want [r bad] then [r good]", gotCalls) + } + gotDel := fs.deletedCalls() + if len(gotDel) != 1 || gotDel[0] != [2]string{"r", "good"} { + t.Fatalf("store.Delete calls = %v, want only [r good]", gotDel) + } +} + +func TestRunImmediateSweepAndStopsOnContext(t *testing.T) { + fs := &fakeStore{expired: []store.Row{row("r", "old")}} + fd := &fakeDeleter{} + rp := New(5*time.Millisecond, fs, fd) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + rp.Run(ctx) + close(done) + }() + + time.Sleep(20 * time.Millisecond) + cancel() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("Run did not return after cancel") + } + + if got := fd.callList(); len(got) < 1 { + t.Fatalf("expected at least one DeleteManifest call, got %d", len(got)) + } + if got := fs.deletedCalls(); len(got) < 1 { + t.Fatalf("expected at least one store.Delete call, got %d", len(got)) + } +} diff --git a/sidecar/internal/redistest/redistest.go b/sidecar/internal/redistest/redistest.go new file mode 100644 index 0000000..268ebca --- /dev/null +++ b/sidecar/internal/redistest/redistest.go @@ -0,0 +1,73 @@ +// Package redistest supplies a real Redis to the tests that need one. +// +// The store exists to get Redis semantics right — transactional writes across +// two keys, score-range queries, lexicographic tie-breaks — so faking it would +// only assert what a reimplementation believes those semantics to be. A package +// opts in with a TestMain and gets a throwaway container for the run: +// +// func TestMain(m *testing.M) { os.Exit(redistest.Run(m)) } +// +// One container is shared by all tests in the package, which are responsible +// for leaving the keyspace as they found it. +package redistest + +import ( + "context" + "fmt" + "os" + "testing" + + "github.com/testcontainers/testcontainers-go" + tcredis "github.com/testcontainers/testcontainers-go/modules/redis" +) + +// image pins the server under test. Keep it in step with the Redis the example +// stack runs. +const image = "redis:8-alpine" + +// Exactly one of these is set before tests run, and neither changes after. +var ( + url string + unavailable string +) + +// Run starts a Redis container, runs the package's tests against it, and +// terminates it. Pass the result to os.Exit from TestMain. A container that +// cannot be started is not fatal here; URL accounts for it per test. +func Run(m *testing.M) int { + ctx := context.Background() + + container, err := tcredis.Run(ctx, image) + if err != nil { + unavailable = fmt.Sprintf("could not start %s: %v", image, err) + return m.Run() + } + defer func() { + if err := testcontainers.TerminateContainer(container); err != nil { + fmt.Fprintf(os.Stderr, "redistest: terminating container: %v\n", err) + } + }() + + url, err = container.ConnectionString(ctx) + if err != nil { + unavailable = fmt.Sprintf("could not read connection string: %v", err) + return m.Run() + } + + return m.Run() +} + +// URL returns the address of the Redis started by Run, in a form store.Open +// accepts. It skips the test when there is no server, or fails it under CI +// where a skip would quietly drop the coverage this suite exists to provide. +func URL(t *testing.T) string { + t.Helper() + if url != "" { + return url + } + if os.Getenv("CI") != "" { + t.Fatalf("no Redis available: %s", unavailable) + } + t.Skipf("no Redis available: %s\nThese tests start their own server — check that Docker is running.", unavailable) + return "" +} diff --git a/sidecar/internal/registry/registry.go b/sidecar/internal/registry/registry.go new file mode 100644 index 0000000..ce18f94 --- /dev/null +++ b/sidecar/internal/registry/registry.go @@ -0,0 +1,51 @@ +// Package registry is the adapter to zot's OCI distribution API. It exposes the +// single operation the reaper needs: delete a manifest by tag. +package registry + +import ( + "context" + "fmt" + "io" + "net/http" + "time" +) + +type Client struct { + baseURL string + http *http.Client +} + +// New returns a Client targeting the given zot base URL (e.g. +// "http://zot:5000"). The base URL should not have a trailing slash. +func New(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + http: &http.Client{ + Timeout: 30 * time.Second, + }, + } +} + +// DeleteManifest issues DELETE /v2//manifests/; zot resolves the tag +// to its digest. 200/202/204/404 all mean the tag is gone; any other status is +// a transient error to retry on the next tick. ctx bounds the request, so a +// wedged registry cannot outlive a shutdown. +func (c *Client) DeleteManifest(ctx context.Context, repo, tag string) error { + endpoint := fmt.Sprintf("%s/v2/%s/manifests/%s", c.baseURL, repo, tag) + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, endpoint, nil) + if err != nil { + return err + } + resp, err := c.http.Do(req) + if err != nil { + return err + } + defer func() { _ = resp.Body.Close() }() + _, _ = io.Copy(io.Discard, resp.Body) + switch resp.StatusCode { + case http.StatusOK, http.StatusAccepted, http.StatusNoContent, http.StatusNotFound: + return nil + default: + return fmt.Errorf("DELETE %s -> %d", endpoint, resp.StatusCode) + } +} diff --git a/sidecar/internal/registry/registry_test.go b/sidecar/internal/registry/registry_test.go new file mode 100644 index 0000000..841c80f --- /dev/null +++ b/sidecar/internal/registry/registry_test.go @@ -0,0 +1,79 @@ +package registry + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +// TestDeleteManifestStatus checks the status-to-error mapping against a fake +// zot, and that the request method and path are what the API expects. +func TestDeleteManifestStatus(t *testing.T) { + cases := []struct { + status int + wantErr bool + }{ + {http.StatusOK, false}, + {http.StatusAccepted, false}, + {http.StatusNoContent, false}, + {http.StatusNotFound, false}, + {http.StatusBadRequest, true}, + {http.StatusInternalServerError, true}, + {http.StatusServiceUnavailable, true}, + } + + for _, tc := range cases { + tc := tc + t.Run(http.StatusText(tc.status), func(t *testing.T) { + var gotMethod, gotPath string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + gotMethod = r.Method + gotPath = r.URL.Path + w.WriteHeader(tc.status) + })) + defer srv.Close() + + c := New(srv.URL) + err := c.DeleteManifest(context.Background(), "foo/bar", "v1") + + if tc.wantErr && err == nil { + t.Fatalf("status %d: expected error, got nil", tc.status) + } + if !tc.wantErr && err != nil { + t.Fatalf("status %d: expected nil, got error: %v", tc.status, err) + } + + if gotMethod != http.MethodDelete { + t.Errorf("method = %q, want %q", gotMethod, http.MethodDelete) + } + if want := "/v2/foo/bar/manifests/v1"; gotPath != want { + t.Errorf("path = %q, want %q", gotPath, want) + } + }) + } +} + +// Port 1 refuses connections, so http.Client.Do fails and the error surfaces. +func TestDeleteManifestTransportError(t *testing.T) { + c := New("http://127.0.0.1:1") + if err := c.DeleteManifest(context.Background(), "foo/bar", "v1"); err == nil { + t.Fatal("expected transport error, got nil") + } +} + +// TestDeleteManifestRequestBuildError uses a base URL containing a control +// character so that http.NewRequest fails before any network I/O. +func TestDeleteManifestRequestBuildError(t *testing.T) { + // Sanity-check that this base URL really makes http.NewRequest fail, so + // the test exercises the request-build error path and not something else. + badBase := "http://\x7f-bad-host" + if _, err := http.NewRequest(http.MethodDelete, badBase+"/v2/foo/bar/manifests/v1", nil); err == nil { + t.Fatalf("precondition failed: expected http.NewRequest to fail for %q", badBase) + } + + c := New(badBase) + if err := c.DeleteManifest(context.Background(), "foo/bar", "v1"); err == nil { + t.Fatal("expected request-build error, got nil") + } +} diff --git a/sidecar/internal/server/server.go b/sidecar/internal/server/server.go new file mode 100644 index 0000000..3baad2b --- /dev/null +++ b/sidecar/internal/server/server.go @@ -0,0 +1,104 @@ +// Package server implements the HTTP surface for the TTL service: a CloudEvents +// sink for zot image events and a health check. +package server + +import ( + "io" + "log" + "net/http" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/events" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/ttl" +) + +// Store is the subset of the persistence layer the server needs. +// Implementations must be safe for concurrent use: the HTTP handlers call it +// from whatever goroutine net/http hands them, without serializing first. +type Store interface { + Upsert(repo, tag, digest string, expiresAt, createdAt time.Time) error +} + +// Server is the HTTP server: events sink + healthz. It holds no lock of its +// own; concurrent event POSTs are safe by way of the Store contract above. +type Server struct { + store Store + defaultTTL time.Duration + maxTTL time.Duration + clock func() time.Time +} + +// New constructs a Server. defaultTTL and maxTTL drive the expiry policy applied +// to incoming events. +func New(st Store, defaultTTL, maxTTL time.Duration, clock func() time.Time) *Server { + return &Server{ + store: st, + defaultTTL: defaultTTL, + maxTTL: maxTTL, + clock: clock, + } +} + +func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.Header().Set("Allow", http.MethodPost) + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20)) + if err != nil { + http.Error(w, "read body: "+err.Error(), http.StatusBadRequest) + return + } + evt, err := events.Parse(r, body) + if err != nil { + log.Printf("events: parse error: %v", err) + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + if evt.Type != events.ImageUpdatedType { + w.WriteHeader(http.StatusNoContent) + return + } + + if evt.Data.Name == "" || evt.Data.Reference == "" { + log.Printf("events: image.updated missing name/reference; ignoring (digest=%q)", evt.Data.Digest) + w.WriteHeader(http.StatusNoContent) + return + } + + now := s.clock() + expires := ttl.ComputeExpiry(evt.Data.Reference, now, s.defaultTTL, s.maxTTL) + + err = s.store.Upsert(evt.Data.Name, evt.Data.Reference, evt.Data.Digest, expires, now) + if err != nil { + log.Printf("events: upsert %s:%s: %v", evt.Data.Name, evt.Data.Reference, err) + http.Error(w, "store error", http.StatusInternalServerError) + return + } + log.Printf("events: recorded %s:%s digest=%s expires_at=%s", + evt.Data.Name, evt.Data.Reference, shortDigest(evt.Data.Digest), expires.UTC().Format(time.RFC3339)) + w.WriteHeader(http.StatusNoContent) +} + +func (s *Server) handleHealthz(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; charset=utf-8") + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, "ok\n") +} + +// Routes returns the HTTP handler wiring the server's endpoints. +func (s *Server) Routes() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/events", s.handleEvents) + mux.HandleFunc("/healthz", s.handleHealthz) + return mux +} + +func shortDigest(d string) string { + if len(d) > 19 { // "sha256:" + 12 hex chars + return d[:19] + } + return d +} diff --git a/sidecar/internal/server/server_test.go b/sidecar/internal/server/server_test.go new file mode 100644 index 0000000..22f70dc --- /dev/null +++ b/sidecar/internal/server/server_test.go @@ -0,0 +1,219 @@ +package server + +import ( + "bytes" + "encoding/json" + "errors" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/events" + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/store" +) + +// Server.Store is documented as safe for concurrent use, so the fake locks its +// own state rather than leaning on the caller to serialize. +type fakeStore struct { + mu sync.Mutex + rows []store.Row + upsertErr error // if set, Upsert returns it +} + +func (f *fakeStore) Upsert(repo, tag, digest string, expiresAt, createdAt time.Time) error { + f.mu.Lock() + defer f.mu.Unlock() + if f.upsertErr != nil { + return f.upsertErr + } + f.rows = append(f.rows, store.Row{ + Repository: repo, + Tag: tag, + ManifestDigest: digest, + ExpiresAt: expiresAt.Unix(), + CreatedAt: createdAt.Unix(), + }) + return nil +} + +const testEpoch = 1_700_000_000 + +func fixedClock() time.Time { return time.Unix(testEpoch, 0) } + +// newTestServer returns a Server backed by an in-memory fake plus the fake +// itself so tests can assert on what was recorded. +func newTestServer() (*Server, *fakeStore) { + fs := &fakeStore{} + return New(fs, 24*time.Hour, 24*time.Hour, fixedClock), fs +} + +func TestHandleEventsRejectsNonPost(t *testing.T) { + srv, _ := newTestServer() + req := httptest.NewRequest(http.MethodGet, "/events", nil) + rec := httptest.NewRecorder() + srv.handleEvents(rec, req) + if rec.Code != http.StatusMethodNotAllowed { + t.Fatalf("got %d want 405", rec.Code) + } + if rec.Header().Get("Allow") != http.MethodPost { + t.Errorf("Allow = %q", rec.Header().Get("Allow")) + } +} + +func TestHandleEventsImageUpdatedUpserts(t *testing.T) { + srv, fs := newTestServer() + body, _ := json.Marshal(events.ImageUpdatedData{ + Name: "foo/bar", Reference: "1h", Digest: "sha256:deadbeef", + }) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", events.ImageUpdatedType) + rec := httptest.NewRecorder() + + srv.handleEvents(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("got %d want 204", rec.Code) + } + if len(fs.rows) != 1 { + t.Fatalf("got %d rows want 1", len(fs.rows)) + } + r := fs.rows[0] + if r.Repository != "foo/bar" || r.Tag != "1h" || r.ManifestDigest != "sha256:deadbeef" { + t.Errorf("row = %+v", r) + } + wantExpires := fixedClock().Add(time.Hour).Unix() + if r.ExpiresAt != wantExpires { + t.Errorf("expires_at = %d want %d", r.ExpiresAt, wantExpires) + } +} + +func TestHandleEventsOtherTypeAcked(t *testing.T) { + srv, fs := newTestServer() + body := []byte(`{"name":"foo","reference":"1h","digest":"sha256:x"}`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", "zotregistry.image.deleted") + rec := httptest.NewRecorder() + + srv.handleEvents(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("got %d want 204", rec.Code) + } + if len(fs.rows) != 0 { + t.Fatalf("non-update event should not store; got %d rows", len(fs.rows)) + } +} + +func TestHandleEventsMissingFieldsAcked(t *testing.T) { + srv, fs := newTestServer() + body := []byte(`{"digest":"sha256:x"}`) // no name, no reference + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", events.ImageUpdatedType) + rec := httptest.NewRecorder() + + srv.handleEvents(rec, req) + if rec.Code != http.StatusNoContent { + t.Fatalf("got %d want 204", rec.Code) + } + if len(fs.rows) != 0 { + t.Fatalf("missing fields should not store; got %d rows", len(fs.rows)) + } +} + +func TestHandleEventsMalformedReturns400(t *testing.T) { + srv, _ := newTestServer() + body := []byte(`{not json`) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", events.ImageUpdatedType) + rec := httptest.NewRecorder() + + srv.handleEvents(rec, req) + if rec.Code != http.StatusBadRequest { + t.Fatalf("got %d want 400", rec.Code) + } +} + +func TestHandleEventsStoreErrorReturns500(t *testing.T) { + fs := &fakeStore{upsertErr: errors.New("boom")} + srv := New(fs, 24*time.Hour, 24*time.Hour, fixedClock) + body, _ := json.Marshal(events.ImageUpdatedData{ + Name: "foo", Reference: "1h", Digest: "sha256:x", + }) + req := httptest.NewRequest(http.MethodPost, "/events", bytes.NewReader(body)) + req.Header.Set("Ce-Type", events.ImageUpdatedType) + rec := httptest.NewRecorder() + + srv.handleEvents(rec, req) + if rec.Code != http.StatusInternalServerError { + t.Fatalf("got %d want 500", rec.Code) + } +} + +func TestHandleHealthz(t *testing.T) { + srv, _ := newTestServer() + req := httptest.NewRequest(http.MethodGet, "/healthz", nil) + rec := httptest.NewRecorder() + srv.handleHealthz(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("got %d want 200", rec.Code) + } + if !strings.Contains(rec.Body.String(), "ok") { + t.Errorf("body = %q", rec.Body.String()) + } +} + +func TestShortDigest(t *testing.T) { + cases := []struct { + in, want string + }{ + {"", ""}, + {"sha256:abc", "sha256:abc"}, // shorter than threshold + {"sha256:0123456789ab", "sha256:0123456789ab"}, // exactly at threshold (len 19) + {"sha256:0123456789abcdef", "sha256:0123456789ab"}, // longer; truncated + } + for _, tc := range cases { + t.Run(tc.in, func(t *testing.T) { + got := shortDigest(tc.in) + if got != tc.want { + t.Errorf("got %q want %q", got, tc.want) + } + }) + } +} + +// TestRoutesWiring exercises New + Routes over real HTTP round-trips. +func TestRoutesWiring(t *testing.T) { + srv := New(&fakeStore{}, 24*time.Hour, 24*time.Hour, fixedClock) + ts := httptest.NewServer(srv.Routes()) + defer ts.Close() + + resp, err := http.Get(ts.URL + "/healthz") + if err != nil { + t.Fatalf("GET /healthz: %v", err) + } + if resp.StatusCode != http.StatusOK { + t.Errorf("GET /healthz = %d want 200", resp.StatusCode) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() + + body, _ := json.Marshal(events.ImageUpdatedData{ + Name: "foo/bar", Reference: "1h", Digest: "sha256:deadbeef", + }) + req, err := http.NewRequest(http.MethodPost, ts.URL+"/events", bytes.NewReader(body)) + if err != nil { + t.Fatalf("new request: %v", err) + } + req.Header.Set("Ce-Type", events.ImageUpdatedType) + resp, err = http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("POST /events: %v", err) + } + if resp.StatusCode != http.StatusNoContent { + t.Errorf("POST /events = %d want 204", resp.StatusCode) + } + _, _ = io.Copy(io.Discard, resp.Body) + _ = resp.Body.Close() +} diff --git a/sidecar/internal/store/store.go b/sidecar/internal/store/store.go new file mode 100644 index 0000000..7e8470d --- /dev/null +++ b/sidecar/internal/store/store.go @@ -0,0 +1,184 @@ +// Package store is the Redis persistence layer: it records +// (repository, tag, manifest_digest, expires_at) rows and answers which +// of them have expired. +// +// State lives in exactly two keys: +// +// zot-ephemeral-ttl:rows HASH field = \x00, value = Row as JSON +// zot-ephemeral-ttl:index ZSET member = \x00, score = expires_at +// +// The sorted set is the expiry index: "which tags are due" is a range query by +// score against it. The NUL separator keeps (repository, tag) pairs unambiguous +// — no repository or tag can contain one — and makes the sorted set's +// lexicographic tie-break for equal scores fall out as (repository, tag) order. +// +// Rows deliberately do NOT carry a Redis key TTL. The reaper has to *see* an +// expired row in order to issue the manifest delete to zot, and it removes the +// row only once zot confirms; letting Redis expire the data itself would drop +// tags on the floor without ever deleting their manifests. +package store + +import ( + "context" + "encoding/json" + "fmt" + "strconv" + "time" + + "github.com/redis/go-redis/v9" +) + +// The two keys this service owns. The Redis instance is dedicated to the +// sidecar, so they are fixed rather than configurable. +const ( + rowsKey = "zot-ephemeral-ttl:rows" + indexKey = "zot-ephemeral-ttl:index" +) + +// opTimeout bounds every individual Redis round trip. The consumer interfaces +// take no context, so each operation carries its own deadline; a wedged Redis +// therefore surfaces as an error rather than a stuck sweep or event handler. +const opTimeout = 5 * time.Second + +// memberSep joins repository and tag into a single sorted-set member / hash +// field. NUL cannot appear in either, so the encoding is injective. +const memberSep = "\x00" + +// Row is a single tracked tag and its expiry. +type Row struct { + Repository string `json:"repository"` + Tag string `json:"tag"` + ManifestDigest string `json:"manifest_digest"` + ExpiresAt int64 `json:"expires_at"` + CreatedAt int64 `json:"created_at"` +} + +type Store struct { + rdb *redis.Client +} + +// Open dials the Redis server named by rawURL (a redis:// or rediss:// URL as +// understood by redis.ParseURL) and verifies the connection with a PING, so a +// bad address or credential fails at startup rather than on the first push. +func Open(rawURL string) (*Store, error) { + opt, err := redis.ParseURL(rawURL) + if err != nil { + // Deliberately not echoing rawURL: it may carry a password. + return nil, fmt.Errorf("parse redis url: %w", err) + } + rdb := redis.NewClient(opt) + + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + if err := rdb.Ping(ctx).Err(); err != nil { + _ = rdb.Close() + return nil, fmt.Errorf("ping redis: %w", err) + } + + return &Store{rdb: rdb}, nil +} + +func (s *Store) Close() error { return s.rdb.Close() } + +func (s *Store) opCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), opTimeout) +} + +func member(repo, tag string) string { return repo + memberSep + tag } + +// Upsert inserts or refreshes the row for (repo, tag). Timestamps are truncated +// to whole seconds. The row payload and its index entry are written in one +// transaction so a reader never sees an index entry without its row. +func (s *Store) Upsert(repo, tag, digest string, expiresAt, createdAt time.Time) error { + row := Row{ + Repository: repo, + Tag: tag, + ManifestDigest: digest, + ExpiresAt: expiresAt.Unix(), + CreatedAt: createdAt.Unix(), + } + payload, err := json.Marshal(row) + if err != nil { + return err + } + + ctx, cancel := s.opCtx() + defer cancel() + m := member(repo, tag) + _, err = s.rdb.TxPipelined(ctx, func(p redis.Pipeliner) error { + p.HSet(ctx, rowsKey, m, payload) + p.ZAdd(ctx, indexKey, redis.Z{Score: float64(row.ExpiresAt), Member: m}) + return nil + }) + return err +} + +// Expired returns every row whose expiry is at or before now, ordered by expiry +// ascending with ties broken by (repository, tag) — the sorted set orders equal +// scores lexicographically by member, which is exactly that pair. +func (s *Store) Expired(now time.Time) ([]Row, error) { + ctx, cancel := s.opCtx() + defer cancel() + + members, err := s.rdb.ZRangeArgs(ctx, redis.ZRangeArgs{ + Key: indexKey, + Start: "-inf", + Stop: strconv.FormatInt(now.Unix(), 10), // inclusive: expires_at <= now + ByScore: true, + }).Result() + if err != nil { + return nil, err + } + return s.rows(ctx, members) +} + +// Delete removes the row for (repo, tag); a missing key is a no-op. +func (s *Store) Delete(repo, tag string) error { + ctx, cancel := s.opCtx() + defer cancel() + + m := member(repo, tag) + _, err := s.rdb.TxPipelined(ctx, func(p redis.Pipeliner) error { + p.HDel(ctx, rowsKey, m) + p.ZRem(ctx, indexKey, m) + return nil + }) + return err +} + +// rows fetches the row payloads for members, preserving their order. An index +// entry whose payload is missing is dropped from the index (best effort) and +// skipped: the pair is only ever written in one transaction, so a gap means +// something outside this service removed the hash field. +func (s *Store) rows(ctx context.Context, members []string) ([]Row, error) { + if len(members) == 0 { + return nil, nil + } + vals, err := s.rdb.HMGet(ctx, rowsKey, members...).Result() + if err != nil { + return nil, err + } + + var out []Row + var orphans []string + for i, v := range vals { + payload, ok := v.(string) + if !ok { // nil: no such hash field + orphans = append(orphans, members[i]) + continue + } + var r Row + if err := json.Unmarshal([]byte(payload), &r); err != nil { + return nil, fmt.Errorf("decode row %q: %w", members[i], err) + } + out = append(out, r) + } + if len(orphans) > 0 { + args := make([]any, len(orphans)) + for i, m := range orphans { + args[i] = m + } + _ = s.rdb.ZRem(ctx, indexKey, args...).Err() + } + return out, nil +} diff --git a/sidecar/internal/store/store_test.go b/sidecar/internal/store/store_test.go new file mode 100644 index 0000000..df21254 --- /dev/null +++ b/sidecar/internal/store/store_test.go @@ -0,0 +1,467 @@ +package store + +import ( + "context" + "os" + "sync" + "testing" + "time" + + "github.com/redis/go-redis/v9" + + "github.com/nullbytelabs/zot-ephemeral-ttl/internal/redistest" +) + +// These tests run against a real Redis, started for the package by redistest. +func TestMain(m *testing.M) { os.Exit(redistest.Run(m)) } + +// farFuture is later than any expiry these tests set, so Expired(farFuture) +// returns the whole keyspace — the read-back used to inspect stored state. +var farFuture = time.Unix(1<<40, 0) + +// newTestStore opens a store against that server. The keys are fixed and the +// whole package shares one container, so each test starts from a clean slate +// and drops them again on the way out. +func newTestStore(t *testing.T) *Store { + t.Helper() + st, err := Open(redistest.URL(t)) + if err != nil { + t.Fatalf("Open: %v", err) + } + reset := func() { + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + if err := st.rdb.Del(ctx, rowsKey, indexKey).Err(); err != nil { + t.Errorf("clearing test keys: %v", err) + } + } + reset() + t.Cleanup(func() { + reset() + if err := st.Close(); err != nil { + t.Errorf("cleanup close: %v", err) + } + }) + return st +} + +func TestOpenRejectsBadURL(t *testing.T) { + // Not a redis:// URL: ParseURL fails before any dial, so this needs no + // server. + if _, err := Open("http://127.0.0.1:6379"); err == nil { + t.Fatal("Open with a non-redis scheme: expected error, got nil") + } +} + +func TestOpenPingFails(t *testing.T) { + // Port 1 refuses connections, so the startup PING fails and Open reports + // it rather than returning a store that only breaks on first use. + if _, err := Open("redis://127.0.0.1:1"); err == nil { + t.Fatal("Open against a dead address: expected error, got nil") + } +} + +func TestOpenEmptyStore(t *testing.T) { + st := newTestStore(t) + rows, err := st.Expired(farFuture) + if err != nil { + t.Fatalf("Expired on empty store: %v", err) + } + if len(rows) != 0 { + t.Fatalf("expected empty store, got %d rows", len(rows)) + } +} + +// TestKeyLayout pins the documented storage layout: state lives in exactly two +// keys, with (repository, tag) encoded as the hash field. +func TestKeyLayout(t *testing.T) { + st := newTestStore(t) + if rowsKey != "zot-ephemeral-ttl:rows" || indexKey != "zot-ephemeral-ttl:index" { + t.Errorf("keys = %q/%q, want zot-ephemeral-ttl:rows / :index", rowsKey, indexKey) + } + + now := time.Now() + if err := st.Upsert("repo/a", "v1", "sha256:aaa", now.Add(time.Hour), now); err != nil { + t.Fatalf("Upsert: %v", err) + } + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + keys, err := st.rdb.Keys(ctx, "zot-ephemeral-ttl*").Result() + if err != nil { + t.Fatalf("Keys: %v", err) + } + if len(keys) != 2 { + t.Errorf("got keys %v, want exactly the rows hash and the index zset", keys) + } + + // Field/member encoding is NUL. + fields, err := st.rdb.HKeys(ctx, rowsKey).Result() + if err != nil { + t.Fatalf("HKeys: %v", err) + } + if len(fields) != 1 || fields[0] != "repo/a\x00v1" { + t.Errorf("hash fields = %q, want [\"repo/a\\x00v1\"]", fields) + } +} + +func TestUpsertInsertThenUpdate(t *testing.T) { + st := newTestStore(t) + now := time.Now() + + if err := st.Upsert("repo/a", "v1", "sha256:aaa", now.Add(time.Hour), now); err != nil { + t.Fatalf("first upsert: %v", err) + } + rows, err := st.Expired(farFuture) + if err != nil { + t.Fatalf("Expired: %v", err) + } + if len(rows) != 1 { + t.Fatalf("got %d rows want 1", len(rows)) + } + if rows[0].ManifestDigest != "sha256:aaa" { + t.Fatalf("digest = %q", rows[0].ManifestDigest) + } + + // Same (repo, tag): should update digest, expires_at, created_at in place. + later := now.Add(2 * time.Hour) + if err := st.Upsert("repo/a", "v1", "sha256:bbb", later.Add(time.Hour), later); err != nil { + t.Fatalf("second upsert: %v", err) + } + rows, err = st.Expired(farFuture) + if err != nil { + t.Fatalf("Expired: %v", err) + } + if len(rows) != 1 { + t.Fatalf("upsert duplicated row, got %d", len(rows)) + } + if rows[0].ManifestDigest != "sha256:bbb" { + t.Fatalf("digest not updated: %q", rows[0].ManifestDigest) + } + if rows[0].CreatedAt != later.Unix() { + t.Fatalf("created_at not updated: got %d want %d", rows[0].CreatedAt, later.Unix()) + } + if rows[0].ExpiresAt != later.Add(time.Hour).Unix() { + t.Fatalf("expires_at not updated: got %d want %d", rows[0].ExpiresAt, later.Add(time.Hour).Unix()) + } +} + +// TestUpsertRefreshesIndexScore proves the sorted-set score moves with the +// row: a re-push must actually postpone reaping, not just rewrite the payload. +func TestUpsertRefreshesIndexScore(t *testing.T) { + st := newTestStore(t) + now := time.Unix(1_700_000_000, 0) + + if err := st.Upsert("r", "t", "sha256:x", now.Add(-time.Hour), now.Add(-2*time.Hour)); err != nil { + t.Fatalf("upsert: %v", err) + } + expired, err := st.Expired(now) + if err != nil { + t.Fatalf("Expired: %v", err) + } + if len(expired) != 1 { + t.Fatalf("row should be expired before refresh, got %d rows", len(expired)) + } + + if err := st.Upsert("r", "t", "sha256:x", now.Add(time.Hour), now); err != nil { + t.Fatalf("refresh upsert: %v", err) + } + expired, err = st.Expired(now) + if err != nil { + t.Fatalf("Expired after refresh: %v", err) + } + if len(expired) != 0 { + t.Fatalf("row should no longer be expired after refresh, got %d rows", len(expired)) + } +} + +func TestExpiredFiltersByTime(t *testing.T) { + st := newTestStore(t) + now := time.Unix(1_700_000_000, 0) + + // Past, exactly-now, and future rows. + must := func(err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } + } + must(st.Upsert("r", "past", "sha256:1", now.Add(-time.Hour), now.Add(-2*time.Hour))) + must(st.Upsert("r", "now", "sha256:2", now, now.Add(-time.Minute))) + must(st.Upsert("r", "future", "sha256:3", now.Add(time.Hour), now)) + + expired, err := st.Expired(now) + if err != nil { + t.Fatalf("Expired: %v", err) + } + tags := map[string]bool{} + for _, r := range expired { + tags[r.Tag] = true + } + if !tags["past"] { + t.Errorf("past row should be expired") + } + if !tags["now"] { + t.Errorf("now row should be expired (expires_at <= now)") + } + if tags["future"] { + t.Errorf("future row should not be expired") + } +} + +// TestExpiredReturnsFullRow checks the payload survives the round trip: the +// reaper needs repository and tag to build the manifest DELETE. +func TestExpiredReturnsFullRow(t *testing.T) { + st := newTestStore(t) + now := time.Unix(1_700_000_000, 0) + if err := st.Upsert("team/app", "30m", "sha256:abc", now.Add(-time.Second), now.Add(-time.Hour)); err != nil { + t.Fatal(err) + } + rows, err := st.Expired(now) + if err != nil { + t.Fatalf("Expired: %v", err) + } + if len(rows) != 1 { + t.Fatalf("got %d rows want 1", len(rows)) + } + want := Row{ + Repository: "team/app", + Tag: "30m", + ManifestDigest: "sha256:abc", + ExpiresAt: now.Add(-time.Second).Unix(), + CreatedAt: now.Add(-time.Hour).Unix(), + } + if rows[0] != want { + t.Errorf("got %+v want %+v", rows[0], want) + } +} + +func TestDelete(t *testing.T) { + st := newTestStore(t) + now := time.Now() + if err := st.Upsert("r", "t", "sha256:x", now.Add(time.Hour), now); err != nil { + t.Fatal(err) + } + if err := st.Delete("r", "t"); err != nil { + t.Fatalf("Delete: %v", err) + } + rows, err := st.Expired(farFuture) + if err != nil { + t.Fatal(err) + } + if len(rows) != 0 { + t.Fatalf("expected empty after delete, got %d", len(rows)) + } + + // Both keys must be cleaned up, not just the payload. + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + n, err := st.rdb.ZCard(ctx, indexKey).Result() + if err != nil { + t.Fatalf("ZCard: %v", err) + } + if n != 0 { + t.Errorf("index still holds %d member(s) after delete", n) + } + + // Deleting a non-existent row is a no-op (no error). + if err := st.Delete("r", "missing"); err != nil { + t.Fatalf("Delete missing: %v", err) + } +} + +// TestDeleteIsScopedToOneTag guards the (repository, tag) encoding: deleting +// one tag must not disturb its neighbours. +func TestDeleteIsScopedToOneTag(t *testing.T) { + st := newTestStore(t) + now := time.Now() + must := func(err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } + } + must(st.Upsert("r", "keep", "sha256:1", now.Add(time.Hour), now)) + must(st.Upsert("r", "drop", "sha256:2", now.Add(time.Hour), now)) + must(st.Upsert("r/nested", "keep", "sha256:3", now.Add(time.Hour), now)) + + must(st.Delete("r", "drop")) + + rows, err := st.Expired(farFuture) + if err != nil { + t.Fatal(err) + } + if len(rows) != 2 { + t.Fatalf("got %d rows after delete, want 2", len(rows)) + } + for _, r := range rows { + if r.Tag == "drop" { + t.Errorf("deleted row still present: %+v", r) + } + } +} + +func TestOrdering(t *testing.T) { + st := newTestStore(t) + now := time.Now() + // Insert in non-sorted order. + rows := []struct { + repo, tag string + expires time.Time + }{ + {"r", "c", now.Add(3 * time.Hour)}, + {"r", "a", now.Add(1 * time.Hour)}, + {"r", "b", now.Add(2 * time.Hour)}, + } + for _, r := range rows { + if err := st.Upsert(r.repo, r.tag, "sha256:x", r.expires, now); err != nil { + t.Fatal(err) + } + } + got, err := st.Expired(farFuture) + if err != nil { + t.Fatal(err) + } + if len(got) != 3 { + t.Fatalf("got %d rows, want 3", len(got)) + } + wantOrder := []string{"a", "b", "c"} + for i, w := range wantOrder { + if got[i].Tag != w { + t.Errorf("position %d: got %q want %q", i, got[i].Tag, w) + } + } +} + +// TestOrderingTieBreak pins the (repository, tag) tie-break for rows that +// share an expiry — it comes from the sorted set's lexicographic ordering of +// equal-score members, which the NUL separator makes match pair ordering. +func TestOrderingTieBreak(t *testing.T) { + st := newTestStore(t) + exp := time.Unix(1_700_000_000, 0) + now := time.Now() + must := func(err error) { + t.Helper() + if err != nil { + t.Fatal(err) + } + } + must(st.Upsert("repo/b", "x", "sha256:1", exp, now)) + must(st.Upsert("repo/a", "y", "sha256:2", exp, now)) + must(st.Upsert("repo/a", "x", "sha256:3", exp, now)) + + got, err := st.Expired(farFuture) + if err != nil { + t.Fatal(err) + } + want := []struct{ repo, tag string }{ + {"repo/a", "x"}, + {"repo/a", "y"}, + {"repo/b", "x"}, + } + if len(got) != len(want) { + t.Fatalf("got %d rows, want %d", len(got), len(want)) + } + for i, w := range want { + if got[i].Repository != w.repo || got[i].Tag != w.tag { + t.Errorf("position %d: got %s:%s want %s:%s", i, got[i].Repository, got[i].Tag, w.repo, w.tag) + } + } +} + +// TestOrphanedIndexEntryIsHealed covers an index member whose payload has gone +// missing (something outside this service touched the hash): it is skipped and +// dropped from the index rather than surfacing as a zero-valued row. +func TestOrphanedIndexEntryIsHealed(t *testing.T) { + st := newTestStore(t) + now := time.Now() + if err := st.Upsert("r", "t", "sha256:x", now.Add(time.Hour), now); err != nil { + t.Fatal(err) + } + + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + if err := st.rdb.HDel(ctx, rowsKey, member("r", "t")).Err(); err != nil { + t.Fatalf("HDel: %v", err) + } + + rows, err := st.Expired(farFuture) + if err != nil { + t.Fatalf("Expired with an orphaned index entry: %v", err) + } + if len(rows) != 0 { + t.Fatalf("got %d rows, want the orphan skipped", len(rows)) + } + n, err := st.rdb.ZCard(ctx, indexKey).Result() + if err != nil { + t.Fatalf("ZCard: %v", err) + } + if n != 0 { + t.Errorf("orphaned index entry not cleaned up: %d member(s) remain", n) + } +} + +// TestCorruptPayloadErrors: an undecodable payload is real corruption, so it +// surfaces as an error instead of being silently dropped. +func TestCorruptPayloadErrors(t *testing.T) { + st := newTestStore(t) + ctx, cancel := context.WithTimeout(context.Background(), opTimeout) + defer cancel() + + m := member("r", "t") + if err := st.rdb.HSet(ctx, rowsKey, m, "not json").Err(); err != nil { + t.Fatalf("HSet: %v", err) + } + if err := st.rdb.ZAdd(ctx, indexKey, redis.Z{Score: 1, Member: m}).Err(); err != nil { + t.Fatalf("ZAdd: %v", err) + } + + if _, err := st.Expired(farFuture); err == nil { + t.Error("Expired with a corrupt payload: expected error, got nil") + } +} + +func TestOperationsOnClosedStore(t *testing.T) { + st, err := Open(redistest.URL(t)) + if err != nil { + t.Fatalf("Open: %v", err) + } + if err := st.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + + now := time.Now() + if err := st.Upsert("r", "t", "sha256:x", now.Add(time.Hour), now); err == nil { + t.Error("Upsert on closed store: expected error, got nil") + } + if err := st.Delete("r", "t"); err == nil { + t.Error("Delete on closed store: expected error, got nil") + } + if _, err := st.Expired(now); err == nil { + t.Error("Expired on closed store: expected error, got nil") + } +} + +// TestConcurrentAccess runs concurrent writers and readers under the race +// detector: the server's event handler and the reaper's sweep share one store. +func TestConcurrentAccess(t *testing.T) { + st := newTestStore(t) + now := time.Now() + var wg sync.WaitGroup + for i := 0; i < 25; i++ { + wg.Add(2) + go func(n int) { + defer wg.Done() + tag := string(rune('a' + n%26)) + if err := st.Upsert("r", tag, "sha256:x", now.Add(time.Hour), now); err != nil { + t.Errorf("concurrent upsert: %v", err) + } + }(i) + go func() { + defer wg.Done() + if _, err := st.Expired(farFuture); err != nil { + t.Errorf("concurrent Expired: %v", err) + } + }() + } + wg.Wait() +} diff --git a/sidecar/internal/ttl/ttl.go b/sidecar/internal/ttl/ttl.go new file mode 100644 index 0000000..afcfa2d --- /dev/null +++ b/sidecar/internal/ttl/ttl.go @@ -0,0 +1,62 @@ +// Package ttl parses TTL-encoding tag references (e.g. "1h", "30m", "7d") and +// applies the expiry policy: the parsed TTL, or a default when the tag doesn't +// match, clamped to a maximum. +package ttl + +import ( + "math" + "regexp" + "strconv" + "time" +) + +var tagRe = regexp.MustCompile(`^([0-9]+)(s|m|h|d|w)$`) + +// ParseTag returns the TTL encoded in reference and whether it matched the +// `^\d+(s|m|h|d|w)$` form. +func ParseTag(reference string) (time.Duration, bool) { + m := tagRe.FindStringSubmatch(reference) + if m == nil { + return 0, false + } + n, err := strconv.ParseInt(m[1], 10, 64) + if err != nil || n <= 0 { + return 0, false + } + var unit time.Duration + switch m[2] { + case "s": + unit = time.Second + case "m": + unit = time.Minute + case "h": + unit = time.Hour + case "d": + unit = 24 * time.Hour + case "w": + unit = 7 * 24 * time.Hour + default: + return 0, false + } + // A tag can name a period far longer than time.Duration can hold. Saturate + // instead of letting the multiply wrap: a negative duration would slip past + // the clamp in ComputeExpiry and expire the tag immediately, which is the + // opposite of what a huge TTL asks for. + if n > int64(math.MaxInt64)/int64(unit) { + return time.Duration(math.MaxInt64), true + } + return time.Duration(n) * unit, true +} + +// ComputeExpiry applies the policy: the parsed TTL (or defaultTTL if reference +// doesn't encode one), clamped to maxTTL. +func ComputeExpiry(reference string, now time.Time, defaultTTL, maxTTL time.Duration) time.Time { + ttl := defaultTTL + if parsed, ok := ParseTag(reference); ok { + ttl = parsed + } + if ttl > maxTTL { + ttl = maxTTL + } + return now.Add(ttl) +} diff --git a/sidecar/internal/ttl/ttl_test.go b/sidecar/internal/ttl/ttl_test.go new file mode 100644 index 0000000..6cb26bd --- /dev/null +++ b/sidecar/internal/ttl/ttl_test.go @@ -0,0 +1,92 @@ +package ttl + +import ( + "math" + "testing" + "time" +) + +func TestParseTag(t *testing.T) { + cases := []struct { + name string + reference string + want time.Duration + wantOK bool + }{ + {"empty", "", 0, false}, + {"latest", "latest", 0, false}, + {"semver", "v1.2.3", 0, false}, + {"unknown unit", "10y", 0, false}, + {"missing unit", "10", 0, false}, + {"missing number", "h", 0, false}, + {"zero seconds", "0s", 0, false}, + {"negative not allowed by regex", "-5m", 0, false}, + {"trailing junk", "10ss", 0, false}, + {"leading junk", "x10s", 0, false}, + {"upper case unit", "10H", 0, false}, + {"seconds", "30s", 30 * time.Second, true}, + {"minutes", "30m", 30 * time.Minute, true}, + {"hours", "2h", 2 * time.Hour, true}, + {"days", "7d", 7 * 24 * time.Hour, true}, + {"weeks", "2w", 2 * 7 * 24 * time.Hour, true}, + {"single second", "1s", time.Second, true}, + {"leading zeros", "007h", 7 * time.Hour, true}, + {"large", "9999h", 9999 * time.Hour, true}, + {"int64 overflow", "99999999999999999999s", 0, false}, + // 20000w exceeds what time.Duration can represent. Saturating keeps the + // value positive so ComputeExpiry clamps it; wrapping would make it + // negative and expire the tag on the spot. + {"duration overflow saturates", "20000w", time.Duration(math.MaxInt64), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got, ok := ParseTag(tc.reference) + if ok != tc.wantOK { + t.Fatalf("ok=%v want=%v", ok, tc.wantOK) + } + if got != tc.want { + t.Fatalf("got=%v want=%v", got, tc.want) + } + }) + } +} + +func TestComputeExpiry(t *testing.T) { + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + defaultTTL := 24 * time.Hour + maxTTL := 24 * time.Hour + + cases := []struct { + name string + reference string + want time.Duration // expected delta from now + }{ + {"unknown reference uses default", "latest", defaultTTL}, + {"empty reference uses default", "", defaultTTL}, + {"parsed under max", "1h", time.Hour}, + {"parsed equal to max", "24h", 24 * time.Hour}, + {"parsed over max gets clamped", "7d", maxTTL}, + {"parsed weeks clamped", "2w", maxTTL}, + {"overflowing tag clamps to max", "20000w", maxTTL}, + {"parsed seconds untouched", "30s", 30 * time.Second}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := ComputeExpiry(tc.reference, now, defaultTTL, maxTTL) + want := now.Add(tc.want) + if !got.Equal(want) { + t.Fatalf("got=%v want=%v", got, want) + } + }) + } +} + +func TestComputeExpiryNonDefaultMax(t *testing.T) { + now := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + // Default larger than max: default itself should be clamped. + got := ComputeExpiry("latest", now, 48*time.Hour, 12*time.Hour) + want := now.Add(12 * time.Hour) + if !got.Equal(want) { + t.Fatalf("default-clamping: got=%v want=%v", got, want) + } +}