diff --git a/src/alerts/mod.rs b/src/alerts/mod.rs index 6e9422af1..e38c1ad54 100644 --- a/src/alerts/mod.rs +++ b/src/alerts/mod.rs @@ -1214,6 +1214,10 @@ pub enum AlertError { FromStrError(#[from] FromStrError), #[error("Invalid Target ID- {0}")] InvalidTargetID(String), + #[error("Target title {0:?} already exists")] + DuplicateTargetName(String), + #[error("Alert title {0:?} already exists")] + DuplicateAlertTitle(String), #[error("Invalid target modification request: {0}")] InvalidTargetModification(String), #[error("Can't delete a Target which is being used")] @@ -1237,6 +1241,7 @@ pub enum AlertError { } impl actix_web::ResponseError for AlertError { + // Keep every error-to-status mapping explicit. skipcq: RS-R1000 fn status_code(&self) -> StatusCode { match self { Self::ObjectStorage(_) => StatusCode::INTERNAL_SERVER_ERROR, @@ -1252,6 +1257,8 @@ impl actix_web::ResponseError for AlertError { Self::InvalidAlertModifyRequest => StatusCode::BAD_REQUEST, Self::FromStrError(_) => StatusCode::BAD_REQUEST, Self::InvalidTargetID(_) => StatusCode::BAD_REQUEST, + Self::DuplicateTargetName(_) => StatusCode::CONFLICT, + Self::DuplicateAlertTitle(_) => StatusCode::CONFLICT, Self::InvalidTargetModification(_) => StatusCode::BAD_REQUEST, Self::TargetInUse => StatusCode::CONFLICT, Self::ParserError(_) => StatusCode::BAD_REQUEST, diff --git a/src/alerts/target.rs b/src/alerts/target.rs index 8a4851d4f..cb864a30f 100644 --- a/src/alerts/target.rs +++ b/src/alerts/target.rs @@ -29,7 +29,7 @@ use http::header::AUTHORIZATION; use itertools::Itertools; use once_cell::sync::Lazy; use serde_json::{Value, json}; -use tokio::sync::RwLock; +use tokio::sync::{Mutex as AsyncMutex, RwLock}; use tracing::{error, trace, warn}; use ulid::Ulid; use url::Url; @@ -47,11 +47,13 @@ use crate::{ pub static TARGETS: Lazy = Lazy::new(|| TargetConfigs { target_configs: RwLock::new(HashMap::new()), + mutation_lock: AsyncMutex::new(()), }); #[derive(Debug)] pub struct TargetConfigs { pub target_configs: RwLock>>, + mutation_lock: AsyncMutex<()>, } impl TargetConfigs { @@ -73,17 +75,32 @@ impl TargetConfigs { Ok(()) } - pub async fn update(&self, target: Target) -> Result<(), AlertError> { + pub async fn update(&self, target: Target, is_create: bool) -> Result<(), AlertError> { + let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); + let _mutation_guard = self.mutation_lock.lock().await; + if is_create && let Some(targets) = self.target_configs.read().await.get(tenant_id) { + if targets + .values() + .any(|existing| existing.name == target.name) + { + return Err(AlertError::DuplicateTargetName(target.name)); + } + if targets.contains_key(&target.id) { + return Err(AlertError::InvalidTargetModification( + "Target ID already exists".to_string(), + )); + } + } PARSEABLE .metastore .put_target(&target, &target.tenant) .await?; - let mut map = self.target_configs.write().await; - let tenant_id = target.tenant.as_deref().unwrap_or(DEFAULT_TENANT); - map.entry(tenant_id.to_owned()) + self.target_configs + .write() + .await + .entry(tenant_id.to_owned()) .or_default() .insert(target.id, target); - // map.insert(target.id, target.clone()); Ok(()) } @@ -121,6 +138,7 @@ impl TargetConfigs { tenant_id: &Option, ) -> Result { let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); + let _mutation_guard = self.mutation_lock.lock().await; // ensure that the target is not being used by any alert let guard = ALERTS.read().await; let alerts = if let Some(alerts) = guard.as_ref() { @@ -134,9 +152,10 @@ impl TargetConfigs { return Err(AlertError::TargetInUse); } } - let target = if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { + let target = if let Some(targets) = self.target_configs.read().await.get(tenant) { targets - .remove(target_id) + .get(target_id) + .cloned() .ok_or(AlertError::InvalidTargetID(target_id.to_string()))? } else { return Err(AlertError::InvalidTargetID(target_id.to_string())); @@ -145,6 +164,9 @@ impl TargetConfigs { .metastore .delete_target(&target, tenant_id) .await?; + if let Some(targets) = self.target_configs.write().await.get_mut(tenant) { + targets.remove(target_id); + } Ok(target) } } diff --git a/src/handlers/http/alerts.rs b/src/handlers/http/alerts.rs index 0e544ce6b..a952b6405 100644 --- a/src/handlers/http/alerts.rs +++ b/src/handlers/http/alerts.rs @@ -41,6 +41,7 @@ use actix_web::{ web::{self, Json, Path}, }; use chrono::{DateTime, Utc}; +use tokio::sync::Mutex; use ulid::Ulid; // Reserved query parameter names that are not treated as other_fields filters @@ -48,6 +49,9 @@ const RESERVED_PARAMS: [&str; 3] = ["tags", "offset", "limit"]; const MAX_LIMIT: usize = 1000; const DEFAULT_LIMIT: usize = 100; +// Serialize alert creation, updates, and deletion in this process. +static ALERT_MUTATION_LOCK: Mutex<()> = Mutex::const_new(()); + /// Query parameters for listing alerts struct ListQueryParams { tags_list: Vec, @@ -317,17 +321,30 @@ pub async fn post_internal( } }; - let guard = ALERTS.write().await; - let alerts = if let Some(alerts) = guard.as_ref() { - alerts - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; + let alerts = { + let guard = ALERTS.read().await; + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; // validate the incoming alert query // does the user have access to these tables or not? alert.validate(session_key).await?; + if alerts + .get_all_alerts(tenant_id) + .await + .values() + .any(|existing| existing.get_title() == alert.get_title()) + { + return Err(AlertError::DuplicateAlertTitle( + alert.get_title().to_owned(), + )); + } + // update persistent storage first PARSEABLE .metastore @@ -396,11 +413,13 @@ pub async fn delete_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { - let guard = ALERTS.write().await; - let alerts = if let Some(alerts) = guard.as_ref() { - alerts - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; + let alerts = { + let guard = ALERTS.read().await; + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; let alert = alerts.get_alert_by_id(alert_id, tenant_id).await?; @@ -465,6 +484,7 @@ pub async fn update_notification_state( } }; + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -506,6 +526,7 @@ pub async fn disable_alert_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -547,6 +568,7 @@ pub async fn enable_alert_internal( alert_id: Ulid, tenant_id: &Option, ) -> Result { + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let guard = ALERTS.write().await; let alerts = if let Some(alerts) = guard.as_ref() { alerts @@ -603,14 +625,13 @@ pub async fn modify_alert_internal( alert_request: AlertRequest, tenant_id: &Option, ) -> Result { - // Get alerts manager reference without holding the global lock + let _mutation_guard = ALERT_MUTATION_LOCK.lock().await; let alerts = { let guard = ALERTS.read().await; - if let Some(alerts) = guard.as_ref() { - alerts.clone() - } else { - return Err(AlertError::CustomError("No AlertManager set".into())); - } + guard + .as_ref() + .cloned() + .ok_or_else(|| AlertError::CustomError("No AlertManager set".into()))? }; // Validate and prepare the new alert @@ -659,6 +680,20 @@ pub async fn modify_alert_internal( new_alert.validate(session_key).await?; + if new_alert.get_title() != alert.get_title() + && alerts + .get_all_alerts(tenant_id) + .await + .values() + .any(|existing| { + existing.get_id() != &alert_id && existing.get_title() == new_alert.get_title() + }) + { + return Err(AlertError::DuplicateAlertTitle( + new_alert.get_title().to_owned(), + )); + } + // Perform I/O operations PARSEABLE .metastore diff --git a/src/handlers/http/targets.rs b/src/handlers/http/targets.rs index 286bc1ff8..7c4a02633 100644 --- a/src/handlers/http/targets.rs +++ b/src/handlers/http/targets.rs @@ -48,10 +48,13 @@ pub async fn post_internal( tenant_id: &Option, ) -> Result { target.tenant.clone_from(tenant_id); + if target.name.trim().is_empty() { + return Err(AlertError::ValidationFailure( + "Target title is required".into(), + )); + } target.validate_outbound_policy().await?; - // should check for duplicacy and liveness (??) - // add to the map - TARGETS.update(target.clone()).await?; + TARGETS.update(target.clone(), true).await?; Ok(serde_json::to_value(target.mask())?) } @@ -165,9 +168,7 @@ pub async fn update_internal( target.id = target_id; target.tenant.clone_from(tenant_id); target.validate_outbound_policy().await?; - // should check for duplicacy and liveness (??) - // add to the map - TARGETS.update(target.clone()).await?; + TARGETS.update(target.clone(), false).await?; Ok(serde_json::to_value(target.mask())?) }