#!/usr/bin/env python3 import argparse import json import shutil import subprocess import sys from datetime import datetime from pathlib import Path from typing import Any from common import ( CLINIC_METRICS_DATA_TYPE, ClinicDataNotReadyError, ClinicLoginRequiredError, DEFAULT_BASE_URL, DEFAULT_CHROME_PROFILE, attach_wait_result, build_download_url, clinic_data_not_ready_message, clinic_data_timeout_message, clinic_session, normalize_unix_ts, parse_clinic_link, raise_for_status_with_login_hint, wait_for_clinic_data_ready_status, ) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Download one Clinic diagnostic bundle, extract it locally, and summarize meta/log paths" ) parser.add_argument( "target", help="Clinic portal link", ) parser.add_argument( "--item-id", default="", help="Optional Clinic download item id override. By default the script auto-discovers it from Clinic data_status.", ) parser.add_argument( "--download-kind", default="download", choices=["download", "downloadConf"], help="Clinic download endpoint kind when --item-id is used (default: download).", ) 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).", ) parser.add_argument( "--output-root", default=str(Path.home() / "Downloads"), help="Root output directory. The script creates a custom subdirectory when --output-subdir is set, " "otherwise a YYYY-MM-DD subdirectory.", ) parser.add_argument( "--output-subdir", default="", help="Optional output subdirectory name under --output-root, for example TICKET-8412 or GTOC-1234.", ) parser.add_argument( "--force", action="store_true", help="Redownload and re-extract even if the local files already exist.", ) parser.add_argument( "--skip-extract", action="store_true", help="Only download the archive and skip zstd/tar extraction.", ) return parser def ensure_command(name: str) -> str: resolved = shutil.which(name) if not resolved: raise RuntimeError(f"required command not found in PATH: {name}") return resolved def resolve_download_target(args: argparse.Namespace) -> dict[str, Any]: raw_target = str(args.target or "").strip() clinic_context = parse_clinic_link(raw_target) return { "base_url": args.base_url.rstrip("/"), "org_id": clinic_context["org_id"], "cluster_id": clinic_context["cluster_id"], "download_kind": args.download_kind, "item_id": str(args.item_id or "").strip(), "clinic_context": clinic_context, } def infer_download_kind_from_task_type(task_type: Any) -> str | None: text = str(task_type or "").strip().lower() if not text: return None if "conf" in text: return "downloadConf" if "download" in text or "diag" in text or "bundle" in text: return "download" return None def resolve_bundle_item_id( data_status: dict[str, Any], requested_kind: str = "download", ) -> tuple[str, dict[str, Any] | None]: items = list(data_status.get("items") or []) if not items: raise ValueError("Clinic data_status returned no items, so the bundle item could not be discovered.") candidates: list[dict[str, Any]] = [] fallback_candidates: list[dict[str, Any]] = [] requested_kind = str(requested_kind or "download").strip() for item in items: item_id = str(item.get("itemID") or "").strip() if not item_id: continue inferred_kind = infer_download_kind_from_task_type(item.get("taskType")) enriched = dict(item) enriched["download_kind"] = inferred_kind or requested_kind fallback_candidates.append(enriched) if inferred_kind == requested_kind: candidates.append(enriched) if not candidates: candidates = fallback_candidates if not candidates: raise ValueError("Clinic data_status did not expose any downloadable itemID for this window.") candidates.sort( key=lambda item: ( normalize_unix_ts(item.get("endTime")) or 0, normalize_unix_ts(item.get("startTime")) or 0, str(item.get("itemID") or ""), ), reverse=True, ) chosen = candidates[0] return str(chosen.get("itemID") or "").strip(), chosen def sanitize_output_subdir(value: str) -> str: text = str(value or "").strip() if not text: return "" text = text.replace("\\", "_").replace("/", "_") text = text.replace("\x00", "") text = text.strip(" .") return text def resolve_output_dir(output_root: str, output_subdir: str = "") -> Path: root = Path(output_root).expanduser() custom = sanitize_output_subdir(output_subdir) if custom: return root / custom date_dir = datetime.now().astimezone().date().isoformat() return root / date_dir def response_filename(headers: dict[str, str], fallback: str) -> str: content_disposition = str(headers.get("content-disposition") or headers.get("Content-Disposition") or "") if "filename=" not in content_disposition: return fallback filename = content_disposition.split("filename=", 1)[1].strip() if filename.startswith('"') and filename.endswith('"'): filename = filename[1:-1] return filename or fallback def download_archive( session, download_url: str, output_dir: Path, force: bool, ) -> tuple[Path, dict[str, Any]]: output_dir.mkdir(parents=True, exist_ok=True) head_resp = session.get(download_url, stream=True, timeout=60) raise_for_status_with_login_hint( head_resp, "Clinic diagnostic bundle download", profile_name=(getattr(session, "clinic_state", {}) or {}).get("profile_name"), ) fallback_name = Path(download_url.rstrip("/")).name or "clinic-download.bin" filename = response_filename(dict(head_resp.headers), fallback_name) archive_path = output_dir / filename expected_bytes = int(head_resp.headers.get("Content-Length") or 0) reuse_existing = ( archive_path.exists() and not force and (expected_bytes <= 0 or archive_path.stat().st_size == expected_bytes) ) if reuse_existing: head_resp.close() return archive_path, { "filename": filename, "content_type": head_resp.headers.get("Content-Type"), "content_length": expected_bytes, "reused_existing_file": True, } with head_resp: with archive_path.open("wb") as fh: for chunk in head_resp.iter_content(chunk_size=1024 * 1024): if chunk: fh.write(chunk) return archive_path, { "filename": filename, "content_type": head_resp.headers.get("Content-Type"), "content_length": archive_path.stat().st_size, "reused_existing_file": False, } def tar_path_for_archive(archive_path: Path) -> Path: name = archive_path.name if name.endswith(".tar.zst"): return archive_path.with_name(name[: -len(".zst")]) if name.endswith(".zst"): return archive_path.with_name(name[: -len(".zst")]) return archive_path def extract_dir_for_archive(archive_path: Path) -> Path: name = archive_path.name for suffix in [".tar.zst", ".tar.gz", ".tgz", ".tar", ".zst", ".gz"]: if name.endswith(suffix): return archive_path.with_name(name[: -len(suffix)]) return archive_path.with_name(f"{name}.contents") def decompress_archive(archive_path: Path, force: bool) -> Path: tar_path = tar_path_for_archive(archive_path) if archive_path == tar_path: return tar_path if tar_path.exists() and not force: return tar_path zstd_bin = ensure_command("zstd") subprocess.run( [zstd_bin, "-d", "-f", str(archive_path), "-o", str(tar_path)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, ) return tar_path def extract_tar(tar_path: Path, extract_dir: Path, force: bool) -> Path: tar_bin = ensure_command("tar") if extract_dir.exists() and force: shutil.rmtree(extract_dir) if not extract_dir.exists(): extract_dir.mkdir(parents=True, exist_ok=True) subprocess.run( [tar_bin, "-xf", str(tar_path), "-C", str(extract_dir)], check=True, ) return extract_dir def read_text_file(path: Path) -> str: return path.read_text(encoding="utf-8", errors="replace") def load_yaml(path: Path) -> Any: try: import yaml # type: ignore return yaml.safe_load(read_text_file(path)) except Exception: pass ruby = shutil.which("ruby") if ruby: proc = subprocess.run( [ ruby, "-ryaml", "-rjson", "-e", "print JSON.generate(YAML.load_file(ARGV[0]))", str(path), ], capture_output=True, text=True, ) if proc.returncode == 0 and proc.stdout.strip(): return json.loads(proc.stdout) return None def find_files_by_keywords(base_dir: Path, keywords: list[str]) -> list[Path]: hits: list[Path] = [] wanted = [item.lower() for item in keywords] for path in sorted(base_dir.rglob("*")): if not path.is_file(): continue low = str(path).lower() if any(keyword in low for keyword in wanted): hits.append(path) return hits def find_slow_log_files(base_dir: Path) -> list[Path]: hits: list[Path] = [] for path in sorted(base_dir.rglob("*")): if not path.is_file(): continue name = path.name.lower() if "monitor/metrics/" in str(path).lower(): continue if "slow" in name and name.endswith((".log", ".txt")): hits.append(path) continue if name in {"tidb-slow.log", "slow_query.log", "slow-query.log"}: hits.append(path) return hits def summarize_component_logs(extract_dir: Path, meta: dict[str, Any] | None) -> dict[str, Any]: topology = (meta or {}).get("topology") or {} components = { "pd": "pd_servers", "tikv": "tikv_servers", "tidb": "tidb_servers", } result: dict[str, Any] = {} warnings: list[str] = [] for component, meta_key in components.items(): items = topology.get(meta_key) or [] summary_items = [] for item in items: host = str(item.get("host") or "").strip() log_dir = str(item.get("log_dir") or "").strip() if not host or not log_dir: continue extracted_log_dir = extract_dir / host / log_dir.lstrip("/") log_files = [] if extracted_log_dir.exists(): for child in sorted(extracted_log_dir.rglob("*")): if child.is_file(): log_files.append(str(child)) summary_items.append( { "host": host, "port": item.get("port") or item.get("client_port") or item.get("peer_port"), "log_dir": log_dir, "extracted_log_dir": str(extracted_log_dir), "log_files": log_files, } ) missing = [entry["extracted_log_dir"] for entry in summary_items if not entry["log_files"]] if missing: warnings.append( f"{component}: no extracted log files found under {len(missing)} configured log_dir path(s)" ) result[component] = summary_items slow_logs = find_slow_log_files(extract_dir) slow_metrics = [ path for path in find_files_by_keywords(extract_dir, ["slow"]) if "monitor/metrics/" in str(path).lower() ] if not slow_logs and slow_metrics: warnings.append("no raw slow log file was extracted; only slow-related metric json files were found") result["slow"] = [str(path) for path in slow_logs] result["slow_metrics"] = [str(path) for path in slow_metrics] return { "components": result, "warnings": warnings, } def meta_summary(meta_path: Path, meta: dict[str, Any] | None) -> dict[str, Any]: raw_text = read_text_file(meta_path) topology = (meta or {}).get("topology") or {} component_counts = {} for meta_key in ["tidb_servers", "tikv_servers", "pd_servers", "monitoring_servers", "grafana_servers"]: value = topology.get(meta_key) or [] if value: component_counts[meta_key] = len(value) return { "meta_yaml_path": str(meta_path), "tidb_version": (meta or {}).get("tidb_version"), "component_counts": component_counts, "server_configs": topology.get("server_configs") or {}, "raw_excerpt": raw_text[:2000], } def build_result( target_info: dict[str, Any], archive_path: Path, download_info: dict[str, Any], tar_path: Path | None, extract_dir: Path | None, meta_path: Path | None, meta: dict[str, Any] | None, logs: dict[str, Any] | None, ) -> dict[str, Any]: result = { "download": { "base_url": target_info["base_url"], "org_id": target_info["org_id"], "cluster_id": target_info["cluster_id"], "download_kind": target_info["download_kind"], "item_id": target_info["item_id"], "download_url": target_info["source_url"], "archive_path": str(archive_path), **download_info, }, "extraction": { "tar_path": str(tar_path) if tar_path else None, "extract_dir": str(extract_dir) if extract_dir else None, }, } if meta_path: result["meta"] = meta_summary(meta_path, meta) if logs: result["logs"] = logs return result def main() -> int: parser = build_parser() args = parser.parse_args() data_status = None try: target_info = resolve_download_target(args) session = clinic_session(base_url=target_info["base_url"], profile_name=args.profile) clinic_context = target_info.get("clinic_context") if clinic_context: data_status, wait_result = wait_for_clinic_data_ready_status( session, clinic_context, data_type=CLINIC_METRICS_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="Clinic bundle download", waited_seconds=int(wait_result.get("waited_seconds") or 0), max_wait_seconds=int(wait_result.get("max_wait_seconds") or 0), ) ) if not data_status.get("ready"): raise ClinicDataNotReadyError( clinic_data_not_ready_message( clinic_context, data_status, action="Clinic bundle download", ) ) if not target_info.get("item_id"): item_id, chosen_item = resolve_bundle_item_id( data_status, requested_kind=target_info.get("download_kind") or "download", ) target_info["item_id"] = item_id if chosen_item and chosen_item.get("download_kind"): target_info["download_kind"] = chosen_item.get("download_kind") if not target_info.get("item_id"): raise ValueError( "Clinic bundle itemID could not be discovered. Pass --item-id only as a manual override if needed." ) target_info["source_url"] = build_download_url( base_url=target_info["base_url"], org_id=target_info["org_id"], cluster_id=target_info["cluster_id"], item_id=target_info["item_id"], download_kind=target_info["download_kind"], ) output_dir = resolve_output_dir(args.output_root, args.output_subdir) archive_path, download_info = download_archive( session=session, download_url=target_info["source_url"], output_dir=output_dir, force=args.force, ) except (ClinicLoginRequiredError, ClinicDataNotReadyError, ValueError, RuntimeError) as exc: print(str(exc), file=sys.stderr) return 1 if args.skip_extract: result = build_result( target_info=target_info, archive_path=archive_path, download_info=download_info, tar_path=None, extract_dir=None, meta_path=None, meta=None, logs=None, ) if data_status: result["data_status"] = data_status print( json.dumps( result, ensure_ascii=False, indent=2, ) ) return 0 try: tar_path = decompress_archive(archive_path, force=args.force) extract_dir = extract_tar(tar_path, extract_dir_for_archive(archive_path), force=args.force) except (RuntimeError, subprocess.CalledProcessError) as exc: print(f"failed to decompress or extract Clinic bundle: {exc}", file=sys.stderr) return 1 meta_path = extract_dir / "meta.yaml" meta = load_yaml(meta_path) if meta_path.exists() else None logs = summarize_component_logs(extract_dir, meta) result = build_result( target_info=target_info, archive_path=archive_path, download_info=download_info, tar_path=tar_path, extract_dir=extract_dir, meta_path=meta_path if meta_path.exists() else None, meta=meta, logs=logs, ) if data_status: result["data_status"] = data_status print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())