feat(alerts): foundational implementation

server/Cargo.toml

@@ -49,6 +49,7 @@ clokwerk = "0.4.0-rc1"
 actix-web-static-files = "4.0"
 static-files = "0.2.1"
 walkdir = "2"
+ureq = "2.5.0"
 
 [build-dependencies]
 static-files = "0.2.1"

server/src/alerts.rs

@@ -0,0 +1,134 @@
+/*
+ * Parseable Server (C) 2022 Parseable, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+use log::{error, info};
+use serde::{Deserialize, Serialize};
+
+use crate::error::Error;
+
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Alerts {
+    pub alerts: Vec<Alert>,
+}
+
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Alert {
+    pub name: String,
+    pub message: String,
+    pub rule: Rule,
+    pub targets: Vec<Target>,
+}
+
+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<(), Error> {
+        if self.rule.resolves(event).await {
+            info!("Alert triggered; name: {}", self.name);
+            for target in self.targets.clone() {
+                let msg = self.message.clone();
+                actix_web::rt::spawn(async move {
+                    target.call(&msg);
+                });
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Default, Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Rule {
+    pub field: String,
+    /// Field that determines what comparison operator is to be used
+    #[serde(default)]
+    pub operator: Operator,
+    pub value: serde_json::Value,
+    pub repeats: u32,
+    #[serde(skip)]
+    repeated: u32,
+    pub within: String,
+}
+
+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 {
+        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()
+            }
+            Operator::LessThan => {
+                event.get(&self.field).unwrap().as_f64().unwrap() < (self.value).as_f64().unwrap()
+            }
+        };
+
+        // If truthy, increment count of repeated
+        if comparison {
+            self.repeated += 1;
+        }
+
+        // If enough repetitions made, return true
+        if self.repeated >= self.repeats {
+            self.repeated = 0;
+            return true;
+        }
+
+        false
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Operator {
+    EqualTo,
+    GreaterThan,
+    LessThan,
+}
+
+impl Default for Operator {
+    fn default() -> Self {
+        Self::EqualTo
+    }
+}
+
+#[derive(Default, 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,
+}
+
+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)
+        }
+    }
+}

server/src/event.rs

@@ -35,6 +35,7 @@ use crate::response;
 use crate::storage::ObjectStorage;
 use crate::Error;
 
+#[derive(Clone)]
 pub struct Event {
     pub body: String,
     pub stream_name: String,
@@ -91,6 +92,10 @@ impl Event {
             error!("Couldn't update stream stats. {:?}", e);
         }
 
+        if let Err(e) = metadata::STREAM_INFO.check_alerts(self).await {
+            error!("Error checking for alerts. {:?}", e);
+        }
+
         let msg = if is_first_event {
             format!(
                 "Intial Event recieved for log stream {}, schema uploaded successfully",

server/src/handlers/logstream.rs

@@ -19,11 +19,11 @@
 use actix_web::http::StatusCode;
 use actix_web::{web, HttpRequest, HttpResponse, Responder};
 
-use crate::metadata;
+use crate::alerts::Alerts;
 use crate::response;
 use crate::s3::S3;
 use crate::storage::ObjectStorage;
-use crate::validator;
+use crate::{metadata, validator};
 
 pub async fn delete(req: HttpRequest) -> HttpResponse {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
@@ -116,26 +116,23 @@ pub async fn schema(req: HttpRequest) -> HttpResponse {
 pub async fn get_alert(req: HttpRequest) -> HttpResponse {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
 
-    match metadata::STREAM_INFO.alert(&stream_name) {
-        Ok(alert) => response::ServerResponse {
-            msg: alert,
+    match metadata::STREAM_INFO.alert(stream_name.clone()) {
+        Ok(alerts) => response::ServerResponse {
+            msg: serde_json::to_string(&alerts).unwrap(),
             code: StatusCode::OK,
         }
         .to_http(),
-        Err(_) => match S3::new().get_alert(&stream_name).await {
-            Ok(alert) if alert.is_empty() => response::ServerResponse {
+        Err(_) => match S3::new().get_alerts(&stream_name).await {
+            Ok(alerts) if alerts.alerts.is_empty() => response::ServerResponse {
                 msg: "alert configuration not set for log stream {}".to_string(),
                 code: StatusCode::BAD_REQUEST,
             }
             .to_http(),
-            Ok(alert) => {
-                let buf = alert.as_ref();
-                response::ServerResponse {
-                    msg: String::from_utf8(buf.to_vec()).unwrap(),
-                    code: StatusCode::OK,
-                }
-                .to_http()
+            Ok(alerts) => response::ServerResponse {
+                msg: serde_json::to_string(&alerts).unwrap(),
+                code: StatusCode::OK,
             }
+            .to_http(),
             Err(_) => response::ServerResponse {
                 msg: "alert doesn't exist".to_string(),
                 code: StatusCode::BAD_REQUEST,
@@ -164,7 +161,7 @@ pub async fn put(req: HttpRequest) -> HttpResponse {
         if let Err(e) = metadata::STREAM_INFO.add_stream(
             stream_name.to_string(),
             "".to_string(),
-            "".to_string(),
+            Default::default(),
         ) {
             return response::ServerResponse {
                 msg: format!(
@@ -208,47 +205,56 @@ pub async fn put(req: HttpRequest) -> HttpResponse {
 
 pub async fn put_alert(req: HttpRequest, body: web::Json<serde_json::Value>) -> HttpResponse {
     let stream_name: String = req.match_info().get("logstream").unwrap().parse().unwrap();
-    let alert_config = body.clone();
-    match validator::alert(serde_json::to_string(&body.as_object()).unwrap()) {
-        Ok(_) => match S3::new()
-            .create_alert(&stream_name, alert_config.to_string())
-            .await
-        {
-            Ok(_) => {
-                if let Err(e) =
-                    metadata::STREAM_INFO.set_alert(stream_name.clone(), alert_config.to_string())
-                {
-                    return response::ServerResponse {
-                        msg: format!(
-                            "failed to set alert configuration for log stream {} due to err: {}",
-                            stream_name, e
-                        ),
-                        code: StatusCode::INTERNAL_SERVER_ERROR,
-                    }
-                    .to_http();
-                }
-                response::ServerResponse {
-                    msg: format!("set alert configuration for log stream {}", stream_name),
-                    code: StatusCode::OK,
-                }
-                .to_http()
-            }
-            Err(e) => response::ServerResponse {
+    let alerts: Alerts = match serde_json::from_value(body.into_inner()) {
+        Ok(alerts) => alerts,
+        Err(e) => {
+            return response::ServerResponse {
                 msg: format!(
                     "failed to set alert configuration for log stream {} due to err: {}",
                     stream_name, e
                 ),
-                code: StatusCode::INTERNAL_SERVER_ERROR,
+                code: StatusCode::BAD_REQUEST,
             }
-            .to_http(),
-        },
-        Err(e) => response::ServerResponse {
+            .to_http()
+        }
+    };
+
+    if let Err(e) = validator::alert(serde_json::to_string(&alerts).unwrap()) {
+        return response::ServerResponse {
             msg: format!(
                 "failed to set alert configuration for log stream {} due to err: {}",
                 stream_name, e
             ),
             code: StatusCode::BAD_REQUEST,
         }
-        .to_http(),
+        .to_http();
+    }
+
+    if let Err(e) = S3::new().put_alerts(&stream_name, alerts.clone()).await {
+        return response::ServerResponse {
+            msg: format!(
+                "failed to set alert configuration for log stream {} due to err: {}",
+                stream_name, e
+            ),
+            code: StatusCode::INTERNAL_SERVER_ERROR,
+        }
+        .to_http();
+    }
+
+    if let Err(e) = metadata::STREAM_INFO.set_alert(stream_name.to_string(), alerts) {
+        return response::ServerResponse {
+            msg: format!(
+                "failed to set alert configuration for log stream {} due to err: {}",
+                stream_name, e
+            ),
+            code: StatusCode::INTERNAL_SERVER_ERROR,
+        }
+        .to_http();
+    }
+
+    response::ServerResponse {
+        msg: format!("set alert configuration for log stream {}", stream_name),
+        code: StatusCode::OK,
     }
+    .to_http()
 }

server/src/main.rs

@@ -34,6 +34,7 @@ use std::time::Duration;
 use tokio::sync::oneshot;
 use tokio::sync::oneshot::error::TryRecvError;
 
+mod alerts;
 mod banner;
 mod error;
 mod event;