feat(flags): shadow-route $feature_flag_called into flag_evaluations

nodejs/src/common/config/kafka-topics.ts

@@ -67,6 +67,8 @@ export const KAFKA_PERFORMANCE_EVENTS = `${prefix}clickhouse_performance_events$
 export const KAFKA_CLICKHOUSE_HEATMAP_EVENTS = `${prefix}clickhouse_heatmap_events${suffix}`
 // write AI events to ClickHouse
 export const KAFKA_CLICKHOUSE_AI_EVENTS_JSON = `${prefix}clickhouse_ai_events_json${suffix}`
+// write flag evaluations ($feature_flag_called telemetry) to ClickHouse
+export const KAFKA_CLICKHOUSE_FLAG_EVALUATIONS = `${prefix}clickhouse_flag_evaluations${suffix}`
 
 // log entries for ingestion into ClickHouse
 export const KAFKA_LOG_ENTRIES = `${prefix}log_entries${suffix}`

nodejs/src/common/outputs/index.ts

@@ -8,6 +8,9 @@ export type EventOutput = typeof EVENTS_OUTPUT
 export const AI_EVENTS_OUTPUT = 'ai_events' as const
 export type AiEventOutput = typeof AI_EVENTS_OUTPUT
 
+export const FLAG_EVALUATIONS_OUTPUT = 'flag_evaluations' as const
+export type FlagEvaluationsOutput = typeof FLAG_EVALUATIONS_OUTPUT
+
 export const ASYNC_OUTPUT = 'async' as const
 export type AsyncOutput = typeof ASYNC_OUTPUT
 

nodejs/src/ingestion/common/feature-flag-called-dedup/feature-flag-called-dedup-service.ts

@@ -4,8 +4,9 @@ import { Redis } from 'ioredis'
 import { timeoutGuard } from '~/common/utils/db/utils'
 import { parseTeamsList } from '~/common/utils/env-utils'
 import { logger } from '~/common/utils/logger'
+import { buildTeamGate } from '~/ingestion/common/team-gate'
 import { IngestionConsumerConfig, IngestionLane, REALTIME_INGESTION_LANES } from '~/ingestion/config'
-import { RedisPool } from '~/types'
+import { RedisPool, ValueMatcher } from '~/types'
 
 import { featureFlagCalledDedupRedisLatency, featureFlagCalledDedupRedisOpsTotal } from './metrics'
 
@@ -48,8 +49,7 @@ export function parseFeatureFlagCalledDedupConfig(
     }
     const excluded = parseTeamsList(excludedTeams)
     if (excluded === '*') {
-        // An operator reaching for '*' as an exclude-everyone switch means "off";
-        // the escape hatch on a data-dropping feature must not fail toward dropping.
+        // The escape hatch on a data-dropping feature must not fail toward dropping.
         logger.warn('INGESTION_FEATURE_FLAG_CALLED_DEDUP_EXCLUDED_TEAMS is "*", disabling dedup')
         parsedMode = 'disabled'
     }
@@ -207,25 +207,20 @@ export function createFeatureFlagCalledDedupService(
 export class RedisFeatureFlagCalledDedupService implements FeatureFlagCalledDedupService {
     private redisPool: RedisPool
     private config: FeatureFlagCalledDedupConfig
-    private teams: Set<number> | '*'
-    private excludedTeams: Set<number>
+    private isEnabled: ValueMatcher<number>
 
     constructor(options: RedisFeatureFlagCalledDedupServiceOptions) {
         this.redisPool = options.redisPool
         this.config = options.config
-        this.teams = options.config.teams === '*' ? '*' : new Set(options.config.teams)
-        this.excludedTeams = new Set(options.config.excludedTeams)
+        this.isEnabled = buildTeamGate(options.config.teams, options.config.excludedTeams)
     }
 
     get mode(): FeatureFlagCalledDedupMode {
         return this.config.mode
     }
 
     isEnabledForTeam(teamId: number): boolean {
-        if (this.excludedTeams.has(teamId)) {
-            return false
-        }
-        return this.teams === '*' || this.teams.has(teamId)
+        return this.isEnabled(teamId)
     }
 
     async claimKeys(claims: FeatureFlagCalledDedupClaim[]): Promise<boolean[]> {

nodejs/src/ingestion/common/flag-evaluations/flag-evaluation-row.test.ts

@@ -0,0 +1,65 @@
+import { DateTime } from 'luxon'
+
+import { parseJSON } from '~/common/utils/json-parse'
+import { serializeEvent } from '~/ingestion/common/steps/event-processing/emit-event-step'
+import { ISOTimestamp, ProcessedEvent, ProjectId } from '~/types'
+
+import { mapProcessedEventToFlagEvaluationRow } from './flag-evaluation-row'
+
+describe('mapProcessedEventToFlagEvaluationRow', () => {
+    const processedEvent: ProcessedEvent = {
+        uuid: 'event-uuid-1',
+        event: '$feature_flag_called',
+        properties: { $feature_flag: 'my-flag', $feature_flag_response: true, $group_0: 'org-1' },
+        timestamp: '2024-01-15T10:30:00.123Z' as ISOTimestamp,
+        team_id: 42,
+        project_id: 7 as ProjectId,
+        distinct_id: 'distinct-1',
+        elements_chain: '',
+        created_at: DateTime.fromISO('2024-01-15T10:31:00.456Z'),
+        captured_at: null,
+        person_id: 'person-uuid-1',
+        person_properties: { email: 'a@example.com' },
+        person_created_at: DateTime.fromISO('2023-01-01T00:00:00.000Z'),
+        person_mode: 'full',
+    }
+
+    it('narrows the events wire format to the flag_evaluations columns, byte-identical on shared fields', () => {
+        const row = mapProcessedEventToFlagEvaluationRow(processedEvent)
+        const eventsRow = serializeEvent(processedEvent)
+
+        expect(row.uuid).toBe('event-uuid-1')
+        expect(row.event).toBe('$feature_flag_called')
+        expect(row.team_id).toBe(42)
+        expect(row.distinct_id).toBe('distinct-1')
+        expect(row.person_id).toBe('person-uuid-1')
+        expect(row.timestamp).toBe('2024-01-15 10:30:00.123')
+        expect(row.properties).toBe(eventsRow.properties)
+        expect(row.person_properties).toBe(eventsRow.person_properties)
+        // Both rows derive created_at from the event, so they agree. Reading the
+        // wall clock per serialization instead would stamp two different values.
+        expect(row.created_at).toBe(eventsRow.created_at)
+        expect(row.created_at).toBe('2024-01-15 10:31:00.456')
+        expect(parseJSON(row.properties!).$feature_flag).toBe('my-flag')
+        expect(parseJSON(row.person_properties!).email).toBe('a@example.com')
+    })
+
+    it('emits exactly the kafka_flag_evaluations column set', () => {
+        // An extra key (elements_chain, person_mode, a materialized column name)
+        // is not a column on the Kafka table; keep the wire format in lockstep
+        // with posthog/models/flag_evaluations/sql.py.
+        const row = mapProcessedEventToFlagEvaluationRow(processedEvent)
+
+        expect(Object.keys(row).sort()).toEqual([
+            'created_at',
+            'distinct_id',
+            'event',
+            'person_id',
+            'person_properties',
+            'properties',
+            'team_id',
+            'timestamp',
+            'uuid',
+        ])
+    })
+})

nodejs/src/ingestion/common/flag-evaluations/flag-evaluation-row.ts

@@ -0,0 +1,51 @@
+import { serializeEvent } from '~/ingestion/common/steps/event-processing/emit-event-step'
+import { ProcessedEvent, RawKafkaEvent } from '~/types'
+
+/**
+ * One row for the kafka_flag_evaluations ClickHouse table (JSONEachRow).
+ * Keep in sync with posthog/models/flag_evaluations/sql.py.
+ *
+ * The table mirrors the events schema, so the row is the events wire format
+ * narrowed to exactly the columns the Kafka engine table declares. Going
+ * through serializeEvent keeps every shared field byte-identical to what the
+ * events table receives for the same event, rather than a second hand-rolled
+ * mapping that could drift from it.
+ *
+ * Deliberately not sent:
+ * - `inserted_at`: omitted keys zero-fill in ClickHouse, and the MV treats the
+ *   epoch sentinel as "absent" and stamps the Kafka message timestamp instead.
+ * - `group0..group4_properties`: the events producer does not send them either
+ *   (they are not in RawKafkaEvent); they zero-fill empty in both tables.
+ * - `flag_key`, `response`, `session_id`, `request_id`, `$group_0..4`: the Kafka
+ *   table does not declare these, and the shard computes them from `properties`
+ *   as DEFAULT columns. A DEFAULT only fills when the insert omits the column,
+ *   so naming one here would store the sent value even where it contradicts
+ *   `properties`. This row type must never grow them.
+ */
+export type FlagEvaluationRow = Pick<
+    RawKafkaEvent,
+    | 'uuid'
+    | 'event'
+    | 'properties'
+    | 'timestamp'
+    | 'team_id'
+    | 'distinct_id'
+    | 'created_at'
+    | 'person_id'
+    | 'person_properties'
+>
+
+export function mapProcessedEventToFlagEvaluationRow(event: ProcessedEvent): FlagEvaluationRow {
+    const serialized = serializeEvent(event)
+    return {
+        uuid: serialized.uuid,
+        event: serialized.event,
+        properties: serialized.properties,
+        timestamp: serialized.timestamp,
+        team_id: serialized.team_id,
+        distinct_id: serialized.distinct_id,
+        created_at: serialized.created_at,
+        person_id: serialized.person_id,
+        person_properties: serialized.person_properties,
+    }
+}

nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.test.ts

@@ -0,0 +1,55 @@
+import { IngestionLane } from '~/ingestion/config'
+
+import { FlagEvaluationsEnvConfig, createFlagEvaluationsService } from './flag-evaluations-service'
+
+describe('FlagEvaluationsService', () => {
+    const envConfig = (
+        mode: string,
+        teams = '*',
+        excludedTeams = '',
+        topic = 'clickhouse_flag_evaluations',
+        lane: IngestionLane | null = 'main'
+    ): FlagEvaluationsEnvConfig => ({
+        INGESTION_LANE: lane,
+        INGESTION_FLAG_EVALUATIONS_MODE: mode,
+        INGESTION_FLAG_EVALUATIONS_TEAMS: teams,
+        INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS: excludedTeams,
+        INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: topic,
+    })
+
+    describe('createFlagEvaluationsService', () => {
+        // The wildcard-exclusion case is the escape hatch: it must fail toward the
+        // events table, not toward excluding everyone from being excluded. The
+        // empty-topic case guards the fork from producing into a nameless topic.
+        it.each([
+            ['mode is disabled', envConfig('disabled'), false],
+            ['mode is invalid', envConfig('garbage'), false],
+            ['the output topic is empty', envConfig('dual_write', '*', '', ''), false],
+            ['excluded teams is the wildcard', envConfig('dual_write', '*', '*'), false],
+            ['mode is dual_write', envConfig('dual_write'), true],
+            // The backfill owns history, so a delayed lane must never fork even
+            // when the env vars say dual_write.
+            ['the lane is historical', envConfig('dual_write', '*', '', 'topic', 'historical'), false],
+            ['the lane is async', envConfig('dual_write', '*', '', 'topic', 'async'), false],
+            ['the lane is overflow', envConfig('dual_write', '*', '', 'topic', 'overflow'), true],
+            ['no lane is set (local dev)', envConfig('dual_write', '*', '', 'topic', null), true],
+            ['the teams allowlist is empty', envConfig('dual_write', ''), false],
+        ])('builds a service when %s -> %s', (_name, config, expected) => {
+            expect(createFlagEvaluationsService(config) !== undefined).toBe(expected)
+        })
+    })
+
+    describe('isEnabledForTeam', () => {
+        it.each([
+            ['*', '', 5, true],
+            ['*', '5', 5, false],
+            ['1,2', '', 2, true],
+            ['1,2', '', 3, false],
+            ['1,2', '2', 2, false],
+        ])('teams=%s excluded=%s team=%i -> %s', (teams, excluded, teamId, expected) => {
+            const service = createFlagEvaluationsService(envConfig('dual_write', teams, excluded))
+
+            expect(service?.isEnabledForTeam(teamId)).toBe(expected)
+        })
+    })
+})

nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.ts

@@ -0,0 +1,108 @@
+import { parseTeamsList } from '~/common/utils/env-utils'
+import { logger } from '~/common/utils/logger'
+import { buildTeamGate } from '~/ingestion/common/team-gate'
+import {
+    IngestionConsumerConfig,
+    IngestionLane,
+    IngestionOutputsConfig,
+    REALTIME_INGESTION_LANES,
+} from '~/ingestion/config'
+import { ValueMatcher } from '~/types'
+
+export interface FlagEvaluationsConfig {
+    /** '*' for all teams, or an explicit allowlist of team IDs. */
+    teams: number[] | '*'
+    /** Escape hatch: teams never forked, even when `teams` is '*'. */
+    excludedTeams: number[]
+}
+
+/**
+ * Gate for the $feature_flag_called fork that shadow-writes flag evaluations to
+ * the ClickHouse flag_evaluations table (via the clickhouse_flag_evaluations
+ * topic) while the event continues to the events table unchanged.
+ *
+ * The shadow write is never load-bearing for an individual event, but the batch
+ * does not commit its offsets until the broker answers. See
+ * createForkFlagEvaluationsStep for the ack contract and
+ * flagEvaluationsPendingAcks for the stall it can cause.
+ *
+ * Shedding that dependency during an incident takes BOTH env vars, not just the
+ * mode: the mode stops a running consumer from forking, but startup topic
+ * verification is driven by the output registration, which the mode does not
+ * touch. A pod that restarts with INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC still
+ * set fails to start at all while the topic is unreachable. Clear the topic too.
+ */
+export class FlagEvaluationsService {
+    private isEnabled: ValueMatcher<number>
+
+    constructor(config: FlagEvaluationsConfig) {
+        this.isEnabled = buildTeamGate(config.teams, config.excludedTeams)
+    }
+
+    isEnabledForTeam(teamId: number): boolean {
+        return this.isEnabled(teamId)
+    }
+}
+
+export type FlagEvaluationsEnvConfig = Pick<
+    IngestionConsumerConfig,
+    | 'INGESTION_LANE'
+    | 'INGESTION_FLAG_EVALUATIONS_MODE'
+    | 'INGESTION_FLAG_EVALUATIONS_TEAMS'
+    | 'INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS'
+> &
+    Pick<IngestionOutputsConfig, 'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC'>
+
+/**
+ * Lanes that may fork: the real-time lanes plus `null` for local dev, where no
+ * lane is set. Derived from REALTIME_INGESTION_LANES so a new real-time lane is
+ * covered automatically; delayed lanes are excluded by construction.
+ *
+ * The backfill owns history for this table, so a delayed lane that also forked
+ * would write rows the backfill already covers, double-counting the per-flag
+ * usage the table exists to answer. The gate holds even if the env vars leak
+ * into a config shared across lanes.
+ */
+const FLAG_EVALUATIONS_ALLOWED_LANES: readonly (IngestionLane | null)[] = [...REALTIME_INGESTION_LANES, null]
+
+/**
+ * Builds the flag evaluations service, or undefined when the fork is off. The
+ * pipeline composes the fork step out entirely when this returns undefined.
+ */
+export function createFlagEvaluationsService(envConfig: FlagEvaluationsEnvConfig): FlagEvaluationsService | undefined {
+    const mode = envConfig.INGESTION_FLAG_EVALUATIONS_MODE
+    if (mode !== 'dual_write') {
+        if (mode !== 'disabled') {
+            logger.warn('Invalid INGESTION_FLAG_EVALUATIONS_MODE, falling back to disabled', { mode })
+        }
+        return undefined
+    }
+    if (!envConfig.INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC) {
+        // An empty topic also skips the startup topic-existence check, so producing
+        // here would fail at runtime instead. See the enable ordering in config.ts.
+        logger.warn(
+            'INGESTION_FLAG_EVALUATIONS_MODE is set but INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC is empty, not forking'
+        )
+        return undefined
+    }
+    if (!FLAG_EVALUATIONS_ALLOWED_LANES.includes(envConfig.INGESTION_LANE)) {
+        logger.warn('Flag evaluations fork is not supported on this ingestion lane, disabling', {
+            lane: envConfig.INGESTION_LANE,
+        })
+        return undefined
+    }
+    const excludedTeams = parseTeamsList(envConfig.INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS)
+    if (excludedTeams === '*') {
+        // An operator's '*' exclusion means "off": the escape hatch fails toward not forking.
+        logger.warn('INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS is "*", disabling the flag evaluations fork')
+        return undefined
+    }
+    const teams = parseTeamsList(envConfig.INGESTION_FLAG_EVALUATIONS_TEAMS)
+    if (teams !== '*' && teams.length === 0) {
+        // Same shape as the empty-topic case: an allowlist naming nobody is off, so
+        // compose the step out rather than paying it per event to gate every team away.
+        logger.warn('INGESTION_FLAG_EVALUATIONS_MODE is set but INGESTION_FLAG_EVALUATIONS_TEAMS is empty, not forking')
+        return undefined
+    }
+    return new FlagEvaluationsService({ teams, excludedTeams })
+}

nodejs/src/ingestion/common/steps/event-processing/create-event-step.test.ts

@@ -253,7 +253,7 @@ describe('create-event-step', () => {
             if (isOkResult(result)) {
                 const event = result.value.eventsToEmit[0].event
                 expect(event.timestamp).toBeTruthy()
-                expect(event.created_at).toBeNull()
+                expect(event.created_at).toBeTruthy()
                 expect(event.person_created_at).toBeTruthy()
             }
         })

nodejs/src/ingestion/common/steps/event-processing/create-event.ts

@@ -1,3 +1,4 @@
+import { DateTime } from 'luxon'
 import { Counter } from 'prom-client'
 
 import { MAX_GROUP_TYPES_PER_TEAM } from '~/common/groups/group-type-manager'
@@ -98,7 +99,7 @@ export function createEvent(
         project_id: projectId,
         distinct_id: distinctId,
         elements_chain: elementsChain,
-        created_at: null,
+        created_at: DateTime.utc(),
         captured_at: capturedAt,
         person_id: personId,
         person_properties: eventPersonProperties,

nodejs/src/ingestion/common/steps/event-processing/emit-event-step.test.ts

@@ -62,7 +62,7 @@ describe('emit-event-step', () => {
             project_id: 1 as ProjectId,
             distinct_id: 'test-distinct-id',
             elements_chain: '',
-            created_at: null,
+            created_at: DateTime.fromISO('2023-01-01T00:01:00.000Z'),
             captured_at: null,
             person_id: 'person-uuid',
             person_properties: {},

nodejs/src/ingestion/common/steps/event-processing/emit-event-step.ts

@@ -123,7 +123,7 @@ export function serializeEvent(event: ProcessedEvent): RawKafkaEvent {
         project_id: event.project_id,
         distinct_id: safeClickhouseString(event.distinct_id),
         elements_chain: safeClickhouseString(event.elements_chain),
-        created_at: castTimestampOrNow(null, TimestampFormat.ClickHouse),
+        created_at: castTimestampOrNow(event.created_at, TimestampFormat.ClickHouse),
         captured_at:
             event.captured_at !== null
                 ? castTimestampToClickhouseFormat(DateTime.fromJSDate(event.captured_at), TimestampFormat.ClickHouse)