From 591ce722180fccb905de1d3a29fdb8500a7b7066 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 3 Feb 2021 14:40:37 +0100 Subject: [PATCH 1/5] Optimize Kafka queue offset handling --- src/ingestion/kafka-queue.ts | 86 ++++++++++++++++++++++++++---------- 1 file changed, 63 insertions(+), 23 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 26d1cc69..b7214b41 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -1,13 +1,50 @@ import * as Sentry from '@sentry/node' import { Kafka, Consumer, Message, EachBatchPayload } from 'kafkajs' -import { PluginsServer, Queue, RawEventMessage } from 'types' +import { EventMessage, PluginsServer, Queue, RawEventMessage } from 'types' import { KAFKA_EVENTS_INGESTION_HANDOFF } from './topics' import { PluginEvent } from '@posthog/plugin-scaffold' import { status } from '../status' import { killGracefully } from '../utils' +import { parseRawEventMessage } from './utils' export type BatchCallback = (messages: Message[]) => Promise +/** + * We use this to avoid a situation where the event was discarded and because of that its Kafka offset is not resolved, + * potentially causing unnecessary message retries. + * + * Works in the following way: for each event UUID the algorithm sets the Kafka offset to resolve + * based on the last event after it from the Kafka topic that WAS discarded. Example, we've got events: + * ```JS + * [{ uuid: 'a', offset: 1 }, { uuid: 's', offset: 2 }, { uuid: 'd', offset: 3 }, { uuid: 'f', offset: 4 }] + * ``` + * Now some plugin discards the last two! Returned map for use in resolveOffset() will in result look like this: + * ```JS + * { 'a': 1, 's': 4, 'd': 4, 'f': 4} + * ``` + * Because 'd' and 'f' were discarded by a plugin, when we save 's' to the database, we'll know that we also in fact + * covered 'd' and 'f' and should resolve the last offset - belonging to 'f' – not an intermediary one. + * As for 'a', we simply resolve its offset outright, because its next event ('s') is not discarded. + */ +function aliasEventUuidForDiscardedKafkaOffsets( + rawEventMessages: RawEventMessage[], + processedEvents: PluginEvent[] +): Map { + rawEventMessages = [...rawEventMessages] // Shallow copy to avoid side effects + const eventUuidToKafkaOffset = new Map() + const processedUuids: Set = new Set(processedEvents.map((event) => event.uuid!)) + rawEventMessages.reverse() + // This initial value below is in fact the last message, due to the array being reversed + let nextOffsetNotDiscarded: string | null = rawEventMessages.shift()!.kafka_offset + for (const rawEventMessage of rawEventMessages) { + if (processedUuids.has(rawEventMessage.uuid)) { + nextOffsetNotDiscarded = rawEventMessage.kafka_offset + } + eventUuidToKafkaOffset.set(rawEventMessage.uuid, nextOffsetNotDiscarded) + } + return eventUuidToKafkaOffset +} + export class KafkaQueue implements Queue { private pluginsServer: PluginsServer private kafka: Kafka @@ -34,25 +71,20 @@ export class KafkaQueue implements Queue { resolveOffset, heartbeat, commitOffsetsIfNecessary, - uncommittedOffsets, isRunning, isStale, }: EachBatchPayload): Promise { const batchProcessingTimer = new Date() - const rawEvents: RawEventMessage[] = batch.messages.map((message) => ({ + const rawEventMessages: RawEventMessage[] = batch.messages.map((message) => ({ ...JSON.parse(message.value!.toString()), kafka_offset: message.offset, })) - const parsedEvents = rawEvents.map((rawEvent) => ({ - ...rawEvent, - data: JSON.parse(rawEvent.data), - })) - const pluginEvents: PluginEvent[] = rawEvents.map((rawEvent) => { - const { data: dataStr, ...restOfRawEvent } = rawEvent - const event = { ...restOfRawEvent, ...JSON.parse(dataStr) } + const parsedEventMessages: EventMessage[] = rawEventMessages.map(parseRawEventMessage) + const pluginEvents: PluginEvent[] = rawEventMessages.map((rawEventMessage) => { + const { data: dataString, kafka_offset: kafkaOffset, ...restOfRawEventMessage } = rawEventMessage + const event = { ...restOfRawEventMessage, ...JSON.parse(dataString) } return { ...event, - kafka_offset: restOfRawEvent.kafka_offset, site_url: event.site_url || null, ip: event.ip || null, } @@ -60,21 +92,29 @@ export class KafkaQueue implements Queue { const processedEvents: PluginEvent[] = ( await this.processEventBatch(pluginEvents) ).filter((event: PluginEvent[] | false | null | undefined) => Boolean(event)) - for (const event of processedEvents) { - if (!isRunning()) { - status.info('😮', 'Consumer not running anymore, canceling batch processing!') - return - } - if (isStale()) { - status.info('😮', 'Batch stale, canceling batch processing!') - return + if (processedEvents.length) { + const eventUuidToKafkaOffset = aliasEventUuidForDiscardedKafkaOffsets(rawEventMessages, processedEvents) + for (const event of processedEvents) { + if (!isRunning()) { + status.info('😮', 'Consumer not running anymore, canceling batch processing!') + return + } + if (isStale()) { + status.info('😮', 'Batch stale, canceling batch processing!') + return + } + const singleIngestionTimer = new Date() + await this.saveEvent(event) + resolveOffset(eventUuidToKafkaOffset.get(event.uuid!)!) + await heartbeat() + await commitOffsetsIfNecessary() + this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } - const singleIngestionTimer = new Date() - await this.saveEvent(event) - resolveOffset(event.kafka_offset!) + } else { + // If all events were discarded in plugin processing, just resolve the final offset and do nothing else + resolveOffset(rawEventMessages[rawEventMessages.length - 1].kafka_offset) await heartbeat() await commitOffsetsIfNecessary() - this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } this.pluginsServer.statsd?.timing('kafka_queue.each_batch', batchProcessingTimer) } From 39536acd5912928a8ae218c0ce33a1479d90f921 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 3 Feb 2021 14:44:51 +0100 Subject: [PATCH 2/5] Update explanation --- src/ingestion/kafka-queue.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index b7214b41..ce06ca5d 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -20,10 +20,10 @@ export type BatchCallback = (messages: Message[]) => Promise * ``` * Now some plugin discards the last two! Returned map for use in resolveOffset() will in result look like this: * ```JS - * { 'a': 1, 's': 4, 'd': 4, 'f': 4} + * { 'a': 1, 's': 4, 'd': 4, 'f': 4 } * ``` - * Because 'd' and 'f' were discarded by a plugin, when we save 's' to the database, we'll know that we also in fact - * covered 'd' and 'f' and should resolve the last offset - belonging to 'f' – not an intermediary one. + * Because 'd' and 'f' were discarded by a plugin, when we save 's' to the database, we'll know that at the same time + * we in fact also covered 'd' and 'f' and should resolve the last offset - belonging to 'f' – not an intermediary one. * As for 'a', we simply resolve its offset outright, because its next event ('s') is not discarded. */ function aliasEventUuidForDiscardedKafkaOffsets( From 6feda229854bcd5cca5f98f7c00c2b2b0d4335d2 Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 3 Feb 2021 15:41:28 +0100 Subject: [PATCH 3/5] Improve clarity and typing --- src/ingestion/kafka-queue.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index ce06ca5d..00a2d9f6 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -35,12 +35,12 @@ function aliasEventUuidForDiscardedKafkaOffsets( const processedUuids: Set = new Set(processedEvents.map((event) => event.uuid!)) rawEventMessages.reverse() // This initial value below is in fact the last message, due to the array being reversed - let nextOffsetNotDiscarded: string | null = rawEventMessages.shift()!.kafka_offset + let currentNotDiscardedOffset: string = rawEventMessages.shift()!.kafka_offset for (const rawEventMessage of rawEventMessages) { if (processedUuids.has(rawEventMessage.uuid)) { - nextOffsetNotDiscarded = rawEventMessage.kafka_offset + currentNotDiscardedOffset = rawEventMessage.kafka_offset } - eventUuidToKafkaOffset.set(rawEventMessage.uuid, nextOffsetNotDiscarded) + eventUuidToKafkaOffset.set(rawEventMessage.uuid, currentNotDiscardedOffset) } return eventUuidToKafkaOffset } From 7dc24c9f1b21d139b1af4db82b1859ea669c9e7d Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 3 Feb 2021 16:24:56 +0100 Subject: [PATCH 4/5] =?UTF-8?q?Revert=20to=20original=20kafka-queue=20to?= =?UTF-8?q?=20test=20tests=20=F0=9F=A4=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/ingestion/kafka-queue.ts | 107 ++++++++++------------------------- 1 file changed, 30 insertions(+), 77 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 00a2d9f6..957b381a 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -1,50 +1,12 @@ import * as Sentry from '@sentry/node' import { Kafka, Consumer, Message, EachBatchPayload } from 'kafkajs' -import { EventMessage, PluginsServer, Queue, RawEventMessage } from 'types' -import { KAFKA_EVENTS_INGESTION_HANDOFF } from './topics' +import { PluginsServer, Queue, RawEventMessage } from 'types' +import { KAFKA_EVENTS_WAL } from './topics' import { PluginEvent } from '@posthog/plugin-scaffold' import { status } from '../status' -import { killGracefully } from '../utils' -import { parseRawEventMessage } from './utils' export type BatchCallback = (messages: Message[]) => Promise -/** - * We use this to avoid a situation where the event was discarded and because of that its Kafka offset is not resolved, - * potentially causing unnecessary message retries. - * - * Works in the following way: for each event UUID the algorithm sets the Kafka offset to resolve - * based on the last event after it from the Kafka topic that WAS discarded. Example, we've got events: - * ```JS - * [{ uuid: 'a', offset: 1 }, { uuid: 's', offset: 2 }, { uuid: 'd', offset: 3 }, { uuid: 'f', offset: 4 }] - * ``` - * Now some plugin discards the last two! Returned map for use in resolveOffset() will in result look like this: - * ```JS - * { 'a': 1, 's': 4, 'd': 4, 'f': 4 } - * ``` - * Because 'd' and 'f' were discarded by a plugin, when we save 's' to the database, we'll know that at the same time - * we in fact also covered 'd' and 'f' and should resolve the last offset - belonging to 'f' – not an intermediary one. - * As for 'a', we simply resolve its offset outright, because its next event ('s') is not discarded. - */ -function aliasEventUuidForDiscardedKafkaOffsets( - rawEventMessages: RawEventMessage[], - processedEvents: PluginEvent[] -): Map { - rawEventMessages = [...rawEventMessages] // Shallow copy to avoid side effects - const eventUuidToKafkaOffset = new Map() - const processedUuids: Set = new Set(processedEvents.map((event) => event.uuid!)) - rawEventMessages.reverse() - // This initial value below is in fact the last message, due to the array being reversed - let currentNotDiscardedOffset: string = rawEventMessages.shift()!.kafka_offset - for (const rawEventMessage of rawEventMessages) { - if (processedUuids.has(rawEventMessage.uuid)) { - currentNotDiscardedOffset = rawEventMessage.kafka_offset - } - eventUuidToKafkaOffset.set(rawEventMessage.uuid, currentNotDiscardedOffset) - } - return eventUuidToKafkaOffset -} - export class KafkaQueue implements Queue { private pluginsServer: PluginsServer private kafka: Kafka @@ -71,50 +33,42 @@ export class KafkaQueue implements Queue { resolveOffset, heartbeat, commitOffsetsIfNecessary, + uncommittedOffsets, isRunning, isStale, }: EachBatchPayload): Promise { const batchProcessingTimer = new Date() - const rawEventMessages: RawEventMessage[] = batch.messages.map((message) => ({ + const rawEvents: RawEventMessage[] = batch.messages.map((message) => ({ ...JSON.parse(message.value!.toString()), kafka_offset: message.offset, })) - const parsedEventMessages: EventMessage[] = rawEventMessages.map(parseRawEventMessage) - const pluginEvents: PluginEvent[] = rawEventMessages.map((rawEventMessage) => { - const { data: dataString, kafka_offset: kafkaOffset, ...restOfRawEventMessage } = rawEventMessage - const event = { ...restOfRawEventMessage, ...JSON.parse(dataString) } - return { - ...event, - site_url: event.site_url || null, - ip: event.ip || null, - } - }) + const parsedEvents = rawEvents.map((rawEvent) => ({ + ...rawEvent, + data: JSON.parse(rawEvent.data), + })) + const pluginEvents: PluginEvent[] = parsedEvents.map((parsedEvent) => ({ + ...parsedEvent, + event: parsedEvent.data.event, + properties: parsedEvent.data.properties, + })) const processedEvents: PluginEvent[] = ( await this.processEventBatch(pluginEvents) ).filter((event: PluginEvent[] | false | null | undefined) => Boolean(event)) - if (processedEvents.length) { - const eventUuidToKafkaOffset = aliasEventUuidForDiscardedKafkaOffsets(rawEventMessages, processedEvents) - for (const event of processedEvents) { - if (!isRunning()) { - status.info('😮', 'Consumer not running anymore, canceling batch processing!') - return - } - if (isStale()) { - status.info('😮', 'Batch stale, canceling batch processing!') - return - } - const singleIngestionTimer = new Date() - await this.saveEvent(event) - resolveOffset(eventUuidToKafkaOffset.get(event.uuid!)!) - await heartbeat() - await commitOffsetsIfNecessary() - this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) + for (const event of processedEvents) { + if (!isRunning()) { + status.info('😮', 'Consumer not running anymore, canceling batch processing!') + return + } + if (isStale()) { + status.info('😮', 'Batch stale, canceling batch processing!') + return } - } else { - // If all events were discarded in plugin processing, just resolve the final offset and do nothing else - resolveOffset(rawEventMessages[rawEventMessages.length - 1].kafka_offset) + const singleIngestionTimer = new Date() + await this.saveEvent(event) + resolveOffset(event.kafka_offset!) await heartbeat() await commitOffsetsIfNecessary() + this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer) } this.pluginsServer.statsd?.timing('kafka_queue.each_batch', batchProcessingTimer) } @@ -125,7 +79,7 @@ export class KafkaQueue implements Queue { this.consumer.on(this.consumer.events.CRASH, ({ payload: { error } }) => reject(error)) status.info('⏬', `Connecting Kafka consumer to ${this.pluginsServer.KAFKA_HOSTS}...`) this.wasConsumerRan = true - await this.consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) + await this.consumer.subscribe({ topic: KAFKA_EVENTS_WAL }) // KafkaJS batching: https://kafka.js.org/docs/consuming#a-name-each-batch-a-eachbatch await this.consumer.run({ // TODO: eachBatchAutoResolve: false, // don't autoresolve whole batch in case we exit it early @@ -144,7 +98,7 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Pausing Kafka consumer...') - await this.consumer.pause([{ topic: KAFKA_EVENTS_INGESTION_HANDOFF }]) + await this.consumer.pause([{ topic: KAFKA_EVENTS_WAL }]) status.info('⏸', 'Kafka consumer paused!') } @@ -153,12 +107,12 @@ export class KafkaQueue implements Queue { return } status.info('⏳', 'Resuming Kafka consumer...') - await this.consumer.resume([{ topic: KAFKA_EVENTS_INGESTION_HANDOFF }]) + await this.consumer.resume([{ topic: KAFKA_EVENTS_WAL }]) status.info('▶️', 'Kafka consumer resumed!') } isPaused(): boolean { - return this.consumer.paused().some(({ topic }) => topic === KAFKA_EVENTS_INGESTION_HANDOFF) + return this.consumer.paused().some(({ topic }) => topic === KAFKA_EVENTS_WAL) } async stop(): Promise { @@ -176,7 +130,7 @@ export class KafkaQueue implements Queue { private static buildConsumer(kafka: Kafka): Consumer { const consumer = kafka.consumer({ - groupId: 'clickhouse-ingestion', + groupId: 'plugin-server', readUncommitted: false, }) const { GROUP_JOIN, CRASH, CONNECT, DISCONNECT } = consumer.events @@ -186,7 +140,6 @@ export class KafkaQueue implements Queue { consumer.on(CRASH, ({ payload: { error, groupId } }) => { status.error('⚠️', `Kafka consumer group ${groupId} crashed:\n`, error) Sentry.captureException(error) - killGracefully() }) consumer.on(CONNECT, () => { status.info('✅', 'Kafka consumer connected!') From 16d594ca3fdb79c87508b69c837e511deaedc66d Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Wed, 3 Feb 2021 16:45:14 +0100 Subject: [PATCH 5/5] Simplify optimization --- src/ingestion/kafka-queue.ts | 45 ++++++++++++++++++++++++++++++++---- 1 file changed, 41 insertions(+), 4 deletions(-) diff --git a/src/ingestion/kafka-queue.ts b/src/ingestion/kafka-queue.ts index 957b381a..02783165 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/ingestion/kafka-queue.ts @@ -7,6 +7,42 @@ import { status } from '../status' export type BatchCallback = (messages: Message[]) => Promise +/** + * We use this to avoid a situation where the event was discarded and because of that its Kafka offset is not resolved, + * potentially causing unnecessary message retries. + * + * Works in the following way: for each event UUID the algorithm sets the Kafka offset to resolve + * based on the last event after it from the Kafka topic that WAS discarded. Example, we've got events: + * ```JS + * [{ uuid: 'a', offset: 1 }, { uuid: 's', offset: 2 }, { uuid: 'd', offset: 3 }, { uuid: 'f', offset: 4 }] + * ``` + * Now some plugin discards the last two! Returned map for use in resolveOffset() will in result look like this: + * ```JS + * { 'a': 1, 's': 4, 'd': 4, 'f': 4 } + * ``` + * Because 'd' and 'f' were discarded by a plugin, when we save 's' to the database, we'll know that at the same time + * we in fact also covered 'd' and 'f' and should resolve the last offset - belonging to 'f' – not an intermediary one. + * As for 'a', we simply resolve its offset outright, because its next event ('s') is not discarded. + */ +function aliasEventUuidForDiscardedKafkaOffsets( + rawEventMessages: RawEventMessage[], + processedEvents: PluginEvent[] +): Map { + rawEventMessages = [...rawEventMessages] // Shallow copy to avoid side effects + const eventUuidToKafkaOffset = new Map() + const processedUuids: Set = new Set(processedEvents.map((event) => event.uuid!)) + rawEventMessages.reverse() + // This initial value below is in fact the last message, due to the array being reversed + let currentNotDiscardedOffset: string = rawEventMessages.shift()!.kafka_offset + for (const rawEventMessage of rawEventMessages) { + if (processedUuids.has(rawEventMessage.uuid)) { + currentNotDiscardedOffset = rawEventMessage.kafka_offset + } + eventUuidToKafkaOffset.set(rawEventMessage.uuid, currentNotDiscardedOffset) + } + return eventUuidToKafkaOffset +} + export class KafkaQueue implements Queue { private pluginsServer: PluginsServer private kafka: Kafka @@ -38,15 +74,15 @@ export class KafkaQueue implements Queue { isStale, }: EachBatchPayload): Promise { const batchProcessingTimer = new Date() - const rawEvents: RawEventMessage[] = batch.messages.map((message) => ({ + const rawEventMessages: RawEventMessage[] = batch.messages.map((message) => ({ ...JSON.parse(message.value!.toString()), kafka_offset: message.offset, })) - const parsedEvents = rawEvents.map((rawEvent) => ({ + const parsedEventMessages = rawEventMessages.map((rawEvent) => ({ ...rawEvent, data: JSON.parse(rawEvent.data), })) - const pluginEvents: PluginEvent[] = parsedEvents.map((parsedEvent) => ({ + const pluginEvents: PluginEvent[] = parsedEventMessages.map((parsedEvent) => ({ ...parsedEvent, event: parsedEvent.data.event, properties: parsedEvent.data.properties, @@ -54,6 +90,7 @@ export class KafkaQueue implements Queue { const processedEvents: PluginEvent[] = ( await this.processEventBatch(pluginEvents) ).filter((event: PluginEvent[] | false | null | undefined) => Boolean(event)) + const eventUuidToKafkaOffset = aliasEventUuidForDiscardedKafkaOffsets(rawEventMessages, processedEvents) for (const event of processedEvents) { if (!isRunning()) { status.info('😮', 'Consumer not running anymore, canceling batch processing!') @@ -65,7 +102,7 @@ export class KafkaQueue implements Queue { } const singleIngestionTimer = new Date() await this.saveEvent(event) - resolveOffset(event.kafka_offset!) + resolveOffset(eventUuidToKafkaOffset.get(event.uuid!)!) await heartbeat() await commitOffsetsIfNecessary() this.pluginsServer.statsd?.timing('kafka_queue.single_ingestion', singleIngestionTimer)