From ac67f9e120f6534538311416a56bd5f7557ed8d6 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sat, 5 Sep 2026 13:10:41 +0100 Subject: [PATCH] Add PromQL query support for dashboard visualizations (ticket 013) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hand-rolls a minimal PromQL parser (src/promql.rs) for metric-name + label-matcher selection, optional [range] selectors, and top-level sum()/avg()/rate() wrapping, rather than pulling in a full PromQL parser crate — the required subset is small and a hand-rolled parser stays fully engine-independent (mirrors filter_parser.rs) with no unused-AST overhead. Adds EngineQuery::Promql/EngineResponse::Promql (engine parses + evaluates in place against the index, no copying) and GET /api/query, returning Vec (labels + (timestamp_ms, f64) points) for charting, with malformed queries surfaced as 400 + ErrorDto. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o --- src/api.rs | 81 ++++++++++- src/dto.rs | 25 ++++ src/engine.rs | 254 +++++++++++++++++++++++++++++++++- src/main.rs | 1 + src/promql.rs | 369 ++++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 727 insertions(+), 3 deletions(-) create mode 100644 src/promql.rs diff --git a/src/api.rs b/src/api.rs index c17bfa8..4287255 100644 --- a/src/api.rs +++ b/src/api.rs @@ -4,7 +4,7 @@ use std::sync::{Arc, Mutex}; use crate::dto::{ BucketValuesParams, CssEntryDto, ErrorDto, ImportCssRequest, LinkCssRequest, - MetricDeclarationDto, ParamDeclarationDto, LogsParams, TimelineParams, + MetricDeclarationDto, ParamDeclarationDto, LogsParams, PromqlParams, TimelineParams, }; use crate::engine::{ DeclareMetricRequest, DeclareParamRequest, EngineHandle, EngineMessage, EngineQuery, @@ -105,6 +105,7 @@ pub fn router( .route("/api/params", post(declare_param)) .route("/api/metrics", get(list_metrics).post(declare_metric)) .route("/api/metrics/{name}", get(get_metric_bucket_values)) + .route("/api/query", get(run_promql_query)) .route("/api/settings/css", get(list_css).post(import_css)) .route("/api/settings/css/link", post(link_css)) .route("/api/settings/css/{id}", delete(remove_css)) @@ -268,6 +269,31 @@ async fn get_metric_bucket_values( } } +/// `GET /api/query` — evaluates a PromQL query (ticket 013; see `crate::promql` for +/// the supported grammar) against declared metrics. Query params: `promql` (the +/// query string), `start_ms`, `end_ms` (bound which points are included, mirroring +/// `/api/logs`/`/api/metrics/:name`). Returns `Vec` — chart-ready +/// `labels` + `(timestamp_ms, value)` series — on success (`200 OK`), or +/// `400 Bad Request` with `{ "error": "..." }` for a malformed query. +async fn run_promql_query( + State(state): State, + Query(params): Query, +) -> impl IntoResponse { + let query = EngineQuery::Promql { + query: params.promql, + start_ms: params.start_ms, + end_ms: params.end_ms, + }; + match ask(&state.engine, query).await { + Ok(EngineResponse::Promql(Ok(series))) => Json(series).into_response(), + Ok(EngineResponse::Promql(Err(err))) => { + (StatusCode::BAD_REQUEST, Json(ErrorDto { error: err })).into_response() + } + Ok(_) => StatusCode::INTERNAL_SERVER_ERROR.into_response(), + Err(status) => status.into_response(), + } +} + /// Maps a `CssError` from a settings mutation/lookup to a response status /// and message: `NotFound` -> 404, `NotImported` -> 400 (the id is valid but /// refers to a linked entry, which this operation doesn't apply to), `Io` -> @@ -580,6 +606,59 @@ mod tests { assert_eq!(values[0]["sequence"], 2); } + #[tokio::test] + async fn promql_query_returns_series_for_seeded_fixture() { + let (app, engine, _settings, _dir) = test_app(); + + send_json( + &app, + "POST", + "/api/metrics", + json!({ + "name": "requests", + "message_match_regex": r"request path=(?P\S+)", + "params": [ + { "param_name": "path", "capture_group": "path", "kind": "string" } + ] + }), + ) + .await; + + for (sequence, text) in [ + (0u64, "request path=/a"), + (1u64, "request path=/b"), + (2u64, "request path=/a"), + ] { + engine + .send(EngineMessage::Ingest(IngestedLine { + sequence, + timestamp: Timestamp::from_millis(sequence as i64 * 1000), + text: text.to_string(), + })) + .await + .unwrap(); + } + + let (status, body) = + get_json(&app, "/api/query?promql=requests%7Bpath%3D%22%2Fa%22%7D").await; + assert_eq!(status, StatusCode::OK); + let series = body.as_array().unwrap(); + assert_eq!(series.len(), 1); + assert_eq!(series[0]["labels"]["path"], "/a"); + let points = series[0]["points"].as_array().unwrap(); + assert_eq!(points.len(), 2); + assert_eq!(points[0], json!([0, 1.0])); + assert_eq!(points[1], json!([2000, 1.0])); + } + + #[tokio::test] + async fn promql_query_rejects_malformed_query_with_400() { + let (app, _engine, _settings, _dir) = test_app(); + let (status, body) = get_json(&app, "/api/query?promql=not+valid(").await; + assert_eq!(status, StatusCode::BAD_REQUEST); + assert!(body["error"].as_str().is_some_and(|s| !s.is_empty())); + } + #[tokio::test] async fn declaring_param_persists_into_active_profile() { let (app, _engine, settings_handle, dir) = test_app(); diff --git a/src/dto.rs b/src/dto.rs index 551db1c..a16c223 100644 --- a/src/dto.rs +++ b/src/dto.rs @@ -83,6 +83,31 @@ pub struct BucketValuesParams { pub limit: Option, } +/// One chart-ready time series returned by `GET /api/query` (ticket 013's PromQL +/// endpoint): a label combination plus its `(timestamp_ms, value)` points, sorted by +/// timestamp. `labels` uses plain `String` values (not `MetricPrimitive`) since every +/// value has already been through `MetricPrimitive`'s `Display` form — chart code +/// doesn't need the typed value, just something to show in a legend/tooltip. +/// +/// `value` is always `f64` (see `crate::engine::coerce_to_f64` for how a bucket's +/// `MetricPrimitive` is coerced into one), since a chart needs a plottable number +/// regardless of the underlying metric primitive's kind. +#[derive(Debug, Clone, Serialize, PartialEq)] +pub struct PromqlSeriesDto { + pub labels: BTreeMap, + pub points: Vec<(i64, f64)>, +} + +/// Query params for `GET /api/query`. `promql` is the query string (see +/// `crate::promql`); `start_ms`/`end_ms` bound which points are included, mirroring +/// `/api/logs`/`/api/metrics/:name`. +#[derive(Debug, Deserialize)] +pub struct PromqlParams { + pub promql: String, + pub start_ms: Option, + pub end_ms: Option, +} + /// JSON error body returned by API handlers on a 4xx/5xx response, e.g. /// `{ "error": "invalid regex: ..." }`. #[derive(Debug, Serialize)] diff --git a/src/engine.rs b/src/engine.rs index fc3ff50..883c151 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1,10 +1,11 @@ use crate::data_value::{DataValue, MetricPrimitiveKind, RegexToMetricPrimitiveMatcher}; use crate::dto::{ - BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, StatsDto, - TimelineBucketDto, TimelineParams, + BucketValueDto, LogEntryDto, LogsParams, MetricDeclarationDto, ParamDeclarationDto, + PromqlSeriesDto, StatsDto, TimelineBucketDto, TimelineParams, }; use crate::index_key::IndexKey; use crate::metric_primitive::{MetricPrimitive, Number}; +use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; use crate::timestamp::Timestamp; use regex::Regex; use serde::Deserialize; @@ -64,6 +65,13 @@ pub enum EngineQuery { since_sequence: Option, limit: Option, }, + /// PromQL query (ticket 013's `GET /api/query`). `query` is parsed here (rather + /// than by the API handler) so parsing and evaluation both happen inside the + /// single-owner engine loop, matching every other `EngineQuery` variant's + /// pattern of evaluating in place against `index` with no copying. A malformed + /// query is therefore only discovered once this reaches `handle_query`, hence + /// `EngineResponse::Promql` carrying a `Result` rather than a bare `Vec`. + Promql { query: String, start_ms: Option, end_ms: Option }, } #[allow(dead_code)] @@ -99,6 +107,9 @@ pub enum EngineResponse { ParamDeclarations(Vec), MetricDeclarations(Vec), BucketValues(Vec), + /// `Err` holds a human-readable message for a malformed query (see + /// `crate::promql::parse_query`), which the API layer maps to a 400 response. + Promql(Result, String>), } pub type EngineHandle = mpsc::Sender; @@ -486,6 +497,134 @@ fn handle_query(index: &BTreeMap, query: EngineQuery) -> En .collect(); EngineResponse::BucketValues(values) } + EngineQuery::Promql { query, start_ms, end_ms } => { + let response = promql::parse_query(&query) + .map(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms)); + EngineResponse::Promql(response) + } + } +} + +/// Coerces a bucket's stored `MetricPrimitive` into a plottable `f64` for +/// `PromqlSeriesDto`: +/// - `Number` uses its underlying numeric value. +/// - `Timestamp` uses its millisecond value (arbitrary but at least meaningful/ +/// orderable, since there's no other sensible single number for a point in time). +/// - `String` has no numeric reading at all, so it falls back to `1.0` — i.e. it is +/// treated the same as the presence/occurrence marker `apply_metric_to_log` already +/// stores for every bucket value today (ticket 003), meaning a PromQL series over a +/// `String`-valued bucket becomes an occurrence-count series rather than being +/// skipped. This keeps every declared metric queryable (matching the ticket's +/// "documented fallback" requirement) rather than silently omitting non-numeric +/// series. +fn coerce_to_f64(value: &MetricPrimitive) -> f64 { + match value { + MetricPrimitive::Number(n) => match n { + Number::Int(v) => *v as f64, + Number::Float(v) => *v, + }, + MetricPrimitive::Timestamp(ts) => ts.as_millis() as f64, + MetricPrimitive::String(_) => 1.0, + } +} + +/// True when every matcher in `matchers` is satisfied by `labels` (compared against +/// each label value's `Display` form, same convention as `EngineQuery::BucketValues`'s +/// `labels_filter`). Unmatched/extra labels on the bucket are ignored — a matcher list +/// only needs to be a subset. +fn matches_label_matchers( + labels: &BTreeMap, + matchers: &[LabelMatcher], +) -> bool { + matchers.iter().all(|matcher| { + labels.get(&matcher.name).is_some_and(|value| value.to_string() == matcher.value) + }) +} + +/// Evaluates a parsed PromQL query against `index`, producing chart-ready series. +/// +/// Baseline (no aggregation wrapper): one output series per distinct label +/// combination that matches `parsed.metric_name` + `parsed.matchers`, with that +/// combination's bucket values as `(timestamp_ms, value)` points (see +/// `coerce_to_f64`), sorted by timestamp. +/// +/// `sum()`/`avg()`: merges every matching series' points by exact millisecond +/// timestamp (summing, or averaging, the values that land on the same timestamp +/// across series) into a single output series with empty labels — the aggregation +/// collapses the label dimension, same as PromQL's `sum by ()`/`avg by ()` default. +/// +/// `rate()`: requires (and, per `promql::parse_query`, is only ever given) a range +/// selector. This is a simplified, single-point-per-series rate rather than PromQL's +/// true sliding-window-per-evaluation-step semantics: for each matching series, it +/// takes the window `(last_point_ts - range, last_point_ts]` and divides the number +/// of points falling in that window by the range's duration in seconds, producing one +/// `(last_point_ts, rate)` point per series. Since bucket values are occurrence +/// markers rather than cumulative counters (ticket 003), an occurrence-count-per- +/// second reading is the meaningful analogue of PromQL's counter-increase rate here. +fn evaluate_promql( + index: &BTreeMap, + parsed: &PromqlQuery, + start_ms: Option, + end_ms: Option, +) -> Vec { + let mut series_points: BTreeMap, Vec<(i64, f64)>> = BTreeMap::new(); + for (labels, (_, timestamp, value)) in bucket_values(index, &parsed.metric_name, None) + .filter(|(labels, _)| matches_label_matchers(labels, &parsed.matchers)) + .filter(|(_, (_, timestamp, _))| { + start_ms.is_none_or(|start| timestamp.as_millis() >= start) + && end_ms.is_none_or(|end| timestamp.as_millis() <= end) + }) + { + let display_labels: BTreeMap = + labels.iter().map(|(name, value)| (name.clone(), value.to_string())).collect(); + series_points + .entry(display_labels) + .or_default() + .push((timestamp.as_millis(), coerce_to_f64(value))); + } + for points in series_points.values_mut() { + points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms); + } + + match parsed.agg { + None => series_points + .into_iter() + .map(|(labels, points)| PromqlSeriesDto { labels, points }) + .collect(), + Some(AggOp::Sum) | Some(AggOp::Avg) => { + let mut grouped: BTreeMap> = BTreeMap::new(); + for points in series_points.values() { + for (timestamp_ms, value) in points { + grouped.entry(*timestamp_ms).or_default().push(*value); + } + } + let is_avg = parsed.agg == Some(AggOp::Avg); + let points = grouped + .into_iter() + .map(|(timestamp_ms, values)| { + let sum: f64 = values.iter().sum(); + let value = if is_avg { sum / values.len() as f64 } else { sum }; + (timestamp_ms, value) + }) + .collect(); + vec![PromqlSeriesDto { labels: BTreeMap::new(), points }] + } + Some(AggOp::Rate) => { + // Validated by `promql::parse_query`: `rate()` never parses without one. + let range_ms = parsed.range.expect("rate() always carries a range").as_millis() as i64; + let range_seconds = range_ms as f64 / 1000.0; + series_points + .into_iter() + .filter_map(|(labels, points)| { + let last_ts = points.last()?.0; + let window_start = last_ts - range_ms; + let count_in_window = + points.iter().filter(|(ts, _)| *ts > window_start && *ts <= last_ts).count(); + let rate = count_in_window as f64 / range_seconds; + Some(PromqlSeriesDto { labels, points: vec![(last_ts, rate)] }) + }) + .collect() + } } } @@ -835,4 +974,115 @@ mod tests { assert_eq!(bucket_values(&index, "requests", None).count(), 1); assert_eq!(bucket_values(&index, "durations", None).count(), 1); } + + fn declare_requests_metric(index: &mut BTreeMap) { + declare_metric( + index, + metric_request( + "requests", + r"request path=(?P\S+)", + vec![("path", "path", MetricPrimitiveKind::String)], + ), + ) + .unwrap(); + } + + #[test] + fn promql_query_selects_matching_label_combination() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + ingest_log(&mut index, 1, "request path=/b"); + ingest_log(&mut index, 2, "request path=/a"); + + let parsed = promql::parse_query(r#"requests{path="/a"}"#).unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); + // Occurrence markers coerce to 1.0 (see `coerce_to_f64`); timestamps come from + // `ingest_log`'s `sequence as millis` helper. + assert_eq!(series[0].points, vec![(0, 1.0), (2, 1.0)]); + } + + #[test] + fn promql_query_with_no_matching_labels_returns_no_series() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + + let parsed = promql::parse_query(r#"requests{path="/nope"}"#).unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + assert!(series.is_empty()); + } + + #[test] + fn promql_sum_aggregates_across_label_combinations() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + ingest_log(&mut index, 0, "request path=/b"); + + let parsed = promql::parse_query("sum(requests)").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert_eq!(series.len(), 1); + assert!(series[0].labels.is_empty()); + assert_eq!(series[0].points, vec![(0, 2.0)]); + } + + #[test] + fn promql_range_bounds_filter_points_via_start_end_ms() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + ingest_log(&mut index, 5, "request path=/a"); + ingest_log(&mut index, 10, "request path=/a"); + + let parsed = promql::parse_query("requests").unwrap(); + let series = evaluate_promql(&index, &parsed, Some(1), Some(9)); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].points, vec![(5, 1.0)]); + } + + #[test] + fn handle_query_promql_rejects_malformed_query() { + let index = BTreeMap::new(); + let response = handle_query( + &index, + EngineQuery::Promql { + query: "not a valid promql query(".to_string(), + start_ms: None, + end_ms: None, + }, + ); + match response { + EngineResponse::Promql(Err(message)) => assert!(!message.is_empty()), + _ => panic!("expected EngineResponse::Promql(Err(_))"), + } + } + + #[test] + fn handle_query_promql_returns_series_for_valid_query() { + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + + let response = handle_query( + &index, + EngineQuery::Promql { + query: r#"requests{path="/a"}"#.to_string(), + start_ms: None, + end_ms: None, + }, + ); + match response { + EngineResponse::Promql(Ok(series)) => { + assert_eq!(series.len(), 1); + assert_eq!(series[0].points, vec![(0, 1.0)]); + } + _ => panic!("expected EngineResponse::Promql(Ok(_))"), + } + } } diff --git a/src/main.rs b/src/main.rs index 0aab491..51f9036 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ mod index_key; mod ingest; mod metric_primitive; mod profile; +mod promql; mod settings; mod timestamp; diff --git a/src/promql.rs b/src/promql.rs new file mode 100644 index 0000000..407dee6 --- /dev/null +++ b/src/promql.rs @@ -0,0 +1,369 @@ +//! Hand-rolled parser for a small PromQL subset (ticket 013). +//! +//! Design decision: a crates.io PromQL parser (e.g. `promql-parser`) was considered +//! but rejected — the required subset here is tiny (metric name + label matchers, +//! plus an optional `[5m]` range and a single top-level `sum()`/`avg()`/`rate()` +//! wrapper), while a full PromQL parser brings a large, largely-unused AST plus an +//! extra maintenance dependency. Mirroring `filter_parser.rs`'s approach (a small, +//! engine-independent parser module) is simpler and easier to control/extend, and +//! keeps parsing fully decoupled from the engine's index — this module knows nothing +//! about `IndexKey`/`DataValue`, only `PromqlQuery`'s own types below. +//! +//! ## Supported grammar +//! +//! ```text +//! query := aggregation | vector_selector +//! aggregation := agg_op "(" vector_selector ")" +//! agg_op := "sum" | "avg" | "rate" +//! vector_selector := metric_name label_matchers? range_selector? +//! metric_name := [a-zA-Z_][a-zA-Z0-9_]* +//! label_matchers := "{" label_matcher ("," label_matcher)* "}" +//! label_matcher := label_name "=" quoted_string +//! label_name := [a-zA-Z_][a-zA-Z0-9_]* +//! quoted_string := '"' (any char except '"' or '\')* '"' +//! range_selector := "[" digits duration_unit "]" +//! duration_unit := "s" | "m" | "h" | "d" +//! ``` +//! +//! Only exact-match label matchers (`=`) are supported — PromQL's `!=`, `=~`, `!~` +//! are out of scope for this ticket's required subset. `rate()` requires its inner +//! vector selector to carry a range selector (matching real PromQL), and is rejected +//! at parse time otherwise. + +use std::time::Duration; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum AggOp { + Sum, + Avg, + Rate, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LabelMatcher { + pub name: String, + pub value: String, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct PromqlQuery { + pub agg: Option, + pub metric_name: String, + pub matchers: Vec, + pub range: Option, +} + +/// Parses a PromQL query string per the grammar documented above. Returns a +/// human-readable error message (suitable for surfacing directly to an API caller) +/// on any malformed input. +pub fn parse_query(input: &str) -> Result { + let mut parser = Parser::new(input); + let query = parser.parse_query()?; + parser.skip_whitespace(); + if !parser.is_at_end() { + return Err(format!("unexpected trailing input: '{}'", parser.remaining())); + } + if query.agg == Some(AggOp::Rate) && query.range.is_none() { + return Err("rate() requires a range selector, e.g. rate(my_metric[5m])".to_string()); + } + Ok(query) +} + +struct Parser<'a> { + input: &'a str, + pos: usize, +} + +impl<'a> Parser<'a> { + fn new(input: &'a str) -> Self { + Parser { input, pos: 0 } + } + + fn remaining(&self) -> &'a str { + &self.input[self.pos..] + } + + fn is_at_end(&self) -> bool { + self.pos >= self.input.len() + } + + fn skip_whitespace(&mut self) { + while self.remaining().starts_with(|c: char| c.is_whitespace()) { + self.pos += 1; + } + } + + fn peek(&self) -> Option { + self.remaining().chars().next() + } + + fn expect_char(&mut self, expected: char) -> Result<(), String> { + self.skip_whitespace(); + match self.peek() { + Some(c) if c == expected => { + self.pos += c.len_utf8(); + Ok(()) + } + Some(c) => Err(format!("expected '{expected}', found '{c}'")), + None => Err(format!("expected '{expected}', found end of input")), + } + } + + /// `[a-zA-Z_][a-zA-Z0-9_]*`, used for both metric names and label names. + fn parse_identifier(&mut self) -> Result { + self.skip_whitespace(); + let rest = self.remaining(); + let mut chars = rest.char_indices(); + match chars.next() { + Some((_, c)) if c.is_ascii_alphabetic() || c == '_' => {} + Some((_, c)) => return Err(format!("expected an identifier, found '{c}'")), + None => return Err("expected an identifier, found end of input".to_string()), + } + let end = chars + .find(|(_, c)| !(c.is_ascii_alphanumeric() || *c == '_')) + .map(|(idx, _)| idx) + .unwrap_or(rest.len()); + let ident = &rest[..end]; + self.pos += end; + Ok(ident.to_string()) + } + + fn parse_quoted_string(&mut self) -> Result { + self.expect_char('"')?; + let rest = self.remaining(); + let end = rest.find('"').ok_or("unterminated string literal (missing closing '\"')")?; + let value = &rest[..end]; + if value.contains('\\') { + return Err("escape sequences in label values are not supported".to_string()); + } + self.pos += end; + self.expect_char('"')?; + Ok(value.to_string()) + } + + fn parse_label_matcher(&mut self) -> Result { + let name = self.parse_identifier()?; + self.expect_char('=')?; + self.skip_whitespace(); + if self.peek() == Some('~') { + return Err("regex label matchers ('=~') are not supported, only '='".to_string()); + } + let value = self.parse_quoted_string()?; + Ok(LabelMatcher { name, value }) + } + + fn parse_label_matchers(&mut self) -> Result, String> { + self.skip_whitespace(); + if self.peek() != Some('{') { + return Ok(Vec::new()); + } + self.expect_char('{')?; + let mut matchers = Vec::new(); + self.skip_whitespace(); + if self.peek() == Some('}') { + self.expect_char('}')?; + return Ok(matchers); + } + loop { + matchers.push(self.parse_label_matcher()?); + self.skip_whitespace(); + match self.peek() { + Some(',') => { + self.pos += 1; + } + Some('}') => break, + Some(c) => return Err(format!("expected ',' or '}}', found '{c}'")), + None => return Err("unterminated label matcher list (missing '}')".to_string()), + } + } + self.expect_char('}')?; + Ok(matchers) + } + + /// `[]`, e.g. `[5m]`. + fn parse_range_selector(&mut self) -> Result, String> { + self.skip_whitespace(); + if self.peek() != Some('[') { + return Ok(None); + } + self.expect_char('[')?; + let rest = self.remaining(); + let digits_end = rest + .find(|c: char| !c.is_ascii_digit()) + .ok_or("unterminated range selector (missing unit and ']')")?; + if digits_end == 0 { + return Err("range selector must start with a number, e.g. [5m]".to_string()); + } + let digits: i64 = rest[..digits_end] + .parse() + .map_err(|_| "range selector duration is not a valid number".to_string())?; + self.pos += digits_end; + + let unit = self.peek().ok_or("range selector is missing a duration unit")?; + let seconds_per_unit: i64 = match unit { + 's' => 1, + 'm' => 60, + 'h' => 60 * 60, + 'd' => 24 * 60 * 60, + other => { + return Err(format!( + "unknown duration unit '{other}' (expected one of s, m, h, d)" + )); + } + }; + self.pos += unit.len_utf8(); + self.expect_char(']')?; + Ok(Some(Duration::from_secs((digits * seconds_per_unit).max(0) as u64))) + } + + fn parse_vector_selector(&mut self) -> Result { + let metric_name = self.parse_identifier()?; + let matchers = self.parse_label_matchers()?; + let range = self.parse_range_selector()?; + Ok(PromqlQuery { agg: None, metric_name, matchers, range }) + } + + fn parse_query(&mut self) -> Result { + self.skip_whitespace(); + let checkpoint = self.pos; + let ident = self.parse_identifier()?; + let agg = match ident.as_str() { + "sum" => Some(AggOp::Sum), + "avg" => Some(AggOp::Avg), + "rate" => Some(AggOp::Rate), + _ => None, + }; + self.skip_whitespace(); + if agg.is_some() && self.peek() == Some('(') { + self.expect_char('(')?; + let mut inner = self.parse_vector_selector()?; + self.expect_char(')')?; + inner.agg = agg; + Ok(inner) + } else { + // Not (or not followed by) an aggregation call — rewind and parse the + // whole thing as a plain vector selector (covers metric names that + // happen to collide with an agg_op keyword, e.g. a metric named `rate` + // used without a call). + self.pos = checkpoint; + self.parse_vector_selector() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_bare_metric_name() { + let query = parse_query("requests").unwrap(); + assert_eq!(query.agg, None); + assert_eq!(query.metric_name, "requests"); + assert!(query.matchers.is_empty()); + assert_eq!(query.range, None); + } + + #[test] + fn parses_metric_with_single_label_matcher() { + let query = parse_query(r#"requests{path="/a"}"#).unwrap(); + assert_eq!(query.metric_name, "requests"); + assert_eq!( + query.matchers, + vec![LabelMatcher { name: "path".to_string(), value: "/a".to_string() }] + ); + } + + #[test] + fn parses_metric_with_multiple_label_matchers() { + let query = parse_query(r#"requests{path="/a", method="GET"}"#).unwrap(); + assert_eq!(query.matchers.len(), 2); + assert_eq!(query.matchers[0].name, "path"); + assert_eq!(query.matchers[1].name, "method"); + } + + #[test] + fn parses_range_selector() { + let query = parse_query("requests[5m]").unwrap(); + assert_eq!(query.range, Some(Duration::from_secs(300))); + } + + #[test] + fn parses_label_matchers_and_range_together() { + let query = parse_query(r#"requests{path="/a"}[1h]"#).unwrap(); + assert_eq!(query.matchers.len(), 1); + assert_eq!(query.range, Some(Duration::from_secs(3600))); + } + + #[test] + fn parses_sum_aggregation() { + let query = parse_query(r#"sum(requests{path="/a"})"#).unwrap(); + assert_eq!(query.agg, Some(AggOp::Sum)); + assert_eq!(query.metric_name, "requests"); + } + + #[test] + fn parses_avg_aggregation() { + let query = parse_query("avg(requests)").unwrap(); + assert_eq!(query.agg, Some(AggOp::Avg)); + } + + #[test] + fn parses_rate_with_range() { + let query = parse_query("rate(requests[5m])").unwrap(); + assert_eq!(query.agg, Some(AggOp::Rate)); + assert_eq!(query.range, Some(Duration::from_secs(300))); + } + + #[test] + fn rejects_rate_without_range() { + let err = parse_query("rate(requests)").unwrap_err(); + assert!(err.contains("range selector")); + } + + #[test] + fn rejects_empty_input() { + assert!(parse_query("").is_err()); + } + + #[test] + fn rejects_unclosed_brace() { + assert!(parse_query(r#"requests{path="/a""#).is_err()); + } + + #[test] + fn rejects_unclosed_bracket() { + assert!(parse_query("requests[5m").is_err()); + } + + #[test] + fn rejects_unknown_aggregation() { + // `count` isn't a supported agg_op, so this parses as a bare metric named + // `count`, then trips on the unexpected trailing `(...)`. + let err = parse_query("count(requests)").unwrap_err(); + assert!(err.contains("unexpected trailing input")); + } + + #[test] + fn rejects_regex_label_matcher() { + let err = parse_query(r#"requests{path=~"/a"}"#).unwrap_err(); + assert!(err.contains("regex label matchers")); + } + + #[test] + fn rejects_missing_metric_name() { + assert!(parse_query(r#"{path="/a"}"#).is_err()); + } + + #[test] + fn rejects_unknown_duration_unit() { + let err = parse_query("requests[5x]").unwrap_err(); + assert!(err.contains("duration unit")); + } + + #[test] + fn rejects_trailing_garbage() { + let err = parse_query("requests extra").unwrap_err(); + assert!(err.contains("unexpected trailing input")); + } +}