feat: add API for alert preview (#2303)

src/handler/http/request/alerts/mod.rs

@@ -303,3 +303,44 @@ async fn trigger_alert(
         },
     }
 }
+
+/// PreviewAlert
+#[utoipa::path(
+    context_path = "/api",
+    tag = "Alerts",
+    operation_id = "PreviewAlert",
+    security(
+        ("Authorization"= [])
+    ),
+    params(
+        ("org_id" = String, Path, description = "Organization name"),
+        ("stream_name" = String, Path, description = "Stream name"),
+        ("alert_name" = String, Path, description = "Alert name"),
+    ),
+    responses(
+        (status = 200, description = "Success",  content_type = "application/json", body = HttpResponse),
+        (status = 404, description = "NotFound", content_type = "application/json", body = HttpResponse),
+        (status = 500, description = "Failure",  content_type = "application/json", body = HttpResponse),
+    )
+)]
+#[get("/{org_id}/{stream_name}/alerts/{alert_name}/preview")]
+async fn preview_alert(
+    path: web::Path<(String, String, String)>,
+    req: HttpRequest,
+) -> Result<HttpResponse, Error> {
+    let (org_id, stream_name, name) = path.into_inner();
+    let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
+    let stream_type = match get_stream_type_from_request(&query) {
+        Ok(v) => v.unwrap_or_default(),
+        Err(e) => {
+            return Ok(MetaHttpResponse::bad_request(e));
+        }
+    };
+    match alerts::preview(&org_id, stream_type, &stream_name, &name).await {
+        Ok(data) => Ok(MetaHttpResponse::json(data)),
+        Err(e) => match e {
+            (http::StatusCode::NOT_FOUND, e) => Ok(MetaHttpResponse::not_found(e)),
+            (_, e) => Ok(MetaHttpResponse::internal_error(e)),
+        },
+    }
+}

src/handler/http/router/mod.rs

@@ -276,10 +276,11 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) {
             .service(alerts::save_alert)
             .service(alerts::get_alert)
             .service(alerts::list_alerts)
-            .service(alerts::trigger_alert)
             .service(alerts::list_stream_alerts)
             .service(alerts::delete_alert)
             .service(alerts::enable_alert)
+            .service(alerts::trigger_alert)
+            .service(alerts::preview_alert)
             .service(alerts::templates::save_template)
             .service(alerts::templates::get_template)
             .service(alerts::templates::delete_template)

src/handler/http/router/openapi.rs

@@ -98,6 +98,7 @@ use crate::{
         request::alerts::delete_alert,
         request::alerts::enable_alert,
         request::alerts::trigger_alert,
+        request::alerts::preview_alert,
         request::alerts::templates::list_templates,
         request::alerts::templates::get_template,
         request::alerts::templates::save_template,

src/service/alerts/mod.rs

@@ -220,6 +220,28 @@ pub async fn trigger(
         .map_err(|e| (http::StatusCode::INTERNAL_SERVER_ERROR, e))
 }
 
+pub async fn preview(
+    org_id: &str,
+    stream_type: StreamType,
+    stream_name: &str,
+    name: &str,
+) -> Result<Vec<Map<String, Value>>, (http::StatusCode, anyhow::Error)> {
+    let alert = match db::alerts::get(org_id, stream_type, stream_name, name).await {
+        Ok(Some(alert)) => alert,
+        _ => {
+            return Err((
+                http::StatusCode::NOT_FOUND,
+                anyhow::anyhow!("Alert not found"),
+            ));
+        }
+    };
+    let data = alert
+        .preview()
+        .await
+        .map_err(|e| (http::StatusCode::INTERNAL_SERVER_ERROR, e))?;
+    Ok(data.unwrap_or_default())
+}
+
 impl Alert {
     pub async fn evaluate(
         &self,
@@ -228,10 +250,14 @@ impl Alert {
         if self.is_real_time {
             self.query_condition.evaluate_realtime(row).await
         } else {
-            self.query_condition.evaluate_scheduled(self).await
+            self.query_condition.evaluate_scheduled(self, false).await
         }
     }
 
+    pub async fn preview(&self) -> Result<Option<Vec<Map<String, Value>>>, anyhow::Error> {
+        self.query_condition.evaluate_scheduled(self, true).await
+    }
+
     pub async fn send_notification(
         &self,
         rows: &[Map<String, Value>],
@@ -282,8 +308,9 @@ impl QueryCondition {
     pub async fn evaluate_scheduled(
         &self,
         alert: &Alert,
+        preview_mode: bool,
     ) -> Result<Option<Vec<Map<String, Value>>>, anyhow::Error> {
-        let sql = match self.query_type {
+        let mut sql = match self.query_type {
             QueryType::Custom => {
                 if let Some(v) = self.conditions.as_ref() {
                     if self.aggregation.is_none() && v.is_empty() {
@@ -311,6 +338,28 @@ impl QueryCondition {
             }
         };
 
+        // handle preview mode
+        if preview_mode {
+            if sql.contains("SELECT * FROM") {
+                sql = sql["SELECT * FROM".len()..].to_string();
+                sql = format!(
+                    "SELECT histogram(_timestamp, '1 minute') AS zo_sql_key, count(*) AS alert_agg_value FROM {sql} GROUP BY zo_sql_key ORDER BY zo_sql_key"
+                )
+            } else {
+                sql = sql["SELECT ".len()..].to_string();
+                sql = sql[..sql.find(" HAVING ").unwrap()].to_string();
+                if sql.contains("GROUP BY") {
+                    sql = format!(
+                        "SELECT histogram(_timestamp, '1 minute') AS zo_sql_key, {sql} ,zo_sql_key ORDER BY zo_sql_key"
+                    )
+                } else {
+                    sql = format!(
+                        "SELECT histogram(_timestamp, '1 minute') AS zo_sql_key, {sql} GROUP BY zo_sql_key ORDER BY zo_sql_key"
+                    )
+                }
+            }
+        }
+
         // fire the query
         let now = Utc::now().timestamp_micros();
         let req = search::Request {
@@ -338,7 +387,7 @@ impl QueryCondition {
         let session_id = uuid::Uuid::new_v4().to_string();
         let resp =
             SearchService::search(&session_id, &alert.org_id, alert.stream_type, &req).await?;
-        if resp.total < alert.trigger_condition.threshold as usize {
+        if !preview_mode && resp.total < alert.trigger_condition.threshold as usize {
             Ok(None)
         } else {
             Ok(Some(