rws_server/otel/
metrics.rs

1//! OTLP metrics export: `http.server.request.duration` pushed to an
2//! OpenTelemetry collector over OTLP/HTTP (JSON), every `interval`, with
3//! cumulative temporality and the semantic convention's explicit buckets.
4//!
5//! The data is the same store `/metrics` serves as
6//! `http_server_request_duration_seconds` ([`crate::metrics::http`]), so the
7//! two never disagree. Start it with [`start`] or [`start_from_env`] (the
8//! `rws` binary does when `OTEL_EXPORTER_OTLP_ENDPOINT` or
9//! `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` is set); [`super::shutdown`] sends a
10//! last export.
11
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::sync::{Condvar, Mutex, OnceLock};
14use std::time::{Duration, SystemTime, UNIX_EPOCH};
15
16use crate::metrics::http;
17
18/// Where and how often to push metrics.
19#[derive(Clone, Debug)]
20pub struct MetricsExportConfig {
21    /// The full URL metrics are POSTed to, e.g. `http://collector:4318/v1/metrics`.
22    pub endpoint: String,
23    /// Time between exports.
24    pub interval: Duration,
25    /// `service.name` resource attribute.
26    pub service_name: String,
27    /// `service.version` resource attribute.
28    pub service_version: String,
29    /// Extra request headers (`OTEL_EXPORTER_OTLP_HEADERS`), e.g. an API key.
30    pub headers: Vec<(String, String)>,
31}
32
33struct Exporter {
34    config: MetricsExportConfig,
35    start_ns: u64,
36    stop: AtomicBool,
37    wake: (Mutex<()>, Condvar),
38}
39
40static EXPORTER: OnceLock<Exporter> = OnceLock::new();
41
42fn now_ns() -> u64 {
43    SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_nanos() as u64).unwrap_or(0)
44}
45
46/// Start pushing metrics on a background thread. A second call is a no-op.
47pub fn start(config: MetricsExportConfig) {
48    let interval = config.interval.max(Duration::from_millis(100));
49    let exporter = Exporter { config, start_ns: now_ns(), stop: AtomicBool::new(false), wake: (Mutex::new(()), Condvar::new()) };
50    if EXPORTER.set(exporter).is_err() {
51        return;
52    }
53    let _ = std::thread::Builder::new().name("rws-otlp-metrics".to_string()).spawn(move || {
54        let exporter = EXPORTER.get().expect("set above");
55        loop {
56            let guard = exporter.wake.0.lock().unwrap_or_else(|e| e.into_inner());
57            let _ = exporter.wake.1.wait_timeout(guard, interval);
58            if exporter.stop.load(Ordering::Relaxed) {
59                return;
60            }
61            export(exporter);
62        }
63    });
64}
65
66/// [`start`] configured by the standard OpenTelemetry variables, if an
67/// endpoint is set and `OTEL_METRICS_EXPORTER` isn't `none`:
68///
69/// - `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` — full URL, used as is; else
70///   `OTEL_EXPORTER_OTLP_ENDPOINT` + `/v1/metrics`
71/// - `OTEL_METRIC_EXPORT_INTERVAL` — milliseconds (default 60000)
72/// - `OTEL_SERVICE_NAME` (default `rws`), `OTEL_EXPORTER_OTLP_HEADERS`
73///   (`key=value,…`; `OTEL_EXPORTER_OTLP_METRICS_HEADERS` wins)
74///
75/// Returns whether it started.
76pub fn start_from_env() -> bool {
77    let var = |name: &str| std::env::var(name).ok().map(|v| v.trim().to_string()).filter(|v| !v.is_empty());
78    if var("OTEL_METRICS_EXPORTER").is_some_and(|v| v.eq_ignore_ascii_case("none")) {
79        return false;
80    }
81    let Some(endpoint) = var("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT")
82        .or_else(|| var("OTEL_EXPORTER_OTLP_ENDPOINT").map(|base| format!("{}/v1/metrics", base.trim_end_matches('/'))))
83    else {
84        return false;
85    };
86    let interval = var("OTEL_METRIC_EXPORT_INTERVAL").and_then(|v| v.parse().ok()).unwrap_or(60_000);
87    let headers = var("OTEL_EXPORTER_OTLP_METRICS_HEADERS").or_else(|| var("OTEL_EXPORTER_OTLP_HEADERS")).map(|h| super::parse_headers(&h)).unwrap_or_default();
88    start(MetricsExportConfig {
89        endpoint,
90        interval: Duration::from_millis(interval),
91        service_name: var("OTEL_SERVICE_NAME").unwrap_or_else(|| "rws".to_string()),
92        service_version: env!("CARGO_PKG_VERSION").to_string(),
93        headers,
94    });
95    true
96}
97
98/// Stop the background thread after one last export (called by [`super::shutdown`]).
99#[doc(hidden)]
100pub fn shutdown() {
101    if let Some(exporter) = EXPORTER.get() {
102        if !exporter.stop.swap(true, Ordering::Relaxed) {
103            exporter.wake.1.notify_all();
104            export(exporter);
105        }
106    }
107}
108
109fn export(exporter: &Exporter) {
110    let body = body(&exporter.config.service_name, &exporter.config.service_version, exporter.start_ns, now_ns());
111    let client = crate::http_client::Client::new().timeout_ms(10_000);
112    let mut request = client.post(&exporter.config.endpoint).body_json(&body);
113    for (name, value) in &exporter.config.headers {
114        request = request.header(name, value);
115    }
116    match request.send() {
117        Ok(response) if response.is_success() => {}
118        Ok(response) => eprintln!("OTLP metrics export to {}: HTTP {}", exporter.config.endpoint, response.status()),
119        Err(e) => eprintln!("OTLP metrics export to {}: {}", exporter.config.endpoint, e),
120    }
121}
122
123/// The OTLP/HTTP JSON `ExportMetricsServiceRequest` for the current data.
124pub fn body(service_name: &str, service_version: &str, start_ns: u64, now_ns: u64) -> String {
125    let points: Vec<String> = http::snapshot()
126        .into_iter()
127        .map(|(series, histogram)| {
128            let attributes = [
129                string_attr("http.request.method", series.method()),
130                int_attr("http.response.status_code", series.status as i64),
131                string_attr("network.protocol.name", "http"),
132                string_attr("network.protocol.version", series.protocol_version()),
133                string_attr("url.scheme", series.scheme()),
134            ]
135            .join(",");
136            let counts: Vec<String> = histogram.counts.iter().map(|c| format!("\"{c}\"")).collect();
137            let bounds: Vec<String> = http::BOUNDS.iter().map(|b| b.to_string()).collect();
138            format!(
139                "{{\"attributes\":[{attributes}],\"startTimeUnixNano\":\"{start_ns}\",\"timeUnixNano\":\"{now_ns}\",\
140                 \"count\":\"{}\",\"sum\":{},\"bucketCounts\":[{}],\"explicitBounds\":[{}]}}",
141                histogram.count,
142                histogram.sum,
143                counts.join(","),
144                bounds.join(","),
145            )
146        })
147        .collect();
148    let metrics = if points.is_empty() {
149        String::new()
150    } else {
151        format!(
152            "{{\"name\":\"http.server.request.duration\",\"description\":\"Duration of HTTP server requests.\",\"unit\":\"s\",\
153             \"histogram\":{{\"aggregationTemporality\":2,\"dataPoints\":[{}]}}}}",
154            points.join(",")
155        )
156    };
157    format!(
158        "{{\"resourceMetrics\":[{{\"resource\":{{\"attributes\":[{},{}]}},\
159         \"scopeMetrics\":[{{\"scope\":{{\"name\":\"rust-web-server\",\"version\":\"{}\"}},\"metrics\":[{metrics}]}}]}}]}}",
160        string_attr("service.name", service_name),
161        string_attr("service.version", service_version),
162        env!("CARGO_PKG_VERSION"),
163    )
164}
165
166fn string_attr(key: &str, value: &str) -> String {
167    format!("{{\"key\":\"{key}\",\"value\":{{\"stringValue\":\"{}\"}}}}", super::json_escape(value))
168}
169
170fn int_attr(key: &str, value: i64) -> String {
171    format!("{{\"key\":\"{key}\",\"value\":{{\"intValue\":\"{value}\"}}}}")
172}
173
174#[cfg(test)]
175mod tests;
rws_server/otel/metrics.rs

rws_server/otel/
metrics.rs
