feat(supervisor): add cluster pod-count dequeue backpressure source (#4027)

Adds an in-process backpressure signal that pauses dequeuing when the
Kubernetes cluster is saturated, so work overflows cheaply in the queue
instead of piling up as unschedulable pods. Saturation is read by
scraping the apiserver's total pod-object count
(`apiserver_storage_objects{resource="pods"}`) and applying an
engage/release threshold with hysteresis - a single lightweight
aggregate scrape, not a pod listing.

Backpressure sources are now evaluated independently and OR'd: each
source has its own enable and dry-run flag, and the supervisor engages
if any enabled source trips. This adds the pod-count source alongside
the existing one without changing it, and is extensible to more sources
later. Off by default.

The scrape uses the in-cluster kubeconfig over `https` so TLS verifies
against the cluster CA (the fetch-options helper attaches the CA as an
`https.Agent`, which the global `fetch` ignores - that path silently
dropped the CA). Enabling the pod-count source requires the supervisor's
service account to be granted `get` on the `/metrics` non-resource URL;
that RBAC and the per-deployment env wiring are operator-side and live
elsewhere.

New config (pod-count source):
`TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED` (default false),
`_POD_COUNT_DRY_RUN` (default true), `_POD_COUNT_ENGAGE` /
`_POD_COUNT_RELEASE` (hysteresis thresholds), `_POD_COUNT_REFRESH_MS`
(scrape interval, default 5s). The existing source's flags are
unchanged.

Observability: a `supervisor_cluster_pod_count` gauge, and the pod-count
monitor's metrics are namespaced (`supervisor_backpressure_pod_count_*`)
so the existing backpressure metrics keep their names.

Author: nicktrn

.server-changes/supervisor-pod-count-backpressure.md

@@ -0,0 +1,6 @@
+---
+area: supervisor
+type: feature
+---
+
+The supervisor can pause dequeuing when the Kubernetes cluster is saturated, based on the cluster's total pod count. Opt-in and off by default.

apps/supervisor/src/backpressure/k8sPodCountSignalSource.test.ts

@@ -0,0 +1,95 @@
+import { describe, it, expect } from "vitest";
+import { parsePodCount, K8sPodCountSignalSource } from "./k8sPodCountSignalSource.js";
+
+describe("parsePodCount", () => {
+  it("reads the pods object count", () => {
+    const text = [
+      "# HELP apiserver_storage_objects Number of stored objects",
+      "# TYPE apiserver_storage_objects gauge",
+      'apiserver_storage_objects{resource="pods"} 8421',
+      'apiserver_storage_objects{resource="configmaps"} 17',
+    ].join("\n");
+    expect(parsePodCount(text)).toBe(8421);
+  });
+
+  it("is tolerant of extra labels in any order", () => {
+    const text = 'apiserver_storage_objects{group="",resource="pods",extra="x"} 12';
+    expect(parsePodCount(text)).toBe(12);
+  });
+
+  it("parses scientific notation", () => {
+    const text = 'apiserver_storage_objects{resource="pods"} 1.2e+04';
+    expect(parsePodCount(text)).toBe(12000);
+  });
+
+  it("throws when the pods metric is absent", () => {
+    const text = 'apiserver_storage_objects{resource="configmaps"} 17';
+    expect(() => parsePodCount(text)).toThrow(/not found/);
+  });
+
+  it("throws on a non-finite value (e.g. 1e999)", () => {
+    const text = 'apiserver_storage_objects{resource="pods"} 1e999';
+    expect(() => parsePodCount(text)).toThrow();
+  });
+
+  it("throws on a negative value", () => {
+    const text = 'apiserver_storage_objects{resource="pods"} -5';
+    expect(() => parsePodCount(text)).toThrow();
+  });
+});
+
+function metrics(count: number): string {
+  return `apiserver_storage_objects{resource="pods"} ${count}`;
+}
+
+describe("K8sPodCountSignalSource", () => {
+  it("engages at the engage threshold and reports the count", async () => {
+    const counts: number[] = [];
+    const source = new K8sPodCountSignalSource({
+      fetchMetrics: async () => metrics(10000),
+      engageThreshold: 10000,
+      releaseThreshold: 5000,
+      reportPodCount: (c) => counts.push(c),
+    });
+    const verdict = await source.read();
+    expect(verdict.engaged).toBe(true);
+    expect(typeof verdict.ts).toBe("number");
+    expect(counts).toEqual([10000]);
+  });
+
+  it("does not engage below the engage threshold", async () => {
+    const source = new K8sPodCountSignalSource({
+      fetchMetrics: async () => metrics(9999),
+      engageThreshold: 10000,
+      releaseThreshold: 5000,
+    });
+    expect((await source.read()).engaged).toBe(false);
+  });
+
+  it("stays engaged in the hysteresis band, releases only below release threshold", async () => {
+    let count = 10000;
+    const source = new K8sPodCountSignalSource({
+      fetchMetrics: async () => metrics(count),
+      engageThreshold: 10000,
+      releaseThreshold: 5000,
+    });
+    expect((await source.read()).engaged).toBe(true);  // engage
+    count = 7000;
+    expect((await source.read()).engaged).toBe(true);  // band -> still engaged
+    count = 4999;
+    expect((await source.read()).engaged).toBe(false); // below release -> off
+    count = 7000;
+    expect((await source.read()).engaged).toBe(false); // band again -> stays off
+  });
+
+  it("propagates scrape failures (monitor fails open on throw)", async () => {
+    const source = new K8sPodCountSignalSource({
+      fetchMetrics: async () => {
+        throw new Error("connection refused");
+      },
+      engageThreshold: 10000,
+      releaseThreshold: 5000,
+    });
+    await expect(source.read()).rejects.toThrow("connection refused");
+  });
+});

apps/supervisor/src/backpressure/k8sPodCountSignalSource.ts

@@ -0,0 +1,46 @@
+import type { BackpressureSignalSource, BackpressureVerdict } from "./backpressureMonitor.js";
+
+// Reads the apiserver's stored-pod-object count from a Prometheus /metrics scrape.
+const POD_COUNT_RE = /^apiserver_storage_objects\{[^}]*resource="pods"[^}]*\}\s+([0-9.eE+]+)/m;
+
+export function parsePodCount(metricsText: string): number {
+  const match = metricsText.match(POD_COUNT_RE);
+  if (!match) {
+    throw new Error('apiserver_storage_objects{resource="pods"} not found in metrics');
+  }
+  const value = Number(match[1]);
+  if (!Number.isFinite(value)) {
+    throw new Error(`unparseable pod count: ${match[1]}`);
+  }
+  return value;
+}
+
+export type K8sPodCountSignalSourceOptions = {
+  fetchMetrics: () => Promise<string>;
+  engageThreshold: number;
+  releaseThreshold: number;
+  reportPodCount?: (count: number) => void;
+};
+
+// Engage/release with hysteresis so a count hovering near the line doesn't flap.
+export class K8sPodCountSignalSource implements BackpressureSignalSource {
+  private engaged = false;
+
+  constructor(private readonly opts: K8sPodCountSignalSourceOptions) {}
+
+  async read(): Promise<BackpressureVerdict> {
+    const text = await this.opts.fetchMetrics();
+    const count = parsePodCount(text);
+    this.opts.reportPodCount?.(count);
+
+    if (this.engaged) {
+      if (count < this.opts.releaseThreshold) {
+        this.engaged = false;
+      }
+    } else if (count >= this.opts.engageThreshold) {
+      this.engaged = true;
+    }
+
+    return { engaged: this.engaged, ts: Date.now() };
+  }
+}

