Introduce docker_fixtures

tests/parametric/test_otel_logs.py

@@ -53,10 +53,10 @@ def find_attributes(proto_object):
 
 
 @pytest.fixture
-def otlp_endpoint_library_env(library_env, endpoint_env, test_agent_container_name, test_agent_otlp_grpc_port):
+def otlp_endpoint_library_env(library_env, endpoint_env, test_agent, test_agent_otlp_grpc_port):
     """Set up a custom endpoint for OTLP logs."""
     prev_value = library_env.get(endpoint_env)
-    library_env[endpoint_env] = f"http://{test_agent_container_name}:{test_agent_otlp_grpc_port}"
+    library_env[endpoint_env] = f"http://{test_agent.container_name}:{test_agent_otlp_grpc_port}"
     yield library_env
     if prev_value is None:
         del library_env[endpoint_env]

tests/parametric/test_otel_metrics.py

@@ -45,7 +45,7 @@
 
 @pytest.fixture
 def otlp_metrics_endpoint_library_env(
-    library_env, endpoint_env, test_agent_otlp_http_port, test_agent_otlp_grpc_port, test_agent_container_name
+    library_env, endpoint_env, test_agent, test_agent_otlp_http_port, test_agent_otlp_grpc_port
 ):
     """Set up a custom endpoint for OTLP metrics."""
     prev_value = library_env.get(endpoint_env)
