Scaffold Phase 0: agent -> Redpanda -> ingest -> ClickHouse -> api -> web
End-to-end log pipeline for Linux hosts, per /docs/architecture.md: - proto: shared gRPC contract (agent <-> ingest), Go bindings checked in - agent: Rust, musl-targeted, journald/file sourcing, RFC5424 parser, mTLS gRPC client, no required config for the common case - ingest: Go, single binary with --mode server|consumer|all; gRPC front end forwards to Redpanda unchanged, consumer normalizes and batch-writes to ClickHouse with at-least-once delivery - storage: ClickHouse schema + a plain SQL-file migration runner - api: minimal SELECT-only query endpoint, plain REST (not gRPC+gateway yet -- see api/README.md) - web: SvelteKit static SPA, one query page - transport: Redpanda compose + topic provisioning - cli: sentryctl ping stub - hack/dev-certs: throwaway CA + cert generation for local mTLS - root docker-compose.yml + docs/phase-0-runbook.md tie it together Not yet run end-to-end against real Docker/ClickHouse/Redpanda -- see the runbook's caveats section before relying on this working as-is.
This commit is contained in:
@@ -0,0 +1,124 @@
|
||||
// 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()
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,178 @@
|
||||
package consumer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"testing"
|
||||
"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"
|
||||
)
|
||||
|
||||
type fakeReader struct {
|
||||
msgs chan kafka.Message
|
||||
|
||||
mu sync.Mutex
|
||||
committed [][]kafka.Message
|
||||
}
|
||||
|
||||
func newFakeReader() *fakeReader {
|
||||
return &fakeReader{msgs: make(chan kafka.Message, 16)}
|
||||
}
|
||||
|
||||
func (f *fakeReader) push(m kafka.Message) { f.msgs <- m }
|
||||
|
||||
func (f *fakeReader) FetchMessage(ctx context.Context) (kafka.Message, error) {
|
||||
select {
|
||||
case m := <-f.msgs:
|
||||
return m, nil
|
||||
case <-ctx.Done():
|
||||
return kafka.Message{}, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeReader) CommitMessages(_ context.Context, msgs ...kafka.Message) error {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.committed = append(f.committed, msgs)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeReader) Close() error { return nil }
|
||||
|
||||
func (f *fakeReader) commitCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.committed)
|
||||
}
|
||||
|
||||
type fakeWriter struct {
|
||||
mu sync.Mutex
|
||||
batches [][]*logsv1.LogRecord
|
||||
failNext bool
|
||||
}
|
||||
|
||||
func (f *fakeWriter) WriteBatch(_ context.Context, records []*logsv1.LogRecord) 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))
|
||||
copy(batch, records)
|
||||
f.batches = append(f.batches, batch)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeWriter) batchCount() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.batches)
|
||||
}
|
||||
|
||||
func newTestConsumer(r reader, w chWriter, batchCfg config.BatchConfig) *Consumer {
|
||||
return &Consumer{
|
||||
logger: slog.New(slog.NewTextHandler(io.Discard, nil)),
|
||||
reader: r,
|
||||
writer: w,
|
||||
batchCfg: batchCfg,
|
||||
}
|
||||
}
|
||||
|
||||
func mustMarshal(t *testing.T, rec *logsv1.LogRecord) []byte {
|
||||
t.Helper()
|
||||
b, err := proto.Marshal(rec)
|
||||
if err != nil {
|
||||
t.Fatalf("marshal: %v", err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
func waitFor(t *testing.T, timeout time.Duration, cond func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatal("condition not met before timeout")
|
||||
}
|
||||
|
||||
func TestConsumerFlushesOnBatchSize(t *testing.T) {
|
||||
fr := newFakeReader()
|
||||
fw := &fakeWriter{}
|
||||
c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 2, FlushIntervalMS: 60_000})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- c.Run(ctx) }()
|
||||
|
||||
fr.push(kafka.Message{Value: mustMarshal(t, &logsv1.LogRecord{Message: "a"})})
|
||||
fr.push(kafka.Message{Value: mustMarshal(t, &logsv1.LogRecord{Message: "b"})})
|
||||
|
||||
waitFor(t, time.Second, func() bool { return fw.batchCount() == 1 })
|
||||
|
||||
fw.mu.Lock()
|
||||
if len(fw.batches[0]) != 2 {
|
||||
t.Fatalf("expected batch of 2 records, got %d", len(fw.batches[0]))
|
||||
}
|
||||
fw.mu.Unlock()
|
||||
|
||||
waitFor(t, time.Second, func() bool { return fr.commitCount() == 1 })
|
||||
}
|
||||
|
||||
func TestConsumerFlushesOnTimeout(t *testing.T) {
|
||||
fr := newFakeReader()
|
||||
fw := &fakeWriter{}
|
||||
c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 1000, FlushIntervalMS: 20})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- c.Run(ctx) }()
|
||||
|
||||
fr.push(kafka.Message{Value: mustMarshal(t, &logsv1.LogRecord{Message: "only-one"})})
|
||||
|
||||
waitFor(t, time.Second, func() bool { return fw.batchCount() == 1 })
|
||||
|
||||
fw.mu.Lock()
|
||||
if len(fw.batches[0]) != 1 {
|
||||
t.Fatalf("expected batch of 1 record, got %d", len(fw.batches[0]))
|
||||
}
|
||||
fw.mu.Unlock()
|
||||
}
|
||||
|
||||
func TestConsumerDoesNotCommitOnWriteFailure(t *testing.T) {
|
||||
fr := newFakeReader()
|
||||
fw := &fakeWriter{failNext: true}
|
||||
c := newTestConsumer(fr, fw, config.BatchConfig{MaxSize: 1, FlushIntervalMS: 60_000})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- c.Run(ctx) }()
|
||||
|
||||
fr.push(kafka.Message{Value: mustMarshal(t, &logsv1.LogRecord{Message: "will-fail"})})
|
||||
|
||||
// Give the flush a moment to run and fail.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
if got := fr.commitCount(); got != 0 {
|
||||
t.Fatalf("expected no commits after a failed clickhouse write, got %d", got)
|
||||
}
|
||||
// The batch was attempted even though writer returned an error.
|
||||
if fw.batchCount() != 0 {
|
||||
t.Fatalf("fakeWriter should not record a failed batch, got %d recorded", fw.batchCount())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user