diff --git a/.github/workflows/deploy-latest.yml b/.github/workflows/deploy-latest.yml
index 3fc83438..b7f2361f 100644
--- a/.github/workflows/deploy-latest.yml
+++ b/.github/workflows/deploy-latest.yml
@@ -117,7 +117,7 @@ jobs:
--plugins-dir=plugins \
--branch=$GITHUB_SHA \
--version-note=$GITHUB_SHA \
- --extra-options="--setting template_debug 1 --setting trace_debug 1 --crossdb --root" \
+ --extra-options="--setting template_debug 1 --crossdb --root" \
--install 'datasette-ephemeral-tables>=0.2.2' \
--service "datasette-latest$SUFFIX" \
--secret $LATEST_DATASETTE_SECRET
diff --git a/datasette/app.py b/datasette/app.py
index 42be7425..8cee9b74 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -49,8 +49,14 @@ from .events import Event
from .plugins import DEFAULT_PLUGINS, get_plugins, pm
from .renderer import json_renderer
from .resources import DatabaseResource, TableResource
+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
from .utils import (
SPATIALITE_FUNCTIONS,
@@ -287,11 +293,6 @@ SETTINGS = (
False,
"Allow display of template debug information with ?_context=1",
),
- Setting(
- "trace_debug",
- False,
- "Allow display of SQL trace debug information with ?_trace=1",
- ),
Setting("base_url", "/", "Datasette URLs should use this base path"),
)
_HASH_URLS_REMOVED = "The hash_urls setting has been removed, try the datasette-hashed-urls plugin instead"
@@ -778,57 +779,73 @@ class Datasette:
# This must be called for Datasette to be in a usable state
if self._startup_invoked:
return
- # Register event classes
- event_classes = []
- for hook in pm.hook.register_events(datasette=self):
- extra_classes = await await_me_maybe(hook)
- if extra_classes:
- event_classes.extend(extra_classes)
- self.event_classes = tuple(event_classes)
+ # `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.
+ with tracer.start_as_current_span(STARTUP):
+ # Register event classes
+ event_classes = []
+ for hook in pm.hook.register_events(datasette=self):
+ extra_classes = await await_me_maybe(hook)
+ if extra_classes:
+ event_classes.extend(extra_classes)
+ self.event_classes = tuple(event_classes)
- # Register actions, but watch out for duplicate name/abbr
- action_names = {}
- action_abbrs = {}
- for hook in pm.hook.register_actions(datasette=self):
- if hook:
- for action in hook:
- if (
- action.name in action_names
- and action != action_names[action.name]
- ):
- raise StartupError(f"Duplicate action name: {action.name}")
- if (
- action.abbr
- and action.abbr in action_abbrs
- and action != action_abbrs[action.abbr]
- ):
- raise StartupError(f"Duplicate action abbr: {action.abbr}")
- action_names[action.name] = action
- if action.abbr:
- action_abbrs[action.abbr] = action
- self.actions[action.name] = action
+ # Register actions, but watch out for duplicate name/abbr
+ action_names = {}
+ action_abbrs = {}
+ for hook in pm.hook.register_actions(datasette=self):
+ if hook:
+ for action in hook:
+ if (
+ action.name in action_names
+ and action != action_names[action.name]
+ ):
+ raise StartupError(f"Duplicate action name: {action.name}")
+ if (
+ action.abbr
+ and action.abbr in action_abbrs
+ and action != action_abbrs[action.abbr]
+ ):
+ raise StartupError(f"Duplicate action abbr: {action.abbr}")
+ action_names[action.name] = action
+ if action.abbr:
+ action_abbrs[action.abbr] = action
+ self.actions[action.name] = action
- # Register column types (classes, not instances)
- self._column_types = {}
- for hook in pm.hook.register_column_types(datasette=self):
- if hook:
- for ct_cls in hook:
- if ct_cls.name in self._column_types:
- raise StartupError(f"Duplicate column type name: {ct_cls.name}")
- self._column_types[ct_cls.name] = ct_cls
+ # Register column types (classes, not instances)
+ self._column_types = {}
+ for hook in pm.hook.register_column_types(datasette=self):
+ if hook:
+ for ct_cls in hook:
+ if ct_cls.name in self._column_types:
+ raise StartupError(
+ f"Duplicate column type name: {ct_cls.name}"
+ )
+ self._column_types[ct_cls.name] = ct_cls
- for hook in pm.hook.prepare_jinja2_environment(
- env=self._jinja_env, datasette=self
- ):
- await await_me_maybe(hook)
- # Ensure internal tables and metadata are populated before startup hooks
- await self._refresh_schemas()
- await self._save_queries_from_config()
- # Load column_types from config into internal DB
- await self._apply_column_types_config()
- for hook in pm.hook.startup(datasette=self):
- await await_me_maybe(hook)
- self._startup_invoked = True
+ for hook in pm.hook.prepare_jinja2_environment(
+ env=self._jinja_env, datasette=self
+ ):
+ await await_me_maybe(hook)
+ # Ensure internal tables and metadata are populated before startup hooks
+ await self._refresh_schemas()
+ await self._save_queries_from_config()
+ # Load column_types from config into internal DB
+ await self._apply_column_types_config()
+ for hook in pm.hook.startup(datasette=self):
+ await await_me_maybe(hook)
+ self._startup_invoked = True
def sign(self, value, namespace="default"):
return URLSafeSerializer(self._secret, namespace).dumps(value)
@@ -2844,8 +2861,6 @@ class Datasette:
self.close()
asgi = CrossOriginProtectionMiddleware(DatasetteRouter(self, routes), self)
- if self.setting("trace_debug"):
- asgi = AsgiTracer(asgi)
asgi = AsgiLifespan(
asgi,
on_startup=[self._startup_sequence],
@@ -2854,6 +2869,12 @@ class Datasette:
asgi = AsgiRunOnFirstRequest(asgi, on_startup=[self._startup_sequence])
for wrapper in pm.hook.asgi_wrapper(datasette=self):
asgi = wrapper(asgi)
+ # Outermost, deliberately: plugin asgi_wrapper() middleware, the
+ # CSRF layer and the first-request startup fallback all run *inside*
+ # this span, so a span created by an instrumented plugin - or by
+ # startup work triggered by the first request - parents to the
+ # request instead of becoming its own orphan root trace.
+ asgi = TelemetryMiddleware(asgi)
return asgi
@@ -2934,8 +2955,26 @@ class DatasetteRouter:
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:
diff --git a/datasette/database.py b/datasette/database.py
index e162d34e..efb8c56e 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -1,19 +1,45 @@
import asyncio
import atexit
+import contextvars
import inspect
import os
import queue
import sys
import tempfile
import threading
+import time
import uuid
from collections import namedtuple
from pathlib import Path
import sqlite_utils
+from opentelemetry import context as otel_context_api
+from opentelemetry.trace import Link, Status, StatusCode, get_current_span
from .inspect import inspect_hash
-from .tracer import trace
+from .telemetry import sql_attribute, sql_operation_name, tracer
+from .telemetry_registry import (
+ DB_COLLECTION_NAME,
+ DB_NAMESPACE,
+ DB_OPERATION_NAME,
+ DB_QUERY,
+ DB_QUERY_EXECUTE,
+ DB_QUERY_TEXT,
+ DB_SYSTEM,
+ DB_WRITE_EXECUTE,
+ DB_WRITE_QUEUE_WAIT,
+ EXECUTEMANY,
+ EXECUTESCRIPT,
+ INTERRUPTED,
+ ISOLATED_CONNECTION,
+ PARAM_COUNT,
+ PARAM_SETS,
+ ROWS_RETURNED,
+ SQL_ERROR_SUPPRESSED,
+ TIME_LIMIT_MS,
+ TRANSACTION,
+ TRUNCATED,
+)
from .utils import (
call_with_supported_arguments,
detect_fts,
@@ -257,7 +283,15 @@ class Database:
cursor, return_all=return_all, returning_limit=returning_limit
)
- with trace("sql", database=self.name, sql=sql.strip(), params=params):
+ with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+ span.set_attribute(DB_SYSTEM, "sqlite")
+ span.set_attribute(DB_NAMESPACE, self.name)
+ span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+ operation_name = sql_operation_name(sql)
+ if operation_name:
+ span.set_attribute(DB_OPERATION_NAME, operation_name)
+ if params:
+ span.set_attribute(PARAM_COUNT, len(params))
results = await self.execute_write_fn(
_inner, block=block, request=request, transaction=transaction
)
@@ -269,7 +303,15 @@ class Database:
def _inner(conn):
return conn.executescript(sql)
- with trace("sql", database=self.name, sql=sql.strip(), executescript=True):
+ # No db.operation.name here, deliberately: executescript() runs
+ # several semicolon-separated statements, and semantic conventions
+ # say the attribute should not be extracted from query text that
+ # can hold more than one operation - see sql_operation_name().
+ with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+ span.set_attribute(DB_SYSTEM, "sqlite")
+ span.set_attribute(DB_NAMESPACE, self.name)
+ span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+ span.set_attribute(EXECUTESCRIPT, True)
results = await self.execute_write_fn(
_inner, block=block, transaction=False, request=request
)
@@ -289,13 +331,22 @@ class Database:
return conn.executemany(sql, count_params(params_seq)), count
- with trace(
- "sql", database=self.name, sql=sql.strip(), executemany=True
- ) as kwargs:
+ with tracer.start_as_current_span(DB_QUERY, kind=DB_QUERY.kind) as span:
+ span.set_attribute(DB_SYSTEM, "sqlite")
+ span.set_attribute(DB_NAMESPACE, self.name)
+ span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+ span.set_attribute(EXECUTEMANY, True)
+ # A single statement run with many parameter sets, so unlike
+ # execute_write_script() there is exactly one operation to name.
+ operation_name = sql_operation_name(sql)
+ if operation_name:
+ span.set_attribute(DB_OPERATION_NAME, operation_name)
results, count = await self.execute_write_fn(
_inner, block=block, request=request
)
- kwargs["count"] = count
+ # count is the number of parameter *sets* consumed by
+ # executemany(), not a row count - executemany returns no rows.
+ span.set_attribute(PARAM_SETS, count)
return results
async def execute_isolated_fn(self, fn):
@@ -321,9 +372,18 @@ class Database:
return _run()
if not write:
# Immutable database - no writes can ever occur, so there is no
- # write queue to block; run against a fresh read-only connection
+ # write queue to block; run against a fresh read-only connection.
+ # A fresh copy_context() is required per submit (not one shared
+ # copy reused across calls): concurrent execution of the same
+ # Context raises "RuntimeError: cannot enter context ... already
+ # entered". This propagates the caller's otel context (e.g. the
+ # enclosing db.query span) onto the worker thread.
+ #
+ # It also propagates every *other* ContextVar - see the note in
+ # execute_fn() for why that is safe.
+ ctx = contextvars.copy_context()
return await asyncio.get_running_loop().run_in_executor(
- self.ds.executor, _run
+ self.ds.executor, ctx.run, _run
)
# Threaded mode - send to write thread
return await self._send_to_write_thread(fn, isolated_connection=True)
@@ -428,8 +488,24 @@ class Database:
task_id = uuid.uuid5(uuid.NAMESPACE_DNS, "datasette.io")
loop = asyncio.get_running_loop()
reply_future = loop.create_future()
+ # Captured here, on the event loop, at enqueue time: the otel
+ # Context (carrying the enclosing db.query span, if any) and the
+ # timestamp used to build the db.write.queue_wait span once this
+ # task is dequeued on the write thread. `block` travels with the
+ # task too, because it decides whether that context is this task's
+ # parent or only a link target - see `_execute_writes`.
self._write_queue.put(
- WriteTask(fn, task_id, loop, reply_future, isolated_connection, transaction)
+ WriteTask(
+ fn,
+ task_id,
+ loop,
+ reply_future,
+ isolated_connection,
+ transaction,
+ otel_context_api.get_current(),
+ time.time_ns(),
+ block,
+ )
)
if block:
return await reply_future
@@ -443,6 +519,16 @@ class Database:
conn = None
try:
conn = self.connect(write=True)
+ # This warm-up runs before any write has ever been queued, so
+ # there is no captured caller context to attach - and a raw
+ # threading.Thread does not inherit the context of whoever started
+ # it. Spans created by plugin hooks here are therefore roots even
+ # when the write thread is started from inside invoke_startup():
+ # its datasette.startup span is current on the event loop but does
+ # not cross this thread boundary. Read connections differ - they
+ # warm up inside executor tasks submitted with copy_context(), so
+ # their prepare_connection spans do nest under whoever triggered
+ # them.
self.ds._prepare_connection(conn, self.name)
except Exception as e: # noqa: BLE001
# Stored and re-raised to whoever queues the next write
@@ -457,40 +543,119 @@ class Database:
# Best-effort close as the write thread exits
pass
return
- exception = None
- result = None
- if conn_exception is not None:
- exception = conn_exception
- elif task.isolated_connection:
- try:
- isolated_connection = self.connect(write=True)
- try:
- result = task.fn(isolated_connection)
- finally:
- isolated_connection.close()
- try:
- self._all_file_connections.remove(isolated_connection)
- except ValueError:
- # Was probably a memory connection
- pass
- except Exception as e: # noqa: BLE001
- # Write thread must survive any task failure or the database wedges
- sys.stderr.write(f"{e}\n")
- sys.stderr.flush()
- exception = e
+ # `task.block` decides how this task's spans relate to the
+ # context captured at enqueue time:
+ #
+ # - block=True: the caller genuinely awaits the reply, so
+ # containment is accurate. Restore that context as current
+ # (attach below) so db.write.queue_wait/db.write.execute parent
+ # normally to the request that queued them. The token must be
+ # detached below in `finally` - a leaked token silently
+ # poisons this thread's ambient context for every write
+ # processed after it, and a *wrong*-token detach only logs a
+ # warning rather than raising, so this pairing is load-bearing
+ # and easy to get wrong silently.
+ # - block=False: the caller returned already without awaiting,
+ # so the enqueueing span may already have closed (and
+ # exported) before this task's spans even start - parenting to
+ # it would make a child appear to outlive its already-closed
+ # parent, which OTel allows but which renders badly in most
+ # trace UIs. The enqueueing request *caused* this write
+ # without *containing* it, so nothing is attached here -
+ # instead each write span is started as its own root (explicit
+ # empty `context=`, so the write thread's ambient context
+ # cannot supply a parent either) carrying one `Link` back to
+ # the enqueueing span's context, built once into
+ # `write_span_kwargs` and spread into every start_span call
+ # below.
+ token = None
+ write_span_kwargs = {}
+ if task.block:
+ token = otel_context_api.attach(task.otel_context)
else:
- try:
- if task.transaction:
- with conn:
- conn.execute("BEGIN IMMEDIATE")
- result = task.fn(conn)
- else:
- result = task.fn(conn)
- except Exception as e: # noqa: BLE001
- sys.stderr.write(f"{e}\n")
- sys.stderr.flush()
- exception = e
- _deliver_write_result(task, result, exception)
+ enqueueing_span_context = get_current_span(
+ task.otel_context
+ ).get_span_context()
+ # No attributes on the link: there is only one kind of link
+ # here, so naming the relationship would be a constant that
+ # carries no information a consumer does not already have
+ # from the link's existence.
+ links = (
+ [Link(enqueueing_span_context)]
+ if enqueueing_span_context.is_valid
+ else []
+ )
+ write_span_kwargs = {
+ "context": otel_context_api.Context(),
+ "links": links,
+ }
+ try:
+ exception = None
+ result = None
+ # Explicit start_time/end_time rather than a `with` block:
+ # this span's duration is the time the task actually spent
+ # waiting in the queue (enqueue -> dequeue), not the near-
+ # zero time spent constructing/ending the span object here.
+ tracer.start_span(
+ DB_WRITE_QUEUE_WAIT,
+ start_time=task.enqueued_at_ns,
+ **write_span_kwargs,
+ ).end(end_time=time.time_ns())
+ if conn_exception is not None:
+ # fn never runs in this branch, so there is nothing to
+ # wrap in a db.write.execute span.
+ exception = conn_exception
+ elif task.isolated_connection:
+ try:
+ with tracer.start_as_current_span(
+ DB_WRITE_EXECUTE, **write_span_kwargs
+ ) as span:
+ span.set_attribute(
+ ISOLATED_CONNECTION,
+ task.isolated_connection,
+ )
+ span.set_attribute(TRANSACTION, task.transaction)
+ isolated_connection = self.connect(write=True)
+ try:
+ result = task.fn(isolated_connection)
+ finally:
+ isolated_connection.close()
+ try:
+ self._all_file_connections.remove(
+ isolated_connection
+ )
+ except ValueError:
+ # Was probably a memory connection
+ pass
+ except Exception as e: # noqa: BLE001
+ # Write thread must survive any task failure or the database wedges
+ sys.stderr.write(f"{e}\n")
+ sys.stderr.flush()
+ exception = e
+ else:
+ try:
+ with tracer.start_as_current_span(
+ DB_WRITE_EXECUTE, **write_span_kwargs
+ ) as span:
+ span.set_attribute(
+ ISOLATED_CONNECTION,
+ task.isolated_connection,
+ )
+ span.set_attribute(TRANSACTION, task.transaction)
+ if task.transaction:
+ with conn:
+ conn.execute("BEGIN IMMEDIATE")
+ result = task.fn(conn)
+ else:
+ result = task.fn(conn)
+ except Exception as e: # noqa: BLE001
+ sys.stderr.write(f"{e}\n")
+ sys.stderr.flush()
+ exception = e
+ _deliver_write_result(task, result, exception)
+ finally:
+ if token is not None:
+ otel_context_api.detach(token)
async def execute_fn(self, fn):
self._check_not_closed()
@@ -512,7 +677,28 @@ class Database:
with self._pending_execute_futures_lock:
self._check_not_closed()
- future = self.ds.executor.submit(in_thread)
+ # A fresh copy_context() is required per submit (not one shared
+ # copy reused across calls): concurrent execution of the same
+ # Context raises "RuntimeError: cannot enter context ...
+ # already entered". This propagates the caller's otel context
+ # (e.g. the enclosing db.query span) onto the worker thread.
+ #
+ # copy_context() is not selective: it also carries Datasette's own
+ # ContextVars - _skip_permission_checks and _permission_check_cache
+ # (datasette/permissions.py) and _in_datasette_client (app.py) -
+ # into worker threads, where they previously took their defaults.
+ # That is safe, for two reasons. Nothing reads them on a worker
+ # thread: the permission code that reads the first two is async and
+ # only ever runs on the event loop. And Context.run() restores the
+ # thread's previous context when the callable returns, so a value
+ # cannot outlive the submit that carried it and reach the next task
+ # on this shared pool - "skip permission checks" in particular can
+ # never bleed from one request into another's query. Where a value
+ # would be read - a plugin calling datasette.in_client() from inside
+ # an execute_fn callable - seeing the submitting request's value is
+ # the more accurate answer, not a leak.
+ ctx = contextvars.copy_context()
+ future = self.ds.executor.submit(ctx.run, in_thread)
self._pending_execute_futures.add(future)
future.add_done_callback(self._remove_pending_execute_future)
return await asyncio.wrap_future(future)
@@ -525,48 +711,143 @@ class Database:
custom_time_limit=None,
page_size=None,
log_sql_errors=True,
+ table=None,
):
- """Executes sql against db_name in a thread"""
+ """Executes sql against db_name in a thread
+
+ `table`, if passed, is recorded as the `db.collection.name` span
+ attribute. It exists for callers that already know which table the
+ query targets - the table and row views - and is never derived from
+ `sql` itself: deriving it would be a parse, and on an instance where
+ anyone can create a table the resulting value set has no ceiling.
+ """
self._check_not_closed()
page_size = page_size or self.ds.page_size
+ time_limit_ms = self.ds.sql_time_limit_ms
+ # A caller that hands in a budget shorter than the instance-wide
+ # sql_time_limit_ms is saying "this may not finish, and that is an
+ # answer I can use" - and every such caller in core does treat the
+ # timeout as normal: table_counts() stores None per table, facet
+ # suggestion moves on to the next column, autocomplete falls back to a
+ # prefix query. Those timeouts are therefore not span errors. Without
+ # this, the homepage alone emits one red span per table (it counts
+ # every table under a 10ms budget) on every single hit.
+ #
+ # A query that runs out the instance-wide limit is a different event -
+ # nobody asked for a short budget, so it stays an error.
+ timeout_expected = bool(custom_time_limit) and custom_time_limit < time_limit_ms
+ if timeout_expected:
+ time_limit_ms = custom_time_limit
def sql_operation_in_thread(conn):
- time_limit_ms = self.ds.sql_time_limit_ms
- if custom_time_limit and custom_time_limit < time_limit_ms:
- time_limit_ms = custom_time_limit
-
- with sqlite_timelimit(conn, time_limit_ms):
+ # This span is created inside the worker thread. Its parent is
+ # resolved from the ambient otel context, which was propagated
+ # onto this thread via copy_context() at the executor.submit()
+ # boundary in execute_fn() (or run_in_executor() for immutable
+ # databases) - so it parents correctly to the enclosing
+ # db.query span despite running on a different thread.
+ #
+ # Exception handling is explicit rather than left to the context
+ # manager's flags, which apply to every exception type alike. This
+ # span needs to tell two apart: an expected timeout is never an
+ # error, while a genuine SQL failure is one unless the caller
+ # passed log_sql_errors=False, meaning it was probing and treats
+ # failure as an expected answer. Without the latter, facet
+ # suggestion marks two spans per text column as failed on every
+ # table page; without the former, so does every homepage hit.
+ with tracer.start_as_current_span(
+ DB_QUERY_EXECUTE,
+ record_exception=False,
+ set_status_on_exception=False,
+ ) as execute_span:
try:
- cursor = conn.cursor()
- cursor.execute(sql, params if params is not None else {})
- max_returned_rows = self.ds.max_returned_rows
- if max_returned_rows == page_size:
- max_returned_rows += 1
- if max_returned_rows and truncate:
- rows = cursor.fetchmany(max_returned_rows + 1)
- truncated = len(rows) > max_returned_rows
- rows = rows[:max_returned_rows]
- else:
- rows = cursor.fetchall()
- truncated = False
- except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
- if e.args == ("interrupted",):
- raise QueryInterrupted(e, sql, params)
+ with sqlite_timelimit(conn, time_limit_ms):
+ try:
+ cursor = conn.cursor()
+ cursor.execute(sql, params if params is not None else {})
+ max_returned_rows = self.ds.max_returned_rows
+ if max_returned_rows == page_size:
+ max_returned_rows += 1
+ if max_returned_rows and truncate:
+ rows = cursor.fetchmany(max_returned_rows + 1)
+ truncated = len(rows) > max_returned_rows
+ rows = rows[:max_returned_rows]
+ else:
+ rows = cursor.fetchall()
+ truncated = False
+ except (sqlite3.OperationalError, sqlite3.DatabaseError) as e:
+ if e.args == ("interrupted",):
+ raise QueryInterrupted(e, sql, params)
+ if log_sql_errors:
+ sys.stderr.write(
+ f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n"
+ )
+ sys.stderr.flush()
+ raise
+ except QueryInterrupted as e:
+ if not timeout_expected:
+ execute_span.record_exception(e)
+ execute_span.set_status(Status(StatusCode.ERROR, str(e)))
+ raise
+ except Exception as e:
if log_sql_errors:
- sys.stderr.write(
- f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n"
- )
- sys.stderr.flush()
+ execute_span.record_exception(e)
+ execute_span.set_status(Status(StatusCode.ERROR, str(e)))
raise
- if truncate:
- return Results(rows, truncated, cursor.description)
+ if truncate:
+ return Results(rows, truncated, cursor.description)
- else:
- return Results(rows, False, cursor.description)
+ else:
+ return Results(rows, False, cursor.description)
- with trace("sql", database=self.name, sql=sql.strip(), params=params):
- results = await self.execute_fn(sql_operation_in_thread)
+ # Exception handling is explicit rather than left to the context
+ # manager's defaults, so that callers passing log_sql_errors=False
+ # can be honoured - see the comment on the generic handler below.
+ with tracer.start_as_current_span(
+ DB_QUERY,
+ kind=DB_QUERY.kind,
+ record_exception=False,
+ set_status_on_exception=False,
+ ) as span:
+ span.set_attribute(DB_SYSTEM, "sqlite")
+ span.set_attribute(DB_NAMESPACE, self.name)
+ span.set_attribute(DB_QUERY_TEXT, sql_attribute(sql))
+ span.set_attribute(TIME_LIMIT_MS, time_limit_ms)
+ operation_name = sql_operation_name(sql)
+ if operation_name:
+ span.set_attribute(DB_OPERATION_NAME, operation_name)
+ if table:
+ span.set_attribute(DB_COLLECTION_NAME, table)
+ if params:
+ span.set_attribute(PARAM_COUNT, len(params))
+ try:
+ results = await self.execute_fn(sql_operation_in_thread)
+ except QueryInterrupted as e:
+ # datasette.interrupted is set either way - it is the
+ # signal worth having. Only the ERROR status is
+ # conditional; see the timeout_expected comment above.
+ span.set_attribute(INTERRUPTED, True)
+ if not timeout_expected:
+ span.set_status(Status(StatusCode.ERROR, str(e)))
+ span.record_exception(e)
+ raise
+ except Exception as e:
+ # log_sql_errors=False means the caller is probing and
+ # treats failure as an expected answer, not an error.
+ # Facet suggestion is the big one: it runs json_type()
+ # against every column precisely to find out which ones
+ # raise, so a table with N text columns would otherwise
+ # mark N queries per page as failed - burying real errors
+ # and setting off any alerting based on span status.
+ if log_sql_errors:
+ span.record_exception(e)
+ span.set_status(Status(StatusCode.ERROR, str(e)))
+ else:
+ span.set_attribute(SQL_ERROR_SUPPRESSED, True)
+ raise
+ span.set_attribute(TRUNCATED, results.truncated)
+ span.set_attribute(ROWS_RETURNED, len(results.rows))
return results
@property
@@ -854,16 +1135,28 @@ def _apply_write_wrapper(fn, wrapper_factory, track_event):
class WriteTask:
__slots__ = (
+ "block",
+ "enqueued_at_ns",
"fn",
"isolated_connection",
"loop",
+ "otel_context",
"reply_future",
"task_id",
"transaction",
)
def __init__(
- self, fn, task_id, loop, reply_future, isolated_connection, transaction
+ self,
+ fn,
+ task_id,
+ loop,
+ reply_future,
+ isolated_connection,
+ transaction,
+ otel_context,
+ enqueued_at_ns,
+ block,
):
self.fn = fn
self.task_id = task_id
@@ -871,6 +1164,14 @@ class WriteTask:
self.reply_future = reply_future
self.isolated_connection = isolated_connection
self.transaction = transaction
+ self.otel_context = otel_context
+ self.enqueued_at_ns = enqueued_at_ns
+ # Whether the enqueueing caller awaits the reply future. Decides how
+ # `_execute_writes` relates this task's spans to `otel_context`:
+ # parent (block=True) or span-link target (block=False). See the
+ # comment at the WriteTask construction site in
+ # `_send_to_write_thread`.
+ self.block = block
def _deliver_write_result(task, result, exception):
diff --git a/datasette/telemetry.py b/datasette/telemetry.py
new file mode 100644
index 00000000..311d661f
--- /dev/null
+++ b/datasette/telemetry.py
@@ -0,0 +1,347 @@
+"""
+OpenTelemetry integration for Datasette core.
+
+Core depends on `opentelemetry-api` only. It never creates a
+`TracerProvider`, never configures an exporter, and never touches
+sampling - that is the responsibility of whoever is running Datasette
+(an `opentelemetry-instrument` agent, a future plugin, or a test
+harness).
+
+With no provider installed every span produced here is a
+`NonRecordingSpan`. That is not free - a table page emits ~58 spans -
+but it is below what an end-to-end page benchmark can resolve: measured
+across 15 runs of a 5,000-row table page, the median moved 9.80ms to
+9.98ms while run-to-run spread was 1.4ms. Installing an SDK provider is
+what costs something measurable.
+"""
+
+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
+# actually emits. Deliberately NOT the latest release.
+#
+# A schema URL is a machine-readable claim: a consumer doing schema
+# translation replays the renames between the declared version and the one
+# it wants, so the claim has to name the version whose spellings are on the
+# wire. A wrong one makes translation wrong rather than merely uninformative.
+#
+# Datasette emits `db.system`, which was renamed to `db.system.name` in
+# semconv 1.30.0. Everything else it emits (`db.namespace`, `db.query.text`,
+# `db.operation.name`, `db.collection.name`) has been current since 1.26.0.
+# So 1.29.0 is the highest version at which every name emitted here is the
+# current spelling. Everything under `datasette.*` is Datasette's own and
+# outside semconv, so it is unaffected either way.
+#
+# Declaring 1.43.0 would be false about `db.system`, and would actively STOP
+# a consumer translating it forward, because it asserts the rename already
+# happened. Bump this deliberately, in the same commit as the attribute
+# renames it implies - it is a claim about the names, not decoration.
+SCHEMA_URL = "https://opentelemetry.io/schemas/1.29.0"
+
+tracer = otel_trace.get_tracer("datasette", __version__, schema_url=SCHEMA_URL)
+
+MAX_SQL_LENGTH = 2048
+
+
+def sql_attribute(sql: str) -> str:
+ "Truncate SQL text so it is safe to attach to a span as an attribute."
+ sql = sql.strip()
+ if len(sql) <= MAX_SQL_LENGTH:
+ return sql
+ return sql[:MAX_SQL_LENGTH] + "…[truncated]"
+
+
+# db.operation.name is the leading keyword of a statement matched against a
+# fixed allowlist - deliberately not a parse.
+#
+# This runs against arbitrary user-supplied SQL (the `?sql=` query string,
+# canned queries, anything typed into the query editor), and the attribute is
+# a candidate dimension on a query-duration metric in a later phase. A metric
+# series is keyed by its attribute values, so echoing back an arbitrary first
+# token would let one visitor's typo mint a new, permanent series. The
+# allowlist bounds that at a fixed, small set regardless of what anyone sends.
+DB_OPERATION_ALLOWLIST = frozenset(
+ {
+ "SELECT",
+ "INSERT",
+ "UPDATE",
+ "DELETE",
+ "CREATE",
+ "DROP",
+ "ALTER",
+ "PRAGMA",
+ "EXPLAIN",
+ "REPLACE",
+ "VACUUM",
+ "ANALYZE",
+ "WITH",
+ }
+)
+
+_LEADING_KEYWORD = re.compile(r"^\s*([A-Za-z]+)")
+
+
+def sql_operation_name(sql: str) -> str | None:
+ """
+ The statement's leading keyword, if it is one we recognise.
+
+ Returns None - never a guess - for anything not on the allowlist,
+ including a statement that opens with a comment or with punctuation such
+ as the "(" of a parenthesised SELECT.
+
+ Known limitation: a statement beginning with a CTE reports `WITH` rather
+ than the operation inside it, and a substantial share of Datasette's own
+ reads take that form. Extracting more than the leading keyword means
+ handling comment stripping, parenthesised `(SELECT ...) UNION` and
+ compound names like `CREATE TABLE` - each a special case a hand-rolled
+ matcher would accrete and eventually get wrong. Omitting a name beats
+ guessing at one.
+
+ Only safe to call with a single statement: `execute_write_script()` runs
+ several separated by semicolons, and semantic conventions say
+ `db.operation.name` "SHOULD NOT be extracted from db.query.text, when the
+ database system supports query text with multiple operations in non-batch
+ operations" - so that call site does not use this at all rather than
+ reporting only the first statement's operation.
+ """
+ match = _LEADING_KEYWORD.match(sql)
+ if not match:
+ return None
+ keyword = match.group(1).upper()
+ 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))
diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py
new file mode 100644
index 00000000..4e750605
--- /dev/null
+++ b/datasette/telemetry_registry.py
@@ -0,0 +1,388 @@
+"""
+The single source of truth for every span and span attribute that Datasette
+core emits.
+
+Three things read this module, which is the point of it existing:
+
+1. **The instrumentation itself.** `Attribute` and `SpanName` subclass `str`,
+ so a registry entry *is* the string OpenTelemetry wants. Call sites pass
+ `DB_NAMESPACE` where they used to pass `"db.namespace"` - no wrapper API
+ over the OTel calls, no parallel structure to keep in step, and a typo is
+ now an `ImportError` instead of a silently misnamed attribute.
+
+2. **The documentation.** `docs/telemetry_doc.py` renders the span reference
+ in `docs/internals.rst` from these definitions using cog, and
+ `cog --check` runs in CI - so the docs cannot drift from the code.
+
+3. **A conformance test.** `tests/test_telemetry_registry.py` makes real
+ requests, collects every span and attribute actually emitted, and compares
+ both directions: emitted-but-unregistered catches instrumentation added
+ without documentation, registered-but-never-emitted catches documentation
+ describing something that no longer exists. Neither the type system nor
+ the generated docs can catch that second case.
+"""
+
+from opentelemetry.trace import SpanKind
+
+
+class Attribute(str):
+ """
+ A span attribute key, carrying its own documentation.
+
+ Subclasses `str` so it can be handed straight to `set_attribute()`.
+ """
+
+ __slots__ = ("description", "optional")
+
+ def __new__(cls, name, description, optional=False):
+ self = super().__new__(cls, name)
+ self.description = description
+ self.optional = optional
+ return self
+
+ def __repr__(self):
+ return f"Attribute({str(self)!r})"
+
+
+class SpanName(str):
+ "A span name, carrying its documentation and the attributes it may set."
+
+ __slots__ = ("attributes", "description", "dynamic", "kind", "prefix")
+
+ def __new__(
+ cls,
+ name,
+ description,
+ attributes=(),
+ prefix=False,
+ dynamic=False,
+ kind=SpanKind.INTERNAL,
+ ):
+ self = super().__new__(cls, name)
+ self.description = description
+ self.attributes = tuple(attributes)
+ # True for a span family whose emitted names carry a variable suffix,
+ # so the conformance test matches by prefix rather than equality.
+ # Nothing sets it yet.
+ self.prefix = prefix
+ # True when the emitted name is composed at runtime and shares no
+ # fixed prefix with the registry entry - the HTTP request span, whose
+ # name is the request method followed by the matched route. There is
+ # no substring of the entry that could be matched against the wire, so
+ # `span_for()` resolves these by span kind instead, and the entry's own
+ # string is a template written for a human reading the generated
+ # reference.
+ self.dynamic = dynamic
+ # SpanKind.INTERNAL by default - every span Datasette emits describes
+ # its own internal work. db.query is the one exception: it is a real
+ # database call, so semantic conventions (and trace UIs, which key
+ # their database styling off this) expect SpanKind.CLIENT.
+ self.kind = kind
+ return self
+
+ def __repr__(self):
+ return f"SpanName({str(self)!r})"
+
+
+# --- Attributes -----------------------------------------------------------
+#
+# Shared attributes are defined once and referenced by every span that sets
+# them, so "which spans carry db.namespace?" is answerable by grep.
+
+HTTP_REQUEST_METHOD = Attribute(
+ "http.request.method",
+ "The HTTP method, clamped to the nine methods RFC 9110 and RFC 5789 "
+ "define. Anything else is reported as ``_OTHER``: the method is a "
+ "client-controlled string, so echoing it back unbounded would be a "
+ "cardinality hazard.",
+)
+HTTP_RESPONSE_STATUS_CODE = Attribute(
+ "http.response.status_code",
+ "The status of the response, read from the ASGI ``http.response.start`` "
+ "message rather than from a :ref:`internals_response` object - several "
+ "views, including static files, file downloads and streaming CSV, send "
+ "that message themselves and never build one. Omitted if the connection "
+ "closed before anything was sent.",
+ optional=True,
+)
+HTTP_ROUTE = Attribute(
+ "http.route",
+ "The route the request matched, as the compiled regular expression "
+ "pattern Datasette routes with - for example "
+ "``/(?P[^\\/\\.]+)/(?P[^\\/\\.]+)(\\.(?P\\w+))?$`` "
+ "for a table page. It is deliberately the pattern rather than a prettified "
+ "``/{database}/{table}`` template: the route table is fixed when the app "
+ "is built, so the pattern is exact, bounded and needs no parsing, whereas "
+ "the transform into something prettier accretes edge cases. Unlike "
+ "``url.path`` this is low cardinality, so it is the attribute to group by. "
+ "Omitted when no route matched - a 404 - which is also when the span name "
+ "falls back to the bare method.",
+ optional=True,
+)
+URL_PATH = Attribute(
+ "url.path",
+ "The path portion of the URL. The query string is deliberately **not** "
+ "recorded, on this or any other span: Datasette puts user-supplied SQL in "
+ "``?sql=`` and canned query parameters in the query string, so exporting "
+ "it by default would export exactly the data the rest of this "
+ "instrumentation is careful with.",
+)
+URL_SCHEME = Attribute("url.scheme", "``http`` or ``https``.")
+SERVER_ADDRESS = Attribute(
+ "server.address",
+ "The ``Host`` header. Client-controlled, so treat it as untrusted input "
+ "rather than as the identity of the server.",
+ optional=True,
+)
+USER_AGENT_ORIGINAL = Attribute(
+ "user_agent.original",
+ "The ``User-Agent`` header, verbatim. Omitted if the client sent none. "
+ "The client's IP address is deliberately not recorded: core records no "
+ "identifier that would tie a span to a person.",
+ optional=True,
+)
+ERROR_TYPE = Attribute(
+ "error.type",
+ "Set when the request failed: the exception class name if one escaped the "
+ "application, otherwise the status code as a string for a 5xx response. "
+ "A 4xx does **not** set this and does not set an error status - per "
+ "semantic conventions a client error is not a server span's failure.",
+ optional=True,
+)
+
+DB_SYSTEM = Attribute("db.system", "Always ``sqlite``.")
+DB_NAMESPACE = Attribute("db.namespace", "Name of the database being queried.")
+DB_QUERY_TEXT = Attribute(
+ "db.query.text",
+ "The SQL, truncated to 2048 characters. Never the parameter values.",
+)
+DB_OPERATION_NAME = Attribute(
+ "db.operation.name",
+ "The statement's leading keyword - ``SELECT``, ``INSERT``, ``CREATE``, and "
+ "so on - matched against a small fixed allowlist. Omitted rather than set "
+ "to an arbitrary value: the allowlist exists because this attribute is a "
+ "candidate dimension for a query-duration metric in a later phase, and "
+ "echoing an unrecognised first token from user-supplied SQL would be an "
+ "unbounded-cardinality hazard. Also omitted for "
+ "``execute_write_script()``, which runs multiple statements - per "
+ "semantic conventions, the operation name should not be extracted from "
+ "query text that can contain more than one operation. Note that a "
+ "statement beginning with a CTE reports ``WITH``, not the operation "
+ "inside it - a substantial share of Datasette's own reads take that "
+ "form. Resolving it further would mean parsing.",
+ optional=True,
+)
+DB_COLLECTION_NAME = Attribute(
+ "db.collection.name",
+ "The primary table, set only where the view already knows it - the table "
+ "and row pages. Omitted for arbitrary ``?sql=`` queries, where determining "
+ "the table would mean parsing the query.",
+ optional=True,
+)
+
+PARAM_COUNT = Attribute(
+ "datasette.param_count",
+ "Number of bound parameters. Recorded instead of the values themselves.",
+ optional=True,
+)
+PARAM_SETS = Attribute(
+ "datasette.param_sets",
+ "Number of parameter sets consumed by ``execute_write_many()``. Not a row "
+ "count - ``executemany()`` returns no rows. The parameter values "
+ "themselves are never recorded: that sequence can hold thousands of rows.",
+ optional=True,
+)
+TIME_LIMIT_MS = Attribute(
+ "datasette.time_limit_ms",
+ "The :ref:`setting_sql_time_limit_ms` value this query ran under. Set on "
+ "reads, which are the queries that time limit applies to.",
+ optional=True,
+)
+ROWS_RETURNED = Attribute(
+ "datasette.rows_returned",
+ "Number of rows a read returned. Set on the read path only, and only when "
+ "the read succeeded.",
+ optional=True,
+)
+TRUNCATED = Attribute(
+ "datasette.truncated",
+ "True if the result was cut short by :ref:`setting_max_returned_rows`.",
+ optional=True,
+)
+INTERRUPTED = Attribute(
+ "datasette.interrupted",
+ "True if the query was cancelled for exceeding the time limit. The span "
+ "status is also set to ``ERROR``, unless the caller asked for a budget "
+ "shorter than :ref:`setting_sql_time_limit_ms` - as table counts, facet "
+ "suggestion and autocomplete all do - in which case running out of time "
+ "is an expected answer rather than a failure and the status is left "
+ "unset.",
+ optional=True,
+)
+SQL_ERROR_SUPPRESSED = Attribute(
+ "datasette.sql_error_suppressed",
+ "True when the query failed but the caller passed ``log_sql_errors=False``, "
+ "meaning it was probing and treats failure as an expected answer. Facet "
+ "suggestion does this against every column.",
+ optional=True,
+)
+EXECUTESCRIPT = Attribute(
+ "datasette.executescript",
+ "True for ``execute_write_script()``, which runs multiple statements.",
+ optional=True,
+)
+EXECUTEMANY = Attribute(
+ "datasette.executemany",
+ "True for ``execute_write_many()``, which runs one statement against many "
+ "parameter sets.",
+ optional=True,
+)
+ISOLATED_CONNECTION = Attribute(
+ "datasette.isolated_connection",
+ "True if the write ran on its own connection rather than the shared write "
+ "connection.",
+)
+TRANSACTION = Attribute(
+ "datasette.transaction",
+ "False for statements such as ``VACUUM`` that cannot run inside a transaction.",
+)
+
+
+# --- Spans ----------------------------------------------------------------
+
+HTTP_REQUEST = SpanName(
+ "{http.request.method} {http.route}",
+ "One span per HTTP request, created by the outermost layer of the ASGI "
+ "stack - so plugin ``asgi_wrapper()`` middleware, CSRF protection and "
+ "every database span raised while serving the request all nest inside "
+ "it. Without it each of those would be its own root trace. The span name "
+ "is not a fixed string: it is the method followed by the matched route, "
+ "and just the method for a request that matched no route. The span starts "
+ "at the ASGI edge, before routing has happened, so it is named for the "
+ "method there and renamed once the route is known. "
+ "W3C ``traceparent`` and ``baggage`` headers are extracted using the "
+ "global propagator, so a request arriving from an already-traced caller "
+ "continues that trace; set ``OTEL_PROPAGATORS=none`` to turn that off, "
+ "and strip those headers at your proxy if your instance is public.",
+ (
+ HTTP_REQUEST_METHOD,
+ HTTP_ROUTE,
+ URL_PATH,
+ URL_SCHEME,
+ SERVER_ADDRESS,
+ USER_AGENT_ORIGINAL,
+ HTTP_RESPONSE_STATUS_CODE,
+ ERROR_TYPE,
+ ),
+ dynamic=True,
+ kind=SpanKind.SERVER,
+)
+
+DB_QUERY = SpanName(
+ "db.query",
+ "A SQL operation issued by Datasette, covering the full round trip "
+ "including any time spent queued for a thread.",
+ (
+ DB_SYSTEM,
+ DB_NAMESPACE,
+ DB_QUERY_TEXT,
+ DB_OPERATION_NAME,
+ DB_COLLECTION_NAME,
+ PARAM_COUNT,
+ PARAM_SETS,
+ TIME_LIMIT_MS,
+ ROWS_RETURNED,
+ TRUNCATED,
+ INTERRUPTED,
+ SQL_ERROR_SUPPRESSED,
+ EXECUTESCRIPT,
+ EXECUTEMANY,
+ ),
+ kind=SpanKind.CLIENT,
+)
+
+DB_QUERY_EXECUTE = SpanName(
+ "db.query.execute",
+ "The read executing inside a SQL worker thread. Child of ``db.query``; the "
+ "gap between the two is time spent waiting for a thread.",
+)
+
+DB_WRITE_QUEUE_WAIT = SpanName(
+ "db.write.queue_wait",
+ "Time a write spent waiting in its database's write queue before the write "
+ "thread picked it up. Child of ``db.query`` for a ``block=True`` write, "
+ "where the caller awaits the write and containment is accurate. For a "
+ "``block=False`` write the caller does not await it - the enqueueing "
+ "request *caused* the write without *containing* it, and the write's "
+ "spans can outlive the request's own - so this is a root span instead, "
+ "carrying an OpenTelemetry link back to the enqueueing span rather than "
+ "a parent. A link records causation without asserting containment, which "
+ "is exactly the distinction here.",
+)
+
+DB_WRITE_EXECUTE = SpanName(
+ "db.write.execute",
+ "The write executing on the write thread. Child of ``db.query`` for a "
+ "``block=True`` write; for ``block=False`` a root span with a link back "
+ "to the enqueueing span instead - see ``db.write.queue_wait`` above.",
+ (ISOLATED_CONNECTION, TRANSACTION),
+)
+
+STARTUP = SpanName(
+ "datasette.startup",
+ "``invoke_startup()`` running: ``register_events``, ``register_actions``, "
+ "``register_column_types``, ``prepare_jinja2_environment``, internal-database "
+ "schema catalog refresh (including the ``prepare_connection`` warm-up this "
+ "triggers for each database touched for the first time), saved queries, "
+ "column type config and the ``startup`` hook. Runs once per process, before "
+ "any request exists, so without this span every child it creates would be "
+ "its own orphan root trace. A connection warmed later - lazily, the first "
+ "time a *request* touches a new database or thread - nests under that "
+ "request's own span instead, not under this one, since this span has "
+ "already ended by then.",
+)
+
+SPANS = (
+ HTTP_REQUEST,
+ DB_QUERY,
+ DB_QUERY_EXECUTE,
+ DB_WRITE_QUEUE_WAIT,
+ DB_WRITE_EXECUTE,
+ STARTUP,
+)
+
+
+def span_for(emitted_name, kind=None):
+ """
+ Resolve an emitted span name to its registry entry, or None.
+
+ Handles the two entry kinds whose emitted names are not knowable in
+ advance:
+
+ - `prefix=True` - the name carries a variable suffix, matched by prefix.
+ Phase 1 registers none.
+ - `dynamic=True` - the name has no fixed part at all, so it is matched on
+ `kind` instead and the caller has to supply one. Exact and prefix
+ entries are tried first, so a dynamic entry can never shadow a span
+ that does have a registered name.
+ """
+ for span in SPANS:
+ if span.dynamic:
+ continue
+ if span.prefix:
+ if emitted_name.startswith(span):
+ return span
+ elif emitted_name == span:
+ return span
+ if kind is not None:
+ for span in SPANS:
+ if span.dynamic and span.kind == kind:
+ return span
+ return None
+
+
+def attribute_allowed(span, emitted_key):
+ "Whether `emitted_key` is a registered attribute of `span`."
+ if span is None:
+ return False
+ return emitted_key in span.attributes
diff --git a/datasette/tracer.py b/datasette/tracer.py
deleted file mode 100644
index 1fbda6f9..00000000
--- a/datasette/tracer.py
+++ /dev/null
@@ -1,156 +0,0 @@
-import asyncio
-import json
-import time
-import traceback
-from contextlib import contextmanager
-from contextvars import ContextVar
-
-from markupsafe import escape
-
-tracers = {}
-
-TRACE_RESERVED_KEYS = {"type", "start", "end", "duration_ms", "traceback"}
-
-trace_task_id = ContextVar("trace_task_id", default=None)
-
-
-def get_task_id():
- current = trace_task_id.get(None)
- if current is not None:
- return current
- try:
- loop = asyncio.get_event_loop()
- except RuntimeError:
- return None
- return id(asyncio.current_task(loop=loop))
-
-
-@contextmanager
-def trace_child_tasks():
- token = trace_task_id.set(get_task_id())
- try:
- yield
- finally:
- trace_task_id.reset(token)
-
-
-@contextmanager
-def trace(trace_type, **kwargs):
- assert not TRACE_RESERVED_KEYS.intersection(
- kwargs.keys()
- ), f".trace() keyword parameters cannot include {TRACE_RESERVED_KEYS}"
- task_id = get_task_id()
- if task_id is None:
- yield kwargs
- return
- tracer = tracers.get(task_id)
- if tracer is None:
- yield kwargs
- return
- start = time.perf_counter()
- captured_error = None
- try:
- yield kwargs
- except Exception as ex:
- captured_error = ex
- raise
- finally:
- end = time.perf_counter()
- trace_info = {
- "type": trace_type,
- "start": start,
- "end": end,
- "duration_ms": (end - start) * 1000,
- "traceback": traceback.format_list(traceback.extract_stack(limit=6)[:-3]),
- "error": str(captured_error) if captured_error else None,
- }
- trace_info.update(kwargs)
- tracer.append(trace_info)
-
-
-@contextmanager
-def capture_traces(tracer):
- # tracer is a list
- task_id = get_task_id()
- if task_id is None:
- yield
- return
- tracers[task_id] = tracer
- yield
- del tracers[task_id]
-
-
-class AsgiTracer:
- # If the body is larger than this we don't attempt to append the trace
- max_body_bytes = 1024 * 256 # 256 KB
-
- def __init__(self, app):
- self.app = app
-
- async def __call__(self, scope, receive, send):
- if b"_trace=1" not in scope.get("query_string", b"").split(b"&"):
- await self.app(scope, receive, send)
- return
- trace_start = time.perf_counter()
- traces = []
-
- accumulated_body = b""
- size_limit_exceeded = False
- response_headers = []
-
- async def wrapped_send(message):
- nonlocal accumulated_body, size_limit_exceeded, response_headers
-
- if message["type"] == "http.response.start":
- response_headers = message["headers"]
- await send(message)
- return
-
- if message["type"] != "http.response.body" or size_limit_exceeded:
- await send(message)
- return
-
- # Accumulate body until the end or until size is exceeded
- accumulated_body += message["body"]
- if len(accumulated_body) > self.max_body_bytes:
- # Send what we have accumulated so far
- await send(
- {
- "type": "http.response.body",
- "body": accumulated_body,
- "more_body": bool(message.get("more_body")),
- }
- )
- size_limit_exceeded = True
- return
-
- if not message.get("more_body"):
- # We have all the body - modify it and send the result
- # TODO: What to do about Content-Type or other cases?
- trace_info = {
- "request_duration_ms": 1000 * (time.perf_counter() - trace_start),
- "sum_trace_duration_ms": sum(t["duration_ms"] for t in traces),
- "num_traces": len(traces),
- "traces": traces,
- }
- content_type = next(
- (
- v.decode("utf8")
- for k, v in response_headers
- if k.lower() == b"content-type"
- ),
- "",
- )
- if "text/html" in content_type and b"
" in accumulated_body:
- extra = escape(json.dumps(trace_info, indent=2))
- extra_html = f"