Wizard/wizard_server/app/load.py

"""Load monitoring: detects when the server runs out of compute.

Every ``load_sample_interval`` seconds the monitor measures

- **event loop lag**: how much later than scheduled the loop woke up. This is
  the most direct signal that the server is slow for players;
- **CPU usage** of the process relative to what it may use (the container's
  CPU quota, at most one core because Python runs game logic on one core);
- **CPU throttling** from cgroup v2 (``/sys/fs/cgroup/cpu.stat``): the share
  of scheduling periods in which the container hit its CPU limit.

Each sample is classified as normal / busy / overloaded, and ``next_level``
turns the recent classifications into the current level with hysteresis, so a
single slow second does not flip the state back and forth.
"""

from __future__ import annotations

import asyncio
import logging
import time
from collections import deque
from collections.abc import Callable, Sequence
from dataclasses import dataclass
from enum import Enum
from pathlib import Path

from .config import Settings

logger = logging.getLogger(__name__)

CGROUP_DIR = Path("/sys/fs/cgroup")


class LoadLevel(str, Enum):
    NORMAL = "normal"
    BUSY = "busy"
    OVERLOADED = "overloaded"


@dataclass(frozen=True, slots=True)
class LoadSample:
    loop_lag: float
    # Share of the available CPU used (None if unknown).
    cpu: float | None = None
    # Share of CPU periods in which the container was throttled (None if unknown).
    throttled: float | None = None


@dataclass(frozen=True, slots=True)
class LoadThresholds:
    lag_busy: float
    lag_overload: float
    lag_severe: float
    cpu_busy: float
    cpu_overload: float
    throttle_overload: float
    overload_samples: int
    recover_samples: int

    @classmethod
    def from_settings(cls, settings: Settings) -> LoadThresholds:
        return cls(
            lag_busy=settings.load_lag_busy,
            lag_overload=settings.load_lag_overload,
            lag_severe=settings.load_lag_severe,
            cpu_busy=settings.load_cpu_busy,
            cpu_overload=settings.load_cpu_overload,
            throttle_overload=settings.load_throttle_overload,
            overload_samples=max(1, settings.load_overload_samples),
            recover_samples=max(1, settings.load_recover_samples),
        )


def classify(sample: LoadSample, t: LoadThresholds) -> LoadLevel:
    """Level of a single sample."""
    if (
        sample.loop_lag >= t.lag_overload
        or (sample.cpu is not None and sample.cpu >= t.cpu_overload)
        or (sample.throttled is not None and sample.throttled >= t.throttle_overload)
    ):
        return LoadLevel.OVERLOADED
    if sample.loop_lag >= t.lag_busy or (sample.cpu is not None and sample.cpu >= t.cpu_busy):
        return LoadLevel.BUSY
    return LoadLevel.NORMAL


def next_level(
    current: LoadLevel, recent: Sequence[LoadLevel], t: LoadThresholds, severe: bool = False
) -> LoadLevel:
    """Current level from the recent sample levels (oldest first), with hysteresis.

    - Overloaded after ``overload_samples`` overloaded samples in a row, or at
      once when the latest sample was ``severe`` (extreme event loop lag).
    - Leaving overload (and busy) requires ``recover_samples`` samples in a row
      below that level.
    - Busy is entered immediately (it is only a hint for players).
    """
    if severe:
        return LoadLevel.OVERLOADED
    if not recent:
        return current
    last_bad = recent[-t.overload_samples:]
    if len(last_bad) == t.overload_samples and all(level is LoadLevel.OVERLOADED for level in last_bad):
        return LoadLevel.OVERLOADED

    window = recent[-t.recover_samples:]
    full_window = len(window) == t.recover_samples
    if current is LoadLevel.OVERLOADED and not (
        full_window and all(level is not LoadLevel.OVERLOADED for level in window)
    ):
        return LoadLevel.OVERLOADED
    if recent[-1] is not LoadLevel.NORMAL:
        return LoadLevel.BUSY
    if current is not LoadLevel.NORMAL and not (full_window and all(level is LoadLevel.NORMAL for level in window)):
        return LoadLevel.BUSY
    return LoadLevel.NORMAL


# --- cgroup v2 helpers ------------------------------------------------------


def parse_cpu_max(text: str) -> float | None:
    """CPU quota in cores from ``cpu.max`` ("max 100000" = unlimited)."""
    parts = text.split()
    if len(parts) != 2 or parts[0] == "max":
        return None
    try:
        return int(parts[0]) / int(parts[1])
    except (ValueError, ZeroDivisionError):
        return None


