Add optional OpenTelemetry traces, metrics and OTLP logs
This commit is contained in:
@@ -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
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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=
|
||||
|
||||
+2
-2
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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/<signal>``, 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",
|
||||
]
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user