# Grafana Prometheus > Implements Grafana and Prometheus integration (metrics collection, querying, alerting rules, Grafana dashboards as code, PromQL patterns, and Grafana HTTP API for dashboard management, using prometheus-api-client and grafana-api Python SDKs. - Skill: `paulpas/grafana-prometheus` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/grafana-prometheus` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/grafana-prometheus/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Integrations & APIs - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/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) ```python """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 ```python """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 ```python """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)