#!/usr/bin/env python3 from __future__ import annotations from typing import Any, Callable from common import clinic_metrics_query, grafana_query def _safe_float(value: Any) -> float | None: try: return float(value) except (TypeError, ValueError): return None def _extract_vector_rows(payload: dict[str, Any]) -> list[dict[str, Any]]: rows = [] for item in (((payload or {}).get("data") or {}).get("result")) or []: metric = item.get("metric") or {} value = item.get("value") or [] last_value = value[1] if len(value) >= 2 else None rows.append( { "metric": metric, "value": _safe_float(last_value), "raw_value": last_value, } ) return rows def _scalar_query(query_fn: Callable[[str], dict[str, Any]], expr: str) -> float | None: rows = _extract_vector_rows(query_fn(expr)) if not rows: return None return rows[0].get("value") def _vector_query(query_fn: Callable[[str], dict[str, Any]], expr: str, top_n: int = 10) -> list[dict[str, Any]]: rows = _extract_vector_rows(query_fn(expr)) rows.sort(key=lambda item: item.get("value") if item.get("value") is not None else float("-inf"), reverse=True) return rows[:top_n] def _infer_cluster_name(instance_rows: list[dict[str, Any]]) -> str | None: for row in instance_rows: instance = str(((row.get("metric") or {}).get("instance")) or "").strip() if not instance: continue host = instance.split(":", 1)[0] for token in ["--tikv-", "--tidb-", "--pd-"]: if token in host: return host.split(token, 1)[0] if "-" in host: return host return None def _rows_to_instance_values(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out = [] for row in rows: metric = row.get("metric") or {} instance = str(metric.get("instance") or "").strip() if not instance: continue out.append( { "instance": instance, "value": row.get("value"), } ) return out def _rows_to_type_values(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: out = [] for row in rows: metric = row.get("metric") or {} metric_type = str(metric.get("type") or "").strip() if not metric_type: continue out.append( { "type": metric_type, "value": row.get("value"), } ) return out def _build_cluster_summary( *, cluster_name: str | None, cluster_version: str | None, store_rows: list[dict[str, Any]], store_count: float | None, tidb_count: float | None, pd_count: float | None, region_count: float | None, leader_count: float | None, used_bytes: float | None, available_bytes: float | None, capacity_bytes: float | None, qps_1m: float | None, qps_5m: float | None, sql_type_rows: list[dict[str, Any]], command_type_rows: list[dict[str, Any]], ) -> dict[str, Any]: return { "cluster_name": cluster_name or _infer_cluster_name(store_rows), "cluster_version": cluster_version, "store_count": int(store_count) if store_count is not None else None, "tidb_count": int(tidb_count) if tidb_count is not None else None, "pd_count": int(pd_count) if pd_count is not None else None, "region_count": int(region_count) if region_count is not None else None, "leader_count": int(leader_count) if leader_count is not None else None, "store_used_bytes": used_bytes, "store_available_bytes": available_bytes, "store_capacity_bytes": capacity_bytes, "qps_1m": qps_1m, "qps_5m": qps_5m, "stores_used_by_instance": _rows_to_instance_values(store_rows), "top_sql_types_5m": _rows_to_type_values(sql_type_rows), "top_tikv_commands_5m": _rows_to_type_values(command_type_rows), } def summarize_clinic_cluster(session, clinic_context: dict[str, Any], cluster_meta: dict[str, Any] | None = None) -> dict[str, Any]: def query(expr: str) -> dict[str, Any]: return clinic_metrics_query(session, clinic_context, expr, step="60s") store_rows = _vector_query(query, 'sum(tikv_store_size_bytes{type="used"}) by (instance)', top_n=20) store_count = _scalar_query(query, 'count(count by (instance) (tikv_store_size_bytes{type="used"}))') tidb_count = _scalar_query(query, 'count(count by (instance) (tidb_server_connections))') pd_count = _scalar_query(query, 'count(count by (instance) (pd_cluster_status))') region_count = _scalar_query(query, 'sum(tikv_raftstore_region_count{type="region"})') leader_count = _scalar_query(query, 'sum(tikv_raftstore_region_count{type="leader"})') used_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="used"})') available_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="available"})') capacity_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="capacity"})') qps_1m = _scalar_query(query, 'sum(rate(tidb_executor_statement_total[1m]))') qps_5m = _scalar_query(query, 'sum(rate(tidb_executor_statement_total[5m]))') sql_type_rows = _vector_query(query, 'topk(10, sum(rate(tidb_executor_statement_total[5m])) by (type))', top_n=10) command_type_rows = _vector_query(query, 'topk(10, sum(rate(tikv_storage_command_total[5m])) by (type))', top_n=10) return _build_cluster_summary( cluster_name=str((cluster_meta or {}).get("name") or "").strip() or None, cluster_version=str((cluster_meta or {}).get("version") or "").strip() or None, store_rows=store_rows, store_count=store_count, tidb_count=tidb_count, pd_count=pd_count, region_count=region_count, leader_count=leader_count, used_bytes=used_bytes, available_bytes=available_bytes, capacity_bytes=capacity_bytes, qps_1m=qps_1m, qps_5m=qps_5m, sql_type_rows=sql_type_rows, command_type_rows=command_type_rows, ) def summarize_grafana_cluster( session, *, datasource_id: str | int, time_ts: int | None = None, ) -> dict[str, Any]: def query(expr: str) -> dict[str, Any]: return grafana_query(session, datasource_id, expr, time_ts=time_ts) store_rows = _vector_query(query, 'sum(tikv_store_size_bytes{type="used"}) by (instance)', top_n=20) store_count = _scalar_query(query, 'count(count by (instance) (tikv_store_size_bytes{type="used"}))') tidb_count = _scalar_query(query, 'count(count by (instance) (tidb_server_connections))') pd_count = _scalar_query(query, 'count(count by (instance) (pd_cluster_status))') region_count = _scalar_query(query, 'sum(tikv_raftstore_region_count{type="region"})') leader_count = _scalar_query(query, 'sum(tikv_raftstore_region_count{type="leader"})') used_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="used"})') available_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="available"})') capacity_bytes = _scalar_query(query, 'sum(tikv_store_size_bytes{type="capacity"})') qps_1m = _scalar_query(query, 'sum(rate(tidb_executor_statement_total[1m]))') qps_5m = _scalar_query(query, 'sum(rate(tidb_executor_statement_total[5m]))') sql_type_rows = _vector_query(query, 'topk(10, sum(rate(tidb_executor_statement_total[5m])) by (type))', top_n=10) command_type_rows = _vector_query(query, 'topk(10, sum(rate(tikv_storage_command_total[5m])) by (type))', top_n=10) return _build_cluster_summary( cluster_name=None, cluster_version=None, store_rows=store_rows, store_count=store_count, tidb_count=tidb_count, pd_count=pd_count, region_count=region_count, leader_count=leader_count, used_bytes=used_bytes, available_bytes=available_bytes, capacity_bytes=capacity_bytes, qps_1m=qps_1m, qps_5m=qps_5m, sql_type_rows=sql_type_rows, command_type_rows=command_type_rows, )