Add live Prometheus telemetry endpoint

status-api/status_api.py

@@ -0,0 +1,91 @@
+import json
+import os
+import time
+import urllib.parse
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+PROMETHEUS_URL = os.getenv("PROMETHEUS_URL", "http://host.docker.internal:9090").rstrip("/")
+TIMEOUT = float(os.getenv("PROMETHEUS_TIMEOUT", "2.5"))
+
+QUERIES = {
+    "cpu_percent": '100-(avg(rate(node_cpu_seconds_total{mode="idle"}[2m]))*100)',
+    "memory_percent": '(1-(node_memory_MemAvailable_bytes/node_memory_MemTotal_bytes))*100',
+    "host_uptime_seconds": 'time()-node_boot_time_seconds',
+    "running_containers": 'count(count by (name) (container_last_seen{name!=""}))',
+}
+
+def get_json(url):
+    request = urllib.request.Request(url, headers={"Accept": "application/json"})
+    with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
+        return json.load(response)
+
+def prometheus_query(expression):
+    url = f"{PROMETHEUS_URL}/api/v1/query?{urllib.parse.urlencode({'query': expression})}"
+    payload = get_json(url)
+    if payload.get("status") != "success":
+        raise RuntimeError("Prometheus query failed")
+    results = payload.get("data", {}).get("result", [])
+    if not results:
+        raise RuntimeError("Prometheus query returned no data")
+    return float(results[0]["value"][1])
+
+def target_health():
+    payload = get_json(f"{PROMETHEUS_URL}/api/v1/targets")
+    targets = payload.get("data", {}).get("activeTargets", [])
+    jobs = {}
+    for target in targets:
+        job = target.get("labels", {}).get("job", "unknown")
+        jobs[job] = target.get("health") == "up"
+    return {
+        "up": sum(1 for target in targets if target.get("health") == "up"),
+        "total": len(targets),
+        "jobs": jobs,
+    }
+
+def build_status():
+    values = {name: prometheus_query(query) for name, query in QUERIES.items()}
+    targets = target_health()
+    return {
+        "ok": True,
+        "source": "Prometheus / node_exporter / cAdvisor",
+        "sampled_at": int(time.time()),
+        "host": {
+            "cpu_percent": round(values["cpu_percent"], 1),
+            "memory_percent": round(values["memory_percent"], 1),
+            "uptime_seconds": int(values["host_uptime_seconds"]),
+        },
+        "containers": {"running": int(values["running_containers"])},
+        "prometheus": targets,
+    }
+
+class Handler(BaseHTTPRequestHandler):
+    def send_json(self, status_code, payload):
+        body = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        self.send_response(status_code)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Cache-Control", "no-store, max-age=0")
+        self.send_header("Content-Length", str(len(body)))
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_GET(self):
+        if self.path == "/healthz":
+            self.send_json(200, {"ok": True})
+            return
+        if self.path != "/status":
+            self.send_json(404, {"ok": False, "error": "not_found"})
+            return
+        try:
+            self.send_json(200, build_status())
+        except Exception as exc:
+            self.send_json(503, {"ok": False, "sampled_at": int(time.time()), "error": type(exc).__name__})
+
+    def log_message(self, fmt, *args):
+        print(f"{self.address_string()} - {fmt % args}", flush=True)
+
+if __name__ == "__main__":
+    server = ThreadingHTTPServer(("0.0.0.0", 8080), Handler)
+    print("ByteGeist status API listening on :8080", flush=True)
+    server.serve_forever()