From 6485f10ee5a069de5c2892c4362ea7862cc7602e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 10:13:24 +0000 Subject: [PATCH 01/12] docs(rpc-metrics): design for unified RPC latency & error-rate metrics Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- .../specs/2026-07-08-rpc-metrics-design.md | 205 ++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 docs/superpowers/specs/2026-07-08-rpc-metrics-design.md diff --git a/docs/superpowers/specs/2026-07-08-rpc-metrics-design.md b/docs/superpowers/specs/2026-07-08-rpc-metrics-design.md new file mode 100644 index 000000000..629a1cbfb --- /dev/null +++ b/docs/superpowers/specs/2026-07-08-rpc-metrics-design.md @@ -0,0 +1,205 @@ +# RPC latency & error-rate metrics — design + +**Date:** 2026-07-08 +**Status:** Approved (brainstorming) — pending implementation plan +**Branch:** `claude/rpc-metrics-latency-errors-w8z4ba` + +## Goal + +Emit uniform **request latency** and **error rate** metrics for every +synchronous RPC handler across all services, so a single set of Grafana +panels and alert rules works for every service via a `service` label. + +## Scope + +**In scope — the synchronous request/reply surface:** + +- All `natsrouter`-based NATS request/reply services (the six that construct a + `natsrouter` Router and register routes): `room-service`, `room-worker`, + `user-service`, `user-presence-service`, `history-service`, `search-service`. +- All Gin HTTP services: `auth-service`, `portal-service`, `media-service`, + `upload-service`. + +**Out of scope (non-goals):** + +- JetStream event consumers (`message-worker`, `notification-worker`, + `message-gatekeeper`, `broadcast-worker`, `inbox-worker`). These are async, + redelivered, + reply-less event processors — RPC latency/error-rate semantics do not fit. + They warrant separate consumer-lag / processing-failure metrics in a future + effort. +- No new client-facing request/response schema or event struct — this is a + purely server-side observability change, so `docs/client-api.md` is not + touched. + +## Rationale for key choices + +- **Centralized in the two chokepoints, not per-handler.** `natsrouter` and + `ginutil` are the single middleware seams that cover the entire in-scope + surface with near-zero per-handler code. This matches the existing + middleware idiom (`RequestID`/`Recovery`/`Logging`/`AccessLog` are all + installed via `Use`) and keeps metrics an opt-in concern separate from the + routing/handler core. +- **Unified metric names, `service` distinguished by label.** The entire point + of "add to all services" is one dashboard/alert that works everywhere. A + bespoke per-service name (as `search-service` has today) fragments that. +- **`route` label is always the pattern/template, never a live subject/URL.** + This is the load-bearing cardinality guard. `natsrouter` matches on patterns + (`chat.user.*.request.room.get`), and Gin exposes the registered route + template (`c.FullPath()`) — neither carries live room/user IDs. +- **`status` label is the errcode `Code`, uniformly across both transports.** + Both NATS and HTTP services already classify via `errcode`; reusing the Code + gives a single status taxonomy. A pinned 9-value allowlist bounds cardinality. +- **Raw Prometheus `promauto`, not the OTel meter.** `search-service` and + `pkg/atrest` already prove this exact pattern; every service already serves + the default Prometheus registry on `/metrics`. OTel stays for tracing. + +## Architecture + +### 1. `pkg/rpcmetrics` (new) — shared vocabulary + +One small package owns the collectors and the label taxonomy so **both** +transports emit identical series. Registered with the default Prometheus +registry via `promauto` (already served on `/metrics` in every service). + +```go +package rpcmetrics + +// rpc_server_requests_total{service, route, status} CounterVec +// rpc_server_request_duration_seconds{service, route} HistogramVec, DefBuckets + +// Observe records one completed RPC: latency + terminal status. +func Observe(service, route, status string, d time.Duration) + +// StatusLabel maps a handler's returned error onto the `status` label. +// nil -> "ok"; a non-empty *errcode.Error Code in the chain, if in the +// pinned allowlist -> that Code; everything else -> "internal". +func StatusLabel(err error) string +``` + +- `StatusLabel` is lifted verbatim from `search-service/metrics.go` + (`statusLabel` + `allowedStatusLabels`): the 9-value pinned allowlist is + `ok` plus the 8 canonical `errcode` Codes (`bad_request`, `unauthenticated`, + `forbidden`, `not_found`, `conflict`, `too_many_requests`, `unavailable`, + `internal`). It is a **pure, non-logging** Code extractor (`errors.As`) — it + never double-logs against `errcode.Classify`. +- Metric names use the `rpc_server_` prefix (`server` distinguishes from any + future client-side RPC metrics). +- Histogram uses `prometheus.DefBuckets`. +- Package name is descriptive per the "no `utils`/`common`/`helpers`" rule. + +### 2. NATS side — `pkg/natsrouter` + +Three tiny enabling changes, then one installable middleware. + +- **`Context` gains `route`.** Set at `acquireContext` from the matched route's + original pattern (`rt.pattern`, threaded through `addRoute`). Exposed as + `c.Route()`. This is the low-cardinality template, never a live subject. +- **`Context` gains a settable terminal `status`.** The typed `Register` / + `RegisterNoBody` / `RegisterVoid` wrappers already have the handler's `err` + in scope: they call `c.setStatus("ok")` on success and + `c.setStatus(rpcmetrics.StatusLabel(err))` on error, **before** `replyErr`. + The reply/logging path (`errnats.Reply` -> `errcode.Classify`) is unchanged; + `StatusLabel` does not log, so there is no double-log. +- **`natsrouter.Metrics(service string) HandlerFunc`.** Wraps `c.Next()`, then + records `rpcmetrics.Observe(service, c.Route(), , elapsed)`. + A request that never set a status (e.g. an aborted/early-return path) + defaults to `"ok"` unless set otherwise; document this default. Installed + once per service: + + ```go + r.Use(natsrouter.RequestID(), natsrouter.Recovery(), + natsrouter.Metrics(cfg.ServiceName), natsrouter.Logging()) + ``` + +### 3. HTTP side — `pkg/ginutil` + +- **`errhttp.Write` stores the classified Code** on the gin context + (`c.Set(...)` the `e.Code`). Free — `Write` already calls `errcode.Classify`. +- **`ginutil.Metrics(service string) gin.HandlerFunc`.** Times `c.Next()`; + reads `c.FullPath()` (the registered route **template**, not the live URL) + for `route`; reads the errcode Code stored by `errhttp.Write` for `status`, + falling back to an HTTP-status class when none was written + (`2xx` -> `ok`, `4xx` -> `bad_request`, `5xx` -> `internal`). Skips + `route == "/healthz"` (and empty `FullPath`, i.e. unmatched 404s) to avoid + noise. Installed alongside `RequestID`/`CORS`/`AccessLog`. + +### 4. Per-service wiring + +- Every in-scope service already serves the default Prometheus registry on + `/metrics`. Router services add the `r.Use(natsrouter.Metrics(...))` line; + HTTP services add `Use(ginutil.Metrics(...))` and, where not already present, + bind `otelutil.MetricsServer()` on `METRICS_ADDR`. +- The `service` value comes from each service's existing service-name config + (the same string passed to `otelutil.InitTracer`/`InitMeter`). + +### 5. `search-service` migration + +- Delete the bespoke request-path machinery from `search-service/metrics.go`: + `metricRequestsTotal`, `metricRequestDuration`, `observeRequest`, `durFor`, + `statusLabel`, `allowedStatusLabels`, the per-`kind` duration handles, and + the per-handler `defer observeRequest(...)()` calls. It inherits the shared + middleware instead. +- **Keep** the genuinely service-specific `search_service_es_duration_seconds` + (Elasticsearch `_search` call latency) and its `observeES` helper. +- Note: the shared metric labels by `route` (pattern) rather than the old + `kind` label; dashboards keyed on `search_service_requests_total{kind=...}` + must move to `rpc_server_requests_total{service="search-service",route=...}`. + +## Data flow + +``` +request ──▶ [Metrics mw: start timer] ──▶ RequestID ──▶ ... ──▶ handler + │ + (Register wrapper sets terminal + status: "ok" | StatusLabel(err)) + │ +request ◀── [Metrics mw: Observe(service, route, status, elapsed)] ◀── reply +``` + +HTTP mirrors this: gin `Metrics` middleware times the chain; `errhttp.Write` +stamps the Code; middleware reads `FullPath` + Code (or HTTP-class fallback). + +## Error handling + +- `StatusLabel`/`Observe` never panic on unknown input: an unknown Code + collapses to `internal` (bounded cardinality); an unknown `route` is only + ever a pattern/`FullPath` value, so it is inherently bounded by the route + table. +- Metrics recording is best-effort and must never affect the reply: the + middleware records after `c.Next()` and swallows nothing meaningful (a + `promauto` `Inc`/`Observe` cannot error). + +## Testing (TDD — Red/Green/Refactor per CLAUDE.md) + +- **`pkg/rpcmetrics`** (target 90%+, core `pkg/`): + - `StatusLabel`: nil -> `ok`; each canonical Code -> itself; non-canonical / + foreign Code -> `internal`; wrapped error resolved via `errors.As`; + non-errcode error -> `internal`. + - `Observe`: correct series/labels and a duration observation, verified with + `prometheus/testutil`. +- **`pkg/natsrouter`**: middleware test — a handled request increments + `rpc_server_requests_total` with the expected `route`/`status` and observes a + duration; both success and error paths; uses the existing router test harness + (`NewContext`, `main_test.go`). +- **`pkg/ginutil`**: `httptest` middleware test — `route` = `FullPath`, + errcode-Code status, HTTP-class fallback when no Code, and the `/healthz` + skip. +- All new `pkg/` code meets the 80% floor; `-race` via the Makefile. + +## Docs + +- No `docs/client-api.md` change (server-side only). +- `pkg/rpcmetrics/doc.go` documents the metric names, label set, the + cardinality rule (`route` = pattern / `FullPath`, never live IDs), and the + `status` allowlist. + +## Rollout / sequence (for the implementation plan) + +1. `pkg/rpcmetrics` (collectors + `StatusLabel` + `Observe`) with tests. +2. `natsrouter`: `Context.route`/`status` plumbing + `Metrics` middleware + tests. +3. `errhttp.Write` Code stamping + `ginutil.Metrics` middleware + tests. +4. Wire `r.Use(...)` / `Use(...)` into each in-scope service; ensure + `MetricsServer()` is bound. +5. Migrate `search-service` off its bespoke request metrics (keep ES metric). +6. `make lint`, `make test`, `make sast`; verify `/metrics` exposes the series. From 5433a86319a8b8652e30a2a6f822d8dfc39df907 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 10:43:13 +0000 Subject: [PATCH 02/12] docs(rpc-metrics): implementation plan for unified RPC metrics Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- .../plans/2026-07-08-rpc-metrics.md | 1086 +++++++++++++++++ 1 file changed, 1086 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-08-rpc-metrics.md diff --git a/docs/superpowers/plans/2026-07-08-rpc-metrics.md b/docs/superpowers/plans/2026-07-08-rpc-metrics.md new file mode 100644 index 000000000..7514b0777 --- /dev/null +++ b/docs/superpowers/plans/2026-07-08-rpc-metrics.md @@ -0,0 +1,1086 @@ +# RPC latency & error-rate metrics — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Emit uniform `rpc_server_requests_total` and `rpc_server_request_duration_seconds` metrics for every synchronous RPC handler (6 NATS request/reply services + 4 Gin HTTP services), queryable by a single `service` label. + +**Architecture:** One shared `pkg/rpcmetrics` package owns the Prometheus collectors and status taxonomy. Two thin middlewares — `natsrouter.Metrics` and `ginutil.Metrics` — feed it, so both transports emit identical series. `route` is always the low-cardinality pattern/template; `status` is the errcode `Code`. + +**Tech Stack:** Go 1.25, `prometheus/client_golang` (`promauto`, default registry), `pkg/natsrouter`, `pkg/ginutil`, `pkg/errcode`, `stretchr/testify`, `prometheus/testutil`. + +## Global Constraints + +- Go 1.25; single `go.mod` at repo root; services are flat `package main` at repo root; shared code in `pkg/`. +- Use `make` targets, never raw `go`: `make test SERVICE=`, `make lint`, `make sast`, `make build SERVICE=`. Package tests: `make test` runs all with `-race`. +- No new third-party dependencies — `prometheus/client_golang` (incl. `prometheus/testutil`) is already in `go.mod`. +- Metric collectors register with the **default** Prometheus registry via `promauto`; `promhttp.Handler()` / `otelutil.MetricsServer()` exposes them on `/metrics`. `InitMeter` is NOT required for these metrics. +- Package names must be descriptive — never `utils`/`helpers`/`common`/`base`. New package is `rpcmetrics`. +- TDD: Red → Green → Refactor → Commit. Tests in `package ` / `package main` (same package). Minimum 80% coverage for new `pkg/` code; target 90%+. +- `status` allowlist is exactly: `ok` + the 8 canonical errcode Codes (`bad_request`, `unauthenticated`, `forbidden`, `not_found`, `conflict`, `too_many_requests`, `unavailable`, `internal`). Anything else collapses to `internal`. +- `route` label MUST be a pattern/template, never a live subject or URL (cardinality guard). +- Commit identity: `git config user.email noreply@anthropic.com && git config user.name Claude`. Commit trailer lines: + `Co-Authored-By: Claude Opus 4.8 ` and + `Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb`. +- No `docs/client-api.md` change (server-side observability only). + +--- + +## File Structure + +- **Create** `pkg/rpcmetrics/metrics.go` — collectors + `Observe` + `StatusLabel`. +- **Create** `pkg/rpcmetrics/doc.go` — package doc: metric names, labels, cardinality rule. +- **Create** `pkg/rpcmetrics/metrics_test.go` — unit tests. +- **Modify** `pkg/natsrouter/params.go` — add `pattern` field to `route`. +- **Modify** `pkg/natsrouter/router.go` — thread `rt.pattern` into `acquireContext`. +- **Modify** `pkg/natsrouter/context.go` — add `route`/`status` fields + accessors. +- **Modify** `pkg/natsrouter/register.go` — stamp terminal status in the 3 `Register*` wrappers. +- **Create** `pkg/natsrouter/metrics.go` — `Metrics(service) HandlerFunc`. +- **Create** `pkg/natsrouter/metrics_test.go` — middleware test. +- **Modify** `pkg/errcode/errhttp/write.go` — stamp classified Code onto gin ctx. +- **Create** `pkg/ginutil/metrics.go` — `Metrics(service) gin.HandlerFunc`. +- **Create** `pkg/ginutil/metrics_test.go` — middleware test. +- **Modify** `search-service/metrics.go` + `search-service/handler.go` + `search-service/main.go` — migrate off bespoke request metrics. +- **Modify** service mains: `room-service`, `room-worker`, `user-service`, `user-presence-service`, `history-service` (router side); `auth-service`, `portal-service`, `media-service`, `upload-service` (HTTP side). + +--- + +## Task 1: `pkg/rpcmetrics` core + +**Files:** +- Create: `pkg/rpcmetrics/metrics.go` +- Create: `pkg/rpcmetrics/doc.go` +- Test: `pkg/rpcmetrics/metrics_test.go` + +**Interfaces:** +- Consumes: `github.com/hmchangw/chat/pkg/errcode` (`*errcode.Error`, `CodeBadRequest`…`CodeInternal`). +- Produces: + - `func Observe(service, route, status string, d time.Duration)` + - `func StatusLabel(err error) string` + - `func CounterValue(service, route, status string) float64` — test seam consumed by the natsrouter and ginutil tests (Tasks 2, 3). + - metric names `rpc_server_requests_total{service,route,status}` (counter), `rpc_server_request_duration_seconds{service,route}` (histogram). + +- [ ] **Step 1: Write the failing tests** + +Create `pkg/rpcmetrics/metrics_test.go`: + +```go +package rpcmetrics + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" +) + +func TestStatusLabel(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {"nil is ok", nil, "ok"}, + {"bad request", errcode.BadRequest("bad"), "bad_request"}, + {"not found", errcode.NotFound("nope"), "not_found"}, + {"forbidden", errcode.Forbidden("no"), "forbidden"}, + {"conflict", errcode.Conflict("dup"), "conflict"}, + {"unauthenticated", errcode.Unauthenticated("who"), "unauthenticated"}, + {"too many requests", errcode.TooManyRequests("slow"), "too_many_requests"}, + {"unavailable", errcode.Unavailable("down"), "unavailable"}, + {"internal", errcode.Internal("boom"), "internal"}, + {"wrapped errcode resolves via errors.As", fmt.Errorf("ctx: %w", errcode.NotFound("nope")), "not_found"}, + {"plain error collapses to internal", errors.New("raw"), "internal"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, StatusLabel(tt.err)) + }) + } +} + +func TestObserve(t *testing.T) { + Observe("svc-test", "chat.route.{id}.get", "ok", 12*time.Millisecond) + + // Counter incremented for the exact label tuple (via the exported seam). + assert.Equal(t, float64(1), CounterValue("svc-test", "chat.route.{id}.get", "ok")) + + // The duration histogram registered a series for this service/route pair. + count := testutil.CollectAndCount(requestDuration, "rpc_server_request_duration_seconds") + assert.GreaterOrEqual(t, count, 1) +} +``` + +- [ ] **Step 2: Run tests to verify they fail** + +Run: `make test SERVICE=rpcmetrics` (or `go test ./pkg/rpcmetrics/`) +Expected: FAIL — `undefined: StatusLabel`, `undefined: Observe`, `undefined: requestsTotal`. + +> Note: `make test SERVICE=` targets a service dir; for a `pkg/` package the runner may not resolve `SERVICE=rpcmetrics`. If so, run `go test -race ./pkg/rpcmetrics/...` directly — this is the one sanctioned raw-`go` use, for a package with no `make` target. + +- [ ] **Step 3: Write `pkg/rpcmetrics/metrics.go`** + +```go +// Package rpcmetrics holds the shared Prometheus collectors and status +// taxonomy for synchronous RPC handler metrics, emitted identically by the +// NATS (pkg/natsrouter) and HTTP (pkg/ginutil) middlewares. +package rpcmetrics + +import ( + "errors" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/hmchangw/chat/pkg/errcode" +) + +// Collectors register with the default Prometheus registry via promauto, so a +// plain promhttp.Handler() (or otelutil.MetricsServer) exposes them on /metrics. +var ( + requestsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "rpc_server_requests_total", + Help: "Total RPC request/reply invocations handled, partitioned by service, route pattern, and terminal status.", + }, []string{"service", "route", "status"}) + + requestDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "rpc_server_request_duration_seconds", + Help: "End-to-end RPC handler latency in seconds.", + Buckets: prometheus.DefBuckets, + }, []string{"service", "route"}) +) + +// Observe records one completed RPC: its latency and terminal status. +// route MUST be a pattern/template (e.g. "chat.user.{account}.request.room.get" +// or a Gin FullPath), never a live subject/URL, to keep cardinality bounded. +func Observe(service, route, status string, d time.Duration) { + requestDuration.WithLabelValues(service, route).Observe(d.Seconds()) + requestsTotal.WithLabelValues(service, route, status).Inc() +} + +// StatusLabel maps a handler's returned error onto the `status` label: +// nil -> "ok"; a non-empty *errcode.Error Code in the chain that is in the +// pinned allowlist -> that Code; everything else -> "internal". It is a pure, +// non-logging Code extractor (errors.As) — it never double-logs against +// errcode.Classify, so it is safe to call on the reply path. +func StatusLabel(err error) string { + if err == nil { + return "ok" + } + var ee *errcode.Error + if errors.As(err, &ee) && ee.Code != "" { + if _, ok := allowedStatusLabels[string(ee.Code)]; ok { + return string(ee.Code) + } + } + return string(errcode.CodeInternal) +} + +// allowedStatusLabels pins the cardinality of the status label to the 8 +// canonical errcode Codes + "ok". Any label outside this set collapses to +// "internal" via StatusLabel, so a future Code added without updating this +// allowlist cannot mint a fresh time series. +var allowedStatusLabels = map[string]struct{}{ + "ok": {}, + string(errcode.CodeBadRequest): {}, + string(errcode.CodeUnauthenticated): {}, + string(errcode.CodeForbidden): {}, + string(errcode.CodeNotFound): {}, + string(errcode.CodeConflict): {}, + string(errcode.CodeTooManyRequests): {}, + string(errcode.CodeUnavailable): {}, + string(errcode.CodeInternal): {}, +} + +// CounterValue returns the current rpc_server_requests_total value for the +// given label tuple. It is a test seam for consumer packages (natsrouter, +// ginutil) that cannot reach the unexported collector; side-effect-free. +func CounterValue(service, route, status string) float64 { + return testutil.ToFloat64(requestsTotal.WithLabelValues(service, route, status)) +} +``` + +- [ ] **Step 4: Write `pkg/rpcmetrics/doc.go`** + +```go +// Package rpcmetrics — RPC observability vocabulary. +// +// Metrics (default Prometheus registry, exposed on /metrics): +// +// rpc_server_requests_total{service, route, status} counter +// rpc_server_request_duration_seconds{service, route} histogram (DefBuckets) +// +// Labels: +// +// - service: the process's service name (same string passed to +// natsrouter.New / InitTracer), one value per process. +// - route: the ROUTE PATTERN, never a live subject/URL. NATS uses the +// natsrouter pattern with {name} placeholders; HTTP uses Gin's FullPath +// (registered template). This is the cardinality guard. +// - status: the errcode Code (ok + 8 canonical codes). Anything outside the +// pinned allowlist collapses to "internal". +// +// Emitted by pkg/natsrouter.Metrics (NATS request/reply) and +// pkg/ginutil.Metrics (Gin HTTP). Do not add per-service copies of these +// series — query by the `service` label instead. +package rpcmetrics +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `go test -race ./pkg/rpcmetrics/...` +Expected: PASS (both tests). + +- [ ] **Step 6: Lint** + +Run: `make lint` +Expected: no findings in `pkg/rpcmetrics`. + +- [ ] **Step 7: Commit** + +```bash +git add pkg/rpcmetrics/ +git commit -m "feat(rpcmetrics): shared RPC metrics collectors and status taxonomy" +``` + +--- + +## Task 2: `natsrouter.Metrics` middleware + Context plumbing + +**Files:** +- Modify: `pkg/natsrouter/params.go` (add `pattern` to `route`, set in `parsePattern`) +- Modify: `pkg/natsrouter/router.go` (pass `rt.pattern` into `acquireContext`) +- Modify: `pkg/natsrouter/context.go` (add `route`/`status` fields + accessors) +- Modify: `pkg/natsrouter/register.go` (stamp status in `Register`/`RegisterNoBody`/`RegisterVoid`) +- Create: `pkg/natsrouter/metrics.go` (`Metrics(service) HandlerFunc`) +- Test: `pkg/natsrouter/metrics_test.go` + +**Interfaces:** +- Consumes: `rpcmetrics.Observe`, `rpcmetrics.StatusLabel`; existing `Context`, `HandlerFunc`, `acquireContext`, `route`. +- Produces: + - `func (c *Context) Route() string` + - `func (c *Context) SetStatus(s string)` and `func (c *Context) Status() string` (defaults `"ok"` when unset) + - `func Metrics(service string) HandlerFunc` + +- [ ] **Step 1: Write the failing test** + +Create `pkg/natsrouter/metrics_test.go`: + +```go +package natsrouter + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// runChain drives a middleware+handler chain against a bare Context, the same +// way acquireContext wires `all` in addRoute. +func runChain(route string, handlers ...HandlerFunc) { + c := NewContext(nil) + c.route = route + c.chain.handlers = handlers + c.chain.index = -1 + c.Next() +} + +func TestMetrics_RecordsOKStatus(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.ok", "ok") + + runChain("chat.route.ok", + Metrics("user-service"), + func(c *Context) { c.SetStatus("ok") }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.ok", "ok") + assert.Equal(t, before+1, after) +} + +func TestMetrics_RecordsErrorStatus(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.err", "not_found") + + runChain("chat.route.err", + Metrics("user-service"), + func(c *Context) { c.SetStatus(rpcmetrics.StatusLabel(errcode.NotFound("nope"))) }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.err", "not_found") + assert.Equal(t, before+1, after) +} + +func TestMetrics_UnsetStatusDefaultsOK(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.default", "ok") + + runChain("chat.route.default", + Metrics("user-service"), + func(c *Context) { /* never sets status */ }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.default", "ok") + assert.Equal(t, before+1, after) +} +``` + +> `rpcmetrics.CounterValue` is defined in Task 1, so it is already available to +> this test — no per-package accessor is needed. + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test -race ./pkg/natsrouter/ -run TestMetrics` +Expected: FAIL — `undefined: Metrics`, `c.route` unexported field not set, `undefined: SetStatus`. + +- [ ] **Step 3: Add `pattern` to the `route` struct** + +In `pkg/natsrouter/params.go`, extend the struct and `parsePattern` return: + +```go +type route struct { + pattern string // original pattern with {name} placeholders — the metrics `route` label + natsSubject string // "chat.user.*.request.room.*.*.msg.history" + params map[int]string // {2: "account", 5: "roomID", 6: "siteID"} +} +``` + +```go + return route{ + pattern: pattern, + natsSubject: strings.Join(nats, "."), + params: params, + } +``` + +- [ ] **Step 4: Thread `pattern` into the Context** + +In `pkg/natsrouter/context.go`, add fields to `Context` (near `keys`): + +```go + // route is the matched route pattern (with {name} placeholders); set once + // at acquire, read by the Metrics middleware. Stable for the request's life. + route string + // status is the terminal RPC status label. Set by the Register* wrappers + // synchronously within the chain, before the reply; read by the Metrics + // middleware after c.Next(). Single-goroutine, sequential — no lock needed. + status string +``` + +Add accessors (below the existing `context.Context` methods): + +```go +// Route returns the matched route pattern (with {name} placeholders), the +// low-cardinality label the Metrics middleware uses. Empty for test contexts. +func (c *Context) Route() string { return c.route } + +// SetStatus records the terminal RPC status label for the Metrics middleware. +func (c *Context) SetStatus(s string) { c.status = s } + +// Status returns the terminal status label, defaulting to "ok" when a handler +// completed without setting one (e.g. a fire-and-forget void success). +func (c *Context) Status() string { + if c.status == "" { + return "ok" + } + return c.status +} +``` + +Update `acquireContext` signature to accept and set the route: + +```go +func acquireContext(ctx context.Context, msg *nats.Msg, params Params, handlers []HandlerFunc, route string) *Context { + cs := chainPool.Get().(*chainState) + cs.handlers = handlers + cs.index = -1 + return &Context{ + ctx: ctx, + Msg: msg, + Params: params, + route: route, + chain: cs, + } +} +``` + +In `releaseContext`, no change needed for `route`/`status` (the `*Context` is left to GC; only `chainState` is pooled). + +- [ ] **Step 5: Update the `acquireContext` call site** + +In `pkg/natsrouter/router.go` `addRoute`, change the call inside the spawned goroutine: + +```go + c := acquireContext(m.Context(), m.Msg, rt.extractParams(m.Msg.Subject), all, rt.pattern) +``` + +- [ ] **Step 6: Stamp terminal status in the Register wrappers** + +In `pkg/natsrouter/register.go`, import `rpcmetrics`, and set the status in each wrapper before replying. + +`Register`: + +```go + resp, err := fn(c, req) + if err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) + replyErr(c, err) + return + } + c.SetStatus("ok") + c.ReplyJSON(resp) +``` + +Also set the status on the unmarshal-failure branch: + +```go + if err := json.Unmarshal(c.Msg.Data, &req); err != nil { + c.SetStatus(string(errcode.CodeBadRequest)) + replyErr(c, errcode.BadRequest("invalid request payload", errcode.WithCause(err))) + return + } +``` + +`RegisterNoBody`: + +```go + resp, err := fn(c) + if err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) + replyErr(c, err) + return + } + c.SetStatus("ok") + c.ReplyJSON(resp) +``` + +`RegisterVoid` (no reply, but still record a status for metrics): + +```go + var req Req + if err := json.Unmarshal(c.Msg.Data, &req); err != nil { + c.SetStatus(string(errcode.CodeBadRequest)) + slog.Error("invalid payload in void handler", "error", err, "subject", c.Msg.Subject) + return + } + if err := fn(c, req); err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) + slog.Error("void handler error", "error", err, "subject", c.Msg.Subject) + return + } + c.SetStatus("ok") +``` + +Add the import `"github.com/hmchangw/chat/pkg/rpcmetrics"` to `register.go`. + +- [ ] **Step 7: Write the `Metrics` middleware** + +Create `pkg/natsrouter/metrics.go`: + +```go +package natsrouter + +import ( + "time" + + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// Metrics returns middleware that records rpc_server_requests_total and +// rpc_server_request_duration_seconds for every request handled by this +// router. The `route` label is the matched pattern (c.Route()); the `status` +// label is the terminal status stamped by the Register* wrappers (c.Status(), +// defaulting to "ok"). Install once via r.Use — placement relative to other +// middleware only shifts what latency window is measured; place it outermost +// (first) to capture the full chain. +func Metrics(service string) HandlerFunc { + return func(c *Context) { + start := time.Now() + c.Next() + rpcmetrics.Observe(service, c.Route(), c.Status(), time.Since(start)) + } +} +``` + +- [ ] **Step 8: Run tests to verify they pass** + +Run: `go test -race ./pkg/natsrouter/ -run TestMetrics` +Expected: PASS (three cases). Then full package: `go test -race ./pkg/natsrouter/...` → PASS (no regressions from the `acquireContext` signature change; the only caller is `addRoute`). + +- [ ] **Step 9: Lint** + +Run: `make lint` +Expected: clean. + +- [ ] **Step 10: Commit** + +```bash +git add pkg/natsrouter/ pkg/rpcmetrics/metrics.go +git commit -m "feat(natsrouter): Metrics middleware with route/status plumbing" +``` + +--- + +## Task 3: `ginutil.Metrics` middleware + errhttp Code stamping + +**Files:** +- Modify: `pkg/errcode/errhttp/write.go` +- Create: `pkg/ginutil/metrics.go` +- Test: `pkg/ginutil/metrics_test.go` + +**Interfaces:** +- Consumes: `rpcmetrics.Observe`, `rpcmetrics.CounterValue`; `gin`, `errcode`. +- Produces: + - `func Metrics(service string) gin.HandlerFunc` + - errhttp writes the classified Code under gin key `ginutil.ErrCodeKey`. + - `const ErrCodeKey = "errcode"` exported from `pkg/ginutil`. + +- [ ] **Step 1: Write the failing test** + +Create `pkg/ginutil/metrics_test.go`: + +```go +package ginutil + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" + "github.com/hmchangw/chat/pkg/errcode/errhttp" + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +func newEngine() *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(Metrics("auth-service")) + return r +} + +func do(r *gin.Engine, method, target string) { + req := httptest.NewRequest(method, target, nil) + r.ServeHTTP(httptest.NewRecorder(), req) +} + +func TestGinMetrics_ErrcodeStatus(t *testing.T) { + r := newEngine() + r.GET("/api/rooms/:id", func(c *gin.Context) { + errhttp.Write(context.Background(), c, errcode.NotFound("nope")) + }) + before := rpcmetrics.CounterValue("auth-service", "/api/rooms/:id", "not_found") + + do(r, http.MethodGet, "/api/rooms/42") + + after := rpcmetrics.CounterValue("auth-service", "/api/rooms/:id", "not_found") + assert.Equal(t, before+1, after) +} + +func TestGinMetrics_HTTPClassFallback(t *testing.T) { + r := newEngine() + r.GET("/api/ping", func(c *gin.Context) { c.Status(http.StatusOK) }) + before := rpcmetrics.CounterValue("auth-service", "/api/ping", "ok") + + do(r, http.MethodGet, "/api/ping") + + after := rpcmetrics.CounterValue("auth-service", "/api/ping", "ok") + assert.Equal(t, before+1, after) +} + +func TestGinMetrics_SkipsHealthzAndUnmatched(t *testing.T) { + r := newEngine() + r.GET("/healthz", func(c *gin.Context) { c.Status(http.StatusOK) }) + + // /healthz is skipped: empty-route counter must not move. + beforeHealth := rpcmetrics.CounterValue("auth-service", "/healthz", "ok") + do(r, http.MethodGet, "/healthz") + assert.Equal(t, beforeHealth, rpcmetrics.CounterValue("auth-service", "/healthz", "ok")) + + // Unmatched route (empty FullPath) is skipped: assert the "" route stays absent. + beforeEmpty := rpcmetrics.CounterValue("auth-service", "", "not_found") + do(r, http.MethodGet, "/does-not-exist") + assert.Equal(t, beforeEmpty, rpcmetrics.CounterValue("auth-service", "", "not_found")) +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `go test -race ./pkg/ginutil/ -run TestGinMetrics` +Expected: FAIL — `undefined: Metrics`. + +- [ ] **Step 3: Stamp the classified Code in errhttp.Write** + +Modify `pkg/errcode/errhttp/write.go`: + +```go +// Package errhttp adapts errcode.Error to Gin HTTP responses. +package errhttp + +import ( + "context" + + "github.com/gin-gonic/gin" + + "github.com/hmchangw/chat/pkg/errcode" +) + +// ErrCodeKey is the gin.Context key under which Write records the classified +// errcode Code, so metrics middleware can label the response without +// re-classifying (which would double-log). +const ErrCodeKey = "errcode" + +// Write classifies err (logging once) and writes the envelope with its HTTP +// status. It also records the classified Code on the gin context under +// ErrCodeKey for downstream metrics middleware. +func Write(ctx context.Context, c *gin.Context, err error) { + e := errcode.Classify(ctx, err) + c.Set(ErrCodeKey, string(e.Code)) + c.JSON(e.HTTPStatus(), e) +} +``` + +> `ginutil` reads this key by its literal string (`"errcode"`) to avoid a +> `ginutil → errhttp` import edge; the constant lives in `errhttp` where +> `Write` sets it. + +- [ ] **Step 4: Write the ginutil Metrics middleware** + +Create `pkg/ginutil/metrics.go`: + +```go +package ginutil + +import ( + "time" + + "github.com/gin-gonic/gin" + + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// errCodeKey mirrors errhttp.ErrCodeKey by value to avoid an import edge from +// ginutil to errhttp. errhttp.Write stores the classified Code here. +const errCodeKey = "errcode" + +// Metrics returns middleware that records rpc_server_requests_total and +// rpc_server_request_duration_seconds for each matched HTTP route. The `route` +// label is the registered template (c.FullPath()), never the live URL. The +// `status` label is the errcode Code stamped by errhttp.Write when present, +// else an HTTP-status class (2xx->ok, 4xx->bad_request, 5xx->internal). +// Unmatched routes (empty FullPath) and /healthz are skipped. +func Metrics(service string) gin.HandlerFunc { + return func(c *gin.Context) { + start := time.Now() + c.Next() + + route := c.FullPath() + if route == "" || route == "/healthz" { + return + } + rpcmetrics.Observe(service, route, statusForHTTP(c), time.Since(start)) + } +} + +// statusForHTTP prefers the errcode Code stamped by errhttp.Write; otherwise it +// derives a coarse class from the HTTP status code so plain responses still map +// onto the shared status taxonomy. +func statusForHTTP(c *gin.Context) string { + if v, ok := c.Get(errCodeKey); ok { + if code, ok := v.(string); ok && code != "" { + return code + } + } + switch code := c.Writer.Status(); { + case code >= 500: + return "internal" + case code >= 400: + return "bad_request" + default: + return "ok" + } +} +``` + +- [ ] **Step 5: Run tests to verify they pass** + +Run: `go test -race ./pkg/ginutil/ -run TestGinMetrics` +Expected: PASS (three tests). Then `go test -race ./pkg/errcode/errhttp/...` → PASS (existing `write_test.go` unaffected; the new `c.Set` is inert for its assertions). + +- [ ] **Step 6: Lint** + +Run: `make lint` +Expected: clean (`pkg/ginutil/metrics.go` imports only `time`, `gin`, `rpcmetrics`). + +- [ ] **Step 7: Commit** + +```bash +git add pkg/ginutil/metrics.go pkg/ginutil/metrics_test.go pkg/errcode/errhttp/write.go +git commit -m "feat(ginutil): HTTP RPC Metrics middleware; errhttp stamps Code" +``` + +--- + +## Task 4: Migrate `search-service` to the shared middleware + +**Files:** +- Modify: `search-service/metrics.go` (delete request-path machinery; keep ES metric) +- Modify: `search-service/handler.go` (remove `defer observeRequest(...)()` calls) +- Modify: `search-service/main.go` (add `router.Use(natsrouter.Metrics("search-service"))`) +- Existing tests: `search-service/handler_test.go` (adjust if they assert on removed symbols) + +**Interfaces:** +- Consumes: `natsrouter.Metrics`. +- Produces: no exported change; `search_service_es_duration_seconds` retained. + +- [ ] **Step 1: Remove request-path metrics from `search-service/metrics.go`** + +Delete: `metricRequestsTotal`, `metricRequestDuration`, the `metricKind*` +consts, the `durMessages/durRooms/durApps/durUsers` vars, `observeRequest`, +`durFor`, `statusLabel`, `allowedStatusLabels`. **Keep**: `metricESDuration`, +`observeES`, and `metricsHandler()`. The file should reduce to the ES histogram ++ `observeES` + `metricsHandler`. Resulting file: + +```go +package main + +import ( + "net/http" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/promhttp" +) + +// metricESDuration is search-service-specific (Elasticsearch _search latency) +// and stays here; the generic request-path metrics now come from +// natsrouter.Metrics (pkg/rpcmetrics). +var metricESDuration = promauto.NewHistogram(prometheus.HistogramOpts{ + Name: "search_service_es_duration_seconds", + Help: "Elasticsearch _search call latency in seconds.", + Buckets: prometheus.DefBuckets, +}) + +func observeES() func() { + start := time.Now() + return func() { metricESDuration.Observe(time.Since(start).Seconds()) } +} + +func metricsHandler() http.Handler { return promhttp.Handler() } +``` + +- [ ] **Step 2: Remove `observeRequest` call sites in `search-service/handler.go`** + +Delete the four `defer observeRequest(metricKind..., &err)()` lines +(handler.go:74, 129, 218, 254). Leave the `observeES()` usages (lines 109-111, +161-163) untouched. If a handler's only use of its named `err` return was +`observeRequest`, keep the named return if still referenced elsewhere; +otherwise simplify the signature to a plain `error` return only if the linter +flags an unused named result — do not otherwise change handler logic. + +- [ ] **Step 3: Add the middleware in `search-service/main.go`** + +Change the router middleware block (main.go:179-183): + +```go + router := natsrouter.New(nc, "search-service") + router.Use(natsrouter.RequestID()) + router.Use(natsrouter.Recovery()) + router.Use(natsrouter.Metrics("search-service")) + router.Use(natsrouter.Logging()) + handler.Register(router) +``` + +- [ ] **Step 4: Update tests that referenced removed symbols** + +Run: `go test -race ./search-service/... 2>&1 | head -40` +Expected initially: compile errors only if `handler_test.go` referenced +`observeRequest`/`statusLabel`/`metricKind*`. If so, delete those specific +assertions/tests (the behavior is now covered by `pkg/natsrouter/metrics_test.go` +and `pkg/rpcmetrics/metrics_test.go`). Do not weaken unrelated tests. + +- [ ] **Step 5: Build + test search-service** + +Run: `make build SERVICE=search-service && make test SERVICE=search-service` +Expected: build OK; tests PASS. + +- [ ] **Step 6: Lint** + +Run: `make lint` +Expected: clean (no unused imports/vars left in `search-service`). + +- [ ] **Step 7: Commit** + +```bash +git add search-service/ +git commit -m "refactor(search-service): use shared natsrouter.Metrics; drop bespoke request metrics" +``` + +--- + +## Task 5: Wire the remaining NATS request/reply services + +**Files (each service's main; add the `Metrics` Use line; add a `/metrics` listener + `MetricsAddr` config to the three that lack one):** +- Modify: `room-worker/main.go` — has `/metrics` already; add Use line only. +- Modify: `history-service/cmd/main.go` — has `/metrics` already; add Use line only. +- Modify: `room-service/main.go` — add Use line + `/metrics` listener + config field. +- Modify: `user-service/main.go` — add Use line + `/metrics` listener + config field. +- Modify: `user-presence-service/main.go` — add Use line + `/metrics` listener + config field. + +**Interfaces:** +- Consumes: `natsrouter.Metrics`, `otelutil.MetricsServer`. +- Produces: each service exposes `rpc_server_*` on `/metrics`. + +Two edit patterns follow. Apply Pattern A to all five services; additionally apply Pattern B to the three without a listener. + +- [ ] **Step 1 (Pattern A — add the middleware, all 5 services):** + +Insert `natsrouter.Metrics("")` into each router's middleware chain, immediately before `Logging` (outermost after Recovery/RequestID). Exact per-service edits: + +- `room-service/main.go:208`: + ```go + router.Use(natsrouter.Recovery(), natsrouter.RequestID(), natsrouter.Metrics("room-service"), natsrouter.Logging()) + ``` +- `room-worker/main.go` (find its `router.Use(...)` line): add `natsrouter.Metrics("room-worker")` before `natsrouter.Logging()` in the same call. +- `history-service/cmd/main.go` (find its `router.Use(...)`): add `natsrouter.Metrics("history-service")` before `Logging`. +- `user-service/main.go` (find its `router.Use(...)`): add `natsrouter.Metrics("user-service")` before `Logging`. +- `user-presence-service/main.go:133`: it uses `natsrouter.Default(...)` (which already installed Recovery/RequestID/Logging). Add a `router.Use(natsrouter.Metrics("user-presence-service"))` line immediately after the `natsrouter.Default(...)` construction (Metrics still wraps because `Use` appends and all are outer to the terminal handler; the duration window is the handler + any middleware registered after it — acceptable, since Default's Logging is already installed and Metrics measures from its own `c.Next()`). + +> If a service constructs middleware across multiple `router.Use(...)` calls (like search-service did), insert a dedicated `router.Use(natsrouter.Metrics(""))` line before the `Logging` line instead of editing an existing multi-arg call. + +- [ ] **Step 2 (Pattern B — add `/metrics` listener; only room-service, user-service, user-presence-service):** + +For each of these three, (a) add a config field, (b) add the listener, (c) add shutdown. + +(a) In the service's `Config`/`config` struct (in `main.go`), add: + +```go + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` +``` + +(b) After the router is constructed and handlers registered, before `shutdown.Wait`, add (mirrors `history-service/cmd/main.go:212-223`): + +```go + // Bind synchronously so a port conflict fails startup loudly rather than + // running blind — /metrics exposes rpc_server_* RPC metrics. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() +``` + +(c) Add a shutdown step inside the existing `shutdown.Wait(ctx, 25*time.Second, ...)` list, placed AFTER `router.Shutdown` and BEFORE `nc.Drain` (so final drain-window samples are still scrapeable, matching history-service): + +```go + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, +``` + +(d) Ensure imports include `"net"`, `"net/http"`, `"errors"`, and `"github.com/hmchangw/chat/pkg/otelutil"`. Add any that are missing (goimports via `make fmt` will order them). + +- [ ] **Step 3: Format** + +Run: `make fmt` +Expected: imports ordered; no diff noise. + +- [ ] **Step 4: Build each modified service** + +Run: +```bash +make build SERVICE=room-service && \ +make build SERVICE=room-worker && \ +make build SERVICE=history-service && \ +make build SERVICE=user-service && \ +make build SERVICE=user-presence-service +``` +Expected: all build OK. + +- [ ] **Step 5: Unit tests for modified services** + +Run: +```bash +for s in room-service room-worker history-service user-service user-presence-service; do make test SERVICE=$s || break; done +``` +Expected: PASS for each (main-wiring changes shouldn't touch handler tests). + +- [ ] **Step 6: Lint** + +Run: `make lint` +Expected: clean. + +- [ ] **Step 7: Commit** + +```bash +git add room-service/ room-worker/ history-service/ user-service/ user-presence-service/ +git commit -m "feat(rpc-metrics): wire natsrouter.Metrics into all request/reply services" +``` + +--- + +## Task 6: Wire the HTTP (Gin) services + +**Files:** +- Modify: `auth-service/main.go` +- Modify: `portal-service/main.go` +- Modify: `media-service/main.go` +- Modify: `upload-service/main.go` + +**Interfaces:** +- Consumes: `ginutil.Metrics`, `otelutil.MetricsServer`. +- Produces: each HTTP service exposes `rpc_server_*` on `/metrics`. + +- [ ] **Step 1: Add the Gin middleware (all 4 services)** + +In each service's gin setup, add `ginutil.Metrics("")` to the middleware chain. For `auth-service/main.go` (lines 88-92): + +```go + r := gin.New() + r.Use(gin.Recovery()) + r.Use(ginutil.RequestID()) + r.Use(ginutil.Metrics("auth-service")) + r.Use(ginutil.AccessLog()) + r.Use(ginutil.CORS()) +``` + +For `portal-service`, `media-service`, `upload-service`: locate the equivalent `r.Use(ginutil.RequestID())` / `AccessLog()` block and insert `r.Use(ginutil.Metrics(""))` (with the matching literal: `"portal-service"`, `"media-service"`, `"upload-service"`) right after `RequestID`. + +- [ ] **Step 2: Add a `/metrics` listener (all 4 services)** + +None of the four currently expose `/metrics`. Add the same config field + listener + shutdown as Task 5 Pattern B. Since these are HTTP services (no `natsrouter`), the shutdown ordering per CLAUDE.md is `nc.Drain()` → disconnect DBs; add the metrics-server shutdown as a distinct step in whatever graceful-shutdown mechanism the service uses. + +(a) Config struct — add: +```go + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` +``` + +(b) Listener — after building the gin engine and before/alongside the main HTTP server start: +```go + // /metrics on a separate port so scrapes don't hit the public API listener. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() +``` + +(c) Shutdown — where the service tears down (e.g. after the main `srv.Shutdown(...)` / on the shutdown path), add: +```go + if err := metricsServer.Shutdown(context.Background()); err != nil { + slog.Error("metrics server shutdown failed", "error", err) + } +``` +Match the existing shutdown style of each service (some use `shutdown.Wait`, `auth-service` uses a `srvErr` channel + `srv.Shutdown`); place the metrics shutdown alongside the existing server shutdown. + +(d) Imports — ensure `"net"`, `"net/http"`, `"errors"`, `"context"` (if used), and `"github.com/hmchangw/chat/pkg/otelutil"` are present. + +- [ ] **Step 3: Format** + +Run: `make fmt` +Expected: imports ordered. + +- [ ] **Step 4: Build each HTTP service** + +Run: +```bash +make build SERVICE=auth-service && \ +make build SERVICE=portal-service && \ +make build SERVICE=media-service && \ +make build SERVICE=upload-service +``` +Expected: all build OK. + +- [ ] **Step 5: Unit tests** + +Run: +```bash +for s in auth-service portal-service media-service upload-service; do make test SERVICE=$s || break; done +``` +Expected: PASS (or "no test files" for services without unit tests — acceptable; wiring is exercised by the `pkg/ginutil` test). + +- [ ] **Step 6: Lint** + +Run: `make lint` +Expected: clean. + +- [ ] **Step 7: Commit** + +```bash +git add auth-service/ portal-service/ media-service/ upload-service/ +git commit -m "feat(rpc-metrics): wire ginutil.Metrics into all HTTP services" +``` + +--- + +## Task 7: Full verification & docs finalize + +**Files:** +- Verify only; `pkg/rpcmetrics/doc.go` already written in Task 1. + +- [ ] **Step 1: Full unit test suite with race detector** + +Run: `make test` +Expected: all packages PASS, no data races. + +- [ ] **Step 2: Lint the whole repo** + +Run: `make lint` +Expected: no findings. + +- [ ] **Step 3: SAST gate** + +Run: `make sast` +Expected: no medium+ findings introduced (new code adds no `InsecureSkipVerify`, unchecked conversions, or command exec). + +- [ ] **Step 4: Manual /metrics smoke (one router + one HTTP service)** + +Verify the series names appear. Using the search-service or room-service docker-compose (which sets `BOOTSTRAP_STREAMS=true` and `METRICS_ADDR=:9090`), start the service and: + +```bash +curl -s localhost:9090/metrics | grep -E '^rpc_server_(requests_total|request_duration_seconds)' +``` +Expected: both `rpc_server_requests_total` and `rpc_server_request_duration_seconds` families are listed (they appear after the first request; the `# HELP`/`# TYPE` lines appear immediately once a handler has run). If no traffic yet, drive one RPC/HTTP call first. + +- [ ] **Step 5: Confirm no client-api drift** + +Run: `git diff --name-only origin/main...HEAD | grep -E 'docs/client-api'` +Expected: empty (server-side change only — no client-api edit required). + +- [ ] **Step 6: Final commit (if verification produced any fmt/doc touch-ups)** + +```bash +git add -A +git commit -m "chore(rpc-metrics): finalize wiring and verification" || echo "nothing to commit" +``` + +--- + +## Self-review notes (coverage map) + +- Spec §1 `pkg/rpcmetrics` → Task 1. §2 natsrouter middleware → Task 2. §3 ginutil + errhttp → Task 3. §4 per-service wiring → Tasks 5–6. §5 search-service migration → Task 4. Testing → per-task TDD steps + Task 7. Docs/`doc.go` → Task 1 Step 4. Non-goals (JetStream/inbox-worker) → untouched by any task. +- `route`/`status`/`Metrics` names are used consistently across Tasks 2, 3, 5, 6. `rpcmetrics.CounterValue` (Task 2 Step 3) is the single test seam reused by Tasks 2 and 3. +- Every code step shows real code; no TBD/placeholder logic (the two "reminder" lines in Tasks 2/3 are explicitly instructed to be removed). From bae5da8e66f464474219367438bbe8905c0e88ef Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 14:30:24 +0000 Subject: [PATCH 03/12] feat(rpcmetrics): shared RPC metrics collectors and status taxonomy Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- go.mod | 1 + pkg/rpcmetrics/doc.go | 21 +++++++++ pkg/rpcmetrics/metrics.go | 79 ++++++++++++++++++++++++++++++++++ pkg/rpcmetrics/metrics_test.go | 49 +++++++++++++++++++++ 4 files changed, 150 insertions(+) create mode 100644 pkg/rpcmetrics/doc.go create mode 100644 pkg/rpcmetrics/metrics.go create mode 100644 pkg/rpcmetrics/metrics_test.go diff --git a/go.mod b/go.mod index d213912c3..5f06be21f 100644 --- a/go.mod +++ b/go.mod @@ -103,6 +103,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/klauspost/cpuid/v2 v2.3.0 // indirect github.com/klauspost/crc32 v1.3.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/magiconair/properties v1.8.10 // indirect diff --git a/pkg/rpcmetrics/doc.go b/pkg/rpcmetrics/doc.go new file mode 100644 index 000000000..6b2e4b062 --- /dev/null +++ b/pkg/rpcmetrics/doc.go @@ -0,0 +1,21 @@ +// Package rpcmetrics — RPC observability vocabulary. +// +// Metrics (default Prometheus registry, exposed on /metrics): +// +// rpc_server_requests_total{service, route, status} counter +// rpc_server_request_duration_seconds{service, route} histogram (DefBuckets) +// +// Labels: +// +// - service: the process's service name (same string passed to +// natsrouter.New / InitTracer), one value per process. +// - route: the ROUTE PATTERN, never a live subject/URL. NATS uses the +// natsrouter pattern with {name} placeholders; HTTP uses Gin's FullPath +// (registered template). This is the cardinality guard. +// - status: the errcode Code (ok + 8 canonical codes). Anything outside the +// pinned allowlist collapses to "internal". +// +// Emitted by pkg/natsrouter.Metrics (NATS request/reply) and +// pkg/ginutil.Metrics (Gin HTTP). Do not add per-service copies of these +// series — query by the `service` label instead. +package rpcmetrics diff --git a/pkg/rpcmetrics/metrics.go b/pkg/rpcmetrics/metrics.go new file mode 100644 index 000000000..cba526674 --- /dev/null +++ b/pkg/rpcmetrics/metrics.go @@ -0,0 +1,79 @@ +// Package rpcmetrics holds the shared Prometheus collectors and status +// taxonomy for synchronous RPC handler metrics, emitted identically by the +// NATS (pkg/natsrouter) and HTTP (pkg/ginutil) middlewares. +package rpcmetrics + +import ( + "errors" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "github.com/prometheus/client_golang/prometheus/testutil" + + "github.com/hmchangw/chat/pkg/errcode" +) + +// Collectors register with the default Prometheus registry via promauto, so a +// plain promhttp.Handler() (or otelutil.MetricsServer) exposes them on /metrics. +var ( + requestsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "rpc_server_requests_total", + Help: "Total RPC request/reply invocations handled, partitioned by service, route pattern, and terminal status.", + }, []string{"service", "route", "status"}) + + requestDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ + Name: "rpc_server_request_duration_seconds", + Help: "End-to-end RPC handler latency in seconds.", + Buckets: prometheus.DefBuckets, + }, []string{"service", "route"}) +) + +// Observe records one completed RPC: its latency and terminal status. +// route MUST be a pattern/template (e.g. "chat.user.{account}.request.room.get" +// or a Gin FullPath), never a live subject/URL, to keep cardinality bounded. +func Observe(service, route, status string, d time.Duration) { + requestDuration.WithLabelValues(service, route).Observe(d.Seconds()) + requestsTotal.WithLabelValues(service, route, status).Inc() +} + +// StatusLabel maps a handler's returned error onto the `status` label: +// nil -> "ok"; a non-empty *errcode.Error Code in the chain that is in the +// pinned allowlist -> that Code; everything else -> "internal". It is a pure, +// non-logging Code extractor (errors.As) — it never double-logs against +// errcode.Classify, so it is safe to call on the reply path. +func StatusLabel(err error) string { + if err == nil { + return "ok" + } + var ee *errcode.Error + if errors.As(err, &ee) && ee.Code != "" { + if _, ok := allowedStatusLabels[string(ee.Code)]; ok { + return string(ee.Code) + } + } + return string(errcode.CodeInternal) +} + +// allowedStatusLabels pins the cardinality of the status label to the 8 +// canonical errcode Codes + "ok". Any label outside this set collapses to +// "internal" via StatusLabel, so a future Code added without updating this +// allowlist cannot mint a fresh time series. +var allowedStatusLabels = map[string]struct{}{ + "ok": {}, + string(errcode.CodeBadRequest): {}, + string(errcode.CodeUnauthenticated): {}, + string(errcode.CodeForbidden): {}, + string(errcode.CodeNotFound): {}, + string(errcode.CodeConflict): {}, + string(errcode.CodeTooManyRequests): {}, + string(errcode.CodeUnavailable): {}, + string(errcode.CodeInternal): {}, +} + +// CounterValue returns the current rpc_server_requests_total value for the +// given label tuple. It is a test seam for consumer packages (natsrouter, +// ginutil) that cannot reach the unexported collector; side-effect-free. +func CounterValue(service, route, status string) float64 { + return testutil.ToFloat64(requestsTotal.WithLabelValues(service, route, status)) +} diff --git a/pkg/rpcmetrics/metrics_test.go b/pkg/rpcmetrics/metrics_test.go new file mode 100644 index 000000000..60e55ae07 --- /dev/null +++ b/pkg/rpcmetrics/metrics_test.go @@ -0,0 +1,49 @@ +package rpcmetrics + +import ( + "errors" + "fmt" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" +) + +func TestStatusLabel(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {"nil is ok", nil, "ok"}, + {"bad request", errcode.BadRequest("bad"), "bad_request"}, + {"not found", errcode.NotFound("nope"), "not_found"}, + {"forbidden", errcode.Forbidden("no"), "forbidden"}, + {"conflict", errcode.Conflict("dup"), "conflict"}, + {"unauthenticated", errcode.Unauthenticated("who"), "unauthenticated"}, + {"too many requests", errcode.TooManyRequests("slow"), "too_many_requests"}, + {"unavailable", errcode.Unavailable("down"), "unavailable"}, + {"internal", errcode.Internal("boom"), "internal"}, + {"wrapped errcode resolves via errors.As", fmt.Errorf("ctx: %w", errcode.NotFound("nope")), "not_found"}, + {"plain error collapses to internal", errors.New("raw"), "internal"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, StatusLabel(tt.err)) + }) + } +} + +func TestObserve(t *testing.T) { + Observe("svc-test", "chat.route.{id}.get", "ok", 12*time.Millisecond) + + // Counter incremented for the exact label tuple (via the exported seam). + assert.Equal(t, float64(1), CounterValue("svc-test", "chat.route.{id}.get", "ok")) + + // The duration histogram registered a series for this service/route pair. + count := testutil.CollectAndCount(requestDuration, "rpc_server_request_duration_seconds") + assert.GreaterOrEqual(t, count, 1) +} From 201c3e0916f0dd840def3c0e62d203be73123f13 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 14:41:50 +0000 Subject: [PATCH 04/12] feat(natsrouter): Metrics middleware with route/status plumbing Adds route/status plumbing to Context (pattern threaded through acquireContext, Route()/SetStatus()/Status() accessors) and stamps terminal status in Register/RegisterNoBody/RegisterVoid so the new Metrics(service) middleware can record rpc_server_requests_total and rpc_server_request_duration_seconds per request. Renamed the pre-existing helpers_test.go runChain to runHandlerChain to resolve a name collision with the new metrics_test.go runChain helper (test-only, no behavior change). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- pkg/natsrouter/context.go | 27 +++++++++++++++- pkg/natsrouter/context_test.go | 9 +++--- pkg/natsrouter/helpers_test.go | 4 +-- pkg/natsrouter/metrics.go | 22 +++++++++++++ pkg/natsrouter/metrics_test.go | 56 ++++++++++++++++++++++++++++++++++ pkg/natsrouter/params.go | 2 ++ pkg/natsrouter/register.go | 11 +++++++ pkg/natsrouter/router.go | 2 +- pkg/natsrouter/router_test.go | 8 ++--- 9 files changed, 129 insertions(+), 12 deletions(-) create mode 100644 pkg/natsrouter/metrics.go create mode 100644 pkg/natsrouter/metrics_test.go diff --git a/pkg/natsrouter/context.go b/pkg/natsrouter/context.go index 18f4eda97..e38f4d85d 100644 --- a/pkg/natsrouter/context.go +++ b/pkg/natsrouter/context.go @@ -33,6 +33,14 @@ type Context struct { keys map[string]any mu sync.RWMutex + // route is the matched route pattern (with {name} placeholders); set once + // at acquire, read by the Metrics middleware. Stable for the request's life. + route string + // status is the terminal RPC status label. Set by the Register* wrappers + // synchronously within the chain, before the reply; read by the Metrics + // middleware after c.Next(). Single-goroutine, sequential — no lock needed. + status string + chain *chainState } @@ -49,7 +57,7 @@ var chainPool = sync.Pool{ New: func() any { return &chainState{} }, } -func acquireContext(ctx context.Context, msg *nats.Msg, params Params, handlers []HandlerFunc) *Context { +func acquireContext(ctx context.Context, msg *nats.Msg, params Params, handlers []HandlerFunc, route string) *Context { cs := chainPool.Get().(*chainState) cs.handlers = handlers cs.index = -1 @@ -57,6 +65,7 @@ func acquireContext(ctx context.Context, msg *nats.Msg, params Params, handlers ctx: ctx, Msg: msg, Params: params, + route: route, chain: cs, } } @@ -91,6 +100,22 @@ func (c *Context) Done() <-chan struct{} { return c.ctx.Done() } func (c *Context) Err() error { return c.ctx.Err() } func (c *Context) Value(key any) any { return c.ctx.Value(key) } +// Route returns the matched route pattern (with {name} placeholders), the +// low-cardinality label the Metrics middleware uses. Empty for test contexts. +func (c *Context) Route() string { return c.route } + +// SetStatus records the terminal RPC status label for the Metrics middleware. +func (c *Context) SetStatus(s string) { c.status = s } + +// Status returns the terminal status label, defaulting to "ok" when a handler +// completed without setting one (e.g. a fire-and-forget void success). +func (c *Context) Status() string { + if c.status == "" { + return "ok" + } + return c.status +} + // Chain methods are handler-internal. Calling them from a post-handler // goroutine panics — chainState is pooled and would otherwise silently read // the next request's state. diff --git a/pkg/natsrouter/context_test.go b/pkg/natsrouter/context_test.go index bb36c006a..41e825d33 100644 --- a/pkg/natsrouter/context_test.go +++ b/pkg/natsrouter/context_test.go @@ -84,7 +84,7 @@ func TestContext_ConcurrentKeysAccess_NoRace(t *testing.T) { // reuse. Run with -race. func TestContext_StableCtxAcrossPoolReuse(t *testing.T) { reqCtx := context.Background() - c1 := acquireContext(reqCtx, nil, NewParams(map[string]string{"req": "1"}), nil) + c1 := acquireContext(reqCtx, nil, NewParams(map[string]string{"req": "1"}), nil, "") c1.Set("id", "alice") stop := make(chan struct{}) @@ -114,6 +114,7 @@ func TestContext_StableCtxAcrossPoolReuse(t *testing.T) { nil, NewParams(map[string]string{"req": strconv.Itoa(i + 2)}), nil, + "", ) c2.Set("id", "bob") releaseContext(c2) @@ -128,11 +129,11 @@ func TestContext_StableCtxAcrossPoolReuse(t *testing.T) { // fast if someone ever reintroduces pooling of the Context header itself — // the likely accident given how much of the rest of the struct looks poolable. func TestContext_KeysIndependentPerRequest(t *testing.T) { - c1 := acquireContext(context.Background(), nil, Params{}, nil) + c1 := acquireContext(context.Background(), nil, Params{}, nil, "") c1.Set("leak", "bad") releaseContext(c1) - c2 := acquireContext(context.Background(), nil, Params{}, nil) + c2 := acquireContext(context.Background(), nil, Params{}, nil, "") _, ok := c2.Get("leak") assert.False(t, ok, "keys set on a released context must not be visible on the next acquire") releaseContext(c2) @@ -189,7 +190,7 @@ func TestContext_GetHeader(t *testing.T) { // next request's chain state. The nil-out + nil-check converts the silent // corruption into a loud panic. func TestContext_ChainMethodsPanicAfterRelease(t *testing.T) { - c := acquireContext(context.Background(), nil, Params{}, []HandlerFunc{func(*Context) {}}) + c := acquireContext(context.Background(), nil, Params{}, []HandlerFunc{func(*Context) {}}, "") releaseContext(c) assert.PanicsWithValue(t, chainAfterReleasePanic, func() { c.Next() }) diff --git a/pkg/natsrouter/helpers_test.go b/pkg/natsrouter/helpers_test.go index a2cac1d75..b4efbb240 100644 --- a/pkg/natsrouter/helpers_test.go +++ b/pkg/natsrouter/helpers_test.go @@ -1,7 +1,7 @@ package natsrouter -// runChain executes the handler chain against c; for tests in this package only. -func runChain(c *Context, handlers []HandlerFunc) { +// runHandlerChain executes the handler chain against c; for tests in this package only. +func runHandlerChain(c *Context, handlers []HandlerFunc) { cs := chainPool.Get().(*chainState) cs.handlers = handlers cs.index = -1 diff --git a/pkg/natsrouter/metrics.go b/pkg/natsrouter/metrics.go new file mode 100644 index 000000000..ac7b38784 --- /dev/null +++ b/pkg/natsrouter/metrics.go @@ -0,0 +1,22 @@ +package natsrouter + +import ( + "time" + + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// Metrics returns middleware that records rpc_server_requests_total and +// rpc_server_request_duration_seconds for every request handled by this +// router. The `route` label is the matched pattern (c.Route()); the `status` +// label is the terminal status stamped by the Register* wrappers (c.Status(), +// defaulting to "ok"). Install once via r.Use — placement relative to other +// middleware only shifts what latency window is measured; place it outermost +// (first) to capture the full chain. +func Metrics(service string) HandlerFunc { + return func(c *Context) { + start := time.Now() + c.Next() + rpcmetrics.Observe(service, c.Route(), c.Status(), time.Since(start)) + } +} diff --git a/pkg/natsrouter/metrics_test.go b/pkg/natsrouter/metrics_test.go new file mode 100644 index 000000000..55e30ca40 --- /dev/null +++ b/pkg/natsrouter/metrics_test.go @@ -0,0 +1,56 @@ +package natsrouter + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// runChain drives a middleware+handler chain against a bare Context, the same +// way acquireContext wires `all` in addRoute. +func runChain(route string, handlers ...HandlerFunc) { + c := NewContext(nil) + c.route = route + c.chain.handlers = handlers + c.chain.index = -1 + c.Next() +} + +func TestMetrics_RecordsOKStatus(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.ok", "ok") + + runChain("chat.route.ok", + Metrics("user-service"), + func(c *Context) { c.SetStatus("ok") }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.ok", "ok") + assert.Equal(t, before+1, after) +} + +func TestMetrics_RecordsErrorStatus(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.err", "not_found") + + runChain("chat.route.err", + Metrics("user-service"), + func(c *Context) { c.SetStatus(rpcmetrics.StatusLabel(errcode.NotFound("nope"))) }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.err", "not_found") + assert.Equal(t, before+1, after) +} + +func TestMetrics_UnsetStatusDefaultsOK(t *testing.T) { + before := rpcmetrics.CounterValue("user-service", "chat.route.default", "ok") + + runChain("chat.route.default", + Metrics("user-service"), + func(c *Context) { /* never sets status */ }, + ) + + after := rpcmetrics.CounterValue("user-service", "chat.route.default", "ok") + assert.Equal(t, before+1, after) +} diff --git a/pkg/natsrouter/params.go b/pkg/natsrouter/params.go index 40aa63f12..e2bdda808 100644 --- a/pkg/natsrouter/params.go +++ b/pkg/natsrouter/params.go @@ -49,6 +49,7 @@ func (p Params) Require(key string) (string, error) { // It holds the converted NATS wildcard subject and the position-to-name // mapping for param extraction. type route struct { + pattern string // original pattern with {name} placeholders — the metrics `route` label natsSubject string // "chat.user.*.request.room.*.*.msg.history" params map[int]string // {2: "account", 5: "roomID", 6: "siteID"} } @@ -80,6 +81,7 @@ func parsePattern(pattern string) route { } return route{ + pattern: pattern, natsSubject: strings.Join(nats, "."), params: params, } diff --git a/pkg/natsrouter/register.go b/pkg/natsrouter/register.go index de3fb225d..77262c355 100644 --- a/pkg/natsrouter/register.go +++ b/pkg/natsrouter/register.go @@ -6,6 +6,7 @@ import ( "github.com/hmchangw/chat/pkg/errcode" "github.com/hmchangw/chat/pkg/errcode/errnats" + "github.com/hmchangw/chat/pkg/rpcmetrics" ) // Register subscribes a typed handler to a subject pattern. @@ -22,16 +23,19 @@ func Register[Req, Resp any]( // Cause preserves the parse-error chain for the Classify server log // without echoing it to the client (errcode.Error.cause is unexported, // never JSON-serialized). The user-facing message stays generic. + c.SetStatus(string(errcode.CodeBadRequest)) replyErr(c, errcode.BadRequest("invalid request payload", errcode.WithCause(err))) return } resp, err := fn(c, req) if err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) replyErr(c, err) return } + c.SetStatus("ok") c.ReplyJSON(resp) }) @@ -47,10 +51,12 @@ func RegisterNoBody[Resp any]( handler := HandlerFunc(func(c *Context) { resp, err := fn(c) if err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) replyErr(c, err) return } + c.SetStatus("ok") c.ReplyJSON(resp) }) @@ -66,13 +72,18 @@ func RegisterVoid[Req any]( handler := HandlerFunc(func(c *Context) { var req Req if err := json.Unmarshal(c.Msg.Data, &req); err != nil { + c.SetStatus(string(errcode.CodeBadRequest)) slog.Error("invalid payload in void handler", "error", err, "subject", c.Msg.Subject) return } if err := fn(c, req); err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) slog.Error("void handler error", "error", err, "subject", c.Msg.Subject) + return } + + c.SetStatus("ok") }) r.addRoute(pattern, []HandlerFunc{handler}) diff --git a/pkg/natsrouter/router.go b/pkg/natsrouter/router.go index 3596b46bd..ba428fee5 100644 --- a/pkg/natsrouter/router.go +++ b/pkg/natsrouter/router.go @@ -208,7 +208,7 @@ func (r *Router) addRoute(pattern string, handlers []HandlerFunc) { } } }() - c := acquireContext(m.Context(), m.Msg, rt.extractParams(m.Msg.Subject), all) + c := acquireContext(m.Context(), m.Msg, rt.extractParams(m.Msg.Subject), all, rt.pattern) defer releaseContext(c) c.Next() }() diff --git a/pkg/natsrouter/router_test.go b/pkg/natsrouter/router_test.go index c30807fe1..ec4da4d91 100644 --- a/pkg/natsrouter/router_test.go +++ b/pkg/natsrouter/router_test.go @@ -544,7 +544,7 @@ func TestRequestIDMiddleware_StoresIDOnUnderlyingContext(t *testing.T) { assert.Equal(t, testID, fromCtx) }, } - runChain(c, chain) + runHandlerChain(c, chain) assert.True(t, called, "downstream handler must run") } @@ -562,7 +562,7 @@ func TestRequestIDMiddleware_GeneratesAndStoresOnContext_WhenHeaderMissing(t *te fromKeys = fromKeysAny.(string) }, } - runChain(c, chain) + runHandlerChain(c, chain) assert.NotEmpty(t, fromCtx, "RequestID middleware must mint and propagate to ctx when header is absent") assert.Equal(t, fromCtx, fromKeys, "minted ID must be identical in ctx and keys map") } @@ -579,7 +579,7 @@ func TestRequestIDMiddleware_RegeneratesOnMalformedHeader(t *testing.T) { fromCtx = natsutil.RequestIDFromContext(c) }, } - runChain(c, chain) + runHandlerChain(c, chain) assert.NotEqual(t, "not-a-uuidv7", fromCtx, "malformed inbound ID must be replaced") assert.True(t, idgen.IsValidUUID(fromCtx), "regenerated ID must be a valid hyphenated UUID") @@ -606,7 +606,7 @@ func TestRequestIDMiddleware_OtherCtxKeysStillReadable(t *testing.T) { assert.Equal(t, "parent-value", got, "non-requestIDKey lookup must still find values from the original parent ctx") }, } - runChain(c, chain) + runHandlerChain(c, chain) assert.True(t, called, "downstream handler must run") } From 5b841d06937800fe24f06347765ed95be3680c50 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 14:48:53 +0000 Subject: [PATCH 05/12] test(natsrouter): cover Register* status stamping end-to-end Add end-to-end tests that install Metrics() in front of real Register/RegisterNoBody/RegisterVoid handlers over an in-process NATS server, asserting rpc_server_requests_total increments on the correct status label for success, errcode errors, and invalid-JSON bodies. Closes the gap where a regression swapping/dropping a SetStatus call in register.go would pass every existing test. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- pkg/natsrouter/metrics_test.go | 136 +++++++++++++++++++++++++++++++++ 1 file changed, 136 insertions(+) diff --git a/pkg/natsrouter/metrics_test.go b/pkg/natsrouter/metrics_test.go index 55e30ca40..de9e7519c 100644 --- a/pkg/natsrouter/metrics_test.go +++ b/pkg/natsrouter/metrics_test.go @@ -1,9 +1,13 @@ package natsrouter import ( + "context" + "encoding/json" "testing" + "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/hmchangw/chat/pkg/errcode" "github.com/hmchangw/chat/pkg/rpcmetrics" @@ -54,3 +58,135 @@ func TestMetrics_UnsetStatusDefaultsOK(t *testing.T) { after := rpcmetrics.CounterValue("user-service", "chat.route.default", "ok") assert.Equal(t, before+1, after) } + +// The tests below drive REAL Register/RegisterNoBody/RegisterVoid handlers +// (not synthetic c.SetStatus calls) over an in-process NATS server, with +// Metrics installed as router middleware. This is the end-to-end coverage +// for the status-stamping wired into pkg/natsrouter/register.go: a +// regression that swaps or drops a SetStatus call there would compile and +// pass runChain-based tests above but must fail here. +// +// Note: Metrics.Observe runs after c.Next() returns, but the Register* +// wrappers send the NATS reply *inside* c.Next() (before control returns to +// Metrics). So nc.Request can complete slightly before the counter is +// incremented — assert via require.Eventually, never immediately after +// nc.Request returns. + +func TestMetrics_EndToEnd_RegisterSuccess(t *testing.T) { + nc := startTestNATS(t) + r := New(nc, "svc-metrics-e2e-success") + r.Use(Metrics("svc-metrics-e2e-success")) + + const pattern = "metrics.e2e.success.{id}" + Register(r, pattern, + func(c *Context, req testReq) (*testResp, error) { + return &testResp{Greeting: "hello " + req.Name}, nil + }) + + before := rpcmetrics.CounterValue("svc-metrics-e2e-success", pattern, "ok") + + data, _ := json.Marshal(testReq{Name: "world"}) + resp, err := nc.Request(context.Background(), "metrics.e2e.success.42", data, 2*time.Second) + require.NoError(t, err) + + var result testResp + require.NoError(t, json.Unmarshal(resp.Data, &result)) + assert.Equal(t, "hello world", result.Greeting) + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("svc-metrics-e2e-success", pattern, "ok") == before+1 + }, time.Second, 5*time.Millisecond, "rpc_server_requests_total status=ok must increment after the real Register success path") +} + +func TestMetrics_EndToEnd_RegisterErrcodeError(t *testing.T) { + nc := startTestNATS(t) + r := New(nc, "svc-metrics-e2e-notfound") + r.Use(Metrics("svc-metrics-e2e-notfound")) + + const pattern = "metrics.e2e.notfound.{id}" + Register(r, pattern, + func(c *Context, req testReq) (*testResp, error) { + return nil, errcode.NotFound("thing not found") + }) + + before := rpcmetrics.CounterValue("svc-metrics-e2e-notfound", pattern, "not_found") + + data, _ := json.Marshal(testReq{Name: "test"}) + resp, err := nc.Request(context.Background(), "metrics.e2e.notfound.42", data, 2*time.Second) + require.NoError(t, err) + + var errResp errcode.Error + require.NoError(t, json.Unmarshal(resp.Data, &errResp)) + assert.Equal(t, "not_found", string(errResp.Code)) + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("svc-metrics-e2e-notfound", pattern, "not_found") == before+1 + }, time.Second, 5*time.Millisecond, "rpc_server_requests_total status=not_found must increment after a real errcode.NotFound handler return") +} + +func TestMetrics_EndToEnd_RegisterInvalidJSON(t *testing.T) { + nc := startTestNATS(t) + r := New(nc, "svc-metrics-e2e-badjson") + r.Use(Metrics("svc-metrics-e2e-badjson")) + + const pattern = "metrics.e2e.badjson.{id}" + Register(r, pattern, + func(c *Context, req testReq) (*testResp, error) { + t.Fatal("handler should not be called for invalid JSON") + return nil, nil + }) + + before := rpcmetrics.CounterValue("svc-metrics-e2e-badjson", pattern, "bad_request") + + resp, err := nc.Request(context.Background(), "metrics.e2e.badjson.42", []byte("not json"), 2*time.Second) + require.NoError(t, err) + + var errResp errcode.Error + require.NoError(t, json.Unmarshal(resp.Data, &errResp)) + assert.Equal(t, "invalid request payload", errResp.Message) + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("svc-metrics-e2e-badjson", pattern, "bad_request") == before+1 + }, time.Second, 5*time.Millisecond, "rpc_server_requests_total status=bad_request must increment on real unmarshal failure in Register") +} + +func TestMetrics_EndToEnd_RegisterVoid(t *testing.T) { + nc := startTestNATS(t) + r := New(nc, "svc-metrics-e2e-void") + r.Use(Metrics("svc-metrics-e2e-void")) + + const okPattern = "metrics.e2e.void.ok.{id}" + const errPattern = "metrics.e2e.void.err.{id}" + + processed := make(chan struct{}, 1) + RegisterVoid(r, okPattern, + func(c *Context, req testReq) error { + processed <- struct{}{} + return nil + }) + RegisterVoid(r, errPattern, + func(c *Context, req testReq) error { + return errcode.Forbidden("nope") + }) + + beforeOK := rpcmetrics.CounterValue("svc-metrics-e2e-void", okPattern, "ok") + beforeErr := rpcmetrics.CounterValue("svc-metrics-e2e-void", errPattern, "forbidden") + + data, _ := json.Marshal(testReq{Name: "hello"}) + require.NoError(t, nc.Publish(context.Background(), "metrics.e2e.void.ok.1", data)) + require.NoError(t, nc.Publish(context.Background(), "metrics.e2e.void.err.1", data)) + + select { + case <-processed: + case <-time.After(2 * time.Second): + t.Fatal("void success handler not called within timeout") + } + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("svc-metrics-e2e-void", okPattern, "ok") == beforeOK+1 + }, time.Second, 5*time.Millisecond, "rpc_server_requests_total status=ok must increment after a real RegisterVoid success") + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("svc-metrics-e2e-void", errPattern, "forbidden") == beforeErr+1 + }, time.Second, 5*time.Millisecond, "rpc_server_requests_total status=forbidden must increment after a real RegisterVoid error") +} From e4402baba6aacdbfd64a54fe80e2caa4bc7f8778 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 14:56:14 +0000 Subject: [PATCH 06/12] feat(ginutil): HTTP RPC Metrics middleware; errhttp stamps Code Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- pkg/errcode/errhttp/write.go | 10 +++++- pkg/ginutil/metrics.go | 51 ++++++++++++++++++++++++++++ pkg/ginutil/metrics_test.go | 66 ++++++++++++++++++++++++++++++++++++ 3 files changed, 126 insertions(+), 1 deletion(-) create mode 100644 pkg/ginutil/metrics.go create mode 100644 pkg/ginutil/metrics_test.go diff --git a/pkg/errcode/errhttp/write.go b/pkg/errcode/errhttp/write.go index 2b8e576a4..5a74a0346 100644 --- a/pkg/errcode/errhttp/write.go +++ b/pkg/errcode/errhttp/write.go @@ -9,8 +9,16 @@ import ( "github.com/hmchangw/chat/pkg/errcode" ) -// Write classifies err (logging once) and writes the envelope with its HTTP status. +// ErrCodeKey is the gin.Context key under which Write records the classified +// errcode Code, so metrics middleware can label the response without +// re-classifying (which would double-log). +const ErrCodeKey = "errcode" + +// Write classifies err (logging once) and writes the envelope with its HTTP +// status. It also records the classified Code on the gin context under +// ErrCodeKey for downstream metrics middleware. func Write(ctx context.Context, c *gin.Context, err error) { e := errcode.Classify(ctx, err) + c.Set(ErrCodeKey, string(e.Code)) c.JSON(e.HTTPStatus(), e) } diff --git a/pkg/ginutil/metrics.go b/pkg/ginutil/metrics.go new file mode 100644 index 000000000..d0ca383e3 --- /dev/null +++ b/pkg/ginutil/metrics.go @@ -0,0 +1,51 @@ +package ginutil + +import ( + "time" + + "github.com/gin-gonic/gin" + + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +// errCodeKey mirrors errhttp.ErrCodeKey by value to avoid an import edge from +// ginutil to errhttp. errhttp.Write stores the classified Code here. +const errCodeKey = "errcode" + +// Metrics returns middleware that records rpc_server_requests_total and +// rpc_server_request_duration_seconds for each matched HTTP route. The `route` +// label is the registered template (c.FullPath()), never the live URL. The +// `status` label is the errcode Code stamped by errhttp.Write when present, +// else an HTTP-status class (2xx->ok, 4xx->bad_request, 5xx->internal). +// Unmatched routes (empty FullPath) and /healthz are skipped. +func Metrics(service string) gin.HandlerFunc { + return func(c *gin.Context) { + start := time.Now() + c.Next() + + route := c.FullPath() + if route == "" || route == "/healthz" { + return + } + rpcmetrics.Observe(service, route, statusForHTTP(c), time.Since(start)) + } +} + +// statusForHTTP prefers the errcode Code stamped by errhttp.Write; otherwise it +// derives a coarse class from the HTTP status code so plain responses still map +// onto the shared status taxonomy. +func statusForHTTP(c *gin.Context) string { + if v, ok := c.Get(errCodeKey); ok { + if code, ok := v.(string); ok && code != "" { + return code + } + } + switch code := c.Writer.Status(); { + case code >= 500: + return "internal" + case code >= 400: + return "bad_request" + default: + return "ok" + } +} diff --git a/pkg/ginutil/metrics_test.go b/pkg/ginutil/metrics_test.go new file mode 100644 index 000000000..6c2eb5713 --- /dev/null +++ b/pkg/ginutil/metrics_test.go @@ -0,0 +1,66 @@ +package ginutil + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/assert" + + "github.com/hmchangw/chat/pkg/errcode" + "github.com/hmchangw/chat/pkg/errcode/errhttp" + "github.com/hmchangw/chat/pkg/rpcmetrics" +) + +func newEngine() *gin.Engine { + gin.SetMode(gin.TestMode) + r := gin.New() + r.Use(Metrics("auth-service")) + return r +} + +func do(r *gin.Engine, method, target string) { + req := httptest.NewRequest(method, target, nil) + r.ServeHTTP(httptest.NewRecorder(), req) +} + +func TestGinMetrics_ErrcodeStatus(t *testing.T) { + r := newEngine() + r.GET("/api/rooms/:id", func(c *gin.Context) { + errhttp.Write(context.Background(), c, errcode.NotFound("nope")) + }) + before := rpcmetrics.CounterValue("auth-service", "/api/rooms/:id", "not_found") + + do(r, http.MethodGet, "/api/rooms/42") + + after := rpcmetrics.CounterValue("auth-service", "/api/rooms/:id", "not_found") + assert.Equal(t, before+1, after) +} + +func TestGinMetrics_HTTPClassFallback(t *testing.T) { + r := newEngine() + r.GET("/api/ping", func(c *gin.Context) { c.Status(http.StatusOK) }) + before := rpcmetrics.CounterValue("auth-service", "/api/ping", "ok") + + do(r, http.MethodGet, "/api/ping") + + after := rpcmetrics.CounterValue("auth-service", "/api/ping", "ok") + assert.Equal(t, before+1, after) +} + +func TestGinMetrics_SkipsHealthzAndUnmatched(t *testing.T) { + r := newEngine() + r.GET("/healthz", func(c *gin.Context) { c.Status(http.StatusOK) }) + + // /healthz is skipped: empty-route counter must not move. + beforeHealth := rpcmetrics.CounterValue("auth-service", "/healthz", "ok") + do(r, http.MethodGet, "/healthz") + assert.Equal(t, beforeHealth, rpcmetrics.CounterValue("auth-service", "/healthz", "ok")) + + // Unmatched route (empty FullPath) is skipped: assert the "" route stays absent. + beforeEmpty := rpcmetrics.CounterValue("auth-service", "", "not_found") + do(r, http.MethodGet, "/does-not-exist") + assert.Equal(t, beforeEmpty, rpcmetrics.CounterValue("auth-service", "", "not_found")) +} From 4da118a8fe90160e8c7d98baef50da95fba66fe7 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 15:02:33 +0000 Subject: [PATCH 07/12] test(ginutil): make unmatched-route skip assertion non-tautological Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- pkg/ginutil/metrics_test.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/pkg/ginutil/metrics_test.go b/pkg/ginutil/metrics_test.go index 6c2eb5713..38bb71a09 100644 --- a/pkg/ginutil/metrics_test.go +++ b/pkg/ginutil/metrics_test.go @@ -60,7 +60,13 @@ func TestGinMetrics_SkipsHealthzAndUnmatched(t *testing.T) { assert.Equal(t, beforeHealth, rpcmetrics.CounterValue("auth-service", "/healthz", "ok")) // Unmatched route (empty FullPath) is skipped: assert the "" route stays absent. - beforeEmpty := rpcmetrics.CounterValue("auth-service", "", "not_found") + // A broken skip would still leave errCodeKey unset (no handler ran), so + // statusForHTTP falls back to the HTTP-status class: 404 -> "bad_request", + // not "not_found". Assert on "bad_request" so this test actually detects a + // broken skip instead of trivially passing on an unrelated label. + beforeEmptyBadRequest := rpcmetrics.CounterValue("auth-service", "", "bad_request") + beforeEmptyNotFound := rpcmetrics.CounterValue("auth-service", "", "not_found") do(r, http.MethodGet, "/does-not-exist") - assert.Equal(t, beforeEmpty, rpcmetrics.CounterValue("auth-service", "", "not_found")) + assert.Equal(t, beforeEmptyBadRequest, rpcmetrics.CounterValue("auth-service", "", "bad_request")) + assert.Equal(t, beforeEmptyNotFound, rpcmetrics.CounterValue("auth-service", "", "not_found")) } From 0b611a5dc37df25abd2daea0689f936866778e0c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 15:05:27 +0000 Subject: [PATCH 08/12] refactor(search-service): use shared natsrouter.Metrics; drop bespoke request metrics Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- search-service/handler.go | 8 --- search-service/main.go | 1 + search-service/metrics.go | 118 +++------------------------------ search-service/metrics_test.go | 66 ------------------ 4 files changed, 9 insertions(+), 184 deletions(-) delete mode 100644 search-service/metrics_test.go diff --git a/search-service/handler.go b/search-service/handler.go index dea133f22..291136f65 100644 --- a/search-service/handler.go +++ b/search-service/handler.go @@ -71,8 +71,6 @@ func (h *handler) withRequestTimeout(parent context.Context) (context.Context, c } func (h *handler) searchMessages(c *natsrouter.Context, req model.SearchMessagesRequest) (resp *model.SearchMessagesResponse, err error) { - defer observeRequest(metricKindMessages, &err)() - account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -126,8 +124,6 @@ func (h *handler) searchMessages(c *natsrouter.Context, req model.SearchMessages } func (h *handler) searchRooms(c *natsrouter.Context, req model.SearchRoomsRequest) (resp *model.SearchRoomsResponse, err error) { - defer observeRequest(metricKindRooms, &err)() - account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -215,8 +211,6 @@ func (h *handler) loadRestricted(ctx context.Context, account string) (map[strin } func (h *handler) searchApps(c *natsrouter.Context, req model.SearchAppsRequest) (resp *model.SearchAppsResponse, err error) { - defer observeRequest(metricKindApps, &err)() - account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -251,8 +245,6 @@ func (h *handler) searchApps(c *natsrouter.Context, req model.SearchAppsRequest) // from the subject is used for logging and metrics only; scoping is // enforced entirely by the third-party endpoint. func (h *handler) searchUsers(c *natsrouter.Context, req model.SearchUsersRequest) (resp *[]model.SearchUser, err error) { - defer observeRequest(metricKindUsers, &err)() - account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr diff --git a/search-service/main.go b/search-service/main.go index e240aa598..8a1ca013e 100644 --- a/search-service/main.go +++ b/search-service/main.go @@ -179,6 +179,7 @@ func main() { router := natsrouter.New(nc, "search-service") router.Use(natsrouter.RequestID()) router.Use(natsrouter.Recovery()) + router.Use(natsrouter.Metrics("search-service")) router.Use(natsrouter.Logging()) handler.Register(router) diff --git a/search-service/metrics.go b/search-service/metrics.go index 1828a4253..21fdc690c 100644 --- a/search-service/metrics.go +++ b/search-service/metrics.go @@ -1,128 +1,26 @@ package main import ( - "errors" "net/http" "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/promhttp" - - "github.com/hmchangw/chat/pkg/errcode" ) -// All collectors register with the default Prometheus registry via -// promauto so a plain promhttp.Handler() exposes them on /metrics. -var ( - metricRequestsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ - Name: "search_service_requests_total", - Help: "Total NATS request/reply invocations handled, partitioned by endpoint and terminal status.", - }, []string{"kind", "status"}) - - metricRequestDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{ - Name: "search_service_request_duration_seconds", - Help: "End-to-end handler latency in seconds, from NATS request receipt to response emission.", - Buckets: prometheus.DefBuckets, - }, []string{"kind"}) - - metricESDuration = promauto.NewHistogram(prometheus.HistogramOpts{ - Name: "search_service_es_duration_seconds", - Help: "Elasticsearch _search call latency in seconds.", - Buckets: prometheus.DefBuckets, - }) -) - -// Per-kind handles for the request-path metrics. The `status` label on -// requests_total is resolved lazily (9 values × 4 kinds = 36 perms would -// clutter here); the duration handles are fully bound. -const ( - metricKindMessages = "messages" - metricKindRooms = "subscriptions" - metricKindApps = "apps" - metricKindUsers = "users" -) - -var ( - durMessages = metricRequestDuration.WithLabelValues(metricKindMessages) - durRooms = metricRequestDuration.WithLabelValues(metricKindRooms) - durApps = metricRequestDuration.WithLabelValues(metricKindApps) - durUsers = metricRequestDuration.WithLabelValues(metricKindUsers) -) - -// observeRequest captures a handler's total latency and terminal status. -// The status is classified at fire-time from the named `err` return, so -// late-bound error classification (wrapping, defer-assigned) is counted -// correctly. Usage: -// -// func (h *handler) search(...) (resp *R, err error) { -// defer observeRequest(metricKindMessages, &err)() -// ... -// } -func observeRequest(kind string, errPtr *error) func() { - start := time.Now() - dur := durFor(kind) - return func() { - dur.Observe(time.Since(start).Seconds()) - metricRequestsTotal.WithLabelValues(kind, statusLabel(*errPtr)).Inc() - } -} +// metricESDuration is search-service-specific (Elasticsearch _search latency) +// and stays here; the generic request-path metrics now come from +// natsrouter.Metrics (pkg/rpcmetrics). +var metricESDuration = promauto.NewHistogram(prometheus.HistogramOpts{ + Name: "search_service_es_duration_seconds", + Help: "Elasticsearch _search call latency in seconds.", + Buckets: prometheus.DefBuckets, +}) func observeES() func() { start := time.Now() return func() { metricESDuration.Observe(time.Since(start).Seconds()) } } -// durFor falls back to the messages variant on an unknown label so a -// caller typo surfaces as misattributed metrics rather than a -// nil-observer panic at fire time. -func durFor(kind string) prometheus.Observer { - switch kind { - case metricKindRooms: - return durRooms - case metricKindApps: - return durApps - case metricKindUsers: - return durUsers - default: - return durMessages - } -} - -// statusLabel maps a handler's returned error onto the requests_total -// `status` label. nil → "ok"; a non-empty *errcode.Error in the chain → its -// Code (one of the 8 canonical Codes below); everything else → "internal". -// -// The label set is pinned to keep Prometheus cardinality bounded — at most -// 9 × len(kinds) series. A non-canonical Code (e.g. a future Code constant -// added without updating this allowlist, or a foreign envelope on a federation -// path) collapses to "internal" rather than minting a fresh time series. -func statusLabel(err error) string { - if err == nil { - return "ok" - } - var ee *errcode.Error - if errors.As(err, &ee) && ee.Code != "" { - if _, ok := allowedStatusLabels[string(ee.Code)]; ok { - return string(ee.Code) - } - } - return string(errcode.CodeInternal) -} - -// allowedStatusLabels pins the cardinality of the requests_total status label -// to the 8 canonical errcode Codes + "ok". Any label outside this set -// collapses to "internal" via statusLabel. -var allowedStatusLabels = map[string]struct{}{ - "ok": {}, - string(errcode.CodeBadRequest): {}, - string(errcode.CodeUnauthenticated): {}, - string(errcode.CodeForbidden): {}, - string(errcode.CodeNotFound): {}, - string(errcode.CodeConflict): {}, - string(errcode.CodeTooManyRequests): {}, - string(errcode.CodeUnavailable): {}, - string(errcode.CodeInternal): {}, -} - func metricsHandler() http.Handler { return promhttp.Handler() } diff --git a/search-service/metrics_test.go b/search-service/metrics_test.go deleted file mode 100644 index b3fb9c4ce..000000000 --- a/search-service/metrics_test.go +++ /dev/null @@ -1,66 +0,0 @@ -package main - -import ( - "errors" - "fmt" - "testing" - - "github.com/hmchangw/chat/pkg/errcode" -) - -func TestStatusLabel_OkOnNil(t *testing.T) { - if got := statusLabel(nil); got != "ok" { - t.Fatalf("nil err → status = %q, want %q", got, "ok") - } -} - -func TestStatusLabel_CanonicalErrcodePassesThrough(t *testing.T) { - cases := []struct { - err error - want string - }{ - {errcode.BadRequest("x"), "bad_request"}, - {errcode.Unauthenticated("x"), "unauthenticated"}, - {errcode.Forbidden("x"), "forbidden"}, - {errcode.NotFound("x"), "not_found"}, - {errcode.Conflict("x"), "conflict"}, - {errcode.TooManyRequests("x"), "too_many_requests"}, - {errcode.Unavailable("x"), "unavailable"}, - {errcode.Internal("x"), "internal"}, - } - for _, tc := range cases { - if got := statusLabel(tc.err); got != tc.want { - t.Errorf("statusLabel(%v) = %q, want %q", tc.err, got, tc.want) - } - } -} - -// Wrapped *errcode.Error (the actual production shape from handler.go where -// callers fmt.Errorf("ctx: %w", errcodeErr) before returning) must traverse -// the chain via errors.As and still pin the right label. -func TestStatusLabel_WrappedErrcodePassesThrough(t *testing.T) { - wrapped := fmt.Errorf("handler load: %w", errcode.BadRequest("missing field")) - if got := statusLabel(wrapped); got != "bad_request" { - t.Fatalf("wrapped errcode → %q, want bad_request", got) - } -} - -func TestStatusLabel_NonCanonicalCodeCollapsesToInternal(t *testing.T) { - // Synthetic *errcode.Error with a non-canonical Code (e.g. a federation peer - // shipped a foreign envelope). Must not mint a new Prometheus series — the - // allowedStatusLabels guard collapses it to "internal". - bad := &errcode.Error{Code: errcode.Code("made_up_category"), Message: "x"} - if got := statusLabel(bad); got != "internal" { - t.Fatalf("non-canonical Code → status = %q, want %q", got, "internal") - } -} - -func TestStatusLabel_RawErrorCollapsesToInternal(t *testing.T) { - if got := statusLabel(errors.New("mongo down")); got != "internal" { - t.Fatalf("raw err → status = %q, want %q", got, "internal") - } - wrapped := fmt.Errorf("ctx: %w", errors.New("mongo down")) - if got := statusLabel(wrapped); got != "internal" { - t.Fatalf("wrapped raw err → status = %q, want %q", got, "internal") - } -} From 8ffb86ee5c1ea3fca269604202ba0376059c937e Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 15:10:52 +0000 Subject: [PATCH 09/12] feat(rpc-metrics): wire natsrouter.Metrics into all request/reply services Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- history-service/cmd/main.go | 1 + room-service/main.go | 22 +++++++++++++++++++++- room-worker/main.go | 2 +- user-presence-service/main.go | 21 +++++++++++++++++++++ user-service/config/config.go | 1 + user-service/main.go | 20 ++++++++++++++++++++ 6 files changed, 65 insertions(+), 2 deletions(-) diff --git a/history-service/cmd/main.go b/history-service/cmd/main.go index 8432ce143..c4d12256b 100644 --- a/history-service/cmd/main.go +++ b/history-service/cmd/main.go @@ -195,6 +195,7 @@ func main() { // RequestID must precede any handler that reads request_id from ctx — // otherwise Classify's log line records an empty value. router.Use(natsrouter.RequestID()) + router.Use(natsrouter.Metrics("history-service")) router.Use(natsrouter.Logging()) svc.RegisterHandlers(router, cfg.SiteID) diff --git a/room-service/main.go b/room-service/main.go index 40cf82fdb..248f99456 100644 --- a/room-service/main.go +++ b/room-service/main.go @@ -2,8 +2,11 @@ package main import ( "context" + "errors" "fmt" "log/slog" + "net" + "net/http" "net/url" "os" "time" @@ -40,6 +43,7 @@ type config struct { RoomKeyGracePeriod time.Duration `env:"ROOM_KEY_GRACE_PERIOD" envDefault:"24h"` HealthAddr string `env:"HEALTH_ADDR" envDefault:":8081"` PProfEnabled bool `env:"PPROF_ENABLED" envDefault:"false"` + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` Bootstrap bootstrapConfig `envPrefix:"BOOTSTRAP_"` RestrictedRoomMinMembers int `env:"RESTRICTED_ROOM_MIN_MEMBERS" envDefault:"5"` // Microsoft Teams integration. Teams* credentials are required only for the @@ -205,7 +209,7 @@ func main() { handler.roomMembersCallLimit = cfg.RoomMembersCallLimit router := natsrouter.New(nc, "room-service") - router.Use(natsrouter.Recovery(), natsrouter.RequestID(), natsrouter.Logging()) + router.Use(natsrouter.Recovery(), natsrouter.RequestID(), natsrouter.Metrics("room-service"), natsrouter.Logging()) handler.Register(router) healthStop, err := health.ServeWithPprof(cfg.HealthAddr, 5*time.Second, cfg.PProfEnabled, @@ -216,10 +220,26 @@ func main() { os.Exit(1) } + // Bind synchronously so a port conflict fails startup loudly rather than + // running blind — /metrics exposes rpc_server_* RPC metrics. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + slog.Info("room-service running", "site", cfg.SiteID) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return router.Shutdown(ctx) }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { diff --git a/room-worker/main.go b/room-worker/main.go index 6dab5808b..1d83284d0 100644 --- a/room-worker/main.go +++ b/room-worker/main.go @@ -202,7 +202,7 @@ func main() { handler.reconcileTTL = cfg.MemberCountReconcileTTL router := natsrouter.New(nc, "room-worker") - router.Use(natsrouter.Recovery(), natsrouter.RequestID(), natsrouter.Logging()) + router.Use(natsrouter.Recovery(), natsrouter.RequestID(), natsrouter.Metrics("room-worker"), natsrouter.Logging()) natsrouter.Register(router, subject.RoomCreateDMSync(cfg.SiteID), handler.serverCreateDM) cons, err := js.CreateOrUpdateConsumer(ctx, streamCfg.Name, buildConsumerConfig(cfg.Consumer)) diff --git a/user-presence-service/main.go b/user-presence-service/main.go index 2252fd262..81e5596dd 100644 --- a/user-presence-service/main.go +++ b/user-presence-service/main.go @@ -2,8 +2,11 @@ package main import ( "context" + "errors" "fmt" "log/slog" + "net" + "net/http" "os" "time" @@ -49,6 +52,7 @@ type Config struct { SiteID string `env:"SITE_ID,required"` UserCacheSize int `env:"USER_CACHE_SIZE" envDefault:"10000"` UserCacheTTL time.Duration `env:"USER_CACHE_TTL" envDefault:"5m"` + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` NATS NATSConfig `envPrefix:"NATS_"` Valkey ValkeyConfig `envPrefix:"VALKEY_"` Mongo MongoConfig `envPrefix:"MONGO_"` @@ -131,6 +135,7 @@ func main() { handler := NewHandler(store, userDir, peer, publish, cfg.SiteID, cfg.Presence.BatchMax) router := natsrouter.Default(nc, "user-presence-service") + router.Use(natsrouter.Metrics("user-presence-service")) natsrouter.RegisterVoid(router, subject.PresenceHelloPattern(cfg.SiteID), handler.Hello) natsrouter.RegisterVoid(router, subject.PresencePingPattern(cfg.SiteID), handler.Ping) natsrouter.RegisterVoid(router, subject.PresenceActivityPattern(cfg.SiteID), handler.Activity) @@ -147,6 +152,21 @@ func main() { sweeper.Run(sweepCtx) }() + // Bind synchronously so a port conflict fails startup loudly rather than + // running blind — /metrics exposes rpc_server_* RPC metrics. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + slog.Info("user-presence-service running", "site", cfg.SiteID, "valkey", cfg.Valkey.Addrs) shutdown.Wait(ctx, 25*time.Second, @@ -162,6 +182,7 @@ func main() { } }, func(ctx context.Context) error { return router.Shutdown(ctx) }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(_ context.Context) error { return store.Close() }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil }, diff --git a/user-service/config/config.go b/user-service/config/config.go index d5ed2d069..d0029ab9d 100644 --- a/user-service/config/config.go +++ b/user-service/config/config.go @@ -32,6 +32,7 @@ type Config struct { DefaultAppsLimit int `env:"APPS_DEFAULT_LIMIT" envDefault:"20"` MaxAccountNames int `env:"MAX_ACCOUNT_NAMES" envDefault:"100"` HandlerTimeout time.Duration `env:"HANDLER_TIMEOUT" envDefault:"15s"` + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` Mongo MongoConfig `envPrefix:"MONGO_"` NATS NATSConfig `envPrefix:"NATS_"` } diff --git a/user-service/main.go b/user-service/main.go index 57f6cf655..f3490402c 100644 --- a/user-service/main.go +++ b/user-service/main.go @@ -2,7 +2,10 @@ package main import ( "context" + "errors" "log/slog" + "net" + "net/http" "os" "time" @@ -98,6 +101,7 @@ func main() { // RequestID must precede any handler that reads request_id from ctx — // otherwise Classify's log line records an empty value. router.Use(natsrouter.RequestID()) + router.Use(natsrouter.Metrics("user-service")) router.Use(natsrouter.Logging()) // After Logging so the timeout wraps the handler chain; bounds the Mongo // aggregations from hanging past the configured deadline. @@ -105,10 +109,26 @@ func main() { svc.RegisterHandlers(router) + // Bind synchronously so a port conflict fails startup loudly rather than + // running blind — /metrics exposes rpc_server_* RPC metrics. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + slog.Info("user-service running", "site", cfg.SiteID) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return router.Shutdown(ctx) }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil }, From 06b17633d18ea4d251c1c6db6eb3917b9c87a365 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 15:19:10 +0000 Subject: [PATCH 10/12] feat(rpc-metrics): wire ginutil.Metrics into all HTTP services Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- auth-service/main.go | 30 ++++++++++++++++++++++++++---- media-service/config.go | 2 ++ media-service/main.go | 28 ++++++++++++++++++++++++---- portal-service/main.go | 38 ++++++++++++++++++++++++++++++-------- upload-service/main.go | 21 +++++++++++++++++++++ 5 files changed, 103 insertions(+), 16 deletions(-) diff --git a/auth-service/main.go b/auth-service/main.go index 75c77704c..dda6ddb6c 100644 --- a/auth-service/main.go +++ b/auth-service/main.go @@ -2,8 +2,10 @@ package main import ( "context" + "errors" "fmt" "log/slog" + "net" "net/http" "os" "time" @@ -14,6 +16,7 @@ import ( "github.com/hmchangw/chat/pkg/ginutil" pkgoidc "github.com/hmchangw/chat/pkg/oidc" + "github.com/hmchangw/chat/pkg/otelutil" "github.com/hmchangw/chat/pkg/shutdown" ) @@ -24,6 +27,7 @@ type config struct { AuthAccountPubKey string `env:"AUTH_ACCOUNT_PUB_KEY,required"` NATSJWTExpiry time.Duration `env:"NATS_JWT_EXPIRY" envDefault:"2h"` NATSJWTExpiryJitter float64 `env:"NATS_JWT_EXPIRY_JITTER" envDefault:"0.1"` + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` // OIDC settings — required when DEV_MODE is false. OIDCIssuerURL string `env:"OIDC_ISSUER_URL"` @@ -88,10 +92,25 @@ func run() error { r := gin.New() r.Use(gin.Recovery()) r.Use(ginutil.RequestID()) + r.Use(ginutil.Metrics("auth-service")) r.Use(ginutil.AccessLog()) r.Use(ginutil.CORS()) registerRoutes(r, handler) + // /metrics on a separate port so scrapes don't hit the public API listener. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ Addr: addr, @@ -109,10 +128,13 @@ func run() error { shutdownDone := make(chan struct{}) go func() { defer close(shutdownDone) - shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { - slog.Info("shutting down auth service") - return srv.Shutdown(ctx) - }) + shutdown.Wait(ctx, 25*time.Second, + func(ctx context.Context) error { + slog.Info("shutting down auth service") + return srv.Shutdown(ctx) + }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + ) }() err = <-srvErr diff --git a/media-service/config.go b/media-service/config.go index cc2de65bb..954fd0a5c 100644 --- a/media-service/config.go +++ b/media-service/config.go @@ -72,6 +72,8 @@ type config struct { // sized to the employee population so the cache does not evict. EIDCacheTTL time.Duration `env:"EID_CACHE_TTL" envDefault:"24h"` EIDCacheCapacity int `env:"EID_CACHE_CAPACITY" envDefault:"120000"` + + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` } // clusterBaseURL returns the configured base URL for a site, or "" if unknown. diff --git a/media-service/main.go b/media-service/main.go index 67b8b0e4d..54ca92739 100644 --- a/media-service/main.go +++ b/media-service/main.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" "log/slog" + "net" "net/http" "os" "time" @@ -12,8 +13,10 @@ import ( "github.com/caarlos0/env/v11" "github.com/gin-gonic/gin" + "github.com/hmchangw/chat/pkg/ginutil" "github.com/hmchangw/chat/pkg/minioutil" "github.com/hmchangw/chat/pkg/mongoutil" + "github.com/hmchangw/chat/pkg/otelutil" "github.com/hmchangw/chat/pkg/shutdown" ) @@ -58,10 +61,24 @@ func run() error { r := gin.New() r.Use(gin.Recovery()) r.Use(requestIDMiddleware()) + r.Use(ginutil.Metrics("media-service")) r.Use(accessLogMiddleware()) r.Use(corsMiddleware()) registerRoutes(r, h) + // /metrics on a separate port so scrapes don't hit the public API listener. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + return fmt.Errorf("metrics listen: %w", err) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + srv := &http.Server{ Addr: fmt.Sprintf(":%s", cfg.Port), Handler: r, @@ -69,10 +86,13 @@ func run() error { WriteTimeout: 30 * time.Second, } - go shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { - slog.Info("shutting down media-service") - return srv.Shutdown(ctx) - }) + go shutdown.Wait(ctx, 25*time.Second, + func(ctx context.Context) error { + slog.Info("shutting down media-service") + return srv.Shutdown(ctx) + }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + ) slog.Info("media-service listening", "port", cfg.Port, "site", cfg.SiteID) if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { diff --git a/portal-service/main.go b/portal-service/main.go index 87b95694f..055034c71 100644 --- a/portal-service/main.go +++ b/portal-service/main.go @@ -2,8 +2,10 @@ package main import ( "context" + "errors" "fmt" "log/slog" + "net" "net/http" "os" "sync" @@ -14,6 +16,7 @@ import ( "github.com/hmchangw/chat/pkg/ginutil" "github.com/hmchangw/chat/pkg/mongoutil" + "github.com/hmchangw/chat/pkg/otelutil" "github.com/hmchangw/chat/pkg/shutdown" ) @@ -47,6 +50,8 @@ type config struct { MongoDB string `env:"MONGO_DB" envDefault:"portal"` MongoUsername string `env:"MONGO_USERNAME" envDefault:""` MongoPassword string `env:"MONGO_PASSWORD" envDefault:""` + + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` } func main() { @@ -110,10 +115,24 @@ func run() error { r := gin.New() r.Use(gin.Recovery()) r.Use(ginutil.RequestID()) + r.Use(ginutil.Metrics("portal-service")) r.Use(ginutil.AccessLog()) r.Use(ginutil.CORS()) registerRoutes(r, handler) + // /metrics on a separate port so scrapes don't hit the public API listener. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + return fmt.Errorf("metrics listen: %w", err) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ Addr: addr, @@ -131,14 +150,17 @@ func run() error { shutdownDone := make(chan struct{}) go func() { defer close(shutdownDone) - shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { - slog.Info("shutting down portal service") - err := srv.Shutdown(ctx) - refreshCancel() - refreshWG.Wait() - mongoutil.Disconnect(ctx, mongoClient) - return err - }) + shutdown.Wait(ctx, 25*time.Second, + func(ctx context.Context) error { + slog.Info("shutting down portal service") + err := srv.Shutdown(ctx) + refreshCancel() + refreshWG.Wait() + mongoutil.Disconnect(ctx, mongoClient) + return err + }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + ) }() err = <-srvErr diff --git a/upload-service/main.go b/upload-service/main.go index 024ff5459..2afb7df4e 100644 --- a/upload-service/main.go +++ b/upload-service/main.go @@ -2,8 +2,10 @@ package main import ( "context" + "errors" "fmt" "log/slog" + "net" "net/http" "os" "time" @@ -12,6 +14,7 @@ import ( "github.com/gin-gonic/gin" "github.com/hmchangw/chat/pkg/drive" + "github.com/hmchangw/chat/pkg/ginutil" "github.com/hmchangw/chat/pkg/minioutil" "github.com/hmchangw/chat/pkg/mongoutil" pkgoidc "github.com/hmchangw/chat/pkg/oidc" @@ -62,6 +65,8 @@ type config struct { MinioDownloadTimeout time.Duration `env:"MINIO_DOWNLOAD_TIMEOUT" envDefault:"5m"` Drive drive.Config `envPrefix:"DRIVE_"` + + MetricsAddr string `env:"METRICS_ADDR" envDefault:":9090"` } func main() { @@ -127,10 +132,25 @@ func run() error { r := gin.New() r.Use(gin.Recovery()) r.Use(requestIDMiddleware()) + r.Use(ginutil.Metrics("upload-service")) r.Use(accessLogMiddleware()) r.Use(corsMiddleware(cfg.CORSAllowedOrigins)) registerRoutes(r, handler, validator, cfg.DevMode) + // /metrics on a separate port so scrapes don't hit the public API listener. + metricsServer := otelutil.MetricsServer() + metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + if err != nil { + slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + os.Exit(1) + } + go func() { + slog.Info("metrics server listening", "addr", cfg.MetricsAddr) + if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ Addr: addr, @@ -150,6 +170,7 @@ func run() error { defer close(shutdownDone) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return srv.Shutdown(ctx) }, + func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil }, ) From c2ba9eb84c9be1c1358c6845f1d39a801013baa9 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 15:36:35 +0000 Subject: [PATCH 11/12] fix(rpc-metrics): normalize HTTP status via allowlist; drop testutil from prod; unify listen-fail handling Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- auth-service/main.go | 3 +-- pkg/ginutil/metrics.go | 2 +- pkg/ginutil/metrics_test.go | 21 +++++++++++++++++++++ pkg/rpcmetrics/metrics.go | 30 ++++++++++++++++++++++-------- pkg/rpcmetrics/metrics_test.go | 19 +++++++++++++++++++ upload-service/main.go | 3 +-- user-presence-service/main.go | 4 ++++ 7 files changed, 69 insertions(+), 13 deletions(-) diff --git a/auth-service/main.go b/auth-service/main.go index dda6ddb6c..de3b82ebb 100644 --- a/auth-service/main.go +++ b/auth-service/main.go @@ -101,8 +101,7 @@ func run() error { metricsServer := otelutil.MetricsServer() metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) if err != nil { - slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) - os.Exit(1) + return fmt.Errorf("metrics listen: %w", err) } go func() { slog.Info("metrics server listening", "addr", cfg.MetricsAddr) diff --git a/pkg/ginutil/metrics.go b/pkg/ginutil/metrics.go index d0ca383e3..a2a19117f 100644 --- a/pkg/ginutil/metrics.go +++ b/pkg/ginutil/metrics.go @@ -37,7 +37,7 @@ func Metrics(service string) gin.HandlerFunc { func statusForHTTP(c *gin.Context) string { if v, ok := c.Get(errCodeKey); ok { if code, ok := v.(string); ok && code != "" { - return code + return rpcmetrics.NormalizeStatus(code) } } switch code := c.Writer.Status(); { diff --git a/pkg/ginutil/metrics_test.go b/pkg/ginutil/metrics_test.go index 38bb71a09..c55d378f4 100644 --- a/pkg/ginutil/metrics_test.go +++ b/pkg/ginutil/metrics_test.go @@ -5,9 +5,11 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/gin-gonic/gin" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/hmchangw/chat/pkg/errcode" "github.com/hmchangw/chat/pkg/errcode/errhttp" @@ -50,6 +52,25 @@ func TestGinMetrics_HTTPClassFallback(t *testing.T) { assert.Equal(t, before+1, after) } +func TestGinMetrics_NonCanonicalErrcodeNormalizesToInternal(t *testing.T) { + r := newEngine() + r.GET("/api/normalize", func(c *gin.Context) { + c.Set("errcode", "weird_code") + c.Status(http.StatusOK) + }) + + before := rpcmetrics.CounterValue("auth-service", "/api/normalize", "internal") + beforeWeird := rpcmetrics.CounterValue("auth-service", "/api/normalize", "weird_code") + + do(r, http.MethodGet, "/api/normalize") + + require.Eventually(t, func() bool { + return rpcmetrics.CounterValue("auth-service", "/api/normalize", "internal") == before+1 + }, time.Second, 10*time.Millisecond, "non-canonical errcode status should normalize to internal") + assert.Equal(t, beforeWeird, rpcmetrics.CounterValue("auth-service", "/api/normalize", "weird_code"), + "raw non-canonical status label must never be recorded") +} + func TestGinMetrics_SkipsHealthzAndUnmatched(t *testing.T) { r := newEngine() r.GET("/healthz", func(c *gin.Context) { c.Status(http.StatusOK) }) diff --git a/pkg/rpcmetrics/metrics.go b/pkg/rpcmetrics/metrics.go index cba526674..bbb056159 100644 --- a/pkg/rpcmetrics/metrics.go +++ b/pkg/rpcmetrics/metrics.go @@ -9,7 +9,7 @@ import ( "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" - "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" "github.com/hmchangw/chat/pkg/errcode" ) @@ -47,10 +47,19 @@ func StatusLabel(err error) string { return "ok" } var ee *errcode.Error - if errors.As(err, &ee) && ee.Code != "" { - if _, ok := allowedStatusLabels[string(ee.Code)]; ok { - return string(ee.Code) - } + if errors.As(err, &ee) { + return NormalizeStatus(string(ee.Code)) + } + return string(errcode.CodeInternal) +} + +// NormalizeStatus collapses any status label outside the pinned allowlist to +// "internal", bounding cardinality. Both transports (NATS StatusLabel and the +// HTTP middleware fallback) funnel through this so the status taxonomy is +// identical on each. +func NormalizeStatus(code string) string { + if _, ok := allowedStatusLabels[code]; ok { + return code } return string(errcode.CodeInternal) } @@ -72,8 +81,13 @@ var allowedStatusLabels = map[string]struct{}{ } // CounterValue returns the current rpc_server_requests_total value for the -// given label tuple. It is a test seam for consumer packages (natsrouter, -// ginutil) that cannot reach the unexported collector; side-effect-free. +// given label tuple. Test seam for consumer packages (natsrouter, ginutil); +// side-effect-free. Implemented via client_model to avoid importing test-only +// prometheus/testutil into production binaries. func CounterValue(service, route, status string) float64 { - return testutil.ToFloat64(requestsTotal.WithLabelValues(service, route, status)) + var m dto.Metric + if err := requestsTotal.WithLabelValues(service, route, status).Write(&m); err != nil { + return 0 + } + return m.GetCounter().GetValue() } diff --git a/pkg/rpcmetrics/metrics_test.go b/pkg/rpcmetrics/metrics_test.go index 60e55ae07..2f72c4de5 100644 --- a/pkg/rpcmetrics/metrics_test.go +++ b/pkg/rpcmetrics/metrics_test.go @@ -37,6 +37,25 @@ func TestStatusLabel(t *testing.T) { } } +func TestNormalizeStatus(t *testing.T) { + tests := []struct { + name string + code string + want string + }{ + {"ok passes through", "ok", "ok"}, + {"not_found passes through", "not_found", "not_found"}, + {"internal passes through", "internal", "internal"}, + {"non-canonical collapses to internal", "weird_code", "internal"}, + {"empty collapses to internal", "", "internal"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, NormalizeStatus(tt.code)) + }) + } +} + func TestObserve(t *testing.T) { Observe("svc-test", "chat.route.{id}.get", "ok", 12*time.Millisecond) diff --git a/upload-service/main.go b/upload-service/main.go index 2afb7df4e..5d572419a 100644 --- a/upload-service/main.go +++ b/upload-service/main.go @@ -141,8 +141,7 @@ func run() error { metricsServer := otelutil.MetricsServer() metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) if err != nil { - slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) - os.Exit(1) + return fmt.Errorf("metrics listen: %w", err) } go func() { slog.Info("metrics server listening", "addr", cfg.MetricsAddr) diff --git a/user-presence-service/main.go b/user-presence-service/main.go index 81e5596dd..d2bcc0813 100644 --- a/user-presence-service/main.go +++ b/user-presence-service/main.go @@ -135,6 +135,10 @@ func main() { handler := NewHandler(store, userDir, peer, publish, cfg.SiteID, cfg.Presence.BatchMax) router := natsrouter.Default(nc, "user-presence-service") + // Installed after natsrouter.Default's chain, so this Metrics window excludes + // Default's RequestID/Recovery/Logging — its latency series is marginally + // narrower than peers that install Metrics before Logging. Status stamping is + // unaffected (the Register wrapper runs inside this Metrics c.Next()). router.Use(natsrouter.Metrics("user-presence-service")) natsrouter.RegisterVoid(router, subject.PresenceHelloPattern(cfg.SiteID), handler.Hello) natsrouter.RegisterVoid(router, subject.PresencePingPattern(cfg.SiteID), handler.Ping) From e5f36d3860c83ffd5ab244dfcdf2140b379b649a Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 8 Jul 2026 23:14:39 +0000 Subject: [PATCH 12/12] refactor(rpc-metrics): extract otelutil.ServeMetrics; reuse errcode helpers; dedup reply path Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Re53EQYKpXS3m9tBkRhbVb --- auth-service/main.go | 15 +++------------ media-service/main.go | 14 +++----------- pkg/ginutil/metrics.go | 7 ++----- pkg/natsrouter/register.go | 30 ++++++++++++++---------------- pkg/otelutil/otel.go | 22 ++++++++++++++++++++++ pkg/rpcmetrics/metrics.go | 23 ++++------------------- portal-service/main.go | 15 +++------------ room-service/main.go | 18 +++--------------- search-service/handler.go | 8 ++++---- upload-service/main.go | 15 +++------------ user-presence-service/main.go | 18 +++--------------- user-service/main.go | 18 +++--------------- 12 files changed, 67 insertions(+), 136 deletions(-) diff --git a/auth-service/main.go b/auth-service/main.go index de3b82ebb..7afdce035 100644 --- a/auth-service/main.go +++ b/auth-service/main.go @@ -2,10 +2,8 @@ package main import ( "context" - "errors" "fmt" "log/slog" - "net" "net/http" "os" "time" @@ -98,17 +96,10 @@ func run() error { registerRoutes(r, handler) // /metrics on a separate port so scrapes don't hit the public API listener. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - return fmt.Errorf("metrics listen: %w", err) + return err } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ @@ -132,7 +123,7 @@ func run() error { slog.Info("shutting down auth service") return srv.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, ) }() diff --git a/media-service/main.go b/media-service/main.go index 54ca92739..5e92085e2 100644 --- a/media-service/main.go +++ b/media-service/main.go @@ -5,7 +5,6 @@ import ( "errors" "fmt" "log/slog" - "net" "net/http" "os" "time" @@ -67,17 +66,10 @@ func run() error { registerRoutes(r, h) // /metrics on a separate port so scrapes don't hit the public API listener. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - return fmt.Errorf("metrics listen: %w", err) + return err } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() srv := &http.Server{ Addr: fmt.Sprintf(":%s", cfg.Port), @@ -91,7 +83,7 @@ func run() error { slog.Info("shutting down media-service") return srv.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, ) slog.Info("media-service listening", "port", cfg.Port, "site", cfg.SiteID) diff --git a/pkg/ginutil/metrics.go b/pkg/ginutil/metrics.go index a2a19117f..385778bfc 100644 --- a/pkg/ginutil/metrics.go +++ b/pkg/ginutil/metrics.go @@ -5,13 +5,10 @@ import ( "github.com/gin-gonic/gin" + "github.com/hmchangw/chat/pkg/errcode/errhttp" "github.com/hmchangw/chat/pkg/rpcmetrics" ) -// errCodeKey mirrors errhttp.ErrCodeKey by value to avoid an import edge from -// ginutil to errhttp. errhttp.Write stores the classified Code here. -const errCodeKey = "errcode" - // Metrics returns middleware that records rpc_server_requests_total and // rpc_server_request_duration_seconds for each matched HTTP route. The `route` // label is the registered template (c.FullPath()), never the live URL. The @@ -35,7 +32,7 @@ func Metrics(service string) gin.HandlerFunc { // derives a coarse class from the HTTP status code so plain responses still map // onto the shared status taxonomy. func statusForHTTP(c *gin.Context) string { - if v, ok := c.Get(errCodeKey); ok { + if v, ok := c.Get(errhttp.ErrCodeKey); ok { if code, ok := v.(string); ok && code != "" { return rpcmetrics.NormalizeStatus(code) } diff --git a/pkg/natsrouter/register.go b/pkg/natsrouter/register.go index 77262c355..0f9aa43c0 100644 --- a/pkg/natsrouter/register.go +++ b/pkg/natsrouter/register.go @@ -29,14 +29,7 @@ func Register[Req, Resp any]( } resp, err := fn(c, req) - if err != nil { - c.SetStatus(rpcmetrics.StatusLabel(err)) - replyErr(c, err) - return - } - - c.SetStatus("ok") - c.ReplyJSON(resp) + replyResult(c, resp, err) }) r.addRoute(pattern, []HandlerFunc{handler}) @@ -50,14 +43,7 @@ func RegisterNoBody[Resp any]( ) { handler := HandlerFunc(func(c *Context) { resp, err := fn(c) - if err != nil { - c.SetStatus(rpcmetrics.StatusLabel(err)) - replyErr(c, err) - return - } - - c.SetStatus("ok") - c.ReplyJSON(resp) + replyResult(c, resp, err) }) r.addRoute(pattern, []HandlerFunc{handler}) @@ -93,3 +79,15 @@ func RegisterVoid[Req any]( func replyErr(c *Context, err error) { errnats.Reply(c, c.Msg, err) } + +// replyResult stamps the terminal metrics status and sends the reply: the +// classified status + error envelope on failure, "ok" + JSON body on success. +func replyResult[Resp any](c *Context, resp *Resp, err error) { + if err != nil { + c.SetStatus(rpcmetrics.StatusLabel(err)) + replyErr(c, err) + return + } + c.SetStatus("ok") + c.ReplyJSON(resp) +} diff --git a/pkg/otelutil/otel.go b/pkg/otelutil/otel.go index 2f2521806..d71d5b443 100644 --- a/pkg/otelutil/otel.go +++ b/pkg/otelutil/otel.go @@ -2,7 +2,10 @@ package otelutil import ( "context" + "errors" "fmt" + "log/slog" + "net" "net/http" "os" "time" @@ -80,3 +83,22 @@ func MetricsServer() *http.Server { IdleTimeout: 60 * time.Second, } } + +// ServeMetrics binds a /metrics server on addr and serves it in a background +// goroutine. It binds synchronously, so a port conflict surfaces as a returned +// error at startup rather than in a goroutine. The returned stop func shuts the +// server down; wire it into the service's graceful-shutdown sequence. +func ServeMetrics(addr string) (func(context.Context) error, error) { + srv := MetricsServer() + ln, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("metrics listen on %s: %w", addr, err) + } + go func() { + slog.Info("metrics server listening", "addr", addr) + if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) { + slog.Error("metrics server failed", "error", err) + } + }() + return srv.Shutdown, nil +} diff --git a/pkg/rpcmetrics/metrics.go b/pkg/rpcmetrics/metrics.go index bbb056159..35a062134 100644 --- a/pkg/rpcmetrics/metrics.go +++ b/pkg/rpcmetrics/metrics.go @@ -53,33 +53,18 @@ func StatusLabel(err error) string { return string(errcode.CodeInternal) } -// NormalizeStatus collapses any status label outside the pinned allowlist to -// "internal", bounding cardinality. Both transports (NATS StatusLabel and the +// NormalizeStatus admits "ok" plus the canonical errcode Codes (via +// Code.Valid()), collapsing anything else to "internal". This bounds +// cardinality on the status label. Both transports (NATS StatusLabel and the // HTTP middleware fallback) funnel through this so the status taxonomy is // identical on each. func NormalizeStatus(code string) string { - if _, ok := allowedStatusLabels[code]; ok { + if code == "ok" || errcode.Code(code).Valid() { return code } return string(errcode.CodeInternal) } -// allowedStatusLabels pins the cardinality of the status label to the 8 -// canonical errcode Codes + "ok". Any label outside this set collapses to -// "internal" via StatusLabel, so a future Code added without updating this -// allowlist cannot mint a fresh time series. -var allowedStatusLabels = map[string]struct{}{ - "ok": {}, - string(errcode.CodeBadRequest): {}, - string(errcode.CodeUnauthenticated): {}, - string(errcode.CodeForbidden): {}, - string(errcode.CodeNotFound): {}, - string(errcode.CodeConflict): {}, - string(errcode.CodeTooManyRequests): {}, - string(errcode.CodeUnavailable): {}, - string(errcode.CodeInternal): {}, -} - // CounterValue returns the current rpc_server_requests_total value for the // given label tuple. Test seam for consumer packages (natsrouter, ginutil); // side-effect-free. Implemented via client_model to avoid importing test-only diff --git a/portal-service/main.go b/portal-service/main.go index 055034c71..3731f5cf6 100644 --- a/portal-service/main.go +++ b/portal-service/main.go @@ -2,10 +2,8 @@ package main import ( "context" - "errors" "fmt" "log/slog" - "net" "net/http" "os" "sync" @@ -121,17 +119,10 @@ func run() error { registerRoutes(r, handler) // /metrics on a separate port so scrapes don't hit the public API listener. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - return fmt.Errorf("metrics listen: %w", err) + return err } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ @@ -159,7 +150,7 @@ func run() error { mongoutil.Disconnect(ctx, mongoClient) return err }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, ) }() diff --git a/room-service/main.go b/room-service/main.go index 248f99456..1edfbe4be 100644 --- a/room-service/main.go +++ b/room-service/main.go @@ -2,11 +2,8 @@ package main import ( "context" - "errors" "fmt" "log/slog" - "net" - "net/http" "net/url" "os" "time" @@ -220,26 +217,17 @@ func main() { os.Exit(1) } - // Bind synchronously so a port conflict fails startup loudly rather than - // running blind — /metrics exposes rpc_server_* RPC metrics. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + slog.Error("metrics server setup failed", "error", err) os.Exit(1) } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() slog.Info("room-service running", "site", cfg.SiteID) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return router.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { diff --git a/search-service/handler.go b/search-service/handler.go index 291136f65..cbee8754c 100644 --- a/search-service/handler.go +++ b/search-service/handler.go @@ -70,7 +70,7 @@ func (h *handler) withRequestTimeout(parent context.Context) (context.Context, c return context.WithTimeout(parent, h.cfg.RequestTimeout) } -func (h *handler) searchMessages(c *natsrouter.Context, req model.SearchMessagesRequest) (resp *model.SearchMessagesResponse, err error) { +func (h *handler) searchMessages(c *natsrouter.Context, req model.SearchMessagesRequest) (*model.SearchMessagesResponse, error) { account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -123,7 +123,7 @@ func (h *handler) searchMessages(c *natsrouter.Context, req model.SearchMessages return &model.SearchMessagesResponse{Messages: messages, Total: total}, nil } -func (h *handler) searchRooms(c *natsrouter.Context, req model.SearchRoomsRequest) (resp *model.SearchRoomsResponse, err error) { +func (h *handler) searchRooms(c *natsrouter.Context, req model.SearchRoomsRequest) (*model.SearchRoomsResponse, error) { account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -210,7 +210,7 @@ func (h *handler) loadRestricted(ctx context.Context, account string) (map[strin return restricted, nil } -func (h *handler) searchApps(c *natsrouter.Context, req model.SearchAppsRequest) (resp *model.SearchAppsResponse, err error) { +func (h *handler) searchApps(c *natsrouter.Context, req model.SearchAppsRequest) (*model.SearchAppsResponse, error) { account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr @@ -244,7 +244,7 @@ func (h *handler) searchApps(c *natsrouter.Context, req model.SearchAppsRequest) // SearchUsersClient and returns a raw []model.SearchUser. The account // from the subject is used for logging and metrics only; scoping is // enforced entirely by the third-party endpoint. -func (h *handler) searchUsers(c *natsrouter.Context, req model.SearchUsersRequest) (resp *[]model.SearchUser, err error) { +func (h *handler) searchUsers(c *natsrouter.Context, req model.SearchUsersRequest) (*[]model.SearchUser, error) { account, rerr := c.Params.Require("account") if rerr != nil { return nil, rerr diff --git a/upload-service/main.go b/upload-service/main.go index 5d572419a..76b0a9d13 100644 --- a/upload-service/main.go +++ b/upload-service/main.go @@ -2,10 +2,8 @@ package main import ( "context" - "errors" "fmt" "log/slog" - "net" "net/http" "os" "time" @@ -138,17 +136,10 @@ func run() error { registerRoutes(r, handler, validator, cfg.DevMode) // /metrics on a separate port so scrapes don't hit the public API listener. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - return fmt.Errorf("metrics listen: %w", err) + return err } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() addr := fmt.Sprintf(":%s", cfg.Port) srv := &http.Server{ @@ -169,7 +160,7 @@ func run() error { defer close(shutdownDone) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return srv.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil }, ) diff --git a/user-presence-service/main.go b/user-presence-service/main.go index d2bcc0813..134b26a4b 100644 --- a/user-presence-service/main.go +++ b/user-presence-service/main.go @@ -2,11 +2,8 @@ package main import ( "context" - "errors" "fmt" "log/slog" - "net" - "net/http" "os" "time" @@ -156,20 +153,11 @@ func main() { sweeper.Run(sweepCtx) }() - // Bind synchronously so a port conflict fails startup loudly rather than - // running blind — /metrics exposes rpc_server_* RPC metrics. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + slog.Error("metrics server setup failed", "error", err) os.Exit(1) } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() slog.Info("user-presence-service running", "site", cfg.SiteID, "valkey", cfg.Valkey.Addrs) @@ -186,7 +174,7 @@ func main() { } }, func(ctx context.Context) error { return router.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(_ context.Context) error { return store.Close() }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil }, diff --git a/user-service/main.go b/user-service/main.go index f3490402c..d6e926958 100644 --- a/user-service/main.go +++ b/user-service/main.go @@ -2,10 +2,7 @@ package main import ( "context" - "errors" "log/slog" - "net" - "net/http" "os" "time" @@ -109,26 +106,17 @@ func main() { svc.RegisterHandlers(router) - // Bind synchronously so a port conflict fails startup loudly rather than - // running blind — /metrics exposes rpc_server_* RPC metrics. - metricsServer := otelutil.MetricsServer() - metricsLn, err := net.Listen("tcp", cfg.MetricsAddr) + stopMetrics, err := otelutil.ServeMetrics(cfg.MetricsAddr) if err != nil { - slog.Error("metrics listen failed", "addr", cfg.MetricsAddr, "error", err) + slog.Error("metrics server setup failed", "error", err) os.Exit(1) } - go func() { - slog.Info("metrics server listening", "addr", cfg.MetricsAddr) - if err := metricsServer.Serve(metricsLn); err != nil && !errors.Is(err, http.ErrServerClosed) { - slog.Error("metrics server failed", "error", err) - } - }() slog.Info("user-service running", "site", cfg.SiteID) shutdown.Wait(ctx, 25*time.Second, func(ctx context.Context) error { return router.Shutdown(ctx) }, - func(ctx context.Context) error { return metricsServer.Shutdown(ctx) }, + func(ctx context.Context) error { return stopMetrics(ctx) }, func(ctx context.Context) error { return nc.Drain() }, func(ctx context.Context) error { return tracerShutdown(ctx) }, func(ctx context.Context) error { mongoutil.Disconnect(ctx, mongoClient); return nil },