From 03b525322dfdab5a2bca23defabe6e3a6a395f6e Mon Sep 17 00:00:00 2001 From: Hengfei Yang Date: Thu, 8 Oct 2026 09:01:04 +0000 Subject: [PATCH] fix(alerts): key alert dedup state by org so one org's state cannot suppress another's (#15169) --- src/config/src/config.rs | 3 +- src/core/src/alerts/deduplication.rs | 61 +++- src/core/src/alerts/scheduler/handlers.rs | 12 +- .../src/table/entity/alert_dedup_state.rs | 7 +- ...7_000001_scope_alert_dedup_state_by_org.rs | 325 ++++++++++++++++++ src/infra/src/table/migration/mod.rs | 3 + 6 files changed, 397 insertions(+), 14 deletions(-) create mode 100644 src/infra/src/table/migration/m20261007_000001_scope_alert_dedup_state_by_org.rs diff --git a/src/config/src/config.rs b/src/config/src/config.rs index 0fefd3d0df..9ef9eefbd9 100644 --- a/src/config/src/config.rs +++ b/src/config/src/config.rs @@ -94,7 +94,8 @@ pub type RwBTreeMap = tokio::sync::RwLock>; // 95: add band settings to anomaly_detection_config. // 96: create rum_pa_named_events, rum_pa_funnels and rum_pa_tombstones. // 97: create query_history. -pub const DB_SCHEMA_VERSION: u64 = 97; +// 98: key alert_dedup_state by (org_id, fingerprint). +pub const DB_SCHEMA_VERSION: u64 = 98; pub const DB_SCHEMA_KEY: &str = "/db_schema_version/"; // global version variables diff --git a/src/core/src/alerts/deduplication.rs b/src/core/src/alerts/deduplication.rs index 2b35660486..2386f10776 100644 --- a/src/core/src/alerts/deduplication.rs +++ b/src/core/src/alerts/deduplication.rs @@ -156,6 +156,7 @@ pub struct DeduplicationOutcome { /// through — no retry bookkeeping required. pub async fn confirm_notification_sent( db: &DatabaseConnection, + org_id: &str, fingerprints: &[String], ) -> Result<(), sea_orm::DbErr> { if fingerprints.is_empty() { @@ -166,6 +167,7 @@ pub async fn confirm_notification_sent( alert_dedup_state::Column::NotificationSent, sea_orm::sea_query::Expr::value(true), ) + .filter(alert_dedup_state::Column::OrgId.eq(org_id)) .filter(alert_dedup_state::Column::Fingerprint.is_in(fingerprints.to_vec())) .exec(db) .await?; @@ -175,9 +177,10 @@ pub async fn confirm_notification_sent( /// Get or create deduplication state pub async fn get_dedup_state( db: &DatabaseConnection, + org_id: &str, fingerprint: &str, ) -> Result, sea_orm::DbErr> { - alert_dedup_state::Entity::find_by_id(fingerprint) + alert_dedup_state::Entity::find_by_id((org_id.to_string(), fingerprint.to_string())) .one(db) .await } @@ -198,7 +201,7 @@ pub async fn save_dedup_state( params: DedupStateParams<'_>, ) -> Result { // Try to find existing record - if let Some(existing) = get_dedup_state(db, params.fingerprint).await? { + if let Some(existing) = get_dedup_state(db, params.org_id, params.fingerprint).await? { // Update existing let mut active: alert_dedup_state::ActiveModel = existing.clone().into(); active.last_seen_at = Set(params.last_seen_at); @@ -250,6 +253,7 @@ pub async fn cleanup_expired_state( /// Used to suppress alerts that share semantic dimensions with recently fired alerts. pub async fn find_matching_semantic_fingerprints( db: &DatabaseConnection, + org_id: &str, semantic_dimensions: &std::collections::HashMap, time_window_minutes: i64, ) -> Result, sea_orm::DbErr> { @@ -258,6 +262,7 @@ pub async fn find_matching_semantic_fingerprints( // Get all recent fingerprints let recent_states = alert_dedup_state::Entity::find() + .filter(alert_dedup_state::Column::OrgId.eq(org_id)) .filter(alert_dedup_state::Column::LastSeenAt.gt(cutoff_time)) .all(db) .await?; @@ -370,7 +375,7 @@ async fn apply_deduplication_impl( // error swallows the page for the whole window. `notification_sent` // has existed on this table since the feature shipped and was never // read; it is the confirm flag. - let should_send = match get_dedup_state(db, &fingerprint).await? { + let should_send = match get_dedup_state(db, org_id, &fingerprint).await? { Some(existing_state) if config::meta::alerts::deduplication::reservation_suppresses( existing_state.notification_sent, @@ -479,9 +484,27 @@ mod tests { use config::meta::alerts::{ AggFunction, Aggregation, Condition, Operator, QueryType, alert::Alert, }; + use sea_orm::{ConnectionTrait, Database, Schema}; use serde_json::{Map, Value, json}; - use super::row_group_key; + use super::{ + DedupStateParams, confirm_notification_sent, get_dedup_state, row_group_key, + save_dedup_state, + }; + + async fn dedup_db() -> sea_orm::DatabaseConnection { + let db = Database::connect("sqlite::memory:").await.unwrap(); + for sql in [ + "CREATE TABLE alerts (id CHAR(27) PRIMARY KEY)", + "INSERT INTO alerts (id) VALUES ('alert_a'), ('alert_b')", + ] { + db.execute_unprepared(sql).await.unwrap(); + } + let backend = db.get_database_backend(); + let table = Schema::new(backend).create_table_from_entity(super::alert_dedup_state::Entity); + db.execute(backend.build(&table)).await.unwrap(); + db + } fn row(pairs: &[(&str, Value)]) -> Map { pairs @@ -600,4 +623,34 @@ mod tests { alert.query_condition.query_type = QueryType::Slo; assert_eq!(row_group_key(&alert, &row(&[("group", json!("g"))])), None); } + + #[tokio::test] + async fn test_dedup_state_is_scoped_by_org() { + let db = dedup_db().await; + let params = |org_id, alert_id| DedupStateParams { + org_id, + fingerprint: "fp", + alert_id, + first_seen_at: 1, + last_seen_at: 1, + occurrence_count: 1, + }; + save_dedup_state(&db, params("org_a", "alert_a")) + .await + .unwrap(); + assert!(get_dedup_state(&db, "org_b", "fp").await.unwrap().is_none()); + + save_dedup_state(&db, params("org_b", "alert_b")) + .await + .unwrap(); + confirm_notification_sent(&db, "org_a", &["fp".to_string()]) + .await + .unwrap(); + + let org_a = get_dedup_state(&db, "org_a", "fp").await.unwrap().unwrap(); + let org_b = get_dedup_state(&db, "org_b", "fp").await.unwrap().unwrap(); + assert!(org_a.notification_sent); + assert!(!org_b.notification_sent); + assert_eq!(org_b.alert_id, "alert_b"); + } } diff --git a/src/core/src/alerts/scheduler/handlers.rs b/src/core/src/alerts/scheduler/handlers.rs index 7c624e1cb4..afa7382b01 100644 --- a/src/core/src/alerts/scheduler/handlers.rs +++ b/src/core/src/alerts/scheduler/handlers.rs @@ -565,21 +565,21 @@ async fn dispatch_per_group( /// exactly what allows the next evaluation through. Best-effort: failing to /// confirm costs at most one duplicate notification, while failing the /// evaluation over it would cost the alert. -async fn confirm_dedup_reservations(fingerprints: &[String], delivered: bool) { - if !delivered || fingerprints.is_empty() { +async fn confirm_dedup_reservations(org_id: &str, fingerprints: &[String]) { + if fingerprints.is_empty() { return; } #[cfg(feature = "enterprise")] { let db = infra::db::get_orm_client_rw().await; if let Err(e) = - crate::alerts::deduplication::confirm_notification_sent(db, fingerprints).await + crate::alerts::deduplication::confirm_notification_sent(db, org_id, fingerprints).await { log::warn!("[SCHEDULER] could not confirm dedup reservations: {e}"); } } #[cfg(not(feature = "enterprise"))] - let _ = fingerprints; + let _ = (org_id, fingerprints); } pub async fn handle_triggers( @@ -3363,7 +3363,7 @@ async fn handle_alert_triggers( }) .map(|(_, fingerprint)| fingerprint.clone()) .collect(); - confirm_dedup_reservations(&confirmable, !confirmable.is_empty()).await; + confirm_dedup_reservations(&alert.org_id, &confirmable).await; record_delivery(&mut trigger_data); episode_delivered = true; trigger_data.period_end_time = if should_store_last_end_time { @@ -3412,7 +3412,7 @@ async fn handle_alert_triggers( .iter() .map(|(_, fingerprint)| fingerprint.clone()) .collect(); - confirm_dedup_reservations(&fingerprints, true).await; + confirm_dedup_reservations(&alert.org_id, &fingerprints).await; let success_msg = success_msg.trim().to_owned(); let err_msg = err_msg.trim().to_owned(); if !err_msg.is_empty() { diff --git a/src/infra/src/table/entity/alert_dedup_state.rs b/src/infra/src/table/entity/alert_dedup_state.rs index 422ae2e9a6..4b3a26ff87 100644 --- a/src/infra/src/table/entity/alert_dedup_state.rs +++ b/src/infra/src/table/entity/alert_dedup_state.rs @@ -20,13 +20,14 @@ use sea_orm::entity::prelude::*; #[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] #[sea_orm(table_name = "alert_dedup_state")] pub struct Model { - /// Fingerprint hash (primary key) + /// Part of the key because the fingerprint itself carries no org. + #[sea_orm(primary_key, auto_increment = false)] + pub org_id: String, + /// Fingerprint hash #[sea_orm(primary_key, auto_increment = false)] pub fingerprint: String, /// Alert ID reference pub alert_id: String, - /// Organization ID - pub org_id: String, /// First time this fingerprint was seen (microseconds) pub first_seen_at: i64, /// Last time this fingerprint was seen (microseconds) diff --git a/src/infra/src/table/migration/m20261007_000001_scope_alert_dedup_state_by_org.rs b/src/infra/src/table/migration/m20261007_000001_scope_alert_dedup_state_by_org.rs new file mode 100644 index 0000000000..c0f6ad3923 --- /dev/null +++ b/src/infra/src/table/migration/m20261007_000001_scope_alert_dedup_state_by_org.rs @@ -0,0 +1,325 @@ +// Copyright 2026 OpenObserve Inc. +// +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU Affero General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// This program is distributed in the hope that it will be useful +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Affero General Public License for more details. +// +// You should have received a copy of the GNU Affero General Public License +// along with this program. If not, see . + +//! Keys `alert_dedup_state` by `(org_id, fingerprint)`: the fingerprint carries no org. + +use sea_orm::{ConnectionTrait, DbBackend, TransactionTrait}; +use sea_orm_migration::prelude::*; + +const OLD_TABLE: &str = "alert_dedup_state_old"; +const ORG_INDEX: &str = "idx_alert_dedup_org_id"; +const COLUMNS: &str = "fingerprint, alert_id, org_id, first_seen_at, last_seen_at, \ + occurrence_count, notification_sent, created_at"; + +#[derive(DeriveMigrationName)] +pub struct Migration; + +#[async_trait::async_trait] +impl MigrationTrait for Migration { + async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> { + set_primary_key( + manager, + &[AlertDedupState::OrgId, AlertDedupState::Fingerprint], + ) + .await?; + // The key now leads with org_id, so this index only adds write cost. + manager + .drop_index( + Index::drop() + .if_exists() + .name(ORG_INDEX) + .table(AlertDedupState::Table) + .to_owned(), + ) + .await + } + + async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> { + // Orgs may now share a fingerprint; the rows are a short-lived cache, so drop them all. + manager + .get_connection() + .execute_unprepared("DELETE FROM alert_dedup_state") + .await?; + set_primary_key(manager, &[AlertDedupState::Fingerprint]).await?; + manager + .create_index( + Index::create() + .if_not_exists() + .name(ORG_INDEX) + .table(AlertDedupState::Table) + .col(AlertDedupState::OrgId) + .to_owned(), + ) + .await + } +} + +#[derive(DeriveIden, Clone, Copy)] +enum AlertDedupState { + Table, + Fingerprint, + AlertId, + OrgId, + FirstSeenAt, + LastSeenAt, + OccurrenceCount, + NotificationSent, + CreatedAt, +} + +#[derive(DeriveIden)] +enum Alerts { + Table, + Id, +} + +async fn set_primary_key( + manager: &SchemaManager<'_>, + key: &[AlertDedupState], +) -> Result<(), DbErr> { + if manager.get_database_backend() == DbBackend::Postgres { + let columns = key + .iter() + .map(|column| column.to_string()) + .collect::>() + .join(", "); + manager + .get_connection() + .execute_unprepared(&format!( + "ALTER TABLE alert_dedup_state DROP CONSTRAINT alert_dedup_state_pkey, \ + ADD PRIMARY KEY ({columns})" + )) + .await?; + return Ok(()); + } + + // SQLite can't alter a PK; rebuild in a transaction, which the migrator gives only Postgres. + let txn = manager.get_connection().begin().await?; + let schema = SchemaManager::new(&txn); + schema + .rename_table( + Table::rename() + .table(AlertDedupState::Table, Alias::new(OLD_TABLE)) + .to_owned(), + ) + .await?; + schema.create_table(create_table(key)).await?; + txn.execute_unprepared(&format!( + "INSERT INTO alert_dedup_state ({COLUMNS}) SELECT {COLUMNS} FROM {OLD_TABLE}" + )) + .await?; + schema + .drop_table(Table::drop().table(Alias::new(OLD_TABLE)).to_owned()) + .await?; + for (name, column) in [ + ("idx_alert_dedup_alert_id", AlertDedupState::AlertId), + ("idx_alert_dedup_last_seen", AlertDedupState::LastSeenAt), + ] { + schema + .create_index( + Index::create() + .name(name) + .table(AlertDedupState::Table) + .col(column) + .to_owned(), + ) + .await?; + } + txn.commit().await +} + +fn create_table(key: &[AlertDedupState]) -> TableCreateStatement { + let mut primary_key = Index::create(); + for column in key { + primary_key.col(*column); + } + Table::create() + .table(AlertDedupState::Table) + .col( + ColumnDef::new(AlertDedupState::Fingerprint) + .string_len(64) + .not_null(), + ) + .col( + ColumnDef::new(AlertDedupState::AlertId) + .char_len(27) + .not_null(), + ) + .col( + ColumnDef::new(AlertDedupState::OrgId) + .string_len(100) + .not_null(), + ) + .col( + ColumnDef::new(AlertDedupState::FirstSeenAt) + .big_integer() + .not_null(), + ) + .col( + ColumnDef::new(AlertDedupState::LastSeenAt) + .big_integer() + .not_null(), + ) + .col( + ColumnDef::new(AlertDedupState::OccurrenceCount) + .big_integer() + .not_null() + .default(1), + ) + .col( + ColumnDef::new(AlertDedupState::NotificationSent) + .boolean() + .not_null() + .default(false), + ) + .col( + ColumnDef::new(AlertDedupState::CreatedAt) + .big_integer() + .not_null(), + ) + .primary_key(&mut primary_key) + .foreign_key( + ForeignKey::create() + .name("fk_alert_dedup_alert") + .from(AlertDedupState::Table, AlertDedupState::AlertId) + .to(Alerts::Table, Alerts::Id) + .on_delete(ForeignKeyAction::Cascade), + ) + .to_owned() +} + +#[cfg(test)] +mod tests { + use sea_orm::{Database, DatabaseConnection, EntityTrait, Statement}; + + use super::*; + use crate::table::entity::alert_dedup_state; + + type ColumnInfo = (String, String, i32, Option); + + async fn db_with_original_table() -> DatabaseConnection { + let db = Database::connect("sqlite::memory:").await.unwrap(); + for sql in [ + "CREATE TABLE alerts (id CHAR(27) PRIMARY KEY)", + "INSERT INTO alerts (id) VALUES ('alert_a'), ('alert_b')", + ] { + db.execute_unprepared(sql).await.unwrap(); + } + super::super::m20251024_000001_add_alert_deduplication::Migration + .up(&SchemaManager::new(&db)) + .await + .unwrap(); + db + } + + async fn insert(db: &DatabaseConnection, org: &str, alert: &str) -> Result<(), DbErr> { + db.execute_unprepared(&format!( + "INSERT INTO alert_dedup_state ({COLUMNS}) \ + VALUES ('fp', '{alert}', '{org}', 1, 2, 3, true, 4)" + )) + .await + .map(|_| ()) + } + + async fn columns(db: &DatabaseConnection) -> Vec { + db.query_all(Statement::from_string( + DbBackend::Sqlite, + "SELECT name, type, \"notnull\", dflt_value \ + FROM pragma_table_info('alert_dedup_state') ORDER BY cid", + )) + .await + .unwrap() + .iter() + .map(|row| { + ( + row.try_get_by_index(0).unwrap(), + row.try_get_by_index(1).unwrap(), + row.try_get_by_index(2).unwrap(), + row.try_get_by_index(3).unwrap(), + ) + }) + .collect() + } + + #[tokio::test] + async fn test_up_lets_two_orgs_share_a_fingerprint() { + let db = db_with_original_table().await; + insert(&db, "org_a", "alert_a").await.unwrap(); + assert!(insert(&db, "org_b", "alert_b").await.is_err()); + + Migration.up(&SchemaManager::new(&db)).await.unwrap(); + + insert(&db, "org_b", "alert_b").await.unwrap(); + assert!(insert(&db, "org_a", "alert_a").await.is_err()); + let kept = alert_dedup_state::Entity::find_by_id(("org_a".to_string(), "fp".to_string())) + .one(&db) + .await + .unwrap() + .unwrap(); + assert_eq!( + ( + kept.alert_id.as_str(), + kept.first_seen_at, + kept.last_seen_at, + kept.occurrence_count, + kept.notification_sent, + kept.created_at + ), + ("alert_a", 1, 2, 3, true, 4) + ); + } + + #[tokio::test] + async fn test_up_keeps_columns_indexes_and_foreign_key() { + let db = db_with_original_table().await; + let manager = SchemaManager::new(&db); + let original = columns(&db).await; + + Migration.up(&manager).await.unwrap(); + + assert_eq!(columns(&db).await, original); + for index in ["idx_alert_dedup_alert_id", "idx_alert_dedup_last_seen"] { + assert!(manager.has_index("alert_dedup_state", index).await.unwrap()); + } + assert!( + !manager + .has_index("alert_dedup_state", ORG_INDEX) + .await + .unwrap() + ); + assert!(!manager.has_table(OLD_TABLE).await.unwrap()); + assert!(insert(&db, "org_a", "no_such_alert").await.is_err()); + } + + #[tokio::test] + async fn test_down_restores_fingerprint_key() { + let db = db_with_original_table().await; + let manager = SchemaManager::new(&db); + Migration.up(&manager).await.unwrap(); + insert(&db, "org_a", "alert_a").await.unwrap(); + insert(&db, "org_b", "alert_b").await.unwrap(); + + Migration.down(&manager).await.unwrap(); + + insert(&db, "org_a", "alert_a").await.unwrap(); + assert!(insert(&db, "org_b", "alert_b").await.is_err()); + assert!( + manager + .has_index("alert_dedup_state", ORG_INDEX) + .await + .unwrap() + ); + } +} diff --git a/src/infra/src/table/migration/mod.rs b/src/infra/src/table/migration/mod.rs index c2483e55fa..857a1e2b61 100644 --- a/src/infra/src/table/migration/mod.rs +++ b/src/infra/src/table/migration/mod.rs @@ -208,6 +208,7 @@ mod m20260928_000001_add_alert_recovery_destinations; mod m20261001_000001_add_anomaly_band_settings; mod m20261003_000001_create_rum_pa_tables; mod m20261004_000001_create_query_history; +mod m20261007_000001_scope_alert_dedup_state_by_org; /// Shared body of the two `folder_id` migrations above; not a migration itself. mod workflow_folder_id; @@ -519,6 +520,7 @@ impl MigratorTrait for Migrator { Box::new(m20261001_000001_add_anomaly_band_settings::Migration), Box::new(m20261003_000001_create_rum_pa_tables::Migration), Box::new(m20261004_000001_create_query_history::Migration), + Box::new(m20261007_000001_scope_alert_dedup_state_by_org::Migration), ] } } @@ -575,6 +577,7 @@ mod tests { (95, "m20261001_000001_add_anomaly_band_settings"), (96, "m20261003_000001_create_rum_pa_tables"), (97, "m20261004_000001_create_query_history"), + (98, "m20261007_000001_scope_alert_dedup_state_by_org"), ]; #[test]