soma_infra/
process_compose_down.rs

1use async_trait::async_trait;
2use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
3use soma_ops::{MutationSendState, Timestamp};
4use tokio_util::sync::CancellationToken;
5
6use crate::{
7    CommandComposeInspector, ComposeDownMutator, ComposeDownReceipt, ComposeDownRequest,
8    InfraError, MutationFailure, MutationResult,
9};
10
11const OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
12
13#[async_trait]
14impl<E> ComposeDownMutator for CommandComposeInspector<E>
15where
16    E: CommandExecutor,
17{
18    async fn down_compose(
19        &self,
20        host: &HostRecord,
21        request: &ComposeDownRequest,
22        cancellation: &CancellationToken,
23    ) -> MutationResult<ComposeDownReceipt> {
24        ensure_admitted(request.deadline(), cancellation)?;
25        let mut args = vec![
26            "compose".into(),
27            "-f".into(),
28            request
29                .project()
30                .config_file()
31                .to_string_lossy()
32                .into_owned(),
33            "down".into(),
34        ];
35        if request.remove_volumes() {
36            args.push("--volumes".into());
37        }
38        let command = CommandRequest::new("docker", args, request.deadline())
39            .map_err(soma_fleet::FleetError::from)
40            .and_then(|command| {
41                command
42                    .with_output_limits(OUTPUT_LIMIT, OUTPUT_LIMIT)
43                    .map_err(soma_fleet::FleetError::from)
44            })
45            .map_err(|error| {
46                MutationFailure::new(MutationSendState::NotSent, InfraError::from(error))
47            })?;
48        let output = self
49            .executor
50            .execute(host, &command, cancellation)
51            .await
52            .map_err(|error| {
53                MutationFailure::new(MutationSendState::Unknown, InfraError::from(error))
54            })?;
55        if output.exit_code() != Some(0) {
56            return Err(MutationFailure::new(
57                MutationSendState::Sent,
58                InfraError::CommandFailed {
59                    domain: "compose-down",
60                    host: host.id().clone(),
61                    exit_code: output.exit_code(),
62                    stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
63                },
64            ));
65        }
66        Ok(ComposeDownReceipt {
67            host: host.id().clone(),
68            topology_revision: host.revision().clone(),
69            project: request.project().name().to_owned(),
70            remove_volumes: request.remove_volumes(),
71            send_state: MutationSendState::Sent,
72            stdout: String::from_utf8_lossy(output.stdout()).trim().to_owned(),
73            stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
74            output_truncated: output.truncated(),
75        })
76    }
77}
78
79fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
80    if cancellation.is_cancelled() {
81        return Err(MutationFailure::new(
82            MutationSendState::NotSent,
83            soma_fleet::FleetError::Cancelled.into(),
84        ));
85    }
86    if deadline <= Timestamp::now() {
87        return Err(MutationFailure::new(
88            MutationSendState::NotSent,
89            soma_fleet::FleetError::DeadlineExceeded.into(),
90        ));
91    }
92    Ok(())
93}
94
95#[cfg(test)]
96#[path = "process_compose_down_tests.rs"]
97mod tests;
soma_infra/process_compose_down.rs

soma_infra/
process_compose_down.rs
