diff --git a/tests/parametric/conftest.py b/tests/parametric/conftest.py index 5b3cb343844..0c8d37874a2 100644 --- a/tests/parametric/conftest.py +++ b/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, diff --git a/tests/test_the_test/test_start_stop_agent.py b/tests/test_the_test/test_start_stop_agent.py new file mode 100644 index 00000000000..79ba068dbc2 --- /dev/null +++ b/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() diff --git a/tests/test_the_test/test_test_agent_pool.py b/tests/test_the_test/test_test_agent_pool.py new file mode 100644 index 00000000000..f996c8285ac --- /dev/null +++ b/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 diff --git a/utils/_context/_scenarios/_docker_fixtures.py b/utils/_context/_scenarios/_docker_fixtures.py index 136a730edd4..39d093feb33 100644 --- a/utils/_context/_scenarios/_docker_fixtures.py +++ b/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,6 +84,53 @@ 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, @@ -83,8 +138,8 @@ def get_test_agent_api( 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, diff --git a/utils/docker_fixtures/_test_agent.py b/utils/docker_fixtures/_test_agent.py index 92845e75931..a8e3a2593f8 100644 --- a/utils/docker_fixtures/_test_agent.py +++ b/utils/docker_fixtures/_test_agent.py @@ -27,6 +27,14 @@ from ._core import HOST_GATEWAY_EXTRA_HOSTS, get_host_port, get_docker_client, docker_run +# Default container-internal OTLP ports for the test-agent. The pooled parametric agent +# (DockerFixturesScenario.get_agent_pool) is created with these, and the poolable check +# in tests/parametric/conftest.py only pools tests that use them — a test that +# parametrizes a custom OTLP port needs a fresh agent listening on that port. Single +# source of truth so the pool creator and the poolable check cannot drift apart. +DEFAULT_OTLP_HTTP_PORT = 4318 +DEFAULT_OTLP_GRPC_PORT = 4317 + def _request_token(request: pytest.FixtureRequest) -> str: token = "" @@ -36,6 +44,11 @@ def _request_token(request: pytest.FixtureRequest) -> str: return token +def _agent_log_path(host_log_folder: str, request: "pytest.FixtureRequest") -> str: + cls_name = request.cls.__name__ if request.cls else "NoClass" + return f"{host_log_folder}/outputs/{cls_name}/{request.node.name}/agent_log.log" + + class AgentRequest(TypedDict): method: str url: str @@ -72,8 +85,9 @@ def pull(self) -> None: get_docker_client().images.pull(self.image) @contextlib.contextmanager - def get_test_agent_api( + def start_agent( self, + *, request: pytest.FixtureRequest, worker_id: str, container_name: str, @@ -81,27 +95,34 @@ def get_test_agent_api( agent_env: dict[str, str], container_otlp_http_port: int, container_otlp_grpc_port: int, + agent_port_base: int = 4600, + otlp_http_port_base: int = 4701, + otlp_grpc_port_base: int = 4802, ) -> Generator["TestAgentAPI", None, None]: - # (meta_tracer_version_header) Not all clients (go for example) submit the tracer version - # (trace_content_length) go client doesn't submit content length header + """Start a test-agent container and yield its API, tearing it down on exit. + + A context manager so the container + log file are released cleanly on any error + during setup or teardown. The fresh-per-test path uses it directly with `with`; + the pooled path (WorkerAgentPool) keeps it open across tests via an ExitStack and + closes it at pool shutdown. + """ env = { "ENABLED_CHECKS": "trace_count_header", "OTLP_HTTP_PORT": str(container_otlp_http_port), "OTLP_GRPC_PORT": str(container_otlp_grpc_port), - "VCR_CASSETTES_DIRECTORY": "/vcr-cassettes", # TODO comment + "VCR_CASSETTES_DIRECTORY": "/vcr-cassettes", } if os.getenv("DEV_MODE") is not None: env["SNAPSHOT_CI"] = "0" - env |= agent_env - host_port = get_host_port(worker_id, 4600) - container_port = 8126 - otlp_http_host_port = get_host_port(worker_id, 4701) - otlp_grpc_host_port = get_host_port(worker_id, 4802) + host_port = get_host_port(worker_id, agent_port_base) + otlp_http_host_port = get_host_port(worker_id, otlp_http_port_base) + otlp_grpc_host_port = get_host_port(worker_id, otlp_grpc_port_base) - log_path = f"{self.host_log_folder}/outputs/{request.cls.__name__}/{request.node.name}/agent_log.log" + log_path = _agent_log_path(self.host_log_folder, request) Path(log_path).parent.mkdir(parents=True, exist_ok=True) + logger.stdout(f"REUSE-POC agent container created: {container_name}") with ( open(log_path, "w+", encoding="utf-8") as log_file, @@ -114,7 +135,7 @@ def get_test_agent_api( "./tests/integration_frameworks/utils/vcr-cassettes": "/vcr-cassettes", }, ports={ - f"{container_port}/tcp": host_port, + "8126/tcp": host_port, f"{container_otlp_http_port}/tcp": otlp_http_host_port, f"{container_otlp_grpc_port}/tcp": otlp_grpc_host_port, }, @@ -125,14 +146,14 @@ def get_test_agent_api( ): client = TestAgentAPI( container_name, - container_port, + 8126, self.host_log_folder, pytest_request=request, otlp_http_host_port=otlp_http_host_port, host_port=host_port, network=docker_network, ) - time.sleep(0.2) # initial wait time, the trace agent takes 200ms to start + time.sleep(0.2) # the trace agent takes ~200ms to start expected_version = agent_env.get("TEST_AGENT_VERSION", "test") for _ in range(100): try: @@ -142,30 +163,53 @@ def get_test_agent_api( time.sleep(0.1) else: if resp["version"] != expected_version: - message = f"""Agent version {resp["version"]} is running instead of the test agent. - Stop the agent on port {container_port} and try again.""" - pytest.fail(message, pytrace=False) - + pytest.fail( + f"Agent version {resp['version']} is running instead of the test agent.", + pytrace=False, + ) logger.info("Test agent is ready") break else: - logger.error("Could not connect to test agent") pytest.fail(f"Could not connect to test agent, check the log file {log_file.name}.", pytrace=False) - # If the snapshot mark is on the test case then do a snapshot test - marks = list(request.node.iter_markers(name="snapshot")) - assert len(marks) <= 1, "Multiple snapshot marks detected" - if marks: - snap = marks[0] - assert len(snap.args) == 0, "only keyword arguments are supported by the snapshot decorator" - if "token" not in snap.kwargs: - snap.kwargs["token"] = _request_token(request).replace(" ", "_").replace(os.path.sep, "_") - with client.snapshot_context(**snap.kwargs): - yield client - else: - yield client + yield client - request.node.add_report_section("teardown", "Test Agent Output", f"Log file:\n./{log_path}") + @contextlib.contextmanager + def get_test_agent_api( + self, + request: pytest.FixtureRequest, + worker_id: str, + container_name: str, + docker_network: str, + agent_env: dict[str, str], + container_otlp_http_port: int, + container_otlp_grpc_port: int, + ) -> Generator["TestAgentAPI", None, None]: + try: + with self.start_agent( + request=request, + worker_id=worker_id, + container_name=container_name, + docker_network=docker_network, + agent_env=agent_env, + container_otlp_http_port=container_otlp_http_port, + container_otlp_grpc_port=container_otlp_grpc_port, + ) as client: + # If the snapshot mark is on the test case then do a snapshot test + marks = list(request.node.iter_markers(name="snapshot")) + assert len(marks) <= 1, "Multiple snapshot marks detected" + if marks: + snap = marks[0] + assert len(snap.args) == 0, "only keyword arguments are supported by the snapshot decorator" + if "token" not in snap.kwargs: + snap.kwargs["token"] = _request_token(request).replace(" ", "_").replace(os.path.sep, "_") + with client.snapshot_context(**snap.kwargs): + yield client + else: + yield client + finally: + log_path = _agent_log_path(self.host_log_folder, request) + request.node.add_report_section("teardown", "Test Agent Output", f"Log file:\n./{log_path}") class TestAgentAPI: @@ -186,6 +230,7 @@ def __init__( self.container_name = container_name self.container_port = container_port self.network = network + self._host_log_folder = host_log_folder self.host = "localhost" self.host_port = host_port @@ -193,9 +238,15 @@ def __init__( self._session = requests.Session() self._pytest_request = pytest_request - self.log_path = ( - f"{host_log_folder}/outputs/{pytest_request.cls.__name__}/{pytest_request.node.name}/agent_api.log" - ) + cls_name = pytest_request.cls.__name__ if pytest_request.cls else "NoClass" + self.log_path = f"{host_log_folder}/outputs/{cls_name}/{pytest_request.node.name}/agent_api.log" + Path(self.log_path).parent.mkdir(parents=True, exist_ok=True) + + def rebind_request(self, pytest_request: "pytest.FixtureRequest") -> None: + """Re-point per-test API logging at the current test (used on reuse).""" + self._pytest_request = pytest_request + cls_name = pytest_request.cls.__name__ if pytest_request.cls else "NoClass" + self.log_path = f"{self._host_log_folder}/outputs/{cls_name}/{pytest_request.node.name}/agent_api.log" Path(self.log_path).parent.mkdir(parents=True, exist_ok=True) def _url(self, path: str) -> str: @@ -396,9 +447,31 @@ def wait_for_num_v06_stats(self, num: int, *, wait_loops: int = 200) -> list[Age raise ValueError(f"Number ({num}) of /v0.6/stats requests not received, got {len(requests)}") def clear(self) -> None: - self._session.get(self._url("/test/session/clear")) + # Drops recorded requests (traces, telemetry, integrations) only. Also used + # mid-test by the clear=True helpers, so it must NOT touch the remote-config the + # agent serves -- wiping a test's active RC mid-test would change tracer behavior + # before the test asserts it. The pooled-reuse RC reset lives in + # reset_remote_config(), called only between tests. Safety-critical for reuse (a + # silent failure leaves the next test on dirty state), so surface a bad status. + self._session.get(self._url("/test/session/clear")).raise_for_status() + # The OTLP test-agent's clear is best-effort and can return non-2xx (e.g. 400); + # don't fail the reset on it (matches the original fire-and-forget behavior). self._session.get(self._otlp_url("/test/session/clear")) + def reset_remote_config(self) -> None: + """Restore the served remote-config to the empty `{}` state a fresh agent has. + + Kept out of `clear()` on purpose: `clear()` is also invoked mid-test by the + `clear=True` helpers, where wiping the active config would flip what the tracer + polls before the test asserts on it. Only the pooled-reuse path (between tests) + calls this, so a pooled non-snapshot test starts from a clean RC baseline -- + e.g. the dynamic-config sampling tests assert the default rate before applying + their own RC. `/test/session/clear` does not reset RC (RemoteConfigServer + keeps `_responses` keyed by token), and non-snapshot tests share the default + token, so a prior test's RC would otherwise leak in. + """ + self._session.post(self._url("/test/session/responses/config"), json={}).raise_for_status() + def info(self): resp = self._session.get(self._url("/info")) diff --git a/utils/docker_fixtures/_test_agent_pool.py b/utils/docker_fixtures/_test_agent_pool.py new file mode 100644 index 00000000000..32534d433ab --- /dev/null +++ b/utils/docker_fixtures/_test_agent_pool.py @@ -0,0 +1,88 @@ +from collections.abc import Callable +import contextlib +from contextlib import AbstractContextManager +from typing import Any + +from utils._logger import logger + + +def agent_env_key(agent_env: dict[str, str]) -> tuple: + """Stable, hashable, order-independent key for an agent_env dict.""" + return tuple(sorted(agent_env.items())) + + +class WorkerAgentPool(contextlib.AbstractContextManager): + """Per-worker cache of test-agents keyed by agent_env. + + First request for a given env opens an agent via `creator` (a context manager + yielding a TestAgentAPI) and keeps it open for the whole worker session; later + requests for the same env reuse it after clear()/reset_remote_config() (reset state) + and rebind_request() (re-point per-test logging at the current test). + + Use as a context manager so every pooled agent is torn down exactly once at exit -- + consumers never call a shutdown method directly: + + with WorkerAgentPool(creator) as pool: + api = pool.acquire(request, agent_env) + + Resilience: an agent is cached only after its reset succeeds, and a reset failure on + a reused agent evicts and recreates it. That stops one dead/blipped agent from + cascading into setup failures for every later poolable test on the worker. + """ + + # creator(request, agent_env) -> context manager yielding a TestAgentAPI. Duck-typed + # (a real FixtureRequest + TestAgentAPI in prod, fakes in tests); Any keeps this + # module Docker-free. Each entry keeps its own ExitStack so it can be torn down + # individually (eviction) or in bulk (pool exit). + def __init__(self, creator: Callable[[Any, dict[str, str]], AbstractContextManager[Any]]) -> None: + self._creator = creator + self._entries: dict[tuple, tuple[Any, contextlib.ExitStack]] = {} + + def acquire(self, request: Any, agent_env: dict[str, str]) -> Any: # noqa: ANN401 + key = agent_env_key(agent_env) + entry = self._entries.get(key) + if entry is not None: + api = entry[0] + api.rebind_request(request) + try: + self._reset(api) + except Exception as e: + # Any reset failure means a dirty/dead agent. Don't let it cascade: drop + # it and recreate below. + logger.info(f"Pooled agent reset failed, recreating it: {e}") + self._close_entry(key) + else: + return api + return self._open_entry(key, request, agent_env) + + def _open_entry(self, key: tuple, request: Any, agent_env: dict[str, str]) -> Any: # noqa: ANN401 + # Reset before caching so a create whose reset fails doesn't leave a broken entry. + stack = contextlib.ExitStack() + try: + api = stack.enter_context(self._creator(request, agent_env)) + self._reset(api) + except BaseException: + stack.close() + raise + self._entries[key] = (api, stack) + return api + + @staticmethod + def _reset(api: Any) -> None: # noqa: ANN401 + # Drop recorded requests, then restore the served remote-config to a fresh-agent + # state. The RC reset is separate from clear() so mid-test clear=True helpers + # don't wipe a test's active config. + api.clear() + api.reset_remote_config() + + def _close_entry(self, key: tuple) -> None: + entry = self._entries.pop(key, None) + if entry is not None: + try: + entry[1].close() + except Exception as e: + logger.info(f"Error tearing down pooled agent, ignoring: {e}") + + def __exit__(self, *exc_info: object) -> None: + for key in list(self._entries): + self._close_entry(key)