fix(webapp): emit realtime backend metrics through OpenTelemetry

The realtime backend's metrics were registered with the in-process
Prometheus registry, which is no longer how the webapp ships metrics.
They now emit through the OpenTelemetry meter (realtime_native.*,
realtime_notifier.*, realtime_shadow.*) so they flow through the
internal metrics exporter like the rest of the webapp's
instrumentation. Gauges become observable gauges sampled at export
time; names move from prom-style _total suffixes to the meter's
dot-namespaced convention.

Author: ericallam

apps/webapp/app/services/realtime/nativeRealtimeClientInstance.server.ts

@@ -1,7 +1,6 @@
-import { Counter, Gauge, Histogram } from "prom-client";
+import { getMeter } from "@internal/tracing";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
-import { metricsRegister } from "~/metrics.server";
 import { singleton } from "~/utils/singleton";
 import { getCachedLimit } from "../platform.v3.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
@@ -15,87 +14,67 @@ import { RunHydrator } from "./runReader.server";
 // Process-singleton wiring for the native realtime client; only constructed when a
 // request actually routes to it, so a disabled webapp never instantiates it.
 function initializeNativeRealtimeClient(): NativeRealtimeClient {
-  const wakeups = new Counter({
-    name: "realtime_native_wakeups_total",
-    help: "Live realtime wakeups by reason. A rising 'timeout' share suggests a write site is missing its publishChangeRecord delegate.",
-    labelNames: ["reason"] as const,
-    registers: [metricsRegister],
+  const meter = getMeter("realtime-native");
+
+  const wakeups = meter.createCounter("realtime_native.wakeups", {
+    description:
+      "Live realtime wakeups by reason. A rising 'timeout' share suggests a write site is missing its publishChangeRecord delegate.",
   });
 
-  const runSetResolves = new Counter({
-    name: "realtime_native_runset_resolve_total",
-    help: "Multi-run (tag-list/batch) resolve+hydrate outcomes. 'hit'/'coalesced' vs 'miss' shows how effectively concurrent same-filter feeds share a single ClickHouse + Postgres query under an env-wide wake.",
-    labelNames: ["result"] as const,
-    registers: [metricsRegister],
+  const runSetResolves = meter.createCounter("realtime_native.runset_resolves", {
+    description:
+      "Multi-run (tag-list/batch) resolve+hydrate outcomes. 'hit'/'coalesced' vs 'miss' shows how effectively concurrent same-filter feeds share a single ClickHouse + Postgres query.",
   });
 
-  const runSetQueryMs = new Histogram({
-    name: "realtime_native_runset_query_ms",
-    help: "Latency of the multi-run resolve (ClickHouse) and hydrate (Postgres) stages.",
-    labelNames: ["stage"] as const,
-    buckets: [1, 5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000],
-    registers: [metricsRegister],
+  const runSetQueryMs = meter.createHistogram("realtime_native.runset_query_ms", {
+    description: "Latency of the multi-run resolve (ClickHouse) and hydrate (Postgres) stages.",
+    unit: "ms",
   });
 
-  const livePollPaths = new Counter({
-    name: "realtime_native_live_poll_total",
-    help: "How live polls resolved. 'fast-hydrate' = the router woke the feed with matched runs hydrated by id (no ClickHouse); 'full-resolve' = the backstop timeout did a ClickHouse resolve. A high fast-path share is the local-membership routing working.",
-    labelNames: ["path"] as const,
-    registers: [metricsRegister],
+  const livePollPaths = meter.createCounter("realtime_native.live_polls", {
+    description:
+      "How live polls resolved. 'fast-hydrate' = router wake with rows hydrated by id (no ClickHouse); 'full-resolve' = backstop; 'cold-resolve' = fresh env subscription probed once.",
   });
 
-  const routerHydrates = new Counter({
-    name: "realtime_native_router_hydrated_runs_total",
-    help: "Runs hydrated by the EnvChangeRouter's batch-hydrate (one query per column set per wake, shared across all feeds matching the same run — the hot-shared-tag fan-out collapse).",
-    registers: [metricsRegister],
+  const routerHydrates = meter.createCounter("realtime_native.router_hydrated_runs", {
+    description:
+      "Runs hydrated by the EnvChangeRouter's batch-hydrate (one query per column set per wake, shared across all feeds matching the same run).",
   });
 
-  const resolveAdmissionWaits = new Counter({
-    name: "realtime_native_resolve_admission_waits_total",
-    help: "Fresh ClickHouse resolves that had to queue for an admission permit. A rising count means a distinct-filter reconnect stampede is being throttled (the gate is doing its job).",
-    registers: [metricsRegister],
+  const resolveAdmissionWaits = meter.createCounter("realtime_native.resolve_admission_waits", {
+    description:
+      "Fresh ClickHouse resolves that had to queue for an admission permit. A rising count means a distinct-filter reconnect stampede is being throttled (the gate is doing its job).",
   });
 
-  const replays = new Counter({
-    name: "realtime_native_replays_total",
-    help: "Buffered change records replayed to a newly-armed feed (inter-poll gap recovery). 'delivered' = rows reached the feed; 'empty' = candidates hydrated but none survived the filter/diff.",
-    labelNames: ["result"] as const,
-    registers: [metricsRegister],
+  const replays = meter.createCounter("realtime_native.replays", {
+    description:
+      "Buffered change records replayed to a newly-armed feed (inter-poll gap recovery). 'delivered' = rows reached the feed; 'empty' = candidates hydrated but none survived the filter/diff.",
   });
 
-  const deliveryLagMs = new Histogram({
-    name: "realtime_native_delivery_lag_ms",
-    help: "Live emissions: now minus the newest emitted row's updatedAt (PG clock vs app clock, so approximate). The end-to-end delivery SLI — a p99 near the backstop hold means wakes are being missed.",
-    labelNames: ["path"] as const,
-    buckets: [5, 10, 25, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000, 30_000],
-    registers: [metricsRegister],
+  const replayEvictions = meter.createCounter("realtime_native.replay_evictions", {
+    description:
+      "Replay-buffer evictions. 'window' expiry is normal; 'cap' means an env churns more runs inside the window than the buffer holds (replay guarantee degrading — retune the knobs).",
   });
 
-  const emittedRows = new Histogram({
-    name: "realtime_native_emitted_rows",
-    help: "Rows per live emission. Deltas should be small; a fat tail means working-set/offset-floor fallbacks are re-emitting full sets.",
-    buckets: [1, 2, 5, 10, 25, 50, 100, 250, 1_000],
-    registers: [metricsRegister],
+  const deliveryLagMs = meter.createHistogram("realtime_native.delivery_lag_ms", {
+    description:
+      "Live emissions: now minus the newest emitted row's updatedAt (PG clock vs app clock, so approximate). The end-to-end delivery SLI — a p99 near the backstop hold means wakes are being missed.",
+    unit: "ms",
   });
 
-  const backstops = new Counter({
-    name: "realtime_native_backstop_total",
-    help: "Backstop full resolves by outcome. 'empty' is normal idle behavior; sustained 'delivered' means the notify/replay path missed changes — alert on it.",
-    labelNames: ["result"] as const,
-    registers: [metricsRegister],
+  const emittedRows = meter.createHistogram("realtime_native.emitted_rows", {
+    description:
+      "Rows per live emission. Deltas should be small; a fat tail means working-set/offset-floor fallbacks are re-emitting full sets.",
+    unit: "rows",
   });
 
-  const concurrencyRejections = new Counter({
-    name: "realtime_native_concurrency_rejections_total",
-    help: "Polls rejected (429) by the per-env concurrency limiter.",
-    registers: [metricsRegister],
+  const backstops = meter.createCounter("realtime_native.backstops", {
+    description:
+      "Backstop full resolves by outcome. 'empty' is normal idle behavior; sustained 'delivered' means the notify/replay path missed changes — alert on it.",
   });
 
-  const replayEvictions = new Counter({
-    name: "realtime_native_replay_evictions_total",
-    help: "Replay-buffer evictions. 'window' expiry is normal; 'cap' means an env churns more runs inside the window than the buffer holds (replay guarantee degrading — retune the knobs).",
-    labelNames: ["reason"] as const,
-    registers: [metricsRegister],
+  const concurrencyRejections = meter.createCounter("realtime_native.concurrency_rejections", {
+    description: "Polls rejected (429) by the per-env concurrency limiter.",
   });
 
   const limiter = new RealtimeConcurrencyLimiter({
@@ -120,12 +99,12 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
   const router = new EnvChangeRouter({
     source: getRunChangeNotifier(),
     hydrator: runReader,
-    onHydrate: (runCount) => routerHydrates.inc(runCount),
+    onHydrate: (runCount) => routerHydrates.add(runCount),
     replayWindowMs: env.REALTIME_BACKEND_NATIVE_REPLAY_WINDOW_MS,
     replayMaxRunsPerEnv: env.REALTIME_BACKEND_NATIVE_REPLAY_MAX_RUNS,
     unsubscribeLingerMs: env.REALTIME_BACKEND_NATIVE_UNSUBSCRIBE_LINGER_MS,
-    onReplay: (result) => replays.inc({ result }),
-    onReplayEviction: (reason) => replayEvictions.inc({ reason }),
+    onReplay: (result) => replays.add(1, { result }),
+    onReplayEviction: (reason) => replayEvictions.add(1, { reason }),
   });
 
   const client = new NativeRealtimeClient({
@@ -159,58 +138,50 @@ function initializeNativeRealtimeClient(): NativeRealtimeClient {
     runSetCreatedAtBucketMs: env.REALTIME_BACKEND_NATIVE_RUNSET_CREATED_AT_BUCKET_MS,
     holdOnEmpty: env.REALTIME_BACKEND_NATIVE_HOLD_ON_EMPTY === "1",
     resolveAdmissionLimit: env.REALTIME_BACKEND_NATIVE_RESOLVE_ADMISSION_LIMIT,
-    onWakeup: (reason) => wakeups.inc({ reason }),
-    onLivePollPath: (path) => livePollPaths.inc({ path }),
-    onRunSetResolve: (result) => runSetResolves.inc({ result }),
-    onRunSetQuery: (stage, ms) => runSetQueryMs.observe({ stage }, ms),
-    onResolveAdmissionWait: () => resolveAdmissionWaits.inc(),
+    onWakeup: (reason) => wakeups.add(1, { reason }),
+    onLivePollPath: (path) => livePollPaths.add(1, { path }),
+    onRunSetResolve: (result) => runSetResolves.add(1, { result }),
+    onRunSetQuery: (stage, ms) => runSetQueryMs.record(ms, { stage }),
+    onResolveAdmissionWait: () => resolveAdmissionWaits.add(1),
     onEmit: (path, lagMs, rowCount) => {
-      deliveryLagMs.observe({ path }, Math.max(lagMs, 0));
-      emittedRows.observe(rowCount);
-    },
-    onBackstopResult: (result) => backstops.inc({ result }),
-    onConcurrencyRejected: () => concurrencyRejections.inc(),
-  });
-
-  new Gauge({
-    name: "realtime_native_working_set_size",
-    help: "Entries in the per-handle working-set cache (one per active multi-run feed session).",
-    registers: [metricsRegister],
-    collect() {
-      this.set(client.workingSetCacheSize);
-    },
-  });
-
-  new Gauge({
-    name: "realtime_native_resolve_admission_in_use",
-    help: "Fresh ClickHouse resolves currently holding an admission permit (live concurrency against the gate's limit).",
-    registers: [metricsRegister],
-    collect() {
-      this.set(client.resolveAdmissionInUse);
+      deliveryLagMs.record(Math.max(lagMs, 0), { path });
+      emittedRows.record(rowCount);
     },
-  });
-
-  new Gauge({
-    name: "realtime_native_held_feeds",
-    help: "Long-polls currently held, by feed kind — the system's capacity unit.",
-    labelNames: ["kind"] as const,
-    registers: [metricsRegister],
-    collect() {
+    onBackstopResult: (result) => backstops.add(1, { result }),
+    onConcurrencyRejected: () => concurrencyRejections.add(1),
+  });
+
+  meter
+    .createObservableGauge("realtime_native.working_set_size", {
+      description:
+        "Entries in the per-handle working-set cache (one per active multi-run feed session).",
+    })
+    .addCallback((result) => result.observe(client.workingSetCacheSize));
+
+  meter
+    .createObservableGauge("realtime_native.resolve_admission_in_use", {
+      description:
+        "Fresh ClickHouse resolves currently holding an admission permit (live concurrency against the gate's limit).",
+    })
+    .addCallback((result) => result.observe(client.resolveAdmissionInUse));
+
+  meter
+    .createObservableGauge("realtime_native.held_feeds", {
+      description: "Long-polls currently held, by feed kind — the system's capacity unit.",
+    })
+    .addCallback((result) => {
       const counts = router.heldFeedCounts;
-      this.set({ kind: "run" }, counts.run);
-      this.set({ kind: "tag" }, counts.tag);
-      this.set({ kind: "batch" }, counts.batch);
-    },
-  });
-
-  new Gauge({
-    name: "realtime_native_active_envs",
-    help: "Environments currently routed on this instance (held feeds + lingering subscriptions).",
-    registers: [metricsRegister],
-    collect() {
-      this.set(router.activeEnvCount);
-    },
-  });
+      result.observe(counts.run, { kind: "run" });
+      result.observe(counts.tag, { kind: "tag" });
+      result.observe(counts.batch, { kind: "batch" });
+    });
+
+  meter
+    .createObservableGauge("realtime_native.active_envs", {
+      description:
+        "Environments currently routed on this instance (held feeds + lingering subscriptions).",
+    })
+    .addCallback((result) => result.observe(router.activeEnvCount));
 
   return client;
 }

apps/webapp/app/services/realtime/runChangeNotifierInstance.server.ts

@@ -1,6 +1,5 @@
-import { Counter, Gauge } from "prom-client";
+import { getMeter } from "@internal/tracing";
 import { env } from "~/env.server";
-import { metricsRegister } from "~/metrics.server";
 import { singleton } from "~/utils/singleton";
 import { RunChangeNotifier, type ChangeRecordInput } from "./runChangeNotifier.server";
 
@@ -16,23 +15,20 @@ function initializeRunChangeNotifier(): RunChangeNotifier {
   // broadcast every message to every node, so this is what actually shards load.
   const shardedPubSub = clusterMode && env.REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_SHARDED_ENABLED === "1";
 
-  const publishes = new Counter({
-    name: "realtime_run_change_notifier_publishes_total",
-    help: "Change-record publishes by outcome. Failures are the leading indicator that feeds are degrading to their backstops (pub/sub Redis trouble).",
-    labelNames: ["result"] as const,
-    registers: [metricsRegister],
+  const meter = getMeter("realtime-notifier");
+
+  const publishes = meter.createCounter("realtime_notifier.publishes", {
+    description:
+      "Change-record publishes by outcome. Failures are the leading indicator that feeds are degrading to their backstops (pub/sub Redis trouble).",
   });
 
-  const received = new Counter({
-    name: "realtime_run_change_notifier_messages_received_total",
-    help: "Raw channel messages received by this instance's subscriber, pre-coalesce.",
-    registers: [metricsRegister],
+  const received = meter.createCounter("realtime_notifier.messages_received", {
+    description: "Raw channel messages received by this instance's subscriber, pre-coalesce.",
   });
 
-  const delivered = new Counter({
-    name: "realtime_run_change_notifier_batches_delivered_total",
-    help: "Coalesced batches delivered to listeners. received/batches = the coalesce ratio (how hard a busy env is being collapsed).",
-    registers: [metricsRegister],
+  const delivered = meter.createCounter("realtime_notifier.batches_delivered", {
+    description:
+      "Coalesced batches delivered to listeners. received/batches = the coalesce ratio (how hard a busy env is being collapsed).",
   });
 
   const notifier = new RunChangeNotifier({
@@ -48,19 +44,16 @@ function initializeRunChangeNotifier(): RunChangeNotifier {
     },
     envWakeCoalesceWindowMs: env.REALTIME_BACKEND_NATIVE_ENV_WAKE_COALESCE_WINDOW_MS,
     shardedPubSub,
-    onPublishResult: (ok) => publishes.inc({ result: ok ? "ok" : "error" }),
-    onMessageReceived: () => received.inc(),
-    onBatchDelivered: () => delivered.inc(),
+    onPublishResult: (ok) => publishes.add(1, { result: ok ? "ok" : "error" }),
+    onMessageReceived: () => received.add(1),
+    onBatchDelivered: () => delivered.add(1),
   });
 
-  new Gauge({
-    name: "realtime_run_change_notifier_active_subscriptions",
-    help: "Distinct runs currently subscribed for realtime change notifications",
-    collect() {
-      this.set(notifier.activeSubscriptionCount);
-    },
-    registers: [metricsRegister],
-  });
+  meter
+    .createObservableGauge("realtime_notifier.active_subscriptions", {
+      description: "Distinct env channels currently subscribed for realtime change notifications.",
+    })
+    .addCallback((result) => result.observe(notifier.activeSubscriptionCount));
 
   return notifier;
 }

apps/webapp/app/services/realtime/shadowRealtimeClientInstance.server.ts

@@ -1,7 +1,6 @@
-import { Counter } from "prom-client";
+import { getMeter } from "@internal/tracing";
 import { $replica } from "~/db.server";
 import { env } from "~/env.server";
-import { metricsRegister } from "~/metrics.server";
 import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server";
 import { singleton } from "~/utils/singleton";
 import { realtimeClient } from "../realtimeClientGlobal.server";
@@ -15,11 +14,9 @@ import { ShadowRealtimeClient } from "./shadowRealtimeClient.server";
  * when an org's `realtimeBackend` flag is set to "shadow".
  */
 function initializeShadowRealtimeClient(): ShadowRealtimeClient {
-  const compares = new Counter({
-    name: "realtime_shadow_compare_total",
-    help: "Dual-run shadow-compare outcomes (Electric vs native). kind=serialization|membership, result=match|diverge|skew.",
-    labelNames: ["feed", "kind", "result"] as const,
-    registers: [metricsRegister],
+  const compares = getMeter("realtime-shadow").createCounter("realtime_shadow.compares", {
+    description:
+      "Dual-run shadow-compare outcomes (Electric vs native). kind=serialization|membership, result=match|diverge|skew.",
   });
 
   const comparator = new RealtimeShadowComparator({
@@ -39,19 +36,20 @@ function initializeShadowRealtimeClient(): ShadowRealtimeClient {
     onOutcome: (outcome) => {
       const { feed } = outcome;
       if (outcome.serializationMatched) {
-        compares.inc({ feed, kind: "serialization", result: "match" }, outcome.serializationMatched);
+        compares.add(outcome.serializationMatched, { feed, kind: "serialization", result: "match" });
       }
       if (outcome.serializationDiverged) {
-        compares.inc(
-          { feed, kind: "serialization", result: "diverge" },
-          outcome.serializationDiverged
-        );
+        compares.add(outcome.serializationDiverged, {
+          feed,
+          kind: "serialization",
+          result: "diverge",
+        });
       }
       if (outcome.serializationSkew) {
-        compares.inc({ feed, kind: "serialization", result: "skew" }, outcome.serializationSkew);
+        compares.add(outcome.serializationSkew, { feed, kind: "serialization", result: "skew" });
       }
       if (outcome.membershipMatch !== undefined) {
-        compares.inc({
+        compares.add(1, {
           feed,
           kind: "membership",
           result: outcome.membershipMatch ? "match" : "diverge",