diff --git a/src/config.rs b/src/config.rs index aa3f708..196fedc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,40 +1,62 @@ +//! Configuration management. +//! +//! This module defines the application's configuration structure and provides +//! functionality to load configuration values from a TOML file. +//! +//! Missing fields are automatically populated with sensible default values +//! through Serde's `default` attribute. + use crate::filter::FilterRule; use anyhow::Result; use serde::Deserialize; use std::path::PathBuf; +/// Application configuration loaded from TOML file. #[derive(Debug, Deserialize, Clone)] pub struct Config { + /// Path to the input souce. pub source_path: PathBuf, + + /// URL of the destination sink. pub sink_url: String, + /// Number of events processed in each batch. #[serde(default = "default_batch_size")] pub batch_size: usize, + /// Time in seconds before flushing pending events. #[serde(default = "default_flush_interval")] pub flush_interval_secs: u64, + /// Maximum number off buffered messages. #[serde(default = "default_channel_capacity")] pub channel_capacity: usize, + /// Maximum retry attempts after a failed send. #[serde(default = "default_max_retries")] pub max_retries: u32, + /// Initial retry delay in millyseconds. #[serde(default = "default_retry_backoff_initial_ms")] pub retry_backoff_initial_ms: u64, + /// Maximum retry backoff delay in seconds. #[serde(default = "default_retry_backoff_max_secs")] pub retry_backoff_max_secs: u64, + /// Event filtering rules. Empty means no filtering. #[serde(default)] pub filter_rules: Vec, + /// Enable masking of common sensitive patterns. #[serde(default = "default_mask_common_patterns")] pub mask_common_patterns: bool, + /// Maximum dead-letter storage size in bytes. #[serde(default = "default_dead_letter_max_bytes")] pub dead_letter_max_bytes: usize, + /// Maximum number of dead-letter files kept; #[serde(default = "default_dead_letter_max_files")] pub dead_letter_max_files: u64, } @@ -70,6 +92,9 @@ fn default_dead_letter_max_files() -> u64 { } impl Config { + /// Loads configuration from a TOML file. + /// + /// Missing optional fields use default values. pub fn load(path: &str) -> Result { let text = std::fs::read_to_string(path) .map_err(|e| anyhow::anyhow!("não consegui ler {path}: {e}"))?; diff --git a/src/filter.rs b/src/filter.rs index d9f9e5e..ada0d02 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -3,29 +3,64 @@ use regex::Regex; use serde::Deserialize; use tokio::sync::mpsc::{Receiver, Sender}; +/// Operation used to compare a field value. #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "lowercase")] pub enum RuleOp { + /// Matches when the field value is exactly. Equals, + + /// Matches when the field value contains the configured value. Contains, + + /// Matches when the field value satisfies the configured regular expression. Regex, } +/// Action performed when a filter rule matches. #[derive(Debug, Deserialize, Clone)] #[serde(rename_all = "lowercase")] pub enum RuleAction { + /// Remove the entire log entry from the processing pipeline. Drop, + + /// Replaces the matching field value with a masked value. Mask, } +/// A single filtering rule applied to a log entry. +/// +/// Rules define which field should be checked, how the value should be +/// compared, and what action should be performed when a match occurs. #[derive(Debug, Deserialize, Clone)] pub struct FilterRule { + /// JSON field name to inspect. pub field: String, + /// Comparison operation used to evaluate the field. + pub op: RuleOp, + /// Value or pattern used for maching. pub value: String, + + /// Action executed when the rule matches pub action: RuleAction, } +/// Runs the log filtering pipeline. +/// +/// Receives log entries from a channel, applies built-in masking patterns +/// and configured filter rules, then forwards accepted logs to the output channel. +/// +/// Logs can either be dropped, masked, or passed through unchanged depending +/// on the configured rules. +/// +/// # Arguments +/// +/// * `rx` - Input channel receiving log entries. +/// * `tx` - Output channel for filtered log entries. +/// * `rules` - User-defined filtering rules. +/// * `mask_common_patterns` - Enables built-in masking of sensitive patterns +/// such as emails, credit card numbers, and API keys. pub async fn run_filter( mut rx: Receiver, tx: Sender, diff --git a/src/lib.rs b/src/lib.rs index 35782c0..07afec5 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,18 @@ pub mod source; pub use config::Config; pub use record::LogLine; +/// Application entry point that starts the processing pipeline. +/// +/// Creates the communication channels and launches the source, parser, +/// filter, and sink tasks. +/// +/// The pipeline flow is: +/// +/// ```text +/// source -> parser -> filter -> sink +/// ``` +/// +/// Tasks communicate through bounded channels using the configured capacity. pub async fn run(cfg: Config) -> anyhow::Result<()> { let (raw_tx, raw_rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity); let (parsed_tx, parsed_rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity); diff --git a/src/parser.rs b/src/parser.rs index 18760f8..f960c56 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -2,6 +2,10 @@ use crate::record::LogLine; use serde_json::json; use tokio::sync::mpsc::{Receiver, Sender}; +/// Parses incoming log lines and forwards structured log entries. +/// +/// Receives raw text lines, converts them into `LogLine` values, and sends +/// the parsed entries to the next stage of the pipeline. pub async fn run_parser(mut rx: Receiver, tx: Sender) { while let Some(line) = rx.recv().await { let log_line = parse_line(&line); @@ -11,6 +15,11 @@ pub async fn run_parser(mut rx: Receiver, tx: Sender) { } } +/// Converts a raw log line into a structured JSON log entry. +/// +/// Valid JSON objects are returned unchanged. +/// Valid JSON values that are not objects and invalid JSON are wrapped into +/// a fallback log entry containing the original message and a parse issue. pub fn parse_line(line: &str) -> LogLine { match serde_json::from_str::(line) { Ok(value) if value.is_object() => value, diff --git a/src/record.rs b/src/record.rs index f4f1a32..b14e992 100644 --- a/src/record.rs +++ b/src/record.rs @@ -1 +1,5 @@ +// Represents a single log entry as a JSON value. +/// +/// Each log line is expected to contain a JSON object with fields that can be +/// processed by filters and sent to the configured sink. pub type LogLine = serde_json::Value; diff --git a/src/sink.rs b/src/sink.rs index 8362425..8b264f0 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -8,8 +8,18 @@ use std::time::Duration; use tokio::sync::mpsc::Receiver; use tokio::time::{interval, sleep}; +/// Path used to store events that could not be delivered. const DEAD_LETTER_PATH: &str = "logtap.failed.jsonl"; +/// Runs the sink pipeline. +/// +/// Receives log entries from a channel, groups them into batches, and sends +/// them to the configured destination. Failed batches are handled by the +/// retry/dead-letter mechanism. +/// +/// The sink flushes data when either: +/// - the configured batch size is reached; +/// - the flush interval expires. pub async fn run_sink(cfg: Config, mut rx: Receiver) { let client = reqwest::Client::new(); let mut batch: Vec = Vec::with_capacity(cfg.batch_size); diff --git a/src/source.rs b/src/source.rs index 7ab7cf3..6c3bb89 100644 --- a/src/source.rs +++ b/src/source.rs @@ -9,57 +9,19 @@ use std::sync::mpsc as std_mpsc; use std::time::Duration; use tokio::sync::mpsc::Sender; -// How often the watch loop wakes up even without a filesystem event, just to -// check whether the downstream receiver is gone, and whether the file at -// source_path has been swapped out from under us (log rotation). Without -// this, the loop can block forever on `notify_rx.recv()` once there's no -// more filesystem activity on the file it's actually still holding open. +/// Interval used to periodically check whether the output channel is closed. +/// +/// Without this timeout, the watcher loop could block forever waiting for a +/// filesystem event after the rest of the pipeline has already shut down. const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(200); -/// Whether the caller should keep going or the downstream receiver is gone. -enum DrainOutcome { - Continue, - ReceiverGone, -} - -/// Reads every complete line available from `offset` onward and forwards it -/// down the channel, advancing `offset` as it goes. -fn drain_new_lines( - reader: &mut BufReader, - offset: &mut u64, - tx: &Sender, - line: &mut String, -) -> Result { - reader.seek(SeekFrom::Start(*offset))?; - - loop { - line.clear(); - let bytes_read = reader.read_line(line)?; - - if bytes_read == 0 { - return Ok(DrainOutcome::Continue); - } - - *offset += bytes_read as u64; - let trimmed = line.trim_end().to_string(); - - if tx.blocking_send(trimmed).is_err() { - return Ok(DrainOutcome::ReceiverGone); - } - } -} - -fn watch( - path: &Path, - notify_tx: std_mpsc::Sender>, -) -> Result { - let mut watcher: RecommendedWatcher = notify::recommended_watcher(move |res| { - let _ = notify_tx.send(res); - })?; - watcher.watch(path, RecursiveMode::NonRecursive)?; - Ok(watcher) -} - +/// Reads new lines from a file as it changes and sends them to the pipeline. +/// +/// The source starts reading from the current end of the file and only +/// processes lines appended after startup. +/// +/// The function watches the file for modifications and stops when the +/// receiver side is closed. pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { let file = File::open(&cfg.source_path)?; let mut inode = file.metadata()?.ino();