dm_database_driver_log/formats/
jdbc.rs

1//! DM（达梦）JDBC 驱动日志行的解析器。
2//!
3//! 设计约束：不使用正则表达式，全部是单次线性扫描；除极少数需要拼接的字段外不分配内存。
4//!
5//! 单行结构（两种形态）：
6//!
7//! ```text
8//! [LEVEL - ts] tid:N - [thread] { conn-x, pstmt-y, rs-z } method(args): ret;  [PARAMS]: p;  [USED TIME]: t ms; [EXEC_ID]: e;
9//! [DEBUG - ts] tid:N - [thread] { conn-x } access();  CMD_EXECUTE2
10//! ```
11//!
12//! 三个关键点保证不用正则也能切得准：
13//!   1. 头部按固定分隔符逐段前进（`']'` -> `'tid:'` -> `'-'` -> `'[thread]'` -> `'{ids}'`），失败即报错，不回溯。
14//!   2. 尾部标记（`[PARAMS]` / `[USED TIME]` / `[EXEC_ID]` / `CMD_x`）必须出现在「分号 + 若干空格」之后，
15//!      因此列值、SQL 文本里即使含有分号或方括号也不会被误切。
16//!   3. 字段终点由「下一个标记的起点」决定，天然避开值里含分号的情况（例如 SQL 里的 ';'）。
17
18use crate::core::{LogFormat, LogRecord, RecordFraming};
19use crate::error::ParseError;
20
21/// 一行日志解析出的结构化事件（字段全部借用原始行，零拷贝）。
22#[derive(Debug, Clone, Default)]
23pub struct Event<'a> {
24    pub level: &'a str,
25    pub event_time_text: &'a str,
26    /// 由 event_time_text 换算出的 epoch 毫秒
27    pub event_time_ms: Option<i64>,
28    pub tid: Option<i32>,
29    pub thread: &'a str,
30    pub conn_id: Option<i32>,
31    pub pstmt_id: Option<i32>,
32    pub rs_id: Option<i32>,
33    /// 统一口径的结果集编号：executeQuery 行取返回值里的 rs-N，其余行取 id 块里的 rs-N
34    pub result_set_id: Option<i32>,
35    pub handle_id: Option<i32>,
36    pub session_id_hex: Option<&'a str>,
37    pub method: &'a str,
38    pub arg_types: &'a str,
39    pub return_value: Option<&'a str>,
40    pub params: Option<&'a str>,
41    pub param_index: Option<i32>,
42    pub param_value: Option<&'a str>,
43    pub column_name: Option<&'a str>,
44    pub used_time_text: Option<&'a str>,
45    pub used_time_ms: Option<f64>,
46    pub exec_id: Option<i64>,
47    pub cmd: Option<&'a str>,
48    pub category: &'static str,
49}
50
51impl<'a> Event<'a> {
52    /// 将当前行的零拷贝事件复制成可脱离输入缓冲区保存的记录。
53    pub fn to_owned(&self, raw: &str, line_number: u64) -> DriverLogEvent {
54        DriverLogEvent {
55            line_number,
56            raw: raw.to_owned(),
57            level: self.level.to_owned(),
58            event_time_text: self.event_time_text.to_owned(),
59            event_time_ms: self.event_time_ms,
60            tid: self.tid,
61            thread: self.thread.to_owned(),
62            conn_id: self.conn_id,
63            pstmt_id: self.pstmt_id,
64            rs_id: self.rs_id,
65            result_set_id: self.result_set_id,
66            handle_id: self.handle_id,
67            session_id_hex: self.session_id_hex.map(str::to_owned),
68            method: self.method.to_owned(),
69            arg_types: self.arg_types.to_owned(),
70            return_value: self.return_value.map(str::to_owned),
71            params: self.params.map(str::to_owned),
72            param_index: self.param_index,
73            param_value: self.param_value.map(str::to_owned),
74            column_name: self.column_name.map(str::to_owned),
75            used_time_text: self.used_time_text.map(str::to_owned),
76            used_time_ms: self.used_time_ms,
77            exec_id: self.exec_id,
78            cmd: self.cmd.map(str::to_owned),
79            category: self.category,
80        }
81    }
82}
83
84/// 一条可独立保存的驱动日志事件。
85///
86/// 文件迭代器返回此类型；统一入口会将 JDBC 记录包装为
87/// [`crate::LogEvent::Jdbc`]。
88#[derive(Debug, Clone, PartialEq)]
89pub struct DriverLogEvent {
90    pub line_number: u64,
91    pub raw: String,
92    pub level: String,
93    pub event_time_text: String,
94    pub event_time_ms: Option<i64>,
95    pub tid: Option<i32>,
96    pub thread: String,
97    pub conn_id: Option<i32>,
98    pub pstmt_id: Option<i32>,
99    pub rs_id: Option<i32>,
100    pub result_set_id: Option<i32>,
101    pub handle_id: Option<i32>,
102    pub session_id_hex: Option<String>,
103    pub method: String,
104    pub arg_types: String,
105    pub return_value: Option<String>,
106    pub params: Option<String>,
107    pub param_index: Option<i32>,
108    pub param_value: Option<String>,
109    pub column_name: Option<String>,
110    pub used_time_text: Option<String>,
111    pub used_time_ms: Option<f64>,
112    pub exec_id: Option<i64>,
113    pub cmd: Option<String>,
114    pub category: &'static str,
115}
116
117/// JDBC 日志格式适配器。
118#[derive(Copy, Clone, Debug, Default)]
119pub struct JdbcFormat;
120
121impl LogRecord for DriverLogEvent {
122    fn method(&self) -> &str {
123        &self.method
124    }
125
126    fn category(&self) -> &str {
127        self.category
128    }
129
130    fn used_time_ms(&self) -> Option<f64> {
131        self.used_time_ms
132    }
133
134    fn exec_id(&self) -> Option<i64> {
135        self.exec_id
136    }
137}
138
139impl LogFormat for JdbcFormat {
140    type Event = DriverLogEvent;
141
142    const FRAMING: RecordFraming = RecordFraming::Line;
143
144    fn is_record_start(line: &str) -> bool {
145        line.starts_with('[')
146    }
147
148    fn parse_record(record: &str, line_number: u64) -> Result<Self::Event, ParseError> {
149        parse(record).map(|event| event.to_owned(record, line_number))
150    }
151}
152
153const MARK_PARAMS: &[u8] = b"[PARAMS]: ";
154const MARK_USED_TIME: &[u8] = b"[USED TIME]: ";
155const MARK_EXEC_ID: &[u8] = b"[EXEC_ID]: ";
156const MARK_CMD: &[u8] = b"CMD_";
157const MARK_SESSION: &[u8] = b"sessionID-";
158
159/// 解析一行（不含换行符）。line 必须是合法 UTF-8（调用方校验），
160/// 这样所有切片都落在字符边界上——所有切分点都是 ASCII。
161pub fn parse(line: &str) -> Result<Event<'_>, ParseError> {
162    parse_inner(line).map_err(|e| e.with_context(line, 0))
163}
164
165fn parse_inner(line: &str) -> Result<Event<'_>, ParseError> {
166    let b = line.as_bytes();
167    if b.is_empty() {
168        return Err(ParseError::invalid(0, "empty line"));
169    }
170    if b[0] != b'[' {
171        return Err(ParseError::invalid(0, "line does not start with '['"));
172    }
173
174    // ---- 头部：[LEVEL - ts] ----
175    let head_end = find(b, b"]", 1).ok_or(ParseError::invalid(1, "unterminated level header"))?;
176    let head = &b[1..head_end];
177    let sep = find(head, b" - ", 0).ok_or(ParseError::invalid(1, "missing ' - ' in header"))?;
178    let level = trim(&head[..sep]);
179    let ts = trim(&head[sep + 3..]);
180
181    // ---- tid / thread / id 块 ----
182    let mut i = skip_spaces(b, head_end + 1);
183    expect(b, i, b"tid:")?;
184    i += 4;
185    let (tid, ni) = take_i64(b, i).ok_or(ParseError::invalid(i, "bad tid"))?;
186    i = skip_spaces(b, ni);
187    expect(b, i, b"-")?;
188    i = skip_spaces(b, i + 1);
189    expect(b, i, b"[")?;
190    let thread_end = find(b, b"]", i + 1).ok_or(ParseError::invalid(i, "unterminated thread"))?;
191    let thread = trim(&b[i + 1..thread_end]);
192    i = skip_spaces(b, thread_end + 1);
193    expect(b, i, b"{")?;
194    let ids_end = find(b, b"}", i + 1).ok_or(ParseError::invalid(i, "unterminated id block"))?;
195    let ids = trim(&b[i + 1..ids_end]);
196    let rest = trim_start(&b[ids_end + 1..]);
197    let rb = rest.as_bytes();
198
199    // ---- method(args) ----
200    let popen = find(rb, b"(", 0).ok_or(ParseError::invalid(0, "missing '(' after method"))?;
201    let method = trim(&rb[..popen]);
202    let pclose =
203        find(rb, b")", popen + 1).ok_or(ParseError::invalid(popen, "unterminated method args"))?;
204    let arg_types = trim(&rb[popen + 1..pclose]);
205    let after = pclose + 1;
206
207    // ---- 尾部标记（只在字段边界上匹配）----
208    let m_params = find_marker(rb, MARK_PARAMS, after);
209    let m_time = find_marker(rb, MARK_USED_TIME, after);
210    let m_exec = find_marker(rb, MARK_EXEC_ID, after);
211    let m_cmd = find_marker(rb, MARK_CMD, after);
212    let first_tail = [m_params, m_time, m_exec, m_cmd]
213        .into_iter()
214        .flatten()
215        .min();
216
217    let mut ev = Event {
218        level,
219        event_time_text: ts,
220        event_time_ms: epoch_millis(ts),
221        tid: i32::try_from(tid).ok(),
222        thread,
223        method,
224        arg_types,
225        category: category_of(method),
226        ..Default::default()
227    };
228
229    // 返回值：只有 "): x" 形态才有；") ;" 表示 void 方法
230    if rb.get(after) == Some(&b':') {
231        let end = first_tail.unwrap_or(rb.len());
232        if end > after + 1 {
233            let raw = trim_semi(trim(&rb[after + 1..end]));
234            if !raw.is_empty() {
235                ev.return_value = Some(raw);
236            }
237        }
238    }
239
240    // [PARAMS]: 直到下一个标记
241    if let Some(p) = m_params {
242        let start = p + MARK_PARAMS.len();
243        let end = [m_time, m_exec, m_cmd]
244            .into_iter()
245            .flatten()
246            .filter(|x| *x > start)
247            .min()
248            .unwrap_or(rb.len());
249        if end > start {
250            let raw = trim_semi(trim(&rb[start..end]));
251            if !raw.is_empty() {
252                ev.params = Some(raw);
253            }
254        }
255    }
256
257    // [USED TIME]: 1.23ms / 4.5E-4ms
258    if let Some(p) = m_time {
259        let start = p + MARK_USED_TIME.len();
260        if let Some(m) = find(rb, b"ms", start) {
261            let num = trim(&rb[start..m]);
262            if !num.is_empty() {
263                ev.used_time_text = Some(num);
264                ev.used_time_ms = num.parse::<f64>().ok();
265            }
266        }
267    }
268
269    // [EXEC_ID]: 19010657
270    if let Some(p) = m_exec
271        && let Some((v, _)) = take_i64(rb, p + MARK_EXEC_ID.len())
272    {
273        ev.exec_id = Some(v);
274    }
275
276    // CMD_EXECUTE2 / CMD_FETCH / CMD_COMMIT
277    if let Some(p) = m_cmd {
278        let seg = &rb[p..];
279        let mut e = 0;
280        while e < seg.len()
281            && (seg[e].is_ascii_uppercase() || seg[e].is_ascii_digit() || seg[e] == b'_')
282        {
283            e += 1;
284        }
285        if e > MARK_CMD.len() {
286            ev.cmd = rest.get(p..p + e);
287        }
288    }
289
290    // ---- id 块：conn-3, pstmt-854, rs-2216 ----
291    for token in ids.split(',') {
292        let t = trim(token.as_bytes());
293        let tb = t.as_bytes();
294        if let Some(v) = strip_digits(tb, b"conn-") {
295            ev.conn_id = i32::try_from(v).ok();
296        } else if let Some(v) = strip_digits(tb, b"pstmt-") {
297            ev.pstmt_id = i32::try_from(v).ok();
298        } else if let Some(v) = strip_digits(tb, b"rs-") {
299            ev.rs_id = i32::try_from(v).ok();
300        } else if let Some(v) = strip_digits(tb, b"handle-") {
301            ev.handle_id = i32::try_from(v).ok();
302        }
303    }
304
305    // ---- 返回值里的对象编号 ----
306    if let Some(rv) = ev.return_value {
307        let vrb = rv.as_bytes();
308        if let Some(v) = digits_after(vrb, b"rs-") {
309            // executeQuery(): rs-2216
310            ev.result_set_id = i32::try_from(v).ok();
311        } else if let Some(v) = digits_after(vrb, b"pstmt-") {
312            // prepareStatement(): pstmt-854, handle-3, sessionID-0x6c0504a0(1812268192)
313            ev.pstmt_id = i32::try_from(v).ok();
314            if let Some(h) = find(vrb, b"handle-", 0).and_then(|i| take_i64(vrb, i + 7)) {
315                ev.handle_id = i32::try_from(h.0).ok();
316            }
317            if let Some(s) = find(vrb, MARK_SESSION, 0) {
318                let start = s + MARK_SESSION.len();
319                let seg = &vrb[start..];
320                let mut e = 0;
321                while e < seg.len()
322                    && (seg[e].is_ascii_hexdigit() || seg[e] == b'x' || seg[e] == b'X')
323                {
324                    e += 1;
325                }
326                ev.session_id_hex = rv.get(start..start + e);
327            }
328        }
329    }
330    ev.result_set_id = ev.result_set_id.or(ev.rs_id);
331
332    // ---- 绑定参数（setXxx）：PARAMS 形如  1, "A06093910597367000594686" ----
333    if ev.category == "bind"
334        && let Some(p) = ev.params
335    {
336        let pb = p.as_bytes();
337        if let Some((idx, ni)) = take_i64(pb, 0) {
338            ev.param_index = i32::try_from(idx).ok();
339            let mut value_start = ni.min(pb.len());
340            while value_start < pb.len()
341                && (pb[value_start] == b',' || pb[value_start].is_ascii_whitespace())
342            {
343                value_start += 1;
344            }
345            let v = trim(&pb[value_start..]);
346            if !v.is_empty() {
347                ev.param_value = Some(v);
348            }
349        }
350    }
351
352    // ---- 读取列（getXxx）：PARAMS 形如  "id_pirec1_119_" ----
353    if ev.method.starts_with("get")
354        && let Some(p) = ev.params
355        && p.len() >= 2
356        && p.starts_with('"')
357        && p.ends_with('"')
358    {
359        ev.column_name = Some(&p[1..p.len() - 1]);
360    }
361
362    Ok(ev)
363}
364
365/// 方法名 -> 事件分类。
366pub fn category_of(method: &str) -> &'static str {
367    if method.is_empty() {
368        return "unparsed";
369    }
370    if method == "prepareStatement" {
371        return "prepare";
372    }
373    if method.starts_with("set") {
374        return "bind";
375    }
376    if method.starts_with("execute") {
377        return "execute";
378    }
379    if method == "next" {
380        return "fetch_next";
381    }
382    if method == "wasNull" {
383        return "null_check";
384    }
385    if method.starts_with("clear") {
386        return "clear";
387    }
388    if method == "close" {
389        return "close";
390    }
391    if method == "access" {
392        return "driver_access";
393    }
394    if method.starts_with("get") {
395        return match method {
396            "getString" | "getTimestamp" | "getLong" | "getInt" | "getShort" | "getDouble"
397            | "getFloat" | "getBoolean" | "getBytes" | "getBigDecimal" | "getDate" | "getTime"
398            | "getObject" | "getBlob" | "getClob" | "getNString" | "getNClob" | "getURL"
399            | "getAsciiStream" | "getBinaryStream" | "getCharacterStream" => "read_value",
400            _ => "read_meta",
401        };
402    }
403    "other"
404}
405
406// ------------------------------------------------------------------ 基础扫描工具
407
408/// 朴素子串查找：先定位首字节再逐字节比较。样例日志行很短（p99 = 197 字节），
409/// 无需 BM/KMP，实测比正则快一到两个数量级。
410pub fn find(hay: &[u8], needle: &[u8], from: usize) -> Option<usize> {
411    if needle.is_empty() || hay.len() < needle.len() || from > hay.len() {
412        return None;
413    }
414    let first = needle[0];
415    let last_start = hay.len() - needle.len();
416    let mut i = from;
417    while i <= last_start {
418        if hay[i] == first && &hay[i..i + needle.len()] == needle {
419            return Some(i);
420        }
421        i += 1;
422    }
423    None
424}
425
426/// 只在字段边界且不在引号字符串内匹配标记：标记左边必须是分号
427/// （中间允许空格）。这样 SQL 参数即使包含 `; [USED TIME]` 也不会被误切。
428fn find_marker(hay: &[u8], needle: &[u8], from: usize) -> Option<usize> {
429    let mut i = from;
430    let mut quote: Option<u8> = None;
431    let mut escaped = false;
432    while i < hay.len() {
433        let ch = hay[i];
434        if let Some(q) = quote {
435            if escaped {
436                escaped = false;
437            } else if ch == b'\\' {
438                escaped = true;
439            } else if ch == q {
440                quote = None;
441            }
442            i += 1;
443            continue;
444        }
445        if ch == b'\'' || ch == b'"' {
446            quote = Some(ch);
447            i += 1;
448            continue;
449        }
450        if hay[i..].starts_with(needle) && is_field_boundary(hay, i) {
451            return Some(i);
452        }
453        i += 1;
454    }
455    None
456}
457
458fn is_field_boundary(b: &[u8], pos: usize) -> bool {
459    let mut j = pos;
460    while j > 0 && b[j - 1] == b' ' {
461        j -= 1;
462    }
463    j > 0 && b[j - 1] == b';'
464}
465
466fn skip_spaces(b: &[u8], mut i: usize) -> usize {
467    while i < b.len() && b[i] == b' ' {
468        i += 1;
469    }
470    i
471}
472
473fn expect(b: &[u8], i: usize, lit: &[u8]) -> Result<(), ParseError> {
474    if b.len() >= i + lit.len() && &b[i..i + lit.len()] == lit {
475        Ok(())
476    } else {
477        let what = match lit {
478            b"tid:" => "expected 'tid:'",
479            b"-" => "expected '-'",
480            b"[" => "expected '['",
481            b"{" => "expected '{'",
482            _ => "unexpected token",
483        };
484        Err(ParseError::invalid(i, what))
485    }
486}
487
488/// 去掉首尾空白。切分点均为 ASCII，故一定是合法 UTF-8 边界。
489pub fn trim(b: &[u8]) -> &str {
490    let mut s = 0;
491    let mut e = b.len();
492    while s < e && (b[s] == b' ' || b[s] == b'\t' || b[s] == b'\r' || b[s] == b'\n') {
493        s += 1;
494    }
495    while e > s && (b[e - 1] == b' ' || b[e - 1] == b'\t' || b[e - 1] == b'\r' || b[e - 1] == b'\n')
496    {
497        e -= 1;
498    }
499    unsafe { std::str::from_utf8_unchecked(&b[s..e]) }
500}
501
502fn trim_start(b: &[u8]) -> &str {
503    let mut s = 0;
504    while s < b.len() && (b[s] == b' ' || b[s] == b'\t') {
505        s += 1;
506    }
507    unsafe { std::str::from_utf8_unchecked(&b[s..]) }
508}
509
510/// 去掉行尾的分号与空白（字段的收尾符）。
511fn trim_semi(s: &str) -> &str {
512    let b = s.as_bytes();
513    let mut e = b.len();
514    while e > 0 && (b[e - 1] == b';' || b[e - 1] == b' ') {
515        e -= 1;
516    }
517    &s[..e]
518}
519
520/// 读取从 i 开始的连续十进制数字。
521fn take_i64(b: &[u8], i: usize) -> Option<(i64, usize)> {
522    let mut j = i;
523    let mut v: i64 = 0;
524    while j < b.len() && b[j].is_ascii_digit() {
525        v = v.checked_mul(10)?.checked_add((b[j] - b'0') as i64)?;
526        j += 1;
527    }
528    if j == i { None } else { Some((v, j)) }
529}
530
531/// prefix + 十进制数字，且必须整段匹配（用于 id 块里的 token）。
532fn strip_digits(b: &[u8], prefix: &[u8]) -> Option<i64> {
533    if b.len() > prefix.len() && b.starts_with(prefix) {
534        let (v, end) = take_i64(b, prefix.len())?;
535        if end == b.len() { Some(v) } else { None }
536    } else {
537        None
538    }
539}
540
541/// prefix + 十进制数字，允许后面还有别的内容（用于返回值）。
542fn digits_after(b: &[u8], prefix: &[u8]) -> Option<i64> {
543    if b.starts_with(prefix) {
544        take_i64(b, prefix.len()).map(|(v, _)| v)
545    } else {
546        None
547    }
548}
549
550/// YYYY-MM-DD HH:MM:SS.mmm -> epoch 毫秒（按 UTC 解释，无时区）。
551pub fn epoch_millis(ts: &str) -> Option<i64> {
552    let b = ts.as_bytes();
553    if b.len() < 23 {
554        return None;
555    }
556    if b[4] != b'-'
557        || b[7] != b'-'
558        || b[10] != b' '
559        || b[13] != b':'
560        || b[16] != b':'
561        || b[19] != b'.'
562    {
563        return None;
564    }
565    let num = |off: usize, len: usize| -> Option<i64> {
566        take_i64(b, off).and_then(|(v, end)| if end == off + len { Some(v) } else { None })
567    };
568    let (y, mo, d) = (num(0, 4)?, num(5, 2)?, num(8, 2)?);
569    let (h, mi, s) = (num(11, 2)?, num(14, 2)?, num(17, 2)?);
570    let ms = num(20, 3)?;
571    let days = days_from_civil(y, mo, d);
572    Some((days * 86_400 + h * 3_600 + mi * 60 + s) * 1_000 + ms)
573}
574
575/// Howard Hinnant 的 civil -> days 算法（1970-01-01 记为 0）。
576fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
577    let y = if m <= 2 { y - 1 } else { y };
578    let era = if y >= 0 { y } else { y - 399 } / 400;
579    let yoe = y - era * 400;
580    let mp = if m > 2 { m - 3 } else { m + 9 };
581    let doy = (153 * mp + 2) / 5 + d - 1;
582    let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
583    era * 146_097 + doe - 719_468
584}
dm_database_driver_log/formats/jdbc.rs

dm_database_driver_log/formats/
jdbc.rs
