fix(alerts): keep every label set on the OTLP and JSON metrics paths (#14085)

Co-authored-by: Hengfei Yang <hengfei.yang@gmail.com>

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<String, super::TriggerSlot> = 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

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 <http://www.gnu.org/licenses/>.
 
+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<u64>,
+}
+
 /// 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<String, config::utils::json::Value>) -> 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::<Vec<String>>()
+        .join("|");
+    gxhash::new().sum64(&key)
+}
+
+/// Whether this label set would add anything to `key`'s pending notification.
+fn trigger_wants_labels(
+    trigger_slots: &HashMap<String, TriggerSlot>,
+    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<String, TriggerSlot>,
+    key: &str,
+    labels: u64,
+    alert: &alert::Alert,
+    rows: Vec<config::utils::json::Map<String, config::utils::json::Value>>,
+) {
+    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<String, json::Value> {
+        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<String, json::Value> {
+        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));
+    }
 }

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<String, super::TriggerSlot> = 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

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<String, TriggerSlot> = HashMap::new();
+        let mut trigger_slots: HashMap<String, super::TriggerSlot> = 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<u64>,
-}
-
-/// Whether this label set would add anything to `key`'s pending notification.
-fn trigger_wants_labels(
-    trigger_slots: &HashMap<String, TriggerSlot>,
-    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<String, TriggerSlot>,
-    key: &str,
-    labels: u64,
-    alert: &alert::Alert,
-    rows: Vec<json::Map<String, json::Value>>,
-) {
-    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<ResponseMetadata> {
     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<String, json::Value> {
-        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())]