Show the resolved default settings path in --help; add Prometheus/OpenMetrics polling
--settings-path's help text now names the actual resolved default path (computed once via LazyLock) instead of only describing where to look for one. Also adds an optional --metrics-url flag that polls a Prometheus/ OpenMetrics-compatible endpoint on an interval, pushing samples straight into a metric's bucket (bypassing log-line regex matching) so they're queryable via PromQL/dashboards alongside log-derived metrics. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
18c2d72d22
commit
12381545c8
@@ -23,6 +23,15 @@ pub struct IngestedLine {
|
|||||||
pub text: String,
|
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<String, String>,
|
||||||
|
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.
|
/// 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)]
|
#[derive(Debug, Clone, Deserialize)]
|
||||||
pub struct DeclareParamRequest {
|
pub struct DeclareParamRequest {
|
||||||
@@ -69,6 +78,7 @@ pub enum EngineQuery {
|
|||||||
#[allow(dead_code)]
|
#[allow(dead_code)]
|
||||||
pub enum EngineMessage {
|
pub enum EngineMessage {
|
||||||
Ingest(IngestedLine),
|
Ingest(IngestedLine),
|
||||||
|
IngestMetricSample(MetricSample),
|
||||||
DeclareParam {
|
DeclareParam {
|
||||||
request: DeclareParamRequest,
|
request: DeclareParamRequest,
|
||||||
respond_to: oneshot::Sender<Result<(), String>>,
|
respond_to: oneshot::Sender<Result<(), String>>,
|
||||||
@@ -114,6 +124,8 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
|||||||
let mut index: BTreeMap<IndexKey, DataValue> = BTreeMap::new();
|
let mut index: BTreeMap<IndexKey, DataValue> = BTreeMap::new();
|
||||||
let mut max_logs: Option<usize> = None;
|
let mut max_logs: Option<usize> = None;
|
||||||
let mut log_count: usize = 0;
|
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).
|
// 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<String, AlertRuntimeState> = HashMap::new();
|
let mut alert_states: HashMap<String, AlertRuntimeState> = HashMap::new();
|
||||||
|
|
||||||
@@ -139,6 +151,22 @@ pub async fn run(mut rx: mpsc::Receiver<EngineMessage>, settings: SettingsHandle
|
|||||||
tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone()));
|
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) => {
|
EngineMessage::SetMaxLogs(new_max) => {
|
||||||
max_logs = new_max;
|
max_logs = new_max;
|
||||||
evict_excess_logs(&mut index, &mut log_count, max_logs);
|
evict_excess_logs(&mut index, &mut log_count, max_logs);
|
||||||
|
|||||||
+118
-1
@@ -1,7 +1,8 @@
|
|||||||
|
use std::collections::BTreeMap;
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use crate::engine::{EngineMessage, IngestedLine};
|
use crate::engine::{EngineMessage, IngestedLine, MetricSample};
|
||||||
use crate::timestamp::Timestamp;
|
use crate::timestamp::Timestamp;
|
||||||
use tokio::fs::File;
|
use tokio::fs::File;
|
||||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||||
@@ -9,6 +10,8 @@ use tokio::sync::mpsc;
|
|||||||
|
|
||||||
const POLL_INTERVAL: Duration = Duration::from_millis(200);
|
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
|
/// Reads stdin line by line, sequences and timestamps each line, and forwards it
|
||||||
/// to the engine task.
|
/// to the engine task.
|
||||||
pub async fn run(engine: mpsc::Sender<EngineMessage>) {
|
pub async fn run(engine: mpsc::Sender<EngineMessage>) {
|
||||||
@@ -118,6 +121,120 @@ pub async fn run_file(engine: mpsc::Sender<EngineMessage>, 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<EngineMessage>, 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<MetricSample> {
|
||||||
|
body.lines().filter_map(|line| parse_openmetrics_line(line, now)).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn parse_openmetrics_line(line: &str, now: Timestamp) -> Option<MetricSample> {
|
||||||
|
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::<f64>().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<String, String> {
|
||||||
|
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)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|||||||
+21
-3
@@ -14,7 +14,7 @@ mod settings;
|
|||||||
mod timestamp;
|
mod timestamp;
|
||||||
|
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::{Arc, Mutex};
|
use std::sync::{Arc, LazyLock, Mutex};
|
||||||
|
|
||||||
use clap::Parser;
|
use clap::Parser;
|
||||||
use tokio::net::TcpListener;
|
use tokio::net::TcpListener;
|
||||||
@@ -23,6 +23,17 @@ use tokio::sync::mpsc;
|
|||||||
use profile::ProfileStore;
|
use profile::ProfileStore;
|
||||||
use settings::Settings;
|
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.
|
/// Observability dashboard from terminal output.
|
||||||
#[derive(Parser, Debug)]
|
#[derive(Parser, Debug)]
|
||||||
struct Args {
|
struct Args {
|
||||||
@@ -31,13 +42,16 @@ struct Args {
|
|||||||
#[arg(long, default_value_t = 0)]
|
#[arg(long, default_value_t = 0)]
|
||||||
port: u16,
|
port: u16,
|
||||||
|
|
||||||
/// Path to the settings file. Defaults to a per-OS config location.
|
#[arg(long, help = SETTINGS_PATH_HELP.as_str())]
|
||||||
#[arg(long)]
|
|
||||||
settings_path: Option<PathBuf>,
|
settings_path: Option<PathBuf>,
|
||||||
|
|
||||||
/// File to tail for log lines, instead of reading from stdin. Behaves like `tail -f`.
|
/// File to tail for log lines, instead of reading from stdin. Behaves like `tail -f`.
|
||||||
#[arg(long)]
|
#[arg(long)]
|
||||||
file: Option<PathBuf>,
|
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]
|
#[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 app = api::router(engine_tx, settings, settings_path);
|
||||||
let listener = TcpListener::bind(("0.0.0.0", args.port))
|
let listener = TcpListener::bind(("0.0.0.0", args.port))
|
||||||
.await
|
.await
|
||||||
|
|||||||
Reference in New Issue
Block a user