#!/usr/bin/env python3 import argparse import json import sys from pathlib import Path from common import parse_clinic_link, parse_grafana_link, write_json SCRIPT_DIR = Path(__file__).resolve().parent ANALYZE_CLUSTER_SCRIPT = SCRIPT_DIR / "analyze_cluster.py" ANALYZE_GRAFANA_SCRIPT = SCRIPT_DIR / "analyze_online_grafana.py" DOWNLOAD_BUNDLE_SCRIPT = SCRIPT_DIR / "download_bundle.py" METRICS_ONLY_SCRIPT = SCRIPT_DIR / "metrics_only_perf_diagnosis.py" DEEP_TIKV_WRITE_PATH_SCRIPT = SCRIPT_DIR / "deep_tikv_write_path_rca.py" def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Dispatch one Clinic link, online Grafana link, or Clinic bundle target to the appropriate access workflow" ) parser.add_argument("target", help="Clinic portal link or online Grafana link/base URL") parser.add_argument( "--route", default="auto", choices=["auto", "clinic", "grafana", "bundle"], help="Optional explicit route override. Defaults to auto-detect from the target and flags.", ) parser.add_argument("--base-url", default="", help="Optional Clinic base URL override for Clinic or bundle routes") parser.add_argument("--profile", default="", help="Optional Chrome profile override") parser.add_argument("--dashboard-uid", default="", help="Grafana dashboard uid or Clinic discovery override") parser.add_argument("--dashboard-name", default="", help="Clinic dashboard name override for discovery") parser.add_argument("--datasource-id", default="1", help="Grafana datasource proxy id") parser.add_argument("--from", dest="from_ts", default="", help="Optional unix timestamp override for Grafana start") parser.add_argument("--to", dest="to_ts", default="", help="Optional unix timestamp override for Grafana end") parser.add_argument("--item-id", default="", help="Optional Clinic download item id override for bundle route") parser.add_argument( "--download-kind", default="download", choices=["download", "downloadConf"], help="Clinic bundle endpoint kind when --item-id is used", ) parser.add_argument("--output-root", default="", help="Optional bundle output root directory") parser.add_argument("--output-subdir", default="", help="Optional bundle output subdirectory") parser.add_argument("--force", action="store_true", help="Force bundle redownload and re-extraction") parser.add_argument("--skip-extract", action="store_true", help="Only download the archive and skip extraction") parser.add_argument( "--include-metrics-evidence", action="store_true", help="For the clinic route, also collect metrics-only evidence through the dispatcher.", ) parser.add_argument( "--include-tikv-write-path-evidence", action="store_true", help="For the clinic route, also collect deep TiKV write-path evidence through the dispatcher.", ) return parser def detect_route(args) -> str: if args.route != "auto": return args.route target = str(args.target or "").strip() try: parse_clinic_link(target) return "clinic" except Exception: pass try: parse_grafana_link(target, dashboard_uid=str(args.dashboard_uid or "").strip() or None) return "grafana" except Exception: pass raise ValueError( "Could not infer the route from the target. Use --route clinic, --route grafana, or --route bundle." ) def run_json_command(cmd: list[str], *, route: str) -> dict: proc = __import__("subprocess").run(cmd, text=True, capture_output=True) stdout = (proc.stdout or "").strip() stderr = (proc.stderr or "").strip() if proc.returncode != 0: raise RuntimeError(stderr or stdout or f"{route} route failed with exit code {proc.returncode}") try: return json.loads(stdout) except Exception as exc: raise RuntimeError(f"{route} route returned non-JSON output: {exc}") from exc def build_clinic_followup_argv(args) -> list[str]: argv: list[str] = [] if args.base_url: argv.extend(["--base-url", args.base_url]) if args.profile: argv.extend(["--profile", args.profile]) if args.dashboard_uid: argv.extend(["--dashboard-uid", args.dashboard_uid]) if args.dashboard_name: argv.extend(["--dashboard-name", args.dashboard_name]) return argv def dispatch(args) -> dict: route = detect_route(args) target = str(args.target or "").strip() if route == "clinic": cmd = [sys.executable, str(ANALYZE_CLUSTER_SCRIPT), target, *build_clinic_followup_argv(args)] elif route == "grafana": cmd = [sys.executable, str(ANALYZE_GRAFANA_SCRIPT), target] if args.dashboard_uid: cmd.extend(["--dashboard-uid", args.dashboard_uid]) if args.profile: cmd.extend(["--profile", args.profile]) if args.datasource_id: cmd.extend(["--datasource-id", args.datasource_id]) if args.from_ts: cmd.extend(["--from", args.from_ts]) if args.to_ts: cmd.extend(["--to", args.to_ts]) else: cmd = [sys.executable, str(DOWNLOAD_BUNDLE_SCRIPT), target] if args.item_id: cmd.extend(["--item-id", args.item_id]) if args.download_kind: cmd.extend(["--download-kind", args.download_kind]) if args.base_url: cmd.extend(["--base-url", args.base_url]) if args.profile: cmd.extend(["--profile", args.profile]) if args.output_root: cmd.extend(["--output-root", args.output_root]) if args.output_subdir: cmd.extend(["--output-subdir", args.output_subdir]) if args.force: cmd.append("--force") if args.skip_extract: cmd.append("--skip-extract") payload = run_json_command(cmd, route=route) if route == "clinic": data_status = payload.get("data_status") or {} payload["metrics_evidence"] = {} payload["tikv_write_path_evidence"] = {} payload["access_warnings"] = list(payload.get("access_warnings") or []) if bool(data_status.get("ready")): clinic_followup_argv = [target, *build_clinic_followup_argv(args)] if args.include_metrics_evidence: try: payload["metrics_evidence"] = run_json_command( [sys.executable, str(METRICS_ONLY_SCRIPT), *clinic_followup_argv], route="clinic metrics-only evidence", ) except Exception as exc: payload["access_warnings"].append(str(exc)) if args.include_tikv_write_path_evidence: try: payload["tikv_write_path_evidence"] = run_json_command( [sys.executable, str(DEEP_TIKV_WRITE_PATH_SCRIPT), *clinic_followup_argv], route="clinic deep TiKV write-path RCA", ) except Exception as exc: payload["access_warnings"].append(str(exc)) payload["dispatcher"] = { "route": route, "target": target, } return payload def main() -> int: parser = build_parser() args = parser.parse_args() try: payload = dispatch(args) except Exception as exc: print(exc, file=sys.stderr) return 1 write_json(payload) return 0 if __name__ == "__main__": raise SystemExit(main())