Skip to content

Commit 722e240

Browse files
nicktrnnicktrn
authored
feat(supervisor): add prometheus metric for outbound http requests (#4350)
Adds Prometheus metrics so the supervisor's outbound HTTP calls are observable - including client-side failures that previously only surfaced as a log line. - `supervisor_outbound_request_total{name, method, status, outcome}` - counts every outbound request. `outcome` separates a transport failure (`network_error`), an HTTP error response (`http_error`), a response that failed schema validation (`invalid_response`), and success (`ok`). - `supervisor_outbound_request_duration_seconds{name, outcome}` - latency histogram. Leaner labels than the counter (no `status`) to avoid bucket×label cardinality; buckets match the existing dequeue-latency histogram since these calls share the same retrying HTTP client and long-poll envelope. Coverage: - The warm-start request (a one-off `fetch`) - instrumented inline; the response status code is now also included in the failure log (it was previously dropped). - All worker API client calls (`SupervisorHttpClient`: dequeue, run attempt start/complete, heartbeats, snapshots, continue, suspend, debug-log, connect) - routed through a single instrumented `request()` helper that reports via an optional `onHttpRequestComplete` callback on the client, which the supervisor wires into the counter + histogram. Low cardinality by design: `name` is a **static per-endpoint label** (e.g. `dequeue`, `start_run_attempt`), never the interpolated URL - so no run/snapshot IDs land in labels, mirroring the templated `route` labels on the inbound HTTP server. Registered on the existing metrics registry, exposed on `/metrics` with no new wiring. Internal-only change (no package release needed), so the changelog note is a single `.server-changes` entry.
1 parent 88ca009 commit 722e240

4 files changed

Lines changed: 127 additions & 15 deletions

File tree

‎.server-changes/supervisor-outbound-request-metrics.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,6 @@
1+
---
2+
area: supervisor
3+
type: improvement
4+
---
5+
6+
Improved supervisor observability: it now reports metrics for its outbound requests, making failed calls to upstream services easier to monitor.

‎apps/supervisor/src/index.ts‎

Lines changed: 38 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ import {
2222
isKubernetesEnvironment,
2323
} from "@trigger.dev/core/v3/serverOnly";
2424
import { createK8sApi, createApiserverMetricsFetcher } from "./clients/kubernetes.js";
25-
import { collectDefaultMetrics, Gauge, Histogram } from "prom-client";
25+
import { collectDefaultMetrics, Counter, Gauge, Histogram } from "prom-client";
2626
import { register } from "./metrics.js";
2727
import { PodCleaner } from "./services/podCleaner.js";
2828
import { FailedPodHandler } from "./services/failedPodHandler.js";
@@ -60,6 +60,21 @@ const workloadCreateDuration = new Histogram({
6060
registers: [register],
6161
});
6262

63+
const outboundRequestsTotal = new Counter({
64+
name: "supervisor_outbound_request_total",
65+
help: "Count of outbound HTTP requests from the supervisor, by target name, method, response status, and outcome (ok, http_error, invalid_response, network_error).",
66+
labelNames: ["name", "method", "status", "outcome"],
67+
registers: [register],
68+
});
69+
70+
const outboundRequestDuration = new Histogram({
71+
name: "supervisor_outbound_request_duration_seconds",
72+
help: "Duration of outbound HTTP requests from the supervisor, by target name and outcome. Includes the HTTP client's internal retries and backoff.",
73+
labelNames: ["name", "outcome"],
74+
buckets: [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 11, 12.5, 15, 20, 30, 60],
75+
registers: [register],
76+
});
77+
6378
class ManagedSupervisor {
6479
private readonly workerSession: SupervisorSession;
6580
private readonly metricsServer?: HttpServer;
@@ -322,6 +337,10 @@ class ManagedSupervisor {
322337
runNotificationsEnabled: env.TRIGGER_WORKLOAD_API_ENABLED,
323338
heartbeatIntervalSeconds: env.TRIGGER_WORKER_HEARTBEAT_INTERVAL_SECONDS,
324339
sendRunDebugLogs: env.SEND_RUN_DEBUG_LOGS,
340+
onHttpRequestComplete: ({ name, method, status, outcome, durationMs }) => {
341+
outboundRequestsTotal.inc({ name, method, status, outcome });
342+
outboundRequestDuration.observe({ name, outcome }, durationMs / 1000);
343+
},
325344
preDequeue: async () => {
326345
// Synchronous, hot-path-safe cached read; false when no monitors are active.
327346
const skipForBackpressure = this.backpressureMonitors.some((m) => m.shouldSkipDequeue());
@@ -692,6 +711,18 @@ class ManagedSupervisor {
692711
headers.traceparent = traceparent;
693712
}
694713

714+
const requestStart = performance.now();
715+
const record = (
716+
status: string,
717+
outcome: "ok" | "http_error" | "invalid_response" | "network_error"
718+
) => {
719+
outboundRequestsTotal.inc({ name: "warm_start", method: "POST", status, outcome });
720+
outboundRequestDuration.observe(
721+
{ name: "warm_start", outcome },
722+
(performance.now() - requestStart) / 1000
723+
);
724+
};
725+
695726
try {
696727
const res = await fetch(warmStartUrlWithPath.href, {
697728
method: "POST",
@@ -700,8 +731,10 @@ class ManagedSupervisor {
700731
});
701732

702733
if (!res.ok) {
734+
record(String(res.status), "http_error");
703735
this.logger.error("Warm start failed", {
704736
runId: dequeuedMessage.run.id,
737+
statusCode: res.status,
705738
});
706739
return false;
707740
}
@@ -710,15 +743,19 @@ class ManagedSupervisor {
710743
const parsedData = z.object({ didWarmStart: z.boolean() }).safeParse(data);
711744

712745
if (!parsedData.success) {
746+
record(String(res.status), "invalid_response");
713747
this.logger.error("Warm start response invalid", {
714748
runId: dequeuedMessage.run.id,
715749
data,
716750
});
717751
return false;
718752
}
719753

754+
record(String(res.status), "ok");
755+
720756
return parsedData.data.didWarmStart;
721757
} catch (error) {
758+
record("none", "network_error");
722759
this.logger.error("Warm start error", {
723760
runId: dequeuedMessage.run.id,
724761
error,

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

Lines changed: 74 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,9 @@ import {
2121
WorkerApiSuspendRunResponseBody,
2222
WorkerApiRunSnapshotsSinceResponseBody,
2323
} from "./schemas.js";
24-
import type { SupervisorClientCommonOptions } from "./types.js";
24+
import type { SupervisorClientCommonOptions, SupervisorHttpRequestMetric } from "./types.js";
2525
import { getDefaultWorkerHeaders } from "./util.js";
26-
import { wrapZodFetch } from "../../zodfetch.js";
26+
import { wrapZodFetch, type ApiResult, type ZodFetchOptions } from "../../zodfetch.js";
2727
import { createHeaders } from "../util.js";
2828
import { WORKER_HEADERS } from "../consts.js";
2929
import { SimpleStructuredLogger } from "../../utils/structuredLogger.js";
@@ -36,6 +36,7 @@ export class SupervisorHttpClient {
3636
private readonly instanceName: string;
3737
private readonly defaultHeaders: Record<string, string>;
3838
private readonly sendRunDebugLogs: boolean;
39+
private readonly onHttpRequestComplete?: (metric: SupervisorHttpRequestMetric) => void;
3940

4041
private readonly logger = new SimpleStructuredLogger("supervisor-http-client");
4142

@@ -45,6 +46,7 @@ export class SupervisorHttpClient {
4546
this.instanceName = opts.instanceName;
4647
this.defaultHeaders = getDefaultWorkerHeaders(opts);
4748
this.sendRunDebugLogs = opts.sendRunDebugLogs ?? false;
49+
this.onHttpRequestComplete = opts.onHttpRequestComplete;
4850

4951
if (!this.apiUrl) {
5052
throw new Error("apiURL is required and needs to be a non-empty string");
@@ -59,8 +61,55 @@ export class SupervisorHttpClient {
5961
}
6062
}
6163

64+
private async request<T extends z.ZodTypeAny>(
65+
name: string,
66+
schema: T,
67+
url: string,
68+
requestInit?: RequestInit,
69+
options?: ZodFetchOptions<z.output<T>>
70+
): Promise<ApiResult<z.infer<T>>> {
71+
const start = performance.now();
72+
const result = await wrapZodFetch(schema, url, requestInit, options);
73+
74+
if (this.onHttpRequestComplete) {
75+
const durationMs = performance.now() - start;
76+
const method = requestInit?.method ?? "GET";
77+
78+
if (result.success) {
79+
this.onHttpRequestComplete({ name, method, status: "2xx", outcome: "ok", durationMs });
80+
} else if (result.statusCode === 200) {
81+
this.onHttpRequestComplete({
82+
name,
83+
method,
84+
status: "200",
85+
outcome: "invalid_response",
86+
durationMs,
87+
});
88+
} else if (typeof result.statusCode === "number") {
89+
this.onHttpRequestComplete({
90+
name,
91+
method,
92+
status: String(result.statusCode),
93+
outcome: "http_error",
94+
durationMs,
95+
});
96+
} else {
97+
this.onHttpRequestComplete({
98+
name,
99+
method,
100+
status: "none",
101+
outcome: "network_error",
102+
durationMs,
103+
});
104+
}
105+
}
106+
107+
return result;
108+
}
109+
62110
async connect(body: WorkerApiConnectRequestBody) {
63-
return wrapZodFetch(
111+
return this.request(
112+
"connect",
64113
WorkerApiConnectResponseBody,
65114
`${this.apiUrl}/engine/v1/worker-actions/connect`,
66115
{
@@ -75,7 +124,8 @@ export class SupervisorHttpClient {
75124
}
76125

77126
async dequeue(body: WorkerApiDequeueRequestBody) {
78-
return wrapZodFetch(
127+
return this.request(
128+
"dequeue",
79129
WorkerApiDequeueResponseBody,
80130
`${this.apiUrl}/engine/v1/worker-actions/dequeue`,
81131
{
@@ -91,7 +141,8 @@ export class SupervisorHttpClient {
91141

92142
/** @deprecated Not currently used */
93143
async dequeueFromVersion(deploymentId: string, maxRunCount = 1, runnerId?: string) {
94-
return wrapZodFetch(
144+
return this.request(
145+
"dequeue_from_version",
95146
WorkerApiDequeueResponseBody,
96147
`${this.apiUrl}/engine/v1/worker-actions/deployments/${deploymentId}/dequeue?maxRunCount=${maxRunCount}`,
97148
{
@@ -104,7 +155,8 @@ export class SupervisorHttpClient {
104155
}
105156

106157
async heartbeatWorker(body: WorkerApiHeartbeatRequestBody) {
107-
return wrapZodFetch(
158+
return this.request(
159+
"heartbeat_worker",
108160
WorkerApiHeartbeatResponseBody,
109161
`${this.apiUrl}/engine/v1/worker-actions/heartbeat`,
110162
{
@@ -124,7 +176,8 @@ export class SupervisorHttpClient {
124176
body: WorkerApiRunHeartbeatRequestBody,
125177
runnerId?: string
126178
) {
127-
return wrapZodFetch(
179+
return this.request(
180+
"heartbeat_run",
128181
WorkerApiRunHeartbeatResponseBody,
129182
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/heartbeat`,
130183
{
@@ -146,7 +199,8 @@ export class SupervisorHttpClient {
146199
runnerId?: string,
147200
environmentId?: string
148201
) {
149-
return wrapZodFetch(
202+
return this.request(
203+
"start_run_attempt",
150204
WorkerApiRunAttemptStartResponseBody,
151205
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/start`,
152206
{
@@ -168,7 +222,8 @@ export class SupervisorHttpClient {
168222
runnerId?: string,
169223
environmentId?: string
170224
) {
171-
return wrapZodFetch(
225+
return this.request(
226+
"complete_run_attempt",
172227
WorkerApiRunAttemptCompleteResponseBody,
173228
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/attempts/complete`,
174229
{
@@ -184,7 +239,8 @@ export class SupervisorHttpClient {
184239
}
185240

186241
async getLatestSnapshot(runId: string, runnerId?: string, environmentId?: string) {
187-
return wrapZodFetch(
242+
return this.request(
243+
"get_latest_snapshot",
188244
WorkerApiRunLatestSnapshotResponseBody,
189245
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/latest`,
190246
{
@@ -204,7 +260,8 @@ export class SupervisorHttpClient {
204260
runnerId?: string,
205261
environmentId?: string
206262
) {
207-
return wrapZodFetch(
263+
return this.request(
264+
"get_snapshots_since",
208265
WorkerApiRunSnapshotsSinceResponseBody,
209266
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/since/${snapshotId}`,
210267
{
@@ -224,7 +281,8 @@ export class SupervisorHttpClient {
224281
}
225282

226283
try {
227-
const res = await wrapZodFetch(
284+
const res = await this.request(
285+
"send_debug_log",
228286
z.unknown(),
229287
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/logs/debug`,
230288
{
@@ -252,7 +310,8 @@ export class SupervisorHttpClient {
252310
runnerId?: string,
253311
environmentId?: string
254312
) {
255-
return wrapZodFetch(
313+
return this.request(
314+
"continue_run_execution",
256315
WorkerApiContinueRunExecutionRequestBody,
257316
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/continue`,
258317
{
@@ -292,7 +351,8 @@ export class SupervisorHttpClient {
292351
runnerId?: string;
293352
body: WorkerApiSuspendRunRequestBody;
294353
}) {
295-
return wrapZodFetch(
354+
return this.request(
355+
"submit_suspend_completion",
296356
WorkerApiSuspendRunResponseBody,
297357
`${this.apiUrl}/engine/v1/worker-actions/runs/${runId}/snapshots/${snapshotId}/suspend`,
298358
{

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

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,21 @@
11
import type { MachineResources } from "../../schemas/runEngine.js";
22

3+
export type SupervisorHttpRequestMetric = {
4+
name: string;
5+
method: string;
6+
status: string;
7+
outcome: "ok" | "http_error" | "invalid_response" | "network_error";
8+
durationMs: number;
9+
};
10+
311
export type SupervisorClientCommonOptions = {
412
apiUrl: string;
513
workerToken: string;
614
instanceName: string;
715
deploymentId?: string;
816
managedWorkerSecret?: string;
917
sendRunDebugLogs?: boolean;
18+
onHttpRequestComplete?: (metric: SupervisorHttpRequestMetric) => void;
1019
};
1120

1221
export type PreDequeueFn = () => Promise<{

0 commit comments

Comments
 (0)