Grafana Prometheus

Grafana & Prometheus Integration

Implements production-grade Prometheus metrics querying, PromQL patterns, Grafana dashboard management via HTTP API, and alerting rules as code. When loaded, this skill makes the model implement PromQL queries for time-series analysis, Grafana dashboard JSON templating, alerting rules with YAML, Prometheus HTTP API calls, and Grafana datasource management. All implementations follow Prometheus and Grafana best practices: use label-based filtering, avoid high-cardinality labels, use range vectors for aggregations, implement dashboard version control, and validate PromQL syntax before deployment.

TL;DR Checklist

• Use prometheus-api-client for Prometheus HTTP API or direct HTTP calls to /api/v1/query
• Use grafana-api Python SDK or direct HTTP calls to Grafana HTTP API
• Read connection details from PROMETHEUS_URL, GRAFANA_URL, GRAFANA_API_KEY env vars
• Avoid high-cardinality labels (unique IDs, high-cardinality strings)
• Use rate() for counters, irate() for short-lived spikes
• Use sum by (label) instead of sum without (label) for clarity
• Always set time bounds (start, end, step) in range queries
• Use offset for comparison queries (week-over-week)
• Validate PromQL syntax before deployment
• Store Grafana dashboard JSON in version control (as code)
• Include __name__ and job/instance labels for metric identification

When to Use

Use this skill when:

• Querying Prometheus metrics using PromQL for time-series analysis
• Creating Grafana dashboards programmatically via API
• Managing Prometheus alerting rules as code (YAML format)
• Building automation that queries metrics for SLI/SLO calculations
• Creating recording rules for pre-aggregated metrics
• Migrating or copying dashboards between Grafana instances
• Setting up datasources programmatically
• Implementing custom alert notification channels
• Querying long-term trends via Prometheus-compatible stores (Thanos, Mimir, Cortex)
• Building dashboard templates with template variables

When NOT to Use

• For Datadog-specific monitoring — use coding-datadog-api instead
• For New Relic APM — use coding-newrelic-api instead
• When you need error tracking only — use coding-sentry-api instead
• For push-based metrics (Graphite, StatsD push model) — Prometheus is pull-based
• When you need SaaS-hosted Grafana Cloud only (can use but other skills for SaaS)

Core Workflow

1. Initialize Connections — Configure Prometheus client using PROMETHEUS_URL and Grafana client using GRAFANA_URL + GRAFANA_API_KEY from environment variables. Checkpoint: Validate connectivity with a simple query like up or Grafana GET /api/health.

2. Select Metric & Label Strategy — Define low-cardinality labels first: job, instance, env, service, version. Avoid high-cardinality: request_id, user_id (use only when necessary). Checkpoint: Every label must have known bounded cardinality — document expected value count.

3. Construct PromQL Queries — Use rate() for counters over time, sum by (group) for aggregations, topk() for ranking, histogram_quantile() for percentiles. Use range vectors with appropriate time windows. Checkpoint: Range queries always have step parameter; instant queries return single points.

4. Build Dashboard as JSON — Create Grafana dashboard JSON with panels, targets (PromQL queries), axes, legend, template variables. Use templating.list for variables. Store in version control. Checkpoint: Dashboard JSON is valid JSON, targets reference datasource by name or uid.

5. Define Alerting Rules — Write alerting rules in YAML format with expr (PromQL), for duration, labels for routing, annotations for context. Use alertmanager_config in Alertmanager. Checkpoint: Every alert has at least severity label and summary annotation.

6. Validate & Deploy — Validate PromQL using promtool check rules or API syntax check. Deploy rules via config reload or API. Checkpoint: No high-cardinality aggregations; all queries return in test queries return values.

Implementation Patterns

Pattern 1: Prometheus Client Initialization (BAD vs GOOD)

"""Prometheus and Grafana client initialization patterns.

Two primary approaches:
1. prometheus-api-client: Official Python SDK for HTTP API
2. Direct HTTP requests: Simple HTTP calls (works everywhere)
3. grafana-api: Python SDK for Grafana HTTP API

Endpoints:
- Prometheus: http://prometheus:9090/api/v1/
- Grafana: http://grafana:3000/api/
"""

from __future__ import annotations

import os
import json
import logging
import time
from typing import Any, Optional
from datetime import datetime, timedelta, timezone
from urllib.parse import urljoin

