chore: hotfix for alerts (#1283)

- Added helper methods to fetch base query, eval details
- Ensured DataFusion accepts columns and table names as is
- Removed clockwerk scheduler
- Modifed alert config JSON
- Modified triggered alert message

Author: parmesant

.github/workflows/coverage.yaml

@@ -62,9 +62,35 @@ jobs:
             exit 1
           fi
 
+      - name: Find and fix librdkafka CMakeLists.txt
+        run: |
+          # Download the package first so it's in the registry
+          cargo fetch
+          
+          # Find the rdkafka-sys package directory
+          RDKAFKA_SYS_DIR=$(find ~/.cargo/registry/src -name "rdkafka-sys-*" -type d | head -n 1)
+          echo "Found rdkafka-sys at: $RDKAFKA_SYS_DIR"
+          
+          # Find the librdkafka CMakeLists.txt file
+          CMAKE_FILE="$RDKAFKA_SYS_DIR/librdkafka/CMakeLists.txt"
+          
+          if [ -f "$CMAKE_FILE" ]; then
+            echo "Found CMakeLists.txt at: $CMAKE_FILE"
+            
+            # Make a backup of the original file
+            cp "$CMAKE_FILE" "$CMAKE_FILE.bak"
+            
+            # Replace the minimum required version
+            sed -i 's/cmake_minimum_required(VERSION 3.2)/cmake_minimum_required(VERSION 3.5)/' "$CMAKE_FILE"
+            
+            echo "Modified CMakeLists.txt - before and after comparison:"
+            diff "$CMAKE_FILE.bak" "$CMAKE_FILE" || true
+          else
+            echo "Could not find librdkafka CMakeLists.txt file!"
+            exit 1
+          fi
+
       - name: Check with clippy
-        env:
-          CMAKE_FLAGS: "-DCMAKE_POLICY_VERSION_MINIMUM=3.5"
         run: cargo hack clippy --verbose --each-feature --no-dev-deps -- -D warnings
 
       - name: Test default feature set

src/alerts/alerts_utils.rs

@@ -25,12 +25,13 @@ use datafusion::{
         min_max::{max, min},
         sum::sum,
     },
+    logical_expr::{BinaryExpr, Literal, Operator},
     prelude::{col, lit, DataFrame, Expr},
 };
 use tracing::trace;
 
 use crate::{
-    alerts::AggregateCondition,
+    alerts::LogicalOperator,
     parseable::PARSEABLE,
     query::{TableScanVisitor, QUERY_SESSION},
     rbac::{
@@ -42,8 +43,8 @@ use crate::{
 };
 
 use super::{
-    AggregateConfig, AggregateOperation, AggregateResult, Aggregations, AlertConfig, AlertError,
-    AlertOperator, AlertState, ConditionConfig, Conditions, ALERTS,
+    AggregateConfig, AggregateFunction, AggregateResult, Aggregates, AlertConfig, AlertError,
+    AlertOperator, AlertState, ConditionConfig, Conditions, WhereConfigOperator, ALERTS,
 };
 
 async fn get_tables_from_query(query: &str) -> Result<TableScanVisitor, AlertError> {
@@ -102,23 +103,23 @@ pub async fn evaluate_alert(alert: &AlertConfig) -> Result<(), AlertError> {
     trace!("RUNNING EVAL TASK FOR- {alert:?}");
 
     let query = prepare_query(alert).await?;
-    let base_df = execute_base_query(&query, &alert.query).await?;
-    let agg_results = evaluate_aggregates(&alert.aggregate_config, &base_df).await?;
-    let final_res = calculate_final_result(&alert.aggregate_config, &agg_results);
+    let select_query = alert.get_base_query();
+    let base_df = execute_base_query(&query, &select_query).await?;
+    let agg_results = evaluate_aggregates(&alert.aggregates, &base_df).await?;
+    let final_res = calculate_final_result(&alert.aggregates, &agg_results);
 
     update_alert_state(alert, final_res, &agg_results).await?;
     Ok(())
 }
 
 async fn prepare_query(alert: &AlertConfig) -> Result<crate::query::Query, AlertError> {
-    let (start_time, end_time) = match &alert.eval_type {
-        super::EvalConfig::RollingWindow(rolling_window) => {
-            (&rolling_window.eval_start, &rolling_window.eval_end)
-        }
+    let (start_time, end_time) = match &alert.eval_config {
+        super::EvalConfig::RollingWindow(rolling_window) => (&rolling_window.eval_start, "now"),
     };
 
     let session_state = QUERY_SESSION.state();
-    let raw_logical_plan = session_state.create_logical_plan(&alert.query).await?;
+    let select_query = alert.get_base_query();
+    let raw_logical_plan = session_state.create_logical_plan(&select_query).await?;
 
     let time_range = TimeRange::parse_human_time(start_time, end_time)
         .map_err(|err| AlertError::CustomError(err.to_string()))?;
@@ -146,15 +147,15 @@ async fn execute_base_query(
 }
 
 async fn evaluate_aggregates(
-    agg_config: &Aggregations,
+    agg_config: &Aggregates,
     base_df: &DataFrame,
 ) -> Result<Vec<AggregateResult>, AlertError> {
     let agg_filter_exprs = get_exprs(agg_config);
     let mut results = Vec::new();
 
     let conditions = match &agg_config.operator {
-        Some(_) => &agg_config.aggregate_conditions[0..2],
-        None => &agg_config.aggregate_conditions[0..1],
+        Some(_) => &agg_config.aggregate_config[0..2],
+        None => &agg_config.aggregate_config[0..1],
     };
 
     for ((agg_expr, filter), agg) in agg_filter_exprs.into_iter().zip(conditions) {
@@ -186,10 +187,10 @@ async fn evaluate_single_aggregate(
     let result = evaluate_condition(&agg.operator, final_value, agg.value);
 
     let message = if result {
-        agg.condition_config
+        agg.conditions
             .as_ref()
             .map(|config| config.generate_filter_message())
-            .or(Some(String::default()))
+            .or(None)
     } else {
         None
     };
@@ -206,18 +207,17 @@ fn evaluate_condition(operator: &AlertOperator, actual: f64, expected: f64) -> b
     match operator {
         AlertOperator::GreaterThan => actual > expected,
         AlertOperator::LessThan => actual < expected,
-        AlertOperator::EqualTo => actual == expected,
-        AlertOperator::NotEqualTo => actual != expected,
-        AlertOperator::GreaterThanEqualTo => actual >= expected,
-        AlertOperator::LessThanEqualTo => actual <= expected,
-        _ => unreachable!(),
+        AlertOperator::Equal => actual == expected,
+        AlertOperator::NotEqual => actual != expected,
+        AlertOperator::GreaterThanOrEqual => actual >= expected,
+        AlertOperator::LessThanOrEqual => actual <= expected,
     }
 }
 
-fn calculate_final_result(agg_config: &Aggregations, results: &[AggregateResult]) -> bool {
+fn calculate_final_result(agg_config: &Aggregates, results: &[AggregateResult]) -> bool {
     match &agg_config.operator {
-        Some(AggregateCondition::And) => results.iter().all(|r| r.result),
-        Some(AggregateCondition::Or) => results.iter().any(|r| r.result),
+        Some(LogicalOperator::And) => results.iter().all(|r| r.result),
+        Some(LogicalOperator::Or) => results.iter().any(|r| r.result),
         None => results.first().is_some_and(|r| r.result),
     }
 }
@@ -228,8 +228,12 @@ async fn update_alert_state(
     agg_results: &[AggregateResult],
 ) -> Result<(), AlertError> {
     if final_res {
-        trace!("ALERT!!!!!!");
         let message = format_alert_message(agg_results);
+        let message = format!(
+            "{message}\nEvaluation Window: {}\nEvaluation Frequency: {}m",
+            alert.get_eval_window(),
+            alert.get_eval_frequency()
+        );
         ALERTS
             .update_state(alert.id, AlertState::Triggered, Some(message))
             .await
@@ -249,8 +253,8 @@ fn format_alert_message(agg_results: &[AggregateResult]) -> String {
     for result in agg_results {
         if let Some(msg) = &result.message {
             message.extend([format!(
-                "|{}({}) WHERE ({}) {} {} (ActualValue: {})|",
-                result.config.agg,
+                "\nCondition: {}({}) WHERE ({}) {} {}\nActualValue: {}\n",
+                result.config.aggregate_function,
                 result.config.column,
                 msg,
                 result.config.operator,
@@ -259,8 +263,8 @@ fn format_alert_message(agg_results: &[AggregateResult]) -> String {
             )]);
         } else {
             message.extend([format!(
-                "|{}({}) {} {} (ActualValue: {})",
-                result.config.agg,
+                "\nCondition: {}({}) {} {}\nActualValue: {}\n",
+                result.config.aggregate_function,
                 result.config.column,
                 result.config.operator,
                 result.config.value,
@@ -305,17 +309,17 @@ fn get_final_value(aggregated_rows: Vec<RecordBatch>) -> f64 {
 /// returns a tuple of (aggregate expressions, filter expressions)
 ///
 /// It calls get_filter_expr() to get filter expressions
-fn get_exprs(aggregate_config: &Aggregations) -> Vec<(Expr, Option<Expr>)> {
+fn get_exprs(aggregate_config: &Aggregates) -> Vec<(Expr, Option<Expr>)> {
     let mut agg_expr = Vec::new();
 
     match &aggregate_config.operator {
         Some(op) => match op {
-            AggregateCondition::And | AggregateCondition::Or => {
-                let agg1 = &aggregate_config.aggregate_conditions[0];
-                let agg2 = &aggregate_config.aggregate_conditions[1];
+            LogicalOperator::And | LogicalOperator::Or => {
+                let agg1 = &aggregate_config.aggregate_config[0];
+                let agg2 = &aggregate_config.aggregate_config[1];
 
                 for agg in [agg1, agg2] {
-                    let filter_expr = if let Some(where_clause) = &agg.condition_config {
+                    let filter_expr = if let Some(where_clause) = &agg.conditions {
                         let fe = get_filter_expr(where_clause);
 
                         trace!("filter_expr-\n{fe:?}");
@@ -331,9 +335,9 @@ fn get_exprs(aggregate_config: &Aggregations) -> Vec<(Expr, Option<Expr>)> {
             }
         },
         None => {
-            let agg = &aggregate_config.aggregate_conditions[0];
+            let agg = &aggregate_config.aggregate_config[0];
 
-            let filter_expr = if let Some(where_clause) = &agg.condition_config {
+            let filter_expr = if let Some(where_clause) = &agg.conditions {
                 let fe = get_filter_expr(where_clause);
 
                 trace!("filter_expr-\n{fe:?}");
@@ -353,23 +357,23 @@ fn get_exprs(aggregate_config: &Aggregations) -> Vec<(Expr, Option<Expr>)> {
 fn get_filter_expr(where_clause: &Conditions) -> Expr {
     match &where_clause.operator {
         Some(op) => match op {
-            AggregateCondition::And => {
+            LogicalOperator::And => {
                 let mut expr = Expr::Literal(datafusion::scalar::ScalarValue::Boolean(Some(true)));
 
-                let expr1 = &where_clause.conditions[0];
-                let expr2 = &where_clause.conditions[1];
+                let expr1 = &where_clause.condition_config[0];
+                let expr2 = &where_clause.condition_config[1];
 
                 for e in [expr1, expr2] {
                     let ex = match_alert_operator(e);
                     expr = expr.and(ex);
                 }
                 expr
             }
-            AggregateCondition::Or => {
+            LogicalOperator::Or => {
                 let mut expr = Expr::Literal(datafusion::scalar::ScalarValue::Boolean(Some(false)));
 
-                let expr1 = &where_clause.conditions[0];
-                let expr2 = &where_clause.conditions[1];
+                let expr1 = &where_clause.condition_config[0];
+                let expr2 = &where_clause.condition_config[1];
 
                 for e in [expr1, expr2] {
                     let ex = match_alert_operator(e);
@@ -379,30 +383,86 @@ fn get_filter_expr(where_clause: &Conditions) -> Expr {
             }
         },
         None => {
-            let expr = &where_clause.conditions[0];
+            let expr = &where_clause.condition_config[0];
             match_alert_operator(expr)
         }
     }
 }
 
 fn match_alert_operator(expr: &ConditionConfig) -> Expr {
+    // the form accepts value as a string
+    // if it can be parsed as a number, then parse it
+    // else keep it as a string
+    let value = NumberOrString::from_string(expr.value.clone());
+
+    // for maintaining column case
+    let column = format!(r#""{}""#, expr.column);
     match expr.operator {
-        AlertOperator::GreaterThan => col(&expr.column).gt(lit(&expr.value)),
-        AlertOperator::LessThan => col(&expr.column).lt(lit(&expr.value)),
-        AlertOperator::EqualTo => col(&expr.column).eq(lit(&expr.value)),
-        AlertOperator::NotEqualTo => col(&expr.column).not_eq(lit(&expr.value)),
-        AlertOperator::GreaterThanEqualTo => col(&expr.column).gt_eq(lit(&expr.value)),
-        AlertOperator::LessThanEqualTo => col(&expr.column).lt_eq(lit(&expr.value)),
-        AlertOperator::Like => col(&expr.column).like(lit(&expr.value)),
-        AlertOperator::NotLike => col(&expr.column).not_like(lit(&expr.value)),
+        WhereConfigOperator::Equal => col(column).eq(lit(value)),
+        WhereConfigOperator::NotEqual => col(column).not_eq(lit(value)),
+        WhereConfigOperator::LessThan => col(column).lt(lit(value)),
+        WhereConfigOperator::GreaterThan => col(column).gt(lit(value)),
+        WhereConfigOperator::LessThanOrEqual => col(column).lt_eq(lit(value)),
+        WhereConfigOperator::GreaterThanOrEqual => col(column).gt_eq(lit(value)),
+        WhereConfigOperator::IsNull => col(column).is_null(),
+        WhereConfigOperator::IsNotNull => col(column).is_not_null(),
+        WhereConfigOperator::ILike => col(column).ilike(lit(&expr.value)),
+        WhereConfigOperator::Contains => col(column).like(lit(&expr.value)),
+        WhereConfigOperator::BeginsWith => Expr::BinaryExpr(BinaryExpr::new(
+            Box::new(col(column)),
+            Operator::RegexIMatch,
+            Box::new(lit(format!("^{}", expr.value))),
+        )),
+        WhereConfigOperator::EndsWith => Expr::BinaryExpr(BinaryExpr::new(
+            Box::new(col(column)),
+            Operator::RegexIMatch,
+            Box::new(lit(format!("{}$", expr.value))),
+        )),
+        WhereConfigOperator::DoesNotContain => col(column).not_ilike(lit(&expr.value)),
+        WhereConfigOperator::DoesNotBeginWith => Expr::BinaryExpr(BinaryExpr::new(
+            Box::new(col(column)),
+            Operator::RegexNotIMatch,
+            Box::new(lit(format!("^{}", expr.value))),
+        )),
+        WhereConfigOperator::DoesNotEndWith => Expr::BinaryExpr(BinaryExpr::new(
+            Box::new(col(column)),
+            Operator::RegexNotIMatch,
+            Box::new(lit(format!("{}$", expr.value))),
+        )),
     }
 }
+
 fn match_aggregate_operation(agg: &AggregateConfig) -> Expr {
-    match agg.agg {
-        AggregateOperation::Avg => avg(col(&agg.column)),
-        AggregateOperation::Count => count(col(&agg.column)),
-        AggregateOperation::Min => min(col(&agg.column)),
-        AggregateOperation::Max => max(col(&agg.column)),
-        AggregateOperation::Sum => sum(col(&agg.column)),
+    // for maintaining column case
+    let column = format!(r#""{}""#, agg.column);
+    match agg.aggregate_function {
+        AggregateFunction::Avg => avg(col(column)),
+        AggregateFunction::Count => count(col(column)),
+        AggregateFunction::Min => min(col(column)),
+        AggregateFunction::Max => max(col(column)),
+        AggregateFunction::Sum => sum(col(column)),
+    }
+}
+
+enum NumberOrString {
+    Number(f64),
+    String(String),
+}
+
+impl Literal for NumberOrString {
+    fn lit(&self) -> Expr {
+        match self {
+            NumberOrString::Number(expr) => lit(*expr),
+            NumberOrString::String(expr) => lit(expr.clone()),
+        }
+    }
+}
+impl NumberOrString {
+    fn from_string(value: String) -> Self {
+        if let Ok(num) = value.parse::<f64>() {
+            NumberOrString::Number(num)
+        } else {
+            NumberOrString::String(value)
+        }
     }
 }