feat: add ability to optionally disable internal HTTP send and receive spans

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `opentelemetry-instrumentation-kafka-python` Instrument temporary fork, kafka-python-ng
   inside kafka-python's instrumentation
   ([#2537](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2537))
+- `opentelemetry-instrumentation-asgi` Add ability to disable internal HTTP send and receive spans
+  ([#2802](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2802))
 
 ## Breaking changes
 

instrumentation/opentelemetry-instrumentation-asgi/src/opentelemetry/instrumentation/asgi/__init__.py

@@ -481,7 +481,7 @@ def get_default_span_details(scope: dict) -> Tuple[str, dict]:
 
 
 def _collect_target_attribute(
-    scope: typing.Dict[str, typing.Any]
+    scope: typing.Dict[str, typing.Any],
 ) -> typing.Optional[str]:
     """
     Returns the target path as defined by the Semantic Conventions.
@@ -527,6 +527,8 @@ class OpenTelemetryMiddleware:
             the current globally configured one is used.
         meter_provider: The optional meter provider to use. If omitted
             the current globally configured one is used.
+        exclude_receive_span: Optional flag to exclude the http receive span from the trace.
+        exclude_send_span: Optional flag to exclude the http send span from the trace.
     """
 
     # pylint: disable=too-many-branches
@@ -545,6 +547,8 @@ def __init__(
         http_capture_headers_server_request: list[str] | None = None,
         http_capture_headers_server_response: list[str] | None = None,
         http_capture_headers_sanitize_fields: list[str] | None = None,
+        exclude_receive_span: bool = False,
+        exclude_send_span: bool = False,
     ):
         # initialize semantic conventions opt-in if needed
         _OpenTelemetrySemanticConventionStability._initialize()
@@ -651,6 +655,8 @@ def __init__(
             )
             or []
         )
+        self.exclude_receive_span = exclude_receive_span
+        self.exclude_send_span = exclude_send_span
 
     # pylint: disable=too-many-statements
     async def __call__(
@@ -796,6 +802,9 @@ async def __call__(
     # pylint: enable=too-many-branches
 
     def _get_otel_receive(self, server_span_name, scope, receive):
+        if self.exclude_receive_span:
+            return receive
+
         @wraps(receive)
         async def otel_receive():
             with self.tracer.start_as_current_span(
@@ -832,74 +841,82 @@ def _get_otel_send(
         @wraps(send)
         async def otel_send(message: dict[str, Any]):
             nonlocal expecting_trailers
-            with self.tracer.start_as_current_span(
-                " ".join((server_span_name, scope["type"], "send"))
-            ) as send_span:
-                if callable(self.client_response_hook):
-                    self.client_response_hook(send_span, scope, message)
-
-                status_code = None
-                if message["type"] == "http.response.start":
-                    status_code = message["status"]
-                elif message["type"] == "websocket.send":
-                    status_code = 200
-
-                if send_span.is_recording():
-                    if message["type"] == "http.response.start":
-                        expecting_trailers = message.get("trailers", False)
-                    send_span.set_attribute("asgi.event.type", message["type"])
-                    if (
-                        server_span.is_recording()
-                        and server_span.kind == trace.SpanKind.SERVER
-                        and "headers" in message
-                    ):
-                        custom_response_attributes = (
-                            collect_custom_headers_attributes(
-                                message,
-                                self.http_capture_headers_sanitize_fields,
-                                self.http_capture_headers_server_response,
-                                normalise_response_header_name,
-                            )
-                            if self.http_capture_headers_server_response
-                            else {}
+
+            status_code = None
+            if message["type"] == "http.response.start":
+                status_code = message["status"]
+                expecting_trailers = message.get("trailers", False)
+            elif message["type"] == "websocket.send":
+                status_code = 200
+
+            if not self.exclude_send_span:
+                with self.tracer.start_as_current_span(
+                    " ".join((server_span_name, scope["type"], "send"))
+                ) as send_span:
+                    if callable(self.client_response_hook):
+                        self.client_response_hook(send_span, scope, message)
+
+                    if send_span.is_recording():
+                        if message["type"] == "http.response.start":
+                            expecting_trailers = message.get("trailers", False)
+                        send_span.set_attribute(
+                            "asgi.event.type", message["type"]
                         )
-                        if len(custom_response_attributes) > 0:
-                            server_span.set_attributes(
-                                custom_response_attributes
+                        if status_code:
+                            set_status_code(
+                                send_span,
+                                status_code,
+                                None,
+                                self._sem_conv_opt_in_mode,
                             )
-                if status_code:
-                    # We record metrics only once
-                    set_status_code(
-                        server_span,
-                        status_code,
-                        duration_attrs,
-                        self._sem_conv_opt_in_mode,
-                    )
-                    set_status_code(
-                        send_span,
-                        status_code,
-                        None,
-                        self._sem_conv_opt_in_mode,
-                    )
 
-                propagator = get_global_response_propagator()
-                if propagator:
-                    propagator.inject(
-                        message,
-                        context=set_span_in_context(
-                            server_span, trace.context_api.Context()
-                        ),
-                        setter=asgi_setter,
-                    )
+                        if (
+                            server_span.is_recording()
+                            and server_span.kind == trace.SpanKind.SERVER
+                            and "headers" in message
+                        ):
+                            custom_response_attributes = (
+                                collect_custom_headers_attributes(
+                                    message,
+                                    self.http_capture_headers_sanitize_fields,
+                                    self.http_capture_headers_server_response,
+                                    normalise_response_header_name,
+                                )
+                                if self.http_capture_headers_server_response
+                                else {}
+                            )
+                            if len(custom_response_attributes) > 0:
+                                server_span.set_attributes(
+                                    custom_response_attributes
+                                )
+
+            if status_code:
+                set_status_code(
+                    server_span,
+                    status_code,
+                    duration_attrs,
+                    self._sem_conv_opt_in_mode,
+                )
 
-                content_length = asgi_getter.get(message, "content-length")
-                if content_length:
-                    try:
-                        self.content_length_header = int(content_length[0])
-                    except ValueError:
-                        pass
+            propagator = get_global_response_propagator()
+            if propagator:
+                propagator.inject(
+                    message,
+                    context=set_span_in_context(
+                        server_span, trace.context_api.Context()
+                    ),
+                    setter=asgi_setter,
+                )
+
+            content_length = asgi_getter.get(message, "content-length")
+            if content_length:
+                try:
+                    self.content_length_header = int(content_length[0])
+                except ValueError:
+                    pass
+
+            await send(message)
 
-                await send(message)
             # pylint: disable=too-many-boolean-expressions
             if (
                 not expecting_trailers

instrumentation/opentelemetry-instrumentation-asgi/tests/test_asgi_middleware.py

@@ -572,6 +572,27 @@ def test_background_execution(self):
             _SIMULATED_BACKGROUND_TASK_EXECUTION_TIME_S * 10**9,
         )
 
+    def test_exclude_internal_spans(self):
+        """Test that internal spans are excluded from the emitted spans when
+        the `exclude_receive_span` or `exclude_send_span` attributes are set.
+        """
+        app = otel_asgi.OpenTelemetryMiddleware(simple_asgi)
+        self.seed_app(app)
+        cases = [
+            (True, True, ["GET / http receive", "GET / http send"]),
+            (False, True, ["GET / http send"]),
+            (True, False, ["GET / http receive"]),
+            (False, False, []),
+        ]
+        for exclude_receive_span, exclude_send_span, excluded_spans in cases:
+            app.exclude_receive_span = exclude_receive_span
+            app.exclude_send_span = exclude_send_span
+            self.send_default_request()
+            span_list = self.memory_exporter.get_finished_spans()
+            for span in span_list:
+                for excluded_span in excluded_spans:
+                    self.assertNotEqual(span.name, excluded_span)
+
     def test_trailers(self):
         """Test that trailers are emitted as expected and that the server span is ended
         BEFORE the background task is finished."""