diff --git a/.env.example b/.env.example index 9f2b838..e150de6 100644 --- a/.env.example +++ b/.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 diff --git a/Dockerfile b/Dockerfile index 223d23c..2f7ea75 100644 --- a/Dockerfile +++ b/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"] diff --git a/cmd/processor/main.go b/cmd/processor/main.go index a455cb5..e4361a3 100644 --- a/cmd/processor/main.go +++ b/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") } } diff --git a/go.mod b/go.mod index cd5aa03..1f59f3c 100644 --- a/go.mod +++ b/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 ) diff --git a/go.sum b/go.sum index 2ab3c17..3fed9b6 100644 --- a/go.sum +++ b/go.sum @@ -1,80 +1,139 @@ -github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= -github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/ClickHouse/ch-go v0.71.0 h1:bUdZ/EZj/LcVHsMqaRUP2holqygrPWQKeMjc6nZoyRM= +github.com/ClickHouse/ch-go v0.71.0/go.mod h1:NwbNc+7jaqfY58dmdDUbG4Jl22vThgx1cYjBw0vtgXw= +github.com/ClickHouse/clickhouse-go/v2 v2.43.0 h1:fUR05TrF1GyvLDa/mAQjkx7KbgwdLRffs2n9O3WobtE= +github.com/ClickHouse/clickhouse-go/v2 v2.43.0/go.mod h1:o6jf7JM/zveWC/PP277BLxjHy5KjnGX/jfljhM4s34g= +github.com/andybalholm/brotli v1.2.0 h1:ukwgCxwYrmACq68yiUqwIWnGY0cTPox/M94sVwToPjQ= +github.com/andybalholm/brotli v1.2.0/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= -github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= -github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= -github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= -github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= -github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= -github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/go-faster/city v1.0.1 h1:4WAxSZ3V2Ws4QRDrscLEDcibJY8uf41H6AhXDrNDcGw= +github.com/go-faster/city v1.0.1/go.mod h1:jKcUJId49qdW3L1qKHH/3wPeUstCVpVSXTM6vO3VcTw= +github.com/go-faster/errors v0.7.1 h1:MkJTnDoEdi9pDabt1dpWf7AA8/BaSYZqibYyhZ20AYg= +github.com/go-faster/errors v0.7.1/go.mod h1:5ySTjWFiphBs07IKuiL69nxdfd5+fzh1u7FPGZP2quo= +github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q= +github.com/google/go-cmp v0.5.2/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= -github.com/klauspost/compress v1.18.4 h1:RPhnKRAQ4Fh8zU2FY/6ZFDwTVTxgJ/EMydqSTzE9a2c= -github.com/klauspost/compress v1.18.4/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.13.6/go.mod h1:/3/Vjq9QcHkK5uEr5lBEmyoZ1iFhe47etQ6QUkpK6sk= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/pierrec/lz4/v4 v4.1.25 h1:kocOqRffaIbU5djlIBr7Wh+cx82C0vtFb0fOurZHqD0= -github.com/pierrec/lz4/v4 v4.1.25/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= +github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= +github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= +github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APPA= +github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y= +github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJuTMNUDYhXwkmfOly8iTdp5TEcJFWZD2D7SIkUc= +github.com/paulmach/orb v0.12.0 h1:z+zOwjmG3MyEEqzv92UN49Lg1JFYx0L9GpGKNVDKk1s= +github.com/paulmach/orb v0.12.0/go.mod h1:5mULz1xQfs3bmQm63QEJA6lNGujuRafwA5S/EnuLaLU= +github.com/paulmach/protoscan v0.2.1/go.mod h1:SpcSwydNLrxUGSDvXvO0P7g7AuhJ7lcKfDlhJCDw2gY= +github.com/pierrec/lz4/v4 v4.1.26 h1:GrpZw1gZttORinvzBdXPUXATeqlJjqUG/D87TKMnhjY= +github.com/pierrec/lz4/v4 v4.1.26/go.mod h1:EoQMVJgeeEOMsCqCzqFm2O0cJvljX2nGZjcRIPL34O4= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/rs/xid v1.6.0/go.mod h1:7XoLgs4eV+QndskICGsho+ADou8ySMSjJKDIan90Nz0= +github.com/rs/zerolog v1.34.0 h1:k43nTLIwcTVQAncfCw4KZ2VY6ukYoZaBPNOE8txlOeY= +github.com/rs/zerolog v1.34.0/go.mod h1:bJsvje4Z08ROH4Nhs5iH600c3IkWhwp44iRc54W6wYQ= +github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= +github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/shopspring/decimal v1.4.0 h1:bxl37RwXBklmTi0C79JfXCEBD1cqqHt0bbgBAGFp81k= +github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk= github.com/twmb/franz-go v1.20.7 h1:P4MGSXJjjAPP3NRGPCks/Lrq+j+twWMVl1qYCVgNmWY= github.com/twmb/franz-go v1.20.7/go.mod h1:0bRX9HZVaoueqFWhPZNi2ODnJL7DNa6mK0HeCrC2bNU= github.com/twmb/franz-go/pkg/kmsg v1.12.0 h1:CbatD7ers1KzDNgJqPbKOq0Bz/WLBdsTH75wgzeVaPc= github.com/twmb/franz-go/pkg/kmsg v1.12.0/go.mod h1:+DPt4NC8RmI6hqb8G09+3giKObE6uD2Eya6CfqBpeJY= -go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= -go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI= +github.com/xdg-go/scram v1.1.1/go.mod h1:RaEWvsqvNKKvBPvcKeFjrG2cJqOkHTiyTpzz23ni57g= +github.com/xdg-go/stringprep v1.0.3/go.mod h1:W3f5j4i+9rC0kuIEJL0ky1VpHXQU3ocBgklLGvcBnW8= +github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= +github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= +github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +go.mongodb.org/mongo-driver v1.11.4/go.mod h1:PTSz5yu21bkT/wXpkS7WR5f0ddqw5quethTUn9WM+2g= go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0 h1:THuZiwpQZuHPul65w4WcwEnkX2QIuMT+UFoOrygtoJw= -go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.42.0/go.mod h1:J2pvYM5NGHofZ2/Ru6zw/TNWnEQp5crgyDeSrYpXkAw= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0 h1:uLXP+3mghfMf7XmV4PkGfFhFKuNWoCvvx5wP/wOXo0o= -go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.42.0/go.mod h1:v0Tj04armyT59mnURNUJf7RCKcKzq+lgJs6QSjHjaTc= -go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= -go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= -go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= -go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts= -go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9RKCAZ3YGuA= -go.opentelemetry.io/otel/sdk/metric v1.42.0/go.mod h1:Ua6AAlDKdZ7tdvaQKfSmnFTdHx37+J4ba8MwVCYM5hc= go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= -go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= -go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= -go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= -go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= +go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/crypto v0.0.0-20220622213112-05595931fe9d/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4= golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= -golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= -golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= -gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= -gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57 h1:JLQynH/LBHfCTSbDWl+py8C+Rg/k1OVH3xfcaiANuF0= -google.golang.org/genproto/googleapis/api v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:kSJwQxqmFXeo79zOmbrALdflXQeAYcUbgS7PbpMknCY= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57 h1:mWPCjDEyshlQYzBpMNHaEof6UX1PmHcaUODUywQ0uac= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260209200024-4cfbd4190f57/go.mod h1:j9x/tPzZkyxcgEFkiKEEGxfvyumM01BEtsW8xzOahRQ= -google.golang.org/grpc v1.79.2 h1:fRMD94s2tITpyJGtBBn7MkMseNpOZU8ZxgC3MMBaXRU= -google.golang.org/grpc v1.79.2/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= -google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= -google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/internal/config/config.go b/internal/config/config.go index a3e9375..05ee298 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -3,7 +3,13 @@ package config import ( "log" "os" + "strconv" "strings" + "time" +) + +const ( + invalidEnvVarLogFormat = "invalid value for environment variable %s: %v" ) type ProcessorConfig struct { @@ -14,11 +20,29 @@ type ProcessorConfig struct { // RedpandaSASLMechanism is the SASL mechanism to use. // Supported values: "SCRAM-SHA-256" (default), "SCRAM-SHA-512". // Leave empty to disable SASL. - RedpandaSASLMechanism string - RedisURL string - LogLevel string - OTelEndpoint string - ConsumerGroup string + RedpandaSASLMechanism string + ConsumerGroup string + ConsumerTopic string + ConsumerResetOffset string + ConsumerFetchMinBytes int32 + ConsumerFetchMaxBytes int32 + ConsumerFetchMaxWait time.Duration + ConsumerFetchMaxPartitionBytes int32 + ConsumerBlockRebalanceOnPoll bool + ConsumerMaxConcurrentFetches int + ClickhouseAddrs []string + ClickhouseDatabase string + ClickhouseUsername string + ClickhousePassword string + ClickhouseDialTimeout time.Duration + ClickhouseBatchSize int + ClickhouseBatchTimeout time.Duration + ClickhouseMaxOpenConns int + ClickhouseMaxIdleConns int + ClickhouseConnOpenStrategy string + RedisURL string + AnonymizationSalt string + LogLevel string } func LoadProcessorConfig() ProcessorConfig { @@ -28,10 +52,32 @@ func LoadProcessorConfig() ProcessorConfig { RedpandaSASLUser: getEnv("REDPANDA_SASL_USER"), RedpandaSASLPassword: getEnv("REDPANDA_SASL_PASSWORD"), RedpandaSASLMechanism: getEnv("REDPANDA_SASL_MECHANISM"), - RedisURL: getEnv("REDIS_URL"), - LogLevel: getEnv("LOG_LEVEL"), - OTelEndpoint: getEnv("OTEL_EXPORTER_OTLP_ENDPOINT"), ConsumerGroup: getEnv("CONSUMER_GROUP"), + ConsumerTopic: getEnv("CONSUMER_TOPIC"), + ConsumerResetOffset: getEnv("CONSUMER_RESET_OFFSET"), + ConsumerFetchMinBytes: getEnvInt32("CONSUMER_FETCH_MIN_BYTES"), + ConsumerFetchMaxBytes: getEnvInt32("CONSUMER_FETCH_MAX_BYTES"), + ConsumerFetchMaxWait: getEnvDuration("CONSUMER_FETCH_MAX_WAIT"), + ConsumerFetchMaxPartitionBytes: getEnvInt32( + "CONSUMER_FETCH_MAX_PARTITION_BYTES", + ), + ConsumerBlockRebalanceOnPoll: getEnvBool( + "CONSUMER_BLOCK_REBALANCE_ON_POLL", + ), + ConsumerMaxConcurrentFetches: getEnvInt("CONSUMER_MAX_CONCURRENT_FETCHES"), + ClickhouseAddrs: getEnvCSV("CLICKHOUSE_ADDRS"), + ClickhouseDatabase: getEnv("CLICKHOUSE_DATABASE"), + ClickhouseUsername: getEnv("CLICKHOUSE_USERNAME"), + ClickhousePassword: getEnv("CLICKHOUSE_PASSWORD"), + ClickhouseDialTimeout: getEnvDuration("CLICKHOUSE_DIAL_TIMEOUT"), + ClickhouseBatchSize: getEnvInt("CH_BATCH_SIZE"), + ClickhouseBatchTimeout: getEnvDuration("CH_BATCH_TIMEOUT"), + ClickhouseMaxOpenConns: getEnvInt("CLICKHOUSE_MAX_OPEN_CONNS"), + ClickhouseMaxIdleConns: getEnvInt("CLICKHOUSE_MAX_IDLE_CONNS"), + ClickhouseConnOpenStrategy: getEnv("CLICKHOUSE_CONN_OPEN_STRATEGY"), + RedisURL: getEnv("REDIS_URL"), + LogLevel: getEnv("LOG_LEVEL"), + AnonymizationSalt: getEnv("ANONYMIZATION_SALT"), } } @@ -44,3 +90,71 @@ func getEnv(key string) string { return v } + +func getEnvInt(key string) int { + v := getEnv(key) + + i, err := strconv.Atoi(strings.TrimSpace(v)) + if err != nil { + log.Printf(invalidEnvVarLogFormat, key, err) + os.Exit(1) + } + + return i +} + +func getEnvInt32(key string) int32 { + v := getEnv(key) + + i, err := strconv.ParseInt(strings.TrimSpace(v), 10, 32) + if err != nil { + log.Printf(invalidEnvVarLogFormat, key, err) + os.Exit(1) + } + + return int32(i) +} + +func getEnvDuration(key string) time.Duration { + v := getEnv(key) + + d, err := time.ParseDuration(strings.TrimSpace(v)) + if err != nil { + log.Printf(invalidEnvVarLogFormat, key, err) + os.Exit(1) + } + + return d +} + +func getEnvBool(key string) bool { + v := getEnv(key) + + b, err := strconv.ParseBool(strings.TrimSpace(v)) + if err != nil { + log.Printf(invalidEnvVarLogFormat, key, err) + os.Exit(1) + } + + return b +} + +func getEnvCSV(key string) []string { + v := getEnv(key) + parts := strings.Split(v, ",") + values := make([]string, 0, len(parts)) + + for _, part := range parts { + trimmed := strings.TrimSpace(part) + if trimmed != "" { + values = append(values, trimmed) + } + } + + if len(values) == 0 { + log.Printf("invalid value for environment variable %s: must contain at least one address", key) + os.Exit(1) + } + + return values +} diff --git a/internal/consumer/consumer.go b/internal/consumer/consumer.go index 340d3b0..fa8e158 100644 --- a/internal/consumer/consumer.go +++ b/internal/consumer/consumer.go @@ -3,35 +3,109 @@ package consumer import ( "context" "encoding/json" - "log/slog" + "errors" + "fmt" + "strings" + "time" "github.com/Tanilytics/processing/internal/models" + "github.com/Tanilytics/processing/internal/pipeline" + "github.com/rs/zerolog" "github.com/twmb/franz-go/pkg/kgo" "github.com/twmb/franz-go/pkg/sasl/scram" ) +type Options struct { + Topic string + ResetOffset string + FetchMinBytes int32 + FetchMaxBytes int32 + FetchMaxWait time.Duration + FetchMaxPartitionBytes int32 + BlockRebalanceOnPoll bool + MaxConcurrentFetches int + BatchSize int + BatchTimeout time.Duration +} + +type ConnectionOptions struct { + GroupID string + SASL SASLOptions +} + +type SASLOptions struct { + User string + Password string + Mechanism string +} + type EventConsumer struct { - client *kgo.Client - logger *slog.Logger + client *kgo.Client + pipeline *pipeline.Pipeline + logger *zerolog.Logger + batchSize int + batchTimeout time.Duration + blockRebalanceOnPoll bool +} + +type consumerBatch struct { + pending []*models.InternalEvent + startedAt time.Time } -func NewEventConsumer(brokers []string, groupID, saslUser, saslPassword, saslMechanism string, logger *slog.Logger) (*EventConsumer, error) { +func NewEventConsumer(brokers []string, connection ConnectionOptions, options Options, p *pipeline.Pipeline, logger *zerolog.Logger) (*EventConsumer, error) { + if strings.TrimSpace(options.Topic) == "" { + return nil, fmt.Errorf("topic must not be empty") + } + + resetOffset, err := parseResetOffset(options.ResetOffset) + if err != nil { + return nil, err + } + if options.FetchMinBytes < 0 { + return nil, fmt.Errorf("fetch min bytes must be >= 0") + } + if options.FetchMaxBytes <= 0 { + return nil, fmt.Errorf("fetch max bytes must be > 0") + } + if options.FetchMaxWait < 0 { + return nil, fmt.Errorf("fetch max wait must be >= 0") + } + if options.FetchMaxPartitionBytes < 0 { + return nil, fmt.Errorf("fetch max partition bytes must be >= 0") + } + if options.MaxConcurrentFetches < 0 { + return nil, fmt.Errorf("max concurrent fetches must be >= 0") + } + if options.BatchSize <= 0 { + return nil, fmt.Errorf("batch size must be > 0") + } + if options.BatchTimeout <= 0 { + return nil, fmt.Errorf("batch timeout must be > 0") + } + opts := []kgo.Opt{ kgo.SeedBrokers(brokers...), - kgo.ConsumerGroup(groupID), - kgo.ConsumeTopics("raw-events"), + kgo.ConsumerGroup(connection.GroupID), + kgo.ConsumeTopics(options.Topic), kgo.DisableAutoCommit(), // manual commit after processing - kgo.FetchMinBytes(1), - kgo.FetchMaxBytes(50 * 1024 * 1024), // 50MB max fetch - kgo.ConsumeResetOffset(kgo.NewOffset().AtStart()), // start from beginning on new group + kgo.FetchMinBytes(options.FetchMinBytes), + kgo.FetchMaxBytes(options.FetchMaxBytes), + kgo.FetchMaxWait(options.FetchMaxWait), + kgo.FetchMaxPartitionBytes(options.FetchMaxPartitionBytes), + kgo.MaxConcurrentFetches(options.MaxConcurrentFetches), + kgo.ConsumeResetOffset(resetOffset), + } + if options.BlockRebalanceOnPoll { + opts = append(opts, kgo.BlockRebalanceOnPoll()) } - if saslUser != "" { + if connection.SASL.User != "" { auth := scram.Auth{ - User: saslUser, - Pass: saslPassword, + User: connection.SASL.User, + Pass: connection.SASL.Password, } - switch saslMechanism { + switch connection.SASL.Mechanism { case "SCRAM-SHA-512": opts = append(opts, kgo.SASL(auth.AsSha512Mechanism())) default: @@ -45,51 +119,263 @@ func NewEventConsumer(brokers []string, groupID, saslUser, saslPassword, saslMec if err != nil { return nil, err } - return &EventConsumer{client: client, logger: logger}, nil + return &EventConsumer{ + client: client, + pipeline: p, + logger: logger, + batchSize: options.BatchSize, + batchTimeout: options.BatchTimeout, + blockRebalanceOnPoll: options.BlockRebalanceOnPoll, + }, nil +} + +func parseResetOffset(value string) (kgo.Offset, error) { + switch strings.ToLower(strings.TrimSpace(value)) { + case "end", "latest": + return kgo.NewOffset().AtEnd(), nil + case "start", "earliest": + return kgo.NewOffset().AtStart(), nil + default: + return kgo.Offset{}, fmt.Errorf("invalid consumer reset offset %q: must be start or end", value) + } } func (c *EventConsumer) Run(ctx context.Context) error { + if c.blockRebalanceOnPoll { + return c.runWithBlockedRebalances(ctx) + } + + batch := consumerBatch{} + for { + if err := c.runOnce(ctx, &batch); err != nil { + return err + } + } +} + +func (c *EventConsumer) runWithBlockedRebalances(ctx context.Context) error { for { - fetches := c.client.PollFetches(ctx) - if ctx.Err() != nil { - return ctx.Err() + if err := ctx.Err(); err != nil { + return err } - if errs := fetches.Errors(); len(errs) > 0 { - for _, e := range errs { - c.logger.Error("fetch error", - "topic", e.Topic, - "partition", e.Partition, - "err", e.Err, - ) - } + if err := c.runBlockedOnce(ctx); err != nil { + return err } + } +} + +func (c *EventConsumer) runBlockedOnce(ctx context.Context) error { + pollCtx, cancel := context.WithTimeout(ctx, c.batchTimeout) + fetches := c.client.PollRecords(pollCtx, c.batchSize) + cancel() + defer c.client.AllowRebalance() + + if err := ctx.Err(); err != nil { + return err + } + + if isIgnorablePollError(fetches.Err()) && fetches.NumRecords() == 0 { + return nil + } + + batch := consumerBatch{} + recordsFetched := c.collectPending(fetches, &batch) + if len(batch.pending) == 0 { + c.commitFetchedOffsets(ctx, recordsFetched) + return nil + } + + if err := c.flushPending(ctx, batch.pending); err != nil { + c.logger.Error().Err(err).Int("batch_size", len(batch.pending)).Msg("pipeline processing failed") + } + + return nil +} + +func (c *EventConsumer) runOnce(ctx context.Context, batch *consumerBatch) error { + if err := c.handleShutdown(ctx, batch); err != nil { + return err + } + + if c.flushReadyBatch(ctx, batch, nil, time.Now()) { + return nil + } + + fetches, pollErr, ok := c.pollFetches(ctx, batch) + if !ok { + return nil + } + + recordsFetched := c.collectPending(fetches, batch) + if len(batch.pending) == 0 { + c.commitFetchedOffsets(ctx, recordsFetched) + return nil + } + + c.flushReadyBatch(ctx, batch, pollErr, time.Now()) + return nil +} + +func (c *EventConsumer) handleShutdown(ctx context.Context, batch *consumerBatch) error { + if ctx.Err() == nil { + return nil + } + if len(batch.pending) == 0 { + return ctx.Err() + } - var events []*models.InternalEvent - fetches.EachRecord(func(record *kgo.Record) { - var event models.InternalEvent - if err := json.Unmarshal(record.Value, &event); err != nil { - c.logger.Error("unmarshal event", - "err", err, - "partition", record.Partition, - "offset", record.Offset, - ) - return // skip malformed records (will be DLQ'd when I feel like doing it) - } - events = append(events, &event) - }) - - if len(events) > 0 { - // Process Events - c.logger.Info("Process Events") + shutdownCtx, shutdownCancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second) + defer shutdownCancel() + + if err := c.flushPending(shutdownCtx, batch.pending); err != nil { + return fmt.Errorf("flush pending events on shutdown: %w", err) + } + + return ctx.Err() +} + +func (c *EventConsumer) flushReadyBatch(ctx context.Context, batch *consumerBatch, pollErr error, now time.Time) bool { + if !batch.shouldFlush(c.batchSize, c.batchTimeout, now) && pollErr != context.DeadlineExceeded { + return false + } + + if err := c.flushPending(ctx, batch.pending); err != nil { + c.logger.Error().Err(err).Int("batch_size", len(batch.pending)).Msg("pipeline processing failed") + return true + } + + batch.reset() + return true +} + +func (c *EventConsumer) pollFetches(ctx context.Context, batch *consumerBatch) (kgo.Fetches, error, bool) { + pollCtx := ctx + var cancel context.CancelFunc + if len(batch.pending) > 0 { + remaining := time.Until(batch.startedAt.Add(c.batchTimeout)) + if remaining <= 0 { + return kgo.Fetches{}, nil, false } - // Commit offsets after successful processing - if err := c.client.CommitUncommittedOffsets(ctx); err != nil { - c.logger.Error("offset commit failed", "err", err) + pollCtx, cancel = context.WithTimeout(ctx, remaining) + } + + fetches := c.client.PollFetches(pollCtx) + pollErr := pollCtx.Err() + if cancel != nil { + cancel() + } + + if ctx.Err() != nil { + return fetches, pollErr, false + } + + return fetches, pollErr, true +} + +func (c *EventConsumer) collectPending(fetches kgo.Fetches, batch *consumerBatch) int { + c.logFetchErrors(fetches) + + batchWasEmpty := len(batch.pending) == 0 + recordsFetched := 0 + fetches.EachRecord(func(record *kgo.Record) { + recordsFetched++ + + var event models.InternalEvent + if err := json.Unmarshal(record.Value, &event); err != nil { + c.logger.Error(). + Err(err). + Int32("partition", record.Partition). + Int64("offset", record.Offset). + Msg("unmarshal event") + return // skip malformed records (will be DLQ'd when I feel like doing it) + } + + batch.pending = append(batch.pending, &event) + }) + + if batchWasEmpty && len(batch.pending) > 0 { + batch.startedAt = time.Now() + } + + return recordsFetched +} + +func (c *EventConsumer) logFetchErrors(fetches kgo.Fetches) { + if errs := fetches.Errors(); len(errs) > 0 { + for _, e := range errs { + c.logger.Error(). + Str("topic", e.Topic). + Int32("partition", e.Partition). + Err(e.Err). + Msg("fetch error") } } } +func (c *EventConsumer) commitFetchedOffsets(ctx context.Context, recordsFetched int) { + if recordsFetched == 0 { + return + } + + if err := c.client.CommitUncommittedOffsets(ctx); err != nil { + c.logger.Error().Err(err).Msg("offset commit failed") + } +} + +func (c *EventConsumer) flushPending(ctx context.Context, events []*models.InternalEvent) error { + if len(events) == 0 { + return nil + } + + if err := c.pipeline.Process(ctx, events); err != nil { + return err + } + + if err := c.client.CommitUncommittedOffsets(ctx); err != nil { + return fmt.Errorf("offset commit failed: %w", err) + } + + return nil +} + +func shouldFlushBatch(batchLen int, batchStartedAt time.Time, batchSize int, batchTimeout time.Duration, now time.Time) bool { + if batchLen == 0 { + return false + } + if batchLen >= batchSize { + return true + } + if batchStartedAt.IsZero() { + return false + } + + return now.Sub(batchStartedAt) >= batchTimeout +} + +func (b *consumerBatch) reset() { + b.pending = nil + b.startedAt = time.Time{} +} + +func (b *consumerBatch) shouldFlush(batchSize int, batchTimeout time.Duration, now time.Time) bool { + if len(b.pending) == 0 { + return false + } + if len(b.pending) >= batchSize { + return true + } + if b.startedAt.IsZero() { + return false + } + + return now.Sub(b.startedAt) >= batchTimeout +} + func (c *EventConsumer) Close() { c.client.Close() } + +func isIgnorablePollError(err error) bool { + return errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) +} diff --git a/internal/consumer/consumer_test.go b/internal/consumer/consumer_test.go index 20834de..c589a92 100644 --- a/internal/consumer/consumer_test.go +++ b/internal/consumer/consumer_test.go @@ -3,12 +3,17 @@ package consumer import ( "context" "encoding/json" - "log/slog" "testing" + "time" "github.com/Tanilytics/processing/internal/models" + "github.com/Tanilytics/processing/internal/pipeline" + "github.com/Tanilytics/processing/internal/processors" + "github.com/Tanilytics/processing/internal/storage" + "github.com/rs/zerolog" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "github.com/twmb/franz-go/pkg/kgo" ) const ( @@ -17,22 +22,36 @@ const ( ) func TestNewEventConsumerWithoutSASL(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) require.NoError(t, err) assert.NotNil(t, consumer) assert.NotNil(t, consumer.client) + assert.NotNil(t, consumer.pipeline) assert.NotNil(t, consumer.logger) } func TestNewEventConsumerWithSASLSHA256(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "SCRAM-SHA-256", logger) + consumer, err := NewEventConsumer( + brokers, + ConnectionOptions{ + GroupID: testGroupID, + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "SCRAM-SHA-256", + }, + }, + testOptions(), + testPipeline(), + &logger, + ) require.NoError(t, err) assert.NotNil(t, consumer) @@ -40,10 +59,23 @@ func TestNewEventConsumerWithSASLSHA256(t *testing.T) { } func TestNewEventConsumerWithSASLSHA512(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "SCRAM-SHA-512", logger) + consumer, err := NewEventConsumer( + brokers, + ConnectionOptions{ + GroupID: testGroupID, + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "SCRAM-SHA-512", + }, + }, + testOptions(), + testPipeline(), + &logger, + ) require.NoError(t, err) assert.NotNil(t, consumer) @@ -51,40 +83,65 @@ func TestNewEventConsumerWithSASLSHA512(t *testing.T) { } func TestNewEventConsumerWithEmptyMechanismFallsBackToSHA256(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "", logger) + consumer, err := NewEventConsumer( + brokers, + ConnectionOptions{ + GroupID: testGroupID, + SASL: SASLOptions{ + User: "user", + Password: "pass", + }, + }, + testOptions(), + testPipeline(), + &logger, + ) require.NoError(t, err) assert.NotNil(t, consumer) } func TestNewEventConsumerWithUnknownMechanismFallsBackToSHA256(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "user", "pass", "UNKNOWN-MECH", logger) + consumer, err := NewEventConsumer( + brokers, + ConnectionOptions{ + GroupID: testGroupID, + SASL: SASLOptions{ + User: "user", + Password: "pass", + Mechanism: "UNKNOWN-MECH", + }, + }, + testOptions(), + testPipeline(), + &logger, + ) require.NoError(t, err) assert.NotNil(t, consumer) } func TestNewEventConsumerWithMultipleBrokers(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{"broker1:9092", "broker2:9092", "broker3:9092"} - consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) require.NoError(t, err) assert.NotNil(t, consumer) } func TestEventConsumerClose(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) require.NoError(t, err) assert.NotPanics(t, func() { @@ -187,10 +244,10 @@ func TestInternalEventUnmarshalInvalidEventType(t *testing.T) { } func TestRunContextCancellation(t *testing.T) { - logger := slog.Default() + logger := zerolog.Nop() brokers := []string{testBroker} - consumer, err := NewEventConsumer(brokers, testGroupID, "", "", "", logger) + consumer, err := NewEventConsumer(brokers, ConnectionOptions{GroupID: testGroupID}, testOptions(), testPipeline(), &logger) require.NoError(t, err) defer consumer.Close() @@ -200,3 +257,164 @@ func TestRunContextCancellation(t *testing.T) { err = consumer.Run(ctx) assert.ErrorIs(t, err, context.Canceled) } + +func TestNewEventConsumerAppliesCustomOptions(t *testing.T) { + logger := zerolog.Nop() + opts := Options{ + Topic: "replay-events", + ResetOffset: "start", + FetchMinBytes: 128 * 1024, + FetchMaxBytes: 8 * 1024 * 1024, + FetchMaxWait: 500 * time.Millisecond, + FetchMaxPartitionBytes: 2 * 1024 * 1024, + BlockRebalanceOnPoll: true, + MaxConcurrentFetches: 4, + BatchSize: 100, + BatchTimeout: 2 * time.Second, + } + + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + require.NoError(t, err) + defer consumer.Close() + + fetchMinBytes, ok := consumer.client.OptValue(kgo.FetchMinBytes).(int32) + require.True(t, ok) + assert.Equal(t, int32(128*1024), fetchMinBytes) + + fetchMaxBytes, ok := consumer.client.OptValue(kgo.FetchMaxBytes).(int32) + require.True(t, ok) + assert.Equal(t, int32(8*1024*1024), fetchMaxBytes) + + fetchMaxWait, ok := consumer.client.OptValue(kgo.FetchMaxWait).(time.Duration) + require.True(t, ok) + assert.Equal(t, 500*time.Millisecond, fetchMaxWait) + + fetchMaxPartitionBytes, ok := consumer.client.OptValue(kgo.FetchMaxPartitionBytes).(int32) + require.True(t, ok) + assert.Equal(t, int32(2*1024*1024), fetchMaxPartitionBytes) + + maxConcurrentFetches, ok := consumer.client.OptValue(kgo.MaxConcurrentFetches).(int) + require.True(t, ok) + assert.Equal(t, 4, maxConcurrentFetches) + + blockRebalanceOnPoll, ok := consumer.client.OptValue(kgo.BlockRebalanceOnPoll).(bool) + require.True(t, ok) + assert.True(t, blockRebalanceOnPoll) + + assert.Equal(t, 100, consumer.batchSize) + assert.Equal(t, 2*time.Second, consumer.batchTimeout) + assert.True(t, consumer.blockRebalanceOnPoll) +} + +func TestParseResetOffset(t *testing.T) { + start, err := parseResetOffset("start") + require.NoError(t, err) + assert.Equal(t, kgo.NewOffset().AtStart().String(), start.String()) + + end, err := parseResetOffset("end") + require.NoError(t, err) + assert.Equal(t, kgo.NewOffset().AtEnd().String(), end.String()) +} + +func TestNewEventConsumerRejectsInvalidResetOffset(t *testing.T) { + logger := zerolog.Nop() + + consumer, err := NewEventConsumer( + []string{testBroker}, + ConnectionOptions{GroupID: testGroupID}, + invalidResetOptions(), + testPipeline(), + &logger, + ) + + require.Error(t, err) + assert.Nil(t, consumer) + assert.ErrorContains(t, err, "must be start or end") +} + +func TestNewEventConsumerRejectsNonPositiveFetchMaxBytes(t *testing.T) { + logger := zerolog.Nop() + opts := testOptions() + opts.FetchMaxBytes = 0 + + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + + require.Error(t, err) + assert.Nil(t, consumer) + assert.ErrorContains(t, err, "fetch max bytes must be > 0") +} + +func TestNewEventConsumerRejectsNonPositiveBatchSize(t *testing.T) { + logger := zerolog.Nop() + opts := testOptions() + opts.BatchSize = 0 + + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + + require.Error(t, err) + assert.Nil(t, consumer) + assert.ErrorContains(t, err, "batch size must be > 0") +} + +func TestNewEventConsumerRejectsNonPositiveBatchTimeout(t *testing.T) { + logger := zerolog.Nop() + opts := testOptions() + opts.BatchTimeout = 0 + + consumer, err := NewEventConsumer([]string{testBroker}, ConnectionOptions{GroupID: testGroupID}, opts, testPipeline(), &logger) + + require.Error(t, err) + assert.Nil(t, consumer) + assert.ErrorContains(t, err, "batch timeout must be > 0") +} + +func TestShouldFlushBatch(t *testing.T) { + now := time.Now() + + assert.False(t, shouldFlushBatch(0, time.Time{}, 100, time.Second, now)) + assert.False(t, shouldFlushBatch(10, now, 100, time.Second, now.Add(500*time.Millisecond))) + assert.True(t, shouldFlushBatch(100, now, 100, time.Second, now)) + assert.True(t, shouldFlushBatch(10, now, 100, time.Second, now.Add(time.Second))) +} + +func testOptions() Options { + return Options{ + Topic: "raw-events", + ResetOffset: "end", + FetchMinBytes: 64 * 1024, + FetchMaxBytes: 50 * 1024 * 1024, + FetchMaxWait: 250 * time.Millisecond, + FetchMaxPartitionBytes: 4 * 1024 * 1024, + MaxConcurrentFetches: 2, + BatchSize: 1000, + BatchTimeout: time.Second, + } +} + +func invalidResetOptions() Options { + opts := testOptions() + opts.ResetOffset = "middle" + return opts +} + +func testPipeline() *pipeline.Pipeline { + writer, err := storage.NewClickHouseWriter(storage.Options{ + Addrs: []string{"some-addr"}, + Database: "default", + Username: "default", + Password: "test-password", + MaxOpenConns: 10, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + if err != nil { + panic(err) + } + logger := zerolog.Nop() + return pipeline.NewPipeline( + processors.NewAnonymizer("test-salt"), + processors.NewUserAgentParser(), + writer, + &logger, + ) +} diff --git a/internal/observability/logger.go b/internal/observability/logger.go index 5c2d818..77f2a29 100644 --- a/internal/observability/logger.go +++ b/internal/observability/logger.go @@ -1,27 +1,27 @@ package observability import ( - "log/slog" "os" + "strings" + "time" + + "github.com/rs/zerolog" ) -func NewLogger(service, level string) *slog.Logger { - var lvl slog.Level - switch level { - case "debug": - lvl = slog.LevelDebug - case "warn": - lvl = slog.LevelWarn - case "error": - lvl = slog.LevelError - default: - lvl = slog.LevelInfo +func NewLogger(service, level string) *zerolog.Logger { + lvl, err := zerolog.ParseLevel(strings.ToLower(level)) + if err != nil { + lvl = zerolog.InfoLevel } - handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{ - Level: lvl, - }) - return slog.New(handler).With( - "service", service, - ) + zerolog.TimeFieldFormat = time.RFC3339Nano + + logger := zerolog.New(os.Stdout). + Level(lvl). + With(). + Timestamp(). + Str("service", service). + Logger() + + return &logger } diff --git a/internal/observability/tracing.go b/internal/observability/tracing.go deleted file mode 100644 index 1406a0a..0000000 --- a/internal/observability/tracing.go +++ /dev/null @@ -1,33 +0,0 @@ -package observability - -import ( - "context" - - "go.opentelemetry.io/otel" - "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp" - "go.opentelemetry.io/otel/sdk/resource" - sdktrace "go.opentelemetry.io/otel/sdk/trace" - semconv "go.opentelemetry.io/otel/semconv/v1.24.0" -) - -func InitTracer(ctx context.Context, serviceName, otlpEndpoint string) (func(), error) { - exporter, err := otlptracehttp.New(ctx, - otlptracehttp.WithEndpoint(otlpEndpoint), - otlptracehttp.WithInsecure(), - ) - if err != nil { - return nil, err - } - - tp := sdktrace.NewTracerProvider( - sdktrace.WithBatcher(exporter), - sdktrace.WithResource(resource.NewWithAttributes( - semconv.SchemaURL, - semconv.ServiceNameKey.String(serviceName), - )), - sdktrace.WithSampler(sdktrace.AlwaysSample()), // sample everything in dev - ) - otel.SetTracerProvider(tp) - - return func() { _ = tp.Shutdown(ctx) }, nil // nolint: errcheck -} diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go new file mode 100644 index 0000000..710c787 --- /dev/null +++ b/internal/pipeline/pipeline.go @@ -0,0 +1,118 @@ +package pipeline + +import ( + "context" + "fmt" + "time" + + "github.com/Tanilytics/processing/internal/models" + "github.com/Tanilytics/processing/internal/processors" + "github.com/Tanilytics/processing/internal/storage" + "github.com/google/uuid" + "github.com/rs/zerolog" +) + +type Pipeline struct { + anonymizer *processors.Anonymizer + uaParser *processors.UserAgentParser + chWriter *storage.ClickHouseWriter + logger *zerolog.Logger +} + +func NewPipeline( + anonymizer *processors.Anonymizer, + uaParser *processors.UserAgentParser, + chWriter *storage.ClickHouseWriter, + logger *zerolog.Logger, +) *Pipeline { + return &Pipeline{ + anonymizer: anonymizer, + uaParser: uaParser, + chWriter: chWriter, + logger: logger, + } +} + +func (p *Pipeline) Process(ctx context.Context, events []*models.InternalEvent) error { + processed := make([]*models.ProcessedEvent, 0, len(events)) + + for _, raw := range events { + select { + case <-ctx.Done(): + return ctx.Err() + default: + } + + pe, err := p.processEvent(raw) + if err != nil { + return err + } + processed = append(processed, pe) + } + + // Step 4: Batch write to ClickHouse + if err := p.chWriter.WriteBatch(ctx, processed); err != nil { + return fmt.Errorf("clickhouse write: %w", err) + } + + return nil +} + +func (p *Pipeline) processEvent(raw *models.InternalEvent) (*models.ProcessedEvent, error) { + // Step 1: Anonymize IP + ipHash, country, region := p.anonymizer.Anonymize(raw.IP) + + // Step 2: Parse User-Agent + browser, os, deviceType := p.uaParser.Parse(raw.UserAgent) + + // Step 3: Parse eventID (in case it fails it derives a uuid instead of failing) + eventID, derived := parseUUID(raw.EventID) + if derived { + p.logger.Debug().Str("event_id", raw.EventID).Str("derived_uuid", eventID.String()).Msg("derived uuid from event id") + } + + return &models.ProcessedEvent{ + EventID: eventID, + SiteID: raw.SiteID, + VisitorID: raw.VisitorID, + SessionID: "unknown", + EventType: raw.EventType, + Timestamp: time.UnixMilli(raw.Timestamp), + URL: raw.URL, + Referrer: raw.Referrer, + UTMSource: raw.UTMSource, + UTMMedium: raw.UTMMedium, + UTMCampaign: raw.UTMCampaign, + Country: country, + Region: region, + DeviceType: deviceType, + Browser: browser, + OS: os, + ScreenWidth: clampUint16(raw.SessionContext.ScreenWidth), + Properties: raw.Properties, + IPHash: ipHash, + ConsentGiven: false, + }, nil +} + +func parseUUID(raw string) (uuid.UUID, bool) { + parsed, err := uuid.Parse(raw) + if err == nil { + return parsed, false + } + + return uuid.NewSHA1(uuid.NameSpaceOID, []byte(raw)), true +} + +func clampUint16(value int) uint16 { + const maxUint16 = 1<<16 - 1 + + if value <= 0 { + return 0 + } + if value > maxUint16 { + return maxUint16 + } + + return uint16(value) +} diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go new file mode 100644 index 0000000..e7a9e85 --- /dev/null +++ b/internal/pipeline/pipeline_test.go @@ -0,0 +1,121 @@ +package pipeline + +import ( + "testing" + + "github.com/Tanilytics/processing/internal/models" + "github.com/Tanilytics/processing/internal/processors" + "github.com/Tanilytics/processing/internal/storage" + "github.com/google/uuid" + "github.com/rs/zerolog" +) + +const invalidEventID = "evt-123" + +func TestParseUUIDKeepsValidUUID(t *testing.T) { + expected := uuid.New() + + parsed, derived := parseUUID(expected.String()) + if derived { + t.Fatal("expected valid UUID to be preserved") + } + if parsed != expected { + t.Fatalf("parsed = %s, want %s", parsed, expected) + } +} + +func TestParseUUIDDerivesDeterministicUUID(t *testing.T) { + first, firstDerived := parseUUID(invalidEventID) + second, secondDerived := parseUUID(invalidEventID) + + if !firstDerived || !secondDerived { + t.Fatal("expected invalid event IDs to derive UUIDs") + } + if first == uuid.Nil { + t.Fatal("expected derived UUID to be non-nil") + } + if first != second { + t.Fatalf("derived UUIDs differ: %s vs %s", first, second) + } +} + +func TestProcessEventDerivesUUIDAndClampsScreenWidth(t *testing.T) { + writer, err := storage.NewClickHouseWriter(storage.Options{ + Addrs: []string{"some-addr"}, + Database: "default", + Username: "default", + Password: "test-password", + MaxOpenConns: 10, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + if err != nil { + t.Fatalf("failed to create writer: %v", err) + } + logger := zerolog.Nop() + p := NewPipeline( + processors.NewAnonymizer("test-salt"), + processors.NewUserAgentParser(), + writer, + &logger, + ) + + processed, err := p.processEvent(&models.InternalEvent{ + EventID: invalidEventID, + SiteID: "site-1", + VisitorID: "visitor-1", + EventType: models.EventPageView, + Timestamp: 1700000000000, + URL: "https://example.com", + IP: "192.168.1.12", + UserAgent: "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1", + SessionContext: models.SessionContext{ + ScreenWidth: 100000, + }, + }) + if err != nil { + t.Fatalf("processEvent error = %v", err) + } + + expectedID, _ := parseUUID(invalidEventID) + if processed.EventID != expectedID { + t.Fatalf("EventID = %s, want %s", processed.EventID, expectedID) + } + if processed.ScreenWidth != 65535 { + t.Fatalf("ScreenWidth = %d, want 65535", processed.ScreenWidth) + } + if processed.DeviceType != "mobile" { + t.Fatalf("DeviceType = %q, want %q", processed.DeviceType, "mobile") + } + if processed.Browser != "Safari" { + t.Fatalf("Browser = %q, want %q", processed.Browser, "Safari") + } + if processed.OS != "iOS" { + t.Fatalf("OS = %q, want %q", processed.OS, "iOS") + } + if processed.IPHash == "" { + t.Fatal("expected IPHash to be set") + } +} + +func TestClampUint16(t *testing.T) { + tests := []struct { + name string + input int + want uint16 + }{ + {name: "negative", input: -1, want: 0}, + {name: "zero", input: 0, want: 0}, + {name: "fits", input: 1024, want: 1024}, + {name: "too large", input: 70000, want: 65535}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := clampUint16(tt.input) + if got != tt.want { + t.Fatalf("clampUint16(%d) = %d, want %d", tt.input, got, tt.want) + } + }) + } +} diff --git a/internal/processors/anonymizer.go b/internal/processors/anonymizer.go new file mode 100644 index 0000000..9f77b86 --- /dev/null +++ b/internal/processors/anonymizer.go @@ -0,0 +1,53 @@ +package processors + +import ( + "crypto/sha256" + "fmt" + "net" + "strings" +) + +type Anonymizer struct { + dailySalt string // use a static salt for simplicity. will add rotation later +} + +func NewAnonymizer(salt string) *Anonymizer { + return &Anonymizer{dailySalt: salt} +} + +// Anonymize truncates the IP and returns a hash + geo data. +// geo lookup is stubbed. will add MaxMind GeoIP later +func (a *Anonymizer) Anonymize(rawIP string) (ipHash, country, region string) { + truncated := truncateIP(rawIP) + if truncated == "" { + // Keep IP as unknown when parsing fails. + country = "" + region = "" + return + } + + hash := sha256.Sum256([]byte(truncated + a.dailySalt)) + ipHash = fmt.Sprintf("%x", hash[:16]) // 32-char hex + + // returns empty strings + country = "" + region = "" + return +} + +func truncateIP(ip string) string { + parsed := net.ParseIP(ip) + if parsed == nil { + return "" + } + if parsed.To4() != nil { + // IPv4: zero last octet (192.168.1.123 → 192.168.1.0) + parts := strings.Split(parsed.String(), ".") + parts[3] = "0" + return strings.Join(parts, ".") + } + // IPv6: zero last 80 bits (/48 mask) + mask := net.CIDRMask(48, 128) + masked := parsed.Mask(mask) + return masked.String() +} diff --git a/internal/processors/anonymizer_test.go b/internal/processors/anonymizer_test.go new file mode 100644 index 0000000..e557f79 --- /dev/null +++ b/internal/processors/anonymizer_test.go @@ -0,0 +1,63 @@ +package processors + +import ( + "encoding/hex" + "testing" +) + +func TestTruncateIP(t *testing.T) { + tests := []struct { + name string + input string + want string + }{ + {name: "ipv4", input: "192.168.1.123", want: "192.168.1.0"}, + {name: "ipv6", input: "2001:db8:abcd:1234:5678:9abc:def0:1234", want: "2001:db8:abcd::"}, + {name: "invalid", input: "not-an-ip", want: ""}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := truncateIP(tt.input) + if got != tt.want { + t.Fatalf("truncateIP(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestAnonymizeUsesTruncatedIPv4Range(t *testing.T) { + a := NewAnonymizer("test-salt") + + firstHash, firstCountry, firstRegion := a.Anonymize("192.168.1.123") + secondHash, secondCountry, secondRegion := a.Anonymize("192.168.1.200") + thirdHash, _, _ := a.Anonymize("192.168.2.10") + + if firstHash != secondHash { + t.Fatalf("expected hashes from same /24 to match, got %q and %q", firstHash, secondHash) + } + if firstHash == thirdHash { + t.Fatalf("expected hashes from different /24 ranges to differ, both were %q", firstHash) + } + if len(firstHash) != 32 { + t.Fatalf("hash length = %d, want 32", len(firstHash)) + } + if _, err := hex.DecodeString(firstHash); err != nil { + t.Fatalf("hash %q is not valid hex: %v", firstHash, err) + } + if firstCountry != "" || firstRegion != "" || secondCountry != "" || secondRegion != "" { + t.Fatalf("expected geo fields to be empty, got %q/%q and %q/%q", firstCountry, firstRegion, secondCountry, secondRegion) + } +} + +func TestAnonymizeIncludesSaltInHash(t *testing.T) { + first := NewAnonymizer("salt-a") + second := NewAnonymizer("salt-b") + + firstHash, _, _ := first.Anonymize("203.0.113.7") + secondHash, _, _ := second.Anonymize("203.0.113.7") + + if firstHash == secondHash { + t.Fatalf("expected different salts to produce different hashes, both were %q", firstHash) + } +} diff --git a/internal/processors/useragent.go b/internal/processors/useragent.go new file mode 100644 index 0000000..a2cbcc9 --- /dev/null +++ b/internal/processors/useragent.go @@ -0,0 +1,95 @@ +package processors + +import "strings" + +type UserAgentParser struct{} + +func NewUserAgentParser() *UserAgentParser { + return &UserAgentParser{} +} + +func (p *UserAgentParser) Parse(rawUA string) (browser, os, deviceType string) { + ua := strings.ToLower(strings.TrimSpace(rawUA)) + if ua == "" { + return "unknown", "unknown", "unknown" + } + + if isBotUserAgent(ua) { + return "Bot", "unknown", "unknown" + } + + return detectBrowser(ua), detectOS(ua), detectDeviceType(ua) +} + +func detectBrowser(ua string) string { + switch { + case strings.Contains(ua, "edg/"): + return "Edge" + case strings.Contains(ua, "opr/") || strings.Contains(ua, "opera"): + return "Opera" + case strings.Contains(ua, "fxios/") || strings.Contains(ua, "firefox/"): + return "Firefox" + case strings.Contains(ua, "crios/"): + return "Chrome" + case (strings.Contains(ua, "chrome/") || strings.Contains(ua, "chromium/")): + return "Chrome" + case strings.Contains(ua, "trident/") || strings.Contains(ua, "msie "): + return "Internet Explorer" + case strings.Contains(ua, "safari/") && !strings.Contains(ua, "android"): + return "Safari" + case strings.HasPrefix(ua, "curl/"): + return "cURL" + default: + return "unknown" + } +} + +func detectOS(ua string) string { + switch { + case strings.Contains(ua, "windows phone"): + return "Windows Phone" + case strings.Contains(ua, "android"): + return "Android" + case strings.Contains(ua, "iphone") || strings.Contains(ua, "ipad") || strings.Contains(ua, "ipod") || strings.Contains(ua, "cpu iphone os") || strings.Contains(ua, "cpu os"): + return "iOS" + case strings.Contains(ua, "cros"): + return "ChromeOS" + case strings.Contains(ua, "windows nt"): + return "Windows" + case strings.Contains(ua, "mac os x") || strings.Contains(ua, "macintosh"): + return "macOS" + case strings.Contains(ua, "linux") || strings.Contains(ua, "x11"): + return "Linux" + default: + return "unknown" + } +} + +func detectDeviceType(ua string) string { + switch { + case strings.Contains(ua, "ipad") || strings.Contains(ua, "tablet") || strings.Contains(ua, "kindle") || strings.Contains(ua, "silk/") || strings.Contains(ua, "playbook") || strings.Contains(ua, "sm-t"): + return "tablet" + case strings.Contains(ua, "android"): + if strings.Contains(ua, "mobile") { + return "mobile" + } + return "tablet" + case strings.Contains(ua, "mobi") || strings.Contains(ua, "iphone") || strings.Contains(ua, "ipod") || strings.Contains(ua, "windows phone"): + return "mobile" + case strings.Contains(ua, "windows nt") || strings.Contains(ua, "macintosh") || strings.Contains(ua, "linux") || strings.Contains(ua, "x11"): + return "desktop" + default: + return "unknown" + } +} + +func isBotUserAgent(ua string) bool { + botTokens := []string{"bot", "crawler", "spider", "slurp", "bingpreview", "mediapartners", "headless", "phantomjs"} + for _, token := range botTokens { + if strings.Contains(ua, token) { + return true + } + } + + return false +} diff --git a/internal/processors/useragent_test.go b/internal/processors/useragent_test.go new file mode 100644 index 0000000..4581549 --- /dev/null +++ b/internal/processors/useragent_test.go @@ -0,0 +1,66 @@ +package processors + +import "testing" + +func TestUserAgentParserParse(t *testing.T) { + parser := NewUserAgentParser() + + tests := []struct { + name string + ua string + browser string + os string + deviceType string + }{ + { + name: "desktop chrome", + ua: "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", + browser: "Chrome", + os: "Windows", + deviceType: "desktop", + }, + { + name: "iphone safari", + ua: "Mozilla/5.0 (iPhone; CPU iPhone OS 17_3 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1", + browser: "Safari", + os: "iOS", + deviceType: "mobile", + }, + { + name: "android tablet chrome", + ua: "Mozilla/5.0 (Linux; Android 13; SM-T970) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", + browser: "Chrome", + os: "Android", + deviceType: "tablet", + }, + { + name: "bot", + ua: "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)", + browser: "Bot", + os: "unknown", + deviceType: "unknown", + }, + { + name: "empty", + ua: "", + browser: "unknown", + os: "unknown", + deviceType: "unknown", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + browser, os, deviceType := parser.Parse(tt.ua) + if browser != tt.browser { + t.Fatalf("browser = %q, want %q", browser, tt.browser) + } + if os != tt.os { + t.Fatalf("os = %q, want %q", os, tt.os) + } + if deviceType != tt.deviceType { + t.Fatalf("deviceType = %q, want %q", deviceType, tt.deviceType) + } + }) + } +} diff --git a/internal/server/health.go b/internal/server/health.go index 04989ff..f6bbf63 100644 --- a/internal/server/health.go +++ b/internal/server/health.go @@ -3,40 +3,17 @@ package server import ( "encoding/json" "net/http" - "sync/atomic" ) -var ready atomic.Bool - type statusResponse struct { Status string `json:"status"` } -// SetReady marks the service as ready to serve traffic. -func SetReady() { - ready.Store(true) -} - -// SetNotReady marks the service as not ready to serve traffic. -func SetNotReady() { - ready.Store(false) -} - // Healthz is a liveness probe. Returns 200 OK if the process is running. func Healthz(w http.ResponseWriter, _ *http.Request) { writeJSON(w, http.StatusOK, statusResponse{Status: "ok"}) } -// Readyz is a readiness probe. Returns 200 OK only when service is ready. -func Readyz(w http.ResponseWriter, _ *http.Request) { - if !ready.Load() { - writeJSON(w, http.StatusServiceUnavailable, statusResponse{Status: "not ready"}) - return - } - - writeJSON(w, http.StatusOK, statusResponse{Status: "ready"}) -} - func writeJSON(w http.ResponseWriter, status int, payload statusResponse) { body, err := json.Marshal(payload) if err != nil { diff --git a/internal/server/server.go b/internal/server/server.go index 0446f35..cb206b2 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -4,8 +4,7 @@ import "net/http" func NewServer(addr string) *http.Server { mux := http.NewServeMux() - mux.HandleFunc("/healthz", Healthz) - mux.HandleFunc("/readyz", Readyz) + mux.HandleFunc("/livez", Healthz) return &http.Server{ Addr: addr, diff --git a/internal/storage/clickhouse.go b/internal/storage/clickhouse.go new file mode 100644 index 0000000..415a857 --- /dev/null +++ b/internal/storage/clickhouse.go @@ -0,0 +1,113 @@ +package storage + +import ( + "context" + "fmt" + "strings" + "time" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/Tanilytics/processing/internal/models" +) + +type Options struct { + Addrs []string + Database string + Username string + Password string + DialTimeout time.Duration + MaxOpenConns int + MaxIdleConns int + ConnOpenStrategy string +} + +type ClickHouseWriter struct { + conn clickhouse.Conn +} + +func NewClickHouseWriter(options Options) (*ClickHouseWriter, error) { + strategy, err := parseConnOpenStrategy(options.ConnOpenStrategy) + if err != nil { + return nil, err + } + if len(options.Addrs) == 0 { + return nil, fmt.Errorf("clickhouse addrs must not be empty") + } + if options.MaxOpenConns <= 0 { + return nil, fmt.Errorf("clickhouse max open conns must be > 0") + } + if options.MaxIdleConns < 0 { + return nil, fmt.Errorf("clickhouse max idle conns must be >= 0") + } + + conn, err := clickhouse.Open(&clickhouse.Options{ + Addr: options.Addrs, + Auth: clickhouse.Auth{ + Database: options.Database, + Username: options.Username, + Password: options.Password, + }, + Settings: clickhouse.Settings{ + "max_execution_time": 60, + }, + Compression: &clickhouse.Compression{ + Method: clickhouse.CompressionLZ4, + }, + DialTimeout: options.DialTimeout, + MaxOpenConns: options.MaxOpenConns, + MaxIdleConns: options.MaxIdleConns, + ConnOpenStrategy: strategy, + }) + if err != nil { + return nil, fmt.Errorf("clickhouse open: %w", err) + } + return &ClickHouseWriter{conn: conn}, nil +} + +func parseConnOpenStrategy(value string) (clickhouse.ConnOpenStrategy, error) { + switch strings.ToLower(strings.TrimSpace(value)) { + case "in_order": + return clickhouse.ConnOpenInOrder, nil + case "round_robin": + return clickhouse.ConnOpenRoundRobin, nil + case "random": + return clickhouse.ConnOpenRandom, nil + default: + return 0, fmt.Errorf("invalid clickhouse conn open strategy %q: must be in_order, round_robin, or random", value) + } +} + +func (w *ClickHouseWriter) WriteBatch(ctx context.Context, events []*models.ProcessedEvent) error { + batch, err := w.conn.PrepareBatch(ctx, ` + INSERT INTO events ( + event_id, site_id, visitor_id, session_id, event_type, timestamp, + url, referrer, utm_source, utm_medium, utm_campaign, + country, region, device_type, browser, os, screen_width, + properties, ip_hash, consent_given + ) + `) + if err != nil { + return fmt.Errorf("prepare batch: %w", err) + } + + for _, e := range events { + if err := batch.Append( + e.EventID, e.SiteID, e.VisitorID, e.SessionID, string(e.EventType), + e.Timestamp, + e.URL, e.Referrer, e.UTMSource, e.UTMMedium, e.UTMCampaign, + e.Country, e.Region, e.DeviceType, e.Browser, e.OS, e.ScreenWidth, + string(e.Properties), e.IPHash, e.ConsentGiven, + ); err != nil { + return fmt.Errorf("append to batch: %w", err) + } + } + + if err := batch.Send(); err != nil { + return fmt.Errorf("send batch: %w", err) + } + return nil +} + +func (w *ClickHouseWriter) Close() error { + return w.conn.Close() +} diff --git a/internal/storage/clickhouse_test.go b/internal/storage/clickhouse_test.go new file mode 100644 index 0000000..18b15e5 --- /dev/null +++ b/internal/storage/clickhouse_test.go @@ -0,0 +1,57 @@ +package storage + +import ( + "testing" + + "github.com/ClickHouse/clickhouse-go/v2" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestParseConnOpenStrategy(t *testing.T) { + strategy, err := parseConnOpenStrategy("in_order") + require.NoError(t, err) + assert.Equal(t, clickhouse.ConnOpenInOrder, strategy) + + strategy, err = parseConnOpenStrategy("round_robin") + require.NoError(t, err) + assert.Equal(t, clickhouse.ConnOpenRoundRobin, strategy) + + strategy, err = parseConnOpenStrategy("random") + require.NoError(t, err) + assert.Equal(t, clickhouse.ConnOpenRandom, strategy) +} + +func TestParseConnOpenStrategyRejectsInvalidValue(t *testing.T) { + strategy, err := parseConnOpenStrategy("weighted") + require.Error(t, err) + assert.Equal(t, clickhouse.ConnOpenStrategy(0), strategy) + assert.ErrorContains(t, err, "must be in_order, round_robin, or random") +} + +func TestNewClickHouseWriterRejectsInvalidOptions(t *testing.T) { + writer, err := NewClickHouseWriter(Options{ + Database: "default", + Username: "default", + Password: "change-me", + MaxOpenConns: 10, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + require.Error(t, err) + assert.Nil(t, writer) + assert.ErrorContains(t, err, "clickhouse addrs must not be empty") + + writer, err = NewClickHouseWriter(Options{ + Addrs: []string{"clickhouse:9000"}, + Database: "default", + Username: "default", + Password: "change-me", + MaxOpenConns: 0, + MaxIdleConns: 5, + ConnOpenStrategy: "in_order", + }) + require.Error(t, err) + assert.Nil(t, writer) + assert.ErrorContains(t, err, "clickhouse max open conns must be > 0") +} diff --git a/mise.toml b/mise.toml index fc229eb..0a9806e 100644 --- a/mise.toml +++ b/mise.toml @@ -21,5 +21,8 @@ run = "./bin/processor" [tasks.test] run = "go test ./..." +[tasks.lint] +run = "golangci-lint run" + [tasks.clean] run = "rm -rf tmp bin"