From f73128dea6f7e944fa654552d133e230168e44d6 Mon Sep 17 00:00:00 2001 From: Alex Garcia Date: Wed, 2 Sep 2026 11:48:56 -0700 Subject: [PATCH] Trace callback-style calls: execute_fn, execute_write_fn, execute_isolated_fn The database instrumentation covered the four SQL-string entry points but not the callback entry points, which are the documented way for plugins to run arbitrary SQL - so the JSON write API's inserts and deletes, the startup catalog scan, and every plugin built on execute_fn/execute_write_fn were invisible to a trace, or worse, showed orphan-looking db.write.* spans with no db.query above them. Each callback method now opens the same db.query CLIENT span as its SQL-string sibling, carrying a new optional datasette.callback attribute (the callable's qualified name, captured before _wrap_fn_with_hooks() can rename it) in place of db.query.text, which is now marked optional. A bare execute_fn() also wraps the callback in a db.query.execute child, so the "gap between the spans is thread-wait" story holds for plugin callbacks too. No db.operation.name: there is no statement to take a keyword from, and the registry says that attribute is omitted rather than guessed. The previous bodies move to private _execute_fn()/_execute_write_fn() and the SQL-string methods call those, so an execute() emits exactly the spans it did before - pinned by test_execute_does_not_double_wrap. Database's own introspection helpers stay on the public method deliberately: they are real SQLite round trips, which lifts a table page from ~58 to ~100 (no-op) spans. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_012U7coQfVu8nK2R4q2mCULA --- datasette/database.py | 106 ++++++++++++---- datasette/telemetry.py | 12 +- datasette/telemetry_registry.py | 24 +++- docs/changelog.rst | 2 +- docs/internals.rst | 9 +- tests/test_telemetry.py | 206 ++++++++++++++++++++++++++++++- tests/test_telemetry_registry.py | 12 ++ 7 files changed, 342 insertions(+), 29 deletions(-) diff --git a/datasette/database.py b/datasette/database.py index 300a71f6..f8adc5fa 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -17,8 +17,9 @@ 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 callback_name, sql_attribute, sql_operation_name, tracer from .telemetry_registry import ( + CALLBACK, DB_COLLECTION_NAME, DB_NAMESPACE, DB_OPERATION_NAME, @@ -299,7 +300,7 @@ 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( + results = await self._execute_write_fn( _inner, block=block, request=request, transaction=transaction ) return results @@ -323,7 +324,7 @@ 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( + results = await self._execute_write_fn( _inner, block=block, transaction=False, request=request ) return results @@ -356,7 +357,7 @@ 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( + results, count = await self._execute_write_fn( _inner, block=block, request=request ) # count is the number of parameter *sets* consumed by @@ -383,21 +384,31 @@ class Database: # Was probably a memory connection pass - 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) + # One db.query span here, like execute_fn() / execute_write_fn(). + # The wrap must NOT move into _send_to_write_thread(): that is the + # shared tail for every write, and for block=False it is where the + # link back to this span is captured - a span opened there would be + # the link target for its own children. + 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(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) async def analyze_sql(self, sql, params=None) -> SQLAnalysis: self._check_not_closed() @@ -407,6 +418,30 @@ class Database: ) async def execute_write_fn(self, fn, block=True, transaction=True, request=None): + """Run `fn(conn)` on the write connection, traced as one database call. + + The public entry point for callback-style writes. Instrumented like + `execute_write()`: one `db.query` span (with `datasette.callback` in + place of `db.query.text`) above the `db.write.queue_wait` and + `db.write.execute` spans the write thread emits. The SQL-string write + methods call `_execute_write_fn()` directly, so they never get a + second span. For `block=False` this span ends at enqueue and the + write-thread spans become roots carrying a link back to it, exactly + as for `execute_write(block=False)`. + """ + self._check_not_closed() + # The raw fn's name, before _wrap_fn_with_hooks() replaces it with a + # wrapper - otherwise every write would report the wrapper's name. + name = callback_name(fn) + 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(CALLBACK, name) + 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() pending_events = [] @@ -666,6 +701,35 @@ class Database: otel_context_api.detach(token) async def execute_fn(self, fn): + """Run `fn(conn)` on a read connection, traced as one database call. + + The public entry point for callback-style reads - plugins and core + both use it to run arbitrary Python against a connection. It is + instrumented exactly like `execute()`: one `db.query` span (with + `datasette.callback` in place of `db.query.text`, since there is no + SQL string to record) and a `db.query.execute` child covering the + time actually spent on the worker thread. `execute()` itself calls + `_execute_fn()` directly, so a SQL read never gets a second span. + """ + self._check_not_closed() + + def fn_in_execute_span(conn): + # Created on the worker thread; parents to the db.query span via + # the copy_context() propagation in _execute_fn(). The gap + # between the two spans is time spent waiting for a free thread. + with tracer.start_as_current_span(DB_QUERY_EXECUTE): + return fn(conn) + + 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(CALLBACK, callback_name(fn)) + # 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) + + async def _execute_fn(self, fn): self._check_not_closed() if self.ds.executor is None: # non-threaded mode @@ -752,7 +816,7 @@ class Database: # 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 + # 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. # @@ -837,7 +901,7 @@ class Database: if params: span.set_attribute(PARAM_COUNT, len(params)) try: - results = await self.execute_fn(sql_operation_in_thread) + 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 diff --git a/datasette/telemetry.py b/datasette/telemetry.py index 3705d89f..641dfeae 100644 --- a/datasette/telemetry.py +++ b/datasette/telemetry.py @@ -8,7 +8,7 @@ sampling - that is the responsibility of whoever is running Datasette harness). With no provider installed every span produced here is a -`NonRecordingSpan`. That is not free - a table page emits ~58 spans - +`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. @@ -54,6 +54,16 @@ def sql_attribute(sql: str) -> str: return sql[:MAX_SQL_LENGTH] + "…[truncated]" +def callback_name(fn) -> str: + """ + The name recorded as `datasette.callback` for a callback-style call. + + `functools.partial` objects (and other callables) have no `__qualname__`, + so fall back to the type's name rather than fail the query over telemetry. + """ + return getattr(fn, "__qualname__", type(fn).__name__) + + # db.operation.name is the leading keyword of a statement matched against a # fixed allowlist - deliberately not a parse. # diff --git a/datasette/telemetry_registry.py b/datasette/telemetry_registry.py index c2b9db32..183d514a 100644 --- a/datasette/telemetry_registry.py +++ b/datasette/telemetry_registry.py @@ -73,7 +73,23 @@ 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.", + "The SQL, truncated to 2048 characters. Never the parameter values. " + "Absent for a callback-style call (``execute_fn()`` and friends), where " + "there is no SQL string to record - ``datasette.callback`` is set " + "instead.", + optional=True, +) +CALLBACK = Attribute( + "datasette.callback", + "The qualified name of the Python callable passed to ``execute_fn()``, " + "``execute_write_fn()`` or ``execute_isolated_fn()`` - for example " + "``TableInsertView.post..insert_or_upsert_rows``. Set instead of " + "``db.query.text``, which does not exist for a callback: the SQL is " + "whatever the function chooses to run. A lambda reports ````, " + "which is why callers wanting a recognisable span should pass a named " + "function. Bounded cardinality: the set of callables is fixed by the " + "installed code, not by request input.", + optional=True, ) DB_OPERATION_NAME = Attribute( "db.operation.name", @@ -171,11 +187,15 @@ TRANSACTION = Attribute( DB_QUERY = SpanName( "db.query", "A SQL operation issued by Datasette, covering the full round trip " - "including any time spent queued for a thread.", + "including any time spent queued for a thread. Callback-style calls - " + "``execute_fn()``, ``execute_write_fn()`` and ``execute_isolated_fn()`` - " + "appear here too, distinguished by ``datasette.callback`` in place of " + "``db.query.text``.", ( DB_SYSTEM, DB_NAMESPACE, DB_QUERY_TEXT, + CALLBACK, DB_OPERATION_NAME, DB_COLLECTION_NAME, PARAM_COUNT, diff --git a/docs/changelog.rst b/docs/changelog.rst index 03d913b1..1488f3a7 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -9,7 +9,7 @@ Changelog Unreleased ---------- -- Datasette's database layer now emits `OpenTelemetry `__ 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`) +- Datasette's database layer now emits `OpenTelemetry `__ 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. Callback-style calls - :ref:`db.execute_fn() `, :ref:`db.execute_write_fn() ` and ``db.execute_isolated_fn()``, the documented way for plugins to run arbitrary SQL - are covered too, carrying ``datasette.callback`` in place of the SQL text. 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) ` 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`) Nothing is removed by the OpenTelemetry work: the ``?_trace=1`` query string parameter, the ``trace_debug`` setting and the :ref:`internals_tracer` module all continue to work as before. diff --git a/docs/internals.rst b/docs/internals.rst index aee51969..bc3d2fb1 100644 --- a/docs/internals.rst +++ b/docs/internals.rst @@ -2024,6 +2024,8 @@ Example usage: version = await db.execute_fn(get_version) +The call is traced as a ``db.query`` OpenTelemetry span carrying ``datasette.callback`` (the function's qualified name) rather than ``db.query.text``, since the SQL is whatever the function chooses to run - see :ref:`internals_telemetry`. Passing a named function gives the span a readable identity; a lambda reports ````. + .. _database_execute_write: await db.execute_write(sql, params=None, block=True, request=None, return_all=False, returning_limit=10, transaction=True) @@ -2100,6 +2102,8 @@ This method works like ``.execute_write()``, but instead of a SQL statement you The function can then perform multiple actions, safe in the knowledge that it has exclusive access to the single writable connection for as long as it is executing. +Like ``execute_fn()``, the call is traced as a ``db.query`` OpenTelemetry span carrying ``datasette.callback`` rather than ``db.query.text``, above the write-queue spans - see :ref:`internals_telemetry`. A named function gives the span a readable identity; a lambda reports ````. + .. warning:: ``fn`` needs to be a regular function, not an ``async def`` function. @@ -2373,7 +2377,7 @@ Spans are ``SpanKind.INTERNAL`` unless a kind is listed below. Only ``db.query`` .. ]]] ``db.query`` - A SQL operation issued by Datasette, covering the full round trip including any time spent queued for a thread. + A SQL operation issued by Datasette, covering the full round trip including any time spent queued for a thread. Callback-style calls - ``execute_fn()``, ``execute_write_fn()`` and ``execute_isolated_fn()`` - appear here too, distinguished by ``datasette.callback`` in place of ``db.query.text``. Kind: ``CLIENT``. @@ -2381,7 +2385,8 @@ Spans are ``SpanKind.INTERNAL`` unless a kind is listed below. Only ``db.query`` - ``db.system`` - Always ``sqlite``. - ``db.namespace`` - Name of the database being queried. - - ``db.query.text`` - The SQL, truncated to 2048 characters. Never the parameter values. + - ``db.query.text`` *(optional)* - The SQL, truncated to 2048 characters. Never the parameter values. Absent for a callback-style call (``execute_fn()`` and friends), where there is no SQL string to record - ``datasette.callback`` is set instead. + - ``datasette.callback`` *(optional)* - The qualified name of the Python callable passed to ``execute_fn()``, ``execute_write_fn()`` or ``execute_isolated_fn()`` - for example ``TableInsertView.post..insert_or_upsert_rows``. Set instead of ``db.query.text``, which does not exist for a callback: the SQL is whatever the function chooses to run. A lambda reports ````, which is why callers wanting a recognisable span should pass a named function. Bounded cardinality: the set of callables is fixed by the installed code, not by request input. - ``db.operation.name`` *(optional)* - 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 attribute must stay safe to use as a metric dimension, 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. - ``db.collection.name`` *(optional)* - 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. - ``datasette.param_count`` *(optional)* - Number of bound parameters. Recorded instead of the values themselves. diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index 83e13b6e..a5ccc3fb 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -178,7 +178,14 @@ async def test_facetable_request_produces_db_query_spans(ds_client, otel_spans): spans = _db_query_spans(otel_spans) assert spans, "expected at least one db.query span" assert all(span.attributes["db.system"] == "sqlite" for span in spans) - assert all(span.attributes["db.query.text"] for span in spans) + # Every db.query names what ran: SQL text for the string methods, + # datasette.callback for callback-style calls (schema introspection here). + assert all( + span.attributes.get("db.query.text") + or span.attributes.get("datasette.callback") + for span in spans + ) + assert any(span.attributes.get("db.query.text") for span in spans) # Rendering the page also queries the internal database, so only some of # these spans belong to "fixtures". assert any(span.attributes["db.namespace"] == "fixtures" for span in spans) @@ -519,8 +526,14 @@ async def test_immutable_database_propagates_context(tmp_path, otel_spans): for span in otel_spans.get_finished_spans() if span.name == "t04-child-in-isolated-worker" ], "expected a span created inside execute_isolated_fn's worker thread" + # execute_isolated_fn() now opens its own db.query span, so the chain is + # event-loop parent -> db.query -> worker child. The worker child + # parenting to that db.query span, across the thread, is the propagation + # this test exists to prove. + query_spans = _children_named(otel_spans, "db.query", parent_context) + assert len(query_spans) == 1 children = _children_named( - otel_spans, "t04-child-in-isolated-worker", parent_context + otel_spans, "t04-child-in-isolated-worker", query_spans[0].context ) assert len(children) == 1 @@ -1036,3 +1049,192 @@ async def test_table_and_row_pages_set_db_collection_name( assert any( span.attributes.get("db.collection.name") == table for span in spans ), f"expected a db.query span from {path} carrying db.collection.name" + + +# --- Callback-style calls: execute_fn / execute_write_fn / execute_isolated_fn + + +@pytest.mark.asyncio +async def test_execute_fn_produces_db_query_span(otel_spans): + db = Datasette(memory=True).add_memory_database("t16_execute_fn") + await db.execute_write("create table t (id integer primary key)") + + def count_rows(conn): + return conn.execute("select count(*) from t").fetchone()[0] + + otel_spans.clear() + assert await db.execute_fn(count_rows) == 0 + + spans = _spans_for_namespace(otel_spans, "t16_execute_fn") + assert len(spans) == 1 + span = spans[0] + assert span.kind == SpanKind.CLIENT + assert span.attributes["db.system"] == "sqlite" + assert ( + span.attributes["datasette.callback"] + == "test_execute_fn_produces_db_query_span..count_rows" + ) + # There is no SQL string for a callback, and no statement to take a + # leading keyword from - absent beats guessed. + assert "db.query.text" not in span.attributes + assert "db.operation.name" not in span.attributes + children = _children_named(otel_spans, "db.query.execute", span.context) + assert len(children) == 1 + + +@pytest.mark.asyncio +async def test_execute_fn_lambda_reports_lambda(otel_spans): + # Pins the documented behaviour rather than pretending lambdas have names. + db = Datasette(memory=True).add_memory_database("t16_lambda") + otel_spans.clear() + await db.execute_fn(lambda conn: conn.execute("select 1").fetchone()) + spans = _spans_for_namespace(otel_spans, "t16_lambda") + assert len(spans) == 1 + assert spans[0].attributes["datasette.callback"].endswith("") + + +@pytest.mark.asyncio +async def test_execute_write_fn_produces_db_query_span(otel_spans): + db = Datasette(memory=True).add_memory_database("t16_write_fn") + + def create_table(conn): + conn.execute("create table t (id integer primary key)") + + otel_spans.clear() + await db.execute_write_fn(create_table) + + spans = _spans_for_namespace(otel_spans, "t16_write_fn") + assert len(spans) == 1 + span = spans[0] + assert span.kind == SpanKind.CLIENT + assert ( + span.attributes["datasette.callback"] + == "test_execute_write_fn_produces_db_query_span..create_table" + ) + assert "db.query.text" not in span.attributes + # The write-thread spans are this span's children, same as execute_write() + for name in ("db.write.queue_wait", "db.write.execute"): + assert len(_children_named(otel_spans, name, span.context)) == 1, name + + +@pytest.mark.asyncio +async def test_execute_write_fn_callback_name_is_not_the_hook_wrapper(otel_spans): + # A callback that declares track_event is the case where + # _wrap_fn_with_hooks() actually replaces fn with a wrapper - the span + # must still report the caller's function, not the wrapper's name. + db = Datasette(memory=True).add_memory_database("t16_wrapper_name") + + def create_with_events(conn, track_event): + conn.execute("create table t (id integer primary key)") + + otel_spans.clear() + await db.execute_write_fn(create_with_events) + spans = _spans_for_namespace(otel_spans, "t16_wrapper_name") + assert len(spans) == 1 + assert spans[0].attributes["datasette.callback"] == ( + "test_execute_write_fn_callback_name_is_not_the_hook_wrapper" + "..create_with_events" + ) + + +@pytest.mark.asyncio +async def test_execute_write_fn_nonblocking_spans_link_to_the_new_span(otel_spans): + # For block=False the public db.query span ends at enqueue and the + # write-thread spans become roots. Their link must target that new span, + # not whatever was current around the execute_write_fn() call. + db = Datasette(memory=True).add_memory_database("t16_nonblocking") + await db.execute_write("create table docs (id integer primary key)") + + def insert(conn): + conn.execute("insert into docs (id) values (1)") + + otel_spans.clear() + with tracer.start_as_current_span("t16-enqueueing-span") as enqueuer: + enqueuer_context = enqueuer.get_span_context() + await db.execute_write_fn(insert, block=False) + # Writes are serialized on the write thread, so a blocking write behind + # the non-blocking one waits for it deterministically. + await db.execute_write("insert into docs (id) values (2)") + + query_spans = [ + span + for span in _spans_for_namespace(otel_spans, "t16_nonblocking") + if span.attributes.get("datasette.callback") + ] + assert len(query_spans) == 1 + fn_span_context = query_spans[0].context + linked = [ + span + for span in otel_spans.get_finished_spans() + if span.name in ("db.write.queue_wait", "db.write.execute") and span.links + ] + assert len(linked) == 2 + for span in linked: + assert span.parent is None, f"{span.name} is still parented" + assert span.links[0].context.span_id == fn_span_context.span_id, span.name + assert span.links[0].context.span_id != enqueuer_context.span_id, span.name + + +@pytest.mark.asyncio +async def test_execute_does_not_double_wrap(otel_spans): + # The regression guard for the refactor: execute() and the SQL-string + # write methods call the private _execute_fn/_execute_write_fn, so they + # must not gain a second db.query span from the public wrappers. + db = Datasette(memory=True).add_memory_database("t16_no_double_wrap") + otel_spans.clear() + await db.execute_write("create table t (id integer primary key)") + assert len(_spans_for_namespace(otel_spans, "t16_no_double_wrap")) == 1 + otel_spans.clear() + await db.execute("select * from t") + spans = _spans_for_namespace(otel_spans, "t16_no_double_wrap") + assert len(spans) == 1 + assert len(_children_named(otel_spans, "db.query.execute", spans[0].context)) == 1 + + +@pytest.mark.asyncio +async def test_execute_isolated_fn_span_on_mutable_and_immutable(tmp_path, otel_spans): + def read_one(conn): + return conn.execute("select 1").fetchone()[0] + + mutable = Datasette(memory=True).add_memory_database("t16_isolated_mutable") + otel_spans.clear() + assert await mutable.execute_isolated_fn(read_one) == 1 + spans = _spans_for_namespace(otel_spans, "t16_isolated_mutable") + assert len(spans) == 1 + assert spans[0].attributes["datasette.callback"].endswith("read_one") + # Mutable databases route through the write thread, so the write spans + # appear as children; immutable ones run on the pool and get none. + assert _children_named(otel_spans, "db.write.execute", spans[0].context) + + db_path = tmp_path / "t16_isolated_immutable.db" + sqlite_utils.Database(str(db_path))["t"].insert({"id": 1}) + ds = Datasette() + immutable = Database(ds, path=str(db_path), is_mutable=False) + ds.add_database(immutable, name="t16_isolated_immutable") + try: + otel_spans.clear() + assert await immutable.execute_isolated_fn(read_one) == 1 + finally: + ds.remove_database("t16_isolated_immutable") + spans = _spans_for_namespace(otel_spans, "t16_isolated_immutable") + assert len(spans) == 1 + assert spans[0].attributes["datasette.callback"].endswith("read_one") + assert not _children_named(otel_spans, "db.write.execute", spans[0].context) + + +@pytest.mark.asyncio +async def test_execute_fn_exception_marks_span_error(otel_spans): + # Unlike execute(), there is no probing caller on this path - a callback + # that raises is an error, with the default record_exception behaviour. + db = Datasette(memory=True).add_memory_database("t16_fn_error") + + def boom(conn): + raise ValueError("callback failed") + + otel_spans.clear() + with pytest.raises(ValueError): + await db.execute_fn(boom) + spans = _spans_for_namespace(otel_spans, "t16_fn_error") + assert len(spans) == 1 + assert spans[0].status.status_code == StatusCode.ERROR + assert any(event.name == "exception" for event in spans[0].events) diff --git a/tests/test_telemetry_registry.py b/tests/test_telemetry_registry.py index 75a183b4..ef7c93a3 100644 --- a/tests/test_telemetry_registry.py +++ b/tests/test_telemetry_registry.py @@ -44,6 +44,7 @@ EXPECTED_ATTRIBUTES = { "db.system", "db.namespace", "db.query.text", + "datasette.callback", "db.operation.name", "db.collection.name", "datasette.param_count", @@ -108,6 +109,17 @@ async def exercise(): # datasette.isolated_connection=True await db.execute_isolated_fn(lambda conn: conn.execute("select 1").fetchone()) + # datasette.callback, with named functions so the conformance run sees the + # attribute's documented value shape (a qualname, not just "") + def registry_read_callback(conn): + return conn.execute("select count(*) from t").fetchone() + + def registry_write_callback(conn): + conn.execute("insert into t (id, v) values (100, 'callback')") + + await db.execute_fn(registry_read_callback) + await db.execute_write_fn(registry_write_callback) + # Reads: db.query.execute, datasette.rows_returned, datasette.truncated, # datasette.param_count, datasette.time_limit_ms await db.execute("select * from t where id > :n", {"n": 5})