feat: add system stats API for dashboard monitoring
Build and Push / build (release) Successful in 1m44s

- GET /tools/system/stats - Real-time host system statistics
- CPU usage, memory, all mounted disks, network I/O
- GPU/VRAM stats via nvidia-smi (if available)
- Uses psutil for cross-platform host metrics
- Auto-discovers and filters real filesystems

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Jeroen Schweitzer
2026-01-03 17:30:45 +01:00
co-authored by Claude Opus 4.5
parent 49be935b5d
commit 075b0ec297
8 changed files with 468 additions and 3 deletions
+13
View File
@@ -5,6 +5,19 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [1.7.0] - 2026-01-03
### Added
- **System Stats API** - Host system resource monitoring for dashboard widgets
- `GET /tools/system/stats` - Real-time host system statistics
- CPU: usage percentage, core count, load averages
- Memory: usage percentage, total/used/available bytes
- Disks: all mounted filesystems with usage stats (auto-discovers mounts)
- Network: total bytes sent/received
- GPU/VRAM: NVIDIA GPU memory usage (via nvidia-smi if available)
- `psutil` dependency for cross-platform system metrics
## [1.6.1] - 2026-01-03
### Added
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "core-api"
version = "1.6.1"
version = "1.7.0"
description = "Core Code API - Infrastructure management and tools API"
readme = "README.md"
requires-python = ">=3.12"
+1
View File
@@ -21,6 +21,7 @@ python-dotenv~=1.0.0
python-json-logger~=2.0.0
pytz~=2024.1
dnspython~=2.7.0
psutil~=6.1.0
# Authentication & Security
PyJWT[crypto]>=2.9.0
+9 -2
View File
@@ -1,9 +1,16 @@
"""
Tools Domain
Provides utility tool endpoints including DNS lookups.
Provides utility tool endpoints including DNS lookups and system stats.
"""
from src.domains.tools.controller import tools_controller
from src.domains.tools.dns import DNSService, DNSQueryError
from src.domains.tools.system import SystemStatsService, SystemStatsResponse
__all__ = ["tools_controller", "DNSService", "DNSQueryError"]
__all__ = [
"tools_controller",
"DNSService",
"DNSQueryError",
"SystemStatsService",
"SystemStatsResponse",
]
+51
View File
@@ -3,6 +3,7 @@ Tools Controller
Provides utility tool endpoints including:
- DNS lookups
- System stats
"""
from fastapi import APIRouter, HTTPException, status
@@ -11,6 +12,8 @@ from src.shared.logging import get_logger
from src.domains.tools.dns.schemas import DNSLookupRequest, DNSLookupResponse
from src.domains.tools.dns.service import DNSService
from src.domains.tools.dns.exceptions import DNSQueryError
from src.domains.tools.system.schemas import SystemStatsResponse
from src.domains.tools.system.service import SystemStatsService
logger = get_logger(__name__)
@@ -21,11 +24,13 @@ class ToolsController(BaseController):
Provides endpoints for:
- DNS lookups
- System stats
"""
def __init__(self):
super().__init__(prefix="/tools", tags=["Tools"])
self.dns_service = DNSService()
self.system_stats_service = SystemStatsService()
def create_router(self) -> APIRouter:
"""Create and configure the router"""
@@ -95,6 +100,52 @@ class ToolsController(BaseController):
detail="An unexpected error occurred during DNS lookup"
)
@router.get(
"/system/stats",
response_model=SystemStatsResponse,
status_code=status.HTTP_200_OK,
summary="Get host system statistics",
description="""
Get real-time host system resource statistics.
Returns CPU, memory, disk, network, and GPU/VRAM usage for the host machine
(not Docker container metrics).
**Metrics Returned:**
- **CPU:** Usage percentage, core count, load averages
- **Memory:** Usage percentage, total/used/available bytes
- **Disk:** Usage percentage, total/used/free bytes (root partition)
- **Network:** Total bytes sent/received
- **GPU:** VRAM usage (if NVIDIA GPU available via nvidia-smi)
**Use Cases:**
- Dashboard system monitoring widgets
- Health checks and alerting
- Capacity planning
"""
)
async def get_system_stats() -> SystemStatsResponse:
"""
Get current host system statistics
Returns:
System statistics including CPU, memory, disk, network, and GPU
Raises:
HTTPException: 500 for processing errors
"""
try:
logger.info("Fetching system stats")
result = await self.system_stats_service.get_stats()
return result
except Exception as e:
logger.error(f"Failed to get system stats: {str(e)}", exc_info=True)
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Failed to collect system stats: {str(e)}"
)
return router
+6
View File
@@ -0,0 +1,6 @@
"""System stats module for host system resource monitoring."""
from src.domains.tools.system.service import SystemStatsService
from src.domains.tools.system.schemas import SystemStatsResponse
__all__ = ["SystemStatsService", "SystemStatsResponse"]
+197
View File
@@ -0,0 +1,197 @@
"""
Pydantic schemas for system stats module
"""
from pydantic import Field
from typing import Optional, List
from datetime import datetime
from src.shared.base import BaseSchema
class CpuStats(BaseSchema):
"""CPU usage statistics"""
usage_percent: float = Field(
...,
description="CPU usage percentage (0-100)",
ge=0,
le=100
)
cores: int = Field(
...,
description="Number of CPU cores"
)
load_1m: Optional[float] = Field(
default=None,
description="1-minute load average"
)
load_5m: Optional[float] = Field(
default=None,
description="5-minute load average"
)
load_15m: Optional[float] = Field(
default=None,
description="15-minute load average"
)
class MemoryStats(BaseSchema):
"""Memory usage statistics"""
usage_percent: float = Field(
...,
description="Memory usage percentage (0-100)",
ge=0,
le=100
)
total_bytes: int = Field(
...,
description="Total memory in bytes"
)
used_bytes: int = Field(
...,
description="Used memory in bytes"
)
available_bytes: int = Field(
...,
description="Available memory in bytes"
)
class DiskStats(BaseSchema):
"""Disk usage statistics for a single mount point"""
mount_point: str = Field(
...,
description="Mount point path"
)
device: str = Field(
...,
description="Device name (e.g., /dev/sda1)"
)
fstype: str = Field(
...,
description="Filesystem type (e.g., ext4, xfs)"
)
usage_percent: float = Field(
...,
description="Disk usage percentage (0-100)",
ge=0,
le=100
)
total_bytes: int = Field(
...,
description="Total disk space in bytes"
)
used_bytes: int = Field(
...,
description="Used disk space in bytes"
)
free_bytes: int = Field(
...,
description="Free disk space in bytes"
)
class NetworkStats(BaseSchema):
"""Network I/O statistics"""
bytes_sent: int = Field(
...,
description="Total bytes sent"
)
bytes_recv: int = Field(
...,
description="Total bytes received"
)
bytes_total: int = Field(
...,
description="Total bytes (sent + received)"
)
class GpuStats(BaseSchema):
"""GPU/VRAM statistics (if available)"""
available: bool = Field(
...,
description="Whether GPU stats are available"
)
name: Optional[str] = Field(
default=None,
description="GPU name"
)
usage_percent: Optional[float] = Field(
default=None,
description="VRAM usage percentage (0-100)"
)
total_bytes: Optional[int] = Field(
default=None,
description="Total VRAM in bytes"
)
used_bytes: Optional[int] = Field(
default=None,
description="Used VRAM in bytes"
)
free_bytes: Optional[int] = Field(
default=None,
description="Free VRAM in bytes"
)
class SystemStatsResponse(BaseSchema):
"""Response model for system stats"""
cpu: CpuStats = Field(
...,
description="CPU statistics"
)
memory: MemoryStats = Field(
...,
description="Memory statistics"
)
disks: List[DiskStats] = Field(
...,
description="Disk statistics for all mounted filesystems"
)
network: NetworkStats = Field(
...,
description="Network I/O statistics"
)
gpu: GpuStats = Field(
...,
description="GPU/VRAM statistics"
)
hostname: str = Field(
...,
description="System hostname"
)
queried_at: datetime = Field(
...,
description="UTC timestamp when stats were collected"
)
+190
View File
@@ -0,0 +1,190 @@
"""
System stats service for collecting host system metrics
"""
import subprocess
import socket
from datetime import datetime, timezone
import psutil
from src.shared.logging import get_logger
from typing import List
from src.domains.tools.system.schemas import (
SystemStatsResponse,
CpuStats,
MemoryStats,
DiskStats,
NetworkStats,
GpuStats,
)
# Filesystem types to exclude (virtual/system filesystems)
EXCLUDED_FSTYPES = {
"tmpfs", "devtmpfs", "devfs", "squashfs", "overlay",
"aufs", "proc", "sysfs", "cgroup", "cgroup2",
"debugfs", "tracefs", "securityfs", "pstore",
"hugetlbfs", "mqueue", "binfmt_misc", "autofs",
"fuse.lxcfs", "nsfs", "efivarfs",
}
logger = get_logger(__name__)
class SystemStatsService:
"""Service for collecting host system statistics"""
async def get_stats(self) -> SystemStatsResponse:
"""
Collect current system statistics.
Returns:
SystemStatsResponse with CPU, memory, disks, network, and GPU stats
"""
cpu = self._get_cpu_stats()
memory = self._get_memory_stats()
disks = self._get_all_disk_stats()
network = self._get_network_stats()
gpu = self._get_gpu_stats()
return SystemStatsResponse(
cpu=cpu,
memory=memory,
disks=disks,
network=network,
gpu=gpu,
hostname=socket.gethostname(),
queried_at=datetime.now(timezone.utc),
)
def _get_cpu_stats(self) -> CpuStats:
"""Get CPU usage statistics"""
# Get CPU percentage (blocking call with interval for accuracy)
cpu_percent = psutil.cpu_percent(interval=0.1)
cpu_count = psutil.cpu_count()
# Get load averages (Unix only)
try:
load_1, load_5, load_15 = psutil.getloadavg()
except (AttributeError, OSError):
load_1 = load_5 = load_15 = None
return CpuStats(
usage_percent=cpu_percent,
cores=cpu_count or 1,
load_1m=load_1,
load_5m=load_5,
load_15m=load_15,
)
def _get_memory_stats(self) -> MemoryStats:
"""Get memory usage statistics"""
mem = psutil.virtual_memory()
return MemoryStats(
usage_percent=mem.percent,
total_bytes=mem.total,
used_bytes=mem.used,
available_bytes=mem.available,
)
def _get_all_disk_stats(self) -> List[DiskStats]:
"""Get disk usage statistics for all mounted real filesystems"""
disks = []
seen_devices = set()
for partition in psutil.disk_partitions(all=False):
# Skip excluded filesystem types
if partition.fstype.lower() in EXCLUDED_FSTYPES:
continue
# Skip duplicate devices (same device mounted multiple times)
if partition.device in seen_devices:
continue
seen_devices.add(partition.device)
# Skip Docker/container overlays
if partition.mountpoint.startswith("/var/lib/docker"):
continue
try:
usage = psutil.disk_usage(partition.mountpoint)
disks.append(DiskStats(
mount_point=partition.mountpoint,
device=partition.device,
fstype=partition.fstype,
usage_percent=usage.percent,
total_bytes=usage.total,
used_bytes=usage.used,
free_bytes=usage.free,
))
except (PermissionError, OSError) as e:
logger.debug(f"Skipping {partition.mountpoint}: {e}")
continue
# Sort by mount point for consistent ordering
disks.sort(key=lambda d: d.mount_point)
return disks
def _get_network_stats(self) -> NetworkStats:
"""Get network I/O statistics"""
net_io = psutil.net_io_counters()
return NetworkStats(
bytes_sent=net_io.bytes_sent,
bytes_recv=net_io.bytes_recv,
bytes_total=net_io.bytes_sent + net_io.bytes_recv,
)
def _get_gpu_stats(self) -> GpuStats:
"""Get GPU/VRAM statistics using nvidia-smi"""
try:
# Query nvidia-smi for GPU memory info
result = subprocess.run(
[
"nvidia-smi",
"--query-gpu=name,memory.total,memory.used,memory.free",
"--format=csv,noheader,nounits",
],
capture_output=True,
text=True,
timeout=5,
)
if result.returncode != 0:
logger.debug("nvidia-smi not available or failed")
return GpuStats(available=False)
# Parse output: "NVIDIA GeForce RTX 3080, 10240, 2048, 8192"
line = result.stdout.strip().split("\n")[0] # First GPU
parts = [p.strip() for p in line.split(",")]
if len(parts) >= 4:
name = parts[0]
total_mb = int(parts[1])
used_mb = int(parts[2])
free_mb = int(parts[3])
total_bytes = total_mb * 1024 * 1024
used_bytes = used_mb * 1024 * 1024
free_bytes = free_mb * 1024 * 1024
usage_percent = (used_mb / total_mb * 100) if total_mb > 0 else 0
return GpuStats(
available=True,
name=name,
usage_percent=round(usage_percent, 1),
total_bytes=total_bytes,
used_bytes=used_bytes,
free_bytes=free_bytes,
)
except FileNotFoundError:
logger.debug("nvidia-smi not found - no NVIDIA GPU available")
except subprocess.TimeoutExpired:
logger.warning("nvidia-smi timed out")
except Exception as e:
logger.warning(f"Failed to get GPU stats: {e}")
return GpuStats(available=False)