lyon
365 lines
16 KiB
TypeScript
365 lines
16 KiB
TypeScript
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, 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;
|
|
}
|