Files
pikasTech-unidesk/scripts/src/platform-infra-observability.ts
T

3193 lines
134 KiB
TypeScript

// SPEC: PJ2026-01060501 OTel追踪 draft-2026-06-19-p0.
// Responsibility: YAML-first platform-infra OpenTelemetry tracing control commands.
import { Buffer } from "node:buffer";
import { readFileSync } from "node:fs";
import type { UniDeskConfig } from "./config";
import { rootPath } from "./config";
import type { RenderedCliResult } from "./output";
import {
compactCapture,
createYamlFieldReader,
numberField,
parseJsonOutput,
redactSensitiveUnknown,
shQuote,
capture,
} from "./platform-infra-ops-library";
const configFile = rootPath("config", "platform-infra", "observability.yaml");
const configLabel = "config/platform-infra/observability.yaml";
const fieldManager = "unidesk-platform-observability";
const {
asRecord,
objectField,
arrayOfRecords,
stringField,
integerField,
booleanField,
stringArrayField,
numberArrayField,
enumField,
kubernetesNameField,
portField,
apiPathField,
} = createYamlFieldReader(configLabel);
interface ObservabilityConfig {
version: number;
kind: "platform-infra-observability";
metadata: { id: string; owner: string; spec: string; relatedIssues: number[] };
defaults: { targetId: string };
images: {
collector: ImageSpec;
tempo: ImageSpec;
};
targets: ObservabilityTarget[];
collector: {
deploymentName: string;
serviceName: string;
configMapName: string;
replicas: number;
healthPort: number;
otlp: OtlpPorts;
};
traceBackend: {
type: "tempo";
deploymentName: string;
serviceName: string;
configMapName: string;
replicas: number;
httpPort: number;
otlp: OtlpPorts;
storage: { mode: "emptyDir"; retention: string };
};
sampling: { mode: "parentbased_traceidratio"; ratio: number };
instrumentation: {
contextPropagation: string[];
serviceConnections: ServiceConnection[];
};
resourceAttributes: {
required: string[];
businessCorrelationAttributes: string[];
};
probes: {
readinessPath: string;
traceQueryPathTemplate: string;
statusEndpoints: StatusEndpoint[];
};
}
interface ImageSpec {
repository: string;
tag: string;
pullPolicy: "Always" | "IfNotPresent" | "Never";
}
interface ObservabilityTarget {
id: string;
route: string;
namespace: string;
role: "active" | "standby";
enabled: boolean;
createNamespace: boolean;
}
interface OtlpPorts {
grpcPort: number;
httpPort: number;
}
interface ServiceConnection {
serviceName: string;
owningRepo: string;
targetNode: string;
lane: string;
namespace: string;
requiredSpans: string[];
}
interface StatusEndpoint {
name: string;
service: string;
portName: string;
path: string;
}
interface CommonOptions {
targetId: string | null;
full: boolean;
raw: boolean;
}
interface ApplyOptions extends CommonOptions {
confirm: boolean;
dryRun: boolean;
wait: boolean;
}
interface TraceOptions extends CommonOptions {
traceId: string | null;
grep: string | null;
limit: number;
}
interface SearchOptions extends CommonOptions {
grep: string | null;
query: string | null;
limit: number;
candidateLimit: number;
lookbackMinutes: number;
}
interface DiagnoseCodeAgentOptions extends CommonOptions {
businessTraceId: string | null;
traceId: string | null;
limit: number;
candidateLimit: number;
lookbackMinutes: number;
}
export function observabilityHelp(): Record<string, unknown> {
return {
command: "platform-infra observability plan|apply|status|validate|trace|search|diagnose-code-agent",
output: "json",
configTruth: "config/platform-infra/observability.yaml",
spec: "PJ2026-01060501 OTel追踪 draft-2026-06-19-p0",
usage: [
"bun scripts/cli.ts platform-infra observability plan --target D601",
"bun scripts/cli.ts platform-infra observability apply --target D601 --dry-run",
"bun scripts/cli.ts platform-infra observability apply --target D601 --confirm",
"bun scripts/cli.ts platform-infra observability status --target D601 [--full|--raw]",
"bun scripts/cli.ts platform-infra observability validate --target D601 [--full|--raw]",
"bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id <traceId> [--grep provider-stream-disconnected] [--limit 40] [--full|--raw]",
"bun scripts/cli.ts platform-infra observability search --target D601 --grep 'no rollout found' [--lookback-minutes 360] [--candidate-limit 80] [--limit 20] [--full|--raw]",
"bun scripts/cli.ts platform-infra observability diagnose-code-agent --target D601 --business-trace-id <trc_...> [--full|--raw]",
],
boundary: "Prometheus remains the metrics source; this command owns only platform-infra OTel Collector, trace backend readiness, and trace lookup.",
};
}
export async function runPlatformObservabilityCommand(config: UniDeskConfig, args: string[]): Promise<Record<string, unknown>> {
const [action = "plan"] = args;
if (action === "plan") return plan(parseCommonOptions(args.slice(1)));
if (action === "apply") return await apply(config, parseApplyOptions(args.slice(1)));
if (action === "status") return await status(config, parseCommonOptions(args.slice(1)));
if (action === "validate") return await validate(config, parseCommonOptions(args.slice(1)));
if (action === "trace") return await trace(config, parseTraceOptions(args.slice(1)));
if (action === "search") return await search(config, parseSearchOptions(args.slice(1)));
if (action === "diagnose-code-agent") return await diagnoseCodeAgent(config, parseDiagnoseCodeAgentOptions(args.slice(1)));
return { ok: false, error: "unsupported-platform-infra-observability-command", args, help: observabilityHelp() };
}
function parseCommonOptions(args: string[]): CommonOptions {
let targetId: string | null = null;
let full = false;
let raw = false;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--target") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--target requires a value");
if (!/^[A-Za-z0-9._-]+$/u.test(value)) throw new Error("--target must be a simple target id");
targetId = value;
index += 1;
} else if (arg === "--full") {
full = true;
} else if (arg === "--raw") {
raw = true;
full = true;
} else {
throw new Error(`unsupported observability option: ${arg}`);
}
}
return { targetId, full, raw };
}
function parseApplyOptions(args: string[]): ApplyOptions {
const commonArgs: string[] = [];
let confirm = false;
let dryRun = false;
let wait = false;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--confirm") confirm = true;
else if (arg === "--dry-run") dryRun = true;
else if (arg === "--wait") wait = true;
else {
commonArgs.push(arg);
if (arg === "--target") {
commonArgs.push(args[index + 1] ?? "");
index += 1;
}
}
}
if (confirm && dryRun) throw new Error("observability apply accepts only one of --confirm or --dry-run");
return { ...parseCommonOptions(commonArgs), confirm, dryRun: dryRun || !confirm, wait };
}
function parseTraceOptions(args: string[]): TraceOptions {
const commonArgs: string[] = [];
let traceId: string | null = null;
let grep: string | null = null;
let limit = 40;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--trace-id") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--trace-id requires a value");
if (!/^[0-9a-fA-F]{32}$/u.test(value)) throw new Error("--trace-id must be a 32-character OpenTelemetry trace id encoded as hex");
traceId = value;
index += 1;
} else if (arg === "--grep") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--grep requires a value");
grep = value;
index += 1;
} else if (arg === "--limit") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 500) throw new Error("--limit must be an integer from 1 to 500");
limit = parsed;
index += 1;
} else {
commonArgs.push(arg);
if (arg === "--target") {
commonArgs.push(args[index + 1] ?? "");
index += 1;
}
}
}
return { ...parseCommonOptions(commonArgs), traceId, grep, limit };
}
function parseSearchOptions(args: string[]): SearchOptions {
const commonArgs: string[] = [];
let grep: string | null = null;
let query: string | null = null;
let limit = 20;
let candidateLimit = 80;
let lookbackMinutes = 360;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--grep" || arg === "--path") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
grep = value;
index += 1;
} else if (arg === "--query" || arg === "--q") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
query = value;
index += 1;
} else if (arg === "--limit") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 100) throw new Error("--limit must be an integer from 1 to 100");
limit = parsed;
index += 1;
} else if (arg === "--candidate-limit") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--candidate-limit requires a value");
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 500) throw new Error("--candidate-limit must be an integer from 1 to 500");
candidateLimit = parsed;
index += 1;
} else if (arg === "--lookback-minutes" || arg === "--since-minutes") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 10080) throw new Error(`${arg} must be an integer from 1 to 10080`);
lookbackMinutes = parsed;
index += 1;
} else {
commonArgs.push(arg);
if (arg === "--target") {
commonArgs.push(args[index + 1] ?? "");
index += 1;
}
}
}
if (grep === null && query === null) throw new Error("observability search requires --grep <text> or --query <tempo-query>");
return { ...parseCommonOptions(commonArgs), grep, query, limit, candidateLimit, lookbackMinutes };
}
function parseDiagnoseCodeAgentOptions(args: string[]): DiagnoseCodeAgentOptions {
const commonArgs: string[] = [];
let businessTraceId: string | null = null;
let traceId: string | null = null;
let limit = 40;
let candidateLimit = 20;
let lookbackMinutes = 10080;
for (let index = 0; index < args.length; index += 1) {
const arg = args[index];
if (arg === "--business-trace-id" || arg === "--trace") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
if (!/^trc_[A-Za-z0-9_-]+$/u.test(value)) throw new Error(`${arg} must look like trc_<id>`);
businessTraceId = value;
index += 1;
} else if (arg === "--trace-id" || arg === "--otel-trace-id") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
if (!/^[0-9a-fA-F]{32}$/u.test(value)) throw new Error(`${arg} must be a 32-character OpenTelemetry trace id encoded as hex`);
traceId = value.toLowerCase();
index += 1;
} else if (arg === "--limit") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 200) throw new Error("--limit must be an integer from 1 to 200");
limit = parsed;
index += 1;
} else if (arg === "--candidate-limit") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--candidate-limit requires a value");
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 100) throw new Error("--candidate-limit must be an integer from 1 to 100");
candidateLimit = parsed;
index += 1;
} else if (arg === "--lookback-minutes" || arg === "--since-minutes") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
const parsed = Number(value);
if (!Number.isInteger(parsed) || parsed < 1 || parsed > 10080) throw new Error(`${arg} must be an integer from 1 to 10080`);
lookbackMinutes = parsed;
index += 1;
} else {
commonArgs.push(arg);
if (arg === "--target") {
commonArgs.push(args[index + 1] ?? "");
index += 1;
}
}
}
if (businessTraceId === null && traceId === null) throw new Error("observability diagnose-code-agent requires --business-trace-id <trc_...> or --trace-id <otelTraceId>");
return { ...parseCommonOptions(commonArgs), businessTraceId, traceId, limit, candidateLimit, lookbackMinutes };
}
function readObservabilityConfig(): ObservabilityConfig {
const parsed = Bun.YAML.parse(readFileSync(configFile, "utf8")) as unknown;
const root = asRecord(parsed, configLabel);
const version = integerField(root, "version", "");
const kind = stringField(root, "kind", "");
if (kind !== "platform-infra-observability") throw new Error(`${configLabel}.kind must be platform-infra-observability`);
const metadata = objectField(root, "metadata", "");
const defaults = objectField(root, "defaults", "");
const images = objectField(root, "images", "");
const collector = objectField(root, "collector", "");
const collectorOtlp = objectField(collector, "otlp", "collector");
const traceBackend = objectField(root, "traceBackend", "");
const traceBackendOtlp = objectField(traceBackend, "otlp", "traceBackend");
const traceBackendStorage = objectField(traceBackend, "storage", "traceBackend");
const sampling = objectField(root, "sampling", "");
const instrumentation = objectField(root, "instrumentation", "");
const resourceAttributes = objectField(root, "resourceAttributes", "");
const probes = objectField(root, "probes", "");
const config: ObservabilityConfig = {
version,
kind,
metadata: {
id: stringField(metadata, "id", "metadata"),
owner: stringField(metadata, "owner", "metadata"),
spec: stringField(metadata, "spec", "metadata"),
relatedIssues: numberArrayField(metadata, "relatedIssues", "metadata"),
},
defaults: { targetId: stringField(defaults, "targetId", "defaults") },
images: {
collector: imageSpec(objectField(images, "collector", "images"), "images.collector"),
tempo: imageSpec(objectField(images, "tempo", "images"), "images.tempo"),
},
targets: arrayOfRecords(root.targets, "targets").map(parseTarget),
collector: {
deploymentName: kubernetesNameField(collector, "deploymentName", "collector"),
serviceName: kubernetesNameField(collector, "serviceName", "collector"),
configMapName: kubernetesNameField(collector, "configMapName", "collector"),
replicas: integerField(collector, "replicas", "collector"),
healthPort: portField(collector, "healthPort", "collector"),
otlp: parseOtlpPorts(collectorOtlp, "collector.otlp"),
},
traceBackend: {
type: enumField(traceBackend, "type", "traceBackend", ["tempo"] as const),
deploymentName: kubernetesNameField(traceBackend, "deploymentName", "traceBackend"),
serviceName: kubernetesNameField(traceBackend, "serviceName", "traceBackend"),
configMapName: kubernetesNameField(traceBackend, "configMapName", "traceBackend"),
replicas: integerField(traceBackend, "replicas", "traceBackend"),
httpPort: portField(traceBackend, "httpPort", "traceBackend"),
otlp: parseOtlpPorts(traceBackendOtlp, "traceBackend.otlp"),
storage: {
mode: enumField(traceBackendStorage, "mode", "traceBackend.storage", ["emptyDir"] as const),
retention: stringField(traceBackendStorage, "retention", "traceBackend.storage"),
},
},
sampling: {
mode: enumField(sampling, "mode", "sampling", ["parentbased_traceidratio"] as const),
ratio: numberField(sampling, "ratio", "sampling"),
},
instrumentation: {
contextPropagation: stringArrayField(instrumentation, "contextPropagation", "instrumentation"),
serviceConnections: arrayOfRecords(instrumentation.serviceConnections, "instrumentation.serviceConnections").map(parseServiceConnection),
},
resourceAttributes: {
required: stringArrayField(resourceAttributes, "required", "resourceAttributes"),
businessCorrelationAttributes: stringArrayField(resourceAttributes, "businessCorrelationAttributes", "resourceAttributes"),
},
probes: {
readinessPath: apiPathField(probes, "readinessPath", "probes"),
traceQueryPathTemplate: stringField(probes, "traceQueryPathTemplate", "probes"),
statusEndpoints: arrayOfRecords(probes.statusEndpoints, "probes.statusEndpoints").map(parseStatusEndpoint),
},
};
if (config.targets.length === 0) throw new Error(`${configLabel}.targets must not be empty`);
assertKnownEnabledTarget(config.targets, config.defaults.targetId, "defaults.targetId");
if (config.collector.replicas < 0 || config.traceBackend.replicas < 0) throw new Error(`${configLabel} replicas must be >= 0`);
if (config.sampling.ratio < 0 || config.sampling.ratio > 1) throw new Error(`${configLabel}.sampling.ratio must be between 0 and 1`);
if (!config.probes.traceQueryPathTemplate.includes("{{traceId}}")) throw new Error(`${configLabel}.probes.traceQueryPathTemplate must include {{traceId}}`);
return config;
}
function imageSpec(record: Record<string, unknown>, path: string): ImageSpec {
const image = {
repository: stringField(record, "repository", path),
tag: stringField(record, "tag", path),
pullPolicy: enumField(record, "pullPolicy", path, ["Always", "IfNotPresent", "Never"] as const),
};
if (!/^[A-Za-z0-9._/:@-]+$/u.test(`${image.repository}:${image.tag}`)) throw new Error(`${configLabel}.${path} must render a valid image reference`);
return image;
}
function parseTarget(record: Record<string, unknown>, index: number): ObservabilityTarget {
const path = `targets[${index}]`;
return {
id: stringField(record, "id", path),
route: stringField(record, "route", path),
namespace: kubernetesNameField(record, "namespace", path),
role: enumField(record, "role", path, ["active", "standby"] as const),
enabled: booleanField(record, "enabled", path),
createNamespace: booleanField(record, "createNamespace", path),
};
}
function parseOtlpPorts(record: Record<string, unknown>, path: string): OtlpPorts {
return {
grpcPort: portField(record, "grpcPort", path),
httpPort: portField(record, "httpPort", path),
};
}
function parseServiceConnection(record: Record<string, unknown>, index: number): ServiceConnection {
const path = `instrumentation.serviceConnections[${index}]`;
return {
serviceName: stringField(record, "serviceName", path),
owningRepo: stringField(record, "owningRepo", path),
targetNode: stringField(record, "targetNode", path),
lane: stringField(record, "lane", path),
namespace: kubernetesNameField(record, "namespace", path),
requiredSpans: stringArrayField(record, "requiredSpans", path),
};
}
function parseStatusEndpoint(record: Record<string, unknown>, index: number): StatusEndpoint {
const path = `probes.statusEndpoints[${index}]`;
return {
name: stringField(record, "name", path),
service: kubernetesNameField(record, "service", path),
portName: stringField(record, "portName", path),
path: apiPathField(record, "path", path),
};
}
function assertKnownEnabledTarget(targets: ObservabilityTarget[], targetId: string, path: string): void {
const target = targets.find((item) => item.id.toLowerCase() === targetId.toLowerCase());
if (target === undefined) throw new Error(`${configLabel}.${path} references unknown target ${targetId}; known targets: ${targets.map((item) => item.id).join(", ")}`);
if (!target.enabled) throw new Error(`${configLabel}.${path} references disabled target ${target.id}`);
}
function resolveTarget(observability: ObservabilityConfig, targetId: string | null): ObservabilityTarget {
const resolved = targetId ?? observability.defaults.targetId;
const target = observability.targets.find((item) => item.id.toLowerCase() === resolved.toLowerCase());
if (target === undefined) throw new Error(`unknown observability target ${resolved}; known targets: ${observability.targets.map((item) => item.id).join(", ")}`);
if (!target.enabled) throw new Error(`observability target ${target.id} is disabled in ${configLabel}`);
return target;
}
function plan(options: CommonOptions): Record<string, unknown> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const yaml = renderManifest(observability, target);
const policy = policyChecks(yaml, target);
return {
ok: policy.every((check) => check.ok),
action: "platform-infra-observability-plan",
mutation: false,
config: configSummary(observability, target),
renderPlan: {
target: targetSummary(target),
objects: manifestObjectSummary(yaml),
otlp: {
collectorGrpcEndpoint: `${observability.collector.serviceName}.${target.namespace}.svc.cluster.local:${observability.collector.otlp.grpcPort}`,
collectorHttpEndpoint: `http://${observability.collector.serviceName}.${target.namespace}.svc.cluster.local:${observability.collector.otlp.httpPort}`,
backendGrpcEndpoint: `${observability.traceBackend.serviceName}.${target.namespace}.svc.cluster.local:${observability.traceBackend.otlp.grpcPort}`,
},
instrumentation: observability.instrumentation.serviceConnections,
resourceAttributes: observability.resourceAttributes,
},
policy,
next: {
dryRun: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --dry-run`,
apply: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --confirm`,
status: `bun scripts/cli.ts platform-infra observability status --target ${target.id}`,
validate: `bun scripts/cli.ts platform-infra observability validate --target ${target.id}`,
trace: `bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id <traceId>`,
},
};
}
async function apply(config: UniDeskConfig, options: ApplyOptions): Promise<Record<string, unknown>> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const yaml = renderManifest(observability, target);
const policy = policyChecks(yaml, target);
if (!policy.every((check) => check.ok)) return { ok: false, action: "platform-infra-observability-apply", mode: "policy-blocked", policy };
const result = await capture(config, target.route, ["sh"], applyScript({
yaml,
target,
dryRun: options.dryRun,
wait: options.wait,
collectorDeploymentName: observability.collector.deploymentName,
backendDeploymentName: observability.traceBackend.deploymentName,
}));
const parsed = parseJsonOutput(result.stdout);
return {
ok: result.exitCode === 0 && parsed?.ok === true,
action: "platform-infra-observability-apply",
mode: options.dryRun ? "dry-run" : "confirmed",
mutation: !options.dryRun,
target: targetSummary(target),
policy,
remote: parsed ?? compactCapture(result, { full: true }),
};
}
async function status(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const result = await capture(config, target.route, ["sh"], statusScript(observability, target, options.full));
const parsed = parseJsonOutput(result.stdout);
const summary = parsed === null ? null : statusSummary(parsed);
return {
ok: result.exitCode === 0 && summary?.ready === true,
action: "platform-infra-observability-status",
mutation: false,
target: targetSummary(target),
summary,
remote: options.raw ? parsed : compactStatus(parsed, options.full) ?? compactCapture(result, { full: true }),
next: {
plan: `bun scripts/cli.ts platform-infra observability plan --target ${target.id}`,
apply: `bun scripts/cli.ts platform-infra observability apply --target ${target.id} --confirm`,
validate: `bun scripts/cli.ts platform-infra observability validate --target ${target.id}`,
},
};
}
async function validate(config: UniDeskConfig, options: CommonOptions): Promise<Record<string, unknown>> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const result = await capture(config, target.route, ["sh"], statusScript(observability, target, options.full, true));
const parsed = parseJsonOutput(result.stdout);
const summary = parsed === null ? null : statusSummary(parsed);
const ready = summary?.ready === true;
const validationTrace = parsed !== null && typeof parsed.validationTrace === "object" && parsed.validationTrace !== null && !Array.isArray(parsed.validationTrace)
? parsed.validationTrace as Record<string, unknown>
: null;
const validationTraceId = typeof validationTrace?.traceId === "string" ? validationTrace.traceId : null;
const validationTraceOk = validationTrace?.ok === true;
return {
ok: result.exitCode === 0 && ready && validationTraceOk,
action: "platform-infra-observability-validate",
mutation: false,
target: targetSummary(target),
summary,
validation: {
readiness: ready ? "passed" : "failed",
testTrace: validationTrace ?? "not-generated-runtime-not-ready",
traceQuery: validationTraceId !== null
? `bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${validationTraceId}`
: "blocked-until-validation-trace-generated",
metricsBoundary: "Prometheus/RUM remains outside this trace readiness check.",
},
remote: options.raw ? parsed : compactStatus(parsed, options.full) ?? compactCapture(result, { full: true }),
};
}
async function trace(config: UniDeskConfig, options: TraceOptions): Promise<Record<string, unknown> | RenderedCliResult> {
if (options.traceId === null) throw new Error("observability trace requires --trace-id <traceId>");
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const tracePath = observability.probes.traceQueryPathTemplate.replaceAll("{{traceId}}", encodeURIComponent(options.traceId));
const result = await capture(config, target.route, ["sh"], traceScript(observability, target, tracePath, options));
const parsed = parseJsonOutput(result.stdout);
const response = {
ok: result.exitCode === 0 && parsed?.ok === true,
action: "platform-infra-observability-trace",
mutation: false,
target: targetSummary(target),
traceId: options.traceId,
query: {
backend: observability.traceBackend.type,
service: observability.traceBackend.serviceName,
path: tracePath,
},
result: redactSensitiveUnknown(parsed),
};
if (options.raw || options.full) return response;
return renderTraceResult(response);
}
async function search(config: UniDeskConfig, options: SearchOptions): Promise<Record<string, unknown> | RenderedCliResult> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const endSeconds = Math.floor(Date.now() / 1000);
const startSeconds = endSeconds - (options.lookbackMinutes * 60);
const params = new URLSearchParams({
start: String(startSeconds),
end: String(endSeconds),
limit: String(options.candidateLimit),
});
if (options.query !== null) params.set("q", options.query);
const searchPath = `/api/search?${params.toString()}`;
const result = await capture(config, target.route, ["sh"], searchScript(observability, target, searchPath, options));
const parsed = parseJsonOutput(result.stdout);
const response = {
ok: result.exitCode === 0 && parsed?.ok === true,
action: "platform-infra-observability-search",
mutation: false,
target: targetSummary(target),
query: {
backend: observability.traceBackend.type,
service: observability.traceBackend.serviceName,
path: searchPath,
grep: options.grep,
tempoQuery: options.query,
lookbackMinutes: options.lookbackMinutes,
candidateLimit: options.candidateLimit,
limit: options.limit,
},
result: redactSensitiveUnknown(parsed) ?? compactCapture(result, { full: true }),
};
if (options.raw || options.full) return response;
return renderSearchResult(response);
}
async function diagnoseCodeAgent(config: UniDeskConfig, options: DiagnoseCodeAgentOptions): Promise<Record<string, unknown> | RenderedCliResult> {
const observability = readObservabilityConfig();
const target = resolveTarget(observability, options.targetId);
const endSeconds = Math.floor(Date.now() / 1000);
const startSeconds = endSeconds - (options.lookbackMinutes * 60);
let searchPath: string | null = null;
if (options.traceId === null && options.businessTraceId !== null) {
const params = new URLSearchParams({
start: String(startSeconds),
end: String(endSeconds),
limit: String(options.candidateLimit),
q: `{ .traceId = "${options.businessTraceId}" }`,
});
searchPath = `/api/search?${params.toString()}`;
}
const result = await capture(config, target.route, ["sh"], diagnoseCodeAgentScript(observability, target, searchPath, options));
const parsed = parseJsonOutput(result.stdout);
const response = {
ok: result.exitCode === 0 && parsed?.ok === true,
action: "platform-infra-observability-diagnose-code-agent",
mutation: false,
target: targetSummary(target),
query: {
backend: observability.traceBackend.type,
service: observability.traceBackend.serviceName,
businessTraceId: options.businessTraceId,
traceId: options.traceId,
path: searchPath,
lookbackMinutes: options.lookbackMinutes,
candidateLimit: options.candidateLimit,
limit: options.limit,
},
result: redactSensitiveUnknown(parsed) ?? compactCapture(result, { full: true }),
};
if (options.raw || options.full) return response;
return renderDiagnoseCodeAgentResult(response);
}
function renderDiagnoseCodeAgentResult(response: Record<string, unknown>): RenderedCliResult {
const target = recordValue(response.target);
const query = recordValue(response.query);
const result = recordValue(response.result);
const mapping = recordValue(result.mapping);
const summary = recordValue(result.summary);
const identity = recordValue(result.identity);
const agentrun = recordValue(result.agentrun);
const http = recordValue(result.http);
const servicePath = recordValue(result.servicePath);
const selection = recordValue(mapping.selection);
const candidates = arrayValue(mapping.candidateScores);
const rootCandidates = arrayValue(summary.rootCauseCandidates);
const problemCounts = arrayValue(http.problemCounts);
const lines: string[] = [];
lines.push("COMMAND TARGET OK BUSINESS_TRACE OTEL_TRACE");
lines.push(`${pad("platform-infra observability diagnose", 44)} ${pad(textValue(target.id) || "-", 7)} ${pad(response.ok === true ? "true" : "false", 6)} ${pad(textValue(query.businessTraceId) || "-", 27)} ${textValue(mapping.otelTraceId) || "-"}`);
lines.push("");
lines.push("TRACE_SELECTION");
lines.push(`mode: ${textValue(selection.mode) || textValue(mapping.mode) || "-"}`);
lines.push(`candidateTraceCount: ${textValue(mapping.candidateTraceCount) || "-"}`);
lines.push(`selectedScore: ${textValue(selection.score) || "-"}`);
lines.push(`selectedReason: ${arrayValue(selection.reasons).map((item) => textValue(item)).filter(Boolean).join(", ") || "-"}`);
if (textValue(mapping.selectionWarning)) lines.push(`warning: ${textValue(mapping.selectionWarning)}`);
lines.push("");
lines.push("SERVICE_PATH");
for (const service of ["hwlab-cloud-api", "agentrun-manager", "agentrun-runner"]) {
lines.push(`${pad(service, 22)} ${textValue(servicePath[service]) || "missing"}`);
}
lines.push("");
lines.push("AGENTRUN");
lines.push(`terminalStatus: ${textValue(agentrun.terminalStatus) || "-"}`);
lines.push(`terminalEventType: ${textValue(agentrun.terminalEventType) || "-"}`);
lines.push(`runnerProviderClassification: ${textValue(agentrun.runnerProviderClassification) || "-"}`);
lines.push("");
lines.push("IDENTITY");
for (const key of ["runId", "commandId", "runnerJobId", "runnerId", "sessionId", "turnId", "backendProfile", "sourceCommit"]) {
const value = textValue(identity[key]);
if (value) lines.push(`${pad(key, 16)} ${value}`);
}
if (lines[lines.length - 1] === "IDENTITY") lines.push("-");
lines.push("");
lines.push("ROOT_CAUSE");
lines.push(textValue(summary.rootCause) || "-");
for (const item of rootCandidates.slice(0, 5)) {
const row = recordValue(item);
lines.push(`- ${textValue(row.code) || textValue(row.label) || "candidate"} confidence=${textValue(row.confidence) || "-"} ${textValue(row.summary) || ""}`.trim());
}
lines.push("");
lines.push("HTTP_PROBLEMS");
if (problemCounts.length === 0) {
lines.push("-");
} else {
lines.push("METHOD ROUTE STATUS COUNT");
for (const item of problemCounts.slice(0, 8)) {
const row = recordValue(item);
lines.push(`${pad(textValue(row.method) || "-", 7)} ${pad(truncate(textValue(row.route) || "-", 29), 29)} ${pad(textValue(row.status) || "-", 7)} ${textValue(row.count) || "-"}`);
}
}
if (candidates.length > 0) {
lines.push("");
lines.push("CANDIDATE_TRACE_SCORES");
lines.push("SCORE SPANS SERVICES TRACE");
for (const item of candidates.slice(0, 8)) {
const row = recordValue(item);
lines.push(`${pad(textValue(row.score) || "0", 6)} ${pad(textValue(row.spanCount) || "-", 6)} ${pad(truncate(arrayValue(row.services).map((value) => textValue(value)).filter(Boolean).join(","), 32), 32)} ${textValue(row.traceId) || "-"}`);
}
}
lines.push("");
lines.push("Next:");
if (textValue(mapping.otelTraceId)) {
lines.push(` bun scripts/cli.ts platform-infra observability trace --target ${textValue(target.id) || textValue(query.target) || "D601"} --trace-id ${textValue(mapping.otelTraceId)} --limit 40`);
lines.push(` bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${textValue(target.id) || "D601"} --business-trace-id ${textValue(query.businessTraceId) || "<trc_...>"} --full`);
} else {
lines.push(` bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${textValue(target.id) || "D601"} --business-trace-id ${textValue(query.businessTraceId) || "<trc_...>"} --lookback-minutes 10080 --candidate-limit 100`);
}
return {
ok: response.ok === true,
command: "platform-infra observability diagnose-code-agent",
contentType: "text/plain",
renderedText: `${lines.join("\n")}\n`,
};
}
function renderSearchResult(response: Record<string, unknown>): RenderedCliResult {
const target = recordValue(response.target);
const query = recordValue(response.query);
const result = recordValue(response.result);
const traces = arrayValue(result.traces);
const lines: string[] = [];
lines.push("COMMAND TARGET OK GREP_OR_QUERY MATCHED SCANNED CANDIDATES");
lines.push(`${pad("platform-infra observability search", 38)} ${pad(textValue(target.id) || "-", 7)} ${pad(response.ok === true ? "true" : "false", 6)} ${pad(truncate(textValue(query.grep) || textValue(query.tempoQuery) || "-", 35), 35)} ${pad(textValue(result.matchedTraceCount) || "-", 8)} ${pad(textValue(result.scannedTraceCount) || "-", 8)} ${textValue(result.candidateTraceCount) || "-"}`);
if (textValue(result.scanStopped)) lines.push(`warning: scanStopped=${textValue(result.scanStopped)}`);
const truncated = recordValue(result.truncated);
if (truncated.candidateTraces === true || truncated.matchedTraces === true) {
lines.push(`warning: truncated candidateTraces=${textValue(truncated.candidateTraces)} matchedTraces=${textValue(truncated.matchedTraces)}`);
}
lines.push("");
lines.push("TRACES");
if (traces.length === 0) {
lines.push("-");
} else {
lines.push("TRACE SPANS ERRORS MATCH SERVICES ROOT");
for (const item of traces.slice(0, 20)) {
const row = recordValue(item);
const meta = recordValue(row.meta);
const services = arrayValue(row.services).map((value) => textValue(value)).filter(Boolean).join(",");
lines.push(`${pad(textValue(row.traceId) || "-", 34)} ${pad(textValue(row.spanCount) || "-", 6)} ${pad(textValue(row.errorSpanCount) || "-", 7)} ${pad(textValue(row.matchedSpanCount) || "-", 6)} ${pad(truncate(services || textValue(meta.rootServiceName) || "-", 32), 32)} ${truncate(textValue(meta.rootTraceName) || "-", 48)}`);
}
}
lines.push("");
lines.push("Next:");
lines.push(` bun scripts/cli.ts platform-infra observability search --target ${textValue(target.id) || "D601"} --grep <text> --lookback-minutes ${textValue(query.lookbackMinutes) || "360"} --candidate-limit ${textValue(query.candidateLimit) || "80"} --limit ${textValue(query.limit) || "20"} --full`);
lines.push(` bun scripts/cli.ts platform-infra observability trace --target ${textValue(target.id) || "D601"} --trace-id <traceId> --grep <text> --limit 80`);
return {
ok: response.ok === true,
command: "platform-infra observability search",
contentType: "text/plain",
renderedText: `${lines.join("\n")}\n`,
};
}
function renderTraceResult(response: Record<string, unknown>): RenderedCliResult {
const target = recordValue(response.target);
const traceId = textValue(response.traceId);
const result = recordValue(response.result);
const services = arrayValue(result.services).map((value) => textValue(value)).filter(Boolean);
const businessTraceIds = arrayValue(result.businessTraceIds).map((value) => textValue(value)).filter(Boolean);
const spanNameCounts = arrayValue(result.spanNameCounts);
const errorSpans = arrayValue(result.errorSpans);
const matchedSpans = arrayValue(result.spans).length > 0 ? arrayValue(result.spans) : arrayValue(result.matchedSpans);
const lines: string[] = [];
lines.push("COMMAND TARGET OK TRACE SPANS ERRORS MATCH");
lines.push(`${pad("platform-infra observability trace", 37)} ${pad(textValue(target.id) || "-", 7)} ${pad(response.ok === true ? "true" : "false", 6)} ${pad(traceId || "-", 34)} ${pad(textValue(result.spanCount) || "-", 6)} ${pad(textValue(result.errorSpanCount) || "-", 7)} ${textValue(result.matchedSpanCount) || "-"}`);
lines.push("");
lines.push(`services: ${services.join(", ") || "-"}`);
lines.push(`businessTraceIds: ${businessTraceIds.join(", ") || "-"}`);
if (textValue(result.grep)) lines.push(`grep: ${textValue(result.grep)}`);
const truncated = recordValue(result.truncated);
if (truncated.errorSpans === true || truncated.spans === true) {
lines.push(`warning: truncated errorSpans=${textValue(truncated.errorSpans)} spans=${textValue(truncated.spans)}`);
}
lines.push("");
lines.push("SPAN_NAME_COUNTS");
if (spanNameCounts.length === 0) {
lines.push("-");
} else {
lines.push("COUNT NAME");
for (const item of spanNameCounts.slice(0, 12)) {
const row = recordValue(item);
lines.push(`${pad(textValue(row.count) || "-", 6)} ${truncate(textValue(row.name) || "-", 80)}`);
}
}
lines.push("");
lines.push("ERROR_SPANS");
if (errorSpans.length === 0) {
lines.push("-");
} else {
lines.push("SERVICE DURATION_MS NAME DETAIL");
for (const item of errorSpans.slice(0, 8)) {
const row = recordValue(item);
lines.push(`${pad(truncate(textValue(row.service) || "-", 23), 23)} ${pad(textValue(row.durationMs) || "-", 12)} ${pad(truncate(textValue(row.name) || "-", 40), 40)} ${truncate(traceSpanDetail(row), 96)}`);
}
}
if (matchedSpans.length > 0) {
lines.push("");
lines.push("MATCHED_SPANS");
lines.push("SERVICE DURATION_MS NAME DETAIL");
for (const item of matchedSpans.slice(0, 8)) {
const row = recordValue(item);
lines.push(`${pad(truncate(textValue(row.service) || "-", 23), 23)} ${pad(textValue(row.durationMs) || "-", 12)} ${pad(truncate(textValue(row.name) || "-", 40), 40)} ${truncate(traceSpanDetail(row), 96)}`);
}
}
lines.push("");
lines.push("Next:");
lines.push(` bun scripts/cli.ts platform-infra observability trace --target ${textValue(target.id) || "D601"} --trace-id ${traceId || "<otelTraceId>"} --grep <text> --limit 80`);
lines.push(` bun scripts/cli.ts platform-infra observability trace --target ${textValue(target.id) || "D601"} --trace-id ${traceId || "<otelTraceId>"} --full --limit 200`);
return {
ok: response.ok === true,
command: "platform-infra observability trace",
contentType: "text/plain",
renderedText: `${lines.join("\n")}\n`,
};
}
function traceSpanDetail(span: Record<string, unknown>): string {
const attrs = recordValue(span.attributes);
const parts: string[] = [];
for (const key of [
"failureKind",
"message",
"eventType",
"toolName",
"exitCode",
"durationMs",
"waitingFor",
"lastEventLabel",
"idleMs",
"idleSeconds",
"idleThresholdMs",
"lastEventAgeMs",
"terminalStatus",
"http.method",
"http.route",
"http.status_code",
"runId",
"commandId",
"runnerJobId",
]) {
const value = textValue(attrs[key]);
if (value) parts.push(`${key}=${value}`);
}
return parts.join(" ") || "-";
}
function recordValue(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as Record<string, unknown> : {};
}
function arrayValue(value: unknown): unknown[] {
return Array.isArray(value) ? value : [];
}
function textValue(value: unknown): string {
if (value === null || value === undefined) return "";
if (typeof value === "string") return value;
if (typeof value === "number" || typeof value === "boolean") return String(value);
return "";
}
function pad(value: string, width: number): string {
return value.length >= width ? value : `${value}${" ".repeat(width - value.length)}`;
}
function truncate(value: string, width: number): string {
if (value.length <= width) return value;
if (width <= 1) return value.slice(0, width);
if (width <= 3) return value.slice(0, width);
return `${value.slice(0, width - 3)}...`;
}
function renderManifest(observability: ObservabilityConfig, target: ObservabilityTarget): string {
const collectorImage = imageReference(observability.images.collector);
const tempoImage = imageReference(observability.images.tempo);
return [
target.createNamespace ? namespaceManifest(target) : "",
allowAllNetworkPolicy(target),
collectorConfigMap(observability, target),
collectorDeployment(observability, target, collectorImage),
collectorService(observability, target),
tempoConfigMap(observability, target),
tempoDeployment(observability, target, tempoImage),
tempoService(observability, target),
].filter((item) => item.trim().length > 0).join("\n---\n");
}
function namespaceManifest(target: ObservabilityTarget): string {
return `apiVersion: v1
kind: Namespace
metadata:
name: ${target.namespace}
labels:
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
unidesk.ai/runtime-node: ${target.id}
`;
}
function allowAllNetworkPolicy(target: ObservabilityTarget): string {
return `apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: allow-all
namespace: ${target.namespace}
labels:
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
spec:
podSelector: {}
policyTypes:
- Ingress
- Egress
ingress:
- {}
egress:
- {}
`;
}
function collectorConfigMap(observability: ObservabilityConfig, target: ObservabilityTarget): string {
const config = `receivers:
otlp:
protocols:
grpc:
endpoint: 0.0.0.0:${observability.collector.otlp.grpcPort}
http:
endpoint: 0.0.0.0:${observability.collector.otlp.httpPort}
processors:
batch: {}
exporters:
otlp/tempo:
endpoint: ${observability.traceBackend.serviceName}.${target.namespace}.svc.cluster.local:${observability.traceBackend.otlp.grpcPort}
tls:
insecure: true
extensions:
health_check:
endpoint: 0.0.0.0:${observability.collector.healthPort}
service:
extensions: [health_check]
pipelines:
traces:
receivers: [otlp]
processors: [batch]
exporters: [otlp/tempo]
`;
return `apiVersion: v1
kind: ConfigMap
metadata:
name: ${observability.collector.configMapName}
namespace: ${target.namespace}
labels:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
annotations:
unidesk.ai/spec: "${observability.metadata.spec}"
data:
collector.yaml: |
${indent(config, 4)}
`;
}
function collectorDeployment(observability: ObservabilityConfig, target: ObservabilityTarget, image: string): string {
return `apiVersion: apps/v1
kind: Deployment
metadata:
name: ${observability.collector.deploymentName}
namespace: ${target.namespace}
labels:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
spec:
replicas: ${observability.collector.replicas}
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 0
maxUnavailable: 1
selector:
matchLabels:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
template:
metadata:
labels:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
app.kubernetes.io/part-of: platform-infra
annotations:
unidesk.ai/spec: "${observability.metadata.spec}"
spec:
containers:
- name: collector
image: ${image}
imagePullPolicy: ${observability.images.collector.pullPolicy}
args:
- --config=/etc/otelcol/collector.yaml
ports:
- name: otlp-grpc
containerPort: ${observability.collector.otlp.grpcPort}
- name: otlp-http
containerPort: ${observability.collector.otlp.httpPort}
- name: health
containerPort: ${observability.collector.healthPort}
readinessProbe:
httpGet:
path: /
port: health
volumeMounts:
- name: config
mountPath: /etc/otelcol/collector.yaml
subPath: collector.yaml
readOnly: true
volumes:
- name: config
configMap:
name: ${observability.collector.configMapName}
`;
}
function collectorService(observability: ObservabilityConfig, target: ObservabilityTarget): string {
return `apiVersion: v1
kind: Service
metadata:
name: ${observability.collector.serviceName}
namespace: ${target.namespace}
labels:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
spec:
type: ClusterIP
selector:
app.kubernetes.io/name: ${observability.collector.deploymentName}
app.kubernetes.io/component: tracing
ports:
- name: otlp-grpc
port: ${observability.collector.otlp.grpcPort}
targetPort: otlp-grpc
- name: otlp-http
port: ${observability.collector.otlp.httpPort}
targetPort: otlp-http
- name: health
port: ${observability.collector.healthPort}
targetPort: health
`;
}
function tempoConfigMap(observability: ObservabilityConfig, _target: ObservabilityTarget): string {
const config = `server:
http_listen_port: ${observability.traceBackend.httpPort}
distributor:
receivers:
otlp:
protocols:
grpc:
endpoint: 0.0.0.0:${observability.traceBackend.otlp.grpcPort}
http:
endpoint: 0.0.0.0:${observability.traceBackend.otlp.httpPort}
ingester:
trace_idle_period: 10s
max_block_duration: 5m
compactor:
compaction:
block_retention: ${observability.traceBackend.storage.retention}
storage:
trace:
backend: local
wal:
path: /var/tempo/wal
local:
path: /var/tempo/traces
`;
return `apiVersion: v1
kind: ConfigMap
metadata:
name: ${observability.traceBackend.configMapName}
namespace: ${_target.namespace}
labels:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
annotations:
unidesk.ai/spec: "${observability.metadata.spec}"
data:
tempo.yaml: |
${indent(config, 4)}
`;
}
function tempoDeployment(observability: ObservabilityConfig, target: ObservabilityTarget, image: string): string {
return `apiVersion: apps/v1
kind: Deployment
metadata:
name: ${observability.traceBackend.deploymentName}
namespace: ${target.namespace}
labels:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
spec:
replicas: ${observability.traceBackend.replicas}
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 0
maxUnavailable: 1
selector:
matchLabels:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
template:
metadata:
labels:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
app.kubernetes.io/part-of: platform-infra
annotations:
unidesk.ai/spec: "${observability.metadata.spec}"
spec:
containers:
- name: tempo
image: ${image}
imagePullPolicy: ${observability.images.tempo.pullPolicy}
args:
- -config.file=/etc/tempo/tempo.yaml
ports:
- name: http
containerPort: ${observability.traceBackend.httpPort}
- name: otlp-grpc
containerPort: ${observability.traceBackend.otlp.grpcPort}
- name: otlp-http
containerPort: ${observability.traceBackend.otlp.httpPort}
readinessProbe:
httpGet:
path: ${observability.probes.readinessPath}
port: http
volumeMounts:
- name: config
mountPath: /etc/tempo/tempo.yaml
subPath: tempo.yaml
readOnly: true
- name: data
mountPath: /var/tempo
volumes:
- name: config
configMap:
name: ${observability.traceBackend.configMapName}
- name: data
emptyDir: {}
`;
}
function tempoService(observability: ObservabilityConfig, target: ObservabilityTarget): string {
return `apiVersion: v1
kind: Service
metadata:
name: ${observability.traceBackend.serviceName}
namespace: ${target.namespace}
labels:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
app.kubernetes.io/part-of: platform-infra
app.kubernetes.io/managed-by: unidesk
spec:
type: ClusterIP
selector:
app.kubernetes.io/name: ${observability.traceBackend.deploymentName}
app.kubernetes.io/component: trace-backend
ports:
- name: http
port: ${observability.traceBackend.httpPort}
targetPort: http
- name: otlp-grpc
port: ${observability.traceBackend.otlp.grpcPort}
targetPort: otlp-grpc
- name: otlp-http
port: ${observability.traceBackend.otlp.httpPort}
targetPort: otlp-http
`;
}
function applyScript(params: {
yaml: string;
target: ObservabilityTarget;
dryRun: boolean;
wait: boolean;
collectorDeploymentName: string;
backendDeploymentName: string;
}): string {
const encoded = Buffer.from(params.yaml, "utf8").toString("base64");
const dryRunArg = params.dryRun ? "--dry-run=server" : "";
const waitDisposition = params.wait ? "skipped-short-connection-use-status" : "skipped";
return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
manifest="$tmp/platform-infra-observability.yaml"
printf '%s' '${encoded}' | base64 -d > "$manifest"
kubectl apply --server-side --field-manager=${fieldManager} ${dryRunArg} -f "$manifest" >"$tmp/apply.out" 2>"$tmp/apply.err"
apply_rc=$?
collector_rollout_rc=0
backend_rollout_rc=0
wait_disposition=${waitDisposition}
python3 - "$apply_rc" "$collector_rollout_rc" "$backend_rollout_rc" "$wait_disposition" "$tmp/apply.out" "$tmp/apply.err" "$tmp/collector-rollout.out" "$tmp/collector-rollout.err" "$tmp/backend-rollout.out" "$tmp/backend-rollout.err" <<'PY'
import json, os, sys
def text(path, limit=6000):
try:
return open(path, encoding="utf-8", errors="replace").read()[-limit:]
except FileNotFoundError:
return ""
apply_rc = int(sys.argv[1])
collector_rc = int(sys.argv[2])
backend_rc = int(sys.argv[3])
payload = {
"ok": apply_rc == 0 and collector_rc == 0 and backend_rc == 0,
"target": "${params.target.id}",
"namespace": "${params.target.namespace}",
"dryRun": ${params.dryRun ? "True" : "False"},
"apply": {"exitCode": apply_rc, "stdout": text(sys.argv[5]), "stderr": text(sys.argv[6])},
"rollout": {
"disposition": sys.argv[4],
"collector": {"exitCode": collector_rc, "stdout": text(sys.argv[7]), "stderr": text(sys.argv[8])},
"backend": {"exitCode": backend_rc, "stdout": text(sys.argv[9]), "stderr": text(sys.argv[10])},
},
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}
function statusScript(observability: ObservabilityConfig, target: ObservabilityTarget, full: boolean, generateValidationTrace = false): string {
const endpointsJson = JSON.stringify(observability.probes.statusEndpoints);
return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
capture_json() {
name="$1"
shift
"$@" >"$tmp/$name.json" 2>"$tmp/$name.err"
echo $? >"$tmp/$name.rc"
}
capture_raw() {
name="$1"
shift
"$@" >"$tmp/$name.out" 2>"$tmp/$name.err"
echo $? >"$tmp/$name.rc"
}
capture_json namespace kubectl get namespace ${shQuote(target.namespace)} -o json
capture_json deployments kubectl -n ${shQuote(target.namespace)} get deployment ${shQuote(observability.collector.deploymentName)} ${shQuote(observability.traceBackend.deploymentName)} -o json
capture_json services kubectl -n ${shQuote(target.namespace)} get service ${shQuote(observability.collector.serviceName)} ${shQuote(observability.traceBackend.serviceName)} -o json
capture_json pods kubectl -n ${shQuote(target.namespace)} get pods -l ${shQuote(`app.kubernetes.io/name in (${observability.collector.deploymentName},${observability.traceBackend.deploymentName})`)} -o json
capture_json events kubectl -n ${shQuote(target.namespace)} get events -o json
python3 - "$tmp" '${endpointsJson.replaceAll("'", "'\"'\"'")}' <<'PY'
import json, random, socket, subprocess, sys, time, urllib.error, urllib.request
tmp = sys.argv[1]
endpoints = json.loads(sys.argv[2])
namespace = "${target.namespace}"
def read(path, binary=False, limit=8000):
try:
mode = "rb" if binary else "r"
with open(path, mode, encoding=None if binary else "utf-8", errors=None if binary else "replace") as fh:
data = fh.read()
if binary:
data = data.decode("utf-8", errors="replace")
return data[-limit:]
except FileNotFoundError:
return ""
def rc(name):
try:
return int(read(f"{tmp}/{name}.rc").strip() or "1")
except ValueError:
return 1
def parsed_json(name):
raw = read(f"{tmp}/{name}.json", limit=1000000)
try:
return json.loads(raw) if raw else None
except Exception:
return None
def compact_section(name):
return {
"exitCode": rc(name),
"stderrTail": read(f"{tmp}/{name}.err", limit=2000),
"json": parsed_json(name),
}
probe_results = []
for ep in endpoints:
path = ep.get("path") or "/"
if not path.startswith("/"):
path = "/" + path
proxy_path = f"/api/v1/namespaces/{namespace}/services/http:{ep['service']}:{ep['portName']}/proxy{path}"
proc = subprocess.run(["kubectl", "get", "--raw", proxy_path], text=True, capture_output=True, timeout=20)
probe_results.append({
"name": ep["name"],
"service": ep["service"],
"portName": ep["portName"],
"path": path,
"exitCode": proc.returncode,
"stdoutTail": proc.stdout[-1000:],
"stderrTail": proc.stderr[-1000:],
"ok": proc.returncode == 0,
})
def free_port():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(("127.0.0.1", 0))
port = sock.getsockname()[1]
sock.close()
return port
def parse_body(raw):
try:
return json.loads(raw) if raw else None
except Exception:
return raw[-2000:]
def http_request(method, url, body=None, timeout=10):
data = None
headers = {}
if body is not None:
data = json.dumps(body).encode("utf-8")
headers["Content-Type"] = "application/json"
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
raw = resp.read(200000).decode("utf-8", errors="replace")
status = int(getattr(resp, "status", 0) or 0)
return {"ok": 200 <= status < 300, "status": status, "body": parse_body(raw)}
except urllib.error.HTTPError as exc:
raw = exc.read(200000).decode("utf-8", errors="replace")
return {"ok": False, "status": exc.code, "body": parse_body(raw)}
except Exception as exc:
return {"ok": False, "error": type(exc).__name__ + ": " + str(exc)}
def wait_proxy(base_url):
last = None
for _ in range(30):
last = http_request("GET", base_url + "/version", timeout=1)
if last.get("ok") is True:
return last
time.sleep(0.2)
return last or {"ok": False, "error": "kubectl proxy did not answer"}
def compact_http(result):
if not isinstance(result, dict):
return {"ok": False, "error": "missing-result"}
compact = {"ok": result.get("ok") is True}
if "status" in result:
compact["status"] = result.get("status")
if "error" in result:
compact["error"] = result.get("error")
body = result.get("body")
if isinstance(body, dict):
compact["bodyKeys"] = sorted(list(body.keys()))[:12]
elif isinstance(body, str) and body:
compact["bodyTail"] = body[-1000:]
return compact
def validation_trace_payload(trace_id, span_id):
now = time.time_ns()
return {
"resourceSpans": [{
"resource": {"attributes": [
{"key": "service.name", "value": {"stringValue": "unidesk-observability-validate"}},
{"key": "deployment.environment", "value": {"stringValue": "${target.id}"}},
{"key": "unidesk.node", "value": {"stringValue": "${target.id}"}},
{"key": "k8s.namespace.name", "value": {"stringValue": namespace}},
]},
"scopeSpans": [{
"scope": {"name": "unidesk-cli-platform-infra-observability"},
"spans": [{
"traceId": trace_id,
"spanId": span_id,
"name": "unidesk.observability.validate",
"kind": 1,
"startTimeUnixNano": str(now),
"endTimeUnixNano": str(now + 1000000),
"attributes": [
{"key": "unidesk.issue", "value": {"stringValue": "489"}},
{"key": "unidesk.spec", "value": {"stringValue": "${observability.metadata.spec}"}},
{"key": "unidesk.validation", "value": {"stringValue": "platform-infra-observability"}},
],
"status": {"code": 1},
}],
}],
}],
}
def generate_test_trace():
trace_id = ("%032x" % random.getrandbits(128))
span_id = ("%016x" % random.getrandbits(64))
port = free_port()
base_url = "http://127.0.0.1:%s" % port
stdout_path = f"{tmp}/kubectl-proxy.out"
stderr_path = f"{tmp}/kubectl-proxy.err"
with open(stdout_path, "w", encoding="utf-8") as stdout, open(stderr_path, "w", encoding="utf-8") as stderr:
proc = subprocess.Popen(
["kubectl", "proxy", "--address=127.0.0.1", "--port", str(port), "--accept-hosts=^127\\\\.0\\\\.0\\\\.1$"],
stdout=stdout,
stderr=stderr,
text=True,
)
try:
proxy_ready = wait_proxy(base_url)
if proxy_ready.get("ok") is not True:
return {
"ok": False,
"stage": "proxy-start",
"traceId": trace_id,
"proxy": compact_http(proxy_ready),
"proxyStderrTail": read(stderr_path, limit=2000),
}
collector_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.collector.serviceName}:otlp-http/proxy/v1/traces" % namespace
tempo_url = base_url + "/api/v1/namespaces/%s/services/http:${observability.traceBackend.serviceName}:http/proxy/api/traces/%s" % (namespace, trace_id)
ingest = http_request("POST", collector_url, validation_trace_payload(trace_id, span_id), timeout=15)
query = None
if ingest.get("ok") is True:
for _ in range(20):
query = http_request("GET", tempo_url, timeout=10)
if query.get("ok") is True:
break
time.sleep(1)
return {
"ok": ingest.get("ok") is True and isinstance(query, dict) and query.get("ok") is True,
"stage": "query" if ingest.get("ok") is True else "ingest",
"traceId": trace_id,
"spanId": span_id,
"collectorEndpoint": "${observability.collector.serviceName}:otlp-http",
"backendEndpoint": "${observability.traceBackend.serviceName}:http",
"ingest": compact_http(ingest),
"query": compact_http(query),
"valuesPrinted": False,
}
finally:
proc.terminate()
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait(timeout=3)
runtime_ready = rc("namespace") == 0 and rc("deployments") == 0 and rc("services") == 0 and all(item["ok"] for item in probe_results)
validation_trace = generate_test_trace() if (${generateValidationTrace ? "True" : "False"} and runtime_ready) else None
payload = {
"ok": runtime_ready and (validation_trace is None or validation_trace.get("ok") is True),
"target": "${target.id}",
"namespace": namespace,
"sections": {
"namespace": compact_section("namespace"),
"deployments": compact_section("deployments"),
"services": compact_section("services"),
"pods": compact_section("pods"),
"events": compact_section("events"),
},
"probes": probe_results,
"validationTrace": validation_trace,
}
print(json.dumps(payload, ensure_ascii=False, indent=2 if ${full ? "True" : "False"} else None))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}
function traceScript(observability: ObservabilityConfig, target: ObservabilityTarget, tracePath: string, options: TraceOptions): string {
const proxyPath = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy${tracePath}`;
const grepLiteral = options.grep === null ? "None" : JSON.stringify(options.grep);
return `
set -u
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
kubectl get --raw ${shQuote(proxyPath)} >"$tmp/trace.out" 2>"$tmp/trace.err"
rc=$?
python3 - "$rc" "$tmp/trace.out" "$tmp/trace.err" <<'PY'
import collections, json, sys
FULL = ${options.full ? "True" : "False"}
RAW = ${options.raw ? "True" : "False"}
GREP = ${grepLiteral}
LIMIT = ${options.limit}
IMPORTANT_ATTRS = [
"traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
"sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry",
"terminalStatus", "phase", "message", "http.route", "http.status_code",
"http.method", "eventType", "runnerId", "attemptId", "backendProfile",
"sourceCommit", "jobName", "podName", "logPath", "storageKind",
"storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
"returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
"totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
"maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
"waitingFor", "lastEventLabel", "idleMs", "idleSeconds", "idleThresholdMs",
"lastEventAgeMs", "toolName", "exitCode", "durationMs", "command",
"error", "error.message", "exception.type", "exception.message",
"valuesPrinted",
]
IMPORTANT_NAMES = {
"durable_admission", "billing_preflight", "agentrun_dispatch",
"projection_write", "trace_events_read", "turn_status_read",
"run_created", "runner_job_created", "runner_job_status",
"command_result", "projection_sync",
}
def text(path, limit=12000):
try:
return open(path, encoding="utf-8", errors="replace").read()[-limit:]
except FileNotFoundError:
return ""
def read_all(path):
try:
return open(path, encoding="utf-8", errors="replace").read()
except FileNotFoundError:
return ""
def attr_value(value):
if not isinstance(value, dict):
return value
inner = value.get("value", value)
if not isinstance(inner, dict):
return inner
for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
if key in inner:
return inner.get(key)
if "arrayValue" in inner:
return "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
return inner
def attrs_to_dict(attrs):
if not isinstance(attrs, list):
return {}
output = {}
for item in attrs:
if not isinstance(item, dict):
continue
key = item.get("key")
if isinstance(key, str):
output[key] = attr_value(item.get("value"))
return output
def nanos_to_ms(start, end):
try:
return round((int(end) - int(start)) / 1000000, 3)
except Exception:
return None
def span_status_code(status):
if not isinstance(status, dict):
return None
return status.get("code")
def selected_attrs(attrs):
output = {}
for key in IMPORTANT_ATTRS:
if key in attrs:
output[key] = attrs[key]
return output
def is_error_span(span, attrs):
status = span.get("status") if isinstance(span, dict) else {}
code = span_status_code(status)
name = str(span.get("name", "")).lower() if isinstance(span, dict) else ""
failure = str(attrs.get("failureKind", "")).strip()
terminal = str(attrs.get("terminalStatus", "")).strip()
return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked")
def compact_span(span, service, resource_attrs, scope_name):
attrs = attrs_to_dict(span.get("attributes"))
status = span.get("status") if isinstance(span.get("status"), dict) else {}
item = {
"name": span.get("name"),
"service": service,
"scope": scope_name,
"statusCode": span_status_code(status),
"statusMessage": status.get("message"),
"durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")),
"attributes": selected_attrs(attrs),
}
if "deployment.environment" in resource_attrs:
item["environment"] = resource_attrs.get("deployment.environment")
if "git.commit" in resource_attrs:
item["gitCommit"] = resource_attrs.get("git.commit")
return item
def batches_from_body(body):
if not isinstance(body, dict):
return []
for key in ("batches", "resourceSpans"):
value = body.get(key)
if isinstance(value, list):
return value
return []
def scope_spans_from_batch(batch):
if not isinstance(batch, dict):
return []
for key in ("scopeSpans", "instrumentationLibrarySpans"):
value = batch.get(key)
if isinstance(value, list):
return value
return []
def grep_matches(item):
if GREP is None:
return False
return GREP.lower() in json.dumps(item, ensure_ascii=False, sort_keys=True).lower()
def key_signature(item):
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
return json.dumps({
"name": item.get("name"),
"service": item.get("service"),
"statusCode": item.get("statusCode"),
"runId": attrs.get("runId"),
"commandId": attrs.get("commandId"),
"failureKind": attrs.get("failureKind"),
"terminalStatus": attrs.get("terminalStatus"),
"phase": attrs.get("phase"),
"http.route": attrs.get("http.route"),
}, ensure_ascii=False, sort_keys=True)
body = read_all(sys.argv[2])
body_bytes = len(body.encode("utf-8"))
try:
parsed = json.loads(body) if body else None
except Exception:
parsed = body[-12000:]
if not isinstance(parsed, dict):
payload = {
"ok": int(sys.argv[1]) == 0,
"path": "${tracePath}",
"proxyPath": "${proxyPath}",
"bodyBytes": body_bytes,
"parseOk": False,
"bodyTail": text(sys.argv[2], 12000),
"stderrTail": text(sys.argv[3], 4000),
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
spans = []
services = set()
business_trace_ids = set()
name_counts = collections.Counter()
error_spans = []
matched_spans = []
key_spans = []
seen_key_spans = set()
for batch in batches_from_body(parsed):
resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {}
resource_attrs = attrs_to_dict(resource.get("attributes"))
service = resource_attrs.get("service.name") or "<unknown-service>"
services.add(service)
for scope_span in scope_spans_from_batch(batch):
scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {}
scope_name = scope.get("name")
raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else []
for span in raw_spans:
if not isinstance(span, dict):
continue
item = compact_span(span, service, resource_attrs, scope_name)
attrs = item.get("attributes", {})
if isinstance(attrs.get("traceId"), str):
business_trace_ids.add(attrs.get("traceId"))
name = str(item.get("name") or "")
name_counts[name] += 1
spans.append(item)
error = is_error_span(span, attrs if isinstance(attrs, dict) else {})
matched = grep_matches(item)
if error:
error_spans.append(item)
if matched:
matched_spans.append(item)
if error or name in IMPORTANT_NAMES or name.startswith("runner_error.") or name.startswith("runner_terminal."):
signature = key_signature(item)
if signature not in seen_key_spans:
seen_key_spans.add(signature)
key_spans.append(item)
display_source = matched_spans if GREP is not None else key_spans
payload = {
"ok": int(sys.argv[1]) == 0,
"path": "${tracePath}",
"proxyPath": "${proxyPath}",
"bodyBytes": body_bytes,
"parseOk": True,
"traceFound": len(spans) > 0,
"spanCount": len(spans),
"serviceCount": len(services),
"services": sorted(services),
"businessTraceIds": sorted(business_trace_ids)[:20],
"errorSpanCount": len(error_spans),
"matchedSpanCount": len(matched_spans) if GREP is not None else None,
"grep": GREP,
"limit": LIMIT,
"spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(20)],
"errorSpans": error_spans[:LIMIT],
"spans": (spans if FULL and GREP is None else display_source)[:LIMIT],
"truncated": {
"errorSpans": len(error_spans) > LIMIT,
"spans": (len(spans) if FULL and GREP is None else len(display_source)) > LIMIT,
},
"next": {
"fullSummary": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? "<traceId>"} --full --limit 200",
"grepProviderStream": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? "<traceId>"} --grep provider-stream-disconnected --limit 40",
"raw": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? "<traceId>"} --raw",
},
"stderrTail": text(sys.argv[3], 4000),
}
if RAW:
payload["rawBody"] = parsed
print(json.dumps(payload, ensure_ascii=False, indent=2))
sys.exit(0 if payload["ok"] else 1)
PY
`;
}
function searchScript(observability: ObservabilityConfig, target: ObservabilityTarget, searchPath: string, options: SearchOptions): string {
const proxyPrefix = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy`;
const searchProxyPath = `${proxyPrefix}${searchPath}`;
const grepLiteral = options.grep === null ? "None" : JSON.stringify(options.grep);
const queryLiteral = options.query === null ? "None" : JSON.stringify(options.query);
return `
set -u
python3 - <<'PY'
import collections, json, subprocess, time
SEARCH_PROXY_PATH = ${JSON.stringify(searchProxyPath)}
TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)}
TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)}
FULL = ${options.full ? "True" : "False"}
RAW = ${options.raw ? "True" : "False"}
GREP = ${grepLiteral}
QUERY = ${queryLiteral}
LIMIT = ${options.limit}
CANDIDATE_LIMIT = ${options.candidateLimit}
DEADLINE = time.time() + 50
IMPORTANT_ATTRS = [
"traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
"sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry",
"terminalStatus", "phase", "message", "http.route", "http.status_code",
"http.method", "eventType", "runnerId", "attemptId", "backendProfile",
"sourceCommit", "jobName", "podName", "logPath", "storageKind",
"storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
"returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
"totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
"maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
"waitingFor", "lastEventLabel", "idleMs", "idleSeconds", "idleThresholdMs",
"lastEventAgeMs", "toolName", "exitCode", "durationMs", "command",
"error", "error.message", "exception.type", "exception.message",
"valuesPrinted",
]
def run_kubectl(proxy_path, timeout=10):
try:
proc = subprocess.run(
["kubectl", "get", "--raw", proxy_path],
capture_output=True,
text=True,
timeout=timeout,
)
return proc.returncode, proc.stdout, proc.stderr
except subprocess.TimeoutExpired as exc:
return 124, exc.stdout or "", "timeout while querying tempo"
except Exception as exc:
return 125, "", str(exc)
def parse_json(body):
try:
return json.loads(body) if body else None
except Exception:
return None
def attr_value(value):
if not isinstance(value, dict):
return value
inner = value.get("value", value)
if not isinstance(inner, dict):
return inner
for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
if key in inner:
return inner.get(key)
if "arrayValue" in inner:
return "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
return inner
def attrs_to_dict(attrs):
if not isinstance(attrs, list):
return {}
output = {}
for item in attrs:
if not isinstance(item, dict):
continue
key = item.get("key")
if isinstance(key, str):
output[key] = attr_value(item.get("value"))
return output
def selected_attrs(attrs):
output = {}
for key in IMPORTANT_ATTRS:
if key in attrs:
output[key] = attrs[key]
return output
def nanos_to_ms(start, end):
try:
return round((int(end) - int(start)) / 1000000, 3)
except Exception:
return None
def span_status_code(status):
if not isinstance(status, dict):
return None
return status.get("code")
def is_error_span(span, attrs):
status = span.get("status") if isinstance(span, dict) else {}
code = span_status_code(status)
name = str(span.get("name", "")).lower() if isinstance(span, dict) else ""
failure = str(attrs.get("failureKind", "")).strip()
terminal = str(attrs.get("terminalStatus", "")).strip()
return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked")
def batches_from_body(body):
if not isinstance(body, dict):
return []
for key in ("batches", "resourceSpans"):
value = body.get(key)
if isinstance(value, list):
return value
return []
def scope_spans_from_batch(batch):
if not isinstance(batch, dict):
return []
for key in ("scopeSpans", "instrumentationLibrarySpans"):
value = batch.get(key)
if isinstance(value, list):
return value
return []
def compact_span(span, service, resource_attrs, scope_name):
attrs = attrs_to_dict(span.get("attributes"))
status = span.get("status") if isinstance(span.get("status"), dict) else {}
item = {
"name": span.get("name"),
"service": service,
"scope": scope_name,
"statusCode": span_status_code(status),
"statusMessage": status.get("message"),
"durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")),
"attributes": selected_attrs(attrs),
}
if "deployment.environment" in resource_attrs:
item["environment"] = resource_attrs.get("deployment.environment")
if "git.commit" in resource_attrs:
item["gitCommit"] = resource_attrs.get("git.commit")
return item
def grep_matches_text(text):
return GREP is not None and GREP.lower() in text.lower()
def grep_matches_item(item):
return GREP is not None and grep_matches_text(json.dumps(item, ensure_ascii=False, sort_keys=True))
def extract_traces(search_body):
if not isinstance(search_body, dict):
return []
value = search_body.get("traces")
if isinstance(value, list):
return value
value = search_body.get("data")
if isinstance(value, list):
return value
return []
def trace_id_from_meta(meta):
if not isinstance(meta, dict):
return None
for key in ("traceID", "traceId", "trace_id"):
value = meta.get(key)
if isinstance(value, str) and value:
return value
return None
def compact_meta(meta):
if not isinstance(meta, dict):
return {}
output = {}
for key in ("traceID", "traceId", "rootServiceName", "rootTraceName", "startTimeUnixNano", "durationMs", "spanSet", "spanSets"):
if key in meta:
output[key] = meta[key]
return output
def trace_summary(trace_id, meta, body, rc, stderr):
parsed = parse_json(body)
raw_matched = grep_matches_text(body)
if not isinstance(parsed, dict):
return {
"traceId": trace_id,
"traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id,
"meta": compact_meta(meta),
"queryOk": rc == 0,
"parseOk": False,
"rawMatched": raw_matched,
"bodyBytes": len(body.encode("utf-8")),
"stderrTail": (stderr or "")[-1000:],
}
services = set()
business_trace_ids = set()
name_counts = collections.Counter()
spans = []
error_spans = []
matched_spans = []
for batch in batches_from_body(parsed):
resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {}
resource_attrs = attrs_to_dict(resource.get("attributes"))
service = resource_attrs.get("service.name") or "<unknown-service>"
services.add(service)
for scope_span in scope_spans_from_batch(batch):
scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {}
scope_name = scope.get("name")
raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else []
for span in raw_spans:
if not isinstance(span, dict):
continue
item = compact_span(span, service, resource_attrs, scope_name)
attrs = item.get("attributes", {})
if isinstance(attrs, dict) and isinstance(attrs.get("traceId"), str):
business_trace_ids.add(attrs.get("traceId"))
name = str(item.get("name") or "")
name_counts[name] += 1
spans.append(item)
if is_error_span(span, attrs if isinstance(attrs, dict) else {}):
error_spans.append(item)
if grep_matches_item(item):
matched_spans.append(item)
return {
"traceId": trace_id,
"traceCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % trace_id,
"grepCommand": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep %s --limit 80" % (trace_id, json.dumps(GREP) if GREP is not None else "<text>"),
"meta": compact_meta(meta),
"queryOk": rc == 0,
"parseOk": True,
"rawMatched": raw_matched,
"bodyBytes": len(body.encode("utf-8")),
"spanCount": len(spans),
"serviceCount": len(services),
"services": sorted(services),
"businessTraceIds": sorted(business_trace_ids)[:20],
"errorSpanCount": len(error_spans),
"matchedSpanCount": len(matched_spans) if GREP is not None else None,
"spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(10)],
"errorSpans": error_spans[:LIMIT] if FULL else error_spans[: min(3, LIMIT)],
"matchedSpans": matched_spans[:LIMIT],
"stderrTail": (stderr or "")[-1000:],
}
search_rc, search_body, search_err = run_kubectl(SEARCH_PROXY_PATH, timeout=15)
search_parsed = parse_json(search_body)
search_parse_ok = isinstance(search_parsed, dict)
trace_metas = extract_traces(search_parsed)
candidate_trace_ids = []
seen = set()
for meta in trace_metas:
trace_id = trace_id_from_meta(meta)
if trace_id is None or trace_id in seen:
continue
seen.add(trace_id)
candidate_trace_ids.append((trace_id, meta))
if len(candidate_trace_ids) >= CANDIDATE_LIMIT:
break
matched = []
scanned = []
scan_stopped = None
for trace_id, meta in candidate_trace_ids:
if time.time() > DEADLINE:
scan_stopped = "deadline"
break
trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", trace_id)
trace_rc, trace_body, trace_err = run_kubectl(TRACE_PROXY_PREFIX + trace_path, timeout=8)
summary = trace_summary(trace_id, meta, trace_body, trace_rc, trace_err)
scanned.append(summary)
if GREP is None or summary.get("rawMatched") is True or (summary.get("matchedSpanCount") or 0) > 0:
matched.append(summary)
if len(matched) >= LIMIT and GREP is not None:
break
payload = {
"ok": search_rc == 0 and search_parse_ok,
"searchPath": "${searchPath}",
"searchProxyPath": SEARCH_PROXY_PATH,
"grep": GREP,
"tempoQuery": QUERY,
"limit": LIMIT,
"candidateLimit": CANDIDATE_LIMIT,
"searchParseOk": search_parse_ok,
"candidateTraceCount": len(candidate_trace_ids),
"scannedTraceCount": len(scanned),
"matchedTraceCount": len(matched),
"scanStopped": scan_stopped,
"traces": matched[:LIMIT] if GREP is not None else scanned[:LIMIT],
"truncated": {
"candidateTraces": len(trace_metas) > len(candidate_trace_ids),
"matchedTraces": len(matched) > LIMIT,
},
"next": {
"expandWindow": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --grep <text> --lookback-minutes 1440 --candidate-limit 200 --limit 40",
"traceDetail": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id <traceId> --grep <text> --limit 80",
"raw": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --grep <text> --raw",
},
"searchStderrTail": (search_err or "")[-2000:],
}
if RAW:
payload["rawSearchBody"] = search_parsed if search_parse_ok else search_body[-12000:]
print(json.dumps(payload, ensure_ascii=False, indent=2))
raise SystemExit(0 if payload["ok"] else 1)
PY
`;
}
function diagnoseCodeAgentScript(observability: ObservabilityConfig, target: ObservabilityTarget, searchPath: string | null, options: DiagnoseCodeAgentOptions): string {
const proxyPrefix = `/api/v1/namespaces/${target.namespace}/services/http:${observability.traceBackend.serviceName}:http/proxy`;
const searchProxyPath = searchPath === null ? null : `${proxyPrefix}${searchPath}`;
const traceIdLiteral = options.traceId === null ? "None" : JSON.stringify(options.traceId);
const businessTraceIdLiteral = options.businessTraceId === null ? "None" : JSON.stringify(options.businessTraceId);
const searchPathLiteral = searchPath === null ? "None" : JSON.stringify(searchPath);
const searchProxyPathLiteral = searchProxyPath === null ? "None" : JSON.stringify(searchProxyPath);
const businessTraceIdForNext = options.businessTraceId ?? "<trc_...>";
return `
set -u
python3 - <<'PY'
import collections, json, subprocess, time
BUSINESS_TRACE_ID = ${businessTraceIdLiteral}
TRACE_ID = ${traceIdLiteral}
SEARCH_PATH = ${searchPathLiteral}
SEARCH_PROXY_PATH = ${searchProxyPathLiteral}
TRACE_PROXY_PREFIX = ${JSON.stringify(proxyPrefix)}
TRACE_PATH_TEMPLATE = ${JSON.stringify(observability.probes.traceQueryPathTemplate)}
FULL = ${options.full ? "True" : "False"}
RAW = ${options.raw ? "True" : "False"}
LIMIT = ${options.limit}
DEADLINE = time.time() + 50
IMPORTANT_ATTRS = [
"traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
"sessionId", "turnId", "threadId", "runnerJobId", "failureKind", "willRetry",
"terminalStatus", "phase", "message", "http.route", "http.status_code",
"http.method", "eventType", "runnerId", "attemptId", "backendProfile",
"sourceCommit", "jobName", "podName", "logPath", "storageKind",
"storagePvcName", "storagePvcPhase", "resumeMode", "rolloutId",
"returnedEvents", "sinceSeq", "afterSeq", "limit", "fromSeq", "toSeq",
"totalEvents", "hasMore", "fullTraceLoaded", "rawEventCount",
"maxSeq", "traceLastSeq", "endSeq", "commandFiltered",
"seq", "eventSeq", "sourceSeq", "sourceLatestSeq", "latestSeq",
"lastSeq", "sourceEventCount", "projectedSeq", "lastProjectedSeq",
"waitingFor", "lastEventLabel", "idleMs", "idleSeconds", "idleThresholdMs",
"lastEventAgeMs", "toolName", "exitCode", "durationMs", "command",
"status", "turnStatus", "error", "error.message", "exception.type",
"exception.message", "valuesPrinted",
]
IDENTITY_KEYS = [
"runId", "commandId", "sessionId", "turnId", "threadId",
"runnerJobId", "runnerId", "attemptId", "backendProfile",
"sourceCommit", "jobName", "podName",
]
def run_kubectl(proxy_path, timeout=15):
try:
proc = subprocess.run(
["kubectl", "get", "--raw", proxy_path],
capture_output=True,
text=True,
timeout=timeout,
)
return proc.returncode, proc.stdout, proc.stderr
except subprocess.TimeoutExpired as exc:
return 124, exc.stdout or "", "timeout while querying tempo"
except Exception as exc:
return 125, "", str(exc)
def parse_json(body):
try:
return json.loads(body) if body else None
except Exception:
return None
def attr_value(value):
if not isinstance(value, dict):
return value
inner = value.get("value", value)
if not isinstance(inner, dict):
return inner
for key in ("stringValue", "intValue", "doubleValue", "boolValue"):
if key in inner:
return inner.get(key)
if "arrayValue" in inner:
return "<array>"
if "kvlistValue" in inner:
return "<kvlist>"
if "bytesValue" in inner:
return "<bytes>"
return inner
def attrs_to_dict(attrs):
if not isinstance(attrs, list):
return {}
output = {}
for item in attrs:
if not isinstance(item, dict):
continue
key = item.get("key")
if isinstance(key, str):
output[key] = attr_value(item.get("value"))
return output
def selected_attrs(attrs):
output = {}
for key in IMPORTANT_ATTRS:
if key in attrs:
output[key] = attrs[key]
return output
def to_int(value):
if value is None:
return None
if isinstance(value, bool):
return 1 if value else 0
if isinstance(value, int):
return value
if isinstance(value, float) and value.is_integer():
return int(value)
if isinstance(value, str):
try:
return int(value.strip())
except Exception:
return None
return None
def to_bool(value):
if isinstance(value, bool):
return value
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in ("true", "1", "yes"):
return True
if lowered in ("false", "0", "no"):
return False
return None
def nanos_to_ms(start, end):
try:
return round((int(end) - int(start)) / 1000000, 3)
except Exception:
return None
def span_status_code(status):
if not isinstance(status, dict):
return None
return status.get("code")
def is_error_span(span, attrs):
status = span.get("status") if isinstance(span, dict) else {}
code = span_status_code(status)
name = str(span.get("name", "")).lower() if isinstance(span, dict) else ""
failure = str(attrs.get("failureKind", "")).strip()
terminal = str(attrs.get("terminalStatus", "")).strip()
return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked")
def batches_from_body(body):
if not isinstance(body, dict):
return []
for key in ("batches", "resourceSpans"):
value = body.get(key)
if isinstance(value, list):
return value
return []
def scope_spans_from_batch(batch):
if not isinstance(batch, dict):
return []
for key in ("scopeSpans", "instrumentationLibrarySpans"):
value = batch.get(key)
if isinstance(value, list):
return value
return []
def extract_traces(search_body):
if not isinstance(search_body, dict):
return []
value = search_body.get("traces")
if isinstance(value, list):
return value
value = search_body.get("data")
if isinstance(value, list):
return value
return []
def trace_id_from_meta(meta):
if not isinstance(meta, dict):
return None
for key in ("traceID", "traceId", "trace_id"):
value = meta.get(key)
if isinstance(value, str) and value:
return value
return None
def compact_meta(meta):
if not isinstance(meta, dict):
return {}
output = {}
for key in ("traceID", "traceId", "rootServiceName", "rootTraceName", "startTimeUnixNano", "durationMs"):
if key in meta:
output[key] = meta[key]
return output
def compact_span(span, service, resource_attrs, scope_name, index):
raw_attrs = attrs_to_dict(span.get("attributes"))
attrs = selected_attrs(raw_attrs)
status = span.get("status") if isinstance(span.get("status"), dict) else {}
item = {
"name": span.get("name"),
"service": service,
"scope": scope_name,
"statusCode": span_status_code(status),
"statusMessage": status.get("message"),
"durationMs": nanos_to_ms(span.get("startTimeUnixNano"), span.get("endTimeUnixNano")),
"attributes": attrs,
"_index": index,
"_start": to_int(span.get("startTimeUnixNano")) or index,
}
if "deployment.environment" in resource_attrs:
item["environment"] = resource_attrs.get("deployment.environment")
if "git.commit" in resource_attrs:
item["gitCommit"] = resource_attrs.get("git.commit")
return item, raw_attrs
def public_span(item):
if not isinstance(item, dict):
return item
return {key: value for key, value in item.items() if not key.startswith("_")}
def tiny_span(item):
if not isinstance(item, dict):
return item
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
keep_attrs = {}
for key in (
"http.route", "http.status_code", "http.method", "status",
"terminalStatus", "eventType", "returnedEvents", "sinceSeq",
"fromSeq", "toSeq", "totalEvents", "hasMore", "fullTraceLoaded",
"afterSeq", "rawEventCount", "failureKind",
):
if key in attrs:
keep_attrs[key] = attrs[key]
return {
"name": item.get("name"),
"service": item.get("service"),
"attributes": keep_attrs,
}
def tiny_terminal(terminal):
if not isinstance(terminal, dict):
return terminal
return {
"status": terminal.get("status"),
"eventType": terminal.get("eventType"),
"span": tiny_span(terminal.get("span")),
}
def identity_from_spans(spans):
identity = {}
for key in IDENTITY_KEYS:
preferred = None
fallback = None
for item in spans:
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
value = attrs.get(key)
if value in (None, ""):
continue
if fallback is None:
fallback = value
service = str(item.get("service") or "")
if service.startswith("agentrun"):
preferred = value
break
if preferred is not None or fallback is not None:
identity[key] = preferred if preferred is not None else fallback
return identity
def flatten_trace(trace_body):
services = set()
business_trace_ids = set()
name_counts = collections.Counter()
spans = []
error_spans = []
raw_index = 0
for batch in batches_from_body(trace_body):
resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {}
resource_attrs = attrs_to_dict(resource.get("attributes"))
service = resource_attrs.get("service.name") or "<unknown-service>"
services.add(service)
for scope_span in scope_spans_from_batch(batch):
scope = scope_span.get("scope") if isinstance(scope_span.get("scope"), dict) else {}
scope_name = scope.get("name")
raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else []
for span in raw_spans:
if not isinstance(span, dict):
continue
item, raw_attrs = compact_span(span, service, resource_attrs, scope_name, raw_index)
raw_index += 1
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
if isinstance(attrs.get("traceId"), str):
business_trace_ids.add(attrs.get("traceId"))
name = str(item.get("name") or "")
name_counts[name] += 1
spans.append(item)
if is_error_span(span, raw_attrs):
error_spans.append(item)
spans.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0)))
error_spans.sort(key=lambda item: (item.get("_start", 0), item.get("_index", 0)))
return {
"spans": spans,
"services": sorted(services),
"businessTraceIds": sorted(business_trace_ids),
"spanNameCounts": [{"name": name, "count": count} for name, count in name_counts.most_common(20)],
"errorSpans": error_spans,
}
def raw_trace_shape(trace_body):
batches = batches_from_body(trace_body)
scope_count = 0
span_count = 0
first_resource_attrs = {}
first_scope_keys = []
first_span_keys = []
first_span_attr_keys = []
for batch_index, batch in enumerate(batches):
resource = batch.get("resource") if isinstance(batch.get("resource"), dict) else {}
resource_attrs = attrs_to_dict(resource.get("attributes"))
if batch_index == 0:
for key in ("service.name", "deployment.environment", "git.commit"):
if key in resource_attrs:
first_resource_attrs[key] = resource_attrs[key]
for scope_span in scope_spans_from_batch(batch):
scope_count += 1
if not first_scope_keys and isinstance(scope_span, dict):
first_scope_keys = sorted(scope_span.keys())
raw_spans = scope_span.get("spans") if isinstance(scope_span.get("spans"), list) else []
span_count += len(raw_spans)
if not first_span_keys and raw_spans and isinstance(raw_spans[0], dict):
first_span_keys = sorted(raw_spans[0].keys())
first_span_attr_keys = sorted(attrs_to_dict(raw_spans[0].get("attributes")).keys())[:50]
return {
"mode": "bounded-raw-shape",
"note": "Full Tempo body is intentionally not printed to avoid secrets/raw transcript and SSH 60s output timeouts.",
"topLevelKeys": sorted(trace_body.keys()) if isinstance(trace_body, dict) else [],
"resourceSpanCount": len(batches),
"scopeSpanCount": scope_count,
"spanCount": span_count,
"firstResourceAttributes": first_resource_attrs,
"firstScopeKeys": first_scope_keys,
"firstSpanKeys": first_span_keys,
"firstSpanAttributeKeys": first_span_attr_keys,
}
def http_status_summary(spans):
grouped = collections.Counter()
problem_spans = []
for item in spans:
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
code = to_int(attrs.get("http.status_code"))
if code is None:
continue
method = attrs.get("http.method") or ""
route = attrs.get("http.route") or item.get("name") or "<unknown-route>"
grouped[(str(method), str(route), code)] += 1
if code in (401, 403) or code >= 500:
problem_spans.append(item)
status_counts = [
{"method": method or None, "route": route, "status": status, "count": count}
for (method, route, status), count in grouped.most_common()
]
problem_counts = [
{"method": method or None, "route": route, "status": status, "count": count}
for (method, route, status), count in grouped.most_common()
if status in (401, 403) or status >= 500
]
return {
"statusCounts": status_counts[:20],
"problemCounts": problem_counts[:20],
"problemSpans": problem_spans,
"actorForbidden": any(to_int((item.get("attributes") or {}).get("http.status_code")) == 403 and item.get("service") == "hwlab-cloud-api" for item in problem_spans),
}
def agentrun_summary(spans):
terminal_spans = []
latest_seq = None
seq_keys = ("maxSeq", "traceLastSeq", "endSeq", "seq", "eventSeq", "sourceSeq", "sourceLatestSeq", "latestSeq", "lastSeq")
for item in spans:
service = str(item.get("service") or "")
name = str(item.get("name") or "")
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
if service.startswith("agentrun"):
for key in seq_keys:
value = to_int(attrs.get(key))
if value is not None:
latest_seq = value if latest_seq is None else max(latest_seq, value)
terminal_status = attrs.get("terminalStatus")
if terminal_status or name.startswith("runner_terminal."):
derived = terminal_status
if not derived and name.startswith("runner_terminal."):
derived = name.split(".", 1)[1]
terminal_spans.append({
"status": derived,
"eventType": attrs.get("eventType"),
"span": public_span(item),
})
terminal_status = terminal_spans[-1]["status"] if terminal_spans else None
terminal_event_type = None
for terminal in reversed(terminal_spans):
if terminal.get("eventType"):
terminal_event_type = terminal.get("eventType")
break
if terminal_status in ("completed", "succeeded", "success"):
runner_classification = "completed"
elif terminal_status in ("failed", "error", "timeout", "cancelled"):
runner_classification = "failed"
elif terminal_status in ("blocked",):
runner_classification = "blocked"
elif any(str(item.get("service") or "") == "agentrun-runner" for item in spans):
runner_classification = "running"
else:
runner_classification = "unknown"
return {
"terminalStatus": terminal_status,
"latestSeq": latest_seq,
"terminalEventType": terminal_event_type,
"terminalSpans": terminal_spans,
"runnerProviderClassification": runner_classification,
}
def hwlab_read_model_summary(spans):
trace_read_spans = []
turn_read_spans = []
projection_spans = []
source_event_count = None
requested_since_seq = None
latest_projected_seq = None
turn_statuses = []
turn_status_counts = collections.Counter()
for item in spans:
if item.get("service") != "hwlab-cloud-api":
continue
name = str(item.get("name") or "")
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
if name == "trace_events_read":
trace_read_spans.append(public_span(item))
for key in ("totalEvents", "sourceEventCount"):
value = to_int(attrs.get(key))
if value is not None:
source_event_count = value if source_event_count is None else max(source_event_count, value)
for key in ("sinceSeq", "fromSeq"):
value = to_int(attrs.get(key))
if value is not None:
requested_since_seq = value if requested_since_seq is None else max(requested_since_seq, value)
if name == "turn_status_read":
turn_read_spans.append(public_span(item))
if attrs.get("turnStatus") is not None or attrs.get("status") is not None:
status_value = attrs.get("turnStatus") if attrs.get("turnStatus") is not None else attrs.get("status")
turn_statuses.append(status_value)
turn_status_counts[str(status_value)] += 1
if name == "projection_sync":
projection_spans.append(public_span(item))
after_seq = to_int(attrs.get("afterSeq"))
raw_count = to_int(attrs.get("rawEventCount"))
projected = None
for key in ("maxSeq", "traceLastSeq", "endSeq", "projectedSeq", "lastProjectedSeq"):
value = to_int(attrs.get(key))
if value is not None:
projected = value if projected is None else max(projected, value)
if after_seq is not None:
projected = after_seq + (raw_count or 0)
if projected is not None:
latest_projected_seq = projected if latest_projected_seq is None else max(latest_projected_seq, projected)
source_event_count = projected if source_event_count is None else max(source_event_count, projected)
has_more_values = []
full_loaded_values = []
for item in trace_read_spans:
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
if "hasMore" in attrs:
has_more_values.append(to_bool(attrs.get("hasMore")))
if "fullTraceLoaded" in attrs:
full_loaded_values.append(to_bool(attrs.get("fullTraceLoaded")))
return {
"sourceEventCount": source_event_count,
"requestedSinceSeq": requested_since_seq,
"latestProjectedSeq": latest_projected_seq,
"traceStatus": {
"readCount": len(trace_read_spans),
"anyHasMore": any(value is True for value in has_more_values),
"anyFullTraceLoaded": any(value is True for value in full_loaded_values),
},
"turnStatus": turn_statuses[-1] if turn_statuses else None,
"turnStatusCounts": [{"status": status, "count": count} for status, count in turn_status_counts.most_common()],
"traceEventReadSpans": trace_read_spans,
"turnStatusReadSpans": turn_read_spans,
"projectionSpans": projection_spans,
}
def projection_lag_summary(agentrun, read_model):
source_event_count = read_model.get("sourceEventCount")
requested_since_seq = read_model.get("requestedSinceSeq")
latest_projected_seq = read_model.get("latestProjectedSeq")
agent_latest_seq = agentrun.get("latestSeq")
reasons = []
status = "none"
if requested_since_seq is not None and source_event_count is not None and requested_since_seq > source_event_count:
status = "confirmed"
reasons.append("requested sinceSeq %s is greater than HWLAB sourceEventCount %s" % (requested_since_seq, source_event_count))
if latest_projected_seq is not None and requested_since_seq is not None and latest_projected_seq < requested_since_seq:
status = "confirmed"
reasons.append("latestProjectedSeq %s is behind requested sinceSeq %s" % (latest_projected_seq, requested_since_seq))
if agent_latest_seq is not None and source_event_count is not None and agent_latest_seq > source_event_count:
status = "confirmed"
reasons.append("AgentRun latestSeq %s is greater than HWLAB sourceEventCount %s" % (agent_latest_seq, source_event_count))
if status == "none" and agentrun.get("terminalStatus") == "completed" and source_event_count is not None and read_model.get("traceStatus", {}).get("readCount", 0) > 0:
status = "suspected"
reasons.append("AgentRun completed while HWLAB read model still needs endpoint/status confirmation")
return {
"status": status,
"reasons": reasons,
}
def root_cause_candidates(http_summary, agentrun, read_model, lag_summary, error_spans):
candidates = []
if http_summary.get("actorForbidden"):
candidates.append({
"code": "actor_forbidden",
"label": "actor forbidden",
"confidence": 0.98,
"summary": "actor forbidden: HWLAB Code Agent read endpoint returned HTTP 403, likely actor/owner mismatch for this trace or turn.",
"evidence": http_summary.get("problemCounts", [])[:5],
})
if lag_summary.get("status") == "confirmed":
candidates.append({
"code": "hwlab_projection_stale",
"label": "HWLAB projection stale",
"confidence": 0.94,
"summary": "HWLAB projection stale: read model sequence is behind the caller/requested event window.",
"evidence": {
"sourceEventCount": read_model.get("sourceEventCount"),
"requestedSinceSeq": read_model.get("requestedSinceSeq"),
"latestProjectedSeq": read_model.get("latestProjectedSeq"),
"reasons": lag_summary.get("reasons", []),
},
})
elif lag_summary.get("status") == "suspected":
candidates.append({
"code": "hwlab_projection_lag_suspected",
"label": "HWLAB projection stale",
"confidence": 0.64,
"summary": "HWLAB projection stale is suspected because AgentRun is terminal but HWLAB read-model evidence is incomplete.",
"evidence": lag_summary.get("reasons", []),
})
if agentrun.get("terminalStatus") == "completed":
candidates.append({
"code": "agentrun_completed_not_provider_blocked",
"label": "AgentRun completed",
"confidence": 0.88,
"summary": "AgentRun completed: runner/provider reached terminal completed, so this trace is not explained by an active runner hang.",
"evidence": {
"terminalStatus": agentrun.get("terminalStatus"),
"terminalEventType": agentrun.get("terminalEventType"),
"runnerProviderClassification": agentrun.get("runnerProviderClassification"),
},
})
if error_spans and agentrun.get("terminalStatus") == "completed":
candidates.append({
"code": "tool_call_failures_recovered",
"label": "tool call failures recovered",
"confidence": 0.42,
"summary": "tool call failure spans exist, but AgentRun still completed; treat them as secondary unless correlated with user-visible failure.",
"evidence": {
"errorSpanCount": len(error_spans),
"sampleNames": sorted(set(str(item.get("name") or "") for item in error_spans))[:5],
},
})
candidates.sort(key=lambda item: item.get("confidence", 0), reverse=True)
return candidates
def score_trace_candidate(trace_id, meta, index):
trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", trace_id)
trace_proxy_path = TRACE_PROXY_PREFIX + trace_path
trace_rc, trace_body, trace_err = run_kubectl(trace_proxy_path, timeout=5)
trace_parsed = parse_json(trace_body)
score = 0
reasons = []
services = []
span_count = 0
error_span_count = 0
business_trace_ids = []
root_name = None
fetch_ok = trace_rc == 0 and isinstance(trace_parsed, dict)
if not fetch_ok:
return {
"traceId": trace_id,
"index": index,
"score": -100,
"reasons": ["trace_fetch_failed"],
"services": services,
"spanCount": span_count,
"errorSpanCount": error_span_count,
"businessTraceIds": business_trace_ids,
"rootName": root_name,
"fetchOk": False,
"selectedMeta": compact_meta(meta),
"stderrTail": (trace_err or "")[-1000:],
}
flat = flatten_trace(trace_parsed)
spans = flat.get("spans", [])
services = flat.get("services", [])
business_trace_ids = flat.get("businessTraceIds", [])
error_span_count = len(flat.get("errorSpans", []))
span_count = len(spans)
if spans:
root_name = spans[0].get("name")
service_set = set(str(service) for service in services)
span_names = [str(item.get("name") or "") for item in spans]
attr_fragments = []
for item in spans:
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
for key in ("runId", "commandId", "runnerJobId", "terminalStatus", "eventType", "failureKind", "traceId"):
value = attrs.get(key)
if value not in (None, ""):
attr_fragments.append("%s=%s" % (key, value))
haystack = " ".join(span_names + attr_fragments).lower()
if BUSINESS_TRACE_ID and BUSINESS_TRACE_ID in business_trace_ids:
score += 20
reasons.append("business_trace_id_matched")
if "hwlab-cloud-api" in service_set:
score += 20
reasons.append("hwlab_cloud_api_reached")
if "agentrun-manager" in service_set:
score += 80
reasons.append("agentrun_manager_reached")
if "agentrun-runner" in service_set:
score += 120
reasons.append("agentrun_runner_reached")
if "codex_stdio" in haystack:
score += 80
reasons.append("codex_stdio_spans_present")
if "runner_terminal" in haystack or "terminalstatus=" in haystack:
score += 60
reasons.append("runner_terminal_present")
if "runid=" in haystack:
score += 30
reasons.append("run_id_present")
if "commandid=" in haystack:
score += 30
reasons.append("command_id_present")
if error_span_count > 0:
score += min(30, error_span_count * 5)
reasons.append("error_spans_present")
if span_count > 1:
score += min(40, span_count)
if service_set == {"hwlab-cloud-api"} and span_count <= 2 and any(name.startswith("GET /v1/workbench/events") for name in span_names):
score -= 120
reasons.append("penalized_workbench_events_sse_only")
if any(name.startswith("GET /v1/workbench/events") for name in span_names):
score -= 20
reasons.append("workbench_events_candidate")
return {
"traceId": trace_id,
"index": index,
"score": score,
"reasons": reasons,
"services": services,
"spanCount": span_count,
"errorSpanCount": error_span_count,
"businessTraceIds": business_trace_ids,
"rootName": root_name,
"fetchOk": True,
"selectedMeta": compact_meta(meta),
}
def resolve_trace():
if TRACE_ID is not None:
return TRACE_ID, {
"mode": "trace-id",
"businessTraceId": BUSINESS_TRACE_ID,
"otelTraceId": TRACE_ID,
"searchPath": None,
"searchOk": None,
"searchParseOk": None,
"candidateTraceCount": None,
"selectedMeta": None,
}, None
if SEARCH_PROXY_PATH is None:
return None, None, "missing trace id and search path"
search_rc, search_body, search_err = run_kubectl(SEARCH_PROXY_PATH, timeout=15)
search_parsed = parse_json(search_body)
metas = extract_traces(search_parsed)
raw_candidates = []
for index, meta in enumerate(metas):
trace_id = trace_id_from_meta(meta)
if trace_id:
raw_candidates.append((index, trace_id, meta))
candidate_probe_limit = 60
probe_deadline_seconds = 45
probe_deadline = time.time() + probe_deadline_seconds
candidate_probe_stopped = None
candidate_scores = []
for index, trace_id, meta in raw_candidates[:candidate_probe_limit]:
if time.time() > probe_deadline:
candidate_probe_stopped = "deadline"
break
scored = score_trace_candidate(trace_id, meta, index)
candidate_scores.append(scored)
if scored.get("score", 0) >= 260 and len(candidate_scores) >= 12:
candidate_probe_stopped = "strong-candidate-found"
break
candidate_scores.sort(key=lambda item: (item.get("score", -9999), item.get("spanCount", 0), -item.get("index", 0)), reverse=True)
selected_score = candidate_scores[0] if candidate_scores else None
selected_trace_id = selected_score.get("traceId") if isinstance(selected_score, dict) else None
selected = selected_score.get("selectedMeta") if isinstance(selected_score, dict) else None
selection_warning = None
if selected_score is not None:
selected_services = set(str(service) for service in selected_score.get("services", []))
if "agentrun-runner" not in selected_services:
selection_warning = "no candidate contained agentrun-runner; selected best available trace"
if selected_score.get("score", 0) < 0:
selection_warning = "candidate fetch/scoring did not find a strong trace; selected best available trace"
mapping = {
"mode": "business-trace-id",
"businessTraceId": BUSINESS_TRACE_ID,
"otelTraceId": selected_trace_id,
"searchPath": SEARCH_PATH,
"searchProxyPath": SEARCH_PROXY_PATH,
"searchOk": search_rc == 0,
"searchParseOk": isinstance(search_parsed, dict),
"candidateTraceCount": len(metas),
"candidateProbeLimit": candidate_probe_limit,
"candidateProbeDeadlineSeconds": probe_deadline_seconds,
"candidateProbeStopped": candidate_probe_stopped,
"probedCandidateCount": len(candidate_scores),
"selectedMeta": selected,
"selection": {
"mode": "scored-candidates",
"score": selected_score.get("score") if isinstance(selected_score, dict) else None,
"reasons": selected_score.get("reasons") if isinstance(selected_score, dict) else [],
"rootName": selected_score.get("rootName") if isinstance(selected_score, dict) else None,
"services": selected_score.get("services") if isinstance(selected_score, dict) else [],
"spanCount": selected_score.get("spanCount") if isinstance(selected_score, dict) else None,
},
"selectionWarning": selection_warning,
"candidateScores": candidate_scores[:12],
"searchStderrTail": (search_err or "")[-2000:],
}
if RAW:
mapping["rawSearchBody"] = search_parsed if isinstance(search_parsed, dict) else search_body[-12000:]
if selected_trace_id is None:
return None, mapping, "no OTel trace found for business trace"
return selected_trace_id, mapping, None
resolved_trace_id, mapping, resolve_error = resolve_trace()
if resolved_trace_id is None:
payload = {
"ok": False,
"mapping": mapping,
"error": resolve_error,
"summary": {
"rootCause": "otel trace not found",
"facts": [],
},
"next": {
"expandWindow": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --business-trace-id ${businessTraceIdForNext} --lookback-minutes 10080 --candidate-limit 100 --full",
"search": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --query '{ .traceId = \\"${businessTraceIdForNext}\\" }' --lookback-minutes 10080 --candidate-limit 100 --limit 10",
},
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
raise SystemExit(1)
trace_path = TRACE_PATH_TEMPLATE.replace("{{traceId}}", resolved_trace_id)
trace_proxy_path = TRACE_PROXY_PREFIX + trace_path
trace_rc, trace_body, trace_err = run_kubectl(trace_proxy_path, timeout=25)
trace_parsed = parse_json(trace_body)
if not isinstance(trace_parsed, dict):
payload = {
"ok": False,
"mapping": mapping,
"tracePath": trace_path,
"traceProxyPath": trace_proxy_path,
"traceQueryOk": trace_rc == 0,
"traceParseOk": False,
"bodyBytes": len(trace_body.encode("utf-8")),
"bodyTail": trace_body[-12000:],
"stderrTail": (trace_err or "")[-4000:],
}
print(json.dumps(payload, ensure_ascii=False, indent=2))
raise SystemExit(1)
flat = flatten_trace(trace_parsed)
spans = flat["spans"]
services = flat["services"]
business_trace_ids = flat["businessTraceIds"]
error_spans = flat["errorSpans"]
http_summary = http_status_summary(spans)
agentrun = agentrun_summary(spans)
read_model = hwlab_read_model_summary(spans)
lag = projection_lag_summary(agentrun, read_model)
identity = identity_from_spans(spans)
candidates = root_cause_candidates(http_summary, agentrun, read_model, lag, error_spans)
expected_services = ["hwlab-cloud-api", "agentrun-manager", "agentrun-runner"]
service_path = {
service: ("reached" if service in services else "missing")
for service in expected_services
}
service_path["complete"] = all(service in services for service in expected_services)
facts = []
if http_summary.get("actorForbidden"):
facts.append("actor forbidden")
if agentrun.get("terminalStatus") == "completed":
facts.append("AgentRun completed")
if lag.get("status") in ("confirmed", "suspected"):
facts.append("HWLAB projection stale")
if not facts:
facts.append("no dominant Code Agent root cause classified")
summary = {
"rootCause": "; ".join(facts),
"facts": facts,
"classification": {
"projectionLag": lag.get("status"),
"runnerProvider": agentrun.get("runnerProviderClassification"),
"actorForbidden": http_summary.get("actorForbidden"),
},
}
evidence = {
"httpProblemSpanCount": len(http_summary.get("problemSpans", [])),
"httpProblemSpanSamples": [tiny_span(item) for item in http_summary.get("problemSpans", [])[:3]],
"terminalSpanCount": len(agentrun.get("terminalSpans", [])),
"terminalSpanSamples": [tiny_terminal(item) for item in agentrun.get("terminalSpans", [])[:3]],
"traceEventReadSpans": [tiny_span(item) for item in read_model.get("traceEventReadSpans", [])[:3]],
"turnStatusReadSpanCount": len(read_model.get("turnStatusReadSpans", [])),
"turnStatusReadSpanSamples": [tiny_span(item) for item in read_model.get("turnStatusReadSpans", [])[:3]],
"projectionSpanCount": len(read_model.get("projectionSpans", [])),
"projectionSpanTail": [tiny_span(item) for item in read_model.get("projectionSpans", [])[-3:]],
"errorSpanCount": len(error_spans),
"errorSpanSampleNames": sorted(set(str(item.get("name") or "") for item in error_spans))[:5],
}
payload = {
"ok": trace_rc == 0 and len(spans) > 0,
"mapping": mapping,
"tracePath": trace_path,
"traceProxyPath": trace_proxy_path,
"bodyBytes": len(trace_body.encode("utf-8")),
"traceParseOk": True,
"spanCount": len(spans),
"services": services,
"servicePath": service_path,
"businessTraceIds": business_trace_ids[:20],
"identity": identity,
"agentrun": {
"terminalStatus": agentrun.get("terminalStatus"),
"latestSeq": agentrun.get("latestSeq"),
"terminalEventType": agentrun.get("terminalEventType"),
"runnerProviderClassification": agentrun.get("runnerProviderClassification"),
},
"hwlabReadModel": {
"sourceEventCount": read_model.get("sourceEventCount"),
"requestedSinceSeq": read_model.get("requestedSinceSeq"),
"latestProjectedSeq": read_model.get("latestProjectedSeq"),
"traceStatus": read_model.get("traceStatus"),
"turnStatus": read_model.get("turnStatus"),
"turnStatusCounts": read_model.get("turnStatusCounts"),
},
"http": {
"statusCounts": http_summary.get("statusCounts", [])[:20],
"problemCounts": http_summary.get("problemCounts", [])[:20],
"actorForbidden": http_summary.get("actorForbidden"),
},
"projectionLag": lag,
"summary": summary,
"rootCauseCandidates": candidates,
"spanNameCounts": flat["spanNameCounts"][:12],
"evidence": evidence,
"next": {
"diagnoseFull": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --business-trace-id ${businessTraceIdForNext} --full",
"traceSummary": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --limit 80" % resolved_trace_id,
"traceReads": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep trace_events_read --limit 20 --full" % resolved_trace_id,
"turnStatus": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep turn_status_read --limit 20 --full" % resolved_trace_id,
"projection": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep projection_sync --limit 120 --full" % resolved_trace_id,
"runnerTerminal": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --grep runner_terminal --limit 20 --full" % resolved_trace_id,
"raw": "bun scripts/cli.ts platform-infra observability diagnose-code-agent --target ${target.id} --trace-id %s --raw" % resolved_trace_id,
},
"stderrTail": (trace_err or "")[-4000:],
}
if FULL:
payload["full"] = {
"traceEventReadSpans": read_model.get("traceEventReadSpans", []),
"turnStatusReadSpans": read_model.get("turnStatusReadSpans", []),
"projectionSpans": read_model.get("projectionSpans", [])[:LIMIT],
"terminalSpans": agentrun.get("terminalSpans", [])[:LIMIT],
"errorSpans": [public_span(item) for item in error_spans[:LIMIT]],
"httpProblemSpans": [public_span(item) for item in http_summary.get("problemSpans", [])[:LIMIT]],
"selectedSpans": [public_span(item) for item in spans if item.get("name") in ("trace_events_read", "turn_status_read", "projection_sync", "command_result")][:LIMIT],
}
if RAW:
payload["raw"] = raw_trace_shape(trace_parsed)
print(json.dumps(payload, ensure_ascii=False, indent=2 if FULL or RAW else None))
raise SystemExit(0 if payload["ok"] else 1)
PY
`;
}
function configSummary(observability: ObservabilityConfig, target: ObservabilityTarget): Record<string, unknown> {
return {
configPath: configLabel,
spec: observability.metadata.spec,
target: targetSummary(target),
images: {
collector: imageReference(observability.images.collector),
traceBackend: imageReference(observability.images.tempo),
},
collector: observability.collector,
traceBackend: observability.traceBackend,
sampling: observability.sampling,
};
}
function targetSummary(target: ObservabilityTarget): Record<string, unknown> {
return {
id: target.id,
route: target.route,
namespace: target.namespace,
role: target.role,
createNamespace: target.createNamespace,
};
}
function policyChecks(yaml: string, target: ObservabilityTarget): Array<Record<string, unknown>> {
return [
{ name: "yaml-source-of-truth", ok: true, detail: "All concrete images, routes, namespace, ports, retention and sampling values are read from config/platform-infra/observability.yaml." },
{ name: "clusterip-only", ok: !/^\s*type:\s*(NodePort|LoadBalancer)\s*$/mu.test(yaml), detail: "Collector and trace backend stay ClusterIP-only." },
{ name: "no-ingress", ok: !/^\s*kind:\s*Ingress\s*$/mu.test(yaml), detail: "No public ingress is rendered for the first tracing backend." },
{ name: "no-host-network", ok: !/^\s*hostNetwork:\s*true\s*$/mu.test(yaml), detail: "Pods must not use host network." },
{ name: "allow-all-network-policy", ok: yaml.includes("kind: NetworkPolicy") && yaml.includes("name: allow-all") && yaml.includes(`namespace: ${target.namespace}`), detail: `NetworkPolicy/allow-all is rendered in ${target.namespace}.` },
];
}
function statusSummary(payload: Record<string, unknown>): Record<string, unknown> {
const sections = asRecord(payload.sections, "status.sections");
const deployments = objectList(sectionJson(sections, "deployments"));
const services = objectList(sectionJson(sections, "services"));
const pods = objectList(sectionJson(sections, "pods"));
const events = objectList(sectionJson(sections, "events"));
const podEvents = latestPodEvents(events);
const probes = Array.isArray(payload.probes) ? payload.probes as Array<Record<string, unknown>> : [];
const readyDeployments = deployments.map((item) => deploymentSummary(item));
return {
ready: payload.ok === true,
namespace: payload.namespace,
deployments: readyDeployments,
services: services.map((item) => metadataName(item)),
pods: pods.map((item) => podSummary(item, podEvents)),
probes: probes.map((item) => ({
name: item.name,
ok: item.ok === true,
service: item.service,
path: item.path,
stderrTail: item.ok === true ? "" : item.stderrTail,
})),
};
}
function compactStatus(payload: Record<string, unknown> | null, full: boolean): Record<string, unknown> | null {
if (payload === null) return null;
if (full) return redactSensitiveUnknown(payload) as Record<string, unknown>;
return statusSummary(payload);
}
function sectionJson(sections: Record<string, unknown>, name: string): unknown {
const section = asRecord(sections[name], `sections.${name}`);
return section.json;
}
function objectList(value: unknown): Record<string, unknown>[] {
if (typeof value !== "object" || value === null) return [];
const items = (value as Record<string, unknown>).items;
if (!Array.isArray(items)) return [];
return items.filter((item): item is Record<string, unknown> => typeof item === "object" && item !== null && !Array.isArray(item));
}
function deploymentSummary(item: Record<string, unknown>): Record<string, unknown> {
const spec = item.spec as Record<string, unknown> | undefined;
const status = item.status as Record<string, unknown> | undefined;
const replicas = typeof spec?.replicas === "number" ? spec.replicas : null;
const available = typeof status?.availableReplicas === "number" ? status.availableReplicas : 0;
return {
name: metadataName(item),
replicas,
availableReplicas: available,
ready: replicas !== null && available >= replicas,
};
}
function podSummary(item: Record<string, unknown>, podEvents: Map<string, Record<string, unknown>>): Record<string, unknown> {
const status = item.status as Record<string, unknown> | undefined;
const waiting = firstWaitingContainer(status);
const scheduled = conditionSummary(status, "PodScheduled");
const ready = conditionSummary(status, "Ready");
const name = metadataName(item);
const showEvent = status?.phase !== "Running" || waiting !== null || scheduled !== null || ready !== null;
const latestEvent = showEvent && name !== null ? podEvents.get(name) ?? null : null;
return {
name,
phase: status?.phase ?? null,
reason: waiting?.reason ?? scheduled?.reason ?? ready?.reason ?? null,
message: waiting?.message ?? scheduled?.message ?? ready?.message ?? null,
latestEvent: latestEvent === null ? null : {
reason: stringValue(latestEvent.reason),
message: stringValue(latestEvent.message),
type: stringValue(latestEvent.type),
count: typeof latestEvent.count === "number" ? latestEvent.count : null,
timestamp: eventTimestamp(latestEvent),
},
};
}
function latestPodEvents(events: Record<string, unknown>[]): Map<string, Record<string, unknown>> {
const byPod = new Map<string, Record<string, unknown>>();
for (const item of events) {
const involved = item.involvedObject;
if (typeof involved !== "object" || involved === null || Array.isArray(involved)) continue;
const podName = (involved as Record<string, unknown>).name;
const kind = (involved as Record<string, unknown>).kind;
if (kind !== "Pod" || typeof podName !== "string") continue;
const current = byPod.get(podName);
if (current === undefined || eventTimestamp(item) > eventTimestamp(current)) byPod.set(podName, item);
}
return byPod;
}
function eventTimestamp(item: Record<string, unknown>): string | null {
const candidate = stringValue(item.eventTime)
?? stringValue(item.lastTimestamp)
?? stringValue(item.deprecatedLastTimestamp)
?? (typeof item.metadata === "object" && item.metadata !== null && !Array.isArray(item.metadata)
? stringValue((item.metadata as Record<string, unknown>).creationTimestamp)
: null);
return candidate;
}
function stringValue(value: unknown): string | null {
return typeof value === "string" && value.length > 0 ? value : null;
}
function firstWaitingContainer(status: Record<string, unknown> | undefined): { reason: string | null; message: string | null } | null {
const statuses = Array.isArray(status?.containerStatuses) ? status.containerStatuses : [];
for (const item of statuses) {
if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
const state = (item as Record<string, unknown>).state;
if (typeof state !== "object" || state === null || Array.isArray(state)) continue;
const waiting = (state as Record<string, unknown>).waiting;
if (typeof waiting !== "object" || waiting === null || Array.isArray(waiting)) continue;
const record = waiting as Record<string, unknown>;
return {
reason: typeof record.reason === "string" ? record.reason : null,
message: typeof record.message === "string" ? record.message : null,
};
}
return null;
}
function conditionSummary(status: Record<string, unknown> | undefined, type: string): { reason: string | null; message: string | null } | null {
const conditions = Array.isArray(status?.conditions) ? status.conditions : [];
for (const item of conditions) {
if (typeof item !== "object" || item === null || Array.isArray(item)) continue;
const record = item as Record<string, unknown>;
if (record.type !== type || record.status === "True") continue;
return {
reason: typeof record.reason === "string" ? record.reason : null,
message: typeof record.message === "string" ? record.message : null,
};
}
return null;
}
function metadataName(item: Record<string, unknown>): string | null {
const metadata = item.metadata as Record<string, unknown> | undefined;
return typeof metadata?.name === "string" ? metadata.name : null;
}
function manifestObjectSummary(yaml: string): Array<Record<string, string>> {
const docs = yaml.split(/^---$/mu);
return docs.map((doc) => {
const kind = doc.match(/^\s*kind:\s*(.+)$/mu)?.[1]?.trim() ?? "unknown";
const name = doc.match(/^\s*name:\s*(.+)$/mu)?.[1]?.trim() ?? "unknown";
return { kind, name };
});
}
function imageReference(image: ImageSpec): string {
return `${image.repository}:${image.tag}`;
}
function indent(value: string, spaces: number): string {
const prefix = " ".repeat(spaces);
return value.split("\n").map((line) => `${prefix}${line}`).join("\n");
}