refactor(agent): Address review feedback on sysfs-first stats

- Extend changelog to call out the memory metric shift (sysfs
  excludes inactive file cache; dashboards may step-down).
- Bound the _cgroup_fallback_warned set to prevent unbounded growth
  on hosts with high container churn.
- Replace RuntimeError("should not reach here") with assert_never()
  to satisfy the BackendAIError-only rule and strengthen exhaustive
  match-statement type checking.
- Add parallel MemoryPlugin fallback + linuxkit tests and a
  dedup-warn test.

Refs #11220
Refs #11223

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: rapsealk

changes/11223.enhance.md

@@ -1 +1,3 @@
 Default to cgroup (sysfs) stat collection on native Linux hosts, falling back to the Docker API only on linuxkit or read failure.
+Note: reported container memory usage may step down on hosts previously using `stats-type: docker`, because sysfs excludes inactive file cache (matching `docker stats`).
+Dashboards or autoscaling thresholds tuned to the old (higher) values should be re-evaluated after upgrade.

src/ai/backend/agent/docker/intrinsic.py

@@ -8,12 +8,13 @@
 from collections.abc import Collection, Iterable, Mapping, Sequence
 from decimal import Decimal
 from pathlib import Path
-from typing import Any, cast
+from typing import Any, assert_never, cast
 
 import aiohttp
 import psutil
 from aiodocker.docker import Docker, DockerContainer
 from aiodocker.exceptions import DockerError
+from cachetools import LRUCache
 
 from ai.backend.agent import __version__  # pants: no-infer-dep
 from ai.backend.agent.alloc_map import AllocationStrategy
@@ -75,14 +76,20 @@
 
 # Tracks containers for which we have already logged a cgroup->Docker-API
 # fallback warning, to avoid log spam on persistent read failures.
-_cgroup_fallback_warned: set[str] = set()
+# Bounded LRU cache to prevent unbounded growth on hosts with high container churn;
+# oldest entries are evicted on overflow, so a recreated container may re-warn.
+# Shared across CPUPlugin / MemoryPlugin since keys are namespaced via the `plugin:` prefix.
+_CGROUP_FALLBACK_WARN_CACHE_SIZE = 1024
+_cgroup_fallback_warned: LRUCache[str, None] = LRUCache(
+    maxsize=_CGROUP_FALLBACK_WARN_CACHE_SIZE,
+)
 
 
 def _warn_cgroup_fallback_once(plugin: str, container_id: str) -> None:
     key = f"{plugin}:{container_id}"
     if key in _cgroup_fallback_warned:
         return
