diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8171178..0db2736 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -11,12 +11,62 @@ jobs: defaults: run: working-directory: ingestion-service + # This service talks to Postgres and Redis, so its tests need both to be + # real. Without them the db and cache suites skip themselves and the job + # still goes green while testing almost nothing. + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: pipelineops + POSTGRES_PASSWORD: pipelineops + POSTGRES_DB: pipelineops + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 5s + --health-timeout 5s + --health-retries 5 + redis: + image: redis:7 + ports: + - 6379:6379 + options: >- + --health-cmd "redis-cli ping" + --health-interval 5s + --health-timeout 5s + --health-retries 5 + env: + INGESTION_TEST_DATABASE_URL: postgres://pipelineops:pipelineops@localhost:5432/pipelineops + INGESTION_TEST_REDIS_URL: redis://localhost:6379/0 steps: - uses: actions/checkout@v4 - uses: actions/setup-go@v5 with: go-version: "1.24" cache-dependency-path: ingestion-service/go.sum + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + cache-dependency-path: core-api/requirements.txt + + # The ingestion service reads and writes tables Django owns the + # migrations for, and it deliberately never runs migrations itself. So + # the schema its tests need has to come from the same place production's + # does — running Django's migrations — rather than from a copy of the + # DDL kept in this service, which would drift the first time a model + # changed. + - name: apply the schema Django owns + working-directory: core-api + env: + DATABASE_URL: postgres://pipelineops:pipelineops@localhost:5432/pipelineops + DJANGO_SECRET_KEY: ci-test-key + run: | + pip install -r requirements.txt + python manage.py migrate --noinput + - name: golangci-lint uses: golangci/golangci-lint-action@v7 with: @@ -24,6 +74,17 @@ jobs: working-directory: ingestion-service - run: go build ./... - run: go test ./... -v -coverprofile=coverage.out -covermode=atomic + + # The suites skip themselves when Postgres or Redis is unreachable, + # which is the right behaviour locally and a silent failure here. If the + # containers above ever stop being wired up correctly, this fails the + # job instead of quietly reporting a much smaller test run. + - name: no test skipped for a missing dependency + run: | + if go test ./... -v 2>&1 | grep -E "skipping: no reachable (postgres|redis)"; then + echo "::error::integration tests skipped — the postgres/redis service containers are not reachable" + exit 1 + fi - name: Upload coverage uses: codecov/codecov-action@v5 with: diff --git a/codecov.yml b/codecov.yml index b2c7108..89cb8dc 100644 --- a/codecov.yml +++ b/codecov.yml @@ -22,3 +22,12 @@ flags: frontend: paths: - frontend/src/ + +ignore: + # main() is now nothing but wiring: read the environment, dial Postgres and + # Redis, serve, wait for a signal, shut down. Everything it used to decide + # moved to cmd/server/bootstrap.go, which is tested directly — the retry + # loops, the request logger, the env fallbacks and the route table. What is + # left here could only be covered by starting the process and killing it, + # which would measure the test harness rather than the code. + - "ingestion-service/cmd/server/main.go" diff --git a/ingestion-service/cmd/server/bootstrap.go b/ingestion-service/cmd/server/bootstrap.go new file mode 100644 index 0000000..f011cec --- /dev/null +++ b/ingestion-service/cmd/server/bootstrap.go @@ -0,0 +1,90 @@ +package main + +import ( + "log/slog" + "os" + "time" + + "github.com/gin-gonic/gin" + "github.com/prometheus/client_golang/prometheus/promhttp" + + "pipelineops/ingestion-service/internal/cache" + "pipelineops/ingestion-service/internal/db" + "pipelineops/ingestion-service/internal/handlers" +) + +// Everything main() decides lives here rather than inside main() itself, so it +// can be tested. What's left in main.go is the part that genuinely can't be: +// binding a port, waiting on a signal, and exiting. + +func getenv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} + +// newRouter builds the full route table. It's separated from main so the +// wiring is covered by a test — a route registered at the wrong path or bound +// to the wrong handler is a real bug, and one that a compiler can't catch and +// a running service only reveals in production. +func newRouter(deps handlers.Deps, logger *slog.Logger) *gin.Engine { + router := gin.New() + router.Use(gin.Recovery(), slogRequestLogger(logger)) + + router.GET("/healthz", handlers.Healthz(deps)) + router.GET("/metrics", gin.WrapH(promhttp.Handler())) + + v1 := router.Group("/v1") + { + v1.POST("/heartbeat", deps.PostHeartbeat) + v1.GET("/jobs/:job/heartbeat/latest", deps.GetLatestHeartbeat) + } + return router +} + +// connectWithRetry calls connect up to 15 times, sleeping 2 seconds between +// attempts, until it succeeds or the attempts are exhausted. connect and +// sleep are injected so tests can substitute a fake connector and a +// non-blocking sleep; main passes the real db.Connect and time.Sleep. +func connectWithRetry(logger *slog.Logger, connect func() (*db.Pool, error), sleep func(time.Duration)) (*db.Pool, error) { + var lastErr error + for i := 0; i < 15; i++ { + pool, err := connect() + if err == nil { + return pool, nil + } + lastErr = err + logger.Warn("postgres not ready, retrying", "attempt", i+1, "error", err) + sleep(2 * time.Second) + } + return nil, lastErr +} + +// connectRedisWithRetry mirrors connectWithRetry for the Redis client. +func connectRedisWithRetry(logger *slog.Logger, connect func() (*cache.Client, error), sleep func(time.Duration)) (*cache.Client, error) { + var lastErr error + for i := 0; i < 15; i++ { + client, err := connect() + if err == nil { + return client, nil + } + lastErr = err + logger.Warn("redis not ready, retrying", "attempt", i+1, "error", err) + sleep(2 * time.Second) + } + return nil, lastErr +} + +func slogRequestLogger(logger *slog.Logger) gin.HandlerFunc { + return func(c *gin.Context) { + start := time.Now() + c.Next() + logger.Info("request", + "method", c.Request.Method, + "path", c.Request.URL.Path, + "status", c.Writer.Status(), + "duration_ms", time.Since(start).Milliseconds(), + ) + } +} diff --git a/ingestion-service/cmd/server/main_test.go b/ingestion-service/cmd/server/bootstrap_test.go similarity index 56% rename from ingestion-service/cmd/server/main_test.go rename to ingestion-service/cmd/server/bootstrap_test.go index f48d58f..137f1de 100644 --- a/ingestion-service/cmd/server/main_test.go +++ b/ingestion-service/cmd/server/bootstrap_test.go @@ -2,6 +2,7 @@ package main import ( "bytes" + "context" "encoding/json" "errors" "log/slog" @@ -13,9 +14,11 @@ import ( "time" "github.com/gin-gonic/gin" + "github.com/google/uuid" "pipelineops/ingestion-service/internal/cache" "pipelineops/ingestion-service/internal/db" + "pipelineops/ingestion-service/internal/handlers" ) func init() { @@ -270,3 +273,158 @@ func TestConnectRedisWithRetry_ExhaustsRetriesAndReturnsLastError(t *testing.T) t.Fatalf("sleep called %d times, want 15", sleeps) } } + +// --- route wiring ----------------------------------------------------------- +// +// newRouter is the one part of startup where a mistake is invisible until +// production: a route registered at the wrong path, or bound to the wrong +// handler, compiles perfectly and fails only when a real client calls it. +// These tests assert the route table and then prove each route reaches the +// handler it claims to, by checking behaviour only that handler produces. + +type stubDB struct { + job *db.Job + findErr error + pingErr error +} + +func (s stubDB) FindJobByNameOrID(_ context.Context, _ string) (*db.Job, error) { + return s.job, s.findErr +} + +func (s stubDB) InsertHeartbeat(_ context.Context, _ db.InsertHeartbeatParams) (int64, time.Time, error) { + return 1, time.Unix(0, 0).UTC(), nil +} + +func (s stubDB) Ping(_ context.Context) error { return s.pingErr } + +type stubCache struct { + hb *cache.LastHeartbeat + pingErr error +} + +func (s stubCache) SetLastHeartbeat(_ context.Context, _ cache.LastHeartbeat) error { return nil } + +func (s stubCache) GetLastHeartbeat(_ context.Context, _ string) (*cache.LastHeartbeat, error) { + return s.hb, nil +} + +func (s stubCache) Ping(_ context.Context) error { return s.pingErr } + +func testDeps(d stubDB, c stubCache) handlers.Deps { + return handlers.Deps{DB: d, Cache: c, Logger: discardLogger()} +} + +func TestNewRouter_RegistersEveryRoute(t *testing.T) { + router := newRouter(testDeps(stubDB{}, stubCache{}), discardLogger()) + + registered := map[string]bool{} + for _, r := range router.Routes() { + registered[r.Method+" "+r.Path] = true + } + + for _, want := range []string{ + "GET /healthz", + "GET /metrics", + "POST /v1/heartbeat", + "GET /v1/jobs/:job/heartbeat/latest", + } { + if !registered[want] { + t.Errorf("route %q is not registered; got %v", want, registered) + } + } + if len(router.Routes()) != 4 { + t.Errorf("router has %d routes, want exactly 4 — an unintended route is as much a bug as a missing one", len(router.Routes())) + } +} + +func TestNewRouter_HealthzReachesTheHealthHandler(t *testing.T) { + t.Run("healthy when both dependencies answer", func(t *testing.T) { + router := newRouter(testDeps(stubDB{}, stubCache{}), discardLogger()) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + if !strings.Contains(rec.Body.String(), `"status":"ok"`) { + t.Fatalf("body = %s", rec.Body.String()) + } + }) + + // A 503 here can only come from Healthz itself, which is what makes this + // a test of the binding rather than of the path. + t.Run("degraded when postgres is down", func(t *testing.T) { + router := newRouter(testDeps(stubDB{pingErr: errors.New("down")}, stubCache{}), discardLogger()) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/healthz", nil)) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusServiceUnavailable) + } + if !strings.Contains(rec.Body.String(), `"db":"down"`) { + t.Fatalf("body = %s", rec.Body.String()) + } + }) +} + +func TestNewRouter_MetricsServesPrometheus(t *testing.T) { + router := newRouter(testDeps(stubDB{}, stubCache{}), discardLogger()) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + if !strings.Contains(rec.Body.String(), "# HELP") { + t.Fatalf("/metrics did not return a Prometheus exposition body: %s", rec.Body.String()[:200]) + } +} + +func TestNewRouter_HeartbeatRouteReachesPostHeartbeat(t *testing.T) { + router := newRouter(testDeps(stubDB{}, stubCache{}), discardLogger()) + + // An empty body fails PostHeartbeat's binding on the required "job" + // field. Only that handler can produce this 400. + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodPost, "/v1/heartbeat", strings.NewReader(`{}`)) + req.Header.Set("Content-Type", "application/json") + router.ServeHTTP(rec, req) + + if rec.Code != http.StatusBadRequest { + t.Fatalf("status = %d, want %d (body: %s)", rec.Code, http.StatusBadRequest, rec.Body.String()) + } +} + +func TestNewRouter_LatestHeartbeatRouteBindsTheJobParam(t *testing.T) { + jobID := uuid.New() + router := newRouter( + testDeps(stubDB{job: &db.Job{ID: jobID, Name: "nightly-etl"}}, stubCache{}), + discardLogger(), + ) + + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/v1/jobs/nightly-etl/heartbeat/latest", nil)) + + if rec.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusOK) + } + // The handler echoes the resolved job id, so seeing it here proves the + // :job path segment was actually bound and passed through. + if !strings.Contains(rec.Body.String(), jobID.String()) { + t.Fatalf("body = %s, want it to contain the resolved job id %s", rec.Body.String(), jobID) + } + if got := rec.Header().Get("X-Cache"); got != "MISS" { + t.Fatalf("X-Cache = %q, want MISS", got) + } +} + +func TestNewRouter_UnknownRouteIs404(t *testing.T) { + router := newRouter(testDeps(stubDB{}, stubCache{}), discardLogger()) + rec := httptest.NewRecorder() + router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/v1/nope", nil)) + + if rec.Code != http.StatusNotFound { + t.Fatalf("status = %d, want %d", rec.Code, http.StatusNotFound) + } +} diff --git a/ingestion-service/cmd/server/main.go b/ingestion-service/cmd/server/main.go index 44dbee3..31f61c1 100644 --- a/ingestion-service/cmd/server/main.go +++ b/ingestion-service/cmd/server/main.go @@ -2,6 +2,10 @@ // receives job heartbeats and writes them to Postgres, updating Redis so // dashboard reads stay fast. Runs independently of the Django core-api, // sharing only the Postgres schema Django's migrations own. +// +// main() is deliberately nothing but wiring: read the environment, dial the +// two dependencies, serve, and shut down cleanly. Every decision it makes +// lives in bootstrap.go, where it is tested. package main import ( @@ -14,28 +18,19 @@ import ( "time" "github.com/gin-gonic/gin" - "github.com/prometheus/client_golang/prometheus/promhttp" "pipelineops/ingestion-service/internal/cache" "pipelineops/ingestion-service/internal/db" "pipelineops/ingestion-service/internal/handlers" ) -func getenv(key, fallback string) string { - if v := os.Getenv(key); v != "" { - return v - } - return fallback -} - func main() { logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) port := getenv("PORT", "8080") databaseURL := getenv("DATABASE_URL", "postgres://pipelineops:pipelineops@localhost:5432/pipelineops") redisURL := getenv("REDIS_URL", "redis://localhost:6379/0") - ginMode := getenv("GIN_MODE", "release") - gin.SetMode(ginMode) + gin.SetMode(getenv("GIN_MODE", "release")) ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() @@ -58,22 +53,9 @@ func main() { } }() - deps := handlers.Deps{DB: pool, Cache: redisClient, Logger: logger} - - router := gin.New() - router.Use(gin.Recovery(), slogRequestLogger(logger)) - - router.GET("/healthz", handlers.Healthz(deps)) - router.GET("/metrics", gin.WrapH(promhttp.Handler())) - v1 := router.Group("/v1") - { - v1.POST("/heartbeat", deps.PostHeartbeat) - v1.GET("/jobs/:job/heartbeat/latest", deps.GetLatestHeartbeat) - } - srv := &http.Server{ Addr: ":" + port, - Handler: router, + Handler: newRouter(handlers.Deps{DB: pool, Cache: redisClient, Logger: logger}, logger), ReadHeaderTimeout: 5 * time.Second, } @@ -94,49 +76,3 @@ func main() { logger.Error("graceful shutdown failed", "error", err) } } - -// connectWithRetry calls connect up to 15 times, sleeping 2 seconds between -// attempts, until it succeeds or the attempts are exhausted. connect and -// sleep are injected so tests can substitute a fake connector and a -// non-blocking sleep; main passes the real db.Connect and time.Sleep. -func connectWithRetry(logger *slog.Logger, connect func() (*db.Pool, error), sleep func(time.Duration)) (*db.Pool, error) { - var lastErr error - for i := 0; i < 15; i++ { - pool, err := connect() - if err == nil { - return pool, nil - } - lastErr = err - logger.Warn("postgres not ready, retrying", "attempt", i+1, "error", err) - sleep(2 * time.Second) - } - return nil, lastErr -} - -// connectRedisWithRetry mirrors connectWithRetry for the Redis client. -func connectRedisWithRetry(logger *slog.Logger, connect func() (*cache.Client, error), sleep func(time.Duration)) (*cache.Client, error) { - var lastErr error - for i := 0; i < 15; i++ { - client, err := connect() - if err == nil { - return client, nil - } - lastErr = err - logger.Warn("redis not ready, retrying", "attempt", i+1, "error", err) - sleep(2 * time.Second) - } - return nil, lastErr -} - -func slogRequestLogger(logger *slog.Logger) gin.HandlerFunc { - return func(c *gin.Context) { - start := time.Now() - c.Next() - logger.Info("request", - "method", c.Request.Method, - "path", c.Request.URL.Path, - "status", c.Writer.Status(), - "duration_ms", time.Since(start).Milliseconds(), - ) - } -} diff --git a/ingestion-service/internal/db/db.go b/ingestion-service/internal/db/db.go index c255a1d..2967780 100644 --- a/ingestion-service/internal/db/db.go +++ b/ingestion-service/internal/db/db.go @@ -95,7 +95,15 @@ func (p *Pool) InsertHeartbeat(ctx context.Context, params InsertHeartbeatParams // matters if it fires before Commit, which the final `return err` below covers. defer func() { _ = tx.Rollback(ctx) }() - receivedAt := time.Now().UTC() + // Postgres stores timestamptz at microsecond precision, so a + // nanosecond-precision time.Now() is silently truncated on the way in. + // Truncating here instead means the timestamp handed back to the caller is + // the same one a later read of this row returns — without it, the + // received_at in a heartbeat response disagrees with the stored value in + // the last three digits, which is the kind of discrepancy that surfaces + // much later as a confusing off-by-a-microsecond in a dashboard or an + // alert comparison. + receivedAt := time.Now().UTC().Truncate(time.Microsecond) var heartbeatID int64 err = tx.QueryRow(ctx, `