feat(flags): deliver flag calls to destinations on flag_evaluations_mode 2

docker/kafka/topics.txt

@@ -54,3 +54,4 @@ heatmaps_ingestion
 ingestion-clientwarnings-main-1
 log_entries
 plugin_log_entries
+realtime_only_events_json

nodejs/src/cdp/config.ts

@@ -53,6 +53,10 @@ export type CdpConfig = ClickhouseConfig & {
     CDP_LEGACY_EVENT_CONSUMER_TOPIC: string
     CDP_LEGACY_EVENT_CONSUMER_INCLUDE_WEBHOOKS: boolean
 
+    CDP_EVENTS_CONSUMER_TOPIC: string
+    CDP_EVENTS_CONSUMER_GROUP_ID: string
+    CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC: string
+
     CDP_CYCLOTRON_BATCH_DELAY_MS: number
     CDP_CYCLOTRON_HEARTBEAT_INTERVAL_MS: number
     CDP_CYCLOTRON_INSERT_MAX_BATCH_SIZE: number
@@ -272,6 +276,10 @@ export function getDefaultCdpConfig(): CdpConfig {
         CDP_LEGACY_EVENT_CONSUMER_TOPIC: KAFKA_EVENTS_JSON,
         CDP_LEGACY_EVENT_CONSUMER_INCLUDE_WEBHOOKS: false,
 
+        CDP_EVENTS_CONSUMER_TOPIC: KAFKA_EVENTS_JSON,
+        CDP_EVENTS_CONSUMER_GROUP_ID: 'cdp-processed-events-consumer',
+        CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC: '',
+
         CDP_CYCLOTRON_BATCH_DELAY_MS: 50,
         CDP_CYCLOTRON_HEARTBEAT_INTERVAL_MS: 10000,
         CDP_CYCLOTRON_INSERT_MAX_BATCH_SIZE: 100,

nodejs/src/cdp/consumers/cdp-events.consumer.ts

@@ -1,6 +1,5 @@
 import { Message } from 'node-rdkafka'
 
-import { KAFKA_EVENTS_JSON } from '~/common/config/kafka-topics'
 import { KafkaConsumerInterface, createKafkaConsumer } from '~/common/kafka/consumer'
 import { instrumentFn, instrumented } from '~/common/tracing/tracing-utils'
 import { parseJSON } from '~/common/utils/json-parse'
@@ -38,8 +37,8 @@ export class CdpEventsConsumer<
         config: TConfig,
         deps: CdpConsumerBaseDeps,
         jobQueues: { hogQueue: JobQueue; hogflowQueue: JobQueue },
-        topic: string = KAFKA_EVENTS_JSON,
-        groupId: string = 'cdp-processed-events-consumer'
+        topic: string = config.CDP_EVENTS_CONSUMER_TOPIC,
+        groupId: string = config.CDP_EVENTS_CONSUMER_GROUP_ID
     ) {
         super(config, deps)
         this.hogQueue = jobQueues.hogQueue

nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.test.ts

@@ -259,6 +259,28 @@ describe('CdpHogflowSubscriptionMatcherConsumer', () => {
             expect(byTopic[KAFKA_CDP_INTERNAL_EVENTS]).toMatchObject({ 'metadata.broker.list': 'cyclotron:9092' })
             expect(byTopic[KAFKA_EVENTS_JSON]?.['metadata.broker.list']).toBeUndefined()
         })
+
+        it.each(['', 'realtime_only_events_json'])(
+            'creates the realtime-only events consumer only when its topic is configured (%j)',
+            (topic) => {
+                const createConsumer = jest.mocked(createKafkaConsumer)
+                createConsumer.mockClear()
+                new CdpHogflowSubscriptionMatcherConsumer(
+                    {
+                        CYCLOTRON_NODE_DATABASE_URL: 'postgres://test',
+                        CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC: topic,
+                    } as any,
+                    {} as any
+                )
+                const realtimeOnlyTopics = createConsumer.mock.calls
+                    .filter(
+                        ([config]) =>
+                            config.groupId === 'cdp-hogflow-subscription-matcher-realtime-only-events-consumer'
+                    )
+                    .map(([config]) => config.topic)
+                expect(realtimeOnlyTopics).toEqual(topic ? [topic] : [])
+            }
+        )
     })
 
     describe('wakeMatchingWorkflows', () => {

nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.ts

@@ -208,6 +208,9 @@ export class CdpHogflowSubscriptionMatcherConsumer<
     // merged-away person's distinct_ids onto the survivor here; we consume it to re-key parked waits onto
     // the survivor's id so the survivor's person/event updates can wake them.
     private personDistinctIdKafkaConsumer: KafkaConsumerInterface
+    // realtime_only_events_json carries the $feature_flag_called events that ingestion stops writing to
+    // clickhouse_events_json for a FLAG_EVALUATIONS_ONLY organization.
+    private realtimeOnlyEventsKafkaConsumer?: KafkaConsumerInterface
     private cyclotronPool: Pool
     private watcherTeamsRefreshTimer: NodeJS.Timeout | null = null
     // Teams with at least one unexpired watcher. Empty until the first refresh, which start() awaits
@@ -249,6 +252,15 @@ export class CdpHogflowSubscriptionMatcherConsumer<
             },
             startAtLatest
         )
