1009 lines
37 KiB
Python
1009 lines
37 KiB
Python
"""Linux system metric collection using /proc, /sys and statvfs.
|
|
|
|
The module intentionally depends only on the Python standard library so the
|
|
dashboard can run on a bare server. Every parser is a pure function and can be
|
|
tested with synthetic /proc data (see tests/test_metrics.py).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import glob
|
|
import json
|
|
import os
|
|
import platform
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Dict, List, Optional, Sequence, Tuple
|
|
|
|
try: # package import
|
|
from . import __version__
|
|
except ImportError: # pragma: no cover - direct module execution
|
|
__version__ = "1.0.0"
|
|
|
|
|
|
# Filesystems worth showing on a storage dashboard. Pseudo filesystems are
|
|
# deliberately excluded so the panel does not fill with tmpfs/proc entries.
|
|
REAL_FS_TYPES = {
|
|
"ext2", "ext3", "ext4", "xfs", "btrfs", "zfs", "f2fs", "jfs", "reiserfs",
|
|
"bcachefs", "vfat", "exfat", "ntfs", "ntfs3", "fuseblk", "overlay", "aufs",
|
|
"nfs", "nfs4", "cifs", "smb3", "ceph", "glusterfs", "lustre", "9p",
|
|
}
|
|
|
|
VIRTUAL_IFACE_PREFIXES = (
|
|
"lo", "veth", "docker", "br-", "virbr", "cni", "flannel", "kube", "cali",
|
|
"tunl", "vxlan", "dummy", "ifb", "lxc", "podman",
|
|
)
|
|
|
|
# Linux /proc/stat fields, in order after the "cpu" label.
|
|
_CPU_FIELDS = ("user", "nice", "system", "idle", "iowait", "irq", "softirq", "steal")
|
|
|
|
_MEMINFO_KEYS = {
|
|
"MemTotal": "total",
|
|
"MemFree": "free",
|
|
"MemAvailable": "available",
|
|
"Buffers": "buffers",
|
|
"Cached": "cached",
|
|
"SReclaimable": "sreclaimable",
|
|
"Shmem": "shmem",
|
|
"SwapTotal": "swap_total",
|
|
"SwapFree": "swap_free",
|
|
"Dirty": "dirty",
|
|
"Writeback": "writeback",
|
|
"Slab": "slab",
|
|
}
|
|
|
|
_STAT_RE = re.compile(r"^(\d+) \((.+)\) ([A-Za-z]) (.*)$")
|
|
|
|
|
|
def _read_text(path: Path) -> str:
|
|
try:
|
|
return path.read_text(encoding="utf-8", errors="replace")
|
|
except (OSError, ValueError):
|
|
return ""
|
|
|
|
|
|
def _as_int(value: str, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _pct(used: float, total: float) -> float:
|
|
if total <= 0:
|
|
return 0.0
|
|
return max(0.0, min(100.0, used / total * 100.0))
|
|
|
|
|
|
def _round(value: Optional[float], digits: int = 1) -> Optional[float]:
|
|
if value is None:
|
|
return None
|
|
return round(float(value), digits)
|
|
|
|
|
|
def parse_cpu_stat(text: str) -> Tuple[Optional[Dict[str, int]], List[Dict[str, int]]]:
|
|
"""Parse /proc/stat CPU times into aggregate and per-core dictionaries."""
|
|
|
|
aggregate: Optional[Dict[str, int]] = None
|
|
cores: List[Dict[str, int]] = []
|
|
for line in text.splitlines():
|
|
if not line.startswith("cpu"):
|
|
continue
|
|
parts = line.split()
|
|
if len(parts) < 5:
|
|
continue
|
|
label = parts[0]
|
|
try:
|
|
values = [int(v) for v in parts[1:9]]
|
|
except ValueError:
|
|
continue
|
|
values += [0] * (8 - len(values))
|
|
item = dict(zip(_CPU_FIELDS, values))
|
|
item["total"] = sum(values)
|
|
item["idle_all"] = item["idle"] + item["iowait"]
|
|
if label == "cpu":
|
|
aggregate = item
|
|
elif label[3:].isdigit():
|
|
item["index"] = int(label[3:])
|
|
cores.append(item)
|
|
cores.sort(key=lambda c: c["index"])
|
|
return aggregate, cores
|
|
|
|
|
|
def parse_cpuinfo(text: str) -> Dict[str, Any]:
|
|
"""Extract model name, logical cores and physical core count from cpuinfo."""
|
|
|
|
model = ""
|
|
processors = 0
|
|
physical = set()
|
|
physical_id: Optional[str] = None
|
|
core_id: Optional[str] = None
|
|
for line in text.splitlines() + [""]:
|
|
if not line.strip():
|
|
if physical_id is not None and core_id is not None:
|
|
physical.add((physical_id, core_id))
|
|
physical_id = core_id = None
|
|
continue
|
|
if ":" not in line:
|
|
continue
|
|
key, value = line.split(":", 1)
|
|
key = key.strip().lower()
|
|
value = value.strip()
|
|
if key == "processor":
|
|
processors += 1
|
|
elif key in ("model name", "hardware", "processor model") and not model:
|
|
model = value
|
|
elif key == "physical id":
|
|
physical_id = value
|
|
elif key == "core id":
|
|
core_id = value
|
|
return {
|
|
"model": model,
|
|
"logical_cores": processors,
|
|
"physical_cores": len(physical) if physical else processors,
|
|
}
|
|
|
|
|
|
def parse_meminfo(text: str) -> Dict[str, int]:
|
|
"""Parse /proc/meminfo into a canonical byte-valued dictionary."""
|
|
|
|
data: Dict[str, int] = {}
|
|
for line in text.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
key, raw = line.split(":", 1)
|
|
key = key.strip()
|
|
target = _MEMINFO_KEYS.get(key)
|
|
if not target:
|
|
continue
|
|
match = re.search(r"(\d+)", raw)
|
|
if not match:
|
|
continue
|
|
# All tracked fields are reported in kB by the kernel.
|
|
data[target] = int(match.group(1)) * 1024
|
|
return data
|
|
|
|
|
|
def parse_loadavg(text: str) -> Dict[str, Any]:
|
|
parts = text.split()
|
|
if len(parts) < 5:
|
|
return {"load1": 0.0, "load5": 0.0, "load15": 0.0, "running": 0, "threads": 0}
|
|
running = threads = 0
|
|
if "/" in parts[3]:
|
|
left, right = parts[3].split("/", 1)
|
|
running, threads = _as_int(left), _as_int(right)
|
|
return {
|
|
"load1": float(parts[0]),
|
|
"load5": float(parts[1]),
|
|
"load15": float(parts[2]),
|
|
"running": running,
|
|
"threads": threads,
|
|
}
|
|
|
|
|
|
def parse_uptime(text: str) -> float:
|
|
parts = text.split()
|
|
if not parts:
|
|
return 0.0
|
|
try:
|
|
return max(0.0, float(parts[0]))
|
|
except ValueError:
|
|
return 0.0
|
|
|
|
|
|
def parse_net_dev(text: str) -> Dict[str, Dict[str, int]]:
|
|
"""Parse /proc/net/dev, returning cumulative counters by interface."""
|
|
|
|
result: Dict[str, Dict[str, int]] = {}
|
|
for line in text.splitlines():
|
|
if ":" not in line:
|
|
continue
|
|
name, raw = line.split(":", 1)
|
|
name = name.strip()
|
|
fields = raw.split()
|
|
if not name or len(fields) < 16:
|
|
continue
|
|
try:
|
|
values = [int(v) for v in fields[:16]]
|
|
except ValueError:
|
|
continue
|
|
result[name] = {
|
|
"rx_bytes": values[0],
|
|
"rx_packets": values[1],
|
|
"rx_errors": values[2],
|
|
"rx_drops": values[3],
|
|
"tx_bytes": values[8],
|
|
"tx_packets": values[9],
|
|
"tx_errors": values[10],
|
|
"tx_drops": values[11],
|
|
}
|
|
return result
|
|
|
|
|
|
def parse_diskstats(text: str) -> Dict[str, Dict[str, int]]:
|
|
"""Parse /proc/diskstats into cumulative counters by block device."""
|
|
|
|
result: Dict[str, Dict[str, int]] = {}
|
|
for line in text.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 14:
|
|
continue
|
|
name = parts[2]
|
|
try:
|
|
values = [int(v) for v in parts[3:14]]
|
|
except ValueError:
|
|
continue
|
|
result[name] = {
|
|
"reads": values[0],
|
|
"reads_merged": values[1],
|
|
"sectors_read": values[2],
|
|
"read_ms": values[3],
|
|
"writes": values[4],
|
|
"writes_merged": values[5],
|
|
"sectors_written": values[6],
|
|
"write_ms": values[7],
|
|
"io_in_progress": values[8],
|
|
"io_ms": values[9],
|
|
"weighted_io_ms": values[10],
|
|
}
|
|
return result
|
|
|
|
|
|
def _unescape_mount(value: str) -> str:
|
|
def replace(match: re.Match[str]) -> str:
|
|
return chr(int(match.group(1), 8))
|
|
|
|
return re.sub(r"\\([0-7]{3})", replace, value)
|
|
|
|
|
|
def parse_mounts(text: str) -> List[Dict[str, str]]:
|
|
mounts = []
|
|
for line in text.splitlines():
|
|
parts = line.split()
|
|
if len(parts) < 4:
|
|
continue
|
|
mounts.append({
|
|
"device": _unescape_mount(parts[0]),
|
|
"mount": _unescape_mount(parts[1]),
|
|
"fstype": parts[2],
|
|
"options": parts[3],
|
|
})
|
|
return mounts
|
|
|
|
|
|
def build_alerts(
|
|
cpu_percent: float,
|
|
memory_percent: float,
|
|
swap_percent: float,
|
|
load1: float,
|
|
cores: int,
|
|
disks: Sequence[Dict[str, Any]],
|
|
temp_c: Optional[float],
|
|
) -> List[Dict[str, str]]:
|
|
"""Build compact, user-facing alert messages for the status ticker."""
|
|
|
|
alerts: List[Dict[str, str]] = []
|
|
|
|
def add(severity: str, message: str) -> None:
|
|
alerts.append({"severity": severity, "message": message})
|
|
|
|
if cpu_percent >= 95:
|
|
add("crit", f"CPU 使用率 {cpu_percent:.0f}%")
|
|
elif cpu_percent >= 85:
|
|
add("warn", f"CPU 使用率 {cpu_percent:.0f}%")
|
|
|
|
if memory_percent >= 95:
|
|
add("crit", f"内存使用率 {memory_percent:.0f}%")
|
|
elif memory_percent >= 85:
|
|
add("warn", f"内存使用率 {memory_percent:.0f}%")
|
|
|
|
if swap_percent >= 80:
|
|
add("crit", f"Swap 已用 {swap_percent:.0f}%")
|
|
elif swap_percent >= 50:
|
|
add("warn", f"Swap 已用 {swap_percent:.0f}%")
|
|
|
|
if cores > 0 and load1 >= cores * 3:
|
|
add("crit", f"负载 {load1:.2f} / {cores} 核")
|
|
elif cores > 0 and load1 >= cores * 1.5:
|
|
add("warn", f"负载 {load1:.2f} / {cores} 核")
|
|
|
|
for disk in sorted(disks, key=lambda d: d.get("percent", 0), reverse=True):
|
|
pct = float(disk.get("percent", 0))
|
|
if pct >= 95:
|
|
add("crit", f"磁盘 {disk.get('mount', '?')} 已用 {pct:.0f}%")
|
|
elif pct >= 85:
|
|
add("warn", f"磁盘 {disk.get('mount', '?')} 已用 {pct:.0f}%")
|
|
if len([a for a in alerts if a["message"].startswith("磁盘")]) >= 2:
|
|
break
|
|
|
|
if temp_c is not None:
|
|
if temp_c >= 90:
|
|
add("crit", f"CPU 温度 {temp_c:.0f}°C")
|
|
elif temp_c >= 80:
|
|
add("warn", f"CPU 温度 {temp_c:.0f}°C")
|
|
|
|
order = {"crit": 0, "warn": 1, "info": 2}
|
|
alerts.sort(key=lambda a: order.get(a["severity"], 3))
|
|
return alerts
|
|
|
|
|
|
class SystemCollector:
|
|
"""Collect one coherent snapshot from a Linux-like /proc and /sys tree."""
|
|
|
|
def __init__(
|
|
self,
|
|
proc_root: str = "/proc",
|
|
sys_root: str = "/sys",
|
|
clock: Callable[[], float] = time.monotonic,
|
|
wall_clock: Callable[[], float] = time.time,
|
|
) -> None:
|
|
self.proc_root = Path(proc_root)
|
|
self.sys_root = Path(sys_root)
|
|
self.clock = clock
|
|
self.wall_clock = wall_clock
|
|
self._ncpu = os.cpu_count() or 1
|
|
self._page_size = self._get_page_size()
|
|
self._static = self._read_static()
|
|
|
|
self._prev_cpu: Optional[Dict[str, int]] = None
|
|
self._prev_cpu_cores: List[Dict[str, int]] = []
|
|
self._cpu_total_now = 0
|
|
self._prev_net: Dict[str, Dict[str, int]] = {}
|
|
self._prev_net_ts: Optional[float] = None
|
|
self._prev_disk: Dict[str, Dict[str, int]] = {}
|
|
self._prev_disk_ts: Optional[float] = None
|
|
self._proc_ticks: Dict[int, int] = {}
|
|
self._prev_proc_total = 0
|
|
self._proc_cache: List[Dict[str, Any]] = []
|
|
self._proc_cache_ts = 0.0
|
|
self._cpu_meta: Dict[str, Any] = {"freq_mhz": None, "temp_c": None, "temp_label": None}
|
|
self._cpu_meta_ts = 0.0
|
|
self._disk_cache: List[Dict[str, Any]] = []
|
|
self._disk_cache_ts = 0.0
|
|
self._addr_cache: Dict[str, Any] = {"interfaces": {}, "primary": None, "all": []}
|
|
self._addr_cache_ts = 0.0
|
|
|
|
def _get_page_size(self) -> int:
|
|
try:
|
|
return int(os.sysconf("SC_PAGE_SIZE"))
|
|
except (AttributeError, ValueError, OSError):
|
|
return 4096
|
|
|
|
def _read_static(self) -> Dict[str, Any]:
|
|
cpuinfo = parse_cpuinfo(_read_text(self.proc_root / "cpuinfo"))
|
|
osr: Dict[str, str] = {}
|
|
for line in _read_text(Path("/etc/os-release")).splitlines():
|
|
if "=" in line:
|
|
key, value = line.split("=", 1)
|
|
osr[key] = value.strip().strip('"')
|
|
if not osr:
|
|
for candidate in (self.proc_root.parent / "etc" / "os-release",):
|
|
for line in _read_text(candidate).splitlines():
|
|
if "=" in line:
|
|
key, value = line.split("=", 1)
|
|
osr[key] = value.strip().strip('"')
|
|
os_name = osr.get("PRETTY_NAME") or osr.get("NAME") or platform.platform()
|
|
if osr.get("VERSION") and osr.get("VERSION") not in os_name:
|
|
os_name = f"{os_name} {osr['VERSION']}"
|
|
|
|
cores = len(self._per_core_names()) or self._ncpu
|
|
physical = cpuinfo.get("physical_cores") or cores
|
|
now = dt.datetime.now().astimezone()
|
|
offset = now.utcoffset()
|
|
tz_offset = int(offset.total_seconds()) if offset else -time.timezone
|
|
tz_name = now.tzname() or time.tzname[0] or "LOCAL"
|
|
|
|
return {
|
|
"hostname": socket.gethostname() or "unknown",
|
|
"os": os_name,
|
|
"kernel": platform.release(),
|
|
"arch": platform.machine() or "unknown",
|
|
"cpu_model": cpuinfo.get("model") or platform.processor() or "Unknown CPU",
|
|
"cores": cores,
|
|
"physical_cores": physical,
|
|
"tz_offset": tz_offset,
|
|
"tz_name": tz_name,
|
|
"version": __version__,
|
|
}
|
|
|
|
def _per_core_names(self) -> List[str]:
|
|
return sorted(p.name for p in self.sys_root.glob("devices/system/cpu/cpu[0-9]*") if p.is_dir())
|
|
|
|
# ------------------------------------------------------------------
|
|
# Host / address helpers
|
|
# ------------------------------------------------------------------
|
|
def _default_interface(self) -> Optional[str]:
|
|
text = _read_text(self.proc_root / "net" / "route")
|
|
best: Optional[Tuple[int, str]] = None
|
|
for line in text.splitlines()[1:]:
|
|
fields = line.split()
|
|
if len(fields) < 8:
|
|
continue
|
|
iface, dest, _gw, flags, _ref, _use, metric, _mask = fields[:8]
|
|
try:
|
|
flag_value = int(flags, 16)
|
|
except ValueError:
|
|
continue
|
|
if dest == "00000000" and flag_value & 0x2:
|
|
try:
|
|
value = int(metric)
|
|
except ValueError:
|
|
value = 0
|
|
if best is None or value < best[0]:
|
|
best = (value, iface)
|
|
return best[1] if best else None
|
|
|
|
def _read_addresses(self) -> Dict[str, Any]:
|
|
mapping: Dict[str, List[str]] = {}
|
|
ip_bin = shutil.which("ip")
|
|
if ip_bin:
|
|
try:
|
|
proc = subprocess.run(
|
|
[ip_bin, "-j", "addr", "show"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=1.5,
|
|
check=False,
|
|
)
|
|
if proc.returncode == 0 and proc.stdout.strip():
|
|
for item in json.loads(proc.stdout):
|
|
ifname = item.get("ifname")
|
|
if not ifname:
|
|
continue
|
|
addresses = []
|
|
for addr in item.get("addr_info", []):
|
|
if addr.get("family") == "inet" and addr.get("local"):
|
|
addresses.append(addr["local"])
|
|
if addresses:
|
|
mapping[ifname] = addresses
|
|
except (OSError, ValueError, subprocess.SubprocessError):
|
|
mapping = {}
|
|
|
|
primary: Optional[str] = None
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.settimeout(0.2)
|
|
sock.connect(("1.1.1.1", 80))
|
|
primary = sock.getsockname()[0]
|
|
except OSError:
|
|
primary = None
|
|
|
|
default_iface = self._default_interface()
|
|
if primary and default_iface and primary not in mapping.get(default_iface, []):
|
|
mapping.setdefault(default_iface, []).insert(0, primary)
|
|
|
|
all_ips = sorted({ip for values in mapping.values() for ip in values})
|
|
if not primary and all_ips:
|
|
primary = all_ips[0]
|
|
return {"interfaces": mapping, "primary": primary, "all": all_ips}
|
|
|
|
def _addresses(self, now: float) -> Dict[str, Any]:
|
|
if now - self._addr_cache_ts >= 30.0 or not self._addr_cache_ts:
|
|
self._addr_cache = self._read_addresses()
|
|
self._addr_cache_ts = now
|
|
return self._addr_cache
|
|
|
|
# ------------------------------------------------------------------
|
|
# CPU / memory / load
|
|
# ------------------------------------------------------------------
|
|
def _read_cpu(self, now: float) -> Dict[str, Any]:
|
|
aggregate, cores = parse_cpu_stat(_read_text(self.proc_root / "stat"))
|
|
if aggregate is None:
|
|
return {
|
|
"percent": 0.0,
|
|
"per_core": [],
|
|
"breakdown": {},
|
|
"freq_mhz": None,
|
|
"temp_c": None,
|
|
"temp_label": None,
|
|
"model": self._static["cpu_model"],
|
|
"cores": self._static["cores"],
|
|
"physical_cores": self._static["physical_cores"],
|
|
}
|
|
|
|
self._cpu_total_now = aggregate["total"]
|
|
first = self._prev_cpu is None
|
|
if first:
|
|
used = aggregate["total"] - aggregate["idle_all"]
|
|
total = aggregate["total"]
|
|
percent = _pct(used, total)
|
|
breakdown = {
|
|
key: _pct(aggregate.get(key, 0), total)
|
|
for key in ("user", "nice", "system", "iowait", "irq", "softirq", "steal")
|
|
}
|
|
per_core = [
|
|
_pct(core["total"] - core["idle_all"], core["total"]) for core in cores
|
|
]
|
|
else:
|
|
prev = self._prev_cpu
|
|
total_delta = max(0, aggregate["total"] - prev["total"])
|
|
idle_delta = max(0, aggregate["idle_all"] - prev["idle_all"])
|
|
percent = _pct(total_delta - idle_delta, total_delta)
|
|
breakdown = {}
|
|
for key in ("user", "nice", "system", "iowait", "irq", "softirq", "steal"):
|
|
breakdown[key] = _pct(
|
|
max(0, aggregate.get(key, 0) - prev.get(key, 0)), total_delta
|
|
)
|
|
per_core = []
|
|
prev_cores = {c["index"]: c for c in self._prev_cpu_cores}
|
|
for core in cores:
|
|
old = prev_cores.get(core["index"])
|
|
if not old:
|
|
per_core.append(0.0)
|
|
continue
|
|
dt = max(0, core["total"] - old["total"])
|
|
di = max(0, core["idle_all"] - old["idle_all"])
|
|
per_core.append(_pct(dt - di, dt))
|
|
|
|
self._prev_cpu = aggregate
|
|
self._prev_cpu_cores = cores
|
|
meta = self._read_cpu_meta(now)
|
|
return {
|
|
"percent": _round(percent, 1),
|
|
"per_core": [_round(v, 1) for v in per_core],
|
|
"breakdown": {k: _round(v, 1) for k, v in breakdown.items()},
|
|
"freq_mhz": meta["freq_mhz"],
|
|
"temp_c": meta["temp_c"],
|
|
"temp_label": meta["temp_label"],
|
|
"model": self._static["cpu_model"],
|
|
"cores": len(cores) or self._static["cores"],
|
|
"physical_cores": self._static["physical_cores"],
|
|
}
|
|
|
|
def _read_cpu_meta(self, now: float) -> Dict[str, Any]:
|
|
if now - self._cpu_meta_ts < 3.0 and self._cpu_meta_ts:
|
|
return self._cpu_meta
|
|
|
|
freqs: List[float] = []
|
|
for path in glob.glob(str(self.sys_root / "devices/system/cpu/cpu*/cpufreq/scaling_cur_freq")):
|
|
try:
|
|
freqs.append(int(Path(path).read_text().strip()) / 1000.0)
|
|
except (OSError, ValueError):
|
|
continue
|
|
if not freqs:
|
|
for line in _read_text(self.proc_root / "cpuinfo").splitlines():
|
|
if line.lower().startswith("cpu mhz"):
|
|
try:
|
|
freqs.append(float(line.split(":", 1)[1].strip()))
|
|
except (IndexError, ValueError):
|
|
pass
|
|
freq = round(sum(freqs) / len(freqs)) if freqs else None
|
|
|
|
zones: List[Tuple[float, str]] = []
|
|
for path in glob.glob(str(self.sys_root / "class/thermal/thermal_zone*")):
|
|
zone = Path(path)
|
|
raw_text = _read_text(zone / "temp").strip()
|
|
try:
|
|
raw = float(raw_text)
|
|
except ValueError:
|
|
continue
|
|
celsius = raw / 1000.0 if raw > 1000 else raw
|
|
if not -20.0 <= celsius <= 150.0:
|
|
continue
|
|
label = _read_text(zone / "type").strip() or zone.name
|
|
zones.append((celsius, label))
|
|
if not zones:
|
|
for path in glob.glob(str(self.sys_root / "class/hwmon/hwmon*/temp*_input")):
|
|
pw = Path(path)
|
|
name = _read_text(pw.parent / "name").strip()
|
|
label = _read_text(pw.with_name(pw.stem.replace("_input", "_label"))).strip() or name
|
|
if not any(token in f"{name} {label}".lower() for token in ("cpu", "package", "tctl", "tdie", "core")):
|
|
continue
|
|
try:
|
|
raw = float(_read_text(pw).strip())
|
|
except ValueError:
|
|
continue
|
|
celsius = raw / 1000.0 if raw > 1000 else raw
|
|
if -20.0 <= celsius <= 150.0:
|
|
zones.append((celsius, label))
|
|
if zones:
|
|
preferred = [z for z in zones if any(token in z[1].lower() for token in ("cpu", "package", "tctl", "tdie", "core"))]
|
|
temp_c, temp_label = max(preferred or zones, key=lambda z: z[0])
|
|
temp_c = round(temp_c, 1)
|
|
else:
|
|
temp_c, temp_label = None, None
|
|
|
|
self._cpu_meta = {"freq_mhz": freq, "temp_c": temp_c, "temp_label": temp_label}
|
|
self._cpu_meta_ts = now
|
|
return self._cpu_meta
|
|
|
|
def _read_memory(self) -> Dict[str, Any]:
|
|
info = parse_meminfo(_read_text(self.proc_root / "meminfo"))
|
|
total = info.get("total", 0)
|
|
free = info.get("free", 0)
|
|
available = info.get("available") or free
|
|
buffers = info.get("buffers", 0)
|
|
cached = info.get("cached", 0) + info.get("sreclaimable", 0)
|
|
swap_total = info.get("swap_total", 0)
|
|
swap_free = info.get("swap_free", 0)
|
|
|
|
used = max(0, total - available) if total else 0
|
|
percent = _pct(used, total)
|
|
swap_used = max(0, swap_total - swap_free)
|
|
swap_percent = _pct(swap_used, swap_total)
|
|
|
|
apps = max(0, total - free - buffers - cached)
|
|
cache_value = min(cached, max(0, total - free - buffers - apps))
|
|
composition = {
|
|
"apps": apps,
|
|
"cache": max(0, cache_value),
|
|
"buffers": buffers,
|
|
"free": free,
|
|
}
|
|
comp_total = sum(composition.values())
|
|
if comp_total > total and comp_total > 0:
|
|
scale = total / comp_total
|
|
composition = {k: int(v * scale) for k, v in composition.items()}
|
|
|
|
return {
|
|
"total": total,
|
|
"used": used,
|
|
"free": free,
|
|
"available": available,
|
|
"buffers": buffers,
|
|
"cached": cached,
|
|
"apps": apps,
|
|
"percent": _round(percent, 1),
|
|
"composition": composition,
|
|
"swap_total": swap_total,
|
|
"swap_used": swap_used,
|
|
"swap_free": swap_free,
|
|
"swap_percent": _round(swap_percent, 1),
|
|
"dirty": info.get("dirty", 0),
|
|
"writeback": info.get("writeback", 0),
|
|
}
|
|
|
|
def _read_load(self, cores: int) -> Dict[str, Any]:
|
|
load = parse_loadavg(_read_text(self.proc_root / "loadavg"))
|
|
load["percent"] = _round(_pct(load["load1"], max(1, cores)), 1)
|
|
return load
|
|
|
|
# ------------------------------------------------------------------
|
|
# Disks
|
|
# ------------------------------------------------------------------
|
|
def _stat_disk(self, mount: Dict[str, str]) -> Optional[Dict[str, Any]]:
|
|
mountpoint = mount["mount"]
|
|
try:
|
|
stat = os.statvfs(mountpoint)
|
|
except OSError:
|
|
return None
|
|
frsize = stat.f_frsize or stat.f_bsize
|
|
total = frsize * stat.f_blocks
|
|
free = frsize * stat.f_bavail
|
|
used = total - frsize * stat.f_bfree
|
|
if total <= 0:
|
|
return None
|
|
if total < 1024 * 1024:
|
|
return None
|
|
percent = _pct(used, used + free)
|
|
inodes_total = stat.f_files
|
|
inodes_free = stat.f_ffree
|
|
return {
|
|
"device": mount["device"],
|
|
"mount": mountpoint,
|
|
"fstype": mount["fstype"],
|
|
"total": int(total),
|
|
"used": int(max(0, used)),
|
|
"free": int(max(0, free)),
|
|
"percent": _round(percent, 1),
|
|
"readonly": "ro" in mount["options"].split(","),
|
|
"inodes_total": int(inodes_total),
|
|
"inodes_free": int(inodes_free),
|
|
}
|
|
|
|
def _read_disks(self, now: float) -> List[Dict[str, Any]]:
|
|
if now - self._disk_cache_ts < 5.0 and self._disk_cache_ts:
|
|
return self._disk_cache
|
|
|
|
mounts = parse_mounts(_read_text(self.proc_root / "mounts"))
|
|
disks: List[Dict[str, Any]] = []
|
|
seen: set = set()
|
|
for mount in mounts:
|
|
fstype = mount["fstype"].lower()
|
|
device = mount["device"]
|
|
is_real = (
|
|
fstype in REAL_FS_TYPES
|
|
or device.startswith("/dev/")
|
|
or (fstype.startswith("fuse.") and fstype not in ("fuse.gvfsd-fuse", "fuse.lxcfs"))
|
|
)
|
|
if not is_real:
|
|
continue
|
|
if mount["mount"] in ("/dev", "/proc", "/sys", "/run"):
|
|
continue
|
|
disk = self._stat_disk(mount)
|
|
if not disk:
|
|
continue
|
|
key = (disk["device"], disk["total"], disk["used"])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
disks.append(disk)
|
|
|
|
disks.sort(key=lambda d: (0 if d["mount"] == "/" else 1, -d["percent"]))
|
|
self._disk_cache = disks
|
|
self._disk_cache_ts = now
|
|
return disks
|
|
|
|
def _is_partition(self, name: str) -> bool:
|
|
try:
|
|
if (self.sys_root / "class" / "block" / name / "partition").exists():
|
|
return True
|
|
except OSError:
|
|
pass
|
|
patterns = (
|
|
r"^(sd|hd|vd|xvd)[a-z]+\d+$",
|
|
r"^nvme\d+n\d+p\d+$",
|
|
r"^mmcblk\d+p\d+$",
|
|
r"^md\d+p\d+$",
|
|
)
|
|
return any(re.match(pattern, name) for pattern in patterns)
|
|
|
|
def _read_disk_io(self, now: float) -> Dict[str, Any]:
|
|
stats = parse_diskstats(_read_text(self.proc_root / "diskstats"))
|
|
dt = (now - self._prev_disk_ts) if self._prev_disk_ts is not None else 0.0
|
|
devices: List[Dict[str, Any]] = []
|
|
new_prev: Dict[str, Dict[str, int]] = {}
|
|
|
|
for name, current in stats.items():
|
|
if name.startswith(("loop", "ram", "fd", "sr", "zram")):
|
|
continue
|
|
if self._is_partition(name):
|
|
continue
|
|
new_prev[name] = current
|
|
old = self._prev_disk.get(name)
|
|
if old and dt > 0:
|
|
read_sectors = max(0, current["sectors_read"] - old["sectors_read"])
|
|
write_sectors = max(0, current["sectors_written"] - old["sectors_written"])
|
|
reads = max(0, current["reads"] - old["reads"])
|
|
writes = max(0, current["writes"] - old["writes"])
|
|
read_ms = max(0, current["read_ms"] - old["read_ms"])
|
|
write_ms = max(0, current["write_ms"] - old["write_ms"])
|
|
io_ms = max(0, current["io_ms"] - old["io_ms"])
|
|
devices.append({
|
|
"device": name,
|
|
"read_bps": read_sectors * 512.0 / dt,
|
|
"write_bps": write_sectors * 512.0 / dt,
|
|
"read_iops": reads / dt,
|
|
"write_iops": writes / dt,
|
|
"util_percent": min(100.0, io_ms / (dt * 1000.0) * 100.0),
|
|
"read_await_ms": read_ms / reads if reads else 0.0,
|
|
"write_await_ms": write_ms / writes if writes else 0.0,
|
|
})
|
|
elif current["sectors_read"] or current["sectors_written"]:
|
|
devices.append({
|
|
"device": name,
|
|
"read_bps": 0.0,
|
|
"write_bps": 0.0,
|
|
"read_iops": 0.0,
|
|
"write_iops": 0.0,
|
|
"util_percent": 0.0,
|
|
"read_await_ms": 0.0,
|
|
"write_await_ms": 0.0,
|
|
})
|
|
|
|
self._prev_disk = new_prev
|
|
self._prev_disk_ts = now
|
|
devices.sort(key=lambda d: d["read_bps"] + d["write_bps"], reverse=True)
|
|
devices = devices[:8]
|
|
for item in devices:
|
|
item["read_bps"] = round(item["read_bps"])
|
|
item["write_bps"] = round(item["write_bps"])
|
|
item["read_iops"] = round(item["read_iops"], 1)
|
|
item["write_iops"] = round(item["write_iops"], 1)
|
|
item["util_percent"] = round(item["util_percent"], 1)
|
|
item["read_await_ms"] = round(item["read_await_ms"], 2)
|
|
item["write_await_ms"] = round(item["write_await_ms"], 2)
|
|
return {
|
|
"devices": devices,
|
|
"read_bps": sum(d["read_bps"] for d in devices),
|
|
"write_bps": sum(d["write_bps"] for d in devices),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Network
|
|
# ------------------------------------------------------------------
|
|
def _iface_speed(self, name: str) -> Optional[int]:
|
|
try:
|
|
raw = _read_text(self.sys_root / "class" / "net" / name / "speed").strip()
|
|
value = int(raw)
|
|
return value if value > 0 else None
|
|
except (ValueError, OSError):
|
|
return None
|
|
|
|
def _read_network(self, now: float) -> Dict[str, Any]:
|
|
stats = parse_net_dev(_read_text(self.proc_root / "net" / "dev"))
|
|
addresses = self._addresses(now)
|
|
dt = (now - self._prev_net_ts) if self._prev_net_ts is not None else 0.0
|
|
interfaces: List[Dict[str, Any]] = []
|
|
default_iface = self._default_interface()
|
|
|
|
for name, current in stats.items():
|
|
if name == "lo":
|
|
continue
|
|
old = self._prev_net.get(name)
|
|
rx_bps = (current["rx_bytes"] - old["rx_bytes"]) / dt if old and dt > 0 else 0.0
|
|
tx_bps = (current["tx_bytes"] - old["tx_bytes"]) / dt if old and dt > 0 else 0.0
|
|
rx_bps = max(0.0, rx_bps)
|
|
tx_bps = max(0.0, tx_bps)
|
|
is_virtual = name.startswith(VIRTUAL_IFACE_PREFIXES)
|
|
ips = addresses["interfaces"].get(name, [])
|
|
if not ips and name == default_iface and addresses.get("primary"):
|
|
ips = [addresses["primary"]]
|
|
interfaces.append({
|
|
"name": name,
|
|
"ip": ips[0] if ips else "",
|
|
"virtual": is_virtual,
|
|
"primary": name == default_iface,
|
|
"rx_bps": rx_bps,
|
|
"tx_bps": tx_bps,
|
|
"rx_total": current["rx_bytes"],
|
|
"tx_total": current["tx_bytes"],
|
|
"errors": current["rx_errors"] + current["tx_errors"],
|
|
"drops": current["rx_drops"] + current["tx_drops"],
|
|
"speed_mbps": self._iface_speed(name),
|
|
"link": bool(ips) or rx_bps > 0 or tx_bps > 0,
|
|
})
|
|
|
|
physical = [i for i in interfaces if not i["virtual"]]
|
|
aggregate_source = physical or interfaces
|
|
total_rx = sum(i["rx_bps"] for i in aggregate_source)
|
|
total_tx = sum(i["tx_bps"] for i in aggregate_source)
|
|
|
|
for item in interfaces:
|
|
item["rx_bps"] = round(item["rx_bps"])
|
|
item["tx_bps"] = round(item["tx_bps"])
|
|
interfaces.sort(key=lambda i: (not i["primary"], i["virtual"], -(i["rx_bps"] + i["tx_bps"])))
|
|
self._prev_net = stats
|
|
self._prev_net_ts = now
|
|
|
|
return {
|
|
"interfaces": interfaces[:8],
|
|
"interface_count": len(interfaces),
|
|
"total_rx_bps": round(total_rx),
|
|
"total_tx_bps": round(total_tx),
|
|
"primary_iface": default_iface,
|
|
"primary_ip": addresses.get("primary"),
|
|
"all_ips": addresses.get("all", []),
|
|
"errors": sum(i["errors"] for i in interfaces),
|
|
"drops": sum(i["drops"] for i in interfaces),
|
|
}
|
|
|
|
# ------------------------------------------------------------------
|
|
# Processes
|
|
# ------------------------------------------------------------------
|
|
def _read_processes(self, now: float, mem_total: int) -> List[Dict[str, Any]]:
|
|
if self._proc_cache_ts and now - self._proc_cache_ts < 2.0:
|
|
return self._proc_cache
|
|
|
|
pids: List[int] = []
|
|
try:
|
|
with os.scandir(self.proc_root) as entries:
|
|
for entry in entries:
|
|
if entry.is_dir(follow_symlinks=False) and entry.name.isdigit():
|
|
pids.append(int(entry.name))
|
|
except OSError:
|
|
return self._proc_cache
|
|
|
|
total_jiffies = self._cpu_total_now
|
|
delta_total = max(0, total_jiffies - self._prev_proc_total)
|
|
previous = self._proc_ticks
|
|
new_ticks: Dict[int, int] = {}
|
|
candidates: List[Dict[str, Any]] = []
|
|
|
|
for pid in pids:
|
|
text = _read_text(self.proc_root / str(pid) / "stat")
|
|
match = _STAT_RE.match(text)
|
|
if not match:
|
|
continue
|
|
try:
|
|
rest = match.group(4).split()
|
|
utime = int(rest[10])
|
|
stime = int(rest[11])
|
|
rss_pages = int(rest[20])
|
|
threads = int(rest[16])
|
|
except (IndexError, ValueError):
|
|
continue
|
|
ticks = utime + stime
|
|
new_ticks[pid] = ticks
|
|
old_ticks = previous.get(pid)
|
|
cpu_percent = 0.0
|
|
if old_ticks is not None and delta_total > 0:
|
|
cpu_percent = max(0.0, (ticks - old_ticks) / delta_total * self._ncpu * 100.0)
|
|
rss = max(0, rss_pages) * self._page_size
|
|
candidates.append({
|
|
"pid": pid,
|
|
"name": match.group(2).strip(),
|
|
"cmd": "",
|
|
"cpu_percent": cpu_percent,
|
|
"mem_percent": _pct(rss, mem_total),
|
|
"rss_bytes": rss,
|
|
"threads": threads,
|
|
})
|
|
|
|
self._proc_ticks = new_ticks
|
|
self._prev_proc_total = total_jiffies
|
|
|
|
by_cpu = sorted(candidates, key=lambda p: p["cpu_percent"], reverse=True)[:8]
|
|
by_mem = sorted(candidates, key=lambda p: p["rss_bytes"], reverse=True)[:8]
|
|
selected: List[Dict[str, Any]] = []
|
|
seen = set()
|
|
for item in by_cpu + by_mem:
|
|
if item["pid"] in seen:
|
|
continue
|
|
seen.add(item["pid"])
|
|
selected.append(item)
|
|
if len(selected) >= 12:
|
|
break
|
|
|
|
for item in selected:
|
|
raw = _read_text(self.proc_root / str(item["pid"]) / "cmdline")
|
|
cmd = " ".join(part for part in raw.split("\x00") if part).strip()
|
|
if not cmd:
|
|
cmd = item["name"]
|
|
item["cmd"] = cmd[:90]
|
|
item["cpu_percent"] = round(item["cpu_percent"], 1)
|
|
item["mem_percent"] = round(item["mem_percent"], 2)
|
|
|
|
selected.sort(key=lambda p: p["cpu_percent"], reverse=True)
|
|
self._proc_cache = selected
|
|
self._proc_cache_ts = now
|
|
return selected
|
|
|
|
# ------------------------------------------------------------------
|
|
# Snapshot
|
|
# ------------------------------------------------------------------
|
|
def snapshot(self) -> Dict[str, Any]:
|
|
now = self.clock()
|
|
wall = self.wall_clock()
|
|
cpu = self._read_cpu(now)
|
|
memory = self._read_memory()
|
|
load = self._read_load(cpu["cores"] or self._static["cores"])
|
|
uptime = parse_uptime(_read_text(self.proc_root / "uptime"))
|
|
disks = self._read_disks(now)
|
|
disk_io = self._read_disk_io(now)
|
|
network = self._read_network(now)
|
|
addresses = self._addresses(now)
|
|
processes = self._read_processes(now, memory["total"])
|
|
|
|
alerts = build_alerts(
|
|
cpu_percent=float(cpu["percent"]),
|
|
memory_percent=float(memory["percent"]),
|
|
swap_percent=float(memory["swap_percent"]),
|
|
load1=float(load["load1"]),
|
|
cores=int(cpu["cores"] or self._static["cores"]),
|
|
disks=disks,
|
|
temp_c=cpu["temp_c"],
|
|
)
|
|
|
|
host = dict(self._static)
|
|
local_now = dt.datetime.now().astimezone()
|
|
local_offset = local_now.utcoffset()
|
|
host.update({
|
|
"tz_offset": int(local_offset.total_seconds()) if local_offset else host["tz_offset"],
|
|
"tz_name": local_now.tzname() or host["tz_name"],
|
|
"boot_epoch": wall - uptime,
|
|
"uptime_seconds": round(uptime, 1),
|
|
"primary_ip": addresses.get("primary"),
|
|
"ips": addresses.get("all", []),
|
|
"process_count": len(self._proc_ticks),
|
|
"thread_count": load.get("threads", 0),
|
|
})
|
|
|
|
return {
|
|
"ts": wall,
|
|
"version": __version__,
|
|
"host": host,
|
|
"cpu": cpu,
|
|
"memory": memory,
|
|
"load": load,
|
|
"disks": disks,
|
|
"disk_io": disk_io,
|
|
"network": network,
|
|
"processes": processes,
|
|
"alerts": alerts,
|
|
} |