app/system_monitor.py

"""Supervision matérielle et réseau du conteneur, sans dépendance.

Tout est lu dans `/proc` et `/sys` — ce que le noyau expose déjà. Ajouter
`psutil` aurait été plus court mais impose une roue compilée à installer au
build, pour des informations qui tiennent en quelques fichiers texte.

Deux précautions valables partout, et particulièrement sur un Space Hugging
Face :

- **Les limites cgroup priment.** Un conteneur voit la RAM et les cœurs de la
  machine hôte, pas son quota. `/sys/fs/cgroup/memory.max` donne la vraie
  limite ; l'ignorer afficherait « 3 Go utilisés sur 500 Go » et masquerait la
  saturation qui approche.
- **Les compteurs sont cumulatifs.** CPU et réseau ne se lisent pas en valeur
  absolue : on garde l'échantillon précédent et on calcule le débit entre deux
  appels. Le premier appel renvoie donc des débits nuls.

Sur un système sans `/proc` (macOS en développement), chaque sonde renvoie
`None` plutôt que d'échouer : le tableau de bord affiche « indisponible ».
"""

from __future__ import annotations

import os
import time
from pathlib import Path

PROC = Path("/proc")
CGROUP = Path("/sys/fs/cgroup")

# Échantillon précédent, pour transformer les compteurs cumulatifs en débits.
_previous: dict[str, tuple[float, float]] = {}


def _read(path: Path) -> str | None:
    try:
        return path.read_text().strip()
    except (OSError, ValueError):
        return None


def _read_int(path: Path) -> int | None:
    raw = _read(path)
    if raw is None:
        return None
    try:
        return int(raw.split()[0])
    except (ValueError, IndexError):
        return None


def _rate(key: str, total: float, now: float) -> float | None:
    """Débit par seconde entre cet appel et le précédent."""
    previous = _previous.get(key)
    _previous[key] = (total, now)
    if previous is None:
        return None
    last_total, last_time = previous
    elapsed = now - last_time
    if elapsed <= 0 or total < last_total:  # compteur remis à zéro
        return None
    return (total - last_total) / elapsed


# --- Processeur -----------------------------------------------------------


def _cpu_quota() -> float | None:
    """Nombre de cœurs réellement alloués au conteneur (cgroup v2 puis v1)."""
    raw = _read(CGROUP / "cpu.max")
    if raw:
        parts = raw.split()
        if len(parts) == 2 and parts[0] != "max":
            try:
                return int(parts[0]) / int(parts[1])
            except (ValueError, ZeroDivisionError):
                pass
    quota = _read_int(CGROUP / "cpu" / "cpu.cfs_quota_us")
    period = _read_int(CGROUP / "cpu" / "cpu.cfs_period_us")
    if quota and period and quota > 0:
        return quota / period
    return None


def _cgroup_cpu_usec() -> int | None:
    """Temps processeur consommé par le conteneur, en microsecondes."""
    raw = _read(CGROUP / "cpu.stat")
    if raw:
        for line in raw.splitlines():
            if line.startswith("usage_usec"):
                try:
                    return int(line.split()[1])
                except (ValueError, IndexError):
                    return None
    # cgroup v1 : nanosecondes.
    nanos = _read_int(CGROUP / "cpuacct" / "cpuacct.usage")
    return nanos // 1000 if nanos is not None else None


def cpu() -> dict:
    now = time.monotonic()
    cores = os.cpu_count()
    quota = _cpu_quota()
    percent = None
    scope = "conteneur"

    # `/proc/stat` n'est pas isolé : dans un conteneur il compte tout ce que
    # fait la machine hôte, voisins compris. Il afficherait 99 % pendant que
    # ce service dort. Le compteur cgroup, lui, ne mesure que nous.
    usec = _cgroup_cpu_usec()
    if usec is not None:
        used_rate = _rate("cpu_cgroup", usec, now)  # µs de CPU par seconde
        if used_rate is not None:
            budget = (quota or cores or 1) * 1_000_000
            percent = max(0.0, min(100.0, used_rate / budget * 100))
    else:
        # Sans cgroup (poste de développement), on retombe sur la mesure
        # globale, en le signalant : elle décrit la machine, pas le service.
        scope = "machine"
        stat = _read(PROC / "stat")
        if stat:
            for line in stat.splitlines():
                if line.startswith("cpu "):
                    values = [int(v) for v in line.split()[1:]]
                    total = sum(values)
                    idle = values[3] + (values[4] if len(values) > 4 else 0)
                    busy_rate = _rate("cpu_busy", total - idle, now)
                    total_rate = _rate("cpu_total", total, now)
                    if busy_rate is not None and total_rate:
                        percent = max(0.0, min(100.0, busy_rate / total_rate * 100))
                    break

    load = None
    loadavg = _read(PROC / "loadavg")
    if loadavg:
        try:
            load = [float(v) for v in loadavg.split()[:3]]
        except ValueError:
            load = None

    return {
        "percent": round(percent, 1) if percent is not None else None,
        "scope": scope,
        "load": load,
        "cores": cores,
        "allocated_cores": round(quota, 2) if quota else None,
        # `loadavg` est celle de la machine hôte : dans un conteneur, elle
        # agrège tout ce qui tourne sur le serveur, y compris les voisins.
        # Elle est donnée à titre indicatif et n'est pas rapportée aux cœurs
        # alloués — le ratio ferait passer pour une saturation de ce service
        # une charge qui ne lui appartient pas.
        "load_is_host_wide": True,
    }


