feat(consumoor): add core infrastructure for Kafka-to-ClickHouse pipeline (#755)

* feat(consumoor): add core infrastructure for Kafka-to-ClickHouse pipeline

Introduce the consumoor service — a Kafka consumer that decodes
DecoratedEvent protobufs and writes flat rows to ClickHouse using typed
columnar batches (ch-go), replacing Vector's VRL transforms.

This commit contains all design-critical code:

- CLI entry point and service orchestrator
- Kafka ingestion via Benthos kafka_franz with batch/message delivery
- ClickHouse writer with per-table batched inserts, retries, and pooling
- Event-to-route dispatch engine with conditional routing support
- Core interfaces: Route, ColumnarBatch, Writer, WriteErrorClassifier
- CommonMetadata extraction from DecoratedEvent proto fields
- Conversion helpers for IP/UUID/UInt128/UInt256/decimal types
- Vector-compatible SeaHash64 implementation
- Route registry (empty — domain packages register in later commits)
- chgo-rowgen code generator for typed batch structs
- Prometheus telemetry metrics
- Example configuration

The service compiles and starts but processes 0 events until table
routes are registered in subsequent commits.

Removes unused BatchBytes config field (writer does not use it).

* refactor(consumoor): split large files and improve code organization

Split writer.go (1017 lines) into 6 focused files and benthos.go
(909 lines) into 4 focused files. Merge catalog package into flattener,
split Snapshot() into separate Snapshotter interface, remove redundant
resolvedChGoConfig defaults, fix silent flush error discard on shutdown,
fix buffer gauge leak in shutdown drain, and harden Benthos error check.

Author: samcm

.gitattributes

@@ -0,0 +1 @@
+*.gen.go linguist-generated=true

cmd/consumoor.go

@@ -0,0 +1,142 @@
+//nolint:dupl // disable duplicate code warning for cmds
+package cmd
+
+import (
+	"os"
+
+	"github.com/creasty/defaults"
+	"github.com/ethpandaops/xatu/pkg/consumoor"
+	"github.com/spf13/cobra"
+	yaml "gopkg.in/yaml.v3"
+)
+
+var (
+	consumoorCfgFile string
+)
+
+// ConsumoorOverride defines a CLI/env override for consumoor config.
+type ConsumoorOverride struct {
+	FlagHelper func(cmd *cobra.Command)
+	Setter     func(cmd *cobra.Command, overrides *consumoor.Override) error
+}
+
+// ConsumoorOverrideConfig defines how a single override is wired.
+type ConsumoorOverrideConfig struct {
+	FlagName     string
+	EnvName      string
+	Description  string
+	OverrideFunc func(val string, overrides *consumoor.Override)
+}
+
+func createConsumoorOverride(config ConsumoorOverrideConfig) ConsumoorOverride {
+	return ConsumoorOverride{
+		FlagHelper: func(cmd *cobra.Command) {
+			cmd.Flags().String(config.FlagName, "", config.Description+` (env: `+config.EnvName+`)`)
+		},
+		Setter: func(cmd *cobra.Command, overrides *consumoor.Override) error {
+			val := ""
+
+			if cmd.Flags().Changed(config.FlagName) {
+				val = cmd.Flags().Lookup(config.FlagName).Value.String()
+			}
+
+			if os.Getenv(config.EnvName) != "" {
+				val = os.Getenv(config.EnvName)
+			}
+
+			if val == "" {
+				return nil
+			}
+
+			config.OverrideFunc(val, overrides)
+
+			return nil
+		},
+	}
+}
+
+// ConsumoorOverrides is the list of CLI/env overrides for consumoor.
+var ConsumoorOverrides = []ConsumoorOverride{
+	createConsumoorOverride(ConsumoorOverrideConfig{
+		FlagName:    "metrics-addr",
+		EnvName:     "METRICS_ADDR",
+		Description: "sets the metrics address",
+		OverrideFunc: func(val string, overrides *consumoor.Override) {
+			overrides.MetricsAddr.Enabled = true
+			overrides.MetricsAddr.Value = val
+		},
+	}),
+}
+
+// consumoorCmd represents the consumoor command.
+var consumoorCmd = &cobra.Command{
+	Use:   "consumoor",
+	Short: "Runs Xatu in consumoor mode.",
+	Long: `Runs Xatu in consumoor mode, which consumes events from Kafka
+	and writes them directly to ClickHouse, replacing Vector.`,
+	Run: func(cmd *cobra.Command, args []string) {
+		initCommon()
+
+		config, err := loadConsumoorConfigFromFile(consumoorCfgFile)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		log = getLogger(config.LoggingLevel, "")
+
+		log.WithField("location", consumoorCfgFile).Info("Loaded config")
+
+		overrides := &consumoor.Override{}
+		for _, override := range ConsumoorOverrides {
+			if errr := override.Setter(cmd, overrides); errr != nil {
+				log.Fatal(errr)
+			}
+		}
+
+		c, err := consumoor.New(cmd.Context(), log, config, overrides)
+		if err != nil {
+			log.Fatal(err)
+		}
+
+		if err := c.Start(cmd.Context()); err != nil {
+			log.Fatal(err)
+		}
+
+		log.Info("Xatu consumoor exited - cya!")
+	},
+}
+
+func init() {
+	rootCmd.AddCommand(consumoorCmd)
+
+	consumoorCmd.Flags().StringVar(&consumoorCfgFile, "config", "consumoor.yaml", "config file (default is consumoor.yaml)")
+
+	for _, override := range ConsumoorOverrides {
+		override.FlagHelper(consumoorCmd)
+	}
+}
+
+func loadConsumoorConfigFromFile(file string) (*consumoor.Config, error) {
+	if file == "" {
+		file = "consumoor.yaml"
+	}
+
+	config := &consumoor.Config{}
+
+	if err := defaults.Set(config); err != nil {
+		return nil, err
+	}
+
+	yamlFile, err := os.ReadFile(file)
+	if err != nil {
+		return nil, err
+	}
+
+	type plain consumoor.Config
+
+	if err := yaml.Unmarshal(yamlFile, (*plain)(config)); err != nil {
+		return nil, err
+	}
+
+	return config, nil
+}

example_consumoor.yaml

@@ -0,0 +1,65 @@
+logging: "info"
+metricsAddr: ":9090"
+# pprofAddr: ":6060"
+
+kafka:
+  brokers:
+    - localhost:9092
+  topics:
+    - "^general-.+"
+  consumerGroup: xatu-consumoor-general
+  encoding: json  # switch to "protobuf" after Vector removal
+  # offsetDefault: oldest
+  # fetchMinBytes: 1
+  # fetchWaitMaxMs: 500
+  # maxPartitionFetchBytes: 10485760
+  # sessionTimeoutMs: 30000
+  # heartbeatIntervalMs: 3000
+  commitInterval: 5s      # Kafka offset commit interval
+  # deliveryMode: batch   # batch (faster) or message (safer, per-message flush)
+  # rejectedTopic: xatu-consumoor-rejected
+  # tls: false
+  # sasl:
+  #   mechanism: PLAIN
+  #   user: "username"
+  #   password: "password"
+
+clickhouse:
+  dsn: "clickhouse://localhost:9000/default"
+  # chgo:
+  #   queryTimeout: 30s
+  #   maxRetries: 3
+  #   retryBaseDelay: 100ms
+  #   retryMaxDelay: 2s
+  #   maxConns: 8
+  #   minConns: 1
+  #   connMaxLifetime: 1h
+  #   connMaxIdleTime: 10m
+  #   healthCheckPeriod: 30s
+  #   poolMetricsInterval: 15s
+  defaults:
+    batchSize: 200000
+    flushInterval: 1s
+    bufferSize: 200000
+  tables:
+    # Canonical tables default to insertSettings.insert_quorum: auto (majority).
+    # Set insertSettings.insert_quorum explicitly to override that behavior.
+    beacon_api_eth_v1_beacon_committee:
+      batchSize: 1000000
+      bufferSize: 1000000
+      # insertSettings:
+      #   insert_quorum: 2
+      #   insert_quorum_timeout: 60000
+    beacon_api_eth_v1_events_attestation:
+      batchSize: 1000000
+      flushInterval: 5s
+      bufferSize: 1000000
+    # canonical_beacon_block:
+    #   insertSettings:
+    #     insert_quorum: 3
+    #     insert_quorum_timeout: 60000
+
+# Optional: disable specific events (replaces Vector blackhole sinks)
+# disabledEvents:
+#   - BEACON_API_ETH_V1_DEBUG_FORK_CHOICE
+#   - BEACON_API_ETH_V1_DEBUG_FORK_CHOICE_V2