Auto derived format
This commit is contained in:
+141
-11
@@ -13,7 +13,7 @@ use crate::engine::{
|
||||
EngineResponse,
|
||||
};
|
||||
use crate::alerts::{deliver_ntfy, AlertError};
|
||||
use crate::profile::{self, Profile, ProfileError, ProfileStore};
|
||||
use crate::profile::{self, ParamDeclarationRecord, Profile, ProfileError, ProfileStore};
|
||||
use crate::settings::{CssError, SettingsHandle};
|
||||
use axum::extract::{Path as AxumPath, Query, State};
|
||||
use axum::http::{header, StatusCode};
|
||||
@@ -55,13 +55,21 @@ impl ApiState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Same as `persist_param_declaration`, for metric declarations.
|
||||
fn persist_metric_declaration(&self, request: &DeclareMetricRequest) {
|
||||
/// Same as `persist_param_declaration`, for metric declarations. Also promotes any of the
|
||||
/// metric's referenced params that only exist as auto-derived engine state (see
|
||||
/// `promoted_param_records`) into persisted profile params, so a metric built against an
|
||||
/// auto-detected logfmt key survives a restart instead of being rejected as "undeclared" once
|
||||
/// declarations replay before any log line re-derives it.
|
||||
async fn persist_metric_declaration(&self, request: &DeclareMetricRequest) {
|
||||
let Some((store_path, profile_name)) = self.active_profile_location() else {
|
||||
return;
|
||||
};
|
||||
let mut store = ProfileStore::load(&store_path);
|
||||
if let Err(err) = store.record_metric_declaration(&store_path, &profile_name, request) {
|
||||
let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default();
|
||||
let promoted = promoted_param_records(&self.engine, &existing_params, request).await;
|
||||
if let Err(err) =
|
||||
store.record_metric_declaration(&store_path, &profile_name, request, &promoted)
|
||||
{
|
||||
eprintln!(
|
||||
"warning: failed to persist metric declaration '{}' into profile '{profile_name}': {err}",
|
||||
request.name
|
||||
@@ -70,14 +78,20 @@ impl ApiState {
|
||||
}
|
||||
|
||||
/// Same as `persist_metric_declaration`, for an in-place edit (`PUT /api/metrics/{name}`) — also drops the persisted record under `old_name` when the edit renames the metric, so a rename doesn't leave a stale duplicate declaration in the profile file.
|
||||
fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) {
|
||||
async fn persist_metric_declaration_update(&self, old_name: &str, request: &DeclareMetricRequest) {
|
||||
let Some((store_path, profile_name)) = self.active_profile_location() else {
|
||||
return;
|
||||
};
|
||||
let mut store = ProfileStore::load(&store_path);
|
||||
if let Err(err) =
|
||||
store.record_metric_declaration_update(&store_path, &profile_name, old_name, request)
|
||||
{
|
||||
let existing_params = store.get(&profile_name).map(|p| p.params.clone()).unwrap_or_default();
|
||||
let promoted = promoted_param_records(&self.engine, &existing_params, request).await;
|
||||
if let Err(err) = store.record_metric_declaration_update(
|
||||
&store_path,
|
||||
&profile_name,
|
||||
old_name,
|
||||
request,
|
||||
&promoted,
|
||||
) {
|
||||
eprintln!(
|
||||
"warning: failed to persist metric declaration update '{}' into profile '{profile_name}': {err}",
|
||||
request.name
|
||||
@@ -177,6 +191,47 @@ async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse
|
||||
response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)
|
||||
}
|
||||
|
||||
/// Any of `request`'s referenced params (`label_params`/`value_param`) not already present in
|
||||
/// `existing_params` are looked up live in the engine and returned as records ready to persist
|
||||
/// alongside the metric (see `ProfileStore::record_metric_declaration`'s `promoted_params`). A
|
||||
/// metric can validly reference a param that's only auto-derived engine state (see
|
||||
/// `crate::engine::auto_declare_params`) — that param was never written into any profile, so without
|
||||
/// this promotion the metric would be rejected as "undeclared" the next time declarations replay on
|
||||
/// startup, before any log line has re-derived it.
|
||||
async fn promoted_param_records(
|
||||
engine: &EngineHandle,
|
||||
existing_params: &[ParamDeclarationRecord],
|
||||
request: &DeclareMetricRequest,
|
||||
) -> Vec<ParamDeclarationRecord> {
|
||||
let missing: Vec<&str> = request
|
||||
.label_params
|
||||
.iter()
|
||||
.map(String::as_str)
|
||||
.chain(request.value_param.as_deref())
|
||||
.filter(|name| !existing_params.iter().any(|p| p.name == **name))
|
||||
.collect();
|
||||
if missing.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
let Ok(EngineResponse::ParamDeclarations(declarations)) =
|
||||
ask(engine, EngineQuery::ParamDeclarations).await
|
||||
else {
|
||||
return Vec::new();
|
||||
};
|
||||
declarations
|
||||
.into_iter()
|
||||
.filter(|param| missing.contains(¶m.name.as_str()))
|
||||
.map(|param| ParamDeclarationRecord {
|
||||
name: param.name,
|
||||
message_match_regex: param.message_match_regex,
|
||||
capture_group: param.capture_group,
|
||||
json_field_path: param.json_field_path,
|
||||
kind: param.kind.into(),
|
||||
elapsed_since_last: param.elapsed_since_last,
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn get_logs(
|
||||
State(state): State<ApiState>,
|
||||
Query(params): Query<LogsParams>,
|
||||
@@ -245,6 +300,9 @@ async fn declare_param(
|
||||
json_field_path: request.json_field_path,
|
||||
kind: request.kind,
|
||||
elapsed_since_last: request.elapsed_since_last,
|
||||
// A request that reached here was accepted by `declare_param`, which always stores
|
||||
// it as non-auto (see `engine::declare_param`).
|
||||
auto_derived: false,
|
||||
};
|
||||
(StatusCode::CREATED, Json(dto)).into_response()
|
||||
}
|
||||
@@ -277,7 +335,7 @@ async fn declare_metric(
|
||||
}
|
||||
match response.await {
|
||||
Ok(Ok(())) => {
|
||||
state.persist_metric_declaration(&request);
|
||||
state.persist_metric_declaration(&request).await;
|
||||
let dto = MetricDeclarationDto {
|
||||
name: request.name,
|
||||
message_match_regex: request.message_match_regex,
|
||||
@@ -311,7 +369,7 @@ async fn update_metric(
|
||||
}
|
||||
match response.await {
|
||||
Ok(Ok(())) => {
|
||||
state.persist_metric_declaration_update(&name, &request);
|
||||
state.persist_metric_declaration_update(&name, &request).await;
|
||||
let dto = MetricDeclarationDto {
|
||||
name: request.name,
|
||||
message_match_regex: request.message_match_regex,
|
||||
@@ -602,11 +660,14 @@ async fn delete_profile(
|
||||
}
|
||||
|
||||
/// Removes every currently-declared param/metric from the engine, then replays `profile`'s declarations into it.
|
||||
/// Auto-derived params (`ParamDeclarationDto::auto_derived`) are left alone — they aren't part of
|
||||
/// any profile, just facts discovered from the log stream, so a profile switch shouldn't discard
|
||||
/// them (they'd otherwise only reappear once a new line re-introduces that key).
|
||||
async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) {
|
||||
if let Ok(EngineResponse::ParamDeclarations(params)) =
|
||||
ask(engine, EngineQuery::ParamDeclarations).await
|
||||
{
|
||||
for param in params {
|
||||
for param in params.into_iter().filter(|param| !param.auto_derived) {
|
||||
let (respond_to, response) = oneshot::channel();
|
||||
if engine
|
||||
.send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to })
|
||||
@@ -1206,6 +1267,75 @@ mod tests {
|
||||
assert_eq!(profile.params[0].name, "level");
|
||||
}
|
||||
|
||||
/// Regression test: declaring a metric against a param that only exists as auto-derived engine
|
||||
/// state (discovered from an ingested logfmt key, never manually declared via `POST
|
||||
/// /api/params`) must persist that param into the active profile too, not just the metric —
|
||||
/// otherwise a restart replays the metric before any log line re-derives the auto param, and the
|
||||
/// engine rejects it as referencing an undeclared param (see `promoted_param_records`).
|
||||
#[tokio::test]
|
||||
async fn declaring_metric_against_auto_derived_param_promotes_it_into_the_profile() {
|
||||
let (app, engine, settings_handle, dir) = test_app();
|
||||
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
let profiles_dir = {
|
||||
let settings = settings_handle.lock().unwrap();
|
||||
profile::resolve_profiles_dir(&settings, &settings_path)
|
||||
};
|
||||
let store_path = profile::profiles_store_path(&profiles_dir);
|
||||
let mut store = ProfileStore::load(&store_path);
|
||||
{
|
||||
let mut settings = settings_handle.lock().unwrap();
|
||||
store
|
||||
.create(&store_path, &mut settings, &settings_path, "web".to_string())
|
||||
.unwrap();
|
||||
store
|
||||
.set_active(&mut settings, &settings_path, Some("web".to_string()))
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
// Ingest a line carrying a logfmt key before any param is manually declared, so "dx" only
|
||||
// exists as engine-internal auto-derived state (never persisted anywhere).
|
||||
ingest_lines(&engine, 0, &["dx=1 moved"]).await;
|
||||
|
||||
let (status, _body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/metrics",
|
||||
json!({
|
||||
"name": "direction_x",
|
||||
"message_match_regex": r"dx=(?P<dx>\S+) moved",
|
||||
"label_params": ["dx"]
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::CREATED);
|
||||
|
||||
let reloaded_store = ProfileStore::load(&store_path);
|
||||
let profile = reloaded_store.get("web").unwrap();
|
||||
assert_eq!(profile.metrics.len(), 1);
|
||||
assert_eq!(profile.metrics[0].name, "direction_x");
|
||||
// The auto-derived "dx" param got promoted into the profile alongside the metric.
|
||||
assert_eq!(profile.params.len(), 1);
|
||||
assert_eq!(profile.params[0].name, "dx");
|
||||
|
||||
// Simulate a restart: replay the persisted profile into a fresh engine with no prior ingestion.
|
||||
let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16);
|
||||
let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default()));
|
||||
tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle));
|
||||
profile::load_into_engine(&fresh_engine_tx, profile).await;
|
||||
|
||||
let (respond_to, response) = oneshot::channel();
|
||||
fresh_engine_tx
|
||||
.send(EngineMessage::Query { query: EngineQuery::MetricDeclarations, respond_to })
|
||||
.await
|
||||
.unwrap();
|
||||
let EngineResponse::MetricDeclarations(metrics) = response.await.unwrap() else {
|
||||
panic!("unexpected response");
|
||||
};
|
||||
assert_eq!(metrics.len(), 1);
|
||||
assert_eq!(metrics[0].name, "direction_x");
|
||||
}
|
||||
|
||||
async fn delete_request(app: &Router, uri: &str) -> StatusCode {
|
||||
let request = Request::builder().method("DELETE").uri(uri).body(Body::empty()).unwrap();
|
||||
app.clone().oneshot(request).await.unwrap().status()
|
||||
|
||||
Reference in New Issue
Block a user