Disable ASGI send/receive spans by default (#371)

Author: alexmojaki

logfire-api/logfire_api/_internal/exporters/processor_wrapper.pyi

@@ -1,7 +1,7 @@
 from ..constants import ATTRIBUTES_LOG_LEVEL_NUM_KEY as ATTRIBUTES_LOG_LEVEL_NUM_KEY, ATTRIBUTES_MESSAGE_KEY as ATTRIBUTES_MESSAGE_KEY, ATTRIBUTES_MESSAGE_TEMPLATE_KEY as ATTRIBUTES_MESSAGE_TEMPLATE_KEY, ATTRIBUTES_SPAN_TYPE_KEY as ATTRIBUTES_SPAN_TYPE_KEY, LEVEL_NUMBERS as LEVEL_NUMBERS, PENDING_SPAN_NAME_SUFFIX as PENDING_SPAN_NAME_SUFFIX, log_level_attributes as log_level_attributes
 from ..db_statement_summary import message_from_db_statement as message_from_db_statement
 from ..scrubbing import BaseScrubber as BaseScrubber
-from ..utils import ReadableSpanDict as ReadableSpanDict, is_instrumentation_suppressed as is_instrumentation_suppressed, span_to_dict as span_to_dict, truncate_string as truncate_string
+from ..utils import ReadableSpanDict as ReadableSpanDict, is_asgi_send_receive_span_name as is_asgi_send_receive_span_name, is_instrumentation_suppressed as is_instrumentation_suppressed, span_to_dict as span_to_dict, truncate_string as truncate_string
 from .wrapper import WrapperSpanProcessor as WrapperSpanProcessor
 from _typeshed import Incomplete
 from opentelemetry import context

logfire-api/logfire_api/_internal/integrations/asgi.pyi

@@ -0,0 +1,20 @@
+from dataclasses import dataclass
+from logfire import Logfire as Logfire
+from logfire._internal.utils import is_asgi_send_receive_span_name as is_asgi_send_receive_span_name
+from opentelemetry.context import Context
+from opentelemetry.trace import Span, Tracer, TracerProvider
+from typing import Any
+
+def tweak_asgi_spans_tracer_provider(logfire_instance: Logfire, record_send_receive: bool) -> TracerProvider:
+    """If record_send_receive is False, return a TracerProvider that skips spans for ASGI send and receive events."""
+
+@dataclass
+class TweakAsgiTracerProvider(TracerProvider):
+    tracer_provider: TracerProvider
+    def get_tracer(self, *args: Any, **kwargs: Any) -> Tracer: ...
+
+@dataclass
+class TweakAsgiSpansTracer(Tracer):
+    tracer: Tracer
+    def start_span(self, name: str, context: Context | None = None, *args: Any, **kwargs: Any) -> Span: ...
+    start_as_current_span = ...

logfire-api/logfire_api/_internal/integrations/fastapi.pyi

@@ -1,6 +1,7 @@
 from ..main import Logfire as Logfire
 from ..stack_info import StackInfo as StackInfo, get_code_object_info as get_code_object_info
 from ..utils import maybe_capture_server_headers as maybe_capture_server_headers
+from .asgi import tweak_asgi_spans_tracer_provider as tweak_asgi_spans_tracer_provider
 from _typeshed import Incomplete
 from fastapi import FastAPI
 from starlette.requests import Request
@@ -9,7 +10,7 @@ from typing import Any, Awaitable, Callable, ContextManager, Iterable
 
 def find_mounted_apps(app: FastAPI) -> list[FastAPI]:
     """Fetch all sub-apps mounted to a FastAPI app, including nested sub-apps."""
-def instrument_fastapi(logfire_instance: Logfire, app: FastAPI, *, capture_headers: bool = False, request_attributes_mapper: Callable[[Request | WebSocket, dict[str, Any]], dict[str, Any] | None] | None = None, use_opentelemetry_instrumentation: bool = True, excluded_urls: str | Iterable[str] | None = None, **opentelemetry_kwargs: Any) -> ContextManager[None]:
+def instrument_fastapi(logfire_instance: Logfire, app: FastAPI, *, capture_headers: bool = False, request_attributes_mapper: Callable[[Request | WebSocket, dict[str, Any]], dict[str, Any] | None] | None = None, use_opentelemetry_instrumentation: bool = True, excluded_urls: str | Iterable[str] | None = None, record_send_receive: bool = False, **opentelemetry_kwargs: Any) -> ContextManager[None]:
     """Instrument a FastAPI app so that spans and logs are automatically created for each request.
 
     See `Logfire.instrument_fastapi` for more details.

logfire-api/logfire_api/_internal/integrations/starlette.pyi

@@ -1,24 +1,16 @@
+from logfire import Logfire as Logfire
+from logfire._internal.integrations.asgi import tweak_asgi_spans_tracer_provider as tweak_asgi_spans_tracer_provider
 from logfire._internal.utils import maybe_capture_server_headers as maybe_capture_server_headers
-from opentelemetry.trace import Span
+from opentelemetry.instrumentation.asgi.types import ClientRequestHook, ClientResponseHook, ServerRequestHook
 from starlette.applications import Starlette
-from typing import Any
-from typing_extensions import Protocol, TypedDict, Unpack
-
-class ServerRequestHook(Protocol):
-    def __call__(self, span: Span, scope: dict[str, Any]): ...
-
-class ClientRequestHook(Protocol):
-    def __call__(self, span: Span, scope: dict[str, Any]): ...
-
-class ClientResponseHook(Protocol):
-    def __call__(self, span: Span, message: dict[str, Any]): ...
+from typing_extensions import TypedDict, Unpack
 
 class StarletteInstrumentKwargs(TypedDict, total=False):
     server_request_hook: ServerRequestHook | None
     client_request_hook: ClientRequestHook | None
     client_response_hook: ClientResponseHook | None
 
-def instrument_starlette(app: Starlette, *, capture_headers: bool = False, **kwargs: Unpack[StarletteInstrumentKwargs]):
+def instrument_starlette(logfire_instance: Logfire, app: Starlette, *, record_send_receive: bool = False, capture_headers: bool = False, **kwargs: Unpack[StarletteInstrumentKwargs]):
     """Instrument `app` so that spans are automatically created for each request.
 
     See the `Logfire.instrument_starlette` method for details.

logfire-api/logfire_api/_internal/main.pyi

@@ -368,7 +368,7 @@ class Logfire:
                 Otherwise, the first time(s) each function is called, it will be timed but not traced.
                 Only after the function has run for at least `min_duration` will it be traced in subsequent calls.
         """
-    def instrument_fastapi(self, app: FastAPI, *, capture_headers: bool = False, request_attributes_mapper: Callable[[Request | WebSocket, dict[str, Any]], dict[str, Any] | None] | None = None, use_opentelemetry_instrumentation: bool = True, excluded_urls: str | Iterable[str] | None = None, **opentelemetry_kwargs: Any) -> ContextManager[None]:
+    def instrument_fastapi(self, app: FastAPI, *, capture_headers: bool = False, request_attributes_mapper: Callable[[Request | WebSocket, dict[str, Any]], dict[str, Any] | None] | None = None, use_opentelemetry_instrumentation: bool = True, excluded_urls: str | Iterable[str] | None = None, record_send_receive: bool = False, **opentelemetry_kwargs: Any) -> ContextManager[None]:
         """Instrument a FastAPI app so that spans and logs are automatically created for each request.
 
         Args:
@@ -398,6 +398,10 @@ class Logfire:
                 will also instrument the app.
 
                 See [OpenTelemetry FastAPI Instrumentation](https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/fastapi/fastapi.html).
+            record_send_receive: Set to True to allow the OpenTelemetry ASGI to create send/receive spans.
+                These are disabled by default to reduce overhead and the number of spans created,
+                since many can be created for a single request, and they are not often useful.
+                If enabled, they will be set to debug level, meaning they will usually still be hidden in the UI.
             opentelemetry_kwargs: Additional keyword arguments to pass to the OpenTelemetry FastAPI instrumentation.
 
         Returns:
@@ -588,11 +592,16 @@ class Logfire:
         [OpenTelemetry Flask Instrumentation](https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/flask/flask.html)
         library, specifically `FlaskInstrumentor().instrument_app()`, to which it passes `**kwargs`.
         """
-    def instrument_starlette(self, app: Starlette, *, capture_headers: bool = False, **kwargs: Unpack[StarletteInstrumentKwargs]) -> None:
+    def instrument_starlette(self, app: Starlette, *, capture_headers: bool = False, record_send_receive: bool = False, **kwargs: Unpack[StarletteInstrumentKwargs]) -> None:
         """Instrument `app` so that spans are automatically created for each request.
 
         Set `capture_headers` to `True` to capture all request and response headers.
 
+        Set `record_send_receive` to `True` to allow the OpenTelemetry ASGI to create send/receive spans.
+        These are disabled by default to reduce overhead and the number of spans created,
+        since many can be created for a single request, and they are not often useful.
+        If enabled, they will be set to debug level, meaning they will usually still be hidden in the UI.
+
         Uses the
         [OpenTelemetry Starlette Instrumentation](https://opentelemetry-python-contrib.readthedocs.io/en/latest/instrumentation/starlette/starlette.html)
         library, specifically `StarletteInstrumentor.instrument_app()`, to which it passes `**kwargs`.

logfire-api/logfire_api/_internal/utils.pyi

@@ -87,3 +87,4 @@ def suppress_instrumentation() -> Generator[None, None, None]:
 def log_internal_error() -> None: ...
 def handle_internal_errors() -> Generator[None, None, None]: ...
 def maybe_capture_server_headers(capture: bool): ...
+def is_asgi_send_receive_span_name(name: str) -> bool: ...

logfire/_internal/exporters/processor_wrapper.py

@@ -21,7 +21,13 @@
 )
 from ..db_statement_summary import message_from_db_statement
 from ..scrubbing import BaseScrubber
-from ..utils import ReadableSpanDict, is_instrumentation_suppressed, span_to_dict, truncate_string
+from ..utils import (
+    ReadableSpanDict,
+    is_asgi_send_receive_span_name,
+    is_instrumentation_suppressed,
+    span_to_dict,
+    truncate_string,
+)
 from .wrapper import WrapperSpanProcessor
 
 
@@ -143,7 +149,7 @@ def _is_asgi_send_receive_span(name: str, instrumentation_scope: Instrumentation
             'opentelemetry.instrumentation.starlette',
             'opentelemetry.instrumentation.fastapi',
         )
-    ) and (name.endswith((' http send', ' http receive', ' websocket send', ' websocket receive')))
+    ) and is_asgi_send_receive_span_name(name)
 
 
 def _tweak_http_spans(span: ReadableSpanDict):

