feat: add alert_url variable for alerts (#2482)

src/common/infra/cluster.rs

@@ -247,6 +247,14 @@ pub async fn leave() -> Result<()> {
     Ok(())
 }
 
+pub async fn update_node(uuid: &str, node: &Node) -> Result<()> {
+    let mut client = etcd::get_etcd_client().await.clone();
+    let key = format!("{}nodes/{}", &CONFIG.etcd.prefix, uuid);
+    let val = json::to_string(node).unwrap();
+    let _resp = client.put(key, val, None).await?;
+    Ok(())
+}
+
 pub fn get_cached_nodes(cond: fn(&Node) -> bool) -> Option<Vec<Node>> {
     if NODES.is_empty() {
         return None;

src/common/utils/base64.rs

@@ -48,6 +48,10 @@ pub(crate) fn decode_raw(s: &str) -> Result<Vec<u8>, Error> {
         .map_err(|e| Error::new(ErrorKind::InvalidData, format!("base64 decode error: {e}")))
 }
 
+pub(crate) fn encode(s: &str) -> String {
+    base64::engine::general_purpose::STANDARD.encode(s.as_bytes())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/config/src/config.rs

@@ -230,6 +230,10 @@ pub struct Common {
     pub cluster_name: String,
     #[env_config(name = "ZO_INSTANCE_NAME", default = "")]
     pub instance_name: String,
+    #[env_config(name = "ZO_WEB_URL", default = "")] // http://localhost:5080
+    pub web_url: String,
+    #[env_config(name = "ZO_BASE_URI", default = "")] // /abc
+    pub base_uri: String,
     #[env_config(name = "ZO_INGESTER_SIDECAR_ENABLED", default = false)]
     pub ingester_sidecar_enabled: bool,
     #[env_config(name = "ZO_INGESTER_SIDECAR_QUERIER", default = false)]
@@ -244,8 +248,6 @@ pub struct Common {
     pub data_db_dir: String,
     #[env_config(name = "ZO_DATA_CACHE_DIR", default = "")] // ./data/openobserve/cache/
     pub data_cache_dir: String,
-    #[env_config(name = "ZO_BASE_URI", default = "")]
-    pub base_uri: String,
     #[env_config(name = "ZO_WAL_MEMORY_MODE_ENABLED", default = false)]
     pub wal_memory_mode_enabled: bool,
     #[env_config(name = "ZO_WAL_LINE_MODE_ENABLED", default = true)]
@@ -756,6 +758,14 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
 }
 
 fn check_path_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
+    // for web
+    if cfg.common.web_url.ends_with('/') {
+        cfg.common.web_url = cfg.common.web_url.trim_end_matches('/').to_string();
+    }
+    if cfg.common.base_uri.ends_with('/') {
+        cfg.common.base_uri = cfg.common.base_uri.trim_end_matches('/').to_string();
+    }
+    // for data
     if cfg.common.data_dir.is_empty() {
         cfg.common.data_dir = "./data/openobserve/".to_string();
     }
@@ -786,9 +796,6 @@ fn check_path_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
     if !cfg.common.data_cache_dir.ends_with('/') {
         cfg.common.data_cache_dir = format!("{}/", cfg.common.data_cache_dir);
     }
-    if cfg.common.base_uri.ends_with('/') {
-        cfg.common.base_uri = cfg.common.base_uri.trim_end_matches('/').to_string();
-    }
     if cfg.sled.data_dir.is_empty() {
         cfg.sled.data_dir = format!("{}db/", cfg.common.data_dir);
     }

src/handler/http/request/status/mod.rs

@@ -15,7 +15,7 @@
 
 use std::io::Error;
 
-use actix_web::{get, HttpResponse};
+use actix_web::{get, put, web, HttpRequest, HttpResponse};
 use ahash::AHashMap as HashMap;
 use config::{CONFIG, HAS_FUNCTIONS, INSTANCE_ID, SQL_FULL_TEXT_SEARCH_FIELDS};
 use datafusion::arrow::datatypes::{Field, Schema};
@@ -36,7 +36,7 @@ use {
 use crate::{
     common::{
         infra::{cache, cluster, config::*, file_list},
-        meta::functions::ZoFunction,
+        meta::{functions::ZoFunction, http::HttpResponse as MetaHttpResponse},
         utils::json,
     },
     service::{db, search::datafusion::DEFAULT_FUNCTIONS},
@@ -264,3 +264,34 @@ async fn refresh_token_with_dex(req: actix_web::HttpRequest) -> HttpResponse {
         Err(_) => HttpResponse::Unauthorized().finish(),
     }
 }
+
+#[put("/node/enable")]
+async fn enable_node(req: HttpRequest) -> Result<HttpResponse, Error> {
+    let node_id = cluster::LOCAL_NODE_UUID.clone();
+    let Some(mut node) = cluster::get_node_by_uuid(&node_id) else {
+        return Ok(MetaHttpResponse::not_found("node not found"));
+    };
+
+    let query = web::Query::<HashMap<String, String>>::from_query(req.query_string()).unwrap();
+    let enable = match query.get("value") {
+        Some(v) => v.parse::<bool>().unwrap_or_default(),
+        None => false,
+    };
+    node.scheduled = enable;
+    match cluster::update_node(&node_id, &node).await {
+        Ok(_) => Ok(MetaHttpResponse::json(true)),
+        Err(e) => Ok(MetaHttpResponse::internal_error(e)),
+    }
+}
+
+#[put("/node/flush")]
+async fn flush_node() -> Result<HttpResponse, Error> {
+    if !cluster::is_ingester(&cluster::LOCAL_NODE_ROLE) {
+        return Ok(MetaHttpResponse::not_found("local node is not an ingester"));
+    };
+
+    match ingester::flush_all().await {
+        Ok(_) => Ok(MetaHttpResponse::json(true)),
+        Err(e) => Ok(MetaHttpResponse::internal_error(e)),
+    }
+}

src/handler/http/router/mod.rs

@@ -201,6 +201,8 @@ pub fn get_service_routes(cfg: &mut web::ServiceConfig) {
             ))
             .wrap(cors.clone())
             .service(status::cache_status)
+            .service(status::enable_node)
+            .service(status::flush_node)
             .service(users::list)
             .service(users::save)
             .service(users::delete)

src/ingester/src/lib.rs

@@ -25,7 +25,7 @@ mod writer;
 
 pub use entry::Entry;
 pub use immutable::read_from_immutable;
-pub use writer::{check_memtable_size, get_writer, read_from_memtable, Writer};
+pub use writer::{check_memtable_size, flush_all, get_writer, read_from_memtable, Writer};
 
 pub async fn init() -> errors::Result<()> {
     // check uncompleted parquet files, need delete those files

src/ingester/src/writer.rs

@@ -62,7 +62,9 @@ pub struct Writer {
 
 // check total memory size
 pub fn check_memtable_size() -> Result<()> {
-    let total_mem_size = metrics::INGEST_MEMTABLE_BYTES.with_label_values(&[]).get();
+    let total_mem_size = metrics::INGEST_MEMTABLE_ARROW_BYTES
+        .with_label_values(&[])
+        .get();
     if total_mem_size >= CONFIG.limit.mem_table_max_size as i64 {
         Err(Error::MemoryTableOverflowError {})
     } else {
@@ -105,6 +107,21 @@ pub async fn read_from_memtable(
     Ok(batches)
 }
 
+pub async fn flush_all() -> Result<()> {
+    for w in WRITERS.iter() {
+        let mut w = w.write().await;
+        let keys = w.keys().cloned().collect::<Vec<_>>();
+        for r in w.values() {
+            r.close().await?; // close writer
+            metrics::INGEST_MEMTABLE_FILES.with_label_values(&[]).dec();
+        }
+        for key in keys {
+            w.remove(&key);
+        }
+    }
+    Ok(())
+}
+
 impl Writer {
     pub(crate) fn new(thread_id: usize, key: WriterKey) -> Self {
         let now = Utc::now().timestamp_micros();
@@ -195,6 +212,28 @@ impl Writer {
         Ok(())
     }
 
+    pub async fn close(&self) -> Result<()> {
+        // rotation wal
+        let wal = self.wal.lock().await;
+        wal.sync().context(WalSnafu)?;
+        let path = wal.path().clone();
+        drop(wal);
+
+        // rotation memtable
+        let mut mem = self.memtable.write().await;
+        let new_mem = MemTable::new();
+        let old_mem = std::mem::replace(&mut *mem, new_mem);
+        drop(mem);
+
+        let thread_id = self.thread_id;
+        let key = self.key.clone();
+        IMMUTABLES
+            .write()
+            .await
+            .insert(path, immutable::Immutable::new(thread_id, key, old_mem));
+        Ok(())
+    }
+
     pub async fn sync(&self) -> Result<()> {
         let wal = self.wal.lock().await;
         wal.sync().context(WalSnafu)

src/service/alerts/alert_manager.rs

@@ -142,8 +142,8 @@ pub async fn handle_triggers(
     }
 
     // send notification
-    if let Some(ret) = ret {
-        alert.send_notification(&ret).await?;
+    if let Some(data) = ret {
+        alert.send_notification(&data).await?;
     }
 
     // update trigger

src/service/alerts/mod.rs

@@ -33,6 +33,7 @@ use crate::{
         },
         utils::{
             auth::{remove_ownership, set_ownership},
+            base64,
             json::{self, Map, Value},
         },
     },
@@ -331,7 +332,7 @@ impl QueryCondition {
                             &Operator::EqualTo => "==".to_string(),
                             _ => condition.operator.to_string(),
                         },
-                        condition.value.as_f64().unwrap_or_default()
+                        to_float(&condition.value)
                     ),
                     start,
                     end,
@@ -735,7 +736,7 @@ pub async fn send_notification(
         process_row_template(&alert.row_template, alert, rows)
     };
     let resp = json::to_string(&dest.template.body)?;
-    let resp = process_dest_template(&resp, alert, rows, &rows_tpl_val);
+    let resp = process_dest_template(&resp, alert, rows, &rows_tpl_val).await;
     let msg: Value = json::from_str(&resp)?;
     let msg: Value = match &msg {
         Value::String(obj) => match json::from_str(obj) {
@@ -791,6 +792,8 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
         for (key, value) in row.iter() {
             let value = if value.is_string() {
                 value.as_str().unwrap_or_default().to_string()
+            } else if value.is_f64() {
+                format!("{:.2}", value.as_f64().unwrap_or_default())
             } else {
                 value.to_string()
             };
@@ -821,7 +824,7 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
                 }
             }
         }
-        let alert_start_time = if alert_start_time > 0 {
+        let alert_start_time_str = if alert_start_time > 0 {
             Local
                 .timestamp_nanos(alert_start_time * 1000)
                 .format("%Y-%m-%dT%H:%M:%S")
@@ -829,7 +832,7 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
         } else {
             String::from("N/A")
         };
-        let alert_end_time = if alert_end_time > 0 {
+        let alert_end_time_str = if alert_end_time > 0 {
             Local
                 .timestamp_nanos(alert_end_time * 1000)
                 .format("%Y-%m-%dT%H:%M:%S")
@@ -837,6 +840,7 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
         } else {
             String::from("N/A")
         };
+
         resp = resp
             .replace("{org_name}", &alert.org_id)
             .replace("{stream_type}", &alert.stream_type.to_string())
@@ -856,8 +860,8 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
                 &alert.trigger_condition.threshold.to_string(),
             )
             .replace("{alert_count}", &alert_count.to_string())
-            .replace("{alert_start_time}", &alert_start_time)
-            .replace("{alert_end_time}", &alert_end_time);
+            .replace("{alert_start_time}", &alert_start_time_str)
+            .replace("{alert_end_time}", &alert_end_time_str);
 
         if let Some(attrs) = &alert.context_attributes {
             for (key, value) in attrs.iter() {
@@ -871,7 +875,7 @@ fn process_row_template(tpl: &String, alert: &Alert, rows: &[Map<String, Value>]
     rows_tpl.join("\\\\n")
 }
 
-fn process_dest_template(
+async fn process_dest_template(
     tpl: &str,
     alert: &Alert,
     rows: &[Map<String, Value>],
@@ -884,6 +888,8 @@ fn process_dest_template(
         for (key, value) in row.iter() {
             let value = if value.is_string() {
                 value.as_str().unwrap_or_default().to_string()
+            } else if value.is_f64() {
+                format!("{:.2}", value.as_f64().unwrap_or_default())
             } else {
                 value.to_string()
             };
@@ -922,7 +928,7 @@ fn process_dest_template(
             }
         }
     }
-    let alert_start_time = if alert_start_time > 0 {
+    let alert_start_time_str = if alert_start_time > 0 {
         Local
             .timestamp_nanos(alert_start_time * 1000)
             .format("%Y-%m-%dT%H:%M:%S")
@@ -930,7 +936,7 @@ fn process_dest_template(
     } else {
         String::from("N/A")
     };
-    let alert_end_time = if alert_end_time > 0 {
+    let alert_end_time_str = if alert_end_time > 0 {
         Local
             .timestamp_nanos(alert_end_time * 1000)
             .format("%Y-%m-%dT%H:%M:%S")
@@ -945,6 +951,60 @@ fn process_dest_template(
         "scheduled"
     };
 
+    // Hack time range for alert url
+    if alert_start_time == alert_end_time {
+        alert_start_time = alert_end_time
+            - Duration::minutes(alert.trigger_condition.period)
+                .num_microseconds()
+                .unwrap();
+    }
+
+    let mut alert_query = String::new();
+    let alert_url = if alert.query_condition.query_type == QueryType::PromQL {
+        if let Some(promql) = &alert.query_condition.promql {
+            let condition = alert.query_condition.promql_condition.as_ref().unwrap();
+            alert_query = format!(
+                "({}) {} {}",
+                promql,
+                match condition.operator {
+                    Operator::EqualTo => "==".to_string(),
+                    _ => condition.operator.to_string(),
+                },
+                to_float(&condition.value)
+            );
+        }
+        // http://localhost:5080/web/metrics?stream=zo_http_response_time_bucket&from=1705248000000000&to=1705334340000000&query=em9faHR0cF9yZXNwb25zZV90aW1lX2J1Y2tldHt9&org_identifier=default
+        format!(
+            "{}{}/web/metrics?org_identifier={}&stream_type={}&stream={}&from={}&to={}&query={}",
+            CONFIG.common.web_url,
+            CONFIG.common.base_uri,
+            alert.org_id,
+            alert.stream_type,
+            alert.stream_name,
+            alert_start_time,
+            alert_end_time,
+            base64::encode(&alert_query),
+        )
+    } else {
+        if let Some(conditions) = &alert.query_condition.conditions {
+            if let Ok(v) = build_sql(alert, conditions).await {
+                alert_query = v;
+            }
+        }
+        // http://localhost:5080/web/logs?stream_type=logs&stream=default&from=1705248000000000&to=1705334340000000&sql_mode=true&query=U0VMRUNUICogRlJPTSAiZGVmYXVsdCIg&org_identifier=default
+        format!(
+            "{}{}/web/logs?org_identifier={}&stream_type={}&stream={}&from={}&to={}&sql_mode=true&query={}",
+            CONFIG.common.web_url,
+            CONFIG.common.base_uri,
+            alert.org_id,
+            alert.stream_type,
+            alert.stream_name,
+            alert_start_time,
+            alert_end_time,
+            base64::encode(&alert_query),
+        )
+    };
+
     let mut resp = tpl
         .replace("{org_name}", &alert.org_id)
         .replace("{stream_type}", &alert.stream_type.to_string())
@@ -964,8 +1024,9 @@ fn process_dest_template(
             &alert.trigger_condition.threshold.to_string(),
         )
         .replace("{alert_count}", &alert_count.to_string())
-        .replace("{alert_start_time}", &alert_start_time)
-        .replace("{alert_end_time}", &alert_end_time)
+        .replace("{alert_start_time}", &alert_start_time_str)
+        .replace("{alert_end_time}", &alert_end_time_str)
+        .replace("{alert_url}", &alert_url)
         .replace("{rows}", rows_tpl_val);
     for (key, value) in vars.iter() {
         if resp.contains(&format!("{{{key}}}")) {
@@ -990,3 +1051,11 @@ fn format_variable_value(val: &str) -> String {
         .replace('\r', "\\\\r")
         .replace('\"', "\\\\\\\"")
 }
+
+fn to_float(val: &Value) -> f64 {
+    if val.is_number() {
+        val.as_f64().unwrap_or_default()
+    } else {
+        val.as_str().unwrap_or_default().parse().unwrap_or_default()
+    }
+}

src/service/ingestion/mod.rs

@@ -50,7 +50,7 @@ use crate::{
 
 pub mod grpc;
 
-pub type TriggerAlertData = Option<Vec<(Alert, Vec<Map<String, Value>>)>>;
+pub type TriggerAlertData = Vec<(Alert, Vec<Map<String, Value>>)>;
 
 pub fn compile_vrl_function(func: &str, org_id: &str) -> Result<VRLRuntimeConfig, std::io::Error> {
     if func.contains("get_env_var") {
@@ -176,7 +176,7 @@ pub async fn get_stream_alerts(
     stream_alerts_map.insert(key, alerts);
 }
 
-pub async fn evaluate_trigger(trigger: TriggerAlertData) {
+pub async fn evaluate_trigger(trigger: Option<TriggerAlertData>) {
     if trigger.is_none() {
         return;
     }

src/service/logs/bulk.rs

@@ -95,7 +95,7 @@ pub async fn ingest(
     let mut action = String::from("");
     let mut stream_name = String::from("");
     let mut doc_id = String::from("");
-    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
 
     let mut next_line_is_data = false;
     let reader = BufReader::new(body.as_ref());

src/service/logs/ingest.rs

@@ -98,7 +98,7 @@ pub async fn ingest(
 
     let mut stream_status = StreamStatus::new(stream_name);
     let mut distinct_values = Vec::with_capacity(16);
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let partition_det =
         crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map).await;

src/service/logs/mod.rs

@@ -151,8 +151,8 @@ async fn add_valid_record(
     write_buf: &mut HashMap<String, SchemaRecords>,
     record_val: &mut Map<String, Value>,
     need_trigger: bool,
-) -> Result<TriggerAlertData, anyhow::Error> {
-    let mut trigger: Vec<(Alert, Vec<Map<String, Value>>)> = Vec::new();
+) -> Result<Option<TriggerAlertData>, anyhow::Error> {
+    let mut trigger: TriggerAlertData = Vec::new();
     let timestamp: i64 = record_val
         .get(&CONFIG.common.column_timestamp)
         .unwrap()

src/service/logs/multi.rs

@@ -84,7 +84,7 @@ async fn ingest_inner(
 
     let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
     let mut stream_status = StreamStatus::new(stream_name);
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     // Start Register Transforms for stream
     let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms(

src/service/logs/otlp_grpc.rs

@@ -84,7 +84,7 @@ pub async fn usage_ingest(
     let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
     let mut stream_status = StreamStatus::new(stream_name);
 
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let partition_det =
         crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map).await;
@@ -301,7 +301,7 @@ pub async fn handle_grpc_request(
     );
     // End Register Transforms for stream
 
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let mut data_buf: HashMap<String, SchemaRecords> = HashMap::new();
 

src/service/logs/otlp_http.rs

@@ -134,7 +134,7 @@ pub async fn logs_json_handler(
     let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
     let mut distinct_values = Vec::with_capacity(16);
     let mut stream_status = StreamStatus::new(stream_name);
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let min_ts =
         (Utc::now() - Duration::hours(CONFIG.limit.ingest_allowed_upto)).timestamp_micros();

src/service/logs/syslog.rs

@@ -89,7 +89,7 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result<HttpResponse, anyhow:
     let mut stream_status = StreamStatus::new(stream_name);
     let mut distinct_values = Vec::with_capacity(16);
 
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let partition_det =
         crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map).await;

src/service/metrics/otlp_grpc.rs

@@ -31,7 +31,7 @@ use crate::{
     common::{
         infra::cluster,
         meta::{
-            alerts::{self, Alert},
+            alerts,
             http::HttpResponse as MetaHttpResponse,
             prom::*,
             stream::{PartitioningDetails, SchemaRecords},
@@ -91,7 +91,7 @@ pub async fn handle_grpc_request(
     let mut metric_schema_map: HashMap<String, Schema> = HashMap::new();
     let mut schema_evoluted: HashMap<String, bool> = HashMap::new();
     let mut stream_alerts_map: HashMap<String, Vec<alerts::Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
     let mut stream_partitioning_map: HashMap<String, PartitioningDetails> = HashMap::new();
 
     for resource_metric in &request.resource_metrics {
@@ -336,10 +336,7 @@ pub async fn handle_grpc_request(
                             local_metric_name.clone()
                         );
                         if let Some(alerts) = stream_alerts_map.get(&key) {
-                            let mut trigger_alerts: Vec<(
-                                Alert,
-                                Vec<json::Map<String, json::Value>>,
-                            )> = Vec::new();
+                            let mut trigger_alerts: TriggerAlertData = Vec::new();
                             for alert in alerts {
                                 if let Ok(Some(v)) = alert.evaluate(Some(val_map)).await {
                                     trigger_alerts.push((alert.clone(), v));

src/service/metrics/otlp_http.rs

@@ -113,7 +113,7 @@ pub async fn metrics_json_handler(
     let mut metric_schema_map: HashMap<String, Schema> = HashMap::new();
     let mut schema_evoluted: HashMap<String, bool> = HashMap::new();
     let mut stream_alerts_map: HashMap<String, Vec<Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
     let mut stream_partitioning_map: HashMap<String, PartitioningDetails> = HashMap::new();
 
     let body: json::Value = match json::from_slice(body.as_ref()) {
@@ -435,10 +435,7 @@ pub async fn metrics_json_handler(
                                 local_metric_name
                             );
                             if let Some(alerts) = stream_alerts_map.get(&key) {
-                                let mut trigger_alerts: Vec<(
-                                    Alert,
-                                    Vec<json::Map<String, json::Value>>,
-                                )> = Vec::new();
+                                let mut trigger_alerts: TriggerAlertData = Vec::new();
                                 for alert in alerts {
                                     if let Ok(Some(v)) = alert.evaluate(Some(val_map)).await {
                                         trigger_alerts.push((alert.clone(), v));

src/service/metrics/prom.rs

@@ -31,7 +31,7 @@ use crate::{
             errors::{Error, Result},
         },
         meta::{
-            alerts::{self, Alert},
+            alerts,
             functions::StreamTransform,
             prom::*,
             search,
@@ -86,7 +86,7 @@ pub async fn remote_write(
     let mut metric_schema_map: HashMap<String, Schema> = HashMap::new();
     let mut schema_evoluted: HashMap<String, bool> = HashMap::new();
     let mut stream_alerts_map: HashMap<String, Vec<alerts::Alert>> = HashMap::new();
-    let mut stream_trigger_map: HashMap<String, TriggerAlertData> = HashMap::new();
+    let mut stream_trigger_map: HashMap<String, Option<TriggerAlertData>> = HashMap::new();
     let mut stream_transform_map: HashMap<String, Vec<StreamTransform>> = HashMap::new();
     let mut stream_partitioning_map: HashMap<String, PartitioningDetails> = HashMap::new();
 
@@ -373,8 +373,7 @@ pub async fn remote_write(
                     metric_name.clone()
                 );
                 if let Some(alerts) = stream_alerts_map.get(&key) {
-                    let mut trigger_alerts: Vec<(Alert, Vec<json::Map<String, json::Value>>)> =
-                        Vec::new();
+                    let mut trigger_alerts: TriggerAlertData = Vec::new();
                     for alert in alerts {
                         if let Ok(Some(v)) = alert.evaluate(Some(val_map)).await {
                             trigger_alerts.push((alert.clone(), v));

src/service/traces/mod.rs

@@ -147,7 +147,7 @@ pub async fn handle_trace_request(
     );
     // End Register Transforms for stream
 
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let min_ts =
         (Utc::now() - Duration::hours(CONFIG.limit.ingest_allowed_upto)).timestamp_micros();
@@ -309,8 +309,7 @@ pub async fn handle_trace_request(
                     // Start check for alert trigger
                     let key = format!("{}/{}/{}", &org_id, StreamType::Traces, traces_stream_name);
                     if let Some(alerts) = stream_alerts_map.get(&key) {
-                        let mut trigger_alerts: Vec<(Alert, Vec<json::Map<String, json::Value>>)> =
-                            Vec::new();
+                        let mut trigger_alerts: TriggerAlertData = Vec::new();
                         for alert in alerts {
                             if let Ok(Some(v)) = alert.evaluate(Some(record_val)).await {
                                 trigger_alerts.push((alert.clone(), v));

src/service/traces/otlp_http.rs

@@ -152,7 +152,7 @@ pub async fn traces_json(
     );
     // End Register Transforms for stream
 
-    let mut trigger: TriggerAlertData = None;
+    let mut trigger: Option<TriggerAlertData> = None;
 
     let mut service_name: String = traces_stream_name.to_string();
     // let export_req: ExportTraceServiceRequest =
@@ -374,10 +374,7 @@ pub async fn traces_json(
                         let key =
                             format!("{}/{}/{}", &org_id, StreamType::Traces, traces_stream_name);
                         if let Some(alerts) = stream_alerts_map.get(&key) {
-                            let mut trigger_alerts: Vec<(
-                                Alert,
-                                Vec<json::Map<String, json::Value>>,
-                            )> = Vec::new();
+                            let mut trigger_alerts: TriggerAlertData = Vec::new();
                             for alert in alerts {
                                 if let Ok(Some(v)) = alert.evaluate(Some(record_val)).await {
                                     trigger_alerts.push((alert.clone(), v));

web/src/components/alerts/AddTemplate.vue

@@ -121,7 +121,7 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
             <div>alert_name, alert_type</div>
             <div>alert_period, alert_operator, alert_threshold</div>
             <div>alert_count, alert_agg_value</div>
-            <div>alert_start_time, alert_end_time</div>
+            <div>alert_start_time, alert_end_time, alert_url</div>
             <div><b>rows</b> multiple lines of row template</div>
             <div><b>All of the stream fields are variables.</b></div>
           </div>