Merge branch 'worktree-agent-a050e355d6c42f2c2'
This commit is contained in:
+34
@@ -7,6 +7,7 @@ mod filter_parser;
|
|||||||
mod index_key;
|
mod index_key;
|
||||||
mod ingest;
|
mod ingest;
|
||||||
mod metric_primitive;
|
mod metric_primitive;
|
||||||
|
mod profile;
|
||||||
mod settings;
|
mod settings;
|
||||||
mod timestamp;
|
mod timestamp;
|
||||||
|
|
||||||
@@ -17,6 +18,7 @@ use clap::Parser;
|
|||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
|
|
||||||
|
use profile::ProfileStore;
|
||||||
use settings::Settings;
|
use settings::Settings;
|
||||||
|
|
||||||
/// Observability dashboard from terminal output.
|
/// Observability dashboard from terminal output.
|
||||||
@@ -48,6 +50,38 @@ async fn main() {
|
|||||||
|
|
||||||
let (engine_tx, engine_rx) = mpsc::channel(1024);
|
let (engine_tx, engine_rx) = mpsc::channel(1024);
|
||||||
tokio::spawn(engine::run(engine_rx));
|
tokio::spawn(engine::run(engine_rx));
|
||||||
|
|
||||||
|
// Ticket 009: reload the active profile's param/metric declarations into the
|
||||||
|
// engine BEFORE starting ingest below. Ordering matters: declarations must be
|
||||||
|
// in place before any line is ingested so that backfill/live-derivation
|
||||||
|
// semantics (ticket 003) apply uniformly to every line, including ones that
|
||||||
|
// arrive immediately once ingest starts (there's no "declare after the fact
|
||||||
|
// and go back to reprocess early lines" step, so doing it after ingest starts
|
||||||
|
// could silently miss lines that raced ahead of the declarations).
|
||||||
|
{
|
||||||
|
let (active_profile_name, profiles_dir) = {
|
||||||
|
let settings_guard = settings.lock().unwrap();
|
||||||
|
(
|
||||||
|
settings_guard.profiles.active_profile.clone(),
|
||||||
|
profile::resolve_profiles_dir(&settings_guard, &settings_path),
|
||||||
|
)
|
||||||
|
};
|
||||||
|
if let Some(profile_name) = active_profile_name {
|
||||||
|
let store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir));
|
||||||
|
match store.get(&profile_name) {
|
||||||
|
Some(active_profile) => {
|
||||||
|
profile::load_into_engine(&engine_tx, active_profile).await;
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
eprintln!(
|
||||||
|
"warning: active profile '{profile_name}' not found under {}; skipping",
|
||||||
|
profiles_dir.display()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
tokio::spawn(ingest::run(engine_tx.clone()));
|
tokio::spawn(ingest::run(engine_tx.clone()));
|
||||||
|
|
||||||
let app = api::router(engine_tx, &args.static_dir, settings, settings_path);
|
let app = api::router(engine_tx, &args.static_dir, settings, settings_path);
|
||||||
|
|||||||
+761
@@ -0,0 +1,761 @@
|
|||||||
|
//! Named, persistable bundles of parsing/indexing rules ("profiles" / "presets"),
|
||||||
|
//! ticket 009.
|
||||||
|
//!
|
||||||
|
//! A `Profile` bundles the param declarations (ticket 001) and metric
|
||||||
|
//! declarations (ticket 002) needed to derive a given CLI app's structure from
|
||||||
|
//! its raw log output, plus any saved dashboards for that app. Profiles are
|
||||||
|
//! collected in a `ProfileStore`, persisted as a single `profiles.json` file
|
||||||
|
//! under `Settings.profiles.profiles_dir` (resolved relative to the settings
|
||||||
|
//! file's directory — the same convention ticket 010 uses for CSS storage). A
|
||||||
|
//! single file (rather than one file per profile) is simplest and avoids
|
||||||
|
//! filename-sanitization concerns for profile names.
|
||||||
|
//!
|
||||||
|
//! `Settings.profiles` (see `settings.rs`) only tracks lightweight pointers —
|
||||||
|
//! known profile names and which one is active — while this module owns the
|
||||||
|
//! actual declaration content. That mirrors how `Settings.css_entries` holds
|
||||||
|
//! lightweight `CssEntry` records while the CSS content itself lives in
|
||||||
|
//! separate files (ticket 010).
|
||||||
|
//!
|
||||||
|
//! On startup, `main.rs` loads the active profile's declarations into a fresh
|
||||||
|
//! engine by replaying `EngineMessage::DeclareParam`/`DeclareMetric` for each
|
||||||
|
//! stored declaration (see `load_into_engine` below) — this reproduces the
|
||||||
|
//! same parsing/indexing rules the profile had when last saved, without
|
||||||
|
//! requiring the user to re-declare anything by hand after a restart.
|
||||||
|
//!
|
||||||
|
//! Several public methods here (e.g. `ProfileStore::delete`,
|
||||||
|
//! `record_param_declaration`) have no caller yet within this worktree — CRUD
|
||||||
|
//! over profiles isn't exposed over HTTP until ticket 004 lands the API layer
|
||||||
|
//! and calls into this module. `#![allow(dead_code)]` mirrors how
|
||||||
|
//! `engine.rs`/`data_value.rs` handle the same not-yet-wired situation for
|
||||||
|
//! tickets 001/002.
|
||||||
|
#![allow(dead_code)]
|
||||||
|
|
||||||
|
use std::fmt;
|
||||||
|
use std::fs;
|
||||||
|
use std::io;
|
||||||
|
use std::path::{Path, PathBuf};
|
||||||
|
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
use crate::data_value::{MetricPrimitiveKind, RegexToMetricPrimitiveMatcher};
|
||||||
|
use crate::engine::{DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage};
|
||||||
|
use crate::settings::Settings;
|
||||||
|
use tokio::sync::oneshot;
|
||||||
|
|
||||||
|
/// Mirrors `data_value::MetricPrimitiveKind` with `Serialize`/`Deserialize` so
|
||||||
|
/// profiles can round-trip through JSON. `MetricPrimitiveKind` itself only
|
||||||
|
/// derives `Serialize` (it's an engine-internal type owned by tickets
|
||||||
|
/// 001/002/003, out of scope for this ticket to modify), so this module keeps
|
||||||
|
/// its own mirror and converts at the boundary.
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
#[serde(rename_all = "lowercase")]
|
||||||
|
pub enum ProfileMetricPrimitiveKind {
|
||||||
|
String,
|
||||||
|
Number,
|
||||||
|
Timestamp,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<MetricPrimitiveKind> for ProfileMetricPrimitiveKind {
|
||||||
|
fn from(kind: MetricPrimitiveKind) -> Self {
|
||||||
|
match kind {
|
||||||
|
MetricPrimitiveKind::String => ProfileMetricPrimitiveKind::String,
|
||||||
|
MetricPrimitiveKind::Number => ProfileMetricPrimitiveKind::Number,
|
||||||
|
MetricPrimitiveKind::Timestamp => ProfileMetricPrimitiveKind::Timestamp,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<ProfileMetricPrimitiveKind> for MetricPrimitiveKind {
|
||||||
|
fn from(kind: ProfileMetricPrimitiveKind) -> Self {
|
||||||
|
match kind {
|
||||||
|
ProfileMetricPrimitiveKind::String => MetricPrimitiveKind::String,
|
||||||
|
ProfileMetricPrimitiveKind::Number => MetricPrimitiveKind::Number,
|
||||||
|
ProfileMetricPrimitiveKind::Timestamp => MetricPrimitiveKind::Timestamp,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A stored param declaration, matching `DeclareParamRequest`'s shape (ticket 001).
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub struct ParamDeclarationRecord {
|
||||||
|
pub name: String,
|
||||||
|
pub message_match_regex: String,
|
||||||
|
pub capture_group: String,
|
||||||
|
pub kind: ProfileMetricPrimitiveKind,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<&DeclareParamRequest> for ParamDeclarationRecord {
|
||||||
|
fn from(request: &DeclareParamRequest) -> Self {
|
||||||
|
ParamDeclarationRecord {
|
||||||
|
name: request.name.clone(),
|
||||||
|
message_match_regex: request.message_match_regex.clone(),
|
||||||
|
capture_group: request.capture_group.clone(),
|
||||||
|
kind: request.kind.into(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<&ParamDeclarationRecord> for DeclareParamRequest {
|
||||||
|
fn from(record: &ParamDeclarationRecord) -> Self {
|
||||||
|
DeclareParamRequest {
|
||||||
|
name: record.name.clone(),
|
||||||
|
message_match_regex: record.message_match_regex.clone(),
|
||||||
|
capture_group: record.capture_group.clone(),
|
||||||
|
kind: record.kind.into(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A stored metric param matcher, matching `RegexToMetricPrimitiveMatcher`'s shape.
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub struct MetricParamRecord {
|
||||||
|
pub param_name: String,
|
||||||
|
pub capture_group: String,
|
||||||
|
pub kind: ProfileMetricPrimitiveKind,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A stored metric declaration, matching `DeclareMetricRequest`'s shape (ticket 002).
|
||||||
|
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
pub struct MetricDeclarationRecord {
|
||||||
|
pub name: String,
|
||||||
|
pub message_match_regex: String,
|
||||||
|
pub params: Vec<MetricParamRecord>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<&DeclareMetricRequest> for MetricDeclarationRecord {
|
||||||
|
fn from(request: &DeclareMetricRequest) -> Self {
|
||||||
|
MetricDeclarationRecord {
|
||||||
|
name: request.name.clone(),
|
||||||
|
message_match_regex: request.message_match_regex.clone(),
|
||||||
|
params: request
|
||||||
|
.params
|
||||||
|
.iter()
|
||||||
|
.map(|matcher| MetricParamRecord {
|
||||||
|
param_name: matcher.param_name.clone(),
|
||||||
|
capture_group: matcher.capture_group.clone(),
|
||||||
|
kind: matcher.kind.into(),
|
||||||
|
})
|
||||||
|
.collect(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl From<&MetricDeclarationRecord> for DeclareMetricRequest {
|
||||||
|
fn from(record: &MetricDeclarationRecord) -> Self {
|
||||||
|
DeclareMetricRequest {
|
||||||
|
name: record.name.clone(),
|
||||||
|
message_match_regex: record.message_match_regex.clone(),
|
||||||
|
params: record
|
||||||
|
.params
|
||||||
|
.iter()
|
||||||
|
.map(|param| RegexToMetricPrimitiveMatcher {
|
||||||
|
param_name: param.param_name.clone(),
|
||||||
|
capture_group: param.capture_group.clone(),
|
||||||
|
kind: param.kind.into(),
|
||||||
|
})
|
||||||
|
.collect(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A named bundle of parsing/indexing rules for a particular CLI app: which
|
||||||
|
/// params and metrics to derive from its log output, plus any saved
|
||||||
|
/// dashboards.
|
||||||
|
///
|
||||||
|
/// `dashboards` is stubbed as opaque JSON: dashboards (ticket 014) don't exist
|
||||||
|
/// yet, so there's no concrete schema to store. Ticket 014 should replace this
|
||||||
|
/// with whatever concrete `Dashboard` type it introduces (or keep it loosely
|
||||||
|
/// typed, if that turns out to be the right call for that ticket).
|
||||||
|
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||||
|
pub struct Profile {
|
||||||
|
pub name: String,
|
||||||
|
#[serde(default)]
|
||||||
|
pub params: Vec<ParamDeclarationRecord>,
|
||||||
|
#[serde(default)]
|
||||||
|
pub metrics: Vec<MetricDeclarationRecord>,
|
||||||
|
/// Placeholder for ticket 014's dashboard storage. See the struct doc above.
|
||||||
|
#[serde(default)]
|
||||||
|
pub dashboards: Vec<serde_json::Value>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Profile {
|
||||||
|
/// A new, empty profile with the given name.
|
||||||
|
pub fn new(name: String) -> Profile {
|
||||||
|
Profile {
|
||||||
|
name,
|
||||||
|
params: Vec::new(),
|
||||||
|
metrics: Vec::new(),
|
||||||
|
dashboards: Vec::new(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sends every param/metric declaration stored in `profile` to the engine at
|
||||||
|
/// `engine`, as `EngineMessage::DeclareParam`/`DeclareMetric`. Used on startup
|
||||||
|
/// to reload the active profile's rules into a fresh engine (see `main.rs`),
|
||||||
|
/// and in tests to simulate that reload.
|
||||||
|
///
|
||||||
|
/// A stored declaration should generally already be valid — it round-tripped
|
||||||
|
/// through the engine's own validation when originally declared (see
|
||||||
|
/// `record_param_declaration`/`record_metric_declaration` below) — but this
|
||||||
|
/// doesn't assume that: any declaration the engine rejects (or any send/recv
|
||||||
|
/// failure, e.g. the engine channel closing) is logged to stderr and skipped
|
||||||
|
/// rather than panicking, so one bad stored declaration can't take the whole
|
||||||
|
/// process down.
|
||||||
|
pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) {
|
||||||
|
for record in &profile.params {
|
||||||
|
let (respond_to, response) = oneshot::channel();
|
||||||
|
let request: DeclareParamRequest = record.into();
|
||||||
|
let name = request.name.clone();
|
||||||
|
if engine
|
||||||
|
.send(EngineMessage::DeclareParam { request, respond_to })
|
||||||
|
.await
|
||||||
|
.is_err()
|
||||||
|
{
|
||||||
|
eprintln!("warning: engine unavailable while loading profile '{}'", profile.name);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
match response.await {
|
||||||
|
Ok(Ok(())) => {}
|
||||||
|
Ok(Err(err)) => eprintln!(
|
||||||
|
"warning: profile '{}' param declaration '{name}' rejected by engine: {err}",
|
||||||
|
profile.name
|
||||||
|
),
|
||||||
|
Err(_) => eprintln!(
|
||||||
|
"warning: engine dropped response while declaring param '{name}' from profile '{}'",
|
||||||
|
profile.name
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
for record in &profile.metrics {
|
||||||
|
let (respond_to, response) = oneshot::channel();
|
||||||
|
let request: DeclareMetricRequest = record.into();
|
||||||
|
let name = request.name.clone();
|
||||||
|
if engine
|
||||||
|
.send(EngineMessage::DeclareMetric { request, respond_to })
|
||||||
|
.await
|
||||||
|
.is_err()
|
||||||
|
{
|
||||||
|
eprintln!("warning: engine unavailable while loading profile '{}'", profile.name);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
match response.await {
|
||||||
|
Ok(Ok(())) => {}
|
||||||
|
Ok(Err(err)) => eprintln!(
|
||||||
|
"warning: profile '{}' metric declaration '{name}' rejected by engine: {err}",
|
||||||
|
profile.name
|
||||||
|
),
|
||||||
|
Err(_) => eprintln!(
|
||||||
|
"warning: engine dropped response while declaring metric '{name}' from profile '{}'",
|
||||||
|
profile.name
|
||||||
|
),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Error returned by `ProfileStore` mutations and lookups.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub enum ProfileError {
|
||||||
|
/// No profile with the given name exists.
|
||||||
|
NotFound,
|
||||||
|
/// A profile with the given name already exists.
|
||||||
|
DuplicateName,
|
||||||
|
/// Underlying I/O failure reading/writing the profile store or settings file.
|
||||||
|
Io(io::Error),
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for ProfileError {
|
||||||
|
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
match self {
|
||||||
|
ProfileError::NotFound => write!(f, "no profile with that name"),
|
||||||
|
ProfileError::DuplicateName => write!(f, "a profile with that name already exists"),
|
||||||
|
ProfileError::Io(err) => write!(f, "I/O error: {err}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::error::Error for ProfileError {}
|
||||||
|
|
||||||
|
impl From<io::Error> for ProfileError {
|
||||||
|
fn from(err: io::Error) -> Self {
|
||||||
|
ProfileError::Io(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Resolves `settings.profiles.profiles_dir` relative to `settings_path`'s
|
||||||
|
/// parent directory (or treats it as absolute if it already is one) — the
|
||||||
|
/// same directory convention ticket 010 uses for CSS storage.
|
||||||
|
pub fn resolve_profiles_dir(settings: &Settings, settings_path: &Path) -> PathBuf {
|
||||||
|
let profiles_dir = Path::new(&settings.profiles.profiles_dir);
|
||||||
|
if profiles_dir.is_absolute() {
|
||||||
|
profiles_dir.to_path_buf()
|
||||||
|
} else {
|
||||||
|
settings_path
|
||||||
|
.parent()
|
||||||
|
.map(|parent| parent.join(profiles_dir))
|
||||||
|
.unwrap_or_else(|| profiles_dir.to_path_buf())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Path to the single file holding all profiles, within `profiles_dir` (see
|
||||||
|
/// `resolve_profiles_dir`).
|
||||||
|
pub fn profiles_store_path(profiles_dir: &Path) -> PathBuf {
|
||||||
|
profiles_dir.join("profiles.json")
|
||||||
|
}
|
||||||
|
|
||||||
|
/// On-disk collection of all known profiles, persisted as a single JSON file
|
||||||
|
/// (see module docs for why one file rather than one-per-profile).
|
||||||
|
#[derive(Debug, Clone, Default)]
|
||||||
|
pub struct ProfileStore {
|
||||||
|
profiles: Vec<Profile>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ProfileStore {
|
||||||
|
/// Loads the store from `store_path`. A missing file is treated as an
|
||||||
|
/// empty store (not an error) — there's no reason to fail startup just
|
||||||
|
/// because no profiles have been created yet. Any I/O or parse error
|
||||||
|
/// while reading an existing file is logged to stderr and treated as an
|
||||||
|
/// empty store, mirroring `Settings::load`'s non-fatal-error handling.
|
||||||
|
pub fn load(store_path: &Path) -> ProfileStore {
|
||||||
|
match fs::read_to_string(store_path) {
|
||||||
|
Ok(contents) => match serde_json::from_str::<Vec<Profile>>(&contents) {
|
||||||
|
Ok(profiles) => ProfileStore { profiles },
|
||||||
|
Err(err) => {
|
||||||
|
eprintln!(
|
||||||
|
"warning: failed to parse profile store at {}: {err}; using empty store",
|
||||||
|
store_path.display()
|
||||||
|
);
|
||||||
|
ProfileStore::default()
|
||||||
|
}
|
||||||
|
},
|
||||||
|
Err(err) if err.kind() == io::ErrorKind::NotFound => ProfileStore::default(),
|
||||||
|
Err(err) => {
|
||||||
|
eprintln!(
|
||||||
|
"warning: failed to read profile store at {}: {err}; using empty store",
|
||||||
|
store_path.display()
|
||||||
|
);
|
||||||
|
ProfileStore::default()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Persists the store to `store_path` as pretty-printed JSON, creating
|
||||||
|
/// any missing parent directories first.
|
||||||
|
pub fn save(&self, store_path: &Path) -> io::Result<()> {
|
||||||
|
if let Some(parent) = store_path.parent() {
|
||||||
|
fs::create_dir_all(parent)?;
|
||||||
|
}
|
||||||
|
let contents = serde_json::to_string_pretty(&self.profiles)
|
||||||
|
.map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
|
||||||
|
fs::write(store_path, contents)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// All known profiles.
|
||||||
|
pub fn list(&self) -> &[Profile] {
|
||||||
|
&self.profiles
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Looks up a profile by name.
|
||||||
|
pub fn get(&self, name: &str) -> Option<&Profile> {
|
||||||
|
self.profiles.iter().find(|p| p.name == name)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Creates a new, empty profile named `name`, persists the store to
|
||||||
|
/// `store_path`, and records the name in `settings.profiles.names`
|
||||||
|
/// (persisting `settings` to `settings_path` too, keeping the two files in
|
||||||
|
/// sync). Fails with `ProfileError::DuplicateName` if a profile with that
|
||||||
|
/// name already exists.
|
||||||
|
pub fn create(
|
||||||
|
&mut self,
|
||||||
|
store_path: &Path,
|
||||||
|
settings: &mut Settings,
|
||||||
|
settings_path: &Path,
|
||||||
|
name: String,
|
||||||
|
) -> Result<(), ProfileError> {
|
||||||
|
if self.get(&name).is_some() {
|
||||||
|
return Err(ProfileError::DuplicateName);
|
||||||
|
}
|
||||||
|
self.profiles.push(Profile::new(name.clone()));
|
||||||
|
self.save(store_path)?;
|
||||||
|
|
||||||
|
if !settings.profiles.names.contains(&name) {
|
||||||
|
settings.profiles.names.push(name);
|
||||||
|
}
|
||||||
|
settings.save(settings_path)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Deletes the profile named `name`, persists the store to `store_path`,
|
||||||
|
/// and drops it from `settings.profiles.names` (persisting `settings` to
|
||||||
|
/// `settings_path` too). If the deleted profile was active, clears
|
||||||
|
/// `settings.profiles.active_profile`. Fails with `ProfileError::NotFound`
|
||||||
|
/// if no such profile exists.
|
||||||
|
pub fn delete(
|
||||||
|
&mut self,
|
||||||
|
store_path: &Path,
|
||||||
|
settings: &mut Settings,
|
||||||
|
settings_path: &Path,
|
||||||
|
name: &str,
|
||||||
|
) -> Result<(), ProfileError> {
|
||||||
|
let index = self.profiles.iter().position(|p| p.name == name).ok_or(ProfileError::NotFound)?;
|
||||||
|
self.profiles.remove(index);
|
||||||
|
self.save(store_path)?;
|
||||||
|
|
||||||
|
settings.profiles.names.retain(|n| n != name);
|
||||||
|
if settings.profiles.active_profile.as_deref() == Some(name) {
|
||||||
|
settings.profiles.active_profile = None;
|
||||||
|
}
|
||||||
|
settings.save(settings_path)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sets (or clears, with `name: None`) the active profile: updates
|
||||||
|
/// `settings.profiles.active_profile` and persists `settings` to
|
||||||
|
/// `settings_path`. Fails with `ProfileError::NotFound` if `name` is
|
||||||
|
/// `Some` and no such profile exists in this store.
|
||||||
|
///
|
||||||
|
/// This only updates `Settings`; it does not itself reload the newly
|
||||||
|
/// active profile's declarations into a running engine (that happens on
|
||||||
|
/// the next startup — see `load_into_engine` and `main.rs`).
|
||||||
|
pub fn set_active(
|
||||||
|
&self,
|
||||||
|
settings: &mut Settings,
|
||||||
|
settings_path: &Path,
|
||||||
|
name: Option<String>,
|
||||||
|
) -> Result<(), ProfileError> {
|
||||||
|
if let Some(name) = &name {
|
||||||
|
if self.get(name).is_none() {
|
||||||
|
return Err(ProfileError::NotFound);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
settings.profiles.active_profile = name;
|
||||||
|
settings.save(settings_path)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Records a newly-declared param into `profile_name`'s stored
|
||||||
|
/// declarations (replacing any existing record with the same param name,
|
||||||
|
/// so re-declaring updates it in place), persisting the store to
|
||||||
|
/// `store_path`.
|
||||||
|
///
|
||||||
|
/// **Integration hook for ticket 004 (HTTP API)**: when the HTTP API
|
||||||
|
/// declares a param against the engine (sending
|
||||||
|
/// `EngineMessage::DeclareParam`), it should also call this — with the
|
||||||
|
/// same `DeclareParamRequest` — against the currently-active profile, so
|
||||||
|
/// the declaration survives a restart (see `load_into_engine`). This
|
||||||
|
/// ticket (009) cannot wire that call itself: there's no declare-param
|
||||||
|
/// HTTP route yet for it to hook into. Call shape:
|
||||||
|
///
|
||||||
|
/// ```ignore
|
||||||
|
/// profile_store.record_param_declaration(&store_path, &active_profile_name, &request)?;
|
||||||
|
/// ```
|
||||||
|
pub fn record_param_declaration(
|
||||||
|
&mut self,
|
||||||
|
store_path: &Path,
|
||||||
|
profile_name: &str,
|
||||||
|
request: &DeclareParamRequest,
|
||||||
|
) -> Result<(), ProfileError> {
|
||||||
|
let profile = self
|
||||||
|
.profiles
|
||||||
|
.iter_mut()
|
||||||
|
.find(|p| p.name == profile_name)
|
||||||
|
.ok_or(ProfileError::NotFound)?;
|
||||||
|
let record: ParamDeclarationRecord = request.into();
|
||||||
|
match profile.params.iter_mut().find(|r| r.name == record.name) {
|
||||||
|
Some(existing) => *existing = record,
|
||||||
|
None => profile.params.push(record),
|
||||||
|
}
|
||||||
|
self.save(store_path)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Records a newly-declared metric into `profile_name`'s stored
|
||||||
|
/// declarations (replacing any existing record with the same metric name),
|
||||||
|
/// persisting the store to `store_path`.
|
||||||
|
///
|
||||||
|
/// **Integration hook for ticket 004 (HTTP API)**: same as
|
||||||
|
/// `record_param_declaration` above, but for
|
||||||
|
/// `EngineMessage::DeclareMetric`/`DeclareMetricRequest`.
|
||||||
|
pub fn record_metric_declaration(
|
||||||
|
&mut self,
|
||||||
|
store_path: &Path,
|
||||||
|
profile_name: &str,
|
||||||
|
request: &DeclareMetricRequest,
|
||||||
|
) -> Result<(), ProfileError> {
|
||||||
|
let profile = self
|
||||||
|
.profiles
|
||||||
|
.iter_mut()
|
||||||
|
.find(|p| p.name == profile_name)
|
||||||
|
.ok_or(ProfileError::NotFound)?;
|
||||||
|
let record: MetricDeclarationRecord = request.into();
|
||||||
|
match profile.metrics.iter_mut().find(|r| r.name == record.name) {
|
||||||
|
Some(existing) => *existing = record,
|
||||||
|
None => profile.metrics.push(record),
|
||||||
|
}
|
||||||
|
self.save(store_path)?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::*;
|
||||||
|
use crate::data_value::MetricPrimitiveKind;
|
||||||
|
use crate::dto::LogsParams;
|
||||||
|
use crate::engine::{EngineMessage, EngineQuery, EngineResponse, IngestedLine};
|
||||||
|
use crate::timestamp::Timestamp;
|
||||||
|
use tokio::sync::mpsc;
|
||||||
|
|
||||||
|
fn param_request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest {
|
||||||
|
DeclareParamRequest {
|
||||||
|
name: name.to_string(),
|
||||||
|
message_match_regex: pattern.to_string(),
|
||||||
|
capture_group: capture_group.to_string(),
|
||||||
|
kind: MetricPrimitiveKind::String,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn metric_request(name: &str, pattern: &str, param_name: &str, capture_group: &str) -> DeclareMetricRequest {
|
||||||
|
DeclareMetricRequest {
|
||||||
|
name: name.to_string(),
|
||||||
|
message_match_regex: pattern.to_string(),
|
||||||
|
params: vec![RegexToMetricPrimitiveMatcher {
|
||||||
|
param_name: param_name.to_string(),
|
||||||
|
capture_group: capture_group.to_string(),
|
||||||
|
kind: MetricPrimitiveKind::Number,
|
||||||
|
}],
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse {
|
||||||
|
let (respond_to, response) = oneshot::channel();
|
||||||
|
engine.send(EngineMessage::Query { query, respond_to }).await.unwrap();
|
||||||
|
response.await.unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resolve_profiles_dir_is_relative_to_settings_parent() {
|
||||||
|
let settings_path = Path::new("/config/ottyl/settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
settings.profiles.profiles_dir = "profiles".to_string();
|
||||||
|
|
||||||
|
let dir = resolve_profiles_dir(&settings, settings_path);
|
||||||
|
assert_eq!(dir, Path::new("/config/ottyl/profiles"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resolve_profiles_dir_respects_absolute_path() {
|
||||||
|
let settings_path = Path::new("/config/ottyl/settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
settings.profiles.profiles_dir = "/elsewhere/profiles".to_string();
|
||||||
|
|
||||||
|
let dir = resolve_profiles_dir(&settings, settings_path);
|
||||||
|
assert_eq!(dir, Path::new("/elsewhere/profiles"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn create_list_get_delete_round_trip() {
|
||||||
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
let store_path = dir.path().join("profiles.json");
|
||||||
|
let settings_path = dir.path().join("settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
|
||||||
|
let mut store = ProfileStore::load(&store_path);
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap();
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "worker".to_string()).unwrap();
|
||||||
|
|
||||||
|
assert_eq!(store.list().len(), 2);
|
||||||
|
assert!(store.get("web").is_some());
|
||||||
|
assert_eq!(settings.profiles.names, vec!["web".to_string(), "worker".to_string()]);
|
||||||
|
|
||||||
|
// Duplicate creation is rejected.
|
||||||
|
let dup = store.create(&store_path, &mut settings, &settings_path, "web".to_string());
|
||||||
|
assert!(matches!(dup, Err(ProfileError::DuplicateName)));
|
||||||
|
|
||||||
|
// Reload from disk: profiles and settings both persisted.
|
||||||
|
let reloaded = ProfileStore::load(&store_path);
|
||||||
|
assert_eq!(reloaded.list().len(), 2);
|
||||||
|
let reloaded_settings = Settings::load(&settings_path);
|
||||||
|
assert_eq!(reloaded_settings.profiles.names, settings.profiles.names);
|
||||||
|
|
||||||
|
store.delete(&store_path, &mut settings, &settings_path, "worker").unwrap();
|
||||||
|
assert!(store.get("worker").is_none());
|
||||||
|
assert_eq!(settings.profiles.names, vec!["web".to_string()]);
|
||||||
|
|
||||||
|
let missing = store.delete(&store_path, &mut settings, &settings_path, "worker");
|
||||||
|
assert!(matches!(missing, Err(ProfileError::NotFound)));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn set_active_updates_and_persists_settings() {
|
||||||
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
let store_path = dir.path().join("profiles.json");
|
||||||
|
let settings_path = dir.path().join("settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
|
||||||
|
let mut store = ProfileStore::load(&store_path);
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap();
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "worker".to_string()).unwrap();
|
||||||
|
|
||||||
|
store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap();
|
||||||
|
assert_eq!(settings.profiles.active_profile, Some("web".to_string()));
|
||||||
|
|
||||||
|
// Switching to another known profile works and persists.
|
||||||
|
store.set_active(&mut settings, &settings_path, Some("worker".to_string())).unwrap();
|
||||||
|
let reloaded = Settings::load(&settings_path);
|
||||||
|
assert_eq!(reloaded.profiles.active_profile, Some("worker".to_string()));
|
||||||
|
|
||||||
|
// Setting an unknown profile active is rejected, leaving the previous value.
|
||||||
|
let err = store.set_active(&mut settings, &settings_path, Some("ghost".to_string()));
|
||||||
|
assert!(matches!(err, Err(ProfileError::NotFound)));
|
||||||
|
assert_eq!(settings.profiles.active_profile, Some("worker".to_string()));
|
||||||
|
|
||||||
|
// Clearing the active profile works.
|
||||||
|
store.set_active(&mut settings, &settings_path, None).unwrap();
|
||||||
|
assert_eq!(settings.profiles.active_profile, None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn record_param_and_metric_declarations_persist_into_profile() {
|
||||||
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
let store_path = dir.path().join("profiles.json");
|
||||||
|
let settings_path = dir.path().join("settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
|
||||||
|
let mut store = ProfileStore::load(&store_path);
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap();
|
||||||
|
|
||||||
|
let param_req = param_request("level", r"level=(?P<level>\w+)", "level");
|
||||||
|
store.record_param_declaration(&store_path, "web", ¶m_req).unwrap();
|
||||||
|
|
||||||
|
let metric_req = metric_request(
|
||||||
|
"requests",
|
||||||
|
r"duration=(?P<duration>\d+)",
|
||||||
|
"duration",
|
||||||
|
"duration",
|
||||||
|
);
|
||||||
|
store.record_metric_declaration(&store_path, "web", &metric_req).unwrap();
|
||||||
|
|
||||||
|
let profile = store.get("web").unwrap();
|
||||||
|
assert_eq!(profile.params.len(), 1);
|
||||||
|
assert_eq!(profile.params[0].name, "level");
|
||||||
|
assert_eq!(profile.metrics.len(), 1);
|
||||||
|
assert_eq!(profile.metrics[0].name, "requests");
|
||||||
|
|
||||||
|
// Re-declaring the same name updates in place rather than duplicating.
|
||||||
|
let param_req_v2 = param_request("level", r"lvl=(?P<level>\w+)", "level");
|
||||||
|
store.record_param_declaration(&store_path, "web", ¶m_req_v2).unwrap();
|
||||||
|
let profile = store.get("web").unwrap();
|
||||||
|
assert_eq!(profile.params.len(), 1);
|
||||||
|
assert_eq!(profile.params[0].message_match_regex, r"lvl=(?P<level>\w+)");
|
||||||
|
|
||||||
|
// Reload from disk: recorded declarations survive a restart.
|
||||||
|
let reloaded = ProfileStore::load(&store_path);
|
||||||
|
let profile = reloaded.get("web").unwrap();
|
||||||
|
assert_eq!(profile.params.len(), 1);
|
||||||
|
assert_eq!(profile.metrics.len(), 1);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Acceptance criterion: declaring params/metrics into a profile, "restarting"
|
||||||
|
/// (dropping and reloading the store from disk, rather than a literal process
|
||||||
|
/// restart), and replaying the saved profile against a fresh engine reproduces
|
||||||
|
/// the same declarations and still ingests lines correctly — without
|
||||||
|
/// re-declaring anything by hand.
|
||||||
|
///
|
||||||
|
/// Note: ticket 003 (metric bucket derivation) isn't implemented yet in this
|
||||||
|
/// worktree, so there's no bucket index to compare directly. What this test
|
||||||
|
/// verifies instead is the part ticket 009 owns: that a profile's declarations
|
||||||
|
/// round-trip through disk byte-for-byte and, once replayed via
|
||||||
|
/// `load_into_engine`, produce identical engine-visible declarations (which is
|
||||||
|
/// exactly the state ticket 003's derivation logic will read from) — and that
|
||||||
|
/// the engine keeps ingesting lines normally afterwards.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn profile_round_trip_reproduces_declarations_in_fresh_engine() {
|
||||||
|
let dir = tempfile::tempdir().unwrap();
|
||||||
|
let store_path = dir.path().join("profiles.json");
|
||||||
|
let settings_path = dir.path().join("settings.json");
|
||||||
|
let mut settings = Settings::default();
|
||||||
|
|
||||||
|
// Declare directly into a profile (as ticket 004's HTTP API will do via the
|
||||||
|
// `record_*_declaration` hooks once it exists).
|
||||||
|
let mut store = ProfileStore::load(&store_path);
|
||||||
|
store.create(&store_path, &mut settings, &settings_path, "web".to_string()).unwrap();
|
||||||
|
store
|
||||||
|
.record_param_declaration(
|
||||||
|
&store_path,
|
||||||
|
"web",
|
||||||
|
¶m_request("level", r"level=(?P<level>\w+)", "level"),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
store
|
||||||
|
.record_metric_declaration(
|
||||||
|
&store_path,
|
||||||
|
"web",
|
||||||
|
&metric_request(
|
||||||
|
"requests",
|
||||||
|
r"request duration=(?P<duration>\d+)",
|
||||||
|
"duration",
|
||||||
|
"duration",
|
||||||
|
),
|
||||||
|
)
|
||||||
|
.unwrap();
|
||||||
|
store.set_active(&mut settings, &settings_path, Some("web".to_string())).unwrap();
|
||||||
|
|
||||||
|
// Simulate a restart: drop everything in memory, reload from disk.
|
||||||
|
drop(store);
|
||||||
|
drop(settings);
|
||||||
|
let settings = Settings::load(&settings_path);
|
||||||
|
let active_name = settings.profiles.active_profile.clone().unwrap();
|
||||||
|
let reloaded_store = ProfileStore::load(&store_path);
|
||||||
|
let profile = reloaded_store.get(&active_name).unwrap().clone();
|
||||||
|
|
||||||
|
// Spawn a fresh engine and replay the profile's declarations into it.
|
||||||
|
let (engine_tx, engine_rx) = mpsc::channel(16);
|
||||||
|
tokio::spawn(crate::engine::run(engine_rx));
|
||||||
|
load_into_engine(&engine_tx, &profile).await;
|
||||||
|
|
||||||
|
// Ingest a line and confirm the engine is functioning with the replayed
|
||||||
|
// declarations in place (no re-declaration needed).
|
||||||
|
engine_tx
|
||||||
|
.send(EngineMessage::Ingest(IngestedLine {
|
||||||
|
sequence: 0,
|
||||||
|
timestamp: Timestamp::from_millis(0),
|
||||||
|
text: "level=info request duration=42".to_string(),
|
||||||
|
}))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let params = match ask(&engine_tx, EngineQuery::ParamDeclarations).await {
|
||||||
|
EngineResponse::ParamDeclarations(params) => params,
|
||||||
|
_ => panic!("unexpected response"),
|
||||||
|
};
|
||||||
|
assert_eq!(params.len(), 1);
|
||||||
|
assert_eq!(params[0].name, "level");
|
||||||
|
assert_eq!(params[0].message_match_regex, r"level=(?P<level>\w+)");
|
||||||
|
assert_eq!(params[0].capture_group, "level");
|
||||||
|
assert!(matches!(params[0].kind, MetricPrimitiveKind::String));
|
||||||
|
|
||||||
|
let metrics = match ask(&engine_tx, EngineQuery::MetricDeclarations).await {
|
||||||
|
EngineResponse::MetricDeclarations(metrics) => metrics,
|
||||||
|
_ => panic!("unexpected response"),
|
||||||
|
};
|
||||||
|
assert_eq!(metrics.len(), 1);
|
||||||
|
assert_eq!(metrics[0].name, "requests");
|
||||||
|
assert_eq!(metrics[0].message_match_regex, r"request duration=(?P<duration>\d+)");
|
||||||
|
assert_eq!(metrics[0].params.len(), 1);
|
||||||
|
assert_eq!(metrics[0].params[0].param_name, "duration");
|
||||||
|
|
||||||
|
let logs = match ask(
|
||||||
|
&engine_tx,
|
||||||
|
EngineQuery::Logs(LogsParams { since_sequence: None, start_ms: None, end_ms: None, limit: None }),
|
||||||
|
)
|
||||||
|
.await
|
||||||
|
{
|
||||||
|
EngineResponse::Logs(logs) => logs,
|
||||||
|
_ => panic!("unexpected response"),
|
||||||
|
};
|
||||||
|
assert_eq!(logs.len(), 1);
|
||||||
|
assert_eq!(logs[0].text, "level=info request duration=42");
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user