Add scenario duplicates for dd-trace-java with forced v1 protocol

.github/workflows/run-end-to-end.yml

@@ -101,6 +101,11 @@ on:
         default: false
         required: false
         type: boolean
+      weblog_env:
+        description: "JSON object of environment variables to force on the weblog"
+        default: "{}"
+        required: false
+        type: string
 
 jobs:
   main:
@@ -113,6 +118,7 @@ jobs:
       SYSTEM_TESTS_FORCE_EXECUTE: ${{ inputs.force_execute }}
       SYSTEM_TESTS_DEV_MODE: ${{ inputs._system_tests_dev_mode }}
       SYSTEM_TESTS_WEBLOG:  ${{ inputs.weblog }}
+      SYSTEM_TESTS_WEBLOG_ENV: ${{ inputs.weblog_env }}
     steps:
     - name: Compute ref
       id: compute_ref

.github/workflows/system-tests.yml

@@ -263,6 +263,7 @@ jobs:
       weblog_instance: ${{ matrix.job.weblog_instance }}
       weblog_build_required: ${{ matrix.job.weblog_build_required }}
       scenarios: ${{ toJson(matrix.job.scenarios) }}
+      weblog_env: ${{ toJson(matrix.job.weblog_env) }}
       force_execute: ${{ inputs.force_execute }}
       binaries_artifact: ${{ matrix.job.binaries_artifact }}
       ci_environment: ${{ needs.compute_parameters.outputs.ci_environment }}

tests/test_the_test/test_ci_orchestrator.py

@@ -6,6 +6,8 @@
 from utils._context.weblog_metadata import WeblogMetaData
 from utils._context._scenarios import get_all_scenarios, Scenario
 from utils.scripts.ci_orchestrators.workflow_data import (
+    Job,
+    _duplicate_jobs,
     _get_endtoend_weblogs,
     get_endtoend_definitions,
 )
@@ -141,10 +143,41 @@ def test_otel_collector():
             "weblog": "otel_collector",
             "weblog_build_required": False,
             "weblog_instance": 1,
+            "weblog_env": {},
         }
     ]
 
 
+@scenarios.test_the_test
+def test_duplicate_jobs_for_selected_weblogs():
+    weblog = get_weblog("java", "spring-boot-jetty")
+    other_weblog = get_weblog("java", "spring-boot")
+    jobs = [
+        Job("java", weblog, 1, {"DEFAULT": 1.0, "JETTY_SCENARIO_1": 1.0}, 1.0, build_base_images=False),
+        Job("java", weblog, 2, {"JETTY_SCENARIO_2": 1.0}, 1.0, build_base_images=False),
+        Job("java", weblog, 3, {"JETTY_SCENARIO_3": 1.0}, 1.0, build_base_images=False),
+        Job("java", other_weblog, 1, {"DEFAULT": 1.0, "NON_DEFAULT_SCENARIO": 1.0}, 1.0, build_base_images=False),
+    ]
+
+    result = _duplicate_jobs(
+        jobs,
+        weblog_names=("spring-boot-jetty",),
+        name_suffix="_v1",
+        weblog_env={"DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0"},
+    )
+    v1_jobs = {f"{job.weblog.name} {job.serialize()['weblog_instance']}": job for job in result}
+
+    assert set(v1_jobs) == {
+        "spring-boot-jetty 1_v1",
+        "spring-boot-jetty 2_v1",
+        "spring-boot-jetty 3_v1",
+    }
+    assert v1_jobs["spring-boot-jetty 1_v1"].scenarios == ("DEFAULT", "JETTY_SCENARIO_1")
+    assert v1_jobs["spring-boot-jetty 2_v1"].scenarios == ("JETTY_SCENARIO_2",)
+    assert v1_jobs["spring-boot-jetty 3_v1"].scenarios == ("JETTY_SCENARIO_3",)
+    assert all(job.weblog_env == {"DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0"} for job in v1_jobs.values())
+
+
 @scenarios.test_the_test
 def test_legacy_scenario_matrix():
     has_error = False

tests/test_the_test/test_docker_scenario.py

@@ -4,7 +4,7 @@
 import pytest
 
 from utils import interfaces, scenarios
-from utils._context._scenarios.endtoend import DdTraceEndToEndScenario, DockerScenario
+from utils._context._scenarios.endtoend import DdTraceEndToEndScenario, DockerScenario, _load_environment_overrides
 from utils._context.containers import TestedContainer as _TestedContainer
 
 
@@ -24,6 +24,17 @@ def remove(self):
         pass
 
 
+@scenarios.test_the_test
+def test_load_environment_overrides(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("SYSTEM_TESTS_WEBLOG_ENV", '{"DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0"}')
+
+    assert _load_environment_overrides("SYSTEM_TESTS_WEBLOG_ENV") == {"DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0"}
+
+    monkeypatch.setenv("SYSTEM_TESTS_WEBLOG_ENV", '["not", "an", "object"]')
+    with pytest.raises(ValueError, match="must be a JSON object"):
+        _load_environment_overrides("SYSTEM_TESTS_WEBLOG_ENV")
+
+
 @scenarios.test_the_test
 def test_main():
     events: list[str] = []

utils/_context/_scenarios/__init__.py

@@ -840,9 +840,6 @@ class _Scenarios:
             "DD_TRACE_SAMPLE_RATE": "1.0",
             "DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0",
         },
-        agent_env={
-            "DD_APM_ENABLE_V1_TRACE_ENDPOINT": "true",
-        },
         backend_interface_timeout=5,
         doc="End-to-end testing scenario focused on efficient payload handling and v1 trace format validation",
     )

utils/_context/_scenarios/endtoend.py

