diff --git a/src/engine.rs b/src/engine.rs index 97b2f2e..741cf0c 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -23,6 +23,15 @@ pub struct IngestedLine { pub text: String, } +/// One sample scraped from a Prometheus/OpenMetrics endpoint, pushed straight into a metric's bucket +/// without going through log-line regex matching — see `crate::ingest::run_metrics_poll`. +pub struct MetricSample { + pub metric_name: String, + pub labels: BTreeMap, + pub value: f64, + pub timestamp: Timestamp, +} + /// Also the request body for `POST /api/params` — `Deserialize` is derived directly here rather than adding a separate DTO, since the shape already matches. #[derive(Debug, Clone, Deserialize)] pub struct DeclareParamRequest { @@ -69,6 +78,7 @@ pub enum EngineQuery { #[allow(dead_code)] pub enum EngineMessage { Ingest(IngestedLine), + IngestMetricSample(MetricSample), DeclareParam { request: DeclareParamRequest, respond_to: oneshot::Sender>, @@ -114,6 +124,8 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle let mut index: BTreeMap = BTreeMap::new(); let mut max_logs: Option = None; let mut log_count: usize = 0; + // Its own counter, independent of log `sequence` — scraped samples aren't tied to any ingested line. + let mut metric_sample_sequence: u64 = 0; // Per-rule rolling windows/seen-value sets, keyed by rule id; lives only as long as this loop (no restart persistence). let mut alert_states: HashMap = HashMap::new(); @@ -139,6 +151,22 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); } } + EngineMessage::IngestMetricSample(sample) => { + let labels = sample + .labels + .into_iter() + .map(|(name, value)| (name, MetricPrimitive::String(value))) + .collect(); + push_bucket_value( + &mut index, + &sample.metric_name, + labels, + metric_sample_sequence, + sample.timestamp, + MetricPrimitive::Number(Number::Float(sample.value)), + ); + metric_sample_sequence += 1; + } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; evict_excess_logs(&mut index, &mut log_count, max_logs); diff --git a/src/ingest.rs b/src/ingest.rs index c152651..83b6acc 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -1,7 +1,8 @@ +use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::time::Duration; -use crate::engine::{EngineMessage, IngestedLine}; +use crate::engine::{EngineMessage, IngestedLine, MetricSample}; use crate::timestamp::Timestamp; use tokio::fs::File; use tokio::io::{AsyncBufReadExt, BufReader}; @@ -9,6 +10,8 @@ use tokio::sync::mpsc; const POLL_INTERVAL: Duration = Duration::from_millis(200); +const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(15); + /// Reads stdin line by line, sequences and timestamps each line, and forwards it /// to the engine task. pub async fn run(engine: mpsc::Sender) { @@ -118,6 +121,120 @@ pub async fn run_file(engine: mpsc::Sender, path: PathBuf) { } } +/// Polls `url` on a fixed interval for a Prometheus/OpenMetrics-format response, pushing every parsed +/// sample straight into the engine. Errors (unreachable endpoint, non-2xx response) are only logged — +/// polling keeps retrying on the next tick rather than exiting. +pub async fn run_metrics_poll(engine: mpsc::Sender, url: String) { + let client = reqwest::Client::new(); + loop { + match client.get(&url).send().await { + Ok(response) => match response.error_for_status() { + Ok(response) => match response.text().await { + Ok(body) => { + let now = Timestamp::now(); + for sample in parse_openmetrics(&body, now) { + if engine.send(EngineMessage::IngestMetricSample(sample)).await.is_err() { + return; + } + } + } + Err(err) => eprintln!("warning: failed to read metrics response from {url}: {err}"), + }, + Err(err) => eprintln!("warning: metrics endpoint {url} returned an error: {err}"), + }, + Err(err) => eprintln!("warning: failed to poll metrics endpoint {url}: {err}"), + } + tokio::time::sleep(METRICS_POLL_INTERVAL).await; + } +} + +/// Parses the Prometheus text exposition / OpenMetrics format: one sample per non-comment, +/// non-empty line, as `metric_name{label="value",...} value [timestamp_ms]` (labels and the +/// timestamp are both optional). A line that doesn't parse is skipped rather than aborting the +/// whole scrape. +fn parse_openmetrics(body: &str, now: Timestamp) -> Vec { + body.lines().filter_map(|line| parse_openmetrics_line(line, now)).collect() +} + +fn parse_openmetrics_line(line: &str, now: Timestamp) -> Option { + let line = line.trim(); + if line.is_empty() || line.starts_with('#') { + return None; + } + + let (name_and_labels, rest) = line.split_once(char::is_whitespace)?; + let mut fields = rest.split_whitespace(); + let value: f64 = fields.next()?.parse().ok()?; + let timestamp = fields + .next() + .and_then(|raw| raw.parse::().ok()) + .map(|ms| Timestamp::from_millis(ms as i64)) + .unwrap_or(now); + + let (metric_name, labels) = match name_and_labels.find('{') { + Some(open) => { + let close = name_and_labels.rfind('}')?; + (name_and_labels[..open].to_string(), parse_openmetrics_labels(&name_and_labels[open + 1..close])) + } + None => (name_and_labels.to_string(), BTreeMap::new()), + }; + if metric_name.is_empty() { + return None; + } + + Some(MetricSample { metric_name, labels, value, timestamp }) +} + +/// Splits on commas without handling an escaped comma inside a quoted label value — fine for the +/// common case this targets, not a full OpenMetrics-grammar parser. +fn parse_openmetrics_labels(raw: &str) -> BTreeMap { + raw.split(',') + .filter_map(|pair| pair.split_once('=')) + .map(|(name, value)| (name.trim().to_string(), value.trim().trim_matches('"').to_string())) + .collect() +} + +#[cfg(test)] +mod openmetrics_tests { + use super::*; + + #[test] + fn parses_bare_metric_without_labels_or_timestamp() { + let now = Timestamp::from_millis(1000); + let samples = parse_openmetrics("requests_total 42\n", now); + assert_eq!(samples.len(), 1); + assert_eq!(samples[0].metric_name, "requests_total"); + assert_eq!(samples[0].value, 42.0); + assert!(samples[0].labels.is_empty()); + assert_eq!(samples[0].timestamp, now); + } + + #[test] + fn parses_labels_and_explicit_timestamp() { + let now = Timestamp::from_millis(1000); + let samples = parse_openmetrics( + r#"http_requests_total{method="GET",path="/a"} 7 5000"#, + now, + ); + assert_eq!(samples.len(), 1); + let sample = &samples[0]; + assert_eq!(sample.metric_name, "http_requests_total"); + assert_eq!(sample.value, 7.0); + assert_eq!(sample.labels.get("method").map(String::as_str), Some("GET")); + assert_eq!(sample.labels.get("path").map(String::as_str), Some("/a")); + assert_eq!(sample.timestamp, Timestamp::from_millis(5000)); + } + + #[test] + fn skips_comments_blank_lines_and_malformed_lines() { + let now = Timestamp::from_millis(0); + let body = "# HELP requests_total docs\n# TYPE requests_total counter\n\nnot a valid line\nrequests_total 1\n"; + let samples = parse_openmetrics(body, now); + assert_eq!(samples.len(), 1); + assert_eq!(samples[0].metric_name, "requests_total"); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/main.rs b/src/main.rs index 811d166..bb3fdf4 100644 --- a/src/main.rs +++ b/src/main.rs @@ -14,7 +14,7 @@ mod settings; mod timestamp; use std::path::PathBuf; -use std::sync::{Arc, Mutex}; +use std::sync::{Arc, LazyLock, Mutex}; use clap::Parser; use tokio::net::TcpListener; @@ -23,6 +23,17 @@ 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 = 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 { @@ -31,13 +42,16 @@ struct Args { #[arg(long, default_value_t = 0)] port: u16, - /// Path to the settings file. Defaults to a per-OS config location. - #[arg(long)] + #[arg(long, help = SETTINGS_PATH_HELP.as_str())] settings_path: Option, /// File to tail for log lines, instead of reading from stdin. Behaves like `tail -f`. #[arg(long)] file: Option, + + /// URL of a Prometheus/OpenMetrics-compatible endpoint to poll for metrics, in addition to log ingestion. + #[arg(long)] + metrics_url: Option, } #[tokio::main] @@ -94,6 +108,10 @@ async fn main() { } } + 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