ottyl/src/main.rs

mod alerts;
mod api;
mod data_value;
mod dto;
mod engine;
mod filter_datetime;
mod filter_parser;
mod index_key;
mod ingest;
mod metric_primitive;
mod profile;
mod promql;
mod settings;
mod timestamp;

use std::path::PathBuf;
use std::sync::{Arc, LazyLock, Mutex};

use clap::Parser;
use tokio::net::TcpListener;
use tokio::sync::mpsc;

use profile::ProfileStore;
use settings::Settings;

/// The resolved default settings path, rendered into `--help` so `--settings-path`'s help text
/// names a real, concrete location instead of only describing where to look for one.
static SETTINGS_PATH_HELP: LazyLock<String> = 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())
    )
});

/// Observability dashboard from terminal output.
#[derive(Parser, Debug)]
struct Args {
    /// Port to serve the API and webapp on. Defaults to 0, letting the OS assign
    /// an ephemeral free port.
    #[arg(long, default_value_t = 0)]
    port: u16,

    #[arg(long, help = SETTINGS_PATH_HELP.as_str())]
    settings_path: Option<PathBuf>,

    /// File to tail for log lines, instead of reading from stdin. Behaves like `tail -f`.
    #[arg(long)]
    file: Option<PathBuf>,

    /// URL of a Prometheus/OpenMetrics-compatible endpoint to poll for metrics, in addition to log ingestion.
    #[arg(long)]
    metrics_url: Option<String>,
}

#[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")
    });
    let settings = Arc::new(Mutex::new(Settings::load(&settings_path)));

    let (engine_tx, engine_rx) = mpsc::channel(1024);
    tokio::spawn(engine::run(engine_rx, settings.clone()));
    {
        let max_logs = settings.lock().unwrap().max_logs;
        let _ = engine_tx.send(engine::EngineMessage::SetMaxLogs(max_logs)).await;
    }

    // Reload the active profile's declarations before ingest starts: there's no way to
    // reprocess early lines later, so declaring after ingest starts could miss lines that raced ahead.
    {
        let (active_profile_name, profiles_dir) = {
            let mut settings_guard = settings.lock().unwrap();
            let profiles_dir = profile::resolve_profiles_dir(&settings_guard, &settings_path);
            let mut store = ProfileStore::load(&profile::profiles_store_path(&profiles_dir));
            store
                .ensure_default(&profile::profiles_store_path(&profiles_dir), &mut settings_guard, &settings_path)
                .expect("failed to ensure the Default profile exists");
            (settings_guard.profiles.active_profile.clone(), profiles_dir)
        };
        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()
                    );
                }
            }
        }
    }

    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);
    let listener = TcpListener::bind(("0.0.0.0", 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(),
    };
    eprintln!(
        "ottyl serving on http://localhost:{} ({})",
        bound_port, source
    );

    axum::serve(listener, app)
        .await
        .expect("server error");
}

#[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"))
        );
    }
}