from __future__ import annotations import argparse import json import math import os import re import urllib.error import urllib.parse import urllib.request from datetime import datetime from datetime import timedelta from datetime import timezone from typing import Any DEFAULT_BASE_URL = "https://clinic.pingcap.com" DEFAULT_TIMEOUT_S = 60 WINDOW_START_HOUR_LOCAL = 15 WINDOW_START_MINUTE_LOCAL = 10 WINDOW_DURATION_MINUTES = 50 BEIJING_OFFSET = timedelta(hours=8) NODE_PATTERN = re.compile(r"\bdb(?:-[A-Za-z0-9]+)?-tikv-[A-Za-z0-9-]+\b") PER_INSTANCE_QUERY = 'sum(rate(tikv_thread_cpu_seconds_total{name=~"unified_read.*"}[1m])) by (instance)' CLUSTER_TOTAL_QUERY = 'sum(sum by (instance) (rate(tikv_thread_cpu_seconds_total{name=~"unified_read.*"}[1m])))' class ReferenceBuildError(RuntimeError): pass class ClinicClient: def __init__(self, *, api_key: str, base_url: str, timeout_s: int = DEFAULT_TIMEOUT_S) -> None: self._api_key = api_key self._base_url = base_url.rstrip("/") self._timeout_s = timeout_s def query_range( self, *, cluster_id: str, promql: str, start: int, end: int, step_seconds: int, ) -> dict[str, Any]: return self._get( f"/data-proxy/metrics/{cluster_id}/api/v1/query_range", params={ "query": promql, "start": str(start), "end": str(end), "step": str(step_seconds), }, ) def _get(self, path: str, *, params: dict[str, Any]) -> dict[str, Any]: url = f"{self._base_url}{path}?{urllib.parse.urlencode(params, doseq=True)}" request = urllib.request.Request( url, headers={"Authorization": f"Bearer {self._api_key}"}, method="GET", ) try: with urllib.request.urlopen(request, timeout=self._timeout_s) as response: payload = response.read().decode("utf-8") except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") raise ReferenceBuildError(f"HTTP {exc.code} for {path}: {body}") from exc except urllib.error.URLError as exc: raise ReferenceBuildError(f"request failed for {path}: {exc}") from exc try: data = json.loads(payload) except json.JSONDecodeError as exc: raise ReferenceBuildError(f"invalid JSON response for {path}: {exc}") from exc status = str(data.get("status") or "") if status and status != "success": raise ReferenceBuildError(f"unexpected response status for {path}: {status}") return data def _env_api_key() -> str: for key in ("SKILL_CLINIC_API_KEY", "CLINIC_API_KEY", "INVESTIBOOK_CLINIC_API_KEY"): value = os.environ.get(key, "").strip() if value: return value raise ReferenceBuildError( "missing Clinic API key in SKILL_CLINIC_API_KEY/CLINIC_API_KEY/INVESTIBOOK_CLINIC_API_KEY" ) def _parse_iso8601(value: str) -> int: normalized = value.strip() if normalized.endswith("Z"): normalized = normalized[:-1] + "+00:00" return int(datetime.fromisoformat(normalized).timestamp()) def _isoformat_utc(ts: int) -> str: return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat().replace("+00:00", "Z") def _safe_float(value: Any) -> float | None: try: numeric = float(value) except (TypeError, ValueError): return None if not math.isfinite(numeric): return None return numeric def _result_series(payload: dict[str, Any]) -> list[dict[str, Any]]: data = payload.get("data") if not isinstance(data, dict): return [] result = data.get("result") return [item for item in result if isinstance(item, dict)] if isinstance(result, list) else [] def _metric_node_name(metric: dict[str, Any]) -> str: for label_name in ("pod", "pod_name", "kubernetes_pod_name"): pod = str(metric.get(label_name) or "").strip() if NODE_PATTERN.search(pod): return pod for label_name in ("instance", "job", "exported_instance", "pod", "pod_name", "kubernetes_pod_name"): raw_value = str(metric.get(label_name) or "").strip() match = NODE_PATTERN.search(raw_value) if match: return match.group(0) return "" def _series_numeric_values(item: dict[str, Any]) -> list[float]: values = item.get("values") if not isinstance(values, list): return [] observed: list[float] = [] for row in values: if not isinstance(row, list) or len(row) < 2: continue numeric = _safe_float(row[1]) if numeric is not None: observed.append(numeric) return observed def _latest_completed_window() -> tuple[str, str]: now_utc = datetime.now(tz=timezone.utc) now_local = now_utc + BEIJING_OFFSET candidate_local = now_local.replace( hour=WINDOW_START_HOUR_LOCAL, minute=WINDOW_START_MINUTE_LOCAL, second=0, microsecond=0, ) candidate_end_local = candidate_local + timedelta(minutes=WINDOW_DURATION_MINUTES) if now_local < candidate_end_local: candidate_local -= timedelta(days=1) candidate_end_local = candidate_local + timedelta(minutes=WINDOW_DURATION_MINUTES) candidate_start_utc = candidate_local - BEIJING_OFFSET candidate_end_utc = candidate_end_local - BEIJING_OFFSET return ( candidate_start_utc.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z"), candidate_end_utc.replace(tzinfo=timezone.utc).isoformat().replace("+00:00", "Z"), ) def _build_case_input(cluster_id: str, metrics_cluster_id: str) -> dict[str, Any]: start, end = _latest_completed_window() return { "cluster_id": cluster_id, "metrics_cluster_id": metrics_cluster_id, "start": start, "end": end, } def _build_reference(*, cluster_id: str, metrics_cluster_id: str, start: str, end: str) -> dict[str, Any]: client = ClinicClient(api_key=_env_api_key(), base_url=os.environ.get("CLINIC_ENDPOINT", DEFAULT_BASE_URL)) start_ts = _parse_iso8601(start) end_ts = _parse_iso8601(end) per_instance_payload = client.query_range( cluster_id=metrics_cluster_id, promql=PER_INSTANCE_QUERY, start=start_ts, end=end_ts, step_seconds=60, ) cluster_total_payload = client.query_range( cluster_id=metrics_cluster_id, promql=CLUSTER_TOTAL_QUERY, start=start_ts, end=end_ts, step_seconds=60, ) best_node = "" best_peak: float | None = None for item in _result_series(per_instance_payload): metric = item.get("metric") if not isinstance(metric, dict): continue node = _metric_node_name(metric) values = _series_numeric_values(item) peak = max(values) if values else None if not node or peak is None: continue if best_peak is None or peak > best_peak or (peak == best_peak and node < best_node): best_node = node best_peak = peak total_peaks: list[float] = [] for item in _result_series(cluster_total_payload): total_peaks.extend(_series_numeric_values(item)) cluster_total_peak = max(total_peaks) if total_peaks else None evidence_sufficient = bool(best_node and best_peak is not None and cluster_total_peak is not None) return { "cluster_id": cluster_id, "metrics_cluster_id": metrics_cluster_id, "window": { "start": start, "end": end, }, "evidence_sufficient": evidence_sufficient, "peak_instance_node": best_node if evidence_sufficient else None, "peak_instance_read_pool_cpu_cores": round(best_peak, 4) if evidence_sufficient and best_peak is not None else None, "cluster_total_peak_read_pool_cpu_cores": round(cluster_total_peak, 4) if evidence_sufficient and cluster_total_peak is not None else None, } def main() -> None: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) prepare = subparsers.add_parser("prepare-case-input") prepare.add_argument("--cluster-id", required=True) prepare.add_argument("--metrics-cluster-id", required=True) reference = subparsers.add_parser("build-reference") reference.add_argument("--cluster-id", required=True) reference.add_argument("--metrics-cluster-id", required=True) reference.add_argument("--start", required=True) reference.add_argument("--end", required=True) args = parser.parse_args() if args.command == "prepare-case-input": payload = _build_case_input(args.cluster_id, args.metrics_cluster_id) else: payload = _build_reference( cluster_id=args.cluster_id, metrics_cluster_id=args.metrics_cluster_id, start=args.start, end=args.end, ) print(json.dumps(payload, ensure_ascii=False)) if __name__ == "__main__": main()