From f6ddce51218b9c628d704a2f4e99c91b61eb0044 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 10:44:51 +0200 Subject: [PATCH 01/19] Move schedule.ts under src/main --- src/{ => main}/services/schedule.ts | 8 ++++---- src/server.ts | 2 +- tests/postgres/worker.test.ts | 2 +- tests/schedule.test.ts | 4 ++-- 4 files changed, 8 insertions(+), 8 deletions(-) rename src/{ => main}/services/schedule.ts (97%) diff --git a/src/services/schedule.ts b/src/main/services/schedule.ts similarity index 97% rename from src/services/schedule.ts rename to src/main/services/schedule.ts index 94c6014b..55c47d53 100644 --- a/src/services/schedule.ts +++ b/src/main/services/schedule.ts @@ -3,10 +3,10 @@ import * as Sentry from '@sentry/node' import * as schedule from 'node-schedule' import Redlock from 'redlock' -import { processError } from '../error' -import { status } from '../status' -import { PluginConfigId, PluginsServer, ScheduleControl } from '../types' -import { createRedis, delay } from '../utils' +import { processError } from '../../error' +import { status } from '../../status' +import { PluginConfigId, PluginsServer, ScheduleControl } from '../../types' +import { createRedis, delay } from '../../utils' export const LOCKED_RESOURCE = 'plugin-server:locks:schedule' diff --git a/src/server.ts b/src/server.ts index 92e42048..6390fd31 100644 --- a/src/server.ts +++ b/src/server.ts @@ -17,7 +17,7 @@ import { ConnectionOptions } from 'tls' import { defaultConfig } from './config' import { DB } from './db' import { EventsProcessor } from './ingestion/process-event' -import { startSchedule } from './services/schedule' +import { startSchedule } from './main/services/schedule' import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' import { createPostgresPool, createRedis, delay, UUIDT } from './utils' diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 6bc3cbcf..b0b55fda 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -4,10 +4,10 @@ import { mocked } from 'ts-jest/utils' import Client from '../../src/celery/client' import { ingestEvent } from '../../src/ingestion/ingest-event' +import { loadPluginSchedule } from '../../src/main/services/schedule' import { runPlugins, runPluginsOnBatch, runPluginTask } from '../../src/plugins/run' import { loadSchedule, setupPlugins } from '../../src/plugins/setup' import { ServerInstance, startPluginsServer } from '../../src/server' -import { loadPluginSchedule } from '../../src/services/schedule' import { LogLevel } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' diff --git a/tests/schedule.test.ts b/tests/schedule.test.ts index a4116e97..7c46550c 100644 --- a/tests/schedule.test.ts +++ b/tests/schedule.test.ts @@ -1,13 +1,13 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' -import { createServer } from '../src/server' import { loadPluginSchedule, LOCKED_RESOURCE, runTasksDebounced, startSchedule, waitForTasksToFinish, -} from '../src/services/schedule' +} from '../src/main/services/schedule' +import { createServer } from '../src/server' import { LogLevel, ScheduleControl } from '../src/types' import { delay } from '../src/utils' import { createPromise } from './helpers/promises' From 119af790a8463d58e85d0ed0056e229d60fd75e1 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 10:49:00 +0200 Subject: [PATCH 02/19] Move queue.ts under src/main --- src/{worker => main}/queue.ts | 0 src/server.ts | 2 +- tests/postgres/queue.test.ts | 2 +- 3 files changed, 2 insertions(+), 2 deletions(-) rename src/{worker => main}/queue.ts (100%) diff --git a/src/worker/queue.ts b/src/main/queue.ts similarity index 100% rename from src/worker/queue.ts rename to src/main/queue.ts diff --git a/src/server.ts b/src/server.ts index 6390fd31..f2c3893f 100644 --- a/src/server.ts +++ b/src/server.ts @@ -17,12 +17,12 @@ import { ConnectionOptions } from 'tls' import { defaultConfig } from './config' import { DB } from './db' import { EventsProcessor } from './ingestion/process-event' +import { startQueue } from './main/queue' import { startSchedule } from './main/services/schedule' import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' import { createPostgresPool, createRedis, delay, UUIDT } from './utils' import { startFastifyInstance, stopFastifyInstance } from './web/server' -import { startQueue } from './worker/queue' const { version } = require('../package.json') diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index a81e2a43..3733eec6 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -1,9 +1,9 @@ import Client from '../../src/celery/client' +import { startQueue } from '../../src/main/queue' import { runPlugins } from '../../src/plugins/run' import { createServer } from '../../src/server' import { LogLevel, PluginsServer } from '../../src/types' import { delay } from '../../src/utils' -import { startQueue } from '../../src/worker/queue' jest.setTimeout(60000) // 60 sec timeout From 817c0cda55bda6489ea3a59b319adfc8825cc98f Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 10:49:51 +0200 Subject: [PATCH 03/19] Move kafka-queue logic under src/main --- src/{ => main}/ingestion/kafka-queue.ts | 6 +++--- src/main/queue.ts | 2 +- 2 files changed, 4 insertions(+), 4 deletions(-) rename src/{ => main}/ingestion/kafka-queue.ts (98%) diff --git a/src/ingestion/kafka-queue.ts b/src/main/ingestion/kafka-queue.ts similarity index 98% rename from src/ingestion/kafka-queue.ts rename to src/main/ingestion/kafka-queue.ts index 469b55d8..faee8a0a 100644 --- a/src/ingestion/kafka-queue.ts +++ b/src/main/ingestion/kafka-queue.ts @@ -3,9 +3,9 @@ import * as Sentry from '@sentry/node' import { Consumer, EachBatchPayload, Kafka } from 'kafkajs' import { PluginsServer, Queue } from 'types' -import { status } from '../status' -import { groupIntoBatches, killGracefully } from '../utils' -import { timeoutGuard } from './utils' +import { timeoutGuard } from '../../ingestion/utils' +import { status } from '../../status' +import { groupIntoBatches, killGracefully } from '../../utils' export class KafkaQueue implements Queue { private pluginsServer: PluginsServer diff --git a/src/main/queue.ts b/src/main/queue.ts index 97e3ccb3..836690c1 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -5,10 +5,10 @@ import * as Sentry from '@sentry/node' import Client from '../celery/client' import Worker from '../celery/worker' import { IngestEventResponse } from '../ingestion/ingest-event' -import { KafkaQueue } from '../ingestion/kafka-queue' import { status } from '../status' import { PluginsServer, Queue } from '../types' import { UUIDT } from '../utils' +import { KafkaQueue } from './ingestion/kafka-queue' export type WorkerMethods = { processEvent: (event: PluginEvent) => Promise From 60aa7d6e7a779f24e78fc503b8f99f375a6ce567 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 10:55:42 +0200 Subject: [PATCH 04/19] Move ingest-event file --- src/main/queue.ts | 3 +-- src/types.ts | 2 ++ src/{ => worker}/ingestion/ingest-event.ts | 8 +++----- src/worker/worker.ts | 2 +- tests/postgres/worker.test.ts | 4 ++-- 5 files changed, 9 insertions(+), 10 deletions(-) rename src/{ => worker}/ingestion/ingest-event.ts (86%) diff --git a/src/main/queue.ts b/src/main/queue.ts index 836690c1..82b07ca5 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -4,9 +4,8 @@ import * as Sentry from '@sentry/node' import Client from '../celery/client' import Worker from '../celery/worker' -import { IngestEventResponse } from '../ingestion/ingest-event' import { status } from '../status' -import { PluginsServer, Queue } from '../types' +import { IngestEventResponse, PluginsServer, Queue } from '../types' import { UUIDT } from '../utils' import { KafkaQueue } from './ingestion/kafka-queue' diff --git a/src/types.ts b/src/types.ts index e2bf5629..24208a92 100644 --- a/src/types.ts +++ b/src/types.ts @@ -361,3 +361,5 @@ export interface ScheduleControl { stopSchedule: () => Promise reloadSchedule: () => Promise } + +export type IngestEventResponse = { success?: boolean; error?: string } diff --git a/src/ingestion/ingest-event.ts b/src/worker/ingestion/ingest-event.ts similarity index 86% rename from src/ingestion/ingest-event.ts rename to src/worker/ingestion/ingest-event.ts index 46e83158..1f04ec50 100644 --- a/src/ingestion/ingest-event.ts +++ b/src/worker/ingestion/ingest-event.ts @@ -2,11 +2,9 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import { DateTime } from 'luxon' -import { status } from '../status' -import { PluginsServer } from '../types' -import { timeoutGuard } from './utils' - -export type IngestEventResponse = { success?: boolean; error?: string } +import { timeoutGuard } from '../../ingestion/utils' +import { status } from '../../status' +import { IngestEventResponse, PluginsServer } from '../../types' export async function ingestEvent(server: PluginsServer, event: PluginEvent): Promise { const timeout = timeoutGuard('Still ingesting event inside worker. Timeout warning after 30 sec!', { diff --git a/src/worker/worker.ts b/src/worker/worker.ts index ea4c3071..6202f4c6 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,4 +1,3 @@ -import { ingestEvent } from '../ingestion/ingest-event' import { initApp } from '../init' import { runPlugins, runPluginsOnBatch, runPluginTask } from '../plugins/run' import { loadSchedule, setupPlugins } from '../plugins/setup' @@ -6,6 +5,7 @@ import { createServer } from '../server' import { status } from '../status' import { PluginsServer, PluginsServerConfig } from '../types' import { cloneObject } from '../utils' +import { ingestEvent } from './ingestion/ingest-event' type TaskWorker = ({ task, args }: { task: string; args: any }) => Promise diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index b0b55fda..2b5580b3 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -3,13 +3,13 @@ import IORedis from 'ioredis' import { mocked } from 'ts-jest/utils' import Client from '../../src/celery/client' -import { ingestEvent } from '../../src/ingestion/ingest-event' import { loadPluginSchedule } from '../../src/main/services/schedule' import { runPlugins, runPluginsOnBatch, runPluginTask } from '../../src/plugins/run' import { loadSchedule, setupPlugins } from '../../src/plugins/setup' import { ServerInstance, startPluginsServer } from '../../src/server' import { LogLevel } from '../../src/types' import { delay, UUIDT } from '../../src/utils' +import { ingestEvent } from '../../src/worker/ingestion/ingest-event' import { makePiscina } from '../../src/worker/piscina' import { createTaskRunner } from '../../src/worker/worker' import { resetTestDatabase } from '../helpers/sql' @@ -17,7 +17,7 @@ import { setupPiscina } from '../helpers/worker' jest.mock('../../src/sql') jest.mock('../../src/status') -jest.mock('../../src/ingestion/ingest-event') +jest.mock('../../src/worker/ingestion/ingest-event') jest.mock('../../src/plugins/run') jest.mock('../../src/plugins/setup') jest.setTimeout(600000) // 600 sec timeout From 7f782874209de8f2727a2b5e0de30768da22fc13 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 10:58:44 +0200 Subject: [PATCH 05/19] Move web/server under main.ts --- src/{ => main}/web/server.ts | 2 +- src/server.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) rename src/{ => main}/web/server.ts (96%) diff --git a/src/web/server.ts b/src/main/web/server.ts similarity index 96% rename from src/web/server.ts rename to src/main/web/server.ts index 7effae00..1bf88653 100644 --- a/src/web/server.ts +++ b/src/main/web/server.ts @@ -1,7 +1,7 @@ import { fastify, FastifyInstance } from 'fastify' import { PluginsServer } from 'types' -import { status } from '../status' +import { status } from '../../status' export function buildFastifyInstance(): FastifyInstance { const fastifyInstance = fastify() diff --git a/src/server.ts b/src/server.ts index f2c3893f..508bbdf3 100644 --- a/src/server.ts +++ b/src/server.ts @@ -19,10 +19,10 @@ import { DB } from './db' import { EventsProcessor } from './ingestion/process-event' import { startQueue } from './main/queue' import { startSchedule } from './main/services/schedule' +import { startFastifyInstance, stopFastifyInstance } from './main/web/server' import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' import { createPostgresPool, createRedis, delay, UUIDT } from './utils' -import { startFastifyInstance, stopFastifyInstance } from './web/server' const { version } = require('../package.json') From 035b3e7e7ba7fff69b1f1a7251816d5fe2b6ef16 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:01:33 +0200 Subject: [PATCH 06/19] Move plugins code under src/worker --- src/{ => worker}/plugins/loadPlugin.ts | 6 +++--- src/{ => worker}/plugins/run.ts | 4 ++-- src/{ => worker}/plugins/setup.ts | 8 ++++---- src/worker/worker.ts | 4 ++-- tests/plugins.test.ts | 10 +++++----- tests/postgres/queue.test.ts | 2 +- tests/postgres/worker.test.ts | 8 ++++---- 7 files changed, 21 insertions(+), 21 deletions(-) rename src/{ => worker}/plugins/loadPlugin.ts (96%) rename src/{ => worker}/plugins/run.ts (96%) rename src/{ => worker}/plugins/setup.ts (96%) diff --git a/src/plugins/loadPlugin.ts b/src/worker/plugins/loadPlugin.ts similarity index 96% rename from src/plugins/loadPlugin.ts rename to src/worker/plugins/loadPlugin.ts index e0bc2c74..bdfe6f12 100644 --- a/src/plugins/loadPlugin.ts +++ b/src/worker/plugins/loadPlugin.ts @@ -1,9 +1,9 @@ import * as fs from 'fs' import * as path from 'path' -import { processError } from '../error' -import { PluginConfig, PluginJsonConfig, PluginsServer } from '../types' -import { getFileFromArchive, pluginDigest } from '../utils' +import { processError } from '../../error' +import { PluginConfig, PluginJsonConfig, PluginsServer } from '../../types' +import { getFileFromArchive, pluginDigest } from '../../utils' export async function loadPlugin(server: PluginsServer, pluginConfig: PluginConfig): Promise { const { plugin } = pluginConfig diff --git a/src/plugins/run.ts b/src/worker/plugins/run.ts similarity index 96% rename from src/plugins/run.ts rename to src/worker/plugins/run.ts index 5cf0b33f..3f0d213f 100644 --- a/src/plugins/run.ts +++ b/src/worker/plugins/run.ts @@ -1,7 +1,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold' -import { processError } from '../error' -import { PluginConfig, PluginsServer } from '../types' +import { processError } from '../../error' +import { PluginConfig, PluginsServer } from '../../types' export async function runPlugins(server: PluginsServer, event: PluginEvent): Promise { const pluginsToRun = getPluginsForTeam(server, event.team_id) diff --git a/src/plugins/setup.ts b/src/worker/plugins/setup.ts similarity index 96% rename from src/plugins/setup.ts rename to src/worker/plugins/setup.ts index 777bdd20..23d1866a 100644 --- a/src/plugins/setup.ts +++ b/src/worker/plugins/setup.ts @@ -1,9 +1,9 @@ import { PluginAttachment } from '@posthog/plugin-scaffold' -import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows } from '../sql' -import { status } from '../status' -import { Plugin, PluginConfig, PluginConfigId, PluginId, PluginsServer, TeamId } from '../types' -import { LazyPluginVM } from '../vm/lazy' +import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows } from '../../sql' +import { status } from '../../status' +import { Plugin, PluginConfig, PluginConfigId, PluginId, PluginsServer, TeamId } from '../../types' +import { LazyPluginVM } from '../../vm/lazy' import { loadPlugin } from './loadPlugin' export async function setupPlugins(server: PluginsServer): Promise { diff --git a/src/worker/worker.ts b/src/worker/worker.ts index 6202f4c6..8c897262 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,11 +1,11 @@ import { initApp } from '../init' -import { runPlugins, runPluginsOnBatch, runPluginTask } from '../plugins/run' -import { loadSchedule, setupPlugins } from '../plugins/setup' import { createServer } from '../server' import { status } from '../status' import { PluginsServer, PluginsServerConfig } from '../types' import { cloneObject } from '../utils' import { ingestEvent } from './ingestion/ingest-event' +import { runPlugins, runPluginsOnBatch, runPluginTask } from './plugins/run' +import { loadSchedule, setupPlugins } from './plugins/setup' type TaskWorker = ({ task, args }: { task: string; args: any }) => Promise diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 18f72260..d5f3e38c 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -2,11 +2,11 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { mocked } from 'ts-jest/utils' import { clearError, processError } from '../src/error' -import { loadPlugin } from '../src/plugins/loadPlugin' -import { runPlugins } from '../src/plugins/run' -import { loadSchedule, setupPlugins } from '../src/plugins/setup' import { createServer } from '../src/server' import { LogLevel, PluginsServer } from '../src/types' +import { loadPlugin } from '../src/worker/plugins/loadPlugin' +import { runPlugins } from '../src/worker/plugins/run' +import { loadSchedule, setupPlugins } from '../src/worker/plugins/setup' import { commonOrganizationId, mockPluginTempFolder, @@ -20,8 +20,8 @@ import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows, setError } jest.mock('../src/sql') jest.mock('../src/status') jest.mock('../src/error') -jest.mock('../src/plugins/loadPlugin', () => { - const { loadPlugin } = jest.requireActual('../src/plugins/loadPlugin') +jest.mock('../src/worker/plugins/loadPlugin', () => { + const { loadPlugin } = jest.requireActual('../src/worker/plugins/loadPlugin') return { loadPlugin: jest.fn().mockImplementation(loadPlugin) } }) diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index 3733eec6..1a2086b8 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -1,9 +1,9 @@ import Client from '../../src/celery/client' import { startQueue } from '../../src/main/queue' -import { runPlugins } from '../../src/plugins/run' import { createServer } from '../../src/server' import { LogLevel, PluginsServer } from '../../src/types' import { delay } from '../../src/utils' +import { runPlugins } from '../../src/worker/plugins/run' jest.setTimeout(60000) // 60 sec timeout diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 2b5580b3..d8840a81 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -4,13 +4,13 @@ import { mocked } from 'ts-jest/utils' import Client from '../../src/celery/client' import { loadPluginSchedule } from '../../src/main/services/schedule' -import { runPlugins, runPluginsOnBatch, runPluginTask } from '../../src/plugins/run' -import { loadSchedule, setupPlugins } from '../../src/plugins/setup' import { ServerInstance, startPluginsServer } from '../../src/server' import { LogLevel } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { ingestEvent } from '../../src/worker/ingestion/ingest-event' import { makePiscina } from '../../src/worker/piscina' +import { runPlugins, runPluginsOnBatch, runPluginTask } from '../../src/worker/plugins/run' +import { loadSchedule, setupPlugins } from '../../src/worker/plugins/setup' import { createTaskRunner } from '../../src/worker/worker' import { resetTestDatabase } from '../helpers/sql' import { setupPiscina } from '../helpers/worker' @@ -18,8 +18,8 @@ import { setupPiscina } from '../helpers/worker' jest.mock('../../src/sql') jest.mock('../../src/status') jest.mock('../../src/worker/ingestion/ingest-event') -jest.mock('../../src/plugins/run') -jest.mock('../../src/plugins/setup') +jest.mock('../../src/worker/plugins/run') +jest.mock('../../src/worker/plugins/setup') jest.setTimeout(600000) // 600 sec timeout function createEvent(index = 0): PluginEvent { From 72f47fed28726ebb31a597725b7ce0cdd60cfd05 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:05:50 +0200 Subject: [PATCH 07/19] Move vm code under src/worker --- benchmarks/clickhouse/e2e.kafka.benchmark.ts | 2 +- benchmarks/clickhouse/e2e.timeout.benchmark.ts | 2 +- benchmarks/postgres/e2e.celery.benchmark.ts | 2 +- benchmarks/vm/memory.benchmark.ts | 2 +- src/types.ts | 2 +- src/worker/plugins/setup.ts | 2 +- src/{ => worker}/vm/extensions/cache.ts | 2 +- src/{ => worker}/vm/extensions/console.ts | 0 src/{ => worker}/vm/extensions/google.ts | 0 src/{ => worker}/vm/extensions/posthog.ts | 6 +++--- src/{ => worker}/vm/extensions/storage.ts | 2 +- src/{ => worker}/vm/lazy.ts | 6 +++--- src/{ => worker}/vm/transforms/common.ts | 2 +- src/{ => worker}/vm/transforms/index.ts | 2 +- src/{ => worker}/vm/transforms/loop-timeout.ts | 0 src/{ => worker}/vm/transforms/promise-timeout.ts | 0 src/{ => worker}/vm/vm.ts | 2 +- tests/clickhouse/e2e.test.ts | 2 +- tests/clickhouse/postgres-parity.test.ts | 2 +- tests/postgres/e2e.test.ts | 2 +- tests/postgres/e2e.timeout.test.ts | 2 +- tests/postgres/vm.lazy.test.ts | 6 +++--- tests/postgres/vm.test.ts | 2 +- tests/postgres/vm.timeout.test.ts | 2 +- tests/transforms.test.ts | 2 +- 25 files changed, 27 insertions(+), 27 deletions(-) rename src/{ => worker}/vm/extensions/cache.ts (94%) rename src/{ => worker}/vm/extensions/console.ts (100%) rename src/{ => worker}/vm/extensions/google.ts (100%) rename src/{ => worker}/vm/extensions/posthog.ts (95%) rename src/{ => worker}/vm/extensions/storage.ts (95%) rename src/{ => worker}/vm/lazy.ts (93%) rename src/{ => worker}/vm/transforms/common.ts (79%) rename src/{ => worker}/vm/transforms/index.ts (95%) rename src/{ => worker}/vm/transforms/loop-timeout.ts (100%) rename src/{ => worker}/vm/transforms/promise-timeout.ts (100%) rename src/{ => worker}/vm/vm.ts (99%) diff --git a/benchmarks/clickhouse/e2e.kafka.benchmark.ts b/benchmarks/clickhouse/e2e.kafka.benchmark.ts index 0ab69288..8dbb1ab7 100644 --- a/benchmarks/clickhouse/e2e.kafka.benchmark.ts +++ b/benchmarks/clickhouse/e2e.kafka.benchmark.ts @@ -5,8 +5,8 @@ import { startPluginsServer } from '../../src/server' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../../tests/helpers/clickhouse' import { resetKafka } from '../../tests/helpers/kafka' import { pluginConfig39 } from '../../tests/helpers/plugins' diff --git a/benchmarks/clickhouse/e2e.timeout.benchmark.ts b/benchmarks/clickhouse/e2e.timeout.benchmark.ts index ca18324e..759a1e6c 100644 --- a/benchmarks/clickhouse/e2e.timeout.benchmark.ts +++ b/benchmarks/clickhouse/e2e.timeout.benchmark.ts @@ -5,8 +5,8 @@ import { startPluginsServer } from '../../src/server' import { ClickHouseEvent, LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../../tests/helpers/clickhouse' import { resetKafka } from '../../tests/helpers/kafka' import { pluginConfig39 } from '../../tests/helpers/plugins' diff --git a/benchmarks/postgres/e2e.celery.benchmark.ts b/benchmarks/postgres/e2e.celery.benchmark.ts index f647c80d..63c12405 100644 --- a/benchmarks/postgres/e2e.celery.benchmark.ts +++ b/benchmarks/postgres/e2e.celery.benchmark.ts @@ -5,8 +5,8 @@ import { startPluginsServer } from '../../src/server' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../../tests/helpers/plugins' import { resetTestDatabase } from '../../tests/helpers/sql' import { delayUntilEventIngested } from '../../tests/shared/process-event' diff --git a/benchmarks/vm/memory.benchmark.ts b/benchmarks/vm/memory.benchmark.ts index 412bfa2c..7a4db84d 100644 --- a/benchmarks/vm/memory.benchmark.ts +++ b/benchmarks/vm/memory.benchmark.ts @@ -2,7 +2,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { createServer } from '../../src/server' import { Plugin, PluginConfig, PluginConfigVMReponse } from '../../src/types' -import { createPluginConfigVM } from '../../src/vm/vm' +import { createPluginConfigVM } from '../../src/worker/vm/vm' import { commonOrganizationId } from '../../tests/helpers/plugins' jest.mock('../../src/sql') diff --git a/src/types.ts b/src/types.ts index 24208a92..d2325a9f 100644 --- a/src/types.ts +++ b/src/types.ts @@ -10,7 +10,7 @@ import { Pool } from 'pg' import { VM } from 'vm2' import { DB } from './db' -import { LazyPluginVM } from './vm/lazy' +import { LazyPluginVM } from './worker/vm/lazy' export enum LogLevel { Debug = 'debug', diff --git a/src/worker/plugins/setup.ts b/src/worker/plugins/setup.ts index 23d1866a..80320955 100644 --- a/src/worker/plugins/setup.ts +++ b/src/worker/plugins/setup.ts @@ -3,7 +3,7 @@ import { PluginAttachment } from '@posthog/plugin-scaffold' import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows } from '../../sql' import { status } from '../../status' import { Plugin, PluginConfig, PluginConfigId, PluginId, PluginsServer, TeamId } from '../../types' -import { LazyPluginVM } from '../../vm/lazy' +import { LazyPluginVM } from '../vm/lazy' import { loadPlugin } from './loadPlugin' export async function setupPlugins(server: PluginsServer): Promise { diff --git a/src/vm/extensions/cache.ts b/src/worker/vm/extensions/cache.ts similarity index 94% rename from src/vm/extensions/cache.ts rename to src/worker/vm/extensions/cache.ts index 00911a6c..ca19b5d2 100644 --- a/src/vm/extensions/cache.ts +++ b/src/worker/vm/extensions/cache.ts @@ -1,6 +1,6 @@ import { CacheExtension } from '@posthog/plugin-scaffold' -import { PluginsServer } from '../../types' +import { PluginsServer } from '../../../types' export function createCache(server: PluginsServer, pluginId: number, teamId: number): CacheExtension { const getKey = (key: string) => `@plugin/${pluginId}/${typeof teamId === 'undefined' ? '@all' : teamId}/${key}` diff --git a/src/vm/extensions/console.ts b/src/worker/vm/extensions/console.ts similarity index 100% rename from src/vm/extensions/console.ts rename to src/worker/vm/extensions/console.ts diff --git a/src/vm/extensions/google.ts b/src/worker/vm/extensions/google.ts similarity index 100% rename from src/vm/extensions/google.ts rename to src/worker/vm/extensions/google.ts diff --git a/src/vm/extensions/posthog.ts b/src/worker/vm/extensions/posthog.ts similarity index 95% rename from src/vm/extensions/posthog.ts rename to src/worker/vm/extensions/posthog.ts index 203d563d..a39d345b 100644 --- a/src/vm/extensions/posthog.ts +++ b/src/worker/vm/extensions/posthog.ts @@ -2,10 +2,10 @@ import { Properties } from '@posthog/plugin-scaffold' import { DateTime } from 'luxon' import { PluginConfig, PluginsServer, RawEventMessage } from 'types' -import Client from '../../celery/client' -import { UUIDT } from '../../utils' +import Client from '../../../celery/client' +import { UUIDT } from '../../../utils' -const { version } = require('../../../package.json') +const { version } = require('../../../../package.json') interface InternalData { distinct_id: string diff --git a/src/vm/extensions/storage.ts b/src/worker/vm/extensions/storage.ts similarity index 95% rename from src/vm/extensions/storage.ts rename to src/worker/vm/extensions/storage.ts index 80f862ac..a0b2f1a7 100644 --- a/src/vm/extensions/storage.ts +++ b/src/worker/vm/extensions/storage.ts @@ -1,6 +1,6 @@ import { StorageExtension } from '@posthog/plugin-scaffold' -import { PluginConfig, PluginsServer } from '../../types' +import { PluginConfig, PluginsServer } from '../../../types' export function createStorage(server: PluginsServer, pluginConfig: PluginConfig): StorageExtension { const get = async function (key: string, defaultValue: unknown): Promise { diff --git a/src/vm/lazy.ts b/src/worker/vm/lazy.ts similarity index 93% rename from src/vm/lazy.ts rename to src/worker/vm/lazy.ts index a0d81dd5..7d0688a5 100644 --- a/src/vm/lazy.ts +++ b/src/worker/vm/lazy.ts @@ -1,6 +1,6 @@ -import { clearError, processError } from '../error' -import { status } from '../status' -import { PluginConfig, PluginConfigVMReponse, PluginsServer, PluginTask } from '../types' +import { clearError, processError } from '../../error' +import { status } from '../../status' +import { PluginConfig, PluginConfigVMReponse, PluginsServer, PluginTask } from '../../types' import { createPluginConfigVM } from './vm' export class LazyPluginVM { diff --git a/src/vm/transforms/common.ts b/src/worker/vm/transforms/common.ts similarity index 79% rename from src/vm/transforms/common.ts rename to src/worker/vm/transforms/common.ts index 6c60ed2a..32f1e785 100644 --- a/src/vm/transforms/common.ts +++ b/src/worker/vm/transforms/common.ts @@ -1,6 +1,6 @@ import { PluginObj } from '@babel/core' import * as types from '@babel/types' -import { PluginsServer } from '../../types' +import { PluginsServer } from '../../../types' export type PluginGen = (server: PluginsServer) => (param: { types: typeof types }) => PluginObj diff --git a/src/vm/transforms/index.ts b/src/worker/vm/transforms/index.ts similarity index 95% rename from src/vm/transforms/index.ts rename to src/worker/vm/transforms/index.ts index 10c62059..05ebe8c2 100644 --- a/src/vm/transforms/index.ts +++ b/src/worker/vm/transforms/index.ts @@ -1,6 +1,6 @@ import { transform } from '@babel/standalone' -import { PluginsServer } from '../../types' +import { PluginsServer } from '../../../types' import { loopTimeout } from './loop-timeout' import { promiseTimeout } from './promise-timeout' diff --git a/src/vm/transforms/loop-timeout.ts b/src/worker/vm/transforms/loop-timeout.ts similarity index 100% rename from src/vm/transforms/loop-timeout.ts rename to src/worker/vm/transforms/loop-timeout.ts diff --git a/src/vm/transforms/promise-timeout.ts b/src/worker/vm/transforms/promise-timeout.ts similarity index 100% rename from src/vm/transforms/promise-timeout.ts rename to src/worker/vm/transforms/promise-timeout.ts diff --git a/src/vm/vm.ts b/src/worker/vm/vm.ts similarity index 99% rename from src/vm/vm.ts rename to src/worker/vm/vm.ts index 6ff353c3..25100f0e 100644 --- a/src/vm/vm.ts +++ b/src/worker/vm/vm.ts @@ -2,7 +2,7 @@ import { randomBytes } from 'crypto' import fetch from 'node-fetch' import { VM } from 'vm2' -import { PluginConfig, PluginConfigVMReponse, PluginsServer } from '../types' +import { PluginConfig, PluginConfigVMReponse, PluginsServer } from '../../types' import { createCache } from './extensions/cache' import { createConsole } from './extensions/console' import { createGoogle } from './extensions/google' diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index 737c8b77..c649a829 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -3,8 +3,8 @@ import { startPluginsServer } from '../../src/server' import { LogLevel, PluginsServerConfig } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { pluginConfig39 } from '../helpers/plugins' diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 9844ab37..459328d8 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -3,8 +3,8 @@ import { DateTime } from 'luxon' import { startPluginsServer } from '../../src/server' import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team, TimestampFormat } from '../../src/types' import { castTimestampOrNow, UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' import { pluginConfig39 } from '../helpers/plugins' diff --git a/tests/postgres/e2e.test.ts b/tests/postgres/e2e.test.ts index 0dd83233..f8004481 100644 --- a/tests/postgres/e2e.test.ts +++ b/tests/postgres/e2e.test.ts @@ -4,8 +4,8 @@ import { startPluginsServer } from '../../src/server' import { LogLevel } from '../../src/types' import { PluginsServer } from '../../src/types' import { UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' import { delayUntilEventIngested } from '../shared/process-event' diff --git a/tests/postgres/e2e.timeout.test.ts b/tests/postgres/e2e.timeout.test.ts index 23152002..9013e976 100644 --- a/tests/postgres/e2e.timeout.test.ts +++ b/tests/postgres/e2e.timeout.test.ts @@ -1,8 +1,8 @@ import { startPluginsServer } from '../../src/server' import { LogLevel, PluginsServer } from '../../src/types' import { UUIDT } from '../../src/utils' -import { createPosthog, DummyPostHog } from '../../src/vm/extensions/posthog' import { makePiscina } from '../../src/worker/piscina' +import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' import { delayUntilEventIngested } from '../shared/process-event' diff --git a/tests/postgres/vm.lazy.test.ts b/tests/postgres/vm.lazy.test.ts index 00ccfea5..bf57936f 100644 --- a/tests/postgres/vm.lazy.test.ts +++ b/tests/postgres/vm.lazy.test.ts @@ -2,10 +2,10 @@ import { mocked } from 'ts-jest/utils' import { clearError, processError } from '../../src/error' import { status } from '../../src/status' -import { LazyPluginVM } from '../../src/vm/lazy' -import { createPluginConfigVM } from '../../src/vm/vm' +import { LazyPluginVM } from '../../src/worker/vm/lazy' +import { createPluginConfigVM } from '../../src/worker/vm/vm' -jest.mock('../../src/vm/vm') +jest.mock('../../src/worker/vm/vm') jest.mock('../../src/error') jest.mock('../../src/status') diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index 75d78ebb..2ef4bcbd 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -5,7 +5,7 @@ import Client from '../../src/celery/client' import { createServer } from '../../src/server' import { PluginsServer } from '../../src/types' import { delay } from '../../src/utils' -import { createPluginConfigVM } from '../../src/vm/vm' +import { createPluginConfigVM } from '../../src/worker/vm/vm' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' diff --git a/tests/postgres/vm.timeout.test.ts b/tests/postgres/vm.timeout.test.ts index 996e250d..f3d8cd61 100644 --- a/tests/postgres/vm.timeout.test.ts +++ b/tests/postgres/vm.timeout.test.ts @@ -1,6 +1,6 @@ import { createServer } from '../../src/server' import { PluginsServer } from '../../src/types' -import { createPluginConfigVM } from '../../src/vm/vm' +import { createPluginConfigVM } from '../../src/worker/vm/vm' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' diff --git a/tests/transforms.test.ts b/tests/transforms.test.ts index ed4aa3a1..6992728f 100644 --- a/tests/transforms.test.ts +++ b/tests/transforms.test.ts @@ -1,7 +1,7 @@ import { createServer } from '../src/server' import { PluginsServer } from '../src/types' import { code } from '../src/utils' -import { transformCode } from '../src/vm/transforms' +import { transformCode } from '../src/worker/vm/transforms' import { resetTestDatabase } from './helpers/sql' let server: PluginsServer From 308aa7d636c2b0e50f420b7fe408b80ceb10d669 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:08:36 +0200 Subject: [PATCH 08/19] Move celery worker under src/main --- src/{ => main}/celery/worker.ts | 8 ++++---- src/main/queue.ts | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) rename src/{ => main}/celery/worker.ts (97%) diff --git a/src/celery/worker.ts b/src/main/celery/worker.ts similarity index 97% rename from src/celery/worker.ts rename to src/main/celery/worker.ts index 0fa7c1dc..8e69983a 100644 --- a/src/celery/worker.ts +++ b/src/main/celery/worker.ts @@ -1,7 +1,7 @@ -import { status } from '../status' -import { Queue } from '../types' -import Base from './base' -import { Message } from './message' +import Base from '../../celery/base' +import { Message } from '../../celery/message' +import { status } from '../../status' +import { Queue } from '../../types' type Handler = (...args: any[]) => Promise diff --git a/src/main/queue.ts b/src/main/queue.ts index 82b07ca5..de835cf1 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -3,10 +3,10 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import Client from '../celery/client' -import Worker from '../celery/worker' import { status } from '../status' import { IngestEventResponse, PluginsServer, Queue } from '../types' import { UUIDT } from '../utils' +import Worker from './celery/worker' import { KafkaQueue } from './ingestion/kafka-queue' export type WorkerMethods = { From 2bdb24aed234680e8f389a763a23502fb1cdf47b Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:14:46 +0200 Subject: [PATCH 09/19] move process-event under src/worker This one is sort of weird since it gets called in server.ts --- src/server.ts | 3 ++- src/types.ts | 2 +- src/{ => worker}/ingestion/process-event.ts | 16 ++++++++-------- 3 files changed, 11 insertions(+), 10 deletions(-) rename src/{ => worker}/ingestion/process-event.ts (98%) diff --git a/src/server.ts b/src/server.ts index 508bbdf3..41e330dd 100644 --- a/src/server.ts +++ b/src/server.ts @@ -16,13 +16,13 @@ import { ConnectionOptions } from 'tls' import { defaultConfig } from './config' import { DB } from './db' -import { EventsProcessor } from './ingestion/process-event' import { startQueue } from './main/queue' import { startSchedule } from './main/services/schedule' import { startFastifyInstance, stopFastifyInstance } from './main/web/server' import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' import { createPostgresPool, createRedis, delay, UUIDT } from './utils' +import { EventsProcessor } from './worker/ingestion/process-event' const { version } = require('../package.json') @@ -154,6 +154,7 @@ export async function createServer( pluginSchedulePromises: { runEveryMinute: {}, runEveryHour: {}, runEveryDay: {} }, } + // :TODO: This is only used on worker threads, not main server.eventsProcessor = new EventsProcessor(server as PluginsServer) const closeServer = async () => { diff --git a/src/types.ts b/src/types.ts index d2325a9f..c650e544 100644 --- a/src/types.ts +++ b/src/types.ts @@ -2,7 +2,6 @@ import ClickHouse from '@posthog/clickhouse' import { PluginAttachment, PluginConfigSchema, PluginEvent, Properties } from '@posthog/plugin-scaffold' import { Pool as GenericPool } from 'generic-pool' import { StatsD } from 'hot-shots' -import { EventsProcessor } from 'ingestion/process-event' import { Redis } from 'ioredis' import { Kafka, Producer } from 'kafkajs' import { DateTime } from 'luxon' @@ -10,6 +9,7 @@ import { Pool } from 'pg' import { VM } from 'vm2' import { DB } from './db' +import { EventsProcessor } from './worker/ingestion/process-event' import { LazyPluginVM } from './worker/vm/lazy' export enum LogLevel { diff --git a/src/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts similarity index 98% rename from src/ingestion/process-event.ts rename to src/worker/ingestion/process-event.ts index 9c19f3c8..ecef76fb 100644 --- a/src/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -7,10 +7,12 @@ import { DateTime, Duration } from 'luxon' import * as fetch from 'node-fetch' import { nodePostHog } from 'posthog-js-lite/dist/src/targets/node' -import Client from '../celery/client' -import { DB } from '../db' -import { Event as EventProto, IEvent } from '../idl/protos' -import { status } from '../status' +import Client from '../../celery/client' +import { DB } from '../../db' +import { Event as EventProto, IEvent } from '../../idl/protos' +import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from '../../ingestion/topics' +import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../ingestion/utils' +import { status } from '../../status' import { CohortPeople, Element, @@ -21,10 +23,8 @@ import { SessionRecordingEvent, Team, TimestampFormat, -} from '../types' -import { castTimestampOrNow, UUID, UUIDT } from '../utils' -import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from './topics' -import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from './utils' +} from '../../types' +import { castTimestampOrNow, UUID, UUIDT } from '../../utils' export class EventsProcessor { pluginsServer: PluginsServer From f823d122671616af468b5fe7adb36f10940dd31a Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:47:13 +0200 Subject: [PATCH 10/19] Move src/celery under src/shared --- src/main/celery/worker.ts | 4 ++-- src/main/queue.ts | 2 +- src/{ => shared}/celery/base.ts | 2 +- src/{ => shared}/celery/broker.ts | 6 +++--- src/{ => shared}/celery/client.ts | 0 src/{ => shared}/celery/conf.ts | 0 src/{ => shared}/celery/message.ts | 0 src/{ => shared}/celery/task.ts | 0 src/worker/ingestion/process-event.ts | 2 +- src/worker/vm/extensions/posthog.ts | 2 +- tests/postgres/queue.test.ts | 2 +- tests/postgres/vm.test.ts | 4 ++-- tests/postgres/worker.test.ts | 2 +- 13 files changed, 13 insertions(+), 13 deletions(-) rename src/{ => shared}/celery/base.ts (96%) rename src/{ => shared}/celery/broker.ts (98%) rename src/{ => shared}/celery/client.ts (100%) rename src/{ => shared}/celery/conf.ts (100%) rename src/{ => shared}/celery/message.ts (100%) rename src/{ => shared}/celery/task.ts (100%) diff --git a/src/main/celery/worker.ts b/src/main/celery/worker.ts index 8e69983a..7915da06 100644 --- a/src/main/celery/worker.ts +++ b/src/main/celery/worker.ts @@ -1,5 +1,5 @@ -import Base from '../../celery/base' -import { Message } from '../../celery/message' +import Base from '../../shared/celery/base' +import { Message } from '../../shared/celery/message' import { status } from '../../status' import { Queue } from '../../types' diff --git a/src/main/queue.ts b/src/main/queue.ts index de835cf1..ce57e30c 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -2,7 +2,7 @@ import Piscina from '@posthog/piscina' import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' -import Client from '../celery/client' +import Client from '../shared/celery/client' import { status } from '../status' import { IngestEventResponse, PluginsServer, Queue } from '../types' import { UUIDT } from '../utils' diff --git a/src/celery/base.ts b/src/shared/celery/base.ts similarity index 96% rename from src/celery/base.ts rename to src/shared/celery/base.ts index 14df8383..ce6e4ad1 100644 --- a/src/celery/base.ts +++ b/src/shared/celery/base.ts @@ -2,7 +2,7 @@ * writes here Base Parent class of Celery client and worker * @author SunMyeong Lee */ -import { DB } from '../db' +import { DB } from '../../db' import { Broker } from './broker' import { CeleryConf, defaultConf } from './conf' diff --git a/src/celery/broker.ts b/src/shared/celery/broker.ts similarity index 98% rename from src/celery/broker.ts rename to src/shared/celery/broker.ts index 83308fb6..ea39d36b 100644 --- a/src/celery/broker.ts +++ b/src/shared/celery/broker.ts @@ -1,8 +1,8 @@ import { v4 } from 'uuid' -import { DB } from '../db' -import { status } from '../status' -import { Pausable } from '../types' +import { DB } from '../../db' +import { status } from '../../status' +import { Pausable } from '../../types' import { Message } from './message' type BrokerSubscription = { queue: string; callback: (message: Message) => any } diff --git a/src/celery/client.ts b/src/shared/celery/client.ts similarity index 100% rename from src/celery/client.ts rename to src/shared/celery/client.ts diff --git a/src/celery/conf.ts b/src/shared/celery/conf.ts similarity index 100% rename from src/celery/conf.ts rename to src/shared/celery/conf.ts diff --git a/src/celery/message.ts b/src/shared/celery/message.ts similarity index 100% rename from src/celery/message.ts rename to src/shared/celery/message.ts diff --git a/src/celery/task.ts b/src/shared/celery/task.ts similarity index 100% rename from src/celery/task.ts rename to src/shared/celery/task.ts diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index ecef76fb..3a8c37d0 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -7,11 +7,11 @@ import { DateTime, Duration } from 'luxon' import * as fetch from 'node-fetch' import { nodePostHog } from 'posthog-js-lite/dist/src/targets/node' -import Client from '../../celery/client' import { DB } from '../../db' import { Event as EventProto, IEvent } from '../../idl/protos' import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from '../../ingestion/topics' import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../ingestion/utils' +import Client from '../../shared/celery/client' import { status } from '../../status' import { CohortPeople, diff --git a/src/worker/vm/extensions/posthog.ts b/src/worker/vm/extensions/posthog.ts index a39d345b..37280bab 100644 --- a/src/worker/vm/extensions/posthog.ts +++ b/src/worker/vm/extensions/posthog.ts @@ -2,7 +2,7 @@ import { Properties } from '@posthog/plugin-scaffold' import { DateTime } from 'luxon' import { PluginConfig, PluginsServer, RawEventMessage } from 'types' -import Client from '../../../celery/client' +import Client from '../../../shared/celery/client' import { UUIDT } from '../../../utils' const { version } = require('../../../../package.json') diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index 1a2086b8..596de225 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -1,6 +1,6 @@ -import Client from '../../src/celery/client' import { startQueue } from '../../src/main/queue' import { createServer } from '../../src/server' +import Client from '../../src/shared/celery/client' import { LogLevel, PluginsServer } from '../../src/types' import { delay } from '../../src/utils' import { runPlugins } from '../../src/worker/plugins/run' diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index 2ef4bcbd..cc86a0ab 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -1,15 +1,15 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as fetch from 'node-fetch' -import Client from '../../src/celery/client' import { createServer } from '../../src/server' +import Client from '../../src/shared/celery/client' import { PluginsServer } from '../../src/types' import { delay } from '../../src/utils' import { createPluginConfigVM } from '../../src/worker/vm/vm' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' -jest.mock('../../src/celery/client') +jest.mock('../../src/shared/celery/client') const defaultEvent = { distinct_id: 'my_id', diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index d8840a81..f30c910d 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -2,9 +2,9 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import IORedis from 'ioredis' import { mocked } from 'ts-jest/utils' -import Client from '../../src/celery/client' import { loadPluginSchedule } from '../../src/main/services/schedule' import { ServerInstance, startPluginsServer } from '../../src/server' +import Client from '../../src/shared/celery/client' import { LogLevel } from '../../src/types' import { delay, UUIDT } from '../../src/utils' import { ingestEvent } from '../../src/worker/ingestion/ingest-event' From a12cbf50e1d381fb95f2deeff7f99afdea95a862 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:51:21 +0200 Subject: [PATCH 11/19] Move remaining src/ingestion under src/shared --- benchmarks/clickhouse/e2e.kafka.benchmark.ts | 2 +- benchmarks/clickhouse/e2e.timeout.benchmark.ts | 2 +- benchmarks/postgres/ingestion.benchmark.ts | 2 +- src/config.ts | 2 +- src/db.ts | 4 ++-- src/main/ingestion/kafka-queue.ts | 2 +- src/{ => shared}/ingestion/topics.ts | 0 src/{ => shared}/ingestion/utils.ts | 4 ++-- src/worker/ingestion/ingest-event.ts | 2 +- src/worker/ingestion/process-event.ts | 9 +++++++-- tests/clickhouse/e2e.test.ts | 2 +- tests/clickhouse/ingestion-utils.test.ts | 2 +- tests/clickhouse/process-event.test.ts | 2 +- tests/helpers/kafka.ts | 2 +- tests/shared/process-event.ts | 4 ++-- 15 files changed, 23 insertions(+), 18 deletions(-) rename src/{ => shared}/ingestion/topics.ts (100%) rename src/{ => shared}/ingestion/utils.ts (98%) diff --git a/benchmarks/clickhouse/e2e.kafka.benchmark.ts b/benchmarks/clickhouse/e2e.kafka.benchmark.ts index 8dbb1ab7..788c22ca 100644 --- a/benchmarks/clickhouse/e2e.kafka.benchmark.ts +++ b/benchmarks/clickhouse/e2e.kafka.benchmark.ts @@ -1,7 +1,7 @@ import { performance } from 'perf_hooks' -import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' import { startPluginsServer } from '../../src/server' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' diff --git a/benchmarks/clickhouse/e2e.timeout.benchmark.ts b/benchmarks/clickhouse/e2e.timeout.benchmark.ts index 759a1e6c..002698c7 100644 --- a/benchmarks/clickhouse/e2e.timeout.benchmark.ts +++ b/benchmarks/clickhouse/e2e.timeout.benchmark.ts @@ -1,7 +1,7 @@ import { performance } from 'perf_hooks' -import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' import { startPluginsServer } from '../../src/server' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { ClickHouseEvent, LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index 8ead9203..e46584a3 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -4,8 +4,8 @@ import os from 'os' import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' -import { EventsProcessor } from '../../src/ingestion/process-event' import { createServer } from '../../src/server' +import { EventsProcessor } from '../../src/shared/ingestion/process-event' import { LogLevel, PluginsServer, SessionRecordingEvent, Team } from '../../src/types' import { UUIDT } from '../../src/utils' import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' diff --git a/src/config.ts b/src/config.ts index 0b1a0588..795816dd 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,6 +1,6 @@ import os from 'os' -import { KAFKA_EVENTS_PLUGIN_INGESTION } from './ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from './shared/ingestion/topics' import { LogLevel, PluginsServerConfig } from './types' export const defaultConfig = overrideWithEnv(getDefaultConfig()) diff --git a/src/db.ts b/src/db.ts index 34bff1af..6e978ace 100644 --- a/src/db.ts +++ b/src/db.ts @@ -7,8 +7,8 @@ import { Producer, ProducerRecord } from 'kafkajs' import { DateTime } from 'luxon' import { Pool, PoolClient, QueryConfig, QueryResult, QueryResultRow } from 'pg' -import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' -import { chainToElements, hashElements, timeoutGuard, unparsePersonPartial } from './ingestion/utils' +import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './shared/ingestion/topics' +import { chainToElements, hashElements, timeoutGuard, unparsePersonPartial } from './shared/ingestion/utils' import { ClickHouseEvent, ClickHousePerson, diff --git a/src/main/ingestion/kafka-queue.ts b/src/main/ingestion/kafka-queue.ts index faee8a0a..8125b1e8 100644 --- a/src/main/ingestion/kafka-queue.ts +++ b/src/main/ingestion/kafka-queue.ts @@ -3,7 +3,7 @@ import * as Sentry from '@sentry/node' import { Consumer, EachBatchPayload, Kafka } from 'kafkajs' import { PluginsServer, Queue } from 'types' -import { timeoutGuard } from '../../ingestion/utils' +import { timeoutGuard } from '../../shared/ingestion/utils' import { status } from '../../status' import { groupIntoBatches, killGracefully } from '../../utils' diff --git a/src/ingestion/topics.ts b/src/shared/ingestion/topics.ts similarity index 100% rename from src/ingestion/topics.ts rename to src/shared/ingestion/topics.ts diff --git a/src/ingestion/utils.ts b/src/shared/ingestion/utils.ts similarity index 98% rename from src/ingestion/utils.ts rename to src/shared/ingestion/utils.ts index fba7c96d..8ee9b31c 100644 --- a/src/ingestion/utils.ts +++ b/src/shared/ingestion/utils.ts @@ -2,8 +2,8 @@ import { Properties } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import crypto from 'crypto' -import { defaultConfig } from '../config' -import { BasePerson, Element, Person, RawPerson } from '../types' +import { defaultConfig } from '../../config' +import { BasePerson, Element, Person, RawPerson } from '../../types' export function unparsePersonPartial(person: Partial): Partial { return { ...(person as BasePerson), ...(person.created_at ? { created_at: person.created_at.toISO() } : {}) } diff --git a/src/worker/ingestion/ingest-event.ts b/src/worker/ingestion/ingest-event.ts index 1f04ec50..42a79533 100644 --- a/src/worker/ingestion/ingest-event.ts +++ b/src/worker/ingestion/ingest-event.ts @@ -2,7 +2,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import { DateTime } from 'luxon' -import { timeoutGuard } from '../../ingestion/utils' +import { timeoutGuard } from '../../shared/ingestion/utils' import { status } from '../../status' import { IngestEventResponse, PluginsServer } from '../../types' diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index 3a8c37d0..b1c0f989 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -9,9 +9,14 @@ import { nodePostHog } from 'posthog-js-lite/dist/src/targets/node' import { DB } from '../../db' import { Event as EventProto, IEvent } from '../../idl/protos' -import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from '../../ingestion/topics' -import { elementsToString, personInitialAndUTMProperties, sanitizeEventName, timeoutGuard } from '../../ingestion/utils' import Client from '../../shared/celery/client' +import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from '../../shared/ingestion/topics' +import { + elementsToString, + personInitialAndUTMProperties, + sanitizeEventName, + timeoutGuard, +} from '../../shared/ingestion/utils' import { status } from '../../status' import { CohortPeople, diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index c649a829..54ec883d 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -1,5 +1,5 @@ -import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' import { startPluginsServer } from '../../src/server' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { LogLevel, PluginsServerConfig } from '../../src/types' import { PluginsServer } from '../../src/types' import { delay, UUIDT } from '../../src/utils' diff --git a/tests/clickhouse/ingestion-utils.test.ts b/tests/clickhouse/ingestion-utils.test.ts index 4227ab4a..66816d9a 100644 --- a/tests/clickhouse/ingestion-utils.test.ts +++ b/tests/clickhouse/ingestion-utils.test.ts @@ -1,4 +1,4 @@ -import { chainToElements, elementsToString } from '../../src/ingestion/utils' +import { chainToElements, elementsToString } from '../../src/shared/ingestion/utils' test('elementsToString and chainToElements', () => { const elementsString = elementsToString([ diff --git a/tests/clickhouse/process-event.test.ts b/tests/clickhouse/process-event.test.ts index 8d3e5068..e7175ec2 100644 --- a/tests/clickhouse/process-event.test.ts +++ b/tests/clickhouse/process-event.test.ts @@ -1,4 +1,4 @@ -import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/ingestion/topics' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { Event, PluginsServerConfig } from '../../src/types' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' import { resetKafka } from '../helpers/kafka' diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index 5d110e75..de303c94 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -8,7 +8,7 @@ import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID, KAFKA_SESSION_RECORDING_EVENTS, -} from '../../src/ingestion/topics' +} from '../../src/shared/ingestion/topics' import { PluginsServerConfig } from '../../src/types' import { delay, UUIDT } from '../../src/utils' diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 02a74f58..32ed3442 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -4,9 +4,9 @@ import { DateTime } from 'luxon' import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' -import { EventsProcessor } from '../../src/ingestion/process-event' -import { hashElements } from '../../src/ingestion/utils' import { createServer } from '../../src/server' +import { EventsProcessor } from '../../src/shared/ingestion/process-event' +import { hashElements } from '../../src/shared/ingestion/utils' import { Database, Event, From 1f1d73e2a55812cf942af68b38158f4f262c386f Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 11:53:30 +0200 Subject: [PATCH 12/19] move utils under src/shared --- benchmarks/clickhouse/e2e.kafka.benchmark.ts | 2 +- benchmarks/clickhouse/e2e.timeout.benchmark.ts | 2 +- benchmarks/postgres/e2e.celery.benchmark.ts | 2 +- benchmarks/postgres/helpers/piscina.ts | 2 +- benchmarks/postgres/ingestion.benchmark.ts | 2 +- src/db.ts | 16 ++++++++-------- src/init.ts | 2 +- src/main/ingestion/kafka-queue.ts | 2 +- src/main/queue.ts | 2 +- src/main/services/schedule.ts | 2 +- src/server.ts | 2 +- src/{ => shared}/utils.ts | 4 ++-- src/worker/ingestion/process-event.ts | 2 +- src/worker/plugins/loadPlugin.ts | 2 +- src/worker/vm/extensions/posthog.ts | 2 +- src/worker/worker.ts | 2 +- tests/clickhouse/e2e.test.ts | 2 +- tests/clickhouse/postgres-parity.test.ts | 2 +- tests/helpers/kafka.ts | 2 +- tests/helpers/sql.ts | 2 +- tests/postgres/e2e.test.ts | 2 +- tests/postgres/e2e.timeout.test.ts | 2 +- tests/postgres/queue.test.ts | 2 +- tests/postgres/vm.test.ts | 2 +- tests/postgres/worker.test.ts | 2 +- tests/schedule.test.ts | 2 +- tests/shared/process-event.ts | 2 +- tests/transforms.test.ts | 2 +- tests/utils.test.ts | 4 ++-- 29 files changed, 38 insertions(+), 38 deletions(-) rename src/{ => shared}/utils.ts (99%) diff --git a/benchmarks/clickhouse/e2e.kafka.benchmark.ts b/benchmarks/clickhouse/e2e.kafka.benchmark.ts index 788c22ca..c020eeb7 100644 --- a/benchmarks/clickhouse/e2e.kafka.benchmark.ts +++ b/benchmarks/clickhouse/e2e.kafka.benchmark.ts @@ -2,9 +2,9 @@ import { performance } from 'perf_hooks' import { startPluginsServer } from '../../src/server' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' +import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../../tests/helpers/clickhouse' diff --git a/benchmarks/clickhouse/e2e.timeout.benchmark.ts b/benchmarks/clickhouse/e2e.timeout.benchmark.ts index 002698c7..5562e969 100644 --- a/benchmarks/clickhouse/e2e.timeout.benchmark.ts +++ b/benchmarks/clickhouse/e2e.timeout.benchmark.ts @@ -2,9 +2,9 @@ import { performance } from 'perf_hooks' import { startPluginsServer } from '../../src/server' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' +import { delay, UUIDT } from '../../src/shared/utils' import { ClickHouseEvent, LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../../tests/helpers/clickhouse' diff --git a/benchmarks/postgres/e2e.celery.benchmark.ts b/benchmarks/postgres/e2e.celery.benchmark.ts index 63c12405..968b1695 100644 --- a/benchmarks/postgres/e2e.celery.benchmark.ts +++ b/benchmarks/postgres/e2e.celery.benchmark.ts @@ -2,9 +2,9 @@ import * as IORedis from 'ioredis' import { performance } from 'perf_hooks' import { startPluginsServer } from '../../src/server' +import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../../tests/helpers/plugins' diff --git a/benchmarks/postgres/helpers/piscina.ts b/benchmarks/postgres/helpers/piscina.ts index 589b8c87..61e62394 100644 --- a/benchmarks/postgres/helpers/piscina.ts +++ b/benchmarks/postgres/helpers/piscina.ts @@ -2,8 +2,8 @@ import Piscina from '@posthog/piscina' import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { defaultConfig } from '../../../src/config' +import { UUIDT } from '../../../src/shared/utils' import { LogLevel } from '../../../src/types' -import { UUIDT } from '../../../src/utils' import { makePiscina } from '../../../src/worker/piscina' export function setupPiscina(workers: number, tasksPerWorker: number): Piscina { diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index e46584a3..ca58342e 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -6,8 +6,8 @@ import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' import { createServer } from '../../src/server' import { EventsProcessor } from '../../src/shared/ingestion/process-event' +import { UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServer, SessionRecordingEvent, Team } from '../../src/types' -import { UUIDT } from '../../src/utils' import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' import { endLog, startLog } from './helpers/log' import { ingestCountEvents, setupPiscina } from './helpers/piscina' diff --git a/src/db.ts b/src/db.ts index 6e978ace..1dd782d9 100644 --- a/src/db.ts +++ b/src/db.ts @@ -9,6 +9,14 @@ import { Pool, PoolClient, QueryConfig, QueryResult, QueryResultRow } from 'pg' import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './shared/ingestion/topics' import { chainToElements, hashElements, timeoutGuard, unparsePersonPartial } from './shared/ingestion/utils' +import { + castTimestampOrNow, + clickHouseTimestampToISO, + createRedis, + escapeClickHouseString, + sanitizeSqlIdentifier, + tryTwice, +} from './shared/utils' import { ClickHouseEvent, ClickHousePerson, @@ -26,14 +34,6 @@ import { SessionRecordingEvent, TimestampFormat, } from './types' -import { - castTimestampOrNow, - clickHouseTimestampToISO, - createRedis, - escapeClickHouseString, - sanitizeSqlIdentifier, - tryTwice, -} from './utils' /** The recommended way of accessing the database. */ export class DB { diff --git a/src/init.ts b/src/init.ts index d7e1948d..6ff02452 100644 --- a/src/init.ts +++ b/src/init.ts @@ -1,7 +1,7 @@ import * as Sentry from '@sentry/node' +import { setLogLevel } from './shared/utils' import { PluginsServerConfig } from './types' -import { setLogLevel } from './utils' // Must require as `tsc` strips unused `import` statements and just requiring this seems to init some globals require('@sentry/tracing') diff --git a/src/main/ingestion/kafka-queue.ts b/src/main/ingestion/kafka-queue.ts index 8125b1e8..36be61f1 100644 --- a/src/main/ingestion/kafka-queue.ts +++ b/src/main/ingestion/kafka-queue.ts @@ -4,8 +4,8 @@ import { Consumer, EachBatchPayload, Kafka } from 'kafkajs' import { PluginsServer, Queue } from 'types' import { timeoutGuard } from '../../shared/ingestion/utils' +import { groupIntoBatches, killGracefully } from '../../shared/utils' import { status } from '../../status' -import { groupIntoBatches, killGracefully } from '../../utils' export class KafkaQueue implements Queue { private pluginsServer: PluginsServer diff --git a/src/main/queue.ts b/src/main/queue.ts index ce57e30c..b8ea3c1e 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -3,9 +3,9 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import Client from '../shared/celery/client' +import { UUIDT } from '../shared/utils' import { status } from '../status' import { IngestEventResponse, PluginsServer, Queue } from '../types' -import { UUIDT } from '../utils' import Worker from './celery/worker' import { KafkaQueue } from './ingestion/kafka-queue' diff --git a/src/main/services/schedule.ts b/src/main/services/schedule.ts index 55c47d53..37e5faf4 100644 --- a/src/main/services/schedule.ts +++ b/src/main/services/schedule.ts @@ -4,9 +4,9 @@ import * as schedule from 'node-schedule' import Redlock from 'redlock' import { processError } from '../../error' +import { createRedis, delay } from '../../shared/utils' import { status } from '../../status' import { PluginConfigId, PluginsServer, ScheduleControl } from '../../types' -import { createRedis, delay } from '../../utils' export const LOCKED_RESOURCE = 'plugin-server:locks:schedule' diff --git a/src/server.ts b/src/server.ts index 41e330dd..57aa2125 100644 --- a/src/server.ts +++ b/src/server.ts @@ -19,9 +19,9 @@ import { DB } from './db' import { startQueue } from './main/queue' import { startSchedule } from './main/services/schedule' import { startFastifyInstance, stopFastifyInstance } from './main/web/server' +import { createPostgresPool, createRedis, delay, UUIDT } from './shared/utils' import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' -import { createPostgresPool, createRedis, delay, UUIDT } from './utils' import { EventsProcessor } from './worker/ingestion/process-event' const { version } = require('../package.json') diff --git a/src/utils.ts b/src/shared/utils.ts similarity index 99% rename from src/utils.ts rename to src/shared/utils.ts index 2b241d15..fcd54ac3 100644 --- a/src/utils.ts +++ b/src/shared/utils.ts @@ -8,8 +8,8 @@ import { Readable } from 'stream' import * as tar from 'tar-stream' import * as zlib from 'zlib' -import { status } from './status' -import { LogLevel, Plugin, PluginsServerConfig, TimestampFormat } from './types' +import { status } from '../status' +import { LogLevel, Plugin, PluginsServerConfig, TimestampFormat } from '../types' /** Time until autoexit (due to error) gives up on graceful exit and kills the process right away. */ const GRACEFUL_EXIT_PERIOD_SECONDS = 5 diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index b1c0f989..027f4cad 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -17,6 +17,7 @@ import { sanitizeEventName, timeoutGuard, } from '../../shared/ingestion/utils' +import { castTimestampOrNow, UUID, UUIDT } from '../../shared/utils' import { status } from '../../status' import { CohortPeople, @@ -29,7 +30,6 @@ import { Team, TimestampFormat, } from '../../types' -import { castTimestampOrNow, UUID, UUIDT } from '../../utils' export class EventsProcessor { pluginsServer: PluginsServer diff --git a/src/worker/plugins/loadPlugin.ts b/src/worker/plugins/loadPlugin.ts index bdfe6f12..23ad803a 100644 --- a/src/worker/plugins/loadPlugin.ts +++ b/src/worker/plugins/loadPlugin.ts @@ -2,8 +2,8 @@ import * as fs from 'fs' import * as path from 'path' import { processError } from '../../error' +import { getFileFromArchive, pluginDigest } from '../../shared/utils' import { PluginConfig, PluginJsonConfig, PluginsServer } from '../../types' -import { getFileFromArchive, pluginDigest } from '../../utils' export async function loadPlugin(server: PluginsServer, pluginConfig: PluginConfig): Promise { const { plugin } = pluginConfig diff --git a/src/worker/vm/extensions/posthog.ts b/src/worker/vm/extensions/posthog.ts index 37280bab..4c057df0 100644 --- a/src/worker/vm/extensions/posthog.ts +++ b/src/worker/vm/extensions/posthog.ts @@ -3,7 +3,7 @@ import { DateTime } from 'luxon' import { PluginConfig, PluginsServer, RawEventMessage } from 'types' import Client from '../../../shared/celery/client' -import { UUIDT } from '../../../utils' +import { UUIDT } from '../../../shared/utils' const { version } = require('../../../../package.json') diff --git a/src/worker/worker.ts b/src/worker/worker.ts index 8c897262..bae618a5 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,8 +1,8 @@ import { initApp } from '../init' import { createServer } from '../server' +import { cloneObject } from '../shared/utils' import { status } from '../status' import { PluginsServer, PluginsServerConfig } from '../types' -import { cloneObject } from '../utils' import { ingestEvent } from './ingestion/ingest-event' import { runPlugins, runPluginsOnBatch, runPluginTask } from './plugins/run' import { loadSchedule, setupPlugins } from './plugins/setup' diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index 54ec883d..d0f5bd5d 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -1,8 +1,8 @@ import { startPluginsServer } from '../../src/server' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' +import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig } from '../../src/types' import { PluginsServer } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index 459328d8..d5fc3c0c 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -1,8 +1,8 @@ import { DateTime } from 'luxon' import { startPluginsServer } from '../../src/server' +import { castTimestampOrNow, UUIDT } from '../../src/shared/utils' import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team, TimestampFormat } from '../../src/types' -import { castTimestampOrNow, UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { resetTestDatabaseClickhouse } from '../helpers/clickhouse' diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index de303c94..eb222ee2 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -9,8 +9,8 @@ import { KAFKA_PERSON_UNIQUE_ID, KAFKA_SESSION_RECORDING_EVENTS, } from '../../src/shared/ingestion/topics' +import { delay, UUIDT } from '../../src/shared/utils' import { PluginsServerConfig } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' /** Clear the kafka queue */ export async function resetKafka(extraServerConfig: Partial, delayMs = 2000): Promise { diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 140d7d05..712e9b89 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -1,8 +1,8 @@ import { Pool, PoolClient } from 'pg' import { defaultConfig } from '../../src/config' +import { delay, UUIDT } from '../../src/shared/utils' import { PluginsServer, PluginsServerConfig, Team } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { commonOrganizationId, commonOrganizationMembershipId, commonUserId, makePluginObjects } from './plugins' export async function resetTestDatabase( diff --git a/tests/postgres/e2e.test.ts b/tests/postgres/e2e.test.ts index f8004481..e17bf4fe 100644 --- a/tests/postgres/e2e.test.ts +++ b/tests/postgres/e2e.test.ts @@ -1,9 +1,9 @@ import * as IORedis from 'ioredis' import { startPluginsServer } from '../../src/server' +import { UUIDT } from '../../src/shared/utils' import { LogLevel } from '../../src/types' import { PluginsServer } from '../../src/types' -import { UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../helpers/plugins' diff --git a/tests/postgres/e2e.timeout.test.ts b/tests/postgres/e2e.timeout.test.ts index 9013e976..9385240a 100644 --- a/tests/postgres/e2e.timeout.test.ts +++ b/tests/postgres/e2e.timeout.test.ts @@ -1,6 +1,6 @@ import { startPluginsServer } from '../../src/server' +import { UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServer } from '../../src/types' -import { UUIDT } from '../../src/utils' import { makePiscina } from '../../src/worker/piscina' import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog' import { pluginConfig39 } from '../helpers/plugins' diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index 596de225..f804d5c7 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -1,8 +1,8 @@ import { startQueue } from '../../src/main/queue' import { createServer } from '../../src/server' import Client from '../../src/shared/celery/client' +import { delay } from '../../src/shared/utils' import { LogLevel, PluginsServer } from '../../src/types' -import { delay } from '../../src/utils' import { runPlugins } from '../../src/worker/plugins/run' jest.setTimeout(60000) // 60 sec timeout diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index cc86a0ab..d6862e37 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -3,8 +3,8 @@ import * as fetch from 'node-fetch' import { createServer } from '../../src/server' import Client from '../../src/shared/celery/client' +import { delay } from '../../src/shared/utils' import { PluginsServer } from '../../src/types' -import { delay } from '../../src/utils' import { createPluginConfigVM } from '../../src/worker/vm/vm' import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index f30c910d..4aae2204 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -5,8 +5,8 @@ import { mocked } from 'ts-jest/utils' import { loadPluginSchedule } from '../../src/main/services/schedule' import { ServerInstance, startPluginsServer } from '../../src/server' import Client from '../../src/shared/celery/client' +import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { ingestEvent } from '../../src/worker/ingestion/ingest-event' import { makePiscina } from '../../src/worker/piscina' import { runPlugins, runPluginsOnBatch, runPluginTask } from '../../src/worker/plugins/run' diff --git a/tests/schedule.test.ts b/tests/schedule.test.ts index 7c46550c..ae1b23c0 100644 --- a/tests/schedule.test.ts +++ b/tests/schedule.test.ts @@ -8,8 +8,8 @@ import { waitForTasksToFinish, } from '../src/main/services/schedule' import { createServer } from '../src/server' +import { delay } from '../src/shared/utils' import { LogLevel, ScheduleControl } from '../src/types' -import { delay } from '../src/utils' import { createPromise } from './helpers/promises' import { resetTestDatabase } from './helpers/sql' import { setupPiscina } from './helpers/worker' diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 32ed3442..09e389d2 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -7,6 +7,7 @@ import { IEvent } from '../../src/idl/protos' import { createServer } from '../../src/server' import { EventsProcessor } from '../../src/shared/ingestion/process-event' import { hashElements } from '../../src/shared/ingestion/utils' +import { delay, UUIDT } from '../../src/shared/utils' import { Database, Event, @@ -17,7 +18,6 @@ import { SessionRecordingEvent, Team, } from '../../src/types' -import { delay, UUIDT } from '../../src/utils' import { createUserTeamAndOrganization, getFirstTeam, getTeams, onQuery, resetTestDatabase } from '../helpers/sql' jest.setTimeout(600000) // 600 sec timeout. diff --git a/tests/transforms.test.ts b/tests/transforms.test.ts index 6992728f..763a4d1a 100644 --- a/tests/transforms.test.ts +++ b/tests/transforms.test.ts @@ -1,6 +1,6 @@ import { createServer } from '../src/server' +import { code } from '../src/shared/utils' import { PluginsServer } from '../src/types' -import { code } from '../src/utils' import { transformCode } from '../src/worker/vm/transforms' import { resetTestDatabase } from './helpers/sql' diff --git a/tests/utils.test.ts b/tests/utils.test.ts index ac397f48..d9ba4e2e 100644 --- a/tests/utils.test.ts +++ b/tests/utils.test.ts @@ -1,6 +1,5 @@ import { randomBytes } from 'crypto' -import { LogLevel } from '../src/types' import { bufferToStream, cloneObject, @@ -12,7 +11,8 @@ import { setLogLevel, UUID, UUIDT, -} from '../src/utils' +} from '../src/shared/utils' +import { LogLevel } from '../src/types' // .zip in Base64: github repo posthog/helloworldplugin const zip = From ba6216f76b7f80c68e83bff02b5a4b11c7b0c55a Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:01:11 +0200 Subject: [PATCH 13/19] Move some top-level code under shared --- benchmarks/postgres/helpers/piscina.ts | 2 +- benchmarks/postgres/ingestion.benchmark.ts | 2 +- benchmarks/vm/memory.benchmark.ts | 2 +- benchmarks/vm/worker.benchmark.ts | 4 ++-- src/healthcheck.ts | 4 ++-- src/index.ts | 2 +- src/main/celery/worker.ts | 2 +- src/main/ingestion/kafka-queue.ts | 2 +- src/main/queue.ts | 2 +- src/main/services/schedule.ts | 4 ++-- src/main/web/server.ts | 2 +- src/server.ts | 6 +++--- src/shared/celery/base.ts | 2 +- src/shared/celery/broker.ts | 4 ++-- src/{ => shared}/config.ts | 4 ++-- src/{ => shared}/db.ts | 22 +++++++++++----------- src/{ => shared}/error.ts | 2 +- src/shared/ingestion/utils.ts | 2 +- src/{ => shared}/sql.ts | 2 +- src/{ => shared}/status.ts | 0 src/shared/utils.ts | 2 +- src/types.ts | 2 +- src/worker/ingestion/ingest-event.ts | 2 +- src/worker/ingestion/process-event.ts | 4 ++-- src/worker/plugins/loadPlugin.ts | 2 +- src/worker/plugins/run.ts | 2 +- src/worker/plugins/setup.ts | 4 ++-- src/worker/vm/lazy.ts | 4 ++-- src/worker/worker.ts | 2 +- tests/config.test.ts | 2 +- tests/helpers/clickhouse.ts | 2 +- tests/helpers/kafka.ts | 2 +- tests/helpers/sql.ts | 2 +- tests/helpers/sqlMock.ts | 2 +- tests/helpers/worker.ts | 2 +- tests/plugins.test.ts | 8 ++++---- tests/postgres/e2e.test.ts | 2 +- tests/postgres/vm.lazy.test.ts | 8 ++++---- tests/postgres/worker.test.ts | 4 ++-- tests/schedule.test.ts | 4 ++-- tests/server.test.ts | 2 +- tests/sql.test.ts | 2 +- 42 files changed, 69 insertions(+), 69 deletions(-) rename src/{ => shared}/config.ts (97%) rename src/{ => shared}/db.ts (99%) rename src/{ => shared}/error.ts (98%) rename src/{ => shared}/sql.ts (97%) rename src/{ => shared}/status.ts (100%) diff --git a/benchmarks/postgres/helpers/piscina.ts b/benchmarks/postgres/helpers/piscina.ts index 61e62394..6ecc0143 100644 --- a/benchmarks/postgres/helpers/piscina.ts +++ b/benchmarks/postgres/helpers/piscina.ts @@ -1,7 +1,7 @@ import Piscina from '@posthog/piscina' import { PluginEvent } from '@posthog/plugin-scaffold/src/types' -import { defaultConfig } from '../../../src/config' +import { defaultConfig } from '../../../src/shared/config' import { UUIDT } from '../../../src/shared/utils' import { LogLevel } from '../../../src/types' import { makePiscina } from '../../../src/worker/piscina' diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index ca58342e..96a6bceb 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -12,7 +12,7 @@ import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' import { endLog, startLog } from './helpers/log' import { ingestCountEvents, setupPiscina } from './helpers/piscina' -jest.mock('../../src/sql') +jest.mock('../../src/shared/sql') jest.setTimeout(600000) // 600 sec timeout describe('ingestion benchmarks', () => { diff --git a/benchmarks/vm/memory.benchmark.ts b/benchmarks/vm/memory.benchmark.ts index 7a4db84d..66034986 100644 --- a/benchmarks/vm/memory.benchmark.ts +++ b/benchmarks/vm/memory.benchmark.ts @@ -5,7 +5,7 @@ import { Plugin, PluginConfig, PluginConfigVMReponse } from '../../src/types' import { createPluginConfigVM } from '../../src/worker/vm/vm' import { commonOrganizationId } from '../../tests/helpers/plugins' -jest.mock('../../src/sql') +jest.mock('../../src/shared/sql') jest.setTimeout(600000) // 600 sec timeout function createEvent(index: number): PluginEvent { diff --git a/benchmarks/vm/worker.benchmark.ts b/benchmarks/vm/worker.benchmark.ts index 4c75e504..ed824b14 100644 --- a/benchmarks/vm/worker.benchmark.ts +++ b/benchmarks/vm/worker.benchmark.ts @@ -2,12 +2,12 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import * as os from 'os' import { performance } from 'perf_hooks' -import { defaultConfig } from '../../src/config' +import { defaultConfig } from '../../src/shared/config' import { LogLevel } from '../../src/types' import { makePiscina } from '../../src/worker/piscina' import { resetTestDatabase } from '../../tests/helpers/sql' -jest.mock('../../src/sql') +jest.mock('../../src/shared/sql') jest.setTimeout(600000) // 600 sec timeout function processOneEvent( diff --git a/src/healthcheck.ts b/src/healthcheck.ts index 9a684a69..90e96a19 100644 --- a/src/healthcheck.ts +++ b/src/healthcheck.ts @@ -1,7 +1,7 @@ import Redis from 'ioredis' -import { defaultConfig } from './config' -import { Status } from './status' +import { defaultConfig } from './shared/config' +import { Status } from './shared/status' const healthStatus = new Status('HLTH') diff --git a/src/index.ts b/src/index.ts index 71a6ec5e..5981b00f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,8 +1,8 @@ import * as yargs from 'yargs' -import { configHelp, defaultConfig } from './config' import { initApp } from './init' import { startPluginsServer } from './server' +import { configHelp, defaultConfig } from './shared/config' import { PluginsServerConfig } from './types' import { makePiscina } from './worker/piscina' diff --git a/src/main/celery/worker.ts b/src/main/celery/worker.ts index 7915da06..715d80a3 100644 --- a/src/main/celery/worker.ts +++ b/src/main/celery/worker.ts @@ -1,6 +1,6 @@ import Base from '../../shared/celery/base' import { Message } from '../../shared/celery/message' -import { status } from '../../status' +import { status } from '../../shared/status' import { Queue } from '../../types' type Handler = (...args: any[]) => Promise diff --git a/src/main/ingestion/kafka-queue.ts b/src/main/ingestion/kafka-queue.ts index 36be61f1..13b96712 100644 --- a/src/main/ingestion/kafka-queue.ts +++ b/src/main/ingestion/kafka-queue.ts @@ -4,8 +4,8 @@ import { Consumer, EachBatchPayload, Kafka } from 'kafkajs' import { PluginsServer, Queue } from 'types' import { timeoutGuard } from '../../shared/ingestion/utils' +import { status } from '../../shared/status' import { groupIntoBatches, killGracefully } from '../../shared/utils' -import { status } from '../../status' export class KafkaQueue implements Queue { private pluginsServer: PluginsServer diff --git a/src/main/queue.ts b/src/main/queue.ts index b8ea3c1e..5df6a037 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -3,8 +3,8 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import Client from '../shared/celery/client' +import { status } from '../shared/status' import { UUIDT } from '../shared/utils' -import { status } from '../status' import { IngestEventResponse, PluginsServer, Queue } from '../types' import Worker from './celery/worker' import { KafkaQueue } from './ingestion/kafka-queue' diff --git a/src/main/services/schedule.ts b/src/main/services/schedule.ts index 37e5faf4..a1ac9ce3 100644 --- a/src/main/services/schedule.ts +++ b/src/main/services/schedule.ts @@ -3,9 +3,9 @@ import * as Sentry from '@sentry/node' import * as schedule from 'node-schedule' import Redlock from 'redlock' -import { processError } from '../../error' +import { processError } from '../../shared/error' +import { status } from '../../shared/status' import { createRedis, delay } from '../../shared/utils' -import { status } from '../../status' import { PluginConfigId, PluginsServer, ScheduleControl } from '../../types' export const LOCKED_RESOURCE = 'plugin-server:locks:schedule' diff --git a/src/main/web/server.ts b/src/main/web/server.ts index 1bf88653..ffdde008 100644 --- a/src/main/web/server.ts +++ b/src/main/web/server.ts @@ -1,7 +1,7 @@ import { fastify, FastifyInstance } from 'fastify' import { PluginsServer } from 'types' -import { status } from '../../status' +import { status } from '../../shared/status' export function buildFastifyInstance(): FastifyInstance { const fastifyInstance = fastify() diff --git a/src/server.ts b/src/server.ts index 57aa2125..1592850b 100644 --- a/src/server.ts +++ b/src/server.ts @@ -14,13 +14,13 @@ import * as path from 'path' import { Pool, types as pgTypes } from 'pg' import { ConnectionOptions } from 'tls' -import { defaultConfig } from './config' -import { DB } from './db' import { startQueue } from './main/queue' import { startSchedule } from './main/services/schedule' import { startFastifyInstance, stopFastifyInstance } from './main/web/server' +import { defaultConfig } from './shared/config' +import { DB } from './shared/db' +import { status } from './shared/status' import { createPostgresPool, createRedis, delay, UUIDT } from './shared/utils' -import { status } from './status' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' import { EventsProcessor } from './worker/ingestion/process-event' diff --git a/src/shared/celery/base.ts b/src/shared/celery/base.ts index ce6e4ad1..14df8383 100644 --- a/src/shared/celery/base.ts +++ b/src/shared/celery/base.ts @@ -2,7 +2,7 @@ * writes here Base Parent class of Celery client and worker * @author SunMyeong Lee */ -import { DB } from '../../db' +import { DB } from '../db' import { Broker } from './broker' import { CeleryConf, defaultConf } from './conf' diff --git a/src/shared/celery/broker.ts b/src/shared/celery/broker.ts index ea39d36b..1ae60b1c 100644 --- a/src/shared/celery/broker.ts +++ b/src/shared/celery/broker.ts @@ -1,8 +1,8 @@ import { v4 } from 'uuid' -import { DB } from '../../db' -import { status } from '../../status' import { Pausable } from '../../types' +import { DB } from '../db' +import { status } from '../status' import { Message } from './message' type BrokerSubscription = { queue: string; callback: (message: Message) => any } diff --git a/src/config.ts b/src/shared/config.ts similarity index 97% rename from src/config.ts rename to src/shared/config.ts index 795816dd..b3671c73 100644 --- a/src/config.ts +++ b/src/shared/config.ts @@ -1,7 +1,7 @@ import os from 'os' -import { KAFKA_EVENTS_PLUGIN_INGESTION } from './shared/ingestion/topics' -import { LogLevel, PluginsServerConfig } from './types' +import { LogLevel, PluginsServerConfig } from '../types' +import { KAFKA_EVENTS_PLUGIN_INGESTION } from './ingestion/topics' export const defaultConfig = overrideWithEnv(getDefaultConfig()) export const configHelp = getConfigHelp() diff --git a/src/db.ts b/src/shared/db.ts similarity index 99% rename from src/db.ts rename to src/shared/db.ts index 1dd782d9..6c2f17e8 100644 --- a/src/db.ts +++ b/src/shared/db.ts @@ -7,16 +7,6 @@ import { Producer, ProducerRecord } from 'kafkajs' import { DateTime } from 'luxon' import { Pool, PoolClient, QueryConfig, QueryResult, QueryResultRow } from 'pg' -import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './shared/ingestion/topics' -import { chainToElements, hashElements, timeoutGuard, unparsePersonPartial } from './shared/ingestion/utils' -import { - castTimestampOrNow, - clickHouseTimestampToISO, - createRedis, - escapeClickHouseString, - sanitizeSqlIdentifier, - tryTwice, -} from './shared/utils' import { ClickHouseEvent, ClickHousePerson, @@ -33,7 +23,17 @@ import { RawPerson, SessionRecordingEvent, TimestampFormat, -} from './types' +} from '../types' +import { KAFKA_PERSON, KAFKA_PERSON_UNIQUE_ID } from './ingestion/topics' +import { chainToElements, hashElements, timeoutGuard, unparsePersonPartial } from './ingestion/utils' +import { + castTimestampOrNow, + clickHouseTimestampToISO, + createRedis, + escapeClickHouseString, + sanitizeSqlIdentifier, + tryTwice, +} from './utils' /** The recommended way of accessing the database. */ export class DB { diff --git a/src/error.ts b/src/shared/error.ts similarity index 98% rename from src/error.ts rename to src/shared/error.ts index 7a57a763..d8a83ad0 100644 --- a/src/error.ts +++ b/src/shared/error.ts @@ -1,7 +1,7 @@ import { PluginEvent } from '@posthog/plugin-scaffold' +import { PluginConfig, PluginConfigId, PluginError, PluginsServer } from '../types' import { setError } from './sql' -import { PluginConfig, PluginConfigId, PluginError, PluginsServer } from './types' export async function processError( server: PluginsServer, diff --git a/src/shared/ingestion/utils.ts b/src/shared/ingestion/utils.ts index 8ee9b31c..8d0bf47d 100644 --- a/src/shared/ingestion/utils.ts +++ b/src/shared/ingestion/utils.ts @@ -2,8 +2,8 @@ import { Properties } from '@posthog/plugin-scaffold' import * as Sentry from '@sentry/node' import crypto from 'crypto' -import { defaultConfig } from '../../config' import { BasePerson, Element, Person, RawPerson } from '../../types' +import { defaultConfig } from '../config' export function unparsePersonPartial(person: Partial): Partial { return { ...(person as BasePerson), ...(person.created_at ? { created_at: person.created_at.toISO() } : {}) } diff --git a/src/sql.ts b/src/shared/sql.ts similarity index 97% rename from src/sql.ts rename to src/shared/sql.ts index f6246dc4..a4ff0c88 100644 --- a/src/sql.ts +++ b/src/shared/sql.ts @@ -1,4 +1,4 @@ -import { Plugin, PluginAttachmentDB, PluginConfig, PluginConfigId, PluginError, PluginsServer } from './types' +import { Plugin, PluginAttachmentDB, PluginConfig, PluginConfigId, PluginError, PluginsServer } from '../types' function pluginConfigsInForceQuery(specificField?: keyof PluginConfig): string { return `SELECT posthog_pluginconfig.${specificField || '*'} diff --git a/src/status.ts b/src/shared/status.ts similarity index 100% rename from src/status.ts rename to src/shared/status.ts diff --git a/src/shared/utils.ts b/src/shared/utils.ts index fcd54ac3..8aca9455 100644 --- a/src/shared/utils.ts +++ b/src/shared/utils.ts @@ -8,8 +8,8 @@ import { Readable } from 'stream' import * as tar from 'tar-stream' import * as zlib from 'zlib' -import { status } from '../status' import { LogLevel, Plugin, PluginsServerConfig, TimestampFormat } from '../types' +import { status } from './status' /** Time until autoexit (due to error) gives up on graceful exit and kills the process right away. */ const GRACEFUL_EXIT_PERIOD_SECONDS = 5 diff --git a/src/types.ts b/src/types.ts index c650e544..5cf8f929 100644 --- a/src/types.ts +++ b/src/types.ts @@ -8,7 +8,7 @@ import { DateTime } from 'luxon' import { Pool } from 'pg' import { VM } from 'vm2' -import { DB } from './db' +import { DB } from './shared/db' import { EventsProcessor } from './worker/ingestion/process-event' import { LazyPluginVM } from './worker/vm/lazy' diff --git a/src/worker/ingestion/ingest-event.ts b/src/worker/ingestion/ingest-event.ts index 42a79533..6bee1f82 100644 --- a/src/worker/ingestion/ingest-event.ts +++ b/src/worker/ingestion/ingest-event.ts @@ -3,7 +3,7 @@ import * as Sentry from '@sentry/node' import { DateTime } from 'luxon' import { timeoutGuard } from '../../shared/ingestion/utils' -import { status } from '../../status' +import { status } from '../../shared/status' import { IngestEventResponse, PluginsServer } from '../../types' export async function ingestEvent(server: PluginsServer, event: PluginEvent): Promise { diff --git a/src/worker/ingestion/process-event.ts b/src/worker/ingestion/process-event.ts index 027f4cad..7a07503e 100644 --- a/src/worker/ingestion/process-event.ts +++ b/src/worker/ingestion/process-event.ts @@ -7,9 +7,9 @@ import { DateTime, Duration } from 'luxon' import * as fetch from 'node-fetch' import { nodePostHog } from 'posthog-js-lite/dist/src/targets/node' -import { DB } from '../../db' import { Event as EventProto, IEvent } from '../../idl/protos' import Client from '../../shared/celery/client' +import { DB } from '../../shared/db' import { KAFKA_EVENTS, KAFKA_SESSION_RECORDING_EVENTS } from '../../shared/ingestion/topics' import { elementsToString, @@ -17,8 +17,8 @@ import { sanitizeEventName, timeoutGuard, } from '../../shared/ingestion/utils' +import { status } from '../../shared/status' import { castTimestampOrNow, UUID, UUIDT } from '../../shared/utils' -import { status } from '../../status' import { CohortPeople, Element, diff --git a/src/worker/plugins/loadPlugin.ts b/src/worker/plugins/loadPlugin.ts index 23ad803a..7cbe9ac8 100644 --- a/src/worker/plugins/loadPlugin.ts +++ b/src/worker/plugins/loadPlugin.ts @@ -1,7 +1,7 @@ import * as fs from 'fs' import * as path from 'path' -import { processError } from '../../error' +import { processError } from '../../shared/error' import { getFileFromArchive, pluginDigest } from '../../shared/utils' import { PluginConfig, PluginJsonConfig, PluginsServer } from '../../types' diff --git a/src/worker/plugins/run.ts b/src/worker/plugins/run.ts index 3f0d213f..e4b6612f 100644 --- a/src/worker/plugins/run.ts +++ b/src/worker/plugins/run.ts @@ -1,6 +1,6 @@ import { PluginEvent } from '@posthog/plugin-scaffold' -import { processError } from '../../error' +import { processError } from '../../shared/error' import { PluginConfig, PluginsServer } from '../../types' export async function runPlugins(server: PluginsServer, event: PluginEvent): Promise { diff --git a/src/worker/plugins/setup.ts b/src/worker/plugins/setup.ts index 80320955..8e95cd85 100644 --- a/src/worker/plugins/setup.ts +++ b/src/worker/plugins/setup.ts @@ -1,7 +1,7 @@ import { PluginAttachment } from '@posthog/plugin-scaffold' -import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows } from '../../sql' -import { status } from '../../status' +import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows } from '../../shared/sql' +import { status } from '../../shared/status' import { Plugin, PluginConfig, PluginConfigId, PluginId, PluginsServer, TeamId } from '../../types' import { LazyPluginVM } from '../vm/lazy' import { loadPlugin } from './loadPlugin' diff --git a/src/worker/vm/lazy.ts b/src/worker/vm/lazy.ts index 7d0688a5..0a77f548 100644 --- a/src/worker/vm/lazy.ts +++ b/src/worker/vm/lazy.ts @@ -1,5 +1,5 @@ -import { clearError, processError } from '../../error' -import { status } from '../../status' +import { clearError, processError } from '../../shared/error' +import { status } from '../../shared/status' import { PluginConfig, PluginConfigVMReponse, PluginsServer, PluginTask } from '../../types' import { createPluginConfigVM } from './vm' diff --git a/src/worker/worker.ts b/src/worker/worker.ts index bae618a5..1e0101e0 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,7 +1,7 @@ import { initApp } from '../init' import { createServer } from '../server' +import { status } from '../shared/status' import { cloneObject } from '../shared/utils' -import { status } from '../status' import { PluginsServer, PluginsServerConfig } from '../types' import { ingestEvent } from './ingestion/ingest-event' import { runPlugins, runPluginsOnBatch, runPluginTask } from './plugins/run' diff --git a/tests/config.test.ts b/tests/config.test.ts index 19ab55e5..63749104 100644 --- a/tests/config.test.ts +++ b/tests/config.test.ts @@ -1,4 +1,4 @@ -import { getDefaultConfig, overrideWithEnv } from '../src/config' +import { getDefaultConfig, overrideWithEnv } from '../src/shared/config' test('overrideWithEnv 1', () => { const defaultConfig = getDefaultConfig() diff --git a/tests/helpers/clickhouse.ts b/tests/helpers/clickhouse.ts index 33af10a8..35c4a5bc 100644 --- a/tests/helpers/clickhouse.ts +++ b/tests/helpers/clickhouse.ts @@ -1,6 +1,6 @@ import ClickHouse from '@posthog/clickhouse' -import { defaultConfig } from '../../src/config' +import { defaultConfig } from '../../src/shared/config' import { PluginsServerConfig } from '../../src/types' export async function resetTestDatabaseClickhouse(extraServerConfig: Partial): Promise { diff --git a/tests/helpers/kafka.ts b/tests/helpers/kafka.ts index eb222ee2..20b60936 100644 --- a/tests/helpers/kafka.ts +++ b/tests/helpers/kafka.ts @@ -1,6 +1,6 @@ import { Kafka, logLevel } from 'kafkajs' -import { defaultConfig, overrideWithEnv } from '../../src/config' +import { defaultConfig, overrideWithEnv } from '../../src/shared/config' import { KAFKA_EVENTS, KAFKA_EVENTS_PLUGIN_INGESTION, diff --git a/tests/helpers/sql.ts b/tests/helpers/sql.ts index 712e9b89..64a330d9 100644 --- a/tests/helpers/sql.ts +++ b/tests/helpers/sql.ts @@ -1,6 +1,6 @@ import { Pool, PoolClient } from 'pg' -import { defaultConfig } from '../../src/config' +import { defaultConfig } from '../../src/shared/config' import { delay, UUIDT } from '../../src/shared/utils' import { PluginsServer, PluginsServerConfig, Team } from '../../src/types' import { commonOrganizationId, commonOrganizationMembershipId, commonUserId, makePluginObjects } from './plugins' diff --git a/tests/helpers/sqlMock.ts b/tests/helpers/sqlMock.ts index 42140942..35392325 100644 --- a/tests/helpers/sqlMock.ts +++ b/tests/helpers/sqlMock.ts @@ -1,4 +1,4 @@ -import * as s from '../../src/sql' +import * as s from '../../src/shared/sql' // mock functions that get data from postgres and give them the right types type UnPromisify = F extends (...args: infer A) => Promise ? (...args: A) => T : never diff --git a/tests/helpers/worker.ts b/tests/helpers/worker.ts index a12de0c2..b9205df6 100644 --- a/tests/helpers/worker.ts +++ b/tests/helpers/worker.ts @@ -1,6 +1,6 @@ import Piscina from '@posthog/piscina' -import { defaultConfig } from '../../src/config' +import { defaultConfig } from '../../src/shared/config' import { LogLevel } from '../../src/types' import { makePiscina } from '../../src/worker/piscina' diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index d5f3e38c..187fcdce 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -1,8 +1,8 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { mocked } from 'ts-jest/utils' -import { clearError, processError } from '../src/error' import { createServer } from '../src/server' +import { clearError, processError } from '../src/shared/error' import { LogLevel, PluginsServer } from '../src/types' import { loadPlugin } from '../src/worker/plugins/loadPlugin' import { runPlugins } from '../src/worker/plugins/run' @@ -17,9 +17,9 @@ import { } from './helpers/plugins' import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows, setError } from './helpers/sqlMock' -jest.mock('../src/sql') -jest.mock('../src/status') -jest.mock('../src/error') +jest.mock('../src/shared/sql') +jest.mock('../src/shared/status') +jest.mock('../src/shared/error') jest.mock('../src/worker/plugins/loadPlugin', () => { const { loadPlugin } = jest.requireActual('../src/worker/plugins/loadPlugin') return { loadPlugin: jest.fn().mockImplementation(loadPlugin) } diff --git a/tests/postgres/e2e.test.ts b/tests/postgres/e2e.test.ts index e17bf4fe..d4ac3662 100644 --- a/tests/postgres/e2e.test.ts +++ b/tests/postgres/e2e.test.ts @@ -10,7 +10,7 @@ import { pluginConfig39 } from '../helpers/plugins' import { resetTestDatabase } from '../helpers/sql' import { delayUntilEventIngested } from '../shared/process-event' -jest.mock('../../src/status') +jest.mock('../../src/shared/status') jest.setTimeout(60000) // 60 sec timeout describe('e2e postgres ingestion', () => { diff --git a/tests/postgres/vm.lazy.test.ts b/tests/postgres/vm.lazy.test.ts index bf57936f..5435652a 100644 --- a/tests/postgres/vm.lazy.test.ts +++ b/tests/postgres/vm.lazy.test.ts @@ -1,13 +1,13 @@ import { mocked } from 'ts-jest/utils' -import { clearError, processError } from '../../src/error' -import { status } from '../../src/status' +import { clearError, processError } from '../../src/shared/error' +import { status } from '../../src/shared/status' import { LazyPluginVM } from '../../src/worker/vm/lazy' import { createPluginConfigVM } from '../../src/worker/vm/vm' jest.mock('../../src/worker/vm/vm') -jest.mock('../../src/error') -jest.mock('../../src/status') +jest.mock('../../src/shared/error') +jest.mock('../../src/shared/status') describe('LazyPluginVM', () => { const createVM = () => new LazyPluginVM() diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 4aae2204..62f09073 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -15,8 +15,8 @@ import { createTaskRunner } from '../../src/worker/worker' import { resetTestDatabase } from '../helpers/sql' import { setupPiscina } from '../helpers/worker' -jest.mock('../../src/sql') -jest.mock('../../src/status') +jest.mock('../../src/shared/sql') +jest.mock('../../src/shared/status') jest.mock('../../src/worker/ingestion/ingest-event') jest.mock('../../src/worker/plugins/run') jest.mock('../../src/worker/plugins/setup') diff --git a/tests/schedule.test.ts b/tests/schedule.test.ts index ae1b23c0..287edd3b 100644 --- a/tests/schedule.test.ts +++ b/tests/schedule.test.ts @@ -14,8 +14,8 @@ import { createPromise } from './helpers/promises' import { resetTestDatabase } from './helpers/sql' import { setupPiscina } from './helpers/worker' -jest.mock('../src/sql') -jest.mock('../src/status') +jest.mock('../src/shared/sql') +jest.mock('../src/shared/status') jest.setTimeout(60000) // 60 sec timeout function createEvent(index = 0): PluginEvent { diff --git a/tests/server.test.ts b/tests/server.test.ts index b29b1b5e..cd2f5f2f 100644 --- a/tests/server.test.ts +++ b/tests/server.test.ts @@ -5,7 +5,7 @@ import { LogLevel } from '../src/types' import { makePiscina } from '../src/worker/piscina' import { resetTestDatabase } from './helpers/sql' -jest.mock('../src/sql') +jest.mock('../src/shared/sql') jest.setTimeout(60000) // 60 sec timeout test('startPluginsServer', async () => { diff --git a/tests/sql.test.ts b/tests/sql.test.ts index 171da780..a0a82d4a 100644 --- a/tests/sql.test.ts +++ b/tests/sql.test.ts @@ -1,5 +1,5 @@ import { createServer } from '../src/server' -import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows, setError } from '../src/sql' +import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows, setError } from '../src/shared/sql' import { PluginConfig, PluginError, PluginsServer } from '../src/types' import { commonOrganizationId } from './helpers/plugins' import { resetTestDatabase } from './helpers/sql' From b3665dab3430c3224d7eaf623888f10015869399 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:07:13 +0200 Subject: [PATCH 14/19] Move main pluginsserver code under src/main --- benchmarks/clickhouse/e2e.kafka.benchmark.ts | 2 +- .../clickhouse/e2e.timeout.benchmark.ts | 2 +- benchmarks/postgres/e2e.celery.benchmark.ts | 2 +- src/index.ts | 2 +- src/main/pluginsServer.ts | 149 ++++++++++++++++++ src/server.ts | 147 +---------------- tests/clickhouse/e2e.test.ts | 2 +- tests/clickhouse/postgres-parity.test.ts | 2 +- tests/postgres/e2e.test.ts | 2 +- tests/postgres/e2e.timeout.test.ts | 2 +- tests/postgres/worker.test.ts | 2 +- tests/server.test.ts | 4 +- 12 files changed, 162 insertions(+), 156 deletions(-) create mode 100644 src/main/pluginsServer.ts diff --git a/benchmarks/clickhouse/e2e.kafka.benchmark.ts b/benchmarks/clickhouse/e2e.kafka.benchmark.ts index c020eeb7..07e3b1db 100644 --- a/benchmarks/clickhouse/e2e.kafka.benchmark.ts +++ b/benchmarks/clickhouse/e2e.kafka.benchmark.ts @@ -1,6 +1,6 @@ import { performance } from 'perf_hooks' -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' diff --git a/benchmarks/clickhouse/e2e.timeout.benchmark.ts b/benchmarks/clickhouse/e2e.timeout.benchmark.ts index 5562e969..f8c43ec7 100644 --- a/benchmarks/clickhouse/e2e.timeout.benchmark.ts +++ b/benchmarks/clickhouse/e2e.timeout.benchmark.ts @@ -1,6 +1,6 @@ import { performance } from 'perf_hooks' -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { delay, UUIDT } from '../../src/shared/utils' import { ClickHouseEvent, LogLevel, PluginsServerConfig, Queue } from '../../src/types' diff --git a/benchmarks/postgres/e2e.celery.benchmark.ts b/benchmarks/postgres/e2e.celery.benchmark.ts index 968b1695..cf11be2a 100644 --- a/benchmarks/postgres/e2e.celery.benchmark.ts +++ b/benchmarks/postgres/e2e.celery.benchmark.ts @@ -1,7 +1,7 @@ import * as IORedis from 'ioredis' import { performance } from 'perf_hooks' -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig, Queue } from '../../src/types' import { PluginsServer } from '../../src/types' diff --git a/src/index.ts b/src/index.ts index 5981b00f..0cb9164f 100644 --- a/src/index.ts +++ b/src/index.ts @@ -1,7 +1,7 @@ import * as yargs from 'yargs' import { initApp } from './init' -import { startPluginsServer } from './server' +import { startPluginsServer } from './main/pluginsServer' import { configHelp, defaultConfig } from './shared/config' import { PluginsServerConfig } from './types' import { makePiscina } from './worker/piscina' diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts new file mode 100644 index 00000000..94bbfb23 --- /dev/null +++ b/src/main/pluginsServer.ts @@ -0,0 +1,149 @@ +import Piscina from '@posthog/piscina' +import * as Sentry from '@sentry/node' +import { FastifyInstance } from 'fastify' +import Redis from 'ioredis' +import * as schedule from 'node-schedule' + +import { createServer } from '../server' +import { defaultConfig } from '../shared/config' +import { status } from '../shared/status' +import { createRedis, delay } from '../shared/utils' +import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from '../types' +import { startQueue } from './queue' +import { startSchedule } from './services/schedule' +import { startFastifyInstance, stopFastifyInstance } from './web/server' + +const { version } = require('../package.json') + +// TODO: refactor this into a class, removing the need for many different Servers +export type ServerInstance = { + server: PluginsServer + piscina: Piscina + queue: Queue + stop: () => Promise +} + +export async function startPluginsServer( + config: Partial, + makePiscina: (config: PluginsServerConfig) => Piscina +): Promise { + const serverConfig: PluginsServerConfig = { + ...defaultConfig, + ...config, + } + + status.info('⚡', `posthog-plugin-server v${version}`) + status.info('ℹī¸', `${serverConfig.WORKER_CONCURRENCY} workers, ${serverConfig.TASKS_PER_WORKER} tasks per worker`) + + let pubSub: Redis.Redis | undefined + let server: PluginsServer | undefined + let fastifyInstance: FastifyInstance | undefined + let pingJob: schedule.Job | undefined + let statsJob: schedule.Job | undefined + let piscina: Piscina | undefined + let queue: Queue | undefined + let closeServer: () => Promise | undefined + let scheduleControl: ScheduleControl | undefined + + let shutdownStatus = 0 + + async function closeJobs(): Promise { + shutdownStatus += 1 + if (shutdownStatus === 2) { + status.info('🔁', 'Try again to shut down forcibly') + return + } + if (shutdownStatus >= 3) { + status.info('❗ī¸', 'Shutting down forcibly!') + void piscina?.destroy() + process.exit() + } + status.info('💤', ' Shutting down gracefully...') + if (fastifyInstance && !serverConfig?.DISABLE_WEB) { + await stopFastifyInstance(fastifyInstance!) + } + await queue?.stop() + await pubSub?.quit() + pingJob && schedule.cancelJob(pingJob) + statsJob && schedule.cancelJob(statsJob) + await scheduleControl?.stopSchedule() + if (piscina) { + await stopPiscina(piscina) + } + await closeServer?.() + + // wait an extra second for any misc async task to finish + await delay(1000) + } + + for (const signal of ['SIGINT', 'SIGTERM', 'SIGHUP']) { + process.on(signal, closeJobs) + } + + try { + ;[server, closeServer] = await createServer(serverConfig, null) + + piscina = makePiscina(serverConfig) + if (!server.DISABLE_WEB) { + fastifyInstance = await startFastifyInstance(server) + } + + scheduleControl = await startSchedule(server, piscina) + queue = await startQueue(server, piscina) + piscina.on('drain', () => { + queue?.resume() + }) + + // use one extra connection for redis pubsub + pubSub = await createRedis(server) + await pubSub.subscribe(server.PLUGINS_RELOAD_PUBSUB_CHANNEL) + pubSub.on('message', async (channel: string, message) => { + if (channel === server!.PLUGINS_RELOAD_PUBSUB_CHANNEL) { + status.info('⚡', 'Reloading plugins!') + + await piscina?.broadcastTask({ task: 'reloadPlugins' }) + await scheduleControl?.reloadSchedule() + } + }) + + // every 5 seconds set Redis keys @posthog-plugin-server/ping and @posthog-plugin-server/version + pingJob = schedule.scheduleJob('*/5 * * * * *', async () => { + await server!.db!.redisSet('@posthog-plugin-server/ping', new Date().toISOString(), 60, { + jsonSerialize: false, + }) + await server!.db!.redisSet('@posthog-plugin-server/version', version, undefined, { jsonSerialize: false }) + }) + + // every 10 seconds sends stuff to StatsD + statsJob = schedule.scheduleJob('*/10 * * * * *', () => { + if (piscina) { + server!.statsd?.gauge(`piscina.utilization`, (piscina?.utilization || 0) * 100) + server!.statsd?.gauge(`piscina.threads`, piscina?.threads.length) + server!.statsd?.gauge(`piscina.queue_size`, piscina?.queueSize) + } + }) + + status.info('🚀', 'All systems go.') + } catch (error) { + Sentry.captureException(error) + status.error('đŸ’Ĩ', 'Launchpad failure!', error) + void Sentry.flush() // flush in the background + await closeJobs() + process.exit(1) + } + + return { + server, + piscina, + queue, + stop: closeJobs, + } +} + +export async function stopPiscina(piscina: Piscina): Promise { + // Wait two seconds for any running workers to stop. + // TODO: better "wait until everything is done" + await delay(2000) + await Promise.race([piscina.broadcastTask({ task: 'flushKafkaMessages' }), delay(2000)]) + await piscina.destroy() +} diff --git a/src/server.ts b/src/server.ts index 1592850b..ebd63355 100644 --- a/src/server.ts +++ b/src/server.ts @@ -1,27 +1,19 @@ import ClickHouse from '@posthog/clickhouse' -import Piscina from '@posthog/piscina' -import { PluginEvent } from '@posthog/plugin-scaffold' -import * as Sentry from '@sentry/node' -import { FastifyInstance } from 'fastify' import * as fs from 'fs' import { createPool } from 'generic-pool' import { StatsD } from 'hot-shots' import Redis from 'ioredis' import { Kafka, logLevel, Producer } from 'kafkajs' import { DateTime } from 'luxon' -import * as schedule from 'node-schedule' import * as path from 'path' -import { Pool, types as pgTypes } from 'pg' +import { types as pgTypes } from 'pg' import { ConnectionOptions } from 'tls' -import { startQueue } from './main/queue' -import { startSchedule } from './main/services/schedule' -import { startFastifyInstance, stopFastifyInstance } from './main/web/server' import { defaultConfig } from './shared/config' import { DB } from './shared/db' import { status } from './shared/status' -import { createPostgresPool, createRedis, delay, UUIDT } from './shared/utils' -import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from './types' +import { createPostgresPool, createRedis, UUIDT } from './shared/utils' +import { PluginsServer, PluginsServerConfig } from './types' import { EventsProcessor } from './worker/ingestion/process-event' const { version } = require('../package.json') @@ -168,136 +160,3 @@ export async function createServer( return [server as PluginsServer, closeServer] } - -// TODO: refactor this into a class, removing the need for many different Servers -export type ServerInstance = { - server: PluginsServer - piscina: Piscina - queue: Queue - stop: () => Promise -} - -export async function startPluginsServer( - config: Partial, - makePiscina: (config: PluginsServerConfig) => Piscina -): Promise { - const serverConfig: PluginsServerConfig = { - ...defaultConfig, - ...config, - } - - status.info('⚡', `posthog-plugin-server v${version}`) - status.info('ℹī¸', `${serverConfig.WORKER_CONCURRENCY} workers, ${serverConfig.TASKS_PER_WORKER} tasks per worker`) - - let pubSub: Redis.Redis | undefined - let server: PluginsServer | undefined - let fastifyInstance: FastifyInstance | undefined - let pingJob: schedule.Job | undefined - let statsJob: schedule.Job | undefined - let piscina: Piscina | undefined - let queue: Queue | undefined - let closeServer: () => Promise | undefined - let scheduleControl: ScheduleControl | undefined - - let shutdownStatus = 0 - - async function closeJobs(): Promise { - shutdownStatus += 1 - if (shutdownStatus === 2) { - status.info('🔁', 'Try again to shut down forcibly') - return - } - if (shutdownStatus >= 3) { - status.info('❗ī¸', 'Shutting down forcibly!') - void piscina?.destroy() - process.exit() - } - status.info('💤', ' Shutting down gracefully...') - if (fastifyInstance && !serverConfig?.DISABLE_WEB) { - await stopFastifyInstance(fastifyInstance!) - } - await queue?.stop() - await pubSub?.quit() - pingJob && schedule.cancelJob(pingJob) - statsJob && schedule.cancelJob(statsJob) - await scheduleControl?.stopSchedule() - if (piscina) { - await stopPiscina(piscina) - } - await closeServer?.() - - // wait an extra second for any misc async task to finish - await delay(1000) - } - - for (const signal of ['SIGINT', 'SIGTERM', 'SIGHUP']) { - process.on(signal, closeJobs) - } - - try { - ;[server, closeServer] = await createServer(serverConfig, null) - - piscina = makePiscina(serverConfig) - if (!server.DISABLE_WEB) { - fastifyInstance = await startFastifyInstance(server) - } - - scheduleControl = await startSchedule(server, piscina) - queue = await startQueue(server, piscina) - piscina.on('drain', () => { - queue?.resume() - }) - - // use one extra connection for redis pubsub - pubSub = await createRedis(server) - await pubSub.subscribe(server.PLUGINS_RELOAD_PUBSUB_CHANNEL) - pubSub.on('message', async (channel: string, message) => { - if (channel === server!.PLUGINS_RELOAD_PUBSUB_CHANNEL) { - status.info('⚡', 'Reloading plugins!') - - await piscina?.broadcastTask({ task: 'reloadPlugins' }) - await scheduleControl?.reloadSchedule() - } - }) - - // every 5 seconds set Redis keys @posthog-plugin-server/ping and @posthog-plugin-server/version - pingJob = schedule.scheduleJob('*/5 * * * * *', async () => { - await server!.db!.redisSet('@posthog-plugin-server/ping', new Date().toISOString(), 60, { - jsonSerialize: false, - }) - await server!.db!.redisSet('@posthog-plugin-server/version', version, undefined, { jsonSerialize: false }) - }) - - // every 10 seconds sends stuff to StatsD - statsJob = schedule.scheduleJob('*/10 * * * * *', () => { - if (piscina) { - server!.statsd?.gauge(`piscina.utilization`, (piscina?.utilization || 0) * 100) - server!.statsd?.gauge(`piscina.threads`, piscina?.threads.length) - server!.statsd?.gauge(`piscina.queue_size`, piscina?.queueSize) - } - }) - - status.info('🚀', 'All systems go.') - } catch (error) { - Sentry.captureException(error) - status.error('đŸ’Ĩ', 'Launchpad failure!', error) - void Sentry.flush() // flush in the background - await closeJobs() - process.exit(1) - } - - return { - server, - piscina, - queue, - stop: closeJobs, - } -} - -export async function stopPiscina(piscina: Piscina): Promise { - // Wait two seconds for any running workers to stop. - // TODO: better "wait until everything is done" - await delay(2000) - await Promise.race([piscina.broadcastTask({ task: 'flushKafkaMessages' }), delay(2000)]) - await piscina.destroy() -} diff --git a/tests/clickhouse/e2e.test.ts b/tests/clickhouse/e2e.test.ts index d0f5bd5d..318718e7 100644 --- a/tests/clickhouse/e2e.test.ts +++ b/tests/clickhouse/e2e.test.ts @@ -1,4 +1,4 @@ -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/shared/ingestion/topics' import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServerConfig } from '../../src/types' diff --git a/tests/clickhouse/postgres-parity.test.ts b/tests/clickhouse/postgres-parity.test.ts index d5fc3c0c..3e8ee533 100644 --- a/tests/clickhouse/postgres-parity.test.ts +++ b/tests/clickhouse/postgres-parity.test.ts @@ -1,6 +1,6 @@ import { DateTime } from 'luxon' -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { castTimestampOrNow, UUIDT } from '../../src/shared/utils' import { Database, LogLevel, PluginsServer, PluginsServerConfig, Team, TimestampFormat } from '../../src/types' import { makePiscina } from '../../src/worker/piscina' diff --git a/tests/postgres/e2e.test.ts b/tests/postgres/e2e.test.ts index d4ac3662..f7c989c8 100644 --- a/tests/postgres/e2e.test.ts +++ b/tests/postgres/e2e.test.ts @@ -1,6 +1,6 @@ import * as IORedis from 'ioredis' -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { UUIDT } from '../../src/shared/utils' import { LogLevel } from '../../src/types' import { PluginsServer } from '../../src/types' diff --git a/tests/postgres/e2e.timeout.test.ts b/tests/postgres/e2e.timeout.test.ts index 9385240a..30923a0f 100644 --- a/tests/postgres/e2e.timeout.test.ts +++ b/tests/postgres/e2e.timeout.test.ts @@ -1,4 +1,4 @@ -import { startPluginsServer } from '../../src/server' +import { startPluginsServer } from '../../src/main/pluginsServer' import { UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServer } from '../../src/types' import { makePiscina } from '../../src/worker/piscina' diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 62f09073..5a030d60 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -2,8 +2,8 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import IORedis from 'ioredis' import { mocked } from 'ts-jest/utils' +import { ServerInstance, startPluginsServer } from '../../src/main/pluginsServer' import { loadPluginSchedule } from '../../src/main/services/schedule' -import { ServerInstance, startPluginsServer } from '../../src/server' import Client from '../../src/shared/celery/client' import { delay, UUIDT } from '../../src/shared/utils' import { LogLevel } from '../../src/types' diff --git a/tests/server.test.ts b/tests/server.test.ts index cd2f5f2f..f6616beb 100644 --- a/tests/server.test.ts +++ b/tests/server.test.ts @@ -1,6 +1,4 @@ -import { PluginEvent } from '@posthog/plugin-scaffold/src/types' - -import { startPluginsServer } from '../src/server' +import { startPluginsServer } from '../src/main/pluginsServer' import { LogLevel } from '../src/types' import { makePiscina } from '../src/worker/piscina' import { resetTestDatabase } from './helpers/sql' From a58ab369767b23e555494c50b806d930bcab3de1 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:08:17 +0200 Subject: [PATCH 15/19] Fix typeerror in tests --- tests/postgres/worker.test.ts | 2 +- tests/shared/process-event.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/postgres/worker.test.ts b/tests/postgres/worker.test.ts index 5a030d60..3cce72f1 100644 --- a/tests/postgres/worker.test.ts +++ b/tests/postgres/worker.test.ts @@ -92,7 +92,7 @@ test('assume that the workerThreads and tasksPerWorker values behave as expected await resetTestDatabase(testCode) const piscina = setupPiscina(workerThreads, tasksPerWorker) const processEvent = (event: PluginEvent) => piscina.runTask({ task: 'processEvent', args: { event } }) - const promises = [] + const promises: Array> = [] // warmup 2x await Promise.all([processEvent(createEvent()), processEvent(createEvent())]) diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 09e389d2..7d1aa69b 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -5,7 +5,6 @@ import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' import { createServer } from '../../src/server' -import { EventsProcessor } from '../../src/shared/ingestion/process-event' import { hashElements } from '../../src/shared/ingestion/utils' import { delay, UUIDT } from '../../src/shared/utils' import { @@ -18,6 +17,7 @@ import { SessionRecordingEvent, Team, } from '../../src/types' +import { EventsProcessor } from '../../src/worker/ingestion/process-event' import { createUserTeamAndOrganization, getFirstTeam, getTeams, onQuery, resetTestDatabase } from '../helpers/sql' jest.setTimeout(600000) // 600 sec timeout. From 43e6988a242ad0b4b4bb13a553eaf96243df3fde Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:10:31 +0200 Subject: [PATCH 16/19] Move createServer under src/shared --- benchmarks/postgres/ingestion.benchmark.ts | 2 +- benchmarks/vm/memory.benchmark.ts | 2 +- src/main/pluginsServer.ts | 2 +- src/{ => shared}/server.ts | 16 ++++++++-------- src/worker/worker.ts | 2 +- tests/plugins.test.ts | 2 +- tests/postgres/queue.test.ts | 2 +- tests/postgres/vm.test.ts | 2 +- tests/postgres/vm.timeout.test.ts | 2 +- tests/schedule.test.ts | 2 +- tests/shared/process-event.ts | 2 +- tests/sql.test.ts | 2 +- tests/transforms.test.ts | 2 +- 13 files changed, 20 insertions(+), 20 deletions(-) rename src/{ => shared}/server.ts (94%) diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index 96a6bceb..1fcf4330 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -4,8 +4,8 @@ import os from 'os' import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' -import { createServer } from '../../src/server' import { EventsProcessor } from '../../src/shared/ingestion/process-event' +import { createServer } from '../../src/shared/server' import { UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServer, SessionRecordingEvent, Team } from '../../src/types' import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' diff --git a/benchmarks/vm/memory.benchmark.ts b/benchmarks/vm/memory.benchmark.ts index 66034986..f2d1a422 100644 --- a/benchmarks/vm/memory.benchmark.ts +++ b/benchmarks/vm/memory.benchmark.ts @@ -1,6 +1,6 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' -import { createServer } from '../../src/server' +import { createServer } from '../../src/shared/server' import { Plugin, PluginConfig, PluginConfigVMReponse } from '../../src/types' import { createPluginConfigVM } from '../../src/worker/vm/vm' import { commonOrganizationId } from '../../tests/helpers/plugins' diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index 94bbfb23..a1e8b851 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -4,8 +4,8 @@ import { FastifyInstance } from 'fastify' import Redis from 'ioredis' import * as schedule from 'node-schedule' -import { createServer } from '../server' import { defaultConfig } from '../shared/config' +import { createServer } from '../shared/server' import { status } from '../shared/status' import { createRedis, delay } from '../shared/utils' import { PluginsServer, PluginsServerConfig, Queue, ScheduleControl } from '../types' diff --git a/src/server.ts b/src/shared/server.ts similarity index 94% rename from src/server.ts rename to src/shared/server.ts index ebd63355..5409de55 100644 --- a/src/server.ts +++ b/src/shared/server.ts @@ -9,14 +9,14 @@ import * as path from 'path' import { types as pgTypes } from 'pg' import { ConnectionOptions } from 'tls' -import { defaultConfig } from './shared/config' -import { DB } from './shared/db' -import { status } from './shared/status' -import { createPostgresPool, createRedis, UUIDT } from './shared/utils' -import { PluginsServer, PluginsServerConfig } from './types' -import { EventsProcessor } from './worker/ingestion/process-event' - -const { version } = require('../package.json') +import { PluginsServer, PluginsServerConfig } from '../types' +import { EventsProcessor } from '../worker/ingestion/process-event' +import { defaultConfig } from './config' +import { DB } from './db' +import { status } from './status' +import { createPostgresPool, createRedis, UUIDT } from './utils' + +const { version } = require('../../package.json') export async function createServer( config: Partial = {}, diff --git a/src/worker/worker.ts b/src/worker/worker.ts index 1e0101e0..38944d7b 100644 --- a/src/worker/worker.ts +++ b/src/worker/worker.ts @@ -1,5 +1,5 @@ import { initApp } from '../init' -import { createServer } from '../server' +import { createServer } from '../shared/server' import { status } from '../shared/status' import { cloneObject } from '../shared/utils' import { PluginsServer, PluginsServerConfig } from '../types' diff --git a/tests/plugins.test.ts b/tests/plugins.test.ts index 187fcdce..11a4fb17 100644 --- a/tests/plugins.test.ts +++ b/tests/plugins.test.ts @@ -1,8 +1,8 @@ import { PluginEvent } from '@posthog/plugin-scaffold/src/types' import { mocked } from 'ts-jest/utils' -import { createServer } from '../src/server' import { clearError, processError } from '../src/shared/error' +import { createServer } from '../src/shared/server' import { LogLevel, PluginsServer } from '../src/types' import { loadPlugin } from '../src/worker/plugins/loadPlugin' import { runPlugins } from '../src/worker/plugins/run' diff --git a/tests/postgres/queue.test.ts b/tests/postgres/queue.test.ts index f804d5c7..a28b43c3 100644 --- a/tests/postgres/queue.test.ts +++ b/tests/postgres/queue.test.ts @@ -1,6 +1,6 @@ import { startQueue } from '../../src/main/queue' -import { createServer } from '../../src/server' import Client from '../../src/shared/celery/client' +import { createServer } from '../../src/shared/server' import { delay } from '../../src/shared/utils' import { LogLevel, PluginsServer } from '../../src/types' import { runPlugins } from '../../src/worker/plugins/run' diff --git a/tests/postgres/vm.test.ts b/tests/postgres/vm.test.ts index d6862e37..d5ca1e70 100644 --- a/tests/postgres/vm.test.ts +++ b/tests/postgres/vm.test.ts @@ -1,8 +1,8 @@ import { PluginEvent } from '@posthog/plugin-scaffold' import * as fetch from 'node-fetch' -import { createServer } from '../../src/server' import Client from '../../src/shared/celery/client' +import { createServer } from '../../src/shared/server' import { delay } from '../../src/shared/utils' import { PluginsServer } from '../../src/types' import { createPluginConfigVM } from '../../src/worker/vm/vm' diff --git a/tests/postgres/vm.timeout.test.ts b/tests/postgres/vm.timeout.test.ts index f3d8cd61..1530333c 100644 --- a/tests/postgres/vm.timeout.test.ts +++ b/tests/postgres/vm.timeout.test.ts @@ -1,4 +1,4 @@ -import { createServer } from '../../src/server' +import { createServer } from '../../src/shared/server' import { PluginsServer } from '../../src/types' import { createPluginConfigVM } from '../../src/worker/vm/vm' import { pluginConfig39 } from '../helpers/plugins' diff --git a/tests/schedule.test.ts b/tests/schedule.test.ts index 287edd3b..e6fabb72 100644 --- a/tests/schedule.test.ts +++ b/tests/schedule.test.ts @@ -7,7 +7,7 @@ import { startSchedule, waitForTasksToFinish, } from '../src/main/services/schedule' -import { createServer } from '../src/server' +import { createServer } from '../src/shared/server' import { delay } from '../src/shared/utils' import { LogLevel, ScheduleControl } from '../src/types' import { createPromise } from './helpers/promises' diff --git a/tests/shared/process-event.ts b/tests/shared/process-event.ts index 7d1aa69b..0dcd12d1 100644 --- a/tests/shared/process-event.ts +++ b/tests/shared/process-event.ts @@ -4,8 +4,8 @@ import { DateTime } from 'luxon' import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' -import { createServer } from '../../src/server' import { hashElements } from '../../src/shared/ingestion/utils' +import { createServer } from '../../src/shared/server' import { delay, UUIDT } from '../../src/shared/utils' import { Database, diff --git a/tests/sql.test.ts b/tests/sql.test.ts index a0a82d4a..571fbc2f 100644 --- a/tests/sql.test.ts +++ b/tests/sql.test.ts @@ -1,4 +1,4 @@ -import { createServer } from '../src/server' +import { createServer } from '../src/shared/server' import { getPluginAttachmentRows, getPluginConfigRows, getPluginRows, setError } from '../src/shared/sql' import { PluginConfig, PluginError, PluginsServer } from '../src/types' import { commonOrganizationId } from './helpers/plugins' diff --git a/tests/transforms.test.ts b/tests/transforms.test.ts index 763a4d1a..726fe7fb 100644 --- a/tests/transforms.test.ts +++ b/tests/transforms.test.ts @@ -1,4 +1,4 @@ -import { createServer } from '../src/server' +import { createServer } from '../src/shared/server' import { code } from '../src/shared/utils' import { PluginsServer } from '../src/types' import { transformCode } from '../src/worker/vm/transforms' From fb326bd6ca3e1b0b84aa8b0b82cc63c12ef5acf3 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:20:17 +0200 Subject: [PATCH 17/19] Rename celery worker to be ingestion-specific --- .../{celery/worker.ts => ingestion/celery-queue-worker.ts} | 4 ++-- src/main/queue.ts | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) rename src/main/{celery/worker.ts => ingestion/celery-queue-worker.ts} (98%) diff --git a/src/main/celery/worker.ts b/src/main/ingestion/celery-queue-worker.ts similarity index 98% rename from src/main/celery/worker.ts rename to src/main/ingestion/celery-queue-worker.ts index 715d80a3..95c2a57b 100644 --- a/src/main/celery/worker.ts +++ b/src/main/ingestion/celery-queue-worker.ts @@ -5,7 +5,7 @@ import { Queue } from '../../types' type Handler = (...args: any[]) => Promise -export class Worker extends Base implements Queue { +export class CeleryQueueWorker extends Base implements Queue { handlers: Record = {} activeTasks: Set> = new Set() @@ -225,4 +225,4 @@ export class Worker extends Base implements Queue { } } -export default Worker +export default CeleryQueueWorker diff --git a/src/main/queue.ts b/src/main/queue.ts index 5df6a037..5e6fc70d 100644 --- a/src/main/queue.ts +++ b/src/main/queue.ts @@ -6,7 +6,7 @@ import Client from '../shared/celery/client' import { status } from '../shared/status' import { UUIDT } from '../shared/utils' import { IngestEventResponse, PluginsServer, Queue } from '../types' -import Worker from './celery/worker' +import CeleryQueueWorker from './ingestion/celery-queue-worker' import { KafkaQueue } from './ingestion/kafka-queue' export type WorkerMethods = { @@ -52,7 +52,7 @@ export async function startQueue( } function startQueueRedis(server: PluginsServer, piscina: Piscina | undefined, workerMethods: WorkerMethods): Queue { - const celeryQueue = new Worker(server.db, server.PLUGINS_CELERY_QUEUE) + const celeryQueue = new CeleryQueueWorker(server.db, server.PLUGINS_CELERY_QUEUE) const client = new Client(server.db, server.CELERY_DEFAULT_QUEUE) celeryQueue.register( From 7259e47ee56cd84ec87ce6b1e409007e6deb0a0c Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:21:15 +0200 Subject: [PATCH 18/19] Fix another package.json import --- src/main/pluginsServer.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/main/pluginsServer.ts b/src/main/pluginsServer.ts index a1e8b851..721a0a21 100644 --- a/src/main/pluginsServer.ts +++ b/src/main/pluginsServer.ts @@ -13,7 +13,7 @@ import { startQueue } from './queue' import { startSchedule } from './services/schedule' import { startFastifyInstance, stopFastifyInstance } from './web/server' -const { version } = require('../package.json') +const { version } = require('../../package.json') // TODO: refactor this into a class, removing the need for many different Servers export type ServerInstance = { From 6c6fef0f9bca0802e1b3236709315a9fd4be1468 Mon Sep 17 00:00:00 2001 From: Karl-Aksel Puulmann Date: Fri, 19 Mar 2021 12:25:52 +0200 Subject: [PATCH 19/19] Fix process-event import --- benchmarks/postgres/ingestion.benchmark.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/benchmarks/postgres/ingestion.benchmark.ts b/benchmarks/postgres/ingestion.benchmark.ts index 1fcf4330..6a58637c 100644 --- a/benchmarks/postgres/ingestion.benchmark.ts +++ b/benchmarks/postgres/ingestion.benchmark.ts @@ -4,10 +4,10 @@ import os from 'os' import { performance } from 'perf_hooks' import { IEvent } from '../../src/idl/protos' -import { EventsProcessor } from '../../src/shared/ingestion/process-event' import { createServer } from '../../src/shared/server' import { UUIDT } from '../../src/shared/utils' import { LogLevel, PluginsServer, SessionRecordingEvent, Team } from '../../src/types' +import { EventsProcessor } from '../../src/worker/ingestion/process-event' import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql' import { endLog, startLog } from './helpers/log' import { ingestCountEvents, setupPiscina } from './helpers/piscina'