#!/usr/bin/env python3 """Read-only alertstore REST helper for Investibook agent skills. Alertstore is the single source of truth for TiDB Cloud operational alerts (ALT-* IDs). It shares the DevOps gateway, so authentication and environment selection work exactly like the builtin ``devops`` skill. This helper exposes read-only (GET) endpoints only. Lifecycle actions (ack / resolve / assign / comment) and subscription / mute-rule mutations are intentionally omitted — perform those in the alertstore Web UI or Jira, per the alert's ``attrs.owned`` flag. """ from __future__ import annotations import argparse import json import os import sys import urllib.error import urllib.parse import urllib.request from typing import Any DEFAULT_BASE_URL = "https://prod.devops.pingcap.com" # Map --env values to gateway base domains (mirrors the nutshell alertstore-api # skill). DEVOPS_BASE_URL / DEVOPS_BASE_DOMAIN override these when set. ENV_BASE_URLS = { "prod": "https://prod.devops.pingcap.com", "staging": "https://staging.devops.pingcap.com", "dev": "https://dev.devops.pingcap.com", } class AlertstoreConfigError(RuntimeError): pass def _env_first(*names: str, default: str = "") -> str: for name in names: value = os.environ.get(name) if value: return value.strip() return default def load_config(env: str | None) -> tuple[str, str]: """Resolve (base_url, api_key) for the alertstore gateway. Precedence for the base URL: explicit ``--env`` > DEVOPS_BASE_URL / DEVOPS_BASE_DOMAIN > prod default. Alertstore requires the DevOps API key; DEVOPS_API_TOKEN remains accepted as a compatibility fallback. """ if env: if env not in ENV_BASE_URLS: raise AlertstoreConfigError( f"Unknown --env value: {env} (valid: prod, staging, dev)." ) base_url = ENV_BASE_URLS[env] else: base_url = _env_first( "DEVOPS_BASE_URL", "DEVOPS_BASE_DOMAIN", default=DEFAULT_BASE_URL ) base_url = base_url.rstrip("/") api_key = _env_first("DEVOPS_API_KEY", "DEVOPS_API_TOKEN") if not api_key: raise AlertstoreConfigError( "Alertstore credentials are missing: set DEVOPS_API_KEY." ) return base_url, api_key def alertstore_request( path: str, *, env: str | None, params: dict[str, Any] | None = None ) -> Any: base_url, api_key = load_config(env) api_path = path.lstrip("/") query = urllib.parse.urlencode( {k: v for k, v in (params or {}).items() if v not in (None, "")}, doseq=True, ) url = f"{base_url}/api/v1/alertstore/{api_path}" if query: url = f"{url}?{query}" request = urllib.request.Request( url=url, headers={ "Accept": "application/json", "X-API-KEY": api_key, }, method="GET", ) with urllib.request.urlopen(request, timeout=30) as response: raw = response.read().decode("utf-8") if not raw: return {} return json.loads(raw) def get_alert(args: argparse.Namespace) -> Any: return alertstore_request(f"/alerts/{args.alert_id}", env=args.env) def get_timeline(args: argparse.Namespace) -> Any: return alertstore_request(f"/alerts/{args.alert_id}/timeline", env=args.env) def list_alerts(args: argparse.Namespace) -> Any: params: dict[str, Any] = { "state": args.state, "classification": args.classification, "region": args.region, "tenant_kind": args.tenant_kind, "assigned_to": args.assigned_to, "cluster_id": args.cluster_id, "tenant_id": args.tenant_id, "limit": args.limit, "offset": args.offset, } return alertstore_request("/alerts", env=args.env, params=params) def search_by_cluster(args: argparse.Namespace) -> Any: params = {"cluster_id": args.cluster_id, "limit": args.limit, "offset": args.offset} if args.state: params["state"] = args.state return alertstore_request("/alerts", env=args.env, params=params) def search_by_tenant(args: argparse.Namespace) -> Any: params = {"tenant_id": args.tenant_id, "limit": args.limit, "offset": args.offset} if args.state: params["state"] = args.state return alertstore_request("/alerts", env=args.env, params=params) def list_events(args: argparse.Namespace) -> Any: params: dict[str, Any] = { "alertname": args.alertname, "classification": args.classification, "cluster_id": args.cluster_id, "tenant_id": args.tenant_id, "status": args.status, "limit": args.limit, "offset": args.offset, } return alertstore_request("/events", env=args.env, params=params) def get_event(args: argparse.Namespace) -> Any: return alertstore_request(f"/events/{args.event_id}", env=args.env) def _add_env_arg(parser: argparse.ArgumentParser) -> None: parser.add_argument( "--env", choices=("prod", "staging", "dev"), default=None, help="Target gateway environment (default: DEVOPS_BASE_URL or prod)", ) def _add_page_args(parser: argparse.ArgumentParser) -> None: parser.add_argument("--limit", type=int, default=50) parser.add_argument("--offset", type=int, default=0) def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( description="Read-only alertstore REST helper (ALT-* alerts)" ) sub = parser.add_subparsers(dest="command", required=True) p_get = sub.add_parser("get-alert", help="Fetch one alert detail by ALT-* id") p_get.add_argument("--alert-id", required=True) _add_env_arg(p_get) p_get.set_defaults(func=get_alert) p_tl = sub.add_parser("get-timeline", help="Fetch the human timeline for an alert") p_tl.add_argument("--alert-id", required=True) _add_env_arg(p_tl) p_tl.set_defaults(func=get_timeline) p_list = sub.add_parser("list-alerts", help="List alerts (newest first)") p_list.add_argument("--state", choices=("open", "acked", "resolved"), default=None) p_list.add_argument( "--classification", choices=("page", "ticket", "logging"), default=None ) p_list.add_argument("--region", default=None) p_list.add_argument("--tenant-kind", choices=("internal", "external"), default=None) p_list.add_argument("--assigned-to", default=None) p_list.add_argument("--cluster-id", default=None) p_list.add_argument("--tenant-id", default=None) _add_page_args(p_list) _add_env_arg(p_list) p_list.set_defaults(func=list_alerts) p_cluster = sub.add_parser("search-by-cluster", help="List alerts for a cluster") p_cluster.add_argument("--cluster-id", required=True) p_cluster.add_argument("--state", choices=("open", "acked", "resolved"), default=None) _add_page_args(p_cluster) _add_env_arg(p_cluster) p_cluster.set_defaults(func=search_by_cluster) p_tenant = sub.add_parser("search-by-tenant", help="List alerts for a tenant") p_tenant.add_argument("--tenant-id", required=True) p_tenant.add_argument("--state", choices=("open", "acked", "resolved"), default=None) _add_page_args(p_tenant) _add_env_arg(p_tenant) p_tenant.set_defaults(func=search_by_tenant) p_events = sub.add_parser("list-events", help="List raw Alertmanager events") p_events.add_argument("--alertname", default=None) p_events.add_argument( "--classification", choices=("page", "ticket", "logging"), default=None ) p_events.add_argument("--cluster-id", default=None) p_events.add_argument("--tenant-id", default=None) p_events.add_argument("--status", choices=("firing", "resolved"), default=None) _add_page_args(p_events) _add_env_arg(p_events) p_events.set_defaults(func=list_events) p_event = sub.add_parser("get-event", help="Fetch one event by id") p_event.add_argument("--event-id", required=True) _add_env_arg(p_event) p_event.set_defaults(func=get_event) return parser def main() -> int: parser = build_parser() args = parser.parse_args() try: result = args.func(args) except AlertstoreConfigError as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace") print(f"ERROR: alertstore HTTP {exc.code}: {body}", file=sys.stderr) return 1 except urllib.error.URLError as exc: print(f"ERROR: alertstore request failed: {exc}", file=sys.stderr) return 1 print(json.dumps(result, indent=2, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())