@@ -59,7 +59,7 @@ def otlp_metrics_endpoint_library_env(
     port = test_agent_otlp_grpc_port if protocol == "grpc" else test_agent_otlp_http_port
     path = "/" if protocol == "grpc" or endpoint_env == "OTEL_EXPORTER_OTLP_ENDPOINT" else "/v1/metrics"
 
-    library_env[endpoint_env] = f"http://{test_agent_container_name}:{port}{path}"
+    library_env[endpoint_env] = f"http://{test_agent.container_name}:{port}{path}"
     yield library_env
     if prev_value is None:
         del library_env[endpoint_env]

utils/_context/_scenarios/parametric.py

@@ -13,13 +13,11 @@
 import pytest
 from _pytest.outcomes import Failed
 from docker.models.containers import Container
-from docker.models.networks import Network
-from retry import retry
 
+from utils.docker_fixtures._test_agent import TestAgentFactory, TestAgentAPI
 from utils._context.component_version import ComponentVersion
-from utils._logger import logger
-
 from utils._context.docker import get_docker_client
+from utils._logger import logger
 from .core import Scenario, scenario_groups
 
 
@@ -56,9 +54,10 @@ class APMLibraryTestServer:
 
 
 class ParametricScenario(Scenario):
-    TEST_AGENT_IMAGE = "ghcr.io/datadog/dd-apm-test-agent/ddapm-test-agent:v1.32.0"
     apm_test_server_definition: APMLibraryTestServer
 
+    _test_agent_factory: TestAgentFactory
+
     class PersistentParametricTestConf(dict):
         """Parametric tests are executed in multiple thread, we need a mechanism to persist
         each parametrized_tests_metadata on a file
@@ -123,13 +122,14 @@ def configure(self, config: pytest.Config):
             "rust": rust_library_factory,
         }[library]
 
+        self._test_agent_factory = TestAgentFactory(self.host_log_folder)
         self.apm_test_server_definition = factory()
 
         if self.is_main_worker:
             # https://github.com/pytest-dev/pytest-xdist/issues/271#issuecomment-826396320
             # we are in the main worker, not in a xdist sub-worker
             self._build_apm_test_server_image(config.option.github_token_file)
-            self._pull_test_agent_image()
+            self._test_agent_factory.pull()
             self._clean_containers()
             self._clean_networks()
 
@@ -161,11 +161,6 @@ def get_warmups(self):
 
         return result
 
-    @retry(delay=10, tries=3)
-    def _pull_test_agent_image(self):
-        logger.stdout("Pulling test agent image...")
-        get_docker_client().images.pull(self.TEST_AGENT_IMAGE)
-
     def _clean_containers(self):
         """Some containers may still exists from previous unfinished sessions"""
 
@@ -258,10 +253,21 @@ def _build_apm_test_server_image(self, github_token_file: str) -> None:
 
             logger.debug("Build tested container finished")
 
-    def create_docker_network(self, test_id: str) -> Network:
+    @contextlib.contextmanager
+    def _get_docker_network(self, test_id: str) -> Generator[str, None, None]:
         docker_network_name = f"{_NETWORK_PREFIX}_{test_id}"
+        network = get_docker_client().networks.create(name=docker_network_name, driver="bridge")
 
-        return get_docker_client().networks.create(name=docker_network_name, driver="bridge")
+        try:
+            yield network.name
+        finally:
+            try:
+                network.remove()
+            except:
+                # It's possible (why?) of having some container not stopped.
+                # If it happens, failing here makes stdout tough to understand.
+                # Let's ignore this, later calls will clean the mess
+                logger.info("Failed to remove network, ignoring the error")
 
     @staticmethod
     def get_host_port(worker_id: str, base_port: int) -> int:
@@ -341,6 +347,28 @@ def get_junit_properties(self) -> dict[str, str]:
 
         return result
 
+    @contextlib.contextmanager
+    def get_test_agent_api(
+        self,
+        worker_id: str,
+        request: pytest.FixtureRequest,
+        test_id: str,
+        container_otlp_http_port: int,
+        container_otlp_grpc_port: int,
+    ) -> Generator[TestAgentAPI, None, None]:
+        with (
+            self._get_docker_network(test_id) as docker_network,
+            self._test_agent_factory.get_test_agent_api(
+                request=request,
+                worker_id=worker_id,
+                docker_network=docker_network,
+                container_name=f"ddapm-test-agent-{test_id}",
+                container_otlp_http_port=container_otlp_http_port,
+                container_otlp_grpc_port=container_otlp_grpc_port,
+            ) as result,
+        ):
+            yield result
+
 
 def _get_base_directory() -> str:
     return str(Path.cwd())

utils/docker_fixtures/_core.py

@@ -0,0 +1,82 @@
+import contextlib
+from collections.abc import Generator
+from pathlib import Path
+from typing import TextIO
+
+from docker.models.containers import Container
+import pytest
+
+from utils._logger import logger
+from utils._context.docker import get_docker_client
+
+
+def get_host_port(worker_id: str, base_port: int) -> int:
+    """Deterministic port allocation for each worker"""
+
+    if worker_id == "master":  # xdist disabled
+        return base_port
+
+    if worker_id.startswith("gw"):
+        return base_port + int(worker_id[2:])
+
+    raise ValueError(f"Unexpected worker_id: {worker_id}")
+
+
+def _compute_volumes(volumes: dict[str, str]) -> dict[str, dict]:
+    """Convert volumes to the format expected by the docker-py API"""
+    fixed_volumes: dict[str, dict] = {}
+    for key, value in volumes.items():
+        # when host path starts with ./, resolve it from cwd()
+        fixed_key = str(Path.cwd().joinpath(key)) if key.startswith("./") else key
+
+        if isinstance(value, dict):
+            fixed_volumes[fixed_key] = value
+        elif isinstance(value, str):
+            fixed_volumes[fixed_key] = {"bind": value, "mode": "rw"}
+        else:
+            raise TypeError(f"Unexpected type for volume {key}: {type(value)}")
+
+    return fixed_volumes
+
+
+@contextlib.contextmanager
+def docker_run(
+    image: str,
+    name: str,
+    env: dict[str, str],
+    volumes: dict[str, str],
+    network: str,
+    ports: dict[str, int],
+    log_file: TextIO,
+    command: list[str] | None = None,
+) -> Generator[Container, None, None]:
+    logger.info(f"Run container {name} from image {image} with ports {ports}")
+
+    try:
+        container: Container = get_docker_client().containers.run(
+            image,
+            name=name,
+            environment=env,
+            volumes=_compute_volumes(volumes),
+            network=network,
+            ports=ports,
+            command=command,
+            detach=True,
+        )
+        logger.debug(f"Container {name} successfully started")
+    except Exception as e:
+        # at this point, even if it failed to start, the container may exists!
+        for container in get_docker_client().containers.list(filters={"name": name}, all=True):
+            container.remove(force=True)
+
+        pytest.fail(f"Failed to run container {name}: {e}")
+
+    try:
+        yield container
+    finally:
+        logger.info(f"Stopping {name}")
+        container.stop(timeout=1)
+        logs = container.logs()
+        log_file.write(logs.decode("utf-8"))
+        log_file.flush()
+        container.remove(force=True)