soma_infra/
process_compose_mutation.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, ComposeMutationAction, ComposeMutationReceipt, ComposeMutationRequest,
8    ComposeMutator, InfraError, MutationFailure, MutationResult,
9};
10
11const MUTATION_OUTPUT_LIMIT: usize = 1024 * 1024;
12
13#[async_trait]
14impl<E> ComposeMutator for CommandComposeInspector<E>
15where
16    E: CommandExecutor,
17{
18    async fn mutate_compose(
19        &self,
20        host: &HostRecord,
21        request: &ComposeMutationRequest,
22        cancellation: &CancellationToken,
23    ) -> MutationResult<ComposeMutationReceipt> {
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            request.action().action_label().into(),
34        ];
35        if request.action() == ComposeMutationAction::Up {
36            args.push("-d".into());
37        }
38        let command = CommandRequest::new("docker", args, request.deadline())
39            .map_err(soma_fleet::FleetError::from)
40            .and_then(|request| {
41                request
42                    .with_output_limits(MUTATION_OUTPUT_LIMIT, MUTATION_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-mutation",
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(ComposeMutationReceipt {
67            host: host.id().clone(),
68            topology_revision: host.revision().clone(),
69            project: request.project().name().to_owned(),
70            action: request.action(),
71            send_state: MutationSendState::Sent,
72        })
73    }
74}
75
76fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
77    if cancellation.is_cancelled() {
78        return Err(MutationFailure::new(
79            MutationSendState::NotSent,
80            soma_fleet::FleetError::Cancelled.into(),
81        ));
82    }
83    if Timestamp::now() >= deadline {
84        return Err(MutationFailure::new(
85            MutationSendState::NotSent,
86            soma_fleet::FleetError::DeadlineExceeded.into(),
87        ));
88    }
89    Ok(())
90}
91
92#[cfg(test)]
93#[path = "process_compose_mutation_tests.rs"]
94mod tests;
soma_infra/process_compose_mutation.rs

soma_infra/
process_compose_mutation.rs
