chore: Configure log batcher via config (#109)

Author: alexluong

internal/config/config.go

@@ -31,34 +31,38 @@ type Config struct {
 	PortalProxyURL   string
 	Topics           []string
 
-	Redis                  *redis.RedisConfig
-	ClickHouse             *clickhouse.ClickHouseConfig
-	OpenTelemetry          *otel.OpenTelemetryConfig
-	PublishQueueConfig     *mqs.QueueConfig
-	DeliveryQueueConfig    *mqs.QueueConfig
-	LogQueueConfig         *mqs.QueueConfig
-	PublishMaxConcurrency  int
-	DeliveryMaxConcurrency int
-	LogMaxConcurrency      int
-	RetryIntervalSeconds   int
-	RetryMaxCount          int
+	Redis                           *redis.RedisConfig
+	ClickHouse                      *clickhouse.ClickHouseConfig
+	OpenTelemetry                   *otel.OpenTelemetryConfig
+	PublishQueueConfig              *mqs.QueueConfig
+	DeliveryQueueConfig             *mqs.QueueConfig
+	LogQueueConfig                  *mqs.QueueConfig
+	PublishMaxConcurrency           int
+	DeliveryMaxConcurrency          int
+	LogMaxConcurrency               int
+	RetryIntervalSeconds            int
+	RetryMaxCount                   int
+	LogBatcherDelayThresholdSeconds int
+	LogBatcherItemCountThreshold    int
 }
 
 var defaultConfig = map[string]any{
-	"PORT":                       3333,
-	"REDIS_HOST":                 "127.0.0.1",
-	"REDIS_PORT":                 6379,
-	"REDIS_PASSWORD":             "",
-	"REDIS_DATABASE":             0,
-	"DELIVERY_RABBITMQ_EXCHANGE": "outpost",
-	"DELIVERY_RABBITMQ_QUEUE":    "outpost.delivery",
-	"LOG_RABBITMQ_EXCHANGE":      "outpost_logs",
-	"LOG_RABBITMQ_QUEUE":         "outpost_logs.log",
-	"PUBLISHMQ_MAX_CONCURRENCY":  1,
-	"DELIVERYMQ_MAX_CONCURRENCY": 1,
-	"LOGMQ_MAX_CONCURRENCY":      1,
-	"RETRY_INTERVAL_SECONDS":     30,
-	"MAX_RETRY_COUNT":            10,
+	"PORT":                                3333,
+	"REDIS_HOST":                          "127.0.0.1",
+	"REDIS_PORT":                          6379,
+	"REDIS_PASSWORD":                      "",
+	"REDIS_DATABASE":                      0,
+	"DELIVERY_RABBITMQ_EXCHANGE":          "outpost",
+	"DELIVERY_RABBITMQ_QUEUE":             "outpost.delivery",
+	"LOG_RABBITMQ_EXCHANGE":               "outpost_logs",
+	"LOG_RABBITMQ_QUEUE":                  "outpost_logs.log",
+	"PUBLISHMQ_MAX_CONCURRENCY":           1,
+	"DELIVERYMQ_MAX_CONCURRENCY":          1,
+	"LOGMQ_MAX_CONCURRENCY":               1,
+	"RETRY_INTERVAL_SECONDS":              30,
+	"MAX_RETRY_COUNT":                     10,
+	"LOG_BATCHER_DELAY_THRESHOLD_SECONDS": 5,
+	"LOG_BATCHER_ITEM_COUNT_THRESHOLD":    100,
 }
 
 var (
@@ -161,16 +165,18 @@ func Parse(flags Flags) (*Config, error) {
 			Password: viper.GetString("REDIS_PASSWORD"),
 			Database: mustInt(viper, "REDIS_DATABASE"),
 		},
-		ClickHouse:             clickHouseConfig,
-		OpenTelemetry:          openTelemetry,
-		PublishQueueConfig:     publishQueueConfig,
-		DeliveryQueueConfig:    deliveryQueueConfig,
-		LogQueueConfig:         logQueueConfig,
-		PublishMaxConcurrency:  mustInt(viper, "PUBLISHMQ_MAX_CONCURRENCY"),
-		DeliveryMaxConcurrency: mustInt(viper, "DELIVERYMQ_MAX_CONCURRENCY"),
-		LogMaxConcurrency:      mustInt(viper, "LOGMQ_MAX_CONCURRENCY"),
-		RetryIntervalSeconds:   mustInt(viper, "RETRY_INTERVAL_SECONDS"),
-		RetryMaxCount:          mustInt(viper, "MAX_RETRY_COUNT"),
+		ClickHouse:                      clickHouseConfig,
+		OpenTelemetry:                   openTelemetry,
+		PublishQueueConfig:              publishQueueConfig,
+		DeliveryQueueConfig:             deliveryQueueConfig,
+		LogQueueConfig:                  logQueueConfig,
+		PublishMaxConcurrency:           mustInt(viper, "PUBLISHMQ_MAX_CONCURRENCY"),
+		DeliveryMaxConcurrency:          mustInt(viper, "DELIVERYMQ_MAX_CONCURRENCY"),
+		LogMaxConcurrency:               mustInt(viper, "LOGMQ_MAX_CONCURRENCY"),
+		RetryIntervalSeconds:            mustInt(viper, "RETRY_INTERVAL_SECONDS"),
+		RetryMaxCount:                   mustInt(viper, "MAX_RETRY_COUNT"),
+		LogBatcherDelayThresholdSeconds: mustInt(viper, "LOG_BATCHER_DELAY_THRESHOLD_SECONDS"),
+		LogBatcherItemCountThreshold:    mustInt(viper, "LOG_BATCHER_ITEM_COUNT_THRESHOLD"),
 	}
 
 	return config, nil

internal/services/log/log.go

@@ -56,7 +56,11 @@ func NewService(ctx context.Context,
 	var eventBatcher *batcher.Batcher[*models.Event]
 	var deliveryBatcher *batcher.Batcher[*models.Delivery]
 	if handler == nil {
-		batcher, err := makeBatcher(ctx, logger, models.NewLogStore(chDB))
+		batcherCfg := batcherConfig{
+			ItemCountThreshold: cfg.LogBatcherItemCountThreshold,
+			DelayThreshold:     time.Duration(cfg.LogBatcherDelayThresholdSeconds) * time.Second,
+		}
+		batcher, err := makeBatcher(ctx, logger, models.NewLogStore(chDB), batcherCfg)
 		if err != nil {
 			return nil, err
 		}
@@ -108,11 +112,16 @@ func (s *LogService) Run(ctx context.Context) error {
 	return nil
 }
 
-func makeBatcher(ctx context.Context, logger *otelzap.Logger, logStore models.LogStore) (*batcher.Batcher[*mqs.Message], error) {
+type batcherConfig struct {
+	ItemCountThreshold int
+	DelayThreshold     time.Duration
+}
+
+func makeBatcher(ctx context.Context, logger *otelzap.Logger, logStore models.LogStore, batcherCfg batcherConfig) (*batcher.Batcher[*mqs.Message], error) {
 	b, err := batcher.NewBatcher(batcher.Config[*mqs.Message]{
 		GroupCountThreshold: 2,
-		ItemCountThreshold:  100,
-		DelayThreshold:      5 * time.Second,
+		ItemCountThreshold:  batcherCfg.ItemCountThreshold,
+		DelayThreshold:      batcherCfg.DelayThreshold,
 		NumGoroutines:       1,
 		Processor: func(_ string, msgs []*mqs.Message) {
 			logger.Ctx(ctx).Info("log batcher processor", zap.Int("msgs", len(msgs)))