@@ -37,6 +37,15 @@
 from .core import Scenario, ScenarioGroup, scenario_groups as all_scenario_groups
 
 
+def _load_environment_overrides(variable_name: str) -> dict[str, str]:
+    value: object = json.loads(os.environ.get(variable_name, "{}"))
+    if not isinstance(value, dict) or not all(
+        isinstance(key, str) and isinstance(item, str) for key, item in value.items()
+    ):
+        raise ValueError(f"{variable_name} must be a JSON object with string keys and values")
+    return value
+
+
 class DockerScenario(Scenario):
     """Scenario that tests docker containers"""
 
@@ -328,6 +337,7 @@ def configure(self, config: pytest.Config):
             pytest.exit("DD_API_KEY is required for this scenario", 1)
 
         self.weblog_infra.configure(config)
+        self.weblog_infra.library_container.environment.update(_load_environment_overrides("SYSTEM_TESTS_WEBLOG_ENV"))
         self._containers += list(self.weblog_infra.get_containers())
 
         self._set_containers_dependancies()

utils/scripts/ci_orchestrators/workflow_data.py

@@ -213,6 +213,8 @@ def __init__(
         build_time: float,
         *,
         build_base_images: bool,
+        name_suffix: str = "",
+        weblog_env: dict[str, str] | None = None,
     ):
         self.library = library
         self.weblog = weblog
@@ -223,6 +225,9 @@ def __init__(
         # as a given weblog can have multiple runner executing its scenarios
         # weblog_instance will be used to differentiate them
         self.weblog_instance = weblog_instance
+        self.name_suffix = name_suffix
+
+        self.weblog_env = weblog_env or {}
 
         # build_time is not directly tight to the job, as another runner will execute it
         # but it's convenient to store this info here, as we'll need it to execute the
@@ -238,8 +243,11 @@ def serialize(self) -> dict:
             "library": self.library,
             "weblog": self.weblog.name,
             "weblog_build_required": self.weblog.require_build,
-            "weblog_instance": self.weblog_instance,
+            "weblog_instance": f"{self.weblog_instance}{self.name_suffix}"
+            if self.name_suffix
+            else self.weblog_instance,
             "scenarios": sorted(self.scenarios),
+            "weblog_env": self.weblog_env,
             "expected_job_time": self.expected_job_time + self.build_time,
             "binaries_artifact": self.weblog.artifact_name,
             "build_weblog_base_image": self.weblog.build_mode == BuildMode.local
@@ -257,6 +265,10 @@ def expected_job_time(self) -> float:
 
     @property
     def sort_key(self) -> tuple:
+        return (self.weblog.name, self.weblog_instance, self.name_suffix)
+
+    @property
+    def identity(self) -> tuple[str, int]:
         return (self.weblog.name, self.weblog_instance)
 
     def get_scenario_time(self, scenario: str) -> float:
@@ -266,6 +278,27 @@ def append_scenario(self, scenario: str, execution_time: float) -> None:
         assert scenario not in self._scenarios_times
         self._scenarios_times[scenario] = execution_time
 
+    def duplicate(
+        self,
+        *,
+        name_suffix: str,
+        scenarios: tuple[str, ...] | None = None,
+        weblog_env: dict[str, str] | None = None,
+    ) -> "Job":
+        selected_scenarios = scenarios or self.scenarios
+        assert set(selected_scenarios) <= set(self.scenarios)
+
+        return Job(
+            library=self.library,
+            weblog=self.weblog,
+            weblog_instance=self.weblog_instance,
+            scenarios_times={scenario: self._scenarios_times[scenario] for scenario in selected_scenarios},
+            build_time=self.build_time,
+            build_base_images=self.build_base_images,
+            name_suffix=name_suffix,
+            weblog_env=weblog_env,
+        )
+
     def split_for_parallel_execution(self, desired_execution_time: float) -> list["Job"]:
         result: list[Job] = []
 
@@ -371,6 +404,19 @@ def get_endtoend_definitions(
     if desired_execution_time > 0:  # 0 or less means that user doesn't want to split jobs
         jobs = _split_jobs_for_parallel_execution(jobs, desired_execution_time, maximum_parallel_jobs)
 
+    # Duplicate selected test jobs to exercise an alternative protocol without changing the original jobs.
+    # This can force v1 while the originals use v0.x, or test a legacy protocol while they use the current one.
+    # Add, remove, or adjust library-specific selections here as protocol coverage evolves.
+    if library == "java" and ci_environment == "prod":
+        jobs.extend(
+            _duplicate_jobs(
+                jobs,
+                weblog_names=("spring-boot-jetty",),
+                name_suffix="_v1",
+                weblog_env={"DD_TRACE_AGENT_PROTOCOL_VERSION": "1.0"},
+            )
+        )
+
     # sort jobs by weblog name and weblog instance
     jobs.sort(key=lambda job: job.sort_key)
 
@@ -427,6 +473,28 @@ def _split_jobs_for_parallel_execution(
     return result
 
 
+def _duplicate_jobs(
+    jobs: list[Job],
+    *,
+    weblog_names: tuple[str, ...],
+    name_suffix: str,
+    weblog_env: dict[str, str] | None = None,
+) -> list[Job]:
+    """Duplicate every job for the selected weblogs."""
+    selected_weblogs = set(weblog_names)
+    return sorted(
+        [
+            job.duplicate(
+                name_suffix=name_suffix,
+                weblog_env=weblog_env,
+            )
+            for job in jobs
+            if job.weblog.name in selected_weblogs
+        ],
+        key=lambda job: job.sort_key,
+    )
+
+
 def _split_scenarios_for_parallel_execution(
     scenario_times: dict[str, float], desired_execution_time: float
 ) -> list[list[str]]: