from __future__ import annotations import argparse import json import math import os import sys import urllib.error import urllib.parse import urllib.request from datetime import UTC from datetime import datetime from datetime import timedelta from typing import Any from zoneinfo import ZoneInfo DEFAULT_BASE_URL = "https://clinic.pingcap.com" DEFAULT_TIMEOUT_S = 60 BEIJING_TZ = ZoneInfo("Asia/Shanghai") class MetricsQueryError(RuntimeError): pass class ClinicMetricsClient: 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_instant( self, *, cluster_id: str, promql: str, query_time: int, ) -> dict[str, Any]: return self._get( f"/data-proxy/metrics/{cluster_id}/api/v1/query", params={ "query": promql, "time": str(query_time), }, ) 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 MetricsQueryError(f"HTTP {exc.code} for {path}: {body}") from exc except urllib.error.URLError as exc: raise MetricsQueryError(f"request failed for {path}: {exc}") from exc try: data = json.loads(payload) except json.JSONDecodeError as exc: raise MetricsQueryError(f"invalid JSON response for {path}: {exc}") from exc status = str(data.get("status") or "") if status and status != "success": raise MetricsQueryError(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 MetricsQueryError( "missing Clinic API key in SKILL_CLINIC_API_KEY/CLINIC_API_KEY/INVESTIBOOK_CLINIC_API_KEY" ) def _isoformat_utc(ts: int) -> str: return datetime.fromtimestamp(ts, tz=UTC).isoformat().replace("+00:00", "Z") 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 _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 _counter_delta_value(payload: dict[str, Any]) -> int | None: for item in _result_series(payload): raw_value = item.get("value") if not isinstance(raw_value, list) or len(raw_value) < 2: continue numeric = _safe_float(raw_value[1]) if numeric is None: continue return max(0, int(round(numeric))) return None def _latest_completed_beijing_window( *, local_hour: int, local_minute: int, duration_minutes: int, ) -> tuple[int, int, str]: now_local = datetime.now(tz=BEIJING_TZ) candidate_start = now_local.replace( hour=local_hour, minute=local_minute, second=0, microsecond=0, ) candidate_end = candidate_start + timedelta(minutes=duration_minutes) if now_local < candidate_end: candidate_start = candidate_start - timedelta(days=1) candidate_end = candidate_start + timedelta(minutes=duration_minutes) return ( int(candidate_start.astimezone(UTC).timestamp()), int(candidate_end.astimezone(UTC).timestamp()), candidate_start.date().isoformat(), ) def _resolve_window(args: argparse.Namespace) -> dict[str, Any]: start_ts, end_ts, _ = _latest_completed_beijing_window( local_hour=args.local_hour, local_minute=args.local_minute, duration_minutes=args.duration_minutes, ) return { "cluster_id": args.cluster_id, "cluster_label": args.cluster_label, "window": { "start": _isoformat_utc(start_ts), "end": _isoformat_utc(end_ts), }, } def _build_reference(args: argparse.Namespace) -> dict[str, Any]: start_ts = _parse_iso8601(args.start) end_ts = _parse_iso8601(args.end) if end_ts <= start_ts: raise MetricsQueryError("end must be greater than start") window_seconds = end_ts - start_ts client = ClinicMetricsClient( api_key=_env_api_key(), base_url=args.base_url, timeout_s=args.timeout_s, ) ttl_insert_rows = _counter_delta_value( client.query_instant( cluster_id=args.cluster_id, promql=f"sum(increase(tidb_server_ttl_insert_rows[{window_seconds}s]))", query_time=end_ts, ) ) ttl_processed_expired_rows = _counter_delta_value( client.query_instant( cluster_id=args.cluster_id, promql=f"sum(increase(tidb_server_ttl_processed_expired_rows[{window_seconds}s]))", query_time=end_ts, ) ) evidence_sufficient = ttl_insert_rows is not None and ttl_processed_expired_rows is not None ttl_activity_present = bool( evidence_sufficient and ( int(ttl_insert_rows or 0) > 0 or int(ttl_processed_expired_rows or 0) > 0 ) ) return { "cluster_id": args.cluster_id, "window": { "start": _isoformat_utc(start_ts), "end": _isoformat_utc(end_ts), }, "evidence_sufficient": evidence_sufficient, "ttl_activity_present": ttl_activity_present, "signals": { "ttl_insert_rows_delta": ttl_insert_rows if evidence_sufficient else None, "ttl_processed_expired_rows_delta": ttl_processed_expired_rows if evidence_sufficient else None, }, } def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Runtime hooks for ttl-row-processing-during-sql-tail-latency-incident-window" ) subparsers = parser.add_subparsers(dest="command", required=True) resolve_window = subparsers.add_parser( "resolve-window", help="Resolve the latest completed Beijing 11:00-11:05 window", ) resolve_window.add_argument("--cluster-id", required=True) resolve_window.add_argument("--cluster-label", required=False) resolve_window.add_argument("--local-hour", type=int, required=True) resolve_window.add_argument("--local-minute", type=int, required=True) resolve_window.add_argument("--duration-minutes", type=int, required=True) resolve_window.set_defaults(handler=_resolve_window) build_reference = subparsers.add_parser( "build-reference", help="Build the live TTL reference for the resolved window", ) build_reference.add_argument("--cluster-id", required=True) build_reference.add_argument("--cluster-label", required=False) build_reference.add_argument("--start", required=True) build_reference.add_argument("--end", required=True) build_reference.add_argument("--base-url", default=DEFAULT_BASE_URL) build_reference.add_argument("--timeout-s", type=int, default=DEFAULT_TIMEOUT_S) build_reference.set_defaults(handler=_build_reference) return parser def main(argv: list[str] | None = None) -> int: parser = _build_parser() args = parser.parse_args(argv) try: output = args.handler(args) except MetricsQueryError as exc: print(str(exc), file=sys.stderr) return 1 json.dump(output, sys.stdout, ensure_ascii=False, indent=2) sys.stdout.write("\n") return 0 if __name__ == "__main__": raise SystemExit(main())