chore(recordings): remove hub dependency on recordings ingestion

plugin-server/functional_tests/session-recordings.test.ts

@@ -462,6 +462,20 @@ test.concurrent(
     20000
 )
 
+test.concurrent(`liveness check endpoint works`, async () => {
+    await waitForExpect(async () => {
+        const response = await fetch('http://localhost:6738/_health')
+        expect(response.status).toBe(200)
+
+        const body = await response.json()
+        expect(body).toEqual(
+            expect.objectContaining({
+                checks: expect.objectContaining({ 'session-recordings': 'ok' }),
+            })
+        )
+    })
+})
+
 test.concurrent(
     `consumer handles empty messages`,
     async () => {

plugin-server/src/capabilities.ts

@@ -8,6 +8,7 @@ export function getPluginServerCapabilities(config: PluginsServerConfig): Plugin
     switch (mode) {
         case null:
             return {
+                mmdb: true,
                 ingestion: true,
                 ingestionOverflow: true,
                 pluginScheduledTasks: true,
@@ -19,40 +20,55 @@ export function getPluginServerCapabilities(config: PluginsServerConfig): Plugin
         case 'ingestion':
             // NOTE: this mode will be removed in the future and replaced with
             // `analytics-ingestion` and `recordings-ingestion` modes.
-            return { ingestion: true, sessionRecordingIngestion: true, ...sharedCapabilities }
+            return {
+                mmdb: true,
+                ingestion: true,
+                sessionRecordingIngestion: true,
+                ...sharedCapabilities,
+            }
         case 'ingestion-overflow':
-            return { ingestionOverflow: true, ...sharedCapabilities }
+            return {
+                mmdb: true,
+                ingestionOverflow: true,
+                ...sharedCapabilities,
+            }
         case 'analytics-ingestion':
             return {
+                mmdb: true,
                 ingestion: true,
                 ...sharedCapabilities,
             }
         case 'recordings-ingestion':
             return {
+                mmdb: false,
                 sessionRecordingIngestion: true,
                 ...sharedCapabilities,
             }
 
         case 'async':
             return {
+                mmdb: true,
                 processPluginJobs: true,
                 processAsyncHandlers: true,
                 pluginScheduledTasks: true,
                 ...sharedCapabilities,
             }
         case 'exports':
             return {
+                mmdb: true,
                 processAsyncHandlers: true,
                 ...sharedCapabilities,
             }
         case 'jobs': {
             return {
+                mmdb: true,
                 processPluginJobs: true,
                 ...sharedCapabilities,
             }
         }
         case 'scheduler':
             return {
+                mmdb: true,
                 pluginScheduledTasks: true,
                 ...sharedCapabilities,
             }

plugin-server/src/index.ts

@@ -1,4 +1,5 @@
 import { Hub } from '../src/types'
+import { getPluginServerCapabilities } from './capabilities'
 import { defaultConfig, formatConfigHelp } from './config/config'
 import { healthcheckWithExit } from './healthcheck'
 import { initApp } from './init'
@@ -62,6 +63,7 @@ switch (alternativeMode) {
     default:
         // void the returned promise
         initApp(defaultConfig)
-        void startPluginsServer(defaultConfig, makePiscina)
+        const capabilities = getPluginServerCapabilities(defaultConfig)
+        void startPluginsServer(defaultConfig, makePiscina, capabilities)
         break
 }

plugin-server/src/main/ingestion-queues/session-recordings-consumer.ts

@@ -1,5 +1,4 @@
 import { PluginEvent } from '@posthog/plugin-scaffold'
-import { StatsD } from 'hot-shots'
 import { EachBatchPayload, Kafka } from 'kafkajs'
 import { exponentialBuckets, Histogram } from 'prom-client'
 
@@ -17,12 +16,10 @@ import { latestOffsetTimestampGauge } from './metrics'
 export const startSessionRecordingEventsConsumer = async ({
     teamManager,
     kafka,
-    statsd,
     partitionsConsumedConcurrently = 5,
 }: {
     teamManager: TeamManager
     kafka: Kafka
-    statsd?: StatsD
     partitionsConsumedConcurrently: number
 }) => {
     /*
@@ -40,10 +37,11 @@ export const startSessionRecordingEventsConsumer = async ({
     // the Kafka consumer handler.
     const producer = kafka.producer()
     await producer.connect()
-    const producerWrapper = new KafkaProducerWrapper(producer, statsd, { KAFKA_FLUSH_FREQUENCY_MS: 0 } as any)
+    const producerWrapper = new KafkaProducerWrapper(producer, undefined, { KAFKA_FLUSH_FREQUENCY_MS: 0 } as any)
 
     const groupId = 'session-recordings'
-    const consumer = kafka.consumer({ groupId: groupId })
+    const sessionTimeout = 30000
+    const consumer = kafka.consumer({ groupId: groupId, sessionTimeout: sessionTimeout })
     setupEventHandlers(consumer)
 
     status.info('🔁', 'Starting session recordings consumer')
@@ -61,7 +59,31 @@ export const startSessionRecordingEventsConsumer = async ({
         },
     })
 
-    return consumer
+    // Subscribe to the heatbeat event to track when the consumer has last
+    // successfully consumed a message. This is used to determine if the
+    // consumer is healthy.
+    const { HEARTBEAT } = consumer.events
+    let lastHeartbeat: number = Date.now()
+    consumer.on(HEARTBEAT, ({ timestamp }) => (lastHeartbeat = timestamp))
+
+    const isHealthy = async () => {
+        // Consumer has heartbeat within the session timeout, so it is healthy.
+        if (Date.now() - lastHeartbeat < sessionTimeout) {
+            return true
+        }
+
+        // Consumer has not heartbeat, but maybe it's because the group is
+        // currently rebalancing.
+        try {
+            const { state } = await consumer.describeGroup()
+
+            return ['CompletingRebalance', 'PreparingRebalance'].includes(state)
+        } catch (error) {
+            return false
+        }
+    }
+
+    return { consumer, isHealthy }
 }
 
 export const eachBatch =