From ddfe314ee34bc7aea2d03601a461b098824de5be Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Tue, 1 Sep 2026 09:35:06 -0700 Subject: [PATCH 01/11] feat(flags): monitor flag_evaluations parity against events Adds a daily Dagster job that compares flag_evaluations with events per team and posts to Slack when a row the fork wrote never reached events. The existing counters increment on the Kafka ack, so nothing past it is observed: the Kafka engine table, the MV, the writable Distributed, and the shard. kafka_flag_evaluations sets kafka_skip_broken_messages, so a malformed row is discarded after a successful produce. Only a table to table comparison sees that. The job takes its team list from flag_evaluations itself rather than a mirror of the deployment allowlist, so the two cannot drift, and reads a lookback window so a day where the fork wrote nothing for a team reads as a deficit rather than as the team not being enabled. Excess never alerts on its own. flag_evaluations carries a permanent surplus because clients re-send event uuids and its plain MergeTree cannot collapse them, so alerting on any difference would fire daily for every team. A deficit alerts at any size, because it has measured zero. Also adds an ack-duration histogram to the fork. The pending-acks gauge only separates stalled from not stalled, which moves once the lane is already holding up offset commits. Generated-By: PostHog Desktop Task-Id: 125630eb-acdf-45c7-9574-ae8c1088540d --- .../fork-flag-evaluations-step.ts | 17 +- posthog/dags/flag_evaluations_parity.py | 242 ++++++++++++++++++ posthog/dags/locations/ingestion.py | 3 + posthog/dags/owners.yaml | 2 + .../tests/test_flag_evaluations_parity.py | 99 +++++++ 5 files changed, 362 insertions(+), 1 deletion(-) create mode 100644 posthog/dags/flag_evaluations_parity.py create mode 100644 posthog/dags/tests/test_flag_evaluations_parity.py diff --git a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts index c7aaac343960..83dcc9b52604 100644 --- a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts +++ b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts @@ -1,4 +1,4 @@ -import { Counter, Gauge } from 'prom-client' +import { Counter, Gauge, Histogram } from 'prom-client' import { FLAG_EVALUATIONS_OUTPUT, FlagEvaluationsOutput } from '~/common/outputs' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' @@ -34,6 +34,18 @@ export const flagEvaluationsPendingAcks = new Gauge({ help: 'flag_evaluations produces queued but not yet acknowledged; sustained non-zero means the fork is holding up batches', }) +/** + * How long a produce takes to be acknowledged. The pending-acks gauge above only + * separates "stalled" from "not stalled", so it moves once the lane is already + * holding up offset commits. This shows the lane slowing down while it still + * works, which is what gates widening the fork onto a lane other teams share. + */ +export const flagEvaluationsAckDuration = new Histogram({ + name: 'ingestion_flag_evaluations_ack_duration_seconds', + help: 'Time from queueing a flag_evaluations produce to the broker acknowledging it', + buckets: [0.01, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30], +}) + export const flagEvaluationsSetPropsTotal = new Counter({ name: 'ingestion_flag_evaluations_set_props_total', help: '$feature_flag_called events carrying $set/$set_once, which a dedicated flag lane bypassing person processing would lose', @@ -106,6 +118,7 @@ export function createForkFlagEvaluationsStep { flagEvaluationsPendingAcks.dec(messages.length) + stopAckTimer() flagEvaluationsEventsTotal.labels('dual_written').inc(messages.length) }, (error: unknown) => { flagEvaluationsPendingAcks.dec(messages.length) + stopAckTimer() if (error instanceof MessageSizeTooLarge) { // An oversized row is a property of the row, not of the broker, so // it gets its own outcome and no warning: retrying or alerting on diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py new file mode 100644 index 000000000000..f3243911312c --- /dev/null +++ b/posthog/dags/flag_evaluations_parity.py @@ -0,0 +1,242 @@ +from dataclasses import dataclass +from datetime import UTC, date, datetime, timedelta + +import dagster +import pydantic +import dagster_slack +from clickhouse_driver import Client + +from posthog import settings +from posthog.clickhouse.cluster import ClickhouseCluster +from posthog.dags.common import JobOwners, settings_with_log_comment +from posthog.models.flag_evaluations.sql import FLAG_EVALUATIONS_SOURCE_EVENT, FLAG_EVALUATIONS_TABLE + +FLAG_EVALUATIONS_SLACK_CHANNEL = "#alerts-ingestion" + +JOB_NAME = "flag_evaluations_parity" + + +class FlagEvaluationsParityConfig(dagster.Config): + """Configuration for the flag_evaluations parity check.""" + + enabled_team_lookback_days: int = pydantic.Field( + default=7, + description=( + "How far back to look for teams with flag_evaluations rows. A team enabled at any " + "point in this window is checked, so a day where the fork wrote nothing at all " + "reads as a total deficit rather than as the team not being enabled." + ), + ) + teams_per_query: int = pydantic.Field( + default=20, + description=( + "Teams compared per query. The comparison groups by uuid, so its memory scales with " + "the rows in the batch, not with the team count. Lower this if a batch containing a " + "high-volume team runs out of memory." + ), + ) + max_teams: int = pydantic.Field( + default=500, + description=( + "Cap on teams checked in one run, highest volume first. Reaching it is reported, " + "never silent, because a truncated run otherwise reads as full coverage." + ), + ) + + +@dataclass(frozen=True) +class TeamParity: + team_id: int + only_in_events: int + only_in_flag_evaluations: int + in_both: int + + +@dataclass(frozen=True) +class ParityResults: + day: date + checked: list[TeamParity] + teams_enabled: int + teams_truncated: int + + +def _target_day() -> date: + return (datetime.now(tz=UTC) - timedelta(days=1)).date() + + +@dagster.op +def measure_flag_evaluations_parity( + context: dagster.OpExecutionContext, + config: FlagEvaluationsParityConfig, + cluster: dagster.ResourceParam[ClickhouseCluster], +) -> ParityResults: + """Compare flag_evaluations against events for every team the fork is writing for. + + The produce-side counters answer "did we hand the row to Kafka", which is not the + question. Everything past the ack -- the Kafka engine table, the MV, the writable + Distributed, the shard -- is unobserved, and kafka_flag_evaluations sets + kafka_skip_broken_messages, so a malformed row is discarded after a successful produce. + Only comparing the two tables catches that. + """ + + day = _target_day() + + def run_queries(client: Client) -> ParityResults: + query_settings = settings_with_log_comment(context) + + # The enabled set comes from the data rather than from a mirror of the charts + # allowlist, so the two cannot drift. + enabled_rows = client.execute( + f""" + SELECT team_id, count() AS rows + FROM {settings.CLICKHOUSE_DATABASE}.{FLAG_EVALUATIONS_TABLE} + WHERE toDate(timestamp) >= %(lookback_start)s AND toDate(timestamp) <= %(day)s + GROUP BY team_id + ORDER BY rows DESC + """, + {"lookback_start": day - timedelta(days=config.enabled_team_lookback_days), "day": day}, + settings=query_settings, + ) + + ranked_teams = [int(row[0]) for row in enabled_rows] + teams = ranked_teams[: config.max_teams] + truncated = len(ranked_teams) - len(teams) + context.log.info(f"{len(ranked_teams)} teams enabled, checking {len(teams)} for {day}") + + checked: list[TeamParity] = [] + for start in range(0, len(teams), config.teams_per_query): + batch = teams[start : start + config.teams_per_query] + rows = client.execute( + f""" + SELECT + team_id, + countIf(in_ev > 0 AND in_fe = 0) AS only_in_events, + countIf(in_fe > 0 AND in_ev = 0) AS only_in_flag_evaluations, + countIf(in_fe > 0 AND in_ev > 0) AS in_both + FROM ( + SELECT team_id, uuid, max(is_fe) AS in_fe, max(is_ev) AS in_ev + FROM ( + SELECT team_id, uuid, 1 AS is_fe, 0 AS is_ev + FROM {settings.CLICKHOUSE_DATABASE}.{FLAG_EVALUATIONS_TABLE} + WHERE team_id IN %(teams)s AND toDate(timestamp) = %(day)s + + UNION ALL + + SELECT team_id, uuid, 0 AS is_fe, 1 AS is_ev + FROM {settings.CLICKHOUSE_DATABASE}.events + WHERE team_id IN %(teams)s AND toDate(timestamp) = %(day)s + AND event = %(event)s + ) + GROUP BY team_id, uuid + ) + GROUP BY team_id + """, + {"teams": batch, "day": day, "event": FLAG_EVALUATIONS_SOURCE_EVENT}, + settings=query_settings, + ) + checked.extend( + TeamParity( + team_id=int(row[0]), + only_in_events=int(row[1]), + only_in_flag_evaluations=int(row[2]), + in_both=int(row[3]), + ) + for row in rows + ) + + return ParityResults( + day=day, + checked=checked, + teams_enabled=len(ranked_teams), + teams_truncated=truncated, + ) + + return cluster.any_host(run_queries).result() + + +def parity_alert_lines(results: ParityResults) -> list[str] | None: + """The Slack message for a run, or None when there is nothing to say. + + Only a deficit and an uncovered team are worth waking someone for. Excess is reported as + context but never triggers the alert on its own: flag_evaluations carries a permanent + surplus of a fraction of a percent, because clients re-send event uuids and its plain + MergeTree cannot collapse them while the ReplacingMergeTree behind events does. Alerting + on any difference would fire every day for every team and teach everyone to ignore it. + + There is no threshold on the deficit for the opposite reason: it has measured zero on + every day checked so far, so any non-zero value is a regression a threshold would hide. + """ + + deficits = sorted( + (team for team in results.checked if team.only_in_events > 0), + key=lambda team: team.only_in_events, + reverse=True, + ) + if not deficits and not results.teams_truncated: + return None + + excess_total = sum(team.only_in_flag_evaluations for team in results.checked) + lines = [ + f"*flag_evaluations parity* for `{results.day}`", + f"Checked {len(results.checked)} of {results.teams_enabled} enabled teams. Excess rows: {excess_total:,}.", + ] + if results.teams_truncated: + lines.append(f"{results.teams_truncated} enabled teams were not checked; raise `max_teams`.") + lines.extend( + f"Team `{team.team_id}`: {team.only_in_events:,} rows in events with no flag_evaluations row" + for team in deficits[:10] + ) + if len(deficits) > 10: + lines.append(f"…and {len(deficits) - 10} more teams with a deficit") + return lines + + +@dagster.op +def report_flag_evaluations_parity( + context: dagster.OpExecutionContext, + results: ParityResults, + slack: dagster.ResourceParam[dagster_slack.SlackResource], +) -> None: + """Post the parity result to Slack when it needs attention.""" + + context.add_output_metadata( + { + "day": str(results.day), + "teams_checked": len(results.checked), + "teams_enabled": results.teams_enabled, + "deficit_rows": sum(team.only_in_events for team in results.checked), + "excess_rows": sum(team.only_in_flag_evaluations for team in results.checked), + } + ) + + lines = parity_alert_lines(results) + if lines is None: + context.log.info(f"Parity holds for {len(results.checked)} teams on {results.day}") + return + + if not settings.CLOUD_DEPLOYMENT: + context.log.info("Skipping Slack notification in non-prod environment") + return + + try: + slack.get_client().chat_postMessage( + channel=FLAG_EVALUATIONS_SLACK_CHANNEL, + blocks=[{"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}}], + ) + except Exception as e: + context.log.exception(f"Failed to send Slack notification: {e}") + + +@dagster.job(tags={"owner": JobOwners.TEAM_INGESTION.value}) +def flag_evaluations_parity(): + """Check that every row the flag-evaluations fork wrote also reached the events table.""" + report_flag_evaluations_parity(measure_flag_evaluations_parity()) + + +flag_evaluations_parity_schedule = dagster.ScheduleDefinition( + job=flag_evaluations_parity, + # After the previous UTC day is complete and its merges have settled. + cron_schedule="0 4 * * *", + execution_timezone="UTC", + name="flag_evaluations_parity_schedule", +) diff --git a/posthog/dags/locations/ingestion.py b/posthog/dags/locations/ingestion.py index 832a10df5129..90f48b0689c1 100644 --- a/posthog/dags/locations/ingestion.py +++ b/posthog/dags/locations/ingestion.py @@ -4,6 +4,7 @@ delete_persons_from_trigger_log, detach_distinct_id, distinct_id_usage, + flag_evaluations_parity, ingestion_assets, person_property_reconciliation, persondistinctids_without_person_cleanup, @@ -21,6 +22,7 @@ delete_persons_from_trigger_log.delete_persons_from_trigger_log_job, detach_distinct_id.detach_distinct_id_job, distinct_id_usage.distinct_id_usage_monitoring, + flag_evaluations_parity.flag_evaluations_parity, person_property_reconciliation.person_property_reconciliation_job, persondistinctids_without_person_cleanup.persondistinctids_without_person_cleanup_job, persons_new_backfill.persons_new_backfill_job, @@ -28,6 +30,7 @@ ], schedules=[ distinct_id_usage.distinct_id_usage_monitoring_schedule, + flag_evaluations_parity.flag_evaluations_parity_schedule, ], sensors=[ person_property_reconciliation.person_property_reconciliation_scheduler, diff --git a/posthog/dags/owners.yaml b/posthog/dags/owners.yaml index af282d98edf0..61d58ffc0bec 100644 --- a/posthog/dags/owners.yaml +++ b/posthog/dags/owners.yaml @@ -3,6 +3,8 @@ owners: [] rules: - match: - '/detach_distinct_id.py' + - '/flag_evaluations_parity.py' + - '/tests/test_flag_evaluations_parity.py' - '/fix_missing_person_overrides.py' - '/fix_person_id_overrides.py' - '/person_property_reconciliation.py' diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py new file mode 100644 index 000000000000..0eee8752c9ed --- /dev/null +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -0,0 +1,99 @@ +from datetime import UTC, datetime, timedelta +from functools import partial +from uuid import UUID + +import pytest + +import dagster +from clickhouse_driver import Client +from parameterized import parameterized + +from posthog.clickhouse.cluster import ClickhouseCluster +from posthog.dags.flag_evaluations_parity import ( + FlagEvaluationsParityConfig, + ParityResults, + TeamParity, + measure_flag_evaluations_parity, + parity_alert_lines, +) +from posthog.dags.tests.conftest import insert_flag_evaluations + + +def _results(checked: list[TeamParity], teams_truncated: int = 0) -> ParityResults: + return ParityResults( + day=datetime.now(tz=UTC).date(), + checked=checked, + teams_enabled=len(checked) + teams_truncated, + teams_truncated=teams_truncated, + ) + + +class TestParityAlertLines: + @parameterized.expand( + [ + ("clean", [TeamParity(1, 0, 0, 100)], 0), + # The permanent duplicate surplus must never alert on its own, or the job fires + # every day for every team and stops being read. + ("excess_only", [TeamParity(1, 0, 250, 100)], 0), + ] + ) + def test_stays_quiet(self, _name: str, checked: list[TeamParity], truncated: int) -> None: + assert parity_alert_lines(_results(checked, truncated)) is None + + @parameterized.expand( + [ + ("deficit", [TeamParity(1, 3, 0, 100)], 0, "3 rows in events"), + # A capped run must not read as full coverage. + ("truncated_without_deficit", [TeamParity(1, 0, 0, 100)], 5, "5 enabled teams were not checked"), + ] + ) + def test_alerts(self, _name: str, checked: list[TeamParity], truncated: int, expected: str) -> None: + lines = parity_alert_lines(_results(checked, truncated)) + assert lines is not None + assert any(expected in line for line in lines) + + +@pytest.mark.django_db +def test_measure_attributes_deficit_and_excess_to_the_right_team(cluster: ClickhouseCluster) -> None: + day = (datetime.now(tz=UTC) - timedelta(days=1)).replace(hour=12, minute=0, second=0, microsecond=0) + matched, also_matched, events_only, fork_only = UUID(int=1), UUID(int=2), UUID(int=3), UUID(int=4) + person = UUID(int=99) + clean_team, deficit_team, excess_team = 8001, 8002, 8003 + + def insert_events(client: Client) -> None: + client.execute( + """INSERT INTO writable_events (team_id, event, distinct_id, person_id, uuid, timestamp) + VALUES + """, + [ + (clean_team, "$feature_flag_called", "d", person, matched, day), + (deficit_team, "$feature_flag_called", "d", person, also_matched, day), + (deficit_team, "$feature_flag_called", "d", person, events_only, day), + # A different event with a uuid the fork never sees must not read as a deficit. + (excess_team, "$pageview", "d", person, UUID(int=5), day), + ], + ) + + cluster.any_host(insert_events).result() + cluster.any_host( + partial( + insert_flag_evaluations, + [ + (clean_team, "d", person, matched, day), + # Enabled, so the op checks it, but one of its two events never arrived. + (deficit_team, "d", person, also_matched, day), + # Two rows for one uuid: a duplicate is excess, and excess is not a deficit. + (excess_team, "d", person, fork_only, day), + (excess_team, "d", person, fork_only, day), + ], + ) + ).result() + + # Two teams per query, three teams enabled, so the batching loop runs more than once. + config = FlagEvaluationsParityConfig(teams_per_query=2) + results = measure_flag_evaluations_parity(dagster.build_op_context(), config, cluster) + + by_team = {team.team_id: team for team in results.checked} + assert by_team[clean_team] == TeamParity(clean_team, only_in_events=0, only_in_flag_evaluations=0, in_both=1) + assert by_team[deficit_team] == TeamParity(deficit_team, only_in_events=1, only_in_flag_evaluations=0, in_both=1) + assert by_team[excess_team] == TeamParity(excess_team, only_in_events=0, only_in_flag_evaluations=1, in_both=0) From f2e75ce413037c3805435ecc9f0e1ac4c9b52b9c Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:17:18 +0000 Subject: [PATCH 02/11] fix(flags): time ack-latency histogram on success only A failed produce called stopAckTimer(), so it recorded a sample into the flag_evaluations ack-latency histogram. Fast local rejects like MessageSizeTooLarge settle before any broker round trip, so those near-zero samples pull the reported ack latency down. That histogram gates widening the fork onto a shared lane, so a failure should not read as a healthy ack. Observe the histogram only in the success arm. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- .../steps/event-processing/fork-flag-evaluations-step.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts index 83dcc9b52604..591a96b6f712 100644 --- a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts +++ b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.ts @@ -129,12 +129,14 @@ export function createForkFlagEvaluationsStep { flagEvaluationsPendingAcks.dec(messages.length) + // Observe latency only on a real ack. A failed produce, especially a + // local reject like MessageSizeTooLarge, settles before any broker + // round trip, so timing it would pull the ack-latency signal down. stopAckTimer() flagEvaluationsEventsTotal.labels('dual_written').inc(messages.length) }, (error: unknown) => { flagEvaluationsPendingAcks.dec(messages.length) - stopAckTimer() if (error instanceof MessageSizeTooLarge) { // An oversized row is a property of the row, not of the broker, so // it gets its own outcome and no warning: retrying or alerting on From 6a44047acc60093dee6a7e10d99a3bf3728e8251 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:22:34 +0000 Subject: [PATCH 03/11] fix(flags): match producer flag predicate in parity events branch The events side of the parity comparison counted every $feature_flag_called row, but the fork only writes a flag_evaluations row when $feature_flag is a non-empty string (hasFlagKey). An event the fork skips on purpose therefore read as a deficit and would fire a false loss alert. Add AND JSONExtractString(properties, '$feature_flag') != '' to the events branch, mirroring the fork's typeof-string check. Update the ClickHouse-backed test so matching events carry a valid $feature_flag, and add missing, empty, and non-string cases that must not read as a deficit. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 4 ++++ .../dags/tests/test_flag_evaluations_parity.py | 18 +++++++++++++----- 2 files changed, 17 insertions(+), 5 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index f3243911312c..47e48ef94463 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -126,6 +126,10 @@ def run_queries(client: Client) -> ParityResults: FROM {settings.CLICKHOUSE_DATABASE}.events WHERE team_id IN %(teams)s AND toDate(timestamp) = %(day)s AND event = %(event)s + -- Same predicate the fork applies before it writes: an event + -- without a non-empty string $feature_flag never produces a row, + -- so counting it here would read as a false deficit. + AND JSONExtractString(properties, '$feature_flag') != '' ) GROUP BY team_id, uuid ) diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py index 0eee8752c9ed..1b442cfff27c 100644 --- a/posthog/dags/tests/test_flag_evaluations_parity.py +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -61,16 +61,24 @@ def test_measure_attributes_deficit_and_excess_to_the_right_team(cluster: Clickh clean_team, deficit_team, excess_team = 8001, 8002, 8003 def insert_events(client: Client) -> None: + flag_props = '{"$feature_flag": "my-flag"}' client.execute( - """INSERT INTO writable_events (team_id, event, distinct_id, person_id, uuid, timestamp) + """INSERT INTO writable_events (team_id, event, distinct_id, person_id, uuid, timestamp, properties) VALUES """, [ - (clean_team, "$feature_flag_called", "d", person, matched, day), - (deficit_team, "$feature_flag_called", "d", person, also_matched, day), - (deficit_team, "$feature_flag_called", "d", person, events_only, day), + (clean_team, "$feature_flag_called", "d", person, matched, day, flag_props), + (deficit_team, "$feature_flag_called", "d", person, also_matched, day, flag_props), + (deficit_team, "$feature_flag_called", "d", person, events_only, day, flag_props), + # The fork only writes for a non-empty string $feature_flag, so a + # $feature_flag_called event without one has no flag_evaluations row by + # design and must not read as a deficit. Missing, empty, and non-string + # all match the fork's typeof-string check. + (deficit_team, "$feature_flag_called", "d", person, UUID(int=6), day, "{}"), + (deficit_team, "$feature_flag_called", "d", person, UUID(int=7), day, '{"$feature_flag": ""}'), + (deficit_team, "$feature_flag_called", "d", person, UUID(int=8), day, '{"$feature_flag": 5}'), # A different event with a uuid the fork never sees must not read as a deficit. - (excess_team, "$pageview", "d", person, UUID(int=5), day), + (excess_team, "$pageview", "d", person, UUID(int=5), day, ""), ], ) From d19456cf306d2c33500a46ad09a749fe81fce1cc Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:24:15 +0000 Subject: [PATCH 04/11] fix(flags): name the deployment in parity Slack alerts prod-US and prod-EU post the parity alert to the same channel, and team ids come from a per-region sequence, so team `12345` names a different team in each region. An operator reading a deficit alert could not tell which region lost rows. Append an Environment line with settings.CLOUD_DEPLOYMENT in the report op, matching the pattern in the sibling dags alert senders. The early non-prod guard above guarantees the value is set. parity_alert_lines stays a pure function with no settings access. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index 47e48ef94463..a40ea52dee2b 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -222,6 +222,10 @@ def report_flag_evaluations_parity( context.log.info("Skipping Slack notification in non-prod environment") return + # Both regions post to the same channel and team ids are per-region, so the alert + # must name the deployment that lost the rows. + lines.append(f"Environment: {settings.CLOUD_DEPLOYMENT}") + try: slack.get_client().chat_postMessage( channel=FLAG_EVALUATIONS_SLACK_CHANNEL, From 2f374ff673f3bbf5a33a10f527c65011dd248fa2 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:29:07 +0000 Subject: [PATCH 05/11] fix(flags): skip a team's activation day in parity check The enabled-team query kept every team with a flag_evaluations row in the lookback window, including a team whose first row lands on the checked day. The fork is gated per team at run time, so a team switched on mid-day has $feature_flag_called events from before the allowlist reached the pods that reach events with no flag_evaluations row. The comparison counted those as a deficit, so the rollout gate this job provides read red on every activation. Return min(toDate(timestamp)) per team and drop a team whose earliest row is the checked day, before applying max_teams. Log how many were skipped. Add a test team switched on during the day whose would-be deficit must not surface, and give the compared teams an earlier-day row so they stay in the enabled set. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 14 +++++++++++--- .../tests/test_flag_evaluations_parity.py | 19 +++++++++++++++++-- 2 files changed, 28 insertions(+), 5 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index a40ea52dee2b..b05e606bfd6c 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -88,7 +88,7 @@ def run_queries(client: Client) -> ParityResults: # allowlist, so the two cannot drift. enabled_rows = client.execute( f""" - SELECT team_id, count() AS rows + SELECT team_id, count() AS rows, min(toDate(timestamp)) AS first_day FROM {settings.CLICKHOUSE_DATABASE}.{FLAG_EVALUATIONS_TABLE} WHERE toDate(timestamp) >= %(lookback_start)s AND toDate(timestamp) <= %(day)s GROUP BY team_id @@ -98,10 +98,18 @@ def run_queries(client: Client) -> ParityResults: settings=query_settings, ) - ranked_teams = [int(row[0]) for row in enabled_rows] + # A team whose earliest row in the lookback window falls on the checked day was + # switched on that day. Its events from before the allowlist reached the pods have + # no fork row, so comparing it would report a deficit for a fork that works. Skip it; + # its first full day is checked on the next run. + ranked_teams = [int(row[0]) for row in enabled_rows if row[2] < day] + skipped_activation = len(enabled_rows) - len(ranked_teams) teams = ranked_teams[: config.max_teams] truncated = len(ranked_teams) - len(teams) - context.log.info(f"{len(ranked_teams)} teams enabled, checking {len(teams)} for {day}") + context.log.info( + f"{len(ranked_teams)} teams enabled, checking {len(teams)} for {day}" + + (f", skipped {skipped_activation} activated that day" if skipped_activation else "") + ) checked: list[TeamParity] = [] for start in range(0, len(teams), config.teams_per_query): diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py index 1b442cfff27c..eb7120b867b5 100644 --- a/posthog/dags/tests/test_flag_evaluations_parity.py +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -56,9 +56,10 @@ def test_alerts(self, _name: str, checked: list[TeamParity], truncated: int, exp @pytest.mark.django_db def test_measure_attributes_deficit_and_excess_to_the_right_team(cluster: ClickhouseCluster) -> None: day = (datetime.now(tz=UTC) - timedelta(days=1)).replace(hour=12, minute=0, second=0, microsecond=0) + earlier = day - timedelta(days=1) matched, also_matched, events_only, fork_only = UUID(int=1), UUID(int=2), UUID(int=3), UUID(int=4) person = UUID(int=99) - clean_team, deficit_team, excess_team = 8001, 8002, 8003 + clean_team, deficit_team, excess_team, activation_team = 8001, 8002, 8003, 8004 def insert_events(client: Client) -> None: flag_props = '{"$feature_flag": "my-flag"}' @@ -79,6 +80,10 @@ def insert_events(client: Client) -> None: (deficit_team, "$feature_flag_called", "d", person, UUID(int=8), day, '{"$feature_flag": 5}'), # A different event with a uuid the fork never sees must not read as a deficit. (excess_team, "$pageview", "d", person, UUID(int=5), day, ""), + # activation_team is switched on during the checked day: this earlier event + # has no fork row, so checking the team would read as a deficit. The op must + # skip the team instead. + (activation_team, "$feature_flag_called", "d", person, UUID(int=9), day, flag_props), ], ) @@ -87,17 +92,25 @@ def insert_events(client: Client) -> None: partial( insert_flag_evaluations, [ + # A row before the checked day marks each team as one the fork was already + # writing for, so it stays in the enabled set and is compared on the day. + (clean_team, "d", person, UUID(int=10), earlier), + (deficit_team, "d", person, UUID(int=11), earlier), + (excess_team, "d", person, UUID(int=12), earlier), (clean_team, "d", person, matched, day), # Enabled, so the op checks it, but one of its two events never arrived. (deficit_team, "d", person, also_matched, day), # Two rows for one uuid: a duplicate is excess, and excess is not a deficit. (excess_team, "d", person, fork_only, day), (excess_team, "d", person, fork_only, day), + # activation_team's first-ever row is on the checked day, so it was switched + # on that day and must be skipped rather than compared. + (activation_team, "d", person, UUID(int=13), day), ], ) ).result() - # Two teams per query, three teams enabled, so the batching loop runs more than once. + # Two teams per query and three teams compared, so the batching loop runs more than once. config = FlagEvaluationsParityConfig(teams_per_query=2) results = measure_flag_evaluations_parity(dagster.build_op_context(), config, cluster) @@ -105,3 +118,5 @@ def insert_events(client: Client) -> None: assert by_team[clean_team] == TeamParity(clean_team, only_in_events=0, only_in_flag_evaluations=0, in_both=1) assert by_team[deficit_team] == TeamParity(deficit_team, only_in_events=1, only_in_flag_evaluations=0, in_both=1) assert by_team[excess_team] == TeamParity(excess_team, only_in_events=0, only_in_flag_evaluations=1, in_both=0) + # Switched on during the checked day, so it is skipped rather than reported as a deficit. + assert activation_team not in by_team From f54cde37f0e3dcc809e6047dd83141f039cb9bbe Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:34:58 +0000 Subject: [PATCH 06/11] fix(flags): report queried team count in parity summary The parity comparison's outer GROUP BY team_id emits a row only for a team that had rows on the day, so an idle team that was queried but wrote nothing that day is dropped from `checked`. The Slack summary used len(checked) as the checked-team count, which understated coverage and could contradict the "N enabled teams were not checked" truncation line beside it. Report teams_enabled - teams_truncated as the checked count and state how many teams had rows on the day, matching the alert format in the PR description. Split the Dagster run metadata the same way. Add a unit test that pins the summary to the queried count. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 10 ++++++++-- posthog/dags/tests/test_flag_evaluations_parity.py | 14 ++++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index b05e606bfd6c..6f035dd12710 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -187,10 +187,15 @@ def parity_alert_lines(results: ParityResults) -> list[str] | None: if not deficits and not results.teams_truncated: return None + # A team the query covered but that had no rows on the day emits no result row, so + # len(checked) counts only teams with rows. Report the queried count so this does not + # understate coverage or contradict the truncation line below. + queried = results.teams_enabled - results.teams_truncated excess_total = sum(team.only_in_flag_evaluations for team in results.checked) lines = [ f"*flag_evaluations parity* for `{results.day}`", - f"Checked {len(results.checked)} of {results.teams_enabled} enabled teams. Excess rows: {excess_total:,}.", + f"Checked {queried} of {results.teams_enabled} enabled teams " + f"({len(results.checked)} had rows on the day). Excess rows: {excess_total:,}.", ] if results.teams_truncated: lines.append(f"{results.teams_truncated} enabled teams were not checked; raise `max_teams`.") @@ -214,7 +219,8 @@ def report_flag_evaluations_parity( context.add_output_metadata( { "day": str(results.day), - "teams_checked": len(results.checked), + "teams_checked": results.teams_enabled - results.teams_truncated, + "teams_with_rows": len(results.checked), "teams_enabled": results.teams_enabled, "deficit_rows": sum(team.only_in_events for team in results.checked), "excess_rows": sum(team.only_in_flag_evaluations for team in results.checked), diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py index eb7120b867b5..fc6c8f8112ef 100644 --- a/posthog/dags/tests/test_flag_evaluations_parity.py +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -52,6 +52,20 @@ def test_alerts(self, _name: str, checked: list[TeamParity], truncated: int, exp assert lines is not None assert any(expected in line for line in lines) + def test_summary_counts_queried_teams_not_only_teams_with_rows(self) -> None: + # Ten teams were queried but only one had rows on the day. The summary must report the + # queried count, not len(checked), or it understates coverage and contradicts the + # truncation line. + results = ParityResults( + day=datetime.now(tz=UTC).date(), + checked=[TeamParity(1, 5, 0, 0)], + teams_enabled=10, + teams_truncated=0, + ) + lines = parity_alert_lines(results) + assert lines is not None + assert any("Checked 10 of 10 enabled teams (1 had rows on the day)" in line for line in lines) + @pytest.mark.django_db def test_measure_attributes_deficit_and_excess_to_the_right_team(cluster: ClickhouseCluster) -> None: From a895bdec40caa031f1e9621ea2238c03e2c55778 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:40:54 +0000 Subject: [PATCH 07/11] fix(flags): cap the parity comparison query's resources The shared Dagster cluster resource disables both ClickHouse ceilings (max_execution_time and max_memory_usage are "0"), so the per-team parity comparison ran with no memory limit and no deadline on the cluster that serves customer queries. Its group-by builds one aggregate state per uuid across both tables for a whole day, and the batch order puts the highest-volume teams first, so as the fork widens the query threatens shared memory. The retry policy re-runs a total-memory or socket-timeout failure up to 8 times, so a heavy query repeated at full cost. Cap the comparison query with max_execution_time, max_memory_usage of 100 GB, and distributed_aggregation_memory_efficient, following the pattern in posthog/dags/sessions.py. A per-query memory or execution-time breach is not retried by the policy, so a heavy batch now fails once. The ceilings are generous guardrails for a teams_per_query-sized batch; that knob still lowers per-query scope. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index 6f035dd12710..219da3d94ebb 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -111,6 +111,23 @@ def run_queries(client: Client) -> ParityResults: + (f", skipped {skipped_activation} activated that day" if skipped_activation else "") ) + # The shared cluster resource disables both ceilings (max_execution_time and + # max_memory_usage are "0"), so this comparison must cap itself: it runs on the + # cluster that serves customer queries and its group-by state grows with the + # $feature_flag_called uuids of the largest teams for a whole day. A per-query memory + # breach reports "for query", which the retry policy leaves alone -- unlike a total + # server breach or a socket timeout, which it retries up to 8 times -- so a heavy + # batch fails once instead of repeating at full cost. distributed_aggregation_memory_efficient + # streams each shard's (team_id, uuid) states rather than merging them all in the + # initiator. The ceilings are generous guardrails for a batch of teams_per_query + # teams; lower teams_per_query if a batch reaches them. + comparison_settings = { + **query_settings, + "max_execution_time": "600", + "max_memory_usage": str(100 * 1024**3), + "distributed_aggregation_memory_efficient": "1", + } + checked: list[TeamParity] = [] for start in range(0, len(teams), config.teams_per_query): batch = teams[start : start + config.teams_per_query] @@ -144,7 +161,7 @@ def run_queries(client: Client) -> ParityResults: GROUP BY team_id """, {"teams": batch, "day": day, "event": FLAG_EVALUATIONS_SOURCE_EVENT}, - settings=query_settings, + settings=comparison_settings, ) checked.extend( TeamParity( From 3bcc37525d3182bc04f9fcb2b6f3d259ccbd5758 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:44:18 +0000 Subject: [PATCH 08/11] fix(flags): skip parity run while one is already active The parity schedule was a bare dagster.ScheduleDefinition, so every tick submitted a run with no active-run check. After a schedule outage Dagster replays up to 5 catch-up ticks close together, which would run several comparisons at once against the shared events cluster and post duplicate alerts for the same day. Convert it to a @dagster.schedule function guarded by @skip_if_already_running, the documented helper already used by peers on the same 0 4 * * * cron. The run behavior is otherwise unchanged. The separate execution-time cap bounds a single run's duration but does not prevent concurrent catch-up runs, which this guard does. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index 219da3d94ebb..ecdffd761830 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -8,7 +8,7 @@ from posthog import settings from posthog.clickhouse.cluster import ClickhouseCluster -from posthog.dags.common import JobOwners, settings_with_log_comment +from posthog.dags.common import JobOwners, settings_with_log_comment, skip_if_already_running from posthog.models.flag_evaluations.sql import FLAG_EVALUATIONS_SOURCE_EVENT, FLAG_EVALUATIONS_TABLE FLAG_EVALUATIONS_SLACK_CHANNEL = "#alerts-ingestion" @@ -272,10 +272,16 @@ def flag_evaluations_parity(): report_flag_evaluations_parity(measure_flag_evaluations_parity()) -flag_evaluations_parity_schedule = dagster.ScheduleDefinition( +@dagster.schedule( job=flag_evaluations_parity, # After the previous UTC day is complete and its merges have settled. cron_schedule="0 4 * * *", execution_timezone="UTC", name="flag_evaluations_parity_schedule", ) +@skip_if_already_running +def flag_evaluations_parity_schedule(context: dagster.ScheduleEvaluationContext) -> dagster.RunRequest: + # Skip while a previous run is still active so catch-up ticks after an outage do not run + # several comparisons at once against the shared events cluster and post duplicate alerts + # for the same day. + return dagster.RunRequest() From 333319fa29cbd7ea9b77fc780f0cbf407c207d48 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:54:03 +0000 Subject: [PATCH 09/11] feat(flags): let a parity run target a chosen day The op derived the checked day only from the wall clock (yesterday, UTC) with no override, so an operator rerunning to re-check the date a past alert named silently checked yesterday instead. Add a target_day config field (YYYY-MM-DD) and resolve the day through a small helper: the explicit override when set, else yesterday. A manual run can now re-check a named date via run config. Scheduled runs pass no config, so their behavior is unchanged. Deriving the day from the scheduled tick, so a delayed or replayed scheduled run checks the day it was scheduled for, is a separate change to the schedule and is not included here. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 14 +++++++++++++- posthog/dags/tests/test_flag_evaluations_parity.py | 8 +++++++- 2 files changed, 20 insertions(+), 2 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index ecdffd761830..7fb5bbeeed41 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -42,6 +42,13 @@ class FlagEvaluationsParityConfig(dagster.Config): "never silent, because a truncated run otherwise reads as full coverage." ), ) + target_day: str | None = pydantic.Field( + default=None, + description=( + "Day to check as YYYY-MM-DD. Defaults to yesterday (UTC). Set it on a manual run to " + "re-check the day a past alert named." + ), + ) @dataclass(frozen=True) @@ -64,6 +71,11 @@ def _target_day() -> date: return (datetime.now(tz=UTC) - timedelta(days=1)).date() +def _resolve_target_day(target_day: str | None) -> date: + """The day to check: an explicit YYYY-MM-DD override, else yesterday (UTC).""" + return date.fromisoformat(target_day) if target_day else _target_day() + + @dagster.op def measure_flag_evaluations_parity( context: dagster.OpExecutionContext, @@ -79,7 +91,7 @@ def measure_flag_evaluations_parity( Only comparing the two tables catches that. """ - day = _target_day() + day = _resolve_target_day(config.target_day) def run_queries(client: Client) -> ParityResults: query_settings = settings_with_log_comment(context) diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py index fc6c8f8112ef..4dfa430eaca2 100644 --- a/posthog/dags/tests/test_flag_evaluations_parity.py +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -1,4 +1,4 @@ -from datetime import UTC, datetime, timedelta +from datetime import UTC, date, datetime, timedelta from functools import partial from uuid import UUID @@ -13,6 +13,7 @@ FlagEvaluationsParityConfig, ParityResults, TeamParity, + _resolve_target_day, measure_flag_evaluations_parity, parity_alert_lines, ) @@ -28,6 +29,11 @@ def _results(checked: list[TeamParity], teams_truncated: int = 0) -> ParityResul ) +def test_resolve_target_day_uses_the_override_when_set() -> None: + # An operator rerunning to re-check a past alert's date must get that date, not yesterday. + assert _resolve_target_day("2026-08-31") == date(2026, 8, 31) + + class TestParityAlertLines: @parameterized.expand( [ From 1abf277220411a243d38414f2964e241202b69d5 Mon Sep 17 00:00:00 2001 From: "posthog[bot]" <206114724+posthog[bot]@users.noreply.github.com> Date: Tue, 1 Sep 2026 20:57:15 +0000 Subject: [PATCH 10/11] fix(flags): add a text fallback to the parity Slack alert The parity alert posted only blocks with no top-level text, so Slack had no notification or accessibility fallback and could drop the message more easily when it rejects the rich blocks. Pass text alongside blocks, reusing the same joined lines, so the fallback carries the full message including the deployment and the truncation clause. Both the text and the block now reference one message value, so they cannot drift. Matches the pattern in posthog/dags/slack_alerts.py. The deployment identifier the same review raised is already present in the alert lines from an earlier commit this review pass. Generated-By: PostHog Desktop Task-Id: 67542d94-107a-43d2-81da-16cea3b8b99e --- posthog/dags/flag_evaluations_parity.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index 7fb5bbeeed41..163cd8949cd5 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -269,10 +269,14 @@ def report_flag_evaluations_parity( # must name the deployment that lost the rows. lines.append(f"Environment: {settings.CLOUD_DEPLOYMENT}") + # text is the notification and accessibility fallback, and what Slack shows if it rejects + # the rich blocks. It carries the same lines, so the truncation clause survives that reject. + message = "\n".join(lines) try: slack.get_client().chat_postMessage( channel=FLAG_EVALUATIONS_SLACK_CHANNEL, - blocks=[{"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}}], + text=message, + blocks=[{"type": "section", "text": {"type": "mrkdwn", "text": message}}], ) except Exception as e: context.log.exception(f"Failed to send Slack notification: {e}") From 2552e5f7ac0fa95c879948377023374a65eafd5c Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Tue, 1 Sep 2026 13:55:08 -0700 Subject: [PATCH 11/11] fix(flags): reject non-string flag keys in the parity events branch The events branch kept any $feature_flag whose JSON value was not an empty string, but JSONExtractString renders a number, boolean, object or array as its text, so 123 came back as '123' and passed the non-empty check. The fork writes a row only for a non-empty string, so each of those events had no flag_evaluations row by design and was counted as a lost one. A deficit alerts at any size, so a single such event a day would alert every morning for a fork that is working. Measured against ClickHouse 26.6.2.158: only null, "" and a missing key return ''. JSONType now rejects the other types before the empty-string check. Removing the JSONType line makes the seeded numeric row read as a deficit, so the existing ClickHouse case covers the regression. Also alert when no team is enabled, rather than reporting the run as clean. A cleared allowlist stops the fork writing, and once the lookback window passes those teams leave the query entirely, which produces no deficit and no message. Log the alert body before posting, because a Slack failure is logged and swallowed, and the per-team deficit lines exist nowhere else. Cover the max_teams truncation counts, which had no test. The excess docstring now cites a uuid-level production measurement. The re-sent-uuid explanation it replaces cannot hold, because the comparison groups by (team_id, uuid) before counting. --- posthog/dags/flag_evaluations_parity.py | 33 +++++++++++++++---- .../tests/test_flag_evaluations_parity.py | 16 +++++++++ 2 files changed, 42 insertions(+), 7 deletions(-) diff --git a/posthog/dags/flag_evaluations_parity.py b/posthog/dags/flag_evaluations_parity.py index 163cd8949cd5..bd5ac180c77e 100644 --- a/posthog/dags/flag_evaluations_parity.py +++ b/posthog/dags/flag_evaluations_parity.py @@ -163,9 +163,13 @@ def run_queries(client: Client) -> ParityResults: FROM {settings.CLICKHOUSE_DATABASE}.events WHERE team_id IN %(teams)s AND toDate(timestamp) = %(day)s AND event = %(event)s - -- Same predicate the fork applies before it writes: an event - -- without a non-empty string $feature_flag never produces a row, - -- so counting it here would read as a false deficit. + -- The fork writes a row only when $feature_flag is a non-empty string, + -- and lets every other $feature_flag_called event continue to the events + -- table untouched. Without the same condition here, each one reads as a + -- lost row. JSONExtractString on its own is not that condition, because + -- it renders a number, boolean, object or array as its text, so + -- JSONType has to reject those types before the empty-string check. + AND JSONType(properties, '$feature_flag') = 'String' AND JSONExtractString(properties, '$feature_flag') != '' ) GROUP BY team_id, uuid @@ -199,15 +203,26 @@ def parity_alert_lines(results: ParityResults) -> list[str] | None: """The Slack message for a run, or None when there is nothing to say. Only a deficit and an uncovered team are worth waking someone for. Excess is reported as - context but never triggers the alert on its own: flag_evaluations carries a permanent - surplus of a fraction of a percent, because clients re-send event uuids and its plain - MergeTree cannot collapse them while the ReplacingMergeTree behind events does. Alerting - on any difference would fire every day for every team and teach everyone to ignore it. + context but never triggers the alert on its own. It counts uuids the fork wrote that the + events table has no row for. Over the week to 2026-08-31 in prod-US it stayed at or below + 0.09 percent of the uuids the fork wrote, and was zero on four of the seven days. Re-sent + uuids are not what it counts: the comparison groups by (team_id, uuid), so the duplicate + rows that flag_evaluations keeps and the ReplacingMergeTree behind events collapses fold + into one entry per side before anything is counted, and a re-sent uuid that reached both + tables lands in in_both. There is no threshold on the deficit for the opposite reason: it has measured zero on every day checked so far, so any non-zero value is a regression a threshold would hide. """ + # No enabled team means the fork wrote nothing anywhere for the whole lookback window, which + # a rolled-back allowlist produces. Staying quiet here would report it as a clean run. + if not results.teams_enabled: + return [ + f"*flag_evaluations parity* for `{results.day}`", + "No team wrote a flag_evaluations row in the lookback window, so nothing was compared.", + ] + deficits = sorted( (team for team in results.checked if team.only_in_events > 0), key=lambda team: team.only_in_events, @@ -261,6 +276,10 @@ def report_flag_evaluations_parity( context.log.info(f"Parity holds for {len(results.checked)} teams on {results.day}") return + # A Slack failure is logged and swallowed below, and the per-team deficit lines exist + # nowhere else, so record them before the post is attempted. + context.log.warning("\n".join(lines)) + if not settings.CLOUD_DEPLOYMENT: context.log.info("Skipping Slack notification in non-prod environment") return diff --git a/posthog/dags/tests/test_flag_evaluations_parity.py b/posthog/dags/tests/test_flag_evaluations_parity.py index 4dfa430eaca2..8279193e3762 100644 --- a/posthog/dags/tests/test_flag_evaluations_parity.py +++ b/posthog/dags/tests/test_flag_evaluations_parity.py @@ -58,6 +58,14 @@ def test_alerts(self, _name: str, checked: list[TeamParity], truncated: int, exp assert lines is not None assert any(expected in line for line in lines) + def test_alerts_when_no_team_is_enabled(self) -> None: + # A cleared allowlist leaves no enabled team, which produces no deficit. Reporting that + # as a clean run would hide the fork having stopped writing everywhere. + results = ParityResults(day=datetime.now(tz=UTC).date(), checked=[], teams_enabled=0, teams_truncated=0) + lines = parity_alert_lines(results) + assert lines is not None + assert any("No team wrote a flag_evaluations row in the lookback window" in line for line in lines) + def test_summary_counts_queried_teams_not_only_teams_with_rows(self) -> None: # Ten teams were queried but only one had rows on the day. The summary must report the # queried count, not len(checked), or it understates coverage and contradicts the @@ -140,3 +148,11 @@ def insert_events(client: Client) -> None: assert by_team[excess_team] == TeamParity(excess_team, only_in_events=0, only_in_flag_evaluations=1, in_both=0) # Switched on during the checked day, so it is skipped rather than reported as a deficit. assert activation_team not in by_team + + # A run capped by max_teams has to report the teams it left out, or it reads as full + # coverage. Only the counts are asserted, because ORDER BY count() DESC does not settle + # which of the equally-sized teams the cap drops. + capped = measure_flag_evaluations_parity( + dagster.build_op_context(), FlagEvaluationsParityConfig(max_teams=2), cluster + ) + assert (capped.teams_enabled, capped.teams_truncated) == (3, 1)