From 93f0187e253c629583d92819a431eb71138e57ae Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Sat, 18 Jul 2026 02:07:54 -0300 Subject: [PATCH] docs: add rustdoc documentation --- src/config.rs | 39 +++++++++++++++++ src/filter.rs | 35 +++++++++++++++ src/lib.rs | 12 ++++++ src/parser.rs | 9 ++++ src/record.rs | 4 ++ src/sink.rs | 117 +++++++++++++++++++++++++++++++++++++++++++++++--- src/source.rs | 70 +++++++++++++++++++----------- 7 files changed, 256 insertions(+), 30 deletions(-) diff --git a/src/config.rs b/src/config.rs index 0d67b6d..196fedc 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,36 +1,64 @@ +//! 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, } fn default_batch_size() -> usize { @@ -55,7 +83,18 @@ fn default_mask_common_patterns() -> bool { true } +fn default_dead_letter_max_bytes() -> usize { + 1024 * 1024 * 1024 +} + +fn default_dead_letter_max_files() -> u64 { + 5 +} + 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 fb3178d..8b264f0 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -1,9 +1,25 @@ use crate::config::Config; use crate::record::LogLine; +use std::fs; +use std::fs::OpenOptions; +use std::io::Write; +use std::path::PathBuf; 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); @@ -41,7 +57,7 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve } Ok(resp) => { eprintln!( - "logtap: tentativa {}/{} falhou — destino respondeu com status {}", + "logtap: attempt {}/{} failed — server responded with status {}", attempt + 1, cfg.max_retries, resp.status() @@ -49,7 +65,7 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve } Err(err) => { eprintln!( - "logtap: tentativa {}/{} falhou — erro ao enviar lote ({} itens): {err}", + "logtap: attempt {}/{} failed — error sending batch ({} items): {err}", attempt + 1, cfg.max_retries, batch.len() @@ -60,13 +76,13 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve attempt += 1; if attempt >= cfg.max_retries { - // TODO(fase 1 do roadmap): dead-letter em vez de descartar aqui. - // Por enquanto o lote é perdido depois de esgotar as tentativas. eprintln!( - "logtap: desistindo do lote ({} itens) após {} tentativas", + "logtap: abandoning batch of {} items after {} attempts — writing to {DEAD_LETTER_PATH}", batch.len(), cfg.max_retries ); + + write_dead_letter(cfg, batch); batch.clear(); return; } @@ -77,7 +93,96 @@ async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Ve let backoff = Duration::from_millis(backoff_ms).min(Duration::from_secs(cfg.retry_backoff_max_secs)); - eprintln!("logtap: esperando {backoff:?} antes da próxima tentativa"); + eprintln!("logtap: waiting {backoff:?} before next attempt"); sleep(backoff).await; } } + +// 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. +fn write_dead_letter(cfg: &Config, batch: &[LogLine]) { + rotate_dead_letter_if_full(cfg); + + let mut file = match OpenOptions::new() + .create(true) + .append(true) + .open(DEAD_LETTER_PATH) + { + Ok(file) => file, + Err(err) => { + eprintln!( + "logtap: could not open {DEAD_LETTER_PATH} ({err}) — {} item(s) lost for good", + batch.len() + ); + return; + } + }; + + for log in batch { + if let Err(err) = writeln!(file, "{log}") { + eprintln!("logtap: failed writing to {DEAD_LETTER_PATH}: {err}"); + return; + } + } +} + +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. +fn rotate_dead_letter_if_full(cfg: &Config) { + let current_size = match fs::metadata(DEAD_LETTER_PATH) { + Ok(meta) => meta.len(), + Err(_) => return, // no file yet — nothing to rotate + }; + + if current_size < cfg.dead_letter_max_bytes as u64 { + return; + } + + if cfg.dead_letter_max_files == 0 { + if let Err(err) = fs::remove_file(DEAD_LETTER_PATH) { + eprintln!("logtap: failed to reset full dead-letter file: {err}"); + } else { + eprintln!( + "logtap: dead-letter file hit {current_size} bytes and dead_letter_max_files is 0 — discarding it entirely" + ); + } + return; + } + + let oldest = rotated_dead_letter_path(cfg.dead_letter_max_files); + if oldest.exists() { + eprintln!( + "logtap: dead-letter rotation limit ({} files) reached — discarding oldest file {}", + cfg.dead_letter_max_files, + oldest.display() + ); + } + + for n in (1..cfg.dead_letter_max_files).rev() { + let from = rotated_dead_letter_path(n); + if from.exists() { + let to = rotated_dead_letter_path(n + 1); + if let Err(err) = fs::rename(&from, &to) { + eprintln!( + "logtap: failed to rotate {} -> {}: {err}", + from.display(), + to.display() + ); + } + } + } + + if let Err(err) = fs::rename(DEAD_LETTER_PATH, rotated_dead_letter_path(1)) { + eprintln!("logtap: failed to rotate {DEAD_LETTER_PATH}: {err}"); + } +} diff --git a/src/source.rs b/src/source.rs index 449377a..6c3bb89 100644 --- a/src/source.rs +++ b/src/source.rs @@ -3,26 +3,33 @@ use anyhow::Result; use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use std::fs::File; use std::io::{BufRead, BufReader, Seek, SeekFrom}; +use std::os::unix::fs::MetadataExt; +use std::path::Path; 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. Without this, the loop can -// block forever on `notify_rx.recv()` after the rest of the pipeline shut -// down, since there is no more filesystem activity left to notice it. +/// 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); +/// 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(); let mut reader = BufReader::new(file); let mut offset = reader.seek(SeekFrom::End(0))?; let (notify_tx, notify_rx) = std_mpsc::channel::>(); - let mut watcher: RecommendedWatcher = notify::recommended_watcher(move |res| { - let _ = notify_tx.send(res); - })?; - watcher.watch(&cfg.source_path, RecursiveMode::NonRecursive)?; + let mut _watcher = watch(&cfg.source_path, notify_tx.clone())?; let mut line = String::new(); @@ -33,6 +40,33 @@ pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { if tx.is_closed() { return Ok(()); } + + // The path may briefly not resolve mid-rotation (rename in + // progress) — that's transient, just retry on the next tick. + let Ok(current_inode) = std::fs::metadata(&cfg.source_path).map(|m| m.ino()) else { + continue; + }; + + if current_inode == inode { + continue; + } + + // The file at source_path is no longer the one we have open + // (standard logrotate "rename + create new" behavior). Flush + // whatever was written to the old file right before the + // swap, then switch to the new one from the start. + if let DrainOutcome::ReceiverGone = + drain_new_lines(&mut reader, &mut offset, &tx, &mut line)? + { + return Ok(()); + } + + let new_file = File::open(&cfg.source_path)?; + inode = new_file.metadata()?.ino(); + reader = BufReader::new(new_file); + offset = 0; + _watcher = watch(&cfg.source_path, notify_tx.clone())?; + continue; } Err(std_mpsc::RecvTimeoutError::Disconnected) => return Ok(()), @@ -42,22 +76,10 @@ pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { continue; } - reader.seek(SeekFrom::Start(offset))?; - - loop { - line.clear(); - let bytes_read = reader.read_line(&mut line)?; - - if bytes_read == 0 { - break; - } - - offset += bytes_read as u64; - let trimmed = line.trim_end().to_string(); - - if tx.blocking_send(trimmed).is_err() { - return Ok(()); - } + if let DrainOutcome::ReceiverGone = + drain_new_lines(&mut reader, &mut offset, &tx, &mut line)? + { + return Ok(()); } } }