diff --git a/Cargo.lock b/Cargo.lock index 251a795..c68720e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -531,6 +531,7 @@ dependencies = [ "serde", "serde_json", "tokio", + "toml", ] [[package]] @@ -859,6 +860,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_spanned" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6662b5879511e06e8999a8a235d848113e942c9124f211511b16466ee2995f26" +dependencies = [ + "serde_core", +] + [[package]] name = "serde_urlencoded" version = "0.7.1" @@ -1035,6 +1045,45 @@ dependencies = [ "tokio", ] +[[package]] +name = "toml" +version = "1.1.3+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c96ecdfa941c8fc4fcaed14f99ada8ebed502eef533015095a07e3301d4c3c" +dependencies = [ + "indexmap", + "serde_core", + "serde_spanned", + "toml_datetime", + "toml_parser", + "toml_writer", + "winnow", +] + +[[package]] +name = "toml_datetime" +version = "1.1.1+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3165f65f62e28e0115a00b2ebdd37eb6f3b641855f9d636d3cd4103767159ad7" +dependencies = [ + "serde_core", +] + +[[package]] +name = "toml_parser" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +dependencies = [ + "winnow", +] + +[[package]] +name = "toml_writer" +version = "1.1.2+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" + [[package]] name = "tower-service" version = "0.3.3" @@ -1355,6 +1404,12 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" +[[package]] +name = "winnow" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" + [[package]] name = "winreg" version = "0.50.0" diff --git a/Cargo.toml b/Cargo.toml index 2c6a8f3..43bf336 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,4 +10,5 @@ serde = { version = "1", features = ["derive"] } serde_json = "1" regex = "1" reqwest = { version = "0.11", default-features = false, features = ["json", "rustls-tls"] } -anyhow = "1" \ No newline at end of file +anyhow = "1" +toml = "1.1.3+spec-1.1.0" \ No newline at end of file diff --git a/src/config.rs b/src/config.rs index e4a6dc4..0d67b6d 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,4 +1,5 @@ use crate::filter::FilterRule; +use anyhow::Result; use serde::Deserialize; use std::path::PathBuf; @@ -6,14 +7,59 @@ use std::path::PathBuf; pub struct Config { pub source_path: PathBuf, pub sink_url: String, + + #[serde(default = "default_batch_size")] pub batch_size: usize, + + #[serde(default = "default_flush_interval")] pub flush_interval_secs: u64, + + #[serde(default = "default_channel_capacity")] pub channel_capacity: usize, - #[serde(default = "filter_rules_default")] + #[serde(default = "default_max_retries")] + pub max_retries: u32, + + #[serde(default = "default_retry_backoff_initial_ms")] + pub retry_backoff_initial_ms: u64, + + #[serde(default = "default_retry_backoff_max_secs")] + pub retry_backoff_max_secs: u64, + + #[serde(default)] pub filter_rules: Vec, + + #[serde(default = "default_mask_common_patterns")] + pub mask_common_patterns: bool, +} + +fn default_batch_size() -> usize { + 50 +} +fn default_flush_interval() -> u64 { + 5 +} +fn default_channel_capacity() -> usize { + 1000 +} +fn default_max_retries() -> u32 { + 5 +} +fn default_retry_backoff_initial_ms() -> u64 { + 500 +} +fn default_retry_backoff_max_secs() -> u64 { + 30 +} +fn default_mask_common_patterns() -> bool { + true } -fn filter_rules_default() -> Vec { - vec![] +impl Config { + 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}"))?; + let cfg: Config = toml::from_str(&text)?; + Ok(cfg) + } } diff --git a/src/filter.rs b/src/filter.rs index ff5b779..d9f9e5e 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -1,9 +1,10 @@ -use crate::record::LogLine; +use crate::LogLine; use regex::Regex; use serde::Deserialize; use tokio::sync::mpsc::{Receiver, Sender}; #[derive(Debug, Deserialize, Clone)] +#[serde(rename_all = "lowercase")] pub enum RuleOp { Equals, Contains, @@ -18,7 +19,6 @@ pub enum RuleAction { } #[derive(Debug, Deserialize, Clone)] -#[serde(rename_all = "lowercase")] pub struct FilterRule { pub field: String, pub op: RuleOp, @@ -26,16 +26,39 @@ pub struct FilterRule { pub action: RuleAction, } -pub async fn run_filter(mut rx: Receiver, tx: Sender, rules: Vec) { +pub async fn run_filter( + mut rx: Receiver, + tx: Sender, + rules: Vec, + mask_common_patterns: bool, +) { + let email_re = Regex::new(r"(?i)\b([a-z0-9._%+-])[a-z0-9._%+-]*(@[a-z0-9.-]+\.[a-z]{2,})\b") + .expect("invalid regex for email"); + let card_re = Regex::new(r"\b(?:\d[ -]?){13,19}\b").expect("invalid regex for card"); + let api_key_re = Regex::new(r"\bsk-[A-Za-z0-9]{16,}\b").expect("invalid regex for API key"); + + let builtin_patterns: Vec<(&Regex, &str)> = vec![ + (&email_re, "$1***$2"), + (&card_re, "[card-masked]"), + (&api_key_re, "[api-key-masked]"), + ]; + while let Some(mut log) = rx.recv().await { + if mask_common_patterns { + mask_builtin(&mut log, &builtin_patterns); + } + let mut dropped = false; for rule in &rules { let field_val = log.get(&rule.field).and_then(|v| v.as_str()).unwrap_or(""); + let matched = match rule.op { RuleOp::Equals => field_val == rule.value, RuleOp::Contains => field_val.contains(&rule.value), - RuleOp::Regex => Regex::new(&rule.value).unwrap().is_match(field_val), + RuleOp::Regex => Regex::new(&rule.value) + .map(|re| re.is_match(field_val)) + .unwrap_or(false), }; if matched { @@ -56,3 +79,19 @@ pub async fn run_filter(mut rx: Receiver, tx: Sender, rules: V } } } + +fn mask_builtin(log: &mut LogLine, patterns: &[(&Regex, &str)]) { + if let Some(obj) = log.as_object_mut() { + for (_, v) in obj.iter_mut() { + if let Some(s) = v.as_str() { + let mut masked = s.to_string(); + for (re, replacement) in patterns { + masked = re.replace_all(&masked, *replacement).to_string(); + } + if masked != s { + *v = serde_json::Value::String(masked); + } + } + } + } +} diff --git a/src/lib.rs b/src/lib.rs index ea49d22..35782c0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,13 +14,18 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> { let (clean_tx, clean_rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity); let source_cfg = cfg.clone(); - let source_handle = - tokio::task::spawn_blocking(move || source::run_source(source_cfg, raw_tx)); + let mask_common_patterns = source_cfg.mask_common_patterns; + let source_handle = tokio::task::spawn_blocking(move || source::run_source(source_cfg, raw_tx)); let parser_handle = tokio::spawn(parser::run_parser(raw_rx, parsed_tx)); let rules = cfg.filter_rules.clone(); - let filter_handle = tokio::spawn(filter::run_filter(parsed_rx, clean_tx, rules)); + let filter_handle = tokio::spawn(filter::run_filter( + parsed_rx, + clean_tx, + rules, + mask_common_patterns, + )); let sink_cfg = cfg.clone(); let sink_handle = tokio::spawn(sink::run_sink(sink_cfg, clean_rx)); @@ -32,4 +37,4 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> { } Ok(()) -} \ No newline at end of file +} diff --git a/src/sink.rs b/src/sink.rs index bde31b6..fb3178d 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -2,27 +2,8 @@ use crate::config::Config; use crate::record::LogLine; use std::time::Duration; use tokio::sync::mpsc::Receiver; -use tokio::time::interval; +use tokio::time::{interval, sleep}; -async fn flush(client: &reqwest::Client, url: &str, batch: &mut Vec) { - if batch.is_empty() { - return; - } - - match client.post(url).json(batch).send().await { - Ok(resp) if resp.status().is_success() => {} - Ok(resp) => eprintln!( - "logtap: destination responded with status {}", - resp.status() - ), - Err(err) => eprintln!( - "logtap: failed to send batch ({} items): {err}", - batch.len() - ), - } - - batch.clear(); -} 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); @@ -33,14 +14,70 @@ pub async fn run_sink(cfg: Config, mut rx: Receiver) { Some(log) = rx.recv() => { batch.push(log); if batch.len() >= cfg.batch_size { - flush(&client, &cfg.sink_url, &mut batch).await; + flush_with_retry(&client, &cfg, &mut batch).await; } } _ = ticker.tick() => { if !batch.is_empty() { - flush(&client, &cfg.sink_url, &mut batch).await; + flush_with_retry(&client, &cfg, &mut batch).await; } } } } } + +async fn flush_with_retry(client: &reqwest::Client, cfg: &Config, batch: &mut Vec) { + if batch.is_empty() { + return; + } + + let mut attempt: u32 = 0; + + loop { + match client.post(&cfg.sink_url).json(batch).send().await { + Ok(resp) if resp.status().is_success() => { + batch.clear(); + return; + } + Ok(resp) => { + eprintln!( + "logtap: tentativa {}/{} falhou — destino respondeu com status {}", + attempt + 1, + cfg.max_retries, + resp.status() + ); + } + Err(err) => { + eprintln!( + "logtap: tentativa {}/{} falhou — erro ao enviar lote ({} itens): {err}", + attempt + 1, + cfg.max_retries, + batch.len() + ); + } + } + + 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", + batch.len(), + cfg.max_retries + ); + batch.clear(); + return; + } + + let backoff_ms = cfg + .retry_backoff_initial_ms + .saturating_mul(1u64 << (attempt - 1)); + 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"); + sleep(backoff).await; + } +} diff --git a/src/source.rs b/src/source.rs index eafc6b0..449377a 100644 --- a/src/source.rs +++ b/src/source.rs @@ -4,8 +4,15 @@ use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use std::fs::File; use std::io::{BufRead, BufReader, Seek, SeekFrom}; 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. +const SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(200); + pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { let file = File::open(&cfg.source_path)?; let mut reader = BufReader::new(file); @@ -19,8 +26,17 @@ pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { let mut line = String::new(); - for res in notify_rx { - let event = res?; + loop { + let event = match notify_rx.recv_timeout(SHUTDOWN_POLL_INTERVAL) { + Ok(res) => res?, + Err(std_mpsc::RecvTimeoutError::Timeout) => { + if tx.is_closed() { + return Ok(()); + } + continue; + } + Err(std_mpsc::RecvTimeoutError::Disconnected) => return Ok(()), + }; if !matches!(event.kind, EventKind::Modify(_)) { continue; @@ -44,6 +60,4 @@ pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { } } } - - Ok(()) -} \ No newline at end of file +} diff --git a/tests/filter_test.rs b/tests/filter_test.rs index 748985d..19efc7e 100644 --- a/tests/filter_test.rs +++ b/tests/filter_test.rs @@ -12,7 +12,7 @@ async fn filter_drops_log_when_equals_rule_matches() { action: RuleAction::Drop, }]; - let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules, false)); input_tx .send(serde_json::json!({ @@ -43,7 +43,7 @@ async fn filter_keeps_log_when_no_rule_matches() { action: RuleAction::Drop, }]; - let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules, false)); input_tx .send(serde_json::json!({ @@ -80,7 +80,7 @@ async fn filter_masks_field_when_contains_rule_matches() { action: RuleAction::Mask, }]; - let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules, false)); input_tx .send(serde_json::json!({ @@ -112,7 +112,7 @@ async fn filter_applies_regex_rule() { action: RuleAction::Mask, }]; - let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules, false)); input_tx .send(serde_json::json!({ @@ -129,3 +129,54 @@ async fn filter_applies_regex_rule() { assert_eq!(received["message"], serde_json::json!("***")); } + +#[tokio::test] +async fn filter_masks_builtin_email_pattern_when_enabled() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let filter = tokio::spawn(run_filter(input_rx, output_tx, vec![], true)); + + input_tx + .send(serde_json::json!({ + "message": "user login: user@example.com" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await.unwrap(); + + filter.await.unwrap(); + + let message = received["message"].as_str().unwrap(); + assert!(!message.contains("user@example.com")); + assert!(message.contains("***")); +} + +#[tokio::test] +async fn filter_does_not_mask_builtin_patterns_when_disabled() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let filter = tokio::spawn(run_filter(input_rx, output_tx, vec![], false)); + + input_tx + .send(serde_json::json!({ + "message": "user login: user@example.com" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await.unwrap(); + + filter.await.unwrap(); + + assert_eq!( + received["message"], + serde_json::json!("user login: user@example.com") + ); +} diff --git a/tests/integration_test.rs b/tests/integration_test.rs index a23a7b6..cf21d46 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -75,7 +75,11 @@ async fn integration_source_parser_filter_sink_sends_log_to_http_server() { batch_size: 1, flush_interval_secs: 60, channel_capacity: 10, + max_retries: 0, + retry_backoff_initial_ms: 0, + retry_backoff_max_secs: 0, filter_rules: vec![], + mask_common_patterns: false, }; let app = tokio::spawn(async move { @@ -94,7 +98,7 @@ async fn integration_source_parser_filter_sink_sends_log_to_http_server() { file, r#"{{"level":"info","message":"integration test worked"}}"# ) - .unwrap(); + .unwrap(); } let body = tokio::time::timeout(Duration::from_secs(3), server) @@ -117,4 +121,91 @@ async fn integration_source_parser_filter_sink_sends_log_to_http_server() { } ]) ); -} \ No newline at end of file +} + +#[tokio::test] +async fn integration_loads_real_config_file_and_delivers_filtered_log() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.unwrap(); + + let body = read_http_request_body(&mut stream).await; + + stream + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n") + .await + .unwrap(); + + body + }); + + let source_path = PathBuf::from("integration_config_file.log"); + fs::write(&source_path, "").unwrap(); + + let config_path = PathBuf::from("integration_config_file.toml"); + let config_toml = format!( + r#" +source_path = "{source}" +sink_url = "http://{address}/logs" +batch_size = 1 +flush_interval_secs = 60 +channel_capacity = 10 +max_retries = 0 +retry_backoff_initial_ms = 0 +retry_backoff_max_secs = 0 +mask_common_patterns = false + +[[filter_rules]] +field = "level" +op = "equals" +value = "debug" +action = "drop" +"#, + source = source_path.display(), + address = address + ); + fs::write(&config_path, config_toml).unwrap(); + + let cfg = + logtap::Config::load(config_path.to_str().unwrap()).expect("failed to load logtap.toml"); + + let app = tokio::spawn(async move { + logtap::run(cfg).await.unwrap(); + }); + + tokio::time::sleep(Duration::from_millis(200)).await; + + { + let mut file = fs::OpenOptions::new() + .append(true) + .open(&source_path) + .unwrap(); + + writeln!(file, r#"{{"level":"debug","message":"should be dropped"}}"#).unwrap(); + writeln!(file, r#"{{"level":"info","message":"config file worked"}}"#).unwrap(); + } + + let body = tokio::time::timeout(Duration::from_secs(3), server) + .await + .expect("timeout: sink did not send request to test HTTP server") + .unwrap(); + + app.abort(); + + fs::remove_file(&source_path).ok(); + fs::remove_file(&config_path).ok(); + + let received_logs: serde_json::Value = serde_json::from_str(&body).unwrap(); + + assert_eq!( + received_logs, + serde_json::json!([ + { + "level": "info", + "message": "config file worked" + } + ]) + ); +} diff --git a/tests/notify_test.rs b/tests/notify_test.rs index 72561c5..0b4d66b 100644 --- a/tests/notify_test.rs +++ b/tests/notify_test.rs @@ -23,7 +23,7 @@ fn notify_detects_file_modification() { let mut watcher = notify::recommended_watcher(move |result| { tx.send(result).unwrap(); }) - .unwrap(); + .unwrap(); watcher.watch(&path, RecursiveMode::NonRecursive).unwrap(); @@ -42,7 +42,7 @@ fn notify_detects_file_modification() { .expect("timeout: notify did not detect file modification") .expect("notify returned error"); - received_kinds.push(event.kind.clone()); + received_kinds.push(event.kind); if matches!(event.kind, EventKind::Modify(_)) { saw_modify = true; @@ -57,4 +57,4 @@ fn notify_detects_file_modification() { "expected Modify, but received events were: {:?}", received_kinds ); -} \ No newline at end of file +} diff --git a/tests/sink_test.rs b/tests/sink_test.rs index 72dc400..25e44a2 100644 --- a/tests/sink_test.rs +++ b/tests/sink_test.rs @@ -71,7 +71,11 @@ async fn sink_posts_logs_when_batch_size_is_reached() { batch_size: 2, flush_interval_secs: 60, channel_capacity: 10, + max_retries: 3, + retry_backoff_initial_ms: 100, + retry_backoff_max_secs: 5, filter_rules: vec![], + mask_common_patterns: false, }; let (tx, rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity); diff --git a/tests/source_test.rs b/tests/source_test.rs index d1fe8a3..f4e7d60 100644 --- a/tests/source_test.rs +++ b/tests/source_test.rs @@ -21,7 +21,11 @@ async fn test_run_source_emite_linhas_novas() { batch_size: 100, flush_interval_secs: 5, channel_capacity: 100, + max_retries: 3, + retry_backoff_initial_ms: 100, + retry_backoff_max_secs: 5, filter_rules: vec![], + mask_common_patterns: false, }; let _handle = tokio::task::spawn_blocking(move || run_source(cfg, tx)); @@ -41,4 +45,4 @@ async fn test_run_source_emite_linhas_novas() { assert_eq!(received, "linha de teste"); fs::remove_file(&path).ok(); -} \ No newline at end of file +}