From ca52eca9d8d4db39b1ccbe8ad8567bdcacce77d8 Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Fri, 10 Nov 2023 21:38:34 +0800 Subject: [PATCH] feat: delete API should return 200, 404 or 500 (#1924) --- src/common/utils/notification.rs | 121 +++++++------- .../http/request/alerts/destinations.rs | 22 ++- src/handler/http/request/alerts/mod.rs | 126 ++++++-------- src/handler/http/request/alerts/templates.rs | 1 + .../http/request/dashboards/folders.rs | 5 +- src/handler/http/request/dashboards/mod.rs | 5 +- src/handler/http/request/functions/mod.rs | 41 ++--- src/service/alerts/destinations.rs | 48 +++--- src/service/alerts/mod.rs | 156 ++++++------------ src/service/alerts/templates.rs | 33 ++-- src/service/dashboards/folders.rs | 47 +++--- src/service/dashboards/mod.rs | 61 ++++--- src/service/db/alerts/destinations.rs | 34 ++-- src/service/db/alerts/mod.rs | 11 +- src/service/db/alerts/templates.rs | 38 ++--- src/service/functions.rs | 36 ++-- src/service/triggers.rs | 6 +- 17 files changed, 360 insertions(+), 431 deletions(-) diff --git a/src/common/utils/notification.rs b/src/common/utils/notification.rs index d01e37225f..6daf73a74e 100644 --- a/src/common/utils/notification.rs +++ b/src/common/utils/notification.rs @@ -27,72 +27,71 @@ pub async fn send_notification( false => "Scheduled", }; let curr_ts = chrono::Utc::now().timestamp_micros(); - match db::alerts::destinations::get(&trigger.org, &alert.destination).await { - Ok(dest) => match dest { - Some(local_dest) => { - let body = local_dest.template.unwrap().body; - let resp_str = json::to_string(&body).unwrap(); + let local_dest = match db::alerts::destinations::get(&trigger.org, &alert.destination).await { + Ok(v) => v, + Err(_) => { + log::error!("Destination Not found: {}", &alert.destination); + return Ok(()); + } + }; - let mut resp = resp_str - .replace("{stream_name}", &trigger.stream) - .replace("{org_name}", &trigger.org) - .replace("{alert_name}", &trigger.alert_name) - .replace("{alert_type}", alert_type) - .replace("{timestamp}", &curr_ts.to_string()); + let body = local_dest.template.unwrap().body; + let resp_str = json::to_string(&body).unwrap(); - // Replace contextual information with values if any from alert - if alert.context_attributes.is_some() { - for (key, value) in alert.context_attributes.as_ref().unwrap() { - resp = resp.replace(&format!("{{{key}}}"), value) - } - } + let mut resp = resp_str + .replace("{stream_name}", &trigger.stream) + .replace("{org_name}", &trigger.org) + .replace("{alert_name}", &trigger.alert_name) + .replace("{alert_type}", alert_type) + .replace("{timestamp}", &curr_ts.to_string()); - let msg: Value = json::from_str(&resp).unwrap(); - let msg: Value = match &msg { - Value::String(obj) => match json::from_str(obj) { - Ok(obj) => obj, - Err(_) => msg, - }, - _ => msg, - }; - let client = if local_dest.skip_tls_verify { - reqwest::Client::builder() - .danger_accept_invalid_certs(true) - .build()? - } else { - reqwest::Client::new() - }; - let url = url::Url::parse(&local_dest.url); - let mut req = match local_dest.method { - alert::AlertHTTPType::POST => client.post(url.unwrap()), - alert::AlertHTTPType::PUT => client.put(url.unwrap()), - alert::AlertHTTPType::GET => client.get(url.unwrap()), - } - .header("Content-type", "application/json"); + // Replace contextual information with values if any from alert + if alert.context_attributes.is_some() { + for (key, value) in alert.context_attributes.as_ref().unwrap() { + resp = resp.replace(&format!("{{{key}}}"), value) + } + } - // Add additional headers if any from destination description - if local_dest.headers.is_some() { - for (key, value) in local_dest.headers.unwrap() { - if !key.is_empty() && !value.is_empty() { - req = req.header(key, value); - } - } - }; - - let resp = req.json(&msg).send().await; - match resp { - Ok(resp) => { - if !resp.status().is_success() { - log::error!("Notification sent error: {:?}", resp.bytes().await); - } - } - Err(err) => log::error!("Notification sending error {:?}", err), - } - } - - None => log::error!("Destination Not found"), + let msg: Value = json::from_str(&resp).unwrap(); + let msg: Value = match &msg { + Value::String(obj) => match json::from_str(obj) { + Ok(obj) => obj, + Err(_) => msg, }, - Err(err) => log::error!("Error sending notification {:?}", err), + _ => msg, + }; + let client = if local_dest.skip_tls_verify { + reqwest::Client::builder() + .danger_accept_invalid_certs(true) + .build()? + } else { + reqwest::Client::new() + }; + let url = url::Url::parse(&local_dest.url); + let mut req = match local_dest.method { + alert::AlertHTTPType::POST => client.post(url.unwrap()), + alert::AlertHTTPType::PUT => client.put(url.unwrap()), + alert::AlertHTTPType::GET => client.get(url.unwrap()), + } + .header("Content-type", "application/json"); + + // Add additional headers if any from destination description + if local_dest.headers.is_some() { + for (key, value) in local_dest.headers.unwrap() { + if !key.is_empty() && !value.is_empty() { + req = req.header(key, value); + } + } + }; + + let resp = req.json(&msg).send().await; + match resp { + Ok(resp) => { + if !resp.status().is_success() { + log::error!("Notification sent error: {:?}", resp.bytes().await); + } + } + Err(err) => log::error!("Notification sending error {:?}", err), } Ok(()) diff --git a/src/handler/http/request/alerts/destinations.rs b/src/handler/http/request/alerts/destinations.rs index 149670fb43..2b4b9e6689 100644 --- a/src/handler/http/request/alerts/destinations.rs +++ b/src/handler/http/request/alerts/destinations.rs @@ -45,19 +45,16 @@ pub async fn save_destination( let dest = dest.into_inner(); - match db::alerts::templates::get(org_id.as_str(), &dest.template).await { - Ok(temp) => match temp { - Some(_) => destinations::save_destination(org_id, name, dest).await, - None => Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - "Please specify valid template".to_string(), - ))), - }, - Err(_) => Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + if db::alerts::templates::get(org_id.as_str(), &dest.template) + .await + .is_err() + { + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( http::StatusCode::BAD_REQUEST.into(), "Please specify valid template".to_string(), - ))), + ))); } + destinations::save_destination(org_id, name, dest).await } /** ListDestinations */ @@ -117,8 +114,9 @@ async fn get_destination(path: web::Path<(String, String)>) -> impl Responder { ("destination_name" = String, Path, description = "Destination name"), ), responses( - (status = 200, description="Success", content_type = "application/json", body = HttpResponse), - (status = 404, description="NotFound", content_type = "application/json", body = HttpResponse), + (status = 200, description = "Success", content_type = "application/json", body = HttpResponse), + (status = 404, description = "NotFound", content_type = "application/json", body = HttpResponse), + (status = 500, description = "Error", content_type = "application/json", body = HttpResponse), ) )] #[delete("/{org_id}/alerts/destinations/{destination_name}")] diff --git a/src/handler/http/request/alerts/mod.rs b/src/handler/http/request/alerts/mod.rs index f47440d919..2214110ddb 100644 --- a/src/handler/http/request/alerts/mod.rs +++ b/src/handler/http/request/alerts/mod.rs @@ -12,17 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub mod destinations; -pub mod templates; - use actix_web::{delete, get, http, post, put, web, HttpRequest, HttpResponse, Responder}; use ahash::AHashMap as HashMap; use std::io::Error; -use crate::common::meta::{self, alert::Alert, StreamType}; +use crate::common::meta::{alert::Alert, http::HttpResponse as MetaHttpResponse}; use crate::common::utils::http::get_stream_type_from_request; use crate::service::alerts; +pub mod destinations; +pub mod templates; + /** CreateAlert */ #[utoipa::path( context_path = "/api", @@ -50,25 +50,20 @@ pub async fn save_alert( let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } alerts::save_alert( - org_id, - stream_name, - stream_type.unwrap(), - name, + &org_id, + &stream_name, + stream_type, + &name, alert.into_inner(), ) .await @@ -93,21 +88,16 @@ pub async fn save_alert( async fn list_stream_alerts(path: web::Path<(String, String)>, req: HttpRequest) -> impl Responder { let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { + let stream_type = match get_stream_type_from_request(&query) { Ok(v) => v, Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } - alerts::list_alert(org_id, Some(stream_name.as_str()), stream_type).await + alerts::list_alert(&org_id, Some(stream_name.as_str()), stream_type).await } /** ListAlerts */ @@ -129,21 +119,16 @@ async fn list_stream_alerts(path: web::Path<(String, String)>, req: HttpRequest) async fn list_alerts(path: web::Path, req: HttpRequest) -> impl Responder { let org_id = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { + let stream_type = match get_stream_type_from_request(&query) { Ok(v) => v, Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } - alerts::list_alert(org_id, None, stream_type).await + alerts::list_alert(&org_id, None, stream_type).await } /** GetAlertByName */ @@ -168,21 +153,16 @@ async fn list_alerts(path: web::Path, req: HttpRequest) -> impl Responde async fn get_alert(path: web::Path<(String, String, String)>, req: HttpRequest) -> impl Responder { let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } - alerts::get_alert(org_id, stream_name, stream_type.unwrap(), name).await + alerts::get_alert(&org_id, &stream_name, stream_type, &name).await } /** DeleteAlert */ @@ -201,6 +181,7 @@ async fn get_alert(path: web::Path<(String, String, String)>, req: HttpRequest) responses( (status = 200, description="Success", content_type = "application/json", body = HttpResponse), (status = 404, description="NotFound", content_type = "application/json", body = HttpResponse), + (status = 500, description="Error", content_type = "application/json", body = HttpResponse), ) )] #[delete("/{org_id}/{stream_name}/alerts/{alert_name}")] @@ -210,21 +191,16 @@ async fn delete_alert( ) -> impl Responder { let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } - alerts::delete_alert(org_id, stream_name, stream_type.unwrap(), name).await + alerts::delete_alert(&org_id, &stream_name, stream_type, &name).await } /** TriggerAlert */ @@ -252,20 +228,14 @@ async fn trigger_alert( ) -> impl Responder { let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { - return Ok( - HttpResponse::BadRequest().json(meta::http::HttpResponse::error( - http::StatusCode::BAD_REQUEST.into(), - e.to_string(), - )), - ) + return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error( + http::StatusCode::BAD_REQUEST.into(), + e.to_string(), + ))) } }; - if stream_type.is_none() { - stream_type = Some(StreamType::Logs) - } - - alerts::trigger_alert(org_id, stream_name, stream_type.unwrap(), name).await + alerts::trigger_alert(&org_id, &stream_name, stream_type, &name).await } diff --git a/src/handler/http/request/alerts/templates.rs b/src/handler/http/request/alerts/templates.rs index 15d23d2a48..6dcba24571 100644 --- a/src/handler/http/request/alerts/templates.rs +++ b/src/handler/http/request/alerts/templates.rs @@ -102,6 +102,7 @@ async fn get_template(path: web::Path<(String, String)>) -> impl Responder { responses( (status = 200, description="Success", content_type = "application/json", body = HttpResponse), (status = 404, description="NotFound", content_type = "application/json", body = HttpResponse), + (status = 500, description="Error", content_type = "application/json", body = HttpResponse), ) )] #[delete("/{org_id}/alerts/templates/{template_name}")] diff --git a/src/handler/http/request/dashboards/folders.rs b/src/handler/http/request/dashboards/folders.rs index 7638ddecaf..8378740cb0 100644 --- a/src/handler/http/request/dashboards/folders.rs +++ b/src/handler/http/request/dashboards/folders.rs @@ -142,8 +142,9 @@ pub async fn get_folder(path: web::Path<(String, String)>) -> Result, req: HttpRequest) -> i ("dashboard_id" = String, Path, description = "Dashboard ID"), ), responses( - (status = StatusCode::OK, description = "Dashboard deleted", body = HttpResponse), - (status = StatusCode::NOT_FOUND, description = "Dashboard not found", body = HttpResponse), + (status = StatusCode::OK, description = "Success", body = HttpResponse), + (status = StatusCode::NOT_FOUND, description = "NotFound", body = HttpResponse), + (status = StatusCode::INTERNAL_SERVER_ERROR, description = "Error", body = HttpResponse), ), )] #[delete("/{org_id}/dashboards/{dashboard_id}")] diff --git a/src/handler/http/request/functions/mod.rs b/src/handler/http/request/functions/mod.rs index e9589f9974..c6db8cfa61 100644 --- a/src/handler/http/request/functions/mod.rs +++ b/src/handler/http/request/functions/mod.rs @@ -142,8 +142,8 @@ async fn list_stream_functions( ) -> Result { let (org_id, stream_name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( @@ -153,11 +153,7 @@ async fn list_stream_functions( ) } }; - if stream_type.is_none() { - stream_type = Some(meta::StreamType::Logs); - } - crate::service::functions::list_stream_functions(org_id, stream_type.unwrap(), stream_name) - .await + crate::service::functions::list_stream_functions(&org_id, stream_type, &stream_name).await } /** RemoveStreamFunction*/ @@ -185,8 +181,8 @@ async fn delete_stream_function( ) -> Result { let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( @@ -196,16 +192,8 @@ async fn delete_stream_function( ) } }; - if stream_type.is_none() { - stream_type = Some(meta::StreamType::Logs); - } - crate::service::functions::delete_stream_function( - org_id, - stream_type.unwrap(), - stream_name, - name, - ) - .await + crate::service::functions::delete_stream_function(&org_id, stream_type, &stream_name, &name) + .await } /** ApplyFunctionToStream */ @@ -235,8 +223,8 @@ pub async fn add_function_to_stream( ) -> Result { let (org_id, stream_name, name) = path.into_inner(); let query = web::Query::>::from_query(req.query_string()).unwrap(); - let mut stream_type = match get_stream_type_from_request(&query) { - Ok(v) => v, + let stream_type = match get_stream_type_from_request(&query) { + Ok(v) => v.unwrap_or_default(), Err(e) => { return Ok( HttpResponse::BadRequest().json(meta::http::HttpResponse::error( @@ -246,14 +234,11 @@ pub async fn add_function_to_stream( ) } }; - if stream_type.is_none() { - stream_type = Some(meta::StreamType::Logs); - } crate::service::functions::add_function_to_stream( - org_id, - stream_type.unwrap(), - stream_name, - name, + &org_id, + stream_type, + &stream_name, + &name, stream_order.into_inner(), ) .await diff --git a/src/service/alerts/destinations.rs b/src/service/alerts/destinations.rs index 5baf3bdee5..70296126df 100644 --- a/src/service/alerts/destinations.rs +++ b/src/service/alerts/destinations.rs @@ -12,13 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use actix_web::{http, HttpResponse}; use std::io::Error; -use actix_web::{http, HttpResponse}; - use crate::common::infra::config::STREAM_ALERTS; -use crate::common::meta::alert::AlertDestination; -use crate::common::meta::http::HttpResponse as MetaHttpResponse; +use crate::common::meta::{alert::AlertDestination, http::HttpResponse as MetaHttpResponse}; use crate::service::db; #[tracing::instrument(skip(destination))] @@ -58,28 +56,28 @@ pub async fn delete_destination(org_id: String, name: String) -> Result { - Ok(HttpResponse::Ok().json(MetaHttpResponse::message( - http::StatusCode::OK.into(), - "Alert destination deleted ".to_string(), - ))) - } - Err(e) => { - Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( - http::StatusCode::NOT_FOUND.into(), - e.to_string(), - ))) - } - }; + if db::alerts::destinations::get(org_id.as_str(), name.as_str()) + .await + .is_err() + { + return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + http::StatusCode::NOT_FOUND.into(), + "Alert destination not found".to_string(), + ))); + } + let result = db::alerts::destinations::delete(org_id.as_str(), name.as_str()).await; + match result { + Ok(_) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( + http::StatusCode::OK.into(), + "Alert destination deleted ".to_string(), + ))), + Err(e) => Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ), } - - Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( - http::StatusCode::NOT_FOUND.into(), - "Alert destination not found".to_string(), - ))) } #[tracing::instrument] diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index 7d26af9383..ee4cf466b8 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -33,10 +33,10 @@ pub mod templates; #[tracing::instrument(skip_all)] pub async fn save_alert( - org_id: String, - stream_name: String, + org_id: &str, + stream_name: &str, stream_type: StreamType, - name: String, + name: &str, mut alert: Alert, ) -> Result { alert.stream = stream_name.to_string(); @@ -45,19 +45,18 @@ pub async fn save_alert( let in_dest = alert.clone().destination; // before saving alert check alert destination - let dest = db::alerts::destinations::get(&org_id, &in_dest) + if db::alerts::destinations::get(org_id, &in_dest) .await - .unwrap(); - - if dest.is_none() { + .is_err() + { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( http::StatusCode::NOT_FOUND.into(), format!("Destination with name {in_dest} not found"), ))); - } + }; // before saving alert check column type to decide numeric condition - let schema = db::schema::get(&org_id, &stream_name, stream_type) + let schema = db::schema::get(org_id, stream_name, stream_type) .await .unwrap(); let fields = schema.to_cloned_fields(); @@ -119,24 +118,18 @@ pub async fn save_alert( } } - db::alerts::set( - org_id.as_str(), - stream_name.as_str(), - stream_type, - name.as_str(), - alert.clone(), - ) - .await - .unwrap(); + db::alerts::set(org_id, stream_name, stream_type, name, alert.clone()) + .await + .unwrap(); // For non-ingest alert set trigger immediately if !alert.is_real_time { let trigger = Trigger { timestamp: Utc::now().timestamp_micros(), is_valid: true, alert_name: alert.name, - stream: stream_name, + stream: stream_name.to_string(), stream_type, - org: org_id, + org: org_id.to_string(), last_sent_at: 0, count: 0, is_ingest_time: false, @@ -152,11 +145,11 @@ pub async fn save_alert( #[tracing::instrument] pub async fn list_alert( - org_id: String, + org_id: &str, stream_name: Option<&str>, stream_type: Option, ) -> Result { - let alerts_list = db::alerts::list(org_id.as_str(), stream_name, stream_type) + let alerts_list = db::alerts::list(org_id, stream_name, stream_type) .await .unwrap(); Ok(HttpResponse::Ok().json(AlertList { list: alerts_list })) @@ -164,44 +157,42 @@ pub async fn list_alert( #[tracing::instrument] pub async fn delete_alert( - org_id: String, - stream_name: String, + org_id: &str, + stream_name: &str, stream_type: StreamType, - name: String, + name: &str, ) -> Result { - let result = db::alerts::delete( - org_id.as_str(), - stream_name.as_str(), - stream_type, - name.as_str(), - ) - .await; - match result { + if db::alerts::get(org_id, stream_name, stream_type, name) + .await + .is_err() + { + return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + http::StatusCode::NOT_FOUND.into(), + "Alert not found".to_string(), + ))); + } + match db::alerts::delete(org_id, stream_name, stream_type, name).await { Ok(_) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( http::StatusCode::OK.into(), "Alert deleted ".to_string(), ))), - Err(e) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( - http::StatusCode::NOT_FOUND.into(), - e.to_string(), - ))), + Err(e) => Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ), } } #[tracing::instrument] pub async fn get_alert( - org_id: String, - stream_name: String, + org_id: &str, + stream_name: &str, stream_type: StreamType, - name: String, + name: &str, ) -> Result { - let result = db::alerts::get( - org_id.as_str(), - stream_name.as_str(), - stream_type, - name.as_str(), - ) - .await; + let result = db::alerts::get(org_id, stream_name, stream_type, name).await; match result { Ok(alert) => Ok(HttpResponse::Ok().json(alert)), Err(_) => Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( @@ -213,27 +204,21 @@ pub async fn get_alert( #[tracing::instrument] pub async fn trigger_alert( - org_id: String, - stream_name: String, + org_id: &str, + stream_name: &str, stream_type: StreamType, - name: String, + name: &str, ) -> Result { - let result = db::alerts::get( - org_id.as_str(), - stream_name.as_str(), - stream_type, - name.as_str(), - ) - .await; + let result = db::alerts::get(org_id, stream_name, stream_type, name).await; match result { Ok(Some(alert)) => { let trigger = Trigger { timestamp: Utc::now().timestamp_micros(), is_valid: false, - alert_name: name, - stream: stream_name, - org: org_id, + alert_name: name.to_string(), + stream: stream_name.to_string(), + org: org_id.to_string(), last_sent_at: 0, count: 0, is_ingest_time: alert.is_real_time, @@ -298,43 +283,19 @@ mod test { let alert_name = "500error"; let alert_stream: &str = "olympics"; let alert = prepare_test_alert_object(alert_name, alert_stream); - let res = save_alert( - "nexus".to_string(), - "olympics".to_string(), - StreamType::Logs, - "500error".to_string(), - alert, - ) - .await; + let res = save_alert("nexus", "olympics", StreamType::Logs, "500error", alert).await; assert!(res.is_ok()); - let list_res = list_alert( - "nexus".to_string(), - Some("olympics"), - Some(StreamType::Logs), - ) - .await; + let list_res = list_alert("nexus", Some("olympics"), Some(StreamType::Logs)).await; assert!(list_res.is_ok()); - let list_res = list_alert("nexus".to_string(), None, None).await; + let list_res = list_alert("nexus", None, None).await; assert!(list_res.is_ok()); - let get_res = get_alert( - "nexus".to_string(), - "olympics".to_string(), - StreamType::Logs, - "500error".to_string(), - ) - .await; + let get_res = get_alert("nexus", "olympics", StreamType::Logs, "500error").await; assert!(get_res.is_ok()); - let del_res = delete_alert( - "nexus".to_string(), - "olympics".to_string(), - StreamType::Logs, - "500error".to_string(), - ) - .await; + let del_res = delete_alert("nexus", "olympics", StreamType::Logs, "500error").await; assert!(del_res.is_ok()); } @@ -343,23 +304,10 @@ mod test { let alert_name = "500error"; let alert_stream: &str = "olympics"; let alert = prepare_test_alert_object(alert_name, alert_stream); - let res = save_alert( - "nexus".to_string(), - "olympics".to_string(), - StreamType::Logs, - "500error".to_string(), - alert, - ) - .await; + let res = save_alert("nexus", "olympics", StreamType::Logs, "500error", alert).await; assert!(res.is_ok()); - let del_res = trigger_alert( - "nexus".to_string(), - "olympics".to_string(), - StreamType::Logs, - "500error".to_string(), - ) - .await; + let del_res = trigger_alert("nexus", "olympics", StreamType::Logs, "500error").await; assert!(del_res.is_ok()); } } diff --git a/src/service/alerts/templates.rs b/src/service/alerts/templates.rs index a009f9c197..2828c7297d 100644 --- a/src/service/alerts/templates.rs +++ b/src/service/alerts/templates.rs @@ -12,13 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use actix_web::{http, HttpResponse}; use std::io::Error; -use actix_web::{http, HttpResponse}; - use crate::common::infra::config::ALERTS_DESTINATIONS; -use crate::common::meta::alert::DestinationTemplate; -use crate::common::meta::http::HttpResponse as MetaHttpResponse; +use crate::common::meta::{alert::DestinationTemplate, http::HttpResponse as MetaHttpResponse}; use crate::service::db; #[tracing::instrument(skip_all)] @@ -58,16 +56,27 @@ pub async fn delete_template(org_id: String, name: String) -> Result Ok(HttpResponse::Ok().json(MetaHttpResponse::message( + http::StatusCode::OK.into(), + "Alert template deleted".to_string(), + ))), + Err(e) => Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ), + } } #[tracing::instrument] diff --git a/src/service/dashboards/folders.rs b/src/service/dashboards/folders.rs index 347fe4369d..b8c6d5e4e1 100644 --- a/src/service/dashboards/folders.rs +++ b/src/service/dashboards/folders.rs @@ -12,17 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. -use actix_web::{ - http::{self, StatusCode}, - HttpResponse, -}; +use actix_web::{http, HttpResponse}; use std::io::Error; use crate::common::meta::{ dashboards::{Folder, FolderList, DEFAULT_FOLDER}, http::HttpResponse as MetaHttpResponse, }; - use crate::service::db; #[tracing::instrument(skip(folder))] @@ -35,7 +31,7 @@ pub async fn save_folder( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::message( http::StatusCode::BAD_REQUEST.into(), - "can't update default folder".to_string(), + "can't update default Dashboard folder".to_string(), )), ); } @@ -64,7 +60,7 @@ pub async fn update_folder( return Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::message( http::StatusCode::BAD_REQUEST.into(), - "can't update default folder".to_string(), + "can't update default Dashboard folder".to_string(), )), ); } @@ -81,7 +77,7 @@ pub async fn update_folder( Ok(HttpResponse::Ok().json(MetaHttpResponse::message( http::StatusCode::OK.into(), - "folder updated".to_string(), + "Dashboard folder updated".to_string(), ))) } @@ -100,8 +96,8 @@ pub async fn get_folder(org_id: &str, folder_id: &str) -> Result Result Ok(HttpResponse::Ok().json(MetaHttpResponse::message( + http::StatusCode::OK.into(), + "Dashboard folder deleted".to_string(), + ))), + Err(e) => Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + e.to_string(), + )), + ), + } } diff --git a/src/service/dashboards/mod.rs b/src/service/dashboards/mod.rs index f96cae48e5..507a38f9ce 100644 --- a/src/service/dashboards/mod.rs +++ b/src/service/dashboards/mod.rs @@ -12,14 +12,16 @@ // See the License for the specific language governing permissions and // limitations under the License. +use actix_web::{http, web, HttpResponse}; use std::io; -use actix_web::{http::StatusCode, HttpResponse}; -use actix_web::web; - -use crate::common::meta::{self, http::HttpResponse as MetaHttpResponse}; -use crate::common::meta::dashboards::DEFAULT_FOLDER; -use crate::common::utils::json; +use crate::common::{ + meta::{ + dashboards::{Dashboards, Folder, DEFAULT_FOLDER}, + http::HttpResponse as MetaHttpResponse, + }, + utils::json, +}; use crate::service::db::dashboards; pub mod folders; @@ -40,7 +42,7 @@ pub async fn create_dashboard( } Err(_) => { if folder_id == DEFAULT_FOLDER { - let folder = meta::dashboards::Folder { + let folder = Folder { folder_id: DEFAULT_FOLDER.to_string(), name: DEFAULT_FOLDER.to_string(), description: DEFAULT_FOLDER.to_string(), @@ -50,7 +52,7 @@ pub async fn create_dashboard( save_dashboard(org_id, &dashboard_id, folder_id, body).await } else { Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( - StatusCode::NOT_FOUND.into(), + http::StatusCode::NOT_FOUND.into(), "folder not found".to_string(), ))) } @@ -71,8 +73,6 @@ pub async fn update_dashboard( #[tracing::instrument] pub async fn list_dashboards(org_id: &str, folder_id: &str) -> Result { - use meta::dashboards::Dashboards; - Ok(HttpResponse::Ok().json(Dashboards { dashboards: dashboards::list(org_id, folder_id).await.unwrap(), })) @@ -98,10 +98,27 @@ pub async fn delete_dashboard( dashboard_id: &str, folder_id: &str, ) -> Result { - if dashboards::get(org_id, dashboard_id, folder_id).await.is_ok() && dashboards::delete(org_id, dashboard_id, folder_id).await.is_ok() { - return Ok(Response::OkMessage("Dashboard deleted".to_owned()).into()); + if dashboards::get(org_id, dashboard_id, folder_id) + .await + .is_err() + { + return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( + http::StatusCode::NOT_FOUND.into(), + "Dashboard not found".to_string(), + ))); + } + match dashboards::delete(org_id, dashboard_id, folder_id).await { + Ok(_) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( + http::StatusCode::OK.into(), + "Dashboard deleted".to_string(), + ))), + Err(error) => Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + error.to_string(), + )), + ), } - Ok(Response::NotFound("Dashboard not found".to_string()).into()) } async fn save_dashboard( @@ -163,16 +180,20 @@ enum Response { impl From for HttpResponse { fn from(resp: Response) -> Self { match resp { - Response::OkMessage(message) => { - Self::Ok().json(MetaHttpResponse::message(StatusCode::OK.into(), message)) - } + Response::OkMessage(message) => Self::Ok().json(MetaHttpResponse::message( + http::StatusCode::OK.into(), + message, + )), Response::NotFound(entity) => Self::NotFound().json(MetaHttpResponse::error( - StatusCode::NOT_FOUND.into(), + http::StatusCode::NOT_FOUND.into(), format!("{entity} not found"), )), - Response::InternalServerError(err) => Self::InternalServerError().json( - MetaHttpResponse::error(StatusCode::INTERNAL_SERVER_ERROR.into(), err.to_string()), - ), + Response::InternalServerError(err) => { + Self::InternalServerError().json(MetaHttpResponse::error( + http::StatusCode::INTERNAL_SERVER_ERROR.into(), + err.to_string(), + )) + } } } } diff --git a/src/service/db/alerts/destinations.rs b/src/service/db/alerts/destinations.rs index 51d1d05837..0492bc6fa3 100644 --- a/src/service/db/alerts/destinations.rs +++ b/src/service/db/alerts/destinations.rs @@ -21,29 +21,19 @@ use crate::common::{ }; use crate::service::db; -pub async fn get( - org_id: &str, - name: &str, -) -> Result, anyhow::Error> { +pub async fn get(org_id: &str, name: &str) -> Result { let map_key = format!("{org_id}/{name}"); - let value: Option = if ALERTS_DESTINATIONS.contains_key(&map_key) { - let dest = ALERTS_DESTINATIONS.get(&map_key).unwrap().clone(); - let template = db::alerts::templates::get(org_id, &dest.template).await?; - Some(dest.to_dest_resp(template)) - } else { - let db = &infra_db::DEFAULT; - let key = format!("/destinations/{org_id}/{name}"); - match db.get(&key).await { - Ok(val) => { - let dest: AlertDestination = json::from_slice(&val).unwrap(); - let template = db::alerts::templates::get(org_id, &dest.template).await?; - Some(dest.to_dest_resp(template)) - } - Err(_) => None, - } - }; + if let Some(val) = ALERTS_DESTINATIONS.get(&map_key) { + let template = db::alerts::templates::get(org_id, &val.template).await?; + return Ok(val.to_dest_resp(Some(template))); + } - Ok(value) + let db = &infra_db::DEFAULT; + let key = format!("/destinations/{org_id}/{name}"); + let val = db.get(&key).await?; + let dest: AlertDestination = json::from_slice(&val).unwrap(); + let template = db::alerts::templates::get(org_id, &dest.template).await?; + Ok(dest.to_dest_resp(Some(template))) } pub async fn set( @@ -76,7 +66,7 @@ pub async fn list(org_id: &str) -> Result, anyhow: for item_value in db.list_values(&key).await? { let dest: AlertDestination = json::from_slice(&item_value).unwrap(); let template = db::alerts::templates::get(org_id, &dest.template).await?; - temp_list.push(dest.to_dest_resp(template)) + temp_list.push(dest.to_dest_resp(Some(template))) } Ok(temp_list) } diff --git a/src/service/db/alerts/mod.rs b/src/service/db/alerts/mod.rs index cf4ff93f36..1cfaf40287 100644 --- a/src/service/db/alerts/mod.rs +++ b/src/service/db/alerts/mod.rs @@ -45,7 +45,11 @@ pub async fn get( Err(_) => None, } }; - Ok(value) + if value.is_none() { + Err(anyhow::anyhow!("Alert not found")) + } else { + Ok(value) + } } pub async fn set( @@ -142,7 +146,10 @@ pub async fn watch() -> Result<(), anyhow::Error> { let alert_key = item_key[0..item_key.rfind('/').unwrap()].to_string(); let item_name = item_key[item_key.rfind('/').unwrap() + 1..].to_string(); if alert_key.contains('/') { - let mut group = STREAM_ALERTS.get(&alert_key.to_string()).unwrap().clone(); + let mut group = match STREAM_ALERTS.get(&alert_key) { + Some(v) => v.clone(), + None => continue, + }; group.list.retain(|trans| !trans.name.eq(&item_name)); STREAM_ALERTS.insert(alert_key.to_string(), group); } else { diff --git a/src/service/db/alerts/templates.rs b/src/service/db/alerts/templates.rs index 2532ffd1c4..9780e5f5ab 100644 --- a/src/service/db/alerts/templates.rs +++ b/src/service/db/alerts/templates.rs @@ -20,31 +20,23 @@ use crate::common::{ utils::json, }; -pub async fn get(org_id: &str, name: &str) -> Result, anyhow::Error> { +pub async fn get(org_id: &str, name: &str) -> Result { let map_key = format!("{org_id}/{name}"); + if let Some(v) = ALERTS_TEMPLATES.get(&map_key) { + return Ok(v.value().clone()); + } let default_org_key = format!("{DEFAULT_ORG}/{name}"); - let value: Option = if ALERTS_TEMPLATES.contains_key(&map_key) - || ALERTS_TEMPLATES.contains_key(&default_org_key) - { - match ALERTS_TEMPLATES.get(&map_key) { - Some(template) => Some(template.clone()), - None => Some(ALERTS_TEMPLATES.get(&default_org_key).unwrap().clone()), - } - } else { - let db = &infra_db::DEFAULT; - let key = format!("/templates/{org_id}/{name}"); - match db.get(&key).await { - Ok(val) => json::from_slice(&val).unwrap(), - Err(_) => { - let key = format!("/templates/{DEFAULT_ORG}/{name}"); - match db.get(&key).await { - Ok(val) => json::from_slice(&val).unwrap(), - Err(_) => None, - } - } - } - }; - Ok(value) + if let Some(v) = ALERTS_TEMPLATES.get(&default_org_key) { + return Ok(v.value().clone()); + } + + let db = &infra_db::DEFAULT; + let key = format!("/templates/{org_id}/{name}"); + if let Ok(val) = db.get(&key).await { + return Ok(json::from_slice(&val).unwrap()); + } + let key = format!("/templates/{DEFAULT_ORG}/{name}"); + Ok(json::from_slice(&db.get(&key).await?).unwrap()) } pub async fn set( diff --git a/src/service/functions.rs b/src/service/functions.rs index 228bea926f..613468f051 100644 --- a/src/service/functions.rs +++ b/src/service/functions.rs @@ -173,9 +173,9 @@ pub async fn delete_function(org_id: String, fn_name: String) -> Result Result { if let Some(val) = STREAM_FUNCTIONS.get(&format!("{}/{}/{}", org_id, stream_type, stream_name)) { @@ -187,12 +187,12 @@ pub async fn list_stream_functions( #[tracing::instrument] pub async fn delete_stream_function( - org_id: String, + org_id: &str, stream_type: StreamType, - stream_name: String, - fn_name: String, + stream_name: &str, + fn_name: &str, ) -> Result { - let mut existing_fn = match check_existing_fn(&org_id, &fn_name).await { + let mut existing_fn = match check_existing_fn(org_id, fn_name).await { Some(function) => function, None => { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( @@ -212,7 +212,7 @@ pub async fn delete_stream_function( .collect::>(), ); } - if let Err(error) = db::functions::set(&org_id, &fn_name, &existing_fn).await { + if let Err(error) = db::functions::set(org_id, fn_name, &existing_fn).await { Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::message( http::StatusCode::INTERNAL_SERVER_ERROR.into(), @@ -222,7 +222,7 @@ pub async fn delete_stream_function( } else { // cant be removed from watcher of function as stream name & type wont be available , hence being removed here let key = format!("{}/{}/{}", org_id, stream_type, stream_name); - remove_stream_fn_from_cache(key, fn_name); + remove_stream_fn_from_cache(&key, fn_name); Ok(HttpResponse::Ok().json(MetaHttpResponse::message( http::StatusCode::OK.into(), FN_REMOVED.to_string(), @@ -238,13 +238,13 @@ pub async fn delete_stream_function( #[tracing::instrument] pub async fn add_function_to_stream( - org_id: String, + org_id: &str, stream_type: StreamType, - stream_name: String, - fn_name: String, + stream_name: &str, + fn_name: &str, mut stream_order: StreamOrder, ) -> Result { - let mut existing_fn = match check_existing_fn(&org_id, &fn_name).await { + let mut existing_fn = match check_existing_fn(org_id, fn_name).await { Some(function) => function, None => { return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error( @@ -254,7 +254,7 @@ pub async fn add_function_to_stream( } }; - stream_order.stream = stream_name; + stream_order.stream = stream_name.to_owned(); stream_order.stream_type = stream_type; if let Some(mut val) = existing_fn.streams { @@ -264,7 +264,7 @@ pub async fn add_function_to_stream( existing_fn.streams = Some(vec![stream_order]); } - if let Err(error) = db::functions::set(&org_id, &fn_name, &existing_fn).await { + if let Err(error) = db::functions::set(org_id, fn_name, &existing_fn).await { Ok( HttpResponse::InternalServerError().json(MetaHttpResponse::message( http::StatusCode::INTERNAL_SERVER_ERROR.into(), @@ -303,8 +303,8 @@ async fn check_existing_fn(org_id: &str, fn_name: &str) -> Option { } } -fn remove_stream_fn_from_cache(key: String, fn_name: String) { - if let Some(val) = STREAM_FUNCTIONS.clone().get(&key) { +fn remove_stream_fn_from_cache(key: &str, fn_name: &str) { + if let Some(val) = STREAM_FUNCTIONS.clone().get(key) { if val.list.len() > 1 { let final_list = val .clone() @@ -312,9 +312,9 @@ fn remove_stream_fn_from_cache(key: String, fn_name: String) { .into_iter() .filter(|x| x.transform.name != fn_name) .collect::>(); - STREAM_FUNCTIONS.insert(key, StreamFunctionsList { list: final_list }); + STREAM_FUNCTIONS.insert(key.to_string(), StreamFunctionsList { list: final_list }); } else { - STREAM_FUNCTIONS.remove(&key); + STREAM_FUNCTIONS.remove(key); } } } diff --git a/src/service/triggers.rs b/src/service/triggers.rs index 8b4aa5f8b8..3ffe7c7727 100644 --- a/src/service/triggers.rs +++ b/src/service/triggers.rs @@ -76,10 +76,10 @@ mod tests { assert!(resp.is_ok()); let resp = crate::service::alerts::get_alert( - "dummy".to_string(), - "TestStream".to_string(), + "dummy", + "TestStream", crate::common::meta::StreamType::Logs, - "TestAlert".to_string(), + "TestAlert", ) .await;