Trace Postgres and Redis queries with the contrib instrumentations

Builds the TracerProvider in tavolo.telemetry (kaya-otel's is private) and
injects it into kaya-otel, then instruments asyncpg and redis through
opentelemetry-instrumentation-asyncpg/redis. Redis command values are
redacted and asyncpg spans carry the parameterized SQL only.

server/tests/test_telemetry.py

@@ -9,6 +9,7 @@ import logging
 import os
 import sys
 import unittest
+from types import SimpleNamespace
 from typing import Optional
 from unittest.mock import patch
 
@@ -33,6 +34,11 @@ try:
     from opentelemetry.sdk.metrics import MeterProvider
     from opentelemetry.sdk.metrics.export import InMemoryMetricReader
     from opentelemetry.sdk.resources import Resource
+    from opentelemetry.sdk.trace import TracerProvider
+    from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+    from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
+        InMemorySpanExporter,
+    )
 
     HAVE_OTEL = True
 except ImportError:  # pragma: no cover - installs without the otel extra
@@ -148,19 +154,25 @@ def _metric_points(reader) -> dict:
 
 @unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
 class MetricExportTest(unittest.TestCase):
-    # SystemMetricsInstrumentor is a process-wide singleton whose callbacks
-    # cannot be re-registered, so the provider and mixin are built once for
-    # the whole class (as the application does).
+    # The instrumentation packages are process-wide singletons whose
+    # callbacks cannot be re-registered, so the providers and mixin are
+    # built once for the whole class (as the application does).
     sessions = _FakeSessions(0)
     games = _FakeGames(0)
 
     @classmethod
     def setUpClass(cls) -> None:
-        cls.reader = InMemoryMetricReader()
         resource = Resource.create({"service.name": "test"})
+        cls.reader = InMemoryMetricReader()
         cls.provider = MeterProvider(resource=resource, metric_readers=[cls.reader])
+        cls.span_exporter = InMemorySpanExporter()
+        cls.tracer_provider = TracerProvider(resource=resource)
+        cls.tracer_provider.add_span_processor(
+            SimpleSpanProcessor(cls.span_exporter)
+        )
         cls.mixin = TelemetryMixin(
             meter_provider=cls.provider,
+            tracer_provider=cls.tracer_provider,
             logger_provider=None,
             log_handler=None,
             session_store=cls.sessions,
@@ -171,7 +183,7 @@ class MetricExportTest(unittest.TestCase):
 
     @classmethod
     def tearDownClass(cls) -> None:
-        cls.provider.shutdown()
+        cls.mixin.shutdown(None)  # type: ignore[arg-type]
 
     @async_test
     async def test_app_gauges_report_polled_counts(self) -> None:
@@ -218,6 +230,36 @@ class MetricExportTest(unittest.TestCase):
         # Host-level metrics are deliberately not configured.
         self.assertNotIn("system.cpu.time", points)
 
+    @async_test
+    async def test_database_clients_are_instrumented(self) -> None:
+        import asyncpg.connection
+        import redis.asyncio
+
+        self.assertTrue(hasattr(redis.asyncio.Redis.execute_command, "__wrapped__"))
+        self.assertTrue(hasattr(asyncpg.connection.Connection.fetch, "__wrapped__"))
+
+        # Exercise each wrapper through stubs: the downstream call fails (no
+        # server), but the span is recorded before it does.
+        client = object.__new__(redis.asyncio.Redis)
+        client.connection_pool = SimpleNamespace(
+            connection_kwargs={"host": "127.0.0.1", "port": 6379, "db": 0}
+        )
+        with self.assertRaises(AttributeError):
+            await client.execute_command("GET", "key")
+
+        connection = object.__new__(asyncpg.connection.Connection)
+        connection._aborted = True  # keep __del__ quiet on the stub
+        with self.assertRaises(Exception):
+            await connection.fetch("SELECT 1")
+
+        spans = {
+            span.name: span for span in self.span_exporter.get_finished_spans()
+        }
+        self.assertEqual("GET ?", spans["GET"].attributes["db.statement"])
+        self.assertEqual("redis", spans["GET"].attributes["db.system"])
+        self.assertEqual("SELECT 1", spans["SELECT"].attributes["db.statement"])
+        self.assertEqual("postgresql", spans["SELECT"].attributes["db.system"])
+
 
 @unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
 class LogExportTest(unittest.TestCase):