menzeljandClaude Fable 5
a0dda120f5
- backend/version.py is now the single version source (main.py, agent). - GET /api/system/update: reads the version tags of the backend's own image repo (anonymous v2 token flow, https→http fallback for insecure registries), compares the highest semver tag against APP_VERSION; reports update_supported from the container's compose labels. 10 min cache. - POST /api/system/update (admin, audited): spawns a detached helper container from the current backend image that runs docker compose pull && up -d on StackPilot's own compose project (project name, working dir and config files resolved from its own container labels) — the helper outlives the backend being recreated. Non-compose installs get a 400. - /api/health now returns the version so the UI can detect the switchover. - TopNav version badge: queries the update status on page load; when a newer release exists an amber pill shows the version — one click (admin) confirms, triggers the update and overlays a wait screen that polls /api/health and reloads once the new version answers. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
136 lines
4.1 KiB
Python
136 lines
4.1 KiB
Python
"""Host / Docker system information."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import shutil
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from sqlmodel import Session
|
|
|
|
from auth import get_current_user, require_admin
|
|
from config import settings
|
|
from database import get_session
|
|
from docker_client import DockerError, get_client, safe_call
|
|
from models.user import User
|
|
from services import audit_service, device_service, gpu_service, self_update_service
|
|
|
|
router = APIRouter(prefix="/api/system", tags=["system"])
|
|
|
|
|
|
def _read_proc(path: str) -> str:
|
|
full = os.path.join(settings.HOST_PROC_PATH, path)
|
|
if not os.path.isfile(full):
|
|
full = os.path.join("/proc", path)
|
|
try:
|
|
with open(full, "r", encoding="utf-8") as fh:
|
|
return fh.read()
|
|
except OSError:
|
|
return ""
|
|
|
|
|
|
def _mem_info() -> dict:
|
|
info = {}
|
|
for line in _read_proc("meminfo").splitlines():
|
|
parts = line.split(":")
|
|
if len(parts) == 2:
|
|
key = parts[0].strip()
|
|
val = parts[1].strip().split()[0]
|
|
try:
|
|
info[key] = int(val) * 1024 # kB -> bytes
|
|
except ValueError:
|
|
pass
|
|
total = info.get("MemTotal", 0)
|
|
available = info.get("MemAvailable", info.get("MemFree", 0))
|
|
return {"total": total, "available": available, "used": max(total - available, 0)}
|
|
|
|
|
|
def _uptime() -> float:
|
|
raw = _read_proc("uptime")
|
|
try:
|
|
return float(raw.split()[0])
|
|
except (IndexError, ValueError):
|
|
return 0.0
|
|
|
|
|
|
def _cpu_count() -> int:
|
|
return os.cpu_count() or 0
|
|
|
|
|
|
def _disk_usage() -> dict:
|
|
try:
|
|
usage = shutil.disk_usage(settings.DATA_DIR)
|
|
return {"total": usage.total, "used": usage.used, "free": usage.free}
|
|
except OSError:
|
|
return {"total": 0, "used": 0, "free": 0}
|
|
|
|
|
|
@router.get("/info")
|
|
def system_info(_user: User = Depends(get_current_user)) -> dict:
|
|
docker_version = ""
|
|
host_os = ""
|
|
containers_running = 0
|
|
containers_total = 0
|
|
try:
|
|
client = get_client()
|
|
version = safe_call(client.version)
|
|
docker_version = version.get("Version", "")
|
|
info = safe_call(client.info)
|
|
host_os = info.get("OperatingSystem", "")
|
|
containers_running = info.get("ContainersRunning", 0)
|
|
containers_total = info.get("Containers", 0)
|
|
except DockerError as exc:
|
|
docker_version = f"unavailable ({exc.error})"
|
|
|
|
return {
|
|
"docker_version": docker_version,
|
|
"host_os": host_os,
|
|
"hostname": os.uname().nodename,
|
|
"cpu_cores": _cpu_count(),
|
|
"ram": _mem_info(),
|
|
"disk": _disk_usage(),
|
|
"uptime_seconds": _uptime(),
|
|
"containers_running": containers_running,
|
|
"containers_total": containers_total,
|
|
"gpus": [g.to_dict() for g in gpu_service.detect_gpus()],
|
|
}
|
|
|
|
|
|
@router.get("/gpus")
|
|
def gpus(_user: User = Depends(get_current_user)) -> list[dict]:
|
|
"""Re-detect GPUs live."""
|
|
return [g.to_dict() for g in gpu_service.detect_gpus()]
|
|
|
|
|
|
@router.get("/devices")
|
|
def devices(_user: User = Depends(get_current_user)) -> dict:
|
|
"""List host USB / serial / DRI devices for passthrough."""
|
|
return device_service.detect_devices()
|
|
|
|
|
|
@router.get("/update")
|
|
async def self_update_status(
|
|
refresh: bool = False,
|
|
_user: User = Depends(get_current_user),
|
|
) -> dict:
|
|
"""Is a newer StackPilot release available? (registry check, cached)"""
|
|
return await self_update_service.get_status(refresh=refresh)
|
|
|
|
|
|
@router.post("/update")
|
|
def self_update_apply(
|
|
request: Request,
|
|
session: Session = Depends(get_session),
|
|
user: User = Depends(require_admin),
|
|
) -> dict:
|
|
"""Update this StackPilot in place via a detached compose helper."""
|
|
try:
|
|
result = self_update_service.apply_update()
|
|
except self_update_service.SelfUpdateError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc))
|
|
audit_service.record(
|
|
session, user=user.username, action="system.update",
|
|
target=result.get("helper", ""), detail=result.get("command"),
|
|
ip=request.client.host if request.client else "",
|
|
)
|
|
return result
|