diff --git a/AGENTS.md b/AGENTS.md index c1b3c03e5..31d5c1aa7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -41,7 +41,7 @@ Twenty internal packages under `internal/` (plus `internal/testutil/` for shared - **`ingest/`** — Ingest worker pipeline (`worker.go`: JetStream input → per-table batch INSERT with DLQ output; over a `Sharded` queue, `claims.go`'s `ClaimShards` narrows the worker to the units this process is assigned — membership leases, capped rendezvous, halt-drain-then-release handover and stop, reset at takeover from a dead owner, each unit's share of a 10,000-row budget of unsettled rows). The pipeline is **insert-only**. The wire format `EventMessage` (`types.go`) carries `{table_name, scope, received_timestamp, format, columns, row}` and nothing else — `row` is one positional `JSONCompactEachRow` line and `columns` names its slots, the table's **insertable** columns (a `MATERIALIZED`/`ALIAS` column cannot be named in an `INSERT`); the worker batches per (tenant, table, column list), the tenant read off each message's `mq.Topic`, and inserts each batch into its tenant's own ClickHouse (`chconn.Pools.Target`); the worker accepts whatever table name the envelope carries (table existence was already checked by the HTTP ingest handler, which `404`s an unknown table before publish; the worker doesn't re-validate), then bulk-INSERTs. In the embedded-NATS deployment (the default), the server runs with `DontListen: true` (`internal/mq/embedded.go`), so the only Publishers reachable on the `ingest.>` subjects are in-process Go code — today, only the HTTP `/v1/ingest?table={table}` handler. Non-insert mutations (`DELETE`/`UPDATE`/`TRUNCATE`/…) must go through `POST /v1/ops/query` under the admin role (the same `RequireAdmin` gate as the rest of `/v1/ops/*`, so non-admin callers never reach the proxy) or through an operator-authored pipe that writes, gated only by its `allowed_roles` (#386). A request with no token (or an invalid one) resolves to the `default_role`, which in a production config is not the admin role (setting them equal is a loudly-warned dev-only setting), so it can't reach this endpoint. Plus `Sweeper` (Active Sweeper for NATS message lifecycle) + `EventMessage`/`BufferConsumerName` types (`types.go`) - **`keyenc/`** — the one escaping composite keys are built from: `Escape` keeps `[A-Za-z0-9_-]` (exactly the tenant-id grammar, so a tenant id is its own escaped form) and writes every other byte as `%XX`, `Unescape` is `url.PathUnescape` (lenient: either hex case, and a byte left unescaped reads as itself, so a `%2D` an earlier build wrote still reads), `Join`/`AppendJoin` escape each field and put a separator between them (they panic on no fields, and on a separator the escaping could write or one outside ASCII) and `Split` reverses them. The package that builds a key takes raw names and escapes them itself, so no caller has to and no field reaches a key unescaped: NATS subjects (`Join`/`Split` after the verbatim tenant) and the cache's keys — the version index and the shared backend's Redis keys (`internal/cache`) — and the dedupe keys (`//`) use it; changing what it keeps orphans every stored key (an orphaned dedupe key lets a seen id through again), and on the shared backend, whose keys every process builds for itself, splits them between builds for the length of a rolling upgrade (a bump one build makes misses the entries the other filed, served until their TTL) - **`mq/`** — the message-queue boundary: the **only** package that imports NATS/JetStream (Key Design Decision #20), and the only one that knows how the broker works. Everything else addresses events by `Topic{Tenant, Table, Scope}` (a validated tenant id and raw names — the tenant leads every subject, `ingest..
`, so one wildcard selects a tenant's traffic, and a topic without one is refused) and states intent through the interfaces — `Publisher` (`ErrQueueFull` is the backpressure signal, `ErrUnavailable` a broker that cannot be reached — both a `503`, with `Retry-After` `30` and `5`; `WithIdempotencyKey` makes a republish inside the queue's duplicate window a no-op), `Subscriber`, `ConsumerManager`/`Consumer`/`ConsumerConfig` (the ingest worker's durable consumer), `DeadLetterer` and `DeadLetterStats` (park a message, count what is parked), `Purger` (drop what is both acked and older than a cutoff — the sweeper), `Replayer` (SSE gap-fill) — composed into `Broker`, which adds each tenant's byte budget (`SetMaxBytes`/`MaxBytes`: the `mq.max_bytes_gb` reload, which opens a tenant's queue the first time) and `Stats` (the system gauges' source). A broker whose ingest queue is split into units one consumer at a time owns implements `Sharded` too (`IngestUnits`, `ResetOrphaned`, `Unowned`; `ConsumerConfig.Units` narrows a consumer to some of them, and its consumer is a `Releaser` and a `Halter`, capped per unit by `ConsumerConfig.MaxHeld`, with `ErrConsumerMismatch` when an operator durable no longer fits), and `Message.OnSettled` runs a hook once, at the first ack or nak attempt, confirmed or not. Every interface speaks per tenant, never per stream: the embedded implementation gives each tenant a queue of its own (a stream pair, `INGEST_`/`DLQ_`), and nothing outside the package may assume that layout — an external implementation may keep one shared stream. Subjects, prefixes, wildcards, stream names, sequences, and ack floors are private to the implementations, whose subject tokens are escaped by the shared `internal/keyenc`: `EmbeddedNATS` (`embedded.go`, `subject.go`, `purge.go`, `deadletter.go`), which `internal/app` constructs and hands everything else as a `mq.Broker`, and `ExternalNATS` (`external.go`, `subject_nats.go`, `nats_topology.go`: an operator-owned cluster whose streams, durables and lease bucket it never creates, changes, purges or deletes; `lease.go` holds `coord.backend: nats`'s leases in that bucket), which `internal/app` constructs from the `mq.nats` block when `mq.backend` is `nats` ([#613](https://github.com/Wave-RF/WaveHouse/issues/613)). Every implementation passes the conformance suite in `internal/mq/mqtest` (`mqtest.Run`), which states the `Broker` contract as behavior; a new backend runs it from its own test, with `mqtest.Caps` only where its semantics legitimately differ -- **`observability/`** — OpenTelemetry pipeline: `InitProvider` wires trace/metric/log providers via OTLP gRPC (each signal independently gated). A top-level `Prometheus` config block drives an optional `/metrics` scrape endpoint that runs independently of OTLP push — standalone (Alloy/Mimir scrape, no collector), alongside OTLP, or off. `NewLogger` produces a slog handler that fans out to stdout AND OTLP (stdout always 100%, OTLP sample-rate-aware). `TraceHandler` injects trace_id/span_id from active spans. `tracer.go` provides W3C trace context propagation over message headers (`InjectHeaders`/`ExtractHeaders` on a plain header map; `internal/mq` injects on every publish and extracts on the `Subscribe` path, so this package never sees a NATS type). +- **`observability/`** — OpenTelemetry pipeline: `InitProvider` wires trace/metric/log providers via OTLP gRPC (each signal independently gated). A top-level `Prometheus` config block drives an optional `/metrics` scrape endpoint that runs independently of OTLP push — standalone (Alloy/Mimir scrape, no collector), alongside OTLP, or off. `NewLogger` produces a slog handler that fans out to stdout AND OTLP (stdout always 100%, OTLP sample-rate-aware). `TraceHandler` injects trace_id/span_id from active spans, and `request_id`/`tenant` from the request scope (`request.go`: `WithRequest`, filled by `api`'s `requestID` middleware and its tenant resolution; on log records and span attributes, never a metric label). `tracer.go` provides W3C trace context propagation over message headers (`InjectHeaders`/`ExtractHeaders` on a plain header map; `internal/mq` injects on every publish and extracts on the `Subscribe` path, so this package never sees a NATS type). - **`pipes/`** — Named query pipes: `NamedQuery` type + `BindParams` + `Source` (read per request; `settings.Store` in production, `Static(q...)` in tests) - **`policy/`** — Hasura-style access control, **role-first**: `TablePolicy` is `map[string]RolePermissions`, and a role's grant splits by operation into `SelectPermissions` (columns, row `filter`, aggregations, the `max_*` limits) and `InsertPermissions` (columns, `check`) — so a field only one side honors does not exist on the other. `Evaluate()` resolves ONE operation and leaves the other side **nil** (`Select *ResolvedSelect` / `Insert *ResolvedInsert`), which every accessor fails closed on — nil is "not resolved", distinct from an empty side, which is "unrestricted" (what the admin return builds). Claim templating (`{{ jwt.claim.path }}`) resolves during that call. Policies come from `Source`, a `func() *Policy` read per call (`settings.Store.Policy` in production, `Static(p)` in tests) - **`query/`** — Structured query AST types + SQL builder with schema validation, structural policy predicate/limit emission, timestamp bucketing diff --git a/CHANGELOG.md b/CHANGELOG.md index 01010a788..e84f1eaaf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,10 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ## Unreleased +### Added + +- **Every request's log records and server span say which request and which tenant they belong to** (`internal/observability/{request,logger}.go` (+ tests), `internal/api/correlation.go` (new, + tests), `internal/api/{router,tenant,dlq}.go`, `cmd/wavehouse/main.go`, a comment in `internal/auth/auth.go`, `docs/src/content/docs/{api,architecture,deployment}.md`, `{configuration,reverse-proxy}.mdx`, `AGENTS.md`): closes [#771](https://github.com/Wave-RF/WaveHouse/issues/771), part of the observability epic [#777](https://github.com/Wave-RF/WaveHouse/issues/777), and carries the request-id item of [#333](https://github.com/Wave-RF/WaveHouse/issues/333) that [#25](https://github.com/Wave-RF/WaveHouse/issues/25) asked for. chi's `RequestID` ran on every request, but nothing read the id, so no log line, span or response carried it. A request now gets an id when it arrives: the `X-Request-Id` it was sent with, when it is sent once and is 1 to 128 printable ASCII characters without spaces, otherwise a random 32-hex-character one, which unlike chi's names no host. The response carries it back in `X-Request-Id`, and `TraceHandler` stamps it as `request_id` on every record logged through the request's context. Once the tenant is resolved (`TenantMW`'s `X-Tenant-ID`, or an ops route's `?tenant=`, read ahead of authentication by `opsRequestTenant`; tenant `0`, from the same point, for any other ops route that names none, reads, raw SQL and the schema refresh alike; a settings reload that names none acts on every tenant and carries none) the records carry `tenant` as well, filled in place so the panic recoverer's record, logged through a context taken before resolution, has it too; a record that names its own tenant keeps it. The otelhttp server span gets both as attributes, `request_id` and `tenant`, so the authorization-denied `WARN`, the operator-key audit line and a request's span can be grouped by tenant and joined to a proxy's access log. Neither is a label on the HTTP request metrics. `cmd/wavehouse`'s boot logger is wrapped in `TraceHandler` too, so stdout carries these fields, and `trace_id`/`span_id`, with `otel.logs.enabled` off, where it used to lose the trace ids. `internal/api/ingest.go` is untouched, since [#712](https://github.com/Wave-RF/WaveHouse/pull/712) rewrites it: its `batch ingested` and per-event lines log through the request's context and carry both fields with no change there, and its `ingest` child span keeps only `table`, with the tenant on the server span above it. + ### Changed - **`Wave-RF/cloudflare-preview` is pinned at v1.1.2** (`.github/workflows/ci.yml`, `.github/workflows/docs-preview-cleanup.yml`). The docs preview comment no longer turns an `@name` in a commit subject into a mention that notifies that person: the old `@` escape was decoded by GitHub, and live testing measured a real mention. URLs in commit subjects no longer autolink. A Cloudflare authentication error is no longer retried three times. ([Wave-RF/cloudflare-preview#2](https://github.com/Wave-RF/cloudflare-preview/issues/2)) diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 6a29c30b5..632a2c19d 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -13,6 +13,7 @@ import ( "github.com/Wave-RF/WaveHouse/internal/app" "github.com/Wave-RF/WaveHouse/internal/config" + "github.com/Wave-RF/WaveHouse/internal/observability" ) // Pre-populated build info variables, set via ldflags by the Makefile and by @@ -160,7 +161,10 @@ Run 'wavehouse -h' for command-specific help. func run(ctx context.Context) int { logLevel := &slog.LevelVar{} logLevel.Set(logLevelFromEnv()) - logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel})) + // TraceHandler, so a request's lines carry request_id and tenant (and its + // trace ids when tracing is on) even when the OTLP logger never replaces + // this one. + logger := slog.New(&observability.TraceHandler{Handler: slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{Level: logLevel})}) slog.SetDefault(logger) logger.Info("starting WaveHouse", "version", Version, "build_time", BuildTime, "git_commit", GitCommit) diff --git a/docs/src/content/docs/api.md b/docs/src/content/docs/api.md index 60c3e5e4f..073af251e 100644 --- a/docs/src/content/docs/api.md +++ b/docs/src/content/docs/api.md @@ -80,6 +80,8 @@ This contract holds for: Historically some error paths defaulted to `text/plain` because they were emitted via `http.Error` or chi's default handlers; those paths now route through a shared `writeJSONError` helper so strict clients can branch on `Content-Type` consistently. +Every response, success or failure, also carries `X-Request-Id`: the id the request is logged and traced under. A request that sends one `X-Request-Id` of 1 to 128 printable ASCII characters, with no spaces, gets its own back; any other gets a fresh random 32-hex-character id. Quote it when reporting a failure ([Following one request or one tenant](/deployment#following-one-request-or-one-tenant)). + The per-endpoint error tables below list the bodies you can expect for each status code; the `Content-Type` and `X-Content-Type-Options` headers above apply uniformly and are not repeated. :::caution[Streaming / partial-write responses] diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 6ab4af23a..3d1d37217 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -75,7 +75,7 @@ internal/ ### `api/` — HTTP Layer -The API layer uses [Chi](https://github.com/go-chi/chi) for routing with RequestID, a CORS middleware that decorates each response from the allowlist of the tenant the request names (`corsOrigins`; the tenant-exempt routes and a refused request read tenant `0`'s, and nothing when no tenant `0` is served), and a custom JSON recoverer (`jsonRecoverer`) that emits a JSON `500` on panic instead of chi's plain-text `middleware.Recoverer`. +The API layer uses [Chi](https://github.com/go-chi/chi) for routing with `requestID` (`correlation.go`: keeps an incoming `X-Request-Id` of 1–128 printable ASCII characters with no spaces, sent once, otherwise mints a random one, echoes it on the response, and opens the request's `observability.WithRequest` scope; `annotateSpan`, after the otelhttp middleware, copies it onto the server span; `TenantMW`, and on `/v1/ops` `opsRequestTenant` ahead of authentication, add the tenant through `observability.SetRequestTenant`, any other ops route naming none adding tenant `0` there too, a settings reload naming none acting on every tenant and adding none), a CORS middleware that decorates each response from the allowlist of the tenant the request names (`corsOrigins`; the tenant-exempt routes and a refused request read tenant `0`'s, and nothing when no tenant `0` is served), and a custom JSON recoverer (`jsonRecoverer`) that emits a JSON `500` on panic instead of chi's plain-text `middleware.Recoverer`. - **router.go** — Route definitions. Public: `/livez`, `/readyz`, and the content-free `/v1/health` SDK ping (plus the permanent `/healthz` alias and the deprecated `/health`, `/ready` aliases). Policy-gated: `/v1/ingest?table={table}`, `/v1/query?table={table}` (structured), `/v1/pipes/{name}` (named pipes), `/v1/stream`. Admin-only (`RequireAdmin` — role == `policy.admin_role`, or a request bearing the operator key's operator bit, which passes even under a nil policy; over a nested settings directory `NewRouter` mounts the gate with no policy at all, whatever `Dependencies.PolicySource` was wired, so the operator key alone passes): `/v1/ops/schema/*`, `/v1/ops/dlq/stats`, `GET /v1/ops/pipes[/{name}]`, `/v1/ops/settings/reload`, `/v1/ops/query` (raw SQL — same gate as the rest of `/v1/ops/*`). `NewOpsRouter` is the router of a process without the `api` role: the probes and their aliases, `/version` and the same-port metrics path (the part it shares with `NewRouter`, `newProbeRouter`), and `POST /v1/ops/settings/reload` behind `RequireAdmin(nil)`, so only the operator key passes; every other route is a 404, under `/v1/ops` only once that gate has passed. - **auth middleware** — the JWT/JWKS authentication middleware is its own package, [`auth/`](#auth--authentication); the router runs it on every `/v1/*` route. @@ -189,7 +189,8 @@ The **only** package that imports NATS/JetStream — a `depguard` rule in `.gola ### `observability/` — OpenTelemetry Pipeline - **provider.go** — `InitProvider(ctx, serviceName, ProviderConfig)` wires the OTel pipeline. Each output is independently gated; the W3C TraceContext + Baggage propagator is always installed (cheap, harmless when traces are off). Returns `(shutdown, promHandler http.Handler, err)` — `promHandler` is non-nil only when `PrometheusEnabled` is true and reads from a *private* `prometheus.Registry` to avoid leaking the process/Go collectors that `prometheus.DefaultRegisterer` auto-registers. OTLP-metrics push (`MetricsEnabled`) and Prometheus exposition (`PrometheusEnabled`) are independent: either, both, or neither may be set, and any combination produces a single MeterProvider feeding the active readers. The OTLP exporters (traces / metrics-OTLP / logs) take their endpoint from the `OTEL_EXPORTER_OTLP_*` env vars, not from `ProviderConfig`, and are only constructed when their signal is enabled, so Prometheus-only operation dials nothing even with those vars set. Provider init in `internal/app` runs whenever `otel.enabled` OR `prometheus.enabled` is true, so Prometheus-only operation (Alloy/scrape, no collector) is a first-class mode. -- **logger.go** — `NewLogger(component, level, isJSON, otlpSampleRate)` produces a slog logger that fans out to stdout (always 100%) and the OTLP log exporter (DEBUG/INFO sampled at `otlpSampleRate`, WARN/ERROR always 100% as a non-configurable safety floor). `TraceHandler` injects `trace_id`/`span_id` from the active span when one exists. `otlpSamplerFn` is exposed (lowercase) for unit testing the per-level rate logic without driving through the slogmulti middleware. +- **logger.go** — `NewLogger(component, level, isJSON, otlpSampleRate)` produces a slog logger that fans out to stdout (always 100%) and the OTLP log exporter (DEBUG/INFO sampled at `otlpSampleRate`, WARN/ERROR always 100% as a non-configurable safety floor). `TraceHandler` injects `trace_id`/`span_id` from the active span when one exists, and `request_id` and `tenant` from the request scope; `cmd/wavehouse` wraps its boot logger in it too, so stdout carries them whether or not OTLP logs are on. `otlpSamplerFn` is exposed (lowercase) for unit testing the per-level rate logic without driving through the slogmulti middleware. +- **request.go** — The request scope: `WithRequest(ctx, id)` puts it in the context, `SetRequestTenant` fills in the tenant once it is resolved (in place, so a record logged through a context taken earlier, the panic recoverer's, carries it too) and sets it on the span in the context, and `AnnotateSpan` copies both onto the server span. `request_id` and `tenant` are the keys on records and spans alike; a record that names its own tenant keeps it. They are never metric labels. - **metrics.go** — `RegisterSystemMetrics(mqStats, pebbleStats)` registers observable gauges for the MQ's NATS connections and in-msgs (the embedded server's, or under `nats` this process's client connection), and Pebble dedupe storage stats. Both are functions read on every scrape — `mqStats` a `func() (MQStats, error)` (`mq.Broker.Stats` in production; nil skips the MQ gauges), `pebbleStats` a `func() map[string]int64` (`dedupe.Embedded.Stats`, the one instance's figures; nil, or a nil map while it is closed, skips the Pebble gauges) — this package never holds the NATS server or a store. Wired in `internal/app` after the providers are up. - **tracer.go** — W3C TraceContext propagation over message headers (`InjectHeaders` / `ExtractHeaders` on a plain `map[string][]string`, the shape NATS and HTTP headers share) — `internal/mq` injects on every publish and extracts onto the delivered `Message.Ctx` on the `Subscribe` path; no consumer reads it yet (the SSE hub bridge forwards the bytes and starts no span), and the ingest worker's `Consumer` path skips extraction entirely (see `embedded.go` above), so nothing downstream of the queue is linked to the originating request span. diff --git a/docs/src/content/docs/configuration.mdx b/docs/src/content/docs/configuration.mdx index 714a1d7ab..4c8a0c83f 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -295,7 +295,7 @@ The master switch is `otel.enabled`. When `true`, each signal (traces/metrics/lo | `otel.traces.enabled` | `WH_OTEL_TRACES_ENABLED` | `true` | Export traces via OTLP gRPC. | | `otel.traces.sample_rate` | `WH_OTEL_TRACES_SAMPLE_RATE` | `1.0` | Head-based trace sampling rate in `[0.0, 1.0]`. `1.0` exports every trace; `0.0` exports none. Defaults to 100% (matches the OpenTelemetry SDK default); lower it for high-QPS production services where collector or backend cost is a concern. Best practice is "100% at the source, downsample at the collector" via tail-based sampling. Validated at config load. | | `otel.metrics.enabled` | `WH_OTEL_METRICS_ENABLED` | `true` | Export metrics + Go runtime metrics via OTLP gRPC. Periodic reader interval is fixed at 15s. Metrics are pre-aggregated so there is no sampling knob. | -| `otel.logs.enabled` | `WH_OTEL_LOGS_ENABLED` | `true` | Export logs via OTLP gRPC. Disabling it leaves stdout with 100% of records, but from the plain boot logger: lines lose `trace_id`/`span_id`, `component`, `source` and the build fields. | +| `otel.logs.enabled` | `WH_OTEL_LOGS_ENABLED` | `true` | Export logs via OTLP gRPC. Disabling it leaves stdout with 100% of records, but from the boot logger: lines keep `trace_id`/`span_id`, `request_id` and `tenant`, and lose `component`, `source` and the build fields. | | `otel.logs.sample_rate` | `WH_OTEL_LOGS_SAMPLE_RATE` | `1.0` | OTLP export rate for `DEBUG`/`INFO` records, in `[0.0, 1.0]`. Validated at config load. `WARN` and `ERROR` records always export at 100% — dropping them silently during incidents is too dangerous to expose as a knob. **Stdout always receives 100% of records regardless of this rate** (see the scraper note above). | ### Prometheus diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index 23fe747e3..ed4311e4d 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -749,6 +749,10 @@ Set `otel.enabled: true` (or `WH_OTEL_ENABLED=true`) to export traces, metrics, WaveHouse **pushes** to an OTel collector; scraping-style pipelines (Promtail/Grafana Alloy → Loki, Vector, Fluent Bit) read stdout directly and own their own sample rates. The `otel.{traces,logs}.sample_rate` knobs apply only to the OTLP push path. Stdout always emits 100%. The logger fans out to both stdout and OTLP, so stdout output never disappears regardless of collector state. gRPC exporters are lazy, so an unreachable collector does not block startup — transient export errors are surfaced via the OTel SDK's error handler instead. +### Following one request or one tenant + +Every log record written through a request's context (as the request path logs) carries `request_id`, and `tenant` once the request's tenant is known: the `X-Tenant-ID` a `/v1` route resolves, or the `?tenant=` an ops route names, from authentication on (tenant `0` for any other ops route that names none, reads, raw SQL and the schema refresh alike; a settings reload that names none acts on every tenant and carries none). The request's server span carries the same two attributes under the same names, so one query finds both (SSE streams on `/v1/stream`, the probes, `/v1/health` and the metrics path open no server span, so their correlation is on log lines only: `/v1/stream` and `/v1/health` carry `request_id` and `tenant`, while the bare probes and the metrics path resolve no tenant and carry `request_id` alone). A record that already names a tenant keeps its own. The request id is the `X-Request-Id` the request arrived with, when it was sent once and is 1 to 128 printable ASCII characters with no spaces, and otherwise a fresh random 32-hex-character id; either way the response carries it back in `X-Request-Id`, so a log line can be joined to a proxy's access log or to a client's report. Have the proxy set the header (see [Reverse proxy](/reverse-proxy#header-and-auth-forwarding)). Neither is a label on the HTTP request metrics: with many tenants per process that would be a series per tenant per route. The ingest worker's lines are not request-scoped and carry the tenant only where they name it. + ### Pattern: Local collector (SigNoz, OTel Collector, Alloy) A local collector almost always speaks **plaintext** gRPC, but the SDK's unset default endpoint is **TLS** at `localhost:4317` — so enabling OTel alone is not enough. Point it at the collector with an explicit `http://` scheme: `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317` (or leave the endpoint unset and set `OTEL_EXPORTER_OTLP_INSECURE=true`). All three signals (traces, metrics, logs) push to the same endpoint, each over its own connection. This is the simplest setup. @@ -792,7 +796,7 @@ export OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:4317 # plaintext gRPC; DD_ The Grafana stack typically wants Prometheus-style scraping for metrics, stdout scraping for logs, and OTLP push for traces. Wire it like this: -- **Logs**: Alloy scrapes stdout via the Docker socket / file tail / k8s logs API — stdout always emits 100%. If traces also go via OTLP (`otel.enabled: true`), keep `otel.logs.enabled` on: it is what puts `trace_id`/`span_id` on the stdout lines, so Loki-to-Tempo links resolve. It also pushes logs over OTLP, so drop that stream in Alloy, or ship logs over OTLP instead of scraping stdout. +- **Logs**: Alloy scrapes stdout via the Docker socket / file tail / k8s logs API — stdout always emits 100%. If traces also go via OTLP (`otel.enabled: true`), the stdout lines carry `trace_id`/`span_id` for Loki-to-Tempo links whether or not `otel.logs.enabled` is on. Leaving it on also pushes logs over OTLP, so drop that stream in Alloy, or turn it off and scrape stdout alone. - **Traces**: Set `OTEL_EXPORTER_OTLP_ENDPOINT` to Alloy's `otelcol.receiver.otlp` listener (`http://alloy:4317`). Alloy forwards to Tempo. - **Metrics**: Set `prometheus.enabled: true`. Alloy's `prometheus.scrape` reads `http://wavehouse:8080/metrics` (or whatever port you configured). The `prometheus` block is independent of `otel.*` — you can leave `otel.enabled: false` if Alloy is only scraping (no OTLP push at all), or combine the two if traces still go via OTLP, with `otel.metrics.enabled: false` so metrics aren't also pushed. diff --git a/docs/src/content/docs/reverse-proxy.mdx b/docs/src/content/docs/reverse-proxy.mdx index 3d55c9b40..7dfe9ac4e 100644 --- a/docs/src/content/docs/reverse-proxy.mdx +++ b/docs/src/content/docs/reverse-proxy.mdx @@ -195,6 +195,7 @@ These limit the *whole* request regardless of traffic, so no keepalive extends t - **`Authorization`** — forward verbatim. WaveHouse validates a `Bearer` JWT (resolving the role from it) or an `Operator ` [operator credential](/access-control#operator-key). Most proxies forward `Authorization` unchanged, so it's the transport to prefer for the operator key — the exception to check for is an auth-terminating layer that consumes or rewrites the header (e.g. a gateway doing its own auth). - **`X-Operator-Key`** (optional) — only relevant if you present the [operator key](/access-control#operator-key) via this alias header instead of `Authorization: Operator `. Custom request headers are forwarded by default, but confirm your proxy doesn't strip it — and note nginx silently drops header names containing **underscores** (this one uses hyphens, so it's fine as named). The `Authorization` form needs none of this. - **`X-Tenant-ID`** — **strip it at the edge unless you are deliberately using it.** It selects the [tenant](/deployment#multi-tenant-deployments) (absent means tenant `0`), and a settings directory that isn't nested defines no tenant other than `0`, so a gateway or mesh that stamps `X-Tenant-ID` with any other value turns every `/v1` route outside `/v1/ops/*` into a `404` (a `400` when the value is not a tenant id at all, a dotted hostname, say) — the SDK's `/v1/health` reachability ping included. It is resolved before authentication and never derived from the token, so a proxy that owns tenant selection must **replace** a client-supplied value, not add to it: two `X-Tenant-ID` lines are refused with `400`. That is a stricter version of the append-vs-replace hazard of `Content-Type` below — here even two agreeing lines are refused. Over a [nested settings directory](/deployment#the-nested-settings-directory) the proxy owns the tenant: derive it from the host or path and set it on **every** request, the CORS `OPTIONS` preflight included — the browser sends no custom header there, and WaveHouse reads the preflight's tenant the same way ([per-tenant CORS](/deployment#multi-tenant-deployments)). The name is hyphenated, so nginx's underscore rule does not apply. +- **`X-Request-Id`** (optional) — set it to your proxy's request id (nginx `$request_id`, Envoy's `x-request-id`) and WaveHouse logs and traces the request under the same id and returns it on the response, so its log lines join your access log; **replace** a client-supplied value rather than appending, since two lines are ignored and a fresh id minted. An id longer than 128 characters, or with spaces or non-ASCII characters, is replaced the same way ([details](/deployment#following-one-request-or-one-tenant)). - **`X-Forwarded-For` / `X-Forwarded-Proto` / `Host`** — set these for your own logs and any upstream that reads them. WaveHouse does not currently derive a client IP from `X-Forwarded-For` (see the caution below); forwarding it is good hygiene and is what the trusted-proxy client-IP work ([#333](https://github.com/Wave-RF/WaveHouse/issues/333)) will consume. :::caution[Don't expose `:8080` directly] diff --git a/internal/api/correlation.go b/internal/api/correlation.go new file mode 100644 index 000000000..5b24fd541 --- /dev/null +++ b/internal/api/correlation.go @@ -0,0 +1,94 @@ +package api + +import ( + "crypto/rand" + "encoding/hex" + "net/http" + "net/url" + + "github.com/Wave-RF/WaveHouse/internal/observability" + "github.com/Wave-RF/WaveHouse/internal/tenant" +) + +// requestIDHeader carries the request id both ways: an upstream proxy's id +// is kept, and the response names the id the request was logged under. +const requestIDHeader = "X-Request-Id" + +// maxRequestIDLen bounds a caller's id, which is copied onto every log +// record and span of its request. +const maxRequestIDLen = 128 + +// requestID gives every request an id and puts it in the request scope the +// log handler and AnnotateSpan read (observability.WithRequest). An incoming +// X-Request-Id is kept when it is sent once and is 1–128 printable ASCII +// characters without spaces; otherwise the request gets a fresh random id, +// so a repeated header cannot pick which value the logs carry. +func requestID(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var id string + if values := r.Header.Values(requestIDHeader); len(values) == 1 && validRequestID(values[0]) { + id = values[0] + } else { + id = newRequestID() + } + w.Header().Set(requestIDHeader, id) + next.ServeHTTP(w, r.WithContext(observability.WithRequest(r.Context(), id))) + }) +} + +func validRequestID(id string) bool { + if id == "" || len(id) > maxRequestIDLen { + return false + } + for i := range len(id) { + if id[i] < '!' || id[i] > '~' { + return false + } + } + return true +} + +// newRequestID is 128 random bits in hex, the shape of a trace id. Unlike +// chi's middleware.RequestID it names no host, since it goes back to the +// caller. +func newRequestID() string { + var b [16]byte + _, _ = rand.Read(b[:]) // never fails (crypto/rand.Read) + return hex.EncodeToString(b[:]) +} + +// opsReloadPath is the one ops route that, naming no tenant, acts on every +// tenant rather than on tenant.Default. +const opsReloadPath = "/v1/ops/settings/reload" + +// opsRequestTenant records the tenant an ops request addresses before +// authentication, so the operator-key audit line and an admin-gate denial +// carry it: the one ?tenant= names, or tenant.Default when it names none, +// except on the reload, which then acts on every tenant. It writes nothing: +// a query opsTenant refuses is left without a tenant, and its handler +// answers the 400. +func opsRequestTenant(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if params, err := url.ParseQuery(r.URL.RawQuery); err == nil { + values, named := params[opsTenantParam] + switch { + case !named && r.URL.Path != opsReloadPath: + observability.SetRequestTenant(r.Context(), tenant.Default.String()) + case len(values) == 1: + if id, err := tenant.Parse(values[0]); err == nil { + observability.SetRequestTenant(r.Context(), id.String()) + } + } + } + next.ServeHTTP(w, r) + }) +} + +// annotateSpan copies the request scope onto the server span, which the +// otelhttp middleware starts after requestID has run. +func annotateSpan(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + observability.AnnotateSpan(r.Context()) + next.ServeHTTP(w, r) + }) +} diff --git a/internal/api/correlation_test.go b/internal/api/correlation_test.go new file mode 100644 index 000000000..a157d4cb0 --- /dev/null +++ b/internal/api/correlation_test.go @@ -0,0 +1,274 @@ +package api + +import ( + "bytes" + "encoding/json" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + + "github.com/Wave-RF/WaveHouse/internal/observability" + "github.com/Wave-RF/WaveHouse/internal/policy" + "github.com/Wave-RF/WaveHouse/internal/stream" + "github.com/Wave-RF/WaveHouse/internal/tenant" + "github.com/Wave-RF/WaveHouse/internal/testutil" + "github.com/Wave-RF/WaveHouse/internal/testutil/logtest" +) + +func TestRequestID(t *testing.T) { + t.Parallel() + tests := []struct { + name string + header []string // every X-Request-Id line sent + keep bool + }{ + {name: "none sent"}, + {name: "empty", header: []string{""}}, + {name: "kept", header: []string{"proxy-7f3a/42"}, keep: true}, + {name: "longest kept", header: []string{strings.Repeat("a", maxRequestIDLen)}, keep: true}, + {name: "too long", header: []string{strings.Repeat("a", maxRequestIDLen+1)}}, + {name: "space", header: []string{"a b"}}, + {name: "control character", header: []string{"a\tb"}}, + {name: "non-ASCII", header: []string{"idé"}}, + {name: "sent twice", header: []string{"one", "two"}}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + var seen string + h := requestID(http.HandlerFunc(func(_ http.ResponseWriter, r *http.Request) { + seen = observability.RequestID(r.Context()) + })) + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil) + for _, v := range tt.header { + req.Header.Add(requestIDHeader, v) + } + w := httptest.NewRecorder() + h.ServeHTTP(w, req) + + got := w.Header().Get(requestIDHeader) + assert.Equal(t, seen, got, "the response names the id the request was logged under") + if tt.keep { + assert.Equal(t, tt.header[0], got) + } else { + assert.Regexp(t, `^[0-9a-f]{32}$`, got) + } + }) + } +} + +func TestRequestID_FreshPerRequest(t *testing.T) { + t.Parallel() + h := requestID(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})) + ids := map[string]bool{} + for range 3 { + w := httptest.NewRecorder() + h.ServeHTTP(w, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)) + ids[w.Header().Get(requestIDHeader)] = true + } + assert.Len(t, ids, 3) +} + +func TestAnnotateSpanMiddleware(t *testing.T) { + t.Parallel() + rec := tracetest.NewSpanRecorder() + tracer := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)).Tracer("test") + + h := requestID(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ctx, span := tracer.Start(r.Context(), "server") + annotateSpan(http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})).ServeHTTP(w, r.WithContext(ctx)) + span.End() + })) + req := httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil) + req.Header.Set(requestIDHeader, "req-1") + h.ServeHTTP(httptest.NewRecorder(), req) + + require.Len(t, rec.Ended(), 1) + assert.Contains(t, rec.Ended()[0].Attributes(), attribute.String(observability.RequestIDKey, "req-1")) +} + +// captureCorrelated routes the default logger through the production +// TraceHandler into a buffer until the test ends. Like logtest.Capture, the +// caller must not run in parallel. +func captureCorrelated(t *testing.T) *logtest.Buffer { + t.Helper() + buf := &logtest.Buffer{} + prev := slog.Default() + slog.SetDefault(slog.New(&observability.TraceHandler{Handler: slog.NewJSONHandler(buf, nil)})) + t.Cleanup(func() { slog.SetDefault(prev) }) + return buf +} + +// records decodes the captured JSON lines whose msg is msg. +func records(t *testing.T, buf *logtest.Buffer, msg string) []map[string]any { + t.Helper() + var out []map[string]any + for line := range bytes.SplitSeq(bytes.TrimSpace(buf.Bytes()), []byte("\n")) { + if len(line) == 0 { + continue + } + var rec map[string]any + require.NoError(t, json.Unmarshal(line, &rec)) + if rec["msg"] == msg { + out = append(out, rec) + } + } + return out +} + +// A tenant route's log lines, the authentication middleware's and the +// handler's authorization denial alike, carry the request's id and tenant. +func TestNewRouter_TenantRouteLogsCarryRequestAndTenant(t *testing.T) { + buf := captureCorrelated(t) + reg := testRegistry(t) + ingest := NewIngestHandler(fixedRegistry(reg), &testutil.MockPublisher{}) + ingest.PolicySource = staticPolicy(&policy.Policy{}) // grants nothing + router := NewRouter(Dependencies{ + Tenants: testTenants(), + Ingest: ingest, + Query: &QueryHandler{}, + SSE: NewStreamHandler(stream.NewHub(nil, nil, nil), nil), + Health: &HealthHandler{}, + Version: NewVersionHandler("test", "test", "test"), + Schema: schemaHandlerOver(reg, testTenants()), + AuthMW: func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + slog.InfoContext(r.Context(), "auth ran") + next.ServeHTTP(w, r) + }) + }, + PolicySource: policy.Static(&policy.Policy{}), + }) + + req := httptest.NewRequestWithContext(t.Context(), http.MethodPost, "/v1/ingest?table=clicks", strings.NewReader(`{"page":"/"}`)) + req.Header.Set("Content-Type", "application/json") + req.Header.Set(tenant.Header, tenant.Default.String()) + req.Header.Set(requestIDHeader, "req-ingest") + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + require.Equal(t, http.StatusForbidden, w.Code, "body: %s", w.Body.String()) + assert.Equal(t, "req-ingest", w.Header().Get(requestIDHeader)) + + for _, msg := range []string{"auth ran", "authorization denied"} { + recs := records(t, buf, msg) + require.Len(t, recs, 1, msg) + assert.Equal(t, "req-ingest", recs[0][observability.RequestIDKey], msg) + assert.Equal(t, tenant.Default.String(), recs[0][observability.TenantKey], msg) + } +} + +// An ops route names its tenant in ?tenant=, or serves tenant 0 without one; +// either way its lines carry it from the handler's read on. +func TestOpsStore_StampsTheTenant(t *testing.T) { + for _, target := range []string{"/v1/ops/schema?tenant=0", "/v1/ops/schema"} { + t.Run(target, func(t *testing.T) { + buf := captureCorrelated(t) + req := httptest.NewRequestWithContext(observability.WithRequest(t.Context(), "req-ops"), http.MethodGet, target, nil) + _, ok := opsStore(httptest.NewRecorder(), req, testTenants()) + require.True(t, ok) + slog.InfoContext(req.Context(), "ops read") + + recs := records(t, buf, "ops read") + require.Len(t, recs, 1) + assert.Equal(t, tenant.Default.String(), recs[0][observability.TenantKey]) + }) + } +} + +// Every response names its request id, the probes' included. +func TestNewRouter_EchoesRequestID(t *testing.T) { + t.Parallel() + var sawStore []bool + router := tenantProbeRouter(t, &sawStore) + for _, path := range []string{"/livez", "/v1/health", "/no-such-route"} { + w := httptest.NewRecorder() + router.ServeHTTP(w, httptest.NewRequestWithContext(t.Context(), http.MethodGet, path, nil)) + assert.Regexp(t, `^[0-9a-f]{32}$`, w.Header().Get(requestIDHeader), path) + } +} + +// The dead-letter read serves tenant 0 when ?tenant= names none, and its +// lines say so. +func TestDLQStats_StampsTheDefaultTenant(t *testing.T) { + buf := captureCorrelated(t) + req := httptest.NewRequestWithContext(observability.WithRequest(t.Context(), "req-dlq"), http.MethodGet, "/v1/ops/dlq/stats", nil) + NewDLQHandler(&testutil.MockDeadLetterStats{}).Stats(httptest.NewRecorder(), req) + slog.InfoContext(req.Context(), "dlq read") + + recs := records(t, buf, "dlq read") + require.Len(t, recs, 1) + assert.Equal(t, tenant.Default.String(), recs[0][observability.TenantKey]) +} + +// An ops request's tenant is on its lines from authentication on, so the +// operator-key audit line carries it: the one ?tenant= names, tenant 0 for a +// read naming none, and none for a reload naming none (it acts on every +// tenant) or for a query opsTenant refuses. +func TestNewRouter_OpsLogsCarryTheTenant(t *testing.T) { + tests := []struct { + query string + wantRead any + wantReload any + }{ + {query: "?tenant=acme", wantRead: "acme", wantReload: "acme"}, + {query: "", wantRead: tenant.Default.String(), wantReload: nil}, + {query: "?tenant=acme&tenant=globex", wantRead: nil, wantReload: nil}, + {query: "?tenant=..%2Fetc", wantRead: nil, wantReload: nil}, + {query: "?tenant=acme;x=1", wantRead: nil, wantReload: nil}, + } + auth := func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + slog.InfoContext(r.Context(), "auth ran", "path", r.URL.Path) + next.ServeHTTP(w, r) + }) + } + for _, tt := range tests { + t.Run(tt.query, func(t *testing.T) { + buf := captureCorrelated(t) + reg := testRegistry(t) + api := NewRouter(Dependencies{ + Tenants: testTenants(), + Ingest: NewIngestHandler(fixedRegistry(reg), &testutil.MockPublisher{}), + Query: &QueryHandler{}, + SSE: NewStreamHandler(stream.NewHub(nil, nil, nil), nil), + Health: &HealthHandler{}, + Version: NewVersionHandler("test", "test", "test"), + Schema: schemaHandlerOver(reg, testTenants()), + Settings: &SettingsHandler{}, + AuthMW: auth, + PolicySource: policy.Static(&policy.Policy{}), + }) + ops := NewOpsRouter(OpsDependencies{ + Health: &HealthHandler{}, + Version: NewVersionHandler("test", "test", "test"), + Settings: &SettingsHandler{}, + AuthMW: auth, + }) + serve := func(h http.Handler, method, path string) { + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequestWithContext(t.Context(), method, path+tt.query, nil)) + } + serve(api, http.MethodGet, "/v1/ops/schema") + serve(api, http.MethodPost, opsReloadPath) + serve(ops, http.MethodPost, opsReloadPath) + + recs := records(t, buf, "auth ran") + require.Len(t, recs, 3) + for _, rec := range recs { + want := tt.wantRead + if rec["path"] == opsReloadPath { + want = tt.wantReload + } + assert.Equal(t, want, rec[observability.TenantKey], rec["path"]) + assert.NotEmpty(t, rec[observability.RequestIDKey]) + } + }) + } +} diff --git a/internal/api/dlq.go b/internal/api/dlq.go index 267d2dc47..ad1d3aa12 100644 --- a/internal/api/dlq.go +++ b/internal/api/dlq.go @@ -7,6 +7,7 @@ import ( "net/http" "github.com/Wave-RF/WaveHouse/internal/mq" + "github.com/Wave-RF/WaveHouse/internal/observability" "github.com/Wave-RF/WaveHouse/internal/tenant" ) @@ -33,6 +34,7 @@ func (h *DLQHandler) Stats(w http.ResponseWriter, r *http.Request) { } if !named { id = tenant.Default + observability.SetRequestTenant(r.Context(), id.String()) } counts, err := h.Counts.DeadLetterCounts(r.Context(), id, r.URL.Query().Get("table")) if err != nil { diff --git a/internal/api/router.go b/internal/api/router.go index 53489ca7d..b7b8c5a61 100644 --- a/internal/api/router.go +++ b/internal/api/router.go @@ -120,6 +120,7 @@ func NewRouter(deps Dependencies) http.Handler { if deps.Tenants != nil && deps.Tenants.Nested() { adminPolicy = nil } + r.Use(opsRequestTenant) r.Use(deps.AuthMW) r.Use(refuseUnverifiable) r.Use(RequireAdmin(adminPolicy)) @@ -163,7 +164,7 @@ func NewRouter(deps Dependencies) http.Handler { func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, version *VersionHandler, metrics http.Handler, metricsPath string) chi.Router { r := chi.NewRouter() - r.Use(middleware.RequestID) + r.Use(requestID) // 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 @@ -210,6 +211,7 @@ func newProbeRouter(origins func(*http.Request) []string, health *HealthHandler, otelhttp.NewMiddleware("wavehouse-api")(next).ServeHTTP(w, r) }) }) + r.Use(annotateSpan) // Public endpoints. /livez and /readyz are the canonical probe names // (current Kubernetes convention — the kube-apiserver split that replaced @@ -257,6 +259,7 @@ func NewOpsRouter(deps OpsDependencies) http.Handler { r := newProbeRouter(nil, deps.Health, deps.Version, deps.MetricsHandler, deps.MetricsPath) if deps.Settings != nil { r.Route("/v1/ops", func(r chi.Router) { + r.Use(opsRequestTenant) r.Use(deps.AuthMW) r.Use(refuseUnverifiable) r.Use(RequireAdmin(nil)) diff --git a/internal/api/tenant.go b/internal/api/tenant.go index 95fe9a0ff..f5512cf44 100644 --- a/internal/api/tenant.go +++ b/internal/api/tenant.go @@ -7,6 +7,7 @@ import ( "net/http" "net/url" + "github.com/Wave-RF/WaveHouse/internal/observability" "github.com/Wave-RF/WaveHouse/internal/policy" "github.com/Wave-RF/WaveHouse/internal/settings" "github.com/Wave-RF/WaveHouse/internal/tenant" @@ -98,6 +99,7 @@ func opsTenant(w http.ResponseWriter, r *http.Request) (id tenant.ID, named, ok writeJSONError(w, http.StatusBadRequest, "invalid ?"+opsTenantParam+": "+err.Error()) return "", false, false } + observability.SetRequestTenant(r.Context(), id.String()) return id, true, true } @@ -111,6 +113,7 @@ func opsStore(w http.ResponseWriter, r *http.Request, tenants *settings.Registry } if !named { id = tenant.Default + observability.SetRequestTenant(r.Context(), id.String()) } return resolveStore(w, tenants, id) } @@ -151,6 +154,7 @@ func TenantMW(tenants *settings.Registry) func(http.Handler) http.Handler { writeJSONError(w, http.StatusBadRequest, err.Error()) return } + observability.SetRequestTenant(r.Context(), id.String()) store, ok := resolveStore(w, tenants, id) if !ok { return diff --git a/internal/auth/auth.go b/internal/auth/auth.go index 553206f68..21f777afc 100644 --- a/internal/auth/auth.go +++ b/internal/auth/auth.go @@ -497,12 +497,11 @@ func middleware(current func(context.Context) (tenant.ID, *verifier), operatorKe // privileged credential in the system — full data-plane + // admin, honored even when the policy is wiped — so its use // must be visible in production logs (Info+), mirroring the - // WARN emitted on an authz denial. Correlation fields - // (request_id, and eventually the trusted-proxy client IP) are - // deliberately NOT stamped per-call-site — they belong in the - // global TraceHandler (internal/observability) so every log line - // gets them uniformly; tracked in #333. When OTel is enabled this - // line already carries trace_id/span_id from that handler. + // WARN emitted on an authz denial. Correlation fields are + // deliberately NOT stamped per-call-site: the global + // TraceHandler (internal/observability) adds request_id, + // tenant and, when a span is active, trace_id/span_id to + // every line; the trusted-proxy client IP is #333. slog.LogAttrs(r.Context(), slog.LevelInfo, "operator key authenticated request", slog.String("path", r.URL.Path), slog.String("method", r.Method), @@ -524,7 +523,8 @@ func middleware(current func(context.Context) (tenant.ID, *verifier), operatorKe // other unauthenticated caller. WARN + a counter because a // mismatch is a strong probing/brute-force signal: nobody // sends a wrong operator key by accident. Same correlation-field - // deferral (request_id / client IP → #333) as the audit line above. + // handling as the audit line above: request_id and tenant from + // TraceHandler, the client IP in #333. operatorKeyFailures.Add(r.Context(), 1) slog.LogAttrs(r.Context(), slog.LevelWarn, "operator key authentication failed", slog.String("path", r.URL.Path), diff --git a/internal/observability/logger.go b/internal/observability/logger.go index 567735ff4..2123c67ee 100644 --- a/internal/observability/logger.go +++ b/internal/observability/logger.go @@ -12,13 +12,16 @@ import ( "go.opentelemetry.io/otel/trace" ) -// TraceHandler intercepts every log call to inject OpenTelemetry IDs +// TraceHandler intercepts every log call to inject OpenTelemetry IDs and the +// request's correlation fields. type TraceHandler struct { slog.Handler } -// Handle pulls the trace_id and span_id from the context if they exist +// Handle pulls the trace_id and span_id from the context if they exist, and +// request_id and tenant from the request scope (WithRequest) when there is one. func (h *TraceHandler) Handle(ctx context.Context, r slog.Record) error { + addRequestAttrs(ctx, &r) if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() { r.AddAttrs( slog.String("trace_id", span.SpanContext().TraceID().String()), diff --git a/internal/observability/request.go b/internal/observability/request.go new file mode 100644 index 000000000..2c0ef3f75 --- /dev/null +++ b/internal/observability/request.go @@ -0,0 +1,98 @@ +package observability + +import ( + "context" + "log/slog" + "sync/atomic" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/trace" +) + +// The keys a request's correlation goes under, the same on a log record and +// on a span, so one query finds both. +const ( + RequestIDKey = "request_id" + TenantKey = "tenant" +) + +// requestScope is what a request's log records and spans say about it: its +// id, known when it arrives, and its tenant, known only once a later +// middleware resolves it. The tenant is filled in place, so a record logged +// through a context taken before resolution (the panic recoverer's, say) +// still carries it. +type requestScope struct { + id string + tenant atomic.Pointer[string] +} + +type requestScopeKey struct{} + +// WithRequest returns ctx carrying a request scope with the given id. Every +// record TraceHandler handles under it, or under a context derived from it, +// carries request_id, and tenant once SetRequestTenant has run. +func WithRequest(ctx context.Context, requestID string) context.Context { + return context.WithValue(ctx, requestScopeKey{}, &requestScope{id: requestID}) +} + +func scopeFrom(ctx context.Context) *requestScope { + s, _ := ctx.Value(requestScopeKey{}).(*requestScope) + return s +} + +// RequestID returns the id WithRequest stored, or "" outside a request. +func RequestID(ctx context.Context) string { + if s := scopeFrom(ctx); s != nil { + return s.id + } + return "" +} + +// SetRequestTenant records the request's tenant on its scope and on the span +// in ctx. Outside a request it still annotates the span. +func SetRequestTenant(ctx context.Context, tenant string) { + if s := scopeFrom(ctx); s != nil { + s.tenant.Store(&tenant) + } + trace.SpanFromContext(ctx).SetAttributes(attribute.String(TenantKey, tenant)) +} + +// AnnotateSpan copies the request scope in ctx onto the span in ctx. Call it +// once the server span exists: the scope is set up before it. +func AnnotateSpan(ctx context.Context) { + s := scopeFrom(ctx) + if s == nil { + return + } + span := trace.SpanFromContext(ctx) + if !span.IsRecording() { + return + } + attrs := []attribute.KeyValue{attribute.String(RequestIDKey, s.id)} + if t := s.tenant.Load(); t != nil { + attrs = append(attrs, attribute.String(TenantKey, *t)) + } + span.SetAttributes(attrs...) +} + +// addRequestAttrs stamps r with the request scope in ctx. A record that +// already names a tenant keeps its own: it may be about another one. +func addRequestAttrs(ctx context.Context, r *slog.Record) { + s := scopeFrom(ctx) + if s == nil { + return + } + r.AddAttrs(slog.String(RequestIDKey, s.id)) + t := s.tenant.Load() + if t == nil { + return + } + named := false + r.Attrs(func(a slog.Attr) bool { + named = a.Key == TenantKey + return !named + }) + if !named { + r.AddAttrs(slog.String(TenantKey, *t)) + } +} diff --git a/internal/observability/request_test.go b/internal/observability/request_test.go new file mode 100644 index 000000000..d3936f5a1 --- /dev/null +++ b/internal/observability/request_test.go @@ -0,0 +1,163 @@ +package observability + +import ( + "bytes" + "context" + "encoding/json" + "log/slog" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" +) + +// logLines runs log against a TraceHandler over a JSON handler and returns +// each record decoded. +func logLines(t *testing.T, log func(*slog.Logger)) []map[string]any { + t.Helper() + var buf bytes.Buffer + log(slog.New(&TraceHandler{Handler: slog.NewJSONHandler(&buf, nil)})) + var lines []map[string]any + for line := range strings.SplitSeq(strings.TrimSpace(buf.String()), "\n") { + var rec map[string]any + require.NoError(t, json.Unmarshal([]byte(line), &rec)) + lines = append(lines, rec) + } + return lines +} + +// countKey counts the occurrences of key in a JSON line, which a decoded map +// would collapse. +func countKey(t *testing.T, log func(*slog.Logger), key string) int { + t.Helper() + var buf bytes.Buffer + log(slog.New(&TraceHandler{Handler: slog.NewJSONHandler(&buf, nil)})) + return strings.Count(buf.String(), `"`+key+`":`) +} + +func TestTraceHandler_RequestScope(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + ctx func() context.Context + attrs []any + wantID any + wantTenant any + }{ + { + name: "outside a request", + ctx: context.Background, + wantID: nil, + wantTenant: nil, + }, + { + name: "before the tenant is resolved", + ctx: func() context.Context { return WithRequest(context.Background(), "req-1") }, + wantID: "req-1", + wantTenant: nil, + }, + { + name: "after the tenant is resolved", + ctx: func() context.Context { + ctx := WithRequest(context.Background(), "req-2") + SetRequestTenant(ctx, "acme") + return ctx + }, + wantID: "req-2", + wantTenant: "acme", + }, + { + name: "a record naming its own tenant keeps it", + ctx: func() context.Context { + ctx := WithRequest(context.Background(), "req-3") + SetRequestTenant(ctx, "acme") + return ctx + }, + attrs: []any{TenantKey, "globex"}, + wantID: "req-3", + wantTenant: "globex", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + lines := logLines(t, func(l *slog.Logger) { l.InfoContext(tt.ctx(), "hello", tt.attrs...) }) + require.Len(t, lines, 1) + assert.Equal(t, tt.wantID, lines[0][RequestIDKey]) + assert.Equal(t, tt.wantTenant, lines[0][TenantKey]) + }) + } +} + +// The tenant is resolved after the context an outer middleware logs with was +// taken; the record still carries it. +func TestTraceHandler_TenantSetAfterTheContextWasTaken(t *testing.T) { + t.Parallel() + outer := WithRequest(context.Background(), "req-1") + inner, cancel := context.WithCancel(outer) + defer cancel() + SetRequestTenant(inner, "acme") + + lines := logLines(t, func(l *slog.Logger) { l.ErrorContext(outer, "panic recovered") }) + require.Len(t, lines, 1) + assert.Equal(t, "acme", lines[0][TenantKey]) +} + +func TestTraceHandler_NoDuplicateTenant(t *testing.T) { + t.Parallel() + ctx := WithRequest(context.Background(), "req-1") + SetRequestTenant(ctx, "acme") + n := countKey(t, func(l *slog.Logger) { l.InfoContext(ctx, "hello", TenantKey, "acme", "table", "clicks") }, TenantKey) + assert.Equal(t, 1, n) +} + +func TestRequestID(t *testing.T) { + t.Parallel() + assert.Empty(t, RequestID(context.Background())) + assert.Equal(t, "req-1", RequestID(WithRequest(context.Background(), "req-1"))) +} + +func TestAnnotateSpan(t *testing.T) { + t.Parallel() + rec := tracetest.NewSpanRecorder() + tracer := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)).Tracer("test") + + // No scope: the span is left alone. + ctx, span := tracer.Start(context.Background(), "bare") + AnnotateSpan(ctx) + span.End() + + ctx, span = tracer.Start(WithRequest(context.Background(), "req-1"), "server") + AnnotateSpan(ctx) + SetRequestTenant(ctx, "acme") + span.End() + + ended := rec.Ended() + require.Len(t, ended, 2) + assert.Empty(t, ended[0].Attributes()) + assert.ElementsMatch(t, []attribute.KeyValue{ + attribute.String(RequestIDKey, "req-1"), + attribute.String(TenantKey, "acme"), + }, ended[1].Attributes()) +} + +// A tenant resolved before the span starts reaches it through AnnotateSpan. +func TestAnnotateSpan_TenantAlreadyResolved(t *testing.T) { + t.Parallel() + rec := tracetest.NewSpanRecorder() + tracer := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(rec)).Tracer("test") + + ctx := WithRequest(context.Background(), "req-1") + SetRequestTenant(ctx, "acme") + ctx, span := tracer.Start(ctx, "server") + AnnotateSpan(ctx) + span.End() + + require.Len(t, rec.Ended(), 1) + assert.Contains(t, rec.Ended()[0].Attributes(), attribute.String(TenantKey, "acme")) +} diff --git a/scripts/ci/remote-ci.test.sh b/scripts/ci/remote-ci.test.sh index aed0f98fc..2ee2b9daf 100755 --- a/scripts/ci/remote-ci.test.sh +++ b/scripts/ci/remote-ci.test.sh @@ -123,7 +123,10 @@ run() { dog=$! wait "$pid" rc=$? - kill "$dog" 2> /dev/null + # KILL, not TERM: a run that is refused exits before the watchdog has been + # scheduled once, and a subshell signaled that early still handles it with + # this script's EXIT trap, which removes the scratch directory. + kill -KILL "$dog" 2> /dev/null return "$rc" }