diff --git a/Cargo.lock b/Cargo.lock index 7fb22c4..de5035a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,15 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "aho-corasick" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba" +dependencies = [ + "memchr", +] + [[package]] name = "anstream" version = "1.0.0" @@ -414,6 +423,7 @@ version = "0.1.0" dependencies = [ "axum", "clap", + "regex", "serde", "serde_json", "tokio", @@ -482,6 +492,35 @@ dependencies = [ "bitflags", ] +[[package]] +name = "regex" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f020237b6c8eed93db2e2cb53c00c60a8e1bc73da7d073199a1180401450218d" +dependencies = [ + "aho-corasick", + "memchr", + "regex-automata", + "regex-syntax", +] + +[[package]] +name = "regex-automata" +version = "0.4.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2" +dependencies = [ + "aho-corasick", + "memchr", + "regex-syntax", +] + +[[package]] +name = "regex-syntax" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" + [[package]] name = "ryu" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index 24df485..caaf46c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,3 +10,4 @@ tower-http = { version = "0.6", features = ["fs", "cors"] } serde = { version = "1", features = ["derive"] } serde_json = "1" clap = { version = "4", features = ["derive"] } +regex = "1" diff --git a/src/data_value.rs b/src/data_value.rs index 0438035..86287b0 100644 --- a/src/data_value.rs +++ b/src/data_value.rs @@ -25,6 +25,12 @@ pub enum MetricPrimitiveKind { pub enum DataValue { /// Paired with `IndexKey::IngestedLog`. LogLiteral { text: String }, + /// Paired with `IndexKey::ParamDeclaration`. + ParamDeclaration { + message_match_regex: String, + capture_group: String, + kind: MetricPrimitiveKind, + }, /// Paired with `IndexKey::MetricDeclaration`. MetricDeclaration { message_match_regex: String, diff --git a/src/dto.rs b/src/dto.rs index 80854cd..6780771 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -1,3 +1,4 @@ +use crate::data_value::MetricPrimitiveKind; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Serialize)] @@ -7,6 +8,14 @@ pub struct LogEntryDto { pub text: String, } +#[derive(Debug, Clone, Serialize)] +pub struct ParamDeclarationDto { + pub name: String, + pub message_match_regex: String, + pub capture_group: String, + pub kind: MetricPrimitiveKind, +} + #[derive(Debug, Clone, Serialize)] pub struct TimelineBucketDto { pub start_ms: i64, diff --git a/src/engine.rs b/src/engine.rs index 64feb83..b82ce04 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1,7 +1,10 @@ -use crate::data_value::DataValue; -use crate::dto::{LogEntryDto, LogsParams, StatsDto, TimelineBucketDto, TimelineParams}; +use crate::data_value::{DataValue, MetricPrimitiveKind}; +use crate::dto::{ + LogEntryDto, LogsParams, ParamDeclarationDto, StatsDto, TimelineBucketDto, TimelineParams, +}; use crate::index_key::IndexKey; use crate::timestamp::Timestamp; +use regex::Regex; use std::collections::BTreeMap; use tokio::sync::{mpsc, oneshot}; @@ -11,24 +14,44 @@ pub struct IngestedLine { pub text: String, } +pub struct DeclareParamRequest { + pub name: String, + pub message_match_regex: String, + pub capture_group: String, + pub kind: MetricPrimitiveKind, +} + +#[allow(dead_code)] pub enum EngineQuery { Logs(LogsParams), Timeline(TimelineParams), Stats, + ParamDeclarations, } +#[allow(dead_code)] pub enum EngineMessage { Ingest(IngestedLine), + DeclareParam { + request: DeclareParamRequest, + respond_to: oneshot::Sender>, + }, + RemoveParamDeclaration { + name: String, + respond_to: oneshot::Sender, + }, Query { query: EngineQuery, respond_to: oneshot::Sender, }, } +#[allow(dead_code)] pub enum EngineResponse { Logs(Vec), Timeline(Vec), Stats(StatsDto), + ParamDeclarations(Vec), } pub type EngineHandle = mpsc::Sender; @@ -50,6 +73,14 @@ pub async fn run(mut rx: mpsc::Receiver) { DataValue::LogLiteral { text: line.text }, ); } + EngineMessage::DeclareParam { request, respond_to } => { + let result = declare_param(&mut index, request); + let _ = respond_to.send(result); + } + EngineMessage::RemoveParamDeclaration { name, respond_to } => { + let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some(); + let _ = respond_to.send(existed); + } EngineMessage::Query { query, respond_to } => { let response = handle_query(&index, query); let _ = respond_to.send(response); @@ -58,6 +89,47 @@ pub async fn run(mut rx: mpsc::Receiver) { } } +/// Validates the regex and capture group before storing, so a bad declaration is +/// rejected up front rather than silently stored (or discovered later at derivation +/// time). +fn declare_param( + index: &mut BTreeMap, + request: DeclareParamRequest, +) -> Result<(), String> { + let re = Regex::new(&request.message_match_regex) + .map_err(|err| format!("invalid regex: {err}"))?; + if !re.capture_names().flatten().any(|name| name == request.capture_group) { + return Err(format!( + "regex has no capture group named '{}'", + request.capture_group + )); + } + + index.insert( + IndexKey::ParamDeclaration { name: request.name }, + DataValue::ParamDeclaration { + message_match_regex: request.message_match_regex, + capture_group: request.capture_group, + kind: request.kind, + }, + ); + Ok(()) +} + +fn param_declarations_range( + index: &BTreeMap, +) -> impl Iterator { + index + .range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit) + .filter_map(|(key, value)| match (key, value) { + ( + IndexKey::ParamDeclaration { name }, + DataValue::ParamDeclaration { message_match_regex, capture_group, kind }, + ) => Some((name.as_str(), message_match_regex.as_str(), capture_group.as_str(), *kind)), + _ => None, + }) +} + fn logs_range(index: &BTreeMap) -> impl Iterator { index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| { match (key, value) { @@ -125,5 +197,95 @@ fn handle_query(index: &BTreeMap, query: EngineQuery) -> En last_timestamp_ms, }) } + EngineQuery::ParamDeclarations => { + let declarations = param_declarations_range(index) + .map(|(name, message_match_regex, capture_group, kind)| ParamDeclarationDto { + name: name.to_string(), + message_match_regex: message_match_regex.to_string(), + capture_group: capture_group.to_string(), + kind, + }) + .collect(); + EngineResponse::ParamDeclarations(declarations) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { + DeclareParamRequest { + name: name.to_string(), + message_match_regex: pattern.to_string(), + capture_group: capture_group.to_string(), + kind: MetricPrimitiveKind::String, + } + } + + #[test] + fn declares_valid_param() { + let mut index = BTreeMap::new(); + let result = declare_param(&mut index, request("level", r"level=(?P\w+)", "level")); + assert!(result.is_ok()); + + let declarations: Vec<_> = param_declarations_range(&index).collect(); + assert_eq!(declarations.len(), 1); + assert_eq!(declarations[0].0, "level"); + assert_eq!(declarations[0].1, r"level=(?P\w+)"); + assert_eq!(declarations[0].2, "level"); + } + + #[test] + fn rejects_invalid_regex() { + let mut index = BTreeMap::new(); + let result = declare_param(&mut index, request("level", "level=(", "level")); + assert!(result.is_err()); + assert!(index.is_empty()); + } + + #[test] + fn rejects_unknown_capture_group() { + let mut index = BTreeMap::new(); + let result = declare_param( + &mut index, + request("level", r"level=(?P\w+)", "missing"), + ); + assert!(result.is_err()); + assert!(index.is_empty()); + } + + #[test] + fn declarations_stay_within_their_range() { + let mut index = BTreeMap::new(); + index.insert( + IndexKey::IngestedLog { sequence: 0, timestamp: Timestamp::from_millis(0) }, + DataValue::LogLiteral { text: "hello".to_string() }, + ); + declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); + index.insert( + IndexKey::MetricDeclaration { name: "requests".to_string() }, + DataValue::MetricDeclaration { + message_match_regex: "x".to_string(), + params: vec![], + }, + ); + + let declarations: Vec<_> = param_declarations_range(&index).collect(); + assert_eq!(declarations.len(), 1); + assert_eq!(declarations[0].0, "level"); + + for key in index.range(IndexKey::LogParamSplit..IndexKey::ParamMetricSplit).map(|(k, _)| k) { + assert!(matches!(key, IndexKey::ParamDeclaration { .. })); + } + } + + #[test] + fn removes_declaration() { + let mut index = BTreeMap::new(); + declare_param(&mut index, request("level", r"level=(?P\w+)", "level")).unwrap(); + assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some()); + assert!(param_declarations_range(&index).next().is_none()); } }