"""Shared implementation for SQL TiKV hotspot health check items.""" import json import logging import math import os import re import time import uuid from dataclasses import asdict, dataclass from datetime import UTC, datetime from typing import Any from urllib.parse import quote from service.datasource import DataProxyQueryError, ServiceDataSource from service.metrics_client import MetricSeries, MetricsDataProxyClient, MetricsQueryError from service.s3_uploader import upload_result_to_s3 logger = logging.getLogger(__name__) DEFAULT_CLINIC_BASE_URL = "https://clinic.pingcap.com/portal/dashboard/cloud/ngm.html" DEFAULT_CLINIC_WEB_HOST = "https://clinic.pingcap.com" DEFAULT_TIKV_INSTANCE_REGEX = "db-tikv-.*" DEFAULT_SHARED_SCOPE_TIKV_INSTANCE_REGEX = ".*-tikv-.*" @dataclass(frozen=True, slots=True) class HotspotParams: metric_step_seconds: int = 120 metric_rate_window: str = "2m" window_min_duration_seconds: int = 180 window_merge_gap_seconds: int = 180 window_expand_before_seconds: int = 120 window_expand_after_seconds: int = 120 attribution_window_padding_seconds: int = 120 topsql_top_n: int = 5 topsql_top_n_max: int = 20 max_anomaly_windows: int = 20 max_time_range_hours: int = 24 topsql_cpu_percent_threshold: float = 10.0 topsql_dominance_ratio_threshold: float = 2.0 critical_window_duration_seconds: int = 600 include_links: bool = True include_promql: bool = False tikv_instance_regex: str = DEFAULT_TIKV_INSTANCE_REGEX duration_threshold_ms: float = 1000.0 critical_duration_threshold_ms: float = 4000.0 grpc_type: str = "coprocessor" exclude_grpc_type_regex: str = "kv_gc" slow_log_match_policy: str = "annotate" absolute_cpu_threshold_percent: float = 75.0 relative_cpu_delta_threshold_percent: float = 10.0 min_cpu_for_relative_outlier_percent: float = 30.0 critical_cpu_threshold_percent: float = 85.0 def build_s3_key(cluster_id: str, report_id: str, item_name: str) -> str: return f"healthcheck-report/{cluster_id}_{report_id}_{item_name}.json" def validate_request_params(body: dict[str, Any], mode: str) -> None: params = resolve_hotspot_params(body, mode) start_time, end_time = parse_time_range(body, params) if start_time >= end_time: raise ValueError("timeRange.end must be later than timeRange.start") def resolve_hotspot_params(body: dict[str, Any], mode: str) -> HotspotParams: raw_params = _parse_params(body.get("params", "{}")) top_n_raw = _get_param(raw_params, "topsql_top_n", "topN", default=body.get("topN")) topsql_top_n_max = _parse_int( _get_param(raw_params, "topsql_top_n_max", default=20), default=20, min_value=1, name="params.topsql_top_n_max", ) topsql_top_n = _parse_int( top_n_raw, default=5, min_value=1, name="params.topsql_top_n", ) if topsql_top_n > topsql_top_n_max: raise ValueError("params.topsql_top_n must be <= params.topsql_top_n_max") instance_regex_default = default_tikv_instance_regex(body) raw_tikv_instance_regex = _get_param( raw_params, "tikv_instance_regex", "instance_regex", "cpu_instance_regex", default=None, ) tikv_instance_regex = str(raw_tikv_instance_regex).strip() if raw_tikv_instance_regex is not None else "" if not tikv_instance_regex: tikv_instance_regex = instance_regex_default slow_log_default = "annotate" slow_log_match_policy = str(_get_param( raw_params, "slow_log_match_policy", default=slow_log_default, )).strip().lower() if slow_log_match_policy == "require": # Legacy configs may still send "require"; slow log is enrichment-only now. slow_log_match_policy = "annotate" if slow_log_match_policy not in {"annotate", "none"}: raise ValueError("params.slow_log_match_policy must be annotate or none") return HotspotParams( metric_step_seconds=_parse_int( _get_param(raw_params, "metric_step_seconds", default=120), default=120, min_value=60, name="params.metric_step_seconds", ), metric_rate_window=str(_get_param(raw_params, "metric_rate_window", default="2m")).strip() or "2m", window_min_duration_seconds=_parse_int( _get_param(raw_params, "window_min_duration_seconds", default=180), default=180, min_value=0, name="params.window_min_duration_seconds", ), window_merge_gap_seconds=_parse_int( _get_param(raw_params, "window_merge_gap_seconds", default=180), default=180, min_value=0, name="params.window_merge_gap_seconds", ), window_expand_before_seconds=_parse_int( _get_param(raw_params, "window_expand_before_seconds", default=120), default=120, min_value=0, name="params.window_expand_before_seconds", ), window_expand_after_seconds=_parse_int( _get_param(raw_params, "window_expand_after_seconds", default=120), default=120, min_value=0, name="params.window_expand_after_seconds", ), attribution_window_padding_seconds=_parse_int( _get_param(raw_params, "attribution_window_padding_seconds", default=120), default=120, min_value=0, name="params.attribution_window_padding_seconds", ), topsql_top_n=topsql_top_n, topsql_top_n_max=topsql_top_n_max, max_anomaly_windows=_parse_int( _get_param(raw_params, "max_anomaly_windows", default=20), default=20, min_value=1, name="params.max_anomaly_windows", ), max_time_range_hours=_parse_int( _get_param(raw_params, "max_time_range_hours", default=24), default=24, min_value=1, name="params.max_time_range_hours", ), topsql_cpu_percent_threshold=_parse_float( _get_param(raw_params, "topsql_cpu_percent_threshold", default=10.0), default=10.0, min_value=0.0, name="params.topsql_cpu_percent_threshold", ), topsql_dominance_ratio_threshold=_parse_float( _get_param(raw_params, "topsql_dominance_ratio_threshold", default=2.0), default=2.0, min_value=1.0, name="params.topsql_dominance_ratio_threshold", ), critical_window_duration_seconds=_parse_int( _get_param(raw_params, "critical_window_duration_seconds", default=600), default=600, min_value=0, name="params.critical_window_duration_seconds", ), include_links=_parse_bool( _get_param(raw_params, "include_links", default=True), default=True, name="params.include_links", ), include_promql=_parse_bool( _get_param(raw_params, "include_promql", default=False), default=False, name="params.include_promql", ), tikv_instance_regex=tikv_instance_regex, duration_threshold_ms=_parse_float( _get_param(raw_params, "duration_threshold_ms", default=1000.0), default=1000.0, min_value=0.0, name="params.duration_threshold_ms", ), critical_duration_threshold_ms=_parse_float( _get_param(raw_params, "critical_duration_threshold_ms", default=4000.0), default=4000.0, min_value=0.0, name="params.critical_duration_threshold_ms", ), grpc_type=str(_get_param(raw_params, "grpc_type", default="coprocessor")).strip() or "coprocessor", exclude_grpc_type_regex=str(_get_param( raw_params, "exclude_grpc_type_regex", default="kv_gc", )).strip(), slow_log_match_policy=slow_log_match_policy, absolute_cpu_threshold_percent=_parse_float( _get_param(raw_params, "absolute_cpu_threshold_percent", default=75.0), default=75.0, min_value=0.0, name="params.absolute_cpu_threshold_percent", ), relative_cpu_delta_threshold_percent=_parse_float( _get_param(raw_params, "relative_cpu_delta_threshold_percent", default=10.0), default=10.0, min_value=0.0, name="params.relative_cpu_delta_threshold_percent", ), min_cpu_for_relative_outlier_percent=_parse_float( _get_param(raw_params, "min_cpu_for_relative_outlier_percent", default=30.0), default=30.0, min_value=0.0, name="params.min_cpu_for_relative_outlier_percent", ), critical_cpu_threshold_percent=_parse_float( _get_param(raw_params, "critical_cpu_threshold_percent", default=85.0), default=85.0, min_value=0.0, name="params.critical_cpu_threshold_percent", ), ) def parse_time_range(body: dict[str, Any], params: HotspotParams) -> tuple[datetime, datetime]: time_range = body["timeRange"] start_time = parse_iso_time(time_range["start"]) end_time = parse_iso_time(time_range["end"]) if start_time >= end_time: raise ValueError("timeRange.end must be later than timeRange.start") max_seconds = params.max_time_range_hours * 3600 if (end_time - start_time).total_seconds() > max_seconds: raise ValueError(f"timeRange duration must be <= {params.max_time_range_hours} hours") return start_time, end_time def public_params(params: HotspotParams, mode: str) -> dict[str, Any]: # Keep report/UI-facing params intentionally small. The handler still accepts # the internal knobs above for compatibility and emergency tuning. common_keys = [ "window_min_duration_seconds", "critical_window_duration_seconds", "topsql_cpu_percent_threshold", "topsql_dominance_ratio_threshold", ] mode_keys = ( [ "duration_threshold_ms", "critical_duration_threshold_ms", ] if mode == "duration" else [ "absolute_cpu_threshold_percent", "relative_cpu_delta_threshold_percent", "critical_cpu_threshold_percent", ] ) params_dict = asdict(params) return {key: params_dict[key] for key in [*mode_keys, *common_keys] if key in params_dict} @dataclass(frozen=True, slots=True) class HotspotRuntimeProfile: """Runtime routing for cluster-scoped and shared-scope data sources.""" cluster_id: str core_metrics_cluster_id: str topsql_cluster_id: str deploy_type: str shared_scope: bool def resolve_runtime_profile(cluster_id: str, body: dict[str, Any]) -> HotspotRuntimeProfile: deploy_type = str(body.get("deployType") or _lookup_meta(body, "deployType") or "dedicated") physical_cluster_id = str( body.get("physicalClusterId") or body.get("physicalClusterID") or _lookup_meta(body, "physicalClusterId") or "" ).strip() shared_scope = is_shared_scope_deploy_type(deploy_type) and bool(physical_cluster_id) return HotspotRuntimeProfile( cluster_id=cluster_id, core_metrics_cluster_id=physical_cluster_id if shared_scope else cluster_id, topsql_cluster_id=cluster_id, deploy_type=deploy_type, shared_scope=shared_scope, ) def default_tikv_instance_regex(body: dict[str, Any]) -> str: deploy_type = str(body.get("deployType") or _lookup_meta(body, "deployType") or "") physical_cluster_id = str( body.get("physicalClusterId") or body.get("physicalClusterID") or _lookup_meta(body, "physicalClusterId") or "" ).strip() if is_shared_scope_deploy_type(deploy_type) and physical_cluster_id: return DEFAULT_SHARED_SCOPE_TIKV_INSTANCE_REGEX return DEFAULT_TIKV_INSTANCE_REGEX def is_shared_scope_deploy_type(deploy_type: str) -> bool: normalized = str(deploy_type or "").strip().lower().replace("_", "-").replace(" ", "-") if not normalized: return False return any(token in normalized for token in ("premium", "byoc", "nextgen", "ng-host", "resource-pool")) def parse_iso_time(time_str: str) -> datetime: if time_str.endswith("Z"): time_str = time_str[:-1] + "+00:00" parsed = datetime.fromisoformat(time_str) if parsed.tzinfo is None: parsed = parsed.replace(tzinfo=UTC) return parsed.astimezone(UTC) def run_hotspot_inspection( body: dict[str, Any], *, mode: str, item_name: str, item_description: str, ) -> dict[str, Any]: started_at = time.monotonic() cluster_id = str(body["clusterId"]) report_id = body.get("reportId", str(uuid.uuid4())) params = resolve_hotspot_params(body, mode) start_time, end_time = parse_time_range(body, params) profile = resolve_runtime_profile(cluster_id, body) options = body.get("options", {}) if isinstance(body.get("options"), dict) else {} data_source = body.get("dataSource") vendor = str(body.get("provider") or _lookup_meta(body, "vendor") or _lookup_meta(body, "provider") or "aws") datasource_endpoint = None datasource_timeout = options.get("dataProxyTimeoutSeconds") datasource_page_size = options.get("dataProxyPageSize") metrics_base_url = options.get("metricsDataProxyBaseURL") or options.get("metrics_data_proxy_base_url") metrics_timeout = options.get("metricsDataProxyTimeoutSeconds") if isinstance(data_source, dict): datasource_endpoint = data_source.get("endpoint") datasource_timeout = data_source.get("timeoutSeconds", data_source.get("timeout_seconds", datasource_timeout)) datasource_page_size = data_source.get("pageSize", data_source.get("page_size", datasource_page_size)) metrics_base_url = data_source.get("metricsBaseURL", data_source.get("metrics_base_url", metrics_base_url)) metrics_timeout = data_source.get("metricsTimeoutSeconds", data_source.get("metrics_timeout_seconds", metrics_timeout)) elif data_source: logger.info("Ignoring legacy S3 dataSource value for cluster=%s", cluster_id) core_metrics_client = MetricsDataProxyClient( cluster_id=profile.core_metrics_cluster_id, base_url=metrics_base_url, timeout_seconds=metrics_timeout, ) topsql_metrics_client = MetricsDataProxyClient( cluster_id=profile.topsql_cluster_id, base_url=metrics_base_url, timeout_seconds=metrics_timeout, ) datasource = ServiceDataSource( cluster_id=cluster_id, endpoint=datasource_endpoint, timeout_seconds=datasource_timeout, page_size=datasource_page_size, ) link_builder = ClinicLinkBuilder(cluster_id, body, options) link_builder.include_links = params.include_links anomaly_promql = ( build_duration_promql(params, profile.core_metrics_cluster_id) if mode == "duration" else build_cpu_promql(params, profile.core_metrics_cluster_id) ) data_quality: dict[str, Any] = { "core_metrics_available": True, "topsql_available": True, "slow_log_available": True, "truncated": False, "truncated_windows": 0, } inspection_error: Exception | None = None windows: list[dict[str, Any]] = [] results: list[dict[str, Any]] = [] try: datasource.initialize() metric_series = core_metrics_client.range_matrix( anomaly_promql, start=start_time, end=end_time, step_seconds=params.metric_step_seconds, ) if mode == "duration": windows = detect_duration_windows(metric_series, params, start_time.timestamp(), end_time.timestamp()) else: windows = detect_cpu_windows(metric_series, params, start_time.timestamp(), end_time.timestamp()) windows, truncated_count = truncate_windows(windows, params, mode) if truncated_count: data_quality["truncated"] = True data_quality["truncated_windows"] = truncated_count logger.info( "Detected TiKV hotspot windows cluster=%s item=%s windows=%d truncated=%d", cluster_id, item_name, len(windows), truncated_count, ) for index, window in enumerate(windows, start=1): window_result = build_window_result( cluster_id=cluster_id, mode=mode, window=window, params=params, metrics_client=topsql_metrics_client, datasource=datasource, vendor=vendor, range_start_ts=start_time.timestamp(), range_end_ts=end_time.timestamp(), link_builder=link_builder, data_quality=data_quality, ) window_result["rank"] = index results.append(window_result) except (MetricsQueryError, DataProxyQueryError, ValueError) as exc: inspection_error = exc data_quality["core_metrics_available"] = False if isinstance(exc, MetricsQueryError) else data_quality["core_metrics_available"] logger.error( "TiKV hotspot inspection failed cluster=%s item=%s report_id=%s error=%s", cluster_id, item_name, report_id, exc, ) finally: datasource.close() digest_summary = build_digest_summary(mode, results) summary = summarize_results(results, data_quality, digest_summary) result: dict[str, Any] = { "name": item_name, "description": item_description, "status": "failed" if inspection_error else "success", "params": public_params(params, mode), "summary": summary, "digest_summary": digest_summary, "results": results, } if params.include_promql: result["promql"] = {"anomaly": anomaly_promql} if inspection_error: result["error"] = str(inspection_error) s3_key = build_s3_key(cluster_id, report_id, item_name) s3_url = upload_result_to_s3(result, s3_key) logger.info( "TiKV hotspot inspection completed cluster=%s item=%s report_id=%s status=%s windows=%d duration_s=%.3f s3_url=%s", cluster_id, item_name, report_id, result["status"], len(results), time.monotonic() - started_at, s3_url, ) return result def build_window_result( *, cluster_id: str, mode: str, window: dict[str, Any], params: HotspotParams, metrics_client: MetricsDataProxyClient, datasource: ServiceDataSource, vendor: str, range_start_ts: float, range_end_ts: float, link_builder: "ClinicLinkBuilder", data_quality: dict[str, Any], ) -> dict[str, Any]: attribution: dict[str, Any] topsql_error = "" try: attribution = fetch_topsql_attribution( metrics_client, window, params, range_start_ts=range_start_ts, range_end_ts=range_end_ts, ) except MetricsQueryError as exc: data_quality["topsql_available"] = False topsql_error = str(exc) attribution = empty_attribution(window, params, range_start_ts, range_end_ts) candidate_digests = select_topsql_candidate_digests(attribution, params) slow_log_error = "" slow_logs: dict[str, dict[str, Any]] = {} if params.slow_log_match_policy != "none" and candidate_digests: try: attr_start = datetime.fromtimestamp(attribution["window_start"], UTC) attr_end = datetime.fromtimestamp(attribution["window_end"], UTC) slow_logs = datasource.query_tikv_hotspot_slow_log_enrichment( attr_start, attr_end, candidate_digests, instance=window["instance"], vendor=vendor, limit=max(len(candidate_digests), params.topsql_top_n), include_instance_cop_candidates=False, ) except DataProxyQueryError as exc: data_quality["slow_log_available"] = False slow_log_error = str(exc) candidates = build_sql_candidates(attribution, slow_logs, params, link_builder) decision = decide_window(topsql_error, candidates) node_anomaly = build_node_anomaly(cluster_id, mode, window, params, link_builder) result: dict[str, Any] = { "window_id": f"{short_instance(window['instance'])}_{int(window['start'])}_{int(window['end'])}", "decision": decision, "node_severity": node_severity(mode, window, params), "node_anomaly": node_anomaly, "topsql_summary": build_topsql_summary(attribution), "sql_candidates": candidates, } if params.include_links: result["links"] = { "top_sql": link_builder.top_sql_url( attribution["window_start"], attribution["window_end"], window["instance"], ), } errors = {} if topsql_error: errors["topsql"] = topsql_error if slow_log_error: errors["slow_log"] = slow_log_error if errors: result["data_errors"] = errors if params.include_promql: result["promql"] = attribution.get("promql", {}) return result def detect_duration_windows( series_list: list[MetricSeries], params: HotspotParams, range_start_ts: float, range_end_ts: float, ) -> list[dict[str, Any]]: windows: list[dict[str, Any]] = [] for series in series_list: instance = str(series.metric.get("instance") or "") if not instance: continue hits = [] for sample in series.values: value = _finite_float(sample.value) if value is None or value <= params.duration_threshold_ms: continue hits.append({ "timestamp": sample.timestamp, "value": value, "rule_tags": ["duration_high"], }) windows.extend(_merge_hits(instance, hits, params, range_start_ts, range_end_ts, _duration_window_metrics)) return windows def detect_cpu_windows( series_list: list[MetricSeries], params: HotspotParams, range_start_ts: float, range_end_ts: float, ) -> list[dict[str, Any]]: p50_by_ts = _p50_by_timestamp(series_list) windows: list[dict[str, Any]] = [] for series in series_list: instance = str(series.metric.get("instance") or "") if not instance: continue hits = [] for sample in series.values: value = _finite_float(sample.value) if value is None: continue p50_value = p50_by_ts.get(sample.timestamp, value) relative_delta = value - p50_value rule_tags = [] if value > params.absolute_cpu_threshold_percent: rule_tags.append("abs_high") if ( value > params.min_cpu_for_relative_outlier_percent and relative_delta > params.relative_cpu_delta_threshold_percent ): rule_tags.append("rel_outlier") if not rule_tags: continue hits.append({ "timestamp": sample.timestamp, "value": value, "p50": p50_value, "relative_delta": relative_delta, "rule_tags": rule_tags, }) windows.extend(_merge_hits(instance, hits, params, range_start_ts, range_end_ts, _cpu_window_metrics)) return windows def truncate_windows( windows: list[dict[str, Any]], params: HotspotParams, mode: str, ) -> tuple[list[dict[str, Any]], int]: if mode == "duration": sorted_windows = sorted( windows, key=lambda item: ( item.get("max_duration_ms", 0.0), item.get("anomaly_duration_seconds", item.get("window_duration_seconds", 0)), ), reverse=True, ) else: sorted_windows = sorted( windows, key=lambda item: ( item.get("max_cpu_percent", 0.0), item.get("max_relative_delta_percent", 0.0), item.get("anomaly_duration_seconds", item.get("window_duration_seconds", 0)), ), reverse=True, ) if len(sorted_windows) <= params.max_anomaly_windows: return sorted_windows, 0 return sorted_windows[:params.max_anomaly_windows], len(sorted_windows) - params.max_anomaly_windows def fetch_topsql_attribution( metrics_client: MetricsDataProxyClient, window: dict[str, Any], params: HotspotParams, *, range_start_ts: float, range_end_ts: float, ) -> dict[str, Any]: attr_start = max(range_start_ts, window["start"] - params.attribution_window_padding_seconds) attr_end = min(range_end_ts, window["end"] + params.attribution_window_padding_seconds) if attr_end <= attr_start: attr_start = window["start"] attr_end = window["end"] interval_seconds = max(1, int(math.ceil(attr_end - attr_start))) instance_regex = topsql_instance_regex(window["instance"]) top_promql = build_topsql_topn_promql(params, instance_regex, interval_seconds) total_promql = build_topsql_total_promql(instance_regex, interval_seconds) top_vector = metrics_client.instant_vector(top_promql, query_time=attr_end) total_vector = metrics_client.instant_vector(total_promql, query_time=attr_end) top_digests = [] for metric, value in top_vector: digest = str(metric.get("sql_digest") or metric.get("digest") or "") cpu_seconds = max(0.0, value) if not digest or cpu_seconds <= 0: continue top_digests.append({ "digest": digest, "cpu_time_seconds": cpu_seconds, }) top_digests.sort(key=lambda item: item["cpu_time_seconds"], reverse=True) for rank, item in enumerate(top_digests, start=1): item["rank"] = rank topsql_total = _sum_vector_values(total_vector) topn_total = sum(item["cpu_time_seconds"] for item in top_digests) return { "window_start": attr_start, "window_end": attr_end, "window_duration_seconds": int(round(attr_end - attr_start)), "instance_regex": instance_regex, "top_digests": top_digests, "topsql_total_cpu_seconds": topsql_total, "topn_topsql_cpu_seconds": topn_total, "promql": { "topsql_topn": top_promql, "topsql_total": total_promql, }, } def empty_attribution( window: dict[str, Any], params: HotspotParams, range_start_ts: float, range_end_ts: float, ) -> dict[str, Any]: attr_start = max(range_start_ts, window["start"] - params.attribution_window_padding_seconds) attr_end = min(range_end_ts, window["end"] + params.attribution_window_padding_seconds) return { "window_start": attr_start, "window_end": attr_end, "window_duration_seconds": int(round(max(0, attr_end - attr_start))), "instance_regex": topsql_instance_regex(window["instance"]), "top_digests": [], "topsql_total_cpu_seconds": 0.0, "topn_topsql_cpu_seconds": 0.0, "promql": {}, } def select_topsql_candidate_digests( attribution: dict[str, Any], params: HotspotParams, ) -> list[str]: return [item["digest"] for item in _select_topsql_candidate_items(attribution, params)] def _select_topsql_candidate_items( attribution: dict[str, Any], params: HotspotParams, ) -> list[dict[str, Any]]: topsql_total = float(attribution.get("topsql_total_cpu_seconds") or 0.0) top_by_digest: dict[str, dict[str, Any]] = {} for item in attribution.get("top_digests", []): digest = canonical_digest(item.get("digest")) if not digest: continue cpu_seconds = float(item.get("cpu_time_seconds") or 0.0) if cpu_seconds <= 0: continue existing = top_by_digest.get(digest) if existing: existing["cpu_time_seconds"] += cpu_seconds existing["topsql_rank"] = min(existing["topsql_rank"], int(item.get("rank") or existing["topsql_rank"])) continue top_by_digest[digest] = { "digest": digest, "topsql_rank": int(item.get("rank") or len(top_by_digest) + 1), "cpu_time_seconds": cpu_seconds, } filtered_items: list[dict[str, Any]] = [] for item in top_by_digest.values(): item["digest_topsql_cpu_share_percent"] = _percent(item["cpu_time_seconds"], topsql_total) if item["digest_topsql_cpu_share_percent"] >= params.topsql_cpu_percent_threshold: filtered_items.append(item) sorted_items = sorted( filtered_items, key=lambda item: ( item["cpu_time_seconds"], item["digest_topsql_cpu_share_percent"], ), reverse=True, ) selected: list[dict[str, Any]] = [] for index, item in enumerate(sorted_items): current = dict(item) current["dominance_cutoff"] = False selected.append(current) next_item = sorted_items[index + 1] if index + 1 < len(sorted_items) else None if not next_item: break if current["cpu_time_seconds"] >= next_item["cpu_time_seconds"] * params.topsql_dominance_ratio_threshold: current["dominance_cutoff"] = True selected[-1] = current break return selected def build_sql_candidates( attribution: dict[str, Any], slow_logs: dict[str, dict[str, Any]], params: HotspotParams, link_builder: "ClinicLinkBuilder", ) -> list[dict[str, Any]]: selected_items = _select_topsql_candidate_items(attribution, params) candidates: list[dict[str, Any]] = [] for item in selected_items: digest = item["digest"] cpu_seconds = float(item.get("cpu_time_seconds") or 0.0) slow_log = slow_logs.get(digest) or slow_logs.get(canonical_digest(digest), empty_slow_log()) candidate = { "digest": digest, "rank": len(candidates) + 1, "topsql_rank": int(item.get("topsql_rank") or len(candidates) + 1), "cpu_time_seconds": round(cpu_seconds, 3), "cpu_time_minutes": round(cpu_seconds / 60.0, 3), "digest_topsql_cpu_share_percent": round(float(item.get("digest_topsql_cpu_share_percent") or 0.0), 3), "matched_topsql_cpu_percent_threshold": True, "dominance_cutoff": bool(item.get("dominance_cutoff")), "is_slow_query": bool(slow_log.get("is_slow_query")), "slow_log": slow_log, } if link_builder.include_links: candidate["links"] = link_builder.candidate_links( digest, attribution["window_start"], attribution["window_end"], attribution["instance_regex"], slow_log, ) candidates.append(candidate) return candidates def decide_window(topsql_error: str, candidates: list[dict[str, Any]]) -> str: if topsql_error: return "unknown" if candidates: return "sql_candidate" return "not_attributed" def build_node_anomaly( cluster_id: str, mode: str, window: dict[str, Any], params: HotspotParams, link_builder: "ClinicLinkBuilder", ) -> dict[str, Any]: common = { "type": "high_duration" if mode == "duration" else "high_cpu", "cluster_id": cluster_id, "instance": short_instance(window["instance"]), "source_instance": window["instance"], "window_start": int(window["start"]), "window_end": int(window["end"]), "window_duration_seconds": int(window["window_duration_seconds"]), "anomaly_duration_seconds": int(window.get("anomaly_duration_seconds", window["window_duration_seconds"])), "sample_count": int(window.get("sample_count", 0)), } if mode == "duration": common.update({ "max_duration_ms": round(float(window.get("max_duration_ms", 0.0)), 3), "avg_duration_ms": round(float(window.get("avg_duration_ms", 0.0)), 3), }) metric_name = "P99TiKVLatency_coprocessor" else: common.update({ "max_cpu_percent": round(float(window.get("max_cpu_percent", 0.0)), 3), "avg_cpu_percent": round(float(window.get("avg_cpu_percent", 0.0)), 3), "p50_cluster_cpu_percent": round(float(window.get("p50_cluster_cpu_percent", 0.0)), 3), "max_relative_delta_percent": round(float(window.get("max_relative_delta_percent", 0.0)), 3), "rule_tag": window.get("rule_tag", ""), }) metric_name = "TiKV_CPU" if params.include_links: common["links"] = { "metrics": link_builder.grafana_metric_url( metric_name, window["start"], window["end"], window["instance"], ), } return common def build_topsql_summary(attribution: dict[str, Any]) -> dict[str, Any]: topsql_total = float(attribution.get("topsql_total_cpu_seconds") or 0.0) topn_total = float(attribution.get("topn_topsql_cpu_seconds") or 0.0) return { "topsql_total_cpu_seconds": round(topsql_total, 3), "topn_topsql_cpu_seconds": round(topn_total, 3), "topn_topsql_share_percent": round(_percent(topn_total, topsql_total), 3), "others_topsql_cpu_seconds": round(max(0.0, topsql_total - topn_total), 3), "attribution_window_start": int(attribution.get("window_start", 0)), "attribution_window_end": int(attribution.get("window_end", 0)), "attribution_window_duration_seconds": int(attribution.get("window_duration_seconds", 0)), } def node_severity(mode: str, window: dict[str, Any], params: HotspotParams) -> str: severity_duration = int(window.get("anomaly_duration_seconds") or window.get("window_duration_seconds", 0)) if severity_duration >= params.critical_window_duration_seconds: return "Critical" if mode == "duration": if float(window.get("max_duration_ms", 0.0)) >= params.critical_duration_threshold_ms: return "Critical" else: if float(window.get("max_cpu_percent", 0.0)) >= params.critical_cpu_threshold_percent: return "Critical" return "Major" def build_digest_summary(mode: str, results: list[dict[str, Any]]) -> list[dict[str, Any]]: summary_by_digest: dict[str, dict[str, Any]] = {} for result in results: window_id = str(result.get("window_id") or "") severity = str(result.get("node_severity") or "Major") anomaly = result.get("node_anomaly") or {} instance = str(anomaly.get("instance") or "") window_start = int(anomaly.get("window_start") or 0) window_end = int(anomaly.get("window_end") or 0) metric_value = ( float(anomaly.get("max_duration_ms") or 0.0) if mode == "duration" else float(anomaly.get("max_cpu_percent") or 0.0) ) for candidate in result.get("sql_candidates", []): digest = canonical_digest(candidate.get("digest")) if not digest: continue entry = summary_by_digest.setdefault(digest, { "digest": digest, "_window_ids": set(), "_instances": set(), "_share_sum": 0.0, "_share_count": 0, "_representative_score": None, "max_severity": severity, "total_topsql_cpu_seconds": 0.0, "max_topsql_share_percent": 0.0, "max_duration_ms": None, "max_cpu_percent": None, "slow_log_count": 0, "first_seen": window_start, "last_seen": window_end, "representative_window_id": window_id, "representative_links": {}, }) entry["_window_ids"].add(window_id) if instance: entry["_instances"].add(instance) cpu_seconds = float(candidate.get("cpu_time_seconds") or 0.0) share_percent = float(candidate.get("digest_topsql_cpu_share_percent") or 0.0) entry["total_topsql_cpu_seconds"] += cpu_seconds entry["_share_sum"] += share_percent entry["_share_count"] += 1 entry["max_topsql_share_percent"] = max(entry["max_topsql_share_percent"], share_percent) entry["slow_log_count"] += int((candidate.get("slow_log") or {}).get("count") or 0) if window_start: entry["first_seen"] = min(int(entry["first_seen"] or window_start), window_start) if window_end: entry["last_seen"] = max(int(entry["last_seen"] or window_end), window_end) if _severity_rank(severity) > _severity_rank(str(entry.get("max_severity") or "")): entry["max_severity"] = severity if mode == "duration": entry["max_duration_ms"] = max(float(entry["max_duration_ms"] or 0.0), metric_value) else: entry["max_cpu_percent"] = max(float(entry["max_cpu_percent"] or 0.0), metric_value) representative_score = ( _severity_rank(severity), cpu_seconds, metric_value, share_percent, ) if entry["_representative_score"] is None or representative_score > entry["_representative_score"]: entry["_representative_score"] = representative_score entry["representative_window_id"] = window_id entry["representative_links"] = _representative_links(result, candidate) digest_summary = [] for entry in summary_by_digest.values(): share_count = int(entry.pop("_share_count")) share_sum = float(entry.pop("_share_sum")) window_ids = entry.pop("_window_ids") instances = entry.pop("_instances") entry.pop("_representative_score", None) entry["affected_windows"] = len(window_ids) entry["affected_instances"] = len(instances) entry["instances"] = sorted(instances) entry["total_topsql_cpu_seconds"] = round(float(entry["total_topsql_cpu_seconds"]), 3) entry["avg_topsql_share_percent"] = round(share_sum / share_count, 3) if share_count else 0.0 entry["max_topsql_share_percent"] = round(float(entry["max_topsql_share_percent"]), 3) if entry["max_duration_ms"] is not None: entry["max_duration_ms"] = round(float(entry["max_duration_ms"]), 3) if entry["max_cpu_percent"] is not None: entry["max_cpu_percent"] = round(float(entry["max_cpu_percent"]), 3) digest_summary.append(entry) digest_summary.sort( key=lambda item: ( item["affected_windows"], _severity_rank(item.get("max_severity")), item["total_topsql_cpu_seconds"], float(item.get("max_duration_ms") or 0.0), float(item.get("max_cpu_percent") or 0.0), ), reverse=True, ) return digest_summary def summarize_results( results: list[dict[str, Any]], data_quality: dict[str, Any], digest_summary: list[dict[str, Any]], ) -> dict[str, Any]: decisions = [result.get("decision") for result in results] return { "node_anomaly_windows": len(results), "sql_candidate_windows": decisions.count("sql_candidate"), "not_attributed_windows": decisions.count("not_attributed"), "unknown_windows": decisions.count("unknown"), "sql_candidates": sum(len(result.get("sql_candidates", [])) for result in results), "distinct_sql_digests": len(digest_summary), "data_quality": data_quality, } def build_duration_promql(params: HotspotParams, metrics_cluster_id: str | None = None) -> str: labels = [ f'tidb_cluster_id="{_promql_escape(str(metrics_cluster_id))}"' if metrics_cluster_id else "", 'component="tikv"', f'instance=~"{_promql_escape(params.tikv_instance_regex)}"', f'type="{_promql_escape(params.grpc_type)}"', ] labels = [label for label in labels if label] if params.exclude_grpc_type_regex: labels.append(f'type!~"{_promql_escape(params.exclude_grpc_type_regex)}"') label_text = ",".join(labels) return ( "1000 * histogram_quantile(0.99, " "sum(rate(tikv_grpc_msg_duration_seconds_bucket{" f"{label_text}" f"}}[{params.metric_rate_window}])) by (le, instance)" ")" ) def build_cpu_promql(params: HotspotParams, metrics_cluster_id: str | None = None) -> str: cluster_filter = f'tidb_cluster_id="{_promql_escape(str(metrics_cluster_id))}",' if metrics_cluster_id else "" instance_filter = f'instance=~"{_promql_escape(params.tikv_instance_regex)}"' return ( "(" f'sum(rate(process_cpu_seconds_total{{{cluster_filter}component="tikv",{instance_filter}}}[{params.metric_rate_window}])) by (instance)' " / " f'(sum(tikv_server_cpu_cores_quota{{{cluster_filter}component=~".*tikv",{instance_filter}}}) by (instance) > 0)' ") * 100" ) def build_topsql_topn_promql(params: HotspotParams, instance_regex: str, interval_seconds: int) -> str: return ( f"topk({params.topsql_top_n}, " "sum by (sql_digest) (" "sum_over_time(topsql_cpu_time_ms{" f'sql_digest!="",instance_type="tikv",instance=~"{_promql_escape(instance_regex)}"' f"}}[{interval_seconds}s])" ")" ") / 1000" ) def build_topsql_total_promql(instance_regex: str, interval_seconds: int) -> str: return ( "sum(sum_over_time(topsql_cpu_time_ms{" f'instance_type="tikv",instance=~"{_promql_escape(instance_regex)}"' f"}}[{interval_seconds}s])) / 1000" ) def topsql_instance_regex(instance: str) -> str: source = str(instance or "") short = short_instance(source) if source and source != short and ("." in source or ":" in source): return _safe_regex_literal(source) return _safe_regex_literal(short) + r"(\..*|:.*)?" def short_instance(instance: str) -> str: source = str(instance or "") if ":" in source: source = source.split(":", 1)[0] if "." in source: source = source.split(".", 1)[0] return source def full_tikv_instance(instance: str, cluster_id: str) -> str: source = str(instance or "") if ":" in source and "." in source: return source short = short_instance(source) return f"{short}.db-tikv-peer.tidb{cluster_id}.svc:20160" def empty_slow_log() -> dict[str, Any]: return { "is_slow_query": False, "count": 0, "max_query_time_s": 0.0, "avg_query_time_s": 0.0, "max_mem_bytes": 0, "avg_mem_bytes": 0.0, "max_process_keys": 0, "max_total_keys": 0, "max_request_count": 0, "cop_proc_addr_match": False, "cop_wait_addr_match": False, "cop_proc_addr_match_count": 0, "cop_wait_addr_match_count": 0, "cop_proc_max_s": 0.0, "cop_wait_max_s": 0.0, "matched_cop_proc_max_s": 0.0, "matched_cop_wait_max_s": 0.0, "sample_conn_id": "", "sample_time": None, "sample_db": "", "sample_query": "", } def canonical_digest(digest: Any) -> str: return str(digest or "").strip().lower() def _clinic_web_host(base_url: str) -> str: if "/portal/" in base_url: return base_url.split("/portal/", 1)[0] if base_url.endswith("/portal"): return base_url[: -len("/portal")] return base_url.rstrip("/") or DEFAULT_CLINIC_WEB_HOST def _dashboard_version_tag(version: str) -> str: text = str(version or "").strip() if not text or text == "unknown": return "8_5_x" if text.startswith("v"): text = text[1:] version_match = re.match(r"^(\d+)\.(\d+)(?:\.(?:\d+|x))?", text) if version_match: return f"{version_match.group(1)}_{version_match.group(2)}_x" return text.replace(".", "_") def _dashboard_version_value(version: str) -> str: text = str(version or "").strip() if not text or text == "unknown": return "v8.5.x" version_match = re.match(r"^v?(\d+)\.(\d+)(?:\.(?:\d+|x))?", text) if version_match: return f"v{version_match.group(1)}.{version_match.group(2)}.x" return text def _time_range_unix_seconds(body: dict[str, Any]) -> tuple[int | None, int | None]: time_range = body.get("timeRange") if not isinstance(time_range, dict): return None, None try: return ( int(parse_iso_time(str(time_range["start"])).timestamp()), int(parse_iso_time(str(time_range["end"])).timestamp()), ) except (KeyError, TypeError, ValueError): return None, None def _is_uint_text(value: str) -> bool: text = str(value or "").strip() return bool(text) and text.isdigit() def _first_uint_text(*values: Any) -> str: for value in values: text = str(value or "").strip() if _is_uint_text(text): return text return "" class ClinicLinkBuilder: def __init__(self, cluster_id: str, body: dict[str, Any], options: dict[str, Any]): self.cluster_id = cluster_id self.include_links = True self.base_url = ( options.get("clinicBaseURL") or body.get("clinicBaseURL") or os.environ.get("CLINIC_BASE_URL") or DEFAULT_CLINIC_BASE_URL ).rstrip("/") self.web_host = _clinic_web_host(self.base_url) self.provider = str(body.get("provider") or _lookup_meta(body, "vendor") or _lookup_meta(body, "provider") or "aws") self.region = str(options.get("region") or body.get("region") or _lookup_meta(body, "region") or "us-east-1") self.org_name = str(body.get("orgName") or _lookup_meta(body, "orgName") or "") self.org_id = str(body.get("orgId") or _lookup_meta(body, "orgId") or "") self.clinic_org_id = str( body.get("clinicOrgId") or body.get("clinicOrgID") or _lookup_meta(body, "clinicOrgId") or self.org_id ) self.project_id = _first_uint_text(body.get("projectId"), _lookup_meta(body, "projectId")) self.cluster_name = str(body.get("clusterName") or _lookup_meta(body, "clusterName") or "") self.deploy_type = str(body.get("deployType") or _lookup_meta(body, "deployType") or "dedicated") self.physical_cluster_id = str( body.get("physicalClusterId") or body.get("physicalClusterID") or _lookup_meta(body, "physicalClusterId") or "" ).strip() self.shared_scope = is_shared_scope_deploy_type(self.deploy_type) and bool(self.physical_cluster_id) self.tidb_version = str(body.get("version") or _lookup_meta(body, "version") or "unknown") self.dashboard_version = str( body.get("dashboardVersion") or _lookup_meta(body, "dashboardVersion") or self.tidb_version ) self.inspection_start_ts, self.inspection_end_ts = _time_range_unix_seconds(body) def query_prefix(self) -> str: params = [ ("provider", self.provider), ("region", self.region), ("orgName", self.org_name), ("orgId", self.org_id), ("clusterId", self.cluster_id), ("clusterName", self.cluster_name), ("deployType", self.deploy_type), ] if _is_uint_text(self.project_id): params.insert(4, ("projectId", self.project_id)) return self.base_url + "?" + "&".join( f"{quote(key)}={quote(str(value))}" for key, value in params ) def top_sql_url(self, start_ts: float, end_ts: float, instance: str) -> str: full_instance = self.top_sql_instance(instance) return ( f"{self.query_prefix()}#/topsql?" f"from={int(start_ts)}&to={int(end_ts)}&instance={quote(full_instance, safe='')}" ) def top_sql_instance(self, instance: str) -> str: if self.shared_scope: return short_instance(instance) return full_tikv_instance(instance, self.cluster_id) def slow_query_list_url(self, digest: str, start_ts: float, end_ts: float) -> str: return ( f"{self.query_prefix()}#/slow_query?" "show_internal=true" f"&digest={quote(digest, safe='')}" f"&from={int(start_ts)}&row=0&to={int(end_ts)}" ) def slow_query_detail_url(self, digest: str, conn_id: Any, timestamp: Any) -> str: ts_text = _format_timestamp_param(timestamp) return ( f"{self.query_prefix()}#/slow_query/detail?" f"digest={quote(digest, safe='')}" f"&connection_id={quote(str(conn_id or ''), safe='')}" f"×tamp={ts_text}" ) def candidate_links( self, digest: str, start_ts: float, end_ts: float, instance_or_regex: str, slow_log: dict[str, Any], ) -> dict[str, str]: links = { "top_sql": self.top_sql_url(start_ts, end_ts, _instance_from_regex(instance_or_regex)), } if slow_log.get("is_slow_query"): links["slow_query_list"] = self.slow_query_list_url(digest, start_ts, end_ts) if slow_log.get("is_slow_query") and slow_log.get("sample_conn_id") and slow_log.get("sample_time") is not None: links["slow_query_detail"] = self.slow_query_detail_url( digest, slow_log.get("sample_conn_id"), slow_log.get("sample_time"), ) return links def grafana_metric_url(self, metric_name: str, start_ts: float, end_ts: float, instance: str) -> str: metrics_cluster_id = self.physical_cluster_id if self.shared_scope else self.cluster_id if self.shared_scope and metric_name == "P99TiKVLatency_coprocessor": template = ( "{webHost}/grafana/d/e53523ea9ee7fcf4nextgenv{version}/sharedscope-tikv-details-cluster-clinic?" "orgId=1&var-tidb_cluster={clusterId}&var-instance=All&var-store=All" "&from={from}&to={to}&var-tidb_version={tidbVersion}&var-org={varOrgId}" "&var-org_type=cloud&viewPanel=55" ) elif self.shared_scope and metric_name == "TiKV_CPU": template = ( "{webHost}/grafana/d/e53523ea9ee7fcf4nextgenv{version}/sharedscope-tikv-details-cluster-clinic?" "orgId=1&var-tidb_cluster={clusterId}&var-instance=All&var-store=All" "&from={from}&to={to}&var-tidb_version={tidbVersion}&var-org={varOrgId}" "&var-org_type=cloud&viewPanel=8" ) elif metric_name == "P99TiKVLatency_coprocessor": template = ( "{webHost}/grafana/d/459a3ee585229796tidbcloudv{version}/tikv-summary-cluster-clinic?" "orgId=1&var-tidb_cluster={clusterId}&var-instance=All&var-store=All" "&from={from}&to={to}&var-tidb_version={tidbVersion}&var-org={varOrgId}" "&var-org_type=cloud&viewPanel=98" ) elif metric_name == "TiKV_CPU": template = ( "{webHost}/grafana/d/e53523ea9ee7fcf4tidbcloudv{version}/tikv-details-cluster-clinic?" "orgId=1&var-tidb_cluster={clusterId}&var-instance=All&var-store=All" "&from={from}&to={to}&var-tidb_version={tidbVersion}&var-org={varOrgId}" "&var-org_type=cloud&viewPanel=8" ) else: return "" grafana_start_ts = self.inspection_start_ts if self.inspection_start_ts is not None else int(start_ts) grafana_end_ts = self.inspection_end_ts if self.inspection_end_ts is not None else int(end_ts) replacements = { "{webHost}": self.web_host, "{version}": _dashboard_version_tag(self.dashboard_version), "{clusterId}": metrics_cluster_id, "{from}": str(int(grafana_start_ts * 1000)), "{to}": str(int(grafana_end_ts * 1000)), "{tidbVersion}": quote(_dashboard_version_value(self.dashboard_version), safe=""), "{varOrgId}": quote(self.clinic_org_id, safe=""), } link = template for key, value in replacements.items(): link = link.replace(key, value) return link def _merge_hits( instance: str, hits: list[dict[str, Any]], params: HotspotParams, range_start_ts: float, range_end_ts: float, metrics_builder, ) -> list[dict[str, Any]]: if not hits: return [] sorted_hits = sorted(hits, key=lambda item: item["timestamp"]) windows = [] current: dict[str, Any] | None = None for hit in sorted_hits: start = max(range_start_ts, hit["timestamp"] - params.window_expand_before_seconds) end = min(range_end_ts, hit["timestamp"] + params.window_expand_after_seconds) if current is None or hit["timestamp"] > current["last_hit_timestamp"] + params.window_merge_gap_seconds: if current is not None: windows.append(current) current = { "instance": instance, "start": start, "end": end, "first_hit_timestamp": hit["timestamp"], "last_hit_timestamp": hit["timestamp"], "points": [hit], } else: current["end"] = max(current["end"], end) current["last_hit_timestamp"] = hit["timestamp"] current["points"].append(hit) if current is not None: windows.append(current) built = [] for window in windows: duration = max(0, window["end"] - window["start"]) anomaly_duration = max( params.metric_step_seconds, window["last_hit_timestamp"] - window["first_hit_timestamp"] + params.metric_step_seconds, ) if anomaly_duration < params.window_min_duration_seconds: continue metrics = metrics_builder(window["points"]) metrics.update({ "instance": instance, "start": window["start"], "end": window["end"], "window_duration_seconds": int(round(duration)), "anomaly_duration_seconds": int(round(anomaly_duration)), "sample_count": len(window["points"]), }) built.append(metrics) return built def _duration_window_metrics(points: list[dict[str, Any]]) -> dict[str, Any]: values = [float(point["value"]) for point in points] return { "max_duration_ms": max(values), "avg_duration_ms": sum(values) / len(values), } def _cpu_window_metrics(points: list[dict[str, Any]]) -> dict[str, Any]: values = [float(point["value"]) for point in points] p50_values = [float(point.get("p50", point["value"])) for point in points] relative_values = [float(point.get("relative_delta", 0.0)) for point in points] tags: list[str] = [] for point in points: for tag in point.get("rule_tags", []): if tag not in tags: tags.append(tag) return { "max_cpu_percent": max(values), "avg_cpu_percent": sum(values) / len(values), "p50_cluster_cpu_percent": sum(p50_values) / len(p50_values), "max_relative_delta_percent": max(relative_values), "rule_tag": ",".join(tags), } def _p50_by_timestamp(series_list: list[MetricSeries]) -> dict[float, float]: values_by_ts: dict[float, list[float]] = {} for series in series_list: for sample in series.values: value = _finite_float(sample.value) if value is not None: values_by_ts.setdefault(sample.timestamp, []).append(value) return { timestamp: _median(values) for timestamp, values in values_by_ts.items() if values } def _median(values: list[float]) -> float: ordered = sorted(values) mid = len(ordered) // 2 if len(ordered) % 2 == 1: return ordered[mid] return (ordered[mid - 1] + ordered[mid]) / 2.0 def _sum_vector_values(vector: list[tuple[dict[str, Any], float]]) -> float: total = 0.0 for _, value in vector: if math.isfinite(value): total += max(0.0, value) return total def _parse_params(raw: Any) -> dict[str, Any]: if raw is None: return {} if isinstance(raw, str): if not raw.strip(): return {} try: parsed = json.loads(raw) except json.JSONDecodeError as exc: raise ValueError("params must be valid JSON") from exc if not isinstance(parsed, dict): raise ValueError("params must be a JSON object") return parsed if isinstance(raw, dict): return raw return {} def _get_param(params: dict[str, Any], *names: str, default: Any = None) -> Any: for name in names: if name in params: return params[name] return default def _parse_int(value: Any, *, default: int, min_value: int, name: str) -> int: if value is None: return default if isinstance(value, bool): raise ValueError(f"{name} must be an integer") if isinstance(value, int): parsed = value elif isinstance(value, float): if not value.is_integer(): raise ValueError(f"{name} must be an integer") parsed = int(value) elif isinstance(value, str): text = value.strip() if not text or not text.lstrip("-").isdigit(): raise ValueError(f"{name} must be an integer") parsed = int(text) else: raise ValueError(f"{name} must be an integer") if parsed < min_value: raise ValueError(f"{name} must be >= {min_value}") return parsed def _parse_float(value: Any, *, default: float, min_value: float, name: str) -> float: if value is None: return default if isinstance(value, bool): raise ValueError(f"{name} must be a number") try: parsed = float(value) except (TypeError, ValueError) as exc: raise ValueError(f"{name} must be a number") from exc if not math.isfinite(parsed): raise ValueError(f"{name} must be finite") if parsed < min_value: raise ValueError(f"{name} must be >= {min_value}") return parsed def _parse_bool(value: Any, *, default: bool, name: str) -> bool: if value is None: return default if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): text = value.strip().lower() if text in {"1", "true", "yes", "y", "on"}: return True if text in {"0", "false", "no", "n", "off"}: return False raise ValueError(f"{name} must be a boolean") def _finite_float(value: Any) -> float | None: try: parsed = float(value) except (TypeError, ValueError): return None if not math.isfinite(parsed): return None return parsed def _percent(numerator: float, denominator: float) -> float: if denominator <= 0: return 0.0 return numerator / denominator * 100.0 def _severity_rank(severity: Any) -> int: ranks = { "Critical": 3, "Major": 2, "Warning": 1, } return ranks.get(str(severity or ""), 0) def _representative_links(result: dict[str, Any], candidate: dict[str, Any]) -> dict[str, str]: links: dict[str, str] = {} candidate_links = candidate.get("links") or {} for key in ("top_sql", "slow_query_list", "slow_query_detail"): if candidate_links.get(key): links[key] = candidate_links[key] if "top_sql" not in links and (result.get("links") or {}).get("top_sql"): links["top_sql"] = result["links"]["top_sql"] return links def _promql_escape(value: str) -> str: return str(value).replace("\\", "\\\\").replace('"', '\\"') def _safe_regex_literal(value: str) -> str: return re.escape(value).replace(r"\-", "-") def _lookup_meta(body: dict[str, Any], key: str) -> Any: cluster_meta = body.get("clusterMeta") if isinstance(cluster_meta, dict) and key in cluster_meta: return cluster_meta[key] return body.get(key) def _format_timestamp_param(value: Any) -> str: try: return f"{float(value):.6f}" except (TypeError, ValueError): return str(value or "") def _instance_from_regex(instance_or_regex: str) -> str: value = str(instance_or_regex or "") if r"(\..*|:.*)?" in value: value = value.split(r"(\..*|:.*)?", 1)[0] return value.replace(r"\.", ".").replace(r"\:", ":").replace("\\", "")