Merge branch 'main' into alerts-updates

Author: nikhilsinhaparseable

src/alerts/alert_enums.rs

@@ -235,6 +235,7 @@ pub enum EvalConfig {
 pub enum AlertState {
     Triggered,
     #[default]
+    #[serde(rename = "not-triggered")]
     NotTriggered,
     Disabled,
 }
@@ -254,7 +255,7 @@ impl Display for AlertState {
 pub enum NotificationState {
     #[default]
     Notify,
-    /// Snoozed means the alert will evaluate but no notifications will be sent out
+    /// Mute means the alert will evaluate but no notifications will be sent out
     ///
     /// It is a state which can only be set manually
     ///

src/alerts/alert_structs.rs

@@ -16,7 +16,7 @@
  *
  */
 
-use std::collections::HashMap;
+use std::{collections::HashMap, time::Duration};
 
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
@@ -431,12 +431,56 @@ pub struct ForecastConfig {
     pub forecast_duration: String,
 }
 
+impl ForecastConfig {
+    pub fn calculate_eval_window(&self) -> Result<String, AlertError> {
+        let parsed_historic_duration =
+            if let Ok(historic_duration) = humantime::parse_duration(&self.historic_duration) {
+                historic_duration
+            } else {
+                return Err(AlertError::Metadata(
+                    "historicDuration should be of type humantime",
+                ));
+            };
+
+        let eval_window = if parsed_historic_duration.lt(&Duration::from_secs(60 * 60 * 24 * 3)) {
+            // less than 3 days = 10 mins
+            "10m"
+        } else {
+            "30m"
+        };
+
+        Ok(eval_window.into())
+    }
+}
+
 #[derive(Debug, serde::Serialize, serde::Deserialize, Clone, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct AnomalyConfig {
     pub historic_duration: String,
 }
 
+impl AnomalyConfig {
+    pub fn calculate_eval_window(&self) -> Result<String, AlertError> {
+        let parsed_historic_duration =
+            if let Ok(historic_duration) = humantime::parse_duration(&self.historic_duration) {
+                historic_duration
+            } else {
+                return Err(AlertError::Metadata(
+                    "historicDuration should be of type humantime",
+                ));
+            };
+
+        let eval_window = if parsed_historic_duration.lt(&Duration::from_secs(60 * 60 * 24 * 3)) {
+            // less than 3 days = 10 mins
+            "10m"
+        } else {
+            "30m"
+        };
+
+        Ok(eval_window.into())
+    }
+}
+
 /// Result structure for alert query execution with group support
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct AlertQueryResult {

src/alerts/alerts_utils.rs

@@ -275,10 +275,9 @@ async fn update_alert_state(
         } else {
             return Err(AlertError::CustomError("No AlertManager set".into()));
         }
-        // Lock is released here
     };
 
-    // Now perform the state update without holding the ALERTS lock
+    // Now perform the state update
     if let Some(msg) = message {
         alerts
             .update_state(*alert.get_id(), AlertState::Triggered, Some(msg))

src/alerts/target.rs

@@ -233,7 +233,6 @@ impl Target {
                     // call once and then start sleeping
                     // reduce repeats by 1
                     call_target(self.target.clone(), context.clone());
-                    // trace!("state not timed out- {state:?}");
                     // set state
                     state.timed_out = true;
                     state.awaiting_resolve = true;
@@ -539,17 +538,6 @@ impl CallableTarget for AlertManager {
 
         let alert = &mut alerts[0];
 
-        // alert["labels"].as_object_mut().expect("is object").extend(
-        //     payload
-        //         .additional_labels
-        //         .as_object()
-        //         .expect("is object")
-        //         .iter()
-        //         // filter non null values for alertmanager and only pass strings
-        //         .filter(|(_, value)| !value.is_null())
-        //         .map(|(k, value)| (k.to_owned(), json::convert_to_string(value))),
-        // );
-
         // fill in status label accordingly
         match payload.alert_info.alert_state {
             AlertState::Triggered => alert["labels"]["status"] = "triggered".into(),

src/handlers/http/alerts.rs

@@ -81,6 +81,12 @@ pub async fn post(
 ) -> Result<impl Responder, AlertError> {
     let mut alert: AlertConfig = alert.into().await?;
 
+    if alert.notification_config.interval > alert.get_eval_frequency() {
+        return Err(AlertError::ValidationFailure(
+            "Notification interval cannot exceed evaluation frequency".into(),
+        ));
+    }
+
     if alert.get_eval_frequency().eq(&0) {
         return Err(AlertError::ValidationFailure(
             "Eval frequency cannot be 0".into(),
@@ -386,6 +392,8 @@ pub async fn modify_alert(
         }
     };
 
+    new_alert.validate(&session_key).await?;
+
     // Perform I/O operations
     let path = alert_json_path(*new_alert.get_id());
     let store = PARSEABLE.storage.get_object_store();

src/static_schema.rs

@@ -124,7 +124,11 @@ pub fn convert_static_schema_to_arrow_schema(
                     "boolean_list" => {
                         DataType::List(Arc::new(Field::new("item", DataType::Boolean, true)))
                     }
-                    _ => DataType::Null,
+                    _ => {
+                        return Err(StaticSchemaError::UnrecognizedDataType(
+                            field.data_type.clone(),
+                        ));
+                    }
                 }
             },
             nullable: default_nullable(),
@@ -216,6 +220,9 @@ pub enum StaticSchemaError {
 
     #[error("duplicate field name: {0}")]
     DuplicateField(String),
+
+    #[error("unrecognized data type: {0}")]
+    UnrecognizedDataType(String),
 }
 
 #[cfg(test)]
@@ -233,4 +240,24 @@ mod tests {
         let _ = validate_field_names("test_field", &mut existing_field_names);
         assert!(validate_field_names("test_field", &mut existing_field_names).is_err());
     }
+
+    #[test]
+    fn unrecognized_data_type() {
+        let static_schema = StaticSchema {
+            fields: vec![SchemaFields {
+                name: "test_field".to_string(),
+                data_type: "unknown_type".to_string(),
+            }],
+        };
+
+        let result = convert_static_schema_to_arrow_schema(static_schema, "", None);
+
+        assert!(result.is_err());
+        match result.unwrap_err() {
+            StaticSchemaError::UnrecognizedDataType(data_type) => {
+                assert_eq!(data_type, "unknown_type");
+            }
+            _ => panic!("Expected UnrecognizedDataType error"),
+        }
+    }
 }

src/sync.rs

@@ -340,9 +340,12 @@ pub async fn alert_runtime(mut rx: mpsc::Receiver<AlertTask>) -> Result<(), anyh
                 if let Some(handle) = alert_tasks.remove(&ulid) {
                     // cancel the task
                     handle.abort();
-                    warn!("Alert with id {} deleted", ulid);
+                    warn!("Alert with id {} deleted from evaluation tasks list", ulid);
                 } else {
-                    error!("Alert with id {} does not exist", ulid);
+                    error!(
+                        "Alert with id {} does not exist in evaluation tasks list",
+                        ulid
+                    );
                 }
             }
         }