#!/usr/bin/env python3 """Fetch TopSQL summary data from Clinic API for all instances. Required auth and cluster context via environment variables or CLI flags: export CLINIC_COOKIE='clinic.session_id=...' export CLINIC_CSRF_TOKEN='your-csrf-token' export CLINIC_CLUSTER_ID='...' export CLINIC_ORG_ID='...' export CLINIC_PROJECT_ID='...' export CLINIC_PROVIDER='gcp' export CLINIC_REGION='asia-northeast1' python scripts/fetch_topsql_summary.py output.json \\ --start 2026-07-03 --end 2026-07-16 """ from __future__ import annotations import argparse import json import os import sys import time import urllib.parse import urllib.request from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime, timezone, timedelta BASE_URL = "https://clinic.pingcap.com/ngm/api/v1/topsql" TZ = timezone(timedelta(hours=8)) WINDOW = "1h" TOP = 5 MAX_WORKERS = 15 def env_or_arg(env_name: str, arg_value: str | None) -> str: value = arg_value or os.environ.get(env_name, "").strip() if not value: raise SystemExit(f"Missing required value: set {env_name} or pass the matching CLI flag.") return value def parse_date(value: str) -> datetime: for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"): try: dt = datetime.strptime(value, fmt) return dt.replace(tzinfo=TZ) except ValueError: continue raise argparse.ArgumentTypeError(f"invalid date format: {value!r} (use YYYY-MM-DD)") def build_headers(args: argparse.Namespace) -> dict[str, str]: return { "accept": "application/json, text/plain, */*", "cookie": env_or_arg("CLINIC_COOKIE", args.cookie), "user-agent": "topsql-summary-fetch/1.0", "x-cluster-id": env_or_arg("CLINIC_CLUSTER_ID", args.cluster_id), "x-csrf-token": env_or_arg("CLINIC_CSRF_TOKEN", args.csrf_token), "x-deploy-type": args.deploy_type, "x-org-id": env_or_arg("CLINIC_ORG_ID", args.org_id), "x-project-id": env_or_arg("CLINIC_PROJECT_ID", args.project_id), "x-provider": env_or_arg("CLINIC_PROVIDER", args.provider), "x-region": env_or_arg("CLINIC_REGION", args.region), } def api_get(path: str, params: dict, headers: dict[str, str]) -> dict: query = urllib.parse.urlencode(params) url = f"{BASE_URL}/{path}?{query}" req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=120) as resp: body = resp.read().decode("utf-8") return json.loads(body) def fetch_instances(start_ts: int, end_ts: int, headers: dict[str, str]) -> list[dict]: result = api_get("instances", {"start": start_ts, "end": end_ts}, headers) if result.get("status") != "ok": raise RuntimeError(f"instances API failed: {result}") return result["data"] def fetch_summary( instance: str, instance_type: str, start_ts: int, end_ts: int, headers: dict[str, str], ) -> dict: params = { "start": start_ts, "end": end_ts, "instance": instance, "instance_type": instance_type, "top": TOP, "window": WINDOW, } try: result = api_get("summary", params, headers) return { "instance": instance, "instance_type": instance_type, "status": result.get("status", "unknown"), "data": result.get("data", []), "error": result.get("message") if result.get("status") != "ok" else None, } except Exception as e: return { "instance": instance, "instance_type": instance_type, "status": "error", "data": [], "error": str(e), } def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Fetch TopSQL summary data from Clinic API.") parser.add_argument( "output", nargs="?", default="topsql_summary.json", help="Output JSON path (default: topsql_summary.json)", ) parser.add_argument("--start", type=parse_date, required=True, help="Start date (YYYY-MM-DD, Asia/Shanghai)") parser.add_argument( "--end", type=parse_date, required=True, help="End date (YYYY-MM-DD, inclusive through 23:59:59 Asia/Shanghai)", ) parser.add_argument("--cookie", help="Clinic cookie header value (or CLINIC_COOKIE)") parser.add_argument("--csrf-token", dest="csrf_token", help="Clinic CSRF token (or CLINIC_CSRF_TOKEN)") parser.add_argument("--cluster-id", dest="cluster_id", help="Cluster ID (or CLINIC_CLUSTER_ID)") parser.add_argument("--org-id", dest="org_id", help="Organization ID (or CLINIC_ORG_ID)") parser.add_argument("--project-id", dest="project_id", help="Project ID (or CLINIC_PROJECT_ID)") parser.add_argument("--provider", help="Cloud provider (or CLINIC_PROVIDER)") parser.add_argument("--region", help="Cloud region (or CLINIC_REGION)") parser.add_argument( "--deploy-type", dest="deploy_type", default=os.environ.get("CLINIC_DEPLOY_TYPE", "dedicated"), help="Deploy type header (default: dedicated or CLINIC_DEPLOY_TYPE)", ) return parser.parse_args() def main() -> None: args = parse_args() headers = build_headers(args) start_ts = int(args.start.timestamp()) end_ts = int(args.end.replace(hour=23, minute=59, second=59).timestamp()) print(f"Fetching instances ({start_ts} - {end_ts})...") instances = fetch_instances(start_ts, end_ts, headers) print(f"Found {len(instances)} instances") summaries = [] errors = 0 done = 0 t0 = time.time() with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: futures = { pool.submit(fetch_summary, inst["instance"], inst["instance_type"], start_ts, end_ts, headers): inst for inst in instances } for future in as_completed(futures): result = future.result() summaries.append(result) done += 1 if result["status"] != "ok": errors += 1 if done % 50 == 0 or done == len(instances): elapsed = time.time() - t0 print(f" Progress: {done}/{len(instances)} ({errors} errors, {elapsed:.0f}s)") summaries.sort(key=lambda x: (x["instance_type"], x["instance"])) output = { "metadata": { "cluster_id": headers["x-cluster-id"], "org_id": headers["x-org-id"], "project_id": headers["x-project-id"], "provider": headers["x-provider"], "region": headers["x-region"], "start": start_ts, "end": end_ts, "start_time": datetime.fromtimestamp(start_ts, tz=TZ).isoformat(), "end_time": datetime.fromtimestamp(end_ts, tz=TZ).isoformat(), "window": WINDOW, "top": TOP, "instance_count": len(instances), "fetched_at": datetime.now(tz=TZ).isoformat(), }, "instances": summaries, } with open(args.output, "w", encoding="utf-8") as f: json.dump(output, f, ensure_ascii=False, indent=2) ok_count = sum(1 for s in summaries if s["status"] == "ok") print(f"\nDone. {ok_count}/{len(instances)} succeeded, {errors} failed.") print(f"Output: {args.output}") print(f"File size: {len(json.dumps(output)) / 1024 / 1024:.1f} MB (approx)") if __name__ == "__main__": main()