rws_server/otel/
logs.rs

1//! OTLP logs export (18.33.0, #102): every access-log record pushed to an
2//! OpenTelemetry collector over OTLP/HTTP (JSON, `/v1/logs`), with the
3//! request's trace and span ids when it had a server span.
4//!
5//! Off until [`start`] (the `rws` binary calls [`start_from_env`]: on with
6//! `OTEL_LOGS_EXPORTER=otlp`). Then [`crate::log::Log::log_access`] formats
7//! each record into a buffer of its thread's own — no shared lock, no I/O on
8//! the request path — and the thread `rws-otlp-logs` collects every buffer
9//! and POSTs one batch every `OTEL_BLRP_SCHEDULE_DELAY` (1 s by default).
10//! A thread holding more than [`MAX_PENDING`] bytes drops records (counted
11//! by [`dropped`]) rather than grow without bound while a collector is down.
12//! [`super::shutdown`] sends what is left.
13//!
14//! A record's body is the access-log line (`RWS_CONFIG_LOG_FORMAT`; JSON
15//! when that is `off`), with the access log's privacy settings applied; its
16//! attributes are `http.request.method`, `url.path`,
17//! `http.response.status_code`, `client.address` and
18//! `network.protocol.version`. Severity is INFO, WARN for 4xx and ERROR for
19//! 5xx.
20
21use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
22use std::sync::{Arc, Condvar, Mutex, OnceLock};
23use std::time::Duration;
24
25use crate::request::Request;
26use crate::response::Response;
27use crate::server_config::LogPrivacy;
28
29/// A thread's unsent records past this many bytes are dropped.
30pub const MAX_PENDING: usize = 1 << 20;
31/// A POST carries at most about this many bytes of records.
32pub const MAX_BATCH: usize = 4 << 20;
33
34/// The response header `OtelLayer` hands the request's trace context to the
35/// access log in, while export is on; the server's connection loops remove
36/// it before the response is written ([`take_trace`]).
37#[doc(hidden)]
38pub const TRACE_HEADER: &str = "x-rws-log-trace";
39
40/// The trace a logged request belonged to.
41#[derive(Clone, Copy, Debug, PartialEq, Eq)]
42pub struct TraceIds {
43    pub trace_id: [u8; 16],
44    pub span_id: [u8; 8],
45    pub sampled: bool,
46}
47
48static ENABLED: AtomicBool = AtomicBool::new(false);
49static DROPPED: AtomicU64 = AtomicU64::new(0);
50
51/// Whether access-log records are exported (after [`start`]).
52pub fn enabled() -> bool {
53    #[cfg(test)]
54    if TEST_ENABLED.with(|e| e.get()) {
55        return true;
56    }
57    ENABLED.load(Ordering::Relaxed)
58}
59
60/// Records dropped because a thread's buffer was full.
61pub fn dropped() -> u64 {
62    DROPPED.load(Ordering::Relaxed)
63}
64
65fn hex(bytes: &[u8], out: &mut String) {
66    const HEX: &[u8; 16] = b"0123456789abcdef";
67    for b in bytes {
68        out.push(HEX[(b >> 4) as usize] as char);
69        out.push(HEX[(b & 15) as usize] as char);
70    }
71}
72
73fn unhex<const N: usize>(text: &str) -> Option<[u8; N]> {
74    if text.len() != N * 2 {
75        return None;
76    }
77    let mut out = [0u8; N];
78    for (i, byte) in out.iter_mut().enumerate() {
79        *byte = u8::from_str_radix(text.get(i * 2..i * 2 + 2)?, 16).ok()?;
80    }
81    Some(out)
82}
83
84/// Put `ids` on `response` for the access log (only while export is on).
85#[doc(hidden)]
86pub fn put_trace(response: &mut Response, ids: TraceIds) {
87    if !enabled() {
88        return;
89    }
90    let mut value = String::with_capacity(52);
91    hex(&ids.trace_id, &mut value);
92    value.push('-');
93    hex(&ids.span_id, &mut value);
94    value.push_str(if ids.sampled { "-01" } else { "-00" });
95    response.headers.push(crate::header::Header::new(TRACE_HEADER, value));
96}
97
98/// Remove the trace context [`put_trace`] left on `response` and return
99/// it. One atomic load while export is off.
100pub fn take_trace(response: &mut Response) -> Option<TraceIds> {
101    if !enabled() {
102        return None;
103    }
104    let at = response.headers.iter().position(|h| h.name.eq_ignore_ascii_case(TRACE_HEADER))?;
105    let header = response.headers.remove(at);
106    parse_trace(&header.value)
107}
108
109fn parse_trace(value: &str) -> Option<TraceIds> {
110    let mut parts = value.split('-');
111    let trace_id = unhex::<16>(parts.next()?)?;
112    let span_id = unhex::<8>(parts.next()?)?;
113    let sampled = parts.next()? == "01";
114    Some(TraceIds { trace_id, span_id, sampled })
115}
116
117// ── per-thread buffers ───────────────────────────────────────────────────────
118
119type Shard = Arc<Mutex<Vec<u8>>>;
120
121fn registry() -> &'static Mutex<Vec<Shard>> {
122    static REGISTRY: OnceLock<Mutex<Vec<Shard>>> = OnceLock::new();
123    REGISTRY.get_or_init(|| Mutex::new(Vec::new()))
124}
125
126/// Records of threads that have exited, not sent yet.
127fn orphans() -> &'static Mutex<Vec<u8>> {
128    static ORPHANS: Mutex<Vec<u8>> = Mutex::new(Vec::new());
129    &ORPHANS
130}
131
132struct ShardHandle(Shard);
133
134impl Drop for ShardHandle {
135    fn drop(&mut self) {
136        let pending = std::mem::take(&mut *self.0.lock().unwrap_or_else(|e| e.into_inner()));
137        append(&mut orphans().lock().unwrap_or_else(|e| e.into_inner()), &pending);
138        registry().lock().unwrap_or_else(|e| e.into_inner()).retain(|s| !Arc::ptr_eq(s, &self.0));
139    }
140}
141
142thread_local! {
143    static SHARD: ShardHandle = {
144        let shard: Shard = Arc::new(Mutex::new(Vec::with_capacity(4096)));
145        registry().lock().unwrap_or_else(|e| e.into_inner()).push(Arc::clone(&shard));
146        ShardHandle(shard)
147    };
148    static SCRATCH: std::cell::RefCell<Vec<u8>> = const { std::cell::RefCell::new(Vec::new()) };
149}
150
151/// Append comma-separated `records` to `buf`, unless that takes it past
152/// [`MAX_PENDING`].
153fn append(buf: &mut Vec<u8>, records: &[u8]) {
154    if records.is_empty() {
155        return;
156    }
157    if buf.len() + records.len() > MAX_PENDING {
158        DROPPED.fetch_add(1 + memchr::memmem::find_iter(records, b"},{\"timeUnixNano\"").count() as u64, Ordering::Relaxed);
159        return;
160    }
161    if !buf.is_empty() {
162        buf.push(b',');
163    }
164    buf.extend_from_slice(records);
165}
166
167/// Record one request: `write_line` writes the access-log line that
168/// becomes the body. Does nothing until [`start`].
169pub fn record(
170    request: &Request,
171    response: &Response,
172    client: std::net::IpAddr,
173    privacy: &LogPrivacy,
174    trace: Option<TraceIds>,
175    write_line: impl FnOnce(&mut Vec<u8>),
176) {
177    if !enabled() {
178        return;
179    }
180    let mut write_line = Some(write_line);
181    let mut build = |out: &mut Vec<u8>, line: &mut Vec<u8>| {
182        line.clear();
183        if let Some(write_line) = write_line.take() {
184            write_line(line);
185        }
186        write_record(out, request, response, client, privacy, trace, line, super::now_ns());
187    };
188    let done = SHARD.try_with(|handle| {
189        SCRATCH.with(|scratch| {
190            let mut record = Vec::with_capacity(512);
191            build(&mut record, &mut scratch.borrow_mut());
192            let mut buf = handle.0.lock().unwrap_or_else(|e| e.into_inner());
193            append(&mut buf, &record);
194        })
195    });
196    if done.is_err() {
197        // The thread is exiting: straight to the orphans.
198        let (mut record, mut line) = (Vec::new(), Vec::new());
199        build(&mut record, &mut line);
200        append(&mut orphans().lock().unwrap_or_else(|e| e.into_inner()), &record);
201    }
202}
203
204/// One OTLP JSON `LogRecord`.
205#[allow(clippy::too_many_arguments)]
206fn write_record(
207    out: &mut Vec<u8>,
208    request: &Request,
209    response: &Response,
210    client: std::net::IpAddr,
211    privacy: &LogPrivacy,
212    trace: Option<TraceIds>,
213    line: &[u8],
214    now_ns: u64,
215) {
216    use std::io::Write;
217    let (number, text) = match response.status_code {
218        500.. => (17, "ERROR"),
219        400.. => (13, "WARN"),
220        _ => (9, "INFO"),
221    };
222    let _ = write!(
223        out,
224        "{{\"timeUnixNano\":\"{now_ns}\",\"observedTimeUnixNano\":\"{now_ns}\",\"severityNumber\":{number},\"severityText\":\"{text}\",\"body\":{{\"stringValue\":\""
225    );
226    escape_into(out, line);
227    out.extend_from_slice(b"\"},\"attributes\":[");
228    string_attr(out, "http.request.method", &request.method);
229    out.push(b',');
230    let path = request.request_uri.split(['?', '#']).next().unwrap_or("");
231    string_attr(out, "url.path", path);
232    let _ = write!(out, ",{{\"key\":\"http.response.status_code\",\"value\":{{\"intValue\":\"{}\"}}}},", response.status_code);
233    string_attr(out, "client.address", &privacy.ip_string(client));
234    out.push(b',');
235    let version = request.http_version.strip_prefix("HTTP/").unwrap_or(&request.http_version);
236    string_attr(out, "network.protocol.version", version);
237    out.push(b']');
238    if let Some(ids) = trace {
239        let mut trace_hex = String::with_capacity(32);
240        hex(&ids.trace_id, &mut trace_hex);
241        let mut span_hex = String::with_capacity(16);
242        hex(&ids.span_id, &mut span_hex);
243        let _ = write!(out, ",\"traceId\":\"{trace_hex}\",\"spanId\":\"{span_hex}\",\"flags\":{}", ids.sampled as u8);
244    }
245    out.push(b'}');
246}
247
248fn string_attr(out: &mut Vec<u8>, key: &str, value: &str) {
249    out.extend_from_slice(b"{\"key\":\"");
250    out.extend_from_slice(key.as_bytes());
251    out.extend_from_slice(b"\",\"value\":{\"stringValue\":\"");
252    escape_into(out, value.as_bytes());
253    out.extend_from_slice(b"\"}}");
254}
255
256/// `bytes` as the inside of a JSON string: every control character
257/// escaped, invalid UTF-8 replaced.
258fn escape_into(out: &mut Vec<u8>, bytes: &[u8]) {
259    use std::io::Write;
260    for c in String::from_utf8_lossy(bytes).chars() {
261        match c {
262            '"' => out.extend_from_slice(b"\\\""),
263            '\\' => out.extend_from_slice(b"\\\\"),
264            '\n' => out.extend_from_slice(b"\\n"),
265            '\r' => out.extend_from_slice(b"\\r"),
266            '\t' => out.extend_from_slice(b"\\t"),
267            c if (c as u32) < 0x20 => {
268                let _ = write!(out, "\\u{:04x}", c as u32);
269            }
270            c => {
271                let mut utf8 = [0u8; 4];
272                out.extend_from_slice(c.encode_utf8(&mut utf8).as_bytes());
273            }
274        }
275    }
276}
277
278/// Every buffered record, taken: batches of at most about [`MAX_BATCH`]
279/// bytes, each a comma-separated list of `LogRecord`s.
280#[doc(hidden)]
281pub fn drain() -> Vec<Vec<u8>> {
282    let shards: Vec<Shard> = registry().lock().unwrap_or_else(|e| e.into_inner()).clone();
283    let mut pieces = vec![std::mem::take(&mut *orphans().lock().unwrap_or_else(|e| e.into_inner()))];
284    for shard in shards {
285        pieces.push(std::mem::take(&mut *shard.lock().unwrap_or_else(|e| e.into_inner())));
286    }
287    let mut batches: Vec<Vec<u8>> = Vec::new();
288    for piece in pieces.into_iter().filter(|p| !p.is_empty()) {
289        match batches.last_mut() {
290            Some(batch) if batch.len() + piece.len() < MAX_BATCH => {
291                batch.push(b',');
292                batch.extend_from_slice(&piece);
293            }
294            _ => batches.push(piece),
295        }
296    }
297    batches
298}
299
300/// The OTLP/HTTP JSON `ExportLogsServiceRequest` for `records` (a
301/// comma-separated list of `LogRecord`s).
302pub fn body(records: &[u8], service_name: &str, service_version: &str) -> Vec<u8> {
303    let mut out = Vec::with_capacity(records.len() + 256);
304    out.extend_from_slice(b"{\"resourceLogs\":[{\"resource\":{\"attributes\":[");
305    string_attr(&mut out, "service.name", service_name);
306    out.push(b',');
307    string_attr(&mut out, "service.version", service_version);
308    out.extend_from_slice(b"]},\"scopeLogs\":[{\"scope\":{\"name\":\"rust-web-server\",\"version\":\"");
309    out.extend_from_slice(env!("CARGO_PKG_VERSION").as_bytes());
310    out.extend_from_slice(b"\"},\"logRecords\":[");
311    out.extend_from_slice(records);
312    out.extend_from_slice(b"]}]}]}");
313    out
314}
315
316// ── the exporter ─────────────────────────────────────────────────────────────
317
318/// Where and how often to push logs.
319#[derive(Clone, Debug)]
320pub struct LogsExportConfig {
321    /// The full URL logs are POSTed to, e.g. `http://collector:4318/v1/logs`.
322    pub endpoint: String,
323    /// Time between exports.
324    pub interval: Duration,
325    /// `service.name` resource attribute.
326    pub service_name: String,
327    /// `service.version` resource attribute.
328    pub service_version: String,
329    /// Extra request headers (`OTEL_EXPORTER_OTLP_HEADERS`), e.g. an API key.
330    pub headers: Vec<(String, String)>,
331}
332
333struct Exporter {
334    config: LogsExportConfig,
335    stop: AtomicBool,
336    wake: (Mutex<()>, Condvar),
337    /// Held while a batch is sent, so [`shutdown`] waits for one in flight.
338    sending: Mutex<()>,
339}
340
341static EXPORTER: OnceLock<Exporter> = OnceLock::new();
342
343/// Start exporting on a background thread. A second call is a no-op.
344pub fn start(config: LogsExportConfig) {
345    let interval = config.interval.max(Duration::from_millis(100));
346    let exporter = Exporter { config, stop: AtomicBool::new(false), wake: (Mutex::new(()), Condvar::new()), sending: Mutex::new(()) };
347    if EXPORTER.set(exporter).is_err() {
348        return;
349    }
350    ENABLED.store(true, Ordering::SeqCst);
351    let _ = std::thread::Builder::new().name("rws-otlp-logs".to_string()).spawn(move || {
352        let exporter = EXPORTER.get().expect("set above");
353        loop {
354            let guard = exporter.wake.0.lock().unwrap_or_else(|e| e.into_inner());
355            let _ = exporter.wake.1.wait_timeout(guard, interval);
356            if exporter.stop.load(Ordering::Relaxed) {
357                return;
358            }
359            export(exporter);
360        }
361    });
362}
363
364/// [`start`] configured by the standard OpenTelemetry variables, when
365/// `OTEL_LOGS_EXPORTER` names `otlp`:
366///
367/// - `OTEL_EXPORTER_OTLP_LOGS_ENDPOINT` — full URL, used as is; else
368///   `OTEL_EXPORTER_OTLP_ENDPOINT` + `/v1/logs`; else
369///   `http://localhost:4318/v1/logs`
370/// - `OTEL_BLRP_SCHEDULE_DELAY` — milliseconds between exports (default 1000)
371/// - `OTEL_SERVICE_NAME` (default `rws`), `OTEL_EXPORTER_OTLP_HEADERS`
372///   (`key=value,…`; `OTEL_EXPORTER_OTLP_LOGS_HEADERS` wins)
373///
374/// Returns whether it started.
375pub fn start_from_env() -> bool {
376    let var = |name: &str| std::env::var(name).ok().map(|v| v.trim().to_string()).filter(|v| !v.is_empty());
377    let Some(config) = config_from(var) else {
378        return false;
379    };
380    start(config);
381    true
382}
383
384/// The configuration [`start_from_env`] would use, from `var`.
385#[doc(hidden)]
386pub fn config_from(var: impl Fn(&str) -> Option<String>) -> Option<LogsExportConfig> {
387    if !var("OTEL_LOGS_EXPORTER").is_some_and(|v| v.split(',').any(|e| e.trim().eq_ignore_ascii_case("otlp"))) {
388        return None;
389    }
390    let endpoint = var("OTEL_EXPORTER_OTLP_LOGS_ENDPOINT")
391        .or_else(|| var("OTEL_EXPORTER_OTLP_ENDPOINT").map(|base| format!("{}/v1/logs", base.trim_end_matches('/'))))
392        .unwrap_or_else(|| "http://localhost:4318/v1/logs".to_string());
393    let interval = var("OTEL_BLRP_SCHEDULE_DELAY").and_then(|v| v.parse().ok()).unwrap_or(1000);
394    let headers = var("OTEL_EXPORTER_OTLP_LOGS_HEADERS").or_else(|| var("OTEL_EXPORTER_OTLP_HEADERS")).map(|h| super::parse_headers(&h)).unwrap_or_default();
395    Some(LogsExportConfig {
396        endpoint,
397        interval: Duration::from_millis(interval),
398        service_name: var("OTEL_SERVICE_NAME").unwrap_or_else(|| "rws".to_string()),
399        service_version: env!("CARGO_PKG_VERSION").to_string(),
400        headers,
401    })
402}
403
404/// Stop the background thread after sending what is buffered (called by
405/// [`super::shutdown`]).
406#[doc(hidden)]
407pub fn shutdown() {
408    if let Some(exporter) = EXPORTER.get() {
409        if !exporter.stop.swap(true, Ordering::Relaxed) {
410            exporter.wake.1.notify_all();
411            export(exporter);
412        }
413    }
414}
415
416fn export(exporter: &Exporter) {
417    let _sending = exporter.sending.lock().unwrap_or_else(|e| e.into_inner());
418    for records in drain() {
419        let body = body(&records, &exporter.config.service_name, &exporter.config.service_version);
420        let client = crate::http_client::Client::new().timeout_ms(10_000);
421        let mut request = client.post(&exporter.config.endpoint).header("Content-Type", "application/json").body(body);
422        for (name, value) in &exporter.config.headers {
423            request = request.header(name, value);
424        }
425        match request.send() {
426            Ok(response) if response.is_success() => {}
427            Ok(response) => eprintln!("OTLP logs export to {}: HTTP {}", exporter.config.endpoint, response.status()),
428            Err(e) => eprintln!("OTLP logs export to {}: {}", exporter.config.endpoint, e),
429        }
430    }
431}
432
433#[cfg(test)]
434thread_local! {
435    /// Recording on for this test thread only (no exporter): the process-wide
436    /// switch would change other tests' responses (`OtelLayer`'s header).
437    static TEST_ENABLED: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
438}
439
440#[cfg(test)]
441fn enable_on_this_thread(on: bool) {
442    TEST_ENABLED.with(|e| e.set(on));
443}
444
445#[cfg(test)]
446mod tests;
rws_server/otel/logs.rs

rws_server/otel/
logs.rs
