pikasTech-unidesk/scripts/src/egress-proxy-traffic.ts

import type { CommandResult } from "./command";

export interface EgressProxyTrafficSpec {
  scope: "platform-infra";
  targetId: string;
  route: string;
  namespace: string;
  deploymentName: string;
  serviceName: string;
  port: number;
  sourceType: string;
  sourceRef: string;
  sourceConfigRef: string | null;
  sourceFingerprint: string | null;
}

export function egressProxyTrafficScript(spec: EgressProxyTrafficSpec, sampleSeconds: number): string {
  return `
set +e
namespace=${shQuote(spec.namespace)}
deployment=${shQuote(spec.deploymentName)}
service_name=${shQuote(spec.serviceName)}
service_port=${shQuote(String(spec.port))}
target_id=${shQuote(spec.targetId)}
sample_seconds=${shQuote(String(sampleSeconds))}
source_type=${shQuote(spec.sourceType)}
source_ref=${shQuote(spec.sourceRef)}
source_config_ref=${shQuote(spec.sourceConfigRef ?? "")}
source_fingerprint=${shQuote(spec.sourceFingerprint ?? "")}
selector="app.kubernetes.io/name=$deployment,app.kubernetes.io/component=egress-proxy"
pod="$(kubectl -n "$namespace" get pod -l "$selector" -o jsonpath='{.items[0].metadata.name}' 2>/dev/null)"
if [ -z "$pod" ]; then
  python3 - "$target_id" "$namespace" "$deployment" "$service_name" "$service_port" "$source_type" "$source_ref" "$source_config_ref" "$source_fingerprint" <<'PY'
import json, sys
target, namespace, deployment, service_name, service_port, source_type, source_ref, source_config_ref, source_fingerprint = sys.argv[1:10]
print(json.dumps({
  "ok": False,
  "reason": "egress-proxy-pod-missing",
  "target": target,
  "namespace": namespace,
  "deployment": deployment,
  "serviceName": service_name,
  "servicePort": int(service_port),
  "source": {"sourceType": source_type, "sourceRef": source_ref, "sourceConfigRef": source_config_ref or None, "sourceFingerprint": source_fingerprint or None, "valuesPrinted": False},
  "next": {"status": f"bun scripts/cli.ts platform-infra sub2api status --target {target} --full"},
}, ensure_ascii=False))
PY
  exit 0
fi
tmp="$(mktemp -d)"
trap 'rm -rf "$tmp"' EXIT
fetch_one() {
  path="$1"
  kubectl -n "$namespace" exec "$pod" -- sh -c 'if command -v wget >/dev/null 2>&1; then wget -qO- http://127.0.0.1:9090/connections; elif command -v curl >/dev/null 2>&1; then curl -fsS --max-time 3 http://127.0.0.1:9090/connections; else exit 127; fi' >"$path" 2>"$path.err"
  return "$?"
}
idx=0
while [ "$idx" -le "$sample_seconds" ]; do
  snapshot="$tmp/snapshot-$idx.json"
  date +%s%3N >"$tmp/snapshot-$idx.ms" 2>/dev/null || python3 - <<'PY' >"$tmp/snapshot-$idx.ms"
import time
print(int(time.time() * 1000))
PY
  fetch_one "$snapshot"
  printf '%s' "$?" >"$tmp/snapshot-$idx.rc"
  if [ "$idx" -lt "$sample_seconds" ]; then sleep 1; fi
  idx=$((idx + 1))
done
python3 - "$tmp" "$target_id" "$namespace" "$deployment" "$pod" "$service_name" "$service_port" "$sample_seconds" "$source_type" "$source_ref" "$source_config_ref" "$source_fingerprint" <<'PY'
from collections import Counter, defaultdict
import glob
import json
import pathlib
import sys

tmp = pathlib.Path(sys.argv[1])
target, namespace, deployment, pod, service_name, service_port = sys.argv[2:8]
sample_seconds = int(sys.argv[8])
source_type, source_ref, source_config_ref, source_fingerprint = sys.argv[9:13]

def read_text(path, limit=2000):
    try:
        return pathlib.Path(path).read_text(errors="replace")[-limit:]
    except FileNotFoundError:
        return ""

def number(value):
    try:
        return int(value)
    except Exception:
        return 0

def field(record, names, default=None):
    if not isinstance(record, dict):
        return default
    for name in names:
        value = record.get(name)
        if value not in (None, ""):
            return value
    return default

def metadata(conn):
    value = conn.get("metadata") if isinstance(conn, dict) else None
    return value if isinstance(value, dict) else {}

def client_of(conn):
    meta = metadata(conn)
    value = field(meta, ["sourceIP", "source_ip", "source", "clientIP", "client_ip", "srcIP", "src_ip"])
    if value is None:
        value = field(conn, ["sourceIP", "source_ip", "source"])
    text = str(value or "unknown")
    if text.count(":") == 1 and text.rsplit(":", 1)[1].isdigit():
        text = text.rsplit(":", 1)[0]
    return text

def destination_of(conn):
    meta = metadata(conn)
    host = field(meta, ["host", "destinationIP", "destination_ip", "destination", "dstIP", "dst_ip"])
    port = field(meta, ["destinationPort", "destination_port", "dstPort", "dst_port"])
    if host is None:
        host = field(conn, ["host", "destination"])
    if host is None:
        return "-"
    host_text = str(host)
    return f"{host_text}:{port}" if port not in (None, "") else host_text

def conn_id(conn):
    value = field(conn, ["id", "uuid", "connId", "connectionId"])
    if value not in (None, ""):
        return str(value)
    meta = metadata(conn)
    return "|".join([
        client_of(conn),
        str(field(meta, ["sourcePort", "source_port"], "")),
        destination_of(conn),
        str(field(meta, ["network", "type"], "")),
    ])

def upload_of(conn):
    return number(field(conn, ["upload", "up", "uploadBytes", "upload_bytes"], 0))

def download_of(conn):
    return number(field(conn, ["download", "down", "downloadBytes", "download_bytes"], 0))

def connections(data):
    items = data.get("connections") if isinstance(data, dict) else None
    return items if isinstance(items, list) else []

snapshots = []
fetch_failures = []
for json_path in sorted(glob.glob(str(tmp / "snapshot-*.json")), key=lambda path: int(path.rsplit("-", 1)[1].split(".", 1)[0])):
    index = int(json_path.rsplit("-", 1)[1].split(".", 1)[0])
    rc = read_text(tmp / f"snapshot-{index}.rc", 200).strip()
    raw = read_text(json_path, 8_000)
    ms = number(read_text(tmp / f"snapshot-{index}.ms", 200).strip())
    if rc != "0":
        fetch_failures.append({"index": index, "exitCode": number(rc), "stderrTail": read_text(f"{json_path}.err", 1200)})
        continue
    try:
        parsed = json.loads(raw)
    except Exception as exc:
        fetch_failures.append({"index": index, "exitCode": 0, "parseError": str(exc), "stdoutTail": raw[-1200:]})
        continue
    if isinstance(parsed, dict):
        snapshots.append({"index": index, "ms": ms, "data": parsed, "connections": connections(parsed)})

if not snapshots:
    print(json.dumps({
        "ok": False,
        "reason": "clash-api-unavailable",
        "target": target,
        "namespace": namespace,
        "deployment": deployment,
        "pod": pod,
        "controller": "127.0.0.1:9090",
        "fetchFailures": fetch_failures[-3:],
        "source": {"sourceType": source_type, "sourceRef": source_ref, "sourceConfigRef": source_config_ref or None, "sourceFingerprint": source_fingerprint or None, "valuesPrinted": False},
        "next": {"apply": f"bun scripts/cli.ts platform-infra sub2api apply --target {target} --confirm --wait"},
    }, ensure_ascii=False))
    sys.exit(0)

first, last = snapshots[0], snapshots[-1]
duration = max(0.001, (last["ms"] - first["ms"]) / 1000.0)
clients = defaultdict(lambda: {
    "client": "",
    "windowUploadBytes": 0,
    "windowDownloadBytes": 0,
    "activeUploadBytes": 0,
    "activeDownloadBytes": 0,
    "activeConnections": 0,
    "destinations": Counter(),
})
previous = {}
for snap_index, snap in enumerate(snapshots):
    current = {}
    for conn in snap["connections"]:
        cid = conn_id(conn)
        current[cid] = {
            "client": client_of(conn),
            "destination": destination_of(conn),
            "upload": upload_of(conn),
            "download": download_of(conn),
        }
    if snap_index > 0:
        for cid, curr in current.items():
            client = curr["client"]
            clients[client]["client"] = client
            prev = previous.get(cid)
            if prev is None:
                du = max(0, curr["upload"])
                dd = max(0, curr["download"])
            else:
                du = max(0, curr["upload"] - prev["upload"])
                dd = max(0, curr["download"] - prev["download"])
            clients[client]["windowUploadBytes"] += du
            clients[client]["windowDownloadBytes"] += dd
            if curr["destination"] != "-":
                clients[client]["destinations"][curr["destination"]] += 1
    previous = current

for conn in last["connections"]:
    client = client_of(conn)
    clients[client]["client"] = client
    clients[client]["activeConnections"] += 1
    clients[client]["activeUploadBytes"] += upload_of(conn)
    clients[client]["activeDownloadBytes"] += download_of(conn)
    dest = destination_of(conn)
    if dest != "-":
        clients[client]["destinations"][dest] += 1

def top_level_total(data, names):
    return number(field(data, names, 0))

first_data, last_data = first["data"], last["data"]
process_upload_first = top_level_total(first_data, ["uploadTotal", "upload_total", "upload"])
process_download_first = top_level_total(first_data, ["downloadTotal", "download_total", "download"])
process_upload_last = top_level_total(last_data, ["uploadTotal", "upload_total", "upload"])
process_download_last = top_level_total(last_data, ["downloadTotal", "download_total", "download"])

rows = []
for client, values in clients.items():
    upload = values["windowUploadBytes"]
    download = values["windowDownloadBytes"]
    active_upload = values["activeUploadBytes"]
    active_download = values["activeDownloadBytes"]
    rows.append({
        "client": client,
        "activeConnections": values["activeConnections"],
        "windowUploadBytes": upload,
        "windowDownloadBytes": download,
        "windowTotalBytes": upload + download,
        "uploadBps": upload / duration,
        "downloadBps": download / duration,
        "totalBps": (upload + download) / duration,
        "activeUploadBytes": active_upload,
        "activeDownloadBytes": active_download,
        "activeTotalBytes": active_upload + active_download,
        "topDestinations": [{"destination": name, "samples": count} for name, count in values["destinations"].most_common(5)],
    })
rows.sort(key=lambda item: (item["totalBps"], item["windowTotalBytes"], item["activeTotalBytes"]), reverse=True)

window_upload = sum(item["windowUploadBytes"] for item in rows)
window_download = sum(item["windowDownloadBytes"] for item in rows)
active_upload = sum(item["activeUploadBytes"] for item in rows)
active_download = sum(item["activeDownloadBytes"] for item in rows)
process_window_upload = max(0, process_upload_last - process_upload_first)
process_window_download = max(0, process_download_last - process_download_first)
payload = {
    "ok": True,
    "target": target,
    "namespace": namespace,
    "deployment": deployment,
    "pod": pod,
    "serviceName": service_name,
    "servicePort": int(service_port),
    "controller": "127.0.0.1:9090",
    "sampleSecondsRequested": sample_seconds,
    "sampleSecondsActual": duration,
    "snapshots": len(snapshots),
    "fetchFailures": fetch_failures[-3:],
    "source": {"sourceType": source_type, "sourceRef": source_ref, "sourceConfigRef": source_config_ref or None, "sourceFingerprint": source_fingerprint or None, "valuesPrinted": False},
    "totals": {
        "activeConnections": len(last["connections"]),
        "clientWindowUploadBytes": window_upload,
        "clientWindowDownloadBytes": window_download,
        "clientWindowTotalBytes": window_upload + window_download,
        "clientUploadBps": window_upload / duration,
        "clientDownloadBps": window_download / duration,
        "clientTotalBps": (window_upload + window_download) / duration,
        "clientActiveUploadBytes": active_upload,
        "clientActiveDownloadBytes": active_download,
        "clientActiveTotalBytes": active_upload + active_download,
        "processUploadTotalBytes": process_upload_last,
        "processDownloadTotalBytes": process_download_last,
        "processTotalBytes": process_upload_last + process_download_last,
        "processWindowUploadBytes": process_window_upload,
        "processWindowDownloadBytes": process_window_download,
        "processWindowTotalBytes": process_window_upload + process_window_download,
        "processWindowBps": (process_window_upload + process_window_download) / duration,
    },
    "clients": rows,
    "disclosure": "Per-client cumulative bytes are measured over this bounded sample window from proxy API connection counters; proxy process totals are included when the API reports them.",
}
print(json.dumps(payload, ensure_ascii=False))
PY
`;
}

export function egressProxyTrafficCompactResult(result: CommandResult): Record<string, unknown> {
  return {
    exitCode: result.exitCode,
    stdoutBytes: Buffer.byteLength(result.stdout, "utf8"),
    stderrBytes: Buffer.byteLength(result.stderr, "utf8"),
    stdoutTail: result.stdout.slice(-2000),
    stderrTail: result.stderr.slice(-2000),
    timedOut: result.timedOut,
  };
}

function shQuote(value: string): string {
  return `'${value.replaceAll("'", "'\"'\"'")}'`;
}