+        if (config.CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC) {
+            this.realtimeOnlyEventsKafkaConsumer = createKafkaConsumer(
+                {
+                    groupId: 'cdp-hogflow-subscription-matcher-realtime-only-events-consumer',
+                    topic: config.CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC,
+                },
+                startAtLatest
+            )
+        }
 
         // The matcher does nothing but read/write cyclotron_jobs, so a missing connection
         // string means it would silently consume the event stream and wake nothing. Fail
@@ -1154,14 +1166,10 @@ export class CdpHogflowSubscriptionMatcherConsumer<
         this.watcherTeamsRefreshTimer.unref()
         // Surface failures to each kafka consumer so the offset doesn't advance past a batch we
         // couldn't match. The pod will crash and replay; the SELECT is read-only and the UPDATE
-        // (with `status = 'available'` guards) is idempotent, so replay is safe. All three streams
+        // (with `status = 'available'` guards) is idempotent, so replay is safe. All the streams
         // funnel into the same input-agnostic wakeMatchingWorkflows via processBatch.
         await Promise.all([
-            this.kafkaConsumer.connect(async (messages) => {
-                return await instrumentFn('cdpHogflowSubscriptionMatcher.handleEachBatch', async () => {
-                    return { backgroundTask: this.processBatch(await this._parseKafkaBatch(messages), 'events') }
-                })
-            }),
+            this.kafkaConsumer.connect(this.eventsBatchHandler('cdpHogflowSubscriptionMatcher.handleEachBatch')),
             this.personKafkaConsumer.connect(async (messages) => {
                 return await instrumentFn('cdpHogflowSubscriptionMatcher.handlePersonBatch', async () => {
                     return { backgroundTask: this.processBatch(await this._parsePersonBatch(messages), 'person') }
@@ -1188,9 +1196,21 @@ export class CdpHogflowSubscriptionMatcherConsumer<
                     })
                 )
             }),
+            this.realtimeOnlyEventsKafkaConsumer?.connect(
+                this.eventsBatchHandler('cdpHogflowSubscriptionMatcher.handleRealtimeOnlyEventsBatch')
+            ),
         ])
     }
 
+    private eventsBatchHandler(
+        instrumentKey: string
+    ): (messages: Message[]) => Promise<{ backgroundTask: Promise<void> }> {
+        return async (messages) =>
+            await instrumentFn(instrumentKey, async () => ({
+                backgroundTask: this.processBatch(await this._parseKafkaBatch(messages), 'events'),
+            }))
+    }
+
     public override async stop(): Promise<void> {
         logger.info('💤', `Stopping ${this.name}...`)
         if (this.watcherTeamsRefreshTimer) {
@@ -1201,6 +1221,7 @@ export class CdpHogflowSubscriptionMatcherConsumer<
             this.personKafkaConsumer.disconnect(),
             this.internalEventsKafkaConsumer.disconnect(),
             this.personDistinctIdKafkaConsumer.disconnect(),
+            this.realtimeOnlyEventsKafkaConsumer?.disconnect(),
         ])
         await this.cyclotronPool.end()
         await super.stop()
@@ -1214,6 +1235,7 @@ export class CdpHogflowSubscriptionMatcherConsumer<
             this.personKafkaConsumer.isHealthy(),
             this.internalEventsKafkaConsumer.isHealthy(),
             this.personDistinctIdKafkaConsumer.isHealthy(),
+            ...(this.realtimeOnlyEventsKafkaConsumer ? [this.realtimeOnlyEventsKafkaConsumer.isHealthy()] : []),
         ]
         return results.find((r) => r.status !== 'ok') ?? results[0]
     }

nodejs/src/cdp/utils.test.ts

@@ -1,10 +1,12 @@
 import { DateTime } from 'luxon'
 
 import { Team } from '../types'
