from __future__ import annotations import json from typing import Any from urllib.parse import quote, urlencode SOURCE = "clinic_health_check" SOURCE_TYPE = "inspection" DETECTOR_CODE = "clinic.health_check.sql_digest_report" DETECTOR_VERSION = "clinic-data-proxy-v1" METRIC_SCHEMA_VERSION = "clinic.health_check.sql_digest_report/v2" CREATED_BY = "clinic-health-check-sync" GOVERNANCE_SIGNAL_ANOMALY = "anomaly" GOVERNANCE_SIGNAL_RANKING_OBSERVATION = "ranking_observation" SUPPORTED_SQL_CHECKS = ( "slow_query_hugeResourceUsage", "tablefullscan_slow_query", "slowlog_top_digest", "sql_tikv_high_duration", "sql_tikv_high_cpu", ) SUPPORTED_SQL_CHECK_SET = frozenset(SUPPORTED_SQL_CHECKS) SEVERITY_RANK = { "info": 0, "low": 1, "medium": 2, "high": 3, "critical": 4, } TOP_DIMENSIONS = { "top_by_count": ("slow_query_frequency", "slowlog_slow_count"), "top_by_total_latency": ("high_total_latency", "slowlog_total_latency_s"), "top_by_total_keys": ("large_key_scan", "slowlog_total_keys"), "top_by_process_keys": ("large_key_scan", "slowlog_process_keys"), "top_by_request_count": ("high_kv_request_count", "slowlog_request_count"), "top_by_sum_ru": ("high_total_latency", "slowlog_sum_ru"), } TOP_METRICS = { "slow_count": "slowlog_slow_count", "total_latency_s": "slowlog_total_latency_s", "total_keys": "slowlog_total_keys", "process_keys": "slowlog_process_keys", "request_count": "slowlog_request_count", "sum_ru": "slowlog_sum_ru", "max_latency_s": "slowlog_max_latency_s", } def importable_check_names(check_items: Any) -> list[str]: if not isinstance(check_items, list): return [] names = { str(item.get("name") or "").strip() for item in check_items if isinstance(item, dict) } return [name for name in SUPPORTED_SQL_CHECKS if name in names] def build_report_source_url( portal_base: str, report: dict[str, Any], portal_org_id: str | None = None, ) -> str | None: org_id = _text(report.get("portalOrgId")) or _text(portal_org_id) cluster_id = _text(report.get("clusterId")) report_id = _text(report.get("reportId")) start = _text(report.get("timeRangeAStart")) end = _text(report.get("timeRangeAEnd")) if not all((org_id, cluster_id, report_id, start, end)): return None query = urlencode({"from": start, "to": end}) return ( f"{portal_base.rstrip('/')}/orgs/{quote(org_id, safe='')}" f"/clusters/{quote(cluster_id, safe='')}/health-check/" f"{quote(report_id, safe='')}?{query}" ) def build_report_import( report: dict[str, Any], cluster: dict[str, Any], portal_base: str, ) -> dict[str, Any]: report_id = _required_text(report, "reportId") org_id = _required_text(report, "orgId") cluster_id = _required_text(report, "clusterId") created_at = _required_text(report, "createdAt") window_start = _text(report.get("timeRangeAStart")) or created_at window_end = ( _text(report.get("timeRangeAEnd")) or _text(report.get("updatedAt")) or created_at ) if cluster.get("cluster_id") and str(cluster["cluster_id"]) != cluster_id: raise ValueError("Clinic report does not belong to the requested cluster") portal_org_id = _text(report.get("portalOrgId")) or _text( cluster.get("portal_org_id") ) source_url = build_report_source_url(portal_base, report, portal_org_id) report_snapshot = { "report_id": report_id, "result_uuid": report_id, "strategy_id": _text(report.get("strategyId")), "org_id": org_id, "dbaas_org_id": org_id, "portal_org_id": portal_org_id, "cluster_id": cluster_id, "status": _text(report.get("status")) or "unknown", "created_at": created_at, "updated_at": _text(report.get("updatedAt")), "time_range_a_start": window_start, "time_range_a_end": window_end, "time_range_b_start": _text(report.get("timeRangeBStart")), "time_range_b_end": _text(report.get("timeRangeBEnd")), "source_url": source_url, } context = { "org_id": org_id, "cluster_id": cluster_id, "cluster_name": cluster.get("cluster_name"), "org_name": cluster.get("org_name"), "project_id": cluster.get("project_id"), "provider": cluster.get("provider"), "region": cluster.get("region"), "report": report_snapshot, "source_url": source_url, "created_at": created_at, "window_start": window_start, "window_end": window_end, } events: dict[str, dict[str, Any]] = {} supported_items: list[str] = [] failed_items: list[str] = [] completed_items = 0 items = report.get("items") if not isinstance(items, list): raise ValueError("Clinic report items must be an array") for item in items: if not isinstance(item, dict): continue name = _text(item.get("name")) if name not in SUPPORTED_SQL_CHECK_SET: continue supported_items.append(name) content = _object(item.get("content")) item_status = (_text(item.get("status")) or _text(content.get("status"))).lower() if item_status != "success": failed_items.append(name) continue completed_items += 1 if name == "slow_query_hugeResourceUsage": _add_extreme_slow_queries(events, context, item, content) elif name == "tablefullscan_slow_query": _add_table_full_scans(events, context, item, content) elif name == "slowlog_top_digest": _add_top_digests(events, context, item, content) elif name in {"sql_tikv_high_duration", "sql_tikv_high_cpu"}: _add_tikv_hotspots(events, context, item, content) findings: list[dict[str, Any]] = [] for digest in sorted(events): event = events[digest] event["dimensions"]["check_names"].sort() event["dimensions"]["governance_signal_class"] = ( GOVERNANCE_SIGNAL_RANKING_OBSERVATION if event["dimensions"]["check_names"] == ["slowlog_top_digest"] else GOVERNANCE_SIGNAL_ANOMALY ) event["symptoms"] = sorted( event.pop("_symptoms").values(), key=lambda symptom: (-SEVERITY_RANK[symptom["severity"]], symptom["code"]), ) if event["symptoms"]: findings.append(event) run_status = _finding_run_status( _text(report.get("status")), len(supported_items), completed_items, ) run = { "source": SOURCE, "source_type": SOURCE_TYPE, "source_run_id": report_id, "org_id": org_id, "project_id": cluster.get("project_id"), "cluster_id": cluster_id, "time_window_start": window_start, "time_window_end": window_end, "status": run_status, "detector_version": DETECTOR_VERSION, "expected_detector_count": len(supported_items), "completed_detector_count": completed_items, "started_at": window_start, "completed_at": _text(report.get("updatedAt")) or window_end, "metadata": { **report_snapshot, "report_created_at": created_at, "selected_item_names": supported_items, "failed_item_names": failed_items, "source_format": "clinic-data-proxy-health-check-report/v1", "synthetic": False, }, } return { "run": run, "findings": findings, "supported_item_names": supported_items, "failed_item_names": failed_items, "report": report_snapshot, } def _new_event(context: dict[str, Any], digest: str) -> dict[str, Any]: evidence_refs: list[dict[str, Any]] = [] if context.get("source_url"): evidence_refs.append( { "type": "clinic_health_check_report", "report_id": context["report"]["report_id"], "url": context["source_url"], } ) return { "source": SOURCE, "source_type": SOURCE_TYPE, "source_event_id": ( f"clinic:healthcheck:{context['report']['report_id']}:sql_digest:{digest}" ), "source_batch_id": f"{context['report']['report_id']}:sync-v1", "source_run_id": context["report"]["report_id"], "org_id": context["org_id"], "project_id": context.get("project_id"), "cluster_id": context["cluster_id"], "sql_digest": digest, "normalized_sql": "", "reported_at": context["created_at"], "observed_at": context["window_end"], "time_window_start": context["window_start"], "time_window_end": context["window_end"], "detector_code": DETECTOR_CODE, "detector_version": DETECTOR_VERSION, "metric_schema_version": METRIC_SCHEMA_VERSION, "metrics": {}, "dimensions": { "check_names": [], "top_digest_ranks": {}, "cluster": { "name": context.get("cluster_name"), "org_name": context.get("org_name"), "dbaas_org_id": context.get("org_id"), "provider": context.get("provider"), "region": context.get("region"), }, "report": context["report"], }, "evidence_refs": evidence_refs, "raw_payload": { "report": context["report"], "extraction": { "source": "clinic_data_proxy", "synthetic": False, }, "checks": {}, }, "created_by": CREATED_BY, "_symptoms": {}, } def _event(events: dict[str, dict[str, Any]], context: dict[str, Any], digest: Any) -> dict[str, Any] | None: key = _text(digest) if not key: return None if key not in events: events[key] = _new_event(context, key) return events[key] def _set_sql(event: dict[str, Any], value: Any) -> None: sql = _text(value) if sql and not event["normalized_sql"]: event["normalized_sql"] = sql def _add_check(event: dict[str, Any], name: str, value: dict[str, Any]) -> None: if name not in event["dimensions"]["check_names"]: event["dimensions"]["check_names"].append(name) event["raw_payload"]["checks"][name] = value def _add_metric(event: dict[str, Any], key: str, value: Any) -> None: number = _number(value) if number is None: return current = event["metrics"].get(key) event["metrics"][key] = number if not isinstance(current, (int, float)) else max(current, number) def _add_symptom( event: dict[str, Any], *, code: str, severity: str, signal_code: str, metric_keys: list[str], assessment: dict[str, Any], evidence: dict[str, Any], ) -> None: symptoms = event["_symptoms"] current = symptoms.get(code) if current is None: current = { "code": code, "severity": severity, "signal_codes": [], "metric_keys": [], "assessment": assessment, "evidence_refs": [], } symptoms[code] = current elif SEVERITY_RANK[severity] > SEVERITY_RANK[current["severity"]]: current["severity"] = severity current["assessment"] = assessment if signal_code not in current["signal_codes"]: current["signal_codes"].append(signal_code) for key in metric_keys: if key in event["metrics"] and key not in current["metric_keys"]: current["metric_keys"].append(key) if evidence not in current["evidence_refs"]: current["evidence_refs"].append(evidence) def _add_external_link(event: dict[str, Any], name: str, url: Any) -> None: link = _text(url) if not link: return evidence = {"type": "clinic_link", "name": name, "url": link} if evidence not in event["evidence_refs"]: event["evidence_refs"].append(evidence) def _add_extreme_slow_queries( events: dict[str, dict[str, Any]], context: dict[str, Any], item: dict[str, Any], content: dict[str, Any], ) -> None: for row in _objects(content.get("results")): event = _event(events, context, row.get("digest")) if event is None: continue _set_sql(event, row.get("sample_query")) metrics = { "extreme_hit_records": row.get("hit_records"), "extreme_avg_query_time_s": row.get("avg_query_time_s"), "extreme_max_query_time_s": row.get("max_query_time_s"), "extreme_hit_query_time_records": row.get("hit_query_time_records"), "extreme_avg_max_mem_bytes": row.get("avg_max_mem_b"), "extreme_max_max_mem_bytes": row.get("max_max_mem_b"), "extreme_hit_mem_records": row.get("hit_mem_records"), } for key, value in metrics.items(): _add_metric(event, key, value) details = _objects(row.get("digest_detail")) for detail in details: _add_external_link(event, "Clinic slow query", detail.get("link")) evidence = { "type": "clinic_health_check_result", "check_name": "slow_query_hugeResourceUsage", "hit_records": _number(row.get("hit_records")), "hit_query_time_records": _number(row.get("hit_query_time_records")), "hit_mem_records": _number(row.get("hit_mem_records")), } _add_symptom( event, code="high_total_latency", severity="high", signal_code="slow_query.extreme_threshold_hit", metric_keys=list(metrics), assessment=_external_assessment( "clinic.slow_query_hugeResourceUsage", "Clinic returned this digest from the Extreme Slow Query check. " "The source does not expose per-digest severity, so a threshold hit is recorded as High.", {"adapter_mapping": "extreme_hit_to_high"}, ), evidence=evidence, ) _add_check( event, "slow_query_hugeResourceUsage", { "status": _text(item.get("status")) or "success", "params": _object(content.get("params")), "result": {key: value for key, value in row.items() if key != "digest_detail"}, "evidence_count": len(details), }, ) def _add_table_full_scans( events: dict[str, dict[str, Any]], context: dict[str, Any], item: dict[str, Any], content: dict[str, Any], ) -> None: for row in _objects(content.get("results")): event = _event(events, context, row.get("digest")) if event is None: continue _set_sql(event, row.get("query")) anomaly_count = next( ( value for key, value in row.items() if str(key).startswith("anomaly_logs(") ), None, ) metrics = { "table_full_anomaly_count": anomaly_count, "table_full_total_scan_rows": row.get("total_scanned_rows"), "table_full_slow_count": row.get("slow_count"), "table_full_max_scanned_rows": row.get("max_scanned_rows"), "table_full_p50_scanned_rows": row.get("p50_scanned_rows"), "table_full_p95_scanned_rows": row.get("p95_scanned_rows"), "table_full_p50_latency_s": row.get("p50_latency_s"), "table_full_p95_latency_s": row.get("p95_latency_s"), "table_full_total_latency_s": row.get("total_latency_s"), } for key, value in metrics.items(): _add_metric(event, key, value) evidence = { "type": "clinic_health_check_result", "check_name": "tablefullscan_slow_query", "total_scan_rows": _number(row.get("total_scanned_rows")), "slow_count": _number(row.get("slow_count")), } _add_symptom( event, code="large_key_scan", severity="high", signal_code="slow_query.table_full_scan", metric_keys=list(metrics), assessment=_external_assessment( "clinic.tablefullscan_slow_query", "Clinic returned this digest from the TableFullScan anomaly results. " "The adapter records the detector hit as High.", {"adapter_mapping": "table_full_scan_hit_to_high"}, ), evidence=evidence, ) samples = _objects(row.get("samples")) _add_check( event, "tablefullscan_slow_query", { "status": _text(item.get("status")) or "success", "params": _object(content.get("params")), "result": {key: value for key, value in row.items() if key != "samples"}, "sample": samples[0] if samples else None, "sample_count": len(samples), }, ) def _add_top_digests( events: dict[str, dict[str, Any]], context: dict[str, Any], item: dict[str, Any], content: dict[str, Any], ) -> None: results = _object(content.get("results")) for dimension, (symptom_code, primary_metric) in TOP_DIMENSIONS.items(): for row in _objects(results.get(dimension)): event = _event(events, context, row.get("digest")) if event is None: continue _set_sql(event, row.get("query")) rank = _number(row.get("rank")) event["dimensions"]["top_digest_ranks"][dimension] = rank metric_keys: list[str] = [] for source_key, metric_key in TOP_METRICS.items(): value = row.get(source_key) _add_metric(event, metric_key, value) if _number(value) is not None: metric_keys.append(metric_key) evidence = { "type": "clinic_health_check_result", "check_name": "slowlog_top_digest", "dimension": dimension, "rank": rank, } _add_symptom( event, code=symptom_code, severity="medium", signal_code=f"slowlog.{dimension}", metric_keys=[primary_metric, *metric_keys], assessment=_external_assessment( f"clinic.slowlog_top_digest.{dimension}", "Clinic returned this digest in a configured Top-N dimension. " "Ranking-only signals do not carry source severity, so the adapter records Medium.", {"adapter_mapping": "top_n_hit_to_medium", "dimension": dimension}, ), evidence=evidence, ) check = event["raw_payload"]["checks"].setdefault( "slowlog_top_digest", { "status": _text(item.get("status")) or "success", "params": _object(content.get("params")), "dimensions": {}, }, ) check["dimensions"][dimension] = row if "slowlog_top_digest" not in event["dimensions"]["check_names"]: event["dimensions"]["check_names"].append("slowlog_top_digest") def _add_tikv_hotspots( events: dict[str, dict[str, Any]], context: dict[str, Any], item: dict[str, Any], content: dict[str, Any], ) -> None: check_name = _text(item.get("name")) symptom_code = "tikv_cpu_hotspot" if check_name == "sql_tikv_high_cpu" else "high_total_latency" windows = _objects(content.get("results")) for summary in _objects(content.get("digest_summary")): event = _event(events, context, summary.get("digest")) if event is None: continue digest = _text(summary.get("digest")) matching_windows, candidate = _hotspot_evidence(windows, digest) slow_log = _object(candidate.get("slow_log")) if candidate else {} _set_sql(event, slow_log.get("query") or slow_log.get("sample_query")) links = _object(summary.get("representative_links")) if candidate: links = {**links, **_object(candidate.get("links"))} for link_name, link in links.items(): _add_external_link(event, f"Clinic {link_name.replace('_', ' ')}", link) metric_prefix = check_name metrics = { f"{metric_prefix}_affected_windows": summary.get("affected_windows"), f"{metric_prefix}_affected_instances": summary.get("affected_instances"), f"{metric_prefix}_total_topsql_cpu_seconds": summary.get("total_topsql_cpu_seconds"), f"{metric_prefix}_avg_topsql_share_percent": summary.get("avg_topsql_share_percent"), f"{metric_prefix}_max_topsql_share_percent": summary.get("max_topsql_share_percent"), f"{metric_prefix}_slow_log_count": summary.get("slow_log_count"), f"{metric_prefix}_matching_window_count": len(matching_windows), ( "tikv_cpu_max_percent" if check_name == "sql_tikv_high_cpu" else "tikv_duration_max_ms" ): ( summary.get("max_cpu_percent") if check_name == "sql_tikv_high_cpu" else summary.get("max_duration_ms") ), } for key, value in metrics.items(): _add_metric(event, key, value) source_severity = _text(summary.get("max_severity")) severity = _normalize_source_severity(source_severity) evidence = { "type": "clinic_health_check_result", "check_name": check_name, "source_severity": source_severity, "affected_windows": _number(summary.get("affected_windows")), "affected_instances": _number(summary.get("affected_instances")), } _add_symptom( event, code=symptom_code, severity=severity, signal_code=( "tikv.high_cpu.topsql_attribution" if check_name == "sql_tikv_high_cpu" else "tikv.high_duration.topsql_attribution" ), metric_keys=list(metrics), assessment=_external_assessment( f"clinic.{check_name}", f"Severity is mapped directly from Clinic {check_name} Max Severity: " f"{source_severity or 'not provided'}.", {"source_severity": source_severity}, ), evidence=evidence, ) _add_check( event, check_name, { "status": _text(item.get("status")) or "success", "params": _object(content.get("params")), "digest_summary": summary, "representative_window": matching_windows[0] if matching_windows else None, "matching_window_count": len(matching_windows), }, ) def _hotspot_evidence( windows: list[dict[str, Any]], digest: str ) -> tuple[list[dict[str, Any]], dict[str, Any]]: matching: list[dict[str, Any]] = [] first_candidate: dict[str, Any] = {} for window in windows: candidate = next( ( value for value in _objects(window.get("sql_candidates")) if _text(value.get("digest")) == digest ), None, ) if candidate is None: continue if not first_candidate: first_candidate = candidate matching.append( { "window_id": window.get("window_id"), "decision": window.get("decision"), "node_severity": window.get("node_severity"), "node_anomaly": window.get("node_anomaly"), "candidate": candidate, } ) return matching, first_candidate def _external_assessment( rule_code: str, summary: str, parameters: dict[str, Any], ) -> dict[str, Any]: return { "method": "external", "summary": summary, "rule_code": rule_code, "rule_version": DETECTOR_VERSION, "policy_id": "clinic-health-check-report", "parameters": parameters, } def _finding_run_status(report_status: str, expected: int, completed: int) -> str: status = report_status.lower() if status in {"pending", "running"}: return "running" if status == "failed": return "failed" if status in {"partial", "partial_failed"} or completed < expected: return "partial" return "success" def _normalize_source_severity(value: str) -> str: return { "critical": "critical", "major": "high", "high": "high", "warning": "medium", "medium": "medium", "low": "low", "info": "info", }.get(value.lower(), "medium") def _required_text(value: dict[str, Any], key: str) -> str: result = _text(value.get(key)) if not result: raise ValueError(f"Clinic report {key} is required") return result def _text(value: Any) -> str: return str(value).strip() if value is not None else "" def _object(value: Any) -> dict[str, Any]: if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) except json.JSONDecodeError: return {} return parsed if isinstance(parsed, dict) else {} return {} def _objects(value: Any) -> list[dict[str, Any]]: return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else [] def _number(value: Any) -> int | float | None: if isinstance(value, bool): return None if isinstance(value, (int, float)): return value if isinstance(value, str): try: number = float(value.replace(",", "").strip()) except ValueError: return None return int(number) if number.is_integer() else number return None