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/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,66 @@ 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
+
+        // The TS library seem to demand you declare labels up-front (for some reason?),
+        // but we have know good way of knowing the set of labels used up-front, so
+        // idk what to do here really - recreate the metric each time? That seems...
+        // not good, maybe not feasible for histograms. Ingoring this problem for now,
+        // will return to it later (maybe get some help)
+        // let 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(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(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(value[i])
+            }
+        }
+    }
+}

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()