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

feat: delete API should return 200, 404 or 500 (#1924)

Browse Source
This commit is contained in:
Hengfei Yang
2023-11-10 21:38:34 +08:00
committed by GitHub
parent dd49bf0a91
commit ca52eca9d8
17 changed files with 360 additions and 431 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/utils/notification.rs
+60 -61
View File
@@ -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(())
src/handler/http/request/alerts/destinations.rs
+10 -12
View File
@@ -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}")]
src/handler/http/request/alerts/mod.rs
+48 -78
View File
@@ -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::<HashMap<String, String>>::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::<HashMap<String, String>>::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<String>, req: HttpRequest) -> impl Responder {
let org_id = path.into_inner();
let query = web::Query::<HashMap<String, String>>::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<String>, 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::<HashMap<String, String>>::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::<HashMap<String, String>>::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::<HashMap<String, String>>::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
}
src/handler/http/request/alerts/templates.rs
+1
View File
@@ -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}")]
src/handler/http/request/dashboards/folders.rs
+3 -2
View File
@@ -142,8 +142,9 @@ pub async fn get_folder(path: web::Path<(String, String)>) -> Result<HttpRespons
("folder_id" = String, Path, description = "Folder ID"),
),
responses(
(status = StatusCode::OK, description = "Folder deleted", body = HttpResponse),
(status = StatusCode::NOT_FOUND, description = "Folder 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}/folders/{folder_id}")]
src/handler/http/request/dashboards/mod.rs
+3 -2
View File
@@ -147,8 +147,9 @@ async fn get_dashboard(path: web::Path<(String, String)>, 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}")]
src/handler/http/request/functions/mod.rs
+13 -28
View File
@@ -142,8 +142,8 @@ async fn list_stream_functions(
) -> Result<HttpResponse, Error> {
let (org_id, stream_name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::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<HttpResponse, Error> {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::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<HttpResponse, Error> {
let (org_id, stream_name, name) = path.into_inner();
let query = web::Query::<HashMap<String, String>>::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
src/service/alerts/destinations.rs
+23 -25
View File
@@ -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<HttpResp
}
}
if db::alerts::destinations::get(org_id.as_str(), name.as_str()).await.is_ok() {
let result = db::alerts::destinations::delete(org_id.as_str(), name.as_str()).await;
return match result {
Ok(_) => {
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]
src/service/alerts/mod.rs
+52 -104
View File
@@ -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<HttpResponse, Error> {
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<StreamType>,
) -> Result<HttpResponse, Error> {
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<HttpResponse, Error> {
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<HttpResponse, Error> {
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<HttpResponse, Error> {
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());
}
}
src/service/alerts/templates.rs
+21 -12
View File
@@ -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<HttpRespons
}
}
if db::alerts::templates::get(org_id.as_str(), name.as_str()).await.is_ok() && db::alerts::templates::delete(org_id.as_str(), name.as_str()).await.is_ok() {
return Ok(HttpResponse::Ok().json(MetaHttpResponse::message(
http::StatusCode::OK.into(),
"Alert template deleted".to_string(),
if db::alerts::templates::get(org_id.as_str(), name.as_str())
.await
.is_err()
{
return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error(
http::StatusCode::NOT_FOUND.into(),
"Alert template not found".to_string(),
)));
}
Ok(HttpResponse::NotFound().json(MetaHttpResponse::error(
http::StatusCode::NOT_FOUND.into(),
"Alert template not found".to_string(),
)))
match db::alerts::templates::delete(org_id.as_str(), name.as_str()).await {
Ok(_) => 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]
src/service/dashboards/folders.rs
+28 -19
View File
@@ -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<HttpResponse, E
HttpResponse::Ok().json(folder)
} else {
return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error(
StatusCode::NOT_FOUND.into(),
"folder not found".to_string(),
http::StatusCode::NOT_FOUND.into(),
"Dashboard folder not found".to_string(),
)));
};
Ok(resp)
@@ -112,18 +108,31 @@ pub async fn delete_folder(org_id: &str, folder_id: &str) -> Result<HttpResponse
let dashboards = db::dashboards::list(org_id, folder_id).await.unwrap();
if !dashboards.is_empty() {
return Ok(HttpResponse::BadRequest().json(MetaHttpResponse::error(
StatusCode::BAD_REQUEST.into(),
"folder contains dashboards, please move/delete dashboards from folder ".to_string(),
http::StatusCode::BAD_REQUEST.into(),
"Dashboard folder contains dashboards, please move/delete dashboards from folder "
.to_string(),
)));
}
let resp = if let Ok(folder) = db::dashboards::folders::delete(org_id, folder_id).await {
HttpResponse::Ok().json(folder)
} else {
if db::dashboards::folders::get(org_id, folder_id)
.await
.is_err()
{
return Ok(HttpResponse::NotFound().json(MetaHttpResponse::error(
StatusCode::NOT_FOUND.into(),
"folder not found".to_string(),
http::StatusCode::NOT_FOUND.into(),
"Dashboard folder not found".to_string(),
)));
};
Ok(resp)
}
match db::dashboards::folders::delete(org_id, folder_id).await {
Ok(_) => 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(),
)),
),
}
}
src/service/dashboards/mod.rs
+41 -20
View File
@@ -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<HttpResponse, io::Error> {
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<HttpResponse, io::Error> {
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<Response> 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(),
))
}
}
}
}
src/service/db/alerts/destinations.rs
+12 -22
View File
@@ -21,29 +21,19 @@ use crate::common::{
};
use crate::service::db;
pub async fn get(
org_id: &str,
name: &str,
) -> Result<Option<AlertDestinationResponse>, anyhow::Error> {
pub async fn get(org_id: &str, name: &str) -> Result<AlertDestinationResponse, anyhow::Error> {
let map_key = format!("{org_id}/{name}");
let value: Option<AlertDestinationResponse> = 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<Vec<AlertDestinationResponse>, 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)
}
src/service/db/alerts/mod.rs
+9 -2
View File
@@ -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 {
src/service/db/alerts/templates.rs
+15 -23
View File
@@ -20,31 +20,23 @@ use crate::common::{
utils::json,
};
pub async fn get(org_id: &str, name: &str) -> Result<Option<DestinationTemplate>, anyhow::Error> {
pub async fn get(org_id: &str, name: &str) -> Result<DestinationTemplate, anyhow::Error> {
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<DestinationTemplate> = 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(
src/service/functions.rs
+18 -18
View File
@@ -173,9 +173,9 @@ pub async fn delete_function(org_id: String, fn_name: String) -> Result<HttpResp
#[tracing::instrument]
pub async fn list_stream_functions(
org_id: String,
org_id: &str,
stream_type: StreamType,
stream_name: String,
stream_name: &str,
) -> Result<HttpResponse, Error> {
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<HttpResponse, Error> {
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::<Vec<StreamOrder>>(),
);
}
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<HttpResponse, Error> {
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<Transform> {
}
}
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::<Vec<StreamTransform>>();
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);
}
}
}
src/service/triggers.rs
+3 -3
View File
@@ -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;