From eab82b43d528a0d490dbf756e943b6d95e3f6298 Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Sat, 18 Jul 2026 01:45:02 -0300 Subject: [PATCH] docs: add rustdoc documentation --- src/config.rs | 25 +++++++++++++++++++++++++ src/filter.rs | 35 +++++++++++++++++++++++++++++++++++ src/lib.rs | 12 ++++++++++++ src/parser.rs | 9 +++++++++ src/record.rs | 4 ++++ src/sink.rs | 10 ++++++++++ src/source.rs | 15 +++++++++++---- 7 files changed, 106 insertions(+), 4 deletions(-) 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 449377a..10064fb 100644 --- a/src/source.rs +++ b/src/source.rs @@ -7,12 +7,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. 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 reader = BufReader::new(file);