ClickHouse Batch Insert

.env.example

@@ -4,7 +4,25 @@ REDPANDA_SASL_USER=superuser
 REDPANDA_SASL_PASSWORD=12345678
 # SCRAM-SHA-256 (default) or SCRAM-SHA-512
 REDPANDA_SASL_MECHANISM=SCRAM-SHA-256
+CONSUMER_GROUP=event-processor
+CONSUMER_TOPIC=raw-events
+CONSUMER_RESET_OFFSET=end
+CONSUMER_FETCH_MIN_BYTES=65536
+CONSUMER_FETCH_MAX_BYTES=52428800
+CONSUMER_FETCH_MAX_WAIT=250ms
+CONSUMER_FETCH_MAX_PARTITION_BYTES=4194304
+CONSUMER_BLOCK_REBALANCE_ON_POLL=false
+CONSUMER_MAX_CONCURRENT_FETCHES=2
+CLICKHOUSE_ADDRS=localhost:9000
+CLICKHOUSE_DATABASE=default
+CLICKHOUSE_USERNAME=default
+CLICKHOUSE_PASSWORD=change-me
+CLICKHOUSE_DIAL_TIMEOUT=5s
+CLICKHOUSE_MAX_OPEN_CONNS=10
+CLICKHOUSE_MAX_IDLE_CONNS=5
+CLICKHOUSE_CONN_OPEN_STRATEGY=in_order
+CH_BATCH_SIZE=10000
+CH_BATCH_TIMEOUT=5s
 REDIS_URL=redis://localhost:6379
+ANONYMIZATION_SALT=change-me
 LOG_LEVEL=info
-OTEL_EXPORTER_OTLP_ENDPOINT=localhost:4318
-CONSUMER_GROUP=event-processor

Dockerfile

@@ -27,6 +27,6 @@ USER appuser
 
 EXPOSE 3000
 
-HEALTHCHECK --interval=10s --timeout=3s CMD wget -qO- http://127.0.0.1:3000/healthz || exit 1
+HEALTHCHECK --interval=10s --timeout=3s CMD wget -qO- http://127.0.0.1:3000/livez || exit 1
 
 CMD ["/bin/processor"]

cmd/processor/main.go