apps/supervisor/src/clients/kubernetes.ts

@@ -1,3 +1,4 @@
+import * as https from "node:https";
 import * as k8s from "@kubernetes/client-node";
 import { Informer } from "@kubernetes/client-node";
 import { ListPromise } from "@kubernetes/client-node";
@@ -53,3 +54,57 @@ function getKubeConfig() {
 }
 
 export { k8s };
+
+/**
+ * Builds a function that scrapes the apiserver's Prometheus /metrics endpoint.
+ * One lightweight aggregate read - not a pod listing. Requires the service
+ * account to be granted GET on the /metrics non-resource URL.
+ */
+export function createApiserverMetricsFetcher(timeoutMs: number): () => Promise<string> {
+  const kubeConfig = getKubeConfig();
+
+  return async () => {
+    const cluster = kubeConfig.getCurrentCluster();
+    if (!cluster) {
+      throw new Error("no current cluster in kubeconfig");
+    }
+    const url = new URL(`${cluster.server}/metrics`);
+    const opts: https.RequestOptions = {
+      method: "GET",
+      protocol: url.protocol,
+      hostname: url.hostname,
+      port: url.port,
+      path: url.pathname,
+    };
+    // applyToHTTPSOptions sets the cluster CA, client cert/key, and auth headers
+    // (incl. exec plugins) on the request - so TLS verifies against the cluster
+    // CA, not the system store. The fetch-options path attaches the CA as an
+    // https.Agent, which global fetch (undici) ignores.
+    await kubeConfig.applyToHTTPSOptions(opts);
+
+    return new Promise<string>((resolve, reject) => {
+      const req = https.request(opts, (res) => {
+        const status = res.statusCode ?? 0;
+        let body = "";
+        res.setEncoding("utf8");
+        res.on("data", (chunk) => {
+          body += chunk;
+        });
+        res.on("end", () => {
+          if (status >= 200 && status < 300) {
+            resolve(body);
+          } else {
+            reject(new Error(`apiserver /metrics scrape failed: ${status}`));
+          }
+        });
+      });
+      // Without this a hung connect/TLS/read never settles, and the monitor's
+      // refreshInFlight guard would freeze the source (silent fail-open).
+      req.setTimeout(timeoutMs, () => {
+        req.destroy(new Error(`apiserver /metrics scrape timed out after ${timeoutMs}ms`));
+      });
+      req.on("error", reject);
+      req.end();
+    });
+  };
+}

