tatara_core/domain/
traced_event.rs

1//! Structured events with correlation IDs for workload lifecycle tracing.
2//!
3//! Every phase transition emits a TracedEvent with a correlation_id that
4//! traces the workload from submission → placement → warming → executing
5//! → contraction → terminal. This enables Charity Majors-style
6//! observability: structured events > metrics > logs.
7//!
8//! Events flow through NATS → Vector → Loki/DataFusion for querying.
9
10use chrono::{DateTime, Utc};
11use serde::{Deserialize, Serialize};
12use std::collections::HashMap;
13use uuid::Uuid;
14
15/// A structured event carrying full context for observability.
16#[derive(Debug, Clone, Serialize, Deserialize)]
17pub struct TracedEvent {
18    /// Unique event ID.
19    pub id: Uuid,
20    /// When this event occurred.
21    pub timestamp: DateTime<Utc>,
22    /// Correlation ID — traces the workload through its entire lifecycle.
23    /// Same correlation_id from job submission through final terminal state.
24    pub correlation_id: Uuid,
25    /// Parent event ID (for causal ordering within a correlation).
26    pub parent_id: Option<Uuid>,
27    /// Event category.
28    pub category: EventCategory,
29    /// Event action.
30    pub action: String,
31    /// Structured context fields.
32    pub fields: HashMap<String, serde_json::Value>,
33    /// Which node emitted this event.
34    pub node_id: Option<u64>,
35    /// Duration of the operation (if applicable).
36    pub duration_ms: Option<u64>,
37    /// Severity level.
38    pub level: EventLevel,
39}
40
41/// Event categories for structured querying.
42#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
43#[serde(rename_all = "snake_case")]
44pub enum EventCategory {
45    /// Job lifecycle events.
46    Job,
47    /// Allocation lifecycle events.
48    Allocation,
49    /// Task lifecycle events.
50    Task,
51    /// Node lifecycle events.
52    Node,
53    /// Scheduling events.
54    Scheduling,
55    /// Networking events (mesh, policy, flow).
56    Network,
57    /// Health check events.
58    Health,
59    /// Secret/security events.
60    Security,
61    /// Build/cache events.
62    Build,
63}
64
65/// Event severity levels.
66#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
67#[serde(rename_all = "snake_case")]
68pub enum EventLevel {
69    #[default]
70    Info,
71    Warning,
72    Error,
73    Debug,
74}
75
76impl TracedEvent {
77    /// Create a new traced event.
78    pub fn new(correlation_id: Uuid, category: EventCategory, action: impl Into<String>) -> Self {
79        Self {
80            id: Uuid::new_v4(),
81            timestamp: Utc::now(),
82            correlation_id,
83            parent_id: None,
84            category,
85            action: action.into(),
86            fields: HashMap::new(),
87            node_id: None,
88            duration_ms: None,
89            level: EventLevel::Info,
90        }
91    }
92
93    /// Set parent event for causal ordering.
94    pub fn with_parent(mut self, parent_id: Uuid) -> Self {
95        self.parent_id = Some(parent_id);
96        self
97    }
98
99    /// Set the emitting node.
100    pub fn with_node(mut self, node_id: u64) -> Self {
101        self.node_id = Some(node_id);
102        self
103    }
104
105    /// Set operation duration.
106    pub fn with_duration(mut self, ms: u64) -> Self {
107        self.duration_ms = Some(ms);
108        self
109    }
110
111    /// Set severity level.
112    pub fn with_level(mut self, level: EventLevel) -> Self {
113        self.level = level;
114        self
115    }
116
117    /// Add a structured field.
118    pub fn field(mut self, key: impl Into<String>, value: impl Into<serde_json::Value>) -> Self {
119        self.fields.insert(key.into(), value.into());
120        self
121    }
122
123    /// Convenience: create an allocation phase transition event.
124    pub fn allocation_phase(
125        correlation_id: Uuid,
126        alloc_id: Uuid,
127        from_phase: &str,
128        to_phase: &str,
129    ) -> Self {
130        Self::new(
131            correlation_id,
132            EventCategory::Allocation,
133            "phase_transition",
134        )
135        .field("alloc_id", alloc_id.to_string())
136        .field("from_phase", from_phase)
137        .field("to_phase", to_phase)
138    }
139
140    /// Convenience: create a scheduling decision event.
141    pub fn scheduling_decision(
142        correlation_id: Uuid,
143        job_id: &str,
144        node_id: u64,
145        driver: &str,
146    ) -> Self {
147        Self::new(
148            correlation_id,
149            EventCategory::Scheduling,
150            "allocation_placed",
151        )
152        .field("job_id", job_id)
153        .field("node_id", serde_json::Value::Number(node_id.into()))
154        .field("driver", driver)
155    }
156
157    /// Convenience: create a health check event.
158    pub fn health_check(
159        correlation_id: Uuid,
160        service_name: &str,
161        healthy: bool,
162        latency_ms: u64,
163    ) -> Self {
164        Self::new(correlation_id, EventCategory::Health, "probe_result")
165            .field("service_name", service_name)
166            .field("healthy", healthy)
167            .with_duration(latency_ms)
168    }
169}
170
171/// A correlation context that tracks a workload through its lifecycle.
172///
173/// # Thread Safety
174/// This type is NOT thread-safe. It uses interior mutation via `&mut self`
175/// in `emit()`. Each workload should own a single CorrelationContext.
176/// For concurrent access, wrap in `Arc<Mutex<CorrelationContext>>`.
177#[derive(Debug, Clone, Serialize, Deserialize)]
178pub struct CorrelationContext {
179    /// The correlation ID for this workload.
180    pub correlation_id: Uuid,
181    /// Job ID.
182    pub job_id: String,
183    /// Events emitted so far (for causal chain).
184    pub event_count: u64,
185    /// Last event ID (for parent linking).
186    pub last_event_id: Option<Uuid>,
187}
188
189impl CorrelationContext {
190    pub fn new(job_id: &str) -> Self {
191        Self {
192            correlation_id: Uuid::new_v4(),
193            job_id: job_id.to_string(),
194            event_count: 0,
195            last_event_id: None,
196        }
197    }
198
199    /// Create a new event in this correlation context.
200    pub fn emit(&mut self, category: EventCategory, action: impl Into<String>) -> TracedEvent {
201        let event = TracedEvent::new(self.correlation_id, category, action);
202        let event = if let Some(parent) = self.last_event_id {
203            event.with_parent(parent)
204        } else {
205            event
206        };
207        self.last_event_id = Some(event.id);
208        self.event_count += 1;
209        event
210    }
211}
212
213#[cfg(test)]
214mod tests {
215    use super::*;
216
217    #[test]
218    fn test_traced_event_creation() {
219        let corr_id = Uuid::new_v4();
220        let event = TracedEvent::new(corr_id, EventCategory::Job, "submitted")
221            .field("job_id", "web-service")
222            .with_node(42);
223
224        assert_eq!(event.correlation_id, corr_id);
225        assert_eq!(event.category, EventCategory::Job);
226        assert_eq!(event.action, "submitted");
227        assert_eq!(event.node_id, Some(42));
228        assert_eq!(event.fields["job_id"], "web-service");
229    }
230
231    #[test]
232    fn test_correlation_context() {
233        let mut ctx = CorrelationContext::new("web-service");
234
235        let e1 = ctx.emit(EventCategory::Job, "submitted");
236        assert!(e1.parent_id.is_none());
237        assert_eq!(ctx.event_count, 1);
238
239        let e2 = ctx.emit(EventCategory::Scheduling, "placed");
240        assert_eq!(e2.parent_id, Some(e1.id));
241        assert_eq!(ctx.event_count, 2);
242
243        let e3 = ctx.emit(EventCategory::Allocation, "warming");
244        assert_eq!(e3.parent_id, Some(e2.id));
245        assert_eq!(e3.correlation_id, e1.correlation_id);
246    }
247
248    #[test]
249    fn test_convenience_constructors() {
250        let corr = Uuid::new_v4();
251        let alloc = Uuid::new_v4();
252
253        let phase = TracedEvent::allocation_phase(corr, alloc, "warming", "executing");
254        assert_eq!(phase.fields["from_phase"], "warming");
255        assert_eq!(phase.fields["to_phase"], "executing");
256
257        let sched = TracedEvent::scheduling_decision(corr, "web", 42, "wasi");
258        assert_eq!(sched.fields["driver"], "wasi");
259
260        let health = TracedEvent::health_check(corr, "web", true, 5);
261        assert_eq!(health.duration_ms, Some(5));
262    }
263
264    #[test]
265    fn test_serde_roundtrip() {
266        let event = TracedEvent::new(Uuid::new_v4(), EventCategory::Network, "flow_detected")
267            .field("bytes", 1024)
268            .with_level(EventLevel::Debug);
269
270        let json = serde_json::to_string(&event).unwrap();
271        let back: TracedEvent = serde_json::from_str(&json).unwrap();
272        assert_eq!(back.category, EventCategory::Network);
273        assert_eq!(back.level, EventLevel::Debug);
274    }
275}
tatara_core/domain/traced_event.rs

tatara_core/domain/
traced_event.rs
