fix(BA-4938): pre-validate namespace path before netstat_ns to prevent thread pool exhaustion (#9782)

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: fregataa

changes/9782.fix.md

@@ -0,0 +1 @@
+Pre-validate namespace path before `netstat_ns()` to prevent thread pool exhaustion from hung threads on stale network namespaces

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

@@ -715,20 +715,36 @@ async def sysfs_impl(
             sandbox_key = data["NetworkSettings"]["SandboxKey"]
             net_rx_bytes = 0
             net_tx_bytes = 0
-            try:
-                nstat = await netstat_ns(sandbox_key)
-            except OSError as e:
+            if not sandbox_key:
                 log.warning(
-                    "MemoryPlugin: cannot read net stats for container {0}: {1!r}",
+                    "MemoryPlugin: empty SandboxKey for container {0},"
+                    " skipping net stat collection",
                     container_id[:7],
-                    e,
                 )
-                return None
-            for name, net_stat in nstat.items():
-                if name == "lo":
-                    continue
-                net_rx_bytes += net_stat.bytes_recv
-                net_tx_bytes += net_stat.bytes_sent
+            else:
+                ns_path = Path(sandbox_key)
+                if not ns_path.exists():
+                    log.warning(
+                        "MemoryPlugin: network namespace path does not exist for container"
+                        " {0} (sandbox_key={1!r}), skipping net stat collection",
+                        container_id[:7],
+                        sandbox_key,
+                    )
+                else:
+                    try:
+                        nstat = await netstat_ns(ns_path)
+                    except OSError as e:
+                        log.warning(
+                            "MemoryPlugin: cannot read net stats for container {0}: {1!r}",
+                            container_id[:7],
+                            e,
+                        )
+                        return None
+                    for name, net_stat in nstat.items():
+                        if name == "lo":
+                            continue
+                        net_rx_bytes += net_stat.bytes_recv
+                        net_tx_bytes += net_stat.bytes_sent
             loop = current_loop()
             scratch_sz = await loop.run_in_executor(None, get_scratch_size, container_id)
             return (

tests/unit/agent/test_docker_intrinsic.py

@@ -5,6 +5,7 @@
 import time
 from collections.abc import Generator, Iterator
 from concurrent.futures import ProcessPoolExecutor
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -148,7 +149,7 @@ def memory_plugin(self) -> MemoryPlugin:
 
     @pytest.fixture
     def memory_cgroup_context(
-        self, cgroup_stat_context: MagicMock
+        self, cgroup_stat_context: MagicMock, tmp_path: Path
     ) -> Generator[MagicMock, None, None]:
         """CGROUP stat context with memory/io cgroup v2 path mocks and related patches."""
         ctx = cgroup_stat_context
@@ -170,8 +171,10 @@ def mock_get_cgroup_path(subsys: str, cid: str) -> MagicMock:
 
         ctx.agent.get_cgroup_path = mock_get_cgroup_path
 
+        sandbox_file = tmp_path / "fake_netns"
+        sandbox_file.touch()
         mock_container_data = {
-            "NetworkSettings": {"SandboxKey": "/var/run/docker/netns/fake"},
+            "NetworkSettings": {"SandboxKey": str(sandbox_file)},
         }
 
         with (
@@ -234,6 +237,135 @@ async def test_sysfs_mode_uses_instance_docker_client(
             mock_docker_cls.assert_not_called()
 
 
+class TestMemoryPluginNamespaceValidation(BaseDockerIntrinsicTest):
+    """Tests for namespace path pre-validation before netstat_ns call."""
+
+    @pytest.fixture
+    def memory_plugin(self) -> MemoryPlugin:
+        plugin = MemoryPlugin.__new__(MemoryPlugin)
+        plugin.local_config = {"agent": {"docker-mode": "default"}}
+        plugin._docker = AsyncMock()
+        return plugin
+
+    @contextmanager
+    def _make_cgroup_context(
+        self,
+        cgroup_stat_context: MagicMock,
+        sandbox_key: str,
+    ) -> Generator[tuple[MagicMock, MagicMock], None, None]:
+        """Build a CGROUP stat context with configurable sandbox_key.
+
+        Pass a real filesystem path for sandbox_key — an existing path
+        triggers netstat_ns, a non-existent one skips it.
+        """
+        ctx = cgroup_stat_context
+        ctx.agent.get_cgroup_version = MagicMock(return_value="2")
+
+        mem_path = MagicMock()
+        mem_stat = MagicMock()
+        mem_stat.read_text.return_value = "inactive_file 0\n"
+        mem_path.__truediv__ = MagicMock(return_value=mem_stat)
+        io_path = MagicMock()
+        io_stat = MagicMock()
+        io_stat.read_text.return_value = ""
+        io_path.__truediv__ = MagicMock(return_value=io_stat)
+
+        def mock_get_cgroup_path(subsys: str, cid: str) -> MagicMock:
+            if subsys == "memory":
+                return mem_path
+            return io_path
+
+        ctx.agent.get_cgroup_path = mock_get_cgroup_path
+
+        mock_container_data = {
+            "NetworkSettings": {"SandboxKey": sandbox_key},
+        }
+
+        with (
+            patch(
+                "ai.backend.agent.docker.intrinsic.DockerContainer",
+            ) as mock_container_cls,
+            patch(
+                "ai.backend.agent.docker.intrinsic.read_sysfs",
+                return_value=1048576,
+            ),
+            patch(
+                "ai.backend.agent.docker.intrinsic.netstat_ns",
+                new_callable=AsyncMock,
+            ) as mock_netstat,
+            patch(
+                "ai.backend.agent.docker.intrinsic.current_loop",
+            ) as mock_loop,
+        ):
+            mock_netstat.return_value = {
+                "eth0": MagicMock(bytes_recv=4096, bytes_sent=8192),
+            }
+            mock_container_instance = AsyncMock()
+            mock_container_instance.show.return_value = mock_container_data
+            mock_container_cls.return_value = mock_container_instance
+            mock_loop.return_value.run_in_executor = AsyncMock(return_value=0)
+            yield ctx, mock_netstat
+
+    async def test_nonexistent_namespace_path_returns_zero_net_stats(
+        self,
+        memory_plugin: MemoryPlugin,
+        cgroup_stat_context: MagicMock,
+        tmp_path: Path,
+    ) -> None:
+        """When namespace path does not exist, net stats should be 0 but other stats collected."""
+        gone_path = tmp_path / "nonexistent_netns"
+        with self._make_cgroup_context(
+            cgroup_stat_context,
+            sandbox_key=str(gone_path),
+        ) as (ctx, mock_netstat):
+            results = await memory_plugin.gather_container_measures(ctx, ["cid_001"])
+            mock_netstat.assert_not_called()
+            # mem stats should be collected (read_sysfs returns 1048576)
+            assert results[0].per_container["cid_001"].value == 1048576
+            # net_rx and net_tx should be 0
+            assert results[3].per_container["cid_001"].value == 0
+            assert results[4].per_container["cid_001"].value == 0
+
+    async def test_empty_sandbox_key_returns_zero_net_stats(
+        self,
+        memory_plugin: MemoryPlugin,
+        cgroup_stat_context: MagicMock,
+    ) -> None:
+        """When sandbox_key is empty string, net stats should be 0 but other stats collected."""
+        with self._make_cgroup_context(
+            cgroup_stat_context,
+            sandbox_key="",
+        ) as (ctx, mock_netstat):
+            results = await memory_plugin.gather_container_measures(ctx, ["cid_001"])
+            mock_netstat.assert_not_called()
+            # mem stats should be collected
+            assert results[0].per_container["cid_001"].value == 1048576
+            # net_rx and net_tx should be 0
+            assert results[3].per_container["cid_001"].value == 0
+            assert results[4].per_container["cid_001"].value == 0
+
+    async def test_valid_namespace_path_calls_netstat_ns(
+        self,
+        memory_plugin: MemoryPlugin,
+        cgroup_stat_context: MagicMock,
+        tmp_path: Path,
+    ) -> None:
+        """When namespace path exists, netstat_ns should be called and net stats collected."""
+        valid_path = tmp_path / "valid_netns"
+        valid_path.touch()
+        with self._make_cgroup_context(
+            cgroup_stat_context,
+            sandbox_key=str(valid_path),
+        ) as (ctx, mock_netstat):
+            results = await memory_plugin.gather_container_measures(ctx, ["cid_001"])
+            mock_netstat.assert_called()
+            # mem stats should be collected
+            assert results[0].per_container["cid_001"].value == 1048576
+            # net_rx and net_tx should have values from mock netstat_ns
+            assert results[3].per_container["cid_001"].value == 4096
+            assert results[4].per_container["cid_001"].value == 8192
+
+
 @pytest.mark.skipif(sys.platform != "linux", reason="Network namespaces require Linux")
 class TestNetstatNsWork:
     """Tests for netstat_ns_work with real namespace switching."""