Grafana Dashboards for DR Drill Observability

#!/usr/bin/env python3
"""Provision a DR-drill observability dashboard into Grafana as code.

Usage:
    GRAFANA_TOKEN=... python3 provision_dashboard.py <grafana_url> <ds_uid>

Exit codes:
    0  dashboard (and alert rule) provisioned successfully
    1  Grafana rejected the dashboard or alert payload
    2  usage / configuration error (missing token or arguments)
"""
import json
import os
import sys
from typing import Dict, List

import requests

RTO_P95_SECONDS = 600      # restore-duration objective for this workload
RPO_LAG_SECONDS = 30       # replication-lag ceiling for this workload


def _target(expr: str, legend: str, ds_uid: str) -> Dict:
    """One Prometheus query target inside a panel."""
    return {
        "expr": expr,
        "legendFormat": legend,
        "refId": "A",
        "datasource": {"type": "prometheus", "uid": ds_uid},
    }


def build_panels(ds_uid: str) -> List[Dict]:
    """Four PromQL panels covering the DR-drill validation SLOs."""
    return [
        {
            "id": 1, "title": "Restore duration p95 (s)", "type": "timeseries",
            "gridPos": {"h": 8, "w": 12, "x": 0, "y": 0},
            "fieldConfig": {"defaults": {"unit": "s", "thresholds": {
                "mode": "absolute",
                "steps": [{"color": "green", "value": None},
                          {"color": "red", "value": RTO_P95_SECONDS}]}}},
            "targets": [_target(
                "histogram_quantile(0.95, sum by (le, engine) "
                "(rate(restore_duration_seconds_bucket[30m])))",
                " p95", ds_uid)],
        },
        {
            "id": 2, "title": "Corrupt page rate (per hour)", "type": "timeseries",
            "gridPos": {"h": 8, "w": 12, "x": 12, "y": 0},
            "targets": [_target(
                "sum by (engine) (increase(corrupt_pages_found_total[1h]))",
                "", ds_uid)],
        },
        {
            "id": 3, "title": "Replication lag vs RPO (s)", "type": "timeseries",
            "gridPos": {"h": 8, "w": 12, "x": 0, "y": 8},
            "fieldConfig": {"defaults": {"unit": "s", "thresholds": {
                "mode": "absolute",
                "steps": [{"color": "green", "value": None},
                          {"color": "red", "value": RPO_LAG_SECONDS}]}}},
            "targets": [_target(
                "max by (engine) (replication_lag_seconds)",
                " lag", ds_uid)],
        },
        {
            "id": 4, "title": "Drill success ratio", "type": "stat",
            "gridPos": {"h": 8, "w": 12, "x": 12, "y": 8},
            "fieldConfig": {"defaults": {"unit": "percentunit"}},
            "targets": [_target(
                "sum(rate(artifacts_validated_total{verdict=\"valid\"}[6h])) "
                "/ sum(rate(artifacts_validated_total[6h]))",
                "success ratio", ds_uid)],
        },
    ]


def build_dashboard(ds_uid: str) -> Dict:
    """Assemble the full dashboard model Grafana's API expects."""
    return {
        "dashboard": {
            "uid": "dr-drill-observability",
            "title": "DR Drill Observability",
            "tags": ["dr", "backup-validation"],
            "timezone": "utc",
            "schemaVersion": 39,
            "version": 0,
            "refresh": "1m",
            "panels": build_panels(ds_uid),
        },
        "overwrite": True,
        "message": "provisioned from provision_dashboard.py",
    }

def _headers() -> Dict[str, str]:
    token = os.environ.get("GRAFANA_TOKEN")
    if not token:
        raise KeyError("GRAFANA_TOKEN not set")
    return {"Authorization": f"Bearer {token}",
            "Content-Type": "application/json"}


def post_dashboard(base_url: str, model: Dict) -> int:
    """POST the dashboard model to Grafana; return its numeric id."""
    resp = requests.post(f"{base_url}/api/dashboards/db",
                         headers=_headers(), data=json.dumps(model),
                         timeout=30)
    resp.raise_for_status()
    return resp.json()["id"]


def provision_alert(base_url: str, ds_uid: str) -> None:
    """Create a burn-style alert rule on the restore-duration SLO."""
    rule = {
        "title": "Restore p95 over RTO",
        "condition": "C",
        "data": [{
            "refId": "A",
            "datasourceUid": ds_uid,
            "model": {"expr": "histogram_quantile(0.95, sum by (le) "
                              "(rate(restore_duration_seconds_bucket[30m])))",
                      "instant": True, "refId": "A"},
        }],
        "noDataState": "NoData",
        "execErrState": "Alerting",
        "for": "15m",
    }
    resp = requests.post(f"{base_url}/api/v1/provisioning/alert-rules",
                         headers=_headers(), data=json.dumps(rule),
                         timeout=30)
    resp.raise_for_status()


def main(argv: List[str]) -> int:
    if len(argv) != 3:
        print("usage: provision_dashboard.py <grafana_url> <ds_uid>",
              file=sys.stderr)
        return 2
    base_url, ds_uid = argv[1].rstrip("/"), argv[2]
    model = build_dashboard(ds_uid)

    # Version the exact JSON that will be sent, before sending it.
    with open("dr_drill_dashboard.json", "w", encoding="utf-8") as fh:
        json.dump(model, fh, indent=2, sort_keys=True)

    try:
        dashboard_id = post_dashboard(base_url, model)
        provision_alert(base_url, ds_uid)
    except KeyError as exc:
        print(f"config error: {exc}", file=sys.stderr)
        return 2
    except requests.RequestException as exc:
        print(f"grafana rejected provisioning: {exc}", file=sys.stderr)
        return 1

    print(f"provisioned dashboard id={dashboard_id} uid=dr-drill-observability")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))

$ GRAFANA_TOKEN=*** python3 provision_dashboard.py https://grafana.dr.internal ceabc123
provisioned dashboard id=42 uid=dr-drill-observability
$ echo $?
0