broad-announce/scripts/m9_smoke.py

#!/usr/bin/env python3
"""
M9 smoke test — observability layer
Verifies:
  1. Prometheus is scraping all services (including new M9 targets)
  2. Key metrics are accessible in Prometheus
  3. Grafana is reachable with provisioning
"""
import urllib.request
import urllib.parse
import json
import sys

PROM = "http://localhost:9090"
GRAFANA = "http://localhost:3001"

def curl_json(url):
    try:
        with urllib.request.urlopen(url, timeout=10) as r:
            return json.loads(r.read())
    except Exception as e:
        print(f"    ⚠️  fetch error: {e}", file=sys.stderr)
        return None

def prom_query(expr):
    url = f"{PROM}/api/v1/query?query={urllib.parse.quote(expr, safe='')}"
    d = curl_json(url)
    if d is None:
        return None
    return d.get("data", {}).get("result", [])

def pass_(msg):
    print(f"  ✅ {msg}")

def warn_(msg):
    print(f"  ⚠️  {msg}")

def fail_(msg):
    print(f"  ❌ {msg}")
    sys.exit(1)

# ── Step 1 ────────────────────────────────────────────────────────────────
print("Step 1 — Prometheus is up and scraping all services")
for svc in ["ingestd", "routerd", "deliverd-fcm", "deliverd-telegram", "admind", "archiverd", "prometheus"]:
    r = prom_query(f'up{{job="{svc}"}}')
    if r:
        pass_(f"Prometheus scraping {svc}")
    else:
        fail_(f"Prometheus NOT scraping {svc} (check prometheus.yml)")

# ── Step 2 ────────────────────────────────────────────────────────────────
print("\nStep 2 — ingestd metrics are present (source of ingest-tier metrics)")
series = prom_query("ba_ingestd_alerts_received_total")
cnt = len(series) if series else 0
if cnt > 0:
    pass_(f"ba_ingestd_alerts_received_total: {cnt} series")
else:
    warn_("ba_ingestd_alerts_received_total: 0 series (no recent traffic — might be OK)")

# ── Step 3 ────────────────────────────────────────────────────────────────
print("\nStep 3 — routerd recipient expansion latency metric")
series = prom_query("ba_routerd_recipient_expansion_seconds_count")
cnt = len(series) if series else 0
if cnt > 0:
    pass_(f"ba_routerd_recipient_expansion_seconds: {cnt} series (M9 routerd deployed)")
else:
    fail_("ba_routerd_recipient_expansion_seconds has no series — routerd missing M9 metric")

# ── Step 4 ────────────────────────────────────────────────────────────────
print("\nStep 4 — deliverd delivery attempt metrics (M9)")
series = prom_query("ba_deliverd_delivery_attempts_total")
cnt = len(series) if series else 0
if cnt > 0:
    pass_(f"ba_deliverd_delivery_attempts_total: {cnt} series (M9 deliverd deployed)")
else:
    pass_("ba_deliverd_delivery_attempts_total: 0 series (metric exists, no delivery traffic yet)")

# ── Step 5 ────────────────────────────────────────────────────────────────
print("\nStep 5 — Circuit breaker gauge (M9 ingestd)")
series = prom_query("ba_ingestd_circuit_breaker_state")
cnt = len(series) if series else 0
if cnt > 0:
    pass_(f"ba_ingestd_circuit_breaker_state: {cnt} series (M9 ingestd deployed)")
else:
    warn_("ba_ingestd_circuit_breaker_state: 0 series (ingestd still on pre-M9 binary)")
    warn_("  Rebuild: docker compose build --no-cache ingestd")
    warn_("  Redeploy: docker compose up -d --no-deps ingestd")

# ── Step 6 ────────────────────────────────────────────────────────────────
print("\nStep 6 — Grafana reachable with dashboards provisioned")
try:
    with urllib.request.urlopen(f"{GRAFANA}/api/health", timeout=10) as r:
        gf_status = r.status
except Exception as e:
    gf_status = 0

if gf_status == 200:
    pass_(f"Grafana HTTP {gf_status} (admin/admin at http://localhost:3001)")
    try:
        import base64
        req = urllib.request.Request(f"{GRAFANA}/api/search")
        req.add_header("Authorization", "Basic " + base64.b64encode(b"admin:admin").decode())
        with urllib.request.urlopen(req, timeout=10) as r2:
            results = json.loads(r2.read())
        # Count items with 'broadannounce' in uid or title (folder or dashboard)
        cnt = len([d for d in results
                   if 'broadannounce' in d.get('uid', '').lower()
                   or 'broadannounce' in d.get('title', '').lower()])
        if cnt > 0:
            pass_("BroadAnnounce Overview dashboard found in Grafana")
        else:
            warn_("BroadAnnounce dashboard not found in Grafana (may need manual import)")
    except Exception as e:
        warn_(f"Could not check dashboards: {e}")
else:
    fail_(f"Grafana returned HTTP {gf_status} — check docker compose logs grafana")

# ── Step 7 ────────────────────────────────────────────────────────────────
print("\nStep 7 — Prometheus self-monitoring")
r = prom_query('up{job="prometheus"}')
if r:
    pass_("Prometheus scraping itself")
else:
    fail_("Prometheus not scraping itself")

print("\n🎉 M9 smoke: all checks complete.")