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

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. The pool is a context
+    # manager: exiting it tears down every pooled agent this worker created.
+    with scenarios.parametric.get_agent_pool(worker_id) as pool:
+        yield pool
 
 
 @pytest.fixture
@@ -82,7 +92,27 @@ 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:
+        # agent_env is empty here (poolable requires `not agent_env`); pass the default
+        # explicitly rather than the always-falsy variable.
+        api = test_agent_pool.acquire(request=request, 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,35 @@
+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:
+        with 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,
+        ) as api:
+            assert api.info()["version"] == "test"  # agent answered → it is ready
+    finally:
+        client.remove()

tests/test_the_test/test_test_agent_pool.py

@@ -0,0 +1,73 @@
+import contextlib
+from collections.abc import Iterator
+
+from utils.docker_fixtures._test_agent_pool import 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:
+    """Context-manager creator: yields a _FakeApi and counts teardowns on exit."""
+
+    def __init__(self) -> None:
+        self.created_envs: list[dict] = []
+        self.stopped = 0
+
+    @contextlib.contextmanager
+    def __call__(self, request: object, agent_env: dict[str, str]) -> Iterator[_FakeApi]:  # noqa: ARG002
+        self.created_envs.append(dict(agent_env))
+        try:
+            yield _FakeApi()
+        finally:
+            self.stopped += 1
+
+
+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()
+    with WorkerAgentPool(creator) as pool:
+        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()
+    with WorkerAgentPool(creator) as pool:
+        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_exit_tears_down_every_agent():
+    creator = _FakeCreator()
+    with WorkerAgentPool(creator) as pool:
+        pool.acquire(request="r", agent_env={})
+        pool.acquire(request="r", agent_env={"DD_ENV": "prod"})
+        assert creator.stopped == 0  # still open inside the context
+
+    assert creator.stopped == 2  # both agents torn down on exit

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, agent_env_key
 from utils._context.docker import get_docker_client
 from utils._context.constants import WeblogCategory
 from utils._logger import logger
@@ -35,6 +42,7 @@ def __init__(
         )
 
         self._test_agent_factory = TestAgentFactory(agent_image)
+        self._agent_pool: WorkerAgentPool | None = None
 
     def _clean(self):
         if self.is_main_worker:
@@ -76,15 +84,62 @@ 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:
+
+            @contextlib.contextmanager
+            def _creator(
+                request: pytest.FixtureRequest, agent_env: dict[str, str]
+            ) -> Generator[TestAgentAPI, None, None]:
+                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 (5000/5100/5200 + worker offset)
+                # so a worker's persistent pooled agent never collides, on that worker, with the
+                # fresh-path agent (4600/4701/4802) or the FFE mock backend (4900, added on main
+                # in MockFFEAgentlessBackendServer). Bands stay non-overlapping for up to ~97
+                # concurrent xdist workers, well above any real run.
+                try:
+                    with 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=5000,
+                        otlp_http_port_base=5100,
+                        otlp_grpc_port_base=5200,
+                    ) as api:
+                        yield api
+                finally:
+                    try:
+                        network.remove()
+                    except Exception as e:
+                        logger.info(f"Failed to remove worker network, ignoring: {e}")
+
+            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,