soma_infra/
process_compose.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use soma_fleet::{CommandExecutor, CommandOutput, CommandRequest, HostRecord};
5use soma_ops::Timestamp;
6use tokio_util::sync::CancellationToken;
7
8use crate::compose_parse::{parse_config, parse_project_list, parse_status, validate_service};
9use crate::{
10    ComposeConfig, ComposeInspector, ComposeLogRequest, ComposeLogs, ComposeProject,
11    ComposeProjectRef, ComposeStatus, InfraError, InfraResult,
12};
13
14const COMPOSE_OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
15
16/// Compose inspector backed by a `soma-fleet` command executor.
17pub struct CommandComposeInspector<E> {
18    pub(crate) executor: Arc<E>,
19}
20
21impl<E> CommandComposeInspector<E> {
22    /// Creates a Compose inspector using the supplied fleet executor.
23    #[must_use]
24    pub fn new(executor: Arc<E>) -> Self {
25        Self { executor }
26    }
27}
28
29#[async_trait]
30impl<E> ComposeInspector for CommandComposeInspector<E>
31where
32    E: CommandExecutor,
33{
34    async fn list_projects(
35        &self,
36        host: &HostRecord,
37        deadline: Timestamp,
38        cancellation: &CancellationToken,
39    ) -> InfraResult<Vec<ComposeProject>> {
40        let raw = self
41            .run(
42                host,
43                ["compose", "ls", "--format", "json"],
44                deadline,
45                cancellation,
46            )
47            .await?;
48        parse_project_list(host, &raw)
49    }
50
51    async fn status(
52        &self,
53        host: &HostRecord,
54        project: &ComposeProjectRef,
55        service: Option<&str>,
56        deadline: Timestamp,
57        cancellation: &CancellationToken,
58    ) -> InfraResult<ComposeStatus> {
59        let config = project
60            .config_file()
61            .to_str()
62            .expect("ComposeProjectRef validates UTF-8 paths")
63            .to_owned();
64        let mut args = vec![
65            "compose".to_owned(),
66            "-f".to_owned(),
67            config,
68            "--project-name".to_owned(),
69            project.name().to_owned(),
70            "ps".to_owned(),
71            "--format".to_owned(),
72            "json".to_owned(),
73        ];
74        if let Some(service) = service {
75            validate_service(service)?;
76            args.push("--".into());
77            args.push(service.to_owned());
78        }
79        let raw = self.run_owned(host, args, deadline, cancellation).await?;
80        parse_status(host, project, &raw)
81    }
82
83    async fn config(
84        &self,
85        host: &HostRecord,
86        project: &ComposeProjectRef,
87        deadline: Timestamp,
88        cancellation: &CancellationToken,
89    ) -> InfraResult<ComposeConfig> {
90        let raw = self
91            .run_owned(
92                host,
93                vec![
94                    "compose".into(),
95                    "-f".into(),
96                    project
97                        .config_file()
98                        .to_str()
99                        .expect("ComposeProjectRef validates UTF-8 paths")
100                        .to_owned(),
101                    "--project-name".into(),
102                    project.name().to_owned(),
103                    "config".into(),
104                    "--format".into(),
105                    "json".into(),
106                ],
107                deadline,
108                cancellation,
109            )
110            .await?;
111        parse_config(host, project, &raw)
112    }
113
114    async fn logs(
115        &self,
116        host: &HostRecord,
117        project: &ComposeProjectRef,
118        request: &ComposeLogRequest,
119        cancellation: &CancellationToken,
120    ) -> InfraResult<ComposeLogs> {
121        let mut args = vec![
122            "compose".into(),
123            "-f".into(),
124            project
125                .config_file()
126                .to_str()
127                .expect("ComposeProjectRef validates UTF-8 paths")
128                .to_owned(),
129            "--project-name".into(),
130            project.name().to_owned(),
131            "logs".into(),
132            "--no-color".into(),
133            "--tail".into(),
134            request.lines().to_string(),
135        ];
136        if let Some(since) = request.since() {
137            args.extend(["--since".into(), since.to_owned()]);
138        }
139        if let Some(service) = request.service() {
140            validate_service(service)?;
141            args.extend(["--".into(), service.to_owned()]);
142        }
143        let output = self
144            .execute_owned(host, args, request.deadline(), cancellation)
145            .await?;
146        if output.exit_code() != Some(0) {
147            return Err(InfraError::CommandFailed {
148                domain: "compose",
149                host: host.id().clone(),
150                exit_code: output.exit_code(),
151                stderr: crate::error::public_diagnostic(output.stderr()),
152            });
153        }
154        let text = std::str::from_utf8(output.stdout()).map_err(|error| InfraError::Parse {
155            domain: "compose",
156            message: format!("Compose log output was not UTF-8: {error}"),
157        })?;
158        Ok(ComposeLogs {
159            host: host.id().clone(),
160            topology_revision: host.revision().clone(),
161            project: project.name().to_owned(),
162            lines: text.lines().map(str::to_owned).collect(),
163            truncated: output.truncated(),
164        })
165    }
166}
167
168impl<E> CommandComposeInspector<E>
169where
170    E: CommandExecutor,
171{
172    async fn run<const N: usize>(
173        &self,
174        host: &HostRecord,
175        args: [&str; N],
176        deadline: Timestamp,
177        cancellation: &CancellationToken,
178    ) -> InfraResult<String> {
179        self.run_owned(
180            host,
181            args.into_iter().map(str::to_owned).collect(),
182            deadline,
183            cancellation,
184        )
185        .await
186    }
187
188    async fn run_owned(
189        &self,
190        host: &HostRecord,
191        args: Vec<String>,
192        deadline: Timestamp,
193        cancellation: &CancellationToken,
194    ) -> InfraResult<String> {
195        let output = self
196            .execute_owned(host, args, deadline, cancellation)
197            .await?;
198        checked_output(host, output)
199    }
200
201    async fn execute_owned(
202        &self,
203        host: &HostRecord,
204        args: Vec<String>,
205        deadline: Timestamp,
206        cancellation: &CancellationToken,
207    ) -> InfraResult<CommandOutput> {
208        let request = CommandRequest::new("docker", args, deadline)
209            .map_err(soma_fleet::FleetError::from)?
210            .with_output_limits(COMPOSE_OUTPUT_LIMIT, COMPOSE_OUTPUT_LIMIT)
211            .map_err(soma_fleet::FleetError::from)?;
212        self.executor
213            .execute(host, &request, cancellation)
214            .await
215            .map_err(InfraError::from)
216    }
217}
218
219fn checked_output(host: &HostRecord, output: CommandOutput) -> InfraResult<String> {
220    if output.exit_code() != Some(0) {
221        return Err(InfraError::CommandFailed {
222            domain: "compose",
223            host: host.id().clone(),
224            exit_code: output.exit_code(),
225            stderr: crate::error::public_diagnostic(output.stderr()),
226        });
227    }
228    if output.truncated() {
229        return Err(InfraError::Parse {
230            domain: "compose",
231            message: "bounded Compose output was truncated".into(),
232        });
233    }
234    String::from_utf8(output.stdout().to_vec()).map_err(|error| InfraError::Parse {
235        domain: "compose",
236        message: format!("Compose output was not UTF-8: {error}"),
237    })
238}
239
240#[cfg(test)]
241#[path = "process_compose_tests.rs"]
242mod tests;
soma_infra/process_compose.rs

soma_infra/
process_compose.rs
