tatara_core/domain/
event.rs

1use chrono::{DateTime, Utc};
2use serde::{Deserialize, Serialize};
3use std::collections::VecDeque;
4use uuid::Uuid;
5
6/// An event representing a state change in the cluster.
7#[derive(Debug, Clone, Serialize, Deserialize)]
8pub struct Event {
9    pub id: Uuid,
10    pub timestamp: DateTime<Utc>,
11    pub kind: EventKind,
12    pub payload: serde_json::Value,
13}
14
15/// Categories of events emitted by the cluster.
16#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
17#[serde(rename_all = "snake_case")]
18pub enum EventKind {
19    JobSubmitted,
20    JobUpdated,
21    JobStopped,
22    AllocationPlaced,
23    AllocationStarted,
24    AllocationFailed,
25    AllocationCompleted,
26    NodeJoined,
27    NodeLeft,
28    NodeDraining,
29    NodeReady,
30    EvaluationCompleted,
31    DeploymentStarted,
32    DeploymentCompleted,
33    AllocationRestarted,
34    AllocationLost,
35    AllocationRescheduled,
36    ReconcileCompleted,
37    SpecDriftDetected,
38    RollingUpdateStarted,
39    RollingUpdateCompleted,
40    SourceCreated,
41    SourceReconciled,
42    SourceFailed,
43    SourceSuspended,
44    SourceResumed,
45    SourceJobCreated,
46    SourceJobUpdated,
47    SourceJobRemoved,
48}
49
50impl std::fmt::Display for EventKind {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        match self {
53            Self::JobSubmitted => write!(f, "job_submitted"),
54            Self::JobUpdated => write!(f, "job_updated"),
55            Self::JobStopped => write!(f, "job_stopped"),
56            Self::AllocationPlaced => write!(f, "allocation_placed"),
57            Self::AllocationStarted => write!(f, "allocation_started"),
58            Self::AllocationFailed => write!(f, "allocation_failed"),
59            Self::AllocationCompleted => write!(f, "allocation_completed"),
60            Self::NodeJoined => write!(f, "node_joined"),
61            Self::NodeLeft => write!(f, "node_left"),
62            Self::NodeDraining => write!(f, "node_draining"),
63            Self::NodeReady => write!(f, "node_ready"),
64            Self::EvaluationCompleted => write!(f, "evaluation_completed"),
65            Self::DeploymentStarted => write!(f, "deployment_started"),
66            Self::DeploymentCompleted => write!(f, "deployment_completed"),
67            Self::AllocationRestarted => write!(f, "allocation_restarted"),
68            Self::AllocationLost => write!(f, "allocation_lost"),
69            Self::AllocationRescheduled => write!(f, "allocation_rescheduled"),
70            Self::ReconcileCompleted => write!(f, "reconcile_completed"),
71            Self::SpecDriftDetected => write!(f, "spec_drift_detected"),
72            Self::RollingUpdateStarted => write!(f, "rolling_update_started"),
73            Self::RollingUpdateCompleted => write!(f, "rolling_update_completed"),
74            Self::SourceCreated => write!(f, "source_created"),
75            Self::SourceReconciled => write!(f, "source_reconciled"),
76            Self::SourceFailed => write!(f, "source_failed"),
77            Self::SourceSuspended => write!(f, "source_suspended"),
78            Self::SourceResumed => write!(f, "source_resumed"),
79            Self::SourceJobCreated => write!(f, "source_job_created"),
80            Self::SourceJobUpdated => write!(f, "source_job_updated"),
81            Self::SourceJobRemoved => write!(f, "source_job_removed"),
82        }
83    }
84}
85
86impl EventKind {
87    pub fn from_str_opt(s: &str) -> Option<Self> {
88        match s {
89            "job_submitted" => Some(Self::JobSubmitted),
90            "job_updated" => Some(Self::JobUpdated),
91            "job_stopped" => Some(Self::JobStopped),
92            "allocation_placed" => Some(Self::AllocationPlaced),
93            "allocation_started" => Some(Self::AllocationStarted),
94            "allocation_failed" => Some(Self::AllocationFailed),
95            "allocation_completed" => Some(Self::AllocationCompleted),
96            "node_joined" => Some(Self::NodeJoined),
97            "node_left" => Some(Self::NodeLeft),
98            "node_draining" => Some(Self::NodeDraining),
99            "node_ready" => Some(Self::NodeReady),
100            "evaluation_completed" => Some(Self::EvaluationCompleted),
101            "deployment_started" => Some(Self::DeploymentStarted),
102            "deployment_completed" => Some(Self::DeploymentCompleted),
103            "allocation_restarted" => Some(Self::AllocationRestarted),
104            "allocation_lost" => Some(Self::AllocationLost),
105            "allocation_rescheduled" => Some(Self::AllocationRescheduled),
106            "reconcile_completed" => Some(Self::ReconcileCompleted),
107            "spec_drift_detected" => Some(Self::SpecDriftDetected),
108            "rolling_update_started" => Some(Self::RollingUpdateStarted),
109            "rolling_update_completed" => Some(Self::RollingUpdateCompleted),
110            "source_created" => Some(Self::SourceCreated),
111            "source_reconciled" => Some(Self::SourceReconciled),
112            "source_failed" => Some(Self::SourceFailed),
113            "source_suspended" => Some(Self::SourceSuspended),
114            "source_resumed" => Some(Self::SourceResumed),
115            "source_job_created" => Some(Self::SourceJobCreated),
116            "source_job_updated" => Some(Self::SourceJobUpdated),
117            "source_job_removed" => Some(Self::SourceJobRemoved),
118            _ => None,
119        }
120    }
121}
122
123impl Event {
124    pub fn new(kind: EventKind, payload: serde_json::Value) -> Self {
125        Self {
126            id: Uuid::new_v4(),
127            timestamp: Utc::now(),
128            kind,
129            payload,
130        }
131    }
132}
133
134/// Ring buffer for events with a configurable capacity.
135#[derive(Debug, Clone, Serialize, Deserialize)]
136pub struct EventRing {
137    events: VecDeque<Event>,
138    capacity: usize,
139}
140
141impl Default for EventRing {
142    fn default() -> Self {
143        Self {
144            events: VecDeque::new(),
145            capacity: 10_000,
146        }
147    }
148}
149
150impl EventRing {
151    pub fn with_capacity(capacity: usize) -> Self {
152        Self {
153            events: VecDeque::with_capacity(capacity),
154            capacity,
155        }
156    }
157
158    pub fn push(&mut self, event: Event) {
159        if self.events.len() >= self.capacity {
160            self.events.pop_front();
161        }
162        self.events.push_back(event);
163    }
164
165    pub fn list(&self) -> &VecDeque<Event> {
166        &self.events
167    }
168
169    /// List events filtered by kind and/or since timestamp.
170    pub fn query(&self, kind: Option<&EventKind>, since: Option<DateTime<Utc>>) -> Vec<&Event> {
171        self.events
172            .iter()
173            .filter(|e| {
174                kind.map_or(true, |k| &e.kind == k) && since.map_or(true, |s| e.timestamp >= s)
175            })
176            .collect()
177    }
178
179    pub fn len(&self) -> usize {
180        self.events.len()
181    }
182
183    pub fn is_empty(&self) -> bool {
184        self.events.is_empty()
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191
192    #[test]
193    fn test_ring_buffer_capacity() {
194        let mut ring = EventRing::with_capacity(3);
195        for i in 0..5 {
196            ring.push(Event::new(
197                EventKind::JobSubmitted,
198                serde_json::json!({ "i": i }),
199            ));
200        }
201        assert_eq!(ring.len(), 3);
202        // Oldest events should be evicted
203        let events: Vec<_> = ring.list().iter().collect();
204        assert_eq!(events[0].payload["i"], 2);
205        assert_eq!(events[1].payload["i"], 3);
206        assert_eq!(events[2].payload["i"], 4);
207    }
208
209    #[test]
210    fn test_query_by_kind() {
211        let mut ring = EventRing::default();
212        ring.push(Event::new(EventKind::JobSubmitted, serde_json::json!({})));
213        ring.push(Event::new(EventKind::NodeJoined, serde_json::json!({})));
214        ring.push(Event::new(EventKind::JobSubmitted, serde_json::json!({})));
215
216        let filtered = ring.query(Some(&EventKind::JobSubmitted), None);
217        assert_eq!(filtered.len(), 2);
218    }
219}
tatara_core/domain/event.rs

tatara_core/domain/
event.rs
