diff --git a/scripts/src/platform-infra-sub2api-ops/projection.ts b/scripts/src/platform-infra-sub2api-ops/projection.ts index c1c22fed..606027c3 100644 --- a/scripts/src/platform-infra-sub2api-ops/projection.ts +++ b/scripts/src/platform-infra-sub2api-ops/projection.ts @@ -4,6 +4,8 @@ const upstreamVersion = "v0.1.155"; const diagnosisRuleSource = "frontend/src/views/admin/ops/components/OpsDashboardHeader.vue"; const diagnosisTextSource = "frontend/src/i18n/locales/zh/admin/ops.ts"; const channelHistoryPageSize = 20; +const monitorResponseHeaderTimeoutMs = 30_000; +const monitorTotalRequestTimeoutMs = 45_000; export function projectSub2ApiOps(raw: Record, options: Sub2ApiOpsOptions): Record { const target = record(raw.target); @@ -25,6 +27,7 @@ export function projectSub2ApiOps(raw: Record, options: Sub2Api } function projectDiagnosis(target: Record, data: Record, options: Sub2ApiOpsOptions): Record { + const runtimeVersion = record(data.runtimeVersion); const groups = arrayRecords(data.groups).map((entry) => { const group = record(entry.group); const overview = record(entry.overview); @@ -53,7 +56,7 @@ function projectDiagnosis(target: Record, data: Record): Record, data: Record, options: Sub2ApiOpsOptions): Record { + const runtimeVersion = record(data.runtimeVersion); const channels = arrayRecords(data.channels).map((entry) => projectChannel(entry, options.window)); const operational = channels.every((channel) => channel.status === "OPERATIONAL"); const history = arrayRecords(data.history) @@ -157,14 +161,15 @@ function projectChannels(target: Record, data: Record String(left.checkedAt).localeCompare(String(right.checkedAt))); - const historyProjection = options.channel === null ? null : paginateHistory(history, options); + const historyProjection = options.channel === null ? null : paginateHistory(history, options, record(data.correlation), runtimeVersion); const historyError = historyProjection === null ? null : historyProjection.error; return { ok: historyError === null, action: "platform-infra-sub2api-ops-channels", target, query: querySummary(options), - source: sourceSummary("channels", "available"), + source: sourceSummary("channels", "available", runtimeVersion), + timeoutPolicy: monitorTimeoutPolicy(runtimeVersion), overallStatus: operational ? "OPERATIONAL" : "DEGRADED", channels, history: historyProjection, @@ -173,9 +178,10 @@ function projectChannels(target: Record, data: Record[], options: Sub2ApiOpsOptions): Record { +function paginateHistory(history: Record[], options: Sub2ApiOpsOptions, correlation: Record, runtimeVersion: Record): Record { if (options.recordId !== null) { - const selectedRecord = history.find((item) => String(item.id) === options.recordId) ?? null; + const selectedRecordRaw = history.find((item) => String(item.id) === options.recordId) ?? null; + const selectedRecord = selectedRecordRaw === null ? null : projectSelectedRecord(selectedRecordRaw, correlation, runtimeVersion); return { order: "PAST_TO_NOW", pageSize: channelHistoryPageSize, @@ -221,6 +227,53 @@ function paginateHistory(history: Record[], options: Sub2ApiOps }; } +function projectSelectedRecord(item: Record, correlation: Record, runtimeVersion: Record): Record { + const latencyMs = typeof item.latencyMs === "number" ? item.latencyMs : null; + const selected = record(correlation.selected); + const final = record(selected.final); + const failovers = arrayRecords(selected.failovers); + const selectFailures = arrayRecords(selected.selectFailures); + const totalLatencyMs = typeof final.latencyMs === "number" ? final.latencyMs : null; + const runtimeTag = typeof runtimeVersion.version === "string" && runtimeVersion.version.length > 0 ? runtimeVersion.version : null; + const runtimeVersionMismatch = runtimeTag?.replace(/^v/u, "") !== upstreamVersion.replace(/^v/u, ""); + return { + ...item, + timeout: { + observedKind: String(item.message ?? "").toLowerCase().includes("timeout awaiting response headers") ? "response-header" : "unknown", + responseHeaderTimeoutMs: monitorResponseHeaderTimeoutMs, + totalRequestTimeoutMs: monitorTotalRequestTimeoutMs, + sourceReferenceVersion: upstreamVersion, + runtimeVersion: runtimeTag, + runtimeVersionMismatch, + runtimePolicyStatus: runtimeVersionMismatch ? "unverified" : "source-version-verified", + remainingAfterResponseHeaderMs: monitorTotalRequestTimeoutMs - monitorResponseHeaderTimeoutMs, + observedLatencyDeltaMs: latencyMs === null ? null : latencyMs - monitorResponseHeaderTimeoutMs, + perUpstreamTimeout: { status: "unsupported", reason: "native monitor history does not expose per-upstream timeout" }, + overallGatewayDeadline: { status: "unsupported", reason: "native monitor history does not expose the gateway request deadline" }, + failoverReserve: { status: "unsupported", reason: "native monitor history does not expose a dedicated failover reserve" }, + }, + correlation: { + status: correlation.status ?? "unsupported", + reason: correlation.reason ?? null, + requestId: selected.requestId ?? null, + match: selected.match ?? null, + apiKeyName: selected.apiKeyName ?? null, + model: selected.model ?? null, + accounts: selected.accounts ?? [], + upstreamErrors: selected.upstreamErrors ?? [], + failovers, + selectFailures, + final: Object.keys(final).length === 0 ? null : final, + switchingObserved: failovers.length > 0, + selectionFailureObserved: selectFailures.length > 0, + sufficientForObservedCompletion: runtimeVersionMismatch || totalLatencyMs === null ? null : monitorTotalRequestTimeoutMs >= totalLatencyMs, + sufficiencyStatus: runtimeVersionMismatch ? "unverified-runtime-version-mismatch" : totalLatencyMs === null ? "unsupported-no-final-latency" : "evaluated-against-verified-source-version", + traceNext: selected.requestId ? `bun scripts/cli.ts platform-infra sub2api codex-pool trace --request-id ${String(selected.requestId)}` : null, + candidates: correlation.status === "ambiguous" ? correlation.candidates ?? [] : [], + }, + }; +} + function projectChannel(entry: Record, window: "7d" | "15d" | "30d"): Record { const summary = record(entry.summary); const detail = record(entry.detail); @@ -261,7 +314,33 @@ function projectChannel(entry: Record, window: "7d" | "15d" | " collectedAt: timeline.length === 0 ? null : timeline[timeline.length - 1]?.checkedAt ?? null, refreshRemainingSeconds: null, refreshSourceStatus: "unsupported-by-native-response", - timeline: { order: "PAST_TO_NOW", records: timeline }, + timeline: { + order: "PAST_TO_NOW", + recordCount: timeline.length, + firstAt: timeline[0]?.checkedAt ?? null, + lastAt: timeline[timeline.length - 1]?.checkedAt ?? null, + records: [], + detail: "use --channel for native history pagination", + }, + }; +} + +function monitorTimeoutPolicy(runtimeVersion: Record): Record { + const runtimeTag = typeof runtimeVersion.version === "string" && runtimeVersion.version.length > 0 ? runtimeVersion.version : null; + const runtimeVersionMismatch = runtimeTag?.replace(/^v/u, "") !== upstreamVersion.replace(/^v/u, ""); + return { + runtime: { image: runtimeVersion.image ?? null, version: runtimeTag }, + sourceReferenceVersion: upstreamVersion, + runtimeVersionMismatch, + runtimePolicyStatus: runtimeVersionMismatch ? "unverified" : "source-version-verified", + sourceReference: { + responseHeaderTimeoutMs: monitorResponseHeaderTimeoutMs, + totalRequestTimeoutMs: monitorTotalRequestTimeoutMs, + sourceStatus: "official-source-constant-verified-for-reference-version", + }, + perUpstreamTimeout: { status: "unsupported", reason: "native monitor API does not expose this field" }, + overallGatewayDeadline: { status: "unsupported", reason: "native monitor API does not expose this field" }, + failoverReserve: { status: "unsupported", reason: "native monitor API does not expose this field" }, }; } @@ -269,11 +348,12 @@ function availabilityFor(model: Record, window: "7d" | "15d" | return model[`availability_${window}`] ?? model.availability ?? null; } -function sourceSummary(action: "diagnosis" | "channels", status: string): Record { +function sourceSummary(action: "diagnosis" | "channels", status: string, runtimeVersion: Record = {}): Record { if (action === "diagnosis") { return { status, - upstreamVersion, + sourceReferenceVersion: upstreamVersion, + runtimeVersion, metricsEndpoint: "/api/v1/admin/ops/dashboard/overview", nativeDiagnosisEndpoint: { status: "unsupported", reason: "v0.1.155 has no diagnosis/advice response schema" }, projectionKind: "native-frontend-projection", @@ -284,10 +364,12 @@ function sourceSummary(action: "diagnosis" | "channels", status: string): Record } return { status, - upstreamVersion, + sourceReferenceVersion: upstreamVersion, + runtimeVersion, listEndpoint: "/api/v1/channel-monitors", detailEndpoint: "/api/v1/channel-monitors/:id/status", historyEndpoint: "/api/v1/admin/channel-monitors/:id/history", + correlationSource: "heuristic projection by native record checked_at, latency_ms and model; request/failover facts come from request rows and bounded runtime logs when available", windows: ["7d", "15d", "30d"], overallStatusKind: "native-frontend-projection", refreshRemaining: { status: "unsupported", reason: "native response does not include a refresh countdown" }, diff --git a/scripts/src/platform-infra-sub2api-ops/query.ts b/scripts/src/platform-infra-sub2api-ops/query.ts index a71e12ef..e5ea7017 100644 --- a/scripts/src/platform-infra-sub2api-ops/query.ts +++ b/scripts/src/platform-infra-sub2api-ops/query.ts @@ -51,6 +51,7 @@ import base64 import json import subprocess import sys +from datetime import datetime from urllib.parse import urlencode TARGET = ${pyJson(target)} @@ -59,6 +60,18 @@ OPTIONS = ${pyJson(options)} def run(command, input_bytes=None): return subprocess.run(command, input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE) +def docker(args): + proc = run(["docker", *args]) + if proc.returncode == 0: + return proc + sudo_proc = run(["sudo", "-n", "docker", *args]) + return sudo_proc if sudo_proc.returncode == 0 else proc + +def runtime_logs(since="24h", tail=20000): + if TARGET["runtimeMode"] == "host-docker": + return docker(["logs", f"--since={since}", f"--tail={tail}", "sub2api-app"]) + return run(["kubectl", "-n", TARGET["namespace"], "logs", "deployment/sub2api", f"--since={since}", f"--tail={tail}"]) + def tail_text(value, limit=800): if isinstance(value, bytes): value = value.decode("utf-8", errors="replace") @@ -103,6 +116,17 @@ def select_app_pod(): APP_POD = select_app_pod() +def runtime_version(): + if TARGET["runtimeMode"] == "host-docker": + proc = docker(["inspect", "--format", "{{.Config.Image}}", "sub2api-app"]) + image = proc.stdout.decode("utf-8", errors="replace").strip() if proc.returncode == 0 else "" + else: + pod = kube_json(["-n", TARGET["namespace"], "get", "pod", APP_POD], "sub2api pod") + containers = ((pod.get("spec") or {}).get("containers") or []) + image = containers[0].get("image") if containers else "" + tag = image.rsplit(":", 1)[1] if isinstance(image, str) and ":" in image else None + return {"image": image or None, "version": tag} + def config_value(key): if TARGET["runtimeMode"] == "host-docker": return read_host_env().get(key) @@ -199,6 +223,116 @@ def find_named(items, selector, label): raise RuntimeError(f"unknown {label} {selector}; available={available}") return matches +def log_item(line): + json_start = line.find("{") + if json_start < 0: + return None + try: + item = json.loads(line[json_start:]) + except Exception: + return None + if not isinstance(item, dict): + return None + prefix = line[:json_start].strip().split() + item["_at"] = prefix[0] if prefix else None + item["_message"] = " ".join(prefix[3:]) if len(prefix) >= 4 else " ".join(prefix[2:]) + return item + +def epoch_of(value): + if not isinstance(value, str) or not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp() + except Exception: + return None + +def account_names(token, platform): + params = {"page": 1, "page_size": 500} + if platform: + params["platform"] = platform + rows = items_of(get(token, "/api/v1/admin/accounts", params, "list accounts for monitor correlation")) + return {item.get("id"): item.get("name") for item in rows if isinstance(item, dict) and item.get("id") is not None} + +def correlate_record(token, record, monitor): + checked_at = record.get("checked_at") + checked_epoch = epoch_of(checked_at) + latency_ms = record.get("latency_ms") + if checked_epoch is None or not isinstance(latency_ms, (int, float)): + return {"status": "unsupported", "reason": "native record has no usable checked_at/latency_ms", "candidates": []} + expected_start = checked_epoch - latency_ms / 1000.0 + record_model = str(record.get("model") or "") + request_rows = items_of(get(token, "/api/v1/admin/ops/requests", {"time_range": "24h", "page": 1, "page_size": 500, "kind": "all"}, "list request correlation candidates")) + request_candidates = [] + for item in request_rows: + request_id = item.get("request_id") + created_epoch = epoch_of(item.get("created_at")) + model_match = not record_model or str(item.get("model") or "") == record_model + if not isinstance(request_id, str) or not request_id or created_epoch is None or not model_match: + continue + start_delta_ms = round(abs(created_epoch - expected_start) * 1000) + if start_delta_ms <= 15000: + request_candidates.append((request_id, item, start_delta_ms)) + proc = runtime_logs() + names = account_names(token, monitor.get("provider")) + grouped = {} + candidate_ids = {item[0] for item in request_candidates} + for line in proc.stdout.decode("utf-8", errors="replace").splitlines() if proc.returncode == 0 else []: + item = log_item(line) + if item is None: + continue + request_id = item.get("request_id") + at_epoch = epoch_of(item.get("_at")) + if not isinstance(request_id, str) or not request_id: + continue + if request_id not in candidate_ids and (at_epoch is None or at_epoch < expected_start - 15 or at_epoch > checked_epoch + 90): + continue + grouped.setdefault(request_id, []).append(item) + candidates = [] + rows_by_id = {item[0]: (item[1], item[2]) for item in request_candidates} + all_request_ids = set(grouped.keys()) | set(rows_by_id.keys()) + for request_id in all_request_ids: + events = grouped.get(request_id, []) + events.sort(key=lambda item: epoch_of(item.get("_at")) or 0) + request_row, row_delta_ms = rows_by_id.get(request_id, ({}, None)) + first = events[0] if events else None + models = [str(item.get("model")) for item in events if item.get("model")] + first_epoch = epoch_of(first.get("_at")) if first is not None else epoch_of(request_row.get("created_at")) + start_delta_ms = row_delta_ms if row_delta_ms is not None else round(abs((first_epoch or expected_start) - expected_start) * 1000) + model_match = not record_model or record_model in models or str(request_row.get("model") or "") == record_model + if start_delta_ms > 15000 or not model_match: + continue + failovers = [item for item in events if "upstream_failover_switching" in str(item.get("_message") or "")] + select_failures = [item for item in events if "account_select_failed" in str(item.get("_message") or "")] + upstream_errors = [item for item in events if "account_upstream_error" in str(item.get("_message") or "")] + finals = [item for item in events if "http request completed" in str(item.get("_message") or "")] + final = finals[-1] if finals else None + account_ids = [] + for item in events: + account_id = item.get("account_id") + if isinstance(account_id, str) and account_id.isdigit(): + account_id = int(account_id) + if isinstance(account_id, int) and account_id not in account_ids: + account_ids.append(account_id) + candidates.append({ + "requestId": request_id, + "match": {"kind": "native-request-start-window", "startDeltaMs": start_delta_ms, "modelMatched": model_match}, + "firstAt": first.get("_at") if first is not None else request_row.get("created_at"), + "lastAt": events[-1].get("_at") if events else request_row.get("created_at"), + "model": next((item for item in models if item), request_row.get("model")), + "apiKeyName": next((item.get("api_key_name") for item in events if item.get("api_key_name")), None), + "accounts": [{"id": account_id, "name": names.get(account_id)} for account_id in account_ids] or ([{"id": request_row.get("account_id"), "name": request_row.get("account_name") or names.get(request_row.get("account_id"))}] if request_row.get("account_id") is not None else []), + "upstreamErrors": [{"at": item.get("_at"), "accountId": item.get("account_id"), "error": item.get("error")} for item in upstream_errors], + "failovers": [{"at": item.get("_at"), "accountId": item.get("account_id"), "upstreamStatus": item.get("upstream_status"), "switchCount": item.get("switch_count"), "maxSwitches": item.get("max_switches")} for item in failovers], + "selectFailures": [{"at": item.get("_at"), "error": item.get("error"), "excludedAccountCount": item.get("excluded_account_count")} for item in select_failures], + "final": {"at": request_row.get("created_at"), "statusCode": request_row.get("status_code") or request_row.get("upstream_status_code"), "accountId": request_row.get("account_id"), "latencyMs": request_row.get("duration_ms"), "path": request_row.get("path")} if final is None else {"at": final.get("_at"), "statusCode": final.get("status_code"), "accountId": final.get("account_id"), "latencyMs": final.get("latency_ms"), "path": final.get("path")}, + }) + candidates.sort(key=lambda item: item["match"]["startDeltaMs"]) + if not candidates: + return {"status": "unsupported", "reason": "no unique native request_id is available in the record; bounded log correlation found no candidate", "candidates": []} + best_delta = candidates[0]["match"]["startDeltaMs"] + best = [item for item in candidates if item["match"]["startDeltaMs"] == best_delta] + return {"status": "inferred" if len(best) == 1 else "ambiguous", "reason": "heuristic correlation by record start window and model; the native monitor record has no request_id" if len(best) == 1 else "multiple requests have the same nearest start time", "selected": best[0] if len(best) == 1 else None, "candidates": candidates[:5]} + def diagnosis(token): group_params = {"platform": OPTIONS["platform"]} if OPTIONS.get("platform") else None groups = items_of(get(token, "/api/v1/admin/groups/all", group_params, "list groups")) @@ -247,17 +381,24 @@ def channels(token): detail = get(token, f"/api/v1/channel-monitors/{monitor['id']}/status", None, "channel monitor status") output.append({"summary": monitor, "detail": detail}) history = None + correlation = None if OPTIONS.get("channel") and output: monitor_id = output[0]["summary"]["id"] params = {"limit": 1000} if OPTIONS.get("model"): params["model"] = OPTIONS["model"] history = items_of(get(token, f"/api/v1/admin/channel-monitors/{monitor_id}/history", params, "channel monitor history")) - return {"channels": output, "history": history} + record_id = OPTIONS.get("recordId") + if record_id: + selected = next((item for item in history if str(item.get("id")) == str(record_id)), None) + if selected is not None: + correlation = correlate_record(token, selected, output[0]["summary"]) + return {"channels": output, "history": history, "correlation": correlation} try: token = login() data = diagnosis(token) if OPTIONS["action"] == "diagnosis" else channels(token) + data["runtimeVersion"] = runtime_version() print(json.dumps({"ok": True, "sourceStatus": "available", "data": data}, ensure_ascii=False, separators=(",", ":"))) except Exception as error: print(json.dumps({"ok": False, "sourceStatus": "unavailable", "error": str(error), "data": None}, ensure_ascii=False, separators=(",", ":"))) diff --git a/scripts/src/platform-infra-sub2api-ops/render.ts b/scripts/src/platform-infra-sub2api-ops/render.ts index bf583153..c2d8fed9 100644 --- a/scripts/src/platform-infra-sub2api-ops/render.ts +++ b/scripts/src/platform-infra-sub2api-ops/render.ts @@ -125,6 +125,12 @@ function renderChannels(result: Record): string[] { } const selectedRecord = record(history.selectedRecord); if (Object.keys(selectedRecord).length > 0) { + const timeout = record(selectedRecord.timeout); + const correlation = record(selectedRecord.correlation); + const final = record(correlation.final); + const accounts = arrayRecords(correlation.accounts); + const failovers = arrayRecords(correlation.failovers); + const selectFailures = arrayRecords(correlation.selectFailures); lines.push( "", "RECORD", @@ -136,8 +142,25 @@ function renderChannels(result: Record): string[] { ["pingLatencyMs", stringValue(selectedRecord.pingLatencyMs)], ["checkedAt", stringValue(selectedRecord.checkedAt)], ["message", stringValue(selectedRecord.message)], + ["runtimeVersion", stringValue(timeout.runtimeVersion)], + ["sourceReferenceVersion", stringValue(timeout.sourceReferenceVersion)], + ["runtimePolicyStatus", stringValue(timeout.runtimePolicyStatus)], + ["sourceResponseHeaderMs", stringValue(timeout.responseHeaderTimeoutMs)], + ["sourceTotalRequestMs", stringValue(timeout.totalRequestTimeoutMs)], + ["remainingAfterHeaderMs", stringValue(timeout.remainingAfterResponseHeaderMs)], + ["correlationStatus", stringValue(correlation.status)], + ["requestId", stringValue(correlation.requestId)], + ["accounts", accounts.map((item) => `${stringValue(item.id)}:${stringValue(item.name)}`).join(",") || "-"], + ["failoverCount", String(failovers.length)], + ["selectFailureCount", String(selectFailures.length)], + ["finalStatus", stringValue(final.statusCode)], + ["finalLatencyMs", stringValue(final.latencyMs)], + ["timeoutSufficient", stringValue(correlation.sufficientForObservedCompletion)], + ["sufficiencyStatus", stringValue(correlation.sufficiencyStatus)], ]), ); + if (correlation.traceNext) lines.push(`Trace: ${stringValue(correlation.traceNext)}`); + if (correlation.reason) lines.push(`Correlation: ${stringValue(correlation.reason)}`); } lines.push( "",