Merge pull request #578 from pikasTech/fix-agentrun-gitmirror-retry-1823

feat: search OTel HTTP path status spans
2026-06-21 14:58:47 +08:00
parent 007f93588a 942499d0d2
commit 359dfa865a
1 changed files with 92 additions and 12 deletions
@@ -133,6 +133,8 @@ interface TraceOptions extends CommonOptions {
 interface SearchOptions extends CommonOptions {
  grep: string | null;
  query: string | null;
+  path: string | null;
+  status: number | null;
  limit: number;
  candidateLimit: number;
  lookbackMinutes: number;
@@ -160,6 +162,7 @@ export function observabilityHelp(): Record<string, unknown> {
      "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 search --target D601 --path /v1/workbench/sessions --status 502 [--lookback-minutes 120] [--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.",
@@ -264,6 +267,8 @@ function parseSearchOptions(args: string[]): SearchOptions {
  const commonArgs: string[] = [];
  let grep: string | null = null;
  let query: string | null = null;
+  let path: string | null = null;
+  let status: number | null = null;
  let limit = 20;
  let candidateLimit = 80;
  let lookbackMinutes = 360;
@@ -279,6 +284,19 @@ function parseSearchOptions(args: string[]): SearchOptions {
      if (value === undefined || value.startsWith("--")) throw new Error(`${arg} requires a value`);
      query = value;
      index += 1;
+    } else if (arg === "--path") {
+      const value = args[index + 1];
+      if (value === undefined || value.startsWith("--")) throw new Error("--path requires a value");
+      if (!value.startsWith("/") || value.includes("\n") || value.length > 300) throw new Error("--path must be an absolute HTTP path up to 300 characters");
+      path = value;
+      index += 1;
+    } else if (arg === "--status") {
+      const value = args[index + 1];
+      if (value === undefined || value.startsWith("--")) throw new Error("--status requires a value");
+      const parsed = Number(value);
+      if (!Number.isInteger(parsed) || parsed < 100 || parsed > 599) throw new Error("--status must be an integer HTTP status from 100 to 599");
+      status = parsed;
+      index += 1;
    } else if (arg === "--limit") {
      const value = args[index + 1];
      if (value === undefined || value.startsWith("--")) throw new Error("--limit requires a value");
@@ -308,8 +326,8 @@ function parseSearchOptions(args: string[]): SearchOptions {
      }
    }
  }
-  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 };
+  if (grep === null && query === null && path === null && status === null) throw new Error("observability search requires --grep <text>, --query <tempo-query>, --path <route>, or --status <code>");
+  return { ...parseCommonOptions(commonArgs), grep, query, path, status, limit, candidateLimit, lookbackMinutes };
 }

 function parseDiagnoseCodeAgentOptions(args: string[]): DiagnoseCodeAgentOptions {
@@ -647,12 +665,13 @@ async function search(config: UniDeskConfig, options: SearchOptions): Promise<Re
  const target = resolveTarget(observability, options.targetId);
  const endSeconds = Math.floor(Date.now() / 1000);
  const startSeconds = endSeconds - (options.lookbackMinutes * 60);
+  const tempoQuery = options.query ?? inferSearchTempoQuery(options);
  const params = new URLSearchParams({
    start: String(startSeconds),
    end: String(endSeconds),
    limit: String(options.candidateLimit),
  });
-  if (options.query !== null) params.set("q", options.query);
+  if (tempoQuery !== null) params.set("q", tempoQuery);
  const searchPath = `/api/search?${params.toString()}`;
  const result = await capture(config, target.route, ["sh"], searchScript(observability, target, searchPath, options));
  const parsed = parseJsonOutput(result.stdout);
@@ -666,7 +685,10 @@ async function search(config: UniDeskConfig, options: SearchOptions): Promise<Re
      service: observability.traceBackend.serviceName,
      path: searchPath,
      grep: options.grep,
-      tempoQuery: options.query,
+      tempoQuery,
+      explicitTempoQuery: options.query,
+      pathFilter: options.path,
+      statusFilter: options.status,
      lookbackMinutes: options.lookbackMinutes,
      candidateLimit: options.candidateLimit,
      limit: options.limit,
@@ -675,6 +697,12 @@ async function search(config: UniDeskConfig, options: SearchOptions): Promise<Re
  };
 }

