add prom metrics to redis worker

Author: nicktrn

packages/redis-worker/package.json

@@ -38,21 +38,22 @@
     "test": "vitest --sequence.concurrent=false --no-file-parallelism"
   },
   "dependencies": {
-    "@internal/tracing": "workspace:*",
     "@internal/redis": "workspace:*",
+    "@internal/tracing": "workspace:*",
     "@trigger.dev/core": "workspace:*",
     "lodash.omit": "^4.5.0",
     "nanoid": "^5.0.7",
     "p-limit": "^6.2.0",
+    "prom-client": "^15.1.0",
     "zod": "3.23.8"
   },
   "devDependencies": {
     "@internal/testcontainers": "workspace:*",
     "@types/lodash.omit": "^4.5.7",
-    "vitest": "^1.4.0",
     "rimraf": "6.0.1",
     "tshy": "^3.0.2",
-    "tsx": "4.17.0"
+    "tsx": "4.17.0",
+    "vitest": "^1.4.0"
   },
   "engines": {
     "node": ">=18.20.0"

packages/redis-worker/src/worker.ts

@@ -9,6 +9,7 @@ import { nanoid } from "nanoid";
 import pLimit from "p-limit";
 import { createRedisClient } from "@internal/redis";
 import { shutdownManager } from "@trigger.dev/core/v3/serverOnly";
+import { Registry, Histogram } from "prom-client";
 
 export type WorkerCatalog = {
   [key: string]: {
@@ -48,6 +49,9 @@ type WorkerOptions<TCatalog extends WorkerCatalog> = {
   shutdownTimeoutMs?: number;
   logger?: Logger;
   tracer?: Tracer;
+  metrics?: {
+    register: Registry;
+  };
 };
 
 // This results in attempt 12 being a delay of 1 hour
@@ -65,6 +69,16 @@ class Worker<TCatalog extends WorkerCatalog> {
   private subscriber: Redis | undefined;
   private tracer: Tracer;
 
+  private metrics: {
+    register?: Registry;
+    enqueueDuration?: Histogram;
+    dequeueDuration?: Histogram;
+    jobDuration?: Histogram;
+    ackDuration?: Histogram;
+    redriveDuration?: Histogram;
+    rescheduleDuration?: Histogram;
+  } = {};
+
   queue: SimpleQueue<QueueCatalogFromWorkerCatalog<TCatalog>>;
   private jobs: WorkerOptions<TCatalog>["jobs"];
   private logger: Logger;
@@ -100,6 +114,61 @@ class Worker<TCatalog extends WorkerCatalog> {
 
     // Create a p-limit instance using this limit.
     this.limiter = pLimit(this.concurrency.limit);
+
+    this.metrics.register = options.metrics?.register;
+
+    if (!this.metrics.register) {
+      return;
+    }
+
+    this.metrics.enqueueDuration = new Histogram({
+      name: "redis_worker_enqueue_duration_seconds",
+      help: "The duration of enqueue operations",
+      labelNames: ["worker_name", "job_type", "has_available_at"],
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
+      registers: [this.metrics.register],
+    });
+
+    this.metrics.dequeueDuration = new Histogram({
+      name: "redis_worker_dequeue_duration_seconds",
+      help: "The duration of dequeue operations",
+      labelNames: ["worker_name", "worker_id", "task_count"],
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
+      registers: [this.metrics.register],
+    });
+
+    this.metrics.jobDuration = new Histogram({
+      name: "redis_worker_job_duration_seconds",
+      help: "The duration of job operations",
+      labelNames: ["worker_name", "worker_id", "batch_size", "job_type", "attempt"],
+      // use different buckets here as jobs can take a while to run
+      buckets: [0.1, 0.25, 0.5, 1, 2.5, 5, 10, 20, 30, 45, 60],
+      registers: [this.metrics.register],
+    });
+
+    this.metrics.ackDuration = new Histogram({
+      name: "redis_worker_ack_duration_seconds",
+      help: "The duration of ack operations",
+      labelNames: ["worker_name"],
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
+      registers: [this.metrics.register],
+    });
+
+    this.metrics.redriveDuration = new Histogram({
+      name: "redis_worker_redrive_duration_seconds",
+      help: "The duration of redrive operations",
+      labelNames: ["worker_name"],
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
+      registers: [this.metrics.register],
+    });
+
+    this.metrics.rescheduleDuration = new Histogram({
+      name: "redis_worker_reschedule_duration_seconds",
+      help: "The duration of reschedule operations",
+      labelNames: ["worker_name"],
+      buckets: [0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1],
+      registers: [this.metrics.register],
+    });
   }
 
   public start() {
@@ -160,18 +229,25 @@ class Worker<TCatalog extends WorkerCatalog> {
 
         span.setAttribute("job_visibility_timeout_ms", timeout);
 
-        return this.queue.enqueue({
-          id,
-          job,
-          item: payload,
-          visibilityTimeoutMs: timeout,
-          availableAt,
-        });
+        return this.withHistogram(
+          this.metrics.enqueueDuration,
+          this.queue.enqueue({
+            id,
+            job,
+            item: payload,
+            visibilityTimeoutMs: timeout,
+            availableAt,
+          }),
+          {
+            job_type: String(job),
+            has_available_at: availableAt ? "true" : "false",
+          }
+        );
       },
       {
         kind: SpanKind.PRODUCER,
         attributes: {
-          job_type: job as string,
+          job_type: String(job),
           job_id: id,
         },
       }
@@ -187,7 +263,10 @@ class Worker<TCatalog extends WorkerCatalog> {
       this.tracer,
       "reschedule",
       async (span) => {
-        return this.queue.reschedule(id, availableAt);
+        return this.withHistogram(
+          this.metrics.rescheduleDuration,
+          this.queue.reschedule(id, availableAt)
+        );
       },
       {
         kind: SpanKind.PRODUCER,
@@ -203,7 +282,7 @@ class Worker<TCatalog extends WorkerCatalog> {
       this.tracer,
       "ack",
       () => {
-        return this.queue.ack(id);
+        return this.withHistogram(this.metrics.ackDuration, this.queue.ack(id));
       },
       {
         attributes: {
@@ -229,7 +308,14 @@ class Worker<TCatalog extends WorkerCatalog> {
       }
 
       try {
-        const items = await this.queue.dequeue(taskCount);
+        const items = await this.withHistogram(
+          this.metrics.dequeueDuration,
+          this.queue.dequeue(taskCount),
+          {
+            worker_id: workerId,
+            task_count: taskCount,
+          }
+        );
 
         if (items.length === 0) {
           await Worker.delay(pollIntervalMs);
@@ -274,7 +360,17 @@ class Worker<TCatalog extends WorkerCatalog> {
       this.tracer,
       "processItem",
       async () => {
-        await handler({ id, payload: item, visibilityTimeoutMs, attempt });
+        await this.withHistogram(
+          this.metrics.jobDuration,
+          handler({ id, payload: item, visibilityTimeoutMs, attempt }),
+          {
+            worker_id: workerId,
+            batch_size: batchSize,
+            job_type: job,
+            attempt,
+          }
+        );
+
         // On success, acknowledge the item.
         await this.queue.ack(id);
       },
@@ -363,6 +459,23 @@ class Worker<TCatalog extends WorkerCatalog> {
     });
   }
 
+  private async withHistogram<T>(
+    histogram: Histogram<string> | undefined,
+    promise: Promise<T>,
+    labels?: Record<string, string | number>
+  ): Promise<T> {
+    if (!histogram || !this.metrics.register) {
+      return promise;
+    }
+
+    const end = histogram.startTimer({ worker_name: this.options.name, ...labels });
+    try {
+      return await promise;
+    } finally {
+      end();
+    }
+  }
+
   // A simple helper to delay for a given number of milliseconds.
   private static delay(ms: number): Promise<void> {
     return new Promise((resolve) => setTimeout(resolve, ms));
@@ -387,7 +500,10 @@ class Worker<TCatalog extends WorkerCatalog> {
       if (typeof id !== "string") {
         throw new Error("Invalid message format: id must be a string");
       }
-      await this.queue.redriveFromDeadLetterQueue(id);
+      await this.withHistogram(
+        this.metrics.redriveDuration,
+        this.queue.redriveFromDeadLetterQueue(id)
+      );
       this.logger.log(`Redrived item ${id} from Dead Letter Queue`);
     } catch (error) {
       this.logger.error("Error processing redrive message", { error, message });