fix: sync follower sources before native status
This commit is contained in:
Codex
@@ -0,0 +1,145 @@
|
||||
// SPEC: PJ2026-01060703 CI/CD branch follower native object bundle reader.
|
||||
// Responsibility: read compact Kubernetes-native source/Tekton/Argo/runtime state through file-backed scripts.
|
||||
import { readFileSync } from "node:fs";
|
||||
import { rootPath } from "./config";
|
||||
import type { CommandResult } from "./command";
|
||||
import { resolveAgentRunLaneTarget } from "./agentrun-lanes";
|
||||
import type { BranchFollowerRegistry, FollowerSpec, NativeObjectBundle, ParsedOptions } from "./cicd-types";
|
||||
import { hwlabRuntimeLaneSpecForNode } from "./hwlab-node-lanes";
|
||||
import { redactText, shQuote } from "./platform-infra-ops-library";
|
||||
|
||||
type KubeScriptRunner = (registry: BranchFollowerRegistry, options: ParsedOptions, script: string, input: string, timeoutMs: number) => CommandResult;
|
||||
|
||||
const NATIVE_BUNDLE_SCRIPT_NAMES = [
|
||||
"read-native-bundle.sh",
|
||||
"kube-get.mjs",
|
||||
"compact-native-object.mjs",
|
||||
"compact-git-mirror.mjs",
|
||||
"plan-artifacts.mjs",
|
||||
] as const;
|
||||
|
||||
export function readNativeObjectBundle(registry: BranchFollowerRegistry, follower: FollowerSpec, options: ParsedOptions, timeoutSeconds: number, runKubeScript: KubeScriptRunner): NativeObjectBundle {
|
||||
const native = follower.nativeStatus;
|
||||
const source = native.source;
|
||||
const tekton = native.tekton;
|
||||
const argo = native.argo;
|
||||
const runtime = native.runtime;
|
||||
const gitopsBranch = nativeGitMirrorGitopsBranch(follower);
|
||||
const workloadRefs = (runtime?.workloads ?? []).map((workload, index) => {
|
||||
const resource = workload.kind === "Deployment" ? "deployments" : "statefulsets";
|
||||
return `workload${index}\t/apis/apps/v1/namespaces/${runtime?.namespace ?? follower.target.namespace}/${resource}/${workload.name}`;
|
||||
});
|
||||
const workloadRefsText = workloadRefs.length === 0 ? "" : `${workloadRefs.join("\n")}\n`;
|
||||
const script = [
|
||||
"set +e",
|
||||
"tmpdir=$(mktemp -d)",
|
||||
"cleanup() { rm -rf \"$tmpdir\"; }",
|
||||
"trap cleanup EXIT INT TERM",
|
||||
nativeBundleScriptLoadShell(),
|
||||
`REPO_PATH=${shQuote(source.repoPath)}`,
|
||||
`SOURCE_BRANCH=${shQuote(follower.source.branch)}`,
|
||||
`REPOSITORY=${shQuote(follower.source.repository)}`,
|
||||
`SNAPSHOT_PREFIX=${shQuote(follower.source.snapshotPrefix)}`,
|
||||
`GITOPS_BRANCH=${shQuote(gitopsBranch ?? "")}`,
|
||||
`TEKTON_NAMESPACE=${shQuote(tekton?.namespace ?? "")}`,
|
||||
`PIPELINE_RUN_PREFIX=${shQuote(tekton?.pipelineRunPrefix ?? "")}`,
|
||||
`ARGO_NAMESPACE=${shQuote(argo?.namespace ?? "")}`,
|
||||
`ARGO_APPLICATION=${shQuote(argo?.application ?? "")}`,
|
||||
`WORKLOAD_REFS_B64=${shQuote(Buffer.from(workloadRefsText, "utf8").toString("base64"))}`,
|
||||
"NATIVE_CICD_SCRIPT_DIR=\"$tmpdir\"",
|
||||
"export NATIVE_CICD_SCRIPT_DIR REPO_PATH SOURCE_BRANCH REPOSITORY SNAPSHOT_PREFIX GITOPS_BRANCH TEKTON_NAMESPACE PIPELINE_RUN_PREFIX ARGO_NAMESPACE ARGO_APPLICATION WORKLOAD_REFS_B64",
|
||||
"\"$tmpdir/read-native-bundle.sh\"",
|
||||
].join("\n");
|
||||
const startedAt = Date.now();
|
||||
const result = runKubeScript(registry, options, script, "", timeoutSeconds * 1000);
|
||||
const parsed = parseNativeBundleLines(result.stdout);
|
||||
const sourceRecord = asOptionalRecord(parsed.objects.source);
|
||||
return {
|
||||
ok: result.exitCode === 0 && sourceRecord !== null && parsed.fatalErrors.length === 0,
|
||||
source: sourceRecord,
|
||||
gitMirror: asOptionalRecord(parsed.objects.gitMirror),
|
||||
pipelineRun: asOptionalRecord(parsed.objects.pipelineRun),
|
||||
taskRuns: asOptionalRecord(parsed.objects.taskRuns),
|
||||
planArtifacts: asOptionalRecord(parsed.objects.planArtifacts),
|
||||
argoApplication: asOptionalRecord(parsed.objects.argoApplication),
|
||||
workloads: Object.entries(parsed.objects)
|
||||
.filter(([key]) => /^workload\d+$/u.test(key))
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([, value]) => asOptionalRecord(value))
|
||||
.filter((item): item is Record<string, unknown> => item !== null),
|
||||
errors: [
|
||||
...parsed.errors,
|
||||
...(result.exitCode === 0 ? [] : [`native bundle command failed: exitCode=${result.exitCode}`]),
|
||||
...parsed.fatalErrors,
|
||||
],
|
||||
exitCode: result.exitCode,
|
||||
timedOut: result.timedOut,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
stdoutTail: redactText(tailText(result.stdout, 1000)),
|
||||
stderrTail: redactText(tailText(result.stderr, 1000)),
|
||||
};
|
||||
}
|
||||
|
||||
function nativeBundleScriptLoadShell(): string {
|
||||
return nativeScriptLoadShell(NATIVE_BUNDLE_SCRIPT_NAMES);
|
||||
}
|
||||
|
||||
function nativeScriptLoadShell(names: readonly string[]): string {
|
||||
return names.map((name) => {
|
||||
const encoded = Buffer.from(readFileSync(rootPath("scripts/native/cicd", name), "utf8"), "utf8").toString("base64");
|
||||
return [
|
||||
`printf '%s' ${shQuote(encoded)} | base64 -d > "$tmpdir/${name}"`,
|
||||
`chmod +x "$tmpdir/${name}"`,
|
||||
].join("\n");
|
||||
}).join("\n");
|
||||
}
|
||||
|
||||
function parseNativeBundleLines(stdout: string): { objects: Record<string, unknown>; errors: string[]; fatalErrors: string[] } {
|
||||
const objects: Record<string, unknown> = {};
|
||||
const errors: string[] = [];
|
||||
const fatalErrors: string[] = [];
|
||||
for (const line of stdout.split(/\r?\n/u)) {
|
||||
if (!line.startsWith("UNIDESK_NATIVE_")) continue;
|
||||
const [kind, key, payload] = line.split("\t");
|
||||
if (kind === undefined || key === undefined || payload === undefined) continue;
|
||||
const decoded = Buffer.from(payload, "base64").toString("utf8").trim();
|
||||
if (kind === "UNIDESK_NATIVE_JSON") {
|
||||
const parsed = parseJsonObject(decoded);
|
||||
if (parsed !== null) objects[key] = parsed;
|
||||
else errors.push(`${key}: invalid native JSON payload`);
|
||||
} else if (kind === "UNIDESK_NATIVE_ERROR") {
|
||||
const message = `${key}: ${redactText(tailText(decoded, 500)) || "not found"}`;
|
||||
errors.push(message);
|
||||
if (key === "source") fatalErrors.push(message);
|
||||
}
|
||||
}
|
||||
return { objects, errors, fatalErrors };
|
||||
}
|
||||
|
||||
function nativeGitMirrorGitopsBranch(follower: FollowerSpec): string | null {
|
||||
if (follower.adapter === "hwlab-node-runtime") {
|
||||
return hwlabRuntimeLaneSpecForNode(follower.target.lane, follower.target.node).gitopsBranch;
|
||||
}
|
||||
if (follower.adapter === "agentrun-yaml-lane") {
|
||||
return resolveAgentRunLaneTarget({ node: follower.target.node, lane: follower.target.lane }).spec.gitops.branch;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function parseJsonObject(text: string): Record<string, unknown> | null {
|
||||
const trimmed = text.trim();
|
||||
if (trimmed.length === 0) return null;
|
||||
try {
|
||||
return asOptionalRecord(JSON.parse(trimmed));
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function asOptionalRecord(value: unknown): Record<string, unknown> | null {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : null;
|
||||
}
|
||||
|
||||
function tailText(text: string, maxChars: number): string {
|
||||
return text.length <= maxChars ? text : text.slice(text.length - maxChars);
|
||||
}
|
||||
+22
-112
@@ -22,8 +22,9 @@ import { transPath } from "./hwlab-node/runtime-common";
|
||||
import { configRefGraph, resolveConfigRefString } from "./ops/config-refs";
|
||||
import { renderControllerManifests, renderControllerReconcileJob, waitForJobShell } from "./cicd-controller-render";
|
||||
import { runNativeHwlabControlPlaneRefresh } from "./cicd-hwlab-refresh";
|
||||
import { readNativeObjectBundle } from "./cicd-native-bundle";
|
||||
import { runNativeK8sJob, runNativeTektonPipelineRun } from "./cicd-native";
|
||||
import type { AdapterSummary, BranchFollowerPhase, BranchFollowerRegistry, ControllerSpec, FollowerSpec, FollowerState, K8sFollowerStateRead, K8sStateRead, NativeCloseoutWaitResult, NativeK8sJobResult, NativeObjectBundle, NativeStatusSpec, NativeWorkloadSpec, OutputMode, ParsedOptions, StageTiming, TriggerResult } from "./cicd-types";
|
||||
import type { AdapterSummary, BranchFollowerPhase, BranchFollowerRegistry, ControllerSpec, FollowerSpec, FollowerState, K8sFollowerStateRead, K8sStateRead, NativeCloseoutWaitResult, NativeK8sJobResult, NativeStatusSpec, NativeWorkloadSpec, OutputMode, ParsedOptions, StageTiming, TriggerResult } from "./cicd-types";
|
||||
import {
|
||||
arrayField,
|
||||
asRecord,
|
||||
@@ -1472,8 +1473,12 @@ function runNativeArgoRefresh(argo: NonNullable<NativeStatusSpec["argo"]>): Comm
|
||||
|
||||
async function readAdapterStatus(registry: BranchFollowerRegistry, follower: FollowerSpec, options: ParsedOptions): Promise<AdapterSummary> {
|
||||
const timeoutSeconds = options.timeoutSeconds ?? follower.budgets.statusSeconds;
|
||||
const bundle = readNativeObjectBundle(registry, follower, options, timeoutSeconds);
|
||||
const observedSha = stringOrNull(bundle.source?.commit);
|
||||
const startedAt = Date.now();
|
||||
const sourceSync = runNativeSourceObservationSync(registry, follower, options, Math.min(timeoutSeconds, follower.budgets.sourceSyncSeconds));
|
||||
const sourceSyncDetail = sourceSync === null || sourceSync.result.ok ? null : redactText(tailText(sourceSync.result.conditionMessage ?? sourceSync.result.logsTail ?? "unknown", 800));
|
||||
const sourceSyncError = sourceSyncDetail === null ? null : `native source sync failed: ${sourceSyncDetail}`;
|
||||
const bundle = readNativeObjectBundle(registry, follower, options, remainingSeconds(startedAt, timeoutSeconds), runKubeScript);
|
||||
const observedSha = sourceSyncError === null ? stringOrNull(bundle.source?.commit) : null;
|
||||
const runtimeTargetSha = runtimeTargetShaFromWorkloads(follower.nativeStatus.runtime, bundle.workloads);
|
||||
const pipelineRunName = stringOrNull(asOptionalRecord(bundle.pipelineRun?.metadata)?.name) ?? expectedPipelineRunName(follower, observedSha);
|
||||
const pipelineRunPresent = follower.nativeStatus.tekton === null ? null : bundle.pipelineRun !== null;
|
||||
@@ -1497,8 +1502,9 @@ async function readAdapterStatus(registry: BranchFollowerRegistry, follower: Fol
|
||||
&& argoReady !== false
|
||||
&& runtimeReady !== false;
|
||||
const targetSha = hasRuntimeTarget && runtimeTargetSha === observedSha && aligned ? runtimeTargetSha : runtimeTargetSha;
|
||||
const ok = bundle.ok;
|
||||
const ok = bundle.ok && sourceSyncError === null;
|
||||
const phase = inferPhase(ok, aligned, observedSha, targetSha, bundle.timedOut);
|
||||
const errors = sourceSyncError === null ? bundle.errors : [sourceSyncError, ...bundle.errors];
|
||||
return {
|
||||
ok,
|
||||
command: "native:k8s-git-mirror+tekton+argocd+runtime",
|
||||
@@ -1518,10 +1524,11 @@ async function readAdapterStatus(registry: BranchFollowerRegistry, follower: Fol
|
||||
gitMirrorReady,
|
||||
argoReady,
|
||||
runtimeReady,
|
||||
errors: bundle.errors,
|
||||
errors,
|
||||
}),
|
||||
payload: {
|
||||
source: bundle.source,
|
||||
sourceSync: sourceSync === null ? null : sourceSync.result,
|
||||
gitMirror: nativeGitMirrorSummary(bundle.gitMirror),
|
||||
tekton: nativePipelineRunSummary(bundle.pipelineRun),
|
||||
taskRuns: bundle.taskRuns,
|
||||
@@ -1529,7 +1536,7 @@ async function readAdapterStatus(registry: BranchFollowerRegistry, follower: Fol
|
||||
argo: nativeArgoSummary(bundle.argoApplication),
|
||||
runtime: nativeRuntimeSummary(follower.nativeStatus.runtime, bundle.workloads, observedSha),
|
||||
timings: { statusRead: { elapsedMs: bundle.elapsedMs, budgetSeconds: timeoutSeconds } },
|
||||
errors: bundle.errors,
|
||||
errors,
|
||||
statusAuthority: "k8s-native",
|
||||
parsedDownstreamCliOutput: false,
|
||||
},
|
||||
@@ -1538,6 +1545,13 @@ async function readAdapterStatus(registry: BranchFollowerRegistry, follower: Fol
|
||||
};
|
||||
}
|
||||
|
||||
function runNativeSourceObservationSync(registry: BranchFollowerRegistry, follower: FollowerSpec, options: ParsedOptions, timeoutSeconds: number): { jobName: string; namespace: string; result: NativeK8sJobResult } | null {
|
||||
if (!options.inCluster) return null;
|
||||
if (follower.adapter !== "hwlab-node-runtime" && follower.adapter !== "agentrun-yaml-lane") return null;
|
||||
const bucket = Math.floor(Date.now() / (registry.controller.loop.intervalSeconds * 1000)).toString(36);
|
||||
return runNativeGitMirrorStage(registry, follower, "source", "sync", timeoutSeconds, `source-${bucket}`);
|
||||
}
|
||||
|
||||
function inferPhase(ok: boolean, aligned: boolean | null, observedSha: string | null, targetSha: string | null, timedOut: boolean): BranchFollowerPhase {
|
||||
if (!ok || timedOut) return "Blocked";
|
||||
if (aligned === true) return "Succeeded";
|
||||
@@ -1562,112 +1576,6 @@ function nativeStatusMessage(ok: boolean, phase: BranchFollowerPhase, observedSh
|
||||
return "k8s-native status did not expose observed source sha";
|
||||
}
|
||||
|
||||
function readNativeObjectBundle(registry: BranchFollowerRegistry, follower: FollowerSpec, options: ParsedOptions, timeoutSeconds: number): NativeObjectBundle {
|
||||
const native = follower.nativeStatus;
|
||||
const source = native.source;
|
||||
const tekton = native.tekton;
|
||||
const argo = native.argo;
|
||||
const runtime = native.runtime;
|
||||
const gitopsBranch = nativeGitMirrorGitopsBranch(follower);
|
||||
const workloadRefs = (runtime?.workloads ?? []).map((workload, index) => {
|
||||
const resource = workload.kind === "Deployment" ? "deployments" : "statefulsets";
|
||||
return `workload${index}\t/apis/apps/v1/namespaces/${runtime?.namespace ?? follower.target.namespace}/${resource}/${workload.name}`;
|
||||
});
|
||||
const workloadRefsText = workloadRefs.length === 0 ? "" : `${workloadRefs.join("\n")}\n`;
|
||||
const script = [
|
||||
"set +e",
|
||||
"tmpdir=$(mktemp -d)",
|
||||
"cleanup() { rm -rf \"$tmpdir\"; }",
|
||||
"trap cleanup EXIT INT TERM",
|
||||
nativeBundleScriptLoadShell(),
|
||||
`REPO_PATH=${shQuote(source.repoPath)}`,
|
||||
`SOURCE_BRANCH=${shQuote(follower.source.branch)}`,
|
||||
`REPOSITORY=${shQuote(follower.source.repository)}`,
|
||||
`SNAPSHOT_PREFIX=${shQuote(follower.source.snapshotPrefix)}`,
|
||||
`GITOPS_BRANCH=${shQuote(gitopsBranch ?? "")}`,
|
||||
`TEKTON_NAMESPACE=${shQuote(tekton?.namespace ?? "")}`,
|
||||
`PIPELINE_RUN_PREFIX=${shQuote(tekton?.pipelineRunPrefix ?? "")}`,
|
||||
`ARGO_NAMESPACE=${shQuote(argo?.namespace ?? "")}`,
|
||||
`ARGO_APPLICATION=${shQuote(argo?.application ?? "")}`,
|
||||
`WORKLOAD_REFS_B64=${shQuote(Buffer.from(workloadRefsText, "utf8").toString("base64"))}`,
|
||||
"NATIVE_CICD_SCRIPT_DIR=\"$tmpdir\"",
|
||||
"export NATIVE_CICD_SCRIPT_DIR REPO_PATH SOURCE_BRANCH REPOSITORY SNAPSHOT_PREFIX GITOPS_BRANCH TEKTON_NAMESPACE PIPELINE_RUN_PREFIX ARGO_NAMESPACE ARGO_APPLICATION WORKLOAD_REFS_B64",
|
||||
"\"$tmpdir/read-native-bundle.sh\"",
|
||||
].join("\n");
|
||||
const startedAt = Date.now();
|
||||
const result = runKubeScript(registry, options, script, "", timeoutSeconds * 1000);
|
||||
const parsed = parseNativeBundleLines(result.stdout);
|
||||
const sourceRecord = asOptionalRecord(parsed.objects.source);
|
||||
return {
|
||||
ok: result.exitCode === 0 && sourceRecord !== null && parsed.fatalErrors.length === 0,
|
||||
source: sourceRecord,
|
||||
gitMirror: asOptionalRecord(parsed.objects.gitMirror),
|
||||
pipelineRun: asOptionalRecord(parsed.objects.pipelineRun),
|
||||
taskRuns: asOptionalRecord(parsed.objects.taskRuns),
|
||||
planArtifacts: asOptionalRecord(parsed.objects.planArtifacts),
|
||||
argoApplication: asOptionalRecord(parsed.objects.argoApplication),
|
||||
workloads: Object.entries(parsed.objects)
|
||||
.filter(([key]) => /^workload\d+$/u.test(key))
|
||||
.sort(([left], [right]) => left.localeCompare(right))
|
||||
.map(([, value]) => asOptionalRecord(value))
|
||||
.filter((item): item is Record<string, unknown> => item !== null),
|
||||
errors: [
|
||||
...parsed.errors,
|
||||
...(result.exitCode === 0 ? [] : [`native bundle command failed: exitCode=${result.exitCode}`]),
|
||||
...parsed.fatalErrors,
|
||||
],
|
||||
exitCode: result.exitCode,
|
||||
timedOut: result.timedOut,
|
||||
elapsedMs: Date.now() - startedAt,
|
||||
stdoutTail: redactText(tailText(result.stdout, 1000)),
|
||||
stderrTail: redactText(tailText(result.stderr, 1000)),
|
||||
};
|
||||
}
|
||||
|
||||
const NATIVE_BUNDLE_SCRIPT_NAMES = [
|
||||
"read-native-bundle.sh",
|
||||
"kube-get.mjs",
|
||||
"compact-native-object.mjs",
|
||||
"compact-git-mirror.mjs",
|
||||
"plan-artifacts.mjs",
|
||||
] as const;
|
||||
|
||||
function nativeBundleScriptLoadShell(): string {
|
||||
return nativeScriptLoadShell(NATIVE_BUNDLE_SCRIPT_NAMES);
|
||||
}
|
||||
|
||||
function nativeScriptLoadShell(names: readonly string[]): string {
|
||||
return names.map((name) => {
|
||||
const encoded = Buffer.from(readFileSync(rootPath("scripts/native/cicd", name), "utf8"), "utf8").toString("base64");
|
||||
return [
|
||||
`printf '%s' ${shQuote(encoded)} | base64 -d > "$tmpdir/${name}"`,
|
||||
`chmod +x "$tmpdir/${name}"`,
|
||||
].join("\n");
|
||||
}).join("\n");
|
||||
}
|
||||
|
||||
function parseNativeBundleLines(stdout: string): { objects: Record<string, unknown>; errors: string[]; fatalErrors: string[] } {
|
||||
const objects: Record<string, unknown> = {};
|
||||
const errors: string[] = [];
|
||||
const fatalErrors: string[] = [];
|
||||
for (const line of stdout.split(/\r?\n/u)) {
|
||||
if (!line.startsWith("UNIDESK_NATIVE_")) continue;
|
||||
const [kind, key, payload] = line.split("\t");
|
||||
if (kind === undefined || key === undefined || payload === undefined) continue;
|
||||
const decoded = Buffer.from(payload, "base64").toString("utf8").trim();
|
||||
if (kind === "UNIDESK_NATIVE_JSON") {
|
||||
const parsed = parseJsonObject(decoded);
|
||||
if (parsed !== null) objects[key] = parsed;
|
||||
else errors.push(`${key}: invalid native JSON payload`);
|
||||
} else if (kind === "UNIDESK_NATIVE_ERROR") {
|
||||
const message = `${key}: ${redactText(tailText(decoded, 500)) || "not found"}`;
|
||||
errors.push(message);
|
||||
if (key === "source") fatalErrors.push(message);
|
||||
}
|
||||
}
|
||||
return { objects, errors, fatalErrors };
|
||||
}
|
||||
|
||||
function expectedPipelineRunName(follower: FollowerSpec, observedSha: string | null): string | null {
|
||||
if (observedSha === null || follower.nativeStatus.tekton === null) return null;
|
||||
return `${follower.nativeStatus.tekton.pipelineRunPrefix}-${shortSha(observedSha)}`;
|
||||
@@ -2340,6 +2248,8 @@ function stageTimingsFromNativePayload(payload: Record<string, unknown> | null):
|
||||
const stages: StageTiming[] = [];
|
||||
const statusRead = asOptionalRecord(asOptionalRecord(payload.timings)?.statusRead);
|
||||
stages.push(stageTiming("status-read", "ok", secondsFromMs(numberOrNull(statusRead?.elapsedMs)), numberOrNull(statusRead?.budgetSeconds), "native-status", null));
|
||||
const sourceSyncStage = k8sJobTiming("git-mirror-sync", asOptionalRecord(payload.sourceSync));
|
||||
if (sourceSyncStage !== null) stages.push(sourceSyncStage);
|
||||
const gitMirror = asOptionalRecord(payload.gitMirror);
|
||||
if (gitMirror !== null) {
|
||||
const status = gitMirror.pendingFlush === true ? "pending-flush" : gitMirror.githubInSync === true && gitMirror.sourceSnapshotReady === true ? "ready" : "not-ready";
|
||||
|
||||
Reference in New Issue
Block a user