diff --git a/src/job/alert_manager.rs b/src/job/alert_manager.rs index 1b552cc041..9f2e29f6ad 100644 --- a/src/job/alert_manager.rs +++ b/src/job/alert_manager.rs @@ -22,7 +22,7 @@ pub async fn run() -> Result<(), anyhow::Error> { return Ok(()); } // should run it every 10 seconds - let mut interval = time::interval(time::Duration::from_secs(10)); + let mut interval = time::interval(time::Duration::from_secs(30)); interval.tick().await; // trigger the first run loop { interval.tick().await; diff --git a/src/service/alert_manager.rs b/src/service/alert_manager.rs index 398f1ef2ec..0c4a60f118 100644 --- a/src/service/alert_manager.rs +++ b/src/service/alert_manager.rs @@ -42,7 +42,7 @@ pub async fn handle_triggers(alert_name: &str, trigger: Trigger) { match super::db::alerts::get(&trigger.org, &trigger.stream, &trigger.alert_name).await { Ok(result) => { if let Some(alert) = result { - if TRIGGERS_IN_PROCESS.contains_key(alert_name) { + if TRIGGERS_IN_PROCESS.clone().contains_key(alert_name) { let mut curr_time = TRIGGERS_IN_PROCESS.get_mut(&alert_name.to_owned()).unwrap(); let delay = trigger.timestamp - curr_time.updated_at; @@ -69,7 +69,7 @@ pub async fn handle_triggers(alert_name: &str, trigger: Trigger) { } } } - Err(_) => log::error!("[ALERT MANAGER] Error fectching alert",), + Err(_) => log::error!("[ALERT MANAGER] Error fetching alert",), } } @@ -81,65 +81,67 @@ pub async fn handle_trigger(alert_name: &str, alert: Alert) { loop { interval.tick().await; - let curr_ts = Utc::now().timestamp_micros(); - let trigger = TRIGGERS.get(&alert_name.to_owned()).unwrap(); - if TRIGGERS_IN_PROCESS.contains_key(alert_name) { - let trigger_time = TRIGGERS_IN_PROCESS.get(&alert_name.to_owned()).unwrap(); - if curr_ts <= trigger_time.expires_at { - let mut query = alert.query.clone().unwrap(); - let curr_ts = Utc::now().timestamp_micros(); - query.end_time = curr_ts; - query.start_time = curr_ts - get_micros_from_min(alert.duration); - let req: meta::search::Request = Request { - query, - aggs: HashMap::new(), - encoding: meta::search::RequestEncoding::Empty, - }; - //let time_elpased = curr_ts - trigger.clone().last_sent_at; + //let curr_ts = Utc::now().timestamp_micros(); + let loc_triggers = TRIGGERS.clone(); + let trigger = loc_triggers.get(&alert_name.to_owned()).unwrap(); + if TRIGGERS_IN_PROCESS.clone().contains_key(alert_name) { + //let proc_triggers = TRIGGERS_IN_PROCESS.clone(); + //let trigger_time = proc_triggers.get(&alert_name.to_owned()).unwrap(); + //if curr_ts <= trigger_time.expires_at { + let mut query = alert.query.clone().unwrap(); + let curr_ts = Utc::now().timestamp_micros(); + query.end_time = curr_ts; + query.start_time = curr_ts - get_micros_from_min(alert.duration); + let req: meta::search::Request = Request { + query, + aggs: HashMap::new(), + encoding: meta::search::RequestEncoding::Empty, + }; + //let time_elpased = curr_ts - trigger.clone().last_sent_at; - // do search - match SearchService::search(&trigger.org, meta::StreamType::Logs, &req).await { - Ok(res) => { - if !res.hits.is_empty() { - let record = res.hits.first().unwrap().as_object().unwrap(); - if alert.condition.clone().evaluate(record.clone()) { - let curr_ts = Utc::now().timestamp_micros(); - let mut local_trigger = trigger.clone(); + // do search + match SearchService::search(&trigger.org, meta::StreamType::Logs, &req).await { + Ok(res) => { + if !res.hits.is_empty() { + let record = res.hits.first().unwrap().as_object().unwrap(); + if alert.condition.clone().evaluate(record.clone()) { + let curr_ts = Utc::now().timestamp_micros(); + let mut local_trigger = trigger.clone(); - if trigger.clone().last_sent_at == 0 - || (trigger.clone().last_sent_at > 0 - && curr_ts - trigger.clone().last_sent_at - > get_micros_from_min(alert.time_between_alerts)) - { - let _ = send_notification(&alert.destination, &trigger.clone()) - .await; - local_trigger.last_sent_at = curr_ts; - } - //Update trigger for last sent - - local_trigger.count += 1; - let _ = triggers::save_trigger( - alert.name.clone(), - local_trigger.clone(), - ) - .await; + if trigger.clone().last_sent_at == 0 + || (trigger.clone().last_sent_at > 0 + && curr_ts - trigger.clone().last_sent_at + > get_micros_from_min(alert.time_between_alerts)) + { + let _ = + send_notification(&alert.destination, &trigger.clone()).await; + local_trigger.last_sent_at = curr_ts; } - } - } - Err(err) => { - log::error!("search error: {:?}", err); + //Update trigger for last sent + + local_trigger.count += 1; + let _ = + triggers::save_trigger(alert.name.clone(), local_trigger.clone()) + .await; + } /* else { + log::info!( + "Setting to in-active expired trigger with name {}", + &alert_name.to_owned() + ); + let mut local_trigger = trigger.clone(); + local_trigger.is_valid = false; + let _ = + triggers::save_trigger(alert.name.clone(), local_trigger.clone()) + .await; + break; + } */ } } - } else { - log::info!( - "Setting to in-active expired trigger with name {}", - &alert_name.to_owned() - ); - let mut local_trigger = trigger.clone(); - local_trigger.is_valid = false; - let _ = triggers::save_trigger(alert.name.clone(), local_trigger.clone()).await; - break; + Err(err) => { + log::error!("search error: {:?}", err); + } } + //} } } } diff --git a/src/service/db/triggers.rs b/src/service/db/triggers.rs index 9ce56f9127..f02528d313 100644 --- a/src/service/db/triggers.rs +++ b/src/service/db/triggers.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use crate::common::json; -use crate::infra::config::{TRIGGERS, TRIGGERS_IN_PROCESS}; +use crate::infra::config::TRIGGERS; use crate::infra::db::Event; use crate::meta::alert::Trigger; @@ -54,9 +54,9 @@ pub async fn cache() -> Result<(), anyhow::Error> { let item_key = item_key.strip_prefix(key).unwrap(); let json_val: Trigger = json::from_slice(&item_value).unwrap(); //let trigger_key = &item_key[0..item_key.rfind('/').unwrap()]; - if json_val.is_valid { - TRIGGERS.insert(item_key.to_owned(), json_val); - } + //if json_val.is_valid { + TRIGGERS.insert(item_key.to_owned(), json_val); + //} } log::info!("[TRACE] Triggers Cached"); Ok(()) @@ -80,13 +80,13 @@ pub async fn watch() -> Result<(), anyhow::Error> { Event::Put(ev) => { let item_key = ev.key.strip_prefix(key).unwrap(); let item_value: Trigger = json::from_slice(&ev.value.unwrap()).unwrap(); - // to avoid timeour error from etcd - if item_value.is_valid { - TRIGGERS.insert(item_key.to_string(), item_value.clone()); - } else { + //if item_value.is_valid { + TRIGGERS.insert(item_key.to_string(), item_value.clone()); + //TRIGGERS_IN_PROCESS.remove(item_key); + /* } else { TRIGGERS_IN_PROCESS.remove(item_key); TRIGGERS.remove(item_key); - } + } */ } Event::Delete(ev) => { let item_key = ev.key.strip_prefix(key).unwrap(); diff --git a/tests/integration_test.rs b/tests/integration_test.rs index d3f7ae8b2a..2eb100e9c9 100644 --- a/tests/integration_test.rs +++ b/tests/integration_test.rs @@ -818,7 +818,10 @@ mod tests { "duration": 5, "frequency": 1, "time_between_alerts": 10, - "destination": "https://dummy/alert" + "destination":[{ + "url":"https://dummy/alert", + "type":"slack" + }] }"#; let app = test::init_service( App::new()