From 8c7141b3b8ae6cdd0486ac29e46a5ab438b7fcae Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 16:07:44 +0100 Subject: [PATCH 01/21] fix deprecations in tests --- jest.setup.fetch-mock.js | 3 ++- tests/queue.test.ts | 4 ++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/jest.setup.fetch-mock.js b/jest.setup.fetch-mock.js index 9f684662..c9b2f01b 100644 --- a/jest.setup.fetch-mock.js +++ b/jest.setup.fetch-mock.js @@ -6,7 +6,8 @@ jest.mock('node-fetch', () => { (url) => new Promise((resolve) => resolve({ - json: () => new Promise((resolve) => resolve(responsesToUrls[url])), + json: () => new Promise((resolve) => resolve(responsesToUrls[url]) || { fetch: 'mock' }), + text: () => new Promise((resolve) => resolve(JSON.stringify(responsesToUrls[url])) || 'fetchmock'), }) ) ) diff --git a/tests/queue.test.ts b/tests/queue.test.ts index a91c49fe..ee2ff387 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -53,7 +53,7 @@ test('worker and task passing via redis', async () => { expect(item['headers']['task']).toBe('posthog.tasks.process_event.process_event_with_plugins') expect(item['properties']['body_encoding']).toBe('base64') - const body = new Buffer(item['body'], 'base64').toString() + const body = Buffer.from(item['body'], 'base64').toString() const [args2, kwargs2] = JSON.parse(body) expect(args2).toEqual(args) @@ -74,7 +74,7 @@ test('worker and task passing via redis', async () => { expect(processedItem['headers']['task']).toBe('posthog.tasks.process_event.process_event') expect(processedItem['properties']['body_encoding']).toBe('base64') - const processedBody = new Buffer(processedItem['body'], 'base64').toString() + const processedBody = Buffer.from(processedItem['body'], 'base64').toString() const [args3, kwargs3] = JSON.parse(processedBody) expect(args3).toEqual([]) From a1cfdbbd31b3661c5e51e3d61ff141ce895452aa Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 16:07:57 +0100 Subject: [PATCH 02/21] console.table test results --- tests/piscina.test.ts | 79 +++++++++++++++++++++++++------------------ 1 file changed, 47 insertions(+), 32 deletions(-) diff --git a/tests/piscina.test.ts b/tests/piscina.test.ts index 074d10c5..2f9b17b2 100644 --- a/tests/piscina.test.ts +++ b/tests/piscina.test.ts @@ -45,51 +45,66 @@ async function processCountEvents(count: number, piscina: ReturnType { - const cpuCount = os.cpus().length + const testName = '' + const coreCount = os.cpus().length - const workers = [1, 2, 4, 8, 12, 16, 24, 32, 48, 64].filter((cores) => - cpuCount === 2 ? cores <= cpuCount : cores < cpuCount - ) - const events = 10000 + const workers = [1, 2, 4, 8].filter((cores) => cores <= coreCount) + const events = 1000 const rounds = 5 - const results: Record = {} - for (const cores of workers) { - const piscina = setupPiscina(cores) - - // warmup - await processCountEvents(cpuCount * 4, piscina) + const tests: Record = { + simple: ` + function processEvent (event, meta) { + event.properties = { "somewhere": "over the rainbow" }; + return event + } + `, + for200k: ` + function processEvent (event, meta) { + let j = 0; for(let i = 0; i < 200000; i++) { j = i }; + event.properties = { "somewhere": "over the rainbow" }; + return event + } + `, + } - // start - let throughput = 0 - for (let i = 0; i < rounds; i++) { - const { eventsPerSecond } = await processCountEvents(events, piscina) - throughput += eventsPerSecond + const results: Array> = [] + for (const [testName, testCode] of Object.entries(tests)) { + const result: Record = { + testName, + coreCount, } - results[cores] = Math.round(throughput / rounds) - await piscina.destroy() - } + for (const cores of workers) { + const piscina = setupPiscina(cores, testCode) - console.log({ cpuCount }) - console.log(JSON.stringify(results, null, 2)) + // warmup + await processCountEvents(coreCount * 4, piscina) - // expect that adding more cores (up to cpuCount) increases throughput - for (let i = 1; i < workers.length; i++) { - expect(results[workers[i - 1]]).toBeLessThan(results[workers[i]]) + // start + let throughput = 0 + for (let i = 0; i < rounds; i++) { + const { eventsPerSecond } = await processCountEvents(events * cores, piscina) + throughput += eventsPerSecond + } + result[`${cores} cores`] = Math.round(throughput / rounds) + await piscina.destroy() + } + console.log(JSON.stringify({ result }, null, 2)) + // for (let i = 1; i < workers.length; i++) { + // expect(result[`${workers[i - 1]} cores`]).toBeLessThan(result[`${workers[i]} cores`]) + // } + results.push(result) } + console.table(results) + + // expect that adding more cores (up to coreCount) increases throughput }) From 65368dc46a28cc73ba76c80c160769e29f4c9ada Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 16:23:48 +0100 Subject: [PATCH 03/21] add test for async/setTimeout/fetch --- src/vm.ts | 7 +++ tests/{piscina.test.ts => worker.test.ts} | 65 ++++++++++++++--------- 2 files changed, 48 insertions(+), 24 deletions(-) rename tests/{piscina.test.ts => worker.test.ts} (64%) diff --git a/src/vm.ts b/src/vm.ts index bbc0951c..5db05b2c 100644 --- a/src/vm.ts +++ b/src/vm.ts @@ -7,6 +7,10 @@ import { createCache } from './extensions/cache' import { createInternalPostHogInstance } from 'posthog-js-lite' import { performance } from 'perf_hooks' +function areWeTestingWithJest() { + return process.env.JEST_WORKER_ID !== undefined +} + export function createPluginConfigVM( server: PluginsServer, pluginConfig: PluginConfig, // NB! might have team_id = 0 @@ -18,6 +22,9 @@ export function createPluginConfigVM( }) vm.freeze(createConsole(), 'console') vm.freeze(fetch, 'fetch') + if (areWeTestingWithJest()) { + vm.freeze(setTimeout, '__jestSetTimeout') + } vm.freeze( { cache: createCache( diff --git a/tests/piscina.test.ts b/tests/worker.test.ts similarity index 64% rename from tests/piscina.test.ts rename to tests/worker.test.ts index 2f9b17b2..9bc95cc6 100644 --- a/tests/piscina.test.ts +++ b/tests/worker.test.ts @@ -53,32 +53,50 @@ function setupPiscina(workers: number, code: string) { }) } -test('piscina 2-24 workers', async () => { +test('piscina worker test', async () => { const testName = '' const coreCount = os.cpus().length const workers = [1, 2, 4, 8].filter((cores) => cores <= coreCount) - const events = 1000 const rounds = 5 - const tests: Record = { - simple: ` - function processEvent (event, meta) { - event.properties = { "somewhere": "over the rainbow" }; - return event - } - `, - for200k: ` - function processEvent (event, meta) { - let j = 0; for(let i = 0; i < 200000; i++) { j = i }; - event.properties = { "somewhere": "over the rainbow" }; - return event - } - `, - } + const tests: { testName: string; events: number; testCode: string }[] = [ + { + testName: 'simple', + events: 10000, + testCode: ` + function processEvent (event, meta) { + event.properties = { "somewhere": "over the rainbow" }; + return event + } + `, + }, + { + testName: 'for200k', + events: 10000, + testCode: ` + function processEvent (event, meta) { + let j = 0; for(let i = 0; i < 200000; i++) { j = i }; + event.properties = { "somewhere": "over the rainbow" }; + return event + } + `, + }, + { + testName: 'timeout100ms', + events: 5, + testCode: ` + async function processEvent (event, meta) { + await new Promise(resolve => __jestSetTimeout(() => resolve(), 100)) + event.properties = { "somewhere": "over the rainbow" }; + return event + } + `, + }, + ] const results: Array> = [] - for (const [testName, testCode] of Object.entries(tests)) { + for (const { testName, events, testCode } of tests) { const result: Record = { testName, coreCount, @@ -87,7 +105,7 @@ test('piscina 2-24 workers', async () => { const piscina = setupPiscina(cores, testCode) // warmup - await processCountEvents(coreCount * 4, piscina) + await processCountEvents(cores * 3, piscina) // start let throughput = 0 @@ -98,13 +116,12 @@ test('piscina 2-24 workers', async () => { result[`${cores} cores`] = Math.round(throughput / rounds) await piscina.destroy() } - console.log(JSON.stringify({ result }, null, 2)) - // for (let i = 1; i < workers.length; i++) { - // expect(result[`${workers[i - 1]} cores`]).toBeLessThan(result[`${workers[i]} cores`]) - // } results.push(result) + console.log(JSON.stringify({ result }, null, 2)) } console.table(results) - // expect that adding more cores (up to coreCount) increases throughput + // for (let i = 1; i < workers.length; i++) { + // expect(result[`${workers[i - 1]} cores`]).toBeLessThan(result[`${workers[i]} cores`]) + // } }) From 941606574293b1bd940af6352058858317164d95 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 16:56:34 +0100 Subject: [PATCH 04/21] tasks per worker --- src/server.ts | 1 + src/types.ts | 1 + src/worker/config.ts | 4 ++++ tests/worker.test.ts | 5 +++-- 4 files changed, 9 insertions(+), 2 deletions(-) diff --git a/src/server.ts b/src/server.ts index 7faecd5a..75b119f1 100644 --- a/src/server.ts +++ b/src/server.ts @@ -31,6 +31,7 @@ export const defaultConfig: PluginsServerConfig = overrideWithEnv({ WEB_PORT: 3008, WEB_HOSTNAME: '0.0.0.0', WORKER_CONCURRENCY: 0, // use all cores + TASKS_PER_WORKER: 1, }) export async function createServer( diff --git a/src/types.ts b/src/types.ts index 227b6dcc..2619c89d 100644 --- a/src/types.ts +++ b/src/types.ts @@ -5,6 +5,7 @@ import { VM, VMScript } from 'vm2' export interface PluginsServerConfig { WORKER_CONCURRENCY: number + TASKS_PER_WORKER: number CELERY_DEFAULT_QUEUE: string DATABASE_URL: string PLUGINS_CELERY_QUEUE: string diff --git a/src/worker/config.ts b/src/worker/config.ts index 35d82d37..aa1a9e52 100644 --- a/src/worker/config.ts +++ b/src/worker/config.ts @@ -31,5 +31,9 @@ export function createConfig(serverConfig: PluginsServerConfig, filename: string config.maxThreads = serverConfig.WORKER_CONCURRENCY } + if (serverConfig.TASKS_PER_WORKER > 1) { + config.concurrentTasksPerWorker = serverConfig.TASKS_PER_WORKER + } + return config } diff --git a/tests/worker.test.ts b/tests/worker.test.ts index 9bc95cc6..d86b2126 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -45,10 +45,11 @@ async function processCountEvents(count: number, piscina: ReturnType { coreCount, } for (const cores of workers) { - const piscina = setupPiscina(cores, testCode) + const piscina = setupPiscina(cores, testCode, 1) // warmup await processCountEvents(cores * 3, piscina) From 4d471d698ff660963d478d24cdfcd40f79fb2562 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 18:03:35 +0100 Subject: [PATCH 05/21] reset numbers --- tests/worker.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/worker.test.ts b/tests/worker.test.ts index d86b2126..a907cb77 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -106,12 +106,12 @@ test('piscina worker test', async () => { const piscina = setupPiscina(cores, testCode, 1) // warmup - await processCountEvents(cores * 3, piscina) + await processCountEvents(cores * 4, piscina) // start let throughput = 0 for (let i = 0; i < rounds; i++) { - const { eventsPerSecond } = await processCountEvents(events * cores, piscina) + const { eventsPerSecond } = await processCountEvents(events, piscina) throughput += eventsPerSecond } result[`${cores} cores`] = Math.round(throughput / rounds) From d930d756454c8db2eed3bd0d3b1e9cf42f88ea0d Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 22:08:52 +0100 Subject: [PATCH 06/21] default config via cli and env --- src/config.ts | 48 +++++++++++++++++++++++++++++++++++++++++++++ src/index.ts | 54 +++++++++++++++++++++++++++------------------------ src/server.ts | 25 +----------------------- src/types.ts | 3 +++ 4 files changed, 81 insertions(+), 49 deletions(-) create mode 100644 src/config.ts diff --git a/src/config.ts b/src/config.ts new file mode 100644 index 00000000..378336f8 --- /dev/null +++ b/src/config.ts @@ -0,0 +1,48 @@ +import { PluginsServerConfig } from './types' + +export const defaultConfig = overrideWithEnv(getDefaultConfig()) +export const configHelp = getConfigHelp() + +function getDefaultConfig(): PluginsServerConfig { + return { + CELERY_DEFAULT_QUEUE: 'celery', + DATABASE_URL: 'postgres://localhost:5432/posthog', + PLUGINS_CELERY_QUEUE: 'posthog-plugins', + REDIS_URL: 'redis://localhost/', + BASE_DIR: '.', + PLUGINS_RELOAD_PUBSUB_CHANNEL: 'reload-plugins', + DISABLE_WEB: false, + WEB_PORT: 3008, + WEB_HOSTNAME: '0.0.0.0', + WORKER_CONCURRENCY: 0, // use all cores + TASKS_PER_WORKER: 1, + LOG_LEVEL: 'log', + } +} + +function getConfigHelp(): Record { + return { + CELERY_DEFAULT_QUEUE: 'celery outgoing queue', + DATABASE_URL: 'url for postgres', + PLUGINS_CELERY_QUEUE: 'celery incoming queue', + REDIS_URL: 'url for redis', + BASE_DIR: 'base path for resolving local plugins', + PLUGINS_RELOAD_PUBSUB_CHANNEL: 'redis channel for reload events', + DISABLE_WEB: 'do not start the web service', + WEB_PORT: 'port for web server', + WEB_HOSTNAME: 'hostname for web server', + WORKER_CONCURRENCY: 'number of concurrent worker threads', + TASKS_PER_WORKER: 'number of parallel tasks per worker thread', + LOG_LEVEL: 'minimum log level', + } +} + +function overrideWithEnv(config: PluginsServerConfig): PluginsServerConfig { + const newConfig: Record = { ...config } + for (const [key, value] of Object.entries(config)) { + if (process.env[key]) { + newConfig[key] = process.env[key] + } + } + return newConfig as PluginsServerConfig +} diff --git a/src/index.ts b/src/index.ts index dda7ee66..3daec4c8 100644 --- a/src/index.ts +++ b/src/index.ts @@ -2,6 +2,7 @@ import * as yargs from 'yargs' import { PluginsServerConfig } from './types' import { startPluginsServer } from './server' import { makePiscina } from './worker/piscina' +import { defaultConfig, configHelp } from './config' type Argv = { config: string @@ -11,31 +12,34 @@ type Argv = { concurrency: number } -yargs +let app: any = yargs + .wrap(yargs.terminalWidth()) .scriptName('posthog-plugins') .option('config', { alias: 'c', describe: 'Config options JSON.', type: 'string' }) - .option('disable-web', { describe: 'Whether web server should be disabled.', type: 'boolean' }) - .option('web-port', { alias: 'p', describe: 'Web server port.', type: 'number' }) - .option('web-hostname', { alias: 'h', describe: 'Web server hostname.', type: 'string' }) - .option('concurrency', { describe: 'Concurrenct Worker Threads', type: 'number' }) - .help() - .command({ - command: ['start', '$0'], - describe: 'start the server', - handler: ({ config, disableWeb, webPort, webHostname, concurrency }: Argv) => { - const parsedConfig: PluginsServerConfig = config ? JSON.parse(config) : {} - if (typeof webHostname !== 'undefined') { - parsedConfig['WEB_HOSTNAME'] = webHostname - } - if (typeof webPort !== 'undefined') { - parsedConfig['WEB_PORT'] = webPort - } - if (typeof disableWeb !== 'undefined') { - parsedConfig['DISABLE_WEB'] = disableWeb - } - if (typeof concurrency !== 'undefined') { - parsedConfig['WORKER_CONCURRENCY'] = concurrency + +for (const [key, value] of Object.entries(defaultConfig)) { + app = app.option(key.toLowerCase().replaceAll('_', '-'), { + describe: `${configHelp[key] || key} [${value}]`, + type: typeof value, + }) +} + +app = app.help().command({ + command: ['start', '$0'], + describe: 'start the server', + handler: ({ config, ...otherArgs }: Argv) => { + const parsedConfig: Record = config ? JSON.parse(config) : {} + for (const [key, value] of Object.entries(otherArgs)) { + if (typeof value !== 'undefined') { + const newKey = key + .replace(/(?:^|\.?)([A-Z])/g, (x, y) => '_' + y.toUpperCase()) + .replace(/^_/, '') + .toUpperCase() + if (newKey in defaultConfig) { + parsedConfig[newKey] = value + } } - startPluginsServer(parsedConfig, makePiscina) - }, - }).argv + } + startPluginsServer(parsedConfig as PluginsServerConfig, makePiscina) + }, +}).argv diff --git a/src/server.ts b/src/server.ts index 75b119f1..e6704270 100644 --- a/src/server.ts +++ b/src/server.ts @@ -8,32 +8,9 @@ import { startFastifyInstance, stopFastifyInstance } from './web/server' import { Worker } from 'celery/worker' import { version } from '../package.json' import { PluginEvent } from 'posthog-plugins' +import { defaultConfig } from './config' import Piscina from 'piscina' -function overrideWithEnv(config: PluginsServerConfig): PluginsServerConfig { - const newConfig: Record = { ...config } - for (const [key, value] of Object.entries(config)) { - if (process.env[key]) { - newConfig[key] = process.env[key] - } - } - return newConfig as PluginsServerConfig -} - -export const defaultConfig: PluginsServerConfig = overrideWithEnv({ - CELERY_DEFAULT_QUEUE: 'celery', - DATABASE_URL: 'postgres://localhost:5432/posthog', - PLUGINS_CELERY_QUEUE: 'posthog-plugins', - REDIS_URL: 'redis://localhost/', - BASE_DIR: '.', - PLUGINS_RELOAD_PUBSUB_CHANNEL: 'reload-plugins', - DISABLE_WEB: false, - WEB_PORT: 3008, - WEB_HOSTNAME: '0.0.0.0', - WORKER_CONCURRENCY: 0, // use all cores - TASKS_PER_WORKER: 1, -}) - export async function createServer( config: Partial = {} ): Promise<[PluginsServer, () => Promise]> { diff --git a/src/types.ts b/src/types.ts index 2619c89d..a3fa1277 100644 --- a/src/types.ts +++ b/src/types.ts @@ -3,6 +3,8 @@ import { Redis } from 'ioredis' import { PluginEvent, PluginAttachment, PluginConfigSchema } from 'posthog-plugins' import { VM, VMScript } from 'vm2' +export type LogLevel = 'debug' | 'info' | 'log' | 'warn' | 'error' | 'none' + export interface PluginsServerConfig { WORKER_CONCURRENCY: number TASKS_PER_WORKER: number @@ -15,6 +17,7 @@ export interface PluginsServerConfig { DISABLE_WEB: boolean WEB_PORT: number WEB_HOSTNAME: string + LOG_LEVEL: LogLevel __jestMock?: { getPluginRows: Plugin[] From 4037c91f8b64083ef6a99f7ff34bab8cfdb332ab Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 22:21:50 +0100 Subject: [PATCH 07/21] set log level --- src/celery/broker.ts | 2 +- src/config.ts | 2 +- src/index.ts | 2 ++ src/plugins.ts | 2 +- src/server.ts | 5 +++++ src/utils.ts | 10 ++++++++++ src/worker/piscina.js | 2 +- src/worker/worker.ts | 3 +++ 8 files changed, 24 insertions(+), 4 deletions(-) diff --git a/src/celery/broker.ts b/src/celery/broker.ts index da3559de..cf391376 100644 --- a/src/celery/broker.ts +++ b/src/celery/broker.ts @@ -131,7 +131,7 @@ export default class RedisBroker { Promise.resolve() }) .then(() => this.receive(index, resolve, queue, callback)) - .catch((err) => console.log(err)) + .catch((err) => console.error(err)) } /** diff --git a/src/config.ts b/src/config.ts index 378336f8..c4a861fc 100644 --- a/src/config.ts +++ b/src/config.ts @@ -16,7 +16,7 @@ function getDefaultConfig(): PluginsServerConfig { WEB_HOSTNAME: '0.0.0.0', WORKER_CONCURRENCY: 0, // use all cores TASKS_PER_WORKER: 1, - LOG_LEVEL: 'log', + LOG_LEVEL: 'info', } } diff --git a/src/index.ts b/src/index.ts index 3daec4c8..19a9c847 100644 --- a/src/index.ts +++ b/src/index.ts @@ -3,6 +3,7 @@ import { PluginsServerConfig } from './types' import { startPluginsServer } from './server' import { makePiscina } from './worker/piscina' import { defaultConfig, configHelp } from './config' +import { setLogLevel } from './utils' type Argv = { config: string @@ -40,6 +41,7 @@ app = app.help().command({ } } } + setLogLevel(parsedConfig.LOG_LEVEL || defaultConfig.LOG_LEVEL) startPluginsServer(parsedConfig as PluginsServerConfig, makePiscina) }, }).argv diff --git a/src/plugins.ts b/src/plugins.ts index 94585670..ca634167 100644 --- a/src/plugins.ts +++ b/src/plugins.ts @@ -161,7 +161,7 @@ async function loadPlugin(server: PluginsServer, pluginConfig: PluginConfig): Pr if (indexJs) { try { pluginConfig.vm = createPluginConfigVM(server, pluginConfig, indexJs, libJs || '') - console.log(`Loaded plugin "${plugin.name}"!`) + console.info(`Loaded plugin "${plugin.name}"!`) await clearError(server, pluginConfig) return true } catch (error) { diff --git a/src/server.ts b/src/server.ts index e6704270..c6adebbf 100644 --- a/src/server.ts +++ b/src/server.ts @@ -25,6 +25,11 @@ export async function createServer( const redis = new Redis(serverConfig.REDIS_URL) + redis.on('error', (error) => { + console.info('🔴 Redis error!', error) + process.kill(process.pid, 'SIGTERM') + }) + const server: PluginsServer = { ...serverConfig, db, diff --git a/src/utils.ts b/src/utils.ts index e8825644..c27bfffe 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -2,6 +2,7 @@ import { Readable } from 'stream' import * as tar from 'tar-stream' import AdmZip from 'adm-zip' import * as zlib from 'zlib' +import { LogLevel } from 'types' /** * @param binary Buffer @@ -94,3 +95,12 @@ export function getFileFromZip(archive: Buffer, file: string): string | null { return null } + +export function setLogLevel(logLevel: LogLevel) { + for (const loopLevel of ['debug', 'info', 'log', 'warn', 'error', 'none']) { + if (loopLevel === logLevel) { + break + } + ;(console as any)[loopLevel] = () => {} + } +} diff --git a/src/worker/piscina.js b/src/worker/piscina.js index 2d19e3db..1e0dcdde 100644 --- a/src/worker/piscina.js +++ b/src/worker/piscina.js @@ -15,7 +15,7 @@ if (isMainThread) { }, } } else { - console.log('🧵 Starting Piscina Worker Thread') + console.info('🧵 Starting Piscina Worker Thread') if (areWeTestingWithJest()) { require('ts-node').register() diff --git a/src/worker/worker.ts b/src/worker/worker.ts index 155bc222..acb0f1ac 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,3 +1,4 @@ +import { setLogLevel } from '../utils' import { runPlugins, setupPlugins } from '../plugins' import { createServer } from '../server' import { PluginsServerConfig } from '../types' @@ -5,6 +6,8 @@ import { PluginsServerConfig } from '../types' type TaskWorker = ({ task, args }: { task: string; args: any }) => Promise export async function createWorker(config: PluginsServerConfig): Promise { + setLogLevel(config.LOG_LEVEL) + const [server, closeServer] = await createServer(config) await setupPlugins(server) From 28d5f7d603effba9de72d42fb848ed114d5c5484 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 22:32:38 +0100 Subject: [PATCH 08/21] log level test --- src/utils.ts | 2 + tests/utils.test.ts | 90 ++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/src/utils.ts b/src/utils.ts index c27bfffe..984c6baf 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -101,6 +101,8 @@ export function setLogLevel(logLevel: LogLevel) { if (loopLevel === logLevel) { break } + const originalFunction = (console as any)[loopLevel]._original || (console as any)[loopLevel] ;(console as any)[loopLevel] = () => {} + ;(console as any)[loopLevel]._original = originalFunction } } diff --git a/tests/utils.test.ts b/tests/utils.test.ts index 871c8098..b1cf411e 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -1,4 +1,4 @@ -import { getFileFromTGZ, getFileFromZip, getFileFromArchive, bufferToStream } from '../src/utils' +import { getFileFromTGZ, getFileFromZip, getFileFromArchive, bufferToStream, setLogLevel } from '../src/utils' // .zip in Base64: github repo posthog/helloworldplugin const zip = @@ -55,3 +55,91 @@ test('bufferToStream', async () => { const stream = bufferToStream(buffer) expect(stream.read()).toEqual(buffer) }) + +test('setLogLevel', async () => { + function resetMocks() { + console.debug = jest.fn() + console.info = jest.fn() + console.log = jest.fn() + console.warn = jest.fn() + console.error = jest.fn() + } + + resetMocks() + setLogLevel('debug') + console.debug('debug') + console.info('debug') + console.log('debug') + console.warn('debug') + console.error('debug') + expect(console.debug).toHaveBeenCalledWith('debug') + expect(console.info).toHaveBeenCalledWith('debug') + expect(console.log).toHaveBeenCalledWith('debug') + expect(console.warn).toHaveBeenCalledWith('debug') + expect(console.error).toHaveBeenCalledWith('debug') + + resetMocks() + setLogLevel('info') + console.debug('info') + console.info('info') + console.log('info') + console.warn('info') + console.error('info') + expect((console.debug as any)._original).toBeDefined() + expect(console.info).toHaveBeenCalledWith('info') + expect(console.log).toHaveBeenCalledWith('info') + expect(console.warn).toHaveBeenCalledWith('info') + expect(console.error).toHaveBeenCalledWith('info') + + resetMocks() + setLogLevel('log') + console.debug('log') + console.info('log') + console.log('log') + console.warn('log') + console.error('log') + expect((console.debug as any)._original).toBeDefined() + expect((console.info as any)._original).toBeDefined() + expect(console.log).toHaveBeenCalledWith('log') + expect(console.warn).toHaveBeenCalledWith('log') + expect(console.error).toHaveBeenCalledWith('log') + + resetMocks() + setLogLevel('warn') + console.debug('warn') + console.info('warn') + console.log('warn') + console.warn('warn') + console.error('warn') + expect((console.debug as any)._original).toBeDefined() + expect((console.info as any)._original).toBeDefined() + expect((console.log as any)._original).toBeDefined() + expect(console.warn).toHaveBeenCalledWith('warn') + expect(console.error).toHaveBeenCalledWith('warn') + + resetMocks() + setLogLevel('error') + console.debug('error') + console.info('error') + console.log('error') + console.warn('error') + console.error('error') + expect((console.debug as any)._original).toBeDefined() + expect((console.info as any)._original).toBeDefined() + expect((console.log as any)._original).toBeDefined() + expect((console.warn as any)._original).toBeDefined() + expect(console.error).toHaveBeenCalledWith('error') + + resetMocks() + setLogLevel('none') + console.debug('none') + console.info('none') + console.log('none') + console.warn('none') + console.error('none') + expect((console.debug as any)._original).toBeDefined() + expect((console.info as any)._original).toBeDefined() + expect((console.log as any)._original).toBeDefined() + expect((console.warn as any)._original).toBeDefined() + expect((console.error as any)._original).toBeDefined() +}) From 7bc8e57a85c62f01177e55e150b6c90a6db382d2 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 23:00:31 +0100 Subject: [PATCH 09/21] log level --- src/utils.ts | 2 +- src/worker/piscina.js | 2 -- src/worker/worker.ts | 2 ++ tests/queue.test.ts | 4 ++-- tests/vm.test.ts | 4 ++-- tests/worker.test.ts | 2 +- 6 files changed, 8 insertions(+), 8 deletions(-) diff --git a/src/utils.ts b/src/utils.ts index 984c6baf..9c8873d8 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -2,7 +2,7 @@ import { Readable } from 'stream' import * as tar from 'tar-stream' import AdmZip from 'adm-zip' import * as zlib from 'zlib' -import { LogLevel } from 'types' +import { LogLevel } from './types' /** * @param binary Buffer diff --git a/src/worker/piscina.js b/src/worker/piscina.js index 1e0dcdde..56614345 100644 --- a/src/worker/piscina.js +++ b/src/worker/piscina.js @@ -15,8 +15,6 @@ if (isMainThread) { }, } } else { - console.info('🧵 Starting Piscina Worker Thread') - if (areWeTestingWithJest()) { require('ts-node').register() } diff --git a/src/worker/worker.ts b/src/worker/worker.ts index acb0f1ac..08bf2113 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -8,6 +8,8 @@ type TaskWorker = ({ task, args }: { task: string; args: any }) => Promise export async function createWorker(config: PluginsServerConfig): Promise { setLogLevel(config.LOG_LEVEL) + console.info('🧵 Starting Piscina Worker Thread') + const [server, closeServer] = await createServer(config) await setupPlugins(server) diff --git a/tests/queue.test.ts b/tests/queue.test.ts index ee2ff387..9e3f75e6 100644 --- a/tests/queue.test.ts +++ b/tests/queue.test.ts @@ -1,5 +1,5 @@ import { startQueue } from '../src/worker/queue' -import { createServer, defaultConfig } from '../src/server' +import { createServer } from '../src/server' import { PluginsServer } from '../src/types' import Client from '../src/celery/client' import { runPlugins } from '../src/plugins' @@ -14,7 +14,7 @@ beforeEach(async () => { // silence logs console.info = jest.fn() - mockServer = (await createServer(defaultConfig))[0] + mockServer = (await createServer())[0] }) test('worker and task passing via redis', async () => { diff --git a/tests/vm.test.ts b/tests/vm.test.ts index 168b84b3..ea009912 100644 --- a/tests/vm.test.ts +++ b/tests/vm.test.ts @@ -1,7 +1,7 @@ import { createPluginConfigVM, prepareForRun } from '../src/vm' import { PluginConfig, PluginsServer, Plugin } from '../src/types' import { PluginEvent } from 'posthog-plugins' -import { createServer, defaultConfig } from '../src/server' +import { createServer } from '../src/server' import * as fetch from 'node-fetch' const defaultEvent = { @@ -40,7 +40,7 @@ const mockConfig: PluginConfig = { } beforeEach(async () => { - mockServer = (await createServer(defaultConfig))[0] + mockServer = (await createServer())[0] }) afterEach(async () => { diff --git a/tests/worker.test.ts b/tests/worker.test.ts index a907cb77..72e59832 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -1,5 +1,5 @@ -import { defaultConfig } from '../src/server' import { makePiscina } from '../src/worker/piscina' +import { defaultConfig } from '../src/config' import { PluginEvent } from 'posthog-plugins/src/types' import { performance } from 'perf_hooks' import { mockJestWithIndex } from './helpers/plugins' From 9cf70e818d0fc1412677bd4f76e85ddd35924991 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 23:21:26 +0100 Subject: [PATCH 10/21] error on error --- src/server.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/server.ts b/src/server.ts index c6adebbf..b8bc172c 100644 --- a/src/server.ts +++ b/src/server.ts @@ -26,7 +26,7 @@ export async function createServer( const redis = new Redis(serverConfig.REDIS_URL) redis.on('error', (error) => { - console.info('🔴 Redis error!', error) + console.error('🔴 Redis error!', error) process.kill(process.pid, 'SIGTERM') }) From 471a7488c3494f9e371e1df4aa0fc41e44d96918 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 23:39:14 +0100 Subject: [PATCH 11/21] less logs for plugins --- tests/plugins.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 0d7dd617..0fda69fd 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -14,7 +14,7 @@ jest.mock('../src/sql') let mockServer: PluginsServer beforeEach(async () => { - ;[mockServer] = await createServer() + ;[mockServer] = await createServer({ LOG_LEVEL: 'log' }) }) test('setupPlugins and runPlugins', async () => { From ce4e679e4825c2db55cfbdf694065c3ac077ba07 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Mon, 7 Dec 2020 23:39:54 +0100 Subject: [PATCH 12/21] chunk event promise awaits --- tests/worker.test.ts | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/tests/worker.test.ts b/tests/worker.test.ts index 72e59832..b1869e13 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -23,14 +23,19 @@ function processOneEvent(processEvent: (event: PluginEvent) => Promise) { + const maxPromises = 1000 const startTime = performance.now() - const promises = Array(count) + const promises = Array(maxPromises) const processEvent = (event: PluginEvent) => piscina.runTask({ task: 'processEvent', args: { event } }) - for (let i = 0; i < count; i++) { - promises[i] = processOneEvent(processEvent) + + const groups = Math.ceil(count / maxPromises) + for (let j = 0; j < groups; j++) { + const groupCount = j === groups - 1 ? count % maxPromises : maxPromises + for (let i = 0; i < groupCount; i++) { + promises[i] = processOneEvent(processEvent) + } + await Promise.all(promises) } - // this will get heavy for tests > 10k events, should chunk them somehow... - await Promise.all(promises) const ms = Math.round((performance.now() - startTime) * 1000) / 1000 @@ -50,6 +55,7 @@ function setupPiscina(workers: number, code: string, tasksPerWorker: number) { ...defaultConfig, WORKER_CONCURRENCY: workers, TASKS_PER_WORKER: tasksPerWorker, + LOG_LEVEL: 'log', __jestMock: mockJestWithIndex(code), }) } From 22bdb8278c37121ad822e4a02cb3c4011cc46d57 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 08:53:13 +0100 Subject: [PATCH 13/21] remove dead cod --- tests/worker.test.ts | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/worker.test.ts b/tests/worker.test.ts index b1869e13..79f995d0 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -61,7 +61,6 @@ function setupPiscina(workers: number, code: string, tasksPerWorker: number) { } test('piscina worker test', async () => { - const testName = '' const coreCount = os.cpus().length const workers = [1, 2, 4, 8].filter((cores) => cores <= coreCount) From 7e37b5c074fd3c9270f44f618a1576d52f0fcb38 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 09:13:43 +0100 Subject: [PATCH 14/21] fix linter --- .eslintrc.js | 1 + src/utils.ts | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/.eslintrc.js b/.eslintrc.js index d16e41f2..22d9ea61 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -9,6 +9,7 @@ module.exports = { '@typescript-eslint/no-non-null-assertion': 'off', '@typescript-eslint/no-var-requires': 'off', '@typescript-eslint/no-explicit-any': 'off', + '@typescript-eslint/no-empty-function': 'off', curly: 'error', }, overrides: [ diff --git a/src/utils.ts b/src/utils.ts index 9c8873d8..aa9f5a45 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -96,7 +96,7 @@ export function getFileFromZip(archive: Buffer, file: string): string | null { return null } -export function setLogLevel(logLevel: LogLevel) { +export function setLogLevel(logLevel: LogLevel): void { for (const loopLevel of ['debug', 'info', 'log', 'warn', 'error', 'none']) { if (loopLevel === logLevel) { break From efa21ac89ab989fcdf5333f070194a5e8e41fca7 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 09:16:20 +0100 Subject: [PATCH 15/21] add comment --- src/index.ts | 1 + 1 file changed, 1 insertion(+) diff --git a/src/index.ts b/src/index.ts index 19a9c847..95e1d89e 100644 --- a/src/index.ts +++ b/src/index.ts @@ -32,6 +32,7 @@ app = app.help().command({ const parsedConfig: Record = config ? JSON.parse(config) : {} for (const [key, value] of Object.entries(otherArgs)) { if (typeof value !== 'undefined') { + // convert camelCase argument keys to under_score const newKey = key .replace(/(?:^|\.?)([A-Z])/g, (x, y) => '_' + y.toUpperCase()) .replace(/^_/, '') From 28a9e492898272c93715f99a6ec4d1a507d2af10 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:26:28 +0100 Subject: [PATCH 16/21] default to 100 async concurrency, which even makes fully sync tasks slightly faster --- src/config.ts | 2 +- tests/worker.test.ts | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/src/config.ts b/src/config.ts index c4a861fc..ba7de28b 100644 --- a/src/config.ts +++ b/src/config.ts @@ -15,7 +15,7 @@ function getDefaultConfig(): PluginsServerConfig { WEB_PORT: 3008, WEB_HOSTNAME: '0.0.0.0', WORKER_CONCURRENCY: 0, // use all cores - TASKS_PER_WORKER: 1, + TASKS_PER_WORKER: 100, LOG_LEVEL: 'info', } } diff --git a/tests/worker.test.ts b/tests/worker.test.ts index 79f995d0..97f581e0 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -63,7 +63,7 @@ function setupPiscina(workers: number, code: string, tasksPerWorker: number) { test('piscina worker test', async () => { const coreCount = os.cpus().length - const workers = [1, 2, 4, 8].filter((cores) => cores <= coreCount) + const workers = [1, 2, 4, 8, 12, 16].filter((cores) => cores <= coreCount) const rounds = 5 const tests: { testName: string; events: number; testCode: string }[] = [ @@ -90,7 +90,7 @@ test('piscina worker test', async () => { }, { testName: 'timeout100ms', - events: 5, + events: 2000, testCode: ` async function processEvent (event, meta) { await new Promise(resolve => __jestSetTimeout(() => resolve(), 100)) @@ -108,7 +108,7 @@ test('piscina worker test', async () => { coreCount, } for (const cores of workers) { - const piscina = setupPiscina(cores, testCode, 1) + const piscina = setupPiscina(cores, testCode, 100) // warmup await processCountEvents(cores * 4, piscina) From 5237333dd070c3aad325c6f3100c81e87aac0d50 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:27:51 +0100 Subject: [PATCH 17/21] disable just for line --- .eslintrc.js | 1 - src/utils.ts | 1 + 2 files changed, 1 insertion(+), 1 deletion(-) diff --git a/.eslintrc.js b/.eslintrc.js index 22d9ea61..d16e41f2 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -9,7 +9,6 @@ module.exports = { '@typescript-eslint/no-non-null-assertion': 'off', '@typescript-eslint/no-var-requires': 'off', '@typescript-eslint/no-explicit-any': 'off', - '@typescript-eslint/no-empty-function': 'off', curly: 'error', }, overrides: [ diff --git a/src/utils.ts b/src/utils.ts index aa9f5a45..e35cf5e1 100644 --- a/src/utils.ts +++ b/src/utils.ts @@ -102,6 +102,7 @@ export function setLogLevel(logLevel: LogLevel): void { break } const originalFunction = (console as any)[loopLevel]._original || (console as any)[loopLevel] + // eslint-disable-next-line @typescript-eslint/no-empty-function ;(console as any)[loopLevel] = () => {} ;(console as any)[loopLevel]._original = originalFunction } From 7753456b7221565cc89e6c38e5171e4adb625f89 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:42:51 +0100 Subject: [PATCH 18/21] add type for config key, parse env types properly --- src/config.ts | 29 ++++++++++++++++++++--------- src/types.ts | 1 + tests/config.test.ts | 29 +++++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 9 deletions(-) create mode 100644 tests/config.test.ts diff --git a/src/config.ts b/src/config.ts index ba7de28b..b500adc0 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,9 +1,9 @@ -import { PluginsServerConfig } from './types' +import { PluginsServerConfig, PluginsServerConfigKey } from './types' export const defaultConfig = overrideWithEnv(getDefaultConfig()) export const configHelp = getConfigHelp() -function getDefaultConfig(): PluginsServerConfig { +export function getDefaultConfig(): PluginsServerConfig { return { CELERY_DEFAULT_QUEUE: 'celery', DATABASE_URL: 'postgres://localhost:5432/posthog', @@ -20,7 +20,7 @@ function getDefaultConfig(): PluginsServerConfig { } } -function getConfigHelp(): Record { +export function getConfigHelp(): Record { return { CELERY_DEFAULT_QUEUE: 'celery outgoing queue', DATABASE_URL: 'url for postgres', @@ -37,12 +37,23 @@ function getConfigHelp(): Record { } } -function overrideWithEnv(config: PluginsServerConfig): PluginsServerConfig { - const newConfig: Record = { ...config } - for (const [key, value] of Object.entries(config)) { - if (process.env[key]) { - newConfig[key] = process.env[key] +export function overrideWithEnv( + config: PluginsServerConfig, + env: Record = process.env +): PluginsServerConfig { + const defaultConfig = getDefaultConfig() + + const newConfig: Record = { ...config } + for (const key of Object.keys(config) as PluginsServerConfigKey[]) { + if (typeof env[key] !== 'undefined') { + if (typeof defaultConfig[key] === 'number') { + newConfig[key] = env[key]?.indexOf('.') ? parseFloat(env[key]!) : parseInt(env[key]!) + } else if (typeof defaultConfig[key] === 'boolean') { + newConfig[key] = env[key] === 'true' || env[key] === 'True' || env[key] === '1' + } else { + newConfig[key] = env[key] + } } } - return newConfig as PluginsServerConfig + return newConfig } diff --git a/src/types.ts b/src/types.ts index a3fa1277..b21f5c1a 100644 --- a/src/types.ts +++ b/src/types.ts @@ -25,6 +25,7 @@ export interface PluginsServerConfig { getPluginAttachmentRows: PluginAttachmentDB[] } } +export type PluginsServerConfigKey = Exclude export interface PluginsServer extends PluginsServerConfig { // active connections to postgres and redis diff --git a/tests/config.test.ts b/tests/config.test.ts new file mode 100644 index 00000000..84d62432 --- /dev/null +++ b/tests/config.test.ts @@ -0,0 +1,29 @@ +import { getDefaultConfig, overrideWithEnv } from '../src/config' + +test('overrideWithEnv 1', async () => { + const defaultConfig = getDefaultConfig() + const env = { + DISABLE_WEB: 'false', + WEB_PORT: '3008', + WEB_HOSTNAME: '0.0.0.0', + BASE_DIR: undefined, + } + const config = overrideWithEnv(getDefaultConfig(), env) + + expect(config.DISABLE_WEB).toEqual(false) + expect(config.WEB_PORT).toEqual(3008) + expect(config.WEB_HOSTNAME).toEqual('0.0.0.0') + expect(config.BASE_DIR).toEqual(defaultConfig.BASE_DIR) +}) + +test('overrideWithEnv 2', async () => { + const defaultConfig = getDefaultConfig() + const env = { + DISABLE_WEB: '1', + WEB_PORT: '3008.12', + } + const config = overrideWithEnv(getDefaultConfig(), env) + + expect(config.DISABLE_WEB).toEqual(true) + expect(config.WEB_PORT).toEqual(3008.12) +}) From ad29c7e995b78a5eb4457b08a93dbed4f8854210 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:48:44 +0100 Subject: [PATCH 19/21] just always start --- src/index.ts | 42 +++++++++++++++++++----------------------- 1 file changed, 19 insertions(+), 23 deletions(-) diff --git a/src/index.ts b/src/index.ts index 95e1d89e..b9fc22a7 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,5 +1,5 @@ import * as yargs from 'yargs' -import { PluginsServerConfig } from './types' +import { PluginsServerConfig, PluginsServerConfigKey } from './types' import { startPluginsServer } from './server' import { makePiscina } from './worker/piscina' import { defaultConfig, configHelp } from './config' @@ -19,30 +19,26 @@ let app: any = yargs .option('config', { alias: 'c', describe: 'Config options JSON.', type: 'string' }) for (const [key, value] of Object.entries(defaultConfig)) { - app = app.option(key.toLowerCase().replaceAll('_', '-'), { - describe: `${configHelp[key] || key} [${value}]`, + app = app.option(key.toLowerCase().split('_').join('-'), { + describe: `${configHelp[key as PluginsServerConfigKey] || key} [${value}]`, type: typeof value, }) } -app = app.help().command({ - command: ['start', '$0'], - describe: 'start the server', - handler: ({ config, ...otherArgs }: Argv) => { - const parsedConfig: Record = config ? JSON.parse(config) : {} - for (const [key, value] of Object.entries(otherArgs)) { - if (typeof value !== 'undefined') { - // convert camelCase argument keys to under_score - const newKey = key - .replace(/(?:^|\.?)([A-Z])/g, (x, y) => '_' + y.toUpperCase()) - .replace(/^_/, '') - .toUpperCase() - if (newKey in defaultConfig) { - parsedConfig[newKey] = value - } - } +const { config, ...otherArgs }: Argv = app.help().argv + +const parsedConfig: Record = config ? JSON.parse(config) : {} +for (const [key, value] of Object.entries(otherArgs)) { + if (typeof value !== 'undefined') { + // convert camelCase argument keys to under_score + const newKey = key + .replace(/(?:^|\.?)([A-Z])/g, (x, y) => '_' + y.toUpperCase()) + .replace(/^_/, '') + .toUpperCase() + if (newKey in defaultConfig) { + parsedConfig[newKey] = value } - setLogLevel(parsedConfig.LOG_LEVEL || defaultConfig.LOG_LEVEL) - startPluginsServer(parsedConfig as PluginsServerConfig, makePiscina) - }, -}).argv + } +} +setLogLevel(parsedConfig.LOG_LEVEL || defaultConfig.LOG_LEVEL) +startPluginsServer(parsedConfig as PluginsServerConfig, makePiscina) From 19b1bb7eebde29897e182acf020b5ac07bb33c17 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:49:12 +0100 Subject: [PATCH 20/21] remove dead code --- tests/worker.test.ts | 4 ---- 1 file changed, 4 deletions(-) diff --git a/tests/worker.test.ts b/tests/worker.test.ts index 97f581e0..5b1dc627 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -126,8 +126,4 @@ test('piscina worker test', async () => { console.log(JSON.stringify({ result }, null, 2)) } console.table(results) - // expect that adding more cores (up to coreCount) increases throughput - // for (let i = 1; i < workers.length; i++) { - // expect(result[`${workers[i - 1]} cores`]).toBeLessThan(result[`${workers[i]} cores`]) - // } }) From b805f470ed3555989ce9dde89307d3d324bf1082 Mon Sep 17 00:00:00 2001 From: Marius Andra Date: Tue, 8 Dec 2020 11:54:05 +0100 Subject: [PATCH 21/21] enum for loglevel --- src/config.ts | 4 ++-- src/types.ts | 9 ++++++++- tests/plugins.test.ts | 4 ++-- tests/utils.test.ts | 13 +++++++------ tests/worker.test.ts | 3 ++- 5 files changed, 21 insertions(+), 12 deletions(-) diff --git a/src/config.ts b/src/config.ts index b500adc0..ee66b571 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,4 +1,4 @@ -import { PluginsServerConfig, PluginsServerConfigKey } from './types' +import { LogLevel, PluginsServerConfig, PluginsServerConfigKey } from './types' export const defaultConfig = overrideWithEnv(getDefaultConfig()) export const configHelp = getConfigHelp() @@ -16,7 +16,7 @@ export function getDefaultConfig(): PluginsServerConfig { WEB_HOSTNAME: '0.0.0.0', WORKER_CONCURRENCY: 0, // use all cores TASKS_PER_WORKER: 100, - LOG_LEVEL: 'info', + LOG_LEVEL: LogLevel.Info, } } diff --git a/src/types.ts b/src/types.ts index b21f5c1a..be331d86 100644 --- a/src/types.ts +++ b/src/types.ts @@ -3,7 +3,14 @@ import { Redis } from 'ioredis' import { PluginEvent, PluginAttachment, PluginConfigSchema } from 'posthog-plugins' import { VM, VMScript } from 'vm2' -export type LogLevel = 'debug' | 'info' | 'log' | 'warn' | 'error' | 'none' +export enum LogLevel { + Debug = 'debug', + Info = 'info', + Log = 'log', + Warn = 'warn', + Error = 'error', + None = 'none', +} export interface PluginsServerConfig { WORKER_CONCURRENCY: number diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 0fda69fd..0f58f3cc 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -1,6 +1,6 @@ import { runPlugins, setupPlugins } from '../src/plugins' import { createServer } from '../src/server' -import { PluginsServer } from '../src/types' +import { LogLevel, PluginsServer } from '../src/types' import { PluginEvent } from 'posthog-plugins/src/types' import { mockPluginTempFolder, @@ -14,7 +14,7 @@ jest.mock('../src/sql') let mockServer: PluginsServer beforeEach(async () => { - ;[mockServer] = await createServer({ LOG_LEVEL: 'log' }) + ;[mockServer] = await createServer({ LOG_LEVEL: LogLevel.Log }) }) test('setupPlugins and runPlugins', async () => { diff --git a/tests/utils.test.ts b/tests/utils.test.ts index b1cf411e..a89b2c8a 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -1,4 +1,5 @@ import { getFileFromTGZ, getFileFromZip, getFileFromArchive, bufferToStream, setLogLevel } from '../src/utils' +import { LogLevel } from '../src/types' // .zip in Base64: github repo posthog/helloworldplugin const zip = @@ -66,7 +67,7 @@ test('setLogLevel', async () => { } resetMocks() - setLogLevel('debug') + setLogLevel(LogLevel.Debug) console.debug('debug') console.info('debug') console.log('debug') @@ -79,7 +80,7 @@ test('setLogLevel', async () => { expect(console.error).toHaveBeenCalledWith('debug') resetMocks() - setLogLevel('info') + setLogLevel(LogLevel.Info) console.debug('info') console.info('info') console.log('info') @@ -92,7 +93,7 @@ test('setLogLevel', async () => { expect(console.error).toHaveBeenCalledWith('info') resetMocks() - setLogLevel('log') + setLogLevel(LogLevel.Log) console.debug('log') console.info('log') console.log('log') @@ -105,7 +106,7 @@ test('setLogLevel', async () => { expect(console.error).toHaveBeenCalledWith('log') resetMocks() - setLogLevel('warn') + setLogLevel(LogLevel.Warn) console.debug('warn') console.info('warn') console.log('warn') @@ -118,7 +119,7 @@ test('setLogLevel', async () => { expect(console.error).toHaveBeenCalledWith('warn') resetMocks() - setLogLevel('error') + setLogLevel(LogLevel.Error) console.debug('error') console.info('error') console.log('error') @@ -131,7 +132,7 @@ test('setLogLevel', async () => { expect(console.error).toHaveBeenCalledWith('error') resetMocks() - setLogLevel('none') + setLogLevel(LogLevel.None) console.debug('none') console.info('none') console.log('none') diff --git a/tests/worker.test.ts b/tests/worker.test.ts index 5b1dc627..ee9c0612 100644 --- a/tests/worker.test.ts +++ b/tests/worker.test.ts @@ -4,6 +4,7 @@ import { PluginEvent } from 'posthog-plugins/src/types' import { performance } from 'perf_hooks' import { mockJestWithIndex } from './helpers/plugins' import * as os from 'os' +import { LogLevel } from '../src/types' jest.mock('../src/sql') jest.setTimeout(300000) // 300 sec timeout @@ -55,7 +56,7 @@ function setupPiscina(workers: number, code: string, tasksPerWorker: number) { ...defaultConfig, WORKER_CONCURRENCY: workers, TASKS_PER_WORKER: tasksPerWorker, - LOG_LEVEL: 'log', + LOG_LEVEL: LogLevel.Log, __jestMock: mockJestWithIndex(code), }) }