[parametric] reuse the test-agent per xdist worker to cut container churn (POC)

tests/parametric/conftest.py

@@ -11,6 +11,8 @@
 
 from utils import scenarios, logger
 from utils.docker_fixtures import TestAgentAPI, ParametricTestClientApi as APMLibrary
+from utils.docker_fixtures._test_agent import DEFAULT_OTLP_HTTP_PORT, DEFAULT_OTLP_GRPC_PORT
+from utils.docker_fixtures._test_agent_pool import WorkerAgentPool
 
 
 # Max timeout in seconds to keep a container running
@@ -66,12 +68,20 @@ def docker() -> str | None:
 
 @pytest.fixture
 def test_agent_otlp_http_port() -> int:
-    return 4318
+    return DEFAULT_OTLP_HTTP_PORT
 
 
 @pytest.fixture
 def test_agent_otlp_grpc_port() -> int:
-    return 4317
+    return DEFAULT_OTLP_GRPC_PORT
+
+
+@pytest.fixture(scope="session")
+def test_agent_pool(worker_id: str) -> Generator[WorkerAgentPool, None, None]:
+    # scope="session" under pytest-xdist == once per worker.
+    pool = scenarios.parametric.get_agent_pool(worker_id)
+    yield pool
+    pool.shutdown()
 
 
 @pytest.fixture
@@ -82,7 +92,25 @@ def test_agent(
     agent_env: dict[str, str],
     test_agent_otlp_http_port: int,
     test_agent_otlp_grpc_port: int,
+    test_agent_pool: WorkerAgentPool,
 ) -> Generator[TestAgentAPI, None, None]:
