from __future__ import annotations import argparse import json import sys from collections import defaultdict from datetime import datetime, timezone from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from backend.sql_catalog.server import create_finding_batch, create_finding_run, init_db # noqa: E402 SOURCE = "clinic_health_check" DETECTOR_VERSION = "clinic-health-check-v1" SLOWLOG_SYMPTOMS = { "top_by_count": "slow_query_frequency", "top_by_total_latency": "high_total_latency", "top_by_total_keys": "large_key_scan", "top_by_process_keys": "large_key_scan", "top_by_request_count": "high_kv_request_count", } SLOWLOG_SYMPTOM_METRICS = { "slow_query_frequency": ["slow_count"], "high_total_latency": ["total_latency_s", "max_latency_s"], "large_key_scan": ["total_keys", "process_keys"], "high_kv_request_count": ["request_count"], } def iso_from_epoch(value: Any) -> str | None: if value is None: return None try: return datetime.fromtimestamp(float(value), timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") except (TypeError, ValueError, OSError): return None def normalize_severity(value: Any) -> str: severity = str(value or "").strip().lower() if severity in {"critical", "high", "medium", "low", "info"}: return severity return "medium" def slowlog_symptom_severity(code: str, metrics: dict[str, Any]) -> str: if code == "high_total_latency": if float(metrics.get("max_latency_s") or 0) >= 30 or float(metrics.get("total_latency_s") or 0) >= 30_000: return "critical" if float(metrics.get("max_latency_s") or 0) >= 10 or float(metrics.get("total_latency_s") or 0) >= 3_000: return "high" elif code == "large_key_scan": maximum = max(float(metrics.get("total_keys") or 0), float(metrics.get("process_keys") or 0)) if maximum >= 10_000_000_000: return "critical" if maximum >= 1_000_000_000: return "high" elif code == "high_kv_request_count": if float(metrics.get("request_count") or 0) >= 10_000_000: return "critical" if float(metrics.get("request_count") or 0) >= 1_000_000: return "high" elif code == "slow_query_frequency" and float(metrics.get("slow_count") or 0) >= 1_000: return "high" return "medium" def slowlog_assessment(code: str, metrics: dict[str, Any]) -> dict[str, Any]: parameters: dict[str, Any] = {"adapter_profile": "clinic-report-import-v1"} if code == "high_total_latency": parameters.update( { "high_max_latency_s": 10, "critical_max_latency_s": 30, "high_total_latency_s": 3_000, "critical_total_latency_s": 30_000, } ) elif code == "large_key_scan": parameters.update({"high_keys": 1_000_000_000, "critical_keys": 10_000_000_000}) elif code == "high_kv_request_count": parameters.update({"high_requests": 1_000_000, "critical_requests": 10_000_000}) elif code == "slow_query_frequency": parameters.update({"high_slow_count": 1_000}) return { "method": "rule", "summary": "The Clinic report adapter classified this report hit before intake; SQL Catalog did not re-evaluate it.", "rule_code": f"clinic.slowlog_top_digest.{code}", "rule_version": DETECTOR_VERSION, "policy_id": "clinic-report-import-v1", "policy_version": "1", "parameters": parameters, } def max_number(rows: list[dict[str, Any]], key: str) -> float | int: values = [row.get(key) for row in rows if isinstance(row.get(key), (int, float))] if not values: return 0 return max(values) def first_query(rows: list[dict[str, Any]]) -> str: ranked = sorted(rows, key=lambda row: int(row.get("rank") or 999999)) for row in ranked: query = row.get("query") if isinstance(query, str) and query: return query return "" def report_context(report: dict[str, Any]) -> dict[str, Any]: return { "cluster_id": report.get("clusterId"), "org_id": report.get("dbaasOrgId"), "strategy_uuid": report.get("strategyUuid"), "created_at": report.get("createdAt"), "status": report.get("status"), "time_range_a_start": report.get("timeRangeAStart"), "time_range_a_end": report.get("timeRangeAEnd"), } def event_prefix(report: dict[str, Any]) -> str: strategy_uuid = report.get("strategyUuid") or "unknown_strategy" report_created_at = report.get("createdAt") or report.get("timeRangeAEnd") or "unknown_time" return f"{SOURCE}:{strategy_uuid}:{report_created_at}" def source_run_id(report: dict[str, Any]) -> str: strategy_uuid = report.get("strategyUuid") or "unknown_strategy" report_created_at = report.get("createdAt") or report.get("timeRangeAEnd") or "unknown_time" return f"{strategy_uuid}:{report_created_at}" def build_slowlog_findings(report: dict[str, Any], item: dict[str, Any]) -> list[dict[str, Any]]: content = item.get("content") or {} results = content.get("results") or {} rows_by_digest: dict[str, list[dict[str, Any]]] = defaultdict(list) for dimension, rows in results.items(): if not isinstance(rows, list): continue for row in rows: if not isinstance(row, dict) or not row.get("digest"): continue enriched = dict(row) enriched["dimension"] = dimension rows_by_digest[str(row["digest"])].append(enriched) findings: list[dict[str, Any]] = [] context = report_context(report) prefix = event_prefix(report) for digest, rows in sorted(rows_by_digest.items()): hit_dimensions = sorted({str(row["dimension"]) for row in rows}) top_ranks = { dimension: min(int(row.get("rank") or 999999) for row in rows if row.get("dimension") == dimension) for dimension in hit_dimensions } latest_seen = max_number(rows, "latest_time") metrics = { "slow_count": max_number(rows, "slow_count"), "total_latency_s": max_number(rows, "total_latency_s"), "max_latency_s": max_number(rows, "max_latency_s"), "total_keys": max_number(rows, "total_keys"), "process_keys": max_number(rows, "process_keys"), "request_count": max_number(rows, "request_count"), "sum_ru": max_number(rows, "sum_ru"), "latest_time": latest_seen, "result_row_count": len(rows), "dimension_count": len(hit_dimensions), } evidence_refs = [ { "type": "clinic_health_check_result", "check_name": item.get("name"), "dimension": row.get("dimension"), "rank": row.get("rank"), "latest_time": row.get("latest_time"), } for row in sorted(rows, key=lambda row: (str(row.get("dimension")), int(row.get("rank") or 999999))) ] symptoms = [] for symptom_code in sorted({SLOWLOG_SYMPTOMS[value] for value in hit_dimensions if value in SLOWLOG_SYMPTOMS}): symptom_dimensions = [ dimension for dimension in hit_dimensions if SLOWLOG_SYMPTOMS.get(dimension) == symptom_code ] symptoms.append( { "code": symptom_code, "severity": slowlog_symptom_severity(symptom_code, metrics), "signal_codes": [f"slowlog.{dimension}" for dimension in symptom_dimensions], "metric_keys": SLOWLOG_SYMPTOM_METRICS[symptom_code], "assessment": slowlog_assessment(symptom_code, metrics), "evidence_refs": [ ref for ref in evidence_refs if ref.get("dimension") in symptom_dimensions ], } ) findings.append( { "source": SOURCE, "source_type": "inspection", "source_event_id": f"{prefix}:slowlog_top_digest:{digest}", "source_run_id": source_run_id(report), "org_id": report.get("dbaasOrgId"), "cluster_id": report.get("clusterId"), "sql_digest": digest, "normalized_sql": first_query(rows), "time_window_start": report.get("timeRangeAStart"), "time_window_end": report.get("timeRangeAEnd"), "observed_at": report.get("timeRangeAEnd"), "detector_code": "clinic.slowlog_top_digest", "detector_version": DETECTOR_VERSION, "symptoms": symptoms, "metric_schema_version": "clinic.slowlog_top_digest/v1", "metrics": metrics, "dimensions": { "check_name": item.get("name"), "check_type": item.get("checkType"), "hit_dimensions": hit_dimensions, "top_ranks": top_ranks, "report": context, }, "evidence_refs": evidence_refs, "raw_payload": { "report": context, "check": { "name": item.get("name"), "description": item.get("description"), "params": item.get("params"), "summary": content.get("summary"), }, "rows": rows, }, "created_by": "clinic-health-report-importer", } ) return findings def find_candidate_query(item: dict[str, Any], digest: str) -> str: results = (item.get("content") or {}).get("results") or [] if not isinstance(results, list): return "" for window in results: if not isinstance(window, dict): continue for candidate in window.get("sql_candidates") or []: if not isinstance(candidate, dict) or candidate.get("digest") != digest: continue slow_log = candidate.get("slow_log") or {} for key in ("sample_query", "query"): query = slow_log.get(key) if isinstance(query, str) and query: return query return "" def matching_cpu_windows(item: dict[str, Any], digest: str) -> list[dict[str, Any]]: results = (item.get("content") or {}).get("results") or [] if not isinstance(results, list): return [] matches: list[dict[str, Any]] = [] for window in results: if not isinstance(window, dict): continue candidates = [ candidate for candidate in window.get("sql_candidates") or [] if isinstance(candidate, dict) and candidate.get("digest") == digest ] if candidates: matches.append( { "window_id": window.get("window_id"), "rank": window.get("rank"), "node_severity": window.get("node_severity"), "decision": window.get("decision"), "node_anomaly": window.get("node_anomaly"), "topsql_summary": window.get("topsql_summary"), "sql_candidates": candidates, "links": window.get("links"), } ) return matches def build_high_cpu_findings(report: dict[str, Any], item: dict[str, Any]) -> list[dict[str, Any]]: content = item.get("content") or {} digest_summary = content.get("digest_summary") or [] if not isinstance(digest_summary, list): return [] findings: list[dict[str, Any]] = [] context = report_context(report) prefix = event_prefix(report) for summary in digest_summary: if not isinstance(summary, dict) or not summary.get("digest"): continue digest = str(summary["digest"]) windows = matching_cpu_windows(item, digest) links = summary.get("representative_links") or {} evidence_refs = [ { "type": "clinic_link", "name": name, "url": url, } for name, url in links.items() if isinstance(url, str) and url ] if summary.get("representative_window_id"): evidence_refs.append( { "type": "clinic_health_check_window", "window_id": summary.get("representative_window_id"), } ) metrics = { "affected_instances": summary.get("affected_instances"), "affected_windows": summary.get("affected_windows"), "avg_topsql_share_percent": summary.get("avg_topsql_share_percent"), "max_topsql_share_percent": summary.get("max_topsql_share_percent"), "max_cpu_percent": summary.get("max_cpu_percent"), "slow_log_count": summary.get("slow_log_count"), "total_topsql_cpu_seconds": summary.get("total_topsql_cpu_seconds"), "matching_window_count": len(windows), } findings.append( { "source": SOURCE, "source_type": "inspection", "source_event_id": f"{prefix}:sql_tikv_high_cpu:{digest}", "source_run_id": source_run_id(report), "org_id": report.get("dbaasOrgId"), "cluster_id": report.get("clusterId"), "sql_digest": digest, "normalized_sql": find_candidate_query(item, digest), "time_window_start": iso_from_epoch(summary.get("first_seen")) or report.get("timeRangeAStart"), "time_window_end": iso_from_epoch(summary.get("last_seen")) or report.get("timeRangeAEnd"), "observed_at": iso_from_epoch(summary.get("last_seen")) or report.get("timeRangeAEnd"), "detector_code": "clinic.sql_tikv_high_cpu", "detector_version": DETECTOR_VERSION, "symptoms": [ { "code": "tikv_cpu_hotspot", "severity": normalize_severity(summary.get("max_severity")), "signal_codes": ["tikv.high_cpu", "topsql.cpu_attribution"], "metric_keys": list(metrics), "assessment": { "method": "external", "summary": "Severity is the max severity reported by the Clinic sql_tikv_high_cpu result.", "rule_code": "clinic.sql_tikv_high_cpu", "rule_version": DETECTOR_VERSION, }, "evidence_refs": evidence_refs, } ], "metric_schema_version": "clinic.sql_tikv_high_cpu/v1", "metrics": metrics, "dimensions": { "check_name": item.get("name"), "check_type": item.get("checkType"), "instances": summary.get("instances") or [], "representative_window_id": summary.get("representative_window_id"), "report": context, "data_quality": content.get("summary", {}).get("data_quality"), }, "evidence_refs": evidence_refs, "raw_payload": { "report": context, "check": { "name": item.get("name"), "description": item.get("description"), "params": item.get("params"), "summary": content.get("summary"), }, "digest_summary": summary, "windows": windows, }, "created_by": "clinic-health-report-importer", } ) return findings def build_findings(report: dict[str, Any]) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] for item in report.get("items") or []: if not isinstance(item, dict) or item.get("checkType") != "sql": continue if item.get("name") == "slowlog_top_digest": findings.extend(build_slowlog_findings(report, item)) elif item.get("name") == "sql_tikv_high_cpu": findings.extend(build_high_cpu_findings(report, item)) return findings def import_report(path: Path) -> dict[str, Any]: report = json.loads(path.read_text(encoding="utf-8")) init_db() findings = build_findings(report) sql_checks = [ item for item in report.get("items") or [] if isinstance(item, dict) and item.get("checkType") == "sql" ] run_status, run_response = create_finding_run( { "source": SOURCE, "source_type": "inspection", "source_run_id": source_run_id(report), "org_id": report.get("dbaasOrgId"), "cluster_id": report.get("clusterId"), "time_window_start": report.get("timeRangeAStart"), "time_window_end": report.get("timeRangeAEnd"), "status": report.get("status") or "success", "detector_version": DETECTOR_VERSION, "expected_detector_count": len(sql_checks), "completed_detector_count": len(sql_checks), "metadata": { "strategy_uuid": report.get("strategyUuid"), "report_created_at": report.get("createdAt"), "report_path": str(path), "fixture_derivation": "none", }, } ) if run_status not in {200, 201}: raise RuntimeError(f"failed to create finding run: {run_status} {run_response}") by_check: dict[str, int] = defaultdict(int) for payload in findings: check_name = str(payload.get("dimensions", {}).get("check_name") or "unknown") by_check[check_name] += 1 batch_status, batch_response = create_finding_batch( { "source": SOURCE, "source_type": "inspection", "source_batch_id": f"{source_run_id(report)}:import", "findings": findings, "created_by": "clinic-health-report-importer", } ) if batch_status not in {200, 201} or batch_response["summary"]["rejected"]: raise RuntimeError(f"failed to create finding batch: {batch_status} {batch_response}") return { "input": str(path), "candidate_findings": len(findings), "created": batch_response["summary"]["created"], "existing": batch_response["summary"]["duplicate"], "source_batch_id": batch_response["source_batch_id"], "finding_run_id": run_response["run"]["id"], "by_check": dict(sorted(by_check.items())), } def main() -> None: parser = argparse.ArgumentParser(description="Import SQL findings from a TiDB Clinic health check report.") parser.add_argument("report", type=Path, help="Path to TiDB Clinic Health Check Report.json") args = parser.parse_args() result = import_report(args.report) print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True)) if __name__ == "__main__": main()