Kafka Consumer Worker

.github/workflows/ci.yaml

@@ -29,8 +29,11 @@ jobs:
       - name: Prepare Nix shell
         run: nix develop --impure .#ci
 
-      - name: Build
-        run: nix develop --impure .#ci -c make build
+      - name: Build Server
+        run: nix develop --impure .#ci -c make build-server
+
+      - name: Build Sink Worker
+        run: nix develop --impure .#ci -c make build-sink-worker
 
   test:
     name: Test
@@ -159,8 +162,12 @@ jobs:
         run: docker compose -f docker-compose.yaml -f docker-compose.ci.yaml up -d
         working-directory: quickstart
 
-      - name: Wait for OpenMeter to become ready
-        run: curl --retry 10 --retry-max-time 120 --retry-all-errors http://localhost:8888/version
+      - name: Wait for worker to become ready
+        run: curl --retry 10 --retry-max-time 120 --retry-all-errors http://localhost:10000/healthz
+
+      # This is probably overkill but it can happen that OpenMeter creates namespaces slower
+      - name: Wait to pickup namespace
+        run: sleep 5
 
       - name: Ingest sample data
         run: ./ingest.sh

Dockerfile

@@ -27,14 +27,20 @@ ARG VERSION
 
 # See https://github.com/confluentinc/confluent-kafka-go#librdkafka
 # See https://github.com/confluentinc/confluent-kafka-go#static-builds-on-linux
-RUN go build -ldflags "-linkmode external -extldflags \"-static\" -X main.version=${VERSION}" -tags musl -o /usr/local/bin/openmeter .
+# Build server binary (default)
+RUN go build -ldflags "-linkmode external -extldflags \"-static\" -X main.version=${VERSION}" -tags musl -o /usr/local/bin/openmeter ./cmd/server
 RUN xx-verify /usr/local/bin/openmeter
 
+# Build sink-worker binary
+RUN go build -ldflags "-linkmode external -extldflags \"-static\" -X main.version=${VERSION}" -tags musl -o /usr/local/bin/openmeter-sink-worker ./cmd/sink-worker
+RUN xx-verify /usr/local/bin/openmeter-sink-worker
+
 FROM gcr.io/distroless/base-debian11:latest@sha256:b31a6e02605827e77b7ebb82a0ac9669ec51091edd62c2c076175e05556f4ab9 AS distroless
 
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 
 COPY --from=builder /usr/local/bin/openmeter /usr/local/bin/
+COPY --from=builder /usr/local/bin/openmeter-sink-worker /usr/local/bin/
 COPY --from=builder /usr/local/src/openmeter/go.* /usr/local/src/openmeter/
 
 CMD openmeter
@@ -44,6 +50,7 @@ FROM redhat/ubi8-micro:8.8-7@sha256:6fa456671239c7ac791dac2537425d1ba5612df36355
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/
 
 COPY --from=builder /usr/local/bin/openmeter /usr/local/bin/
+COPY --from=builder /usr/local/bin/openmeter-sink-worker /usr/local/bin/
 COPY --from=builder /usr/local/src/openmeter/go.* /usr/local/src/openmeter/
 
 CMD openmeter
@@ -55,6 +62,7 @@ RUN apk add --update --no-cache ca-certificates tzdata bash
 SHELL ["/bin/bash", "-c"]
 
 COPY --from=builder /usr/local/bin/openmeter /usr/local/bin/
+COPY --from=builder /usr/local/bin/openmeter-sink-worker /usr/local/bin/
 COPY --from=builder /usr/local/src/openmeter/go.* /usr/local/src/openmeter/
 
 CMD openmeter

Makefile

@@ -15,19 +15,30 @@ generate: ## Generate code
 	$(call print-target)
 	go generate ./...
 
-.PHONY: build
-build: ## Build binary
+.PHONY: build-server
+build-server: ## Build server binary
 	$(call print-target)
-	go build -tags dynamic -o build/ .
+	go build -tags dynamic -o build/server ./cmd/server
+
+.PHONY: build-sink-worker
+build-sink-worker: ## Build binary
+	$(call print-target)
+	go build -tags dynamic -o build/sink-worker ./cmd/sink-worker
 
 config.yaml:
 	cp config.example.yaml config.yaml
 
-run: config.yaml
-run: ## Run OpenMeter
+.PHONY: server
+server: ## Run sink-worker
+	@ if [ config.yaml -ot config.example.yaml ]; then diff -u config.yaml config.example.yaml || (echo "!!! The configuration example changed. Please update your config.yaml file accordingly (or at least touch it). !!!" && false); fi
+	$(call print-target)
+	air -c ./cmd/server/.air.toml
+
+.PHONY: sink-worker
+sink-worker: ## Run sink-worker
 	@ if [ config.yaml -ot config.example.yaml ]; then diff -u config.yaml config.example.yaml || (echo "!!! The configuration example changed. Please update your config.yaml file accordingly (or at least touch it). !!!" && false); fi
 	$(call print-target)
-	air
+	air -c ./cmd/sink-worker/.air.toml
 
 .PHONY: test
 test: ## Run tests

.air.toml → cmd/server/.air.toml

@@ -3,9 +3,9 @@ testdata_dir = "testdata"
 tmp_dir = "tmp"
 
 [build]
-  args_bin = ["--config", "config.yaml"]
-  bin = "./tmp/openmeter"
-  cmd = "go build -o ./tmp/openmeter ."
+  args_bin = ["--config", "./config.yaml"]
+  bin = "./tmp/openmeter-server"
+  cmd = "go build -o ./tmp/openmeter-server ./cmd/server"
   delay = 0
   exclude_dir = ["assets", "ci", "deploy", "docs", "examples", "testdata", "quickstart", "tmp", "vendor", "api/client", "node_modules"]
   exclude_file = []

