axon/
request_log.rs

1//! Request Logger — structured audit log for AxonServer API requests.
2//!
3//! Records each API request as a structured `RequestLogEntry` with:
4//!   - method, path, status code, latency
5//!   - client key (from auth token or "anonymous")
6//!   - timestamp (monotonic + wall clock)
7//!
8//! The log is an in-memory ring buffer with configurable capacity.
9//! Entries can be queried by path, status, or time range.
10//!
11//! Endpoints:
12//!   - `GET /v1/logs` — query recent request logs
13//!   - `GET /v1/logs/stats` — aggregate request statistics
14
15use std::collections::VecDeque;
16use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
17
18use serde::Serialize;
19
20// ── Entry ────────────────────────────────────────────────────────────────
21
22/// A single request log entry.
23#[derive(Debug, Clone, Serialize)]
24pub struct RequestLogEntry {
25    /// Wall-clock timestamp (Unix seconds).
26    pub timestamp: u64,
27    /// HTTP method (GET, POST, DELETE).
28    pub method: String,
29    /// Request path (e.g., "/v1/deploy").
30    pub path: String,
31    /// HTTP status code.
32    pub status: u16,
33    /// Request latency in microseconds.
34    pub latency_us: u64,
35    /// Client identifier (auth token or "anonymous").
36    pub client_key: String,
37}
38
39// ── Config ───────────────────────────────────────────────────────────────
40
41/// Request logger configuration.
42#[derive(Debug, Clone)]
43pub struct RequestLogConfig {
44    /// Maximum entries in the ring buffer.
45    pub capacity: usize,
46    /// Whether logging is enabled.
47    pub enabled: bool,
48}
49
50impl RequestLogConfig {
51    /// Default: 1000 entries, enabled.
52    pub fn default_config() -> Self {
53        RequestLogConfig {
54            capacity: 1000,
55            enabled: true,
56        }
57    }
58
59    /// Disabled logger.
60    pub fn disabled() -> Self {
61        RequestLogConfig {
62            capacity: 0,
63            enabled: false,
64        }
65    }
66}
67
68// ── Logger ───────────────────────────────────────────────────────────────
69
70/// In-memory ring buffer request logger.
71pub struct RequestLogger {
72    config: RequestLogConfig,
73    entries: VecDeque<RequestLogEntry>,
74    started_at: Instant,
75    total_requests: u64,
76    total_errors: u64,
77}
78
79impl RequestLogger {
80    /// Create a new request logger.
81    pub fn new(config: RequestLogConfig) -> Self {
82        RequestLogger {
83            entries: VecDeque::with_capacity(config.capacity.min(1024)),
84            config,
85            started_at: Instant::now(),
86            total_requests: 0,
87            total_errors: 0,
88        }
89    }
90
91    /// Record a request.
92    pub fn record(&mut self, method: &str, path: &str, status: u16, latency: Duration, client_key: &str) {
93        if !self.config.enabled {
94            return;
95        }
96
97        self.total_requests += 1;
98        if status >= 400 {
99            self.total_errors += 1;
100        }
101
102        let entry = RequestLogEntry {
103            timestamp: wall_clock_secs(),
104            method: method.to_string(),
105            path: path.to_string(),
106            status,
107            latency_us: latency.as_micros() as u64,
108            client_key: client_key.to_string(),
109        };
110
111        if self.entries.len() >= self.config.capacity && self.config.capacity > 0 {
112            self.entries.pop_front();
113        }
114        if self.config.capacity > 0 {
115            self.entries.push_back(entry);
116        }
117    }
118
119    /// Get the configuration.
120    pub fn config(&self) -> &RequestLogConfig {
121        &self.config
122    }
123
124    /// Update the configuration at runtime.
125    pub fn update_config(&mut self, capacity: Option<usize>, enabled: Option<bool>) {
126        if let Some(cap) = capacity {
127            self.config.capacity = cap;
128            // Trim entries if new capacity is smaller
129            while self.entries.len() > cap {
130                self.entries.pop_front();
131            }
132        }
133        if let Some(en) = enabled {
134            self.config.enabled = en;
135        }
136    }
137
138    /// Get recent entries (newest first), optionally filtered.
139    pub fn recent(&self, limit: usize, filter: Option<&LogFilter>) -> Vec<&RequestLogEntry> {
140        let result: Vec<&RequestLogEntry> = self.entries.iter().rev()
141            .filter(|e| match filter {
142                Some(f) => f.matches(e),
143                None => true,
144            })
145            .take(limit)
146            .collect();
147        // Already newest-first from rev()
148        result
149    }
150
151    /// Compute aggregate statistics.
152    pub fn stats(&self) -> LogStats {
153        let entries: Vec<&RequestLogEntry> = self.entries.iter().collect();
154
155        let mut path_counts: std::collections::HashMap<String, u64> = std::collections::HashMap::new();
156        let mut status_counts: std::collections::HashMap<u16, u64> = std::collections::HashMap::new();
157        let mut total_latency_us: u64 = 0;
158        let mut max_latency_us: u64 = 0;
159
160        for e in &entries {
161            *path_counts.entry(e.path.clone()).or_insert(0) += 1;
162            *status_counts.entry(e.status).or_insert(0) += 1;
163            total_latency_us += e.latency_us;
164            if e.latency_us > max_latency_us {
165                max_latency_us = e.latency_us;
166            }
167        }
168
169        let avg_latency_us = if entries.is_empty() {
170            0
171        } else {
172            total_latency_us / entries.len() as u64
173        };
174
175        let mut top_paths: Vec<(String, u64)> = path_counts.into_iter().collect();
176        top_paths.sort_by(|a, b| b.1.cmp(&a.1));
177        top_paths.truncate(10);
178
179        let mut status_breakdown: Vec<(u16, u64)> = status_counts.into_iter().collect();
180        status_breakdown.sort_by_key(|(k, _)| *k);
181
182        LogStats {
183            total_requests: self.total_requests,
184            total_errors: self.total_errors,
185            buffered_entries: self.entries.len(),
186            avg_latency_us,
187            max_latency_us,
188            top_paths,
189            status_breakdown,
190            uptime_secs: self.started_at.elapsed().as_secs(),
191        }
192    }
193
194    /// Number of buffered entries.
195    pub fn len(&self) -> usize {
196        self.entries.len()
197    }
198
199    /// Whether the buffer is empty.
200    pub fn is_empty(&self) -> bool {
201        self.entries.is_empty()
202    }
203
204    /// Total requests recorded (including evicted).
205    pub fn total_requests(&self) -> u64 {
206        self.total_requests
207    }
208
209    /// Clear all entries.
210    pub fn clear(&mut self) {
211        self.entries.clear();
212    }
213}
214
215// ── Filter ───────────────────────────────────────────────────────────────
216
217/// Filter for querying log entries.
218#[derive(Debug, Clone, Default, Serialize)]
219pub struct LogFilter {
220    /// Filter by path prefix (e.g., "/v1/deploy").
221    pub path_prefix: Option<String>,
222    /// Filter by minimum status code.
223    pub min_status: Option<u16>,
224    /// Filter by maximum status code.
225    pub max_status: Option<u16>,
226    /// Filter by client key.
227    pub client_key: Option<String>,
228}
229
230impl LogFilter {
231    /// Check if an entry matches this filter.
232    pub fn matches(&self, entry: &RequestLogEntry) -> bool {
233        if let Some(ref prefix) = self.path_prefix {
234            if !entry.path.starts_with(prefix) {
235                return false;
236            }
237        }
238        if let Some(min) = self.min_status {
239            if entry.status < min {
240                return false;
241            }
242        }
243        if let Some(max) = self.max_status {
244            if entry.status > max {
245                return false;
246            }
247        }
248        if let Some(ref key) = self.client_key {
249            if entry.client_key != *key {
250                return false;
251            }
252        }
253        true
254    }
255}
256
257// ── Stats ────────────────────────────────────────────────────────────────
258
259/// Aggregate request statistics.
260#[derive(Debug, Clone, Serialize)]
261pub struct LogStats {
262    pub total_requests: u64,
263    pub total_errors: u64,
264    pub buffered_entries: usize,
265    pub avg_latency_us: u64,
266    pub max_latency_us: u64,
267    pub top_paths: Vec<(String, u64)>,
268    pub status_breakdown: Vec<(u16, u64)>,
269    pub uptime_secs: u64,
270}
271
272// ── Helpers ──────────────────────────────────────────────────────────────
273
274fn wall_clock_secs() -> u64 {
275    SystemTime::now()
276        .duration_since(UNIX_EPOCH)
277        .unwrap_or_default()
278        .as_secs()
279}
280
281// ── Tests ────────────────────────────────────────────────────────────────
282
283#[cfg(test)]
284mod tests {
285    use super::*;
286
287    #[test]
288    fn record_and_retrieve() {
289        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
290        logger.record("GET", "/v1/health", 200, Duration::from_micros(500), "anon");
291        logger.record("POST", "/v1/deploy", 200, Duration::from_micros(1500), "token_a");
292
293        assert_eq!(logger.len(), 2);
294        assert_eq!(logger.total_requests(), 2);
295
296        let recent = logger.recent(10, None);
297        assert_eq!(recent.len(), 2);
298        // Newest first
299        assert_eq!(recent[0].path, "/v1/deploy");
300        assert_eq!(recent[1].path, "/v1/health");
301    }
302
303    #[test]
304    fn ring_buffer_eviction() {
305        let config = RequestLogConfig { capacity: 3, enabled: true };
306        let mut logger = RequestLogger::new(config);
307
308        for i in 0..5 {
309            logger.record("GET", &format!("/v1/req{}", i), 200, Duration::from_micros(100), "c");
310        }
311
312        assert_eq!(logger.len(), 3);
313        assert_eq!(logger.total_requests(), 5);
314
315        let recent = logger.recent(10, None);
316        assert_eq!(recent[0].path, "/v1/req4");
317        assert_eq!(recent[2].path, "/v1/req2");
318    }
319
320    #[test]
321    fn disabled_logger_no_recording() {
322        let mut logger = RequestLogger::new(RequestLogConfig::disabled());
323        logger.record("GET", "/v1/health", 200, Duration::from_micros(100), "c");
324        assert_eq!(logger.len(), 0);
325        assert_eq!(logger.total_requests(), 0);
326    }
327
328    #[test]
329    fn error_counting() {
330        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
331        logger.record("GET", "/v1/a", 200, Duration::from_micros(100), "c");
332        logger.record("GET", "/v1/b", 401, Duration::from_micros(100), "c");
333        logger.record("GET", "/v1/c", 500, Duration::from_micros(100), "c");
334        logger.record("GET", "/v1/d", 429, Duration::from_micros(100), "c");
335
336        let stats = logger.stats();
337        assert_eq!(stats.total_requests, 4);
338        assert_eq!(stats.total_errors, 3); // 401, 500, 429
339    }
340
341    #[test]
342    fn filter_by_path_prefix() {
343        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
344        logger.record("GET", "/v1/health", 200, Duration::from_micros(100), "c");
345        logger.record("POST", "/v1/deploy", 200, Duration::from_micros(100), "c");
346        logger.record("GET", "/v1/health/live", 200, Duration::from_micros(100), "c");
347
348        let filter = LogFilter { path_prefix: Some("/v1/health".into()), ..Default::default() };
349        let result = logger.recent(10, Some(&filter));
350        assert_eq!(result.len(), 2);
351    }
352
353    #[test]
354    fn filter_by_status_range() {
355        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
356        logger.record("GET", "/v1/a", 200, Duration::from_micros(100), "c");
357        logger.record("GET", "/v1/b", 401, Duration::from_micros(100), "c");
358        logger.record("GET", "/v1/c", 500, Duration::from_micros(100), "c");
359
360        let filter = LogFilter { min_status: Some(400), ..Default::default() };
361        let result = logger.recent(10, Some(&filter));
362        assert_eq!(result.len(), 2);
363    }
364
365    #[test]
366    fn filter_by_client_key() {
367        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
368        logger.record("GET", "/v1/a", 200, Duration::from_micros(100), "alice");
369        logger.record("GET", "/v1/b", 200, Duration::from_micros(100), "bob");
370        logger.record("GET", "/v1/c", 200, Duration::from_micros(100), "alice");
371
372        let filter = LogFilter { client_key: Some("alice".into()), ..Default::default() };
373        let result = logger.recent(10, Some(&filter));
374        assert_eq!(result.len(), 2);
375    }
376
377    #[test]
378    fn stats_computation() {
379        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
380        logger.record("GET", "/v1/health", 200, Duration::from_micros(100), "c");
381        logger.record("GET", "/v1/health", 200, Duration::from_micros(300), "c");
382        logger.record("POST", "/v1/deploy", 200, Duration::from_micros(500), "c");
383
384        let stats = logger.stats();
385        assert_eq!(stats.total_requests, 3);
386        assert_eq!(stats.total_errors, 0);
387        assert_eq!(stats.buffered_entries, 3);
388        assert_eq!(stats.avg_latency_us, 300); // (100+300+500)/3
389        assert_eq!(stats.max_latency_us, 500);
390        assert_eq!(stats.top_paths[0].0, "/v1/health");
391        assert_eq!(stats.top_paths[0].1, 2);
392    }
393
394    #[test]
395    fn stats_empty_logger() {
396        let logger = RequestLogger::new(RequestLogConfig::default_config());
397        let stats = logger.stats();
398        assert_eq!(stats.total_requests, 0);
399        assert_eq!(stats.avg_latency_us, 0);
400        assert_eq!(stats.max_latency_us, 0);
401    }
402
403    #[test]
404    fn clear_entries() {
405        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
406        logger.record("GET", "/v1/a", 200, Duration::from_micros(100), "c");
407        logger.record("GET", "/v1/b", 200, Duration::from_micros(100), "c");
408        assert_eq!(logger.len(), 2);
409
410        logger.clear();
411        assert_eq!(logger.len(), 0);
412        assert!(logger.is_empty());
413        // total_requests preserved
414        assert_eq!(logger.total_requests(), 2);
415    }
416
417    #[test]
418    fn entry_serializes_to_json() {
419        let entry = RequestLogEntry {
420            timestamp: 1700000000,
421            method: "POST".into(),
422            path: "/v1/deploy".into(),
423            status: 200,
424            latency_us: 1500,
425            client_key: "token_abc".into(),
426        };
427        let json = serde_json::to_string(&entry).unwrap();
428        assert!(json.contains("\"method\":\"POST\""));
429        assert!(json.contains("\"path\":\"/v1/deploy\""));
430        assert!(json.contains("\"status\":200"));
431        assert!(json.contains("\"latency_us\":1500"));
432    }
433
434    #[test]
435    fn stats_serializes_to_json() {
436        let stats = LogStats {
437            total_requests: 100,
438            total_errors: 5,
439            buffered_entries: 50,
440            avg_latency_us: 250,
441            max_latency_us: 5000,
442            top_paths: vec![("/v1/health".into(), 40)],
443            status_breakdown: vec![(200, 95), (500, 5)],
444            uptime_secs: 3600,
445        };
446        let json = serde_json::to_string(&stats).unwrap();
447        assert!(json.contains("\"total_requests\":100"));
448        assert!(json.contains("\"total_errors\":5"));
449    }
450
451    #[test]
452    fn recent_with_limit() {
453        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
454        for i in 0..10 {
455            logger.record("GET", &format!("/v1/r{}", i), 200, Duration::from_micros(100), "c");
456        }
457        let recent = logger.recent(3, None);
458        assert_eq!(recent.len(), 3);
459        assert_eq!(recent[0].path, "/v1/r9");
460    }
461
462    #[test]
463    fn combined_filter() {
464        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
465        logger.record("POST", "/v1/deploy", 200, Duration::from_micros(100), "alice");
466        logger.record("POST", "/v1/deploy", 500, Duration::from_micros(100), "alice");
467        logger.record("POST", "/v1/deploy", 200, Duration::from_micros(100), "bob");
468        logger.record("GET", "/v1/health", 200, Duration::from_micros(100), "alice");
469
470        let filter = LogFilter {
471            path_prefix: Some("/v1/deploy".into()),
472            client_key: Some("alice".into()),
473            min_status: Some(200),
474            max_status: Some(299),
475        };
476        let result = logger.recent(10, Some(&filter));
477        assert_eq!(result.len(), 1);
478        assert_eq!(result[0].status, 200);
479    }
480
481    #[test]
482    fn default_config_values() {
483        let cfg = RequestLogConfig::default_config();
484        assert_eq!(cfg.capacity, 1000);
485        assert!(cfg.enabled);
486    }
487
488    #[test]
489    fn timestamp_is_recent() {
490        let mut logger = RequestLogger::new(RequestLogConfig::default_config());
491        logger.record("GET", "/v1/a", 200, Duration::from_micros(100), "c");
492        let recent = logger.recent(1, None);
493        assert!(recent[0].timestamp > 1700000000);
494    }
495}
axon/request_log.rs

axon/
request_log.rs
