soma_infra/
process.rs

1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3use soma_fleet::{HostId, HostRecord, TopologyRevision};
4use soma_ops::Timestamp;
5use tokio_util::sync::CancellationToken;
6
7use crate::{InfraError, InfraResult};
8
9const MAX_FILTER_CHARS: usize = 1024;
10const MAX_PROCESS_ROWS: u32 = 500;
11
12/// Supported deterministic process sort orders.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
14#[serde(rename_all = "snake_case")]
15pub enum ProcessSort {
16    /// Highest CPU utilization first.
17    #[default]
18    Cpu,
19    /// Highest memory utilization first.
20    Memory,
21    /// Lowest process identifier first.
22    Pid,
23    /// Greatest accumulated CPU time first.
24    Time,
25}
26
27impl ProcessSort {
28    #[cfg(any(feature = "process-driver", test))]
29    pub(crate) const fn ps_argument(self) -> &'static str {
30        match self {
31            Self::Cpu => "-cpu",
32            Self::Memory => "-mem",
33            Self::Pid => "pid",
34            Self::Time => "-time",
35        }
36    }
37}
38
39/// Closed request for one process snapshot.
40#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
41pub struct ProcessListRequest {
42    sort: ProcessSort,
43    grep: Option<String>,
44    user: Option<String>,
45    limit: u32,
46    deadline: Timestamp,
47}
48
49impl ProcessListRequest {
50    /// Creates a CPU-sorted request limited to 50 rows.
51    #[must_use]
52    pub const fn new(deadline: Timestamp) -> Self {
53        Self {
54            sort: ProcessSort::Cpu,
55            grep: None,
56            user: None,
57            limit: 50,
58            deadline,
59        }
60    }
61
62    /// Selects the process sort order.
63    #[must_use]
64    pub const fn with_sort(mut self, sort: ProcessSort) -> Self {
65        self.sort = sort;
66        self
67    }
68
69    /// Adds a case-sensitive substring filter over rendered command rows.
70    pub fn with_grep(mut self, value: impl Into<String>) -> InfraResult<Self> {
71        self.grep = Some(validate_filter("grep", value.into())?);
72        Ok(self)
73    }
74
75    /// Adds an exact user-column filter.
76    pub fn with_user(mut self, value: impl Into<String>) -> InfraResult<Self> {
77        self.user = Some(validate_filter("user", value.into())?);
78        Ok(self)
79    }
80
81    /// Sets the maximum returned row count.
82    pub fn with_limit(mut self, limit: u32) -> InfraResult<Self> {
83        if limit == 0 || limit > MAX_PROCESS_ROWS {
84            return Err(InfraError::InvalidRequest {
85                domain: "process",
86                message: format!("limit must be 1-{MAX_PROCESS_ROWS}"),
87            });
88        }
89        self.limit = limit;
90        Ok(self)
91    }
92
93    /// Returns the selected sort order.
94    #[must_use]
95    pub const fn sort(&self) -> ProcessSort {
96        self.sort
97    }
98
99    /// Returns the optional command substring filter.
100    #[must_use]
101    pub fn grep(&self) -> Option<&str> {
102        self.grep.as_deref()
103    }
104
105    /// Returns the optional user filter.
106    #[must_use]
107    pub fn user(&self) -> Option<&str> {
108        self.user.as_deref()
109    }
110
111    /// Returns the row limit.
112    #[must_use]
113    pub const fn limit(&self) -> u32 {
114        self.limit
115    }
116
117    /// Returns the absolute request deadline.
118    #[must_use]
119    pub const fn deadline(&self) -> Timestamp {
120        self.deadline
121    }
122}
123
124/// Typed row from a process snapshot.
125#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
126pub struct ProcessRow {
127    /// Process owner.
128    pub user: String,
129    /// Process identifier.
130    pub pid: u32,
131    /// CPU percentage reported by ps.
132    pub cpu_percent: f64,
133    /// Memory percentage reported by ps.
134    pub memory_percent: f64,
135    /// Virtual memory size in KiB.
136    pub virtual_size_kib: u64,
137    /// Resident memory size in KiB.
138    pub resident_size_kib: u64,
139    /// Controlling terminal.
140    pub tty: String,
141    /// Process state flags.
142    pub state: String,
143    /// Process start field reported by ps.
144    pub start: String,
145    /// Accumulated CPU time.
146    pub cpu_time: String,
147    /// Command and arguments.
148    pub command: String,
149}
150
151/// Bounded process snapshot for one host revision.
152#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
153pub struct ProcessSnapshot {
154    /// Target host.
155    pub host: HostId,
156    /// Exact topology revision.
157    pub topology_revision: TopologyRevision,
158    /// Applied sort order.
159    pub sort: ProcessSort,
160    /// Returned process rows.
161    pub rows: Vec<ProcessRow>,
162    /// Whether rows were omitted by the request limit.
163    pub truncated: bool,
164}
165
166/// Product-neutral process inspection engine.
167#[async_trait]
168pub trait ProcessInspector: Send + Sync {
169    /// Lists and parses a bounded process snapshot.
170    async fn list_processes(
171        &self,
172        host: &HostRecord,
173        request: &ProcessListRequest,
174        cancellation: &CancellationToken,
175    ) -> InfraResult<ProcessSnapshot>;
176}
177
178#[cfg(any(feature = "process-driver", test))]
179pub(crate) fn parse_process_rows(
180    host: &HostRecord,
181    request: &ProcessListRequest,
182    raw: &str,
183) -> InfraResult<ProcessSnapshot> {
184    let mut rows = raw
185        .lines()
186        .filter(|line| !line.trim().is_empty())
187        .map(parse_process_row)
188        .collect::<InfraResult<Vec<_>>>()?;
189
190    if let Some(user) = request.user() {
191        rows.retain(|row| row.user == user);
192    }
193    if let Some(pattern) = request.grep() {
194        rows.retain(|row| row.command.contains(pattern));
195    }
196
197    let truncated = rows.len() > request.limit() as usize;
198    rows.truncate(request.limit() as usize);
199    Ok(ProcessSnapshot {
200        host: host.id().clone(),
201        topology_revision: host.revision().clone(),
202        sort: request.sort(),
203        rows,
204        truncated,
205    })
206}
207
208#[cfg(any(feature = "process-driver", test))]
209fn parse_process_row(line: &str) -> InfraResult<ProcessRow> {
210    let mut fields = line.split_whitespace();
211    let user = next_field(&mut fields, "user")?.to_owned();
212    let pid = parse_field(next_field(&mut fields, "pid")?, "pid")?;
213    let cpu_percent = parse_field(next_field(&mut fields, "cpu")?, "cpu")?;
214    let memory_percent = parse_field(next_field(&mut fields, "memory")?, "memory")?;
215    let virtual_size_kib = parse_field(next_field(&mut fields, "vsz")?, "vsz")?;
216    let resident_size_kib = parse_field(next_field(&mut fields, "rss")?, "rss")?;
217    let tty = next_field(&mut fields, "tty")?.to_owned();
218    let state = next_field(&mut fields, "state")?.to_owned();
219    let start = next_field(&mut fields, "start")?.to_owned();
220    let cpu_time = next_field(&mut fields, "time")?.to_owned();
221    let command = fields.collect::<Vec<_>>().join(" ");
222    if command.is_empty() {
223        return Err(parse_error("process row has no command"));
224    }
225    Ok(ProcessRow {
226        user,
227        pid,
228        cpu_percent,
229        memory_percent,
230        virtual_size_kib,
231        resident_size_kib,
232        tty,
233        state,
234        start,
235        cpu_time,
236        command,
237    })
238}
239
240#[cfg(any(feature = "process-driver", test))]
241fn next_field<'a>(fields: &mut impl Iterator<Item = &'a str>, name: &str) -> InfraResult<&'a str> {
242    fields
243        .next()
244        .ok_or_else(|| parse_error(&format!("process row has no {name} field")))
245}
246
247#[cfg(any(feature = "process-driver", test))]
248fn parse_field<T>(value: &str, name: &str) -> InfraResult<T>
249where
250    T: std::str::FromStr,
251{
252    value
253        .parse()
254        .map_err(|_| parse_error(&format!("invalid process {name} field: {value:?}")))
255}
256
257fn validate_filter(name: &'static str, value: String) -> InfraResult<String> {
258    let count = value.chars().count();
259    if count == 0 || count > MAX_FILTER_CHARS || value.chars().any(char::is_control) {
260        Err(InfraError::InvalidRequest {
261            domain: "process",
262            message: format!("{name} must contain 1-{MAX_FILTER_CHARS} printable characters"),
263        })
264    } else {
265        Ok(value)
266    }
267}
268
269#[cfg(any(feature = "process-driver", test))]
270fn parse_error(message: &str) -> InfraError {
271    InfraError::Parse {
272        domain: "process",
273        message: message.to_owned(),
274    }
275}
276
277#[cfg(test)]
278#[path = "process_tests.rs"]
279mod tests;
soma_infra/process.rs

soma_infra/
process.rs
