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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA

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()

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,
 )

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``.
 

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()