soma_infra/
process_host_system.rs

1use std::path::{Component, Path};
2use std::sync::Arc;
3
4use async_trait::async_trait;
5use soma_fleet::{CommandExecutor, CommandOutput, CommandRequest, HostRecord};
6use soma_ops::Timestamp;
7use tokio_util::sync::CancellationToken;
8
9use crate::host_system_parse::{
10    parse_mounts, parse_network, parse_ports, parse_services, parse_usage,
11};
12use crate::{
13    DoctorCheck, DoctorReport, FilesystemUsage, HostSystemInspector, InfraError, InfraResult,
14    MountInfo, NetworkInterface, PortInfo, PortListRequest, ServiceListRequest, ServiceStatus,
15};
16
17const HOST_SYSTEM_OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
18
19/// Host-system inspector backed by a fleet command executor.
20pub struct CommandHostSystemInspector<E> {
21    executor: Arc<E>,
22}
23
24impl<E> CommandHostSystemInspector<E> {
25    /// Creates an inspector using the supplied executor.
26    #[must_use]
27    pub fn new(executor: Arc<E>) -> Self {
28        Self { executor }
29    }
30}
31
32#[async_trait]
33impl<E> HostSystemInspector for CommandHostSystemInspector<E>
34where
35    E: CommandExecutor,
36{
37    async fn services(
38        &self,
39        host: &HostRecord,
40        request: &ServiceListRequest,
41        cancellation: &CancellationToken,
42    ) -> InfraResult<Vec<ServiceStatus>> {
43        let raw = self
44            .run(
45                host,
46                "systemctl",
47                vec![
48                    "list-units".into(),
49                    "--type=service".into(),
50                    "--all".into(),
51                    "--no-legend".into(),
52                    "--no-pager".into(),
53                ],
54                request.deadline(),
55                cancellation,
56            )
57            .await?;
58        Ok(parse_services(&raw)
59            .into_iter()
60            .filter(|row| {
61                request
62                    .service()
63                    .is_none_or(|value| row.unit.contains(value))
64                    && request.state().is_none_or(|value| row.active == value)
65            })
66            .collect())
67    }
68
69    async fn network(
70        &self,
71        host: &HostRecord,
72        deadline: Timestamp,
73        cancellation: &CancellationToken,
74    ) -> InfraResult<Vec<NetworkInterface>> {
75        let raw = self
76            .run(
77                host,
78                "ip",
79                vec!["-j".into(), "address".into()],
80                deadline,
81                cancellation,
82            )
83            .await?;
84        parse_network(&raw)
85    }
86
87    async fn mounts(
88        &self,
89        host: &HostRecord,
90        deadline: Timestamp,
91        cancellation: &CancellationToken,
92    ) -> InfraResult<Vec<MountInfo>> {
93        let raw = self
94            .run(
95                host,
96                "findmnt",
97                vec![
98                    "-J".into(),
99                    "-b".into(),
100                    "-o".into(),
101                    "TARGET,SOURCE,FSTYPE,OPTIONS,SIZE,USED,AVAIL".into(),
102                ],
103                deadline,
104                cancellation,
105            )
106            .await?;
107        parse_mounts(&raw)
108    }
109
110    async fn ports(
111        &self,
112        host: &HostRecord,
113        request: &PortListRequest,
114        cancellation: &CancellationToken,
115    ) -> InfraResult<Vec<PortInfo>> {
116        let mut args = vec!["-H".into(), "-l".into(), "-n".into(), "-p".into()];
117        match request.protocol() {
118            Some(protocol) => args.push(protocol.as_ss_filter().into()),
119            None => {
120                args.push("-t".into());
121                args.push("-u".into());
122            }
123        }
124        let raw = self
125            .run(host, "ss", args, request.deadline(), cancellation)
126            .await?;
127        Ok(parse_ports(&raw)
128            .into_iter()
129            .skip(request.offset() as usize)
130            .take(request.limit() as usize)
131            .collect())
132    }
133
134    async fn filesystem_usage(
135        &self,
136        host: &HostRecord,
137        path: Option<&str>,
138        deadline: Timestamp,
139        cancellation: &CancellationToken,
140    ) -> InfraResult<FilesystemUsage> {
141        let mut args = vec![
142            "-B1".into(),
143            "--output=source,fstype,size,used,avail,pcent,target".into(),
144        ];
145        if let Some(path) = path {
146            validate_path(path)?;
147            args.push(path.to_owned());
148        }
149        let raw = self.run(host, "df", args, deadline, cancellation).await?;
150        parse_usage(&raw)
151    }
152
153    async fn doctor(
154        &self,
155        host: &HostRecord,
156        deadline: Timestamp,
157        cancellation: &CancellationToken,
158    ) -> InfraResult<DoctorReport> {
159        let mut checks = Vec::new();
160        checks.push(check(
161            "network",
162            self.network(host, deadline, cancellation)
163                .await
164                .map(|rows| format!("{} interface(s)", rows.len())),
165        ));
166        checks.push(check(
167            "services",
168            self.services(host, &ServiceListRequest::new(deadline), cancellation)
169                .await
170                .map(|rows| format!("{} service(s)", rows.len())),
171        ));
172        checks.push(check(
173            "storage",
174            self.filesystem_usage(host, None, deadline, cancellation)
175                .await
176                .map(|usage| format!("{}% used on {}", usage.usage_percent, usage.target)),
177        ));
178        let all_ok = checks.iter().all(|check| check.ok);
179        Ok(DoctorReport {
180            host: host.id().clone(),
181            topology_revision: host.revision().clone(),
182            overall: if all_ok { "ok" } else { "degraded" }.into(),
183            checks,
184        })
185    }
186}
187
188impl<E> CommandHostSystemInspector<E>
189where
190    E: CommandExecutor,
191{
192    async fn run(
193        &self,
194        host: &HostRecord,
195        program: &str,
196        args: Vec<String>,
197        deadline: Timestamp,
198        cancellation: &CancellationToken,
199    ) -> InfraResult<String> {
200        let request = CommandRequest::new(program, args, deadline)
201            .map_err(soma_fleet::FleetError::from)?
202            .with_output_limits(HOST_SYSTEM_OUTPUT_LIMIT, HOST_SYSTEM_OUTPUT_LIMIT)
203            .map_err(soma_fleet::FleetError::from)?;
204        let output = self.executor.execute(host, &request, cancellation).await?;
205        checked_output(host, "host-system", output)
206    }
207}
208
209fn checked_output(
210    host: &HostRecord,
211    domain: &'static str,
212    output: CommandOutput,
213) -> InfraResult<String> {
214    if output.exit_code() != Some(0) {
215        return Err(InfraError::CommandFailed {
216            domain,
217            host: host.id().clone(),
218            exit_code: output.exit_code(),
219            stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
220        });
221    }
222    if output.truncated() {
223        return Err(InfraError::Parse {
224            domain,
225            message: "bounded command output was truncated".into(),
226        });
227    }
228    String::from_utf8(output.stdout().to_vec()).map_err(|error| InfraError::Parse {
229        domain,
230        message: format!("command output was not UTF-8: {error}"),
231    })
232}
233
234fn validate_path(value: &str) -> InfraResult<()> {
235    let path = Path::new(value);
236    if !path.is_absolute()
237        || path
238            .components()
239            .any(|part| matches!(part, Component::ParentDir | Component::CurDir))
240    {
241        Err(InfraError::InvalidRequest {
242            domain: "filesystem",
243            message: format!("path must be absolute and normalized: {value}"),
244        })
245    } else {
246        Ok(())
247    }
248}
249
250fn check(name: &str, result: InfraResult<String>) -> DoctorCheck {
251    match result {
252        Ok(summary) => DoctorCheck {
253            name: name.into(),
254            ok: true,
255            summary,
256        },
257        Err(error) => DoctorCheck {
258            name: name.into(),
259            ok: false,
260            summary: error.to_string(),
261        },
262    }
263}
264
265#[cfg(test)]
266#[path = "process_host_system_tests.rs"]
267mod tests;
soma_infra/process_host_system.rs

soma_infra/
process_host_system.rs
