starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 00:09:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add support for OR and NOT operators in alert conditions (#6540)

Browse Source
This commit is contained in:
udev
2025-04-21 16:26:47 +05:30
committed by GitHub
parent 7e335c38fc
commit e5dfcee2dc
4 changed files with 442 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/src/meta/alerts/mod.rs
+263 -3
View File
@@ -219,7 +219,7 @@ pub struct QueryCondition {
#[serde(default)]
#[serde(rename = "type")]
pub query_type: QueryType,
pub conditions: Option<Vec<Condition>>,
pub conditions: Option<ConditionList>,
pub sql: Option<String>,
pub promql: Option<String>, // (cpu usage / cpu total)
pub promql_condition: Option<Condition>, // value >= 80
@@ -232,6 +232,85 @@ pub struct QueryCondition {
pub multi_time_range: Option<Vec<CompareHistoricData>>,
}
#[derive(Clone, Debug, Eq, PartialEq, Deserialize, Serialize, ToSchema)]
#[serde(untagged)]
pub enum ConditionList {
OrNode{or: Vec<ConditionList>},
AndNode{and: Vec<ConditionList>},
NotNode{not: Box<ConditionList>},
/// This variant handles data serialized in `Vec<Condition>`
/// where all conditions are evaluated as conjunction
LegacyConditions(Vec<ConditionList>),
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<ConditionList>,
index: usize
}
impl Iterator for ConditionListIterator {
type Item = ConditionList;
fn next(&mut self) -> Option<Self::Item> {
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<Vec<String>>,
@@ -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,
})
]
}
);
}
}
src/handler/http/models/alerts/mod.rs
+41 -7
View File
@@ -149,7 +149,9 @@ pub struct QueryCondition {
#[serde(default)]
#[serde(rename = "type")]
pub query_type: QueryType,
pub conditions: Option<Vec<Condition>>,
#[serde(with = "condition_list_depth_validator")]
pub conditions: Option<meta_alerts::ConditionList>,
pub sql: Option<String>,
pub promql: Option<String>,
pub promql_condition: Option<Condition>,
@@ -162,6 +164,42 @@ pub struct QueryCondition {
pub multi_time_range: Option<Vec<CompareHistoricData>>,
}
/// 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<S>(value: &Option<meta_alerts::ConditionList>, serializer: S) -> Result<S::Ok, S::Error>
where
S: ser::Serializer,
{
value.serialize(serializer)
}
/// Deserialize implementation with depth check
pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<meta_alerts::ConditionList>, D::Error>
where
D: de::Deserializer<'de>,
{
// First deserialize to the normal type
let condition_list: std::option::Option<meta_alerts::ConditionList> = 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<Vec<String>>,
@@ -340,9 +378,7 @@ impl From<meta_alerts::QueryCondition> 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<QueryCondition> 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()),
src/infra/src/table/alerts/mod.rs
+3 -7
View File
@@ -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<alerts::Model> for MetaAlert {
.context_attributes
.map(serde_json::from_value)
.transpose()?;
let query_conditions: Option<Vec<intermediate::QueryCondition>> = value
let query_conditions: Option<MetaConditionList> = value
.query_conditions
.map(serde_json::from_value)
.transpose()?;
@@ -115,7 +116,7 @@ impl TryFrom<alerts::Model> 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());
src/service/alerts/mod.rs
+135 -29
View File
@@ -14,7 +14,7 @@
// along with this program. If not, see <http://www.gnu.org/licenses/>.
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<String, anyhow::Error>;
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<String, anyhow::Error> {
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<String, Value>) -> bool;
}
#[async_trait]
impl ConditionExt for ConditionList {
async fn evaluate(&self, row: &Map<String, Value>) -> 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<String, Value>) -> bool {
@@ -561,31 +680,18 @@ async fn build_sql(
stream_name: &str,
stream_type: StreamType,
query_condition: &QueryCondition,
conditions: &[Condition],
conditions: &ConditionList,
) -> Result<String, anyhow::Error> {
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