soma_infra/
host.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use soma_fleet::{
6    CommandExecutor, CommandOutput, CommandRequest, HostId, HostRecord, TopologyRevision,
7};
8use soma_ops::Timestamp;
9use tokio_util::sync::CancellationToken;
10
11use crate::{InfraError, InfraResult};
12
13const HOST_OUTPUT_LIMIT: usize = 64 * 1024;
14
15/// Deadline-bound request for one host inspection.
16#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
17pub struct HostInspectRequest {
18    deadline: Timestamp,
19}
20
21impl HostInspectRequest {
22    /// Creates a host inspection request.
23    #[must_use]
24    pub const fn new(deadline: Timestamp) -> Self {
25        Self { deadline }
26    }
27
28    /// Returns the absolute request deadline.
29    #[must_use]
30    pub const fn deadline(self) -> Timestamp {
31        self.deadline
32    }
33}
34
35/// Stable host identity fields collected from the operating system.
36#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
37pub struct HostIdentity {
38    /// Reported hostname.
39    pub hostname: String,
40    /// Operating-system family reported by `uname -s`.
41    pub operating_system: String,
42    /// Kernel release reported by `uname -r`.
43    pub kernel_release: String,
44    /// Machine architecture reported by `uname -m`.
45    pub architecture: String,
46}
47
48/// Parsed host memory counters.
49#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
50pub struct HostMemory {
51    /// Total physical memory in bytes.
52    pub total_bytes: u64,
53    /// Currently available memory in bytes.
54    pub available_bytes: u64,
55    /// Derived used memory in bytes.
56    pub used_bytes: u64,
57    /// Rounded integer utilization percentage.
58    pub usage_percent: u8,
59}
60
61/// Parsed Linux load averages.
62#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
63pub struct HostLoadAverage {
64    /// One-minute load average.
65    pub one: f64,
66    /// Five-minute load average.
67    pub five: f64,
68    /// Fifteen-minute load average.
69    pub fifteen: f64,
70}
71
72/// Complete read-only host inspection result.
73#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
74pub struct HostInspection {
75    /// Stable fleet host identity.
76    pub host: HostId,
77    /// Exact topology revision used for collection.
78    pub topology_revision: TopologyRevision,
79    /// Operating-system identity.
80    pub identity: HostIdentity,
81    /// Uptime in fractional seconds.
82    pub uptime_seconds: f64,
83    /// Memory counters.
84    pub memory: HostMemory,
85    /// Load averages.
86    pub load: HostLoadAverage,
87}
88
89/// Product-neutral host inspection engine.
90#[async_trait]
91pub trait HostInspector: Send + Sync {
92    /// Collects one typed snapshot from the exact host revision.
93    async fn inspect(
94        &self,
95        host: &HostRecord,
96        request: HostInspectRequest,
97        cancellation: &CancellationToken,
98    ) -> InfraResult<HostInspection>;
99}
100
101/// Host inspector backed by a `soma-fleet` command executor.
102pub struct LinuxCommandHostInspector<E> {
103    executor: Arc<E>,
104}
105
106impl<E> LinuxCommandHostInspector<E> {
107    /// Creates an inspector using the supplied fleet command executor.
108    #[must_use]
109    pub fn new(executor: Arc<E>) -> Self {
110        Self { executor }
111    }
112}
113
114#[async_trait]
115impl<E> HostInspector for LinuxCommandHostInspector<E>
116where
117    E: CommandExecutor,
118{
119    async fn inspect(
120        &self,
121        host: &HostRecord,
122        request: HostInspectRequest,
123        cancellation: &CancellationToken,
124    ) -> InfraResult<HostInspection> {
125        if cancellation.is_cancelled() {
126            return Err(soma_fleet::FleetError::Cancelled.into());
127        }
128        let hostname = self
129            .run_text(host, "hostname", &[], request.deadline, cancellation)
130            .await?;
131        let operating_system = self
132            .run_text(host, "uname", &["-s"], request.deadline, cancellation)
133            .await?;
134        let kernel_release = self
135            .run_text(host, "uname", &["-r"], request.deadline, cancellation)
136            .await?;
137        let architecture = self
138            .run_text(host, "uname", &["-m"], request.deadline, cancellation)
139            .await?;
140        let uptime = self
141            .run_text(
142                host,
143                "cat",
144                &["/proc/uptime"],
145                request.deadline,
146                cancellation,
147            )
148            .await?;
149        let meminfo = self
150            .run_text(
151                host,
152                "cat",
153                &["/proc/meminfo"],
154                request.deadline,
155                cancellation,
156            )
157            .await?;
158        let loadavg = self
159            .run_text(
160                host,
161                "cat",
162                &["/proc/loadavg"],
163                request.deadline,
164                cancellation,
165            )
166            .await?;
167
168        Ok(HostInspection {
169            host: host.id().clone(),
170            topology_revision: host.revision().clone(),
171            identity: HostIdentity {
172                hostname,
173                operating_system,
174                kernel_release,
175                architecture,
176            },
177            uptime_seconds: parse_uptime(&uptime)?,
178            memory: parse_meminfo(&meminfo)?,
179            load: parse_loadavg(&loadavg)?,
180        })
181    }
182}
183
184impl<E> LinuxCommandHostInspector<E>
185where
186    E: CommandExecutor,
187{
188    async fn run_text(
189        &self,
190        host: &HostRecord,
191        program: &str,
192        args: &[&str],
193        deadline: Timestamp,
194        cancellation: &CancellationToken,
195    ) -> InfraResult<String> {
196        let request = CommandRequest::new(program, args.iter().copied(), deadline)
197            .map_err(soma_fleet::FleetError::from)?
198            .with_output_limits(HOST_OUTPUT_LIMIT, HOST_OUTPUT_LIMIT)
199            .map_err(soma_fleet::FleetError::from)?;
200        let output = self.executor.execute(host, &request, cancellation).await?;
201        checked_text(host, output)
202    }
203}
204
205fn checked_text(host: &HostRecord, output: CommandOutput) -> InfraResult<String> {
206    if output.exit_code() != Some(0) {
207        return Err(InfraError::CommandFailed {
208            domain: "host",
209            host: host.id().clone(),
210            exit_code: output.exit_code(),
211            stderr: crate::error::public_diagnostic(output.stderr()),
212        });
213    }
214    if output.truncated() {
215        return Err(InfraError::Parse {
216            domain: "host",
217            message: "bounded command output was truncated".into(),
218        });
219    }
220    String::from_utf8(output.stdout().to_vec())
221        .map(|value| value.trim().to_owned())
222        .map_err(|error| InfraError::Parse {
223            domain: "host",
224            message: format!("output was not UTF-8: {error}"),
225        })
226}
227
228fn parse_uptime(raw: &str) -> InfraResult<f64> {
229    raw.split_whitespace()
230        .next()
231        .and_then(|value| value.parse::<f64>().ok())
232        .filter(|value| value.is_finite() && *value >= 0.0)
233        .ok_or_else(|| InfraError::Parse {
234            domain: "host",
235            message: "invalid /proc/uptime value".into(),
236        })
237}
238
239fn parse_meminfo(raw: &str) -> InfraResult<HostMemory> {
240    let mut total_kib = None;
241    let mut available_kib = None;
242    for line in raw.lines() {
243        let mut fields = line.split_whitespace();
244        let key = fields.next().unwrap_or_default().trim_end_matches(':');
245        let value = fields.next().and_then(|value| value.parse::<u64>().ok());
246        let unit = fields.next();
247        match key {
248            "MemTotal" if unit == Some("kB") => total_kib = value,
249            "MemAvailable" if unit == Some("kB") => available_kib = value,
250            _ => {}
251        }
252    }
253    let total_bytes = total_kib
254        .and_then(|value| value.checked_mul(1024))
255        .ok_or_else(|| InfraError::Parse {
256            domain: "host",
257            message: "missing or overflowing MemTotal".into(),
258        })?;
259    let available_bytes = available_kib
260        .and_then(|value| value.checked_mul(1024))
261        .ok_or_else(|| InfraError::Parse {
262            domain: "host",
263            message: "missing or overflowing MemAvailable".into(),
264        })?;
265    if total_bytes == 0 || available_bytes > total_bytes {
266        return Err(InfraError::Parse {
267            domain: "host",
268            message: "inconsistent MemTotal and MemAvailable".into(),
269        });
270    }
271    let used_bytes = total_bytes - available_bytes;
272    let usage_percent = ((used_bytes as f64 / total_bytes as f64) * 100.0)
273        .round()
274        .clamp(0.0, 100.0) as u8;
275    Ok(HostMemory {
276        total_bytes,
277        available_bytes,
278        used_bytes,
279        usage_percent,
280    })
281}
282
283fn parse_loadavg(raw: &str) -> InfraResult<HostLoadAverage> {
284    let values = raw
285        .split_whitespace()
286        .take(3)
287        .map(str::parse::<f64>)
288        .collect::<Result<Vec<_>, _>>()
289        .map_err(|error| InfraError::Parse {
290            domain: "host",
291            message: format!("invalid /proc/loadavg value: {error}"),
292        })?;
293    if values.len() != 3
294        || values
295            .iter()
296            .any(|value| !value.is_finite() || *value < 0.0)
297    {
298        return Err(InfraError::Parse {
299            domain: "host",
300            message: "expected three non-negative load averages".into(),
301        });
302    }
303    Ok(HostLoadAverage {
304        one: values[0],
305        five: values[1],
306        fifteen: values[2],
307    })
308}
309
310#[cfg(test)]
311#[path = "host_tests.rs"]
312mod tests;
soma_infra/host.rs

soma_infra/
host.rs
