diff --git a/deploy/k8s/tavolo.yaml b/deploy/k8s/tavolo.yaml index 839d130..e65bcbc 100644 --- a/deploy/k8s/tavolo.yaml +++ b/deploy/k8s/tavolo.yaml @@ -71,13 +71,16 @@ data: # CORS_ALLOW_CREDENTIALS: "false" # CORS_EXPOSE_HEADERS: "" # CORS_MAX_AGE: "600" - # OpenTelemetry (kaya-otel): traces + metrics via OTLP/HTTP, disabled - # unless OTEL_ENABLED is set. The image bundles the otel extra. + # OpenTelemetry (kaya-otel + system-metrics): traces, metrics and logs + # via OTLP/HTTP, disabled unless OTEL_ENABLED is set. The image bundles + # the otel extra. # OTEL_ENABLED: "true" # OTEL_SERVICE_NAME: "tavolo" # OTEL_EXPORTER_OTLP_ENDPOINT: "http://otel-collector.observability:4318" # OTEL_EXPORTER_OTLP_HEADERS: "Authorization=Bearer ..." # OTEL_EXCLUDED_PATHS: "/api/health" # default; paths skipped by tracing + # OTEL_METRIC_EXPORT_INTERVAL_MILLIS: "60000" + # OTEL_METRICS_POLL_INTERVAL_SECONDS: "15" # OIDC (provider lives in another namespace). OIDC_CLIENT_ID: tavolo OIDC_POST_LOGIN_REDIRECT: / diff --git a/docker-compose.yml b/docker-compose.yml index 58f5cbc..f88ef2d 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -122,7 +122,8 @@ services: CORS_ALLOW_CREDENTIALS: ${CORS_ALLOW_CREDENTIALS:-} CORS_EXPOSE_HEADERS: ${CORS_EXPOSE_HEADERS:-} CORS_MAX_AGE: ${CORS_MAX_AGE:-} - # OpenTelemetry (kaya-otel): disabled unless OTEL_ENABLED is set. + # OpenTelemetry (kaya-otel + system-metrics): traces, metrics and + # logs via OTLP/HTTP, disabled unless OTEL_ENABLED is set. # The image bundles the otel extra (see packages/tavolo-app/pyproject.toml # and server/requirements.txt). OTEL_ENABLED: ${OTEL_ENABLED:-} diff --git a/server/.env.example b/server/.env.example index 641f488..d80e54a 100644 --- a/server/.env.example +++ b/server/.env.example @@ -61,6 +61,22 @@ TURN_TIMEOUT_SECONDS=30 # Seconds browsers may cache the preflight response (default 600): #CORS_MAX_AGE=600 +# OpenTelemetry (via kaya-otel + the contrib system-metrics package). Off +# unless OTEL_ENABLED is truthy; exports traces, HTTP/app/process metrics +# and logs over OTLP/HTTP (requires the tavolo-app "otel" extra, bundled +# in the Docker image). The endpoint defaults to the standard +# OTEL_EXPORTER_OTLP_ENDPOINT (localhost:4318) when unset. +#OTEL_ENABLED=true +#OTEL_SERVICE_NAME=tavolo +#OTEL_EXPORTER_OTLP_ENDPOINT=http://otel-collector:4318 +#OTEL_EXPORTER_OTLP_HEADERS=Authorization=Bearer token,X-Tenant=one +# Paths excluded from tracing and metrics (default /api/health). +#OTEL_EXCLUDED_PATHS=/api/health +# Metric export interval (ms, default 60000) and session/game count poll +# interval (seconds, default 15). +#OTEL_METRIC_EXPORT_INTERVAL_MILLIS=60000 +#OTEL_METRICS_POLL_INTERVAL_SECONDS=15 + # App server APP_HOST=0.0.0.0 APP_PORT=8080 diff --git a/server/README.md b/server/README.md index cfd90fe..7d98bd5 100644 --- a/server/README.md +++ b/server/README.md @@ -81,11 +81,13 @@ All configuration comes from environment variables (see `.env.example`): | `CORS_ALLOW_CREDENTIALS` | `false` | `1`/`true`/`yes`/`on` allow cookies/credentials on cross-origin requests | | `CORS_EXPOSE_HEADERS` | unset | Comma-separated response headers exposed to the browser | | `CORS_MAX_AGE` | `600` | Seconds browsers may cache the preflight response | -| `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces and metrics (the Docker image bundles the `otel` extra; non-image installs need `pip install tavolo-app[otel]`) | +| `OTEL_ENABLED` | `false` | `1`/`true`/`yes`/`on` enable OpenTelemetry traces, metrics and logs (the Docker image bundles the `otel` extra; non-image installs need `pip install tavolo-app[otel]`) | | `OTEL_SERVICE_NAME` | `tavolo` | `service.name` resource attribute of the exported telemetry | | `OTEL_EXPORTER_OTLP_ENDPOINT` | unset | Base URL of an OTLP/HTTP collector (e.g. `http://localhost:4318`); unset uses the exporter default | | `OTEL_EXPORTER_OTLP_HEADERS` | unset | Comma-separated `key=value` headers sent to the collector (e.g. authentication) | | `OTEL_EXCLUDED_PATHS` | `/api/health` | Comma-separated paths excluded from tracing and metrics (exact matches) | +| `OTEL_METRIC_EXPORT_INTERVAL_MILLIS` | `60000` | How often the OTLP metric reader exports | +| `OTEL_METRICS_POLL_INTERVAL_SECONDS` | `15` | How often the session/game count gauges are refreshed | | `APP_HOST` / `APP_PORT` | `0.0.0.0` / `8080` | Bind address | ## Logging @@ -123,6 +125,29 @@ loggers: level: INFO ``` +## OpenTelemetry + +With `OTEL_ENABLED` set, the app exports over OTLP/HTTP (kaya-otel plus +`tavolo.telemetry`): + +- **Traces**: HTTP server spans and one span per WebSocket connection. +- **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 + game store), and process/CPython metrics from + `opentelemetry-instrumentation-system-metrics`: `process.cpu.time`, + `process.cpu.utilization`, `process.memory.usage`, + `process.memory.virtual`, `cpython.gc.collections`, + `cpython.gc.collected_objects` and `cpython.gc.uncollectable_objects`. +- **Logs**: forwarded from the root logger at the configured console level + (OpenTelemetry's own records are filtered out to avoid export loops). + +The session/game gauges are polled every +`OTEL_METRICS_POLL_INTERVAL_SECONDS` (Redis-backed stores report +cluster-wide counts, in-memory stores the process's own). Note that with +multiple workers or replicas each process reports the same global counts: +aggregate with `max`, not `sum`. + ## Data model ### Redis (live games) @@ -356,6 +381,7 @@ packages/tavolo-app/ # deployable app: composition root (own pyproject) src/tavolo/ ├── app.py # session/OIDC/Tortoise/OpenAPI/Platform/DeadlineScheduler mixins ├── config.py # env -> frozen Settings + ├── telemetry.py # OTLP metric/log providers, app gauges, count poller ├── aerich_config.py # aerich CLI configuration ├── logging_config.py # logging setup └── static.py # SPA shell hosting diff --git a/server/packages/tavolo-app/README.md b/server/packages/tavolo-app/README.md index 2b3fc34..3ec1626 100644 --- a/server/packages/tavolo-app/README.md +++ b/server/packages/tavolo-app/README.md @@ -11,6 +11,10 @@ The deployable tavolo application: the composition root wiring `ScoponeEngine` with its configured timeouts, and exposes `tavolo.app:app` for granian. - `config.py` — environment-driven frozen `Settings`. +- `telemetry.py` — the optional OpenTelemetry wiring beyond kaya-otel: + meter/logger providers, the `tavolo.active_users`/`tavolo.ongoing_games` + gauges (polled from the stores), process/CPython metrics via + `opentelemetry-instrumentation-system-metrics` and OTLP log export. - `aerich_config.py` — Tortoise ORM configuration consumed by the aerich CLI (migrations live in `server/migrations/`; aerich itself is configured via the `[tool.aerich]` section of `server/pyproject.toml`). @@ -23,6 +27,7 @@ The deployable tavolo application: the composition root wiring .venv/bin/python -m mypy --namespace-packages --explicit-package-bases \ src/tavolo/app.py \ src/tavolo/config.py \ + src/tavolo/telemetry.py \ src/tavolo/static.py \ src/tavolo/aerich_config.py \ src/tavolo/logging_config.py diff --git a/server/packages/tavolo-app/pyproject.toml b/server/packages/tavolo-app/pyproject.toml index e9b624b..8b0cdbd 100644 --- a/server/packages/tavolo-app/pyproject.toml +++ b/server/packages/tavolo-app/pyproject.toml @@ -35,6 +35,7 @@ dev = [ ] otel = [ "kaya-otel>=0.0.4", + "opentelemetry-instrumentation-system-metrics>=0.66b0", ] [tool.setuptools.packages.find] diff --git a/server/packages/tavolo-app/src/tavolo/app.py b/server/packages/tavolo-app/src/tavolo/app.py index 8b5f812..5e9e631 100644 --- a/server/packages/tavolo-app/src/tavolo/app.py +++ b/server/packages/tavolo-app/src/tavolo/app.py @@ -31,14 +31,13 @@ from __future__ import annotations from importlib.metadata import version as _pkg_version from logging import getLogger -from typing import Optional +from typing import Any, Dict, Optional from kaya.core import KayaApp, KayaMixin from kaya.cors import CorsMixin from kaya.oidc import OIDCConfig, OIDCMixin from kaya.openapi import OpenAPIMixin -from kaya.session import InMemorySessionStore, SessionMixin, SessionStore -from kaya.session.redis import RedisSessionStore +from kaya.session import SessionMixin from redis.asyncio import Redis from tavolo.platform import GameRegistry, Platform, PlatformMixin from tavolo.platform.deadlines import DeadlineScheduler, DeadlineSchedulerMixin @@ -48,6 +47,13 @@ from tavolo.scopone import ScoponeEngine from .config import Settings, settings from .logging_config import configure_logging +from .telemetry import ( + CountableSessionStore, + CountingInMemorySessionStore, + CountingRedisSessionStore, + otel_headers, + telemetry_from_settings, +) configure_logging(settings.logging_config) log = getLogger(__name__) @@ -73,13 +79,20 @@ def cors_mixin_from_settings(settings: Settings) -> Optional[CorsMixin]: ) -def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]: +def otel_mixin_from_settings( + settings: Settings, meter_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. + + ``meter_provider`` is the provider built by + :func:`tavolo.telemetry.telemetry_from_settings`; injecting it lets the + custom gauges share the same OTLP pipeline (kaya-otel then leaves its + shutdown to the telemetry mixin). """ if not settings.otel_enabled: return None @@ -90,17 +103,15 @@ def otel_mixin_from_settings(settings: Settings) -> Optional[KayaMixin]: "OTEL_ENABLED is set but kaya-otel is not installed; " "install tavolo-app with the 'otel' extra" ) from exc - headers = dict( - pair.split("=", 1) - for pair in (settings.otel_exporter_headers or ()) - if "=" in pair - ) - return OTelMixin( - service_name=settings.otel_service_name, - endpoint=settings.otel_exporter_endpoint, - headers=headers or None, - excluded_paths=settings.otel_excluded_paths, - ) + 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, + } + if meter_provider is not None: + kwargs["meter_provider"] = meter_provider + return OTelMixin(**kwargs) registry = GameRegistry() @@ -117,12 +128,12 @@ log.debug( settings.turn_timeout_seconds, ) -session_store: SessionStore +session_store: CountableSessionStore game_store: GameStore if settings.redis_url is not None: # Lazy client: no connection is opened until a session is actually # loaded/saved, so importing this module never requires a live Redis. - session_store = RedisSessionStore(Redis.from_url(settings.redis_url)) + session_store = CountingRedisSessionStore(Redis.from_url(settings.redis_url)) game_store = RedisGameStore( Redis.from_url(settings.redis_url, decode_responses=False), registry, @@ -130,7 +141,7 @@ if settings.redis_url is not None: ) log.info("using Redis stores (sessions + live games, game TTL %ds)", settings.game_ttl_seconds) else: - session_store = InMemorySessionStore() + session_store = CountingInMemorySessionStore() game_store = InMemoryGameStore(registry) log.info("REDIS_URL unset: using in-memory stores (sessions + live games)") @@ -176,15 +187,22 @@ platform = Platform( mixins: list[KayaMixin] = [session_mixin, oidc_mixin, tortoise_mixin, openapi_mixin, PlatformMixin(platform), DeadlineSchedulerMixin(scheduler)] -otel_mixin = otel_mixin_from_settings(settings) -if otel_mixin is not None: - # First in the list: before hooks run in registration order (after hooks +telemetry = telemetry_from_settings(settings, session_store, game_store) +if telemetry is not None: + otel_mixin = otel_mixin_from_settings( + settings, meter_provider=telemetry.meter_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. + # preflight short-circuits stay untraced. The telemetry mixin carries no + # hooks but 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", + "OpenTelemetry enabled: service=%s endpoint=%s (traces, metrics, logs)", settings.otel_service_name, settings.otel_exporter_endpoint or "(OTLP default)", ) diff --git a/server/packages/tavolo-app/src/tavolo/config.py b/server/packages/tavolo-app/src/tavolo/config.py index bbf4dd7..a71fe66 100644 --- a/server/packages/tavolo-app/src/tavolo/config.py +++ b/server/packages/tavolo-app/src/tavolo/config.py @@ -125,6 +125,10 @@ class Settings: # Paths excluded from tracing and metrics (exact matches). Defaults to # the health endpoint, which k8s probes would otherwise spam. otel_excluded_paths: Tuple[str, ...] + # How often the OTLP metric reader exports (milliseconds) and how often + # the telemetry gauges refresh the session/game counts (seconds). + otel_metric_export_interval_millis: int + otel_metrics_poll_interval_seconds: int @staticmethod def from_env() -> "Settings": @@ -171,14 +175,20 @@ class Settings: cors_allow_credentials=_env_bool("CORS_ALLOW_CREDENTIALS"), cors_expose_headers=_env_list("CORS_EXPOSE_HEADERS"), cors_max_age=int(_env("CORS_MAX_AGE", "600")), - # OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces - # and metrics via OTLP/HTTP (requires the ``otel`` extra). + # OpenTelemetry is opt-in: set OTEL_ENABLED=1 to export traces, + # metrics and logs via OTLP/HTTP (requires the ``otel`` extra). otel_enabled=_env_bool("OTEL_ENABLED"), otel_service_name=_env("OTEL_SERVICE_NAME", "tavolo"), otel_exporter_endpoint=os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or None, # Comma-separated key=value pairs, e.g. "Authorization=Bearer x". otel_exporter_headers=_env_list("OTEL_EXPORTER_OTLP_HEADERS"), otel_excluded_paths=_env_list("OTEL_EXCLUDED_PATHS") or ("/api/health",), + otel_metric_export_interval_millis=int( + _env("OTEL_METRIC_EXPORT_INTERVAL_MILLIS", "60000") + ), + otel_metrics_poll_interval_seconds=int( + _env("OTEL_METRICS_POLL_INTERVAL_SECONDS", "15") + ), ) diff --git a/server/packages/tavolo-app/src/tavolo/telemetry.py b/server/packages/tavolo-app/src/tavolo/telemetry.py new file mode 100644 index 0000000..af11095 --- /dev/null +++ b/server/packages/tavolo-app/src/tavolo/telemetry.py @@ -0,0 +1,300 @@ +"""OpenTelemetry metrics and log export beyond kaya-otel. + +kaya-otel instruments HTTP requests and websockets 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): + +* application gauges: ``tavolo.active_users`` (sessions held in the + session store) and ``tavolo.ongoing_games`` (live matches that have not + finished yet); +* 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. + +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 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, SessionStore +from kaya.session.redis import RedisSessionStore +from redis.asyncio import Redis +from tavolo.platform.store import GameStore + +from .config import Settings + +if TYPE_CHECKING: + from opentelemetry.sdk._logs import LoggerProvider + from opentelemetry.sdk.metrics import MeterProvider + +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.cpu.utilization`` ignores its configuration value upstream +# (the callback always emits one aggregate observation), so its ``None`` +# is only for readability. +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}/{signal}" if base else None + + +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 + + +# The session stores the application builds: both expose ``count()`` on top +# of kaya's ``SessionStore`` interface. Game stores need no alias — the +# platform's ``GameStore`` declares ``count_ongoing()`` itself. +CountableSessionStore = Union[CountingInMemorySessionStore, CountingRedisSessionStore] + + +class TelemetryMixin(KayaMixin): + """Application gauges and log export on top of kaya-otel. + + The meter 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. + """ + + def __init__( + self, + *, + meter_provider: "MeterProvider", + logger_provider: Optional["LoggerProvider"], + log_handler: Optional[logging.Handler], + session_store: CountableSessionStore, + game_store: GameStore, + poll_interval_seconds: float, + ) -> None: + self._meter_provider = meter_provider + self._logger_provider = logger_provider + self._log_handler = log_handler + self._session_store = session_store + self._game_store = game_store + self._poll_interval = poll_interval_seconds + self._active_users: Optional[int] = None + self._ongoing_games: Optional[int] = None + self._poll_task: Optional[asyncio.Task] = None + self._instrumentor: Optional[Any] = None + + def apply(self, app: KayaApp) -> None: + 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) + + meter = self._meter_provider.get_meter(__name__) + + def active_users(options: CallbackOptions): + if self._active_users is not None: + yield Observation(self._active_users) + + def ongoing_games(options: CallbackOptions): + if self._ongoing_games is not None: + yield Observation(self._ongoing_games) + + meter.create_observable_gauge( + "tavolo.active_users", + callbacks=[active_users], + unit="{user}", + description="Sessions currently held in the session store.", + ) + meter.create_observable_gauge( + "tavolo.ongoing_games", + callbacks=[ongoing_games], + unit="{game}", + description="Live games that have not finished yet.", + ) + + # -- 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 + 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._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_users = await self._session_store.count() + except Exception: + log.warning("active_users refresh failed", exc_info=True) + try: + self._ongoing_games = await self._game_store.count_ongoing() + except Exception: + log.warning("ongoing_games refresh failed", exc_info=True) + + +@dataclass +class Telemetry: + """The providers and mixin built from the ``OTEL_*`` settings.""" + + meter_provider: "MeterProvider" + logger_provider: Optional["LoggerProvider"] + log_handler: Optional[logging.Handler] + mixin: TelemetryMixin + + +def telemetry_from_settings( + settings: Settings, + session_store: CountableSessionStore, + game_store: GameStore, +) -> Optional[Telemetry]: + """Build the metric/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.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 + except ImportError as exc: + raise RuntimeError( + "OTEL_ENABLED is set but the OpenTelemetry dependencies are not " + "installed; install tavolo-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}) + meter_provider = MeterProvider( + resource=resource, + metric_readers=[ + PeriodicExportingMetricReader( + OTLPMetricExporter( + endpoint=_signal_endpoint(endpoint, "v1/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, "v1/logs"), + headers=headers, + ) + ) + ) + handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider) + handler.addFilter(_NoOtelLogs()) + logging.getLogger().addHandler(handler) + mixin = TelemetryMixin( + meter_provider=meter_provider, + logger_provider=logger_provider, + log_handler=handler, + session_store=session_store, + game_store=game_store, + poll_interval_seconds=settings.otel_metrics_poll_interval_seconds, + ) + return Telemetry( + meter_provider=meter_provider, + logger_provider=logger_provider, + log_handler=handler, + mixin=mixin, + ) diff --git a/server/packages/tavolo-platform/src/tavolo/platform/store.py b/server/packages/tavolo-platform/src/tavolo/platform/store.py index 7dc9e62..fa242c5 100644 --- a/server/packages/tavolo-platform/src/tavolo/platform/store.py +++ b/server/packages/tavolo-platform/src/tavolo/platform/store.py @@ -38,6 +38,7 @@ from __future__ import annotations import asyncio import contextlib import json +import time from abc import ABC, abstractmethod from datetime import datetime from logging import getLogger @@ -54,6 +55,10 @@ GAME_KEY_PREFIX = "tavolo:game:" CODE_KEY_PREFIX = "tavolo:code:" CHANNEL_PREFIX = "tavolo:game:" DEADLINES_KEY = "tavolo:deadlines" +# Sorted set of running games (score = expiry timestamp), maintained by the +# stores so the telemetry gauge can count ongoing matches without loading +# every session. Finished games are removed on their final save. +ONGOING_GAMES_KEY = "tavolo:games:ongoing" # Sentinel pushed into in-memory subscriber queues to signal a change. _BUMP = b"update" @@ -120,6 +125,14 @@ class GameStore(ABC): async def find_by_code(self, code: str) -> Optional[GameSession]: """Return the live session for a join ``code`` or ``None``.""" + @abstractmethod + async def count_ongoing(self) -> int: + """Return the number of live games that have not finished yet. + + Lobbies and matches in progress count; matches that are over but + still retained for the final-board view do not. + """ + @abstractmethod def lock(self, game_id: str) -> AsyncContextManager[None]: """Async context manager serializing mutations of one game.""" @@ -181,9 +194,15 @@ class RedisGameStore(GameStore): async def save(self, session: GameSession) -> None: payload = json.dumps(session_to_json(session, self._registry)) + engine = self._registry.require(session.game_type) async with self._redis.pipeline(transaction=True) as pipe: pipe.set(f"{GAME_KEY_PREFIX}{session.id}", payload, ex=self._ttl) pipe.set(f"{CODE_KEY_PREFIX}{session.join_code}", session.id, ex=self._ttl) + if engine.is_finished(session): + pipe.zrem(ONGOING_GAMES_KEY, session.id) + else: + pipe.zadd(ONGOING_GAMES_KEY, {session.id: time.time() + self._ttl}) + pipe.expire(ONGOING_GAMES_KEY, self._ttl) await pipe.execute() log.debug("redis save %s (game %s, ttl %ds)", session.id, session.game_type, self._ttl) @@ -195,6 +214,17 @@ class RedisGameStore(GameStore): game_id = game_id.decode("utf-8") return await self.load(str(game_id)) + async def count_ongoing(self) -> int: + # Prune entries whose game has expired since the last poll, then + # count what is left. Both commands run in one transaction so the + # count cannot see a half-pruned set. + now = time.time() + async with self._redis.pipeline(transaction=True) as pipe: + pipe.zremrangebyscore(ONGOING_GAMES_KEY, "-inf", now) + pipe.zcard(ONGOING_GAMES_KEY) + _, count = await pipe.execute() + return int(count) + @contextlib.asynccontextmanager async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]: pubsub = self._redis.pubsub() @@ -243,6 +273,7 @@ class InMemoryGameStore(GameStore): self._registry = registry self._games: Dict[str, str] = {} self._codes: Dict[str, str] = {} + self._ongoing: Set[str] = set() self._locks: Dict[str, asyncio.Lock] = {} self._subscribers: Dict[str, Set[asyncio.Queue]] = {} self._deadlines: Dict[str, float] = {} @@ -268,6 +299,11 @@ class InMemoryGameStore(GameStore): async def save(self, session: GameSession) -> None: self._games[session.id] = json.dumps(session_to_json(session, self._registry)) self._codes[session.join_code] = session.id + engine = self._registry.require(session.game_type) + if engine.is_finished(session): + self._ongoing.discard(session.id) + else: + self._ongoing.add(session.id) async def find_by_code(self, code: str) -> Optional[GameSession]: game_id = self._codes.get(code.upper()) @@ -275,6 +311,9 @@ class InMemoryGameStore(GameStore): return None return await self.load(game_id) + async def count_ongoing(self) -> int: + return len(self._ongoing) + @contextlib.asynccontextmanager async def subscribe(self, game_id: str) -> AsyncIterator[AsyncIterator[None]]: queue: asyncio.Queue = asyncio.Queue() diff --git a/server/packages/tavolo-platform/tests/test_store.py b/server/packages/tavolo-platform/tests/test_store.py index 13c0803..5b64236 100644 --- a/server/packages/tavolo-platform/tests/test_store.py +++ b/server/packages/tavolo-platform/tests/test_store.py @@ -2,11 +2,16 @@ from __future__ import annotations import asyncio +import time import unittest from tavolo.platform import GameSession, Seat from tavolo.platform.registry import GameRegistry -from tavolo.platform.store import InMemoryGameStore +from tavolo.platform.store import ( + ONGOING_GAMES_KEY, + InMemoryGameStore, + RedisGameStore, +) from helpers import DummyEngine, async_test @@ -27,6 +32,81 @@ def _session(game_id: str = "g1", code: str = "CODE01") -> GameSession: return session +class _FakePipeline: + """Records commands and applies them to the fake Redis on execute.""" + + def __init__(self, redis: "_FakeRedis") -> None: + self._redis = redis + self._ops: list = [] + + async def __aenter__(self) -> "_FakePipeline": + return self + + async def __aexit__(self, *exc: object) -> bool: + return False + + def set(self, key: str, value: str, ex: object = None) -> None: + self._ops.append(("set", key)) + + def zadd(self, key: str, mapping: dict) -> None: + self._ops.append(("zadd", key, mapping)) + + def zrem(self, key: str, member: str) -> None: + self._ops.append(("zrem", key, member)) + + def expire(self, key: str, ttl: int) -> None: + self._ops.append(("expire", key)) + + def zremrangebyscore(self, key: str, minimum: object, maximum: object) -> None: + self._ops.append(("zremrangebyscore", key)) + + def zcard(self, key: str) -> None: + self._ops.append(("zcard", key)) + + async def execute(self) -> list: + results = [self._redis.apply(op) for op in self._ops] + self._ops = [] + return results + + +class _FakeRedis: + """The slice of the Redis client the game store uses.""" + + def __init__(self) -> None: + self.keys: dict = {} + self.zsets: dict = {} + + def pipeline(self, transaction: bool = True) -> _FakePipeline: + return _FakePipeline(self) + + def apply(self, op: tuple) -> object: + name, *args = op + if name == "set": + self.keys[args[0]] = "payload" + return True + if name == "zadd": + self.zsets.setdefault(args[0], {}).update(args[1]) + return 1 + if name == "zrem": + self.zsets.get(args[0], {}).pop(args[1], None) + return 1 + if name == "expire": + return True + if name == "zremrangebyscore": + zset = self.zsets.get(args[0], {}) + for member in [ + member + for member, score in zset.items() + if score <= time.time() + ]: + del zset[member] + return 0 + if name == "zcard": + return len(self.zsets.get(args[0], {})) + raise AssertionError(f"unexpected command: {name}") + + + class InMemoryGameStoreTest(unittest.TestCase): @async_test async def test_save_load_roundtrip(self) -> None: @@ -114,6 +194,28 @@ class InMemoryGameStoreTest(unittest.TestCase): ["holder-enter", "holder-exit", "contender"], order ) + @async_test + async def test_count_ongoing_tracks_unfinished_games(self) -> None: + store = InMemoryGameStore(_registry()) + self.assertEqual(0, await store.count_ongoing()) + + session = _session() + await store.save(session) + self.assertEqual(1, await store.count_ongoing()) + + other = _session(game_id="g2", code="CODE02") + await store.save(other) + self.assertEqual(2, await store.count_ongoing()) + + # A finished match is retained for the final board but no longer + # counts as ongoing. + session.state["finished"] = True + await store.save(session) + self.assertEqual(1, await store.count_ongoing()) + other.state["finished"] = True + await store.save(other) + self.assertEqual(0, await store.count_ongoing()) + @async_test async def test_deadline_queue(self) -> None: store = InMemoryGameStore(_registry()) @@ -138,5 +240,35 @@ class InMemoryGameStoreTest(unittest.TestCase): self.assertEqual(["b"], await store.due_deadlines(now=100.0)) +class RedisGameStoreTest(unittest.TestCase): + @async_test + async def test_save_marks_games_ongoing_and_finished_ones_stop_counting(self) -> None: + redis = _FakeRedis() + store = RedisGameStore(redis, _registry(), ttl_seconds=3600) # type: ignore[arg-type] + session = _session() + + await store.save(session) + self.assertEqual(1, await store.count_ongoing()) + score = redis.zsets[ONGOING_GAMES_KEY][session.id] + self.assertGreater(score, time.time()) + self.assertLessEqual(score, time.time() + 3600) + + session.state["finished"] = True + await store.save(session) + self.assertNotIn(session.id, redis.zsets[ONGOING_GAMES_KEY]) + self.assertEqual(0, await store.count_ongoing()) + + @async_test + async def test_count_ongoing_prunes_expired_entries(self) -> None: + redis = _FakeRedis() + redis.zsets[ONGOING_GAMES_KEY] = { + "stale": time.time() - 10, + "live": time.time() + 100, + } + store = RedisGameStore(redis, _registry()) # type: ignore[arg-type] + self.assertEqual(1, await store.count_ongoing()) + self.assertNotIn("stale", redis.zsets[ONGOING_GAMES_KEY]) + + if __name__ == "__main__": unittest.main() diff --git a/server/requirements.txt b/server/requirements.txt index 38e52a3..397a3d9 100644 --- a/server/requirements.txt +++ b/server/requirements.txt @@ -93,6 +93,8 @@ opentelemetry-api==1.45.0 # via # opentelemetry-exporter-http-transport # opentelemetry-exporter-otlp-proto-http + # opentelemetry-instrumentation + # opentelemetry-instrumentation-system-metrics # opentelemetry-sdk # opentelemetry-semantic-conventions opentelemetry-exporter-http-transport[urllib3]==0.66b0 @@ -103,6 +105,10 @@ opentelemetry-exporter-otlp-proto-common==1.45.0 # via opentelemetry-exporter-otlp-proto-http opentelemetry-exporter-otlp-proto-http==1.45.0 # via kaya-otel +opentelemetry-instrumentation==0.66b0 + # via opentelemetry-instrumentation-system-metrics +opentelemetry-instrumentation-system-metrics==0.66b0 + # via tavolo-app (pyproject.toml) opentelemetry-proto==1.45.0 # via # opentelemetry-exporter-otlp-proto-common @@ -113,11 +119,18 @@ opentelemetry-sdk==1.45.0 # opentelemetry-exporter-otlp-common # opentelemetry-exporter-otlp-proto-http opentelemetry-semantic-conventions==0.66b0 - # via opentelemetry-sdk + # via + # opentelemetry-instrumentation + # opentelemetry-instrumentation-system-metrics + # opentelemetry-sdk +packaging==26.3 + # via opentelemetry-instrumentation protobuf==7.36.2 # via # googleapis-common-protos # opentelemetry-proto +psutil==7.2.2 + # via opentelemetry-instrumentation-system-metrics pwo==0.1.2 # via # kaya-core @@ -151,3 +164,5 @@ typing-extensions==4.16.0 # pwo urllib3==2.8.0 # via opentelemetry-exporter-http-transport +wrapt==2.5.0 + # via opentelemetry-instrumentation diff --git a/server/tests/test_config.py b/server/tests/test_config.py index 87ba845..3cdcdd4 100644 --- a/server/tests/test_config.py +++ b/server/tests/test_config.py @@ -164,6 +164,19 @@ class OTelSettingsTests(unittest.TestCase): settings = _settings({"OTEL_EXCLUDED_PATHS": "/api/health, /metrics"}) self.assertEqual(("/api/health", "/metrics"), settings.otel_excluded_paths) + def test_otel_interval_defaults(self): + settings = _settings({}) + self.assertEqual(60000, settings.otel_metric_export_interval_millis) + self.assertEqual(15, settings.otel_metrics_poll_interval_seconds) + + def test_otel_intervals_parse(self): + settings = _settings({ + "OTEL_METRIC_EXPORT_INTERVAL_MILLIS": "5000", + "OTEL_METRICS_POLL_INTERVAL_SECONDS": "3", + }) + self.assertEqual(5000, settings.otel_metric_export_interval_millis) + self.assertEqual(3, settings.otel_metrics_poll_interval_seconds) + if __name__ == "__main__": unittest.main() diff --git a/server/tests/test_otel.py b/server/tests/test_otel.py index a204a2f..e350a91 100644 --- a/server/tests/test_otel.py +++ b/server/tests/test_otel.py @@ -69,6 +69,15 @@ class OTelMixinFromSettingsTests(unittest.TestCase): self.assertIsNone(mixin.kwargs["headers"]) self.assertEqual(("/api/health",), mixin.kwargs["excluded_paths"]) + def test_meter_provider_is_passed_through(self): + with _stub_kaya_otel(): + provider = object() + mixin = otel_mixin_from_settings( + _settings({"OTEL_ENABLED": "1"}), meter_provider=provider + ) + assert isinstance(mixin, _StubOTelMixin) + self.assertIs(provider, mixin.kwargs["meter_provider"]) + def test_excluded_paths_are_passed_through(self): with _stub_kaya_otel(): mixin = otel_mixin_from_settings(_settings({ diff --git a/server/tests/test_telemetry.py b/server/tests/test_telemetry.py new file mode 100644 index 0000000..54b3e90 --- /dev/null +++ b/server/tests/test_telemetry.py @@ -0,0 +1,253 @@ +"""Telemetry unit tests: app gauges, process metrics, log export and counts. + +The OpenTelemetry packages are an optional extra (``tavolo-app[otel]``), +so the provider-dependent tests skip when they are not installed. +""" +from __future__ import annotations + +import logging +import os +import sys +import unittest +from typing import Optional +from unittest.mock import patch + +from kaya.session import Session +from tavolo.telemetry import ( + CountingInMemorySessionStore, + CountingRedisSessionStore, + TelemetryMixin, + _NoOtelLogs, + telemetry_from_settings, +) +from tavolo.config import Settings + +from tests.helpers import async_test + +try: + from opentelemetry.sdk._logs import LoggingHandler, LoggerProvider + from opentelemetry.sdk._logs.export import ( + InMemoryLogRecordExporter, + SimpleLogRecordProcessor, + ) + from opentelemetry.sdk.metrics import MeterProvider + from opentelemetry.sdk.metrics.export import InMemoryMetricReader + from opentelemetry.sdk.resources import Resource + + 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 + self.error: Optional[Exception] = None + + async def count(self) -> int: + if self.error is not None: + raise self.error + assert self.value is not None + return self.value + + +class _FakeGames: + def __init__(self, count: int = 0) -> None: + self.value: Optional[int] = count + + async def count_ongoing(self) -> int: + assert self.value is not None + return self.value + + +class _FakeRedis: + """Minimal async Redis double covering the session-count scan.""" + + 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 TelemetrySetupTest(unittest.TestCase): + def test_disabled_returns_none(self) -> None: + self.assertIsNone( + telemetry_from_settings( + _settings({}), _FakeSessions(), _FakeGames() + ) + ) + + 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(), _FakeGames()) + + +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"})) # no expiry + self.assertEqual(3, await store.count()) + + now[0] += 11 # a and b have expired + 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", "tavolo:code:X"]) + ) + self.assertEqual(2, await store.count()) + + +class NoOtelLogsFilterTest(unittest.TestCase): + def test_filters_opentelemetry_records_only(self) -> None: + filter_ = _NoOtelLogs() + self.assertFalse( + filter_.filter(logging.makeLogRecord({"name": "opentelemetry.sdk"})) + ) + self.assertTrue( + filter_.filter(logging.makeLogRecord({"name": "tavolo.app"})) + ) + self.assertTrue( + filter_.filter(logging.makeLogRecord({"name": "granian.server"})) + ) + + +def _metric_points(reader) -> dict: + """Map metric name -> list of (attributes, value) data points.""" + data = reader.get_metrics_data() + assert data is not None + points = {} + for resource_metrics in data.resource_metrics: + for scope_metrics in resource_metrics.scope_metrics: + for metric in scope_metrics.metrics: + points[metric.name] = [ + (dict(point.attributes), point.value) + for point in metric.data.data_points + ] + return points + + +@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). + sessions = _FakeSessions(0) + games = _FakeGames(0) + + @classmethod + def setUpClass(cls) -> None: + cls.reader = InMemoryMetricReader() + resource = Resource.create({"service.name": "test"}) + cls.provider = MeterProvider(resource=resource, metric_readers=[cls.reader]) + cls.mixin = TelemetryMixin( + meter_provider=cls.provider, + logger_provider=None, + log_handler=None, + session_store=cls.sessions, + game_store=cls.games, + poll_interval_seconds=1, + ) + cls.mixin.apply(None) # type: ignore[arg-type] + + @classmethod + def tearDownClass(cls) -> None: + cls.provider.shutdown() + + @async_test + async def test_app_gauges_report_polled_counts(self) -> None: + self.sessions.value = 3 + self.sessions.error = None + self.games.value = 2 + await self.mixin.refresh_counts() + points = _metric_points(self.reader) + self.assertEqual([({}, 3)], points["tavolo.active_users"]) + self.assertEqual([({}, 2)], points["tavolo.ongoing_games"]) + + @async_test + async def test_failed_refresh_keeps_last_value(self) -> None: + self.sessions.value = 5 + self.sessions.error = None + await self.mixin.refresh_counts() + self.sessions.error = RuntimeError("redis down") + await self.mixin.refresh_counts() + points = _metric_points(self.reader) + self.assertEqual([({}, 5)], points["tavolo.active_users"]) + + @async_test + async def test_process_and_gc_metrics_are_exported(self) -> None: + await self.mixin.refresh_counts() + points = _metric_points(self.reader) + for name in ( + "process.cpu.time", + "process.cpu.utilization", + "process.memory.usage", + "process.memory.virtual", + "cpython.gc.collections", + "cpython.gc.collected_objects", + "cpython.gc.uncollectable_objects", + ): + self.assertIn(name, points) + self.assertGreater(points["process.memory.usage"][0][1], 0) + modes = {attrs["type"] for attrs, _ in points["process.cpu.time"]} + self.assertEqual({"user", "system"}, modes) + generations = { + attrs["cpython.gc.generation"] + for attrs, _ in points["cpython.gc.collections"] + } + self.assertEqual({0, 1, 2}, generations) + # Host-level metrics are deliberately not configured. + self.assertNotIn("system.cpu.time", points) + + +@unittest.skipUnless(HAVE_OTEL, "tavolo-app[otel] not installed") +class LogExportTest(unittest.TestCase): + def test_logs_are_exported_and_otel_records_filtered(self) -> None: + exporter = InMemoryLogRecordExporter() + resource = Resource.create({"service.name": "test"}) + provider = LoggerProvider(resource=resource) + provider.add_log_record_processor(SimpleLogRecordProcessor(exporter)) + handler = LoggingHandler(level=logging.NOTSET, logger_provider=provider) + handler.addFilter(_NoOtelLogs()) + root = logging.getLogger() + previous_level = root.level + root.addHandler(handler) + root.setLevel(logging.DEBUG) + try: + logging.getLogger("tavolo.test").info("exported message") + logging.getLogger("opentelemetry.test").info("filtered message") + bodies = [r.log_record.body for r in exporter.get_finished_logs()] + self.assertIn("exported message", bodies) + self.assertNotIn("filtered message", bodies) + records = exporter.get_finished_logs() + self.assertEqual( + "test", records[0].resource.attributes["service.name"] + ) + finally: + root.removeHandler(handler) + handler.close() + root.setLevel(previous_level) + provider.shutdown() + + +if __name__ == "__main__": + unittest.main()