Feat/filter sink

.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

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<FilterRule>,
+}
+
+fn filter_rules_default() -> Vec<FilterRule> {
+    vec![]
+}

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<LogLine>, tx: Sender<LogLine>, rules: Vec<FilterRule>) {
+    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;
+        }
+    }
+}

src/lib.rs

@@ -1,2 +1,4 @@
+pub mod config;
+pub mod filter;
 pub mod record;
-pub mod config;
+pub mod sink;

src/record.rs

@@ -1 +1 @@
-pub type LogLine = serde_json::Value;
+pub type LogLine = serde_json::Value;

src/sink.rs

@@ -0,0 +1,46 @@
+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<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);
+    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;
+                }
+            }
+        }
+    }
+}

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!("***"));
+}