feat(supervisor): optionally create a Runner k8s custom resource instead of building a pod

Adds an alternative Kubernetes workload manager behind a new
`KUBERNETES_RUN_CRD_ENABLED` setting, off by default. Instead of
building a run pod directly, it creates a `Runner` object in the
`compute.trigger.dev/v1alpha1` API group and stops, leaving pod
construction to the operator that reconciles that object. This keeps the
reasoning about uid pinning, node selection and the label set a network
policy selects on in one place instead of copying it a third time.

The Helm chart exposes it as
`supervisor.config.kubernetes.runCrdEnabled`, also off by default, which
sets that variable and grants the supervisor's Role the two writes the
backend needs: creating the `Runner`, and creating and patching the
Secret that carries the deployment token. Turning it on requires the
`Runner` CRD and a controller reconciling it to be installed, or the
create fails.

The existing pod-building path is unchanged and remains the default. The
two are mutually exclusive by construction: whichever one runs is the
only thing that creates a workload for a cold start.

Mono-RevId: 0665f61da14cbc8ec45c7fb75b013cf45f93fdc8

Author: 1stvamp

apps/supervisor/src/clients/kubernetes.ts

@@ -23,6 +23,9 @@ export function createK8sApi() {
     core: kubeConfig.makeApiClient(k8s.CoreV1Api),
     batch: kubeConfig.makeApiClient(k8s.BatchV1Api),
     apps: kubeConfig.makeApiClient(k8s.AppsV1Api),
+    custom: kubeConfig.makeApiClient(k8s.CustomObjectsApi),
+    // Patching needs a content type the generated clients do not offer.
+    objects: k8s.KubernetesObjectApi.makeApiClient(kubeConfig),
     makeInformer,
   };
 

apps/supervisor/src/env.ts

@@ -179,6 +179,10 @@ export const Env = z
 
     // Kubernetes settings
     KUBERNETES_FORCE_ENABLED: BoolEnv.default(false),
+    // Create a Runner and let the operator build the pod, instead of building
+    // the pod here. The two are mutually exclusive by construction: whichever
+    // one runs is the only thing that creates a workload for a cold start.
+    KUBERNETES_RUN_CRD_ENABLED: BoolEnv.default(false),
     KUBERNETES_NAMESPACE: z.string().default("default"),
     KUBERNETES_WORKER_NODETYPE_LABEL: NodeLabelValue.default("v4-worker"),
     KUBERNETES_IMAGE_PULL_SECRETS: z.string().optional(), // csv

apps/supervisor/src/index.ts

@@ -14,6 +14,7 @@ import {
   type ResourceMonitor,
 } from "./resourceMonitor.js";
 import { KubernetesWorkloadManager } from "./workloadManager/kubernetes.js";
