pikasTech-unidesk/scripts/src/platform-infra-sub2api-ops/query.ts

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 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 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 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 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
    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}

try:
    token = login()
    data = diagnosis(token) if OPTIONS["action"] == "diagnosis" else channels(token)
    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
`;
}