diff --git a/package.json b/package.json index 54c44996..a65927cf 100644 --- a/package.json +++ b/package.json @@ -86,6 +86,6 @@ } }, "lint-staged": { - "*.{js,css,scss}": "prettier --write" + "*.{js,ts,css,scss}": "prettier --write" } } diff --git a/src/plugins.ts b/src/plugins.ts index ca634167..7231fcc3 100644 --- a/src/plugins.ts +++ b/src/plugins.ts @@ -1,6 +1,6 @@ import * as path from 'path' import * as fs from 'fs' -import { createPluginConfigVM, prepareForRun } from './vm' +import { createPluginConfigVM } from './vm' import { PluginsServer, PluginConfig, PluginJsonConfig, TeamId } from './types' import { PluginEvent, PluginAttachment } from 'posthog-plugins' import { clearError, processError } from './error' @@ -180,26 +180,21 @@ async function loadPlugin(server: PluginsServer, pluginConfig: PluginConfig): Pr } export async function runPlugins(server: PluginsServer, event: PluginEvent): Promise { - const pluginsToRun = server.pluginConfigsPerTeam.get(event.team_id) || server.defaultConfigs - + const pluginsToRun = getPluginsForTeam(server, event.team_id) let returnedEvent: PluginEvent | null = event for (const pluginConfig of pluginsToRun.reverse()) { - if (pluginConfig.vm) { - const processEvent = prepareForRun(server, event.team_id, pluginConfig, 'processEvent', event) - - if (processEvent) { - const startTime = performance.now() - try { - returnedEvent = (await processEvent(returnedEvent)) || null - const ms = Math.round((performance.now() - startTime) * 1000) / 1000 - logTime(pluginConfig.plugin?.name || 'noname', ms) - } catch (error) { - await processError(server, pluginConfig, error, returnedEvent) - const ms = Math.round((performance.now() - startTime) * 1000) / 1000 - logTime(pluginConfig.plugin?.name || 'noname', ms, true) - } + if (pluginConfig.vm?.methods?.processEvent) { + let errored = false + const { processEvent } = pluginConfig.vm.methods + const startTime = performance.now() + try { + returnedEvent = (await processEvent(returnedEvent)) || null + } catch (error) { + errored = true + await processError(server, pluginConfig, error, returnedEvent) } + logTime(pluginConfig.plugin?.name || 'noname', performance.now() - startTime, errored) if (!returnedEvent) { return null @@ -209,3 +204,46 @@ export async function runPlugins(server: PluginsServer, event: PluginEvent): Pro return returnedEvent } + +export async function runPluginsOnBatch(server: PluginsServer, 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 + + for (const pluginConfig of pluginsToRun.reverse()) { + const { processEventBatch } = pluginConfig.vm?.methods || {} + if (processEventBatch && returnedEvents.length > 0) { + const startTime = performance.now() + let errored = false + try { + returnedEvents = (await processEventBatch(returnedEvents)) || [] + } catch (error) { + errored = true + await processError(server, pluginConfig, error, returnedEvents[0]) + } + logTime(pluginConfig.plugin?.name || 'noname', performance.now() - startTime, errored) + } + } + + allReturnedEvents = allReturnedEvents.concat(returnedEvents) + } + + return allReturnedEvents +} + +function getPluginsForTeam(server: PluginsServer, teamId: number): PluginConfig[] { + return server.pluginConfigsPerTeam.get(teamId) || server.defaultConfigs +} diff --git a/src/stats.ts b/src/stats.ts index 33e6d456..e955d981 100644 --- a/src/stats.ts +++ b/src/stats.ts @@ -3,7 +3,7 @@ const histories = new Map>() const historyIndex = new Map() export function logTime(name: string, time: number, error?: boolean): void { - const ms = Math.round(time * 1000) / 1000 + // const ms = Math.round(time * 1000) / 1000 // TODO: add this back with better dev logging. Disabling since this trashes performance tests. // console.log(`Running plugin ${name}: ${error ? 'ERROR IN ' : ''}${ms}ms`) if (!histories.has(name)) { diff --git a/src/types.ts b/src/types.ts index be331d86..d753cd41 100644 --- a/src/types.ts +++ b/src/types.ts @@ -113,5 +113,6 @@ export interface PluginConfigVMReponse { vm: VM methods: { processEvent: (event: PluginEvent) => Promise + processEventBatch: (batch: PluginEvent[]) => Promise } } diff --git a/src/vm.ts b/src/vm.ts index 5db05b2c..fba4f34c 100644 --- a/src/vm.ts +++ b/src/vm.ts @@ -65,9 +65,32 @@ export function createPluginConfigVM( // run the plugin setup script, if present __callWithMeta('setupPlugin'); + // we have processEvent, but not processEventBatch + if (!__getExported('processEventBatch') && __getExported('processEvent')) { + exports.processEventBatch = async function processEventBatch (batch, meta) { + const processEvent = __getExported('processEvent'); + let waitFor = false + const processedEvents = batch.map(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')) { + exports.processEvent = async function processEvent (event, meta) { + return (await (__getExported('processEventBatch'))([event], meta))?.[0] + } + } + // export various functions const __methods = { - processEvent: __bindMeta('processEvent') + processEvent: __bindMeta('processEvent'), + processEventBatch: __bindMeta('processEventBatch') }; ` ) @@ -77,33 +100,3 @@ export function createPluginConfigVM( methods: vm.run('__methods'), } } - -export function prepareForRun( - server: PluginsServer, - teamId: number, - pluginConfig: PluginConfig, // might have team_id=0 - method: 'processEvent', - event?: PluginEvent -): null | ((event: PluginEvent) => Promise) | (() => Promise) { - if (!pluginConfig.vm?.methods[method]) { - return null - } - - const { vm } = pluginConfig.vm - - if (event?.properties?.token) { - // TODO: this should be nicer... and it's not optimised for batch processing - const posthog = createInternalPostHogInstance( - event.properties.token, - { apiHost: event.site_url, fetch }, - { - performance: performance, - } - ) - vm.freeze(posthog, 'posthog') - } else { - vm.freeze(null, 'posthog') - } - - return pluginConfig.vm.methods[method] -} diff --git a/src/worker/worker.ts b/src/worker/worker.ts index 4e27a753..a80781a7 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,5 +1,5 @@ +import { runPlugins, runPluginsOnBatch, setupPlugins } from '../plugins' import { cloneObject, setLogLevel } from '../utils' -import { runPlugins, setupPlugins } from '../plugins' import { createServer } from '../server' import { PluginsServerConfig } from '../types' @@ -29,5 +29,10 @@ export async function createWorker(config: PluginsServerConfig): Promise) } + if (task === 'processEventBatch') { + const processedEvents = await runPluginsOnBatch(server, args.batch) + // must clone the object, as we may get from VM2 something like { ..., properties: Proxy {} } + return cloneObject(processedEvents as any[]) + } } } diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 0f58f3cc..88a55510 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -52,7 +52,7 @@ test('setupPlugins and runPlugins', async () => { }, }) expect(pluginConfig.vm).toBeDefined() - expect(Object.keys(pluginConfig.vm!.methods)).toEqual(['processEvent']) + expect(Object.keys(pluginConfig.vm!.methods)).toEqual(['processEvent', 'processEventBatch']) expect(setError).toHaveBeenCalled() expect(setError.mock.calls[0][0]).toEqual(mockServer) diff --git a/tests/vm.test.ts b/tests/vm.test.ts index ea009912..9b505520 100644 --- a/tests/vm.test.ts +++ b/tests/vm.test.ts @@ -1,4 +1,4 @@ -import { createPluginConfigVM, prepareForRun } from '../src/vm' +import { createPluginConfigVM } from '../src/vm' import { PluginConfig, PluginsServer, Plugin } from '../src/types' import { PluginEvent } from 'posthog-plugins' import { createServer } from '../src/server' @@ -55,8 +55,9 @@ test('empty plugins', async () => { const vm = createPluginConfigVM(mockServer, mockConfig, indexJs, libJs) expect(Object.keys(vm).sort()).toEqual(['methods', 'vm']) - expect(Object.keys(vm.methods).sort()).toEqual(['processEvent']) + expect(Object.keys(vm.methods).sort()).toEqual(['processEvent', 'processEventBatch']) expect(vm.methods.processEvent).toEqual(undefined) + expect(vm.methods.processEventBatch).toEqual(undefined) }) test('processEvent', async () => { @@ -68,17 +69,165 @@ test('processEvent', async () => { ` const vm = createPluginConfigVM(mockServer, mockConfig, indexJs) expect(vm.methods.processEvent).not.toEqual(undefined) + expect(vm.methods.processEventBatch).not.toEqual(undefined) const event: PluginEvent = { ...defaultEvent, event: 'original event', } + const newEvent = await vm.methods.processEvent(event) + 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 () => { + const indexJs = ` + async function processEvent (event, meta) { + event.event = 'changed event' + return event + } + ` + const vm = createPluginConfigVM(mockServer, mockConfig, indexJs) + expect(vm.methods.processEvent).not.toEqual(undefined) + expect(vm.methods.processEventBatch).not.toEqual(undefined) + + const event: PluginEvent = { + ...defaultEvent, + event: 'original event', + } + const newEvent = await vm.methods.processEvent(event) + 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('processEventBatch', async () => { + const indexJs = ` + function processEventBatch (events, meta) { + return events.map(event => { + event.event = 'changed event' + return event + }) + } + ` + const vm = createPluginConfigVM(mockServer, mockConfig, indexJs) + expect(vm.methods.processEvent).not.toEqual(undefined) + expect(vm.methods.processEventBatch).not.toEqual(undefined) + const event: PluginEvent = { + ...defaultEvent, + event: 'original event', + } const newEvent = await vm.methods.processEvent(event) + 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 () => { + const indexJs = ` + async function processEventBatch (events, meta) { + return events.map(event => { + event.event = 'changed event' + return event + }) + } + ` + const vm = createPluginConfigVM(mockServer, mockConfig, indexJs) + expect(vm.methods.processEvent).not.toEqual(undefined) + expect(vm.methods.processEventBatch).not.toEqual(undefined) + + const event: PluginEvent = { + ...defaultEvent, + event: 'original event', + } + const newEvent = await vm.methods.processEvent(event) 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 () => { + const indexJs = ` + function processEvent (event, meta) { + event.event = 'changed event 1' + return event + } + function processEventBatch (events, meta) { + return events.map(event => { + event.event = 'changed event 2' + return event + }) + } + ` + const vm = createPluginConfigVM(mockServer, mockConfig, indexJs) + expect(vm.methods.processEvent).not.toEqual(undefined) + expect(vm.methods.processEventBatch).not.toEqual(undefined) + + const event: PluginEvent = { + ...defaultEvent, + event: 'original event', + } + const newEvent = await vm.methods.processEvent(event) + 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 () => { @@ -343,90 +492,3 @@ test('attachments', async () => { expect(event.properties).toEqual(attachments) }) - -test('prepareForRun without token', async () => { - const indexJs = ` - async function processEvent (event, meta) { - event.properties = { - posthog: posthog - } - return event - } - ` - const pluginConfig = { ...mockConfig } - const vm = createPluginConfigVM(mockServer, pluginConfig, indexJs) - pluginConfig.vm = vm - const event: PluginEvent = { - ...defaultEvent, - event: 'prepareForRun event', - properties: {}, - } - const processEvent = prepareForRun(mockServer, pluginConfig.team_id, pluginConfig, 'processEvent', event) - - expect(processEvent).toBeDefined() - - await processEvent!(event) - - expect(event.properties!.posthog).toEqual(null) -}) - -test('prepareForRun with token gets posthog', async () => { - const indexJs = ` - async function processEvent (event, meta) { - event.properties = { - posthog: posthog - } - return event - } - ` - const pluginConfig = { ...mockConfig } - const vm = createPluginConfigVM(mockServer, pluginConfig, indexJs) - pluginConfig.vm = vm - const event: PluginEvent = { - ...defaultEvent, - event: 'prepareForRun event', - properties: { - token: 'posthog-token', - }, - } - const processEvent = prepareForRun(mockServer, pluginConfig.team_id, pluginConfig, 'processEvent', event) - expect(processEvent).toBeDefined() - - await processEvent!(event) - - expect(event.properties!.posthog.capture).toBeDefined() - expect(event.properties!.posthog.identify).toBeDefined() -}) - -test('posthog.capture', async () => { - const indexJs = ` - async function processEvent (event, meta) { - posthog.capture('random-event', { prop: 'value' }) - return event - } - ` - const pluginConfig = { ...mockConfig } - const vm = createPluginConfigVM(mockServer, pluginConfig, indexJs) - pluginConfig.vm = vm - const event: PluginEvent = { - ...defaultEvent, - event: 'prepareForRun event', - properties: { - // needs a token in the event - token: 'posthog-token', - }, - } - const processEvent = prepareForRun(mockServer, pluginConfig.team_id, pluginConfig, 'processEvent', event) - expect(processEvent).toBeDefined() - await processEvent!(event) - - expect((fetch as any).mock.calls[0][0]).toContain('http://localhost/e/?ip=1&_=') - expect((fetch as any).mock.calls[0][1].body).toContain('data=') - expect((fetch as any).mock.calls[0][1].body).toContain('&compression=lz64') - expect((fetch as any).mock.calls[0][1]).toMatchObject({ - credentials: 'omit', - headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, - method: 'POST', - mode: 'no-cors', - }) -}) diff --git a/tests/worker.test.ts b/tests/worker.test.ts index ee9c0612..7d0adcc0 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -7,9 +7,12 @@ import * as os from 'os' import { LogLevel } from '../src/types' jest.mock('../src/sql') -jest.setTimeout(300000) // 300 sec timeout +jest.setTimeout(600000) // 600 sec timeout -function processOneEvent(processEvent: (event: PluginEvent) => Promise): Promise { +function processOneEvent( + processEvent: (event: PluginEvent) => Promise, + index: number +): Promise { const defaultEvent = { distinct_id: 'my_id', ip: '127.0.0.1', @@ -17,38 +20,45 @@ function processOneEvent(processEvent: (event: PluginEvent) => Promise) { +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 startTime = performance.now() 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) for (let j = 0; j < groups; j++) { - const groupCount = j === groups - 1 ? count % maxPromises : maxPromises + const groupCount = groups === 1 ? count : j === groups - 1 ? count % maxPromises : maxPromises for (let i = 0; i < groupCount; i++) { - promises[i] = processOneEvent(processEvent) + promises[i] = + batchSize === 1 ? processOneEvent(processEvent, i) : processOneBatch(processEventBatch, batchSize, i) } await Promise.all(promises) } - - const ms = Math.round((performance.now() - startTime) * 1000) / 1000 - - const log = { - eventsPerSecond: 1000 / (ms / count), - events: count, - concurrency: piscina.threads.length, - totalMs: ms, - averageEventMs: ms / count, - } - - return log } function setupPiscina(workers: number, code: string, tasksPerWorker: number) { @@ -62,10 +72,15 @@ function setupPiscina(workers: number, code: string, tasksPerWorker: number) { } test('piscina worker test', async () => { - const coreCount = os.cpus().length + // Uncomment this to become a 10x developer and make the test run just as fast! + // Reduces events by 10x and limits threads to max 8 for quicker development + const isLightDevRun = false - const workers = [1, 2, 4, 8, 12, 16].filter((cores) => cores <= coreCount) - const rounds = 5 + const coreCount = os.cpus().length + const workerThreads = [1, 2, 4, 8, 12, 16].filter((threads) => + isLightDevRun ? threads <= 8 : threads <= coreCount + ) + const rounds = 1 const tests: { testName: string; events: number; testCode: string }[] = [ { @@ -91,7 +106,7 @@ test('piscina worker test', async () => { }, { testName: 'timeout100ms', - events: 2000, + events: 10000, testCode: ` async function processEvent (event, meta) { await new Promise(resolve => __jestSetTimeout(() => resolve(), 100)) @@ -103,28 +118,35 @@ test('piscina worker test', async () => { ] const results: Array> = [] - for (const { testName, events, testCode } of tests) { - const result: Record = { - testName, - coreCount, - } - for (const cores of workers) { - const piscina = setupPiscina(cores, testCode, 100) + for (const { testName, events: _events, testCode } of tests) { + const events = isLightDevRun ? _events / 10 : _events + for (const batchSize of [1, 10, 100].filter((size) => size <= events)) { + const result: Record = { + testName, + coreCount, + events, + batchSize, + } + for (const threads of workerThreads) { + const piscina = setupPiscina(threads, testCode, 100) - // warmup - await processCountEvents(cores * 4, piscina) + // warmup + await processCountEvents(piscina, threads * 4) + + // start + const startTime = performance.now() + for (let i = 0; i < rounds; i++) { + await processCountEvents(piscina, events / batchSize, batchSize) + } + result[`${threads} thread${threads === 1 ? '' : 's'}`] = Math.round( + 1000 / ((startTime - performance.now()) / events / rounds) + ) - // start - let throughput = 0 - for (let i = 0; i < rounds; i++) { - const { eventsPerSecond } = await processCountEvents(events, piscina) - throughput += eventsPerSecond + await piscina.destroy() } - result[`${cores} cores`] = Math.round(throughput / rounds) - await piscina.destroy() + results.push(result) + console.log(JSON.stringify({ result }, null, 2)) } - results.push(result) - console.log(JSON.stringify({ result }, null, 2)) } console.table(results) })