Adding lock metrics

Author: ericallam

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

@@ -97,6 +97,7 @@ export class RunEngine {
       redis: this.runLockRedis,
       logger: this.logger,
       tracer: trace.getTracer("RunLocker"),
+      meter: options.meter,
     });
 
     const keys = new RunQueueFullKeyProducer();

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

@@ -5,20 +5,30 @@ 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 } from "@internal/tracing";
+import { startSpan, Tracer, Meter, getMeter, ValueType, ObservableResult, Attributes, Histogram } from "@internal/tracing";
+
+const SemanticAttributes = {
+  LOCK_TYPE: "run_engine.lock.type",
+  LOCK_RESOURCES: "run_engine.lock.resources",
+  LOCK_SUCCESS: "run_engine.lock.success",
+};
 
 interface LockContext {
   resources: string;
   signal: redlock.RedlockAbortSignal;
+  lockType: string;
 }
 
 export class RunLocker {
   private redlock: InstanceType<typeof redlock.default>;
   private asyncLocalStorage: AsyncLocalStorage<LockContext>;
   private logger: Logger;
   private tracer: Tracer;
+  private meter: Meter;
+  private activeLocks: Map<string, { lockType: string; resources: string[] }> = new Map();
+  private lockDurationHistogram: Histogram;
 
-  constructor(options: { redis: Redis; logger: Logger; tracer: Tracer }) {
+  constructor(options: { redis: Redis; logger: Logger; tracer: Tracer; meter?: Meter }) {
     this.redlock = new Redlock([options.redis], {
       driftFactor: 0.01,
       retryCount: 10,
@@ -29,6 +39,45 @@ export class RunLocker {
     this.asyncLocalStorage = new AsyncLocalStorage<LockContext>();
     this.logger = options.logger;
     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 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;
+  }
+
+  async #updateActiveLocksCount(observableResult: ObservableResult<Attributes>) {
+    // Group active locks by type
+    const lockCountsByType = new Map<string, number>();
+
+    for (const [_, lockInfo] of this.activeLocks) {
+      const count = lockCountsByType.get(lockInfo.lockType) || 0;
+      lockCountsByType.set(lockInfo.lockType, count + 1);
+    }
+
+    // Report metrics for each lock type
+    for (const [lockType, count] of lockCountsByType) {
+      observableResult.observe(count, {
+        [SemanticAttributes.LOCK_TYPE]: lockType,
+      });
+    }
   }
 
   /** Locks resources using RedLock. It won't lock again if we're already inside a lock with the same resources. */
@@ -54,17 +103,53 @@ export class RunLocker {
         span.setAttribute("nested", false);
 
         // Different resources or not in a lock, proceed with new lock
+        const lockId = `${name}:${joinedResources}:${Date.now()}`;
+        const lockStartTime = performance.now();
+
         const [error, result] = await tryCatch(
           this.redlock.using(resources, duration, async (signal) => {
-            const newContext: LockContext = { resources: joinedResources, signal };
+            const newContext: LockContext = { 
+              resources: joinedResources, 
+              signal,
+              lockType: name
+            };
 
-            return this.asyncLocalStorage.run(newContext, async () => {
-              return routine(signal);
+            // Track active lock
+            this.activeLocks.set(lockId, {
+              lockType: name,
+              resources: resources,
             });
+
+            let lockSuccess = true;
+            try {
+              return this.asyncLocalStorage.run(newContext, async () => {
+                return routine(signal);
+              });
+            } catch (lockError) {
+              lockSuccess = false;
+              throw lockError;
+            } finally {
+              // Record lock duration
+              const lockDuration = performance.now() - lockStartTime;
+              this.lockDurationHistogram.record(lockDuration, {
+                [SemanticAttributes.LOCK_TYPE]: name,
+                [SemanticAttributes.LOCK_SUCCESS]: lockSuccess.toString(),
+              });
+
+              // Remove from active locks when done
+              this.activeLocks.delete(lockId);
+            }
           })
         );
 
         if (error) {
+          // Record failed lock acquisition
+          const lockDuration = performance.now() - lockStartTime;
+          this.lockDurationHistogram.record(lockDuration, {
+            [SemanticAttributes.LOCK_TYPE]: name,
+            [SemanticAttributes.LOCK_SUCCESS]: "false",
+          });
+
           this.logger.error("[RunLocker] Error locking resources", { error, resources, duration });
           throw error;
         }

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

@@ -45,6 +45,7 @@ import { z } from "zod";
 const SemanticAttributes = {
   QUEUE: "runqueue.queue",
   WORKER_QUEUE: "runqueue.workerQueue",
+  MASTER_QUEUE_SHARD: "runqueue.masterQueueShard",
   CONSUMER_ID: "runqueue.consumerId",
   RUN_ID: "runqueue.runId",
   RESULT_COUNT: "runqueue.resultCount",
@@ -141,7 +142,17 @@ export class RunQueue {
       }
     );
 
+    const masterQueueObservableGauge = this._meter.createObservableGauge(
+      "runqueue.masterQueue.length",
+      {
+        description: "The number of queues in the master queue shard",
+        unit: "1",
+        valueType: ValueType.INT,
+      }
+    );
+
     workerQueueObservableGauge.addCallback(this.#updateWorkerQueueLength.bind(this));
+    masterQueueObservableGauge.addCallback(this.#updateMasterQueueLength.bind(this));
 
     this.abortController = new AbortController();
 
@@ -206,6 +217,17 @@ export class RunQueue {
     }
   }
 
+  async #updateMasterQueueLength(observableResult: ObservableResult<Attributes>) {
+    for (let shard = 0; shard < this.shardCount; shard++) {
+      const masterQueueKey = this.keys.masterQueueKeyForShard(shard);
+      const masterQueueLength = await this.redis.zcard(masterQueueKey);
+
+      observableResult.observe(masterQueueLength, {
+        [SemanticAttributes.MASTER_QUEUE_SHARD]: shard.toString(),
+      });
+    }
+  }
+
   public async updateQueueConcurrencyLimits(
     env: MinimalAuthenticatedEnvironment,
     queue: string,