logfire/_internal/integrations/asgi.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Any
+
+from opentelemetry.context import Context
+from opentelemetry.sdk.trace import Tracer as SDKTracer
+from opentelemetry.trace import NonRecordingSpan, Span, Tracer, TracerProvider
+from opentelemetry.trace.propagation import get_current_span
+
+from logfire._internal.utils import is_asgi_send_receive_span_name
+
+if TYPE_CHECKING:
+    from logfire import Logfire
+
+
+def tweak_asgi_spans_tracer_provider(logfire_instance: Logfire, record_send_receive: bool) -> TracerProvider:
+    """If record_send_receive is False, return a TracerProvider that skips spans for ASGI send and receive events."""
+    tracer_provider = logfire_instance.config.get_tracer_provider()
+    if record_send_receive:
+        return tracer_provider
+    else:
+        return TweakAsgiTracerProvider(tracer_provider)
+
+
+@dataclass
+class TweakAsgiTracerProvider(TracerProvider):
+    tracer_provider: TracerProvider
+
+    def get_tracer(self, *args: Any, **kwargs: Any) -> Tracer:
+        return TweakAsgiSpansTracer(self.tracer_provider.get_tracer(*args, **kwargs))
+
+
+@dataclass
+class TweakAsgiSpansTracer(Tracer):
+    tracer: Tracer
+
+    def start_span(self, name: str, context: Context | None = None, *args: Any, **kwargs: Any) -> Span:
+        if is_asgi_send_receive_span_name(name):
+            # These are the noisy spans we want to skip.
+            # Create a no-op span with the same SpanContext as the current span.
+            # This means that any spans created within will have the current span as their parent,
+            # as if this span didn't exist at all.
+            return NonRecordingSpan(get_current_span(context).get_span_context())
+
+        return self.tracer.start_span(name, context, *args, **kwargs)
+
+    # This means that `with start_as_current_span(...):`
+    # is roughly equivalent to `with use_span(start_span(...)):`
+    start_as_current_span = SDKTracer.start_as_current_span

