links_logs_vault/
pipeline.rs

1use crate::{LogLevel, LogRecord, Vault, VaultError};
2
3/// A Logstash-style event transformation. `None` drops the event.
4pub trait Filter {
5    /// Transforms, passes, or drops one event.
6    fn apply(&mut self, record: LogRecord) -> Option<LogRecord>;
7}
8
9/// Drops events below a configured severity.
10pub struct MinimumLevel {
11    minimum: LogLevel,
12}
13
14impl MinimumLevel {
15    /// Creates a severity filter.
16    #[must_use]
17    pub const fn new(minimum: LogLevel) -> Self {
18        Self { minimum }
19    }
20}
21
22impl Filter for MinimumLevel {
23    fn apply(&mut self, record: LogRecord) -> Option<LogRecord> {
24        (record.level >= self.minimum).then_some(record)
25    }
26}
27
28/// Enriches every passing event with a fixed field.
29pub struct AddField {
30    key: String,
31    value: String,
32}
33
34impl AddField {
35    /// Creates an enrichment filter.
36    #[must_use]
37    pub fn new(key: impl Into<String>, value: impl Into<String>) -> Self {
38        Self {
39            key: key.into(),
40            value: value.into(),
41        }
42    }
43}
44
45impl Filter for AddField {
46    fn apply(&mut self, mut record: LogRecord) -> Option<LogRecord> {
47        record.fields.insert(self.key.clone(), self.value.clone());
48        Some(record)
49    }
50}
51
52/// Ordered filters followed by an associative-vault output.
53pub struct Pipeline {
54    vault: Vault,
55    filters: Vec<Box<dyn Filter>>,
56}
57
58impl Pipeline {
59    /// Creates a pipeline whose output is `vault`.
60    #[must_use]
61    pub fn new(vault: Vault) -> Self {
62        Self {
63            vault,
64            filters: Vec::new(),
65        }
66    }
67
68    /// Appends a filter. Filters run in insertion order.
69    pub fn add_filter(&mut self, filter: impl Filter + 'static) {
70        self.filters.push(Box::new(filter));
71    }
72
73    /// Runs one event through all filters and stores it unless it is dropped.
74    pub fn process(&mut self, mut record: LogRecord) -> Result<Option<u64>, VaultError> {
75        for filter in &mut self.filters {
76            let Some(filtered) = filter.apply(record) else {
77                return Ok(None);
78            };
79            record = filtered;
80        }
81        self.vault.append(record).map(Some)
82    }
83
84    /// Borrows the pipeline output.
85    #[must_use]
86    pub const fn vault(&self) -> &Vault {
87        &self.vault
88    }
89
90    /// Mutably borrows the pipeline output.
91    #[must_use]
92    pub const fn vault_mut(&mut self) -> &mut Vault {
93        &mut self.vault
94    }
95
96    /// Returns the output vault.
97    #[must_use]
98    pub fn into_vault(self) -> Vault {
99        self.vault
100    }
101}
links_logs_vault/pipeline.rs

links_logs_vault/
pipeline.rs
