soma_infra/
process_process.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use soma_fleet::{CommandExecutor, CommandRequest, HostRecord};
5use tokio_util::sync::CancellationToken;
6
7use crate::process::parse_process_rows;
8use crate::{InfraError, InfraResult, ProcessInspector, ProcessListRequest, ProcessSnapshot};
9
10const PROCESS_OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
11
12/// Process inspector backed by a fleet command executor.
13pub struct CommandProcessInspector<E> {
14    executor: Arc<E>,
15}
16
17impl<E> CommandProcessInspector<E> {
18    /// Creates an inspector using the supplied fleet executor.
19    #[must_use]
20    pub fn new(executor: Arc<E>) -> Self {
21        Self { executor }
22    }
23}
24
25#[async_trait]
26impl<E> ProcessInspector for CommandProcessInspector<E>
27where
28    E: CommandExecutor,
29{
30    async fn list_processes(
31        &self,
32        host: &HostRecord,
33        request: &ProcessListRequest,
34        cancellation: &CancellationToken,
35    ) -> InfraResult<ProcessSnapshot> {
36        let command = CommandRequest::new(
37            "ps",
38            ["aux", "--sort", request.sort().ps_argument()],
39            request.deadline(),
40        )
41        .map_err(soma_fleet::FleetError::from)?
42        .with_output_limits(PROCESS_OUTPUT_LIMIT, PROCESS_OUTPUT_LIMIT)
43        .map_err(soma_fleet::FleetError::from)?;
44        let output = self.executor.execute(host, &command, cancellation).await?;
45        if output.truncated() {
46            return Err(InfraError::InvalidRequest {
47                domain: "process",
48                message: format!("process output exceeded {PROCESS_OUTPUT_LIMIT} bytes"),
49            });
50        }
51        if output.exit_code() != Some(0) {
52            return Err(InfraError::CommandFailed {
53                domain: "process",
54                host: host.id().clone(),
55                exit_code: output.exit_code(),
56                stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
57            });
58        }
59        let stdout = std::str::from_utf8(output.stdout()).map_err(|error| InfraError::Parse {
60            domain: "process",
61            message: format!("process output is not UTF-8: {error}"),
62        })?;
63        let mut lines = stdout.lines();
64        let _header = lines.next();
65        parse_process_rows(
66            host,
67            request,
68            &lines.collect::<Vec<_>>().join(
69                "
70",
71            ),
72        )
73    }
74}
75
76#[cfg(test)]
77#[path = "process_process_tests.rs"]
78mod tests;
soma_infra/process_process.rs

soma_infra/
process_process.rs