+import { RunCrdWorkloadManager } from "./workloadManager/runCrd.js";
 import { DockerWorkloadManager } from "./workloadManager/docker.js";
 import { ComputeWorkloadManager } from "./workloadManager/compute.js";
 import {
@@ -80,7 +81,7 @@ class ManagedSupervisor {
   private readonly metricsServer?: HttpServer;
   private readonly workloadServer: WorkloadServer;
   private readonly workloadManager: WorkloadManager;
-  private readonly workloadManagerBackend: "compute" | "kubernetes" | "docker";
+  private readonly workloadManagerBackend: "compute" | "kubernetes" | "run-crd" | "docker";
   private readonly computeManager?: ComputeWorkloadManager;
   private readonly logger = new SimpleStructuredLogger("managed-supervisor");
   private readonly resourceMonitor: ResourceMonitor;
@@ -189,6 +190,12 @@ class ManagedSupervisor {
       this.computeManager = computeManager;
       this.workloadManager = computeManager;
       this.workloadManagerBackend = "compute";
+    } else if (this.isKubernetes && env.KUBERNETES_RUN_CRD_ENABLED) {
+      this.workloadManager = new RunCrdWorkloadManager({
+        ...workloadManagerOptions,
+        namespace: env.KUBERNETES_NAMESPACE,
+      });
+      this.workloadManagerBackend = "run-crd";
     } else if (this.isKubernetes) {
       this.workloadManager = new KubernetesWorkloadManager(workloadManagerOptions);
       this.workloadManagerBackend = "kubernetes";
@@ -197,7 +204,8 @@ class ManagedSupervisor {
       this.workloadManagerBackend = "docker";
     }
 
-    if (this.isKubernetes) {
+    // Not run-crd: the operator owns those pods and fails a Runner without one.
+    if (this.isKubernetes && this.workloadManagerBackend !== "run-crd") {
       if (env.POD_CLEANER_ENABLED) {
         this.logger.log("🧹 Pod cleaner enabled", {
           namespace: env.KUBERNETES_NAMESPACE,
@@ -227,6 +235,8 @@ class ManagedSupervisor {
       } else {
         this.logger.warn("Failed pod handler disabled");
       }
+    } else if (this.isKubernetes) {
+      this.logger.log("🧹 Pod cleaner and failed pod handler disabled, the operator owns run pods");
     }
 
     if (env.TRIGGER_DEQUEUE_INTERVAL_MS > env.TRIGGER_DEQUEUE_IDLE_INTERVAL_MS) {

apps/supervisor/src/services/failedPodHandler.test.ts

@@ -176,6 +176,43 @@ describe.skipIf(!process.env.K8S_INTEGRATION_TESTS)("FailedPodHandler Integratio
     }
   }, 30000);
 
+  it("should ignore operator-owned pods", async () => {
+    const handler = new FailedPodHandler({ namespace, k8s, register });
+
+    try {
+      // Operator-owned pods carry app=task-run as well, so only the runner
+      // label separates them. This handler deletes the pod that carries the
+      // termination reason, so processing one loses that reason too.
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        shouldFail: true,
+        labels: { app: "task-run", "compute.trigger.dev/runner": "runner-abc123" },
+      });
+
+      await waitForPodsPhase({
+        k8sApi: k8s,
+        namespace,
+        podNames,
+        phase: "Failed",
+      });
+
+      await handler.start();
+
+      await setTimeout(5000);
+
+      const exists = await podExists({ k8sApi: k8s, namespace, podName: podNames[0]! });
+      expect(exists).toBe(true);
+
+      const metrics = handler.getMetrics();
+      const processedPods = await metrics.processedPodsTotal.get();
+      expect(processedPods.values).toHaveLength(0);
+    } finally {
+      await handler.stop();
+    }
+  }, 30000);
+
   it("should not process pods that are being deleted", async () => {
     const handler = new FailedPodHandler({ namespace, k8s, register });
 

apps/supervisor/src/services/failedPodHandler.ts

@@ -7,6 +7,12 @@ import type { K8sApi } from "../clients/kubernetes.js";
 import { createK8sApi } from "../clients/kubernetes.js";
 import { register } from "../metrics.js";
 
+// Operator-owned pods carry app=task-run too, since the network policy selects
+// on it, so they have to be excluded by their own label instead. Unconditional:
+// a rolling update runs this against pods the next version created. One
+// constant because the list and the watch must select the same set.
+const RUN_POD_SELECTOR = "app=task-run,!compute.trigger.dev/runner";
+
 type PodStatus = "Pending" | "Running" | "Succeeded" | "Failed" | "Unknown" | "GracefulShutdown";
 
 export type FailedPodHandlerOptions = {
@@ -54,10 +60,10 @@ export class FailedPodHandler {
       () =>
         this.k8s.core.listNamespacedPod({
           namespace: this.namespace,
-          labelSelector: "app=task-run",
+          labelSelector: RUN_POD_SELECTOR,
           fieldSelector: "status.phase=Failed",
         }),
-      "app=task-run",
+      RUN_POD_SELECTOR,
       "status.phase=Failed"
     );
 

apps/supervisor/src/services/podCleaner.test.ts

@@ -232,6 +232,42 @@ describe.skipIf(!process.env.K8S_INTEGRATION_TESTS)("PodCleaner Integration Test
     }
   }, 30000);
 
+  it("should not delete operator-owned pods", async () => {
+    const podCleaner = new PodCleaner({ namespace, k8s, register });
+
+    try {
+      // Operator-owned pods carry app=task-run as well, so only the runner
+      // label separates them. Deleting one fails the run it was serving.
+      const podNames = await createTestPods({
+        k8sApi: k8s,
+        namespace,
+        count: 1,
+        labels: { app: "task-run", "compute.trigger.dev/runner": "runner-abc123" },
+        namePrefix: "operator-owned-pod",
+      });
+
+      if (!podNames[0]) {
+        throw new Error("Failed to create test pod");
+      }
+      const podName = podNames[0];
+
+      await waitForPodPhase({
+        k8sApi: k8s,
+        namespace,
+        podName,
+        phase: "Succeeded",
+      });
+
+      await podCleaner.start();
+
+      await setTimeout(5000);
+
+      expect(await podExists({ k8sApi: k8s, namespace, podName })).toBe(true);
+    } finally {
+      await podCleaner.stop();
+    }
+  }, 30000);
+
   it("should not delete pods that are still running", async () => {
     const podCleaner = new PodCleaner({ namespace, k8s, register });
 

apps/supervisor/src/services/podCleaner.ts

@@ -6,6 +6,11 @@ import type { K8sApi } from "../clients/kubernetes.js";
 import { createK8sApi } from "../clients/kubernetes.js";
 import { register } from "../metrics.js";
 
+// Operator-owned pods carry app=task-run too, since the network policy selects
+// on it, so they have to be excluded by their own label instead. Unconditional:
+// a rolling update runs this against pods the next version created.
+const RUN_POD_SELECTOR = "app=task-run,!compute.trigger.dev/runner";
+
 export type PodCleanerOptions = {
   namespace: string;
   k8s?: K8sApi;
@@ -72,7 +77,7 @@ export class PodCleaner {
       try {
         const result = await this.k8s.core.deleteCollectionNamespacedPod({
           namespace: this.namespace,
-          labelSelector: "app=task-run",
+          labelSelector: RUN_POD_SELECTOR,
           fieldSelector: "status.phase=Succeeded",
           limit: this.batchSize,
           _continue: continuationToken,