main.go → cmd/server/main.go

@@ -41,12 +41,10 @@ import (
 	"github.com/openmeterio/openmeter/internal/namespace"
 	"github.com/openmeterio/openmeter/internal/server"
 	"github.com/openmeterio/openmeter/internal/server/router"
-	"github.com/openmeterio/openmeter/internal/sink"
 	"github.com/openmeterio/openmeter/internal/streaming"
 	"github.com/openmeterio/openmeter/internal/streaming/clickhouse_connector"
 	"github.com/openmeterio/openmeter/pkg/gosundheit"
 	pkgkafka "github.com/openmeterio/openmeter/pkg/kafka"
-	"github.com/openmeterio/openmeter/pkg/kafkaconnect"
 	"github.com/openmeterio/openmeter/pkg/models"
 	"github.com/openmeterio/openmeter/pkg/slicesx"
 )
@@ -178,15 +176,6 @@ func main() {
 		return *meter
 	}))
 
-	// Initialize Kafka Connect sink
-	if conf.Sink.KafkaConnect.Enabled {
-		err := initKafkaConnect(conf.Sink.KafkaConnect, logger)
-		if err != nil {
-			logger.Error("failed to initialize kafka connect sink", "error", err)
-			os.Exit(1)
-		}
-	}
-
 	// Initialize ClickHouse Aggregation
 	clickhouseStreamingConnector, err := initClickHouseStreaming(conf, meterRepository, logger)
 	if err != nil {
@@ -359,7 +348,7 @@ func initKafkaIngest(ctx context.Context, config config.Configuration, logger *s
 	return collector, namespaceHandler, nil
 }
 
-func initClickHouseStreaming(config config.Configuration, meterRepository meter.Repository, logger *slog.Logger) (*clickhouse_connector.ClickhouseConnector, error) {
+func initClickHouseClient(config config.Configuration) (clickhouse.Conn, error) {
 	options := &clickhouse.Options{
 		Addr: []string{config.Aggregation.ClickHouse.Address},
 		Auth: clickhouse.Auth{
@@ -386,6 +375,15 @@ func initClickHouseStreaming(config config.Configuration, meterRepository meter.
 		return nil, fmt.Errorf("init clickhouse client: %w", err)
 	}
 
+	return clickHouseClient, nil
+}
+
+func initClickHouseStreaming(config config.Configuration, meterRepository meter.Repository, logger *slog.Logger) (*clickhouse_connector.ClickhouseConnector, error) {
+	clickHouseClient, err := initClickHouseClient(config)
+	if err != nil {
+		return nil, fmt.Errorf("init clickhouse client: %w", err)
+	}
+
 	streamingConnector, err := clickhouse_connector.NewClickhouseConnector(clickhouse_connector.ClickhouseConnectorConfig{
 		Logger:     logger,
 		ClickHouse: clickHouseClient,
@@ -399,32 +397,6 @@ func initClickHouseStreaming(config config.Configuration, meterRepository meter.
 	return streamingConnector, nil
 }
 
-func initKafkaConnect(config config.KafkaConnectSinkConfiguration, logger *slog.Logger) error {
-	client, err := kafkaconnect.NewClient(config.URL)
-	if err != nil {
-		return fmt.Errorf("init kafka connect client: %w", err)
-	}
-
-	kafkaConnectSink := sink.KafkaConnect{
-		Client: client,
-		Logger: logger.With(slog.String("subsystem", "sink.kafkaconnect")),
-	}
-
-	for _, connector := range config.Connectors {
-		connectorConfig, err := connector.ConnectorConfig()
-		if err != nil {
-			return fmt.Errorf("create kafka connector config %q: %w", connector.Name, err)
-		}
-
-		err = kafkaConnectSink.ConfigureConnector(context.Background(), connector.Name, connectorConfig)
-		if err != nil {
-			return fmt.Errorf("init kafka connector %q: %w", connector.Name, err)
-		}
-	}
-
-	return nil
-}
-
 func initNamespace(config config.Configuration, namespaces ...namespace.Handler) (*namespace.Manager, error) {
 	namespaceManager, err := namespace.NewManager(namespace.ManagerConfig{
 		Handlers:          namespaces,

cmd/sink-worker/.air.toml

@@ -0,0 +1,44 @@
+root = "."
+testdata_dir = "testdata"
+tmp_dir = "tmp"
+
+[build]
+  args_bin = ["--config", "./config.yaml", "--telemetry-address", ":1001"]
+  bin = "./tmp/openmeter-sink-worker"
+  cmd = "go build -o ./tmp/openmeter-sink-worker ./cmd/sink-worker"
+  delay = 0
+  exclude_dir = ["assets", "ci", "deploy", "docs", "examples", "testdata", "quickstart", "tmp", "vendor", "api/client", "node_modules"]
+  exclude_file = []
+  exclude_regex = ["_test.go"]
+  exclude_unchanged = false
+  follow_symlink = false
+  full_bin = ""
+  include_dir = []
+  include_ext = ["go", "tpl", "tmpl", "html", "yml", "yaml", "sql", "json"]
+  include_file = []
+  kill_delay = "0s"
+  log = "build-errors.log"
+  poll = false
+  poll_interval = 0
+  rerun = false
+  rerun_delay = 500
+  send_interrupt = false
+  stop_on_error = false
+
+[color]
+  app = ""
+  build = "yellow"
+  main = "magenta"
+  runner = "green"
+  watcher = "cyan"
+
+[log]
+  main_only = false
+  time = false
+
+[misc]
+  clean_on_exit = false
+
+[screen]
+  clear_on_rebuild = false
+  keep_scroll = true