Codex
1272 lines
61 KiB
TypeScript
1272 lines
61 KiB
TypeScript
export interface RuntimeRemoteTarget {
|
|
runtimeMode: string;
|
|
namespace: string;
|
|
hostDockerAppPort: number | null;
|
|
hostDockerEnvPath: string | null;
|
|
appSecretName: string;
|
|
}
|
|
|
|
function pyJson(value: unknown): string {
|
|
return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
|
|
}
|
|
|
|
export function runtimeRemoteScriptHead(payload: unknown, target: RuntimeRemoteTarget): string {
|
|
return `
|
|
set -u
|
|
python3 - <<'PY'
|
|
import json
|
|
import re
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from urllib.parse import quote, urlsplit
|
|
|
|
RUNTIME_MODE = ${pyJson(target.runtimeMode)}
|
|
NAMESPACE = ${pyJson(target.namespace)}
|
|
HOST_DOCKER_APP_PORT = ${pyJson(target.hostDockerAppPort)}
|
|
HOST_DOCKER_ENV_PATH = ${pyJson(target.hostDockerEnvPath)}
|
|
HOST_DOCKER_APP_CONTAINER = "sub2api-app"
|
|
APP_SECRET_NAME = ${pyJson(target.appSecretName)}
|
|
PAYLOAD = ${pyJson(payload)}
|
|
|
|
def run(cmd, input_bytes=None):
|
|
return subprocess.run(cmd, input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
|
|
|
|
def text(value, limit=1000):
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8", errors="replace")
|
|
return str(value)[-limit:]
|
|
|
|
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 kubectl(args, input_bytes=None):
|
|
return run(["kubectl", *args], input_bytes)
|
|
|
|
def kube_json(args, label):
|
|
proc = kubectl([*args, "-o", "json"])
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(f"{label} failed: {text(proc.stderr)}")
|
|
return json.loads(proc.stdout.decode("utf-8"))
|
|
|
|
def read_host_env():
|
|
try:
|
|
with open(HOST_DOCKER_ENV_PATH, "r", encoding="utf-8") as handle:
|
|
lines = handle.read().splitlines()
|
|
except PermissionError:
|
|
proc = run(["sudo", "-n", "cat", HOST_DOCKER_ENV_PATH])
|
|
if proc.returncode != 0:
|
|
raise RuntimeError("read host env failed: " + text(proc.stderr))
|
|
lines = proc.stdout.decode("utf-8", errors="replace").splitlines()
|
|
values = {}
|
|
for line in lines:
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#") or "=" not in stripped:
|
|
continue
|
|
key, value = stripped.split("=", 1)
|
|
value = value.strip()
|
|
if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
|
|
value = value[1:-1]
|
|
values[key.strip()] = value
|
|
return values
|
|
|
|
def select_app_pod():
|
|
if RUNTIME_MODE == "host-docker":
|
|
return HOST_DOCKER_APP_CONTAINER
|
|
pods = kube_json(["-n", NAMESPACE, "get", "pods", "-l", "app.kubernetes.io/name=sub2api"], "sub2api pods").get("items") or []
|
|
running = [item for item in pods if (item.get("status") or {}).get("phase") == "Running"]
|
|
if not running:
|
|
raise RuntimeError("sub2api app pod not found")
|
|
return running[0]["metadata"]["name"]
|
|
|
|
APP_POD = select_app_pod()
|
|
|
|
def config_value(name, key, fallback=None):
|
|
if RUNTIME_MODE == "host-docker":
|
|
return read_host_env().get(key) or fallback
|
|
data = kube_json(["-n", NAMESPACE, "get", "configmap", name], f"configmap/{name}").get("data") or {}
|
|
return data.get(key) or fallback
|
|
|
|
def secret_value(name, key):
|
|
if RUNTIME_MODE == "host-docker":
|
|
return read_host_env().get(key)
|
|
import base64
|
|
data = kube_json(["-n", NAMESPACE, "get", "secret", name], f"secret/{name}").get("data") or {}
|
|
return base64.b64decode(data[key]).decode("utf-8") if key in data else None
|
|
|
|
def parse_curl(proc):
|
|
output = proc.stdout.decode("utf-8", errors="replace")
|
|
marker = "\\n__HTTP_CODE__:"
|
|
pos = output.rfind(marker)
|
|
if pos < 0:
|
|
return {"ok": False, "httpStatus": 0, "json": None, "body": output, "stderr": text(proc.stderr)}
|
|
body = output[:pos]
|
|
try:
|
|
status = int(output[pos + len(marker):].strip()[-3:])
|
|
except ValueError:
|
|
status = 0
|
|
try:
|
|
parsed = json.loads(body) if body.strip() else None
|
|
except json.JSONDecodeError:
|
|
parsed = None
|
|
return {"ok": proc.returncode == 0 and 200 <= status < 300, "httpStatus": status, "json": parsed, "body": body, "stderr": text(proc.stderr)}
|
|
|
|
def curl_api(method, path, bearer=None, payload=None):
|
|
body = b"" if payload is None else json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
script = r'''
|
|
set -eu
|
|
method="$1"
|
|
url="$2"
|
|
token="\${3:-}"
|
|
tmp="$(mktemp)"
|
|
trap 'rm -f "$tmp"' EXIT
|
|
cat > "$tmp"
|
|
if [ -n "$token" ] && [ "$method" = GET ] && [ ! -s "$tmp" ]; then
|
|
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H "Authorization: Bearer $token" "$url"
|
|
elif [ -n "$token" ]; then
|
|
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' -H "Authorization: Bearer $token" --data-binary @"$tmp" "$url"
|
|
elif [ "$method" = GET ] && [ ! -s "$tmp" ]; then
|
|
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" "$url"
|
|
else
|
|
curl -sS -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' --data-binary @"$tmp" "$url"
|
|
fi
|
|
'''
|
|
if RUNTIME_MODE == "host-docker":
|
|
proc = run(["sh", "-c", script, "sh", method, f"http://127.0.0.1:{HOST_DOCKER_APP_PORT}{path}", bearer or ""], body)
|
|
else:
|
|
proc = run(["kubectl", "-n", NAMESPACE, "exec", "-i", APP_POD, "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or ""], body)
|
|
return parse_curl(proc)
|
|
|
|
def data_of(response, label):
|
|
parsed = response.get("json")
|
|
code = parsed.get("code") if isinstance(parsed, dict) else None
|
|
if response.get("ok") is not True or (code is not None and code != 0):
|
|
message = parsed.get("message") if isinstance(parsed, dict) else text(response.get("body"), 500)
|
|
raise RuntimeError(f"{label} failed: http={response.get('httpStatus')} message={message}")
|
|
return parsed.get("data") if isinstance(parsed, dict) and "data" in parsed else parsed
|
|
|
|
def items_of(data):
|
|
if isinstance(data, list):
|
|
return data
|
|
if isinstance(data, dict):
|
|
for key in ("items", "groups", "accounts"):
|
|
if isinstance(data.get(key), list):
|
|
return data[key]
|
|
return []
|
|
|
|
def login():
|
|
email = config_value("sub2api-config", "ADMIN_EMAIL", PAYLOAD["adminEmailDefault"])
|
|
password = secret_value(APP_SECRET_NAME, "ADMIN_PASSWORD")
|
|
if not password:
|
|
raise RuntimeError("ADMIN_PASSWORD missing")
|
|
data = data_of(curl_api("POST", "/api/v1/auth/login", payload={"email": email, "password": password}), "admin login")
|
|
token = None
|
|
if isinstance(data, dict):
|
|
token = data.get("access_token") or data.get("token")
|
|
if not token:
|
|
raise RuntimeError("admin login response has no token")
|
|
return token
|
|
|
|
def list_groups(token, platform=None):
|
|
suffix = "?platform=" + quote(str(platform), safe="") if platform else ""
|
|
return items_of(data_of(curl_api("GET", "/api/v1/admin/groups/all" + suffix, bearer=token), "list groups"))
|
|
|
|
def list_group_accounts(token, group_id, platform=None):
|
|
path = f"/api/v1/admin/accounts?group_id={group_id}&page=1&page_size=500"
|
|
if platform:
|
|
path += "&platform=" + quote(str(platform), safe="")
|
|
return items_of(data_of(curl_api("GET", path, bearer=token), "list group accounts"))
|
|
|
|
def account_detail(token, account_id):
|
|
return data_of(curl_api("GET", f"/api/v1/admin/accounts/{account_id}", bearer=token), "get account")
|
|
|
|
def normalize_policy(credentials):
|
|
if not isinstance(credentials, dict):
|
|
credentials = {}
|
|
enabled = credentials.get("temp_unschedulable_enabled") is True
|
|
raw = credentials.get("temp_unschedulable_rules")
|
|
if isinstance(raw, str):
|
|
try:
|
|
raw = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
raw = []
|
|
rules = raw if isinstance(raw, list) else []
|
|
codes = sorted(set(item.get("error_code") for item in rules if isinstance(item, dict) and isinstance(item.get("error_code"), int)))
|
|
return {"enabled": enabled, "ruleCount": len(rules), "statusCodes": codes, "rules": rules}
|
|
|
|
def template_summary(template):
|
|
policy = normalize_policy(template.get("credentials"))
|
|
return {"id": template.get("id"), "kind": template.get("kind"), "description": template.get("description"), **{key: policy[key] for key in ("enabled", "ruleCount", "statusCodes")}}
|
|
|
|
def matching_template(credentials):
|
|
current = normalize_policy(credentials)
|
|
for template in PAYLOAD["templates"]:
|
|
if normalize_policy(template.get("credentials")) == current:
|
|
return template.get("id")
|
|
return None
|
|
|
|
def quota_percent(value):
|
|
if not isinstance(value, (int, float)):
|
|
return None
|
|
return round(value * 100, 4) if 0 <= value <= 1 else round(value, 4)
|
|
|
|
def cached_codex_window(extra, label):
|
|
return {
|
|
"window": label,
|
|
"usedPercent": extra.get(f"codex_{label}_used_percent"),
|
|
"resetAfterSeconds": extra.get(f"codex_{label}_reset_after_seconds"),
|
|
"windowMinutes": extra.get(f"codex_{label}_window_minutes"),
|
|
"resetAt": extra.get(f"codex_{label}_reset_at"),
|
|
"source": "sub2api-account-extra",
|
|
}
|
|
|
|
def unix_time_text(value):
|
|
if not isinstance(value, (int, float)) or value <= 0:
|
|
return value
|
|
try:
|
|
return datetime.fromtimestamp(value, tz=timezone.utc).isoformat()
|
|
except (ValueError, OSError, OverflowError):
|
|
return value
|
|
|
|
def openai_quota_auto_pause_settings(token):
|
|
try:
|
|
advanced = data_of(curl_api("GET", "/api/v1/admin/ops/advanced-settings", bearer=token), "get advanced settings")
|
|
settings = advanced.get("openai_account_quota_auto_pause") if isinstance(advanced, dict) else None
|
|
return {"status": "available", "settings": settings if isinstance(settings, dict) else {}}
|
|
except Exception as exc:
|
|
return {"status": "unavailable", "settings": {}, "reason": compact_error(str(exc))}
|
|
|
|
def openai_quota_summary(token, detail):
|
|
extra = detail.get("extra") if isinstance(detail.get("extra"), dict) else {}
|
|
global_result = openai_quota_auto_pause_settings(token)
|
|
global_settings = global_result.get("settings") or {}
|
|
cached_windows = [cached_codex_window(extra, label) for label in ("5h", "7d")]
|
|
probe = {"status": "unsupported", "source": "sub2api-native-openai-quota", "windows": []}
|
|
if detail.get("platform") == "openai" or detail.get("type") in ("oauth", "setup-token"):
|
|
try:
|
|
usage = data_of(curl_api("GET", f"/api/v1/admin/openai/accounts/{detail.get('id')}/quota", bearer=token), "query OpenAI quota")
|
|
rate_limit = usage.get("rate_limit") if isinstance(usage, dict) else None
|
|
rows = []
|
|
if isinstance(rate_limit, dict):
|
|
for name in ("primary_window", "secondary_window"):
|
|
item = rate_limit.get(name)
|
|
if not isinstance(item, dict):
|
|
continue
|
|
seconds = item.get("limit_window_seconds")
|
|
label = "5h" if seconds == 18000 else "7d" if seconds == 604800 else name.replace("_window", "")
|
|
rows.append({
|
|
"window": label,
|
|
"usedPercent": item.get("used_percent"),
|
|
"resetAfterSeconds": item.get("reset_after_seconds"),
|
|
"resetAt": unix_time_text(item.get("reset_at")),
|
|
"source": "sub2api-native-openai-quota",
|
|
})
|
|
probe = {
|
|
"status": "available",
|
|
"source": "sub2api-native-openai-quota",
|
|
"allowed": rate_limit.get("allowed") if isinstance(rate_limit, dict) else None,
|
|
"limitReached": rate_limit.get("limit_reached") if isinstance(rate_limit, dict) else None,
|
|
"fetchedAt": unix_time_text(usage.get("fetched_at")) if isinstance(usage, dict) else None,
|
|
"windows": rows,
|
|
}
|
|
except Exception as exc:
|
|
probe = {"status": "unavailable", "source": "sub2api-native-openai-quota", "reason": compact_error(str(exc)), "windows": []}
|
|
probed_by_window = {item.get("window"): item for item in probe.get("windows", []) if isinstance(item, dict)}
|
|
windows = [probed_by_window.get(item.get("window"), item) for item in cached_windows]
|
|
protections = []
|
|
for window in windows:
|
|
label = window.get("window")
|
|
account_threshold = extra.get(f"auto_pause_{label}_threshold")
|
|
global_threshold = global_settings.get(f"default_threshold_{label}")
|
|
if isinstance(account_threshold, (int, float)) and account_threshold > 0:
|
|
threshold = account_threshold
|
|
source = "account-extra"
|
|
elif isinstance(global_threshold, (int, float)) and global_threshold > 0:
|
|
threshold = global_threshold
|
|
source = "global-ops-default"
|
|
elif global_result.get("status") == "available":
|
|
threshold = None
|
|
source = "disabled"
|
|
else:
|
|
threshold = None
|
|
source = "global-settings-unavailable"
|
|
used = window.get("usedPercent")
|
|
threshold_percent = quota_percent(threshold)
|
|
protections.append({
|
|
"window": label,
|
|
"thresholdPercent": threshold_percent,
|
|
"source": source,
|
|
"triggered": isinstance(used, (int, float)) and isinstance(threshold_percent, (int, float)) and used >= threshold_percent,
|
|
})
|
|
return {
|
|
"windows": windows,
|
|
"cachedWindows": cached_windows,
|
|
"protections": protections,
|
|
"nativeProbe": probe,
|
|
"globalSettings": {"status": global_result.get("status"), "reason": global_result.get("reason")},
|
|
"updatedAt": extra.get("codex_usage_updated_at"),
|
|
"attribution": "Window values prefer the native read-only quota probe and fall back to account cache. Protection uses account threshold first, then the global Ops default.",
|
|
}
|
|
|
|
def account_summary(detail, include_rules=False, token=None):
|
|
credentials = detail.get("credentials") if isinstance(detail.get("credentials"), dict) else {}
|
|
policy = normalize_policy(credentials)
|
|
name = detail.get("name")
|
|
management = "yaml-managed" if name in PAYLOAD["managedAccountNames"] else ("yaml-protected-manual" if name in PAYLOAD["protectedManualAccountNames"] else "runtime-manual")
|
|
result = {
|
|
"accountName": name,
|
|
"accountId": detail.get("id"),
|
|
"management": management,
|
|
"type": detail.get("type"),
|
|
"status": detail.get("status"),
|
|
"schedulable": detail.get("schedulable"),
|
|
"proxyId": detail.get("proxy_id"),
|
|
"concurrencyLimit": detail.get("concurrency"),
|
|
"currentConcurrency": detail.get("current_concurrency"),
|
|
"loadFactor": detail.get("load_factor"),
|
|
"priority": detail.get("priority"),
|
|
"createdAt": detail.get("created_at"),
|
|
"updatedAt": detail.get("updated_at"),
|
|
"prioritySemantics": {"order": "lower-is-higher", "highest": 0},
|
|
"tempUnschedulable": {key: policy[key] for key in ("enabled", "ruleCount", "statusCodes")},
|
|
"matchingTemplate": matching_template(credentials),
|
|
"poolModeEnabled": credentials.get("pool_mode") is True,
|
|
"credentialsPrinted": False,
|
|
}
|
|
if include_rules:
|
|
result["tempUnschedulable"]["rules"] = policy["rules"]
|
|
result["upstreamLimits"] = openai_quota_summary(token, detail) if token else None
|
|
return result
|
|
|
|
def list_proxies(token):
|
|
return items_of(data_of(curl_api("GET", "/api/v1/admin/proxies?page=1&page_size=500", bearer=token), "list proxies"))
|
|
|
|
def proxy_summary(token, proxy_id):
|
|
if proxy_id is None:
|
|
return {"configured": False, "route": "direct", "credentialsPrinted": False}
|
|
matches = [item for item in list_proxies(token) if isinstance(item, dict) and item.get("id") == proxy_id]
|
|
if len(matches) != 1:
|
|
return {"configured": True, "id": proxy_id, "status": "record-not-found", "credentialsPrinted": False}
|
|
item = matches[0]
|
|
return {
|
|
"configured": True,
|
|
"route": "account-proxy",
|
|
"id": item.get("id"),
|
|
"name": item.get("name"),
|
|
"protocol": item.get("protocol"),
|
|
"host": item.get("host"),
|
|
"port": item.get("port"),
|
|
"status": item.get("status"),
|
|
"fallbackMode": item.get("fallback_mode"),
|
|
"lastCheckAt": item.get("last_check_at"),
|
|
"lastError": text(item.get("last_error"), 240) if item.get("last_error") else None,
|
|
"credentialsPrinted": False,
|
|
}
|
|
|
|
def proxy_environment_summary():
|
|
rows = []
|
|
for name in ("HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "http_proxy", "https_proxy", "all_proxy"):
|
|
value = app_text(["sh", "-c", f"printenv {name} 2>/dev/null || true"])
|
|
parsed = urlsplit(value) if value else None
|
|
rows.append({
|
|
"name": name,
|
|
"present": bool(value),
|
|
"scheme": parsed.scheme if parsed else None,
|
|
"host": parsed.hostname if parsed else None,
|
|
"port": parsed.port if parsed else None,
|
|
"credentialsRedacted": bool(parsed and (parsed.username or parsed.password)),
|
|
})
|
|
no_proxy = app_text(["sh", "-c", "printenv NO_PROXY 2>/dev/null || printenv no_proxy 2>/dev/null || true"])
|
|
return rows, {
|
|
"present": bool(no_proxy),
|
|
"entryCount": len([item for item in no_proxy.split(",") if item.strip()]),
|
|
"hyueapiPresent": any(item.strip() in ("hyueapi.com", ".hyueapi.com") for item in no_proxy.split(",")),
|
|
"valuesPrinted": False,
|
|
}
|
|
|
|
def app_runtime_summary():
|
|
if RUNTIME_MODE == "host-docker":
|
|
proc = docker(["inspect", APP_POD])
|
|
if proc.returncode != 0:
|
|
return {"runtimeMode": RUNTIME_MODE, "name": APP_POD, "status": "unavailable", "reason": text(proc.stderr, 240)}
|
|
values = json.loads(proc.stdout.decode("utf-8"))
|
|
item = values[0] if values else {}
|
|
state = item.get("State") or {}
|
|
host_config = item.get("HostConfig") or {}
|
|
networks = ((item.get("NetworkSettings") or {}).get("Networks") or {})
|
|
return {
|
|
"runtimeMode": RUNTIME_MODE,
|
|
"name": APP_POD,
|
|
"state": state.get("Status"),
|
|
"health": ((state.get("Health") or {}).get("Status")),
|
|
"restartCount": item.get("RestartCount"),
|
|
"oomKilled": state.get("OOMKilled"),
|
|
"startedAt": state.get("StartedAt"),
|
|
"networkMode": host_config.get("NetworkMode"),
|
|
"networks": [
|
|
{"name": name, "ipAddress": value.get("IPAddress"), "gateway": value.get("Gateway")}
|
|
for name, value in sorted(networks.items()) if isinstance(value, dict)
|
|
],
|
|
}
|
|
pod = kube_json(["-n", NAMESPACE, "get", "pod", APP_POD], "sub2api pod")
|
|
spec = pod.get("spec") or {}
|
|
status = pod.get("status") or {}
|
|
restarts = sum(item.get("restartCount") or 0 for item in status.get("containerStatuses") or [] if isinstance(item, dict))
|
|
return {
|
|
"runtimeMode": RUNTIME_MODE,
|
|
"name": APP_POD,
|
|
"state": status.get("phase"),
|
|
"restartCount": restarts,
|
|
"oomKilled": None,
|
|
"startedAt": status.get("startTime"),
|
|
"networkMode": "host" if spec.get("hostNetwork") is True else "pod",
|
|
"nodeName": spec.get("nodeName"),
|
|
"podIp": status.get("podIP"),
|
|
"dnsPolicy": spec.get("dnsPolicy"),
|
|
}
|
|
|
|
def dns_summary():
|
|
content = app_text(["cat", "/etc/resolv.conf"])
|
|
nameservers = []
|
|
search = []
|
|
options = []
|
|
for line in content.splitlines():
|
|
fields = line.strip().split()
|
|
if not fields or fields[0].startswith("#"):
|
|
continue
|
|
if fields[0] == "nameserver": nameservers.extend(fields[1:])
|
|
elif fields[0] == "search": search.extend(fields[1:])
|
|
elif fields[0] == "options": options.extend(fields[1:])
|
|
return {"nameservers": nameservers, "search": search, "options": options}
|
|
|
|
def default_route_summary():
|
|
content = app_text(["cat", "/proc/net/route"])
|
|
for line in content.splitlines()[1:]:
|
|
fields = line.split()
|
|
if len(fields) < 4 or fields[1] != "00000000":
|
|
continue
|
|
try:
|
|
gateway = socket.inet_ntoa(struct.pack("<I", int(fields[2], 16)))
|
|
except (ValueError, OSError):
|
|
gateway = None
|
|
return {"interface": fields[0], "gateway": gateway, "flags": fields[3]}
|
|
return {"interface": None, "gateway": None, "status": "default-route-not-found"}
|
|
|
|
def interface_summaries():
|
|
script = 'for path in /sys/class/net/*; do [ -d "$path/statistics" ] || continue; name="$(basename "$path")"; printf "%s" "$name"; for key in rx_bytes tx_bytes rx_errors tx_errors rx_dropped tx_dropped; do value="$(cat "$path/statistics/$key" 2>/dev/null || printf 0)"; printf "\\t%s" "$value"; done; printf "\\n"; done'
|
|
rows = []
|
|
for line in app_text(["sh", "-c", script]).splitlines():
|
|
fields = line.split("\t")
|
|
if len(fields) != 7:
|
|
continue
|
|
values = []
|
|
for value in fields[1:]:
|
|
try:
|
|
values.append(int(value))
|
|
except ValueError:
|
|
values.append(None)
|
|
rows.append({
|
|
"name": fields[0], "rxBytes": values[0], "txBytes": values[1],
|
|
"rxErrors": values[2], "txErrors": values[3], "rxDropped": values[4], "txDropped": values[5],
|
|
})
|
|
return rows
|
|
|
|
def proxy_socket_summary(proxy):
|
|
host = proxy.get("host") if isinstance(proxy, dict) else None
|
|
port = proxy.get("port") if isinstance(proxy, dict) else None
|
|
if not isinstance(host, str) or not isinstance(port, int):
|
|
return {"status": "not-applicable", "resolvedIps": [], "total": 0, "established": 0}
|
|
try:
|
|
resolved = sorted(set(item[4][0] for item in socket.getaddrinfo(host, port, socket.AF_INET, socket.SOCK_STREAM)))
|
|
resolution_status = "resolved"
|
|
except OSError as exc:
|
|
resolved = []
|
|
resolution_status = "failed: " + text(exc, 160)
|
|
state_names = {"01": "established", "02": "syn-sent", "03": "syn-recv", "04": "fin-wait-1", "05": "fin-wait-2", "06": "time-wait", "08": "close-wait", "09": "last-ack", "0A": "listen", "0B": "closing"}
|
|
states = {}
|
|
content = app_text(["cat", "/proc/net/tcp"])
|
|
for line in content.splitlines()[1:]:
|
|
fields = line.split()
|
|
if len(fields) < 4 or ":" not in fields[2]:
|
|
continue
|
|
address_hex, port_hex = fields[2].split(":", 1)
|
|
try:
|
|
address = socket.inet_ntoa(struct.pack("<I", int(address_hex, 16)))
|
|
remote_port = int(port_hex, 16)
|
|
except (ValueError, OSError):
|
|
continue
|
|
if address not in resolved or remote_port != port:
|
|
continue
|
|
state = state_names.get(fields[3], fields[3])
|
|
states[state] = states.get(state, 0) + 1
|
|
return {
|
|
"status": resolution_status,
|
|
"resolvedIps": resolved,
|
|
"port": port,
|
|
"total": sum(states.values()),
|
|
"established": states.get("established", 0),
|
|
"states": [{"state": key, "count": value} for key, value in sorted(states.items())],
|
|
"attribution": "Socket counts are a query-time snapshot. DNS resolution is performed without opening an application connection.",
|
|
}
|
|
|
|
def proxy_endpoint_owner(proxy):
|
|
host = proxy.get("host") if isinstance(proxy, dict) else None
|
|
port = proxy.get("port") if isinstance(proxy, dict) else None
|
|
if RUNTIME_MODE != "host-docker" or host not in ("127.0.0.1", "localhost", "::1") or not isinstance(port, int):
|
|
return {"status": "not-applicable", "listeners": []}
|
|
proc = run(["ss", "-ltnpH"])
|
|
if proc.returncode != 0:
|
|
return {"status": "unavailable", "reason": text(proc.stderr, 240), "listeners": []}
|
|
rows = []
|
|
for line in proc.stdout.decode("utf-8", errors="replace").splitlines():
|
|
fields = line.split()
|
|
if len(fields) < 4:
|
|
continue
|
|
local_endpoint = fields[3]
|
|
try:
|
|
local_port = int(local_endpoint.rsplit(":", 1)[1])
|
|
except (ValueError, IndexError):
|
|
continue
|
|
if local_port != port:
|
|
continue
|
|
process_match = re.search(r'\\("([^\"]+)",pid=(\\d+)', line)
|
|
process_name = process_match.group(1) if process_match else None
|
|
pid = int(process_match.group(2)) if process_match else None
|
|
executable = None
|
|
peers = []
|
|
udp_peers = []
|
|
systemd_unit = None
|
|
active_state = None
|
|
service_restarts = None
|
|
if isinstance(pid, int):
|
|
exe_proc = run(["readlink", f"/proc/{pid}/exe"])
|
|
if exe_proc.returncode == 0:
|
|
executable = exe_proc.stdout.decode("utf-8", errors="replace").strip()
|
|
peer_proc = run(["ss", "-tnpH"])
|
|
if peer_proc.returncode == 0:
|
|
for peer_line in peer_proc.stdout.decode("utf-8", errors="replace").splitlines():
|
|
if f"pid={pid}," not in peer_line:
|
|
continue
|
|
peer_fields = peer_line.split()
|
|
if len(peer_fields) >= 5 and peer_fields[0] == "ESTAB":
|
|
peers.append(peer_fields[4])
|
|
udp_proc = run(["ss", "-unpH"])
|
|
if udp_proc.returncode == 0:
|
|
for udp_line in udp_proc.stdout.decode("utf-8", errors="replace").splitlines():
|
|
if f"pid={pid}," not in udp_line:
|
|
continue
|
|
udp_fields = udp_line.split()
|
|
if len(udp_fields) >= 5 and udp_fields[4] not in ("*:*", "0.0.0.0:*"):
|
|
udp_peers.append(udp_fields[4])
|
|
cgroup_proc = run(["cat", f"/proc/{pid}/cgroup"])
|
|
cgroup_text = cgroup_proc.stdout.decode("utf-8", errors="replace") if cgroup_proc.returncode == 0 else ""
|
|
unit_match = re.search(r'/([^/]+\\.service)(?:/|$)', cgroup_text)
|
|
systemd_unit = unit_match.group(1) if unit_match else None
|
|
if systemd_unit:
|
|
show_proc = run(["systemctl", "show", systemd_unit, "--property=ActiveState", "--property=SubState", "--property=NRestarts"])
|
|
if show_proc.returncode == 0:
|
|
service_values = dict(line.split("=", 1) for line in show_proc.stdout.decode("utf-8", errors="replace").splitlines() if "=" in line)
|
|
active_state = "/".join(value for value in (service_values.get("ActiveState"), service_values.get("SubState")) if value)
|
|
service_restarts = service_values.get("NRestarts")
|
|
rows.append({
|
|
"localEndpoint": local_endpoint,
|
|
"processName": process_name,
|
|
"pid": pid,
|
|
"executable": executable,
|
|
"establishedPeers": sorted(set(peers))[:12],
|
|
"udpPeers": sorted(set(udp_peers))[:12],
|
|
"systemdUnit": systemd_unit,
|
|
"activeState": active_state,
|
|
"serviceRestarts": service_restarts,
|
|
})
|
|
return {"status": "available", "listeners": rows, "credentialsPrinted": False}
|
|
|
|
def proxy_service_evidence(owner):
|
|
listeners = owner.get("listeners") if isinstance(owner, dict) else []
|
|
units = sorted(set(item.get("systemdUnit") for item in listeners if isinstance(item, dict) and item.get("systemdUnit")))
|
|
if not units:
|
|
return {"status": "not-applicable", "buckets": [], "samples": []}
|
|
buckets = {}
|
|
samples = []
|
|
for unit in units:
|
|
proc = run(["journalctl", "--unit", unit, "--since", since_start_time(PAYLOAD.get("since")), "--no-pager", "--output", "json", "--lines", "2000"])
|
|
if proc.returncode != 0:
|
|
continue
|
|
for line in proc.stdout.decode("utf-8", errors="replace").splitlines():
|
|
try:
|
|
item = json.loads(line)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
message = str(item.get("MESSAGE") or "")
|
|
lowered = message.lower()
|
|
priority = int(item.get("PRIORITY")) if str(item.get("PRIORITY") or "").isdigit() else 7
|
|
cause = None
|
|
if "reconnect" in lowered or "retry" in lowered: cause = "reconnect"
|
|
elif "disconnect" in lowered or "closed" in lowered: cause = "disconnected"
|
|
elif "timeout" in lowered or "deadline" in lowered: cause = "timeout"
|
|
elif "error" in lowered or "failed" in lowered or priority <= 3: cause = "service-error"
|
|
if cause is None:
|
|
continue
|
|
buckets[cause] = buckets.get(cause, 0) + 1
|
|
if len(samples) < (30 if PAYLOAD.get("full") is True else 12):
|
|
samples.append({"at": item.get("__REALTIME_TIMESTAMP"), "cause": cause, "message": text(message, 400)})
|
|
return {
|
|
"status": "available",
|
|
"units": units,
|
|
"buckets": [{"cause": key, "count": value} for key, value in sorted(buckets.items(), key=lambda pair: (-pair[1], pair[0]))],
|
|
"samples": samples,
|
|
}
|
|
|
|
def classify_network_message(message, status_code):
|
|
lowered = message.lower()
|
|
checks = (
|
|
("dns-failure", ("no such host", "name resolution", "temporary failure in name resolution")),
|
|
("proxy-connect-failure", ("proxyconnect", "connect tunnel", "http connect")),
|
|
("connection-reset", ("connection reset", "reset by peer")),
|
|
("unexpected-eof", ("unexpected eof", " eof")),
|
|
("broken-pipe", ("broken pipe",)),
|
|
("connection-refused", ("connection refused",)),
|
|
("network-unreachable", ("network is unreachable", "no route to host")),
|
|
("transport-timeout", ("i/o timeout", "tls handshake timeout", "context deadline exceeded")),
|
|
("context-canceled", ("context canceled", "context cancelled")),
|
|
("stream-incomplete", ("missing terminal event", "stream usage incomplete", "response failed")),
|
|
)
|
|
for cause, markers in checks:
|
|
if any(marker in lowered for marker in markers):
|
|
return cause
|
|
return "upstream-503" if status_code == 503 or " 503" in lowered else None
|
|
|
|
def recent_network_evidence(account_id):
|
|
lines = runtime_log_lines()
|
|
buckets = {}
|
|
samples = []
|
|
matched = 0
|
|
sample_limit = 30 if PAYLOAD.get("full") is True else 12
|
|
for raw in lines:
|
|
candidate = raw[raw.find("{"):] if "{" in raw else raw
|
|
try:
|
|
item = json.loads(candidate)
|
|
except json.JSONDecodeError:
|
|
item = {}
|
|
extra = item.get("extra") if isinstance(item.get("extra"), dict) else {}
|
|
observed_account = item.get("account_id") if item.get("account_id") is not None else extra.get("account_id")
|
|
if str(observed_account) != str(account_id):
|
|
continue
|
|
status_code = item.get("status_code") if isinstance(item.get("status_code"), int) else extra.get("status_code")
|
|
if not isinstance(status_code, int):
|
|
upstream_status = extra.get("upstream_status")
|
|
status_code = upstream_status if isinstance(upstream_status, int) else None
|
|
message_parts = [item.get("message"), item.get("error"), extra.get("error"), extra.get("message")]
|
|
message = " | ".join(str(value) for value in message_parts if value not in (None, ""))
|
|
cause = classify_network_message(message, status_code)
|
|
if cause is None:
|
|
continue
|
|
matched += 1
|
|
buckets[cause] = buckets.get(cause, 0) + 1
|
|
if len(samples) < sample_limit:
|
|
samples.append({
|
|
"at": item.get("time") or item.get("created_at") or extra.get("_at"),
|
|
"requestId": item.get("request_id") or extra.get("request_id"),
|
|
"cause": cause,
|
|
"statusCode": status_code,
|
|
"message": text(message, 360),
|
|
})
|
|
return {
|
|
"source": "sub2api-target-runtime-logs",
|
|
"window": PAYLOAD.get("since"),
|
|
"scannedLines": len(lines),
|
|
"matchedEvents": matched,
|
|
"buckets": [{"cause": key, "count": value} for key, value in sorted(buckets.items(), key=lambda pair: (-pair[1], pair[0]))],
|
|
"samples": samples,
|
|
}
|
|
|
|
def infrastructure_snapshot(token, detail):
|
|
proxy = proxy_summary(token, detail.get("proxy_id"))
|
|
proxy_env, no_proxy = proxy_environment_summary()
|
|
owner = proxy_endpoint_owner(proxy)
|
|
return {
|
|
"source": "sub2api-admin-api-and-target-runtime",
|
|
"accountProxy": proxy,
|
|
"appRuntime": app_runtime_summary(),
|
|
"proxyEnvironment": proxy_env,
|
|
"noProxy": no_proxy,
|
|
"dns": dns_summary(),
|
|
"defaultRoute": default_route_summary(),
|
|
"interfaces": interface_summaries(),
|
|
"proxySockets": proxy_socket_summary(proxy),
|
|
"proxyEndpointOwner": owner,
|
|
"proxyServiceEvidence": proxy_service_evidence(owner),
|
|
"recentNetworkEvidence": recent_network_evidence(detail.get("id")),
|
|
"attribution": "The account proxy record controls this account's transport. Container proxy environment is reported separately. Interface counters and sockets are query-time snapshots; no upstream account test or application request is sent.",
|
|
"valuesPrinted": False,
|
|
}
|
|
|
|
def policy_summary(policy, include_rules=False):
|
|
keys = ("enabled", "ruleCount", "statusCodes", "rules") if include_rules else ("enabled", "ruleCount", "statusCodes")
|
|
return {key: policy[key] for key in keys}
|
|
|
|
def find_account(token, accounts, selector):
|
|
exact = [item for item in accounts if str(item.get("id")) == selector or item.get("name") == selector]
|
|
if len(exact) != 1:
|
|
raise RuntimeError("account-not-found-or-ambiguous: " + selector)
|
|
return account_detail(token, exact[0]["id"])
|
|
|
|
def runtime_log_lines():
|
|
since = PAYLOAD.get("since", "24h")
|
|
tail = str(PAYLOAD.get("tail", 50000))
|
|
if RUNTIME_MODE == "host-docker":
|
|
proc = docker(["logs", "--since", since, "--tail", tail, APP_POD])
|
|
else:
|
|
proc = kubectl(["-n", NAMESPACE, "logs", APP_POD, "--since=" + since, "--tail=" + tail])
|
|
if proc.returncode != 0:
|
|
raise RuntimeError("read Sub2API runtime logs failed: " + text(proc.stderr))
|
|
output = proc.stdout + b"\\n" + proc.stderr
|
|
return output.decode("utf-8", errors="replace").splitlines()
|
|
|
|
def runtime_policy_events():
|
|
markers = (
|
|
"account_temp_unschedulable", "openai.upstream_failover_switching",
|
|
"openai.account_select_failed", "openai.forward_failed",
|
|
"openai.pool_mode_same_account_retry", "gateway.failover_same_account_retry",
|
|
)
|
|
lines = runtime_log_lines()
|
|
policy = []
|
|
request_ids = set()
|
|
|
|
def parsed(line):
|
|
start = line.find("{")
|
|
if start < 0:
|
|
return None
|
|
try:
|
|
value = json.loads(line[start:])
|
|
return value if isinstance(value, dict) else None
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def normalized(item, marker, line):
|
|
prefix = line[:line.find("{")]
|
|
timestamp = re.search(r"\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}(?:\\.\\d+)?(?:Z|[+-]\\d{2}:?\\d{2})", prefix)
|
|
return {
|
|
"createdAt": item.get("created_at") or item.get("time") or item.get("timestamp") or (timestamp.group(0) if timestamp else None),
|
|
"marker": marker,
|
|
"requestId": item.get("request_id") or item.get("client_request_id"),
|
|
"accountId": item.get("account_id"),
|
|
"groupId": item.get("group_id"),
|
|
"statusCode": item.get("status_code"),
|
|
"upstreamStatus": item.get("upstream_status"),
|
|
"path": item.get("path"),
|
|
}
|
|
|
|
for line in lines:
|
|
marker = next((value for value in markers if value in line), None)
|
|
if marker is None:
|
|
continue
|
|
item = parsed(line)
|
|
if item is None:
|
|
continue
|
|
event = normalized(item, marker, line)
|
|
policy.append(event)
|
|
if isinstance(event.get("requestId"), str) and event.get("requestId"):
|
|
request_ids.add(event.get("requestId"))
|
|
|
|
completed = []
|
|
if request_ids:
|
|
for line in lines:
|
|
if "http request completed" not in line:
|
|
continue
|
|
item = parsed(line)
|
|
if item is None:
|
|
continue
|
|
request_id = item.get("request_id") or item.get("client_request_id")
|
|
if request_id not in request_ids or not gateway_request_path(item.get("path")):
|
|
continue
|
|
completed.append(normalized(item, "http request completed", line))
|
|
|
|
events = sorted(policy + completed, key=lambda item: str(item.get("createdAt") or ""))
|
|
return {
|
|
"ok": True,
|
|
"operation": "events",
|
|
"mutation": False,
|
|
"source": "sub2api-runtime-policy-events",
|
|
"window": PAYLOAD.get("since"),
|
|
"tail": PAYLOAD.get("tail"),
|
|
"scannedLineCount": len(lines),
|
|
"policyEventCount": len(policy),
|
|
"completionEventCount": len(completed),
|
|
"eventCount": len(events),
|
|
"events": events,
|
|
"valuesPrinted": False,
|
|
}
|
|
|
|
def app_exec(args):
|
|
if RUNTIME_MODE == "host-docker":
|
|
return docker(["exec", APP_POD, *args])
|
|
return kubectl(["-n", NAMESPACE, "exec", APP_POD, "--", *args])
|
|
|
|
def app_text(args):
|
|
proc = app_exec(args)
|
|
return proc.stdout.decode("utf-8", errors="replace").strip() if proc.returncode == 0 else ""
|
|
|
|
def config_scalar(path, key):
|
|
script = r'''awk -v wanted="$2" '
|
|
BEGIN { in_gateway=0 }
|
|
/^gateway:[[:space:]]*($|#)/ { in_gateway=1; next }
|
|
in_gateway && /^[^[:space:]#]/ { exit }
|
|
in_gateway && $0 ~ "^[[:space:]]+" wanted ":[[:space:]]*" {
|
|
value=$0
|
|
sub("^[[:space:]]+" wanted ":[[:space:]]*", "", value)
|
|
sub("[[:space:]]*#.*$", "", value)
|
|
gsub("^[[:space:]\\\"'\''']+|[[:space:]\\\"'\''']+$", "", value)
|
|
print value
|
|
exit
|
|
}' "$1"'''
|
|
return app_text(["sh", "-c", script, "sh", path, key])
|
|
|
|
def runtime_response_header_timeout():
|
|
env_openai = app_text(["sh", "-c", "printenv GATEWAY_OPENAI_RESPONSE_HEADER_TIMEOUT 2>/dev/null || true"])
|
|
env_generic = app_text(["sh", "-c", "printenv GATEWAY_RESPONSE_HEADER_TIMEOUT 2>/dev/null || true"])
|
|
config_file = app_text(["sh", "-c", "printenv CONFIG_FILE 2>/dev/null || true"])
|
|
paths = [value for value in (config_file, "/app/data/config.yaml", "/etc/sub2api/config.yaml", "/app/config.yaml") if value]
|
|
config_openai = ""
|
|
config_generic = ""
|
|
config_source = None
|
|
for path in dict.fromkeys(paths):
|
|
openai_value = config_scalar(path, "openai_response_header_timeout")
|
|
generic_value = config_scalar(path, "response_header_timeout")
|
|
if openai_value or generic_value:
|
|
config_openai = openai_value
|
|
config_generic = generic_value
|
|
config_source = path
|
|
break
|
|
def non_negative_int(value):
|
|
try:
|
|
parsed = int(str(value).strip())
|
|
return parsed if parsed >= 0 else None
|
|
except (TypeError, ValueError):
|
|
return None
|
|
openai_configured = non_negative_int(config_openai)
|
|
openai_env = non_negative_int(env_openai)
|
|
generic_configured = non_negative_int(config_generic)
|
|
generic_env = non_negative_int(env_generic)
|
|
effective = openai_configured if openai_configured is not None else openai_env if openai_env is not None else 0
|
|
source = "config-file:" + config_source if openai_configured is not None else "process-env" if openai_env is not None else "sub2api-default"
|
|
generic = generic_configured if generic_configured is not None else generic_env if generic_env is not None else 600
|
|
return {
|
|
"effectiveSeconds": effective,
|
|
"disabled": effective == 0,
|
|
"source": source,
|
|
"genericResponseHeaderTimeoutSeconds": generic,
|
|
"genericAppliesToOpenAI": False,
|
|
"configFileChecked": config_source,
|
|
"valuesPrinted": False,
|
|
}
|
|
|
|
def compact_error(value):
|
|
if not isinstance(value, str):
|
|
return ""
|
|
compact = " ".join(value.replace("\x00", "").split())
|
|
compact = re.sub(r"(?i)\\bBearer\\s+[A-Za-z0-9._~+/=-]+", "Bearer [REDACTED]", compact)
|
|
compact = re.sub(r"\\bsk-[A-Za-z0-9_-]{8,}\\b", "sk-[REDACTED]", compact)
|
|
compact = re.sub(r"(?i)\\b(api[_ -]?key|token)[=:]\\s*[^\\s,;]+", r"\\1=[REDACTED]", compact)
|
|
return compact[:240]
|
|
|
|
def since_start_time(value):
|
|
match = re.fullmatch(r"(\\d+)([mhd])", str(value or ""))
|
|
if not match:
|
|
raise RuntimeError("unsupported runtime error window: " + str(value))
|
|
amount = int(match.group(1))
|
|
unit = match.group(2)
|
|
duration = timedelta(minutes=amount) if unit == "m" else timedelta(hours=amount) if unit == "h" else timedelta(days=amount)
|
|
return (datetime.now(timezone.utc) - duration).isoformat().replace("+00:00", "Z")
|
|
|
|
def ops_error_cause(value):
|
|
lower = value.lower()
|
|
markers = {
|
|
"concurrency": ("concurrency limit", "concurrent request", "too many concurrent", "maximum concurrency", "max concurrency"),
|
|
"overload": ("overloaded", "server busy", "temporarily unavailable", "capacity exhausted"),
|
|
"model-route-unavailable": ("no available channel for model", "unsupported model", "model not found", "model_not_found"),
|
|
"context-window": ("context window", "context length", "maximum context"),
|
|
"continuation-payload": ("encrypted_content", "missing_required_parameter", "missing required parameter"),
|
|
"invalid-request": ("invalid_request", "invalid request"),
|
|
"rate-limit": ("rate limit", "too many requests"),
|
|
"access-denied": ("forbidden", "access denied", "permission denied"),
|
|
"not-found": ("not found", "does not exist"),
|
|
"incomplete-stream": ("ended before a terminal event", "missing terminal event", "before response.completed"),
|
|
"transport": ("do request failed", "unexpected eof", "connection reset", "stream read error", "stream closed", "internal_error; received from peer"),
|
|
"upstream-api-key-validation": ("failed to validate api key", "api key validation failed"),
|
|
"authentication": ("unauthorized", "invalid token", "token revoked", "invalid_grant"),
|
|
"generic-upstream-failure": ("upstream request failed", "upstream gateway error"),
|
|
}
|
|
for cause, values in markers.items():
|
|
if any(marker in lower for marker in values):
|
|
return cause
|
|
if re.match(r"recovered upstream error \\d+\\b", lower):
|
|
return "status-only-upstream-failure"
|
|
return "unknown"
|
|
|
|
def ops_error_analysis(token, account, log_statuses):
|
|
account_id = account.get("id")
|
|
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
|
|
path = f"/api/v1/admin/ops/upstream-errors?account_id={account_id}&view=all&page=1&page_size=500&start_time={start_time}"
|
|
response = curl_api("GET", path, bearer=token)
|
|
try:
|
|
data = data_of(response, "list account ops errors")
|
|
except Exception as exc:
|
|
return {
|
|
"source": "ops-upstream-errors",
|
|
"status": "unavailable",
|
|
"logObservedStatusCodes": log_statuses,
|
|
"reason": compact_error(str(exc)),
|
|
"assessment": "insufficient-evidence",
|
|
}
|
|
rows = items_of(data)
|
|
total = data.get("total", len(rows)) if isinstance(data, dict) else len(rows)
|
|
detail_limit = 24 if PAYLOAD.get("full") is True else 8
|
|
analyzed = []
|
|
for row in rows:
|
|
message = compact_error(row.get("message") if isinstance(row, dict) else "")
|
|
analyzed.append({"row": row, "evidence": message, "cause": ops_error_cause(message)})
|
|
for entry in analyzed[:detail_limit]:
|
|
row = entry["row"]
|
|
error_id = row.get("id") if isinstance(row, dict) else None
|
|
if not isinstance(error_id, int):
|
|
continue
|
|
detail_response = curl_api("GET", f"/api/v1/admin/ops/upstream-errors/{error_id}", bearer=token)
|
|
try:
|
|
detail = data_of(detail_response, "get account 500 ops error")
|
|
except Exception:
|
|
continue
|
|
fields = []
|
|
for key in ("message", "upstream_error_message", "upstream_error_detail", "error_body", "upstream_errors"):
|
|
value = detail.get(key) if isinstance(detail, dict) else None
|
|
if isinstance(value, str) and value.strip():
|
|
fields.append(value)
|
|
evidence = compact_error(" ".join(fields))
|
|
entry["evidence"] = evidence
|
|
entry["cause"] = ops_error_cause(evidence)
|
|
|
|
causes = {}
|
|
statuses = {}
|
|
messages = {}
|
|
models = {}
|
|
hours = {}
|
|
stream_modes = {"stream": 0, "synchronous": 0, "unknown": 0}
|
|
concurrency_count = 0
|
|
for entry in analyzed:
|
|
row = entry["row"]
|
|
error_id = row.get("id") if isinstance(row, dict) else None
|
|
evidence = entry["evidence"]
|
|
cause = entry["cause"]
|
|
causes[cause] = causes.get(cause, 0) + 1
|
|
list_message = compact_error(row.get("message") if isinstance(row, dict) else "")
|
|
if list_message:
|
|
messages[list_message] = messages.get(list_message, 0) + 1
|
|
status_code = row.get("status_code") if isinstance(row, dict) else None
|
|
if isinstance(status_code, int):
|
|
statuses[str(status_code)] = statuses.get(str(status_code), 0) + 1
|
|
if cause == "concurrency":
|
|
concurrency_count += 1
|
|
model = str(row.get("model") or "unknown") if isinstance(row, dict) else "unknown"
|
|
models[model] = models.get(model, 0) + 1
|
|
created_at = str(row.get("created_at") or "") if isinstance(row, dict) else ""
|
|
hour = created_at[:13] + ":00" if len(created_at) >= 13 else "unknown"
|
|
hours[hour] = hours.get(hour, 0) + 1
|
|
stream = row.get("stream") if isinstance(row, dict) else None
|
|
stream_key = "stream" if stream is True else "synchronous" if stream is False else "unknown"
|
|
stream_modes[stream_key] += 1
|
|
sample = {
|
|
"errorId": error_id,
|
|
"createdAt": row.get("created_at") if isinstance(row, dict) else None,
|
|
"requestId": row.get("request_id") if isinstance(row, dict) else None,
|
|
"model": row.get("model") if isinstance(row, dict) else None,
|
|
"stream": row.get("stream") if isinstance(row, dict) else None,
|
|
"cause": cause,
|
|
"evidence": evidence,
|
|
}
|
|
entry["sample"] = sample
|
|
analyzed_count = len(analyzed)
|
|
if analyzed_count == 0:
|
|
assessment = "insufficient-evidence"
|
|
elif concurrency_count == analyzed_count:
|
|
assessment = "concurrency-related"
|
|
elif concurrency_count > 0:
|
|
assessment = "mixed"
|
|
else:
|
|
assessment = "no-direct-concurrency-evidence"
|
|
dominant_cause = None
|
|
dominant_count = 0
|
|
if causes:
|
|
dominant_cause, dominant_count = max(causes.items(), key=lambda pair: pair[1])
|
|
dominant_share = round(dominant_count / analyzed_count, 4) if analyzed_count > 0 else None
|
|
if dominant_cause == "upstream-api-key-validation":
|
|
finding = "The upstream gateway returned HTTP 500 while validating its inbound API key; investigate its key lookup and storage dependencies first."
|
|
elif dominant_cause == "concurrency":
|
|
finding = "Direct concurrency-limit messages dominate the observed Ops records."
|
|
elif dominant_cause is not None:
|
|
finding = "The dominant observed Ops cause is " + dominant_cause + "."
|
|
else:
|
|
finding = "No structured HTTP 500 Ops records were available for cause analysis."
|
|
if concurrency_count > 0:
|
|
concurrency_action = "consider-lower-limit-after-controlled-comparison"
|
|
else:
|
|
concurrency_action = "do-not-lower-limit-as-primary-fix"
|
|
ranked_messages = sorted(messages.items(), key=lambda pair: (-pair[1], pair[0]))
|
|
message_limit = 20 if PAYLOAD.get("full") is True else 5
|
|
return {
|
|
"source": "ops-upstream-errors",
|
|
"status": "available",
|
|
"logObservedStatusCodes": log_statuses,
|
|
"opsRecordCount": total,
|
|
"analyzedRecordCount": analyzed_count,
|
|
"detailRecordCount": min(detail_limit, analyzed_count),
|
|
"recordsTruncated": total > analyzed_count,
|
|
"statusBuckets": [{"statusCode": int(code), "count": count} for code, count in sorted(statuses.items(), key=lambda pair: int(pair[0]))],
|
|
"causeBuckets": [{"cause": cause, "count": count} for cause, count in sorted(causes.items())],
|
|
"messageBuckets": [{"message": message, "count": count} for message, count in ranked_messages[:message_limit]],
|
|
"concurrencyRelatedRecordCount": concurrency_count,
|
|
"assessment": assessment,
|
|
"rootCause": {
|
|
"dominantCause": dominant_cause,
|
|
"dominantCount": dominant_count,
|
|
"dominantShare": dominant_share,
|
|
"finding": finding,
|
|
},
|
|
"errorRateEvidence": {
|
|
"errorRecordCount": total,
|
|
"requestAttemptDenominator": None,
|
|
"errorRate": None,
|
|
"reason": "The available Ops endpoint lists errors but does not provide all routing attempts for this account.",
|
|
},
|
|
"concurrencyRecommendation": {
|
|
"action": concurrency_action,
|
|
"reason": "No error-time concurrency history is available; only explicit concurrency error text is treated as direct evidence.",
|
|
},
|
|
"runtimeConcurrency": {
|
|
"limit": account.get("concurrency"),
|
|
"current": account.get("current_concurrency"),
|
|
"loadFactor": account.get("load_factor"),
|
|
"historicalAtErrorAvailable": False,
|
|
},
|
|
**({
|
|
"modelBuckets": [{"model": model, "count": count} for model, count in sorted(models.items(), key=lambda pair: (-pair[1], pair[0]))],
|
|
"hourBuckets": [{"hour": hour, "count": count} for hour, count in sorted(hours.items())],
|
|
"streamBuckets": [{"mode": mode, "count": count} for mode, count in stream_modes.items() if count > 0],
|
|
"samples": [entry["sample"] for entry in analyzed[:detail_limit]],
|
|
} if PAYLOAD.get("full") is True else {
|
|
"samples": [entry["sample"] for entry in analyzed[:1]],
|
|
}),
|
|
"disclosure": "Ops records are independently sampled evidence and are not positionally joined to account_upstream_error log lines.",
|
|
}
|
|
|
|
def gateway_request_path(value):
|
|
path = str(value or "").split("?", 1)[0].rstrip("/")
|
|
return path in (
|
|
"/responses", "/v1/responses", "/responses/compact", "/v1/responses/compact",
|
|
"/messages", "/v1/messages", "/chat/completions", "/v1/chat/completions",
|
|
)
|
|
|
|
def native_ops_snapshot(token, group_id, platform=None):
|
|
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
|
|
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
|
|
base_query = f"group_id={group_id}&start_time={start_time}{platform_query}"
|
|
|
|
def optional(path, label):
|
|
response = curl_api("GET", path, bearer=token)
|
|
try:
|
|
return {"status": "available", "data": data_of(response, label)}
|
|
except Exception as exc:
|
|
return {"status": "unavailable", "reason": compact_error(str(exc))}
|
|
|
|
overview_result = optional(f"/api/v1/admin/ops/dashboard/overview?{base_query}&query_mode=raw", "get ops dashboard overview")
|
|
availability_result = optional(f"/api/v1/admin/ops/account-availability?group_id={group_id}{platform_query}", "get ops account availability")
|
|
concurrency_result = optional(f"/api/v1/admin/ops/concurrency?group_id={group_id}{platform_query}", "get ops concurrency")
|
|
sink_result = optional("/api/v1/admin/ops/system-logs/health", "get ops system log health")
|
|
|
|
overview = overview_result.get("data") if overview_result.get("status") == "available" else None
|
|
overview_summary = None
|
|
if isinstance(overview, dict):
|
|
overview_summary = {
|
|
key: overview.get(key)
|
|
for key in (
|
|
"start_time", "end_time", "health_score", "success_count", "error_count_total",
|
|
"business_limited_count", "error_count_sla", "request_count_total", "request_count_sla",
|
|
"error_rate", "upstream_error_rate", "upstream_error_count_excl_429_529",
|
|
"upstream_429_count", "upstream_529_count",
|
|
"duration", "ttft",
|
|
)
|
|
}
|
|
|
|
availability = availability_result.get("data") if availability_result.get("status") == "available" else None
|
|
availability_summary = None
|
|
if isinstance(availability, dict):
|
|
group_values = availability.get("group") if isinstance(availability.get("group"), dict) else {}
|
|
account_values = availability.get("account") if isinstance(availability.get("account"), dict) else {}
|
|
selected_group = group_values.get(str(group_id)) or group_values.get(group_id)
|
|
unavailable = []
|
|
for value in account_values.values():
|
|
if not isinstance(value, dict) or value.get("group_id") != group_id or value.get("is_available") is True:
|
|
continue
|
|
unavailable.append({
|
|
"accountId": value.get("account_id"),
|
|
"accountName": value.get("account_name"),
|
|
"status": value.get("status"),
|
|
"rateLimited": value.get("is_rate_limited"),
|
|
"rateLimitRemainingSeconds": value.get("rate_limit_remaining_sec"),
|
|
"overloaded": value.get("is_overloaded"),
|
|
"overloadRemainingSeconds": value.get("overload_remaining_sec"),
|
|
"hasError": value.get("has_error"),
|
|
"error": compact_error(value.get("error_message")),
|
|
})
|
|
availability_summary = {
|
|
"enabled": availability.get("enabled"),
|
|
"timestamp": availability.get("timestamp"),
|
|
"group": selected_group,
|
|
"unavailableAccounts": unavailable[:20 if PAYLOAD.get("full") is True else 8],
|
|
"unavailableAccountCount": max(
|
|
len(unavailable),
|
|
max(0, selected_group.get("total_accounts") - selected_group.get("available_count"))
|
|
if isinstance(selected_group, dict)
|
|
and isinstance(selected_group.get("total_accounts"), int)
|
|
and isinstance(selected_group.get("available_count"), int)
|
|
else 0,
|
|
),
|
|
"unavailableAccountDetailsComplete": (
|
|
not isinstance(selected_group, dict)
|
|
or not isinstance(selected_group.get("total_accounts"), int)
|
|
or not isinstance(selected_group.get("available_count"), int)
|
|
or len(unavailable) >= max(0, selected_group.get("total_accounts") - selected_group.get("available_count"))
|
|
),
|
|
}
|
|
|
|
concurrency = concurrency_result.get("data") if concurrency_result.get("status") == "available" else None
|
|
concurrency_summary = None
|
|
if isinstance(concurrency, dict):
|
|
group_values = concurrency.get("group") if isinstance(concurrency.get("group"), dict) else {}
|
|
account_values = concurrency.get("account") if isinstance(concurrency.get("account"), dict) else {}
|
|
selected_group = group_values.get(str(group_id)) or group_values.get(group_id)
|
|
loaded = []
|
|
for value in account_values.values():
|
|
if not isinstance(value, dict) or value.get("group_id") != group_id:
|
|
continue
|
|
if (value.get("current_in_use") or 0) <= 0 and (value.get("waiting_in_queue") or 0) <= 0:
|
|
continue
|
|
loaded.append({
|
|
"accountId": value.get("account_id"),
|
|
"accountName": value.get("account_name"),
|
|
"currentInUse": value.get("current_in_use"),
|
|
"maxCapacity": value.get("max_capacity"),
|
|
"loadPercentage": value.get("load_percentage"),
|
|
"waitingInQueue": value.get("waiting_in_queue"),
|
|
})
|
|
concurrency_summary = {
|
|
"enabled": concurrency.get("enabled"),
|
|
"timestamp": concurrency.get("timestamp"),
|
|
"group": selected_group,
|
|
"activeAccounts": sorted(loaded, key=lambda item: (-(item.get("loadPercentage") or 0), str(item.get("accountName") or "")))[:20 if PAYLOAD.get("full") is True else 8],
|
|
}
|
|
|
|
sink = sink_result.get("data") if sink_result.get("status") == "available" else None
|
|
return {
|
|
"source": "sub2api-native-admin-ops",
|
|
"overview": {"status": overview_result.get("status"), "summary": overview_summary, "reason": overview_result.get("reason")},
|
|
"accountAvailability": {"status": availability_result.get("status"), "summary": availability_summary, "reason": availability_result.get("reason")},
|
|
"concurrency": {"status": concurrency_result.get("status"), "summary": concurrency_summary, "reason": concurrency_result.get("reason")},
|
|
"systemLogSink": {"status": sink_result.get("status"), "health": sink if isinstance(sink, dict) else None, "reason": sink_result.get("reason")},
|
|
}
|
|
|
|
def native_request_timing(token, group_id, platform, overview):
|
|
start_time = quote(since_start_time(PAYLOAD.get("since")), safe="")
|
|
platform_query = "&platform=" + quote(str(platform), safe="") if platform else ""
|
|
base = f"group_id={group_id}&start_time={start_time}{platform_query}"
|
|
all_counts = {}
|
|
error_counts = {}
|
|
status = "available"
|
|
reason = None
|
|
duration = overview.get("duration") if isinstance(overview.get("duration"), dict) else {}
|
|
ttft = overview.get("ttft") if isinstance(overview.get("ttft"), dict) else {}
|
|
ttft_p99 = ttft.get("p99_ms")
|
|
observed_floor = max(60, ((ttft_p99 + 29999) // 15000) * 15) if isinstance(ttft_p99, int) else None
|
|
candidate_seconds = sorted(set([30, 45, 60] + ([observed_floor] if isinstance(observed_floor, int) else [])))
|
|
error_duration_response = curl_api("GET", f"/api/v1/admin/ops/requests?{base}&page=1&page_size=1&kind=error&min_duration_ms=0", bearer=token)
|
|
try:
|
|
error_duration_data = data_of(error_duration_response, "get error duration coverage")
|
|
error_duration_record_count = error_duration_data.get("total") if isinstance(error_duration_data, dict) else None
|
|
except Exception as exc:
|
|
error_duration_record_count = None
|
|
status = "unavailable"
|
|
reason = compact_error(str(exc))
|
|
for seconds in candidate_seconds:
|
|
threshold = seconds * 1000
|
|
for kind, target in ((None, all_counts), ("error", error_counts)):
|
|
kind_query = "&kind=error" if kind else ""
|
|
response = curl_api("GET", f"/api/v1/admin/ops/requests?{base}&page=1&page_size=1&min_duration_ms={threshold}{kind_query}", bearer=token)
|
|
try:
|
|
data = data_of(response, "get long request count")
|
|
value = data.get("total") if isinstance(data, dict) else None
|
|
target[threshold] = None if kind == "error" and error_duration_record_count == 0 else value
|
|
except Exception as exc:
|
|
status = "unavailable"
|
|
reason = compact_error(str(exc))
|
|
target[threshold] = None
|
|
candidates = []
|
|
for seconds in candidate_seconds:
|
|
headroom = seconds * 1000 - ttft_p99 if isinstance(ttft_p99, int) else None
|
|
candidates.append({
|
|
"seconds": seconds,
|
|
"observedFloor": seconds == observed_floor,
|
|
"ttftP99HeadroomMs": headroom,
|
|
"ttftP99Below": headroom >= 0 if isinstance(headroom, int) else None,
|
|
"totalDurationAtLeastCount": all_counts.get(seconds * 1000),
|
|
"errorDurationAtLeastCount": error_counts.get(seconds * 1000),
|
|
"assessment": "observed-ttft-p99-plus-15s-rounded" if seconds == observed_floor else "ttft-p99-within-candidate" if isinstance(headroom, int) and headroom >= 0 else "ttft-p99-exceeds-candidate" if isinstance(headroom, int) else "insufficient-ttft-evidence",
|
|
})
|
|
since = str(PAYLOAD.get("since") or "")
|
|
token_range = {"30m": "30m", "60m": "1h", "1h": "1h", "24h": "1d", "1d": "1d"}.get(since)
|
|
model_timing = []
|
|
model_timing_status = "unsupported-window"
|
|
if token_range:
|
|
response = curl_api("GET", f"/api/v1/admin/ops/dashboard/openai-token-stats?group_id={group_id}&time_range={token_range}&top_n=100{platform_query}", bearer=token)
|
|
try:
|
|
data = data_of(response, "get OpenAI model token stats")
|
|
rows = items_of(data)
|
|
for row in rows:
|
|
if not isinstance(row, dict):
|
|
continue
|
|
request_count = row.get("request_count")
|
|
first_count = row.get("requests_with_first_token")
|
|
model_timing.append({
|
|
"model": row.get("model"), "requestCount": request_count,
|
|
"requestsWithFirstToken": first_count,
|
|
"firstTokenCoverage": round(first_count / request_count, 6) if isinstance(first_count, int) and isinstance(request_count, int) and request_count > 0 else None,
|
|
"avgFirstTokenMs": row.get("avg_first_token_ms"), "avgDurationMs": row.get("avg_duration_ms"),
|
|
"avgTokensPerSec": row.get("avg_tokens_per_sec"),
|
|
})
|
|
model_timing.sort(key=lambda item: (-(item.get("avgFirstTokenMs") or -1), str(item.get("model") or "")))
|
|
model_timing_status = "available"
|
|
except Exception as exc:
|
|
model_timing_status = "unavailable"
|
|
reason = compact_error(str(exc))
|
|
return {
|
|
"status": status,
|
|
"reason": reason,
|
|
"source": "sub2api-native-admin-ops-dashboard-and-requests",
|
|
"ttft": ttft,
|
|
"duration": duration,
|
|
"observedFloorSeconds": observed_floor,
|
|
"errorDurationRecordCount": error_duration_record_count,
|
|
"modelTimingStatus": model_timing_status,
|
|
"modelTimingWindow": token_range,
|
|
"modelTiming": model_timing[:20 if PAYLOAD.get("full") is True else 8],
|
|
"candidates": candidates,
|
|
"attribution": "TTFT covers successful streaming requests with first_token_ms. Total and error duration are end-to-end request durations, not response-header wait or guaranteed failover lead time. The observed floor is TTFT P99 plus 15 seconds rounded up to 15 seconds; it is not a deployment recommendation. A dash in ERROR_DUR>= means error duration coverage is unavailable.",
|
|
}
|
|
|
|
`;
|
|
}
|