Repository navigation
feat(metrics): add config, subscription check and health probes to metrics ingester - #1003
Merged
AntonTyb merged 17 commits intoOct 23, 2026
Merged
Conversation
This was referenced Sep 28, 2026
There was a problem hiding this comment.
Code Review
This pull request introduces a health server with liveness and readiness probes for the metrics ingester, refactors configuration loading, and adds consistency tests for the ClickHouse schema alongside Pub/Sub pipeline integration tests. Feedback suggests adding a timeout context to the subscription existence check at startup to prevent blocking indefinitely, and checking for nil messages in the test helper awaitAllAcked to avoid potential nil pointer dereference panics.
janzak-dev
previously approved these changes
Oct 2, 2026
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
…trics ingester Make the metrics ingester deployable: - Settings are resolved from the flag, then the environment variable (PORT, GOOGLE_CLOUD_PROJECT, PUBSUB_SUBSCRIPTION_ID), then the default, and are validated at startup. - Startup fails fast if the subscription doesn't exist. The client leases at most twice the batch size. - /healthz reports whether the receive loop is running. /readyz also requires ClickHouse to answer a ping. Add a Python schema test that fails if xmc_metrics_pubsub_mv or its queue table drift from the Kafka equivalents. Add a manual end-to-end test that publishes to the Pub/Sub fake and verifies the rows landing in a real ClickHouse. TAG=agy CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4
Replace the duplicate Flags struct field listing with `type Flags Config` so raw CLI flag bindings and validated runtime configuration stay automatically in sync when options are added. Comment link: #1003 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ie4091718d863bc3fb615e38fb18c98c4ef80b8f0
AntonTyb
force-pushed
the
feat/pubsub-ingester-config
branch
from
October 22, 2026 14:26
20b7bd4 to
51df0ce
Compare
…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
janzak-dev
previously approved these changes
Oct 23, 2026
…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
…dpoints Update RegisterFlags docstring in config.go to state that string flags default to empty so LoadConfig falls back to environment variables when flags are omitted (matching firstNonEmpty semantics), and extend cmd/metrics_ingester/README.md for -port, /healthz, /readyz, pubsub.subscriptions.get, and schema_test. Comment link: #1003 (comment) TAG=agy CONV=0fafd02b-f3b1-4563-a73d-49ec0845afbd Change-Id: Ic3c615861d5a7d51ef8bdbc64f8c25876214dd20
There was a problem hiding this comment.
🤖 Incremental PR Approval
PR author is a code owner and changes since last approval (12 lines) are within the 50-line grace delta.
Preserving prior approval by @janzak-dev on commit 6ef9b77.
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
AntonTyb
added a commit
that referenced
this pull request
Oct 23, 2026
…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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Makes the metrics ingester deployable. Settings follow flag → env → default precedence and are validated. Startup fails if the subscription is missing, the client has a lease limit, and the service exposes
/healthzand/readyz. It also adds a schema drift test and a manual end-to-end test against real ClickHouse.Motivation & Context
The core PR's entrypoint only works when run by hand. To run under Kubernetes, the ingester needs to:
The drift test protects the key invariant from #1002: the Pub/Sub view has to stay identical to the Kafka view.
Key Changes
config.go:RegisterFlagsandLoadConfig.PORT,GOOGLE_CLOUD_PROJECT,PUBSUB_SUBSCRIPTION_ID), then defaults.subscription.go:OpenSubscriptionchecksExists, so it needspubsub.subscriptions.get. It also setsMaxOutstandingMessages = 2 × batch_size.health.go:/healthzreports whether the receive loop is running./readyzalso requires a ClickHousePing.Ingester.IsRunningis added to back the probes.main.gois switched to the above and runs the health server.deploy/clickhouse/schema_test.py(py_test): fails ifxmc_metrics_pubsub_mv's SELECT (ignoring the source table) orxmc_metrics_pubsub_queue's columns drift from the Kafka equivalents.test/metrics/pubsub_pipeline_test.go: publishes topstestand runs the real writer against ClickHouse in containers. It checks that:replica_iddefaults to'0'xid = 0records are filtered outIt's added to the existing
clickhouse_pipeline_testtarget (tags: local, manual, requires-network).Assumptions
Existscheck is worth the extrapubsub.subscriptions.getpermission — it turns a silent misconfiguration into an immediate crash loop, matching the gateway's topic check — if the deployment grants onlyroles/pubsub.subscriber, startup fails untilroles/pubsub.viewer(or a custom role) is added./readyzgates on ClickHouse, while/healthzdoesn't — a ClickHouse outage should take the pod out of rotation without restarting it — the ingester serves no traffic, so readiness only matters for rollouts and dashboards.2 × batch_sizeis a reasonable lease limit — it lets one batch fill while the previous one is being written — heavy traffic might need tuning.Verification & Testing
go build,go vet, andgo test -race -count=30 -coveron the package.python3 deploy/clickhouse/schema_test.pypasses, as does./.githooks/pre-commit.-short. It has not yet been run against real containers because the authoring environment has no Docker. Run it withbazel test //test/metrics:clickhouse_pipeline_test --test_filter=TestPubSubPipelineIngestion.Related Issues / Bugs
N/A
🤖 Generated with Claude Code
TAG=agy
CONV=c72e85d4-da3a-4185-b361-f01b12d0e3b4