Files
ottyl/src/api.rs
T
Hugh KaznowskiandClaude Sonnet 5 1d0f324c43 Add settings-configurable cap on logs kept in memory (ticket 026)
Settings.max_logs (default None = unlimited) is pushed to the engine
via a new EngineMessage::SetMaxLogs, enforced on every ingest and
whenever the setting changes at runtime (no restart required).
Eviction drops the oldest (lowest-sequence) logs; sequence numbering
isn't renumbered and buckets already derived from an evicted log are
left in place. Exposed via PATCH /api/settings (max_logs: number |
null) and a NavMenu numeric field (blank = unlimited).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
2026-09-05 15:12:12 +01:00

1219 lines
49 KiB
Rust

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use crate::dto::{
BucketValuesParams, CreateDashboardRequest, CreateProfileRequest, CssEntryDto, DashboardDto,
ErrorDto, ImportCssRequest, LinkCssRequest, MetricDeclarationDto, ParamDeclarationDto,
LogsParams, ProfileSummaryDto, PromqlParams, SaveDashboardRequest, SettingsDto,
TimelineParams, UpdateSettingsRequest,
};
use crate::engine::{
DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery,
EngineResponse,
};
use crate::profile::{self, Profile, ProfileError, ProfileStore};
use crate::settings::{CssError, Settings};
use axum::extract::{Path as AxumPath, Query, State};
use axum::http::{header, StatusCode};
use axum::response::{IntoResponse, Json};
use axum::routing::{delete, get, post};
use axum::Router;
use tokio::sync::oneshot;
use tower_http::services::{ServeDir, ServeFile};
/// Shared, process-wide settings handle. Cheap to clone; lock is only held
/// briefly since settings changes are rare (unlike the engine's hot path).
pub type SettingsHandle = Arc<Mutex<Settings>>;
#[derive(Clone)]
struct ApiState {
engine: EngineHandle,
settings: SettingsHandle,
/// Path to the settings JSON file on disk. Imported CSS files are stored
/// alongside it, in its parent directory (see `ApiState::css_dir`).
settings_path: Arc<PathBuf>,
}
impl ApiState {
fn css_dir(&self) -> PathBuf {
self.settings_path
.parent()
.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(
engine: EngineHandle,
static_dir: &str,
settings: SettingsHandle,
settings_path: PathBuf,
) -> Router {
let state = ApiState {
engine,
settings,
settings_path: Arc::new(settings_path),
};
let index_html = format!("{static_dir}/index.html");
let static_service = ServeDir::new(static_dir).fallback(ServeFile::new(index_html));
Router::new()
.route("/api/logs", get(get_logs))
.route("/api/timeline", get(get_timeline))
.route("/api/stats", get(get_stats))
.route("/api/params", get(list_params).post(declare_param))
.route("/api/metrics", get(list_metrics).post(declare_metric))
.route("/api/metrics/{name}", get(get_metric_bucket_values))
.route("/api/query", get(run_promql_query))
.route("/api/settings", get(get_settings).patch(update_settings))
.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))
.route("/api/settings/css/{id}/content", get(get_css_content))
.route("/api/settings/profiles", get(list_profiles).post(create_profile))
.route("/api/settings/profiles/{name}", delete(delete_profile))
.route("/api/settings/profiles/{name}/activate", post(activate_profile))
.route(
"/api/settings/profiles/{name}/dashboards",
get(list_dashboards).post(create_dashboard),
)
.route(
"/api/settings/profiles/{name}/dashboards/{dashboard_id}",
get(get_dashboard).put(save_dashboard).delete(delete_dashboard),
)
.with_state(state)
.fallback_service(static_service)
}
async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result<EngineResponse, StatusCode> {
let (respond_to, response) = oneshot::channel();
engine
.send(EngineMessage::Query { query, respond_to })
.await
.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?;
response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)
}
/// `GET /api/logs` — also accepts a `filter` query param (ticket 007): a
/// FilterBar query string, parsed and evaluated inside the engine. A
/// malformed filter string comes back as `EngineResponse::Error` and is
/// surfaced here as `400 Bad Request` with `{ "error": "..." }`, matching
/// ticket 004's `declare_param`/`declare_metric` error-mapping pattern.
async fn get_logs(
State(state): State<ApiState>,
Query(params): Query<LogsParams>,
) -> impl IntoResponse {
match ask(&state.engine, EngineQuery::Logs(params)).await {
Ok(EngineResponse::Logs(entries)) => Json(entries).into_response(),
Ok(EngineResponse::Error(err)) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
}
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(),
}
}
/// `GET /api/timeline` — same `filter` query param and error mapping as
/// `get_logs` above.
async fn get_timeline(
State(state): State<ApiState>,
Query(params): Query<TimelineParams>,
) -> impl IntoResponse {
match ask(&state.engine, EngineQuery::Timeline(params)).await {
Ok(EngineResponse::Timeline(buckets)) => Json(buckets).into_response(),
Ok(EngineResponse::Error(err)) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
}
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(),
}
}
async fn get_stats(State(state): State<ApiState>) -> impl IntoResponse {
match ask(&state.engine, EngineQuery::Stats).await {
Ok(EngineResponse::Stats(stats)) => Json(stats).into_response(),
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(),
}
}
/// `GET /api/params` — lists all declared params (name, regex, capture
/// group, kind) as `Vec<ParamDeclarationDto>`. Mirrors `list_metrics` below
/// (ticket 015 needed this for LogPreview's column picker; ticket 004 only
/// exposed the equivalent for metrics).
async fn list_params(State(state): State<ApiState>) -> impl IntoResponse {
match ask(&state.engine, EngineQuery::ParamDeclarations).await {
Ok(EngineResponse::ParamDeclarations(declarations)) => Json(declarations).into_response(),
Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(),
Err(status) => status.into_response(),
}
}
/// `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(),
}
}
/// `GET /api/query` — evaluates a PromQL query (ticket 013; see `crate::promql` for
/// the supported grammar) against declared metrics. Query params: `promql` (the
/// query string), `start_ms`, `end_ms` (bound which points are included, mirroring
/// `/api/logs`/`/api/metrics/:name`). Returns `Vec<PromqlSeriesDto>` — chart-ready
/// `labels` + `(timestamp_ms, value)` series — on success (`200 OK`), or
/// `400 Bad Request` with `{ "error": "..." }` for a malformed query.
async fn run_promql_query(
State(state): State<ApiState>,
Query(params): Query<PromqlParams>,
) -> impl IntoResponse {
let query = EngineQuery::Promql {
query: params.promql,
start_ms: params.start_ms,
end_ms: params.end_ms,
};
match ask(&state.engine, query).await {
Ok(EngineResponse::Promql(Ok(series))) => Json(series).into_response(),
Ok(EngineResponse::Promql(Err(err))) => {
(StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response()
}
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` ->
/// 500.
fn css_error_response(err: CssError) -> axum::response::Response {
match err {
CssError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(),
CssError::NotImported => (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
CssError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
}
}
/// `GET /api/settings/css` — lists all configured CSS entries (imported and
/// linked) as `Vec<CssEntryDto>`.
async fn list_css(State(state): State<ApiState>) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let entries: Vec<CssEntryDto> = settings.css_entries.iter().map(CssEntryDto::from).collect();
Json(entries).into_response()
}
/// `POST /api/settings/css` — imports raw CSS content. Body:
/// `{ "filename": "theme.css", "content": "body { ... }" }`. Stores the
/// content as a file alongside the settings file, adds a `CssEntry::Imported`
/// to settings, and returns the new entry as `CssEntryDto` (201 Created).
async fn import_css(
State(state): State<ApiState>,
Json(body): Json<ImportCssRequest>,
) -> impl IntoResponse {
let css_dir = state.css_dir();
let mut settings = state.settings.lock().unwrap();
match settings.import_css(&state.settings_path, &css_dir, body.filename, &body.content) {
Ok(id) => {
let entry = settings
.css_entries
.iter()
.find(|entry| crate::settings::css_entry_id(entry) == id)
.expect("entry was just inserted by import_css");
(StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response()
}
Err(err) => css_error_response(err),
}
}
/// `POST /api/settings/css/link` — registers a reference to an external CSS
/// path/URL. Body: `{ "path_or_url": "https://example.com/style.css" }`.
/// Content is never copied; the entry is never deletable via
/// `DELETE /api/settings/css/:id`. Returns the new entry as `CssEntryDto`
/// (201 Created).
async fn link_css(
State(state): State<ApiState>,
Json(body): Json<LinkCssRequest>,
) -> impl IntoResponse {
let mut settings = state.settings.lock().unwrap();
match settings.link_css(&state.settings_path, body.path_or_url.clone()) {
Ok(()) => {
let entry = settings
.css_entries
.last()
.expect("entry was just pushed by link_css");
(StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response()
}
Err(err) => css_error_response(err),
}
}
/// `DELETE /api/settings/css/:id` — removes an imported CSS entry (file +
/// settings reference). Rejects with 400 if `id` refers to a `Linked` entry
/// (never deletable this way) and 404 if `id` is unknown. On success returns
/// 204 No Content.
async fn remove_css(
State(state): State<ApiState>,
AxumPath(id): AxumPath<String>,
) -> impl IntoResponse {
let css_dir = state.css_dir();
let mut settings = state.settings.lock().unwrap();
match settings.remove_css(&state.settings_path, &css_dir, &id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => css_error_response(err),
}
}
/// `GET /api/settings/css/:id/content` — serves back the raw CSS content of
/// an imported entry (`content-type: text/css`). Rejects with 400 for a
/// linked entry (there's no local content to serve — the frontend should use
/// `path_or_url` directly) and 404 for an unknown id.
async fn get_css_content(
State(state): State<ApiState>,
AxumPath(id): AxumPath<String>,
) -> impl IntoResponse {
let css_dir = state.css_dir();
let settings = state.settings.lock().unwrap();
match settings.css_content(&css_dir, &id) {
Ok(content) => (
StatusCode::OK,
[(header::CONTENT_TYPE, "text/css; charset=utf-8")],
content,
)
.into_response(),
Err(err) => css_error_response(err),
}
}
/// `GET /api/settings` — returns the subset of `Settings` the NavMenu UI
/// needs (colour scheme, active profile name) as `SettingsDto`. CSS entries
/// have their own dedicated list endpoint (`GET /api/settings/css`).
async fn get_settings(State(state): State<ApiState>) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
Json(SettingsDto::from(&*settings)).into_response()
}
/// `PATCH /api/settings` — updates process-wide settings. Body:
/// `{ "color_scheme": "light" | "dark" | "system", "max_logs": number |
/// null }` (omit a field to leave it unchanged; `max_logs: null` clears the
/// cap back to unlimited). Persists to `settings_path`, pushes a changed
/// `max_logs` to the engine so it takes effect on the next ingest without a
/// restart, and returns the updated settings as `SettingsDto`.
async fn update_settings(
State(state): State<ApiState>,
Json(body): Json<UpdateSettingsRequest>,
) -> impl IntoResponse {
let max_logs_changed = body.max_logs.is_some();
let new_max_logs = {
let mut settings = state.settings.lock().unwrap();
if let Some(color_scheme) = body.color_scheme {
settings.color_scheme = color_scheme;
}
if let Some(max_logs) = body.max_logs {
settings.max_logs = max_logs;
}
match settings.save(&state.settings_path) {
Ok(()) => (),
Err(err) => return (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
}
(SettingsDto::from(&*settings), settings.max_logs)
};
if max_logs_changed {
let _ = state
.engine
.send(EngineMessage::SetMaxLogs(new_max_logs.1))
.await;
}
Json(new_max_logs.0).into_response()
}
/// Maps a `ProfileError` to a response status/message: `NotFound` -> 404,
/// `DuplicateName`/`CannotDeleteDefault` -> 400, `Io` -> 500. Mirrors
/// `css_error_response` above.
fn profile_error_response(err: ProfileError) -> axum::response::Response {
match err {
ProfileError::NotFound => (StatusCode::NOT_FOUND, err.to_string()).into_response(),
ProfileError::DuplicateName => (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
ProfileError::CannotDeleteDefault => (StatusCode::BAD_REQUEST, err.to_string()).into_response(),
ProfileError::Io(_) => (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response(),
}
}
/// `GET /api/settings/profiles` — lists all known profiles with their
/// param/metric declaration and dashboard counts, as `Vec<ProfileSummaryDto>`.
/// Use `GET /api/settings`'s `active_profile` field to know which one (if
/// any) is currently active.
async fn list_profiles(State(state): State<ApiState>) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir));
let summaries: Vec<ProfileSummaryDto> = store.list().iter().map(ProfileSummaryDto::from).collect();
Json(summaries).into_response()
}
/// `POST /api/settings/profiles` — creates a new, empty profile. Body:
/// `{ "name": "..." }`. Returns `201 Created` with the new profile as
/// `ProfileSummaryDto`, or `400 Bad Request` if a profile with that name
/// already exists.
async fn create_profile(
State(state): State<ApiState>,
Json(body): Json<CreateProfileRequest>,
) -> impl IntoResponse {
let mut settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let mut store = ProfileStore::load(&store_path);
match store.create(&store_path, &mut settings, &state.settings_path, body.name.clone()) {
Ok(()) => {
let profile = store.get(&body.name).expect("profile was just created by store.create");
(StatusCode::CREATED, Json(ProfileSummaryDto::from(profile))).into_response()
}
Err(err) => profile_error_response(err),
}
}
/// `DELETE /api/settings/profiles/:name` — deletes a profile. If it was the
/// active profile, `active_profile` is cleared (but the engine's currently
/// declared params/metrics are left as-is — they simply won't be reloaded on
/// the next restart). Returns `204 No Content`, or `404 Not Found` if no
/// profile with that name exists.
async fn delete_profile(
State(state): State<ApiState>,
AxumPath(name): AxumPath<String>,
) -> impl IntoResponse {
let mut settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let mut store = ProfileStore::load(&store_path);
match store.delete(&store_path, &mut settings, &state.settings_path, &name) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => profile_error_response(err),
}
}
/// Removes every currently-declared param/metric from the running engine
/// (via `EngineMessage::RemoveParamDeclaration`/`RemoveMetricDeclaration`),
/// then replays `profile`'s stored declarations into it (see
/// `profile::load_into_engine`). Used by `activate_profile` so switching the
/// active profile takes effect on the running engine immediately, without
/// requiring a process restart.
async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) {
if let Ok(EngineResponse::ParamDeclarations(params)) =
ask(engine, EngineQuery::ParamDeclarations).await
{
for param in params {
let (respond_to, response) = oneshot::channel();
if engine
.send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to })
.await
.is_ok()
{
let _ = response.await;
}
}
}
if let Ok(EngineResponse::MetricDeclarations(metrics)) =
ask(engine, EngineQuery::MetricDeclarations).await
{
for metric in metrics {
let (respond_to, response) = oneshot::channel();
if engine
.send(EngineMessage::RemoveMetricDeclaration { name: metric.name, respond_to })
.await
.is_ok()
{
let _ = response.await;
}
}
}
profile::load_into_engine(engine, profile).await;
}
/// `POST /api/settings/profiles/:name/activate` — sets `name` as the active
/// profile (persisted, see `ProfileStore::set_active`) and hot-swaps the
/// running engine's declared params/metrics to match it (see
/// `reload_engine_declarations`), so the switch is visible without a
/// restart. Returns the updated settings as `SettingsDto`, or `404 Not
/// Found` if no profile with that name exists.
async fn activate_profile(
State(state): State<ApiState>,
AxumPath(name): AxumPath<String>,
) -> impl IntoResponse {
let activated_profile = {
let mut settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let store = ProfileStore::load(&store_path);
match store.set_active(&mut settings, &state.settings_path, Some(name.clone())) {
Ok(()) => Ok(store.get(&name).cloned()),
Err(err) => Err(err),
}
};
match activated_profile {
Ok(profile) => {
if let Some(profile) = profile {
reload_engine_declarations(&state.engine, &profile).await;
}
let settings = state.settings.lock().unwrap();
Json(SettingsDto::from(&*settings)).into_response()
}
Err(err) => profile_error_response(err),
}
}
/// `GET /api/settings/profiles/:name/dashboards` — lists all dashboards saved
/// on the profile `name`, as `Vec<DashboardDto>`. Returns `404 Not Found` if
/// no profile with that name exists.
async fn list_dashboards(
State(state): State<ApiState>,
AxumPath(name): AxumPath<String>,
) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir));
match store.list_dashboards(&name) {
Ok(dashboards) => {
let dtos: Vec<DashboardDto> = dashboards.iter().map(DashboardDto::from).collect();
Json(dtos).into_response()
}
Err(err) => profile_error_response(err),
}
}
/// `POST /api/settings/profiles/:name/dashboards` — creates a new, empty
/// dashboard on profile `name`. Body: `{ "name": "..." }`. Returns
/// `201 Created` with the new dashboard as `DashboardDto`, or `404 Not
/// Found` if no profile with that name exists.
async fn create_dashboard(
State(state): State<ApiState>,
AxumPath(name): AxumPath<String>,
Json(body): Json<CreateDashboardRequest>,
) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let mut store = ProfileStore::load(&store_path);
match store.create_dashboard(&store_path, &name, body.name) {
Ok(dashboard) => (StatusCode::CREATED, Json(DashboardDto::from(&dashboard))).into_response(),
Err(err) => profile_error_response(err),
}
}
/// `GET /api/settings/profiles/:name/dashboards/:dashboard_id` — fetches one
/// dashboard by id. Returns `404 Not Found` if the profile or dashboard
/// doesn't exist.
async fn get_dashboard(
State(state): State<ApiState>,
AxumPath((name, dashboard_id)): AxumPath<(String, String)>,
) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir));
match store.list_dashboards(&name) {
Ok(dashboards) => match dashboards.iter().find(|d| d.id == dashboard_id) {
Some(dashboard) => Json(DashboardDto::from(dashboard)).into_response(),
None => profile_error_response(ProfileError::NotFound),
},
Err(err) => profile_error_response(err),
}
}
/// `PUT /api/settings/profiles/:name/dashboards/:dashboard_id` — replaces the
/// dashboard's name/panels. Body: `{ "name": "...", "panels": [{ "id": "...",
/// "title": "...", "promql": "...", "chart_type": "bar" | "line" | "pie" }] }`.
/// Returns the updated dashboard as `DashboardDto`, or `404 Not Found` if the
/// profile or dashboard doesn't exist.
async fn save_dashboard(
State(state): State<ApiState>,
AxumPath((name, dashboard_id)): AxumPath<(String, String)>,
Json(body): Json<SaveDashboardRequest>,
) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let mut store = ProfileStore::load(&store_path);
let dashboard = crate::profile::Dashboard {
id: dashboard_id.clone(),
name: body.name,
panels: body.panels.iter().map(|p| p.into()).collect(),
};
match store.save_dashboard(&store_path, &name, &dashboard_id, dashboard) {
Ok(dashboard) => Json(DashboardDto::from(&dashboard)).into_response(),
Err(err) => profile_error_response(err),
}
}
/// `DELETE /api/settings/profiles/:name/dashboards/:dashboard_id` — deletes a
/// dashboard. Returns `204 No Content`, or `404 Not Found` if the profile or
/// dashboard doesn't exist.
async fn delete_dashboard(
State(state): State<ApiState>,
AxumPath((name, dashboard_id)): AxumPath<(String, String)>,
) -> impl IntoResponse {
let settings = state.settings.lock().unwrap();
let profiles_dir = profile::resolve_profiles_dir(&settings, &state.settings_path);
let store_path = profile::profiles_store_path(&profiles_dir);
let mut store = ProfileStore::load(&store_path);
match store.delete_dashboard(&store_path, &name, &dashboard_id) {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => profile_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 promql_query_returns_series_for_seeded_fixture() {
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/query?promql=requests%7Bpath%3D%22%2Fa%22%7D").await;
assert_eq!(status, StatusCode::OK);
let series = body.as_array().unwrap();
assert_eq!(series.len(), 1);
assert_eq!(series[0]["labels"]["path"], "/a");
let points = series[0]["points"].as_array().unwrap();
assert_eq!(points.len(), 2);
assert_eq!(points[0], json!([0, 1.0]));
assert_eq!(points[1], json!([2000, 1.0]));
}
#[tokio::test]
async fn promql_query_rejects_malformed_query_with_400() {
let (app, _engine, _settings, _dir) = test_app();
let (status, body) = get_json(&app, "/api/query?promql=not+valid(").await;
assert_eq!(status, StatusCode::BAD_REQUEST);
assert!(body["error"].as_str().is_some_and(|s| !s.is_empty()));
}
#[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");
}
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()
}
async fn get_status(app: &Router, uri: &str) -> StatusCode {
let request = Request::builder().method("GET").uri(uri).body(Body::empty()).unwrap();
app.clone().oneshot(request).await.unwrap().status()
}
#[tokio::test]
async fn dashboard_crud_round_trips_through_http() {
let (app, _engine, _settings, _dir) = test_app();
let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await;
assert_eq!(status, StatusCode::CREATED);
let (status, body) = send_json(
&app,
"POST",
"/api/settings/profiles/web/dashboards",
json!({ "name": "Overview" }),
)
.await;
assert_eq!(status, StatusCode::CREATED);
assert_eq!(body["name"], "Overview");
assert!(body["panels"].as_array().unwrap().is_empty());
let dashboard_id = body["id"].as_str().unwrap().to_string();
// Listing shows the new (empty) dashboard.
let (status, body) = get_json(&app, "/api/settings/profiles/web/dashboards").await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body.as_array().unwrap().len(), 1);
// Save adds a panel.
let (status, body) = send_json(
&app,
"PUT",
&format!("/api/settings/profiles/web/dashboards/{dashboard_id}"),
json!({
"name": "Overview",
"panels": [
{ "id": "p1", "title": "Requests", "promql": "requests", "chart_type": "line" }
]
}),
)
.await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["id"], dashboard_id);
let panels = body["panels"].as_array().unwrap();
assert_eq!(panels.len(), 1);
assert_eq!(panels[0]["chart_type"], "line");
// Get by id reflects the saved panel.
let (status, body) =
get_json(&app, &format!("/api/settings/profiles/web/dashboards/{dashboard_id}")).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["panels"].as_array().unwrap().len(), 1);
// Delete removes it.
let status = delete_request(&app, &format!("/api/settings/profiles/web/dashboards/{dashboard_id}")).await;
assert_eq!(status, StatusCode::NO_CONTENT);
let (status, body) = get_json(&app, "/api/settings/profiles/web/dashboards").await;
assert_eq!(status, StatusCode::OK);
assert!(body.as_array().unwrap().is_empty());
}
#[tokio::test]
async fn dashboard_routes_404_for_unknown_profile_or_dashboard() {
let (app, _engine, _settings, _dir) = test_app();
let status = get_status(&app, "/api/settings/profiles/ghost/dashboards").await;
assert_eq!(status, StatusCode::NOT_FOUND);
let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await;
assert_eq!(status, StatusCode::CREATED);
let status = get_status(&app, "/api/settings/profiles/web/dashboards/does-not-exist").await;
assert_eq!(status, StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn saved_dashboard_persists_across_simulated_restart() {
let (app, _engine, settings_handle, dir) = test_app();
let (status, _body) = send_json(&app, "POST", "/api/settings/profiles", json!({ "name": "web" })).await;
assert_eq!(status, StatusCode::CREATED);
let (status, body) = send_json(
&app,
"POST",
"/api/settings/profiles/web/dashboards",
json!({ "name": "Overview" }),
)
.await;
assert_eq!(status, StatusCode::CREATED);
let dashboard_id = body["id"].as_str().unwrap().to_string();
send_json(
&app,
"PUT",
&format!("/api/settings/profiles/web/dashboards/{dashboard_id}"),
json!({
"name": "Overview",
"panels": [
{ "id": "p1", "title": "Requests", "promql": "requests", "chart_type": "bar" }
]
}),
)
.await;
// Simulate a restart: reload the profile store from disk directly,
// independent of the running app/engine.
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 reloaded_store = ProfileStore::load(&store_path);
let dashboards = reloaded_store.list_dashboards("web").unwrap();
assert_eq!(dashboards.len(), 1);
assert_eq!(dashboards[0].name, "Overview");
assert_eq!(dashboards[0].panels.len(), 1);
assert_eq!(dashboards[0].panels[0].promql, "requests");
}
async fn ingest_lines(engine: &EngineHandle, start_sequence: u64, texts: &[&str]) {
for (offset, text) in texts.iter().enumerate() {
let sequence = start_sequence + offset as u64;
engine
.send(EngineMessage::Ingest(IngestedLine {
sequence,
timestamp: Timestamp::from_millis(sequence as i64 * 1000),
text: text.to_string(),
}))
.await
.unwrap();
}
}
#[tokio::test]
async fn setting_max_logs_evicts_oldest_on_next_ingest() {
let (app, engine, _settings, _dir) = test_app();
ingest_lines(&engine, 0, &["a", "b", "c"]).await;
let (status, body) = get_json(&app, "/api/logs").await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body.as_array().unwrap().len(), 3);
let (status, body) = send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": 2 })).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["max_logs"], 2);
// Existing over-cap logs are evicted as soon as the setting change
// reaches the engine, not just on the next ingest.
let (_, body) = get_json(&app, "/api/logs").await;
assert_eq!(body.as_array().unwrap().len(), 2);
ingest_lines(&engine, 3, &["d"]).await;
let (_, body) = get_json(&app, "/api/logs").await;
let texts: Vec<_> = body
.as_array()
.unwrap()
.iter()
.map(|entry| entry["text"].as_str().unwrap())
.collect();
assert_eq!(texts, vec!["c", "d"]);
}
#[tokio::test]
async fn max_logs_defaults_to_unlimited() {
let (app, engine, _settings, _dir) = test_app();
let (_, body) = get_json(&app, "/api/settings").await;
assert_eq!(body["max_logs"], Value::Null);
ingest_lines(&engine, 0, &["a", "b", "c", "d", "e"]).await;
let (_, body) = get_json(&app, "/api/logs").await;
assert_eq!(body.as_array().unwrap().len(), 5);
}
#[tokio::test]
async fn max_logs_clears_back_to_unlimited() {
let (app, engine, _settings, _dir) = test_app();
send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": 1 })).await;
ingest_lines(&engine, 0, &["a", "b"]).await;
let (_, body) = get_json(&app, "/api/logs").await;
assert_eq!(body.as_array().unwrap().len(), 1);
let (status, body) = send_json(&app, "PATCH", "/api/settings", json!({ "max_logs": null })).await;
assert_eq!(status, StatusCode::OK);
assert_eq!(body["max_logs"], Value::Null);
ingest_lines(&engine, 2, &["c", "d"]).await;
let (_, body) = get_json(&app, "/api/logs").await;
assert_eq!(body.as_array().unwrap().len(), 3);
}
}