From e5c2f7ec4e7c8a1515ea758c25f431230cadbfa9 Mon Sep 17 00:00:00 2001 From: taitelee Date: Thu, 8 Oct 2026 11:19:55 -0400 Subject: [PATCH] fix(chconn): release a tuple's HTTP clients with its pool --- AGENTS.md | 2 +- CHANGELOG.md | 1 + docs/src/content/docs/architecture.md | 4 +- internal/api/query.go | 4 + internal/api/query_test.go | 30 +++ internal/app/app.go | 4 + internal/app/app_test.go | 59 +++++ internal/app/wire.go | 57 +++-- internal/chconn/chconn.go | 58 ++++- internal/chconn/chconn_test.go | 278 +++++++++++++++++++++--- internal/ingest/worker.go | 19 +- internal/ingest/worker_test.go | 24 +- tests/e2e/sdk/admin.test.ts | 32 ++- tests/e2e/sdk/settings.ts | 19 ++ tests/integration/ingest_outage_test.go | 2 +- tests/integration/shard_order_test.go | 2 +- 16 files changed, 518 insertions(+), 77 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6e8f6d6b..44a847d7 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,7 +32,7 @@ Twenty internal packages under `internal/` (plus `internal/testutil/` for shared - **`app/`** — the process wiring: `New` builds every component from the boot config and the settings directory (each one wired in one place — what it opens, what it loops, what it releases — with the settings registry handed to its wiring function whole, the injection point of the per-tenant registry of #583: store-keyed getters for the handlers, `perTenant` for the async paths (with the tenant each message's `mq.Topic` names for the stream hub and the ingest worker), the `chconn.Pools`, the per-tenant `discoveries` and the per-tenant keepalive wheels (`keepalives`: one `stream.Heartbeater` per served tenant, at its own `stream.keepalive_*`, turned by the `keepalive` component, [#597](https://github.com/Wave-RF/WaveHouse/issues/597)) reconciled from `AfterAdopt`, `gapWindows` handing the sweeper each tenant's own gap window (a rejected tenant's as its folder last had it, unbounded for one rejected since boot) and the `mq.max_bytes_gb` reconcile each served tenant's byte budget, and `defaultPolicy` for the one setting that still follows tenant `0`, a flat directory's ops-gate admin role; the auth verifiers are per tenant, reconfigured (rebuilt only on changed wiring) and pruned from `AfterAdopt`, the same hook's `Hub.Prune` ends the open streams of a tenant no longer served, and `wireCache`'s hook drops, through `LocalCache.Prune`, the cache version index of a tenant no longer served ([#262](https://github.com/Wave-RF/WaveHouse/issues/262))), `Run` drives the long-lived ones under one `errgroup` until the context is cancelled or one fails, `Close` releases them in reverse order. `New` wires only what the process's `roles` need (discovery, dedupe, auth verifiers, the hub bridge and keepalive per API process; the ingest worker per ingest process; the sweeper under its lease through `elected`, on the embedded MQ only); a process without `api` serves `api.NewOpsRouter` — probes, `/version`, metrics, and the settings reload behind the operator key alone. `cmd/wavehouse` and `tests/integration` both boot through it - **`auth/`** — JWT auth middleware: HMAC **or** JWKS verification with `alg` pinned to the active verifier, role extraction from a configurable claim path; always runs, never rejects (bad token → empty role + stashed reason). One verifier per tenant ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 9): `Authenticator` keys them by `tenant.ID` — the request store's `settings.Store.Tenant()`, through an injected `TenantSource`; `tenant.Default` on the tenant-exempt routes — built from each tenant's `auth` block by `Reconfigure`, dropped by `Prune` once the tenant stops being served (rejected or removed), released by `Close`; the secrets (`Config`) are boot-level and shared. A JWKS key set is fetched off the boot and reload paths: until one has been stored the verifier is pending and a token-bearing request gets `503` + `Retry-After` from `api.refuseUnverifiable` (`auth.ErrVerifierPending`), never a `default_role` evaluation; refresh is library-managed (Eric, 2026-09-22), response capped at 1 MiB; the operator key's admin role is the request tenant's - **`cache/`** — `Cache` interface → `LocalCache` (Ristretto: one pool for every tenant) + `VersionManager` (the invalidation index), and `RedisCache`, the Redis-compatible shared backend (random version tokens under the tenant's hash tag, one-round-trip lookups, bypass on failure behind a circuit breaker, deferred invalidations retried; selected by `cache.backend: redis`, configured by the boot config's `cache.redis` block — [#613](https://github.com/Wave-RF/WaveHouse/issues/613)). Every key carries the tenant (in `RedisCache`, after the key prefix: `:{}:…` for a version token, `:q::…` for a value); in `LocalCache` and the version index it leads ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 8) — `:query:` for the caller's query key and its singleflight, escaped whole as the lead field of the stored key `|.||…`, where each raw table and scope name is escaped by `keyenc` (a `Namespace` carries them raw, so no caller escapes); the index holds a version per tenant, per (tenant, table) and per (tenant, table, scope), keyed by raw name and bumped in place (one entry per live namespace however often it is bumped, [#262](https://github.com/Wave-RF/WaveHouse/issues/262)) — so no cached read or coalesced flight crosses tenants, a bump through `Invalidate` names one tenant's namespaces and no other's, and `InvalidateTenant` drops the tenant's index so its next key gets a process-unique generation, orphaning its every cached result in one step, pipe results included (no insert reaches a pipe result until [#343](https://github.com/Wave-RF/WaveHouse/pull/343)); `Lookup` returns a `Snapshot` of the versions it read, taken before the handler chooses any input a bump invalidates — the tenant's connection included — and `Set` files the fill under it, so a write landing mid-query, or a reload moving the tenant to another address or database after the request took its connection, orphans the fill ([#382](https://github.com/Wave-RF/WaveHouse/issues/382)), and every backend runs the conformance suite `internal/testutil/cachetest`; the one crossing is the wiring's, above the package: `internal/app` hands the ingest worker the cache through `sharedTables`, which repeats each of the worker's bumps under every tenant on the same ClickHouse address and database (`chconn.Pools.SharingTables`, whatever their user or tls block — they read the same tables), and orphans the whole cache of a tenant back on a pool after an absence, since it was out of that fan-out while away, or moved to another address or database, since it now reads other tables (story 6) -- **`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 +- **`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, and drops a released tuple's, its idle connections closed, after the grace the tuple's pool gets (`Release`, handed the tuples `Reconcile` returns as released by the wiring's pools hook; the tuples with a zero `tls` block share one client, which stays; [#713](https://github.com/Wave-RF/WaveHouse/issues/713)). `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 - **`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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 4552f117..89d276df 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -103,6 +103,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), ### Fixed +- **A ClickHouse tuple's HTTP clients are released with its pool** (`internal/chconn/chconn.go` (+ tests), `internal/app/{app,wire}.go` (+ tests), `internal/ingest/worker.go` (+ tests), `internal/api/query.go` (+ tests), `tests/integration/{ingest_outage,shard_order}_test.go`, `tests/e2e/sdk/{admin.test,settings}.ts`, `docs/src/content/docs/architecture.md`, `AGENTS.md`): closes [#713](https://github.com/Wave-RF/WaveHouse/issues/713), part of [#583](https://github.com/Wave-RF/WaveHouse/issues/583). The ingest worker and the `/v1/ops/query` proxy each keep one HTTP client per TLS config (`chconn.HTTPClients`), and a connection tuple with a `tls` block gets a config of its own each time its pool opens, so the client each had made for such a tuple stayed for the life of the process once the tuple was gone, its idle connections open until the transport's own 90-second idle timeout rather than closing with the pool: a test that opens and releases one tuple five times counted five clients, each still holding its connection after the native pool had closed. `Pools.Reconcile` now also returns the tuples it released, whether their tenants were removed or moved to another address, database, user or `tls` block, and the pools hook has both caches drop each one's client and close its idle connections after the grace the native pool gets, the longest `query_timeout` among the tenants the tuple had, so a request in flight finishes. A tuple another tenant still names keeps its client, and one that opens again gets a fresh one. The tuples with a zero `tls` block are unchanged: they share one client per cache, which stays. - **`make ci` no longer kills healthy test runs at the `go test` time limits** (`Makefile`, `docs/src/content/docs/development.md`): its parallel phase runs the unit suite beside the lint and build jobs on every core, where `internal/app` and `internal/mq`, about 10s each alone, run past the suite's 15s per-package budget, so that phase now gives the suite 60s (`CI_UNIT_TIMEOUT`). `make test-unit` and the unit job in CI, which run the suite on its own, keep the 15s (`UNIT_TIMEOUT`). The integration suite's limit goes from 480s to 900s (`INTEGRATION_TIMEOUT`): `tests/integration` takes about 6m on a CI runner and up to 8m under Docker Desktop, which 480s left half a minute. Each limit is a Makefile variable, so one can be raised for a run (`make ci CI_UNIT_TIMEOUT=90s`). - **Four `internal/mq` tests no longer fail when the embedded NATS server removes its streams directory under them** (`internal/mq/embedded_test.go`): after a stream is deleted, nats-server 2.14.6 removes its streams directory and the account's, once they are empty, on a goroutine of its own. A test that deleted every stream and then opened one raced that removal ([nats-io/nats-server#8725](https://github.com/nats-io/nats-server/issues/8725)), and the open was refused as `error creating store for stream`: rare (one failure in 6,400 local runs), but enough to fail the unit job on a pull request that changed no Go. Each of the four now keeps another stream on the server, so the directory is never empty, as `TestEmbeddedNATS_PacesTheRetriesOfAQueueThatCannotOpen` already did. - **`TestExternalNATS_AcksUnderBothAckSubjectLayouts` no longer fails on an ack that races the end of the test** (`internal/mq/external_perms_test.go`): the test published a second row to read the ack subject's layout off the history stream, and the worker's consumer, left running by the drain before it, took that row from the work queue and acked it as the test's context ended (`context canceled`, 9 of 40 runs alone). It now reads the layout off the history stream's copy of the row already drained. diff --git a/docs/src/content/docs/architecture.md b/docs/src/content/docs/architecture.md index 1399fb90..1c8e59ce 100644 --- a/docs/src/content/docs/architecture.md +++ b/docs/src/content/docs/architecture.md @@ -93,7 +93,7 @@ The API layer uses [Chi](https://github.com/go-chi/chi) for routing with Request ### `app/` — Process wiring - **app.go** — `New(ctx, Options)` builds every component from the boot config (`Options.Config`) and the settings directory it names, in dependency order: settings registry, observability (after which each `Config.Warnings` line is logged at `WARN`), ClickHouse pools, schema discovery, the dedupe stores, the MQ (embedded NATS with its ingest + DLQ streams, or the external NATS), cache, the lease coordinator, sweeper, streaming (hub, MQ→hub bridge, keepalive wheels), ingest worker, auth, reload triggers, HTTP. The boot config's `roles` decide which of them a process wires: every process gets the settings registry, observability, the MQ, the coordinator, the reload triggers and a listener; `api` adds schema discovery, the dedupe stores, streaming, auth and the full router; `ingest` adds the ingest worker; `sweeper` adds the sweeper on `mq.backend: embedded` only (under `nats` the streams' retention replaces it); the ClickHouse pools and the cache come with `api` or `ingest`. A process without `api` serves `api.NewOpsRouter` (probes, `/version`, the metrics path, and the settings reload behind the operator key alone, `wireOpsAuth`) on `server.port`. `config.Validate` refuses a role set the backends cannot serve (a split over the embedded MQ, ingest on a shared MQ over a local coordinator, a sweeper-only process on a shared MQ, or `api` without `ingest` and the reverse over a local cache), and `New` refuses a `Config` with no roles, which only one built without `config.Load` can have. Each is one `component` value — what it opens, what it loops, what it releases — so a failure part-way releases what was already opened and returns the error. `Run(ctx)` drives every loop under one `errgroup` until `ctx` is canceled (a clean stop: every loop drains, the API server and the ingest worker within `server.shutdown_timeout`; open SSE streams are ended as the drain begins rather than waited on) or a component fails, which stops the rest and returns that error. `Close(ctx)` releases what `New` opened, newest first, under the caller's release budget (`ReleaseTimeout`, 5s), a real bound: a remote implementation's close gives up at the deadline itself, and a close that ignores the context (the local stores) is abandoned at it, with the components below it left unreleased rather than overlapping it, both named in the error — and then flushes telemetry under its own 3s budget, so the flush that reports on the stop is never handed a deadline a slow close already spent. The SIGHUP registration is released last of all. `Handler`, `Registry`, and `MQ` expose the pieces a harness needs; `Options.Listener` lets one serve the API on its own listener instead of `server.port`. -- **wire.go** — one `wire*` function per component, each handed the settings registry whole and deriving the per-call getters the internal packages take (`DLQFor`, `DedupeFor`, `GapWindow`, …) and registering its `AfterAdopt` hook there where it has one. A layer with a choice of implementation — `wireMQ`, `wireCache`, `wireDedupe`, `wireCoord` — picks it there and nowhere else, in a `switch` on the boot config's `.backend` with one case per backend; the default case refuses boot, which only a `config.Config` built without `config.Load` reaches, since `Validate` refuses a value no case handles. `wireCache` has two: `local`, the in-process `LocalCache`, and `redis`, the shared `RedisCache` built from the `cache.redis` block, which boots bypassed rather than failing when its server is unreachable. Those wiring functions are where the per-tenant registry of [#583](https://github.com/Wave-RF/WaveHouse/issues/583) is injected, not `main`: `wireSettings` opens the `settings.Registry`, the HTTP handlers get store-keyed getters (method expressions such as `(*settings.Store).Policy`), and `perTenant` adapts a store accessor into the `func(tenant.ID) T` getter the async packages take, with the tenant each message's `mq.Topic` names for the stream hub and the ingest worker — a tenant the registry is not serving is logged and read as the zero value, except in `dlqFor`, the ingest worker's DLQ switch, where it reads as on so a message the worker cannot read is parked rather than dropped, and a removed or rejected tenant's queued rows are parked rather than left unacked, where each would be redelivered every ack wait for as long as the tenant is away and would hold that tenant's ack floor, so the sweeper could purge none of its queue past it. The ClickHouse pools (`chconn.Pools`) and the per-tenant schema registries (`discoveries`, in `discoveries.go`) are reconciled from `AfterAdopt` after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): `wireClickHouse` builds each served tenant's `chconn.Member` from its store and logs what the reconcile refused; `wireDiscovery` builds a registry over `pools.For` for each newly served tenant — a flat directory's tenant `0` refreshed synchronously first, as before — runs its loop under the App's stop context, stops the loop of a tenant no longer served, and drives the `BootState` from the first tenant's first discovery, sticky from there; before that, a diagnostic naming a tenant a reload stopped serving goes back to the no-tenant one. The handlers resolve both per request through store-keyed getters (`chConnFor`, `registryFor`, `chTargetFor`, `queryTimeout`), the hub and the ingest worker through tenant-keyed ones (`discoveries.For`, `pools.Target`) called with the tenant the message's topic names; a tenant on no pool is an untyped nil connection, the handlers' `503`. The ingest worker is handed the cache through `sharedTables`, which bumps each namespace the worker invalidates under every tenant on the same ClickHouse address and database (`pools.SharingTables`), and the pools hook orphans the whole cache — structured-query and pipe results — of a tenant back on a pool after an absence (`Cache.InvalidateTenant`), since it was out of that fan-out while away, and of a tenant moved to another address or database, since it now reads other tables (both returned by `Pools.Reconcile`). The one setting that still follows the default tenant is read per request, the admin role of a flat directory's ops gate: `defaultPolicy` reads it through the registry, where a flat directory's tenant `0` is always served. The auth verifiers are per tenant: `wireAuth` builds one for each tenant being served, its `AfterAdopt` hook reconfigures the adopted tenants' (rebuilt only when their wiring changed) and prunes the ones no longer served, and the operator key's admin role is read from the request tenant's policy. `wireStreaming`'s hook prunes the stream hub the same way (`Hub.Prune`, with the one `served` predicate the auth, dedupe and cache hooks use too), ending the open streams of a tenant no longer served, and `wireCache`'s hook prunes the cache's version index the same way (`LocalCache.Prune`, [#262](https://github.com/Wave-RF/WaveHouse/issues/262)), so a tenant no longer served stops holding it. The keepalive wheels are per tenant as well (`keepalives`, in `keepalives.go`, [#597](https://github.com/Wave-RF/WaveHouse/issues/597)): a second `wireStreaming` hook gives each served tenant a `stream.Heartbeater` in the shape of its own `stream.keepalive_interval` and `stream.keepalive_buckets`, reshapes a served tenant's in place (`Reconfigure`, a no-op when the pair did not change, so one tenant's reload never moves another's cadence), and stops the wheel of a tenant no longer served before the reload returns. The `keepalive` component's loop turns the wheels for as long as `Run` does, and returns once each has stopped; one built before `Run` holds its streams until then. The stream handler asks for its request tenant's wheel through a store-keyed getter (`keepaliveFor`), and a lookup that lands after a reload adopted a tenant and before the hook ran reconciles first, so a served tenant always has a wheel. The ingest worker's queue, over a `mq.Sharded` broker (`mq.backend: nats`), is wrapped in `ingest.ClaimShards` over the same coordinator, so each process consumes only its share of the shards; the `nats` coordinator holds the ingest processes' membership leases there, since no sweeper is wired under `nats`. The sweeper is handed each tenant's own `stream.gap_window_minutes` (`gapWindows`, read every sweep over `Registry.Known`, so a rejected tenant keeps the window its folder last had, and all of its history if the folder has been rejected since boot), since each tenant's events have a queue of their own, and runs only while its process holds the `sweeper` lease (`elected`, which wraps `coord.RunElected` over the coordinator `wireCoord` opens: `local` keeps leases in the process, so the one process always holds it; `nats` calls `ExternalNATS.Leases` on the MQ's own connection with the bucket `coordBucket` names — `coord.nats.bucket`, or `mq.DefaultNATSCoordBucket` of the subject prefix — and `instance_id` as the holder. `wireNATSMQ` hands the same bucket name to the topology, so boot waits for it with the streams. The coordinator is added after the MQ, so it closes first and resigns its terms while the connection is still up). The dedupe stores are per tenant ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 7): `wireDedupe`'s `pebble` case builds a `dedupe.Stores` over the `Tenant` factory of the embedded Pebble implementation (`dedupe.NewEmbedded`), handing it `data_dir` once; the implementation decides where every tenant's store lives — one instance, each key led by its tenant (story 3) and then its table — and one reconcile closure, the boot apply and the `AfterAdopt` hook alike, sets every store to what the registry says: open exactly when its tenant is served with `dedupe.enabled` on, closed with its seen ids kept when the tenant is switched off, rejected, or removed. An instance that cannot open follows the registry's rule for the shape: fatal at boot over a flat directory, fail-closed for every tenant with dedupe on over a nested one. The system gauges report that one instance's figures (`Embedded.Stats`), not a sum over tenants. `wireDedupe`'s `dynamodb` case is `wireDynamoDedupe`, in wire_dynamodb.go (below). `wireHTTP` hands the ingest handler `dedupe.lease` (`IngestHandler.DedupeLease`) whichever backend is chosen. The ingest handler picks the tenant's store off the request's `settings.Store` (`Store.Tenant()`). The reload triggers only start in `Run`, after `New` has registered every hook, so the watcher's first reload already drives all of them: SIGHUP in both shapes, the directory watcher for a flat directory only. `wireMQ`'s `nats` case builds an `mq.NATSConfig` from the boot config's `mq.nats` block and calls `mq.NewNATS`, which waits for the operator's topology under `New`'s context; it hands over no budget, since the operator's streams set every limit. Both cases end in `adoptMQ`, which registers the MQ's close and the system gauges. The `embedded` case hands each served tenant's `mq.max_bytes_gb` to `mq.Broker.SetMaxBytes` at boot, under `New`'s context (so a stop signaled mid-boot is not held up by opening many queues), and again after every reload, under the App's stop context; the first apply opens that tenant's queue. A queue that cannot be opened or resized follows the registry's rule for the shape — fatal at boot over a flat directory, logged over a nested one — and is retried by the next reload, a queue that did not open by the next publish too. How the budget is split across the tenant's streams, the time bounds, the rollback, and the dead-letter shrink guard are `internal/mq`'s. +- **wire.go** — one `wire*` function per component, each handed the settings registry whole and deriving the per-call getters the internal packages take (`DLQFor`, `DedupeFor`, `GapWindow`, …) and registering its `AfterAdopt` hook there where it has one. A layer with a choice of implementation — `wireMQ`, `wireCache`, `wireDedupe`, `wireCoord` — picks it there and nowhere else, in a `switch` on the boot config's `.backend` with one case per backend; the default case refuses boot, which only a `config.Config` built without `config.Load` reaches, since `Validate` refuses a value no case handles. `wireCache` has two: `local`, the in-process `LocalCache`, and `redis`, the shared `RedisCache` built from the `cache.redis` block, which boots bypassed rather than failing when its server is unreachable. Those wiring functions are where the per-tenant registry of [#583](https://github.com/Wave-RF/WaveHouse/issues/583) is injected, not `main`: `wireSettings` opens the `settings.Registry`, the HTTP handlers get store-keyed getters (method expressions such as `(*settings.Store).Policy`), and `perTenant` adapts a store accessor into the `func(tenant.ID) T` getter the async packages take, with the tenant each message's `mq.Topic` names for the stream hub and the ingest worker — a tenant the registry is not serving is logged and read as the zero value, except in `dlqFor`, the ingest worker's DLQ switch, where it reads as on so a message the worker cannot read is parked rather than dropped, and a removed or rejected tenant's queued rows are parked rather than left unacked, where each would be redelivered every ack wait for as long as the tenant is away and would hold that tenant's ack floor, so the sweeper could purge none of its queue past it. The ClickHouse pools (`chconn.Pools`) and the per-tenant schema registries (`discoveries`, in `discoveries.go`) are reconciled from `AfterAdopt` after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): `wireClickHouse` builds each served tenant's `chconn.Member` from its store (`chMembers`) and logs what the reconcile refused; `wireDiscovery` builds a registry over `pools.For` for each newly served tenant — a flat directory's tenant `0` refreshed synchronously first, as before — runs its loop under the App's stop context, stops the loop of a tenant no longer served, and drives the `BootState` from the first tenant's first discovery, sticky from there; before that, a diagnostic naming a tenant a reload stopped serving goes back to the no-tenant one. The handlers resolve both per request through store-keyed getters (`chConnFor`, `registryFor`, `chTargetFor`, `queryTimeout`), the hub and the ingest worker through tenant-keyed ones (`discoveries.For`, `pools.Target`) called with the tenant the message's topic names; a tenant on no pool is an untyped nil connection, the handlers' `503`. The ingest worker is handed the cache through `sharedTables`, which bumps each namespace the worker invalidates under every tenant on the same ClickHouse address and database (`pools.SharingTables`), and the pools hook orphans the whole cache — structured-query and pipe results — of a tenant back on a pool after an absence (`Cache.InvalidateTenant`), since it was out of that fan-out while away, and of a tenant moved to another address or database, since it now reads other tables (both returned by `Pools.Reconcile`). The same hook has the client caches of the HTTP-interface consumers, the ingest worker's and the query proxy's, release the clients of the tuples that reconcile released (`HTTPClients.Release`); the worker's cache is built by `wireIngestWorker`, since the worker itself exists only once `Run` starts it. The one setting that still follows the default tenant is read per request, the admin role of a flat directory's ops gate: `defaultPolicy` reads it through the registry, where a flat directory's tenant `0` is always served. The auth verifiers are per tenant: `wireAuth` builds one for each tenant being served, its `AfterAdopt` hook reconfigures the adopted tenants' (rebuilt only when their wiring changed) and prunes the ones no longer served, and the operator key's admin role is read from the request tenant's policy. `wireStreaming`'s hook prunes the stream hub the same way (`Hub.Prune`, with the one `served` predicate the auth, dedupe and cache hooks use too), ending the open streams of a tenant no longer served, and `wireCache`'s hook prunes the cache's version index the same way (`LocalCache.Prune`, [#262](https://github.com/Wave-RF/WaveHouse/issues/262)), so a tenant no longer served stops holding it. The keepalive wheels are per tenant as well (`keepalives`, in `keepalives.go`, [#597](https://github.com/Wave-RF/WaveHouse/issues/597)): a second `wireStreaming` hook gives each served tenant a `stream.Heartbeater` in the shape of its own `stream.keepalive_interval` and `stream.keepalive_buckets`, reshapes a served tenant's in place (`Reconfigure`, a no-op when the pair did not change, so one tenant's reload never moves another's cadence), and stops the wheel of a tenant no longer served before the reload returns. The `keepalive` component's loop turns the wheels for as long as `Run` does, and returns once each has stopped; one built before `Run` holds its streams until then. The stream handler asks for its request tenant's wheel through a store-keyed getter (`keepaliveFor`), and a lookup that lands after a reload adopted a tenant and before the hook ran reconciles first, so a served tenant always has a wheel. The ingest worker's queue, over a `mq.Sharded` broker (`mq.backend: nats`), is wrapped in `ingest.ClaimShards` over the same coordinator, so each process consumes only its share of the shards; the `nats` coordinator holds the ingest processes' membership leases there, since no sweeper is wired under `nats`. The sweeper is handed each tenant's own `stream.gap_window_minutes` (`gapWindows`, read every sweep over `Registry.Known`, so a rejected tenant keeps the window its folder last had, and all of its history if the folder has been rejected since boot), since each tenant's events have a queue of their own, and runs only while its process holds the `sweeper` lease (`elected`, which wraps `coord.RunElected` over the coordinator `wireCoord` opens: `local` keeps leases in the process, so the one process always holds it; `nats` calls `ExternalNATS.Leases` on the MQ's own connection with the bucket `coordBucket` names — `coord.nats.bucket`, or `mq.DefaultNATSCoordBucket` of the subject prefix — and `instance_id` as the holder. `wireNATSMQ` hands the same bucket name to the topology, so boot waits for it with the streams. The coordinator is added after the MQ, so it closes first and resigns its terms while the connection is still up). The dedupe stores are per tenant ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 7): `wireDedupe`'s `pebble` case builds a `dedupe.Stores` over the `Tenant` factory of the embedded Pebble implementation (`dedupe.NewEmbedded`), handing it `data_dir` once; the implementation decides where every tenant's store lives — one instance, each key led by its tenant (story 3) and then its table — and one reconcile closure, the boot apply and the `AfterAdopt` hook alike, sets every store to what the registry says: open exactly when its tenant is served with `dedupe.enabled` on, closed with its seen ids kept when the tenant is switched off, rejected, or removed. An instance that cannot open follows the registry's rule for the shape: fatal at boot over a flat directory, fail-closed for every tenant with dedupe on over a nested one. The system gauges report that one instance's figures (`Embedded.Stats`), not a sum over tenants. `wireDedupe`'s `dynamodb` case is `wireDynamoDedupe`, in wire_dynamodb.go (below). `wireHTTP` hands the ingest handler `dedupe.lease` (`IngestHandler.DedupeLease`) whichever backend is chosen. The ingest handler picks the tenant's store off the request's `settings.Store` (`Store.Tenant()`). The reload triggers only start in `Run`, after `New` has registered every hook, so the watcher's first reload already drives all of them: SIGHUP in both shapes, the directory watcher for a flat directory only. `wireMQ`'s `nats` case builds an `mq.NATSConfig` from the boot config's `mq.nats` block and calls `mq.NewNATS`, which waits for the operator's topology under `New`'s context; it hands over no budget, since the operator's streams set every limit. Both cases end in `adoptMQ`, which registers the MQ's close and the system gauges. The `embedded` case hands each served tenant's `mq.max_bytes_gb` to `mq.Broker.SetMaxBytes` at boot, under `New`'s context (so a stop signaled mid-boot is not held up by opening many queues), and again after every reload, under the App's stop context; the first apply opens that tenant's queue. A queue that cannot be opened or resized follows the registry's rule for the shape — fatal at boot over a flat directory, logged over a nested one — and is retried by the next reload, a queue that did not open by the next publish too. How the budget is split across the tenant's streams, the time bounds, the rollback, and the dead-letter shrink guard are `internal/mq`'s. - **wire_dynamodb.go** — `wireDedupe`'s `dynamodb` case, split out of wire.go so the e2e suite's coverage exclude for it (the e2e binary always runs Pebble dedupe, never DynamoDB) doesn't have to blanket wire.go itself: builds the same `dedupe.Stores` over `Dynamo.Tenant`, gated (`Factory.Gated`) on the table's check: boot runs `Dynamo.Check` (after `CreateTable`, when `dedupe.dynamodb.create_table` is on) whether or not any tenant has dedupe on. Boot is refused only for a misconfigured table (an error that is not `ErrUnavailable`) over a flat directory whose tenant has dedupe on; every other failure boots with the switched-on stores closed, the check retried until it passes by a background component that backs off from one second to thirty (a nested directory has no watcher, and a flat one's table can come good with no settings change). The `AfterAdopt` hook never runs the check, since it holds the lock that serializes reloads, and it does not wait on a tenant whose `dedupe.enabled` is unchanged either — `Managed.Apply`'s no-op fast path settles that case under its own read lock, so the hook only takes a store's write lock, and so waits for that tenant's in-flight `Reserve`/`Commit`/`Release` calls to finish, on a genuine flip. It applies every store against the last check's result, so a tenant a reload switches on fails closed meanwhile, and wakes the retry, so a reload still retries at once. It has no Pebble gauges. - **wire_nats.go** — the `nats` cases of `wireMQ` and `wireCoord` (`wireNATSMQ`, `wireNATSCoord`, and the lease bucket's name, `coordBucket`), split out of wire.go for the same reason as `wire_dynamodb.go`: the e2e binary never runs `mq.backend: nats`. `wireNATSMQ` hands the topology check the dedupe lease and, in a process running `api`, warns at boot and after every reload about a tenant with dedupe on whose finite retention is under the partitions' duplicate window (`ExternalNATS.DuplicateWindow`), and runs `ExternalNATS.CheckReplayWindows(servedGapWindows(...))` at boot and from an `AfterAdopt` hook, the check of a gap window longer than the history keeps that the sweeper made on the embedded MQ. @@ -230,7 +230,7 @@ The hot-reloadable half of configuration: a directory of four JSON files (`confi ### `chconn/` — ClickHouse Connection Pools -- **chconn.go** — `Pools` holds one `Manager` per distinct connection tuple among the served tenants — `Identity{Addr, Database, Username, Password, TLS}`, a plain comparable value, the map key — reconciled from the settings registry's `AfterAdopt` hook after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): tenants naming one tuple share its pool, sized to the largest `max_open_conns` and `max_idle_conns` among them (`Sizes`); a tenant whose tuple changed is repointed; a tuple no tenant names is released after the longest `query_timeout` among the tenants it had; a changed largest ask is a `Resize` with the same grace. The walk keeps the boot config's `clickhouse.max_total_conns` — the ceiling on the open pools' `max_open_conns` together — at every step, tenants no longer served leaving first and what was refused placed once more at the end: a refused resize keeps the pool's size, and a tuple that cannot be opened (the ceiling, a certificate file that cannot be read, or options the driver refuses — the pool opens as the walk places its first tenant, at the largest ask among the tenants naming it when that fits the ceiling and otherwise at that tenant's own, so each of these is undone in place) leaves its tenants on the pool they had — `Params` and all, so their `Target` stays whole — or on none; `NewPools` refuses boot on any refusal, `Reconcile` returns them joined for the wiring to log, and the next reload retries. `Manager` is a `driver.Conn` over one tuple's pool whose backing connection `Resize` swaps; like `clickhouse.Open` it never dials, so boot tolerates an unreachable ClickHouse (schema discovery retries) and a bad address surfaces where reachability is already handled (`/readyz`, query errors). The `tls` block is the tuple's, read once into one `tls.Config` handed to the driver when `tls.enabled` and carried on each tenant's `Target` for the https hop. Resolution is per tenant: `For` (the `driver.Conn`, nil for a tenant on no pool — the wiring returns an untyped nil), `Target` (the tenant's own `http_port`, `http_scheme` and `headers` over its pool's host, credentials, database and TLS config, from the `Params` last applied for it), `SharingTables` (the tenants on the same address and database, whatever their user — the cache fan-out's rule) and `Ping` (every pool at once, nil at the first answer). The HTTP-interface consumers (ingest INSERTs, raw-SQL proxy) take their `http.Client` from an `HTTPClients` cache, one client per TLS config ever handed to it, since the proxy serves tenants on different configs in alternation. +- **chconn.go** — `Pools` holds one `Manager` per distinct connection tuple among the served tenants — `Identity{Addr, Database, Username, Password, TLS}`, a plain comparable value, the map key — reconciled from the settings registry's `AfterAdopt` hook after every reload ([#583](https://github.com/Wave-RF/WaveHouse/issues/583) story 6): tenants naming one tuple share its pool, sized to the largest `max_open_conns` and `max_idle_conns` among them (`Sizes`); a tenant whose tuple changed is repointed; a tuple no tenant names is released after the longest `query_timeout` among the tenants it had; a changed largest ask is a `Resize` with the same grace. The walk keeps the boot config's `clickhouse.max_total_conns` — the ceiling on the open pools' `max_open_conns` together — at every step, tenants no longer served leaving first and what was refused placed once more at the end: a refused resize keeps the pool's size, and a tuple that cannot be opened (the ceiling, a certificate file that cannot be read, or options the driver refuses — the pool opens as the walk places its first tenant, at the largest ask among the tenants naming it when that fits the ceiling and otherwise at that tenant's own, so each of these is undone in place) leaves its tenants on the pool they had — `Params` and all, so their `Target` stays whole — or on none; `NewPools` refuses boot on any refusal, `Reconcile` returns them joined for the wiring to log, and the next reload retries. `Manager` is a `driver.Conn` over one tuple's pool whose backing connection `Resize` swaps; like `clickhouse.Open` it never dials, so boot tolerates an unreachable ClickHouse (schema discovery retries) and a bad address surfaces where reachability is already handled (`/readyz`, query errors). The `tls` block is the tuple's, read once into one `tls.Config` handed to the driver when `tls.enabled` and carried on each tenant's `Target` for the https hop. Resolution is per tenant: `For` (the `driver.Conn`, nil for a tenant on no pool — the wiring returns an untyped nil), `Target` (the tenant's own `http_port`, `http_scheme` and `headers` over its pool's host, credentials, database and TLS config, from the `Params` last applied for it), `SharingTables` (the tenants on the same address and database, whatever their user — the cache fan-out's rule) and `Ping` (every pool at once, nil at the first answer). The HTTP-interface consumers (ingest INSERTs, raw-SQL proxy) take their `http.Client` from an `HTTPClients` cache, one client per TLS config, since the proxy serves tenants on different configs in alternation. A tuple with a `tls` block has a config of its own, and so a client of its own in each cache, which lasts as long as its pool: `Reconcile` also returns the tuples it released, and `HTTPClients.Release` drops each one's client and closes its idle connections after the grace its pool closes after, so a request in flight finishes ([#713](https://github.com/Wave-RF/WaveHouse/issues/713)). The tuples with a zero `tls` block share the nil config and with it one client, which stays. - **errclass.go** — `Classify`, what a failed ClickHouse request says about the request: `Unavailable` (connection refused/reset, timeouts, and the exception codes of a server that cannot take work — `TIMEOUT_EXCEEDED`, `TOO_MANY_SIMULTANEOUS_QUERIES`, `MEMORY_LIMIT_EXCEEDED`, `TOO_MANY_PARTS`, `READONLY`, `TABLE_IS_READ_ONLY`, `KEEPER_EXCEPTION`, …), `Denied` (`AUTHENTICATION_FAILED`, `ACCESS_DENIED`, …: the identity, not the request), `Rejected` (any other exception code — the server read the request and refused it), or `Unknown` (no exception code, and no failure recognizable as the way to ClickHouse). It reads the driver's `*clickhouse.Exception`/`*clickhouse.HTTPError` and the HTTP interface's `HTTPError` (`NewHTTPError`: the code from `X-ClickHouse-Exception-Code`, else the body's `Code: NNN.`), so it is one answer for the ingest worker and the query handlers (`api/ch_errors.go`); the worker retries every class but `Rejected`. `TableScoped` and `Splittable` refine a retried failure: the first names a failure of one table (read-only, too many parts or mutations, a missing grant), the second one a multi-row batch can earn by its size alone (`TOO_MANY_PARTS` for too many partitions in one INSERT, `MEMORY_LIMIT_EXCEEDED`), which the worker splits row by row before retrying. ### `chsql/` — ClickHouse SQL Helpers diff --git a/internal/api/query.go b/internal/api/query.go index fa782b0c..5bb60abd 100644 --- a/internal/api/query.go +++ b/internal/api/query.go @@ -127,6 +127,10 @@ func NewQueryHandler(target func(*settings.Store) chconn.Target, queryTimeout fu } } +// Clients is the proxy's client cache, for the wiring to release a tuple's +// client from as a reload releases the tuple's pool. +func (h *QueryHandler) Clients() *chconn.HTTPClients { return h.clients } + // proxyHTTPClient is the proxy's client for the ClickHouse HTTP interface: // net/http's default transport with the target's TLS config for an https // target. ClickHouse's HTTP interface doesn't 3xx in normal operation, and diff --git a/internal/api/query_test.go b/internal/api/query_test.go index 30b41933..a81e4d71 100644 --- a/internal/api/query_test.go +++ b/internal/api/query_test.go @@ -7,9 +7,11 @@ import ( "crypto/x509" "encoding/json" "io" + "net" "net/http" "net/http/httptest" "strings" + "sync/atomic" "testing" "time" @@ -60,6 +62,34 @@ func postQuery(h *QueryHandler, body []byte) *httptest.ResponseRecorder { return w } +// The cache the proxy hands the wiring is the one its queries go through: a +// client released from it takes the connection a query left idle with it. +func TestQueryHandler_ClientsIsTheCacheItsQueriesUse(t *testing.T) { + t.Parallel() + var closed atomic.Int32 + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte(`{"data":[]}`)) + })) + srv.Config.ConnState = func(_ net.Conn, state http.ConnState) { + if state == http.StateClosed { + closed.Add(1) + } + } + srv.Start() + t.Cleanup(srv.Close) + tlsCfg := &tls.Config{MinVersion: tls.VersionTLS12} + h := newTestQueryHandler(func(*settings.Store) chconn.Target { + return chconn.Target{URL: srv.URL, TLS: tlsCfg} + }, func(*settings.Store) time.Duration { return 30 * time.Second }) + + rec := postQuery(h, []byte(`{"sql":"SELECT 1"}`)) + require.Equal(t, http.StatusOK, rec.Code, "body: %s", rec.Body.String()) + require.Zero(t, closed.Load(), "the query's connection is kept for the next one") + + h.Clients().Release(chconn.Released{TLS: tlsCfg}) + assert.Eventually(t, func() bool { return closed.Load() == 1 }, 5*time.Second, 5*time.Millisecond) +} + // assertSecurityHeaders confirms Cache-Control: no-store and // X-Content-Type-Options: nosniff are set. Both are set unconditionally at // handler entry (see query.go:Handle), so every response — 200, 4xx, 5xx, diff --git a/internal/app/app.go b/internal/app/app.go index 6762a303..f6878ec9 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -101,6 +101,10 @@ type App struct { pools *chconn.Pools bootState *api.BootState discoveries *discoveries + // httpClients is the client caches of the HTTP-interface consumers this + // process runs, the ingest worker's and the query proxy's: a reload + // releases from each the clients of the tuples it released. + httpClients []*chconn.HTTPClients // dedup is one store per tenant, each following its own folder's switch, // and dedupeStats the figures of the one Pebble instance they share. dedup *dedupe.Stores diff --git a/internal/app/app_test.go b/internal/app/app_test.go index 7fbc1c1b..5474fe5b 100644 --- a/internal/app/app_test.go +++ b/internal/app/app_test.go @@ -2041,6 +2041,65 @@ func TestReload_TenantGoneReleasesItsPoolAndRegistry(t *testing.T) { assert.True(t, dup, "an id acme sent before the removal is still a duplicate") } +// tlsSettings is poolSettings with both ClickHouse hops on TLS, so the +// tenant's tuple has a TLS config of its own for its HTTP clients to be keyed +// on (a zero tls block's is the nil every such tuple shares). +func tlsSettings(addr string) map[string]any { + p := poolSettings(addr, 10) + ch := p["clickhouse"].(map[string]any) + ch["http_scheme"] = "https" + ch["tls"].(map[string]any)["enabled"] = true + return p +} + +// A tenant's HTTP clients go with its pool (#713): once a removed tenant's +// tuple is released and its grace has passed, neither the ingest worker's +// cache nor the query proxy's holds the client it made for that tuple. The +// tenant beside it keeps its clients. +func TestReload_RemovedTenantsHTTPClientsGoWithItsPool(t *testing.T) { + root := writeNestedSettings(t, map[string]map[string]any{ + "acme": tlsSettings(closedAddr(t)), + "globex": tlsSettings(closedAddr(t)), + }) + a := newApp(t, testConfig(t, root), Options{}) + // A tuple is released after the longest query_timeout among its tenants, + // a second at the least in a settings directory: the pools are asked for + // a millisecond instead, so the test does not wait one out. + members := a.chMembers() + for i := range members { + members[i].Params.QueryTimeout = time.Millisecond + } + _, _, err := a.pools.Reconcile(members) + require.NoError(t, err) + require.Len(t, a.httpClients, 2, "the ingest worker's and the query proxy's") + acme, globex := a.pools.Target("acme"), a.pools.Target("globex") + require.NotNil(t, acme.TLS) + require.NotSame(t, acme.TLS, globex.TLS, "a config per tuple") + type made struct{ acme, globex *http.Client } + before := make([]made, len(a.httpClients)) + for i, clients := range a.httpClients { + before[i] = made{acme: clients.For(acme), globex: clients.For(globex)} + } + + require.NoError(t, os.RemoveAll(filepath.Join(root, "acme"))) + a.tenants.Reload("test") + require.Nil(t, a.pools.For("acme"), "its pool is released") + // For makes a client for a config it holds none for, so a new one means + // the one made before the reload is gone. + assert.Eventually(t, func() bool { + for i, clients := range a.httpClients { + if clients.For(acme) == before[i].acme { + return false + } + } + return true + }, 5*time.Second, 5*time.Millisecond, "a cache still holds the removed tenant's client past its grace") + for i, clients := range a.httpClients { + assert.NotSame(t, before[i].acme, clients.For(acme), "cache %d", i) + assert.Same(t, before[i].globex, clients.For(globex), "cache %d keeps the other tenant's", i) + } +} + // Over a nested directory the probes read every tenant together: /livez is // degraded while no tenant has completed a first discovery, names the tenant // in its diagnostic, and turns 200 for good at the first success, whatever diff --git a/internal/app/wire.go b/internal/app/wire.go index 12b4105e..80c8d7ab 100644 --- a/internal/app/wire.go +++ b/internal/app/wire.go @@ -257,6 +257,24 @@ func (a *App) wireObservability(ctx context.Context) { } } +// chMembers is what each served tenant asks of the ClickHouse pools: its +// settings' clickhouse block, with the boot-config password. +func (a *App) chMembers() []chconn.Member { + var ms []chconn.Member + for id, store := range a.tenants.All() { + c := store.ClickHouse() + ms = append(ms, chconn.Member{Tenant: id, Params: chconn.Params{ + Addr: c.Addr, HTTPPort: c.HTTPPort, HTTPScheme: c.HTTPScheme, + Database: c.Database, Username: c.Username, Password: a.cfg.ClickHouse.Password, + QueryTimeout: c.QueryTimeout, + TLS: chconn.TLS(c.TLS), + Headers: c.Headers, + MaxOpenConns: c.MaxOpenConns, MaxIdleConns: c.MaxIdleConns, + }}) + } + return ms +} + // wireClickHouse opens the ClickHouse pools: one per distinct address, // database, user, password and tls tuple among the served tenants' clickhouse // blocks (with the boot-config password), shared by the tenants naming it and @@ -274,29 +292,14 @@ func (a *App) wireObservability(ctx context.Context) { // where it already does (schema discovery retries, /readyz, query errors). // Every consumer resolves its tenant's pool per call (chConn, chTargetFor). func (a *App) wireClickHouse() error { - members := func() []chconn.Member { - var ms []chconn.Member - for id, store := range a.tenants.All() { - c := store.ClickHouse() - ms = append(ms, chconn.Member{Tenant: id, Params: chconn.Params{ - Addr: c.Addr, HTTPPort: c.HTTPPort, HTTPScheme: c.HTTPScheme, - Database: c.Database, Username: c.Username, Password: a.cfg.ClickHouse.Password, - QueryTimeout: c.QueryTimeout, - TLS: chconn.TLS(c.TLS), - Headers: c.Headers, - MaxOpenConns: c.MaxOpenConns, MaxIdleConns: c.MaxIdleConns, - }}) - } - return ms - } - pools, err := chconn.NewPools(a.cfg.ClickHouse.MaxTotalConns, members()) + pools, err := chconn.NewPools(a.cfg.ClickHouse.MaxTotalConns, a.chMembers()) if err != nil { return fmt.Errorf("clickhouse open: %w", err) } a.pools = pools a.add(component{name: "clickhouse", close: withoutContext(pools.Close)}) a.tenants.AfterAdopt(func([]tenant.ID) { - stale, err := pools.Reconcile(members()) + stale, released, err := pools.Reconcile(a.chMembers()) if err != nil { slog.Error("clickhouse pools reconciled in part; the next reload retries", "error", err) } @@ -310,6 +313,16 @@ func (a *App) wireClickHouse() error { if a.discoveries != nil { a.discoveries.drop(stale) } + // A released tuple's HTTP clients go with its pool (#713): each + // consumer's cache drops the client it made for the tuple, idle + // connections closed, once the pool's grace has passed. By the + // tuples released, not the stale tenants: a tenant moved to another + // user reads the same tables and still left a tuple behind. + for _, r := range released { + for _, clients := range a.httpClients { + clients.Release(r) + } + } // A tenant back on a pool after an absence was out of the cache // fan-out (sharedTables) while away, and one moved to another // address or database now reads other tables: either way what it @@ -800,6 +813,10 @@ func (a *App) wireStreaming() { // it consumes only the shards this process is assigned (ingest.ClaimShards), // so each table has one writer. func (a *App) wireIngestWorker() { + // Built here rather than by the worker, which exists only once Run + // starts it, so the pools hook releases from it from the first reload on. + clients := ingest.NewHTTPClients() + a.httpClients = append(a.httpClients, clients) a.add(component{name: "ingest worker", run: func(ctx context.Context) error { var queue ingest.Queue = a.mq if _, ok := a.mq.(mq.Sharded); ok { @@ -810,7 +827,7 @@ func (a *App) wireIngestWorker() { return err } } - stop, failed, err := ingest.StartIngestWorker(ctx, queue, sharedTables{Cache: a.cache, sharing: a.pools.SharingTables}, a.pools.Target, dlqFor(a.tenants)) + stop, failed, err := ingest.StartIngestWorker(ctx, queue, sharedTables{Cache: a.cache, sharing: a.pools.SharingTables}, clients, a.pools.Target, dlqFor(a.tenants)) if err != nil { return err } @@ -1008,9 +1025,11 @@ func (a *App) wireHTTP(authMW func(http.Handler) http.Handler) { // /v1/ops/query proxies straight to ClickHouse over HTTP — no native // driver involvement. The HTTP target of the tenant ?tenant= names, - // resolved per request like the ingest worker's. + // resolved per request like the ingest worker's; like the worker's too, + // its clients are released with their tuples' pools (httpClients). queryHandler := api.NewQueryHandler(a.chTargetFor, queryTimeout) queryHandler.Tenants = a.tenants + a.httpClients = append(a.httpClients, queryHandler.Clients()) deps := api.Dependencies{ Ingest: ingestHandler, diff --git a/internal/chconn/chconn.go b/internal/chconn/chconn.go index 07d9ae59..1ce25718 100644 --- a/internal/chconn/chconn.go +++ b/internal/chconn/chconn.go @@ -154,8 +154,9 @@ type Target struct { Password string Database string // TLS configures an https URL; nil means net/http's defaults. The - // pointer is the tuple's, so it changes only when the tenant moves to a - // tuple with another tls block, which is what HTTPClients keys on. + // pointer is the tuple's — nil for a zero tls block, else the config + // built when the tuple's pool opened — so it changes only when the + // tenant moves to another tuple, which is what HTTPClients keys on. TLS *tls.Config // Headers go on every request ahead of the consumer's own; read-only. Headers map[string]string @@ -388,6 +389,14 @@ func (s *snapshot) clone() *snapshot { return next } +// Released is a tuple a reconcile released: the TLS config its tenants' +// Targets carried, which an HTTPClients keys the tuple's client on, and the +// grace its pool closes after. +type Released struct { + TLS *tls.Config + Grace time.Duration +} + // Pools holds one Manager per tuple the served tenants name, reconciled // after every settings reload: a new tuple opens a pool (no dial), a tenant // whose tuple changed is repointed, a tuple no tenant names any more is @@ -413,7 +422,7 @@ type Pools struct { // boot: nothing is left open. func NewPools(ceiling int, want []Member) (*Pools, error) { p := newPools(ceiling, dial) - if _, err := p.Reconcile(want); err != nil { + if _, _, err := p.Reconcile(want); err != nil { _ = p.Close() return nil, err } @@ -433,7 +442,10 @@ func newPools(ceiling int, d dialer) *Pools { // fan-out (SharingTables) while away — and the ones it moved to another // address or database, whose cached results were read from other tables. A // move that keeps the address and database (a username or tls change) reads -// the same tables and is not stale. The walk keeps the ceiling at every +// the same tables and is not stale. It returns the tuples it released as +// well, for the caller to release their HTTP clients (HTTPClients.Release): +// each one no tenant names any more, its tenants removed or moved, a move +// that is not stale included. The walk keeps the ceiling at every // step: tenants no longer wanted leave first, then each wanted tenant is // placed in turn, and a refused placement — the ceiling, or a pool that // cannot be opened — is undone before the next, so a refused move leaves the @@ -441,7 +453,7 @@ func newPools(ceiling int, d dialer) *Pools { // placed once more at the end, since a shrink or a move placed after it may // have freed the budget it needed; only what the second pass refuses is // reported. -func (p *Pools) Reconcile(want []Member) (stale []tenant.ID, err error) { +func (p *Pools) Reconcile(want []Member) (stale []tenant.ID, released []Released, err error) { p.mu.Lock() defer p.mu.Unlock() cur := p.cur.Load() @@ -492,6 +504,7 @@ func (p *Pools) Reconcile(want []Member) (stale []tenant.ID, err error) { _ = t.manager.Close() } else { t.manager.Release(grace) + released = append(released, Released{TLS: t.manager.tlsCfg, Grace: grace}) } delete(w.next.tuples, id) case s != t.manager.Sizes(): @@ -508,7 +521,7 @@ func (p *Pools) Reconcile(want []Member) (stale []tenant.ID, err error) { stale = append(stale, id) } } - return stale, errors.Join(w.errs...) + return stale, released, errors.Join(w.errs...) } // walk is one Reconcile's working state: the snapshot being built, and the @@ -774,11 +787,13 @@ func (p *Pools) Close() error { // HTTPClients hands an HTTP-interface consumer the client for a target: one // per TLS config, made by the consumer's own factory (the worker's tuned -// transport, the proxy's redirect policy) and kept for the process lifetime. -// A config is built each time a pool opens on a tls block, so the set grows -// with the pools ever opened on one — a block whose pool closed and -// reopened adds another — not with requests; a client whose pool is gone -// keeps only its transport, whose idle connections time out on their own. +// transport, the proxy's redirect policy). A tuple with a tls block has a +// config of its own, built when its pool opened, so its client lasts as long +// as its pool does: Release drops it once the tuple is released, and a tuple +// that opens again gets another. The tuples with a zero tls block share the +// nil config and so one client, which stays — nothing in it is one tuple's, +// and its connections to an address no tuple names any more idle out on the +// transport's own timeout. // // The factory gets a copy of the config, never the target's own: net/http // appends its HTTP/2 protocols to TLSClientConfig.NextProtos in place when @@ -809,3 +824,24 @@ func (c *HTTPClients) For(t Target) *http.Client { c.clients[t.TLS] = cl return cl } + +// Release drops the client of a tuple a reconcile released and closes its +// idle connections, after the grace the tuple's pool closes after +// (Manager.Release): until then a consumer that resolved the tuple's Target +// just before the reload's swap still finds its client. A request in flight +// when the grace ends finishes: only idle connections are closed. The nil +// config is no one tuple's, so its client is not dropped. +func (c *HTTPClients) Release(r Released) { + if r.TLS == nil { + return + } + time.AfterFunc(r.Grace, func() { + c.mu.Lock() + cl, ok := c.clients[r.TLS] + delete(c.clients, r.TLS) + c.mu.Unlock() + if ok { + cl.CloseIdleConnections() + } + }) +} diff --git a/internal/chconn/chconn_test.go b/internal/chconn/chconn_test.go index 96b1c538..c9672fb3 100644 --- a/internal/chconn/chconn_test.go +++ b/internal/chconn/chconn_test.go @@ -10,7 +10,10 @@ import ( "crypto/x509/pkix" "encoding/pem" "errors" + "io" + "maps" "math/big" + "net" "net/http" "net/http/httptest" "os" @@ -333,7 +336,7 @@ func member(id tenant.ID, addr string, open, idle int) Member { func TestPools_DifferentTuplesGetDifferentPools(t *testing.T) { t.Parallel() p := newPools(0, fakeDial) - stale, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) + stale, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) require.NoError(t, err) assert.Equal(t, []tenant.ID{"acme", "globex"}, stale) require.NotNil(t, p.For("acme")) @@ -356,7 +359,7 @@ func TestPools_SharedTupleGetsOnePool(t *testing.T) { p := newPools(0, rec.dial) globex := member("globex", "a:9000", 8, 8) globex.Params.HTTPPort, globex.Params.Headers = 8443, map[string]string{"X-Proxy-Token": "g"} - _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), globex}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), globex}) require.NoError(t, err) require.NotNil(t, p.For("acme")) assert.Same(t, p.For("acme"), p.For("globex")) @@ -377,7 +380,7 @@ func TestPools_NewPoolNeverOpensAboveTheCeiling(t *testing.T) { t.Parallel() rec := &dialRecorder{} p := newPools(10, rec.dial) - _, err := p.Reconcile([]Member{member("acme", "a:9000", 1, 1), member("globex", "a:9000", 20, 5)}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 1, 1), member("globex", "a:9000", 20, 5)}) require.ErrorContains(t, err, "not grown for tenant globex") require.Equal(t, 1, rec.count("a:9000")) assert.Equal(t, Sizes{MaxOpenConns: 1, MaxIdleConns: 1}, rec.latest("a:9000").sizes, "opened at acme's ask, within the ceiling") @@ -396,7 +399,7 @@ func TestPools_TupleChangeRepointsWithoutTouchingTheOther(t *testing.T) { p := newPools(40, rec.dial) acme, globex := member("acme", "a:9000", 10, 5), member("globex", "a:9000", 20, 8) acme.Params.QueryTimeout, globex.Params.QueryTimeout = 30*time.Millisecond, 10*time.Millisecond - _, err := p.Reconcile([]Member{acme, globex}) + _, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) shared := p.For("acme") require.Same(t, shared, p.For("globex")) @@ -404,7 +407,7 @@ func TestPools_TupleChangeRepointsWithoutTouchingTheOther(t *testing.T) { before := rec.latest("a:9000") globex.Params.Username = "reporting" - stale, err := p.Reconcile([]Member{acme, globex}) + stale, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) assert.Empty(t, stale, "a new username reads the same tables: not stale") assert.Same(t, shared, p.For("acme"), "acme keeps its Manager") @@ -426,20 +429,20 @@ func TestPools_MoveToOtherTablesIsStale(t *testing.T) { t.Parallel() p := newPools(0, fakeDial) acme, globex := member("acme", "a:9000", 10, 5), member("globex", "a:9000", 10, 5) - _, err := p.Reconcile([]Member{acme, globex}) + _, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) acme.Params.Addr = "b:9000" - stale, err := p.Reconcile([]Member{acme, globex}) + stale, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) assert.Equal(t, []tenant.ID{"acme"}, stale, "another address") globex.Params.Database = "other" - stale, err = p.Reconcile([]Member{acme, globex}) + stale, _, err = p.Reconcile([]Member{acme, globex}) require.NoError(t, err) assert.Equal(t, []tenant.ID{"globex"}, stale, "another database") - stale, err = p.Reconcile([]Member{acme, globex}) + stale, _, err = p.Reconcile([]Member{acme, globex}) require.NoError(t, err) assert.Empty(t, stale, "nothing moved") } @@ -457,14 +460,14 @@ func TestPools_UnopenablePoolKeepsTheMoverWhereItWas(t *testing.T) { return rec.dial(id, s, c) }) acme := member("acme", "a:9000", 10, 5) - _, err := p.Reconcile([]Member{acme}) + _, _, err := p.Reconcile([]Member{acme}) require.NoError(t, err) before := p.For("acme") beforeConn := rec.latest("a:9000") moved := member("acme", "bad:9000", 10, 5) moved.Params.HTTPPort = 8124 - stale, err := p.Reconcile([]Member{moved}) + stale, _, err := p.Reconcile([]Member{moved}) require.ErrorContains(t, err, "clickhouse pool bad:9000 database db user u not opened for tenant acme: malformed option") assert.ErrorContains(t, err, "tenant acme keeps its previous pool a:9000 database db user u") assert.Empty(t, stale) @@ -479,20 +482,21 @@ func TestPools_TenantGoneReleasesItsPoolAfterGrace(t *testing.T) { p := newPools(0, rec.dial) acme := member("acme", "a:9000", 10, 5) acme.Params.QueryTimeout = 20 * time.Millisecond - _, err := p.Reconcile([]Member{acme, member("globex", "b:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{acme, member("globex", "b:9000", 10, 5)}) require.NoError(t, err) c := rec.latest("a:9000") globexPool := p.For("globex") - stale, err := p.Reconcile([]Member{member("globex", "b:9000", 10, 5)}) + stale, released, err := p.Reconcile([]Member{member("globex", "b:9000", 10, 5)}) require.NoError(t, err) assert.Empty(t, stale) + assert.Equal(t, []Released{{Grace: 20 * time.Millisecond}}, released, "the tuple it left, with the grace its pool gets") assert.Nil(t, p.For("acme")) assert.Same(t, globexPool, p.For("globex")) assert.False(t, c.closed.Load(), "released after the grace, not at once") assert.Eventually(t, func() bool { return c.closed.Load() }, time.Second, 5*time.Millisecond) - stale, err = p.Reconcile([]Member{acme, member("globex", "b:9000", 10, 5)}) + stale, _, err = p.Reconcile([]Member{acme, member("globex", "b:9000", 10, 5)}) require.NoError(t, err) assert.Equal(t, []tenant.ID{"acme"}, stale, "back after an absence: its cache is stale") require.NotNil(t, p.For("acme")) @@ -506,7 +510,7 @@ func TestPools_CeilingRefusesBoot(t *testing.T) { t.Parallel() rec := &dialRecorder{} p := newPools(15, rec.dial) - _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) require.Error(t, err) assert.ErrorContains(t, err, "not opened for tenant globex") assert.ErrorContains(t, err, "clickhouse.max_open_conns 10") @@ -527,11 +531,11 @@ func TestPools_CeilingRefusesAThirdTupleThenOpensIt(t *testing.T) { rec := &dialRecorder{} p := newPools(25, rec.dial) acme, globex, initech := member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5), member("initech", "c:9000", 10, 5) - _, err := p.Reconcile([]Member{acme, globex}) + _, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) acmePool, globexPool := p.For("acme"), p.For("globex") - stale, err := p.Reconcile([]Member{acme, globex, initech}) + stale, _, err := p.Reconcile([]Member{acme, globex, initech}) require.ErrorContains(t, err, "clickhouse pool c:9000 database db user u not opened for tenant initech") assert.ErrorContains(t, err, "at 30, above clickhouse.max_total_conns 25") assert.Empty(t, stale) @@ -542,7 +546,7 @@ func TestPools_CeilingRefusesAThirdTupleThenOpensIt(t *testing.T) { assert.Equal(t, 1, rec.count("a:9000")) acme.Params.MaxOpenConns = 5 - stale, err = p.Reconcile([]Member{acme, globex, initech}) + stale, _, err = p.Reconcile([]Member{acme, globex, initech}) require.NoError(t, err) assert.Equal(t, []tenant.ID{"initech"}, stale) require.NotNil(t, p.For("initech")) @@ -557,11 +561,11 @@ func TestPools_RefusedResizeKeepsTheSize(t *testing.T) { t.Parallel() p := newPools(20, fakeDial) acme, globex := member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5) - _, err := p.Reconcile([]Member{acme, globex}) + _, _, err := p.Reconcile([]Member{acme, globex}) require.NoError(t, err) acme.Params.MaxOpenConns, acme.Params.HTTPPort = 15, 8124 - _, err = p.Reconcile([]Member{acme, globex}) + _, _, err = p.Reconcile([]Member{acme, globex}) require.ErrorContains(t, err, "clickhouse pool a:9000 database db user u not resized for tenants [acme]") assert.ErrorContains(t, err, "clickhouse.max_open_conns 15") assert.ErrorContains(t, err, "at 25, above clickhouse.max_total_conns 20") @@ -570,7 +574,7 @@ func TestPools_RefusedResizeKeepsTheSize(t *testing.T) { assert.Equal(t, "http://a:8124", p.Target("acme").URL, "the HTTP wiring applied all the same") globex.Params.MaxOpenConns = 5 - _, err = p.Reconcile([]Member{acme, globex}) + _, _, err = p.Reconcile([]Member{acme, globex}) require.NoError(t, err) assert.Equal(t, 15, p.For("acme").Sizes().MaxOpenConns, "retried by the next reload") assert.Equal(t, 5, p.For("globex").Sizes().MaxOpenConns) @@ -582,10 +586,10 @@ func TestPools_SharedGrowthIsOneRefusal(t *testing.T) { t.Parallel() p := newPools(20, fakeDial) acme, globex, initech := member("acme", "a:9000", 10, 5), member("globex", "a:9000", 10, 5), member("initech", "b:9000", 10, 5) - _, err := p.Reconcile([]Member{acme, globex, initech}) + _, _, err := p.Reconcile([]Member{acme, globex, initech}) require.NoError(t, err) acme.Params.MaxOpenConns, globex.Params.MaxOpenConns = 12, 12 - _, err = p.Reconcile([]Member{acme, globex, initech}) + _, _, err = p.Reconcile([]Member{acme, globex, initech}) require.Error(t, err) joined, ok := err.(interface{ Unwrap() []error }) require.True(t, ok) @@ -603,14 +607,14 @@ func TestPools_RefusedMoveKeepsThePreviousPool(t *testing.T) { p := newPools(15, rec.dial) acme := member("acme", "a:9000", 10, 5) acme.Params.QueryTimeout = 10 * time.Millisecond - _, err := p.Reconcile([]Member{acme}) + _, _, err := p.Reconcile([]Member{acme}) require.NoError(t, err) before := p.For("acme") beforeConn := rec.latest("a:9000") moved := member("acme", "b:9000", 20, 5) moved.Params.HTTPPort = 8124 - stale, err := p.Reconcile([]Member{moved}) + stale, _, err := p.Reconcile([]Member{moved}) require.ErrorContains(t, err, "clickhouse pool b:9000 database db user u not opened for tenant acme") assert.ErrorContains(t, err, "tenant acme keeps its previous pool a:9000 database db user u") assert.Empty(t, stale, "a refused move stays on the tables it had") @@ -620,7 +624,7 @@ func TestPools_RefusedMoveKeepsThePreviousPool(t *testing.T) { assert.Equal(t, 0, rec.count("b:9000")) moved.Params.MaxOpenConns = 15 - _, err = p.Reconcile([]Member{moved}) + _, _, err = p.Reconcile([]Member{moved}) require.NoError(t, err) assert.Equal(t, "b:9000", p.For("acme").Identity().Addr, "the next reload that fits moves it") assert.Equal(t, "http://b:8124", p.Target("acme").URL) @@ -633,9 +637,9 @@ func TestPools_RefusedMoveKeepsThePreviousPool(t *testing.T) { func TestPools_MoveAtTheCeilingIsAllowed(t *testing.T) { t.Parallel() p := newPools(10, fakeDial) - _, err := p.Reconcile([]Member{member("0", "a:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{member("0", "a:9000", 10, 5)}) require.NoError(t, err) - _, err = p.Reconcile([]Member{member("0", "b:9000", 10, 5)}) + _, _, err = p.Reconcile([]Member{member("0", "b:9000", 10, 5)}) require.NoError(t, err) assert.Equal(t, "b:9000", p.For("0").Identity().Addr) } @@ -647,18 +651,18 @@ func TestPools_UnreadableCertificateRefusesTheTuple(t *testing.T) { t.Parallel() p := newPools(0, fakeDial) acme := member("acme", "a:9000", 10, 5) - _, err := p.Reconcile([]Member{acme}) + _, _, err := p.Reconcile([]Member{acme}) require.NoError(t, err) before := p.For("acme") acme.Params.TLS = TLS{Enabled: true, CAFile: "/nowhere/ca.pem"} - _, err = p.Reconcile([]Member{acme}) + _, _, err = p.Reconcile([]Member{acme}) require.ErrorContains(t, err, "clickhouse.tls.ca_file") assert.ErrorContains(t, err, "/nowhere/ca.pem") assert.ErrorContains(t, err, "keeps its previous pool") assert.Same(t, before, p.For("acme")) - _, err = newPools(0, fakeDial).Reconcile([]Member{acme}) + _, _, err = newPools(0, fakeDial).Reconcile([]Member{acme}) require.ErrorContains(t, err, "not opened for tenant acme") } @@ -672,7 +676,7 @@ func TestPools_PingFirstSuccessWins(t *testing.T) { t.Parallel() rec := &dialRecorder{ping: func(context.Context) error { return errors.New("connection refused") }} p := newPools(0, rec.dial) - _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) require.NoError(t, err) err = p.Ping(context.Background()) require.Error(t, err) @@ -688,7 +692,7 @@ func TestPools_PingFirstSuccessWins(t *testing.T) { return ctx.Err() }} p := newPools(0, rec.dial) - _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) require.NoError(t, err) rec.latest("b:9000").ping = nil ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) @@ -703,7 +707,7 @@ func TestPools_CloseClosesEveryPool(t *testing.T) { t.Parallel() rec := &dialRecorder{} p := newPools(0, rec.dial) - _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) + _, _, err := p.Reconcile([]Member{member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5)}) require.NoError(t, err) require.NoError(t, p.Close()) assert.True(t, rec.latest("a:9000").closed.Load()) @@ -793,3 +797,209 @@ func TestHTTPClients_HandsEachClientItsOwnTLSConfig(t *testing.T) { assert.Nil(t, shared.NextProtos, "the target's config, which the driver dials with, is untouched") assert.NotSame(t, handed[0], handed[1]) } + +// tlsMember is member with a tls block, so its tuple has a TLS config of its +// own (a zero block's is the nil every such tuple shares), released after +// grace. +func tlsMember(id tenant.ID, addr string, grace time.Duration) Member { + m := member(id, addr, 10, 5) + m.Params.TLS = TLS{ServerName: "clickhouse.internal"} + m.Params.QueryTimeout = grace + return m +} + +// countingClient is a client over a closeCounter. +func countingClient(*tls.Config) *http.Client { + return &http.Client{Transport: &closeCounter{}} +} + +func idleCloses(c *http.Client) int32 { return c.Transport.(*closeCounter).closed.Load() } + +// held is the clients c holds, by TLS config. Look a config up by index: +// testify compares pointer keys by what they point to. +func held(c *HTTPClients) map[*tls.Config]*http.Client { + c.mu.Lock() + defer c.mu.Unlock() + return maps.Clone(c.clients) +} + +// TestHTTPClients_FollowTheReleasedTuple is #713: a tuple a reconcile +// releases takes its HTTP client with it once the pool's grace has passed, +// idle connections closed. Here the tenant rotates to another user on the +// same tables, the release no stale list reports. The tuple another tenant +// still names keeps its client, and the released one, opened again, gets a +// fresh client rather than one more. +func TestHTTPClients_FollowTheReleasedTuple(t *testing.T) { + t.Parallel() + const grace = 20 * time.Millisecond + p := newPools(0, fakeDial) + clients := NewHTTPClients(countingClient) + // The wiring's hook: the pools, then the clients of what they released. + reconcile := func(want ...Member) ([]tenant.ID, []Released) { + t.Helper() + stale, released, err := p.Reconcile(want) + require.NoError(t, err) + for _, r := range released { + clients.Release(r) + } + return stale, released + } + acme, globex, initech := tlsMember("acme", "a:9000", grace), tlsMember("globex", "b:9000", grace), tlsMember("initech", "b:9000", grace) + _, released := reconcile(acme, globex, initech) + require.Empty(t, released, "nothing was open to release") + left, shared := p.Target("acme"), p.Target("initech") + leftClient, sharedClient := clients.For(left), clients.For(shared) + require.NotSame(t, leftClient, sharedClient) + require.Same(t, sharedClient, clients.For(p.Target("globex")), "one tuple, one client") + + rotated := acme + rotated.Params.Username = "reporting" + stale, released := reconcile(rotated, initech) + assert.Empty(t, stale, "the same tables: no stale tenant reports this release") + require.Len(t, released, 1) + assert.Same(t, left.TLS, released[0].TLS, "the tuple acme left, not the one initech still names") + assert.Equal(t, grace, released[0].Grace) + require.Eventually(t, func() bool { return idleCloses(leftClient) == 1 }, time.Second, 5*time.Millisecond, "its idle connections close after the grace") + assert.Nil(t, held(clients)[left.TLS], "and its client is gone") + assert.Same(t, sharedClient, clients.For(shared), "a tuple another tenant still names keeps its client") + assert.Zero(t, idleCloses(sharedClient)) + rotatedClient := clients.For(p.Target("acme")) + + _, released = reconcile(acme, initech) + require.Len(t, released, 1, "the tuple of the user it rotated to") + reopened := p.Target("acme") + require.NotSame(t, left.TLS, reopened.TLS, "the pool opened again, on a config of its own") + assert.NotSame(t, leftClient, clients.For(reopened), "a reopened tuple gets a fresh client") + require.Eventually(t, func() bool { return idleCloses(rotatedClient) == 1 }, time.Second, 5*time.Millisecond) + assert.Len(t, held(clients), 2, "only the open tuples have a client: the set does not grow with the pools ever opened") +} + +// A released tuple's client is dropped after the grace, not at once: a +// consumer that resolved the Target just before the reload's swap still +// finds it. +func TestHTTPClients_ReleaseWaitsOutTheGrace(t *testing.T) { + t.Parallel() + clients := NewHTTPClients(countingClient) + target := Target{TLS: &tls.Config{MinVersion: tls.VersionTLS12}} + client := clients.For(target) + clients.Release(Released{TLS: target.TLS, Grace: time.Hour}) + assert.Same(t, client, clients.For(target)) + assert.Zero(t, idleCloses(client)) +} + +// Tuples with a zero tls block share the nil config and so one client, which +// the release of one of them must not take from the others. +func TestHTTPClients_ZeroTLSBlockClientStays(t *testing.T) { + t.Parallel() + p := newPools(0, fakeDial) + clients := NewHTTPClients(countingClient) + acme, globex := member("acme", "a:9000", 10, 5), member("globex", "b:9000", 10, 5) + acme.Params.QueryTimeout = 0 + _, _, err := p.Reconcile([]Member{acme, globex}) + require.NoError(t, err) + plain := clients.For(p.Target("acme")) + require.Same(t, plain, clients.For(p.Target("globex")), "one client for every zero block") + own := Target{TLS: &tls.Config{MinVersion: tls.VersionTLS12}} + ownClient := clients.For(own) + + _, released, err := p.Reconcile([]Member{globex}) + require.NoError(t, err) + require.Equal(t, []Released{{}}, released, "acme's tuple: no config of its own, no grace") + clients.Release(released[0]) + // Released after it with the same grace: once this one has landed, a + // release of the nil config would have too. + clients.Release(Released{TLS: own.TLS}) + require.Eventually(t, func() bool { return idleCloses(ownClient) == 1 }, time.Second, 5*time.Millisecond) + assert.Same(t, plain, clients.For(p.Target("globex"))) + assert.Zero(t, idleCloses(plain)) +} + +// connStates is a test server's connections, each by its latest state. +type connStates struct { + mu sync.Mutex + states map[net.Conn]http.ConnState +} + +func (c *connStates) track(conn net.Conn, state http.ConnState) { + c.mu.Lock() + defer c.mu.Unlock() + if c.states == nil { + c.states = map[net.Conn]http.ConnState{} + } + c.states[conn] = state +} + +// count is how many connections are open and how many closed. +func (c *connStates) count() (open, closed int) { + c.mu.Lock() + defer c.mu.Unlock() + for _, state := range c.states { + if state == http.StateClosed { + closed++ + } else { + open++ + } + } + return open, closed +} + +// TestHTTPClients_ReleaseClosesIdleConnectionsOnly runs Release against +// net/http itself, with the server counting its connections: the idle one a +// released tuple's client held is closed when the grace ends, and a request +// in flight then finishes, its connection closing once it has. +func TestHTTPClients_ReleaseClosesIdleConnectionsOnly(t *testing.T) { + t.Parallel() + var conns connStates + entered, finish := make(chan struct{}), make(chan struct{}) + srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/slow" { + close(entered) + <-finish + } + w.WriteHeader(http.StatusOK) + })) + srv.Config.ConnState = conns.track + srv.Start() + t.Cleanup(srv.Close) + // Registered after srv.Close, so run before it: a failed assertion must + // not leave the slow request holding the server open. + finishSlow := sync.OnceFunc(func() { close(finish) }) + t.Cleanup(finishSlow) + + clients := NewHTTPClients(func(cfg *tls.Config) *http.Client { + transport := http.DefaultTransport.(*http.Transport).Clone() + transport.TLSClientConfig = cfg + return &http.Client{Transport: transport} + }) + target := Target{URL: srv.URL, TLS: &tls.Config{MinVersion: tls.VersionTLS12}} + get := func(path string) error { + req, err := http.NewRequestWithContext(t.Context(), http.MethodGet, target.URL+path, nil) + if err != nil { + return err + } + resp, err := clients.For(target).Do(req) + if err != nil { + return err + } + _, _ = io.Copy(io.Discard, resp.Body) + return resp.Body.Close() + } + + slow := make(chan error, 1) + go func() { slow <- get("/slow") }() + <-entered + require.NoError(t, get("/"), "a second connection, idle once answered") + open, closed := conns.count() + require.Equal(t, 2, open) + require.Zero(t, closed) + + clients.Release(Released{TLS: target.TLS}) + require.Eventually(t, func() bool { _, closed := conns.count(); return closed == 1 }, 5*time.Second, 5*time.Millisecond, "the idle connection closes when the grace ends") + assert.Empty(t, held(clients)) + open, _ = conns.count() + assert.Equal(t, 1, open, "the request in flight keeps its connection") + + finishSlow() + require.NoError(t, <-slow, "and finishes on it") + assert.Eventually(t, func() bool { open, closed := conns.count(); return open == 0 && closed == 2 }, 5*time.Second, 5*time.Millisecond, "its connection closes once it has") +} diff --git a/internal/ingest/worker.go b/internal/ingest/worker.go index 523eb1c8..e89d4bbf 100644 --- a/internal/ingest/worker.go +++ b/internal/ingest/worker.go @@ -57,8 +57,9 @@ type IngestWorker struct { // consumer could not start, or delivery ended underneath it. Buffered so // the dispatch loop never blocks on a caller that has already gone. failed chan error - // clients follows the target's TLS config: one client per config, - // replaced when a reload changes it (chconn.HTTPClients). + // clients follows the target's TLS config: one client per config + // (NewHTTPClients), the caller's to release from as a reload releases a + // tuple's pool. clients *chconn.HTTPClients cache cache.Cache // target resolves a tenant's ClickHouse HTTP wiring per insert @@ -153,9 +154,12 @@ const ( // the caller must treat it as fatal (internal/app returns it from Run, which // stops the process — the next boot recreates the consumer). The worker has // already flushed and acked what it held by the time failed fires; stop is -// still the caller's to call. +// still the caller's to call. clients is the worker's client cache +// (NewHTTPClients), built by the caller so it can release a tuple's client +// from it whether or not the worker has started. func StartIngestWorker( ctx context.Context, queue Queue, cache cache.Cache, + clients *chconn.HTTPClients, target func(tenant.ID) chconn.Target, dlqEnabled func(id tenant.ID, table string) bool, ) (stop func(context.Context) error, failed <-chan error, err error) { @@ -165,6 +169,9 @@ func StartIngestWorker( if cache == nil { return nil, nil, fmt.Errorf("cache is nil") } + if clients == nil { + return nil, nil, fmt.Errorf("clickhouse http clients is nil") + } if target == nil { return nil, nil, fmt.Errorf("clickhouse target is nil") } @@ -181,7 +188,7 @@ func StartIngestWorker( worker := &IngestWorker{ dlq: queue, failed: make(chan error, 1), - clients: chconn.NewHTTPClients(ingestHTTPClient), + clients: clients, cache: cache, target: target, maxBatch: defaultMaxBatch, @@ -202,6 +209,10 @@ func StartIngestWorker( return stopFunc, worker.failed, nil } +// NewHTTPClients returns the client cache a worker inserts through +// (ingestHTTPClient). +func NewHTTPClients() *chconn.HTTPClients { return chconn.NewHTTPClients(ingestHTTPClient) } + // ingestHTTPClient is the worker's client for the ClickHouse HTTP // interface: a transport tuned for high-throughput inserts, with the // target's TLS config for an https target. diff --git a/internal/ingest/worker_test.go b/internal/ingest/worker_test.go index 26a6f747..7c3573b7 100644 --- a/internal/ingest/worker_test.go +++ b/internal/ingest/worker_test.go @@ -114,11 +114,13 @@ func TestStartIngestWorker_Validation(t *testing.T) { tests := []struct { name string setup func(t *testing.T) (Queue, cache.Cache) + clients *chconn.HTTPClients wantErrSub string }{ { name: "nil queue", setup: func(*testing.T) (Queue, cache.Cache) { return nil, &testutil.MockCache{} }, + clients: NewHTTPClients(), wantErrSub: "message queue is nil", }, { @@ -127,15 +129,23 @@ func TestStartIngestWorker_Validation(t *testing.T) { emb := testutil.NewEmbeddedMQ(t, 1024*1024) return emb, nil }, + clients: NewHTTPClients(), wantErrSub: "cache is nil", }, + { + name: "nil clients", + setup: func(t *testing.T) (Queue, cache.Cache) { + return testutil.NewEmbeddedMQ(t, 1024*1024), &testutil.MockCache{} + }, + wantErrSub: "clickhouse http clients is nil", + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() q, c := tt.setup(t) - _, _, err := StartIngestWorker(context.Background(), q, c, + _, _, err := StartIngestWorker(context.Background(), q, c, tt.clients, func(tenant.ID) chconn.Target { return chconn.Target{URL: "http://localhost:8123"} }, nil) require.Error(t, err) assert.Contains(t, err.Error(), tt.wantErrSub) @@ -261,7 +271,14 @@ func TestStartIngestWorker_StopFunc_RespectsShutdownDeadline(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) t.Cleanup(cancel) - stopFn, _, err := StartIngestWorker(ctx, emb, &testutil.MockCache{}, + // The cache is the caller's, which releases a tuple's client from it: + // the worker must insert through the one it is handed. + var built atomic.Int32 + clients := chconn.NewHTTPClients(func(cfg *tls.Config) *http.Client { + built.Add(1) + return ingestHTTPClient(cfg) + }) + stopFn, _, err := StartIngestWorker(ctx, emb, &testutil.MockCache{}, clients, func(tenant.ID) chconn.Target { return chconn.Target{URL: fmt.Sprintf("http://%s:%s", host, port), Username: "u", Password: "p", Database: "db"} }, nil) @@ -288,6 +305,7 @@ func TestStartIngestWorker_StopFunc_RespectsShutdownDeadline(t *testing.T) { err = stopFn(shutCtx) assert.ErrorIs(t, err, context.DeadlineExceeded) + assert.Eventually(t, func() bool { return built.Load() == 1 }, 5*time.Second, 10*time.Millisecond, "the insert went through the caller's cache") } // TestStartIngestWorker_StopFunc_CleanShutdown verifies the graceful path: with @@ -301,7 +319,7 @@ func TestStartIngestWorker_StopFunc_CleanShutdown(t *testing.T) { // chURL is never dialed: with no messages there is no flush, so a dummy // host/port is fine. - stopFn, _, err := StartIngestWorker(context.Background(), emb, &testutil.MockCache{}, + stopFn, _, err := StartIngestWorker(context.Background(), emb, &testutil.MockCache{}, NewHTTPClients(), func(tenant.ID) chconn.Target { return chconn.Target{URL: "http://localhost:8123"} }, nil) require.NoError(t, err) diff --git a/tests/e2e/sdk/admin.test.ts b/tests/e2e/sdk/admin.test.ts index 2d9a77c4..82a59a8d 100644 --- a/tests/e2e/sdk/admin.test.ts +++ b/tests/e2e/sdk/admin.test.ts @@ -1,7 +1,14 @@ import type { Pipe, Policy } from "@wavehouse/sdk"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; import { adminClient } from "./helpers.js"; -import { readPipesFile, readPolicyFile, setPipes, setPolicy } from "./settings.js"; +import { + readConfigFile, + readPipesFile, + readPolicyFile, + setConfig, + setPipes, + setPolicy, +} from "./settings.js"; import { suiteTables } from "./tables.js"; describe("Admin", () => { @@ -138,6 +145,29 @@ describe("Admin", () => { const unknown = await wh.settings.reload({ tenant: "acme" }); expect(unknown.error?.status).toBe(404); }); + + // A connection tuple is the address, database, user and tls block, so a + // tls.server_name the plaintext hops never read still names another one. + // Each reload opens the tuple it names and releases the one it left, the + // HTTP clients made for it included (#713); the next raw-SQL query runs + // on the tuple the tenant is on. query_timeout is the grace a released + // tuple gets, so the one moved to is gone five seconds after the reload + // back rather than thirty. + it("keeps serving across a reload that moves the tenant to another connection tuple", async () => { + const baseline = readConfigFile(); + const moved = structuredClone(baseline); + moved.clickhouse.tls.server_name = "clickhouse.e2e"; + moved.clickhouse.query_timeout = 5; + try { + await setConfig(moved); + const onMoved = await wh.sql("SELECT 1 AS one"); + expect(onMoved.error).toBeNull(); + } finally { + await setConfig(baseline); + } + const onRestored = await wh.sql("SELECT 1 AS one"); + expect(onRestored.error).toBeNull(); + }); }); describe("Pipes", () => { diff --git a/tests/e2e/sdk/settings.ts b/tests/e2e/sdk/settings.ts index 5387265b..46983078 100644 --- a/tests/e2e/sdk/settings.ts +++ b/tests/e2e/sdk/settings.ts @@ -38,6 +38,25 @@ function writeJSON(name: string, doc: unknown): void { writeFileSync(join(settingsDir(), name), `${JSON.stringify(doc, null, 2)}\n`); } +/** + * `config.json` as far as the suite reaches into it: the `clickhouse` block + * and the `tls` block inside it. Every other key rides along untouched. + */ +export type ConfigFile = Record & { + clickhouse: Record & { tls: Record }; +}; + +/** The configuration currently on disk (`config.json`). */ +export function readConfigFile(): ConfigFile { + return readJSON("config.json"); +} + +/** Replace `config.json` and reload. */ +export async function setConfig(config: ConfigFile): Promise { + writeJSON("config.json", config); + await reloadSettings(); +} + /** The policy currently on disk (`policies.json`). */ export function readPolicyFile(): Policy { return readJSON("policies.json"); diff --git a/tests/integration/ingest_outage_test.go b/tests/integration/ingest_outage_test.go index 53faa2d0..39b26ddf 100644 --- a/tests/integration/ingest_outage_test.go +++ b/tests/integration/ingest_outage_test.go @@ -53,7 +53,7 @@ func TestIngest_ClickHouseOutage_RetriedNotDeadLettered(t *testing.T) { target := func(tenant.ID) chconn.Target { return chconn.Target{URL: *chURL.Load(), Username: testCHUser, Password: testCHPassword, Database: testCHDatabase} } - stop, _, err := ingest.StartIngestWorker(ctx, broker, &testutil.MockCache{}, target, nil) + stop, _, err := ingest.StartIngestWorker(ctx, broker, &testutil.MockCache{}, ingest.NewHTTPClients(), target, nil) require.NoError(t, err) t.Cleanup(func() { stopCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) diff --git a/tests/integration/shard_order_test.go b/tests/integration/shard_order_test.go index f3827bbf..c4ae90bf 100644 --- a/tests/integration/shard_order_test.go +++ b/tests/integration/shard_order_test.go @@ -135,7 +135,7 @@ func startWorkerProc(t *testing.T, url, id string, lease time.Duration, cfg inge require.NoError(t, err) q, err := ingest.ClaimShards(p.broker, leases, cfg) require.NoError(t, err) - p.stop, p.failed, err = ingest.StartIngestWorker(context.WithoutCancel(ctx), q, &testutil.MockCache{}, + p.stop, p.failed, err = ingest.StartIngestWorker(context.WithoutCancel(ctx), q, &testutil.MockCache{}, ingest.NewHTTPClients(), func(tenant.ID) chconn.Target { return chconn.Target{URL: chURL, Database: "db"} }, nil) require.NoError(t, err) t.Cleanup(func() { _ = p.stop(context.Background()) })