[bciaraldi]

Problem

• The repo places the prod logs Kafka ingest on the ingestion-events nodes, but every cloud env now runs it on the ingestion-apm nodes.
• A cluster built from the migrations, or an HCL reconcile, would put the ingest back on ingestion-events.
• The ingest moved on the clusters by hand, so no issue tracks it.

Changes

• The prod-us and prod-eu apm compositions now include roles/coshared/logs_avro_ingest, so the HCL matches the clusters. Dev already had it.
• Migration 0359_logs_ingest_to_apm creates the ingest on APM with IF NOT EXISTS, then drops it from INGESTION_EVENTS with IF EXISTS.
  • It runs in the cloud envs only. The multinode stack keeps the ingest on its ingestion-events node.
  • The apm side comes first, so the shared consumer group keeps members during a rebuild.
  • On the current clusters it is a no-op.
  • The apm creates set require_hosts=True, so a topology without apm hosts fails before any drop.
  • The ingestion-events writable_logs34 stays, so blocks it still queues finish sending. Drop it by hand once its distribution queue is empty.
• kafka_logs_avro now uses 4 consumers per node in the base layer and in KAFKA_LOGS_AVRO_TABLE_SQL, which matches both prod regions. Dev keeps its own patch.
• Mechanical: regenerated prod apm goldens and SQL, and comment fixes in logs34.py.

Important

Depends on PostHog/posthog-cloud-infra#11480, which composes logs_avro_ingest into the cloud prod apm stacks. The cloud compose gate fails here until that merges. Re-run the gate after it does.

Warning

#121979 also adds a 0359 migration and still targets INGESTION_EVENTS for the prod logs ingest. The PR that lands second renumbers, and #121979 needs APM for all cloud envs.

Note

The prod-eu writable_logs34 has no background_insert_batch setting, while the golden has background_insert_batch = 1. A Distributed table cannot change settings with ALTER, so it converges when the table is next recreated.

How did you test this code?

• posthog/clickhouse/hcl/check.sh passed locally after gen-golden.sh and gen-sql.sh.
• check-cloud.sh: the posthog-cloud-infra compose gate composes dev, prod-us and prod-eu with this tree.
• posthog/clickhouse/test/test_migrations.py passed locally.
• The generated apm SQL matches the live table definitions in dev, prod-us and prod-eu: columns, MV query and Kafka settings.
• Not run: the multinode migration smoke, which CI runs, and the migration against a live cluster.

Test rationale: No new test. test_migrations.py already checks each deployment branch of a migration for ON CLUSTER, and check.sh guards the HCL composition.

👉 Stay up-to-date with PostHog coding conventions for a smoother review.

Release status

• No feature flag controls this change

Docs update

None.

🤖 Agent context

Autonomy: Human-driven (agent-assisted)

Agent: Claude Code, Claude Opus 5.5

• Skills used: clickhouse-migrations, writing-code-comments and writing-pr-descriptions, read from .agents/skills/.
• The consumer count started as a prod-us patch. It became the base value after prod-eu moved to the same count.
• The duplicate search found no PR for this move. feat(logs): add internal_source column to logs34 #121979 overlaps, as the warning above describes.

[trunk-io]

😎 Merged successfully - details.

[github-actions]

🤖 CI report

⚠️ Trunk lane — backend Python lane

This PR is assigned to the backend Python lane. It runs backend Python tests and may merge in parallel with PRs in other lanes.

✅ Duplication (Python) — clean

New Python code duplication introduced by this branch. Fails at 70+ tokens in app code, or 150+ tokens when both copies live in test files. Advisory while the gate proves itself: extract a shared helper instead of copying.

✅ Duplication (TypeScript) — clean

New TypeScript code duplication introduced by this branch. Fails at 70+ tokens in app code, or 150+ tokens when both copies live in test files. Advisory while the gate proves itself: extract a shared helper instead of copying.

⚠️ Playwright — 2 flaky

🎭 Playwright report · View test results →

⚠️ 2 flaky tests:

