From 4b3965092849ee8ebf5b1091945a63152eed3852 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 19:32:42 +0000 Subject: [PATCH 1/4] feat(observability): identify the process in telemetry --- AGENTS.md | 4 +- CHANGELOG.md | 2 + cmd/wavehouse/main.go | 4 + config.yaml | 4 +- docs/src/content/docs/architecture.md | 5 +- docs/src/content/docs/configuration.mdx | 8 +- docs/src/content/docs/deployment.md | 12 ++- internal/app/wire.go | 6 +- internal/app/wire_identity.go | 21 ++++ internal/app/wire_identity_test.go | 26 +++++ internal/config/config.go | 6 +- internal/observability/identity.go | 102 +++++++++++++++++++ internal/observability/identity_test.go | 127 ++++++++++++++++++++++++ internal/observability/provider.go | 19 ++-- internal/observability/provider_test.go | 4 +- internal/testutil/otlp.go | 2 +- tests/integration/otel_test.go | 9 +- 17 files changed, 329 insertions(+), 32 deletions(-) create mode 100644 internal/app/wire_identity.go create mode 100644 internal/app/wire_identity_test.go create mode 100644 internal/observability/identity.go create mode 100644 internal/observability/identity_test.go diff --git a/AGENTS.md b/AGENTS.md index 36b16d15..1c98d3eb 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,14 +34,14 @@ Twenty internal packages under `internal/` (plus `internal/testutil/` for shared - **`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) - **`chconn/`** — `Pools`, one `Manager` (a `driver.Conn`) per distinct `Identity{Addr, Database, Username, Password, TLS}` tuple among the served tenants, reconciled from the settings registry's `AfterAdopt` after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): tenants naming one tuple share its pool, sized to their largest `max_open_conns`/`max_idle_conns`; a tenant whose tuple changed is repointed; a tuple no tenant names is released after the longest `query_timeout` among the tenants it had (never dials; a resize swaps the connection with the same grace). The boot config's `clickhouse.max_total_conns` bounds the open pools' `max_open_conns` together: boot refuses naming sum and ceiling; at a reload a resize above it keeps the pool's size, and a tuple that cannot be opened (the ceiling, an unreadable certificate, or options the driver refuses) leaves its tenants on the pool they had or on none — logged, retried by the next reload. Every consumer resolves its tenant's pool per call: `For` (nil for a tenant on no pool, a `503`), `Target` (the tenant's own HTTP wiring over its pool's TLS config), `SharingTables`, `Ping` (every pool at once, ready at the first answer). `HTTPClients` keeps one `http.Client` per TLS config. `Classify` (`errclass.go`) says what a failed ClickHouse request means for the request — `Unavailable`, `Denied`, `Rejected` (any unlisted exception code: the server read it and refused it), or `Unknown` (no code, no recognizable transport failure) — over the driver's error types and the HTTP interface's `HTTPError`; the ingest worker and the query handlers (`api/ch_errors.go` `writeCHError`, [#403](https://github.com/Wave-RF/WaveHouse/issues/403), [#271](https://github.com/Wave-RF/WaveHouse/issues/271)) both use it - **`chsql/`** — dependency-free ClickHouse SQL helpers shared by `query`/`policy` (avoids an import cycle): `QuoteIdent` (backtick-quote every identifier) + `BindUnsafe` (reject names with a literal `?`) -- **`config/`** — YAML + env var config loading (cleanenv); strict on both sides (undeclared YAML key, unbound `WH_*` variable) and probes `data_dir` writability when a selected backend keeps state there (`NeedsDataDir`); `backends.go` holds each layer's `.backend` (the in-process value by default; `mq.backend` also takes `nats`, with its `mq.nats` sub-block of file-path-only credentials; `coord.backend` takes `nats`, whose `coord.nats` block names only the lease bucket and rides `mq.nats`'s connection (both blocks, their rules and warnings are `mq_nats.go`); `cache.backend` takes `redis`, whose sub-block is `cache_redis.go`; and `dedupe.backend` takes `dynamodb`, with its `dedupe.dynamodb` sub-block) and `Warnings`, the valid combinations boot logs at `WARN`; `config.go` holds `roles` (`Has(Role)`) and `instance_id`, and `Validate` refuses a role split the backends cannot serve (any split over the embedded MQ; `api` without `ingest`, or the reverse, over a local cache; `coord.backend=nats` without `mq.backend=nats`; `mq.backend=nats` with `coord.backend=local` in a process running `ingest`; a process running only `sweeper` under `mq.backend=nats`) — boot is the validator, there is no dry run +- **`config/`** — YAML + env var config loading (cleanenv); strict on both sides (undeclared YAML key, unbound `WH_*` variable) and probes `data_dir` writability when a selected backend keeps state there (`NeedsDataDir`); `backends.go` holds each layer's `.backend` (the in-process value by default; `mq.backend` also takes `nats`, with its `mq.nats` sub-block of file-path-only credentials; `coord.backend` takes `nats`, whose `coord.nats` block names only the lease bucket and rides `mq.nats`'s connection (both blocks, their rules and warnings are `mq_nats.go`); `cache.backend` takes `redis`, whose sub-block is `cache_redis.go`; and `dedupe.backend` takes `dynamodb`, with its `dedupe.dynamodb` sub-block) and `Warnings`, the valid combinations boot logs at `WARN`; `config.go` holds `roles` (`Has(Role)`) and `instance_id` (on every log line, and `service.instance.id` in the telemetry), and `Validate` refuses a role split the backends cannot serve (any split over the embedded MQ; `api` without `ingest`, or the reverse, over a local cache; `coord.backend=nats` without `mq.backend=nats`; `mq.backend=nats` with `coord.backend=local` in a process running `ingest`; a process running only `sweeper` under `mq.backend=nats`) — boot is the validator, there is no dry run - **`coord/`** — leases for work that must run in one process at a time (`Observer.Held` reads whether one is held without campaigning): `Coordinator.TryAcquire(ctx, name)` → a `Term` (fencing `Token`, strictly increasing per name; `Done`/`Err`, `ErrLost` on loss; `Resign`), `ErrHeld` while another holder's — or this coordinator's own — term is live; `RunElected` runs a loop only while holding its lease, resigning when the loop returns and campaigning again every `RetryPeriod`. `Local` is the in-process implementation (first taker wins, never expires; `Peer` is a second handle over the same table for tests); every implementation runs `coordtest.Conformance`. Imports only the standard library, so a distributed backend lives beside its connection: `coord.backend: nats` is `internal/mq/lease.go` (`ExternalNATS.Leases`), a key per lease in the operator's KV bucket, the KV revision as the fencing token, and expiry judged on the candidate's own clock (the same revision seen unchanged for 15s), never by a server TTL. `internal/app`'s `wireCoord` opens the one `coord.backend` selects and the sweeper runs through `RunElected` under the `sweeper` lease - **`dedupe/`** — `Deduplicator` interface (two-phase `Reserve`/`Commit`/`Release` over `Key{Table, ID}`; every backend passes the `dedupetest` conformance suite) → `Embedded` (Pebble: every tenant's seen ids in one instance at `data_dir/pebble`, each key led by its tenant and table, pending claims in memory, committed ids stored with their expiry and deleted by an hourly background sweep along with the version-0 keys from before the table joined the key, open while any tenant's store is — the layout is the implementation's call, and the wiring hands it `data_dir` once; its `Stats` feed the system gauges) or `Dynamo` (one shared DynamoDB table, conditional `PutItem` claims; conformance-tested against dynamodb-local, selected by `dedupe.backend: dynamodb`; boot checks the table and never creates it outside dynamodb-local), wrapped by `Managed` whose open/closed state follows the hot-reloadable `dedupe.enabled` in the settings directory's `config.json`; `Stores` holds one `Managed` per tenant, built through a `Factory` (`func(tenant.ID) *Managed`, `Embedded.Tenant` or, gated on the table check (`Factory.Gated`), `Dynamo.Tenant` in production; `Managed` opens its store through a function, so every backend gets the same switch), and reconciled from the registry's `AfterAdopt` hook — open exactly when the tenant is served with its switch on, closed with its seen ids kept otherwise ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) stories 7 and 3) - **`discovery/`** — `SchemaRegistry`, one per served tenant over a `Source` read once per refresh — the tenant's pool's connection and the database that pool was opened for, one snapshot, so a refused move keeps discovering the database the tenant's queries still use (`internal/app`'s `discoveries` builds, runs and stops them from `AfterAdopt` and `App.Close`, and starts a tenant over on a fresh registry when a reload moves it to another address or database: `RetryRefresh` until the first success, then `StartAutoRefresh` with a random first tick; `Lookup` answers `ErrNotLoaded` before the first success — the handlers' `503` with `Retry-After` — and `ErrUnknownTable` after; a failed loop attempt counts in `wavehouse_schema_refresh_failures_total{tenant}`), that introspects ClickHouse `system.columns` (name/type/nullability plus `default_expression` and 1-based `position`) and `system.tables` (each table's `create_table_query`, kept in-process and never serialized — an external-engine table renders its wiring there unconditionally — endpoint, bucket/host, database, username, S3 access key id; ClickHouse masks the password as `[HIDDEN]` from ~23.9, so the exposure is the topology, not the secret), records the server version, + `Validate()` for ingest payloads + `CanonicalizeTimestamps()` rewriting top-level `DateTime`/`DateTime64` column values to the canonical RFC 3339 UTC wire form pre-publish (Key Design Decision #19) - **`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. `Identity` (`identity.go`) is the process's name in its telemetry: one resource for every signal (service name, version, `instance_id`, roles, commit, build time; `OTEL_SERVICE_NAME`/`OTEL_RESOURCE_ATTRIBUTES` override it) and the `wavehouse_build_info` gauge (no instance label; the resource carries it). `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 2cdd5b08..f139fcc5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,6 +16,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added +- **Every span, metric series and log record names the process that sent it** (`internal/observability/identity.go` (new, + tests), `internal/observability/provider.go` (+ tests), `internal/app/wire_identity.go` (new, + test), `internal/app/wire.go`, `cmd/wavehouse/main.go`, `internal/config/config.go`, `internal/testutil/otlp.go`, `tests/integration/otel_test.go`, `config.yaml`, `docs/src/content/docs/{configuration.mdx,deployment.md,architecture.md}`, `AGENTS.md`; closes [#772](https://github.com/Wave-RF/WaveHouse/issues/772)): the OpenTelemetry resource carried only `service.name`, so series and spans pushed by two replicas of one role were indistinguishable unless the collector added an instance, and which build each process ran was visible only on `/version`. The resource now also carries `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles`, `wavehouse.git_commit` and `wavehouse.build_time`, and the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` variables override or extend it (a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, never fatal). A new `wavehouse_build_info` gauge, always `1`, is labeled `version`, `git_commit`, `build_time` and `roles`, so a mixed-version rollout shows on a dashboard. Every log line carries `instance_id`, stdout included. + - **`/readyz` consults the external queue, not just ClickHouse** (`internal/api/health.go` (+ tests), `internal/app/{app,wire,wire_ops,wire_nats}.go` (+ tests), `internal/mq/external.go` (+ test), `internal/mq/natstest/natstest.go`, `tests/integration/boot_resilience_test.go`, `docs/src/content/docs/{api,deployment,architecture}.md`, `reverse-proxy.mdx`, `AGENTS.md`): closes [#775](https://github.com/Wave-RF/WaveHouse/issues/775), the dependency checks [#144](https://github.com/Wave-RF/WaveHouse/issues/144) asked for and [#213](https://github.com/Wave-RF/WaveHouse/pull/213) deferred. The readiness handler runs a list of named `Check`s instead of one ClickHouse ping, every one on every call, and its `503` body's `error` names each that failed, one per line: `clickhouse` (`chconn.Pools.Ping`, as before, now under that prefix) and `mq` under `mq.backend: nats` (`ExternalNATS.Ready`: not connected, or the topology failed its last check — the states of the `wavehouse_mq_connected` and `wavehouse_mq_topology_ok` gauges; the lease bucket of `coord.backend: nats` rides that connection and is part of that check, so it is covered; the check also runs again as soon as the client reconnects, so a cluster that came back without a stream fails readiness without waiting for the periodic recheck). A process whose NATS connection has closed for good is now taken out of rotation instead of answering `503` on every request, and a process without the `api` role consults the queue too, where it used to be ready once booted without pools. Deliberately not checks: the shared cache (a tripped breaker falls back to ClickHouse) and the dedupe stores, Pebble and DynamoDB alike (a store that is not answering fails that tenant's ingest closed per request, and is only needed to dedupe, so it should not take the whole process out of rotation; parking those events on NATS until it answers is [#800](https://github.com/Wave-RF/WaveHouse/issues/800)). The probe bodies keep their diagnostic detail, the failing pool's ClickHouse address included: the bare probe paths are documented as internal-only, and the routes meant for a reverse proxy (`/v1/health`) carry none. - **Each tenant's streams are kept alive at its own `stream.keepalive_interval`** (`internal/app/keepalives.go` (new), `internal/app/{app,wire}.go` (+ tests), `internal/api/stream.go` (+ tests), comments in `internal/stream/{heartbeat,hub}.go`, `internal/chconn/chconn.go`, `internal/config/config.go` and `internal/settings/settings.go`, `docs/src/content/docs/{architecture,deployment}.md`, `{settings-directory,configuration,reverse-proxy}.mdx`, `AGENTS.md`): a story 2 follow-up of the multi-tenant epic ([#583](https://github.com/Wave-RF/WaveHouse/issues/583)), closing [#597](https://github.com/Wave-RF/WaveHouse/issues/597). `stream.keepalive_interval` and `stream.keepalive_buckets` are per tenant like every other key of a folder's `config.json`, but one keepalive wheel nudged every tenant's `GET /v1/stream` connections, at the shortest interval among the tenants being served: a tenant asking for one second set every tenant's cadence, and a reload of one tenant's settings moved another's. Each served tenant now has a wheel of its own, in the shape of its own pair. A connection is nudged at its tenant's interval whatever the others ask for; a reload of a tenant's pair rebuilds that tenant's ring in place under its open streams, as before, and leaves every other tenant's wheel untouched; and a tenant a reload removes or rejects has its wheel stopped as its streams are ended, and starts over on a fresh one when it is served again. A wheel exists from the reload that adopts its tenant, whether or not a stream is open: idle, it is one parked goroutine, about 5 KB, and a wake of about 3 µs every `keepalive_interval ÷ keepalive_buckets` (ten seconds at the seed's `30` and `3`), so a thousand idle tenants cost some 5 MB and 0.03% of one core, less than starting a wheel with a tenant's first stream and stopping it after its last would be worth. A flat directory's one tenant is kept alive exactly as before. Since the pair is a tenant's own, `stream.keepalive_buckets` is now bounded at `100` (`settings.MaxKeepaliveBuckets`; it had no upper bound): a wheel ticks once per bucket per interval and holds a set per bucket, so the cap bounds each tenant's tick rate and ring, and a hundred buckets already spreads the writes to 1% of its streams per tick; a `config.json` asking for more is refused at boot and on reload like any other out-of-range key. - **Codegen reads any tenant of a nested settings directory** (`clients/ts/src/cli/codegen.ts` (+ tests), `clients/ts/README.md`, `docs/src/content/docs/sdk/reference.md`): a story 6 follow-up of the multi-tenant epic ([#583](https://github.com/Wave-RF/WaveHouse/issues/583)), closing [#640](https://github.com/Wave-RF/WaveHouse/issues/640). `wavehouse-codegen` takes `--tenant ` (`-t`), sent as the `?tenant=` the `/v1/ops/schema` route reads, and `--operator-key ` (`-k`), sent as `X-Operator-Key` — the one credential a nested directory's `/v1/ops/*` routes admit, which `--auth`'s bearer token is not. The flags are independent: `--auth` stays the flat-directory credential, with or without `--tenant`, and the operator key without `--tenant` reads tenant `0` on either shape. Every flag that takes a value, `--url`, `--out` and `--auth` included, is refused client-side when its value is missing, empty, or another flag (what an unset shell variable leaves behind), so the next flag is never sent as the tenant or the credential; a malformed tenant id is the server's `400`. A request carrying either credential does not follow a redirect, where across origins `fetch` would drop the bearer token and re-send `X-Operator-Key` to the redirect's target: codegen stops and names the target instead. diff --git a/cmd/wavehouse/main.go b/cmd/wavehouse/main.go index 6a29c30b..7d3c8a7e 100644 --- a/cmd/wavehouse/main.go +++ b/cmd/wavehouse/main.go @@ -174,6 +174,10 @@ func run(ctx context.Context) int { logger.Error("load config", "error", err) return 1 } + // Every later line names the process; the OTLP-aware logger app.New may + // swap in carries it too (observability.Identity.LogAttrs). + logger = logger.With("instance_id", cfg.InstanceID) + slog.SetDefault(logger) // data_dir, when a selected backend keeps state there, must be writable // before anything dials out, so the refusal (and, for the typical cause — diff --git a/config.yaml b/config.yaml index a332a7cc..f67f3ad9 100644 --- a/config.yaml +++ b/config.yaml @@ -13,8 +13,8 @@ data_dir: ./data # coord.backend: nats (there is no sweeper under nats); separate api and ingest processes also need # cache.backend: redis. Boot refuses a split the backends cannot serve. roles: [api, ingest, sweeper] -# Names this process: logged at boot, and a lease's holder under -# coord.backend: nats. Empty means -<8 hex>, fresh at every boot. +# Names this process: on every log line, as service.instance.id in its +# telemetry, and a lease's holder under coord.backend: nats. Empty means -<8 hex>, fresh at every boot. instance_id: "" server: diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 6ab4af23..fdc5b992 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -133,7 +133,7 @@ The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https:/ - **backends.go** — the `.backend` keys: one string type per layer (`MQBackend`, `CacheBackend`, `DedupeBackend`, `CoordBackend`), each with its list of the backends this build has, and a `validate` per layer block (`checkBackend`, run by `validateBackends` in `Validate`, between `validateRoles` and `validateTopology`), which refuses a value not on the list and names the ones that are. One rule spans two layers: while `mq.backend` is `embedded`, `dedupe.lease` plus its own ceiling to the next whole second (`ceilSecond`) plus one more second must fit the embedded MQ's 2m duplicate window (`embeddedDuplicateWindow`), a cap of 59s (`maxEmbeddedLease`), because a client obeying the in-flight `503`'s `Retry-After` can republish as late as the lease plus that ceiling plus a second after the claim, since DynamoDB rounds a claim's expiry up to the second. A backend's own settings go in a `.` sub-block that is its `validate`'s case to check. `Distributed` reports whether the MQ is shared with other processes, `NeedsDataDir` whether a selected backend keeps state under `data_dir`, and `Warnings` returns what a valid configuration is still likely to get wrong — the combinations that are harmless or correct for one replica only (a shared MQ over a local cache or Pebble dedupe; under `nats`, `mq.max_bytes_gb` not applied; an `mq.nats` or `coord.nats` block its layer's backend ignores), a `cache.redis` block that is not read, certificate verification turned off — which `app.New` logs at `WARN`. `mq.backend` has two values, `embedded` and `nats` (`MQNATS`), and `nats` reads the `mq.nats` sub-block, which `MQ.validate` checks only when it is selected. `coord.backend` takes `nats` (`CoordNATS`) too, whose `coord.nats` block holds only the bucket name: the leases ride `mq.nats`'s connection. - **cache_redis.go** — `CacheRedisConfig`, the `cache.redis` sub-block, and its checks: an address (exactly one in `standalone` mode, which dials only the first), each `host:port` with a port from 1 to 65535 (a URL or `user:password@` form refused without repeating it, since it may hold a password), a known mode (`sentinel` is refused until [#656](https://github.com/Wave-RF/WaveHouse/issues/656)), `db` 0 in cluster mode, positive timeouts and sizes, a `timeout` and `dial_timeout` of at most 1 s each (boot and `Close` each wait out a dial: a connect and a handshake bounded by `dial_timeout`, and a cluster's topology read bounded by the larger of the two), a `version_ttl` of at least 2 s, and a `compress_min_bytes` that is not negative (`0` never compresses); its defaults are in `defaults()` with the rest. `CacheRedisTLS.Config` builds the `tls.Config`, reading the files; `Validate` calls it so an unreadable file refuses boot, and `internal/app` calls it again to build the connection. A TLS key set while `tls.enabled` is off is an error rather than a plaintext connection. - **mq_nats.go** — the two nats blocks and their rules, apart from backends.go for the e2e coverage gate as `cache_redis.go` is: `MQNATSConfig` (URLs, file-path-only credentials, TLS, and the topology to expect; its defaults, `defaultMQNATS`, are in `defaults()`) and its checks, `CoordNATSConfig` (the bucket name), the nats lines of `Warnings` (`natsWarnings`) and the two cross-layer rules `validateTopology` delegates (`validateNATSTopology`: `coord.backend=nats` needs `mq.backend=nats`, a process running `ingest` on a shared queue needs `coord.backend=nats`, and a process running only `sweeper` there is refused). -- **config.go**, roles — `roles` (`[]Role`: `api`, `ingest`, `sweeper`; `AllRoles` by default; `Has(Role)`) picks which components `internal/app` wires, and `instance_id` names the process (`-<8 hex>` when empty, resolved in `Load`; logged at boot, and the holder a NATS lease's value names). `validateRoles` refuses an empty list, an empty entry, an unknown or a repeated role; `validateTopology` refuses a role set the backends cannot serve: any split over the embedded MQ, `coord.backend=nats` without `mq.backend=nats` (the leases ride its connection), `mq.backend=nats` with `coord.backend=local` in a process running `ingest` (the ingest processes share the queue through leases), a process running only `sweeper` under `mq.backend=nats` (it would run nothing), and a process with exactly one of `api` and `ingest` over a local cache. `NeedsDataDir` counts Pebble only for a process running `api`, and the cache and dedupe warnings are skipped without `api`, since only that role opens a cache it reads or a dedupe store. +- **config.go**, roles — `roles` (`[]Role`: `api`, `ingest`, `sweeper`; `AllRoles` by default; `Has(Role)`) picks which components `internal/app` wires, and `instance_id` names the process (`-<8 hex>` when empty, resolved in `Load`; a field on every log line, `service.instance.id` in the telemetry, and the holder a NATS lease's value names). `validateRoles` refuses an empty list, an empty entry, an unknown or a repeated role; `validateTopology` refuses a role set the backends cannot serve: any split over the embedded MQ, `coord.backend=nats` without `mq.backend=nats` (the leases ride its connection), `mq.backend=nats` with `coord.backend=local` in a process running `ingest` (the ingest processes share the queue through leases), a process running only `sweeper` under `mq.backend=nats` (it would run nothing), and a process with exactly one of `api` and `ingest` over a local cache. `NeedsDataDir` counts Pebble only for a process running `api`, and the cache and dedupe warnings are skipped without `api`, since only that role opens a cache it reads or a dedupe store. - **strict.go** — `rejectUnknownKeys`, the YAML half: re-reads the file as a generic tree and walks it against the struct's `yaml` tags, listing every key the struct doesn't declare. cleanenv itself is lenient by design, which is exactly wrong for boot config once keys have moved to the settings directory. - **persistence.go** — `WarnIfFreshDataDir` logs the startup `WARN` when `data_dir` is missing or empty (on a redeploy, the sign that the volume didn't persist); `LogStorageInitError` attaches the UID-65532 `permissionHint` to a NATS or Pebble open failure that looks like a permission denial — the same hint string `CheckDataDir` uses. @@ -188,7 +188,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. +- **provider.go** — `InitProvider(ctx, Identity, ProviderConfig)` wires the OTel pipeline, every signal over one resource built from the `Identity`. 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. +- **identity.go** — `Identity`, the process's name in its telemetry ([#772](https://github.com/Wave-RF/WaveHouse/issues/772)): service name, version, commit, build time, `instance_id` and roles, built by `internal/app`'s `identity` (`wire_identity.go`). `newResource` turns it into the resource (`service.name`, `service.version`, `service.instance.id`, `wavehouse.roles`, `wavehouse.git_commit`, `wavehouse.build_time`, each left out when empty), then merges `resource.WithFromEnv`, so `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` override it key by key; a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, not fatal. `registerBuildInfo` exports `wavehouse_build_info` (always `1`; `version`, `git_commit`, `build_time`, `roles`) on the meter provider, with no instance label, since the resource carries it. `LogAttrs` are the build fields and `instance_id` the OTLP-aware logger carries on every record; `cmd/wavehouse` puts `instance_id` on the plain boot logger once the config is loaded. - **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. - **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 714a1d7a..97ede560 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -133,7 +133,7 @@ By default one process does all the work. `roles` splits it, so that the API and | YAML Key | Env Var | Default | Description | | --- | --- | ------- | ----------- | | `roles` | `WH_ROLES` | `api,ingest,sweeper` | The roles this process runs: a YAML list, or a comma-separated variable. Order does not matter. An empty list, an empty entry, an unknown role, or a role named twice refuses boot. | -| `instance_id` | `WH_INSTANCE_ID` | *(empty)* | Names this process: it is logged at boot (the `process roles` line), and under `coord.backend=nats` it is the `holder` a lease's value names, so you can see which process holds a lease. Empty resolves at boot to `-<8 hex>`, with a fresh random suffix every time, so a restarted process is a new instance. | +| `instance_id` | `WH_INSTANCE_ID` | *(empty)* | Names this process: every log line carries it as `instance_id`, its spans, metric series and OTLP log records carry it as the `service.instance.id` resource attribute (see [OTel](#otel)), and under `coord.backend=nats` it is the `holder` a lease's value names, so you can see which process holds a lease. Empty resolves at boot to `-<8 hex>`, with a fresh random suffix every time, so a restarted process is a new instance. | | Role | Runs | | --- | --- | @@ -289,13 +289,15 @@ The master switch is `otel.enabled`. When `true`, each signal (traces/metrics/lo **Direct-to-cloud OTLP.** The OTLP destination is configured through the standard [`OTEL_EXPORTER_OTLP_*` environment variables](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) read by the OpenTelemetry SDK, not WaveHouse config: `OTEL_EXPORTER_OTLP_ENDPOINT` (always include a scheme — `https://` selects TLS via system root CAs, `http://` selects plaintext), `OTEL_EXPORTER_OTLP_HEADERS` for per-RPC auth, `OTEL_EXPORTER_OTLP_CERTIFICATE` to trust a custom/private CA, and `OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE` / `OTEL_EXPORTER_OTLP_CLIENT_KEY` for mutual TLS. Without a scheme the SDK misreads the endpoint and each signal handles it differently (the logs exporter, for one, can be left with no destination at all), so always write the scheme. An unset endpoint means **TLS** to `localhost:4317`, so a plaintext local collector needs `http://localhost:4317`, or `OTEL_EXPORTER_OTLP_INSECURE=true` with the endpoint left unset. The custom/private CA and mutual TLS variables apply to all three signals (traces, metrics and logs). With these set, WaveHouse ships telemetry straight to a TLS-protected cloud gateway (Grafana Cloud's OTLP gateway, Honeycomb, etc.) — no sidecar required (a sidecar is still useful for egress queuing, batching, and tail-based sampling; it's just no longer mandatory). A malformed entry in `OTEL_EXPORTER_OTLP_HEADERS` never fails startup, but the signals treat it differently: traces and metrics log and skip just that entry, while the logs exporter logs it and discards the whole variable, so logs are sent with no headers at all and an authenticating gateway rejects them. Datadog has no public direct-to-cloud OTLP endpoint — use the local DDOT Collector path in the [deployment guide](/deployment#observability). See [Deployment → Observability](/deployment#observability) for worked Honeycomb / Grafana Cloud examples. +**Which process sent it.** Every span, metric series and OTLP log record carries the process's identity as OpenTelemetry resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the [`instance_id`](#process-roles)), `wavehouse.roles` (the [`roles`](#process-roles), comma-separated, such as `api,ingest`), `wavehouse.git_commit` and `wavehouse.build_time`. Set the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables to override any of them or add your own (`OTEL_RESOURCE_ATTRIBUTES=deployment.environment=prod`); they win on a shared key. A malformed `OTEL_RESOURCE_ATTRIBUTES` never fails startup: the entries that can be read apply, and the error is logged. The Prometheus endpoint reports the resource as `target_info`, and every metrics pipeline exports `wavehouse_build_info` (see [Deployment → Telling processes apart](/deployment#telling-processes-apart)). + | YAML Key | Env Var | Default | Description | | -------- | ------- | ------- | ----------- | | `otel.enabled` | `WH_OTEL_ENABLED` | `false` | Master switch. When `false`, no signals are initialized regardless of the sub-toggles below. The OTLP endpoint, TLS, custom CA, mutual TLS, and auth headers are configured via the standard `OTEL_EXPORTER_OTLP_*` env vars (see the note above), not a WaveHouse key. | | `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 plain boot logger: lines lose `trace_id`/`span_id`, `component`, `source` and the build fields, and keep `instance_id`. | | `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 @@ -325,7 +327,7 @@ Every key, with its default. Save the YAML as `config.yaml` next to the binary ( data_dir: ./data # nats → ./data/nats, pebble → ./data/pebble roles: [api, ingest, sweeper] # the work this process runs; a split needs shared backends -instance_id: "" # logged at boot; empty = -<8 hex>, fresh at every boot +instance_id: "" # on every log line and span; empty = -<8 hex>, fresh at every boot server: port: 8080 diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index f4152d58..2f07df67 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -504,7 +504,7 @@ Run every role in one process, the default, until you need more than one. A pod without the `api` role serves an ops listener on `:8080`: `/livez`, `/readyz` and their aliases, `/version`, the metrics path when `prometheus.port` is `0`, and `POST /v1/ops/settings/reload`. Every other route answers 404 (under `/v1/ops`, 403 without the operator key, and 401 for a bearer token). Point the same probes at it as at an API pod. `/livez` does not wait for schema discovery there, because only the API runs it. `/readyz` checks ClickHouse in an ingest pod, and the NATS connection in every pod. Every pod reads the settings directory, so mount it in every Deployment. The reload route on the ops listener accepts only the operator key, so whatever reloads your API pods over HTTP must send the operator key to the worker pods too, or rely on `SIGHUP` (or, over a flat directory, the directory watcher) instead. -Give each pod a stable `WH_INSTANCE_ID` only if you need one in the logs or in the lease's `holder`. The default, the pod's hostname with a random suffix, already names each pod uniquely. +Give each pod a stable `WH_INSTANCE_ID` only if you need one in the logs, the telemetry or the lease's `holder`. The default, the pod's hostname with a random suffix, already names each pod uniquely. ## Behind a reverse proxy @@ -800,6 +800,16 @@ export OTEL_EXPORTER_OTLP_ENDPOINT=http://alloy:4317 For the metrics path specifically: WaveHouse uses the OTel SDK's Prometheus exporter under the hood, which translates OTel metric names to Prometheus conventions automatically (dots and dashes become underscores; counters get a `_total` suffix). Existing OTel instruments don't need renaming. +### Telling processes apart + +Every span, metric series and OTLP log record names the process that sent it, so replicas of one role, or two builds during a rollout, never merge into one source. The identity rides as resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles`, `wavehouse.git_commit` and `wavehouse.build_time`. A Prometheus-compatible backend that ingests OTLP (Mimir, Grafana Cloud) maps `service.name` and `service.instance.id` to a series' `job` and `instance` labels; the Prometheus endpoint reports them on `target_info`. Override any of them, or add your own, with the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables, which win on a shared key; on Kubernetes, `OTEL_RESOURCE_ATTRIBUTES=k8s.pod.name=$(POD_NAME)` from the downward API is the usual addition. Every log line carries `instance_id` as a field too, stdout included. + +Which build each process runs is the `wavehouse_build_info` gauge, always `1`, labeled `version`, `git_commit`, `build_time` and `roles`. A mixed-version rollout shows as more than one `version`: + +```promql +count by (version) (wavehouse_build_info) +``` + ### Separating the `/metrics` listener By default, `prometheus.port` is `0`, which mounts `/metrics` on the main API server port (typically `8080`). This is the friendliest setup for compose / quick-start use. diff --git a/internal/app/wire.go b/internal/app/wire.go index 5905a85c..748e5edd 100644 --- a/internal/app/wire.go +++ b/internal/app/wire.go @@ -215,7 +215,7 @@ func (a *App) wireObservability(ctx context.Context) { if !cfg.OTel.Enabled && !cfg.Prometheus.Enabled { return } - shutdown, promHandler, err := observability.InitProvider(ctx, serviceName, observability.ProviderConfig{ + shutdown, promHandler, err := observability.InitProvider(ctx, a.identity(), observability.ProviderConfig{ TracesEnabled: cfg.OTel.Enabled && cfg.OTel.Traces.Enabled, TracesSampleRate: cfg.OTel.Traces.SampleRate, MetricsEnabled: cfg.OTel.Enabled && cfg.OTel.Metrics.Enabled, @@ -237,9 +237,7 @@ func (a *App) wireObservability(ctx context.Context) { // Prometheus-only mode keeps the stdout-only handler main installed. if cfg.OTel.Enabled && cfg.OTel.Logs.Enabled { slog.SetDefault(observability.NewLogger(serviceName, a.logLevel, true, cfg.OTel.Logs.SampleRate).With( - "version", a.build.Version, - "build_time", a.build.BuildTime, - "git_commit", a.build.GitCommit, + a.identity().LogAttrs()..., )) } otlpEndpoint := os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") diff --git a/internal/app/wire_identity.go b/internal/app/wire_identity.go new file mode 100644 index 00000000..1d041260 --- /dev/null +++ b/internal/app/wire_identity.go @@ -0,0 +1,21 @@ +package app + +import "github.com/Wave-RF/WaveHouse/internal/observability" + +// identity is how this process names itself in its telemetry: the build it +// runs, its instance_id and its roles (#772). A file of its own so the +// observability wiring in wire.go only calls it. +func (a *App) identity() observability.Identity { + roles := make([]string, len(a.cfg.Roles)) + for i, r := range a.cfg.Roles { + roles[i] = string(r) + } + return observability.Identity{ + ServiceName: serviceName, + Version: a.build.Version, + GitCommit: a.build.GitCommit, + BuildTime: a.build.BuildTime, + InstanceID: a.cfg.InstanceID, + Roles: roles, + } +} diff --git a/internal/app/wire_identity_test.go b/internal/app/wire_identity_test.go new file mode 100644 index 00000000..f5b61d3f --- /dev/null +++ b/internal/app/wire_identity_test.go @@ -0,0 +1,26 @@ +package app + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/Wave-RF/WaveHouse/internal/config" + "github.com/Wave-RF/WaveHouse/internal/observability" +) + +func TestIdentity_NamesTheBuildInstanceAndRoles(t *testing.T) { + t.Parallel() + a := &App{ + cfg: &config.Config{InstanceID: "pod-a-1234abcd", Roles: []config.Role{config.RoleAPI, config.RoleIngest}}, + build: BuildInfo{Version: "v1.2.3", GitCommit: "abc123", BuildTime: "2026-10-12T00:00:00Z"}, + } + assert.Equal(t, observability.Identity{ + ServiceName: "wavehouse", + Version: "v1.2.3", + GitCommit: "abc123", + BuildTime: "2026-10-12T00:00:00Z", + InstanceID: "pod-a-1234abcd", + Roles: []string{"api", "ingest"}, + }, a.identity()) +} diff --git a/internal/config/config.go b/internal/config/config.go index 587c6fe9..5d648ea2 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,9 +24,9 @@ type Config struct { // Roles are the components this process runs (every role by default); // a Deployment per role differs only in this. See Role. Roles []Role `yaml:"roles" env:"WH_ROLES"` - // InstanceID names this process: logged at boot, and the holder a - // coord.backend=nats lease names. Empty resolves to -<8 hex> at - // Load. + // InstanceID names this process: a field on every log line, + // service.instance.id in its telemetry, and the holder a coord.backend=nats + // lease names. Empty resolves to -<8 hex> at Load. InstanceID string `yaml:"instance_id" env:"WH_INSTANCE_ID"` Server Server `yaml:"server"` ClickHouse ClickHouse `yaml:"clickhouse"` diff --git a/internal/observability/identity.go b/internal/observability/identity.go new file mode 100644 index 00000000..5c4b2eb6 --- /dev/null +++ b/internal/observability/identity.go @@ -0,0 +1,102 @@ +package observability + +import ( + "context" + "log/slog" + "strings" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" + "go.opentelemetry.io/otel/sdk/resource" + semconv "go.opentelemetry.io/otel/semconv/v1.24.0" +) + +// Identity names the process in its telemetry: the resource every span, +// metric series and OTLP log record carries, and the wavehouse_build_info +// series, so replicas of one role and builds in a mixed-version rollout can +// be told apart. +type Identity struct { + // ServiceName is service.name unless OTEL_SERVICE_NAME overrides it. + ServiceName string + Version string + GitCommit string + BuildTime string + // InstanceID is service.instance.id: the boot config's instance_id, the + // name a lease's holder carries too. + InstanceID string + Roles []string +} + +// The resource attributes with no OpenTelemetry semantic convention. +const ( + attrRoles = attribute.Key("wavehouse.roles") + attrGitCommit = attribute.Key("wavehouse.git_commit") + attrBuildTime = attribute.Key("wavehouse.build_time") +) + +// BuildInfoMetric is the gauge naming the build and roles each process runs. +const BuildInfoMetric = "wavehouse_build_info" + +// roles is the role set as one value, "api,ingest,sweeper", so it reads the +// same as a resource attribute, a metric label and a log field. +func (id Identity) roles() string { return strings.Join(id.Roles, ",") } + +// LogAttrs are the fields every log record carries to name the process and +// its build. +func (id Identity) LogAttrs() []any { + return []any{ + "version", id.Version, + "build_time", id.BuildTime, + "git_commit", id.GitCommit, + "instance_id", id.InstanceID, + } +} + +// newResource builds the resource from id, then lets OTEL_RESOURCE_ATTRIBUTES +// and OTEL_SERVICE_NAME override it key by key, the standard OpenTelemetry +// precedence. A malformed OTEL_RESOURCE_ATTRIBUTES is never fatal, like a +// malformed OTEL_EXPORTER_OTLP_HEADERS: the entries it could read apply, and +// the rest is logged. +func newResource(ctx context.Context, id Identity) *resource.Resource { + attrs := []attribute.KeyValue{semconv.ServiceNameKey.String(id.ServiceName)} + for _, kv := range []attribute.KeyValue{ + semconv.ServiceVersionKey.String(id.Version), + semconv.ServiceInstanceIDKey.String(id.InstanceID), + attrRoles.String(id.roles()), + attrGitCommit.String(id.GitCommit), + attrBuildTime.String(id.BuildTime), + } { + if kv.Value.AsString() != "" { + attrs = append(attrs, kv) + } + } + res, err := resource.New(ctx, resource.WithAttributes(attrs...), resource.WithFromEnv()) + if err != nil { + slog.Warn("OTEL_RESOURCE_ATTRIBUTES is malformed; the entries that could be read apply", "error", err) + } + return res +} + +// registerBuildInfo exports wavehouse_build_info, always 1, labeled with the +// build and the roles. The instance is no label: it is service.instance.id on +// the resource, which an OTLP backend maps to the series' instance and the +// Prometheus exporter puts on target_info. +func registerBuildInfo(mp metric.MeterProvider, id Identity) error { + meter := mp.Meter("wavehouse-build") + gauge, err := meter.Int64ObservableGauge(BuildInfoMetric, + metric.WithDescription("Always 1; labeled with the version, commit, build time and roles this process runs")) + if err != nil { + return err + } + labels := metric.WithAttributes( + attribute.String("version", id.Version), + attribute.String("git_commit", id.GitCommit), + attribute.String("build_time", id.BuildTime), + attribute.String("roles", id.roles()), + ) + _, err = meter.RegisterCallback(func(_ context.Context, o metric.Observer) error { + o.ObserveInt64(gauge, 1, labels) + return nil + }, gauge) + return err +} diff --git a/internal/observability/identity_test.go b/internal/observability/identity_test.go new file mode 100644 index 00000000..f252b7b7 --- /dev/null +++ b/internal/observability/identity_test.go @@ -0,0 +1,127 @@ +package observability + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/sdk/resource" + + "github.com/Wave-RF/WaveHouse/internal/testutil/logtest" +) + +// testIdentity names the process under test in its telemetry. +var testIdentity = Identity{ + ServiceName: "wavehouse-test", + Version: "v1.2.3", + GitCommit: "abc123", + BuildTime: "2026-10-12T00:00:00Z", + InstanceID: "pod-a-1234abcd", + Roles: []string{"api", "ingest"}, +} + +func attrs(res *resource.Resource) map[attribute.Key]string { + m := map[attribute.Key]string{} + for _, kv := range res.Attributes() { + m[kv.Key] = kv.Value.Emit() + } + return m +} + +func TestNewResource_NamesTheProcess(t *testing.T) { + t.Setenv("OTEL_SERVICE_NAME", "") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "") + + assert.Equal(t, map[attribute.Key]string{ + "service.name": "wavehouse-test", + "service.version": "v1.2.3", + "service.instance.id": "pod-a-1234abcd", + "wavehouse.roles": "api,ingest", + "wavehouse.git_commit": "abc123", + "wavehouse.build_time": "2026-10-12T00:00:00Z", + }, attrs(newResource(t.Context(), testIdentity))) +} + +func TestNewResource_EmptyFieldsAreLeftOut(t *testing.T) { + t.Setenv("OTEL_SERVICE_NAME", "") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "") + + assert.Equal(t, map[attribute.Key]string{"service.name": "wavehouse"}, + attrs(newResource(t.Context(), Identity{ServiceName: "wavehouse"}))) +} + +func TestNewResource_EnvironmentOverridesKeyByKey(t *testing.T) { + t.Setenv("OTEL_SERVICE_NAME", "wavehouse-ingest") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "service.instance.id=ingest-0,deployment.environment=prod") + + got := attrs(newResource(t.Context(), testIdentity)) + assert.Equal(t, "wavehouse-ingest", got["service.name"], "OTEL_SERVICE_NAME wins") + assert.Equal(t, "ingest-0", got["service.instance.id"], "OTEL_RESOURCE_ATTRIBUTES wins on a shared key") + assert.Equal(t, "prod", got["deployment.environment"], "OTEL_RESOURCE_ATTRIBUTES adds keys") + assert.Equal(t, "v1.2.3", got["service.version"], "a key the environment leaves alone keeps WaveHouse's value") + assert.Equal(t, "api,ingest", got["wavehouse.roles"]) +} + +func TestNewResource_MalformedEnvironmentIsNotFatal(t *testing.T) { + logs := logtest.Capture(t, slog.LevelWarn) + t.Setenv("OTEL_SERVICE_NAME", "") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "broken,team=ops") + + got := attrs(newResource(t.Context(), testIdentity)) + assert.Equal(t, "ops", got["team"], "the entries that could be read apply") + assert.Equal(t, "pod-a-1234abcd", got["service.instance.id"]) + assert.Contains(t, logs.String(), "OTEL_RESOURCE_ATTRIBUTES is malformed") +} + +func TestIdentity_LogAttrs(t *testing.T) { + t.Parallel() + assert.Equal(t, []any{ + "version", "v1.2.3", + "build_time", "2026-10-12T00:00:00Z", + "git_commit", "abc123", + "instance_id", "pod-a-1234abcd", + }, testIdentity.LogAttrs()) +} + +// TestInitProvider_ExportsBuildInfoAndTargetInfo scrapes the Prometheus +// handler: wavehouse_build_info names the build and roles, and target_info +// carries the resource, the instance included. +func TestInitProvider_ExportsBuildInfoAndTargetInfo(t *testing.T) { + // t.Setenv keeps this serial, which InitProvider's global state needs. + t.Setenv("OTEL_SERVICE_NAME", "") + t.Setenv("OTEL_RESOURCE_ATTRIBUTES", "") + savedProp := otel.GetTextMapPropagator() + savedMP := otel.GetMeterProvider() + t.Cleanup(func() { + otel.SetTextMapPropagator(savedProp) + otel.SetMeterProvider(savedMP) + }) + + shutdown, promHandler, err := InitProvider(t.Context(), testIdentity, ProviderConfig{PrometheusEnabled: true}) + require.NoError(t, err) + t.Cleanup(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = shutdown(ctx) + }) + require.NotNil(t, promHandler) + + rec := httptest.NewRecorder() + promHandler.ServeHTTP(rec, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/metrics", nil)) + require.Equal(t, http.StatusOK, rec.Code) + body, err := io.ReadAll(rec.Body) + require.NoError(t, err) + out := string(body) + + assert.Regexp(t, `(?m)^wavehouse_build_info\{[^}]*build_time="2026-10-12T00:00:00Z"[^}]*git_commit="abc123"[^}]*roles="api,ingest"[^}]*version="v1.2.3"[^}]*\} 1$`, out) + assert.Regexp(t, `(?m)^target_info\{[^}]*service_instance_id="pod-a-1234abcd"[^}]*\} 1$`, out) + assert.Regexp(t, `(?m)^target_info\{[^}]*wavehouse_roles="api,ingest"[^}]*\} 1$`, out) +} diff --git a/internal/observability/provider.go b/internal/observability/provider.go index 4d9353b6..f311e37c 100644 --- a/internal/observability/provider.go +++ b/internal/observability/provider.go @@ -20,9 +20,7 @@ import ( "go.opentelemetry.io/otel/propagation" "go.opentelemetry.io/otel/sdk/log" "go.opentelemetry.io/otel/sdk/metric" - "go.opentelemetry.io/otel/sdk/resource" "go.opentelemetry.io/otel/sdk/trace" - semconv "go.opentelemetry.io/otel/semconv/v1.24.0" ) // runtimeStartOnce gates the `runtime.Start` call in InitProvider so that @@ -60,10 +58,13 @@ type ProviderConfig struct { // propagator (cheap, harmless when traces are off — in-process span // extraction still works against a no-op tracer provider). // +// Every signal carries id as its resource, and a meter provider exports +// wavehouse_build_info (see Identity). +// // Returns a shutdown function (always non-nil) and a Prometheus HTTP handler // (non-nil iff PrometheusEnabled). The handler reads from a private // prometheus.Registry — global registry pollution is avoided. -func InitProvider(ctx context.Context, serviceName string, cfg ProviderConfig) (func(context.Context) error, http.Handler, error) { +func InitProvider(ctx context.Context, id Identity, cfg ProviderConfig) (func(context.Context) error, http.Handler, error) { // Snapshot the OTel globals on entry. On a partial init failure we want to // roll them back to whatever was installed before — otherwise the caller // (which continues on init error per internal/app) keeps using shut-down @@ -134,13 +135,7 @@ func InitProvider(ctx context.Context, serviceName string, cfg ProviderConfig) ( } } - res, err := resource.New(ctx, - resource.WithAttributes(semconv.ServiceNameKey.String(serviceName)), - ) - if err != nil { - handleErr(err) - return shutdown, nil, err - } + res := newResource(ctx, id) otel.SetTextMapPropagator( propagation.NewCompositeTextMapPropagator( @@ -201,6 +196,10 @@ func InitProvider(ctx context.Context, serviceName string, cfg ProviderConfig) ( meterProvider := metric.NewMeterProvider(mpOpts...) shutdownFuncs = append(shutdownFuncs, meterProvider.Shutdown) otel.SetMeterProvider(meterProvider) + if err := registerBuildInfo(meterProvider, id); err != nil { + handleErr(err) + return shutdown, nil, err + } // `runtime.Start` spawns goroutines that the upstream package exposes // no way to stop. In production InitProvider is called exactly once, diff --git a/internal/observability/provider_test.go b/internal/observability/provider_test.go index 11204d45..7a03e3af 100644 --- a/internal/observability/provider_test.go +++ b/internal/observability/provider_test.go @@ -34,7 +34,7 @@ func TestInitProvider_Shutdown(t *testing.T) { // No OTEL_EXPORTER_OTLP_ENDPOINT set → the SDK uses its default target and // the gRPC exporters dial lazily, so InitProvider succeeds regardless of // whether anything is listening. - shutdown, promHandler, err := InitProvider(ctx, "wavehouse-test", ProviderConfig{ + shutdown, promHandler, err := InitProvider(ctx, testIdentity, ProviderConfig{ TracesEnabled: true, TracesSampleRate: 0.10, MetricsEnabled: true, @@ -83,7 +83,7 @@ func TestInitProvider_ShutdownParallelBounded(t *testing.T) { otel.SetLoggerProvider(savedLP) }) - shutdown, _, err := InitProvider(context.Background(), "wavehouse-test", ProviderConfig{ + shutdown, _, err := InitProvider(context.Background(), testIdentity, ProviderConfig{ TracesEnabled: true, MetricsEnabled: true, LogsEnabled: true, diff --git a/internal/testutil/otlp.go b/internal/testutil/otlp.go index 78512b59..55d04455 100644 --- a/internal/testutil/otlp.go +++ b/internal/testutil/otlp.go @@ -8,7 +8,7 @@ // // r := testutil.NewFakeOTLP(t) // t.Setenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://"+r.Addr()) -// shutdown, _, _ := observability.InitProvider(ctx, "svc", observability.ProviderConfig{...}) +// shutdown, _, _ := observability.InitProvider(ctx, observability.Identity{ServiceName: "svc"}, observability.ProviderConfig{...}) // ... emit spans/metrics/logs ... // _ = shutdown(ctx) // forces a final flush // assert.Equal(t, expected, r.SpanCount()) diff --git a/tests/integration/otel_test.go b/tests/integration/otel_test.go index c992ac87..b059a00c 100644 --- a/tests/integration/otel_test.go +++ b/tests/integration/otel_test.go @@ -37,6 +37,9 @@ import ( "github.com/Wave-RF/WaveHouse/internal/testutil" ) +// testIdentity names the process under test in its telemetry. +var testIdentity = observability.Identity{ServiceName: "wavehouse-test", Version: "v0.0.0-test", InstanceID: "test-1", Roles: []string{"api"}} + // guardOTelGlobals saves and restores the package-global OTel providers so a // test's InitProvider call doesn't leak its state into later tests. func guardOTelGlobals(t *testing.T) { @@ -61,7 +64,7 @@ func initAndShutdown(t *testing.T, cfg observability.ProviderConfig) (func(conte t.Helper() ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - shutdown, promHandler, err := observability.InitProvider(ctx, "wavehouse-test", cfg) + shutdown, promHandler, err := observability.InitProvider(ctx, testIdentity, cfg) require.NoError(t, err) require.NotNil(t, shutdown) return shutdown, promHandler @@ -230,6 +233,8 @@ func TestOTel_PrometheusScrape_ExposesMetrics(t *testing.T) { // Counter ending in _total is the standard Prometheus convention; the // OTel exporter applies it automatically. assert.Contains(t, bodyStr, "test_widget_received_total", "/metrics must list the custom counter we recorded") + assert.Contains(t, bodyStr, observability.BuildInfoMetric+"{", "/metrics must carry the build_info series") + assert.Contains(t, bodyStr, `service_instance_id="test-1"`, "target_info must name the instance") assert.Contains(t, bodyStr, "# HELP", "Prometheus format includes HELP lines") assert.Contains(t, bodyStr, "# TYPE", "Prometheus format includes TYPE lines") } @@ -279,7 +284,7 @@ func TestOTel_UnreachableEndpoint_DoesNotBlockStartupOrEmits(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - shutdown, _, err := observability.InitProvider(ctx, "wavehouse-test", cfg) + shutdown, _, err := observability.InitProvider(ctx, testIdentity, cfg) require.NoError(t, err, "InitProvider must not fail on unreachable endpoint") require.NotNil(t, shutdown) From fdb46849258f406db530552fda8e695d97058edb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 19:37:37 +0000 Subject: [PATCH 2/4] fix(observability): sort roles and drop the repeated instance_id --- AGENTS.md | 2 +- CHANGELOG.md | 2 +- config.yaml | 2 +- docs/src/content/docs/architecture.md | 4 ++-- docs/src/content/docs/configuration.mdx | 4 ++-- docs/src/content/docs/deployment.md | 4 ++-- internal/app/app.go | 2 +- internal/app/wire_identity.go | 9 ++++++++- internal/app/wire_identity_test.go | 2 +- internal/config/config.go | 2 +- 10 files changed, 20 insertions(+), 13 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 1c98d3eb..b08d252b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -34,7 +34,7 @@ Twenty internal packages under `internal/` (plus `internal/testutil/` for shared - **`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) - **`chconn/`** — `Pools`, one `Manager` (a `driver.Conn`) per distinct `Identity{Addr, Database, Username, Password, TLS}` tuple among the served tenants, reconciled from the settings registry's `AfterAdopt` after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): tenants naming one tuple share its pool, sized to their largest `max_open_conns`/`max_idle_conns`; a tenant whose tuple changed is repointed; a tuple no tenant names is released after the longest `query_timeout` among the tenants it had (never dials; a resize swaps the connection with the same grace). The boot config's `clickhouse.max_total_conns` bounds the open pools' `max_open_conns` together: boot refuses naming sum and ceiling; at a reload a resize above it keeps the pool's size, and a tuple that cannot be opened (the ceiling, an unreadable certificate, or options the driver refuses) leaves its tenants on the pool they had or on none — logged, retried by the next reload. Every consumer resolves its tenant's pool per call: `For` (nil for a tenant on no pool, a `503`), `Target` (the tenant's own HTTP wiring over its pool's TLS config), `SharingTables`, `Ping` (every pool at once, ready at the first answer). `HTTPClients` keeps one `http.Client` per TLS config. `Classify` (`errclass.go`) says what a failed ClickHouse request means for the request — `Unavailable`, `Denied`, `Rejected` (any unlisted exception code: the server read it and refused it), or `Unknown` (no code, no recognizable transport failure) — over the driver's error types and the HTTP interface's `HTTPError`; the ingest worker and the query handlers (`api/ch_errors.go` `writeCHError`, [#403](https://github.com/Wave-RF/WaveHouse/issues/403), [#271](https://github.com/Wave-RF/WaveHouse/issues/271)) both use it - **`chsql/`** — dependency-free ClickHouse SQL helpers shared by `query`/`policy` (avoids an import cycle): `QuoteIdent` (backtick-quote every identifier) + `BindUnsafe` (reject names with a literal `?`) -- **`config/`** — YAML + env var config loading (cleanenv); strict on both sides (undeclared YAML key, unbound `WH_*` variable) and probes `data_dir` writability when a selected backend keeps state there (`NeedsDataDir`); `backends.go` holds each layer's `.backend` (the in-process value by default; `mq.backend` also takes `nats`, with its `mq.nats` sub-block of file-path-only credentials; `coord.backend` takes `nats`, whose `coord.nats` block names only the lease bucket and rides `mq.nats`'s connection (both blocks, their rules and warnings are `mq_nats.go`); `cache.backend` takes `redis`, whose sub-block is `cache_redis.go`; and `dedupe.backend` takes `dynamodb`, with its `dedupe.dynamodb` sub-block) and `Warnings`, the valid combinations boot logs at `WARN`; `config.go` holds `roles` (`Has(Role)`) and `instance_id` (on every log line, and `service.instance.id` in the telemetry), and `Validate` refuses a role split the backends cannot serve (any split over the embedded MQ; `api` without `ingest`, or the reverse, over a local cache; `coord.backend=nats` without `mq.backend=nats`; `mq.backend=nats` with `coord.backend=local` in a process running `ingest`; a process running only `sweeper` under `mq.backend=nats`) — boot is the validator, there is no dry run +- **`config/`** — YAML + env var config loading (cleanenv); strict on both sides (undeclared YAML key, unbound `WH_*` variable) and probes `data_dir` writability when a selected backend keeps state there (`NeedsDataDir`); `backends.go` holds each layer's `.backend` (the in-process value by default; `mq.backend` also takes `nats`, with its `mq.nats` sub-block of file-path-only credentials; `coord.backend` takes `nats`, whose `coord.nats` block names only the lease bucket and rides `mq.nats`'s connection (both blocks, their rules and warnings are `mq_nats.go`); `cache.backend` takes `redis`, whose sub-block is `cache_redis.go`; and `dedupe.backend` takes `dynamodb`, with its `dedupe.dynamodb` sub-block) and `Warnings`, the valid combinations boot logs at `WARN`; `config.go` holds `roles` (`Has(Role)`) and `instance_id` (on every log line once loaded, and `service.instance.id` in the telemetry), and `Validate` refuses a role split the backends cannot serve (any split over the embedded MQ; `api` without `ingest`, or the reverse, over a local cache; `coord.backend=nats` without `mq.backend=nats`; `mq.backend=nats` with `coord.backend=local` in a process running `ingest`; a process running only `sweeper` under `mq.backend=nats`) — boot is the validator, there is no dry run - **`coord/`** — leases for work that must run in one process at a time (`Observer.Held` reads whether one is held without campaigning): `Coordinator.TryAcquire(ctx, name)` → a `Term` (fencing `Token`, strictly increasing per name; `Done`/`Err`, `ErrLost` on loss; `Resign`), `ErrHeld` while another holder's — or this coordinator's own — term is live; `RunElected` runs a loop only while holding its lease, resigning when the loop returns and campaigning again every `RetryPeriod`. `Local` is the in-process implementation (first taker wins, never expires; `Peer` is a second handle over the same table for tests); every implementation runs `coordtest.Conformance`. Imports only the standard library, so a distributed backend lives beside its connection: `coord.backend: nats` is `internal/mq/lease.go` (`ExternalNATS.Leases`), a key per lease in the operator's KV bucket, the KV revision as the fencing token, and expiry judged on the candidate's own clock (the same revision seen unchanged for 15s), never by a server TTL. `internal/app`'s `wireCoord` opens the one `coord.backend` selects and the sweeper runs through `RunElected` under the `sweeper` lease - **`dedupe/`** — `Deduplicator` interface (two-phase `Reserve`/`Commit`/`Release` over `Key{Table, ID}`; every backend passes the `dedupetest` conformance suite) → `Embedded` (Pebble: every tenant's seen ids in one instance at `data_dir/pebble`, each key led by its tenant and table, pending claims in memory, committed ids stored with their expiry and deleted by an hourly background sweep along with the version-0 keys from before the table joined the key, open while any tenant's store is — the layout is the implementation's call, and the wiring hands it `data_dir` once; its `Stats` feed the system gauges) or `Dynamo` (one shared DynamoDB table, conditional `PutItem` claims; conformance-tested against dynamodb-local, selected by `dedupe.backend: dynamodb`; boot checks the table and never creates it outside dynamodb-local), wrapped by `Managed` whose open/closed state follows the hot-reloadable `dedupe.enabled` in the settings directory's `config.json`; `Stores` holds one `Managed` per tenant, built through a `Factory` (`func(tenant.ID) *Managed`, `Embedded.Tenant` or, gated on the table check (`Factory.Gated`), `Dynamo.Tenant` in production; `Managed` opens its store through a function, so every backend gets the same switch), and reconciled from the registry's `AfterAdopt` hook — open exactly when the tenant is served with its switch on, closed with its seen ids kept otherwise ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) stories 7 and 3) - **`discovery/`** — `SchemaRegistry`, one per served tenant over a `Source` read once per refresh — the tenant's pool's connection and the database that pool was opened for, one snapshot, so a refused move keeps discovering the database the tenant's queries still use (`internal/app`'s `discoveries` builds, runs and stops them from `AfterAdopt` and `App.Close`, and starts a tenant over on a fresh registry when a reload moves it to another address or database: `RetryRefresh` until the first success, then `StartAutoRefresh` with a random first tick; `Lookup` answers `ErrNotLoaded` before the first success — the handlers' `503` with `Retry-After` — and `ErrUnknownTable` after; a failed loop attempt counts in `wavehouse_schema_refresh_failures_total{tenant}`), that introspects ClickHouse `system.columns` (name/type/nullability plus `default_expression` and 1-based `position`) and `system.tables` (each table's `create_table_query`, kept in-process and never serialized — an external-engine table renders its wiring there unconditionally — endpoint, bucket/host, database, username, S3 access key id; ClickHouse masks the password as `[HIDDEN]` from ~23.9, so the exposure is the topology, not the secret), records the server version, + `Validate()` for ingest payloads + `CanonicalizeTimestamps()` rewriting top-level `DateTime`/`DateTime64` column values to the canonical RFC 3339 UTC wire form pre-publish (Key Design Decision #19) diff --git a/CHANGELOG.md b/CHANGELOG.md index f139fcc5..212db648 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Added -- **Every span, metric series and log record names the process that sent it** (`internal/observability/identity.go` (new, + tests), `internal/observability/provider.go` (+ tests), `internal/app/wire_identity.go` (new, + test), `internal/app/wire.go`, `cmd/wavehouse/main.go`, `internal/config/config.go`, `internal/testutil/otlp.go`, `tests/integration/otel_test.go`, `config.yaml`, `docs/src/content/docs/{configuration.mdx,deployment.md,architecture.md}`, `AGENTS.md`; closes [#772](https://github.com/Wave-RF/WaveHouse/issues/772)): the OpenTelemetry resource carried only `service.name`, so series and spans pushed by two replicas of one role were indistinguishable unless the collector added an instance, and which build each process ran was visible only on `/version`. The resource now also carries `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles`, `wavehouse.git_commit` and `wavehouse.build_time`, and the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` variables override or extend it (a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, never fatal). A new `wavehouse_build_info` gauge, always `1`, is labeled `version`, `git_commit`, `build_time` and `roles`, so a mixed-version rollout shows on a dashboard. Every log line carries `instance_id`, stdout included. +- **Every span, metric series and log record names the process that sent it** (`internal/observability/identity.go` (new, + tests), `internal/observability/provider.go` (+ tests), `internal/app/wire_identity.go` (new, + test), `internal/app/{wire,app}.go`, `cmd/wavehouse/main.go`, `internal/config/config.go`, `internal/testutil/otlp.go`, `tests/integration/otel_test.go`, `config.yaml`, `docs/src/content/docs/{configuration.mdx,deployment.md,architecture.md}`, `AGENTS.md`; closes [#772](https://github.com/Wave-RF/WaveHouse/issues/772)): the OpenTelemetry resource carried only `service.name`, so series and spans pushed by two replicas of one role were indistinguishable unless the collector added an instance, and which build each process ran was visible only on `/version`. The resource now also carries `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles` (sorted), `wavehouse.git_commit` and `wavehouse.build_time`, and the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` variables override or extend it (a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, never fatal). A new `wavehouse_build_info` gauge, always `1`, is labeled `version`, `git_commit`, `build_time` and `roles`, so a mixed-version rollout shows on a dashboard. Every log line after the boot config loads carries `instance_id`, stdout included, so the `process roles` boot line no longer repeats it. - **`/readyz` consults the external queue, not just ClickHouse** (`internal/api/health.go` (+ tests), `internal/app/{app,wire,wire_ops,wire_nats}.go` (+ tests), `internal/mq/external.go` (+ test), `internal/mq/natstest/natstest.go`, `tests/integration/boot_resilience_test.go`, `docs/src/content/docs/{api,deployment,architecture}.md`, `reverse-proxy.mdx`, `AGENTS.md`): closes [#775](https://github.com/Wave-RF/WaveHouse/issues/775), the dependency checks [#144](https://github.com/Wave-RF/WaveHouse/issues/144) asked for and [#213](https://github.com/Wave-RF/WaveHouse/pull/213) deferred. The readiness handler runs a list of named `Check`s instead of one ClickHouse ping, every one on every call, and its `503` body's `error` names each that failed, one per line: `clickhouse` (`chconn.Pools.Ping`, as before, now under that prefix) and `mq` under `mq.backend: nats` (`ExternalNATS.Ready`: not connected, or the topology failed its last check — the states of the `wavehouse_mq_connected` and `wavehouse_mq_topology_ok` gauges; the lease bucket of `coord.backend: nats` rides that connection and is part of that check, so it is covered; the check also runs again as soon as the client reconnects, so a cluster that came back without a stream fails readiness without waiting for the periodic recheck). A process whose NATS connection has closed for good is now taken out of rotation instead of answering `503` on every request, and a process without the `api` role consults the queue too, where it used to be ready once booted without pools. Deliberately not checks: the shared cache (a tripped breaker falls back to ClickHouse) and the dedupe stores, Pebble and DynamoDB alike (a store that is not answering fails that tenant's ingest closed per request, and is only needed to dedupe, so it should not take the whole process out of rotation; parking those events on NATS until it answers is [#800](https://github.com/Wave-RF/WaveHouse/issues/800)). The probe bodies keep their diagnostic detail, the failing pool's ClickHouse address included: the bare probe paths are documented as internal-only, and the routes meant for a reverse proxy (`/v1/health`) carry none. - **Each tenant's streams are kept alive at its own `stream.keepalive_interval`** (`internal/app/keepalives.go` (new), `internal/app/{app,wire}.go` (+ tests), `internal/api/stream.go` (+ tests), comments in `internal/stream/{heartbeat,hub}.go`, `internal/chconn/chconn.go`, `internal/config/config.go` and `internal/settings/settings.go`, `docs/src/content/docs/{architecture,deployment}.md`, `{settings-directory,configuration,reverse-proxy}.mdx`, `AGENTS.md`): a story 2 follow-up of the multi-tenant epic ([#583](https://github.com/Wave-RF/WaveHouse/issues/583)), closing [#597](https://github.com/Wave-RF/WaveHouse/issues/597). `stream.keepalive_interval` and `stream.keepalive_buckets` are per tenant like every other key of a folder's `config.json`, but one keepalive wheel nudged every tenant's `GET /v1/stream` connections, at the shortest interval among the tenants being served: a tenant asking for one second set every tenant's cadence, and a reload of one tenant's settings moved another's. Each served tenant now has a wheel of its own, in the shape of its own pair. A connection is nudged at its tenant's interval whatever the others ask for; a reload of a tenant's pair rebuilds that tenant's ring in place under its open streams, as before, and leaves every other tenant's wheel untouched; and a tenant a reload removes or rejects has its wheel stopped as its streams are ended, and starts over on a fresh one when it is served again. A wheel exists from the reload that adopts its tenant, whether or not a stream is open: idle, it is one parked goroutine, about 5 KB, and a wake of about 3 µs every `keepalive_interval ÷ keepalive_buckets` (ten seconds at the seed's `30` and `3`), so a thousand idle tenants cost some 5 MB and 0.03% of one core, less than starting a wheel with a tenant's first stream and stopping it after its last would be worth. A flat directory's one tenant is kept alive exactly as before. Since the pair is a tenant's own, `stream.keepalive_buckets` is now bounded at `100` (`settings.MaxKeepaliveBuckets`; it had no upper bound): a wheel ticks once per bucket per interval and holds a set per bucket, so the cap bounds each tenant's tick rate and ring, and a hundred buckets already spreads the writes to 1% of its streams per tick; a `config.json` asking for more is refused at boot and on reload like any other out-of-range key. diff --git a/config.yaml b/config.yaml index f67f3ad9..e2d172b9 100644 --- a/config.yaml +++ b/config.yaml @@ -13,7 +13,7 @@ data_dir: ./data # coord.backend: nats (there is no sweeper under nats); separate api and ingest processes also need # cache.backend: redis. Boot refuses a split the backends cannot serve. roles: [api, ingest, sweeper] -# Names this process: on every log line, as service.instance.id in its +# Names this process: on every log line after boot config loads, as service.instance.id in its # telemetry, and a lease's holder under coord.backend: nats. Empty means -<8 hex>, fresh at every boot. instance_id: "" diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index fdc5b992..6f49a5c5 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -133,7 +133,7 @@ The SSE fan-out, factored out of `api/` so the delivery hot path ([#294](https:/ - **backends.go** — the `.backend` keys: one string type per layer (`MQBackend`, `CacheBackend`, `DedupeBackend`, `CoordBackend`), each with its list of the backends this build has, and a `validate` per layer block (`checkBackend`, run by `validateBackends` in `Validate`, between `validateRoles` and `validateTopology`), which refuses a value not on the list and names the ones that are. One rule spans two layers: while `mq.backend` is `embedded`, `dedupe.lease` plus its own ceiling to the next whole second (`ceilSecond`) plus one more second must fit the embedded MQ's 2m duplicate window (`embeddedDuplicateWindow`), a cap of 59s (`maxEmbeddedLease`), because a client obeying the in-flight `503`'s `Retry-After` can republish as late as the lease plus that ceiling plus a second after the claim, since DynamoDB rounds a claim's expiry up to the second. A backend's own settings go in a `.` sub-block that is its `validate`'s case to check. `Distributed` reports whether the MQ is shared with other processes, `NeedsDataDir` whether a selected backend keeps state under `data_dir`, and `Warnings` returns what a valid configuration is still likely to get wrong — the combinations that are harmless or correct for one replica only (a shared MQ over a local cache or Pebble dedupe; under `nats`, `mq.max_bytes_gb` not applied; an `mq.nats` or `coord.nats` block its layer's backend ignores), a `cache.redis` block that is not read, certificate verification turned off — which `app.New` logs at `WARN`. `mq.backend` has two values, `embedded` and `nats` (`MQNATS`), and `nats` reads the `mq.nats` sub-block, which `MQ.validate` checks only when it is selected. `coord.backend` takes `nats` (`CoordNATS`) too, whose `coord.nats` block holds only the bucket name: the leases ride `mq.nats`'s connection. - **cache_redis.go** — `CacheRedisConfig`, the `cache.redis` sub-block, and its checks: an address (exactly one in `standalone` mode, which dials only the first), each `host:port` with a port from 1 to 65535 (a URL or `user:password@` form refused without repeating it, since it may hold a password), a known mode (`sentinel` is refused until [#656](https://github.com/Wave-RF/WaveHouse/issues/656)), `db` 0 in cluster mode, positive timeouts and sizes, a `timeout` and `dial_timeout` of at most 1 s each (boot and `Close` each wait out a dial: a connect and a handshake bounded by `dial_timeout`, and a cluster's topology read bounded by the larger of the two), a `version_ttl` of at least 2 s, and a `compress_min_bytes` that is not negative (`0` never compresses); its defaults are in `defaults()` with the rest. `CacheRedisTLS.Config` builds the `tls.Config`, reading the files; `Validate` calls it so an unreadable file refuses boot, and `internal/app` calls it again to build the connection. A TLS key set while `tls.enabled` is off is an error rather than a plaintext connection. - **mq_nats.go** — the two nats blocks and their rules, apart from backends.go for the e2e coverage gate as `cache_redis.go` is: `MQNATSConfig` (URLs, file-path-only credentials, TLS, and the topology to expect; its defaults, `defaultMQNATS`, are in `defaults()`) and its checks, `CoordNATSConfig` (the bucket name), the nats lines of `Warnings` (`natsWarnings`) and the two cross-layer rules `validateTopology` delegates (`validateNATSTopology`: `coord.backend=nats` needs `mq.backend=nats`, a process running `ingest` on a shared queue needs `coord.backend=nats`, and a process running only `sweeper` there is refused). -- **config.go**, roles — `roles` (`[]Role`: `api`, `ingest`, `sweeper`; `AllRoles` by default; `Has(Role)`) picks which components `internal/app` wires, and `instance_id` names the process (`-<8 hex>` when empty, resolved in `Load`; a field on every log line, `service.instance.id` in the telemetry, and the holder a NATS lease's value names). `validateRoles` refuses an empty list, an empty entry, an unknown or a repeated role; `validateTopology` refuses a role set the backends cannot serve: any split over the embedded MQ, `coord.backend=nats` without `mq.backend=nats` (the leases ride its connection), `mq.backend=nats` with `coord.backend=local` in a process running `ingest` (the ingest processes share the queue through leases), a process running only `sweeper` under `mq.backend=nats` (it would run nothing), and a process with exactly one of `api` and `ingest` over a local cache. `NeedsDataDir` counts Pebble only for a process running `api`, and the cache and dedupe warnings are skipped without `api`, since only that role opens a cache it reads or a dedupe store. +- **config.go**, roles — `roles` (`[]Role`: `api`, `ingest`, `sweeper`; `AllRoles` by default; `Has(Role)`) picks which components `internal/app` wires, and `instance_id` names the process (`-<8 hex>` when empty, resolved in `Load`; a field on every log line once loaded, `service.instance.id` in the telemetry, and the holder a NATS lease's value names). `validateRoles` refuses an empty list, an empty entry, an unknown or a repeated role; `validateTopology` refuses a role set the backends cannot serve: any split over the embedded MQ, `coord.backend=nats` without `mq.backend=nats` (the leases ride its connection), `mq.backend=nats` with `coord.backend=local` in a process running `ingest` (the ingest processes share the queue through leases), a process running only `sweeper` under `mq.backend=nats` (it would run nothing), and a process with exactly one of `api` and `ingest` over a local cache. `NeedsDataDir` counts Pebble only for a process running `api`, and the cache and dedupe warnings are skipped without `api`, since only that role opens a cache it reads or a dedupe store. - **strict.go** — `rejectUnknownKeys`, the YAML half: re-reads the file as a generic tree and walks it against the struct's `yaml` tags, listing every key the struct doesn't declare. cleanenv itself is lenient by design, which is exactly wrong for boot config once keys have moved to the settings directory. - **persistence.go** — `WarnIfFreshDataDir` logs the startup `WARN` when `data_dir` is missing or empty (on a redeploy, the sign that the volume didn't persist); `LogStorageInitError` attaches the UID-65532 `permissionHint` to a NATS or Pebble open failure that looks like a permission denial — the same hint string `CheckDataDir` uses. @@ -189,7 +189,7 @@ The **only** package that imports NATS/JetStream — a `depguard` rule in `.gola ### `observability/` — OpenTelemetry Pipeline - **provider.go** — `InitProvider(ctx, Identity, ProviderConfig)` wires the OTel pipeline, every signal over one resource built from the `Identity`. 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. -- **identity.go** — `Identity`, the process's name in its telemetry ([#772](https://github.com/Wave-RF/WaveHouse/issues/772)): service name, version, commit, build time, `instance_id` and roles, built by `internal/app`'s `identity` (`wire_identity.go`). `newResource` turns it into the resource (`service.name`, `service.version`, `service.instance.id`, `wavehouse.roles`, `wavehouse.git_commit`, `wavehouse.build_time`, each left out when empty), then merges `resource.WithFromEnv`, so `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` override it key by key; a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, not fatal. `registerBuildInfo` exports `wavehouse_build_info` (always `1`; `version`, `git_commit`, `build_time`, `roles`) on the meter provider, with no instance label, since the resource carries it. `LogAttrs` are the build fields and `instance_id` the OTLP-aware logger carries on every record; `cmd/wavehouse` puts `instance_id` on the plain boot logger once the config is loaded. +- **identity.go** — `Identity`, the process's name in its telemetry ([#772](https://github.com/Wave-RF/WaveHouse/issues/772)): service name, version, commit, build time, `instance_id` and roles, built by `internal/app`'s `identity` (`wire_identity.go`). `newResource` turns it into the resource (`service.name`, `service.version`, `service.instance.id`, `wavehouse.roles` (sorted by `identity`), `wavehouse.git_commit`, `wavehouse.build_time`, each left out when empty), then merges `resource.WithFromEnv`, so `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` override it key by key; a malformed `OTEL_RESOURCE_ATTRIBUTES` is logged, not fatal. `registerBuildInfo` exports `wavehouse_build_info` (always `1`; `version`, `git_commit`, `build_time`, `roles`) on the meter provider, with no instance label, since the resource carries it. `LogAttrs` are the build fields and `instance_id` the OTLP-aware logger carries on every record; `cmd/wavehouse` puts `instance_id` on the plain boot logger once the config is loaded. - **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. - **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 97ede560..474de0c7 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -133,7 +133,7 @@ By default one process does all the work. `roles` splits it, so that the API and | YAML Key | Env Var | Default | Description | | --- | --- | ------- | ----------- | | `roles` | `WH_ROLES` | `api,ingest,sweeper` | The roles this process runs: a YAML list, or a comma-separated variable. Order does not matter. An empty list, an empty entry, an unknown role, or a role named twice refuses boot. | -| `instance_id` | `WH_INSTANCE_ID` | *(empty)* | Names this process: every log line carries it as `instance_id`, its spans, metric series and OTLP log records carry it as the `service.instance.id` resource attribute (see [OTel](#otel)), and under `coord.backend=nats` it is the `holder` a lease's value names, so you can see which process holds a lease. Empty resolves at boot to `-<8 hex>`, with a fresh random suffix every time, so a restarted process is a new instance. | +| `instance_id` | `WH_INSTANCE_ID` | *(empty)* | Names this process: every log line after the boot config loads carries it as `instance_id`, its spans, metric series and OTLP log records carry it as the `service.instance.id` resource attribute (see [OTel](#otel)), and under `coord.backend=nats` it is the `holder` a lease's value names, so you can see which process holds a lease. Empty resolves at boot to `-<8 hex>`, with a fresh random suffix every time, so a restarted process is a new instance. | | Role | Runs | | --- | --- | @@ -289,7 +289,7 @@ The master switch is `otel.enabled`. When `true`, each signal (traces/metrics/lo **Direct-to-cloud OTLP.** The OTLP destination is configured through the standard [`OTEL_EXPORTER_OTLP_*` environment variables](https://opentelemetry.io/docs/specs/otel/protocol/exporter/) read by the OpenTelemetry SDK, not WaveHouse config: `OTEL_EXPORTER_OTLP_ENDPOINT` (always include a scheme — `https://` selects TLS via system root CAs, `http://` selects plaintext), `OTEL_EXPORTER_OTLP_HEADERS` for per-RPC auth, `OTEL_EXPORTER_OTLP_CERTIFICATE` to trust a custom/private CA, and `OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE` / `OTEL_EXPORTER_OTLP_CLIENT_KEY` for mutual TLS. Without a scheme the SDK misreads the endpoint and each signal handles it differently (the logs exporter, for one, can be left with no destination at all), so always write the scheme. An unset endpoint means **TLS** to `localhost:4317`, so a plaintext local collector needs `http://localhost:4317`, or `OTEL_EXPORTER_OTLP_INSECURE=true` with the endpoint left unset. The custom/private CA and mutual TLS variables apply to all three signals (traces, metrics and logs). With these set, WaveHouse ships telemetry straight to a TLS-protected cloud gateway (Grafana Cloud's OTLP gateway, Honeycomb, etc.) — no sidecar required (a sidecar is still useful for egress queuing, batching, and tail-based sampling; it's just no longer mandatory). A malformed entry in `OTEL_EXPORTER_OTLP_HEADERS` never fails startup, but the signals treat it differently: traces and metrics log and skip just that entry, while the logs exporter logs it and discards the whole variable, so logs are sent with no headers at all and an authenticating gateway rejects them. Datadog has no public direct-to-cloud OTLP endpoint — use the local DDOT Collector path in the [deployment guide](/deployment#observability). See [Deployment → Observability](/deployment#observability) for worked Honeycomb / Grafana Cloud examples. -**Which process sent it.** Every span, metric series and OTLP log record carries the process's identity as OpenTelemetry resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the [`instance_id`](#process-roles)), `wavehouse.roles` (the [`roles`](#process-roles), comma-separated, such as `api,ingest`), `wavehouse.git_commit` and `wavehouse.build_time`. Set the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables to override any of them or add your own (`OTEL_RESOURCE_ATTRIBUTES=deployment.environment=prod`); they win on a shared key. A malformed `OTEL_RESOURCE_ATTRIBUTES` never fails startup: the entries that can be read apply, and the error is logged. The Prometheus endpoint reports the resource as `target_info`, and every metrics pipeline exports `wavehouse_build_info` (see [Deployment → Telling processes apart](/deployment#telling-processes-apart)). +**Which process sent it.** Every span, metric series and OTLP log record carries the process's identity as OpenTelemetry resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the [`instance_id`](#process-roles)), `wavehouse.roles` (the [`roles`](#process-roles), sorted and comma-separated, such as `api,ingest`), `wavehouse.git_commit` and `wavehouse.build_time`. Set the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables to override any of them or add your own (`OTEL_RESOURCE_ATTRIBUTES=deployment.environment=prod`); they win on a shared key. A malformed `OTEL_RESOURCE_ATTRIBUTES` never fails startup: the entries that can be read apply, and the error is logged. The Prometheus endpoint reports the resource as `target_info`, and every metrics pipeline exports `wavehouse_build_info` (see [Deployment → Telling processes apart](/deployment#telling-processes-apart)). | YAML Key | Env Var | Default | Description | | -------- | ------- | ------- | ----------- | diff --git a/docs/src/content/docs/deployment.md b/docs/src/content/docs/deployment.md index 2f07df67..2c85c34a 100644 --- a/docs/src/content/docs/deployment.md +++ b/docs/src/content/docs/deployment.md @@ -802,9 +802,9 @@ For the metrics path specifically: WaveHouse uses the OTel SDK's Prometheus expo ### Telling processes apart -Every span, metric series and OTLP log record names the process that sent it, so replicas of one role, or two builds during a rollout, never merge into one source. The identity rides as resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles`, `wavehouse.git_commit` and `wavehouse.build_time`. A Prometheus-compatible backend that ingests OTLP (Mimir, Grafana Cloud) maps `service.name` and `service.instance.id` to a series' `job` and `instance` labels; the Prometheus endpoint reports them on `target_info`. Override any of them, or add your own, with the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables, which win on a shared key; on Kubernetes, `OTEL_RESOURCE_ATTRIBUTES=k8s.pod.name=$(POD_NAME)` from the downward API is the usual addition. Every log line carries `instance_id` as a field too, stdout included. +Every span, metric series and OTLP log record names the process that sent it, so replicas of one role, or two builds during a rollout, never merge into one source. The identity rides as resource attributes: `service.name` (`wavehouse`), `service.version`, `service.instance.id` (the boot config's `instance_id`), `wavehouse.roles`, `wavehouse.git_commit` and `wavehouse.build_time`. A Prometheus-compatible backend that ingests OTLP (Mimir, Grafana Cloud) maps `service.name` and `service.instance.id` to a series' `job` and `instance` labels; the Prometheus endpoint reports them on `target_info`. Override any of them, or add your own, with the standard `OTEL_SERVICE_NAME` and `OTEL_RESOURCE_ATTRIBUTES` environment variables, which win on a shared key; on Kubernetes, `OTEL_RESOURCE_ATTRIBUTES=k8s.pod.name=$(POD_NAME)` from the downward API is the usual addition. Every log line after the boot config loads carries `instance_id` as a field too, stdout included. -Which build each process runs is the `wavehouse_build_info` gauge, always `1`, labeled `version`, `git_commit`, `build_time` and `roles`. A mixed-version rollout shows as more than one `version`: +Which build each process runs is the `wavehouse_build_info` gauge, always `1`, labeled `version`, `git_commit`, `build_time` and `roles` (sorted, comma-separated, whatever order `roles` was written in). A mixed-version rollout shows as more than one `version`: ```promql count by (version) (wavehouse_build_info) diff --git a/internal/app/app.go b/internal/app/app.go index b9ac2004..98532568 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -181,7 +181,7 @@ func New(ctx context.Context, opts Options) (app *App, err error) { for _, w := range a.cfg.Warnings() { slog.Warn(w) } - slog.Info("process roles", "roles", a.cfg.Roles, "instance_id", a.cfg.InstanceID) + slog.Info("process roles", "roles", a.cfg.Roles) // What each role wires; config.Validate refused a set these cannot serve. // The API's discovery, dedupe, auth verifiers, hub bridge and keepalive // wheels are per process: every API process runs its own. diff --git a/internal/app/wire_identity.go b/internal/app/wire_identity.go index 1d041260..12b0e2eb 100644 --- a/internal/app/wire_identity.go +++ b/internal/app/wire_identity.go @@ -1,15 +1,22 @@ package app -import "github.com/Wave-RF/WaveHouse/internal/observability" +import ( + "slices" + + "github.com/Wave-RF/WaveHouse/internal/observability" +) // identity is how this process names itself in its telemetry: the build it // runs, its instance_id and its roles (#772). A file of its own so the // observability wiring in wire.go only calls it. func (a *App) identity() observability.Identity { + // Sorted: the order roles is written in means nothing, so it must not + // split a dashboard's series. roles := make([]string, len(a.cfg.Roles)) for i, r := range a.cfg.Roles { roles[i] = string(r) } + slices.Sort(roles) return observability.Identity{ ServiceName: serviceName, Version: a.build.Version, diff --git a/internal/app/wire_identity_test.go b/internal/app/wire_identity_test.go index f5b61d3f..12914289 100644 --- a/internal/app/wire_identity_test.go +++ b/internal/app/wire_identity_test.go @@ -12,7 +12,7 @@ import ( func TestIdentity_NamesTheBuildInstanceAndRoles(t *testing.T) { t.Parallel() a := &App{ - cfg: &config.Config{InstanceID: "pod-a-1234abcd", Roles: []config.Role{config.RoleAPI, config.RoleIngest}}, + cfg: &config.Config{InstanceID: "pod-a-1234abcd", Roles: []config.Role{config.RoleIngest, config.RoleAPI}}, build: BuildInfo{Version: "v1.2.3", GitCommit: "abc123", BuildTime: "2026-10-12T00:00:00Z"}, } assert.Equal(t, observability.Identity{ diff --git a/internal/config/config.go b/internal/config/config.go index 5d648ea2..0daa62cd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -24,7 +24,7 @@ type Config struct { // Roles are the components this process runs (every role by default); // a Deployment per role differs only in this. See Role. Roles []Role `yaml:"roles" env:"WH_ROLES"` - // InstanceID names this process: a field on every log line, + // InstanceID names this process: a field on every log line once loaded, // service.instance.id in its telemetry, and the holder a coord.backend=nats // lease names. Empty resolves to -<8 hex> at Load. InstanceID string `yaml:"instance_id" env:"WH_INSTANCE_ID"` From 9a244d7dc57bdc2f21d98328e8c4e907379438cb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 19:40:24 +0000 Subject: [PATCH 3/4] docs(observability): name the init errors the fallback still catches --- docs/src/content/docs/configuration.mdx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/src/content/docs/configuration.mdx b/docs/src/content/docs/configuration.mdx index 474de0c7..f15c801e 100644 --- a/docs/src/content/docs/configuration.mdx +++ b/docs/src/content/docs/configuration.mdx @@ -283,7 +283,7 @@ The one boot-config key for the [Settings Directory](/settings-directory) is whe ### OTel -The master switch is `otel.enabled`. When `true`, each signal (traces/metrics/logs) is then individually gated by its own `enabled` flag — you can run traces-only, logs-only, etc. Prometheus exposition is configured in its own top-level [`prometheus`](#prometheus) block; it operates independently and works in any combination with OTel (OTLP push only, Prometheus only, both, or neither). Stdout is always active for logs (the logger fans out to both stdout and the OTLP exporter), so logs never disappear 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. The `if err != nil` fallback in `internal/app` only fires for genuine init errors (malformed options, resource construction failure). +The master switch is `otel.enabled`. When `true`, each signal (traces/metrics/logs) is then individually gated by its own `enabled` flag — you can run traces-only, logs-only, etc. Prometheus exposition is configured in its own top-level [`prometheus`](#prometheus) block; it operates independently and works in any combination with OTel (OTLP push only, Prometheus only, both, or neither). Stdout is always active for logs (the logger fans out to both stdout and the OTLP exporter), so logs never disappear 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. The `if err != nil` fallback in `internal/app` only fires for genuine init errors (malformed exporter options, or the meter provider failing to register an instrument such as `wavehouse_build_info`); a malformed `OTEL_RESOURCE_ATTRIBUTES` is only logged (see **Which process sent it** below). **Sampling rates apply only to the OTLP push path.** Stdout always emits 100% of records — operators using a scraping-style pipeline (Promtail/Grafana Alloy → Loki, Vector, Fluent Bit, etc.) set the collection rate at the scraper, not the application. WaveHouse pushes telemetry to an OTel collector; the scraper world owns its own ingest policy. If you want to throttle OTLP volume for cost, lower the rates below. If you want to throttle Loki/Datadog Logs/etc., do it at that pipeline. From 8cad471d8e3f28ffa82b00e2ef6540c80aa8fcb8 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 12 Oct 2026 19:44:09 +0000 Subject: [PATCH 4/4] test(observability): read resource values with Value.String --- internal/observability/identity_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/observability/identity_test.go b/internal/observability/identity_test.go index f252b7b7..b794a8f0 100644 --- a/internal/observability/identity_test.go +++ b/internal/observability/identity_test.go @@ -31,7 +31,7 @@ var testIdentity = Identity{ func attrs(res *resource.Resource) map[attribute.Key]string { m := map[attribute.Key]string{} for _, kv := range res.Attributes() { - m[kv.Key] = kv.Value.Emit() + m[kv.Key] = kv.Value.String() } return m }