commonware_runtime/tokio/
telemetry.rs

1//! Utilities for collecting and reporting telemetry data.
2
3use super::{
4    tracing::{export, Config},
5    Context,
6};
7use crate::{Metrics as _, Spawner, Supervisor as _};
8use axum::{
9    body::Body,
10    http::{header, Response, StatusCode},
11    routing::get,
12    serve, Extension, Router,
13};
14use cfg_if::cfg_if;
15use std::{net::SocketAddr, sync::Arc};
16use tokio::net::TcpListener;
17use tracing::Level;
18use tracing_subscriber::{filter::filter_fn, layer::SubscriberExt, Layer, Registry};
19
20/// Log configuration.
21pub struct Logs {
22    /// The level of logging to use.
23    pub level: Level,
24
25    /// Whether to log in JSON format.
26    ///
27    /// This is useful for structured logging in server-based environments.
28    /// If you are running things locally, it is recommended to use
29    /// `json = false` to get a human-readable format.
30    pub json: bool,
31}
32
33/// Initialize telemetry with the given configuration.
34///
35/// If `metrics` is provided, starts serving metrics at the given address at `/metrics`.
36/// If `traces` is provided, enables OpenTelemetry trace export. When `None`, span
37/// callsites in the runtime-provided subscriber are disabled and only log events
38/// are emitted (per [`Logs::level`]).
39pub fn init(context: Context, logs: Logs, metrics: Option<SocketAddr>, traces: Option<Config>) {
40    // Create fmt layer for logging
41    let log_layer = tracing_subscriber::fmt::layer()
42        .with_line_number(true)
43        .with_thread_ids(true)
44        .with_file(true);
45
46    // Set the format to JSON (if specified)
47    let log_layer = if logs.json {
48        log_layer.json().boxed()
49    } else {
50        log_layer.compact().boxed()
51    };
52    let log_layer = match &traces {
53        None => log_layer
54            .with_filter(filter_fn(move |metadata| {
55                metadata.is_event() && *metadata.level() <= logs.level
56            }))
57            .boxed(),
58        Some(_) => log_layer
59            .with_filter(tracing_subscriber::EnvFilter::new(logs.level.to_string()))
60            .boxed(),
61    };
62
63    // Create OpenTelemetry layer for tracing
64    let trace_layer = traces.map(|cfg| {
65        let tracer = export(cfg).expect("Failed to initialize tracer");
66        tracing_opentelemetry::layer()
67            .with_tracer(tracer)
68            .with_filter(tracing_subscriber::EnvFilter::new(logs.level.to_string()))
69    });
70
71    // Create tracing registry.
72    cfg_if! {
73        if #[cfg(feature = "tokio-console")] {
74            let console_layer = console_subscriber::spawn();
75            let registry = Registry::default()
76                .with(log_layer)
77                .with(trace_layer)
78                .with(console_layer);
79        } else {
80            let registry = Registry::default().with(log_layer).with(trace_layer);
81        }
82    }
83
84    // Set the global subscriber
85    tracing::subscriber::set_global_default(registry).expect("Failed to set subscriber");
86
87    // Expose metrics over HTTP
88    if let Some(cfg) = metrics {
89        context.child("metrics").spawn(move |context| async move {
90            // Create a tokio listener for the metrics server.
91            //
92            // We explicitly avoid using a runtime `Listener` because
93            // it will track bandwidth used for metrics and apply a policy
94            // for read/write timeouts fit for a p2p network.
95            let listener = TcpListener::bind(cfg)
96                .await
97                .expect("Failed to bind metrics server");
98
99            // Create a router for the metrics server
100            let shared = Arc::new(context);
101            let app = Router::new()
102                .route(
103                    "/metrics",
104                    get(|Extension(ctx): Extension<Arc<Context>>| async move {
105                        Response::builder()
106                            .status(StatusCode::OK)
107                            .header(header::CONTENT_TYPE, "text/plain; version=0.0.4")
108                            .body(Body::from(ctx.encode()))
109                            .expect("Failed to create response")
110                    }),
111                )
112                .layer(Extension(shared));
113
114            // Serve the metrics over HTTP.
115            //
116            // `serve` will spawn its own tasks using `tokio::spawn` (and there is no way to specify
117            // it to do otherwise). These tasks will not be tracked like metrics spawned using `Spawner`.
118            serve(listener, app.into_make_service())
119                .await
120                .expect("Could not serve metrics");
121        });
122    }
123}
commonware_runtime/tokio/telemetry.rs

commonware_runtime/tokio/
telemetry.rs
