Merge branch 'main' into alerts-updates

Signed-off-by: parmesant <[email protected]>

Author: parmesant

src/alerts/alert_structs.rs

@@ -88,7 +88,7 @@ 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
         )
     }

src/alerts/mod.rs

@@ -766,25 +766,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
                 }
             }
         }

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();

src/handlers/http/alerts.rs

@@ -415,7 +415,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 +431,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

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)
         }