distro: override otlp exporters user agent headers

For GRPC we leverage the recently added channel_options, for HTTP we
temporarily need to monkey patch the exporter default headers in the
meantime we have an sdk where we can override them.

Author: xrmx

src/elasticotel/distro/__init__.py

@@ -23,6 +23,14 @@
     OTEL_METRICS_EXPORTER,
     OTEL_TRACES_EXPORTER,
 )
+from opentelemetry.exporter.otlp.proto.grpc import _USER_AGENT_HEADER_VALUE
+from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter as GRPCOTLPLogExporter
+from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter as GRPCOTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter as GRPCOTLPSpanExporter
+from opentelemetry.exporter.otlp.proto.http import _OTLP_HTTP_HEADERS
+from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter as HTTPOTLPLogExporter
+from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter as HTTPOTLPMetricExporter
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter as HTTPOTLPSpanExporter
 from opentelemetry.instrumentation.distro import BaseDistro
 from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
 from opentelemetry.instrumentation.system_metrics import (
@@ -44,16 +52,43 @@
 from opentelemetry._opamp.client import OpAMPClient
 from opentelemetry._opamp.proto import opamp_pb2 as opamp_pb2
 
+from elasticotel.distro import version
 from elasticotel.distro.environment_variables import ELASTIC_OTEL_OPAMP_ENDPOINT, ELASTIC_OTEL_SYSTEM_METRICS_ENABLED
 from elasticotel.distro.resource_detectors import get_cloud_resource_detectors
 from elasticotel.distro.config import opamp_handler
 
 
 logger = logging.getLogger(__name__)
 
+EDOT_GRPC_USER_AGENT_HEADER_VALUE = "elastic-otlp-grpc-python/" + version.__version__
+EDOT_HTTP_USER_AGENT_HEADER_VALUE = "elastic-otlp-http-python/" + version.__version__
+
 
 class ElasticOpenTelemetryConfigurator(_OTelSDKConfigurator):
     def _configure(self, **kwargs):
+        # override GRPC and HTTP user agent headers, GRPC works since OTel SDK 1.35.0, HTTP currently requires an hack
+        otlp_grpc_exporter_options = {
+            "channel_options": (
+                ("grpc.primary_user_agent", f"{EDOT_GRPC_USER_AGENT_HEADER_VALUE} {_USER_AGENT_HEADER_VALUE}"),
+            )
+        }
+        otlp_http_exporter_options = {
+            "headers": {
+                **_OTLP_HTTP_HEADERS,
+                "User-Agent": f"{EDOT_HTTP_USER_AGENT_HEADER_VALUE} {_OTLP_HTTP_HEADERS['User-Agent']}",
+            }
+        }
+        kwargs["exporter_args_map"] = {
+            GRPCOTLPLogExporter: otlp_grpc_exporter_options,
+            GRPCOTLPMetricExporter: otlp_grpc_exporter_options,
+            GRPCOTLPSpanExporter: otlp_grpc_exporter_options,
+            HTTPOTLPLogExporter: otlp_http_exporter_options,
+            HTTPOTLPMetricExporter: otlp_http_exporter_options,
+            HTTPOTLPSpanExporter: otlp_http_exporter_options,
+        }
+        # TODO: Remove the following line after rebasing on top of upstream 1.37.0
+        _OTLP_HTTP_HEADERS["User-Agent"] = otlp_http_exporter_options["headers"]["User-Agent"]
+
         super()._configure(**kwargs)
 
         enable_opamp = False

tests/integration/test_integration.py

@@ -22,6 +22,7 @@
     ContainerResourceDetector,
 )
 
+from elasticotel.distro import version
 from .utils import ElasticIntegrationTestCase, OTEL_INSTRUMENTATION_VERSION, ROOT_DIR
 
 
@@ -136,8 +137,7 @@ def test_metrics_with_system_metrics(self):
 
     def test_log_events_are_sent(self):
         def send_event():
-            from opentelemetry._events import Event
-            from opentelemetry._events import get_event_logger
+            from opentelemetry._events import Event, get_event_logger
 
             event = Event(name="test.event", attributes={}, body={"key": "value", "dict": {"nestedkey": "nestedvalue"}})
             event_logger = get_event_logger(__name__)
@@ -150,6 +150,38 @@ def send_event():
         self.assertEqual(log["attributes"]["event.name"], "test.event")
         self.assertEqual(log["body"], {"key": "value", "dict": {"nestedkey": "nestedvalue"}})
 
+    def test_edot_user_agent_is_used_in_otlp_grpc_exporter(self):
+        def test_script():
+            import sqlite3
+
+            from opentelemetry._events import Event, get_event_logger
+
+            connection = sqlite3.connect(":memory:")
+            cursor = connection.cursor()
+            cursor.execute("CREATE TABLE movie(title, year, score)")
+
+            event = Event(name="test.event", attributes={}, body={"key": "value"})
+            event_logger = get_event_logger(__name__)
+            event_logger.emit(event)
+
+        stdout, stderr, returncode = self.run_script(test_script, wrapper_script="opentelemetry-instrument")
+
+        telemetry = self.get_telemetry()
+        (metrics_headers, logs_headers, traces_headers) = (
+            telemetry["metrics_headers"],
+            telemetry["logs_headers"],
+            telemetry["traces_headers"],
+        )
+
+        assert metrics_headers
+        assert traces_headers
+        assert logs_headers
+
+        edot_user_agent = "elastic-otlp-grpc-python/" + version.__version__
+        self.assertIn(edot_user_agent, metrics_headers[0]["user-agent"])
+        self.assertIn(edot_user_agent, traces_headers[0]["user-agent"])
+        self.assertIn(edot_user_agent, logs_headers[0]["user-agent"])
+
 
 @pytest.mark.integration
 class OperatorTestCase(ElasticIntegrationTestCase):

tests/integration/utils.py

@@ -156,6 +156,7 @@ def normalize_kvlist(body) -> dict:
             return dict_values
 
         metrics = []
+        metrics_headers = []
         for request in telemetry["metric_requests"]:
             elems = []
             for proto_elem in request["pbreq"]["resourceMetrics"]:
@@ -176,7 +177,10 @@ def normalize_kvlist(body) -> dict:
             metric = {"resourceMetrics": elems}
             metrics.append(metric)
 
+            metrics_headers.append(request["headers"])
+
         traces = []
+        traces_headers = []
         for request in telemetry["trace_requests"]:
             for resource_span in request["pbreq"]["resourceSpans"]:
                 resource_attributes = normalize_attributes(resource_span["resource"]["attributes"])
@@ -188,8 +192,10 @@ def normalize_kvlist(body) -> dict:
                         span["spanId"] = decode_id(span["spanId"])
                         span["traceId"] = decode_id(span["traceId"])
                         traces.append(span)
+            traces_headers.append(request["headers"])
 
         logs = []
+        logs_headers = []
         for request in telemetry["log_requests"]:
             for resource_log in request["pbreq"]["resourceLogs"]:
                 resource_attributes = normalize_attributes(resource_log["resource"]["attributes"])
@@ -200,11 +206,15 @@ def normalize_kvlist(body) -> dict:
                         log["body"] = normalize_kvlist(log["body"])
                         log["resource"] = resource_attributes
                         logs.append(log)
+            logs_headers.append(request["headers"])
 
         return {
             "logs": logs,
+            "logs_headers": logs_headers,
             "metrics": metrics,
+            "metrics_headers": metrics_headers,
             "traces": traces,
+            "traces_headers": traces_headers,
         }
 
     def run_script(