"""System metrics ingestion."""
import subprocess
import logging
from datetime import datetime, timezone
import psutil
from ..models import MetricSnapshot
logger = logging.getLogger(__name__)
class SystemMetricsIngestor:
"""Collect system metrics."""
def get_snapshot(self) -> MetricSnapshot:
"""Capture current system metrics."""
try:
cpu = psutil.cpu_percent(interval=0.1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage("/")
load = psutil.getloadavg()
net = psutil.net_io_counters()
# Top processes by CPU
processes = []
for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]):
try:
processes.append({
"pid": proc.info["pid"],
"name": proc.info["name"],
"cpu": proc.info["cpu_percent"],
"memory": proc.info["memory_percent"],
})
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
# Sort by CPU, take top 10
processes.sort(key=lambda p: p.get("cpu", 0) or 0, reverse=True)
return MetricSnapshot(
timestamp=datetime.now(timezone.utc),
cpu_percent=cpu,
memory_percent=memory.percent,
disk_percent=disk.percent,
load_avg=load,
network_bytes_in=net.bytes_sent,
network_bytes_out=net.bytes_recv,
top_processes=processes[:10],
)
except Exception as e:
logger.error("Failed to collect metrics: %s", e)
return MetricSnapshot(timestamp=datetime.now(timezone.utc))
def get_process_info(self, pattern: str) -> list[dict]:
"""Get processes matching a name pattern."""
processes = []
for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent", "create_time", "status"]):
try:
if pattern.lower() in (proc.info["name"] or "").lower():
processes.append({
"pid": proc.info["pid"],
"name": proc.info["name"],
"cpu": proc.info["cpu_percent"],
"memory": proc.info["memory_percent"],
"status": proc.info["status"],
"created": datetime.fromtimestamp(
proc.info["create_time"], tz=timezone.utc
).isoformat() if proc.info["create_time"] else "unknown",
})
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
return processes
def get_disk_io(self) -> dict:
"""Get disk I/O counters."""
try:
counters = psutil.disk_io_counters()
if counters:
return {
"read_count": counters.read_count,
"write_count": counters.write_count,
"read_bytes": counters.read_bytes,
"write_bytes": counters.write_bytes,
"read_time": counters.read_time,
"write_time": counters.write_time,
}
except Exception as e:
logger.warning("Failed to get disk IO: %s", e)
return {}
def get_open_files_summary(self) -> dict:
"""Get summary of open file descriptors."""
try:
import resource
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
return {"soft_limit": soft, "hard_limit": hard}
except Exception:
return {}