diff --git a/src/dto.rs b/src/dto.rs index 6780771..2a8b6ec 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -1,4 +1,4 @@ -use crate::data_value::MetricPrimitiveKind; +use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher}; use serde::{Deserialize, Serialize}; #[derive(Debug, Clone, Serialize)] @@ -16,6 +16,13 @@ pub struct ParamDeclarationDto { pub kind: MetricPrimitiveKind, } +#[derive(Debug, Clone, Serialize)] +pub struct MetricDeclarationDto { + pub name: String, + pub message_match_regex: String, + pub params: Vec, +} + #[derive(Debug, Clone, Serialize)] pub struct TimelineBucketDto { pub start_ms: i64, diff --git a/src/engine.rs b/src/engine.rs index b82ce04..10cbe55 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1,6 +1,7 @@ -use crate::data_value::{DataValue, MetricPrimitiveKind}; +use crate::data_value::{DataValue, MetricPrimitiveKind, RegexToMetricPrimitiveMatcher}; use crate::dto::{ - LogEntryDto, LogsParams, ParamDeclarationDto, StatsDto, TimelineBucketDto, TimelineParams, + LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, StatsDto, + TimelineBucketDto, TimelineParams, }; use crate::index_key::IndexKey; use crate::timestamp::Timestamp; @@ -21,12 +22,24 @@ pub struct DeclareParamRequest { pub kind: MetricPrimitiveKind, } +/// v1 mechanism: a metric declaration's params are derived purely from inline named +/// capture groups on the metric's own `message_match_regex` (each +/// `RegexToMetricPrimitiveMatcher::capture_group` must name a capture group on that +/// regex). Referencing an already-declared `ParamDeclaration` (001) by name is not +/// supported yet — that's left for a future ticket. +pub struct DeclareMetricRequest { + pub name: String, + pub message_match_regex: String, + pub params: Vec, +} + #[allow(dead_code)] pub enum EngineQuery { Logs(LogsParams), Timeline(TimelineParams), Stats, ParamDeclarations, + MetricDeclarations, } #[allow(dead_code)] @@ -40,6 +53,14 @@ pub enum EngineMessage { name: String, respond_to: oneshot::Sender, }, + DeclareMetric { + request: DeclareMetricRequest, + respond_to: oneshot::Sender>, + }, + RemoveMetricDeclaration { + name: String, + respond_to: oneshot::Sender, + }, Query { query: EngineQuery, respond_to: oneshot::Sender, @@ -52,6 +73,7 @@ pub enum EngineResponse { Timeline(Vec), Stats(StatsDto), ParamDeclarations(Vec), + MetricDeclarations(Vec), } pub type EngineHandle = mpsc::Sender; @@ -81,6 +103,14 @@ pub async fn run(mut rx: mpsc::Receiver) { let existed = index.remove(&IndexKey::ParamDeclaration { name }).is_some(); let _ = respond_to.send(existed); } + EngineMessage::DeclareMetric { request, respond_to } => { + let result = declare_metric(&mut index, request); + let _ = respond_to.send(result); + } + EngineMessage::RemoveMetricDeclaration { name, respond_to } => { + let existed = index.remove(&IndexKey::MetricDeclaration { name }).is_some(); + let _ = respond_to.send(existed); + } EngineMessage::Query { query, respond_to } => { let response = handle_query(&index, query); let _ = respond_to.send(response); @@ -116,6 +146,57 @@ fn declare_param( Ok(()) } +/// Validates the regex and every param matcher's capture-group reference before +/// storing. All matchers are checked (not just the first bad one) so a rejection can +/// report every problem at once. +fn declare_metric( + index: &mut BTreeMap, + request: DeclareMetricRequest, +) -> Result<(), String> { + let re = Regex::new(&request.message_match_regex) + .map_err(|err| format!("invalid regex: {err}"))?; + + let errors: Vec = request + .params + .iter() + .filter(|matcher| { + !re.capture_names().flatten().any(|name| name == matcher.capture_group) + }) + .map(|matcher| { + format!( + "param '{}' references capture group '{}', which the regex does not have", + matcher.param_name, matcher.capture_group + ) + }) + .collect(); + if !errors.is_empty() { + return Err(errors.join("; ")); + } + + index.insert( + IndexKey::MetricDeclaration { name: request.name }, + DataValue::MetricDeclaration { + message_match_regex: request.message_match_regex, + params: request.params, + }, + ); + Ok(()) +} + +fn metric_declarations_range( + index: &BTreeMap, +) -> impl Iterator { + index + .range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit) + .filter_map(|(key, value)| match (key, value) { + ( + IndexKey::MetricDeclaration { name }, + DataValue::MetricDeclaration { message_match_regex, params }, + ) => Some((name.as_str(), message_match_regex.as_str(), params.as_slice())), + _ => None, + }) +} + fn param_declarations_range( index: &BTreeMap, ) -> impl Iterator { @@ -208,6 +289,16 @@ fn handle_query(index: &BTreeMap, query: EngineQuery) -> En .collect(); EngineResponse::ParamDeclarations(declarations) } + EngineQuery::MetricDeclarations => { + let declarations = metric_declarations_range(index) + .map(|(name, message_match_regex, params)| MetricDeclarationDto { + name: name.to_string(), + message_match_regex: message_match_regex.to_string(), + params: params.to_vec(), + }) + .collect(); + EngineResponse::MetricDeclarations(declarations) + } } } @@ -288,4 +379,120 @@ mod tests { assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some()); assert!(param_declarations_range(&index).next().is_none()); } + + fn metric_request( + name: &str, + pattern: &str, + params: Vec<(&str, &str, MetricPrimitiveKind)>, + ) -> DeclareMetricRequest { + DeclareMetricRequest { + name: name.to_string(), + message_match_regex: pattern.to_string(), + params: params + .into_iter() + .map(|(param_name, capture_group, kind)| RegexToMetricPrimitiveMatcher { + param_name: param_name.to_string(), + capture_group: capture_group.to_string(), + kind, + }) + .collect(), + } + } + + #[test] + fn declares_valid_metric() { + let mut index = BTreeMap::new(); + let result = declare_metric( + &mut index, + metric_request( + "requests", + r"request path=(?P\S+) duration=(?P\d+)", + vec![ + ("path", "path", MetricPrimitiveKind::String), + ("duration", "duration", MetricPrimitiveKind::Number), + ], + ), + ); + assert!(result.is_ok()); + + let declarations: Vec<_> = metric_declarations_range(&index).collect(); + assert_eq!(declarations.len(), 1); + assert_eq!(declarations[0].0, "requests"); + assert_eq!( + declarations[0].1, + r"request path=(?P\S+) duration=(?P\d+)" + ); + assert_eq!(declarations[0].2.len(), 2); + } + + #[test] + fn rejects_invalid_metric_regex() { + let mut index = BTreeMap::new(); + let result = declare_metric(&mut index, metric_request("requests", "path=(", vec![])); + assert!(result.is_err()); + assert!(index.is_empty()); + } + + #[test] + fn rejects_metric_matcher_with_unknown_capture_group() { + let mut index = BTreeMap::new(); + let result = declare_metric( + &mut index, + metric_request( + "requests", + r"request path=(?P\S+)", + vec![ + ("path", "path", MetricPrimitiveKind::String), + ("duration", "duration", MetricPrimitiveKind::Number), + ], + ), + ); + assert!(result.is_err()); + // Both matchers are validated: the error mentions the one that's actually bad. + assert!(result.unwrap_err().contains("duration")); + assert!(index.is_empty()); + } + + #[test] + fn metric_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(); + declare_metric( + &mut index, + metric_request( + "requests", + r"duration=(?P\d+)", + vec![("duration", "duration", MetricPrimitiveKind::Number)], + ), + ) + .unwrap(); + + let declarations: Vec<_> = metric_declarations_range(&index).collect(); + assert_eq!(declarations.len(), 1); + assert_eq!(declarations[0].0, "requests"); + + for key in index.range(IndexKey::ParamMetricSplit..IndexKey::MetricBucketSplit).map(|(k, _)| k) { + assert!(matches!(key, IndexKey::MetricDeclaration { .. })); + } + } + + #[test] + fn removes_metric_declaration() { + let mut index = BTreeMap::new(); + declare_metric( + &mut index, + metric_request( + "requests", + r"duration=(?P\d+)", + vec![("duration", "duration", MetricPrimitiveKind::Number)], + ), + ) + .unwrap(); + assert!(index.remove(&IndexKey::MetricDeclaration { name: "requests".to_string() }).is_some()); + assert!(metric_declarations_range(&index).next().is_none()); + } }