diff --git a/scripts/src/platform-infra-observability.ts b/scripts/src/platform-infra-observability.ts index 7d4d14a1..76cd3ffd 100644 --- a/scripts/src/platform-infra-observability.ts +++ b/scripts/src/platform-infra-observability.ts @@ -6,7 +6,6 @@ import type { UniDeskConfig } from "./config"; import { rootPath } from "./config"; import { compactCapture, - compactUnknown, createYamlFieldReader, numberField, parseJsonOutput, @@ -127,6 +126,8 @@ interface ApplyOptions extends CommonOptions { interface TraceOptions extends CommonOptions { traceId: string | null; + grep: string | null; + limit: number; } export function observabilityHelp(): Record { @@ -141,7 +142,7 @@ export function observabilityHelp(): Record { "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 [--full|--raw]", + "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id [--grep provider-stream-disconnected] [--limit 40] [--full|--raw]", ], boundary: "Prometheus remains the metrics source; this command owns only platform-infra OTel Collector, trace backend readiness, and trace lookup.", }; @@ -206,6 +207,8 @@ function parseApplyOptions(args: string[]): ApplyOptions { 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") { @@ -214,6 +217,18 @@ function parseTraceOptions(args: string[]): TraceOptions { 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") { @@ -222,7 +237,7 @@ function parseTraceOptions(args: string[]): TraceOptions { } } } - return { ...parseCommonOptions(commonArgs), traceId }; + return { ...parseCommonOptions(commonArgs), traceId, grep, limit }; } function readObservabilityConfig(): ObservabilityConfig { @@ -484,7 +499,7 @@ async function trace(config: UniDeskConfig, options: TraceOptions): Promise"$tmp/trace.out" 2>"$tmp/trace.err" rc=$? python3 - "$rc" "$tmp/trace.out" "$tmp/trace.err" <<'PY' -import json, sys +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", "failureKind", "willRetry", + "terminalStatus", "phase", "message", "http.route", "http.status_code", + "http.method", "eventType", "runnerId", "attemptId", +] +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 "" -body = text(sys.argv[2], 200000) +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 "" + if "kvlistValue" in inner: + return "" + if "bytesValue" in inner: + return "" + 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 "" + 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}", - "body": parsed, + "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 ?? ""} --full --limit 200", + "grepProviderStream": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.traceId ?? ""} --grep provider-stream-disconnected --limit 40", + "raw": "bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id ${options.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 diff --git a/scripts/src/platform-infra.ts b/scripts/src/platform-infra.ts index edc8a5b2..14896573 100644 --- a/scripts/src/platform-infra.ts +++ b/scripts/src/platform-infra.ts @@ -302,7 +302,7 @@ export function platformInfraHelp(): unknown { "bun scripts/cli.ts platform-infra observability apply --target D601 --confirm", "bun scripts/cli.ts platform-infra observability status --target D601", "bun scripts/cli.ts platform-infra observability validate --target D601", - "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id ", + "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id [--grep text] [--limit 40] [--full|--raw]", ], description: "Operate YAML-controlled platform-infra services such as Sub2API, LangBot, n8n, WeChat archive workflows and OpenTelemetry tracing. Public services use PK01 Caddy+FRP rather than Kubernetes Ingress, NodePort, or LoadBalancer.", target,