metrics now working, configure the run queue settings, additional metrics for run engine and redis-worker

Author: ericallam

apps/webapp/app/env.server.ts

@@ -330,8 +330,8 @@ const EnvironmentSchema = z.object({
   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),
+  INTERNAL_OTEL_METRIC_EXPORTER_ENABLED: z.string().default("0"),
+  INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS: z.coerce.number().int().default(30_000),
 
   ORG_SLACK_INTEGRATION_CLIENT_ID: z.string().optional(),
   ORG_SLACK_INTEGRATION_CLIENT_SECRET: z.string().optional(),
@@ -468,8 +468,10 @@ const EnvironmentSchema = z.object({
   RUN_ENGINE_QUEUE_AGE_RANDOMIZATION_BIAS: z.coerce.number().default(0.25),
   RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
   RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
+  RUN_ENGINE_RUN_QUEUE_SHARD_COUNT: z.coerce.number().int().default(4),
   RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
   RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS: z.coerce.number().int().default(30_000),
+  RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
 
   RUN_ENGINE_WORKER_REDIS_HOST: z
     .string()

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

@@ -59,6 +59,8 @@ function createRunEngine() {
         maximumEnvCount: env.RUN_ENGINE_MAXIMUM_ENV_COUNT,
         tracer,
       },
+      shardCount: env.RUN_ENGINE_RUN_QUEUE_SHARD_COUNT,
+      processWorkerQueueDebounceMs: env.RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS,
     },
     runLock: {
       redis: {

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

@@ -26,7 +26,7 @@ import {
   ConsoleMetricExporter,
   PeriodicExportingMetricReader,
 } from "@opentelemetry/sdk-metrics";
-import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-http";
+import { OTLPMetricExporter } from "@opentelemetry/exporter-metrics-otlp-proto";
 import { Resource } from "@opentelemetry/resources";
 import {
   BatchSpanProcessor,
@@ -289,17 +289,11 @@ function setupTelemetry() {
 }
 
 function setupMetrics() {
-  if (env.INTERNAL_OTEL_METRIC_EXPORTER_DISABLED === "1") {
+  if (env.INTERNAL_OTEL_METRIC_EXPORTER_ENABLED === "0") {
     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 exporter = createMetricsExporter();
 
   const meterProvider = new MeterProvider({
     resource: new Resource({
@@ -309,8 +303,8 @@ function setupMetrics() {
     readers: [
       new PeriodicExportingMetricReader({
         exporter,
-        exportIntervalMillis: env.INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL,
-        exportTimeoutMillis: 30_000,
+        exportIntervalMillis: env.INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS,
+        exportTimeoutMillis: env.INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS,
       }),
     ],
   });
@@ -412,3 +406,23 @@ function parseInternalMetricsHeaders(): Record<string, string> | undefined {
     return;
   }
 }
+
+function createMetricsExporter() {
+  if (env.INTERNAL_OTEL_METRIC_EXPORTER_URL) {
+    const headers = parseInternalMetricsHeaders() ?? {};
+
+    console.log(
+      `🔦 Tracer: OTLP metric exporter enabled to ${
+        env.INTERNAL_OTEL_METRIC_EXPORTER_URL
+      } with headers: ${Object.keys(headers)}`
+    );
+
+    return new OTLPMetricExporter({
+      url: env.INTERNAL_OTEL_METRIC_EXPORTER_URL,
+      timeoutMillis: 30_000,
+      headers,
+    });
+  } else {
+    return new ConsoleMetricExporter();
+  }
+}

apps/webapp/package.json

@@ -61,7 +61,7 @@
     "@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/exporter-metrics-otlp-proto": "0.52.1",
     "@opentelemetry/instrumentation": "0.52.1",
     "@opentelemetry/instrumentation-express": "^0.36.1",
     "@opentelemetry/instrumentation-http": "0.52.1",

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

@@ -122,13 +122,14 @@ export class RunEngine {
         immediatePollIntervalMs: options.worker.immediatePollIntervalMs,
         shutdownTimeoutMs: options.worker.shutdownTimeoutMs,
       },
-      masterQueueConsumersDisabled: options.queue?.masterQueueConsumersDisabled,
+      shardCount: options.queue?.shardCount,
+      masterQueueConsumersDisabled: options.worker.disabled,
       processWorkerQueueDebounceMs: options.queue?.processWorkerQueueDebounceMs,
       meter: options.meter,
     });
 
     this.worker = new Worker({
-      name: "worker",
+      name: "run-engine-worker",
       redisOptions: {
         ...options.worker.redis,
         keyPrefix: `${options.worker.redis.keyPrefix}worker:`,

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

@@ -5,7 +5,16 @@ import { Redis } from "@internal/redis";
 import * as redlock from "redlock";
 import { tryCatch } from "@trigger.dev/core";
 import { Logger } from "@trigger.dev/core/logger";
-import { startSpan, Tracer, Meter, getMeter, ValueType, ObservableResult, Attributes, Histogram } from "@internal/tracing";
+import {
+  startSpan,
+  Tracer,
+  Meter,
+  getMeter,
+  ValueType,
+  ObservableResult,
+  Attributes,
+  Histogram,
+} from "@internal/tracing";
 
 const SemanticAttributes = {
   LOCK_TYPE: "run_engine.lock.type",
@@ -41,23 +50,17 @@ export class RunLocker {
     this.tracer = options.tracer;
     this.meter = options.meter ?? getMeter("run-engine");
 
-    const activeLocksObservableGauge = this.meter.createObservableGauge(
-      "run_engine.locks.active",
-      {
-        description: "The number of active locks by type",
-        unit: "1",
-        valueType: ValueType.INT,
-      }
-    );
+    const activeLocksObservableGauge = this.meter.createObservableGauge("run_engine.locks.active", {
+      description: "The number of active locks by type",
+      unit: "locks",
+      valueType: ValueType.INT,
+    });
 
-    const lockDurationHistogram = this.meter.createHistogram(
-      "run_engine.lock.duration",
-      {
-        description: "The duration of lock operations",
-        unit: "ms",
-        valueType: ValueType.DOUBLE,
-      }
-    );
+    const lockDurationHistogram = this.meter.createHistogram("run_engine.lock.duration", {
+      description: "The duration of lock operations",
+      unit: "ms",
+      valueType: ValueType.DOUBLE,
+    });
 
     activeLocksObservableGauge.addCallback(this.#updateActiveLocksCount.bind(this));
     this.lockDurationHistogram = lockDurationHistogram;
@@ -108,10 +111,10 @@ export class RunLocker {
 
         const [error, result] = await tryCatch(
           this.redlock.using(resources, duration, async (signal) => {
-            const newContext: LockContext = { 
-              resources: joinedResources, 
+            const newContext: LockContext = {
+              resources: joinedResources,
               signal,
-              lockType: name
+              lockType: name,
             };
 
             // Track active lock

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

@@ -1,5 +1,13 @@
 import { Callback, createRedisClient, Redis, Result, type RedisOptions } from "@internal/redis";
-import { startSpan, Tracer, Meter, getMeter, ValueType, ObservableResult, Attributes } from "@internal/tracing";
+import {
+  startSpan,
+  Tracer,
+  Meter,
+  getMeter,
+  ValueType,
+  ObservableResult,
+  Attributes,
+} from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { z } from "zod";
 import { setInterval } from "node:timers/promises";
@@ -100,16 +108,16 @@ export class ReleaseConcurrencyTokenBucketQueue<T> {
       "release_concurrency.releasings.length",
       {
         description: "Number of items in the releasings sorted set",
-        unit: "1",
+        unit: "items",
         valueType: ValueType.INT,
       }
     );
 
     const masterQueueLengthGauge = this.meter.createObservableGauge(
       "release_concurrency.master_queue.length",
       {
-        description: "Number of queues in the master queue sorted set",
-        unit: "1",
+        description: "Number of items in the master queue sorted set",
+        unit: "items",
         valueType: ValueType.INT,
       }
     );

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

@@ -30,7 +30,7 @@ export type RunEngineOptions = {
   };
   queue: {
     redis: RedisOptions;
-    masterQueueConsumersDisabled?: boolean;
+    shardCount?: number;
     processWorkerQueueDebounceMs?: number;
     workerOptions?: WorkerConcurrencyOptions;
     retryOptions?: RetryOptions;

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

@@ -137,7 +137,7 @@ export class RunQueue {
       "runqueue.workerQueue.length",
       {
         description: "The number of messages in the worker queue",
-        unit: "1",
+        unit: "messages",
         valueType: ValueType.INT,
       }
     );
@@ -146,7 +146,7 @@ export class RunQueue {
       "runqueue.masterQueue.length",
       {
         description: "The number of queues in the master queue shard",
-        unit: "1",
+        unit: "queues",
         valueType: ValueType.INT,
       }
     );
@@ -791,8 +791,11 @@ export class RunQueue {
     projectId: string
   ) {
     // Calculate the master queue shard for this environment
-    const masterQueue = this.keys.masterQueueKeyForEnvironment(runtimeEnvironmentId, this.shardCount);
-    
+    const masterQueue = this.keys.masterQueueKeyForEnvironment(
+      runtimeEnvironmentId,
+      this.shardCount
+    );
+
     // Use scanStream to find all matching members
     const stream = this.redis.zscanStream(masterQueue, {
       match: this.keys.queueKey(organizationId, projectId, "*", "*"),

packages/redis-worker/package.json

@@ -27,7 +27,6 @@
     "lodash.omit": "^4.5.0",
     "nanoid": "^5.0.7",
     "p-limit": "^6.2.0",
-    "prom-client": "^15.1.0",
     "zod": "3.23.8"
   },
   "devDependencies": {
@@ -52,4 +51,4 @@
       "require": "./dist/index.cjs"
     }
   }
-}
+}