Merge branch 'main' into alerts-updates

parmesant · parmesant · commit 838191a763e0 · 2025-08-12T10:06:55.000+05:30
Signed-off-by: parmesant &lt;anant.v09@proton.me&gt;
diff --git a/src/alerts/alert_structs.rs b/src/alerts/alert_structs.rs
@@ -88,17 +88,10 @@ impl Context {
 
     pub(crate) fn default_disabled_string(&self) -> String {
         format!(
-            "{} is now `disabled`. No more evals will be run till it is `disabled`.",
+            "{} is now `disabled`. No more evals will be run till the sate is `disabled`.",
             self.alert_info.alert_name
         )
     }
-
-    // fn default_silenced_string(&self) -> String {
-    //     format!(
-    //         "Notifications for {} have been silenced ",
-    //         self.alert_info.alert_name
-    //     )
-    // }
 }
 
 #[derive(Debug, Clone)]
diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs
@@ -561,7 +561,6 @@ impl AlertConfig {
         let state_str = alert_json["state"].as_str().unwrap_or("resolved");
         match state_str.to_lowercase().as_str() {
             "triggered" => AlertState::Triggered,
-            // "silenced" => AlertState::Silenced,
             "resolved" => AlertState::NotTriggered,
             _ => AlertState::NotTriggered,
         }
@@ -766,25 +765,22 @@ fn extract_alias_from_expr(expr: &Expr) -> Option<(String, Option<String>)> {
             // This is an aliased expression
             let alias_name = alias_expr.name.clone();
 
-            match alias_expr.expr.as_ref() {
-                Expr::AggregateFunction(agg_func) => {
-                    let agg_name = format!("{:?}", agg_func.func);
-                    Some((agg_name, Some(alias_name)))
-                }
+            if let Expr::AggregateFunction(agg_func) = alias_expr.expr.as_ref() {
+                let agg_name = format!("{:?}", agg_func.func);
+                Some((agg_name, Some(alias_name)))
+            } else {
                 // Handle other aggregate expressions like Count, etc.
-                _ => {
-                    // Check if the inner expression is an aggregate
-                    let expr_str = format!("{:?}", alias_expr.expr);
-                    if expr_str.contains("count")
-                        || expr_str.contains("sum")
-                        || expr_str.contains("avg")
-                        || expr_str.contains("min")
-                        || expr_str.contains("max")
-                    {
-                        Some((expr_str, Some(alias_name)))
-                    } else {
-                        None
-                    }
+                // Check if the inner expression is an aggregate
+                let expr_str = format!("{:?}", alias_expr.expr);
+                if expr_str.contains("count")
+                    || expr_str.contains("sum")
+                    || expr_str.contains("avg")
+                    || expr_str.contains("min")
+                    || expr_str.contains("max")
+                {
+                    Some((expr_str, Some(alias_name)))
+                } else {
+                    None
                 }
             }
         }
diff --git a/src/alerts/target.rs b/src/alerts/target.rs
@@ -37,15 +37,13 @@ use ulid::Ulid;
 use url::Url;
 
 use crate::{
-    alerts::{AlertError, alert_traits::CallableTarget},
+    alerts::{AlertError, AlertState, Context, alert_traits::CallableTarget},
     parseable::PARSEABLE,
     storage::object_storage::target_json_path,
 };
 
 use super::ALERTS;
 
-use super::{AlertState, Context};
-
 pub static TARGETS: Lazy<TargetConfigs> = Lazy::new(|| TargetConfigs {
     target_configs: RwLock::new(HashMap::new()),
 });
@@ -571,14 +569,7 @@ impl CallableTarget for AlertManager {
                     .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
                     .into();
             }
-            AlertState::Disabled => alert["labels"]["status"] = "disabled".into(), // AlertState::Silenced => {
-                                                                                   //     alert["labels"]["status"] = "silenced".into();
-                                                                                   //     alert["annotations"]["reason"] =
-                                                                                   //         serde_json::Value::String(payload.default_silenced_string());
-                                                                                   //     // alert["endsAt"] = Utc::now()
-                                                                                   //     //     .to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
-                                                                                   //     //     .into();
-                                                                                   // }
+            AlertState::Disabled => alert["labels"]["status"] = "disabled".into(),
         };
 
         if let Err(e) = client
diff --git a/src/catalog/mod.rs b/src/catalog/mod.rs
@@ -366,8 +366,8 @@ pub async fn get_first_event(
         Mode::All | Mode::Ingest => {
             // get current snapshot
             let stream_first_event = PARSEABLE.get_stream(stream_name)?.get_first_event();
-            if let Some(event) = stream_first_event {
-                first_event_at = event;
+            if let Some(first_event) = stream_first_event {
+                first_event_at = first_event;
             } else {
                 let mut meta = storage.get_object_store_format(stream_name).await?;
                 let meta_clone = meta.clone();
diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs
@@ -344,8 +344,8 @@ pub async fn modify_alert(
     let alert = alerts.get_alert_by_id(alert_id).await?;
 
     // validate that the user has access to the tables mentioned in the query
-
     user_auth_for_query(&session_key, alert.get_query()).await?;
+
     // validate the request
     let mut new_config = alert_request.into().await?;
 
@@ -391,21 +391,19 @@ pub async fn modify_alert(
         }
     };
 
-    // remove the task
-    alerts.delete_task(alert_id).await?;
-
-    // remove alert from memory
-    alerts.delete(alert_id).await?;
-
-    // move on to saving the alert in ObjectStore
+    // save the new alert in ObjectStore
     alerts.update(new_alert).await;
-
     let path = alert_json_path(*new_alert.get_id());
-
     let store = PARSEABLE.storage.get_object_store();
     let alert_bytes = serde_json::to_vec(&new_alert.to_alert_config())?;
     store.put_object(&path, Bytes::from(alert_bytes)).await?;
 
+    // remove the task
+    alerts.delete_task(alert_id).await?;
+
+    // remove alert from memory
+    alerts.delete(alert_id).await?;
+
     let config = new_alert.to_alert_config().to_response();
 
     // start the task
@@ -415,7 +413,11 @@ pub async fn modify_alert(
 }
 
 // PUT /alerts/{alert_id}/evaluate_alert
-pub async fn evaluate_alert(alert_id: Path<Ulid>) -> Result<impl Responder, AlertError> {
+pub async fn evaluate_alert(
+    req: HttpRequest,
+    alert_id: Path<Ulid>,
+) -> Result<impl Responder, AlertError> {
+    let session_key = extract_session_key_from_req(&req)?;
     let alert_id = alert_id.into_inner();
 
     let guard = ALERTS.write().await;
@@ -427,6 +429,8 @@ pub async fn evaluate_alert(alert_id: Path<Ulid>) -> Result<impl Responder, Aler
 
     let alert = alerts.get_alert_by_id(alert_id).await?;
 
+    user_auth_for_query(&session_key, alert.get_query()).await?;
+
     let config = alert.to_alert_config().to_response();
 
     // remove task
diff --git a/src/livetail.rs b/src/livetail.rs
@@ -161,7 +161,8 @@ impl Stream for ReceiverPipe {
 impl Drop for ReceiverPipe {
     fn drop(&mut self) {
         if let Some(map) = self._ref.upgrade()
-            && let Some(pipes) = map.write().unwrap().get_mut(&self.stream)
+            && let Ok(mut guard) = map.write()
+            && let Some(pipes) = guard.get_mut(&self.stream)
         {
             pipes.retain(|x| x.id != self.id)
         }
diff --git a/src/migration/metadata_migration.rs b/src/migration/metadata_migration.rs
@@ -86,7 +86,7 @@ pub fn v2_v3(mut storage_metadata: JsonValue) -> JsonValue {
             for privilege in privileges.iter_mut() {
                 let privilege_value = privilege.get_mut("privilege");
                 if let Some(value) = privilege_value
-                    && value.as_str().unwrap() == "ingester"
+                    && matches!(value.as_str(), Some("ingester"))
                 {
                     *value = JsonValue::String("ingestor".to_string());
                 }
@@ -125,7 +125,7 @@ pub fn v3_v4(mut storage_metadata: JsonValue) -> JsonValue {
         for privilege in privileges.iter_mut() {
             let privilege_value = privilege.get_mut("privilege");
             if let Some(value) = privilege_value
-                && value.as_str().unwrap() == "ingester"
+                && matches!(value.as_str(), Some("ingester"))
             {
                 *value = JsonValue::String("ingestor".to_string());
             }

Original file line number	Diff line number	Diff line change
`@@ -88,17 +88,10 @@ impl Context {`
`88`	`88`
`89`	`89`	`pub(crate) fn default_disabled_string(&self) -> String {`
`90`	`90`	`format!(`
`91`		- "{} is now `disabled`. No more evals will be run till it is `disabled`.",
	`91`	+ "{} is now `disabled`. No more evals will be run till the sate is `disabled`.",
`92`	`92`	`self.alert_info.alert_name`
`93`	`93`	`)`
`94`	`94`	`}`
`95`		`-`
`96`		`- // fn default_silenced_string(&self) -> String {`
`97`		`- // format!(`
`98`		`- // "Notifications for {} have been silenced ",`
`99`		`- // self.alert_info.alert_name`
`100`		`- // )`
`101`		`- // }`
`102`	`95`	`}`
`103`	`96`
`104`	`97`	`#[derive(Debug, Clone)]`
Original file line number	Diff line number	Diff line change
`@@ -161,7 +161,8 @@ impl Stream for ReceiverPipe {`
`161`	`161`	`impl Drop for ReceiverPipe {`
`162`	`162`	`fn drop(&mut self) {`
`163`	`163`	`if let Some(map) = self._ref.upgrade()`
`164`		`- && let Some(pipes) = map.write().unwrap().get_mut(&self.stream)`
	`164`	`+ && let Ok(mut guard) = map.write()`
	`165`	`+ && let Some(pipes) = guard.get_mut(&self.stream)`
`165`	`166`	`{`
`166`	`167`	`pipes.retain(\|x\| x.id != self.id)`
`167`	`168`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,7 @@ pub fn v2_v3(mut storage_metadata: JsonValue) -> JsonValue {`
`86`	`86`	`for privilege in privileges.iter_mut() {`
`87`	`87`	`let privilege_value = privilege.get_mut("privilege");`
`88`	`88`	`if let Some(value) = privilege_value`
`89`		`- && value.as_str().unwrap() == "ingester"`
	`89`	`+ && matches!(value.as_str(), Some("ingester"))`
`90`	`90`	`{`
`91`	`91`	`*value = JsonValue::String("ingestor".to_string());`
`92`	`92`	`}`
`@@ -125,7 +125,7 @@ pub fn v3_v4(mut storage_metadata: JsonValue) -> JsonValue {`
`125`	`125`	`for privilege in privileges.iter_mut() {`
`126`	`126`	`let privilege_value = privilege.get_mut("privilege");`
`127`	`127`	`if let Some(value) = privilege_value`
`128`		`- && value.as_str().unwrap() == "ingester"`
	`128`	`+ && matches!(value.as_str(), Some("ingester"))`
`129`	`129`	`{`
`130`	`130`	`*value = JsonValue::String("ingestor".to_string());`
`131`	`131`	`}`