fix(worker): discard impossible cgroup v2 cpu samples

livekit-agents/livekit/agents/utils/hw/cpu.py

@@ -39,6 +39,9 @@ def cpu_percent(self, interval: float = 0.5) -> float:
 
 
 class CGroupV2CPUMonitor(CPUMonitor):
+    def __init__(self) -> None:
+        self._last_cpu_percent = 0.0
+
     def cpu_count(self) -> float:
         # quota: The maximum CPU time in microseconds that the cgroup can use within a given period.
         # period: The period of time in microseconds over which the quota applies.
@@ -53,18 +56,28 @@ def cpu_count(self) -> float:
         return 1.0 * int(quota) / period
 
     def cpu_percent(self, interval: float = 0.5) -> float:
+        start = time.monotonic()
         cpu_usage_start = self._read_cpu_usage()
         time.sleep(interval)
         cpu_usage_end = self._read_cpu_usage()
-        cpu_usage_diff = cpu_usage_end - cpu_usage_start
+        elapsed = time.monotonic() - start
 
         # microseconds to seconds
-        cpu_usage_seconds = cpu_usage_diff / 1_000_000
-
-        num_cpus = self.cpu_count()
-        cpu_usage_percent = cpu_usage_seconds / (interval * num_cpus)
-
-        return min(cpu_usage_percent, 1)
+        cpu_usage_seconds = (cpu_usage_end - cpu_usage_start) / 1_000_000
+
+        # some hypervisors serve a torn per-cpu sum, so discard a delta the host cannot have produced
+        max_cpu_usage_seconds = elapsed * (psutil.cpu_count() or 1)
+        if not 0 <= cpu_usage_seconds <= max_cpu_usage_seconds:
+            logger.warning(
+                "discarding impossible cgroup cpu usage delta of %.3fs (ceiling %.3fs)",
+                cpu_usage_seconds,
+                max_cpu_usage_seconds,
+            )
+            return self._last_cpu_percent
+
+        cpu_usage_percent = cpu_usage_seconds / (elapsed * self.cpu_count())
+        self._last_cpu_percent = min(cpu_usage_percent, 1.0)
+        return self._last_cpu_percent
 
     def _read_cpu_max(self) -> tuple[str, int]:
         try:

tests/test_cpu_monitor.py

@@ -0,0 +1,121 @@
+import logging
+import time
+
+import psutil
+import pytest
+
+from livekit.agents import utils
+from livekit.agents.utils.hw.cpu import CGroupV2CPUMonitor
+
+pytestmark = pytest.mark.unit
+
+HOST_CPUS = 8
+INTERVAL = 0.5
+# one idle 0.5 s sample on the reporter's host, about 0.5 % of 8 CPUs
+IDLE_USAGE_USEC = 21_600
+
+# raw (start, end) usage_usec pairs logged on the affected Xen HVM guest
+TORN_READS = [
+    (5260817467, 2548880500),
+    (5268770613, 5860175581),
+    (5889766280, 5270782758),
+    (1210759425, 5271033390),
+    (5271381786, 5798443673),
+    (5288672329, 1215868554),
+]
+
+
+@pytest.fixture
+def monitor(monkeypatch: pytest.MonkeyPatch) -> CGroupV2CPUMonitor:
+    monkeypatch.delenv("NUM_CPUS", raising=False)
+    monkeypatch.setattr(psutil, "cpu_count", lambda: HOST_CPUS)
+    monkeypatch.setattr(time, "sleep", lambda _: None)
+    monitor = CGroupV2CPUMonitor()
+    monkeypatch.setattr(monitor, "_read_cpu_max", lambda: ("max", 100000))
+    return monitor
+
+
+def sample(
+    monitor: CGroupV2CPUMonitor,
+    monkeypatch: pytest.MonkeyPatch,
+    usage_start: int,
+    usage_end: int,
+    *,
+    elapsed: float = INTERVAL,
+) -> float:
+    reads = iter([usage_start, usage_end])
+    clock = iter([0.0, elapsed])
+    monkeypatch.setattr(monitor, "_read_cpu_usage", lambda: next(reads))
+    monkeypatch.setattr(time, "monotonic", lambda: next(clock))
+    return monitor.cpu_percent(INTERVAL)
+
+
+@pytest.mark.parametrize("elapsed", [0.5, 0.6])
+def test_normal_delta_uses_measured_elapsed(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch, elapsed: float
+) -> None:
+    # one cpu-second of usage over the measured interval on 8 CPUs
+    pct = sample(monitor, monkeypatch, 5_000_000_000, 5_001_000_000, elapsed=elapsed)
+    assert pct == pytest.approx(1.0 / (elapsed * HOST_CPUS))
+
+
+def test_negative_delta_holds_previous_sample(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    good = sample(monitor, monkeypatch, 5_000_000_000, 5_001_000_000)
+    with caplog.at_level(logging.WARNING, logger="livekit.agents"):
+        pct = sample(monitor, monkeypatch, *TORN_READS[0])
+    assert pct == good
+    assert "impossible" in caplog.text
+
+
+def test_over_ceiling_delta_holds_previous_sample(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    good = sample(monitor, monkeypatch, 5_000_000_000, 5_001_000_000)
+    with caplog.at_level(logging.WARNING, logger="livekit.agents"):
+        pct = sample(monitor, monkeypatch, *TORN_READS[1])
+    assert pct == good
+    assert pct != 1.0
+    assert "impossible" in caplog.text
+
+
+def test_first_discarded_sample_reads_idle(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    assert sample(monitor, monkeypatch, *TORN_READS[0]) == 0.0
+
+
+def test_reporter_pattern_stays_below_threshold(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    idle = 5_270_000_000
+    reads: list[tuple[int, int]] = []
+    for torn in [None, None, *TORN_READS[:3], None, *TORN_READS[3:], None, None]:
+        reads.append(torn or (idle, idle + IDLE_USAGE_USEC))
+        idle += IDLE_USAGE_USEC
+
+    avg = utils.MovingAverage(5)
+    idle_pct = IDLE_USAGE_USEC / 1_000_000 / (INTERVAL * HOST_CPUS)
+    for usage_start, usage_end in reads:
+        avg.add_sample(sample(monitor, monkeypatch, usage_start, usage_end))
+        assert avg.get_avg() == pytest.approx(idle_pct)
+
+
+def test_delta_at_ceiling_is_full_load(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    full = int(INTERVAL * HOST_CPUS * 1_000_000)
+    assert sample(monitor, monkeypatch, 5_000_000_000, 5_000_000_000 + full) == 1.0
+
+
+def test_burst_above_quota_clamps_instead_of_discarding(
+    monitor: CGroupV2CPUMonitor, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    # a 2-CPU cpu.max quota on an 8-CPU host, bursting to 4 CPUs for the interval
+    monkeypatch.setattr(monitor, "_read_cpu_max", lambda: ("200000", 100000))
+    assert monitor.cpu_count() == 2.0
+    with caplog.at_level(logging.WARNING, logger="livekit.agents"):
+        pct = sample(monitor, monkeypatch, 5_000_000_000, 5_002_000_000)
+    assert pct == 1.0
+    assert "impossible" not in caplog.text