refactor: split Sub2API remote Python modules
This commit is contained in:
Codex
@@ -0,0 +1,345 @@
|
||||
export const remotePythonTraceScript = `def parse_log_line(line):
|
||||
json_start = line.find("{")
|
||||
if json_start < 0:
|
||||
return None
|
||||
prefix = line[:json_start].rstrip()
|
||||
try:
|
||||
item = json.loads(line[json_start:])
|
||||
except Exception:
|
||||
return None
|
||||
if not isinstance(item, dict):
|
||||
return None
|
||||
at = None
|
||||
parts = prefix.split()
|
||||
if parts:
|
||||
at = parts[0]
|
||||
message = ""
|
||||
if len(parts) >= 4:
|
||||
message = " ".join(parts[3:])
|
||||
elif len(parts) >= 3:
|
||||
message = parts[2]
|
||||
elif len(parts) >= 1:
|
||||
message = parts[-1]
|
||||
item["_at"] = at
|
||||
item["_message"] = message
|
||||
item["_line"] = line
|
||||
return item
|
||||
|
||||
def log_time_epoch(item):
|
||||
at = item.get("_at") if isinstance(item, dict) else None
|
||||
if not isinstance(at, str) or not at:
|
||||
return None
|
||||
try:
|
||||
return datetime.strptime(at, "%Y-%m-%dT%H:%M:%S.%f%z").timestamp()
|
||||
except Exception:
|
||||
try:
|
||||
return datetime.fromisoformat(at.replace("Z", "+00:00")).timestamp()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def event_base(item, account_names_by_id):
|
||||
account_id = item.get("account_id")
|
||||
if isinstance(account_id, str) and account_id.isdigit():
|
||||
account_id = int(account_id)
|
||||
account_name = account_names_by_id.get(account_id)
|
||||
return {
|
||||
"at": item.get("_at"),
|
||||
"message": item.get("_message"),
|
||||
"requestId": item.get("request_id"),
|
||||
"clientRequestId": item.get("client_request_id"),
|
||||
"path": item.get("path"),
|
||||
"method": item.get("method"),
|
||||
"model": item.get("model"),
|
||||
"accountId": account_id,
|
||||
"accountName": account_name,
|
||||
"statusCode": item.get("status_code"),
|
||||
"upstreamStatus": item.get("upstream_status"),
|
||||
"latencyMs": item.get("latency_ms"),
|
||||
}
|
||||
|
||||
def classify_trace_event(item, account_names_by_id):
|
||||
message = str(item.get("_message") or "")
|
||||
event = event_base(item, account_names_by_id)
|
||||
if "content_moderation.gateway_check_start" in message:
|
||||
event.update({
|
||||
"type": "request-start",
|
||||
"stream": item.get("stream"),
|
||||
"bodyBytes": item.get("body_bytes"),
|
||||
"groupId": item.get("group_id"),
|
||||
"groupName": item.get("group_name"),
|
||||
"apiKeyName": item.get("api_key_name"),
|
||||
})
|
||||
elif "content_moderation.gateway_check_done" in message:
|
||||
event.update({
|
||||
"type": "gateway-check",
|
||||
"allowed": item.get("allowed"),
|
||||
"blocked": item.get("blocked"),
|
||||
"action": item.get("action"),
|
||||
})
|
||||
elif "openai.upstream_failover_switching" in message:
|
||||
event.update({
|
||||
"type": "failover",
|
||||
"switchCount": item.get("switch_count"),
|
||||
"maxSwitches": item.get("max_switches"),
|
||||
})
|
||||
elif "openai.account_select_failed" in message:
|
||||
event.update({
|
||||
"type": "select-failed",
|
||||
"error": item.get("error"),
|
||||
"excludedAccountCount": item.get("excluded_account_count"),
|
||||
})
|
||||
elif "account_upstream_error" in message:
|
||||
event.update({
|
||||
"type": "upstream-error",
|
||||
"error": item.get("error"),
|
||||
})
|
||||
elif "account_temp_unschedulable" in message:
|
||||
event.update({
|
||||
"type": "temp-unschedulable",
|
||||
"until": item.get("until") or item.get("temp_unschedulable_until"),
|
||||
"ruleIndex": item.get("rule_index"),
|
||||
"matchedKeyword": item.get("matched_keyword"),
|
||||
"reason": item.get("reason") or item.get("error"),
|
||||
})
|
||||
elif "http request completed" in message:
|
||||
event.update({
|
||||
"type": "final",
|
||||
"clientIp": item.get("client_ip"),
|
||||
"protocol": item.get("protocol"),
|
||||
"platform": item.get("platform"),
|
||||
"completedAt": item.get("completed_at"),
|
||||
})
|
||||
elif "admin account schedulable updated" in message or "account schedulable updated" in message or "/schedulable" in str(item.get("path") or ""):
|
||||
event.update({
|
||||
"type": "admin-schedulable",
|
||||
"schedulable": item.get("schedulable"),
|
||||
})
|
||||
else:
|
||||
event.update({"type": "other"})
|
||||
return event
|
||||
|
||||
def with_trace_phase(event, first_epoch, last_epoch):
|
||||
epoch = None
|
||||
at = event.get("at") if isinstance(event, dict) else None
|
||||
if isinstance(at, str) and at:
|
||||
epoch = log_time_epoch({"_at": at})
|
||||
if epoch is None or first_epoch is None:
|
||||
phase = "unknown"
|
||||
elif epoch < first_epoch:
|
||||
phase = "before-request"
|
||||
elif last_epoch is not None and epoch > last_epoch:
|
||||
phase = "after-request"
|
||||
else:
|
||||
phase = "during-request"
|
||||
event["phase"] = phase
|
||||
return event
|
||||
|
||||
def account_snapshot_from_runtime(token):
|
||||
try:
|
||||
accounts = list_accounts(token)
|
||||
except Exception as exc:
|
||||
return [], {"error": str(exc)}
|
||||
rows = []
|
||||
for item in accounts:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
rows.append({
|
||||
"accountId": item.get("id"),
|
||||
"accountName": item.get("name"),
|
||||
"schedulable": item.get("schedulable"),
|
||||
"status": item.get("status"),
|
||||
"concurrency": item.get("concurrency"),
|
||||
"priority": item.get("priority"),
|
||||
"tempUnschedulableUntil": item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil"),
|
||||
"tempUnschedulableSet": (item.get("temp_unschedulable_until") or item.get("tempUnschedulableUntil")) is not None or bool(item.get("temp_unschedulable_reason") or item.get("tempUnschedulableReason")),
|
||||
})
|
||||
rows.sort(key=lambda row: (str(row.get("accountName") or ""), int(row.get("accountId") or 0)))
|
||||
return rows, None
|
||||
|
||||
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"]
|
||||
if failover_budget_exhausted(failovers, final_event):
|
||||
return "failover-budget-exhausted"
|
||||
if failovers and select_failures:
|
||||
return "failover-attempted-no-candidate"
|
||||
if failovers:
|
||||
return "failover-attempted"
|
||||
if select_failures:
|
||||
return "account-select-failed"
|
||||
if upstream_errors:
|
||||
return "upstream-error"
|
||||
if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int) and final_event.get("statusCode") >= 400:
|
||||
return "final-http-error"
|
||||
if isinstance(final_event, dict) and isinstance(final_event.get("statusCode"), int):
|
||||
return "completed"
|
||||
return "unknown"
|
||||
|
||||
def failover_budget_exhausted(failovers, final_event):
|
||||
if not failovers or not isinstance(final_event, dict):
|
||||
return False
|
||||
last = failovers[-1]
|
||||
switch_count = last.get("switchCount")
|
||||
max_switches = last.get("maxSwitches")
|
||||
final_status = final_event.get("statusCode")
|
||||
return (
|
||||
isinstance(switch_count, int)
|
||||
and isinstance(max_switches, int)
|
||||
and max_switches > 0
|
||||
and switch_count >= max_switches
|
||||
and isinstance(final_status, int)
|
||||
and final_status >= 500
|
||||
)
|
||||
|
||||
def trace_untried_schedulable_accounts(failovers, final_event, account_snapshot):
|
||||
if not failover_budget_exhausted(failovers, final_event):
|
||||
return []
|
||||
tried = set()
|
||||
for item in failovers:
|
||||
account_id = item.get("accountId")
|
||||
if isinstance(account_id, int):
|
||||
tried.add(account_id)
|
||||
final_account = final_event.get("accountId") if isinstance(final_event, dict) else None
|
||||
if isinstance(final_account, int):
|
||||
tried.add(final_account)
|
||||
result = []
|
||||
for item in account_snapshot:
|
||||
account_id = item.get("accountId")
|
||||
if not isinstance(account_id, int) or account_id in tried:
|
||||
continue
|
||||
if item.get("schedulable") is True and item.get("status") == "active" and item.get("tempUnschedulableSet") is not True:
|
||||
result.append({
|
||||
"accountId": account_id,
|
||||
"accountName": item.get("accountName"),
|
||||
"priority": item.get("priority"),
|
||||
"concurrency": item.get("concurrency"),
|
||||
})
|
||||
return result
|
||||
|
||||
def run_trace():
|
||||
payload = json.loads(base64.b64decode(PAYLOAD_B64).decode("utf-8")) if PAYLOAD_B64 else {}
|
||||
request_id = payload.get("requestId")
|
||||
since = payload.get("since") or "24h"
|
||||
tail = int(payload.get("tail") or 20000)
|
||||
context_seconds = int(payload.get("contextSeconds") or 300)
|
||||
show_lines = bool(payload.get("showLines"))
|
||||
if not isinstance(request_id, str) or not request_id:
|
||||
raise RuntimeError("trace payload missing requestId")
|
||||
admin_email, token, admin_compliance = login()
|
||||
account_snapshot, account_snapshot_error = account_snapshot_from_runtime(token)
|
||||
account_names_by_id = {}
|
||||
for row in account_snapshot:
|
||||
account_id = row.get("accountId")
|
||||
if isinstance(account_id, str) and account_id.isdigit():
|
||||
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")
|
||||
parsed_lines = []
|
||||
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)
|
||||
first_epoch = None
|
||||
last_epoch = None
|
||||
for item in matched:
|
||||
epoch = log_time_epoch(item)
|
||||
if epoch is None:
|
||||
continue
|
||||
first_epoch = epoch if first_epoch is None else min(first_epoch, epoch)
|
||||
last_epoch = epoch if last_epoch is None else max(last_epoch, epoch)
|
||||
window_lines = []
|
||||
if first_epoch is not None:
|
||||
start_epoch = first_epoch - context_seconds
|
||||
end_epoch = (last_epoch if last_epoch is not None else first_epoch) + context_seconds
|
||||
for item in parsed_lines:
|
||||
epoch = log_time_epoch(item)
|
||||
if epoch is not None and start_epoch <= epoch <= end_epoch:
|
||||
window_lines.append(item)
|
||||
else:
|
||||
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)
|
||||
final_event = next((item for item in reversed(events) if item.get("type") == "final"), 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"]
|
||||
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"]
|
||||
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"
|
||||
elif isinstance(final_event, dict):
|
||||
outcome = "failed"
|
||||
else:
|
||||
outcome = "incomplete"
|
||||
return {
|
||||
"ok": proc.returncode == 0 and len(matched) > 0,
|
||||
"mode": "trace",
|
||||
"namespace": NAMESPACE,
|
||||
"serviceDns": SERVICE_DNS,
|
||||
"appPod": APP_POD,
|
||||
"admin": {"email": admin_email, "tokenPrinted": False, "compliance": admin_compliance},
|
||||
"requestId": request_id,
|
||||
"summary": {
|
||||
"outcome": outcome,
|
||||
"reason": reason,
|
||||
"eventCount": len(events),
|
||||
"matchedLineCount": len(matched),
|
||||
"firstAt": events[0].get("at") if events else None,
|
||||
"lastAt": events[-1].get("at") if events else None,
|
||||
},
|
||||
"window": {
|
||||
"since": since,
|
||||
"tail": tail,
|
||||
"beforeSeconds": context_seconds,
|
||||
"afterSeconds": context_seconds,
|
||||
"lineCount": len(window_lines),
|
||||
},
|
||||
"request": request_start or {},
|
||||
"final": final_event or {},
|
||||
"events": events,
|
||||
"failovers": failovers,
|
||||
"selectFailures": select_failures,
|
||||
"upstreamErrors": upstream_errors,
|
||||
"tempUnschedulable": temp_unsched,
|
||||
"adminSchedulable": admin_sched[-20:],
|
||||
"windowStats": {
|
||||
"matchedLines": len(matched),
|
||||
"eventCount": len(window_events),
|
||||
"finalErrorCount": len(final_errors),
|
||||
"failoverCount": len(window_failovers),
|
||||
"selectFailedCount": len(window_select_failures),
|
||||
"tempUnschedulableCount": len(temp_unsched),
|
||||
"adminSchedulableCount": len(admin_sched),
|
||||
},
|
||||
"diagnostics": {
|
||||
"failoverBudgetExhausted": failover_budget_exhausted(failovers, final_event or {}),
|
||||
"untriedSchedulableAccounts": untried_schedulable_accounts,
|
||||
},
|
||||
"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": {
|
||||
"exitCode": proc.returncode,
|
||||
"stderrTail": text(proc.stderr, 1000),
|
||||
"stdoutLineCount": len(stdout.splitlines()),
|
||||
},
|
||||
"valuesPrinted": False,
|
||||
}
|
||||
|
||||
`;
|
||||
Reference in New Issue
Block a user