#!/usr/bin/env python3 import argparse import base64 import json import os import re import time from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse import requests from pycookiecheat import chrome_cookies SKILL_ROOT = Path(__file__).resolve().parent.parent SKILLS_ROOT = SKILL_ROOT.parent.parent DEFAULT_BASE_URL = os.environ.get("CLINIC_BASE_URL", "https://clinic.pingcap.com.cn").rstrip("/") DEFAULT_CLINIC_API_BASE = os.environ.get("CLINIC_API_BASE", "https://clinic.pingcap.com/clinic/api/v1").rstrip("/") DEFAULT_CHROME_PROFILE = os.environ.get("CLINIC_CHROME_PROFILE", "auto") DEFAULT_CHROME_DATA_DIR = Path( os.environ.get( "CLINIC_CHROME_DATA_DIR", str(Path.home() / "Library/Application Support/Google/Chrome"), ) ).expanduser() SESSION_COOKIE = "clinic.session_id" KNOWLEDGE_DIR = SKILLS_ROOT / "diagnosis" / "tidb-perf-diagnosis" / "knowledge" class ClinicLoginRequiredError(RuntimeError): pass class ClinicDataNotReadyError(RuntimeError): pass class ClinicApiPreflightRequiredError(RuntimeError): pass CLINIC_READY_STATUS = 100 CLINIC_METRICS_DATA_TYPE = 1 CLINIC_WAITABLE_STATUS_VALUES = {-1} def is_perimeter81_host(hostname: str) -> bool: """Return True if hostname is behind Perimeter81 gateway (pzero.perimeter81.com).""" return hostname.endswith(".pzero.perimeter81.com") if hostname else False _cached_chrome_version: str | None = None def detect_local_chrome_version() -> str: """Detect the installed Google Chrome major version. Currently supports macOS only. Returns cached result after first detection. Falls back to '130' if detection fails or on unsupported platforms. """ global _cached_chrome_version if _cached_chrome_version is not None: return _cached_chrome_version import platform import subprocess if platform.system() != "Darwin": _cached_chrome_version = "130" return _cached_chrome_version try: result = subprocess.run( ['/Applications/Google Chrome.app/Contents/MacOS/Google Chrome', '--version'], capture_output=True, text=True, timeout=10 ) ver = result.stdout.strip() m = re.search(r'\d+', ver) if m: major = m.group(0) # Ensure the result is numeric (paranoid guard for malformed version strings) int(major) # raises ValueError if not numeric _cached_chrome_version = major return _cached_chrome_version except Exception: pass _cached_chrome_version = "130" return _cached_chrome_version def parse_int_env(name: str, default: int, minimum: int = 0) -> int: raw = str(os.environ.get(name) or "").strip() if not raw: return max(default, minimum) try: value = int(raw) except ValueError: return max(default, minimum) return max(value, minimum) CLINIC_DATA_STATUS_WAIT_TIMEOUT_SECONDS = parse_int_env( "CLINIC_DATA_STATUS_WAIT_TIMEOUT_SECONDS", 600, minimum=0, ) CLINIC_DATA_STATUS_POLL_INTERVAL_SECONDS = parse_int_env( "CLINIC_DATA_STATUS_POLL_INTERVAL_SECONDS", 15, minimum=1, ) def parse_json_arg(text: str | None) -> dict[str, Any]: if not text: return {} return json.loads(text) def normalize_unix_ts(value: str | int | None) -> int | None: if value in (None, ""): return None text = str(value).strip() if not text: return None raw = int(text) if raw > 10**12: return raw // 1000 return raw def load_env_value(name: str, env_file: str | None = None) -> str: value = str(os.environ.get(name) or "").strip() if value: return value candidate = Path(env_file or os.environ.get("ENV_FILE") or ".env").expanduser() if not candidate.exists(): return "" try: for line in candidate.read_text(encoding="utf-8").splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#") or "=" not in stripped: continue key, raw = stripped.split("=", 1) if key.strip() != name: continue parsed = raw.strip().strip('"').strip("'") if parsed: return parsed except OSError: return "" return "" def login_hint(base_url: str, profile_name: str | None = None, cookie_file: Path | None = None) -> str: parts = [ f"Open {base_url} in Google Chrome on this machine and log in once, then rerun the skill.", "Local Clinic uses the browser session cookie path here instead of API-key auth.", ] if profile_name and profile_name != "auto": parts.append(f"If you logged in with a different Chrome profile, rerun with --profile {profile_name!r}.") else: parts.append("If you logged in with a non-default Chrome profile, rerun with --profile ''.") return " ".join(parts) def chrome_cookie_path(profile_name: str, chrome_data_dir: Path = DEFAULT_CHROME_DATA_DIR) -> Path: return chrome_data_dir / profile_name / "Cookies" def candidate_chrome_profiles( profile_name: str = DEFAULT_CHROME_PROFILE, chrome_data_dir: Path = DEFAULT_CHROME_DATA_DIR, ) -> list[str]: requested = str(profile_name or "").strip() if requested and requested.lower() != "auto": return [requested] if not chrome_data_dir.exists(): return [] candidates: list[str] = [] if (chrome_data_dir / "Default").exists(): candidates.append("Default") for path in sorted(chrome_data_dir.glob("Profile *")): if path.is_dir(): candidates.append(path.name) return candidates def load_chrome_cookies( base_url: str = DEFAULT_BASE_URL, profile_name: str = DEFAULT_CHROME_PROFILE, chrome_data_dir: Path = DEFAULT_CHROME_DATA_DIR, ) -> dict[str, Any]: candidates = candidate_chrome_profiles(profile_name=profile_name, chrome_data_dir=chrome_data_dir) errors = [] if not candidates: raise ClinicLoginRequiredError( f"No Chrome profiles were found under {chrome_data_dir}. " f"{login_hint(base_url=base_url, profile_name=profile_name)}" ) for candidate in candidates: cookie_file = chrome_cookie_path(candidate, chrome_data_dir=chrome_data_dir) if not cookie_file.exists(): continue try: cookies = chrome_cookies(base_url, cookie_file=str(cookie_file)) except Exception as exc: # pragma: no cover - depends on local keychain/browser state errors.append(f"{candidate}: failed to read Chrome cookies ({type(exc).__name__})") continue session_id = str(cookies.get(SESSION_COOKIE) or "").strip() if session_id: return { "cookies": cookies, "profile_name": candidate, "cookie_file": cookie_file, } details = f" Failed attempts: {'; '.join(errors)}." if errors else "" raise ClinicLoginRequiredError( f"{SESSION_COOKIE} was not found in any usable Chrome profile.{details} " f"{login_hint(base_url=base_url, profile_name=profile_name)}" ) def load_browser_cookies( base_url: str, profile_name: str = DEFAULT_CHROME_PROFILE, chrome_data_dir: Path = DEFAULT_CHROME_DATA_DIR, ) -> dict[str, Any] | None: """Load any available cookies for base_url from Chrome profiles. For Grafana access, we don't require a specific cookie (unlike Clinic which requires clinic.session_id). This function tries all available profiles and returns the first one with any cookies for the URL. Returns None if no cookies found (which is acceptable for Grafana — the request will proceed without auth). """ candidates = candidate_chrome_profiles(profile_name=profile_name, chrome_data_dir=chrome_data_dir) for candidate in candidates: cookie_file = chrome_cookie_path(candidate, chrome_data_dir=chrome_data_dir) if not cookie_file.exists(): continue try: cookies = chrome_cookies(base_url, cookie_file=str(cookie_file)) except Exception as exc: # Log the error but continue — this is for Grafana, not Clinic continue if cookies: return { "cookies": cookies, "profile_name": candidate, "cookie_file": cookie_file, } return None def decode_jwt_payload(token: str) -> dict[str, Any]: parts = str(token or "").split(".") if len(parts) < 2: raise ValueError("invalid clinic.session_id JWT format") payload = parts[1] payload += "=" * (-len(payload) % 4) return json.loads(base64.urlsafe_b64decode(payload.encode())) def clinic_login_state( base_url: str = DEFAULT_BASE_URL, profile_name: str = DEFAULT_CHROME_PROFILE, ) -> dict[str, Any]: loaded = load_chrome_cookies(base_url=base_url, profile_name=profile_name) cookies = loaded["cookies"] session_id = str(cookies.get(SESSION_COOKIE) or "").strip() if not session_id: raise ClinicLoginRequiredError( f"{SESSION_COOKIE} not found in Chrome profile {loaded['profile_name']}. " f"{login_hint(base_url=base_url, profile_name=loaded['profile_name'], cookie_file=loaded['cookie_file'])}" ) try: payload = decode_jwt_payload(session_id) except Exception as exc: raise ClinicLoginRequiredError( f"Failed to decode {SESSION_COOKIE} from Chrome profile {loaded['profile_name']}: {exc}. " f"{login_hint(base_url=base_url, profile_name=loaded['profile_name'], cookie_file=loaded['cookie_file'])}" ) from exc csrf_token = ( payload.get("user", {}).get("csrfToken") or payload.get("csrfToken") or "" ) if not csrf_token: raise ClinicLoginRequiredError( f"csrf token not found in {SESSION_COOKIE} payload. " f"{login_hint(base_url=base_url, profile_name=loaded['profile_name'], cookie_file=loaded['cookie_file'])}" ) expires_at = payload.get("exp") try: expires_at = int(expires_at) if expires_at is not None else None except (TypeError, ValueError): expires_at = None if expires_at is not None and expires_at <= int(time.time()) + 60: raise ClinicLoginRequiredError( f"Clinic browser session in Chrome profile {loaded['profile_name']} is expired or about to expire. " f"{login_hint(base_url=base_url, profile_name=loaded['profile_name'], cookie_file=loaded['cookie_file'])}" ) return { "base_url": base_url.rstrip("/"), "profile_name": loaded["profile_name"], "cookie_file": str(loaded["cookie_file"]), "session_id": session_id, "csrf_token": csrf_token, "expires_at": expires_at, "jwt_payload": payload, } def clinic_portal_host_requires_meta_preflight(clinic_context: dict[str, Any]) -> bool: host = (urlparse(str(clinic_context.get("base_url") or "")).hostname or "").lower() return host == "clinic.pingcap.com" def first_present_value(payload: dict[str, Any], *keys: str) -> str: for key in keys: value = payload.get(key) text = str(value or "").strip() if text: return text return "" def normalize_clinic_api_meta(meta: dict[str, Any]) -> dict[str, Any]: provider = first_present_value(meta, "provider", "cloudProvider", "clusterProviderName") region = first_present_value(meta, "region", "cloudRegion", "clusterRegionName") cluster_type = first_present_value(meta, "clusterType") return { "clusterID": meta.get("clusterID"), "clusterName": meta.get("clusterName"), "orgID": meta.get("orgID"), "tenantID": meta.get("tenantID"), "tenantName": meta.get("tenantName"), "clusterType": cluster_type, "clusterDeployType": meta.get("clusterDeployType"), "clusterDeployTypeV2": meta.get("clusterDeployTypeV2"), "provider": provider, "region": region, "status": meta.get("status"), "version": meta.get("version"), } def clinic_api_preflight_meta(clinic_context: dict[str, Any]) -> dict[str, Any]: if not clinic_portal_host_requires_meta_preflight(clinic_context): return { "required": False, "used": False, "handoff_to_tidb_cloud_clinic": False, "reason": "portal host does not require clinic-api meta preflight", "cluster_meta": {}, } api_key = load_env_value("CLINIC_API_KEY") if not api_key: raise ClinicApiPreflightRequiredError( "clinic.pingcap.com portal analysis now requires a clinic-api metadata preflight first, " "but CLINIC_API_KEY is not configured in the environment or .env." ) session = requests.Session() session.headers.update( { "Authorization": f"Bearer {api_key}", "accept": "application/json", "user-agent": "op-clinic-grafana-access/0.1 clinic-api-preflight", } ) resp = session.get( f"{DEFAULT_CLINIC_API_BASE}/dashboard/clusters", params={ "cluster_id": str(clinic_context.get("cluster_id") or ""), "show_deleted": "true", "limit": "10", "page": "1", }, timeout=30, ) resp.raise_for_status() payload = resp.json() items = payload.get("items") or [] cluster_id = str(clinic_context.get("cluster_id") or "") matched = next((item for item in items if str(item.get("clusterID") or "") == cluster_id), None) if not matched: raise ClinicApiPreflightRequiredError( f"clinic-api metadata preflight could not find cluster {cluster_id} for clinic.pingcap.com portal routing." ) normalized = normalize_clinic_api_meta(matched) cluster_type = str(normalized.get("clusterType") or "").strip().lower() is_cloud_cluster = cluster_type == "cloud" return { "required": True, "used": True, "handoff_to_tidb_cloud_clinic": is_cloud_cluster, "reason": ( "clinic-api metadata says clusterType=cloud, so this portal link should be analyzed by the TiDB Cloud Clinic flow" if is_cloud_cluster else "clinic-api metadata is not clusterType=cloud, so continue with local Chrome Clinic access only" ), "cluster_meta": normalized, } def clinic_session(base_url: str = DEFAULT_BASE_URL, profile_name: str = DEFAULT_CHROME_PROFILE) -> requests.Session: state = clinic_login_state(base_url=base_url, profile_name=profile_name) session = requests.Session() session.cookies.set(SESSION_COOKIE, state["session_id"], domain=urlparse(state["base_url"]).hostname) session.headers.update( { "x-csrf-token": state["csrf_token"], "accept": "application/json", "user-agent": "op-clinic-grafana-access/0.1", } ) session.base_url = state["base_url"] # type: ignore[attr-defined] session.clinic_state = state # type: ignore[attr-defined] return session def grafana_session( base_url: str, profile_name: str = DEFAULT_CHROME_PROFILE, path_prefix: str = "", source_link: str = "", ) -> requests.Session: session = requests.Session() session.trust_env = False session.headers.update( { "accept": "application/json", "user-agent": "op-clinic-grafana-access/0.1", } ) normalized_base_url = base_url.rstrip("/") parsed = urlparse(normalized_base_url) cookie_candidates = [normalized_base_url] host_only_url = f"{parsed.scheme}://{parsed.hostname}" if parsed.scheme and parsed.hostname else "" if host_only_url and host_only_url not in cookie_candidates: cookie_candidates.append(host_only_url) loaded = None for candidate in cookie_candidates: loaded = load_browser_cookies(candidate, profile_name=profile_name) if loaded: break if loaded: domain = urlparse(normalized_base_url).hostname for name, value in dict(loaded["cookies"]).items(): session.cookies.set(name, value, domain=domain) session.grafana_state = { # type: ignore[attr-defined] "profile_name": loaded["profile_name"], "cookie_file": str(loaded["cookie_file"]), } # Expose whether cookies were successfully loaded from Chrome DB. # Downstream callers (e.g. P81 probe) can use this to distinguish # "no cookies found" from "cookies found but session expired server-side". session.grafana_cookies_loaded = loaded is not None # type: ignore[attr-defined] session.base_url = normalized_base_url # type: ignore[attr-defined] normalized_path_prefix = str(path_prefix or "").strip() if normalized_path_prefix in {"", "/"}: normalized_path_prefix = "" else: normalized_path_prefix = "/" + normalized_path_prefix.strip("/") session.grafana_path_prefix = normalized_path_prefix # type: ignore[attr-defined] session.grafana_api_base_url = f"{normalized_base_url}{normalized_path_prefix}" # type: ignore[attr-defined] # Perimeter81 / SSO gateway requires browser-grade headers. # Without these (especially accept wildcards and full Referer path), # the P81 gateway redirects to the login page even with valid cookies. if is_perimeter81_host(parsed.hostname): p81_referer = str(source_link or normalized_base_url).strip() # Use a full dashboard path as Referer when source_link is a Grafana URL if source_link and "/d/" in source_link: p81_referer = source_link # Browser accept header with wildcard - P81 requires this session.headers["accept"] = "application/json, text/plain, */*" # Use the browser-brand sec-ch-ua format seen in real Chrome traffic major = detect_local_chrome_version().split(".", 1)[0] session.headers["sec-ch-ua"] = ( f'"Google Chrome";v="{major}", "Chromium";v="{major}", "Not)A;Brand";v="24"' ) # P81 gateway checks the User-Agent version. Use major.0.0.0 format # matching what the actual browser reports via navigator.userAgent, # not the full installed version (e.g. 149.0.7827.115) that # build_browser_like_user_agent() generates from the Info.plist. browser_ua = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) " f"Chrome/{major}.0.0.0 Safari/537.36" ) session.headers["user-agent"] = browser_ua session.headers.setdefault("x-grafana-org-id", "1") session.headers.setdefault("Origin", normalized_base_url) session.headers.setdefault("Referer", p81_referer) session.headers.setdefault("Accept-Language", "en,zh;q=0.9,zh-CN;q=0.8") session.headers.setdefault("Connection", "keep-alive") session.headers.setdefault("Sec-Fetch-Dest", "empty") session.headers.setdefault("Sec-Fetch-Mode", "cors") session.headers.setdefault("Sec-Fetch-Site", "same-origin") return session def raise_for_status_with_login_hint( resp: requests.Response, action: str, profile_name: str | None = None, ) -> None: if resp.ok: return if resp.status_code in {401, 403}: request_url = urlparse(resp.request.url) base_url = f"{request_url.scheme}://{request_url.netloc}" raise ClinicLoginRequiredError( f"{action} returned HTTP {resp.status_code}. " f"{login_hint(base_url=base_url, profile_name=profile_name or DEFAULT_CHROME_PROFILE)}" ) resp.raise_for_status() def parse_clinic_link(link: str) -> dict[str, Any]: parsed = urlparse(link) fragment = parsed.fragment or "" fragment_path, _, fragment_query = fragment.partition("?") match = re.search(r"/orgs/(?P\d+)/clusters/(?P\d+)", fragment_path) if not match: raise ValueError(f"unsupported clinic link: {link}") query = parse_qs(fragment_query, keep_blank_values=True) from_ts = normalize_unix_ts(first_query_value(query, "from")) to_ts = normalize_unix_ts(first_query_value(query, "to")) if from_ts is None or to_ts is None: raise ValueError("clinic link must include from/to query parameters") return { "base_url": f"{parsed.scheme}://{parsed.netloc}", "source_link": link, "org_id": match.group("org_id"), "cluster_id": match.group("cluster_id"), "from": from_ts, "to": to_ts, "from_ms": from_ts * 1000, "to_ms": to_ts * 1000, "fragment_path": fragment_path, "fragment_query": query, } def parse_download_url(url: str) -> dict[str, Any]: parsed = urlparse(url) match = re.search( r"/clinic/api/v1/orgs/(?P\d+)/clusters/(?P\d+)/" r"(?Pdownload|downloadConf)/(?P[^/?#]+)", parsed.path, ) if not match: raise ValueError(f"unsupported clinic download url: {url}") return { "base_url": f"{parsed.scheme}://{parsed.netloc}", "source_url": url, "org_id": match.group("org_id"), "cluster_id": match.group("cluster_id"), "download_kind": match.group("download_kind"), "item_id": match.group("item_id"), } def parse_grafana_link(link: str, dashboard_uid: str | None = None) -> dict[str, Any]: parsed = urlparse(link) base_url = f"{parsed.scheme}://{parsed.netloc}".rstrip("/") path = str(parsed.path or "") query = parse_qs(parsed.query, keep_blank_values=True) uid = str(dashboard_uid or "").strip() if not uid: match = re.search(r"/d/([^/]+)", path) if match: uid = match.group(1) path_prefix = "" if "/d/" in path: path_prefix = path.split("/d/", 1)[0].rstrip("/") elif path and path not in {"", "/"}: path_prefix = path.rstrip("/") if path_prefix in {"", "/"}: path_prefix = "" from_ts = normalize_unix_ts(first_query_value(query, "from")) to_ts = normalize_unix_ts(first_query_value(query, "to")) return { "base_url": base_url, "path_prefix": path_prefix, "api_base_url": f"{base_url}{path_prefix}", "source_link": link, "dashboard_uid": uid, "path": path, "query": query, "from": from_ts, "to": to_ts, "from_ms": from_ts * 1000 if from_ts is not None else None, "to_ms": to_ts * 1000 if to_ts is not None else None, } def build_download_url( base_url: str, org_id: str, cluster_id: str, item_id: str, download_kind: str = "download", ) -> str: kind = str(download_kind or "download").strip() if kind not in {"download", "downloadConf"}: raise ValueError(f"unsupported download kind: {download_kind}") return ( f"{base_url.rstrip('/')}/clinic/api/v1/orgs/{org_id}/clusters/{cluster_id}/" f"{kind}/{item_id}" ) def first_query_value(query: dict[str, list[str]], key: str) -> str | None: values = query.get(key) or [] return values[0] if values else None def clinic_dashboard_list(session: requests.Session, clinic_context: dict[str, Any]) -> list[dict[str, Any]]: url = ( f"{session.base_url}/clinic/api/v1/orgs/{clinic_context['org_id']}" f"/clusters/{clinic_context['cluster_id']}/dashboard" ) resp = session.get( url, params={"startTime": clinic_context["from"], "endTime": clinic_context["to"]}, timeout=30, ) raise_for_status_with_login_hint( resp, "Clinic dashboard list", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) return resp.json() def clinic_cluster_meta(session: requests.Session, clinic_context: dict[str, Any]) -> dict[str, Any]: url = ( f"{session.base_url}/clinic/api/v1/orgs/{clinic_context['org_id']}" f"/clusters/{clinic_context['cluster_id']}" ) resp = session.get(url, timeout=30) raise_for_status_with_login_hint( resp, "Clinic cluster metadata fetch", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) return resp.json() def normalize_data_status_item(item: dict[str, Any]) -> dict[str, Any]: status = item.get("status") try: status = int(status) if status is not None else None except (TypeError, ValueError): status = None return { "itemID": item.get("itemID"), "status": status, "startTime": normalize_unix_ts(item.get("startTime")), "endTime": normalize_unix_ts(item.get("endTime")), "taskType": item.get("taskType"), } def clinic_data_status( session: requests.Session, clinic_context: dict[str, Any], data_type: int = CLINIC_METRICS_DATA_TYPE, use_cache: bool = True, ) -> dict[str, Any]: cache_key = ( str(clinic_context["org_id"]), str(clinic_context["cluster_id"]), int(clinic_context["from"]), int(clinic_context["to"]), int(data_type), ) cache = getattr(session, "clinic_data_status_cache", None) if use_cache and isinstance(cache, dict) and cache_key in cache: return cache[cache_key] url = ( f"{session.base_url}/clinic/api/v1/orgs/{clinic_context['org_id']}" f"/clusters/{clinic_context['cluster_id']}/data_status" ) resp = session.get( url, params={ "startTime": clinic_context["from"], "endTime": clinic_context["to"], "data_type": data_type, }, timeout=30, ) raise_for_status_with_login_hint( resp, "Clinic data status query", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) payload = resp.json() or {} items = [normalize_data_status_item(item) for item in (payload.get("items") or [])] statuses = [item["status"] for item in items if item.get("status") is not None] ready = bool(items) and all(status == CLINIC_READY_STATUS for status in statuses) and len(statuses) == len(items) result = { "data_type": int(data_type), "ready_status": CLINIC_READY_STATUS, "ready": ready, "items": items, "status_values": statuses, "max_status": max(statuses) if statuses else None, "min_status": min(statuses) if statuses else None, "raw": payload, } if not isinstance(cache, dict): cache = {} setattr(session, "clinic_data_status_cache", cache) cache[cache_key] = result return result def clinic_data_not_ready_message( clinic_context: dict[str, Any], data_status: dict[str, Any], action: str, ) -> str: statuses = data_status.get("status_values") or [] status_text = ", ".join(str(value) for value in statuses) if statuses else "no status item yet" return ( f"{action} is blocked because Clinic data is not ready for this window yet. " f"Current status: {status_text}. Wait until status reaches {CLINIC_READY_STATUS}, then rerun. " f"Window: from={clinic_context['from']} to={clinic_context['to']}." ) def clinic_data_should_wait(data_status: dict[str, Any]) -> bool: statuses = data_status.get("status_values") or [] return bool(statuses) and all(status in CLINIC_WAITABLE_STATUS_VALUES for status in statuses) def clinic_data_timeout_message( clinic_context: dict[str, Any], data_status: dict[str, Any], action: str, *, waited_seconds: int, max_wait_seconds: int, ) -> str: statuses = data_status.get("status_values") or [] status_text = ", ".join(str(value) for value in statuses) if statuses else "no status item yet" return ( f"Critical evidence missing: {action} cannot continue because Clinic data status stayed at " f"{status_text} for {waited_seconds}s and did not reach {CLINIC_READY_STATUS} within the " f"{max_wait_seconds}s wait window. Analysis cannot continue. " f"Window: from={clinic_context['from']} to={clinic_context['to']}." ) def wait_for_clinic_data_ready_status( session: requests.Session, clinic_context: dict[str, Any], data_type: int = CLINIC_METRICS_DATA_TYPE, *, max_wait_seconds: int = CLINIC_DATA_STATUS_WAIT_TIMEOUT_SECONDS, poll_interval_seconds: int = CLINIC_DATA_STATUS_POLL_INTERVAL_SECONDS, ) -> tuple[dict[str, Any], dict[str, Any]]: data_status = clinic_data_status( session, clinic_context, data_type=data_type, use_cache=False, ) if data_status.get("ready") or not clinic_data_should_wait(data_status) or max_wait_seconds <= 0: return data_status, { "waited": False, "timed_out": False, "waited_seconds": 0, "max_wait_seconds": max_wait_seconds, "poll_interval_seconds": poll_interval_seconds, } started_at = time.monotonic() polls = 0 while not data_status.get("ready") and clinic_data_should_wait(data_status): elapsed = time.monotonic() - started_at remaining = max_wait_seconds - elapsed if remaining <= 0: break time.sleep(min(float(poll_interval_seconds), remaining)) polls += 1 data_status = clinic_data_status( session, clinic_context, data_type=data_type, use_cache=False, ) waited_seconds = int(round(time.monotonic() - started_at)) timed_out = bool(not data_status.get("ready") and clinic_data_should_wait(data_status)) return data_status, { "waited": True, "timed_out": timed_out, "waited_seconds": waited_seconds, "max_wait_seconds": max_wait_seconds, "poll_interval_seconds": poll_interval_seconds, "polls": polls, } def attach_wait_result(data_status: dict[str, Any], wait_result: dict[str, Any]) -> dict[str, Any]: if not wait_result: return data_status annotated = dict(data_status or {}) annotated["wait_result"] = wait_result return annotated def ensure_clinic_data_ready( session: requests.Session, clinic_context: dict[str, Any], data_type: int = CLINIC_METRICS_DATA_TYPE, action: str = "Clinic metrics query", ) -> dict[str, Any]: data_status, wait_result = wait_for_clinic_data_ready_status( session, clinic_context, data_type=data_type, ) data_status = attach_wait_result(data_status, wait_result) if wait_result.get("timed_out"): raise ClinicDataNotReadyError( clinic_data_timeout_message( clinic_context, data_status, action, waited_seconds=int(wait_result.get("waited_seconds") or 0), max_wait_seconds=int(wait_result.get("max_wait_seconds") or CLINIC_DATA_STATUS_WAIT_TIMEOUT_SECONDS), ) ) if data_status.get("ready"): return data_status raise ClinicDataNotReadyError(clinic_data_not_ready_message(clinic_context, data_status, action)) def grafana_uid_from_path(url: str) -> str | None: match = re.search(r"/d/([^/]+)(?:/|$)", str(url or "")) return match.group(1) if match else None def parse_dashboard_url_vars(url: str) -> dict[str, Any]: parsed = urlparse(url) query = parse_qs(parsed.query, keep_blank_values=True) variable_context = {} for key, values in query.items(): value = values[0] if values else None if key.startswith("var-"): variable_context[key[4:]] = {"value": value, "source": "dashboard_url"} if "from" in query: from_ts = normalize_unix_ts(query["from"][0]) if from_ts is not None: variable_context["from_ms"] = {"value": from_ts * 1000, "source": "dashboard_url"} if "to" in query: to_ts = normalize_unix_ts(query["to"][0]) if to_ts is not None: variable_context["to_ms"] = {"value": to_ts * 1000, "source": "dashboard_url"} return variable_context def select_dashboard( dashboards: list[dict[str, Any]], dashboard_uid: str | None = None, dashboard_name: str | None = None, ) -> dict[str, Any]: if dashboard_uid: for item in dashboards: if grafana_uid_from_path(item.get("url", "")) == dashboard_uid: return item raise ValueError(f"dashboard uid not found: {dashboard_uid}") if dashboard_name: name_lower = dashboard_name.lower() for item in dashboards: if name_lower in str(item.get("name") or "").lower(): return item raise ValueError(f"dashboard name not found: {dashboard_name}") if not dashboards: raise ValueError("no dashboards returned by Clinic API") preferred_names = [ "performance-overview-cluster-clinic", "tidb-cluster-clinic", "overview-cluster-clinic", "tikv-details-cluster-clinic", ] for preferred in preferred_names: for item in dashboards: name = str(item.get("name") or "").strip().lower() if preferred in name: return item return dashboards[0] def grafana_dashboard_json(session: requests.Session, uid: str) -> dict[str, Any]: api_base_url = str( getattr(session, "grafana_api_base_url", "") or getattr(session, "base_url", "") or "" ).rstrip("/") url = f"{api_base_url}/api/dashboards/uid/{uid}" resp = session.get(url, timeout=30) clinic_state = getattr(session, "clinic_state", None) if clinic_state: raise_for_status_with_login_hint( resp, f"Grafana dashboard fetch for uid={uid}", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) else: resp.raise_for_status() return resp.json() def grafana_query( session: requests.Session, datasource_id: str | int, expr: str, *, time_ts: int | None = None, ) -> dict[str, Any]: base_url = str( getattr(session, "grafana_api_base_url", "") or getattr(session, "base_url", "") or "" ).rstrip("/") datasource_id = str(datasource_id).strip() if not datasource_id: raise ValueError("datasource_id is required") url = f"{base_url}/api/datasources/proxy/{datasource_id}/api/v1/query" params: dict[str, Any] = {"query": expr} if time_ts is not None: params["time"] = int(time_ts) resp = session.get(url, params=params, timeout=30) resp.raise_for_status() return resp.json() def grafana_query_range( session: requests.Session, datasource_id: str | int, expr: str, *, start_ts: int, end_ts: int, step: str = "60s", ) -> dict[str, Any]: base_url = str( getattr(session, "grafana_api_base_url", "") or getattr(session, "base_url", "") or "" ).rstrip("/") datasource_id = str(datasource_id).strip() if not datasource_id: raise ValueError("datasource_id is required") url = f"{base_url}/api/datasources/proxy/{datasource_id}/api/v1/query_range" params = { "query": expr, "start": to_rfc3339(start_ts), "end": to_rfc3339(end_ts), "step": step, } resp = session.get(url, params=params, timeout=30) resp.raise_for_status() return resp.json() def flatten_panels(panels: list[dict[str, Any]]) -> list[dict[str, Any]]: flat: list[dict[str, Any]] = [] for panel in panels or []: children = panel.get("panels") or [] if children: flat.extend(flatten_panels(children)) elif panel.get("targets"): flat.append(panel) return flat def normalize_templating(templating_list: list[dict[str, Any]]) -> list[dict[str, Any]]: out = [] for item in templating_list or []: query = item.get("query") if isinstance(query, dict): query = query.get("query") current = item.get("current") current_value = current.get("value") if isinstance(current, dict) else current out.append( { "name": item.get("name"), "type": item.get("type"), "label": item.get("label"), "query": query, "current": current_value, } ) return out def normalize_panels(panels: list[dict[str, Any]]) -> list[dict[str, Any]]: out = [] for panel in flatten_panels(panels): targets = [] for target in panel.get("targets") or []: expr = str(target.get("expr") or "").strip() if not expr: continue targets.append( { "ref_id": target.get("refId"), "expr": expr, "legend_format": target.get("legendFormat"), "step": target.get("step"), "datasource": target.get("datasource") or panel.get("datasource"), } ) if not targets: continue out.append( { "id": panel.get("id"), "title": panel.get("title"), "type": panel.get("type"), "description": panel.get("description"), "targets": targets, } ) return out def build_variable_context( clinic_context: dict[str, Any], dashboard_item: dict[str, Any], dashboard_doc: dict[str, Any], ) -> dict[str, Any]: variable_context = { "tidb_cluster": {"value": clinic_context["cluster_id"], "source": "clinic_link"}, "org": {"value": clinic_context["org_id"], "source": "clinic_link"}, "from": {"value": clinic_context["from"], "source": "clinic_link"}, "to": {"value": clinic_context["to"], "source": "clinic_link"}, } variable_context.update(parse_dashboard_url_vars(dashboard_item.get("url", ""))) for item in normalize_templating((dashboard_doc.get("templating") or {}).get("list") or []): name = item.get("name") if not name or name in variable_context: continue current = item.get("current") if current in ("", None): variable_context[name] = {"value": None, "source": "unresolved"} else: variable_context[name] = {"value": current, "source": "templating.current"} return variable_context def build_grafana_variable_context( grafana_context: dict[str, Any], dashboard_doc: dict[str, Any], ) -> dict[str, Any]: variable_context = {} variable_context.update(parse_dashboard_url_vars(grafana_context.get("source_link") or "")) for item in normalize_templating((dashboard_doc.get("templating") or {}).get("list") or []): name = item.get("name") if not name or name in variable_context: continue current = item.get("current") if current in ("", None): variable_context[name] = {"value": None, "source": "unresolved"} else: variable_context[name] = {"value": current, "source": "templating.current"} return variable_context def discover_dashboard( link: str, dashboard_uid: str | None = None, dashboard_name: str | None = None, base_url: str = DEFAULT_BASE_URL, profile_name: str = DEFAULT_CHROME_PROFILE, ) -> dict[str, Any]: clinic_context = parse_clinic_link(link) clinic_context["base_url"] = base_url.rstrip("/") clinic_api_session = clinic_session(base_url=base_url, profile_name=profile_name) dashboards = clinic_dashboard_list(clinic_api_session, clinic_context) dashboard_item = select_dashboard(dashboards, dashboard_uid=dashboard_uid, dashboard_name=dashboard_name) uid = grafana_uid_from_path(dashboard_item.get("url", "")) if not uid: raise ValueError(f"failed to extract Grafana uid from dashboard url: {dashboard_item.get('url')}") path_prefix = str(dashboard_item.get("url") or "").split("/d/", 1)[0].rstrip("/") grafana_api_session = grafana_session( base_url=base_url, profile_name=profile_name, path_prefix=path_prefix, ) dashboard_payload = grafana_dashboard_json(grafana_api_session, uid) dashboard_doc = dashboard_payload.get("dashboard") or {} return { "clinic_context": clinic_context, "dashboard": { "uid": uid, "title": dashboard_doc.get("title") or dashboard_item.get("name"), "url": dashboard_item.get("url"), "slug": dashboard_payload.get("meta", {}).get("slug"), }, "variable_context": build_variable_context(clinic_context, dashboard_item, dashboard_doc), "templating": normalize_templating((dashboard_doc.get("templating") or {}).get("list") or []), "panels": normalize_panels(dashboard_doc.get("panels") or []), } def discover_grafana_dashboard( link: str, *, dashboard_uid: str | None = None, profile_name: str = DEFAULT_CHROME_PROFILE, session: requests.Session | None = None, ) -> dict[str, Any]: """Discover Grafana dashboard metadata and panels from a dashboard link. If session is None, creates a new session with source_link for P81 header support. If session is provided, reuses it for the dashboard queries. Args: link: Grafana dashboard URL or base URL dashboard_uid: Optional dashboard UID (auto-detected from link if not provided) profile_name: Chrome profile name for cookie reuse session: Optional pre-configured requests.Session (if None, creates a new one) Returns: Dict with grafana_context, dashboard metadata, variables, templating, and panels """ grafana_context = parse_grafana_link(link, dashboard_uid=dashboard_uid) if session is None: session = grafana_session( grafana_context["base_url"], profile_name=profile_name, path_prefix=str(grafana_context.get("path_prefix") or ""), source_link=link, ) dashboard_payload = grafana_dashboard_json(session, grafana_context["dashboard_uid"]) dashboard_doc = dashboard_payload.get("dashboard") or {} return { "grafana_context": grafana_context, "dashboard": { "uid": grafana_context["dashboard_uid"], "title": dashboard_doc.get("title"), "url": link, "slug": dashboard_payload.get("meta", {}).get("slug"), }, "variable_context": build_grafana_variable_context(grafana_context, dashboard_doc), "templating": normalize_templating((dashboard_doc.get("templating") or {}).get("list") or []), "panels": normalize_panels(dashboard_doc.get("panels") or []), } def iter_knowledge_files() -> list[Path]: return sorted(KNOWLEDGE_DIR.glob("*.json")) def load_knowledge_file(path: Path) -> dict[str, Any]: with path.open() as fh: return json.load(fh) def find_metric_definition(dashboard_title: str, panel_title: str) -> dict[str, Any]: want_dashboard = str(dashboard_title or "").strip().lower() want_panel = str(panel_title or "").strip().lower() if not want_dashboard or not want_panel: raise ValueError("dashboard_title and panel_title are required") for path in iter_knowledge_files(): data = load_knowledge_file(path) if want_dashboard not in str(data.get("dashboard_title") or "").strip().lower(): continue for row in data.get("rows") or []: for panel in row.get("panels") or []: if want_panel == str(panel.get("title") or "").strip().lower(): return { "knowledge_file": str(path), "dashboard_title": data.get("dashboard_title"), "row_title": row.get("title"), "panel_title": panel.get("title"), "subtitle": panel.get("subtitle"), "expressions": panel.get("expressions") or [], } raise ValueError(f"metric definition not found for dashboard={dashboard_title!r}, panel={panel_title!r}") def replace_simple_variables(expr: str, variables: dict[str, str | None]) -> str: result = str(expr or "") for name, value in variables.items(): regex_value = value if value not in (None, "") else ".*" result = result.replace(f"${{{name}:regex}}", regex_value) result = result.replace(f"${name}", value or "") return result def cleanup_empty_label_matchers(expr: str) -> str: result = str(expr or "") patterns = [ r',?\s*k8s_cluster\s*=~\s*""', r',?\s*k8s_cluster\s*=\s*""', r',?\s*\w+\s*=\s*""(?=[,}])', ] for pattern in patterns: result = re.sub(pattern, "", result) result = re.sub(r'{\s*,', "{", result) result = re.sub(r',\s*,', ",", result) result = re.sub(r',\s*}', "}", result) result = re.sub(r'{\s*}', "{}", result) return result def dedupe_label_matchers(expr: str) -> str: def repl(match): body = match.group(1) parts = [part.strip() for part in body.split(",") if part.strip()] seen = set() kept = [] for part in parts: if part in seen: continue seen.add(part) kept.append(part) return "{" + ",".join(kept) + "}" return re.sub(r"{([^{}]+)}", repl, expr) def remove_label_matcher(expr: str, label: str) -> str: label = str(label or "").strip() if not label: return expr def repl(match): body = match.group(1) parts = [part.strip() for part in body.split(",") if part.strip()] kept = [part for part in parts if not re.match(rf"^{re.escape(label)}\s*(=~|!~|=|!=)\s*", part)] if not kept: return "" return "{" + ",".join(kept) + "}" return re.sub(r"{([^{}]*)}", repl, expr) def inject_expression_variables(expr: str, clinic_context: dict[str, Any], variable_rules: dict[str, Any] | None = None) -> str: rules = variable_rules or {} variables = { "tidb_cluster": clinic_context["cluster_id"], "k8s_cluster": rules.get("k8s_cluster", ""), "instance": rules.get("instance", ".*"), "store": rules.get("store", ".*"), "worker_instance": rules.get("worker_instance", ".*"), "additional_groupby": rules.get("additional_groupby", "instance"), "optional_quantile": rules.get("optional_quantile", "0.99"), "__rate_interval": rules.get("__rate_interval", "1m"), "keyspace_id": rules.get("keyspace_id", clinic_context["cluster_id"]), } result = replace_simple_variables(expr, variables) result = cleanup_empty_label_matchers(result) result = dedupe_label_matchers(result) return result def inject_store_id_for_instance( session: requests.Session, clinic_context: dict[str, Any], expr: str, *, instance: str = "", store: str = "", ) -> tuple[str, str]: final_expr = str(expr or "") final_store = str(store or "").strip() final_instance = str(instance or "").strip() needs_store_mapping = False if not final_store and final_instance and final_instance != ".*": if "$store" in final_expr: needs_store_mapping = True if re.search(r'store\s*(=~|=)\s*"\$instance"', final_expr): needs_store_mapping = True if re.search(r'store\s*(=~|=)\s*"[^"]*:\d+.*"', final_expr): needs_store_mapping = True if needs_store_mapping: final_store = str(resolve_store_id_for_instance(session, clinic_context, final_instance) or "").strip() if final_store: final_expr = final_expr.replace('"$store"', f'"{final_store}"') final_expr = re.sub(r'store\s*(=~|=)\s*"\$instance"', f'store="{final_store}"', final_expr) return final_expr, final_store def clinic_metrics_query( session: requests.Session, clinic_context: dict[str, Any], expr: str, step: str = "60s", ) -> dict[str, Any]: ensure_clinic_data_ready( session, clinic_context, data_type=CLINIC_METRICS_DATA_TYPE, action="Clinic metrics query", ) url = f"{session.base_url}/clinic/api/v1/data/metrics" headers = { "X-OrgType": "org", "X-OrgID": str(clinic_context["org_id"]), "X-ClusterID": str(clinic_context["cluster_id"]), } params = { "query": expr, "start": to_rfc3339(clinic_context["from"]), "end": to_rfc3339(clinic_context["to"]), "step": step, } resp = session.get(url, params=params, headers=headers, timeout=30) raise_for_status_with_login_hint( resp, "Clinic metrics query", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) return resp.json() def resolve_store_id_for_instance( session: requests.Session, clinic_context: dict[str, Any], instance: str, ) -> str | None: want_instance = str(instance or "").strip() if not want_instance: raise ValueError("instance is required") def split_host_port(value: str) -> tuple[str, int | None]: host, sep, port = str(value).strip().rpartition(":") if not sep: return str(value).strip(), None try: return host, int(port) except ValueError: return host, None # Prefer PD scheduler metrics because they reliably carry both address and store labels. candidate_exprs = [ f'pd_scheduler_store_status{{address="{want_instance}",type="leader_count"}}', f'pd_scheduler_store_status{{address="{want_instance}"}}', f'pd_hotspot_status{{address="{want_instance}"}}', ] for expr in candidate_exprs: payload = clinic_metrics_query(session, clinic_context, expr, step="30s") result = (((payload or {}).get("data") or {}).get("result")) or [] for item in result: metric = item.get("metric") or {} store_id = str(metric.get("store") or "").strip() address = str(metric.get("address") or "").strip() if store_id and address == want_instance: return store_id # PD address is often the service port while TiKV instance in dashboards is the status port. # Fall back to host-based matching and prefer the address whose port differs by 20. host, port = split_host_port(want_instance) payload = clinic_metrics_query( session, clinic_context, 'pd_scheduler_store_status{type="leader_count"}', step="30s", ) result = (((payload or {}).get("data") or {}).get("result")) or [] candidates: list[tuple[int, str]] = [] for item in result: metric = item.get("metric") or {} store_id = str(metric.get("store") or "").strip() address = str(metric.get("address") or "").strip() address_host, address_port = split_host_port(address) if not store_id or address_host != host: continue score = 1000 if port is not None and address_port is not None: diff = abs(port - address_port) score = 0 if diff == 20 else diff candidates.append((score, store_id)) if candidates: candidates.sort(key=lambda item: item[0]) return candidates[0][1] return None def to_rfc3339(ts: int) -> str: from datetime import datetime, timezone return datetime.fromtimestamp(int(ts), tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def series_last_values(metrics_payload: dict[str, Any]) -> list[dict[str, Any]]: result = (((metrics_payload or {}).get("data") or {}).get("result")) or [] items = [] for series in result: metric = series.get("metric") or {} values = series.get("values") or [] value = series.get("value") or [] last = values[-1][1] if values else (value[1] if len(value) >= 2 else None) items.append( { "metric": metric, "instance": metric.get("instance"), "last_value": last, } ) return items def grafana_frames_last_values(ds_query_payload: dict[str, Any]) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] results = (ds_query_payload or {}).get("results") or {} for ref_id, result in results.items(): for frame in result.get("frames") or []: schema = frame.get("schema") or {} fields = schema.get("fields") or [] data_values = (frame.get("data") or {}).get("values") or [] if len(fields) < 2 or len(data_values) < 2: continue labels = ((fields[1] or {}).get("labels") or {}) series_name = str(schema.get("name") or labels.get("instance") or labels.get("type") or ref_id) values = data_values[1] or [] timestamps = data_values[0] or [] if not values: continue last_value = values[-1] last_timestamp = timestamps[-1] if timestamps else None items.append( { "ref_id": ref_id, "series_name": series_name, "labels": labels, "last_value": last_value, "last_timestamp_ms": last_timestamp, } ) return items def is_json_content_type(content_type: str) -> bool: """Return True if the Content-Type header indicates JSON. Handles common variations: - application/json - text/json - custom+json (e.g., application/vnd.api+json) - Headers with charset parameters (e.g., application/json; charset=utf-8) Args: content_type: Value from Content-Type header, or empty string Returns: True if content type is JSON, False otherwise Example: >>> is_json_content_type("application/json") True >>> is_json_content_type("application/json; charset=utf-8") True >>> is_json_content_type("text/html") False """ if not content_type: return False ct = content_type.lower().split(';')[0].strip() return ct in ('application/json', 'text/json') or ct.endswith('+json') def probe_grafana_session(session: requests.Session, base_url: str) -> dict[str, Any]: """Probe whether the Grafana session is valid by hitting /api/org without redirects. This is used to detect expired P81 sessions before attempting dashboard queries. A valid session returns HTTP 200 with JSON content. Redirects (301/302) indicate the session has expired and the user needs to re-authenticate. Args: session: requests.Session object with cookies loaded base_url: Grafana base URL (e.g., https://xxxxx.pzero.perimeter81.com) Returns: Dict with keys: - "ok" (bool): True if session is valid - "status" (int): HTTP status code (0 if request failed) - "detail" (str): Human-readable status message Example valid response: {"ok": True, "status": 200, "detail": "session valid"} Example expired response: {"ok": False, "status": 302, "detail": "HTTP 302; redirect to https://....; non-JSON response"} Raises: No exceptions; returns error dict on any failure """ probe_url = base_url.rstrip("/") + "/api/org" try: resp = session.get(probe_url, timeout=15, allow_redirects=False) except Exception as exc: return {"ok": False, "status": 0, "detail": f"Request failed: {exc}"} content_type = resp.headers.get("Content-Type", "") if resp.status_code == 200 and is_json_content_type(content_type): return {"ok": True, "status": 200, "detail": "session valid"} detail_parts = [f"HTTP {resp.status_code}"] if resp.status_code in (301, 302, 307, 308): location = resp.headers.get("Location", "") detail_parts.append(f"redirect to {location}") if not is_json_content_type(content_type): detail_parts.append("non-JSON response (likely sign-in page)") return {"ok": False, "status": resp.status_code, "detail": "; ".join(detail_parts)} def write_json(data: Any, pretty: bool = True) -> None: if pretty: print(json.dumps(data, ensure_ascii=False, indent=2)) else: print(json.dumps(data, ensure_ascii=False)) def build_common_arg_parser(description: str) -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=description) parser.add_argument("link", help="Clinic portal link") parser.add_argument( "--base-url", default=DEFAULT_BASE_URL, help="Clinic base URL (default: env CLINIC_BASE_URL or https://clinic.pingcap.com.cn)", ) parser.add_argument( "--profile", default=DEFAULT_CHROME_PROFILE, help="Chrome profile name. Use 'auto' to scan profiles for clinic.session_id " "(default: env CLINIC_CHROME_PROFILE or auto).", ) return parser