apps/supervisor/src/env.test.ts

@@ -0,0 +1,64 @@
+import { describe, it, expect, vi } from "vitest";
+
+// Mock std-env before importing env.ts so the module-level `Env.parse(stdEnv)`
+// doesn't fail in a test environment that lacks required vars.
+vi.mock("std-env", () => ({
+  env: {
+    TRIGGER_API_URL: "http://localhost:3030",
+    TRIGGER_WORKER_TOKEN: "test-token",
+    MANAGED_WORKER_SECRET: "test-secret",
+    OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+  },
+}));
+
+const { Env } = await import("./env.js");
+
+// Minimal env that satisfies all required fields; everything else has defaults.
+const base = {
+  TRIGGER_API_URL: "http://localhost:3030",
+  TRIGGER_WORKER_TOKEN: "test-token",
+  MANAGED_WORKER_SECRET: "test-secret",
+  OTEL_EXPORTER_OTLP_ENDPOINT: "http://localhost:4318",
+};
+
+describe("Env superRefine - backpressure source awareness", () => {
+  it("pod-count source can be enabled without a Redis host", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED: "true",
+      })
+    ).not.toThrow();
+  });
+
+  it("redis source requires a Redis host", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: "true",
+      })
+    ).toThrow();
+  });
+
+  it("both sources can be enabled together (with a Redis host)", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_ENABLED: "true",
+        TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_HOST: "localhost",
+        TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED: "true",
+      })
+    ).not.toThrow();
+  });
+
+  it("rejects pod-count release >= engage when the source is enabled", () => {
+    expect(() =>
+      Env.parse({
+        ...base,
+        TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED: "true",
+        TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE: "100",
+        TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE: "100",
+      })
+    ).toThrow();
+  });
+});

apps/supervisor/src/env.ts

@@ -3,7 +3,7 @@ import { env as stdEnv } from "std-env";
 import { z } from "zod";
 import { AdditionalEnvVars, BoolEnv } from "./envUtil.js";
 
-const Env = z
+export const Env = z
   .object({
     // This will come from `spec.nodeName` in k8s
     TRIGGER_WORKER_INSTANCE_NAME: z.string().default(randomUUID()),
@@ -73,6 +73,18 @@ const Env = z
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_USERNAME: z.string().optional(),
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_PASSWORD: z.string().optional(),
     TRIGGER_DEQUEUE_BACKPRESSURE_REDIS_TLS_DISABLED: BoolEnv.default(false),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED: BoolEnv.default(false),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_DRY_RUN: BoolEnv.default(true),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE: z.coerce.number().int().positive().default(10_000),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE: z.coerce.number().int().positive().default(5_000),
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_REFRESH_MS: z.coerce.number().int().positive().default(5_000),
+    // Hard timeout on the apiserver /metrics scrape. A hung request would otherwise
+    // never settle and freeze the monitor's refresh loop (fail-open silently).
+    TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_SCRAPE_TIMEOUT_MS: z.coerce
+      .number()
+      .int()
+      .positive()
+      .default(10_000),
 
     // Optional services
     TRIGGER_WARM_START_URL: z.string().optional(),
@@ -312,6 +324,18 @@ const Env = z
     TRIGGER_WIDE_EVENTS_NOISY_ROUTES: BoolEnv.default(false),
   })
   .superRefine((data, ctx) => {
+    if (
+      data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENABLED &&
+      data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE >=
+        data.TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE
+    ) {
+      ctx.addIssue({
+        code: z.ZodIssueCode.custom,
+        message:
+          "TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE must be less than TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_ENGAGE",
+        path: ["TRIGGER_DEQUEUE_BACKPRESSURE_POD_COUNT_RELEASE"],
+      });
+    }
     if (data.COMPUTE_SNAPSHOTS_ENABLED && !data.TRIGGER_METADATA_URL) {
       ctx.addIssue({
         code: z.ZodIssueCode.custom,