starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-06 08:09:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Alert Manager & integration test fixes (#239)

Browse Source
This commit is contained in:
Ashish Kolhe
2023-03-09 22:58:31 +08:00
committed by GitHub
parent c04a5a567b
commit 373d33f831
4 changed files with 71 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/job/alert_manager.rs
+1 -1
View File
@@ -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;
src/service/alert_manager.rs
+57 -55
View File
@@ -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);
}
}
//}
}
}
}
src/service/db/triggers.rs
+9 -9
View File
@@ -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();
tests/integration_test.rs
+4 -1
View File
@@ -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()