From 34b99c9282bd84c686f0c29346c2817de21feb4c Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Tue, 14 Jul 2026 22:21:42 -0300 Subject: [PATCH 01/11] feat: implement source tail --- logtap.toml | 0 src/lib.rs | 3 ++- src/source.rs | 34 ++++++++++++++++++++++++++++++++++ tests/test_source.rs | 26 ++++++++++++++++++++++++++ 4 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 logtap.toml create mode 100644 src/source.rs create mode 100644 tests/test_source.rs diff --git a/logtap.toml b/logtap.toml new file mode 100644 index 0000000..e69de29 diff --git a/src/lib.rs b/src/lib.rs index 7531c95..5189d9b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,2 +1,3 @@ pub mod record; -pub mod config; \ No newline at end of file +pub mod config; +pub mod source; \ No newline at end of file diff --git a/src/source.rs b/src/source.rs new file mode 100644 index 0000000..1aa6918 --- /dev/null +++ b/src/source.rs @@ -0,0 +1,34 @@ +use std::fs::File; +use tokio::time::{sleep, Duration}; +use std::io::{BufRead, BufReader, Seek, SeekFrom}; + +pub async fn run_source(cfg: crate::config::Config, tx: tokio::sync::mpsc::Sender) -> anyhow::Result<()>{ + let mut offset: u64 = 0; + + loop { + let file = File::open(&cfg.source_path)?; + let mut reader = BufReader::new(file); + + reader.seek(SeekFrom::Start(offset))?; + + let mut line = String::new(); + 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.send(trimmed).await.is_err() { + return Ok(()); + } + } + + sleep(Duration::from_millis(500)).await; + + } +} \ No newline at end of file diff --git a/tests/test_source.rs b/tests/test_source.rs new file mode 100644 index 0000000..d191d6d --- /dev/null +++ b/tests/test_source.rs @@ -0,0 +1,26 @@ +use std::path::PathBuf; + +use logtap::config::Config; +use logtap::source::run_source; +use tokio::sync::mpsc; + +#[tokio::test] +async fn test_run_source() { + let (tx, mut rx) = mpsc::channel::(100); + + let cfg = Config { + source_path: PathBuf::from("teste.log"), + sink_url: "http://localhost:8080".to_string(), + batch_size: 100, + flush_interval_secs: 5, + channel_capacity: 100, + }; + + tokio::spawn(async move { + run_source(cfg, tx).await.unwrap(); + }); + + while let Some(line) = rx.recv().await { + println!("{line}"); + } +} From 61c51dd95d5b9f111f90bf931ea266e72dc98c05 Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Tue, 14 Jul 2026 23:48:34 -0300 Subject: [PATCH 02/11] feat: implement parser --- src/lib.rs | 3 +- src/parser.rs | 32 ++++++++++++++++ src/record.rs | 8 +++- src/source.rs | 1 - tests/parser_test.rs | 49 ++++++++++++++++++++++++ tests/{test_source.rs => source_test.rs} | 0 6 files changed, 90 insertions(+), 3 deletions(-) create mode 100644 src/parser.rs create mode 100644 tests/parser_test.rs rename tests/{test_source.rs => source_test.rs} (100%) diff --git a/src/lib.rs b/src/lib.rs index 5189d9b..ec93ae2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ pub mod record; pub mod config; -pub mod source; \ No newline at end of file +pub mod source; +pub mod parser; \ No newline at end of file diff --git a/src/parser.rs b/src/parser.rs new file mode 100644 index 0000000..28792ff --- /dev/null +++ b/src/parser.rs @@ -0,0 +1,32 @@ +use std::collections::HashMap; +use serde_json::Value; +use crate::record::LogLine; +use tokio::sync::mpsc::{Receiver, Sender}; + +pub async fn run_parser(mut rx: Receiver, tx: Sender) { + while let Some(line) = rx.recv().await { + let log_line = parse_line(&line); + if tx.send(log_line).await.is_err() { + break; + } + } +} + +pub fn parse_line(line: &str) -> LogLine { + match serde_json::from_str::>(line) { + Ok(fields) => LogLine{ + raw: line.to_string(), + fields, + }, + Err(_) => { + let mut fields = HashMap::new(); + fields.insert("message".to_string(), Value::String(line.to_string())); + fields.insert("level".to_string(), Value::String("UNKNOWN".to_string())); + + LogLine { + raw: line.to_string(), + fields, + } + } + } +} diff --git a/src/record.rs b/src/record.rs index a753002..66ce774 100644 --- a/src/record.rs +++ b/src/record.rs @@ -1 +1,7 @@ -pub type LogLine = serde_json::Value; \ No newline at end of file +use serde_json::Value; +use std::collections::HashMap; + +pub struct LogLine { + pub raw: String, + pub fields: HashMap, +} \ No newline at end of file diff --git a/src/source.rs b/src/source.rs index 1aa6918..da561a6 100644 --- a/src/source.rs +++ b/src/source.rs @@ -29,6 +29,5 @@ pub async fn run_source(cfg: crate::config::Config, tx: tokio::sync::mpsc::Sende } sleep(Duration::from_millis(500)).await; - } } \ No newline at end of file diff --git a/tests/parser_test.rs b/tests/parser_test.rs new file mode 100644 index 0000000..333235d --- /dev/null +++ b/tests/parser_test.rs @@ -0,0 +1,49 @@ +use logtap::parser::parse_line; +use serde_json::Value; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_run_parse() { + let linha = r#"{"level": "info", "msg": "server started"}"#; + + let resultado = parse_line(linha); + + assert_eq!( + resultado.fields.get("level"), + Some(&Value::String("info".to_string())) + ); + assert_eq!( + resultado.fields.get("msg"), + Some(&Value::String("server started".to_string())) + ); + assert_eq!(resultado.raw, linha); + } + + #[test] + fn parseline_with_text() { + let linha = "isso aqui nao e json"; + + let resultado = parse_line(linha); + + assert_eq!( + resultado.fields.get("message"), + Some(&Value::String(linha.to_string())) + ); + assert_eq!( + resultado.fields.get("level"), + Some(&Value::String("UNKNOWN".to_string())) + ); + } + + #[test] + fn parse_empty_json_line_returns_empty_object() { + let linha = "{}"; + + let resultado = parse_line(linha); + + assert!(resultado.fields.is_empty()); + } +} \ No newline at end of file diff --git a/tests/test_source.rs b/tests/source_test.rs similarity index 100% rename from tests/test_source.rs rename to tests/source_test.rs From 0b1fa6d500e2b69db39c24d2762eb667db5b3866 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:10:19 -0300 Subject: [PATCH 03/11] fix: use the correct type for logline --- src/parser.rs | 31 ++++++++++++++----------------- src/record.rs | 7 +------ tests/parser_test.rs | 29 ++++++++++++++++------------- 3 files changed, 31 insertions(+), 36 deletions(-) diff --git a/src/parser.rs b/src/parser.rs index 28792ff..abdcb1b 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -1,6 +1,5 @@ -use std::collections::HashMap; -use serde_json::Value; use crate::record::LogLine; +use serde_json::json; use tokio::sync::mpsc::{Receiver, Sender}; pub async fn run_parser(mut rx: Receiver, tx: Sender) { @@ -12,21 +11,19 @@ pub async fn run_parser(mut rx: Receiver, tx: Sender) { } } + pub fn parse_line(line: &str) -> LogLine { - match serde_json::from_str::>(line) { - Ok(fields) => LogLine{ - raw: line.to_string(), - fields, - }, - Err(_) => { - let mut fields = HashMap::new(); - fields.insert("message".to_string(), Value::String(line.to_string())); - fields.insert("level".to_string(), Value::String("UNKNOWN".to_string())); - - LogLine { - raw: line.to_string(), - fields, - } + match serde_json::from_str::(line) { + Ok(value) if value.is_object() => value, + + Ok(value) => { + eprintln!("logtap: line was valid JSON, but is not object: {value}"); + json!({ "message": line, "level": "UNKNOWN", "parse_issue": "not_an_object" }) + } + + Err(err) => { + eprintln!("logtap: falied to process as JSON: {err}"); + json!({ "message": line, "level": "UNKNOWN", "parse_issue": "invalid_json" }) } } -} +} \ No newline at end of file diff --git a/src/record.rs b/src/record.rs index 66ce774..d4a58df 100644 --- a/src/record.rs +++ b/src/record.rs @@ -1,7 +1,2 @@ -use serde_json::Value; -use std::collections::HashMap; -pub struct LogLine { - pub raw: String, - pub fields: HashMap, -} \ No newline at end of file +pub type LogLine = serde_json::Value; \ No newline at end of file diff --git a/tests/parser_test.rs b/tests/parser_test.rs index 333235d..faf94c3 100644 --- a/tests/parser_test.rs +++ b/tests/parser_test.rs @@ -7,43 +7,46 @@ mod tests { #[test] fn test_run_parse() { - let linha = r#"{"level": "info", "msg": "server started"}"#; + let line = r#"{"level": "info", "msg": "server started"}"#; - let resultado = parse_line(linha); + let result = parse_line(line); assert_eq!( - resultado.fields.get("level"), + result.get("level"), Some(&Value::String("info".to_string())) ); assert_eq!( - resultado.fields.get("msg"), + result.get("msg"), Some(&Value::String("server started".to_string())) ); - assert_eq!(resultado.raw, linha); } #[test] fn parseline_with_text() { - let linha = "isso aqui nao e json"; + let line = "isso aqui nao e json"; - let resultado = parse_line(linha); + let result = parse_line(line); assert_eq!( - resultado.fields.get("message"), - Some(&Value::String(linha.to_string())) + result.get("message"), + Some(&Value::String(line.to_string())) ); assert_eq!( - resultado.fields.get("level"), + result.get("level"), Some(&Value::String("UNKNOWN".to_string())) ); + assert_eq!( + result.get("parse_issue"), + Some(&Value::String("invalid_json".to_string())) + ); } #[test] fn parse_empty_json_line_returns_empty_object() { - let linha = "{}"; + let line = "{}"; - let resultado = parse_line(linha); + let result = parse_line(line); - assert!(resultado.fields.is_empty()); + assert!(result.as_object().is_some_and(|obj| obj.is_empty())); } } \ No newline at end of file From ff9f0ce637731a95d3d4799dc80ca4eb37fb409e Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:17:49 -0300 Subject: [PATCH 04/11] fix: corrige workflow --- src/parser.rs | 3 +-- src/source.rs | 11 +++++++---- tests/parser_test.rs | 2 +- tests/source_test.rs | 1 + 4 files changed, 10 insertions(+), 7 deletions(-) diff --git a/src/parser.rs b/src/parser.rs index abdcb1b..18760f8 100644 --- a/src/parser.rs +++ b/src/parser.rs @@ -11,7 +11,6 @@ pub async fn run_parser(mut rx: Receiver, tx: Sender) { } } - pub fn parse_line(line: &str) -> LogLine { match serde_json::from_str::(line) { Ok(value) if value.is_object() => value, @@ -26,4 +25,4 @@ pub fn parse_line(line: &str) -> LogLine { json!({ "message": line, "level": "UNKNOWN", "parse_issue": "invalid_json" }) } } -} \ No newline at end of file +} diff --git a/src/source.rs b/src/source.rs index da561a6..7f2ce82 100644 --- a/src/source.rs +++ b/src/source.rs @@ -1,8 +1,11 @@ use std::fs::File; -use tokio::time::{sleep, Duration}; use std::io::{BufRead, BufReader, Seek, SeekFrom}; +use tokio::time::{Duration, sleep}; -pub async fn run_source(cfg: crate::config::Config, tx: tokio::sync::mpsc::Sender) -> anyhow::Result<()>{ +pub async fn run_source( + cfg: crate::config::Config, + tx: tokio::sync::mpsc::Sender, +) -> anyhow::Result<()> { let mut offset: u64 = 0; loop { @@ -24,10 +27,10 @@ pub async fn run_source(cfg: crate::config::Config, tx: tokio::sync::mpsc::Sende let trimmed = line.trim_end().to_string(); if tx.send(trimmed).await.is_err() { - return Ok(()); + return Ok(()); } } sleep(Duration::from_millis(500)).await; } -} \ No newline at end of file +} diff --git a/tests/parser_test.rs b/tests/parser_test.rs index faf94c3..d6cf559 100644 --- a/tests/parser_test.rs +++ b/tests/parser_test.rs @@ -49,4 +49,4 @@ mod tests { assert!(result.as_object().is_some_and(|obj| obj.is_empty())); } -} \ No newline at end of file +} diff --git a/tests/source_test.rs b/tests/source_test.rs index d191d6d..0652e1c 100644 --- a/tests/source_test.rs +++ b/tests/source_test.rs @@ -14,6 +14,7 @@ async fn test_run_source() { batch_size: 100, flush_interval_secs: 5, channel_capacity: 100, + filter_rules: vec![], }; tokio::spawn(async move { From 5371f83c6618760b9e9c133493be8b0af99ec4eb Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:19:50 -0300 Subject: [PATCH 05/11] fix: format rust code --- src/lib.rs | 4 ++-- src/record.rs | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/src/lib.rs b/src/lib.rs index 1ffd430..b216ac4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,6 +1,6 @@ pub mod config; pub mod filter; +pub mod parser; pub mod record; +pub mod sink; pub mod source; -pub mod parser; -pub mod sink; \ No newline at end of file diff --git a/src/record.rs b/src/record.rs index a753002..f4f1a32 100644 --- a/src/record.rs +++ b/src/record.rs @@ -1 +1 @@ -pub type LogLine = serde_json::Value; \ No newline at end of file +pub type LogLine = serde_json::Value; From 92e312b3f459b7acc39b3e624033c92388986eb8 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:33:21 -0300 Subject: [PATCH 06/11] feat: create run function and integration_test --- src/filter.rs | 2 + src/lib.rs | 29 +++++++++ src/source.rs | 12 ++-- tests/integration_test.rs | 120 ++++++++++++++++++++++++++++++++++++++ tests/source_test.rs | 33 ++++++++--- 5 files changed, 182 insertions(+), 14 deletions(-) create mode 100644 tests/integration_test.rs diff --git a/src/filter.rs b/src/filter.rs index 0ee86a0..ff5b779 100644 --- a/src/filter.rs +++ b/src/filter.rs @@ -11,12 +11,14 @@ pub enum RuleOp { } #[derive(Debug, Deserialize, Clone)] +#[serde(rename_all = "lowercase")] pub enum RuleAction { Drop, Mask, } #[derive(Debug, Deserialize, Clone)] +#[serde(rename_all = "lowercase")] pub struct FilterRule { pub field: String, pub op: RuleOp, diff --git a/src/lib.rs b/src/lib.rs index b216ac4..ea49d22 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -4,3 +4,32 @@ pub mod parser; pub mod record; pub mod sink; pub mod source; + +pub use config::Config; +pub use record::LogLine; + +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); + 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 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 sink_cfg = cfg.clone(); + let sink_handle = tokio::spawn(sink::run_sink(sink_cfg, clean_rx)); + + let (src_res, _, _, _) = tokio::join!(source_handle, parser_handle, filter_handle, sink_handle); + + if let Ok(Err(e)) = src_res { + eprintln!("logtap: source encerrou com erro: {e}"); + } + + Ok(()) +} \ No newline at end of file diff --git a/src/source.rs b/src/source.rs index 7f2ce82..0d34ba8 100644 --- a/src/source.rs +++ b/src/source.rs @@ -1,8 +1,9 @@ use std::fs::File; use std::io::{BufRead, BufReader, Seek, SeekFrom}; -use tokio::time::{Duration, sleep}; +use std::thread::sleep; +use std::time::Duration; -pub async fn run_source( +pub fn run_source( cfg: crate::config::Config, tx: tokio::sync::mpsc::Sender, ) -> anyhow::Result<()> { @@ -11,7 +12,6 @@ pub async fn run_source( loop { let file = File::open(&cfg.source_path)?; let mut reader = BufReader::new(file); - reader.seek(SeekFrom::Start(offset))?; let mut line = String::new(); @@ -26,11 +26,11 @@ pub async fn run_source( offset += bytes_read as u64; let trimmed = line.trim_end().to_string(); - if tx.send(trimmed).await.is_err() { + if tx.blocking_send(trimmed).is_err() { return Ok(()); } } - sleep(Duration::from_millis(500)).await; + sleep(Duration::from_millis(500)); } -} +} \ No newline at end of file diff --git a/tests/integration_test.rs b/tests/integration_test.rs new file mode 100644 index 0000000..a23a7b6 --- /dev/null +++ b/tests/integration_test.rs @@ -0,0 +1,120 @@ +use std::fs; +use std::io::Write; +use std::path::PathBuf; +use std::time::Duration; + +use logtap::config::Config; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::{TcpListener, TcpStream}; + +async fn read_http_request_body(stream: &mut TcpStream) -> String { + let mut buffer = Vec::new(); + let mut temp = [0_u8; 1024]; + + loop { + let read = stream.read(&mut temp).await.unwrap(); + assert!(read > 0, "connection closed before headers were read"); + + buffer.extend_from_slice(&temp[..read]); + + if buffer.windows(4).any(|window| window == b"\r\n\r\n") { + break; + } + } + + let headers_end = buffer + .windows(4) + .position(|window| window == b"\r\n\r\n") + .unwrap() + + 4; + + let headers = String::from_utf8_lossy(&buffer[..headers_end]); + let content_length = headers + .lines() + .find_map(|line| { + line.to_ascii_lowercase() + .strip_prefix("content-length:") + .map(|value| value.trim().parse::().unwrap()) + }) + .unwrap(); + + while buffer.len() < headers_end + content_length { + let read = stream.read(&mut temp).await.unwrap(); + assert!(read > 0, "connection closed before body was read"); + + buffer.extend_from_slice(&temp[..read]); + } + + String::from_utf8(buffer[headers_end..headers_end + content_length].to_vec()).unwrap() +} + +#[tokio::test] +async fn integration_source_parser_filter_sink_sends_log_to_http_server() { + 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_source_parser_filter_sink.log"); + fs::write(&source_path, "").unwrap(); + + let cfg = Config { + source_path: source_path.clone(), + sink_url: format!("http://{address}/logs"), + batch_size: 1, + flush_interval_secs: 60, + channel_capacity: 10, + filter_rules: vec![], + }; + + 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":"info","message":"integration test 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(); + + let received_logs: serde_json::Value = serde_json::from_str(&body).unwrap(); + + assert_eq!( + received_logs, + serde_json::json!([ + { + "level": "info", + "message": "integration test worked" + } + ]) + ); +} \ No newline at end of file diff --git a/tests/source_test.rs b/tests/source_test.rs index 0652e1c..d1fe8a3 100644 --- a/tests/source_test.rs +++ b/tests/source_test.rs @@ -1,15 +1,22 @@ +use std::fs; +use std::io::Write; use std::path::PathBuf; +use std::time::Duration; use logtap::config::Config; use logtap::source::run_source; use tokio::sync::mpsc; +use tokio::time::timeout; #[tokio::test] -async fn test_run_source() { +async fn test_run_source_emite_linhas_novas() { + let path = PathBuf::from("teste_run_source.log"); + fs::write(&path, "").unwrap(); + let (tx, mut rx) = mpsc::channel::(100); let cfg = Config { - source_path: PathBuf::from("teste.log"), + source_path: path.clone(), sink_url: "http://localhost:8080".to_string(), batch_size: 100, flush_interval_secs: 5, @@ -17,11 +24,21 @@ async fn test_run_source() { filter_rules: vec![], }; - tokio::spawn(async move { - run_source(cfg, tx).await.unwrap(); - }); + let _handle = tokio::task::spawn_blocking(move || run_source(cfg, tx)); + + tokio::time::sleep(Duration::from_millis(100)).await; - while let Some(line) = rx.recv().await { - println!("{line}"); + { + let mut file = fs::OpenOptions::new().append(true).open(&path).unwrap(); + writeln!(file, "linha de teste").unwrap(); } -} + + let received = timeout(Duration::from_secs(2), rx.recv()) + .await + .expect("timeout: nenhuma linha chegou em 2s") + .expect("canal fechou antes de receber a linha"); + + assert_eq!(received, "linha de teste"); + + fs::remove_file(&path).ok(); +} \ No newline at end of file From 61489af6e4e1f01bb786d9b92f1f86e7009a33db Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:43:52 -0300 Subject: [PATCH 07/11] docs: create readme --- LICENSE | 202 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ README.md | 156 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 357 insertions(+), 1 deletion(-) create mode 100644 LICENSE diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..fa609a0 --- /dev/null +++ b/LICENSE @@ -0,0 +1,202 @@ + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright 2026 logtap contributors + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/README.md b/README.md index f3a6924..0d27fcf 100644 --- a/README.md +++ b/README.md @@ -1 +1,155 @@ -# logtap \ No newline at end of file +# logtap + +[![License: Apache 2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](https://opensource.org/licenses/Apache-2.0) +[![CI](https://github.com/dwego/logtap/actions/workflows/ci.yml/badge.svg)](https://github.com/dwego/logtap/actions/workflows/ci.yml) + +An asynchronous, minimalist log aggregator written in Rust, built around one core idea: **real backpressure**. The pipeline should never blow up memory just because the destination got slow. + +> **Status:** early-stage / MVP. The core pipeline works end-to-end and is covered by tests, but the project is still small on purpose — see [Roadmap](#roadmap) for what's deliberately not built yet. + +## What it does + +`logtap` watches a log source (today, a file; `stdin` is planned), turns each line into a structured record, applies filter and masking rules, and delivers everything in batches to an HTTP destination — without ever letting the internal work queue grow without bound. + +Think of it as a small, readable take on tools like Fluentd or Vector: it doesn't compete on feature count, it competes on being easy to understand end to end, from the first line of code to the last. + +## Architecture + +The project is organized as a pipeline of independent stages, each running as its own async task, connected only by channels: + +``` +log file + │ + ▼ +source.rs ──String──▶ parser.rs ──LogLine──▶ filter.rs ──LogLine──▶ sink.rs +(tails the (raw text becomes (drops and (batches records + file via structured JSON) masks fields and sends them + notify/inotify) per rule) over HTTP) +``` + +Every arrow is a `tokio::sync::mpsc::channel` with a bounded capacity (`channel_capacity`, configurable). That bounded capacity is the entire design in one sentence: when a stage produces faster than the next one can consume, the channel fills up and the sending call (`send().await` / `blocking_send()`) simply waits — no unbounded queue, no runaway `Vec`, no process getting OOM-killed. + +**Why separate stages instead of one big function.** Each piece (`source`, `parser`, `filter`, `sink`) only knows about its input channel and its output channel. That means swapping the internal implementation of any stage — for example, changing how `source.rs` watches the file, or swapping the HTTP client used in `sink.rs` — doesn't require touching any other file, as long as the function signature stays the same. + +**The central type, `LogLine`.** Rather than a fixed struct with predefined fields, `LogLine` is a type alias for `serde_json::Value`. Real-world logs rarely share a single schema — every application logs different fields — and forcing a rigid struct would mean the parser silently drops or truncates whatever doesn't fit. `Value` accepts any JSON log shape without `logtap` needing to know the field set up front. + +**Filtering as configuration, not code.** What to drop and what to mask lives in `logtap.toml`, not in `if` statements scattered through the codebase. `FilterRule` entries match on a field using `Equals`, `Contains`, or `Regex`, and either `Drop` the record or `Mask` the field (replacing its value with `"***"`). This lets operators change behavior without recompiling — including masking sensitive fields such as emails or API keys — by adding a rule, rather than depending on someone remembering to write one into the source. + +## How this is meant to scale + +"Scaling" here doesn't just mean "handle more volume" — it means three distinct things, and the project is structured so each can evolve independently. + +### 1. Volume (more logs per second) + +The natural bottleneck in a staged pipeline is its slowest stage — usually the sink, since it depends on the network. The current architecture already absorbs this in two ways: + +- **Batching amortizes network cost.** Instead of one HTTP request per line, logs accumulate until `batch_size` or `flush_interval_secs` is reached, whichever comes first. The fixed cost of each request (handshake, round-trip latency) is spread across dozens or hundreds of log lines instead of paid on every single one. +- **Backpressure protects the process, not just the sink.** If inbound volume spikes (a traffic burst, an incident generating excessive logs), the bounded channels absorb the pressure without letting memory grow — the source's own read rate slows down automatically as a side effect. + +The natural next step for volume is parallelizing the sink itself: today there's a single `run_sink` task consuming from one channel. Nothing prevents running multiple sink tasks reading from the same channel (`tokio::mpsc` already supports multiple concurrent consumers), as long as the destination can handle concurrent requests. + +### 2. Surface area (more sources, more destinations, more formats) + +Today `source.rs` and `sink.rs` are single files with no trait abstraction — a deliberate choice for the MVP, to avoid paying the cost of indirection before a second real implementation exists. This is exactly where the project is expected to grow: + +``` +source/ sink/ +├── mod.rs (trait) ├── mod.rs (trait) +├── file.rs ├── http.rs +└── stdin.rs ├── stdout.rs + └── file.rs +``` + +Once there's a second source (`stdin`, alongside the file tail) or a second destination (a local audit log as backup, alongside HTTP), it's worth introducing the trait — and the refactor stays small, because the rest of the pipeline (`parser`, `filter`, the channels) doesn't change at all. The same logic applies to fanning out to multiple sinks at once — shipping the same log to HTTP and to a local audit file without duplicating filter logic. + +### 3. Reliability (the system must not silently lose data) + +This is the most important axis, and the first one that needs work. Today, if a batch fails to send, it's dropped — no retry, no persistence. That's acceptable for an MVP, but it's the first thing that needs to change before `logtap` is trusted with data that actually matters: + +- **Retry with exponential backoff** in the sink before giving up on a batch. +- **A local dead-letter file** — batches that fail repeatedly get written to disk (e.g. `logtap.failed.jsonl`) instead of lost, so they can be reprocessed later. +- **Log rotation detection** — when the watched file is renamed and recreated (standard `logrotate` behavior in production), the source needs to notice and start reading the new file, instead of continuing to hold a handle to a file that no longer exists. + +### 4. Operability (visibility into the pipeline itself) + +A log shipper nobody can observe is a blind spot inside another observability system — which is a bit ironic. The natural direction here: + +- **Internal metrics**: lines read, parsed, dropped by filter, dropped by parse error, sent successfully, failed. +- **A Prometheus-format `/metrics` endpoint**, so `logtap` can be monitored by the same tooling that watches the rest of the infrastructure. +- **Degradation signals, not just outright failure** — for example, exposing when a channel is consistently near capacity, which is the earliest symptom of a downstream stage slowing down, before it turns into actual data loss. + +## What stays fixed as the project grows + +Whichever of these directions the project tackles first, two architectural decisions stay fixed, because they're the foundation everything else depends on: + +- **Communication only through bounded channels.** No unbounded queue anywhere in the pipeline — this is the guarantee that keeps the rest of the system predictable under load. +- **Stages that know nothing about each other beyond the data type they exchange.** A new stage, a new source, a new destination — all of them fit the same shape of "receive from a channel, process, send to the next one," with no cascading changes through the rest of the code. + +That discipline, more than any specific feature, is what determines whether the project can keep growing without turning into something no one can maintain. + +## Getting started + +`logtap` is currently a library crate — there is no CLI binary yet (see [Roadmap](#roadmap)). It's driven through `logtap::run`: + +```rust +use logtap::Config; + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + let cfg = Config { + source_path: "app.log".into(), + sink_url: "http://localhost:8080/logs".to_string(), + batch_size: 100, + flush_interval_secs: 5, + channel_capacity: 1000, + filter_rules: vec![], + }; + + logtap::run(cfg).await +} +``` + +### Configuration (`logtap.toml`) + +`Config` is deserializable from TOML via `serde`: + +```toml +source_path = "app.log" +sink_url = "http://localhost:8080/logs" +batch_size = 100 +flush_interval_secs = 5 +channel_capacity = 1000 + +[[filter_rules]] +field = "level" +op = "equals" +value = "debug" +action = "drop" + +[[filter_rules]] +field = "email" +op = "regex" +value = "^[^@]+@[^@]+\\.[^@]+$" +action = "mask" +``` + +| Field | Type | Description | +|---|---|---| +| `source_path` | path | File to tail. | +| `sink_url` | string | HTTP endpoint the batches are `POST`ed to as a JSON array. | +| `batch_size` | integer | Max records buffered before a flush is triggered. | +| `flush_interval_secs` | integer | Max time between flushes, even if `batch_size` isn't reached. | +| `channel_capacity` | integer | Bound applied to every internal channel — the backpressure knob. | +| `filter_rules` | array | Ordered `FilterRule` entries (`field`, `op`, `value`, `action`); optional, defaults to empty. | + +`op` accepts `equals`, `contains`, or `regex`. `action` accepts `drop` or `mask`. + +## Development + +```bash +cargo test --all # unit + integration tests +cargo fmt --all # formatting +cargo clippy --all-targets --all-features -- -D warnings +``` + +CI runs all three on every pull request and on pushes to `main`. \ No newline at end of file From 2c29df19b4deee2f036f4b755c84bfc775230835 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 00:50:33 -0300 Subject: [PATCH 08/11] docs: add roadmap at the bottom of the readme --- README.md | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 0d27fcf..1ad3085 100644 --- a/README.md +++ b/README.md @@ -152,4 +152,23 @@ cargo fmt --all # formatting cargo clippy --all-targets --all-features -- -D warnings ``` -CI runs all three on every pull request and on pushes to `main`. \ No newline at end of file +CI runs all three on every pull request and on pushes to `main`. + +## Roadmap + +### v1 acceptance criteria + +v1 is done when all of the following hold: + +- [ ] No batch is ever lost silently — every failed send is retried or persisted locally. +- [ ] The process survives log rotation without manual intervention. +- [ ] No channel grows unbounded under any tested load condition. +- [ ] Every config field is validated with a clear error message at startup — never fails silently at runtime. +- [ ] Sensitive data is masked by default, even with no user-configured rules. +- [ ] External visibility (metrics) into what the pipeline is doing, without reading logtap's own stderr output. +- [ ] End-to-end integration test covering the full pipeline, plus unit tests per stage. +- [ ] Installing and running the project requires no source reading — README and example config are enough. + +## License + +Licensed under the [Apache License, Version 2.0](LICENSE). From 5ede5626c369cae9123be8e1cf8913945c8eb44d Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 01:02:03 -0300 Subject: [PATCH 09/11] feat: add notify instead of 500 ms polling --- src/source.rs | 43 ++++++++++++++++++++----------- tests/notify_test.rs | 60 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 15 deletions(-) create mode 100644 tests/notify_test.rs diff --git a/src/source.rs b/src/source.rs index 0d34ba8..eafc6b0 100644 --- a/src/source.rs +++ b/src/source.rs @@ -1,20 +1,33 @@ +use crate::config::Config; +use anyhow::Result; +use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use std::fs::File; use std::io::{BufRead, BufReader, Seek, SeekFrom}; -use std::thread::sleep; -use std::time::Duration; - -pub fn run_source( - cfg: crate::config::Config, - tx: tokio::sync::mpsc::Sender, -) -> anyhow::Result<()> { - let mut offset: u64 = 0; - - loop { - let file = File::open(&cfg.source_path)?; - let mut reader = BufReader::new(file); +use std::sync::mpsc as std_mpsc; +use tokio::sync::mpsc::Sender; + +pub fn run_source(cfg: Config, tx: Sender) -> Result<()> { + let file = File::open(&cfg.source_path)?; + 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 line = String::new(); + + for res in notify_rx { + let event = res?; + + if !matches!(event.kind, EventKind::Modify(_)) { + continue; + } + reader.seek(SeekFrom::Start(offset))?; - let mut line = String::new(); loop { line.clear(); let bytes_read = reader.read_line(&mut line)?; @@ -30,7 +43,7 @@ pub fn run_source( return Ok(()); } } - - sleep(Duration::from_millis(500)); } + + Ok(()) } \ No newline at end of file diff --git a/tests/notify_test.rs b/tests/notify_test.rs new file mode 100644 index 0000000..72561c5 --- /dev/null +++ b/tests/notify_test.rs @@ -0,0 +1,60 @@ +use std::fs; +use std::path::PathBuf; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +use notify::{EventKind, RecursiveMode, Watcher}; + +#[test] +fn notify_detects_file_modification() { + let file_name = format!( + "notify_test_{}.log", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + ); + + let path = PathBuf::from(file_name); + + fs::write(&path, "").unwrap(); + + let (tx, rx) = std::sync::mpsc::channel(); + + let mut watcher = notify::recommended_watcher(move |result| { + tx.send(result).unwrap(); + }) + .unwrap(); + + watcher.watch(&path, RecursiveMode::NonRecursive).unwrap(); + + fs::write(&path, "changed line\n").unwrap(); + + let timeout = Duration::from_secs(3); + let deadline = Instant::now() + timeout; + let mut received_kinds = Vec::new(); + let mut saw_modify = false; + + while Instant::now() < deadline { + let remaining = deadline.saturating_duration_since(Instant::now()); + + let event = rx + .recv_timeout(remaining) + .expect("timeout: notify did not detect file modification") + .expect("notify returned error"); + + received_kinds.push(event.kind.clone()); + + if matches!(event.kind, EventKind::Modify(_)) { + saw_modify = true; + break; + } + } + + fs::remove_file(&path).ok(); + + assert!( + saw_modify, + "expected Modify, but received events were: {:?}", + received_kinds + ); +} \ No newline at end of file From 740b3b86673da19969cbc7707bd59819b87f6a6b Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 01:10:48 -0300 Subject: [PATCH 10/11] ci: replace check formating with format code --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 968dcbc..a66ff5c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,8 +21,8 @@ jobs: - name: Cache Cargo dependencies uses: Swatinem/rust-cache@v2 - - name: Check formatting - run: cargo fmt --all -- --check + - name: Format code + run: cargo fmt --all - name: Run Clippy run: cargo clippy --all-targets --all-features -- -D warnings From 0f12fa135b5611c68ecca3a970a65157fe673021 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Wed, 15 Jul 2026 02:35:21 -0300 Subject: [PATCH 11/11] feat: add sink backoff --- Cargo.lock | 55 +++++++++++++++++++++++ Cargo.toml | 3 +- src/config.rs | 52 +++++++++++++++++++-- src/filter.rs | 47 +++++++++++++++++-- src/lib.rs | 13 ++++-- src/sink.rs | 81 ++++++++++++++++++++++++--------- src/source.rs | 24 +++++++--- tests/filter_test.rs | 59 ++++++++++++++++++++++-- tests/integration_test.rs | 95 ++++++++++++++++++++++++++++++++++++++- tests/notify_test.rs | 6 +-- tests/sink_test.rs | 4 ++ tests/source_test.rs | 6 ++- 12 files changed, 396 insertions(+), 49 deletions(-) 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 +}