From 3c2995acc180110dae31e9f1c7af75faddc61628 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sun, 6 Sep 2026 17:08:29 +0100 Subject: [PATCH] Convert remaining clean-receiver free functions to methods Profile (legacy-shape migration, load_into_engine, load_legacy_store), MetricDeclarationRecord/Dashboard seed constructors, PromqlQuery::parse, LogStreamHeader (detect/split_row/column_pattern/message_pattern), and api.rs's EngineHandle::ask/*Error::into_response/ParamDeclarationDto::reserved all had a single obvious receiver and are now methods on it instead of free functions taking that type as a parameter. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS --- src/api.rs | 201 +++++++++++++++-------------- src/engine.rs | 90 ++++++------- src/logfmt.rs | 152 +++++++++++----------- src/main.rs | 2 +- src/profile.rs | 328 +++++++++++++++++++++++++----------------------- src/promql.rs | 100 +++++++-------- src/settings.rs | 2 +- 7 files changed, 449 insertions(+), 426 deletions(-) diff --git a/src/api.rs b/src/api.rs index 144be96..e4e8f6f 100644 --- a/src/api.rs +++ b/src/api.rs @@ -14,7 +14,7 @@ use crate::engine::{ EngineResponse, }; use crate::alerts::{deliver_ntfy, AlertError}; -use crate::profile::{self, Dashboard, ParamDeclarationRecord, Profile, ProfileError}; +use crate::profile::{Dashboard, ParamDeclarationRecord, Profile, ProfileError}; use crate::settings::{CssError, SettingsHandle}; use axum::extract::{Path as AxumPath, Query, State}; use axum::http::{header, StatusCode}; @@ -215,25 +215,26 @@ fn mime_for_extension(extension: &str) -> &'static str { } } -async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result { - let read_handle = engine.reader_factory.handle(); - if let Some(sort_by) = query.sort_by() { - let needs_build = match read_handle.enter() { - Some(guard) => !guard.has_sort_helper_index(sort_by), - None => return Err(StatusCode::SERVICE_UNAVAILABLE), - }; - if needs_build { - let (respond_to, response) = oneshot::channel(); - engine - .tx - .send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to }) - .await - .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; - response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; +impl EngineHandle { + async fn ask(&self, query: EngineQuery) -> Result { + let read_handle = self.reader_factory.handle(); + if let Some(sort_by) = query.sort_by() { + let needs_build = match read_handle.enter() { + Some(guard) => !guard.has_sort_helper_index(sort_by), + None => return Err(StatusCode::SERVICE_UNAVAILABLE), + }; + if needs_build { + let (respond_to, response) = oneshot::channel(); + self.tx + .send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to }) + .await + .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; + response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; + } } + let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + Ok(guard.handle_query_ro(query)) } - let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; - Ok(guard.handle_query_ro(query)) } async fn promoted_param_records( @@ -252,7 +253,7 @@ async fn promoted_param_records( return Vec::new(); } let Ok(EngineResponse::ParamDeclarations(declarations)) = - ask(engine, EngineQuery::ParamDeclarations).await + engine.ask(EngineQuery::ParamDeclarations).await else { return Vec::new(); }; @@ -274,7 +275,7 @@ async fn get_logs( State(state): State, Query(params): Query, ) -> impl IntoResponse { - match ask(&state.engine, EngineQuery::Logs(params)).await { + match state.engine.ask(EngineQuery::Logs(params)).await { Ok(EngineResponse::Logs(entries)) => Json(entries).into_response(), Ok(EngineResponse::Error(err)) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() @@ -288,7 +289,7 @@ async fn get_timeline( State(state): State, Query(params): Query, ) -> impl IntoResponse { - match ask(&state.engine, EngineQuery::Timeline(params)).await { + match state.engine.ask(EngineQuery::Timeline(params)).await { Ok(EngineResponse::Timeline(buckets)) => Json(buckets).into_response(), Ok(EngineResponse::Error(err)) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() @@ -299,7 +300,7 @@ async fn get_timeline( } async fn get_stats(State(state): State) -> impl IntoResponse { - match ask(&state.engine, EngineQuery::Stats).await { + match state.engine.ask(EngineQuery::Stats).await { Ok(EngineResponse::Stats(stats)) => Json(stats).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), @@ -307,9 +308,9 @@ async fn get_stats(State(state): State) -> impl IntoResponse { } async fn list_params(State(state): State) -> impl IntoResponse { - match ask(&state.engine, EngineQuery::ParamDeclarations).await { + match state.engine.ask(EngineQuery::ParamDeclarations).await { Ok(EngineResponse::ParamDeclarations(mut declarations)) => { - declarations.extend(reserved_param_declaration_dtos()); + declarations.extend(ParamDeclarationDto::reserved()); Json(declarations).into_response() } Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), @@ -317,39 +318,41 @@ async fn list_params(State(state): State) -> impl IntoResponse { } } -fn reserved_param_declaration_dtos() -> [ParamDeclarationDto; 3] { - [ - ParamDeclarationDto { - name: "ottyl_seq".to_string(), - message_match_regex: String::new(), - capture_group: String::new(), - json_field_path: None, - kind: MetricPrimitiveKind::Number, - elapsed_since_last: false, - auto_derived: false, - reserved: true, - }, - ParamDeclarationDto { - name: "ottyl_ts".to_string(), - message_match_regex: String::new(), - capture_group: String::new(), - json_field_path: None, - kind: MetricPrimitiveKind::Timestamp, - elapsed_since_last: false, - auto_derived: false, - reserved: true, - }, - ParamDeclarationDto { - name: "ottyl_fmt".to_string(), - message_match_regex: String::new(), - capture_group: String::new(), - json_field_path: None, - kind: MetricPrimitiveKind::String, - elapsed_since_last: false, - auto_derived: false, - reserved: true, - }, - ] +impl ParamDeclarationDto { + fn reserved() -> [ParamDeclarationDto; 3] { + [ + ParamDeclarationDto { + name: "ottyl_seq".to_string(), + message_match_regex: String::new(), + capture_group: String::new(), + json_field_path: None, + kind: MetricPrimitiveKind::Number, + elapsed_since_last: false, + auto_derived: false, + reserved: true, + }, + ParamDeclarationDto { + name: "ottyl_ts".to_string(), + message_match_regex: String::new(), + capture_group: String::new(), + json_field_path: None, + kind: MetricPrimitiveKind::Timestamp, + elapsed_since_last: false, + auto_derived: false, + reserved: true, + }, + ParamDeclarationDto { + name: "ottyl_fmt".to_string(), + message_match_regex: String::new(), + capture_group: String::new(), + json_field_path: None, + kind: MetricPrimitiveKind::String, + elapsed_since_last: false, + auto_derived: false, + reserved: true, + }, + ] + } } async fn declare_param( @@ -402,7 +405,7 @@ async fn clear_params(State(state): State) -> impl IntoResponse { } async fn list_metrics(State(state): State) -> impl IntoResponse { - match ask(&state.engine, EngineQuery::MetricDeclarations).await { + match state.engine.ask(EngineQuery::MetricDeclarations).await { Ok(EngineResponse::MetricDeclarations(declarations)) => Json(declarations).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), @@ -521,7 +524,7 @@ async fn get_metric_bucket_values( since_sequence: params.since_sequence, limit: params.limit, }; - match ask(&state.engine, query).await { + match state.engine.ask(query).await { Ok(EngineResponse::BucketValues(values)) => Json(values).into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Err(status) => status.into_response(), @@ -537,7 +540,7 @@ async fn run_promql_query( start_ms: params.start_ms, end_ms: params.end_ms, }; - match ask(&state.engine, query).await { + match state.engine.ask(query).await { Ok(EngineResponse::Promql(Ok(series))) => Json(series).into_response(), Ok(EngineResponse::Promql(Err(err))) => { (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() @@ -547,11 +550,13 @@ async fn run_promql_query( } } -fn css_error_response(err: CssError) -> axum::response::Response { - match err { - CssError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), - CssError::NotImported => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), - CssError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), +impl CssError { + fn into_response(self) -> axum::response::Response { + match self { + CssError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(), + CssError::NotImported => (StatusCode::BAD_REQUEST, self.to_string()).into_response(), + CssError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(), + } } } @@ -576,7 +581,7 @@ async fn import_css( .expect("entry was just inserted by import_css"); (StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response() } - Err(err) => css_error_response(err), + Err(err) => err.into_response(), } } @@ -593,7 +598,7 @@ async fn link_css( .expect("entry was just pushed by link_css"); (StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response() } - Err(err) => css_error_response(err), + Err(err) => err.into_response(), } } @@ -605,7 +610,7 @@ async fn remove_css( let mut settings = state.settings.lock().unwrap(); match settings.remove_css(&state.settings_path, &css_dir, &id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => css_error_response(err), + Err(err) => err.into_response(), } } @@ -622,14 +627,16 @@ async fn get_css_content( content, ) .into_response(), - Err(err) => css_error_response(err), + Err(err) => err.into_response(), } } -fn alert_error_response(err: AlertError) -> axum::response::Response { - match err { - AlertError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), - AlertError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), +impl AlertError { + fn into_response(self) -> axum::response::Response { + match self { + AlertError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(), + AlertError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(), + } } } @@ -645,7 +652,7 @@ async fn create_alert_rule( let mut settings = state.settings.lock().unwrap(); match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition) { Ok(rule) => (StatusCode::CREATED, Json(rule)).into_response(), - Err(err) => alert_error_response(err), + Err(err) => err.into_response(), } } @@ -657,7 +664,7 @@ async fn update_alert_rule( let mut settings = state.settings.lock().unwrap(); match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition) { Ok(rule) => Json(rule).into_response(), - Err(err) => alert_error_response(err), + Err(err) => err.into_response(), } } @@ -665,7 +672,7 @@ async fn delete_alert_rule(State(state): State, AxumPath(id): AxumPath let mut settings = state.settings.lock().unwrap(); match settings.remove_alert_rule(&state.settings_path, &id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => alert_error_response(err), + Err(err) => err.into_response(), } } @@ -721,12 +728,14 @@ async fn update_settings( Json(new_max_logs.0).into_response() } -fn profile_error_response(err: ProfileError) -> axum::response::Response { - match err { - ProfileError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), - ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), - ProfileError::CannotDeleteDefault => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), - ProfileError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), +impl ProfileError { + fn into_response(self) -> axum::response::Response { + match self { + ProfileError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(), + ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, self.to_string()).into_response(), + ProfileError::CannotDeleteDefault => (StatusCode::BAD_REQUEST, self.to_string()).into_response(), + ProfileError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, self.to_string()).into_response(), + } } } @@ -746,7 +755,7 @@ async fn create_profile( let profile = settings.profile(&body.name).expect("profile was just created by create_profile"); (StatusCode::CREATED, Json(ProfileSummaryDto::from(profile))).into_response() } - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -757,13 +766,13 @@ async fn delete_profile( let mut settings = state.settings.lock().unwrap(); match settings.delete_profile(&state.settings_path, &name) { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { if let Ok(EngineResponse::ParamDeclarations(params)) = - ask(engine, EngineQuery::ParamDeclarations).await + engine.ask(EngineQuery::ParamDeclarations).await { for param in params.into_iter().filter(|param| !param.auto_derived) { let (respond_to, response) = oneshot::channel(); @@ -778,7 +787,7 @@ async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { } } if let Ok(EngineResponse::MetricDeclarations(metrics)) = - ask(engine, EngineQuery::MetricDeclarations).await + engine.ask(EngineQuery::MetricDeclarations).await { for metric in metrics { let (respond_to, response) = oneshot::channel(); @@ -792,7 +801,7 @@ async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { } } } - profile::load_into_engine(engine, profile).await; + profile.load_into_engine(engine).await; } async fn activate_profile( @@ -815,7 +824,7 @@ async fn activate_profile( let settings = state.settings.lock().unwrap(); Json(SettingsDto::from(&*settings)).into_response() } - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -829,7 +838,7 @@ async fn list_dashboards( let dtos: Vec = dashboards.iter().map(DashboardDto::from).collect(); Json(dtos).into_response() } - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -841,7 +850,7 @@ async fn create_dashboard( let mut settings = state.settings.lock().unwrap(); match settings.create_dashboard(&state.settings_path, &name, body.name) { Ok(dashboard) => (StatusCode::CREATED, Json(DashboardDto::from(&dashboard))).into_response(), - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -853,9 +862,9 @@ async fn get_dashboard( match settings.list_dashboards(&name) { Ok(dashboards) => match dashboards.iter().find(|d| d.id == dashboard_id) { Some(dashboard) => Json(DashboardDto::from(dashboard)).into_response(), - None => profile_error_response(ProfileError::NotFound), + None => ProfileError::NotFound.into_response(), }, - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -872,7 +881,7 @@ async fn save_dashboard( }; match settings.save_dashboard(&state.settings_path, &name, &dashboard_id, dashboard) { Ok(dashboard) => Json(DashboardDto::from(&dashboard)).into_response(), - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -883,7 +892,7 @@ async fn delete_dashboard( let mut settings = state.settings.lock().unwrap(); match settings.delete_dashboard(&state.settings_path, &name, &dashboard_id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), - Err(err) => profile_error_response(err), + Err(err) => err.into_response(), } } @@ -1512,9 +1521,9 @@ mod tests { let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); - profile::load_into_engine(&fresh_engine_tx, profile).await; + profile.load_into_engine(&fresh_engine_tx).await; - let response = ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await.unwrap(); + let response = fresh_engine_tx.ask(EngineQuery::MetricDeclarations).await.unwrap(); let EngineResponse::MetricDeclarations(metrics) = response else { panic!("unexpected response"); }; diff --git a/src/engine.rs b/src/engine.rs index 4a98fee..a008953 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -13,7 +13,7 @@ use crate::filter_datetime; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::index_key::IndexKey; use crate::metric_primitive::{MetricPrimitive, Number}; -use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; +use crate::promql::{AggOp, LabelMatcher, PromqlQuery}; use crate::rate; use crate::settings::SettingsHandle; use crate::timestamp::Timestamp; @@ -208,7 +208,7 @@ async fn run( let ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at); - let is_new_header = crate::logfmt::detect_log_stream_header(&text).is_some(); + let is_new_header = crate::logfmt::LogStreamHeader::detect(&text).is_some(); log_stream_header = log_stream_header_transition(&log_stream_header, &text); let row_header = if is_new_header { None } else { log_stream_header.clone() }; @@ -527,7 +527,7 @@ fn log_stream_header_transition( current: &Option, text: &str, ) -> Option { - crate::logfmt::detect_log_stream_header(text).or_else(|| current.clone()) + crate::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone()) } fn derive_log_stream_params_from_row( @@ -538,7 +538,7 @@ fn derive_log_stream_params_from_row( if !crate::logfmt::is_log_stream_line(text) { return; } - let row = crate::logfmt::split_log_stream_row(text, header); + let row = header.split_row(text); for (i, column) in header.columns.iter().enumerate() { let Some(value) = row.get(i) else { continue }; @@ -550,7 +550,7 @@ fn derive_log_stream_params_from_row( index_key, DataValue::ParamDeclaration(ParamDeclaration { name: column.name.clone(), - message_match_regex: crate::logfmt::log_stream_column_pattern(header, i, &column.name), + message_match_regex: header.column_pattern(i, &column.name), capture_group: column.name.clone(), json_field_path: None, kind: infer_kind(value.trim()), @@ -565,7 +565,7 @@ fn derive_log_stream_params_from_row( index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| { DataValue::ParamDeclaration(ParamDeclaration { name: "message".to_string(), - message_match_regex: crate::logfmt::log_stream_message_pattern(header, "message"), + message_match_regex: header.message_pattern("message"), capture_group: "message".to_string(), json_field_path: None, kind: infer_kind(message.trim()), @@ -583,7 +583,7 @@ fn auto_declare_log_stream_params( header: &mut Option, text: &str, ) { - let is_new_header = crate::logfmt::detect_log_stream_header(text).is_some(); + let is_new_header = crate::logfmt::LogStreamHeader::detect(text).is_some(); *header = log_stream_header_transition(header, text); if is_new_header { return; @@ -1802,7 +1802,7 @@ fn handle_query_inner( EngineResponse::BucketValues(values) } EngineQuery::Promql { query, start_ms, end_ms } => { - let response = promql::parse_query(&query) + let response = PromqlQuery::parse(&query) .and_then(|parsed| snapshot.evaluate_promql(&parsed, start_ms, end_ms)); EngineResponse::Promql(response) } @@ -2797,7 +2797,7 @@ mod tests { declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); - let parsed = promql::parse_query("requests").unwrap(); + let parsed = PromqlQuery::parse("requests").unwrap(); assert!(index.evaluate_promql(&parsed, None, None).unwrap().len() == 1); assert!(index.remove_metric_declaration("requests")); @@ -2816,7 +2816,7 @@ mod tests { Timestamp::from_millis(0), MetricPrimitive::Number(Number::Float(1.0)), ); - let parsed = promql::parse_query("scraped_metric").unwrap(); + let parsed = PromqlQuery::parse("scraped_metric").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); } @@ -2824,7 +2824,7 @@ mod tests { #[test] fn promql_rejects_metric_with_no_declaration_and_no_data() { let index = EngineSnapshot::default(); - let parsed = promql::parse_query("never_declared").unwrap(); + let parsed = PromqlQuery::parse("never_declared").unwrap(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err(); assert!(err.contains("unknown metric"), "unexpected error message: {err}"); } @@ -3053,7 +3053,7 @@ mod tests { ingest_log(&mut index, 1, "request path=/b"); ingest_log(&mut index, 2, "request path=/a"); - let parsed = promql::parse_query(r#"requests{path="/a"}"#).unwrap(); + let parsed = PromqlQuery::parse(r#"requests{path="/a"}"#).unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3067,7 +3067,7 @@ mod tests { declare_requests_metric(&mut index); ingest_log(&mut index, 0, "request path=/a"); - let parsed = promql::parse_query(r#"requests{path="/nope"}"#).unwrap(); + let parsed = PromqlQuery::parse(r#"requests{path="/nope"}"#).unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert!(series.is_empty()); } @@ -3079,7 +3079,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); - let parsed = promql::parse_query("sum(requests)").unwrap(); + let parsed = PromqlQuery::parse("sum(requests)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3095,7 +3095,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/b"); ingest_log(&mut index, 1, "request path=/a"); - let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); + let parsed = PromqlQuery::parse("sum(requests) by (path)").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); @@ -3154,7 +3154,7 @@ mod tests { declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); - let parsed = promql::parse_query("sum(statuses)").unwrap(); + let parsed = PromqlQuery::parse("sum(statuses)").unwrap(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err(); assert!(err.contains("statuses"), "error should name the offending metric: {err}"); @@ -3167,7 +3167,7 @@ mod tests { declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); - let parsed = promql::parse_query("avg(statuses)").unwrap(); + let parsed = PromqlQuery::parse("avg(statuses)").unwrap(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err(); assert!(err.contains("statuses")); @@ -3180,7 +3180,7 @@ mod tests { ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 1, "request status=error"); - let parsed = promql::parse_query("count(statuses)").unwrap(); + let parsed = PromqlQuery::parse("count(statuses)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3195,7 +3195,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); - let parsed = promql::parse_query("sum(requests)").unwrap(); + let parsed = PromqlQuery::parse("sum(requests)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3209,7 +3209,7 @@ mod tests { ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 1000, "request status=error"); - let parsed = promql::parse_query("sum(rate(statuses[2s]))").unwrap(); + let parsed = PromqlQuery::parse("sum(rate(statuses[2s]))").unwrap(); let series = index.evaluate_promql(&parsed, None, None); assert!(series.is_ok(), "sum(rate(...)) never reads the coerced value, so it should not be rejected: {series:?}"); @@ -3221,7 +3221,7 @@ mod tests { declare_string_valued_metric(&mut index); ingest_log(&mut index, 0, "request status=ok"); - let parsed = promql::parse_query("statuses").unwrap(); + let parsed = PromqlQuery::parse("statuses").unwrap(); let series = index.evaluate_promql(&parsed, None, None); assert!(series.is_ok()); @@ -3235,7 +3235,7 @@ mod tests { ingest_log(&mut index, 1, "request path=/a host=web2"); ingest_log(&mut index, 2, "request path=/a host=web3"); - let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); + let parsed = PromqlQuery::parse("sum(requests) by (path)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3256,7 +3256,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a host=web1"); ingest_log(&mut index, 1, "request path=/a host=web2"); - let parsed = promql::parse_query("avg(requests) by (path)").unwrap(); + let parsed = PromqlQuery::parse("avg(requests) by (path)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3271,7 +3271,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/a"); - let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); + let parsed = PromqlQuery::parse("sum(requests) by (path)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3285,7 +3285,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); - let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let parsed = PromqlQuery::parse("rate(requests[2s])").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3300,7 +3300,7 @@ mod tests { ingest_log(&mut index, 1000, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); - let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); + let parsed = PromqlQuery::parse("sum(rate(requests[2s])) by (path)").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); @@ -3318,7 +3318,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1, "request path=/a"); - let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let parsed = PromqlQuery::parse("rate(requests[2s])").unwrap(); let series = index.evaluate_promql(&parsed, None, Some(100_000)).unwrap(); assert_eq!(series.len(), 1); @@ -3335,7 +3335,7 @@ mod tests { declare_requests_metric(&mut index); ingest_log(&mut index, 50_000, "request path=/a"); - let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let parsed = PromqlQuery::parse("rate(requests[2s])").unwrap(); let series = index.evaluate_promql(&parsed, Some(0), None).unwrap(); assert_eq!(series[0].points, vec![(0, 0.0), (50_000, 0.5)]); @@ -3349,7 +3349,7 @@ mod tests { ingest_log(&mut index, 1, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); - let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); + let parsed = PromqlQuery::parse("sum(rate(requests[2s])) by (path)").unwrap(); let mut series = index.evaluate_promql(&parsed, None, Some(100_000)).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); @@ -3365,7 +3365,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a"); - let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let parsed = PromqlQuery::parse("rate(requests[2s])").unwrap(); let series = index.evaluate_promql(&parsed, Some(0), Some(1000)).unwrap(); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); @@ -3387,7 +3387,7 @@ mod tests { ingest_log(&mut index, *ms, text); } - let parsed = promql::parse_query("rate(requests[2s])").unwrap(); + let parsed = PromqlQuery::parse("rate(requests[2s])").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); let (_, dashboard_rate) = *series[0].points.last().unwrap(); @@ -3431,7 +3431,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/b"); - let parsed = promql::parse_query("sum(rate(requests[2s]))").unwrap(); + let parsed = PromqlQuery::parse("sum(rate(requests[2s]))").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3447,7 +3447,7 @@ mod tests { ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 10, "request path=/a"); - let parsed = promql::parse_query("requests").unwrap(); + let parsed = PromqlQuery::parse("requests").unwrap(); let series = index.evaluate_promql(&parsed, Some(1), Some(9)).unwrap(); assert_eq!(series.len(), 1); @@ -3503,7 +3503,7 @@ mod tests { ingest_log(&mut index, 1, "request path=/a duration=9"); ingest_log(&mut index, 2, "request path=/b duration=42"); - let parsed = promql::parse_query("gauge(durations)").unwrap(); + let parsed = PromqlQuery::parse("gauge(durations)").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); @@ -3522,7 +3522,7 @@ mod tests { ingest_log(&mut index, 5, "request path=/a duration=9"); ingest_log(&mut index, 10, "request path=/a duration=42"); - let parsed = promql::parse_query("gauge(durations)").unwrap(); + let parsed = PromqlQuery::parse("gauge(durations)").unwrap(); let series = index.evaluate_promql(&parsed, None, Some(9)).unwrap(); assert_eq!(series.len(), 1); @@ -3537,7 +3537,7 @@ mod tests { ingest_log(&mut index, seq, &format!("request path=/a duration={duration}")); } - let parsed = promql::parse_query("histogram(durations[10])").unwrap(); + let parsed = PromqlQuery::parse("histogram(durations[10])").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le"))); @@ -3563,7 +3563,7 @@ mod tests { ingest_log(&mut index, seq, &format!("request path={path} duration={duration}")); } - let parsed = promql::parse_query("histogram(durations[10]) by (path)").unwrap(); + let parsed = PromqlQuery::parse("histogram(durations[10]) by (path)").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| { a.labels.get("path").cmp(&b.labels.get("path")).then(a.labels.get("le").cmp(&b.labels.get("le"))) @@ -3631,7 +3631,7 @@ mod tests { ingest_log(&mut index, 0, "user=alice logged in"); ingest_log(&mut index, 10, "user=alice logged in"); - let parsed = promql::parse_query("login_gap").unwrap(); + let parsed = PromqlQuery::parse("login_gap").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); @@ -3678,7 +3678,7 @@ mod tests { ingest_log(&mut index, 1, "all good"); ingest_log(&mut index, 2, "all good again"); - let parsed = promql::parse_query("log_category").unwrap(); + let parsed = PromqlQuery::parse("log_category").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category"))); @@ -3750,7 +3750,7 @@ mod tests { ingest_log(&mut index, 1, "line two"); ingest_log(&mut index, 2, "line three"); - let parsed = promql::parse_query("count(ottyl_seq)").unwrap(); + let parsed = PromqlQuery::parse("count(ottyl_seq)").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0), (2, 1.0)]); @@ -3762,7 +3762,7 @@ mod tests { ingest_log(&mut index, 0, "line zero"); ingest_log(&mut index, 5, "line five"); - let parsed = promql::parse_query("ottyl_seq").unwrap(); + let parsed = PromqlQuery::parse("ottyl_seq").unwrap(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); series[0].points.sort_by_key(|(ts, _)| *ts); @@ -3775,7 +3775,7 @@ mod tests { ingest_log(&mut index, 0, "a"); ingest_log(&mut index, 1, "b"); - let parsed = promql::parse_query("ottyl_ts").unwrap(); + let parsed = PromqlQuery::parse("ottyl_ts").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 0.0), (1, 1.0)]); @@ -3788,7 +3788,7 @@ mod tests { ingest_log(&mut index, sequence, "line"); } - let parsed = promql::parse_query("rate(ottyl_ts[5s])").unwrap(); + let parsed = PromqlQuery::parse("rate(ottyl_ts[5s])").unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(!series[0].points.is_empty(), "rate(ottyl_ts[...]) must not silently return no data"); @@ -3799,7 +3799,7 @@ mod tests { let mut index = EngineSnapshot::default(); ingest_log(&mut index, 0, "line"); - let parsed = promql::parse_query(r#"ottyl_seq{path="/a"}"#).unwrap(); + let parsed = PromqlQuery::parse(r#"ottyl_seq{path="/a"}"#).unwrap(); let series = index.evaluate_promql(&parsed, None, None).unwrap(); assert!(series.is_empty()); } @@ -3809,7 +3809,7 @@ mod tests { let mut index = EngineSnapshot::default(); ingest_log(&mut index, 0, r#"{"level":"error"}"#); - let parsed = promql::parse_query("count(ottyl_fmt)").unwrap(); + let parsed = PromqlQuery::parse("count(ottyl_fmt)").unwrap(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err(); assert!(err.contains("ottyl_fmt"), "error should name the rejected metric: {err}"); assert!(err.contains("categorical") || err.contains("filter"), "error should explain why: {err}"); diff --git a/src/logfmt.rs b/src/logfmt.rs index 57bf137..f9ad9b7 100644 --- a/src/logfmt.rs +++ b/src/logfmt.rs @@ -138,78 +138,80 @@ fn char_slice(s: &str, start: usize, end: Option) -> &str { &s[byte_start..byte_end.max(byte_start)] } -pub fn detect_log_stream_header(line: &str) -> Option { - if is_log_stream_line(line) { - return None; - } - let mut columns = Vec::new(); - for (byte_start, slice) in split_columns_with_offsets(line, None) { - let trimmed = slice.trim(); - if trimmed.is_empty() { - continue; +impl LogStreamHeader { + pub fn detect(line: &str) -> Option { + if is_log_stream_line(line) { + return None; } - let leading_ws = slice.len() - slice.trim_start().len(); - let start = line[..byte_start + leading_ws].chars().count(); - columns.push(LogStreamColumn { name: trimmed.to_string(), start }); + let mut columns = Vec::new(); + for (byte_start, slice) in split_columns_with_offsets(line, None) { + let trimmed = slice.trim(); + if trimmed.is_empty() { + continue; + } + let leading_ws = slice.len() - slice.trim_start().len(); + let start = line[..byte_start + leading_ws].chars().count(); + columns.push(LogStreamColumn { name: trimmed.to_string(), start }); + } + if columns.len() < 2 || !columns.iter().all(|c| COLUMN_NAME.is_match(&c.name)) { + return None; + } + let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t')); + Some(LogStreamHeader { columns, tab_delimited }) } - if columns.len() < 2 || !columns.iter().all(|c| COLUMN_NAME.is_match(&c.name)) { - return None; - } - let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t')); - Some(LogStreamHeader { columns, tab_delimited }) -} -pub fn split_log_stream_row<'a>(line: &'a str, header: &LogStreamHeader) -> Vec<&'a str> { - if header.tab_delimited { - return split_columns(line, Some(header.columns.len() + 1)); - } - let mut result = Vec::with_capacity(header.columns.len() + 1); - for (i, column) in header.columns.iter().enumerate() { - match header.columns.get(i + 1) { - Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()), - None => { - let tail = char_slice(line, column.start, None); - let pieces = split_columns(tail, Some(2)); - result.push(pieces[0].trim()); - if let Some(message) = pieces.get(1) { - result.push(message); + pub fn split_row<'a>(&self, line: &'a str) -> Vec<&'a str> { + if self.tab_delimited { + return split_columns(line, Some(self.columns.len() + 1)); + } + let mut result = Vec::with_capacity(self.columns.len() + 1); + for (i, column) in self.columns.iter().enumerate() { + match self.columns.get(i + 1) { + Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()), + None => { + let tail = char_slice(line, column.start, None); + let pieces = split_columns(tail, Some(2)); + result.push(pieces[0].trim()); + if let Some(message) = pieces.get(1) { + result.push(message); + } } } } + result } - result -} -pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture_group: &str) -> String { - if header.tab_delimited { - let mut pattern = String::from("^"); - for _ in 0..index { - pattern.push_str(COLUMN_VALUE_PATTERN); - pattern.push_str(COLUMN_SEPARATOR_PATTERN); + pub fn column_pattern(&self, index: usize, capture_group: &str) -> String { + if self.tab_delimited { + let mut pattern = String::from("^"); + for _ in 0..index { + pattern.push_str(COLUMN_VALUE_PATTERN); + pattern.push_str(COLUMN_SEPARATOR_PATTERN); + } + pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})")); + return pattern; } - pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})")); - return pattern; + let start = self.columns[index].start; + let value_pattern = match self.columns.get(index + 1) { + Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)), + None => COLUMN_VALUE_PATTERN.to_string(), + }; + format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})") } - let start = header.columns[index].start; - let value_pattern = match header.columns.get(index + 1) { - Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)), - None => COLUMN_VALUE_PATTERN.to_string(), - }; - format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})") -} -pub fn log_stream_message_pattern(header: &LogStreamHeader, capture_group: &str) -> String { - if header.tab_delimited { - let mut pattern = String::from("^"); - for _ in 0..header.columns.len() { - pattern.push_str(COLUMN_VALUE_PATTERN); - pattern.push_str(COLUMN_SEPARATOR_PATTERN); + pub fn message_pattern(&self, capture_group: &str) -> String { + if self.tab_delimited { + let mut pattern = String::from("^"); + for _ in 0..self.columns.len() { + pattern.push_str(COLUMN_VALUE_PATTERN); + pattern.push_str(COLUMN_SEPARATOR_PATTERN); + } + pattern.push_str(&format!("(?P<{capture_group}>.*)$")); + return pattern; } - pattern.push_str(&format!("(?P<{capture_group}>.*)$")); - return pattern; + let start = self.columns.last().map(|c| c.start).unwrap_or(0); + format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$") } - let start = header.columns.last().map(|c| c.start).unwrap_or(0); - format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$") } fn go_log_prefix(input: &str) -> IResult<&str, (&str, &str, &str)> { @@ -368,7 +370,7 @@ mod tests { #[test] fn detects_log_stream_header_columns_generically() { let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); - let header = detect_log_stream_header(header_line).expect("fixture header must be detected"); + let header = LogStreamHeader::detect(header_line).expect("fixture header must be detected"); assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type", "Activity", "PID", "TTL"]); assert!(!header.tab_delimited); } @@ -376,21 +378,21 @@ mod tests { #[test] fn a_log_stream_data_line_is_never_detected_as_its_own_header() { let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap(); - assert!(detect_log_stream_header(data_line).is_none()); + assert!(LogStreamHeader::detect(data_line).is_none()); } #[test] fn plain_prose_is_not_detected_as_a_log_stream_header() { - assert!(detect_log_stream_header("just some plain prose here").is_none()); - assert!(detect_log_stream_header("ts=1788648105773 msg=\"entity_moved\"").is_none()); + assert!(LogStreamHeader::detect("just some plain prose here").is_none()); + assert!(LogStreamHeader::detect("ts=1788648105773 msg=\"entity_moved\"").is_none()); } #[test] fn splits_log_stream_row_into_named_columns_plus_trailing_message() { let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); - let header = detect_log_stream_header(header_line).unwrap(); + let header = LogStreamHeader::detect(header_line).unwrap(); let data_line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened"; - let row = split_log_stream_row(data_line, &header); + let row = header.split_row(data_line); assert_eq!(row.len(), header.columns.len() + 1); assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100"); assert_eq!(row[1], "0x4dfed2"); @@ -403,12 +405,12 @@ mod tests { #[test] fn a_single_tab_separates_columns_exactly_like_a_run_of_spaces() { - let header = detect_log_stream_header("Timestamp\tThread\tType").unwrap(); + let header = LogStreamHeader::detect("Timestamp\tThread\tType").unwrap(); assert_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type"]); assert!(header.tab_delimited); - let spaced = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message", &header); - let tabbed = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message", &header); + let spaced = header.split_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message"); + let tabbed = header.split_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message"); assert_eq!(spaced[0], "2026-09-06 01:48:54"); assert_eq!(spaced[1], "0x4dfed2"); assert_eq!(spaced[2], "Default trailing message"); @@ -420,22 +422,22 @@ mod tests { #[test] fn column_and_message_patterns_capture_the_same_fields_split_log_stream_row_finds() { let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); - let header = detect_log_stream_header(header_line).unwrap(); + let header = LogStreamHeader::detect(header_line).unwrap(); let line = "2026-09-06 01:48:54.518117+0100 0x4dfed2 Default 0x0 606 0 WindowServer: (HSTouchHIDService) something happened"; - let type_re = Regex::new(&log_stream_column_pattern(&header, 2, "Type")).unwrap(); + let type_re = Regex::new(&header.column_pattern(2, "Type")).unwrap(); assert_eq!(type_re.captures(line).unwrap().name("Type").unwrap().as_str(), "Default"); - let pid_re = Regex::new(&log_stream_column_pattern(&header, 4, "PID")).unwrap(); + let pid_re = Regex::new(&header.column_pattern(4, "PID")).unwrap(); assert_eq!(pid_re.captures(line).unwrap().name("PID").unwrap().as_str(), "606"); - let message_re = Regex::new(&log_stream_message_pattern(&header, "message")).unwrap(); + let message_re = Regex::new(&header.message_pattern("message")).unwrap(); assert_eq!( message_re.captures(line).unwrap().name("message").unwrap().as_str(), "WindowServer: (HSTouchHIDService) something happened" ); - let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap(); + let timestamp_re = Regex::new(&header.column_pattern(0, "Timestamp")).unwrap(); assert_eq!( timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(), "2026-09-06 01:48:54.518117+0100" @@ -445,8 +447,8 @@ mod tests { #[test] fn is_log_stream_line_rejects_non_columnar_lines_that_a_fixed_offset_column_regex_would_still_match() { let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); - let header = detect_log_stream_header(header_line).unwrap(); - let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap(); + let header = LogStreamHeader::detect(header_line).unwrap(); + let timestamp_re = Regex::new(&header.column_pattern(0, "Timestamp")).unwrap(); let continuation_line = "so_gencnt: 4042707 so_state: 0x0"; assert!(timestamp_re.captures(continuation_line).is_some(), "sanity: regex alone still matches garbage"); diff --git a/src/main.rs b/src/main.rs index 1f0fedd..28bdda6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -110,7 +110,7 @@ async fn main() { }; match (active_profile_name, active_profile) { (Some(_), Some(active_profile)) => { - profile::load_into_engine(&engine_tx, &active_profile).await; + active_profile.load_into_engine(&engine_tx).await; } (Some(profile_name), None) => { eprintln!( diff --git a/src/profile.rs b/src/profile.rs index d2b4dab..2267e50 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -153,78 +153,80 @@ impl From<&MetricDeclarationRecord> for DeclareMetricRequest { } } -fn migrate_legacy_metric_params(profile: &mut Profile) -> bool { - let mut migrated = false; - for metric_index in 0..profile.metrics.len() { - if profile.metrics[metric_index].legacy_params.is_empty() { - continue; - } - let metric_name = profile.metrics[metric_index].name.clone(); - let message_match_regex = profile.metrics[metric_index].message_match_regex.clone(); - let legacy = std::mem::take(&mut profile.metrics[metric_index].legacy_params); - - let mut label_params = Vec::new(); - let mut value_param = None; - for matcher in &legacy { - let param_name = - resolve_or_create_legacy_param(profile, &metric_name, &message_match_regex, matcher); - match matcher.role { - LegacyParamRole::Label => label_params.push(param_name), - LegacyParamRole::Value => value_param = Some(param_name), +impl Profile { + fn migrate_legacy_metric_params(&mut self) -> bool { + let mut migrated = false; + for metric_index in 0..self.metrics.len() { + if self.metrics[metric_index].legacy_params.is_empty() { + continue; } - } - profile.metrics[metric_index].label_params = label_params; - profile.metrics[metric_index].value_param = value_param; - migrated = true; - } - migrated -} + let metric_name = self.metrics[metric_index].name.clone(); + let message_match_regex = self.metrics[metric_index].message_match_regex.clone(); + let legacy = std::mem::take(&mut self.metrics[metric_index].legacy_params); -fn resolve_or_create_legacy_param( - profile: &mut Profile, - metric_name: &str, - message_match_regex: &str, - matcher: &LegacyMetricParamRecord, -) -> String { - if let Some(existing) = profile.params.iter().find(|p| p.name == matcher.param_name) { - if existing.message_match_regex == message_match_regex - && existing.capture_group == matcher.capture_group - && existing.json_field_path == matcher.json_field_path - && existing.kind == matcher.kind - { - return existing.name.clone(); - } - } - - let implicit_name = format!("{metric_name}.{}", matcher.param_name); - if !profile.params.iter().any(|p| p.name == implicit_name) { - profile.params.push(ParamDeclarationRecord { - name: implicit_name.clone(), - message_match_regex: message_match_regex.to_string(), - capture_group: matcher.capture_group.clone(), - json_field_path: matcher.json_field_path.clone(), - kind: matcher.kind, - elapsed_since_last: matcher.role == LegacyParamRole::Value - && matcher.kind == ProfileMetricPrimitiveKind::Duration, - }); - } - implicit_name -} - -fn migrate_duration_value_params(profile: &mut Profile) -> bool { - let value_param_names: std::collections::HashSet<&str> = - profile.metrics.iter().filter_map(|metric| metric.value_param.as_deref()).collect(); - let mut migrated = false; - for param in &mut profile.params { - if !param.elapsed_since_last - && param.kind == ProfileMetricPrimitiveKind::Duration - && value_param_names.contains(param.name.as_str()) - { - param.elapsed_since_last = true; + let mut label_params = Vec::new(); + let mut value_param = None; + for matcher in &legacy { + let param_name = + self.resolve_or_create_legacy_param(&metric_name, &message_match_regex, matcher); + match matcher.role { + LegacyParamRole::Label => label_params.push(param_name), + LegacyParamRole::Value => value_param = Some(param_name), + } + } + self.metrics[metric_index].label_params = label_params; + self.metrics[metric_index].value_param = value_param; migrated = true; } + migrated + } + + fn resolve_or_create_legacy_param( + &mut self, + metric_name: &str, + message_match_regex: &str, + matcher: &LegacyMetricParamRecord, + ) -> String { + if let Some(existing) = self.params.iter().find(|p| p.name == matcher.param_name) { + if existing.message_match_regex == message_match_regex + && existing.capture_group == matcher.capture_group + && existing.json_field_path == matcher.json_field_path + && existing.kind == matcher.kind + { + return existing.name.clone(); + } + } + + let implicit_name = format!("{metric_name}.{}", matcher.param_name); + if !self.params.iter().any(|p| p.name == implicit_name) { + self.params.push(ParamDeclarationRecord { + name: implicit_name.clone(), + message_match_regex: message_match_regex.to_string(), + capture_group: matcher.capture_group.clone(), + json_field_path: matcher.json_field_path.clone(), + kind: matcher.kind, + elapsed_since_last: matcher.role == LegacyParamRole::Value + && matcher.kind == ProfileMetricPrimitiveKind::Duration, + }); + } + implicit_name + } + + fn migrate_duration_value_params(&mut self) -> bool { + let value_param_names: std::collections::HashSet<&str> = + self.metrics.iter().filter_map(|metric| metric.value_param.as_deref()).collect(); + let mut migrated = false; + for param in &mut self.params { + if !param.elapsed_since_last + && param.kind == ProfileMetricPrimitiveKind::Duration + && value_param_names.contains(param.name.as_str()) + { + param.elapsed_since_last = true; + migrated = true; + } + } + migrated } - migrated } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -269,35 +271,39 @@ impl Dashboard { } } -fn seeded_error_split_metric() -> MetricDeclarationRecord { - MetricDeclarationRecord { - name: SEEDED_METRIC_NAME.to_string(), - message_match_regex: r"(?i)error".to_string(), - label_params: Vec::new(), - value_param: None, - matched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "error".to_string())])), - unmatched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "other".to_string())])), - legacy_params: Vec::new(), +impl MetricDeclarationRecord { + fn seeded_error_split() -> MetricDeclarationRecord { + MetricDeclarationRecord { + name: SEEDED_METRIC_NAME.to_string(), + message_match_regex: r"(?i)error".to_string(), + label_params: Vec::new(), + value_param: None, + matched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "error".to_string())])), + unmatched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "other".to_string())])), + legacy_params: Vec::new(), + } } } -fn seeded_default_dashboard() -> Dashboard { - let mut dashboard = Dashboard::new("Overview".to_string()); - dashboard.panels = vec![ - Panel { - id: generate_id("error-volume-line"), - title: "Error vs. non-error volume".to_string(), - promql: format!("sum(rate({SEEDED_METRIC_NAME}[1m])) by ({SEEDED_METRIC_LABEL})"), - chart_type: ChartType::Line, - }, - Panel { - id: generate_id("error-volume-pie"), - title: "Error vs. non-error ratio".to_string(), - promql: SEEDED_METRIC_NAME.to_string(), - chart_type: ChartType::Pie, - }, - ]; - dashboard +impl Dashboard { + fn seeded_default() -> Dashboard { + let mut dashboard = Dashboard::new("Overview".to_string()); + dashboard.panels = vec![ + Panel { + id: generate_id("error-volume-line"), + title: "Error vs. non-error volume".to_string(), + promql: format!("sum(rate({SEEDED_METRIC_NAME}[1m])) by ({SEEDED_METRIC_LABEL})"), + chart_type: ChartType::Line, + }, + Panel { + id: generate_id("error-volume-pie"), + title: "Error vs. non-error ratio".to_string(), + promql: SEEDED_METRIC_NAME.to_string(), + chart_type: ChartType::Pie, + }, + ]; + dashboard + } } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -322,56 +328,58 @@ impl Profile { } } -pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) { - for record in &profile.params { - let (respond_to, response) = oneshot::channel(); - let request: DeclareParamRequest = record.into(); - let name = request.name.clone(); - if engine - .tx - .send(EngineMessage::DeclareParam { request, respond_to }) - .await - .is_err() - { - eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); - return; +impl Profile { + pub async fn load_into_engine(&self, engine: &EngineHandle) { + for record in &self.params { + let (respond_to, response) = oneshot::channel(); + let request: DeclareParamRequest = record.into(); + let name = request.name.clone(); + if engine + .tx + .send(EngineMessage::DeclareParam { request, respond_to }) + .await + .is_err() + { + eprintln!("warning: engine unavailable while loading profile '{}'", self.name); + return; + } + match response.await { + Ok(Ok(())) => {} + Ok(Err(err)) => eprintln!( + "warning: profile '{}' param declaration '{name}' rejected by engine: {err}", + self.name + ), + Err(_) => eprintln!( + "warning: engine dropped response while declaring param '{name}' from profile '{}'", + self.name + ), + } } - match response.await { - Ok(Ok(())) => {} - Ok(Err(err)) => eprintln!( - "warning: profile '{}' param declaration '{name}' rejected by engine: {err}", - profile.name - ), - Err(_) => eprintln!( - "warning: engine dropped response while declaring param '{name}' from profile '{}'", - profile.name - ), - } - } - for record in &profile.metrics { - let (respond_to, response) = oneshot::channel(); - let request: DeclareMetricRequest = record.into(); - let name = request.name.clone(); - if engine - .tx - .send(EngineMessage::DeclareMetric { request, respond_to }) - .await - .is_err() - { - eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); - return; - } - match response.await { - Ok(Ok(())) => {} - Ok(Err(err)) => eprintln!( - "warning: profile '{}' metric declaration '{name}' rejected by engine: {err}", - profile.name - ), - Err(_) => eprintln!( - "warning: engine dropped response while declaring metric '{name}' from profile '{}'", - profile.name - ), + for record in &self.metrics { + let (respond_to, response) = oneshot::channel(); + let request: DeclareMetricRequest = record.into(); + let name = request.name.clone(); + if engine + .tx + .send(EngineMessage::DeclareMetric { request, respond_to }) + .await + .is_err() + { + eprintln!("warning: engine unavailable while loading profile '{}'", self.name); + return; + } + match response.await { + Ok(Ok(())) => {} + Ok(Err(err)) => eprintln!( + "warning: profile '{}' metric declaration '{name}' rejected by engine: {err}", + self.name + ), + Err(_) => eprintln!( + "warning: engine dropped response while declaring metric '{name}' from profile '{}'", + self.name + ), + } } } } @@ -403,16 +411,18 @@ impl From for ProfileError { } } -pub(crate) fn load_legacy_profile_store(path: &Path) -> Option> { - let contents = fs::read_to_string(path).ok()?; - match serde_json::from_str::>(&contents) { - Ok(profiles) => Some(profiles), - Err(err) => { - eprintln!( - "warning: failed to parse legacy profile store at {}: {err}; skipping migration", - path.display() - ); - None +impl Profile { + pub(crate) fn load_legacy_store(path: &Path) -> Option> { + let contents = fs::read_to_string(path).ok()?; + match serde_json::from_str::>(&contents) { + Ok(profiles) => Some(profiles), + Err(err) => { + eprintln!( + "warning: failed to parse legacy profile store at {}: {err}; skipping migration", + path.display() + ); + None + } } } } @@ -420,8 +430,8 @@ pub(crate) fn load_legacy_profile_store(path: &Path) -> Option> { pub(crate) fn migrate_legacy_profile_shapes(profiles: &mut [Profile]) -> bool { let mut any_migrated = false; for profile in profiles { - let legacy_migrated = migrate_legacy_metric_params(profile); - let duration_migrated = migrate_duration_value_params(profile); + let legacy_migrated = profile.migrate_legacy_metric_params(); + let duration_migrated = profile.migrate_duration_value_params(); if legacy_migrated || duration_migrated { any_migrated = true; } @@ -465,8 +475,8 @@ impl Settings { pub fn ensure_default_profile(&mut self, settings_path: &Path) -> Result<(), ProfileError> { if self.profile(DEFAULT_PROFILE_NAME).is_none() { let mut profile = Profile::new(DEFAULT_PROFILE_NAME.to_string()); - profile.metrics.push(seeded_error_split_metric()); - profile.dashboards.push(seeded_default_dashboard()); + profile.metrics.push(MetricDeclarationRecord::seeded_error_split()); + profile.dashboards.push(Dashboard::seeded_default()); self.profiles.push(profile); } if self.active_profile.is_none() { @@ -798,7 +808,7 @@ mod tests { let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); let engine_tx = engine::spawn(settings_handle, 128); - load_into_engine(&engine_tx, &default_profile).await; + default_profile.load_into_engine(&engine_tx).await; for (sequence, text) in [(0u64, "all good"), (1, "an ERROR happened"), (2, "fine again")] { engine_tx @@ -981,7 +991,7 @@ mod tests { let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); let engine_tx = crate::engine::spawn(settings_handle, 16); - load_into_engine(&engine_tx, &profile).await; + profile.load_into_engine(&engine_tx).await; engine_tx .tx @@ -1133,7 +1143,7 @@ mod tests { let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); tokio::runtime::Runtime::new().unwrap().block_on(async { let engine_tx = crate::engine::spawn(settings_handle, 16); - load_into_engine(&engine_tx, &profile).await; + profile.load_into_engine(&engine_tx).await; engine_tx .tx .send(EngineMessage::Ingest(IngestedLine { @@ -1334,7 +1344,7 @@ mod tests { let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); - load_into_engine(&fresh_engine_tx, &profile).await; + profile.load_into_engine(&fresh_engine_tx).await; let EngineResponse::MetricDeclarations(metrics) = ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await diff --git a/src/promql.rs b/src/promql.rs index 227ef33..6091619 100644 --- a/src/promql.rs +++ b/src/promql.rs @@ -32,22 +32,24 @@ enum RangeSelector { BareWidth(f64), } -pub fn parse_query(input: &str) -> Result { - let mut parser = Parser::new(input); - let query = parser.parse_query()?; - parser.skip_whitespace(); - if !parser.is_at_end() { - return Err(format!("unexpected trailing input: '{}'", parser.remaining())); +impl PromqlQuery { + pub fn parse(input: &str) -> Result { + let mut parser = Parser::new(input); + let query = parser.parse_query()?; + parser.skip_whitespace(); + if !parser.is_at_end() { + return Err(format!("unexpected trailing input: '{}'", parser.remaining())); + } + if query.agg == Some(AggOp::Rate) && query.range.is_none() { + return Err("rate() requires a range selector, e.g. rate(my_metric[5m])".to_string()); + } + if query.agg == Some(AggOp::Histogram) && query.bucket_width.is_none() { + return Err( + "histogram() requires a bucket width, e.g. histogram(my_metric[10])".to_string(), + ); + } + Ok(query) } - if query.agg == Some(AggOp::Rate) && query.range.is_none() { - return Err("rate() requires a range selector, e.g. rate(my_metric[5m])".to_string()); - } - if query.agg == Some(AggOp::Histogram) && query.bucket_width.is_none() { - return Err( - "histogram() requires a bucket width, e.g. histogram(my_metric[10])".to_string(), - ); - } - Ok(query) } struct Parser<'a> { @@ -328,7 +330,7 @@ mod tests { #[test] fn parses_bare_metric_name() { - let query = parse_query("requests").unwrap(); + let query = PromqlQuery::parse("requests").unwrap(); assert_eq!(query.agg, None); assert_eq!(query.metric_name, "requests"); assert!(query.matchers.is_empty()); @@ -337,7 +339,7 @@ mod tests { #[test] fn parses_metric_with_single_label_matcher() { - let query = parse_query(r#"requests{path="/a"}"#).unwrap(); + let query = PromqlQuery::parse(r#"requests{path="/a"}"#).unwrap(); assert_eq!(query.metric_name, "requests"); assert_eq!( query.matchers, @@ -347,7 +349,7 @@ mod tests { #[test] fn parses_metric_with_multiple_label_matchers() { - let query = parse_query(r#"requests{path="/a", method="GET"}"#).unwrap(); + let query = PromqlQuery::parse(r#"requests{path="/a", method="GET"}"#).unwrap(); assert_eq!(query.matchers.len(), 2); assert_eq!(query.matchers[0].name, "path"); assert_eq!(query.matchers[1].name, "method"); @@ -355,67 +357,67 @@ mod tests { #[test] fn parses_range_selector() { - let query = parse_query("requests[5m]").unwrap(); + let query = PromqlQuery::parse("requests[5m]").unwrap(); assert_eq!(query.range, Some(Duration::from_secs(300))); } #[test] fn parses_label_matchers_and_range_together() { - let query = parse_query(r#"requests{path="/a"}[1h]"#).unwrap(); + let query = PromqlQuery::parse(r#"requests{path="/a"}[1h]"#).unwrap(); assert_eq!(query.matchers.len(), 1); assert_eq!(query.range, Some(Duration::from_secs(3600))); } #[test] fn parses_sum_aggregation() { - let query = parse_query(r#"sum(requests{path="/a"})"#).unwrap(); + let query = PromqlQuery::parse(r#"sum(requests{path="/a"})"#).unwrap(); assert_eq!(query.agg, Some(AggOp::Sum)); assert_eq!(query.metric_name, "requests"); } #[test] fn parses_avg_aggregation() { - let query = parse_query("avg(requests)").unwrap(); + let query = PromqlQuery::parse("avg(requests)").unwrap(); assert_eq!(query.agg, Some(AggOp::Avg)); } #[test] fn parses_rate_with_range() { - let query = parse_query("rate(requests[5m])").unwrap(); + let query = PromqlQuery::parse("rate(requests[5m])").unwrap(); assert_eq!(query.agg, Some(AggOp::Rate)); assert_eq!(query.range, Some(Duration::from_secs(300))); } #[test] fn rejects_rate_without_range() { - let err = parse_query("rate(requests)").unwrap_err(); + let err = PromqlQuery::parse("rate(requests)").unwrap_err(); assert!(err.contains("range selector")); } #[test] fn rejects_empty_input() { - assert!(parse_query("").is_err()); + assert!(PromqlQuery::parse("").is_err()); } #[test] fn rejects_unclosed_brace() { - assert!(parse_query(r#"requests{path="/a""#).is_err()); + assert!(PromqlQuery::parse(r#"requests{path="/a""#).is_err()); } #[test] fn rejects_unclosed_bracket() { - assert!(parse_query("requests[5m").is_err()); + assert!(PromqlQuery::parse("requests[5m").is_err()); } #[test] fn rejects_unknown_aggregation() { - let err = parse_query("median(requests)").unwrap_err(); + let err = PromqlQuery::parse("median(requests)").unwrap_err(); assert!(err.contains("unexpected trailing input")); } #[test] fn parses_count_aggregation() { - let query = parse_query("count(requests)").unwrap(); + let query = PromqlQuery::parse("count(requests)").unwrap(); assert_eq!(query.agg, Some(AggOp::Count)); assert_eq!(query.metric_name, "requests"); assert!(!query.inner_rate); @@ -423,37 +425,37 @@ mod tests { #[test] fn parses_count_with_by_clause() { - let query = parse_query("count(requests) by (path)").unwrap(); + let query = PromqlQuery::parse("count(requests) by (path)").unwrap(); assert_eq!(query.agg, Some(AggOp::Count)); assert_eq!(query.by_labels, vec!["path".to_string()]); } #[test] fn rejects_regex_label_matcher() { - let err = parse_query(r#"requests{path=~"/a"}"#).unwrap_err(); + let err = PromqlQuery::parse(r#"requests{path=~"/a"}"#).unwrap_err(); assert!(err.contains("regex label matchers")); } #[test] fn rejects_missing_metric_name() { - assert!(parse_query(r#"{path="/a"}"#).is_err()); + assert!(PromqlQuery::parse(r#"{path="/a"}"#).is_err()); } #[test] fn rejects_unknown_duration_unit() { - let err = parse_query("requests[5x]").unwrap_err(); + let err = PromqlQuery::parse("requests[5x]").unwrap_err(); assert!(err.contains("duration unit")); } #[test] fn rejects_trailing_garbage() { - let err = parse_query("requests extra").unwrap_err(); + let err = PromqlQuery::parse("requests extra").unwrap_err(); assert!(err.contains("unexpected trailing input")); } #[test] fn parses_histogram_with_bare_width() { - let query = parse_query("histogram(latency_ms[10])").unwrap(); + let query = PromqlQuery::parse("histogram(latency_ms[10])").unwrap(); assert_eq!(query.agg, Some(AggOp::Histogram)); assert_eq!(query.metric_name, "latency_ms"); assert_eq!(query.bucket_width, Some(10.0)); @@ -462,27 +464,27 @@ mod tests { #[test] fn rejects_histogram_without_width() { - let err = parse_query("histogram(latency_ms)").unwrap_err(); + let err = PromqlQuery::parse("histogram(latency_ms)").unwrap_err(); assert!(err.contains("bucket width")); } #[test] fn parses_gauge_with_no_range_selector() { - let query = parse_query(r#"gauge(connections{host="a"})"#).unwrap(); + let query = PromqlQuery::parse(r#"gauge(connections{host="a"})"#).unwrap(); assert_eq!(query.agg, Some(AggOp::Gauge)); assert_eq!(query.matchers, vec![LabelMatcher { name: "host".to_string(), value: "a".to_string() }]); } #[test] fn rate_range_selector_is_still_a_duration_not_a_bare_width() { - let query = parse_query("rate(requests[5m])").unwrap(); + let query = PromqlQuery::parse("rate(requests[5m])").unwrap(); assert_eq!(query.range, Some(Duration::from_secs(300))); assert_eq!(query.bucket_width, None); } #[test] fn parses_sum_of_rate() { - let query = parse_query("sum(rate(requests[1m]))").unwrap(); + let query = PromqlQuery::parse("sum(rate(requests[1m]))").unwrap(); assert_eq!(query.agg, Some(AggOp::Sum)); assert!(query.inner_rate); assert_eq!(query.metric_name, "requests"); @@ -492,7 +494,7 @@ mod tests { #[test] fn parses_avg_of_rate_with_label_matchers() { - let query = parse_query(r#"avg(rate(requests{path="/a"}[5m]))"#).unwrap(); + let query = PromqlQuery::parse(r#"avg(rate(requests{path="/a"}[5m]))"#).unwrap(); assert_eq!(query.agg, Some(AggOp::Avg)); assert!(query.inner_rate); assert_eq!(query.matchers, vec![LabelMatcher { name: "path".to_string(), value: "/a".to_string() }]); @@ -500,13 +502,13 @@ mod tests { #[test] fn rejects_sum_of_rate_without_range() { - let err = parse_query("sum(rate(requests))").unwrap_err(); + let err = PromqlQuery::parse("sum(rate(requests))").unwrap_err(); assert!(err.contains("range selector")); } #[test] fn parses_sum_with_by_clause() { - let query = parse_query("sum(requests) by (path)").unwrap(); + let query = PromqlQuery::parse("sum(requests) by (path)").unwrap(); assert_eq!(query.agg, Some(AggOp::Sum)); assert!(!query.inner_rate); assert_eq!(query.by_labels, vec!["path".to_string()]); @@ -514,44 +516,44 @@ mod tests { #[test] fn parses_sum_of_rate_with_by_clause_multiple_labels() { - let query = parse_query("sum(rate(requests[1m])) by (path, method)").unwrap(); + let query = PromqlQuery::parse("sum(rate(requests[1m])) by (path, method)").unwrap(); assert!(query.inner_rate); assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]); } #[test] fn sum_without_by_clause_has_empty_by_labels() { - let query = parse_query("sum(requests)").unwrap(); + let query = PromqlQuery::parse("sum(requests)").unwrap(); assert!(query.by_labels.is_empty()); } #[test] fn rejects_empty_by_clause() { - let err = parse_query("sum(requests) by ()").unwrap_err(); + let err = PromqlQuery::parse("sum(requests) by ()").unwrap_err(); assert!(err.contains("at least one label")); } #[test] fn rejects_unterminated_by_clause() { - assert!(parse_query("sum(requests) by (path").is_err()); + assert!(PromqlQuery::parse("sum(requests) by (path").is_err()); } #[test] fn parses_sum_with_prefix_by_clause() { - let query = parse_query("sum by (path) (requests)").unwrap(); + let query = PromqlQuery::parse("sum by (path) (requests)").unwrap(); assert_eq!(query.agg, Some(AggOp::Sum)); assert_eq!(query.by_labels, vec!["path".to_string()]); } #[test] fn parses_sum_with_prefix_by_clause_multiple_labels() { - let query = parse_query("sum by (path, method) (requests)").unwrap(); + let query = PromqlQuery::parse("sum by (path, method) (requests)").unwrap(); assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]); } #[test] fn rejects_by_clause_both_before_and_after_aggregation() { - let err = parse_query("sum by (path) (requests) by (method)").unwrap_err(); + let err = PromqlQuery::parse("sum by (path) (requests) by (method)").unwrap_err(); assert!(err.contains("only appear once")); } } diff --git a/src/settings.rs b/src/settings.rs index 1828919..42e37b7 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -197,7 +197,7 @@ impl Settings { .map(|parent| parent.join("profiles")) .unwrap_or_else(|| PathBuf::from("profiles")) .join("profiles.json"); - let Some(profiles) = crate::profile::load_legacy_profile_store(&legacy_path) else { + let Some(profiles) = crate::profile::Profile::load_legacy_store(&legacy_path) else { return; }; self.profiles = profiles;