diff --git a/CLAUDE.md b/CLAUDE.md index 3585e5c..da71abf 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -237,21 +237,30 @@ presents (minted via `enterprise-auth new `POST /internal/authorize-ingest` endpoint — never an `enterprise/` import, same boundary shape as `api/authz.Authorizer`), and the resolved tenant ID is attached to every record as a `tenant_id` Kafka -message header before it's produced. **What's still deferred, clearly**: -nothing downstream reads that header yet — neither `ingest`'s own -ClickHouse writer nor `search`'s independent Redpanda consumer route a -record's write into a per-tenant destination, so every record still -lands in the one shared ClickHouse database/Tantivy index regardless of -which tenant it's now correctly tagged with. That write-routing split -(likely another "second binary," mirroring `enterprise-api`) is real, -scoped, remaining work — attaching a verified tenant identity as early -as possible was deliberately built as a self-contained first step, not -the whole feature. What still keeps this phase from being done: the -actual tenant-picker *page* doesn't exist (`web` has no session/cookie- -handling code at all yet, and `enterprise-auth` has no CORS middleware -for a cross-origin `fetch` with credentials — both real, separately- -scoped frontend gaps), and per-tenant write-routing for ingest per the -above. Full accounting: +message header before it's produced. **ClickHouse write-routing is now +built too**: `enterprise/cmd/enterprise-ingest` (another "second binary," +mirroring `enterprise-api`) reuses `ingest/consumer`'s own flush loop +with `enterprise/internal/chwriter.Registry` swapped in as the writer — +one dedicated ClickHouse connection per tenant, routing each batch's +records by their `tenant_id` tag, fail-closed on an untagged or +unprovisioned tenant. Building it found and fixed a real bug: +`tenantprovision.ProvisionClickHouse` originally granted a tenant's +ClickHouse user `SELECT` only, which would have made every real +per-tenant write fail with a permission error — fixed by granting +`SELECT, INSERT` (one credential, both directions; no cross-tenant +boundary is crossed by also allowing INSERT within a tenant's own +database). **What's still deferred, clearly**: Tantivy's side. +`search/src/consumer.rs` is a completely independent Redpanda consumer +(not called through `ingest` or `enterprise-ingest` at all) and still +writes every record into the one shared (default) index regardless of +tenant — a different codebase (Rust) and a different process, real, +disclosed, separately-scoped remaining work, not just "the same fix +applied twice." What still keeps this phase from being done: the actual +tenant-picker *page* doesn't exist (`web` has no session/cookie-handling +code at all yet, and `enterprise-auth` has no CORS middleware for a +cross-origin `fetch` with credentials — both real, separately-scoped +frontend gaps), and Tantivy write-routing per the above. Full +accounting: `/docs/security/threat-model.md`; step-by-step verification procedure (not yet run against a live cluster in this environment): `/docs/phase-4-runbook.md`. The rest of this section describes the exit diff --git a/deploy/helm/sentry/templates/enterprise-ingest.yaml b/deploy/helm/sentry/templates/enterprise-ingest.yaml new file mode 100644 index 0000000..b668188 --- /dev/null +++ b/deploy/helm/sentry/templates/enterprise-ingest.yaml @@ -0,0 +1,64 @@ +{{/* +Only rendered once per-tenant write-routing is actually turned on (see +ingest.yaml's -mode=server comment) -- this Deployment is what takes +over consuming sentry.logs.raw once ingest.yaml's own consumer half +stops, routing each record to its own tenant's dedicated ClickHouse +database (enterprise/internal/chwriter) instead of the one shared table. +No Service: this is a pure background worker, nothing calls it, only +kubelet's own probes talk to its /healthz. +*/}} +{{- if and .Values.enterprise.enabled .Values.ingest.requireTenantCredential }} +apiVersion: apps/v1 +kind: Deployment +metadata: + name: {{ .Release.Name }}-enterprise-ingest + labels: + {{- include "sentry.labels" . | nindent 4 }} + {{- include "sentry.selectorLabels" (list $ "enterprise-ingest") | nindent 4 }} +spec: + replicas: {{ .Values.ingest.replicas }} + selector: + matchLabels: + {{- include "sentry.selectorLabels" (list $ "enterprise-ingest") | nindent 6 }} + template: + metadata: + labels: + {{- include "sentry.selectorLabels" (list $ "enterprise-ingest") | nindent 8 }} + spec: + initContainers: + {{- include "sentry.waitForTCP" (list "redpanda" (printf "%s-redpanda" .Release.Name) "9092") | nindent 8 }} + {{- include "sentry.waitForTCP" (list "clickhouse" (printf "%s-clickhouse" .Release.Name) "9000") | nindent 8 }} + {{- include "sentry.waitForTCP" (list "postgres" (printf "%s-postgres" .Release.Name) "5432") | nindent 8 }} + containers: + - name: enterprise-ingest + image: "{{ .Values.enterprise.ingestImage.repository }}:{{ .Values.enterprise.ingestImage.tag }}" + imagePullPolicy: {{ .Values.global.imagePullPolicy }} + env: + - name: REDPANDA_BROKERS + value: "{{ .Release.Name }}-redpanda:9092" + - name: CLICKHOUSE_ADDR + value: "{{ .Release.Name }}-clickhouse:9000" + - name: POSTGRES_ADDR + value: "{{ .Release.Name }}-postgres:5432" + - name: POSTGRES_DATABASE + value: sentry_metadata + - name: POSTGRES_USERNAME + value: sentry + - name: POSTGRES_PASSWORD + valueFrom: + secretKeyRef: + name: {{ .Release.Name }}-postgres + key: password + readinessProbe: + exec: + command: ["/enterprise-ingest", "-healthcheck"] + initialDelaySeconds: 5 + periodSeconds: 5 + livenessProbe: + exec: + command: ["/enterprise-ingest", "-healthcheck"] + initialDelaySeconds: 10 + periodSeconds: 10 + resources: + {{- toYaml .Values.ingest.resources | nindent 12 }} +{{- end }} diff --git a/deploy/helm/sentry/templates/ingest.yaml b/deploy/helm/sentry/templates/ingest.yaml index a567683..a0889e0 100644 --- a/deploy/helm/sentry/templates/ingest.yaml +++ b/deploy/helm/sentry/templates/ingest.yaml @@ -22,6 +22,18 @@ spec: - name: ingest image: "{{ .Values.ingest.image.repository }}:{{ .Values.ingest.image.tag }}" imagePullPolicy: {{ .Values.global.imagePullPolicy }} + {{- if and .Values.enterprise.enabled .Values.ingest.requireTenantCredential }} + # -mode=server only: this Deployment stops running the + # ClickHouse-writing consumer half (the default -mode=all) + # once per-tenant write-routing is on -- enterprise-ingest.yaml + # (below) takes over consuming sentry.logs.raw instead, so it + # can write each tenant's records to their own database rather + # than the one shared table ingest's own consumer always + # writes to. The agent-facing server half (PushBatch, tenant + # tagging via TenantResolver) keeps running here unconditionally + # either way -- only which process consumes the topic changes. + args: ["-mode=server"] + {{- end }} env: - name: REDPANDA_BROKERS value: "{{ .Release.Name }}-redpanda:9092" diff --git a/deploy/helm/sentry/values.yaml b/deploy/helm/sentry/values.yaml index dbf5146..b867464 100644 --- a/deploy/helm/sentry/values.yaml +++ b/deploy/helm/sentry/values.yaml @@ -76,7 +76,13 @@ ingest: # its own deliberate opt-in, not folded into enterprise.enabled # directly: turning it on requires every agent to already present a # valid ingest credential (`enterprise-auth - # -create-ingest-credential-tenant=`) or be refused outright. + # -create-ingest-credential-tenant=`) or be refused outright. Also + # controls per-tenant write-routing: true switches ingest.yaml's + # Deployment to -mode=server only and renders enterprise-ingest.yaml + # to take over consuming sentry.logs.raw, writing each tenant's + # records into their own ClickHouse database instead of the one + # shared table -- both flags gate together since write-routing is only + # meaningful once records actually carry a tenant_id to route on. requireTenantCredential: false search: @@ -156,6 +162,15 @@ enterprise: apiImage: repository: sentry-enterprise-api tag: latest + # enterprise-ingest (templates/enterprise-ingest.yaml) -- only + # rendered when ingest.requireTenantCredential is also true (see that + # value's comment); takes over consuming sentry.logs.raw from + # ingest.yaml's own consumer once per-tenant write-routing is on. Same + # "repo root build context" reasoning as apiImage above -- see + # enterprise/cmd/enterprise-ingest/Dockerfile. + ingestImage: + repository: sentry-enterprise-ingest + tag: latest replicas: 1 resources: {} # Leave empty to auto-generate (>= 32 bytes) and persist across diff --git a/docker-compose.yml b/docker-compose.yml index 828deb1..059af88 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -365,6 +365,46 @@ services: timeout: 5s retries: 30 + # Per-tenant write-routing for ingest (enterprise/internal/chwriter) -- + # see enterprise/cmd/enterprise-ingest/main.go's doc comment. Unlike + # api/enterprise-api's COMPOSE_PROFILES trick, this is NOT wired to be + # mutually exclusive with `ingest`'s own consumer half in this file: + # `ingest` always runs -mode=all here regardless of profile (splitting + # its server/consumer halves into separate compose services isn't done + # -- real, disclosed scope, see /docs/phase-4-runbook.md), so with the + # "enterprise" profile active both this service AND ingest's own + # consumer independently read every message (different consumer + # groups) and write it -- ingest into the one shared `logs` table, + # this into each tenant's own database. Harmless duplication for local + # testing/verification purposes, not what a real deployment does (see + # deploy/helm/sentry's ingest.yaml/enterprise-ingest.yaml, which + # actually achieve exclusivity via -mode=server/-mode=consumer). + enterprise-ingest: + profiles: ["enterprise"] + build: + context: . + dockerfile: enterprise/cmd/enterprise-ingest/Dockerfile + container_name: sentry-enterprise-ingest + depends_on: + redpanda-provision: + condition: service_completed_successfully + clickhouse-migrate: + condition: service_completed_successfully + metadata-migrate: + condition: service_completed_successfully + environment: + REDPANDA_BROKERS: "redpanda:9092" + CLICKHOUSE_ADDR: "clickhouse:9000" + POSTGRES_ADDR: "metadata-postgres:5432" + POSTGRES_DATABASE: "sentry_metadata" + POSTGRES_USERNAME: "sentry" + POSTGRES_PASSWORD: "sentry-dev-only" + healthcheck: + test: ["CMD", "/enterprise-ingest", "-healthcheck"] + interval: 5s + timeout: 5s + retries: 30 + web: build: context: web diff --git a/docs/architecture.md b/docs/architecture.md index c61d583..96f3cd1 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -139,23 +139,34 @@ escape hatch is opaque to any compiler-injected filter. ran in the environment it was built in — Tantivy is an embedded library, so the cross-tenant isolation probe needed no live database or Docker to execute for real, and it passed. -- **Ingest identity is now built, though write-routing isn't.** `ingest` - (AGPL core) gained an optional `TenantResolver` +- **Ingest identity and ClickHouse write-routing are now built; Tantivy's + isn't.** `ingest` (AGPL core) gained an optional `TenantResolver` (`ingest/internal/grpcserver`): an agent presents a per-tenant bearer credential (`enterprise-auth -create-ingest-credential-tenant=` mints one, only its hash stored), validated over the network via a new `POST /internal/authorize-ingest` endpoint (never an `enterprise/` import — same "network boundary, not import boundary" shape `api/authz.Authorizer` already uses), and the resolved tenant ID rides - as a `tenant_id` Kafka message header on every record produced. What - isn't built yet: neither `ingest`'s own ClickHouse writer nor - `search`'s independent Redpanda consumer reads that header back to - route the write anywhere per-tenant — every record still lands in the - one shared ClickHouse database and Tantivy index regardless of tenant, - correctly tagged but not yet isolated at write time. That per-tenant - write-routing split is real, scoped remaining work (likely another - "second binary," mirroring `enterprise-api`), not something this - change claims to have closed. + as a `tenant_id` Kafka message header on every record produced. + `enterprise/cmd/enterprise-ingest` (another "second binary," mirroring + `enterprise-api`) consumes that tag: it reuses `ingest/consumer`'s own + flush loop with `enterprise/internal/chwriter.Registry` swapped in as + the writer, routing each batch's records to a per-tenant ClickHouse + connection (one per tenant, built from `rbacstore. + ListProvisionedDataSources` — the same source `chrunner` already uses + for reads), fail-closed on an untagged or unprovisioned tenant. + Building it found and fixed a real bug: `tenantprovision. + ProvisionClickHouse` originally granted a tenant's ClickHouse user + `SELECT` only, which would have made every real per-tenant write fail + with a permission error — fixed by granting `SELECT, INSERT` (one + credential, both directions; no cross-tenant boundary is crossed by + also allowing INSERT within a tenant's own database). What isn't built + yet: `search`'s independent Redpanda consumer (a different codebase — + Rust — and a different process, not reachable through either `ingest` + or `enterprise-ingest`) still writes every record into the one shared + Tantivy index regardless of tenant. That's real, disclosed, separately + scoped remaining work, not something this change claims to have + closed. - `deploy/operator`'s `Tenant` CRD and `enterprise-api -provision-tenant` are now unified, deliberately lightweight: `-provision-tenant` stays the sole real actor (ClickHouse + `rbacstore`), and now also syncs its diff --git a/docs/phase-4-runbook.md b/docs/phase-4-runbook.md index 9a62f8b..b5837bb 100644 --- a/docs/phase-4-runbook.md +++ b/docs/phase-4-runbook.md @@ -554,7 +554,7 @@ all -- a cross-origin `fetch` with credentials from `web`'s origin to actual picker UI is real, separately-scoped frontend work; this section only closes the backend half. -## 13. Ingest tenant identity (no per-tenant write-routing yet) +## 13. Ingest tenant identity The identity mechanism was chosen deliberately (config-supplied tenant_id + a shared-secret token ingest validates, not per-tenant @@ -592,24 +592,77 @@ go test ./internal/grpcserver/... -run 'Resolver|TenantHeader' -v # tenant." ``` -**Not built, and explicitly scoped out for now**: per-tenant write -routing. Neither `ingest/internal/consumer` (the ClickHouse writer) nor -`search/src/consumer.rs` (a completely independent Redpanda consumer, -not called through `ingest` at all -- see that file) reads the -`tenant_id` Kafka header back to route a record's write into a -per-tenant ClickHouse database or Tantivy index. Every record still -lands in the one shared destination regardless of tenant, correctly -tagged but not yet isolated at write time -- see CLAUDE.md and -`/docs/security/threat-model.md`'s "Read this first" for the full -disclosure. Also not built: any Helm/`docker-compose.yml` wiring that -issues an agent a real ingest credential automatically (`enterprise- -auth -create-ingest-credential-tenant=` is, like every other -credential-minting flag in this codebase, a manual operator action) -- -`deploy/helm/sentry/values.yaml`'s `ingest.requireTenantCredential` -(default `false`) only turns on *validation*, deliberately not folded -into `enterprise.enabled` directly, since flipping that flag with no -agents holding a credential yet would refuse all ingest traffic outright -rather than degrading gracefully. +Also not built as part of the identity mechanism itself: any Helm/ +`docker-compose.yml` wiring that issues an agent a real ingest credential +automatically (`enterprise-auth -create-ingest-credential-tenant=` +is, like every other credential-minting flag in this codebase, a manual +operator action) -- `deploy/helm/sentry/values.yaml`'s +`ingest.requireTenantCredential` (default `false`) only turns on +*validation*, deliberately not folded into `enterprise.enabled` +directly, since flipping that flag with no agents holding a credential +yet would refuse all ingest traffic outright rather than degrading +gracefully. See §14 below for what the identity is actually used for on +the write path. + +## 14. Ingest write-routing (ClickHouse built, Tantivy not yet) + +`enterprise/cmd/enterprise-ingest` (mirrors `enterprise-api`'s "second +binary" shape) consumes the `tenant_id` Kafka header §13 attaches and +routes each record's ClickHouse write into that tenant's own database, +via `enterprise/internal/chwriter.Registry` -- a per-tenant +`*ingest/clickhousewriter.Writer` registry that reuses `ingest/ +consumer`'s own flush loop unchanged. Verified in this environment +without Docker, using the same fake-transport discipline as §13: + +```sh +cd enterprise +go test ./internal/chwriter/... -run TestRegistry -v +# Docker-free: constructs a Registry directly (bypassing New(), the only +# part that dials ClickHouse) to prove the fail-closed paths -- an empty +# TenantID, or a TenantID with no registered writer, refuses the WHOLE +# batch rather than silently dropping just those records or falling back +# to a default destination. + +go test ./internal/tenantprovision/... -run TestProvisionedUserCanInsertIntoOwnDatabase -v +# skip-gated (needs a live ClickHouse) -- regression test for the bug +# found while building this: the per-tenant credential chwriter reuses +# from chrunner only had SELECT granted, which would make every real +# write fail with a permission error. Fixed by granting SELECT, INSERT +# (tenantprovision.go). This test proves the fix, not just documents it, +# whenever a real ClickHouse is available to run it against. + +go test ./internal/chwriter/... -run TestRegistryRoutesToCorrectTenant -v +# skip-gated (CHWRITER_TEST_CLICKHOUSE_ADDR) -- writes a mixed batch +# spanning two tenants in one WriteBatch call and confirms each row +# lands in its own tenant's database, none in the other's. +``` + +**Not run in this environment**: no live ClickHouse was available while +this was built, so the skip-gated tests above are correct Go that has +never actually executed -- "the test exists" is not the same claim as +"write-routing is confirmed," same caveat §8 already states for the +read-side `chrunner` tests. + +**Not built**: Tantivy's side of this. `search/src/consumer.rs` is a +completely independent Redpanda consumer, not called through `ingest` or +`enterprise-ingest` at all, and does not read the `tenant_id` header -- +every record still lands in the one shared (default) Tantivy index +regardless of tenant. See CLAUDE.md and `/docs/security/threat-model.md`'s +"Read this first" for the full disclosure. + +**Also not built**: Helm/`docker-compose.yml` do gate *whether* +`enterprise-ingest` runs at all (`ingest.requireTenantCredential`, same +flag §13 uses for validation -- see `deploy/helm/sentry/templates/ +enterprise-ingest.yaml`), but `docker-compose.yml`'s version is a +disclosed, weaker approximation of Helm's: Helm achieves genuine +`-mode=server`/`-mode=consumer` mutual exclusivity between `ingest` and +`enterprise-ingest`; compose's `enterprise-ingest` service is a +profile-gated opt-in extra that does NOT split `ingest`'s own +server/consumer halves, so with the `enterprise` profile active, both +`ingest -mode=all` and `enterprise-ingest` independently consume every +message via different consumer groups -- harmless duplication for local +verification, not a topology compose actually enforces the way Helm +does. ## Known gaps (do not treat this phase as done without reading these) @@ -635,17 +688,22 @@ Full accounting: `/docs/security/threat-model.md`. Headline items: that split (declarative request vs. imperative provisioning action) is intentional, not the "two disconnected sources of truth" gap this bullet used to describe. -- **Ingest now has a real tenant identity (§13), but no per-tenant - write-routing yet.** An agent presents a bearer credential - (`enterprise-auth -create-ingest-credential-tenant=`), - `ingest/internal/grpcserver.TenantResolver` validates it (fail-closed) - and attaches the resolved tenant ID to every record as a `tenant_id` - Kafka message header. Nothing downstream reads that header back yet -- - every record still lands in the one shared ClickHouse database and the - one shared Tantivy index no matter what. A newly-provisioned tenant's - storage is real, isolated at query time, and permanently empty until - the write-routing split is built (a real, scoped follow-up, no longer - an undesigned one). +- **Ingest now has a real tenant identity (§13), and ClickHouse + write-routing is built (§14) -- Tantivy write-routing is the one gap + left.** An agent presents a bearer credential (`enterprise-auth + -create-ingest-credential-tenant=`), `ingest/internal/grpcserver. + TenantResolver` validates it (fail-closed) and attaches the resolved + tenant ID to every record as a `tenant_id` Kafka message header. + `enterprise-ingest` reads that header back and routes each record's + ClickHouse write into its own tenant's database (not yet confirmed + against a real ClickHouse in this environment -- see §14). Nothing + reads that header on the Tantivy side yet -- every record still lands + in the one shared Tantivy index no matter what. A newly-provisioned + tenant's ClickHouse database is real, isolated, and actually + populated by write-routed traffic (once confirmed live); its Tantivy + index remains real, isolated at query time, and permanently empty + until Tantivy's write-routing split is built (a real, scoped + follow-up, no longer an undesigned one). - **Human SSO login now works for both OIDC (§3a) and SAML (§3b)** -- each verified with a real fake IdP (genuine cryptographic signing and verification), not yet a real external IdP or a running diff --git a/docs/security/threat-model.md b/docs/security/threat-model.md index 7d5c848..77a4ab8 100644 --- a/docs/security/threat-model.md +++ b/docs/security/threat-model.md @@ -9,13 +9,15 @@ for the full design rationale behind the controls described here. ## Read this first: the single most important open finding -**Updated a second time.** This section originally read "log data +**Updated a third 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." Both ClickHouse *and* -Tantivy connection/index-layer isolation are now built. What's left is +"ClickHouse is isolated but Tantivy isn't," then "ingest tags records +with a tenant identity but nothing routes the write." Both ClickHouse +*and* Tantivy connection/index-layer isolation are built on the *read* +path, and ClickHouse write-routing is now built too. What's left is narrower but still real: **whether a given deployment actually runs the -isolated binary**, and **whether ingest itself is tenant-aware** (it -isn't, for either storage engine). +isolated binaries**, and **Tantivy's write path**, which still has no +per-tenant routing at all. **ClickHouse (the SQL path) is built.** `enterprise/internal/ tenantprovision` (real `CREATE DATABASE`/`CREATE USER`/`GRANT`) and @@ -68,30 +70,43 @@ provisioned, pointing at the same ClickHouse/Postgres. The Helm chart makes the *default*, chart-managed path correct; it isn't a runtime guard against misconfiguration. -**Ingest now has a real tenant identity, but no per-tenant write -routing yet** — a narrower, more precise gap than "not tenant-aware at -all." `chrunner`/`searchclient` prove *read* isolation given tenant- -scoped data exists; a new optional `ingest/internal/grpcserver. -TenantResolver` closes the "does a record know which tenant it belongs -to" half by validating a per-tenant bearer credential an agent presents +**Ingest now has a real tenant identity, and ClickHouse write-routing is +built — Tantivy write-routing is the one gap left.** `chrunner`/ +`searchclient` prove *read* isolation given tenant-scoped data exists; an +optional `ingest/internal/grpcserver.TenantResolver` closes the "does a +record know which tenant it belongs to" half by validating a per-tenant +bearer credential an agent presents (`enterprise-auth -create-ingest-credential-tenant=` mints one; only -its SHA-256 hash is ever stored) against a new `POST +its SHA-256 hash is ever stored) against a `POST /internal/authorize-ingest` endpoint, and attaching the resolved tenant ID to every record as a `tenant_id` Kafka message header before producing it — fail-closed: once a resolver is configured, a missing or invalid credential refuses the whole batch, never falls back to "no -tenant." What's still missing is the "does that identity actually -change where the record is written" half: neither `ingest`'s own -ClickHouse writer nor `search`'s independent Redpanda consumer reads -that header back to route the write anywhere per-tenant yet. Every -record still lands in the one shared ClickHouse database and the one -shared (default) Tantivy index, regardless of tenant — correctly tagged, -not yet isolated at write time. A newly-provisioned tenant's ClickHouse -database and Tantivy index remain real, isolated, and queryable through -`enterprise-api` — and permanently empty, until that write-routing split -is built (likely another "second binary," mirroring `enterprise-api` -itself), which is now scoped, disclosed remaining work, not an -undesigned gap. +tenant." `enterprise/cmd/enterprise-ingest` (a second binary, mirroring +`enterprise-api`) now consumes that header: it reuses `ingest/consumer`'s +own flush loop with `enterprise/internal/chwriter.Registry` — a +per-tenant `*clickhousewriter.Writer` registry, built the same way +`chrunner`'s per-tenant connections are — swapped in as the writer, so a +tagged batch's records are grouped by tenant and each group INSERTed +through that tenant's own ClickHouse connection, fail-closed on an +untagged or unprovisioned tenant. Building it surfaced a real gap in an +already-shipped control: `tenantprovision.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`. What's still missing: +`search`'s independent Redpanda consumer does not read the `tenant_id` +header at all — every ingested record still lands in the one shared +(default) Tantivy index regardless of tenant. A newly-provisioned +tenant's ClickHouse database is now real, isolated, and actually +populated by write-routed agent traffic (once confirmed against a real +cluster); its Tantivy index remains real, isolated, and queryable +through `enterprise-api` — and permanently empty, until Tantivy's own +write-routing split is built, which is now scoped, disclosed remaining +work, not an undesigned gap. ## System overview @@ -129,14 +144,15 @@ sentryctl ──▶ api, alerting (Bearer token when SENTRYCTL_TOKEN is set) ``` Ingest path (agent → Redpanda → ingest → ClickHouse, and Redpanda → -search → Tantivy): `ingest` now resolves and tags each record with a -real tenant ID (see "Read this first" above), but nothing downstream -routes on it yet — every ingested log record still lands in the one -shared `logs` table/index. Tenant isolation for the *write* path is -still out of scope for what's built so far and is not separately -designed in -`/docs/phase-4-isolation-design.md`; named here as a gap that design doc -doesn't yet cover, not just an implementation gap. +search → Tantivy): `ingest` resolves and tags each record with a real +tenant ID (see "Read this first" above). `enterprise/cmd/enterprise-ingest` +now consumes that tag and routes ClickHouse writes to each tenant's own +database. Tantivy's independent Redpanda consumer (`search/src/ +consumer.rs`) still does not consume the tag at all — every record still +lands in the one shared Tantivy index regardless of tenant. That +narrower gap is still out of scope for what's built so far and is not +separately designed in `/docs/phase-4-isolation-design.md`; named here as +a gap that design doc doesn't yet cover, not just an implementation gap. ## Module boundary (trust boundary #1) @@ -438,7 +454,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 and Tantivy both) | **Not implemented, now scoped** — every record still lands in the single shared database/index regardless of tenant; consuming the tenant_id header to route the write is real, disclosed remaining work | +| 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 | +| Ingest tenant *write-routing*, Tantivy | **Not implemented, now scoped** — every record still lands in the single shared Tantivy index regardless of tenant; `search/src/consumer.rs` consuming the tenant_id header to route the write is real, disclosed remaining work | | 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 1b5190d..e5d10b5 100644 --- a/enterprise/README.md +++ b/enterprise/README.md @@ -182,18 +182,14 @@ silently left out: a cross-origin `fetch` with credentials from `web`'s origin needs it), neither of which is verifiable in this environment without a live backend and a browser session to exercise. -- **Ingest write-routing, for either storage engine** -- identity is now - real (see "Ingest tenant identity" below), but nothing consumes it - yet: `chrunner`/`searchclient` prove read isolation given tenant- - scoped data exists, and every record `ingest` produces is now tagged - with a real tenant ID, but neither `ingest/internal/consumer` (the - ClickHouse writer) nor `search/src/consumer.rs` (a completely - independent Redpanda consumer) reads that tag back to route the write - anywhere per-tenant. Every record still lands in the single shared - ClickHouse database and the single shared Tantivy index regardless of - tenant. A newly-provisioned tenant's storage is real and isolated, and - permanently empty. Now scoped, disclosed remaining work, not an - undesigned gap -- see `/docs/security/threat-model.md`. +- **Ingest write-routing, for Tantivy** -- `search/src/consumer.rs` (a + completely independent Redpanda consumer, not called through `ingest` + or `enterprise-ingest` at all) still writes every record into the one + shared (default) Tantivy index, regardless of tenant. The ClickHouse + half is now built (see "Ingest write-routing" below); Tantivy's is + real, disclosed, separate follow-up work -- a different codebase + (Rust) and a different consumer process, not just "the same fix + applied twice." Deployment-topology routing (does traffic actually reach `enterprise-api` instead of `api`) is no longer deferred -- both `deploy/helm/sentry` and @@ -226,17 +222,67 @@ import (`ingest` is AGPL core), same "network boundary, not import boundary" shape `api/authz.HTTPAuthorizer` already uses for the query path. -**What this does not do**: change where a record is actually written. -See "Deliberately deferred" above -- attaching a verified tenant -identity as early as possible (right where the credential is presented) -was built as a self-contained first step; per-tenant write-routing for -both storage engines is separate, scoped follow-up work. +## Ingest write-routing (ClickHouse) + +`enterprise/cmd/enterprise-ingest` is `ingest -mode=consumer`'s multi- +tenant alternative -- same "second binary" shape as `enterprise-api` +next to `api/cmd/api` (AGPL core must never import `enterprise/`, so the +tenant-aware wiring has to live in a binary that imports *into* core, +not the reverse). It reuses `ingest/consumer.Consumer`'s exact flush +loop unchanged, swapping in `enterprise/internal/chwriter.Registry` -- +chrunner's write-side counterpart -- as the writer: one fully separate +`*ingest/clickhousewriter.Writer` (and the `driver.Conn` under it) per +tenant, built once at startup from `rbacstore. +ListProvisionedDataSources` (the same source of truth `chrunner` already +uses for reads). `WriteBatch` groups a Kafka batch's records by their +`tenant_id` tag and writes each tenant's group through its own +dedicated connection, refusing the whole call (matching `ingest/ +consumer`'s existing all-or-nothing batch contract -- no offsets commit, +the batch redelivers) if any record is untagged or tagged with a tenant +that isn't provisioned. `ingest/consumer` and `ingest/clickhousewriter` +moved out of `internal/` for this -- same Go compiler-enforced +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). + +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, +but it would have made every real per-tenant write from `chwriter` fail +with a permission error, since it's the *same* credential used for +both. Fixed by granting `SELECT, INSERT` (not a second, separate +write-only credential -- there's no cross-tenant boundary crossed by +also granting INSERT within a tenant's own database, so one credential +for both directions is the simpler, still-correctly-scoped choice). + +A real multi-tenant deployment runs `ingest -mode=server` (agent-facing, +tags records, unchanged) alongside `enterprise-ingest` (consumer, +per-tenant writes) *instead of* `ingest -mode=consumer` -- see `deploy/ +helm/sentry`'s `ingest.requireTenantCredential` value (gates both the +credential-validation requirement and this mode split together, since +write-routing is only meaningful once records actually carry a +tenant_id to route on) and `docker-compose.yml`'s `enterprise-ingest` +service (a simpler opt-in there -- true `-mode=server`/`-mode=consumer` +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. ## Package layout ``` cmd/enterprise-auth/ config loading, OIDC discovery at startup, health/authorize/features/authorize-ingest endpoints, -mint-service-token, -create-tenant, -grant-membership-*, -revoke-membership-*, -list-memberships-tenant, -create-ingest-credential-tenant, -list-ingest-credentials-tenant, -revoke-ingest-credential cmd/enterprise-api/ multi-tenant-aware alternative to api/cmd/api -- see its own doc comment +cmd/enterprise-ingest/ multi-tenant-aware alternative to ingest -mode=consumer -- see its own doc comment internal/tenant/ the ID type -- see its package doc comment before touching it internal/oidc/ coreos/go-oidc wiring: discovery, login redirect, code exchange + ID token verification internal/saml/ crewjam/saml wiring: SP setup, login redirect, response parsing/validation @@ -247,10 +293,12 @@ internal/rbacstore/ users/tenants/tenant_memberships/data_sources/dashb internal/tenantprovision/ real ClickHouse CREATE DATABASE/USER/GRANT internal/tenantcrd/ syncs -provision-tenant's real result into deploy/operator's Tenant CRD (K8s dynamic client, no cluster needed to test) internal/chrunner/ tenant-scoped api/querylang/executor.SQLRunner +internal/chwriter/ tenant-scoped ingest/consumer.chWriter -- chrunner's write-side counterpart internal/searchclient/ tenant-scoped api/querylang/executor.SearchClient internal/audit/ append-only, hash-chained query audit log, plus the api/queryapi.AuditLogger adapter (queryapi_adapter.go) internal/apiconfig/ enterprise-api's own env-var config +internal/ingestconfig/ enterprise-ingest's own env-var config internal/config/ enterprise-auth's env-var config ``` diff --git a/enterprise/cmd/enterprise-ingest/Dockerfile b/enterprise/cmd/enterprise-ingest/Dockerfile new file mode 100644 index 0000000..352dba3 --- /dev/null +++ b/enterprise/cmd/enterprise-ingest/Dockerfile @@ -0,0 +1,14 @@ +# Same shape as every other Go service's Dockerfile in this repo -- +# context must be the repo root (needs ingest/, proto/, and enterprise/, +# like enterprise-api/Dockerfile does for api/ + proto/ + enterprise/), +# not enterprise/ alone. +# docker build -f enterprise/cmd/enterprise-ingest/Dockerfile -t sentry-enterprise-ingest . +FROM golang:1.25-alpine AS builder +WORKDIR /src +COPY . . +WORKDIR /src/enterprise +RUN CGO_ENABLED=0 GOOS=linux go build -o /out/enterprise-ingest ./cmd/enterprise-ingest + +FROM gcr.io/distroless/static-debian12 +COPY --from=builder /out/enterprise-ingest /enterprise-ingest +ENTRYPOINT ["/enterprise-ingest"] diff --git a/enterprise/cmd/enterprise-ingest/main.go b/enterprise/cmd/enterprise-ingest/main.go new file mode 100644 index 0000000..0e9ef8d --- /dev/null +++ b/enterprise/cmd/enterprise-ingest/main.go @@ -0,0 +1,159 @@ +// Command enterprise-ingest is the multi-tenant-aware alternative to +// running `ingest -mode=consumer` -- reads the same shared +// sentry.logs.raw Redpanda topic ingest/cmd/ingest's agent-facing +// server half (PushBatch) produces onto (see that binary's doc +// comment), but writes each record into its own tenant's dedicated +// ClickHouse database (enterprise/internal/chwriter) instead of the one +// shared table `ingest -mode=consumer` always writes to. +// +// Why a second binary, not a flag on ingest/cmd/ingest: ingest is AGPL +// core and must never import enterprise/ (hack/check-tenant-boundary.sh +// enforces this) -- there is no way for ingest's own binary to +// construct an enterprise-supplied chwriter.Registry (which needs +// rbacstore's per-tenant ClickHouse credentials) without that import. +// enterprise/ importing ingest/ is the allowed direction, so this +// binary lives here instead, reusing ingest/consumer.Consumer's own +// flush loop unchanged with a tenant-aware writer swapped in -- the +// exact same "second binary" shape as enterprise/cmd/enterprise-api +// next to api/cmd/api. +// +// A real multi-tenant deployment runs this binary INSTEAD OF (not +// alongside) `ingest -mode=consumer` -- `ingest -mode=server` (the +// agent-facing half, which tags records with a tenant_id via +// TenantResolver) keeps running unchanged and unconditionally either +// way; only which process consumes sentry.logs.raw and where it writes +// changes. +package main + +import ( + "context" + "fmt" + "log/slog" + "net/http" + "os" + "os/signal" + "strings" + "syscall" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "golang.org/x/sync/errgroup" + + "github.com/sentry/sentry/enterprise/internal/chwriter" + "github.com/sentry/sentry/enterprise/internal/ingestconfig" + "github.com/sentry/sentry/enterprise/internal/rbacstore" + "github.com/sentry/sentry/ingest/consumer" +) + +func main() { + logger := slog.New(slog.NewJSONHandler(os.Stdout, nil)) + + cfg, err := ingestconfig.Load() + if err != nil { + logger.Error("loading config", "error", err) + os.Exit(1) + } + + if len(os.Args) > 1 && os.Args[1] == "-healthcheck" { + os.Exit(runHealthcheck(cfg.HTTPListenAddr)) + } + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + pgDSN := fmt.Sprintf("postgres://%s:%s@%s/%s", cfg.Postgres.Username, cfg.Postgres.Password, cfg.Postgres.Addr, cfg.Postgres.Database) + pgPool, err := pgxpool.New(ctx, pgDSN) + if err != nil { + logger.Error("opening postgres pool", "error", err) + os.Exit(1) + } + defer pgPool.Close() + if err := pgPool.Ping(ctx); err != nil { + logger.Error("pinging postgres", "error", err) + os.Exit(1) + } + rbac := rbacstore.NewStore(pgPool) + + // Same source of truth chrunner.Registry (the read side) already + // uses -- active+credentialed tenants only, see + // rbacstore.ListProvisionedDataSources's doc comment. A tenant + // 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) + 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) + if err != nil { + logger.Error("building tenant write registry", "error", err) + os.Exit(1) + } + defer registry.Close() + + c := consumer.New(logger, consumer.Config{ + Brokers: cfg.Redpanda.Brokers, Topic: cfg.Redpanda.Topic, ConsumerGroup: cfg.Redpanda.ConsumerGroup, + BatchMaxSize: cfg.Batch.MaxSize, FlushIntervalMS: cfg.Batch.FlushIntervalMS, + }, registry) + + mux := http.NewServeMux() + mux.HandleFunc("GET /healthz", func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }) + srv := &http.Server{Addr: cfg.HTTPListenAddr, Handler: mux} + + g, ctx := errgroup.WithContext(ctx) + g.Go(func() error { return c.Run(ctx) }) + g.Go(func() error { + logger.Info("enterprise-ingest healthz listening", "addr", cfg.HTTPListenAddr) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + return err + } + return nil + }) + g.Go(func() error { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + return srv.Shutdown(shutdownCtx) + }) + + logger.Info("enterprise-ingest started") + if err := g.Wait(); err != nil { + logger.Error("enterprise-ingest exited with error", "error", err) + os.Exit(1) + } +} + +// 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 +// runHealthcheck doc comment). +func runHealthcheck(listenAddr string) int { + addr := listenAddr + if strings.HasPrefix(addr, ":") { + addr = "localhost" + addr + } + client := http.Client{Timeout: 3 * time.Second} + resp, err := client.Get("http://" + addr + "/healthz") + if err != nil { + return 1 + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return 1 + } + return 0 +} diff --git a/enterprise/go.mod b/enterprise/go.mod index 4cdfa15..a6aa5d7 100644 --- a/enterprise/go.mod +++ b/enterprise/go.mod @@ -9,6 +9,12 @@ go 1.25.0 // the package that defines it. replace github.com/sentry/sentry/api => ../api +// Same allowed direction, against ingest/ instead -- enterprise/internal/ +// chwriter implements ingest/consumer's chWriter interface, which +// structurally requires importing the package that defines it (see +// that package's doc comment). +replace github.com/sentry/sentry/ingest => ../ingest + // api's own go.mod replace directive for proto/ is module-local and // doesn't propagate here -- enterprise/ needs its own, or `go build` // tries to fetch github.com/sentry/sentry/proto from a real (nonexistent) @@ -16,7 +22,10 @@ replace github.com/sentry/sentry/api => ../api // the generated search gRPC stubs. replace github.com/sentry/sentry/proto => ../proto -require github.com/sentry/sentry/api v0.0.0-00010101000000-000000000000 +require ( + github.com/sentry/sentry/api v0.0.0-00010101000000-000000000000 + github.com/sentry/sentry/ingest v0.0.0-00010101000000-000000000000 +) require ( github.com/ClickHouse/clickhouse-go/v2 v2.48.0 @@ -71,6 +80,7 @@ require ( github.com/pkg/errors v0.9.1 // indirect github.com/russellhaering/goxmldsig v1.4.0 // indirect github.com/segmentio/asm v1.2.1 // indirect + github.com/segmentio/kafka-go v0.4.51 // indirect github.com/shopspring/decimal v1.4.0 // indirect github.com/spf13/pflag v1.0.5 // indirect github.com/x448/float16 v0.8.4 // indirect diff --git a/enterprise/go.sum b/enterprise/go.sum index 477c7b0..0f53b67 100644 --- a/enterprise/go.sum +++ b/enterprise/go.sum @@ -121,6 +121,8 @@ github.com/russellhaering/goxmldsig v1.4.0 h1:8UcDh/xGyQiyrW+Fq5t8f+l2DLB1+zlhYz github.com/russellhaering/goxmldsig v1.4.0/go.mod h1:gM4MDENBQf7M+V824SGfyIUVFWydB7n0KkEubVJl+Tw= github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/segmentio/kafka-go v0.4.51 h1:JgDPPG75tC1rWIS2Me6MwcvXJ6f49UQ4HjAOef71Hno= +github.com/segmentio/kafka-go v0.4.51/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E= github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= @@ -138,6 +140,12 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= +github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY= +github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4= +github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8= +github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= diff --git a/enterprise/internal/chwriter/chwriter.go b/enterprise/internal/chwriter/chwriter.go new file mode 100644 index 0000000..b3f7118 --- /dev/null +++ b/enterprise/internal/chwriter/chwriter.go @@ -0,0 +1,125 @@ +// Package chwriter is enterprise/internal/chrunner's write-side +// counterpart -- the tenant-scoped implementation ingest/consumer. +// Consumer needs to route each batch's records into their own tenant's +// dedicated ClickHouse database, instead of the one shared table +// ingest/cmd/ingest's single-tenant mode always writes to. Requires +// importing ingest/clickhousewriter and ingest/consumer directly (see +// enterprise/go.mod's replace directive) -- same allowed +// "enterprise -> core" import direction chrunner uses for +// api/querylang/executor, just against a different core module. +// +// Design mirrors chrunner.Registry closely: one fully separate +// *clickhousewriter.Writer (and the driver.Conn under it) per tenant, +// built once at construction from an immutable map -- never a shared +// pool with session-level USE, for the same concurrency reasons +// chrunner's doc comment explains. The one real difference: +// chrunner.RunSQL resolves exactly one tenant per call from ctx (a +// single request always belongs to one identity); WriteBatch resolves +// per *record*, since one Kafka batch pulled off the shared +// sentry.logs.raw topic can freely mix records from many different +// tenants -- see ingest/internal/grpcserver's doc comment for why +// there's one shared topic, not topic-per-tenant. +package chwriter + +import ( + "context" + "fmt" + + "github.com/sentry/sentry/ingest/clickhousewriter" + "github.com/sentry/sentry/ingest/consumer" + logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" +) + +// DataSource mirrors chrunner.DataSource -- deliberately not +// enterprise/internal/rbacstore.DataSource itself, so this package +// doesn't need to import rbacstore just to describe "an address and a +// credential." Callers (enterprise-ingest's main.go) adapt rbacstore +// rows into this. +type DataSource struct { + TenantID string + Database string + Username string + Password string +} + +// 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. +type Registry struct { + writers map[string]*clickhousewriter.Writer + closers []func() +} + +// New opens one real ClickHouse connection per DataSource (same native +// address for all of them, different per-tenant credentials -- tenants +// sharing a physical ClickHouse server today, same as chrunner). Fails +// closed: if any one tenant's connection can't be opened, the whole +// 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))} + for _, src := range sources { + w, err := clickhousewriter.New(ctx, clickhousewriter.Config{ + Addr: addr, Database: src.Database, Username: src.Username, Password: src.Password, + }) + if err != nil { + reg.Close() + 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. +func (r *Registry) Close() { + for _, c := range r.closers { + c() + } +} + +// WriteBatch implements ingest/consumer's chWriter interface. Groups +// records by TenantID and writes each tenant's group through its own +// dedicated connection -- fails the *whole* call (matching +// ingest/consumer's existing all-or-nothing batch contract: a failed +// WriteBatch means no offsets are committed and the entire batch is +// redelivered, never partial credit) if any record's tenant is empty +// (no TenantResolver was configured for the PushBatch call that +// produced it -- a multi-tenant deployment must never silently write an +// untagged record somewhere) or unrecognized (not yet provisioned, or +// provisioning failed). Fail closed, same reasoning +// chrunner.Registry.RunSQL's doc comment gives for the read side. +// +// A permanently-unprovisioned or permanently-mistagged tenant would +// stall this consumer's offset progress entirely (every redelivery of +// that batch fails the same way) -- a real, disclosed limitation of +// reusing ingest/consumer's existing all-or-nothing contract rather +// than building new partial-batch-success semantics nothing else in +// this codebase has either. See /docs/phase-4-runbook.md. +func (r *Registry) WriteBatch(ctx context.Context, records []consumer.Record) error { + byTenant := make(map[string][]consumer.Record, len(records)) + for _, rec := range records { + byTenant[rec.TenantID] = append(byTenant[rec.TenantID], rec) + } + + 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)) + } + writer, ok := r.writers[tenantID] + if !ok { + return fmt.Errorf("chwriter: tenant %q has no provisioned ClickHouse connection, refusing to write %d record(s)", tenantID, len(group)) + } + plain := make([]*logsv1.LogRecord, len(group)) + for i, rec := range group { + plain[i] = rec.Record + } + if err := writer.WriteBatch(ctx, plain); err != nil { + return fmt.Errorf("chwriter: writing batch for tenant %q: %w", tenantID, err) + } + } + return nil +} diff --git a/enterprise/internal/chwriter/chwriter_test.go b/enterprise/internal/chwriter/chwriter_test.go new file mode 100644 index 0000000..fe9067a --- /dev/null +++ b/enterprise/internal/chwriter/chwriter_test.go @@ -0,0 +1,157 @@ +// Fail-closed behavior (empty/unknown tenant_id) needs no live +// ClickHouse at all -- Registry.WriteBatch returns before ever touching +// a connection for those cases, so those tests run unconditionally. +// Everything that actually writes data is a real integration test +// against a live ClickHouse (same CHWRITER_TEST_CLICKHOUSE_ADDR +// convention as enterprise/internal/chrunner's own tests), skipped +// unless that's set; run via: +// +// docker run --rm --network sentry_default -v $(pwd)/../../..:/src -w /src/enterprise \ +// -e CHWRITER_TEST_CLICKHOUSE_ADDR=clickhouse:9000 \ +// -e CHWRITER_TEST_CLICKHOUSE_PASSWORD=sentry-dev-only \ +// golang:1.25-alpine go test ./internal/chwriter/... -v +package chwriter + +import ( + "context" + "fmt" + "os" + "testing" + + chdriver "github.com/ClickHouse/clickhouse-go/v2" + "github.com/google/uuid" + + "github.com/sentry/sentry/enterprise/internal/tenantprovision" + "github.com/sentry/sentry/ingest/clickhousewriter" + "github.com/sentry/sentry/ingest/consumer" + logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" +) + +// TestWriteBatchRefusesEmptyTenantID and +// TestWriteBatchRefusesUnknownTenantWithEmptyRegistry construct a +// Registry directly (bypassing New, which would dial ClickHouse) so +// they genuinely run without Docker: WriteBatch's fail-closed checks +// happen before ever touching a real connection, purely a map lookup. +func TestWriteBatchRefusesEmptyTenantID(t *testing.T) { + reg := &Registry{writers: map[string]*clickhousewriter.Writer{}} + err := reg.WriteBatch(context.Background(), []consumer.Record{ + {TenantID: "", Record: &logsv1.LogRecord{Message: "untagged"}}, + }) + if err == nil { + t.Fatal("expected WriteBatch to refuse a record with no tenant_id, not silently drop the tag") + } +} + +func TestWriteBatchRefusesUnknownTenantWithEmptyRegistry(t *testing.T) { + reg := &Registry{writers: map[string]*clickhousewriter.Writer{}} + err := reg.WriteBatch(context.Background(), []consumer.Record{ + {TenantID: "acme", Record: &logsv1.LogRecord{Message: "m"}}, + }) + if err == nil { + t.Fatal("expected WriteBatch to refuse a tenant with no entry in the registry") + } +} + +func testAddr(t *testing.T) string { + t.Helper() + addr := os.Getenv("CHWRITER_TEST_CLICKHOUSE_ADDR") + if addr == "" { + t.Skip("CHWRITER_TEST_CLICKHOUSE_ADDR not set -- skipping live-ClickHouse integration test") + } + return addr +} + +func provisionTestTenant(t *testing.T, addr string) (tenantID string, creds tenantprovision.Credentials) { + t.Helper() + admin, err := chdriver.Open(&chdriver.Options{ + Addr: []string{addr}, + Auth: chdriver.Auth{Database: "default", Username: "default", Password: os.Getenv("CHWRITER_TEST_CLICKHOUSE_PASSWORD")}, + }) + if err != nil { + t.Fatalf("opening admin connection: %v", err) + } + t.Cleanup(func() { admin.Close() }) + + tenantID = "cw" + uuid.NewString()[:8] + creds, err = tenantprovision.New(admin).ProvisionClickHouse(context.Background(), tenantID) + if err != nil { + t.Fatalf("provisioning tenant %s: %v", tenantID, err) + } + if err := admin.Exec(context.Background(), fmt.Sprintf( + "CREATE TABLE `%s`.logs (timestamp DateTime64(9), host String, service String, severity String, message String, attributes Map(String, String), record_id UUID) ENGINE = MergeTree ORDER BY timestamp", + tenantID)); err != nil { + t.Fatalf("creating logs table for tenant %s: %v", tenantID, err) + } + return tenantID, creds +} + +// TestRegistryWritesEachTenantToItsOwnDatabase is the core adversarial +// probe for the write side, complementing chrunner's own read-side +// version: two tenants, two connections inside one Registry, one +// WriteBatch call mixing records from both, and a direct check (via an +// admin connection, not through Registry) that each tenant's row landed +// only in its own database. +func TestRegistryWritesEachTenantToItsOwnDatabase(t *testing.T) { + addr := testAddr(t) + ctx := context.Background() + tenantA, credsA := provisionTestTenant(t, addr) + tenantB, credsB := provisionTestTenant(t, addr) + + reg, err := New(ctx, addr, []DataSource{ + {TenantID: tenantA, Database: tenantA, Username: credsA.Username, Password: credsA.Password}, + {TenantID: tenantB, Database: tenantB, Username: credsB.Username, Password: credsB.Password}, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + defer reg.Close() + + err = reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: tenantA, Record: &logsv1.LogRecord{Host: "h1", Message: "for-a", RecordId: uuid.NewString()}}, + {TenantID: tenantB, Record: &logsv1.LogRecord{Host: "h1", Message: "for-b", RecordId: uuid.NewString()}}, + }) + if err != nil { + t.Fatalf("WriteBatch: %v", err) + } + + admin, err := chdriver.Open(&chdriver.Options{ + Addr: []string{addr}, + Auth: chdriver.Auth{Database: "default", Username: "default", Password: os.Getenv("CHWRITER_TEST_CLICKHOUSE_PASSWORD")}, + }) + if err != nil { + t.Fatalf("opening admin connection: %v", err) + } + defer admin.Close() + + for tenantID, wantMessage := range map[string]string{tenantA: "for-a", tenantB: "for-b"} { + row := admin.QueryRow(ctx, fmt.Sprintf("SELECT message FROM `%s`.logs", tenantID)) + var got string + if err := row.Scan(&got); err != nil { + t.Fatalf("querying %s's logs: %v", tenantID, err) + } + if got != wantMessage { + t.Fatalf("tenant %s's logs.message = %q, want %q", tenantID, got, wantMessage) + } + } +} + +func TestRegistryRefusesUnprovisionedTenant(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() + + err = reg.WriteBatch(ctx, []consumer.Record{ + {TenantID: "some-other-tenant-never-provisioned", Record: &logsv1.LogRecord{Host: "h1", Message: "m", RecordId: uuid.NewString()}}, + }) + if err == nil { + t.Fatal("expected WriteBatch to refuse a tenant with no provisioned connection, not silently drop or misroute it") + } +} diff --git a/enterprise/internal/ingestconfig/ingestconfig.go b/enterprise/internal/ingestconfig/ingestconfig.go new file mode 100644 index 0000000..106a098 --- /dev/null +++ b/enterprise/internal/ingestconfig/ingestconfig.go @@ -0,0 +1,101 @@ +// Package ingestconfig loads enterprise-ingest's configuration from +// environment variables -- same convention as every other Go service in +// this repo. Named ingestconfig, not config, to avoid colliding with +// enterprise/internal/config (enterprise-auth's own, differently-shaped +// config) within the same module -- mirrors enterprise/internal/ +// apiconfig's own naming reasoning exactly. +package ingestconfig + +import ( + "fmt" + "os" + "strconv" + "strings" +) + +type Config struct { + // HTTPListenAddr serves only /healthz -- this binary's actual job + // (Redpanda -> per-tenant ClickHouse) has no other HTTP surface, + // same "just enough for Docker's HEALTHCHECK" shape as every other + // binary in this repo's -healthcheck self-check mode. + HTTPListenAddr string + // ClickHouseAddr is the shared physical ClickHouse server's native + // address -- every tenant's connection (enterprise/internal/ + // chwriter.Registry) dials this same address, just with different + // per-tenant credentials rbacstore already has on file from + // enterprise-api -provision-tenant. Mirrors apiconfig.Config. + // ClickHouseAddr's own doc comment. + ClickHouseAddr string + Postgres PostgresConfig + Redpanda RedpandaConfig + Batch BatchConfig +} + +type PostgresConfig struct { + Addr string + Database string + Username string + Password string +} + +type RedpandaConfig struct { + Brokers []string + Topic string + ConsumerGroup string +} + +type BatchConfig struct { + MaxSize int + FlushIntervalMS int +} + +func Load() (Config, error) { + cfg := Config{ + HTTPListenAddr: getenv("HTTP_LISTEN_ADDR", ":8084"), + ClickHouseAddr: getenv("CLICKHOUSE_ADDR", "localhost:9000"), + Postgres: PostgresConfig{ + Addr: getenv("POSTGRES_ADDR", "localhost:5432"), + Database: getenv("POSTGRES_DATABASE", "sentry_metadata"), + Username: getenv("POSTGRES_USERNAME", "sentry"), + Password: getenv("POSTGRES_PASSWORD", ""), + }, + Redpanda: RedpandaConfig{ + Brokers: strings.Split(getenv("REDPANDA_BROKERS", "localhost:9092"), ","), + // Same default topic ingest/internal/config uses -- this + // binary reads the identical shared sentry.logs.raw topic + // ingest/cmd/ingest's server half (agent-facing PushBatch) + // produces onto; there's no per-tenant topic, see + // ingest/internal/grpcserver's doc comment. + Topic: getenv("REDPANDA_TOPIC", "sentry.logs.raw"), + // A distinct consumer group from ingest/cmd/ingest's own + // default ("sentry-ingest") -- this binary and a + // single-tenant `ingest -mode=consumer` must never share a + // group (each message would only ever reach one of them, + // silently splitting traffic) even though in practice a + // real multi-tenant deployment runs this binary *instead + // of*, not alongside, `ingest -mode=consumer`. + ConsumerGroup: getenv("REDPANDA_CONSUMER_GROUP", "sentry-enterprise-ingest"), + }, + } + + maxSize, err := strconv.Atoi(getenv("CONSUMER_BATCH_MAX_SIZE", "500")) + if err != nil { + return Config{}, fmt.Errorf("CONSUMER_BATCH_MAX_SIZE: %w", err) + } + cfg.Batch.MaxSize = maxSize + + flushMS, err := strconv.Atoi(getenv("CONSUMER_BATCH_FLUSH_INTERVAL_MS", "2000")) + if err != nil { + return Config{}, fmt.Errorf("CONSUMER_BATCH_FLUSH_INTERVAL_MS: %w", err) + } + cfg.Batch.FlushIntervalMS = flushMS + + return cfg, nil +} + +func getenv(key, fallback string) string { + if v := os.Getenv(key); v != "" { + return v + } + return fallback +} diff --git a/enterprise/internal/tenantprovision/tenantprovision.go b/enterprise/internal/tenantprovision/tenantprovision.go index e60dc05..f718b86 100644 --- a/enterprise/internal/tenantprovision/tenantprovision.go +++ b/enterprise/internal/tenantprovision/tenantprovision.go @@ -59,8 +59,16 @@ func New(admin driver.Conn) *Provisioner { } // ProvisionClickHouse creates tenantID's database and a fresh user for -// it, granted SELECT on exactly that database and nothing else. -// Database creation is idempotent (CREATE DATABASE IF NOT EXISTS -- +// it, granted SELECT and INSERT on exactly that database and nothing +// else -- one credential covers both enterprise/internal/chrunner's +// query path and enterprise/internal/chwriter's ingest-write path +// (found while building chwriter: the grant here originally covered +// SELECT only, which would have made every real per-tenant ClickHouse +// write fail with a permission error -- there's no cross-tenant +// boundary crossed by also granting INSERT within a tenant's own +// database, so a single credential for both directions is the simpler, +// still-correctly-scoped choice over provisioning a second write-only +// credential). Database creation is idempotent (CREATE DATABASE IF NOT EXISTS -- // harmless to repeat). User creation deliberately is NOT idempotent // (plain CREATE USER, no IF NOT EXISTS): ClickHouse has no way to read // back an existing user's password, so silently succeeding on a second @@ -108,8 +116,8 @@ func (p *Provisioner) ProvisionClickHouse(ctx context.Context, tenantID string) return Credentials{}, fmt.Errorf("tenantprovision: creating user (already provisioned? this call is not safe to retry): %w", err) } - if err := p.admin.Exec(ctx, fmt.Sprintf("GRANT SELECT ON `%s`.* TO `%s`", database, username)); err != nil { - return Credentials{}, fmt.Errorf("tenantprovision: granting select: %w", err) + if err := p.admin.Exec(ctx, fmt.Sprintf("GRANT SELECT, INSERT ON `%s`.* TO `%s`", database, username)); err != nil { + return Credentials{}, fmt.Errorf("tenantprovision: granting select/insert: %w", err) } return Credentials{Username: username, Password: password}, nil diff --git a/enterprise/internal/tenantprovision/tenantprovision_test.go b/enterprise/internal/tenantprovision/tenantprovision_test.go index 6995f30..42fd736 100644 --- a/enterprise/internal/tenantprovision/tenantprovision_test.go +++ b/enterprise/internal/tenantprovision/tenantprovision_test.go @@ -80,6 +80,41 @@ func TestProvisionClickHouseCreatesUsableTenantConnection(t *testing.T) { } } +// TestProvisionedUserCanInsertIntoOwnDatabase is the regression test for +// a real bug found while building enterprise/internal/chwriter: this +// credential is also the one chwriter.Registry uses to write ingested +// records, so it must be able to INSERT into its own database, not just +// SELECT from it -- the grant originally only covered SELECT, which +// would have made every real per-tenant ClickHouse write fail with a +// permission error. +func TestProvisionedUserCanInsertIntoOwnDatabase(t *testing.T) { + admin := testAdminConn(t) + p := New(admin) + tenantID := testTenantID() + ctx := context.Background() + + creds, err := p.ProvisionClickHouse(ctx, tenantID) + if err != nil { + t.Fatalf("ProvisionClickHouse: %v", err) + } + if err := admin.Exec(ctx, fmt.Sprintf("CREATE TABLE `%s`.marker (id UInt8) ENGINE = Memory", tenantID)); err != nil { + t.Fatalf("creating marker table: %v", err) + } + + tenantConn, err := clickhouse.Open(&clickhouse.Options{ + Addr: []string{os.Getenv("TENANTPROVISION_TEST_CLICKHOUSE_ADDR")}, + Auth: clickhouse.Auth{Database: tenantID, Username: creds.Username, Password: creds.Password}, + }) + if err != nil { + t.Fatalf("opening tenant connection: %v", err) + } + defer tenantConn.Close() + + if err := tenantConn.Exec(ctx, "INSERT INTO marker VALUES (42)"); err != nil { + t.Fatalf("INSERT as the provisioned tenant user into its own database: %v", err) + } +} + // TestProvisionedUserCannotReadOtherTenantDatabase is one of the four // adversarial probes /docs/phase-4-isolation-design.md's verification // plan names for Phase 4 task 8 (see api/queryapi/ diff --git a/ingest/README.md b/ingest/README.md index fe34bf6..e33167e 100644 --- a/ingest/README.md +++ b/ingest/README.md @@ -14,17 +14,46 @@ binary, selected with `--mode`: and the ClickHouse writer both read the same Redpanda messages and need to agree on the same ID for the same record to join search hits back to rows — two consumers generating their own IDs would produce mismatched - ones for what's supposed to be the same record. + ones for what's supposed to be the same record. Also (Phase 4): + resolves an optional per-tenant `Authorization: Bearer ` + credential via `internal/grpcserver.TenantResolver` (nil by default -- + single-tenant behavior unchanged) and attaches the resolved tenant ID + to every produced Kafka message as a `tenant_id` header + (`consumer.TenantIDHeaderKey`) -- see "Multi-tenant write-routing" + below. - **consumer** — reads back off Redpanda, normalizes into the ClickHouse row shape (`internal/normalize`), and batch-writes via the native protocol driver. Commits Redpanda offsets only after a successful ClickHouse write, so a ClickHouse outage causes redelivery on restart rather than - data loss. + data loss. Reads each message's `tenant_id` header (if any) but this + package's own writer (`clickhousewriter.Writer`, used by + `cmd/ingest`'s single-tenant mode) ignores it -- every record still + lands in the one shared ClickHouse database regardless of tag. See + "Multi-tenant write-routing" below for where the tag actually gets + used. - **all** (default) — both, in one process. This is what docker-compose runs. Splitting into two deployments later (e.g. to scale them independently in k8s) is a manifest change, not a code change — see `--mode`. +## Multi-tenant write-routing + +This package (AGPL core) only ever writes to one shared ClickHouse +database, regardless of any `tenant_id` tag a message carries -- routing +a tagged record into its own tenant's dedicated database is +`enterprise/internal/chwriter` and `enterprise/cmd/enterprise-ingest`'s +job (commercial-licensed, per `/CLAUDE.md`'s licensing boundary), not +this package's. `consumer` and `clickhousewriter` live outside +`internal/` (moved there once `enterprise/internal/chwriter` needed to +import them directly -- Go's compiler-enforced `internal/` visibility +rule blocks a separate module from importing anything under +`ingest/internal/...`, the same reason several `api/internal/...` +packages moved out earlier in Phase 4) specifically so `enterprise/` can +reuse this package's own flush loop and ClickHouse batch-insert logic +unchanged, rather than reimplementing either. See +`/enterprise/README.md`'s "Ingest tenant identity"/write-routing +sections for the full story, including what's still not built. + ## Why Redpanda stays in the path Confirmed with the project owner during Phase 0 planning: the gRPC front @@ -65,6 +94,7 @@ full list and defaults) — no config file format for Phase 0: | `CLICKHOUSE_DATABASE` / `_USERNAME` / `_PASSWORD` | `sentry` / `default` / `` | | | `CONSUMER_BATCH_MAX_SIZE` | `500` | Records per ClickHouse batch insert | | `CONSUMER_BATCH_FLUSH_INTERVAL_MS` | `2000` | Max time a partial batch waits before flushing | +| `ENTERPRISE_AUTH_URL` | (empty) | Enables `internal/grpcserver.TenantResolver` -- empty means PushBatch never requires a bearer credential and no `tenant_id` header is ever attached, same as every Phase 0-3 deployment | ## Building & testing @@ -87,11 +117,14 @@ docker build -f ingest/Dockerfile -t sentry-ingest . ## Testing notes -`internal/consumer` and `internal/grpcserver` depend on Redpanda and -ClickHouse only through small interfaces (`reader`/`chWriter` in consumer, +`consumer` and `internal/grpcserver` depend on Redpanda and ClickHouse +only through small interfaces (`reader`/`chWriter` in consumer, `batchProducer` in grpcserver), so the flush/commit/error-handling logic is -unit-tested against fakes — no embedded broker or database needed. What's -*not* covered by these tests: the real `kafka.Reader`/`kafka.Writer` -wiring and the ClickHouse native-protocol driver itself. Those are only -exercised by the docker-compose end-to-end flow described in -`/docs/phase-0-runbook.md`. +unit-tested against fakes — no embedded broker or database needed. This +includes the tenant_id tagging/extraction round trip end to end (a fake +`TenantResolver` in grpcserver's tests, a fake Kafka header in +consumer's) — real logic, fake transport, no live enterprise-auth or +Redpanda needed. What's *not* covered by these tests: the real +`kafka.Reader`/`kafka.Writer` wiring and the ClickHouse native-protocol +driver itself. Those are only exercised by the docker-compose end-to-end +flow described in `/docs/phase-0-runbook.md`. diff --git a/ingest/internal/clickhousewriter/writer.go b/ingest/clickhousewriter/writer.go similarity index 55% rename from ingest/internal/clickhousewriter/writer.go rename to ingest/clickhousewriter/writer.go index ac32637..e162ffd 100644 --- a/ingest/internal/clickhousewriter/writer.go +++ b/ingest/clickhousewriter/writer.go @@ -1,5 +1,12 @@ // Package clickhousewriter batch-inserts normalized log rows into -// ClickHouse using the native protocol driver's batch API. +// ClickHouse using the native protocol driver's batch API. Moved out of +// internal/ (was ingest/internal/clickhousewriter) once +// enterprise/internal/chwriter needed to construct one *Writer per +// tenant -- same reasoning api/internal/dashboards and friends moved out +// of internal/ earlier in Phase 4: Go's compiler-enforced internal/ +// visibility blocks a separate module (enterprise/) from importing +// anything under ingest/internal/..., regardless of what the AGPL/ +// commercial licensing boundary itself would otherwise allow. package clickhousewriter import ( @@ -9,16 +16,30 @@ import ( "github.com/ClickHouse/clickhouse-go/v2" "github.com/ClickHouse/clickhouse-go/v2/lib/driver" - "github.com/sentry/sentry/ingest/internal/config" "github.com/sentry/sentry/ingest/internal/normalize" logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) +// Config is deliberately a local type, not ingest/internal/config. +// ClickHouseConfig -- this package needs to be importable from +// enterprise/ (see the package doc comment), and internal/config must +// stay internal (nothing outside ingest/ needs its other fields, e.g. +// TLSConfig/GRPCConfig, and moving the whole package out just for this +// one struct would be a wider hole than necessary). Same "narrow local +// type, not the storage/config type itself" precedent as +// enterprise/internal/chrunner.DataSource. +type Config struct { + Addr string + Database string + Username string + Password string +} + type Writer struct { conn driver.Conn } -func New(ctx context.Context, cfg config.ClickHouseConfig) (*Writer, error) { +func New(ctx context.Context, cfg Config) (*Writer, error) { conn, err := clickhouse.Open(&clickhouse.Options{ Addr: []string{cfg.Addr}, Auth: clickhouse.Auth{ diff --git a/ingest/cmd/ingest/main.go b/ingest/cmd/ingest/main.go index e3e9c1f..8c35db7 100644 --- a/ingest/cmd/ingest/main.go +++ b/ingest/cmd/ingest/main.go @@ -21,14 +21,36 @@ import ( "golang.org/x/sync/errgroup" - "github.com/sentry/sentry/ingest/internal/clickhousewriter" + "github.com/sentry/sentry/ingest/clickhousewriter" + "github.com/sentry/sentry/ingest/consumer" "github.com/sentry/sentry/ingest/internal/config" - "github.com/sentry/sentry/ingest/internal/consumer" "github.com/sentry/sentry/ingest/internal/grpcserver" "github.com/sentry/sentry/ingest/internal/producer" "github.com/sentry/sentry/ingest/internal/tenantresolver" + logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) +// singleTenantWriter adapts *clickhousewriter.Writer -- which only +// knows how to write to the one ClickHouse database it was constructed +// with -- to consumer.chWriter's tenant-tagged signature, by simply +// ignoring the tag. This is this binary's single-tenant behavior, +// unchanged from before per-tenant ingest credentials existed: every +// record lands in the same shared database regardless of which tenant +// (if any) it was resolved to. enterprise/cmd/enterprise-ingest is +// where a tag-respecting writer (enterprise/internal/chwriter.Registry) +// actually routes per tenant instead. +type singleTenantWriter struct { + w *clickhousewriter.Writer +} + +func (s singleTenantWriter) WriteBatch(ctx context.Context, records []consumer.Record) error { + plain := make([]*logsv1.LogRecord, len(records)) + for i, r := range records { + plain[i] = r.Record + } + return s.w.WriteBatch(ctx, plain) +} + func main() { mode := flag.String("mode", "all", "which half of ingest to run: server | consumer | all") flag.Parse() @@ -70,13 +92,19 @@ func main() { } if *mode == "consumer" || *mode == "all" { - chw, err := clickhousewriter.New(ctx, cfg.ClickHouse) + chw, err := clickhousewriter.New(ctx, clickhousewriter.Config{ + Addr: cfg.ClickHouse.Addr, Database: cfg.ClickHouse.Database, + Username: cfg.ClickHouse.Username, Password: cfg.ClickHouse.Password, + }) if err != nil { logger.Error("connecting to clickhouse", "error", err) os.Exit(1) } defer chw.Close() - c := consumer.New(logger, cfg.Redpanda, cfg.Batch, chw) + c := consumer.New(logger, consumer.Config{ + Brokers: cfg.Redpanda.Brokers, Topic: cfg.Redpanda.Topic, ConsumerGroup: cfg.Redpanda.ConsumerGroup, + BatchMaxSize: cfg.Batch.MaxSize, FlushIntervalMS: cfg.Batch.FlushIntervalMS, + }, singleTenantWriter{w: chw}) g.Go(func() error { return c.Run(ctx) }) } diff --git a/ingest/cmd/ingest/main_test.go b/ingest/cmd/ingest/main_test.go new file mode 100644 index 0000000..1dddb87 --- /dev/null +++ b/ingest/cmd/ingest/main_test.go @@ -0,0 +1,21 @@ +package main + +import ( + "testing" + + "github.com/sentry/sentry/ingest/consumer" + "github.com/sentry/sentry/ingest/internal/grpcserver" +) + +// TestTenantIDHeaderKeyConstantsMatch guards against the literal drift +// grpcserver.TenantIDHeaderKey's doc comment warns about: the producer +// side (grpcserver) and the consumer side (consumer) each define their +// own copy of this Kafka header key name rather than importing across +// that producer/consumer boundary, so nothing else catches a typo in +// either one at compile time. +func TestTenantIDHeaderKeyConstantsMatch(t *testing.T) { + if grpcserver.TenantIDHeaderKey != consumer.TenantIDHeaderKey { + t.Fatalf("grpcserver.TenantIDHeaderKey = %q, consumer.TenantIDHeaderKey = %q -- these must match", + grpcserver.TenantIDHeaderKey, consumer.TenantIDHeaderKey) + } +} diff --git a/ingest/consumer/consumer.go b/ingest/consumer/consumer.go new file mode 100644 index 0000000..e5fcafe --- /dev/null +++ b/ingest/consumer/consumer.go @@ -0,0 +1,180 @@ +// Package consumer reads normalized-on-write LogRecords back off Redpanda +// and batch-writes them into ClickHouse. Offsets are committed only after +// a successful ClickHouse write, so a ClickHouse outage causes redelivery +// on restart rather than silent data loss (at-least-once, not exactly-once +// — Phase 0 doesn't dedupe on the consumer side). +// +// Moved out of internal/ (was ingest/internal/consumer) once +// enterprise/cmd/enterprise-ingest needed to run this same flush loop +// against a per-tenant chWriter -- see clickhousewriter's doc comment +// for why (same Go internal/-visibility reasoning as every other +// package this phase moved out of internal/ for a cross-module +// import). Each record's TenantID (Record.TenantID below) is read from +// the tenant_id Kafka message header grpcserver.TenantIDHeaderKey +// documents -- empty when no TenantResolver was configured for the +// PushBatch call that produced it, exactly as before per-tenant ingest +// credentials existed. What a chWriter implementation *does* with that +// tag varies: ingest/cmd/ingest's single-tenant clickhousewriter.Writer +// ignores it (writes everything to its one configured database, per +// Phase 0-3 behavior, unchanged); enterprise/internal/chwriter.Registry +// (only ever wired into enterprise/cmd/enterprise-ingest, never this +// core binary) routes each record to its tenant's dedicated ClickHouse +// database instead. +package consumer + +import ( + "context" + "log/slog" + "time" + + "github.com/segmentio/kafka-go" + "google.golang.org/protobuf/proto" + + logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" +) + +// TenantIDHeaderKey mirrors ingest/internal/grpcserver.TenantIDHeaderKey +// -- kept as its own constant (not an import of grpcserver, which is +// the agent-facing *producer* side, a different concern from this +// package's consumer side) so this package's dependency list stays +// narrow. Both must name the same literal; a mismatch would silently +// stop tenant_id from ever reaching a consumer, so grpcserver's own +// doc comment on TenantIDHeaderKey cross-references this one. +const TenantIDHeaderKey = "tenant_id" + +// Record pairs a parsed LogRecord with the tenant it was tagged with at +// ingest time (see the package doc comment). +type Record struct { + TenantID string + Record *logsv1.LogRecord +} + +// chWriter is the subset of a ClickHouse writer this package depends +// on, kept as an interface so the flush loop is unit-testable without a +// real ClickHouse connection, and so both the single-tenant +// (clickhousewriter.Writer, adapted) and multi-tenant +// (enterprise/internal/chwriter.Registry) implementations can share +// this exact same consumer loop. +type chWriter interface { + WriteBatch(ctx context.Context, records []Record) error +} + +// reader is the subset of *kafka.Reader used here, as an interface so the +// flush/commit logic can be tested against a fake without a real broker. +type reader interface { + FetchMessage(ctx context.Context) (kafka.Message, error) + CommitMessages(ctx context.Context, msgs ...kafka.Message) error + Close() error +} + +// Config is deliberately a local type, not ingest/internal/config's +// RedpandaConfig/BatchConfig -- same "this package must be importable +// from enterprise/, so it can't depend on ingest/internal/..." reasoning +// as clickhousewriter.Config. +type Config struct { + Brokers []string + Topic string + ConsumerGroup string + BatchMaxSize int + FlushIntervalMS int +} + +type Consumer struct { + logger *slog.Logger + reader reader + writer chWriter + cfg Config +} + +func New(logger *slog.Logger, cfg Config, w chWriter) *Consumer { + r := kafka.NewReader(kafka.ReaderConfig{ + Brokers: cfg.Brokers, + Topic: cfg.Topic, + GroupID: cfg.ConsumerGroup, + }) + return &Consumer{logger: logger, reader: r, writer: w, cfg: cfg} +} + +func (c *Consumer) Run(ctx context.Context) error { + defer c.reader.Close() + + flushInterval := time.Duration(c.cfg.FlushIntervalMS) * time.Millisecond + ticker := time.NewTicker(flushInterval) + defer ticker.Stop() + + msgCh := make(chan kafka.Message) + fetchErrCh := make(chan error, 1) + + go func() { + for { + m, err := c.reader.FetchMessage(ctx) + if err != nil { + fetchErrCh <- err + return + } + select { + case msgCh <- m: + case <-ctx.Done(): + return + } + } + }() + + var records []Record + var pending []kafka.Message + + flush := func() { + if len(records) == 0 { + return + } + if err := c.writer.WriteBatch(ctx, records); err != nil { + c.logger.Error("clickhouse batch write failed, offsets not committed, will redeliver", + "records", len(records), "error", err) + } else if err := c.reader.CommitMessages(ctx, pending...); err != nil { + c.logger.Error("committing offsets after clickhouse write", "error", err) + } else { + c.logger.Debug("batch flushed to clickhouse", "records", len(records)) + } + records = records[:0] + pending = pending[:0] + } + + for { + select { + case <-ctx.Done(): + flush() + return nil + case err := <-fetchErrCh: + flush() + if ctx.Err() != nil { + return nil + } + return err + case <-ticker.C: + flush() + case m := <-msgCh: + var rec logsv1.LogRecord + if err := proto.Unmarshal(m.Value, &rec); err != nil { + c.logger.Warn("skipping unparseable message", "error", err, "offset", m.Offset) + if cerr := c.reader.CommitMessages(ctx, m); cerr != nil { + c.logger.Error("committing offset for poison message", "error", cerr) + } + continue + } + records = append(records, Record{TenantID: tenantIDFromHeaders(m.Headers), Record: &rec}) + pending = append(pending, m) + if len(records) >= c.cfg.BatchMaxSize { + flush() + } + } + } +} + +func tenantIDFromHeaders(headers []kafka.Header) string { + for _, h := range headers { + if h.Key == TenantIDHeaderKey { + return string(h.Value) + } + } + return "" +} diff --git a/ingest/internal/consumer/consumer_test.go b/ingest/consumer/consumer_test.go similarity index 66% rename from ingest/internal/consumer/consumer_test.go rename to ingest/consumer/consumer_test.go index 8624d83..35125ec 100644 --- a/ingest/internal/consumer/consumer_test.go +++ b/ingest/consumer/consumer_test.go @@ -12,7 +12,6 @@ import ( "github.com/segmentio/kafka-go" "google.golang.org/protobuf/proto" - "github.com/sentry/sentry/ingest/internal/config" logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" ) @@ -55,18 +54,18 @@ func (f *fakeReader) commitCount() int { type fakeWriter struct { mu sync.Mutex - batches [][]*logsv1.LogRecord + batches [][]Record failNext bool } -func (f *fakeWriter) WriteBatch(_ context.Context, records []*logsv1.LogRecord) error { +func (f *fakeWriter) WriteBatch(_ context.Context, records []Record) error { f.mu.Lock() defer f.mu.Unlock() if f.failNext { f.failNext = false return errors.New("simulated clickhouse failure") } - batch := make([]*logsv1.LogRecord, len(records)) + batch := make([]Record, len(records)) copy(batch, records) f.batches = append(f.batches, batch) return nil @@ -78,12 +77,12 @@ func (f *fakeWriter) batchCount() int { return len(f.batches) } -func newTestConsumer(r reader, w chWriter, batchCfg config.BatchConfig) *Consumer { +func newTestConsumer(r reader, w chWriter, cfg Config) *Consumer { return &Consumer{ - logger: slog.New(slog.NewTextHandler(io.Discard, nil)), - reader: r, - writer: w, - batchCfg: batchCfg, + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + reader: r, + writer: w, + cfg: cfg, } } @@ -111,7 +110,7 @@ func waitFor(t *testing.T, timeout time.Duration, cond func() bool) { func TestConsumerFlushesOnBatchSize(t *testing.T) { fr := newFakeReader() fw := &fakeWriter{} - c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 2, FlushIntervalMS: 60_000}) + c := newTestConsumer(fr, fw, Config{BatchMaxSize: 2, FlushIntervalMS: 60_000}) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -135,7 +134,7 @@ func TestConsumerFlushesOnBatchSize(t *testing.T) { func TestConsumerFlushesOnTimeout(t *testing.T) { fr := newFakeReader() fw := &fakeWriter{} - c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 1000, FlushIntervalMS: 20}) + c := newTestConsumer(fr, fw, Config{BatchMaxSize: 1000, FlushIntervalMS: 20}) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -156,7 +155,7 @@ func TestConsumerFlushesOnTimeout(t *testing.T) { func TestConsumerDoesNotCommitOnWriteFailure(t *testing.T) { fr := newFakeReader() fw := &fakeWriter{failNext: true} - c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 1, FlushIntervalMS: 60_000}) + c := newTestConsumer(fr, fw, Config{BatchMaxSize: 1, FlushIntervalMS: 60_000}) ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -176,3 +175,40 @@ func TestConsumerDoesNotCommitOnWriteFailure(t *testing.T) { t.Fatalf("fakeWriter should not record a failed batch, got %d recorded", fw.batchCount()) } } + +// TestConsumerExtractsTenantIDFromHeader is the read-side half of the +// producer/consumer tenant_id contract -- ingest/internal/grpcserver +// attaches this header on the way in; this proves the consumer reads it +// back correctly (and that a message with no header at all -- the +// single-tenant/no-resolver case -- gets an empty TenantID, not an +// error). +func TestConsumerExtractsTenantIDFromHeader(t *testing.T) { + fr := newFakeReader() + fw := &fakeWriter{} + c := newTestConsumer(fr, fw, Config{BatchMaxSize: 2, FlushIntervalMS: 60_000}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { _ = c.Run(ctx) }() + + fr.push(kafka.Message{ + Value: mustMarshal(t, &logsv1.LogRecord{Message: "tagged"}), + Headers: []kafka.Header{{Key: TenantIDHeaderKey, Value: []byte("acme")}}, + }) + fr.push(kafka.Message{Value: mustMarshal(t, &logsv1.LogRecord{Message: "untagged"})}) + + waitFor(t, time.Second, func() bool { return fw.batchCount() == 1 }) + + fw.mu.Lock() + defer fw.mu.Unlock() + byMessage := map[string]string{} + for _, r := range fw.batches[0] { + byMessage[r.Record.GetMessage()] = r.TenantID + } + if byMessage["tagged"] != "acme" { + t.Fatalf("TenantID for the tagged message = %q, want acme", byMessage["tagged"]) + } + if byMessage["untagged"] != "" { + t.Fatalf("TenantID for the untagged message = %q, want empty", byMessage["untagged"]) + } +} diff --git a/ingest/internal/consumer/consumer.go b/ingest/internal/consumer/consumer.go deleted file mode 100644 index 7692d3d..0000000 --- a/ingest/internal/consumer/consumer.go +++ /dev/null @@ -1,124 +0,0 @@ -// Package consumer reads normalized-on-write LogRecords back off Redpanda -// and batch-writes them into ClickHouse. Offsets are committed only after -// a successful ClickHouse write, so a ClickHouse outage causes redelivery -// on restart rather than silent data loss (at-least-once, not exactly-once -// — Phase 0 doesn't dedupe on the consumer side). -package consumer - -import ( - "context" - "log/slog" - "time" - - "github.com/segmentio/kafka-go" - "google.golang.org/protobuf/proto" - - "github.com/sentry/sentry/ingest/internal/config" - logsv1 "github.com/sentry/sentry/proto/sentry/logs/v1" -) - -// chWriter is the subset of *clickhousewriter.Writer this package depends -// on, kept as an interface so the flush loop is unit-testable without a -// real ClickHouse connection. -type chWriter interface { - WriteBatch(ctx context.Context, records []*logsv1.LogRecord) error -} - -// reader is the subset of *kafka.Reader used here, as an interface so the -// flush/commit logic can be tested against a fake without a real broker. -type reader interface { - FetchMessage(ctx context.Context) (kafka.Message, error) - CommitMessages(ctx context.Context, msgs ...kafka.Message) error - Close() error -} - -type Consumer struct { - logger *slog.Logger - reader reader - writer chWriter - batchCfg config.BatchConfig -} - -func New(logger *slog.Logger, redpandaCfg config.RedpandaConfig, batchCfg config.BatchConfig, w chWriter) *Consumer { - r := kafka.NewReader(kafka.ReaderConfig{ - Brokers: redpandaCfg.Brokers, - Topic: redpandaCfg.Topic, - GroupID: redpandaCfg.ConsumerGroup, - }) - return &Consumer{logger: logger, reader: r, writer: w, batchCfg: batchCfg} -} - -func (c *Consumer) Run(ctx context.Context) error { - defer c.reader.Close() - - flushInterval := time.Duration(c.batchCfg.FlushIntervalMS) * time.Millisecond - ticker := time.NewTicker(flushInterval) - defer ticker.Stop() - - msgCh := make(chan kafka.Message) - fetchErrCh := make(chan error, 1) - - go func() { - for { - m, err := c.reader.FetchMessage(ctx) - if err != nil { - fetchErrCh <- err - return - } - select { - case msgCh <- m: - case <-ctx.Done(): - return - } - } - }() - - var records []*logsv1.LogRecord - var pending []kafka.Message - - flush := func() { - if len(records) == 0 { - return - } - if err := c.writer.WriteBatch(ctx, records); err != nil { - c.logger.Error("clickhouse batch write failed, offsets not committed, will redeliver", - "records", len(records), "error", err) - } else if err := c.reader.CommitMessages(ctx, pending...); err != nil { - c.logger.Error("committing offsets after clickhouse write", "error", err) - } else { - c.logger.Debug("batch flushed to clickhouse", "records", len(records)) - } - records = records[:0] - pending = pending[:0] - } - - for { - select { - case <-ctx.Done(): - flush() - return nil - case err := <-fetchErrCh: - flush() - if ctx.Err() != nil { - return nil - } - return err - case <-ticker.C: - flush() - case m := <-msgCh: - var rec logsv1.LogRecord - if err := proto.Unmarshal(m.Value, &rec); err != nil { - c.logger.Warn("skipping unparseable message", "error", err, "offset", m.Offset) - if cerr := c.reader.CommitMessages(ctx, m); cerr != nil { - c.logger.Error("committing offset for poison message", "error", cerr) - } - continue - } - records = append(records, &rec) - pending = append(pending, m) - if len(records) >= c.batchCfg.MaxSize { - flush() - } - } - } -} diff --git a/ingest/internal/grpcserver/server.go b/ingest/internal/grpcserver/server.go index 398691a..7012515 100644 --- a/ingest/internal/grpcserver/server.go +++ b/ingest/internal/grpcserver/server.go @@ -43,9 +43,11 @@ import ( ) // TenantIDHeaderKey is the Kafka message header a resolved tenant ID is -// attached under -- exported so internal/consumer (or a future per- -// tenant write-routing consumer) can read it back by the same name -// without duplicating the literal. +// attached under. ingest/consumer.TenantIDHeaderKey names the identical +// literal on the read side -- duplicated rather than imported (this +// package is the agent-facing producer side; consumer is a different +// concern, and importing across them for one string constant isn't +// worth the coupling), so a change here must be mirrored there. const TenantIDHeaderKey = "tenant_id" type Server struct {