diff --git a/src/common/infra/cluster.rs b/src/common/infra/cluster.rs index 77a1b0b6b2..03ea31999b 100644 --- a/src/common/infra/cluster.rs +++ b/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> { if NODES.is_empty() { return None; diff --git a/src/common/utils/base64.rs b/src/common/utils/base64.rs index e603931efb..fe0513809b 100644 --- a/src/common/utils/base64.rs +++ b/src/common/utils/base64.rs @@ -48,6 +48,10 @@ pub(crate) fn decode_raw(s: &str) -> Result, 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::*; diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 3d054ffab6..325b95c9a5 100644 --- a/src/config/src/config.rs +++ b/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); } diff --git a/src/handler/http/request/status/mod.rs b/src/handler/http/request/status/mod.rs index dbcac97a42..a22450f721 100644 --- a/src/handler/http/request/status/mod.rs +++ b/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 { + 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::>::from_query(req.query_string()).unwrap(); + let enable = match query.get("value") { + Some(v) => v.parse::().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 { + 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)), + } +} diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index dae85377fc..d952de8898 100644 --- a/src/handler/http/router/mod.rs +++ b/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) diff --git a/src/ingester/src/lib.rs b/src/ingester/src/lib.rs index 4f17ad3db1..56c3cda0e3 100644 --- a/src/ingester/src/lib.rs +++ b/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 diff --git a/src/ingester/src/writer.rs b/src/ingester/src/writer.rs index d32b1f92d7..3d4dc4437d 100644 --- a/src/ingester/src/writer.rs +++ b/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::>(); + 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) diff --git a/src/service/alerts/alert_manager.rs b/src/service/alerts/alert_manager.rs index d70d7b5904..416b6db354 100644 --- a/src/service/alerts/alert_manager.rs +++ b/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 diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index 2a7b589bdc..9550d77539 100644 --- a/src/service/alerts/mod.rs +++ b/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] 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] } } } - 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] } 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] } 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] &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] rows_tpl.join("\\\\n") } -fn process_dest_template( +async fn process_dest_template( tpl: &str, alert: &Alert, rows: &[Map], @@ -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() + } +} diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index e54258d8b0..1e0ce033d4 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -50,7 +50,7 @@ use crate::{ pub mod grpc; -pub type TriggerAlertData = Option>)>>; +pub type TriggerAlertData = Vec<(Alert, Vec>)>; pub fn compile_vrl_function(func: &str, org_id: &str) -> Result { 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) { if trigger.is_none() { return; } diff --git a/src/service/logs/bulk.rs b/src/service/logs/bulk.rs index ec23c47214..bf60c8e387 100644 --- a/src/service/logs/bulk.rs +++ b/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 = HashMap::new(); + let mut stream_trigger_map: HashMap> = HashMap::new(); let mut next_line_is_data = false; let reader = BufReader::new(body.as_ref()); diff --git a/src/service/logs/ingest.rs b/src/service/logs/ingest.rs index 3e57553043..d653932493 100644 --- a/src/service/logs/ingest.rs +++ b/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 = None; let partition_det = crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map).await; diff --git a/src/service/logs/mod.rs b/src/service/logs/mod.rs index 24501cc767..4e1e7f313f 100644 --- a/src/service/logs/mod.rs +++ b/src/service/logs/mod.rs @@ -151,8 +151,8 @@ async fn add_valid_record( write_buf: &mut HashMap, record_val: &mut Map, need_trigger: bool, -) -> Result { - let mut trigger: Vec<(Alert, Vec>)> = Vec::new(); +) -> Result, anyhow::Error> { + let mut trigger: TriggerAlertData = Vec::new(); let timestamp: i64 = record_val .get(&CONFIG.common.column_timestamp) .unwrap() diff --git a/src/service/logs/multi.rs b/src/service/logs/multi.rs index e657a22ec6..e9b685d4ab 100644 --- a/src/service/logs/multi.rs +++ b/src/service/logs/multi.rs @@ -84,7 +84,7 @@ async fn ingest_inner( let mut stream_alerts_map: HashMap> = HashMap::new(); let mut stream_status = StreamStatus::new(stream_name); - let mut trigger: TriggerAlertData = None; + let mut trigger: Option = None; // Start Register Transforms for stream let (local_trans, stream_vrl_map) = crate::service::ingestion::register_stream_transforms( diff --git a/src/service/logs/otlp_grpc.rs b/src/service/logs/otlp_grpc.rs index 4046110941..9b7c3533a5 100644 --- a/src/service/logs/otlp_grpc.rs +++ b/src/service/logs/otlp_grpc.rs @@ -84,7 +84,7 @@ pub async fn usage_ingest( let mut stream_alerts_map: HashMap> = HashMap::new(); let mut stream_status = StreamStatus::new(stream_name); - let mut trigger: TriggerAlertData = None; + let mut trigger: Option = 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 = None; let mut data_buf: HashMap = HashMap::new(); diff --git a/src/service/logs/otlp_http.rs b/src/service/logs/otlp_http.rs index 964a07525a..9aa29f621f 100644 --- a/src/service/logs/otlp_http.rs +++ b/src/service/logs/otlp_http.rs @@ -134,7 +134,7 @@ pub async fn logs_json_handler( let mut stream_alerts_map: HashMap> = 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 = None; let min_ts = (Utc::now() - Duration::hours(CONFIG.limit.ingest_allowed_upto)).timestamp_micros(); diff --git a/src/service/logs/syslog.rs b/src/service/logs/syslog.rs index ffe64d7adf..96270ba7dc 100644 --- a/src/service/logs/syslog.rs +++ b/src/service/logs/syslog.rs @@ -89,7 +89,7 @@ pub async fn ingest(msg: &str, addr: SocketAddr) -> Result = None; let partition_det = crate::service::ingestion::get_stream_partition_keys(stream_name, &stream_schema_map).await; diff --git a/src/service/metrics/otlp_grpc.rs b/src/service/metrics/otlp_grpc.rs index 4c1003bca4..0ba9761e9f 100644 --- a/src/service/metrics/otlp_grpc.rs +++ b/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 = HashMap::new(); let mut schema_evoluted: HashMap = HashMap::new(); let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap = HashMap::new(); + let mut stream_trigger_map: HashMap> = HashMap::new(); let mut stream_partitioning_map: HashMap = 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>, - )> = 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)); diff --git a/src/service/metrics/otlp_http.rs b/src/service/metrics/otlp_http.rs index cfa8e08f60..a9ba7d021d 100644 --- a/src/service/metrics/otlp_http.rs +++ b/src/service/metrics/otlp_http.rs @@ -113,7 +113,7 @@ pub async fn metrics_json_handler( let mut metric_schema_map: HashMap = HashMap::new(); let mut schema_evoluted: HashMap = HashMap::new(); let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap = HashMap::new(); + let mut stream_trigger_map: HashMap> = HashMap::new(); let mut stream_partitioning_map: HashMap = 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>, - )> = 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)); diff --git a/src/service/metrics/prom.rs b/src/service/metrics/prom.rs index 9ecda5b826..881e1274c0 100644 --- a/src/service/metrics/prom.rs +++ b/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 = HashMap::new(); let mut schema_evoluted: HashMap = HashMap::new(); let mut stream_alerts_map: HashMap> = HashMap::new(); - let mut stream_trigger_map: HashMap = HashMap::new(); + let mut stream_trigger_map: HashMap> = HashMap::new(); let mut stream_transform_map: HashMap> = HashMap::new(); let mut stream_partitioning_map: HashMap = 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>)> = - 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)); diff --git a/src/service/traces/mod.rs b/src/service/traces/mod.rs index e2a480a64e..6ffedcca82 100644 --- a/src/service/traces/mod.rs +++ b/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 = 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>)> = - 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)); diff --git a/src/service/traces/otlp_http.rs b/src/service/traces/otlp_http.rs index 6435c110f8..1ab379f3fe 100644 --- a/src/service/traces/otlp_http.rs +++ b/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 = 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>, - )> = 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)); diff --git a/web/src/components/alerts/AddTemplate.vue b/web/src/components/alerts/AddTemplate.vue index 61335d525a..62f2cb6320 100644 --- a/web/src/components/alerts/AddTemplate.vue +++ b/web/src/components/alerts/AddTemplate.vue @@ -121,7 +121,7 @@ along with this program. If not, see .
alert_name, alert_type
alert_period, alert_operator, alert_threshold
alert_count, alert_agg_value
-
alert_start_time, alert_end_time
+
alert_start_time, alert_end_time, alert_url
rows multiple lines of row template
All of the stream fields are variables.