Refactor PluginsServer to Hub

README.md

@@ -86,7 +86,8 @@ Let's talk about the main thread first. This has:
 
 1. `pubSub` – Redis powered pub-sub mechanism for reloading plugins whenever a message is published by the main PostHog app.
 
-2. `server` – Handler of connections to required DBs and queues (ClickHouse, Kafka, Postgres, Redis), created via `server.ts -> createServer`. Every thread has its own instance.
+2. `hub` – Handler of connections to required DBs and queues (ClickHouse, Kafka, Postgres, Redis), holds loaded plugins.
+   Created via `hub.ts -> createHub`. Every thread has its own instance.
 
 3. `fastifyInstance` – Web server. Unused for now.
 
@@ -110,7 +111,7 @@ Let's talk about the main thread first. This has:
 
 This begins with `worker.ts` and `createWorker()`.
 
-`server` is the same setup as in the main thread.
+`hub` is the same setup as in the main thread.
 
 New functions called here are:
 

benchmarks/clickhouse/e2e.kafka.benchmark.ts

@@ -2,8 +2,7 @@ import { performance } from 'perf_hooks'
 
 import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/config/kafka-topics'
 import { startPluginsServer } from '../../src/main/pluginsServer'
-import { LogLevel, PluginsServerConfig, Queue } from '../../src/types'
-import { PluginsServer } from '../../src/types'
+import { Hub, LogLevel, PluginsServerConfig, Queue } from '../../src/types'
 import { delay, UUIDT } from '../../src/utils/utils'
 import { makePiscina } from '../../src/worker/piscina'
 import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog'
