diff --git a/src/engine.rs b/src/engine.rs index 7b2dd95..206eef2 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1202,13 +1202,13 @@ fn handle_query( } EngineQuery::Promql { query, start_ms, end_ms } => { let response = promql::parse_query(&query) - .map(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms)); + .and_then(|parsed| evaluate_promql(index, &parsed, start_ms, end_ms)); EngineResponse::Promql(response) } } } -/// Coerces a bucket's `MetricPrimitive` into a plottable `f64`: Number/Timestamp/Duration use their natural numeric value; `String` has no numeric reading, so it falls back to `1.0` (matching the presence/occurrence marker used for every bucket) rather than being skipped, keeping every declared metric queryable. +/// Coerces a bucket's `MetricPrimitive` into a plottable `f64`: Number/Timestamp/Duration use their natural numeric value; `String` has no numeric reading, so it falls back to `1.0` (matching the presence/occurrence marker used for every bucket) rather than being skipped, keeping every declared metric queryable as a bare series or under `rate()`/`gauge()`, which only ever count or sample points rather than summing this value. `sum()`/`avg()` do read it, so `evaluate_promql` rejects those aggregations up front for a `String`-kind metric (see `declared_value_kind`) rather than letting this 1.0 stand in for a sum. fn coerce_to_f64(value: &MetricPrimitive) -> f64 { match value { MetricPrimitive::Number(n) => match n { @@ -1313,13 +1313,48 @@ fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, .collect() } +/// The value kind a metric's buckets carry: the kind of its `Value`-role param, or `Number` when +/// it has none (every occurrence then falls back to a fixed `Number::Int(1)` marker — see +/// `apply_metric_to_log`). Returns `None` when `metric_name` has no declaration at all, in which +/// case there's nothing to validate against and the query simply yields no series. +fn declared_value_kind( + index: &BTreeMap, + metric_name: &str, +) -> Option { + match index.get(&IndexKey::MetricDeclaration { name: metric_name.to_string() }) { + Some(DataValue::MetricDeclaration { params, .. }) => Some( + params + .iter() + .find(|param| param.role == ParamRole::Value) + .map(|param| param.kind) + .unwrap_or(MetricPrimitiveKind::Number), + ), + _ => None, + } +} + /// Evaluates a parsed PromQL query against `index`, producing chart-ready series: baseline is one series per label combination; `sum()`/`avg()` merge series into one series per `by_labels` group (the whole result collapses to a single label-less series when `by_labels` is empty); `rate()` — bare or wrapped in `sum()`/`avg()` — turns each series into a windowed occurrences-per-second reading via `windowed_rate_series`, so it plots as a real line rather than one trailing point. +/// `sum()`/`avg()` over a `String`-kind metric are rejected up front — `coerce_to_f64` has no +/// numeric reading for a string, so silently treating every occurrence as `1.0` would let a sum +/// pass for a count with no signal to the caller. `sum(rate(...))`/`avg(rate(...))` are exempt: +/// the inner `rate()` only counts occurrences per window and never reads the coerced value, so a +/// string-kind metric under `rate()` is already the count semantic the ticket asks for. fn evaluate_promql( index: &BTreeMap, parsed: &PromqlQuery, start_ms: Option, end_ms: Option, -) -> Vec { +) -> Result, String> { + if matches!(parsed.agg, Some(AggOp::Sum) | Some(AggOp::Avg)) + && !parsed.inner_rate + && declared_value_kind(index, &parsed.metric_name) == Some(MetricPrimitiveKind::String) + { + return Err(format!( + "cannot sum/avg a string-typed metric \"{}\"; use count() or a numeric-typed metric", + parsed.metric_name + )); + } + 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)) @@ -1339,11 +1374,30 @@ fn evaluate_promql( points.sort_by_key(|(timestamp_ms, _)| *timestamp_ms); } - match parsed.agg { + Ok(match parsed.agg { None => series_points .into_iter() .map(|(labels, points)| PromqlSeriesDto { labels, points }) .collect(), + Some(AggOp::Count) => { + // Occurrence count per group, ignoring each point's coerced value entirely — the + // explicit opt-in for the count-shaped result `sum()`/`avg()` used to give a + // `String`-kind metric by accident (see `coerce_to_f64`). + let mut groups: BTreeMap, Vec>> = BTreeMap::new(); + for (labels, points) in series_points { + let count_points: Vec<(i64, f64)> = + points.iter().map(|(timestamp_ms, _)| (*timestamp_ms, 1.0)).collect(); + let group_key = project_labels(&labels, &parsed.by_labels); + groups.entry(group_key).or_default().push(count_points); + } + groups + .into_iter() + .map(|(group_labels, sub_series)| { + let points = merge_series_exact(&sub_series, false); + PromqlSeriesDto { labels: group_labels, points } + }) + .collect() + } Some(AggOp::Sum) | Some(AggOp::Avg) => { let is_avg = parsed.agg == Some(AggOp::Avg); // Validated by `promql::parse_query`: an inner rate() always carries a range. @@ -1421,7 +1475,7 @@ fn evaluate_promql( }) .collect() } - } + }) } #[cfg(test)] @@ -1947,7 +2001,7 @@ mod tests { 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); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); @@ -1962,7 +2016,7 @@ mod tests { 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); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert!(series.is_empty()); } @@ -1974,7 +2028,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(requests)").unwrap(); - let series = evaluate_promql(&index, &parsed, None, None); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); @@ -1990,7 +2044,7 @@ mod tests { ingest_log(&mut index, 1, "request path=/a"); let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); - let mut series = evaluate_promql(&index, &parsed, None, None); + let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); @@ -2000,6 +2054,99 @@ mod tests { assert_eq!(series[1].points, vec![(0, 1.0)]); } + /// Declares a metric whose captured `status` param is `String`-kind and holds the `Value` role, + /// so every bucket entry for it is a `MetricPrimitive::String` rather than the default occurrence marker. + fn declare_string_valued_metric(index: &mut BTreeMap) { + declare_metric( + index, + metric_request_with_roles( + "statuses", + r"request status=(?P\S+)", + vec![("status", "status", MetricPrimitiveKind::String, ParamRole::Value)], + ), + ) + .unwrap(); + } + + #[test] + fn sum_over_string_kind_metric_is_rejected_with_an_actionable_error() { + let mut index = BTreeMap::new(); + declare_string_valued_metric(&mut index); + ingest_log(&mut index, 0, "request status=ok"); + + let parsed = promql::parse_query("sum(statuses)").unwrap(); + let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); + + assert!(err.contains("statuses"), "error should name the offending metric: {err}"); + assert!(err.to_lowercase().contains("string"), "error should explain why: {err}"); + } + + #[test] + fn avg_over_string_kind_metric_is_rejected_with_an_actionable_error() { + let mut index = BTreeMap::new(); + declare_string_valued_metric(&mut index); + ingest_log(&mut index, 0, "request status=ok"); + + let parsed = promql::parse_query("avg(statuses)").unwrap(); + let err = evaluate_promql(&index, &parsed, None, None).unwrap_err(); + + assert!(err.contains("statuses")); + } + + #[test] + fn count_over_string_kind_metric_counts_occurrences_by_name_not_by_accident() { + let mut index = BTreeMap::new(); + declare_string_valued_metric(&mut index); + ingest_log(&mut index, 0, "request status=ok"); + ingest_log(&mut index, 1, "request status=error"); + + let parsed = promql::parse_query("count(statuses)").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); + + assert_eq!(series.len(), 1); + assert!(series[0].labels.is_empty()); + assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); + } + + #[test] + fn sum_over_numeric_kind_metric_is_unaffected_by_the_string_kind_guard() { + 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).unwrap(); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].points, vec![(0, 2.0)]); + } + + #[test] + fn sum_of_rate_over_string_kind_metric_is_still_allowed_since_rate_only_counts_occurrences() { + let mut index = BTreeMap::new(); + declare_string_valued_metric(&mut index); + ingest_log(&mut index, 0, "request status=ok"); + ingest_log(&mut index, 1000, "request status=error"); + + let parsed = promql::parse_query("sum(rate(statuses[2s]))").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert!(series.is_ok(), "sum(rate(...)) never reads the coerced value, so it should not be rejected: {series:?}"); + } + + #[test] + fn bare_query_over_string_kind_metric_is_still_allowed() { + let mut index = BTreeMap::new(); + declare_string_valued_metric(&mut index); + ingest_log(&mut index, 0, "request status=ok"); + + let parsed = promql::parse_query("statuses").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert!(series.is_ok()); + } + #[test] fn bare_rate_produces_a_windowed_point_per_timestamp_not_a_single_point() { let mut index = BTreeMap::new(); @@ -2008,7 +2155,7 @@ mod tests { ingest_log(&mut index, 1000, "request path=/a"); let parsed = promql::parse_query("rate(requests[2s])").unwrap(); - let series = evaluate_promql(&index, &parsed, None, None); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(0, 0.5), (1000, 1.0)]); @@ -2023,7 +2170,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s])) by (path)").unwrap(); - let mut series = evaluate_promql(&index, &parsed, None, None); + let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); @@ -2041,7 +2188,7 @@ mod tests { ingest_log(&mut index, 0, "request path=/b"); let parsed = promql::parse_query("sum(rate(requests[2s]))").unwrap(); - let series = evaluate_promql(&index, &parsed, None, None); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert!(series[0].labels.is_empty()); @@ -2057,7 +2204,7 @@ mod tests { 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)); + let series = evaluate_promql(&index, &parsed, Some(1), Some(9)).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 1.0)]); @@ -2122,7 +2269,7 @@ mod tests { ingest_log(&mut index, 2, "request path=/b duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); - let mut series = evaluate_promql(&index, &parsed, None, None); + let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("path").cmp(&b.labels.get("path"))); assert_eq!(series.len(), 2); @@ -2141,7 +2288,7 @@ mod tests { ingest_log(&mut index, 10, "request path=/a duration=42"); let parsed = promql::parse_query("gauge(durations)").unwrap(); - let series = evaluate_promql(&index, &parsed, None, Some(9)); + let series = evaluate_promql(&index, &parsed, None, Some(9)).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(5, 9.0)]); @@ -2156,7 +2303,7 @@ mod tests { } let parsed = promql::parse_query("histogram(durations[10])").unwrap(); - let mut series = evaluate_promql(&index, &parsed, None, None); + let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("le").cmp(&b.labels.get("le"))); // durations 2,4 -> le=10 (count 2); 12,15 -> le=20 (count 2); 21 -> le=30 (count 1). @@ -2222,7 +2369,7 @@ mod tests { ingest_log(&mut index, 10, "user=alice logged in"); let parsed = promql::parse_query("login_gap").unwrap(); - let series = evaluate_promql(&index, &parsed, None, None); + let series = evaluate_promql(&index, &parsed, None, None).unwrap(); assert_eq!(series.len(), 1); assert_eq!(series[0].points, vec![(10, 10.0)]); @@ -2272,7 +2419,7 @@ mod tests { ingest_log(&mut index, 2, "all good again"); let parsed = promql::parse_query("log_category").unwrap(); - let mut series = evaluate_promql(&index, &parsed, None, None); + let mut series = evaluate_promql(&index, &parsed, None, None).unwrap(); series.sort_by(|a, b| a.labels.get("category").cmp(&b.labels.get("category"))); assert_eq!(series.len(), 2); diff --git a/src/promql.rs b/src/promql.rs index 3a2ea0c..0c3d237 100644 --- a/src/promql.rs +++ b/src/promql.rs @@ -1,6 +1,7 @@ -//! Hand-rolled parser for a small PromQL subset: metric name + label matchers, an optional `[5m]` range, and a top-level `sum()`/`avg()`/`rate()`/`histogram()`/`gauge()` wrapper. +//! Hand-rolled parser for a small PromQL subset: metric name + label matchers, an optional `[5m]` range, and a top-level `sum()`/`avg()`/`count()`/`rate()`/`histogram()`/`gauge()` wrapper. //! Only exact-match (`=`) label matchers are supported; `rate()` requires a duration range selector, while `histogram(metric[width])` reuses that `[...]` slot for a bare bucket width instead. //! `sum()`/`avg()` may additionally wrap a `rate(...)` call directly (e.g. `sum(rate(requests[5m]))`), and either may be followed by a `by (label, ...)` clause to preserve one series per distinct value of the named labels instead of collapsing to a single series. +//! `count()` is a plain vector selector wrapper only (no `rate()` nesting) — it counts occurrences per group regardless of the metric's value, which is the deliberate opt-in for occurrence counting that `sum()`/`avg()` no longer fall back to for a `String`-kind metric (see `crate::engine::evaluate_promql`). use std::time::Duration; @@ -8,6 +9,9 @@ use std::time::Duration; pub enum AggOp { Sum, Avg, + /// Occurrence count per group, ignoring the metric's stored value entirely — the explicit, + /// opt-in-by-name alternative to summing a `String`-kind metric's coerced-to-1.0 values. + Count, Rate, /// Buckets counter values into fixed-width buckets; requires `bucket_width`. Histogram, @@ -278,6 +282,7 @@ impl<'a> Parser<'a> { let agg = match ident.as_str() { "sum" => Some(AggOp::Sum), "avg" => Some(AggOp::Avg), + "count" => Some(AggOp::Count), "rate" => Some(AggOp::Rate), "histogram" => Some(AggOp::Histogram), "gauge" => Some(AggOp::Gauge), @@ -415,11 +420,26 @@ mod tests { #[test] fn rejects_unknown_aggregation() { - // `count` isn't a supported agg_op, so this parses as a bare metric name and trips on the trailing `(...)`. - let err = parse_query("count(requests)").unwrap_err(); + // `median` isn't a supported agg_op, so this parses as a bare metric name and trips on the trailing `(...)`. + let err = parse_query("median(requests)").unwrap_err(); assert!(err.contains("unexpected trailing input")); } + #[test] + fn parses_count_aggregation() { + let query = parse_query("count(requests)").unwrap(); + assert_eq!(query.agg, Some(AggOp::Count)); + assert_eq!(query.metric_name, "requests"); + assert!(!query.inner_rate); + } + + #[test] + fn parses_count_with_by_clause() { + let query = parse_query("count(requests) by (path)").unwrap(); + assert_eq!(query.agg, Some(AggOp::Count)); + assert_eq!(query.by_labels, vec!["path".to_string()]); + } + #[test] fn rejects_regex_label_matcher() { let err = parse_query(r#"requests{path=~"/a"}"#).unwrap_err();