fix: force aiohttp's ThreadedResolver to stop the c-ares channel leak

changes/12985.fix.md

@@ -0,0 +1 @@
+Fix a native memory and file-descriptor leak in every long-running service making aiohttp client requests: since aiodns 3.2, aiohttp implicitly defaulted to the aiodns/pycares resolver, leaking one c-ares channel (native heap plus a `/dev/urandom` fd) per ephemeral client session. All service entrypoints now force aiohttp's threaded resolver at startup.

src/ai/backend/account_manager/server.py

@@ -39,6 +39,7 @@
 from ai.backend.common.metrics.multiprocess_setup import cleanup_prometheus_multiprocess_dir
 from ai.backend.common.metrics.profiler import Profiler, PyroscopeArgs
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.types import HostPortPair
 from ai.backend.common.utils import env_info
 from ai.backend.logging import BraceStyleAdapter, Logger, LogLevel
@@ -431,6 +432,7 @@ def main(
     """
     Start the account-manager service as a foreground process.
     """
+    force_threaded_dns_resolver()
     log_level = LogLevel.DEBUG if debug else log_level
     server_config = load_config(config_path, log_level)
 

src/ai/backend/agent/server.py

@@ -103,6 +103,7 @@
 )
 from ai.backend.common.metrics.metric import CommonMetricRegistry
 from ai.backend.common.metrics.profiler import Profiler, PyroscopeArgs
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.service_discovery.etcd_discovery.service_discovery import (
     ETCDServiceDiscovery,
     ETCDServiceDiscoveryArgs,
@@ -1717,6 +1718,7 @@ def main(
     log_level: LogLevel,
 ) -> int:
     """Start the agent service as a foreground process."""
+    force_threaded_dns_resolver()
     if debug:
         log_level = LogLevel.DEBUG
 

src/ai/backend/appproxy/coordinator/server.py

@@ -99,6 +99,7 @@
 from ai.backend.common.metrics.http import build_api_metric_middleware
 from ai.backend.common.metrics.multiprocess_setup import cleanup_prometheus_multiprocess_dir
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.service_discovery.event_publisher import ServiceDiscoveryEventPublisher
 from ai.backend.common.service_discovery.redis_discovery.service_discovery import (
     RedisServiceDiscovery,
@@ -1173,6 +1174,7 @@ def main(ctx: click.Context, config_path: Path | None, debug: bool, log_level: L
     """
     Start the proxy-coordinator service as a foreground process.
     """
+    force_threaded_dns_resolver()
     log_level = LogLevel.DEBUG if debug else log_level
     server_config = load_config(config_path, log_level)
 

src/ai/backend/appproxy/worker/server.py

@@ -100,6 +100,7 @@
 from ai.backend.common.metrics.http import build_api_metric_middleware
 from ai.backend.common.metrics.multiprocess_setup import cleanup_prometheus_multiprocess_dir
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.service_discovery.event_publisher import ServiceDiscoveryEventPublisher
 from ai.backend.common.service_discovery.redis_discovery.service_discovery import (
     RedisServiceDiscovery,
@@ -982,6 +983,7 @@ def main(ctx: click.Context, config_path: Path, debug: bool, log_level: LogLevel
     """
     Start the proxy-worker service as a foreground process.
     """
+    force_threaded_dns_resolver()
     log_level = LogLevel.DEBUG if debug else log_level
     server_config = load_config(config_path, log_level)
 

src/ai/backend/common/networking.py

@@ -6,18 +6,45 @@
 from typing import TYPE_CHECKING, TypeVar, overload
 
 import aiohttp
+import aiohttp.connector
+import aiohttp.resolver
 
 if TYPE_CHECKING:
     import yarl
 
 __all__ = (
     "curl",
     "find_free_port",
+    "force_threaded_dns_resolver",
 )
 
 T = TypeVar("T")
 
 
+def force_threaded_dns_resolver() -> None:
+    """Make aiohttp use its ThreadedResolver regardless of aiodns presence.
+
+    Since aiodns 3.2, aiohttp selects the aiodns/pycares ``AsyncResolver`` as
+    its default resolver whenever the package is importable. Every ephemeral
+    ``ClientSession``/``TCPConnector`` then creates a c-ares channel that is
+    never destroyed, leaking native heap memory and one ``/dev/urandom`` file
+    descriptor per channel on long-running services (see also
+    https://github.com/aio-libs/aiodns/issues/191 for the aiodns >= 3.3
+    variant that additionally exhausts inotify watches).
+
+    aiohttp offers no configuration knob for this choice and injecting
+    ``resolver=`` at every connector creation site is easy to miss, so we
+    override the module-level defaults once at service startup. Our outbound
+    HTTP traffic targets a small set of fixed hosts, making threaded
+    ``getaddrinfo`` resolution entirely sufficient.
+    """
+    aiohttp.resolver.DefaultResolver = aiohttp.resolver.ThreadedResolver
+    # aiohttp.connector re-imports DefaultResolver at import time, so its copy
+    # must be overridden as well; setattr avoids mypy's attr-defined complaint
+    # about the non-re-exported name.
+    setattr(aiohttp.connector, "DefaultResolver", aiohttp.resolver.ThreadedResolver)
+
+
 @overload
 async def curl(
     url: str | yarl.URL,

src/ai/backend/manager/server.py

@@ -46,6 +46,7 @@
 from ai.backend.common.metrics.metric import CommonMetricRegistry
 from ai.backend.common.metrics.multiprocess_setup import cleanup_prometheus_multiprocess_dir
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.utils import env_info
 from ai.backend.logging import BraceStyleAdapter, Logger, LogLevel
 from ai.backend.logging.otel import (
@@ -499,6 +500,7 @@ def main(
     """
     Start the manager service as a foreground process.
     """
+    force_threaded_dns_resolver()
     log_level = LogLevel.DEBUG if debug else log_level
 
     if config_path is None:

src/ai/backend/storage/server.py

@@ -61,6 +61,7 @@
 from ai.backend.common.metrics.multiprocess_setup import cleanup_prometheus_multiprocess_dir
 from ai.backend.common.metrics.profiler import Profiler, PyroscopeArgs
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.plugin import AbstractPlugin, BasePluginContext
 from ai.backend.common.runner.types import Runner
 from ai.backend.common.service_discovery.etcd_discovery.service_discovery import (
@@ -820,6 +821,7 @@ def main(
     debug: bool = False,
 ) -> int:
     """Start the storage-proxy service as a foreground process."""
+    force_threaded_dns_resolver()
     log_level = LogLevel.DEBUG if debug else log_level
     try:
         local_config = load_local_config(config_path, log_level=log_level)

src/ai/backend/web/server.py

@@ -66,6 +66,7 @@
 from ai.backend.common.health_checker.types import ComponentId
 from ai.backend.common.middlewares.exception import general_exception_middleware
 from ai.backend.common.msgpack import DEFAULT_PACK_OPTS, DEFAULT_UNPACK_OPTS
+from ai.backend.common.networking import force_threaded_dns_resolver
 from ai.backend.common.web.session import (
     Session,
     extra_config_headers,
@@ -1289,6 +1290,7 @@ def main(
     debug: bool,
 ) -> None:
     """Start the webui host service as a foreground process."""
+    force_threaded_dns_resolver()
     # Delete this part when you remove --debug option
     raw_cfg = tomli.loads(Path(config_path).read_text(encoding="utf-8"))
 

tests/unit/common/test_networking.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Any
+
+import aiohttp.connector
+import aiohttp.resolver
+import pytest
+
+from ai.backend.common.networking import force_threaded_dns_resolver
+
+
+@pytest.fixture
+def restore_default_resolver() -> Iterator[None]:
+    orig_resolver = aiohttp.resolver.DefaultResolver
+    orig_connector: Any = vars(aiohttp.connector)["DefaultResolver"]
+    yield
+    aiohttp.resolver.DefaultResolver = orig_resolver
+    setattr(aiohttp.connector, "DefaultResolver", orig_connector)
+
+
+def test_force_threaded_dns_resolver_overrides_defaults(
+    restore_default_resolver: None,
+) -> None:
+    force_threaded_dns_resolver()
+    assert aiohttp.resolver.DefaultResolver is aiohttp.resolver.ThreadedResolver
+    assert vars(aiohttp.connector)["DefaultResolver"] is aiohttp.resolver.ThreadedResolver
+
+
+async def test_new_connector_uses_threaded_resolver(
+    restore_default_resolver: None,
+) -> None:
+    force_threaded_dns_resolver()
+    conn = aiohttp.TCPConnector()
+    try:
+        # The connector must not fall back to the aiodns/pycares AsyncResolver,
+        # whose per-instance c-ares channels leak on long-running services.
+        assert isinstance(conn._resolver, aiohttp.resolver.ThreadedResolver)
+    finally:
+        await conn.close()