# Copyright The OpenTelemetry Authors
# SPDX-License-Identifier: Apache-2.0
"""
Usage
-----
.. code-block:: python
import fastapi
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
app = fastapi.FastAPI()
@app.get("/foobar")
async def foobar():
return {"message": "hello world"}
FastAPIInstrumentor.instrument_app(app)
Configuration
-------------
Exclude lists
*************
To exclude certain URLs from tracking, set the environment variable ``OTEL_PYTHON_FASTAPI_EXCLUDED_URLS``
(or ``OTEL_PYTHON_EXCLUDED_URLS`` to cover all instrumentations) to a string of comma delimited regexes that match the
URLs.
For example,
::
export OTEL_PYTHON_FASTAPI_EXCLUDED_URLS="client/.*/info,healthcheck"
will exclude requests such as ``https://site/client/123/info`` and ``https://site/xyz/healthcheck``.
You can also pass comma delimited regexes directly to the ``instrument_app`` method:
.. code-block:: python
FastAPIInstrumentor.instrument_app(app, excluded_urls="client/.*/info,healthcheck")
Request/Response hooks
**********************
This instrumentation supports request and response hooks. These are functions that get called
right after a span is created for a request and right before the span is finished for the response.
- The server request hook is passed a server span and ASGI scope object for every incoming request.
- The client request hook is called with the internal span, and ASGI scope and event when the method ``receive`` is called.
- The client response hook is called with the internal span, and ASGI scope and event when the method ``send`` is called.
.. code-block:: python
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.trace import Span
from typing import Any
def server_request_hook(span: Span, scope: dict[str, Any]):
if span and span.is_recording():
span.set_attribute("custom_user_attribute_from_request_hook", "some-value")
def client_request_hook(span: Span, scope: dict[str, Any], message: dict[str, Any]):
if span and span.is_recording():
span.set_attribute("custom_user_attribute_from_client_request_hook", "some-value")
def client_response_hook(span: Span, scope: dict[str, Any], message: dict[str, Any]):
if span and span.is_recording():
span.set_attribute("custom_user_attribute_from_response_hook", "some-value")
FastAPIInstrumentor().instrument(server_request_hook=server_request_hook, client_request_hook=client_request_hook, client_response_hook=client_response_hook)
Capture HTTP request and response headers
*****************************************
You can configure the agent to capture specified HTTP headers as span attributes, according to the
`semantic conventions <https://github.com/open-telemetry/semantic-conventions/blob/main/docs/http/http-spans.md#http-server-span>`_.
Request headers
***************
To capture HTTP request headers as span attributes, set the environment variable
``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST`` to a comma delimited list of HTTP header names,
or pass the ``http_capture_headers_server_request`` keyword argument to the ``instrument_app`` method.
For example using the environment variable,
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST="content-type,custom_request_header"
will extract ``content-type`` and ``custom_request_header`` from the request headers and add them as span attributes.
Request header names in FastAPI are case-insensitive. So, giving the header name as ``CUStom-Header`` in the environment
variable will capture the header named ``custom-header``.
Regular expressions may also be used to match multiple headers that correspond to the given pattern. For example:
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST="Accept.*,X-.*"
Would match all request headers that start with ``Accept`` and ``X-``.
To capture all request headers, set ``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST`` to ``".*"``.
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_REQUEST=".*"
The name of the added span attribute will follow the format ``http.request.header.<header_name>`` where ``<header_name>``
is the normalized HTTP header name (lowercase, with ``-`` replaced by ``_``). The value of the attribute will be a
single item list containing all the header values.
For example:
``http.request.header.custom_request_header = ["<value1>", "<value2>"]``
Response headers
****************
To capture HTTP response headers as span attributes, set the environment variable
``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE`` to a comma delimited list of HTTP header names,
or pass the ``http_capture_headers_server_response`` keyword argument to the ``instrument_app`` method.
For example using the environment variable,
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE="content-type,custom_response_header"
will extract ``content-type`` and ``custom_response_header`` from the response headers and add them as span attributes.
Response header names in FastAPI are case-insensitive. So, giving the header name as ``CUStom-Header`` in the environment
variable will capture the header named ``custom-header``.
Regular expressions may also be used to match multiple headers that correspond to the given pattern. For example:
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE="Content.*,X-.*"
Would match all response headers that start with ``Content`` and ``X-``.
To capture all response headers, set ``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE`` to ``".*"``.
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SERVER_RESPONSE=".*"
The name of the added span attribute will follow the format ``http.response.header.<header_name>`` where ``<header_name>``
is the normalized HTTP header name (lowercase, with ``-`` replaced by ``_``). The value of the attribute will be a
list containing the header values.
For example:
``http.response.header.custom_response_header = ["<value1>", "<value2>"]``
Sanitizing headers
******************
In order to prevent storing sensitive data such as personally identifiable information (PII), session keys, passwords,
etc, set the environment variable ``OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS``
to a comma delimited list of HTTP header names to be sanitized, or pass the ``http_capture_headers_sanitize_fields``
keyword argument to the ``instrument_app`` method.
Regexes may be used, and all header names will be matched in a case-insensitive manner.
For example using the environment variable,
::
export OTEL_INSTRUMENTATION_HTTP_CAPTURE_HEADERS_SANITIZE_FIELDS=".*session.*,set-cookie"
will replace the value of headers such as ``session-id`` and ``set-cookie`` with ``[REDACTED]`` in the span.
Note:
The environment variable names used to capture HTTP headers are still experimental, and thus are subject to change.
API
---
"""
from __future__ import annotations
import functools
import logging
import types
from typing import Any, Collection, Literal
from weakref import WeakSet as _WeakSet
import fastapi
from starlette.applications import Starlette
from starlette.background import BackgroundTask
from starlette.middleware.errors import ServerErrorMiddleware
from starlette.routing import Match, Route
from starlette.types import ASGIApp, Receive, Scope, Send
from opentelemetry.instrumentation._semconv import (
_get_schema_url,
_OpenTelemetrySemanticConventionStability,
_OpenTelemetryStabilitySignalType,
_StabilityMode,
)
from opentelemetry.instrumentation.asgi import OpenTelemetryMiddleware
from opentelemetry.instrumentation.asgi.types import (
ClientRequestHook,
ClientResponseHook,
ServerRequestHook,
)
from opentelemetry.instrumentation.fastapi.package import _instruments
from opentelemetry.instrumentation.fastapi.version import __version__
from opentelemetry.instrumentation.instrumentor import BaseInstrumentor
from opentelemetry.metrics import MeterProvider, get_meter
from opentelemetry.semconv.attributes.http_attributes import HTTP_ROUTE
from opentelemetry.trace import TracerProvider, get_current_span, get_tracer
from opentelemetry.trace.status import Status, StatusCode
from opentelemetry.util.http import (
get_excluded_urls,
parse_excluded_urls,
sanitize_method,
)
_excluded_urls_from_env = get_excluded_urls("FASTAPI")
_logger = logging.getLogger(__name__)
[docs]class FastAPIInstrumentor(BaseInstrumentor):
"""An instrumentor for FastAPI
See `BaseInstrumentor`
"""
_original_fastapi = None
[docs] @staticmethod
def instrument_app(
app: fastapi.FastAPI,
server_request_hook: ServerRequestHook = None,
client_request_hook: ClientRequestHook = None,
client_response_hook: ClientResponseHook = None,
tracer_provider: TracerProvider | None = None,
meter_provider: MeterProvider | None = None,
excluded_urls: str | None = None,
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_spans: list[Literal["receive", "send"]] | None = None,
): # pylint: disable=too-many-locals
"""Instrument an uninstrumented FastAPI application.
Args:
app: The fastapi ASGI application callable to forward requests to.
server_request_hook: Optional callback which is called with the server span and ASGI
scope object for every incoming request.
client_request_hook: Optional callback which is called with the internal span, and ASGI
scope and event which are sent as dictionaries for when the method receive is called.
client_response_hook: Optional callback which is called with the internal span, and ASGI
scope and event which are sent as dictionaries for when the method send is called.
tracer_provider: The optional tracer provider to use. If omitted
the current globally configured one is used.
meter_provider: The optional meter provider to use. If omitted
the current globally configured one is used.
excluded_urls: Optional comma delimited string of regexes to match URLs that should not be traced.
http_capture_headers_server_request: Optional list of HTTP headers to capture from the request.
http_capture_headers_server_response: Optional list of HTTP headers to capture from the response.
http_capture_headers_sanitize_fields: Optional list of HTTP headers to sanitize.
exclude_spans: Optionally exclude HTTP `send` and/or `receive` spans from the trace.
"""
if not hasattr(app, "_is_instrumented_by_opentelemetry"):
app._is_instrumented_by_opentelemetry = False
if not getattr(app, "_is_instrumented_by_opentelemetry", False):
# initialize semantic conventions opt-in if needed
_OpenTelemetrySemanticConventionStability._initialize()
sem_conv_opt_in_mode = _OpenTelemetrySemanticConventionStability._get_opentelemetry_stability_opt_in_mode(
_OpenTelemetryStabilitySignalType.HTTP,
)
if excluded_urls is None:
excluded_urls = _excluded_urls_from_env
else:
excluded_urls = parse_excluded_urls(excluded_urls)
tracer = get_tracer(
__name__,
__version__,
tracer_provider,
schema_url=_get_schema_url(sem_conv_opt_in_mode),
)
meter = get_meter(
__name__,
__version__,
meter_provider,
schema_url=_get_schema_url(sem_conv_opt_in_mode),
)
def build_middleware_stack(self: Starlette) -> ASGIApp:
# Define an additional middleware for exception handling
# Normally, `opentelemetry.trace.use_span` covers the recording of
# exceptions into the active span, but `OpenTelemetryMiddleware`
# ends the span too early before the exception can be recorded.
class ExceptionHandlerMiddleware:
def __init__(self, app):
self.app = app
async def __call__(
self, scope: Scope, receive: Receive, send: Send
) -> None:
try:
await self.app(scope, receive, send)
except Exception as exc: # pylint: disable=broad-exception-caught
span = get_current_span()
if span.is_recording():
span.record_exception(exc)
span.set_status(
Status(
status_code=StatusCode.ERROR,
description=f"{type(exc).__name__}: {exc}",
)
)
raise
# For every possible use case of error handling, exception
# handling, trace availability in exception handlers and
# automatic exception recording to work, we need to make a
# series of wrapping and re-wrapping middlewares.
# First, grab the original middleware stack from Starlette. It
# comprises a stack of
# `ServerErrorMiddleware` -> [user defined middlewares] -> `ExceptionMiddleware`
inner_server_error_middleware: ServerErrorMiddleware = ( # type: ignore
self._original_build_middleware_stack() # type: ignore
)
if not isinstance(
inner_server_error_middleware, ServerErrorMiddleware
):
# Oops, something changed about how Starlette creates middleware stacks
_logger.error(
"Skipping FastAPI instrumentation due to unexpected middleware stack: expected %s, got %s",
ServerErrorMiddleware.__name__,
type(inner_server_error_middleware),
)
return inner_server_error_middleware
# We take [user defined middlewares] -> `ExceptionHandlerMiddleware`
# out of the outermost `ServerErrorMiddleware` and instead pass
# it to our own `ExceptionHandlerMiddleware`
exception_middleware = ExceptionHandlerMiddleware(
inner_server_error_middleware.app
)
# Now, we create a new `ServerErrorMiddleware` that wraps
# `ExceptionHandlerMiddleware` but otherwise uses the same
# original `handler` and debug setting. The end result is a
# middleware stack that's identical to the original stack except
# all user middlewares are covered by our
# `ExceptionHandlerMiddleware`.
error_middleware = ServerErrorMiddleware(
app=exception_middleware,
handler=inner_server_error_middleware.handler,
debug=inner_server_error_middleware.debug,
)
# Finally, we wrap the stack above in our actual OTEL
# middleware. As a result, an active tracing context exists for
# every use case of user-defined error and exception handlers as
# well as automatic recording of exceptions in active spans.
otel_middleware = OpenTelemetryMiddleware(
error_middleware,
excluded_urls=excluded_urls,
default_span_details=_get_default_span_details,
server_request_hook=server_request_hook,
client_request_hook=client_request_hook,
client_response_hook=client_response_hook,
# Pass in tracer/meter to get __name__and __version__ of fastapi instrumentation
tracer=tracer,
meter=meter,
http_capture_headers_server_request=http_capture_headers_server_request,
http_capture_headers_server_response=http_capture_headers_server_response,
http_capture_headers_sanitize_fields=http_capture_headers_sanitize_fields,
exclude_spans=exclude_spans,
)
# Ultimately, wrap everything in another default
# `ServerErrorMiddleware` (w/o user handlers) so that any
# exceptions raised in `OpenTelemetryMiddleware` are handled.
#
# This should not happen unless there is a bug in
# OpenTelemetryMiddleware, but if there is we don't want that to
# impact the user's application just because we wrapped the
# middlewares in this order.
return ServerErrorMiddleware(
app=otel_middleware,
)
app._original_build_middleware_stack = app.build_middleware_stack
app.build_middleware_stack = types.MethodType(
functools.wraps(app.build_middleware_stack)(
build_middleware_stack
),
app,
)
if not hasattr(BackgroundTask, "_otel_original_call"):
BackgroundTask._otel_original_call = BackgroundTask.__call__
async def traced_call(self):
span_name = f"BackgroundTask {getattr(self.func, '__name__', self.func.__class__.__name__)}"
with tracer.start_as_current_span(span_name):
return await BackgroundTask._otel_original_call(self)
BackgroundTask.__call__ = traced_call
app._is_instrumented_by_opentelemetry = True
if app not in _InstrumentedFastAPI._instrumented_fastapi_apps:
_InstrumentedFastAPI._instrumented_fastapi_apps.add(app)
else:
_logger.warning(
"Attempting to instrument FastAPI app while already instrumented"
)
[docs] @staticmethod
def uninstrument_app(app: fastapi.FastAPI):
original_build_middleware_stack = getattr(
app, "_original_build_middleware_stack", None
)
if original_build_middleware_stack:
app.build_middleware_stack = original_build_middleware_stack
del app._original_build_middleware_stack
app.middleware_stack = app.build_middleware_stack()
if hasattr(BackgroundTask, "_otel_original_call"):
BackgroundTask.__call__ = BackgroundTask._otel_original_call
del BackgroundTask._otel_original_call
app._is_instrumented_by_opentelemetry = False
# Remove the app from the set of instrumented apps to avoid calling uninstrument twice
# if the instrumentation is later disabled or such
# Use discard to avoid KeyError if already GC'ed
_InstrumentedFastAPI._instrumented_fastapi_apps.discard(app)
[docs] def instrumentation_dependencies(self) -> Collection[str]:
return _instruments
def _instrument(self, **kwargs: Any):
self._original_fastapi = fastapi.FastAPI
_InstrumentedFastAPI._instrument_kwargs = kwargs
fastapi.FastAPI = _InstrumentedFastAPI
def _uninstrument(self, **kwargs):
# Create a copy of the set to avoid RuntimeError during iteration
instances_to_uninstrument = list(
_InstrumentedFastAPI._instrumented_fastapi_apps
)
for instance in instances_to_uninstrument:
self.uninstrument_app(instance)
_InstrumentedFastAPI._instrumented_fastapi_apps.clear()
fastapi.FastAPI = self._original_fastapi
class _InstrumentedFastAPI(fastapi.FastAPI):
_instrument_kwargs: dict[str, Any] = {}
# Track instrumented app instances using weak references to avoid GC leaks
_instrumented_fastapi_apps: _WeakSet[fastapi.FastAPI] = _WeakSet()
_sem_conv_opt_in_mode = _StabilityMode.DEFAULT
def __init__(self, *args: Any, **kwargs: Any):
super().__init__(*args, **kwargs)
FastAPIInstrumentor.instrument_app(
self, **_InstrumentedFastAPI._instrument_kwargs
)
_InstrumentedFastAPI._instrumented_fastapi_apps.add(self)
def _get_route_details(scope):
"""
Function to retrieve Starlette route from scope.
TODO: there is currently no way to retrieve http.route from
a starlette application from scope.
See: https://github.com/encode/starlette/pull/804
Args:
scope: A Starlette scope
Returns:
A string containing the route or None
"""
app = scope["app"]
route = None
for starlette_route in app.routes:
match, _ = (
Route.matches(starlette_route, scope)
if isinstance(starlette_route, Route)
else starlette_route.matches(scope)
)
if match == Match.FULL:
try:
route = starlette_route.path
except AttributeError:
# routes added via host routing won't have a path attribute
route = scope.get("path")
break
if match == Match.PARTIAL:
route = starlette_route.path
return route
def _get_default_span_details(scope):
"""
Callback to retrieve span name and attributes from scope.
Args:
scope: A Starlette scope
Returns:
A tuple of span name and attributes
"""
route = _get_route_details(scope)
method = sanitize_method(scope.get("method", "").strip())
attributes = {}
if method == "_OTHER":
method = "HTTP"
if route:
attributes[HTTP_ROUTE] = route
if method and route: # http
span_name = f"{method} {route}"
elif route: # websocket
span_name = route
else: # fallback
span_name = method
return span_name, attributes