feat: add branch follower gate probes
This commit is contained in:
Codex
@@ -0,0 +1,388 @@
|
||||
import { createHash } from "node:crypto";
|
||||
import { execFileSync } from "node:child_process";
|
||||
import { existsSync, readFileSync } from "node:fs";
|
||||
import http from "node:http";
|
||||
import https from "node:https";
|
||||
|
||||
const gate = requiredEnv("GATE");
|
||||
const follower = requiredEnv("FOLLOWER_ID");
|
||||
const repository = requiredEnv("REPOSITORY");
|
||||
const repoPath = requiredEnv("REPO_PATH");
|
||||
const sourceBranch = requiredEnv("SOURCE_BRANCH");
|
||||
const snapshotPrefix = requiredEnv("SNAPSHOT_PREFIX").replace(/\/+$/u, "");
|
||||
const selectedSource = process.env.SOURCE_COMMIT || "";
|
||||
const gitopsBranch = process.env.GITOPS_BRANCH || "";
|
||||
const tektonNamespace = process.env.TEKTON_NAMESPACE || "";
|
||||
const pipelineRunPrefix = process.env.PIPELINE_RUN_PREFIX || "";
|
||||
const argoNamespace = process.env.ARGO_NAMESPACE || "";
|
||||
const argoApplication = process.env.ARGO_APPLICATION || "";
|
||||
const runtimeNamespace = process.env.RUNTIME_NAMESPACE || "";
|
||||
const workloads = parseWorkloads(process.env.WORKLOADS_B64 || "");
|
||||
const healthUrl = process.env.HEALTH_URL || "";
|
||||
|
||||
const errors = [];
|
||||
const branchCommit = rev(`refs/heads/${sourceBranch}`);
|
||||
const sourceCommit = selectedSource || branchCommit || "";
|
||||
const sourceStageRef = sourceCommit ? `${snapshotPrefix}/${sourceCommit}` : null;
|
||||
const sourceSnapshot = sourceStageRef ? rev(sourceStageRef) : null;
|
||||
const source = {
|
||||
repository,
|
||||
branch: sourceBranch,
|
||||
sourceCommit: sourceCommit || null,
|
||||
stageRef: sourceStageRef,
|
||||
snapshotReady: Boolean(sourceCommit && sourceSnapshot === sourceCommit),
|
||||
authority: "k8s-git-mirror-snapshot",
|
||||
};
|
||||
const gitMirror = gitMirrorSummary(sourceCommit);
|
||||
|
||||
let evidence;
|
||||
if (gate === "reuse-plan") evidence = reusePlanEvidence(sourceCommit);
|
||||
else if (gate === "ci-taskrun-plan") evidence = await ciTaskRunEvidence(sourceCommit);
|
||||
else if (gate === "cd-rollout-plan") evidence = await cdRolloutEvidence(sourceCommit);
|
||||
else if (gate === "post-deploy-health") evidence = await postDeployHealthEvidence(sourceCommit);
|
||||
else fail(`unsupported gate ${gate}`);
|
||||
|
||||
const ok = errors.length === 0 && evidence?.ok === true;
|
||||
console.log(JSON.stringify({
|
||||
ok,
|
||||
gate,
|
||||
follower,
|
||||
source,
|
||||
evidence,
|
||||
errors: errors.slice(0, 6),
|
||||
statusAuthority: "kubernetes-api-serviceaccount",
|
||||
parsedDownstreamCliOutput: false,
|
||||
bounded: true,
|
||||
}));
|
||||
|
||||
function reusePlanEvidence(commit) {
|
||||
const reuse = readReuseConfig(commit);
|
||||
return {
|
||||
ok: source.snapshotReady && gitMirror.ok === true && reuse.present === true && reuse.serviceCount > 0,
|
||||
gitMirror,
|
||||
reuse,
|
||||
};
|
||||
}
|
||||
|
||||
async function ciTaskRunEvidence(commit) {
|
||||
if (!commit || !tektonNamespace || !pipelineRunPrefix) return notConfigured("tekton");
|
||||
const pipelineRunName = `${pipelineRunPrefix}-${commit.slice(0, 12)}`;
|
||||
const pipelineRun = await getJson(`/apis/tekton.dev/v1/namespaces/${encodeURIComponent(tektonNamespace)}/pipelineruns/${encodeURIComponent(pipelineRunName)}`, false);
|
||||
const prStatus = pipelineRunStatus(pipelineRun);
|
||||
const pipelineRef = str(pipelineRun?.spec?.pipelineRef?.name);
|
||||
const pipeline = pipelineRef ? await getJson(`/apis/tekton.dev/v1/namespaces/${encodeURIComponent(tektonNamespace)}/pipelines/${encodeURIComponent(pipelineRef)}`, false) : null;
|
||||
const taskRuns = await getJson(`/apis/tekton.dev/v1/namespaces/${encodeURIComponent(tektonNamespace)}/taskruns?labelSelector=${encodeURIComponent(`tekton.dev/pipelineRun=${pipelineRunName}`)}`, false);
|
||||
const taskSummary = taskRunsSummary(taskRuns);
|
||||
return {
|
||||
ok: prStatus.succeeded === true && taskSummary.failedCount === 0 && taskSummary.activeCount === 0,
|
||||
pipelineRun: prStatus,
|
||||
pipeline: {
|
||||
name: pipelineRef,
|
||||
taskCount: Array.isArray(pipeline?.spec?.tasks) ? pipeline.spec.tasks.length : null,
|
||||
tasks: Array.isArray(pipeline?.spec?.tasks) ? pipeline.spec.tasks.slice(0, 12).map((task) => ({ name: str(task?.name), runAfter: Array.isArray(task?.runAfter) ? task.runAfter.slice(0, 6) : [] })) : [],
|
||||
},
|
||||
taskRuns: taskSummary,
|
||||
};
|
||||
}
|
||||
|
||||
async function cdRolloutEvidence(commit) {
|
||||
const argo = await argoSummary();
|
||||
const runtime = await runtimeSummary(commit);
|
||||
return {
|
||||
ok: gitMirror.githubInSync !== false && argo.ready === true && runtime.ready === true && runtime.aligned === true,
|
||||
gitMirror,
|
||||
argo,
|
||||
runtime,
|
||||
};
|
||||
}
|
||||
|
||||
async function postDeployHealthEvidence(commit) {
|
||||
const runtime = await runtimeSummary(commit);
|
||||
const health = await healthSummary();
|
||||
return {
|
||||
ok: runtime.ready === true && runtime.aligned === true && health.ok === true,
|
||||
runtime,
|
||||
health,
|
||||
};
|
||||
}
|
||||
|
||||
function readReuseConfig(commit) {
|
||||
if (!commit || !sourceStageRef) return { present: false, reason: "source-commit-missing" };
|
||||
try {
|
||||
const text = execFileSync("git", [`--git-dir=${repoPath}`, "show", `${sourceStageRef}:gitops/reuse.ymal`], { encoding: "utf8", maxBuffer: 256 * 1024 });
|
||||
const services = reuseServiceIds(text);
|
||||
return {
|
||||
present: true,
|
||||
path: "gitops/reuse.ymal",
|
||||
bytes: Buffer.byteLength(text, "utf8"),
|
||||
sha256: createHash("sha256").update(text).digest("hex"),
|
||||
serviceCount: services.length,
|
||||
serviceIds: services.slice(0, 12),
|
||||
runtimeReuseMentioned: /\bruntimeReuse\b/u.test(text),
|
||||
envReuseMentioned: /\benvReuse\b/u.test(text),
|
||||
};
|
||||
} catch (error) {
|
||||
return { present: false, path: "gitops/reuse.ymal", reason: shortText(error?.message || String(error)) };
|
||||
}
|
||||
}
|
||||
|
||||
function reuseServiceIds(text) {
|
||||
const ids = new Set([...text.matchAll(/(?:^|\n)\s*-\s*id:\s*([A-Za-z0-9_.-]+)/gu)].map((match) => match[1]).filter(Boolean));
|
||||
const lines = text.split(/\r?\n/u);
|
||||
let inServices = false;
|
||||
let servicesIndent = 0;
|
||||
for (const line of lines) {
|
||||
const services = /^(\s*)services:\s*$/u.exec(line);
|
||||
if (services) {
|
||||
inServices = true;
|
||||
servicesIndent = services[1].length;
|
||||
continue;
|
||||
}
|
||||
if (!inServices) continue;
|
||||
const match = /^(\s*)([A-Za-z0-9_.-]+):\s*$/u.exec(line);
|
||||
if (!match) continue;
|
||||
const indent = match[1].length;
|
||||
if (indent <= servicesIndent) {
|
||||
inServices = false;
|
||||
continue;
|
||||
}
|
||||
if (indent === servicesIndent + 2) ids.add(match[2]);
|
||||
}
|
||||
return Array.from(ids);
|
||||
}
|
||||
|
||||
function gitMirrorSummary(commit) {
|
||||
const localSource = rev(`refs/heads/${sourceBranch}`);
|
||||
const githubSource = rev(`refs/mirror-stage/heads/${sourceBranch}`);
|
||||
const localGitops = gitopsBranch ? rev(`refs/heads/${gitopsBranch}`) : null;
|
||||
const githubGitops = gitopsBranch ? rev(`refs/mirror-stage/heads/${gitopsBranch}`) : null;
|
||||
return {
|
||||
ok: Boolean(localSource) && source.snapshotReady && (!gitopsBranch || !localGitops || localGitops === githubGitops),
|
||||
localSource,
|
||||
githubSource,
|
||||
sourceStageRef,
|
||||
sourceSnapshot,
|
||||
sourceSnapshotReady: source.snapshotReady,
|
||||
gitopsBranch: gitopsBranch || null,
|
||||
localGitops,
|
||||
githubGitops,
|
||||
pendingFlush: gitopsBranch ? Boolean(localGitops && localGitops !== githubGitops) : null,
|
||||
githubInSync: gitopsBranch ? Boolean(!localGitops || localGitops === githubGitops) : null,
|
||||
};
|
||||
}
|
||||
|
||||
async function argoSummary() {
|
||||
if (!argoNamespace || !argoApplication) return { ready: null, reason: "argo-not-configured" };
|
||||
const app = await getJson(`/apis/argoproj.io/v1alpha1/namespaces/${encodeURIComponent(argoNamespace)}/applications/${encodeURIComponent(argoApplication)}`, false);
|
||||
const sync = app?.status?.sync || {};
|
||||
const health = app?.status?.health || {};
|
||||
const op = app?.status?.operationState || {};
|
||||
return {
|
||||
name: argoApplication,
|
||||
namespace: argoNamespace,
|
||||
syncStatus: str(sync.status),
|
||||
healthStatus: str(health.status),
|
||||
revision: str(sync.revision),
|
||||
operationPhase: str(op.phase),
|
||||
operationMessage: str(op.message),
|
||||
ready: sync.status === "Synced" && health.status === "Healthy",
|
||||
};
|
||||
}
|
||||
|
||||
async function runtimeSummary(expected) {
|
||||
if (!runtimeNamespace || workloads.length === 0) return { ready: null, aligned: null, reason: "runtime-not-configured" };
|
||||
const rows = [];
|
||||
for (const workload of workloads) {
|
||||
const resource = workload.kind === "StatefulSet" ? "statefulsets" : "deployments";
|
||||
const item = await getJson(`/apis/apps/v1/namespaces/${encodeURIComponent(runtimeNamespace)}/${resource}/${encodeURIComponent(workload.name)}`, false);
|
||||
rows.push(workloadSummary(workload, item, expected));
|
||||
}
|
||||
return {
|
||||
namespace: runtimeNamespace,
|
||||
expectedSha: shortSha(expected),
|
||||
ready: rows.length > 0 && rows.every((row) => row.ready === true),
|
||||
aligned: rows.length > 0 && rows.every((row) => row.aligned === true),
|
||||
workloads: rows,
|
||||
};
|
||||
}
|
||||
|
||||
async function healthSummary() {
|
||||
if (!healthUrl) return { ok: null, reason: "health-url-not-configured" };
|
||||
const targets = [`${healthUrl.replace(/\/+$/u, "")}/health/readiness`, `${healthUrl.replace(/\/+$/u, "")}/health/live`];
|
||||
const probes = [];
|
||||
for (const url of targets) probes.push(await httpProbe(url));
|
||||
return { ok: probes.every((probe) => probe.ok), probes };
|
||||
}
|
||||
|
||||
function workloadSummary(spec, item, expected) {
|
||||
const status = item?.status || {};
|
||||
const desired = item?.spec?.replicas ?? 1;
|
||||
const readyCount = spec.kind === "StatefulSet" ? status.readyReplicas ?? 0 : status.availableReplicas ?? 0;
|
||||
const updated = status.updatedReplicas ?? readyCount;
|
||||
const commit = workloadCommit(spec, item);
|
||||
return {
|
||||
kind: spec.kind,
|
||||
name: spec.name,
|
||||
ready: readyCount >= desired && updated >= desired,
|
||||
desired,
|
||||
readyReplicas: readyCount,
|
||||
updatedReplicas: updated,
|
||||
sourceCommit: shortSha(commit),
|
||||
aligned: Boolean(expected && commit && commit === expected),
|
||||
};
|
||||
}
|
||||
|
||||
function workloadCommit(spec, item) {
|
||||
const meta = item?.metadata || {};
|
||||
const tmpl = item?.spec?.template || {};
|
||||
const podMeta = tmpl.metadata || {};
|
||||
for (const key of spec.sourceCommit.labels || []) if (sha(meta.labels?.[key])) return meta.labels[key];
|
||||
for (const key of spec.sourceCommit.annotations || []) if (sha(meta.annotations?.[key])) return meta.annotations[key];
|
||||
for (const key of spec.sourceCommit.podLabels || []) if (sha(podMeta.labels?.[key])) return podMeta.labels[key];
|
||||
for (const key of spec.sourceCommit.podAnnotations || []) if (sha(podMeta.annotations?.[key])) return podMeta.annotations[key];
|
||||
for (const envName of spec.sourceCommit.env || []) {
|
||||
for (const container of tmpl.spec?.containers || []) {
|
||||
for (const env of container.env || []) if (env?.name === envName && sha(env.value)) return env.value;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function pipelineRunStatus(value) {
|
||||
const condition = (value?.status?.conditions || []).find((item) => item?.type === "Succeeded") || null;
|
||||
return {
|
||||
name: str(value?.metadata?.name),
|
||||
namespace: str(value?.metadata?.namespace),
|
||||
pipelineRefName: str(value?.spec?.pipelineRef?.name),
|
||||
present: value !== null,
|
||||
succeeded: condition?.status === "True" ? true : condition?.status === "False" ? false : null,
|
||||
reason: str(condition?.reason),
|
||||
startTime: str(value?.status?.startTime),
|
||||
completionTime: str(value?.status?.completionTime),
|
||||
durationSeconds: durationSeconds(value?.status?.startTime, value?.status?.completionTime),
|
||||
};
|
||||
}
|
||||
|
||||
function taskRunsSummary(list) {
|
||||
const items = Array.isArray(list?.items) ? list.items : [];
|
||||
const rows = items.map((item) => {
|
||||
const condition = (item?.status?.conditions || []).find((entry) => entry?.type === "Succeeded") || {};
|
||||
return {
|
||||
name: str(item?.metadata?.name),
|
||||
taskName: str(item?.metadata?.labels?.["tekton.dev/pipelineTask"]) || str(item?.spec?.taskRef?.name),
|
||||
status: str(condition.status) || "Unknown",
|
||||
reason: str(condition.reason),
|
||||
durationSeconds: durationSeconds(item?.status?.startTime, item?.status?.completionTime),
|
||||
};
|
||||
});
|
||||
const failed = rows.filter((item) => item.status === "False");
|
||||
const active = rows.filter((item) => item.status !== "True" && item.status !== "False");
|
||||
const slow = rows.filter((item) => typeof item.durationSeconds === "number" && item.durationSeconds >= 60);
|
||||
return {
|
||||
count: rows.length,
|
||||
failedCount: failed.length,
|
||||
activeCount: active.length,
|
||||
slowCount: slow.length,
|
||||
failedItems: failed.slice(0, 5),
|
||||
activeItems: active.slice(0, 5),
|
||||
slowItems: slow.slice(0, 5),
|
||||
items: rows.slice(0, 12),
|
||||
};
|
||||
}
|
||||
|
||||
async function getJson(path, required) {
|
||||
const host = process.env.KUBERNETES_SERVICE_HOST;
|
||||
const port = Number(process.env.KUBERNETES_SERVICE_PORT || "443");
|
||||
const tokenPath = "/var/run/secrets/kubernetes.io/serviceaccount/token";
|
||||
const caPath = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt";
|
||||
if (!host || !existsSync(tokenPath) || !existsSync(caPath)) fail("kubernetes serviceaccount is unavailable");
|
||||
const token = readFileSync(tokenPath, "utf8").trim();
|
||||
const ca = readFileSync(caPath);
|
||||
return await new Promise((resolve, reject) => {
|
||||
const req = https.request({ host, port, path, method: "GET", ca, headers: { authorization: `Bearer ${token}` } }, (res) => {
|
||||
let body = "";
|
||||
res.setEncoding("utf8");
|
||||
res.on("data", (chunk) => { body += chunk; });
|
||||
res.on("end", () => {
|
||||
if (res.statusCode === 404 && !required) return resolve(null);
|
||||
if ((res.statusCode || 0) < 200 || (res.statusCode || 0) >= 300) return reject(new Error(shortText(body || `kube api ${res.statusCode}`)));
|
||||
try { resolve(JSON.parse(body)); } catch (error) { reject(error); }
|
||||
});
|
||||
});
|
||||
req.on("error", reject);
|
||||
req.end();
|
||||
}).catch((error) => {
|
||||
errors.push(`${path}: ${shortText(error?.message || String(error))}`);
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
async function httpProbe(url) {
|
||||
const client = url.startsWith("https:") ? https : http;
|
||||
const started = Date.now();
|
||||
return await new Promise((resolve) => {
|
||||
const req = client.get(url, { timeout: 5000 }, (res) => {
|
||||
res.resume();
|
||||
res.on("end", () => resolve({ url, ok: (res.statusCode || 0) >= 200 && (res.statusCode || 0) < 300, statusCode: res.statusCode || null, elapsedMs: Date.now() - started }));
|
||||
});
|
||||
req.on("timeout", () => { req.destroy(); resolve({ url, ok: false, statusCode: null, elapsedMs: Date.now() - started, reason: "timeout" }); });
|
||||
req.on("error", (error) => resolve({ url, ok: false, statusCode: null, elapsedMs: Date.now() - started, reason: shortText(error?.message || String(error)) }));
|
||||
});
|
||||
}
|
||||
|
||||
function parseWorkloads(value) {
|
||||
if (!value) return [];
|
||||
try {
|
||||
const parsed = JSON.parse(Buffer.from(value, "base64").toString("utf8"));
|
||||
return Array.isArray(parsed) ? parsed.filter((item) => item && typeof item === "object").slice(0, 8) : [];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function rev(ref) {
|
||||
try {
|
||||
const out = execFileSync("git", [`--git-dir=${repoPath}`, "rev-parse", "--verify", `${ref}^{commit}`], { encoding: "utf8", stdio: ["ignore", "pipe", "ignore"] }).trim();
|
||||
return sha(out) ? out : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
function durationSeconds(start, finish) {
|
||||
if (!start || !finish) return null;
|
||||
const value = (Date.parse(finish) - Date.parse(start)) / 1000;
|
||||
return Number.isFinite(value) && value >= 0 ? Math.round(value * 10) / 10 : null;
|
||||
}
|
||||
|
||||
function notConfigured(name) {
|
||||
return { ok: false, reason: `${name}-not-configured` };
|
||||
}
|
||||
|
||||
function str(value) {
|
||||
return typeof value === "string" && value.length > 0 ? value : null;
|
||||
}
|
||||
|
||||
function sha(value) {
|
||||
return typeof value === "string" && /^[0-9a-f]{40}$/iu.test(value);
|
||||
}
|
||||
|
||||
function shortSha(value) {
|
||||
return sha(value) ? value.slice(0, 12) : null;
|
||||
}
|
||||
|
||||
function shortText(value) {
|
||||
const text = String(value || "").replace(/\s+/gu, " ").trim();
|
||||
return text.length <= 300 ? text : text.slice(0, 300);
|
||||
}
|
||||
|
||||
function requiredEnv(name) {
|
||||
const value = process.env[name];
|
||||
if (!value) fail(`${name} is required`);
|
||||
return value;
|
||||
}
|
||||
|
||||
function fail(message) {
|
||||
console.error(message);
|
||||
process.exit(1);
|
||||
}
|
||||
Reference in New Issue
Block a user