Querying Prometheus API

import httpx
from datetime import datetime, timedelta

class PrometheusClient:
    def __init__(self, base_url: str = "http://localhost:9090"):
        self.base_url = base_url
        self.client = httpx.Client(timeout=30)

    def query(self, promql: str, time: str | None = None) -> dict:
        """Execute an instant query."""
        params = {"query": promql}
        if time:
            params["time"] = time
        resp = self.client.get(f"{self.base_url}/api/v1/query", params=params)
        resp.raise_for_status()
        return resp.json()["data"]

    def query_range(self, promql: str, start: str, end: str, step: str = "60s") -> dict:
        """Execute a range query."""
        params = {"query": promql, "start": start, "end": end, "step": step}
        resp = self.client.get(f"{self.base_url}/api/v1/query_range", params=params)
        resp.raise_for_status()
        return resp.json()["data"]

    def targets(self) -> list[dict]:
        """List all scrape targets."""
        resp = self.client.get(f"{self.base_url}/api/v1/targets")
        resp.raise_for_status()
        return resp.json()["data"]["activeTargets"]

# Usage
prom = PrometheusClient("http://prometheus:9090")
result = prom.query('up{job="api-server"}')
print(result)

prom = PrometheusClient()

# Current CPU usage per instance
result = prom.query('100 - (avg by(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)')
for metric in result["result"]:
    instance = metric["metric"]["instance"]
    cpu = float(metric["value"][1])
    print(f"{instance}: {cpu:.1f}% CPU")

# Current memory usage percentage
result = prom.query(
    '(1 - node_memory_MemAvailable_bytes / node_memory_MemTotal_bytes) * 100'
)
for metric in result["result"]:
    instance = metric["metric"]["instance"]
    mem = float(metric["value"][1])
    print(f"{instance}: {mem:.1f}% memory")

# Current request rate (requests per second)
result = prom.query('rate(http_requests_total[5m])')
for metric in result["result"]:
    endpoint = metric["metric"]["endpoint"]
    rate = float(metric["value"][1])
    print(f"{endpoint}: {rate:.1f} req/s")

from datetime import datetime, timedelta

prom = PrometheusClient()

# Get CPU usage over the last hour
now = datetime.utcnow()
start = (now - timedelta(hours=1)).isoformat() + "Z"
end = now.isoformat() + "Z"

result = prom.query_range(
    promql='100 - (avg by(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)',
    start=start,
    end=end,
    step="60s",
)

for series in result["result"]:
    instance = series["metric"]["instance"]
    values = [(ts, float(val)) for ts, val in series["values"]]
    print(f"Instance: {instance}")
    print(f"  Min CPU: {min(v for _, v in values):.1f}%")
    print(f"  Max CPU: {max(v for _, v in values):.1f}%")
    print(f"  Avg CPU: {sum(v for _, v in values) / len(values):.1f}%")

prom = PrometheusClient()

# List all active alerts
result = prom.query('ALERTS{alertstate="firing"}')
for alert in result["result"]:
    name = alert["metric"]["alertname"]
    severity = alert["metric"].get("severity", "unknown")
    print(f"FIRING: {name} (severity: {severity})")

# Get alert history
result = prom.query_range(
    promql='ALERTS{alertname="HighCPU"}',
    start="2024-01-15T00:00:00Z",
    end="2024-01-15T23:59:59Z",
    step="300s",
)

for series in result["result"]:
    for ts, val in series["values"]:
        if val == "1":
            print(f"Alert fired at {ts}")
        else:
            print(f"Alert resolved at {ts}")

from datetime import datetime, timedelta

def get_dashboard_data(prom: PrometheusClient) -> dict:
    """Collect data for a service dashboard."""
    now = datetime.utcnow()
    hour_ago = (now - timedelta(hours=1)).isoformat() + "Z"

    # Current request rate
    req_rate = prom.query('sum(rate(http_requests_total[5m]))')

    # Error rate
    error_rate = prom.query('sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m])) * 100')

    # P99 latency
    p99_latency = prom.query('histogram_quantile(0.99, sum(rate(http_request_duration_seconds_bucket[5m])) by (le))')

    # Request rate over time
    req_history = prom.query_range(
        promql='sum(rate(http_requests_total[5m]))',
        start=hour_ago,
        end=now.isoformat() + "Z",
        step="60s",
    )

    return {
        "current_rps": float(req_rate["result"][0]["value"][1]) if req_rate["result"] else 0,
        "error_rate": float(error_rate["result"][0]["value"][1]) if error_rate["result"] else 0,
        "p99_latency": float(p99_latency["result"][0]["value"][1]) if p99_latency["result"] else 0,
        "request_history": req_history["result"],
    }

groups:
  - name: api_server_rules
    interval: 30s
    rules:
      - record: job:http_requests:rate5m
        expr: sum(rate(http_requests_total[5m])) by (job)

      - record: job:http_errors:rate5m
        expr: sum(rate(http_requests_total{status=~"5.."}[5m])) by (job)

      - record: job:http_error_ratio:rate5m
        expr: job:http_errors:rate5m / job:http_requests:rate5m * 100

      - record: instance:node_cpu:usage
        expr: 100 - (avg by(instance)(irate(node_cpu_seconds_total{mode="idle"}[5m])) * 100)