Codex
409 lines
20 KiB
TypeScript
409 lines
20 KiB
TypeScript
import type { UniDeskConfig } from "../config";
|
|
import { parseJsonOutput } from "../platform-infra-ops-library";
|
|
import { runSshCommandCapture } from "../ssh";
|
|
import { readSub2ApiConfig } from "../platform-infra/config";
|
|
import { resolveTarget } from "../platform-infra/manifest";
|
|
|
|
import type { Sub2ApiOpsOptions, Sub2ApiOpsTarget } from "./types";
|
|
|
|
export async function querySub2ApiOps(config: UniDeskConfig, options: Sub2ApiOpsOptions): Promise<Record<string, unknown>> {
|
|
const target = resolveOpsTarget(options.targetId);
|
|
const result = await runSshCommandCapture(config, target.route, ["sh"], remoteQueryScript(target, options));
|
|
const parsed = parseJsonOutput(result.stdout);
|
|
if (result.exitCode !== 0 || parsed === null) {
|
|
const stderr = result.stderr.trim().slice(-1200);
|
|
const stdout = result.stdout.trim().slice(-1200);
|
|
return {
|
|
ok: false,
|
|
sourceStatus: "unavailable",
|
|
error: stderr || stdout || "remote Sub2API ops query returned no JSON",
|
|
remote: { exitCode: result.exitCode, stdoutTail: stdout, stderrTail: stderr },
|
|
target,
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
return { ...parsed, target, valuesPrinted: false };
|
|
}
|
|
|
|
function resolveOpsTarget(targetId: string): Sub2ApiOpsTarget {
|
|
const config = readSub2ApiConfig();
|
|
const target = resolveTarget(config, targetId);
|
|
return {
|
|
id: target.id,
|
|
route: target.route,
|
|
namespace: target.namespace,
|
|
runtimeMode: target.runtimeMode,
|
|
hostDockerAppPort: target.hostDocker?.appPort ?? null,
|
|
hostDockerEnvPath: target.hostDocker?.envPath ?? null,
|
|
appSecretName: config.runtime.database.secretName,
|
|
};
|
|
}
|
|
|
|
function pyJson(value: unknown): string {
|
|
return `json.loads(${JSON.stringify(JSON.stringify(value))})`;
|
|
}
|
|
|
|
function remoteQueryScript(target: Sub2ApiOpsTarget, options: Sub2ApiOpsOptions): string {
|
|
return `
|
|
set -u
|
|
python3 - <<'PY'
|
|
import base64
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
from urllib.parse import urlencode
|
|
|
|
TARGET = ${pyJson(target)}
|
|
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")
|
|
return str(value)[-limit:]
|
|
|
|
def kube_json(args, label):
|
|
proc = run(["kubectl", *args, "-o", "json"])
|
|
if proc.returncode != 0:
|
|
raise RuntimeError(f"{label} failed: {tail_text(proc.stderr)}")
|
|
return json.loads(proc.stdout.decode("utf-8"))
|
|
|
|
def read_host_env():
|
|
path = TARGET.get("hostDockerEnvPath")
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as handle:
|
|
lines = handle.read().splitlines()
|
|
except PermissionError:
|
|
proc = run(["sudo", "-n", "cat", path])
|
|
if proc.returncode != 0:
|
|
raise RuntimeError("read host env failed: " + tail_text(proc.stderr))
|
|
lines = proc.stdout.decode("utf-8", errors="replace").splitlines()
|
|
values = {}
|
|
for line in lines:
|
|
text = line.strip()
|
|
if not text or text.startswith("#") or "=" not in text:
|
|
continue
|
|
key, value = text.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 TARGET["runtimeMode"] == "host-docker":
|
|
return "sub2api-app"
|
|
pods = kube_json(["-n", TARGET["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 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)
|
|
data = kube_json(["-n", TARGET["namespace"], "get", "configmap", "sub2api-config"], "configmap/sub2api-config").get("data") or {}
|
|
return data.get(key)
|
|
|
|
def secret_value(key):
|
|
if TARGET["runtimeMode"] == "host-docker":
|
|
return read_host_env().get(key)
|
|
data = kube_json(["-n", TARGET["namespace"], "get", "secret", TARGET["appSecretName"]], "sub2api secret").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__:"
|
|
position = output.rfind(marker)
|
|
if position < 0:
|
|
return {"ok": False, "httpStatus": 0, "json": None, "body": output, "stderr": tail_text(proc.stderr)}
|
|
body = output[:position]
|
|
try:
|
|
status = int(output[position + 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": tail_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 --max-time 20 -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H "Authorization: Bearer $token" "$url"
|
|
elif [ -n "$token" ]; then
|
|
curl -sS --max-time 20 -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 --max-time 20 -w '\n__HTTP_CODE__:%{http_code}' -X "$method" "$url"
|
|
else
|
|
curl -sS --max-time 20 -w '\n__HTTP_CODE__:%{http_code}' -X "$method" -H 'Content-Type: application/json' --data-binary @"$tmp" "$url"
|
|
fi
|
|
'''
|
|
if TARGET["runtimeMode"] == "host-docker":
|
|
command = ["sh", "-c", script, "sh", method, f"http://127.0.0.1:{TARGET['hostDockerAppPort']}{path}", bearer or ""]
|
|
else:
|
|
command = ["kubectl", "-n", TARGET["namespace"], "exec", "-i", APP_POD, "--", "sh", "-c", script, "sh", method, f"http://127.0.0.1:8080{path}", bearer or ""]
|
|
return parse_curl(run(command, body))
|
|
|
|
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 tail_text(response.get("body"), 400)
|
|
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) and isinstance(data.get("items"), list):
|
|
return data["items"]
|
|
return []
|
|
|
|
def get(token, path, params=None, label="GET"):
|
|
suffix = "?" + urlencode(params) if params else ""
|
|
return data_of(curl_api("GET", path + suffix, bearer=token), label)
|
|
|
|
def login():
|
|
email = config_value("ADMIN_EMAIL")
|
|
password = secret_value("ADMIN_PASSWORD")
|
|
if not email or not password:
|
|
raise RuntimeError("Sub2API admin credentials are unavailable")
|
|
data = data_of(curl_api("POST", "/api/v1/auth/login", payload={"email": email, "password": password}), "admin login")
|
|
token = data.get("access_token") or data.get("token") if isinstance(data, dict) else None
|
|
if not token:
|
|
raise RuntimeError("admin login response has no token")
|
|
return token
|
|
|
|
def find_named(items, selector, label):
|
|
if selector is None:
|
|
return items
|
|
text = str(selector).strip().lower()
|
|
numeric = int(text) if text.isdigit() else None
|
|
matches = [item for item in items if (numeric is not None and item.get("id") == numeric) or str(item.get("id", "")).strip().lower() == text or str(item.get("name", "")).strip().lower() == text]
|
|
if not matches:
|
|
available = [f"{item.get('id')}:{item.get('name')}" for item in items]
|
|
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"))
|
|
groups = find_named(groups, OPTIONS.get("group"), "group")
|
|
snapshots = []
|
|
selected = groups if groups else [{"id": None, "name": "all", "platform": OPTIONS.get("platform") or "all"}]
|
|
for group in selected:
|
|
params = {"time_range": OPTIONS["timeRange"]}
|
|
platform = OPTIONS.get("platform") or group.get("platform")
|
|
if platform and platform != "all":
|
|
params["platform"] = platform
|
|
if group.get("id"):
|
|
params["group_id"] = group["id"]
|
|
overview = get(token, "/api/v1/admin/ops/dashboard/overview", params, "ops dashboard overview")
|
|
snapshots.append({"group": {"id": group.get("id"), "name": group.get("name"), "platform": platform or "all"}, "overview": overview})
|
|
evidence = None
|
|
diagnosis_id = OPTIONS.get("diagnosisId")
|
|
if diagnosis_id:
|
|
scope, _, category = diagnosis_id.partition(":")
|
|
group = next((item["group"] for item in snapshots if ("g" + str(item["group"].get("id"))) == scope or scope == "all"), None)
|
|
if group is None:
|
|
raise RuntimeError(f"diagnosis id scope not found: {diagnosis_id}")
|
|
params = {"time_range": OPTIONS["timeRange"], "page": 1, "page_size": 20}
|
|
if group.get("id"):
|
|
params["group_id"] = group["id"]
|
|
if group.get("platform") and group.get("platform") != "all":
|
|
params["platform"] = group["platform"]
|
|
if category.startswith("upstream-"):
|
|
endpoint = "/api/v1/admin/ops/upstream-errors"
|
|
elif category.startswith("error-") or category.startswith("sla-"):
|
|
endpoint = "/api/v1/admin/ops/request-errors"
|
|
else:
|
|
endpoint = "/api/v1/admin/ops/requests"
|
|
params["kind"] = "all"
|
|
params["sort"] = "duration_desc"
|
|
evidence = {"diagnosisId": diagnosis_id, "records": items_of(get(token, endpoint, params, "diagnosis evidence"))}
|
|
return {"groups": snapshots, "evidence": evidence}
|
|
|
|
def channels(token):
|
|
monitors = items_of(get(token, "/api/v1/channel-monitors", None, "list channel monitors"))
|
|
if OPTIONS.get("platform"):
|
|
monitors = [item for item in monitors if str(item.get("provider", "")).lower() == OPTIONS["platform"].lower()]
|
|
monitors = find_named(monitors, OPTIONS.get("channel"), "channel")
|
|
output = []
|
|
for monitor in monitors:
|
|
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"))
|
|
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=(",", ":")))
|
|
sys.exit(1)
|
|
PY
|
|
`;
|
|
}
|