guyg1474andClaude Opus 5.5
c6eef176a2
Misst CPU, RAM, Swap, Platten, Netzwerk, Temperaturen und Prozesse des Hosts und liefert sie als JSON-API (/api/metrics, /api/history) für das P42-Dashboard. Nur Python-Standardbibliothek, Token-geschützt, Container ohne Root und mit schreibgeschütztem Dateisystem. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
109 lines
4.1 KiB
Python
109 lines
4.1 KiB
Python
"""HTTP-API: liefert die zuletzt gemessenen Werte als JSON.
|
||
|
||
GET /api/health ohne Token – für Docker-Healthcheck und Erreichbarkeitsprüfung
|
||
GET /api/metrics aktueller Stand (Token nötig)
|
||
GET /api/history Verlauf, optional ?minutes=N (Token nötig)
|
||
"""
|
||
import hmac
|
||
import json
|
||
import logging
|
||
import os
|
||
import threading
|
||
import time
|
||
from collections import deque
|
||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
from . import collector
|
||
|
||
log = logging.getLogger("p42-sm")
|
||
|
||
TOKEN = os.environ.get("SM_TOKEN", "").strip()
|
||
INTERVAL = max(1, int(os.environ.get("SM_INTERVAL", "5")))
|
||
HISTORY = max(1, int(os.environ.get("SM_HISTORY", "720"))) # 720 × 5 s = 1 Stunde
|
||
|
||
_lock = threading.Lock()
|
||
_latest: dict | None = None
|
||
_history: deque = deque(maxlen=HISTORY)
|
||
_last_update = 0.0
|
||
|
||
|
||
def _collect_loop() -> None:
|
||
global _latest, _last_update
|
||
prev = collector.raw_sample()
|
||
time.sleep(1) # erste Messung schon nach 1 s, damit die API schnell Daten hat
|
||
while True:
|
||
try:
|
||
cur = collector.raw_sample()
|
||
snap = collector.snapshot(prev, cur, INTERVAL)
|
||
with _lock:
|
||
_latest = snap
|
||
_history.append(collector.history_point(snap))
|
||
_last_update = time.monotonic()
|
||
prev = cur
|
||
except Exception:
|
||
log.exception("Messung fehlgeschlagen")
|
||
time.sleep(INTERVAL)
|
||
|
||
|
||
class Handler(BaseHTTPRequestHandler):
|
||
server_version = "P42-SM"
|
||
sys_version = ""
|
||
|
||
def _send(self, status: int, body: dict) -> None:
|
||
data = json.dumps(body, ensure_ascii=False).encode()
|
||
self.send_response(status)
|
||
self.send_header("Content-Type", "application/json; charset=utf-8")
|
||
self.send_header("Content-Length", str(len(data)))
|
||
self.send_header("Cache-Control", "no-store")
|
||
self.end_headers()
|
||
self.wfile.write(data)
|
||
|
||
def _authorized(self) -> bool:
|
||
if not TOKEN:
|
||
return True
|
||
header = self.headers.get("Authorization", "")
|
||
scheme, _, given = header.partition(" ")
|
||
return scheme.lower() == "bearer" and hmac.compare_digest(given.strip().encode(), TOKEN.encode())
|
||
|
||
def do_GET(self) -> None:
|
||
url = urlparse(self.path)
|
||
with _lock:
|
||
latest, history, age = _latest, list(_history), time.monotonic() - _last_update
|
||
|
||
if url.path == "/api/health":
|
||
# Ungesund, wenn die Messung hängt (drei Intervalle ohne neue Werte)
|
||
ok = latest is not None and age < INTERVAL * 3 + 5
|
||
return self._send(200 if ok else 503, {"status": "ok" if ok else "starting"})
|
||
|
||
if url.path not in ("/api/metrics", "/api/history"):
|
||
return self._send(404, {"error": "Nicht gefunden."})
|
||
if not self._authorized():
|
||
return self._send(401, {"error": "Token fehlt oder ist falsch."})
|
||
if latest is None:
|
||
return self._send(503, {"error": "Noch keine Messwerte – bitte gleich erneut versuchen."})
|
||
|
||
if url.path == "/api/metrics":
|
||
return self._send(200, latest)
|
||
|
||
minutes = parse_qs(url.query).get("minutes", [""])[0]
|
||
if minutes.isdigit():
|
||
history = history[-max(1, int(minutes) * 60 // INTERVAL):]
|
||
return self._send(200, {"intervalSeconds": INTERVAL, "points": history})
|
||
|
||
def log_message(self, fmt: str, *args) -> None:
|
||
# Nur Fehler loggen – der Abruf alle paar Sekunden würde das Log sonst fluten
|
||
if len(args) >= 2 and str(args[1]).startswith(("4", "5")):
|
||
log.warning("%s %s", self.address_string(), fmt % args)
|
||
|
||
|
||
def serve() -> None:
|
||
host = os.environ.get("HOST", "0.0.0.0")
|
||
port = int(os.environ.get("PORT", "9142"))
|
||
if not TOKEN:
|
||
log.warning("SM_TOKEN ist leer – die API ist ohne Anmeldung abrufbar!")
|
||
threading.Thread(target=_collect_loop, daemon=True, name="collector").start()
|
||
httpd = ThreadingHTTPServer((host, port), Handler)
|
||
log.info("P42-S.M. läuft auf %s:%s (Messintervall %s s, Verlauf %s Punkte)", host, port, INTERVAL, HISTORY)
|
||
httpd.serve_forever()
|