perf(ingestion): limit concurrency when setting-up plugins

package.json

@@ -128,6 +128,7 @@
         "posthog-js": "1.78.2",
         "posthog-js-lite": "2.0.0-alpha5",
         "prettier": "^2.8.8",
+        "prom-utils": "^0.5.0",
         "prop-types": "^15.7.2",
         "query-selector-shadow-dom": "^1.0.0",
         "rc-field-form": "~1.21.0",

plugin-server/src/config/config.ts

@@ -116,6 +116,7 @@ export function getDefaultConfig(): PluginsServerConfig {
         OBJECT_STORAGE_SECRET_ACCESS_KEY: 'object_storage_root_password',
         OBJECT_STORAGE_BUCKET: 'posthog',
         PLUGIN_SERVER_MODE: null,
+        PLUGIN_LOAD_CONCURRENCY: 50,
         KAFKAJS_LOG_LEVEL: 'WARN',
         HISTORICAL_EXPORTS_ENABLED: true,
         HISTORICAL_EXPORTS_MAX_RETRY_COUNT: 15,

plugin-server/src/types.ts

@@ -188,6 +188,7 @@ export interface PluginsServerConfig {
     OBJECT_STORAGE_SECRET_ACCESS_KEY: string
     OBJECT_STORAGE_BUCKET: string // the object storage bucket name
     PLUGIN_SERVER_MODE: PluginServerMode | null
+    PLUGIN_LOAD_CONCURRENCY: number // concurrency of the queue executing loadPlugin calls, to reduce heap usage spikes
     KAFKAJS_LOG_LEVEL: 'NOTHING' | 'DEBUG' | 'INFO' | 'WARN' | 'ERROR'
     HISTORICAL_EXPORTS_ENABLED: boolean // enables historical exports for export apps
     HISTORICAL_EXPORTS_MAX_RETRY_COUNT: number

plugin-server/src/worker/plugins/setup.ts

@@ -1,3 +1,5 @@
+import { rateLimit } from 'prom-utils'
+
 import { Hub, StatelessVmMap } from '../../types'
 import { LazyPluginVM } from '../vm/lazy'
 import { loadPlugin } from './loadPlugin'
@@ -7,9 +9,9 @@ import { teardownPlugins } from './teardown'
 
 export async function setupPlugins(hub: Hub): Promise<void> {
     const { plugins, pluginConfigs, pluginConfigsPerTeam } = await loadPluginsFromDB(hub)
-    const pluginVMLoadPromises: Array<Promise<any>> = []
     const statelessVms = {} as StatelessVmMap
 
+    const limiter = rateLimit(hub.PLUGIN_LOAD_CONCURRENCY)
     const timer = new Date()
 
     for (const [id, pluginConfig] of pluginConfigs) {
@@ -26,7 +28,7 @@ export async function setupPlugins(hub: Hub): Promise<void> {
             pluginConfig.vm = statelessVms[plugin.id]
         } else {
             pluginConfig.vm = new LazyPluginVM(hub, pluginConfig)
-            pluginVMLoadPromises.push(loadPlugin(hub, pluginConfig))
+            await limiter.add(loadPlugin(hub, pluginConfig)) // Will block if PLUGIN_LOAD_CONCURRENCY is reached
 
             if (prevConfig) {
                 void teardownPlugins(hub, prevConfig)
@@ -38,7 +40,7 @@ export async function setupPlugins(hub: Hub): Promise<void> {
         }
     }
 
-    await Promise.all(pluginVMLoadPromises)
+    await limiter.finish()
     hub.statsd?.timing('setup_plugins.success', timer)
 
     hub.plugins = plugins