phughk/ottyl
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Convert remaining clean-receiver free functions to methods

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS
This commit is contained in:
Hugh Kaznowski
2026-09-06 17:08:29 +01:00
co-authored by Claude Sonnet 5
parent 529f79e9da
commit 3c2995acc1
7 changed files with 449 additions and 426 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api.rs
+60 -51
View File
@@ -14,7 +14,7 @@ use crate::engine::{
EngineResponse, EngineResponse,
}; };
use crate::alerts::{deliver_ntfy, AlertError}; 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 crate::settings::{CssError, SettingsHandle};
use axum::extract::{Path as AxumPath, Query, State}; use axum::extract::{Path as AxumPath, Query, State};
use axum::http::{header, StatusCode}; use axum::http::{header, StatusCode};
@@ -215,8 +215,9 @@ fn mime_for_extension(extension: &str) -> &'static str {
} }
} }
async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse, StatusCode> { impl EngineHandle {
let read_handle = engine.reader_factory.handle(); async fn ask(&self, query: EngineQuery) -> Result<EngineResponse, StatusCode> {
let read_handle = self.reader_factory.handle();
if let Some(sort_by) = query.sort_by() { if let Some(sort_by) = query.sort_by() {
let needs_build = match read_handle.enter() { let needs_build = match read_handle.enter() {
Some(guard) => !guard.has_sort_helper_index(sort_by), Some(guard) => !guard.has_sort_helper_index(sort_by),
@@ -224,8 +225,7 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
}; };
if needs_build { if needs_build {
let (respond_to, response) = oneshot::channel(); let (respond_to, response) = oneshot::channel();
engine self.tx
.tx
.send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to }) .send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to })
.await .await
.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?;
@@ -234,6 +234,7 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
} }
let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?;
Ok(guard.handle_query_ro(query)) Ok(guard.handle_query_ro(query))
}
} }
async fn promoted_param_records( async fn promoted_param_records(
@@ -252,7 +253,7 @@ async fn promoted_param_records(
return Vec::new(); return Vec::new();
} }
let Ok(EngineResponse::ParamDeclarations(declarations)) = let Ok(EngineResponse::ParamDeclarations(declarations)) =
ask(engine, EngineQuery::ParamDeclarations).await engine.ask(EngineQuery::ParamDeclarations).await
else { else {
return Vec::new(); return Vec::new();
}; };
@@ -274,7 +275,7 @@ async fn get_logs(
State(state): State<ApiState>, State(state): State<ApiState>,
Query(params): Query<LogsParams>, Query(params): Query<LogsParams>,
) -> impl IntoResponse { ) -> 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::Logs(entries)) => Json(entries).into_response(),
Ok(EngineResponse::Error(err)) => { Ok(EngineResponse::Error(err)) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
@@ -288,7 +289,7 @@ async fn get_timeline(
State(state): State<ApiState>, State(state): State<ApiState>,
Query(params): Query<TimelineParams>, Query(params): Query<TimelineParams>,
) -> impl IntoResponse { ) -> 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::Timeline(buckets)) => Json(buckets).into_response(),
Ok(EngineResponse::Error(err)) => { Ok(EngineResponse::Error(err)) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
@@ -299,7 +300,7 @@ async fn get_timeline(
} }
async fn get_stats(State(state): State<ApiState>) -> impl IntoResponse { async fn get_stats(State(state): State<ApiState>) -> 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(EngineResponse::Stats(stats)) => Json(stats).into_response(),
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(), Err(status) => status.into_response(),
@@ -307,9 +308,9 @@ async fn get_stats(State(state): State<ApiState>) -> impl IntoResponse {
} }
async fn list_params(State(state): State<ApiState>) -> impl IntoResponse { async fn list_params(State(state): State<ApiState>) -> impl IntoResponse {
match ask(&state.engine, EngineQuery::ParamDeclarations).await { match state.engine.ask(EngineQuery::ParamDeclarations).await {
Ok(EngineResponse::ParamDeclarations(mut declarations)) => { Ok(EngineResponse::ParamDeclarations(mut declarations)) => {
declarations.extend(reserved_param_declaration_dtos()); declarations.extend(ParamDeclarationDto::reserved());
Json(declarations).into_response() Json(declarations).into_response()
} }
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
@@ -317,7 +318,8 @@ async fn list_params(State(state): State<ApiState>) -> impl IntoResponse {
} }
} }
fn reserved_param_declaration_dtos() -> [ParamDeclarationDto; 3] { impl ParamDeclarationDto {
fn reserved() -> [ParamDeclarationDto; 3] {
[ [
ParamDeclarationDto { ParamDeclarationDto {
name: "ottyl_seq".to_string(), name: "ottyl_seq".to_string(),
@@ -350,6 +352,7 @@ fn reserved_param_declaration_dtos() -> [ParamDeclarationDto; 3] {
reserved: true, reserved: true,
}, },
] ]
}
} }
async fn declare_param( async fn declare_param(
@@ -402,7 +405,7 @@ async fn clear_params(State(state): State<ApiState>) -> impl IntoResponse {
} }
async fn list_metrics(State(state): State<ApiState>) -> impl IntoResponse { async fn list_metrics(State(state): State<ApiState>) -> 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(EngineResponse::MetricDeclarations(declarations)) => Json(declarations).into_response(),
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(), Err(status) => status.into_response(),
@@ -521,7 +524,7 @@ async fn get_metric_bucket_values(
since_sequence: params.since_sequence, since_sequence: params.since_sequence,
limit: params.limit, limit: params.limit,
}; };
match ask(&state.engine, query).await { match state.engine.ask(query).await {
Ok(EngineResponse::BucketValues(values)) => Json(values).into_response(), Ok(EngineResponse::BucketValues(values)) => Json(values).into_response(),
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(), Err(status) => status.into_response(),
@@ -537,7 +540,7 @@ async fn run_promql_query(
start_ms: params.start_ms, start_ms: params.start_ms,
end_ms: params.end_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(Ok(series))) => Json(series).into_response(),
Ok(EngineResponse::Promql(Err(err))) => { Ok(EngineResponse::Promql(Err(err))) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() (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 { impl CssError {
match err { fn into_response(self) -> axum::response::Response {
CssError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), match self {
CssError::NotImported => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), CssError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(),
CssError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.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"); .expect("entry was just inserted by import_css");
(StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response() (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"); .expect("entry was just pushed by link_css");
(StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response() (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(); let mut settings = state.settings.lock().unwrap();
match settings.remove_css(&state.settings_path, &css_dir, &id) { match settings.remove_css(&state.settings_path, &css_dir, &id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(), 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, content,
) )
.into_response(), .into_response(),
Err(err) => css_error_response(err), Err(err) => err.into_response(),
} }
} }
fn alert_error_response(err: AlertError) -> axum::response::Response { impl AlertError {
match err { fn into_response(self) -> axum::response::Response {
AlertError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), match self {
AlertError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(), 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(); let mut settings = state.settings.lock().unwrap();
match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition) { match settings.create_alert_rule(&state.settings_path, body.name, body.enabled, body.condition) {
Ok(rule) => (StatusCode::CREATED, Json(rule)).into_response(), 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(); let mut settings = state.settings.lock().unwrap();
match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition) { match settings.update_alert_rule(&state.settings_path, &id, body.name, body.enabled, body.condition) {
Ok(rule) => Json(rule).into_response(), 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<ApiState>, AxumPath(id): AxumPath
let mut settings = state.settings.lock().unwrap(); let mut settings = state.settings.lock().unwrap();
match settings.remove_alert_rule(&state.settings_path, &id) { match settings.remove_alert_rule(&state.settings_path, &id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(), 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() Json(new_max_logs.0).into_response()
} }
fn profile_error_response(err: ProfileError) -> axum::response::Response { impl ProfileError {
match err { fn into_response(self) -> axum::response::Response {
ProfileError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(), match self {
ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), ProfileError::NotFound => (StatusCode::NOT_FOUND, self.to_string()).into_response(),
ProfileError::CannotDeleteDefault => (StatusCode::BAD_REQUEST, err.to_string()).into_response(), ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, self.to_string()).into_response(),
ProfileError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.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"); let profile = settings.profile(&body.name).expect("profile was just created by create_profile");
(StatusCode::CREATED, Json(ProfileSummaryDto::from(profile))).into_response() (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(); let mut settings = state.settings.lock().unwrap();
match settings.delete_profile(&state.settings_path, &name) { match settings.delete_profile(&state.settings_path, &name) {
Ok(()) => StatusCode::NO_CONTENT.into_response(), 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) { async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) {
if let Ok(EngineResponse::ParamDeclarations(params)) = 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) { for param in params.into_iter().filter(|param| !param.auto_derived) {
let (respond_to, response) = oneshot::channel(); 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)) = if let Ok(EngineResponse::MetricDeclarations(metrics)) =
ask(engine, EngineQuery::MetricDeclarations).await engine.ask(EngineQuery::MetricDeclarations).await
{ {
for metric in metrics { for metric in metrics {
let (respond_to, response) = oneshot::channel(); 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( async fn activate_profile(
@@ -815,7 +824,7 @@ async fn activate_profile(
let settings = state.settings.lock().unwrap(); let settings = state.settings.lock().unwrap();
Json(SettingsDto::from(&*settings)).into_response() 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<DashboardDto> = dashboards.iter().map(DashboardDto::from).collect(); let dtos: Vec<DashboardDto> = dashboards.iter().map(DashboardDto::from).collect();
Json(dtos).into_response() 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(); let mut settings = state.settings.lock().unwrap();
match settings.create_dashboard(&state.settings_path, &name, body.name) { match settings.create_dashboard(&state.settings_path, &name, body.name) {
Ok(dashboard) => (StatusCode::CREATED, Json(DashboardDto::from(&dashboard))).into_response(), 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) { match settings.list_dashboards(&name) {
Ok(dashboards) => match dashboards.iter().find(|d| d.id == dashboard_id) { Ok(dashboards) => match dashboards.iter().find(|d| d.id == dashboard_id) {
Some(dashboard) => Json(DashboardDto::from(dashboard)).into_response(), 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) { match settings.save_dashboard(&state.settings_path, &name, &dashboard_id, dashboard) {
Ok(dashboard) => Json(DashboardDto::from(&dashboard)).into_response(), 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(); let mut settings = state.settings.lock().unwrap();
match settings.delete_dashboard(&state.settings_path, &name, &dashboard_id) { match settings.delete_dashboard(&state.settings_path, &name, &dashboard_id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(), 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_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); 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 { let EngineResponse::MetricDeclarations(metrics) = response else {
panic!("unexpected response"); panic!("unexpected response");
}; };
src/engine.rs
+45 -45
View File
@@ -13,7 +13,7 @@ use crate::filter_datetime;
use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue};
use crate::index_key::IndexKey; use crate::index_key::IndexKey;
use crate::metric_primitive::{MetricPrimitive, Number}; use crate::metric_primitive::{MetricPrimitive, Number};
use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; use crate::promql::{AggOp, LabelMatcher, PromqlQuery};
use crate::rate; use crate::rate;
use crate::settings::SettingsHandle; use crate::settings::SettingsHandle;
use crate::timestamp::Timestamp; use crate::timestamp::Timestamp;
@@ -208,7 +208,7 @@ async fn run(
let ingest_lag_high_watermark_ms = let ingest_lag_high_watermark_ms =
ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at); 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); log_stream_header = log_stream_header_transition(&log_stream_header, &text);
let row_header = if is_new_header { None } else { log_stream_header.clone() }; let row_header = if is_new_header { None } else { log_stream_header.clone() };
@@ -527,7 +527,7 @@ fn log_stream_header_transition(
current: &Option<crate::logfmt::LogStreamHeader>, current: &Option<crate::logfmt::LogStreamHeader>,
text: &str, text: &str,
) -> Option<crate::logfmt::LogStreamHeader> { ) -> Option<crate::logfmt::LogStreamHeader> {
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( 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) { if !crate::logfmt::is_log_stream_line(text) {
return; 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() { for (i, column) in header.columns.iter().enumerate() {
let Some(value) = row.get(i) else { continue }; let Some(value) = row.get(i) else { continue };
@@ -550,7 +550,7 @@ fn derive_log_stream_params_from_row(
index_key, index_key,
DataValue::ParamDeclaration(ParamDeclaration { DataValue::ParamDeclaration(ParamDeclaration {
name: column.name.clone(), 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(), capture_group: column.name.clone(),
json_field_path: None, json_field_path: None,
kind: infer_kind(value.trim()), 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(|| { index.entry(IndexKey::ParamDeclaration { name: "message".to_string() }).or_insert_with(|| {
DataValue::ParamDeclaration(ParamDeclaration { DataValue::ParamDeclaration(ParamDeclaration {
name: "message".to_string(), 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(), capture_group: "message".to_string(),
json_field_path: None, json_field_path: None,
kind: infer_kind(message.trim()), kind: infer_kind(message.trim()),
@@ -583,7 +583,7 @@ fn auto_declare_log_stream_params(
header: &mut Option<crate::logfmt::LogStreamHeader>, header: &mut Option<crate::logfmt::LogStreamHeader>,
text: &str, 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); *header = log_stream_header_transition(header, text);
if is_new_header { if is_new_header {
return; return;
@@ -1802,7 +1802,7 @@ fn handle_query_inner(
EngineResponse::BucketValues(values) EngineResponse::BucketValues(values)
} }
EngineQuery::Promql { query, start_ms, end_ms } => { 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)); .and_then(|parsed| snapshot.evaluate_promql(&parsed, start_ms, end_ms));
EngineResponse::Promql(response) EngineResponse::Promql(response)
} }
@@ -2797,7 +2797,7 @@ mod tests {
declare_requests_metric(&mut index); declare_requests_metric(&mut index);
ingest_log(&mut index, 0, "request path=/a"); 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.evaluate_promql(&parsed, None, None).unwrap().len() == 1);
assert!(index.remove_metric_declaration("requests")); assert!(index.remove_metric_declaration("requests"));
@@ -2816,7 +2816,7 @@ mod tests {
Timestamp::from_millis(0), Timestamp::from_millis(0),
MetricPrimitive::Number(Number::Float(1.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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
} }
@@ -2824,7 +2824,7 @@ mod tests {
#[test] #[test]
fn promql_rejects_metric_with_no_declaration_and_no_data() { fn promql_rejects_metric_with_no_declaration_and_no_data() {
let index = EngineSnapshot::default(); 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(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err();
assert!(err.contains("unknown metric"), "unexpected error message: {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, 1, "request path=/b");
ingest_log(&mut index, 2, "request path=/a"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3067,7 +3067,7 @@ mod tests {
declare_requests_metric(&mut index); declare_requests_metric(&mut index);
ingest_log(&mut index, 0, "request path=/a"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert!(series.is_empty()); 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=/a");
ingest_log(&mut index, 0, "request path=/b"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3095,7 +3095,7 @@ mod tests {
ingest_log(&mut index, 0, "request path=/b"); ingest_log(&mut index, 0, "request path=/b");
ingest_log(&mut index, 1, "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 mut series = index.evaluate_promql(&parsed, None, None).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"))); 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); declare_string_valued_metric(&mut index);
ingest_log(&mut index, 0, "request status=ok"); 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(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err();
assert!(err.contains("statuses"), "error should name the offending metric: {err}"); assert!(err.contains("statuses"), "error should name the offending metric: {err}");
@@ -3167,7 +3167,7 @@ mod tests {
declare_string_valued_metric(&mut index); declare_string_valued_metric(&mut index);
ingest_log(&mut index, 0, "request status=ok"); 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(); let err = index.evaluate_promql(&parsed, None, None).unwrap_err();
assert!(err.contains("statuses")); assert!(err.contains("statuses"));
@@ -3180,7 +3180,7 @@ mod tests {
ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 0, "request status=ok");
ingest_log(&mut index, 1, "request status=error"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); 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=/a");
ingest_log(&mut index, 0, "request path=/b"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3209,7 +3209,7 @@ mod tests {
ingest_log(&mut index, 0, "request status=ok"); ingest_log(&mut index, 0, "request status=ok");
ingest_log(&mut index, 1000, "request status=error"); 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); 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:?}"); 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); declare_string_valued_metric(&mut index);
ingest_log(&mut index, 0, "request status=ok"); 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); let series = index.evaluate_promql(&parsed, None, None);
assert!(series.is_ok()); assert!(series.is_ok());
@@ -3235,7 +3235,7 @@ mod tests {
ingest_log(&mut index, 1, "request path=/a host=web2"); ingest_log(&mut index, 1, "request path=/a host=web2");
ingest_log(&mut index, 2, "request path=/a host=web3"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); 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, 0, "request path=/a host=web1");
ingest_log(&mut index, 1, "request path=/a host=web2"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3271,7 +3271,7 @@ mod tests {
ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/a");
ingest_log(&mut index, 1, "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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3285,7 +3285,7 @@ mod tests {
ingest_log(&mut index, 0, "request path=/a"); ingest_log(&mut index, 0, "request path=/a");
ingest_log(&mut index, 1000, "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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3300,7 +3300,7 @@ mod tests {
ingest_log(&mut index, 1000, "request path=/a"); ingest_log(&mut index, 1000, "request path=/a");
ingest_log(&mut index, 0, "request path=/b"); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); 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, 0, "request path=/a");
ingest_log(&mut index, 1, "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(); let series = index.evaluate_promql(&parsed, None, Some(100_000)).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3335,7 +3335,7 @@ mod tests {
declare_requests_metric(&mut index); declare_requests_metric(&mut index);
ingest_log(&mut index, 50_000, "request path=/a"); 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(); let series = index.evaluate_promql(&parsed, Some(0), None).unwrap();
assert_eq!(series[0].points, vec![(0, 0.0), (50_000, 0.5)]); 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, 1, "request path=/a");
ingest_log(&mut index, 0, "request path=/b"); 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(); 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"))); 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, 0, "request path=/a");
ingest_log(&mut index, 1000, "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(); let series = index.evaluate_promql(&parsed, Some(0), Some(1000)).unwrap();
assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]);
@@ -3387,7 +3387,7 @@ mod tests {
ingest_log(&mut index, *ms, text); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
let (_, dashboard_rate) = *series[0].points.last().unwrap(); 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=/a");
ingest_log(&mut index, 0, "request path=/b"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3447,7 +3447,7 @@ mod tests {
ingest_log(&mut index, 5, "request path=/a"); ingest_log(&mut index, 5, "request path=/a");
ingest_log(&mut index, 10, "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(); let series = index.evaluate_promql(&parsed, Some(1), Some(9)).unwrap();
assert_eq!(series.len(), 1); 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, 1, "request path=/a duration=9");
ingest_log(&mut index, 2, "request path=/b duration=42"); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); 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, 5, "request path=/a duration=9");
ingest_log(&mut index, 10, "request path=/a duration=42"); 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(); let series = index.evaluate_promql(&parsed, None, Some(9)).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3537,7 +3537,7 @@ mod tests {
ingest_log(&mut index, seq, &format!("request path=/a duration={duration}")); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le"))); 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}")); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
series.sort_by(|a, b| { series.sort_by(|a, b| {
a.labels.get("path").cmp(&b.labels.get("path")).then(a.labels.get("le").cmp(&b.labels.get("le"))) 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, 0, "user=alice logged in");
ingest_log(&mut index, 10, "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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
@@ -3678,7 +3678,7 @@ mod tests {
ingest_log(&mut index, 1, "all good"); ingest_log(&mut index, 1, "all good");
ingest_log(&mut index, 2, "all good again"); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category"))); 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, 1, "line two");
ingest_log(&mut index, 2, "line three"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0), (2, 1.0)]); 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, 0, "line zero");
ingest_log(&mut index, 5, "line five"); 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(); let mut series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
series[0].points.sort_by_key(|(ts, _)| *ts); series[0].points.sort_by_key(|(ts, _)| *ts);
@@ -3775,7 +3775,7 @@ mod tests {
ingest_log(&mut index, 0, "a"); ingest_log(&mut index, 0, "a");
ingest_log(&mut index, 1, "b"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
assert_eq!(series[0].points, vec![(0, 0.0), (1, 1.0)]); assert_eq!(series[0].points, vec![(0, 0.0), (1, 1.0)]);
@@ -3788,7 +3788,7 @@ mod tests {
ingest_log(&mut index, sequence, "line"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert_eq!(series.len(), 1); assert_eq!(series.len(), 1);
assert!(!series[0].points.is_empty(), "rate(ottyl_ts[...]) must not silently return no data"); 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(); let mut index = EngineSnapshot::default();
ingest_log(&mut index, 0, "line"); 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(); let series = index.evaluate_promql(&parsed, None, None).unwrap();
assert!(series.is_empty()); assert!(series.is_empty());
} }
@@ -3809,7 +3809,7 @@ mod tests {
let mut index = EngineSnapshot::default(); let mut index = EngineSnapshot::default();
ingest_log(&mut index, 0, r#"{"level":"error"}"#); 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(); 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("ottyl_fmt"), "error should name the rejected metric: {err}");
assert!(err.contains("categorical") || err.contains("filter"), "error should explain why: {err}"); assert!(err.contains("categorical") || err.contains("filter"), "error should explain why: {err}");
src/logfmt.rs
+37 -35
View File
@@ -138,7 +138,8 @@ fn char_slice(s: &str, start: usize, end: Option<usize>) -> &str {
&s[byte_start..byte_end.max(byte_start)] &s[byte_start..byte_end.max(byte_start)]
} }
pub fn detect_log_stream_header(line: &str) -> Option<LogStreamHeader> { impl LogStreamHeader {
pub fn detect(line: &str) -> Option<LogStreamHeader> {
if is_log_stream_line(line) { if is_log_stream_line(line) {
return None; return None;
} }
@@ -157,15 +158,15 @@ pub fn detect_log_stream_header(line: &str) -> Option<LogStreamHeader> {
} }
let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t')); let tab_delimited = WHITESPACE_RUN.find_iter(line).any(|m| m.as_str().contains('\t'));
Some(LogStreamHeader { columns, tab_delimited }) 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() { pub fn split_row<'a>(&self, line: &'a str) -> Vec<&'a str> {
match header.columns.get(i + 1) { 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()), Some(next) => result.push(char_slice(line, column.start, Some(next.start)).trim()),
None => { None => {
let tail = char_slice(line, column.start, None); let tail = char_slice(line, column.start, None);
@@ -178,10 +179,10 @@ pub fn split_log_stream_row<'a>(line: &'a str, header: &LogStreamHeader) -> Vec<
} }
} }
result result
} }
pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture_group: &str) -> String { pub fn column_pattern(&self, index: usize, capture_group: &str) -> String {
if header.tab_delimited { if self.tab_delimited {
let mut pattern = String::from("^"); let mut pattern = String::from("^");
for _ in 0..index { for _ in 0..index {
pattern.push_str(COLUMN_VALUE_PATTERN); pattern.push_str(COLUMN_VALUE_PATTERN);
@@ -190,26 +191,27 @@ pub fn log_stream_column_pattern(header: &LogStreamHeader, index: usize, capture
pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})")); pattern.push_str(&format!("(?P<{capture_group}>{COLUMN_VALUE_PATTERN})"));
return pattern; return pattern;
} }
let start = header.columns[index].start; let start = self.columns[index].start;
let value_pattern = match header.columns.get(index + 1) { let value_pattern = match self.columns.get(index + 1) {
Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)), Some(next) => bounded_column_value_pattern(next.start.saturating_sub(start)),
None => COLUMN_VALUE_PATTERN.to_string(), None => COLUMN_VALUE_PATTERN.to_string(),
}; };
format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})") format!("^.{{{start}}}(?P<{capture_group}>{value_pattern})")
} }
pub fn log_stream_message_pattern(header: &LogStreamHeader, capture_group: &str) -> String { pub fn message_pattern(&self, capture_group: &str) -> String {
if header.tab_delimited { if self.tab_delimited {
let mut pattern = String::from("^"); let mut pattern = String::from("^");
for _ in 0..header.columns.len() { for _ in 0..self.columns.len() {
pattern.push_str(COLUMN_VALUE_PATTERN); pattern.push_str(COLUMN_VALUE_PATTERN);
pattern.push_str(COLUMN_SEPARATOR_PATTERN); pattern.push_str(COLUMN_SEPARATOR_PATTERN);
} }
pattern.push_str(&format!("(?P<{capture_group}>.*)$")); pattern.push_str(&format!("(?P<{capture_group}>.*)$"));
return pattern; return pattern;
} }
let start = header.columns.last().map(|c| c.start).unwrap_or(0); let start = self.columns.last().map(|c| c.start).unwrap_or(0);
format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$") format!("^.{{{start}}}{COLUMN_VALUE_PATTERN}{COLUMN_SEPARATOR_PATTERN}(?P<{capture_group}>.*)$")
}
} }
fn go_log_prefix(input: &str) -> IResult<&str, (&str, &str, &str)> { fn go_log_prefix(input: &str) -> IResult<&str, (&str, &str, &str)> {
@@ -368,7 +370,7 @@ mod tests {
#[test] #[test]
fn detects_log_stream_header_columns_generically() { fn detects_log_stream_header_columns_generically() {
let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); 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_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type", "Activity", "PID", "TTL"]);
assert!(!header.tab_delimited); assert!(!header.tab_delimited);
} }
@@ -376,21 +378,21 @@ mod tests {
#[test] #[test]
fn a_log_stream_data_line_is_never_detected_as_its_own_header() { fn a_log_stream_data_line_is_never_detected_as_its_own_header() {
let data_line = LOG_STREAM_FIXTURE.lines().nth(1).unwrap(); 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] #[test]
fn plain_prose_is_not_detected_as_a_log_stream_header() { fn plain_prose_is_not_detected_as_a_log_stream_header() {
assert!(detect_log_stream_header("just some plain prose here").is_none()); assert!(LogStreamHeader::detect("just some plain prose here").is_none());
assert!(detect_log_stream_header("ts=1788648105773 msg=\"entity_moved\"").is_none()); assert!(LogStreamHeader::detect("ts=1788648105773 msg=\"entity_moved\"").is_none());
} }
#[test] #[test]
fn splits_log_stream_row_into_named_columns_plus_trailing_message() { fn splits_log_stream_row_into_named_columns_plus_trailing_message() {
let header_line = LOG_STREAM_FIXTURE.lines().next().unwrap(); 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 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.len(), header.columns.len() + 1);
assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100"); assert_eq!(row[0], "2026-09-06 01:48:54.518117+0100");
assert_eq!(row[1], "0x4dfed2"); assert_eq!(row[1], "0x4dfed2");
@@ -403,12 +405,12 @@ mod tests {
#[test] #[test]
fn a_single_tab_separates_columns_exactly_like_a_run_of_spaces() { 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_eq!(column_names(&header), vec!["Timestamp", "Thread", "Type"]);
assert!(header.tab_delimited); assert!(header.tab_delimited);
let spaced = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message", &header); let spaced = header.split_row("2026-09-06 01:48:54\t0x4dfed2\t\tDefault trailing message");
let tabbed = split_log_stream_row("2026-09-06 01:48:54\t0x4dfed2\tDefault trailing message", &header); 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[0], "2026-09-06 01:48:54");
assert_eq!(spaced[1], "0x4dfed2"); assert_eq!(spaced[1], "0x4dfed2");
assert_eq!(spaced[2], "Default trailing message"); assert_eq!(spaced[2], "Default trailing message");
@@ -420,22 +422,22 @@ mod tests {
#[test] #[test]
fn column_and_message_patterns_capture_the_same_fields_split_log_stream_row_finds() { 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_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 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"); 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"); 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!( assert_eq!(
message_re.captures(line).unwrap().name("message").unwrap().as_str(), message_re.captures(line).unwrap().name("message").unwrap().as_str(),
"WindowServer: (HSTouchHIDService) something happened" "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!( assert_eq!(
timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(), timestamp_re.captures(line).unwrap().name("Timestamp").unwrap().as_str(),
"2026-09-06 01:48:54.518117+0100" "2026-09-06 01:48:54.518117+0100"
@@ -445,8 +447,8 @@ mod tests {
#[test] #[test]
fn is_log_stream_line_rejects_non_columnar_lines_that_a_fixed_offset_column_regex_would_still_match() { 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_line = LOG_STREAM_FIXTURE.lines().next().unwrap();
let header = detect_log_stream_header(header_line).unwrap(); let header = LogStreamHeader::detect(header_line).unwrap();
let timestamp_re = Regex::new(&log_stream_column_pattern(&header, 0, "Timestamp")).unwrap(); let timestamp_re = Regex::new(&header.column_pattern(0, "Timestamp")).unwrap();
let continuation_line = "so_gencnt: 4042707 so_state: 0x0"; let continuation_line = "so_gencnt: 4042707 so_state: 0x0";
assert!(timestamp_re.captures(continuation_line).is_some(), "sanity: regex alone still matches garbage"); assert!(timestamp_re.captures(continuation_line).is_some(), "sanity: regex alone still matches garbage");
src/main.rs
+1 -1
View File
@@ -110,7 +110,7 @@ async fn main() {
}; };
match (active_profile_name, active_profile) { match (active_profile_name, active_profile) {
(Some(_), Some(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) => { (Some(profile_name), None) => {
eprintln!( eprintln!(
src/profile.rs
+50 -40
View File
@@ -153,40 +153,41 @@ impl From<&MetricDeclarationRecord> for DeclareMetricRequest {
} }
} }
fn migrate_legacy_metric_params(profile: &mut Profile) -> bool { impl Profile {
fn migrate_legacy_metric_params(&mut self) -> bool {
let mut migrated = false; let mut migrated = false;
for metric_index in 0..profile.metrics.len() { for metric_index in 0..self.metrics.len() {
if profile.metrics[metric_index].legacy_params.is_empty() { if self.metrics[metric_index].legacy_params.is_empty() {
continue; continue;
} }
let metric_name = profile.metrics[metric_index].name.clone(); let metric_name = self.metrics[metric_index].name.clone();
let message_match_regex = profile.metrics[metric_index].message_match_regex.clone(); let message_match_regex = self.metrics[metric_index].message_match_regex.clone();
let legacy = std::mem::take(&mut profile.metrics[metric_index].legacy_params); let legacy = std::mem::take(&mut self.metrics[metric_index].legacy_params);
let mut label_params = Vec::new(); let mut label_params = Vec::new();
let mut value_param = None; let mut value_param = None;
for matcher in &legacy { for matcher in &legacy {
let param_name = let param_name =
resolve_or_create_legacy_param(profile, &metric_name, &message_match_regex, matcher); self.resolve_or_create_legacy_param(&metric_name, &message_match_regex, matcher);
match matcher.role { match matcher.role {
LegacyParamRole::Label => label_params.push(param_name), LegacyParamRole::Label => label_params.push(param_name),
LegacyParamRole::Value => value_param = Some(param_name), LegacyParamRole::Value => value_param = Some(param_name),
} }
} }
profile.metrics[metric_index].label_params = label_params; self.metrics[metric_index].label_params = label_params;
profile.metrics[metric_index].value_param = value_param; self.metrics[metric_index].value_param = value_param;
migrated = true; migrated = true;
} }
migrated migrated
} }
fn resolve_or_create_legacy_param( fn resolve_or_create_legacy_param(
profile: &mut Profile, &mut self,
metric_name: &str, metric_name: &str,
message_match_regex: &str, message_match_regex: &str,
matcher: &LegacyMetricParamRecord, matcher: &LegacyMetricParamRecord,
) -> String { ) -> String {
if let Some(existing) = profile.params.iter().find(|p| p.name == matcher.param_name) { if let Some(existing) = self.params.iter().find(|p| p.name == matcher.param_name) {
if existing.message_match_regex == message_match_regex if existing.message_match_regex == message_match_regex
&& existing.capture_group == matcher.capture_group && existing.capture_group == matcher.capture_group
&& existing.json_field_path == matcher.json_field_path && existing.json_field_path == matcher.json_field_path
@@ -197,8 +198,8 @@ fn resolve_or_create_legacy_param(
} }
let implicit_name = format!("{metric_name}.{}", matcher.param_name); let implicit_name = format!("{metric_name}.{}", matcher.param_name);
if !profile.params.iter().any(|p| p.name == implicit_name) { if !self.params.iter().any(|p| p.name == implicit_name) {
profile.params.push(ParamDeclarationRecord { self.params.push(ParamDeclarationRecord {
name: implicit_name.clone(), name: implicit_name.clone(),
message_match_regex: message_match_regex.to_string(), message_match_regex: message_match_regex.to_string(),
capture_group: matcher.capture_group.clone(), capture_group: matcher.capture_group.clone(),
@@ -209,13 +210,13 @@ fn resolve_or_create_legacy_param(
}); });
} }
implicit_name implicit_name
} }
fn migrate_duration_value_params(profile: &mut Profile) -> bool { fn migrate_duration_value_params(&mut self) -> bool {
let value_param_names: std::collections::HashSet<&str> = let value_param_names: std::collections::HashSet<&str> =
profile.metrics.iter().filter_map(|metric| metric.value_param.as_deref()).collect(); self.metrics.iter().filter_map(|metric| metric.value_param.as_deref()).collect();
let mut migrated = false; let mut migrated = false;
for param in &mut profile.params { for param in &mut self.params {
if !param.elapsed_since_last if !param.elapsed_since_last
&& param.kind == ProfileMetricPrimitiveKind::Duration && param.kind == ProfileMetricPrimitiveKind::Duration
&& value_param_names.contains(param.name.as_str()) && value_param_names.contains(param.name.as_str())
@@ -225,6 +226,7 @@ fn migrate_duration_value_params(profile: &mut Profile) -> bool {
} }
} }
migrated migrated
}
} }
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -269,7 +271,8 @@ impl Dashboard {
} }
} }
fn seeded_error_split_metric() -> MetricDeclarationRecord { impl MetricDeclarationRecord {
fn seeded_error_split() -> MetricDeclarationRecord {
MetricDeclarationRecord { MetricDeclarationRecord {
name: SEEDED_METRIC_NAME.to_string(), name: SEEDED_METRIC_NAME.to_string(),
message_match_regex: r"(?i)error".to_string(), message_match_regex: r"(?i)error".to_string(),
@@ -279,9 +282,11 @@ fn seeded_error_split_metric() -> MetricDeclarationRecord {
unmatched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "other".to_string())])), unmatched_labels: Some(BTreeMap::from([(SEEDED_METRIC_LABEL.to_string(), "other".to_string())])),
legacy_params: Vec::new(), legacy_params: Vec::new(),
} }
}
} }
fn seeded_default_dashboard() -> Dashboard { impl Dashboard {
fn seeded_default() -> Dashboard {
let mut dashboard = Dashboard::new("Overview".to_string()); let mut dashboard = Dashboard::new("Overview".to_string());
dashboard.panels = vec![ dashboard.panels = vec![
Panel { Panel {
@@ -298,6 +303,7 @@ fn seeded_default_dashboard() -> Dashboard {
}, },
]; ];
dashboard dashboard
}
} }
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -322,8 +328,9 @@ impl Profile {
} }
} }
pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) { impl Profile {
for record in &profile.params { pub async fn load_into_engine(&self, engine: &EngineHandle) {
for record in &self.params {
let (respond_to, response) = oneshot::channel(); let (respond_to, response) = oneshot::channel();
let request: DeclareParamRequest = record.into(); let request: DeclareParamRequest = record.into();
let name = request.name.clone(); let name = request.name.clone();
@@ -333,23 +340,23 @@ pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) {
.await .await
.is_err() .is_err()
{ {
eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); eprintln!("warning: engine unavailable while loading profile '{}'", self.name);
return; return;
} }
match response.await { match response.await {
Ok(Ok(())) => {} Ok(Ok(())) => {}
Ok(Err(err)) => eprintln!( Ok(Err(err)) => eprintln!(
"warning: profile '{}' param declaration '{name}' rejected by engine: {err}", "warning: profile '{}' param declaration '{name}' rejected by engine: {err}",
profile.name self.name
), ),
Err(_) => eprintln!( Err(_) => eprintln!(
"warning: engine dropped response while declaring param '{name}' from profile '{}'", "warning: engine dropped response while declaring param '{name}' from profile '{}'",
profile.name self.name
), ),
} }
} }
for record in &profile.metrics { for record in &self.metrics {
let (respond_to, response) = oneshot::channel(); let (respond_to, response) = oneshot::channel();
let request: DeclareMetricRequest = record.into(); let request: DeclareMetricRequest = record.into();
let name = request.name.clone(); let name = request.name.clone();
@@ -359,21 +366,22 @@ pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) {
.await .await
.is_err() .is_err()
{ {
eprintln!("warning: engine unavailable while loading profile '{}'", profile.name); eprintln!("warning: engine unavailable while loading profile '{}'", self.name);
return; return;
} }
match response.await { match response.await {
Ok(Ok(())) => {} Ok(Ok(())) => {}
Ok(Err(err)) => eprintln!( Ok(Err(err)) => eprintln!(
"warning: profile '{}' metric declaration '{name}' rejected by engine: {err}", "warning: profile '{}' metric declaration '{name}' rejected by engine: {err}",
profile.name self.name
), ),
Err(_) => eprintln!( Err(_) => eprintln!(
"warning: engine dropped response while declaring metric '{name}' from profile '{}'", "warning: engine dropped response while declaring metric '{name}' from profile '{}'",
profile.name self.name
), ),
} }
} }
}
} }
#[derive(Debug)] #[derive(Debug)]
@@ -403,7 +411,8 @@ impl From<io::Error> for ProfileError {
} }
} }
pub(crate) fn load_legacy_profile_store(path: &Path) -> Option<Vec<Profile>> { impl Profile {
pub(crate) fn load_legacy_store(path: &Path) -> Option<Vec<Profile>> {
let contents = fs::read_to_string(path).ok()?; let contents = fs::read_to_string(path).ok()?;
match serde_json::from_str::<Vec<Profile>>(&contents) { match serde_json::from_str::<Vec<Profile>>(&contents) {
Ok(profiles) => Some(profiles), Ok(profiles) => Some(profiles),
@@ -415,13 +424,14 @@ pub(crate) fn load_legacy_profile_store(path: &Path) -> Option<Vec<Profile>> {
None None
} }
} }
}
} }
pub(crate) fn migrate_legacy_profile_shapes(profiles: &mut [Profile]) -> bool { pub(crate) fn migrate_legacy_profile_shapes(profiles: &mut [Profile]) -> bool {
let mut any_migrated = false; let mut any_migrated = false;
for profile in profiles { for profile in profiles {
let legacy_migrated = migrate_legacy_metric_params(profile); let legacy_migrated = profile.migrate_legacy_metric_params();
let duration_migrated = migrate_duration_value_params(profile); let duration_migrated = profile.migrate_duration_value_params();
if legacy_migrated || duration_migrated { if legacy_migrated || duration_migrated {
any_migrated = true; any_migrated = true;
} }
@@ -465,8 +475,8 @@ impl Settings {
pub fn ensure_default_profile(&mut self, settings_path: &Path) -> Result<(), ProfileError> { pub fn ensure_default_profile(&mut self, settings_path: &Path) -> Result<(), ProfileError> {
if self.profile(DEFAULT_PROFILE_NAME).is_none() { if self.profile(DEFAULT_PROFILE_NAME).is_none() {
let mut profile = Profile::new(DEFAULT_PROFILE_NAME.to_string()); let mut profile = Profile::new(DEFAULT_PROFILE_NAME.to_string());
profile.metrics.push(seeded_error_split_metric()); profile.metrics.push(MetricDeclarationRecord::seeded_error_split());
profile.dashboards.push(seeded_default_dashboard()); profile.dashboards.push(Dashboard::seeded_default());
self.profiles.push(profile); self.profiles.push(profile);
} }
if self.active_profile.is_none() { 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 settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone()));
let engine_tx = engine::spawn(settings_handle, 128); 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")] { for (sequence, text) in [(0u64, "all good"), (1, "an ERROR happened"), (2, "fine again")] {
engine_tx engine_tx
@@ -981,7 +991,7 @@ mod tests {
let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone()));
let engine_tx = crate::engine::spawn(settings_handle, 16); 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 engine_tx
.tx .tx
@@ -1133,7 +1143,7 @@ mod tests {
let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
tokio::runtime::Runtime::new().unwrap().block_on(async { tokio::runtime::Runtime::new().unwrap().block_on(async {
let engine_tx = crate::engine::spawn(settings_handle, 16); 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 engine_tx
.tx .tx
.send(EngineMessage::Ingest(IngestedLine { .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_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); 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) = let EngineResponse::MetricDeclarations(metrics) =
ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await
src/promql.rs
+37 -35
View File
@@ -32,7 +32,8 @@ enum RangeSelector {
BareWidth(f64), BareWidth(f64),
} }
pub fn parse_query(input: &str) -> Result<PromqlQuery, String> { impl PromqlQuery {
pub fn parse(input: &str) -> Result<PromqlQuery, String> {
let mut parser = Parser::new(input); let mut parser = Parser::new(input);
let query = parser.parse_query()?; let query = parser.parse_query()?;
parser.skip_whitespace(); parser.skip_whitespace();
@@ -48,6 +49,7 @@ pub fn parse_query(input: &str) -> Result<PromqlQuery, String> {
); );
} }
Ok(query) Ok(query)
}
} }
struct Parser<'a> { struct Parser<'a> {
@@ -328,7 +330,7 @@ mod tests {
#[test] #[test]
fn parses_bare_metric_name() { 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.agg, None);
assert_eq!(query.metric_name, "requests"); assert_eq!(query.metric_name, "requests");
assert!(query.matchers.is_empty()); assert!(query.matchers.is_empty());
@@ -337,7 +339,7 @@ mod tests {
#[test] #[test]
fn parses_metric_with_single_label_matcher() { 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.metric_name, "requests");
assert_eq!( assert_eq!(
query.matchers, query.matchers,
@@ -347,7 +349,7 @@ mod tests {
#[test] #[test]
fn parses_metric_with_multiple_label_matchers() { 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.len(), 2);
assert_eq!(query.matchers[0].name, "path"); assert_eq!(query.matchers[0].name, "path");
assert_eq!(query.matchers[1].name, "method"); assert_eq!(query.matchers[1].name, "method");
@@ -355,67 +357,67 @@ mod tests {
#[test] #[test]
fn parses_range_selector() { 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))); assert_eq!(query.range, Some(Duration::from_secs(300)));
} }
#[test] #[test]
fn parses_label_matchers_and_range_together() { 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.matchers.len(), 1);
assert_eq!(query.range, Some(Duration::from_secs(3600))); assert_eq!(query.range, Some(Duration::from_secs(3600)));
} }
#[test] #[test]
fn parses_sum_aggregation() { 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.agg, Some(AggOp::Sum));
assert_eq!(query.metric_name, "requests"); assert_eq!(query.metric_name, "requests");
} }
#[test] #[test]
fn parses_avg_aggregation() { 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)); assert_eq!(query.agg, Some(AggOp::Avg));
} }
#[test] #[test]
fn parses_rate_with_range() { 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.agg, Some(AggOp::Rate));
assert_eq!(query.range, Some(Duration::from_secs(300))); assert_eq!(query.range, Some(Duration::from_secs(300)));
} }
#[test] #[test]
fn rejects_rate_without_range() { 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")); assert!(err.contains("range selector"));
} }
#[test] #[test]
fn rejects_empty_input() { fn rejects_empty_input() {
assert!(parse_query("").is_err()); assert!(PromqlQuery::parse("").is_err());
} }
#[test] #[test]
fn rejects_unclosed_brace() { fn rejects_unclosed_brace() {
assert!(parse_query(r#"requests{path="/a""#).is_err()); assert!(PromqlQuery::parse(r#"requests{path="/a""#).is_err());
} }
#[test] #[test]
fn rejects_unclosed_bracket() { fn rejects_unclosed_bracket() {
assert!(parse_query("requests[5m").is_err()); assert!(PromqlQuery::parse("requests[5m").is_err());
} }
#[test] #[test]
fn rejects_unknown_aggregation() { 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")); assert!(err.contains("unexpected trailing input"));
} }
#[test] #[test]
fn parses_count_aggregation() { 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.agg, Some(AggOp::Count));
assert_eq!(query.metric_name, "requests"); assert_eq!(query.metric_name, "requests");
assert!(!query.inner_rate); assert!(!query.inner_rate);
@@ -423,37 +425,37 @@ mod tests {
#[test] #[test]
fn parses_count_with_by_clause() { 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.agg, Some(AggOp::Count));
assert_eq!(query.by_labels, vec!["path".to_string()]); assert_eq!(query.by_labels, vec!["path".to_string()]);
} }
#[test] #[test]
fn rejects_regex_label_matcher() { 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")); assert!(err.contains("regex label matchers"));
} }
#[test] #[test]
fn rejects_missing_metric_name() { fn rejects_missing_metric_name() {
assert!(parse_query(r#"{path="/a"}"#).is_err()); assert!(PromqlQuery::parse(r#"{path="/a"}"#).is_err());
} }
#[test] #[test]
fn rejects_unknown_duration_unit() { 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")); assert!(err.contains("duration unit"));
} }
#[test] #[test]
fn rejects_trailing_garbage() { 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")); assert!(err.contains("unexpected trailing input"));
} }
#[test] #[test]
fn parses_histogram_with_bare_width() { 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.agg, Some(AggOp::Histogram));
assert_eq!(query.metric_name, "latency_ms"); assert_eq!(query.metric_name, "latency_ms");
assert_eq!(query.bucket_width, Some(10.0)); assert_eq!(query.bucket_width, Some(10.0));
@@ -462,27 +464,27 @@ mod tests {
#[test] #[test]
fn rejects_histogram_without_width() { 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")); assert!(err.contains("bucket width"));
} }
#[test] #[test]
fn parses_gauge_with_no_range_selector() { 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.agg, Some(AggOp::Gauge));
assert_eq!(query.matchers, vec![LabelMatcher { name: "host".to_string(), value: "a".to_string() }]); assert_eq!(query.matchers, vec![LabelMatcher { name: "host".to_string(), value: "a".to_string() }]);
} }
#[test] #[test]
fn rate_range_selector_is_still_a_duration_not_a_bare_width() { 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.range, Some(Duration::from_secs(300)));
assert_eq!(query.bucket_width, None); assert_eq!(query.bucket_width, None);
} }
#[test] #[test]
fn parses_sum_of_rate() { 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_eq!(query.agg, Some(AggOp::Sum));
assert!(query.inner_rate); assert!(query.inner_rate);
assert_eq!(query.metric_name, "requests"); assert_eq!(query.metric_name, "requests");
@@ -492,7 +494,7 @@ mod tests {
#[test] #[test]
fn parses_avg_of_rate_with_label_matchers() { 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_eq!(query.agg, Some(AggOp::Avg));
assert!(query.inner_rate); assert!(query.inner_rate);
assert_eq!(query.matchers, vec![LabelMatcher { name: "path".to_string(), value: "/a".to_string() }]); assert_eq!(query.matchers, vec![LabelMatcher { name: "path".to_string(), value: "/a".to_string() }]);
@@ -500,13 +502,13 @@ mod tests {
#[test] #[test]
fn rejects_sum_of_rate_without_range() { 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")); assert!(err.contains("range selector"));
} }
#[test] #[test]
fn parses_sum_with_by_clause() { 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_eq!(query.agg, Some(AggOp::Sum));
assert!(!query.inner_rate); assert!(!query.inner_rate);
assert_eq!(query.by_labels, vec!["path".to_string()]); assert_eq!(query.by_labels, vec!["path".to_string()]);
@@ -514,44 +516,44 @@ mod tests {
#[test] #[test]
fn parses_sum_of_rate_with_by_clause_multiple_labels() { 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!(query.inner_rate);
assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]); assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]);
} }
#[test] #[test]
fn sum_without_by_clause_has_empty_by_labels() { 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()); assert!(query.by_labels.is_empty());
} }
#[test] #[test]
fn rejects_empty_by_clause() { 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")); assert!(err.contains("at least one label"));
} }
#[test] #[test]
fn rejects_unterminated_by_clause() { fn rejects_unterminated_by_clause() {
assert!(parse_query("sum(requests) by (path").is_err()); assert!(PromqlQuery::parse("sum(requests) by (path").is_err());
} }
#[test] #[test]
fn parses_sum_with_prefix_by_clause() { 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.agg, Some(AggOp::Sum));
assert_eq!(query.by_labels, vec!["path".to_string()]); assert_eq!(query.by_labels, vec!["path".to_string()]);
} }
#[test] #[test]
fn parses_sum_with_prefix_by_clause_multiple_labels() { 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()]); assert_eq!(query.by_labels, vec!["path".to_string(), "method".to_string()]);
} }
#[test] #[test]
fn rejects_by_clause_both_before_and_after_aggregation() { 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")); assert!(err.contains("only appear once"));
} }
} }
src/settings.rs
+1 -1
View File
@@ -197,7 +197,7 @@ impl Settings {
.map(|parent| parent.join("profiles")) .map(|parent| parent.join("profiles"))
.unwrap_or_else(|| PathBuf::from("profiles")) .unwrap_or_else(|| PathBuf::from("profiles"))
.join("profiles.json"); .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; return;
}; };
self.profiles = profiles; self.profiles = profiles;