mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: stabilize observability runtime apply

Browse Source
This commit is contained in:
Codex
2026-06-19 14:19:55 +00:00
parent d58b46f6ab
commit 66f71bec62
3 changed files with 388 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/platform-infra-observability.ts
+242 -43
View File
@@ -4,7 +4,6 @@ import { Buffer } from "node:buffer";
import { readFileSync } from "node:fs";
import type { UniDeskConfig } from "./config";
import { rootPath } from "./config";
import { startJob } from "./jobs";
import {
compactCapture,
compactUnknown,
@@ -212,7 +211,7 @@ function parseTraceOptions(args: string[]): TraceOptions {
if (arg === "--trace-id") {
const value = args[index + 1];
if (value === undefined || value.startsWith("--")) throw new Error("--trace-id requires a value");
if (!/^[A-Za-z0-9._:-]+$/u.test(value)) throw new Error("--trace-id has an unsupported format");
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 {
@@ -409,27 +408,6 @@ async function apply(config: UniDeskConfig, options: ApplyOptions): Promise<Reco
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 };
if (options.confirm && !options.wait) {
const job = startJob(
`platform_infra_observability_apply_${target.id.toLowerCase()}`,
["bun", "scripts/cli.ts", "platform-infra", "observability", "apply", "--target", target.id, "--confirm", "--wait"],
`Apply ${target.id} platform-infra OTel Collector and trace backend through the controlled UniDesk CLI`,
);
return {
ok: true,
action: "platform-infra-observability-apply",
mode: "async-job",
mutation: true,
target: targetSummary(target),
job,
statusCommand: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
next: {
status: `bun scripts/cli.ts job status ${job.id} --tail-bytes 12000`,
rollout: `bun scripts/cli.ts platform-infra observability status --target ${target.id}`,
validate: `bun scripts/cli.ts platform-infra observability validate --target ${target.id}`,
},
};
}
const result = await capture(config, target.route, ["sh"], applyScript({
yaml,
target,
@@ -474,20 +452,27 @@ async function status(config: UniDeskConfig, options: CommonOptions): Promise<Re
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));
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,
ok: result.exitCode === 0 && ready && validationTraceOk,
action: "platform-infra-observability-validate",
mutation: false,
target: targetSummary(target),
summary,
validation: {
readiness: ready ? "passed" : "failed",
testTrace: "not-generated-by-this-stage",
traceQuery: ready ? `bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id <traceId>` : "blocked-until-runtime-ready",
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 }),
@@ -621,6 +606,11 @@ metadata:
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}
@@ -748,6 +738,11 @@ metadata:
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}
@@ -833,15 +828,7 @@ function applyScript(params: {
}): string {
const encoded = Buffer.from(params.yaml, "utf8").toString("base64");
const dryRunArg = params.dryRun ? "--dry-run=server" : "";
const wait = params.dryRun || !params.wait
? "wait_disposition=skipped"
: [
`kubectl -n ${shQuote(params.target.namespace)} rollout status deployment/${shQuote(params.collectorDeploymentName)} --timeout=180s >"$tmp/collector-rollout.out" 2>"$tmp/collector-rollout.err"`,
"collector_rollout_rc=$?",
`kubectl -n ${shQuote(params.target.namespace)} rollout status deployment/${shQuote(params.backendDeploymentName)} --timeout=180s >"$tmp/backend-rollout.out" 2>"$tmp/backend-rollout.err"`,
"backend_rollout_rc=$?",
"wait_disposition=executed",
].join("\n");
const waitDisposition = params.wait ? "skipped-short-connection-use-status" : "skipped";
return `
set -u
tmp="$(mktemp -d)"
@@ -852,7 +839,7 @@ kubectl apply --server-side --field-manager=${fieldManager} ${dryRunArg} -f "$ma
apply_rc=$?
collector_rollout_rc=0
backend_rollout_rc=0
${wait}
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):
@@ -881,7 +868,7 @@ PY
`;
}
function statusScript(observability: ObservabilityConfig, target: ObservabilityTarget, full: boolean): string {
function statusScript(observability: ObservabilityConfig, target: ObservabilityTarget, full: boolean, generateValidationTrace = false): string {
const endpointsJson = JSON.stringify(observability.probes.statusEndpoints);
return `
set -u
@@ -903,8 +890,9 @@ capture_json namespace kubectl get namespace ${shQuote(target.namespace)} -o jso
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, subprocess, sys
import json, random, socket, subprocess, sys, time, urllib.error, urllib.request
tmp = sys.argv[1]
endpoints = json.loads(sys.argv[2])
namespace = "${target.namespace}"
@@ -952,8 +940,141 @@ for ep in endpoints:
"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": rc("namespace") == 0 and rc("deployments") == 0 and rc("services") == 0 and all(item["ok"] for item in probe_results),
"ok": runtime_ready and (validation_trace is None or validation_trace.get("ok") is True),
"target": "${target.id}",
"namespace": namespace,
"sections": {
@@ -961,8 +1082,10 @@ payload = {
"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)
@@ -1043,6 +1166,8 @@ function statusSummary(payload: Record<string, unknown>): Record<string, unknown
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 {
@@ -1050,7 +1175,7 @@ function statusSummary(payload: Record<string, unknown>): Record<string, unknown
namespace: payload.namespace,
deployments: readyDeployments,
services: services.map((item) => metadataName(item)),
pods: pods.map((item) => podSummary(item)),
pods: pods.map((item) => podSummary(item, podEvents)),
probes: probes.map((item) => ({
name: item.name,
ok: item.ok === true,
@@ -1092,14 +1217,88 @@ function deploymentSummary(item: Record<string, unknown>): Record<string, unknow
};
}
function podSummary(item: Record<string, unknown>): Record<string, unknown> {
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: metadataName(item),
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;