feagi_observability/
ring_layer.rs

1// Copyright 2025 Neuraville Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4//! In-process log ring buffer + tracing layer.
5//!
6//! `init_logging` installs a [`RingBufferLayer`] that captures recent log records
7//! into a fixed-capacity ring buffer. The buffer is reachable through a global
8//! singleton ([`global_ring`]) so REST handlers (e.g. `/v1/system/log_tail`) can
9//! return diagnostics to clients without scraping log files.
10//!
11//! Design notes:
12//! - The buffer is bounded to `FEAGI_LOG_RING_BUFFER_CAPACITY` records (default 2000)
13//!   so memory usage stays predictable regardless of log volume.
14//! - Records are written non-blocking; the `tracing` layer never blocks the caller.
15//! - Reads clone the requested slice so callers do not hold the lock.
16//! - Setting `FEAGI_LOG_RING_BUFFER_CAPACITY=0` disables the layer entirely.
17
18use std::collections::VecDeque;
19use std::fmt::Write as _;
20use std::sync::{Arc, OnceLock};
21use std::time::{SystemTime, UNIX_EPOCH};
22
23use parking_lot::RwLock;
24use serde::Serialize;
25use tracing::field::{Field, Visit};
26use tracing::{Event, Level, Subscriber};
27use tracing_subscriber::layer::Context;
28use tracing_subscriber::Layer;
29
30/// Default ring buffer capacity (records). Override via FEAGI_LOG_RING_BUFFER_CAPACITY.
31pub const DEFAULT_CAPACITY: usize = 2_000;
32
33/// Environment variable name for capacity override (set to "0" to disable layer).
34pub const CAPACITY_ENV_VAR: &str = "FEAGI_LOG_RING_BUFFER_CAPACITY";
35
36/// One captured log record.
37#[derive(Debug, Clone, Serialize)]
38pub struct LogRecord {
39    /// Wall-clock millisecond timestamp when the record was emitted.
40    pub timestamp_ms: i64,
41    /// Severity level: `TRACE` / `DEBUG` / `INFO` / `WARN` / `ERROR`.
42    pub level: String,
43    /// Tracing target (typically `crate_name::module`).
44    pub target: String,
45    /// Source file (best-effort, may be empty in release builds).
46    pub file: String,
47    /// Source line number (0 when unavailable).
48    pub line: u32,
49    /// Human-readable message extracted from the event's `message` field.
50    pub message: String,
51    /// Additional structured fields serialised as JSON.
52    #[serde(skip_serializing_if = "Option::is_none")]
53    pub fields: Option<serde_json::Value>,
54}
55
56/// Bounded ring buffer of log records (oldest are dropped when full).
57pub struct LogRingBuffer {
58    inner: RwLock<VecDeque<LogRecord>>,
59    capacity: usize,
60}
61
62impl LogRingBuffer {
63    /// Create a new ring buffer with the given capacity. A capacity of 0 means the
64    /// buffer is effectively disabled (pushes are no-ops, snapshots are empty).
65    pub fn new(capacity: usize) -> Self {
66        let initial_capacity = capacity.min(64);
67        Self {
68            inner: RwLock::new(VecDeque::with_capacity(initial_capacity)),
69            capacity,
70        }
71    }
72
73    /// Maximum number of records the buffer can hold.
74    pub fn capacity(&self) -> usize {
75        self.capacity
76    }
77
78    /// Append a record, dropping the oldest one if the buffer is full.
79    pub fn push(&self, record: LogRecord) {
80        if self.capacity == 0 {
81            return;
82        }
83        let mut inner = self.inner.write();
84        if inner.len() == self.capacity {
85            inner.pop_front();
86        }
87        inner.push_back(record);
88    }
89
90    /// Return a filtered, ordered (oldest first) snapshot of buffered records.
91    ///
92    /// # Arguments
93    /// * `since_ts_ms` - drop records with `timestamp_ms < since_ts_ms`
94    /// * `min_level`   - drop records below this level (TRACE < DEBUG < INFO < WARN < ERROR)
95    /// * `target_prefix` - drop records whose `target` does not start with this prefix
96    /// * `limit`       - return at most this many records (most recent records win)
97    /// * `message_contains` - case-insensitive substring match on `message`
98    pub fn snapshot(
99        &self,
100        since_ts_ms: Option<i64>,
101        min_level: Option<&str>,
102        target_prefix: Option<&str>,
103        limit: Option<usize>,
104        message_contains: Option<&str>,
105    ) -> Vec<LogRecord> {
106        let inner = self.inner.read();
107        let min_rank = min_level.and_then(level_rank);
108        let message_needle = message_contains
109            .map(str::trim)
110            .filter(|needle| !needle.is_empty())
111            .map(str::to_lowercase);
112        let mut filtered: Vec<LogRecord> = inner
113            .iter()
114            .filter(|r| match since_ts_ms {
115                Some(ts) => r.timestamp_ms >= ts,
116                None => true,
117            })
118            .filter(|r| match (min_rank, level_rank(&r.level)) {
119                (Some(min), Some(level)) => level >= min,
120                _ => true,
121            })
122            .filter(|r| match target_prefix {
123                Some(prefix) => r.target.starts_with(prefix),
124                None => true,
125            })
126            .filter(|r| match message_needle.as_deref() {
127                Some(needle) => r.message.to_lowercase().contains(needle),
128                None => true,
129            })
130            .cloned()
131            .collect();
132
133        if let Some(n) = limit {
134            if filtered.len() > n {
135                let skip = filtered.len() - n;
136                filtered.drain(0..skip);
137            }
138        }
139        filtered
140    }
141
142    /// Clear all buffered records (used in tests).
143    pub fn clear(&self) {
144        self.inner.write().clear();
145    }
146}
147
148/// Convert a level string (case-insensitive) to a numeric rank suitable for
149/// "min level" comparisons. Returns `None` for unknown values.
150fn level_rank(level: &str) -> Option<u8> {
151    match level.to_ascii_uppercase().as_str() {
152        "TRACE" => Some(0),
153        "DEBUG" => Some(1),
154        "INFO" => Some(2),
155        "WARN" | "WARNING" => Some(3),
156        "ERROR" => Some(4),
157        _ => None,
158    }
159}
160
161fn level_to_str(level: &Level) -> &'static str {
162    match *level {
163        Level::TRACE => "TRACE",
164        Level::DEBUG => "DEBUG",
165        Level::INFO => "INFO",
166        Level::WARN => "WARN",
167        Level::ERROR => "ERROR",
168    }
169}
170
171fn now_unix_ms() -> i64 {
172    SystemTime::now()
173        .duration_since(UNIX_EPOCH)
174        .map(|d| d.as_millis() as i64)
175        .unwrap_or(0)
176}
177
178/// Tracing layer that pushes events into the global ring buffer.
179pub struct RingBufferLayer {
180    buffer: Arc<LogRingBuffer>,
181}
182
183impl RingBufferLayer {
184    pub fn new(buffer: Arc<LogRingBuffer>) -> Self {
185        Self { buffer }
186    }
187}
188
189impl<S: Subscriber> Layer<S> for RingBufferLayer {
190    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
191        if self.buffer.capacity() == 0 {
192            return;
193        }
194        let metadata = event.metadata();
195        let mut visitor = MessageVisitor::default();
196        event.record(&mut visitor);
197
198        let fields = if visitor.extra_fields.is_empty() {
199            None
200        } else {
201            Some(serde_json::Value::Object(visitor.extra_fields))
202        };
203
204        let record = LogRecord {
205            timestamp_ms: now_unix_ms(),
206            level: level_to_str(metadata.level()).to_string(),
207            target: metadata.target().to_string(),
208            file: metadata.file().unwrap_or_default().to_string(),
209            line: metadata.line().unwrap_or(0),
210            message: visitor.message,
211            fields,
212        };
213        self.buffer.push(record);
214    }
215}
216
217/// Visitor that extracts the `message` field plus any other structured fields.
218#[derive(Default)]
219struct MessageVisitor {
220    message: String,
221    extra_fields: serde_json::Map<String, serde_json::Value>,
222}
223
224impl Visit for MessageVisitor {
225    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
226        if field.name() == "message" {
227            // Avoid the debug `"..."` wrapping: write directly into the message.
228            let _ = write!(&mut self.message, "{:?}", value);
229        } else {
230            self.extra_fields.insert(
231                field.name().to_string(),
232                serde_json::Value::String(format!("{:?}", value)),
233            );
234        }
235    }
236
237    fn record_str(&mut self, field: &Field, value: &str) {
238        if field.name() == "message" {
239            self.message.push_str(value);
240        } else {
241            self.extra_fields.insert(
242                field.name().to_string(),
243                serde_json::Value::String(value.to_string()),
244            );
245        }
246    }
247
248    fn record_i64(&mut self, field: &Field, value: i64) {
249        self.extra_fields
250            .insert(field.name().to_string(), serde_json::json!(value));
251    }
252
253    fn record_u64(&mut self, field: &Field, value: u64) {
254        self.extra_fields
255            .insert(field.name().to_string(), serde_json::json!(value));
256    }
257
258    fn record_f64(&mut self, field: &Field, value: f64) {
259        self.extra_fields
260            .insert(field.name().to_string(), serde_json::json!(value));
261    }
262
263    fn record_bool(&mut self, field: &Field, value: bool) {
264        self.extra_fields
265            .insert(field.name().to_string(), serde_json::json!(value));
266    }
267}
268
269static GLOBAL_RING: OnceLock<Arc<LogRingBuffer>> = OnceLock::new();
270
271/// Install (once) the global ring buffer with the given capacity. Subsequent
272/// calls are no-ops and return the previously installed instance.
273pub fn install_global_ring(capacity: usize) -> Arc<LogRingBuffer> {
274    GLOBAL_RING
275        .get_or_init(|| Arc::new(LogRingBuffer::new(capacity)))
276        .clone()
277}
278
279/// Resolve the configured capacity from `FEAGI_LOG_RING_BUFFER_CAPACITY`.
280/// Falls back to [`DEFAULT_CAPACITY`] when unset/invalid.
281pub fn capacity_from_env() -> usize {
282    std::env::var(CAPACITY_ENV_VAR)
283        .ok()
284        .and_then(|v| v.parse::<usize>().ok())
285        .unwrap_or(DEFAULT_CAPACITY)
286}
287
288/// Returns the global ring buffer, if installed.
289pub fn global_ring() -> Option<Arc<LogRingBuffer>> {
290    GLOBAL_RING.get().cloned()
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    #[test]
298    fn ring_buffer_respects_capacity() {
299        let ring = LogRingBuffer::new(3);
300        for i in 0..5 {
301            ring.push(LogRecord {
302                timestamp_ms: i as i64,
303                level: "INFO".into(),
304                target: "test".into(),
305                file: String::new(),
306                line: 0,
307                message: format!("msg-{}", i),
308                fields: None,
309            });
310        }
311        let snap = ring.snapshot(None, None, None, None, None);
312        assert_eq!(snap.len(), 3);
313        assert_eq!(snap[0].message, "msg-2");
314        assert_eq!(snap[2].message, "msg-4");
315    }
316
317    #[test]
318    fn snapshot_filters_by_level_and_target() {
319        let ring = LogRingBuffer::new(10);
320        ring.push(LogRecord {
321            timestamp_ms: 100,
322            level: "DEBUG".into(),
323            target: "feagi-api".into(),
324            file: String::new(),
325            line: 0,
326            message: "debug".into(),
327            fields: None,
328        });
329        ring.push(LogRecord {
330            timestamp_ms: 200,
331            level: "ERROR".into(),
332            target: "feagi-burst-engine".into(),
333            file: String::new(),
334            line: 0,
335            message: "error".into(),
336            fields: None,
337        });
338        ring.push(LogRecord {
339            timestamp_ms: 300,
340            level: "WARN".into(),
341            target: "feagi-api".into(),
342            file: String::new(),
343            line: 0,
344            message: "warn".into(),
345            fields: None,
346        });
347
348        let warnings = ring.snapshot(None, Some("warn"), None, None, None);
349        assert_eq!(warnings.len(), 2);
350
351        let api_only = ring.snapshot(None, None, Some("feagi-api"), None, None);
352        assert_eq!(api_only.len(), 2);
353        assert!(api_only.iter().all(|r| r.target.starts_with("feagi-api")));
354
355        let recent = ring.snapshot(Some(250), None, None, None, None);
356        assert_eq!(recent.len(), 1);
357        assert_eq!(recent[0].timestamp_ms, 300);
358
359        let limited = ring.snapshot(None, None, None, Some(2), None);
360        assert_eq!(limited.len(), 2);
361        assert_eq!(limited.last().unwrap().timestamp_ms, 300);
362
363        let by_message = ring.snapshot(None, None, None, None, Some("ERR"));
364        assert_eq!(by_message.len(), 1);
365        assert_eq!(by_message[0].message, "error");
366    }
367
368    #[test]
369    fn capacity_zero_disables_buffer() {
370        let ring = LogRingBuffer::new(0);
371        ring.push(LogRecord {
372            timestamp_ms: 0,
373            level: "INFO".into(),
374            target: "test".into(),
375            file: String::new(),
376            line: 0,
377            message: "x".into(),
378            fields: None,
379        });
380        assert!(ring.snapshot(None, None, None, None, None).is_empty());
381    }
382}
feagi_observability/ring_layer.rs

feagi_observability/
ring_layer.rs
