Build per-tenant ClickHouse write-routing for ingest (Tantivy still deferred)
ingest tags every record with a tenant_id Kafka header (built previously), but nothing consumed it to actually route the write. This closes that for ClickHouse: enterprise/cmd/enterprise-ingest (a second binary, mirroring enterprise-api) reuses ingest/consumer's own flush loop unchanged, with enterprise/internal/chwriter.Registry -- a per-tenant clickhousewriter.Writer registry -- swapped in as the writer. A batch pulled from the single shared Redpanda topic can mix records from many tenants, so WriteBatch groups by TenantID and dispatches each group to its own tenant's connection, fail- closed on an empty or unrecognized tenant_id. ingest/consumer and ingest/clickhousewriter move out of internal/ (same reason api/internal/* moved earlier this phase: enterprise/ can't import anything under another module's internal/). Their New() constructors now take small local Config structs instead of ingest/internal/config types, so enterprise/ doesn't need that import either. Building this surfaced a real bug: tenantprovision.ProvisionClickHouse only granted SELECT on a tenant's ClickHouse user, correct for chrunner's read-only use but not enough for chwriter reusing the same credential to write -- every real per-tenant write would have failed closed with a permission error. Fixed by widening the grant to SELECT, INSERT; no cross-tenant boundary is crossed by also allowing INSERT within a tenant's own database. Helm gates enterprise-ingest's Deployment on the same ingest.requireTenantCredential flag that already gates tag validation -- write-routing is meaningless without tagging already being required, so they're one decision, not two. docker-compose.yml's version is a disclosed, weaker approximation: it can't achieve Helm's genuine -mode=server/-mode=consumer split, so with the enterprise profile active both ingest and enterprise-ingest independently consume every message via different consumer groups -- harmless duplication for local verification only. Not built: Tantivy's independent Redpanda consumer (search/src/consumer.rs) still doesn't read the tenant_id header at all -- every record still lands in the one shared index regardless of tenant. Not run: the live-ClickHouse- gated tests (chwriter's cross-tenant routing test, tenantprovision's INSERT regression test) -- no Docker/database access in this environment; they're correct Go that has never executed, disclosed as such in docs/security/ threat-model.md and docs/phase-4-runbook.md §14.
This commit is contained in:
+42
-9
@@ -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 <token>`
|
||||
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`.
|
||||
|
||||
@@ -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{
|
||||
@@ -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) })
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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 ""
|
||||
}
|
||||
@@ -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"])
|
||||
}
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user