fix: 修复 AgentRun provider 假活性与容量可见性
This commit is contained in:
Codex
@@ -0,0 +1,311 @@
|
||||
import type { AgentRunLaneSpec } from "../agentrun-lanes";
|
||||
|
||||
import { cleanupRunnersFactsNodeScript } from "./secrets";
|
||||
import { shQuote } from "./utils";
|
||||
|
||||
export interface RunnerJobObservationIdentity {
|
||||
readonly runnerJobId: string;
|
||||
readonly runId: string;
|
||||
readonly commandId: string;
|
||||
readonly runnerId: string;
|
||||
readonly namespace: string;
|
||||
readonly jobName: string;
|
||||
}
|
||||
|
||||
export function runnerJobObservationScript(spec: AgentRunLaneSpec, identity: RunnerJobObservationIdentity): string {
|
||||
const identityJsonB64 = Buffer.from(JSON.stringify(identity), "utf8").toString("base64");
|
||||
return [
|
||||
"set -eu",
|
||||
`namespace=${shQuote(spec.runtime.namespace)}`,
|
||||
`manager_deployment=${shQuote(spec.runtime.managerDeployment)}`,
|
||||
`runner_job_id=${shQuote(identity.runnerJobId)}`,
|
||||
`job_name=${shQuote(identity.jobName)}`,
|
||||
`identity_json_b64=${shQuote(identityJsonB64)}`,
|
||||
"tmp_dir=$(mktemp -d)",
|
||||
"trap 'rm -rf \"$tmp_dir\"' EXIT",
|
||||
"facts_exit=0",
|
||||
"set +e",
|
||||
"kubectl -n \"$namespace\" exec -i deploy/\"$manager_deployment\" -- env RETENTION_NAMESPACE=\"$namespace\" RETENTION_RUNNER_JOB_ID=\"$runner_job_id\" sh -lc 'cat >/tmp/agentrun-runner-observation.mjs && bun /tmp/agentrun-runner-observation.mjs' > \"$tmp_dir/runner-facts.json\" 2> \"$tmp_dir/runner-facts.err\" <<'NODE'",
|
||||
cleanupRunnersFactsNodeScript(),
|
||||
"NODE",
|
||||
"facts_exit=$?",
|
||||
"set -e",
|
||||
"if [ \"$facts_exit\" -ne 0 ]; then printf '%s\\n' '{\"ok\":false,\"items\":[],\"failureKind\":\"manager-facts-unavailable\",\"valuesPrinted\":false}' > \"$tmp_dir/runner-facts.json\"; fi",
|
||||
"job_exit=0",
|
||||
"set +e",
|
||||
"kubectl -n \"$namespace\" get job \"$job_name\" -o json > \"$tmp_dir/job.json\" 2> \"$tmp_dir/job.err\"",
|
||||
"job_exit=$?",
|
||||
"set -e",
|
||||
"if [ \"$job_exit\" -ne 0 ]; then printf '%s\\n' '{}' > \"$tmp_dir/job.json\"; fi",
|
||||
"pods_exit=0",
|
||||
"set +e",
|
||||
"kubectl -n \"$namespace\" get pods -l \"job-name=$job_name\" -o json > \"$tmp_dir/pods.json\" 2> \"$tmp_dir/pods.err\"",
|
||||
"pods_exit=$?",
|
||||
"set -e",
|
||||
"if [ \"$pods_exit\" -ne 0 ]; then printf '%s\\n' '{\"items\":[]}' > \"$tmp_dir/pods.json\"; fi",
|
||||
"node -e 'const p=JSON.parse(require(\"node:fs\").readFileSync(process.argv[1],\"utf8\")); for(const x of Array.isArray(p.items)?p.items.slice(0,8):[]){const n=x?.metadata?.name;if(typeof n===\"string\"&&n)console.log(n)}' \"$tmp_dir/pods.json\" > \"$tmp_dir/pod-names.txt\"",
|
||||
"events_failed=0",
|
||||
"event_index=0",
|
||||
"while IFS= read -r pod_name; do",
|
||||
" [ -n \"$pod_name\" ] || continue",
|
||||
" event_file=\"$tmp_dir/event-$event_index.json\"",
|
||||
" set +e",
|
||||
" kubectl -n \"$namespace\" get events --field-selector \"involvedObject.kind=Pod,involvedObject.name=$pod_name,type=Warning\" -o json > \"$event_file\" 2> \"$tmp_dir/event-$event_index.err\"",
|
||||
" event_exit=$?",
|
||||
" set -e",
|
||||
" if [ \"$event_exit\" -ne 0 ]; then events_failed=$((events_failed + 1)); printf '%s\\n' '{\"items\":[]}' > \"$event_file\"; fi",
|
||||
" event_index=$((event_index + 1))",
|
||||
"done < \"$tmp_dir/pod-names.txt\"",
|
||||
"IDENTITY_JSON_B64=\"$identity_json_b64\" FACTS_EXIT=\"$facts_exit\" JOB_EXIT=\"$job_exit\" PODS_EXIT=\"$pods_exit\" EVENTS_FAILED=\"$events_failed\" TMP_DIR=\"$tmp_dir\" NAMESPACE=\"$namespace\" node <<'NODE'",
|
||||
runnerJobObservationNodeScript(),
|
||||
"NODE",
|
||||
].join("\n");
|
||||
}
|
||||
|
||||
function runnerJobObservationNodeScript(): string {
|
||||
return String.raw`
|
||||
const fs = require("node:fs");
|
||||
const path = require("node:path");
|
||||
const tmp = process.env.TMP_DIR;
|
||||
const expected = JSON.parse(Buffer.from(process.env.IDENTITY_JSON_B64, "base64").toString("utf8"));
|
||||
const factsExit = Number(process.env.FACTS_EXIT);
|
||||
const jobExit = Number(process.env.JOB_EXIT);
|
||||
const podsExit = Number(process.env.PODS_EXIT);
|
||||
const eventsFailed = Number(process.env.EVENTS_FAILED);
|
||||
|
||||
function readJson(name) {
|
||||
return JSON.parse(fs.readFileSync(path.join(tmp, name), "utf8"));
|
||||
}
|
||||
|
||||
function iso(value) {
|
||||
if (typeof value !== "string" || value.length === 0) return null;
|
||||
const ms = Date.parse(value);
|
||||
return Number.isFinite(ms) ? new Date(ms).toISOString() : null;
|
||||
}
|
||||
|
||||
function ageMs(value) {
|
||||
const ms = Date.parse(value || "");
|
||||
return Number.isFinite(ms) ? Math.max(0, Date.now() - ms) : null;
|
||||
}
|
||||
|
||||
function bounded(value, max = 320) {
|
||||
if (typeof value !== "string" || value.length === 0) return null;
|
||||
const oneLine = value.replace(/\s+/g, " ").trim();
|
||||
return oneLine.length <= max ? oneLine : oneLine.slice(0, max);
|
||||
}
|
||||
|
||||
function annotationsOf(value) {
|
||||
return value?.metadata?.annotations && typeof value.metadata.annotations === "object" ? value.metadata.annotations : {};
|
||||
}
|
||||
|
||||
function stateKind(state) {
|
||||
if (state?.running) return "running";
|
||||
if (state?.terminated) return "terminated";
|
||||
if (state?.waiting) return "waiting";
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
function waitingFromStatus(status, source) {
|
||||
const waiting = status?.state?.waiting;
|
||||
if (!waiting || typeof waiting.reason !== "string" || waiting.reason.length === 0) return [];
|
||||
return [{
|
||||
source,
|
||||
name: typeof status.name === "string" ? status.name : null,
|
||||
reason: waiting.reason,
|
||||
message: bounded(waiting.message),
|
||||
valuesPrinted: false,
|
||||
}];
|
||||
}
|
||||
|
||||
function warningFromEvent(event) {
|
||||
if (event?.type !== "Warning" || typeof event?.reason !== "string" || event.reason.length === 0) return null;
|
||||
return {
|
||||
source: "event",
|
||||
name: typeof event?.metadata?.name === "string" ? event.metadata.name : null,
|
||||
reason: event.reason,
|
||||
message: bounded(event.message),
|
||||
count: Number.isFinite(event.count) ? event.count : null,
|
||||
lastObservedAt: iso(event?.series?.lastObservedTime || event?.eventTime || event?.lastTimestamp || event?.firstTimestamp),
|
||||
valuesPrinted: false,
|
||||
};
|
||||
}
|
||||
|
||||
function identityMatches(value) {
|
||||
const annotations = annotationsOf(value);
|
||||
const optionalRunnerJobId = annotations["agentrun.pikastech.local/runner-job-id"];
|
||||
return annotations["agentrun.pikastech.local/run-id"] === expected.runId
|
||||
&& annotations["agentrun.pikastech.local/command-id"] === expected.commandId
|
||||
&& annotations["agentrun.pikastech.local/runner-id"] === expected.runnerId
|
||||
&& (optionalRunnerJobId === undefined || optionalRunnerJobId === expected.runnerJobId);
|
||||
}
|
||||
|
||||
const facts = readJson("runner-facts.json");
|
||||
const factItems = Array.isArray(facts.items) ? facts.items : [];
|
||||
const fact = factItems.find((item) => item?.id === expected.runnerJobId) || null;
|
||||
const job = readJson("job.json");
|
||||
const podsPayload = readJson("pods.json");
|
||||
const pods = Array.isArray(podsPayload.items) ? podsPayload.items.slice(0, 8) : [];
|
||||
const eventDocuments = fs.readdirSync(tmp)
|
||||
.filter((name) => /^event-\d+\.json$/u.test(name))
|
||||
.sort()
|
||||
.map(readJson);
|
||||
const warningEvents = eventDocuments.flatMap((value) => Array.isArray(value.items) ? value.items : []);
|
||||
const warningsByPod = new Map();
|
||||
for (const event of warningEvents) {
|
||||
const podName = event?.involvedObject?.kind === "Pod" && typeof event?.involvedObject?.name === "string"
|
||||
? event.involvedObject.name
|
||||
: null;
|
||||
if (podName === null) continue;
|
||||
const items = warningsByPod.get(podName) || [];
|
||||
items.push(event);
|
||||
warningsByPod.set(podName, items);
|
||||
}
|
||||
|
||||
const podObservations = pods.map((pod) => {
|
||||
const statuses = Array.isArray(pod?.status?.containerStatuses) ? pod.status.containerStatuses : [];
|
||||
const initStatuses = Array.isArray(pod?.status?.initContainerStatuses) ? pod.status.initContainerStatuses : [];
|
||||
const runner = statuses.find((status) => status?.name === "runner") || null;
|
||||
const conditions = Array.isArray(pod?.status?.conditions) ? pod.status.conditions : [];
|
||||
const eventWarnings = (warningsByPod.get(pod?.metadata?.name || "") || []).map(warningFromEvent).filter(Boolean);
|
||||
const waiting = [
|
||||
...statuses.flatMap((status) => waitingFromStatus(status, "container")),
|
||||
...initStatuses.flatMap((status) => waitingFromStatus(status, "init-container")),
|
||||
...conditions.flatMap((condition) => condition?.status === "False" && typeof condition?.reason === "string" && condition.reason.length > 0
|
||||
? [{ source: "condition", name: condition.type || null, reason: condition.reason, message: bounded(condition.message), valuesPrinted: false }]
|
||||
: []),
|
||||
...eventWarnings,
|
||||
];
|
||||
const ready = conditions.some((condition) => condition?.type === "Ready" && condition?.status === "True");
|
||||
return {
|
||||
name: pod?.metadata?.name || null,
|
||||
uid: pod?.metadata?.uid || null,
|
||||
resourceVersion: pod?.metadata?.resourceVersion || null,
|
||||
phase: pod?.status?.phase || null,
|
||||
createdAt: iso(pod?.metadata?.creationTimestamp),
|
||||
ageMs: ageMs(pod?.metadata?.creationTimestamp),
|
||||
ready,
|
||||
identityMatches: identityMatches(pod),
|
||||
runnerContainerState: stateKind(runner?.state || {}),
|
||||
runnerContainerReady: runner?.ready === true,
|
||||
runnerContainerRestartCount: Number(runner?.restartCount || 0),
|
||||
waiting: waiting.slice(0, 8),
|
||||
waitingOmittedCount: Math.max(0, waiting.length - 8),
|
||||
eventFactsAvailable: eventsFailed === 0,
|
||||
valuesPrinted: false,
|
||||
};
|
||||
});
|
||||
|
||||
const managerIdentityMatches = fact !== null
|
||||
&& fact.runId === expected.runId
|
||||
&& fact.commandId === expected.commandId
|
||||
&& fact.runnerId === expected.runnerId
|
||||
&& fact.namespace === expected.namespace
|
||||
&& fact.jobName === expected.jobName;
|
||||
const jobPresent = jobExit === 0 && job?.metadata?.name === expected.jobName;
|
||||
const jobIdentityMatches = jobPresent && identityMatches(job);
|
||||
const podIdentityMatches = podObservations.length > 0 && podObservations.every((pod) => pod.identityMatches === true);
|
||||
const podPhases = podObservations.map((pod) => pod.phase).filter((value) => typeof value === "string");
|
||||
const waitingReasons = [...new Set(podObservations.flatMap((pod) => pod.waiting.map((item) => item.reason)).filter(Boolean))].slice(0, 8);
|
||||
const state = podPhases.includes("Pending")
|
||||
? "Pending"
|
||||
: podPhases.includes("Running")
|
||||
? "Running"
|
||||
: podPhases.includes("Failed")
|
||||
? "Failed"
|
||||
: podPhases.includes("Succeeded")
|
||||
? "Succeeded"
|
||||
: jobPresent
|
||||
? "Unknown"
|
||||
: "NotFound";
|
||||
const classification = state === "Pending" && waitingReasons.includes("FailedMount")
|
||||
? "pending-failed-mount"
|
||||
: state.toLowerCase();
|
||||
const warningEventsAvailable = podObservations.length > 0 && eventsFailed === 0;
|
||||
const ok = factsExit === 0
|
||||
&& facts.ok === true
|
||||
&& factItems.length === 1
|
||||
&& managerIdentityMatches
|
||||
&& jobPresent
|
||||
&& jobIdentityMatches
|
||||
&& podsExit === 0
|
||||
&& podIdentityMatches
|
||||
&& warningEventsAvailable;
|
||||
const reason = ok
|
||||
? null
|
||||
: factsExit !== 0 || facts.ok !== true
|
||||
? "manager-runner-facts-query-failed"
|
||||
: factItems.length !== 1 || !managerIdentityMatches
|
||||
? "manager-runner-identity-mismatch"
|
||||
: !jobPresent
|
||||
? "runner-job-not-found"
|
||||
: !jobIdentityMatches
|
||||
? "runner-job-identity-mismatch"
|
||||
: podsExit !== 0 || !podIdentityMatches
|
||||
? "runner-pod-observation-incomplete"
|
||||
: "runner-warning-event-observation-incomplete";
|
||||
|
||||
console.log(JSON.stringify({
|
||||
ok,
|
||||
reason,
|
||||
observedAt: new Date().toISOString(),
|
||||
target: { namespace: process.env.NAMESPACE },
|
||||
managerFacts: {
|
||||
ok: factsExit === 0 && facts.ok === true && factItems.length === 1,
|
||||
queryMode: facts.queryMode || null,
|
||||
runnerJobId: expected.runnerJobId,
|
||||
itemCount: factItems.length,
|
||||
identityMatches: managerIdentityMatches,
|
||||
valuesPrinted: false,
|
||||
},
|
||||
kubernetesFacts: {
|
||||
jobQuery: "exact-name",
|
||||
podQuery: "exact-job-name-label",
|
||||
warningEventQuery: "exact-pod-name",
|
||||
jobPresent,
|
||||
podCount: podObservations.length,
|
||||
warningEventCount: warningEvents.length,
|
||||
warningEventsAvailable,
|
||||
valuesPrinted: false,
|
||||
},
|
||||
runner: {
|
||||
name: expected.jobName,
|
||||
namespace: expected.namespace,
|
||||
runId: expected.runId,
|
||||
commandId: expected.commandId,
|
||||
runnerJobId: expected.runnerJobId,
|
||||
runnerId: expected.runnerId,
|
||||
resourceUid: job?.metadata?.uid || null,
|
||||
resourceVersion: job?.metadata?.resourceVersion || null,
|
||||
createdAt: iso(job?.metadata?.creationTimestamp),
|
||||
lastActiveAt: fact?.runnerHeartbeatAt || fact?.runUpdatedAt || fact?.runnerJobUpdatedAt || null,
|
||||
lastActiveAgeMs: ageMs(fact?.runnerHeartbeatAt || fact?.runUpdatedAt || fact?.runnerJobUpdatedAt),
|
||||
runStatus: fact?.runStatus || null,
|
||||
runClaimedBy: fact?.runClaimedBy || null,
|
||||
commandState: fact?.commandState || null,
|
||||
heartbeatAt: fact?.runnerHeartbeatAt || null,
|
||||
heartbeatAgeMs: ageMs(fact?.runnerHeartbeatAt),
|
||||
jobStatus: {
|
||||
active: Number(job?.status?.active || 0),
|
||||
succeeded: Number(job?.status?.succeeded || 0),
|
||||
failed: Number(job?.status?.failed || 0),
|
||||
ready: Number(job?.status?.ready || 0),
|
||||
},
|
||||
podPhases,
|
||||
waitingReasons,
|
||||
podObservations: podObservations.slice(0, 4),
|
||||
podObservationOmittedCount: Math.max(0, podObservations.length - 4),
|
||||
identityComplete: managerIdentityMatches && jobIdentityMatches && podIdentityMatches,
|
||||
managerFactsComplete: factsExit === 0 && facts.ok === true && factItems.length === 1,
|
||||
podFactsComplete: podsExit === 0 && podIdentityMatches && warningEventsAvailable,
|
||||
state,
|
||||
classification,
|
||||
cleanupAuthority: "agentrun-manager-pre-create-retention",
|
||||
cleanupDecisionComputed: false,
|
||||
mutation: false,
|
||||
valuesPrinted: false,
|
||||
},
|
||||
mutation: false,
|
||||
valuesPrinted: false,
|
||||
}));
|
||||
`;
|
||||
}
|
||||
Reference in New Issue
Block a user