refactor(plugin-server): yeet redis queue

plugin-server/src/main/ingestion-queues/queue.ts

@@ -1,26 +1,13 @@
 import Piscina from '@posthog/piscina'
 import { PluginEvent, ProcessedPluginEvent } from '@posthog/plugin-scaffold'
-import * as Sentry from '@sentry/node'
 
-import {
-    CeleryTriggeredJobOperation,
-    Hub,
-    PluginConfig,
-    PreIngestionEvent,
-    Queue,
-    Team,
-    WorkerMethods,
-} from '../../types'
+import { Hub, PreIngestionEvent, Queue, WorkerMethods } from '../../types'
 import { status } from '../../utils/status'
-import { sanitizeEvent, UUIDT } from '../../utils/utils'
 import { Action } from './../../types'
-import { CeleryQueue } from './celery-queue'
-import { ingestEvent } from './ingest-event'
 import { KafkaQueue } from './kafka-queue'
 
 interface Queues {
     ingestion: Queue | null
-    auxiliary: Queue | null
 }
 
 export function pauseQueueIfWorkerFull(
@@ -53,13 +40,8 @@ export async function startQueues(
     }
 
     try {
-        const redisQueue = startQueueRedis(server, piscina, mergedWorkerMethods)
-        const queues = {
-            ingestion: redisQueue,
-            auxiliary: redisQueue,
-        }
-        if (server.KAFKA_ENABLED) {
-            queues.ingestion = await startQueueKafka(server, mergedWorkerMethods)
+        const queues: Queues = {
+            ingestion: await startQueueKafka(server, mergedWorkerMethods),
         }
         return queues
     } catch (error) {
@@ -68,56 +50,6 @@ export async function startQueues(
     }
 }
 
-function startQueueRedis(server: Hub, piscina: Piscina, workerMethods: WorkerMethods): Queue | null {
-    const celeryQueue = new CeleryQueue(server.db, server.PLUGINS_CELERY_QUEUE)
-
-    let startQueue = false
-
-    // if kafka is enabled, we'll process events from there
-    if (!server.KAFKA_ENABLED && server.capabilities.ingestion) {
-        startQueue = true
-
-        celeryQueue.register(
-            'posthog.tasks.process_event.process_event_with_plugins',
-            async (
-                distinct_id: string,
-                ip: string | null,
-                site_url: string,
-                data: Record<string, unknown>,
-                team_id: number,
-                now: string,
-                sent_at?: string
-            ) => {
-                const event = sanitizeEvent({
-                    distinct_id,
-                    ip,
-                    site_url,
-                    team_id,
-                    now,
-                    sent_at,
-                    uuid: new UUIDT().toString(),
-                    ...data,
-                } as PluginEvent)
-                try {
-                    const checkAndPause = () => pauseQueueIfWorkerFull(() => celeryQueue.pause(), server, piscina)
-                    await ingestEvent(server, workerMethods, event, checkAndPause)
-                } catch (e) {
-                    Sentry.captureException(e)
-                }
-            }
-        )
-    }
-
-    if (startQueue) {
-        // run in the background
-        void celeryQueue.start()
-
-        return celeryQueue
-    } else {
-        return null
-    }
-}
-
 async function startQueueKafka(server: Hub, workerMethods: WorkerMethods): Promise<Queue | null> {
     if (!server.capabilities.ingestion) {
         return null

plugin-server/src/main/pluginsServer.ts

@@ -61,7 +61,6 @@ export async function startPluginsServer(
     let hub: Hub | undefined
     let piscina: Piscina | undefined
     let queue: Queue | undefined | null // ingestion queue
-    let redisQueueForPluginJobs: Queue | undefined | null
     let healthCheckConsumer: Consumer | undefined
     let jobQueueConsumer: JobQueueConsumerControl | undefined
     let closeHub: () => Promise<void> | undefined
@@ -87,7 +86,6 @@ export async function startPluginsServer(
         lastActivityCheck && clearInterval(lastActivityCheck)
         cancelAllScheduledJobs()
         await queue?.stop()
-        await redisQueueForPluginJobs?.stop()
         await pubSub?.stop()
         await jobQueueConsumer?.stop()
         await pluginScheduleControl?.stopSchedule()
@@ -160,16 +158,10 @@ export async function startPluginsServer(
 
         const queues = await startQueues(hub, piscina)
 
-        // `queue` refers to the ingestion queue. With Celery ingestion, we only
-        // have one queue for plugin jobs and ingestion. With Kafka ingestion, we
-        // use Kafka for events but still start Redis for plugin jobs.
-        // Thus, if Kafka is disabled, we don't need to call anything on
-        // redisQueueForPluginJobs, as that will also be the ingestion queue.
+        // `queue` refers to the ingestion queue.
         queue = queues.ingestion
-        redisQueueForPluginJobs = config.KAFKA_ENABLED ? queues.auxiliary : null
         piscina.on('drain', () => {
             void queue?.resume()
-            void redisQueueForPluginJobs?.resume()
             void jobQueueConsumer?.resume()
         })
 

plugin-server/tests/clickhouse/kafka-queue.test.ts

@@ -1,12 +1,10 @@
 import Piscina from '@posthog/piscina'
 
-import { ONE_HOUR } from '../../src/config/constants'
 import { KAFKA_EVENTS_PLUGIN_INGESTION } from '../../src/config/kafka-topics'
 import { ServerInstance, startPluginsServer } from '../../src/main/pluginsServer'
 import { LogLevel, PluginsServerConfig } from '../../src/types'
 import { Hub } from '../../src/types'
-import { Client } from '../../src/utils/celery/client'
-import { delay, UUIDT } from '../../src/utils/utils'
+import { UUIDT } from '../../src/utils/utils'
 import { makePiscina } from '../../src/worker/piscina'
 import { createPosthog, DummyPostHog } from '../../src/worker/vm/extensions/posthog'
 import { writeToFile } from '../../src/worker/vm/extensions/test-utils'