xtrace 0.0.15

An AI observability service for collecting, storing, and querying traces, spans, and metrics across LLM and agent workflows.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133

use axum::{
    middleware::{self},
    routing::{get, post},
    Router,
};
use sqlx::postgres::PgPoolOptions;
use std::{net::SocketAddr, sync::Arc};
use tokio::sync::mpsc;
use tower_http::trace::TraceLayer;

use crate::http::common::healthz;
use crate::http::{
    auth::{auth, rate_limit},
    metrics::{self, metrics_worker, post_metrics_batch, MetricsBatchRequest},
    ops::get_rate_limit_stats,
    projects::get_projects,
    traces,
};
use crate::ingest::batch::{ingest_worker, post_batch, BatchIngestRequest};
use crate::ingest::otlp;
use crate::state::{AppState, RateLimitStats, ServerConfig};

/// Start xtrace server (blocks until shutdown signal)
pub async fn run_server(config: ServerConfig) -> anyhow::Result<()> {
    let pool = PgPoolOptions::new()
        .max_connections(20)
        .connect(&config.database_url)
        .await?;

    sqlx::migrate!("./migrations").run(&pool).await?;

    let (ingest_tx, ingest_rx) = mpsc::channel::<BatchIngestRequest>(1000);
    let (metrics_tx, metrics_rx) = mpsc::channel::<MetricsBatchRequest>(5000);

    let qps = config.rate_limit_qps;
    let burst = config.rate_limit_burst;
    let query_limiter = AppState::build_limiter(qps, burst);
    let rate_limit_stats = Arc::new(RateLimitStats::new());

    let state = AppState {
        pool,
        api_bearer_token: Arc::from(config.api_bearer_token),
        langfuse_public_key: config.langfuse_public_key.map(Arc::from),
        langfuse_secret_key: config.langfuse_secret_key.map(Arc::from),
        default_project_id: Arc::from(config.default_project_id),
        ingest_tx,
        metrics_tx,
        query_limiter,
        rate_limit_stats,
        rate_limit_qps: qps,
        rate_limit_burst: burst,
    };

    tokio::spawn(ingest_worker(
        state.pool.clone(),
        state.default_project_id.clone(),
        ingest_rx,
    ));

    tokio::spawn(metrics_worker(
        state.pool.clone(),
        state.default_project_id.clone(),
        metrics_rx,
    ));

    // Query routes — apply both auth and per-token rate limiting.
    let query_routes = Router::new()
        .route("/api/public/metrics/daily", get(metrics::get_metrics_daily))
        .route("/api/public/metrics/query", get(metrics::get_metrics_query))
        .route("/api/public/metrics/names", get(metrics::get_metrics_names))
        .route("/api/public/traces", get(traces::get_traces))
        .route("/api/public/traces/:traceId", get(traces::get_trace))
        .route_layer(middleware::from_fn_with_state(state.clone(), rate_limit));

    // Write / compat routes — auth only, no rate limit (channel backpressure applies).
    let write_routes = Router::new()
        .route("/v1/l/batch", post(post_batch))
        .route("/v1/metrics/batch", post(post_metrics_batch))
        .route("/api/public/projects", get(get_projects))
        .route("/api/public/otel/v1/traces", post(otlp::post_otel_traces));

    let protected_routes = Router::new()
        .merge(query_routes)
        .merge(write_routes)
        .route_layer(middleware::from_fn_with_state(state.clone(), auth));

    let addr: SocketAddr = config.bind_addr.parse()?;
    tracing::info!(
        "listening on {} (rate_limit: {} qps, burst {})",
        addr,
        qps,
        burst
    );

    let app = Router::new()
        .route("/healthz", get(healthz))
        .route("/api/internal/rate_limit_stats", get(get_rate_limit_stats))
        .merge(protected_routes)
        .with_state(state)
        .layer(TraceLayer::new_for_http());

    let listener = tokio::net::TcpListener::bind(addr).await?;
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())
        .await?;

    Ok(())
}

async fn shutdown_signal() {
    let ctrl_c = async {
        tokio::signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            .expect("failed to install signal handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
}