From 34b99c9282bd84c686f0c29346c2817de21feb4c Mon Sep 17 00:00:00 2001 From: menezes0067 Date: Tue, 14 Jul 2026 22:21:42 -0300 Subject: [PATCH 1/5] 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 2/5] 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 3/5] 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 4/5] 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 5/5] 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;