mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: 合并固定主工作区并行提交

Browse Source
This commit is contained in:
Codex
2026-07-14 15:29:20 +02:00
parent e90a096762 4b55fdb711
commit 72f3589c18
37 changed files with 4174 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/src/platform-infra-sub2api-codex/remote-python-trace.ts
+196 -8
View File
@@ -88,6 +88,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",
@@ -160,10 +165,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:
@@ -217,6 +227,16 @@ def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot)
})
return result
def ops_start_time_for_since(since):
if not isinstance(since, str):
return None
match = re.fullmatch(r"\s*(\d+)\s*([smhd])\s*", since.lower())
if match is None:
return None
value = int(match.group(1))
multiplier = {"s": 1, "m": 60, "h": 3600, "d": 86400}[match.group(2)]
return (datetime.now(timezone.utc) - timedelta(seconds=value * multiplier)).strftime("%Y-%m-%dT%H:%M:%SZ")
def run_trace():
payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {}
request_id = payload.get("requestId")
@@ -227,6 +247,89 @@ 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),
}
if native_request is None:
ops_start_time = ops_start_time_for_since(since)
error_query = f"/api/v1/admin/ops/request-errors?q={request_query}&view=all&page=1&page_size=20"
if isinstance(ops_start_time, str):
error_query += "&start_time=" + quote(ops_start_time, safe="")
error_data = ensure_success(
curl_api("GET", error_query, bearer=token),
"get ops request error fallback",
)
error_items = extract_items(error_data)
native_request = next((item for item in error_items if isinstance(item, dict) and item.get("request_id") == request_id), None)
native_request_status.update({
"fallback": "sub2api-native-admin-ops-request-errors",
"fallbackStartTime": ops_start_time,
"fallbackReturned": len(error_items),
"fallbackTotal": error_data.get("total", len(error_items)) if isinstance(error_data, dict) else len(error_items),
})
if isinstance(native_request, dict):
native_request = dict(native_request)
native_request["kind"] = "error"
error_id = native_request.get("id")
if isinstance(error_id, int):
try:
detail = ensure_success(curl_api("GET", f"/api/v1/admin/ops/request-errors/{error_id}", bearer=token), "get ops request error detail")
if isinstance(detail, dict):
native_request.update(detail)
except Exception:
pass
native_request_status.update({"matched": True, "source": "sub2api-native-admin-ops-request-errors"})
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:
@@ -235,17 +338,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:
@@ -266,34 +373,94 @@ 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": native_request.get("request_path") or native_request.get("inbound_endpoint"),
"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": native_request.get("request_path") or native_request.get("inbound_endpoint"),
"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",
}
reference_at = request_start.get("at") if isinstance(request_start, dict) else None
reference_epoch = log_time_epoch({"_at": reference_at}) if isinstance(reference_at, str) else first_epoch
reference_source = "request-start" if isinstance(reference_at, str) and reference_epoch is not None else "first-matched-event"
native_duration_ms = native_request.get("duration_ms") if isinstance(native_request, dict) else None
if not isinstance(native_duration_ms, int) and isinstance(final_event, dict):
native_duration_ms = final_event.get("latencyMs")
if (
isinstance(request_start, dict)
and request_start.get("source") == "sub2api-native-admin-ops-requests"
and isinstance(native_duration_ms, int)
and native_duration_ms >= 0
and reference_epoch is not None
):
reference_epoch -= native_duration_ms / 1000
reference_at = datetime.fromtimestamp(reference_epoch, timezone.utc).isoformat().replace("+00:00", "Z")
reference_source = "native-completion-minus-duration"
for event in events:
epoch = log_time_epoch({"_at": event.get("at")}) if isinstance(event.get("at"), str) else None
event["elapsedMs"] = round((epoch - reference_epoch) * 1000) if epoch is not None and reference_epoch is not None else None
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)
first_failover_elapsed = failovers[0].get("elapsedMs") if failovers else None
first_select_failed_elapsed = select_failures[0].get("elapsedMs") if select_failures else None
final_epoch = log_time_epoch({"_at": final_event.get("at")}) if isinstance(final_event, dict) and isinstance(final_event.get("at"), str) else None
final_elapsed = round((final_epoch - reference_epoch) * 1000) if final_epoch is not None and reference_epoch is not None else None
failover_to_final = final_elapsed - first_failover_elapsed if isinstance(final_elapsed, int) and isinstance(first_failover_elapsed, int) else None
reason = trace_reason(events, final_event)
if not matched:
if not matched and not isinstance(native_request, dict):
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:
outcome = "incomplete"
return {
"ok": proc.returncode == 0 and len(matched) > 0,
"ok": proc.returncode == 0 and (len(matched) > 0 or isinstance(native_request, dict)),
"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,
@@ -311,10 +478,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": {
@@ -323,20 +496,35 @@ 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),
},
"diagnostics": {
"failoverBudgetExhausted": failover_budget_exhausted(failovers, final_event or {}),
"untriedSchedulableAccounts": untried_schedulable_accounts,
"timing": {
"referenceAt": reference_at if isinstance(reference_at, str) else (events[0].get("at") if events else None),
"referenceSource": reference_source,
"firstFailoverElapsedMs": first_failover_elapsed,
"firstSelectFailedElapsedMs": first_select_failed_elapsed,
"finalElapsedMs": final_elapsed,
"failoverToFinalMs": failover_to_final,
"clientDeadlineAvailable": False,
"clientDeadlineRemainingMs": None,
"attribution": "Elapsed values come from indexed event timestamps. The client deadline is not present in Sub2API Ops records, so remaining request budget cannot be calculated.",
},
},
"accountSnapshot": account_snapshot,
"accountSnapshotError": account_snapshot_error,
"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,