From 4ed14b42a70c16f7ef6b79ce74dd92ff6ef98a36 Mon Sep 17 00:00:00 2001 From: Walter Oggioni Date: Fri, 23 Oct 2026 06:09:43 +0000 Subject: [PATCH] Add optional OpenTelemetry traces, metrics and OTLP logs --- README.md | 36 +- docker-compose.yml | 8 + server/.env.example | 13 + server/Dockerfile | 4 +- server/packages/kca-app/src/kca/app.py | 83 ++++- server/packages/kca-app/src/kca/telemetry.py | 368 +++++++++++++++++++ server/tests/test_otel.py | 112 ++++++ server/tests/test_telemetry.py | 200 ++++++++++ 8 files changed, 813 insertions(+), 11 deletions(-) create mode 100644 server/packages/kca-app/src/kca/telemetry.py create mode 100644 server/tests/test_otel.py create mode 100644 server/tests/test_telemetry.py diff --git a/README.md b/README.md index 1d3d9d8..ce1285e 100644 --- a/README.md +++ b/README.md @@ -204,13 +204,47 @@ DATABASE_URL=postgres://kca:password@localhost:5432/kca uv run aerich migrate -- DATABASE_URL=postgres://kca:password@localhost:5432/kca uv run aerich upgrade ``` +## OpenTelemetry + +OpenTelemetry is optional and disabled by default. The Docker image includes +the `otel` extra; for a non-image install run: + +```sh +uv sync --package kca-app --extra otel +``` + +Enable export with: + +```sh +OTEL_ENABLED=1 +OTEL_SERVICE_NAME=kca +OTEL_EXPORTER_OTLP_ENDPOINT=http://collector:4318 +# Optional comma-separated "Name=Value" pairs: +OTEL_EXPORTER_OTLP_HEADERS="Authorization=Bearer token,X-Tenant=one" +``` + +The OTLP/HTTP endpoints are derived automatically (`/v1/traces`, +`/v1/metrics`, `/v1/logs`). KCA exports: + +- HTTP server spans and metrics through `kaya-otel`, with W3C trace-context + propagation and health-check exclusion; +- asyncpg and Redis client spans; +- the `kca.active_sessions` gauge plus process/GC system metrics; +- application logs through the OTLP log exporter (OpenTelemetry's own SDK + logs are filtered to avoid export loops); +- `session.id` (truncated digest) and `user.id` span attributes for trace + correlation. + +`OTEL_EXCLUDED_PATHS`, `OTEL_METRIC_EXPORT_INTERVAL_MILLIS` and +`OTEL_METRICS_POLL_INTERVAL_SECONDS` are documented in `server/.env.example`. + ## Layout ``` server/ packages/kca-core/ # pure PKI domain (keys, names, extensions, certs, CSRs, # CRLs, templates, imports/exports, .xdb, AES-GCM) - packages/kca-app/ # kaya app, config, Tortoise models, routes, static + packages/kca-app/ # kaya app, config, telemetry, Tortoise models, routes, static migrations/ # aerich schema history tests/ # app integration suite dev/mockoauth/ # dev OIDC provider image + test users diff --git a/docker-compose.yml b/docker-compose.yml index 433ec35..310cdc3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -97,6 +97,14 @@ services: # openssl rand -base64 32 KCA_MASTER_KEY: ${KCA_MASTER_KEY:-MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=} MAX_UPLOAD_BYTES: ${MAX_UPLOAD_BYTES:-20971520} + # Optional OpenTelemetry export (OTLP/HTTP). Set OTEL_ENABLED=1 and + # point OTEL_EXPORTER_OTLP_ENDPOINT at a collector, e.g. + # http://otel-collector:4318 + OTEL_ENABLED: ${OTEL_ENABLED:-false} + OTEL_SERVICE_NAME: ${OTEL_SERVICE_NAME:-kca} + OTEL_EXPORTER_OTLP_ENDPOINT: ${OTEL_EXPORTER_OTLP_ENDPOINT:-} + OTEL_EXPORTER_OTLP_HEADERS: ${OTEL_EXPORTER_OTLP_HEADERS:-} + OTEL_EXCLUDED_PATHS: ${OTEL_EXCLUDED_PATHS:-/api/health} ports: - "127.0.0.1:${APP_PORT:-8080}:8080" diff --git a/server/.env.example b/server/.env.example index 72dcd0b..2700e09 100644 --- a/server/.env.example +++ b/server/.env.example @@ -38,6 +38,19 @@ KEYGEN_CONCURRENCY=2 SESSION_COOKIE_SECURE=false #LOGGING_CONFIG= +# --- OpenTelemetry (optional) ---------------------------------------------- +# The Docker image includes the 'otel' extra. Non-image installs need: +# uv sync --package kca-app --extra otel +# Export targets are OTLP/HTTP; the /v1/traces, /v1/metrics and /v1/logs +# paths are appended automatically. +OTEL_ENABLED=false +OTEL_SERVICE_NAME=kca +#OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318 +#OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer token,X-Tenant=one +OTEL_EXCLUDED_PATHS=/api/health +#OTEL_METRIC_EXPORT_INTERVAL_MILLIS=60000 +#OTEL_METRICS_POLL_INTERVAL_SECONDS=15 + # --- CORS (optional) ------------------------------------------------------- #CORS_ALLOW_ORIGINS= #CORS_ALLOW_ORIGIN_REGEX= diff --git a/server/Dockerfile b/server/Dockerfile index b139052..ccfdbab 100644 --- a/server/Dockerfile +++ b/server/Dockerfile @@ -46,11 +46,11 @@ COPY server/uv.lock server/pyproject.toml ./ COPY server/packages/kca-core/pyproject.toml packages/kca-core/ COPY server/packages/kca-app/pyproject.toml packages/kca-app/ RUN --mount=type=cache,target=/root/.cache/uv \ - uv sync --frozen --no-install-workspace --package kca-app --no-dev + uv sync --frozen --no-install-workspace --package kca-app --extra otel --no-dev COPY server/packages/ ./packages/ RUN --mount=type=cache,target=/root/.cache/uv \ - uv sync --locked --no-editable --package kca-app --no-dev \ + uv sync --locked --no-editable --package kca-app --extra otel --no-dev \ --reinstall-package kca-core --reinstall-package kca-app COPY server/migrations ./migrations diff --git a/server/packages/kca-app/src/kca/app.py b/server/packages/kca-app/src/kca/app.py index d61e253..00a42cd 100644 --- a/server/packages/kca-app/src/kca/app.py +++ b/server/packages/kca-app/src/kca/app.py @@ -1,8 +1,10 @@ """Application entry point. Assembles the kaya app with sessions, OIDC, Tortoise, OpenAPI, the KCA API -and the SPA shell. Optional CORS mirrors spendwise's environment-driven -configuration. +and the SPA shell. Optional CORS mirrors spendwise's environment-driven +configuration. When ``OTEL_ENABLED`` is set, kaya-otel adds HTTP traces and +metrics, and :class:`~kca.telemetry.TelemetryMixin` adds process metrics, +database spans and OTLP log export. """ from __future__ import annotations @@ -19,6 +21,14 @@ from .config import Settings, settings from .logging_config import configure_logging from .mixin import Kca, KcaMixin from .openapi import SCHEMAS, SECURITY_SCHEMES, KcaOpenAPI +from .telemetry import ( + CountableSessionStore, + CountingInMemorySessionStore, + CountingRedisSessionStore, + otel_headers, + session_tag_hook, + telemetry_from_settings, +) from .tortoise_mixin import TortoiseMixin configure_logging(settings.logging_config) @@ -39,6 +49,45 @@ def cors_mixin_from_settings(settings: Settings) -> Optional[CorsMixin]: ) +def otel_mixin_from_settings( + settings: Settings, + meter_provider: Optional[Any] = None, + tracer_provider: Optional[Any] = None, +) -> Optional[KayaMixin]: + """Build a :class:`~kaya.otel.OTelMixin` from the OTEL_* settings. + + Returns ``None`` — telemetry disabled — unless ``OTEL_ENABLED`` is + truthy. kaya-otel is an optional dependency (the ``otel`` extra), so it + is imported lazily here: default installs and the test suite never need + the OpenTelemetry packages. + """ + if not settings.otel_enabled: + return None + try: + from kaya.otel import OTelMixin + except ImportError as exc: + raise RuntimeError( + "OTEL_ENABLED is set but kaya-otel is not installed; " + "install kca-app with the 'otel' extra" + ) from exc + kwargs: Dict[str, Any] = { + "service_name": settings.otel_service_name, + "endpoint": settings.otel_exporter_endpoint, + "headers": otel_headers(settings), + "excluded_paths": settings.otel_excluded_paths, + "metric_export_interval_millis": settings.otel_metric_export_interval_millis, + # 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 + if tracer_provider is not None: + kwargs["tracer_provider"] = tracer_provider + return OTelMixin(**kwargs) + + if settings.master_key_is_dev: log.warning( "KCA_MASTER_KEY is unset: using the built-in development key. " @@ -46,14 +95,10 @@ if settings.master_key_is_dev: ) if settings.redis_url is not None: - from kaya.session.redis import RedisSessionStore - - session_store: Any = RedisSessionStore(Redis.from_url(settings.redis_url)) + session_store: CountableSessionStore = CountingRedisSessionStore(Redis.from_url(settings.redis_url)) log.info("using the Redis session store") else: - from kaya.session import InMemorySessionStore - - session_store = InMemorySessionStore() + session_store = CountingInMemorySessionStore() log.info("REDIS_URL unset: using the in-memory session store") session_mixin = SessionMixin(session_store, secure=settings.session_cookie_secure) @@ -99,6 +144,28 @@ mixins: list[KayaMixin] = [ KcaMixin(kca), ] +telemetry = telemetry_from_settings(settings, session_store) +if telemetry is not None: + otel_mixin = otel_mixin_from_settings( + settings, + meter_provider=telemetry.meter_provider, + tracer_provider=telemetry.tracer_provider, + ) + assert otel_mixin is not None # OTEL_ENABLED was checked above + # kaya-otel first: before hooks run in registration order (after hooks + # in reverse), so the span covers session loading, OIDC handling and + # the handler itself. CORS, when enabled, is still prepended before it + # so preflight short-circuits stay untraced. The telemetry mixin owns + # the metric/logger providers' polling and shutdown, so it must be in + # the list too. + mixins.insert(0, telemetry.mixin) + mixins.insert(0, otel_mixin) + log.info( + "OpenTelemetry enabled: service=%s endpoint=%s (traces, metrics, logs)", + settings.otel_service_name, + settings.otel_exporter_endpoint or "(OTLP default)", + ) + cors_mixin = cors_mixin_from_settings(settings) if cors_mixin is not None: mixins.insert(0, cors_mixin) diff --git a/server/packages/kca-app/src/kca/telemetry.py b/server/packages/kca-app/src/kca/telemetry.py new file mode 100644 index 0000000..e5c31f8 --- /dev/null +++ b/server/packages/kca-app/src/kca/telemetry.py @@ -0,0 +1,368 @@ +"""OpenTelemetry metrics and log export beyond kaya-otel. + +kaya-otel instruments HTTP requests with traces and HTTP metrics; this +module adds what it does not cover, all behind the optional ``otel`` extra +(OpenTelemetry imports are lazy, so default installs and the test suite +never need them): + +* the application gauge ``kca.active_sessions`` (sessions held in the + session store); +* process metrics through ``opentelemetry-instrumentation-system-metrics``: + 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; +* session/user tagging of request spans through :func:`session_tag_hook` + (kaya-otel's response hook), 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 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 kaya.core import KayaApp, KayaMixin +from kaya.session import InMemorySessionStore +from kaya.session.redis import RedisSessionStore +from redis.asyncio import Redis + +from .auth import cached_user, session_user +from .config import Settings + +if TYPE_CHECKING: + from opentelemetry.sdk._logs import LoggerProvider + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.trace import TracerProvider + +log = getLogger(__name__) + +# Session keys live under kaya's default prefix. Keeping it explicit here +# makes the Redis count scan exact and preserves existing sessions. +SESSION_KEY_PREFIX = "kaya:session:" + +# The interpreter metrics to export. The instrumentation package can also +# emit host-level ``system.*`` metrics; those are deliberately left off: +# the application cares about its own process, and per-CPU host metrics +# add cardinality without context. +PROCESS_METRICS_CONFIG: Dict[str, Optional[list]] = { + "process.cpu.time": ["user", "system"], + "process.cpu.utilization": None, + "process.memory.usage": None, + "process.memory.virtual": None, + "cpython.gc.collections": None, + "cpython.gc.collected_objects": None, + "cpython.gc.uncollectable_objects": None, +} + + +def otel_headers(settings: Settings) -> Optional[Dict[str, str]]: + """Parse ``OTEL_EXPORTER_OTLP_HEADERS`` into a mapping (or ``None``).""" + headers = dict( + pair.split("=", 1) + for pair in (settings.otel_exporter_headers or ()) + if "=" in pair + ) + return headers or None + + +def _signal_endpoint(base: Optional[str], signal: str) -> Optional[str]: + """``{base}/v1/``, or ``None`` for the exporter's own default.""" + return f"{base}/v1/{signal}" if base else None + + +def session_tag_hook(span: Any, carrier: Any) -> None: + """Tag an HTTP request span with the request's session/user. + + Wired into kaya-otel as its ``server_response_hook``: it runs 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). + Requests authenticated with a personal access token have no session + user; their ``user.id`` comes from the user the auth guard resolved + and cached on the request. + """ + 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], + ) + sub = None + user = session_user(session) + if user is not None: + sub = user.get("sub") + if sub is None: + cached = cached_user(carrier) + if cached is not None: + sub = cached.sub + if sub: + span.set_attribute("user.id", str(sub)) + + +class _NoOtelLogs(logging.Filter): + """Drop OpenTelemetry's own log records. + + Export failures are logged by the SDK; forwarding those through the + same exporter would feed an endless retry loop. + """ + + def filter(self, record: logging.LogRecord) -> bool: + return not record.name.startswith("opentelemetry") + + +class CountingInMemorySessionStore(InMemorySessionStore): + """In-memory session store that can report its live session count.""" + + async def count(self) -> int: + now = self._clock() + return sum( + 1 + for session_id in self._data + if session_id not in self._expires + or self._expires[session_id] > now + ) + + +class CountingRedisSessionStore(RedisSessionStore): + """Redis session store that can report its live session count.""" + + def __init__(self, client: Redis, prefix: str = SESSION_KEY_PREFIX) -> None: + super().__init__(client, prefix=prefix) + + async def count(self) -> int: + total = 0 + async for _ in self._client.scan_iter(match=f"{self._prefix}*", count=500): + total += 1 + return total + + +CountableSessionStore = Union[CountingInMemorySessionStore, CountingRedisSessionStore] + + +class TelemetryMixin(KayaMixin): + """Application gauge, DB instrumentation and log export. + + The meter, tracer and logger providers are built by + :func:`telemetry_from_settings` and injected: kaya-otel only shuts + down providers it created itself, so this mixin owns their shutdown + (and uninstruments the database clients it instrumented). + """ + + def __init__( + self, + *, + meter_provider: "MeterProvider", + tracer_provider: "TracerProvider", + logger_provider: Optional["LoggerProvider"], + log_handler: Optional[logging.Handler], + session_store: CountableSessionStore, + poll_interval_seconds: float, + ) -> None: + self._meter_provider = meter_provider + self._tracer_provider = tracer_provider + self._logger_provider = logger_provider + self._log_handler = log_handler + self._session_store = session_store + self._poll_interval = poll_interval_seconds + self._active_sessions: Optional[int] = None + self._poll_task: Optional[asyncio.Task] = None + self._instrumentor: Optional[Any] = None + self._db_instrumentors: list = [] + + def apply(self, app: KayaApp) -> None: + from opentelemetry.instrumentation.asyncpg import AsyncPGInstrumentor + from opentelemetry.instrumentation.redis import RedisInstrumentor + from opentelemetry.instrumentation.system_metrics import ( + SystemMetricsInstrumentor, + ) + from opentelemetry.metrics import CallbackOptions, Observation + + instrumentor = SystemMetricsInstrumentor(config=PROCESS_METRICS_CONFIG) + self._instrumentor = instrumentor + instrumentor.instrument(meter_provider=self._meter_provider) + + # Query spans for the Postgres (asyncpg, used by Tortoise) and Redis + # clients. Redis command values are redacted by the instrumentation + # (``SET ? ?``); asyncpg spans carry the parameterized SQL only. + self._db_instrumentors = [ + AsyncPGInstrumentor(capture_parameters=False), + RedisInstrumentor(), + ] + for db_instrumentor in self._db_instrumentors: + db_instrumentor.instrument(tracer_provider=self._tracer_provider) + + meter = self._meter_provider.get_meter(__name__) + + def active_sessions(options: CallbackOptions): + if self._active_sessions is not None: + yield Observation(self._active_sessions) + + meter.create_observable_gauge( + "kca.active_sessions", + callbacks=[active_sessions], + unit="{session}", + description="Sessions currently held in the session store.", + ) + + # -- lifecycle ---------------------------------------------------------- + + def setup(self, loop: asyncio.AbstractEventLoop) -> None: + self._poll_task = loop.create_task(self._poll()) + + def shutdown(self, loop: asyncio.AbstractEventLoop) -> None: + if self._poll_task is not None: + self._poll_task.cancel() + self._poll_task = None + for db_instrumentor in self._db_instrumentors: + db_instrumentor.uninstrument() + self._db_instrumentors = [] + if self._log_handler is not None: + logging.getLogger().removeHandler(self._log_handler) + self._log_handler.close() + self._log_handler = None + if self._logger_provider is not None: + self._logger_provider.shutdown() + self._tracer_provider.shutdown() + self._meter_provider.shutdown() + + async def _poll(self) -> None: + while True: + await self.refresh_counts() + await asyncio.sleep(self._poll_interval) + + async def refresh_counts(self) -> None: + """Refresh the cached store counts (the poll loop calls this).""" + try: + self._active_sessions = await self._session_store.count() + except Exception: + log.warning("active_sessions refresh failed", exc_info=True) + + +@dataclass +class Telemetry: + """The providers and mixin built from the ``OTEL_*`` settings.""" + + meter_provider: "MeterProvider" + tracer_provider: "TracerProvider" + logger_provider: Optional["LoggerProvider"] + log_handler: Optional[logging.Handler] + mixin: TelemetryMixin + + +def telemetry_from_settings( + settings: Settings, + session_store: CountableSessionStore, +) -> Optional[Telemetry]: + """Build the metric/trace/log providers, or ``None`` when telemetry is off. + + Only called when ``OTEL_ENABLED`` is set; the OpenTelemetry imports + are lazy so the default install does not need the ``otel`` extra. + """ + if not settings.otel_enabled: + return None + try: + from opentelemetry.exporter.otlp.proto.http._log_exporter import ( + OTLPLogExporter, + ) + from opentelemetry.exporter.otlp.proto.http.metric_exporter import ( + OTLPMetricExporter, + ) + from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( + OTLPSpanExporter, + ) + from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler + from opentelemetry.sdk._logs.export import BatchLogRecordProcessor + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + except ImportError as exc: + raise RuntimeError( + "OTEL_ENABLED is set but the OpenTelemetry dependencies are not " + "installed; install kca-app with the 'otel' extra" + ) from exc + + endpoint = settings.otel_exporter_endpoint + headers = otel_headers(settings) + resource = Resource.create({"service.name": settings.otel_service_name}) + tracer_provider = TracerProvider(resource=resource) + tracer_provider.add_span_processor( + BatchSpanProcessor( + OTLPSpanExporter( + endpoint=_signal_endpoint(endpoint, "traces"), + headers=headers, + ) + ) + ) + meter_provider = MeterProvider( + resource=resource, + metric_readers=[ + PeriodicExportingMetricReader( + OTLPMetricExporter( + endpoint=_signal_endpoint(endpoint, "metrics"), + headers=headers, + ), + export_interval_millis=settings.otel_metric_export_interval_millis, + ) + ], + ) + logger_provider = LoggerProvider(resource=resource) + logger_provider.add_log_record_processor( + BatchLogRecordProcessor( + OTLPLogExporter( + endpoint=_signal_endpoint(endpoint, "logs"), + headers=headers, + ) + ) + ) + # KCA's console logging defaults to DEBUG, including per-query Tortoise + # records. Export INFO+ so the OTLP log volume stays useful. + handler = LoggingHandler(level=logging.INFO, logger_provider=logger_provider) + handler.addFilter(_NoOtelLogs()) + logging.getLogger().addHandler(handler) + mixin = TelemetryMixin( + meter_provider=meter_provider, + tracer_provider=tracer_provider, + logger_provider=logger_provider, + log_handler=handler, + session_store=session_store, + poll_interval_seconds=settings.otel_metrics_poll_interval_seconds, + ) + return Telemetry( + meter_provider=meter_provider, + tracer_provider=tracer_provider, + logger_provider=logger_provider, + log_handler=handler, + mixin=mixin, + ) + + +__all__ = [ + "CountableSessionStore", + "CountingInMemorySessionStore", + "CountingRedisSessionStore", + "Telemetry", + "TelemetryMixin", + "otel_headers", + "session_tag_hook", + "telemetry_from_settings", +] diff --git a/server/tests/test_otel.py b/server/tests/test_otel.py new file mode 100644 index 0000000..a3105f7 --- /dev/null +++ b/server/tests/test_otel.py @@ -0,0 +1,112 @@ +"""OpenTelemetry wiring tests. + +The actual OpenTelemetry packages are an optional extra (``kca-app[otel]``), +so the kaya-otel mixin is stubbed here. Provider-dependent tests live in +``test_telemetry.py``. +""" +from __future__ import annotations + +import os +import sys +import types +import unittest +from unittest.mock import patch + +from kca.app import otel_mixin_from_settings +from kca.config import Settings +from kca.telemetry import session_tag_hook + + +def _settings(env: dict) -> Settings: + with patch.dict(os.environ, env, clear=True): + return Settings.from_env() + + +class _StubOTelMixin: + def __init__(self, **kwargs): + self.kwargs = kwargs + + +def _stub_kaya_otel(): + module = types.ModuleType("kaya.otel") + module.OTelMixin = _StubOTelMixin # type: ignore[attr-defined] + return patch.dict(sys.modules, {"kaya.otel": module}) + + +class OTelMixinFromSettingsTests(unittest.TestCase): + def test_disabled_by_default(self): + self.assertIsNone(otel_mixin_from_settings(_settings({}))) + + def test_enabled_by_otel_enabled(self): + with _stub_kaya_otel(): + mixin = otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"})) + self.assertIsNotNone(mixin) + + def test_settings_are_passed_through(self): + with _stub_kaya_otel(): + mixin = otel_mixin_from_settings( + _settings( + { + "OTEL_ENABLED": "true", + "OTEL_SERVICE_NAME": "kca-test", + "OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318", + "OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer t, X-Tenant=one", + "OTEL_METRIC_EXPORT_INTERVAL_MILLIS": "15000", + } + ) + ) + assert isinstance(mixin, _StubOTelMixin) + self.assertEqual("kca-test", mixin.kwargs["service_name"]) + self.assertEqual("http://collector:4318", mixin.kwargs["endpoint"]) + self.assertEqual( + {"Authorization": "Bearer t", "X-Tenant": "one"}, + mixin.kwargs["headers"], + ) + self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"]) + self.assertEqual(15000, mixin.kwargs["metric_export_interval_millis"]) + 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"})) + assert isinstance(mixin, _StubOTelMixin) + self.assertEqual("kca", mixin.kwargs["service_name"]) + self.assertIsNone(mixin.kwargs["endpoint"]) + self.assertIsNone(mixin.kwargs["headers"]) + self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"]) + + def test_providers_are_passed_through(self): + with _stub_kaya_otel(): + meter = object() + tracer = object() + mixin = otel_mixin_from_settings( + _settings({"OTEL_ENABLED": "1"}), + meter_provider=meter, + tracer_provider=tracer, + ) + assert isinstance(mixin, _StubOTelMixin) + self.assertIs(meter, mixin.kwargs["meter_provider"]) + self.assertIs(tracer, mixin.kwargs["tracer_provider"]) + + def test_excluded_paths_are_passed_through(self): + with _stub_kaya_otel(): + mixin = otel_mixin_from_settings( + _settings( + { + "OTEL_ENABLED": "1", + "OTEL_EXCLUDED_PATHS": "/api/health,/api/docs", + } + ) + ) + assert isinstance(mixin, _StubOTelMixin) + self.assertEqual(("/api/health", "/api/docs"), mixin.kwargs["excluded_paths"]) + + def test_missing_extra_raises_runtime_error(self): + with patch.dict(sys.modules, {"kaya.otel": None}): + with self.assertRaises(RuntimeError): + otel_mixin_from_settings(_settings({"OTEL_ENABLED": "1"})) + + +if __name__ == "__main__": + unittest.main() diff --git a/server/tests/test_telemetry.py b/server/tests/test_telemetry.py new file mode 100644 index 0000000..9f028a4 --- /dev/null +++ b/server/tests/test_telemetry.py @@ -0,0 +1,200 @@ +"""Telemetry helper tests. + +Provider-dependent tests are skipped unless the optional OpenTelemetry +packages are installed. +""" +from __future__ import annotations + +import hashlib +import logging +import os +import sys +import unittest +from types import SimpleNamespace +from typing import Optional +from unittest.mock import patch + +from kaya.session import Session + +from kca.config import Settings +from kca.models import User +from kca.telemetry import ( + CountingInMemorySessionStore, + CountingRedisSessionStore, + _NoOtelLogs, + _signal_endpoint, + otel_headers, + session_tag_hook, + telemetry_from_settings, +) + +from tests.helpers.asynctest import async_test + +try: + from opentelemetry.sdk._logs import LoggingHandler # noqa: F401 + from opentelemetry.sdk.metrics import MeterProvider # noqa: F401 + from opentelemetry.sdk.trace import TracerProvider # noqa: F401 + + HAVE_OTEL = True +except ImportError: # pragma: no cover - installs without the otel extra + HAVE_OTEL = False + + +def _settings(env: dict) -> Settings: + with patch.dict(os.environ, env, clear=True): + return Settings.from_env() + + +class _FakeSessions: + def __init__(self, count: int = 0) -> None: + self.value: Optional[int] = count + + async def count(self) -> int: + assert self.value is not None + return self.value + + +class _FakeRedis: + def __init__(self, keys: list) -> None: + self.keys = keys + + async def scan_iter(self, match: str = "", count: int = 0): + prefix = match.rstrip("*") + for key in self.keys: + if key.startswith(prefix): + yield key + + +class _RecordedSpan: + def __init__(self) -> None: + self.attributes: dict = {} + + def set_attribute(self, key: str, value) -> None: + self.attributes[key] = value + + +class TelemetrySetupTest(unittest.TestCase): + def test_disabled_returns_none(self) -> None: + self.assertIsNone(telemetry_from_settings(_settings({}), _FakeSessions())) + + def test_missing_otel_dependencies_raise(self) -> None: + settings = _settings({"OTEL_ENABLED": "true"}) + missing = "opentelemetry.exporter.otlp.proto.http._log_exporter" + with patch.dict(sys.modules, {missing: None}): + with self.assertRaises(RuntimeError): + telemetry_from_settings(settings, _FakeSessions()) + + +class ProviderSetupTest(unittest.TestCase): + @unittest.skipUnless(HAVE_OTEL, "OpenTelemetry extra is not installed") + def test_builds_providers_and_installs_log_handler(self) -> None: + telemetry = telemetry_from_settings( + _settings({"OTEL_ENABLED": "1", "OTEL_SERVICE_NAME": "kca-test"}), + _FakeSessions(), + ) + assert telemetry is not None + self.assertIsNotNone(telemetry.meter_provider) + self.assertIsNotNone(telemetry.tracer_provider) + self.assertIsNotNone(telemetry.logger_provider) + self.assertIn(telemetry.log_handler, logging.getLogger().handlers) + try: + self.assertIsNotNone(telemetry.mixin) + finally: + logging.getLogger().removeHandler(telemetry.log_handler) + telemetry.log_handler.close() + telemetry.logger_provider.shutdown() + telemetry.tracer_provider.shutdown() + telemetry.meter_provider.shutdown() + + +class HeaderAndFilterTest(unittest.TestCase): + def test_otel_headers(self) -> None: + settings = _settings( + {"OTEL_EXPORTER_OTLP_HEADERS": "Authorization=Bearer x, X-Tenant=one"} + ) + self.assertEqual( + {"Authorization": "Bearer x", "X-Tenant": "one"}, + otel_headers(settings), + ) + + def test_empty_headers(self) -> None: + self.assertIsNone(otel_headers(_settings({}))) + + def test_signal_endpoint(self) -> None: + self.assertEqual( + "http://collector:4318/v1/traces", + _signal_endpoint("http://collector:4318", "traces"), + ) + self.assertIsNone(_signal_endpoint(None, "metrics")) + + def test_no_otel_logs_filter(self) -> None: + filter_ = _NoOtelLogs() + self.assertFalse(filter_.filter(logging.makeLogRecord({"name": "opentelemetry.sdk"}))) + self.assertTrue(filter_.filter(logging.makeLogRecord({"name": "kca.app"}))) + self.assertTrue(filter_.filter(logging.makeLogRecord({"name": "granian.server"}))) + + +class SessionCountTest(unittest.TestCase): + @async_test + async def test_in_memory_counts_live_sessions(self) -> None: + now = [1000.0] + store = CountingInMemorySessionStore(clock=lambda: now[0]) + await store.save("a", Session("a", {"user": "alice"}), max_age=10) + await store.save("b", Session("b", {"user": "bob"}), max_age=10) + await store.save("c", Session("c", {"user": "carol"})) + self.assertEqual(3, await store.count()) + now[0] += 11 + self.assertEqual(1, await store.count()) + await store.delete("c") + self.assertEqual(0, await store.count()) + + @async_test + async def test_redis_counts_only_session_keys(self) -> None: + store = CountingRedisSessionStore( + _FakeRedis(["kaya:session:a", "kaya:session:b", "kca:other"]) + ) + self.assertEqual(2, await store.count()) + + +class SessionTagHookTest(unittest.TestCase): + def test_tags_authenticated_session(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace( + session=Session("sid", {"oidc:default: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_user_of_any_provider(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace( + session=Session("sid", {"oidc:google:user": {"sub": "alice"}}) + ) + session_tag_hook(span, carrier) + 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_token_user_cached_on_request(self) -> None: + span = _RecordedSpan() + carrier = SimpleNamespace(session=Session("sid", {})) + setattr(carrier, "_kca_user", User(sub="token-user", display_name="Token")) + session_tag_hook(span, carrier) + self.assertEqual("token-user", span.attributes["user.id"]) + + def test_ignores_requests_without_session(self) -> None: + span = _RecordedSpan() + session_tag_hook(span, SimpleNamespace()) + self.assertEqual({}, span.attributes) + + +if __name__ == "__main__": + unittest.main()