soma_infra/
docker_telemetry.rs

1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3use soma_fleet::{HostId, HostRecord, TopologyRevision};
4use tokio_util::sync::CancellationToken;
5
6use crate::{InfraError, InfraResult};
7
8const MAX_LOG_LINES: u32 = 5000;
9const MAX_GREP_CHARS: usize = 1024;
10
11/// Aggregate disk usage for one Docker resource category.
12#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
13pub struct DockerUsageCategory {
14    /// Number of reported resources.
15    pub count: u64,
16    /// Sum of resource size fields in bytes.
17    pub size_bytes: u64,
18}
19
20/// Neutral Docker disk-usage snapshot.
21#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
22pub struct DockerDiskUsage {
23    /// Target host.
24    pub host: HostId,
25    /// Exact topology revision.
26    pub topology_revision: TopologyRevision,
27    /// Shared layer bytes reported by the daemon.
28    pub layers_size_bytes: u64,
29    /// Image usage.
30    pub images: DockerUsageCategory,
31    /// Container writable-layer usage.
32    pub containers: DockerUsageCategory,
33    /// Local volume usage.
34    pub volumes: DockerUsageCategory,
35    /// Build-cache usage.
36    pub build_cache: DockerUsageCategory,
37}
38
39/// Selected Docker log stream.
40#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
41#[serde(rename_all = "snake_case")]
42pub enum DockerLogStream {
43    /// Standard output only.
44    Stdout,
45    /// Standard error only.
46    Stderr,
47    /// Both output streams.
48    #[default]
49    Both,
50}
51
52/// Bounded one-shot Docker log options.
53#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
54pub struct ContainerLogOptions {
55    lines: u32,
56    stream: DockerLogStream,
57    since_unix_seconds: Option<i64>,
58    until_unix_seconds: Option<i64>,
59    grep: Option<String>,
60}
61
62impl Default for ContainerLogOptions {
63    fn default() -> Self {
64        Self {
65            lines: 100,
66            stream: DockerLogStream::Both,
67            since_unix_seconds: None,
68            until_unix_seconds: None,
69            grep: None,
70        }
71    }
72}
73
74impl ContainerLogOptions {
75    /// Sets the requested tail line count.
76    pub fn with_lines(mut self, lines: u32) -> InfraResult<Self> {
77        if lines == 0 || lines > MAX_LOG_LINES {
78            return Err(InfraError::InvalidRequest {
79                domain: "docker",
80                message: format!("container log lines must be 1-{MAX_LOG_LINES}"),
81            });
82        }
83        self.lines = lines;
84        Ok(self)
85    }
86
87    /// Selects the output stream.
88    #[must_use]
89    pub const fn with_stream(mut self, stream: DockerLogStream) -> Self {
90        self.stream = stream;
91        self
92    }
93
94    /// Sets an inclusive lower Unix-second bound.
95    pub fn with_since(mut self, seconds: i64) -> InfraResult<Self> {
96        self.since_unix_seconds = Some(seconds);
97        self.validate_time_order()?;
98        Ok(self)
99    }
100
101    /// Sets an inclusive upper Unix-second bound.
102    pub fn with_until(mut self, seconds: i64) -> InfraResult<Self> {
103        self.until_unix_seconds = Some(seconds);
104        self.validate_time_order()?;
105        Ok(self)
106    }
107
108    /// Adds a local case-sensitive substring filter.
109    pub fn with_grep(mut self, grep: impl Into<String>) -> InfraResult<Self> {
110        let grep = grep.into();
111        let count = grep.chars().count();
112        if count == 0 || count > MAX_GREP_CHARS || grep.chars().any(char::is_control) {
113            return Err(InfraError::InvalidRequest {
114                domain: "docker",
115                message: format!(
116                    "container log grep must be 1-{MAX_GREP_CHARS} printable characters"
117                ),
118            });
119        }
120        self.grep = Some(grep);
121        Ok(self)
122    }
123
124    /// Returns the line count.
125    #[must_use]
126    pub const fn lines(&self) -> u32 {
127        self.lines
128    }
129
130    /// Returns the selected stream.
131    #[must_use]
132    pub const fn stream(&self) -> DockerLogStream {
133        self.stream
134    }
135
136    /// Returns the lower Unix-second bound.
137    #[must_use]
138    pub const fn since_unix_seconds(&self) -> Option<i64> {
139        self.since_unix_seconds
140    }
141
142    /// Returns the upper Unix-second bound.
143    #[must_use]
144    pub const fn until_unix_seconds(&self) -> Option<i64> {
145        self.until_unix_seconds
146    }
147
148    /// Returns the optional local grep filter.
149    #[must_use]
150    pub fn grep(&self) -> Option<&str> {
151        self.grep.as_deref()
152    }
153
154    fn validate_time_order(&self) -> InfraResult<()> {
155        if matches!(
156            (self.since_unix_seconds, self.until_unix_seconds),
157            (Some(since), Some(until)) if since > until
158        ) {
159            Err(InfraError::InvalidRequest {
160                domain: "docker",
161                message: "container log since must not exceed until".into(),
162            })
163        } else {
164            Ok(())
165        }
166    }
167}
168
169/// Bounded one-shot Docker log result.
170#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
171pub struct ContainerLogs {
172    /// Target host.
173    pub host: HostId,
174    /// Exact topology revision.
175    pub topology_revision: TopologyRevision,
176    /// Container identifier.
177    pub container: String,
178    /// Rendered non-empty log lines.
179    pub lines: Vec<String>,
180    /// Whether the client-side byte ceiling omitted output.
181    pub truncated: bool,
182}
183
184/// Neutral one-shot Docker container statistics.
185#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
186pub struct ContainerStatsSnapshot {
187    /// Target host.
188    pub host: HostId,
189    /// Exact topology revision.
190    pub topology_revision: TopologyRevision,
191    /// Container identifier.
192    pub container: String,
193    /// Daemon read timestamp.
194    pub read_at: Option<String>,
195    /// Current process count.
196    pub pids_current: u64,
197    /// Current memory usage.
198    pub memory_usage_bytes: u64,
199    /// Memory limit.
200    pub memory_limit_bytes: u64,
201    /// Total container CPU usage.
202    pub cpu_total_usage: u64,
203    /// Host system CPU usage.
204    pub system_cpu_usage: u64,
205    /// Online CPU count.
206    pub online_cpus: u64,
207    /// Aggregate received network bytes.
208    pub network_rx_bytes: u64,
209    /// Aggregate transmitted network bytes.
210    pub network_tx_bytes: u64,
211    /// Aggregate block-device read bytes.
212    pub block_read_bytes: u64,
213    /// Aggregate block-device write bytes.
214    pub block_write_bytes: u64,
215}
216
217/// Docker telemetry read operations.
218#[async_trait]
219pub trait DockerTelemetryReader: Send + Sync {
220    /// Reads daemon disk usage.
221    async fn disk_usage(
222        &self,
223        host: &HostRecord,
224        cancellation: &CancellationToken,
225    ) -> InfraResult<DockerDiskUsage>;
226
227    /// Reads bounded one-shot container logs.
228    async fn container_logs(
229        &self,
230        host: &HostRecord,
231        container: &str,
232        options: &ContainerLogOptions,
233        cancellation: &CancellationToken,
234    ) -> InfraResult<ContainerLogs>;
235
236    /// Reads one container statistics frame.
237    async fn container_stats(
238        &self,
239        host: &HostRecord,
240        container: &str,
241        cancellation: &CancellationToken,
242    ) -> InfraResult<ContainerStatsSnapshot>;
243}
244
245#[cfg(test)]
246#[path = "docker_telemetry_tests.rs"]
247mod tests;
soma_infra/docker_telemetry.rs

soma_infra/
docker_telemetry.rs
