clawft_kernel/
log_service.rs

1//! Structured log service with ring buffer, query, and subscription support.
2//!
3//! [`LogService`] provides structured logging for all kernel subsystems.
4//! Logs are stored in a bounded ring buffer and queryable by PID, service,
5//! level, time range, and trace ID. Real-time subscribers receive entries
6//! as they are ingested.
7
8use std::collections::{HashMap, VecDeque};
9use std::sync::RwLock;
10
11use chrono::{DateTime, Utc};
12use dashmap::DashMap;
13use serde::{Deserialize, Serialize};
14use tokio::sync::mpsc;
15
16use crate::console::LogLevel;
17use crate::process::Pid;
18
19/// Default maximum number of log entries retained in the ring buffer.
20pub const DEFAULT_LOG_CAPACITY: usize = 100_000;
21
22/// Default subscriber channel capacity.
23const SUBSCRIBER_CHANNEL_CAPACITY: usize = 1024;
24
25/// A structured log entry.
26#[derive(Debug, Clone, Serialize, Deserialize)]
27pub struct LogEntry {
28    /// When the log entry was created.
29    pub timestamp: DateTime<Utc>,
30    /// Log severity level.
31    pub level: LogLevel,
32    /// PID of the source agent (if applicable).
33    pub source_pid: Option<Pid>,
34    /// Name of the source service (if applicable).
35    pub source_service: Option<String>,
36    /// Human-readable log message.
37    pub message: String,
38    /// Structured fields (key-value metadata).
39    pub fields: HashMap<String, serde_json::Value>,
40    /// Distributed trace ID for correlated log queries.
41    pub trace_id: Option<String>,
42}
43
44impl LogEntry {
45    /// Create a new log entry with minimal fields.
46    pub fn new(level: LogLevel, message: impl Into<String>) -> Self {
47        Self {
48            timestamp: Utc::now(),
49            level,
50            source_pid: None,
51            source_service: None,
52            message: message.into(),
53            fields: HashMap::new(),
54            trace_id: None,
55        }
56    }
57
58    /// Set the source PID.
59    pub fn with_pid(mut self, pid: Pid) -> Self {
60        self.source_pid = Some(pid);
61        self
62    }
63
64    /// Set the source service name.
65    pub fn with_service(mut self, service: impl Into<String>) -> Self {
66        self.source_service = Some(service.into());
67        self
68    }
69
70    /// Set the trace ID.
71    pub fn with_trace_id(mut self, trace_id: impl Into<String>) -> Self {
72        self.trace_id = Some(trace_id.into());
73        self
74    }
75
76    /// Add a structured field.
77    pub fn with_field(mut self, key: impl Into<String>, value: serde_json::Value) -> Self {
78        self.fields.insert(key.into(), value);
79        self
80    }
81}
82
83/// Query parameters for filtering log entries.
84#[derive(Debug, Clone, Default, Serialize, Deserialize)]
85pub struct LogQuery {
86    /// Filter by source PID.
87    pub pid: Option<Pid>,
88    /// Filter by source service name.
89    pub service: Option<String>,
90    /// Minimum log level (inclusive).
91    pub level_min: Option<LogLevel>,
92    /// Start of time range (inclusive).
93    pub since: Option<DateTime<Utc>>,
94    /// End of time range (inclusive).
95    pub until: Option<DateTime<Utc>>,
96    /// Maximum number of entries to return.
97    pub limit: usize,
98    /// Filter by trace ID.
99    pub trace_id: Option<String>,
100}
101
102impl LogQuery {
103    /// Create a query with a limit.
104    pub fn new(limit: usize) -> Self {
105        Self {
106            limit,
107            ..Default::default()
108        }
109    }
110}
111
112/// Map LogLevel to a numeric severity for ordering.
113fn level_severity(level: &LogLevel) -> u8 {
114    match level {
115        LogLevel::Debug => 0,
116        LogLevel::Info => 1,
117        LogLevel::Warn => 2,
118        LogLevel::Error => 3,
119    }
120}
121
122/// Structured logging service with ring buffer and query support.
123///
124/// Stores log entries in a bounded ring buffer (oldest evicted when full).
125/// Supports query by PID, service, level, time range, and trace ID.
126/// Real-time subscribers receive entries via `mpsc` channels.
127pub struct LogService {
128    entries: RwLock<VecDeque<LogEntry>>,
129    max_entries: usize,
130    subscribers: DashMap<String, mpsc::Sender<LogEntry>>,
131}
132
133impl LogService {
134    /// Create a new log service with the given capacity.
135    pub fn new(max_entries: usize) -> Self {
136        Self {
137            entries: RwLock::new(VecDeque::with_capacity(max_entries.min(8192))),
138            max_entries,
139            subscribers: DashMap::new(),
140        }
141    }
142
143    /// Create a log service with default capacity (100,000).
144    pub fn with_default_capacity() -> Self {
145        Self::new(DEFAULT_LOG_CAPACITY)
146    }
147
148    /// Ingest a log entry.
149    ///
150    /// Stores in the ring buffer and fans out to all subscribers.
151    /// Write lock is released before subscriber notification to
152    /// prevent blocking on slow consumers.
153    pub fn ingest(&self, entry: LogEntry) {
154        // Store in ring buffer
155        {
156            let mut entries = self.entries.write().expect("log write lock poisoned");
157            while entries.len() >= self.max_entries {
158                entries.pop_front();
159            }
160            entries.push_back(entry.clone());
161        }
162
163        // Fan out to subscribers (non-blocking)
164        let mut dead_subs = Vec::new();
165        for sub in self.subscribers.iter() {
166            if sub.value().try_send(entry.clone()).is_err() {
167                dead_subs.push(sub.key().clone());
168            }
169        }
170        for key in dead_subs {
171            self.subscribers.remove(&key);
172        }
173    }
174
175    /// Log a convenience method: ingest with level, message, and optional PID.
176    pub fn log(&self, level: LogLevel, message: impl Into<String>, pid: Option<Pid>) {
177        let mut entry = LogEntry::new(level, message);
178        entry.source_pid = pid;
179        self.ingest(entry);
180    }
181
182    /// Query log entries matching the given criteria.
183    pub fn query(&self, query: &LogQuery) -> Vec<LogEntry> {
184        let entries = self.entries.read().expect("log read lock poisoned");
185        entries
186            .iter()
187            .filter(|e| query.pid.is_none_or(|pid| e.source_pid == Some(pid)))
188            .filter(|e| {
189                query
190                    .service
191                    .as_ref()
192                    .is_none_or(|s| e.source_service.as_ref() == Some(s))
193            })
194            .filter(|e| {
195                query
196                    .level_min
197                    .as_ref()
198                    .is_none_or(|min| level_severity(&e.level) >= level_severity(min))
199            })
200            .filter(|e| query.since.is_none_or(|since| e.timestamp >= since))
201            .filter(|e| query.until.is_none_or(|until| e.timestamp <= until))
202            .filter(|e| {
203                query
204                    .trace_id
205                    .as_ref()
206                    .is_none_or(|tid| e.trace_id.as_ref() == Some(tid))
207            })
208            .take(query.limit)
209            .cloned()
210            .collect()
211    }
212
213    /// Subscribe to real-time log entries.
214    ///
215    /// Returns a receiver that will receive all new log entries.
216    /// The subscription ID can be used to unsubscribe.
217    pub fn subscribe(&self) -> (String, mpsc::Receiver<LogEntry>) {
218        let id = uuid::Uuid::new_v4().to_string();
219        let (tx, rx) = mpsc::channel(SUBSCRIBER_CHANNEL_CAPACITY);
220        self.subscribers.insert(id.clone(), tx);
221        (id, rx)
222    }
223
224    /// Unsubscribe from real-time log entries.
225    pub fn unsubscribe(&self, id: &str) {
226        self.subscribers.remove(id);
227    }
228
229    /// Number of log entries currently stored.
230    pub fn len(&self) -> usize {
231        self.entries.read().expect("log read lock poisoned").len()
232    }
233
234    /// Whether the log buffer is empty.
235    pub fn is_empty(&self) -> bool {
236        self.len() == 0
237    }
238
239    /// Maximum capacity of the ring buffer.
240    pub fn capacity(&self) -> usize {
241        self.max_entries
242    }
243
244    /// Number of active subscribers.
245    pub fn subscriber_count(&self) -> usize {
246        self.subscribers.len()
247    }
248
249    /// Clear all log entries.
250    pub fn clear(&self) {
251        self.entries
252            .write()
253            .expect("log write lock poisoned")
254            .clear();
255    }
256}
257
258impl Default for LogService {
259    fn default() -> Self {
260        Self::with_default_capacity()
261    }
262}
263
264#[cfg(test)]
265mod tests {
266    use super::*;
267
268    fn info_entry(msg: &str) -> LogEntry {
269        LogEntry::new(LogLevel::Info, msg)
270    }
271
272    #[test]
273    fn ingest_and_len() {
274        let service = LogService::new(100);
275        assert!(service.is_empty());
276
277        service.ingest(info_entry("hello"));
278        assert_eq!(service.len(), 1);
279    }
280
281    #[test]
282    fn ring_buffer_eviction() {
283        let service = LogService::new(3);
284
285        for i in 0..5 {
286            service.ingest(info_entry(&format!("msg-{i}")));
287        }
288
289        assert_eq!(service.len(), 3);
290        let results = service.query(&LogQuery::new(10));
291        assert_eq!(results[0].message, "msg-2");
292        assert_eq!(results[2].message, "msg-4");
293    }
294
295    #[test]
296    fn query_by_pid() {
297        let service = LogService::new(100);
298        service.ingest(info_entry("from-1").with_pid(1));
299        service.ingest(info_entry("from-2").with_pid(2));
300        service.ingest(info_entry("from-1-again").with_pid(1));
301
302        let results = service.query(&LogQuery {
303            pid: Some(1),
304            limit: 10,
305            ..Default::default()
306        });
307        assert_eq!(results.len(), 2);
308        assert!(results.iter().all(|e| e.source_pid == Some(1)));
309    }
310
311    #[test]
312    fn query_by_service() {
313        let service = LogService::new(100);
314        service.ingest(info_entry("a").with_service("auth"));
315        service.ingest(info_entry("b").with_service("db"));
316        service.ingest(info_entry("c").with_service("auth"));
317
318        let results = service.query(&LogQuery {
319            service: Some("auth".into()),
320            limit: 10,
321            ..Default::default()
322        });
323        assert_eq!(results.len(), 2);
324    }
325
326    #[test]
327    fn query_by_level() {
328        let service = LogService::new(100);
329        service.ingest(LogEntry::new(LogLevel::Debug, "debug-msg"));
330        service.ingest(LogEntry::new(LogLevel::Info, "info-msg"));
331        service.ingest(LogEntry::new(LogLevel::Warn, "warn-msg"));
332        service.ingest(LogEntry::new(LogLevel::Error, "error-msg"));
333
334        let results = service.query(&LogQuery {
335            level_min: Some(LogLevel::Warn),
336            limit: 10,
337            ..Default::default()
338        });
339        assert_eq!(results.len(), 2);
340        assert!(
341            results
342                .iter()
343                .all(|e| matches!(e.level, LogLevel::Warn | LogLevel::Error))
344        );
345    }
346
347    #[test]
348    fn query_by_time_range() {
349        let service = LogService::new(100);
350        let before = Utc::now();
351        service.ingest(info_entry("in-range"));
352        let after = Utc::now();
353
354        let results = service.query(&LogQuery {
355            since: Some(before),
356            until: Some(after),
357            limit: 10,
358            ..Default::default()
359        });
360        assert_eq!(results.len(), 1);
361    }
362
363    #[test]
364    fn query_by_trace_id() {
365        let service = LogService::new(100);
366        service.ingest(info_entry("traced").with_trace_id("trace-abc"));
367        service.ingest(info_entry("untraced"));
368        service.ingest(info_entry("other-trace").with_trace_id("trace-def"));
369
370        let results = service.query(&LogQuery {
371            trace_id: Some("trace-abc".into()),
372            limit: 10,
373            ..Default::default()
374        });
375        assert_eq!(results.len(), 1);
376        assert_eq!(results[0].message, "traced");
377    }
378
379    #[test]
380    fn query_limit() {
381        let service = LogService::new(100);
382        for i in 0..10 {
383            service.ingest(info_entry(&format!("msg-{i}")));
384        }
385
386        let results = service.query(&LogQuery::new(3));
387        assert_eq!(results.len(), 3);
388    }
389
390    #[tokio::test]
391    async fn subscriber_receives_entries() {
392        let service = LogService::new(100);
393        let (_id, mut rx) = service.subscribe();
394
395        service.ingest(info_entry("sub-msg-1"));
396        service.ingest(info_entry("sub-msg-2"));
397
398        let m1 = rx.recv().await.unwrap();
399        let m2 = rx.recv().await.unwrap();
400        assert_eq!(m1.message, "sub-msg-1");
401        assert_eq!(m2.message, "sub-msg-2");
402    }
403
404    #[test]
405    fn unsubscribe() {
406        let service = LogService::new(100);
407        let (id, _rx) = service.subscribe();
408        assert_eq!(service.subscriber_count(), 1);
409
410        service.unsubscribe(&id);
411        assert_eq!(service.subscriber_count(), 0);
412    }
413
414    #[test]
415    fn dead_subscriber_removed() {
416        let service = LogService::new(100);
417        let (_id, rx) = service.subscribe();
418        assert_eq!(service.subscriber_count(), 1);
419
420        // Drop the receiver
421        drop(rx);
422
423        // Ingesting should clean up dead subscribers
424        service.ingest(info_entry("after-drop"));
425        assert_eq!(service.subscriber_count(), 0);
426    }
427
428    #[test]
429    fn clear() {
430        let service = LogService::new(100);
431        service.ingest(info_entry("a"));
432        service.ingest(info_entry("b"));
433        assert_eq!(service.len(), 2);
434
435        service.clear();
436        assert!(service.is_empty());
437    }
438
439    #[test]
440    fn log_convenience() {
441        let service = LogService::new(100);
442        service.log(LogLevel::Warn, "warning!", Some(42));
443
444        let results = service.query(&LogQuery {
445            pid: Some(42),
446            limit: 10,
447            ..Default::default()
448        });
449        assert_eq!(results.len(), 1);
450        assert_eq!(results[0].message, "warning!");
451        assert!(matches!(results[0].level, LogLevel::Warn));
452    }
453
454    #[test]
455    fn log_entry_builder() {
456        let entry = LogEntry::new(LogLevel::Info, "test")
457            .with_pid(5)
458            .with_service("auth")
459            .with_trace_id("t-123")
460            .with_field("key", serde_json::json!("value"));
461
462        assert_eq!(entry.source_pid, Some(5));
463        assert_eq!(entry.source_service.as_deref(), Some("auth"));
464        assert_eq!(entry.trace_id.as_deref(), Some("t-123"));
465        assert_eq!(
466            entry.fields.get("key").unwrap(),
467            &serde_json::json!("value")
468        );
469    }
470
471    #[test]
472    fn log_entry_serde_roundtrip() {
473        let entry = LogEntry::new(LogLevel::Error, "boom")
474            .with_pid(7)
475            .with_service("db")
476            .with_trace_id("trace-999")
477            .with_field("code", serde_json::json!(500));
478
479        let json = serde_json::to_string(&entry).unwrap();
480        let restored: LogEntry = serde_json::from_str(&json).unwrap();
481        assert_eq!(restored.message, "boom");
482        assert_eq!(restored.source_pid, Some(7));
483        assert_eq!(restored.fields["code"], serde_json::json!(500));
484    }
485
486    #[test]
487    fn default_service() {
488        let service = LogService::default();
489        assert_eq!(service.capacity(), DEFAULT_LOG_CAPACITY);
490    }
491
492    #[test]
493    fn log_query_default() {
494        let query = LogQuery::default();
495        assert!(query.pid.is_none());
496        assert_eq!(query.limit, 0);
497    }
498}
clawft_kernel/log_service.rs

clawft_kernel/
log_service.rs
