fix: reconcile runner job observations
This commit is contained in:
lyon
@@ -0,0 +1,314 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { AgentRunError } from "../common/errors.js";
|
||||
import { redactJson, redactText } from "../common/redaction.js";
|
||||
import type { CommandRecord, FailureKind, JsonRecord, JsonValue, RunnerJobRecord, TerminalStatus } from "../common/types.js";
|
||||
import { nowIso, stableHash } from "../common/validation.js";
|
||||
import type { AgentRunStore } from "./store.js";
|
||||
import { isTerminalCommandState, isTerminalRunStatus } from "./store.js";
|
||||
|
||||
export interface RunnerReconcilerOptions {
|
||||
store: AgentRunStore;
|
||||
namespace?: string;
|
||||
kubectlCommand?: string;
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
export interface RunnerReconcilerLoopOptions extends RunnerReconcilerOptions {
|
||||
batchSize: number;
|
||||
intervalMs: number;
|
||||
onError?: (error: unknown) => void;
|
||||
}
|
||||
|
||||
interface K8sObject {
|
||||
kind?: string;
|
||||
metadata?: {
|
||||
name?: string;
|
||||
namespace?: string;
|
||||
uid?: string;
|
||||
creationTimestamp?: string;
|
||||
labels?: Record<string, string>;
|
||||
annotations?: Record<string, string>;
|
||||
ownerReferences?: Array<{ kind?: string; name?: string }>;
|
||||
};
|
||||
status?: JsonRecord;
|
||||
}
|
||||
|
||||
interface K8sList {
|
||||
items?: K8sObject[];
|
||||
}
|
||||
|
||||
export async function reconcileRunnerJobsOnce(input: RunnerReconcilerOptions): Promise<JsonRecord> {
|
||||
const limit = clampLimit(input.limit ?? 20);
|
||||
const jobs = await input.store.listRunnerJobsForReconciliation(limit);
|
||||
const items: JsonRecord[] = [];
|
||||
let observeFailedCount = 0;
|
||||
let runClosureCount = 0;
|
||||
|
||||
for (const job of jobs) {
|
||||
const observation = await observeRunnerJob(job, { ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
|
||||
await input.store.updateRunnerJobResult(job.id, { observation });
|
||||
if (stringValue(observation.category) === "runner-job-observe-failed") observeFailedCount++;
|
||||
const runClosure = await closeOpenRunWhenCommandTerminal(input.store, job);
|
||||
if (runClosure.closed === true) runClosureCount++;
|
||||
items.push({
|
||||
runnerJobId: job.id,
|
||||
runId: job.runId,
|
||||
commandId: job.commandId,
|
||||
namespace: stringValue(observation.namespace) ?? job.namespace,
|
||||
jobName: job.jobName,
|
||||
observedRunnerPhase: stringValue(observation.observedRunnerPhase) ?? "unknown",
|
||||
terminalReportState: stringValue(observation.terminalReportState) ?? "unknown",
|
||||
runReportState: stringValue(observation.runReportState) ?? "unknown",
|
||||
runClosure,
|
||||
valuesPrinted: false,
|
||||
});
|
||||
}
|
||||
|
||||
return {
|
||||
action: "runner-job-reconcile",
|
||||
reconciledAt: nowIso(),
|
||||
limit,
|
||||
scannedCount: jobs.length,
|
||||
updatedCount: jobs.length,
|
||||
observeFailedCount,
|
||||
runClosureCount,
|
||||
items,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
export function startRunnerJobReconciler(input: RunnerReconcilerLoopOptions): () => void {
|
||||
const intervalMs = Math.max(1_000, input.intervalMs);
|
||||
let stopped = false;
|
||||
let running = false;
|
||||
const tick = async (): Promise<void> => {
|
||||
if (stopped || running) return;
|
||||
running = true;
|
||||
try {
|
||||
await reconcileRunnerJobsOnce({ store: input.store, limit: input.batchSize, ...(input.namespace ? { namespace: input.namespace } : {}), ...(input.kubectlCommand ? { kubectlCommand: input.kubectlCommand } : {}) });
|
||||
} catch (error) {
|
||||
input.onError?.(error);
|
||||
} finally {
|
||||
running = false;
|
||||
}
|
||||
};
|
||||
const timer = setInterval(() => { void tick(); }, intervalMs);
|
||||
timer.unref?.();
|
||||
void tick();
|
||||
return () => {
|
||||
stopped = true;
|
||||
clearInterval(timer);
|
||||
};
|
||||
}
|
||||
|
||||
async function observeRunnerJob(job: RunnerJobRecord, input: { namespace?: string; kubectlCommand?: string }): Promise<JsonRecord> {
|
||||
const namespace = input.namespace ?? job.namespace;
|
||||
const kubectlCommand = input.kubectlCommand ?? "kubectl";
|
||||
const observedAt = nowIso();
|
||||
try {
|
||||
const [jobObject, podObjects] = await Promise.all([
|
||||
getK8sObject(kubectlCommand, "job", namespace, job.jobName),
|
||||
getK8sList(kubectlCommand, "pods", namespace, ["-l", `job-name=${job.jobName}`]),
|
||||
]);
|
||||
return observationFromObjects(job, namespace, observedAt, jobObject, podObjects);
|
||||
} catch (error) {
|
||||
return {
|
||||
source: "manager-reconciler",
|
||||
namespace,
|
||||
jobName: job.jobName,
|
||||
observedRunnerPhase: "k8s:observe-failed",
|
||||
category: "runner-job-observe-failed",
|
||||
terminalStatus: null,
|
||||
failureKind: "infra-failed",
|
||||
failureMessage: error instanceof Error ? redactText(error.message).slice(0, 300) : "unknown observation failure",
|
||||
terminalReportState: "k8s-observation-failed",
|
||||
runReportState: "not-updated",
|
||||
lastK8sObservedAt: observedAt,
|
||||
evidenceLevel: "medium",
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
function observationFromObjects(job: RunnerJobRecord, namespace: string, observedAt: string, jobObject: K8sObject | null, podObjects: K8sObject[]): JsonRecord {
|
||||
const phase = observedRunnerPhase(jobObject, podObjects);
|
||||
const terminalReportState = terminalReportStateForPhase(phase);
|
||||
const failureKind: FailureKind | null = phase === "k8s:failed" || phase === "k8s:missing" ? "infra-failed" : null;
|
||||
const k8s = k8sSummary(jobObject, podObjects);
|
||||
return {
|
||||
source: "manager-reconciler",
|
||||
namespace,
|
||||
jobName: job.jobName,
|
||||
observedRunnerPhase: phase,
|
||||
phase,
|
||||
category: categoryForPhase(phase),
|
||||
terminalStatus: null,
|
||||
failureKind,
|
||||
terminalReportState,
|
||||
runReportState: "not-updated",
|
||||
lastK8sObservedAt: observedAt,
|
||||
lastObservedAt: observedAt,
|
||||
lastObservedKind: `manager-reconciler:${phase}`,
|
||||
evidenceLevel: phase === "k8s:unknown" ? "medium" : "high",
|
||||
k8s,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
async function closeOpenRunWhenCommandTerminal(store: AgentRunStore, job: RunnerJobRecord): Promise<JsonRecord> {
|
||||
let command: CommandRecord;
|
||||
try {
|
||||
command = await store.getCommand(job.commandId);
|
||||
} catch {
|
||||
return { closed: false, state: "command-missing", valuesPrinted: false };
|
||||
}
|
||||
if (!isTerminalCommandState(command.state)) return { closed: false, state: "command-open", commandState: command.state, valuesPrinted: false };
|
||||
const run = await store.getRun(job.runId);
|
||||
if (isTerminalRunStatus(run.status)) return { closed: false, state: "run-terminal", runStatus: run.status, commandState: command.state, valuesPrinted: false };
|
||||
const terminalStatus = terminalStatusFromCommand(command);
|
||||
const failureKind: FailureKind | null = terminalStatus === "cancelled" ? "cancelled" : terminalStatus === "failed" ? "infra-failed" : null;
|
||||
const failureMessage = terminalStatus === "completed" ? null : "manager reconciler closed open run after terminal command state";
|
||||
const next = await store.finishRun(run.id, { terminalStatus, failureKind, failureMessage });
|
||||
return { closed: true, state: "closed-open-run", runStatus: next.status, terminalStatus, commandState: command.state, valuesPrinted: false };
|
||||
}
|
||||
|
||||
function terminalStatusFromCommand(command: CommandRecord): TerminalStatus {
|
||||
if (command.state === "completed") return "completed";
|
||||
if (command.state === "cancelled") return "cancelled";
|
||||
return "failed";
|
||||
}
|
||||
|
||||
function observedRunnerPhase(jobObject: K8sObject | null, pods: K8sObject[]): string {
|
||||
if (!jobObject && pods.length === 0) return "k8s:missing";
|
||||
const condition = jobCondition(jobObject);
|
||||
if (condition === "Complete") return "k8s:succeeded";
|
||||
if (condition === "Failed") return "k8s:failed";
|
||||
const active = numberPath(jobObject, ["status", "active"]);
|
||||
const succeeded = numberPath(jobObject, ["status", "succeeded"]);
|
||||
const failed = numberPath(jobObject, ["status", "failed"]);
|
||||
const podPhases = pods.map((pod) => stringPath(pod, ["status", "phase"])).filter((phase): phase is string => Boolean(phase));
|
||||
if ((succeeded ?? 0) > 0 || podPhases.some((phase) => phase === "Succeeded")) return "k8s:succeeded";
|
||||
if ((failed ?? 0) > 0 || podPhases.some((phase) => phase === "Failed")) return "k8s:failed";
|
||||
if ((active ?? 0) > 0 || podPhases.some((phase) => phase === "Running")) return "k8s:running";
|
||||
if (podPhases.some((phase) => phase === "Pending")) return "k8s:pending";
|
||||
return "k8s:unknown";
|
||||
}
|
||||
|
||||
function categoryForPhase(phase: string): string {
|
||||
if (phase === "k8s:succeeded") return "runner-job-k8s-succeeded";
|
||||
if (phase === "k8s:failed") return "runner-job-k8s-failed";
|
||||
if (phase === "k8s:missing") return "runner-job-k8s-missing";
|
||||
if (phase === "k8s:running" || phase === "k8s:pending") return "runner-job-running";
|
||||
return "runner-job-observed";
|
||||
}
|
||||
|
||||
function terminalReportStateForPhase(phase: string): string {
|
||||
if (phase === "k8s:succeeded" || phase === "k8s:failed") return "terminal-runner-report-missing";
|
||||
if (phase === "k8s:missing") return "runner-resource-missing";
|
||||
return "not-terminal";
|
||||
}
|
||||
|
||||
function k8sSummary(jobObject: K8sObject | null, pods: K8sObject[]): JsonRecord {
|
||||
const podPhases = pods.map((pod) => stringPath(pod, ["status", "phase"]) ?? "unknown");
|
||||
return {
|
||||
jobPresent: jobObject !== null,
|
||||
jobUid: stringPath(jobObject, ["metadata", "uid"]),
|
||||
condition: jobCondition(jobObject),
|
||||
active: numberPath(jobObject, ["status", "active"]),
|
||||
succeeded: numberPath(jobObject, ["status", "succeeded"]),
|
||||
failed: numberPath(jobObject, ["status", "failed"]),
|
||||
startTime: stringPath(jobObject, ["status", "startTime"]),
|
||||
completionTime: stringPath(jobObject, ["status", "completionTime"]),
|
||||
podCount: pods.length,
|
||||
podPhases,
|
||||
newestPodName: newestPodName(pods),
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function jobCondition(jobObject: K8sObject | null): string | null {
|
||||
const conditions = jobObject?.status?.conditions;
|
||||
if (!Array.isArray(conditions)) return null;
|
||||
for (const condition of [...conditions].reverse()) {
|
||||
if (typeof condition !== "object" || condition === null || Array.isArray(condition)) continue;
|
||||
const record = condition as JsonRecord;
|
||||
if ((record.type === "Complete" || record.type === "Failed") && record.status === "True") return record.type;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function newestPodName(pods: K8sObject[]): string | null {
|
||||
const sorted = [...pods].sort((left, right) => (stringPath(right, ["metadata", "creationTimestamp"]) ?? "").localeCompare(stringPath(left, ["metadata", "creationTimestamp"]) ?? ""));
|
||||
return stringPath(sorted[0] ?? null, ["metadata", "name"]);
|
||||
}
|
||||
|
||||
async function getK8sObject(kubectlCommand: string, resource: string, namespace: string, name: string): Promise<K8sObject | null> {
|
||||
const result = await kubectlRun(kubectlCommand, ["get", resource, name, "-n", namespace, "-o", "json"]);
|
||||
if (result.code === 0) return parseJsonObject(result.stdout, `kubectl get ${resource}`) as K8sObject;
|
||||
if (isNotFound(result.stderr)) return null;
|
||||
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 }) });
|
||||
}
|
||||
|
||||
async function getK8sList(kubectlCommand: string, resource: string, namespace: string, extraArgs: string[]): Promise<K8sObject[]> {
|
||||
const result = await kubectlRun(kubectlCommand, ["get", resource, "-n", namespace, ...extraArgs, "-o", "json"]);
|
||||
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}`) as K8sList;
|
||||
return Array.isArray(parsed.items) ? parsed.items.filter((item) => typeof item === "object" && item !== null && !Array.isArray(item)) : [];
|
||||
}
|
||||
|
||||
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 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 isNotFound(stderr: string): boolean {
|
||||
return /notfound|not found|notfound/i.test(stderr.replace(/\s+/gu, ""));
|
||||
}
|
||||
|
||||
function clampLimit(limit: number): number {
|
||||
return Math.max(1, Math.min(Math.floor(limit), 500));
|
||||
}
|
||||
|
||||
function stringValue(value: JsonValue | undefined): string | null {
|
||||
return typeof value === "string" && value.length > 0 ? value : null;
|
||||
}
|
||||
|
||||
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 numberPath(value: unknown, path: string[]): number | 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 === "number" && Number.isFinite(current) ? current : null;
|
||||
}
|
||||
Reference in New Issue
Block a user