soma_infra/
bollard_driver.rs

1use std::collections::HashMap;
2use std::future::Future;
3use std::path::Path;
4#[cfg(feature = "remote-bollard")]
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use bollard::query_parameters::{
9    ListContainersOptions, ListImagesOptions, ListNetworksOptions, ListVolumesOptions, TopOptions,
10};
11use bollard::{API_DEFAULT_VERSION, Docker};
12#[cfg(feature = "remote-bollard")]
13use soma_fleet::{ForwardedUnixSocket, OpenSshConnection};
14use soma_fleet::{HostEndpoint, HostId, HostRecord, TopologyRevision};
15use tokio_util::sync::CancellationToken;
16
17use crate::docker_map::{
18    map_container_inspect, map_container_summary, map_image, map_network, map_system_info,
19    map_volume, parse_error,
20};
21use crate::{
22    ContainerInspect, ContainerListOptions, ContainerProcessTable, ContainerReader,
23    ContainerSummary, DockerSystemInfo, DockerSystemReader, ImageListOptions, ImageReader,
24    ImageSummary, InfraError, InfraResult, NetworkReader, NetworkSummary, VolumeReader,
25    VolumeSummary,
26};
27
28const MAX_LIST_ITEMS: usize = 10_000;
29const MAX_LIST_ITEM_JSON_BYTES: usize = 256 * 1024;
30const CLIENT_TIMEOUT_SECONDS: u64 = 120;
31
32/// Local Bollard implementation of the neutral Docker read contracts.
33pub struct BollardReadClient {
34    docker: Docker,
35    host: HostId,
36    revision: TopologyRevision,
37    #[cfg(feature = "remote-bollard")]
38    _connection: Option<Arc<OpenSshConnection>>,
39    #[cfg(feature = "remote-bollard")]
40    forward: Option<ForwardedUnixSocket>,
41}
42
43impl BollardReadClient {
44    /// Connects to a local Docker socket and binds the client to one host revision.
45    pub fn connect_local(host: &HostRecord) -> InfraResult<Self> {
46        if !matches!(host.endpoint(), HostEndpoint::Local) {
47            return Err(InfraError::UnsupportedTarget {
48                domain: "docker",
49                host: host.id().clone(),
50            });
51        }
52        let docker = Docker::connect_with_socket_defaults()
53            .map_err(|error| InfraError::Docker(error.to_string()))?;
54        Ok(Self {
55            docker,
56            host: host.id().clone(),
57            revision: host.revision().clone(),
58            #[cfg(feature = "remote-bollard")]
59            _connection: None,
60            #[cfg(feature = "remote-bollard")]
61            forward: None,
62        })
63    }
64
65    /// Connects to a remote Docker Unix socket through a strict OpenSSH forward.
66    #[cfg(feature = "remote-bollard")]
67    pub async fn connect_remote(
68        connection: Arc<OpenSshConnection>,
69        host: &HostRecord,
70        remote_socket: &Path,
71        cancellation: &CancellationToken,
72    ) -> InfraResult<Self> {
73        if !matches!(host.endpoint(), HostEndpoint::Ssh(_)) {
74            return Err(InfraError::UnsupportedTarget {
75                domain: "docker",
76                host: host.id().clone(),
77            });
78        }
79        let forward =
80            ForwardedUnixSocket::open(&connection, host, remote_socket, cancellation).await?;
81        let docker = Docker::connect_with_socket(
82            forward.path().to_string_lossy().as_ref(),
83            CLIENT_TIMEOUT_SECONDS,
84            API_DEFAULT_VERSION,
85        )
86        .map_err(|error| InfraError::Docker(error.to_string()))?;
87        Ok(Self {
88            docker,
89            host: host.id().clone(),
90            revision: host.revision().clone(),
91            _connection: Some(connection),
92            forward: Some(forward),
93        })
94    }
95
96    /// Explicitly closes an owned remote forward when present.
97    #[cfg(feature = "remote-bollard")]
98    pub async fn close(mut self) -> InfraResult<()> {
99        if let Some(forward) = self.forward.take() {
100            forward.close().await?;
101        }
102        Ok(())
103    }
104
105    pub(crate) fn validate_host(&self, host: &HostRecord) -> InfraResult<()> {
106        if host.id() == &self.host && host.revision() == &self.revision {
107            Ok(())
108        } else {
109            Err(InfraError::InvalidRequest {
110                domain: "docker",
111                message: format!(
112                    "client is bound to {}@{}, received {}@{}",
113                    self.host,
114                    self.revision,
115                    host.id(),
116                    host.revision()
117                ),
118            })
119        }
120    }
121
122    pub(crate) fn docker(&self) -> &Docker {
123        &self.docker
124    }
125}
126
127#[async_trait]
128impl DockerSystemReader for BollardReadClient {
129    async fn system_info(
130        &self,
131        host: &HostRecord,
132        cancellation: &CancellationToken,
133    ) -> InfraResult<DockerSystemInfo> {
134        self.validate_host(host)?;
135        let value = serde_json::to_value(cancellable(cancellation, self.docker.info()).await?)
136            .map_err(|error| parse_error(error.to_string()))?;
137        map_system_info(host, &value)
138    }
139}
140
141#[async_trait]
142impl ContainerReader for BollardReadClient {
143    async fn list_containers(
144        &self,
145        host: &HostRecord,
146        options: &ContainerListOptions,
147        cancellation: &CancellationToken,
148    ) -> InfraResult<Vec<ContainerSummary>> {
149        self.validate_host(host)?;
150        let mut query = ListContainersOptions {
151            all: options.all,
152            ..Default::default()
153        };
154        if let Some(label) = options.label.as_deref() {
155            validate_filter(label)?;
156            let mut filters = HashMap::new();
157            filters.insert("label".to_owned(), vec![label.to_owned()]);
158            query.filters = Some(filters);
159        }
160        let rows = cancellable(cancellation, self.docker.list_containers(Some(query))).await?;
161        ensure_list_bound("containers", rows.len())?;
162        rows.into_iter()
163            .map(|row| bounded_json_value("container", row))
164            .map(|value| value.and_then(|value| map_container_summary(host, &value)))
165            .filter(|result| {
166                result.as_ref().map_or(true, |row| {
167                    options
168                        .state
169                        .as_ref()
170                        .is_none_or(|state| &row.state == state)
171                })
172            })
173            .collect()
174    }
175
176    async fn inspect_container(
177        &self,
178        host: &HostRecord,
179        container: &str,
180        cancellation: &CancellationToken,
181    ) -> InfraResult<ContainerInspect> {
182        self.validate_host(host)?;
183        validate_identifier("container", container)?;
184        let row = cancellable(cancellation, self.docker.inspect_container(container, None)).await?;
185        let value = serde_json::to_value(row).map_err(|error| parse_error(error.to_string()))?;
186        map_container_inspect(host, &value)
187    }
188
189    async fn top_container(
190        &self,
191        host: &HostRecord,
192        container: &str,
193        cancellation: &CancellationToken,
194    ) -> InfraResult<ContainerProcessTable> {
195        self.validate_host(host)?;
196        validate_identifier("container", container)?;
197        let response = cancellable(
198            cancellation,
199            self.docker
200                .top_processes(container, Some(TopOptions::default())),
201        )
202        .await?;
203        Ok(ContainerProcessTable {
204            host: host.id().clone(),
205            topology_revision: host.revision().clone(),
206            container: container.to_owned(),
207            titles: response.titles.unwrap_or_default(),
208            processes: response.processes.unwrap_or_default(),
209        })
210    }
211}
212
213#[async_trait]
214impl ImageReader for BollardReadClient {
215    async fn list_images(
216        &self,
217        host: &HostRecord,
218        options: &ImageListOptions,
219        cancellation: &CancellationToken,
220    ) -> InfraResult<Vec<ImageSummary>> {
221        self.validate_host(host)?;
222        let mut query = ListImagesOptions {
223            all: options.all,
224            ..Default::default()
225        };
226        if options.dangling_only {
227            let mut filters = HashMap::new();
228            filters.insert("dangling".to_owned(), vec!["true".to_owned()]);
229            query.filters = Some(filters);
230        }
231        let rows = cancellable(cancellation, self.docker.list_images(Some(query))).await?;
232        ensure_list_bound("images", rows.len())?;
233        rows.into_iter()
234            .map(|row| bounded_json_value("image", row))
235            .map(|value| value.and_then(|value| map_image(host, &value)))
236            .collect()
237    }
238}
239
240#[async_trait]
241impl NetworkReader for BollardReadClient {
242    async fn list_networks(
243        &self,
244        host: &HostRecord,
245        cancellation: &CancellationToken,
246    ) -> InfraResult<Vec<NetworkSummary>> {
247        self.validate_host(host)?;
248        let rows = cancellable(
249            cancellation,
250            self.docker.list_networks(None::<ListNetworksOptions>),
251        )
252        .await?;
253        ensure_list_bound("networks", rows.len())?;
254        rows.into_iter()
255            .map(|row| bounded_json_value("network", row))
256            .map(|value| value.and_then(|value| map_network(host, &value)))
257            .collect()
258    }
259}
260
261#[async_trait]
262impl VolumeReader for BollardReadClient {
263    async fn list_volumes(
264        &self,
265        host: &HostRecord,
266        cancellation: &CancellationToken,
267    ) -> InfraResult<Vec<VolumeSummary>> {
268        self.validate_host(host)?;
269        let response = cancellable(
270            cancellation,
271            self.docker.list_volumes(None::<ListVolumesOptions>),
272        )
273        .await?;
274        let rows = response.volumes.unwrap_or_default();
275        ensure_list_bound("volumes", rows.len())?;
276        rows.into_iter()
277            .map(|row| bounded_json_value("volume", row))
278            .map(|value| value.and_then(|value| map_volume(host, &value)))
279            .collect()
280    }
281}
282
283fn ensure_list_bound(domain: &'static str, count: usize) -> InfraResult<()> {
284    if count > MAX_LIST_ITEMS {
285        Err(InfraError::InvalidRequest {
286            domain: "docker",
287            message: format!("{domain} response exceeds the {MAX_LIST_ITEMS}-item limit"),
288        })
289    } else {
290        Ok(())
291    }
292}
293
294fn bounded_json_value<T: serde::Serialize>(
295    domain: &'static str,
296    row: T,
297) -> InfraResult<serde_json::Value> {
298    let encoded = serde_json::to_vec(&row).map_err(|error| parse_error(error.to_string()))?;
299    if encoded.len() > MAX_LIST_ITEM_JSON_BYTES {
300        return Err(InfraError::InvalidRequest {
301            domain: "docker",
302            message: format!(
303                "{domain} response item exceeds the {MAX_LIST_ITEM_JSON_BYTES}-byte limit"
304            ),
305        });
306    }
307    serde_json::from_slice(&encoded).map_err(|error| parse_error(error.to_string()))
308}
309
310pub(crate) async fn cancellable<T, F>(cancellation: &CancellationToken, future: F) -> InfraResult<T>
311where
312    F: Future<Output = Result<T, bollard::errors::Error>>,
313{
314    tokio::select! {
315        () = cancellation.cancelled() => Err(soma_fleet::FleetError::Cancelled.into()),
316        result = future => result.map_err(|error| InfraError::Docker(error.to_string())),
317    }
318}
319
320fn validate_filter(value: &str) -> InfraResult<()> {
321    if value.is_empty() || value.len() > 1024 || value.chars().any(char::is_control) {
322        Err(InfraError::InvalidRequest {
323            domain: "docker",
324            message: "label filter must contain 1-1024 printable characters".into(),
325        })
326    } else {
327        Ok(())
328    }
329}
330
331fn validate_identifier(kind: &str, value: &str) -> InfraResult<()> {
332    if value.is_empty() || value.len() > 256 || value.chars().any(char::is_control) {
333        Err(InfraError::InvalidRequest {
334            domain: "docker",
335            message: format!("invalid {kind} identifier"),
336        })
337    } else {
338        Ok(())
339    }
340}
341
342#[cfg(test)]
343#[path = "bollard_driver_tests.rs"]
344mod tests;
soma_infra/bollard_driver.rs

soma_infra/
bollard_driver.rs
