remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-11-07 18:58:29 +01:00
Code Issues Projects Releases Packages Wiki Activity

Add OpenTelemetry metrics for SQL thread pool saturation and query latency

Browse source 
Spans describe requests that have finished. They structurally cannot answer
"am I saturating my 3 SQL threads right now", because that is a level rather
than an event - and with num_sql_threads defaulting to 3, it is usually the
first thing worth knowing about a busy Datasette. This adds the metrics that
answer it.

Five observable gauges, computed only when something is collecting, so an
instance with no MeterProvider installed does no work for them at all:

  datasette.sql.threads.limit         num_sql_threads
  datasette.sql.threads.queue_depth   queries waiting for a free thread
  datasette.sql.queries.pending       in-flight reads, by db.namespace
  datasette.write.queue_depth         writes behind the single write thread
  datasette.connections.open          tracked file connections

Three instruments recorded inline, which matters because metrics survive
trace sampling and spans do not - an operator sampling 1% of traces still
gets 100% of the latency distribution:

  db.client.operation.duration        semconv histogram, with error.type
  datasette.write.queue_wait          the metric twin of the existing span
  datasette.sql.queries.interrupted   sql_time_limit_ms kills

The interrupted counter closes a gap the plan called out as unanswerable:
"how often are we killing queries at the limit" is a rate, and a rate cannot
be recovered from sampled spans.

Core still creates no provider of any kind, so the architecture is unchanged;
`grep -rn 'opentelemetry.sdk' datasette/` stays empty. One real difference
from tracing is worth recording: _ProxyMeter and its instruments forward to a
provider installed after they were created, whereas ProxyTracer permanently
caches the first concrete tracer it resolves. Module-level instruments are
therefore safe and the test fixture has no ordering constraint.

Live instances are tracked in a lock-guarded WeakSet so instrumenting an
instance never keeps it alive. The pool gauges carry no attribute saying
which Datasette produced them: production runs one instance per process, and
adding an id to disambiguate the test suite's hundreds of instances would buy
unbounded attribute cardinality to fix a case that does not occur. The
collision is documented instead, and the gauge callbacks are plain generator
functions so tests can assert exact values by calling them directly rather
than through the SDK's last-value aggregation.

demos/otel/metrics_demo.py fires 12 concurrent 40ms queries at a 3-thread
pool and samples the gauges mid-flight: queue_depth peaks at exactly 9, and
the duration histogram reads max=0.1695s for a query whose work is 40ms. That
gap is the queue, and it is the thing traces alone will not show you.

Also corrects the demo README's privacy section, which still claimed
parameter values are never recorded - that stopped being unconditionally true
when trace_sql_parameters landed.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

(cherry picked from 6ef0dd8c and adapted to the rebuilt phase-1 stack:
attribute names now come from telemetry_registry where entries exist, the
meter carries the instrumentation-scope version and schema URL, and the
interrupted-queries counter skips expected timeouts - callers that opted
into a deliberately short budget, like facet suggestion - matching how
those are excluded from span error status. The internals.rst reference
lands with the registry commit that follows.)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01F2h9ANGZ7paWSpqs5DUAcG
This commit is contained in:
Alex Garcia 2026-07-30 09:43:30 -07:00 • committed by GitHub
commit 90b727db73
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 730 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

10
datasette/app.py
View file

@ -53,8 +53,10 @@ from .telemetry import (
TelemetryMiddleware,
_in_datasette_client,
clamp_http_method,
register_datasette,
request_span,
tracer,
unregister_datasette,
)
from .telemetry_registry import HTTP_ROUTE, STARTUP
from .tokens import TokenInvalid
@ -651,6 +653,10 @@ class Datasette:
self.root_enabled = False
self.default_deny = default_deny
self.client = DatasetteClient(self)
# Last, so that the observable-gauge callbacks - which may fire on the
# SDK's collection thread the instant this returns - never see a
# half-built instance.
register_datasette(self)
async def apply_metadata_json(self):
# Apply any metadata entries from metadata.json to the internal tables
@ -990,6 +996,10 @@ class Datasette:
if self._closed:
return
self._closed = True
# Stop reporting gauges before tearing anything down, so a collection
# cycle landing mid-close cannot observe a half-closed instance. The
# WeakSet would drop it eventually anyway; this makes it immediate.
unregister_datasette(self)
first_exception = None
dbs = list(self.databases.values()) + [self._internal_database]
for db in dbs:

