fix: enforce runner retention before job create
This commit is contained in:
lyon
@@ -0,0 +1,364 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactJson, redactText } from "../common/redaction.js";
|
||||
import type { CommandRecord, JsonRecord, JsonValue, RunRecord } from "../common/types.js";
|
||||
import { stableHash } from "../common/validation.js";
|
||||
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
|
||||
import type { AgentRunStore } from "./store.js";
|
||||
|
||||
export interface RunnerRetentionOptions {
|
||||
namespace: string;
|
||||
maxRunners: number;
|
||||
cleanupOrder: "oldest-inactive-last-active-first";
|
||||
activeHeartbeatMaxAgeMs: number;
|
||||
matchLabels: Record<string, string>;
|
||||
jobNamePrefixes: string[];
|
||||
ageBasedCleanup: {
|
||||
enabled: boolean;
|
||||
maxAgeHours: number | null;
|
||||
};
|
||||
kubectlCommand?: string;
|
||||
}
|
||||
|
||||
export interface RunnerRetentionSummary extends JsonRecord {
|
||||
enabled: boolean;
|
||||
namespace: string;
|
||||
maxRunners: number;
|
||||
incomingRunnerCount: number;
|
||||
liveRunnerCountBefore: number;
|
||||
liveRunnerCountAfter: number;
|
||||
runnerJobCountBefore: number;
|
||||
nonTerminalRunnerPodCountBefore: number;
|
||||
orphanNonTerminalRunnerPodCountBefore: number;
|
||||
inactiveCandidateCount: number;
|
||||
protectedActiveRunnerCount: number;
|
||||
selectedRunnerCount: number;
|
||||
deletedRunnerJobCount: number;
|
||||
deletedRunnerPodCount: number;
|
||||
deletedAssociatedResourceCount: number;
|
||||
activeRunRisk: boolean;
|
||||
reason: string;
|
||||
valuesPrinted: false;
|
||||
}
|
||||
|
||||
type CandidateKind = "idle" | "terminal" | "inactive";
|
||||
|
||||
interface K8sObject {
|
||||
kind?: string;
|
||||
metadata?: {
|
||||
name?: string;
|
||||
namespace?: string;
|
||||
creationTimestamp?: string;
|
||||
labels?: Record<string, string>;
|
||||
annotations?: Record<string, string>;
|
||||
ownerReferences?: Array<{ kind?: string; name?: string }>;
|
||||
};
|
||||
status?: JsonRecord;
|
||||
}
|
||||
|
||||
interface RunnerResourceEntry {
|
||||
resourceKind: "job" | "pod";
|
||||
name: string;
|
||||
jobName: string;
|
||||
runId: string | null;
|
||||
commandId: string | null;
|
||||
runnerId: string | null;
|
||||
createdAtMs: number;
|
||||
terminal: boolean;
|
||||
podPhase: string | null;
|
||||
protectedActive: boolean;
|
||||
protectedReason: string | null;
|
||||
candidateKind: CandidateKind;
|
||||
lastActiveAtMs: number;
|
||||
}
|
||||
|
||||
interface Snapshot {
|
||||
jobs: RunnerResourceEntry[];
|
||||
pods: RunnerResourceEntry[];
|
||||
liveRunnerCount: number;
|
||||
runnerJobCount: number;
|
||||
nonTerminalRunnerPodCount: number;
|
||||
orphanNonTerminalRunnerPodCount: number;
|
||||
}
|
||||
|
||||
export async function enforceRunnerRetentionBeforeCreate(input: { store: AgentRunStore; options: RunnerRetentionOptions; incomingRunnerCount?: number }): Promise<RunnerRetentionSummary> {
|
||||
const incomingRunnerCount = input.incomingRunnerCount ?? 1;
|
||||
const before = await runnerSnapshot(input.store, input.options);
|
||||
const targetBeforeCreate = Math.max(0, input.options.maxRunners - incomingRunnerCount);
|
||||
const requiredDeleteCount = Math.max(0, before.liveRunnerCount - targetBeforeCreate);
|
||||
const candidates = [...before.jobs, ...before.pods]
|
||||
.filter((item) => !item.protectedActive)
|
||||
.sort(compareCandidates);
|
||||
const selected = candidates.slice(0, requiredDeleteCount);
|
||||
const protectedActiveRunnerCount = before.jobs.filter((item) => item.protectedActive).length + before.pods.filter((item) => item.protectedActive).length;
|
||||
if (requiredDeleteCount > 0 && selected.length < requiredDeleteCount) {
|
||||
throw new AgentRunError("infra-failed", "runner retention cannot safely make room for a new runner", {
|
||||
httpStatus: 409,
|
||||
details: {
|
||||
reason: "runner-retention-no-safe-candidate",
|
||||
namespace: input.options.namespace,
|
||||
maxRunners: input.options.maxRunners,
|
||||
incomingRunnerCount,
|
||||
liveRunnerCountBefore: before.liveRunnerCount,
|
||||
requiredDeleteCount,
|
||||
inactiveCandidateCount: candidates.length,
|
||||
protectedActiveRunnerCount,
|
||||
activeRunRisk: protectedActiveRunnerCount > 0,
|
||||
valuesPrinted: false,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
let deletedRunnerJobCount = 0;
|
||||
let deletedRunnerPodCount = 0;
|
||||
let deletedAssociatedResourceCount = 0;
|
||||
for (const item of selected) {
|
||||
if (item.resourceKind === "job") {
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "job", item.name, "-n", input.options.namespace, "--ignore-not-found=true"]);
|
||||
deletedRunnerJobCount++;
|
||||
deletedAssociatedResourceCount += await deleteAssociatedResources(input.options, item.jobName);
|
||||
} else {
|
||||
await kubectlRun(input.options.kubectlCommand ?? "kubectl", ["delete", "pod", item.name, "-n", input.options.namespace, "--ignore-not-found=true"]);
|
||||
deletedRunnerPodCount++;
|
||||
}
|
||||
}
|
||||
const after = selected.length > 0 ? await runnerSnapshot(input.store, input.options) : before;
|
||||
return {
|
||||
enabled: true,
|
||||
namespace: input.options.namespace,
|
||||
maxRunners: input.options.maxRunners,
|
||||
incomingRunnerCount,
|
||||
liveRunnerCountBefore: before.liveRunnerCount,
|
||||
liveRunnerCountAfter: after.liveRunnerCount,
|
||||
runnerJobCountBefore: before.runnerJobCount,
|
||||
nonTerminalRunnerPodCountBefore: before.nonTerminalRunnerPodCount,
|
||||
orphanNonTerminalRunnerPodCountBefore: before.orphanNonTerminalRunnerPodCount,
|
||||
inactiveCandidateCount: candidates.length,
|
||||
protectedActiveRunnerCount,
|
||||
selectedRunnerCount: selected.length,
|
||||
deletedRunnerJobCount,
|
||||
deletedRunnerPodCount,
|
||||
deletedAssociatedResourceCount,
|
||||
activeRunRisk: protectedActiveRunnerCount > 0,
|
||||
reason: requiredDeleteCount === 0 ? "within-limit" : "pre-create-retention",
|
||||
selected: selected.map((item) => ({ resourceKind: item.resourceKind, name: item.name, jobName: item.jobName, candidateKind: item.candidateKind, protectedActive: false, lastActiveAt: new Date(item.lastActiveAtMs).toISOString(), valuesPrinted: false })),
|
||||
valuesPrinted: false,
|
||||
} satisfies RunnerRetentionSummary;
|
||||
}
|
||||
|
||||
async function runnerSnapshot(store: AgentRunStore, options: RunnerRetentionOptions): Promise<Snapshot> {
|
||||
const [jobObjects, podObjects] = await Promise.all([
|
||||
kubectlGetList(options, "jobs", labelSelector(options.matchLabels)),
|
||||
kubectlGetList(options, "pods", labelSelector(options.matchLabels)),
|
||||
]);
|
||||
const jobs: RunnerResourceEntry[] = [];
|
||||
for (const item of jobObjects) {
|
||||
const name = objectName(item);
|
||||
if (!name || !matchesJobPrefix(name, options.jobNamePrefixes)) continue;
|
||||
jobs.push(await jobEntry(store, item, options));
|
||||
}
|
||||
const jobNames = new Set(jobs.map((item) => item.jobName));
|
||||
const pods: RunnerResourceEntry[] = [];
|
||||
let nonTerminalRunnerPodCount = 0;
|
||||
let orphanNonTerminalRunnerPodCount = 0;
|
||||
for (const item of podObjects) {
|
||||
const name = objectName(item);
|
||||
if (!name) continue;
|
||||
const jobName = podJobName(item) ?? name;
|
||||
if (!matchesJobPrefix(jobName, options.jobNamePrefixes)) continue;
|
||||
const terminal = podTerminal(item);
|
||||
if (!terminal) nonTerminalRunnerPodCount++;
|
||||
if (jobNames.has(jobName)) continue;
|
||||
if (!terminal) orphanNonTerminalRunnerPodCount++;
|
||||
pods.push(await podEntry(store, item, options, jobName));
|
||||
}
|
||||
return {
|
||||
jobs,
|
||||
pods,
|
||||
liveRunnerCount: jobs.length + pods.filter((item) => !item.terminal).length,
|
||||
runnerJobCount: jobs.length,
|
||||
nonTerminalRunnerPodCount,
|
||||
orphanNonTerminalRunnerPodCount,
|
||||
};
|
||||
}
|
||||
|
||||
async function jobEntry(store: AgentRunStore, item: K8sObject, options: RunnerRetentionOptions): Promise<RunnerResourceEntry> {
|
||||
const name = objectName(item) ?? "unknown";
|
||||
const annotations = item.metadata?.annotations ?? {};
|
||||
const activity = await activityFor(store, {
|
||||
runId: annotations["agentrun.pikastech.local/run-id"] ?? null,
|
||||
commandId: annotations["agentrun.pikastech.local/command-id"] ?? null,
|
||||
runnerId: annotations["agentrun.pikastech.local/runner-id"] ?? null,
|
||||
terminal: jobTerminal(item),
|
||||
createdAtMs: objectCreatedAtMs(item),
|
||||
activeHeartbeatMaxAgeMs: options.activeHeartbeatMaxAgeMs,
|
||||
});
|
||||
return {
|
||||
resourceKind: "job",
|
||||
name,
|
||||
jobName: name,
|
||||
createdAtMs: objectCreatedAtMs(item),
|
||||
terminal: jobTerminal(item),
|
||||
podPhase: null,
|
||||
...activity,
|
||||
};
|
||||
}
|
||||
|
||||
async function podEntry(store: AgentRunStore, item: K8sObject, options: RunnerRetentionOptions, jobName: string): Promise<RunnerResourceEntry> {
|
||||
const name = objectName(item) ?? "unknown";
|
||||
const annotations = item.metadata?.annotations ?? {};
|
||||
const activity = await activityFor(store, {
|
||||
runId: annotations["agentrun.pikastech.local/run-id"] ?? null,
|
||||
commandId: annotations["agentrun.pikastech.local/command-id"] ?? null,
|
||||
runnerId: annotations["agentrun.pikastech.local/runner-id"] ?? null,
|
||||
terminal: podTerminal(item),
|
||||
createdAtMs: objectCreatedAtMs(item),
|
||||
activeHeartbeatMaxAgeMs: options.activeHeartbeatMaxAgeMs,
|
||||
});
|
||||
return {
|
||||
resourceKind: "pod",
|
||||
name,
|
||||
jobName,
|
||||
createdAtMs: objectCreatedAtMs(item),
|
||||
terminal: podTerminal(item),
|
||||
podPhase: stringPath(item, ["status", "phase"]),
|
||||
...activity,
|
||||
};
|
||||
}
|
||||
|
||||
async function activityFor(store: AgentRunStore, input: { runId: string | null; commandId: string | null; runnerId: string | null; terminal: boolean; createdAtMs: number; activeHeartbeatMaxAgeMs: number }): Promise<Pick<RunnerResourceEntry, "runId" | "commandId" | "runnerId" | "protectedActive" | "protectedReason" | "candidateKind" | "lastActiveAtMs">> {
|
||||
let run: RunRecord | null = null;
|
||||
let command: CommandRecord | null = null;
|
||||
if (input.runId) {
|
||||
try { run = await store.getRun(input.runId); } catch { run = null; }
|
||||
}
|
||||
if (input.commandId) {
|
||||
try { command = await store.getCommand(input.commandId); } catch { command = null; }
|
||||
}
|
||||
const lastActiveAtMs = Math.max(input.createdAtMs, isoMs(run?.updatedAt), isoMs(command?.updatedAt), isoMs(run?.leaseExpiresAt));
|
||||
const active = run && !isTerminalRunStatus(run.status) && command && !isTerminalCommandState(command.state) && input.runnerId && run.claimedBy === input.runnerId && heartbeatFresh(run.leaseExpiresAt, input.activeHeartbeatMaxAgeMs);
|
||||
if (active) {
|
||||
return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: true, protectedReason: "fresh-active-run", candidateKind: "inactive", lastActiveAtMs };
|
||||
}
|
||||
const candidateKind: CandidateKind = command && isTerminalCommandState(command.state) && !input.terminal ? "idle" : input.terminal || (run !== null && isTerminalRunStatus(run.status)) ? "terminal" : "inactive";
|
||||
return { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, protectedActive: false, protectedReason: null, candidateKind, lastActiveAtMs };
|
||||
}
|
||||
|
||||
async function kubectlGetList(options: RunnerRetentionOptions, resource: string, selector: string): Promise<K8sObject[]> {
|
||||
const args = ["get", resource, "-n", options.namespace];
|
||||
if (selector) args.push("-l", selector);
|
||||
args.push("-o", "json");
|
||||
const result = await kubectlRun(options.kubectlCommand ?? "kubectl", args);
|
||||
if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl get ${resource} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) });
|
||||
const parsed = parseJsonObject(result.stdout, `kubectl get ${resource}`);
|
||||
const items = parsed.items;
|
||||
return Array.isArray(items) ? items.filter((entry) => typeof entry === "object" && entry !== null && !Array.isArray(entry)).map((entry) => entry as unknown as K8sObject) : [];
|
||||
}
|
||||
|
||||
async function deleteAssociatedResources(options: RunnerRetentionOptions, jobName: string): Promise<number> {
|
||||
let deleted = 0;
|
||||
const selector = `agentrun.pikastech.local/runner-job=${jobName}`;
|
||||
for (const resource of ["secret", "pvc"] as const) {
|
||||
const result = await kubectlRun(options.kubectlCommand ?? "kubectl", ["delete", resource, "-n", options.namespace, "-l", selector, "--ignore-not-found=true"]);
|
||||
if (result.code !== 0) throw new AgentRunError("infra-failed", `kubectl delete associated ${resource} failed with code ${result.code}`, { httpStatus: 502, details: redactJson({ stderr: redactText(result.stderr.slice(-2000)), stdout: redactText(result.stdout.slice(-1000)), valuesPrinted: false }) });
|
||||
deleted += countDeletedLines(result.stdout);
|
||||
}
|
||||
return deleted;
|
||||
}
|
||||
|
||||
async function kubectlRun(kubectlCommand: string, args: string[]): Promise<{ code: number | null; signal: NodeJS.Signals | null; stdout: string; stderr: string }> {
|
||||
const child = spawn(kubectlCommand, args, { stdio: ["ignore", "pipe", "pipe"] });
|
||||
let stdout = "";
|
||||
let stderr = "";
|
||||
child.stdout.setEncoding("utf8");
|
||||
child.stderr.setEncoding("utf8");
|
||||
child.stdout.on("data", (chunk) => { stdout += String(chunk); });
|
||||
child.stderr.on("data", (chunk) => { stderr += String(chunk); });
|
||||
const result = await new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
|
||||
child.on("error", reject);
|
||||
child.on("close", (code, signal) => resolve({ code, signal }));
|
||||
}).catch((error: unknown) => {
|
||||
throw new AgentRunError("infra-failed", `failed to start kubectl: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 503 });
|
||||
});
|
||||
return { ...result, stdout: redactText(stdout), stderr: redactText(stderr) };
|
||||
}
|
||||
|
||||
function compareCandidates(left: RunnerResourceEntry, right: RunnerResourceEntry): number {
|
||||
const priority = (kind: CandidateKind): number => kind === "idle" ? 0 : kind === "terminal" ? 1 : 2;
|
||||
const byKind = priority(left.candidateKind) - priority(right.candidateKind);
|
||||
if (byKind !== 0) return byKind;
|
||||
const byActive = left.lastActiveAtMs - right.lastActiveAtMs;
|
||||
if (byActive !== 0) return byActive;
|
||||
return left.name.localeCompare(right.name);
|
||||
}
|
||||
|
||||
function labelSelector(labels: Record<string, string>): string {
|
||||
return Object.entries(labels).map(([key, value]) => `${key}=${value}`).join(",");
|
||||
}
|
||||
|
||||
function matchesJobPrefix(name: string, prefixes: string[]): boolean {
|
||||
return prefixes.length === 0 || prefixes.some((prefix) => name === prefix || name.startsWith(`${prefix}-`));
|
||||
}
|
||||
|
||||
function podJobName(item: K8sObject): string | null {
|
||||
const labels = item.metadata?.labels ?? {};
|
||||
if (labels["job-name"]) return labels["job-name"];
|
||||
return item.metadata?.ownerReferences?.find((owner) => owner.kind === "Job")?.name ?? null;
|
||||
}
|
||||
|
||||
function jobTerminal(item: K8sObject): boolean {
|
||||
const conditions = item.status?.conditions;
|
||||
if (!Array.isArray(conditions)) return false;
|
||||
return conditions.some((condition) => {
|
||||
const record = condition as JsonRecord;
|
||||
return (record.type === "Complete" || record.type === "Failed") && record.status === "True";
|
||||
});
|
||||
}
|
||||
|
||||
function podTerminal(item: K8sObject): boolean {
|
||||
const phase = stringPath(item, ["status", "phase"]);
|
||||
return phase === "Succeeded" || phase === "Failed";
|
||||
}
|
||||
|
||||
function heartbeatFresh(leaseExpiresAt: string | null, activeHeartbeatMaxAgeMs: number): boolean {
|
||||
const leaseMs = isoMs(leaseExpiresAt);
|
||||
if (leaseMs <= 0) return false;
|
||||
return leaseMs + activeHeartbeatMaxAgeMs >= Date.now();
|
||||
}
|
||||
|
||||
function objectName(item: K8sObject): string | null {
|
||||
return typeof item.metadata?.name === "string" && item.metadata.name.length > 0 ? item.metadata.name : null;
|
||||
}
|
||||
|
||||
function objectCreatedAtMs(item: K8sObject): number {
|
||||
return isoMs(item.metadata?.creationTimestamp) || 0;
|
||||
}
|
||||
|
||||
function isoMs(value: string | null | undefined): number {
|
||||
if (!value) return 0;
|
||||
const parsed = Date.parse(value);
|
||||
return Number.isFinite(parsed) ? parsed : 0;
|
||||
}
|
||||
|
||||
function stringPath(value: unknown, path: string[]): string | null {
|
||||
let current: unknown = value;
|
||||
for (const key of path) {
|
||||
if (typeof current !== "object" || current === null || Array.isArray(current)) return null;
|
||||
current = (current as Record<string, unknown>)[key];
|
||||
}
|
||||
return typeof current === "string" ? current : null;
|
||||
}
|
||||
|
||||
function parseJsonObject(text: string, label: string): JsonRecord {
|
||||
try {
|
||||
const parsed = JSON.parse(text) as JsonValue;
|
||||
if (typeof parsed === "object" && parsed !== null && !Array.isArray(parsed)) return parsed;
|
||||
} catch (error) {
|
||||
throw new AgentRunError("infra-failed", `${label} returned invalid JSON: ${error instanceof Error ? error.message : String(error)}`, { httpStatus: 502, details: { stdoutHash: stableHash(text), valuesPrinted: false } });
|
||||
}
|
||||
throw new AgentRunError("infra-failed", `${label} returned non-object JSON`, { httpStatus: 502, details: { stdoutHash: stableHash(text), valuesPrinted: false } });
|
||||
}
|
||||
|
||||
function countDeletedLines(stdout: string): number {
|
||||
return stdout.split(/\r?\n/u).filter((line) => /\sdeleted$/u.test(line.trim())).length;
|
||||
}
|
||||
Reference in New Issue
Block a user