feat(supervisor): instrument outbound worker api client requests

Author: nicktrn

.changeset/supervisor-http-client-request-metrics.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/core": patch
+---
+
+Add an optional `onHttpRequestComplete` callback to the supervisor worker API client so hosts can record metrics for its outbound requests (method, response status, and outcome).

apps/supervisor/src/index.ts

@@ -329,6 +329,7 @@ class ManagedSupervisor {
       runNotificationsEnabled: env.TRIGGER_WORKLOAD_API_ENABLED,
       heartbeatIntervalSeconds: env.TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS,
       sendRunDebugLogs: env.SEND_RUN_DEBUG_LOGS,
+      onHttpRequestComplete: (metric) => outboundRequestsTotal.inc(metric),
       preDequeue: async () => {
         // Synchronous, hot-path-safe cached read; false when no monitors are active.
         const skipForBackpressure = this.backpressureMonitors.some((m) => m.shouldSkipDequeue());

packages/core/src/v3/runEngineWorker/supervisor/http.ts

@@ -21,9 +21,9 @@ import {
   WorkerApiSuspendRunResponseBody,
   WorkerApiRunSnapshotsSinceResponseBody,
 } from "./schemas.js";
-import type { SupervisorClientCommonOptions } from "./types.js";
+import type { SupervisorClientCommonOptions, SupervisorHttpRequestMetric } from "./types.js";
 import { getDefaultWorkerHeaders } from "./util.js";
-import { wrapZodFetch } from "../../zodfetch.js";
+import { wrapZodFetch, type ApiResult, type ZodFetchOptions } from "../../zodfetch.js";
 import { createHeaders } from "../util.js";
 import { WORKER_HEADERS } from "../consts.js";
 import { SimpleStructuredLogger } from "../../utils/structuredLogger.js";
@@ -36,6 +36,7 @@ export class SupervisorHttpClient {
   private readonly instanceName: string;
   private readonly defaultHeaders: Record<string, string>;
   private readonly sendRunDebugLogs: boolean;
+  private readonly onHttpRequestComplete?: (metric: SupervisorHttpRequestMetric) => void;
 
   private readonly logger = new SimpleStructuredLogger("supervisor-http-client");
 
@@ -45,6 +46,7 @@ export class SupervisorHttpClient {
     this.instanceName = opts.instanceName;
     this.defaultHeaders = getDefaultWorkerHeaders(opts);
     this.sendRunDebugLogs = opts.sendRunDebugLogs ?? false;
+    this.onHttpRequestComplete = opts.onHttpRequestComplete;
 
     if (!this.apiUrl) {
       throw new Error("apiURL is required and needs to be a non-empty string");
@@ -59,8 +61,38 @@ export class SupervisorHttpClient {
     }
   }
 
+  private async request<T extends z.ZodTypeAny>(
+    name: string,
+    schema: T,
+    url: string,
+    requestInit?: RequestInit,
+    options?: ZodFetchOptions<z.output<T>>
+  ): Promise<ApiResult<z.infer<T>>> {
+    const result = await wrapZodFetch(schema, url, requestInit, options);
+
+    if (this.onHttpRequestComplete) {
+      const method = requestInit?.method ?? "GET";
+
+      if (result.success) {
+        this.onHttpRequestComplete({ name, method, status: "2xx", outcome: "ok" });
+      } else if (typeof result.statusCode === "number") {
+        this.onHttpRequestComplete({
+          name,
+          method,
+          status: String(result.statusCode),
+          outcome: "http_error",
+        });
+      } else {
+        this.onHttpRequestComplete({ name, method, status: "none", outcome: "network_error" });
+      }
+    }
+
+    return result;
+  }
+
   async connect(body: WorkerApiConnectRequestBody) {
-    return wrapZodFetch(
+    return this.request(
+      "connect",
       WorkerApiConnectResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/connect`,
       {
@@ -75,7 +107,8 @@ export class SupervisorHttpClient {
   }
 
   async dequeue(body: WorkerApiDequeueRequestBody) {
-    return wrapZodFetch(
+    return this.request(
+      "dequeue",
       WorkerApiDequeueResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/dequeue`,
       {
@@ -91,7 +124,8 @@ export class SupervisorHttpClient {
 
   /** @deprecated Not currently used */
   async dequeueFromVersion(deploymentId: string, maxRunCount = 1, runnerId?: string) {
-    return wrapZodFetch(
+    return this.request(
+      "dequeue_from_version",
       WorkerApiDequeueResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/deployments/${deploymentId}/dequeue?maxRunCount=${maxRunCount}`,
       {
@@ -104,7 +138,8 @@ export class SupervisorHttpClient {
   }
 
   async heartbeatWorker(body: WorkerApiHeartbeatRequestBody) {
-    return wrapZodFetch(
+    return this.request(
+      "heartbeat_worker",
       WorkerApiHeartbeatResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/heartbeat`,
       {
@@ -124,7 +159,8 @@ export class SupervisorHttpClient {
     body: WorkerApiRunHeartbeatRequestBody,
     runnerId?: string
   ) {
-    return wrapZodFetch(
+    return this.request(
+      "heartbeat_run",
       WorkerApiRunHeartbeatResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/heartbeat`,
       {
@@ -146,7 +182,8 @@ export class SupervisorHttpClient {
     runnerId?: string,
     environmentId?: string
   ) {
-    return wrapZodFetch(
+    return this.request(
+      "start_run_attempt",
       WorkerApiRunAttemptStartResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/start`,
       {
@@ -168,7 +205,8 @@ export class SupervisorHttpClient {
     runnerId?: string,
     environmentId?: string
   ) {
-    return wrapZodFetch(
+    return this.request(
+      "complete_run_attempt",
       WorkerApiRunAttemptCompleteResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/complete`,
       {
@@ -184,7 +222,8 @@ export class SupervisorHttpClient {
   }
 
   async getLatestSnapshot(runId: string, runnerId?: string, environmentId?: string) {
-    return wrapZodFetch(
+    return this.request(
+      "get_latest_snapshot",
       WorkerApiRunLatestSnapshotResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/latest`,
       {
@@ -204,7 +243,8 @@ export class SupervisorHttpClient {
     runnerId?: string,
     environmentId?: string
   ) {
-    return wrapZodFetch(
+    return this.request(
+      "get_snapshots_since",
       WorkerApiRunSnapshotsSinceResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/since/${snapshotId}`,
       {
@@ -224,7 +264,8 @@ export class SupervisorHttpClient {
     }
 
     try {
-      const res = await wrapZodFetch(
+      const res = await this.request(
+        "send_debug_log",
         z.unknown(),
         `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/logs/debug`,
         {
@@ -252,7 +293,8 @@ export class SupervisorHttpClient {
     runnerId?: string,
     environmentId?: string
   ) {
-    return wrapZodFetch(
+    return this.request(
+      "continue_run_execution",
       WorkerApiContinueRunExecutionRequestBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/continue`,
       {
@@ -292,7 +334,8 @@ export class SupervisorHttpClient {
     runnerId?: string;
     body: WorkerApiSuspendRunRequestBody;
   }) {
-    return wrapZodFetch(
+    return this.request(
+      "submit_suspend_completion",
       WorkerApiSuspendRunResponseBody,
       `${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/suspend`,
       {

packages/core/src/v3/runEngineWorker/supervisor/types.ts

@@ -1,12 +1,20 @@
 import type { MachineResources } from "../../schemas/runEngine.js";
 
+export type SupervisorHttpRequestMetric = {
+  name: string;
+  method: string;
+  status: string;
+  outcome: "ok" | "http_error" | "network_error";
+};
+
 export type SupervisorClientCommonOptions = {
   apiUrl: string;
   workerToken: string;
   instanceName: string;
   deploymentId?: string;
   managedWorkerSecret?: string;
   sendRunDebugLogs?: boolean;
+  onHttpRequestComplete?: (metric: SupervisorHttpRequestMetric) => void;
 };
 
 export type PreDequeueFn = () => Promise<{