def parse_cpu_stat(text: str) -> dict[str, int]:
    stats: dict[str, int] = {}
    for line in text.splitlines():
        key, _, value = line.partition(" ")
        if value.strip().isdigit():
            stats[key] = int(value)
    return stats


def own_cgroup_dir(root: Path = CGROUP_DIR, proc_cgroup: Path = Path("/proc/self/cgroup")) -> Path:
    """The cgroup v2 directory of this process.

    In a container this is ``/sys/fs/cgroup`` itself (cgroup namespace); on a
    host it is a sub directory such as ``/sys/fs/cgroup/user.slice/…scope``.
    """
    for line in (_read(proc_cgroup) or "").splitlines():
        if line.startswith("0::"):
            candidate = root / line[3:].strip().lstrip("/")
            if candidate.is_dir():
                return candidate
    return root


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


class LoadMonitor:
    def __init__(
        self,
        settings: Settings,
        on_change: Callable[[LoadLevel], None] | None = None,
        cgroup_dir: Path | None = None,
    ) -> None:
        self._settings = settings
        self._thresholds = LoadThresholds.from_settings(settings)
        self._on_change = on_change
        self._cgroup_dir = cgroup_dir if cgroup_dir is not None else own_cgroup_dir()
        self._forced = LoadLevel(settings.load_force_level) if settings.load_force_level else None
        self._level = LoadLevel.NORMAL
        self._recent: deque[LoadLevel] = deque(
            maxlen=max(self._thresholds.overload_samples, self._thresholds.recover_samples)
        )
        self.last_sample = LoadSample(loop_lag=0.0)

        quota = parse_cpu_max(_read(self._cgroup_dir / "cpu.max") or "")
        # Game logic runs on one core, so more than one core cannot be used.
        self._cpu_capacity = min(quota, 1.0) if quota else 1.0
        self._last_cpu_time = time.process_time()
        self._last_wall = time.monotonic()
        self._last_cgroup = self._cgroup_stats()

    @property
    def level(self) -> LoadLevel:
        return self._forced or self._level

    def record(self, sample: LoadSample) -> None:
        """Add a sample and update the level (called by ``run``; public for tests)."""
        self.last_sample = sample
        self._recent.append(classify(sample, self._thresholds))
        severe = sample.loop_lag >= self._thresholds.lag_severe
        new_level = next_level(self._level, list(self._recent), self._thresholds, severe)
        if new_level is self._level:
            return
        logger.log(
            logging.WARNING if new_level is LoadLevel.OVERLOADED else logging.INFO,
            "Load level %s -> %s (lag %.3fs, cpu %s, throttled %s)",
            self._level.value, new_level.value, sample.loop_lag, sample.cpu, sample.throttled,
        )
        self._level = new_level
        if self._on_change is not None and self._forced is None:
            self._on_change(new_level)

    async def run(self) -> None:
        loop = asyncio.get_running_loop()
        interval = self._settings.load_sample_interval
        while True:
            started = loop.time()
            await asyncio.sleep(interval)
            lag = max(0.0, loop.time() - started - interval)
            try:
                self.record(self._measure(lag))
            except Exception:
                logger.exception("Load measurement failed")

    def stats(self) -> dict[str, object]:
        sample = self.last_sample
        return {
            "load": self.level.value,
            "loop_lag": round(sample.loop_lag, 4),
            "cpu": None if sample.cpu is None else round(sample.cpu, 3),
            "throttled": None if sample.throttled is None else round(sample.throttled, 3),
        }

    def _measure(self, lag: float) -> LoadSample:
        now_cpu, now_wall = time.process_time(), time.monotonic()
        wall = now_wall - self._last_wall
        cpu = (now_cpu - self._last_cpu_time) / wall / self._cpu_capacity if wall > 0 else None
        self._last_cpu_time, self._last_wall = now_cpu, now_wall

        throttled = None
        stats = self._cgroup_stats()
        if stats and self._last_cgroup:
            periods = stats.get("nr_periods", 0) - self._last_cgroup.get("nr_periods", 0)
            if periods > 0:
                throttled = (stats.get("nr_throttled", 0) - self._last_cgroup.get("nr_throttled", 0)) / periods
        self._last_cgroup = stats
        return LoadSample(loop_lag=lag, cpu=cpu, throttled=throttled)

    def _cgroup_stats(self) -> dict[str, int]:
        text = _read(self._cgroup_dir / "cpu.stat")
        return parse_cpu_stat(text) if text else {}