[Tracing] Implement initial OTLP traces weblog tests

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

@@ -235,6 +235,9 @@ jobs:
       env:
         DD_API_KEY: ${{ secrets.DD_API_KEY }}
         DD_APP_KEY: ${{ secrets.DD_APPLICATION_KEY }}
+    - name: Run APM_TRACING_OTLP scenario
+      if: always() && steps.build.outcome == 'success' && contains(inputs.scenarios, '"APM_TRACING_OTLP"')
+      run: ./run.sh APM_TRACING_OTLP
     - name: Run APM_TRACING_EFFICIENT_PAYLOAD scenario
       if: always() && steps.build.outcome == 'success' && contains(inputs.scenarios, '"APM_TRACING_EFFICIENT_PAYLOAD"')
       run: ./run.sh APM_TRACING_EFFICIENT_PAYLOAD

manifests/cpp_nginx.yml

@@ -291,6 +291,7 @@ manifest:
   tests/integrations/test_service_overrides.py::Test_SqlServiceNameSource: irrelevant (Only implemented for Java)
   tests/integrations/test_sql.py::Test_Sql: missing_feature (Endpoint is not implemented on weblog)
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api: irrelevant (library does not implement OpenTelemetry)
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: irrelevant (library does not implement OpenTelemetry)
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/dotnet.yml

@@ -756,6 +756,7 @@ manifest:
   tests/k8s_lib_injection/test_k8s_lib_injection_appsec.py::TestK8sLibInjectionAppsecClusterEnabled: v3.36.0
   tests/k8s_lib_injection/test_k8s_lib_injection_appsec.py::TestK8sLibInjectionAppsecDisabledByDefault: v3.36.0
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api: v3.9.0
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/golang.yml

@@ -989,6 +989,7 @@ manifest:
     - weblog_declaration:
         "*": incomplete_test_app (endpoint not implemented)
         net-http: v1.70.1
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/java.yml

@@ -3474,6 +3474,7 @@ manifest:
         spring-boot-openliberty: v1.58.2+06122213c8  # Modified by easy win activation script
         uds-spring-boot: v1.58.2+06122213c8  # Modified by easy win activation script
         spring-boot-payara: v1.58.2+06122213c8  # Modified by easy win activation script
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/nodejs.yml

