From 9e5a4021fd03f2c1bf9e69c2084e786211afccd4 Mon Sep 17 00:00:00 2001 From: Alex Garcia Date: Wed, 2 Sep 2026 11:51:03 -0700 Subject: [PATCH] Count callback-style calls in db.client.operation.duration The callback entry points gained db.query spans in the database-spans PR; this adds their other half - the duration histogram measurement, so a plugin's execute_fn/execute_write_fn work and the JSON write API's inserts and deletes stop being invisible to the one series that survives trace sampling. execute_isolated_fn records "write" when the database is mutable (the call blocks the write queue) and "read" when immutable (it runs on the read pool). error.type comes from the raised exception class, same as the SQL-string paths. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA --- datasette/database.py | 45 +++++++++++---------- datasette/telemetry_registry.py | 4 +- docs/internals.rst | 2 +- tests/test_telemetry_metrics.py | 70 +++++++++++++++++++++++++++++++++ 4 files changed, 99 insertions(+), 22 deletions(-) diff --git a/datasette/database.py b/datasette/database.py index ec83ecad..96d42a21 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -404,22 +404,25 @@ class Database: span.set_attribute(DB_SYSTEM, "sqlite") span.set_attribute(DB_NAMESPACE, self.name) span.set_attribute(CALLBACK, callback_name(fn)) - if self.ds.executor is None: - # non-threaded mode - 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. copy_context() carries the caller's otel context - # onto the worker thread - see the notes in _execute_fn() for - # why it must be a fresh copy per submit and why carrying - # every ContextVar is safe. - ctx = contextvars.copy_context() - return await asyncio.get_running_loop().run_in_executor( - self.ds.executor, ctx.run, _run - ) - # Threaded mode - send to write thread - return await self._send_to_write_thread(fn, isolated_connection=True) + # "write" when mutable because the call blocks the write queue; + # "read" when immutable, where it runs on the read pool. + with record_operation_duration(self.name, "write" if write else "read"): + if self.ds.executor is None: + # non-threaded mode + 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. copy_context() carries the + # caller's otel context onto the worker thread - see the + # notes in _execute_fn() for why it must be a fresh copy + # per submit and why carrying every ContextVar is safe. + ctx = contextvars.copy_context() + return await asyncio.get_running_loop().run_in_executor( + self.ds.executor, ctx.run, _run + ) + # Threaded mode - send to write thread + return await self._send_to_write_thread(fn, isolated_connection=True) async def analyze_sql(self, sql, params=None) -> SQLAnalysis: self._check_not_closed() @@ -449,9 +452,10 @@ class Database: span.set_attribute(DB_SYSTEM, "sqlite") span.set_attribute(DB_NAMESPACE, self.name) span.set_attribute(CALLBACK, name) - return await self._execute_write_fn( - fn, block=block, transaction=transaction, request=request - ) + with record_operation_duration(self.name, "write"): + return await self._execute_write_fn( + fn, block=block, transaction=transaction, request=request + ) async def _execute_write_fn(self, fn, block=True, transaction=True, request=None): self._check_not_closed() @@ -741,7 +745,8 @@ class Database: # Default exception handling applies, unlike execute(): there is # no log_sql_errors=False probing caller and no expected-timeout # budget on this path, so a raised exception is an error. - return await self._execute_fn(fn_in_execute_span) + with record_operation_duration(self.name, "read"): + return await self._execute_fn(fn_in_execute_span) async def _execute_fn(self, fn): self._check_not_closed() diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py index 3c775a7b..e4082347 100644 --- a/datasette/telemetry_registry.py +++ b/datasette/telemetry_registry.py @@ -459,7 +459,9 @@ M_OPERATION_DURATION = MetricName( HISTOGRAM, "s", "Duration of a SQL operation. The standard OpenTelemetry semantic " - "convention metric, and the one that survives trace sampling.", + "convention metric, and the one that survives trace sampling. " + "Callback-style calls (``execute_fn()`` and friends) are counted " + "alongside the SQL-string methods.", (DB_SYSTEM, DB_NAMESPACE, OPERATION, ERROR_TYPE), buckets=DURATION_BUCKETS, ) diff --git a/docs/internals.rst b/docs/internals.rst index 3b1ae285..b82ebfbc 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -2466,7 +2466,7 @@ This reference is generated from ``datasette/telemetry_registry.py``, like the s .. ]]] ``db.client.operation.duration`` - Histogram, unit ``s``. Duration of a SQL operation. The standard OpenTelemetry semantic convention metric, and the one that survives trace sampling. + Histogram, unit ``s``. Duration of a SQL operation. The standard OpenTelemetry semantic convention metric, and the one that survives trace sampling. Callback-style calls (``execute_fn()`` and friends) are counted alongside the SQL-string methods. Bucket boundaries: ``0.0001``, ``0.0005``, ``0.001``, ``0.005``, ``0.01``, ``0.05``, ``0.1``, ``0.5``, ``1``, ``5``, ``10``. diff --git a/tests/test_telemetry_metrics.py b/tests/test_telemetry_metrics.py index f1ac4a56..5aade58b 100644 --- a/tests/test_telemetry_metrics.py +++ b/tests/test_telemetry_metrics.py @@ -460,3 +460,73 @@ def test_histograms_spread_values_across_buckets( f"expected each of {SPREAD} in its own bucket, got bucket counts " f"{list(point.bucket_counts)} for bounds {list(point.explicit_bounds)}" ) + + +@pytest.mark.asyncio +async def test_operation_duration_histogram_records_execute_fn(otel_metrics): + "Callback-style reads land in the same histogram as SQL-string reads." + ds = Datasette(memory=True) + ds.add_memory_database("duration_fn_db") + try: + db = ds.get_database("duration_fn_db") + + def read_one(conn): + return conn.execute("select 1").fetchone()[0] + + assert await db.execute_fn(read_one) == 1 + otel_metrics.collect() + point = otel_metrics.point( + "db.client.operation.duration", + {"db.namespace": "duration_fn_db", "datasette.operation": "read"}, + ) + assert point.count == 1 + assert point.sum > 0 + finally: + ds.close() + + +@pytest.mark.asyncio +async def test_operation_duration_histogram_records_execute_write_fn(otel_metrics): + "Callback-style writes - the JSON write API's whole diet - are counted too." + ds = Datasette(memory=True) + ds.add_memory_database("duration_write_fn_db") + try: + db = ds.get_database("duration_write_fn_db") + + def create_table(conn): + conn.execute("create table t (id integer primary key)") + + await db.execute_write_fn(create_table) + otel_metrics.collect() + point = otel_metrics.point( + "db.client.operation.duration", + {"db.namespace": "duration_write_fn_db", "datasette.operation": "write"}, + ) + assert point.count == 1 + assert point.sum > 0 + finally: + ds.close() + + +@pytest.mark.asyncio +async def test_operation_duration_records_callback_error_type(otel_metrics): + "A callback that raises is still timed, with error.type from the exception." + ds = Datasette(memory=True) + ds.add_memory_database("duration_fn_error_db") + try: + db = ds.get_database("duration_fn_error_db") + + def boom(conn): + raise ValueError("callback failed") + + with pytest.raises(ValueError): + await db.execute_fn(boom) + otel_metrics.collect() + point = otel_metrics.point( + "db.client.operation.duration", + {"db.namespace": "duration_fn_error_db", "datasette.operation": "read"}, + ) + assert point.count == 1 + assert dict(point.attributes)["error.type"] == "ValueError" + finally: + ds.close()