From 7d1831122bfd69656991d243b58ee05fc057e539 Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Wed, 7 Oct 2026 14:59:32 -0700 Subject: [PATCH 1/2] feat(flags): deliver flag calls to destinations on flag_evaluations_mode 2 On FLAG_EVALUATIONS_ONLY, the flag-evaluations fork sends each forked $feature_flag_called to a new realtime_only_events output when INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC is set. ClickHouse does not read that topic, so the call stays out of the events table while the CDP consumers can still deliver it. With the topic unset, the fork removes the call from emission. CdpEventsConsumer takes its topic and group from CDP_EVENTS_CONSUMER_TOPIC and CDP_EVENTS_CONSUMER_GROUP_ID, so a second deployment can read the new topic. The hogflow subscription matcher reads it when CDP_HOGFLOW_SUBSCRIPTION_MATCHER_REALTIME_ONLY_EVENTS_TOPIC is set. Claude-Session: https://claude.ai/code/session_01EvZDsoSPrQzLQit6kYQydV --- docker/kafka/topics.txt | 1 + nodejs/src/cdp/config.ts | 8 ++ .../src/cdp/consumers/cdp-events.consumer.ts | 5 +- ...flow-subscription-matcher.consumer.test.ts | 22 ++++ ...p-hogflow-subscription-matcher.consumer.ts | 34 ++++-- nodejs/src/common/config/kafka-topics.ts | 1 + nodejs/src/common/outputs/index.ts | 3 + .../flag-evaluations-service.test.ts | 15 +++ .../flag-evaluations-service.ts | 9 +- .../fork-flag-evaluations-step.test.ts | 101 ++++++++++++------ .../fork-flag-evaluations-step.ts | 32 ++++-- nodejs/src/ingestion/config.ts | 11 ++ nodejs/src/ingestion/ingestion-consumer.ts | 6 +- .../pipelines/analytics/event-subpipeline.ts | 10 +- .../analytics/joined-ingestion-pipeline.ts | 2 + .../pipelines/analytics/outputs/registry.ts | 5 + nodejs/tests/helpers/ingestion-outputs.ts | 7 ++ nodejs/tests/helpers/kafka.ts | 2 + .../ingestion/flag-evaluations-e2e.test.ts | 20 +++- posthog/kafka_client/topics.py | 1 + 20 files changed, 238 insertions(+), 57 deletions(-) diff --git a/docker/kafka/topics.txt b/docker/kafka/topics.txt index cec348ea44c8..40380d9fcb86 100644 --- a/docker/kafka/topics.txt +++ b/docker/kafka/topics.txt @@ -54,3 +54,4 @@ heatmaps_ingestion ingestion-clientwarnings-main-1 log_entries plugin_log_entries +realtime_only_events_json diff --git a/nodejs/src/cdp/config.ts b/nodejs/src/cdp/config.ts index 989a22974424..88f0ed5a1877 100644 --- a/nodejs/src/cdp/config.ts +++ b/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, diff --git a/nodejs/src/cdp/consumers/cdp-events.consumer.ts b/nodejs/src/cdp/consumers/cdp-events.consumer.ts index f7d367318a80..afb57209d8d8 100644 --- a/nodejs/src/cdp/consumers/cdp-events.consumer.ts +++ b/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 diff --git a/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.test.ts b/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.test.ts index 79030d0b0468..24e7561ca5e3 100644 --- a/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.test.ts +++ b/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', () => { diff --git a/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.ts b/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.ts index d349bfa6d7e0..756d8b88d0e0 100644 --- a/nodejs/src/cdp/consumers/cdp-hogflow-subscription-matcher.consumer.ts +++ b/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 }> { + return async (messages) => + await instrumentFn(instrumentKey, async () => ({ + backgroundTask: this.processBatch(await this._parseKafkaBatch(messages), 'events'), + })) + } + public override async stop(): Promise { 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] } diff --git a/nodejs/src/common/config/kafka-topics.ts b/nodejs/src/common/config/kafka-topics.ts index 5bb4224ce450..0814e69b6abb 100644 --- a/nodejs/src/common/config/kafka-topics.ts +++ b/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}` diff --git a/nodejs/src/common/outputs/index.ts b/nodejs/src/common/outputs/index.ts index 31bbdf7af718..2cefe6bd55d9 100644 --- a/nodejs/src/common/outputs/index.ts +++ b/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 diff --git a/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.test.ts b/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.test.ts index 5922a6730bf8..e05fe461570e 100644 --- a/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.test.ts +++ b/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) + }) + }) }) diff --git a/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.ts b/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.ts index 5ca0a9d67fe7..7414252e0015 100644 --- a/nodejs/src/ingestion/common/flag-evaluations/flag-evaluations-service.ts +++ b/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 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 + 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 !== '', }) } diff --git a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.test.ts b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.test.ts index b6bbf19ccdb0..2e5e4341762f 100644 --- a/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.test.ts +++ b/nodejs/src/ingestion/common/steps/event-processing/fork-flag-evaluations-step.test.ts @@ -1,6 +1,6 @@ import { DateTime } from 'luxon' -import { FlagEvaluationsOutput, IngestionWarningsOutput } from '~/common/outputs' +import { FlagEvaluationsOutput, IngestionWarningsOutput, RealtimeOnlyEventsOutput } from '~/common/outputs' import { MessageSizeTooLarge } from '~/common/utils/db/error' import { parseJSON } from '~/common/utils/json-parse' import { FlagEvaluationsService } from '~/ingestion/common/flag-evaluations/flag-evaluations-service' @@ -19,13 +19,17 @@ import { // Real instances: the service is a pure, synchronous config gate, so mocking // it would only hide drift from the real class. Team 7 matches createInput. -const enabledService = (flagEvaluationsOnlyDisabled = false) => - new FlagEvaluationsService({ teams: '*', excludedTeams: [], flagEvaluationsOnlyDisabled }) -const teamExcludedService = () => - new FlagEvaluationsService({ teams: '*', excludedTeams: [7], flagEvaluationsOnlyDisabled: false }) +const enabledService = ({ + excludedTeams = [] as number[], + flagEvaluationsOnlyDisabled = false, + routesToRealtimeOnlyEvents = true, +} = {}) => + new FlagEvaluationsService({ teams: '*', excludedTeams, flagEvaluationsOnlyDisabled, routesToRealtimeOnlyEvents }) const createStep = (service: FlagEvaluationsService) => { - const outputs = createMockIngestionOutputs() + const outputs = createMockIngestionOutputs< + FlagEvaluationsOutput | IngestionWarningsOutput | RealtimeOnlyEventsOutput + >() const step = createForkFlagEvaluationsStep(outputs, service) return { step, outputs } } @@ -64,6 +68,9 @@ const createInput = ( team: { flag_evaluations_mode: mode }, }) +const emitted = (eventsToEmit: EventToEmit[]) => + eventsToEmit.map(({ event, output }) => ({ uuid: event.uuid, output })) + describe('createForkFlagEvaluationsStep', () => { beforeEach(() => { flagEvaluationsEventsTotal.reset() @@ -79,7 +86,7 @@ describe('createForkFlagEvaluationsStep', () => { }, { name: 'the team is not enabled', - buildService: () => teamExcludedService(), + buildService: () => enabledService({ excludedTeams: [7] }), buildInput: () => createInput(), }, ])('passes through without producing when $name', async ({ buildService, buildInput }) => { @@ -265,14 +272,23 @@ describe('createForkFlagEvaluationsStep', () => { { name: 'a FLAG_EVALUATIONS_ONLY', mode: FlagEvaluationsMode.FlagEvaluationsOnly, + routesToRealtimeOnlyEvents: false, outcome: 'lost_message_too_large', producedOutputs: ['flag_evaluations', 'ingestion_warnings'], warnings: [{ type: 'message_size_too_large', eventUuid: 'event-uuid-1' }], }, + { + name: 'a routed FLAG_EVALUATIONS_ONLY', + mode: FlagEvaluationsMode.FlagEvaluationsOnly, + routesToRealtimeOnlyEvents: true, + outcome: 'lost_message_too_large', + producedOutputs: ['flag_evaluations'], + warnings: [], + }, ])( 'does not block the batch when the row exceeds the broker message limit for $name team', - async ({ mode, outcome, producedOutputs, warnings }) => { - const { step, outputs } = createStep(enabledService()) + async ({ mode, routesToRealtimeOnlyEvents, outcome, producedOutputs, warnings }) => { + const { step, outputs } = createStep(enabledService({ routesToRealtimeOnlyEvents })) outputs.queueMessages.mockRejectedValueOnce( new MessageSizeTooLarge('too large', new Error('too large')) ) @@ -328,28 +344,45 @@ describe('createForkFlagEvaluationsStep', () => { }) describe('FLAG_EVALUATIONS_ONLY mode', () => { - it('removes the queued $feature_flag_called event from eventsToEmit and keeps the $experiment_exposure copy', async () => { - const { step, outputs } = createStep(enabledService()) - const input = createInput( - [createProcessedEvent(), createExposureDuplicate()], - FlagEvaluationsMode.FlagEvaluationsOnly - ) + it.each([ + { + name: 'removes the queued $feature_flag_called event from eventsToEmit', + routesToRealtimeOnlyEvents: false, + expectedEventsToEmit: [{ uuid: 'dup-uuid', output: 'events' }], + }, + { + name: 'sends the queued $feature_flag_called event to the realtime_only_events output', + routesToRealtimeOnlyEvents: true, + expectedEventsToEmit: [ + { uuid: 'event-uuid-1', output: 'realtime_only_events' }, + { uuid: 'dup-uuid', output: 'events' }, + ], + }, + ])( + '$name and keeps the $experiment_exposure copy on events', + async ({ routesToRealtimeOnlyEvents, expectedEventsToEmit }) => { + const { step, outputs } = createStep(enabledService({ routesToRealtimeOnlyEvents })) + const input = createInput( + [createProcessedEvent(), createExposureDuplicate()], + FlagEvaluationsMode.FlagEvaluationsOnly + ) - const result = await step(input) + const result = await step(input) - expect(isOkResult(result)).toBe(true) - if (isOkResult(result)) { - expect(result.value.eventsToEmit.map(({ event }) => event.uuid)).toEqual(['dup-uuid']) - await Promise.all(result.sideEffects) + expect(isOkResult(result)).toBe(true) + if (isOkResult(result)) { + expect(emitted(result.value.eventsToEmit)).toEqual(expectedEventsToEmit) + await Promise.all(result.sideEffects) + } + // For a FLAG_EVALUATIONS_ONLY team the flag_evaluations row is the only stored copy of the call. + expect(outputs.queueMessages).toHaveBeenCalledTimes(1) + expect(outputs.queueMessages.mock.calls[0][1].map((message) => message.key)).toEqual(['event-uuid-1']) + // The counter records one outcome per event, so the call must not also count as dual_written. + expect((await flagEvaluationsEventsTotal.get()).values).toEqual([ + expect.objectContaining({ labels: { outcome: 'flag_evaluations_only' }, value: 1 }), + ]) } - // For a FLAG_EVALUATIONS_ONLY team the flag_evaluations row is the only copy of the call. - expect(outputs.queueMessages).toHaveBeenCalledTimes(1) - expect(outputs.queueMessages.mock.calls[0][1].map((message) => message.key)).toEqual(['event-uuid-1']) - // The counter records one outcome per event, so the removed call must not also count as dual_written. - expect((await flagEvaluationsEventsTotal.get()).values).toEqual([ - expect.objectContaining({ labels: { outcome: 'flag_evaluations_only' }, value: 1 }), - ]) - }) + ) it.each([ { @@ -365,12 +398,12 @@ describe('createForkFlagEvaluationsStep', () => { { name: 'INGESTION_FLAG_EVALUATIONS_ONLY_DISABLED is on', mode: FlagEvaluationsMode.FlagEvaluationsOnly, - buildService: () => enabledService(true), + buildService: () => enabledService({ flagEvaluationsOnlyDisabled: true }), }, { name: 'the team is not enabled', mode: FlagEvaluationsMode.FlagEvaluationsOnly, - buildService: () => teamExcludedService(), + buildService: () => enabledService({ excludedTeams: [7] }), }, { name: 'the flag key is missing', @@ -394,7 +427,7 @@ describe('createForkFlagEvaluationsStep', () => { }), }, ])( - 'keeps the $feature_flag_called event in eventsToEmit when $name', + 'keeps the $feature_flag_called event on the events output when $name', async ({ mode, buildService, flagCalledOverrides, arrange }) => { const deps = createStep(buildService()) arrange?.(deps) @@ -404,9 +437,9 @@ describe('createForkFlagEvaluationsStep', () => { expect(isOkResult(result)).toBe(true) if (isOkResult(result)) { - expect(result.value.eventsToEmit.map(({ event }) => event.uuid)).toEqual([ - 'event-uuid-1', - 'dup-uuid', + expect(emitted(result.value.eventsToEmit)).toEqual([ + { uuid: 'event-uuid-1', output: 'events' }, + { uuid: 'dup-uuid', output: 'events' }, ]) await Promise.all(result.sideEffects) } 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 c96fbc43e39e..2c2590241cf0 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,6 +1,12 @@ import { Counter, Gauge } from 'prom-client' -import { FLAG_EVALUATIONS_OUTPUT, FlagEvaluationsOutput, IngestionWarningsOutput } from '~/common/outputs' +import { + FLAG_EVALUATIONS_OUTPUT, + FlagEvaluationsOutput, + IngestionWarningsOutput, + REALTIME_ONLY_EVENTS_OUTPUT, + RealtimeOnlyEventsOutput, +} from '~/common/outputs' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' import { IngestionOutputMessage } from '~/common/outputs/types' import { MessageSizeTooLarge } from '~/common/utils/db/error' @@ -77,8 +83,9 @@ export interface ForkFlagEvaluationsStepInput { * narrows the assembled events row to the flag_evaluations column set and * produces it to the clickhouse_flag_evaluations topic. The event continues to * the events table unchanged, except for a team whose organization is on - * FLAG_EVALUATIONS_ONLY: this step removes each event it queued for that team - * from eventsToEmit. + * FLAG_EVALUATIONS_ONLY: this step moves each event it queued for that team + * to the realtime-only output, which only the CDP consumers read. Without a + * topic for that output, the step removes the event from eventsToEmit. * * Sits between create-event and emit-event, where the fully assembled * ProcessedEvent exists: person_id and person_properties from person @@ -89,13 +96,15 @@ export interface ForkFlagEvaluationsStepInput { * * Every failure path before the produce is queued continues toward the events * table. A produce that fails permanently is counted and dropped, never - * retried, so for a FLAG_EVALUATIONS_ONLY team the call is lost. The ack + * retried, so for a FLAG_EVALUATIONS_ONLY team the stored row is lost. The ack * rides as a side effect, so the batch does not commit its offsets until the * broker answers. Pipe this step without a retry envelope, because a retry * would queue the produce again. */ export function createForkFlagEvaluationsStep( - outputs: IngestionOutputs, + // This step never produces to the realtime-only output. It requires it because emit-event produces + // every entry this step retargets, so a pipeline without that output fails to compile. + outputs: IngestionOutputs, flagEvaluationsService: FlagEvaluationsService ): ProcessingStep { return function forkFlagEvaluationsStep(input) { @@ -122,6 +131,7 @@ export function createForkFlagEvaluationsStep { if (event.properties['$set'] || event.properties['$set_once']) { @@ -162,7 +172,7 @@ export function createForkFlagEvaluationsStep !eventsToFork.includes(toEmit)) + const eventsToEmit = removesFromEmission + ? input.eventsToEmit.filter((toEmit) => !eventsToFork.includes(toEmit)) + : input.eventsToEmit.map((toEmit) => + eventsToFork.includes(toEmit) ? { ...toEmit, output: REALTIME_ONLY_EVENTS_OUTPUT } : toEmit + ) return Promise.resolve(ok({ ...input, eventsToEmit }, [settled])) } catch (error) { logger.warn('Failed to fork $feature_flag_called event to flag_evaluations, continuing', { diff --git a/nodejs/src/ingestion/config.ts b/nodejs/src/ingestion/config.ts index 26b7bf30b6b3..9c9b31330637 100644 --- a/nodejs/src/ingestion/config.ts +++ b/nodejs/src/ingestion/config.ts @@ -513,6 +513,9 @@ export type IngestionOutputsConfig = { INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: string INGESTION_OUTPUT_FLAG_EVALUATIONS_PRODUCER: ProducerName + INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC: string + INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_PRODUCER: ProducerName + INGESTION_OUTPUT_HEATMAPS_TOPIC: string INGESTION_OUTPUT_HEATMAPS_PRODUCER: ProducerName @@ -564,6 +567,14 @@ export function getDefaultIngestionOutputsConfig(): IngestionOutputsConfig { // createFlagEvaluationsService. INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: '', INGESTION_OUTPUT_FLAG_EVALUATIONS_PRODUCER: INGESTION_DOWNSTREAM_PRODUCER, + // An empty topic skips the startup topic-existence check. While it is empty, a + // FLAG_EVALUATIONS_ONLY team's flag calls reach no realtime destination, so set it + // on every lane that sets INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC. Enable + // ordering: (1) create the realtime_only_events_json topic, (2) start the CDP + // consumers that read it, (3) set INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC. A + // consumer group that joins after step 3 can start past the first routed calls. + INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC: '', + INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_PRODUCER: INGESTION_DOWNSTREAM_PRODUCER, INGESTION_OUTPUT_HEATMAPS_TOPIC: KAFKA_CLICKHOUSE_HEATMAP_EVENTS, INGESTION_OUTPUT_HEATMAPS_PRODUCER: INGESTION_DOWNSTREAM_PRODUCER, INGESTION_OUTPUT_INGESTION_WARNINGS_TOPIC: KAFKA_INGESTION_WARNINGS, diff --git a/nodejs/src/ingestion/ingestion-consumer.ts b/nodejs/src/ingestion/ingestion-consumer.ts index fb3f95e82eea..50131b3bd16f 100644 --- a/nodejs/src/ingestion/ingestion-consumer.ts +++ b/nodejs/src/ingestion/ingestion-consumer.ts @@ -22,6 +22,7 @@ import { PersonDistinctIdsOutput, PersonMergeEventsOutput, PersonsOutput, + RealtimeOnlyEventsOutput, } from '~/common/outputs' import { IngestionOutputs } from '~/common/outputs/ingestion-outputs' import { PersonRepository } from '~/common/persons/repositories/person-repository' @@ -77,7 +78,9 @@ export type IngestionConsumerFullConfig = IngestionConsumerConfig & // than relying on the runtime shape. Pick< IngestionOutputsConfig, - 'INGESTION_OUTPUT_PERSON_MERGE_EVENTS_TOPIC' | 'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC' + | 'INGESTION_OUTPUT_PERSON_MERGE_EVENTS_TOPIC' + | 'INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC' + | 'INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC' > export interface IngestionConsumerDeps { @@ -88,6 +91,7 @@ export interface IngestionConsumerDeps { outputs: IngestionOutputs< | EventOutput | FlagEvaluationsOutput + | RealtimeOnlyEventsOutput | IngestionWarningsOutput | DlqOutput | OverflowOutput diff --git a/nodejs/src/ingestion/pipelines/analytics/event-subpipeline.ts b/nodejs/src/ingestion/pipelines/analytics/event-subpipeline.ts index 8094eb424cb6..d21710a69d18 100644 --- a/nodejs/src/ingestion/pipelines/analytics/event-subpipeline.ts +++ b/nodejs/src/ingestion/pipelines/analytics/event-subpipeline.ts @@ -41,6 +41,7 @@ import { PersonDistinctIdsOutput, PersonMergeEventsOutput, PersonsOutput, + RealtimeOnlyEventsOutput, } from './outputs' // Mirrors the merge condition in PersonMergeService.handleIdentifyOrAlias: an event asks for a person @@ -69,6 +70,7 @@ export interface EventSubpipelineConfig { outputs: IngestionOutputs< | EventOutput | FlagEvaluationsOutput + | RealtimeOnlyEventsOutput | IngestionWarningsOutput | PersonsOutput | PersonDistinctIdsOutput @@ -171,7 +173,13 @@ export function createEventSubpipeline { ) testWithTeamIngester( - 'FLAG_EVALUATIONS_ONLY team: the call lands only in flag_evaluations, and its person update and $experiment_exposure copy still land', + 'FLAG_EVALUATIONS_ONLY team: the call lands only in flag_evaluations and the realtime-only topic, and its person update and $experiment_exposure copy still land', { teamOverrides: { flag_evaluations_mode: FlagEvaluationsMode.FlagEvaluationsOnly }, pluginServerConfig: { INGESTION_FLAG_EVALUATIONS_MODE: 'dual_write', INGESTION_FLAG_EVALUATIONS_TEAMS: '*', INGESTION_OUTPUT_FLAG_EVALUATIONS_TOPIC: KAFKA_CLICKHOUSE_FLAG_EVALUATIONS, + INGESTION_OUTPUT_REALTIME_ONLY_EVENTS_TOPIC: KAFKA_REALTIME_ONLY_EVENTS_JSON, EXPERIMENT_EXPOSURE_DUPLICATION_TEAMS: '*', }, }, async ({ infra, ingester, team, kafkaProducer, token }) => { + const producerObserver = new KafkaProducerObserver(kafkaProducer) // A variant response makes create-event append the $experiment_exposure copy. const event = new EventBuilder(team) .withEvent('$feature_flag_called') @@ -145,6 +149,18 @@ describe('Flag evaluations shadow-routing E2E', () => { // single-partition topic. A call written to events would already be visible here. expect(events.map((row) => row.event)).toEqual(['$experiment_exposure']) expect(events[0].person_id).toBe(persons[0].uuid) + + const realtimeEvents = producerObserver + .getProducedKafkaMessagesForTopic(KAFKA_REALTIME_ONLY_EVENTS_JSON) + .map((message) => message.value) + expect(realtimeEvents).toEqual([ + expect.objectContaining({ + uuid: event.uuid, + event: '$feature_flag_called', + person_id: persons[0].uuid, + }), + ]) + expect(parseJSON(realtimeEvents[0].person_properties as string)).toMatchObject({ plan: 'pro' }) } ) diff --git a/posthog/kafka_client/topics.py b/posthog/kafka_client/topics.py index 6de01fb71f03..fb59f00a966f 100644 --- a/posthog/kafka_client/topics.py +++ b/posthog/kafka_client/topics.py @@ -39,6 +39,7 @@ KAFKA_CLICKHOUSE_HEATMAP_EVENTS = f"{KAFKA_PREFIX}clickhouse_heatmap_events{SUFFIX}" KAFKA_CLICKHOUSE_AI_EVENTS_JSON = f"{KAFKA_PREFIX}clickhouse_ai_events_json{SUFFIX}" KAFKA_CLICKHOUSE_FLAG_EVALUATIONS = f"{KAFKA_PREFIX}clickhouse_flag_evaluations{SUFFIX}" +KAFKA_REALTIME_ONLY_EVENTS_JSON = f"{KAFKA_PREFIX}realtime_only_events_json{SUFFIX}" # Legacy session recording topic — only retained because clickhouse migration 0063 still # DROPs the old materialized view by name. Producers/consumers live in the Node.js services. From f4c00f671e8c17a6edb322f40e28c4a797a2c6d8 Mon Sep 17 00:00:00 2001 From: Phil Haack Date: Thu, 8 Oct 2026 12:58:57 -0700 Subject: [PATCH 2/2] fix(flags): name the event in links to flag calls 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 destination and workflow log links without the name open an empty page. Links to other events stay unchanged. Claude-Session: https://claude.ai/code/session_01EvZDsoSPrQzLQit6kYQydV --- nodejs/src/cdp/utils.test.ts | 19 +++++++++++++++++++ nodejs/src/cdp/utils.ts | 8 +++++++- .../Workflows/logs/log-utils.test.tsx | 19 +++++++++++++++++++ .../frontend/Workflows/logs/log-utils.tsx | 6 +++++- 4 files changed, 50 insertions(+), 2 deletions(-) diff --git a/nodejs/src/cdp/utils.test.ts b/nodejs/src/cdp/utils.test.ts index 237d4b34d6ac..3c849a9ea442 100644 --- a/nodejs/src/cdp/utils.test.ts +++ b/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' } diff --git a/nodejs/src/cdp/utils.ts b/nodejs/src/cdp/utils.ts index c71429b01c02..fb6181e1e765 100644 --- a/nodejs/src/cdp/utils.ts +++ b/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, } diff --git a/products/workflows/frontend/Workflows/logs/log-utils.test.tsx b/products/workflows/frontend/Workflows/logs/log-utils.test.tsx index 72ab85b32add..5695735b9a64 100644 --- a/products/workflows/frontend/Workflows/logs/log-utils.test.tsx +++ b/products/workflows/frontend/Workflows/logs/log-utils.test.tsx @@ -30,4 +30,23 @@ describe('renderWorkflowLogMessage', () => { expect(link?.getAttribute('target')).toBe('_blank') expect(link?.textContent).toContain('View task') }) + + it.each([ + ['$feature_flag_called', '$feature_flag_called'], + ['$pageview', null], + ])( + 'names the event in the link of an [Event:...] token for %s only when it is a flag call', + (eventName, linkedName) => { + const { container } = render( + renderWorkflowLogMessage( + workflow, + `Workflow woken by [Event:018f0000-0000-7000-8000-000000000002|${eventName}|2026-08-24T00:00:00.000Z]` + ) + ) + + const href = container.querySelector('a')?.getAttribute('href') ?? '' + expect(href).toContain('/events/018f0000-0000-7000-8000-000000000002/') + expect(new URL(href, 'http://localhost').searchParams.get('event')).toBe(linkedName) + } + ) }) diff --git a/products/workflows/frontend/Workflows/logs/log-utils.tsx b/products/workflows/frontend/Workflows/logs/log-utils.tsx index 89904da00792..f929bd61a9b3 100644 --- a/products/workflows/frontend/Workflows/logs/log-utils.tsx +++ b/products/workflows/frontend/Workflows/logs/log-utils.tsx @@ -110,12 +110,16 @@ export const renderWorkflowLogMessage = (workflow: HogFlow, message: string): JS const eventUuid = matchesEventRegex[1] const eventName = matchesEventRegex[2] const eventTimestamp = matchesEventRegex[3] + // The events page reads flag calls from flag_evaluations only when the link names the event. + // Other names stay off the link because the log token strips | from event names. + // A name that contained | would then match no event. + const linkEventName = eventName === '$feature_flag_called' ? eventName : undefined elements.push(