fix: 压缩 OTel trace CLI 输出

2026-06-20 01:23:48 +00:00
parent 126271e770
commit ca9c1424fe
2 changed files with 222 additions and 10 deletions
@@ -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<string, unknown> {
@@ -141,7 +142,7 @@ export function observabilityHelp(): Record<string, unknown> {
      "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> [--full|--raw]",
+      "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id <traceId> [--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<Reco
  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));
+  const result = await capture(config, target.route, ["sh"], traceScript(observability, target, tracePath, options));
  const parsed = parseJsonOutput(result.stdout);
  return {
    ok: result.exitCode === 0 && parsed?.ok === true,
@@ -497,7 +512,7 @@ async function trace(config: UniDeskConfig, options: TraceOptions): Promise<Reco
      service: observability.traceBackend.serviceName,
      path: tracePath,
    },
-    result: options.raw ? redactSensitiveUnknown(parsed) : compactUnknown(redactSensitiveUnknown(parsed)),
+    result: redactSensitiveUnknown(parsed),
  };
 }

@@ -1093,8 +1108,9 @@ PY
 `;
 }

-function traceScript(observability: ObservabilityConfig, target: ObservabilityTarget, tracePath: string): string {
+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)"
@@ -1102,24 +1118,220 @@ 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 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 "<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}",
-  "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 ?? "<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
@@ -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 <traceId>",
+      "bun scripts/cli.ts platform-infra observability trace --target D601 --trace-id <traceId> [--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,