From 5c6d169139b9565e405772e53dd14777c9385cae Mon Sep 17 00:00:00 2001 From: Taite Lee <113070390+taitelee@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:28:32 +0000 Subject: [PATCH 1/5] feat(observability): time HTTP requests and query-path ClickHouse calls wavehouse_http_request_duration_seconds{route, method, status_class} times every request but the SSE streams, by route pattern, on every listener. wavehouse_clickhouse_query_duration_seconds{operation, result} times each statement /v1/query, the pipes and /v1/ops/query send, with result ok or the response's error code. First tier of #94. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019YVNGPfArHqjc3Xt6N9App --- AGENTS.md | 2 +- CHANGELOG.md | 2 + docs/src/content/docs/architecture.md | 1 + docs/src/content/docs/deployment.md | 23 +++ internal/api/metrics.go | 161 ++++++++++++++++++ internal/api/metrics_test.go | 225 ++++++++++++++++++++++++++ internal/api/pipes.go | 15 +- internal/api/query.go | 13 ++ internal/api/router.go | 1 + internal/api/structured_query.go | 5 + 10 files changed, 442 insertions(+), 6 deletions(-) create mode 100644 internal/api/metrics.go create mode 100644 internal/api/metrics_test.go diff --git a/AGENTS.md b/AGENTS.md index 4aaa4eff..14992434 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,7 +28,7 @@ One binary: Sixteen internal packages under `internal/` (plus `internal/testutil/` for shared test helpers), and five public ones under `pkg/` — `chsql/`, `pipes/`, `policy/`, `settings/` and `tenant/` — that a module outside this repository imports at a release tag to validate a settings directory with the server's own rules ([#834](https://github.com/Wave-RF/WaveHouse/issues/834)); they import nothing under `internal/` and none of the server's runtime dependencies (the `public-packages` depguard rule in `.golangci.yml`): -- **`api/`** — Chi HTTP router, JWT/JWKS middleware (from `auth/`), ingest/query/structured-query/SSE/schema/DLQ/pipes handlers; `ch_errors.go` (`writeCHError`) is the one mapping from a failed ClickHouse query to status, `code` and `retryable` (`writeCHWriteError` for a write pipe: never retryable, no `Retry-After`, since the write may have run) +- **`api/`** — Chi HTTP router, JWT/JWKS middleware (from `auth/`), ingest/query/structured-query/SSE/schema/DLQ/pipes handlers; `ch_errors.go` (`writeCHError`) is the one mapping from a failed ClickHouse query to status, `code` and `retryable` (`writeCHWriteError` for a write pipe: never retryable, no `Retry-After`, since the write may have run); `metrics.go` holds the API's request-latency and query-path ClickHouse histograms (`wavehouse_http_request_duration_seconds`, `wavehouse_clickhouse_query_duration_seconds`), labeled by route pattern, method, status class, operation and result, never by tenant or path - **`app/`** — the process wiring: `New` builds every component from the boot config and the settings directory (each one wired in one place — what it opens, what it loops, what it releases — with the settings registry handed to its wiring function whole, the injection point of the per-tenant registry of #583: store-keyed getters for the handlers, `perTenant` for the async paths (with the tenant each message's `mq.Topic` names for the stream hub and the ingest worker), the `chconn.Pools`, the per-tenant `discoveries` and the per-tenant keepalive wheels (`keepalives`: one `stream.Heartbeater` per served tenant, at its own `stream.keepalive_*`, turned by the `keepalive` component, [#597](https://github.com/Wave-RF/WaveHouse/issues/597)) reconciled from `AfterAdopt`, `gapWindows` handing the sweeper each tenant's own gap window (a rejected tenant's as its folder last had it, unbounded for one rejected since boot) and the `mq.max_bytes_gb` reconcile each served tenant's byte budget, and `defaultPolicy` for the one setting that still follows tenant `0`, a flat directory's ops-gate admin role; the auth verifiers are per tenant, reconfigured (rebuilt only on changed wiring) and pruned from `AfterAdopt`, the same hook's `Hub.Prune` ends the open streams of a tenant no longer served, and `wireCache`'s hook drops, through `LocalCache.Prune`, the cache version index of a tenant no longer served ([#262](https://github.com/Wave-RF/WaveHouse/issues/262))), `Run` drives the long-lived ones under one `errgroup` until the context is cancelled or one fails, `Close` releases them in reverse order. `New` wires only what the process's `roles` need (discovery, dedupe, auth verifiers, the hub bridge and keepalive per API process; the ingest worker per ingest process; the sweeper under its lease through `elected`, on the embedded MQ only); a process without `api` serves `api.NewOpsRouter` — probes, `/version`, metrics, and the settings reload behind the operator key alone. `cmd/wavehouse` and `tests/integration` both boot through it - **`auth/`** — JWT auth middleware: HMAC **or** JWKS verification with `alg` pinned to the active verifier, role extraction from a configurable claim path; always runs, never rejects (bad token → empty role + stashed reason). One verifier per tenant ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 9): `Authenticator` keys them by `tenant.ID` — the request store's `settings.Store.Tenant()`, through an injected `TenantSource`; `tenant.Default` on the tenant-exempt routes — built from each tenant's `auth` block by `Reconfigure`, dropped by `Prune` once the tenant stops being served (rejected or removed), released by `Close`; the operator key (`Config`) is boot-level and shared, while the HMAC secret is the tenant's own `auth.jwt_secret` (`Wiring`), rotated by a reload ([#529](https://github.com/Wave-RF/WaveHouse/issues/529)). A JWKS key set is fetched off the boot and reload paths: until one has been stored the verifier is pending and a token-bearing request gets `503` + `Retry-After` from `api.refuseUnverifiable` (`auth.ErrVerifierPending`), never a `default_role` evaluation; refresh is library-managed (Eric, 2026-09-22), response capped at 1 MiB; the operator key's admin role is the request tenant's - **`cache/`** — `Cache` interface → `LocalCache` (Ristretto: one pool for every tenant) + `VersionManager` (the invalidation index), and `RedisCache`, the Redis-compatible shared backend (random version tokens under the tenant's hash tag, one-round-trip lookups, bypass on failure behind a circuit breaker, deferred invalidations retried; selected by `cache.backend: redis`, configured by the boot config's `cache.redis` block — [#613](https://github.com/Wave-RF/WaveHouse/issues/613)). Every key carries the tenant (in `RedisCache`, after the key prefix: `:{}:…` for a version token, `:q::…` for a value); in `LocalCache` and the version index it leads ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 8) — `:query:` for the caller's query key and its singleflight, escaped whole as the lead field of the stored key `|.||…`, where each raw table and scope name is escaped by `keyenc` (a `Namespace` carries them raw, so no caller escapes); the index holds a version per tenant, per (tenant, table) and per (tenant, table, scope), keyed by raw name and bumped in place (one entry per live namespace however often it is bumped, [#262](https://github.com/Wave-RF/WaveHouse/issues/262)) — so no cached read or coalesced flight crosses tenants, a bump through `Invalidate` names one tenant's namespaces and no other's, and `InvalidateTenant` drops the tenant's index so its next key gets a process-unique generation, orphaning its every cached result in one step, pipe results included (no insert reaches a pipe result until [#343](https://github.com/Wave-RF/WaveHouse/pull/343)); `Lookup` returns a `Snapshot` of the versions it read, taken before the handler chooses any input a bump invalidates — the tenant's connection included — and `Set` files the fill under it, so a write landing mid-query, or a reload moving the tenant to another address or database after the request took its connection, orphans the fill ([#382](https://github.com/Wave-RF/WaveHouse/issues/382)), and every backend runs the conformance suite `internal/testutil/cachetest`; the one crossing is the wiring's, above the package: `internal/app` hands the ingest worker the cache through `sharedTables`, which repeats each of the worker's bumps under every tenant on the same ClickHouse address and database (`chconn.Pools.SharingTables`, whatever their user or tls block — they read the same tables), and orphans the whole cache of a tenant back on a pool after an absence, since it was out of that fan-out while away, or moved to another address or database, since it now reads other tables (story 6) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0ff721ea..a3831980 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added +- **HTTP request latency and query-path ClickHouse calls are histograms** (`internal/api/metrics.go` (new, + `metrics_test.go`), `internal/api/{router,query,structured_query,pipes}.go`, `docs/src/content/docs/{deployment,architecture}.md`, `AGENTS.md`): the first tier of [#94](https://github.com/Wave-RF/WaveHouse/issues/94), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). The only request metrics were otelhttp's, which carry no route, so a slow or failing endpoint could not be told from the rest, and nothing timed what the query paths sent to ClickHouse. `wavehouse_http_request_duration_seconds{route, method, status_class}` times every request on every listener but the SSE streams, under the route's pattern (`/v1/pipes/{name}`), never the path; `wavehouse_clickhouse_query_duration_seconds{operation, result}` times each statement `POST /v1/query`, a pipe (`pipe`, `pipe_write`) or `POST /v1/ops/query` sends, with `result` `ok` or the response's error code without its `clickhouse.` prefix, so a histogram's `_count` is both the rate and the error rate. A cache hit records no query, and requests coalesced onto one record it once. Neither carries a tenant or table label. The worker's `INSERT` timing is the next tier. + - **Dropped SSE frames say why, on which table and on which connection** (`internal/stream/{metrics,subscriber,hub}.go` (+ tests, `drop_test.go` new), `internal/api/stream.go` (+ `stream_drop_test.go` new), `docs/src/content/docs/{deployment,architecture,configuration}.md*`, `AGENTS.md`): closes [#424](https://github.com/Wave-RF/WaveHouse/issues/424), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). `wavehouse_sse_dropped_frames_total` counted only frames dropped to a full queue, labeled by `kind` alone, and nothing was logged, so a burst of drops could not be tied to a connection or told apart from clients going away. It now also counts the frames a stream left unwritten when it ended, and carries `reason` (`buffer_full`, `client_gone`, `stream_closed`) and `table` (empty for a table the tenant's schema does not have, since the client names it), which `prometheus.per_table_metrics: false` drops like every other table label; a query that sums the counter now also sees the two new reasons, so filter on `reason="buffer_full"` to keep the old meaning. New `wavehouse_sse_queue_peak_frames` gauge: the deepest any open stream's queue has been since it opened, so a fill between scrapes still shows. Every stream logs `SSE stream opened` and `SSE stream closed` (how it ended, how long it lasted, what it dropped), and a slow consumer logs `SSE stream dropping frames` at `WARN`, at most every 10 seconds per stream, with its queue's depth, peak and capacity; each line carries the connection's `request_id` and `tenant` from the request scope. No metric carries a tenant label. - **Settings reloads and the tenants they serve or reject are metrics** (`internal/settings/{stats,registry}.go` (+ tests), `internal/app/wire_settings_metrics.go` (new, + test), `internal/app/{app,wire}.go`, `docs/src/content/docs/{deployment,architecture}.md`, `settings-directory.mdx`, `AGENTS.md`): closes [#776](https://github.com/Wave-RF/WaveHouse/issues/776), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). Reload outcomes were log lines only, so a tenant rejected on one process of several had nothing to alert on but a log grep. Each process now exports `wavehouse_settings_reloads_total{trigger, result}` (`adopted`, `partial`, `rejected`), `wavehouse_settings_last_reload_successful`, `wavehouse_settings_last_reload_success_timestamp_seconds`, `wavehouse_settings_tenants{state}` (`served`, `rejected`), `wavehouse_settings_tenants_removed_total` and `wavehouse_settings_reconcile_failures_total{component}`, none with a tenant label ([#538](https://github.com/Wave-RF/WaveHouse/issues/538)). A reload used to count as adopted even when the work after it failed, since the `AfterAdopt` hooks return nothing: the ClickHouse pool hook only logged, and the dedupe and queue-budget hooks discarded their errors. Those three now report through `settings.Registry.Reconciled`, which makes the reload `partial`, and a failure while boot applies the settings counts against the boot reload. What `POST /v1/ops/settings/reload` returns is unchanged. In `internal/app/wire.go` the change is one call per hook, and `app.go` gains one call at the end of `New`, which open PRs [#712](https://github.com/Wave-RF/WaveHouse/pull/712) and [#770](https://github.com/Wave-RF/WaveHouse/pull/770) also edit. diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index a4051ec6..8ddce241 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -87,6 +87,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with `reques - **pipes.go** — Named query pipe handlers: admin listing (`GET /v1/ops/pipes[/{name}]`, read per request from its `pipes.Source`) and execution with parameter binding. A read is cached and coalesced; a write — bound SQL that `IsMutation` (`clickhouse_exec.go`) classifies as one — bypasses both and runs every call. `pipes.json` is the only way to define or change a pipe. - **structured_query.go** — Handler for `POST /v1/query?table={table}`: validates query AST, enforces permissions, builds and executes SQL. - **ch_errors.go** — `writeCHError`, the one mapping from a failed ClickHouse query to a response, shared by `/v1/query`, pipes and `/v1/ops/query` so they cannot drift apart: `chconn.Classify` decides the class, and the class the status, `code` and `retryable` ([ClickHouse errors on the query paths](/api#clickhouse-errors-on-the-query-paths)). A write pipe answers through `writeCHWriteError`, the same mapping with `retryable` always `false` and no `Retry-After`, since the write may have run. +- **metrics.go** — the API's own instruments (meter `wavehouse-api`, no tenant label): `wavehouse_http_request_duration_seconds` by `route` (the matched chi route pattern: a mount's own, such as `/v1/*`, for a request answered inside it short of a route, and `unmatched` for one that reached no mount), `method` (net/http's methods, `other` for any other) and `status_class`, recorded by a middleware every router mounts outside `jsonRecoverer`, so a recovered panic counts as its `500`; `/v1/stream` is not timed, since a stream's lifetime is `wavehouse_sse_stream_duration_seconds`. `wavehouse_clickhouse_query_duration_seconds` by `operation` (`query`, `pipe`, `pipe_write`, `ops_query`) and `result` (`ok`, or the response's error `code` without its `clickhouse.` prefix, through `chFailureOf`), one value per statement sent: a cache hit records none and singleflight-coalesced callers record one. Nil-safe, so a handler built without its constructor records nothing. - **ingest.go** — Accepts `POST /v1/ingest?table={table}` in three body shapes: one flat JSON object, a JSON array of them, or NDJSON. The **required** `Content-Type` chooses the format *family* — `application/json` versus the four NDJSON spellings — and within the JSON family the body's first non-whitespace byte picks array versus single object; the bytes never choose the family. Anything that is not exactly one readable media type is a `415`, decided before the body is read: the header is parsed per RFC 9110 §8.3, and because `Content-Type` is a singleton field, repeated header lines must all resolve to the same format and a value carrying a comma is refused unless the value as a whole parses as one media type — a comma inside a *quoted* parameter value is data, so `application/json; a=", application/x-ndjson; b="` is accepted. It then reads the whole (`MaxBytesReader`-capped) body into a pooled buffer and runs the per-format record readers over those bytes, so the `413` lands before any record is processed and peak memory per request is O(body) rather than O(record). Then it validates and encodes each record, and runs the records in windows of up to 256 (`ingestWindow`) through three phases: one dedupe `Reserve` for the window's ids, the publishes in record order (a deduped record under `mq.WithIdempotencyKey`, keyed by `dedupe.IdempotencyKey`), and one `Commit` of the published ids — a window is the unit of a dedupe round trip and of Pebble's commit `fsync`. An id another request holds answers `503` with the lease as `Retry-After`, a store that cannot answer (`dedupe.ErrUnavailable`) `503` with `Retry-After: 5`; a publish that fails at a record commits the ones before it and releases the rest, except that a failure other than `mq.ErrQueueFull` may have stored the event, so that record's claim is left to lapse and the idempotency key drops the retry's copy if it comes within the stream's two-minute duplicate window — `mq.ErrUnavailable` (a broker blip) is one such failure, and still answers `503`: with the lease, rounded up to whole seconds, as `Retry-After` when the failing record held a claim left to lapse, else the flat `Retry-After: 5`. Each row goes through `mq.Publisher` on `mq.Topic{Tenant, Table, Scope}` (the request's tenant, read off its resolved store — `store.Tenant()` — and raw names; the subject it becomes is `internal/mq`'s; a full queue comes back as `mq.ErrQueueFull`, which is the `503` + `Retry-After`). When dedup is on, a row missing the configured `id_field` (or setting it to `null`) can't be deduped: it is logged at `WARN` and counted by `wavehouse_ingest_dedupe_missing_id_total` (labeled by `table`), then published un-deduped — or rejected when `dedupe.require_id` is set ([#219](https://github.com/Wave-RF/WaveHouse/issues/219)). - **query.go** — Proxies raw SQL for `POST /v1/ops/query` straight to the `?tenant=`'s ClickHouse HTTP interface (`chconn.Pools.Target` by the resolved store's tenant; the zero target — no pool — is a `503` with `Retry-After`). **Not cached** — sets `Cache-Control: no-store` so every request hits ClickHouse; DateTime is rendered ISO-8601 via `date_time_output_format=iso` (the Go-side type conversion lives in the structured-query / pipes path, not here). - **stream.go** — Real-time streaming via SSE. Callers select a table with the `?table=` query parameter. Each connection registers one `Subscriber` (the `stream/` package) with both the event `Hub` (under its `(topic, role)`) and its tenant's keepalive wheel (`Heartbeater`, a lookup by the request's store; a tenant no longer served has none, and its streams are ending), then drains both from a single byte-pump — so idle streams keep emitting `:` keepalive comments (surviving reverse-proxy idle timeouts) while live events arrive already projected and serialized. Per-event projection/serialization happens **once per role** in the `Hub`, not once per subscriber ([#294](https://github.com/Wave-RF/WaveHouse/issues/294)); the handler also snapshots the connection's JWT claims onto the `Subscriber`, which the `Hub` evaluates per subscriber when the role carries a row-level `filter` ([#319](https://github.com/Wave-RF/WaveHouse/issues/319)). Gap-fill replay (`mq.Replayer.ReplaySince` on the connection's `mq.Topic` — a `DeliverByStartTime` consumer inside `internal/mq`) stays per-connection (low-volume, one-time on connect). A stream ends, a gap-fill in progress included, when the server begins shutting down (`Closing`) or its `Subscriber` is evicted because its tenant is no longer served (`Hub.Prune`); one admitted just before the reload that stopped serving its tenant, and registered just after the prune, is ended right after it registers (`Served`). Each stream logs its open and its close (why it ended, how long it lasted, what it dropped), and a stream dropping frames to a full queue logs a rate-limited `WARN` with its queue state, all through the request's context, so each line carries the connection's `request_id` and `tenant`; on close the handler counts a frame it failed to write, and what is still queued, as dropped (`client_gone` or `stream_closed`). diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index a7b14540..c3325330 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -833,6 +833,29 @@ Which build each process runs is the `wavehouse_build_info` gauge, always `1`, l count by (version) (wavehouse_build_info) ``` +### Request latency and ClickHouse queries + +Every process times its HTTP requests, and an API process times each statement its query paths send to ClickHouse. Both are histograms in seconds, with buckets from 5 ms to 30 s, and neither carries a tenant label: + +| Metric | Labels | +| --- | --- | +| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`, or a tenant that is not served) is labeled with the prefix's pattern, such as `/v1/*`, and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | +| `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | + +The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`, and the queries ClickHouse could not take: + +```promql +sum by (route) (rate(wavehouse_http_request_duration_seconds_count{status_class="5xx"}[5m])) + / sum by (route) (rate(wavehouse_http_request_duration_seconds_count[5m])) +sum by (operation) (rate(wavehouse_clickhouse_query_duration_seconds_count{result="unavailable"}[5m])) > 0 +``` + +The 99th percentile of each route's latency: + +```promql +histogram_quantile(0.99, sum by (route, le) (rate(wavehouse_http_request_duration_seconds_bucket[5m]))) +``` + ### Settings reloads Every process reads the settings directory and reloads it on its own, so a reload can be adopted by one pod and rejected, or only half applied, by the next. These metrics report each process's reloads; none carries a tenant label, and which tenant was rejected is in the reload's log line: diff --git a/internal/api/metrics.go b/internal/api/metrics.go new file mode 100644 index 00000000..df505f17 --- /dev/null +++ b/internal/api/metrics.go @@ -0,0 +1,161 @@ +package api + +import ( + "context" + "net/http" + "strings" + "time" + + "github.com/go-chi/chi/v5" + "github.com/go-chi/chi/v5/middleware" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +// latencyBuckets are the boundaries, in seconds, of the request and query +// latency histograms: Prometheus' default buckets, plus 30s for a query that +// runs up to a long query_timeout. +var latencyBuckets = []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, 30} + +// routeUnmatched is the route label of a request that reached no route or +// mount: an unknown top-level path, a method chi does not route, or a CORS +// preflight answered before routing. One answered inside a mount short of a +// route (an unknown path under /v1, a tenant not served) carries the mount's +// pattern, /v1/*. +const routeUnmatched = "unmatched" + +// httpMetrics records each request's latency by route, method and status +// class. The zero value (a nil *httpMetrics) is a no-op. +type httpMetrics struct { + duration metric.Float64Histogram +} + +// meterName is the meter the api package's instruments are built on. +const meterName = "wavehouse-api" + +// newHTTPMetrics builds the HTTP instruments on the global meter provider. +// Call it after observability.InitProvider, like every other instrument. +func newHTTPMetrics() *httpMetrics { return httpMetricsOn(otel.Meter(meterName)) } + +// httpMetricsOn builds the HTTP instruments on meter. +func httpMetricsOn(meter metric.Meter) *httpMetrics { + duration, _ := meter.Float64Histogram("wavehouse_http_request_duration_seconds", + metric.WithDescription("HTTP request latency by route (the matched route pattern), method and status_class (2xx, 3xx, 4xx, 5xx); SSE streams on /v1/stream are not counted"), + metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(latencyBuckets...)) + return &httpMetrics{duration: duration} +} + +// middleware times every request but the SSE streams, whose lifetime is +// wavehouse_sse_stream_duration_seconds, not a latency. It runs outside +// jsonRecoverer, so a recovered panic counts as the 500 it answers. +func (m *httpMetrics) middleware(next http.Handler) http.Handler { + if m == nil { + return next + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasPrefix(r.URL.Path, "/v1/stream") { + next.ServeHTTP(w, r) + return + } + start := time.Now() + ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor) + next.ServeHTTP(ww, r) + // The route pattern is complete only once routing has run, so it is + // read after the handler, never before. + route := routeUnmatched + if rctx := chi.RouteContext(r.Context()); rctx != nil { + if p := rctx.RoutePattern(); p != "" { + route = p + } + } + m.duration.Record(context.Background(), time.Since(start).Seconds(), metric.WithAttributes( + attribute.String("route", route), + attribute.String("method", methodLabel(r.Method)), + attribute.String("status_class", statusClass(ww.Status())), + )) + }) +} + +// methodLabel is method when it is one of net/http's, and "other" otherwise: +// the method is the client's to choose, and a label must stay a closed set. +func methodLabel(method string) string { + switch method { + case http.MethodGet, http.MethodHead, http.MethodPost, http.MethodPut, http.MethodPatch, + http.MethodDelete, http.MethodConnect, http.MethodOptions, http.MethodTrace: + return method + } + return "other" +} + +// statusClass is status's class, "2xx" for a handler that wrote nothing +// (net/http answers it 200). +func statusClass(status int) string { + switch { + case status == 0 || status >= 200 && status < 300: + return "2xx" + case status < 200: + return "1xx" + case status < 400: + return "3xx" + case status < 500: + return "4xx" + } + return "5xx" +} + +// Operations for the "operation" attribute on +// wavehouse_clickhouse_query_duration_seconds: which query path ran the +// statement. +const ( + opStructuredQuery = "query" // POST /v1/query + opPipe = "pipe" // a pipe that reads, /v1/pipes/{name} + opPipeWrite = "pipe_write" // a pipe that writes, /v1/pipes/{name} + opOpsQuery = "ops_query" // raw SQL, POST /v1/ops/query +) + +// resultOK is the "result" of a query ClickHouse answered. +const resultOK = "ok" + +// queryMetrics records each ClickHouse call the query paths make: one per +// statement sent, so a cache hit records none and callers coalesced onto one +// call record it once. The zero value (a nil *queryMetrics) is a no-op. +type queryMetrics struct { + duration metric.Float64Histogram +} + +// newQueryMetrics builds the query-path instruments on the global meter +// provider. Call it after observability.InitProvider. +func newQueryMetrics() *queryMetrics { return queryMetricsOn(otel.Meter(meterName)) } + +// queryMetricsOn builds the query-path instruments on meter. +func queryMetricsOn(meter metric.Meter) *queryMetrics { + duration, _ := meter.Float64Histogram("wavehouse_clickhouse_query_duration_seconds", + metric.WithDescription("ClickHouse calls on the query paths by operation (query, pipe, pipe_write, ops_query) and result: ok, or the error code the response carries without its clickhouse. prefix (rejected, limit_exceeded, access_denied, misconfigured, unavailable, response_too_large, unknown)"), + metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(latencyBuckets...)) + return &queryMetrics{duration: duration} +} + +// observe records one call of operation that took d and ended in err, +// classed as the response classes it (chFailureOf, under caps). +func (m *queryMetrics) observe(operation string, d time.Duration, err error, caps queryCaps) { + result := resultOK + if err != nil { + result = chResult(chFailureOf(err, http.StatusInternalServerError, caps).code) + } + m.record(operation, d, result) +} + +// record records one call of operation that took d, with result as given. +func (m *queryMetrics) record(operation string, d time.Duration, result string) { + if m == nil { + return + } + m.duration.Record(context.Background(), d.Seconds(), metric.WithAttributes( + attribute.String("operation", operation), attribute.String("result", result))) +} + +// chResult is the "result" label of a failure answered with code. +func chResult(code string) string { + return strings.TrimPrefix(code, "clickhouse.") +} diff --git a/internal/api/metrics_test.go b/internal/api/metrics_test.go new file mode 100644 index 00000000..6617cb69 --- /dev/null +++ b/internal/api/metrics_test.go @@ -0,0 +1,225 @@ +package api + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/Wave-RF/WaveHouse/internal/auth" + "github.com/Wave-RF/WaveHouse/internal/query" + "github.com/Wave-RF/WaveHouse/internal/settings" + "github.com/Wave-RF/WaveHouse/pkg/pipes" + "github.com/Wave-RF/WaveHouse/pkg/policy" + "github.com/go-chi/chi/v5" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +// histogramCounts collects reader and returns, for the histogram name, how +// many values each label set recorded, keyed by the labels as key=value +// pairs joined by "," in key order. +func histogramCounts(t *testing.T, reader *sdkmetric.ManualReader, name string) map[string]uint64 { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + out := map[string]uint64{} + for _, sm := range rm.ScopeMetrics { + for _, md := range sm.Metrics { + if md.Name != name { + continue + } + hist, ok := md.Data.(metricdata.Histogram[float64]) + require.True(t, ok, "%s is a float64 histogram", name) + for _, dp := range hist.DataPoints { + var pairs []string + for _, kv := range dp.Attributes.ToSlice() { + pairs = append(pairs, string(kv.Key)+"="+kv.Value.String()) + } + out[strings.Join(pairs, ",")] += dp.Count + } + } + } + return out +} + +// testMeterReader is a meter provider of the test's own, so a test reads +// only what it recorded and runs in parallel. +func testMeterReader(t *testing.T) (*sdkmetric.MeterProvider, *sdkmetric.ManualReader) { + t.Helper() + reader := sdkmetric.NewManualReader() + mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)) + t.Cleanup(func() { _ = mp.Shutdown(context.Background()) }) + return mp, reader +} + +// TestHTTPMetrics_LabelsByRoutePattern: a request is recorded under the +// route it matched, its pattern and not its path, so a path parameter adds +// no series (the mount's own pattern when it was answered inside the mount +// without reaching a route, unmatched when it reached no mount); under the +// method, a closed set; and under its status class, a panic counting as the +// 500 the recoverer answers. +func TestHTTPMetrics_LabelsByRoutePattern(t *testing.T) { + t.Parallel() + mp, reader := testMeterReader(t) + m := httpMetricsOn(mp.Meter(meterName)) + + r := chi.NewRouter() + r.Use(m.middleware) + r.Use(jsonRecoverer) + r.Route("/v1", func(r chi.Router) { + r.Group(func(r chi.Router) { + r.Use(func(next http.Handler) http.Handler { return next }) + r.Get("/pipes/{name}", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusNoContent) }) + r.Post("/ingest", func(w http.ResponseWriter, _ *http.Request) { writeJSONError(w, http.StatusBadRequest, "bad") }) + r.Get("/stream", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }) + }) + }) + r.Get("/boom", func(http.ResponseWriter, *http.Request) { panic("boom") }) + + for _, req := range []struct{ method, path string }{ + {http.MethodGet, "/v1/pipes/a"}, + {http.MethodGet, "/v1/pipes/b"}, + {http.MethodPost, "/v1/ingest"}, + {http.MethodGet, "/v1/stream"}, + {"PURGE", "/v1/pipes/a"}, + {http.MethodDelete, "/v1/pipes/a"}, + {http.MethodGet, "/v1/nope"}, + {http.MethodGet, "/nope"}, + {http.MethodGet, "/boom"}, + } { + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequestWithContext(t.Context(), req.method, req.path, nil)) + } + + assert.Equal(t, map[string]uint64{ + "method=GET,route=/v1/pipes/{name},status_class=2xx": 2, + "method=POST,route=/v1/ingest,status_class=4xx": 1, + "method=other,route=unmatched,status_class=4xx": 1, + "method=DELETE,route=/v1/*,status_class=4xx": 1, + "method=GET,route=/v1/*,status_class=4xx": 1, + "method=GET,route=unmatched,status_class=4xx": 1, + "method=GET,route=/boom,status_class=5xx": 1, + }, histogramCounts(t, reader, "wavehouse_http_request_duration_seconds"), + "/v1/stream is not timed: its lifetime is wavehouse_sse_stream_duration_seconds") +} + +// TestHTTPMetrics_RouterRecords: the routers every listener serves time +// their requests, the ops-only one included. +func TestHTTPMetrics_RouterRecords(t *testing.T) { + // No t.Parallel(): newHTTPMetrics reads the global meter provider, which + // this test swaps. + mp, reader := testMeterReader(t) + saved := otel.GetMeterProvider() + otel.SetMeterProvider(mp) + t.Cleanup(func() { otel.SetMeterProvider(saved) }) + + h := NewOpsRouter(OpsDependencies{Health: NewHealthHandler(), Version: NewVersionHandler("v", "c", "b")}) + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/version", nil)) + + assert.Equal(t, map[string]uint64{"method=GET,route=/version,status_class=2xx": 1}, + histogramCounts(t, reader, "wavehouse_http_request_duration_seconds")) +} + +func TestStatusClass(t *testing.T) { + t.Parallel() + for status, want := range map[int]string{0: "2xx", 101: "1xx", 200: "2xx", 204: "2xx", 304: "3xx", 404: "4xx", 499: "4xx", 500: "5xx", 503: "5xx"} { + assert.Equal(t, want, statusClass(status), "status %d", status) + } +} + +// TestQueryMetrics_ClickHouseErrors: each failed ClickHouse call on the +// native query paths is recorded once, under the code its response carries. +func TestQueryMetrics_ClickHouseErrors(t *testing.T) { + t.Parallel() + for _, tc := range chErrorCases(t) { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + want := chResult(tc.wantCode) + + mp, reader := testMeterReader(t) + perms := tc.caps + perms.AllowColumns = []string{"*"} + sq := newCapturingHandler(t, &failingConn{err: tc.err}, policyWithViewer(perms)) + sq.metrics = queryMetricsOn(mp.Meter(meterName)) + sq.Handle(httptest.NewRecorder(), withTenant(viewerRequest(t, query.StructuredQuery{Columns: []string{"page"}}))) + assert.Equal(t, map[string]uint64{"operation=query,result=" + want: 1}, + histogramCounts(t, reader, "wavehouse_clickhouse_query_duration_seconds")) + + if tc.caps.MaxExecutionTime > 0 || tc.caps.MaxMemoryUsage > 0 { + return // a pipe carries no role caps + } + mp, reader = testMeterReader(t) + store := staticPipes(&pipes.NamedQuery{Name: "p", SQL: "SELECT 1", AllowedRoles: []string{"viewer"}}) + p := NewPipesHandler(store, staticPolicy(&policy.Policy{}), fixedConn(&failingConn{err: tc.err}), nil, func(*settings.Store) time.Duration { return time.Second }) + p.metrics = queryMetricsOn(mp.Meter(meterName)) + r := pipesRequest(t, http.MethodGet, "/v1/pipes/p", "p", nil) + p.Execute(httptest.NewRecorder(), withTenant(r.WithContext(auth.WithRole(r.Context(), "viewer")))) + assert.Equal(t, map[string]uint64{"operation=pipe,result=" + want: 1}, + histogramCounts(t, reader, "wavehouse_clickhouse_query_duration_seconds")) + + mp, reader = testMeterReader(t) + wp := writerPipesHandler(t, &writeConn{err: tc.err}, nil, &pipes.NamedQuery{Name: "log", SQL: "INSERT INTO audit_log VALUES ({{msg}}, now())"}) + wp.metrics = queryMetricsOn(mp.Meter(meterName)) + pipeCallAs(t, wp, "log") + assert.Equal(t, map[string]uint64{"operation=pipe_write,result=" + want: 1}, + histogramCounts(t, reader, "wavehouse_clickhouse_query_duration_seconds")) + }) + } +} + +// TestQueryMetrics_Proxy: the raw-SQL proxy records each call it makes, ok +// or under the code its response carries, the response cap included; a +// request refused before the call records nothing. +func TestQueryMetrics_Proxy(t *testing.T) { + t.Parallel() + mp, reader := testMeterReader(t) + h := newProxyHandler(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + sql, _ := io.ReadAll(r.Body) + switch string(sql) { + case "SELECT 1": + _, _ = w.Write([]byte(`{"data":[{"1":1}]}`)) + case "SELECT big": + _, _ = w.Write([]byte(`{"data":[` + strings.Repeat(`{"x":1},`, 64) + `{"x":1}]}`)) + default: + w.Header().Set("X-ClickHouse-Exception-Code", "62") + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("Code: 62. DB::Exception: Syntax error. (SYNTAX_ERROR)\n")) + } + })) + h.metrics = queryMetricsOn(mp.Meter(meterName)) + h.maxResponseBytes = 256 + + for _, sql := range []string{"SELECT 1", "SELECT 1", "SELEC", "SELECT big", ""} { + body, err := json.Marshal(queryRequest{SQL: sql}) + require.NoError(t, err) + postQuery(h, body) + } + + assert.Equal(t, map[string]uint64{ + "operation=ops_query,result=ok": 2, + "operation=ops_query,result=rejected": 1, + "operation=ops_query,result=response_too_large": 1, + }, histogramCounts(t, reader, "wavehouse_clickhouse_query_duration_seconds")) +} + +// TestQueryMetrics_NilIsNoop: a handler built without its constructor +// records nothing and does not panic. +func TestQueryMetrics_NilIsNoop(t *testing.T) { + t.Parallel() + var m *queryMetrics + assert.NotPanics(t, func() { + m.observe(opPipe, time.Second, nil, queryCaps{}) + m.record(opOpsQuery, time.Second, resultOK) + }) + var hm *httpMetrics + next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) + assert.NotNil(t, hm.middleware(next)) +} diff --git a/internal/api/pipes.go b/internal/api/pipes.go index 5f9fd526..872e2faf 100644 --- a/internal/api/pipes.go +++ b/internal/api/pipes.go @@ -44,10 +44,13 @@ type PipesHandler struct { // without allocating 1 MiB per run); not a production knob. Mirrors // StructuredQueryHandler / QueryHandler. maxRequestBytes int64 + + // metrics records each ClickHouse call; nil records nothing. + metrics *queryMetrics } func NewPipesHandler(source func(*settings.Store) pipes.Source, policySource PolicySource, conn func(*settings.Store) driver.Conn, c cache.Cache, queryTimeout func(*settings.Store) time.Duration) *PipesHandler { - return &PipesHandler{Source: source, PolicySource: policySource, CHConn: conn, Cache: c, queryTimeout: queryTimeout} + return &PipesHandler{Source: source, PolicySource: policySource, CHConn: conn, Cache: c, queryTimeout: queryTimeout, metrics: newQueryMetrics()} } // List returns all named queries of the ?tenant= (admin endpoint). @@ -190,7 +193,7 @@ func (h *PipesHandler) Execute(w http.ResponseWriter, r *http.Request) { // Execute with singleflight. v, err, _ := h.sf.Do(cacheKey, func() (interface{}, error) { - data, queryDuration, err := h.run(r.Context(), store, conn, sql, params) + data, queryDuration, err := h.run(r.Context(), store, conn, opPipe, sql, params) if err != nil { return nil, err } @@ -221,7 +224,7 @@ func (h *PipesHandler) executeWrite(w http.ResponseWriter, r *http.Request, stor writeUnavailable(w, noConnectionMessage, retryAfterPool) return } - data, _, err := h.run(r.Context(), store, conn, sql, params) + data, _, err := h.run(r.Context(), store, conn, opPipeWrite, sql, params) if err != nil { writeCHWriteError(w, r, err, err.Error()) return @@ -233,13 +236,15 @@ func (h *PipesHandler) executeWrite(w http.ResponseWriter, r *http.Request, stor } // run executes a pipe's bound SQL under the tenant's query timeout and -// returns the rows as JSON with how long ClickHouse took. -func (h *PipesHandler) run(ctx context.Context, store *settings.Store, conn driver.Conn, sql string, params []any) ([]byte, time.Duration, error) { +// returns the rows as JSON with how long ClickHouse took, recording the call +// as operation (opPipe or opPipeWrite). +func (h *PipesHandler) run(ctx context.Context, store *settings.Store, conn driver.Conn, operation, sql string, params []any) ([]byte, time.Duration, error) { queryCtx, cancel := context.WithTimeout(ctx, timeoutOf(h.queryTimeout, store)) defer cancel() start := time.Now() rows, err := executeCHQuery(queryCtx, conn, sql, params) queryDuration := time.Since(start) + h.metrics.observe(operation, queryDuration, err, queryCaps{}) if err != nil { return nil, 0, err } diff --git a/internal/api/query.go b/internal/api/query.go index fa782b0c..f429f80c 100644 --- a/internal/api/query.go +++ b/internal/api/query.go @@ -70,6 +70,9 @@ type QueryHandler struct { // body cap (maxRequestBodyBytes). When 0, the default applies. Same // test-only purpose as maxResponseBytes. maxRequestBytes int64 + + // metrics records each ClickHouse call; nil records nothing. + metrics *queryMetrics } const ( @@ -124,6 +127,7 @@ func NewQueryHandler(target func(*settings.Store) chconn.Target, queryTimeout fu clients: chconn.NewHTTPClients(proxyHTTPClient), target: target, queryTimeout: queryTimeout, + metrics: newQueryMetrics(), } } @@ -262,8 +266,12 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { httpReq.Header.Set("X-ClickHouse-Key", target.Password) } + // The call is timed from the request to the end of the response body: + // ClickHouse streams the result, so the body is most of the query. + start := time.Now() resp, err := h.clients.For(target).Do(httpReq) if err != nil { + h.metrics.observe(opOpsQuery, time.Since(start), err, queryCaps{}) writeCHError(w, r, err, "clickhouse request failed: "+err.Error(), http.StatusBadGateway, queryCaps{}) return } @@ -276,11 +284,14 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { respCap = h.maxResponseBytes } body, err := io.ReadAll(io.LimitReader(resp.Body, respCap+1)) + elapsed := time.Since(start) if err != nil { + h.metrics.observe(opOpsQuery, elapsed, err, queryCaps{}) writeCHError(w, r, err, "read clickhouse response: "+err.Error(), http.StatusBadGateway, queryCaps{}) return } if int64(len(body)) > respCap { + h.metrics.record(opOpsQuery, elapsed, chResult(codeCHResponseTooLarge)) // The same query overflows again, so not retryable. retryable := false writeJSONErrorBody(w, http.StatusBadGateway, errorBody{ @@ -297,6 +308,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { // exception code it sends with it does (#403). The message is // ClickHouse's own text, verbatim. chErr := chconn.NewHTTPError(&http.Response{StatusCode: resp.StatusCode, Header: resp.Header, Body: io.NopCloser(bytes.NewReader(body))}) + h.metrics.observe(opOpsQuery, elapsed, chErr, queryCaps{}) msg := strings.TrimSpace(string(body)) if msg == "" { msg = fmt.Sprintf("clickhouse returned status %d", resp.StatusCode) @@ -304,6 +316,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { writeCHError(w, r, chErr, msg, http.StatusBadGateway, queryCaps{}) return } + h.metrics.record(opOpsQuery, elapsed, resultOK) // ClickHouse returns an empty body for mutations (TRUNCATE/INSERT/ // DELETE/etc. with no result set). Marshal those to `[]` so clients diff --git a/internal/api/router.go b/internal/api/router.go index 44990c13..6705af49 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -165,6 +165,7 @@ func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, r := chi.NewRouter() r.Use(requestID) + r.Use(newHTTPMetrics().middleware) // No middleware.RealIP: it rewrites r.RemoteAddr from spoofable forwarded // headers on every request (chi deprecated it for the IP-spoofing GHSAs), // and nothing here reads RemoteAddr — WaveHouse does no per-IP logic (that's diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index 9210bb0a..efc30210 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -48,6 +48,9 @@ type StructuredQueryHandler struct { // 1 MiB per run; not a production tuning knob, hence unexported. Mirrors // QueryHandler. maxRequestBytes int64 + + // metrics records each ClickHouse call; nil records nothing. + metrics *queryMetrics } func NewStructuredQueryHandler( @@ -67,6 +70,7 @@ func NewStructuredQueryHandler( bucketSecs: bucketSecs, queryTimeout: queryTimeout, defaultMaxRows: defaultMaxRows, + metrics: newQueryMetrics(), } } @@ -253,6 +257,7 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) rows, err := executeCHQuery(queryCtx, conn, result.SQL, result.Params) queryDuration := time.Since(start) + h.metrics.observe(opStructuredQuery, queryDuration, err, caps) if err != nil { // TODO: depending on the error, we may actually want to cache it return nil, err From 2533daed3d38ae84383550d61dfea552f4490fca Mon Sep 17 00:00:00 2001 From: Taite Lee <113070390+taitelee@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:33:01 +0000 Subject: [PATCH 2/5] fix(observability): count a client's disconnect as canceled, not unavailable Also correct where a tenant refusal's route label lands and that only server.port is timed, per pre-push review. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019YVNGPfArHqjc3Xt6N9App --- CHANGELOG.md | 2 +- docs/src/content/docs/architecture.md | 2 +- docs/src/content/docs/deployment.md | 16 +++++++++------ internal/api/metrics.go | 29 +++++++++++++++++++-------- internal/api/metrics_test.go | 23 ++++++++++++++++++++- internal/api/pipes.go | 2 +- internal/api/query.go | 6 +++--- internal/api/structured_query.go | 2 +- 8 files changed, 60 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3831980..ea7e29da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,7 +10,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added -- **HTTP request latency and query-path ClickHouse calls are histograms** (`internal/api/metrics.go` (new, + `metrics_test.go`), `internal/api/{router,query,structured_query,pipes}.go`, `docs/src/content/docs/{deployment,architecture}.md`, `AGENTS.md`): the first tier of [#94](https://github.com/Wave-RF/WaveHouse/issues/94), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). The only request metrics were otelhttp's, which carry no route, so a slow or failing endpoint could not be told from the rest, and nothing timed what the query paths sent to ClickHouse. `wavehouse_http_request_duration_seconds{route, method, status_class}` times every request on every listener but the SSE streams, under the route's pattern (`/v1/pipes/{name}`), never the path; `wavehouse_clickhouse_query_duration_seconds{operation, result}` times each statement `POST /v1/query`, a pipe (`pipe`, `pipe_write`) or `POST /v1/ops/query` sends, with `result` `ok` or the response's error code without its `clickhouse.` prefix, so a histogram's `_count` is both the rate and the error rate. A cache hit records no query, and requests coalesced onto one record it once. Neither carries a tenant or table label. The worker's `INSERT` timing is the next tier. +- **HTTP request latency and query-path ClickHouse calls are histograms** (`internal/api/metrics.go` (new, + `metrics_test.go`), `internal/api/{router,query,structured_query,pipes}.go`, `docs/src/content/docs/{deployment,architecture}.md`, `AGENTS.md`): the first tier of [#94](https://github.com/Wave-RF/WaveHouse/issues/94), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). The only request metrics were otelhttp's, which carry no route, so a slow or failing endpoint could not be told from the rest, and nothing timed what the query paths sent to ClickHouse. `wavehouse_http_request_duration_seconds{route, method, status_class}` times every request on `server.port` but the SSE streams, under the route's pattern (`/v1/pipes/{name}`), never the path; `wavehouse_clickhouse_query_duration_seconds{operation, result}` times each statement `POST /v1/query`, a pipe (`pipe`, `pipe_write`) or `POST /v1/ops/query` sends, with `result` `ok`, `canceled` (the client went away) or the response's error code without its `clickhouse.` prefix, so a histogram's `_count` is both the rate and the error rate. A cache hit records no query, and requests coalesced onto one record it once. Neither carries a tenant or table label. The worker's `INSERT` timing is the next tier. - **Dropped SSE frames say why, on which table and on which connection** (`internal/stream/{metrics,subscriber,hub}.go` (+ tests, `drop_test.go` new), `internal/api/stream.go` (+ `stream_drop_test.go` new), `docs/src/content/docs/{deployment,architecture,configuration}.md*`, `AGENTS.md`): closes [#424](https://github.com/Wave-RF/WaveHouse/issues/424), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777). `wavehouse_sse_dropped_frames_total` counted only frames dropped to a full queue, labeled by `kind` alone, and nothing was logged, so a burst of drops could not be tied to a connection or told apart from clients going away. It now also counts the frames a stream left unwritten when it ended, and carries `reason` (`buffer_full`, `client_gone`, `stream_closed`) and `table` (empty for a table the tenant's schema does not have, since the client names it), which `prometheus.per_table_metrics: false` drops like every other table label; a query that sums the counter now also sees the two new reasons, so filter on `reason="buffer_full"` to keep the old meaning. New `wavehouse_sse_queue_peak_frames` gauge: the deepest any open stream's queue has been since it opened, so a fill between scrapes still shows. Every stream logs `SSE stream opened` and `SSE stream closed` (how it ended, how long it lasted, what it dropped), and a slow consumer logs `SSE stream dropping frames` at `WARN`, at most every 10 seconds per stream, with its queue's depth, peak and capacity; each line carries the connection's `request_id` and `tenant` from the request scope. No metric carries a tenant label. diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 8ddce241..c7eda886 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -87,7 +87,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with `reques - **pipes.go** — Named query pipe handlers: admin listing (`GET /v1/ops/pipes[/{name}]`, read per request from its `pipes.Source`) and execution with parameter binding. A read is cached and coalesced; a write — bound SQL that `IsMutation` (`clickhouse_exec.go`) classifies as one — bypasses both and runs every call. `pipes.json` is the only way to define or change a pipe. - **structured_query.go** — Handler for `POST /v1/query?table={table}`: validates query AST, enforces permissions, builds and executes SQL. - **ch_errors.go** — `writeCHError`, the one mapping from a failed ClickHouse query to a response, shared by `/v1/query`, pipes and `/v1/ops/query` so they cannot drift apart: `chconn.Classify` decides the class, and the class the status, `code` and `retryable` ([ClickHouse errors on the query paths](/api#clickhouse-errors-on-the-query-paths)). A write pipe answers through `writeCHWriteError`, the same mapping with `retryable` always `false` and no `Retry-After`, since the write may have run. -- **metrics.go** — the API's own instruments (meter `wavehouse-api`, no tenant label): `wavehouse_http_request_duration_seconds` by `route` (the matched chi route pattern: a mount's own, such as `/v1/*`, for a request answered inside it short of a route, and `unmatched` for one that reached no mount), `method` (net/http's methods, `other` for any other) and `status_class`, recorded by a middleware every router mounts outside `jsonRecoverer`, so a recovered panic counts as its `500`; `/v1/stream` is not timed, since a stream's lifetime is `wavehouse_sse_stream_duration_seconds`. `wavehouse_clickhouse_query_duration_seconds` by `operation` (`query`, `pipe`, `pipe_write`, `ops_query`) and `result` (`ok`, or the response's error `code` without its `clickhouse.` prefix, through `chFailureOf`), one value per statement sent: a cache hit records none and singleflight-coalesced callers record one. Nil-safe, so a handler built without its constructor records nothing. +- **metrics.go** — the API's own instruments (meter `wavehouse-api`, no tenant label): `wavehouse_http_request_duration_seconds` by `route` (the matched chi route pattern: a mount's own, such as `/v1/*`, for a request answered inside it short of a route, and `unmatched` for one that reached no mount), `method` (net/http's methods, `other` for any other) and `status_class`, recorded by a middleware every router mounts outside `jsonRecoverer`, so a recovered panic counts as its `500`; `/v1/stream` is not timed, since a stream's lifetime is `wavehouse_sse_stream_duration_seconds`. `wavehouse_clickhouse_query_duration_seconds` by `operation` (`query`, `pipe`, `pipe_write`, `ops_query`) and `result` (`ok`, `canceled` for a call cut short by its client going away, which the response would class as `unavailable`, or the response's error `code` without its `clickhouse.` prefix, through `chFailureOf`), one value per statement sent: a cache hit records none and singleflight-coalesced callers record one. Nil-safe, so a handler built without its constructor records nothing. - **ingest.go** — Accepts `POST /v1/ingest?table={table}` in three body shapes: one flat JSON object, a JSON array of them, or NDJSON. The **required** `Content-Type` chooses the format *family* — `application/json` versus the four NDJSON spellings — and within the JSON family the body's first non-whitespace byte picks array versus single object; the bytes never choose the family. Anything that is not exactly one readable media type is a `415`, decided before the body is read: the header is parsed per RFC 9110 §8.3, and because `Content-Type` is a singleton field, repeated header lines must all resolve to the same format and a value carrying a comma is refused unless the value as a whole parses as one media type — a comma inside a *quoted* parameter value is data, so `application/json; a=", application/x-ndjson; b="` is accepted. It then reads the whole (`MaxBytesReader`-capped) body into a pooled buffer and runs the per-format record readers over those bytes, so the `413` lands before any record is processed and peak memory per request is O(body) rather than O(record). Then it validates and encodes each record, and runs the records in windows of up to 256 (`ingestWindow`) through three phases: one dedupe `Reserve` for the window's ids, the publishes in record order (a deduped record under `mq.WithIdempotencyKey`, keyed by `dedupe.IdempotencyKey`), and one `Commit` of the published ids — a window is the unit of a dedupe round trip and of Pebble's commit `fsync`. An id another request holds answers `503` with the lease as `Retry-After`, a store that cannot answer (`dedupe.ErrUnavailable`) `503` with `Retry-After: 5`; a publish that fails at a record commits the ones before it and releases the rest, except that a failure other than `mq.ErrQueueFull` may have stored the event, so that record's claim is left to lapse and the idempotency key drops the retry's copy if it comes within the stream's two-minute duplicate window — `mq.ErrUnavailable` (a broker blip) is one such failure, and still answers `503`: with the lease, rounded up to whole seconds, as `Retry-After` when the failing record held a claim left to lapse, else the flat `Retry-After: 5`. Each row goes through `mq.Publisher` on `mq.Topic{Tenant, Table, Scope}` (the request's tenant, read off its resolved store — `store.Tenant()` — and raw names; the subject it becomes is `internal/mq`'s; a full queue comes back as `mq.ErrQueueFull`, which is the `503` + `Retry-After`). When dedup is on, a row missing the configured `id_field` (or setting it to `null`) can't be deduped: it is logged at `WARN` and counted by `wavehouse_ingest_dedupe_missing_id_total` (labeled by `table`), then published un-deduped — or rejected when `dedupe.require_id` is set ([#219](https://github.com/Wave-RF/WaveHouse/issues/219)). - **query.go** — Proxies raw SQL for `POST /v1/ops/query` straight to the `?tenant=`'s ClickHouse HTTP interface (`chconn.Pools.Target` by the resolved store's tenant; the zero target — no pool — is a `503` with `Retry-After`). **Not cached** — sets `Cache-Control: no-store` so every request hits ClickHouse; DateTime is rendered ISO-8601 via `date_time_output_format=iso` (the Go-side type conversion lives in the structured-query / pipes path, not here). - **stream.go** — Real-time streaming via SSE. Callers select a table with the `?table=` query parameter. Each connection registers one `Subscriber` (the `stream/` package) with both the event `Hub` (under its `(topic, role)`) and its tenant's keepalive wheel (`Heartbeater`, a lookup by the request's store; a tenant no longer served has none, and its streams are ending), then drains both from a single byte-pump — so idle streams keep emitting `:` keepalive comments (surviving reverse-proxy idle timeouts) while live events arrive already projected and serialized. Per-event projection/serialization happens **once per role** in the `Hub`, not once per subscriber ([#294](https://github.com/Wave-RF/WaveHouse/issues/294)); the handler also snapshots the connection's JWT claims onto the `Subscriber`, which the `Hub` evaluates per subscriber when the role carries a row-level `filter` ([#319](https://github.com/Wave-RF/WaveHouse/issues/319)). Gap-fill replay (`mq.Replayer.ReplaySince` on the connection's `mq.Topic` — a `DeliverByStartTime` consumer inside `internal/mq`) stays per-connection (low-volume, one-time on connect). A stream ends, a gap-fill in progress included, when the server begins shutting down (`Closing`) or its `Subscriber` is evicted because its tenant is no longer served (`Hub.Prune`); one admitted just before the reload that stopped serving its tenant, and registered just after the prune, is ended right after it registers (`Served`). Each stream logs its open and its close (why it ended, how long it lasted, what it dropped), and a stream dropping frames to a full queue logs a rate-limited `WARN` with its queue state, all through the request's context, so each line carries the connection's `request_id` and `tenant`; on close the handler counts a frame it failed to write, and what is still queued, as dropped (`client_gone` or `stream_closed`). diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index c3325330..815b8aba 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -835,18 +835,22 @@ count by (version) (wavehouse_build_info) ### Request latency and ClickHouse queries -Every process times its HTTP requests, and an API process times each statement its query paths send to ClickHouse. Both are histograms in seconds, with buckets from 5 ms to 30 s, and neither carries a tenant label: +Every process times the HTTP requests on `server.port` (scrapes on a dedicated `prometheus.port` listener are not timed), and an API process times each statement its query paths send to ClickHouse. Both are histograms in seconds, with buckets from 5 ms to 30 s, and neither carries a tenant label: | Metric | Labels | | --- | --- | -| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`, or a tenant that is not served) is labeled with the prefix's pattern, such as `/v1/*`, and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | -| `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | +| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`; a request refused for its tenant keeps the route it matched, under `4xx`; and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | +| `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, `canceled` when the client went away before the statement finished, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | -The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`, and the queries ClickHouse could not take: +The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`: + +```promql +sum by (route) (rate(wavehouse_http_request_duration_seconds_count{status_class="5xx"}[5m])) / sum by (route) (rate(wavehouse_http_request_duration_seconds_count[5m])) +``` + +Queries ClickHouse could not take: ```promql -sum by (route) (rate(wavehouse_http_request_duration_seconds_count{status_class="5xx"}[5m])) - / sum by (route) (rate(wavehouse_http_request_duration_seconds_count[5m])) sum by (operation) (rate(wavehouse_clickhouse_query_duration_seconds_count{result="unavailable"}[5m])) > 0 ``` diff --git a/internal/api/metrics.go b/internal/api/metrics.go index df505f17..174049f6 100644 --- a/internal/api/metrics.go +++ b/internal/api/metrics.go @@ -2,6 +2,7 @@ package api import ( "context" + "errors" "net/http" "strings" "time" @@ -21,8 +22,9 @@ var latencyBuckets = []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, // routeUnmatched is the route label of a request that reached no route or // mount: an unknown top-level path, a method chi does not route, or a CORS // preflight answered before routing. One answered inside a mount short of a -// route (an unknown path under /v1, a tenant not served) carries the mount's -// pattern, /v1/*. +// route (an unknown path or an unsupported method under /v1) carries the +// mount's pattern, /v1/*; one a group middleware refuses (a tenant not +// served) has matched its route already and carries that. const routeUnmatched = "unmatched" // httpMetrics records each request's latency by route, method and status @@ -114,8 +116,12 @@ const ( opOpsQuery = "ops_query" // raw SQL, POST /v1/ops/query ) -// resultOK is the "result" of a query ClickHouse answered. -const resultOK = "ok" +// Results that are not an error code: a query ClickHouse answered, and one +// cut short because its client went away. +const ( + resultOK = "ok" + resultCanceled = "canceled" +) // queryMetrics records each ClickHouse call the query paths make: one per // statement sent, so a cache hit records none and callers coalesced onto one @@ -131,16 +137,23 @@ func newQueryMetrics() *queryMetrics { return queryMetricsOn(otel.Meter(meterNam // queryMetricsOn builds the query-path instruments on meter. func queryMetricsOn(meter metric.Meter) *queryMetrics { duration, _ := meter.Float64Histogram("wavehouse_clickhouse_query_duration_seconds", - metric.WithDescription("ClickHouse calls on the query paths by operation (query, pipe, pipe_write, ops_query) and result: ok, or the error code the response carries without its clickhouse. prefix (rejected, limit_exceeded, access_denied, misconfigured, unavailable, response_too_large, unknown)"), + metric.WithDescription("ClickHouse calls on the query paths by operation (query, pipe, pipe_write, ops_query) and result: ok, canceled (the client went away), or the error code the response carries without its clickhouse. prefix (rejected, limit_exceeded, access_denied, misconfigured, unavailable, response_too_large, unknown)"), metric.WithUnit("s"), metric.WithExplicitBucketBoundaries(latencyBuckets...)) return &queryMetrics{duration: duration} } // observe records one call of operation that took d and ended in err, -// classed as the response classes it (chFailureOf, under caps). -func (m *queryMetrics) observe(operation string, d time.Duration, err error, caps queryCaps) { +// classed as the response classes it (chFailureOf, under caps); reqCtx is the +// request's own context. A call cut short because the client went away is +// resultCanceled, not the unavailable its response would read as: nobody +// reads that response, and ClickHouse was not the one that failed. +func (m *queryMetrics) observe(reqCtx context.Context, operation string, d time.Duration, err error, caps queryCaps) { result := resultOK - if err != nil { + switch { + case err == nil: + case errors.Is(err, context.Canceled) && reqCtx.Err() != nil: + result = resultCanceled + default: result = chResult(chFailureOf(err, http.StatusInternalServerError, caps).code) } m.record(operation, d, result) diff --git a/internal/api/metrics_test.go b/internal/api/metrics_test.go index 6617cb69..2cce7112 100644 --- a/internal/api/metrics_test.go +++ b/internal/api/metrics_test.go @@ -3,6 +3,7 @@ package api import ( "context" "encoding/json" + "fmt" "io" "net/http" "net/http/httptest" @@ -216,10 +217,30 @@ func TestQueryMetrics_NilIsNoop(t *testing.T) { t.Parallel() var m *queryMetrics assert.NotPanics(t, func() { - m.observe(opPipe, time.Second, nil, queryCaps{}) + m.observe(context.Background(), opPipe, time.Second, nil, queryCaps{}) m.record(opOpsQuery, time.Second, resultOK) }) var hm *httpMetrics next := http.HandlerFunc(func(http.ResponseWriter, *http.Request) {}) assert.NotNil(t, hm.middleware(next)) } + +// TestQueryMetrics_ClientGone: a call cut short because the client went away +// is canceled, not unavailable, which would read as ClickHouse failing; a +// cancel of WaveHouse's own (the time-cap backstop) with the client still +// there stays unavailable. +func TestQueryMetrics_ClientGone(t *testing.T) { + t.Parallel() + mp, reader := testMeterReader(t) + m := queryMetricsOn(mp.Meter(meterName)) + gone, cancel := context.WithCancel(t.Context()) + cancel() + err := fmt.Errorf("clickhouse query: %w", context.Canceled) + m.observe(gone, opStructuredQuery, time.Millisecond, err, queryCaps{}) + m.observe(t.Context(), opStructuredQuery, time.Millisecond, err, queryCaps{time: true}) + + assert.Equal(t, map[string]uint64{ + "operation=query,result=canceled": 1, + "operation=query,result=unavailable": 1, + }, histogramCounts(t, reader, "wavehouse_clickhouse_query_duration_seconds")) +} diff --git a/internal/api/pipes.go b/internal/api/pipes.go index 872e2faf..3fb45239 100644 --- a/internal/api/pipes.go +++ b/internal/api/pipes.go @@ -244,7 +244,7 @@ func (h *PipesHandler) run(ctx context.Context, store *settings.Store, conn driv start := time.Now() rows, err := executeCHQuery(queryCtx, conn, sql, params) queryDuration := time.Since(start) - h.metrics.observe(operation, queryDuration, err, queryCaps{}) + h.metrics.observe(ctx, operation, queryDuration, err, queryCaps{}) if err != nil { return nil, 0, err } diff --git a/internal/api/query.go b/internal/api/query.go index f429f80c..cf9adff4 100644 --- a/internal/api/query.go +++ b/internal/api/query.go @@ -271,7 +271,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { start := time.Now() resp, err := h.clients.For(target).Do(httpReq) if err != nil { - h.metrics.observe(opOpsQuery, time.Since(start), err, queryCaps{}) + h.metrics.observe(r.Context(), opOpsQuery, time.Since(start), err, queryCaps{}) writeCHError(w, r, err, "clickhouse request failed: "+err.Error(), http.StatusBadGateway, queryCaps{}) return } @@ -286,7 +286,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { body, err := io.ReadAll(io.LimitReader(resp.Body, respCap+1)) elapsed := time.Since(start) if err != nil { - h.metrics.observe(opOpsQuery, elapsed, err, queryCaps{}) + h.metrics.observe(r.Context(), opOpsQuery, elapsed, err, queryCaps{}) writeCHError(w, r, err, "read clickhouse response: "+err.Error(), http.StatusBadGateway, queryCaps{}) return } @@ -308,7 +308,7 @@ func (h *QueryHandler) Handle(w http.ResponseWriter, r *http.Request) { // exception code it sends with it does (#403). The message is // ClickHouse's own text, verbatim. chErr := chconn.NewHTTPError(&http.Response{StatusCode: resp.StatusCode, Header: resp.Header, Body: io.NopCloser(bytes.NewReader(body))}) - h.metrics.observe(opOpsQuery, elapsed, chErr, queryCaps{}) + h.metrics.observe(r.Context(), opOpsQuery, elapsed, chErr, queryCaps{}) msg := strings.TrimSpace(string(body)) if msg == "" { msg = fmt.Sprintf("clickhouse returned status %d", resp.StatusCode) diff --git a/internal/api/structured_query.go b/internal/api/structured_query.go index efc30210..1709ba36 100644 --- a/internal/api/structured_query.go +++ b/internal/api/structured_query.go @@ -257,7 +257,7 @@ func (h *StructuredQueryHandler) Handle(w http.ResponseWriter, r *http.Request) rows, err := executeCHQuery(queryCtx, conn, result.SQL, result.Params) queryDuration := time.Since(start) - h.metrics.observe(opStructuredQuery, queryDuration, err, caps) + h.metrics.observe(r.Context(), opStructuredQuery, queryDuration, err, caps) if err != nil { // TODO: depending on the error, we may actually want to cache it return nil, err From fc5e417c7e4bfddca19c53e8f43e3ffa43d818c9 Mon Sep 17 00:00:00 2001 From: Taite Lee <113070390+taitelee@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:33:57 +0000 Subject: [PATCH 3/5] docs(observability): a rejected tenant's refusal is a 5xx Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019YVNGPfArHqjc3Xt6N9App --- docs/src/content/docs/deployment.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index 815b8aba..7753d90c 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -839,7 +839,7 @@ Every process times the HTTP requests on `server.port` (scrapes on a dedicated ` | Metric | Labels | | --- | --- | -| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`; a request refused for its tenant keeps the route it matched, under `4xx`; and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | +| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`; a request refused for its tenant keeps the route it matched (`4xx` for a malformed or unknown tenant, `5xx` for one whose settings were rejected); and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | | `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, `canceled` when the client went away before the statement finished, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`: From e224c63eb86671504d5f3a1d650a4dbea64b562d Mon Sep 17 00:00:00 2001 From: Taite Lee <113070390+taitelee@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:37:04 +0000 Subject: [PATCH 4/5] docs(observability): the /v1/ops gates' refusals are labeled /v1/ops/* Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019YVNGPfArHqjc3Xt6N9App --- docs/src/content/docs/deployment.md | 2 +- internal/api/metrics.go | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index 7753d90c..f08107e9 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -839,7 +839,7 @@ Every process times the HTTP requests on `server.port` (scrapes on a dedicated ` | Metric | Labels | | --- | --- | -| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`; a request refused for its tenant keeps the route it matched (`4xx` for a malformed or unknown tenant, `5xx` for one whose settings were rejected); and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | +| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`. The `/v1/ops` gates run before its routes, so a request they refuse (`401` or `403` from the admin gate, `503` while a token verifier is pending) is labeled `/v1/ops/*`, not the route it named. On the other `/v1` routes a request refused for its tenant keeps the route it matched (`4xx` for a malformed or unknown tenant, `5xx` for one whose settings were rejected); and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | | `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, `canceled` when the client went away before the statement finished, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`: diff --git a/internal/api/metrics.go b/internal/api/metrics.go index 174049f6..a0696af3 100644 --- a/internal/api/metrics.go +++ b/internal/api/metrics.go @@ -23,8 +23,9 @@ var latencyBuckets = []float64{.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10, // mount: an unknown top-level path, a method chi does not route, or a CORS // preflight answered before routing. One answered inside a mount short of a // route (an unknown path or an unsupported method under /v1) carries the -// mount's pattern, /v1/*; one a group middleware refuses (a tenant not -// served) has matched its route already and carries that. +// mount's pattern, /v1/*, as does one the /v1/ops gates refuse, which run +// before its routes (/v1/ops/*); one a group middleware refuses (a tenant +// not served) has matched its route already and carries that. const routeUnmatched = "unmatched" // httpMetrics records each request's latency by route, method and status From 7628776ee6ff98f2043f910acff0ae74b291c427 Mon Sep 17 00:00:00 2001 From: Taite Lee <113070390+taitelee@users.noreply.github.com> Date: Tue, 20 Oct 2026 01:39:13 +0000 Subject: [PATCH 5/5] test(observability): pin an ops gate's refusal under /v1/ops/*; fix the unmatched sentence Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_019YVNGPfArHqjc3Xt6N9App --- docs/src/content/docs/deployment.md | 2 +- internal/api/metrics_test.go | 11 ++++++++++- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index f08107e9..af894450 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -839,7 +839,7 @@ Every process times the HTTP requests on `server.port` (scrapes on a dedicated ` | Metric | Labels | | --- | --- | -| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`. The `/v1/ops` gates run before its routes, so a request they refuse (`401` or `403` from the admin gate, `503` while a token verifier is pending) is labeled `/v1/ops/*`, not the route it named. On the other `/v1` routes a request refused for its tenant keeps the route it matched (`4xx` for a malformed or unknown tenant, `5xx` for one whose settings were rejected); and one that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) with `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | +| `wavehouse_http_request_duration_seconds` | `route` is the route's pattern, such as `/v1/pipes/{name}`, never the path a client sent. A request answered under a prefix without reaching one of its routes (an unknown path or an unsupported method under `/v1`) is labeled with the prefix's pattern, such as `/v1/*`. The `/v1/ops` gates run before its routes, so a request they refuse (`401` or `403` from the admin gate, `503` while a token verifier is pending) is labeled `/v1/ops/*`, not the route it named. On the other `/v1` routes a request refused for its tenant keeps the route it matched (`4xx` for a malformed or unknown tenant, `5xx` for one whose settings were rejected). A request that reached no prefix (an unknown top-level path, a method net/http does not name, or a CORS preflight) is labeled `unmatched`. `method` is the request's, or `other` for a method net/http does not name. `status_class` is `2xx`, `3xx`, `4xx` or `5xx`; a panic counts as the `500` it answers. SSE streams on `/v1/stream` are not counted: `wavehouse_sse_stream_duration_seconds` is their lifetime. | | `wavehouse_clickhouse_query_duration_seconds` | `operation` is the path that sent the statement: `query` (`POST /v1/query`), `pipe` and `pipe_write` (a pipe that reads, and one that writes) or `ops_query` (raw SQL on `POST /v1/ops/query`, timed to the end of the response body). `result` is `ok`, `canceled` when the client went away before the statement finished, or the [error code](/api#clickhouse-errors-on-the-query-paths) the response carries without its `clickhouse.` prefix: `rejected`, `limit_exceeded`, `access_denied`, `misconfigured`, `unavailable`, `response_too_large` or `unknown`. One value per statement sent: a cache hit records none, and requests coalesced onto one query record it once. | The histogram's `_count` series is the request or query rate, so the error rate needs no counter of its own. The share of requests answering `5xx`: diff --git a/internal/api/metrics_test.go b/internal/api/metrics_test.go index 2cce7112..43b8c0d1 100644 --- a/internal/api/metrics_test.go +++ b/internal/api/metrics_test.go @@ -64,7 +64,8 @@ func testMeterReader(t *testing.T) (*sdkmetric.MeterProvider, *sdkmetric.ManualR // TestHTTPMetrics_LabelsByRoutePattern: a request is recorded under the // route it matched, its pattern and not its path, so a path parameter adds // no series (the mount's own pattern when it was answered inside the mount -// without reaching a route, unmatched when it reached no mount); under the +// without reaching a route, as a mount's own gate refuses it, unmatched +// when it reached no mount); under the // method, a closed set; and under its status class, a panic counting as the // 500 the recoverer answers. func TestHTTPMetrics_LabelsByRoutePattern(t *testing.T) { @@ -82,6 +83,12 @@ func TestHTTPMetrics_LabelsByRoutePattern(t *testing.T) { r.Post("/ingest", func(w http.ResponseWriter, _ *http.Request) { writeJSONError(w, http.StatusBadRequest, "bad") }) r.Get("/stream", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }) }) + r.Route("/ops", func(r chi.Router) { + r.Use(func(http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { writeJSONError(w, http.StatusForbidden, "no") }) + }) + r.Post("/query", func(http.ResponseWriter, *http.Request) {}) + }) }) r.Get("/boom", func(http.ResponseWriter, *http.Request) { panic("boom") }) @@ -93,6 +100,7 @@ func TestHTTPMetrics_LabelsByRoutePattern(t *testing.T) { {"PURGE", "/v1/pipes/a"}, {http.MethodDelete, "/v1/pipes/a"}, {http.MethodGet, "/v1/nope"}, + {http.MethodPost, "/v1/ops/query"}, {http.MethodGet, "/nope"}, {http.MethodGet, "/boom"}, } { @@ -106,6 +114,7 @@ func TestHTTPMetrics_LabelsByRoutePattern(t *testing.T) { "method=other,route=unmatched,status_class=4xx": 1, "method=DELETE,route=/v1/*,status_class=4xx": 1, "method=GET,route=/v1/*,status_class=4xx": 1, + "method=POST,route=/v1/ops/*,status_class=4xx": 1, "method=GET,route=unmatched,status_class=4xx": 1, "method=GET,route=/boom,status_class=5xx": 1, }, histogramCounts(t, reader, "wavehouse_http_request_duration_seconds"),