hotfix for alerts

Added new actions

Author: parmesant

.github/workflows/coverage.yaml

@@ -33,10 +33,20 @@ jobs:
             pkg-config \
             build-essential
         if: runner.os == 'Linux'
+      
+      - name: Patch librdkafka CMakeLists.txt
+        run: |
+          RDKAFKA_DIR=$(find ~/.cargo/registry/src -path "*/rdkafka-sys-*/librdkafka" -type d | head -n 1)
+          if [ -n "$RDKAFKA_DIR" ]; then
+            echo "Found librdkafka at $RDKAFKA_DIR"
+            sed -i 's/cmake_minimum_required(VERSION 3.2)/cmake_minimum_required(VERSION 3.5)/' "$RDKAFKA_DIR/CMakeLists.txt"
+            echo "Updated minimum CMake version to 3.5"
+          else
+            echo "Could not find librdkafka directory"
+            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

@@ -102,7 +102,8 @@ 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 select_query = format!("SELECT * FROM {}", alert.stream);
+    let base_df = execute_base_query(&query, &select_query).await?;
     let agg_results = evaluate_aggregates(&alert.aggregate_config, &base_df).await?;
     let final_res = calculate_final_result(&alert.aggregate_config, &agg_results);
 
@@ -118,7 +119,8 @@ async fn prepare_query(alert: &AlertConfig) -> Result<crate::query::Query, Alert
     };
 
     let session_state = QUERY_SESSION.state();
-    let raw_logical_plan = session_state.create_logical_plan(&alert.query).await?;
+    let select_query = format!("SELECT * FROM {}", alert.stream);
+    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()))?;

src/alerts/mod.rs

