From 0a7e38f73989bdc2cb10d5a4a26b29d442b043e1 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Tue, 14 Jul 2026 21:48:40 -0300 Subject: [PATCH 1/4] feat: create sink race multiple futures function with batch size and flush time --- src/lib.rs | 3 +- src/sink.rs | 41 ++++++++++++++++++ tests/sink_test.rs | 106 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 149 insertions(+), 1 deletion(-) create mode 100644 src/sink.rs create mode 100644 tests/sink_test.rs diff --git a/src/lib.rs b/src/lib.rs index 7531c95..28dca05 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 sink; diff --git a/src/sink.rs b/src/sink.rs new file mode 100644 index 0000000..41e781d --- /dev/null +++ b/src/sink.rs @@ -0,0 +1,41 @@ +use tokio::sync::mpsc::Receiver; +use std::time::Duration; +use tokio::time::interval; +use crate::config::Config; +use crate::record::LogLine; + + +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); + let mut ticker = interval(Duration::from_secs(cfg.flush_interval_secs)); + + loop { + tokio::select! { + Some(log) = rx.recv() => { + batch.push(log); + if batch.len() >= cfg.batch_size { + flush(&client, &cfg.sink_url, &mut batch).await; + } + } + _ = ticker.tick() => { + if !batch.is_empty() { + flush(&client, &cfg.sink_url, &mut batch).await; + } + } + } + } +} \ No newline at end of file diff --git a/tests/sink_test.rs b/tests/sink_test.rs new file mode 100644 index 0000000..6d23ef4 --- /dev/null +++ b/tests/sink_test.rs @@ -0,0 +1,106 @@ +use std::path::PathBuf; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tokio::net::TcpListener; + +use logtap::config::Config; +use logtap::record::LogLine; +use logtap::sink::run_sink; + +async fn read_http_request_body(stream: &mut tokio::net::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 sink_posts_logs_when_batch_size_is_reached() { + 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 cfg = Config { + source_path: PathBuf::from("unused.log"), + sink_url: format!("http://{address}/logs"), + batch_size: 2, + flush_interval_secs: 60, + channel_capacity: 10, + }; + + let (tx, rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity); + + let sink = tokio::spawn(run_sink(cfg, rx)); + + tx.send(serde_json::json!({ "message": "first log" })) + .await + .unwrap(); + + tx.send(serde_json::json!({ "message": "second log" })) + .await + .unwrap(); + + let body = tokio::time::timeout(std::time::Duration::from_secs(2), server) + .await + .unwrap() + .unwrap(); + + sink.abort(); + + let received_logs: serde_json::Value = serde_json::from_str(&body).unwrap(); + + assert_eq!( + received_logs, + serde_json::json!([ + { "message": "first log" }, + { "message": "second log" } + ]) + ); +} \ No newline at end of file From b09cfbd14d40f6129bbb5c3918e4065f4658a0c5 Mon Sep 17 00:00:00 2001 From: dweg0 Date: Tue, 14 Jul 2026 21:49:51 -0300 Subject: [PATCH 2/4] ci: add CI to run tests on pull requests --- .github/workflows/ci.yml | 31 +++++++++++++++++++++++++++++++ 1 file changed, 31 insertions(+) create mode 100644 .github/workflows/ci.yml diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..968dcbc --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,31 @@ +name: CI + +on: + pull_request: + push: + branches: + - main + +jobs: + test: + name: Run Rust tests + runs-on: ubuntu-latest + + steps: + - name: Checkout repository + uses: actions/checkout@v4 + + - name: Install Rust toolchain + uses: dtolnay/rust-toolchain@stable + + - name: Cache Cargo dependencies + uses: Swatinem/rust-cache@v2 + + - name: Check formatting + run: cargo fmt --all -- --check + + - name: Run Clippy + run: cargo clippy --all-targets --all-features -- -D warnings + + - name: Run tests + run: cargo test --all \ No newline at end of file From 42d14433878ddb2dc4b7c0eb0daae9bde0455b7c Mon Sep 17 00:00:00 2001 From: dweg0 Date: Tue, 14 Jul 2026 21:52:30 -0300 Subject: [PATCH 3/4] fix: format rust code --- src/config.rs | 2 +- src/lib.rs | 2 +- src/record.rs | 2 +- src/sink.rs | 19 ++++++++++++------- tests/sink_test.rs | 6 ++---- 5 files changed, 17 insertions(+), 14 deletions(-) diff --git a/src/config.rs b/src/config.rs index e5f7129..103c1a7 100644 --- a/src/config.rs +++ b/src/config.rs @@ -6,4 +6,4 @@ pub struct Config { pub batch_size: usize, pub flush_interval_secs: u64, pub channel_capacity: usize, -} \ No newline at end of file +} diff --git a/src/lib.rs b/src/lib.rs index 28dca05..15e668f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,3 @@ -pub mod record; pub mod config; +pub mod record; pub mod sink; 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; diff --git a/src/sink.rs b/src/sink.rs index 41e781d..bde31b6 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -1,9 +1,8 @@ -use tokio::sync::mpsc::Receiver; -use std::time::Duration; -use tokio::time::interval; use crate::config::Config; use crate::record::LogLine; - +use std::time::Duration; +use tokio::sync::mpsc::Receiver; +use tokio::time::interval; async fn flush(client: &reqwest::Client, url: &str, batch: &mut Vec) { if batch.is_empty() { @@ -12,8 +11,14 @@ async fn flush(client: &reqwest::Client, url: &str, batch: &mut Vec) { 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()), + Ok(resp) => eprintln!( + "logtap: destination responded with status {}", + resp.status() + ), + Err(err) => eprintln!( + "logtap: failed to send batch ({} items): {err}", + batch.len() + ), } batch.clear(); @@ -38,4 +43,4 @@ pub async fn run_sink(cfg: Config, mut rx: Receiver) { } } } -} \ No newline at end of file +} diff --git a/tests/sink_test.rs b/tests/sink_test.rs index 6d23ef4..028cbba 100644 --- a/tests/sink_test.rs +++ b/tests/sink_test.rs @@ -58,9 +58,7 @@ async fn sink_posts_logs_when_batch_size_is_reached() { 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", - ) + .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\nconnection: close\r\n\r\n") .await .unwrap(); @@ -103,4 +101,4 @@ async fn sink_posts_logs_when_batch_size_is_reached() { { "message": "second log" } ]) ); -} \ No newline at end of file +} From fa4fa9e343c4d8db42744408d134fe9bb3eb67ba Mon Sep 17 00:00:00 2001 From: dweg0 Date: Tue, 14 Jul 2026 22:10:44 -0300 Subject: [PATCH 4/4] feat: create configurable filter to act before sink --- src/config.rs | 10 ++++ src/filter.rs | 56 ++++++++++++++++++ src/lib.rs | 1 + tests/filter_test.rs | 131 +++++++++++++++++++++++++++++++++++++++++++ tests/sink_test.rs | 1 + 5 files changed, 199 insertions(+) create mode 100644 src/filter.rs create mode 100644 tests/filter_test.rs diff --git a/src/config.rs b/src/config.rs index 103c1a7..e4a6dc4 100644 --- a/src/config.rs +++ b/src/config.rs @@ -1,9 +1,19 @@ +use crate::filter::FilterRule; +use serde::Deserialize; use std::path::PathBuf; +#[derive(Debug, Deserialize, Clone)] pub struct Config { pub source_path: PathBuf, pub sink_url: String, pub batch_size: usize, pub flush_interval_secs: u64, pub channel_capacity: usize, + + #[serde(default = "filter_rules_default")] + pub filter_rules: Vec, +} + +fn filter_rules_default() -> Vec { + vec![] } diff --git a/src/filter.rs b/src/filter.rs new file mode 100644 index 0000000..0ee86a0 --- /dev/null +++ b/src/filter.rs @@ -0,0 +1,56 @@ +use crate::record::LogLine; +use regex::Regex; +use serde::Deserialize; +use tokio::sync::mpsc::{Receiver, Sender}; + +#[derive(Debug, Deserialize, Clone)] +pub enum RuleOp { + Equals, + Contains, + Regex, +} + +#[derive(Debug, Deserialize, Clone)] +pub enum RuleAction { + Drop, + Mask, +} + +#[derive(Debug, Deserialize, Clone)] +pub struct FilterRule { + pub field: String, + pub op: RuleOp, + pub value: String, + pub action: RuleAction, +} + +pub async fn run_filter(mut rx: Receiver, tx: Sender, rules: Vec) { + while let Some(mut log) = rx.recv().await { + 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), + }; + + if matched { + match rule.action { + RuleAction::Drop => { + dropped = true; + break; + } + RuleAction::Mask => { + log[&rule.field] = serde_json::json!("***"); + } + } + } + } + + if !dropped && tx.send(log).await.is_err() { + break; + } + } +} diff --git a/src/lib.rs b/src/lib.rs index 15e668f..ebadaed 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,4 @@ pub mod config; +pub mod filter; pub mod record; pub mod sink; diff --git a/tests/filter_test.rs b/tests/filter_test.rs new file mode 100644 index 0000000..748985d --- /dev/null +++ b/tests/filter_test.rs @@ -0,0 +1,131 @@ +use logtap::filter::{FilterRule, RuleAction, RuleOp, run_filter}; + +#[tokio::test] +async fn filter_drops_log_when_equals_rule_matches() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let rules = vec![FilterRule { + field: "level".to_string(), + op: RuleOp::Equals, + value: "debug".to_string(), + action: RuleAction::Drop, + }]; + + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + + input_tx + .send(serde_json::json!({ + "level": "debug", + "message": "this should be dropped" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await; + + filter.await.unwrap(); + + assert!(received.is_none()); +} + +#[tokio::test] +async fn filter_keeps_log_when_no_rule_matches() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let rules = vec![FilterRule { + field: "level".to_string(), + op: RuleOp::Equals, + value: "debug".to_string(), + action: RuleAction::Drop, + }]; + + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + + input_tx + .send(serde_json::json!({ + "level": "info", + "message": "this should pass" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await.unwrap(); + + filter.await.unwrap(); + + assert_eq!( + received, + serde_json::json!({ + "level": "info", + "message": "this should pass" + }) + ); +} + +#[tokio::test] +async fn filter_masks_field_when_contains_rule_matches() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let rules = vec![FilterRule { + field: "email".to_string(), + op: RuleOp::Contains, + value: "@".to_string(), + action: RuleAction::Mask, + }]; + + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + + input_tx + .send(serde_json::json!({ + "email": "user@example.com", + "message": "user logged in" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await.unwrap(); + + filter.await.unwrap(); + + assert_eq!(received["email"], serde_json::json!("***")); + assert_eq!(received["message"], serde_json::json!("user logged in")); +} + +#[tokio::test] +async fn filter_applies_regex_rule() { + let (input_tx, input_rx) = tokio::sync::mpsc::channel(10); + let (output_tx, mut output_rx) = tokio::sync::mpsc::channel(10); + + let rules = vec![FilterRule { + field: "message".to_string(), + op: RuleOp::Regex, + value: r"\d{3}-\d{2}-\d{4}".to_string(), + action: RuleAction::Mask, + }]; + + let filter = tokio::spawn(run_filter(input_rx, output_tx, rules)); + + input_tx + .send(serde_json::json!({ + "message": "ssn 123-45-6789" + })) + .await + .unwrap(); + + drop(input_tx); + + let received = output_rx.recv().await.unwrap(); + + filter.await.unwrap(); + + assert_eq!(received["message"], serde_json::json!("***")); +} diff --git a/tests/sink_test.rs b/tests/sink_test.rs index 028cbba..72dc400 100644 --- a/tests/sink_test.rs +++ b/tests/sink_test.rs @@ -71,6 +71,7 @@ async fn sink_posts_logs_when_batch_size_is_reached() { batch_size: 2, flush_interval_secs: 60, channel_capacity: 10, + filter_rules: vec![], }; let (tx, rx) = tokio::sync::mpsc::channel::(cfg.channel_capacity);