diff --git a/datasette/database.py b/datasette/database.py index e162d34e..f78eb925 100644 --- a/datasette/database.py +++ b/datasette/database.py @@ -11,8 +11,10 @@ from collections import namedtuple from pathlib import Path import sqlite_utils +from opentelemetry.trace import Status, StatusCode from .inspect import inspect_hash +from .telemetry import sql_attribute, tracer from .tracer import trace from .utils import ( call_with_supported_arguments, @@ -529,12 +531,11 @@ class Database: """Executes sql against db_name in a thread""" self._check_not_closed() page_size = page_size or self.ds.page_size + time_limit_ms = self.ds.sql_time_limit_ms + if custom_time_limit and custom_time_limit < time_limit_ms: + time_limit_ms = custom_time_limit def sql_operation_in_thread(conn): - time_limit_ms = self.ds.sql_time_limit_ms - if custom_time_limit and custom_time_limit < time_limit_ms: - time_limit_ms = custom_time_limit - with sqlite_timelimit(conn, time_limit_ms): try: cursor = conn.cursor() @@ -565,8 +566,49 @@ class Database: else: return Results(rows, False, cursor.description) - with trace("sql", database=self.name, sql=sql.strip(), params=params): - results = await self.execute_fn(sql_operation_in_thread) + # SIM117 wants these two context managers merged. They are kept nested + # deliberately: the hand-rolled tracer's wrapper is on its way out, and + # nesting makes removing it a single-line deletion. + with trace( # noqa: SIM117 + "sql", database=self.name, sql=sql.strip(), params=params + ): + # Exception handling is explicit rather than left to the context + # manager's defaults, so that callers passing log_sql_errors=False + # can be honoured - see the comment on the generic handler below. + with tracer.start_as_current_span( + "db.query", + record_exception=False, + set_status_on_exception=False, + ) as span: + span.set_attribute("db.system", "sqlite") + span.set_attribute("db.namespace", self.name) + span.set_attribute("db.query.text", sql_attribute(sql)) + span.set_attribute("datasette.time_limit_ms", time_limit_ms) + if params: + span.set_attribute("datasette.param_count", len(params)) + try: + results = await self.execute_fn(sql_operation_in_thread) + except QueryInterrupted as e: + span.set_status(Status(StatusCode.ERROR, str(e))) + span.set_attribute("datasette.interrupted", True) + span.record_exception(e) + raise + except Exception as e: + # log_sql_errors=False means the caller is probing and + # treats failure as an expected answer, not an error. + # Facet suggestion is the big one: it runs json_type() + # against every column precisely to find out which ones + # raise, so a table with N text columns would otherwise + # mark N queries per page as failed - burying real errors + # and setting off any alerting based on span status. + if log_sql_errors: + span.record_exception(e) + span.set_status(Status(StatusCode.ERROR, str(e))) + else: + span.set_attribute("datasette.sql_error_suppressed", True) + raise + span.set_attribute("datasette.truncated", results.truncated) + span.set_attribute("datasette.rows_returned", len(results.rows)) return results @property diff --git a/tests/test_telemetry.py b/tests/test_telemetry.py index b71be504..a13898ff 100644 --- a/tests/test_telemetry.py +++ b/tests/test_telemetry.py @@ -1,6 +1,31 @@ +import json +import sqlite3 import subprocess import sys +import pytest +from opentelemetry.trace import StatusCode + +from datasette.telemetry import MAX_SQL_LENGTH, sql_attribute + +SECRET_PARAM_VALUE = "SUPER_SECRET_PARAM_VALUE_XYZ_123" + +INVALID_SQL = "select this_is_not_valid_sql from nowhere" + + +def _db_query_spans(otel_spans): + return [span for span in otel_spans.get_finished_spans() if span.name == "db.query"] + + +def _all_attribute_values(otel_spans): + "Every attribute value across every finished span, for the 'no leaked param values' test." + values = [] + for span in otel_spans.get_finished_spans(): + values.extend((span.attributes or {}).values()) + for event in span.events: + values.extend((event.attributes or {}).values()) + return values + def test_datasette_package_never_imports_the_sdk(): """ @@ -23,3 +48,147 @@ def test_datasette_package_never_imports_the_sdk(): assert ( result.stdout.strip() == "[]" ), f"datasette imported the OpenTelemetry SDK: {result.stdout.strip()}" + + +@pytest.mark.asyncio +async def test_db_query_span_basic_attributes(ds_client, otel_spans): + response = await ds_client.get("/fixtures/-/query.json?sql=select+1") + assert response.status_code == 200 + + spans = _db_query_spans(otel_spans) + assert spans, "expected at least one db.query span" + span = spans[-1] + + assert span.attributes["db.system"] == "sqlite" + assert span.attributes["db.namespace"] == "fixtures" + assert span.attributes["db.query.text"] == "select 1" + assert span.attributes["datasette.rows_returned"] == 1 + assert span.attributes["datasette.truncated"] is False + assert isinstance(span.attributes["datasette.time_limit_ms"], int) + assert span.status.status_code == StatusCode.UNSET + + +@pytest.mark.asyncio +async def test_facetable_request_produces_db_query_spans(ds_client, otel_spans): + response = await ds_client.get("/fixtures/facetable.json") + assert response.status_code == 200 + + 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) + # 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) + + +def test_sql_attribute_truncates_at_2048(): + short_sql = "select 1" + assert sql_attribute(short_sql) == "select 1" + # Whitespace is stripped, so the same query logged twice with different + # surrounding whitespace produces one attribute value, not two. + assert sql_attribute(" select 1\n") == "select 1" + + long_sql = "select 1 -- " + ("x" * 3000) + truncated = sql_attribute(long_sql) + assert len(truncated) == MAX_SQL_LENGTH + len("…[truncated]") + assert truncated.startswith("select 1 -- ") + assert truncated.endswith("…[truncated]") + + +@pytest.mark.asyncio +async def test_db_query_text_is_truncated_in_real_span(ds_client, otel_spans): + # A long trailing SQL comment keeps the query valid and executable while + # pushing db.query.text well past the 2048 char cap. + long_sql = "select 1 -- " + ("x" * 3000) + response = await ds_client.get("/fixtures/-/query.json", params={"sql": long_sql}) + assert response.status_code == 200 + + spans = _db_query_spans(otel_spans) + assert spans + assert any(len(span.attributes["db.query.text"]) > 100 for span in spans), ( + "expected the long query to reach a span - otherwise this test would " + "pass even if truncation were never applied" + ) + for span in spans: + recorded = span.attributes["db.query.text"] + assert len(recorded) <= MAX_SQL_LENGTH + len("…[truncated]") + + +@pytest.mark.asyncio +async def test_no_span_attribute_ever_contains_a_parameter_value(ds_client, otel_spans): + response = await ds_client.get( + "/fixtures/-/query.json", + params={"sql": "select :secret", "secret": SECRET_PARAM_VALUE}, + ) + assert response.status_code == 200 + # Sanity check the value really did flow through as a bound parameter, + # not inlined into the SQL text, otherwise this test would be vacuous. + assert SECRET_PARAM_VALUE in json.dumps(response.json()) + + for value in _all_attribute_values(otel_spans): + if isinstance(value, str): + assert SECRET_PARAM_VALUE not in value + elif isinstance(value, (list, tuple)): + for item in value: + if isinstance(item, str): + assert SECRET_PARAM_VALUE not in item + + spans = _db_query_spans(otel_spans) + assert spans + span = spans[-1] + assert "select :secret" in span.attributes["db.query.text"] + assert span.attributes.get("datasette.param_count") == 1 + + +@pytest.mark.asyncio +async def test_query_interrupted_sets_error_status(ds_client, otel_spans): + response = await ds_client.get( + "/fixtures/-/query.json", + params={"sql": "select sleep(0.05)", "_timelimit": 5}, + ) + assert response.status_code == 400 + + spans = _db_query_spans(otel_spans) + assert spans + span = spans[-1] + assert span.status.status_code == StatusCode.ERROR + assert span.attributes["datasette.interrupted"] is True + assert span.events + assert all(event.name == "exception" for event in span.events) + + +@pytest.mark.asyncio +async def test_unsuppressed_sql_error_is_a_span_error(ds_client, otel_spans): + db = ds_client.ds.get_database("fixtures") + with pytest.raises(sqlite3.OperationalError): + await db.execute(INVALID_SQL) + + spans = _db_query_spans(otel_spans) + assert spans + span = spans[-1] + assert span.status.status_code == StatusCode.ERROR + assert any(event.name == "exception" for event in span.events) + assert "datasette.sql_error_suppressed" not in span.attributes + + +@pytest.mark.asyncio +async def test_suppressed_sql_error_is_not_a_span_error(ds_client, otel_spans): + """ + log_sql_errors=False means the caller is probing and expects failures. + + Facet suggestion runs `json_type(column)` against every column precisely + to discover which ones raise, so marking those spans as errors would put + two red spans per text column on every table page - burying real failures + and tripping any alerting keyed on span status. + """ + db = ds_client.ds.get_database("fixtures") + with pytest.raises(sqlite3.OperationalError): + await db.execute(INVALID_SQL, log_sql_errors=False) + + spans = _db_query_spans(otel_spans) + assert spans + span = spans[-1] + assert span.status.status_code == StatusCode.UNSET + assert span.attributes["datasette.sql_error_suppressed"] is True + assert not [event for event in span.events if event.name == "exception"]