@@ -390,7 +390,7 @@ pub struct AlertRequest {
     #[serde(default = "Severity::default")]
     pub severity: Severity,
     pub title: String,
-    pub query: String,
+    pub stream: String,
     pub alert_type: AlertType,
     pub aggregate_config: Aggregations,
     pub eval_type: EvalConfig,
@@ -404,7 +404,7 @@ impl From<AlertRequest> for AlertConfig {
             id: Ulid::new(),
             severity: val.severity,
             title: val.title,
-            query: val.query,
+            stream: val.stream,
             alert_type: val.alert_type,
             aggregate_config: val.aggregate_config,
             eval_type: val.eval_type,
@@ -422,7 +422,7 @@ pub struct AlertConfig {
     pub id: Ulid,
     pub severity: Severity,
     pub title: String,
-    pub query: String,
+    pub stream: String,
     pub alert_type: AlertType,
     pub aggregate_config: Aggregations,
     pub eval_type: EvalConfig,
@@ -435,7 +435,7 @@ pub struct AlertConfig {
 impl AlertConfig {
     pub fn modify(&mut self, alert: AlertRequest) {
         self.title = alert.title;
-        self.query = alert.query;
+        self.stream = alert.stream;
         self.alert_type = alert.alert_type;
         self.aggregate_config = alert.aggregate_config;
         self.eval_type = alert.eval_type;
@@ -480,28 +480,14 @@ impl AlertConfig {
         self.validate_configs()?;
 
         let session_state = QUERY_SESSION.state();
-        let raw_logical_plan = session_state.create_logical_plan(&self.query).await?;
+        let select_query = format!("SELECT * FROM {}", self.stream);
+
+        let raw_logical_plan = session_state.create_logical_plan(&select_query).await?;
 
         // create a visitor to extract the table names present in query
         let mut visitor = TableScanVisitor::default();
         let _ = raw_logical_plan.visit(&mut visitor);
 
-        let table = visitor.into_inner().first().unwrap().to_owned();
-
-        let lowercase = self.query.split(&table).collect_vec()[0].to_lowercase();
-
-        if lowercase
-            .strip_prefix(" ")
-            .unwrap_or(&lowercase)
-            .strip_suffix(" ")
-            .unwrap_or(&lowercase)
-            .ne("select * from")
-        {
-            return Err(AlertError::Metadata(
-                "Query needs to be select * from <logstream>",
-            ));
-        }
-
         // TODO: Filter tags should be taken care of!!!
         let time_range = TimeRange::parse_human_time("1m", "now")
             .map_err(|err| AlertError::CustomError(err.to_string()))?;
@@ -517,7 +503,7 @@ impl AlertConfig {
         let Some(stream_name) = query.first_table_name() else {
             return Err(AlertError::CustomError(format!(
                 "Table name not found in query- {}",
-                self.query
+                select_query
             )));
         };
 
@@ -766,8 +752,8 @@ impl Alerts {
         let mut alerts: Vec<AlertConfig> = Vec::new();
         for (_, alert) in self.alerts.read().await.iter() {
             // filter based on whether the user can execute this query or not
-            let query = &alert.query;
-            if user_auth_for_query(&session, query).await.is_ok() {
+            let query = format!("SELECT * from {}", &alert.stream);
+            if user_auth_for_query(&session, &query).await.is_ok() {
                 alerts.push(alert.to_owned());
             }
         }

src/handlers/http/alerts.rs

@@ -53,7 +53,8 @@ pub async fn post(
     // validate the incoming alert query
     // does the user have access to these tables or not?
     let session_key = extract_session_key_from_req(&req)?;
-    user_auth_for_query(&session_key, &alert.query).await?;
+    let query = format!("SELECT * FROM {}", alert.stream);
+    user_auth_for_query(&session_key, &query).await?;
 
     // create scheduled tasks
     let (outbox_tx, outbox_rx) = oneshot::channel::<()>();
@@ -89,7 +90,8 @@ pub async fn get(req: HttpRequest, alert_id: Path<Ulid>) -> Result<impl Responde
 
     let alert = ALERTS.get_alert_by_id(alert_id).await?;
     // validate that the user has access to the tables mentioned
-    user_auth_for_query(&session_key, &alert.query).await?;
+    let query = format!("SELECT * FROM {}", alert.stream);
+    user_auth_for_query(&session_key, &query).await?;
 
     Ok(web::Json(alert))
 }
@@ -103,7 +105,8 @@ pub async fn delete(req: HttpRequest, alert_id: Path<Ulid>) -> Result<impl Respo
     let alert = ALERTS.get_alert_by_id(alert_id).await?;
 
     // validate that the user has access to the tables mentioned
-    user_auth_for_query(&session_key, &alert.query).await?;
+    let query = format!("SELECT * FROM {}", alert.stream);
+    user_auth_for_query(&session_key, &query).await?;
 
     let store = PARSEABLE.storage.get_object_store();
     let alert_path = alert_json_path(alert_id);
@@ -139,8 +142,9 @@ pub async fn modify(
 
     // validate that the user has access to the tables mentioned
     // in the old as well as the modified alert
-    user_auth_for_query(&session_key, &alert.query).await?;
-    user_auth_for_query(&session_key, &alert_request.query).await?;
+    let query = format!("SELECT * FROM {}", alert.stream);
+    user_auth_for_query(&session_key, &query).await?;
+    user_auth_for_query(&session_key, &alert_request.stream).await?;
 
     alert.modify(alert_request);
     alert.validate().await?;
@@ -185,7 +189,8 @@ pub async fn update_state(
     let alert = ALERTS.get_alert_by_id(alert_id).await?;
 
     // validate that the user has access to the tables mentioned
-    user_auth_for_query(&session_key, &alert.query).await?;
+    let query = format!("SELECT * FROM {}", alert.stream);
+    user_auth_for_query(&session_key, &query).await?;
 
     // get current state
     let current_state = ALERTS.get_state(alert_id).await?;

src/rbac/role.rs

@@ -48,6 +48,10 @@ pub enum Action {
     ListRole,
     GetAbout,
     QueryLLM,
+    AddLLM,
+    DeleteLLM,
+    GetLLM,
+    ListLLM,
     ListCluster,
     ListClusterMetrics,
     Deleteingestor,
@@ -115,6 +119,10 @@ impl RoleBuilder {
                 | Action::DeleteUser
                 | Action::GetAbout
                 | Action::QueryLLM
+                | Action::AddLLM
+                | Action::DeleteLLM
+                | Action::GetLLM
+                | Action::ListLLM
                 | Action::PutRole
                 | Action::GetRole
                 | Action::DeleteRole
@@ -234,6 +242,7 @@ pub mod model {
                 Action::GetAlert,
                 Action::DeleteAlert,
                 Action::QueryLLM,
+                Action::GetLLM,
                 Action::CreateFilter,
                 Action::ListFilter,
                 Action::GetFilter,
@@ -276,6 +285,7 @@ pub mod model {
                 Action::DeleteDashboard,
                 Action::Ingest,
                 Action::QueryLLM,
+                Action::GetLLM,
                 Action::GetStreamInfo,
                 Action::GetFilter,
                 Action::ListFilter,
@@ -298,6 +308,7 @@ pub mod model {
                 Action::GetSchema,
                 Action::GetStats,
                 Action::QueryLLM,
+                Action::GetLLM,
                 Action::ListFilter,
                 Action::GetFilter,
                 Action::CreateFilter,