tavolo/server/crates/tavolo-app/src/telemetry.rs

//! OpenTelemetry wiring: OTLP traces, metrics and logs.
//!
//! Mirrors `tavolo.telemetry` from the Python implementation within what the
//! Rust ecosystem offers:
//!
//! * **traces** — HTTP request spans (tagged with a hashed `session.id` and
//!   the OIDC `user.id`) and one span per WebSocket connection, exported over
//!   OTLP/HTTP;
//! * **metrics** — `tavolo.active_users` (sessions in the session store),
//!   `tavolo.ongoing_games` (live matches not finished yet) and Rust process
//!   metrics (`process.cpu.time`, `process.cpu.utilization`,
//!   `process.memory.usage`, `process.memory.virtual`) in place of the
//!   CPython-specific gauges;
//! * **logs** — tracing events forwarded to OTLP (OpenTelemetry's own records
//!   are filtered out upstream).
//!
//! Database client instrumentation (asyncpg/redis spans in Python) has no
//! direct Rust equivalent and is omitted.

use std::sync::Arc;

use crate::config::Settings;
use tavolo_platform::session::SharedSessionStore;
use tavolo_platform::store::SharedGameStore;

#[cfg(feature = "otel")]
mod inner {
    use std::collections::HashMap;
    use std::sync::atomic::{AtomicI64, Ordering};
    use std::sync::{Arc, Mutex, OnceLock};
    use std::time::Duration;

    use opentelemetry_otlp::{
        LogExporter, MetricExporter, SpanExporter, WithExportConfig, WithHttpConfig,
    };
    use opentelemetry_sdk::logs::SdkLoggerProvider;
    use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
    use opentelemetry_sdk::trace::SdkTracerProvider;
    use opentelemetry_sdk::Resource;

    use super::{Settings, SharedGameStore, SharedSessionStore};

    pub(super) static HTTP_DURATION: OnceLock<opentelemetry::metrics::Histogram<f64>> =
        OnceLock::new();

    #[derive(Clone, Copy, Default)]
    pub(super) struct ProcessSnapshot {
        pub cpu_seconds: f64,
        pub cpu_utilization: f64,
        pub memory_usage: u64,
        pub memory_virtual: u64,
    }

    pub(super) struct OtelInner {
        pub tracer_provider: SdkTracerProvider,
        pub meter_provider: SdkMeterProvider,
        pub logger_provider: SdkLoggerProvider,
        pub poll_task: tokio::task::JoinHandle<()>,
    }

    impl Drop for OtelInner {
        fn drop(&mut self) {
            self.poll_task.abort();
            let _ = self.logger_provider.shutdown();
            let _ = self.meter_provider.shutdown();
            let _ = self.tracer_provider.shutdown();
        }
    }

    fn signal_endpoint(base: Option<&str>, signal: &str) -> Option<String> {
        base.map(|base| format!("{base}/{signal}"))
    }

    fn parse_headers(raw: Option<&[String]>) -> HashMap<String, String> {
        raw.unwrap_or(&[])
            .iter()
            .filter_map(|pair| pair.split_once('='))
            .map(|(key, value)| (key.to_string(), value.trim().to_string()))
            .collect()
    }