+    # POC: pool only default-agent_env, default-OTLP-port, non-snapshot tests.
+    # Snapshot-marked tests need per-test snapshot_context lifecycle; non-default
+    # agent_env would require a second pooled agent per worker (worker-keyed host ports
+    # would collide); a parametrized custom container OTLP port needs an agent listening
+    # on that port, which the pool's fixed-port agent cannot serve. All fall back to the
+    # fresh-per-test path. Pooled agents are reset with clear() between tests.
+    poolable = (
+        request.node.get_closest_marker("snapshot") is None
+        and not agent_env
+        and test_agent_otlp_http_port == DEFAULT_OTLP_HTTP_PORT
+        and test_agent_otlp_grpc_port == DEFAULT_OTLP_GRPC_PORT
+    )
+    if poolable:
+        api = test_agent_pool.acquire(request=request, agent_env=agent_env)
+        yield api
+        return  # REQUIRED: do not fall through into the fresh-path agent below
+
     with scenarios.parametric.get_test_agent_api(
         request=request,
         worker_id=worker_id,

tests/test_the_test/test_start_stop_agent.py

@@ -0,0 +1,38 @@
+import os
+
+import pytest
+
+from utils._context._scenarios import scenarios
+from utils.docker_fixtures._core import get_docker_client
+from utils.docker_fixtures._test_agent import DEFAULT_OTLP_HTTP_PORT, DEFAULT_OTLP_GRPC_PORT
+
+
+def test_start_agent_then_stop(request: pytest.FixtureRequest):
+    # Docker-backed smoke test. Gate at runtime rather than with a module-level
+    # pytest.mark.skipif: the root conftest's _item_must_pass iterates skipif
+    # marker.args[0] (all(marker.args[0])), which raises on a bool condition and
+    # crashes collection for every session that collects this file.
+    if os.getenv("DOCKER_SMOKE") != "1":
+        pytest.skip("Docker-backed; run with DOCKER_SMOKE=1")
+
+    factory = scenarios.parametric._test_agent_factory  # noqa: SLF001
+    factory.configure("logs_parametric")
+    factory.pull()
+
+    client = get_docker_client().networks.create(name="reuse_smoke_net", driver="bridge")
+    try:
+        api, stop = factory.start_agent(
+            request=request,
+            worker_id="gw0",
+            container_name="ddapm-test-agent-reuse-smoke",
+            docker_network=client.name,
+            agent_env={},
+            container_otlp_http_port=DEFAULT_OTLP_HTTP_PORT,
+            container_otlp_grpc_port=DEFAULT_OTLP_GRPC_PORT,
+        )
+        try:
+            assert api.info()["version"] == "test"  # agent answered → it is ready
+        finally:
+            stop()
+    finally:
+        client.remove()

tests/test_the_test/test_test_agent_pool.py

@@ -0,0 +1,77 @@
+from utils.docker_fixtures._test_agent_pool import AgentLease, WorkerAgentPool, agent_env_key
+
+
+class _FakeApi:
+    def __init__(self) -> None:
+        self.clear_calls = 0
+        self.reset_rc_calls = 0
+        self.rebind_calls: list[object] = []
+
+    def clear(self) -> None:
+        self.clear_calls += 1
+
+    def reset_remote_config(self) -> None:
+        self.reset_rc_calls += 1
+
+    def rebind_request(self, request: object) -> None:
+        self.rebind_calls.append(request)
+
+
+class _FakeCreator:
+    """Records every creation and hands back a lease whose stop() is tracked."""
+
+    def __init__(self) -> None:
+        self.created_envs: list[dict] = []
+        self.stopped = 0
+
+    def __call__(self, request: object, agent_env: dict[str, str]) -> AgentLease:  # noqa: ARG002
+        self.created_envs.append(dict(agent_env))
+        api = _FakeApi()
+
+        def _stop() -> None:
+            self.stopped += 1
+
+        return AgentLease(api=api, stop=_stop)
+
+
+def test_env_key_is_order_independent():
+    assert agent_env_key({"A": "1", "B": "2"}) == agent_env_key({"B": "2", "A": "1"})
+
+
+def test_same_env_reuses_and_clears():
+    creator = _FakeCreator()
+    pool = WorkerAgentPool(creator)
+
+    api1 = pool.acquire(request="req1", agent_env={})
+    api2 = pool.acquire(request="req2", agent_env={})
+
+    assert api1 is api2  # reused, not recreated
+    assert len(creator.created_envs) == 1  # created exactly once
+    assert api1.clear_calls == 2  # cleared on both acquires (first + reuse)
+    assert api1.reset_rc_calls == 2  # remote-config reset on both acquires too
+    assert api1.rebind_calls == ["req2"]  # rebound only on reuse, not on first acquire
+
+
+def test_distinct_env_creates_separate_agents():
+    creator = _FakeCreator()
+    pool = WorkerAgentPool(creator)
+
+    a = pool.acquire(request="r", agent_env={})
+    b = pool.acquire(request="r", agent_env={"DD_ENV": "prod"})
+
+    assert a is not b
+    assert len(creator.created_envs) == 2
+
+
+def test_shutdown_stops_every_lease():
+    creator = _FakeCreator()
+    pool = WorkerAgentPool(creator)
+    pool.acquire(request="r", agent_env={})
+    pool.acquire(request="r", agent_env={"DD_ENV": "prod"})
+
+    pool.shutdown()
+
+    assert creator.stopped == 2
+    # After shutdown the cache is empty: acquiring again creates anew.
+    pool.acquire(request="r", agent_env={})
+    assert len(creator.created_envs) == 3

utils/_context/_scenarios/_docker_fixtures.py

@@ -5,8 +5,15 @@
 
 import pytest
 
-from utils.docker_fixtures._test_agent import TestAgentFactory, TestAgentAPI
 from docker.errors import DockerException
+
+from utils.docker_fixtures._test_agent import (
+    TestAgentFactory,
+    TestAgentAPI,
+    DEFAULT_OTLP_HTTP_PORT,
+    DEFAULT_OTLP_GRPC_PORT,
+)
+from utils.docker_fixtures._test_agent_pool import WorkerAgentPool, AgentLease, agent_env_key
 from utils._context.docker import get_docker_client
 from utils._logger import logger
 from .core import Scenario, ScenarioGroup, scenario_groups as groups
@@ -32,6 +39,7 @@ def __init__(
         )
 
         self._test_agent_factory = TestAgentFactory(agent_image)
+        self._agent_pool: WorkerAgentPool | None = None
 
     def _clean(self):
         if self.is_main_worker:
@@ -73,15 +81,70 @@ def _get_docker_network(self, test_id: str) -> Generator[str, None, None]:
                 # Let's ignore this, later calls will clean the mess
                 logger.info(f"Failed to remove network, ignoring the error: {e}")
 
+    def get_agent_pool(self, worker_id: str) -> WorkerAgentPool:
+        # POC: only the default agent_env ({}) is pooled, so there is at most one
+        # pooled agent per xdist worker.  That is why start_agent's worker-keyed
+        # host ports are safe — no two pooled agents on the same worker can collide.
+        # Supporting multiple envs per worker would require per-(worker, env) port
+        # allocation and is out of scope for this POC.
+        if self._agent_pool is None:
+
+            def _creator(request: pytest.FixtureRequest, agent_env: dict[str, str]) -> AgentLease:
+                key = agent_env_key(agent_env)
+                network_name = f"{_NETWORK_PREFIX}_worker_{worker_id}_{abs(hash(key))}"
+                network = get_docker_client().networks.create(name=network_name, driver="bridge")
+                container_name = f"ddapm-test-agent-worker-{worker_id}-{abs(hash(key))}"
+                # Pooled agents use a separate host-port band (4900/5000/5100 + worker offset)
+                # so a worker's persistent pooled agent never collides with a fresh-path agent
+                # (4600/4701/4802) on that worker. The bands are non-overlapping for up to ~97
+                # concurrent xdist workers (well above any real run): fresh OTLP-gRPC base 4802
+                # + 98 == pooled base 4900 + 0.
+                try:
+                    api, stop_agent = self._test_agent_factory.start_agent(
+                        request=request,
+                        worker_id=worker_id,
+                        container_name=container_name,
+                        docker_network=network.name,
+                        agent_env=agent_env,
+                        # Fixed container OTLP ports for the pooled agent. The poolable check
+                        # in tests/parametric/conftest.py only pools tests using these ports,
+                        # since a parametrized custom OTLP port needs an agent listening on it.
+                        container_otlp_http_port=DEFAULT_OTLP_HTTP_PORT,
+                        container_otlp_grpc_port=DEFAULT_OTLP_GRPC_PORT,
+                        agent_port_base=4900,
+                        otlp_http_port_base=5000,
+                        otlp_grpc_port_base=5100,
+                    )
+                except BaseException:
+                    try:
+                        network.remove()
+                    except Exception as e:
+                        logger.info(f"Failed to remove network after start_agent failure, ignoring: {e}")
+                    raise
+
+                def _stop() -> None:
+                    try:
+                        stop_agent()
+                    finally:
+                        try:
+                            network.remove()
+                        except Exception as e:
+                            logger.info(f"Failed to remove worker network, ignoring: {e}")
+
+                return AgentLease(api=api, stop=_stop)
+
+            self._agent_pool = WorkerAgentPool(_creator)
+        return self._agent_pool
+
     @contextlib.contextmanager
     def get_test_agent_api(
         self,
         worker_id: str,
         request: pytest.FixtureRequest,
         test_id: str,
         agent_env: dict[str, str],
-        container_otlp_http_port: int = 4318,
-        container_otlp_grpc_port: int = 4317,
+        container_otlp_http_port: int = DEFAULT_OTLP_HTTP_PORT,
+        container_otlp_grpc_port: int = DEFAULT_OTLP_GRPC_PORT,
     ) -> Generator[TestAgentAPI, None, None]:
         with (
             self._get_docker_network(test_id) as docker_network,