diff --git a/src/core/src/ingestion/mod.rs b/src/core/src/ingestion/mod.rs index c0ac7ec1eb..e51ef6c17f 100644 --- a/src/core/src/ingestion/mod.rs +++ b/src/core/src/ingestion/mod.rs @@ -58,6 +58,7 @@ use crate::{ pub mod grpc; pub mod ingestion_service; +pub mod realtime_alerts; pub type TriggerAlertData = Vec<(Alert, Vec>)>; @@ -103,13 +104,22 @@ pub async fn get_stream_alerts( continue; } - let stream_alerts_cacher = STREAM_ALERTS.read().await; - let alerts_id_list = stream_alerts_cacher.get(&key); - if alerts_id_list.is_none() { - continue; - } + // Copy the ids out and release this guard before the loop below. + // `get_alert_from_cache` acquires the alert cache read lock, while alert + // deletion acquires the two caches in the opposite order (alert cache + // write, then stream-alert cache write). Holding this guard across that + // await is a lock-order inversion, and because these are write-preferring + // locks a pending writer also blocks new readers -- so it can stall + // ingestion for every stream type, not just metrics. + let alert_ids: Vec = { + let stream_alerts_cacher = STREAM_ALERTS.read().await; + match stream_alerts_cacher.get(&key) { + Some(ids) => ids.clone(), + None => continue, + } + }; let mut alerts_list = vec![]; - for alert_id in alerts_id_list.unwrap().iter() { + for alert_id in alert_ids.iter() { if let Some((_, alert)) = alert::get_alert_from_cache(&stream.org_id, alert_id).await { alerts_list.push(alert); } @@ -127,10 +137,12 @@ pub async fn get_stream_alerts( } }) .collect::>(); - if alerts.is_empty() { - return; + // A stream whose alerts are all filtered out (disabled, scheduled, or + // silenced) must not abort the loop -- the streams after it in this + // request still need their alerts loaded. + if !alerts.is_empty() { + stream_alerts_map.insert(key, alerts); } - stream_alerts_map.insert(key, alerts); } } @@ -1091,4 +1103,245 @@ mod tests { let result = create_log_ingestion_req(99, data); assert!(result.is_err()); } + + mod get_stream_alerts_tests { + use config::meta::folder::Folder; + use svix_ksuid::{Ksuid, KsuidLike}; + + use super::*; + use crate::common::infra::config::ALERTS; + + /// Seed one real-time alert for `org`/`stream` into the in-memory caches + /// `get_stream_alerts` reads from, and return its id. + /// + /// Every test uses a unique org so the process-global caches cannot make + /// tests order-dependent or flaky under parallel execution. + async fn seed_alert(org: &str, stream: &str, enabled: bool, real_time: bool) -> Ksuid { + let id = Ksuid::new(None, None); + // `Alert` carries private fields, so struct-update syntax is unavailable + // outside its defining crate. It is `#[serde(default)]`, so building it + // from JSON is the supported route and mirrors how alerts really arrive. + let mut alert: Alert = from_value(json!({ + "name": format!("alert_{stream}"), + "org_id": org, + "stream_type": "metrics", + "stream_name": stream, + "is_real_time": real_time, + "enabled": enabled, + })) + .expect("valid alert json"); + alert.id = Some(id); + ALERTS.write().await.insert( + alert::cache_alert_key(org, &id.to_string()), + (Folder::default(), alert), + ); + STREAM_ALERTS + .write() + .await + .entry(alert::cache_stream_key(org, StreamType::Metrics, stream)) + .or_default() + .push(id.to_string()); + id + } + + fn params(org: &str, stream: &str) -> StreamParams { + StreamParams::new(org, stream, StreamType::Metrics) + } + + fn key(org: &str, stream: &str) -> String { + alert::cache_stream_key(org, StreamType::Metrics, stream) + } + + #[tokio::test] + async fn test_get_stream_alerts_loads_a_stream_with_alerts() { + let org = "d2_g1"; + seed_alert(org, "alerted", true, true).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "alerted")], &mut map).await; + + assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1)); + } + + #[tokio::test] + async fn test_get_stream_alerts_skips_a_stream_without_alerts() { + let org = "d2_g2"; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "quiet")], &mut map).await; + + assert!( + map.is_empty(), + "a stream with no alerts must not be inserted" + ); + } + + /// Seed a stream that has alert ids registered but whose alerts are all + /// silenced, so the filter leaves an empty list. This is the state that + /// reaches the `alerts.is_empty()` branch. + async fn seed_silenced_alert(org: &str, stream: &str) { + let id = seed_alert(org, stream, true, true).await; + REALTIME_ALERT_TRIGGERS.write().await.insert( + format!("{org}/{id}"), + db::scheduler::Trigger { + is_silenced: true, + ..Default::default() + }, + ); + } + + /// D2: a stream whose alerts are all filtered out must not abort the loop + /// for the streams after it. + #[tokio::test] + async fn test_get_stream_alerts_continues_past_a_fully_filtered_stream() { + let org = "d2_g3"; + seed_silenced_alert(org, "silenced").await; + seed_alert(org, "alerted", true, true).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "silenced"), params(org, "alerted")], &mut map).await; + + assert_eq!( + map.get(&key(org, "alerted")).map(|v| v.len()), + Some(1), + "a stream whose alerts are all silenced must not suppress later streams" + ); + } + + /// An unregistered stream already takes the `continue` path; pin it so the + /// fix does not disturb it. + #[tokio::test] + async fn test_get_stream_alerts_continues_past_an_unregistered_stream() { + let org = "d2_g3b"; + seed_alert(org, "alerted", true, true).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "quiet"), params(org, "alerted")], &mut map).await; + + assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1)); + } + + #[tokio::test] + async fn test_get_stream_alerts_loads_when_alerted_stream_is_first() { + let org = "d2_g4"; + seed_alert(org, "alerted", true, true).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "alerted"), params(org, "quiet")], &mut map).await; + + assert_eq!(map.get(&key(org, "alerted")).map(|v| v.len()), Some(1)); + } + + /// D2: several fully-filtered streams in a row must not abort the loop. + #[tokio::test] + async fn test_get_stream_alerts_continues_past_multiple_filtered_streams() { + let org = "d2_g5"; + seed_silenced_alert(org, "silenced_a").await; + seed_alert(org, "disabled_b", false, true).await; + seed_alert(org, "scheduled_c", true, false).await; + seed_alert(org, "alerted", true, true).await; + + let mut map = HashMap::new(); + get_stream_alerts( + &[ + params(org, "silenced_a"), + params(org, "disabled_b"), + params(org, "scheduled_c"), + params(org, "alerted"), + ], + &mut map, + ) + .await; + + assert_eq!( + map.get(&key(org, "alerted")).map(|v| v.len()), + Some(1), + "alerts must load after any number of fully-filtered streams" + ); + } + + #[tokio::test] + async fn test_get_stream_alerts_handles_all_streams_without_alerts() { + let org = "d2_g6"; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "a"), params(org, "b")], &mut map).await; + + assert!(map.is_empty()); + } + + #[tokio::test] + async fn test_get_stream_alerts_does_not_refetch_cached_keys() { + let org = "d2_g7"; + seed_alert(org, "alerted", true, true).await; + + // Pre-seed the map with a sentinel; the function must leave it alone. + let mut map = HashMap::new(); + map.insert(key(org, "alerted"), Vec::new()); + get_stream_alerts(&[params(org, "alerted")], &mut map).await; + + assert_eq!( + map.get(&key(org, "alerted")).map(|v| v.len()), + Some(0), + "an already-cached key must not be refetched" + ); + } + + #[tokio::test] + async fn test_get_stream_alerts_filters_disabled_alerts() { + let org = "d2_g8"; + seed_alert(org, "alerted", false, true).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "alerted")], &mut map).await; + + assert!(map.is_empty(), "disabled alerts must be filtered out"); + } + + #[tokio::test] + async fn test_get_stream_alerts_filters_scheduled_alerts() { + let org = "d2_g9"; + seed_alert(org, "alerted", true, false).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "alerted")], &mut map).await; + + assert!(map.is_empty(), "non-realtime alerts must be filtered out"); + } + + #[tokio::test] + async fn test_get_stream_alerts_filters_silenced_alerts() { + let org = "d2_g10"; + let id = seed_alert(org, "alerted", true, true).await; + REALTIME_ALERT_TRIGGERS.write().await.insert( + format!("{org}/{id}"), + db::scheduler::Trigger { + is_silenced: true, + ..Default::default() + }, + ); + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "alerted")], &mut map).await; + + assert!(map.is_empty(), "silenced alerts must be filtered out"); + } + + #[tokio::test] + async fn test_get_stream_alerts_keeps_only_enabled_realtime_alerts() { + let org = "d2_g11"; + seed_alert(org, "mixed", true, true).await; + seed_alert(org, "mixed", false, true).await; + seed_alert(org, "mixed", true, false).await; + + let mut map = HashMap::new(); + get_stream_alerts(&[params(org, "mixed")], &mut map).await; + + assert_eq!( + map.get(&key(org, "mixed")).map(|v| v.len()), + Some(1), + "only the enabled real-time alert survives" + ); + } + } } diff --git a/src/core/src/ingestion/realtime_alerts.rs b/src/core/src/ingestion/realtime_alerts.rs new file mode 100644 index 0000000000..7cdef8cd46 --- /dev/null +++ b/src/core/src/ingestion/realtime_alerts.rs @@ -0,0 +1,438 @@ +// Copyright 2025 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +use config::{ + meta::alerts::alert::Alert, + utils::{ + json::{Map, Value}, + time::now_micros, + }, +}; + +use crate::{alerts::alert::AlertExt, ingestion::TriggerAlertData}; + +/// Accumulates real-time alert triggers for the records of a single stream +/// within one ingestion request. +/// +/// Every record is checked against every alert that has not yet matched, and +/// each alert contributes at most one trigger. That cap matters: +/// `evaluate_trigger` sends one notification per trigger and performs no +/// deduplication of its own, and the silence window is applied once up front in +/// `get_stream_alerts`, so a second trigger for the same alert within one +/// request is a second notification. +pub struct RealtimeAlertEvaluator<'a> { + alerts: &'a [Alert], + /// Index-aligned with `alerts`: whether that alert has already fired. + /// Indexing by position keeps the per-record path free of key building. + /// Repeated ids in the input are collapsed once in `new`, so a position that + /// repeats an earlier alert starts out marked as fired. + fired: Vec, + triggers: TriggerAlertData, + /// Alerts that have not yet fired. Reaching zero short-circuits `observe`, + /// which also covers the common "stream has no alerts" case at no cost. + remaining: usize, + /// Captured once per stream; see `observe`. + end_time: i64, +} + +impl<'a> RealtimeAlertEvaluator<'a> { + pub fn new(alerts: &'a [Alert]) -> Self { + let mut fired = vec![false; alerts.len()]; + let mut remaining = alerts.len(); + + // The same alert id can appear twice for one stream: the bulk cache load + // pushes ids unconditionally, while the incremental path checks first. + // Position alone would then treat the repeats as separate alerts and send + // two notifications for one alert from a single request, so mark repeats + // as already fired. Done once per stream, which keeps the per-record path + // free of any key building. + for i in 1..alerts.len() { + let id = alerts[i].id; + if id.is_some() && alerts[..i].iter().any(|prev| prev.id == id) { + fired[i] = true; + remaining -= 1; + } + } + + Self { + alerts, + fired, + triggers: Vec::with_capacity(alerts.len()), + remaining, + end_time: now_micros(), + } + } + + /// Check one record against every alert that has not yet fired. + pub async fn observe(&mut self, record: &Map) { + if self.remaining == 0 { + return; + } + // The real-time branch of `evaluate` ignores this time range entirely and + // takes its own timestamp, so it is computed once per stream rather than + // once per record. + let end_time = self.end_time; + let alerts = self.alerts; + for (i, alert) in alerts.iter().enumerate() { + if self.fired[i] { + continue; + } + match alert.evaluate(Some(record), (None, end_time), None).await { + Ok(res) if res.data.is_some() => { + self.triggers.push((alert.clone(), res.data.unwrap())); + self.fired[i] = true; + self.remaining -= 1; + } + Ok(_) => { + // the record does not satisfy the alert condition + } + Err(e) => { + log::error!("[METRICS] Error while evaluating realtime alert: {e}"); + } + } + } + } + + pub fn finish(self) -> TriggerAlertData { + self.triggers + } +} + +#[cfg(test)] +mod tests { + use config::utils::json::{from_value, json}; + use svix_ksuid::{Ksuid, KsuidLike}; + + use super::*; + + /// Build a real-time alert whose single condition is `column == value`. + /// + /// `Alert` carries private fields, so struct-update syntax is unavailable + /// outside its defining crate. It is `#[serde(default)]`, so building it from + /// JSON is the supported route and mirrors how alerts really arrive. + fn alert_matching(name: &str, column: &str, value: Value) -> Alert { + from_value(json!({ + "name": name, + "org_id": "test_org", + "stream_type": "metrics", + "stream_name": "test_stream", + "is_real_time": true, + "enabled": true, + "query_condition": { + "type": "custom", + "conditions": [ + {"column": column, "operator": "=", "value": value, "ignore_case": false} + ] + } + })) + .expect("valid alert json") + } + + fn record(pairs: &[(&str, Value)]) -> Map { + pairs + .iter() + .map(|(k, v)| ((*k).to_string(), v.clone())) + .collect() + } + + /// One row that matches `alert_matching("..", "condition", "DiskPressure")`. + fn matching_record() -> Map { + record(&[("condition", json!("DiskPressure")), ("value", json!(1))]) + } + + /// One row that does not match. + fn other_record() -> Map { + record(&[("condition", json!("Ready")), ("value", json!(1))]) + } + + fn disk_pressure_alert() -> Alert { + alert_matching("disk_pressure", "condition", json!("DiskPressure")) + } + + // ---- match position ---- + + #[tokio::test] + async fn test_observe_fires_on_the_only_record() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert_eq!(ev.finish().len(), 1); + } + + /// D1: the reported bug -- a match on any record after the first. + #[tokio::test] + async fn test_observe_fires_when_match_is_the_second_record() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&other_record()).await; + ev.observe(&matching_record()).await; + assert_eq!( + ev.finish().len(), + 1, + "a match on the second record must still fire" + ); + } + + #[tokio::test] + async fn test_observe_fires_when_match_is_mid_batch() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + for _ in 0..24 { + ev.observe(&other_record()).await; + } + ev.observe(&matching_record()).await; + for _ in 0..25 { + ev.observe(&other_record()).await; + } + assert_eq!(ev.finish().len(), 1, "a match mid-batch must fire"); + } + + #[tokio::test] + async fn test_observe_fires_when_match_is_the_last_record() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + for _ in 0..49 { + ev.observe(&other_record()).await; + } + ev.observe(&matching_record()).await; + assert_eq!( + ev.finish().len(), + 1, + "a match on the last record of 50 must fire" + ); + } + + #[tokio::test] + async fn test_observe_produces_no_trigger_when_nothing_matches() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + for _ in 0..10 { + ev.observe(&other_record()).await; + } + assert!(ev.finish().is_empty()); + } + + #[tokio::test] + async fn test_finish_returns_empty_when_no_records_observed() { + let alerts = vec![disk_pressure_alert()]; + let ev = RealtimeAlertEvaluator::new(&alerts); + assert!(ev.finish().is_empty()); + } + + // ---- multiplicity: at most one trigger per alert per request ---- + + #[tokio::test] + async fn test_repeated_matches_produce_exactly_one_trigger() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + for _ in 0..10 { + ev.observe(&matching_record()).await; + } + assert_eq!( + ev.finish().len(), + 1, + "ten matching records must still yield exactly one trigger" + ); + } + + #[tokio::test] + async fn test_trigger_payload_holds_only_the_first_matching_record() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + for _ in 0..10 { + ev.observe(&matching_record()).await; + } + let triggers = ev.finish(); + assert_eq!(triggers.len(), 1); + assert_eq!( + triggers[0].1.len(), + 1, + "the payload must carry one row, not every matching row" + ); + } + + #[tokio::test] + async fn test_trigger_payload_holds_the_earliest_match_across_the_batch() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + let mut first = matching_record(); + first.insert("node".to_string(), json!("node_first")); + let mut second = matching_record(); + second.insert("node".to_string(), json!("node_second")); + + ev.observe(&other_record()).await; + ev.observe(&first).await; + ev.observe(&second).await; + + let triggers = ev.finish(); + assert_eq!(triggers.len(), 1); + assert_eq!( + triggers[0].1[0].get("node"), + Some(&json!("node_first")), + "the payload must hold the earliest matching record" + ); + } + + // ---- multiple alerts ---- + + #[tokio::test] + async fn test_two_alerts_matching_the_same_record_each_fire_once() { + let alerts = vec![ + disk_pressure_alert(), + alert_matching("value_one", "value", json!(1)), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert_eq!(ev.finish().len(), 2); + } + + /// D1: a per-stream flag passes the second-record test but fails this one. + #[tokio::test] + async fn test_two_alerts_matching_different_records_both_fire() { + let alerts = vec![ + disk_pressure_alert(), + alert_matching("condition", "condition", json!("MemoryPressure")), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + ev.observe(&record(&[("condition", json!("MemoryPressure"))])) + .await; + assert_eq!( + ev.finish().len(), + 2, + "alerts matching different records must both fire" + ); + } + + #[tokio::test] + async fn test_two_alerts_fire_regardless_of_match_order() { + let alerts = vec![ + disk_pressure_alert(), + alert_matching("condition", "condition", json!("MemoryPressure")), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&record(&[("condition", json!("MemoryPressure"))])) + .await; + ev.observe(&matching_record()).await; + assert_eq!(ev.finish().len(), 2, "match order must not matter"); + } + + #[tokio::test] + async fn test_only_the_matching_alert_fires() { + let alerts = vec![ + disk_pressure_alert(), + alert_matching("condition", "condition", json!("NeverMatches")), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&other_record()).await; + ev.observe(&matching_record()).await; + + let triggers = ev.finish(); + assert_eq!(triggers.len(), 1); + assert_eq!(triggers[0].0.name, "disk_pressure"); + } + + #[tokio::test] + async fn test_no_alert_fires_when_none_match() { + let alerts = vec![ + disk_pressure_alert(), + alert_matching("condition", "condition", json!("MemoryPressure")), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&other_record()).await; + assert!(ev.finish().is_empty()); + } + + #[tokio::test] + async fn test_alerts_with_identical_conditions_fire_independently() { + let alerts = vec![ + alert_matching("copy_a", "condition", json!("DiskPressure")), + alert_matching("copy_b", "condition", json!("DiskPressure")), + ]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert_eq!( + ev.finish().len(), + 2, + "two distinct alerts with the same condition must each fire" + ); + } + + // ---- short-circuit and degenerate input ---- + + #[tokio::test] + async fn test_observe_is_a_noop_without_alerts() { + let alerts: Vec = vec![]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert!(ev.finish().is_empty()); + } + + #[tokio::test] + async fn test_observe_after_saturation_does_not_add_triggers() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + for _ in 0..5 { + ev.observe(&matching_record()).await; + } + assert_eq!(ev.finish().len(), 1); + } + + /// The alert cache can hold the same id twice for one stream, because the + /// bulk load pushes ids without checking while the incremental path checks + /// first. One alert must still produce one notification. + #[tokio::test] + async fn test_duplicate_alert_ids_produce_one_trigger() { + let mut a = disk_pressure_alert(); + a.id = Some(Ksuid::new(None, None)); + let alerts = vec![a.clone(), a]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert_eq!( + ev.finish().len(), + 1, + "the same alert listed twice must fire once" + ); + } + + /// Distinct alerts that happen to share a name are still separate alerts. + #[tokio::test] + async fn test_distinct_ids_with_same_name_both_fire() { + let mut a = disk_pressure_alert(); + let mut b = disk_pressure_alert(); + a.id = Some(Ksuid::new(None, None)); + b.id = Some(Ksuid::new(None, None)); + let alerts = vec![a, b]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&matching_record()).await; + assert_eq!(ev.finish().len(), 2); + } + + /// A record lacking the condition column must not consume the alert -- a + /// later record still has to be able to fire it. + #[tokio::test] + async fn test_record_missing_condition_column_leaves_alert_eligible() { + let alerts = vec![disk_pressure_alert()]; + let mut ev = RealtimeAlertEvaluator::new(&alerts); + ev.observe(&record(&[("unrelated", json!("x"))])).await; + ev.observe(&matching_record()).await; + assert_eq!( + ev.finish().len(), + 1, + "a record missing the column must not disqualify later records" + ); + } +} diff --git a/src/core/src/metrics/json.rs b/src/core/src/metrics/json.rs index f6573dfaef..2362fb240c 100644 --- a/src/core/src/metrics/json.rs +++ b/src/core/src/metrics/json.rs @@ -46,11 +46,10 @@ use schema::check_for_schema; use super::get_exclude_labels; use crate::{ - alerts::alert::AlertExt, common::meta::{authz::Authz, stream::SchemaRecords}, ingestion::{ TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id, - get_write_partition_key, write_file, + get_write_partition_key, realtime_alerts::RealtimeAlertEvaluator, write_file, }, pipeline::batch_execution::ExecutablePipeline, }; @@ -129,7 +128,7 @@ pub async fn ingest( // realtime alerts let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap> = HashMap::new(); + let mut stream_trigger_map: HashMap = HashMap::new(); // records buffer let mut json_data_by_stream: HashMap> = HashMap::new(); @@ -405,21 +404,29 @@ pub async fn ingest( .unwrap_or_default(); let partition_time_level = get_partition_time_level(StreamType::Metrics); - for (mut record, metric_type) in json_data { - // Start get stream alerts - if !stream_alerts_map.contains_key(&stream_name) { - crate::ingestion::get_stream_alerts( - &[StreamParams { - org_id: org_id.to_owned().into(), - stream_name: stream_name.to_owned().into(), - stream_type: StreamType::Metrics, - }], - &mut stream_alerts_map, - ) - .await; - } - // End get stream alert + // Start get stream alerts -- fetched once per stream, not once per record + let alerts_key = + db::alerts::alert::cache_stream_key(org_id, StreamType::Metrics, &stream_name); + if !stream_alerts_map.contains_key(&alerts_key) { + crate::ingestion::get_stream_alerts( + &[StreamParams { + org_id: org_id.to_owned().into(), + stream_name: stream_name.to_owned().into(), + stream_type: StreamType::Metrics, + }], + &mut stream_alerts_map, + ) + .await; + } + let mut alert_evaluator = RealtimeAlertEvaluator::new( + stream_alerts_map + .get(&alerts_key) + .map(Vec::as_slice) + .unwrap_or_default(), + ); + // End get stream alert + for (mut record, metric_type) in json_data { // check value let value = parse_metric_value(record.get(VALUE_LABEL).ok_or(anyhow!("missing value"))?)?; @@ -536,27 +543,14 @@ pub async fn ingest( .or_insert_with(|| StreamStatus::new(&stream_name)); stream_status.status.successful += 1; - // realtime alert - let need_trigger = !stream_trigger_map.contains_key(&stream_name); - if need_trigger && !stream_alerts_map.is_empty() { - // start check for alert trigger - let key = format!("{}/{}/{}", org_id, StreamType::Metrics, stream_name); - if let Some(alerts) = stream_alerts_map.get(&key) { - let mut trigger_alerts: TriggerAlertData = Vec::new(); - let alert_end_time = now_micros(); - for alert in alerts { - if let Ok(Some(data)) = alert - .evaluate(Some(&record), (None, alert_end_time), None) - .await - .map(|res| res.data) - { - trigger_alerts.push((alert.clone(), data)) - } - } - stream_trigger_map.insert(stream_name.clone(), Some(trigger_alerts)); - } - } - // End check for alert trigger + // realtime alert -- every record is checked against every alert that + // has not yet fired in this request + alert_evaluator.observe(&record).await; + } + + let triggers = alert_evaluator.finish(); + if !triggers.is_empty() { + stream_trigger_map.insert(stream_name.clone(), triggers); } } @@ -631,9 +625,9 @@ pub async fn ingest( .inc(); // only one trigger per request - for (_, entry) in stream_trigger_map { - if let Some(entry) = entry { - evaluate_trigger(entry).await; + for (_, triggers) in stream_trigger_map { + if !triggers.is_empty() { + tokio::spawn(async move { evaluate_trigger(triggers).await }); } } diff --git a/src/core/src/metrics/otlp.rs b/src/core/src/metrics/otlp.rs index b7721a940e..cfd26f71dc 100644 --- a/src/core/src/metrics/otlp.rs +++ b/src/core/src/metrics/otlp.rs @@ -40,7 +40,6 @@ use config::{ json, schema::format_stream_name, schema_ext::SchemaExt, - time::now_micros, }, }; use db; @@ -57,11 +56,11 @@ use prost::Message; use schema::{check_for_schema, stream_schema_exists}; use crate::{ - alerts::alert::AlertExt, common::meta::{http::HttpResponse as MetaHttpResponse, stream::SchemaRecords}, ingestion::{ TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id, grpc::{get_exemplar_val, get_metric_val, get_val}, + realtime_alerts::RealtimeAlertEvaluator, write_file, }, metrics::get_exclude_labels, @@ -173,7 +172,7 @@ pub async fn handle_otlp_request( // realtime alerts let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap> = HashMap::new(); + let mut stream_trigger_map: HashMap = HashMap::new(); let mut partial_success = ExportMetricsPartialSuccess::default(); @@ -536,6 +535,20 @@ pub async fn handle_otlp_request( .unwrap_or_default(); let partition_time_level = get_partition_time_level(StreamType::Metrics); + // Realtime alerts for this stream. Alerts were fetched while records were + // built, for the metric name and its derived streams; a stream that only + // appears as a pipeline destination has no entry here and evaluates nothing. + let mut alert_evaluator = RealtimeAlertEvaluator::new( + stream_alerts_map + .get(&db::alerts::alert::cache_stream_key( + org_id, + StreamType::Metrics, + &local_metric_name, + )) + .map(Vec::as_slice) + .unwrap_or_default(), + ); + // check for schema evolution let min_timestamp = batch_min_timestamp(&json_data, Utc::now().timestamp_micros()); @@ -588,27 +601,14 @@ pub async fn handle_otlp_request( .push(Arc::new(json::Value::Object(val_map.to_owned()))); hour_buf.records_size += value_str.len(); - // real time alert - let need_trigger = !stream_trigger_map.contains_key(&local_metric_name); - if need_trigger && !stream_alerts_map.is_empty() { - // Start check for alert trigger - let key = format!("{}/{}/{}", org_id, StreamType::Metrics, local_metric_name); - if let Some(alerts) = stream_alerts_map.get(&key) { - let mut trigger_alerts: TriggerAlertData = Vec::new(); - let alert_end_time = now_micros(); - for alert in alerts { - if let Ok(Some(data)) = alert - .evaluate(Some(&val_map), (None, alert_end_time), None) - .await - .map(|res| res.data) - { - trigger_alerts.push((alert.clone(), data)) - } - } - stream_trigger_map.insert(local_metric_name.clone(), Some(trigger_alerts)); - } - } - // End check for alert trigger + // real time alert -- every record is checked against every alert that + // has not yet fired in this request + alert_evaluator.observe(&val_map).await; + } + + let triggers = alert_evaluator.finish(); + if !triggers.is_empty() { + stream_trigger_map.insert(local_metric_name.clone(), triggers); } } @@ -681,9 +681,9 @@ pub async fn handle_otlp_request( .inc(); // only one trigger per request - for (_, entry) in stream_trigger_map { - if let Some(entry) = entry { - evaluate_trigger(entry).await; + for (_, triggers) in stream_trigger_map { + if !triggers.is_empty() { + tokio::spawn(async move { evaluate_trigger(triggers).await }); } } diff --git a/src/core/src/metrics/prom.rs b/src/core/src/metrics/prom.rs index f12775c151..fd5cab8e21 100644 --- a/src/core/src/metrics/prom.rs +++ b/src/core/src/metrics/prom.rs @@ -56,13 +56,13 @@ use search_service; use super::native_histogram::{CLASSIC_HISTOGRAM_SUFFIXES, expand_native_histogram}; use crate::{ - alerts::alert::AlertExt, common::{ infra::config::{METRIC_CLUSTER_LEADER, METRIC_CLUSTER_MAP}, meta::stream::SchemaRecords, }, ingestion::{ - TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id, write_file, + TriggerAlertData, check_ingestion_allowed, evaluate_trigger, get_thread_id, + realtime_alerts::RealtimeAlertEvaluator, write_file, }, pipeline::batch_execution::ExecutablePipeline, }; @@ -99,7 +99,7 @@ pub async fn remote_write( // realtime alerts let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap> = HashMap::new(); + let mut stream_trigger_map: HashMap = HashMap::new(); let decoded = snap::raw::Decoder::new() .decompress_vec(&body) @@ -559,6 +559,20 @@ pub async fn remote_write( .unwrap_or_default(); let partition_time_level = get_partition_time_level(StreamType::Metrics); + // Realtime alerts for this stream. Alerts were preloaded in bulk for the + // metric names present in the request; a stream that only appears as a + // pipeline destination has no entry here and evaluates nothing. + let mut alert_evaluator = RealtimeAlertEvaluator::new( + stream_alerts_map + .get(&db::alerts::alert::cache_stream_key( + org_id, + StreamType::Metrics, + &stream_name, + )) + .map(Vec::as_slice) + .unwrap_or_default(), + ); + for (mut val_map, timestamp) in json_data { let hash = super::signature_without_labels(&val_map, &[VALUE_LABEL]); val_map.insert(HASH_LABEL.to_string(), json::Value::Number(hash.into())); @@ -631,27 +645,14 @@ pub async fn remote_write( .push(Arc::new(json::Value::Object(val_map.to_owned()))); hour_buf.records_size += value_str.len(); - // real time alert - let need_trigger = !stream_trigger_map.contains_key(&stream_name); - if need_trigger && !stream_alerts_map.is_empty() { - // Start check for alert trigger - let key = format!("{}/{}/{}", org_id, StreamType::Metrics, stream_name); - if let Some(alerts) = stream_alerts_map.get(&key) { - let mut trigger_alerts: TriggerAlertData = Vec::new(); - let alert_end_time = now_micros(); - for alert in alerts { - if let Ok(Some(data)) = alert - .evaluate(Some(&val_map), (None, alert_end_time), None) - .await - .map(|res| res.data) - { - trigger_alerts.push((alert.clone(), data)); - } - } - stream_trigger_map.insert(stream_name.clone(), Some(trigger_alerts)); - } - } - // End check for alert trigger + // real time alert -- every record is checked against every alert that + // has not yet fired in this request + alert_evaluator.observe(&val_map).await; + } + + let triggers = alert_evaluator.finish(); + if !triggers.is_empty() { + stream_trigger_map.insert(stream_name.clone(), triggers); } } let elapsed_ms = step_start.elapsed().as_millis(); @@ -775,9 +776,9 @@ pub async fn remote_write( .inc(); // only one trigger per request - for (_, entry) in stream_trigger_map { - if let Some(entry) = entry { - evaluate_trigger(entry).await; + for (_, triggers) in stream_trigger_map { + if !triggers.is_empty() { + tokio::spawn(async move { evaluate_trigger(triggers).await }); } }