@@ -1860,6 +1860,7 @@ manifest:
         fastify: *ref_5_26_0
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api::test_propagation_extract: incomplete_test_app (Node.js extract endpoint doesn't seem to be working.)
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api::test_propagation_inject: incomplete_test_app (Node.js inject endpoint doesn't seem to be working.)
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/php.yml

@@ -620,6 +620,7 @@ manifest:
   tests/k8s_lib_injection/test_k8s_lib_injection_profiling.py::TestK8sLibInjectioProfilingClusterOverride: v1.9.0
   tests/k8s_lib_injection/test_k8s_lib_injection_profiling.py::TestK8sLibInjectioProfilingDisabledByDefault: v1.9.0
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api: incomplete_test_app (endpoint not implemented)
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/python.yml

@@ -1507,6 +1507,7 @@ manifest:
         flask-poc: v2.19.0
         uds-flask: v4.3.1  # Modified by easy win activation script
         uwsgi-poc: v4.3.1  # Modified by easy win activation script
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/ruby.yml

@@ -1241,6 +1241,7 @@ manifest:
         "*": incomplete_test_app (endpoint not implemented)
         rails72: v2.0.0
   tests/otel/test_context_propagation.py::Test_Otel_Context_Propagation_Default_Propagator_Api::test_propagation_extract: incomplete_test_app (Ruby extract seems to fail even though it should be supported)
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

manifests/rust.yml

@@ -47,6 +47,7 @@ manifest:
   tests/integrations/test_mongo.py::Test_Mongo: missing_feature (Endpoint is not implemented on weblog)
   tests/integrations/test_service_overrides.py::Test_SqlServiceNameSource: irrelevant (Only implemented for Java)
   tests/integrations/test_sql.py::Test_Sql: missing_feature (Endpoint is not implemented on weblog)
+  tests/otel/test_tracing_otlp.py::Test_Otel_Tracing_OTLP: missing_feature
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelLogE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelMetricE2E: irrelevant
   tests/otel_tracing_e2e/test_e2e.py::Test_OTelTracingE2E: irrelevant

tests/otel/test_tracing_otlp.py

@@ -0,0 +1,109 @@
+# Unless explicitly stated otherwise all files in this repository are licensed under the the Apache License Version 2.0.
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2024 Datadog, Inc.
+
+import re
+
+from utils import context, features, interfaces, scenarios, weblog
+from utils._context._scenarios.endtoend import EndToEndScenario
+from utils.dd_constants import SpanKind, StatusCode
+
+
+# @scenarios.apm_tracing_e2e_otel
+@features.otel_api
+@scenarios.apm_tracing_otlp
+class Test_Otel_Tracing_OTLP:
+    def setup_single_server_trace(self):
+        # Get the start time of the weblog container in nanoseconds
+        assert isinstance(context.scenario, EndToEndScenario)
+        exit_code, output = context.scenario.weblog_container.execute_command("date -u +%s%N")
+        assert exit_code == 0, f"self.start_time_ns: date -u +%s%N in weblog container failed: {output!r}"
+        stripped = output.strip()
+        assert stripped, f"empty output from date -u +%s%N in weblog container: {output!r}"
+
+        self.start_time_ns = int(stripped)
+        self.req = weblog.get("/")
+
+    def test_single_server_trace(self):
+        """Validates the required elements of the OTLP payload for a single trace"""
+        data = list(interfaces.open_telemetry.get_otel_spans(self.req))
+
+        # Assert that there is only one OTLP request containing the desired server span
+        assert len(data) == 1
+        request, content, span = data[0]
+
+        # Determine if JSON Protobuf Encoding was used for the OTLP request (rather than Binary Protobuf)
+        # We need to assert that we match the OTLP specification, which has some odd encoding rules when using JSON: https://opentelemetry.io/docs/specs/otlp/#json-protobuf-encoding
+        request_headers = {key.lower(): value for key, value in request.get("headers")}
+        is_json = request_headers.get("content-type") == "application/json"
+
+        # Assert that there is only one resource span (i.e. SDK) in the OTLP request
+        resource_spans = content["resourceSpans"]
+        assert resource_spans is not None, f"missing 'resourceSpans' on content: {content}"
+        assert len(resource_spans) == 1, f"expected 1 resource span, got {len(resource_spans)}"
+        resource_span = resource_spans[0]
+
+        attributes = resource_span.get("resource", {}).get("attributes", {})
+
+        # Assert that the resource attributes contain the service-level attributes and tracer-level attributes we expect
+        # TODO: Assert the following attributes: runtime-id, git.commit.sha, git.repository_url
+        assert attributes.get("service.name") == "weblog"
+        assert attributes.get("service.version") == "1.0.0"
+        assert (
+            attributes.get("deployment.environment.name") == "system-tests"
+            or attributes.get("deployment.environment") == "system-tests"
+        )
+        assert attributes.get("telemetry.sdk.name") == "datadog"
+        assert "telemetry.sdk.language" in attributes
+        # assert "telemetry.sdk.version" in attributes
+
+        # Assert that the `traceId` and `spanId` JSON fields are valid case-insensitive hexadecimal strings, not base64-encoded strings as defined in the standard Protobuf JSON Mapping.
+        # See https://opentelemetry.io/docs/specs/otlp/#json-protobuf-encoding
+        # TODO: Assert against trace_id and span_id fields in the protobuf encoding as well
+        if is_json:
+            assert re.match(r"^[0-9a-fA-F]{32}$", span.get("traceId")), (
+                f"traceId is not a valid case-insensitive hexadecimal string, got {span.get('traceId')}"
+            )
+            assert re.match(r"^[0-9a-fA-F]{16}$", span.get("spanId")), (
+                f"spanId is not a valid case-insensitive hexadecimal string, got {span.get('spanId')}"
+            )
+
+        # Assert that the span fields match the expected values
+        span_start_time_ns = int(span["startTimeUnixNano"])
+        span_end_time_ns = int(span["endTimeUnixNano"])
+        assert span_start_time_ns >= self.start_time_ns
+        assert span_end_time_ns >= span_start_time_ns
+
+        assert span["name"]
+        assert span["kind"] == SpanKind.SERVER.value
+        status = span.get("status", {})
+        # An absent or empty status dict both mean STATUS_CODE_UNSET (protobuf default = 0).
+        assert (
+            not status or status.get("code", StatusCode.STATUS_CODE_UNSET.value) == StatusCode.STATUS_CODE_UNSET.value
+        )
+
+        # Assert core span attributes
+        assert span["attributes"] is not None
+        span_attributes = span["attributes"]
+        assert span_attributes.get("service.name") == "weblog" or span_attributes.get("service.name") is None
+        assert span_attributes["resource.name"] == span["name"]
+        assert span_attributes["span.type"] == "web"
+        assert span_attributes["operation.name"] is not None
+
+        # Assert HTTP tags
+        # Convert attributes list to a dictionary, but for now only handle key_value objects with stringValue
+        method = span_attributes.get("http.method") or span_attributes.get("http.request.method")
+        status_code = span_attributes.get("http.status_code") or span_attributes.get("http.response.status_code")
+        assert method == "GET", f"HTTP method is not GET, got {method}"
+        assert status_code is not None
+        assert int(status_code) == 200, f"HTTP status code is not 200, got {int(status_code)}"
+
+    def setup_unsampled_trace(self):
+        self.req = weblog.get("/", headers={"traceparent": "00-11111111111111110000000000000001-0000000000000001-00"})
+
+    def test_unsampled_trace(self):
+        """Validates that the spans from a non-sampled trace are not exported."""
+        data = list(interfaces.open_telemetry.get_otel_spans(self.req))
+
+        # Assert that the span from this test case was not exported
+        assert len(data) == 0, f"Expected no weblog spans in the OTLP trace payload, got {data}"

utils/_context/_scenarios/__init__.py

@@ -579,6 +579,17 @@ class _Scenarios:
         require_api_key=True,
         doc="",
     )
+    apm_tracing_otlp = EndToEndScenario(
+        "APM_TRACING_OTLP",
+        weblog_env={
+            "OTEL_TRACES_EXPORTER": "otlp",
+            "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": f"http://proxy:{ProxyPorts.open_telemetry_weblog}/v1/traces",
+            "OTEL_EXPORTER_OTLP_TRACES_HEADERS": "dd-protocol=otlp,dd-otlp-path=agent",
+        },
+        backend_interface_timeout=5,
+        include_opentelemetry=True,
+        doc="",
+    )
 
     apm_tracing_efficient_payload = EndToEndScenario(
         "APM_TRACING_EFFICIENT_PAYLOAD",

utils/_context/_scenarios/endtoend.py

@@ -204,6 +204,7 @@ def __init__(
         runtime_metrics_enabled: bool = False,
         backend_interface_timeout: int = 0,
         include_buddies: bool = False,
+        include_opentelemetry: bool = False,
         require_api_key: bool = False,
         other_weblog_containers: tuple[type[TestedContainer], ...] = (),
     ) -> None:
@@ -283,6 +284,7 @@ def __init__(
         self.agent_interface_timeout = agent_interface_timeout
         self.backend_interface_timeout = backend_interface_timeout
         self._library_interface_timeout = library_interface_timeout
+        self.include_opentelemetry = include_opentelemetry
 
     def configure(self, config: pytest.Config):
         if self._require_api_key and "DD_API_KEY" not in os.environ and not self.replay:
@@ -299,6 +301,9 @@ def configure(self, config: pytest.Config):
         interfaces.library_stdout.configure(self.host_log_folder, replay=self.replay)
         interfaces.agent_stdout.configure(self.host_log_folder, replay=self.replay)
 
+        if self.include_opentelemetry:
+            interfaces.open_telemetry.configure(self.host_log_folder, replay=self.replay)
+
         for container in self.buddies:
             container.interface.configure(self.host_log_folder, replay=self.replay)
 
@@ -359,8 +364,13 @@ def _get_weblog_system_info(self):
         logger.stdout("")
 
     def _start_interfaces_watchdog(self):
+        open_telemetry_interfaces: list[ProxyBasedInterfaceValidator] = (
+            [interfaces.open_telemetry] if self.include_opentelemetry else []
+        )
         super().start_interfaces_watchdog(
-            [interfaces.library, interfaces.agent] + [container.interface for container in self.buddies]
+            [interfaces.library, interfaces.agent]
+            + [container.interface for container in self.buddies]
+            + open_telemetry_interfaces
         )
 
     def _set_weblog_domain(self):
@@ -420,6 +430,10 @@ def _wait_and_stop_containers(self, *, force_interface_timout_to_zero: bool):
 
             interfaces.backend.load_data_from_logs()
 
+            if self.include_opentelemetry:
+                interfaces.open_telemetry.load_data_from_logs()
+                interfaces.open_telemetry.check_deserialization_errors()
+
         else:
             self._wait_interface(
                 interfaces.library, 0 if force_interface_timout_to_zero else self.library_interface_timeout
@@ -444,6 +458,11 @@ def _wait_and_stop_containers(self, *, force_interface_timout_to_zero: bool):
                 interfaces.backend, 0 if force_interface_timout_to_zero else self.backend_interface_timeout
             )
 
+            if self.include_opentelemetry:
+                self._wait_interface(
+                    interfaces.open_telemetry, 0 if force_interface_timout_to_zero else self.backend_interface_timeout
+                )
+
     def _wait_interface(self, interface: ProxyBasedInterfaceValidator, timeout: int):
         logger.terminal.write_sep("-", f"Wait for {interface} ({timeout}s)")
         logger.terminal.flush()

utils/dd_constants.py

@@ -108,10 +108,18 @@ class SamplingMechanism(IntEnum):
     AI_GUARD = 13
 
 
+# See https://github.com/open-telemetry/opentelemetry-proto/blob/v1.9.0/opentelemetry/proto/trace/v1/trace.proto#L153
 class SpanKind(IntEnum):
     UNSPECIFIED = 0
     INTERNAL = 1
     SERVER = 2
     CLIENT = 3
     PRODUCER = 4
     CONSUMER = 5
+
+
+# See https://github.com/open-telemetry/opentelemetry-proto/blob/v1.9.0/opentelemetry/proto/trace/v1/trace.proto#L316
+class StatusCode(IntEnum):
+    STATUS_CODE_UNSET = 0
+    STATUS_CODE_OK = 1
+    STATUS_CODE_ERROR = 2

utils/interfaces/_open_telemetry.py

@@ -23,18 +23,44 @@ def ingest_file(self, src_path: str):
         return super().ingest_file(src_path)
 
     def get_otel_trace_id(self, request: HttpResponse):
+        # Paths filter intercepted OTLP export requests (weblog → proxy), not weblog or backend URLs.
         paths = ["/api/v0.2/traces", "/v1/traces"]
         rid = request.get_rid()
 
         if rid:
             logger.debug(f"Try to find traces related to request {rid}")
 
         for data in self.get_data(path_filters=paths):
-            for resource_span in data.get("request").get("content").get("resourceSpans"):
-                for scope_span in resource_span.get("scopeSpans"):
+            content = data.get("request").get("content")
+            resource_spans = content.get("resourceSpans") or []
+            for resource_span in resource_spans:
+                scope_spans = resource_span.get("scopeSpans") or []
+                for scope_span in scope_spans:
+                    for span in scope_span.get("spans", []):
+                        attributes = span.get("attributes", {})
+                        request_headers_user_agent_value = attributes.get("http.request.headers.user-agent", "")
+                        user_agent_value = attributes.get("http.useragent", "")
+                        if rid in request_headers_user_agent_value or rid in user_agent_value:
+                            yield span.get("trace_id") or span.get("traceId")
+
+    def get_otel_spans(self, request: HttpResponse):
+        paths = ["/api/v0.2/traces", "/v1/traces"]
+        rid = request.get_rid()
+
+        if rid:
+            logger.debug(f"Try to find traces related to request {rid}")
+
+        for data in self.get_data(path_filters=paths):
+            content = data.get("request").get("content")
+            logger.debug(f"[get_otel_spans] content: {content}")
+            resource_spans = content.get("resourceSpans") or []
+            for resource_span in resource_spans:
+                scope_spans = resource_span.get("scopeSpans")
+                for scope_span in scope_spans:
                     for span in scope_span.get("spans"):
-                        for attribute in span.get("attributes", []):
-                            attr_key = attribute.get("key")
-                            attr_val = attribute.get("value").get("stringValue")
-                            if attr_key == "http.request.headers.user-agent" and rid in attr_val:
-                                yield span.get("traceId")
+                        attributes = span.get("attributes", {})
+                        request_headers_user_agent_value = attributes.get("http.request.headers.user-agent", "")
+                        user_agent_value = attributes.get("http.useragent", "")
+                        if rid in request_headers_user_agent_value or rid in user_agent_value:
+                            yield data.get("request"), content, span
+                            break  # Skip to next span