feat(metrics): add minimal Pub/Sub metrics ingester for ClickHouse by AntonTyb · Pull Request #1002 · google/xmc

AntonTyb · 2026-09-28T13:34:51Z

Stack 1/3: this PR (core) ← config (#1003) ← hardening (#1004). Review and merge bottom-up.

Summary

Adds cmd/metrics_ingester, a Pub/Sub subscriber that delivers the Ingestion Gateway's -message_bus=pubsub metrics into ClickHouse, along with the ClickHouse objects it writes to. Until now the Pub/Sub producer published correctly, but nothing consumed the topic.

Motivation & Context

In Redpanda mode ClickHouse reads the topic itself through its ENGINE = Kafka table. ClickHouse has no Pub/Sub table engine, so the Pub/Sub bus needs a consumer service.

Schema: the ingester inserts into a new ENGINE = Null table, xmc_metrics_pubsub_queue, which has the same columns as xmc_metrics_queue. A second materialized view, xmc_metrics_pubsub_mv, runs the same SELECT as xmc_metrics_mv into xmc_metrics. Both buses therefore filter invalid records, fill defaults and sanitize timestamps identically, and that logic stays in SQL instead of being duplicated in Go. When the Pub/Sub path is unused, both objects are inert.
Delivery is at least once. A batch is acked only after its INSERT succeeds and is nacked for redelivery otherwise. That's the same guarantee the Kafka engine gives the Redpanda path.
Malformed payloads are acked and dropped with a warning, mirroring kafka_skip_broken_messages. Nacking them would redeliver them forever.
Batching: a batch is flushed when it reaches -batch_size messages (default 1000) or when -batch_interval passes (default 1s, matching kafka_flush_interval_ms), whichever comes first.
Shutdown in this PR: pending messages are nacked so that another replica gets them. feat(metrics): flush pending batch on metrics ingester shutdown #1004 changes this to flush them instead.

Key Changes

deploy/clickhouse/schema.sql: sections 4 and 5, the Null queue table and the Pub/Sub materialized view.
cmd/metrics_ingester/metricsingester/writer.go: a ClickHouse batch writer (PrepareBatch, Append, Send, and Abort on failure).
cmd/metrics_ingester/metricsingester/ingester.go: the receive → batch → write → ack/nack loop. When the receive callback's context is done it nacks the message rather than blocking, because Pub/Sub's Receive doesn't return until every message it delivered has been settled.
cmd/metrics_ingester/main.go: a minimal entrypoint (-project_id, -subscription_id, -batch_size, -batch_interval), plus BUILD.bazel with a binary target and oci_image_for_service.

PR Size Exception

The sizing tool reports 504 production lines (limit 400) and 1,050 total (limit 1,000). This is the smallest slice that can run on its own:

About 150 of the production lines are BUILD files and schema.sql, which the tool counts as production code.
Every Go file carries a 14-line license header.
Splitting further would mean a PR containing only the schema and the writer, with nothing calling the writer, which violates the dead-code rule.

The split was reviewed with the PR split-verification process. The rest of the feature is in #1003 (265 production / 818 total) and #1004 (30 / 84).

Assumptions

Ordering keys don't need to be honoured on the consumer side — xmc_metrics is sorted by its ORDER BY key, so arrival order never shows up in queries — if some downstream reader depended on arrival order, the subscription would need message ordering enabled.
Duplicate rows after a partial failure or crash are acceptable — at-least-once delivery was the agreed choice, and the Redpanda path has the same property — if exactly-once is needed later, it would take a deduplicating table engine or insert tokens.
Deployment is out of scope for this stack — Helm wiring comes in a follow-up, and Terraform/CTK (topic, subscription, IAM) waits for feat(components): migrate ingestion-gateway to 1P native component blueprint #969 to settle — until then the ingester can only be run by hand.

Verification & Testing

Commands Executed: Bazel isn't available in the authoring environment, so the Go code was built in a scratch module with stand-ins for internal imports: go build, go vet, and go test -race -count=30 -cover on the package. ./.githooks/pre-commit (buildifier, gofmt, golangci-lint, the determinism check, addlicense, ruff) passes.
Coverage & Test Status: 100% statement coverage. All tests pass 30 consecutive runs with the race detector. The tests use the in-memory Pub/Sub fake (pstest), so they need no network or credentials. CI's bazel test //cmd/metrics_ingester/... will be the first Bazel run.

Related Issues / Bugs

N/A

🤖 Generated with Claude Code

TAG=agy
CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4

gemini-code-assist

< 10BC8 div class="comment-body markdown-body js-comment-body soft-wrap user-select-contain d-block css-overflow-wrap-anywhere">

Code Review

This pull request introduces the metrics_ingester service, which consumes serialized metric messages from a Google Cloud Pub/Sub subscription, batches them, and writes them to ClickHouse. It includes the main entrypoint, the core batching ingester logic, a ClickHouse writer, corresponding unit tests, and the necessary ClickHouse schema updates (a Null-engine landing table and a materialized view). The review feedback highlights three key areas for improvement: using a deferred function in Run to prevent a potential goroutine leak if Receive exits prematurely, resetting the ticker when a full batch is flushed to avoid frequent small writes to ClickHouse, and adding an early return in flush to avoid unnecessary allocations when there are no pending messages.

VLukiantsev · 2026-10-22T08:11:11Z

Heads-up for the whole stack (#1002 → #1003 → #1004): this branch is based on ff9e76bf7, from before #1124 and #1199. All three PRs still use //:compat.bzl, //internal/compat/google and the old xmanager_cloud/logging/logging importpath, and none of those exist on main anymore.

The PR checks are green only because they ran against that old base. main itself is safe: the merge queue re-runs the required checks on main + PR, and there both the legacy-import ratchet in Code Quality & Linting and the Bazel load in Go & Backend Unit Tests would fail, so the queue would kick these PRs out.

Please rebase the stack and re-run bazel test //cmd/metrics_ingester/... //test/metrics/... and tools/lint/check_legacy_imports.py before queueing. The inline comment on metricsingester/BUILD.bazel L15 has the details.

Run now closes the message channel and waits for the batch loop in a defer, so the loop is always shut down however Receive exits. Resetting the batch ticker after a size-triggered flush means a partial batch waits a full interval from the last insert. Before, a tick shortly after a full flush could insert a tiny batch, and ClickHouse handles frequent small inserts poorly. Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4

…metry Add bounded exponential backoff (250ms to 30s) in batchLoop after failed ClickHouse batch writes so an outage does not tight-loop Receive -> Nack at line rate while remaining interruptible on shutdown cancellation. Detach regular batch writes from Run's signal context using context.WithoutCancel bounded by a 10s per-write timeout so SIGTERM does not abort an in-flight batch.Send() mid-flight, and check ctx.Err() before flushing when select chooses msgCh or ticker concurrently with cancellation. Track malformed payload drops via an atomic counter (DroppedMessages) logged alongside each warning, clarify the decodeBatch comment vs kafka_skip_broken_messages, and update -batch_interval help/comments to reflect target wait time behind slow inserts. Also sync BUILD.bazel and main.go imports with main's rules_go and internal/logging conventions. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ie330e3d44122d4f733391b19593dd7079f8e4c42

…acts Defer //test/metrics:__pkg__ visibility on metricsingester to #1003 where pubsub_pipeline_test.go first consumes the package, keeping #1002 free of forward-looking grants. Clarify at-least-once ClickHouse batch delivery semantics in writer.go and ingester.go (post-commit Send network errors and multi-partition toYYYYMM writes), and document in config.go why metrics_ingester runs as a separate service rather than inside ingestion_gateway. Guard gate channel closures with sync.OnceFunc + t.Cleanup so awaitCalls timeouts fail fast instead of hanging on loopDone, and replace the bare-struct channel drain test with an end-to-end pstest test asserting Nack settlement (AckDeadline=0, Acks=0) and redelivery after cancellation. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Id2d6c9a451db35e55822ef2311f41673c5a7f3c4

…s visibility Merge updated feat/pubsub-ingester-core (42d0d4e) and declare //test/metrics:__pkg__ visibility on //cmd/metrics_ingester/metricsingester here alongside test/metrics/pubsub_pipeline_test.go. Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ie2e071863f92376d2372c006c31649fe63dcaed0

… separation Merge updated feat/pubsub-ingester-config and add a brief rationale in cmd/metrics_ingester/README.md explaining why metrics_ingester runs as a separate service from ingestion_gateway (ClickHouse credential/egress isolation and independent batch-insert replica scaling). Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: I751d6c78cf1088313833074a4f95fcd0a5c7df2e

AntonTyb · 2026-10-23T08:09:53Z

Thanks for catching this! The entire stack (#1002 -> #1003 -> #1004) has now been rebased onto main, migrated from //:compat.bzl, //internal/compat/google (google.Init()), and xmanager_cloud/logging/logging to @io_bazel_rules_go//go:def.bzl and //internal/logging, and verified with tools/lint/check_legacy_imports.py (total: 0) and bazel test --@io_bazel_rules_go//go/config:race //cmd/metrics_ingester/.... All 6 inline review comments on #1002 have also been addressed in follow-up commit 42d0d4e8 and merged forward through #1003 and #1004.

VLukiantsev

LGTM. Thanks for the rebase and the thorough fixes. I checked the head against current main: check_legacy_imports.py is clean, and bazel test //cmd/metrics_ingester/... passes.

Approving with three small follow-ups, all in inline threads:

t.Cleanup ordering in TestIngesterNacksInsteadOfBlockingWhenCallbackContextIsDone. My earlier suggestion was wrong; the correct fix is defer closeGate() inside the script.
DroppedMessages() has only a test caller (dead code).
Move the READMEs from #1004 into this PR so the gateway README on main doesn't go stale.

…tter Add cmd/metrics_ingester/README.md and update cmd/ingestion_gateway/README.md in #1002 so main's gateway documentation and manual setup instructions stay aligned as soon as the ingester binary lands. Remove the unused exported DroppedMessages() getter in favor of the dropped_total log field and direct in-package atomic reads in tests, and defer closeGate() inside scriptedSource so an awaitCalls t.Fatalf releases batchLoop before Run's deferred <-loopDone waits. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: I5dc801d64761a983a289d97867ecd5a94e4da6d7

github-actions

🤖 Incremental PR Approval

PR author is a code owner and changes since last approval (47 lines, excluding 5 test file lines) are within the 50-line grace delta.

Preserving prior approval by @VLukiantsev on commit 42d0d4e.

…tal docs Merge updated feat/pubsub-ingester-config (caa080a) into #1004, retaining #1004's shutdown-flush README details and referencing the dropped_total log field in place of the removed DroppedMessages getter. Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: I3ea0c7f2900179baeb5927323bb3b619331f87c5

Run now closes the message channel and waits for the batch loop in a defer, so the loop is always shut down however Receive exits. Resetting the batch ticker after a size-triggere 10BC8 d flush means a partial batch waits a full interval from the last insert. Before, a tick shortly after a full flush could insert a tiny batch, and ClickHouse handles frequent small inserts poorly. Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4

…metry Add bounded exponential backoff (250ms to 30s) in batchLoop after failed ClickHouse batch writes so an outage does not tight-loop Receive -> Nack at line rate while remaining interruptible on shutdown cancellation. Detach regular batch writes from Run's signal context using context.WithoutCancel bounded by a 10s per-write timeout so SIGTERM does not abort an in-flight batch.Send() mid-flight, and check ctx.Err() before flushing when select chooses msgCh or ticker concurrently with cancellation. Track malformed payload drops via an atomic counter (DroppedMessages) logged alongside each warning, clarify the decodeBatch comment vs kafka_skip_broken_messages, and update -batch_interval help/comments to reflect target wait time behind slow inserts. Also sync BUILD.bazel and main.go imports with main's rules_go and internal/logging conventions. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ie330e3d44122d4f733391b19593dd7079f8e4c42

…acts Defer //test/metrics:__pkg__ visibility on metricsingester to #1003 where pubsub_pipeline_test.go first consumes the package, keeping #1002 free of forward-looking grants. Clarify at-least-once ClickHouse batch delivery semantics in writer.go and ingester.go (post-commit Send network errors and multi-partition toYYYYMM writes), and document in config.go why metrics_ingester runs as a separate service rather than inside ingestion_gateway. Guard gate channel closures with sync.OnceFunc + t.Cleanup so awaitCalls timeouts fail fast instead of hanging on loopDone, and replace the bare-struct channel drain test with an end-to-end pstest test asserting Nack settlement (AckDeadline=0, Acks=0) and redelivery after cancellation. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Id2d6c9a451db35e55822ef2311f41673c5a7f3c4

…tter Add cmd/metrics_ingester/README.md and update cmd/ingestion_gateway/README.md in #1002 so main's gateway documentation and manual setup instructions stay aligned as soon as the ingester binary lands. Remove the unused exported DroppedMessages() getter in favor of the dropped_total log field and direct in-package atomic reads in tests, and defer closeGate() inside scriptedSource so an awaitCalls t.Fatalf releases batchLoop before Run's deferred <-loopDone waits. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: I5dc801d64761a983a289d97867ecd5a94e4da6d7

github-actions

🤖 Incremental PR Approval

PR author is a code owner and changes since last approval (47 lines, excluding 5 test file lines) are within the 50-line grace delta.

Preserving prior approval by @VLukiantsev on commit 42d0d4e.

ClickHouse has no Pub/Sub table engine, so with -message_bus=pubsub the gateway's published metrics never reached ClickHouse. Add cmd/metrics_ingester, which pulls from a subscription on the metrics topic, batches messages and inserts them into a new Null-engine table xmc_metrics_pubsub_queue. A second materialized view, xmc_metrics_pubsub_mv, runs the same SELECT as the Kafka view, so both buses apply the same filtering, defaults and timestamp sanitization. A batch is flushed when it fills or when the batch interval passes. Delivery is at least once: a batch is acked only after its insert succeeds and nacked otherwise. Malformed payloads are acked and dropped, mirroring kafka_skip_broken_messages. On shutdown, pending messages are nacked for redelivery. This is the base of a stack. Config precedence, the startup subscription check, health probes and the shutdown flush follow in later PRs. TAG=agy CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4

Run now closes the message channel and waits for the batch loop in a defer, so the loop is always shut down however Receive exits. Resetting the batch ticker after a size-triggered flush means a partial batch waits a full interval from the last insert. Before, a tick shortly after a full flush could insert a tiny batch, and ClickHouse handles frequent small inserts poorly. Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4

…metry Add bounded exponential backoff (250ms to 30s) in batchLoop after failed ClickHouse batch writes so an outage does not tight-loop Receive -> Nack at line rate while remaining interruptible on shutdown cancellation. Detach regular batch writes from Run's signal context using context.WithoutCancel bounded by a 10s per-write timeout so SIGTERM does not abort an in-flight batch.Send() mid-flight, and check ctx.Err() before flushing when select chooses msgCh or ticker concurrently with cancellation. Track malformed payload drops via an atomic counter (DroppedMessages) logged alongside each warning, clarify the decodeBatch comment vs kafka_skip_broken_messages, and update -batch_interval help/comments to reflect target wait time behind slow inserts. Also sync BUILD.bazel and main.go imports with main's rules_go and internal/logging conventions. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ie330e3d44122d4f733391b19593dd7079f8e4c42

…acts Defer //test/metrics:__pkg__ visibility on metricsingester to #1003 where pubsub_pipeline_test.go first consumes the package, keeping #1002 free of forward-looking grants. Clarify at-least-once ClickHouse batch delivery semantics in writer.go and ingester.go (post-commit Send network errors and multi-partition toYYYYMM writes), and document in config.go why metrics_ingester runs as a separate service rather than inside ingestion_gateway. Guard gate channel closures with sync.OnceFunc + t.Cleanup so awaitCalls timeouts fail fast instead of hanging on loopDone, and replace the bare-struct channel drain test with an end-to-end pstest test asserting Nack settlement (AckDeadline=0, Acks=0) and redelivery after cancellation. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Id2d6c9a451db35e55822ef2311f41673c5a7f3c4

…tter Add cmd/metrics_ingester/README.md and update cmd/ingestion_gateway/README.md in #1002 so main's gateway documentation and manual setup instructions stay aligned as soon as the ingester binary lands. Remove the unused exported DroppedMessages() getter in favor of the dropped_total log field and direct in-package atomic reads in tests, and defer closeGate() inside scriptedSource so an awaitCalls t.Fatalf releases batchLoop before Run's deferred <-loopDone waits. Comment link: #1002 (comment) Comment link: #1002 (comment) Comment link: #1002 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: I5dc801d64761a983a289d97867ecd5a94e4da6d7

github-actions

🤖 Incremental PR Approval

PR author is a code owner and changes since last approval (47 lines, excluding 5 test file lines) are within the 50-line grace delta.

Preserving prior approval by @VLukiantsev on commit 42d0d4e.

This was referenced Sep 28, 2026

feat(metrics): add config, subscription check and health probes to metrics ingester #1003

Merged

feat(metrics): flush pending batch on metrics ingester shutdown #1004

Merged

gemini-code-assist Bot reviewed Sep 28, 2026

View reviewed changes

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

AntonTyb added this pull request to stack #1005 September 28, 2026 13:43

AntonTyb marked this pull request as ready for review September 29, 2026 11:32

github-actions Bot requested a review from pachopepe September 29, 2026 11:33

pachopepe reviewed Sep 30, 2026

View reviewed changes

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

Comment thread cmd/metrics_ingester/metricsingester/ingester.go Resolved

Comment thread cmd/metrics_ingester/main.go Resolved

AntonTyb requested a review from VLukiantsev October 22, 2026 06:58