dm_database_driver_log/core/
mod.rs

1//! 所有驱动日志格式共用的文件流式解析引擎。
2//!
3//! 具体格式只需要实现 [`LogFormat`]，文件生命周期、编码、记录 framing、错误上下文
4//! 和通用过滤器都由这里统一处理。
5
6use std::fs::File;
7use std::io::{BufRead, BufReader};
8use std::marker::PhantomData;
9use std::path::{Path, PathBuf};
10
11use crate::encoding::{self, FileEncodingHint};
12use crate::error::ParseError;
13
14/// 输入文件中一条记录的边界策略。
15#[derive(Copy, Clone, Debug, PartialEq, Eq)]
16pub enum RecordFraming {
17    /// 每个物理行都是一条记录。
18    Line,
19    /// 以格式定义的记录头开始，直到下一条记录头之间的内容属于同一条记录。
20    HeaderDelimited,
21}
22
23/// 通用事件字段访问接口，供格式无关的过滤器使用。
24pub trait LogRecord {
25    fn method(&self) -> &str;
26    fn category(&self) -> &str;
27    fn used_time_ms(&self) -> Option<f64>;
28    fn exec_id(&self) -> Option<i64>;
29}
30
31/// 一个可插拔的驱动日志格式适配器。
32///
33/// 新增格式时只需要实现这个 trait，并为事件实现 [`LogRecord`]；文件迭代和
34/// 通用过滤器无需重新实现。
35pub trait LogFormat: 'static {
36    type Event: LogRecord;
37
38    const FRAMING: RecordFraming;
39
40    /// 判断一条物理行是否是该格式的记录头。
41    fn is_record_start(line: &str) -> bool;
42
43    /// 将一条已按 framing 合并的原始记录转换为拥有数据的事件。
44    /// 实现应将 `line_number` 和 `raw` 写入事件；解析错误可以只描述格式，
45    /// 通用引擎会补充原文和文件行号。
46    fn parse_record(record: &str, line_number: u64) -> Result<Self::Event, ParseError>;
47}
48
49/// 通用的日志文件解析器。
50#[derive(Debug)]
51pub struct LogParser<F: LogFormat> {
52    path: PathBuf,
53    encoding: FileEncodingHint,
54    _format: PhantomData<F>,
55}
56
57impl<F: LogFormat> LogParser<F> {
58    /// 每次调用都会重新打开文件，适合重复执行不同的过滤查询。
59    pub fn iter(&self) -> Result<LogIterator<F>, ParseError> {
60        let file = File::open(&self.path)
61            .map_err(|error| ParseError::IoError(format!("{}: {error}", self.path.display())))?;
62        Ok(LogIterator {
63            reader: BufReader::with_capacity(1 << 20, file),
64            encoding: self.encoding,
65            line_number: 0,
66            line_buf: Vec::with_capacity(4096),
67            lookahead: None,
68            done: false,
69            _format: PhantomData,
70        })
71    }
72
73    pub fn path(&self) -> &Path {
74        &self.path
75    }
76}
77
78/// 通用日志文件解析器构建器。
79pub struct LogParserBuilder<F: LogFormat> {
80    path: PathBuf,
81    encoding: FileEncodingHint,
82    _format: PhantomData<F>,
83}
84
85impl<F: LogFormat> LogParserBuilder<F> {
86    pub fn new<P: AsRef<Path>>(path: P) -> Self {
87        Self {
88            path: path.as_ref().to_path_buf(),
89            encoding: FileEncodingHint::Auto,
90            _format: PhantomData,
91        }
92    }
93
94    /// 设置输入文件编码；默认使用 [`FileEncodingHint::Auto`]。
95    pub fn encoding_hint(mut self, hint: FileEncodingHint) -> Self {
96        self.encoding = hint;
97        self
98    }
99
100    /// 打开并验证输入文件，然后构建一个可重复调用 `iter()` 的解析器。
101    pub fn build(self) -> Result<LogParser<F>, ParseError> {
102        let path = self.path;
103        match File::open(&path) {
104            Ok(_) => Ok(LogParser {
105                path,
106                encoding: self.encoding,
107                _format: PhantomData,
108            }),
109            Err(error) => Err(ParseError::IoError(format!("{}: {error}", path.display()))),
110        }
111    }
112}
113
114/// 通用日志记录迭代器。
115pub struct LogIterator<F: LogFormat> {
116    reader: BufReader<File>,
117    encoding: FileEncodingHint,
118    line_number: u64,
119    line_buf: Vec<u8>,
120    lookahead: Option<(u64, String)>,
121    done: bool,
122    _format: PhantomData<F>,
123}
124
125impl<F: LogFormat> LogIterator<F> {
126    /// 丢弃格式错误的记录，只返回成功解析的事件。
127    pub fn skip_errors(self) -> impl Iterator<Item = F::Event> {
128        self.filter_map(Result::ok)
129    }
130
131    /// 只保留指定方法的事件。
132    pub fn filter_by_method(
133        self,
134        method: &str,
135    ) -> impl Iterator<Item = Result<F::Event, ParseError>> + '_ {
136        self.filter(move |result| match result {
137            Ok(event) => event.method() == method,
138            Err(_) => true,
139        })
140    }
141
142    /// 只保留指定分类的事件。
143    pub fn filter_by_category(
144        self,
145        category: &str,
146    ) -> impl Iterator<Item = Result<F::Event, ParseError>> + '_ {
147        self.filter(move |result| match result {
148            Ok(event) => event.category() == category,
149            Err(_) => true,
150        })
151    }
152
153    /// 只保留驱动耗时大于等于 `min_ms` 的事件。
154    pub fn filter_by_used_time(
155        self,
156        min_ms: f64,
157    ) -> impl Iterator<Item = Result<F::Event, ParseError>> {
158        self.filter(move |result| match result {
159            Ok(event) => event.used_time_ms().is_some_and(|value| value >= min_ms),
160            Err(_) => true,
161        })
162    }
163
164    /// 只保留属于指定执行编号的事件。
165    pub fn filter_by_exec_id(
166        self,
167        exec_id: i64,
168    ) -> impl Iterator<Item = Result<F::Event, ParseError>> {
169        self.filter(move |result| match result {
170            Ok(event) => event.exec_id() == Some(exec_id),
171            Err(_) => true,
172        })
173    }
174
175    fn read_line(&mut self) -> Result<Option<(u64, String)>, ParseError> {
176        self.line_buf.clear();
177        let bytes_read = self
178            .reader
179            .read_until(b'\n', &mut self.line_buf)
180            .map_err(|error| ParseError::IoError(error.to_string()))?;
181        if bytes_read == 0 {
182            return Ok(None);
183        }
184
185        self.line_number += 1;
186        let mut end = self.line_buf.len();
187        while end > 0 && matches!(self.line_buf[end - 1], b'\n' | b'\r') {
188            end -= 1;
189        }
190        Ok(Some((
191            self.line_number,
192            encoding::decode(&self.line_buf[..end], self.encoding),
193        )))
194    }
195
196    fn parse_record(&self, record: &str, line_number: u64) -> Result<F::Event, ParseError> {
197        F::parse_record(record, line_number)
198            .map_err(|error| error.with_context(record, line_number))
199    }
200
201    fn next_line_record(&mut self) -> Option<Result<F::Event, ParseError>> {
202        loop {
203            let (line_number, line) = match self.read_line() {
204                Ok(Some(line)) => line,
205                Ok(None) => {
206                    self.done = true;
207                    return None;
208                }
209                Err(error) => {
210                    self.done = true;
211                    return Some(Err(error));
212                }
213            };
214            if line.is_empty() {
215                continue;
216            }
217            return Some(self.parse_record(&line, line_number));
218        }
219    }
220
221    fn next_header_delimited_record(&mut self) -> Option<Result<F::Event, ParseError>> {
222        let mut record = String::new();
223        let mut start_line = 0;
224
225        loop {
226            let next_line = match self.lookahead.take() {
227                Some(line) => Some(line),
228                None => match self.read_line() {
229                    Ok(line) => line,
230                    Err(error) => {
231                        self.done = true;
232                        return Some(Err(error));
233                    }
234                },
235            };
236            let Some((line_number, line)) = next_line else {
237                self.done = true;
238                if record.is_empty() {
239                    return None;
240                }
241                break;
242            };
243
244            if F::is_record_start(&line) && !record.is_empty() {
245                self.lookahead = Some((line_number, line));
246                break;
247            }
248            if record.is_empty() {
249                if line.is_empty() {
250                    continue;
251                }
252                start_line = line_number;
253                record = line;
254            } else {
255                record.push('\n');
256                record.push_str(&line);
257            }
258        }
259
260        Some(self.parse_record(&record, start_line))
261    }
262}
263
264impl<F: LogFormat> Iterator for LogIterator<F> {
265    type Item = Result<F::Event, ParseError>;
266
267    fn next(&mut self) -> Option<Self::Item> {
268        if self.done {
269            return None;
270        }
271        match F::FRAMING {
272            RecordFraming::Line => self.next_line_record(),
273            RecordFraming::HeaderDelimited => self.next_header_delimited_record(),
274        }
275    }
276}
dm_database_driver_log/core/mod.rs

dm_database_driver_log/core/
mod.rs
