remote/datasette
1
0
Fork 0
mirror of https://github.com/simonw/datasette.git synced 2026-09-03 07:04:16 +02: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
commit 8b35d4a80d
8 changed files with 954 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

10
datasette/app.py
View file

@ -52,8 +52,10 @@ from .resources import DatabaseResource, TableResource
from .telemetry import (
TelemetryMiddleware,
clamp_http_method,
register_datasette,
request_span,
tracer,
unregister_datasette,
)
from .telemetry_registry import HTTP_ROUTE, STARTUP
from .tokens import TokenInvalid
@ -639,6 +641,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
@ -978,6 +984,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:

47
datasette/database.py
View file

@ -17,7 +17,14 @@ 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 import (
record_operation_duration,
record_query_interrupted,
record_write_queue_wait,
sql_attribute,
sql_operation_name,
tracer,
)
from .telemetry_registry import (
DB_COLLECTION_NAME,
DB_NAMESPACE,
@ -292,9 +299,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):
@ -312,9 +320,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):
@ -341,9 +350,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)
@ -596,11 +606,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.
@ -822,7 +836,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
@ -831,6 +846,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,9 +2,9 @@
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
@ -12,17 +12,26 @@ With no provider installed every span produced here is a
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.
what costs something measurable. Every metric instrument is likewise a
no-op without a provider, and the observable-gauge callbacks are never
invoked at all.
"""
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,
@ -55,6 +64,7 @@ from .version import __version__
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
@ -345,3 +355,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 file connections tracked for closing, per database."
for ds in _live_instances():
for db in _databases_of(ds):
yield otel_metrics.Observation(
len(db._all_file_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 file connections tracked for closing",
)

55
demos/otel/README.md
View file

@ -1,16 +1,18 @@
# OpenTelemetry demo
Datasette core depends on `opentelemetry-api` only. It emits spans and nothing else — it never
creates a `TracerProvider`, never configures an exporter, and never sets a sampler. With no SDK
installed every span is a no-op and costs approximately nothing.
Datasette core depends on `opentelemetry-api` only. It emits spans and metrics and nothing
else — it never creates a `TracerProvider` or a `MeterProvider`, never configures an exporter,
and never sets a sampler. With no SDK installed every span and every instrument is a no-op and
costs approximately nothing.
That means "turning tracing on" is entirely the job of whoever runs Datasette. This directory
shows two ways to do it, **neither of which needs Docker**:
That means "turning telemetry on" is entirely the job of whoever runs Datasette. This directory
shows several ways to do it, **none of which needs Docker**:
| | |
|---|---|
| `otlp_receiver.py` | A ~150 line pure-Python OTLP/HTTP receiver — a real protobuf export, summarized in your terminal |
| `just jaeger` | The same export into Jaeger's own binary, for a real trace UI |
| `metrics_demo.py` | Saturates the SQL thread pool and prints the gauges — the question spans cannot answer |
Both listen for OTLP/HTTP on port 4318, so the Datasette side is identical — run one or the
other, not both. The `Justfile` in this directory wraps every command below; bare `just` lists
@ -97,6 +99,49 @@ request produces exactly two traces — the request trace (~67 spans, rooted at
span, with every `db.query` nested inside it across thread boundaries) and the
`datasette.startup` trace (~26 spans of catalog queries and connection warm-up).
## 3. The question spans cannot answer
```bash
uv run python demos/otel/metrics_demo.py
```
A trace tells you a query took 170ms. It does not tell you that 130ms of that was spent waiting for
one of only three threads — "how many threads are busy right now" is a level, not an event, so no
span can carry it. That is what metrics are for.
The script registers a SQL function that sleeps, fires 12 concurrent 40ms queries at a pool of 3
threads, and samples the gauges while they are in flight. Real output:
```
num_sql_threads = 3, firing 12 concurrent 40ms queries
wall clock : 170ms
if fully serialised : 480ms
with 3 threads perfectly used : 160ms
Peak values sampled while the queries were in flight:
datasette.sql.queries.pending {db.namespace=demo} = 12
datasette.sql.threads.queue_depth = 9
Final collection:
datasette.sql.threads.limit
3
db.client.operation.duration {datasette.operation=read, db.namespace=demo, db.system=sqlite}
count=16 sum=1.2605s min=0.0001s max=0.1695s
datasette.write.queue_wait {db.namespace=demo}
count=1 sum=0.0002s min=0.0002s max=0.0002s
```
`queue_depth = 9` is the whole point: 12 queries, 3 threads, 9 of them sitting in a queue. Sustained
above zero in production means requests are backing up on `num_sql_threads`, and no amount of
reading traces would have told you that.
Note also `max=0.1695s` on the duration histogram against a query whose actual work is 40ms. The
gap is queue time. The two numbers together — 170ms observed, 40ms of work — are what distinguishes
"my queries are slow" from "my pool is too small".
## Notes on the environment variables
- **`opentelemetry-instrument` is required.** Setting `OTEL_TRACES_EXPORTER` and running plain

175
demos/otel/metrics_demo.py Normal file
View file

@ -0,0 +1,175 @@
"""
Saturate Datasette's SQL thread pool and print the metrics that show it.
Run it with no arguments and no infrastructure:
uv run python demos/otel/metrics_demo.py
It builds a small database in a temporary directory, registers a deliberately
slow SQL function, installs an in-memory OpenTelemetry SDK metric reader, then
fires more concurrent queries than there are threads in the pool while
sampling the gauges in the background.
The point is the question spans cannot answer. A trace tells you a query took
170ms; it does not tell you that 130ms of that was spent waiting for one of
only three threads, because "how many threads are busy right now" is a level
rather than an event. That is what `datasette.sql.threads.queue_depth` reports.
Datasette core never installs a `MeterProvider` - the block near the top of
this file is doing the job that `opentelemetry-instrument` would normally do.
Core only emits, and whoever runs Datasette decides where it goes.
"""
import asyncio
import sqlite3
import sys
import tempfile
import time
from pathlib import Path
try:
from opentelemetry import metrics as otel_metrics
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
except ImportError:
sys.exit(
"This demo needs the OpenTelemetry SDK:\n"
" pip install opentelemetry-sdk\n"
"(Datasette itself only depends on opentelemetry-api.)"
)
reader = InMemoryMetricReader()
otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
from datasette import hookimpl
from datasette.app import Datasette
from datasette.plugins import pm
NUM_SQL_THREADS = 3
CONCURRENT_QUERIES = 12
QUERY_MS = 40
class SlowQueryPlugin:
"Registers a SQL function that sleeps, so queries occupy a pool thread."
__name__ = "slow-query-plugin"
@hookimpl
def prepare_connection(self, conn):
conn.create_function("slow_ms", 1, lambda ms: time.sleep(ms / 1000) or ms)
def build_database(path):
conn = sqlite3.connect(path)
conn.execute("create table t (id integer primary key)")
conn.executemany("insert into t (id) values (?)", [[i] for i in range(100)])
conn.commit()
conn.close()
def collect():
"One collection cycle -> {(metric name, attributes tuple): data point}."
points = {}
data = reader.get_metrics_data()
if data is None:
return points
for resource_metrics in data.resource_metrics:
for scope_metrics in resource_metrics.scope_metrics:
for metric in scope_metrics.metrics:
for point in metric.data.data_points:
key = (metric.name, tuple(sorted((point.attributes or {}).items())))
points[key] = point
return points
async def sample_during_load(stop, peaks):
"Poll the gauges while the load is running and keep the highest seen."
while not stop.is_set():
points = collect()
for name in (
"datasette.sql.threads.queue_depth",
"datasette.sql.queries.pending",
):
for (metric_name, attributes), point in points.items():
if metric_name != name:
continue
current = getattr(point, "value", 0)
key = (name, attributes)
if current > peaks.get(key, -1):
peaks[key] = current
await asyncio.sleep(0.002)
def format_attributes(attributes):
if not attributes:
return ""
return " {" + ", ".join(f"{k}={v}" for k, v in attributes) + "}"
async def main():
tmpdir = Path(tempfile.mkdtemp())
db_path = tmpdir / "demo.db"
build_database(db_path)
pm.register(SlowQueryPlugin(), name="slow-query-plugin")
ds = Datasette([str(db_path)], settings={"num_sql_threads": NUM_SQL_THREADS})
await ds.invoke_startup()
db = ds.get_database("demo")
# Warm up so connection setup and schema introspection do not land in the
# middle of the measurement.
await db.execute("select 1")
print(
f"num_sql_threads = {NUM_SQL_THREADS}, "
f"firing {CONCURRENT_QUERIES} concurrent {QUERY_MS}ms queries\n"
)
peaks = {}
stop = asyncio.Event()
sampler = asyncio.ensure_future(sample_during_load(stop, peaks))
started = time.perf_counter()
await asyncio.gather(
*[db.execute(f"select slow_ms({QUERY_MS})") for _ in range(CONCURRENT_QUERIES)]
)
elapsed_ms = (time.perf_counter() - started) * 1000
stop.set()
await sampler
# A write, so the write-queue metrics have something to report.
await db.execute_write("create table if not exists written (id integer)")
serial_ms = CONCURRENT_QUERIES * QUERY_MS
ideal_ms = serial_ms / NUM_SQL_THREADS
print(f"wall clock : {elapsed_ms:.0f}ms")
print(f" if fully serialised : {serial_ms}ms")
print(f" with {NUM_SQL_THREADS} threads perfectly used : {ideal_ms:.0f}ms\n")
print("Peak values sampled while the queries were in flight:\n")
for (name, attributes), peak in sorted(peaks.items()):
print(f" {name}{format_attributes(attributes)} = {peak}")
print("\nFinal collection:\n")
points = collect()
for (name, attributes), point in sorted(points.items()):
if hasattr(point, "value"):
rendered = str(point.value)
else:
rendered = (
f"count={point.count} sum={point.sum:.4f}s "
f"min={point.min:.4f}s max={point.max:.4f}s"
)
print(f" {name}{format_attributes(attributes)}")
print(f" {rendered}")
print(
"\nThe queue_depth peak is the number of queries that were sitting "
"\nwaiting for a thread. Raising num_sql_threads is what moves it."
)
ds.close()
if __name__ == "__main__":
asyncio.run(main())

1
docs/changelog.rst
View file

@ -11,6 +11,7 @@ Unreleased
- Datasette's database layer now emits `OpenTelemetry <https://opentelemetry.io/>`__ spans: one per query, covering the full round trip including time spent waiting for a SQL worker thread, plus separate child spans for the execution itself and for time spent in the write queue. Datasette core depends on ``opentelemetry-api`` only and never installs an SDK provider, an exporter or a sampler, so there is no effect and no measurable overhead unless tracing is switched on externally - normally with the standard ``opentelemetry-instrument`` agent. See :ref:`internals_telemetry`. (:issue:`1730`)
- :ref:`db.execute(sql, ..., table=None) <database_execute>` has a new optional ``table=`` parameter, naming the table a query is about so it can be recorded on that query's OpenTelemetry span. It has no effect on query execution, and Datasette never derives it from the SQL. (:issue:`1730`)
- Datasette core now also emits OpenTelemetry **metrics** covering SQL thread pool saturation, per-database write queue depth, open connections, query latency and time-limit interruptions. These answer operational questions that spans structurally cannot - "am I saturating my :ref:`setting_num_sql_threads` threads?" is a level, not an event - and they survive trace sampling. As with spans, core installs no ``MeterProvider``, so there is no cost unless metrics are collected externally. See :ref:`internals_telemetry`. (:issue:`1730`)
- **Breaking change:** Datasette's hand-rolled tracer has been removed, now that OpenTelemetry covers the same ground. The ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the ``datasette.tracer`` module are all gone. ``datasette.tracer.trace()`` and ``datasette.tracer.trace_child_tasks()`` were documented plugin APIs, so any plugin importing them will now raise ``ModuleNotFoundError`` and needs a new release. `datasette-pretty-traces <https://datasette.io/plugins/datasette-pretty-traces>`__ does not import that module, but it renders ``?_trace=1`` output, so it no longer has anything to display. (:issue:`1730`)
.. _v1_0_a38:

100
tests/conftest.py
View file

@ -114,6 +114,106 @@ def otel_spans():
yield _otel_span_exporter
_otel_metric_reader = None
@pytest.fixture(scope="session", autouse=True)
def _otel_meter_provider():
"""
Install a real OTel SDK MeterProvider + InMemoryMetricReader once per
process.
Unlike the tracer, ordering is not load-bearing here: `_ProxyMeter` and
the `_ProxyInstrument`s it hands out forward to a provider installed
*after* they were created, whereas `ProxyTracer` permanently caches the
first concrete tracer it resolves. This fixture is still session-scoped
and autouse for symmetry, and so that a single reader collects for the
whole run.
DELTA temporality is chosen for counters and histograms so that each
collection reports only what happened since the previous one. With the
SDK default of CUMULATIVE, every metrics test would see every query run
by every earlier test in the session.
"""
global _otel_metric_reader
try:
from opentelemetry import metrics as otel_metrics
from opentelemetry.sdk.metrics import Counter, Histogram, MeterProvider
from opentelemetry.sdk.metrics.export import (
AggregationTemporality,
InMemoryMetricReader,
)
except ImportError:
return
reader = InMemoryMetricReader(
preferred_temporality={
Counter: AggregationTemporality.DELTA,
Histogram: AggregationTemporality.DELTA,
}
)
otel_metrics.set_meter_provider(MeterProvider(metric_readers=[reader]))
_otel_metric_reader = reader
class MetricsCollector:
"""
Thin reader over an `InMemoryMetricReader`.
`collect()` runs a collection cycle - which is what invokes the observable
gauge callbacks - and snapshots the result. Queries then run against that
snapshot rather than re-collecting, so a test that inspects several
metrics sees one consistent moment and does not drain delta state twice.
"""
def __init__(self, reader):
self.reader = reader
self.snapshot = {}
def collect(self):
self.snapshot = {}
data = self.reader.get_metrics_data()
if data is None:
return self.snapshot
for resource_metrics in data.resource_metrics:
for scope_metrics in resource_metrics.scope_metrics:
for metric in scope_metrics.metrics:
self.snapshot.setdefault(metric.name, []).extend(
metric.data.data_points
)
return self.snapshot
def points(self, name, attributes=None):
"Data points for `name` whose attributes are a superset of `attributes`."
found = []
for point in self.snapshot.get(name, []):
point_attributes = dict(point.attributes or {})
if all(point_attributes.get(k) == v for k, v in (attributes or {}).items()):
found.append(point)
return found
def point(self, name, attributes=None):
"The single matching data point, asserting there is exactly one."
found = self.points(name, attributes)
assert len(found) == 1, (
f"expected exactly one {name} point matching {attributes}, "
f"got {len(found)}: {found}"
)
return found[0]
@pytest.fixture
def otel_metrics():
"""
Function-scoped metrics collector. Drains any delta state accumulated by
earlier tests before yielding, so counts start from zero.
"""
pytest.importorskip("opentelemetry.sdk")
if _otel_metric_reader is None:
pytest.skip("OpenTelemetry SDK meter provider was not installed")
_otel_metric_reader.get_metrics_data()
yield MetricsCollector(_otel_metric_reader)
@pytest.fixture
def bare_ds():
"""

