Codex
346 lines
14 KiB
TypeScript
346 lines
14 KiB
TypeScript
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,
|
|
}
|
|
|
|
`;
|