+function inferSearchTempoQuery(options: SearchOptions): string | null {
+  if (options.path !== null) return `{ .http.route = ${JSON.stringify(options.path)} }`;
+  if (options.status !== null) return `{ .http.response.status_code = ${options.status} }`;
+  return null;
+}
+
 async function diagnoseCodeAgent(config: UniDeskConfig, options: DiagnoseCodeAgentOptions): Promise<Record<string, unknown>> {
  const observability = readObservabilityConfig();
  const target = resolveTarget(observability, options.targetId);
@@ -1549,7 +1577,10 @@ function searchScript(observability: ObservabilityConfig, target: ObservabilityT
  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);
+  const effectiveQuery = inferSearchTempoQuery(options);
+  const queryLiteral = effectiveQuery === null ? "None" : JSON.stringify(effectiveQuery);
+  const pathLiteral = options.path === null ? "None" : JSON.stringify(options.path);
+  const statusLiteral = options.status === null ? "None" : String(options.status);
  return `
 set -u
 python3 - <<'PY'
@@ -1562,6 +1593,8 @@ FULL = ${options.full ? "True" : "False"}
 RAW = ${options.raw ? "True" : "False"}
 GREP = ${grepLiteral}
 QUERY = ${queryLiteral}
+PATH_FILTER = ${pathLiteral}
+STATUS_FILTER = ${statusLiteral}
 LIMIT = ${options.limit}
 CANDIDATE_LIMIT = ${options.candidateLimit}
 DEADLINE = time.time() + 50
@@ -1571,7 +1604,8 @@ IMPORTANT_ATTRS = [
    "retryAttempt", "retryMax", "retryExhausted", "retryBackoffMs",
    "upstreamHttpStatus", "upstreamHost", "providerErrorClass", "errorSummary",
    "terminalStatus", "phase", "message", "http.route", "http.status_code",
-    "http.method", "eventType", "runnerId", "attemptId", "backendProfile",
+    "http.response.status_code", "http.method", "http.request.method",
+    "http.target", "http.url", "url.path", "eventType", "runnerId", "attemptId", "backendProfile",
    "sourceCommit", "jobName", "podName", "logPath",
    "toolName", "type", "itemType", "itemId", "status", "exitCode",
    "durationMs", "cwd", "processId", "command", "commandFingerprint",
@@ -1651,13 +1685,32 @@ def span_status_code(status):
        return None
    return status.get("code")

+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 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")
+    http_status = to_int(attrs.get("http.response.status_code"))
+    if http_status is None:
+        http_status = to_int(attrs.get("http.status_code"))
+    return code in ("STATUS_CODE_ERROR", 2) or "error" in name or bool(failure) or terminal in ("failed", "blocked") or (http_status is not None and http_status >= 500)

 def batches_from_body(body):
    if not isinstance(body, dict):
@@ -1701,6 +1754,26 @@ def grep_matches_text(text):
 def grep_matches_item(item):
    return GREP is not None and grep_matches_text(json.dumps(item, ensure_ascii=False, sort_keys=True))

+def span_matches_filters(item):
+    attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
+    if PATH_FILTER is not None:
+        candidates = [
+            attrs.get("http.route"),
+            attrs.get("http.target"),
+            attrs.get("http.url"),
+            attrs.get("url.path"),
+            item.get("name"),
+        ]
+        if not any(isinstance(value, str) and (value == PATH_FILTER or value.startswith(PATH_FILTER + "?")) for value in candidates):
+            return False
+    if STATUS_FILTER is not None:
+        status_value = to_int(attrs.get("http.response.status_code"))
+        if status_value is None:
+            status_value = to_int(attrs.get("http.status_code"))
+        if status_value != STATUS_FILTER:
+            return False
+    return True
+
 def extract_traces(search_body):
    if not isinstance(search_body, dict):
        return []
@@ -1733,6 +1806,7 @@ def compact_meta(meta):
 def trace_summary(trace_id, meta, body, rc, stderr):
    parsed = parse_json(body)
    raw_matched = grep_matches_text(body)
+    matching_active = GREP is not None or PATH_FILTER is not None or STATUS_FILTER is not None
    if not isinstance(parsed, dict):
        return {
            "traceId": trace_id,
@@ -1741,6 +1815,7 @@ def trace_summary(trace_id, meta, body, rc, stderr):
            "queryOk": rc == 0,
            "parseOk": False,
            "rawMatched": raw_matched,
+            "matchingActive": matching_active,
            "bodyBytes": len(body.encode("utf-8")),
            "stderrTail": (stderr or "")[-1000:],
        }
@@ -1771,7 +1846,7 @@ def trace_summary(trace_id, meta, body, rc, stderr):
                spans.append(item)
                if is_error_span(span, attrs if isinstance(attrs, dict) else {}):
                    error_spans.append(item)
-                if grep_matches_item(item):
+                if span_matches_filters(item) and (GREP is None or grep_matches_item(item)):
                    matched_spans.append(item)
    return {
        "traceId": trace_id,
@@ -1781,13 +1856,14 @@ def trace_summary(trace_id, meta, body, rc, stderr):
        "queryOk": rc == 0,
        "parseOk": True,
        "rawMatched": raw_matched,
+        "matchingActive": matching_active,
        "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,
+        "matchedSpanCount": len(matched_spans) if matching_active 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],
@@ -1820,9 +1896,9 @@ for trace_id, meta in candidate_trace_ids:
    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:
+    if summary.get("matchingActive") is not True 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:
+    if len(matched) >= LIMIT and summary.get("matchingActive") is True:
        break

 payload = {
@@ -1831,6 +1907,8 @@ payload = {
    "searchProxyPath": SEARCH_PROXY_PATH,
    "grep": GREP,
    "tempoQuery": QUERY,
+    "pathFilter": PATH_FILTER,
+    "statusFilter": STATUS_FILTER,
    "limit": LIMIT,
    "candidateLimit": CANDIDATE_LIMIT,
    "searchParseOk": search_parse_ok,
@@ -1838,13 +1916,15 @@ payload = {
    "scannedTraceCount": len(scanned),
    "matchedTraceCount": len(matched),
    "scanStopped": scan_stopped,
-    "traces": matched[:LIMIT] if GREP is not None else scanned[:LIMIT],
+    "matchingActive": GREP is not None or PATH_FILTER is not None or STATUS_FILTER is not None,
+    "traces": matched[:LIMIT] if (GREP is not None or PATH_FILTER is not None or STATUS_FILTER 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",
+        "pathStatus": "bun scripts/cli.ts platform-infra observability search --target ${target.id} --path /v1/workbench/sessions --status 502 --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",
    },