feat(cyclotron): Add metrics everywhere

plugin-server/package.json

@@ -148,6 +148,6 @@
     },
     "cyclotron": {
         "//This is a short term workaround to ensure that cyclotron changes trigger a rebuild": true,
-        "version": "0.1.6"
+        "version": "0.1.7"
     }
 }

plugin-server/src/cdp/cdp-consumers.ts

@@ -52,6 +52,7 @@ import {
     convertToHogFunctionInvocationGlobals,
     createInvocation,
     cyclotronJobToInvocation,
+    emitCyclotronMetrics,
     gzipObject,
     invocationToCyclotronJobUpdate,
     prepareLogEntriesForClickhouse,
@@ -404,6 +405,7 @@ export class CdpProcessedEventsConsumer extends CdpConsumerBase {
     }
 
     public async processBatch(invocationGlobals: HogFunctionInvocationGlobals[]): Promise<HogFunctionInvocation[]> {
+        emitCyclotronMetrics()
         if (!invocationGlobals.length) {
             return []
         }
@@ -744,6 +746,7 @@ export class CdpCyclotronWorker extends CdpConsumerBase {
     protected queue: 'hog' | 'fetch' = 'hog'
 
     public async processBatch(invocations: HogFunctionInvocation[]): Promise<void> {
+        emitCyclotronMetrics()
         if (!invocations.length) {
             return
         }

plugin-server/src/cdp/utils.ts

@@ -1,8 +1,9 @@
 // NOTE: PostIngestionEvent is our context event - it should never be sent directly to an output, but rather transformed into a lightweight schema
 
-import { CyclotronJob, CyclotronJobUpdate } from '@posthog/cyclotron'
+import { CyclotronJob, CyclotronJobUpdate, getMetricsReport } from '@posthog/cyclotron'
 import { captureException } from '@sentry/node'
 import { DateTime } from 'luxon'
+import { Counter, exponentialBuckets, Gauge, Histogram } from 'prom-client'
 import RE2 from 're2'
 import { gunzip, gzip } from 'zlib'
 
@@ -378,3 +379,61 @@ export function cyclotronJobToInvocation(job: CyclotronJob, hogFunction: HogFunc
         timings: parsedState.timings,
     }
 }
+
+const CYCLOTRON_GAUGES: Map<string, Gauge> = new Map()
+const CYCLOTRON_COUNTERS: Map<string, Counter> = new Map()
+const CYCLOTRON_HISTOGRAMS: Map<string, Histogram> = new Map()
+const CYCLOTRON_HISTOGRAM_BOUNDS: number[] = exponentialBuckets(0.5, 2, 11) // 0.5 -> 512
+
+export function emitCyclotronMetrics() {
+    const report = getMetricsReport()
+    for (const measurement of report.measurements) {
+        const name = measurement.name
+        const type = measurement.type
+
+        const labels = measurement.labels
+
+        if (type === 'counter') {
+            const value = measurement.value as number
+            let counter = CYCLOTRON_COUNTERS.get(name)
+            if (!counter) {
+                counter = new Counter({
+                    name,
+                    help: name,
+                })
+                CYCLOTRON_COUNTERS.set(name, counter)
+            }
+            counter.reset()
+            counter.inc(labels as any, value)
+        }
+
+        if (type === 'gauge') {
+            const value = measurement.value as number
+            let gauge = CYCLOTRON_GAUGES.get(name)
+            if (!gauge) {
+                gauge = new Gauge({
+                    name,
+                    help: name,
+                })
+                CYCLOTRON_GAUGES.set(name, gauge)
+            }
+            gauge.set(labels as any, value)
+        }
+
+        if (type === 'histogram') {
+            const value = measurement.value as number[]
+            let histogram = CYCLOTRON_HISTOGRAMS.get(name)
+            if (!histogram) {
+                histogram = new Histogram({
+                    name,
+                    help: name,
+                    buckets: CYCLOTRON_HISTOGRAM_BOUNDS,
+                })
+                CYCLOTRON_HISTOGRAMS.set(name, histogram)
+            }
+            for (let i = 0; i < value.length; i++) {
+                histogram.observe(labels as any, value[i])
+            }
+        }
+    }
+}

plugin-server/src/main/pluginsServer.ts

@@ -1,3 +1,4 @@
+import { CyclotronMetricsConfig, initCyclotronMetrics } from '@posthog/cyclotron'
 import * as Sentry from '@sentry/node'
 import fs from 'fs'
 import { Server } from 'http'
@@ -503,6 +504,16 @@ export async function startPluginsServer(
             })
         })
 
+        if (cyclotronInUse(await setupHub())) {
+            const labels = new Map()
+            labels.set('plugin_server_version', version)
+            labels.set('plugin_server_mode', serverConfig.PLUGIN_SERVER_MODE)
+            const config: CyclotronMetricsConfig = {
+                defaultLabels: labels,
+            }
+            initCyclotronMetrics(config)
+        }
+
         return serverInstance
     } catch (error) {
         Sentry.captureException(error)
@@ -562,3 +573,11 @@ function runStartupProfiles(config: PluginsServerConfig) {
         }, config.STARTUP_PROFILE_DURATION_SECONDS * 1000)
     }
 }
+
+function cyclotronInUse(hub: Hub): boolean {
+    const urlSet = hub.CYCLOTRON_DATABASE_URL ? true : false
+    const cdpProcessedEvents = hub.capabilities.cdpProcessedEvents ? true : false
+    const cdpCyclotronWorker = hub.capabilities.cdpCyclotronWorker ? true : false
+
+    return (urlSet && cdpProcessedEvents) || cdpCyclotronWorker
+}

rust/Cargo.toml

@@ -90,4 +90,5 @@ url = { version = "2.5.0 " }
 uuid = { version = "1.6.1", features = ["v7", "serde"] }
 neon = "1"
 quick_cache = "0.6.5"
-ahash = "0.8.11"
+ahash = "0.8.11"
+metrics-util = "0.17.0"

rust/common/metrics/Cargo.toml

@@ -10,4 +10,5 @@ workspace = true
 axum = { workspace = true }
 metrics-exporter-prometheus = { workspace = true }
 tokio = { workspace = true }
-metrics = { workspace = true }
+metrics = { workspace = true }
+uuid = { workspace = true }

rust/common/metrics/src/lib.rs

@@ -17,8 +17,8 @@ pub async fn serve(router: Router, bind: &str) -> Result<(), std::io::Error> {
 }
 
 /// Add the prometheus endpoint and middleware to a router, should be called last.
-pub fn setup_metrics_routes(router: Router) -> Router {
-    let recorder_handle = setup_metrics_recorder();
+pub fn setup_metrics_routes(router: Router, service: &str, process: Option<&str>) -> Router {
+    let recorder_handle = setup_metrics_recorder(service, process);
 
     router
         .route(
@@ -28,12 +28,16 @@ pub fn setup_metrics_routes(router: Router) -> Router {
         .layer(axum::middleware::from_fn(track_metrics))
 }
 
-pub fn setup_metrics_recorder() -> PrometheusHandle {
+pub fn setup_metrics_recorder(service: &str, process: Option<&str>) -> PrometheusHandle {
+    let default_process = uuid::Uuid::now_v7().to_string();
+    let process = process.unwrap_or(&default_process);
     const BUCKETS: &[f64] = &[
         0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 50.0, 100.0, 250.0,
     ];
 
     PrometheusBuilder::new()
+        .add_global_label("service", service)
+        .add_global_label("process", process)
         .set_buckets(BUCKETS)
         .unwrap()
         .install_recorder()