@@ -3,7 +3,6 @@ package main
 import (
 	"context"
 	"errors"
-	"log/slog"
 	"net/http"
 	"os"
 	"os/signal"
@@ -14,7 +13,11 @@ import (
 	"github.com/Tanilytics/processing/internal/config"
 	"github.com/Tanilytics/processing/internal/consumer"
 	"github.com/Tanilytics/processing/internal/observability"
+	"github.com/Tanilytics/processing/internal/pipeline"
+	"github.com/Tanilytics/processing/internal/processors"
 	"github.com/Tanilytics/processing/internal/server"
+	"github.com/Tanilytics/processing/internal/storage"
+	"github.com/rs/zerolog"
 )
 
 func main() {
@@ -23,93 +26,119 @@ func main() {
 
 	// 2. Init structured logger
 	logger := observability.NewLogger("processor-service", cfg.LogLevel)
-	logger.Info("starting processor service",
-		"port", cfg.Port,
-		"brokers", cfg.RedpandaBrokers,
-		"redis_url", cfg.RedisURL,
-	)
+	logger.Info().
+		Str("port", cfg.Port).
+		Strs("brokers", cfg.RedpandaBrokers).
+		Strs("clickhouse_addrs", cfg.ClickhouseAddrs).
+		Str("redis_url", cfg.RedisURL).
+		Msg("starting processor service")
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	// 3. Init OTel Tracer
-	shutdownTracer, err := observability.InitTracer(ctx, "processor-service", cfg.OTelEndpoint)
+	// 3. Create HTTP server
+	app := server.NewServer(cfg.Port)
+
+	// 4. Initialize event consumer
+	anonymizer := processors.NewAnonymizer(cfg.AnonymizationSalt)
+	uaParser := processors.NewUserAgentParser()
+	chWriter, err := storage.NewClickHouseWriter(storage.Options{
+		Addrs:            cfg.ClickhouseAddrs,
+		Database:         cfg.ClickhouseDatabase,
+		Username:         cfg.ClickhouseUsername,
+		Password:         cfg.ClickhousePassword,
+		DialTimeout:      cfg.ClickhouseDialTimeout,
+		MaxOpenConns:     cfg.ClickhouseMaxOpenConns,
+		MaxIdleConns:     cfg.ClickhouseMaxIdleConns,
+		ConnOpenStrategy: cfg.ClickhouseConnOpenStrategy,
+	})
 	if err != nil {
-		logger.Error("failed to initialize tracer", "error", err)
-	} else {
-		defer shutdownTracer()
+		logger.Error().Err(err).Msg("failed to initialize clickhouse writer")
+		return
 	}
 
-	// 4. Create HTTP server
-	app := server.NewServer(cfg.Port)
+	processorPipeline := pipeline.NewPipeline(anonymizer, uaParser, chWriter, logger)
 
-	// 5. Initialize event consumer
 	eventConsumer, err := consumer.NewEventConsumer(
 		cfg.RedpandaBrokers,
-		cfg.ConsumerGroup,
-		cfg.RedpandaSASLUser,
-		cfg.RedpandaSASLPassword,
-		cfg.RedpandaSASLMechanism,
+		consumer.ConnectionOptions{
+			GroupID: cfg.ConsumerGroup,
+			SASL: consumer.SASLOptions{
+				User:      cfg.RedpandaSASLUser,
+				Password:  cfg.RedpandaSASLPassword,
+				Mechanism: cfg.RedpandaSASLMechanism,
+			},
+		},
+		consumer.Options{
+			Topic:                  cfg.ConsumerTopic,
+			ResetOffset:            cfg.ConsumerResetOffset,
+			FetchMinBytes:          cfg.ConsumerFetchMinBytes,
+			FetchMaxBytes:          cfg.ConsumerFetchMaxBytes,
+			FetchMaxWait:           cfg.ConsumerFetchMaxWait,
+			FetchMaxPartitionBytes: cfg.ConsumerFetchMaxPartitionBytes,
+			BlockRebalanceOnPoll:   cfg.ConsumerBlockRebalanceOnPoll,
+			MaxConcurrentFetches:   cfg.ConsumerMaxConcurrentFetches,
+			BatchSize:              cfg.ClickhouseBatchSize,
+			BatchTimeout:           cfg.ClickhouseBatchTimeout,
+		},
+		processorPipeline,
 		logger,
 	)
 	if err != nil {
-		logger.Error("failed to initialize consumer", "error", err)
+		logger.Error().Err(err).Msg("failed to initialize consumer")
 		return
 	}
 
 	var consumerWG sync.WaitGroup
 
-	// 6. Start event consumer in a goroutine
+	// 5. Start event consumer in a goroutine
 	consumerWG.Go(func() {
 		if err := eventConsumer.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
-			logger.Error("consumer stopped with error", "error", err)
+			logger.Error().Err(err).Msg("consumer stopped with error")
 			cancel()
 		}
 	})
 
-	// 7. Start HTTP server in a goroutine
+	// 6. Start HTTP server in a goroutine
 	go func() {
-		logger.Info("http server listening", "addr", cfg.Port)
+		logger.Info().Str("addr", cfg.Port).Msg("http server listening")
 		if err := app.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
-			logger.Error("http server error", "error", err)
+			logger.Error().Err(err).Msg("http server error")
 			cancel()
 		}
 	}()
 
-	// 8. Mark service as ready
-	server.SetReady()
-	logger.Info("service is ready")
-
-	// 9. Wait for shutdown signal
+	// 7. Wait for shutdown signal
 	sigCh := make(chan os.Signal, 1)
 	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
 	defer signal.Stop(sigCh)
 	select {
 	case sig := <-sigCh:
-		logger.Info("received signal, shutting down", "signal", sig)
+		logger.Info().Stringer("signal", sig).Msg("received signal, shutting down")
 	case <-ctx.Done():
-		logger.Info("context cancelled, shutting down")
+		logger.Info().Msg("context cancelled, shutting down")
 	}
 
-	// 10. Graceful shutdown
+	// 8. Graceful shutdown
 	cancel()
 	gracefulShutdown(app, logger)
-	eventConsumer.Close()
 	consumerWG.Wait()
+	eventConsumer.Close()
+	if err := chWriter.Close(); err != nil {
+		logger.Error().Err(err).Msg("clickhouse writer shutdown error")
+	}
 
-	logger.Info("shutdown complete")
+	logger.Info().Msg("shutdown complete")
 }
 
 func gracefulShutdown(
 	app interface{ Shutdown(context.Context) error },
-	logger *slog.Logger,
+	logger *zerolog.Logger,
 ) {
-	server.SetNotReady()
-
 	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer shutdownCancel()
 
 	if err := app.Shutdown(shutdownCtx); err != nil {
-		logger.Error("http server shutdown error", "error", err)
+		logger.Error().Err(err).Msg("http server shutdown error")
 	}
 }

go.mod

@@ -3,37 +3,36 @@ module github.com/Tanilytics/processing
 go 1.26.0
 
 require (
+	github.com/ClickHouse/clickhouse-go/v2 v2.43.0
 	github.com/google/uuid v1.6.0
+	github.com/rs/zerolog v1.34.0
 	github.com/stretchr/testify v1.11.1
 	github.com/twmb/franz-go v1.20.7
-	go.opentelemetry.io/otel v1.42.0
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0
-	go.opentelemetry.io/otel/sdk v1.42.0
 )
 
 require (
-	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
+	github.com/ClickHouse/ch-go v0.71.0 // indirect
+	github.com/andybalholm/brotli v1.2.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/go-logr/logr v1.4.3 // indirect
-	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect
-	github.com/klauspost/compress v1.18.4 // indirect
-	github.com/pierrec/lz4/v4 v4.1.25 // indirect
+	github.com/go-faster/city v1.0.1 // indirect
+	github.com/go-faster/errors v0.7.1 // indirect
+	github.com/klauspost/compress v1.18.5 // indirect
+	github.com/kr/pretty v0.3.1 // indirect
+	github.com/mattn/go-colorable v0.1.13 // indirect
+	github.com/mattn/go-isatty v0.0.19 // indirect
+	github.com/paulmach/orb v0.12.0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.26 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/rogpeppe/go-internal v1.14.1 // indirect
+	github.com/segmentio/asm v1.2.1 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
 	github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect
-	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
-	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 // indirect
-	go.opentelemetry.io/otel/metric v1.42.0 // indirect
+	go.opentelemetry.io/otel v1.42.0 // indirect
 	go.opentelemetry.io/otel/trace v1.42.0 // indirect
-	go.opentelemetry.io/proto/otlp v1.9.0 // indirect
+	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.48.0 // indirect
-	golang.org/x/net v0.51.0 // indirect
-	golang.org/x/sys v0.41.0 // indirect
-	golang.org/x/text v0.34.0 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 // indirect
-	google.golang.org/grpc v1.79.2 // indirect
-	google.golang.org/protobuf v1.36.11 // indirect
+	golang.org/x/sys v0.42.0 // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )