diff --git a/tests/parametric/conftest.py b/tests/parametric/conftest.py index 5b3cb343844..6dcc2aea53b 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. + 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, 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..6da403fb806 --- /dev/null +++ b/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() 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..c095329f06c --- /dev/null +++ b/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 diff --git a/utils/_context/_scenarios/_docker_fixtures.py b/utils/_context/_scenarios/_docker_fixtures.py index e1b8389fc6d..84163d5760c 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, 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,6 +81,61 @@ 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, @@ -80,8 +143,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 1d4756bf229..6e8e3f454bb 100644 --- a/utils/docker_fixtures/_test_agent.py +++ b/utils/docker_fixtures/_test_agent.py @@ -1,5 +1,5 @@ import base64 -from collections.abc import Generator +from collections.abc import Generator, Callable import contextlib import datetime import gzip @@ -26,6 +26,14 @@ from ._core import 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 = "" @@ -35,6 +43,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 @@ -69,9 +82,9 @@ def pull(self) -> None: logger.stdout(f"Pull test agent image {self.image}...") 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, @@ -79,76 +92,112 @@ def get_test_agent_api( agent_env: dict[str, str], container_otlp_http_port: int, container_otlp_grpc_port: int, - ) -> 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 + agent_port_base: int = 4600, + otlp_http_port_base: int = 4701, + otlp_grpc_port_base: int = 4802, + ) -> "tuple[TestAgentAPI, Callable[[], None]]": 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) - - with ( - open(log_path, "w+", encoding="utf-8") as log_file, - docker_run( - image=self.image, - name=container_name, - env=env, - volumes={ - "./snapshots": "/snapshots", - "./tests/integration_frameworks/utils/vcr-cassettes": "/vcr-cassettes", - }, - ports={ - f"{container_port}/tcp": host_port, - f"{container_otlp_http_port}/tcp": otlp_http_host_port, - f"{container_otlp_grpc_port}/tcp": otlp_grpc_host_port, - }, - log_file=log_file, - network=docker_network, - ), - ): - client = TestAgentAPI( - container_name, - container_port, - 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 - expected_version = agent_env.get("TEST_AGENT_VERSION", "test") - for _ in range(100): - try: - resp = client.info() - except Exception as e: - logger.debug(f"Wait for 0.1s for the test agent to be ready {e}") - 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) - - logger.info("Test agent is ready") - break + log_file = open(log_path, "w+", encoding="utf-8") # noqa: SIM115 + logger.stdout(f"REUSE-POC agent container created: {container_name}") + + cm = docker_run( + image=self.image, + name=container_name, + env=env, + volumes={ + "./snapshots": "/snapshots", + "./tests/integration_frameworks/utils/vcr-cassettes": "/vcr-cassettes", + }, + ports={ + "8126/tcp": host_port, + f"{container_otlp_http_port}/tcp": otlp_http_host_port, + f"{container_otlp_grpc_port}/tcp": otlp_grpc_host_port, + }, + log_file=log_file, + network=docker_network, + ) + try: + cm.__enter__() + except BaseException: + log_file.close() + raise + + client = TestAgentAPI( + container_name, + 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) # the trace agent takes ~200ms to start + expected_version = agent_env.get("TEST_AGENT_VERSION", "test") + for _ in range(100): + try: + resp = client.info() + except Exception as e: + logger.debug(f"Wait for 0.1s for the test agent to be ready {e}") + time.sleep(0.1) 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 resp["version"] != expected_version: + pytest.fail( + f"Agent version {resp['version']} is running instead of the test agent.", + pytrace=False, + ) + logger.info("Test agent is ready") + break + else: + try: + cm.__exit__(None, None, None) + finally: + log_file.close() + pytest.fail(f"Could not connect to test agent, check the log file {log_path}.", pytrace=False) + + def _stop() -> None: + try: + cm.__exit__(None, None, None) + finally: + log_file.close() + return client, _stop + + @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]: + client, stop = 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, + ) + try: # 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" @@ -161,8 +210,10 @@ def get_test_agent_api( yield client else: yield client - - request.node.add_report_section("teardown", "Test Agent Output", f"Log file:\n./{log_path}") + 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}") + stop() class TestAgentAPI: @@ -183,6 +234,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 @@ -190,9 +242,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: @@ -393,9 +451,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..5d37ed150ae --- /dev/null +++ b/utils/docker_fixtures/_test_agent_pool.py @@ -0,0 +1,62 @@ +from collections.abc import Callable +from dataclasses import dataclass +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())) + + +@dataclass +class AgentLease: + """One running test-agent (+ its network) owned by the pool. + + `api` is a TestAgentAPI (duck-typed here so this module stays Docker-free). + `stop` tears the lease down (container + network); it must be idempotent-safe + to call exactly once at shutdown. + """ + + api: Any + stop: Callable[[], None] + + +class WorkerAgentPool: + """Per-worker cache of test-agents keyed by agent_env. + + First request for a given env creates a lease via `creator`; later requests + for the same env reuse it after `api.clear()` (reset state) and + `api.rebind_request()` (re-point per-test logging at the current test). + """ + + def __init__(self, creator: Callable[[Any, dict[str, str]], AgentLease]) -> None: + self._creator = creator + self._leases: dict[tuple, AgentLease] = {} + + # request/return are duck-typed (a real FixtureRequest + TestAgentAPI in prod, fakes in + # tests); Any keeps this module Docker-free. + def acquire(self, request: Any, agent_env: dict[str, str]) -> Any: # noqa: ANN401 + key = agent_env_key(agent_env) + lease = self._leases.get(key) + if lease is None: + lease = self._creator(request, agent_env) + self._leases[key] = lease + else: + lease.api.rebind_request(request) + # Always hand back a fully reset agent (covers the first acquire too): drop + # recorded requests, then restore the served remote-config to a fresh-agent + # state. The RC reset is done here, not in clear(), so mid-test clear=True + # helpers don't wipe a test's active config. + lease.api.clear() + lease.api.reset_remote_config() + return lease.api + + def shutdown(self) -> None: + for lease in self._leases.values(): + try: + lease.stop() + except Exception as e: + logger.info(f"Error stopping pooled agent lease, ignoring: {e}") + self._leases.clear()