    fn refresh_process(snapshot: &Mutex<ProcessSnapshot>, pid: sysinfo::Pid) {
        use sysinfo::ProcessesToUpdate;
        let mut system = sysinfo::System::new();
        system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true);
        if let Some(process) = system.process(pid) {
            let mut snapshot = snapshot.lock().expect("process snapshot");
            snapshot.cpu_seconds = process.accumulated_cpu_time() as f64 / 1000.0;
            snapshot.cpu_utilization = process.cpu_usage() as f64;
            snapshot.memory_usage = process.memory();
            snapshot.memory_virtual = process.virtual_memory();
        }
    }

    pub(super) fn init(
        settings: &Settings,
        session_store: SharedSessionStore,
        game_store: SharedGameStore,
    ) -> Result<Option<super::Telemetry>, String> {
        let endpoint = settings.otel_exporter_endpoint.clone();
        let headers = parse_headers(settings.otel_exporter_headers.as_deref());
        let resource = Resource::builder()
            .with_service_name(settings.otel_service_name.clone())
            .build();

        let span_exporter = SpanExporter::builder()
            .with_http()
            .with_endpoint(signal_endpoint(endpoint.as_deref(), "v1/traces").unwrap_or_default())
            .with_headers(headers.clone())
            .build()
            .map_err(|err| format!("could not build the OTLP span exporter: {err}"))?;
        let tracer_provider = SdkTracerProvider::builder()
            .with_batch_exporter(span_exporter)
            .with_resource(resource.clone())
            .build();

        let metric_exporter = MetricExporter::builder()
            .with_http()
            .with_endpoint(signal_endpoint(endpoint.as_deref(), "v1/metrics").unwrap_or_default())
            .with_headers(headers.clone())
            .build()
            .map_err(|err| format!("could not build the OTLP metric exporter: {err}"))?;
        let reader = PeriodicReader::builder(metric_exporter)
            .with_interval(Duration::from_millis(
                settings.otel_metric_export_interval_millis.max(1),
            ))
            .build();
        let meter_provider = SdkMeterProvider::builder()
            .with_reader(reader)
            .with_resource(resource.clone())
            .build();

        let log_exporter = LogExporter::builder()
            .with_http()
            .with_endpoint(signal_endpoint(endpoint.as_deref(), "v1/logs").unwrap_or_default())
            .with_headers(headers)
            .build()
            .map_err(|err| format!("could not build the OTLP log exporter: {err}"))?;
        let logger_provider = SdkLoggerProvider::builder()
            .with_batch_exporter(log_exporter)
            .with_resource(resource)
            .build();

        opentelemetry::global::set_meter_provider(meter_provider.clone());
        let meter = opentelemetry::metrics::MeterProvider::meter(&meter_provider, "tavolo");

        let active_users = Arc::new(AtomicI64::new(0));
        let ongoing_games = Arc::new(AtomicI64::new(0));
        let process = Arc::new(Mutex::new(ProcessSnapshot::default()));
        let pid = sysinfo::get_current_pid()
            .map_err(|err| format!("could not determine the current process id: {err}"))?;
        refresh_process(&process, pid);

        let _ = meter
            .u64_observable_gauge("tavolo.active_users")
            .with_description("Sessions currently held in the session store.")
            .with_unit("{user}")
            .with_callback({
                let active_users = active_users.clone();
                move |observer| {
                    observer.observe(active_users.load(Ordering::Relaxed).max(0) as u64, &[])
                }
            })
            .build();
        let _ = meter
            .u64_observable_gauge("tavolo.ongoing_games")
            .with_description("Live games that have not finished yet.")
            .with_unit("{game}")
            .with_callback({
                let ongoing_games = ongoing_games.clone();
                move |observer| {
                    observer.observe(ongoing_games.load(Ordering::Relaxed).max(0) as u64, &[])
                }
            })
            .build();
        let _ = meter
            .f64_observable_gauge("process.cpu.utilization")
            .with_description("Process CPU utilization since the last poll, in percent.")
            .with_callback({
                let process = process.clone();
                move |observer| {
                    let snapshot = process.lock().expect("process snapshot");
                    observer.observe(snapshot.cpu_utilization / 100.0, &[])
                }
            })
            .build();
        let _ = meter
            .u64_observable_gauge("process.memory.usage")
            .with_description("Process resident memory, in bytes.")
            .with_unit("By")
            .with_callback({
                let process = process.clone();
                move |observer| {
                    let snapshot = process.lock().expect("process snapshot");
                    observer.observe(snapshot.memory_usage, &[])
                }
            })
            .build();
        let _ = meter
            .u64_observable_gauge("process.memory.virtual")
            .with_description("Process virtual memory, in bytes.")
            .with_unit("By")
            .with_callback({
                let process = process.clone();
                move |observer| {
                    let snapshot = process.lock().expect("process snapshot");
                    observer.observe(snapshot.memory_virtual, &[])
                }
            })
            .build();
        let _ = meter
            .f64_observable_counter("process.cpu.time")
            .with_description("Process CPU time, in seconds.")
            .with_unit("s")
            .with_callback({
                let process = process.clone();
                move |observer| {
                    let snapshot = process.lock().expect("process snapshot");
                    observer.observe(snapshot.cpu_seconds, &[]);
                }
            })
            .build();

        let _ = HTTP_DURATION.set(
            meter
                .f64_histogram("http.server.request.duration")
                .with_description("HTTP server request duration.")
                .with_unit("s")
                .build(),
        );

        let poll_interval = Duration::from_secs(settings.otel_metrics_poll_interval_seconds.max(1));
        let poll_task = tokio::spawn({
            let session_store = session_store.clone();
            let game_store = game_store.clone();
            let active_users = active_users.clone();
            let ongoing_games = ongoing_games.clone();
            let process = process.clone();
            async move {
                loop {
                    match session_store.count().await {
                        Ok(count) => active_users.store(count as i64, Ordering::Relaxed),
                        Err(err) => tracing::warn!("active_users refresh failed: {err}"),
                    }
                    match game_store.count_ongoing().await {
                        Ok(count) => ongoing_games.store(count as i64, Ordering::Relaxed),
                        Err(err) => tracing::warn!("ongoing_games refresh failed: {err}"),
                    }
                    let process = process.clone();
                    let _ =
                        tokio::task::spawn_blocking(move || refresh_process(&process, pid)).await;
                    tokio::time::sleep(poll_interval).await;
                }
            }
        });

        Ok(Some(super::Telemetry {
            inner: OtelInner {
                tracer_provider,
                meter_provider,
                logger_provider,
                poll_task,
            },
        }))
    }
}

