soma_infra/
process_compose_pull.rs

1use async_trait::async_trait;
2use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
3use soma_ops::{MutationSendState, ProgressEvent, Timestamp};
4use tokio_util::sync::CancellationToken;
5
6use crate::{
7    CommandComposeInspector, ComposePullMutator, ComposePullReceipt, ComposePullRequest,
8    InfraError, MutationFailure, MutationProgressReporter, MutationResult,
9};
10
11const PULL_OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
12const MAX_PROGRESS_ERRORS: usize = 16;
13
14#[async_trait]
15impl<E> ComposePullMutator for CommandComposeInspector<E>
16where
17    E: CommandExecutor,
18{
19    async fn pull_compose_images(
20        &self,
21        host: &HostRecord,
22        request: &ComposePullRequest,
23        progress: &dyn MutationProgressReporter,
24        cancellation: &CancellationToken,
25    ) -> MutationResult<ComposePullReceipt> {
26        ensure_admitted(request.deadline(), cancellation)?;
27        let mut progress_delivery_errors = Vec::new();
28        report_progress(
29            progress,
30            request,
31            1,
32            format!(
33                "pulling images for Compose project {}",
34                request.project().name()
35            ),
36            &mut progress_delivery_errors,
37        );
38        let mut args = vec![
39            "compose".into(),
40            "-f".into(),
41            request
42                .project()
43                .config_file()
44                .to_string_lossy()
45                .into_owned(),
46            "pull".into(),
47        ];
48        if let Some(service) = request.service() {
49            args.extend(["--".into(), service.to_owned()]);
50        }
51        let command = CommandRequest::new("docker", args, request.deadline())
52            .map_err(soma_fleet::FleetError::from)
53            .and_then(|request| {
54                request
55                    .with_output_limits(PULL_OUTPUT_LIMIT, PULL_OUTPUT_LIMIT)
56                    .map_err(soma_fleet::FleetError::from)
57            })
58            .map_err(|error| {
59                MutationFailure::new(MutationSendState::NotSent, InfraError::from(error))
60            })?;
61        let output = self
62            .executor
63            .execute(host, &command, cancellation)
64            .await
65            .map_err(|error| {
66                MutationFailure::new(MutationSendState::Unknown, InfraError::from(error))
67            })?;
68        if output.exit_code() != Some(0) {
69            return Err(MutationFailure::new(
70                MutationSendState::Sent,
71                InfraError::CommandFailed {
72                    domain: "compose-pull",
73                    host: host.id().clone(),
74                    exit_code: output.exit_code(),
75                    stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
76                },
77            ));
78        }
79        report_progress(
80            progress,
81            request,
82            2,
83            format!(
84                "Compose image pull completed for {}",
85                request.project().name()
86            ),
87            &mut progress_delivery_errors,
88        );
89        Ok(ComposePullReceipt {
90            host: host.id().clone(),
91            topology_revision: host.revision().clone(),
92            project: request.project().name().to_owned(),
93            service: request.service().map(str::to_owned),
94            send_state: MutationSendState::Sent,
95            progress_delivery_errors,
96            output_truncated: output.truncated(),
97        })
98    }
99}
100
101fn report_progress(
102    sink: &dyn MutationProgressReporter,
103    request: &ComposePullRequest,
104    sequence: u64,
105    message: String,
106    errors: &mut Vec<String>,
107) {
108    let event = ProgressEvent::new(
109        request.operation_id().clone(),
110        request.operation().clone(),
111        sequence,
112        Timestamp::now(),
113        "pull",
114    )
115    .and_then(|event| event.with_message(message));
116    if let Ok(event) = event
117        && let Err(error) = sink.report(&event)
118        && errors.len() < MAX_PROGRESS_ERRORS
119    {
120        errors.push(error);
121    }
122}
123
124fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
125    if cancellation.is_cancelled() {
126        return Err(MutationFailure::new(
127            MutationSendState::NotSent,
128            soma_fleet::FleetError::Cancelled.into(),
129        ));
130    }
131    if Timestamp::now() >= deadline {
132        return Err(MutationFailure::new(
133            MutationSendState::NotSent,
134            soma_fleet::FleetError::DeadlineExceeded.into(),
135        ));
136    }
137    Ok(())
138}
139
140#[cfg(test)]
141#[path = "process_compose_pull_tests.rs"]
142mod tests;
soma_infra/process_compose_pull.rs

soma_infra/
process_compose_pull.rs
