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    pub fn snapshot(
98        &self,
99        since_ts_ms: Option<i64>,
100        min_level: Option<&str>,
101        target_prefix: Option<&str>,
102        limit: Option<usize>,
103    ) -> Vec<LogRecord> {
104        let inner = self.inner.read();
105        let min_rank = min_level.and_then(level_rank);
106        let mut filtered: Vec<LogRecord> = inner
107            .iter()
108            .filter(|r| match since_ts_ms {
109                Some(ts) => r.timestamp_ms >= ts,
110                None => true,
111            })
112            .filter(|r| match (min_rank, level_rank(&r.level)) {
113                (Some(min), Some(level)) => level >= min,
114                _ => true,
115            })
116            .filter(|r| match target_prefix {
117                Some(prefix) => r.target.starts_with(prefix),
118                None => true,
119            })
120            .cloned()
121            .collect();
122
123        if let Some(n) = limit {
124            if filtered.len() > n {
125                let skip = filtered.len() - n;
126                filtered.drain(0..skip);
127            }
128        }
129        filtered
130    }
131
132    /// Clear all buffered records (used in tests).
133    pub fn clear(&self) {
134        self.inner.write().clear();
135    }
136}
137
138/// Convert a level string (case-insensitive) to a numeric rank suitable for
139/// "min level" comparisons. Returns `None` for unknown values.
140fn level_rank(level: &str) -> Option<u8> {
141    match level.to_ascii_uppercase().as_str() {
142        "TRACE" => Some(0),
143        "DEBUG" => Some(1),
144        "INFO" => Some(2),
145        "WARN" | "WARNING" => Some(3),
146        "ERROR" => Some(4),
147        _ => None,
148    }
149}
150
151fn level_to_str(level: &Level) -> &'static str {
152    match *level {
153        Level::TRACE => "TRACE",
154        Level::DEBUG => "DEBUG",
155        Level::INFO => "INFO",
156        Level::WARN => "WARN",
157        Level::ERROR => "ERROR",
158    }
159}
160
161fn now_unix_ms() -> i64 {
162    SystemTime::now()
163        .duration_since(UNIX_EPOCH)
164        .map(|d| d.as_millis() as i64)
165        .unwrap_or(0)
166}
167
168/// Tracing layer that pushes events into the global ring buffer.
169pub struct RingBufferLayer {
170    buffer: Arc<LogRingBuffer>,
171}
172
173impl RingBufferLayer {
174    pub fn new(buffer: Arc<LogRingBuffer>) -> Self {
175        Self { buffer }
176    }
177}
178
179impl<S: Subscriber> Layer<S> for RingBufferLayer {
180    fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) {
181        if self.buffer.capacity() == 0 {
182            return;
183        }
184        let metadata = event.metadata();
185        let mut visitor = MessageVisitor::default();
186        event.record(&mut visitor);
187
188        let fields = if visitor.extra_fields.is_empty() {
189            None
190        } else {
191            Some(serde_json::Value::Object(visitor.extra_fields))
192        };
193
194        let record = LogRecord {
195            timestamp_ms: now_unix_ms(),
196            level: level_to_str(metadata.level()).to_string(),
197            target: metadata.target().to_string(),
198            file: metadata.file().unwrap_or_default().to_string(),
199            line: metadata.line().unwrap_or(0),
200            message: visitor.message,
201            fields,
202        };
203        self.buffer.push(record);
204    }
205}
206
207/// Visitor that extracts the `message` field plus any other structured fields.
208#[derive(Default)]
209struct MessageVisitor {
210    message: String,
211    extra_fields: serde_json::Map<String, serde_json::Value>,
212}
213
214impl Visit for MessageVisitor {
215    fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
216        if field.name() == "message" {
217            // Avoid the debug `"..."` wrapping: write directly into the message.
218            let _ = write!(&mut self.message, "{:?}", value);
219        } else {
220            self.extra_fields.insert(
221                field.name().to_string(),
222                serde_json::Value::String(format!("{:?}", value)),
223            );
224        }
225    }
226
227    fn record_str(&mut self, field: &Field, value: &str) {
228        if field.name() == "message" {
229            self.message.push_str(value);
230        } else {
231            self.extra_fields.insert(
232                field.name().to_string(),
233                serde_json::Value::String(value.to_string()),
234            );
235        }
236    }
237
238    fn record_i64(&mut self, field: &Field, value: i64) {
239        self.extra_fields
240            .insert(field.name().to_string(), serde_json::json!(value));
241    }
242
243    fn record_u64(&mut self, field: &Field, value: u64) {
244        self.extra_fields
245            .insert(field.name().to_string(), serde_json::json!(value));
246    }
247
248    fn record_f64(&mut self, field: &Field, value: f64) {
249        self.extra_fields
250            .insert(field.name().to_string(), serde_json::json!(value));
251    }
252
253    fn record_bool(&mut self, field: &Field, value: bool) {
254        self.extra_fields
255            .insert(field.name().to_string(), serde_json::json!(value));
256    }
257}
258
259static GLOBAL_RING: OnceLock<Arc<LogRingBuffer>> = OnceLock::new();
260
261/// Install (once) the global ring buffer with the given capacity. Subsequent
262/// calls are no-ops and return the previously installed instance.
263pub fn install_global_ring(capacity: usize) -> Arc<LogRingBuffer> {
264    GLOBAL_RING
265        .get_or_init(|| Arc::new(LogRingBuffer::new(capacity)))
266        .clone()
267}
268
269/// Resolve the configured capacity from `FEAGI_LOG_RING_BUFFER_CAPACITY`.
270/// Falls back to [`DEFAULT_CAPACITY`] when unset/invalid.
271pub fn capacity_from_env() -> usize {
272    std::env::var(CAPACITY_ENV_VAR)
273        .ok()
274        .and_then(|v| v.parse::<usize>().ok())
275        .unwrap_or(DEFAULT_CAPACITY)
276}
277
278/// Returns the global ring buffer, if installed.
279pub fn global_ring() -> Option<Arc<LogRingBuffer>> {
280    GLOBAL_RING.get().cloned()
281}
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286
287    #[test]
288    fn ring_buffer_respects_capacity() {
289        let ring = LogRingBuffer::new(3);
290        for i in 0..5 {
291            ring.push(LogRecord {
292                timestamp_ms: i as i64,
293                level: "INFO".into(),
294                target: "test".into(),
295                file: String::new(),
296                line: 0,
297                message: format!("msg-{}", i),
298                fields: None,
299            });
300        }
301        let snap = ring.snapshot(None, None, None, None);
302        assert_eq!(snap.len(), 3);
303        assert_eq!(snap[0].message, "msg-2");
304        assert_eq!(snap[2].message, "msg-4");
305    }
306
307    #[test]
308    fn snapshot_filters_by_level_and_target() {
309        let ring = LogRingBuffer::new(10);
310        ring.push(LogRecord {
311            timestamp_ms: 100,
312            level: "DEBUG".into(),
313            target: "feagi-api".into(),
314            file: String::new(),
315            line: 0,
316            message: "debug".into(),
317            fields: None,
318        });
319        ring.push(LogRecord {
320            timestamp_ms: 200,
321            level: "ERROR".into(),
322            target: "feagi-burst-engine".into(),
323            file: String::new(),
324            line: 0,
325            message: "error".into(),
326            fields: None,
327        });
328        ring.push(LogRecord {
329            timestamp_ms: 300,
330            level: "WARN".into(),
331            target: "feagi-api".into(),
332            file: String::new(),
333            line: 0,
334            message: "warn".into(),
335            fields: None,
336        });
337
338        let warnings = ring.snapshot(None, Some("warn"), None, None);
339        assert_eq!(warnings.len(), 2);
340
341        let api_only = ring.snapshot(None, None, Some("feagi-api"), None);
342        assert_eq!(api_only.len(), 2);
343        assert!(api_only.iter().all(|r| r.target.starts_with("feagi-api")));
344
345        let recent = ring.snapshot(Some(250), None, None, None);
346        assert_eq!(recent.len(), 1);
347        assert_eq!(recent[0].timestamp_ms, 300);
348
349        let limited = ring.snapshot(None, None, None, Some(2));
350        assert_eq!(limited.len(), 2);
351        assert_eq!(limited.last().unwrap().timestamp_ms, 300);
352    }
353
354    #[test]
355    fn capacity_zero_disables_buffer() {
356        let ring = LogRingBuffer::new(0);
357        ring.push(LogRecord {
358            timestamp_ms: 0,
359            level: "INFO".into(),
360            target: "test".into(),
361            file: String::new(),
362            line: 0,
363            message: "x".into(),
364            fields: None,
365        });
366        assert!(ring.snapshot(None, None, None, None).is_empty());
367    }
368}
feagi_observability/ring_layer.rs

feagi_observability/
ring_layer.rs
