Dynamic message in alert message (#348)

This PR adds a way to send log content (for a specific 
event) from a specific column to the alert target.

Users can specify the column name they want to send
as a part of alert message in the alert config. Like 

`message: "Alert triggered for status: {status_message}"`

Fixes #331

server/Cargo.toml

@@ -76,6 +76,7 @@ once_cell = "1.17.1"
 pyroscope = { version = "0.5.3", optional = true }
 pyroscope_pprofrs = { version = "0.2", optional = true }
 uptime_lib = "0.2.2"
+regex = "1.7.3"
 
 [build-dependencies]
 static-files = "0.2"

server/src/alerts/mod.rs

@@ -17,6 +17,8 @@
  */
 
 use async_trait::async_trait;
+use datafusion::arrow::datatypes::Schema;
+use regex::Regex;
 use serde::{Deserialize, Serialize};
 use std::fmt;
 
@@ -51,7 +53,8 @@ pub struct Alert {
     #[serde(default = "crate::utils::uid::gen")]
     pub id: uid::Uid,
     pub name: String,
-    pub message: String,
+    #[serde(flatten)]
+    pub message: Message,
     pub rule: Rule,
     pub targets: Vec<Target>,
 }
@@ -63,7 +66,7 @@ impl Alert {
         match resolves {
             AlertState::Listening | AlertState::Firing => (),
             alert_state @ (AlertState::SetToFiring | AlertState::Resolved) => {
-                let context = self.get_context(stream_name, alert_state, &self.rule);
+                let context = self.get_context(stream_name, alert_state, &self.rule, event_json);
                 ALERTS_STATES
                     .with_label_values(&[
                         context.stream.as_str(),
@@ -78,7 +81,13 @@ impl Alert {
         }
     }
 
-    fn get_context(&self, stream_name: String, alert_state: AlertState, rule: &Rule) -> Context {
+    fn get_context(
+        &self,
+        stream_name: String,
+        alert_state: AlertState,
+        rule: &Rule,
+        event_json: &serde_json::Value,
+    ) -> Context {
         let deployment_instance = format!(
             "{}://{}",
             CONFIG.parseable.get_scheme(),
@@ -102,7 +111,7 @@ impl Alert {
             stream_name,
             AlertInfo::new(
                 self.name.clone(),
-                self.message.clone(),
+                self.message.get(event_json),
                 rule.trigger_reason(),
                 alert_state,
             ),
@@ -111,6 +120,49 @@ impl Alert {
         )
     }
 }
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct Message {
+    pub message: String,
+}
+
+impl Message {
+    // checks if message (with a column name) is valid (i.e. the column name is present in the schema)
+    pub fn valid(&self, schema: &Schema, column: Option<&str>) -> bool {
+        if let Some(col) = column {
+            return schema.field_with_name(col).is_ok();
+        }
+        true
+    }
+
+    pub fn extract_column_name(&self) -> Option<&str> {
+        let re = Regex::new(r"\{(.*?)\}").unwrap();
+        let tokens: Vec<&str> = re
+            .captures_iter(self.message.as_str())
+            .map(|cap| cap.get(1).unwrap().as_str())
+            .collect();
+        // the message can have either no column name ({column_name} not present) or one column name
+        // return Some only if there is exactly one column name present
+        if tokens.len() == 1 {
+            return Some(tokens[0]);
+        }
+        None
+    }
+
+    // returns the message with the column name replaced with the value of the column
+    fn get(&self, event_json: &serde_json::Value) -> String {
+        if let Some(column) = self.extract_column_name() {
+            if let Some(value) = event_json.get(column) {
+                return self
+                    .message
+                    .replace(&format!("{{{column}}}"), value.to_string().as_str());
+            }
+        }
+        self.message.clone()
+    }
+}
+
 #[async_trait]
 pub trait CallableTarget {
     async fn call(&self, payload: &Context);

server/src/handlers/http/logstream.rs

@@ -153,6 +153,15 @@ pub async fn put_alert(
 
     let schema = STREAM_INFO.merged_schema(&stream_name)?;
     for alert in &alerts.alerts {
+        let column = alert.message.extract_column_name();
+        let is_valid = alert.message.valid(&schema, column);
+        if !is_valid {
+            let col = column.unwrap_or("");
+            return Err(StreamError::InvalidAlertMessage(
+                alert.name.to_owned(),
+                col.to_string(),
+            ));
+        }
         if !alert.rule.valid_for_schema(&schema) {
             return Err(StreamError::InvalidAlert(alert.name.to_owned()));
         }
@@ -301,6 +310,10 @@ pub mod error {
         AlertValidation(#[from] AlertValidationError),
         #[error("alert - \"{0}\" is invalid, please check if alert is valid according to this stream's schema and try again")]
         InvalidAlert(String),
+        #[error(
+            "alert - \"{0}\" is invalid, column \"{1}\" does not exist in this stream's schema"
+        )]
+        InvalidAlertMessage(String, String),
         #[error("failed to set retention configuration due to err: {0}")]
         InvalidRetentionConfig(serde_json::Error),
         #[error("{msg}")]
@@ -319,6 +332,7 @@ pub mod error {
                 StreamError::BadAlertJson { .. } => StatusCode::BAD_REQUEST,
                 StreamError::AlertValidation(_) => StatusCode::BAD_REQUEST,
                 StreamError::InvalidAlert(_) => StatusCode::BAD_REQUEST,
+                StreamError::InvalidAlertMessage(_, _) => StatusCode::BAD_REQUEST,
                 StreamError::InvalidRetentionConfig(_) => StatusCode::BAD_REQUEST,
             }
         }

server/src/validator.rs

@@ -37,7 +37,7 @@ pub fn alert(alerts: &Alerts) -> Result<(), AlertValidationError> {
         if alert.name.is_empty() {
             return Err(AlertValidationError::EmptyName);
         }
-        if alert.message.is_empty() {
+        if alert.message.message.is_empty() {
             return Err(AlertValidationError::EmptyMessage);
         }
         if alert.targets.is_empty() {