From 5fdff20e5c4a6fb679ad5983738eff1533e52b6e Mon Sep 17 00:00:00 2001 From: Amey Pawar <138877912+ameyypawar@users.noreply.github.com> Date: Thu, 3 Sep 2026 07:34:40 +0000 Subject: [PATCH] fix(alerts): keep every label set on the OTLP and JSON metrics paths (#14085) Co-authored-by: Hengfei Yang --- src/core/src/metrics/json.rs | 22 +-- src/core/src/metrics/mod.rs | 295 ++++++++++++++++++++++++++++++++++- src/core/src/metrics/otlp.rs | 22 +-- src/core/src/metrics/prom.rs | 214 +------------------------ 4 files changed, 324 insertions(+), 229 deletions(-) diff --git a/src/core/src/metrics/json.rs b/src/core/src/metrics/json.rs index fb9038f17e..f216fe0ce4 100644 --- a/src/core/src/metrics/json.rs +++ b/src/core/src/metrics/json.rs @@ -431,7 +431,7 @@ pub async fn ingest( )); let mut triggers: TriggerAlertData = Vec::with_capacity(cur_stream_alerts.map_or(0, |v| v.len())); - let mut evaluated_alerts = HashSet::new(); + let mut trigger_slots: HashMap = HashMap::new(); // End get stream alert for (mut record, metric_type) in json_data { @@ -556,10 +556,9 @@ pub async fn ingest( stream_status.status.successful += 1; // start check for alert trigger - if let Some(alerts) = cur_stream_alerts - && triggers.len() < alerts.len() - { + if let Some(alerts) = cur_stream_alerts { let end_time = now_micros(); + let dedup = super::series_signature(&record); for alert in alerts { let key = format!( "{}/{}/{}/{}", @@ -568,15 +567,20 @@ pub async fn ingest( alert.stream_name, alert.get_unique_key() ); - // For one alert, only one trigger per request - // Trigger for this alert is already added. - if evaluated_alerts.contains(&key) { + // One row per label set: a series repeats its labels on every sample. + if !super::trigger_wants_labels(&trigger_slots, &key, dedup) { continue; } match alert.evaluate(Some(&record), (None, end_time), None).await { Ok(trigger_results) if trigger_results.data.is_some() => { - triggers.push((alert.clone(), trigger_results.data.unwrap())); - evaluated_alerts.insert(key); + super::merge_trigger_rows( + &mut triggers, + &mut trigger_slots, + &key, + dedup, + alert, + trigger_results.data.unwrap(), + ); } Ok(_) => { // the data doesn't satisfy the alert condition diff --git a/src/core/src/metrics/mod.rs b/src/core/src/metrics/mod.rs index 3439d2b304..0088f0804e 100644 --- a/src/core/src/metrics/mod.rs +++ b/src/core/src/metrics/mod.rs @@ -13,18 +13,41 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use std::collections::{HashMap, HashSet}; + use config::{ - meta::promql::{METRICS_HASH_EXCLUDED_LABELS, Metadata}, - utils::hash::{Sum64, gxhash}, + meta::{ + alerts::{alert, level::PAYLOAD_SAMPLE_ROWS}, + promql::{METRICS_HASH_EXCLUDED_LABELS, Metadata}, + }, + utils::{ + hash::{Sum64, gxhash}, + json::get_string_value, + }, }; use datafusion::arrow::datatypes::Schema; +use crate::ingestion::TriggerAlertData; + pub mod json; mod native_histogram; pub mod otlp; mod otlp_json_compat; pub mod prom; +/// Distinct label sets one realtime notification carries, matching the scheduled path's sample. +const TRIGGER_LABEL_LIMIT: usize = PAYLOAD_SAMPLE_ROWS as usize; + +/// OTLP writes a per-sample `start_time`, which the shared hash-excluded set does not cover. +const TRIGGER_DEDUP_EXTRA_EXCLUDED_LABELS: &[&str] = &["start_time"]; + +/// An alert's pending notification for the request being ingested. +struct TriggerSlot { + idx: usize, + /// Label sets already represented, so a series repeating across samples adds one row. + labels: HashSet, +} + /// The value policy for every metric we ingest, on every path. /// /// NaN means "no observation": the sample is not recorded. An absent series is how Prometheus @@ -85,9 +108,74 @@ fn get_exclude_labels() -> &'static [&'static str] { METRICS_HASH_EXCLUDED_LABELS } +/// Series identity for alert dedup, rendering every value so a numeric label still separates two +/// series. +fn series_signature(labels: &config::utils::json::Map) -> u64 { + let mut labels: Vec<(&str, String)> = labels + .iter() + .filter(|(key, _value)| { + !METRICS_HASH_EXCLUDED_LABELS.contains(&key.as_str()) + && !TRIGGER_DEDUP_EXTRA_EXCLUDED_LABELS.contains(&key.as_str()) + }) + .map(|(key, value)| (key.as_str(), get_string_value(value))) + .collect(); + labels.sort_by(|a, b| a.0.cmp(b.0)); + + let key = labels + .iter() + .map(|(key, value)| format!("{key}:{value}")) + .collect::>() + .join("|"); + gxhash::new().sum64(&key) +} + +/// Whether this label set would add anything to `key`'s pending notification. +fn trigger_wants_labels( + trigger_slots: &HashMap, + key: &str, + labels: u64, +) -> bool { + match trigger_slots.get(key) { + Some(slot) => slot.labels.len() < TRIGGER_LABEL_LIMIT && !slot.labels.contains(&labels), + None => true, + } +} + +/// Record an evaluation against `key`, one row per distinct label set. +fn merge_trigger_rows( + triggers: &mut TriggerAlertData, + trigger_slots: &mut HashMap, + key: &str, + labels: u64, + alert: &alert::Alert, + rows: Vec>, +) { + match trigger_slots.get_mut(key) { + Some(slot) => { + if slot.labels.len() < TRIGGER_LABEL_LIMIT && slot.labels.insert(labels) { + triggers[slot.idx].1.extend(rows); + } + } + None => { + trigger_slots.insert( + key.to_string(), + TriggerSlot { + idx: triggers.len(), + labels: HashSet::from([labels]), + }, + ); + triggers.push((alert.clone(), rows)); + } + } +} + #[cfg(test)] mod tests { - use config::{meta::promql::METADATA_LABEL, utils::json}; + use config::{ + TIMESTAMP_COL_NAME, + meta::promql::{HASH_LABEL, METADATA_LABEL, VALUE_LABEL}, + utils::json, + }; use super::*; @@ -227,4 +315,205 @@ mod tests { fn test_get_prom_metadata_from_schema_absent_metadata_is_none() { assert!(get_prom_metadata_from_schema(&Schema::empty()).is_none()); } + fn labelled_row(label: &str) -> json::Map { + let mut row = json::Map::new(); + row.insert("label".to_string(), json::Value::String(label.to_string())); + row + } + + #[test] + fn merge_trigger_rows_creates_a_slot_on_first_match() { + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + 1, + &alert::Alert::default(), + vec![labelled_row("east")], + ); + assert_eq!(triggers.len(), 1); + assert_eq!(triggers[0].1.len(), 1); + assert_eq!(slots["k"].idx, 0); + } + + #[test] + fn merge_trigger_rows_adds_a_row_per_distinct_label_set() { + // Every label set is its own series, and the reported bug loses all but the first. + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + let alert = alert::Alert::default(); + for (labels, name) in [(1u64, "east"), (2, "west"), (3, "north")] { + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + labels, + &alert, + vec![labelled_row(name)], + ); + } + assert_eq!(triggers.len(), 1, "still one notification per alert"); + let seen: Vec<_> = triggers[0] + .1 + .iter() + .map(|r| r["label"].as_str().unwrap()) + .collect(); + assert_eq!(seen, ["east", "west", "north"]); + } + + #[test] + fn merge_trigger_rows_ignores_a_repeated_label_set() { + // One series repeats its labels per sample, which must not add a row each time. + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + let alert = alert::Alert::default(); + for value in ["t1", "t2", "t3"] { + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + 7, + &alert, + vec![labelled_row(value)], + ); + } + assert_eq!(triggers[0].1.len(), 1); + assert_eq!(triggers[0].1[0]["label"].as_str().unwrap(), "t1"); + } + + #[test] + fn merge_trigger_rows_keeps_separate_alerts_apart() { + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + merge_trigger_rows( + &mut triggers, + &mut slots, + "a", + 1, + &alert::Alert::default(), + vec![labelled_row("x")], + ); + merge_trigger_rows( + &mut triggers, + &mut slots, + "b", + 1, + &alert::Alert::default(), + vec![labelled_row("y")], + ); + assert_eq!(triggers.len(), 2); + assert_eq!(triggers[0].1.len(), 1); + assert_eq!(triggers[1].1.len(), 1); + } + + #[test] + fn merge_trigger_rows_stops_at_the_label_limit() { + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + let alert = alert::Alert::default(); + for i in 0..(TRIGGER_LABEL_LIMIT + 25) { + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + i as u64, + &alert, + vec![labelled_row(&i.to_string())], + ); + } + assert_eq!(triggers[0].1.len(), TRIGGER_LABEL_LIMIT); + } + + #[test] + fn trigger_wants_labels_gates_on_seen_and_capacity() { + let mut triggers: TriggerAlertData = Vec::new(); + let mut slots = HashMap::new(); + let alert = alert::Alert::default(); + assert!( + trigger_wants_labels(&slots, "k", 1), + "an alert with nothing yet takes any label set" + ); + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + 1, + &alert, + vec![labelled_row("east")], + ); + assert!( + !trigger_wants_labels(&slots, "k", 1), + "a label set already represented adds nothing" + ); + assert!(trigger_wants_labels(&slots, "k", 2), "a new one still does"); + for i in 2..=(TRIGGER_LABEL_LIMIT as u64) { + merge_trigger_rows( + &mut triggers, + &mut slots, + "k", + i, + &alert, + vec![labelled_row("x")], + ); + } + assert!( + !trigger_wants_labels(&slots, "k", 9_999), + "a full notification takes nothing new" + ); + } + fn labels(pairs: &[(&str, json::Value)]) -> json::Map { + let mut m = json::Map::new(); + for (k, v) in pairs { + m.insert((*k).to_string(), v.clone()); + } + m + } + + #[test] + fn series_signature_separates_numeric_labels() { + // `signature_without_labels` renders every non-string as "", so these two collide there. + let a = labels(&[("code", json::json!(200))]); + let b = labels(&[("code", json::json!(500))]); + assert_ne!(series_signature(&a), series_signature(&b)); + assert_eq!( + signature_without_labels(&a, &[]), + signature_without_labels(&b, &[]) + ); + } + + #[test] + fn series_signature_separates_boolean_labels() { + let a = labels(&[("ok", json::json!(true))]); + let b = labels(&[("ok", json::json!(false))]); + assert_ne!(series_signature(&a), series_signature(&b)); + } + + #[test] + fn series_signature_ignores_per_sample_columns() { + let base = [("host", json::json!("a"))]; + let one = labels(&[ + base[0].clone(), + (TIMESTAMP_COL_NAME, json::json!(1_000_i64)), + ("start_time", json::json!("111")), + (HASH_LABEL, json::json!(7_u64)), + (VALUE_LABEL, json::json!(1.5)), + ]); + let two = labels(&[ + base[0].clone(), + (TIMESTAMP_COL_NAME, json::json!(2_000_i64)), + ("start_time", json::json!("222")), + (HASH_LABEL, json::json!(9_u64)), + (VALUE_LABEL, json::json!(9.9)), + ]); + assert_eq!(series_signature(&one), series_signature(&two)); + } + + #[test] + fn series_signature_separates_distinct_series() { + let a = labels(&[("host", json::json!("a")), ("region", json::json!("eu"))]); + let b = labels(&[("host", json::json!("b")), ("region", json::json!("eu"))]); + assert_ne!(series_signature(&a), series_signature(&b)); + } } diff --git a/src/core/src/metrics/otlp.rs b/src/core/src/metrics/otlp.rs index 463e70491a..ff88b829ca 100644 --- a/src/core/src/metrics/otlp.rs +++ b/src/core/src/metrics/otlp.rs @@ -558,7 +558,7 @@ pub async fn handle_otlp_request( )); let mut triggers: TriggerAlertData = Vec::with_capacity(cur_stream_alerts.map_or(0, |v| v.len())); - let mut evaluated_alerts = HashSet::new(); + let mut trigger_slots: HashMap = HashMap::new(); for val_map in json_data { let timestamp = val_map @@ -599,10 +599,9 @@ pub async fn handle_otlp_request( hour_buf.records_size += value_str.len(); // start check for alert trigger - if let Some(alerts) = cur_stream_alerts - && triggers.len() < alerts.len() - { + if let Some(alerts) = cur_stream_alerts { let end_time = now_micros(); + let dedup = super::series_signature(&val_map); for alert in alerts { let key = format!( "{}/{}/{}/{}", @@ -611,15 +610,20 @@ pub async fn handle_otlp_request( alert.stream_name, alert.get_unique_key() ); - // For one alert, only one trigger per request - // Trigger for this alert is already added. - if evaluated_alerts.contains(&key) { + // One row per label set: a series repeats its labels on every sample. + if !super::trigger_wants_labels(&trigger_slots, &key, dedup) { continue; } match alert.evaluate(Some(&val_map), (None, end_time), None).await { Ok(trigger_results) if trigger_results.data.is_some() => { - triggers.push((alert.clone(), trigger_results.data.unwrap())); - evaluated_alerts.insert(key); + super::merge_trigger_rows( + &mut triggers, + &mut trigger_slots, + &key, + dedup, + alert, + trigger_results.data.unwrap(), + ); } Ok(_) => { // the data doesn't satisfy the alert condition diff --git a/src/core/src/metrics/prom.rs b/src/core/src/metrics/prom.rs index 7e6861992e..2e715e7a6e 100644 --- a/src/core/src/metrics/prom.rs +++ b/src/core/src/metrics/prom.rs @@ -25,7 +25,7 @@ use config::{ cluster::LOCAL_NODE, get_config, meta::{ - alerts::{alert, level::PAYLOAD_SAMPLE_ROWS}, + alerts::alert, promql::*, search::default_use_cache, self_reporting::usage::UsageType, @@ -591,7 +591,7 @@ pub async fn remote_write( .collect() }) .unwrap_or_default(); - let mut trigger_slots: HashMap = HashMap::new(); + let mut trigger_slots: HashMap = HashMap::new(); for (mut val_map, timestamp) in json_data { let hash = super::signature_without_labels(&val_map, &[VALUE_LABEL]); @@ -668,19 +668,20 @@ pub async fn remote_write( // start check for alert trigger if let Some(alerts) = cur_stream_alerts { let end_time = now_micros(); + let dedup = super::series_signature(&val_map); for (alert, key) in alerts.iter().zip(alert_keys.iter()) { // One row per label set: a series repeats its labels on // every sample, and only distinct sets reach the template. - if !trigger_wants_labels(&trigger_slots, key, hash) { + if !super::trigger_wants_labels(&trigger_slots, key, dedup) { continue; } match alert.evaluate(Some(&val_map), (None, end_time), None).await { Ok(trigger_results) if trigger_results.data.is_some() => { - merge_trigger_rows( + super::merge_trigger_rows( &mut triggers, &mut trigger_slots, key, - hash, + dedup, alert, trigger_results.data.unwrap(), ); @@ -836,58 +837,6 @@ pub async fn remote_write( Ok(()) } -/// Distinct label sets one realtime notification carries, matching the payload -/// sample the scheduled path truncates to. -const TRIGGER_LABEL_LIMIT: usize = PAYLOAD_SAMPLE_ROWS as usize; - -/// An alert's pending notification for the request being ingested. -struct TriggerSlot { - idx: usize, - /// Label-set signatures already represented, so a series repeating across - /// samples contributes one row rather than one per sample. - labels: HashSet, -} - -/// Whether this label set would add anything to `key`'s pending notification. -fn trigger_wants_labels( - trigger_slots: &HashMap, - key: &str, - labels: u64, -) -> bool { - match trigger_slots.get(key) { - Some(slot) => slot.labels.len() < TRIGGER_LABEL_LIMIT && !slot.labels.contains(&labels), - None => true, - } -} - -/// Record an evaluation against `key`, one row per distinct label set. -fn merge_trigger_rows( - triggers: &mut TriggerAlertData, - trigger_slots: &mut HashMap, - key: &str, - labels: u64, - alert: &alert::Alert, - rows: Vec>, -) { - match trigger_slots.get_mut(key) { - Some(slot) => { - if slot.labels.len() < TRIGGER_LABEL_LIMIT && slot.labels.insert(labels) { - triggers[slot.idx].1.extend(rows); - } - } - None => { - trigger_slots.insert( - key.to_string(), - TriggerSlot { - idx: triggers.len(), - labels: HashSet::from([labels]), - }, - ); - triggers.push((alert.clone(), rows)); - } - } -} - pub async fn get_metadata(org_id: &str, req: RequestMetadata) -> Result { if req.limit == Some(0) { return Ok(hashbrown::HashMap::new()); @@ -1487,157 +1436,6 @@ mod tests { use super::*; - fn labelled_row(label: &str) -> json::Map { - let mut row = json::Map::new(); - row.insert("label".to_string(), json::Value::String(label.to_string())); - row - } - - #[test] - fn merge_trigger_rows_creates_a_slot_on_first_match() { - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - 1, - &alert::Alert::default(), - vec![labelled_row("east")], - ); - assert_eq!(triggers.len(), 1); - assert_eq!(triggers[0].1.len(), 1); - assert_eq!(slots["k"].idx, 0); - } - - #[test] - fn merge_trigger_rows_adds_a_row_per_distinct_label_set() { - // Each label set is its own series and every one has to reach the row - // template, which is what the reported bug loses. - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - let alert = alert::Alert::default(); - for (labels, name) in [(1u64, "east"), (2, "west"), (3, "north")] { - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - labels, - &alert, - vec![labelled_row(name)], - ); - } - assert_eq!(triggers.len(), 1, "still one notification per alert"); - let seen: Vec<_> = triggers[0] - .1 - .iter() - .map(|r| r["label"].as_str().unwrap()) - .collect(); - assert_eq!(seen, ["east", "west", "north"]); - } - - #[test] - fn merge_trigger_rows_ignores_a_repeated_label_set() { - // One series repeats its labels once per sample in a request; those - // samples describe the same dimension, so they must not each add a row. - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - let alert = alert::Alert::default(); - for value in ["t1", "t2", "t3"] { - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - 7, - &alert, - vec![labelled_row(value)], - ); - } - assert_eq!(triggers[0].1.len(), 1); - assert_eq!(triggers[0].1[0]["label"].as_str().unwrap(), "t1"); - } - - #[test] - fn merge_trigger_rows_keeps_separate_alerts_apart() { - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - merge_trigger_rows( - &mut triggers, - &mut slots, - "a", - 1, - &alert::Alert::default(), - vec![labelled_row("x")], - ); - merge_trigger_rows( - &mut triggers, - &mut slots, - "b", - 1, - &alert::Alert::default(), - vec![labelled_row("y")], - ); - assert_eq!(triggers.len(), 2); - assert_eq!(triggers[0].1.len(), 1); - assert_eq!(triggers[1].1.len(), 1); - } - - #[test] - fn merge_trigger_rows_stops_at_the_label_limit() { - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - let alert = alert::Alert::default(); - for i in 0..(TRIGGER_LABEL_LIMIT + 25) { - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - i as u64, - &alert, - vec![labelled_row(&i.to_string())], - ); - } - assert_eq!(triggers[0].1.len(), TRIGGER_LABEL_LIMIT); - } - - #[test] - fn trigger_wants_labels_gates_on_seen_and_capacity() { - let mut triggers: TriggerAlertData = Vec::new(); - let mut slots = HashMap::new(); - let alert = alert::Alert::default(); - assert!( - trigger_wants_labels(&slots, "k", 1), - "an alert with nothing yet takes any label set" - ); - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - 1, - &alert, - vec![labelled_row("east")], - ); - assert!( - !trigger_wants_labels(&slots, "k", 1), - "a label set already represented adds nothing" - ); - assert!(trigger_wants_labels(&slots, "k", 2), "a new one still does"); - for i in 2..=(TRIGGER_LABEL_LIMIT as u64) { - merge_trigger_rows( - &mut triggers, - &mut slots, - "k", - i, - &alert, - vec![labelled_row("x")], - ); - } - assert!( - !trigger_wants_labels(&slots, "k", 9_999), - "a full notification takes nothing new" - ); - } - fn schema_with_metadata(blob: &str) -> Schema { Schema::empty().with_metadata( [(METADATA_LABEL.to_string(), blob.to_string())]