-    _cgroup_fallback_warned.add(key)
+    _cgroup_fallback_warned[key] = None
     log.warning(
         "{0}: cgroup sysfs read failed for container {1}; falling back to Docker API",
         plugin,
@@ -340,7 +347,7 @@ async def cgroup_first_impl(container_id: str) -> float | None:
             case StatModes.CGROUP | StatModes.DOCKER:
                 impl = api_impl
             case _:
-                raise RuntimeError("should not reach here")
+                assert_never(ctx.mode)
 
         tasks = []
         for cid in container_ids:
@@ -863,7 +870,7 @@ async def cgroup_first_impl(
             case StatModes.CGROUP | StatModes.DOCKER:
                 impl = api_impl
             case _:
-                raise RuntimeError("should not reach here")
+                assert_never(ctx.mode)
 
         per_container_mem_used_bytes = {}
         per_container_io_read_bytes = {}

tests/unit/agent/test_docker_intrinsic.py

@@ -10,10 +10,12 @@
 
 import pytest
 
+from ai.backend.agent.docker import intrinsic
 from ai.backend.agent.docker.intrinsic import (
     ContainerNetStat,
     CPUPlugin,
     MemoryPlugin,
+    _warn_cgroup_fallback_once,
     read_proc_net_dev,
 )
 from ai.backend.agent.stats import StatModes
@@ -277,6 +279,101 @@ async def test_sysfs_mode_uses_instance_docker_client(
             await memory_plugin.gather_container_measures(memory_cgroup_context, container_ids)
             mock_docker_cls.assert_not_called()
 
+    async def test_cgroup_mode_falls_back_to_api_on_sysfs_failure(
+        self,
+        memory_plugin: MemoryPlugin,
+        container_ids: list[str],
+        cgroup_stat_context: MagicMock,
+        mock_fetch_api_stats: MagicMock,
+    ) -> None:
+        """When sysfs read fails in CGROUP mode, the Docker API is used
+        as a per-read fallback instead of silently returning zero."""
+        # Arrange: cgroup version that triggers "return None" in sysfs_impl.
+        cgroup_stat_context.agent.get_cgroup_version = MagicMock(return_value="invalid")
+        cgroup_stat_context.agent.get_cgroup_path = MagicMock(return_value=MagicMock())
+
+        results = await memory_plugin.gather_container_measures(cgroup_stat_context, container_ids)
+
+        assert mock_fetch_api_stats.call_count == len(container_ids)
+        # api_impl returns mem_cur = 1024 * 1024 * 100 = 104857600 bytes
+        for cid in container_ids:
+            assert results[0].per_container[cid].value == 1024 * 1024 * 100
+
+    async def test_linuxkit_forces_api_even_in_cgroup_mode(
+        self,
+        container_ids: list[str],
+        cgroup_stat_context: MagicMock,
+        mock_fetch_api_stats: MagicMock,
+    ) -> None:
+        """On linuxkit hosts the API path is used even when mode is CGROUP."""
+        plugin = MemoryPlugin.__new__(MemoryPlugin)
+        plugin.local_config = {"agent": {"docker-mode": "linuxkit"}}
+        plugin._docker = AsyncMock()
+
+        await plugin.gather_container_measures(cgroup_stat_context, container_ids)
+        assert mock_fetch_api_stats.call_count == len(container_ids)
+
+
+class TestWarnCgroupFallbackOnce:
+    """Tests for _warn_cgroup_fallback_once() dedup and bounded-cache semantics."""
+
+    @pytest.fixture(autouse=True)
+    def _reset_warn_cache(self) -> Generator[None, None, None]:
+        """Reset the module-level warn cache between tests to avoid bleed-through."""
+        intrinsic._cgroup_fallback_warned.clear()
+        yield
+        intrinsic._cgroup_fallback_warned.clear()
+
+    def test_deduplicates_per_container(
+        self,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """The helper logs once per (plugin, container_id) and stays silent on
+        subsequent calls for the same container."""
+        with caplog.at_level("WARNING", logger="ai.backend.agent.docker.intrinsic"):
+            _warn_cgroup_fallback_once("CPUPlugin", "container_abc")
+            _warn_cgroup_fallback_once("CPUPlugin", "container_abc")
+            _warn_cgroup_fallback_once("CPUPlugin", "container_abc")
+
+        warn_records = [r for r in caplog.records if r.levelname == "WARNING"]
+        assert len(warn_records) == 1
+
+        # A different container should still warn.
+        caplog.clear()
+        with caplog.at_level("WARNING", logger="ai.backend.agent.docker.intrinsic"):
+            _warn_cgroup_fallback_once("CPUPlugin", "container_xyz")
+        warn_records = [r for r in caplog.records if r.levelname == "WARNING"]
+        assert len(warn_records) == 1
+
+        # Same container under a different plugin namespace should also warn once.
+        caplog.clear()
+        with caplog.at_level("WARNING", logger="ai.backend.agent.docker.intrinsic"):
+            _warn_cgroup_fallback_once("MemoryPlugin", "container_abc")
+        warn_records = [r for r in caplog.records if r.levelname == "WARNING"]
+        assert len(warn_records) == 1
+
+    def test_evicts_beyond_limit(
+        self,
+        caplog: pytest.LogCaptureFixture,
+    ) -> None:
+        """When the bounded cache overflows, the oldest entry is evicted and a
+        previously-seen container may warn again."""
+        cap = intrinsic._CGROUP_FALLBACK_WARN_CACHE_SIZE
+        first_cid = "first_container"
+
+        with caplog.at_level("WARNING", logger="ai.backend.agent.docker.intrinsic"):
+            # First warn for `first_cid`.
+            _warn_cgroup_fallback_once("CPUPlugin", first_cid)
+            # Fill the cache with `cap` distinct new entries to evict `first_cid`.
+            for i in range(cap):
+                _warn_cgroup_fallback_once("CPUPlugin", f"filler_{i}")
+            # `first_cid` should have been evicted and now warn again.
+            _warn_cgroup_fallback_once("CPUPlugin", first_cid)
+
+        warn_records = [r for r in caplog.records if r.levelname == "WARNING"]
+        # 1 (first) + cap (fillers) + 1 (re-warn of first) = cap + 2
+        assert len(warn_records) == cap + 2
+
 
 class TestMemoryPluginContainerPidValidation(BaseDockerIntrinsicTest):
     """Tests for container PID validation before reading /proc/[pid]/net/dev."""