• Redirect to a saved insight after login (chromium)
• Logout in another tab results in logout in the current tab too (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 — 1 migration(s)

ClickHouse migration SQL per cloud environment

• unset
  • 0359_logs_ingest_to_apm — operations list is empty under this environment
• US, EU, DEV
  • apm

    CREATE TABLE IF NOT EXISTS posthog_test.writable_logs34
    (
        `time_bucket` DateTime MATERIALIZED toStartOfDay(timestamp),
        `original_expiry_timestamp` DateTime64(6),
        `uuid` String,
        `team_id` Int32,
        `trace_id` String,
        `span_id` String,
        `trace_flags` Int32,
        `timestamp` DateTime64(6) CODEC(DoubleDelta),
        `observed_timestamp` DateTime64(6),
        `created_at` DateTime64(6) MATERIALIZED now(),
        `body` String,
        `severity_text` LowCardinality(String),
        `severity_number` Int32,
        `service_name` LowCardinality(String),
        `resource_attributes` Map(LowCardinality(String), String),
        `resource_fingerprint` UInt64 MATERIALIZED cityHash64(resource_attributes),
        `instrumentation_scope` String,
        `event_name` String,
        `attributes_map_str` Map(LowCardinality(String), String),
        `level` String ALIAS severity_text,
        `mat_body_ipv4_matches` Array(String) ALIAS extractAll(body, '(\d\.((25[0-5]|(2[0-4]|1(0, 1)[0-9])(0, 1)[0-9])\.)(2, 2)([0-9]))'),
        `time_minute` DateTime ALIAS toStartOfMinute(timestamp),
        `attributes` Map(LowCardinality(String), String) ALIAS mapApply((k, v) -> (left(k, -5), v), attributes_map_str),
        `attributes_map_float` Map(LowCardinality(String), Float64) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__float'), toFloat64OrNull(v)), attributes_map_str)),
        `attributes_map_datetime` Map(LowCardinality(String), DateTime64(6)) MATERIALIZED mapFilter((k, v) -> (v IS NOT NULL), mapApply((k, v) -> (concat(left(k, -5), '__datetime'), parseDateTimeBestEffortOrNull(v, 6)), attributes_map_str)),
        `_partition` UInt32,
        `_topic` String,
        `_offset` UInt64,
        `_bytes_uncompressed` UInt64,
        `_bytes_compressed` UInt64,
        `_record_count` UInt64,
        `pattern` String,
        `pattern_version` UInt8
    )
    ENGINE = Distributed('posthog_single_shard', 'posthog_test', 'logs34')
    SETTINGS background_insert_batch = 1

    CREATE TABLE IF NOT EXISTS posthog_test.kafka_logs_avro
    (
        `uuid` String,
        `trace_id` String,
        `span_id` String,
        `trace_flags` Int32,
        `timestamp` DateTime64(6),
        `observed_timestamp` DateTime64(6),
        `body` String,
        `severity_text` String,
        `severity_number` Int32,
        `service_name` String,
        `resource_attributes` Map(LowCardinality(String), String),
        `instrumentation_scope` String,
        `event_name` String,
        `attributes` Map(LowCardinality(String), String),
        `retention_days` Nullable(Int32),
        `pattern` Nullable(String),
        `pattern_version` Nullable(Int32)
    )
    ENGINE = Kafka(warpstream_logs, kafka_topic_list = 'clickhouse_logs', kafka_group_name = 'clickhouse-logs-avro-new', kafka_format = 'Avro')
    SETTINGS
        kafka_skip_broken_messages = 100,
        kafka_num_consumers = 4,
        kafka_poll_timeout_ms = 3000,
        kafka_poll_max_batch_size = 1000,
        kafka_thread_per_consumer = 1,
        input_format_avro_allow_missing_fields = 1

    CREATE MATERIALIZED VIEW IF NOT EXISTS posthog_test.kafka_logs34_avro_mv TO posthog_test.writable_logs34
    (
        `uuid` String,
        `trace_id` String,
        `span_id` String,
        `trace_flags` Int32,
        `timestamp` DateTime64(6),
        `observed_timestamp` DateTime64(6),
        `body` String,
        `severity_text` String,
        `severity_number` Int32,
        `service_name` String,
        `instrumentation_scope` String,
        `event_name` String,
        `attributes_map_str` Map(String, String),
        `resource_attributes` Map(String, String),
        `team_id` Int32,
        `original_expiry_timestamp` DateTime64(6),
        `_partition` UInt64,
        `_topic` LowCardinality(String),
        `_offset` UInt64,
        `_record_count` Int64,
        `_bytes_uncompressed` Nullable(Int64),
        `_bytes_compressed` Nullable(Int64),
        `pattern` String,
        `pattern_version` UInt8
    )
    AS SELECT
        uuid,
        trace_id,
        span_id,
        trace_flags,
        timestamp,
        observed_timestamp,
        body,
        severity_text,
        severity_number,
        service_name,
        instrumentation_scope,
        event_name,
        mapSort(mapApply((k, v) -> (concat(k, '__str'), JSONExtractString(v)), attributes)) AS attributes_map_str,
        mapSort(mapApply((k, v) -> (k, JSONExtractString(v)), resource_attributes)) AS resource_attributes,
        toInt32OrZero(_headers.value[indexOf(_headers.name, 'team_id')]) AS team_id,
        timestamp + toIntervalDay(if((retention_days IS NOT NULL) AND (retention_days > 0), retention_days, toInt32OrDefault(_headers.value[indexOf(_headers.name, 'retention-days')], toInt32(15)))) AS original_expiry_timestamp,
        _partition,
        _topic,
        _offset,
        toInt64OrDefault(_headers.value[indexOf(_headers.name, 'record_count')], toInt64(1)) AS _record_count,
        toInt64OrNull(_headers.value[indexOf(_headers.name, 'bytes_uncompressed')]) / _record_count AS _bytes_uncompressed,
        toInt64OrNull(_headers.value[indexOf(_headers.name, 'bytes_compressed')]) / _record_count AS _bytes_compressed,
        ifNull(pattern, '') AS pattern,
        toUInt8(ifNull(pattern_version, 0)) AS pattern_version
    FROM posthog_test.kafka_logs_avro

  • events

    DROP TABLE IF EXISTS posthog_test.kafka_logs_avro

    DROP TABLE IF EXISTS posthog_test.kafka_logs34_avro_mv

[posthog]

PostHog Review · Standard · reviewed dab50e7: 1 Must fix, 1 Should fix.

This is the only automatic review while this pull request is a draft. Comment @posthog review to review new changes sooner. Otherwise, a review of the new changes runs when you mark it ready for review.

[parameterai]

Risk: No findings

This is a ClickHouse migration-only change: it moves the logs Kafka ingest from the ingestion-events nodes to the ingestion-apm nodes in cloud environments (migration 0359, gated to deployed cloud), and updates the HCL manifest, goldens and generated SQL to match what the clusters already run by hand. It contains no attacker-facing surface — static DDL over settings-derived identifiers, with no auth, secrets, or user input involved. No security issues found.

_{Sentinel reviewed 48c2ca9 · Review settings}

[trunk-io]

   

Failed Test	Failure Summary	Logs
Auth › Redirect to a saved insight after login		Logs ↗︎
Auth › Logout in another tab results in logout in the current tab too	The page reload was aborted, possibly because the frame was detached.	Logs ↗︎
test_projection_counts_unique_billable_events	The test failed because the expected count of 1 was not met; instead, the count was 0.	Logs ↗︎

View Full Report ↗︎ ⋅ Docs

[deployment-status-posthog]

Deploy status

Environment	Status	Deployed At	Workflow
dev	✅ Deployed	2026-10-22 20:51 UTC	Run
prod-us	✅ Deployed	2026-10-22 21:09 UTC	Run
prod-eu	✅ Deployed	2026-10-22 21:10 UTC	Run