test(otel): observe the SDK's default OTLP protocol instead of hardcoding it

tests/parametric/otel_env_vars/conftest.py

@@ -1,18 +1,30 @@
 from collections.abc import Callable, Generator
 from contextlib import AbstractContextManager, contextmanager
+import time
 
 from tests.parametric.conftest import APMLibrary
+from tests.parametric.utils import generate_default_counter_data_point
 from utils import pytest, scenarios
 from utils.docker_fixtures import TestAgentAPI, new_test_id
+from utils.docker_fixtures._test_agent import AgentRequest
+from utils.docker_fixtures.parametric import LogLevel
 
 
 APMLibraryFactory = Callable[[dict[str, str | None]], AbstractContextManager[APMLibrary]]
+DefaultProtocol = Callable[[str], str]
 
+CONTENT_TYPE_PROTOCOLS = {"application/x-protobuf": "http/protobuf", "application/json": "http/json"}
 
-# This fixture serves two small use cases: the SDK-default comparison in
-# OTEL_PROPAGATORS and OTEL_LOG_LEVEL, and the control runs in
-# OTEL_SDK_DISABLED. Do not use it outside these tests; contact the
-# apm-shared-testing team if another use case needs it.
+# SDK defaults cannot change during a session, so each signal is observed once
+# per worker. None records that the SDK exported nothing with the protocol unset.
+_default_protocols: dict[str, str | None] = {}
+
+
+# This fixture serves three small use cases: the SDK-default comparison in
+# OTEL_PROPAGATORS and OTEL_LOG_LEVEL, the control runs in OTEL_SDK_DISABLED,
+# and the default observation for OTEL_EXPORTER_OTLP_*_PROTOCOL. Do not use it
+# outside these tests; contact the apm-shared-testing team if another use case
+# needs it.
 @pytest.fixture
 def test_library_factory(
     worker_id: str,
@@ -40,3 +52,88 @@ def create(library_env: dict[str, str | None]) -> Generator[APMLibrary, None, No
             yield library
 
     return create
+
+
+def _signal_env(signal: str) -> dict[str, str | None]:
+    if signal == "traces":
+        # Datadog names its trace switch in the singular, traces are only
+        # exported over OTLP when it is selected as the exporter, and OTLP trace
+        # export always runs with OTel semantics.
+        return {
+            "DD_TRACE_OTEL_ENABLED": "true",
+            "DD_TRACE_OTEL_SEMANTICS_ENABLED": "true",
+            "OTEL_TRACES_EXPORTER": "otlp",
+        }
+    return {f"DD_{signal.upper()}_OTEL_ENABLED": "true"}
+
+
+def _export(library: APMLibrary, signal: str) -> None:
+    if signal == "logs":
+        library.create_logger("default-protocol", LogLevel.INFO)
+        library.write_log("default-protocol", LogLevel.INFO, "default-protocol")
+    elif signal == "traces":
+        with library.dd_start_span("default-protocol"):
+            pass
+        library.dd_flush()
+    else:
+        generate_default_counter_data_point(library, "default_protocol_counter")
+
+
+def _export_requests(test_agent: TestAgentAPI, signal: str) -> list[AgentRequest]:
+    deadline = time.monotonic() + 5
+    while True:
+        requests = [request for request in test_agent.requests() if request["url"].endswith(f"/v1/{signal}")]
+        if requests or time.monotonic() > deadline:
+            return requests
+        time.sleep(0.1)
+
+
+@pytest.fixture
+def default_protocol(
+    test_agent: TestAgentAPI,
+    test_agent_otlp_http_port: int,
+    test_agent_otlp_grpc_port: int,
+    test_library_factory: APMLibraryFactory,
+) -> DefaultProtocol:
+    """The protocol this SDK uses for a signal when no protocol variable is set.
+
+    OTel permits retaining a historical gRPC default, and SDKs migrate between
+    defaults, so the default is observed from an unset control run instead of
+    being hardcoded. The returned callable observes lazily, so tests that never
+    need the default do not depend on it.
+    """
+
+    def observe(signal: str) -> str | None:
+        env: dict[str, str | None] = {
+            **_signal_env(signal),
+            "OTEL_EXPORTER_OTLP_PROTOCOL": None,
+            f"OTEL_EXPORTER_OTLP_{signal.upper()}_PROTOCOL": None,
+        }
+        # A gRPC exporter cannot deliver to the HTTP listener, so a request there
+        # proves an HTTP transport and its content type identifies the encoding.
+        endpoint = f"http://{test_agent.container_name}:{test_agent_otlp_http_port}"
+        with test_library_factory({**env, "OTEL_EXPORTER_OTLP_ENDPOINT": endpoint}) as library:
+            _export(library, signal)
+        if requests := _export_requests(test_agent, signal):
+            headers = {name.lower(): value for name, value in requests[0]["headers"].items()}
+            content_type = headers.get("content-type", "").split(";")[0]
+            assert content_type in CONTENT_TYPE_PROTOCOLS, headers
+            return CONTENT_TYPE_PROTOCOLS[content_type]
+
+        # The gRPC listener forwards decoded payloads to the HTTP listener, so a
+        # request recorded after this run proves delivery over gRPC.
+        endpoint = f"http://{test_agent.container_name}:{test_agent_otlp_grpc_port}"
+        with test_library_factory({**env, "OTEL_EXPORTER_OTLP_ENDPOINT": endpoint}) as library:
+            _export(library, signal)
+        return "grpc" if _export_requests(test_agent, signal) else None
+
+    def get(signal: str) -> str:
+        if signal not in _default_protocols:
+            _default_protocols[signal] = observe(signal)
+            # The control runs must not leak payloads into the tested process.
+            test_agent.clear()
+        protocol = _default_protocols[signal]
+        assert protocol is not None, f"No OTLP {signal} export over HTTP or gRPC with the protocol unset"
+        return protocol
+
+    return get

tests/parametric/otel_env_vars/test_otel_exporter_otlp_logs_protocol.py

@@ -12,40 +12,52 @@
 from utils.docker_fixtures.parametric import LogLevel
 from tests.parametric.conftest import APMLibrary
 from tests.parametric.otel_env_vars import utils as otel_env_var_utils
+from tests.parametric.otel_env_vars.conftest import DefaultProtocol
 from tests.parametric.utils import find_log_record
 
 VARIABLE = "OTEL_EXPORTER_OTLP_LOGS_PROTOCOL"
 
 
+# default_protocol is the fixture from otel_env_vars/conftest.py that observes the SDK's default.
+@pytest.fixture
+def expected_protocol(
+    generic_protocol: otel_env_var_utils.ProtocolInput,
+    protocol: otel_env_var_utils.ProtocolInput,
+    signal: str,
+    default_protocol: DefaultProtocol,
+) -> str:
+    return otel_env_var_utils.expected_protocol(generic_protocol, protocol, signal, default_protocol)
+
+
 # The test matrix supplies the protocol value to exercise protocol selection.
 # Default cases leave it unset; the endpoint selects the expected HTTP/gRPC
 # listener so delivery proves the transport, including SDKs that default to gRPC.
 # Use the generic endpoint so each SDK derives its HTTP or gRPC signal path.
 @pytest.fixture
 def library_env(
-    generic_protocol: str | None,
-    protocol: str | None,
+    generic_protocol: otel_env_var_utils.ProtocolInput,
+    protocol: otel_env_var_utils.ProtocolInput,
     signal: str,
+    default_protocol: DefaultProtocol,
+    expected_protocol: str,
     test_agent: TestAgentAPI,
     test_agent_otlp_http_port: int,
     test_agent_otlp_grpc_port: int,
 ) -> dict[str, str | None]:
-    expected_protocol = otel_env_var_utils.expected_protocol(generic_protocol, protocol, signal)
     port = test_agent_otlp_grpc_port if expected_protocol == "grpc" else test_agent_otlp_http_port
     env: dict[str, str | None] = {
         f"DD_{signal.upper()}_OTEL_ENABLED": "true",
         "OTEL_EXPORTER_OTLP_ENDPOINT": f"http://{test_agent.container_name}:{port}",
-        VARIABLE: protocol,
+        VARIABLE: otel_env_var_utils.resolve_protocol(protocol, signal, default_protocol),
     }
     if generic_protocol is not None:
-        env["OTEL_EXPORTER_OTLP_PROTOCOL"] = generic_protocol
+        env["OTEL_EXPORTER_OTLP_PROTOCOL"] = otel_env_var_utils.resolve_protocol(
+            generic_protocol, signal, default_protocol
+        )
     return env
 
 
-def _assert_export(
-    test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, protocol: str | None, generic_protocol: str | None
-) -> None:
-    expected_protocol = otel_env_var_utils.expected_protocol(generic_protocol, protocol, signal)
+def _assert_export(test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str) -> None:
     with test_library as library:
         library.create_logger("protocol-test", LogLevel.INFO)
         library.write_log("protocol-test", LogLevel.INFO, "selected-protocol")
@@ -78,131 +90,88 @@ class Test_OTEL_EXPORTER_OTLP_LOGS_PROTOCOL:
         ("protocol", "generic_protocol"), [pytest.param("http/protobuf", None, id="http-protobuf")]
     )
     def test_http_protobuf(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param("grpc", None, id="grpc")])
     def test_grpc(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param("http/json", None, id="http-json")])
     def test_http_json(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param(None, None, id="unset")])
     def test_default(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        assert otel_env_var_utils.default_protocol(signal) in ("http/protobuf", "grpc")
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        # OTel recommends HTTP/protobuf and permits retaining a historical gRPC default.
+        assert expected_protocol in ("http/protobuf", "grpc")
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param("", None, id="empty")])
     def test_empty(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param("unsupported", None, id="invalid")])
     def test_invalid(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(
         ("protocol", "generic_protocol"),
-        [pytest.param(otel_env_var_utils.non_default_protocol("logs"), None, id="uppercase")],
+        [
+            pytest.param(
+                lambda signal, default: otel_env_var_utils.sdk_non_default(signal, default).upper(),
+                None,
+                id="uppercase",
+            )
+        ],
     )
     def test_case_insensitive(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(
         ("protocol", "generic_protocol"),
         [
             pytest.param(
-                otel_env_var_utils.non_default_protocol("logs").lower(),
-                otel_env_var_utils.default_protocol("logs"),
+                otel_env_var_utils.sdk_non_default,
+                otel_env_var_utils.sdk_default,
                 id="signal-nondefault-generic-default",
             )
         ],
     )
     def test_signal_precedence(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(
         ("protocol", "generic_protocol"),
         [
-            pytest.param(None, otel_env_var_utils.non_default_protocol("logs").lower(), id="unset-generic-nondefault"),
-            pytest.param("", otel_env_var_utils.non_default_protocol("logs").lower(), id="empty-generic-nondefault"),
+            pytest.param(None, otel_env_var_utils.sdk_non_default, id="unset-generic-nondefault"),
+            pytest.param("", otel_env_var_utils.sdk_non_default, id="empty-generic-nondefault"),
         ],
     )
     def test_generic_fallback(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        _assert_export(test_library, test_agent, signal, expected_protocol)
 
     @pytest.mark.parametrize(("protocol", "generic_protocol"), [pytest.param(None, None, id="unset")])
     def test_documented_json_default(
-        self,
-        test_library: APMLibrary,
-        test_agent: TestAgentAPI,
-        signal: str,
-        protocol: str | None,
-        generic_protocol: str | None,
+        self, test_library: APMLibrary, test_agent: TestAgentAPI, signal: str, expected_protocol: str
     ) -> None:
         """Go documents HTTP/JSON as its logs default, independently of the OTel recommendation."""
-        assert otel_env_var_utils.default_protocol(signal) == "http/json"
-        _assert_export(test_library, test_agent, signal, protocol, generic_protocol)
+        assert expected_protocol == "http/json"
+        _assert_export(test_library, test_agent, signal, expected_protocol)