Emit an OpenTelemetry span for each HTTP request

datasette/app.py

@@ -49,8 +49,13 @@
 from .plugins import DEFAULT_PLUGINS, get_plugins, pm
 from .renderer import json_renderer
 from .resources import DatabaseResource, TableResource
-from .telemetry import tracer
-from .telemetry_registry import STARTUP
+from .telemetry import (
+    TelemetryMiddleware,
+    clamp_http_method,
+    request_span,
+    tracer,
+)
+from .telemetry_registry import HTTP_ROUTE, STARTUP
 from .tokens import TokenInvalid
 from .tracer import AsgiTracer
 from .url_builder import Urls
@@ -777,12 +782,16 @@ async def invoke_startup(self):
         # This must be called for Datasette to be in a usable state
         if self._startup_invoked:
             return
-        # invoke_startup() runs before any request exists, so every span its
-        # children create - the register_* hook dispatches, the internal
-        # catalog's db.query/db.write spans, and the prepare_connection
-        # warm-up of the read connections those touch - would otherwise be
-        # its own orphan root trace: around twenty of them on a fresh
-        # instance. Bracketing the whole thing gives them somewhere to belong.
+        # `datasette serve` calls invoke_startup() before uvicorn starts, so
+        # on the CLI path every span its children create - the register_*
+        # hook dispatches, the internal catalog's db.query/db.write spans,
+        # and the prepare_connection warm-up of the read connections those
+        # touch - would otherwise be its own orphan root trace: around twenty
+        # of them on a fresh instance. Bracketing the whole thing gives them
+        # somewhere to belong. An ASGI-hosted or programmatic deployment
+        # reaches here instead through AsgiRunOnFirstRequest, in which case
+        # this span nests under the first request's own span - honest enough,
+        # since it genuinely is that request's latency.
         # A connection warmed lazily later, by a request touching a new
         # database for the first time, nests under that request instead:
         # this span has already ended by then.
@@ -2837,6 +2846,11 @@ async def _close_on_shutdown():
         asgi = AsgiRunOnFirstRequest(asgi, on_startup=[setup_db, self.invoke_startup])
         for wrapper in pm.hook.asgi_wrapper(datasette=self):
             asgi = wrapper(asgi)
+        # Outermost, deliberately: plugin asgi_wrapper() middleware and the
+        # CSRF layer run *inside* this span, so a span created by an
+        # instrumented plugin parents to the request instead of becoming its
+        # own orphan root trace.
+        asgi = TelemetryMiddleware(asgi)
         return asgi
 
 
@@ -2917,8 +2931,26 @@ async def route_path(self, scope, receive, send, path):
         match, view = resolve_routes(self.routes, path)
 
         if match is None:
+            # No route matched, so the span keeps the bare method name it was
+            # given at the edge and gets no http.route. That is what semantic
+            # conventions ask for when the route is unknown.
             return await self.handle_404(request, send)
 
+        # The request span was started at the ASGI edge, before routing, so it
+        # carries only the method as a name. Now that the route is known, give
+        # it the `{method} {route}` shape semantic conventions want, and the
+        # http.route attribute - the low-cardinality counterpart to url.path,
+        # and so the one to group by.
+        span = request_span(scope)
+        if span is not None:
+            route = match.re.pattern
+            span.set_attribute(HTTP_ROUTE, route)
+            # Clamped, for the same reason the middleware clamps it: the method
+            # is a client-controlled string, and an unclamped one here would
+            # put attacker-supplied text back into the span name that the
+            # middleware just kept out of it.
+            span.update_name(f"{clamp_http_method(request.method)} {route}")
+
         new_scope = dict(scope, url_route={"kwargs": match.groupdict()})
         request.scope = new_scope
         try:

datasette/telemetry.py

@@ -18,7 +18,19 @@
 import re
 
 from opentelemetry import trace as otel_trace
+from opentelemetry.propagate import extract
+from opentelemetry.propagators.textmap import Getter
+from opentelemetry.trace import SpanKind, Status, StatusCode
 
+from .telemetry_registry import (
+    ERROR_TYPE,
+    HTTP_REQUEST_METHOD,
+    HTTP_RESPONSE_STATUS_CODE,
+    SERVER_ADDRESS,
+    URL_PATH,
+    URL_SCHEME,
+    USER_AGENT_ORIGINAL,
+)
 from .version import __version__
 
 # The semantic-convention version whose spellings this instrumentation
@@ -115,3 +127,221 @@ def sql_operation_name(sql: str) -> str | None:
     if keyword in DB_OPERATION_ALLOWLIST:
         return keyword
     return None
