From 9d194a7e545e6c9ffa8e39629a9ecadac1373d2c Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Sat, 18 Jul 2026 02:45:59 -0300 Subject: [PATCH 1/2] docs: stardadize rust docs --- src/config.rs | 24 +++++++++++++++++++++++ src/filter.rs | 36 ++++++++++++++++++++++++++++++++++ src/lib.rs | 13 +++++++++++++ src/parser.rs | 9 +++++++++ src/sink.rs | 53 +++++++++++++++++++++++++++++++++++++++++---------- src/source.rs | 15 +++++++++++++++ 6 files changed, 140 insertions(+), 10 deletions(-) diff --git a/src/config.rs b/src/config.rs index aa3f708..e8edd50 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,40 +1,61 @@ +//! 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 size of dead-letter data in bytes. #[serde(default = "default_dead_letter_max_bytes")] pub dead_letter_max_bytes: usize, + /// Maximum number of dead-letter files to retain. #[serde(default = "default_dead_letter_max_files")] pub dead_letter_max_files: u64, } @@ -70,6 +91,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..78ccf0d 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -3,29 +3,65 @@ 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..ac75904 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -8,6 +8,19 @@ 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/sink.rs b/src/sink.rs index 8362425..4fe6c9c 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); @@ -32,6 +42,13 @@ pub async fn run_sink(cfg: Config, mut rx: Receiver) { } } +/// Attempts to deliver a batch to the configured sink endpoint. +/// +/// Failed requests are retried using an exponential backoff strategy limited +/// by the configured maximum retry count and maximum backoff duration. +/// +/// When all retry attempts are exhausted, the batch is moved to the +/// dead-letter file so that failed events are not silently lost. async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Vec) { if batch.is_empty() { return; @@ -88,10 +105,14 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve } } -// Reopens the file on every call instead of keeping a long-lived handle — -// a handle stays bound to the underlying inode even if the file is later -// renamed out from under it (e.g. for a manual replay), which would make -// new failures silently land in the renamed-away file. +/// Writes failed log entries to the dead-letter file. +/// +/// The file is opened on every call instead of keeping a long-lived handle. +/// This ensures new failures are written to the current dead-letter path even +/// if the file has been rotated or renamed externally. +/// +/// Before writing, the current dead-letter file is rotated if it exceeds the +/// configured size limit. fn write_dead_letter(cfg: &Config, batch: &[LogLine]) { rotate_dead_letter_if_full(cfg); @@ -118,16 +139,28 @@ fn write_dead_letter(cfg: &Config, batch: &[LogLine]) { } } + + +/// Rotates the dead-letter file when it exceeds the configured size limit. +/// +/// Rotation keeps a bounded number of historical files by shifting older +/// files to the next index and removing the oldest one when the limit is +/// reached. +/// +/// When `dead_letter_max_files` is set to zero, the current dead-letter file +/// is discarded instead of being rotated.s fn rotated_dead_letter_path(n: u64) -> PathBuf { PathBuf::from(format!("{DEAD_LETTER_PATH}.{n}")) } -// Caps how big logtap.failed.jsonl is allowed to get. Same idea as -// logrotate: once the current file crosses dead_letter_max_bytes, it's -// shifted into .1, the old .1 becomes .2, and so on up to -// dead_letter_max_files — at which point the oldest file is discarded to -// make room. That eviction is real, permanent data loss, so it's always -// logged loudly rather than happening quietly. +/// Rotates the dead-letter file when it exceeds the configured size limit. +/// +/// Rotation keeps a bounded number of historical files by shifting older +/// files to the next index and removing the oldest one when the limit is +/// reached. +/// +/// When `dead_letter_max_files` is set to zero, the current dead-letter file +/// is discarded instead of being rotated. fn rotate_dead_letter_if_full(cfg: &Config) { let current_size = match fs::metadata(DEAD_LETTER_PATH) { Ok(meta) => meta.len(), diff --git a/src/source.rs b/src/source.rs index 7ab7cf3..c677607 100644 --- a/src/source.rs +++ b/src/source.rs @@ -49,6 +49,10 @@ fn drain_new_lines( } } +/// Creates a filesystem watcher for the given path. +/// +/// The watcher sends filesystem events through the provided channel. +/// Only events from the target file itself are monitored. fn watch( path: &Path, notify_tx: std_mpsc::Sender>, @@ -60,6 +64,17 @@ fn watch( Ok(watcher) } +/// Starts watching a source file and forwards newly appended lines to the +/// provided channel. +/// +/// The function begins reading from the current end of the file and waits for +/// filesystem modification events to detect new content. +/// +/// When the file is replaced during log rotation, the old file is drained, +/// the new file is opened, and reading continues from the beginning of the +/// replacement file. +/// +/// Processing stops when the downstream receiver is closed. pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { let file = File::open(&cfg.source_path)?; let mut inode = file.metadata()?.ino(); From b59e3ea8b8a887724d51fd54e29986afe4fd6f39 Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Sun, 26 Jul 2026 18:13:05 -0300 Subject: [PATCH 2/2] docs: fix typos in documentation --- src/config.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/config.rs b/src/config.rs index e8edd50..35b536f 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,7 +14,7 @@ use std::path::PathBuf; /// Application configuration loaded from TOML file. #[derive(Debug, Deserialize, Clone)] pub struct Config { - /// Path to the input souce. + /// Path to the input source. pub source_path: PathBuf, /// URL of the destination sink. pub sink_url: String, @@ -27,7 +27,7 @@ pub struct Config { #[serde(default = "default_flush_interval")] pub flush_interval_secs: u64, - /// Maximum number off buffered messages. + /// Maximum number of buffered messages. #[serde(default = "default_channel_capacity")] pub channel_capacity: usize, @@ -35,7 +35,7 @@ pub struct Config { #[serde(default = "default_max_retries")] pub max_retries: u32, - /// Initial retry delay in millyseconds. + /// Initial retry delay in milliseconds. #[serde(default = "default_retry_backoff_initial_ms")] pub retry_backoff_initial_ms: u64,