feat: restore parallel ops and diagnostics changes

2026-07-14 06:40:02 +02:00
parent fcc5227cc1
commit a91238c200
22 changed files with 2039 additions and 63 deletions
@@ -2867,6 +2867,11 @@ def classify_trace_event(item, account_names_by_id):
            "error": item.get("error"),
            "excludedAccountCount": item.get("excluded_account_count"),
        })
+    elif "openai.forward_failed" in message:
+        event.update({
+            "type": "forward-failed",
+            "error": item.get("error"),
+        })
    elif "account_upstream_error" in message:
        event.update({
            "type": "upstream-error",
@@ -2939,10 +2944,15 @@ def trace_reason(events, final_event):
    failovers = [item for item in events if item.get("type") == "failover"]
    select_failures = [item for item in events if item.get("type") == "select-failed"]
    upstream_errors = [item for item in events if item.get("type") == "upstream-error"]
+    forward_failures = [item for item in events if item.get("type") == "forward-failed"]
    if failover_budget_exhausted(failovers, final_event):
        return "failover-budget-exhausted"
    if failovers and select_failures:
        return "failover-attempted-no-candidate"
+    if forward_failures and isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400:
+        return "forward-failed-http-success"
+    if forward_failures:
+        return "forward-failed"
    if failovers:
        return "failover-attempted"
    if select_failures:
@@ -3006,6 +3016,60 @@ def run_trace():
    if not isinstance(request_id, str) or not request_id:
        raise RuntimeError("trace payload missing requestId")
    admin_email, token, admin_compliance = login()
+    request_query = quote(request_id, safe="")
+    native_request_status = {"status": "unavailable"}
+    native_request = None
+    try:
+        request_data = ensure_success(
+            curl_api("GET", f"/api/v1/admin/ops/requests?request_id={request_query}&kind=all&page=1&page_size=100", bearer=token),
+            "get ops request details",
+        )
+        request_items = extract_items(request_data)
+        native_request = next((item for item in request_items if isinstance(item, dict) and item.get("request_id") == request_id), None)
+        native_request_status = {
+            "status": "available",
+            "matched": native_request is not None,
+            "returned": len(request_items),
+            "total": request_data.get("total", len(request_items)) if isinstance(request_data, dict) else len(request_items),
+        }
+    except Exception as exc:
+        native_request_status = {"status": "unavailable", "reason": text(str(exc), 500)}
+
+    native_system_log_status = {"status": "unavailable"}
+    native_matched = []
+    try:
+        system_log_data = ensure_success(
+            curl_api("GET", f"/api/v1/admin/ops/system-logs?request_id={request_query}&page=1&page_size=200", bearer=token),
+            "get ops request system logs",
+        )
+        system_log_items = extract_items(system_log_data)
+        for row in system_log_items:
+            if not isinstance(row, dict):
+                continue
+            parsed = dict(row.get("extra") or {}) if isinstance(row.get("extra"), dict) else {}
+            for source_key in ("request_id", "client_request_id", "account_id", "platform", "model"):
+                if parsed.get(source_key) is None and row.get(source_key) is not None:
+                    parsed[source_key] = row.get(source_key)
+            parsed["_at"] = row.get("created_at")
+            parsed["_message"] = row.get("message")
+            parsed["_line"] = json.dumps({
+                "created_at": row.get("created_at"),
+                "message": row.get("message"),
+                "request_id": row.get("request_id"),
+                "account_id": row.get("account_id"),
+                "platform": row.get("platform"),
+                "model": row.get("model"),
+                "extra": parsed,
+            }, ensure_ascii=False, default=str)
+            native_matched.append(parsed)
+        native_system_log_status = {
+            "status": "available",
+            "returned": len(system_log_items),
+            "total": system_log_data.get("total", len(system_log_items)) if isinstance(system_log_data, dict) else len(system_log_items),
+        }
+    except Exception as exc:
+        native_system_log_status = {"status": "unavailable", "reason": text(str(exc), 500)}
+
    account_snapshot, account_snapshot_error = account_snapshot_from_runtime(token)
    account_names_by_id = {}
    for row in account_snapshot:
@@ -3014,17 +3078,21 @@ def run_trace():
            account_id = int(account_id)
        if isinstance(account_id, int) and isinstance(row.get("accountName"), str):
            account_names_by_id[account_id] = row.get("accountName")
-    proc = kubectl(["-n", NAMESPACE, "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"])
-    stdout = proc.stdout.decode("utf-8", errors="replace")
+    proc = runtime_logs(since, tail)
+    log_bytes = proc.stdout + b"\\n" + proc.stderr
+    stdout = log_bytes.decode("utf-8", errors="replace")
    parsed_lines = []
-    matched = []
+    runtime_matched = []
    for line in stdout.splitlines():
        parsed = parse_log_line(line)
        if parsed is None:
            continue
        parsed_lines.append(parsed)
        if request_id in line:
-            matched.append(parsed)
+            runtime_matched.append(parsed)
+    matched = native_matched if native_matched else runtime_matched
+    matched.sort(key=lambda item: (log_time_epoch(item) is None, log_time_epoch(item) or 0))
+    trace_event_source = "sub2api-native-admin-ops-system-logs" if native_matched else "runtime-logs-fallback"
    first_epoch = None
    last_epoch = None
    for item in matched:
@@ -3045,22 +3113,55 @@ def run_trace():
        window_lines = matched
    events = [classify_trace_event(item, account_names_by_id) for item in matched]
    request_start = next((item for item in events if item.get("type") == "request-start"), None)
+    if isinstance(native_request, dict):
+        if not isinstance(request_start, dict):
+            request_start = {
+                "type": "request-start",
+                "at": native_request.get("created_at"),
+                "requestId": native_request.get("request_id"),
+                "path": None,
+                "model": native_request.get("model"),
+                "accountId": native_request.get("account_id"),
+                "accountName": account_names_by_id.get(native_request.get("account_id")),
+                "stream": native_request.get("stream"),
+                "source": "sub2api-native-admin-ops-requests",
+            }
+        else:
+            if request_start.get("stream") is None:
+                request_start["stream"] = native_request.get("stream")
+            if not request_start.get("model"):
+                request_start["model"] = native_request.get("model")
    final_event = next((item for item in reversed(events) if item.get("type") == "final"), None)
+    if not isinstance(final_event, dict) and isinstance(native_request, dict) and isinstance(native_request.get("status_code"), int):
+        final_event = {
+            "type": "final",
+            "at": native_request.get("created_at"),
+            "requestId": native_request.get("request_id"),
+            "path": None,
+            "model": native_request.get("model"),
+            "accountId": native_request.get("account_id"),
+            "accountName": account_names_by_id.get(native_request.get("account_id")),
+            "statusCode": native_request.get("status_code"),
+            "latencyMs": native_request.get("duration_ms"),
+            "source": "sub2api-native-admin-ops-requests",
+        }
    failovers = [item for item in events if item.get("type") == "failover"]
    select_failures = [item for item in events if item.get("type") == "select-failed"]
    upstream_errors = [item for item in events if item.get("type") == "upstream-error"]
+    forward_failures = [item for item in events if item.get("type") == "forward-failed"]
    temp_unsched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if "account_temp_unschedulable" in str(item.get("_message") or "")]
    admin_sched = [with_trace_phase(classify_trace_event(item, account_names_by_id), first_epoch, last_epoch) for item in window_lines if ("schedulable" in str(item.get("_message") or "") or "/schedulable" in str(item.get("path") or ""))]
    window_events = [classify_trace_event(item, account_names_by_id) for item in window_lines]
    final_errors = [item for item in window_events if item.get("type") == "final" and isinstance(item.get("statusCode"), int) and item.get("statusCode") >= 400]
    window_failovers = [item for item in window_events if item.get("type") == "failover"]
    window_select_failures = [item for item in window_events if item.get("type") == "select-failed"]
+    window_forward_failures = [item for item in window_events if item.get("type") == "forward-failed"]
    untried_schedulable_accounts = trace_untried_schedulable_accounts(failovers, final_event or {}, account_snapshot)
    reason = trace_reason(events, final_event)
    if not matched:
        outcome = "not-found"
    elif isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") < 400:
-        outcome = "succeeded"
+        outcome = "degraded" if forward_failures else ("succeeded-with-failover" if failovers else "succeeded")
    elif isinstance(final_event, dict):
        outcome = "failed"
    else:
@@ -3068,11 +3169,14 @@ def run_trace():
    return {
        "ok": proc.returncode == 0 and len(matched) > 0,
        "mode": "trace",
+        "targetId": TARGET_ID,
+        "runtimeMode": RUNTIME_MODE,
        "namespace": NAMESPACE,
        "serviceDns": SERVICE_DNS,
        "appPod": APP_POD,
        "admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
        "requestId": request_id,
+        "eventSource": trace_event_source,
        "summary": {
            "outcome": outcome,
            "reason": reason,
@@ -3090,10 +3194,16 @@ def run_trace():
        },
        "request": request_start or {},
        "final": final_event or {},
+        "nativeOps": {
+            "requestDetails": native_request_status,
+            "request": native_request,
+            "systemLogs": native_system_log_status,
+        },
        "events": events,
        "failovers": failovers,
        "selectFailures": select_failures,
        "upstreamErrors": upstream_errors,
+        "forwardFailures": forward_failures,
        "tempUnschedulable": temp_unsched,
        "adminSchedulable": admin_sched[-20:],
        "windowStats": {
@@ -3102,6 +3212,8 @@ def run_trace():
            "finalErrorCount": len(final_errors),
            "failoverCount": len(window_failovers),
            "selectFailedCount": len(window_select_failures),
+            "forwardFailedCount": len(window_forward_failures),
+            "matchedForwardFailedCount": len(forward_failures),
            "tempUnschedulableCount": len(temp_unsched),
            "adminSchedulableCount": len(admin_sched),
        },
@@ -3114,8 +3226,10 @@ def run_trace():
        "rawLines": [{"line": item.get("_line")} for item in matched[-30:]] if show_lines else [],
        "showLines": show_lines,
        "logs": {
+            "source": trace_event_source,
+            "contextSource": f"host-docker:{HOST_DOCKER_APP_CONTAINER}" if RUNTIME_MODE == "host-docker" else f"k3s:{NAMESPACE}/deployment/sub2api",
            "exitCode": proc.returncode,
-            "stderrTail": text(proc.stderr, 1000),
+            "stderrTail": text(proc.stderr, 1000) if proc.returncode != 0 else "",
            "stdoutLineCount": len(stdout.splitlines()),
        },
        "valuesPrinted": False,