soma_infra/
logs.rs

1use std::path::PathBuf;
2
3use async_trait::async_trait;
4use serde::{Deserialize, Serialize};
5use soma_fleet::{HostId, HostRecord, TopologyRevision};
6use soma_ops::Timestamp;
7use tokio_util::sync::CancellationToken;
8
9use crate::{InfraError, InfraResult};
10
11const MAX_LINES: u32 = 500;
12const MAX_FILTER_CHARS: usize = 1024;
13const MAX_UNIT_CHARS: usize = 256;
14const MAX_TIME_CHARS: usize = 64;
15
16/// Supported read-only operating-system log sources.
17#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
18#[serde(rename_all = "snake_case")]
19pub enum LogSource {
20    /// Traditional system log, with messages fallback.
21    Syslog,
22    /// systemd journal.
23    Journal,
24    /// Kernel ring buffer.
25    Dmesg,
26    /// Authentication log, with secure fallback.
27    Auth,
28}
29
30/// Journal priority accepted by journalctl.
31#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(rename_all = "snake_case")]
33pub enum JournalPriority {
34    /// Emergency.
35    Emerg,
36    /// Alert.
37    Alert,
38    /// Critical.
39    Crit,
40    /// Error.
41    Err,
42    /// Warning.
43    Warning,
44    /// Notice.
45    Notice,
46    /// Informational.
47    Info,
48    /// Debug.
49    Debug,
50}
51
52impl JournalPriority {
53    #[cfg(any(feature = "process-driver", test))]
54    pub(crate) const fn as_arg(self) -> &'static str {
55        match self {
56            Self::Emerg => "emerg",
57            Self::Alert => "alert",
58            Self::Crit => "crit",
59            Self::Err => "err",
60            Self::Warning => "warning",
61            Self::Notice => "notice",
62            Self::Info => "info",
63            Self::Debug => "debug",
64        }
65    }
66}
67
68/// Validated journal filters.
69#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
70pub struct JournalFilters {
71    unit: Option<String>,
72    priority: Option<JournalPriority>,
73    since: Option<String>,
74    until: Option<String>,
75}
76
77impl JournalFilters {
78    /// Adds a journal unit filter.
79    pub fn with_unit(mut self, unit: impl Into<String>) -> InfraResult<Self> {
80        let unit = unit.into();
81        if unit.is_empty()
82            || unit.starts_with('-')
83            || unit.chars().count() > MAX_UNIT_CHARS
84            || unit.chars().any(char::is_control)
85        {
86            return Err(InfraError::InvalidRequest {
87                domain: "logs",
88                message: "journal unit must be 1-256 printable characters and not start with '-'"
89                    .into(),
90            });
91        }
92        self.unit = Some(unit);
93        Ok(self)
94    }
95
96    /// Adds a journal priority filter.
97    #[must_use]
98    pub const fn with_priority(mut self, priority: JournalPriority) -> Self {
99        self.priority = Some(priority);
100        self
101    }
102
103    /// Adds a journal lower time bound.
104    pub fn with_since(mut self, since: impl Into<String>) -> InfraResult<Self> {
105        self.since = Some(validate_time_filter(since.into())?);
106        Ok(self)
107    }
108
109    /// Adds a journal upper time bound.
110    pub fn with_until(mut self, until: impl Into<String>) -> InfraResult<Self> {
111        self.until = Some(validate_time_filter(until.into())?);
112        Ok(self)
113    }
114
115    /// Returns the optional unit filter.
116    #[must_use]
117    pub fn unit(&self) -> Option<&str> {
118        self.unit.as_deref()
119    }
120
121    /// Returns the optional priority.
122    #[must_use]
123    pub const fn priority(&self) -> Option<JournalPriority> {
124        self.priority
125    }
126
127    /// Returns the optional lower time bound.
128    #[must_use]
129    pub fn since(&self) -> Option<&str> {
130        self.since.as_deref()
131    }
132
133    /// Returns the optional upper time bound.
134    #[must_use]
135    pub fn until(&self) -> Option<&str> {
136        self.until.as_deref()
137    }
138}
139
140/// Bounded read request for one log source.
141#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
142pub struct LogReadRequest {
143    source: LogSource,
144    lines: u32,
145    grep: Option<String>,
146    journal: JournalFilters,
147    deadline: Timestamp,
148}
149
150impl LogReadRequest {
151    /// Creates a request for 100 lines from the selected source.
152    #[must_use]
153    pub const fn new(source: LogSource, deadline: Timestamp) -> Self {
154        Self {
155            source,
156            lines: 100,
157            grep: None,
158            journal: JournalFilters {
159                unit: None,
160                priority: None,
161                since: None,
162                until: None,
163            },
164            deadline,
165        }
166    }
167
168    /// Sets the maximum returned line count.
169    pub fn with_lines(mut self, lines: u32) -> InfraResult<Self> {
170        if lines == 0 || lines > MAX_LINES {
171            return Err(InfraError::InvalidRequest {
172                domain: "logs",
173                message: format!("line count must be 1-{MAX_LINES}"),
174            });
175        }
176        self.lines = lines;
177        Ok(self)
178    }
179
180    /// Adds a case-sensitive local substring filter.
181    pub fn with_grep(mut self, grep: impl Into<String>) -> InfraResult<Self> {
182        self.grep = Some(validate_filter("grep", grep.into(), MAX_FILTER_CHARS)?);
183        Ok(self)
184    }
185
186    /// Sets journal-specific filters.
187    pub fn with_journal_filters(mut self, filters: JournalFilters) -> InfraResult<Self> {
188        if self.source != LogSource::Journal {
189            return Err(InfraError::InvalidRequest {
190                domain: "logs",
191                message: "journal filters require the journal source".into(),
192            });
193        }
194        self.journal = filters;
195        Ok(self)
196    }
197
198    /// Returns the source.
199    #[must_use]
200    pub const fn source(&self) -> LogSource {
201        self.source
202    }
203
204    /// Returns the line limit.
205    #[must_use]
206    pub const fn lines(&self) -> u32 {
207        self.lines
208    }
209
210    /// Returns the optional local substring filter.
211    #[must_use]
212    pub fn grep(&self) -> Option<&str> {
213        self.grep.as_deref()
214    }
215
216    /// Returns journal-specific filters.
217    #[must_use]
218    pub const fn journal(&self) -> &JournalFilters {
219        &self.journal
220    }
221
222    /// Returns the absolute deadline.
223    #[must_use]
224    pub const fn deadline(&self) -> Timestamp {
225        self.deadline
226    }
227}
228
229/// Structured permission diagnostic for a log source.
230#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
231pub struct LogPermissionDiagnostic {
232    /// Driver-safe failure detail.
233    pub message: String,
234    /// Operator guidance.
235    pub help: String,
236}
237
238/// Bounded log read result.
239#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
240pub struct LogRead {
241    /// Target host.
242    pub host: HostId,
243    /// Exact topology revision.
244    pub topology_revision: TopologyRevision,
245    /// Selected source.
246    pub source: LogSource,
247    /// Source path when file-backed.
248    pub source_path: Option<PathBuf>,
249    /// Filtered result lines.
250    pub lines: Vec<String>,
251    /// Whether the output byte ceiling or line limit omitted data.
252    pub truncated: bool,
253    /// Permission diagnostic for restricted sources such as dmesg.
254    pub permission: Option<LogPermissionDiagnostic>,
255}
256
257/// Product-neutral operating-system log reader.
258#[async_trait]
259pub trait LogReader: Send + Sync {
260    /// Reads one bounded log source.
261    async fn read_logs(
262        &self,
263        host: &HostRecord,
264        request: &LogReadRequest,
265        cancellation: &CancellationToken,
266    ) -> InfraResult<LogRead>;
267}
268
269#[cfg(any(feature = "process-driver", test))]
270pub(crate) fn filtered_tail(raw: &str, grep: Option<&str>, limit: u32) -> (Vec<String>, bool) {
271    let mut lines = raw
272        .lines()
273        .filter(|line| grep.is_none_or(|pattern| line.contains(pattern)))
274        .map(str::to_owned)
275        .collect::<Vec<_>>();
276    let truncated = lines.len() > limit as usize;
277    if truncated {
278        lines = lines.split_off(lines.len() - limit as usize);
279    }
280    (lines, truncated)
281}
282
283fn validate_time_filter(value: String) -> InfraResult<String> {
284    let option_like = value.starts_with("--")
285        || (value.starts_with('-')
286            && !value[1..]
287                .chars()
288                .next()
289                .is_some_and(|character| character.is_ascii_digit()));
290    if value.is_empty()
291        || option_like
292        || value.chars().count() > MAX_TIME_CHARS
293        || value.chars().any(char::is_control)
294    {
295        Err(InfraError::InvalidRequest {
296            domain: "logs",
297            message: "journal time filter is invalid or option-like".into(),
298        })
299    } else {
300        Ok(value)
301    }
302}
303
304fn validate_filter(name: &'static str, value: String, max: usize) -> InfraResult<String> {
305    let count = value.chars().count();
306    if count == 0 || count > max || value.chars().any(char::is_control) {
307        Err(InfraError::InvalidRequest {
308            domain: "logs",
309            message: format!("{name} must contain 1-{max} printable characters"),
310        })
311    } else {
312        Ok(value)
313    }
314}
315
316#[cfg(test)]
317#[path = "logs_tests.rs"]
318mod tests;
soma_infra/logs.rs

soma_infra/
logs.rs
