import { existsSync, readFileSync } from "node:fs"; import https from "node:https"; import { spawnSync } from "node:child_process"; const namespace = requiredEnv("JOB_NAMESPACE"); const jobName = requiredEnv("JOB_NAME"); const stageName = process.env.STAGE_NAME || "job"; const sourceCommit = process.env.SOURCE_COMMIT || ""; const logsTailLines = positiveInt(process.env.LOGS_TAIL_LINES, "LOGS_TAIL_LINES"); const maxLogBytes = positiveInt(process.env.MAX_LOG_BYTES, "MAX_LOG_BYTES"); const maxMessageBytes = positiveInt(process.env.MAX_MESSAGE_BYTES, "MAX_MESSAGE_BYTES"); const maxContainers = positiveInt(process.env.MAX_CONTAINERS, "MAX_CONTAINERS"); const useServiceAccount = Boolean(process.env.KUBERNETES_SERVICE_HOST && process.env.KUBERNETES_SERVICE_PORT) && existsSync("/var/run/secrets/kubernetes.io/serviceaccount/token") && existsSync("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"); class KubeReadError extends Error { constructor(reason, message, details = {}) { super(shortText(message) || reason); this.name = "KubeReadError"; this.reason = reason; this.statusCode = details.statusCode ?? null; this.notFound = details.notFound === true; this.path = details.path || null; } } main().catch((error) => { process.stderr.write(error?.message || String(error)); process.exit(1); }); async function main() { const job = await getJson(`/apis/batch/v1/namespaces/${encodeURIComponent(namespace)}/jobs/${encodeURIComponent(jobName)}`, false); if (job === null) { console.log(JSON.stringify({ ok: false, degradedReason: "job-not-found", message: `Job ${namespace}/${jobName} was not found; it may have expired by ttlSecondsAfterFinished`, query: { namespace, jobName, stage: stageName, sourceCommit: sourceCommit || null }, job: null, pods: [], logs: [], statusAuthority: useServiceAccount ? "kubernetes-api-serviceaccount" : "target-node-kubectl-raw", parsedDownstreamCliOutput: false, })); return; } const pods = await listJobPods(job); const podSummaries = pods.slice(0, Math.max(1, maxContainers)).map(podSummary); const logTargets = selectLogTargets(pods).slice(0, maxContainers); const logs = []; const perContainerBytes = Math.max(1, Math.floor(maxLogBytes / Math.max(1, logTargets.length))); for (const target of logTargets) { const read = await readPodLog(target.podName, target.container, logsTailLines, perContainerBytes); const text = read.tail || ""; logs.push({ ok: read.ok, degradedReason: read.degradedReason, message: read.message, pod: target.podName, container: target.container, lineCount: text.length === 0 ? 0 : text.split(/\r?\n/u).filter((line) => line.length > 0).length, bytes: Buffer.byteLength(text, "utf8"), tail: text, nodeCicdTiming: lastNodeCicdTiming(text), }); } const logFailures = logs.filter((item) => item.ok === false); const status = job.status || {}; const metadata = job.metadata || {}; console.log(JSON.stringify({ ok: logFailures.length === 0, degradedReason: logFailures.length === 0 ? null : "log-read-failed", errors: logFailures.map((item) => ({ pod: item.pod, container: item.container, degradedReason: item.degradedReason, message: item.message })), query: { namespace, jobName, stage: stageName, sourceCommit: sourceCommit || null }, job: { name: metadata.name || jobName, namespace: metadata.namespace || namespace, stage: stageName, sourceCommit: shortSha(sourceCommit), createdAt: metadata.creationTimestamp || null, startTime: status.startTime || null, completionTime: status.completionTime || null, durationSeconds: durationSeconds(status.startTime, status.completionTime), activeDeadlineSeconds: job.spec?.activeDeadlineSeconds ?? null, ttlSecondsAfterFinished: job.spec?.ttlSecondsAfterFinished ?? null, active: integerOrNull(status.active), succeeded: integerOrNull(status.succeeded), failed: integerOrNull(status.failed), condition: compactCondition(latestCondition(status.conditions)), completed: Boolean(status.succeeded && status.succeeded > 0), failedState: Boolean(status.failed && status.failed > 0), }, pods: podSummaries, logs, nodeCicdTiming: lastTiming(logs), statusAuthority: useServiceAccount ? "kubernetes-api-serviceaccount" : "target-node-kubectl-raw", parsedDownstreamCliOutput: false, })); } async function listJobPods(job) { const uid = job?.metadata?.uid || ""; const selectors = [ `batch.kubernetes.io/job-name=${jobName}`, `job-name=${jobName}`, uid ? `controller-uid=${uid}` : null, ].filter(Boolean); for (const selector of selectors) { const list = await getJson(`/api/v1/namespaces/${encodeURIComponent(namespace)}/pods?labelSelector=${encodeURIComponent(selector)}`, false); const items = Array.isArray(list?.items) ? list.items : []; if (items.length > 0) return items.sort((left, right) => String(right.metadata?.creationTimestamp || "").localeCompare(String(left.metadata?.creationTimestamp || ""))); } return []; } function podSummary(pod) { const statuses = allContainerStatuses(pod); return { name: pod.metadata?.name || null, phase: pod.status?.phase || null, createdAt: pod.metadata?.creationTimestamp || null, startTime: pod.status?.startTime || null, ready: conditionByType(pod.status, "Ready")?.status || null, containers: statuses.slice(0, maxContainers).map(containerSummary), }; } function selectLogTargets(pods) { const targets = []; for (const pod of pods) { for (const status of allContainerStatuses(pod)) { if (status.name) targets.push({ podName: pod.metadata?.name, container: status.name, status }); } } targets.sort((left, right) => scoreContainer(right.status) - scoreContainer(left.status)); return targets.filter((item) => item.podName && item.container); } function scoreContainer(status) { if (status.state?.waiting) return 5; const terminated = status.state?.terminated || status.lastState?.terminated; if (terminated && terminated.exitCode !== 0) return 4; if (terminated) return 3; if (status.state?.running) return 2; return 1; } function allContainerStatuses(pod) { const status = pod?.status || {}; return [ ...arrayItems(status.initContainerStatuses), ...arrayItems(status.containerStatuses), ...arrayItems(status.ephemeralContainerStatuses), ]; } function containerSummary(item) { return { name: item.name || null, ready: item.ready === true, restartCount: integerOrNull(item.restartCount), waiting: compactWaiting(item.state?.waiting), running: item.state?.running ? { startedAt: item.state.running.startedAt || null } : null, terminated: compactTerminated(item.state?.terminated || item.lastState?.terminated), imageID: shortImageId(item.imageID || null), }; } async function readPodLog(podName, container, tailLines, limitBytes) { const path = `/api/v1/namespaces/${encodeURIComponent(namespace)}/pods/${encodeURIComponent(podName)}/log?container=${encodeURIComponent(container)}&tailLines=${tailLines}&limitBytes=${limitBytes}`; try { const text = tailBytes(await getText(path, false), limitBytes); return { ok: true, degradedReason: null, message: null, tail: text }; } catch (error) { return { ok: false, degradedReason: isNotFoundError(error) ? "log-not-found" : error instanceof KubeReadError ? error.reason : "log-read-failed", message: shortText(error?.message || String(error)), tail: "", }; } } async function getJson(path, required) { let text = ""; try { text = await getText(path, required); } catch (error) { if (!required && isNotFoundError(error)) return null; throw error; } try { return JSON.parse(text); } catch (error) { throw new KubeReadError("invalid-json", `kube api returned non-JSON for ${path}: ${error?.message || String(error)}`, { path }); } } async function getText(path, required) { if (useServiceAccount) return kubeApiGet(path, required); const result = spawnSync("kubectl", ["get", "--raw", path], { encoding: "utf8", maxBuffer: Math.max(maxLogBytes * 2, 1024 * 1024) }); if (result.error) throw new KubeReadError("transport-error", `kubectl get --raw transport error for ${path}: ${result.error.message}`, { path }); if (result.status === 0) return result.stdout; const body = result.stderr || result.stdout || `kubectl get --raw failed with exit ${result.status}`; throw new KubeReadError(isNotFoundText(body) ? "not-found" : "kube-api-error", `kubectl get --raw failed for ${path}: ${body}`, { path, notFound: isNotFoundText(body), }); } function kubeApiGet(path, required) { const host = process.env.KUBERNETES_SERVICE_HOST; const port = Number(process.env.KUBERNETES_SERVICE_PORT || "443"); const token = readFileSync("/var/run/secrets/kubernetes.io/serviceaccount/token", "utf8").trim(); const ca = readFileSync("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"); return 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", () => { const code = res.statusCode || 0; if (code >= 200 && code < 300) resolve(body); else reject(new KubeReadError(code === 404 || isNotFoundText(body) ? "not-found" : "kube-api-error", `kube api GET ${path} status ${code}: ${body || "-"}`, { path, statusCode: code, notFound: code === 404 || isNotFoundText(body), })); }); }); req.on("error", (error) => reject(new KubeReadError("transport-error", `kube api GET ${path} transport error: ${error?.message || String(error)}`, { path }))); req.end(); }); } function latestCondition(conditions) { return arrayItems(conditions).slice().sort((left, right) => String(right.lastTransitionTime || "").localeCompare(String(left.lastTransitionTime || "")))[0] || null; } function conditionByType(status, type) { return arrayItems(status?.conditions).find((item) => item?.type === type) || null; } function compactCondition(value) { if (!value) return null; return { type: value.type || null, status: value.status || null, reason: value.reason || null, message: shortText(value.message || null), lastTransitionTime: value.lastTransitionTime || null }; } function compactTerminated(value) { if (!value) return null; return { reason: value.reason || null, exitCode: integerOrNull(value.exitCode), message: shortText(value.message || null), startedAt: value.startedAt || null, finishedAt: value.finishedAt || null }; } function compactWaiting(value) { if (!value) return null; return { reason: value.reason || null, message: shortText(value.message || null) }; } function lastNodeCicdTiming(text) { const lines = text.split(/\r?\n/u).filter((line) => line.includes("node-cicd-timing")); for (let index = lines.length - 1; index >= 0; index -= 1) { const parsed = parseTimingLine(lines[index]); if (parsed !== null) return parsed; } return null; } function parseTimingLine(line) { const candidates = [line.trim()]; const brace = line.indexOf("{"); if (brace >= 0) candidates.push(line.slice(brace).trim()); for (const candidate of candidates) { try { const parsed = JSON.parse(candidate); if (parsed && typeof parsed === "object" && JSON.stringify(parsed).includes("node-cicd-timing")) return parsed; } catch { // Try the next bounded candidate. } } return { rawTail: shortText(line) }; } function lastTiming(logs) { for (let index = logs.length - 1; index >= 0; index -= 1) { if (logs[index].nodeCicdTiming !== null) return logs[index].nodeCicdTiming; } return null; } function durationSeconds(start, end) { const s = timestampMs(start); const e = timestampMs(end); return s === null || e === null || e < s ? null : Math.round((e - s) / 1000); } function timestampMs(value) { const parsed = Date.parse(String(value || "")); return Number.isFinite(parsed) ? parsed : null; } function shortImageId(value) { if (!value) return null; const text = String(value); const at = text.lastIndexOf("@"); return at >= 0 ? text.slice(at + 1, at + 25) : shortText(text); } function shortSha(value) { if (!value) return null; const text = String(value); return text.length > 12 ? text.slice(0, 12) : text; } function shortText(value) { if (value === null || value === undefined) return null; const text = String(value).replace(/\s+/gu, " ").trim(); return text.length <= maxMessageBytes ? text : `${text.slice(0, Math.max(0, maxMessageBytes - 3))}...`; } function tailBytes(value, maxBytes) { const buffer = Buffer.from(value, "utf8"); if (buffer.length <= maxBytes) return value; return buffer.subarray(buffer.length - maxBytes).toString("utf8"); } function arrayItems(value) { return Array.isArray(value) ? value : []; } function integerOrNull(value) { return Number.isInteger(value) ? value : null; } function isNotFoundError(error) { return error instanceof KubeReadError && error.notFound === true; } function isNotFoundText(value) { return /\b404\b|not found|NotFound/u.test(String(value || "")); } function positiveInt(value, name) { const parsed = Number(value); if (!Number.isInteger(parsed) || parsed <= 0) throw new Error(`${name} must be a positive integer`); return parsed; } function requiredEnv(name) { const value = process.env[name]; if (!value) throw new Error(`${name} is required`); return value; }