48
datasette/database.py
View file

@ -17,7 +17,15 @@ 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 callback_name, sql_attribute, sql_operation_name, tracer
from .telemetry import (
callback_name,
record_operation_duration,
record_query_interrupted,
record_write_queue_wait,
sql_attribute,
sql_operation_name,
tracer,
)
from .telemetry_registry import (
CALLBACK,
DB_NAMESPACE,
@ -316,9 +324,10 @@ class Database:
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
)
with record_operation_duration(self.name, "write"):
results = await self._execute_write_fn(
_inner, block=block, request=request, transaction=transaction
)
return results
async def execute_write_script(self, sql, block=True, request=None):
@ -340,9 +349,10 @@ class Database:
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
)
with record_operation_duration(self.name, "write"):
results = await self._execute_write_fn(
_inner, block=block, transaction=False, request=request
)
return results
async def execute_write_many(self, sql, params_seq, block=True, request=None):
@ -373,9 +383,10 @@ class Database:
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
)
with record_operation_duration(self.name, "write"):
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)
@ -665,11 +676,15 @@ class Database:
# 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.
dequeued_at_ns = time.time_ns()
tracer.start_span(
DB_WRITE_QUEUE_WAIT,
start_time=task.enqueued_at_ns,
**write_span_kwargs,
).end(end_time=time.time_ns())
).end(end_time=dequeued_at_ns)
record_write_queue_wait(
self.name, dequeued_at_ns - task.enqueued_at_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.
@ -915,7 +930,8 @@ class Database:
if params:
span.set_attribute(PARAM_COUNT, len(params))
try:
results = await self._execute_fn(sql_operation_in_thread)
with record_operation_duration(self.name, "read"):
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
@ -924,6 +940,14 @@ class Database:
if not timeout_expected:
span.set_status(Status(StatusCode.ERROR, str(e)))
span.record_exception(e)
# A counter rather than only a span, because this is the
# one thing an operator wants a rate and an alert on, and
# spans under a 1% sampler cannot provide either. An
# expected timeout - a caller that opted into a shorter
# budget, like facet suggestion - is not counted, for the
# same reason it is not a span error: it fires routinely
# by design and would drown the signal this exists for.
record_query_interrupted(self.name)
raise
except Exception as e:
# log_sql_errors=False means the caller is probing and

264
datasette/telemetry.py
View file

@ -2,27 +2,36 @@
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
`TracerProvider` or a `MeterProvider`, 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 ~100 spans -
but end-to-end page benchmarks put the overhead below their own
run-to-run variation. Installing an SDK provider is what costs
something measurable.
something measurable. Every metric instrument is likewise a no-op
without a provider, and the observable-gauge callbacks are never
invoked at all.
"""
import contextvars
import re
import threading
import time
import weakref
from contextlib import contextmanager
from opentelemetry import metrics as otel_metrics
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 (
DB_NAMESPACE,
DB_SYSTEM,
ERROR_TYPE,
HTTP_REQUEST_METHOD,
HTTP_RESPONSE_STATUS_CODE,
@ -64,6 +73,7 @@ _in_datasette_client = contextvars.ContextVar("in_datasette_client", default=Fal
SCHEMA_URL = "https://opentelemetry.io/schemas/1.29.0"
tracer = otel_trace.get_tracer("datasette", __version__, schema_url=SCHEMA_URL)
meter = otel_metrics.get_meter("datasette", __version__, schema_url=SCHEMA_URL)
MAX_SQL_LENGTH = 2048
@ -359,3 +369,249 @@ class TelemetryMiddleware:
if status >= 500 and not escaped:
span.set_status(Status(StatusCode.ERROR))
span.set_attribute(ERROR_TYPE, str(status))
# --- Metrics --------------------------------------------------------------
#
# Spans answer "what happened during this request". They cannot answer "am I
# saturating my 3 SQL threads right now", because that is a gauge: a level
# sampled at collection time, not an event with a duration. It is also the
# single most useful operational question about a Datasette deployment, since
# num_sql_threads defaults to 3 and every read query in the process competes
# for those threads.
#
# Two shapes are used here:
#
# Observable gauges - a callback the SDK invokes on its own collection
# cycle. Nothing is computed unless something is collecting, so the default
# no-provider install pays literally nothing for them.
#
# Synchronous histograms/counters - recorded inline on the query path. These
# survive trace sampling, which spans do not: an operator sampling 1% of
# traces still gets 100% of the latency distribution and the interrupted
# count.
#
# Note a real difference from tracing: `_ProxyMeter` and its instruments
# forward to a provider installed *after* they were created, whereas
# `ProxyTracer` permanently caches the concrete tracer it first resolves. So
# module-level instruments here are safe, and tests do not need a provider
# installed before this module is imported.
def _duration_attributes(database_name, operation):
return {
DB_SYSTEM: "sqlite",
DB_NAMESPACE: database_name,
"datasette.operation": operation,
}
sql_operation_duration = meter.create_histogram(
"db.client.operation.duration",
unit="s",
description="Duration of a SQL operation issued by Datasette",
)
write_queue_wait = meter.create_histogram(
"datasette.write.queue_wait",
unit="s",
description=(
"Time a write spent queued behind the single write thread for its database"
),
)
queries_interrupted = meter.create_counter(
"datasette.sql.queries.interrupted",
unit="{query}",
description=(
"Queries cancelled for exceeding sql_time_limit_ms. Not derivable from "
"spans under sampling, and the signal that a time limit is too tight"
),
)
@contextmanager
def record_operation_duration(database_name, operation):
"""
Record `db.client.operation.duration` for one SQL operation.
`error.type` is set from the exception class on failure, per semconv, so a
latency distribution can be split by success and failure. For a
`block=False` write this measures the enqueue, not the write - the same
caveat that applies to the surrounding span.
"""
attributes = _duration_attributes(database_name, operation)
started = time.perf_counter()
try:
yield
except BaseException as exception:
attributes[ERROR_TYPE] = type(exception).__qualname__
raise
finally:
sql_operation_duration.record(time.perf_counter() - started, attributes)
def record_write_queue_wait(database_name, waited_ns):
write_queue_wait.record(waited_ns / 1e9, {DB_NAMESPACE: database_name})
def record_query_interrupted(database_name):
queries_interrupted.add(1, {DB_NAMESPACE: database_name})
# Live Datasette instances, weakly held so that instrumenting an instance
# never keeps it alive. Guarded by a lock because the gauge callbacks run on
# the SDK's collection thread while the event loop may be building or closing
# a Datasette.
#
# Known limitation: the pool gauges below carry no attribute identifying which
# Datasette produced them, so if a single process runs more than one instance
# their observations collide and last-one-wins. Production runs one instance
# per process; adding an instance id to make the test suite's hundreds of
# instances distinguishable would mean unbounded attribute cardinality in
# exchange for fixing a case that does not occur in production.
_live_datasettes = weakref.WeakSet()
_live_datasettes_lock = threading.Lock()
def register_datasette(ds):
"Start reporting pool/queue gauges for this Datasette instance."
with _live_datasettes_lock:
_live_datasettes.add(ds)
def unregister_datasette(ds):
"Stop reporting gauges for an instance that has been closed."
with _live_datasettes_lock:
_live_datasettes.discard(ds)
def _live_instances():
with _live_datasettes_lock:
return list(_live_datasettes)
def _databases_of(ds):
"""
Every Database attached to an instance, including the internal database.
The internal database is deliberately included: permission checks run SQL
against it on essentially every request, so its queue depth and connection
count are as operationally interesting as any user database's.
"""
databases = list(ds.databases.values())
internal = getattr(ds, "_internal_database", None)
if internal is not None:
databases.append(internal)
return databases
# Each callback is a plain generator function so it can be unit-tested
# directly, without standing up an SDK provider and a metric reader.
def observe_sql_thread_limit(options=None):
"Size of the shared read-query thread pool (the num_sql_threads setting)."
for ds in _live_instances():
if ds.executor is None:
# num_sql_threads=0 - queries run on the event loop, no pool.
continue
yield otel_metrics.Observation(ds.setting("num_sql_threads"), {})
def observe_sql_thread_queue_depth(options=None):
"""
Read queries waiting for a free thread in the shared pool.
This is the saturation signal: sustained above zero means requests are
queueing on num_sql_threads. `_work_queue` is a private attribute of
ThreadPoolExecutor, so its absence is tolerated rather than fatal - a
missing gauge is much better than a crashed collection cycle.
"""
for ds in _live_instances():
if ds.executor is None:
continue
work_queue = getattr(ds.executor, "_work_queue", None)
if work_queue is None:
continue
yield otel_metrics.Observation(work_queue.qsize(), {})
def observe_pending_queries(options=None):
"""
Read queries submitted to the pool and not yet finished, per database.
Summed across databases and compared against the thread limit, this is the
utilisation half of the saturation picture. `len()` is deliberately taken
without `_pending_execute_futures_lock`: it is atomic, and taking a lock
held on the request path from the collection thread would let telemetry
add latency to queries.
"""
for ds in _live_instances():
for db in _databases_of(ds):
yield otel_metrics.Observation(
len(db._pending_execute_futures), {DB_NAMESPACE: db.name}
)
def observe_write_queue_depth(options=None):
"""
Writes queued behind the single write thread, per database.
Every database serialises its writes through one thread, so this is
unbounded backpressure that no amount of num_sql_threads will relieve.
"""
for ds in _live_instances():
for db in _databases_of(ds):
write_queue = db._write_queue
if write_queue is None:
# No write has ever been queued for this database.
continue
yield otel_metrics.Observation(
write_queue.qsize(), {DB_NAMESPACE: db.name}
)
def observe_open_connections(options=None):
"Open SQLite connections tracked for closing, per database."
for ds in _live_instances():
for db in _databases_of(ds):
yield otel_metrics.Observation(
len(db._all_connections), {DB_NAMESPACE: db.name}
)
sql_thread_limit_gauge = meter.create_observable_gauge(
"datasette.sql.threads.limit",
callbacks=[observe_sql_thread_limit],
unit="{thread}",
description="Maximum concurrent read queries (the num_sql_threads setting)",
)
sql_thread_queue_depth_gauge = meter.create_observable_gauge(
"datasette.sql.threads.queue_depth",
callbacks=[observe_sql_thread_queue_depth],
unit="{query}",
description="Read queries waiting for a free thread in the shared SQL pool",
)
pending_queries_gauge = meter.create_observable_gauge(
"datasette.sql.queries.pending",
callbacks=[observe_pending_queries],
unit="{query}",
description="Read queries submitted to the pool and not yet complete",
)
write_queue_depth_gauge = meter.create_observable_gauge(
"datasette.write.queue_depth",
callbacks=[observe_write_queue_depth],
unit="{write}",
description="Writes queued behind a database's single write thread",
)
open_connections_gauge = meter.create_observable_gauge(
"datasette.connections.open",
callbacks=[observe_open_connections],
unit="{connection}",
description="Open SQLite connections tracked for closing",
)