fix: 隔离代码代理诊断 trace
This commit is contained in:
Codex
@@ -911,7 +911,7 @@ export function diagnoseCodeAgentScript(observability: ObservabilityConfig, targ
|
||||
return `
|
||||
set -u
|
||||
python3 - <<'PY'
|
||||
import collections, json, re, subprocess, time, urllib.parse
|
||||
import collections, json, re, secrets, subprocess, time, urllib.parse
|
||||
|
||||
BUSINESS_TRACE_ID = ${businessTraceIdLiteral}
|
||||
TRACE_ID = ${traceIdLiteral}
|
||||
@@ -932,6 +932,7 @@ FULL = ${options.full ? "True" : "False"}
|
||||
RAW = ${options.raw ? "True" : "False"}
|
||||
LIMIT = ${options.limit}
|
||||
CANDIDATE_LIMIT = ${options.candidateLimit}
|
||||
DIAGNOSTIC_OPERATION = "unidesk.diagnose-code-agent"
|
||||
DEADLINE = time.time() + 50
|
||||
IMPORTANT_ATTRS = [
|
||||
"traceId", "otel.trace_id", "agentrun.stage", "runId", "commandId",
|
||||
@@ -1077,16 +1078,38 @@ def agentrun_service_proxy_paths(namespace, api_path):
|
||||
"/api/v1/namespaces/%s/services/http:agentrun-mgr:8080/proxy%s" % (namespace, api_path),
|
||||
]
|
||||
|
||||
def run_agentrun_mgr_json(namespace, api_path, timeout=12):
|
||||
def diagnostic_header_items(diagnostic_context):
|
||||
if not isinstance(diagnostic_context, dict):
|
||||
return []
|
||||
return [
|
||||
("traceparent", diagnostic_context.get("traceparent")),
|
||||
("x-agentrun-trace-purpose", "diagnostic"),
|
||||
("x-agentrun-diagnostic-operation", diagnostic_context.get("operation")),
|
||||
("x-agentrun-diagnostic-target-business-trace-id", diagnostic_context.get("targetBusinessTraceId")),
|
||||
("x-agentrun-diagnostic-target-command-id", diagnostic_context.get("targetCommandId")),
|
||||
]
|
||||
|
||||
def run_agentrun_mgr_json(namespace, api_path, timeout=12, diagnostic_context=None):
|
||||
url = "http://127.0.0.1:8080%s" % api_path
|
||||
diagnostic_headers = [
|
||||
(name, str(value))
|
||||
for name, value in diagnostic_header_items(diagnostic_context)
|
||||
if value not in (None, "")
|
||||
]
|
||||
script_parts = []
|
||||
script_parts.append("url=" + shell_quote(url) + "; ")
|
||||
script_parts.append("api_key=\\\"\${AGENTRUN_API_KEY:-}\\\"; ")
|
||||
script_parts.append("if [ -z \\\"$api_key\\\" ]; then api_key=\\\"\${HWLAB_API_KEY:-}\\\"; fi; ")
|
||||
script_parts.append("if command -v wget >/dev/null 2>&1; then ")
|
||||
script_parts.append("wget -qO- --header=\\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ")
|
||||
script_parts.append("wget -qO- --header=\\\"Authorization: Bearer $api_key\\\" ")
|
||||
for name, value in diagnostic_headers:
|
||||
script_parts.append("--header=" + shell_quote(name + ": " + value) + " ")
|
||||
script_parts.append("\\\"$url\\\"; ")
|
||||
script_parts.append("else ")
|
||||
script_parts.append("curl -fsS -H \\\"Authorization: Bearer $api_key\\\" \\\"$url\\\"; ")
|
||||
script_parts.append("curl -fsS -H \\\"Authorization: Bearer $api_key\\\" ")
|
||||
for name, value in diagnostic_headers:
|
||||
script_parts.append("-H " + shell_quote(name + ": " + value) + " ")
|
||||
script_parts.append("\\\"$url\\\"; ")
|
||||
script_parts.append("fi")
|
||||
script = "".join(script_parts)
|
||||
try:
|
||||
@@ -1104,18 +1127,23 @@ def run_agentrun_mgr_json(namespace, api_path, timeout=12):
|
||||
"stderrTail": (proc.stderr or "")[-1000:],
|
||||
"bodyBytes": len((proc.stdout or "").encode("utf-8")),
|
||||
"method": "manager-exec-localhost",
|
||||
"diagnosticContextApplied": bool(diagnostic_headers),
|
||||
}
|
||||
except subprocess.TimeoutExpired as exc:
|
||||
return {"ok": False, "rc": 124, "json": None, "stderrTail": "timeout while querying agentrun manager", "bodyBytes": len((exc.stdout or "").encode("utf-8")), "method": "manager-exec-localhost"}
|
||||
return {"ok": False, "rc": 124, "json": None, "stderrTail": "timeout while querying agentrun manager", "bodyBytes": len((exc.stdout or "").encode("utf-8")), "method": "manager-exec-localhost", "diagnosticContextApplied": bool(diagnostic_headers)}
|
||||
except Exception as exc:
|
||||
return {"ok": False, "rc": 125, "json": None, "stderrTail": str(exc), "bodyBytes": 0, "method": "manager-exec-localhost"}
|
||||
return {"ok": False, "rc": 125, "json": None, "stderrTail": str(exc), "bodyBytes": 0, "method": "manager-exec-localhost", "diagnosticContextApplied": bool(diagnostic_headers)}
|
||||
|
||||
def get_agentrun_json(namespace, api_path, timeout=10):
|
||||
def get_agentrun_json(namespace, api_path, timeout=10, diagnostic_context=None):
|
||||
attempts = []
|
||||
exec_response = run_agentrun_mgr_json(namespace, api_path, timeout=timeout)
|
||||
exec_response = run_agentrun_mgr_json(namespace, api_path, timeout=timeout, diagnostic_context=diagnostic_context)
|
||||
attempts.append(exec_response)
|
||||
if exec_response.get("ok"):
|
||||
return exec_response, attempts
|
||||
if isinstance(diagnostic_context, dict):
|
||||
exec_response["error"] = "diagnostic-transport-failed"
|
||||
exec_response["fallbackDenied"] = "headerless-service-proxy"
|
||||
return exec_response, attempts
|
||||
for proxy_path in agentrun_service_proxy_paths(namespace, api_path):
|
||||
response = run_kubectl_json(proxy_path, timeout=timeout)
|
||||
response["proxyPath"] = proxy_path
|
||||
@@ -1196,7 +1224,14 @@ def preferred_failure_kind(values):
|
||||
return kind
|
||||
return kinds[-1]
|
||||
|
||||
def agentrun_authority_summary(identity, provider_decision):
|
||||
def diagnostic_transport_failure_names(responses):
|
||||
return [
|
||||
name
|
||||
for name, response in responses
|
||||
if isinstance(response, dict) and response.get("error") == "diagnostic-transport-failed"
|
||||
]
|
||||
|
||||
def agentrun_authority_summary(identity, provider_decision, diagnostic_context):
|
||||
run_id = identity.get("runId") if isinstance(identity, dict) else None
|
||||
command_id = identity.get("commandId") if isinstance(identity, dict) else None
|
||||
if not run_id or not command_id:
|
||||
@@ -1204,12 +1239,18 @@ def agentrun_authority_summary(identity, provider_decision):
|
||||
"queried": False,
|
||||
"reason": "missing-run-or-command-id",
|
||||
}
|
||||
if not isinstance(diagnostic_context, dict) or diagnostic_context.get("targetBusinessTraceId") in (None, ""):
|
||||
return {
|
||||
"queried": False,
|
||||
"reason": "missing-unambiguous-target-business-trace-id",
|
||||
}
|
||||
namespace, namespace_source = guess_agentrun_namespace(identity, provider_decision)
|
||||
run_q = urllib.parse.quote(str(run_id), safe="")
|
||||
command_q = urllib.parse.quote(str(command_id), safe="")
|
||||
command_response, command_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s" % (run_q, command_q), timeout=8)
|
||||
result_response, result_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s/result" % (run_q, command_q), timeout=12)
|
||||
events_response, events_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/events?afterSeq=0&limit=1000" % run_q, timeout=12)
|
||||
result_response, result_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/commands/%s/result" % (run_q, command_q), timeout=12, diagnostic_context=diagnostic_context)
|
||||
run_result_response, run_result_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/result?commandId=%s" % (run_q, command_q), timeout=12, diagnostic_context=diagnostic_context)
|
||||
events_response, events_attempts = get_agentrun_json(namespace, "/api/v1/runs/%s/events?afterSeq=0&limit=1000" % run_q, timeout=12, diagnostic_context=diagnostic_context)
|
||||
command_payload = unwrap_payload(command_response.get("json"))
|
||||
result_payload = unwrap_payload(result_response.get("json"))
|
||||
events = event_items(events_response.get("json"))
|
||||
@@ -1227,6 +1268,11 @@ def agentrun_authority_summary(identity, provider_decision):
|
||||
terminal_event.get("status") if isinstance(terminal_event, dict) else None,
|
||||
command_state if command_state in ("completed", "failed", "blocked", "cancelled", "canceled") else None,
|
||||
)
|
||||
diagnostic_transport_failures = diagnostic_transport_failure_names([
|
||||
("command-result", result_response),
|
||||
("run-result", run_result_response),
|
||||
("events", events_response),
|
||||
])
|
||||
return {
|
||||
"queried": True,
|
||||
"namespace": namespace,
|
||||
@@ -1235,7 +1281,10 @@ def agentrun_authority_summary(identity, provider_decision):
|
||||
"ok": bool(command_response.get("ok") or result_response.get("ok") or events_response.get("ok")),
|
||||
"commandOk": command_response.get("ok"),
|
||||
"resultOk": result_response.get("ok"),
|
||||
"runResultOk": run_result_response.get("ok"),
|
||||
"eventsOk": events_response.get("ok"),
|
||||
"diagnosticTransportFailed": len(diagnostic_transport_failures) > 0,
|
||||
"diagnosticTransportFailures": diagnostic_transport_failures,
|
||||
"commandState": command_state,
|
||||
"terminalStatus": terminal_status,
|
||||
"resultTerminalStatus": result_status,
|
||||
@@ -1252,9 +1301,10 @@ def agentrun_authority_summary(identity, provider_decision):
|
||||
"eventCount": len(events),
|
||||
"scopedEventCount": len(scoped_events),
|
||||
"attempts": {
|
||||
"command": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in command_attempts],
|
||||
"result": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in result_attempts],
|
||||
"events": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "stderrTail": item.get("stderrTail")} for item in events_attempts],
|
||||
"command": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "diagnosticContextApplied": item.get("diagnosticContextApplied"), "fallbackDenied": item.get("fallbackDenied"), "error": item.get("error"), "stderrTail": item.get("stderrTail")} for item in command_attempts],
|
||||
"result": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "diagnosticContextApplied": item.get("diagnosticContextApplied"), "fallbackDenied": item.get("fallbackDenied"), "error": item.get("error"), "stderrTail": item.get("stderrTail")} for item in result_attempts],
|
||||
"runResult": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "diagnosticContextApplied": item.get("diagnosticContextApplied"), "fallbackDenied": item.get("fallbackDenied"), "error": item.get("error"), "stderrTail": item.get("stderrTail")} for item in run_result_attempts],
|
||||
"events": [{"ok": item.get("ok"), "rc": item.get("rc"), "method": item.get("method"), "proxyPath": item.get("proxyPath"), "diagnosticContextApplied": item.get("diagnosticContextApplied"), "fallbackDenied": item.get("fallbackDenied"), "error": item.get("error"), "stderrTail": item.get("stderrTail")} for item in events_attempts],
|
||||
},
|
||||
}
|
||||
|
||||
@@ -1556,6 +1606,43 @@ def span_business_trace_id(item):
|
||||
return value.strip()
|
||||
return None
|
||||
|
||||
def authority_target_business_trace_id(spans, identity):
|
||||
if isinstance(BUSINESS_TRACE_ID, str) and BUSINESS_TRACE_ID.startswith("trc_"):
|
||||
return BUSINESS_TRACE_ID
|
||||
command_id = identity.get("commandId") if isinstance(identity, dict) else None
|
||||
command_trace_ids = sorted(set(
|
||||
trace_id
|
||||
for item in spans
|
||||
for trace_id in [span_business_trace_id(item)]
|
||||
if trace_id is not None and command_id not in (None, "") and span_identity_value(item, "commandId") == str(command_id)
|
||||
))
|
||||
if len(command_trace_ids) == 1:
|
||||
return command_trace_ids[0]
|
||||
scoped_trace_ids = sorted(set(
|
||||
trace_id
|
||||
for item in spans
|
||||
for trace_id in [span_business_trace_id(item)]
|
||||
if trace_id is not None
|
||||
))
|
||||
return scoped_trace_ids[0] if len(scoped_trace_ids) == 1 else None
|
||||
|
||||
def new_diagnostic_context(target_business_trace_id, target_command_id, target_otel_trace_id):
|
||||
trace_id = secrets.token_hex(16)
|
||||
while trace_id == "0" * 32 or trace_id == target_otel_trace_id:
|
||||
trace_id = secrets.token_hex(16)
|
||||
parent_span_id = secrets.token_hex(8)
|
||||
while parent_span_id == "0" * 16:
|
||||
parent_span_id = secrets.token_hex(8)
|
||||
return {
|
||||
"traceId": trace_id,
|
||||
"parentSpanId": parent_span_id,
|
||||
"traceparent": "00-%s-%s-01" % (trace_id, parent_span_id),
|
||||
"operation": DIAGNOSTIC_OPERATION,
|
||||
"targetBusinessTraceId": target_business_trace_id,
|
||||
"targetCommandId": target_command_id,
|
||||
"expectedTargetOtelTraceId": target_otel_trace_id,
|
||||
}
|
||||
|
||||
def span_identity_value(item, key):
|
||||
attrs = item.get("attributes", {}) if isinstance(item.get("attributes"), dict) else {}
|
||||
value = attrs.get(key)
|
||||
@@ -2287,7 +2374,44 @@ http_summary = http_status_summary(spans)
|
||||
read_model = hwlab_read_model_summary(spans)
|
||||
identity = identity_from_spans(spans)
|
||||
provider_decision = provider_decision_summary(spans)
|
||||
agentrun_authority = agentrun_authority_summary(identity, provider_decision)
|
||||
target_business_trace_id = authority_target_business_trace_id(spans, identity)
|
||||
diagnostic_context = new_diagnostic_context(
|
||||
target_business_trace_id,
|
||||
identity.get("commandId") if isinstance(identity, dict) else None,
|
||||
resolved_trace_id,
|
||||
)
|
||||
diagnostic_context_ready = (
|
||||
diagnostic_context.get("targetBusinessTraceId") not in (None, "")
|
||||
and diagnostic_context.get("targetCommandId") not in (None, "")
|
||||
)
|
||||
diagnostic_trace = {
|
||||
**diagnostic_context,
|
||||
"contextReady": diagnostic_context_ready,
|
||||
"headerReadPaths": ["command-result", "run-result", "events"],
|
||||
"ordinaryReadPaths": ["command-detail"],
|
||||
"traceCommand": (
|
||||
"bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --full"
|
||||
% diagnostic_context.get("traceId")
|
||||
if diagnostic_context_ready
|
||||
else None
|
||||
),
|
||||
"targetTraceCommand": (
|
||||
"bun scripts/cli.ts platform-infra observability trace --target ${target.id} --trace-id %s --full"
|
||||
% resolved_trace_id
|
||||
),
|
||||
"linkQueryClue": (
|
||||
"inspect diagnostic spans[].links for traceId=%s and attributes "
|
||||
"diagnostic.target_business_trace_id=%s, commandId=%s"
|
||||
% (
|
||||
resolved_trace_id,
|
||||
target_business_trace_id,
|
||||
diagnostic_context.get("targetCommandId"),
|
||||
)
|
||||
if diagnostic_context_ready
|
||||
else "diagnostic reads skipped because the business trace and command target were not unambiguous"
|
||||
),
|
||||
}
|
||||
agentrun_authority = agentrun_authority_summary(identity, provider_decision, diagnostic_context)
|
||||
agentrun = agentrun_summary(spans, agentrun_authority)
|
||||
lag = projection_lag_summary(agentrun, read_model)
|
||||
candidates = root_cause_candidates(http_summary, agentrun, read_model, lag, error_spans, idle_warning_spans)
|
||||
@@ -2441,6 +2565,7 @@ payload = {
|
||||
"observabilityGap": observability_gap,
|
||||
"businessTraceIds": business_trace_ids[:20],
|
||||
"businessTraceScope": business_scope,
|
||||
"diagnosticTrace": diagnostic_trace,
|
||||
"providerDecision": provider_decision,
|
||||
"identity": identity,
|
||||
"agentrun": {
|
||||
|
||||
Reference in New Issue
Block a user