#!/usr/bin/env python3 """hard-monitor — агент GHard Monitor (Gnexus Hardware Monitor). Stateless-сборщик: собирает факты о сервере и POST-ит их на hard-panel (POST /api/v1/ingest, заголовок X-Server-Key). Вся логика — пороги, ивенты, расчёт скорости сети — на панели. Конфиг: переменные окружения или .env рядом со скриптом PANEL_URL — адрес панели, напр. https://panel.example.com SERVER_KEY — ключ сервера из панели INTERVAL — период отправки, сек (по умолчанию 30) Использование: hard_monitor.py — бесконечный цикл (штатный режим, systemd) hard_monitor.py --once — один пакет и выход (для теста) hard_monitor.py --dry-run — собрать и напечатать пакет, не отправляя """ from __future__ import annotations import argparse import json import logging import os import platform import shutil import subprocess import sys import time from datetime import datetime, timezone from pathlib import Path import psutil import requests log = logging.getLogger("hard-monitor") # Файловые системы/маунты, которые не показываем как «диски» панели SKIP_FSTYPES = {"squashfs", "tmpfs", "devtmpfs", "overlay", "iso9660", "proc", "sysfs"} # --- Конфиг ----------------------------------------------------------------- def load_env_file(path: Path) -> None: """Мини-парсер .env: KEY=VALUE в переменные окружения (env важнее файла).""" if not path.exists(): return for line in path.read_text(encoding="utf-8").splitlines(): line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key, value = key.strip(), value.strip().strip("'\"") if key and key not in os.environ: os.environ[key] = value def env(name: str, default: str = "") -> str: return os.environ.get(name, default) # --- Сбор фактов ------------------------------------------------------------- def collect_identity() -> dict: os_name = platform.platform() pretty = Path("/etc/os-release") if pretty.exists(): for line in pretty.read_text(encoding="utf-8").splitlines(): if line.startswith("PRETTY_NAME="): os_name = line.partition("=")[2].strip('"') break return { "hostname": platform.node() or "unknown", "os": os_name, "kernel": platform.release(), "ips": collect_ips(), } def collect_ips() -> list[str]: """IPv4-адреса активных интерфейсов. Виртуальные (docker/veth/мосты) не показываем — в карточке сервера нужен реальный адрес, а не 172.17.0.1 каждого docker-моста. """ virtual = ("lo", "docker", "veth", "br-", "virbr") stats = psutil.net_if_stats() result = [] for iface, addrs in psutil.net_if_addrs().items(): if iface.startswith(virtual): continue if not stats.get(iface) or not stats[iface].isup: continue for addr in addrs: if addr.family.name == "AF_INET": result.append(addr.address) return sorted(set(result)) def collect_cpu() -> dict: # psutil считает % от предыдущего вызова — при старте сеем первое чтение load1, load5, load15 = os.getloadavg() return { "percent": psutil.cpu_percent(interval=None), "count": psutil.cpu_count() or 0, "load": [round(load1, 2), round(load5, 2), round(load15, 2)], } def collect_memory() -> dict: ram = psutil.virtual_memory() swap = psutil.swap_memory() return { "ram": {"total": ram.total, "used": ram.used, "available": ram.available}, "swap": {"total": swap.total, "used": swap.used}, } def collect_disks() -> list[dict]: disks = [] seen = set() for part in psutil.disk_partitions(all=False): if part.fstype in SKIP_FSTYPES or "/snap/" in part.mountpoint or part.mountpoint in seen: continue seen.add(part.mountpoint) try: usage = psutil.disk_usage(part.mountpoint) except (PermissionError, OSError): continue disks.append( { "mount": part.mountpoint, "total": usage.total, "used": usage.used, "percent": round(usage.percent, 1), } ) return disks def collect_net() -> list[dict]: counters = psutil.net_io_counters(pernic=True) return [ { "iface": iface, "bytes_sent": counters[iface].bytes_sent, "bytes_recv": counters[iface].bytes_recv, } for iface in sorted(counters) ] # loopback панель отфильтрует сама def collect_processes(limit: int = 5) -> list[dict]: """Топ процессов по CPU (с прошлого цикла).""" procs = [] for proc in psutil.process_iter(["pid", "name", "cpu_percent", "memory_percent"]): try: info = proc.info procs.append( { "name": info["name"] or "?", "pid": info["pid"] or 0, "cpu": info["cpu_percent"] or 0.0, "mem": info["memory_percent"] or 0.0, } ) except (psutil.NoSuchProcess, psutil.AccessDenied): continue procs.sort(key=lambda p: p["cpu"], reverse=True) return procs[:limit] def collect_docker() -> list[dict]: """Все контейнеры (включая упавшие — по ним панель сделает ивенты).""" if not shutil.which("docker"): return [] try: out = subprocess.run( ["docker", "ps", "-a", "--format", "{{.Names}}|{{.Image}}|{{.Status}}"], capture_output=True, text=True, timeout=10, ) except (OSError, subprocess.TimeoutExpired): return [] if out.returncode != 0: # нет доступа к docker-сокету и т.п. return [] containers = [] for line in out.stdout.strip().splitlines(): name, _, rest = line.partition("|") image, _, status = rest.partition("|") containers.append({"name": name, "image": image, "status": status}) return containers def collect_extra() -> dict: """Необязательные факты: температуры. Панель хранит их в extra.""" extra: dict = {} try: temps = psutil.sensors_temperatures() if temps: extra["temps"] = { label: round(entry[0].current or 0, 1) for label, entries in temps.items() if entries } except (OSError, AttributeError, Exception): pass return extra def collect() -> dict: return { **collect_identity(), "interval": int(env("INTERVAL", "30")), "ts": datetime.now(timezone.utc).isoformat(), "cpu": collect_cpu(), "memory": collect_memory(), "disks": collect_disks(), "uptime": int(time.time() - psutil.boot_time()), "net": collect_net(), "processes": collect_processes(), "docker": collect_docker(), "extra": collect_extra(), } # --- Отправка ---------------------------------------------------------------- def send(panel_url: str, server_key: str, payload: dict) -> None: response = requests.post( panel_url.rstrip("/") + "/api/v1/ingest", json=payload, headers={"X-Server-Key": server_key}, timeout=15, ) response.raise_for_status() def main() -> int: parser = argparse.ArgumentParser(description="hard-monitor agent") parser.add_argument("--once", action="store_true", help="отправить один пакет и выйти") parser.add_argument("--dry-run", action="store_true", help="напечатать пакет, не отправляя") parser.add_argument("--env", default=".env", help="путь к .env (по умолчанию рядом со скриптом)") args = parser.parse_args() load_env_file(Path(__file__).resolve().parent / args.env) panel_url = env("PANEL_URL") server_key = env("SERVER_KEY") interval = int(env("INTERVAL", "30")) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout, ) if not args.dry_run: if not panel_url or not server_key: log.error("PANEL_URL и SERVER_KEY обязательны (env или .env)") return 1 # сеем базовые точки, чтобы cpu_percent и process cpu были не нулевые psutil.cpu_percent(interval=None) list(psutil.process_iter(["cpu_percent"])) while True: started = time.monotonic() try: payload = collect() if args.dry_run: print(json.dumps(payload, ensure_ascii=False, indent=2)) return 0 send(panel_url, server_key, payload) log.info("sent: cpu=%.1f%% net_ifaces=%d", payload["cpu"]["percent"], len(payload["net"])) except requests.RequestException as exc: log.warning("send failed: %s", exc) except Exception: log.exception("collect failed") if args.once: return 0 time.sleep(max(interval - (time.monotonic() - started), 0)) if __name__ == "__main__": sys.exit(main())