feat(metrics): create the metrics kafka ingest chain via migration - #89631
Open
jzhu13 wants to merge 2 commits into
Open
feat(metrics): create the metrics kafka ingest chain via migration#89631jzhu13 wants to merge 2 commits into
jzhu13 wants to merge 2 commits into
Conversation
|
Merging to
After your PR is submitted to the merge queue, this comment will be automatically updated with its status. If the PR fails, failure details will also be posted here |
Contributor
🤖 CI report🚨 Trunk lane — universal laneThis PR is assigned to the universal lane. It cannot merge in parallel with other PRs, so it can take longer to merge. Ask dev-ex if you think this is wrong.
|
| File | Size | Δ vs base |
|---|---|---|
render-query/src/render-query/render-query.js |
21.69 MiB | 🔺 +1.8 KiB (+0.0%) |
Posted automatically by build-bundle-size-report · uncompressed bytes from dist-report
✅ Eager graph — within budget
How much code each root ships on the eager path — downloaded and parsed before the surface is interactive. Measured from the esbuild output chunks (post-tree-shake, static imports only); lazy import() / React.lazy chunks are not counted.
| Root | Eager (shipped) | Δ vs base | Budget |
|---|---|---|---|
entry (logged-out pages, app bootstrap)src/index.tsx |
1.36 MiB · 22 files | no change | ███░░░░░░░ 30.3% of 4.51 MiB |
authenticated shell (every logged-in page)src/scenes/AuthenticatedShell.tsx |
8.78 MiB · 3,238 files | 🔺 +1.9 KiB (+0.0%) | █████████░ 90.4% of 9.71 MiB |
🟢 node_modules/monaco-editor/ stays out of src/index.tsx
🟢 src/lib/components/ActivityLog/describers stays out of src/index.tsx
🟢 [object Object] stays out of src/index.tsx
🟢 [object Object] stays out of src/index.tsx
🟢 node_modules/monaco-editor/ stays out of src/scenes/AuthenticatedShell.tsx
🟢 src/lib/components/ActivityLog/describers stays out of src/scenes/AuthenticatedShell.tsx
🟢 [object Object] stays out of src/scenes/AuthenticatedShell.tsx
🟢 [object Object] stays out of src/scenes/AuthenticatedShell.tsx
Largest files eagerly shipped from src/index.tsx
| Size | File |
|---|---|
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 24.6 KiB | ../node_modules/.pnpm/buffer@6.0.3/node_modules/buffer/index.js |
| 6.3 KiB | ../node_modules/.pnpm/react@18.3.1/node_modules/react/cjs/react.production.min.js |
| 4.5 KiB | ../node_modules/.pnpm/@jspm+core@2.1.0/node_modules/@jspm/core/nodelibs/browser/process.js |
| 3.9 KiB | ../node_modules/.pnpm/scheduler@0.23.2/node_modules/scheduler/cjs/scheduler.production.min.js |
| 1.4 KiB | ../node_modules/.pnpm/base64-js@1.5.1/node_modules/base64-js/index.js |
| 1.3 KiB | src/RootErrorBoundary.tsx |
| 912 B | ../node_modules/.pnpm/ieee754@1.2.1/node_modules/ieee754/index.js |
| 789 B | src/scenes/ChunkLoadErrorBoundary.tsx |
| 762 B | src/index.tsx |
Largest files eagerly shipped from src/scenes/AuthenticatedShell.tsx
| Size | File |
|---|---|
| 306.5 KiB | ../node_modules/.pnpm/posthog-js@1.418.17/node_modules/posthog-js/dist/rrweb.js |
| 267.7 KiB | ../node_modules/.pnpm/@posthog+icons@0.38.0_react-dom@18.3.1_react@18.3.1__react@18.3.1/node_modules/@posthog/icons/dist/posthog-icons.es.js |
| 258.5 KiB | ../node_modules/.pnpm/posthog-js@1.418.17/node_modules/posthog-js/dist/module.js |
| 249.5 KiB | src/taxonomy/core-filter-definitions-by-group.json |
| 154.2 KiB | ../node_modules/.pnpm/re2js@0.4.1/node_modules/re2js/build/index.esm.js |
| 126.8 KiB | ../node_modules/.pnpm/react-dom@18.3.1_react@18.3.1/node_modules/react-dom/cjs/react-dom.production.min.js |
| 104.7 KiB | src/lib/api.ts |
| 95.2 KiB | ../packages/quill/packages/quill/dist/index.js |
| 93.3 KiB | ../node_modules/.pnpm/prosemirror-view@1.40.1/node_modules/prosemirror-view/dist/index.js |
| 90.6 KiB | ../node_modules/.pnpm/@tiptap+core@3.20.6_@tiptap+pm@3.20.6/node_modules/@tiptap/core/dist/index.js |
Posted automatically by check-eager-graph · sizes are eager output bytes (shipped, post-tree-shake) from the esbuild metafile · part of #32479
✅ Toolbar bundle — eager 2.25 MiB within budget
What the toolbar ships to customer pages, measured from the esbuild output (minified, post-tree-shake). The eager set is the entry plus everything statically imported from it — fetched before any feature runs; deferred chunks load lazily. The eager guardrail is 5.72 MiB. Each output file must also stay below 10 MB, where CloudFront stops compressing it. The module boundary is enforced separately by check-toolbar-graph.
| Metric | Size | Δ vs base | Budget |
|---|---|---|---|
| Eager (shipped) entry + static imports |
2.25 MiB · 17 files | 🟢 -56 B (-0.0%) | ████░░░░░░ 39.3% of 5.72 MiB |
| Deferred (lazy) | 2.10 MiB · 33 files | no change | n/a — loads on demand |
Loader dist/toolbar.js |
1.1 KiB | no change | █░░░░░░░░░ 5.8% of 19.5 KiB |
Largest eagerly-shipped chunks
| Size | File |
|---|---|
| 746.7 KiB | dist/toolbar/toolbar-app-RBPKVAJT.css |
| 582.7 KiB | dist/toolbar/chunk-chunk-CTIXXM34.js |
| 484.6 KiB | dist/toolbar/chunk-chunk-PQM5GZBZ.js |
| 133.8 KiB | dist/toolbar/chunk-chunk-V5X3WXY7.js |
| 131.8 KiB | dist/toolbar/chunk-chunk-T5KY5WYR.js |
| 71.3 KiB | dist/toolbar/toolbar-app-D5BGFE6I.js |
| 69.0 KiB | dist/toolbar/chunk-chunk-27JL52RE.js |
| 35.6 KiB | dist/toolbar/chunk-chunk-VCDDUI5G.js |
| 20.9 KiB | dist/toolbar/chunk-chunk-HYUHGX5L.js |
| 12.2 KiB | dist/toolbar/chunk-chunk-PIK3PADE.js |
Posted automatically by check-toolbar-size · sizes are toolbar output bytes (shipped, post-tree-shake) from the esbuild metafile
✅ Dist folder size — 🔺 +72.2 KiB (+0.0%)
Total size of the built frontend/dist folder (all assets), compared against the base branch.
Total: 1442.51 MiB · 🔺 +72.2 KiB (+0.0%)
⚠️ Playwright — 1 flaky
🎭 Playwright report · View test results →
- Hover chart to see tooltip with data point values (chromium)
These issues are not necessarily caused by your changes.
Annoyed by this section? Help fix flakies and failures and it will go green!
ℹ️ ClickHouse migration SQL — 2 migration(s)
ClickHouse migration SQL per cloud environment
- unset
- all
ALTER TABLE posthog_test.metric_series1 ADD COLUMN IF NOT EXISTS `aggregation_temporality` LowCardinality(String) AFTER unit, ADD COLUMN IF NOT EXISTS `is_monotonic` Bool DEFAULT false AFTER aggregation_temporality
ALTER TABLE posthog_test.metric_series1 MODIFY TTL toDateTime(last_seen) + INTERVAL 90 DAY DELETE
ALTER TABLE posthog_test.metric_samples1 ADD COLUMN IF NOT EXISTS `count` UInt64 DEFAULT 1 AFTER value, ADD COLUMN IF NOT EXISTS `histogram_bounds` Array(Float64) AFTER count, ADD COLUMN IF NOT EXISTS `histogram_counts` Array(UInt64) AFTER histogram_bounds
ALTER TABLE posthog_test.metric_series ADD COLUMN IF NOT EXISTS `aggregation_temporality` LowCardinality(String) AFTER unit, ADD COLUMN IF NOT EXISTS `is_monotonic` Bool DEFAULT false AFTER aggregation_temporality
ALTER TABLE posthog_test.metric_samples ADD COLUMN IF NOT EXISTS `count` UInt64 DEFAULT 1 AFTER value, ADD COLUMN IF NOT EXISTS `histogram_bounds` Array(Float64) AFTER count, ADD COLUMN IF NOT EXISTS `histogram_counts` Array(UInt64) AFTER histogram_bounds
CREATE TABLE IF NOT EXISTS posthog_test.metrics1 ( `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)), `uuid` String CODEC(ZSTD(1)), `team_id` Int32 CODEC(ZSTD(1)), `trace_id` String CODEC(ZSTD(1)), `span_id` String CODEC(ZSTD(1)), `trace_flags` Int32 CODEC(ZSTD(1)), `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)), `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)), `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)), `service_name` LowCardinality(String) CODEC(ZSTD(1)), `metric_name` LowCardinality(String) CODEC(ZSTD(1)), `metric_type` LowCardinality(String) CODEC(ZSTD(1)), `value` Float64 CODEC(Gorilla, ZSTD(1)), `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)), `histogram_bounds` Array(Float64) CODEC(ZSTD(1)), `histogram_counts` Array(UInt64) CODEC(ZSTD(1)), `unit` LowCardinality(String) CODEC(ZSTD(1)), `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)), `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)), `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)), `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)), `instrumentation_scope` String CODEC(ZSTD(1)), `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)), `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)), `time_minute` DateTime ALIAS toStartOfMinute(timestamp), `attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str), INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1, INDEX idx_metric_type_set metric_type TYPE set(10) GRANULARITY 1, INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1, INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1, PROJECTION projection_aggregate_counts ( SELECT team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint, count() AS event_count, sum(value) AS total_value, min(value) AS min_value, max(value) AS max_value GROUP BY team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint ) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.metrics1', '{replica}-{shard}') PARTITION BY toDate(timestamp) PRIMARY KEY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) ORDER BY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) SETTINGS index_granularity_bytes = 104857600, index_granularity = 8192, ttl_only_drop_parts = 1
CREATE TABLE IF NOT EXISTS posthog_test.metrics AS posthog_test.metrics1 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'metrics1')
CREATE TABLE IF NOT EXISTS posthog_test.metric_attributes ( `team_id` Int32 CODEC(DoubleDelta, ZSTD(1)), `time_bucket` DateTime64(0) CODEC(DoubleDelta, ZSTD(1)), `service_name` LowCardinality(String) CODEC(ZSTD(1)), `resource_fingerprint` UInt64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1)), `attribute_key` LowCardinality(String) CODEC(ZSTD(1)), `attribute_value` String CODEC(ZSTD(1)), `attribute_count` SimpleAggregateFunction(sum, UInt64), `attribute_type` LowCardinality(String), INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1, INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1 ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metric_attributes', '{replica}-{shard}') PARTITION BY toDate(time_bucket) ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value) SETTINGS deduplicate_merge_projection_mode = 'drop', index_granularity = 8192
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.metrics1_to_metric_attributes TO posthog_test.metric_attributes ( `team_id` Int32, `time_bucket` DateTime64(0), `service_name` LowCardinality(String), `resource_fingerprint` UInt64, `attribute_key` LowCardinality(String), `attribute_value` String, `attribute_type` LowCardinality(String), `attribute_count` SimpleAggregateFunction(sum, UInt64) ) AS SELECT team_id, time_bucket, service_name, resource_fingerprint, attribute_key, attribute_value, attribute_type, attribute_count FROM ( SELECT team_id AS team_id, toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket, service_name AS service_name, resource_fingerprint, mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes, arrayJoin(attributes) AS attribute, 'metric' AS attribute_type, attribute.1 AS attribute_key, attribute.2 AS attribute_value, sumSimpleState(1) AS attribute_count FROM posthog_test.metrics1 GROUP BY team_id, time_bucket, service_name, resource_fingerprint, attributes )
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.metrics1_to_resource_attributes TO posthog_test.metric_attributes ( `team_id` Int32, `time_bucket` DateTime64(0), `service_name` LowCardinality(String), `resource_fingerprint` UInt64, `attribute_key` LowCardinality(String), `attribute_value` String, `attribute_type` LowCardinality(String), `attribute_count` SimpleAggregateFunction(sum, UInt64) ) AS SELECT team_id, time_bucket, service_name, resource_fingerprint, attribute_key, attribute_value, attribute_type, attribute_count FROM ( SELECT team_id AS team_id, toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket, service_name AS service_name, resource_fingerprint, arrayJoin(resource_attributes) AS attribute, 'resource' AS attribute_type, attribute.1 AS attribute_key, attribute.2 AS attribute_value, sumSimpleState(1) AS attribute_count FROM posthog_test.metrics1 GROUP BY team_id, time_bucket, service_name, resource_fingerprint, resource_attributes )
CREATE TABLE IF NOT EXISTS posthog_test.metrics_kafka_metrics ( `_partition` UInt32, `_topic` String, `max_offset` SimpleAggregateFunction(max, UInt64), `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)), `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)), `max_created_at` SimpleAggregateFunction(max, DateTime64(9)), `max_lag` SimpleAggregateFunction(max, UInt64) ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metrics_kafka_metrics', '{replica}-{shard}') ORDER BY (_topic, _partition) SETTINGS index_granularity = 8192
CREATE TABLE IF NOT EXISTS posthog_test.kafka_metrics_avro ( `uuid` String, `trace_id` String, `span_id` String, `trace_flags` Nullable(Int32), `timestamp` DateTime64(6), `observed_timestamp` DateTime64(6), `service_name` Nullable(String), `metric_name` Nullable(String), `metric_type` Nullable(String), `value` Nullable(Float64), `count` Nullable(Int64), `histogram_bounds` Array(Float64), `histogram_counts` Array(Int64), `unit` Nullable(String), `aggregation_temporality` Nullable(String), `is_monotonic` Nullable(UInt8), `resource_attributes` Map(String, String), `instrumentation_scope` Nullable(String), `attributes` Map(String, String), `series_fingerprint` Nullable(Int64) ) ENGINE = Kafka(warpstream_metrics, kafka_topic_list = 'clickhouse_metrics', kafka_group_name = 'clickhouse-metrics-avro-new', kafka_format = 'Avro') SETTINGS kafka_skip_broken_messages = 100, kafka_thread_per_consumer = 1, kafka_num_consumers = 8, kafka_poll_timeout_ms = 3000, kafka_poll_max_batch_size = 1000
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_mv TO posthog_test.metrics1 AS SELECT uuid, trace_id, span_id, ifNull(trace_flags, 0) AS trace_flags, timestamp, observed_timestamp, ifNull(service_name, '') AS service_name, ifNull(metric_name, '') AS metric_name, ifNull(metric_type, '') AS metric_type, ifNull(value, 0) AS value, toUInt64(ifNull(count, 1)) AS count, histogram_bounds, arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts, ifNull(unit, '') AS unit, ifNull(aggregation_temporality, '') AS aggregation_temporality, ifNull(is_monotonic, 0) AS is_monotonic, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes, ifNull(instrumentation_scope, '') AS instrumentation_scope, mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str, mapSort(mapFilter((k, v) -> isNotNull(v), mapApply((k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))), attributes))) AS attributes_map_float, toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id FROM posthog_test.kafka_metrics_avro SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_to_metric_samples TO posthog_test.metric_samples1 AS SELECT toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id, ifNull(metric_name, '') AS metric_name, reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint, timestamp, ifNull(value, 0) AS value, toUInt64(ifNull(count, 1)) AS count, histogram_bounds, arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts, trace_id, span_id, ifNull(trace_flags, 0) AS trace_flags FROM posthog_test.kafka_metrics_avro WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_to_metric_series TO posthog_test.metric_series1 AS SELECT toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id, ifNull(metric_name, '') AS metric_name, reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint, ifNull(metric_type, '') AS metric_type, ifNull(unit, '') AS unit, ifNull(aggregation_temporality, '') AS aggregation_temporality, ifNull(is_monotonic, 0) AS is_monotonic, ifNull(service_name, '') AS service_name, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes, timestamp AS last_seen FROM posthog_test.kafka_metrics_avro WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_kafka_metrics_mv TO posthog_test.metrics_kafka_metrics AS SELECT _partition, _topic, maxSimpleState(_offset) AS max_offset, maxSimpleState(observed_timestamp) AS max_observed_timestamp, maxSimpleState(timestamp) AS max_timestamp, maxSimpleState(now()) AS max_created_at, maxSimpleState(now() - observed_timestamp) AS max_lag FROM posthog_test.kafka_metrics_avro GROUP BY _partition, _topic
- all
- US, EU, DEV
- logs
ALTER TABLE posthog_test.metric_series1 ADD COLUMN IF NOT EXISTS `aggregation_temporality` LowCardinality(String) AFTER unit, ADD COLUMN IF NOT EXISTS `is_monotonic` Bool DEFAULT false AFTER aggregation_temporality
ALTER TABLE posthog_test.metric_series1 MODIFY TTL toDateTime(last_seen) + INTERVAL 90 DAY DELETE
ALTER TABLE posthog_test.metric_samples1 ADD COLUMN IF NOT EXISTS `count` UInt64 DEFAULT 1 AFTER value, ADD COLUMN IF NOT EXISTS `histogram_bounds` Array(Float64) AFTER count, ADD COLUMN IF NOT EXISTS `histogram_counts` Array(UInt64) AFTER histogram_bounds
ALTER TABLE posthog_test.metric_series ADD COLUMN IF NOT EXISTS `aggregation_temporality` LowCardinality(String) AFTER unit, ADD COLUMN IF NOT EXISTS `is_monotonic` Bool DEFAULT false AFTER aggregation_temporality
ALTER TABLE posthog_test.metric_samples ADD COLUMN IF NOT EXISTS `count` UInt64 DEFAULT 1 AFTER value, ADD COLUMN IF NOT EXISTS `histogram_bounds` Array(Float64) AFTER count, ADD COLUMN IF NOT EXISTS `histogram_counts` Array(UInt64) AFTER histogram_bounds
CREATE TABLE IF NOT EXISTS posthog_test.metrics1 ( `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp) CODEC(DoubleDelta, ZSTD(1)), `uuid` String CODEC(ZSTD(1)), `team_id` Int32 CODEC(ZSTD(1)), `trace_id` String CODEC(ZSTD(1)), `span_id` String CODEC(ZSTD(1)), `trace_flags` Int32 CODEC(ZSTD(1)), `timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)), `observed_timestamp` DateTime64(6) CODEC(DoubleDelta, ZSTD(1)), `created_at` DateTime64(6) MATERIALIZED now() CODEC(DoubleDelta, ZSTD(1)), `service_name` LowCardinality(String) CODEC(ZSTD(1)), `metric_name` LowCardinality(String) CODEC(ZSTD(1)), `metric_type` LowCardinality(String) CODEC(ZSTD(1)), `value` Float64 CODEC(Gorilla, ZSTD(1)), `count` UInt64 DEFAULT 1 CODEC(T64, ZSTD(1)), `histogram_bounds` Array(Float64) CODEC(ZSTD(1)), `histogram_counts` Array(UInt64) CODEC(ZSTD(1)), `unit` LowCardinality(String) CODEC(ZSTD(1)), `aggregation_temporality` LowCardinality(String) CODEC(ZSTD(1)), `is_monotonic` Bool DEFAULT false CODEC(ZSTD(1)), `resource_attributes` Map(LowCardinality(String), String) CODEC(ZSTD(1)), `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes) CODEC(DoubleDelta, ZSTD(1)), `instrumentation_scope` String CODEC(ZSTD(1)), `attributes_map_str` Map(LowCardinality(String), String) CODEC(ZSTD(1)), `attributes_map_float` Map(LowCardinality(String), Float64) CODEC(ZSTD(1)), `time_minute` DateTime ALIAS toStartOfMinute(timestamp), `attributes` Map(String, String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str), INDEX idx_metric_name_set metric_name TYPE set(100) GRANULARITY 1, INDEX idx_metric_type_set metric_type TYPE set(10) GRANULARITY 1, INDEX idx_attributes_str_keys mapKeys(attributes_map_str) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attributes_str_values mapValues(attributes_map_str) TYPE bloom_filter(0.001) GRANULARITY 1, INDEX idx_observed_minmax observed_timestamp TYPE minmax GRANULARITY 1, PROJECTION projection_aggregate_counts ( SELECT team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint, count() AS event_count, sum(value) AS total_value, min(value) AS min_value, max(value) AS max_value GROUP BY team_id, time_bucket, toStartOfMinute(timestamp), service_name, metric_name, metric_type, resource_fingerprint ) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/noshard/posthog.metrics1', '{replica}-{shard}') PARTITION BY toDate(timestamp) PRIMARY KEY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) ORDER BY (team_id, time_bucket, service_name, metric_name, resource_fingerprint, timestamp) SETTINGS index_granularity_bytes = 104857600, index_granularity = 8192, ttl_only_drop_parts = 1
CREATE TABLE IF NOT EXISTS posthog_test.metrics AS posthog_test.metrics1 ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'metrics1')
CREATE TABLE IF NOT EXISTS posthog_test.metric_attributes ( `team_id` Int32 CODEC(DoubleDelta, ZSTD(1)), `time_bucket` DateTime64(0) CODEC(DoubleDelta, ZSTD(1)), `service_name` LowCardinality(String) CODEC(ZSTD(1)), `resource_fingerprint` UInt64 DEFAULT 0 CODEC(DoubleDelta, ZSTD(1)), `attribute_key` LowCardinality(String) CODEC(ZSTD(1)), `attribute_value` String CODEC(ZSTD(1)), `attribute_count` SimpleAggregateFunction(sum, UInt64), `attribute_type` LowCardinality(String), INDEX idx_attribute_key attribute_key TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attribute_value attribute_value TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_attribute_key_n3 attribute_key TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1, INDEX idx_attribute_value_n3 attribute_value TYPE ngrambf_v1(3, 32768, 3, 0) GRANULARITY 1 ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metric_attributes', '{replica}-{shard}') PARTITION BY toDate(time_bucket) ORDER BY (team_id, attribute_type, time_bucket, resource_fingerprint, attribute_key, attribute_value) SETTINGS deduplicate_merge_projection_mode = 'drop', index_granularity = 8192
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.metrics1_to_metric_attributes TO posthog_test.metric_attributes ( `team_id` Int32, `time_bucket` DateTime64(0), `service_name` LowCardinality(String), `resource_fingerprint` UInt64, `attribute_key` LowCardinality(String), `attribute_value` String, `attribute_type` LowCardinality(String), `attribute_count` SimpleAggregateFunction(sum, UInt64) ) AS SELECT team_id, time_bucket, service_name, resource_fingerprint, attribute_key, attribute_value, attribute_type, attribute_count FROM ( SELECT team_id AS team_id, toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket, service_name AS service_name, resource_fingerprint, mapFilter((k, v) -> ((length(k) < 256) AND (length(v) < 256)), attributes) AS attributes, arrayJoin(attributes) AS attribute, 'metric' AS attribute_type, attribute.1 AS attribute_key, attribute.2 AS attribute_value, sumSimpleState(1) AS attribute_count FROM posthog_test.metrics1 GROUP BY team_id, time_bucket, service_name, resource_fingerprint, attributes )
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.metrics1_to_resource_attributes TO posthog_test.metric_attributes ( `team_id` Int32, `time_bucket` DateTime64(0), `service_name` LowCardinality(String), `resource_fingerprint` UInt64, `attribute_key` LowCardinality(String), `attribute_value` String, `attribute_type` LowCardinality(String), `attribute_count` SimpleAggregateFunction(sum, UInt64) ) AS SELECT team_id, time_bucket, service_name, resource_fingerprint, attribute_key, attribute_value, attribute_type, attribute_count FROM ( SELECT team_id AS team_id, toStartOfInterval(timestamp, toIntervalMinute(10)) AS time_bucket, service_name AS service_name, resource_fingerprint, arrayJoin(resource_attributes) AS attribute, 'resource' AS attribute_type, attribute.1 AS attribute_key, attribute.2 AS attribute_value, sumSimpleState(1) AS attribute_count FROM posthog_test.metrics1 GROUP BY team_id, time_bucket, service_name, resource_fingerprint, resource_attributes )
CREATE TABLE IF NOT EXISTS posthog_test.metrics_kafka_metrics ( `_partition` UInt32, `_topic` String, `max_offset` SimpleAggregateFunction(max, UInt64), `max_observed_timestamp` SimpleAggregateFunction(max, DateTime64(9)), `max_timestamp` SimpleAggregateFunction(max, DateTime64(9)), `max_created_at` SimpleAggregateFunction(max, DateTime64(9)), `max_lag` SimpleAggregateFunction(max, UInt64) ) ENGINE = ReplicatedAggregatingMergeTree('/clickhouse/tables/noshard/posthog.metrics_kafka_metrics', '{replica}-{shard}') ORDER BY (_topic, _partition) SETTINGS index_granularity = 8192
CREATE TABLE IF NOT EXISTS posthog_test.kafka_metrics_avro ( `uuid` String, `trace_id` String, `span_id` String, `trace_flags` Nullable(Int32), `timestamp` DateTime64(6), `observed_timestamp` DateTime64(6), `service_name` Nullable(String), `metric_name` Nullable(String), `metric_type` Nullable(String), `value` Nullable(Float64), `count` Nullable(Int64), `histogram_bounds` Array(Float64), `histogram_counts` Array(Int64), `unit` Nullable(String), `aggregation_temporality` Nullable(String), `is_monotonic` Nullable(UInt8), `resource_attributes` Map(String, String), `instrumentation_scope` Nullable(String), `attributes` Map(String, String), `series_fingerprint` Nullable(Int64) ) ENGINE = Kafka(warpstream_metrics, kafka_topic_list = 'clickhouse_metrics', kafka_group_name = 'clickhouse-metrics-avro-new', kafka_format = 'Avro') SETTINGS kafka_skip_broken_messages = 100, kafka_thread_per_consumer = 1, kafka_num_consumers = 8, kafka_poll_timeout_ms = 3000, kafka_poll_max_batch_size = 1000
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_mv TO posthog_test.metrics1 AS SELECT uuid, trace_id, span_id, ifNull(trace_flags, 0) AS trace_flags, timestamp, observed_timestamp, ifNull(service_name, '') AS service_name, ifNull(metric_name, '') AS metric_name, ifNull(metric_type, '') AS metric_type, ifNull(value, 0) AS value, toUInt64(ifNull(count, 1)) AS count, histogram_bounds, arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts, ifNull(unit, '') AS unit, ifNull(aggregation_temporality, '') AS aggregation_temporality, ifNull(is_monotonic, 0) AS is_monotonic, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes, ifNull(instrumentation_scope, '') AS instrumentation_scope, mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str, mapSort(mapFilter((k, v) -> isNotNull(v), mapApply((k, v) -> (concat(k, '__float'), toFloat64OrNull(JSONExtract(v, 'String'))), attributes))) AS attributes_map_float, toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id FROM posthog_test.kafka_metrics_avro SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_to_metric_samples TO posthog_test.metric_samples1 AS SELECT toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id, ifNull(metric_name, '') AS metric_name, reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint, timestamp, ifNull(value, 0) AS value, toUInt64(ifNull(count, 1)) AS count, histogram_bounds, arrayMap(x -> toUInt64(x), histogram_counts) AS histogram_counts, trace_id, span_id, ifNull(trace_flags, 0) AS trace_flags FROM posthog_test.kafka_metrics_avro WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_to_metric_series TO posthog_test.metric_series1 AS SELECT toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id, ifNull(metric_name, '') AS metric_name, reinterpretAsUInt64(assumeNotNull(series_fingerprint)) AS series_fingerprint, ifNull(metric_type, '') AS metric_type, ifNull(unit, '') AS unit, ifNull(aggregation_temporality, '') AS aggregation_temporality, ifNull(is_monotonic, 0) AS is_monotonic, ifNull(service_name, '') AS service_name, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes, mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), attributes)) AS attributes, timestamp AS last_seen FROM posthog_test.kafka_metrics_avro WHERE kafka_metrics_avro.series_fingerprint IS NOT NULL SETTINGS min_insert_block_size_rows = 0, min_insert_block_size_bytes = 0
CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_metrics_avro_kafka_metrics_mv TO posthog_test.metrics_kafka_metrics AS SELECT _partition, _topic, maxSimpleState(_offset) AS max_offset, maxSimpleState(observed_timestamp) AS max_observed_timestamp, maxSimpleState(timestamp) AS max_timestamp, maxSimpleState(now()) AS max_created_at, maxSimpleState(now() - observed_timestamp) AS max_lag FROM posthog_test.kafka_metrics_avro GROUP BY _partition, _topic
- logs
jzhu13
force-pushed
the
posthog-code/local-metrics-clickhouse-ingest
branch
from
August 26, 2026 19:05
596c7b2 to
536a519
Compare
jzhu13
changed the title
jzhu13
marked this pull request as ready for review
August 26, 2026 19:09
OTLP metrics reached the clickhouse_metrics topic but never ClickHouse outside the cloud logs cluster: no consumer table existed, so the metrics product showed only hand-seeded data on dev stacks. Migration 0305 creates the ingest chain on the LOGS role: metrics1 (raw) and its distributed reader, metrics_kafka_metrics, kafka_metrics_avro, and the materialized views into metrics1, metric_samples1, metric_series1, and metrics_kafka_metrics. metrics1 itself had no creating migration — cloud provisions it out of band and dev laptops via the schema-sync path — so 0305 creates it too (0283 did this for metric_samples1/metric_series1 but left metrics1 out). Every statement is IF NOT EXISTS, so it is a no-op where the objects already exist. Add the warpstream_metrics named collection to the dev and multinode ClickHouse configs so the Kafka table resolves its broker list. These objects are modeled in the roles/logs/shared + cloud HCL layers for dev/prod but not in the local goldens; metrics1 diverges per env (value/count codecs on local and prod, none on dev), so a single all-env HCL declaration cannot serve both yet. Skip them in the local-multi and local-single convergence dumps until that is reconciled, so the gate does not read the migration-created objects as drift. Generated-By: PostHog Code Task-Id: 6522cf9f-0db9-46ac-93ad-12a849ece06e
jzhu13
force-pushed
the
posthog-code/local-metrics-clickhouse-ingest
branch
from
August 26, 2026 19:18
536a519 to
7e2756f
Compare
…igration metrics was the last product provisioning its ClickHouse objects via a hand-run script (bin/clickhouse-metrics.sql); logs and traces already moved to migrations. Delete the script and its hogli command, and extend migration 0305 to cover everything it created that no migration did: metric_attributes and its two feeder views, on top of metrics1 and the Kafka ingest chain. All IF NOT EXISTS, so environments provisioned by the script or the cloud HCL are unaffected. Generated-By: PostHog Code Task-Id: 6522cf9f-0db9-46ac-93ad-12a849ece06e
|
Gates denied this PR (new ClickHouse migration hit the migrations deny-list, and size/scope classified it T2-never); it's a cross-team schema change with no reviews or approvals, so it needs human sign-off rather than automated approval.
Gate mechanics and policy version
|
jzhu13
added
stamphog
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
1 participant
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.
Problem
OTLP metrics sent to a dev or multinode stack never reached the Metrics product — they stopped at the
clickhouse_metricsKafka topic because no ClickHouse consumer table existed outside the cloud logs cluster. Anyone developing against Metrics locally saw only hand-seeded rows.The metrics ingest objects (
kafka_metrics_avro+ its materialized views) were provisioned only via the cloud HCL layer. Andmetrics1(the raw table those views write to) had no creating migration at all — cloud provisions it out of band, dev laptops via the schema-sync path — so it is absent wherever only migrations run (the multinode CI smoke, a from-migrations dev stack).Before
flowchart LR A[otel collector / capture-logs] --> B[(metrics_ingestion)] B --> C[ingestion-metrics consumer] C --> D[(clickhouse_metrics)] D -.->|no consumer table| X[nothing] classDef phBlue fill:#1d4aff,stroke:#1d4aff,color:#fff; classDef phGray fill:#e5e7eb,stroke:#c7ccd1,color:#000; classDef phRed fill:#f54e00,stroke:#f54e00,color:#fff; class A,C phBlue; class B,D phGray; class X phRed;After
flowchart LR A[otel collector / capture-logs] --> B[(metrics_ingestion)] B --> C[ingestion-metrics consumer] C --> D[(clickhouse_metrics)] D --> E[kafka_metrics_avro] E --> F[(metrics1 / metric_samples1 / metric_series1 / metrics_kafka_metrics)] classDef phBlue fill:#1d4aff,stroke:#1d4aff,color:#fff; classDef phGray fill:#e5e7eb,stroke:#c7ccd1,color:#000; classDef phYellow fill:#f9bd2b,stroke:#f9bd2b,color:#000; class A,C,E phBlue; class B,D phGray; class F phYellow;Changes
migrate_clickhouse, instead of only hand-seeded data.0305_metrics_kafka_ingestcreates the ingest chain onNodeRole.LOGS:metrics1+ its distributed reader,metrics_kafka_metrics,kafka_metrics_avro, and the four materialized views. AllIF NOT EXISTS, so it is a no-op where the objects already exist (cloud, seeded laptops).metrics1in a migration closes the gap that made the multinode smoke fail —0283createdmetric_samples1/metric_series1this way but omittedmetrics1.posthog/clickhouse/metrics/kafka_metrics.pyholds the SQL, matching the cloud HCL definitions.bin/clickhouse-metrics.sqlinit script (and itshogli clickhouse:metrics:initcommand) is deleted — metrics was the last product provisioning ClickHouse objects that way; the migration replaces it.0305therefore also createsmetric_attributesand its two feeder views, which only the script covered before.warpstream_metricsnamed collection added to the dev and multinode ClickHouse configs so the Kafka table resolves its broker list.local-multiandlocal-singleconvergence dumps (exclude-*.hcl). They are HCL-modeled for dev/prod but not locally;metrics1diverges per env (value/count codecs on local and prod, none on dev), so a single all-env HCL declaration can't serve both yet.Note
Modeling these objects in an all-env HCL layer (so the local convergence gate checks them too) is a deliberate follow-up. It requires reconciling
metrics1's per-env codec/engine divergence between the Django SQL and the HCLsharedlayer, which touches cloud goldens and is out of scope here.How did you test this code?
bash posthog/clickhouse/hcl/check.sh) passes: duplicates match baseline, all envs validate, no golden drift, generated SQL fresh.hclexp introspecton a running dev ClickHouse → 0 matches), so thelocal-multigate won't read them as drift.clickhouse-metrics-avroconsumer group wentDead→Stableand drained the backlog intometrics1/metric_samples1/metric_series1, with exemplar trace ids resolving againsttrace_spans.metrics1 does not exist, and the HCL duplicate/validation errors).Automatic notifications
Docs update
None — dev-environment schema only.
🤖 Agent context
Autonomy: Human-driven (agent-assisted)
/clickhouse-migrations,/writing-pr-descriptions.metrics1is created by no migration, and the HCL guard rejected the layer (duplicate declarations vs the cloudshared/cloudlayers, plusmetrics1referenced but undeclared in the local compositions).metrics1genuinely diverges per env (Django SQL carriesGorilla/T64codecs and a single-shard engine; the HCLsharedlayer has neither), so it can't be unified into one HCL declaration without changing cloud goldens. This revision instead fixes the real runtime bug (createmetrics1via migration) and scopes the objects out of the local convergence gate, leaving cloud goldens untouched.Created with PostHog Code