import requests

logger = logging.getLogger(__name__)


# ===================================================================
# ❌ BAD — hardcoded URLs, no validation, error handling missing
# ===================================================================

def bad_prometheus_init_bad() -> dict[str, Any]:
    """❌ BAD: Hardcoded URLs, no timeout, no validation."""
    
    # ❌ Hardcoded! Should come from config/env
    prometheus_url = "http://localhost:9090"
    grafana_url = "http://localhost:3000"
    
    # ❌ No API key handling
    # ❌ No timeout
    # ❌ No validation
    return {"prometheus_url": prometheus_url, "grafana_url": grafana_url}


# ===================================================================
# ✅ GOOD — env-based config, validation, typed errors
# ===================================================================


class PrometheusGrafanaError(Exception):
    """Base exception for Prometheus/Grafana client errors."""
    pass


class PrometheusQueryError(PrometheusGrafanaError):
    """Prometheus query returned error status."""
    pass


class GrafanaAPIError(PrometheusGrafanaError):
    """Grafana API call failed."""
    pass


class PrometheusConfig:
    """Prometheus configuration from environment variables.
    
    Environment variables:
        PROMETHEUS_URL: Base URL (http://prometheus:9090)
        PROMETHEUS_USERNAME: Optional basic auth username
        PROMETHEUS_PASSWORD: Optional basic auth password
    """
    
    def __init__(
        self,
        url: str,
        username: Optional[str] = None,
        password: Optional[str] = None,
        timeout: float = 30.0,
    ) -> None:
        if not url or not url.strip():
            raise ValueError("Prometheus URL cannot be empty")
        
        # Normalize URL (ensure no trailing slash)
        self.url = url.rstrip("/")
        self.username = username
        self.password = password
        self.timeout = timeout
    
    @classmethod
    def from_env(cls) -> "PrometheusConfig":
        """Load from environment variables."""
        url = os.environ.get("PROMETHEUS_URL", "http://localhost:9090")
        return cls(
            url=url,
            username=os.environ.get("PROMETHEUS_USERNAME"),
            password=os.environ.get("PROMETHEUS_PASSWORD"),
        )
    
    def get_api_url(self, endpoint: str) -> str:
        """Build full API URL for endpoint.
        
        Args:
            endpoint: e.g., "/api/v1/query" or "api/v1/query"
        """
        endpoint = endpoint.lstrip("/")
        if not endpoint.startswith("api/"):
            endpoint = "api/v1/" + endpoint.lstrip("/")
        return f"{self.url}/{endpoint}"


class GrafanaConfig:
    """Grafana configuration from environment variables.
    
    Environment variables:
        GRAFANA_URL: Base URL (http://grafana:3000)
        GRAFANA_API_KEY: API token or Service Account token
        GRAFANA_USER: Optional basic auth username
        GRAFANA_PASSWORD: Optional basic auth password
    """
    
    AUTH_API_KEY = "api_key"
    AUTH_BASIC = "basic"
    
    def __init__(
        self,
        url: str,
        api_key: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
        timeout: float = 30.0,
    ) -> None:
        if not url or not url.strip():
            raise ValueError("Grafana URL cannot be empty")
        
        self.url = url.rstrip("/")
        self.api_key = api_key.strip() if api_key else None
        self.username = username
        self.password = password
        self.timeout = timeout
        
        # Determine auth method
        if self.api_key:
            self.auth_method = self.AUTH_API_KEY
        elif self.username and self.password:
            self.auth_method = self.AUTH_BASIC
        else:
            self.auth_method = None  # No auth (local dev)
    
    @classmethod
    def from_env(cls) -> "GrafanaConfig":
        """Load from environment variables."""
        url = os.environ.get("GRAFANA_URL", "http://localhost:3000")
        return cls(
            url=url,
            api_key=os.environ.get("GRAFANA_API_KEY"),
            username=os.environ.get("GRAFANA_USER"),
            password=os.environ.get("GRAFANA_PASSWORD"),
        )
    
    def get_api_url(self, endpoint: str) -> str:
        """Build full API URL."""
        endpoint = endpoint.lstrip("/")
        if not endpoint.startswith("api/"):
            endpoint = "api/" + endpoint.lstrip("/")
        return f"{self.url}/{endpoint}"


