From 55f346d7b75d84ea6135fde1745713e449392182 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 00:21:51 +0200 Subject: [PATCH 01/11] mostly remove processEventBatch --- benchmarks/clickhouse/e2e.kafka.benchmark.ts | 11 ++-- benchmarks/vm/worker.benchmark.ts | 26 +------- src/main/ingestion-queues/queue.ts | 5 -- src/worker/plugins/run.ts | 65 -------------------- src/worker/tasks.ts | 5 +- src/worker/vm/lazy.ts | 4 -- src/worker/vm/vm.ts | 18 +----- tests/plugins.test.ts | 35 ++++------- tests/postgres/queue.test.ts | 1 - tests/postgres/vm.lazy.test.ts | 2 - tests/postgres/vm.test.ts | 38 +----------- tests/postgres/worker.test.ts | 16 +---- 12 files changed, 23 insertions(+), 203 deletions(-) diff --git a/benchmarks/clickhouse/e2e.kafka.benchmark.ts b/benchmarks/clickhouse/e2e.kafka.benchmark.ts index ac3027e8..9637dea5 100644 --- a/benchmarks/clickhouse/e2e.kafka.benchmark.ts +++ b/benchmarks/clickhouse/e2e.kafka.benchmark.ts @@ -31,13 +31,10 @@ describe('e2e kafka & clickhouse benchmark', () => { beforeEach(async () => { await resetTestDatabase(` - async function processEventBatch (batch) { - // console.log(\`Received batch of \${batch.length} events\`) - return batch.map(event => { - event.properties.processed = 'hell yes' - event.properties.upperUuid = event.properties.uuid?.toUpperCase() - return event - }) + async function processEvent (event) { + event.properties.processed = 'hell yes' + event.properties.upperUuid = event.properties.uuid?.toUpperCase() + return event } `) await resetKafka(extraServerConfig) diff --git a/benchmarks/vm/worker.benchmark.ts b/benchmarks/vm/worker.benchmark.ts index 676e3ad5..a49aed05 100644 --- a/benchmarks/vm/worker.benchmark.ts +++ b/benchmarks/vm/worker.benchmark.ts @@ -27,36 +27,16 @@ function processOneEvent( return processEvent(defaultEvent) } -function processOneBatch( - processEventBatch: (batch: PluginEvent[]) => Promise, - batchSize: number, - batchIndex: number -): Promise { - const events = [...Array(batchSize)].map((_, i) => ({ - distinct_id: 'my_id', - ip: '127.0.0.1', - site_url: 'http://localhost', - team_id: 2, - now: new Date().toISOString(), - event: 'default event', - properties: { key: 'value', batchIndex, indexInBatch: i }, - })) - - return processEventBatch(events) -} - async function processCountEvents(piscina: ReturnType, count: number, batchSize = 1) { const maxPromises = 1000 const promises = Array(maxPromises) const processEvent = (event: PluginEvent) => piscina.runTask({ task: 'processEvent', args: { event } }) - const processEventBatch = (batch: PluginEvent[]) => piscina.runTask({ task: 'processEventBatch', args: { batch } }) - const groups = Math.ceil(count / maxPromises) + const groups = Math.ceil((count * batchSize) / maxPromises) for (let j = 0; j < groups; j++) { - const groupCount = groups === 1 ? count : j === groups - 1 ? count % maxPromises : maxPromises + const groupCount = groups === 1 ? count : j === groups - 1 ? (count * batchSize) % maxPromises : maxPromises for (let i = 0; i < groupCount; i++) { - promises[i] = - batchSize === 1 ? processOneEvent(processEvent, i) : processOneBatch(processEventBatch, batchSize, i) + promises[i] = processOneEvent(processEvent, i) } await Promise.all(promises) } diff --git a/src/main/ingestion-queues/queue.ts b/src/main/ingestion-queues/queue.ts index 99a156e1..d4f869db 100644 --- a/src/main/ingestion-queues/queue.ts +++ b/src/main/ingestion-queues/queue.ts @@ -40,11 +40,6 @@ export async function startQueue( server.lastActivityType = 'processEvent' return piscina.runTask({ task: 'processEvent', args: { event } }) }, - processEventBatch: (batch: PluginEvent[]) => { - server.lastActivity = new Date().valueOf() - server.lastActivityType = 'processEventBatch' - return piscina.runTask({ task: 'processEventBatch', args: { batch } }) - }, ingestEvent: (event: PluginEvent) => { server.lastActivity = new Date().valueOf() server.lastActivityType = 'ingestEvent' diff --git a/src/worker/plugins/run.ts b/src/worker/plugins/run.ts index 5f02ecd2..632ef078 100644 --- a/src/worker/plugins/run.ts +++ b/src/worker/plugins/run.ts @@ -98,71 +98,6 @@ export async function runProcessEvent(server: Hub, event: PluginEvent): Promise< return returnedEvent } -export async function runProcessEventBatch(server: Hub, batch: PluginEvent[]): Promise { - const eventsByTeam = new Map() - - for (const event of batch) { - if (eventsByTeam.has(event.team_id)) { - eventsByTeam.get(event.team_id)!.push(event) - } else { - eventsByTeam.set(event.team_id, [event]) - } - } - - let allReturnedEvents: PluginEvent[] = [] - - for (const [teamId, teamEvents] of eventsByTeam.entries()) { - const pluginsToRun = getPluginsForTeam(server, teamId) - - let returnedEvents: PluginEvent[] = teamEvents - const pluginsSucceeded = [] - const pluginsFailed = [] - for (const pluginConfig of pluginsToRun) { - const timer = new Date() - const processEventBatch = await pluginConfig.vm?.getProcessEventBatch() - if (processEventBatch && returnedEvents.length > 0) { - try { - returnedEvents = (await processEventBatch(returnedEvents)) || [] - let wasChangedTeamIdFound = false - for (const returnedEvent of returnedEvents) { - if (returnedEvent.team_id != teamId) { - returnedEvent.team_id = teamId - wasChangedTeamIdFound = true - } - } - if (wasChangedTeamIdFound) { - throw new IllegalOperationError('Plugin tried to change event.team_id') - } - pluginsSucceeded.push(`${pluginConfig.plugin?.name} (${pluginConfig.id})`) - } catch (error) { - await processError(server, pluginConfig, error, returnedEvents[0]) - server.statsd?.increment(`plugin.${pluginConfig.plugin?.name}.process_event_batch.ERROR`) - pluginsFailed.push(`${pluginConfig.plugin?.name} (${pluginConfig.id})`) - } - server.statsd?.timing(`plugin.${pluginConfig.plugin?.name}.process_event_batch`, timer) - server.statsd?.timing('plugin.process_event_batch', timer, { - plugin: pluginConfig.plugin?.name ?? '?', - teamId: teamId.toString(), - }) - } - } - - for (const event of returnedEvents) { - if (event && (pluginsSucceeded.length > 0 || pluginsFailed.length > 0)) { - event.properties = { - ...event.properties, - $plugins_succeeded: pluginsSucceeded, - $plugins_failed: pluginsFailed, - } - } - } - - allReturnedEvents = allReturnedEvents.concat(returnedEvents) - } - - return allReturnedEvents.filter(Boolean) -} - export async function runPluginTask( server: Hub, taskName: string, diff --git a/src/worker/tasks.ts b/src/worker/tasks.ts index 823bccfe..30072c75 100644 --- a/src/worker/tasks.ts +++ b/src/worker/tasks.ts @@ -2,7 +2,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { Action, EnqueuedJob, Hub, PluginTaskType, Team } from '../types' import { ingestEvent } from './ingestion/ingest-event' -import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent, runProcessEventBatch } from './plugins/run' +import { runOnEvent, runOnSnapshot, runPluginTask, runProcessEvent } from './plugins/run' import { loadSchedule, setupPlugins } from './plugins/setup' import { teardownPlugins } from './plugins/teardown' @@ -18,9 +18,6 @@ export const workerTasks: Record = { processEvent: (hub, args: { event: PluginEvent }) => { return runProcessEvent(hub, args.event) }, - processEventBatch: (hub, args: { batch: PluginEvent[] }) => { - return runProcessEventBatch(hub, args.batch) - }, runJob: (hub, { job }: { job: EnqueuedJob }) => { return runPluginTask(hub, job.type, PluginTaskType.Job, job.pluginConfigId, job.payload) }, diff --git a/src/worker/vm/lazy.ts b/src/worker/vm/lazy.ts index 115d5ec1..581a7f45 100644 --- a/src/worker/vm/lazy.ts +++ b/src/worker/vm/lazy.ts @@ -72,10 +72,6 @@ export class LazyPluginVM { return (await this.resolveInternalVm)?.methods.processEvent || null } - async getProcessEventBatch(): Promise { - return (await this.resolveInternalVm)?.methods.processEventBatch || null - } - async getTeardownPlugin(): Promise { return (await this.resolveInternalVm)?.methods.teardownPlugin || null } diff --git a/src/worker/vm/vm.ts b/src/worker/vm/vm.ts index 4916bcde..6a9c4cde 100644 --- a/src/worker/vm/vm.ts +++ b/src/worker/vm/vm.ts @@ -140,23 +140,8 @@ export async function createPluginConfigVM( if (func) return func(...args); } - // we have processEvent, but not processEventBatch - if (!__getExported('processEventBatch') && __getExported('processEvent')) { - exports.processEventBatch = async function __processEventBatch${pluginConfigIdentifier} (batch, meta) { - const processEvent = __getExported('processEvent'); - let waitFor = false - const processedEvents = batch.map(function __eventBatchToEvent${pluginConfigIdentifier} (event) { - const e = processEvent(event, meta) - if (e && typeof e.then !== 'undefined') { - waitFor = true - } - return e - }) - const response = waitFor ? (await Promise.all(processedEvents)) : processedEvents; - return response.filter(r => r) - } // we have processEventBatch, but not processEvent - } else if (!__getExported('processEvent') && __getExported('processEventBatch')) { + if (!__getExported('processEvent') && __getExported('processEventBatch')) { exports.processEvent = async function __processEvent${pluginConfigIdentifier} (event, meta) { return (await (__getExported('processEventBatch'))([event], meta))?.[0] } @@ -170,7 +155,6 @@ export async function createPluginConfigVM( onEvent: __asyncFunctionGuard(__bindMeta('onEvent')), onSnapshot: __asyncFunctionGuard(__bindMeta('onSnapshot')), processEvent: __asyncFunctionGuard(__bindMeta('processEvent')), - processEventBatch: __asyncFunctionGuard(__bindMeta('processEventBatch')), }; const __tasks = { diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 3cd8222a..3b0d255c 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -5,7 +5,7 @@ import { Hub, LogLevel, PluginTaskType } from '../src/types' import { clearError, processError } from '../src/utils/db/error' import { createHub } from '../src/utils/db/hub' import { loadPlugin } from '../src/worker/plugins/loadPlugin' -import { IllegalOperationError, runProcessEvent, runProcessEventBatch } from '../src/worker/plugins/run' +import { IllegalOperationError, runProcessEvent } from '../src/worker/plugins/run' import { loadSchedule, setupPlugins } from '../src/worker/plugins/setup' import { commonOrganizationId, @@ -78,7 +78,6 @@ test('setupPlugins and runProcessEvent', async () => { 'onEvent', 'onSnapshot', 'processEvent', - 'processEventBatch', 'setupPlugin', 'teardownPlugin', ]) @@ -90,7 +89,7 @@ test('setupPlugins and runProcessEvent', async () => { 60, { ...plugin60, - capabilities: { jobs: [], scheduled_tasks: [], methods: ['processEvent', 'processEventBatch'] }, + capabilities: { jobs: [], scheduled_tasks: [], methods: ['processEvent'] }, }, ], ]) @@ -251,11 +250,9 @@ test('plugin changing event.team_id throws error (single)', async () => { test('plugin changing event.team_id throws error (batch)', async () => { getPluginRows.mockReturnValueOnce([ mockPluginWithArchive(` - function processEventBatch (events, meta) { - for (const event of events) { - event.team_id = 400 - } - return events + function processEvent (events, meta) { + event.team_id = 400 + return event } `), ]) @@ -266,8 +263,8 @@ test('plugin changing event.team_id throws error (batch)', async () => { await setupPlugins(hub) const { pluginConfigs } = hub - const events = [{ event: '$test', properties: {}, team_id: 2 } as PluginEvent] - const returnedEvent = await runProcessEventBatch(hub, events) + const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent + const returnedEvent = await runProcessEvent(hub, event) const expectedReturnedEvent = { event: '$test', @@ -506,11 +503,7 @@ test('plugin with archive loads capabilities', async () => { await pluginConfig.vm?.resolveInternalVm // async loading of capabilities - expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual([ - 'processEvent', - 'processEventBatch', - 'setupPlugin', - ]) + expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['processEvent', 'setupPlugin']) expect(pluginConfig.plugin!.capabilities!.jobs).toHaveLength(0) expect(pluginConfig.plugin!.capabilities!.scheduled_tasks).toHaveLength(0) }) @@ -540,7 +533,7 @@ test('plugin with archive loads all capabilities, no random caps', async () => { await pluginConfig.vm?.resolveInternalVm // async loading of capabilities - expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['onEvent', 'processEvent', 'processEventBatch']) + expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['onEvent', 'processEvent']) expect(pluginConfig.plugin!.capabilities!.jobs).toEqual(['x']) expect(pluginConfig.plugin!.capabilities!.scheduled_tasks).toEqual(['runEveryHour']) }) @@ -564,7 +557,7 @@ test('plugin with source file loads capabilities', async () => { await pluginConfig.vm?.resolveInternalVm // async loading of capabilities - expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['onEvent', 'processEvent', 'processEventBatch']) + expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['onEvent', 'processEvent']) expect(pluginConfig.plugin!.capabilities!.jobs).toEqual([]) expect(pluginConfig.plugin!.capabilities!.scheduled_tasks).toEqual([]) @@ -590,11 +583,7 @@ test('plugin with source code loads capabilities', async () => { await pluginConfig.vm?.resolveInternalVm // async loading of capabilities - expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual([ - 'onSnapshot', - 'processEvent', - 'processEventBatch', - ]) + expect(pluginConfig.plugin!.capabilities!.methods!.sort()).toEqual(['onSnapshot', 'processEvent']) expect(pluginConfig.plugin!.capabilities!.jobs).toEqual([]) expect(pluginConfig.plugin!.capabilities!.scheduled_tasks).toEqual([]) }) @@ -677,7 +666,7 @@ test("capabilities don't reload without changes", async () => { getPluginRows.mockReturnValueOnce([{ ...plugin60 }]).mockReturnValueOnce([ { ...plugin60, - capabilities: { jobs: [], scheduled_tasks: [], methods: ['processEvent', 'processEventBatch'] }, + capabilities: { jobs: [], scheduled_tasks: [], methods: ['processEvent'] }, }, ]) // updated in DB via first `setPluginCapabilities` call. getPluginAttachmentRows.mockReturnValue([pluginAttachment1]) diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index bbe7f1e4..c34b0a1c 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -63,7 +63,6 @@ test('pause and resume queue', async () => { const piscina = setupPiscina(2, 2) const queue = await startQueue(hub, piscina, { processEvent: (event) => runProcessEvent(hub, event), - processEventBatch: (events) => Promise.all(events.map((event) => runProcessEvent(hub, event))), ingestEvent: () => Promise.resolve({ success: true }), }) await advanceOneTick() diff --git a/tests/postgres/vm.lazy.test.ts b/tests/postgres/vm.lazy.test.ts index ca418499..e99f5d14 100644 --- a/tests/postgres/vm.lazy.test.ts +++ b/tests/postgres/vm.lazy.test.ts @@ -51,7 +51,6 @@ describe('LazyPluginVM', () => { void initializeVm(vm) expect(await vm.getProcessEvent()).toEqual('processEvent') - expect(await vm.getProcessEventBatch()).toEqual(null) expect(await vm.getTask('someTask', PluginTaskType.Schedule)).toEqual(null) expect(await vm.getTask('runEveryMinute', PluginTaskType.Schedule)).toEqual('runEveryMinute') expect(await vm.getTasks(PluginTaskType.Schedule)).toEqual(mockVM.tasks.schedule) @@ -86,7 +85,6 @@ describe('LazyPluginVM', () => { void initializeVm(vm) expect(await vm.getProcessEvent()).toEqual(null) - expect(await vm.getProcessEventBatch()).toEqual(null) expect(await vm.getTask('runEveryMinute', PluginTaskType.Schedule)).toEqual(null) expect(await vm.getTasks(PluginTaskType.Schedule)).toEqual({}) }) diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index dd818c1b..d9dc48bf 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -154,7 +154,6 @@ test('async processEvent', async () => { await resetTestDatabase(indexJs) const vm = await createPluginConfigVM(hub, pluginConfig39, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) - expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, @@ -177,6 +176,7 @@ test('async processEvent', async () => { expect(newBatch[0]).toBe(batch[0]) }) +// this is deprecated, but still works test('processEventBatch', async () => { const indexJs = ` function processEventBatch (events, meta) { @@ -189,7 +189,6 @@ test('processEventBatch', async () => { await resetTestDatabase(indexJs) const vm = await createPluginConfigVM(hub, pluginConfig39, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) - expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, @@ -199,17 +198,6 @@ test('processEventBatch', async () => { expect(event.event).toEqual('changed event') expect(newEvent.event).toEqual('changed event') expect(newEvent).toBe(event) - - const batch: PluginEvent[] = [ - { - ...defaultEvent, - event: 'original event', - }, - ] - const newBatch = await vm.methods.processEventBatch!(batch) - expect(batch[0].event).toEqual('changed event') - expect(newBatch[0].event).toEqual('changed event') - expect(newBatch[0]).toBe(batch[0]) }) test('async processEventBatch', async () => { @@ -224,7 +212,6 @@ test('async processEventBatch', async () => { await resetTestDatabase(indexJs) const vm = await createPluginConfigVM(hub, pluginConfig39, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) - expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, @@ -234,17 +221,6 @@ test('async processEventBatch', async () => { expect(event.event).toEqual('changed event') expect(newEvent.event).toEqual('changed event') expect(newEvent).toBe(event) - - const batch: PluginEvent[] = [ - { - ...defaultEvent, - event: 'original event', - }, - ] - const newBatch = await vm.methods.processEventBatch!(batch) - expect(batch[0].event).toEqual('changed event') - expect(newBatch[0].event).toEqual('changed event') - expect(newBatch[0]).toBe(batch[0]) }) test('processEvent && processEventBatch', async () => { @@ -263,7 +239,6 @@ test('processEvent && processEventBatch', async () => { await resetTestDatabase(indexJs) const vm = await createPluginConfigVM(hub, pluginConfig39, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) - expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, @@ -273,17 +248,6 @@ test('processEvent && processEventBatch', async () => { expect(event.event).toEqual('changed event 1') expect(newEvent.event).toEqual('changed event 1') expect(newEvent).toBe(event) - - const batch: PluginEvent[] = [ - { - ...defaultEvent, - event: 'original event', - }, - ] - const newBatch = await vm.methods.processEventBatch!(batch) - expect(batch[0].event).toEqual('changed event 2') - expect(newBatch[0].event).toEqual('changed event 2') - expect(newBatch[0]).toBe(batch[0]) }) test('processEvent without returning', async () => { diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 51534e87..2694f567 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -12,7 +12,7 @@ import { delay, UUIDT } from '../../src/utils/utils' import { ActionManager } from '../../src/worker/ingestion/action-manager' import { ingestEvent } from '../../src/worker/ingestion/ingest-event' import { makePiscina } from '../../src/worker/piscina' -import { runPluginTask, runProcessEvent, runProcessEventBatch } from '../../src/worker/plugins/run' +import { runPluginTask, runProcessEvent } from '../../src/worker/plugins/run' import { loadSchedule, setupPlugins } from '../../src/worker/plugins/setup' import { teardownPlugins } from '../../src/worker/plugins/teardown' import { createTaskRunner } from '../../src/worker/worker' @@ -62,7 +62,6 @@ test('piscina worker test', async () => { const piscina = setupPiscina(workerThreads, 10) const processEvent = (event: PluginEvent) => piscina.runTask({ task: 'processEvent', args: { event } }) - const processEventBatch = (batch: PluginEvent[]) => piscina.runTask({ task: 'processEventBatch', args: { batch } }) const runEveryDay = (pluginConfigId: number) => piscina.runTask({ task: 'runEveryDay', args: { pluginConfigId } }) const ingestEvent = (event: PluginEvent) => piscina.runTask({ task: 'ingestEvent', args: { event } }) @@ -72,9 +71,6 @@ test('piscina worker test', async () => { const event = await processEvent(createEvent()) expect(event.properties['somewhere']).toBe('over the rainbow') - const eventBatch = await processEventBatch([createEvent()]) - expect(eventBatch[0]!.properties['somewhere']).toBe('over the rainbow') - const everyDayReturn = await runEveryDay(39) expect(everyDayReturn).toBe(4) @@ -257,16 +253,6 @@ describe('createTaskRunner()', () => { expect(runProcessEvent).toHaveBeenCalledWith(hub, 'someEvent') }) - it('handles `processEventBatch` task', async () => { - mocked(runProcessEventBatch).mockReturnValue(['runProcessEventBatch response'] as any) - - expect(await taskRunner({ task: 'processEventBatch', args: { batch: 'someBatch' } })).toEqual([ - 'runProcessEventBatch response', - ]) - - expect(runProcessEventBatch).toHaveBeenCalledWith(hub, 'someBatch') - }) - it('handles `getPluginSchedule` task', async () => { hub.pluginSchedule = { runEveryDay: [66] } From 3f79e39ddfa0fa441eb9a837d82807ff14baf22e Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 00:34:59 +0200 Subject: [PATCH 02/11] fix issues --- src/types.ts | 3 --- tests/plugins.test.ts | 2 +- 2 files changed, 1 insertion(+), 4 deletions(-) diff --git a/src/types.ts b/src/types.ts index 255407c0..6c1377fb 100644 --- a/src/types.ts +++ b/src/types.ts @@ -290,7 +290,6 @@ export type WorkerMethods = { onEvent: (event: PluginEvent) => Promise onSnapshot: (event: PluginEvent) => Promise processEvent: (event: PluginEvent) => Promise - processEventBatch: (batch: PluginEvent[]) => Promise<(PluginEvent | null)[]> ingestEvent: (event: PluginEvent) => Promise } @@ -301,8 +300,6 @@ export type VMMethods = { onSnapshot?: (event: PluginEvent) => Promise exportEvents?: (events: PluginEvent[]) => Promise processEvent?: (event: PluginEvent) => Promise - // DEPRECATED - processEventBatch?: (batch: PluginEvent[]) => Promise } export interface PluginConfigVMResponse { diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 3b0d255c..5239f7b0 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -274,7 +274,7 @@ test('plugin changing event.team_id throws error (batch)', async () => { }, team_id: 2, } - expect(returnedEvent).toEqual([expectedReturnedEvent]) + expect(returnedEvent).toEqual(expectedReturnedEvent) expect(processError).toHaveBeenCalledWith( hub, From eb64f133b03a360683a5f8818816783526b8fca7 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 08:53:18 +0200 Subject: [PATCH 03/11] fix bug, remove dead test code --- tests/plugins.test.ts | 2 +- tests/postgres/vm.test.ts | 24 ------------------------ 2 files changed, 1 insertion(+), 25 deletions(-) diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 5239f7b0..6d4d0e8d 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -250,7 +250,7 @@ test('plugin changing event.team_id throws error (single)', async () => { test('plugin changing event.team_id throws error (batch)', async () => { getPluginRows.mockReturnValueOnce([ mockPluginWithArchive(` - function processEvent (events, meta) { + function processEvent (event, meta) { event.team_id = 400 return event } diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index d9dc48bf..2d8b89ac 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -51,7 +51,6 @@ test('empty plugins', async () => { 'teardownPlugin', ]) expect(vm.methods.processEvent).toEqual(undefined) - expect(vm.methods.processEventBatch).toEqual(undefined) }) test('setupPlugin sync', async () => { @@ -121,7 +120,6 @@ test('processEvent', async () => { await resetTestDatabase(indexJs) const vm = await createPluginConfigVM(hub, pluginConfig39, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) - expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, @@ -131,17 +129,6 @@ test('processEvent', async () => { expect(event.event).toEqual('changed event') expect(newEvent.event).toEqual('changed event') expect(newEvent).toBe(event) - - const batch: PluginEvent[] = [ - { - ...defaultEvent, - event: 'original event', - }, - ] - const newBatch = await vm.methods.processEventBatch!(batch) - expect(batch[0].event).toEqual('changed event') - expect(newBatch[0].event).toEqual('changed event') - expect(newBatch[0]).toBe(batch[0]) }) test('async processEvent', async () => { @@ -163,17 +150,6 @@ test('async processEvent', async () => { expect(event.event).toEqual('changed event') expect(newEvent.event).toEqual('changed event') expect(newEvent).toBe(event) - - const batch: PluginEvent[] = [ - { - ...defaultEvent, - event: 'original event', - }, - ] - const newBatch = await vm.methods.processEventBatch!(batch) - expect(batch[0].event).toEqual('changed event') - expect(newBatch[0].event).toEqual('changed event') - expect(newBatch[0]).toBe(batch[0]) }) // this is deprecated, but still works From 12104f9ae3ebe75ffb03fe20bdb55789db2ee816 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:04:43 +0200 Subject: [PATCH 04/11] add verbosity to posthog/master dist github CI task --- .github/workflows/ci.yml | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e266a956..b1b165bd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -105,33 +105,39 @@ jobs: DATABASE_URL: 'postgres://postgres:postgres@localhost:5432/test_posthog' REDIS_URL: 'redis://localhost' run: | + echo "🔗 Linking this plugin server into posthog(master)/plugins/" cd posthog/plugins yarn link "@posthog/plugin-server" yarn - # start this in the background + echo "🏁 Starting the plugin server dist/ (compiled JS) via posthog(master)/plugins/" yarn start &> tmplog.txt & pid=$! + echo "⏳ Waiting 10 seconds to see if it runs" sleep 10 - # check if it's still running, as it should be + echo "🤔 Checking if it's still running, as it should be" if ! kill $pid > /dev/null 2>&1; then - echo "Could not send SIGTERM to process $pid" >&2 + echo "😵 It was not running!" + echo "🪵 Here's the log:" echo "" cat tmplog.txt exit 1 fi + echo '🤔 Checking if it logged "All systems go"' str=`cat tmplog.txt | grep "All systems go"` if [ ! "$str" ];then - echo 'Did not find "All systems go" in plugin server log output!' + echo "😵 Did not find "All systems go" in plugin server log output! " + echo '🪵 Here's the log:' echo "" cat tmplog.txt exit 1 fi - rm tmplog.txt + echo $str + rm -f tmplog.txt tests-postgres-1: name: Tests / Postgres + Redis (1) From 90d25bb412b282b92248160e3a2e62f7b953ffbd Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:08:25 +0200 Subject: [PATCH 05/11] fix quotes --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b1b165bd..404fb632 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -129,7 +129,7 @@ jobs: echo '🤔 Checking if it logged "All systems go"' str=`cat tmplog.txt | grep "All systems go"` if [ ! "$str" ];then - echo "😵 Did not find "All systems go" in plugin server log output! " + echo '😵 Did not find "All systems go" in plugin server log output!' echo '🪵 Here's the log:' echo "" cat tmplog.txt From c3941ca6c8461fcf3ff3a376c73083719f0e4e0c Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:15:07 +0200 Subject: [PATCH 06/11] fix test --- tests/postgres/vm.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index 2d8b89ac..3823c11e 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -46,7 +46,6 @@ test('empty plugins', async () => { 'onEvent', 'onSnapshot', 'processEvent', - 'processEventBatch', 'setupPlugin', 'teardownPlugin', ]) From e9cd33a39f5500151deb56a77dd5a8bd599b11db Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:18:34 +0200 Subject: [PATCH 07/11] more ci verbosity --- .github/workflows/ci.yml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 404fb632..ab9d152e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -129,6 +129,7 @@ jobs: echo '🤔 Checking if it logged "All systems go"' str=`cat tmplog.txt | grep "All systems go"` if [ ! "$str" ];then + sleep 5 echo '😵 Did not find "All systems go" in plugin server log output!' echo '🪵 Here's the log:' echo "" @@ -136,7 +137,10 @@ jobs: exit 1 fi - echo $str + echo '✅ All systems went!' + echo '🪵 Here's the complete log:' + echo "" + cat tmplog.txt rm -f tmplog.txt tests-postgres-1: From 380aef38be5dc601738f7c9f5e2197bb103d15ca Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:28:43 +0200 Subject: [PATCH 08/11] rename file --- .github/workflows/ci.yml | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ab9d152e..136604e7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -111,7 +111,8 @@ jobs: yarn echo "🏁 Starting the plugin server dist/ (compiled JS) via posthog(master)/plugins/" - yarn start &> tmplog.txt & + mkdir -p tmp + yarn start &> tmp/plugin-server.log & pid=$! echo "⏳ Waiting 10 seconds to see if it runs" @@ -122,26 +123,26 @@ jobs: echo "😵 It was not running!" echo "🪵 Here's the log:" echo "" - cat tmplog.txt + cat tmp/plugin-server.log exit 1 fi echo '🤔 Checking if it logged "All systems go"' - str=`cat tmplog.txt | grep "All systems go"` + str=`cat tmp/plugin-server.log | grep "All systems go"` if [ ! "$str" ];then sleep 5 echo '😵 Did not find "All systems go" in plugin server log output!' echo '🪵 Here's the log:' echo "" - cat tmplog.txt + cat tmp/plugin-server.log exit 1 fi echo '✅ All systems went!' echo '🪵 Here's the complete log:' echo "" - cat tmplog.txt - rm -f tmplog.txt + cat tmp/plugin-server.log + rm -f tmp/plugin-server.log tests-postgres-1: name: Tests / Postgres + Redis (1) From e5fea70e457024d1bfb7815523976729e4e5387d Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:34:51 +0200 Subject: [PATCH 09/11] quotes really are important :) --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 136604e7..758a5cec 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -132,14 +132,14 @@ jobs: if [ ! "$str" ];then sleep 5 echo '😵 Did not find "All systems go" in plugin server log output!' - echo '🪵 Here's the log:' + echo "🪵 Here's the log:" echo "" cat tmp/plugin-server.log exit 1 fi echo '✅ All systems went!' - echo '🪵 Here's the complete log:' + echo "🪵 Here's the complete log:" echo "" cat tmp/plugin-server.log rm -f tmp/plugin-server.log From c77b4f2faf275aacb684588f71e6e2083deef2bc Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Fri, 28 May 2021 09:50:28 +0200 Subject: [PATCH 10/11] give the plugin server time to shut down before displaying the full log --- .github/workflows/ci.yml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 758a5cec..ab3f3a05 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -138,6 +138,7 @@ jobs: exit 1 fi + sleep 5 echo '✅ All systems went!' echo "🪵 Here's the complete log:" echo "" From f062effa6942534562a618b4855cd74271e8d42e Mon Sep 17 00:00:00 2001 From: Michael Matloka Date: Fri, 28 May 2021 12:23:25 +0200 Subject: [PATCH 11/11] Remove now redundant test --- tests/plugins.test.ts | 39 +-------------------------------------- 1 file changed, 1 insertion(+), 38 deletions(-) diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 6d4d0e8d..4f101fd5 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -210,44 +210,7 @@ test('local plugin with broken index.js does not do much', async () => { unlink() }) -test('plugin changing event.team_id throws error (single)', async () => { - getPluginRows.mockReturnValueOnce([ - mockPluginWithArchive(` - function processEvent (event, meta) { - event.team_id = 400 - return event - } - `), - ]) - - getPluginConfigRows.mockReturnValueOnce([pluginConfig39]) - getPluginAttachmentRows.mockReturnValueOnce([]) - - await setupPlugins(hub) - const { pluginConfigs } = hub - - const event = { event: '$test', properties: {}, team_id: 2 } as PluginEvent - const returnedEvent = await runProcessEvent(hub, event) - - const expectedReturnedEvent = { - event: '$test', - properties: { - $plugins_failed: ['test-maxmind-plugin (39)'], - $plugins_succeeded: [], - }, - team_id: 2, - } - expect(returnedEvent).toEqual(expectedReturnedEvent) - - expect(processError).toHaveBeenCalledWith( - hub, - pluginConfigs.get(39)!, - new IllegalOperationError('Plugin tried to change event.team_id'), - expectedReturnedEvent - ) -}) - -test('plugin changing event.team_id throws error (batch)', async () => { +test('plugin changing event.team_id throws error', async () => { getPluginRows.mockReturnValueOnce([ mockPluginWithArchive(` function processEvent (event, meta) {