Replace the Python/kaya server in server/ with a Rust Cargo workspace of three crates mirroring the old layout: tavolo-platform (lobby API, live-game store, websocket endpoint, deadline scheduler, stats/Elo, OIDC sessions), tavolo-scopone (pure scopone scientifico rules plus the platform adapter) and tavolo-app (composition root and the `tavolo` binary with serve/migrate/backfill-elo). The HTTP and WebSocket contracts, error messages, Redis key layout and Postgres schema are unchanged, so the existing web/ frontend and in-flight games keep working. Notable differences: sessions are JSON under tavolo:session: (Python pickle sessions invalidate once), LOGGING_CONFIG is ignored in favour of RUST_LOG, DATABASE_URL must be Postgres, sqlx migrations replace aerich, and OpenTelemetry exports Rust process metrics instead of CPython gc metrics (no Postgres/Redis client instrumentation). Docker Compose and the Kubernetes manifest now run the single Rust image (embedded migrations, /static and /assets served by the app); the release CI keeps building server/Dockerfile. The full test suite (scopone rules, platform store/deadlines/stats/websockets, OIDC login flow, scopone e2e) is ported and runs with cargo test --workspace; Redis/Postgres integration tests are gated on TEST_REDIS_URL/TEST_DATABASE_URL.
This commit is contained in:
1 parent
adc3d340b1
commit
7b3b38ccbc
136 files changed
+16821
-11777
No files matched your search
@@ -0,0 +1,389 @@
|
||||
//! 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);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user