clawft_kernel/
mesh_log.rs

1//! Cross-node log aggregation for mesh networking (K6-G2).
2//!
3//! [`LogAggregator`] collects log entries from mesh peers and provides
4//! a unified query interface that merges logs by timestamp.
5//!
6//! # Status (WEFT-151 audit)
7//!
8//! **Orphan — library only.** No production callers outside this module
9//! and the `clawft_kernel` re-export. Not used by `boot.rs`, daemon RPC,
10//! CLI, or `MeshRuntime`. Unit tests cover the API; no integration path.
11//!
12//! **Disposition:** keep as library surface for mesh observability.
13//! **Wiring schedule (0.8.x+):** attach to mesh admin / observability RPC
14//! (query remote peer logs) once a daemon mesh-status surface lands
15//! (related: WEFT-117 assess mesh-status). Do not remove; do not
16//! `#[deprecated]` (clippy deny-warnings — see WEFT-671).
17
18use chrono::{DateTime, Utc};
19use dashmap::DashMap;
20use serde::{Deserialize, Serialize};
21
22// ---------------------------------------------------------------------------
23// RemoteLogEntry
24// ---------------------------------------------------------------------------
25
26/// A log entry that includes the source node identifier.
27#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
28pub struct RemoteLogEntry {
29    /// Source node identifier (local or remote).
30    pub node_id: String,
31    /// Log level.
32    pub level: String,
33    /// Log message.
34    pub message: String,
35    /// Originating module or service.
36    pub source: String,
37    /// When the log entry was created.
38    pub timestamp: DateTime<Utc>,
39}
40
41impl RemoteLogEntry {
42    /// Create a new log entry.
43    pub fn new(
44        node_id: &str,
45        level: &str,
46        message: &str,
47        source: &str,
48        timestamp: DateTime<Utc>,
49    ) -> Self {
50        Self {
51            node_id: node_id.to_string(),
52            level: level.to_string(),
53            message: message.to_string(),
54            source: source.to_string(),
55            timestamp,
56        }
57    }
58}
59
60// ---------------------------------------------------------------------------
61// LogQuery
62// ---------------------------------------------------------------------------
63
64/// Query parameters for filtering aggregated logs.
65#[derive(Debug, Clone, Default, Serialize, Deserialize)]
66pub struct LogQuery {
67    /// Filter by source node (None = all nodes).
68    pub node_id: Option<String>,
69    /// Filter by log level.
70    pub level: Option<String>,
71    /// Filter by source module/service.
72    pub source: Option<String>,
73    /// Only entries after this time.
74    pub since: Option<DateTime<Utc>>,
75    /// Only entries before this time.
76    pub until: Option<DateTime<Utc>>,
77    /// Maximum number of entries to return.
78    pub limit: Option<usize>,
79}
80
81// ---------------------------------------------------------------------------
82// LogAggregator
83// ---------------------------------------------------------------------------
84
85/// Aggregates log entries from local and remote mesh nodes.
86///
87/// Stores entries per node and provides a merged, time-ordered query
88/// interface.
89pub struct LogAggregator {
90    /// Local node identifier.
91    local_node_id: String,
92    /// Entries per node: node_id -> entries (append-only).
93    entries: DashMap<String, Vec<RemoteLogEntry>>,
94}
95
96impl LogAggregator {
97    /// Create a new log aggregator for the given local node.
98    pub fn new(local_node_id: String) -> Self {
99        Self {
100            local_node_id,
101            entries: DashMap::new(),
102        }
103    }
104
105    /// Add a local log entry.
106    pub fn log_local(&self, level: &str, message: &str, source: &str) {
107        let entry = RemoteLogEntry::new(&self.local_node_id, level, message, source, Utc::now());
108        self.entries
109            .entry(self.local_node_id.clone())
110            .or_default()
111            .push(entry);
112    }
113
114    /// Add a remote log entry received from a mesh peer.
115    pub fn add_remote(&self, entry: RemoteLogEntry) {
116        self.entries
117            .entry(entry.node_id.clone())
118            .or_default()
119            .push(entry);
120    }
121
122    /// Add a batch of remote log entries.
123    pub fn add_remote_batch(&self, entries: Vec<RemoteLogEntry>) {
124        for entry in entries {
125            self.add_remote(entry);
126        }
127    }
128
129    /// Query logs with optional filters, merged by timestamp.
130    pub fn query(&self, query: &LogQuery) -> Vec<RemoteLogEntry> {
131        let mut results: Vec<RemoteLogEntry> = Vec::new();
132
133        for entry_ref in self.entries.iter() {
134            let node_id = entry_ref.key();
135            let entries = entry_ref.value();
136
137            // Filter by node_id.
138            if query.node_id.as_ref().is_some_and(|n| node_id != n) {
139                continue;
140            }
141
142            for e in entries {
143                // Filter by level.
144                if query.level.as_ref().is_some_and(|l| e.level != *l) {
145                    continue;
146                }
147                // Filter by source.
148                if query.source.as_ref().is_some_and(|s| e.source != *s) {
149                    continue;
150                }
151                // Filter by time range.
152                if query.since.is_some_and(|since| e.timestamp < since) {
153                    continue;
154                }
155                if query.until.is_some_and(|until| e.timestamp > until) {
156                    continue;
157                }
158
159                results.push(e.clone());
160            }
161        }
162
163        // Sort by timestamp (merge from multiple nodes).
164        results.sort_by_key(|e| e.timestamp);
165
166        // Apply limit.
167        if let Some(limit) = query.limit {
168            results.truncate(limit);
169        }
170
171        results
172    }
173
174    /// Get all entries for a specific node.
175    pub fn entries_for_node(&self, node_id: &str) -> Vec<RemoteLogEntry> {
176        self.entries
177            .get(node_id)
178            .map(|v| v.value().clone())
179            .unwrap_or_default()
180    }
181
182    /// Total number of log entries across all nodes.
183    pub fn total_entries(&self) -> usize {
184        self.entries.iter().map(|e| e.value().len()).sum()
185    }
186
187    /// List all known node IDs.
188    pub fn known_nodes(&self) -> Vec<String> {
189        self.entries.iter().map(|e| e.key().clone()).collect()
190    }
191}
192
193// ── Tests ─────────────────────────────────────────────────────────────────
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198    use chrono::Duration;
199
200    fn make_entry(node: &str, level: &str, msg: &str, offset_secs: i64) -> RemoteLogEntry {
201        RemoteLogEntry::new(
202            node,
203            level,
204            msg,
205            "test-module",
206            Utc::now() + Duration::seconds(offset_secs),
207        )
208    }
209
210    #[test]
211    fn log_local_entry() {
212        let agg = LogAggregator::new("node-1".into());
213        agg.log_local("info", "hello", "boot");
214        assert_eq!(agg.total_entries(), 1);
215        let entries = agg.entries_for_node("node-1");
216        assert_eq!(entries[0].message, "hello");
217        assert_eq!(entries[0].node_id, "node-1");
218    }
219
220    #[test]
221    fn add_remote_entry() {
222        let agg = LogAggregator::new("node-1".into());
223        let entry = make_entry("node-2", "warn", "remote warning", 0);
224        agg.add_remote(entry);
225        assert_eq!(agg.total_entries(), 1);
226        let entries = agg.entries_for_node("node-2");
227        assert_eq!(entries[0].node_id, "node-2");
228    }
229
230    #[test]
231    fn query_all_merged_by_timestamp() {
232        let agg = LogAggregator::new("node-1".into());
233        agg.add_remote(make_entry("node-2", "info", "second", 2));
234        agg.add_remote(make_entry("node-3", "info", "first", 1));
235        agg.add_remote(make_entry("node-1", "info", "third", 3));
236
237        let results = agg.query(&LogQuery::default());
238        assert_eq!(results.len(), 3);
239        assert_eq!(results[0].message, "first");
240        assert_eq!(results[1].message, "second");
241        assert_eq!(results[2].message, "third");
242    }
243
244    #[test]
245    fn query_filter_by_node() {
246        let agg = LogAggregator::new("node-1".into());
247        agg.add_remote(make_entry("node-1", "info", "local", 0));
248        agg.add_remote(make_entry("node-2", "info", "remote", 0));
249
250        let results = agg.query(&LogQuery {
251            node_id: Some("node-2".into()),
252            ..Default::default()
253        });
254        assert_eq!(results.len(), 1);
255        assert_eq!(results[0].node_id, "node-2");
256    }
257
258    #[test]
259    fn query_filter_by_level() {
260        let agg = LogAggregator::new("node-1".into());
261        agg.add_remote(make_entry("node-1", "info", "info msg", 0));
262        agg.add_remote(make_entry("node-1", "error", "error msg", 0));
263
264        let results = agg.query(&LogQuery {
265            level: Some("error".into()),
266            ..Default::default()
267        });
268        assert_eq!(results.len(), 1);
269        assert_eq!(results[0].level, "error");
270    }
271
272    #[test]
273    fn query_with_limit() {
274        let agg = LogAggregator::new("node-1".into());
275        for i in 0..10 {
276            agg.add_remote(make_entry("node-1", "info", &format!("msg-{i}"), i));
277        }
278
279        let results = agg.query(&LogQuery {
280            limit: Some(3),
281            ..Default::default()
282        });
283        assert_eq!(results.len(), 3);
284    }
285
286    #[test]
287    fn query_filter_by_source() {
288        let agg = LogAggregator::new("n1".into());
289        agg.add_remote(RemoteLogEntry::new("n1", "info", "a", "boot", Utc::now()));
290        agg.add_remote(RemoteLogEntry::new("n1", "info", "b", "mesh", Utc::now()));
291
292        let results = agg.query(&LogQuery {
293            source: Some("mesh".into()),
294            ..Default::default()
295        });
296        assert_eq!(results.len(), 1);
297        assert_eq!(results[0].source, "mesh");
298    }
299
300    #[test]
301    fn add_remote_batch() {
302        let agg = LogAggregator::new("n1".into());
303        let batch = vec![
304            make_entry("n2", "info", "a", 0),
305            make_entry("n3", "info", "b", 0),
306        ];
307        agg.add_remote_batch(batch);
308        assert_eq!(agg.total_entries(), 2);
309    }
310
311    #[test]
312    fn known_nodes() {
313        let agg = LogAggregator::new("n1".into());
314        agg.log_local("info", "x", "y");
315        agg.add_remote(make_entry("n2", "info", "z", 0));
316        let mut nodes = agg.known_nodes();
317        nodes.sort();
318        assert_eq!(nodes, vec!["n1", "n2"]);
319    }
320
321    #[test]
322    fn empty_aggregator() {
323        let agg = LogAggregator::new("n1".into());
324        assert_eq!(agg.total_entries(), 0);
325        assert!(agg.query(&LogQuery::default()).is_empty());
326        assert!(agg.known_nodes().is_empty());
327    }
328
329    #[test]
330    fn remote_log_entry_serialization() {
331        let entry = RemoteLogEntry::new("n1", "info", "msg", "src", Utc::now());
332        let json = serde_json::to_string(&entry).unwrap();
333        let deser: RemoteLogEntry = serde_json::from_str(&json).unwrap();
334        assert_eq!(deser.node_id, "n1");
335        assert_eq!(deser.message, "msg");
336    }
337}
clawft_kernel/mesh_log.rs

clawft_kernel/
mesh_log.rs
