distro: override otlp exporters user agent headers

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,11 +22,17 @@
     ContainerResourceDetector,
 )
 
-from .utils import ElasticIntegrationTestCase, OTEL_INSTRUMENTATION_VERSION, ROOT_DIR
+from elasticotel.distro import version
+from .utils import (
+    ElasticIntegrationGRPCTestCase,
+    ElasticIntegrationHTTPTestCase,
+    OTEL_INSTRUMENTATION_VERSION,
+    ROOT_DIR,
+)
 
 
 @pytest.mark.integration
-class IntegrationTestCase(ElasticIntegrationTestCase):
+class GRPCIntegrationTestCase(ElasticIntegrationGRPCTestCase):
     @classmethod
     def requirements(cls):
         requirements = super().requirements()
@@ -136,8 +142,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,9 +155,81 @@ 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 HTTPIntegrationTestCase(ElasticIntegrationHTTPTestCase):
+    @classmethod
+    def requirements(cls):
+        requirements = super().requirements()
+        return requirements + [f"opentelemetry-instrumentation-sqlite3=={OTEL_INSTRUMENTATION_VERSION}"]
+
+    def test_edot_user_agent_is_used_in_otlp_http_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-http-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):
+class OperatorTestCase(ElasticIntegrationHTTPTestCase):
     @staticmethod
     def _read_operator_requirements():
         requirements = []

tests/integration/utils.py

@@ -16,10 +16,12 @@
 
 import base64
 import inspect
+import random
 import subprocess
 import os
 import tempfile
 import unittest
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from pathlib import Path
 from typing import Callable, Mapping, Optional
 
@@ -33,7 +35,7 @@
 ROOT_DIR = os.path.abspath(os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
 
 
-class ElasticIntegrationTestCase(unittest.TestCase):
+class ElasticIntegrationGRPCTestCase(unittest.TestCase):
     """This is an experimental reimplementation of OtelTest using unittest
 
     The idea is to do integration testing by creating a separate virtualenv for each TestCase inheriting
@@ -42,7 +44,7 @@ class ElasticIntegrationTestCase(unittest.TestCase):
 
     A basic TestCase would look like:
 
-    class MyTestCase(ElasticIntegrationTestCase):
+    class MyTestCase(ElasticIntegrationGRPCTestCase):
         @classmethod
         def requirements(cls):
             requirements = super().requirements()
@@ -156,6 +158,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 +179,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 +194,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 +208,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(
@@ -264,3 +276,101 @@ def run_script(
                     ex.stderr.decode() if ex.stderr else None,
                     proc.returncode,
                 )
+
+
+class HttpSink(ot.HttpSink):
+    """Backport of teardown fixes, drop if we ever rebase on top of oteltest 0.37.0+"""
+
+    # This code is copyright Pablo Collins under Apache-2.0
+
+    def __init__(self, listener, port=4318, daemon=True):
+        self.httpd = None
+        super().__init__(listener, port, daemon)
+
+    def run_server(self):
+        class Handler(BaseHTTPRequestHandler):
+            def do_POST(this):
+                # /v1/traces
+                content_length = int(this.headers["Content-Length"])
+                post_data = this.rfile.read(content_length)
+
+                otlp_handler_func = self.handlers.get(this.path)
+                if otlp_handler_func:
+                    otlp_handler_func(post_data, {k: v for k, v in this.headers.items()})
+
+                this.send_response(200)
+                this.send_header("Content-type", "text/html")
+                this.end_headers()
+
+                this.wfile.write("OK".encode("utf-8"))
+
+        self.httpd = HTTPServer(("", self.port), Handler)
+        self.httpd.serve_forever()
+
+    def stop(self):
+        if self.httpd:
+            self.httpd.shutdown()
+        super().stop()
+
+
+class ElasticIntegrationHTTPTestCase(ElasticIntegrationGRPCTestCase):
+    """This is an experimental reimplementation of OtelTest using unittest
+
+    The idea is to do integration testing by creating a separate virtualenv for each TestCase inheriting
+    from ElasticIntegrationTestCase, run a script, collect OTLP http/protobuf calls and make the received data
+    available in order to add assertions.
+
+    A basic TestCase would look like:
+
+    class MyTestCase(ElasticIntegrationHTTPTestCase):
+        @classmethod
+        def requirements(cls):
+            requirements = super().requirements()
+            return requirements + [f"my-library"]
+
+        def script(self):
+            import sqlite3
+
+            connection = sqlite3.connect(":memory:")
+            cursor = connection.cursor()
+            cursor.execute("CREATE TABLE movie(title, year, score)")
+
+        def test_one_span_generated(self):
+            stdout, stderr, returncode = self.run_script(self.script, wrapper_script="opentelemetry-instrument")
+
+            telemetry.get_telemetry()
+            assert len(telemetry["traces"], 1)
+
+
+    Each TestCase costs around 10 seconds for settings up the virtualenv so split tests accordingly.
+    """
+
+    def set_http_port(self):
+        self._http_port = int(random.uniform(44318, 45000))
+        return self._http_port
+
+    def get_http_port(self):
+        return self._http_port
+
+    def setUp(self):
+        self.handler = ot.AccumulatingHandler()
+        # pick a new port each time because otherwise the next test will fail to bind to the same port, even with SO_REUSEPORT :(
+        port = self.set_http_port()
+        self.sink = HttpSink(self.handler, port=port)
+        self.sink.start()
+
+    def run_script(
+        self,
+        script_func: Callable[[], None],
+        environment_variables: Optional[Mapping[str, str]] = None,
+        wrapper_script: Optional[str] = None,
+        on_start: Optional[Callable[[], Optional[float]]] = None,
+    ):
+        # Add a sane default so that callers don't need to remember to set that
+        if environment_variables is None:
+            otlp_endpoint = f"http://localhost:{self.get_http_port()}"
+            environment_variables = {
+                "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
+                "OTEL_EXPORTER_OTLP_ENDPOINT": otlp_endpoint,
+            }
+        return super().run_script(script_func, environment_variables, wrapper_script, on_start)