dol/
metrics.rs

1//! Metrics collection from Docker containers.
2//!
3//! Defines the [`MetricsCollector`] trait with implementations:
4//! - [`BollardMetricsCollector`] — live stats from bollard
5//! - [`MockMetricsCollector`] — test fixtures
6//! - [`NoopMetricsCollector`] — no-op for non-metric queries
7//! - [`MetricRingBuffer`] — bounded circular buffer for windowed analysis
8//!
9//! # Example
10//!
11//! ```ignore
12//! let collector = BollardMetricsCollector::with_config(Arc::new(docker), &config);
13//! let samples = collector.collect().await?;
14//! ```
15
16use std::collections::{HashMap, VecDeque};
17use std::sync::Arc;
18use std::time::Duration;
19
20use thiserror::Error;
21use tokio::time::timeout;
22
23use crate::config::DolConfig;
24use crate::docker::{DockerClient, DockerError, MetricSample};
25
26pub trait MetricsCollector {
27    fn collect(
28        &self,
29    ) -> impl std::future::Future<Output = Result<Vec<MetricSample>, MetricsError>> + Send;
30}
31
32#[derive(Debug, Error)]
33pub enum MetricsError {
34    #[error("{0}")]
35    Docker(#[from] DockerError),
36    #[error("invalid docker stats field `{field}`: {value}")]
37    InvalidStatsField { field: &'static str, value: String },
38    #[error("tokio runtime error: {0}")]
39    Runtime(String),
40}
41
42/// A metrics collector that uses a `DockerClient` (bollard) to fetch container stats.
43///
44/// Collectors wrap an `Arc<C>` so they can be passed by `Clone` into
45/// `tokio::task::spawn_blocking` and other `'static` contexts.
46#[derive(Debug, Clone)]
47pub struct BollardMetricsCollector<C> {
48    docker: Arc<C>,
49    stats_timeout: Duration,
50}
51
52impl<C> BollardMetricsCollector<C>
53where
54    C: DockerClient + Send + Sync + 'static,
55{
56    /// Create a new collector with the default stats timeout (10s).
57    pub const fn new(docker: Arc<C>) -> Self {
58        Self {
59            docker,
60            stats_timeout: Duration::from_secs(10),
61        }
62    }
63
64    /// Create a new collector with a specific stats timeout from config.
65    #[must_use]
66    pub fn with_config(docker: Arc<C>, config: &DolConfig) -> Self {
67        Self {
68            docker,
69            stats_timeout: Duration::from_secs(config.stats_timeout.unwrap_or(10)),
70        }
71    }
72
73    /// Override the per-container stats call timeout.
74    pub fn set_stats_timeout(&mut self, timeout: Duration) {
75        self.stats_timeout = timeout;
76    }
77}
78
79impl<C> MetricsCollector for BollardMetricsCollector<C>
80where
81    C: DockerClient + Send + Sync + 'static,
82{
83    async fn collect(&self) -> Result<Vec<MetricSample>, MetricsError> {
84        let containers = self
85            .docker
86            .list_containers()
87            .await
88            .map_err(MetricsError::Docker)?;
89
90        let mut samples = Vec::new();
91        for container in &containers {
92            let id_short = &container.id[..12.min(container.id.len())];
93            let id = id_short.to_owned();
94            match timeout(self.stats_timeout, self.docker.container_stats(&id)).await {
95                Ok(Ok(sample)) => samples.push(sample),
96                Ok(Err(e)) => {
97                    eprintln!("[metrics] failed to get stats for {}: {e}", container.name);
98                }
99                Err(_) => {
100                    eprintln!(
101                        "[metrics] stats call timed out after {}s for {}",
102                        self.stats_timeout.as_secs(),
103                        container.name
104                    );
105                }
106            }
107        }
108        Ok(samples)
109    }
110}
111
112#[derive(Debug, Clone, Default)]
113pub struct MockMetricsCollector {
114    pub samples: Vec<MetricSample>,
115}
116
117impl MetricsCollector for MockMetricsCollector {
118    async fn collect(&self) -> Result<Vec<MetricSample>, MetricsError> {
119        Ok(self.samples.clone())
120    }
121}
122
123#[derive(Debug, Clone, Default)]
124pub struct NoopMetricsCollector;
125
126impl MetricsCollector for NoopMetricsCollector {
127    async fn collect(&self) -> Result<Vec<MetricSample>, MetricsError> {
128        Ok(Vec::new())
129    }
130}
131
132#[derive(Debug, Clone, Default)]
133pub struct MetricRingBuffer {
134    samples: VecDeque<MetricSample>,
135    capacity: usize,
136}
137
138impl MetricRingBuffer {
139    #[must_use]
140    pub fn new(capacity: usize) -> Self {
141        Self {
142            samples: VecDeque::with_capacity(capacity),
143            capacity,
144        }
145    }
146
147    pub fn push(&mut self, sample: MetricSample) {
148        if self.capacity == 0 {
149            return;
150        }
151
152        while self.samples.len() >= self.capacity {
153            self.samples.pop_front();
154        }
155        self.samples.push_back(sample);
156    }
157
158    #[must_use]
159    pub fn latest_by_container(&self) -> HashMap<String, MetricSample> {
160        let mut latest = HashMap::new();
161        for sample in &self.samples {
162            latest.insert(sample.container_id.clone(), sample.clone());
163            latest.insert(sample.container_name.clone(), sample.clone());
164        }
165        latest
166    }
167}
168
169/// Parse a percentage string like "85.42%" into a float.
170#[allow(clippy::result_unit_err)]
171pub fn parse_percent(value: &str) -> Result<f64, ()> {
172    value.trim().trim_end_matches('%').parse().map_err(|_| ())
173}
174
175/// Parse a usage pair string like "128MiB / 1GiB" into (used, limit) bytes.
176#[allow(clippy::result_unit_err)]
177pub fn parse_usage_pair(value: &str) -> Result<(Option<u64>, Option<u64>), ()> {
178    let mut parts = value.split('/').map(str::trim);
179    let left = parts.next().ok_or(())?;
180    let right = parts.next();
181
182    Ok((
183        parse_byte_quantity(left)?,
184        right.map(parse_byte_quantity).transpose()?.flatten(),
185    ))
186}
187
188#[allow(clippy::result_unit_err)]
189pub fn parse_byte_quantity(value: &str) -> Result<Option<u64>, ()> {
190    let value = value.trim();
191    if value.is_empty() || value == "--" {
192        return Ok(None);
193    }
194
195    let split_at = value
196        .find(|c: char| !(c.is_ascii_digit() || c == '.'))
197        .unwrap_or(value.len());
198    let number = value[..split_at].trim().parse::<f64>().map_err(|_| ())?;
199    let unit = value[split_at..].trim().to_ascii_lowercase();
200    let multiplier = match unit.as_str() {
201        "" | "b" => 1.0,
202        "kb" => 1_000.0,
203        "mb" => 1_000_000.0,
204        "gb" => 1_000_000_000.0,
205        "tb" => 1_000_000_000_000.0,
206        "kib" => 1024.0,
207        "mib" => 1024.0 * 1024.0,
208        "gib" => 1024.0 * 1024.0 * 1024.0,
209        "tib" => 1024.0 * 1024.0 * 1024.0 * 1024.0,
210        _ => return Err(()),
211    };
212
213    Ok(Some((number * multiplier).round() as u64))
214}
215
216#[cfg(test)]
217mod tests {
218    use super::*;
219
220    #[test]
221    fn parses_percent() {
222        assert_eq!(parse_percent("85.42%"), Ok(85.42));
223    }
224
225    #[test]
226    fn parses_byte_quantities() {
227        assert_eq!(parse_byte_quantity("1KiB"), Ok(Some(1024)));
228        assert_eq!(parse_byte_quantity("1.5MB"), Ok(Some(1_500_000)));
229        assert_eq!(parse_byte_quantity("--"), Ok(None));
230    }
231
232    #[test]
233    fn parses_usage_pairs() {
234        assert_eq!(
235            parse_usage_pair("12.5MiB / 1GiB"),
236            Ok((Some(13_107_200), Some(1_073_741_824)))
237        );
238    }
239
240    #[tokio::test]
241    async fn mock_collector_works() {
242        let collector = MockMetricsCollector {
243            samples: vec![MetricSample {
244                container_id: "abc".to_owned(),
245                container_name: "api".to_owned(),
246                timestamp: "2026-01-01T12:00:00Z".to_owned(),
247                cpu_percent: Some(50.0),
248                memory_usage_bytes: Some(128),
249                memory_limit_bytes: Some(1024),
250                network_rx_bytes: None,
251                network_tx_bytes: None,
252                disk_read_bytes: None,
253                disk_write_bytes: None,
254            }],
255        };
256        let samples = collector.collect().await.unwrap();
257        assert_eq!(samples.len(), 1);
258        assert_eq!(samples[0].container_name, "api");
259    }
260
261    #[tokio::test]
262    async fn noop_collector_returns_empty() {
263        let samples = NoopMetricsCollector.collect().await.unwrap();
264        assert!(samples.is_empty());
265    }
266
267    #[test]
268    fn ring_buffer_keeps_latest_samples_by_container() {
269        let mut buffer = MetricRingBuffer::new(1);
270        buffer.push(MetricSample {
271            container_id: "abc".to_owned(),
272            container_name: "api".to_owned(),
273            timestamp: "t1".to_owned(),
274            cpu_percent: Some(10.0),
275            memory_usage_bytes: None,
276            memory_limit_bytes: None,
277            network_rx_bytes: None,
278            network_tx_bytes: None,
279            disk_read_bytes: None,
280            disk_write_bytes: None,
281        });
282        buffer.push(MetricSample {
283            container_id: "abc".to_owned(),
284            container_name: "api".to_owned(),
285            timestamp: "t2".to_owned(),
286            cpu_percent: Some(20.0),
287            memory_usage_bytes: None,
288            memory_limit_bytes: None,
289            network_rx_bytes: None,
290            network_tx_bytes: None,
291            disk_read_bytes: None,
292            disk_write_bytes: None,
293        });
294
295        let latest = buffer.latest_by_container();
296
297        assert_eq!(latest["abc"].cpu_percent, Some(20.0));
298        assert_eq!(latest["api"].cpu_percent, Some(20.0));
299    }
300}
dol/metrics.rs

dol/
metrics.rs
