1use std::collections::BTreeMap;
17
18use chrono::Utc;
19
20use serde::{Deserialize, Serialize};
21use serde_json::{Number, Value as JsonValue};
22use thiserror::Error;
23
24use crate::{
25 ast::{EventsQuery, InspectQuery, SingularTargetKind, TimeSelector},
26 docker::{Container, DockerEvent, Image, MetricSample, Network, Volume},
27 events::{self, MockEventSource},
28 executor::{ExecutionResult, Row},
29 json_string,
30};
31
32#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
33pub struct TelemetrySnapshot {
34 pub timestamp: String,
35 pub containers: Vec<Container>,
36 pub images: Vec<Image>,
37 pub networks: Vec<Network>,
38 pub volumes: Vec<Volume>,
39}
40
41#[derive(Debug, Error)]
42pub enum TelemetryError {
43 #[error("historical query requires a telemetry store")]
44 MissingStore,
45 #[error("no snapshot found at or before `{timestamp}`")]
46 SnapshotNotFound { timestamp: String },
47 #[error("historical target not found: {target}")]
48 TargetNotFound { target: String },
49 #[error("{0}")]
50 Events(#[from] events::EventsError),
51 #[error("storage error: {0}")]
52 Storage(String),
53}
54
55pub trait TelemetryStore {
56 fn write_metric(&mut self, sample: MetricSample) -> Result<(), TelemetryError>;
57 fn latest_metrics(&self) -> Result<Vec<MetricSample>, TelemetryError>;
58 fn metrics_between(&self, from: &str, to: &str) -> Result<Vec<MetricSample>, TelemetryError>;
59 fn write_event(&mut self, event: DockerEvent) -> Result<(), TelemetryError>;
60 fn events_between(&self, from: &str, to: &str) -> Result<Vec<DockerEvent>, TelemetryError>;
61 fn write_snapshot(&mut self, snapshot: TelemetrySnapshot) -> Result<(), TelemetryError>;
62 fn snapshot_at_or_before(
63 &self,
64 timestamp: &str,
65 ) -> Result<Option<TelemetrySnapshot>, TelemetryError>;
66 fn all_snapshots(&self) -> Result<Vec<TelemetrySnapshot>, TelemetryError>;
67 fn write_alert_event(&mut self, event: AlertHistoryEvent) -> Result<(), TelemetryError>;
68 fn alert_history(&self, from: &str, to: &str)
69 -> Result<Vec<AlertHistoryEvent>, TelemetryError>;
70}
71
72#[derive(Debug, Clone, Serialize, Deserialize)]
73pub struct AlertHistoryEvent {
74 pub timestamp: String,
75 pub container_id: String,
76 pub container_name: String,
77 pub rule_condition: String,
78 pub action_type: String,
79 pub action_detail: String,
80 pub success: bool,
81}
82
83#[derive(Debug, Clone, Default)]
84pub struct InMemoryTelemetryStore {
85 metrics: Vec<MetricSample>,
86 events: Vec<DockerEvent>,
87 snapshots: Vec<TelemetrySnapshot>,
88}
89
90impl TelemetryStore for InMemoryTelemetryStore {
91 fn write_metric(&mut self, sample: MetricSample) -> Result<(), TelemetryError> {
92 self.metrics.push(sample);
93 Ok(())
94 }
95
96 fn latest_metrics(&self) -> Result<Vec<MetricSample>, TelemetryError> {
97 Ok(self.metrics.clone())
98 }
99
100 fn metrics_between(&self, from: &str, to: &str) -> Result<Vec<MetricSample>, TelemetryError> {
101 Ok(self
102 .metrics
103 .iter()
104 .filter(|sample| sample.timestamp.as_str() >= from && sample.timestamp.as_str() <= to)
105 .cloned()
106 .collect())
107 }
108
109 fn write_event(&mut self, event: DockerEvent) -> Result<(), TelemetryError> {
110 self.events.push(event);
111 Ok(())
112 }
113
114 fn events_between(&self, from: &str, to: &str) -> Result<Vec<DockerEvent>, TelemetryError> {
115 Ok(self
116 .events
117 .iter()
118 .filter(|event| event.time.as_str() >= from && event.time.as_str() <= to)
119 .cloned()
120 .collect())
121 }
122
123 fn write_snapshot(&mut self, snapshot: TelemetrySnapshot) -> Result<(), TelemetryError> {
124 self.snapshots.push(snapshot);
125 self.snapshots
126 .sort_by(|left, right| left.timestamp.cmp(&right.timestamp));
127 Ok(())
128 }
129
130 fn snapshot_at_or_before(
131 &self,
132 timestamp: &str,
133 ) -> Result<Option<TelemetrySnapshot>, TelemetryError> {
134 Ok(self
135 .snapshots
136 .iter()
137 .filter(|snapshot| snapshot.timestamp.as_str() <= timestamp)
138 .max_by(|left, right| left.timestamp.cmp(&right.timestamp))
139 .cloned())
140 }
141
142 fn all_snapshots(&self) -> Result<Vec<TelemetrySnapshot>, TelemetryError> {
143 Ok(self.snapshots.clone())
144 }
145
146 fn write_alert_event(&mut self, event: AlertHistoryEvent) -> Result<(), TelemetryError> {
147 let _ = event;
149 Ok(())
150 }
151
152 fn alert_history(
153 &self,
154 _from: &str,
155 _to: &str,
156 ) -> Result<Vec<AlertHistoryEvent>, TelemetryError> {
157 Ok(Vec::new())
158 }
159}
160
161pub fn inspect_at<S>(query: &InspectQuery, store: &S) -> Result<ExecutionResult, TelemetryError>
162where
163 S: TelemetryStore + ?Sized,
164{
165 let timestamp = query.at.as_deref().ok_or(TelemetryError::MissingStore)?;
166 let snapshot = store.snapshot_at_or_before(timestamp)?.ok_or_else(|| {
167 TelemetryError::SnapshotNotFound {
168 timestamp: timestamp.to_owned(),
169 }
170 })?;
171
172 let row = match query.target.kind {
173 SingularTargetKind::Container => snapshot
174 .containers
175 .into_iter()
176 .find(|container| {
177 container.id == query.target.value || container.name == query.target.value
178 })
179 .map(|container| container_snapshot_row(timestamp, container)),
180 SingularTargetKind::Image => snapshot
181 .images
182 .into_iter()
183 .find(|image| {
184 image.id == query.target.value
185 || image.repository == query.target.value
186 || format!("{}:{}", image.repository, image.tag) == query.target.value
187 })
188 .map(|image| image_snapshot_row(timestamp, image)),
189 SingularTargetKind::Network => snapshot
190 .networks
191 .into_iter()
192 .find(|network| network.id == query.target.value || network.name == query.target.value)
193 .map(|network| network_snapshot_row(timestamp, network)),
194 SingularTargetKind::Volume => snapshot
195 .volumes
196 .into_iter()
197 .find(|volume| volume.name == query.target.value)
198 .map(|volume| volume_snapshot_row(timestamp, volume)),
199 }
200 .ok_or_else(|| TelemetryError::TargetNotFound {
201 target: query.target.value.clone(),
202 })?;
203
204 Ok(ExecutionResult { rows: vec![row] })
205}
206
207pub async fn historical_events<S>(
208 query: &EventsQuery,
209 store: &S,
210) -> Result<ExecutionResult, TelemetryError>
211where
212 S: TelemetryStore + ?Sized,
213{
214 let events = match &query.time {
215 Some(TimeSelector::Range { from, to }) => store.events_between(from, to)?,
216 Some(TimeSelector::Last(duration)) => {
217 let now = Utc::now();
218 let seconds = crate::alerts::duration_to_std(*duration).as_secs() as i64;
219 let from_time = now - chrono::Duration::seconds(seconds);
220 let from_str = from_time.to_rfc3339();
221 let to_str = now.to_rfc3339();
222 store.events_between(&from_str, &to_str)?
223 }
224 None => return Err(TelemetryError::MissingStore),
225 };
226 let source = MockEventSource { events };
227 events::collect_events(query, &source)
228 .await
229 .map_err(Into::into)
230}
231
232fn container_snapshot_row(timestamp: &str, container: Container) -> Row {
233 Row {
234 fields: BTreeMap::from([
235 ("snapshot_at".to_owned(), json_string(timestamp)),
236 ("kind".to_owned(), json_string("container")),
237 ("id".to_owned(), json_string(container.id)),
238 ("name".to_owned(), json_string(container.name)),
239 ("image".to_owned(), json_string(container.image)),
240 ("status".to_owned(), json_string(container.status)),
241 ("state".to_owned(), json_string(container.state)),
242 (
243 "restart_count".to_owned(),
244 container.restart_count.map_or(JsonValue::Null, json_u64),
245 ),
246 ]),
247 }
248}
249
250fn image_snapshot_row(timestamp: &str, image: Image) -> Row {
251 Row {
252 fields: BTreeMap::from([
253 ("snapshot_at".to_owned(), json_string(timestamp)),
254 ("kind".to_owned(), json_string("image")),
255 ("id".to_owned(), json_string(image.id)),
256 ("repository".to_owned(), json_string(image.repository)),
257 ("tag".to_owned(), json_string(image.tag)),
258 ("size".to_owned(), json_string(image.size)),
259 ]),
260 }
261}
262
263fn network_snapshot_row(timestamp: &str, network: Network) -> Row {
264 Row {
265 fields: BTreeMap::from([
266 ("snapshot_at".to_owned(), json_string(timestamp)),
267 ("kind".to_owned(), json_string("network")),
268 ("id".to_owned(), json_string(network.id)),
269 ("name".to_owned(), json_string(network.name)),
270 ("driver".to_owned(), json_string(network.driver)),
271 ("scope".to_owned(), json_string(network.scope)),
272 ]),
273 }
274}
275
276fn volume_snapshot_row(timestamp: &str, volume: Volume) -> Row {
277 Row {
278 fields: BTreeMap::from([
279 ("snapshot_at".to_owned(), json_string(timestamp)),
280 ("kind".to_owned(), json_string("volume")),
281 ("name".to_owned(), json_string(volume.name)),
282 ("driver".to_owned(), json_string(volume.driver)),
283 (
284 "mountpoint".to_owned(),
285 volume.mountpoint.map_or(JsonValue::Null, JsonValue::String),
286 ),
287 ]),
288 }
289}
290
291fn json_u64(value: u64) -> JsonValue {
292 JsonValue::Number(Number::from(value))
293}
294
295#[cfg(test)]
296mod tests {
297 use crate::{ast::Query, parser};
298
299 use super::*;
300
301 #[test]
302 fn stores_and_reads_events_by_range() {
303 let mut store = InMemoryTelemetryStore::default();
304 store
305 .write_event(event("2026-01-01T12:00:00Z", "start"))
306 .unwrap();
307 store
308 .write_event(event("2026-01-01T12:05:00Z", "die"))
309 .unwrap();
310 store
311 .write_event(event("2026-01-01T13:00:00Z", "restart"))
312 .unwrap();
313
314 let events = store
315 .events_between("2026-01-01T12:00:00Z", "2026-01-01T12:59:59Z")
316 .unwrap();
317
318 assert_eq!(events.len(), 2);
319 }
320
321 #[test]
322 fn inspects_container_at_historical_snapshot() {
323 let mut store = InMemoryTelemetryStore::default();
324 store
325 .write_snapshot(snapshot("2026-01-01 11:59:00", "old-image"))
326 .unwrap();
327 store
328 .write_snapshot(snapshot("2026-01-01 12:00:00", "new-image"))
329 .unwrap();
330 let Query::Inspect(query) =
331 parser::parse("inspect container api at \"2026-01-01 12:00:30\"")
332 .unwrap()
333 .query
334 else {
335 panic!("expected inspect query");
336 };
337
338 let result = inspect_at(&query, &store).unwrap();
339
340 assert_eq!(result.rows.len(), 1);
341 assert_eq!(
342 result.rows[0].fields["image"],
343 JsonValue::String("new-image".to_owned())
344 );
345 assert_eq!(
346 result.rows[0].fields["snapshot_at"],
347 JsonValue::String("2026-01-01 12:00:30".to_owned())
348 );
349 }
350
351 #[test]
352 fn replays_historical_events_through_event_pipeline() {
353 let rt = tokio::runtime::Runtime::new().unwrap();
354 let mut store = InMemoryTelemetryStore::default();
355 store
356 .write_event(event("2026-01-01T12:00:00Z", "start"))
357 .unwrap();
358 store
359 .write_event(event("2026-01-01T12:05:00Z", "die"))
360 .unwrap();
361 let Query::Events(query) = parser::parse(
362 "events containers from \"2026-01-01T12:00:00Z\" to \"2026-01-01T12:10:00Z\" where action = \"die\" | select time, action",
363 )
364 .unwrap()
365 .query
366 else {
367 panic!("expected events query");
368 };
369
370 let result = rt.block_on(historical_events(&query, &store)).unwrap();
371
372 assert_eq!(result.rows.len(), 1);
373 assert_eq!(
374 result.rows[0].fields["action"],
375 JsonValue::String("die".to_owned())
376 );
377 }
378
379 fn snapshot(timestamp: &str, image: &str) -> TelemetrySnapshot {
380 TelemetrySnapshot {
381 timestamp: timestamp.to_owned(),
382 containers: vec![Container {
383 id: "abc".to_owned(),
384 name: "api".to_owned(),
385 image: image.to_owned(),
386 status: "Up".to_owned(),
387 state: "running".to_owned(),
388 ports: Vec::new(),
389 labels: Vec::new(),
390 created_at: None,
391 started_at: None,
392 finished_at: None,
393 restart_count: Some(1),
394 health: None,
395 }],
396 images: Vec::new(),
397 networks: Vec::new(),
398 volumes: Vec::new(),
399 }
400 }
401
402 fn event(time: &str, action: &str) -> DockerEvent {
403 DockerEvent {
404 time: time.to_owned(),
405 event_type: "container".to_owned(),
406 action: action.to_owned(),
407 actor_id: "abc".to_owned(),
408 container: Some("api".to_owned()),
409 image: Some("api:latest".to_owned()),
410 attributes: vec![("name".to_owned(), "api".to_owned())],
411 }
412 }
413}