diff --git a/CLAUDE.md b/CLAUDE.md index aac35ff..d9d1b82 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -283,8 +283,16 @@ default as everything else optional in this codebase); when they are, startup blocks on the first fetch succeeding and later refresh failures keep serving the last-known-good set rather than clearing it. Verified with real HTTP round trips against a hand-rolled TCP test server in this -environment, no live enterprise-auth needed. **The tenant-picker page is -now built too**: +environment, no live enterprise-auth needed. **This closing move exposed +the ClickHouse side's own gap by comparison** — `chwriter.Registry`'s +writer map was still a startup-only snapshot with no refresh at all, a +real asymmetry once Tantivy's tracker refreshed every minute and +ClickHouse's didn't — so `Registry.StartRefreshing` (new) closes that +too: same one-minute interval, same last-known-good posture on a failed +refresh, opening connections for newly-active tenants and closing ones +no longer active. Both engines now share the same active-tenant +staleness bound instead of one being materially staler than the other. +**The tenant-picker page is now built too**: `web/src/routes/select-tenant` calls `GET /auth/memberships`/ `POST /auth/select-tenant` via `fetch(..., {credentials: 'include'})` (new `$lib/api.ts` functions), which needed a second CORS posture diff --git a/docs/phase-4-runbook.md b/docs/phase-4-runbook.md index 869423a..561affb 100644 --- a/docs/phase-4-runbook.md +++ b/docs/phase-4-runbook.md @@ -785,19 +785,44 @@ go test ./internal/authhandler/... -run TestActiveTenants -v # whole reason to distinguish token kinds. ``` -**One asymmetry remains, disclosed rather than fixed**: `chwriter. -Registry`'s per-tenant writer map is a snapshot built once at -`enterprise-ingest` startup, never refreshed -- a tenant deprovisioned -after startup keeps writing successfully to ClickHouse until the next -restart, a real staleness window `ActiveTenantTracker`'s 60-second -refresh doesn't have on the Tantivy side. Neither is a live per-write -check (a database/HTTP round trip on every record would be a real -throughput cost neither implementation accepts), so both have *some* -staleness window by design -- the gap between the two windows is what's -disclosed as inconsistent here, not a claim that either is fully live. -Closing it would mean adding a periodic refresh to `chwriter.Registry` -too; not done in this pass. See `/docs/security/threat-model.md`'s "Read -this first". +**The ClickHouse/Tantivy asymmetry this section used to disclose is now +closed too.** `chwriter.Registry.StartRefreshing` (new) spawns a +goroutine that re-lists active tenants every minute (matching +`ActiveTenantTracker`'s interval -- see +`enterprise-ingest/main.go`'s `dataSourceRefreshInterval`) and +reconciles the writer map: opens a connection for a newly-active tenant, +closes and removes one no longer active. A tenant deprovisioned after +`enterprise-ingest` startup now loses ClickHouse write access within a +minute, not "until the next restart." A refresh failure (rbacstore +unreachable, or one tenant's new connection failing to open) logs and +leaves the existing map untouched for that tick -- the same +last-known-good posture `ActiveTenantTracker` uses, so a transient +Postgres blip doesn't evict every other tenant's already-working writer. +Neither engine does a live per-write check (a database/HTTP round trip +per record would be a real throughput cost neither implementation +accepts), so a roughly one-minute staleness window remains on both +sides by design, not a gap unique to either anymore. + +```sh +cd enterprise +go test ./internal/chwriter/... -run TestRefreshListerErrorLeavesRegistryUnchanged -v +# Docker-free -- refresh's early-return on a lister error, proven the +# same way the existing fail-closed tests are: a Registry constructed +# directly (bypassing New, so nothing dials ClickHouse), asserting +# WriteBatch still refuses afterward. + +go test ./internal/chwriter/... -run 'TestRefreshAddsNewlyActiveTenant|TestRefreshRemovesNoLongerActiveTenant' -v +# skip-gated (CHWRITER_TEST_CLICKHOUSE_ADDR) -- the actual add/remove +# reconciliation against real connections: a tenant absent at New() time +# gains a working writer after refresh() sees it in a later lister call; +# a tenant present at New() time loses its writer (WriteBatch starts +# refusing it) after refresh() stops seeing it. +``` + +**Not run in this environment**: same live-ClickHouse caveat as +everything else here -- the two new skip-gated tests are correct Go +that's never executed against a real database. See +`/docs/security/threat-model.md`'s "Read this first". **Also not built**: Helm/`docker-compose.yml` do gate *whether* `enterprise-ingest` runs at all (`ingest.requireTenantCredential`, same @@ -849,13 +874,12 @@ Full accounting: `/docs/security/threat-model.md`. Headline items: each record into its own tenant's Tantivy index -- genuinely verified in this environment, unlike the ClickHouse side, since Tantivy needs no Docker to exercise real logic. Both engines now also gate writes on - an active-tenant check: ClickHouse's is a startup-time snapshot with - no refresh (stale until the next `enterprise-ingest` restart), while - Tantivy's `ActiveTenantTracker` polls `enterprise-auth` every 60 - seconds, a tighter staleness bound the ClickHouse side wasn't updated - to match -- a real, disclosed asymmetry between the two, not a gap in - either alone; see §14 and `/docs/security/threat-model.md`'s "Read - this first". A + an active-tenant check that refreshes every minute -- `chwriter. + Registry.StartRefreshing` on the ClickHouse side (new, closing what + was originally a startup-only snapshot with no refresh at all) and + Tantivy's `ActiveTenantTracker` on the other, the same interval on + both, no asymmetry left between them; see §14 and + `/docs/security/threat-model.md`'s "Read this first". A newly-provisioned tenant's ClickHouse database and Tantivy index are both now real, isolated, and actually populated by write-routed traffic (the ClickHouse claim pending live confirmation, the Tantivy diff --git a/docs/security/threat-model.md b/docs/security/threat-model.md index 11b330a..1b89cb2 100644 --- a/docs/security/threat-model.md +++ b/docs/security/threat-model.md @@ -9,27 +9,28 @@ for the full design rationale behind the controls described here. ## Read this first: the single most important open finding -**Updated a fifth time.** This section originally read "log data queried +**Updated a sixth time.** This section originally read "log data queried through `POST /query` is not tenant-isolated at all," then "ClickHouse is isolated but Tantivy isn't," then "ingest tags records with a tenant identity but nothing routes the write," then "ClickHouse write-routing is built but Tantivy's isn't," then "both are write-routed but neither -rechecks tenant-active status live." Both storage engines are now -isolated on both the read and write paths, **and both now gate writes -on an active-tenant check** — ClickHouse's (`chwriter.Registry`) and -Tantivy's (`search/src/tenants.rs`'s `ActiveTenantTracker`, new) differ -in staleness bound, not in whether the gate exists at all: ClickHouse's -is a startup-time snapshot with no refresh (a *deprovisioned* tenant can -keep writing successfully until the next `enterprise-ingest` restart); -Tantivy's polls `enterprise-auth` every 60 seconds and keeps serving the -last-known-good set through a transient refresh failure, a materially -tighter staleness window with no code changed on the ClickHouse side to -match it — a real, disclosed asymmetry between the two, not a claim -they're now identical. What's left is narrower still: **whether a given -deployment actually runs the isolated binaries** (deployment-time, not -code-level), and closing ClickHouse's staleness gap to match Tantivy's -if that inconsistency matters for a given deployment. See below for -both engines' write-routing, in full. +rechecks tenant-active status live," then "both recheck, but ClickHouse's +snapshot never refreshes while Tantivy's does." Both storage engines are +now isolated on both the read and write paths, **both gate writes on an +active-tenant check, and both now refresh that check periodically** — +`chwriter.Registry.StartRefreshing` (new) closes the asymmetry the +previous version of this section named: ClickHouse's writer map now +re-lists active tenants every minute, the same interval +`tenants.ActiveTenantTracker` already used on the Tantivy side, opening +connections for newly-active tenants and closing/removing ones no +longer active — a deprovisioned tenant now loses ClickHouse write access +within a minute, not "until the next `enterprise-ingest` restart." +Neither engine does a live per-write check (a database/HTTP round trip +per record would be a real throughput cost neither implementation +accepts), so a minute-wide staleness window remains on both sides by +design, not by oversight. What's left is narrower still: **whether a +given deployment actually runs the isolated binaries** (deployment-time, +not code-level). See below for both engines' write-routing, in full. **ClickHouse (the SQL path) is built.** `enterprise/internal/ tenantprovision` (real `CREATE DATABASE`/`CREATE USER`/`GRANT`) and @@ -109,10 +110,15 @@ ProvisionClickHouse`'s grant was `SELECT`-only (correct for the read-side credential `chrunner` uses, but `chwriter` reuses the same credential for writes) — every real per-tenant write would have failed with a permission error until this was widened to `SELECT, INSERT`. -**Not yet confirmed against a real ClickHouse**, same caveat as the -read-side chrunner claim above — the Docker-free fail-closed tests pass, -the live-database tests are written but skip-gated, see -`/docs/phase-4-runbook.md`. +`Registry.StartRefreshing` (new) re-lists active tenants every minute +and reconciles the writer map — opens a connection for a newly-active +tenant, closes and removes one no longer active — closing what was +originally a startup-only snapshot with no refresh at all. **Not yet +confirmed against a real ClickHouse**, same caveat as the read-side +chrunner claim above — the Docker-free fail-closed and refresh-error +tests pass, the live-database tests (including the two new ones proving +refresh's add/remove reconciliation against real connections) are +written but skip-gated, see `/docs/phase-4-runbook.md`. **Tantivy**: `search/src/consumer.rs` now resolves each record's `tenant_id` header through the *same* `IndexRegistry` the read side @@ -136,22 +142,19 @@ header actually sent, JSON parsing, both fail-closed paths) against a hand-rolled TCP test server, no live enterprise-auth needed since `reqwest` doesn't care that the other end is real. -**Both engines share one open question** — deployment topology, covered -above — and now differ only in staleness bound, not in whether an -active-tenant gate exists at all: `chwriter.Registry`'s map is built -once at `enterprise-ingest` startup from an active-tenants-only query -and never refreshed, so a tenant deprovisioned after startup keeps -writing successfully until the next restart; `ActiveTenantTracker` -refreshes every 60 seconds, a materially tighter window, with no -corresponding change made to the ClickHouse side to match it. Neither -is a live per-write check (that would mean a database round trip on -every record, a real throughput cost neither implementation accepts), -so both have *some* staleness window by design — the asymmetry between -the two windows is the one thing disclosed as inconsistent, not fixed, -here. A newly-provisioned tenant's ClickHouse database and Tantivy index -are both now real, isolated, and actually populated by write-routed -agent traffic (the ClickHouse claim pending live confirmation, the -Tantivy claim already verified). +**Both engines share one open question now** — deployment topology, +covered above — and no longer differ in active-tenant staleness bound: +`chwriter.Registry.StartRefreshing` re-lists active tenants every +minute, matching `ActiveTenantTracker`'s interval, opening a connection +for a newly-active tenant and closing/removing one no longer active. +Neither is a live per-write check (that would mean a database/HTTP +round trip on every record, a real throughput cost neither +implementation accepts), so a roughly one-minute staleness window +remains on both sides by design, not a gap unique to either engine +anymore. A newly-provisioned tenant's ClickHouse database and Tantivy +index are both now real, isolated, and actually populated by +write-routed agent traffic (the ClickHouse claim pending live +confirmation, the Tantivy claim already verified). ## System overview @@ -509,8 +512,8 @@ terms: | Tantivy per-tenant index routing (`search/src/registry.rs`) | **Enforced, verified live** — real Tantivy indices, real cross-tenant probe, all passing | | Tantivy tenant_id resolution (`enterprise/internal/searchclient`) | **Enforced, verified live** — real gRPC wire-level test | | Ingest tenant *identity* (credential validation, tagging) | **Built and tested** — fail-closed `TenantResolver`, `tenant_id` Kafka header attached per record | -| Ingest tenant *write-routing*, ClickHouse | **Built, not yet confirmed against a real ClickHouse** — `enterprise-ingest`/`chwriter.Registry` route each tagged batch to its tenant's own database, fail-closed on an untagged/unprovisioned tenant; Docker-free tests pass, live-database tests are skip-gated. Startup-time active-tenant snapshot, no live recheck — a deprovisioned tenant can keep writing until the next restart | -| Ingest tenant *write-routing*, Tantivy | **Built and genuinely verified** — `search/src/consumer.rs` routes each record into its own tenant's index via `IndexRegistry`, same registry the (already-verified) read side uses; no Docker needed, real tests pass. Active-tenant-gated too: `tenants::ActiveTenantTracker` polls `enterprise-auth` every 60s (off unless configured), refusing any tenant not in the polled allowlist — tighter staleness bound than ClickHouse's startup-only snapshot, an asymmetry disclosed above, not a gap on Tantivy's side specifically | +| Ingest tenant *write-routing*, ClickHouse | **Built, not yet confirmed against a real ClickHouse** — `enterprise-ingest`/`chwriter.Registry` route each tagged batch to its tenant's own database, fail-closed on an untagged/unprovisioned tenant; Docker-free tests pass, live-database tests are skip-gated. Active-tenant snapshot now refreshes every minute (`Registry.StartRefreshing`) — a deprovisioned tenant loses write access within a minute, not "until the next restart" | +| Ingest tenant *write-routing*, Tantivy | **Built and genuinely verified** — `search/src/consumer.rs` routes each record into its own tenant's index via `IndexRegistry`, same registry the (already-verified) read side uses; no Docker needed, real tests pass. Active-tenant-gated too: `tenants::ActiveTenantTracker` polls `enterprise-auth` every 60s (off unless configured), refusing any tenant not in the polled allowlist — same one-minute staleness bound as ClickHouse's now-refreshing snapshot, no more asymmetry between the two | | Deployment actually routing traffic to `enterprise-api` (Helm) | **Enforced** — `api`/`enterprise-api` are mutually exclusive, same flag as RBAC/audit/SSO | | Deployment actually routing traffic to `enterprise-api` (docker-compose) | **Enforced** — `api`/`enterprise-api` are mutually exclusive via `COMPOSE_PROFILES`, same flag choice as Helm's `enterprise.enabled`; verified via `docker compose config`, not an actual `docker compose up` in this environment | | Human SSO login — OIDC | **Built, verified with a real fake IdP** (not yet tried against a real external IdP) | diff --git a/enterprise/README.md b/enterprise/README.md index 15da22a..53b2243 100644 --- a/enterprise/README.md +++ b/enterprise/README.md @@ -249,6 +249,21 @@ visibility reasoning as every other package this phase moved out of `internal/` for a cross-module import (see `ingest/README.md`'s "Multi- tenant write-routing" section). +The writer map isn't frozen at startup anymore: `Registry. +StartRefreshing` (called from `cmd/enterprise-ingest/main.go` right +after construction) spawns a goroutine that re-lists data sources every +minute and reconciles the map -- opens a connection for a newly-active +tenant, closes and removes one no longer active. This closes the same +staleness gap `search/src/tenants.rs`'s `ActiveTenantTracker` closes on +the Tantivy side (see `/search/README.md`'s "Per-tenant indices" +section), at the same one-minute interval, so a deprovisioned tenant +loses write access on both storage engines within roughly the same +window instead of ClickHouse's writer surviving until the next +`enterprise-ingest` restart. A refresh failure (rbacstore unreachable, +or one tenant's new connection failing to open) logs and leaves the +existing map untouched for that tick, the same last-known-good posture +the Rust tracker uses. + A real bug was found and fixed while wiring this up: `tenantprovision.ProvisionClickHouse` originally granted a tenant's ClickHouse user `SELECT` only -- correct for `chrunner`'s query path, @@ -271,15 +286,17 @@ exclusivity isn't wired in compose, a disclosed local-dev-only gap; see that service's own comment). Verified: `enterprise/internal/chwriter`'s fail-closed paths (empty/ -unknown `tenant_id`) run genuinely without Docker (constructing a -`Registry` directly, bypassing `New`, which is the only part that would -dial ClickHouse); the actual per-tenant write-isolation probe -(`TestRegistryWritesEachTenantToItsOwnDatabase`) and the -`tenantprovision` INSERT-grant regression test are real integration -tests against a live ClickHouse, same `CHWRITER_TEST_CLICKHOUSE_ADDR`/ -`TENANTPROVISION_TEST_CLICKHOUSE_ADDR` convention as every other -ClickHouse-backed test this phase -- not run against a live database in -this environment. +unknown `tenant_id`, and now a failed refresh keeping the last-known-good +map) run genuinely without Docker (constructing a `Registry` directly, +bypassing `New`, which is the only part that would dial ClickHouse); the +actual per-tenant write-isolation probe +(`TestRegistryWritesEachTenantToItsOwnDatabase`), the `tenantprovision` +INSERT-grant regression test, and refresh's add/remove reconciliation +(`TestRefreshAddsNewlyActiveTenant`/`TestRefreshRemovesNoLongerActiveTenant`) +are real integration tests against a live ClickHouse, same +`CHWRITER_TEST_CLICKHOUSE_ADDR`/`TENANTPROVISION_TEST_CLICKHOUSE_ADDR` +convention as every other ClickHouse-backed test this phase -- not run +against a live database in this environment. ## Package layout diff --git a/enterprise/cmd/enterprise-ingest/main.go b/enterprise/cmd/enterprise-ingest/main.go index 0e9ef8d..ec8d4d9 100644 --- a/enterprise/cmd/enterprise-ingest/main.go +++ b/enterprise/cmd/enterprise-ingest/main.go @@ -45,6 +45,12 @@ import ( "github.com/sentry/sentry/ingest/consumer" ) +// dataSourceRefreshInterval matches search/src/tenants.rs's +// REFRESH_INTERVAL -- both close the same disclosed asymmetry +// (/docs/security/threat-model.md's "Read this first") on their +// respective storage engines, so they use the same staleness bound. +const dataSourceRefreshInterval = time.Minute + func main() { logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) @@ -80,22 +86,14 @@ func main() { // that's mid-provisioning simply has no writer in the registry // below, so chwriter.Registry.WriteBatch refuses it the same way // chrunner.Registry.RunSQL already refuses an unprovisioned tenant - // on the read side. - sources, err := rbac.ListProvisionedDataSources(ctx) + // on the read side. lister is reused below for periodic refresh, + // not just this one startup call. + lister := tenantDataSourceLister(rbac) + chwSources, err := lister(ctx) if err != nil { logger.Error("listing provisioned data sources", "error", err) os.Exit(1) } - chwSources := make([]chwriter.DataSource, 0, len(sources)) - for _, s := range sources { - if s.ClickHouseUsername == nil || s.ClickHousePassword == nil { - continue // ListProvisionedDataSources already filters these out; defensive only. - } - chwSources = append(chwSources, chwriter.DataSource{ - TenantID: s.TenantID, Database: s.ClickHouseDatabaseName, - Username: *s.ClickHouseUsername, Password: *s.ClickHousePassword, - }) - } logger.Info("loaded tenant data sources", "count", len(chwSources)) registry, err := chwriter.New(ctx, cfg.ClickHouseAddr, chwSources) @@ -115,6 +113,13 @@ func main() { srv := &http.Server{Addr: cfg.HTTPListenAddr, Handler: mux} g, ctx := errgroup.WithContext(ctx) + // Closes the staleness gap disclosed in + // /docs/security/threat-model.md as an asymmetry with search's + // tenants.ActiveTenantTracker: the writer map built above was a + // startup-only snapshot until this call -- now it re-lists and + // reconciles every dataSourceRefreshInterval, stopping when ctx is + // cancelled (same shutdown path c.Run below uses). + registry.StartRefreshing(ctx, lister, dataSourceRefreshInterval, logger) g.Go(func() error { return c.Run(ctx) }) g.Go(func() error { logger.Info("enterprise-ingest healthz listening", "addr", cfg.HTTPListenAddr) @@ -137,6 +142,31 @@ func main() { } } +// tenantDataSourceLister adapts rbacstore's row shape into +// []chwriter.DataSource -- shared between the initial synchronous load +// above (must succeed before this binary does anything) and +// chwriter.Registry.StartRefreshing's periodic re-list, so the two +// never drift into checking different things. +func tenantDataSourceLister(rbac *rbacstore.Store) chwriter.SourceLister { + return func(ctx context.Context) ([]chwriter.DataSource, error) { + sources, err := rbac.ListProvisionedDataSources(ctx) + if err != nil { + return nil, err + } + chwSources := make([]chwriter.DataSource, 0, len(sources)) + for _, s := range sources { + if s.ClickHouseUsername == nil || s.ClickHousePassword == nil { + continue // ListProvisionedDataSources already filters these out; defensive only. + } + chwSources = append(chwSources, chwriter.DataSource{ + TenantID: s.TenantID, Database: s.ClickHouseDatabaseName, + Username: *s.ClickHouseUsername, Password: *s.ClickHousePassword, + }) + } + return chwSources, nil + } +} + // runHealthcheck mirrors every other binary in this repo's // -healthcheck self-check mode -- execs the binary against itself // rather than using an external tool (see e.g. api/cmd/api/main.go's diff --git a/enterprise/internal/chwriter/chwriter.go b/enterprise/internal/chwriter/chwriter.go index b3f7118..5779c1e 100644 --- a/enterprise/internal/chwriter/chwriter.go +++ b/enterprise/internal/chwriter/chwriter.go @@ -24,6 +24,9 @@ package chwriter import ( "context" "fmt" + "log/slog" + "sync" + "time" "github.com/sentry/sentry/ingest/clickhousewriter" "github.com/sentry/sentry/ingest/consumer" @@ -44,11 +47,17 @@ type DataSource struct { // Registry implements ingest/consumer's chWriter interface // (WriteBatch(ctx, []consumer.Record) error) by routing each record to -// its tenant's dedicated connection. Immutable after New returns -- see -// this file's doc comment. +// its tenant's dedicated connection. The writer map used to be +// immutable after New returned; StartRefreshing (below) makes it +// mutable at runtime, guarded by mu -- WriteBatch takes a read lock (the +// common case, and concurrent reads don't block each other), a refresh +// takes a write lock only for the brief final swap, never while +// actually dialing ClickHouse (see refresh's comment). type Registry struct { + addr string + + mu sync.RWMutex writers map[string]*clickhousewriter.Writer - closers []func() } // New opens one real ClickHouse connection per DataSource (same native @@ -58,7 +67,7 @@ type Registry struct { // Registry fails to construct rather than silently running with a // partial tenant set. func New(ctx context.Context, addr string, sources []DataSource) (*Registry, error) { - reg := &Registry{writers: make(map[string]*clickhousewriter.Writer, len(sources))} + reg := &Registry{addr: addr, writers: make(map[string]*clickhousewriter.Writer, len(sources))} for _, src := range sources { w, err := clickhousewriter.New(ctx, clickhousewriter.Config{ Addr: addr, Database: src.Database, Username: src.Username, Password: src.Password, @@ -68,16 +77,21 @@ func New(ctx context.Context, addr string, sources []DataSource) (*Registry, err return nil, fmt.Errorf("chwriter: opening connection for tenant %q: %w", src.TenantID, err) } reg.writers[src.TenantID] = w - reg.closers = append(reg.closers, func() { _ = w.Close() }) } return reg, nil } // Close releases every underlying connection -- call once at process -// shutdown, same lifecycle as chrunner.Registry.Close. +// shutdown, same lifecycle as chrunner.Registry.Close. Safe to call +// even with StartRefreshing's goroutine still running (it only ever +// adds/removes individual writers under mu, never assumes the whole map +// survives), though callers should still cancel that goroutine's +// context first to stop it from reopening what Close just shut down. func (r *Registry) Close() { - for _, c := range r.closers { - c() + r.mu.Lock() + defer r.mu.Unlock() + for _, w := range r.writers { + _ = w.Close() } } @@ -105,6 +119,8 @@ func (r *Registry) WriteBatch(ctx context.Context, records []consumer.Record) er byTenant[rec.TenantID] = append(byTenant[rec.TenantID], rec) } + r.mu.RLock() + defer r.mu.RUnlock() for tenantID, group := range byTenant { if tenantID == "" { return fmt.Errorf("chwriter: %d record(s) in this batch have no tenant_id, refusing to write any of it", len(group)) @@ -123,3 +139,91 @@ func (r *Registry) WriteBatch(ctx context.Context, records []consumer.Record) er } return nil } + +// SourceLister re-lists the data sources a Registry should have a +// writer for -- a narrow function type, not an rbacstore dependency, +// same reasoning DataSource's doc comment gives for not importing +// rbacstore directly here. enterprise-ingest's main.go supplies one +// backed by rbacstore.ListProvisionedDataSources (the same query New's +// caller already runs once at startup). +type SourceLister func(ctx context.Context) ([]DataSource, error) + +// StartRefreshing closes the staleness gap disclosed in +// /docs/security/threat-model.md as an asymmetry with search's +// tenants.ActiveTenantTracker (Tantivy's write-side active-tenant gate, +// which already refreshes every 60s): spawns a goroutine that +// periodically re-lists data sources via lister and reconciles the +// writer map -- opens a connection for any newly-active tenant, closes +// and removes any tenant no longer present (deprovisioned or suspended +// since the last refresh). Stops when ctx is cancelled; call at most +// once per Registry. A refresh failure (lister error, or one tenant's +// new connection failing to open) logs via logger and leaves the +// existing map alone for that tick -- a transient rbacstore/Postgres +// blip, or one bad tenant's connection, must not evict every other +// tenant's already-working writer, the same "last-known-good" posture +// ActiveTenantTracker's periodic refresh uses. +func (r *Registry) StartRefreshing(ctx context.Context, lister SourceLister, interval time.Duration, logger *slog.Logger) { + go func() { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + r.refresh(ctx, lister, logger) + } + } + }() +} + +// refresh dials any newly-needed connections *before* taking the write +// lock, so a slow/unreachable ClickHouse for one newly-active tenant +// never blocks WriteBatch's read lock for longer than the map swap +// itself takes. +func (r *Registry) refresh(ctx context.Context, lister SourceLister, logger *slog.Logger) { + sources, err := lister(ctx) + if err != nil { + logger.Error("chwriter: refreshing data sources failed, keeping last-known-good writer set", "error", err) + return + } + fresh := make(map[string]DataSource, len(sources)) + for _, src := range sources { + fresh[src.TenantID] = src + } + + r.mu.RLock() + var toOpen []DataSource + for tenantID, src := range fresh { + if _, ok := r.writers[tenantID]; !ok { + toOpen = append(toOpen, src) + } + } + r.mu.RUnlock() + + newWriters := make(map[string]*clickhousewriter.Writer, len(toOpen)) + for _, src := range toOpen { + w, err := clickhousewriter.New(ctx, clickhousewriter.Config{ + Addr: r.addr, Database: src.Database, Username: src.Username, Password: src.Password, + }) + if err != nil { + logger.Error("chwriter: opening connection for newly-active tenant failed, will retry next refresh", "tenant_id", src.TenantID, "error", err) + continue + } + newWriters[src.TenantID] = w + } + + r.mu.Lock() + defer r.mu.Unlock() + for tenantID, w := range newWriters { + r.writers[tenantID] = w + logger.Info("chwriter: added writer for newly-active tenant", "tenant_id", tenantID) + } + for tenantID, w := range r.writers { + if _, ok := fresh[tenantID]; !ok { + _ = w.Close() + delete(r.writers, tenantID) + logger.Info("chwriter: removed writer for tenant no longer active/provisioned", "tenant_id", tenantID) + } + } +} diff --git a/enterprise/internal/chwriter/chwriter_test.go b/enterprise/internal/chwriter/chwriter_test.go index fe9067a..1976ddd 100644 --- a/enterprise/internal/chwriter/chwriter_test.go +++ b/enterprise/internal/chwriter/chwriter_test.go @@ -14,7 +14,10 @@ package chwriter import ( "context" + "errors" "fmt" + "io" + "log/slog" "os" "testing" @@ -27,6 +30,10 @@ import ( logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + // TestWriteBatchRefusesEmptyTenantID and // TestWriteBatchRefusesUnknownTenantWithEmptyRegistry construct a // Registry directly (bypassing New, which would dial ClickHouse) so @@ -52,6 +59,31 @@ func TestWriteBatchRefusesUnknownTenantWithEmptyRegistry(t *testing.T) { } } +// TestRefreshListerErrorLeavesRegistryUnchanged is Docker-free the same +// way the two tests above are: refresh's early-return on a lister error +// happens before anything touches ClickHouse, so this genuinely +// exercises the "keep last-known-good" path -- see refresh's doc +// comment on StartRefreshing. +func TestRefreshListerErrorLeavesRegistryUnchanged(t *testing.T) { + reg := &Registry{writers: map[string]*clickhousewriter.Writer{}} + lister := func(context.Context) ([]DataSource, error) { + return nil, errors.New("rbacstore unreachable") + } + + reg.refresh(context.Background(), lister, discardLogger()) + + // Still refuses -- refresh must not have added a writer for "acme" + // (there's nothing a failed lister call could have legitimately + // learned), and must not have panicked reaching into a nil/partial + // state either. + err := reg.WriteBatch(context.Background(), []consumer.Record{ + {TenantID: "acme", Record: &logsv1.LogRecord{Message: "m"}}, + }) + if err == nil { + t.Fatal("expected WriteBatch to still refuse tenant acme after a failed refresh") + } +} + func testAddr(t *testing.T) string { t.Helper() addr := os.Getenv("CHWRITER_TEST_CLICKHOUSE_ADDR") @@ -155,3 +187,74 @@ func TestRegistryRefusesUnprovisionedTenant(t *testing.T) { t.Fatal("expected WriteBatch to refuse a tenant with no provisioned connection, not silently drop or misroute it") } } + +// TestRefreshAddsNewlyActiveTenant is the live counterpart to +// TestRefreshListerErrorLeavesRegistryUnchanged: proves refresh actually +// opens a real, usable connection for a tenant that appears in a later +// lister call but wasn't present at New() time -- the scenario +// StartRefreshing exists to handle (a tenant provisioned after +// enterprise-ingest already started). +func TestRefreshAddsNewlyActiveTenant(t *testing.T) { + addr := testAddr(t) + ctx := context.Background() + tenantA, credsA := provisionTestTenant(t, addr) + + reg, err := New(ctx, addr, nil) // starts with zero tenants, same as a cold start before any tenant exists + if err != nil { + t.Fatalf("New: %v", err) + } + defer reg.Close() + + if err := reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: tenantA, Record: &logsv1.LogRecord{Message: "m", RecordId: uuid.NewString()}}, + }); err == nil { + t.Fatal("expected WriteBatch to refuse tenantA before the first refresh has run") + } + + lister := func(context.Context) ([]DataSource, error) { + return []DataSource{{TenantID: tenantA, Database: tenantA, Username: credsA.Username, Password: credsA.Password}}, nil + } + reg.refresh(ctx, lister, discardLogger()) + + if err := reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: tenantA, Record: &logsv1.LogRecord{Host: "h1", Message: "after-refresh", RecordId: uuid.NewString()}}, + }); err != nil { + t.Fatalf("expected WriteBatch to succeed for tenantA after refresh added it, got: %v", err) + } +} + +// TestRefreshRemovesNoLongerActiveTenant is TestRefreshAddsNewlyActiveTenant's +// mirror image: a tenant present at New() time that a later lister call +// no longer returns (deprovisioned or suspended) must lose its writer, +// not keep writing indefinitely until process restart -- the exact +// staleness gap this whole mechanism exists to close. +func TestRefreshRemovesNoLongerActiveTenant(t *testing.T) { + addr := testAddr(t) + ctx := context.Background() + tenantA, credsA := provisionTestTenant(t, addr) + + reg, err := New(ctx, addr, []DataSource{ + {TenantID: tenantA, Database: tenantA, Username: credsA.Username, Password: credsA.Password}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + defer reg.Close() + + if err := reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: tenantA, Record: &logsv1.LogRecord{Host: "h1", Message: "before-removal", RecordId: uuid.NewString()}}, + }); err != nil { + t.Fatalf("expected WriteBatch to succeed for tenantA before refresh removes it, got: %v", err) + } + + lister := func(context.Context) ([]DataSource, error) { + return nil, nil // tenantA no longer active/provisioned as of this refresh + } + reg.refresh(ctx, lister, discardLogger()) + + if err := reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: tenantA, Record: &logsv1.LogRecord{Message: "after-removal", RecordId: uuid.NewString()}}, + }); err == nil { + t.Fatal("expected WriteBatch to refuse tenantA after refresh removed it, not keep writing with a stale connection") + } +}