From b2c22e9229bf66d77471c2cb20c33e5433cd4ed8 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 11:52:02 +0100 Subject: [PATCH 1/2] remove _HANDOFF from topic --- src/ingestion/topics.ts | 2 +- src/server.ts | 4 ++-- tests/clickhouse/e2e.test.ts | 4 ++-- tests/clickhouse/process-event.test.ts | 4 ++-- tests/helpers/kafka.ts | 6 +++--- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/src/ingestion/topics.ts b/src/ingestion/topics.ts index 05c6de36..16916752 100644 --- a/src/ingestion/topics.ts +++ b/src/ingestion/topics.ts @@ -3,4 +3,4 @@ export const KAFKA_PERSON = 'clickhouse_person' export const KAFKA_PERSON_UNIQUE_ID = 'clickhouse_person_unique_id' export const KAFKA_SESSION_RECORDING_EVENTS = 'clickhouse_session_recording_events' export const KAFKA_EVENTS_WAL = 'events_write_ahead_log' -export const KAFKA_EVENTS_INGESTION_HANDOFF = 'events_ingestion_handoff' +export const KAFKA_EVENTS_INGESTION = 'events_ingestion' diff --git a/src/server.ts b/src/server.ts index 9442ce31..29d3445e 100644 --- a/src/server.ts +++ b/src/server.ts @@ -20,7 +20,7 @@ import { startSchedule } from './services/schedule' import { ConnectionOptions } from 'tls' import { DB } from './db' import { DateTime } from 'luxon' -import { KAFKA_EVENTS_INGESTION_HANDOFF, KAFKA_EVENTS_WAL } from './ingestion/topics' +import { KAFKA_EVENTS_INGESTION, KAFKA_EVENTS_WAL } from './ingestion/topics' export async function createServer( config: Partial = {}, @@ -86,7 +86,7 @@ export async function createServer( if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) { // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard serverConfig.KAFKA_CONSUMPTION_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION - ? KAFKA_EVENTS_INGESTION_HANDOFF + ? KAFKA_EVENTS_INGESTION : KAFKA_EVENTS_WAL } diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index 51ac85f2..b099784b 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -9,7 +9,7 @@ import { delay, UUIDT } from '../../src/utils' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { delayUntilEventIngested } from '../shared/process-event' -import { KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_INGESTION } from '../../src/ingestion/topics' jest.setTimeout(60000) // 60 sec timeout @@ -18,7 +18,7 @@ const extraServerConfig: Partial = { KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', WORKER_CONCURRENCY: 2, PLUGIN_SERVER_INGESTION: true, - KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION, LOG_LEVEL: LogLevel.Log, } diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 41167ddf..9e93367c 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -2,7 +2,7 @@ import { PluginsServerConfig, Event } from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { createProcessEventTests } from '../shared/process-event' -import { KAFKA_EVENTS_INGESTION_HANDOFF } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_INGESTION } from '../../src/ingestion/topics' jest.setTimeout(180_000) // 3 minute timeout @@ -10,7 +10,7 @@ const extraServerConfig: Partial = { KAFKA_ENABLED: true, KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', PLUGIN_SERVER_INGESTION: true, - KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION_HANDOFF, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION, } describe('process event (clickhouse)', () => { diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 7d1b535a..9bcd31ad 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -2,7 +2,7 @@ import { Kafka, logLevel } from 'kafkajs' import { PluginsServerConfig } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { defaultConfig, overrideWithEnv } from '../../src/config' -import { KAFKA_EVENTS_INGESTION_HANDOFF, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_INGESTION, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' /** Clear the kafka queue */ export async function resetKafka(extraServerConfig: Partial, delayMs = 2000) { @@ -36,7 +36,7 @@ export async function resetKafka(extraServerConfig: Partial } } - await createTopic(KAFKA_EVENTS_INGESTION_HANDOFF) + await createTopic(KAFKA_EVENTS_INGESTION) await createTopic(KAFKA_SESSION_RECORDING_EVENTS) const connected = await new Promise(async (resolve, reject) => { @@ -54,7 +54,7 @@ export async function resetKafka(extraServerConfig: Partial await producer.connect() console.info('subscribing consumer') - await consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION_HANDOFF }) + await consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION }) console.info('running consumer') await consumer.run({ eachMessage: async (payload) => { From 3b80c73c9b930fd846cae91cb539e45076f85bd8 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Thu, 4 Feb 2021 12:21:03 +0100 Subject: [PATCH 2/2] add plugin_ to plugin server ingestion topic --- src/ingestion/topics.ts | 2 +- src/server.ts | 4 ++-- tests/clickhouse/e2e.test.ts | 4 ++-- tests/clickhouse/process-event.test.ts | 4 ++-- tests/helpers/kafka.ts | 6 +++--- 5 files changed, 10 insertions(+), 10 deletions(-) diff --git a/src/ingestion/topics.ts b/src/ingestion/topics.ts index 16916752..a76767cb 100644 --- a/src/ingestion/topics.ts +++ b/src/ingestion/topics.ts @@ -3,4 +3,4 @@ export const KAFKA_PERSON = 'clickhouse_person' export const KAFKA_PERSON_UNIQUE_ID = 'clickhouse_person_unique_id' export const KAFKA_SESSION_RECORDING_EVENTS = 'clickhouse_session_recording_events' export const KAFKA_EVENTS_WAL = 'events_write_ahead_log' -export const KAFKA_EVENTS_INGESTION = 'events_ingestion' +export const KAFKA_EVENTS_PLUGIN_INGESTION = 'events_plugin_ingestion' diff --git a/src/server.ts b/src/server.ts index 29d3445e..b05bec21 100644 --- a/src/server.ts +++ b/src/server.ts @@ -20,7 +20,7 @@ import { startSchedule } from './services/schedule' import { ConnectionOptions } from 'tls' import { DB } from './db' import { DateTime } from 'luxon' -import { KAFKA_EVENTS_INGESTION, KAFKA_EVENTS_WAL } from './ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION, KAFKA_EVENTS_WAL } from './ingestion/topics' export async function createServer( config: Partial = {}, @@ -86,7 +86,7 @@ export async function createServer( if (!serverConfig.KAFKA_CONSUMPTION_TOPIC) { // When ingesting events, listen to the "INGESTION_HANDOFF" topic, otherwise listen to the "WAL" and discard serverConfig.KAFKA_CONSUMPTION_TOPIC = serverConfig.PLUGIN_SERVER_INGESTION - ? KAFKA_EVENTS_INGESTION + ? KAFKA_EVENTS_PLUGIN_INGESTION : KAFKA_EVENTS_WAL } diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index b099784b..78996480 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -9,7 +9,7 @@ import { delay, UUIDT } from '../../src/utils' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { delayUntilEventIngested } from '../shared/process-event' -import { KAFKA_EVENTS_INGESTION } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' jest.setTimeout(60000) // 60 sec timeout @@ -18,7 +18,7 @@ const extraServerConfig: Partial = { KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', WORKER_CONCURRENCY: 2, PLUGIN_SERVER_INGESTION: true, - KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION, LOG_LEVEL: LogLevel.Log, } diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 9e93367c..a6a9b6a6 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -2,7 +2,7 @@ import { PluginsServerConfig, Event } from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { createProcessEventTests } from '../shared/process-event' -import { KAFKA_EVENTS_INGESTION } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' jest.setTimeout(180_000) // 3 minute timeout @@ -10,7 +10,7 @@ const extraServerConfig: Partial = { KAFKA_ENABLED: true, KAFKA_HOSTS: process.env.KAFKA_HOSTS || 'kafka:9092', PLUGIN_SERVER_INGESTION: true, - KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_INGESTION, + KAFKA_CONSUMPTION_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION, } describe('process event (clickhouse)', () => { diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 9bcd31ad..0bf3cff5 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -2,7 +2,7 @@ import { Kafka, logLevel } from 'kafkajs' import { PluginsServerConfig } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { defaultConfig, overrideWithEnv } from '../../src/config' -import { KAFKA_EVENTS_INGESTION, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION, KAFKA_SESSION_RECORDING_EVENTS } from '../../src/ingestion/topics' /** Clear the kafka queue */ export async function resetKafka(extraServerConfig: Partial, delayMs = 2000) { @@ -36,7 +36,7 @@ export async function resetKafka(extraServerConfig: Partial } } - await createTopic(KAFKA_EVENTS_INGESTION) + await createTopic(KAFKA_EVENTS_PLUGIN_INGESTION) await createTopic(KAFKA_SESSION_RECORDING_EVENTS) const connected = await new Promise(async (resolve, reject) => { @@ -54,7 +54,7 @@ export async function resetKafka(extraServerConfig: Partial await producer.connect() console.info('subscribing consumer') - await consumer.subscribe({ topic: KAFKA_EVENTS_INGESTION }) + await consumer.subscribe({ topic: KAFKA_EVENTS_PLUGIN_INGESTION }) console.info('running consumer') await consumer.run({ eachMessage: async (payload) => {