Add test section for OTLP Log ingestion in backend

tests/otel_tracing_e2e/test_e2e.py

@@ -168,6 +168,26 @@ def test_main(self):
             return
         validate_log_trace_correlation(otel_log_trace_attrs, trace_agent)
 
+        # The 2nd account has logs and traces sent via the backend OTLP intake endpoint
+        try:
+            log_intake = interfaces.backend.get_logs(
+                query=f"trace_id:{dd_trace_id}",
+                rid=rid,
+                dd_api_key=os.environ["DD_API_KEY_2"],
+                dd_app_key=os.environ["DD_APP_KEY_2"],
+            )
+            otel_log_trace_attrs = validate_log(log_intake, rid, "backend_endpoint")
+            trace_intake = interfaces.backend.assert_otlp_trace_exist(
+                request=self.r,
+                dd_trace_id=dd_trace_id,
+                dd_api_key=os.environ["DD_API_KEY_2"],
+                dd_app_key=os.environ["DD_APP_KEY_2"],
+            )
+        except ValueError:
+            logger.warning("Backend does not provide logs")
+            return
+        validate_log_trace_correlation(otel_log_trace_attrs, trace_intake)
+
         # The 3rd account has logs and traces sent by OTel Collector
         try:
             log_collector = interfaces.backend.get_logs(

utils/_context/_scenarios/__init__.py

@@ -430,7 +430,7 @@ def all_endtoend_scenarios(test_object):
 
     otel_tracing_e2e = OpenTelemetryScenario("OTEL_TRACING_E2E", doc="")
     otel_metric_e2e = OpenTelemetryScenario("OTEL_METRIC_E2E", doc="")
-    otel_log_e2e = OpenTelemetryScenario("OTEL_LOG_E2E", include_intake=False, doc="")
+    otel_log_e2e = OpenTelemetryScenario("OTEL_LOG_E2E", doc="")
 
     library_conf_custom_header_tags = EndToEndScenario(
         "LIBRARY_CONF_CUSTOM_HEADER_TAGS",

utils/build/docker/java_otel/spring-boot-native/src/main/java/com/datadoghq/springbootnative/App.java

@@ -144,6 +144,18 @@ private static SdkLoggerProvider setupLoggerProvider(Resource resource) {
                             .addHeader("dd-otlp-path", "agent")
                             .build());
         }
+        if (isIntakeEnabled()) {
+            logRecordExporters.add(
+                    OtlpHttpLogRecordExporter.builder()
+                            .setEndpoint("http://proxy:8126/api/v2/logs")
+                            .addHeader("dd-protocol", "otlp")
+                            .addHeader("dd-api-key", System.getenv("DD_API_KEY"))
+                            .addHeader("dd-otlp-path", "intake-logs")
+                            .build());
+            logRecordExporters.add(
+                    OtlpJsonLoggingLogRecordExporter.create()
+            );
+        }
         if (isCollectorEnabled()) {
             logRecordExporters.add(
                     OtlpHttpLogRecordExporter.builder()

utils/otel_validators/validator_log.py

@@ -13,11 +13,11 @@ def validate_log(log: dict, rid: str, otel_source: str) -> dict:
         "source:otlp_log_ingestion",
     ]
     assert expected_attributes_tags <= log["attributes"]["tags"]
-    expected_attributes_attributes = {
-        "http": {"request": {"headers": {"user-agent": f"system_tests rid/{rid}"}}, "method": "GET"},
-        "status": "info",
-    }
-    assert expected_attributes_attributes.items() <= log["attributes"]["attributes"].items()
+
+    assert log["attributes"]["attributes"]["http"]["request"]["headers"]["user-agent"] == f"system_tests rid/{rid}"
+    assert log["attributes"]["attributes"]["http"]["method"] == "GET"
+    assert log["attributes"]["attributes"].get("status") == "info" or log["attributes"].get("status") == "info"
+
     return log["attributes"]["attributes"]["otel"]
 
 
@@ -28,4 +28,4 @@ def validate_log_trace_correlation(otel_log_trace_attrs: dict, trace: dict):
         span = item[1]
     assert otel_log_trace_attrs["trace_id"] == span["meta"]["otel.trace_id"]
     assert int(otel_log_trace_attrs["span_id"], 16) == int(span["span_id"])
-    assert otel_log_trace_attrs["severity_number"] == "9"
+    assert str(otel_log_trace_attrs["severity_number"]) == "9"

utils/proxy/core.py

@@ -150,6 +150,10 @@ def request(self, flow: Flow):
                     flow.request.host = "api." + os.environ.get("DD_SITE", "datad0g.com")
                     flow.request.port = 443
                     flow.request.scheme = "https"
+                elif otlp_path == "intake-logs":
+                    flow.request.host = "http-intake.logs." + os.environ.get("DD_SITE", "datad0g.com")
+                    flow.request.port = 443
+                    flow.request.scheme = "https"
                 else:
                     raise Exception(f"Unknown OTLP ingestion path {otlp_path}")
             else: