sz_orm_logger/
log_pipeline.rs

1//! 日志管道：过滤器链、路由器、格式化器链与输出器链
2//!
3//! 本模块提供可组合的日志处理管道，支持：
4//!
5//! - **过滤器**（[`LogFilter`] trait）：决定日志是否继续传递
6//! - **格式化器**（[`LogFormatter`] trait）：将日志记录格式化为字符串
7//! - **输出器**（[`LogOutput`] trait）：将格式化后的日志写入目标
8//! - **管道**（[`LogPipeline`]）：过滤器链 → 格式化器 → 输出器链
9//! - **路由器**（[`LogRouter`]）：按规则将日志路由到不同管道
10//! - **构建器**（[`LogPipelineBuilder`]）：流式构建管道
11//!
12//! ## 示例
13//!
14//! ```no_run
15//! use sz_orm_logger::LogLevel;
16//! use sz_orm_logger::log_pipeline::{
17//!     LogPipelineBuilder, LogRecord, LevelThresholdFilter,
18//!     JsonFormatter, MemoryOutput,
19//! };
20//!
21//! let output = MemoryOutput::new();
22//! let output_handle = output.handle();
23//! let pipeline = LogPipelineBuilder::new()
24//!     .filter(Box::new(LevelThresholdFilter::new(LogLevel::Info)))
25//!     .formatter(Box::new(JsonFormatter))
26//!     .output(Box::new(output))
27//!     .build();
28//!
29//! let record = LogRecord::new(LogLevel::Info, "app", "hello");
30//! pipeline.process(&record);
31//! assert_eq!(output_handle.entries().len(), 1);
32//! ```
33
34use crate::LogLevel;
35use chrono::{DateTime, Utc};
36use parking_lot::Mutex;
37use serde::{Deserialize, Serialize};
38use std::collections::{HashMap, HashSet};
39use std::fmt;
40use std::sync::Arc;
41
42// ============================================================================
43// 日志记录
44// ============================================================================
45
46/// 日志记录，包含级别、目标、消息、时间戳与结构化字段
47///
48/// 相比 [`crate::LogEntry`]，增加了 `target`（模块名）与 `fields`（键值对），
49/// 便于过滤、路由与结构化输出。
50#[derive(Debug, Clone, Serialize, Deserialize)]
51pub struct LogRecord {
52    /// 日志级别
53    pub level: LogLevel,
54    /// 日志目标（模块名/组件名）
55    pub target: String,
56    /// 日志消息
57    pub message: String,
58    /// 时间戳（UTC）
59    pub timestamp: DateTime<Utc>,
60    /// 结构化字段
61    pub fields: HashMap<String, String>,
62}
63
64impl LogRecord {
65    /// 创建新日志记录，自动填充当前时间戳
66    pub fn new(level: LogLevel, target: impl Into<String>, message: impl Into<String>) -> Self {
67        Self {
68            level,
69            target: target.into(),
70            message: message.into(),
71            timestamp: Utc::now(),
72            fields: HashMap::new(),
73        }
74    }
75
76    /// 添加结构化字段
77    pub fn with_field(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
78        self.fields.insert(key.into(), value.into());
79        self
80    }
81
82    /// 添加多个结构化字段
83    pub fn with_fields(mut self, fields: HashMap<String, String>) -> Self {
84        self.fields.extend(fields);
85        self
86    }
87
88    /// 设置时间戳
89    pub fn with_timestamp(mut self, timestamp: DateTime<Utc>) -> Self {
90        self.timestamp = timestamp;
91        self
92    }
93
94    /// 判断级别是否达到指定阈值
95    pub fn level_at_least(&self, threshold: LogLevel) -> bool {
96        self.level >= threshold
97    }
98}
99
100impl fmt::Display for LogRecord {
101    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
102        write!(
103            f,
104            "[{}] {} {} - {}",
105            self.level.as_str(),
106            self.timestamp.to_rfc3339(),
107            self.target,
108            self.message
109        )
110    }
111}
112
113// ============================================================================
114// 过滤器 trait
115// ============================================================================
116
117/// 日志过滤器接口
118///
119/// 返回 `true` 表示日志应继续传递，`false` 表示丢弃。
120pub trait LogFilter: Send + Sync {
121    /// 判断日志记录是否应保留
122    fn should_keep(&self, record: &LogRecord) -> bool;
123
124    /// 过滤器名称（用于调试）
125    fn name(&self) -> &str {
126        "filter"
127    }
128}
129
130// ============================================================================
131// 内置过滤器
132// ============================================================================
133
134/// 级别阈值过滤器：仅保留级别 >= threshold 的日志
135#[derive(Debug, Clone)]
136pub struct LevelThresholdFilter {
137    /// 最低级别
138    pub threshold: LogLevel,
139}
140
141impl LevelThresholdFilter {
142    /// 创建过滤器，指定最低级别
143    pub fn new(threshold: LogLevel) -> Self {
144        Self { threshold }
145    }
146}
147
148impl LogFilter for LevelThresholdFilter {
149    fn should_keep(&self, record: &LogRecord) -> bool {
150        record.level >= self.threshold
151    }
152
153    fn name(&self) -> &str {
154        "level_threshold"
155    }
156}
157
158/// 目标过滤器：按 target 名称过滤
159///
160/// `allow` 为 `true` 时仅允许列表中的 target，为 `false` 时排除列表中的 target。
161#[derive(Debug, Clone)]
162pub struct TargetFilter {
163    /// 目标名称集合
164    pub targets: HashSet<String>,
165    /// true = 白名单（仅允许），false = 黑名单（排除）
166    pub allow: bool,
167}
168
169impl TargetFilter {
170    /// 创建白名单过滤器（仅允许列表中的 target）
171    pub fn allowlist(targets: &[&str]) -> Self {
172        Self {
173            targets: targets.iter().map(|s| s.to_string()).collect(),
174            allow: true,
175        }
176    }
177
178    /// 创建黑名单过滤器（排除列表中的 target）
179    pub fn blocklist(targets: &[&str]) -> Self {
180        Self {
181            targets: targets.iter().map(|s| s.to_string()).collect(),
182            allow: false,
183        }
184    }
185}
186
187impl LogFilter for TargetFilter {
188    fn should_keep(&self, record: &LogRecord) -> bool {
189        let contains = self.targets.contains(&record.target);
190        if self.allow {
191            contains
192        } else {
193            !contains
194        }
195    }
196
197    fn name(&self) -> &str {
198        if self.allow {
199            "target_allowlist"
200        } else {
201            "target_blocklist"
202        }
203    }
204}
205
206/// 子串过滤器：消息包含指定子串则保留
207#[derive(Debug, Clone)]
208pub struct ContainsFilter {
209    /// 匹配子串
210    pub pattern: String,
211    /// true = 包含则保留，false = 包含则排除
212    pub include: bool,
213}
214
215impl ContainsFilter {
216    /// 创建包含过滤器（消息包含 pattern 则保留）
217    pub fn include(pattern: impl Into<String>) -> Self {
218        Self {
219            pattern: pattern.into(),
220            include: true,
221        }
222    }
223
224    /// 创建排除过滤器（消息包含 pattern 则丢弃）
225    pub fn exclude(pattern: impl Into<String>) -> Self {
226        Self {
227            pattern: pattern.into(),
228            include: false,
229        }
230    }
231}
232
233impl LogFilter for ContainsFilter {
234    fn should_keep(&self, record: &LogRecord) -> bool {
235        let contains = record.message.contains(self.pattern.as_str());
236        if self.include {
237            contains
238        } else {
239            !contains
240        }
241    }
242
243    fn name(&self) -> &str {
244        "contains"
245    }
246}
247
248/// 复合过滤器：所有子过滤器均通过才保留（AND 逻辑）
249pub struct AllFilter {
250    filters: Vec<Box<dyn LogFilter>>,
251}
252
253impl AllFilter {
254    /// 创建复合过滤器
255    pub fn new(filters: Vec<Box<dyn LogFilter>>) -> Self {
256        Self { filters }
257    }
258}
259
260impl fmt::Debug for AllFilter {
261    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
262        f.debug_struct("AllFilter")
263            .field("filter_count", &self.filters.len())
264            .finish()
265    }
266}
267
268impl LogFilter for AllFilter {
269    fn should_keep(&self, record: &LogRecord) -> bool {
270        self.filters.iter().all(|f| f.should_keep(record))
271    }
272
273    fn name(&self) -> &str {
274        "all"
275    }
276}
277
278/// 复合过滤器：任一子过滤器通过则保留（OR 逻辑）
279pub struct AnyFilter {
280    filters: Vec<Box<dyn LogFilter>>,
281}
282
283impl AnyFilter {
284    /// 创建复合过滤器
285    pub fn new(filters: Vec<Box<dyn LogFilter>>) -> Self {
286        Self { filters }
287    }
288}
289
290impl fmt::Debug for AnyFilter {
291    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
292        f.debug_struct("AnyFilter")
293            .field("filter_count", &self.filters.len())
294            .finish()
295    }
296}
297
298impl LogFilter for AnyFilter {
299    fn should_keep(&self, record: &LogRecord) -> bool {
300        self.filters.iter().any(|f| f.should_keep(record))
301    }
302
303    fn name(&self) -> &str {
304        "any"
305    }
306}
307
308// ============================================================================
309// 格式化器 trait
310// ============================================================================
311
312/// 日志格式化器接口
313pub trait LogFormatter: Send + Sync {
314    /// 将日志记录格式化为字符串
315    fn format(&self, record: &LogRecord) -> String;
316
317    /// 格式化器名称
318    fn name(&self) -> &str {
319        "formatter"
320    }
321}
322
323// ============================================================================
324// 内置格式化器
325// ============================================================================
326
327/// JSON 格式化器：输出 JSON 行（JSONL）
328#[derive(Debug, Default)]
329pub struct JsonFormatter;
330
331impl LogFormatter for JsonFormatter {
332    fn format(&self, record: &LogRecord) -> String {
333        // 手动构建 JSON 以避免 serde_json::to_string 的额外分配
334        // 但为正确性起见使用 serde_json
335        serde_json::to_string(record).unwrap_or_else(|_| "{}".to_string())
336    }
337
338    fn name(&self) -> &str {
339        "json"
340    }
341}
342
343/// 文本格式化器：输出可读的文本行
344#[derive(Debug, Clone)]
345pub struct TextFormatter {
346    /// 格式模板，支持占位符 {level}, {target}, {message}, {timestamp}
347    pub template: String,
348}
349
350impl TextFormatter {
351    /// 创建默认文本格式化器
352    pub fn new() -> Self {
353        Self {
354            template: "[{level}] {timestamp} {target} - {message}".to_string(),
355        }
356    }
357
358    /// 创建自定义模板的格式化器
359    pub fn with_template(template: impl Into<String>) -> Self {
360        Self {
361            template: template.into(),
362        }
363    }
364
365    fn render(&self, record: &LogRecord) -> String {
366        self.template
367            .replace("{level}", record.level.as_str())
368            .replace("{timestamp}", &record.timestamp.to_rfc3339())
369            .replace("{target}", &record.target)
370            .replace("{message}", &record.message)
371    }
372}
373
374impl Default for TextFormatter {
375    fn default() -> Self {
376        Self::new()
377    }
378}
379
380impl LogFormatter for TextFormatter {
381    fn format(&self, record: &LogRecord) -> String {
382        self.render(record)
383    }
384
385    fn name(&self) -> &str {
386        "text"
387    }
388}
389
390/// 结构化字段格式化器：输出 `key=value` 对
391#[derive(Debug, Default)]
392pub struct StructuredFormatter;
393
394impl StructuredFormatter {
395    /// 创建结构化格式化器
396    pub fn new() -> Self {
397        Self
398    }
399}
400
401impl LogFormatter for StructuredFormatter {
402    fn format(&self, record: &LogRecord) -> String {
403        let mut parts = Vec::with_capacity(4 + record.fields.len());
404        parts.push(format!("level={}", record.level.as_str()));
405        parts.push(format!("ts={}", record.timestamp.to_rfc3339()));
406        parts.push(format!("target={}", record.target));
407        parts.push(format!("msg={}", record.message));
408        // 字段按键排序以保证输出稳定
409        let mut field_keys: Vec<&String> = record.fields.keys().collect();
410        field_keys.sort();
411        for key in field_keys {
412            parts.push(format!("{}={}", key, record.fields[key]));
413        }
414        parts.join(" ")
415    }
416
417    fn name(&self) -> &str {
418        "structured"
419    }
420}
421
422// ============================================================================
423// 输出器 trait
424// ============================================================================
425
426/// 日志输出器接口
427pub trait LogOutput: Send + Sync {
428    /// 写入格式化后的日志行
429    fn write(&self, formatted: &str);
430
431    /// 刷新输出缓冲（如有）
432    fn flush(&self) {}
433
434    /// 输出器名称
435    fn name(&self) -> &str {
436        "output"
437    }
438}
439
440// ============================================================================
441// 内置输出器
442// ============================================================================
443
444/// 内存输出器：将日志行存入内存缓冲，用于测试与采集
445pub struct MemoryOutput {
446    buffer: Mutex<Vec<String>>,
447}
448
449impl MemoryOutput {
450    /// 创建空内存输出器
451    pub fn new() -> Self {
452        Self {
453            buffer: Mutex::new(Vec::new()),
454        }
455    }
456
457    /// 创建带初始容量的内存输出器
458    pub fn with_capacity(capacity: usize) -> Self {
459        Self {
460            buffer: Mutex::new(Vec::with_capacity(capacity)),
461        }
462    }
463
464    /// 返回可共享的句柄，用于在外部读取输出内容
465    pub fn handle(&self) -> MemoryOutputHandle {
466        MemoryOutputHandle {
467            buffer: Arc::new(Mutex::new(Vec::new())),
468        }
469    }
470}
471
472impl Default for MemoryOutput {
473    fn default() -> Self {
474        Self::new()
475    }
476}
477
478impl fmt::Debug for MemoryOutput {
479    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
480        f.debug_struct("MemoryOutput")
481            .field("count", &self.buffer.lock().len())
482            .finish()
483    }
484}
485
486impl LogOutput for MemoryOutput {
487    fn write(&self, formatted: &str) {
488        self.buffer.lock().push(formatted.to_string());
489    }
490
491    fn name(&self) -> &str {
492        "memory"
493    }
494}
495
496/// 内存输出器句柄，可在管道构建后读取已写入的日志行
497///
498/// 注意：句柄与输出器独立，仅用于测试中预创建句柄并传入管道。
499/// 生产中建议直接使用 `MemoryOutput` 并通过 `Arc` 共享。
500pub struct MemoryOutputHandle {
501    buffer: Arc<Mutex<Vec<String>>>,
502}
503
504impl MemoryOutputHandle {
505    /// 创建句柄与对应的输出器
506    pub fn new() -> (Self, MemoryOutputShared) {
507        let buffer = Arc::new(Mutex::new(Vec::new()));
508        let handle = Self {
509            buffer: buffer.clone(),
510        };
511        let output = MemoryOutputShared { buffer };
512        (handle, output)
513    }
514
515    /// 返回已写入的日志行快照
516    pub fn entries(&self) -> Vec<String> {
517        self.buffer.lock().clone()
518    }
519
520    /// 返回已写入的日志行数
521    pub fn count(&self) -> usize {
522        self.buffer.lock().len()
523    }
524
525    /// 清空缓冲
526    pub fn clear(&self) {
527        self.buffer.lock().clear();
528    }
529}
530
531impl Default for MemoryOutputHandle {
532    fn default() -> Self {
533        Self {
534            buffer: Arc::new(Mutex::new(Vec::new())),
535        }
536    }
537}
538
539/// 共享内存输出器，与 [`MemoryOutputHandle`] 配对使用
540pub struct MemoryOutputShared {
541    buffer: Arc<Mutex<Vec<String>>>,
542}
543
544impl fmt::Debug for MemoryOutputShared {
545    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
546        f.debug_struct("MemoryOutputShared")
547            .field("count", &self.buffer.lock().len())
548            .finish()
549    }
550}
551
552impl LogOutput for MemoryOutputShared {
553    fn write(&self, formatted: &str) {
554        self.buffer.lock().push(formatted.to_string());
555    }
556
557    fn name(&self) -> &str {
558        "memory_shared"
559    }
560}
561
562/// 回调输出器：将日志行传入闭包处理
563pub struct CallbackOutput {
564    callback: Box<dyn Fn(&str) + Send + Sync>,
565}
566
567impl CallbackOutput {
568    /// 创建回调输出器
569    pub fn new(callback: impl Fn(&str) + Send + Sync + 'static) -> Self {
570        Self {
571            callback: Box::new(callback),
572        }
573    }
574}
575
576impl fmt::Debug for CallbackOutput {
577    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
578        f.debug_struct("CallbackOutput").finish()
579    }
580}
581
582impl LogOutput for CallbackOutput {
583    fn write(&self, formatted: &str) {
584        (self.callback)(formatted);
585    }
586
587    fn name(&self) -> &str {
588        "callback"
589    }
590}
591
592/// 计数输出器：仅统计写入次数，不存储内容
593pub struct CountingOutput {
594    count: std::sync::atomic::AtomicU64,
595}
596
597impl CountingOutput {
598    /// 创建计数输出器
599    pub fn new() -> Self {
600        Self {
601            count: std::sync::atomic::AtomicU64::new(0),
602        }
603    }
604
605    /// 返回已写入次数
606    pub fn count(&self) -> u64 {
607        self.count.load(std::sync::atomic::Ordering::Relaxed)
608    }
609}
610
611impl Default for CountingOutput {
612    fn default() -> Self {
613        Self::new()
614    }
615}
616
617impl fmt::Debug for CountingOutput {
618    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
619        f.debug_struct("CountingOutput")
620            .field("count", &self.count())
621            .finish()
622    }
623}
624
625impl LogOutput for CountingOutput {
626    fn write(&self, _formatted: &str) {
627        self.count
628            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
629    }
630
631    fn name(&self) -> &str {
632        "counting"
633    }
634}
635
636// ============================================================================
637// 日志管道
638// ============================================================================
639
640/// 日志管道：过滤器链 → 格式化器 → 输出器链
641///
642/// 处理流程：
643/// 1. 依次执行所有过滤器，任一过滤失败则丢弃
644/// 2. 格式化器将记录转为字符串
645/// 3. 依次写入所有输出器
646pub struct LogPipeline {
647    filters: Vec<Box<dyn LogFilter>>,
648    formatter: Box<dyn LogFormatter>,
649    outputs: Vec<Box<dyn LogOutput>>,
650    /// 已处理记录数
651    processed_count: std::sync::atomic::AtomicU64,
652    /// 已丢弃记录数
653    dropped_count: std::sync::atomic::AtomicU64,
654}
655
656impl LogPipeline {
657    /// 创建管道
658    pub fn new(
659        filters: Vec<Box<dyn LogFilter>>,
660        formatter: Box<dyn LogFormatter>,
661        outputs: Vec<Box<dyn LogOutput>>,
662    ) -> Self {
663        Self {
664            filters,
665            formatter,
666            outputs,
667            processed_count: std::sync::atomic::AtomicU64::new(0),
668            dropped_count: std::sync::atomic::AtomicU64::new(0),
669        }
670    }
671
672    /// 处理一条日志记录
673    pub fn process(&self, record: &LogRecord) {
674        // 过滤器链
675        for filter in &self.filters {
676            if !filter.should_keep(record) {
677                self.dropped_count
678                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
679                return;
680            }
681        }
682        // 格式化
683        let formatted = self.formatter.format(record);
684        // 输出器链
685        for output in &self.outputs {
686            output.write(&formatted);
687        }
688        self.processed_count
689            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
690    }
691
692    /// 批量处理日志记录
693    pub fn process_batch(&self, records: &[LogRecord]) {
694        for record in records {
695            self.process(record);
696        }
697    }
698
699    /// 返回已处理记录数
700    pub fn processed_count(&self) -> u64 {
701        self.processed_count
702            .load(std::sync::atomic::Ordering::Relaxed)
703    }
704
705    /// 返回已丢弃记录数
706    pub fn dropped_count(&self) -> u64 {
707        self.dropped_count
708            .load(std::sync::atomic::Ordering::Relaxed)
709    }
710
711    /// 返回过滤器数量
712    pub fn filter_count(&self) -> usize {
713        self.filters.len()
714    }
715
716    /// 返回输出器数量
717    pub fn output_count(&self) -> usize {
718        self.outputs.len()
719    }
720
721    /// 刷新所有输出器
722    pub fn flush(&self) {
723        for output in &self.outputs {
724            output.flush();
725        }
726    }
727}
728
729impl fmt::Debug for LogPipeline {
730    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
731        f.debug_struct("LogPipeline")
732            .field("filters", &self.filters.len())
733            .field("outputs", &self.outputs.len())
734            .field("processed", &self.processed_count())
735            .field("dropped", &self.dropped_count())
736            .finish()
737    }
738}
739
740// ============================================================================
741// 管道构建器
742// ============================================================================
743
744/// 日志管道构建器，支持流式 API
745pub struct LogPipelineBuilder {
746    filters: Vec<Box<dyn LogFilter>>,
747    formatter: Option<Box<dyn LogFormatter>>,
748    outputs: Vec<Box<dyn LogOutput>>,
749}
750
751impl LogPipelineBuilder {
752    /// 创建空构建器
753    pub fn new() -> Self {
754        Self {
755            filters: Vec::new(),
756            formatter: None,
757            outputs: Vec::new(),
758        }
759    }
760
761    /// 添加过滤器
762    pub fn filter(mut self, filter: Box<dyn LogFilter>) -> Self {
763        self.filters.push(filter);
764        self
765    }
766
767    /// 添加多个过滤器
768    pub fn filters(mut self, filters: Vec<Box<dyn LogFilter>>) -> Self {
769        self.filters.extend(filters);
770        self
771    }
772
773    /// 设置格式化器（默认 JSON）
774    pub fn formatter(mut self, formatter: Box<dyn LogFormatter>) -> Self {
775        self.formatter = Some(formatter);
776        self
777    }
778
779    /// 添加输出器
780    pub fn output(mut self, output: Box<dyn LogOutput>) -> Self {
781        self.outputs.push(output);
782        self
783    }
784
785    /// 添加多个输出器
786    pub fn outputs(mut self, outputs: Vec<Box<dyn LogOutput>>) -> Self {
787        self.outputs.extend(outputs);
788        self
789    }
790
791    /// 构建管道
792    pub fn build(self) -> LogPipeline {
793        let formatter = self.formatter.unwrap_or_else(|| Box::new(JsonFormatter));
794        LogPipeline::new(self.filters, formatter, self.outputs)
795    }
796}
797
798impl Default for LogPipelineBuilder {
799    fn default() -> Self {
800        Self::new()
801    }
802}
803
804impl fmt::Debug for LogPipelineBuilder {
805    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
806        f.debug_struct("LogPipelineBuilder")
807            .field("filters", &self.filters.len())
808            .field("has_formatter", &self.formatter.is_some())
809            .field("outputs", &self.outputs.len())
810            .finish()
811    }
812}
813
814// ============================================================================
815// 路由规则与路由器
816// ============================================================================
817
818/// 路由规则：匹配的日志发送到指定管道
819pub struct RoutingRule {
820    /// 匹配过滤器
821    pub filter: Box<dyn LogFilter>,
822    /// 目标管道
823    pub pipeline: LogPipeline,
824    /// 规则名称
825    pub name: String,
826}
827
828impl fmt::Debug for RoutingRule {
829    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
830        f.debug_struct("RoutingRule")
831            .field("name", &self.name)
832            .field("filter", &self.filter.name())
833            .finish()
834    }
835}
836
837/// 日志路由器：按规则将日志路由到不同管道
838///
839/// 按规则顺序匹配，第一个匹配的规则处理日志。
840/// 若无规则匹配，则发送到默认管道（如有）。
841pub struct LogRouter {
842    rules: Vec<RoutingRule>,
843    default: Option<LogPipeline>,
844    /// 已路由记录数
845    routed_count: std::sync::atomic::AtomicU64,
846    /// 未匹配记录数
847    unmatched_count: std::sync::atomic::AtomicU64,
848}
849
850impl LogRouter {
851    /// 创建空路由器
852    pub fn new() -> Self {
853        Self {
854            rules: Vec::new(),
855            default: None,
856            routed_count: std::sync::atomic::AtomicU64::new(0),
857            unmatched_count: std::sync::atomic::AtomicU64::new(0),
858        }
859    }
860
861    /// 添加路由规则
862    pub fn route(mut self, rule: RoutingRule) -> Self {
863        self.rules.push(rule);
864        self
865    }
866
867    /// 设置默认管道（无规则匹配时使用）
868    pub fn default_pipeline(mut self, pipeline: LogPipeline) -> Self {
869        self.default = Some(pipeline);
870        self
871    }
872
873    /// 路由一条日志记录
874    pub fn route_record(&self, record: &LogRecord) {
875        for rule in &self.rules {
876            if rule.filter.should_keep(record) {
877                rule.pipeline.process(record);
878                self.routed_count
879                    .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
880                return;
881            }
882        }
883        // 无规则匹配
884        if let Some(default) = &self.default {
885            default.process(record);
886            self.routed_count
887                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
888        } else {
889            self.unmatched_count
890                .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
891        }
892    }
893
894    /// 批量路由日志记录
895    pub fn route_batch(&self, records: &[LogRecord]) {
896        for record in records {
897            self.route_record(record);
898        }
899    }
900
901    /// 返回已路由记录数
902    pub fn routed_count(&self) -> u64 {
903        self.routed_count.load(std::sync::atomic::Ordering::Relaxed)
904    }
905
906    /// 返回未匹配记录数
907    pub fn unmatched_count(&self) -> u64 {
908        self.unmatched_count
909            .load(std::sync::atomic::Ordering::Relaxed)
910    }
911
912    /// 返回规则数量
913    pub fn rule_count(&self) -> usize {
914        self.rules.len()
915    }
916
917    /// 是否有默认管道
918    pub fn has_default(&self) -> bool {
919        self.default.is_some()
920    }
921}
922
923impl Default for LogRouter {
924    fn default() -> Self {
925        Self::new()
926    }
927}
928
929impl fmt::Debug for LogRouter {
930    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
931        f.debug_struct("LogRouter")
932            .field("rules", &self.rules.len())
933            .field("has_default", &self.default.is_some())
934            .field("routed", &self.routed_count())
935            .field("unmatched", &self.unmatched_count())
936            .finish()
937    }
938}
939
940// ============================================================================
941// 日志速率限制器
942// ============================================================================
943
944/// 日志速率限制器：在时间窗口内最多允许 N 条日志通过
945pub struct RateLimitFilter {
946    /// 时间窗口（毫秒）
947    window_ms: u64,
948    /// 窗口内最大条数
949    max_count: u32,
950    /// 当前窗口起始时间
951    window_start: parking_lot::Mutex<Option<std::time::Instant>>,
952    /// 当前窗口内计数
953    count: parking_lot::Mutex<u32>,
954}
955
956impl RateLimitFilter {
957    /// 创建速率限制器
958    pub fn new(window_ms: u64, max_count: u32) -> Self {
959        Self {
960            window_ms,
961            max_count,
962            window_start: parking_lot::Mutex::new(None),
963            count: parking_lot::Mutex::new(0),
964        }
965    }
966
967    /// 每秒最多 N 条
968    pub fn per_second(max_per_sec: u32) -> Self {
969        Self::new(1000, max_per_sec)
970    }
971}
972
973impl LogFilter for RateLimitFilter {
974    fn should_keep(&self, _record: &LogRecord) -> bool {
975        let now = std::time::Instant::now();
976        let mut start = self.window_start.lock();
977        let mut count = self.count.lock();
978        match *start {
979            None => {
980                *start = Some(now);
981                *count = 1;
982                true
983            }
984            Some(s) => {
985                let elapsed = now.duration_since(s);
986                if elapsed.as_millis() as u64 >= self.window_ms {
987                    *start = Some(now);
988                    *count = 1;
989                    true
990                } else {
991                    *count += 1;
992                    *count <= self.max_count
993                }
994            }
995        }
996    }
997}
998
999// ============================================================================
1000// 日志采样器
1001// ============================================================================
1002
1003/// 日志采样器：按比例随机采样日志
1004pub struct SamplingFilter {
1005    /// 采样率（0.0-1.0，1.0 = 全部通过）
1006    rate: f64,
1007    /// 伪随机状态
1008    state: parking_lot::Mutex<u64>,
1009}
1010
1011impl SamplingFilter {
1012    /// 创建采样器
1013    ///
1014    /// `rate` 会被 clamp 到 [0.0, 1.0]
1015    pub fn new(rate: f64) -> Self {
1016        Self {
1017            rate: rate.clamp(0.0, 1.0),
1018            state: parking_lot::Mutex::new(0x12345678),
1019        }
1020    }
1021
1022    /// 10% 采样
1023    pub fn ten_percent() -> Self {
1024        Self::new(0.1)
1025    }
1026
1027    /// 1% 采样
1028    pub fn one_percent() -> Self {
1029        Self::new(0.01)
1030    }
1031
1032    /// 获取采样率
1033    pub fn rate(&self) -> f64 {
1034        self.rate
1035    }
1036
1037    /// 简单 LCG 伪随机
1038    fn next_random(&self) -> f64 {
1039        let mut state = self.state.lock();
1040        *state = state
1041            .wrapping_mul(6364136223846793005)
1042            .wrapping_add(1442695040888963407);
1043        (*state >> 11) as f64 / (1u64 << 53) as f64
1044    }
1045}
1046
1047impl LogFilter for SamplingFilter {
1048    fn should_keep(&self, _record: &LogRecord) -> bool {
1049        self.next_random() < self.rate
1050    }
1051}
1052
1053// ============================================================================
1054// 日志聚合器
1055// ============================================================================
1056
1057/// 日志聚合器：相同消息的日志在时间窗口内只输出第一条和计数
1058pub struct LogAggregator {
1059    /// 时间窗口（毫秒）
1060    window_ms: u64,
1061    /// 聚合表：message -> (first_seen, count)
1062    entries: parking_lot::Mutex<HashMap<String, (std::time::Instant, u32)>>,
1063}
1064
1065impl LogAggregator {
1066    /// 创建聚合器
1067    pub fn new(window_ms: u64) -> Self {
1068        Self {
1069            window_ms,
1070            entries: parking_lot::Mutex::new(HashMap::new()),
1071        }
1072    }
1073
1074    /// 处理一条日志，返回是否应输出
1075    ///
1076    /// 首次出现返回 true，窗口内重复出现返回 false，
1077    /// 窗口过期后首次出现返回 true 并携带上一窗口的计数。
1078    pub fn should_output(&self, message: &str) -> bool {
1079        let now = std::time::Instant::now();
1080        let mut entries = self.entries.lock();
1081        match entries.get_mut(message) {
1082            Some((first_seen, count)) => {
1083                let elapsed = now.duration_since(*first_seen);
1084                if elapsed.as_millis() as u64 >= self.window_ms {
1085                    *first_seen = now;
1086                    *count = 1;
1087                    true
1088                } else {
1089                    *count += 1;
1090                    false
1091                }
1092            }
1093            None => {
1094                entries.insert(message.to_string(), (now, 1));
1095                true
1096            }
1097        }
1098    }
1099
1100    /// 获取消息的当前窗口计数
1101    pub fn count(&self, message: &str) -> u32 {
1102        self.entries
1103            .lock()
1104            .get(message)
1105            .map(|(_, c)| *c)
1106            .unwrap_or(0)
1107    }
1108
1109    /// 清除所有聚合条目
1110    pub fn clear(&self) {
1111        self.entries.lock().clear();
1112    }
1113}
1114
1115// ============================================================================
1116// 日志缓冲器
1117// ============================================================================
1118
1119/// 日志缓冲器：积累日志到阈值后批量输出
1120pub struct LogBuffer {
1121    /// 缓冲区容量
1122    capacity: usize,
1123    /// 缓冲的日志记录
1124    buffer: parking_lot::Mutex<Vec<LogRecord>>,
1125}
1126
1127impl LogBuffer {
1128    /// 创建缓冲器
1129    pub fn new(capacity: usize) -> Self {
1130        Self {
1131            capacity: capacity.max(1),
1132            buffer: parking_lot::Mutex::new(Vec::new()),
1133        }
1134    }
1135
1136    /// 添加日志记录，返回是否达到 flush 阈值
1137    pub fn push(&self, record: LogRecord) -> bool {
1138        let mut buf = self.buffer.lock();
1139        buf.push(record);
1140        buf.len() >= self.capacity
1141    }
1142
1143    /// 取出所有缓冲的日志（flush）
1144    pub fn flush(&self) -> Vec<LogRecord> {
1145        let mut buf = self.buffer.lock();
1146        std::mem::take(&mut *buf)
1147    }
1148
1149    /// 当前缓冲区大小
1150    pub fn len(&self) -> usize {
1151        self.buffer.lock().len()
1152    }
1153
1154    /// 缓冲区是否为空
1155    pub fn is_empty(&self) -> bool {
1156        self.len() == 0
1157    }
1158
1159    /// 缓冲区容量
1160    pub fn capacity(&self) -> usize {
1161        self.capacity
1162    }
1163}
1164
1165// ============================================================================
1166// 测试
1167// ============================================================================
1168
1169#[cfg(test)]
1170mod tests {
1171    use super::*;
1172
1173    // ---- LogRecord 测试 ----
1174
1175    #[test]
1176    fn test_log_record_new() {
1177        let record = LogRecord::new(LogLevel::Info, "app", "hello");
1178        assert_eq!(record.level, LogLevel::Info);
1179        assert_eq!(record.target, "app");
1180        assert_eq!(record.message, "hello");
1181        assert!(record.fields.is_empty());
1182    }
1183
1184    #[test]
1185    fn test_log_record_with_field() {
1186        let record = LogRecord::new(LogLevel::Warn, "db", "slow query")
1187            .with_field("duration", "150ms")
1188            .with_field("sql", "SELECT * FROM users");
1189        assert_eq!(record.fields.get("duration"), Some(&"150ms".to_string()));
1190        assert_eq!(
1191            record.fields.get("sql"),
1192            Some(&"SELECT * FROM users".to_string())
1193        );
1194    }
1195
1196    #[test]
1197    fn test_log_record_level_at_least() {
1198        let record = LogRecord::new(LogLevel::Warn, "app", "msg");
1199        assert!(record.level_at_least(LogLevel::Warn));
1200        assert!(record.level_at_least(LogLevel::Info));
1201        assert!(!record.level_at_least(LogLevel::Error));
1202    }
1203
1204    #[test]
1205    fn test_log_record_display() {
1206        let record = LogRecord::new(LogLevel::Error, "app", "crash");
1207        let s = format!("{}", record);
1208        assert!(s.contains("ERROR"));
1209        assert!(s.contains("app"));
1210        assert!(s.contains("crash"));
1211    }
1212
1213    // ---- LevelThresholdFilter 测试 ----
1214
1215    #[test]
1216    fn test_level_threshold_filter_passes() {
1217        let filter = LevelThresholdFilter::new(LogLevel::Info);
1218        let record = LogRecord::new(LogLevel::Info, "app", "msg");
1219        assert!(filter.should_keep(&record));
1220    }
1221
1222    #[test]
1223    fn test_level_threshold_filter_blocks() {
1224        let filter = LevelThresholdFilter::new(LogLevel::Warn);
1225        let record = LogRecord::new(LogLevel::Debug, "app", "msg");
1226        assert!(!filter.should_keep(&record));
1227    }
1228
1229    #[test]
1230    fn test_level_threshold_filter_boundary() {
1231        let filter = LevelThresholdFilter::new(LogLevel::Warn);
1232        assert!(filter.should_keep(&LogRecord::new(LogLevel::Warn, "a", "m")));
1233        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "a", "m")));
1234        assert!(filter.should_keep(&LogRecord::new(LogLevel::Error, "a", "m")));
1235    }
1236
1237    // ---- TargetFilter 测试 ----
1238
1239    #[test]
1240    fn test_target_filter_allowlist() {
1241        let filter = TargetFilter::allowlist(&["app", "db"]);
1242        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "app", "m")));
1243        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "db", "m")));
1244        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "cache", "m")));
1245    }
1246
1247    #[test]
1248    fn test_target_filter_blocklist() {
1249        let filter = TargetFilter::blocklist(&["debug", "trace"]);
1250        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "debug", "m")));
1251        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "app", "m")));
1252    }
1253
1254    // ---- ContainsFilter 测试 ----
1255
1256    #[test]
1257    fn test_contains_filter_include() {
1258        let filter = ContainsFilter::include("error");
1259        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "a", "an error occurred")));
1260        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "a", "all good")));
1261    }
1262
1263    #[test]
1264    fn test_contains_filter_exclude() {
1265        let filter = ContainsFilter::exclude("password");
1266        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "a", "user password leaked")));
1267        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "a", "user logged in")));
1268    }
1269
1270    // ---- AllFilter / AnyFilter 测试 ----
1271
1272    #[test]
1273    fn test_all_filter() {
1274        let filter = AllFilter::new(vec![
1275            Box::new(LevelThresholdFilter::new(LogLevel::Info)),
1276            Box::new(TargetFilter::allowlist(&["app"])),
1277        ]);
1278        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "app", "m")));
1279        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Debug, "app", "m")));
1280        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "db", "m")));
1281    }
1282
1283    #[test]
1284    fn test_any_filter() {
1285        let filter = AnyFilter::new(vec![
1286            Box::new(TargetFilter::allowlist(&["app"])),
1287            Box::new(LevelThresholdFilter::new(LogLevel::Error)),
1288        ]);
1289        // 匹配 target
1290        assert!(filter.should_keep(&LogRecord::new(LogLevel::Info, "app", "m")));
1291        // 匹配 level
1292        assert!(filter.should_keep(&LogRecord::new(LogLevel::Error, "db", "m")));
1293        // 都不匹配
1294        assert!(!filter.should_keep(&LogRecord::new(LogLevel::Info, "db", "m")));
1295    }
1296
1297    // ---- JsonFormatter 测试 ----
1298
1299    #[test]
1300    fn test_json_formatter() {
1301        let formatter = JsonFormatter;
1302        let record = LogRecord::new(LogLevel::Info, "app", "hello");
1303        let json = formatter.format(&record);
1304        assert!(json.contains("\"level\":\"Info\""));
1305        assert!(json.contains("\"target\":\"app\""));
1306        assert!(json.contains("\"message\":\"hello\""));
1307    }
1308
1309    #[test]
1310    fn test_json_formatter_with_fields() {
1311        let formatter = JsonFormatter;
1312        let record = LogRecord::new(LogLevel::Warn, "db", "slow").with_field("duration", "100ms");
1313        let json = formatter.format(&record);
1314        assert!(json.contains("duration"));
1315        assert!(json.contains("100ms"));
1316    }
1317
1318    // ---- TextFormatter 测试 ----
1319
1320    #[test]
1321    fn test_text_formatter_default() {
1322        let formatter = TextFormatter::new();
1323        let record = LogRecord::new(LogLevel::Error, "app", "crash");
1324        let text = formatter.format(&record);
1325        assert!(text.contains("ERROR"));
1326        assert!(text.contains("app"));
1327        assert!(text.contains("crash"));
1328    }
1329
1330    #[test]
1331    fn test_text_formatter_custom_template() {
1332        let formatter = TextFormatter::with_template("{level} - {message}");
1333        let record = LogRecord::new(LogLevel::Info, "app", "hello");
1334        let text = formatter.format(&record);
1335        assert_eq!(text, "INFO - hello");
1336    }
1337
1338    // ---- StructuredFormatter 测试 ----
1339
1340    #[test]
1341    fn test_structured_formatter() {
1342        let formatter = StructuredFormatter::new();
1343        let record = LogRecord::new(LogLevel::Info, "app", "hello").with_field("key", "value");
1344        let text = formatter.format(&record);
1345        assert!(text.contains("level=INFO"));
1346        assert!(text.contains("target=app"));
1347        assert!(text.contains("msg=hello"));
1348        assert!(text.contains("key=value"));
1349    }
1350
1351    // ---- MemoryOutput 测试 ----
1352
1353    #[test]
1354    fn test_memory_output_write() {
1355        let output = MemoryOutput::new();
1356        output.write("line 1");
1357        output.write("line 2");
1358        let entries = output.buffer.lock().clone();
1359        assert_eq!(entries, vec!["line 1", "line 2"]);
1360    }
1361
1362    // ---- MemoryOutputHandle 测试 ----
1363
1364    #[test]
1365    fn test_memory_output_handle() {
1366        let (handle, output) = MemoryOutputHandle::new();
1367        output.write("test line");
1368        assert_eq!(handle.count(), 1);
1369        assert_eq!(handle.entries(), vec!["test line"]);
1370    }
1371
1372    #[test]
1373    fn test_memory_output_handle_clear() {
1374        let (handle, output) = MemoryOutputHandle::new();
1375        output.write("a");
1376        output.write("b");
1377        assert_eq!(handle.count(), 2);
1378        handle.clear();
1379        assert_eq!(handle.count(), 0);
1380    }
1381
1382    // ---- CountingOutput 测试 ----
1383
1384    #[test]
1385    fn test_counting_output() {
1386        let output = CountingOutput::new();
1387        output.write("a");
1388        output.write("b");
1389        output.write("c");
1390        assert_eq!(output.count(), 3);
1391    }
1392
1393    // ---- CallbackOutput 测试 ----
1394
1395    #[test]
1396    fn test_callback_output() {
1397        let counter = Arc::new(std::sync::atomic::AtomicU64::new(0));
1398        let counter_clone = counter.clone();
1399        let output = CallbackOutput::new(move |_s| {
1400            counter_clone.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1401        });
1402        output.write("a");
1403        output.write("b");
1404        assert_eq!(counter.load(std::sync::atomic::Ordering::Relaxed), 2);
1405    }
1406
1407    // ---- LogPipeline 测试 ----
1408
1409    #[test]
1410    fn test_log_pipeline_basic() {
1411        let (handle, output) = MemoryOutputHandle::new();
1412        let pipeline = LogPipeline::new(
1413            vec![],
1414            Box::new(TextFormatter::new()),
1415            vec![Box::new(output)],
1416        );
1417        let record = LogRecord::new(LogLevel::Info, "app", "hello");
1418        pipeline.process(&record);
1419        assert_eq!(handle.count(), 1);
1420        assert_eq!(pipeline.processed_count(), 1);
1421        assert_eq!(pipeline.dropped_count(), 0);
1422    }
1423
1424    #[test]
1425    fn test_log_pipeline_filter_drops() {
1426        let (handle, output) = MemoryOutputHandle::new();
1427        let pipeline = LogPipeline::new(
1428            vec![Box::new(LevelThresholdFilter::new(LogLevel::Warn))],
1429            Box::new(TextFormatter::new()),
1430            vec![Box::new(output)],
1431        );
1432        let record = LogRecord::new(LogLevel::Debug, "app", "debug msg");
1433        pipeline.process(&record);
1434        assert_eq!(handle.count(), 0);
1435        assert_eq!(pipeline.processed_count(), 0);
1436        assert_eq!(pipeline.dropped_count(), 1);
1437    }
1438
1439    #[test]
1440    fn test_log_pipeline_multiple_filters() {
1441        let (handle, output) = MemoryOutputHandle::new();
1442        let pipeline = LogPipeline::new(
1443            vec![
1444                Box::new(LevelThresholdFilter::new(LogLevel::Info)),
1445                Box::new(TargetFilter::allowlist(&["app"])),
1446            ],
1447            Box::new(JsonFormatter),
1448            vec![Box::new(output)],
1449        );
1450        pipeline.process(&LogRecord::new(LogLevel::Info, "app", "ok"));
1451        pipeline.process(&LogRecord::new(LogLevel::Info, "db", "filtered"));
1452        pipeline.process(&LogRecord::new(LogLevel::Debug, "app", "filtered"));
1453        assert_eq!(handle.count(), 1);
1454        assert_eq!(pipeline.processed_count(), 1);
1455        assert_eq!(pipeline.dropped_count(), 2);
1456    }
1457
1458    #[test]
1459    fn test_log_pipeline_multiple_outputs() {
1460        let counter = CountingOutput::new();
1461        let counter_ref = Arc::new(CountingOutput::new());
1462        // 使用 CallbackOutput 作为第二个输出
1463        let count2 = Arc::new(std::sync::atomic::AtomicU64::new(0));
1464        let count2_clone = count2.clone();
1465        let callback = CallbackOutput::new(move |_| {
1466            count2_clone.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
1467        });
1468        let pipeline = LogPipeline::new(
1469            vec![],
1470            Box::new(TextFormatter::new()),
1471            vec![Box::new(counter), Box::new(callback)],
1472        );
1473        pipeline.process(&LogRecord::new(LogLevel::Info, "a", "m"));
1474        pipeline.process(&LogRecord::new(LogLevel::Info, "a", "m"));
1475        assert_eq!(count2.load(std::sync::atomic::Ordering::Relaxed), 2);
1476        let _ = counter_ref;
1477    }
1478
1479    #[test]
1480    fn test_log_pipeline_batch() {
1481        let (handle, output) = MemoryOutputHandle::new();
1482        let pipeline = LogPipeline::new(
1483            vec![],
1484            Box::new(TextFormatter::new()),
1485            vec![Box::new(output)],
1486        );
1487        let records = vec![
1488            LogRecord::new(LogLevel::Info, "a", "1"),
1489            LogRecord::new(LogLevel::Warn, "a", "2"),
1490            LogRecord::new(LogLevel::Error, "a", "3"),
1491        ];
1492        pipeline.process_batch(&records);
1493        assert_eq!(handle.count(), 3);
1494        assert_eq!(pipeline.processed_count(), 3);
1495    }
1496
1497    // ---- LogPipelineBuilder 测试 ----
1498
1499    #[test]
1500    fn test_pipeline_builder_basic() {
1501        let (handle, output) = MemoryOutputHandle::new();
1502        let pipeline = LogPipelineBuilder::new()
1503            .filter(Box::new(LevelThresholdFilter::new(LogLevel::Info)))
1504            .formatter(Box::new(TextFormatter::new()))
1505            .output(Box::new(output))
1506            .build();
1507        pipeline.process(&LogRecord::new(LogLevel::Info, "app", "hello"));
1508        pipeline.process(&LogRecord::new(LogLevel::Debug, "app", "dropped"));
1509        assert_eq!(handle.count(), 1);
1510        assert_eq!(pipeline.processed_count(), 1);
1511        assert_eq!(pipeline.dropped_count(), 1);
1512    }
1513
1514    #[test]
1515    fn test_pipeline_builder_default_formatter() {
1516        let (handle, output) = MemoryOutputHandle::new();
1517        let pipeline = LogPipelineBuilder::new().output(Box::new(output)).build();
1518        pipeline.process(&LogRecord::new(LogLevel::Info, "app", "hello"));
1519        assert_eq!(handle.count(), 1);
1520        // 默认 JSON 格式器
1521        assert!(handle.entries()[0].contains("\"level\""));
1522    }
1523
1524    #[test]
1525    fn test_pipeline_builder_multiple_outputs() {
1526        let (handle1, output1) = MemoryOutputHandle::new();
1527        let (handle2, output2) = MemoryOutputHandle::new();
1528        let pipeline = LogPipelineBuilder::new()
1529            .formatter(Box::new(TextFormatter::new()))
1530            .outputs(vec![Box::new(output1), Box::new(output2)])
1531            .build();
1532        pipeline.process(&LogRecord::new(LogLevel::Info, "a", "m"));
1533        assert_eq!(handle1.count(), 1);
1534        assert_eq!(handle2.count(), 1);
1535    }
1536
1537    // ---- LogRouter 测试 ----
1538
1539    #[test]
1540    fn test_log_router_basic() {
1541        let (handle1, output1) = MemoryOutputHandle::new();
1542        let (handle2, output2) = MemoryOutputHandle::new();
1543
1544        let pipeline1 = LogPipeline::new(
1545            vec![],
1546            Box::new(TextFormatter::new()),
1547            vec![Box::new(output1)],
1548        );
1549        let pipeline2 = LogPipeline::new(
1550            vec![],
1551            Box::new(TextFormatter::new()),
1552            vec![Box::new(output2)],
1553        );
1554
1555        let router = LogRouter::new()
1556            .route(RoutingRule {
1557                filter: Box::new(TargetFilter::allowlist(&["app"])),
1558                pipeline: pipeline1,
1559                name: "app_rule".to_string(),
1560            })
1561            .route(RoutingRule {
1562                filter: Box::new(TargetFilter::allowlist(&["db"])),
1563                pipeline: pipeline2,
1564                name: "db_rule".to_string(),
1565            });
1566
1567        router.route_record(&LogRecord::new(LogLevel::Info, "app", "app msg"));
1568        router.route_record(&LogRecord::new(LogLevel::Info, "db", "db msg"));
1569        router.route_record(&LogRecord::new(LogLevel::Info, "cache", "unmatched"));
1570
1571        assert_eq!(handle1.count(), 1);
1572        assert_eq!(handle2.count(), 1);
1573        assert_eq!(router.routed_count(), 2);
1574        assert_eq!(router.unmatched_count(), 1);
1575    }
1576
1577    #[test]
1578    fn test_log_router_default_pipeline() {
1579        let (handle, output) = MemoryOutputHandle::new();
1580        let default_pipeline = LogPipeline::new(
1581            vec![],
1582            Box::new(TextFormatter::new()),
1583            vec![Box::new(output)],
1584        );
1585
1586        let router = LogRouter::new().default_pipeline(default_pipeline);
1587        router.route_record(&LogRecord::new(LogLevel::Info, "any", "msg"));
1588        assert_eq!(handle.count(), 1);
1589        assert_eq!(router.routed_count(), 1);
1590        assert_eq!(router.unmatched_count(), 0);
1591    }
1592
1593    #[test]
1594    fn test_log_router_no_match_no_default() {
1595        let router = LogRouter::new();
1596        router.route_record(&LogRecord::new(LogLevel::Info, "any", "msg"));
1597        assert_eq!(router.routed_count(), 0);
1598        assert_eq!(router.unmatched_count(), 1);
1599    }
1600
1601    #[test]
1602    fn test_log_router_batch() {
1603        let (handle, output) = MemoryOutputHandle::new();
1604        let pipeline = LogPipeline::new(
1605            vec![],
1606            Box::new(TextFormatter::new()),
1607            vec![Box::new(output)],
1608        );
1609        let router = LogRouter::new().default_pipeline(pipeline);
1610        let records = vec![
1611            LogRecord::new(LogLevel::Info, "a", "1"),
1612            LogRecord::new(LogLevel::Warn, "b", "2"),
1613        ];
1614        router.route_batch(&records);
1615        assert_eq!(handle.count(), 2);
1616        assert_eq!(router.routed_count(), 2);
1617    }
1618
1619    #[test]
1620    fn test_log_router_first_match_wins() {
1621        let (handle1, output1) = MemoryOutputHandle::new();
1622        let (handle2, output2) = MemoryOutputHandle::new();
1623
1624        let pipeline1 = LogPipeline::new(
1625            vec![],
1626            Box::new(TextFormatter::new()),
1627            vec![Box::new(output1)],
1628        );
1629        let pipeline2 = LogPipeline::new(
1630            vec![],
1631            Box::new(TextFormatter::new()),
1632            vec![Box::new(output2)],
1633        );
1634
1635        // 两个规则都能匹配 Error 级别，第一个规则应胜出
1636        let router = LogRouter::new()
1637            .route(RoutingRule {
1638                filter: Box::new(LevelThresholdFilter::new(LogLevel::Warn)),
1639                pipeline: pipeline1,
1640                name: "warn_plus".to_string(),
1641            })
1642            .route(RoutingRule {
1643                filter: Box::new(LevelThresholdFilter::new(LogLevel::Error)),
1644                pipeline: pipeline2,
1645                name: "error_only".to_string(),
1646            });
1647
1648        router.route_record(&LogRecord::new(LogLevel::Error, "a", "m"));
1649        assert_eq!(handle1.count(), 1);
1650        assert_eq!(handle2.count(), 0);
1651    }
1652
1653    // ---- RateLimitFilter 测试 ----
1654
1655    #[test]
1656    fn test_rate_limit_allows_within_limit() {
1657        let filter = RateLimitFilter::per_second(5);
1658        let record = LogRecord::new(LogLevel::Info, "app", "msg");
1659        for _ in 0..5 {
1660            assert!(filter.should_keep(&record));
1661        }
1662    }
1663
1664    #[test]
1665    fn test_rate_limit_blocks_over_limit() {
1666        let filter = RateLimitFilter::per_second(3);
1667        let record = LogRecord::new(LogLevel::Info, "app", "msg");
1668        for _ in 0..3 {
1669            assert!(filter.should_keep(&record));
1670        }
1671        assert!(!filter.should_keep(&record));
1672    }
1673
1674    #[test]
1675    fn test_rate_limit_per_second_constructor() {
1676        let f = RateLimitFilter::per_second(10);
1677        assert_eq!(f.max_count, 10);
1678        assert_eq!(f.window_ms, 1000);
1679    }
1680
1681    // ---- SamplingFilter 测试 ----
1682
1683    #[test]
1684    fn test_sampling_full_rate() {
1685        let filter = SamplingFilter::new(1.0);
1686        let record = LogRecord::new(LogLevel::Info, "app", "msg");
1687        for _ in 0..100 {
1688            assert!(filter.should_keep(&record));
1689        }
1690    }
1691
1692    #[test]
1693    fn test_sampling_zero_rate() {
1694        let filter = SamplingFilter::new(0.0);
1695        let record = LogRecord::new(LogLevel::Info, "app", "msg");
1696        for _ in 0..100 {
1697            assert!(!filter.should_keep(&record));
1698        }
1699    }
1700
1701    #[test]
1702    fn test_sampling_rate_clamped() {
1703        let filter = SamplingFilter::new(2.0);
1704        assert_eq!(filter.rate(), 1.0);
1705        let filter2 = SamplingFilter::new(-1.0);
1706        assert_eq!(filter2.rate(), 0.0);
1707    }
1708
1709    #[test]
1710    fn test_sampling_ten_percent() {
1711        let filter = SamplingFilter::ten_percent();
1712        assert!((filter.rate() - 0.1).abs() < 1e-10);
1713    }
1714
1715    #[test]
1716    fn test_sampling_one_percent() {
1717        let filter = SamplingFilter::one_percent();
1718        assert!((filter.rate() - 0.01).abs() < 1e-10);
1719    }
1720
1721    // ---- LogAggregator 测试 ----
1722
1723    #[test]
1724    fn test_aggregator_first_output() {
1725        let agg = LogAggregator::new(1000);
1726        assert!(agg.should_output("error: db connection failed"));
1727    }
1728
1729    #[test]
1730    fn test_aggregator_suppresses_duplicates() {
1731        let agg = LogAggregator::new(1000);
1732        assert!(agg.should_output("error: timeout"));
1733        assert!(!agg.should_output("error: timeout"));
1734        assert!(!agg.should_output("error: timeout"));
1735        assert_eq!(agg.count("error: timeout"), 3);
1736    }
1737
1738    #[test]
1739    fn test_aggregator_different_messages() {
1740        let agg = LogAggregator::new(1000);
1741        assert!(agg.should_output("error A"));
1742        assert!(agg.should_output("error B"));
1743        assert!(!agg.should_output("error A"));
1744        assert!(!agg.should_output("error B"));
1745    }
1746
1747    #[test]
1748    fn test_aggregator_clear() {
1749        let agg = LogAggregator::new(1000);
1750        agg.should_output("msg");
1751        assert_eq!(agg.count("msg"), 1);
1752        agg.clear();
1753        assert_eq!(agg.count("msg"), 0);
1754    }
1755
1756    #[test]
1757    fn test_aggregator_count_unknown() {
1758        let agg = LogAggregator::new(1000);
1759        assert_eq!(agg.count("unknown"), 0);
1760    }
1761
1762    // ---- LogBuffer 测试 ----
1763
1764    #[test]
1765    fn test_log_buffer_push_and_flush() {
1766        let buf = LogBuffer::new(3);
1767        assert!(buf.is_empty());
1768        buf.push(LogRecord::new(LogLevel::Info, "a", "1"));
1769        buf.push(LogRecord::new(LogLevel::Info, "a", "2"));
1770        assert_eq!(buf.len(), 2);
1771        let flushed = buf.flush();
1772        assert_eq!(flushed.len(), 2);
1773        assert!(buf.is_empty());
1774    }
1775
1776    #[test]
1777    fn test_log_buffer_threshold() {
1778        let buf = LogBuffer::new(2);
1779        assert!(!buf.push(LogRecord::new(LogLevel::Info, "a", "1")));
1780        assert!(buf.push(LogRecord::new(LogLevel::Info, "a", "2")));
1781    }
1782
1783    #[test]
1784    fn test_log_buffer_capacity() {
1785        let buf = LogBuffer::new(5);
1786        assert_eq!(buf.capacity(), 5);
1787    }
1788
1789    #[test]
1790    fn test_log_buffer_capacity_clamped() {
1791        let buf = LogBuffer::new(0);
1792        assert_eq!(buf.capacity(), 1);
1793    }
1794}
sz_orm_logger/log_pipeline.rs

sz_orm_logger/
log_pipeline.rs
