bugfix and alerts retry

Author: parmesant

src/alerts/mod.rs

@@ -32,7 +32,7 @@ use std::thread;
 use tokio::sync::oneshot::{Receiver, Sender};
 use tokio::sync::{mpsc, RwLock};
 use tokio::task::JoinHandle;
-use tracing::{trace, warn};
+use tracing::{error, trace, warn};
 use ulid::Ulid;
 
 pub mod alerts_utils;
@@ -52,7 +52,7 @@ pub type ScheduledTaskHandlers = (JoinHandle<()>, Receiver<()>, Sender<()>);
 pub const CURRENT_ALERTS_VERSION: &str = "v1";
 
 pub static ALERTS: Lazy<Alerts> = Lazy::new(|| {
-    let (tx, rx) = mpsc::channel::<AlertTask>(1);
+    let (tx, rx) = mpsc::channel::<AlertTask>(10);
     let alerts = Alerts {
         alerts: RwLock::new(HashMap::new()),
         sender: tx,
@@ -659,7 +659,7 @@ impl AlertConfig {
 
                 // validate condition config
                 let agg1 = &self.aggregates.aggregate_config[0];
-                let agg2 = &self.aggregates.aggregate_config[0];
+                let agg2 = &self.aggregates.aggregate_config[1];
 
                 validate_condition_config(&agg1.conditions)?;
                 validate_condition_config(&agg2.conditions)?;
@@ -833,7 +833,20 @@ impl Alerts {
         let store = PARSEABLE.storage.get_object_store();
 
         for alert in store.get_alerts().await.unwrap_or_default() {
-            self.sender.send(AlertTask::Create(alert.clone())).await?;
+            match self.sender.send(AlertTask::Create(alert.clone())).await {
+                Ok(_) => {}
+                Err(e) => {
+                    warn!("Failed to create alert task: {e}\nRetrying...");
+                    // Retry sending the task
+                    match self.sender.send(AlertTask::Create(alert.clone())).await {
+                        Ok(_) => {}
+                        Err(e) => {
+                            error!("Failed to create alert task: {e}");
+                            continue;
+                        }
+                    }
+                }
+            };
             map.insert(alert.id, alert);
         }
 

src/alerts/target.rs

@@ -29,6 +29,7 @@ use http::{header::AUTHORIZATION, HeaderMap, HeaderValue};
 use humantime_serde::re::humantime;
 use reqwest::ClientBuilder;
 use tracing::{error, trace, warn};
+use url::Url;
 
 use super::ALERTS;
 
@@ -255,9 +256,9 @@ fn default_client_builder() -> ClientBuilder {
     ClientBuilder::new()
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct SlackWebHook {
-    endpoint: String,
+    endpoint: Url,
 }
 
 #[async_trait]
@@ -279,7 +280,7 @@ impl CallableTarget for SlackWebHook {
             }
         };
 
-        if let Err(e) = client.post(&self.endpoint).json(&alert).send().await {
+        if let Err(e) = client.post(self.endpoint.clone()).json(&alert).send().await {
             error!("Couldn't make call to webhook, error: {}", e)
         }
     }
@@ -288,7 +289,7 @@ impl CallableTarget for SlackWebHook {
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct OtherWebHook {
-    endpoint: String,
+    endpoint: Url,
     #[serde(default)]
     headers: HashMap<String, String>,
     #[serde(default)]
@@ -314,7 +315,7 @@ impl CallableTarget for OtherWebHook {
         };
 
         let request = client
-            .post(&self.endpoint)
+            .post(self.endpoint.clone())
             .headers((&self.headers).try_into().expect("valid_headers"));
 
         if let Err(e) = request.body(alert).send().await {
@@ -326,7 +327,7 @@ impl CallableTarget for OtherWebHook {
 #[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct AlertManager {
-    endpoint: String,
+    endpoint: Url,
     #[serde(default)]
     skip_tls_check: bool,
     #[serde(flatten)]
@@ -404,7 +405,12 @@ impl CallableTarget for AlertManager {
             }
         };
 
-        if let Err(e) = client.post(&self.endpoint).json(&alerts).send().await {
+        if let Err(e) = client
+            .post(self.endpoint.clone())
+            .json(&alerts)
+            .send()
+            .await
+        {
             error!("Couldn't make call to alertmanager, error: {}", e)
         }
     }

src/handlers/http/modal/mod.rs

@@ -34,6 +34,7 @@ use tokio::sync::oneshot;
 use tracing::{error, info, warn};
 
 use crate::{
+    alerts::ALERTS,
     cli::Options,
     correlation::CORRELATIONS,
     oidc::Claims,
@@ -165,7 +166,7 @@ pub trait ParseableServer {
 
 pub async fn load_on_init() -> anyhow::Result<()> {
     // Run all loading operations concurrently
-    let (correlations_result, filters_result, dashboards_result) = future::join3(
+    let (correlations_result, filters_result, dashboards_result, alerts_result) = future::join4(
         async {
             CORRELATIONS
                 .load()
@@ -174,6 +175,7 @@ pub async fn load_on_init() -> anyhow::Result<()> {
         },
         async { FILTERS.load().await.context("Failed to load filters") },
         async { DASHBOARDS.load().await.context("Failed to load dashboards") },
+        async { ALERTS.load().await.context("Failed to load alerts") },
     )
     .await;
 
@@ -190,6 +192,10 @@ pub async fn load_on_init() -> anyhow::Result<()> {
         error!("{err}");
     }
 
+    if let Err(err) = alerts_result {
+        error!("{err}");
+    }
+
     Ok(())
 }
 

src/handlers/http/modal/query_server.rs

@@ -18,7 +18,6 @@
 
 use std::thread;
 
-use crate::alerts::ALERTS;
 use crate::handlers::airplane;
 use crate::handlers::http::cluster::{self, init_cluster_metrics_schedular};
 use crate::handlers::http::middleware::{DisAllowRootUser, RouteExt};
@@ -34,7 +33,7 @@ use actix_web_prometheus::PrometheusMetrics;
 use async_trait::async_trait;
 use bytes::Bytes;
 use tokio::sync::oneshot;
-use tracing::{error, info};
+use tracing::info;
 
 use crate::parseable::PARSEABLE;
 use crate::Server;
@@ -128,17 +127,6 @@ impl ParseableServer for QueryServer {
         let (cancel_tx, cancel_rx) = oneshot::channel();
         thread::spawn(|| sync::handler(cancel_rx));
 
-        // Run the alerts scheduler
-        tokio::spawn(async {
-            match ALERTS.load().await {
-                Ok(_) => info!("Alerts loaded successfully"),
-                Err(e) => {
-                    error!("Failed to load alerts: {}", e);
-                    // return Err(e);
-                }
-            };
-        });
-
         tokio::spawn(airplane::server());
 
         let result = self

src/sync.rs

@@ -246,16 +246,25 @@ pub async fn alert_runtime(mut rx: mpsc::Receiver<AlertTask>) -> Result<(), anyh
                 let alert = alert.clone();
                 let id = alert.id;
                 let handle = tokio::spawn(async move {
+                    let mut retry_counter = 0;
+                    let mut sleep_duration = alert.get_eval_frequency();
                     loop {
                         match alerts_utils::evaluate_alert(&alert).await {
-                            Ok(_) => {}
+                            Ok(_) => {
+                                retry_counter = 0;
+                            }
                             Err(err) => {
-                                error!("Error while evaluation- {err}");
-                                break;
+                                warn!("Error while evaluation- {}\nRetrying after sleeping for 1 minute", err);
+                                sleep_duration = 1;
+                                retry_counter += 1;
+
+                                if retry_counter > 3 {
+                                    error!("Alert with id {} failed to evaluate after 3 retries with err- {}", id, err);
+                                    break;
+                                }
                             }
                         }
-                        tokio::time::sleep(Duration::from_secs(alert.get_eval_frequency() * 60))
-                            .await;
+                        tokio::time::sleep(Duration::from_secs(sleep_duration * 60)).await;
                     }
                 });