agentsight_capture/analyzers/
timestamp_normalizer.rs

1// SPDX-License-Identifier: MIT
2// Copyright (c) 2026 eunomia-bpf org.
3
4//! Timestamp Normalizer Analyzer
5//!
6//! Converts all timestamps from nanoseconds since boot to milliseconds since UNIX epoch.
7//! This ensures timestamps are standardized for frontend consumption.
8
9use super::Analyzer;
10use crate::event::Event;
11use crate::time::boot_ns_to_epoch_ms;
12use async_trait::async_trait;
13use futures::stream::{Stream, StreamExt};
14use std::pin::Pin;
15
16type EventStream = Pin<Box<dyn Stream<Item = Event> + Send>>;
17
18#[derive(Debug)]
19pub struct TimestampNormalizer {}
20
21impl TimestampNormalizer {
22    pub fn new() -> Self {
23        Self {}
24    }
25}
26
27impl Default for TimestampNormalizer {
28    fn default() -> Self {
29        Self::new()
30    }
31}
32
33#[async_trait]
34impl Analyzer for TimestampNormalizer {
35    async fn process(
36        &mut self,
37        stream: EventStream,
38    ) -> Result<EventStream, Box<dyn std::error::Error + Send + Sync>> {
39        let normalized_stream = stream.map(|mut event| {
40            // Convert timestamp from nanoseconds since boot to milliseconds since UNIX epoch
41            let timestamp_ms = boot_ns_to_epoch_ms(event.timestamp);
42            event.timestamp = timestamp_ms;
43            event
44        });
45
46        Ok(Box::pin(normalized_stream))
47    }
48}
49
50#[cfg(test)]
51mod tests {
52    use super::*;
53    use futures::stream;
54    use serde_json::json;
55
56    #[tokio::test]
57    async fn test_timestamp_normalizer() {
58        let mut normalizer = TimestampNormalizer::new();
59
60        // Create test event with nanoseconds since boot (e.g., 1 second after boot)
61        let test_event = Event::new_with_timestamp(
62            1_000_000_000, // 1 second in nanoseconds
63            "test".to_string(),
64            1234,
65            "test_comm".to_string(),
66            json!({"test": "data"}),
67        );
68
69        let input_stream = stream::iter(vec![test_event]);
70        let output_stream = normalizer.process(Box::pin(input_stream)).await.unwrap();
71
72        let results: Vec<Event> = output_stream.collect().await;
73        assert_eq!(results.len(), 1);
74
75        // Timestamp should be converted to milliseconds since epoch
76        // Should be much larger than 1 second (boot time + 1 second)
77        assert!(results[0].timestamp > 1_000_000_000_000); // Should be > year 2001 in ms
78    }
79
80    #[tokio::test]
81    async fn test_timestamp_normalizer_multiple_events() {
82        let mut normalizer = TimestampNormalizer::new();
83
84        let events = vec![
85            Event::new_with_timestamp(
86                1_000_000_000, // 1 second
87                "test".to_string(),
88                1234,
89                "test1".to_string(),
90                json!({"id": 1}),
91            ),
92            Event::new_with_timestamp(
93                2_000_000_000, // 2 seconds
94                "test".to_string(),
95                1234,
96                "test2".to_string(),
97                json!({"id": 2}),
98            ),
99            Event::new_with_timestamp(
100                3_000_000_000, // 3 seconds
101                "test".to_string(),
102                1234,
103                "test3".to_string(),
104                json!({"id": 3}),
105            ),
106        ];
107
108        let input_stream = stream::iter(events);
109        let output_stream = normalizer.process(Box::pin(input_stream)).await.unwrap();
110
111        let results: Vec<Event> = output_stream.collect().await;
112        assert_eq!(results.len(), 3);
113
114        // Verify timestamps are in order and normalized
115        assert!(results[0].timestamp < results[1].timestamp);
116        assert!(results[1].timestamp < results[2].timestamp);
117
118        // All should be in milliseconds since epoch (reasonable timestamp)
119        for result in &results {
120            assert!(result.timestamp > 1_000_000_000_000); // > year 2001
121        }
122    }
123}
agentsight_capture/analyzers/timestamp_normalizer.rs

agentsight_capture/analyzers/
timestamp_normalizer.rs
