Add param declaration storage and validation to engine index
Implements the ParamDeclaration slice of the primary index: DataValue::ParamDeclaration
holds a regex + named capture group + MetricPrimitiveKind, stored under
IndexKey::ParamDeclaration{name} between LogParamSplit and ParamMetricSplit.
EngineMessage gains DeclareParam (validates the regex compiles and the capture
group exists before storing, returning a clear error otherwise) and
RemoveParamDeclaration; EngineQuery gains ParamDeclarations to list them back out
as ParamDeclarationDto. No API wiring yet (separate ticket).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
1fb564d06b
commit
f62830d8f7
Generated
+39
@@ -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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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,
|
||||
|
||||
+164
-2
@@ -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<Result<(), String>>,
|
||||
},
|
||||
RemoveParamDeclaration {
|
||||
name: String,
|
||||
respond_to: oneshot::Sender<bool>,
|
||||
},
|
||||
Query {
|
||||
query: EngineQuery,
|
||||
respond_to: oneshot::Sender<EngineResponse>,
|
||||
},
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
pub enum EngineResponse {
|
||||
Logs(Vec<LogEntryDto>),
|
||||
Timeline(Vec<TimelineBucketDto>),
|
||||
Stats(StatsDto),
|
||||
ParamDeclarations(Vec<ParamDeclarationDto>),
|
||||
}
|
||||
|
||||
pub type EngineHandle = mpsc::Sender<EngineMessage>;
|
||||
@@ -50,6 +73,14 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>) {
|
||||
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<EngineMessage>) {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<IndexKey, DataValue>,
|
||||
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<IndexKey, DataValue>,
|
||||
) -> impl Iterator<Item = (&str, &str, &str, MetricPrimitiveKind)> {
|
||||
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<IndexKey, DataValue>) -> impl Iterator<Item = (u64, Timestamp, &str)> {
|
||||
index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
|
||||
match (key, value) {
|
||||
@@ -125,5 +197,95 @@ fn handle_query(index: &BTreeMap<IndexKey, DataValue>, 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<level>\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<level>\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<level>\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<level>\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<level>\w+)", "level")).unwrap();
|
||||
assert!(index.remove(&IndexKey::ParamDeclaration { name: "level".to_string() }).is_some());
|
||||
assert!(param_declarations_range(&index).next().is_none());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user