bytegeist-portfolio/status-api/status_api.py

import json
import os
import time
import urllib.parse
import urllib.request
from pathlib import Path
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"))
SITE_GIT_DIR = Path(os.getenv("SITE_GIT_DIR", "/site/.git"))

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!=""}))',
}

HTTP_CHECKS = {
    "gitea": "http://host.docker.internal:3000/api/v1/version",
    "woodpecker": "http://host.docker.internal:8010/",
    "uptime_kuma": "http://host.docker.internal:3001/",
    "prometheus": "http://host.docker.internal:9090/-/healthy",
}

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 get_text(url):
    request = urllib.request.Request(url, headers={"Accept": "*/*"})
    with urllib.request.urlopen(request, timeout=TIMEOUT) as response:
        return response.status, response.read(1024).decode("utf-8", "replace")

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 check_services():
    services = {}
    for name, url in HTTP_CHECKS.items():
        try:
            status, body = get_text(url)
            services[name] = {"up": 200 <= status < 400, "status": status}
            if name == "gitea":
                try:
                    services[name]["version"] = json.loads(body).get("version")
                except Exception:
                    pass
        except Exception:
            services[name] = {"up": False, "status": None}
    return services

def deployed_commit():
    try:
        head = (SITE_GIT_DIR / "HEAD").read_text(encoding="utf-8").strip()
        if head.startswith("ref: "):
            ref = head[5:]
            ref_path = SITE_GIT_DIR / ref
            if ref_path.exists():
                value = ref_path.read_text(encoding="utf-8").strip()
            else:
                value = None
                packed = SITE_GIT_DIR / "packed-refs"
                if packed.exists():
                    for line in packed.read_text(encoding="utf-8").splitlines():
                        if line and not line.startswith(("#", "^")):
                            sha, name = line.split(" ", 1)
                            if name == ref:
                                value = sha
                                break
                if not value:
                    return None
        else:
            value = head
        return value[:7]
    except Exception:
        return None

def build_status():
    values = {name: prometheus_query(query) for name, query in QUERIES.items()}
    targets = target_health()
    services = check_services()
    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,
        "services": services,
        "deployment": {
            "commit": deployed_commit(),
            "site": "bytegeist.casko.dev",
        },
    }

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()