feat: New ingestion consumer

.vscode/launch.json

@@ -144,7 +144,8 @@
                 "OBJECT_STORAGE_ENABLED": "True",
                 "HOG_HOOK_URL": "http://localhost:3300/hoghook",
                 "CDP_ASYNC_FUNCTIONS_RUSTY_HOOK_TEAMS": "",
-                "CDP_CYCLOTRON_ENABLED_TEAMS": "*"
+                "CDP_CYCLOTRON_ENABLED_TEAMS": "*",
+                "PLUGIN_SERVER_MODE": "all-v2"
             },
             "presentation": {
                 "group": "main"

frontend/src/scenes/experiments/ExperimentView/ExperimentView.tsx

@@ -19,10 +19,10 @@ import { ReleaseConditionsModal, ReleaseConditionsTable } from './ReleaseConditi
 import { SummaryTable } from './SummaryTable'
 
 const ResultsTab = (): JSX.Element => {
-    const { experiment, metricResults } = useValues(experimentLogic)
+    const { experiment, metricResults, primaryMetricsLengthWithSharedMetrics } = useValues(experimentLogic)
     const hasSomeResults = metricResults?.some((result) => result?.insight)
 
-    const hasSinglePrimaryMetric = experiment.metrics.length === 1
+    const hasSinglePrimaryMetric = primaryMetricsLengthWithSharedMetrics === 1
 
     return (
         <>

frontend/src/scenes/experiments/Metrics/SharedMetricModal.tsx

@@ -2,7 +2,7 @@ import { LemonBanner, LemonButton, LemonModal, Link } from '@posthog/lemon-ui'
 import { useActions, useValues } from 'kea'
 import { IconOpenInNew } from 'lib/lemon-ui/icons'
 import { LemonTable } from 'lib/lemon-ui/LemonTable'
-import { useEffect, useState } from 'react'
+import { useState } from 'react'
 import { urls } from 'scenes/urls'
 
 import { Experiment } from '~/types'
@@ -37,15 +37,7 @@ export function SharedMetricModal({
     } = useActions(experimentLogic({ experimentId }))
 
     const [selectedMetricIds, setSelectedMetricIds] = useState<SharedMetric['id'][]>([])
-    const [selectedMetricId, setSelectedMetricId] = useState<SharedMetric['id'] | null>(null)
-    const [mode, setMode] = useState<'create' | 'edit'>('create')
-
-    useEffect(() => {
-        if (editingSharedMetricId) {
-            setSelectedMetricId(editingSharedMetricId)
-            setMode('edit')
-        }
-    }, [editingSharedMetricId])
+    const mode = editingSharedMetricId ? 'edit' : 'create'
 
     if (!sharedMetrics) {
         return <></>
@@ -196,10 +188,10 @@ export function SharedMetricModal({
                 </div>
             )}
 
-            {selectedMetricId && (
+            {editingSharedMetricId && (
                 <div>
                     {(() => {
-                        const metric = sharedMetrics.find((m: SharedMetric) => m.id === selectedMetricId)
+                        const metric = sharedMetrics.find((m: SharedMetric) => m.id === editingSharedMetricId)
                         if (!metric) {
                             return <></>
                         }

frontend/src/scenes/experiments/experimentLogic.tsx

@@ -534,8 +534,6 @@ export const experimentLogic = kea<experimentLogicType>([
             {
                 openPrimarySharedMetricModal: (_, { sharedMetricId }) => sharedMetricId,
                 openSecondarySharedMetricModal: (_, { sharedMetricId }) => sharedMetricId,
-                closePrimarySharedMetricModal: () => null,
-                closeSecondarySharedMetricModal: () => null,
                 updateExperimentGoal: () => null,
             },
         ],

frontend/utils.mjs

@@ -543,7 +543,7 @@ export function gatherProductUrls(products) {
 }
 
 export function gatherProductManifests() {
-    const products = fse.readdirSync('products').filter((p) => p !== '__pycache__')
+    const products = fse.readdirSync('products').filter((p) => !['__pycache__', 'README.md'].includes(p))
     const allScenes = {}
     const allRoutes = {}
     const allRedirects = {}

plugin-server/src/capabilities.ts

@@ -7,29 +7,49 @@ export function getPluginServerCapabilities(config: PluginsServerConfig): Plugin
         : null
     const sharedCapabilities = !isTestEnv() ? { http: true } : {}
 
+    const singleProcessCapabilities: PluginServerCapabilities = {
+        mmdb: true,
+        ingestion: true,
+        ingestionOverflow: true,
+        ingestionHistorical: true,
+        eventsIngestionPipelines: true, // with null PluginServerMode we run all of them
+        pluginScheduledTasks: true,
+        processPluginJobs: true,
+        processAsyncOnEventHandlers: true,
+        processAsyncWebhooksHandlers: true,
+        sessionRecordingBlobIngestion: true,
+        sessionRecordingBlobOverflowIngestion: config.SESSION_RECORDING_OVERFLOW_ENABLED,
+        sessionRecordingBlobIngestionV2: true,
+        sessionRecordingBlobIngestionV2Overflow: config.SESSION_RECORDING_OVERFLOW_ENABLED,
+        appManagementSingleton: true,
+        preflightSchedules: true,
+        cdpProcessedEvents: true,
+        cdpInternalEvents: true,
+        cdpFunctionCallbacks: true,
+        cdpCyclotronWorker: true,
+        syncInlinePlugins: true,
+        ...sharedCapabilities,
+    }
+
     switch (mode) {
         case null:
+            return {
+                ...singleProcessCapabilities,
+                ...sharedCapabilities,
+            }
+        case PluginServerMode.all_v2:
+            return {
+                ...singleProcessCapabilities,
+                ...sharedCapabilities,
+                ingestionV2Combined: true,
+            }
+
+        case PluginServerMode.ingestion_v2:
+            // NOTE: this mode will be removed in the future and replaced with
+            // `analytics-ingestion` and `recordings-ingestion` modes.
             return {
                 mmdb: true,
-                ingestion: true,
-                ingestionOverflow: true,
-                ingestionHistorical: true,
-                eventsIngestionPipelines: true, // with null PluginServerMode we run all of them
-                pluginScheduledTasks: true,
-                processPluginJobs: true,
-                processAsyncOnEventHandlers: true,
-                processAsyncWebhooksHandlers: true,
-                sessionRecordingBlobIngestion: true,
-                sessionRecordingBlobOverflowIngestion: config.SESSION_RECORDING_OVERFLOW_ENABLED,
-                sessionRecordingBlobIngestionV2: true,
-                sessionRecordingBlobIngestionV2Overflow: config.SESSION_RECORDING_OVERFLOW_ENABLED,
-                appManagementSingleton: true,
-                preflightSchedules: true,
-                cdpProcessedEvents: true,
-                cdpInternalEvents: true,
-                cdpFunctionCallbacks: true,
-                cdpCyclotronWorker: true,
-                syncInlinePlugins: true,
+                ingestionV2: true,
                 ...sharedCapabilities,
             }
         case PluginServerMode.ingestion:

plugin-server/src/config/config.ts

@@ -5,6 +5,7 @@ import {
     KAFKA_CLICKHOUSE_HEATMAP_EVENTS,
     KAFKA_EVENTS_JSON,
     KAFKA_EVENTS_PLUGIN_INGESTION,
+    KAFKA_EVENTS_PLUGIN_INGESTION_DLQ,
     KAFKA_EVENTS_PLUGIN_INGESTION_OVERFLOW,
     KAFKA_EXCEPTION_SYMBOLIFICATION_EVENTS,
 } from './kafka-topics'
@@ -127,6 +128,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         EXTERNAL_REQUEST_TIMEOUT_MS: 10 * 1000, // 10 seconds
         DROP_EVENTS_BY_TOKEN_DISTINCT_ID: '',
         DROP_EVENTS_BY_TOKEN: '',
+        SKIP_PERSONS_PROCESSING_BY_TOKEN_DISTINCT_ID: '',
         PIPELINE_STEP_STALLED_LOG_TIMEOUT: 30,
         RELOAD_PLUGIN_JITTER_MAX_MS: 60000,
         RUSTY_HOOK_FOR_TEAMS: '',
@@ -205,6 +207,12 @@ export function getDefaultConfig(): PluginsServerConfig {
             : '',
 
         CYCLOTRON_SHARD_DEPTH_LIMIT: 1000000,
+
+        // New IngestionConsumer config
+        INGESTION_CONSUMER_GROUP_ID: 'events-ingestion-consumer',
+        INGESTION_CONSUMER_CONSUME_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION,
+        INGESTION_CONSUMER_OVERFLOW_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION_OVERFLOW,
+        INGESTION_CONSUMER_DLQ_TOPIC: KAFKA_EVENTS_PLUGIN_INGESTION_DLQ,
     }
 }
 

plugin-server/src/ingestion/__snapshots__/ingestion-consumer.test.ts.snap

@@ -0,0 +1,49 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`IngestionConsumer general should process a standard event 1`] = `
+[
+  {
+    "key": "<REPLACED-UUID-0>",
+    "topic": "clickhouse_events_json_test",
+    "value": {
+      "created_at": "2025-01-31 23:00:00.000",
+      "distinct_id": "user-1",
+      "elements_chain": "",
+      "event": "$pageview",
+      "person_created_at": "2025-01-31 23:00:00",
+      "person_id": "<REPLACED-UUID-1>",
+      "person_mode": "full",
+      "person_properties": "{"$current_url":"http://localhost:8000","$creator_event_uuid":"<REPLACED-UUID-2>","$initial_current_url":"http://localhost:8000"}",
+      "project_id": 2,
+      "properties": "{"$current_url":"http://localhost:8000","$ip":"127.0.0.1","$set":{"$current_url":"http://localhost:8000"},"$set_once":{"$initial_current_url":"http://localhost:8000"}}",
+      "team_id": 2,
+      "timestamp": "2025-01-31 23:00:00.000",
+      "uuid": "<REPLACED-UUID-1>",
+    },
+  },
+  {
+    "key": null,
+    "topic": "clickhouse_person_test",
+    "value": {
+      "created_at": "2025-01-31 23:00:00",
+      "id": "<REPLACED-UUID-2>",
+      "is_deleted": 0,
+      "is_identified": 0,
+      "properties": "{"$current_url":"http://localhost:8000","$creator_event_uuid":"<REPLACED-UUID-2>","$initial_current_url":"http://localhost:8000"}",
+      "team_id": 2,
+      "version": 0,
+    },
+  },
+  {
+    "key": null,
+    "topic": "clickhouse_person_distinct_id_test",
+    "value": {
+      "distinct_id": "user-1",
+      "is_deleted": 0,
+      "person_id": "<REPLACED-UUID-2>",
+      "team_id": 2,
+      "version": 0,
+    },
+  },
+]
+`;