diff --git a/.github/workflows/deploy-latest.yml b/.github/workflows/deploy-latest.yml
index b7f2361f..3fc83438 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 --crossdb --root" \
+ --extra-options="--setting template_debug 1 --setting trace_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 8cee9b74..42be7425 100644
--- a/datasette/app.py
+++ b/datasette/app.py
@@ -49,14 +49,8 @@ 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,
@@ -293,6 +287,11 @@ 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"
@@ -779,73 +778,57 @@ class Datasette:
# This must be called for Datasette to be in a usable state
if self._startup_invoked:
return
- # `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 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)
@@ -2861,6 +2844,8 @@ 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],
@@ -2869,12 +2854,6 @@ 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
@@ -2955,26 +2934,8 @@ 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 efb8c56e..e162d34e 100644
--- a/datasette/database.py
+++ b/datasette/database.py
@@ -1,45 +1,19 @@
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 .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 .tracer import trace
from .utils import (
call_with_supported_arguments,
detect_fts,
@@ -283,15 +257,7 @@ class Database:
cursor, return_all=return_all, returning_limit=returning_limit
)
- 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))
+ with trace("sql", database=self.name, sql=sql.strip(), params=params):
results = await self.execute_write_fn(
_inner, block=block, request=request, transaction=transaction
)
@@ -303,15 +269,7 @@ class Database:
def _inner(conn):
return conn.executescript(sql)
- # 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)
+ with trace("sql", database=self.name, sql=sql.strip(), executescript=True):
results = await self.execute_write_fn(
_inner, block=block, transaction=False, request=request
)
@@ -331,22 +289,13 @@ class Database:
return conn.executemany(sql, count_params(params_seq)), count
- 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)
+ with trace(
+ "sql", database=self.name, sql=sql.strip(), executemany=True
+ ) as kwargs:
results, count = await self.execute_write_fn(
_inner, block=block, request=request
)
- # count is the number of parameter *sets* consumed by
- # executemany(), not a row count - executemany returns no rows.
- span.set_attribute(PARAM_SETS, count)
+ kwargs["count"] = count
return results
async def execute_isolated_fn(self, fn):
@@ -372,18 +321,9 @@ 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.
- # 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()
+ # write queue to block; run against a fresh read-only connection
return await asyncio.get_running_loop().run_in_executor(
- self.ds.executor, ctx.run, _run
+ self.ds.executor, _run
)
# Threaded mode - send to write thread
return await self._send_to_write_thread(fn, isolated_connection=True)
@@ -488,24 +428,8 @@ 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,
- otel_context_api.get_current(),
- time.time_ns(),
- block,
- )
+ WriteTask(fn, task_id, loop, reply_future, isolated_connection, transaction)
)
if block:
return await reply_future
@@ -519,16 +443,6 @@ 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
@@ -543,119 +457,40 @@ class Database:
# Best-effort close as the write thread exits
pass
return
- # `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)
+ 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
else:
- 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)
+ 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)
async def execute_fn(self, fn):
self._check_not_closed()
@@ -677,28 +512,7 @@ class Database:
with self._pending_execute_futures_lock:
self._check_not_closed()
- # 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)
+ future = self.ds.executor.submit(in_thread)
self._pending_execute_futures.add(future)
future.add_done_callback(self._remove_pending_execute_future)
return await asyncio.wrap_future(future)
@@ -711,143 +525,48 @@ class Database:
custom_time_limit=None,
page_size=None,
log_sql_errors=True,
- table=None,
):
- """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.
- """
+ """Executes sql against db_name in a thread"""
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):
- # 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:
+ 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):
try:
- 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:
+ 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:
- execute_span.record_exception(e)
- execute_span.set_status(Status(StatusCode.ERROR, str(e)))
+ sys.stderr.write(
+ f"ERROR: conn={conn}, sql = {sql!r}, params = {params}: {e}\n"
+ )
+ sys.stderr.flush()
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)
- # 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))
+ with trace("sql", database=self.name, sql=sql.strip(), params=params):
+ results = await self.execute_fn(sql_operation_in_thread)
return results
@property
@@ -1135,28 +854,16 @@ 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,
- otel_context,
- enqueued_at_ns,
- block,
+ self, fn, task_id, loop, reply_future, isolated_connection, transaction
):
self.fn = fn
self.task_id = task_id
@@ -1164,14 +871,6 @@ 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
deleted file mode 100644
index 311d661f..00000000
--- a/datasette/telemetry.py
+++ /dev/null
@@ -1,347 +0,0 @@
-"""
-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
deleted file mode 100644
index 4e750605..00000000
--- a/datasette/telemetry_registry.py
+++ /dev/null
@@ -1,388 +0,0 @@
-"""
-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
new file mode 100644
index 00000000..1fbda6f9
--- /dev/null
+++ b/datasette/tracer.py
@@ -0,0 +1,156 @@
+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"