Tag traced requests with the session id and OIDC sub

This commit is contained in:
2026-09-28 05:11:22 +08:00
parent bcd1371019
commit 834f1f26a9
5 changed files with 173 additions and 3 deletions
+5 -1
View File
@@ -134,7 +134,11 @@ With `OTEL_ENABLED` set, the app exports over OTLP/HTTP (kaya-otel plus
database spans for the Postgres (asyncpg, through Tortoise) and Redis
clients. Redis command values are redacted by the instrumentation
(`GET ? ?`) and asyncpg spans carry the parameterized SQL only — no
parameter values.
parameter values. Each HTTP/WebSocket span is tagged with the request's
session and user, so all traces of a session can be found together:
`session.id` is a truncated SHA-256 of the session id (the cookie is a
credential, so it is never exported verbatim) and `user.id` is the OIDC
`sub`, present once logged in.
- **Metrics**: the HTTP server metrics from kaya-otel, plus
`tavolo.active_users` (sessions in the session store),
`tavolo.ongoing_games` (live matches not finished yet, maintained by the
@@ -52,6 +52,7 @@ from .telemetry import (
CountingInMemorySessionStore,
CountingRedisSessionStore,
otel_headers,
session_tag_hook,
telemetry_from_settings,
)
@@ -111,6 +112,10 @@ def otel_mixin_from_settings(
"endpoint": settings.otel_exporter_endpoint,
"headers": otel_headers(settings),
"excluded_paths": settings.otel_excluded_paths,
# Runs after the session mixin's after hook, so the session id has
# been persisted/minted and the OIDC claims are on the session.
"server_response_hook": session_tag_hook,
"websocket_close_hook": session_tag_hook,
}
if meter_provider is not None:
kwargs["meter_provider"] = meter_provider
@@ -12,7 +12,10 @@ the test suite never need them):
CPU time/utilization, memory usage and CPython garbage-collection
counters;
* OTLP log export through a stdlib :class:`logging.Handler` attached to
the root logger, mirroring what the console sees.
the root logger, mirroring what the console sees;
* session/user tagging of request and websocket spans through
:func:`session_tag_hook` (kaya-otel's response/close hooks), so traces
can be correlated with a session (hashed id) and an OIDC user.
The store counts are asynchronous while OpenTelemetry instrument
callbacks are synchronous, so :class:`TelemetryMixin` polls the stores on
@@ -21,10 +24,11 @@ a background task and its callbacks report the cached values.
from __future__ import annotations
import asyncio
import hashlib
import logging
from dataclasses import dataclass
from logging import getLogger
from typing import TYPE_CHECKING, Any, Dict, Optional, Union
from typing import TYPE_CHECKING, Any, Dict, Mapping, Optional, Union
from kaya.core import KayaApp, KayaMixin
from kaya.session import InMemorySessionStore, SessionStore
@@ -79,6 +83,38 @@ def _signal_endpoint(base: Optional[str], signal: str) -> Optional[str]:
return f"{base}/{signal}" if base else None
def session_tag_hook(span: Any, carrier: Any) -> None:
"""Tag an HTTP request/websocket span with the request's session/user.
Wired into kaya-otel as its ``server_response_hook`` and
``websocket_close_hook``: those run after the session mixin's after
hook has persisted (and, at login, minted) the session, so the final
session id is visible here.
The session id is the session cookie, i.e. a credential, so only a
truncated SHA-256 of it is exported; session ids are
``secrets.token_urlsafe(32)`` (256 bits), which makes the digest
irreversible in practice. The OIDC ``sub`` is a pseudonymous
identifier already persisted in Postgres and served by the API, so it
is exported verbatim as ``user.id`` (OpenTelemetry's current end-user
attribute; ``enduser.id`` is PII-flagged and being superseded).
"""
session = getattr(carrier, "session", None)
if session is None:
return
session_id = getattr(session, "id", None)
if session_id:
span.set_attribute(
"session.id",
hashlib.sha256(session_id.encode()).hexdigest()[:16],
)
user = session.get("oidc_user")
if isinstance(user, Mapping):
sub = user.get("sub")
if sub:
span.set_attribute("user.id", str(sub))
class _NoOtelLogs(logging.Filter):
"""Drop OpenTelemetry's own log records.
+10
View File
@@ -16,6 +16,7 @@ from unittest.mock import patch
from tavolo.app import otel_mixin_from_settings
from tavolo.config import Settings
from tavolo.telemetry import session_tag_hook
def _settings(env: dict) -> Settings:
@@ -58,8 +59,17 @@ class OTelMixinFromSettingsTests(unittest.TestCase):
"endpoint": "http://collector:4318",
"headers": {"Authorization": "Bearer t", "X-Tenant": "one"},
"excluded_paths": ("/api/health",),
"server_response_hook": session_tag_hook,
"websocket_close_hook": session_tag_hook,
}, mixin.kwargs)
def test_session_tag_hooks_are_wired(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"}))
assert isinstance(mixin, _StubOTelMixin)
self.assertIs(session_tag_hook, mixin.kwargs["server_response_hook"])
self.assertIs(session_tag_hook, mixin.kwargs["websocket_close_hook"])
def test_defaults_when_only_enabled(self):
with _stub_kaya_otel():
mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "on"}))
+115
View File
@@ -5,6 +5,7 @@ so the provider-dependent tests skip when they are not installed.
"""
from __future__ import annotations
import hashlib
import logging
import os
import sys
@@ -19,6 +20,7 @@ from tavolo.telemetry import (
CountingRedisSessionStore,
TelemetryMixin,
_NoOtelLogs,
session_tag_hook,
telemetry_from_settings,
)
from tavolo.config import Settings
@@ -137,6 +139,119 @@ class NoOtelLogsFilterTest(unittest.TestCase):
)
class _RecordedSpan:
def __init__(self) -> None:
self.attributes: dict = {}
def set_attribute(self, key: str, value) -> None:
self.attributes[key] = value
class SessionTagHookTest(unittest.TestCase):
def test_tags_authenticated_session(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session("sid", {"oidc_user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertEqual(
hashlib.sha256(b"sid").hexdigest()[:16],
span.attributes["session.id"],
)
self.assertEqual("alice", span.attributes["user.id"])
def test_tags_session_without_user(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace(session=Session("sid", {})))
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
def test_tags_user_when_session_id_not_yet_minted(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(
session=Session(None, {"oidc_user": {"sub": "alice"}})
)
session_tag_hook(span, carrier)
self.assertNotIn("session.id", span.attributes)
self.assertEqual("alice", span.attributes["user.id"])
def test_missing_session_is_a_noop(self) -> None:
span = _RecordedSpan()
session_tag_hook(span, SimpleNamespace())
self.assertEqual({}, span.attributes)
def test_malformed_user_is_ignored(self) -> None:
span = _RecordedSpan()
carrier = SimpleNamespace(session=Session("sid", {"oidc_user": "alice"}))
session_tag_hook(span, carrier)
self.assertIn("session.id", span.attributes)
self.assertNotIn("user.id", span.attributes)
@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed")
class SessionTagIntegrationTest(unittest.TestCase):
"""The hook is wired as kaya-otel's response hook, so it sees the
session after the session mixin persisted and minted it."""
@async_test
async def test_spans_carry_hashed_session_and_oidc_sub(self) -> None:
from httpx import ASGITransport, AsyncClient
from kaya.core import HttpContext, KayaApp
from kaya.otel import OTelMixin
from kaya.session import InMemorySessionStore, SessionMixin
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider(
resource=Resource.create({"service.name": "test"})
)
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
app = KayaApp(
mixins=[
OTelMixin(
service_name="test",
tracer_provider=tracer_provider,
meter_provider=MeterProvider(
metric_readers=[InMemoryMetricReader()]
),
server_response_hook=session_tag_hook,
websocket_close_hook=session_tag_hook,
),
SessionMixin(InMemorySessionStore()),
]
)
@app.GET("/login")
async def login(ctx: HttpContext) -> None:
session = ctx.session
assert session is not None
session["oidc_user"] = {"sub": "alice"}
await ctx.send_str(200, "ok")
@app.GET("/me")
async def me(ctx: HttpContext) -> None:
await ctx.send_str(200, "ok")
transport = ASGITransport(app=app) # type: ignore[arg-type]
async with AsyncClient(
transport=transport, base_url="http://127.0.0.1"
) as client:
first = await client.get("/login")
second = await client.get("/me")
self.assertEqual(200, first.status_code)
self.assertEqual(200, second.status_code)
# The cookie value is the session id; only its digest is exported.
cookie = first.cookies["session_id"]
digest = hashlib.sha256(cookie.encode()).hexdigest()[:16]
spans = {span.name: span for span in span_exporter.get_finished_spans()}
self.assertEqual({"GET /login", "GET /me"}, set(spans))
for name, span in spans.items():
attributes = span.attributes
assert attributes is not None
self.assertEqual(digest, attributes["session.id"], name)
self.assertEqual("alice", attributes["user.id"], name)
def _metric_points(reader) -> dict:
"""Map metric name -> list of (attributes, value) data points."""
data = reader.get_metrics_data()