soma_infra/
process_image_build.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
5use soma_ops::{MutationSendState, ProgressEvent, Timestamp};
6use tokio_util::sync::CancellationToken;
7
8use crate::{
9    ImageBuildMutator, ImageBuildReceipt, ImageBuildRequest, InfraError, MutationFailure,
10    MutationProgressReporter, MutationResult,
11};
12
13const BUILD_OUTPUT_LIMIT: usize = 16 * 1024 * 1024;
14const MAX_PROGRESS_ERRORS: usize = 16;
15
16/// Process-backed Docker build driver.
17pub struct CommandImageBuildMutator<E> {
18    executor: Arc<E>,
19}
20
21impl<E> CommandImageBuildMutator<E> {
22    /// Creates a build driver from a fleet command executor.
23    #[must_use]
24    pub fn new(executor: Arc<E>) -> Self {
25        Self { executor }
26    }
27}
28
29#[async_trait]
30impl<E> ImageBuildMutator for CommandImageBuildMutator<E>
31where
32    E: CommandExecutor,
33{
34    async fn build_image(
35        &self,
36        host: &HostRecord,
37        request: &ImageBuildRequest,
38        progress: &dyn MutationProgressReporter,
39        cancellation: &CancellationToken,
40    ) -> MutationResult<ImageBuildReceipt> {
41        ensure_admitted(request.deadline(), cancellation)?;
42        let mut errors = Vec::new();
43        report(
44            progress,
45            request,
46            1,
47            "build",
48            "starting Docker image build",
49            &mut errors,
50        );
51        let mut args = vec![
52            "build".into(),
53            "--progress=plain".into(),
54            "-t".into(),
55            request.tag().into(),
56        ];
57        if request.no_cache() {
58            args.push("--no-cache".into());
59        }
60        if let Some(dockerfile) = request.dockerfile() {
61            args.push("-f".into());
62            args.push(
63                request
64                    .context()
65                    .join(dockerfile)
66                    .to_string_lossy()
67                    .into_owned(),
68            );
69        }
70        args.push(request.context().to_string_lossy().into_owned());
71        let command = CommandRequest::new("docker", args, request.deadline())
72            .map_err(soma_fleet::FleetError::from)
73            .and_then(|request| {
74                request
75                    .with_output_limits(BUILD_OUTPUT_LIMIT, BUILD_OUTPUT_LIMIT)
76                    .map_err(soma_fleet::FleetError::from)
77            })
78            .map_err(|error| {
79                MutationFailure::new(MutationSendState::NotSent, InfraError::from(error))
80            })?;
81        let output = self
82            .executor
83            .execute(host, &command, cancellation)
84            .await
85            .map_err(|error| {
86                MutationFailure::new(MutationSendState::Unknown, InfraError::from(error))
87            })?;
88        report(
89            progress,
90            request,
91            2,
92            "build",
93            "Docker image build process completed",
94            &mut errors,
95        );
96        if output.exit_code() != Some(0) {
97            return Err(MutationFailure::new(
98                MutationSendState::Sent,
99                InfraError::CommandFailed {
100                    domain: "image-build",
101                    host: host.id().clone(),
102                    exit_code: output.exit_code(),
103                    stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
104                },
105            ));
106        }
107        Ok(ImageBuildReceipt {
108            host: host.id().clone(),
109            topology_revision: host.revision().clone(),
110            tag: request.tag().into(),
111            send_state: MutationSendState::Sent,
112            stdout: String::from_utf8_lossy(output.stdout()).into_owned(),
113            stderr: String::from_utf8_lossy(output.stderr()).into_owned(),
114            output_truncated: output.truncated(),
115            progress_delivery_errors: errors,
116        })
117    }
118}
119
120fn report(
121    sink: &dyn MutationProgressReporter,
122    request: &ImageBuildRequest,
123    sequence: u64,
124    phase: &str,
125    message: &str,
126    errors: &mut Vec<String>,
127) {
128    let event = ProgressEvent::new(
129        request.operation_id().clone(),
130        request.operation().clone(),
131        sequence,
132        Timestamp::now(),
133        phase,
134    )
135    .and_then(|event| event.with_message(message));
136    if let Ok(event) = event
137        && let Err(error) = sink.report(&event)
138        && errors.len() < MAX_PROGRESS_ERRORS
139    {
140        errors.push(error);
141    }
142}
143fn ensure_admitted(deadline: Timestamp, cancellation: &CancellationToken) -> MutationResult<()> {
144    if cancellation.is_cancelled() {
145        return Err(MutationFailure::new(
146            MutationSendState::NotSent,
147            soma_fleet::FleetError::Cancelled.into(),
148        ));
149    }
150    if Timestamp::now() >= deadline {
151        return Err(MutationFailure::new(
152            MutationSendState::NotSent,
153            soma_fleet::FleetError::DeadlineExceeded.into(),
154        ));
155    }
156    Ok(())
157}
158
159#[cfg(test)]
160#[path = "process_image_build_tests.rs"]
161mod tests;
soma_infra/process_image_build.rs

soma_infra/
process_image_build.rs
