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/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/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. 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(