feat: add sink backoff

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"
+anyhow = "1"
+toml = "1.1.3+spec-1.1.0"

src/config.rs

@@ -1,19 +1,65 @@
 use crate::filter::FilterRule;
+use anyhow::Result;
 use serde::Deserialize;
 use std::path::PathBuf;
 
 #[derive(Debug, Deserialize, Clone)]
 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<FilterRule>,
+
+    #[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<FilterRule> {
-    vec![]
+impl Config {
+    pub fn load(path: &str) -> Result<Self> {
+        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)
+    }
 }

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,24 +19,46 @@ pub enum RuleAction {
 }
 
 #[derive(Debug, Deserialize, Clone)]
-#[serde(rename_all = "lowercase")]
 pub struct FilterRule {
     pub field: String,
     pub op: RuleOp,
     pub value: String,
     pub action: RuleAction,
 }
 
-pub async fn run_filter(mut rx: Receiver<LogLine>, tx: Sender<LogLine>, rules: Vec<FilterRule>) {
+pub async fn run_filter(
+    mut rx: Receiver<LogLine>,
+    tx: Sender<LogLine>,
+    rules: Vec<FilterRule>,
+    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<LogLine>, tx: Sender<LogLine>, 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);
+                }
+            }
+        }
+    }
+}

src/lib.rs

@@ -14,13 +14,18 @@ pub async fn run(cfg: Config) -> anyhow::Result<()> {
     let (clean_tx, clean_rx) = tokio::sync::mpsc::channel::<LogLine>(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(())
-}
+}

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<LogLine>) {
-    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<LogLine>) {
     let client = reqwest::Client::new();
     let mut batch: Vec<LogLine> = Vec::with_capacity(cfg.batch_size);
@@ -33,14 +14,70 @@ pub async fn run_sink(cfg: Config, mut rx: Receiver<LogLine>) {
             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<LogLine>) {
+    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;
+    }
+}