hermes-skills/devops/proxmox-ve-administration/scripts/monitoring-validation.py

#!/usr/bin/env python3
"""
Monitoring Stack Validation Script

Validates that all Prometheus targets are UP, key metrics exist for
each dashboard, and Grafana dashboard variables are properly configured.

Transfer to CT 141 via base64 (pitfall 40):
  SCRIPT=$(base64 -w0 monitoring_validation.py)
  ssh -i ~/.ssh/key root@10.0.20.70 "echo '$SCRIPT' | base64 -d | pct exec 141 -- tee /tmp/validate.py >/dev/null && pct exec 141 -- python3 /tmp/validate.py"

Usage: python3 monitoring_validation.py
Exit: 0 if all checks pass, 1 if any target down or metric missing
"""

import json
import sys
import urllib.request
import urllib.parse
import base64

PROM = "http://127.0.0.1:9090"
GRAFANA = "http://127.0.0.1:3000"
GRAFANA_USER = "admin"
GRAFANA_PASS = "Grafana2026!"
AUTH = base64.b64encode(f"{GRAFANA_USER}:{GRAFANA_PASS}".encode()).decode()


def prom_query(expr):
    url = f"{PROM}/api/v1/query?query={urllib.parse.quote(expr)}"
    with urllib.request.urlopen(url) as r:
        d = json.loads(r.read())
    return d.get("data", {}).get("result", [])


def grafana_get(path):
    req = urllib.request.Request(f"{GRAFANA}{path}")
    req.add_header("Authorization", f"Basic {AUTH}")
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())


def grafana_post(path, data):
    body = json.dumps(data).encode()
    req = urllib.request.Request(f"{GRAFANA}{path}", data=body, method="POST")
    req.add_header("Authorization", f"Basic {AUTH}")
    req.add_header("Content-Type", "application/json")
    with urllib.request.urlopen(req) as resp:
        return json.loads(resp.read())


def fix_datasource_var(dash, var_name, prom_uid):
    """Fix a datasource template variable with empty current value."""
    templating = dash.get("templating", {}).get("list", [])
    changed = False
    for v in templating:
        if v.get("name") == var_name and not v.get("current"):
            v["current"] = {"text": "Prometheus", "value": prom_uid}
            changed = True
    return changed


def add_datasource_var(dash, var_name, prom_uid):
    """Add a missing datasource template variable."""
    templating = dash.get("templating", {}).get("list", [])
    existing = [v.get("name") for v in templating]
    if var_name not in existing:
        templating.insert(0, {
            "name": var_name,
            "label": "Prometheus",
            "type": "datasource",
            "query": "prometheus",
            "current": {"text": "Prometheus", "value": prom_uid},
            "includeAll": False,
            "hide": 0,
        })
        return True
    return False


def main():
    errors = []

    print("=" * 60)
    print("MONITORING STACK VALIDATION")
    print("=" * 60)

    # 1. Check all Prometheus targets
    with urllib.request.urlopen(f"{PROM}/api/v1/targets") as r:
        d = json.loads(r.read())
    targets = d.get("data", {}).get("activeTargets", [])
    up = [t for t in targets if t["health"] == "up"]
    down = [t for t in targets if t["health"] != "up"]

    print(f"\n1. Prometheus Targets: {len(up)} UP, {len(down)} DOWN")
    for t in down:
        job = t.get("labels", {}).get("job", "?")
        inst = t.get("scrapeUrl", "?")
        err = t.get("lastError", "")[:60]
        print(f"   DOWN: {job} {inst} — {err}")
        errors.append(f"Target down: {job}/{inst}")

    # 2. Check key metrics per dashboard
    checks = [
        ("Node Exporter", "node_uname_info", 1),
        ("PVE", "pve_cpu_usage_ratio", 1),
        ("Ceph", "ceph_health_status", 1),
        ("MySQL/Galera", "mysql_up", 1),
    ]
    print("\n2. Metrics per dashboard:")
    for name, metric, min_series in checks:
        r = prom_query(metric)
        count = len(r)
        status = "OK" if count >= min_series else "FAIL"
        print(f"   [{status}] {name:20s} {metric:30s} {count:4d} series")
        if count < min_series:
            errors.append(f"No data for {metric}")

    # 3. Get Prometheus datasource UID
    ds_list = grafana_get("/api/datasources")
    prom_uid = None
    for ds in ds_list:
        if ds.get("type") == "prometheus":
            prom_uid = ds.get("uid")
            break
    if not prom_uid:
        print("\n3. ERROR: No Prometheus datasource found in Grafana!")
        errors.append("No Prometheus datasource")
        prom_uid = "PBFA97CFB590B2093"  # fallback from memory

    # 4. Check and fix dashboard variables
    dashboards = {
        "Node Exporter Full": {"uid": "rYdddlPWk", "ds_var": "ds_prometheus"},
        "PVE via Prometheus": {"uid": "Dp7Cd57Zza", "ds_var": "DS_PROMETHEUS"},
        "Ceph Cluster": {"uid": "tbO9LAiZK", "ds_var": "DS_PROMETHEUS"},
        "MySQL Overview": {"uid": "MQWgroiiz", "ds_var": None},
    }

    print("\n3. Dashboard variables:")
    for name, info in dashboards.items():
        uid = info["uid"]
        ds_var = info["ds_var"]
        d = grafana_get(f"/api/dashboards/uid/{uid}")
        dash = d.get("dashboard", {})
        panels = dash.get("panels", [])
        templating = dash.get("templating", {}).get("list", [])

        vars_str = ", ".join(
            f"${v['name']}={v.get('current', {}).get('value', '?')}"
            for v in templating
        )
        print(f"   {name:25s} {len(panels):3d} panels | {vars_str}")

        # Fix missing/empty datasource variables
        if ds_var:
            needs_fix = add_datasource_var(dash, ds_var, prom_uid)
            needs_fix = fix_datasource_var(dash, ds_var, prom_uid) or needs_fix
            if needs_fix:
                result = grafana_post("/api/dashboards/db", {
                    "dashboard": dash,
                    "message": f"Fix {ds_var} datasource variable",
                    "overwrite": True,
                })
                print(f"   → Fixed: {result.get('status')}")

    # 5. Alertmanager
    print("\n4. Alertmanager:")
    try:
        with urllib.request.urlopen("http://127.0.0.1:9093/api/v2/alerts") as r:
            alerts = json.loads(r.read())
        print(f"   Active alerts: {len(alerts)}")
        for a in alerts[:5]:
            lbl = a.get("labels", {}).get("alertname", "?")
            summary = a.get("annotations", {}).get("summary", "")[:60]
            print(f"   - {lbl}: {summary}")
    except Exception as e:
        print(f"   Error: {e}")
        errors.append(f"Alertmanager: {e}")

    # 6. Summary
    print("\n" + "=" * 60)
    if errors:
        print(f"RESULT: {len(errors)} ISSUES FOUND")
        for e in errors:
            print(f"  ! {e}")
        sys.exit(1)
    else:
        print("RESULT: ALL CHECKS PASSED")
        sys.exit(0)


if __name__ == "__main__":
    main()