# --- Mémoire --------------------------------------------------------------


def memory() -> dict:
    limit = _read_int(CGROUP / "memory.max")
    used = _read_int(CGROUP / "memory.current")
    if limit is None:
        limit = _read_int(CGROUP / "memory" / "memory.limit_in_bytes")
        used = _read_int(CGROUP / "memory" / "memory.usage_in_bytes")

    total = available = None
    meminfo = _read(PROC / "meminfo")
    if meminfo:
        values = {}
        for line in meminfo.splitlines():
            parts = line.split(":")
            if len(parts) == 2:
                try:
                    values[parts[0]] = int(parts[1].split()[0]) * 1024
                except (ValueError, IndexError):
                    continue
        total = values.get("MemTotal")
        available = values.get("MemAvailable")

    # Une limite cgroup absurde (valeur sentinelle « pas de limite ») ne doit
    # pas passer pour une vraie limite.
    if limit is not None and total is not None and limit > total * 4:
        limit = None

    effective_total = limit or total
    effective_used = used if limit else (total - available if total and available else None)
    return {
        "total": effective_total,
        "used": effective_used,
        "percent": (
            round(effective_used / effective_total * 100, 1)
            if effective_total and effective_used is not None
            else None
        ),
        "limited_by_cgroup": limit is not None,
    }


# --- Disque ---------------------------------------------------------------


# Au-delà de ce seuil, le point de montage n'est pas un volume dédié mais le
# système de fichiers partagé de l'hôte : ses chiffres ne disent rien de la
# place dont dispose réellement ce service.
DEDICATED_VOLUME_MAX = 4 * 1024**4  # 4 To


def _tree_size(path: str) -> int | None:
    """Octets réellement occupés sous `path`. Le répertoire reste minuscule."""
    total = 0
    try:
        for root, _, files in os.walk(path):
            for name in files:
                try:
                    total += os.stat(os.path.join(root, name)).st_size
                except OSError:
                    continue
    except OSError:
        return None
    return total


def disk(path: str) -> dict:
    try:
        stats = os.statvfs(path)
    except OSError:
        return {"total": None, "used": None, "free": None, "percent": None, "path": path,
                "dedicated": False}
    total = stats.f_blocks * stats.f_frsize
    free = stats.f_bavail * stats.f_frsize
    used = total - stats.f_bfree * stats.f_frsize

    # Sur un Space sans stockage persistant attaché, /data vit sur le système
    # de fichiers de l'hôte : `statvfs` annonce des pétaoctets, ce qui est
    # exact et parfaitement inutile. On rapporte alors ce que ce service
    # occupe vraiment, sans prétendre connaître la place disponible.
    if total > DEDICATED_VOLUME_MAX:
        return {
            "total": None,
            "used": _tree_size(path),
            "free": None,
            "percent": None,
            "path": path,
            "dedicated": False,
        }
    return {
        "total": total,
        "used": used,
        "free": free,
        "percent": round(used / total * 100, 1) if total else None,
        "path": path,
        "dedicated": True,
    }


# --- Réseau ---------------------------------------------------------------


def network() -> dict:
    """Débits entrant et sortant, hors interface de bouclage."""
    raw = _read(PROC / "net" / "dev")
    if not raw:
        return {"rx_rate": None, "tx_rate": None, "rx_total": None, "tx_total": None}
    now = time.monotonic()
    rx_total = tx_total = 0
    for line in raw.splitlines()[2:]:
        name, _, rest = line.partition(":")
        if name.strip() == "lo":
            continue
        fields = rest.split()
        if len(fields) >= 9:
            try:
                rx_total += int(fields[0])
                tx_total += int(fields[8])
            except ValueError:
                continue
    rx_rate = _rate("net_rx", rx_total, now)
    tx_rate = _rate("net_tx", tx_total, now)
    return {
        "rx_rate": round(rx_rate) if rx_rate is not None else None,
        "tx_rate": round(tx_rate) if tx_rate is not None else None,
        "rx_total": rx_total,
        "tx_total": tx_total,
    }


# --- Process --------------------------------------------------------------


def process() -> dict:
    rss = None
    status = _read(PROC / "self" / "status")
    if status:
        for line in status.splitlines():
            if line.startswith("VmRSS:"):
                try:
                    rss = int(line.split()[1]) * 1024
                except (ValueError, IndexError):
                    rss = None
                break
    threads = None
    if status:
        for line in status.splitlines():
            if line.startswith("Threads:"):
                try:
                    threads = int(line.split()[1])
                except (ValueError, IndexError):
                    threads = None
                break
    try:
        descriptors = len(os.listdir("/proc/self/fd"))
    except OSError:
        descriptors = None

    uptime = None
    raw = _read(PROC / "uptime")
    if raw:
        try:
            uptime = float(raw.split()[0])
        except (ValueError, IndexError):
            uptime = None

    return {"rss": rss, "threads": threads, "open_files": descriptors, "host_uptime": uptime}


_started = time.time()


def snapshot(data_path: str = "/data") -> dict:
    """Instantané complet, destiné au tableau de bord temps réel."""
    return {
        "at": int(time.time() * 1000),
        "available": PROC.is_dir(),
        "app_uptime": round(time.time() - _started),
        "cpu": cpu(),
        "memory": memory(),
        "disk": disk(data_path if os.path.isdir(data_path) else "."),
        "network": network(),
        "process": process(),
    }