server-manager/core/grafana_client.py

"""
Grafana API client for ServerManager.

Provides dashboard listing, alert management, datasource queries,
and annotation creation via Grafana HTTP API.
"""

from __future__ import annotations

from typing import Any

from core.logger import log


class GrafanaClient:
    """Client for interacting with a Grafana instance via its HTTP API."""

    def __init__(self, server: dict) -> None:
        """
        Initialize the Grafana client.

        Args:
            server: dict with keys: ip, port, api_token, use_ssl
        """
        self.ip: str = server["ip"]
        self.port: int = int(server["port"])
        self.api_token: str = server["api_token"]
        self.use_ssl: bool = bool(server.get("use_ssl", False))

        scheme = "https" if self.use_ssl else "http"
        self.base_url: str = f"{scheme}://{self.ip}:{self.port}"
        self.headers: dict[str, str] = {
            "Authorization": f"Bearer {self.api_token}",
            "Content-Type": "application/json",
        }
        self.timeout: int = 10

    def _get(self, path: str, params: dict | None = None) -> Any:
        """Send a GET request to the Grafana API."""
        import requests

        url = f"{self.base_url}{path}"
        log.debug("Grafana GET %s", url)
        resp = requests.get(
            url, headers=self.headers, params=params, timeout=self.timeout
        )
        resp.raise_for_status()
        return resp.json()

    def _post(self, path: str, json_data: dict | None = None) -> Any:
        """Send a POST request to the Grafana API."""
        import requests

        url = f"{self.base_url}{path}"
        log.debug("Grafana POST %s", url)
        resp = requests.post(
            url, headers=self.headers, json=json_data, timeout=self.timeout
        )
        resp.raise_for_status()
        return resp.json()

    def check_connection(self) -> bool:
        """
        Check connectivity to Grafana via GET /api/health.

        Returns:
            True if Grafana responds successfully, False otherwise.
        """
        try:
            result = self._get("/api/health")
            healthy = result.get("database", "") == "ok"
            log.info("Grafana health check: %s", "OK" if healthy else "FAIL")
            return healthy
        except Exception as exc:
            log.error("Grafana health check failed: %s", exc)
            return False

    def list_dashboards(self) -> list[dict]:
        """
        List all dashboards via GET /api/search.

        Returns:
            List of dicts with keys: uid, title, folder, url.
        """
        try:
            results = self._get("/api/search", params={"type": "dash-db"})
            dashboards = [
                {
                    "uid": d.get("uid", ""),
                    "title": d.get("title", ""),
                    "folder": d.get("folderTitle", ""),
                    "url": d.get("url", ""),
                }
                for d in results
            ]
            log.info("Grafana: found %d dashboards", len(dashboards))
            return dashboards
        except Exception as exc:
            log.error("Grafana list_dashboards failed: %s", exc)
            return []

    def get_dashboard(self, uid: str) -> dict:
        """
        Get a single dashboard by UID via GET /api/dashboards/uid/{uid}.

        Args:
            uid: Dashboard UID string.

        Returns:
            Full dashboard JSON dict, or empty dict on error.
        """
        try:
            result = self._get(f"/api/dashboards/uid/{uid}")
            log.info("Grafana: loaded dashboard '%s'", uid)
            return result
        except Exception as exc:
            log.error("Grafana get_dashboard(%s) failed: %s", uid, exc)
            return {}

    def list_alerts(self) -> list[dict]:
        """
        List provisioned alert rules via GET /api/v1/provisioning/alert-rules.

        Returns:
            List of alert rule dicts, or empty list on error.
        """
        try:
            results = self._get("/api/v1/provisioning/alert-rules")
            log.info("Grafana: found %d alert rules", len(results))
            return results
        except Exception as exc:
            log.error("Grafana list_alerts failed: %s", exc)
            return []

    def list_datasources(self) -> list[dict]:
        """
        List all datasources via GET /api/datasources.

        Returns:
            List of datasource dicts, or empty list on error.
        """
        try:
            results = self._get("/api/datasources")
            log.info("Grafana: found %d datasources", len(results))
            return results
        except Exception as exc:
            log.error("Grafana list_datasources failed: %s", exc)
            return []

    def create_annotation(self, text: str, tags: list[str] | None = None) -> dict:
        """
        Create a global annotation via POST /api/annotations.

        Args:
            text: Annotation text/description.
            tags: Optional list of tag strings.

        Returns:
            API response dict, or empty dict on error.
        """
        payload: dict[str, Any] = {"text": text}
        if tags:
            payload["tags"] = tags
        try:
            result = self._post("/api/annotations", json_data=payload)
            log.info("Grafana: created annotation id=%s", result.get("id"))
            return result
        except Exception as exc:
            log.error("Grafana create_annotation failed: %s", exc)
            return {}