pikasTech-unidesk/scripts/native/cicd/k8s-job-drilldown.mjs

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;
}