/// The live telemetry pipeline; dropping it shuts the providers down.
pub struct Telemetry {
    #[cfg(feature = "otel")]
    inner: inner::OtelInner,
}

#[cfg(feature = "otel")]
impl Telemetry {
    pub fn tracer(&self) -> opentelemetry_sdk::trace::SdkTracer {
        use opentelemetry::trace::TracerProvider as _;
        self.inner.tracer_provider.tracer("tavolo")
    }

    pub fn logger_provider(&self) -> &opentelemetry_sdk::logs::SdkLoggerProvider {
        &self.inner.logger_provider
    }
}

/// Initialize the telemetry pipeline when `OTEL_ENABLED` is set.
pub fn init(
    settings: &Settings,
    session_store: SharedSessionStore,
    game_store: SharedGameStore,
) -> Result<Option<Telemetry>, String> {
    if !settings.otel_enabled {
        return Ok(None);
    }
    #[cfg(feature = "otel")]
    {
        inner::init(settings, session_store, game_store)
    }
    #[cfg(not(feature = "otel"))]
    {
        let _ = (session_store, game_store);
        Err("OTEL_ENABLED is set but the binary was built without the 'otel' feature".to_string())
    }
}

/// Tag a span with the hashed session id and the authenticated user.
pub fn session_attributes(
    session: &tavolo_platform::session::SessionData,
) -> (Option<String>, Option<String>) {
    let session_id = session.id().map(hash_session_id);
    let user = tavolo_platform::auth::session_user(session).map(|user| user.sub);
    (session_id, user)
}

/// Truncated SHA-256 of the session id (the cookie is a credential, so it is
/// never exported verbatim).
pub fn hash_session_id(session_id: &str) -> String {
    tavolo_platform::auth::session_tag(session_id)
}

/// An HTTP span per request, tagged with session/user; excludes the paths in
/// `OTEL_EXCLUDED_PATHS`. Also records the HTTP server duration histogram.
pub async fn trace_middleware(
    excluded: Arc<Vec<String>>,
    request: axum::extract::Request,
    next: axum::middleware::Next,
) -> axum::response::Response {
    use tracing::Instrument;

    let path = request.uri().path().to_string();
    if excluded.contains(&path) {
        return next.run(request).await;
    }
    let method = request.method().clone();
    let span = tracing::info_span!(
        "http.server.request",
        "http.request.method" = %method,
        "url.path" = %path,
        "http.response.status_code" = tracing::field::Empty,
        "session.id" = tracing::field::Empty,
        "user.id" = tracing::field::Empty,
    );
    let session = request
        .extensions()
        .get::<tavolo_platform::session::SessionHandle>()
        .cloned();
    async move {
        let started = std::time::Instant::now();
        let response = next.run(request).await;
        let status = response.status().as_u16();
        tracing::Span::current().record("http.response.status_code", status);
        // Recorded after the handler ran so logins/mutations are visible;
        // the session middleware mints the cookie (and its id on first use)
        // after this layer returns.
        if let Some(session) = session {
            let guard = session.lock().await;
            let (session_id, user) = session_attributes(&guard);
            if let Some(session_id) = session_id {
                tracing::Span::current().record("session.id", session_id);
            }
            if let Some(user) = user {
                tracing::Span::current().record("user.id", user);
            }
        }
        record_http_duration(started.elapsed(), &method, status);
        response
    }
    .instrument(span)
    .await
}

fn record_http_duration(duration: std::time::Duration, method: &axum::http::Method, status: u16) {
    #[cfg(feature = "otel")]
    {
        if let Some(histogram) = inner::HTTP_DURATION.get() {
            histogram.record(
                duration.as_secs_f64(),
                &[
                    opentelemetry::KeyValue::new("http.request.method", method.to_string()),
                    opentelemetry::KeyValue::new("http.response.status_code", status as i64),
                ],
            );
        }
    }
    #[cfg(not(feature = "otel"))]
    {
        let _ = (duration, method, status);
    }
}