soma_infra/
process_logs.rs

1use std::path::PathBuf;
2use std::sync::Arc;
3
4use async_trait::async_trait;
5use soma_fleet::{CommandExecutor, CommandOutput, CommandRequest, HostRecord};
6use tokio_util::sync::CancellationToken;
7
8use crate::logs::filtered_tail;
9use crate::{
10    InfraError, InfraResult, LogPermissionDiagnostic, LogRead, LogReadRequest, LogReader, LogSource,
11};
12
13const LOG_OUTPUT_LIMIT: usize = 4 * 1024 * 1024;
14
15/// Operating-system log reader backed by a fleet command executor.
16pub struct CommandLogReader<E> {
17    executor: Arc<E>,
18}
19
20impl<E> CommandLogReader<E> {
21    /// Creates a reader using the supplied fleet executor.
22    #[must_use]
23    pub fn new(executor: Arc<E>) -> Self {
24        Self { executor }
25    }
26}
27
28#[async_trait]
29impl<E> LogReader for CommandLogReader<E>
30where
31    E: CommandExecutor,
32{
33    async fn read_logs(
34        &self,
35        host: &HostRecord,
36        request: &LogReadRequest,
37        cancellation: &CancellationToken,
38    ) -> InfraResult<LogRead> {
39        match request.source() {
40            LogSource::Syslog => {
41                self.read_file_logs(
42                    host,
43                    request,
44                    "/var/log/syslog",
45                    "/var/log/messages",
46                    cancellation,
47                )
48                .await
49            }
50            LogSource::Auth => {
51                self.read_file_logs(
52                    host,
53                    request,
54                    "/var/log/auth.log",
55                    "/var/log/secure",
56                    cancellation,
57                )
58                .await
59            }
60            LogSource::Journal => self.read_journal(host, request, cancellation).await,
61            LogSource::Dmesg => self.read_dmesg(host, request, cancellation).await,
62        }
63    }
64}
65
66impl<E> CommandLogReader<E>
67where
68    E: CommandExecutor,
69{
70    async fn read_file_logs(
71        &self,
72        host: &HostRecord,
73        request: &LogReadRequest,
74        primary: &str,
75        fallback: &str,
76        cancellation: &CancellationToken,
77    ) -> InfraResult<LogRead> {
78        let line_count = request.lines().to_string();
79        let primary_output = self
80            .execute(
81                host,
82                "tail",
83                vec!["-n".into(), line_count.clone(), primary.into()],
84                request,
85                cancellation,
86            )
87            .await?;
88        let (output, source_path) = if primary_output.exit_code() == Some(0) {
89            (primary_output, PathBuf::from(primary))
90        } else if missing_file(&primary_output) {
91            let fallback_output = self
92                .execute(
93                    host,
94                    "tail",
95                    vec!["-n".into(), line_count, fallback.into()],
96                    request,
97                    cancellation,
98                )
99                .await?;
100            if fallback_output.exit_code() != Some(0) {
101                return Err(command_error(host, "logs", &fallback_output));
102            }
103            (fallback_output, PathBuf::from(fallback))
104        } else {
105            return Err(command_error(host, "logs", &primary_output));
106        };
107        self.render(host, request, output, Some(source_path), None)
108    }
109
110    async fn read_journal(
111        &self,
112        host: &HostRecord,
113        request: &LogReadRequest,
114        cancellation: &CancellationToken,
115    ) -> InfraResult<LogRead> {
116        let mut args = vec![
117            "-n".into(),
118            request.lines().to_string(),
119            "--no-pager".into(),
120        ];
121        let filters = request.journal();
122        if let Some(unit) = filters.unit() {
123            args.extend(["-u".into(), unit.into()]);
124        }
125        if let Some(priority) = filters.priority() {
126            args.extend(["-p".into(), priority.as_arg().into()]);
127        }
128        if let Some(since) = filters.since() {
129            args.extend(["--since".into(), since.into()]);
130        }
131        if let Some(until) = filters.until() {
132            args.extend(["--until".into(), until.into()]);
133        }
134        let output = self
135            .execute(host, "journalctl", args, request, cancellation)
136            .await?;
137        if output.exit_code() != Some(0) {
138            return Err(command_error(host, "logs", &output));
139        }
140        self.render(host, request, output, None, None)
141    }
142
143    async fn read_dmesg(
144        &self,
145        host: &HostRecord,
146        request: &LogReadRequest,
147        cancellation: &CancellationToken,
148    ) -> InfraResult<LogRead> {
149        let fetch_lines = request.lines().saturating_mul(4).min(1000);
150        let output = self
151            .execute(
152                host,
153                "dmesg",
154                vec![
155                    "--color=never".into(),
156                    "--lines".into(),
157                    fetch_lines.to_string(),
158                ],
159                request,
160                cancellation,
161            )
162            .await?;
163        if output.exit_code() != Some(0) {
164            let detail = String::from_utf8_lossy(output.stderr()).trim().to_owned();
165            if permission_denied(&detail) {
166                return Ok(LogRead {
167                    host: host.id().clone(),
168                    topology_revision: host.revision().clone(),
169                    source: LogSource::Dmesg,
170                    source_path: None,
171                    lines: Vec::new(),
172                    truncated: output.truncated(),
173                    permission: Some(LogPermissionDiagnostic {
174                        message: detail,
175                        help: "dmesg requires root or CAP_SYSLOG on restricted kernels".into(),
176                    }),
177                });
178            }
179            return Err(command_error(host, "logs", &output));
180        }
181        self.render(host, request, output, None, None)
182    }
183
184    async fn execute(
185        &self,
186        host: &HostRecord,
187        program: &str,
188        args: Vec<String>,
189        request: &LogReadRequest,
190        cancellation: &CancellationToken,
191    ) -> InfraResult<CommandOutput> {
192        let command = CommandRequest::new(program, args, request.deadline())
193            .map_err(soma_fleet::FleetError::from)?
194            .with_output_limits(LOG_OUTPUT_LIMIT, LOG_OUTPUT_LIMIT)
195            .map_err(soma_fleet::FleetError::from)?;
196        self.executor
197            .execute(host, &command, cancellation)
198            .await
199            .map_err(InfraError::from)
200    }
201
202    fn render(
203        &self,
204        host: &HostRecord,
205        request: &LogReadRequest,
206        output: CommandOutput,
207        source_path: Option<PathBuf>,
208        permission: Option<LogPermissionDiagnostic>,
209    ) -> InfraResult<LogRead> {
210        let text = std::str::from_utf8(output.stdout()).map_err(|error| InfraError::Parse {
211            domain: "logs",
212            message: format!("log output is not UTF-8: {error}"),
213        })?;
214        let (lines, line_truncated) = filtered_tail(text, request.grep(), request.lines());
215        Ok(LogRead {
216            host: host.id().clone(),
217            topology_revision: host.revision().clone(),
218            source: request.source(),
219            source_path,
220            lines,
221            truncated: output.truncated() || line_truncated,
222            permission,
223        })
224    }
225}
226
227fn missing_file(output: &CommandOutput) -> bool {
228    let stderr = String::from_utf8_lossy(output.stderr()).to_lowercase();
229    stderr.contains("no such file") || stderr.contains("not found")
230}
231
232fn permission_denied(detail: &str) -> bool {
233    let detail = detail.to_lowercase();
234    detail.contains("operation not permitted")
235        || detail.contains("permission denied")
236        || detail.contains("read kernel buffer failed")
237}
238
239fn command_error(host: &HostRecord, domain: &'static str, output: &CommandOutput) -> InfraError {
240    InfraError::CommandFailed {
241        domain,
242        host: host.id().clone(),
243        exit_code: output.exit_code(),
244        stderr: String::from_utf8_lossy(output.stderr()).trim().to_owned(),
245    }
246}
247
248#[cfg(test)]
249#[path = "process_logs_tests.rs"]
250mod tests;
soma_infra/process_logs.rs

soma_infra/
process_logs.rs
