diff --git a/src/engine/handle.rs b/src/engine/handle.rs index 20442d1..f80ef97 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -9,7 +9,7 @@ use crate::engine::data_value::DataValue; use crate::engine::index_key::IndexKey; use crate::engine::message::EngineMessage; use crate::engine::metric_primitive::{MetricPrimitive, Number}; -use crate::engine::snapshot::params::{derive_log_stream_params_from_row, log_stream_header_transition}; +use crate::engine::snapshot::params::log_stream_header_transition; use crate::engine::snapshot::{approx_log_bytes, EngineSnapshot}; use crate::engine::timestamp::Timestamp; use crate::engine::{ClearParamsResult, IngestedLine}; @@ -144,7 +144,7 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu ); snap.log_count += 1; if let Some(header) = row_header.as_ref() { - derive_log_stream_params_from_row(&mut snap.index, header, &text_for_op); + snap.derive_log_stream_params_from_row(header, &text_for_op); } snap.auto_declare_params(&text_for_op); snap.apply_all_metrics(sequence, timestamp, &text_for_op); @@ -209,6 +209,7 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu write.mutate(move |snap: &mut EngineSnapshot| { snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() }); snap.sort_helper_indexes.remove(&name_for_op); + snap.param_regex_cache.remove(&name_for_op); }); let _ = respond_to.send(existed); } diff --git a/src/engine/snapshot/metrics.rs b/src/engine/snapshot/metrics.rs index 18b6233..9fa1bb7 100644 --- a/src/engine/snapshot/metrics.rs +++ b/src/engine/snapshot/metrics.rs @@ -40,6 +40,7 @@ impl EngineSnapshot { pub(crate) fn declare_metric(&mut self, request: DeclareMetricRequest) -> Result<(), String> { let re = self.validate_metric_request(&request)?; + self.metric_regex_cache.insert(request.name.clone(), re.clone()); let resolved_params = self.resolved_param_declarations(); self.backfill_metric( @@ -87,7 +88,9 @@ impl EngineSnapshot { self.clear_metric_data(existing_name); if request.name != existing_name { self.index.remove(&IndexKey::MetricDeclaration { name: existing_name.to_string() }); + self.metric_regex_cache.remove(existing_name); } + self.metric_regex_cache.insert(request.name.clone(), re.clone()); self.backfill_metric( &request.name, @@ -121,6 +124,7 @@ impl EngineSnapshot { pub(crate) fn remove_metric_declaration(&mut self, name: &str) -> bool { let existed = self.index.remove(&IndexKey::MetricDeclaration { name: name.to_string() }).is_some(); + self.metric_regex_cache.remove(name); self.clear_metric_data(name); existed } @@ -251,7 +255,6 @@ impl EngineSnapshot { pub(crate) fn apply_all_metrics(&mut self, sequence: u64, timestamp: Timestamp, text: &str) { type SeededMetric = ( - String, String, Vec, Option, @@ -260,10 +263,9 @@ impl EngineSnapshot { ); let metrics: Vec = self .metric_declarations_range() - .map(|(name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels)| { + .map(|(name, _, label_params, value_param, matched_labels, unmatched_labels)| { ( name.to_string(), - message_match_regex.to_string(), label_params.to_vec(), value_param.map(|s| s.to_string()), matched_labels.cloned(), @@ -272,8 +274,8 @@ impl EngineSnapshot { }) .collect(); let resolved_params = self.resolved_param_declarations(); - for (name, message_match_regex, label_params, value_param, matched_labels, unmatched_labels) in metrics { - if let Ok(re) = Regex::new(&message_match_regex) { + for (name, label_params, value_param, matched_labels, unmatched_labels) in metrics { + if let Some(re) = self.metric_regex_cache.get(&name).cloned() { self.apply_metric_to_log( &name, &re, diff --git a/src/engine/snapshot/mod.rs b/src/engine/snapshot/mod.rs index 0b00c0e..9b1cb83 100644 --- a/src/engine/snapshot/mod.rs +++ b/src/engine/snapshot/mod.rs @@ -17,6 +17,7 @@ use crate::engine::index_key::IndexKey; use crate::engine::message::{EngineQuery, EngineResponse}; use crate::engine::timestamp::Timestamp; use query::handle_query_inner; +use regex::Regex; use sort::{SortHelperAccess, SortHelperIndexEntry}; use std::collections::{BTreeMap, HashMap}; @@ -24,6 +25,13 @@ use std::collections::{BTreeMap, HashMap}; pub(crate) struct EngineSnapshot { pub(crate) index: BTreeMap, pub(crate) sort_helper_indexes: HashMap, + /// Compiled regexes for param/metric declarations, keyed by declaration name. Compiling a + /// regex is far more expensive than matching with one, and `apply_all_metrics`/ + /// `update_sort_helper_indexes_on_ingest` run per ingested line, so every insertion or + /// removal of a declaration (see `params.rs`/`metrics.rs`) must keep these in sync rather + /// than recompiling from the stored pattern string on every line. + pub(crate) param_regex_cache: HashMap, + pub(crate) metric_regex_cache: HashMap, pub(crate) log_count: usize, pub(crate) approx_bytes_total: u64, pub(crate) ingest_lag_high_watermark_ms: i64, @@ -105,7 +113,28 @@ pub(crate) fn approx_log_bytes(text: &str) -> u64 { impl EngineSnapshot { pub(crate) fn from_index(index: BTreeMap) -> Self { - EngineSnapshot { index, ..Default::default() } + let mut snapshot = EngineSnapshot { index, ..Default::default() }; + snapshot.rebuild_regex_caches(); + snapshot + } + + /// Recompiles `param_regex_cache`/`metric_regex_cache` from the declarations currently in + /// `index`. Needed whenever a snapshot's declarations are populated by inserting straight + /// into `index` rather than through the normal declare/insert methods that keep the caches + /// in sync incrementally (e.g. `from_index` building a snapshot from a bare index). + pub(crate) fn rebuild_regex_caches(&mut self) { + self.param_regex_cache = self + .param_declarations_range() + .filter_map(|(name, message_match_regex, ..)| { + regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re)) + }) + .collect(); + self.metric_regex_cache = self + .metric_declarations_range() + .filter_map(|(name, message_match_regex, ..)| { + regex::Regex::new(message_match_regex).ok().map(|re| (name.to_string(), re)) + }) + .collect(); } pub(crate) fn evict_excess_logs(&mut self, max_logs: Option) { diff --git a/src/engine/snapshot/params.rs b/src/engine/snapshot/params.rs index ade76ba..31e1e68 100644 --- a/src/engine/snapshot/params.rs +++ b/src/engine/snapshot/params.rs @@ -51,6 +51,9 @@ impl EngineSnapshot { } pub(crate) fn insert_param_declaration(&mut self, request: DeclareParamRequest) { + if let Ok(re) = Regex::new(&request.message_match_regex) { + self.param_regex_cache.insert(request.name.clone(), re); + } self.index.insert( IndexKey::ParamDeclaration { name: request.name.clone() }, DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }), @@ -146,6 +149,9 @@ impl EngineSnapshot { if self.index.contains_key(&index_key) { continue; } + if let Ok(re) = Regex::new(&candidate.message_match_regex) { + self.param_regex_cache.insert(candidate.name.clone(), re); + } self.index.insert( index_key, DataValue::ParamDeclaration(ParamDeclaration { @@ -172,56 +178,70 @@ pub(crate) fn log_stream_header_transition( crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone()) } -pub(crate) fn derive_log_stream_params_from_row( - index: &mut BTreeMap, - header: &crate::engine::logfmt::LogStreamHeader, - text: &str, -) { - if !crate::engine::logfmt::is_log_stream_line(text) { - return; - } - let row = header.split_row(text); - - for (i, column) in header.columns.iter().enumerate() { - let Some(value) = row.get(i) else { continue }; - let index_key = IndexKey::ParamDeclaration { name: column.name.clone() }; - if index.contains_key(&index_key) { - continue; +impl EngineSnapshot { + pub(crate) fn derive_log_stream_params_from_row( + &mut self, + header: &crate::engine::logfmt::LogStreamHeader, + text: &str, + ) { + if !crate::engine::logfmt::is_log_stream_line(text) { + return; } - index.insert( - index_key, - DataValue::ParamDeclaration(ParamDeclaration { - name: column.name.clone(), - message_match_regex: header.column_pattern(i, &column.name), - capture_group: column.name.clone(), - json_field_path: None, - kind: infer_kind(value.trim()), - elapsed_since_last: false, - auto_derived: true, - log_stream_gated: true, - }), - ); - } + let row = header.split_row(text); - if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) { - index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| { - DataValue::ParamDeclaration(ParamDeclaration { - name: "message".to_string(), - message_match_regex: header.message_pattern("message"), - capture_group: "message".to_string(), - json_field_path: None, - kind: infer_kind(message.trim()), - elapsed_since_last: false, - auto_derived: true, - log_stream_gated: true, - }) - }); + for (i, column) in header.columns.iter().enumerate() { + let Some(value) = row.get(i) else { continue }; + let index_key = IndexKey::ParamDeclaration { name: column.name.clone() }; + if self.index.contains_key(&index_key) { + continue; + } + let message_match_regex = header.column_pattern(i, &column.name); + if let Ok(re) = Regex::new(&message_match_regex) { + self.param_regex_cache.insert(column.name.clone(), re); + } + self.index.insert( + index_key, + DataValue::ParamDeclaration(ParamDeclaration { + name: column.name.clone(), + message_match_regex, + capture_group: column.name.clone(), + json_field_path: None, + kind: infer_kind(value.trim()), + elapsed_since_last: false, + auto_derived: true, + log_stream_gated: true, + }), + ); + } + + if let Some(message) = row.get(header.columns.len()).filter(|m| !m.trim().is_empty()) { + let index_key = IndexKey::ParamDeclaration { name: "message".to_string() }; + if !self.index.contains_key(&index_key) { + let message_match_regex = header.message_pattern("message"); + if let Ok(re) = Regex::new(&message_match_regex) { + self.param_regex_cache.insert("message".to_string(), re); + } + self.index.insert( + index_key, + DataValue::ParamDeclaration(ParamDeclaration { + name: "message".to_string(), + message_match_regex, + capture_group: "message".to_string(), + json_field_path: None, + kind: infer_kind(message.trim()), + elapsed_since_last: false, + auto_derived: true, + log_stream_gated: true, + }), + ); + } + } } } #[cfg(test)] pub(crate) fn auto_declare_log_stream_params( - index: &mut BTreeMap, + snapshot: &mut EngineSnapshot, header: &mut Option, text: &str, ) { @@ -230,8 +250,8 @@ pub(crate) fn auto_declare_log_stream_params( if is_new_header { return; } - if let Some(header) = header.as_ref() { - derive_log_stream_params_from_row(index, header, text); + if let Some(header) = header.clone() { + snapshot.derive_log_stream_params_from_row(&header, text); } } @@ -241,7 +261,8 @@ impl EngineSnapshot { self.param_declarations_range().map(|(name, ..)| name.to_string()).collect(); let removed = names.len(); for name in names { - self.index.remove(&IndexKey::ParamDeclaration { name }); + self.index.remove(&IndexKey::ParamDeclaration { name: name.clone() }); + self.param_regex_cache.remove(&name); } removed } @@ -269,12 +290,12 @@ pub(super) struct ResolvedParam { impl EngineSnapshot { pub(super) fn resolved_param_declarations(&self) -> HashMap { self.param_declarations_range() - .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| { - Regex::new(message_match_regex).ok().map(|regex| { + .filter_map(|(name, _, capture_group, json_field_path, kind, elapsed_since_last, _, log_stream_gated)| { + self.param_regex_cache.get(name).map(|regex| { ( name.to_string(), ResolvedParam { - regex, + regex: regex.clone(), capture_group: capture_group.to_string(), json_field_path: json_field_path.map(|path| path.to_string()), kind, diff --git a/src/engine/snapshot/sort.rs b/src/engine/snapshot/sort.rs index 5b6757e..bca2753 100644 --- a/src/engine/snapshot/sort.rs +++ b/src/engine/snapshot/sort.rs @@ -97,11 +97,11 @@ impl EngineSnapshot { pub(crate) fn compiled_param_declarations(&self) -> Vec { self.param_declarations_range() .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last) - .filter_map(|(name, message_match_regex, capture_group, json_field_path, kind, _, _, log_stream_gated)| { - Regex::new(message_match_regex).ok().map(|re| { + .filter_map(|(name, _, capture_group, json_field_path, kind, _, _, log_stream_gated)| { + self.param_regex_cache.get(name).map(|re| { ( name.to_string(), - re, + re.clone(), capture_group.to_string(), json_field_path.map(|path| path.to_string()), kind, diff --git a/src/engine/tests.rs b/src/engine/tests.rs index e6f12a4..956f26c 100644 --- a/src/engine/tests.rs +++ b/src/engine/tests.rs @@ -119,7 +119,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; fn ingest_log_stream_fixture(index: &mut EngineSnapshot) { let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { - auto_declare_log_stream_params(&mut index.index, &mut header, line); + auto_declare_log_stream_params(index, &mut header, line); index.auto_declare_params(line); } } @@ -197,7 +197,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; for (i, line) in LOG_STREAM_FIXTURE.lines().enumerate() { let sequence = i as u64; let timestamp = Timestamp::from_millis(sequence as i64); - auto_declare_log_stream_params(&mut index.index, &mut header, line); + auto_declare_log_stream_params(&mut index, &mut header, line); index.auto_declare_params(line); index.index.insert( IndexKey::IngestedLog { sequence, timestamp }, @@ -232,7 +232,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; fn non_log_stream_lines_do_not_get_any_columns_declared_without_a_cached_header() { let mut index = EngineSnapshot::default(); let mut header = None; - auto_declare_log_stream_params(&mut index.index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); + auto_declare_log_stream_params(&mut index, &mut header, r#"ts=1788648105773 msg="entity_moved" x1=-4"#); assert!(index.param_declarations_range().next().is_none()); assert!(header.is_none()); @@ -247,7 +247,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; for _ in 0..repeats { let mut header = None; for line in LOG_STREAM_FIXTURE.lines() { - auto_declare_log_stream_params(&mut index.index, &mut header, line); + auto_declare_log_stream_params(&mut index, &mut header, line); index.auto_declare_params(line); } }