report worker queue lengths via opentelemetry metrics

Author: ericallam

apps/webapp/app/env.server.ts

@@ -328,6 +328,10 @@ const EnvironmentSchema = z.object({
   INTERNAL_OTEL_TRACE_DISABLED: z.string().default("0"),
 
   INTERNAL_OTEL_LOG_EXPORTER_URL: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_URL: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_AUTH_HEADERS: z.string().optional(),
+  INTERNAL_OTEL_METRIC_EXPORTER_DISABLED: z.string().default("0"),
+  INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL: z.coerce.number().int().default(30_000),
 
   ORG_SLACK_INTEGRATION_CLIENT_ID: z.string().optional(),
   ORG_SLACK_INTEGRATION_CLIENT_SECRET: z.string().optional(),
@@ -621,6 +625,7 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_RELEASE_CONCURRENCY_BATCH_SIZE: z.coerce.number().int().default(10),
 
   RUN_ENGINE_WORKER_ENABLED: z.string().default("1"),
+  RUN_ENGINE_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
 
   /** How long should the presence ttl last */
   DEV_PRESENCE_SSE_TIMEOUT: z.coerce.number().int().default(30_000),

apps/webapp/app/v3/runEngine.server.ts

@@ -4,7 +4,7 @@ import { prisma } from "~/db.server";
 import { env } from "~/env.server";
 import { singleton } from "~/utils/singleton";
 import { allMachines } from "./machinePresets.server";
-import { tracer } from "./tracer.server";
+import { tracer, meter } from "./tracer.server";
 
 export const engine = singleton("RunEngine", createRunEngine);
 
@@ -13,6 +13,7 @@ export type { RunEngine };
 function createRunEngine() {
   const engine = new RunEngine({
     prisma,
+    logLevel: env.RUN_ENGINE_WORKER_LOG_LEVEL,
     worker: {
       disabled: env.RUN_ENGINE_WORKER_ENABLED === "0",
       workers: env.RUN_ENGINE_WORKER_COUNT,
@@ -71,6 +72,7 @@ function createRunEngine() {
       },
     },
     tracer,
+    meter,
     heartbeatTimeoutsMs: {
       PENDING_EXECUTING: env.RUN_ENGINE_TIMEOUT_PENDING_EXECUTING,
       PENDING_CANCEL: env.RUN_ENGINE_TIMEOUT_PENDING_CANCEL,

apps/webapp/app/v3/tracer.server.ts

@@ -11,6 +11,8 @@ import {
   Tracer,
   diag,
   trace,
+  metrics,
+  Meter,
 } from "@opentelemetry/api";
 import { logs, SeverityNumber } from "@opentelemetry/api-logs";
 import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-http";
@@ -19,6 +21,12 @@ import { BatchLogRecordProcessor, LoggerProvider } from "@opentelemetry/sdk-logs
 import { type Instrumentation, registerInstrumentations } from "@opentelemetry/instrumentation";
 import { ExpressInstrumentation } from "@opentelemetry/instrumentation-express";
 import { HttpInstrumentation } from "@opentelemetry/instrumentation-http";
+import {
+  MeterProvider,
+  ConsoleMetricExporter,
+  PeriodicExportingMetricReader,
+} from "@opentelemetry/sdk-metrics";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
 import { Resource } from "@opentelemetry/resources";
 import {
   BatchSpanProcessor,
@@ -30,17 +38,24 @@ import {
   TraceIdRatioBasedSampler,
 } from "@opentelemetry/sdk-trace-base";
 import { NodeTracerProvider } from "@opentelemetry/sdk-trace-node";
-import { SEMRESATTRS_SERVICE_NAME } from "@opentelemetry/semantic-conventions";
+import {
+  SEMRESATTRS_SERVICE_INSTANCE_ID,
+  SEMRESATTRS_SERVICE_NAME,
+} from "@opentelemetry/semantic-conventions";
 import { PrismaInstrumentation } from "@prisma/instrumentation";
 import { env } from "~/env.server";
 import { AuthenticatedEnvironment } from "~/services/apiAuth.server";
 import { singleton } from "~/utils/singleton";
 import { LoggerSpanExporter } from "./telemetry/loggerExporter.server";
 import { logger } from "~/services/logger.server";
 import { flattenAttributes } from "@trigger.dev/core/v3";
+import { randomUUID } from "node:crypto";
+import { prisma } from "~/db.server";
 
 export const SEMINTATTRS_FORCE_RECORDING = "forceRecording";
 
+const SERVICE_INSTANCE_ID = randomUUID();
+
 class CustomWebappSampler implements Sampler {
   constructor(private readonly _baseSampler: Sampler) {}
 
@@ -83,7 +98,12 @@ class CustomWebappSampler implements Sampler {
   }
 }
 
-export const { tracer, logger: otelLogger, provider } = singleton("tracer", getTracer);
+export const {
+  tracer,
+  logger: otelLogger,
+  provider,
+  meter,
+} = singleton("opentelemetry", setupTelemetry);
 
 export async function startActiveSpan<T>(
   name: string,
@@ -148,14 +168,15 @@ export async function emitWarnLog(message: string, params: Record<string, unknow
   });
 }
 
-function getTracer() {
+function setupTelemetry() {
   if (env.INTERNAL_OTEL_TRACE_DISABLED === "1") {
     console.log(`🔦 Tracer disabled, returning a noop tracer`);
 
     return {
       tracer: trace.getTracer("trigger.dev", "3.3.12"),
       logger: logs.getLogger("trigger.dev", "3.3.12"),
       provider: new NodeTracerProvider(),
+      meter: setupMetrics(),
     };
   }
 
@@ -167,6 +188,7 @@ function getTracer() {
     forceFlushTimeoutMillis: 15_000,
     resource: new Resource({
       [SEMRESATTRS_SERVICE_NAME]: env.SERVICE_NAME,
+      [SEMRESATTRS_SERVICE_INSTANCE_ID]: SERVICE_INSTANCE_ID,
     }),
     sampler: new ParentBasedSampler({
       root: new CustomWebappSampler(new TraceIdRatioBasedSampler(samplingRate)),
@@ -261,10 +283,90 @@ function getTracer() {
   return {
     tracer: provider.getTracer("trigger.dev", "3.3.12"),
     logger: logs.getLogger("trigger.dev", "3.3.12"),
+    meter: setupMetrics(),
     provider,
   };
 }
 
+function setupMetrics() {
+  if (env.INTERNAL_OTEL_METRIC_EXPORTER_DISABLED === "1") {
+    return metrics.getMeter("trigger.dev", "3.3.12");
+  }
+
+  const exporter = env.INTERNAL_OTEL_METRIC_EXPORTER_URL
+    ? new OTLPMetricExporter({
+        url: env.INTERNAL_OTEL_METRIC_EXPORTER_URL,
+        timeoutMillis: 30_000,
+        headers: parseInternalMetricsHeaders() ?? {},
+      })
+    : new ConsoleMetricExporter();
+
+  const meterProvider = new MeterProvider({
+    resource: new Resource({
+      [SEMRESATTRS_SERVICE_NAME]: env.SERVICE_NAME,
+      [SEMRESATTRS_SERVICE_INSTANCE_ID]: SERVICE_INSTANCE_ID,
+    }),
+    readers: [
+      new PeriodicExportingMetricReader({
+        exporter,
+        exportIntervalMillis: env.INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL,
+        exportTimeoutMillis: 30_000,
+      }),
+    ],
+  });
+
+  metrics.setGlobalMeterProvider(meterProvider);
+
+  const meter = meterProvider.getMeter("trigger.dev", "3.3.12");
+
+  configurePrismaMetrics(meter);
+
+  return meter;
+}
+
+function configurePrismaMetrics(meter: Meter) {
+  const totalGauge = meter.createObservableGauge("db.pool.connections.total", {
+    description: "Open Prisma-pool connections",
+    unit: "connections",
+  });
+  const busyGauge = meter.createObservableGauge("db.pool.connections.busy", {
+    description: "Connections currently executing queries",
+    unit: "connections",
+  });
+  const freeGauge = meter.createObservableGauge("db.pool.connections.free", {
+    description: "Idle (free) connections in the pool",
+    unit: "connections",
+  });
+
+  // Single helper so we hit Prisma only once per scrape ---------------------
+  async function readPoolCounters() {
+    const { gauges } = await prisma.$metrics.json();
+
+    const busy = gauges.find((g) => g.key === "prisma_pool_connections_busy")?.value ?? 0;
+    const free = gauges.find((g) => g.key === "prisma_pool_connections_idle")?.value ?? 0;
+    const total =
+      gauges.find((g) => g.key === "prisma_pool_connections_open")?.value ?? busy + free; // fallback compute
+
+    return { total, busy, free };
+  }
+
+  // Register callbacks (one scrape == one DB call) --------------------------
+  totalGauge.addCallback(async (res) => {
+    const { total } = await readPoolCounters();
+    res.observe(total);
+  });
+
+  busyGauge.addCallback(async (res) => {
+    const { busy } = await readPoolCounters();
+    res.observe(busy);
+  });
+
+  freeGauge.addCallback(async (res) => {
+    const { free } = await readPoolCounters();
+    res.observe(free);
+  });
+}
+
 const SemanticEnvResources = {
   ENV_ID: "$trigger.env.id",
   ENV_TYPE: "$trigger.env.type",
@@ -300,3 +402,13 @@ function parseInternalTraceHeaders(): Record<string, string> | undefined {
     return;
   }
 }
+
+function parseInternalMetricsHeaders(): Record<string, string> | undefined {
+  try {
+    return env.INTERNAL_OTEL_METRIC_EXPORTER_AUTH_HEADERS
+      ? (JSON.parse(env.INTERNAL_OTEL_METRIC_EXPORTER_AUTH_HEADERS) as Record<string, string>)
+      : undefined;
+  } catch {
+    return;
+  }
+}

apps/webapp/package.json

@@ -61,12 +61,14 @@
     "@opentelemetry/core": "1.25.1",
     "@opentelemetry/exporter-logs-otlp-http": "0.52.1",
     "@opentelemetry/exporter-trace-otlp-http": "0.52.1",
+    "@opentelemetry/exporter-metrics-otlp-http": "0.52.1",
     "@opentelemetry/instrumentation": "0.52.1",
     "@opentelemetry/instrumentation-express": "^0.36.1",
     "@opentelemetry/instrumentation-http": "0.52.1",
     "@opentelemetry/resources": "1.25.1",
     "@opentelemetry/sdk-logs": "0.52.1",
     "@opentelemetry/sdk-node": "0.52.1",
+    "@opentelemetry/sdk-metrics": "1.25.1",
     "@opentelemetry/sdk-trace-base": "1.25.1",
     "@opentelemetry/sdk-trace-node": "1.25.1",
     "@opentelemetry/semantic-conventions": "1.25.1",

internal-packages/run-engine/src/engine/index.ts

@@ -1,5 +1,5 @@
 import { createRedisClient, Redis } from "@internal/redis";
-import { startSpan, trace, Tracer } from "@internal/tracing";
+import { getMeter, Meter, startSpan, trace, Tracer } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import {
   CheckpointInput,
@@ -55,8 +55,9 @@ export class RunEngine {
   private runLockRedis: Redis;
   private runLock: RunLocker;
   private worker: EngineWorker;
-  private logger = new Logger("RunEngine", "debug");
+  private logger: Logger;
   private tracer: Tracer;
+  private meter: Meter;
   private heartbeatTimeouts: HeartbeatTimeouts;
 
   prisma: PrismaClient;
@@ -76,6 +77,7 @@ export class RunEngine {
   raceSimulationSystem: RaceSimulationSystem = new RaceSimulationSystem();
 
   constructor(private readonly options: RunEngineOptions) {
+    this.logger = options.logger ?? new Logger("RunEngine", this.options.logLevel ?? "info");
     this.prisma = options.prisma;
     this.runLockRedis = createRedisClient(
       {
@@ -109,7 +111,7 @@ export class RunEngine {
         defaultEnvConcurrencyLimit: options.queue?.defaultEnvConcurrency ?? 10,
       }),
       defaultEnvConcurrency: options.queue?.defaultEnvConcurrency ?? 10,
-      logger: new Logger("RunQueue", "debug"),
+      logger: new Logger("RunQueue", this.options.logLevel ?? "info"),
       redis: { ...options.queue.redis, keyPrefix: `${options.queue.redis.keyPrefix}runqueue:` },
       retryOptions: options.queue?.retryOptions,
       workerOptions: {
@@ -121,6 +123,7 @@ export class RunEngine {
       },
       masterQueueConsumersDisabled: options.queue?.masterQueueConsumersDisabled,
       processWorkerQueueDebounceMs: options.queue?.processWorkerQueueDebounceMs,
+      meter: options.meter,
     });
 
     this.worker = new Worker({
@@ -184,6 +187,7 @@ export class RunEngine {
     }
 
     this.tracer = options.tracer;
+    this.meter = options.meter ?? getMeter("run-engine");
 
     const defaultHeartbeatTimeouts: HeartbeatTimeouts = {
       PENDING_EXECUTING: 60_000,
@@ -203,6 +207,7 @@ export class RunEngine {
       eventBus: this.eventBus,
       logger: this.logger,
       tracer: this.tracer,
+      meter: this.meter,
       runLock: this.runLock,
       runQueue: this.runQueue,
       raceSimulationSystem: this.raceSimulationSystem,
@@ -566,6 +571,9 @@ export class RunEngine {
     runnerId?: string;
     tx?: PrismaClientOrTransaction;
   }): Promise<DequeuedMessage[]> {
+    // We only do this with "prod" worker queues because we don't want to observe dev (e.g. environment) worker queues
+    this.runQueue.registerObservableWorkerQueue(workerQueue);
+
     const dequeuedMessage = await this.dequeueSystem.dequeueFromWorkerQueue({
       consumerId,
       workerQueue,

internal-packages/run-engine/src/engine/systems/systems.ts

@@ -1,4 +1,4 @@
-import { Tracer } from "@internal/tracing";
+import { Meter, Tracer } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { PrismaClient } from "@trigger.dev/database";
 import { RunQueue } from "../../run-queue/index.js";
@@ -13,6 +13,7 @@ export type SystemResources = {
   eventBus: EventBus;
   logger: Logger;
   tracer: Tracer;
+  meter: Meter;
   runLock: RunLocker;
   runQueue: RunQueue;
   raceSimulationSystem: RaceSimulationSystem;

internal-packages/run-engine/src/engine/types.ts

@@ -1,6 +1,6 @@
 import { type RedisOptions } from "@internal/redis";
 import { Worker, type WorkerConcurrencyOptions } from "@trigger.dev/redis-worker";
-import { Tracer } from "@internal/tracing";
+import { Meter, Tracer } from "@internal/tracing";
 import {
   MachinePreset,
   MachinePresetName,
@@ -12,6 +12,7 @@ import { PrismaClient } from "@trigger.dev/database";
 import { FairQueueSelectionStrategyOptions } from "../run-queue/fairQueueSelectionStrategy.js";
 import { MinimalAuthenticatedEnvironment } from "../shared/index.js";
 import { workerCatalog } from "./workerCatalog.js";
+import { Logger, LogLevel } from "@trigger.dev/core/logger";
 
 export type RunEngineOptions = {
   prisma: PrismaClient;
@@ -47,6 +48,9 @@ export type RunEngineOptions = {
   heartbeatTimeoutsMs?: Partial<HeartbeatTimeouts>;
   queueRunsWaitingForWorkerBatchSize?: number;
   tracer: Tracer;
+  meter?: Meter;
+  logger?: Logger;
+  logLevel?: LogLevel;
   releaseConcurrency?: {
     disabled?: boolean;
     maxTokensRatio?: number;