+import { createIncomingEvent } from './_tests/fixtures'
 import { CdpInternalEvent } from './schema'
 import { LogEntry, MinimalLogEntry } from './types'
 import {
     convertInternalEventToHogFunctionInvocationGlobals,
+    convertToHogFunctionInvocationGlobals,
     createAddLogFunction,
     fixLogDeduplication,
     getSensitiveValues,
@@ -56,6 +58,23 @@ describe('Utils', () => {
         }
     )
 
+    test.each([
+        ['$feature_flag_called', '?event=%24feature_flag_called'],
+        ['$pageview', ''],
+    ])('names the event in the event link of %s only when it is a flag call', (eventName, expectedQuery) => {
+        const uuid = '018f0000-0000-7000-8000-000000000002'
+
+        const globals = convertToHogFunctionInvocationGlobals(
+            createIncomingEvent(1, { uuid, event: eventName }),
+            { id: 1, name: 'Test project' } as Team,
+            'https://us.posthog.com'
+        )
+
+        expect(globals.event.url).toBe(
+            `https://us.posthog.com/project/1/events/${uuid}/${encodeURIComponent(globals.event.timestamp)}${expectedQuery}`
+        )
+    })
+
     describe('gzip compressions', () => {
         it("should compress and decompress a string using gzip's sync functions", async () => {
             const input = { foo: 'bar', foo2: 'bar' }

nodejs/src/cdp/utils.ts

@@ -79,6 +79,7 @@ export function convertToHogFunctionInvocationGlobals(
 
     const eventTimestamp = parseClickHouseEventTimestamp(event.timestamp)!
     const eventCapturedAt = parseClickHouseEventTimestamp(event.captured_at)
+    const eventUrl = `${projectUrl}/events/${encodeURIComponent(event.uuid)}/${encodeURIComponent(eventTimestamp)}`
 
     const context: HogFunctionInvocationGlobals = {
         project: {
@@ -94,7 +95,12 @@ export function convertToHogFunctionInvocationGlobals(
             properties,
             timestamp: eventTimestamp,
             captured_at: eventCapturedAt,
-            url: `${projectUrl}/events/${encodeURIComponent(event.uuid)}/${encodeURIComponent(eventTimestamp)}`,
+            // The events page reads flag calls from flag_evaluations only when the link names the event.
+            // On flag_evaluations_mode 2 a flag call has no events row, so a link without the name finds nothing.
+            url:
+                event.event === '$feature_flag_called'
+                    ? `${eventUrl}?event=${encodeURIComponent(event.event)}`
+                    : eventUrl,
         },
         person,
     }

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

@@ -69,6 +69,7 @@ export const KAFKA_CLICKHOUSE_HEATMAP_EVENTS = `${prefix}clickhouse_heatmap_even
 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}`
+export const KAFKA_REALTIME_ONLY_EVENTS_JSON = `${prefix}realtime_only_events_json${suffix}`
 
 // log entries for ingestion into ClickHouse
 export const KAFKA_LOG_ENTRIES = `${prefix}log_entries${suffix}`

nodejs/src/common/outputs/index.ts

@@ -11,6 +11,9 @@ 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 REALTIME_ONLY_EVENTS_OUTPUT = 'realtime_only_events' as const
+export type RealtimeOnlyEventsOutput = typeof REALTIME_ONLY_EVENTS_OUTPUT
+
 export const ASYNC_OUTPUT = 'async' as const
 export type AsyncOutput = typeof ASYNC_OUTPUT
 

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

@@ -14,6 +14,7 @@ describe('FlagEvaluationsService', () => {
         INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS: excludedTeams,
         INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED: false,
         INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: topic,
+        INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC: '',
     })
 
     describe('createFlagEvaluationsService', () => {
@@ -64,4 +65,18 @@ describe('FlagEvaluationsService', () => {
             }
         )
     })
+
+    describe('routesToRealtimeOnlyEvents', () => {
+        it.each([
+            ['realtime_only_events_json', true],
+            ['', false],
+        ])('INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC=%j -> %s', (topic, expected) => {
+            const service = createFlagEvaluationsService({
+                ...envConfig('dual_write'),
+                INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC: topic,
+            })
+
+            expect(service?.routesToRealtimeOnlyEvents).toBe(expected)
+        })
+    })
 })

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

@@ -11,6 +11,7 @@ export interface FlagEvaluationsConfig {
     excludedTeams: number[]
     /** Keeps the events writes for teams on FLAG_EVALUATIONS_ONLY. */
     flagEvaluationsOnlyDisabled: boolean
+    routesToRealtimeOnlyEvents: boolean
 }
 
 /**
@@ -31,12 +32,14 @@ export interface FlagEvaluationsConfig {
  * set fails to start at all while the topic is unreachable. Clear the topic too.
  */
 export class FlagEvaluationsService {
+    readonly routesToRealtimeOnlyEvents: boolean
     private isEnabled: ValueMatcher<number>
     private flagEvaluationsOnlyDisabled: boolean
 
     constructor(config: FlagEvaluationsConfig) {
         this.isEnabled = buildTeamGate(config.teams, config.excludedTeams)
         this.flagEvaluationsOnlyDisabled = config.flagEvaluationsOnlyDisabled
+        this.routesToRealtimeOnlyEvents = config.routesToRealtimeOnlyEvents
     }
 
     isEnabledForTeam(teamId: number): boolean {
@@ -57,7 +60,10 @@ export type FlagEvaluationsEnvConfig = Pick<
     | 'INGESTION_FLAG_EVALUATIONS_EXCLUDED_TEAMS'
     | 'INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED'
 > &
-    Pick<IngestionOutputsConfig, 'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC'>
+    Pick<
+        IngestionOutputsConfig,
+        'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC' | 'INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC'
+    >
 
 /**
  * Builds the flag evaluations service, or undefined when the fork is off. The
@@ -96,5 +102,6 @@ export function createFlagEvaluationsService(envConfig: FlagEvaluationsEnvConfig
         teams,
         excludedTeams,
         flagEvaluationsOnlyDisabled: envConfig.INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED,
+        routesToRealtimeOnlyEvents: envConfig.INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC !== '',
     })
 }