soma_infra/
bollard_recreate.rs

1use std::future::Future;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use bollard::models::{ContainerCreateBody, ContainerInspectResponse, NetworkingConfig};
6use bollard::query_parameters::{
7    CreateContainerOptions, CreateImageOptions, StartContainerOptions, StopContainerOptions,
8};
9use futures_util::StreamExt;
10use serde_json::json;
11use soma_fleet::{HostRecord, TopologyRevision};
12use soma_ops::{MutationSendState, Timestamp};
13use tokio_util::sync::CancellationToken;
14
15use crate::docker_map::map_container_inspect;
16use crate::{
17    BollardReadClient, ContainerRecreateFingerprint, ContainerRecreateInspector,
18    ContainerRecreateMutator, ContainerRecreateReceipt, ContainerRecreateRequest,
19    ContainerRecreateStage, InfraError, InfraResult, MutationFailure, MutationResult,
20};
21
22#[async_trait]
23impl ContainerRecreateInspector for BollardReadClient {
24    async fn recreate_fingerprint(
25        &self,
26        host: &HostRecord,
27        container: &str,
28        cancellation: &CancellationToken,
29    ) -> InfraResult<ContainerRecreateFingerprint> {
30        self.validate_host(host)?;
31        let raw = tokio::select! {
32            () = cancellation.cancelled() => return Err(soma_fleet::FleetError::Cancelled.into()),
33            result = self.docker().inspect_container(container, None) => result
34                .map_err(|error| InfraError::Docker(error.to_string()))?,
35        };
36        fingerprint(host, container, &raw)
37    }
38}
39
40#[async_trait]
41impl ContainerRecreateMutator for BollardReadClient {
42    async fn recreate_container(
43        &self,
44        host: &HostRecord,
45        request: &ContainerRecreateRequest,
46        cancellation: &CancellationToken,
47    ) -> MutationResult<ContainerRecreateReceipt> {
48        self.validate_host(host)
49            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
50        ensure_not_expired(request.deadline(), cancellation)?;
51        let raw = self
52            .docker()
53            .inspect_container(&request.expected().container, None)
54            .await
55            .map_err(|error| {
56                MutationFailure::new(
57                    MutationSendState::NotSent,
58                    InfraError::Docker(error.to_string()),
59                )
60            })?;
61        let current = fingerprint(host, &request.expected().container, &raw)
62            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
63        if current != *request.expected() {
64            return Err(MutationFailure::new(
65                MutationSendState::NotSent,
66                InfraError::InvalidRequest {
67                    domain: "container-recreate",
68                    message: "container configuration changed immediately before replacement"
69                        .into(),
70                },
71            ));
72        }
73
74        let image = image_ref(&raw)
75            .map_err(|error| MutationFailure::new(MutationSendState::NotSent, error))?;
76        let name = container_name(&raw, &request.expected().container);
77        if request.pull() {
78            pull_image(self, &image, request.deadline(), cancellation).await?;
79        }
80
81        let mut stage = ContainerRecreateStage::Prepared;
82        await_stage(
83            request.deadline(),
84            cancellation,
85            stage,
86            self.docker()
87                .stop_container(&request.expected().container, None::<StopContainerOptions>),
88        )
89        .await?;
90        stage = ContainerRecreateStage::Stopped;
91
92        await_stage(
93            request.deadline(),
94            cancellation,
95            stage,
96            self.docker()
97                .remove_container(&request.expected().container, None),
98        )
99        .await?;
100        stage = ContainerRecreateStage::Removed;
101
102        let body = create_body(&raw, &image);
103        let created = await_stage(
104            request.deadline(),
105            cancellation,
106            stage,
107            self.docker().create_container(
108                Some(CreateContainerOptions {
109                    name: Some(name.clone()),
110                    platform: String::new(),
111                }),
112                body,
113            ),
114        )
115        .await?;
116        stage = ContainerRecreateStage::Created;
117
118        await_stage(
119            request.deadline(),
120            cancellation,
121            stage,
122            self.docker()
123                .start_container(&created.id, None::<StartContainerOptions>),
124        )
125        .await?;
126        stage = ContainerRecreateStage::Started;
127
128        Ok(ContainerRecreateReceipt {
129            host: host.id().clone(),
130            topology_revision: TopologyRevision::clone(host.revision()),
131            original_container: request.expected().container.clone(),
132            new_container: Some(created.id),
133            name,
134            image,
135            stage,
136            send_state: MutationSendState::Sent,
137            pulled: request.pull(),
138        })
139    }
140}
141
142fn fingerprint(
143    host: &HostRecord,
144    container: &str,
145    raw: &ContainerInspectResponse,
146) -> InfraResult<ContainerRecreateFingerprint> {
147    let value = serde_json::to_value(raw).map_err(|error| InfraError::Parse {
148        domain: "container-recreate",
149        message: error.to_string(),
150    })?;
151    let neutral = map_container_inspect(host, &value)?;
152    let name = container_name(raw, container);
153    let image = image_ref(raw)?;
154    let material = json!({
155        "name": name,
156        "image": image,
157        "config": raw.config,
158        "host_config": raw.host_config,
159        "networks": raw.network_settings.as_ref().and_then(|settings| settings.networks.as_ref()),
160    });
161    let encoded = serde_json::to_vec(&material).map_err(|error| InfraError::Parse {
162        domain: "container-recreate",
163        message: error.to_string(),
164    })?;
165    let sha256 = crate::mutation::sha256_hex(&encoded);
166    ContainerRecreateFingerprint::new(container, name, image, neutral.state, sha256)
167}
168
169fn image_ref(raw: &ContainerInspectResponse) -> InfraResult<String> {
170    raw.config
171        .as_ref()
172        .and_then(|config| config.image.clone())
173        .filter(|image| !image.is_empty())
174        .ok_or_else(|| InfraError::Parse {
175            domain: "container-recreate",
176            message: "container inspection does not include an image reference".into(),
177        })
178}
179
180fn container_name(raw: &ContainerInspectResponse, fallback: &str) -> String {
181    raw.name
182        .as_deref()
183        .map(|name| name.trim_start_matches('/').to_owned())
184        .filter(|name| !name.is_empty())
185        .unwrap_or_else(|| fallback.to_owned())
186}
187
188fn create_body(raw: &ContainerInspectResponse, image: &str) -> ContainerCreateBody {
189    let config = raw.config.as_ref();
190    let networking_config = raw
191        .network_settings
192        .as_ref()
193        .and_then(|settings| settings.networks.as_ref())
194        .map(|networks| NetworkingConfig {
195            endpoints_config: Some(networks.clone()),
196        });
197    ContainerCreateBody {
198        image: Some(image.to_owned()),
199        env: config.and_then(|config| config.env.clone()),
200        cmd: config.and_then(|config| config.cmd.clone()),
201        entrypoint: config.and_then(|config| config.entrypoint.clone()),
202        labels: config.and_then(|config| config.labels.clone()),
203        working_dir: config.and_then(|config| config.working_dir.clone()),
204        user: config.and_then(|config| config.user.clone()),
205        volumes: config.and_then(|config| config.volumes.clone()),
206        host_config: raw.host_config.clone(),
207        networking_config,
208        ..Default::default()
209    }
210}
211
212async fn pull_image(
213    client: &BollardReadClient,
214    image: &str,
215    deadline: Timestamp,
216    cancellation: &CancellationToken,
217) -> MutationResult<()> {
218    let (from_image, tag) = split_image(image);
219    let mut stream = client.docker().create_image(
220        Some(CreateImageOptions {
221            from_image: Some(from_image),
222            tag,
223            ..Default::default()
224        }),
225        None,
226        None,
227    );
228    loop {
229        let remaining = remaining(deadline)?;
230        let item = tokio::select! {
231            () = cancellation.cancelled() => return Err(MutationFailure::new(
232                MutationSendState::Unknown,
233                soma_fleet::FleetError::Cancelled.into(),
234            )),
235            result = tokio::time::timeout(remaining, stream.next()) => match result {
236                Err(_) => return Err(MutationFailure::new(
237                    MutationSendState::Unknown,
238                    soma_fleet::FleetError::DeadlineExceeded.into(),
239                )),
240                Ok(value) => value,
241            }
242        };
243        match item {
244            None => return Ok(()),
245            Some(Ok(frame)) => {
246                let value = serde_json::to_value(frame).map_err(|error| {
247                    MutationFailure::new(
248                        MutationSendState::Sent,
249                        InfraError::Parse {
250                            domain: "container-recreate",
251                            message: error.to_string(),
252                        },
253                    )
254                })?;
255                let error = value
256                    .get("error_detail")
257                    .or_else(|| value.get("errorDetail"))
258                    .and_then(|detail| detail.get("message"))
259                    .and_then(serde_json::Value::as_str);
260                if let Some(error) = error {
261                    return Err(MutationFailure::new(
262                        MutationSendState::Sent,
263                        InfraError::Docker(error.to_owned()),
264                    ));
265                }
266            }
267            Some(Err(error)) => {
268                return Err(MutationFailure::new(
269                    MutationSendState::Unknown,
270                    InfraError::Docker(error.to_string()),
271                ));
272            }
273        }
274    }
275}
276
277fn split_image(image: &str) -> (String, Option<String>) {
278    if image.contains('@') {
279        return (image.to_owned(), None);
280    }
281    let slash = image.rfind('/');
282    let colon = image.rfind(':');
283    match colon.filter(|colon| slash.is_none_or(|slash| *colon > slash)) {
284        Some(colon) => (
285            image[..colon].to_owned(),
286            Some(image[colon + 1..].to_owned()),
287        ),
288        None => (image.to_owned(), Some("latest".into())),
289    }
290}
291
292fn ensure_not_expired(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
293    if cancellation.is_cancelled() {
294        return Err(MutationFailure::new(
295            MutationSendState::NotSent,
296            soma_fleet::FleetError::Cancelled.into(),
297        ));
298    }
299    remaining(deadline).map(|_| ())
300}
301
302fn remaining(deadline: Timestamp) -> MutationResult<Duration> {
303    let millis = deadline
304        .unix_millis()
305        .saturating_sub(Timestamp::now().unix_millis());
306    if millis <= 0 {
307        Err(MutationFailure::new(
308            MutationSendState::NotSent,
309            soma_fleet::FleetError::DeadlineExceeded.into(),
310        ))
311    } else {
312        Ok(Duration::from_millis(millis as u64))
313    }
314}
315
316async fn await_stage<T, F>(
317    deadline: Timestamp,
318    cancellation: &CancellationToken,
319    stage: ContainerRecreateStage,
320    future: F,
321) -> MutationResult<T>
322where
323    F: Future<Output = Result<T, bollard::errors::Error>>,
324{
325    let timeout = remaining(deadline)?;
326    tokio::select! {
327        () = cancellation.cancelled() => Err(MutationFailure::new(
328            MutationSendState::Unknown,
329            InfraError::Docker(format!("container recreate cancelled after stage {stage:?}")),
330        )),
331        result = tokio::time::timeout(timeout, future) => match result {
332            Err(_) => Err(MutationFailure::new(
333                MutationSendState::Unknown,
334                InfraError::Docker(format!("container recreate timed out after stage {stage:?}")),
335            )),
336            Ok(Err(error)) => Err(MutationFailure::new(
337                MutationSendState::Unknown,
338                InfraError::Docker(format!("container recreate failed after stage {stage:?}: {error}")),
339            )),
340            Ok(Ok(value)) => Ok(value),
341        }
342    }
343}
344
345#[cfg(test)]
346#[path = "bollard_recreate_tests.rs"]
347mod tests;
soma_infra/bollard_recreate.rs

soma_infra/
bollard_recreate.rs
