Add Metrics API: declare/list params and metrics, query bucket values (ticket 004)
Exposes param/metric declarations and derived bucket data over HTTP: POST /api/params and POST /api/metrics validate and declare against the engine, persisting into the active profile (if any) so declarations survive a restart; GET /api/metrics lists declarations; GET /api/metrics/:name queries bucket values with label, time-range, sequence, and limit filters. Adds a new EngineQuery::BucketValues variant that layers this filtering on top of ticket 003's bucket_values helper. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
f0348589f5
commit
49e8590cf3
Generated
+2
@@ -601,6 +601,7 @@ dependencies = [
|
||||
"chrono",
|
||||
"clap",
|
||||
"directories",
|
||||
"http-body-util",
|
||||
"humantime",
|
||||
"pest",
|
||||
"pest_derive",
|
||||
@@ -609,6 +610,7 @@ dependencies = [
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
"tower",
|
||||
"tower-http",
|
||||
]
|
||||
|
||||
|
||||
@@ -19,3 +19,5 @@ directories = "6.0.0"
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3.27.0"
|
||||
tower = { version = "0.5", features = ["util"] }
|
||||
http-body-util = "0.1"
|
||||
|
||||
+427
-2
@@ -1,8 +1,16 @@
|
||||
use std::collections::BTreeMap;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::{Arc, Mutex};
|
||||
|
||||
use crate::engine::{EngineHandle, EngineMessage, EngineQuery, EngineResponse};
|
||||
use crate::dto::{CssEntryDto, ImportCssRequest, LinkCssRequest, LogsParams, TimelineParams};
|
||||
use crate::dto::{
|
||||
BucketValuesParams, CssEntryDto, ErrorDto, ImportCssRequest, LinkCssRequest,
|
||||
MetricDeclarationDto, ParamDeclarationDto, LogsParams, TimelineParams,
|
||||
};
|
||||
use crate::engine::{
|
||||
DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery,
|
||||
EngineResponse,
|
||||
};
|
||||
use crate::profile::{self, ProfileStore};
|
||||
use crate::settings::{CssError, Settings};
|
||||
use axum::extract::{Path as AxumPath, Query, State};
|
||||
use axum::http::{header, StatusCode};
|
||||
@@ -32,6 +40,48 @@ impl ApiState {
|
||||
.map(Path::to_path_buf)
|
||||
.unwrap_or_else(|| PathBuf::from("."))
|
||||
}
|
||||
|
||||
/// Persists a newly-declared param into the active profile, if any (see
|
||||
/// `ProfileStore::record_param_declaration`), so it survives a restart.
|
||||
/// A no-op (not an error) when no profile is currently active — the
|
||||
/// engine declaration above has already succeeded regardless. Failures
|
||||
/// (e.g. I/O) are logged to stderr rather than surfaced to the caller: the
|
||||
/// engine declaration is the source of truth for the running process, and
|
||||
/// this is a best-effort persistence step layered on top.
|
||||
fn persist_param_declaration(&self, request: &DeclareParamRequest) {
|
||||
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_param_declaration(&store_path, &profile_name, request) {
|
||||
eprintln!(
|
||||
"warning: failed to persist param declaration '{}' into profile '{profile_name}': {err}",
|
||||
request.name
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Same as `persist_param_declaration`, for metric declarations.
|
||||
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) {
|
||||
eprintln!(
|
||||
"warning: failed to persist metric declaration '{}' into profile '{profile_name}': {err}",
|
||||
request.name
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// The active profile's store path and name, if a profile is currently active.
|
||||
fn active_profile_location(&self) -> Option<(PathBuf, String)> {
|
||||
let settings = self.settings.lock().unwrap();
|
||||
let profile_name = settings.profiles.active_profile.clone()?;
|
||||
let profiles_dir = profile::resolve_profiles_dir(&settings, &self.settings_path);
|
||||
Some((profile::profiles_store_path(&profiles_dir), profile_name))
|
||||
}
|
||||
}
|
||||
|
||||
pub fn router(
|
||||
@@ -52,6 +102,9 @@ pub fn router(
|
||||
.route("/api/logs", get(get_logs))
|
||||
.route("/api/timeline", get(get_timeline))
|
||||
.route("/api/stats", get(get_stats))
|
||||
.route("/api/params", post(declare_param))
|
||||
.route("/api/metrics", get(list_metrics).post(declare_metric))
|
||||
.route("/api/metrics/{name}", get(get_metric_bucket_values))
|
||||
.route("/api/settings/css", get(list_css).post(import_css))
|
||||
.route("/api/settings/css/link", post(link_css))
|
||||
.route("/api/settings/css/{id}", delete(remove_css))
|
||||
@@ -99,6 +152,122 @@ async fn get_stats(State(state): State<ApiState>) -> impl IntoResponse {
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/params` — declares a param. Body mirrors `DeclareParamRequest`:
|
||||
/// `{ "name": "...", "message_match_regex": "...", "capture_group": "...",
|
||||
/// "kind": "string" | "number" | "timestamp" }`. On success, also persists
|
||||
/// the declaration into the active profile (if any — see
|
||||
/// `ApiState::persist_param_declaration`) and returns `201 Created` with the
|
||||
/// declaration as `ParamDeclarationDto`. On a validation failure (bad regex,
|
||||
/// unknown capture group), returns `400 Bad Request` with
|
||||
/// `{ "error": "..." }`.
|
||||
async fn declare_param(
|
||||
State(state): State<ApiState>,
|
||||
Json(body): Json<DeclareParamRequest>,
|
||||
) -> impl IntoResponse {
|
||||
let (respond_to, response) = oneshot::channel();
|
||||
let request = body.clone();
|
||||
if state
|
||||
.engine
|
||||
.send(EngineMessage::DeclareParam { request: body, respond_to })
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return StatusCode::SERVICE_UNAVAILABLE.into_response();
|
||||
}
|
||||
match response.await {
|
||||
Ok(Ok(())) => {
|
||||
state.persist_param_declaration(&request);
|
||||
let dto = ParamDeclarationDto {
|
||||
name: request.name,
|
||||
message_match_regex: request.message_match_regex,
|
||||
capture_group: request.capture_group,
|
||||
kind: request.kind,
|
||||
};
|
||||
(StatusCode::CREATED, Json(dto)).into_response()
|
||||
}
|
||||
Ok(Err(err)) => (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response(),
|
||||
Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// `GET /api/metrics` — lists all declared metrics (name, regex, params) as
|
||||
/// `Vec<MetricDeclarationDto>`.
|
||||
async fn list_metrics(State(state): State<ApiState>) -> impl IntoResponse {
|
||||
match ask(&state.engine, EngineQuery::MetricDeclarations).await {
|
||||
Ok(EngineResponse::MetricDeclarations(declarations)) => Json(declarations).into_response(),
|
||||
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
Err(status) => status.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// `POST /api/metrics` — declares a metric. Body mirrors
|
||||
/// `DeclareMetricRequest`: `{ "name": "...", "message_match_regex": "...",
|
||||
/// "params": [{ "param_name": "...", "capture_group": "...", "kind": "..." }] }`.
|
||||
/// Same success/failure shape as `declare_param` above, persisting into the
|
||||
/// active profile via `ApiState::persist_metric_declaration`.
|
||||
async fn declare_metric(
|
||||
State(state): State<ApiState>,
|
||||
Json(body): Json<DeclareMetricRequest>,
|
||||
) -> impl IntoResponse {
|
||||
let (respond_to, response) = oneshot::channel();
|
||||
let request = body.clone();
|
||||
if state
|
||||
.engine
|
||||
.send(EngineMessage::DeclareMetric { request: body, respond_to })
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return StatusCode::SERVICE_UNAVAILABLE.into_response();
|
||||
}
|
||||
match response.await {
|
||||
Ok(Ok(())) => {
|
||||
state.persist_metric_declaration(&request);
|
||||
let dto = MetricDeclarationDto {
|
||||
name: request.name,
|
||||
message_match_regex: request.message_match_regex,
|
||||
params: request.params,
|
||||
};
|
||||
(StatusCode::CREATED, Json(dto)).into_response()
|
||||
}
|
||||
Ok(Err(err)) => (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response(),
|
||||
Err(_) => StatusCode::SERVICE_UNAVAILABLE.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Parses the `labels` query param (`"foo=bar,baz=qux"`) into a
|
||||
/// `key -> expected value` map. Malformed pairs (no `=`) are skipped rather
|
||||
/// than rejected — see `BucketValuesParams` doc comment for the format.
|
||||
fn parse_labels_filter(raw: &str) -> BTreeMap<String, String> {
|
||||
raw.split(',')
|
||||
.filter_map(|pair| pair.split_once('='))
|
||||
.map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// `GET /api/metrics/:name` — queries bucket values for one metric. Query
|
||||
/// params: `labels` (see `parse_labels_filter`), `start_ms`, `end_ms`,
|
||||
/// `since_sequence`, `limit` (mirroring `/api/logs`/`/api/timeline`).
|
||||
async fn get_metric_bucket_values(
|
||||
State(state): State<ApiState>,
|
||||
AxumPath(metric_name): AxumPath<String>,
|
||||
Query(params): Query<BucketValuesParams>,
|
||||
) -> impl IntoResponse {
|
||||
let labels_filter = params.labels.as_deref().map(parse_labels_filter);
|
||||
let query = EngineQuery::BucketValues {
|
||||
metric_name,
|
||||
labels_filter,
|
||||
start_ms: params.start_ms,
|
||||
end_ms: params.end_ms,
|
||||
since_sequence: params.since_sequence,
|
||||
limit: params.limit,
|
||||
};
|
||||
match ask(&state.engine, query).await {
|
||||
Ok(EngineResponse::BucketValues(values)) => Json(values).into_response(),
|
||||
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|
||||
Err(status) => status.into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Maps a `CssError` from a settings mutation/lookup to a response status
|
||||
/// and message: `NotFound` -> 404, `NotImported` -> 400 (the id is valid but
|
||||
/// refers to a linked entry, which this operation doesn't apply to), `Io` ->
|
||||
@@ -200,3 +369,259 @@ async fn get_css_content(
|
||||
Err(err) => css_error_response(err),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::engine::{self, IngestedLine};
|
||||
use crate::timestamp::Timestamp;
|
||||
use axum::body::Body;
|
||||
use axum::http::Request;
|
||||
use http_body_util::BodyExt;
|
||||
use serde_json::{json, Value};
|
||||
use tokio::sync::mpsc;
|
||||
use tower::ServiceExt;
|
||||
|
||||
/// Spins up a fresh engine + router pair, backed by a temp settings file
|
||||
/// (and empty static dir), for one test. Returns the router, the engine
|
||||
/// handle (so a test can ingest lines directly without an HTTP route for
|
||||
/// it), and the `TempDir` guard (kept alive for the test's duration).
|
||||
fn test_app() -> (Router, EngineHandle, SettingsHandle, tempfile::TempDir) {
|
||||
let (engine_tx, engine_rx) = mpsc::channel(128);
|
||||
tokio::spawn(engine::run(engine_rx));
|
||||
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let settings_path = dir.path().join("settings.json");
|
||||
let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));
|
||||
let static_dir = dir.path().join("static");
|
||||
std::fs::create_dir_all(&static_dir).unwrap();
|
||||
|
||||
let app = router(
|
||||
engine_tx.clone(),
|
||||
static_dir.to_str().unwrap(),
|
||||
settings.clone(),
|
||||
settings_path,
|
||||
);
|
||||
(app, engine_tx, settings, dir)
|
||||
}
|
||||
|
||||
async fn send_json(app: &Router, method: &str, uri: &str, body: Value) -> (StatusCode, Value) {
|
||||
let request = Request::builder()
|
||||
.method(method)
|
||||
.uri(uri)
|
||||
.header(header::CONTENT_TYPE, "application/json")
|
||||
.body(Body::from(body.to_string()))
|
||||
.unwrap();
|
||||
let response = app.clone().oneshot(request).await.unwrap();
|
||||
let status = response.status();
|
||||
let bytes = response.into_body().collect().await.unwrap().to_bytes();
|
||||
let value = if bytes.is_empty() {
|
||||
Value::Null
|
||||
} else {
|
||||
serde_json::from_slice(&bytes).unwrap()
|
||||
};
|
||||
(status, value)
|
||||
}
|
||||
|
||||
async fn get_json(app: &Router, uri: &str) -> (StatusCode, Value) {
|
||||
let request = Request::builder().method("GET").uri(uri).body(Body::empty()).unwrap();
|
||||
let response = app.clone().oneshot(request).await.unwrap();
|
||||
let status = response.status();
|
||||
let bytes = response.into_body().collect().await.unwrap().to_bytes();
|
||||
let value = if bytes.is_empty() {
|
||||
Value::Null
|
||||
} else {
|
||||
serde_json::from_slice(&bytes).unwrap()
|
||||
};
|
||||
(status, value)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn declares_param_and_returns_201() {
|
||||
let (app, _engine, _settings, _dir) = test_app();
|
||||
let (status, body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/params",
|
||||
json!({
|
||||
"name": "level",
|
||||
"message_match_regex": r"level=(?P<level>\w+)",
|
||||
"capture_group": "level",
|
||||
"kind": "string"
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::CREATED);
|
||||
assert_eq!(body["name"], "level");
|
||||
assert_eq!(body["capture_group"], "level");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rejects_invalid_param_regex_with_400_and_error_body() {
|
||||
let (app, _engine, _settings, _dir) = test_app();
|
||||
let (status, body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/params",
|
||||
json!({
|
||||
"name": "level",
|
||||
"message_match_regex": "level=(",
|
||||
"capture_group": "level",
|
||||
"kind": "string"
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert!(body["error"].as_str().unwrap().contains("invalid regex"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn declares_metric_and_shows_up_in_list() {
|
||||
let (app, _engine, _settings, _dir) = test_app();
|
||||
let (status, body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/metrics",
|
||||
json!({
|
||||
"name": "requests",
|
||||
"message_match_regex": r"request path=(?P<path>\S+)",
|
||||
"params": [
|
||||
{ "param_name": "path", "capture_group": "path", "kind": "string" }
|
||||
]
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::CREATED);
|
||||
assert_eq!(body["name"], "requests");
|
||||
|
||||
let (status, body) = get_json(&app, "/api/metrics").await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
let metrics = body.as_array().unwrap();
|
||||
assert_eq!(metrics.len(), 1);
|
||||
assert_eq!(metrics[0]["name"], "requests");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rejects_metric_with_unknown_capture_group() {
|
||||
let (app, _engine, _settings, _dir) = test_app();
|
||||
let (status, body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/metrics",
|
||||
json!({
|
||||
"name": "requests",
|
||||
"message_match_regex": r"request path=(?P<path>\S+)",
|
||||
"params": [
|
||||
{ "param_name": "duration", "capture_group": "duration", "kind": "number" }
|
||||
]
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::BAD_REQUEST);
|
||||
assert!(body["error"].as_str().unwrap().contains("duration"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn queries_bucket_values_with_label_filter_and_range() {
|
||||
let (app, engine, _settings, _dir) = test_app();
|
||||
|
||||
send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/metrics",
|
||||
json!({
|
||||
"name": "requests",
|
||||
"message_match_regex": r"request path=(?P<path>\S+)",
|
||||
"params": [
|
||||
{ "param_name": "path", "capture_group": "path", "kind": "string" }
|
||||
]
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
|
||||
for (sequence, text) in [
|
||||
(0u64, "request path=/a"),
|
||||
(1u64, "request path=/b"),
|
||||
(2u64, "request path=/a"),
|
||||
] {
|
||||
engine
|
||||
.send(EngineMessage::Ingest(IngestedLine {
|
||||
sequence,
|
||||
timestamp: Timestamp::from_millis(sequence as i64 * 1000),
|
||||
text: text.to_string(),
|
||||
}))
|
||||
.await
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
let (status, body) = get_json(&app, "/api/metrics/requests").await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
assert_eq!(body.as_array().unwrap().len(), 3);
|
||||
|
||||
let (status, body) = get_json(&app, "/api/metrics/requests?labels=path=/a").await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
let values = body.as_array().unwrap();
|
||||
assert_eq!(values.len(), 2);
|
||||
for value in values {
|
||||
// `MetricPrimitive` is a Rust enum with derived `Serialize`, so its
|
||||
// JSON form is externally tagged: `{"String": "/a"}`.
|
||||
assert_eq!(value["labels"]["path"]["String"], "/a");
|
||||
}
|
||||
|
||||
// Bucket values are grouped by label combination (ordered by the
|
||||
// labels' `BTreeMap` key, i.e. "/a" before "/b"), not globally by
|
||||
// sequence — so filtering to sequence > 0 leaves sequence 2 (from the
|
||||
// "/a" bucket) ahead of sequence 1 (from the "/b" bucket), and a
|
||||
// limit of 1 picks the former.
|
||||
let (status, body) = get_json(&app, "/api/metrics/requests?since_sequence=0&limit=1").await;
|
||||
assert_eq!(status, StatusCode::OK);
|
||||
let values = body.as_array().unwrap();
|
||||
assert_eq!(values.len(), 1);
|
||||
assert_eq!(values[0]["sequence"], 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn declaring_param_persists_into_active_profile() {
|
||||
let (app, _engine, settings_handle, dir) = test_app();
|
||||
|
||||
// Set up an active profile against the *same* shared `settings`
|
||||
// handle the router was built with — mutating a separate `Settings`
|
||||
// loaded fresh from disk wouldn't be visible to the running app's
|
||||
// in-memory settings.
|
||||
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();
|
||||
}
|
||||
|
||||
let (status, _body) = send_json(
|
||||
&app,
|
||||
"POST",
|
||||
"/api/params",
|
||||
json!({
|
||||
"name": "level",
|
||||
"message_match_regex": r"level=(?P<level>\w+)",
|
||||
"capture_group": "level",
|
||||
"kind": "string"
|
||||
}),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(status, StatusCode::CREATED);
|
||||
|
||||
let reloaded_store = ProfileStore::load(&store_path);
|
||||
let profile = reloaded_store.get("web").unwrap();
|
||||
assert_eq!(profile.params.len(), 1);
|
||||
assert_eq!(profile.params[0].name, "level");
|
||||
}
|
||||
}
|
||||
|
||||
+8
-3
@@ -1,18 +1,23 @@
|
||||
use crate::metric_primitive::MetricPrimitive;
|
||||
use crate::timestamp::Timestamp;
|
||||
use serde::Serialize;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// A single parsed metric primitive extracted from a regex capture group.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct RegexToMetricPrimitiveMatcher {
|
||||
pub param_name: String,
|
||||
pub capture_group: String,
|
||||
pub kind: MetricPrimitiveKind,
|
||||
}
|
||||
|
||||
/// `rename_all = "lowercase"` matches every other `kind`-style enum exposed
|
||||
/// over the API/settings JSON (`ColorScheme`, `CssEntryKind`,
|
||||
/// `ProfileMetricPrimitiveKind`), so `POST /api/params`/`POST /api/metrics`
|
||||
/// bodies use `"string"`/`"number"`/`"timestamp"`.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Copy, Serialize)]
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum MetricPrimitiveKind {
|
||||
String,
|
||||
Number,
|
||||
|
||||
+36
@@ -1,6 +1,8 @@
|
||||
use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher};
|
||||
use crate::metric_primitive::MetricPrimitive;
|
||||
use crate::settings::CssEntry;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct LogEntryDto {
|
||||
@@ -54,6 +56,40 @@ pub struct TimelineParams {
|
||||
pub bucket_ms: Option<i64>,
|
||||
}
|
||||
|
||||
/// One entry from a metric's derived bucket timeseries: the label combination
|
||||
/// that bucket represents, plus one `(sequence, timestamp, value)` occurrence
|
||||
/// within it.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
pub struct BucketValueDto {
|
||||
pub labels: BTreeMap<String, MetricPrimitive>,
|
||||
pub sequence: u64,
|
||||
pub timestamp_ms: i64,
|
||||
pub value: MetricPrimitive,
|
||||
}
|
||||
|
||||
/// Query params for `GET /api/metrics/:name`. `labels` is a single
|
||||
/// comma-separated `key=value` list (e.g. `?labels=path=/a,duration=5`)
|
||||
/// rather than repeated `label.foo=bar` params, since axum's `Query`
|
||||
/// extractor needs a fixed/known set of field names — an arbitrary label set
|
||||
/// doesn't fit that without a custom deserializer. Every listed label must
|
||||
/// match (compared against each label value's `Display` form, see
|
||||
/// `MetricPrimitive`); unlisted labels on a bucket are ignored.
|
||||
#[derive(Debug, Deserialize)]
|
||||
pub struct BucketValuesParams {
|
||||
pub labels: Option<String>,
|
||||
pub since_sequence: Option<u64>,
|
||||
pub start_ms: Option<i64>,
|
||||
pub end_ms: Option<i64>,
|
||||
pub limit: Option<usize>,
|
||||
}
|
||||
|
||||
/// JSON error body returned by API handlers on a 4xx/5xx response, e.g.
|
||||
/// `{ "error": "invalid regex: ..." }`.
|
||||
#[derive(Debug, Serialize)]
|
||||
pub struct ErrorDto {
|
||||
pub error: String,
|
||||
}
|
||||
|
||||
/// Request body for `POST /api/settings/css`: raw CSS content to import and
|
||||
/// store alongside the settings file.
|
||||
#[derive(Debug, Deserialize)]
|
||||
|
||||
+63
-5
@@ -1,12 +1,13 @@
|
||||
use crate::data_value::{DataValue, MetricPrimitiveKind, RegexToMetricPrimitiveMatcher};
|
||||
use crate::dto::{
|
||||
LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, StatsDto,
|
||||
BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, StatsDto,
|
||||
TimelineBucketDto, TimelineParams,
|
||||
};
|
||||
use crate::index_key::IndexKey;
|
||||
use crate::metric_primitive::{MetricPrimitive, Number};
|
||||
use crate::timestamp::Timestamp;
|
||||
use regex::Regex;
|
||||
use serde::Deserialize;
|
||||
use std::collections::BTreeMap;
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
|
||||
@@ -16,6 +17,10 @@ pub struct IngestedLine {
|
||||
pub text: String,
|
||||
}
|
||||
|
||||
/// Also the request body for `POST /api/params` (ticket 004): `Deserialize` is
|
||||
/// derived directly on this engine type rather than adding a separate DTO,
|
||||
/// since its shape is already exactly what the API needs.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct DeclareParamRequest {
|
||||
pub name: String,
|
||||
pub message_match_regex: String,
|
||||
@@ -28,6 +33,11 @@ pub struct DeclareParamRequest {
|
||||
/// `RegexToMetricPrimitiveMatcher::capture_group` must name a capture group on that
|
||||
/// regex). Referencing an already-declared `ParamDeclaration` (001) by name is not
|
||||
/// supported yet — that's left for a future ticket.
|
||||
///
|
||||
/// Also the request body for `POST /api/metrics` (ticket 004) — see
|
||||
/// `DeclareParamRequest`'s doc comment for why `Deserialize` lives here rather
|
||||
/// than on a separate DTO.
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct DeclareMetricRequest {
|
||||
pub name: String,
|
||||
pub message_match_regex: String,
|
||||
@@ -41,6 +51,19 @@ pub enum EngineQuery {
|
||||
Stats,
|
||||
ParamDeclarations,
|
||||
MetricDeclarations,
|
||||
/// Bucket values for one metric (ticket 004's `GET /api/metrics/:name`).
|
||||
/// `labels_filter`, when present, keeps only buckets where every listed
|
||||
/// label matches — compared against each label value's `Display` form
|
||||
/// (see `MetricPrimitive`) rather than a typed value, since the filter
|
||||
/// arrives as plain strings off the query string.
|
||||
BucketValues {
|
||||
metric_name: String,
|
||||
labels_filter: Option<BTreeMap<String, String>>,
|
||||
start_ms: Option<i64>,
|
||||
end_ms: Option<i64>,
|
||||
since_sequence: Option<u64>,
|
||||
limit: Option<usize>,
|
||||
},
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
@@ -75,6 +98,7 @@ pub enum EngineResponse {
|
||||
Stats(StatsDto),
|
||||
ParamDeclarations(Vec<ParamDeclarationDto>),
|
||||
MetricDeclarations(Vec<MetricDeclarationDto>),
|
||||
BucketValues(Vec<BucketValueDto>),
|
||||
}
|
||||
|
||||
pub type EngineHandle = mpsc::Sender<EngineMessage>;
|
||||
@@ -289,10 +313,10 @@ fn apply_all_metrics(
|
||||
}
|
||||
|
||||
/// Bucket values for one metric, in insertion order, optionally filtered to a single
|
||||
/// label combination. Internal helper for now — the metrics-API ticket ([004]) will
|
||||
/// wrap this (or an `EngineQuery` built on it) with time/sequence-range filtering and
|
||||
/// HTTP plumbing.
|
||||
#[allow(dead_code)]
|
||||
/// exact label combination. Used directly by tests (exact-match filtering) and, via
|
||||
/// `EngineQuery::BucketValues` in `handle_query` below, by the metrics API (ticket
|
||||
/// 004) — which layers its own textual label-filter and time/sequence-range
|
||||
/// filtering on top rather than using this function's `labels_filter` parameter.
|
||||
fn bucket_values<'a>(
|
||||
index: &'a BTreeMap<IndexKey, DataValue>,
|
||||
metric_name: &'a str,
|
||||
@@ -428,6 +452,40 @@ fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> En
|
||||
.collect();
|
||||
EngineResponse::MetricDeclarations(declarations)
|
||||
}
|
||||
EngineQuery::BucketValues {
|
||||
metric_name,
|
||||
labels_filter,
|
||||
start_ms,
|
||||
end_ms,
|
||||
since_sequence,
|
||||
limit,
|
||||
} => {
|
||||
let limit = limit.unwrap_or(500).min(10_000);
|
||||
let values: Vec<BucketValueDto> = bucket_values(index, &metric_name, None)
|
||||
.filter(|(labels, _)| {
|
||||
labels_filter.as_ref().is_none_or(|filter| {
|
||||
filter.iter().all(|(name, expected)| {
|
||||
labels
|
||||
.get(name)
|
||||
.is_some_and(|value| &value.to_string() == expected)
|
||||
})
|
||||
})
|
||||
})
|
||||
.filter(|(_, (sequence, timestamp, _))| {
|
||||
since_sequence.is_none_or(|since| *sequence > since)
|
||||
&& start_ms.is_none_or(|start| timestamp.as_millis() >= start)
|
||||
&& end_ms.is_none_or(|end| timestamp.as_millis() <= end)
|
||||
})
|
||||
.take(limit)
|
||||
.map(|(labels, (sequence, timestamp, value))| BucketValueDto {
|
||||
labels: labels.clone(),
|
||||
sequence: *sequence,
|
||||
timestamp_ms: timestamp.as_millis(),
|
||||
value: value.clone(),
|
||||
})
|
||||
.collect();
|
||||
EngineResponse::BucketValues(values)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
use crate::timestamp::Timestamp;
|
||||
use serde::Serialize;
|
||||
use std::cmp::Ordering;
|
||||
use std::fmt;
|
||||
|
||||
/// A simple, self-describing value used by metrics/labels without pulling in JSON.
|
||||
#[allow(dead_code)]
|
||||
@@ -42,6 +43,21 @@ impl Ord for MetricPrimitive {
|
||||
}
|
||||
}
|
||||
|
||||
/// Textual form used for label-filter comparisons over the metrics API
|
||||
/// (`GET /api/metrics/:name?labels=foo=bar`): query-string values are plain
|
||||
/// strings, so a label filter is compared against this formatted form rather
|
||||
/// than trying to parse the filter value into a typed `MetricPrimitive`.
|
||||
impl fmt::Display for MetricPrimitive {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
match self {
|
||||
MetricPrimitive::String(s) => write!(f, "{s}"),
|
||||
MetricPrimitive::Number(Number::Int(n)) => write!(f, "{n}"),
|
||||
MetricPrimitive::Number(Number::Float(n)) => write!(f, "{n}"),
|
||||
MetricPrimitive::Timestamp(ts) => write!(f, "{}", ts.as_millis()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Numbers are kept as a tagged int/float pair so exact integers stay exact.
|
||||
#[allow(dead_code)]
|
||||
#[derive(Debug, Clone, Copy, Serialize)]
|
||||
|
||||
Reference in New Issue
Block a user