@@ -26,7 +25,7 @@ const extraServerConfig: Partial<PluginsServerConfig> = {
 
 describe('e2e kafka & clickhouse benchmark', () => {
     let queue: Queue
-    let server: PluginsServer
+    let hub: Hub
     let stopServer: () => Promise<void>
     let posthog: DummyPostHog
 
@@ -45,11 +44,11 @@ describe('e2e kafka & clickhouse benchmark', () => {
         await resetTestDatabaseClickhouse(extraServerConfig)
 
         const startResponse = await startPluginsServer(extraServerConfig, makePiscina)
-        server = startResponse.server
+        hub = startResponse.hub
         stopServer = startResponse.stop
         queue = startResponse.queue
 
-        posthog = createPosthog(server, pluginConfig39)
+        posthog = createPosthog(hub, pluginConfig39)
     })
 
     afterEach(async () => {
@@ -77,7 +76,7 @@ describe('e2e kafka & clickhouse benchmark', () => {
 
         console.log('Starting timer')
         const startTime = performance.now()
-        await delayUntilEventIngested(() => server.db.fetchEvents(), count, 500, count)
+        await delayUntilEventIngested(() => hub.db.fetchEvents(), count, 500, count)
         const timeMs = performance.now() - startTime
         console.log('Finished!')
 
@@ -88,7 +87,7 @@ describe('e2e kafka & clickhouse benchmark', () => {
             )} events/sec, ${n(timeMs / count)}ms per event)`
         )
 
-        const events = await server.db.fetchEvents()
+        const events = await hub.db.fetchEvents()
         expect(events[count - 1].properties.upperUuid).toEqual(events[count - 1].properties.uuid.toUpperCase())
     })
 })

benchmarks/clickhouse/e2e.timeout.benchmark.ts

@@ -2,8 +2,7 @@ import { performance } from 'perf_hooks'
 
 import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/config/kafka-topics'
 import { startPluginsServer } from '../../src/main/pluginsServer'
-import { ClickHouseEvent, LogLevel, PluginsServerConfig, Queue } from '../../src/types'
-import { PluginsServer } from '../../src/types'
+import { ClickHouseEvent, Hub, LogLevel, PluginsServerConfig, Queue } from '../../src/types'
 import { delay, UUIDT } from '../../src/utils/utils'
 import { makePiscina } from '../../src/worker/piscina'
 import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog'
@@ -27,7 +26,7 @@ const extraServerConfig: Partial<PluginsServerConfig> = {
 
 describe('e2e kafka processing timeout benchmark', () => {
     let queue: Queue
-    let server: PluginsServer
+    let hub: Hub
     let stopServer: () => Promise<void>
     let posthog: DummyPostHog
 
@@ -43,11 +42,11 @@ describe('e2e kafka processing timeout benchmark', () => {
         await resetTestDatabaseClickhouse(extraServerConfig)
 
         const startResponse = await startPluginsServer(extraServerConfig, makePiscina)
-        server = startResponse.server
+        hub = startResponse.hub
         stopServer = startResponse.stop
         queue = startResponse.queue
 
-        posthog = createPosthog(server, pluginConfig39)
+        posthog = createPosthog(hub, pluginConfig39)
     })
 
     afterEach(async () => {
@@ -75,7 +74,7 @@ describe('e2e kafka processing timeout benchmark', () => {
 
         console.log('Starting timer')
         const startTime = performance.now()
-        await delayUntilEventIngested(() => server.db.fetchEvents(), count, 500, count)
+        await delayUntilEventIngested(() => hub.db.fetchEvents(), count, 500, count)
         const timeMs = performance.now() - startTime
         console.log('Finished!')
 
@@ -85,7 +84,7 @@ describe('e2e kafka processing timeout benchmark', () => {
                 1000 / (timeMs / count)
             )} events/sec, ${n(timeMs / count)}ms per event)`
         )
-        const events = (await server.db.fetchEvents()) as ClickHouseEvent[]
+        const events = (await hub.db.fetchEvents()) as ClickHouseEvent[]
         const passedEvents = events.filter((e) => e.properties.timeout).length
         console.log(
             `ℹ️ Out of 3000 events: ${passedEvents} took under 5sec, ${

benchmarks/postgres/e2e.celery.benchmark.ts

@@ -2,8 +2,7 @@ import * as IORedis from 'ioredis'
 import { performance } from 'perf_hooks'
 
 import { startPluginsServer } from '../../src/main/pluginsServer'
-import { LogLevel, PluginsServerConfig, Queue } from '../../src/types'
-import { PluginsServer } from '../../src/types'
+import { Hub, LogLevel, PluginsServerConfig, Queue } from '../../src/types'
 import { delay, UUIDT } from '../../src/utils/utils'
 import { makePiscina } from '../../src/worker/piscina'
 import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog'
@@ -25,7 +24,7 @@ const extraServerConfig: Partial<PluginsServerConfig> = {
 
 describe('e2e celery & postgres benchmark', () => {
     let queue: Queue
-    let server: PluginsServer
+    let hub: Hub
     let stopServer: () => Promise<void>
     let posthog: DummyPostHog
     let redis: IORedis.Redis
@@ -40,19 +39,19 @@ describe('e2e celery & postgres benchmark', () => {
         `)
 
         const startResponse = await startPluginsServer(extraServerConfig, makePiscina)
-        server = startResponse.server
+        hub = startResponse.hub
         stopServer = startResponse.stop
         queue = startResponse.queue
-        redis = await server.redisPool.acquire()
+        redis = await hub.redisPool.acquire()
 
-        await redis.del(server.PLUGINS_CELERY_QUEUE)
-        await redis.del(server.CELERY_DEFAULT_QUEUE)
+        await redis.del(hub.PLUGINS_CELERY_QUEUE)
+        await redis.del(hub.CELERY_DEFAULT_QUEUE)
 
-        posthog = createPosthog(server, pluginConfig39)
+        posthog = createPosthog(hub, pluginConfig39)
     })
 
     afterEach(async () => {
-        await server.redisPool.release(redis)
+        await hub.redisPool.release(redis)
         await stopServer()
     })
 
@@ -67,17 +66,17 @@ describe('e2e celery & postgres benchmark', () => {
             posthog.capture('custom event', { name: 'haha', uuid, randomProperty: 'lololo' })
         }
         await queue.pause()
-        expect(await redis.llen(server.PLUGINS_CELERY_QUEUE)).toEqual(0)
+        expect(await redis.llen(hub.PLUGINS_CELERY_QUEUE)).toEqual(0)
         for (let i = 0; i < count; i++) {
             createEvent()
         }
         await delay(3000)
-        expect(await redis.llen(server.PLUGINS_CELERY_QUEUE)).toEqual(count)
+        expect(await redis.llen(hub.PLUGINS_CELERY_QUEUE)).toEqual(count)
         await queue.resume()
 
         console.log('Starting timer')
         const startTime = performance.now()
-        await delayUntilEventIngested(() => server.db.fetchEvents(), count, 500, count)
+        await delayUntilEventIngested(() => hub.db.fetchEvents(), count, 500, count)
         const timeMs = performance.now() - startTime
         console.log('Finished!')
 
@@ -88,7 +87,7 @@ describe('e2e celery & postgres benchmark', () => {
             )} events/sec, ${n(timeMs / count)}ms per event)`
         )
 
-        const events = await server.db.fetchEvents()
+        const events = await hub.db.fetchEvents()
         expect(events[count - 1].properties.upperUuid).toEqual(events[count - 1].properties.uuid.toUpperCase())
     })
 })

benchmarks/postgres/ingestion.benchmark.ts

@@ -4,8 +4,8 @@ import * as os from 'os'
 import { performance } from 'perf_hooks'
 
 import { IEvent } from '../../src/config/idl/protos'
-import { LogLevel, PluginsServer, SessionRecordingEvent, Team } from '../../src/types'
-import { createServer } from '../../src/utils/db/server'
+import { Hub, LogLevel, SessionRecordingEvent, Team } from '../../src/types'
+import { createHub } from '../../src/utils/db/hub'
 import { UUIDT } from '../../src/utils/utils'
 import { EventsProcessor } from '../../src/worker/ingestion/process-event'
 import { getFirstTeam, resetTestDatabase } from '../../tests/helpers/sql'
@@ -17,8 +17,8 @@ jest.setTimeout(600000) // 600 sec timeout
 
 describe('ingestion benchmarks', () => {
     let team: Team
-    let server: PluginsServer
-    let stopServer: () => Promise<void>
+    let hub: Hub
+    let closeHub: () => Promise<void>
     let eventsProcessor: EventsProcessor
     let now = DateTime.utc()
 
@@ -46,13 +46,13 @@ describe('ingestion benchmarks', () => {
                 return event
             }
         `)
-        ;[server, stopServer] = await createServer({
+        ;[hub, closeHub] = await createHub({
             PLUGINS_CELERY_QUEUE: 'benchmark-plugins-celery-queue',
             CELERY_DEFAULT_QUEUE: 'benchmark-celery-default-queue',
             LOG_LEVEL: LogLevel.Log,
         })
-        eventsProcessor = new EventsProcessor(server)
-        team = await getFirstTeam(server)
+        eventsProcessor = new EventsProcessor(hub)
+        team = await getFirstTeam(hub)
         now = DateTime.utc()
 
         // warmup
@@ -62,7 +62,7 @@ describe('ingestion benchmarks', () => {
     })
 
     afterEach(async () => {
-        await stopServer?.()
+        await closeHub?.()
     })
 
     test('basic sequential ingestion', async () => {

benchmarks/vm/memory.benchmark.ts

@@ -1,7 +1,7 @@
 import { PluginEvent } from '@posthog/plugin-scaffold/src/types'
 
 import { Plugin, PluginConfig, PluginConfigVMReponse } from '../../src/types'
-import { createServer } from '../../src/utils/db/server'
+import { createHub } from '../../src/utils/db/hub'
 import { createPluginConfigVM } from '../../src/worker/vm/vm'
 import { commonOrganizationId } from '../../tests/helpers/plugins'
 
@@ -57,7 +57,7 @@ test('test vm memory usage', async () => {
     const numVMs = 1000
     const numEventsPerVM = 100
 
-    const [server, closeServer] = await createServer()
+    const [hub, closeHub] = await createHub()
     const indexJs = `
         async function processEvent (event, meta) {
             event.event = 'changed event'
@@ -80,7 +80,7 @@ test('test vm memory usage', async () => {
     const vms: PluginConfigVMReponse[] = []
 
     for (let i = 0; i < numVMs; i++) {
-        const vm = await createPluginConfigVM(server, mockConfig, indexJs)
+        const vm = await createPluginConfigVM(hub, mockConfig, indexJs)
         vms.push(vm)
 
         if (debug || i === numVMs - 1) {
@@ -110,5 +110,5 @@ test('test vm memory usage', async () => {
         }
     }
 
-    await closeServer()
+    await closeHub()
 })

src/main/ingestion-queues/ingest-event.ts

@@ -1,12 +1,12 @@
 import { PluginEvent } from '@posthog/plugin-scaffold'
 import * as Sentry from '@sentry/node'
 
-import { PluginsServer, WorkerMethods } from '../../types'
+import { Hub, WorkerMethods } from '../../types'
 import { timeoutGuard } from '../../utils/db/utils'
 import { status } from '../../utils/status'
 
 export async function ingestEvent(
-    server: PluginsServer,
+    server: Hub,
     workerMethods: WorkerMethods,
     event: PluginEvent,
     checkAndPause?: () => void // pause incoming messages if we are slow in getting them out again
@@ -62,7 +62,7 @@ async function runInstrumentedFunction({
     func,
     statsKey,
 }: {
-    server: PluginsServer
+    server: Hub
     event: PluginEvent
     timeoutMessage: string
     statsKey: string

src/main/ingestion-queues/kafka-queue.ts

@@ -2,21 +2,21 @@ import Piscina from '@posthog/piscina'
 import { PluginEvent } from '@posthog/plugin-scaffold'
 import * as Sentry from '@sentry/node'
 import { Consumer, EachBatchPayload, Kafka, KafkaMessage } from 'kafkajs'
-import { PluginsServer, Queue, WorkerMethods } from 'types'
+import { Hub, Queue, WorkerMethods } from 'types'
 
 import { status } from '../../utils/status'
 import { groupIntoBatches, killGracefully, sanitizeEvent } from '../../utils/utils'
 import { ingestEvent } from './ingest-event'
 
 export class KafkaQueue implements Queue {
-    private pluginsServer: PluginsServer
+    private pluginsServer: Hub
     private piscina: Piscina
     private kafka: Kafka
     private consumer: Consumer
     private wasConsumerRan: boolean
     private workerMethods: WorkerMethods
 
-    constructor(pluginsServer: PluginsServer, piscina: Piscina, workerMethods: WorkerMethods) {
+    constructor(pluginsServer: Hub, piscina: Piscina, workerMethods: WorkerMethods) {
         this.pluginsServer = pluginsServer
         this.piscina = piscina
         this.kafka = pluginsServer.kafka!

src/main/ingestion-queues/queue.ts

@@ -2,7 +2,7 @@ import Piscina from '@posthog/piscina'
 import { PluginEvent } from '@posthog/plugin-scaffold'
 import * as Sentry from '@sentry/node'
 
-import { PluginsServer, Queue, WorkerMethods } from '../../types'
+import { Hub, Queue, WorkerMethods } from '../../types'
 import { status } from '../../utils/status'
 import { sanitizeEvent, UUIDT } from '../../utils/utils'
 import { CeleryQueue } from './celery-queue'
@@ -11,7 +11,7 @@ import { KafkaQueue } from './kafka-queue'
 
 export function pauseQueueIfWorkerFull(
     pause: undefined | (() => void | Promise<void>),
-    server: PluginsServer,
+    server: Hub,
     piscina?: Piscina
 ): void {
     if (pause && (piscina?.queueSize || 0) > (server.WORKER_CONCURRENCY || 4) * (server.WORKER_CONCURRENCY || 4)) {
@@ -20,7 +20,7 @@ export function pauseQueueIfWorkerFull(
 }
 
 export async function startQueue(
-    server: PluginsServer,
+    server: Hub,
     piscina: Piscina,
     workerMethods: Partial<WorkerMethods> = {}
 ): Promise<Queue> {
@@ -65,7 +65,7 @@ export async function startQueue(
     }
 }
 
-function startQueueRedis(server: PluginsServer, piscina: Piscina | undefined, workerMethods: WorkerMethods): Queue {
+function startQueueRedis(server: Hub, piscina: Piscina | undefined, workerMethods: WorkerMethods): Queue {
     const celeryQueue = new CeleryQueue(server.db, server.PLUGINS_CELERY_QUEUE)
 
     celeryQueue.register(
@@ -104,7 +104,7 @@ function startQueueRedis(server: PluginsServer, piscina: Piscina | undefined, wo
     return celeryQueue
 }
 
-async function startQueueKafka(server: PluginsServer, piscina: Piscina, workerMethods: WorkerMethods): Promise<Queue> {
+async function startQueueKafka(server: Hub, piscina: Piscina, workerMethods: WorkerMethods): Promise<Queue> {
     const kafkaQueue: Queue = new KafkaQueue(server, piscina, workerMethods)
     await kafkaQueue.start()
 

src/main/job-queues/job-queue-consumer.ts

@@ -1,7 +1,7 @@
 import Piscina from '@posthog/piscina'
 import * as Sentry from '@sentry/node'
 
-import { JobQueueConsumerControl, OnJobCallback, PluginsServer } from '../../types'
+import { Hub, JobQueueConsumerControl, OnJobCallback } from '../../types'
 import { killProcess } from '../../utils/kill'
 import { startRedlock } from '../../utils/redlock'
 import { status } from '../../utils/status'
@@ -10,7 +10,7 @@ import { pauseQueueIfWorkerFull } from '../ingestion-queues/queue'
 
 export const LOCKED_RESOURCE = 'plugin-server:locks:job-queue-consumer'
 
-export async function startJobQueueConsumer(server: PluginsServer, piscina: Piscina): Promise<JobQueueConsumerControl> {
+export async function startJobQueueConsumer(server: Hub, piscina: Piscina): Promise<JobQueueConsumerControl> {
     status.info('🔄', 'Starting job queue consumer, trying to get lock...')
 
     const onJob: OnJobCallback = async (jobs) => {