#!/usr/bin/env python3 import statistics from typing import Any from basic_healthy_check import run_checks from common import ( DEFAULT_BASE_URL, build_common_arg_parser, clinic_cluster_meta, clinic_session, find_metric_definition, inject_expression_variables, parse_clinic_link, series_last_values, write_json, clinic_metrics_query, ) DEFAULT_INSTANCE_REGEX = ".*" SKEW_RATIO_WARN = 1.2 SKEW_RATIO_HIGH = 1.5 SKEW_SPREAD_WARN = 0.30 SKEW_SPREAD_HIGH = 0.50 DEPLOY_TYPE_CLUSTER_FAMILY = { "serverless": "starter-essential", "starter": "starter-essential", "essential": "starter-essential", "dedicated": "dedicated", "dedicated-cluster": "dedicated", "premium": "premium", "tiup-cluster": "premium", } def deploy_type_to_cluster_family(cluster_meta: dict[str, Any]) -> str: deploy_type = str(cluster_meta.get("deployTypeV2") or cluster_meta.get("deployType") or "").strip().lower() if deploy_type in DEPLOY_TYPE_CLUSTER_FAMILY: return DEPLOY_TYPE_CLUSTER_FAMILY[deploy_type] return "premium" def fetch_panel_last_values( session, clinic_context: dict[str, Any], dashboard_title: str, panel_title: str, expression_index: int = 0, instance: str = DEFAULT_INSTANCE_REGEX, k8s_cluster: str = "", step: str = "60s", ) -> dict[str, Any]: metric_def = find_metric_definition(dashboard_title, panel_title) expressions = metric_def["expressions"] if expression_index < 0 or expression_index >= len(expressions): raise ValueError( f"expression_index out of range for dashboard={dashboard_title!r} panel={panel_title!r}: {expression_index}" ) raw_expr = expressions[expression_index] final_expr = inject_expression_variables( raw_expr, clinic_context, variable_rules={ "k8s_cluster": k8s_cluster, "instance": instance, }, ) payload = clinic_metrics_query(session, clinic_context, final_expr, step=step) last_values = series_last_values(payload) return { "knowledge_file": metric_def["knowledge_file"], "dashboard_title": metric_def["dashboard_title"], "panel_title": metric_def["panel_title"], "raw_expression": raw_expr, "final_promql": final_expr, "last_values": last_values, } def first_numeric_last_value(items: list[dict[str, Any]]) -> float | None: for item in items: try: return float(item.get("last_value")) except (TypeError, ValueError): continue return None def summarize_store_balance(last_values: list[dict[str, Any]]) -> dict[str, Any]: pairs = [] for item in last_values: inst = item.get("instance") try: value = float(item.get("last_value")) except (TypeError, ValueError): continue if inst: pairs.append((inst, value)) pairs.sort(key=lambda x: x[1], reverse=True) if not pairs: return { "series_count": 0, "summary": {}, "top5": [], "bottom5": [], "judgement": "No data", "severity": "info", } values = [v for _, v in pairs] max_v = max(values) min_v = min(values) avg_v = statistics.mean(values) ratio = max_v / min_v if min_v else None spread_pct = (max_v - min_v) / avg_v if avg_v else None if ratio is not None and spread_pct is not None and (ratio >= SKEW_RATIO_HIGH or spread_pct >= SKEW_SPREAD_HIGH): judgement = "TiKV store used size is clearly skewed across instances." severity = "high" elif ratio is not None and spread_pct is not None and (ratio >= SKEW_RATIO_WARN or spread_pct >= SKEW_SPREAD_WARN): judgement = "TiKV store used size shows mild skew across instances." severity = "medium" else: judgement = "TiKV store used size looks roughly balanced." severity = "info" return { "series_count": len(pairs), "summary": { "max": max_v, "min": min_v, "avg": avg_v, "max_min_ratio": ratio, "spread_pct_of_avg": spread_pct, }, "top5": pairs[:5], "bottom5": pairs[-5:], "judgement": judgement, "severity": severity, } def summarize_basic_healthy(checks: list[dict[str, Any]]) -> dict[str, Any]: items = [] failed = [] for check in checks: judgement = "passed" if check["passed"] else "failed" severity = "info" if check["passed"] else "high" entry = { "check_item": check["title"], "evidence_metrics": check["queries"], "judgement": judgement, "severity": severity, "observed": check["observed"], "note": check["note"], } items.append(entry) if not check["passed"]: failed.append(entry) return { "summary": { "total": len(items), "passed": len(items) - len(failed), "failed": len(failed), }, "items": items, "overall_judgement": "passed" if not failed else "failed", "severity": "info" if not failed else "high", } def performance_overview_summary(session, clinic_context: dict[str, Any], cluster_family: str) -> dict[str, Any]: dashboard_title = ( "Test-Cluster-Performance-Overview" if cluster_family == "dedicated" else "Test-Cluster-Performance-Overview-KeyspaceName" ) qps = fetch_panel_last_values(session, clinic_context, dashboard_title, "QPS", expression_index=1) connections = fetch_panel_last_values(session, clinic_context, dashboard_title, "Connection Count", expression_index=1) db_time = fetch_panel_last_values(session, clinic_context, dashboard_title, "Database Time By SQL Phase", expression_index=0) qps_value = first_numeric_last_value(qps["last_values"]) connection_value = first_numeric_last_value(connections["last_values"]) db_time_value = first_numeric_last_value(db_time["last_values"]) items = [] if qps_value is not None: items.append( { "check_item": "Cluster QPS", "evidence_metrics": [{"name": "qps", "promql": qps["final_promql"]}], "judgement": f"QPS at end of window is {qps_value:.2f}.", "severity": "info", } ) if connection_value is not None: items.append( { "check_item": "Connection Count", "evidence_metrics": [{"name": "connection_count", "promql": connections["final_promql"]}], "judgement": f"Connection count at end of window is {connection_value:.0f}.", "severity": "info", } ) if db_time_value is not None: items.append( { "check_item": "Database Time By SQL Phase", "evidence_metrics": [{"name": "database_time", "promql": db_time["final_promql"]}], "judgement": f"Database time proxy metric at end of window is {db_time_value:.0f}.", "severity": "info", } ) return { "dashboard_title": dashboard_title, "knowledge_files": sorted({qps["knowledge_file"], connections["knowledge_file"], db_time["knowledge_file"]}), "items": items, } def tikv_risk_summary(session, clinic_context: dict[str, Any], cluster_family: str, checks: list[dict[str, Any]]) -> dict[str, Any]: dashboard_title = "Test-Cluster-TiKV-Details" if cluster_family in {"premium", "dedicated"} else "Serverless-Cluster-TiKV-Details" store_size = fetch_panel_last_values( session, clinic_context, dashboard_title, "Storage Size by Storage Class", expression_index=0, ) balance = summarize_store_balance(store_size["last_values"]) risk_items = [ { "check_item": "TiKV store used size balance", "evidence_metrics": [{"name": "tikv_store_used_size", "promql": store_size["final_promql"]}], "judgement": balance["judgement"], "severity": balance["severity"], "observed": { "series_count": balance["series_count"], "summary": balance["summary"], "top5": balance["top5"], "bottom5": balance["bottom5"], }, } ] for check in checks: if not check["id"].startswith("tikv_") or check["passed"]: continue risk_items.append( { "check_item": check["title"], "evidence_metrics": check["queries"], "judgement": "healthy preflight check failed", "severity": "high", "observed": check["observed"], } ) overall = balance["judgement"] severity = balance["severity"] return { "dashboard_title": dashboard_title, "knowledge_files": sorted({store_size["knowledge_file"]}), "items": risk_items, "overall_judgement": overall, "severity": severity, } def pd_risk_summary(checks: list[dict[str, Any]]) -> dict[str, Any]: pd_metric_def = find_metric_definition("PD", "Abnormal stores") items = [] for check in checks: if not check["id"].startswith("pd_"): continue items.append( { "check_item": check["title"], "evidence_metrics": check["queries"], "judgement": "passed" if check["passed"] else "failed", "severity": "info" if check["passed"] else "high", "observed": check["observed"], } ) failed = [item for item in items if item["judgement"] == "failed"] return { "knowledge_files": [pd_metric_def["knowledge_file"]], "items": items, "overall_judgement": "PD abnormal store checks passed." if not failed else "PD abnormal store risk detected.", "severity": "info" if not failed else "high", } def build_overall_judgement(basic_healthy: dict[str, Any], tikv_risks: dict[str, Any], pd_risks: dict[str, Any]) -> dict[str, Any]: evidence = [] severity = "info" judgement = "No strong risk signal detected from the current metrics-only diagnosis." if basic_healthy["overall_judgement"] == "failed": severity = "high" judgement = "Basic healthy checks failed. There is an immediate infrastructure-level risk." evidence.append("basic_healthy_check") if tikv_risks["severity"] == "high": severity = "high" judgement = "Metrics-only diagnosis found a clear TiKV risk signal, mainly store-size skew across instances." evidence.append("tikv_risk_summary") elif severity != "high" and tikv_risks["severity"] == "medium": severity = "medium" judgement = "Metrics-only diagnosis found a mild TiKV skew signal that deserves follow-up." evidence.append("tikv_risk_summary") if pd_risks["severity"] == "high": severity = "high" if "pd_risk_summary" not in evidence: evidence.append("pd_risk_summary") return { "judgement": judgement, "severity": severity, "evidence_modules": evidence, } def analyze_metrics_only( link: str, base_url: str, profile_name: str, ) -> dict[str, Any]: clinic_context = parse_clinic_link(link) parsed_base_url = str(clinic_context.get("base_url") or "").rstrip("/") requested_base_url = str(base_url or "").rstrip("/") if not requested_base_url or (requested_base_url == DEFAULT_BASE_URL and parsed_base_url and parsed_base_url != DEFAULT_BASE_URL): effective_base_url = parsed_base_url else: effective_base_url = requested_base_url session = clinic_session(base_url=effective_base_url, profile_name=profile_name) cluster_meta = clinic_cluster_meta(session, clinic_context) cluster_family = deploy_type_to_cluster_family(cluster_meta) checks = run_checks(session, clinic_context) healthy = summarize_basic_healthy(checks) perf_overview = performance_overview_summary(session, clinic_context, cluster_family) tikv_risks = tikv_risk_summary(session, clinic_context, cluster_family, checks) pd_risks = pd_risk_summary(checks) overall = build_overall_judgement(healthy, tikv_risks, pd_risks) knowledge_files = sorted( { *(perf_overview.get("knowledge_files") or []), *(tikv_risks.get("knowledge_files") or []), *(pd_risks.get("knowledge_files") or []), } ) return { "clinic_context": clinic_context, "cluster_meta": { "id": cluster_meta.get("id"), "name": cluster_meta.get("name"), "version": cluster_meta.get("version"), "deployType": cluster_meta.get("deployType"), "deployTypeV2": cluster_meta.get("deployTypeV2"), "cluster_family": cluster_family, "topology": cluster_meta.get("topology"), "status": cluster_meta.get("status"), }, "knowledge_files": knowledge_files, "basic_healthy_check": healthy, "performance_overview_summary": perf_overview, "tikv_risk_summary": tikv_risks, "pd_risk_summary": pd_risks, "overall_judgement": overall, "scope_note": "metrics only, no slow log / raw log / TopSQL evidence yet", } def main() -> int: parser = build_common_arg_parser( "Run a metrics-only Clinic performance diagnosis and output fixed diagnosis modules for one clinic link" ) args = parser.parse_args() write_json(analyze_metrics_only(args.link, base_url=args.base_url, profile_name=args.profile)) return 0 if __name__ == "__main__": raise SystemExit(main())