From e07a67b0d0f2dfe420fde8872bfc61e73bd7dd32 Mon Sep 17 00:00:00 2001 From: Taiming Liu Date: Sat, 19 Jul 2025 20:14:21 -0700 Subject: [PATCH] fix: not overwrite pipeline_id when imported (#7179) --- src/handler/http/request/alerts/deprecated.rs | 10 +++++-- src/handler/http/request/alerts/mod.rs | 8 +++--- src/handler/http/request/dashboards/mod.rs | 14 +++++++--- src/handler/http/request/pipeline.rs | 26 +++++++++---------- src/service/alerts/alert.rs | 23 +++++++++++----- src/service/db/alerts/alert.rs | 3 ++- src/service/pipeline/mod.rs | 6 +++++ 7 files changed, 61 insertions(+), 29 deletions(-) diff --git a/src/handler/http/request/alerts/deprecated.rs b/src/handler/http/request/alerts/deprecated.rs index f854616c73..4e26fed2e7 100644 --- a/src/handler/http/request/alerts/deprecated.rs +++ b/src/handler/http/request/alerts/deprecated.rs @@ -63,8 +63,14 @@ pub async fn save_alert( path: web::Path<(String, String)>, alert: web::Json, user_email: UserEmail, + req: HttpRequest, ) -> HttpResponse { let (org_id, stream_name) = path.into_inner(); + let query = web::Query::>::from_query(req.query_string()).unwrap(); + let overwrite = match query.get("overwrite") { + Some(v) => v.parse::().unwrap_or_default(), + None => false, + }; // Hack for frequency: convert minutes to seconds let mut alert = alert.into_inner(); @@ -77,7 +83,7 @@ pub async fn save_alert( alert.set_last_satisfied_at(None); alert.set_last_triggered_at(None); - match alert::save(&org_id, &stream_name, "", alert, true).await { + match alert::save(&org_id, &stream_name, "", alert, true, overwrite).await { Ok(_) => MetaHttpResponse::ok("Alert saved"), Err(e) => e.into(), } @@ -118,7 +124,7 @@ pub async fn update_alert( alert.trigger_condition.frequency *= 60; alert.last_edited_by = Some(user_email.user_id); alert.updated_at = Some(datetime_now()); - match alert::save(&org_id, &stream_name, &name, alert, false).await { + match alert::save(&org_id, &stream_name, &name, alert, false, false).await { Ok(_) => MetaHttpResponse::ok("Alert Updated"), Err(e) => e.into(), } diff --git a/src/handler/http/request/alerts/mod.rs b/src/handler/http/request/alerts/mod.rs index 65faf87956..ce09cdd1d5 100644 --- a/src/handler/http/request/alerts/mod.rs +++ b/src/handler/http/request/alerts/mod.rs @@ -32,7 +32,7 @@ use crate::{ }, responses::{EnableAlertResponseBody, GetAlertResponseBody, ListAlertsResponseBody}, }, - request::dashboards::get_folder, + request::dashboards::{get_folder, is_overwrite}, }, service::{ alerts::alert::{self, AlertError}, @@ -111,7 +111,9 @@ pub async fn create_alert( let org_id = path.into_inner(); let req_body = req_body.into_inner(); - let folder_id = get_folder(req); + let query_str = req.query_string(); + let folder_id = get_folder(query_str); + let overwrite = is_overwrite(query_str); let mut alert: MetaAlert = req_body.into(); if alert.owner.clone().filter(|o| !o.is_empty()).is_none() { alert.owner = Some(user_email.user_id.clone()); @@ -119,7 +121,7 @@ pub async fn create_alert( alert.last_edited_by = Some(user_email.user_id); let client = ORM_CLIENT.get_or_init(connect_to_orm).await; - match alert::create(client, &org_id, &folder_id, alert).await { + match alert::create(client, &org_id, &folder_id, alert, overwrite).await { Ok(v) => MetaHttpResponse::json( MetaHttpResponse::message(StatusCode::OK, "Alert saved") .with_id(v.id.map(|id| id.to_string()).unwrap_or_default()) diff --git a/src/handler/http/request/dashboards/mod.rs b/src/handler/http/request/dashboards/mod.rs index a53dac5668..84b7126041 100644 --- a/src/handler/http/request/dashboards/mod.rs +++ b/src/handler/http/request/dashboards/mod.rs @@ -102,7 +102,7 @@ pub async fn create_dashboard( user_email: UserEmail, ) -> impl Responder { let org_id = path.into_inner(); - let folder = get_folder(req); + let folder = get_folder(req.query_string()); let mut dashboard: config::meta::dashboards::Dashboard = match req_body.into_inner().try_into() { Ok(dashboard) => dashboard, @@ -372,11 +372,19 @@ async fn move_dashboards( } } -pub fn get_folder(req: HttpRequest) -> String { - let query = web::Query::>::from_query(req.query_string()).unwrap(); +pub fn get_folder(query_str: &str) -> String { + let query = web::Query::>::from_query(query_str).unwrap(); crate::common::utils::http::get_folder(&query) } +pub fn is_overwrite(query_str: &str) -> bool { + let query = web::Query::>::from_query(query_str).unwrap(); + match query.get("overwrite") { + Some(v) => v.parse::().unwrap_or_default(), + None => false, + } +} + /// Tries to get the user ID from the request headers. fn get_user_id(req: HttpRequest) -> Option { req.headers() diff --git a/src/handler/http/request/pipeline.rs b/src/handler/http/request/pipeline.rs index 3295187de0..836d180d90 100644 --- a/src/handler/http/request/pipeline.rs +++ b/src/handler/http/request/pipeline.rs @@ -59,13 +59,20 @@ impl From for HttpResponse { pub async fn save_pipeline( path: web::Path, pipeline: web::Json, - _req: HttpRequest, + req: HttpRequest, ) -> Result { let org_id = path.into_inner(); let mut pipeline = pipeline.into_inner(); pipeline.name = pipeline.name.trim().to_lowercase(); pipeline.org = org_id; - pipeline.id = ider::generate(); + let query = web::Query::>::from_query(req.query_string()).unwrap(); + let overwrite = match query.get("overwrite") { + Some(v) => v.parse::().unwrap_or_default(), + None => false, + }; + if !overwrite { + pipeline.id = ider::generate(); + } match pipeline::save_pipeline(pipeline).await { Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( http::StatusCode::OK, @@ -158,10 +165,7 @@ async fn list_pipelines( ) )] #[get("/{org_id}/pipelines/streams")] -async fn list_streams_with_pipeline( - path: web::Path, - _req: HttpRequest, -) -> Result { +async fn list_streams_with_pipeline(path: web::Path) -> Result { let org_id = path.into_inner(); match pipeline::list_streams_with_pipeline(&org_id).await { Ok(stream_params) => Ok(HttpResponse::Ok().json(stream_params)), @@ -189,10 +193,7 @@ async fn list_streams_with_pipeline( ) )] #[delete("/{org_id}/pipelines/{pipeline_id}")] -async fn delete_pipeline( - path: web::Path<(String, String)>, - _req: HttpRequest, -) -> Result { +async fn delete_pipeline(path: web::Path<(String, String)>) -> Result { let (_org_id, pipeline_id) = path.into_inner(); match pipeline::delete_pipeline(&pipeline_id).await { Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( @@ -223,10 +224,7 @@ async fn delete_pipeline( ) )] #[put("/{org_id}/pipelines")] -pub async fn update_pipeline( - pipeline: web::Json, - _req: HttpRequest, -) -> Result { +pub async fn update_pipeline(pipeline: web::Json) -> Result { let pipeline = pipeline.into_inner(); match pipeline::update_pipeline(pipeline).await { Ok(()) => Ok(HttpResponse::Ok().json(MetaHttpResponse::message( diff --git a/src/service/alerts/alert.rs b/src/service/alerts/alert.rs index 6c07211566..dd1ed3f77f 100644 --- a/src/service/alerts/alert.rs +++ b/src/service/alerts/alert.rs @@ -190,6 +190,7 @@ pub async fn save( name: &str, mut alert: Alert, create: bool, + overwrite: bool, ) -> Result<(), AlertError> { // Currently all alerts are stored in the default folder so create the // default folder for the org if it doesn't exist yet. @@ -197,7 +198,7 @@ pub async fn save( create_default_alerts_folder(org_id).await?; }; - prepare_alert(org_id, stream_name, name, &mut alert, create).await?; + prepare_alert(org_id, stream_name, name, &mut alert, create, overwrite).await?; // save the alert // TODO: Get the folder id @@ -239,6 +240,7 @@ async fn prepare_alert( name: &str, alert: &mut Alert, create: bool, + overwrite: bool, ) -> Result<(), AlertError> { if !name.is_empty() { alert.name = name.to_string(); @@ -261,7 +263,7 @@ async fn prepare_alert( if let Some(alert_id) = alert.id { match get_by_id_db(org_id, alert_id).await { Ok(old_alert) => { - if create { + if create && !overwrite { return Err(AlertError::CreateAlreadyExists); } alert.owner = old_alert.owner; @@ -459,6 +461,7 @@ pub async fn create( org_id: &str, folder_id: &str, mut alert: Alert, + overwrite: bool, ) -> Result { if !table::folders::exists(org_id, folder_id, FolderType::Alerts).await? { if folder_id == DEFAULT_FOLDER { @@ -470,9 +473,17 @@ pub async fn create( let alert_name = alert.name.clone(); let stream_name = alert.stream_name.clone(); - prepare_alert(org_id, &stream_name, &alert_name, &mut alert, true).await?; + prepare_alert( + org_id, + &stream_name, + &alert_name, + &mut alert, + true, + overwrite, + ) + .await?; - let alert = db::alerts::alert::create(conn, org_id, folder_id, alert).await?; + let alert = db::alerts::alert::create(conn, org_id, folder_id, alert, overwrite).await?; set_ownership( org_id, @@ -578,7 +589,7 @@ pub async fn update( let alert_name = alert.name.clone(); let stream_name = alert.stream_name.clone(); - prepare_alert(org_id, &stream_name, &alert_name, &mut alert, false).await?; + prepare_alert(org_id, &stream_name, &alert_name, &mut alert, false, false).await?; let alert = db::alerts::alert::update(conn, org_id, dst_folder_id_info, alert).await?; #[cfg(feature = "enterprise")] @@ -1840,7 +1851,7 @@ mod tests { let alert_name = "abc/alert"; let mut alert: Alert = Default::default(); alert.name = alert_name.to_string(); - let ret = save(org_id, stream_name, alert_name, alert, true).await; + let ret = save(org_id, stream_name, alert_name, alert, true, false).await; // alert name should not contain / assert!(ret.is_err()); } diff --git a/src/service/db/alerts/alert.rs b/src/service/db/alerts/alert.rs index dd352da7c2..5293672bb6 100644 --- a/src/service/db/alerts/alert.rs +++ b/src/service/db/alerts/alert.rs @@ -150,8 +150,9 @@ pub async fn create( org_id: &str, folder_id: &str, alert: Alert, + overwrite: bool, ) -> Result { - let alert = table::create(conn, org_id, folder_id, alert, false).await?; + let alert = table::create(conn, org_id, folder_id, alert, overwrite).await?; let schedule_key = scheduler_key(alert.id); cluster::emit_put_event(org_id, &alert, Some(folder_id.to_string())).await?; diff --git a/src/service/pipeline/mod.rs b/src/service/pipeline/mod.rs index 3843ad8b44..de968e7d2c 100644 --- a/src/service/pipeline/mod.rs +++ b/src/service/pipeline/mod.rs @@ -33,6 +33,12 @@ pub mod batch_execution; #[tracing::instrument(skip(pipeline))] pub async fn save_pipeline(mut pipeline: Pipeline) -> Result<(), PipelineError> { + // check if id is missing + if pipeline.id.is_empty() { + return Err(PipelineError::InvalidPipeline( + "Missing pipeline ID".to_string(), + )); + } // check if another realtime pipeline with the same source stream already exists if let PipelineSource::Realtime(stream) = &pipeline.source && pipeline::list_streams_with_pipeline(&pipeline.org)