class PrometheusClient:
    """Client for Prometheus HTTP API.
    
    Implements:
    - Instant queries (single point in time)
    - Range queries (time series over interval)
    - Label queries
    - Series queries
    - Rules queries
    """
    
    def __init__(self, config: PrometheusConfig) -> None:
        self._config = config
        self._session = requests.Session()
    
    def _request(
        self,
        method: str,
        endpoint: str,
        params: Optional[dict[str, Any]] = None,
        data: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        """Make HTTP request to Prometheus API."""
        url = self._config.get_api_url(endpoint)
        
        auth: Optional[tuple[str, str]] = None
        if self._config.username and self._config.password:
            auth = (self._config.username, self._config.password)
        
        try:
            response = self._session.request(
                method=method,
                url=url,
                params=params,
                json=data,
                auth=auth,
                timeout=self._config.timeout,
            )
            
            # Parse response
            try:
                result = response.json()
            except json.JSONDecodeError:
                raise PrometheusQueryError(
                    f"Prometheus returned non-JSON: {response.status_code} {response.text[:200]}"
                )
            
            # Check Prometheus status field
            if result.get("status") == "error":
                error_type = result.get("errorType", "unknown")
                error_msg = result.get("error", "unknown error")
                raise PrometheusQueryError(f"Prometheus error [{error_type}]: {error_msg}")
            
            if response.status_code >= 400:
                raise PrometheusQueryError(
                    f"Prometheus HTTP {response.status_code}: {result}"
                )
            
            return result
            
        except requests.RequestException as e:
            raise PrometheusGrafanaError(f"Network error: {e}") from e
    
    def query_instant(
        self,
        query: str,
        time: Optional[float] = None,
    ) -> dict[str, Any]:
        """Execute an instant query (single point in time).
        
        Args:
            query: PromQL query string
            time: Optional Unix timestamp (defaults to now)
        
        Returns:
            Dict with 'resultType' and 'result' from Prometheus
        """
        params: dict[str, Any] = {"query": query}
        if time is not None:
            params["time"] = time
        
        response = self._request("GET", "query", params=params)
        return response.get("data", {})
    
    def query_range(
        self,
        query: str,
        start: float,
        end: float,
        step: str,  # e.g., "1m", "5m", "1h"
    ) -> dict[str, Any]:
        """Execute a range query (time series over interval).
        
        Args:
            query: PromQL query string
            start: Start Unix timestamp
            end: End Unix timestamp
            step: Resolution step (e.g., "1m", "5m")
        
        Returns:
            Dict with 'resultType' and 'result'
        """
        params: dict[str, Any] = {
            "query": query,
            "start": start,
            "end": end,
            "step": step,
        }
        
        response = self._request("GET", "query_range", params=params)
        return response.get("data", {})
    
    def query_range_relative(
        self,
        query: str,
        duration_seconds: float,
        step: str,
    ) -> dict[str, Any]:
        """Range query with relative time (now - duration to now).
        
        Args:
            query: PromQL query
            duration_seconds: How far back to query
            step: Resolution step
        
        Returns:
            Query results
        """
        now = datetime.now(timezone.utc).timestamp()
        start = now - duration_seconds
        
        return self.query_range(query, start, now, step)
    
    def get_label_values(self, label_name: str) -> list[str]:
        """Get all values for a label."""
        response = self._request("GET", f"label/{label_name}/values")
        return response.get("data", [])
    
    def get_series(
        self,
        matchers: list[str],
        start: Optional[float] = None,
        end: Optional[float] = None,
    ) -> list[dict[str, str]]:
        """Get time series matching selectors.
        
        Args:
            matchers: List of metric selectors, e.g., ['up', 'http_requests_total']
            start: Optional start time
            end: Optional end time
        
        Returns:
            List of label sets
        """
        params: dict[str, Any] = {"match[]": matchers}
        if start is not None:
            params["start"] = start
        if end is not None:
            params["end"] = end
        
        response = self._request("GET", "series", params=params)
        return response.get("data", [])
    
    def validate(self) -> bool:
        """Validate connectivity by querying 'up' metric.
        
        Returns:
            True if connection works.
        """
        try:
            result = self.query_instant("up")
            logger.info("Prometheus connectivity validated")
            return True
        except Exception as e:
            logger.warning("Prometheus validation failed: %s", e)
            raise


class GrafanaClient:
    """Client for Grafana HTTP API.
    
    Implements:
    - Dashboard CRUD
    - Datasource management
    - Folder management
    - Alerting (if using new unified alerting)
    """
    
    def __init__(self, config: GrafanaConfig) -> None:
        self._config = config
        self._session = requests.Session()
    
    def _get_headers(self) -> dict[str, str]:
        """Get authentication headers."""
        headers = {"Content-Type": "application/json"}
        
        if self._config.auth_method == GrafanaConfig.AUTH_API_KEY:
            headers["Authorization"] = f"Bearer {self._config.api_key}"
        # Basic auth handled via auth tuple in request
        
        return headers
    
    def _get_auth(self) -> Optional[tuple[str, str]]:
        """Get basic auth tuple if using basic auth."""
        if self._config.auth_method == GrafanaConfig.AUTH_BASIC:
            return (self._config.username, self._config.password)
        return None
    
    def _request(
        self,
        method: str,
        endpoint: str,
        params: Optional[dict[str, Any]] = None,
        data: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        """Make HTTP request to Grafana API."""
        url = self._config.get_api_url(endpoint)
        headers = self._get_headers()
        auth = self._get_auth()
        
        try:
            response = self._session.request(
                method=method,
                url=url,
                params=params,
                json=data,
                headers=headers,
                auth=auth,
                timeout=self._config.timeout,
            )
            
            if response.status_code == 204:
                return {}
            
            try:
                result = response.json()
            except json.JSONDecodeError:
                raise GrafanaAPIError(
                    f"Grafana returned non-JSON: {response.status_code} {response.text[:200]}"
                )
            
            if response.status_code >= 400:
                raise GrafanaAPIError(
                    f"Grafana HTTP {response.status_code}: {result}"
                )
            
            return result
            
        except requests.RequestException as e:
            raise PrometheusGrafanaError(f"Network error: {e}") from e
    
    def health(self) -> dict[str, Any]:
        """Get Grafana health status."""
        return self._request("GET", "health")
    
    def get_dashboard(self, uid: str) -> dict[str, Any]:
        """Get dashboard by UID.
        
        Returns:
            Dict with 'dashboard' and 'meta' keys.
        """
        return self._request("GET", f"dashboards/uid/{uid}")
    
    def get_dashboard_by_slug(self, slug: str) -> dict[str, Any]:
        """Get dashboard by slug (older method)."""
        return self._request("GET", f"dashboards/db/{slug}")
    
    def create_or_update_dashboard(
        self,
        dashboard: dict[str, Any],
        folder_id: int = 0,
        folder_uid: Optional[str] = None,
        overwrite: bool = False,
        message: Optional[str] = None,
    ) -> dict[str, Any]:
        """Create or update a dashboard.
        
        Args:
            dashboard: Full dashboard JSON dict
            folder_id: Legacy folder ID (0 for General folder)
            folder_uid: Optional folder UID (preferred)
            overwrite: If True, overwrite existing with same title
            message: Optional commit message
        
        Returns:
            Response with id, uid, url, status
        """
        payload: dict[str, Any] = {
            "dashboard": dashboard,
            "overwrite": overwrite,
        }
        
        if folder_uid:
            payload["folderUid"] = folder_uid
        else:
            payload["folderId"] = folder_id
            
        if message:
            payload["message"] = message
        
        return self._request("POST", "dashboards/db", data=payload)
    
    def delete_dashboard(self, uid: str) -> dict[str, Any]:
        """Delete dashboard by UID."""
        return self._request("DELETE", f"dashboards/uid/{uid}")
    
    def search_dashboards(
        self,
        query: Optional[str] = None,
        tag: Optional[str] = None,
        type: str = "dash-db",
    ) -> list[dict[str, Any]]:
        """Search dashboards.
        
        Args:
            query: Search query
            tag: Filter by tag
            type: Type filter (dash-db, dash-folder)
        
        Returns:
            List of dashboard items
        """
        params: dict[str, Any] = {"type": type}
        if query:
            params["query"] = query
        if tag:
            params["tag"] = tag
        
        return self._request("GET", "search", params=params)
    
    def get_datasources(self) -> list[dict[str, Any]]:
        """List all datasources."""
        return self._request("GET", "datasources")
    
    def get_datasource_by_uid(self, uid: str) -> dict[str, Any]:
        """Get datasource by UID."""
        return self._request("GET", f"datasources/uid/{uid}")
    
    def create_datasource(
        self,
        name: str,
        type: str,
        url: str,
        access: str = "proxy",
        is_default: bool = False,
        json_data: Optional[dict[str, Any]] = None,
    ) -> dict[str, Any]:
        """Create a datasource.
        
        Args:
            name: Display name
            type: Datasource type (prometheus, graphite, etc.)
            url: Server URL
            access: Access mode (proxy or direct)
            is_default: Set as default datasource
            json_data: Additional type-specific config
        
        Returns:
            Created datasource
        """
        payload: dict[str, Any] = {
            "name": name,
            "type": type,
            "url": url,
            "access": access,
            "isDefault": is_default,
        }
        
        if json_data:
            payload["jsonData"] = json_data
        
        return self._request("POST", "datasources", data=payload)
    
    def validate(self) -> bool:
        """Validate connectivity via health endpoint.
        
        Returns:
            True if connection works.
        """
        try:
            health = self.health()
            logger.info("Grafana health: %s", health.get("version", "unknown"))
            return True
        except Exception as e:
            logger.warning("Grafana validation failed: %s", e)
            raise


# Global clients (lazy-loaded)
_global_prometheus: Optional[PrometheusClient] = None
_global_grafana: Optional[GrafanaClient] = None


def get_prometheus_client() -> PrometheusClient:
    """Get or create global PrometheusClient."""
    global _global_prometheus
    if _global_prometheus is None:
        config = PrometheusConfig.from_env()
        _global_prometheus = PrometheusClient(config)
    return _global_prometheus


def get_grafana_client() -> GrafanaClient:
    """Get or create global GrafanaClient."""
    global _global_grafana
    if _global_grafana is None:
        config = GrafanaConfig.from_env()
        _global_grafana = GrafanaClient(config)
    return _global_grafana

Pattern 2: PromQL Query Patterns

"""Common PromQL query patterns.

PromQL fundamentals:
- Instant vector: single sample per time series
- Range vector: sample range over time window
- rate(): counter growth rate per second (use for counters)
- irate(): instant rate from last two samples (use for spikes)
- sum by (): aggregate and keep specified labels
- sum without (): aggregate and remove specified labels
- histogram_quantile(): calculate percentile from histogram
- offset: compare with past data
- topk/bottomk: get k highest/lowest series
"""

from __future__ import annotations

import logging
from typing import Any, Optional
from datetime import datetime, timedelta, timezone

logger = logging.getLogger(__name__)


class PromQLQuery:
    """Builder for common PromQL query patterns.
    
    Provides type-safe construction of common query types:
    - Counter rates
    - Gauge aggregations
    - Histogram percentiles
    - Error rates
    - Uptime/SLO queries
    - Comparison queries (week-over-week)
    """
    
    def __init__(self, client: Any) -> None:
        self._client = client
    
    # ===================================================================
    # ❌ BAD — Common PromQL mistakes
    # ===================================================================
    
    @staticmethod
    def bad_counter_example() -> str:
        """❌ BAD: Don't do these things with counters."""
        # ❌ Using sum by (instance) http_requests_total
        #   - Counters only go up (except reset); you want rate()
        
        # ❌ rate(http_requests_total[1m]) > 100
        #   - Comparing raw rate without aggregation
        
        # ❌ sum(http_requests_total) - sum(http_requests_total offset 5m)
        #   - Don't calculate delta manually; use increase()
        
        return "rate(http_requests_total[1m])"  # Actually this part is OK
    
    # ===================================================================
    # ✅ GOOD — Proper PromQL patterns
    # ===================================================================
    
    @staticmethod
    def counter_rate(
        metric: str,
        window: str = "5m",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Rate of counter increase per second.
        
        Use this for: request counts, error counts, bytes transferred.
        
        Args:
            metric: Counter metric name
            window: Rate window (typically 1m-10m, at least 4x scrape interval)
            labels: Optional label filters
        
        Returns:
            PromQL: rate(metric{filters}[window])
        """
        selector = PromQLQuery._build_selector(metric, labels)
        return f"rate({selector}[{window}])"
    
    @staticmethod
    def counter_increase(
        metric: str,
        window: str = "5m",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Total increase of counter over window.
        
        Use this for: total requests in window.
        
        Args:
            metric: Counter metric name
            window: Time window
            labels: Optional filters
        
        Returns:
            PromQL: increase(metric{...}[window])
        """
        selector = PromQLQuery._build_selector(metric, labels)
        return f"increase({selector}[{window}])"
    
    @staticmethod
    def sum_by(
        query: str,
        by_labels: list[str],
    ) -> str:
        """Aggregate and keep specified labels.
        
        Args:
            query: Inner query (e.g., rate(http_requests_total[5m])
            by_labels: Labels to preserve in result
        
        Returns:
            PromQL: sum by (label1, label2) (query)
        """
        labels_str = ", ".join(by_labels)
        return f"sum by ({labels_str}) ({query})"
    
    @staticmethod
    def error_rate(
        total_metric: str,
        error_metric: str,
        window: str = "5m",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Calculate error rate percentage.
        
        Formula: rate(errors) / rate(total)
        
        Args:
            total_metric: Total request counter
            error_metric: Error counter (or same metric with status code label)
            window: Rate window
        
        Returns:
            PromQL for error rate 0-1 (or use < 1 for %)
        """
        # Simple version when using separate metrics:
        # sum by (instance) rate(errors[5m]) / sum by (instance) rate(total[5m])
        
        # Or when using status code label:
        # sum by (instance) rate(http_requests_total{status=~"5.."}[5m])
        # /
        # sum by (instance) rate(http_requests_total[5m])
        
        # This method assumes separate metrics for now, or caller can handle labels
        total_selector = PromQLQuery._build_selector(total_metric, labels)
        error_selector = PromQLQuery._build_selector(error_metric, labels)
        
        return (
            f"sum by (job, instance) rate({error_selector}[{window}])) "
            f"/ "
            f"sum by (job, instance) rate({total_selector}[{window}]))"
        )
    
    @staticmethod
    def error_rate_by_status(
        metric: str,
        window: str = "5m",
        status_label: str = "status",
        error_pattern: str = "5..",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Error rate using status code label.
        
        Args:
            metric: Request counter with status code label
            window: Rate window
            status_label: Label name for status code
            error_pattern: Regex for error codes (e.g., "5..", "4..|5..")
            labels: Additional filters
        
        Returns:
            PromQL: errors / total
        """
        # Build with label filters
        base_filters = dict(labels or {})
        error_filters = dict(base_filters)
        error_filters[status_label] = f"=~\"{error_pattern}\""  # Regex match
        
        # Use PromQL's label matching syntax
        # Actually we need to construct carefully
        # This returns a pattern the caller can use, but real implementation
        # would use separate queries or more complex label matching
        
        # Simplified for now - caller should construct based on their schema
        total = PromQLQuery.counter_rate(metric, window, base_filters)
        return (
            f"sum by (job, instance) ({PromQLQuery.counter_rate(metric, window, error_filters)}) "
            f"/ sum by (job, instance) ({total})"
        )
    
    @staticmethod
    def histogram_quantile(
        quantile: float,
        metric_bucket: str,
        window: str = "5m",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Calculate percentile from histogram.
        
        Use this for: latency percentiles (p50, p95, p99), duration distributions.
        
        Args:
            quantile: 0.0-1.0 (e.g., 0.95 for p95)
            metric_bucket: Histogram bucket metric (without _bucket suffix? or with?)
            window: Rate window
            labels: Optional filters
        
        Returns:
            PromQL: histogram_quantile(q, sum by (le) rate(metric_bucket[5m]))
        """
        # Histogram metrics typically have:
        # metric_bucket{le="0.1"}
        # metric_sum
        # metric_count
        
        selector = PromQLQuery._build_selector(metric_bucket, labels)
        
        return (
            f"histogram_quantile({quantile}, "
            f"sum by (le) (rate({selector}[{window}]))) )"
        )
    
    @staticmethod
    def uptime(
        metric: str = "up",
        window: str = "1h",
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Calculate uptime percentage over window.
        
        Args:
            metric: Typically 'up' (1=up, 0=down)
            window: Time window
            labels: Optional filters
        
        Returns:
            PromQL for uptime 0-1 (0-100% if multiplied)
        """
        selector = PromQLQuery._build_selector(metric, labels)
        return f"avg_over_time({selector}[{window}])"
    
    @staticmethod
    def compare_week_over_week(
        query: str,
    ) -> str:
        """Compare current value with 1 week ago.
        
        Returns:
            PromQL: query / query offset 1w
        """
        return f"({query}) / ({query} offset 1w)"
    
    @staticmethod
    def top_k(
        k: int,
        query: str,
    ) -> str:
        """Get top K time series by value.
        
        Args:
            k: Number to return
            query: Inner query to rank
        
        Returns:
            PromQL: topk(k, query)
        """
        return f"topk({k}, {query})"
    
    @staticmethod
    def _build_selector(
        metric: str,
        labels: Optional[dict[str, str]] = None,
    ) -> str:
        """Build metric selector with label filters.
        
        Handles special label values:
        - = exact match
        - != not equal
        - =~ regex match
        - !~ regex not match
        
        Args:
            metric: Metric name
            labels: Dict of label -> value (value can include operator prefix)
        
        Returns:
            metric{label="value", label2=~"regex"}
        """
        if not labels:
            return metric
        
        label_parts = []
        for key, value in labels.items():
            # Check if value already has operator
            if isinstance(value, str):
                if value.startswith("=~") or value.startswith("!~"):
                    # Already has operator and maybe quotes?
                    # Caller should handle properly
                    label_parts.append(f'{key}{value}')
                elif value.startswith("=") or value.startswith("!="):
                    label_parts.append(f'{key}{value}')
                else:
                    # Simple value - exact match
                    label_parts.append(f'{key}="{value}"')
            else:
                label_parts.append(f'{key}="{value}"')
        
        labels_str = ", ".join(label_parts)
        return f"{metric}{{{labels_str}}}"


# Example usage functions

def query_request_rate(
    client: Any,
    job: str,
    window_minutes: int = 5,
) -> dict[str, Any]:
    """Query request rate aggregated by instance.
    
    Args:
        client: PrometheusClient
        job: Job label value
        window_minutes: Rate window in minutes
    
    Returns:
        Dict with instances and their rates
    """
    window = f"{window_minutes}m"
    query = PromQLQuery.sum_by(
        PromQLQuery.counter_rate(
            "http_requests_total",
            window=window,
            labels={"job": job},
        ),
        by_labels=["instance", "job"],
    )
    
    result = client.query_instant(query)
    
    return {
        "result_type": result.get("resultType"),
        "results": [
            {
                "labels": r.get("metric", {}),
                "value": float(r["value"][1]) if "value" in r else None,
            }
            for r in result.get("result", [])
        ]
    }


def query_error_rate_percent(
    client: Any,
    job: str,
    window_minutes: int = 5,
) -> dict[str, Any]:
    """Query error rate as percentage 0-100.
    
    Assumes metrics:
    - http_requests_total (all requests)
    - http_requests_errors_total (error requests)
    
    Or uses status code label approach.
    """
    window = f"{window_minutes}m"
    
    # Using status code pattern (5xx errors)
    # This is a common pattern where status=~"5.." matches 500, 502, etc.
    
    # Actually simpler: sum errors / sum total
    # We'll construct two queries and divide
    
    # Query 1: errors (status 5xx)
    error_query = PromQLQuery.sum_by(
        PromQLQuery.counter_rate(
            "http_requests_total",
            window=window,
            labels={"job": job, "status": '=~"5.."'}
        ),
        by_labels=["job"],
    )
    
    # Query 2: total
    total_query = PromQLQuery.sum_by(
        PromQLQuery.counter_rate(
            "http_requests_total",
            window=window,
            labels={"job": job},
        ),
        by_labels=["job"],
    )
    
    # Combined in one PromQL: errors / total
    combined = f"({error_query}) / ({total_query}) * 100"
    
    result = client.query_instant(combined)
    
    results = result.get("result", [])
    if results:
        value = float(results[0]["value"][1])
        return {
            "error_rate_pct": value,
            "window_minutes": window_minutes,
        }
    
    return {"error_rate_pct": 0.0, "window_minutes": window_minutes}


def query_latency_percentiles(
    client: Any,
    job: str,
    metric_bucket: str = "http_request_duration_seconds_bucket",
    window_minutes: int = 5,
) -> dict[str, float]:
    """Query p50, p95, p99 latency percentiles from histogram.
    
    Args:
        client: PrometheusClient
        job: Job label
        metric_bucket: Histogram bucket metric name
        window_minutes: Rate window
    
    Returns:
        Dict with p50, p95, p99 in seconds (or whatever unit the histogram uses)
    """
    window = f"{window_minutes}m"
    labels = {"job": job}
    
    result = {}
    
    for quantile, name in [(0.5, "p50"), (0.95, "p95"), (0.99, "p99")]:
        query = PromQLQuery.histogram_quantile(
            quantile, metric_bucket, window, labels)
        query_result = client.query_instant(query)
        values = query_result.get("result", [])
        if values:
            result[name] = float(values[0]["value"][1])
        else:
            result[name] = 0.0
    
    return result

Pattern 3: Grafana Dashboard as Code

"""Grafana dashboard as code patterns.

Grafana dashboards are JSON objects. Key concepts:

- Dashboard JSON structure:
  - id: Numeric ID (assigned by Grafana)
  - uid: Unique ID (string, user-provided or auto-generated)
  - title: Display name
  - tags: List of tags
  - timezone: Browser or UTC
  - panels: Array of panel objects
  - templating: Template variables
  - time: Default time range
  - refresh: Auto-refresh interval

- Panel types:
  - graph: Time series line chart
  - stat: Single stat display
  - table: Table view
  - gauge: Gauge display
  - heatmap: Heatmap for histograms
  - text: Text/markdown panel
"""

from __future__ import annotations

import json
import logging
from typing import Any, Optional
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


@dataclass
class GrafanaDashboardBuilder:
    """Builder for Grafana dashboard JSON.
    
    Provides fluent API for constructing dashboard creation.
    """
    
    title: str
    uid: Optional[str] = None
    tags: list[str] = field(default_factory=list)
    timezone: str = "browser"
    refresh: str = "1m"
    
    _panels: list[dict[str, Any]] = field(default_factory=list)
    _variables: list[dict[str, Any]] = field(default_factory=list)
    _next_panel_id: int = 1
    
    def add_tag(self, tag: str) -> "GrafanaDashboardBuilder":
        """Add a tag."""
        self.tags.append(tag)
        return self
    
    def add_variable(
        self,
        name: str,
        label: str,
        datasource: str,
        query: str,
        include_all: bool = True,
        multi: bool = False,
        default_value: Optional[str] = None,
    ) -> "GrafanaDashboardBuilder":
        """Add a template variable.
        
        Args:
            name: Variable name (used in queries as $name)
            label: Display label
            datasource: Datasource name or uid
            query: Query for values (e.g., label_values(metric, label))
            include_all: Add "All" option
            multi: Allow multiple selection
            default_value: Optional default value
        """
        variable: dict[str, Any] = {
            "name": name,
            "type": "query",
            "label": label,
            "datasource": datasource,
            "query": query,
            "refresh": 2,  # Refresh on time range change
            "includeAll": include_all,
            "multi": multi,
            "options": [],
        }
        
        if include_all:
            variable["allValue"] = ".*"  # Regex for All
        
        if default_value:
            variable["current"] = {
                "text": default_value,
                "value": default_value,
            }
        
        self._variables.append(variable)
        return self
    
    def add_panel_graph(
        self,
        title: str,
        targets: list[dict[str, Any]],
        yaxis_label: Optional[str] = None,
        span: int = 12,  # 6 = half width, 12 = full width
        height: int = 8,
        legend_show: bool = True,
    ) -> "GrafanaDashboardBuilder":
        """Add a graph (time series) panel.
        
        Args:
            title: Panel title
            targets: List of target dicts with expr, refId, legendFormat
            yaxis_label: Optional Y axis label
            span: Width (6=half, 12=full)
            height: Row height
            legend_show: Show legend
        """
        panel_id = self._next_panel_id
        self._next_panel_id += 1
        
        panel: dict[str, Any] = {
            "id": panel_id,
            "type": "graph",
            "title": title,
            "span": span,
            "height": f"{height}h",
            "targets": targets,
            "legend": {
                "show": legend_show,
                "values": False,
                "min": False,
                "max": False,
                "current": True,
                "avg": False,
            },
            "yaxes": [
                {
                    "format": "short",

…(truncated)