- Flask backend + static SPA (HTML/CSS/JS) - Docker container discovery via unix socket (stdlib only, no docker SDK) - Host stats (CPU/mem/disk/load/uptime) via psutil - Live-updating UI with per-state filters and resource gauges - Dockerfile for containerized deployment
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
"""Host resource stats via psutil, with a short cache."""
|
|
|
|
import time
|
|
|
|
import psutil
|
|
|
|
# Cache TTL seconds.
|
|
_CACHE_TTL = 2.0
|
|
|
|
|
|
class HostStats:
|
|
"""Collects CPU, memory, disk and load stats for the host."""
|
|
|
|
def __init__(self, cache_ttl=_CACHE_TTL):
|
|
self.cache_ttl = cache_ttl
|
|
self._cache = None
|
|
self._cached_at = 0.0
|
|
|
|
def _fresh(self):
|
|
return time.time() - self._cached_at >= self.cache_ttl
|
|
|
|
def refresh(self):
|
|
if not self._fresh():
|
|
return self._cache
|
|
|
|
cpu_percent = psutil.cpu_percent(interval=0.3)
|
|
cpu_count = psutil.cpu_count(logical=True)
|
|
|
|
mem = psutil.virtual_memory()
|
|
mem = {
|
|
"total_mb": round(mem.total / 1024 / 1024, 0),
|
|
"used_mb": round(mem.used / 1024 / 1024, 0),
|
|
"available_mb": round(mem.available / 1024 / 1024, 0),
|
|
"percent": mem.percent,
|
|
}
|
|
|
|
disk = {}
|
|
for part in psutil.disk_partitions(all=False):
|
|
if part.opts.split(",")[0] in ("rw", "") or "rw" in part.opts:
|
|
try:
|
|
usage = psutil.disk_usage(part.mountpoint)
|
|
disk[part.mountpoint] = {
|
|
"total_gb": round(usage.total / 1024 / 1024 / 1024, 1),
|
|
"used_gb": round(usage.used / 1024 / 1024 / 1024, 1),
|
|
"percent": usage.percent,
|
|
}
|
|
except (PermissionError, OSError):
|
|
continue
|
|
# Always include / at minimum.
|
|
if "/" not in disk:
|
|
try:
|
|
usage = psutil.disk_usage("/")
|
|
disk["/"] = {
|
|
"total_gb": round(usage.total / 1024 / 1024 / 1024, 1),
|
|
"used_gb": round(usage.used / 1024 / 1024 / 1024, 1),
|
|
"percent": usage.percent,
|
|
}
|
|
except (PermissionError, OSError):
|
|
pass
|
|
|
|
boot_time = psutil.boot_time()
|
|
|
|
load = {}
|
|
try:
|
|
l1, l5, l15 = psutil.getloadavg()
|
|
load = {"1min": round(l1, 2), "5min": round(l5, 2), "15min": round(l15, 2)}
|
|
except (OSError, AttributeError):
|
|
load = {}
|
|
|
|
self._cache = {
|
|
"cpu_percent": cpu_percent,
|
|
"cpu_count": cpu_count,
|
|
"cpu_cores": cpu_count,
|
|
"memory": mem,
|
|
"disk": disk,
|
|
"boot_time": boot_time,
|
|
"uptime_seconds": round(time.time() - boot_time, 0),
|
|
"load": load,
|
|
"timestamp": time.time(),
|
|
}
|
|
self._cached_at = time.time()
|
|
return self._cache
|
|
|
|
def get(self):
|
|
return self.refresh()
|