From e5dfcee2dc6fe69c1e9ff0164402958e1ecea83d Mon Sep 17 00:00:00 2001 From: udev Date: Mon, 21 Apr 2025 16:26:47 +0530 Subject: [PATCH] feat: add support for OR and NOT operators in alert conditions (#6540) --- src/config/src/meta/alerts/mod.rs | 266 +++++++++++++++++++++++++- src/handler/http/models/alerts/mod.rs | 48 ++++- src/infra/src/table/alerts/mod.rs | 10 +- src/service/alerts/mod.rs | 164 +++++++++++++--- 4 files changed, 442 insertions(+), 46 deletions(-) diff --git a/src/config/src/meta/alerts/mod.rs b/src/config/src/meta/alerts/mod.rs index c917e55e0a..7683cb7dec 100644 --- a/src/config/src/meta/alerts/mod.rs +++ b/src/config/src/meta/alerts/mod.rs @@ -219,7 +219,7 @@ pub struct QueryCondition { #[serde(default)] #[serde(rename = "type")] pub query_type: QueryType, - pub conditions: Option>, + pub conditions: Option, pub sql: Option, pub promql: Option, // (cpu usage / cpu total) pub promql_condition: Option, // value >= 80 @@ -232,6 +232,85 @@ pub struct QueryCondition { pub multi_time_range: Option>, } +#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize, ToSchema)] +#[serde(untagged)] +pub enum ConditionList { + OrNode{or: Vec}, + AndNode{and: Vec}, + NotNode{not: Box}, + /// This variant handles data serialized in `Vec` + /// where all conditions are evaluated as conjunction + LegacyConditions(Vec), + EndCondition(Condition) +} + +impl ConditionList { + /// Calculates the depth of the condition list tree + /// An EndCondition has depth 1 + /// Other nodes have depth 1 + max depth of their children + pub fn depth(&self) -> usize { + match self { + ConditionList::OrNode { or } => { + let mut max_child_depth = 0; + for child in or { + let child_depth = child.depth(); + max_child_depth = max_child_depth.max(child_depth); + } + max_child_depth + 1 + }, + ConditionList::AndNode { and } | ConditionList::LegacyConditions(and) => { + let mut max_child_depth = 0; + for child in and { + let child_depth = child.depth(); + max_child_depth = max_child_depth.max(child_depth); + } + max_child_depth + 1 + }, + ConditionList::NotNode { not } => { + not.depth() + 1 + }, + ConditionList::EndCondition(_) => 1, + } + } +} + +// Define a separate iterator struct for ConditionList +pub struct ConditionListIterator { + inner: Vec, + index: usize +} + +impl Iterator for ConditionListIterator { + type Item = ConditionList; + + fn next(&mut self) -> Option { + if self.index < self.inner.len() { + let item = self.inner[self.index].clone(); + self.index += 1; + Some(item) + } else { + None + } + } +} + +// Implement IntoIterator for ConditionList +impl IntoIterator for ConditionList { + type Item = ConditionList; + type IntoIter = ConditionListIterator; + + fn into_iter(self) -> Self::IntoIter { + let inner = match self { + ConditionList::OrNode { or } => or, + ConditionList::AndNode { and } => and, + ConditionList::LegacyConditions(conditions) => conditions, + ConditionList::NotNode { not } => vec![*not], + ConditionList::EndCondition(condition) => vec![ConditionList::EndCondition(condition)], + }; + ConditionListIterator { inner, index: 0 } + } +} + #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq)] pub struct Aggregation { pub group_by: Option>, @@ -385,9 +464,8 @@ impl std::fmt::Display for Operator { } #[cfg(test)] -mod tests { +mod test { use chrono::{DateTime, FixedOffset, TimeZone}; - use super::*; #[test] @@ -540,4 +618,186 @@ mod tests { assert_eq!(dt.minute() % 5, 0); assert_eq!(dt.second(), 0); } + + #[test] + fn test_deserialize_backcompat_condition_list() { + let backcompat_condition_list = r#"[ + { + "column": "level", + "operator": "=", + "value": "error", + "ignore_case": false + }, + { + "column": "job", + "operator": "=", + "value": "something", + "ignore_case": false + } + ]"#; + let expected_legacy_condition_list: ConditionList = serde_json::from_str(backcompat_condition_list).unwrap(); + assert_eq!( + expected_legacy_condition_list, + ConditionList::LegacyConditions( + vec![ + ConditionList::EndCondition(Condition{ + column: "level".into(), + operator: Operator::EqualTo, + value: Value::String("error".into()), + ignore_case: false, + }), + ConditionList::EndCondition(Condition{ + column: "job".to_string(), + operator: Operator::EqualTo, + value: Value::String("something".into()), + ignore_case: false, + }) + ] + ) + ); + } + + #[test] + fn test_deserialize_not_condition_list() { + let and_condition_list = r#"{ + "not": { + "and": [ + { + "column": "level", + "operator": "=", + "value": "error", + "ignore_case": false + }, + { + "column": "job", + "operator": "=", + "value": "something", + "ignore_case": false + } + ] + } + }"#; + let expected_not_condition_list: ConditionList = serde_json::from_str(and_condition_list).unwrap(); + assert_eq!( + expected_not_condition_list, + ConditionList::NotNode { + not: { Box::new(ConditionList::AndNode { + and: vec![ + ConditionList::EndCondition(Condition{ + column: "level".into(), + operator: Operator::EqualTo, + value: Value::String("error".into()), + ignore_case: false, + }), + ConditionList::EndCondition(Condition{ + column: "job".to_string(), + operator: Operator::EqualTo, + value: Value::String("something".into()), + ignore_case: false, + }) + ] + }) + } + } + ); + } + + #[test] + fn test_deserialize_simple_condition_list() { + let and_condition_list = r#"{ + "and": [ + { + "column": "level", + "operator": "=", + "value": "error", + "ignore_case": false + }, + { + "column": "job", + "operator": "=", + "value": "something", + "ignore_case": false + } + ]}"#; + let expected_and_condition_list: ConditionList = serde_json::from_str(and_condition_list).unwrap(); + assert_eq!( + expected_and_condition_list, + ConditionList::AndNode { + and: vec![ + ConditionList::EndCondition(Condition{ + column: "level".into(), + operator: Operator::EqualTo, + value: Value::String("error".into()), + ignore_case: false, + }), + ConditionList::EndCondition(Condition{ + column: "job".to_string(), + operator: Operator::EqualTo, + value: Value::String("something".into()), + ignore_case: false, + }) + ] + } + ); + } + + #[test] + fn test_deserialize_complex_condition_list() { + let complex_condition_list = r#"{ + "or": [ + { + "and": [ + { + "column": "column1", + "operator": "=", + "value": "value1", + "ignore_case": true + }, + { + "column": "level", + "operator": "=", + "value": "error", + "ignore_case": false + } + ] + }, + { + "column": "column3", + "operator": ">", + "value": "value3", + "ignore_case": false + } + ] + }"#; + let expected_complex_condition_list: ConditionList = serde_json::from_str(complex_condition_list).unwrap(); + assert_eq!( + expected_complex_condition_list, + ConditionList::OrNode { + or: vec![ + ConditionList::AndNode{ + and: vec![ + ConditionList::EndCondition(Condition{ + column: "column1".into(), + operator: Operator::EqualTo, + value: Value::String("value1".into()), + ignore_case: true, + }), + ConditionList::EndCondition(Condition{ + column: "level".to_string(), + operator: Operator::EqualTo, + value: Value::String("error".into()), + ignore_case: false, + }) + ] + }, + ConditionList::EndCondition(Condition{ + column: "column3".to_string(), + operator: Operator::GreaterThan, + value: Value::String("value3".into()), + ignore_case: false, + }) + ] + } + ); + } } diff --git a/src/handler/http/models/alerts/mod.rs b/src/handler/http/models/alerts/mod.rs index ccdb5c51db..3de3cc697d 100644 --- a/src/handler/http/models/alerts/mod.rs +++ b/src/handler/http/models/alerts/mod.rs @@ -149,7 +149,9 @@ pub struct QueryCondition { #[serde(default)] #[serde(rename = "type")] pub query_type: QueryType, - pub conditions: Option>, + + #[serde(with = "condition_list_depth_validator")] + pub conditions: Option, pub sql: Option, pub promql: Option, pub promql_condition: Option, @@ -162,6 +164,42 @@ pub struct QueryCondition { pub multi_time_range: Option>, } +/// Validator module to check that condition list doesn't exceed maximum depth +mod condition_list_depth_validator { + use serde::{de, ser, Deserialize, Serialize}; + use config::meta::alerts as meta_alerts; + + const MAX_DEPTH: usize = 3; + + /// Serialize implementation - simply pass through to the default serializer + pub fn serialize(value: &Option, serializer: S) -> Result + where + S: ser::Serializer, + { + value.serialize(serializer) + } + + /// Deserialize implementation with depth check + pub fn deserialize<'de, D>(deserializer: D) -> Result, D::Error> + where + D: de::Deserializer<'de>, + { + // First deserialize to the normal type + let condition_list: std::option::Option = Option::deserialize(deserializer)?; + + // Then check depth if Some + if let Some(ref list) = condition_list { + if list.depth() > MAX_DEPTH { + return Err(de::Error::custom(format!( + "Condition list exceeds maximum allowed depth of {}", MAX_DEPTH + ))); + } + } + + Ok(condition_list) + } +} + #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq)] pub struct Aggregation { pub group_by: Option>, @@ -340,9 +378,7 @@ impl From for QueryCondition { fn from(value: meta_alerts::QueryCondition) -> Self { Self { query_type: value.query_type.into(), - conditions: value - .conditions - .map(|cs| cs.into_iter().map(|c| c.into()).collect()), + conditions: value.conditions, sql: value.sql, promql: value.promql, promql_condition: value.promql_condition.map(|pc| pc.into()), @@ -513,9 +549,7 @@ impl From for meta_alerts::QueryCondition { fn from(value: QueryCondition) -> Self { Self { query_type: value.query_type.into(), - conditions: value - .conditions - .map(|cs| cs.into_iter().map(|c| c.into()).collect()), + conditions: value.conditions, sql: value.sql, promql: value.promql, promql_condition: value.promql_condition.map(|pc| pc.into()), diff --git a/src/infra/src/table/alerts/mod.rs b/src/infra/src/table/alerts/mod.rs index ac3ef26793..abf5dbaa67 100644 --- a/src/infra/src/table/alerts/mod.rs +++ b/src/infra/src/table/alerts/mod.rs @@ -20,6 +20,7 @@ use config::meta::{ alerts::{ QueryCondition as MetaQueryCondition, TriggerCondition as MetaTriggerCondition, alert::{Alert as MetaAlert, ListAlertsParams}, + ConditionList as MetaConditionList, }, folder::{Folder as MetaFolder, FolderType}, stream::StreamType as MetaStreamType, @@ -73,7 +74,7 @@ impl TryFrom for MetaAlert { .context_attributes .map(serde_json::from_value) .transpose()?; - let query_conditions: Option> = value + let query_conditions: Option = value .query_conditions .map(serde_json::from_value) .transpose()?; @@ -115,7 +116,7 @@ impl TryFrom for MetaAlert { alert.updated_at = updated_at_utc; alert.query_condition = MetaQueryCondition { query_type: query_type.into(), - conditions: query_conditions.map(|cs| cs.into_iter().map(|c| c.into()).collect()), + conditions: query_conditions, sql: value.query_sql, promql: value.query_promql, promql_condition: query_promql_condition.map(|c| c.into()), @@ -592,11 +593,6 @@ fn update_mutable_fields( let query_conditions = alert .query_condition .conditions - .map(|cs| { - cs.into_iter() - .map(intermediate::QueryCondition::from) - .collect_vec() - }) .map(serde_json::to_value) .transpose()?; let query_sql = alert.query_condition.sql.filter(|s| !s.is_empty()); diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index dba8469016..f8a0e1ef16 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -14,7 +14,7 @@ // along with this program. If not, see . use alert::to_float; -use arrow_schema::DataType; +use arrow_schema::{Schema, DataType}; use async_trait::async_trait; use chrono::{Duration, Utc}; use config::{ @@ -37,6 +37,7 @@ use config::{ use super::promql; use crate::service::{search as SearchService, self_reporting::http_report_metrics}; +use config::meta::alerts::ConditionList; pub mod alert; pub mod derived_streams; @@ -86,14 +87,9 @@ impl QueryConditionExt for QueryCondition { return Ok(eval_results); } let conditions = self.conditions.as_ref().unwrap(); - if conditions.is_empty() { + if !conditions.evaluate(row).await { return Ok(eval_results); } - for condition in conditions.iter() { - if !condition.evaluate(row).await { - return Ok(eval_results); - } - } eval_results.data = Some(vec![row.to_owned()]); return Ok(eval_results); } @@ -484,11 +480,134 @@ impl QueryConditionExt for QueryCondition { } } + +#[async_trait] +pub trait ConditionListExt: Sync + Send + 'static { + async fn len(&self) -> u32; + async fn to_sql(&self, schema: &Schema) -> Result; + async fn is_empty(&self) -> bool; +} + +#[async_trait] +impl ConditionListExt for ConditionList { + /// Returns end node count of a Condition list + async fn len(&self) -> u32 { + match self { + ConditionList::OrNode{ or: conditions} | + ConditionList::LegacyConditions(conditions) | + ConditionList::AndNode{ and: conditions} => { + let mut count = 0; + for condition in conditions.iter() { + count += condition.len().await + } + count + } + ConditionList::NotNode{ not: inner } => inner.len().await, + ConditionList::EndCondition(_) => 1 + } + } + + /// Converts Condition list to SQL query as per schema + async fn to_sql(&self, schema: &Schema) -> Result { + match self { + ConditionList::OrNode{or:conditions} => { + let mut cond_sql_list = Vec::new(); + for condition in conditions.iter() { + cond_sql_list.push(condition.to_sql(schema).await?); + } + Ok(format!("({})",cond_sql_list.join(" OR "))) + } + ConditionList::LegacyConditions(conditions) | + ConditionList::AndNode{and:conditions} => { + let mut cond_sql_list = Vec::new(); + for condition in conditions.iter() { + cond_sql_list.push(condition.to_sql(schema).await?); + } + Ok(format!("({})",cond_sql_list.join(" AND "))) + } + ConditionList::NotNode{not:inner} => { + Ok(format!("NOT ({})", inner.to_sql(schema).await?)) + } + ConditionList::EndCondition(node) => { + let data_type = match schema.field_with_name(&node.column) { + Ok(field) => field.data_type(), + Err(_) => { + return Err(anyhow::anyhow!( + "Column {} not found", + &node.column, + )); + } + }; + build_expr(node, "", data_type) + } + } + } + + async fn is_empty(&self) -> bool { + match self { + ConditionList::OrNode{or:conditions} => { + for condition in conditions.iter() { + if condition.is_empty().await { + return true; + } + } + false + }, + ConditionList::AndNode{and:conditions} => { + for condition in conditions.iter() { + if !condition.is_empty().await { + return false; + } + } + true + }, + ConditionList::NotNode{not:inner} => inner.is_empty().await, + ConditionList::LegacyConditions(conditions) => { + for condition in conditions.iter() { + if !condition.is_empty().await { + return false; + } + } + true + }, + ConditionList::EndCondition(_) => false + } + } +} + #[async_trait] pub trait ConditionExt: Sync + Send + 'static { async fn evaluate(&self, row: &Map) -> bool; } +#[async_trait] +impl ConditionExt for ConditionList { + async fn evaluate(&self, row: &Map) -> bool { + match self { + ConditionList::OrNode { or: conditions} => { + let mut eval = false; + for condition in conditions { + eval = eval || condition.evaluate(row).await + } + eval + } + ConditionList::AndNode{ and: conditions } | ConditionList::LegacyConditions(conditions) => { + let mut eval = true; + for condition in conditions { + eval = eval && condition.evaluate(row).await + } + eval + } + ConditionList::NotNode{ not: conditions } => { + !conditions.evaluate(row).await + } + ConditionList::EndCondition(condition) => { + condition.evaluate(row).await + } + } + } +} + #[async_trait] impl ConditionExt for Condition { async fn evaluate(&self, row: &Map) -> bool { @@ -561,31 +680,18 @@ async fn build_sql( stream_name: &str, stream_type: StreamType, query_condition: &QueryCondition, - conditions: &[Condition], + conditions: &ConditionList, ) -> Result { let schema = infra::schema::get(org_id, stream_name, stream_type).await?; - let mut wheres = Vec::with_capacity(conditions.len()); - for cond in conditions.iter() { - let data_type = match schema.field_with_name(&cond.column) { - Ok(field) => field.data_type(), - Err(_) => { - return Err(anyhow::anyhow!( - "Column {} not found on stream {}", - &cond.column, - stream_name - )); - } - }; - let expr = build_expr(cond, "", data_type)?; - wheres.push(expr); - } - let where_sql = if !wheres.is_empty() { - format!("WHERE {}", wheres.join(" AND ")) - } else { - String::new() - }; + let where_sql = conditions.to_sql(&schema).await.map_err(|err| { + anyhow::anyhow!( + "Error building SQL on stream {}: {}", + &stream_name, + err + ) + })?; if query_condition.aggregation.is_none() { - return Ok(format!("SELECT * FROM \"{}\" {}", stream_name, where_sql)); + return Ok(format!("SELECT * FROM \"{}\" WHERE {}", stream_name, where_sql)); } // handle aggregation