+
+
+# --- The HTTP request span ------------------------------------------------
+
+
+class _ScopeHeadersGetter(Getter):
+    """
+    Read W3C trace context out of an ASGI scope's headers.
+
+    `scope["headers"]` is a list of `(bytes, bytes)` pairs, lowercased by the
+    server per the ASGI spec - but `.lower()` is applied again here because
+    that is a spec promise about servers, not something this process
+    controls. Header bytes are latin-1 by RFC 9110.
+    """
+
+    def get(self, carrier, key):
+        wanted = key.lower().encode("latin-1")
+        values = [v.decode("latin-1") for k, v in carrier if k.lower() == wanted]
+        return values or None
+
+    def keys(self, carrier):
+        return [k.decode("latin-1") for k, _ in carrier]
+
+
+_HEADERS_GETTER = _ScopeHeadersGetter()
+
+
+# An unclamped method is an unbounded dimension a client controls: anyone can
+# send `FOO / HTTP/1.1`. Semantic conventions say map anything unrecognised to
+# `_OTHER`. These nine are the methods of RFC 9110 plus PATCH (RFC 5789).
+_KNOWN_METHODS = frozenset(
+    {"GET", "HEAD", "POST", "PUT", "DELETE", "CONNECT", "OPTIONS", "TRACE", "PATCH"}
+)
+
+
+def clamp_http_method(method):
+    "The request method if it is one we recognise, else ``_OTHER``."
+    method = (method or "").upper()
+    return method if method in _KNOWN_METHODS else "_OTHER"
+
+
+def _first_header(headers, name):
+    "The first value of a header, decoded, or None."
+    for key, value in headers:
+        if key.lower() == name:
+            return value.decode("latin-1")
+    return None
+
+
+def _url_path(scope):
+    """
+    The request path, with any query string removed.
+
+    `raw_path` is preferred because it is the bytes the client sent, before
+    percent-decoding - Datasette routes on database and table names that can
+    contain encoded slashes, which `scope["path"]` has already collapsed.
+
+    The split on "?" is not decoration. The ASGI spec's `raw_path` excludes
+    the query string, and uvicorn honours that, but the name is used the
+    other way round elsewhere in this same dependency tree: httpx's
+    `URL.raw_path` is documented as "raw bytes of both the path and query".
+    A server that followed that reading would hand us `?sql=...` here, and
+    Datasette's query strings carry user-supplied SQL, which core never
+    records. A literal "?" cannot appear unencoded in a path, so the split
+    costs nothing when the server is well behaved.
+    """
+    raw_path = scope.get("raw_path")
+    if raw_path:
+        if isinstance(raw_path, bytes):
+            raw_path = raw_path.decode("latin-1")
+        return raw_path.split("?", 1)[0]
+    return scope.get("path", "")
+
+
+# The request span is handed to `DatasetteRouter.route_path` through the ASGI
+# scope rather than through `get_current_span()`, because by the time routing
+# happens the current span may well be something else: a plugin
+# `asgi_wrapper()` runs *inside* this middleware, and an instrumented one makes
+# its own span current for the whole request. Reading the current span there
+# would set `http.route` on that plugin's span - and rename it - while leaving
+# the actual request span without the one attribute a trace UI groups by. Not
+# hypothetical: an ordinary tracing plugin triggers it.
+#
+# Namespaced per the ASGI spec's rules for extension keys. Absent when the span
+# is not recording, which is exactly when the router should skip the work too.
+REQUEST_SPAN_SCOPE_KEY = "datasette.telemetry.request_span"
+
+
+def request_span(scope):
+    """
+    The recording request span for an ASGI scope, or None.
+
+    Falls back to the current span so that a `DatasetteRouter` running under
+    some other instrumentation - one that started a SERVER span but of course
+    knows nothing about this scope key - still gets enriched.
+    """
+    span = scope.get(REQUEST_SPAN_SCOPE_KEY)
+    if span is None:
+        span = otel_trace.get_current_span()
+    # is_recording(), not `get_span_context().is_valid`: with no provider but
+    # an inbound `traceparent`, the API's NoOpTracer hands back a
+    # NonRecordingSpan carrying the *remote* context, which is perfectly valid
+    # and still records nothing.
+    return span if span.is_recording() else None
+
+
+class TelemetryMiddleware:
+    """
+    One `SpanKind.SERVER` span per HTTP request.
+
+    Mounted outermost in `Datasette.app()`, so every other span raised while
+    serving a request - database queries, plugin middleware, startup work on
+    a cold ASGI-hosted deployment - has somewhere to belong instead of
+    becoming its own root trace.
+
+    Deliberately much smaller than `opentelemetry-instrumentation-asgi`,
+    which needs several hundred lines of deferred-end machinery for
+    applications that return before their body is sent. Datasette does not:
+    `DatasetteRouter.route_path` awaits `response.asgi_send(send)`, and for a
+    streaming CSV export `AsgiStream.asgi_send` runs the generator inline.
+    All of it happens inside the single `await self.app(...)` below, so
+    ending the span in a `finally` covers the response body too.
+    """
+
+    def __init__(self, app):
+        self.app = app
+
+    async def __call__(self, scope, receive, send):
+        # First, before anything else: `AsgiLifespan` is *inside* this
+        # middleware, so lifespan startup and shutdown have to pass through
+        # untouched or the server never starts. Same for websockets.
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        headers = scope.get("headers") or []
+        # The *global* propagator, deliberately: it leaves the operator in
+        # control with no Datasette-specific setting - OTEL_PROPAGATORS=none
+        # disables extraction entirely, OTEL_PROPAGATORS=tracecontext drops
+        # baggage - and core configuring propagation itself would be the same
+        # mistake as core configuring sampling.
+        context = extract(headers, getter=_HEADERS_GETTER)
+        method = clamp_http_method(scope.get("method", ""))
+        # The method, not the URL: a span name has to be low cardinality, and
+        # the method is what is known out here at the edge, before any routing
+        # has happened.
+        with tracer.start_as_current_span(
+            method, context=context, kind=SpanKind.SERVER
+        ) as span:
+            if not span.is_recording():
+                # No provider installed, or a sampler dropped this trace.
+                # Everything below would be discarded, so skip building the
+                # `send` wrapper and let a default install pay almost
+                # nothing. Note this cannot be `get_span_context().is_valid`:
+                # with no provider but an inbound `traceparent`, the API's
+                # NoOpTracer returns a NonRecordingSpan carrying the *remote*
+                # context, which is perfectly valid and still records nothing.
+                await self.app(scope, receive, send)
+                return
+            span.set_attribute(HTTP_REQUEST_METHOD, method)
+            span.set_attribute(URL_PATH, _url_path(scope))
+            scheme = scope.get("scheme")
+            if scheme:
+                span.set_attribute(URL_SCHEME, scheme)
+            host = _first_header(headers, b"host")
+            if host:
+                span.set_attribute(SERVER_ADDRESS, host)
+            user_agent = _first_header(headers, b"user-agent")
+            if user_agent:
+                span.set_attribute(USER_AGENT_ORIGINAL, user_agent)
+
+            # A copy, not a mutation: the scope belongs to the server, and
+            # every other layer in Datasette extends it the same way.
+            scope = dict(scope, **{REQUEST_SPAN_SCOPE_KEY: span})
+
+            # The status cannot be read off a Response object: `asgi_static`,
+            # the favicon route, `AsgiStream` and `AsgiFileDownload` all call
+            # `send` directly and never build one. Wrapping `send` is the only
+            # thing that sees every response, including the 404 and 500
+            # handlers.
+            status_holder = {}
+
+            async def wrapped_send(message):
+                if (
+                    message["type"] == "http.response.start"
+                    and "status" not in status_holder
+                ):
+                    status_holder["status"] = message["status"]
+                await send(message)
+
+            escaped = False
+            try:
+                # Positional (scope, receive, send) throughout this codebase -
+                # `wrapped_send` is the third argument. `receive` is passed
+                # through unwrapped.
+                await self.app(scope, receive, wrapped_send)
+            except BaseException as exception:
+                # BaseException, not Exception: `route_path` turns almost
+                # everything into a 500 itself, but `asyncio.CancelledError`
+                # on client disconnect is a BaseException its `except
+                # Exception` deliberately does not catch.
+                escaped = True
+                span.set_attribute(ERROR_TYPE, type(exception).__name__)
+                span.set_status(Status(StatusCode.ERROR, str(exception)))
+                raise
+            finally:
+                status = status_holder.get("status")
+                if status is not None:
+                    span.set_attribute(HTTP_RESPONSE_STATUS_CODE, status)
+                    # 4xx is NOT an error for a SERVER span per semantic
+                    # conventions - the client made the mistake, not us.
+                    #
+                    # `not escaped` because this block still runs when an
+                    # exception is on its way out, and a response can have
+                    # started before it: the exception's class name is more
+                    # use than the string "500", so it wins.
+                    if status >= 500 and not escaped:
+                        span.set_status(Status(StatusCode.ERROR))
+                        span.set_attribute(ERROR_TYPE, str(status))