From a810ecf396b6dbe3c5941ae1976c9d2d5f85daab Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sun, 6 Sep 2026 17:15:39 +0100 Subject: [PATCH] Split src/ into an api/cli/engine/config module hierarchy - src/api/: HTTP router + DTOs + the MCP server (the HTTP/MCP-facing surface): mod.rs (formerly api.rs), dto.rs, mcp_server.rs. - src/cli/: argument parsing and process entry (formerly inline in main.rs) plus session registry: mod.rs, sessions.rs. main.rs is now a thin entry point that just declares modules and calls cli::run(). - src/engine/: the actor + everything it depends on to parse/index/ query log lines: mod.rs (formerly engine.rs), alerts.rs, promql.rs, rate.rs, ingest.rs, logfmt.rs, filter_parser.rs/.pest, filter_datetime.rs, index_key.rs, data_value.rs, metric_primitive.rs, timestamp.rs. - src/config/: persisted configuration: mod.rs (formerly settings.rs), profile.rs. - src/consts.rs stays at the crate root, used from every module above. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01A1ZJ3SrayzLzVndhVq9BmS --- src/{ => api}/dto.rs | 12 +- src/{ => api}/mcp_server.rs | 0 src/{api.rs => api/mod.rs} | 21 +-- src/cli/mod.rs | 205 +++++++++++++++++++++++++ src/{ => cli}/sessions.rs | 0 src/{settings.rs => config/mod.rs} | 16 +- src/{ => config}/profile.rs | 12 +- src/{ => engine}/alerts.rs | 12 +- src/{ => engine}/data_value.rs | 4 +- src/{ => engine}/filter_datetime.rs | 2 +- src/{ => engine}/filter_parser.pest | 0 src/{ => engine}/filter_parser.rs | 2 +- src/{ => engine}/index_key.rs | 4 +- src/{ => engine}/ingest.rs | 2 +- src/{ => engine}/logfmt.rs | 6 +- src/{ => engine}/metric_primitive.rs | 2 +- src/{engine.rs => engine/mod.rs} | 76 +++++---- src/{ => engine}/promql.rs | 0 src/{ => engine}/rate.rs | 2 +- src/{ => engine}/timestamp.rs | 0 src/main.rs | 220 +-------------------------- 21 files changed, 302 insertions(+), 296 deletions(-) rename src/{ => api}/dto.rs (96%) rename src/{ => api}/mcp_server.rs (100%) rename src/{api.rs => api/mod.rs} (99%) create mode 100644 src/cli/mod.rs rename src/{ => cli}/sessions.rs (100%) rename src/{settings.rs => config/mod.rs} (97%) rename src/{ => config}/profile.rs (99%) rename src/{ => engine}/alerts.rs (98%) rename src/{ => engine}/data_value.rs (96%) rename src/{ => engine}/filter_datetime.rs (99%) rename src/{ => engine}/filter_parser.pest (100%) rename src/{ => engine}/filter_parser.rs (99%) rename src/{ => engine}/index_key.rs (87%) rename src/{ => engine}/ingest.rs (99%) rename src/{ => engine}/logfmt.rs (98%) rename src/{ => engine}/metric_primitive.rs (98%) rename src/{engine.rs => engine/mod.rs} (98%) rename src/{ => engine}/promql.rs (100%) rename src/{ => engine}/rate.rs (98%) rename src/{ => engine}/timestamp.rs (100%) diff --git a/src/dto.rs b/src/api/dto.rs similarity index 96% rename from src/dto.rs rename to src/api/dto.rs index a45a7d8..50a92a8 100644 --- a/src/dto.rs +++ b/src/api/dto.rs @@ -1,8 +1,8 @@ -use crate::alerts::AlertCondition; -use crate::data_value::MetricPrimitiveKind; -use crate::metric_primitive::MetricPrimitive; -use crate::profile::{ChartType, Dashboard, Panel, Profile}; -use crate::settings::{ColorScheme, CssEntry, Settings}; +use crate::engine::alerts::AlertCondition; +use crate::engine::data_value::MetricPrimitiveKind; +use crate::engine::metric_primitive::MetricPrimitive; +use crate::config::profile::{ChartType, Dashboard, Panel, Profile}; +use crate::config::{ColorScheme, CssEntry, Settings}; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; @@ -144,7 +144,7 @@ impl From<&CssEntry> for CssEntryDto { path_or_url: None, }, CssEntry::Linked { path_or_url } => CssEntryDto { - id: crate::settings::css_entry_id(entry), + id: crate::config::css_entry_id(entry), kind: CssEntryKind::Linked, filename: None, path_or_url: Some(path_or_url.clone()), diff --git a/src/mcp_server.rs b/src/api/mcp_server.rs similarity index 100% rename from src/mcp_server.rs rename to src/api/mcp_server.rs diff --git a/src/api.rs b/src/api/mod.rs similarity index 99% rename from src/api.rs rename to src/api/mod.rs index 23273dc..6d06d88 100644 --- a/src/api.rs +++ b/src/api/mod.rs @@ -1,21 +1,24 @@ +pub mod dto; +pub mod mcp_server; + use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::sync::Arc; -use crate::dto::{ +use crate::api::dto::{ BucketValuesParams, CreateAlertRuleRequest, CreateDashboardRequest, CreateProfileRequest, CssEntryDto, DashboardDto, ErrorDto, ImportCssRequest, LinkCssRequest, MetricDeclarationDto, ParamDeclarationDto, LogsParams, ProfileSummaryDto, PromqlParams, SaveDashboardRequest, SettingsDto, TestAlertNotificationRequest, TimelineParams, UpdateAlertRuleRequest, UpdateSettingsRequest, }; -use crate::data_value::MetricPrimitiveKind; +use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::{ DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery, EngineResponse, }; -use crate::alerts::{deliver_ntfy, AlertError}; -use crate::profile::{Dashboard, ParamDeclarationRecord, Profile, ProfileError}; -use crate::settings::{CssError, SettingsHandle}; +use crate::engine::alerts::{deliver_ntfy, AlertError}; +use crate::config::profile::{Dashboard, ParamDeclarationRecord, Profile, ProfileError}; +use crate::config::{CssError, SettingsHandle}; use axum::extract::{Path as AxumPath, Query, State}; use axum::http::{header, StatusCode}; use axum::response::{IntoResponse, Json, Response}; @@ -26,7 +29,7 @@ use tokio::sync::oneshot; static WEBAPP_DIST: EmbedDir = fs_embed::fs_embed!("webapp/dist"); -const SKILL_MD: &str = include_str!("../SKILL.md"); +const SKILL_MD: &str = include_str!("../../SKILL.md"); #[derive(Clone)] struct ApiState { @@ -577,7 +580,7 @@ async fn import_css( let entry = settings .css_entries .iter() - .find(|entry| crate::settings::css_entry_id(entry) == id) + .find(|entry| crate::config::css_entry_id(entry) == id) .expect("entry was just inserted by import_css"); (StatusCode::CREATED, Json(CssEntryDto::from(entry))).into_response() } @@ -900,8 +903,8 @@ async fn delete_dashboard( mod tests { use super::*; use crate::engine::{self, IngestedLine}; - use crate::settings::Settings; - use crate::timestamp::Timestamp; + use crate::config::Settings; + use crate::engine::timestamp::Timestamp; use axum::body::Body; use axum::http::Request; use http_body_util::BodyExt; diff --git a/src/cli/mod.rs b/src/cli/mod.rs new file mode 100644 index 0000000..1a32304 --- /dev/null +++ b/src/cli/mod.rs @@ -0,0 +1,205 @@ +pub mod sessions; + +use std::path::PathBuf; +use std::sync::{Arc, LazyLock, Mutex}; + +use clap::{Parser, Subcommand}; +use tokio::net::TcpListener; + +use crate::config::Settings; + +static SETTINGS_PATH_HELP: LazyLock = LazyLock::new(|| { + format!( + "Path to the settings file. Defaults to {}.", + crate::config::default_settings_path() + .map(|path| path.display().to_string()) + .unwrap_or_else(|| "a per-OS config location".to_string()) + ) +}); + +#[derive(Parser, Debug)] +struct Args { + #[command(subcommand)] + command: Option, + + #[arg(long, default_value_t = 0)] + port: u16, + + #[arg(long, help = SETTINGS_PATH_HELP.as_str())] + settings_path: Option, + + #[arg(long)] + file: Option, + + #[arg(long)] + metrics_url: Option, + + #[arg(long)] + name: Option, +} + +#[derive(Subcommand, Debug)] +enum Command { + Mcp { + #[arg(long)] + session: Option, + #[arg(long)] + port: Option, + }, + Sessions, +} + +pub async fn run() { + let args = Args::parse(); + + let settings_path = args.settings_path.clone().unwrap_or_else(|| { + crate::config::default_settings_path() + .expect("could not determine a default settings path; pass --settings-path") + }); + + match args.command { + Some(Command::Mcp { session, port }) => { + run_mcp(&settings_path, session, port).await; + return; + } + Some(Command::Sessions) => { + print_sessions(&settings_path); + return; + } + None => {} + } + + let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); + + let engine_tx = crate::engine::spawn(settings.clone(), crate::consts::ENGINE_CHANNEL_CAPACITY); + { + let max_logs = settings.lock().unwrap().max_logs; + let _ = engine_tx.tx.send(crate::engine::EngineMessage::SetMaxLogs(max_logs)).await; + } + + { + let (active_profile_name, active_profile) = { + let mut settings_guard = settings.lock().unwrap(); + settings_guard + .ensure_default_profile(&settings_path) + .expect("failed to ensure the Default profile exists"); + let active_profile_name = settings_guard.active_profile.clone(); + let active_profile = active_profile_name + .as_deref() + .and_then(|name| settings_guard.profile(name).cloned()); + (active_profile_name, active_profile) + }; + match (active_profile_name, active_profile) { + (Some(_), Some(active_profile)) => { + active_profile.load_into_engine(&engine_tx).await; + } + (Some(profile_name), None) => { + eprintln!( + "warning: active profile '{profile_name}' not found in {}; skipping", + settings_path.display() + ); + } + (None, _) => {} + } + } + + match args.file.clone() { + Some(path) => { + tokio::spawn(crate::engine::ingest::run_file(engine_tx.clone(), path)); + } + None => { + tokio::spawn(crate::engine::ingest::run(engine_tx.clone())); + } + } + + if let Some(metrics_url) = args.metrics_url.clone() { + tokio::spawn(crate::engine::ingest::run_metrics_poll(engine_tx.clone(), metrics_url)); + } + + let app = crate::api::router(engine_tx, settings, settings_path.clone()); + let listener = TcpListener::bind((crate::consts::DEFAULT_BIND_HOST, args.port)) + .await + .expect("failed to bind listener"); + let bound_port = listener + .local_addr() + .expect("failed to read back bound address") + .port(); + + let source = match &args.file { + Some(path) => format!("tailing {}", path.display()), + None => "reading logs from stdin".to_string(), + }; + let session_name = args.name.clone().unwrap_or_else(sessions::generate_name); + match sessions::register_session(&settings_path, &session_name, bound_port) { + Ok(_) => eprintln!( + "ottyl serving on http://localhost:{} ({}), session name: {}", + bound_port, source, session_name + ), + Err(err) => eprintln!( + "ottyl serving on http://localhost:{} ({}); warning: failed to register session '{}': {}", + bound_port, source, session_name, err + ), + } + + axum::serve(listener, app) + .await + .expect("server error"); +} + +async fn run_mcp(settings_path: &std::path::Path, session: Option, port: Option) { + use rmcp::ServiceExt; + + let port = match (port, session) { + (Some(port), _) => port, + (None, Some(name)) => match sessions::resolve_session(settings_path, &name) { + Some(info) => info.port, + None => { + eprintln!("error: no running ottyl session named '{name}'"); + std::process::exit(1); + } + }, + (None, None) => { + eprintln!("error: `ottyl mcp` requires --session or --port "); + std::process::exit(1); + } + }; + + let base_url = format!("http://localhost:{port}"); + let service = crate::api::mcp_server::OttylMcp::new(base_url) + .serve(rmcp::transport::stdio()) + .await + .expect("failed to start MCP server"); + service.waiting().await.expect("MCP server error"); +} + +fn print_sessions(settings_path: &std::path::Path) { + let dir = sessions::sessions_dir(settings_path); + let sessions = sessions::list_sessions(&dir); + if sessions.is_empty() { + println!("no known ottyl sessions"); + return; + } + for session in sessions { + println!("{}\tport={}\tpid={}", session.name, session.port, session.pid); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn settings_path_defaults_to_none() { + let args = Args::parse_from(["ottyl"]); + assert_eq!(args.settings_path, None); + } + + #[test] + fn settings_path_flag_overrides_default() { + let args = Args::parse_from(["ottyl", "--settings-path", "/tmp/custom-settings.json"]); + assert_eq!( + args.settings_path, + Some(PathBuf::from("/tmp/custom-settings.json")) + ); + } +} diff --git a/src/sessions.rs b/src/cli/sessions.rs similarity index 100% rename from src/sessions.rs rename to src/cli/sessions.rs diff --git a/src/settings.rs b/src/config/mod.rs similarity index 97% rename from src/settings.rs rename to src/config/mod.rs index 42e37b7..c585401 100644 --- a/src/settings.rs +++ b/src/config/mod.rs @@ -1,3 +1,5 @@ +pub mod profile; + use std::collections::hash_map::DefaultHasher; use std::fmt; use std::fs; @@ -10,8 +12,8 @@ use std::time::{SystemTime, UNIX_EPOCH}; use directories::ProjectDirs; use serde::{Deserialize, Serialize}; -use crate::alerts::AlertRule; -use crate::profile::Profile; +use crate::engine::alerts::AlertRule; +use crate::config::profile::Profile; pub type SettingsHandle = Arc>; @@ -197,7 +199,7 @@ impl Settings { .map(|parent| parent.join("profiles")) .unwrap_or_else(|| PathBuf::from("profiles")) .join("profiles.json"); - let Some(profiles) = crate::profile::Profile::load_legacy_store(&legacy_path) else { + let Some(profiles) = crate::config::profile::Profile::load_legacy_store(&legacy_path) else { return; }; self.profiles = profiles; @@ -218,7 +220,7 @@ impl Settings { } fn migrate_legacy_profile_shapes(&mut self, settings_path: &Path) { - if crate::profile::migrate_legacy_profile_shapes(&mut self.profiles) + if crate::config::profile::migrate_legacy_profile_shapes(&mut self.profiles) && let Err(err) = self.save(settings_path) { eprintln!( @@ -418,13 +420,13 @@ mod tests { settings.css_entries.push(CssEntry::Linked { path_or_url: "https://example.com/style.css".to_string(), }); - settings.profiles.push(crate::profile::Profile::new("default".to_string())); + settings.profiles.push(crate::config::profile::Profile::new("default".to_string())); settings.active_profile = Some("default".to_string()); - settings.alert_rules.push(crate::alerts::AlertRule { + settings.alert_rules.push(crate::engine::alerts::AlertRule { id: "rule1".to_string(), name: "error spike".to_string(), enabled: true, - condition: crate::alerts::AlertCondition::Immediate { + condition: crate::engine::alerts::AlertCondition::Immediate { query: r#"regex("error")"#.to_string(), }, }); diff --git a/src/profile.rs b/src/config/profile.rs similarity index 99% rename from src/profile.rs rename to src/config/profile.rs index c71f83e..fd40450 100644 --- a/src/profile.rs +++ b/src/config/profile.rs @@ -12,9 +12,9 @@ use std::time::{SystemTime, UNIX_EPOCH}; use serde::{Deserialize, Serialize}; use crate::consts::{DEFAULT_PROFILE_NAME, SEEDED_METRIC_LABEL, SEEDED_METRIC_NAME}; -use crate::data_value::MetricPrimitiveKind; +use crate::engine::data_value::MetricPrimitiveKind; use crate::engine::{DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage}; -use crate::settings::Settings; +use crate::config::Settings; use tokio::sync::oneshot; #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] @@ -678,10 +678,10 @@ impl Settings { #[cfg(test)] mod tests { use super::*; - use crate::data_value::MetricPrimitiveKind; - use crate::dto::LogsParams; + use crate::engine::data_value::MetricPrimitiveKind; + use crate::api::dto::LogsParams; use crate::engine::{EngineMessage, EngineQuery, EngineResponse, IngestedLine}; - use crate::timestamp::Timestamp; + use crate::engine::timestamp::Timestamp; fn param_request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { DeclareParamRequest { @@ -798,7 +798,7 @@ mod tests { #[tokio::test] async fn seeded_metric_renders_real_error_vs_other_data() { use crate::engine::{self, EngineMessage, EngineQuery, EngineResponse, IngestedLine}; - use crate::timestamp::Timestamp; + use crate::engine::timestamp::Timestamp; let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); diff --git a/src/alerts.rs b/src/engine/alerts.rs similarity index 98% rename from src/alerts.rs rename to src/engine/alerts.rs index 1c80398..0f75359 100644 --- a/src/alerts.rs +++ b/src/engine/alerts.rs @@ -10,11 +10,11 @@ use regex::Regex; use serde::{Deserialize, Serialize}; use crate::engine::{self, eval_json_comparison}; -use crate::filter_datetime; -use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; -use crate::rate::RateWindow; -use crate::settings::Settings; -use crate::timestamp::Timestamp; +use crate::engine::filter_datetime; +use crate::engine::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; +use crate::engine::rate::RateWindow; +use crate::config::Settings; +use crate::engine::timestamp::Timestamp; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct AlertRule { @@ -190,7 +190,7 @@ fn eval_alert_expr(expr: &FilterExpr, sequence: u64, timestamp: Timestamp, text: FilterValue::Number(n) => n.to_string(), FilterValue::Raw(raw) => raw.clone(), }; - Ok(op.matches(crate::logfmt::detect_format(text).cmp(target.as_str()))) + Ok(op.matches(crate::engine::logfmt::detect_format(text).cmp(target.as_str()))) } FilterExpr::Comparison { metric, .. } => Err(format!( "alert queries only support regex()/ottyl_seq/ottyl_ts/ottyl_fmt (not metric comparisons like \ diff --git a/src/data_value.rs b/src/engine/data_value.rs similarity index 96% rename from src/data_value.rs rename to src/engine/data_value.rs index 07c5ece..3af2af9 100644 --- a/src/data_value.rs +++ b/src/engine/data_value.rs @@ -1,5 +1,5 @@ -use crate::metric_primitive::MetricPrimitive; -use crate::timestamp::Timestamp; +use crate::engine::metric_primitive::MetricPrimitive; +use crate::engine::timestamp::Timestamp; use serde::{Deserialize, Serialize}; use std::collections::BTreeMap; diff --git a/src/filter_datetime.rs b/src/engine/filter_datetime.rs similarity index 99% rename from src/filter_datetime.rs rename to src/engine/filter_datetime.rs index 2c9c706..0236a8c 100644 --- a/src/filter_datetime.rs +++ b/src/engine/filter_datetime.rs @@ -2,7 +2,7 @@ use chrono::{DateTime, NaiveDate, NaiveTime, TimeZone, Utc}; -use crate::timestamp::Timestamp; +use crate::engine::timestamp::Timestamp; #[derive(Debug, Clone, PartialEq, Eq)] pub enum DateTimeParseError { diff --git a/src/filter_parser.pest b/src/engine/filter_parser.pest similarity index 100% rename from src/filter_parser.pest rename to src/engine/filter_parser.pest diff --git a/src/filter_parser.rs b/src/engine/filter_parser.rs similarity index 99% rename from src/filter_parser.rs rename to src/engine/filter_parser.rs index 9c7cf7f..d9734b4 100644 --- a/src/filter_parser.rs +++ b/src/engine/filter_parser.rs @@ -6,7 +6,7 @@ use pest_derive::Parser as PestParser; use serde::{Deserialize, Serialize}; #[derive(PestParser)] -#[grammar = "filter_parser.pest"] +#[grammar = "engine/filter_parser.pest"] struct FilterGrammar; #[derive(Debug, Clone, PartialEq)] diff --git a/src/index_key.rs b/src/engine/index_key.rs similarity index 87% rename from src/index_key.rs rename to src/engine/index_key.rs index 6a228cf..bdba229 100644 --- a/src/index_key.rs +++ b/src/engine/index_key.rs @@ -1,5 +1,5 @@ -use crate::metric_primitive::MetricPrimitive; -use crate::timestamp::Timestamp; +use crate::engine::metric_primitive::MetricPrimitive; +use crate::engine::timestamp::Timestamp; use serde::Serialize; use std::collections::BTreeMap; diff --git a/src/ingest.rs b/src/engine/ingest.rs similarity index 99% rename from src/ingest.rs rename to src/engine/ingest.rs index ad31fc8..1a94383 100644 --- a/src/ingest.rs +++ b/src/engine/ingest.rs @@ -3,7 +3,7 @@ use std::path::{Path, PathBuf}; use crate::consts::{METRICS_POLL_INTERVAL, STDIN_POLL_INTERVAL as POLL_INTERVAL}; use crate::engine::{EngineHandle, EngineMessage, IngestedLine, MetricSample}; -use crate::timestamp::Timestamp; +use crate::engine::timestamp::Timestamp; use tokio::fs::File; use tokio::io::{AsyncBufReadExt, BufReader}; diff --git a/src/logfmt.rs b/src/engine/logfmt.rs similarity index 98% rename from src/logfmt.rs rename to src/engine/logfmt.rs index f9ad9b7..ca77876 100644 --- a/src/logfmt.rs +++ b/src/engine/logfmt.rs @@ -348,7 +348,7 @@ mod tests { ); } - const LOG_STREAM_FIXTURE: &str = include_str!("fixtures/log_stream_sample.txt"); + const LOG_STREAM_FIXTURE: &str = include_str!("../fixtures/log_stream_sample.txt"); #[test] fn detects_log_stream_shaped_lines() { @@ -467,8 +467,8 @@ mod tests { assert!(!extract_pairs(line).is_empty()); } - const GO_LOG_FIXTURE: &str = include_str!("fixtures/go_log_sample.txt"); - const PY_LOG_FIXTURE: &str = include_str!("fixtures/py_log_sample.txt"); + const GO_LOG_FIXTURE: &str = include_str!("../fixtures/go_log_sample.txt"); + const PY_LOG_FIXTURE: &str = include_str!("../fixtures/py_log_sample.txt"); #[test] fn detects_go_log_shaped_lines() { diff --git a/src/metric_primitive.rs b/src/engine/metric_primitive.rs similarity index 98% rename from src/metric_primitive.rs rename to src/engine/metric_primitive.rs index 86fba1e..250c23b 100644 --- a/src/metric_primitive.rs +++ b/src/engine/metric_primitive.rs @@ -1,4 +1,4 @@ -use crate::timestamp::Timestamp; +use crate::engine::timestamp::Timestamp; use serde::Serialize; use std::cmp::Ordering; use std::fmt; diff --git a/src/engine.rs b/src/engine/mod.rs similarity index 98% rename from src/engine.rs rename to src/engine/mod.rs index 40bde36..0dcd6c7 100644 --- a/src/engine.rs +++ b/src/engine/mod.rs @@ -1,22 +1,32 @@ -use crate::alerts::{self, AlertRuntimeState}; +pub mod alerts; +pub mod data_value; +pub mod filter_datetime; +pub mod filter_parser; +pub mod index_key; +pub mod ingest; +pub mod logfmt; +pub mod metric_primitive; +pub mod promql; +pub mod rate; +pub mod timestamp; + +use crate::engine::alerts::AlertRuntimeState; use crate::consts::{ DEFAULT_TIMELINE_BUCKET_MS, INGEST_LAG_WATERMARK_MAX_ENTRIES, INGEST_LAG_WATERMARK_WINDOW_MS, MAX_AUTO_DECLARED_PARAMS, MAX_NEW_AUTO_PARAMS_PER_LINE, MERGE_WINDOW_MS, OTTYL_RESERVED_PARAM_NAMES, SORT_HELPER_INDEX_PURGE_AFTER, }; -use crate::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; -use crate::dto::{ +use crate::engine::data_value::{DataValue, MetricPrimitiveKind, ParamDeclaration}; +use crate::api::dto::{ BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams, }; -use crate::filter_datetime; -use crate::filter_parser::{self, CompareOp, FilterExpr, FilterValue}; -use crate::index_key::IndexKey; -use crate::metric_primitive::{MetricPrimitive, Number}; -use crate::promql::{AggOp, LabelMatcher, PromqlQuery}; -use crate::rate; -use crate::settings::SettingsHandle; -use crate::timestamp::Timestamp; +use crate::engine::filter_parser::{CompareOp, FilterExpr, FilterValue}; +use crate::engine::index_key::IndexKey; +use crate::engine::metric_primitive::{MetricPrimitive, Number}; +use crate::engine::promql::{AggOp, LabelMatcher, PromqlQuery}; +use crate::config::SettingsHandle; +use crate::engine::timestamp::Timestamp; use left_right::Absorb; use regex::Regex; use serde::Deserialize; @@ -227,7 +237,7 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let mut max_logs: Option = None; let mut metric_sample_sequence: u64 = 0; let mut alert_states: HashMap = HashMap::new(); - let mut log_stream_header: Option = None; + let mut log_stream_header: Option = None; let mut ingest_lag_samples: Vec<(Timestamp, i64)> = Vec::new(); while let Some(message) = rx.recv().await { @@ -243,7 +253,7 @@ async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle, mu let ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at); - let is_new_header = crate::logfmt::LogStreamHeader::detect(&text).is_some(); + let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(&text).is_some(); log_stream_header = log_stream_header_transition(&log_stream_header, &text); let row_header = if is_new_header { None } else { log_stream_header.clone() }; @@ -493,8 +503,8 @@ impl EngineSnapshot { if total >= MAX_AUTO_DECLARED_PARAMS { return; } - let candidates: Vec = if crate::logfmt::detect_format(text) == "json" { - crate::data_value::flatten_json_object(text) + let candidates: Vec = if crate::engine::logfmt::detect_format(text) == "json" { + crate::engine::data_value::flatten_json_object(text) .into_iter() .map(|(path, value)| AutoParamCandidate { name: path.clone(), @@ -505,7 +515,7 @@ impl EngineSnapshot { }) .collect() } else { - crate::logfmt::extract_pairs(text) + crate::engine::logfmt::extract_pairs(text) .into_iter() .map(|(key, value, quoted)| { let group = sanitize_group_name(key); @@ -549,18 +559,18 @@ impl EngineSnapshot { } fn log_stream_header_transition( - current: &Option, + current: &Option, text: &str, -) -> Option { - crate::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone()) +) -> Option { + crate::engine::logfmt::LogStreamHeader::detect(text).or_else(|| current.clone()) } fn derive_log_stream_params_from_row( index: &mut BTreeMap, - header: &crate::logfmt::LogStreamHeader, + header: &crate::engine::logfmt::LogStreamHeader, text: &str, ) { - if !crate::logfmt::is_log_stream_line(text) { + if !crate::engine::logfmt::is_log_stream_line(text) { return; } let row = header.split_row(text); @@ -605,10 +615,10 @@ fn derive_log_stream_params_from_row( #[cfg(test)] fn auto_declare_log_stream_params( index: &mut BTreeMap, - header: &mut Option, + header: &mut Option, text: &str, ) { - let is_new_header = crate::logfmt::LogStreamHeader::detect(text).is_some(); + let is_new_header = crate::engine::logfmt::LogStreamHeader::detect(text).is_some(); *header = log_stream_header_transition(header, text); if is_new_header { return; @@ -764,7 +774,7 @@ fn parse_metric_primitive(raw: &str, kind: MetricPrimitiveKind) -> Option { - crate::filter_datetime::parse_datetime(raw, Timestamp::now()) + crate::engine::filter_datetime::parse_datetime(raw, Timestamp::now()) .ok() .map(MetricPrimitive::Timestamp) } @@ -811,11 +821,11 @@ fn extract_param_raw( json_field_path: Option<&str>, log_stream_gated: bool, ) -> Option { - if log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + if log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) { return None; } match json_field_path.filter(|path| !path.is_empty()) { - Some(field_path) => crate::data_value::extract_json_field(text, field_path), + Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path), None => regex.captures(text)?.name(capture_group).map(|m| m.as_str().to_string()), } } @@ -1250,7 +1260,7 @@ impl CompiledFilter { op.matches(timestamp.as_millis().cmp(target_ms)) } CompiledFilter::Fmt { op, target } => { - op.matches(crate::logfmt::detect_format(text).cmp(target.as_str())) + op.matches(crate::engine::logfmt::detect_format(text).cmp(target.as_str())) } CompiledFilter::MetricMatch(sequences) => sequences.contains(&sequence), CompiledFilter::Json { field, op, value } => eval_json_comparison(text, field, *op, value), @@ -1356,7 +1366,7 @@ fn derive_single_param( ) -> Option { let (_, regex, capture_group, json_field_path, kind, log_stream_gated) = declarations.iter().find(|(name, ..)| name == param_name)?; - if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) { return None; } if *kind == MetricPrimitiveKind::Boolean { @@ -1364,7 +1374,7 @@ fn derive_single_param( } let captures = regex.captures(text)?; let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { - Some(field_path) => crate::data_value::extract_json_field(text, field_path)?, + Some(field_path) => crate::engine::data_value::extract_json_field(text, field_path)?, None => captures.name(capture_group)?.as_str().to_string(), }; parse_metric_primitive(&raw, *kind) @@ -1437,7 +1447,7 @@ impl EngineSnapshot { fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap { let mut params = BTreeMap::new(); for (name, regex, capture_group, json_field_path, kind, log_stream_gated) in declarations { - if *log_stream_gated && !crate::logfmt::is_log_stream_line(text) { + if *log_stream_gated && !crate::engine::logfmt::is_log_stream_line(text) { continue; } if *kind == MetricPrimitiveKind::Boolean { @@ -1448,7 +1458,7 @@ fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> B continue; }; let raw: String = match json_field_path.as_deref().filter(|path| !path.is_empty()) { - Some(field_path) => match crate::data_value::extract_json_field(text, field_path) { + Some(field_path) => match crate::engine::data_value::extract_json_field(text, field_path) { Some(value) => value, None => continue, }, @@ -1467,7 +1477,7 @@ fn derive_log_params(text: &str, declarations: &[CompiledParamDeclaration]) -> B fn log_entry_params(text: &str, declarations: &[CompiledParamDeclaration]) -> BTreeMap { let mut params = derive_log_params(text, declarations); - params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::logfmt::detect_format(text).to_string())); + params.insert("ottyl_fmt".to_string(), MetricPrimitive::String(crate::engine::logfmt::detect_format(text).to_string())); params } @@ -2244,7 +2254,7 @@ mod tests { assert!(declarations[0].6, "the original auto-derived declaration must be untouched"); } - const LOG_STREAM_FIXTURE: &str = include_str!("fixtures/log_stream_sample.txt"); + const LOG_STREAM_FIXTURE: &str = include_str!("../fixtures/log_stream_sample.txt"); fn ingest_log_stream_fixture(index: &mut EngineSnapshot) { let mut header = None; diff --git a/src/promql.rs b/src/engine/promql.rs similarity index 100% rename from src/promql.rs rename to src/engine/promql.rs diff --git a/src/rate.rs b/src/engine/rate.rs similarity index 98% rename from src/rate.rs rename to src/engine/rate.rs index 42fde8f..8f4c0a5 100644 --- a/src/rate.rs +++ b/src/engine/rate.rs @@ -1,6 +1,6 @@ use std::collections::VecDeque; -use crate::timestamp::Timestamp; +use crate::engine::timestamp::Timestamp; fn count_in_window(points: &[(Timestamp, f64)], now: Timestamp, window_ms: i64) -> usize { let now_ms = now.as_millis(); diff --git a/src/timestamp.rs b/src/engine/timestamp.rs similarity index 100% rename from src/timestamp.rs rename to src/engine/timestamp.rs diff --git a/src/main.rs b/src/main.rs index 28bdda6..cad58b9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,224 +1,10 @@ -mod alerts; mod api; +mod cli; +mod config; mod consts; -mod data_value; -mod dto; mod engine; -mod filter_datetime; -mod filter_parser; -mod index_key; -mod ingest; -mod logfmt; -mod mcp_server; -mod metric_primitive; -mod profile; -mod promql; -mod rate; -mod sessions; -mod settings; -mod timestamp; - -use std::path::PathBuf; -use std::sync::{Arc, LazyLock, Mutex}; - -use clap::{Parser, Subcommand}; -use tokio::net::TcpListener; - -use settings::Settings; - -static SETTINGS_PATH_HELP: LazyLock = LazyLock::new(|| { - format!( - "Path to the settings file. Defaults to {}.", - settings::default_settings_path() - .map(|path| path.display().to_string()) - .unwrap_or_else(|| "a per-OS config location".to_string()) - ) -}); - -#[derive(Parser, Debug)] -struct Args { - #[command(subcommand)] - command: Option, - - #[arg(long, default_value_t = 0)] - port: u16, - - #[arg(long, help = SETTINGS_PATH_HELP.as_str())] - settings_path: Option, - - #[arg(long)] - file: Option, - - #[arg(long)] - metrics_url: Option, - - #[arg(long)] - name: Option, -} - -#[derive(Subcommand, Debug)] -enum Command { - Mcp { - #[arg(long)] - session: Option, - #[arg(long)] - port: Option, - }, - Sessions, -} #[tokio::main] async fn main() { - let args = Args::parse(); - - let settings_path = args.settings_path.clone().unwrap_or_else(|| { - settings::default_settings_path() - .expect("could not determine a default settings path; pass --settings-path") - }); - - match args.command { - Some(Command::Mcp { session, port }) => { - run_mcp(&settings_path, session, port).await; - return; - } - Some(Command::Sessions) => { - print_sessions(&settings_path); - return; - } - None => {} - } - - let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); - - let engine_tx = engine::spawn(settings.clone(), consts::ENGINE_CHANNEL_CAPACITY); - { - let max_logs = settings.lock().unwrap().max_logs; - let _ = engine_tx.tx.send(engine::EngineMessage::SetMaxLogs(max_logs)).await; - } - - { - let (active_profile_name, active_profile) = { - let mut settings_guard = settings.lock().unwrap(); - settings_guard - .ensure_default_profile(&settings_path) - .expect("failed to ensure the Default profile exists"); - let active_profile_name = settings_guard.active_profile.clone(); - let active_profile = active_profile_name - .as_deref() - .and_then(|name| settings_guard.profile(name).cloned()); - (active_profile_name, active_profile) - }; - match (active_profile_name, active_profile) { - (Some(_), Some(active_profile)) => { - active_profile.load_into_engine(&engine_tx).await; - } - (Some(profile_name), None) => { - eprintln!( - "warning: active profile '{profile_name}' not found in {}; skipping", - settings_path.display() - ); - } - (None, _) => {} - } - } - - match args.file.clone() { - Some(path) => { - tokio::spawn(ingest::run_file(engine_tx.clone(), path)); - } - None => { - tokio::spawn(ingest::run(engine_tx.clone())); - } - } - - if let Some(metrics_url) = args.metrics_url.clone() { - tokio::spawn(ingest::run_metrics_poll(engine_tx.clone(), metrics_url)); - } - - let app = api::router(engine_tx, settings, settings_path.clone()); - let listener = TcpListener::bind((consts::DEFAULT_BIND_HOST, args.port)) - .await - .expect("failed to bind listener"); - let bound_port = listener - .local_addr() - .expect("failed to read back bound address") - .port(); - - let source = match &args.file { - Some(path) => format!("tailing {}", path.display()), - None => "reading logs from stdin".to_string(), - }; - let session_name = args.name.clone().unwrap_or_else(sessions::generate_name); - match sessions::register_session(&settings_path, &session_name, bound_port) { - Ok(_) => eprintln!( - "ottyl serving on http://localhost:{} ({}), session name: {}", - bound_port, source, session_name - ), - Err(err) => eprintln!( - "ottyl serving on http://localhost:{} ({}); warning: failed to register session '{}': {}", - bound_port, source, session_name, err - ), - } - - axum::serve(listener, app) - .await - .expect("server error"); -} - -async fn run_mcp(settings_path: &std::path::Path, session: Option, port: Option) { - use rmcp::ServiceExt; - - let port = match (port, session) { - (Some(port), _) => port, - (None, Some(name)) => match sessions::resolve_session(settings_path, &name) { - Some(info) => info.port, - None => { - eprintln!("error: no running ottyl session named '{name}'"); - std::process::exit(1); - } - }, - (None, None) => { - eprintln!("error: `ottyl mcp` requires --session or --port "); - std::process::exit(1); - } - }; - - let base_url = format!("http://localhost:{port}"); - let service = mcp_server::OttylMcp::new(base_url) - .serve(rmcp::transport::stdio()) - .await - .expect("failed to start MCP server"); - service.waiting().await.expect("MCP server error"); -} - -fn print_sessions(settings_path: &std::path::Path) { - let dir = sessions::sessions_dir(settings_path); - let sessions = sessions::list_sessions(&dir); - if sessions.is_empty() { - println!("no known ottyl sessions"); - return; - } - for session in sessions { - println!("{}\tport={}\tpid={}", session.name, session.port, session.pid); - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn settings_path_defaults_to_none() { - let args = Args::parse_from(["ottyl"]); - assert_eq!(args.settings_path, None); - } - - #[test] - fn settings_path_flag_overrides_default() { - let args = Args::parse_from(["ottyl", "--settings-path", "/tmp/custom-settings.json"]); - assert_eq!( - args.settings_path, - Some(PathBuf::from("/tmp/custom-settings.json")) - ); - } + cli::run().await; }