homelab/site/metrics.py

#!/usr/bin/env python3
"""Expose a small, sanitized snapshot of host CPU, memory, disk, and NVIDIA GPU usage."""

import json
import os
import subprocess
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path


PORT = int(os.environ.get("PORT", "8092"))
HOST_PROC = Path(os.environ.get("HOST_PROC", "/host/proc"))
DISK_PATH = Path(os.environ.get("DISK_PATH", "/host/disk"))
CACHE_TTL_SECONDS = float(os.environ.get("CACHE_TTL_SECONDS", "2"))

cache_lock = threading.Lock()
cached_snapshot = None
cached_at = 0.0
previous_cpu = None


def read_text(path):
    return path.read_text(encoding="utf-8", errors="replace")


def read_cpu_times():
    first_line = read_text(HOST_PROC / "stat").splitlines()[0]
    values = [int(value) for value in first_line.split()[1:]]
    idle = values[3] + (values[4] if len(values) > 4 else 0)
    return sum(values), idle


def cpu_usage_percent():
    global previous_cpu

    current = read_cpu_times()
    if previous_cpu is None:
        previous_cpu = current
        time.sleep(0.12)
        current = read_cpu_times()

    total_delta = current[0] - previous_cpu[0]
    idle_delta = current[1] - previous_cpu[1]
    previous_cpu = current
    if total_delta <= 0:
        return 0.0
    return round(max(0.0, min(100.0, (1 - idle_delta / total_delta) * 100)), 1)


def cpu_info():
    cpuinfo = read_text(HOST_PROC / "cpuinfo")
    model = "Unknown CPU"
    logical_cores = 0
    for line in cpuinfo.splitlines():
        if line.startswith("processor"):
            logical_cores += 1
        elif line.startswith("model name") and model == "Unknown CPU":
            model = line.split(":", 1)[1].strip()

    load_values = read_text(HOST_PROC / "loadavg").split()
    return {
        "usagePercent": cpu_usage_percent(),
        "logicalCores": logical_cores,
        "model": model,
        "load1": float(load_values[0]),
        "load5": float(load_values[1]),
        "load15": float(load_values[2]),
    }


def memory_info():
    values = {}
    for line in read_text(HOST_PROC / "meminfo").splitlines():
        key, raw_value = line.split(":", 1)
        values[key] = int(raw_value.strip().split()[0]) * 1024

    total = values["MemTotal"]
    available = values["MemAvailable"]
    used = total - available
    return {
        "usedBytes": used,
        "availableBytes": available,
        "totalBytes": total,
        "usagePercent": round(used / total * 100, 1),
    }


def root_device():
    for line in read_text(HOST_PROC / "1" / "mountinfo").splitlines():
        fields = line.split()
        if len(fields) < 10 or fields[4] != "/" or "-" not in fields:
            continue
        separator = fields.index("-")
        if len(fields) > separator + 2:
            return fields[separator + 2]
    return "root filesystem"


def disk_info():
    stats = os.statvfs(DISK_PATH)
    total = stats.f_blocks * stats.f_frsize
    free = stats.f_bfree * stats.f_frsize
    available = stats.f_bavail * stats.f_frsize
    used = total - free
    return {
        "device": root_device(),
        "mount": "/",
        "usedBytes": used,
        "availableBytes": available,
        "totalBytes": total,
        "usagePercent": round(used / total * 100, 1),
    }


def optional_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def gpu_info():
    query = ",".join([
        "index",
        "name",
        "utilization.gpu",
        "memory.used",
        "memory.total",
        "temperature.gpu",
        "power.draw",
        "power.limit",
    ])

    try:
        result = subprocess.run(
            [
                "nvidia-smi",
                f"--query-gpu={query}",
                "--format=csv,noheader,nounits",
            ],
            capture_output=True,
            check=True,
            text=True,
            timeout=5,
        )
    except (FileNotFoundError, subprocess.SubprocessError):
        return {"available": False, "devices": []}

    devices = []
    for line in result.stdout.splitlines():
        fields = [field.strip() for field in line.split(",")]
        if len(fields) != 8:
            continue
        devices.append({
            "index": int(fields[0]),
            "name": fields[1],
            "utilizationPercent": optional_float(fields[2]),
            "memoryUsedMiB": optional_float(fields[3]),
            "memoryTotalMiB": optional_float(fields[4]),
            "temperatureCelsius": optional_float(fields[5]),
            "powerDrawWatts": optional_float(fields[6]),
            "powerLimitWatts": optional_float(fields[7]),
        })

    return {"available": bool(devices), "devices": devices}


def host_name():
    hostname_path = Path("/host/hostname")
    if hostname_path.exists():
        return read_text(hostname_path).strip()
    return "homelab"


def collect_snapshot():
    global cached_snapshot, cached_at

    now = time.monotonic()
    with cache_lock:
        if cached_snapshot is not None and now - cached_at < CACHE_TTL_SECONDS:
            return cached_snapshot

        cached_snapshot = {
            "generatedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
            "hostname": host_name(),
            "cpu": cpu_info(),
            "memory": memory_info(),
            "disk": disk_info(),
            "gpu": gpu_info(),
        }
        cached_at = time.monotonic()
        return cached_snapshot


class MetricsHandler(BaseHTTPRequestHandler):
    def send_json(self, status_code, payload):
        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode()
        self.send_response(status_code)
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Type", "application/json; charset=utf-8")
        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, {"status": "ok"})
            return
        if self.path != "/metrics":
            self.send_json(404, {"error": "Not found"})
            return

        try:
            self.send_json(200, collect_snapshot())
        except Exception as error:
            print(f"metrics collection failed: {error}", flush=True)
            self.send_json(503, {"error": "Metrics unavailable"})

    def log_message(self, message_format, *args):
        return


ThreadingHTTPServer(("0.0.0.0", PORT), MetricsHandler).serve_forever()