323
tests/test_telemetry_metrics.py Normal file
View file

@ -0,0 +1,323 @@
"""
Tests for the OpenTelemetry metrics Datasette core emits.
Two layers are tested separately and deliberately:
- The gauge callbacks are plain generator functions, so they are called
directly for exact-value assertions. Going through the SDK for those would
be unreliable: the pool gauges carry no attribute identifying which
Datasette produced them, and a pytest session has many live instances, so
the SDK's last-value aggregation would report whichever one happened to be
observed last.
- The SDK pipeline (instrument -> reader -> data points) is tested through
the `otel_metrics` fixture, using metrics that carry `db.namespace` - a
uniquely named in-memory database is enough to isolate those from every
other instance alive in the session.
"""
import asyncio
import pytest
from datasette import telemetry
from datasette.app import Datasette
from datasette.database import Database
from datasette.utils.sqlite import sqlite3
pytestmark = pytest.mark.filterwarnings("ignore::ResourceWarning")
def observations(callback, datasette=None):
"""
Run a gauge callback, optionally keeping only observations produced by one
Datasette's databases. Returns a list of (attributes dict, value).
"""
results = []
names = None
if datasette is not None:
names = {db.name for db in telemetry._databases_of(datasette)}
for observation in callback():
attributes = dict(observation.attributes or {})
namespace = attributes.get("db.namespace")
if names is not None and namespace is not None and namespace not in names:
continue
results.append((attributes, observation.value))
return results
@pytest.fixture
def metrics_ds():
"A Datasette with a distinctive thread count and a uniquely named database."
ds = Datasette(
memory=True,
settings={"num_sql_threads": 7},
)
ds.add_memory_database("metrics_test_db")
try:
yield ds
finally:
ds.close()
@pytest.mark.asyncio
async def test_sql_thread_limit_gauge_reports_num_sql_threads(metrics_ds):
values = [value for _, value in observations(telemetry.observe_sql_thread_limit)]
# Other instances are alive in this session, so assert membership rather
# than uniqueness - 7 is distinctive enough to only come from metrics_ds.
assert 7 in values
@pytest.mark.asyncio
async def test_no_thread_gauges_in_non_threaded_mode():
"""
num_sql_threads=0 means there is no pool at all, so the pool gauges must
skip the instance rather than report a bogus limit of 0.
The pool gauges carry no attributes, so the assertion is that adding this
instance produces no *additional* observations.
"""
before_limits = len(list(telemetry.observe_sql_thread_limit()))
before_depths = len(list(telemetry.observe_sql_thread_queue_depth()))
ds = Datasette(memory=True, settings={"num_sql_threads": 0})
try:
assert ds.executor is None
assert len(list(telemetry.observe_sql_thread_limit())) == before_limits
assert len(list(telemetry.observe_sql_thread_queue_depth())) == before_depths
# Per-database gauges are unaffected - they do not depend on the pool.
assert observations(telemetry.observe_pending_queries, ds)
finally:
ds.close()
@pytest.mark.asyncio
async def test_pending_queries_gauge_tracks_in_flight_queries(metrics_ds):
db = metrics_ds.get_database("metrics_test_db")
attributes = {"db.namespace": "metrics_test_db"}
def value():
points = [
v
for a, v in observations(telemetry.observe_pending_queries, metrics_ds)
if a == attributes
]
assert len(points) == 1
return points[0]
assert value() == 0
# sqlite3.sleep is not a thing, so block the worker thread on an event we
# control from the event loop and sample the gauge while it is held.
release = asyncio.Event()
loop = asyncio.get_running_loop()
entered = asyncio.Event()
def blocking_fn(conn):
loop.call_soon_threadsafe(entered.set)
asyncio.run_coroutine_threadsafe(release.wait(), loop).result()
return "done"
task = asyncio.ensure_future(db.execute_fn(blocking_fn))
await entered.wait()
assert value() == 1, "a query occupying a pool thread must be counted as pending"
release.set()
assert await task == "done"
assert value() == 0, "the count must drop once the query completes"
@pytest.mark.asyncio
async def test_write_queue_depth_gauge(metrics_ds):
db = metrics_ds.get_database("metrics_test_db")
attributes = {"db.namespace": "metrics_test_db"}
def depths():
return [
v
for a, v in observations(telemetry.observe_write_queue_depth, metrics_ds)
if a == attributes
]
# No write has ever been queued, so there is no queue and no observation -
# rather than a fabricated zero for a queue that does not exist.
assert depths() == []
await db.execute_write("create table t (id integer primary key)")
assert depths() == [0], "an idle write queue reports zero, not nothing"
@pytest.mark.asyncio
async def test_open_connections_gauge(metrics_ds, tmp_path):
path = str(tmp_path / "conns.db")
sqlite3.connect(path).execute("create table t (id integer primary key)")
db = metrics_ds.add_database(Database(metrics_ds, path=path), name="conns_db")
attributes = {"db.namespace": "conns_db"}
def open_connections():
points = [
v
for a, v in observations(telemetry.observe_open_connections, metrics_ds)
if a == attributes
]
assert len(points) == 1
return points[0]
assert open_connections() == 0
await db.execute("select 1")
assert open_connections() >= 1, "executing a query opens a tracked file connection"
@pytest.mark.asyncio
async def test_operation_duration_histogram_read(otel_metrics):
ds = Datasette(memory=True)
ds.add_memory_database("duration_read_db")
try:
db = ds.get_database("duration_read_db")
await db.execute("select 1")
otel_metrics.collect()
point = otel_metrics.point(
"db.client.operation.duration",
{"db.namespace": "duration_read_db", "datasette.operation": "read"},
)
assert point.count == 1
assert point.sum > 0
assert dict(point.attributes)["db.system"] == "sqlite"
assert "error.type" not in dict(point.attributes)
finally:
ds.close()
@pytest.mark.asyncio
async def test_operation_duration_histogram_write(otel_metrics):
ds = Datasette(memory=True)
ds.add_memory_database("duration_write_db")
try:
db = ds.get_database("duration_write_db")
await db.execute_write("create table t (id integer primary key)")
otel_metrics.collect()
point = otel_metrics.point(
"db.client.operation.duration",
{"db.namespace": "duration_write_db", "datasette.operation": "write"},
)
assert point.count == 1
assert point.sum > 0
finally:
ds.close()
@pytest.mark.asyncio
async def test_operation_duration_records_error_type(otel_metrics):
"A failed query is still timed, and is separable from a successful one."
ds = Datasette(memory=True)
ds.add_memory_database("duration_error_db")
try:
db = ds.get_database("duration_error_db")
with pytest.raises(sqlite3.OperationalError):
await db.execute("select * from nope")
otel_metrics.collect()
point = otel_metrics.point(
"db.client.operation.duration",
{"db.namespace": "duration_error_db", "datasette.operation": "read"},
)
assert point.count == 1
assert dict(point.attributes)["error.type"] == "OperationalError"
finally:
ds.close()
@pytest.mark.asyncio
async def test_write_queue_wait_histogram(otel_metrics):
ds = Datasette(memory=True)
ds.add_memory_database("queue_wait_db")
try:
db = ds.get_database("queue_wait_db")
await db.execute_write("create table t (id integer primary key)")
await db.execute_write("insert into t (id) values (1)")
otel_metrics.collect()
point = otel_metrics.point(
"datasette.write.queue_wait", {"db.namespace": "queue_wait_db"}
)
assert point.count == 2, "one measurement per write dequeued"
assert point.sum >= 0
finally:
ds.close()
@pytest.mark.asyncio
async def test_interrupted_queries_counter(otel_metrics):
"The count of time-limit kills, which sampled traces cannot provide."
ds = Datasette(memory=True, settings={"sql_time_limit_ms": 1})
ds.add_memory_database("interrupted_db")
try:
db = ds.get_database("interrupted_db")
from datasette.database import QueryInterrupted
with pytest.raises(QueryInterrupted):
await db.execute("""
with recursive counter(x) as (
select 0 union all select x + 1 from counter
)
select * from counter
""")
otel_metrics.collect()
point = otel_metrics.point(
"datasette.sql.queries.interrupted", {"db.namespace": "interrupted_db"}
)
assert point.value == 1
finally:
ds.close()
@pytest.mark.asyncio
async def test_metrics_are_reported_through_the_sdk_for_gauges(otel_metrics):
"End-to-end: a gauge callback reaches the reader as a data point."
ds = Datasette(memory=True)
ds.add_memory_database("gauge_pipeline_db")
try:
await ds.get_database("gauge_pipeline_db").execute("select 1")
otel_metrics.collect()
point = otel_metrics.point(
"datasette.sql.queries.pending", {"db.namespace": "gauge_pipeline_db"}
)
assert point.value == 0
assert otel_metrics.points("datasette.sql.threads.limit")
finally:
ds.close()
def test_closed_datasette_stops_being_observed():
ds = Datasette(memory=True)
ds.add_memory_database("closed_db")
assert observations(telemetry.observe_pending_queries, ds)
ds.close()
names = [
attributes.get("db.namespace")
for attributes, _ in observations(telemetry.observe_pending_queries)
]
assert "closed_db" not in names
def test_registry_holds_instances_weakly():
"""
Registering an instance must never be the thing that keeps it alive.
A stand-in object is used rather than a real Datasette because a Datasette
with a temp-disk internal database is pinned for the life of the process
by `Database.__init__`'s `atexit.register(self._cleanup_temp_file)`, which
holds the Database, which holds the Datasette. That is pre-existing and
unrelated to telemetry; what is tested here is that this registry adds no
reference of its own.
"""
import gc
import weakref
class FakeDatasette:
pass
fake = FakeDatasette()
telemetry.register_datasette(fake)
assert fake in telemetry._live_instances()
ref = weakref.ref(fake)
del fake
gc.collect()
assert ref() is None
assert not any(isinstance(ds, FakeDatasette) for ds in telemetry._live_instances())