logfire/_internal/integrations/fastapi.py

@@ -17,6 +17,7 @@
 from ..main import Logfire
 from ..stack_info import StackInfo, get_code_object_info
 from ..utils import maybe_capture_server_headers
+from .asgi import tweak_asgi_spans_tracer_provider
 
 try:
     from opentelemetry.instrumentation.asgi import get_host_port_url_tuple  # type: ignore
@@ -58,6 +59,7 @@ def instrument_fastapi(
     | None = None,
     use_opentelemetry_instrumentation: bool = True,
     excluded_urls: str | Iterable[str] | None = None,
+    record_send_receive: bool = False,
     **opentelemetry_kwargs: Any,
 ) -> ContextManager[None]:
     """Instrument a FastAPI app so that spans and logs are automatically created for each request.
@@ -71,7 +73,12 @@ def instrument_fastapi(
 
     if use_opentelemetry_instrumentation:  # pragma: no branch
         maybe_capture_server_headers(capture_headers)
-        FastAPIInstrumentor.instrument_app(app, excluded_urls=excluded_urls, **opentelemetry_kwargs)  # type: ignore
+        FastAPIInstrumentor.instrument_app(  # type: ignore
+            app,
+            excluded_urls=excluded_urls,
+            tracer_provider=tweak_asgi_spans_tracer_provider(logfire_instance, record_send_receive),
+            **opentelemetry_kwargs,
+        )
 
     registry = patch_fastapi()
     if app in registry:  # pragma: no cover

logfire/_internal/integrations/starlette.py

@@ -1,35 +1,39 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from opentelemetry.instrumentation.starlette import StarletteInstrumentor
 from starlette.applications import Starlette
 
+from logfire import Logfire
+from logfire._internal.integrations.asgi import tweak_asgi_spans_tracer_provider
 from logfire._internal.utils import maybe_capture_server_headers
 
 if TYPE_CHECKING:
-    from opentelemetry.trace import Span
-    from typing_extensions import Protocol, TypedDict, Unpack
-
-    class ServerRequestHook(Protocol):
-        def __call__(self, span: Span, scope: dict[str, Any]): ...
-
-    class ClientRequestHook(Protocol):
-        def __call__(self, span: Span, scope: dict[str, Any]): ...
-
-    class ClientResponseHook(Protocol):
-        def __call__(self, span: Span, message: dict[str, Any]): ...
+    from opentelemetry.instrumentation.asgi.types import ClientRequestHook, ClientResponseHook, ServerRequestHook
+    from typing_extensions import TypedDict, Unpack
 
     class StarletteInstrumentKwargs(TypedDict, total=False):
         server_request_hook: ServerRequestHook | None
         client_request_hook: ClientRequestHook | None
         client_response_hook: ClientResponseHook | None
 
 
-def instrument_starlette(app: Starlette, *, capture_headers: bool = False, **kwargs: Unpack[StarletteInstrumentKwargs]):
+def instrument_starlette(
+    logfire_instance: Logfire,
+    app: Starlette,
+    *,
+    record_send_receive: bool = False,
+    capture_headers: bool = False,
+    **kwargs: Unpack[StarletteInstrumentKwargs],
+):
     """Instrument `app` so that spans are automatically created for each request.
 
     See the `Logfire.instrument_starlette` method for details.
     """
     maybe_capture_server_headers(capture_headers)
-    StarletteInstrumentor().instrument_app(app, **kwargs)  # type: ignore[reportUnknownMemberType]
+    StarletteInstrumentor().instrument_app(  # type: ignore[reportUnknownMemberType]
+        app,
+        tracer_provider=tweak_asgi_spans_tracer_provider(logfire_instance, record_send_receive),
+        **kwargs,
+    )