Refactor alerts (#179)

Alerts are stored on metadata map which is behind and rwlock.
This is fine but having to hold writer lock for alert check is not ideal.
This PR refactors code around alerts implementation.

Each alerts rule should be self contained with its own interior mutable state
if it requires any. Interior mutability pattern allows for alert to be resolved
with shared reference. An alert can now contain different variant of rules
which can be state or stateless.

Author: trueleo

server/Cargo.toml

@@ -53,6 +53,7 @@ actix-web-static-files = "4.0"
 static-files = "0.2.1"
 walkdir = "2"
 ureq = { version = "2.5.0", features = ["json"] }
+uuid = { version = "1.2.1", features = ["v4", "fast-rng", "serde"] }
 
 [build-dependencies]
 static-files = "0.2.1"

server/src/alerts.rs

@@ -16,18 +16,24 @@
  *
  */
 
-use log::{error, info};
+use std::sync::atomic::{AtomicU32, Ordering};
+
 use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+use crate::event::Event;
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Default, Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct Alerts {
     pub alerts: Vec<Alert>,
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct Alert {
+    #[serde(default = "crate::utils::uuid::gen")]
+    pub id: Uuid,
     pub name: String,
     pub message: String,
     pub rule: Rule,
@@ -37,13 +43,20 @@ pub struct Alert {
 impl Alert {
     // TODO: spawn async tasks to call webhooks if alert rules are met
     // This is done to ensure that threads aren't blocked by calls to the webhook
-    pub async fn check_alert(&mut self, event: &serde_json::Value) -> Result<(), ()> {
-        if self.rule.resolves(event).await {
-            info!("Alert triggered; name: {}", self.name);
+    pub async fn check_alert(&self, event: &Event) -> Result<(), ()> {
+        let event_json: serde_json::Value = serde_json::from_str(&event.body).map_err(|_| ())?;
+
+        if self.rule.resolves(&event_json) {
+            log::info!("Alert triggered for stream {}", self.name);
             for target in self.targets.clone() {
-                let msg = self.message.clone();
+                let context = Context::new(
+                    event.stream_name.clone(),
+                    self.name.clone(),
+                    self.message.clone(),
+                    self.rule.trigger_reason(),
+                );
                 actix_web::rt::spawn(async move {
-                    target.call(&msg);
+                    target.call(&context);
                 });
             }
         }
@@ -52,81 +65,270 @@ impl Alert {
     }
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum Rule {
+    Numeric(NumericRule),
+}
+
+impl Rule {
+    fn resolves(&self, event: &serde_json::Value) -> bool {
+        match self {
+            Rule::Numeric(rule) => rule.resolves(event),
+        }
+    }
+
+    pub fn valid_for_schema(&self, schema: &arrow_schema::Schema) -> bool {
+        match self {
+            Rule::Numeric(NumericRule { column, .. }) => match schema.column_with_name(column) {
+                Some((_, column)) => matches!(
+                    column.data_type(),
+                    arrow_schema::DataType::Int8
+                        | arrow_schema::DataType::Int16
+                        | arrow_schema::DataType::Int32
+                        | arrow_schema::DataType::Int64
+                        | arrow_schema::DataType::UInt8
+                        | arrow_schema::DataType::UInt16
+                        | arrow_schema::DataType::UInt32
+                        | arrow_schema::DataType::UInt64
+                        | arrow_schema::DataType::Float16
+                        | arrow_schema::DataType::Float32
+                        | arrow_schema::DataType::Float64
+                ),
+                None => false,
+            },
+        }
+    }
+
+    pub fn trigger_reason(&self) -> String {
+        match self {
+            Rule::Numeric(NumericRule {
+                column,
+                operator,
+                value,
+                repeats,
+                ..
+            }) => match operator {
+                NumericOperator::EqualTo => format!(
+                    "{} column was equal to {}, {} times",
+                    column, value, repeats
+                ),
+                NumericOperator::NotEqualTo => format!(
+                    "{} column was not equal to {}, {} times",
+                    column, value, repeats
+                ),
+                NumericOperator::GreaterThan => format!(
+                    "{} column was greater than {}, {} times",
+                    column, value, repeats
+                ),
+                NumericOperator::GreaterThanEquals => format!(
+                    "{} column was greater than or equal to {}, {} times",
+                    column, value, repeats
+                ),
+                NumericOperator::LessThan => format!(
+                    "{} column was less than {}, {} times",
+                    column, value, repeats
+                ),
+                NumericOperator::LessThanEquals => format!(
+                    "{} column was less than or equal to {}, {} times",
+                    column, value, repeats
+                ),
+            },
+        }
+    }
+}
+
+#[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct Rule {
-    pub field: String,
+pub struct NumericRule {
+    pub column: String,
     /// Field that determines what comparison operator is to be used
     #[serde(default)]
-    pub operator: Operator,
-    pub value: serde_json::Value,
+    pub operator: NumericOperator,
+    pub value: serde_json::Number,
     pub repeats: u32,
     #[serde(skip)]
-    repeated: u32,
-    pub within: String,
+    repeated: AtomicU32,
 }
 
-impl Rule {
-    // TODO: utilise `within` to set a range for validity of rule to trigger alert
-    pub async fn resolves(&mut self, event: &serde_json::Value) -> bool {
+impl NumericRule {
+    fn resolves(&self, event: &serde_json::Value) -> bool {
+        let number = match event.get(&self.column).expect("column exists") {
+            serde_json::Value::Number(number) => number,
+            _ => unreachable!("right rule is set for right column type"),
+        };
+
         let comparison = match self.operator {
-            Operator::EqualTo => event.get(&self.field).unwrap() == &self.value,
-            // TODO: currently this is a hack, ensure checks are performed in the right way
-            Operator::GreaterThan => {
-                event.get(&self.field).unwrap().as_f64().unwrap() > (self.value).as_f64().unwrap()
+            NumericOperator::EqualTo => number == &self.value,
+            NumericOperator::NotEqualTo => number != &self.value,
+            NumericOperator::GreaterThan => number.as_f64().unwrap() > self.value.as_f64().unwrap(),
+            NumericOperator::GreaterThanEquals => {
+                number.as_f64().unwrap() >= self.value.as_f64().unwrap()
             }
-            Operator::LessThan => {
-                event.get(&self.field).unwrap().as_f64().unwrap() < (self.value).as_f64().unwrap()
+            NumericOperator::LessThan => number.as_f64().unwrap() < self.value.as_f64().unwrap(),
+            NumericOperator::LessThanEquals => {
+                number.as_f64().unwrap() <= self.value.as_f64().unwrap()
             }
         };
 
         // If truthy, increment count of repeated
+        // acquire lock and load
+        let mut repeated = self.repeated.load(Ordering::Acquire);
+
         if comparison {
-            self.repeated += 1;
+            repeated += 1
         }
 
         // If enough repetitions made, return true
-        if self.repeated >= self.repeats {
-            self.repeated = 0;
-            return true;
-        }
+        let ret = if repeated >= self.repeats {
+            repeated = 0;
+            true
+        } else {
+            false
+        };
+        // store the value back to repeated and release
+        self.repeated.store(repeated, Ordering::Release);
 
-        false
+        ret
     }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub enum Operator {
+pub enum NumericOperator {
+    #[serde(alias = "=")]
     EqualTo,
+    #[serde(alias = "!=")]
+    NotEqualTo,
+    #[serde(alias = ">")]
     GreaterThan,
+    #[serde(alias = ">=")]
+    GreaterThanEquals,
+    #[serde(alias = "<")]
     LessThan,
+    #[serde(alias = "<=")]
+    LessThanEquals,
 }
 
-impl Default for Operator {
+impl Default for NumericOperator {
     fn default() -> Self {
         Self::EqualTo
     }
 }
 
-#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
-pub struct Target {
-    pub name: String,
-    #[serde(rename = "server_url")]
-    pub server_url: String,
-    #[serde(rename = "api_key")]
-    pub api_key: String,
+#[serde(tag = "type")]
+pub enum Target {
+    Slack(targets::slack::SlackWebHook),
+    #[serde(alias = "webhook")]
+    Other(targets::other::OtherWebHook),
 }
 
 impl Target {
-    pub fn call(&self, msg: &str) {
-        if let Err(e) = ureq::post(&self.server_url)
-            .set("Content-Type", "text/plain; charset=iso-8859-1")
-            .set("X-API-Key", &self.api_key)
-            .send_string(msg)
-        {
-            error!("Couldn't make call to webhook, error: {}", e)
+    pub fn call(&self, payload: &Context) {
+        match self {
+            Target::Slack(target) => target.call(payload),
+            Target::Other(target) => target.call(payload),
+        }
+    }
+}
+
+pub trait CallableTarget {
+    fn call(&self, payload: &Context);
+}
+
+pub mod targets {
+    pub mod slack {
+        use serde::{Deserialize, Serialize};
+
+        use crate::alerts::{CallableTarget, Context};
+
+        #[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+        pub struct SlackWebHook {
+            #[serde(rename = "server_url")]
+            server_url: String,
+        }
+
+        impl CallableTarget for SlackWebHook {
+            fn call(&self, payload: &Context) {
+                if let Err(e) = ureq::post(&self.server_url)
+                    .set("Content-Type", "application/json")
+                    .send_json(ureq::json!({ "text": payload.default_alert_string() }))
+                {
+                    log::error!("Couldn't make call to webhook, error: {}", e)
+                }
+            }
+        }
+    }
+
+    pub mod other {
+        use serde::{Deserialize, Serialize};
+
+        use crate::alerts::{CallableTarget, Context};
+
+        #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+        #[serde(untagged)]
+        pub enum OtherWebHook {
+            ApiKey {
+                #[serde(rename = "server_url")]
+                server_url: String,
+                #[serde(rename = "api_key")]
+                api_key: String,
+            },
+            Simple {
+                #[serde(rename = "server_url")]
+                server_url: String,
+            },
         }
+
+        impl CallableTarget for OtherWebHook {
+            fn call(&self, payload: &Context) {
+                let res = match self {
+                    OtherWebHook::Simple { server_url } => ureq::post(server_url)
+                        .set("Content-Type", "text/plain; charset=iso-8859-1")
+                        .send_string(&payload.default_alert_string()),
+                    OtherWebHook::ApiKey {
+                        server_url,
+                        api_key,
+                    } => ureq::post(server_url)
+                        .set("Content-Type", "text/plain; charset=iso-8859-1")
+                        .set("X-API-Key", api_key)
+                        .send_string(&payload.default_alert_string()),
+                };
+
+                if let Err(e) = res {
+                    log::error!("Couldn't make call to webhook, error: {}", e)
+                }
+            }
+        }
+    }
+}
+
+pub struct Context {
+    stream: String,
+    alert_name: String,
+    message: String,
+    reason: String,
+}
+
+impl Context {
+    pub fn new(stream: String, alert_name: String, message: String, reason: String) -> Self {
+        Self {
+            stream,
+            alert_name,
+            message,
+            reason,
+        }
+    }
+
+    // <Alert_Name> Triggered on <Log_stream>
+    // Message: Ting
+    // Failing Condition: Status column was equal to 500, 5 times
+    fn default_alert_string(&self) -> String {
+        format!(
+            "{} triggered on {}\nMessage: {}\nFailing Condition: {}",
+            self.alert_name, self.stream, self.message, self.reason
+        )
     }
 }