From 44011fbebcaca0a4058c76e2ffc3b095a8dcb226 Mon Sep 17 00:00:00 2001 From: sktrpathi <42913098+ByteBaker@users.noreply.github.com> Date: Wed, 26 Nov 2025 12:09:10 +0530 Subject: [PATCH] feat: add organization-level alert deduplication with semantic field groups (#9209) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary Implements org-level alert deduplication with semantic field groups for intelligent alert suppression and batched notifications. This feature prevents alert storms by deduplicating alerts based on configurable fingerprints and grouping related alerts together. ### Core Capabilities - **Semantic Field Groups**: Map field name variations (`hostname`, `host`, `node`) to canonical dimensions for consistent deduplication across different data sources - **Organization-Level Configuration**: Global deduplication settings with cross-alert support to suppress alerts from different rules sharing semantic dimensions - **Alert Grouping & Batching**: Wait-and-collect logic with three send strategies (`FirstWithCount`, `Summary`, `All`) for batched notifications - **Per-Alert Control**: Individual alert deduplication settings with optional fingerprint field customization ### Features **Deduplication Modes:** - Per-alert exact matching using fingerprint-based TTL expiration - Cross-alert semantic matching using org-defined dimension groups - Configurable time windows (per-alert override or org-level default) **Alert Grouping:** - Batch alerts with same fingerprint before sending notifications - Three strategies: first alert with count, summary with names, or all details - Configurable wait time (1-300 seconds) and max batch size (1-100 alerts) - Background processor runs every 1 second for expired batches **Semantic Field Groups:** - Built-in presets for common fields (host, IP address, service) - Kubernetes resource groups (cluster, namespace, pod, node, container) - Field normalization support (lowercase + trim) - Overlap handling with first-defined precedence ### Configuration Types - `SemanticFieldGroup`: Field name equivalence definitions with normalization - `GlobalDeduplicationConfig`: Org-level settings with semantic groups and cross-alert dedup - `DeduplicationConfig`: Per-alert fingerprint fields and grouping options ### API Endpoints - `GET /{org_id}/alerts/deduplication/config` - Retrieve org-level dedup config - `POST /{org_id}/alerts/deduplication/config` - Set org-level dedup config - `DELETE /{org_id}/alerts/deduplication/config` - Delete org-level dedup config ### Implementation Details - Pure business logic algorithms in enterprise layer (fingerprinting, matching, calculations) - OSS service layer orchestrates database operations with enterprise delegation - Feature-gated with `#[cfg(feature = "enterprise")]` for enterprise-only functionality - Background job architecture for async batch processing - Storage: Org config in KV DB at `/alert_config/{org_id}/deduplication` ### Architecture ``` Alert Triggered ↓ Query Evaluation ↓ Deduplication Check (fingerprint + TTL) ↓ Grouping Enabled? → Add to Batch → Wait/Collect → Send Grouped ↓ No Send Individual Notification ``` --------- Co-authored-by: Ashish Kolhe Co-authored-by: Shrinath Rao --- .github/workflows/playwright.yml | 4 +- src/config/src/meta/alerts/deduplication.rs | 586 +++++++++++++++++- src/config/src/meta/self_reporting/mod.rs | 20 + src/config/src/meta/self_reporting/usage.rs | 54 ++ src/config/src/metrics.rs | 136 ++++ src/config/src/utils/flatten.rs | 79 ++- .../http/request/alerts/dedup_stats.rs | 154 +++++ .../http/request/alerts/deduplication.rs | 203 ++++++ src/handler/http/request/alerts/history.rs | 38 ++ src/handler/http/request/alerts/mod.rs | 2 + src/handler/http/router/mod.rs | 7 + src/job/alert_grouping.rs | 256 ++++++++ src/job/alert_manager.rs | 3 +- src/job/mod.rs | 4 + src/service/alerts/alert.rs | 7 +- src/service/alerts/deduplication.rs | 127 +++- src/service/alerts/grouping.rs | 249 ++++++++ src/service/alerts/mod.rs | 4 + src/service/alerts/org_config.rs | 81 +++ src/service/alerts/scheduler/handlers.rs | 156 ++++- src/service/ingestion/mod.rs | 2 +- src/service/self_reporting/queues.rs | 5 + src/service/self_reporting/triggers_schema.rs | 6 +- .../ui-testing/pages/generalPages/homePage.js | 4 +- web/src/components/alerts/AlertHistory.vue | 54 ++ web/src/components/alerts/AlertList.vue | 96 ++- .../components/alerts/DedupSummaryCards.vue | 158 +++++ .../OrganizationDeduplicationSettings.vue | 334 ++++++++++ .../alerts/SemanticFieldGroupsConfig.vue | 398 ++++++++++++ .../components/alerts/SemanticGroupItem.vue | 210 +++++++ web/src/components/alerts/TagInput.vue | 194 ++++++ web/src/locales/languages/en.json | 1 + web/src/services/alerts.ts | 12 + 33 files changed, 3541 insertions(+), 103 deletions(-) create mode 100644 src/handler/http/request/alerts/dedup_stats.rs create mode 100644 src/handler/http/request/alerts/deduplication.rs create mode 100644 src/job/alert_grouping.rs create mode 100644 src/service/alerts/grouping.rs create mode 100644 src/service/alerts/org_config.rs create mode 100644 web/src/components/alerts/DedupSummaryCards.vue create mode 100644 web/src/components/alerts/OrganizationDeduplicationSettings.vue create mode 100644 web/src/components/alerts/SemanticFieldGroupsConfig.vue create mode 100644 web/src/components/alerts/SemanticGroupItem.vue create mode 100644 web/src/components/alerts/TagInput.vue diff --git a/.github/workflows/playwright.yml b/.github/workflows/playwright.yml index ada7117a1a..3a82331984 100644 --- a/.github/workflows/playwright.yml +++ b/.github/workflows/playwright.yml @@ -327,8 +327,8 @@ jobs: # Try to get last failed tests from TestDino # Redirect stderr to /dev/null to avoid capturing npm warnings if LAST_FAILED_FLAGS="$(npx tdpw last-failed --cache-id "$CACHE_ID" 2>/dev/null)"; then - # Trim whitespace (original working logic - no filtering) - LAST_FAILED_FLAGS="$(echo "$LAST_FAILED_FLAGS" | xargs)" + # Trim whitespace using parameter expansion (safer than xargs for quoted strings) + LAST_FAILED_FLAGS="$(echo "$LAST_FAILED_FLAGS" | sed -e 's/^[[:space:]]*//' -e 's/[[:space:]]*$//')" if [ -n "$LAST_FAILED_FLAGS" ] && [ "$LAST_FAILED_FLAGS" != "No failed tests found" ]; then echo "::notice::✅ Found failed tests from previous run." diff --git a/src/config/src/meta/alerts/deduplication.rs b/src/config/src/meta/alerts/deduplication.rs index ed6fbb54f7..ae80df049a 100644 --- a/src/config/src/meta/alerts/deduplication.rs +++ b/src/config/src/meta/alerts/deduplication.rs @@ -13,9 +13,216 @@ // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . +use itertools::Itertools; use serde::{Deserialize, Serialize}; use utoipa::ToSchema; +/// Semantic field group - defines field names that represent the same entity +/// +/// # Field Name Overlaps +/// +/// Field names can appear in multiple semantic groups. When extracting dimensions, +/// the **first-defined group** takes precedence. This allows flexible configurations +/// where a field like "service" can have different meanings in different contexts. +/// +/// Example: +/// ``` +/// // Group 1: service-primary (defined first) +/// // field_names: ["service", "primary_service"] +/// // +/// // Group 2: service-backup (defined second) +/// // field_names: ["service", "backup_service"] +/// // +/// // When extracting dimensions from {"service": "api"}: +/// // → Matches "service-primary" group (first occurrence wins) +/// ``` +#[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq)] +pub struct SemanticFieldGroup { + /// Unique identifier (lowercase, dash-separated, e.g., "host", "k8s-cluster") + pub id: String, + + /// Human-readable display name (e.g., "Host", "K8s Cluster") + pub display: String, + + /// List of field names that are equivalent (e.g., ["host", "hostname", "node"]) + /// + /// Note: Field names can overlap with other groups. First-defined group wins. + pub fields: Vec, + + /// Whether to normalize values (lowercase + trim) + #[serde(default)] + pub normalize: bool, +} + +impl SemanticFieldGroup { + pub fn new( + id: impl Into, + display: impl Into, + fields: &[&str], + normalize: bool, + ) -> Self { + let fields = fields.iter().map(|v| v.to_string()).collect_vec(); + + Self { + id: id.into(), + display: display.into(), + fields, + normalize, + } + } + /// Get default semantic groups for common use cases + pub fn default_presets() -> Vec { + vec![ + Self::new( + "host", + "Host", + &["host", "hostname", "node", "node_name"], + true, + ), + Self::new( + "ip-address", + "IP Address", + &[ + "ip", + "ipaddr", + "ip_address", + "ip_addr", + "client_ip", + "source_ip", + "host_ip", + ], + true, + ), + Self::new( + "service", + "Service", + &[ + "service", + "service_name", + "svc", + "app", + "application", + "app_name", + ], + true, + ), + // Kubernetes groups + Self::new( + "k8s-cluster", + "K8s Cluster", + &["k8s_cluster", "cluster", "cluster_name", "cluster_id"], + false, + ), + Self::new( + "k8s-namespace", + "K8s Namespace", + &["k8s_namespace", "namespace", "k8s_ns", "ns"], + false, + ), + Self::new( + "k8s-pod", + "K8s Pod", + &["k8s_pod", "pod", "pod_name", "pod_id"], + false, + ), + Self::new( + "k8s-node", + "K8s Node", + &["k8s_node", "node", "node_name", "kubernetes_node"], + false, + ), + Self::new( + "k8s-container", + "K8s Container", + &[ + "container", + "container_name", + "container_id", + "k8s_container", + ], + false, + ), + ] + } + + /// Validate semantic field group ID format + pub fn validate_id(id: &str) -> bool { + if id.is_empty() { + return false; + } + // Lowercase, alphanumeric, dash-separated + id.chars() + .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-') + && !id.starts_with('-') + && !id.ends_with('-') + && !id.contains("--") + } +} + +/// Organization-level deduplication configuration (Global settings) +/// +/// This configuration is stored at the organization level and defines: +/// - Whether deduplication is enabled globally +/// - Semantic field groups that map field name variations to canonical dimensions +/// - Cross-alert deduplication based on shared semantic dimensions +/// - Default time window for deduplication +/// +/// Per-alert fingerprint fields are stored in the Alert.deduplication field. +#[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq, Default)] +#[serde(default)] +pub struct GlobalDeduplicationConfig { + /// Enable/disable deduplication globally for this organization + #[serde(default)] + pub enabled: bool, + + /// Semantic field groups - defines field name equivalences + /// + /// Defines how field names from different data sources map to canonical dimensions. + /// Example: `["host", "hostname", "server"]` all map to semantic group ID "host". + /// + /// These groups are used for: + /// 1. Mapping per-alert fingerprint_fields to semantic dimensions + /// 2. Cross-alert deduplication (if enabled) + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub semantic_field_groups: Vec, + + /// Enable cross-alert deduplication based on shared semantic dimensions + /// + /// When true, alerts from DIFFERENT alert rules that share semantic dimensions + /// will be deduplicated if one already fired recently. + /// + /// Example: + /// - Alert A fires: {host: "srv01", service: "api"} + /// - Alert B fires 30s later: {host: "srv01", region: "us-east"} + /// - If cross_alert_dedup=true: Alert B suppressed (shares "host" dimension) + /// - If cross_alert_dedup=false: Both sent (per-alert dedup only) + #[serde(default)] + pub alert_dedup_enabled: bool, + + /// Semantic group IDs to use for cross-alert fingerprinting + /// + /// When cross_alert_dedup is enabled, these semantic group IDs are used + /// to generate fingerprints across all alerts (instead of per-alert fingerprint_fields). + /// Must reference IDs from semantic_field_groups. + /// + /// Example: ["host", "service"] means fingerprint = hash(host_value + service_value) + /// Required when cross_alert_dedup is true. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub alert_fingerprint_groups: Vec, + + /// Default time window in minutes for deduplication + /// + /// Alerts with the same fingerprint within this window are suppressed. + /// If None, defaults to 2x the alert evaluation frequency. + /// Can be overridden per-alert. + #[serde(skip_serializing_if = "Option::is_none")] + pub time_window_minutes: Option, +} + +/// Per-alert deduplication configuration (from main branch) +/// +/// This is stored on each Alert and specifies which fields to use for fingerprinting. +/// This matches the structure from the main branch exactly. /// Configuration for alert deduplication #[derive(Clone, Debug, Serialize, Deserialize, ToSchema, PartialEq, Default)] #[serde(default)] @@ -110,12 +317,224 @@ fn default_group_wait() -> i64 { 30 } +impl GlobalDeduplicationConfig { + /// Validate the organization-level deduplication configuration + pub fn validate(&self) -> Result<(), String> { + if !self.enabled { + return Ok(()); + } + + // Validate cross-alert fingerprint groups reference existing semantic groups + if self.alert_dedup_enabled && !self.alert_fingerprint_groups.is_empty() { + let semantic_group_ids: std::collections::HashSet<_> = + self.semantic_field_groups.iter().map(|g| &g.id).collect(); + + for group_id in &self.alert_fingerprint_groups { + if !semantic_group_ids.contains(group_id) { + return Err(format!( + "Cross-alert fingerprint group '{}' not found in semantic_field_groups", + group_id + )); + } + } + } + + // Require at least one fingerprint group when cross-alert dedup is enabled + if self.alert_dedup_enabled && self.alert_fingerprint_groups.is_empty() { + return Err( + "cross_alert_fingerprint_groups is required when cross_alert_dedup is enabled" + .to_string(), + ); + } + + // Validate semantic groups + if self.semantic_field_groups.len() > 50 { + return Err("Maximum 50 semantic groups allowed per organization".to_string()); + } + + let mut seen_ids = std::collections::HashSet::new(); + let mut field_name_to_group: std::collections::HashMap<&String, &String> = + std::collections::HashMap::new(); + + for group in &self.semantic_field_groups { + // Validate ID format + if !SemanticFieldGroup::validate_id(&group.id) { + return Err(format!( + "Invalid semantic group ID '{}': must be lowercase, alphanumeric, dash-separated", + group.id + )); + } + + // Validate ID uniqueness + if !seen_ids.insert(&group.id) { + return Err(format!("Duplicate semantic group ID: {}", group.id)); + } + + // Validate display name + if group.display.is_empty() { + return Err(format!( + "Display name required for semantic group '{}'", + group.id + )); + } + + // Validate field names + if group.fields.is_empty() { + return Err(format!( + "Semantic group '{}' must have at least one field name", + group.id + )); + } + + if group.fields.len() > 20 { + return Err(format!( + "Semantic group '{}' has too many field names (max 20)", + group.id + )); + } + + // Track field name overlaps (warn but allow) + // Precedence: first-defined group wins when extracting dimensions + for field_name in &group.fields { + if let Some(existing_group_id) = field_name_to_group.get(field_name) { + log::warn!( + "[deduplication] Field name '{field_name}' appears in multiple semantic groups: '{existing_group_id}' and '{}'. Using first occurrence (group '{existing_group_id}').", + group.id + ); + } else { + field_name_to_group.insert(field_name, &group.id); + } + } + } + + Ok(()) + } + + /// Get default configuration with common semantic groups + pub fn default_with_presets() -> Self { + Self { + enabled: false, + semantic_field_groups: SemanticFieldGroup::default_presets(), + alert_dedup_enabled: false, + alert_fingerprint_groups: vec![], + time_window_minutes: None, + } + } + + /// Map a field name to its semantic group ID + /// + /// Returns the first matching semantic group ID, or None if no match. + /// Used for reverse lookup: actual field name → semantic dimension. + pub fn get_semantic_group_id(&self, field_name: &str) -> Option<&str> { + for group in &self.semantic_field_groups { + if group.fields.iter().any(|f| f == field_name) { + return Some(&group.id); + } + } + None + } + + /// Extract semantic dimensions from a result row + /// + /// Maps actual field names to semantic group IDs and extracts their values. + /// Returns a map of semantic_group_id → value. + /// + /// Example: + /// - Input fields: ["hostname", "service_name"] + /// - Input row: {"hostname": "srv01", "service_name": "api", "other": "data"} + /// - Output: {"host": "srv01", "service": "api"} + pub fn extract_semantic_dimensions( + &self, + field_names: &[String], + result_row: &std::collections::HashMap, + ) -> std::collections::HashMap { + let mut dimensions = std::collections::HashMap::new(); + + for field_name in field_names { + if let Some(semantic_id) = self.get_semantic_group_id(field_name) { + if let Some(value) = result_row.get(field_name) { + // Find the semantic group to check if normalization is needed + if let Some(group) = self + .semantic_field_groups + .iter() + .find(|g| g.id == semantic_id) + { + let normalized_value = if group.normalize { + value.to_lowercase().trim().to_string() + } else { + value.clone() + }; + dimensions.insert(semantic_id.to_string(), normalized_value); + } + } + } else { + // Field not in semantic groups - use as-is + if let Some(value) = result_row.get(field_name) { + dimensions.insert(field_name.clone(), value.clone()); + } + } + } + + dimensions + } +} + +impl DeduplicationConfig { + /// Validate the per-alert deduplication configuration + pub fn validate(&self) -> Result<(), String> { + if !self.enabled { + return Ok(()); + } + + // Fingerprint fields can be empty (will auto-detect) + if self.fingerprint_fields.len() > 10 { + return Err("Maximum 10 fingerprint fields allowed per alert".to_string()); + } + + // Time window must be positive if set + if let Some(window) = self.time_window_minutes + && window <= 0 + { + return Err("Time window must be positive".to_string()); + } + + // Validate grouping config if present + if let Some(grouping) = &self.grouping + && grouping.enabled + { + if grouping.max_group_size == 0 { + return Err("Max group size must be at least 1".to_string()); + } + if grouping.group_wait_seconds < 0 { + return Err("Group wait seconds must be non-negative".to_string()); + } + } + + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; #[test] - fn test_deduplication_config_default() { + fn test_semantic_field_group_id_validation() { + assert!(SemanticFieldGroup::validate_id("host")); + assert!(SemanticFieldGroup::validate_id("k8s-cluster")); + assert!(SemanticFieldGroup::validate_id("ip-address")); + assert!(SemanticFieldGroup::validate_id("service-123")); + + assert!(!SemanticFieldGroup::validate_id("")); + assert!(!SemanticFieldGroup::validate_id("Host")); + assert!(!SemanticFieldGroup::validate_id("k8s_cluster")); + assert!(!SemanticFieldGroup::validate_id("-host")); + assert!(!SemanticFieldGroup::validate_id("host-")); + assert!(!SemanticFieldGroup::validate_id("host--name")); + } + + #[test] + fn test_per_alert_deduplication_config_default() { let config = DeduplicationConfig::default(); assert!(!config.enabled); assert!(config.fingerprint_fields.is_empty()); @@ -124,31 +543,122 @@ mod tests { } #[test] - fn test_grouping_config_default() { - let config = GroupingConfig::default(); + fn test_organization_deduplication_config_default() { + let config = GlobalDeduplicationConfig::default(); assert!(!config.enabled); - assert_eq!(config.max_group_size, 100); - assert_eq!(config.send_strategy, SendStrategy::FirstWithCount); - assert_eq!(config.group_wait_seconds, 30); + assert!(!config.alert_dedup_enabled); + assert!(config.semantic_field_groups.is_empty()); + assert_eq!(config.time_window_minutes, None); } #[test] - fn test_send_strategy_serialization() { - let strategy = SendStrategy::FirstWithCount; - let json = serde_json::to_string(&strategy).unwrap(); - assert_eq!(json, r#""first_with_count""#); + fn test_organization_deduplication_config_validation() { + let mut config = GlobalDeduplicationConfig { + enabled: true, + alert_dedup_enabled: false, + semantic_field_groups: vec![ + SemanticFieldGroup::new("service", "Service", &["service", "service_name"], true), + SemanticFieldGroup::new("host", "Host", &["host", "hostname"], true), + ], + alert_fingerprint_groups: vec![], + time_window_minutes: Some(10), + }; - let strategy = SendStrategy::Summary; - let json = serde_json::to_string(&strategy).unwrap(); - assert_eq!(json, r#""summary""#); + assert!(config.validate().is_ok()); - let strategy = SendStrategy::All; - let json = serde_json::to_string(&strategy).unwrap(); - assert_eq!(json, r#""all""#); + // Test invalid ID + config.semantic_field_groups[0].id = "Invalid_ID".to_string(); + assert!(config.validate().is_err()); + config.semantic_field_groups[0].id = "service".to_string(); + + // Test duplicate ID + config.semantic_field_groups.push(SemanticFieldGroup::new( + "service", + "Service 2", + &["svc"], + true, + )); + assert!(config.validate().is_err()); + config.semantic_field_groups.pop(); + + // Test overlapping field names - allowed with warning + config.semantic_field_groups[1] + .fields + .push("service".to_string()); + assert!(config.validate().is_ok()); // Should succeed, just warns + config.semantic_field_groups[1].fields.pop(); } #[test] - fn test_deduplication_config_serialization() { + fn test_per_alert_deduplication_config_validation() { + let mut config = DeduplicationConfig { + enabled: true, + fingerprint_fields: vec!["hostname".to_string(), "service".to_string()], + time_window_minutes: Some(10), + grouping: None, + }; + + assert!(config.validate().is_ok()); + + // Test too many fingerprint fields + config.fingerprint_fields = (0..15).map(|i| format!("field{}", i)).collect(); + assert!(config.validate().is_err()); + + // Test negative time window + config.fingerprint_fields = vec!["hostname".to_string()]; + config.time_window_minutes = Some(-1); + assert!(config.validate().is_err()); + } + + #[test] + fn test_organization_config_overlapping_field_names() { + // Test that overlapping field names are allowed (with warning) + // Precedence: first-defined group wins + let config = GlobalDeduplicationConfig { + enabled: true, + alert_dedup_enabled: false, + semantic_field_groups: vec![ + SemanticFieldGroup::new( + "service-primary", + "Primary Service", + &["service", "primary_service"], + true, + ), + SemanticFieldGroup::new( + "service-backup", + "Backup Service", + &["service", "backup_service"], + true, + ), // "service" overlaps + ], + alert_fingerprint_groups: vec![], + time_window_minutes: Some(10), + }; + + // Should succeed - overlaps are allowed, just logged as warnings + assert!(config.validate().is_ok()); + } + + #[test] + fn test_organization_deduplication_config_serialization() { + let config = GlobalDeduplicationConfig { + enabled: true, + alert_dedup_enabled: true, + semantic_field_groups: vec![ + SemanticFieldGroup::new("service", "Service", &["service", "service_name"], true), + SemanticFieldGroup::new("host", "Host", &["host", "hostname"], true), + ], + alert_fingerprint_groups: vec![], + time_window_minutes: Some(10), + }; + + let json = serde_json::to_string_pretty(&config).unwrap(); + let deserialized: GlobalDeduplicationConfig = serde_json::from_str(&json).unwrap(); + assert_eq!(config, deserialized); + } + + #[test] + fn test_per_alert_deduplication_config_serialization() { let config = DeduplicationConfig { enabled: true, fingerprint_fields: vec!["service".to_string(), "hostname".to_string()], @@ -167,7 +677,7 @@ mod tests { } #[test] - fn test_deduplication_config_minimal_serialization() { + fn test_per_alert_config_minimal_serialization() { let config = DeduplicationConfig { enabled: true, ..Default::default() @@ -179,5 +689,45 @@ mod tests { assert!(json.get("fingerprint_fields").is_none()); // time_window_minutes should be omitted when None assert!(json.get("time_window_minutes").is_none()); + // grouping should be omitted when None + assert!(json.get("grouping").is_none()); + } + + #[test] + fn test_organization_config_minimal_serialization() { + let config = GlobalDeduplicationConfig { + enabled: true, + ..Default::default() + }; + + let json = serde_json::to_value(&config).unwrap(); + assert_eq!(json["enabled"], true); + // semantic_field_groups should be omitted when empty + assert!(json.get("semantic_field_groups").is_none()); + // time_window_minutes should be omitted when None + assert!(json.get("time_window_minutes").is_none()); + // alert_dedup_enabled should be false by default + assert_eq!( + json.get("alert_dedup_enabled").and_then(|v| v.as_bool()), + Some(false) + ); + } + + #[test] + fn test_default_presets() { + let presets = SemanticFieldGroup::default_presets(); + assert!(!presets.is_empty()); + + // Check that we have common groups + assert!(presets.iter().any(|g| g.id == "host")); + assert!(presets.iter().any(|g| g.id == "service")); + assert!(presets.iter().any(|g| g.id == "k8s-cluster")); + + // Validate all preset IDs + for preset in &presets { + assert!(SemanticFieldGroup::validate_id(&preset.id)); + assert!(!preset.display.is_empty()); + assert!(!preset.fields.is_empty()); + } } } diff --git a/src/config/src/meta/self_reporting/mod.rs b/src/config/src/meta/self_reporting/mod.rs index 1c3b087c02..393bfb9a5b 100644 --- a/src/config/src/meta/self_reporting/mod.rs +++ b/src/config/src/meta/self_reporting/mod.rs @@ -372,6 +372,11 @@ mod tests { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; let result = queue @@ -893,6 +898,11 @@ mod tests { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; let error_data = error::ErrorData { @@ -956,6 +966,11 @@ mod tests { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; let trigger_data2 = TriggerData { @@ -1009,6 +1024,11 @@ mod tests { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; // Should succeed when queue has space diff --git a/src/config/src/meta/self_reporting/usage.rs b/src/config/src/meta/self_reporting/usage.rs index 83539097f6..e3526e4d23 100644 --- a/src/config/src/meta/self_reporting/usage.rs +++ b/src/config/src/meta/self_reporting/usage.rs @@ -79,6 +79,50 @@ pub struct TriggerData { pub query_took: Option, pub scheduler_trace_id: Option, pub time_in_queue_ms: Option, + // Deduplication tracking fields + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dedup_enabled: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dedup_suppressed: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dedup_count: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub grouped: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub group_size: Option, +} + +impl Default for TriggerData { + fn default() -> Self { + Self { + _timestamp: 0, + org: String::new(), + module: TriggerDataType::Alert, + key: String::new(), + next_run_at: 0, + is_realtime: false, + is_silenced: false, + status: TriggerDataStatus::Completed, + start_time: 0, + end_time: 0, + retries: 0, + skipped_alerts_count: None, + error: None, + success_response: None, + is_partial: None, + delay_in_secs: None, + evaluation_took_in_secs: None, + source_node: None, + query_took: None, + scheduler_trace_id: None, + time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, + } + } } impl TriggerData { @@ -114,6 +158,11 @@ impl TriggerData { query_took: Some(0), scheduler_trace_id: Some(String::new()), time_in_queue_ms: Some(0), + dedup_enabled: Some(false), + dedup_suppressed: Some(false), + dedup_count: Some(0), + grouped: Some(false), + group_size: Some(0), } } @@ -715,6 +764,11 @@ mod tests { query_took: Some(500), scheduler_trace_id: Some("trace123".to_string()), time_in_queue_ms: Some(100), + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; let json = serde_json::to_string(&trigger_data).unwrap(); diff --git a/src/config/src/metrics.rs b/src/config/src/metrics.rs index 51f6757bcf..c337068b3a 100644 --- a/src/config/src/metrics.rs +++ b/src/config/src/metrics.rs @@ -691,6 +691,114 @@ pub static META_NUM_ALERTS: Lazy = Lazy::new(|| { ) .expect("Metric created") }); + +// Alert deduplication metrics +pub static ALERT_DEDUP_SUPPRESSED_TOTAL: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new( + "alert_dedup_suppressed_total", + "Total number of alerts suppressed by deduplication", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization", "alert_name", "dedup_type"], + ) + .expect("Metric created") +}); + +pub static ALERT_DEDUP_PASSED_TOTAL: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new( + "alert_dedup_passed_total", + "Total number of alerts that passed deduplication", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization", "alert_name"], + ) + .expect("Metric created") +}); + +pub static ALERT_DEDUP_ERRORS_TOTAL: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new( + "alert_dedup_errors_total", + "Total deduplication processing errors", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization", "error_type"], + ) + .expect("Metric created") +}); + +// Alert grouping/batching metrics +pub static ALERT_GROUPING_BATCHES_PENDING: Lazy = Lazy::new(|| { + IntGaugeVec::new( + Opts::new( + "alert_grouping_batches_pending", + "Current number of pending alert batches", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization"], + ) + .expect("Metric created") +}); + +pub static ALERT_GROUPING_NOTIFICATIONS_SENT_TOTAL: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new( + "alert_grouping_notifications_sent_total", + "Total number of grouped notifications sent", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization", "send_strategy", "reason"], + ) + .expect("Metric created") +}); + +pub static ALERT_GROUPING_BATCH_SIZE: Lazy = Lazy::new(|| { + HistogramVec::new( + HistogramOpts::new( + "alert_grouping_batch_size", + "Number of alerts per grouped notification", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()) + .buckets(vec![1.0, 2.0, 3.0, 5.0, 10.0, 25.0, 50.0, 100.0]), + &["organization", "send_strategy"], + ) + .expect("Metric created") +}); + +pub static ALERT_GROUPING_WAIT_TIME: Lazy = Lazy::new(|| { + HistogramVec::new( + HistogramOpts::new( + "alert_grouping_wait_time_seconds", + "Time batches waited before sending (seconds)", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()) + .buckets(vec![5.0, 10.0, 15.0, 20.0, 30.0, 45.0, 60.0, 90.0, 120.0]), + &["organization"], + ) + .expect("Metric created") +}); + +pub static ALERT_GROUPING_SEND_ERRORS_TOTAL: Lazy = Lazy::new(|| { + IntCounterVec::new( + Opts::new( + "alert_grouping_send_errors_total", + "Total grouped notification send failures", + ) + .namespace(NAMESPACE) + .const_labels(create_const_labels()), + &["organization", "error_type"], + ) + .expect("Metric created") +}); pub static META_NUM_DASHBOARDS: Lazy = Lazy::new(|| { IntGaugeVec::new( Opts::new("meta_num_dashboards", "Metadata dashboard nums") @@ -1376,6 +1484,34 @@ fn register_metrics(registry: &Registry) { .register(Box::new(META_NUM_DASHBOARDS.clone())) .expect("Metric registered"); + // alert deduplication metrics + registry + .register(Box::new(ALERT_DEDUP_SUPPRESSED_TOTAL.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_DEDUP_PASSED_TOTAL.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_DEDUP_ERRORS_TOTAL.clone())) + .expect("Metric registered"); + + // alert grouping metrics + registry + .register(Box::new(ALERT_GROUPING_BATCHES_PENDING.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_GROUPING_NOTIFICATIONS_SENT_TOTAL.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_GROUPING_BATCH_SIZE.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_GROUPING_WAIT_TIME.clone())) + .expect("Metric registered"); + registry + .register(Box::new(ALERT_GROUPING_SEND_ERRORS_TOTAL.clone())) + .expect("Metric registered"); + // db stats registry .register(Box::new(DB_QUERY_NUMS.clone())) diff --git a/src/config/src/utils/flatten.rs b/src/config/src/utils/flatten.rs index 12b2802547..b5f47253f0 100644 --- a/src/config/src/utils/flatten.rs +++ b/src/config/src/utils/flatten.rs @@ -417,19 +417,62 @@ mod tests { ] }); - let expected_output = json!({ - "firstname": "John", - "lastname": "Doe", - "age": 25, - "address_streetaddress": "123 Main St", - "address_city": "Anytown", - "address_state": "CA", - "address_postalcode": "12345", - "phonenumbers":"[{\"number\":\"555-555-1234\",\"type\":\"home\"},{\"number\":\"555-555-5678\",\"type\":\"work\"}]" - }); - let output = flatten(input).unwrap(); - assert_eq!(output, expected_output); + + // Check all fields except phonenumbers + assert_eq!(output["firstname"], "John"); + assert_eq!(output["lastname"], "Doe"); + assert_eq!(output["age"], 25); + assert_eq!(output["address_streetaddress"], "123 Main St"); + assert_eq!(output["address_city"], "Anytown"); + assert_eq!(output["address_state"], "CA"); + assert_eq!(output["address_postalcode"], "12345"); + + // Parse and compare phonenumbers JSON to handle key ordering + let phonenumbers_str = output["phonenumbers"].as_str().unwrap(); + let phonenumbers: serde_json::Value = serde_json::from_str(phonenumbers_str).unwrap(); + let expected_phonenumbers = json!([ + {"type": "home", "number": "555-555-1234"}, + {"type": "work", "number": "555-555-5678"} + ]); + assert_eq!(phonenumbers, expected_phonenumbers); + } + + fn compare_flattened_json(actual: &serde_json::Value, expected: &serde_json::Value) { + // Helper to compare JSON values, parsing embedded JSON strings + let actual_obj = actual.as_object().unwrap(); + let expected_obj = expected.as_object().unwrap(); + + assert_eq!( + actual_obj.len(), + expected_obj.len(), + "Different number of fields" + ); + + for (key, expected_val) in expected_obj { + let actual_val = actual_obj + .get(key) + .unwrap_or_else(|| panic!("Missing key: {}", key)); + + if let (Some(actual_str), Some(expected_str)) = + (actual_val.as_str(), expected_val.as_str()) + { + // Both are strings, try to parse as JSON + if let (Ok(actual_json), Ok(expected_json)) = ( + serde_json::from_str::(actual_str), + serde_json::from_str::(expected_str), + ) { + // Both strings contain valid JSON, compare them structurally + assert_eq!(actual_json, expected_json, "JSON mismatch for key: {}", key); + } else { + // Not JSON or parse failed, compare as strings + assert_eq!(actual_str, expected_str, "String mismatch for key: {}", key); + } + } else { + // Non-string values, compare directly + assert_eq!(actual_val, expected_val, "Value mismatch for key: {}", key); + } + } } #[test] @@ -512,17 +555,17 @@ mod tests { }); let output = flatten_with_level(input.clone(), 0).unwrap(); - assert_eq!(output, expected_output_level0); + compare_flattened_json(&output, &expected_output_level0); let output = flatten_with_level(input.clone(), 1).unwrap(); - assert_eq!(output, expected_output_level1); + compare_flattened_json(&output, &expected_output_level1); let output = flatten_with_level(input.clone(), 2).unwrap(); - assert_eq!(output, expected_output_level2); + compare_flattened_json(&output, &expected_output_level2); let output = flatten_with_level(input.clone(), 3).unwrap(); - assert_eq!(output, expected_output_level3); + compare_flattened_json(&output, &expected_output_level3); let output = flatten_with_level(input.clone(), 4).unwrap(); - assert_eq!(output, expected_output_level4); + compare_flattened_json(&output, &expected_output_level4); let output = flatten_with_level(input, 5).unwrap(); - assert_eq!(output, expected_output_level4); + compare_flattened_json(&output, &expected_output_level4); } #[test] diff --git a/src/handler/http/request/alerts/dedup_stats.rs b/src/handler/http/request/alerts/dedup_stats.rs new file mode 100644 index 0000000000..052f73e996 --- /dev/null +++ b/src/handler/http/request/alerts/dedup_stats.rs @@ -0,0 +1,154 @@ +// Copyright 2025 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 . + +use actix_web::{HttpResponse, get, web}; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::common::meta::http::HttpResponse as MetaHttpResponse; + +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct DedupSummaryResponse { + /// Total number of alerts in organization + pub total_alerts: usize, + /// Number of alerts with deduplication enabled + pub alerts_with_dedup: usize, + /// Total alerts suppressed in time period + pub suppressions_total: i64, + /// Total alerts that passed dedup in time period + pub passed_total: i64, + /// Suppression rate (0.0 to 1.0) + pub suppression_rate: f64, + /// Current pending batches (grouping) + pub pending_batches: i64, + /// Timestamp when data was collected + pub timestamp: i64, +} + +/// Get deduplication summary statistics for an organization +#[utoipa::path( + context_path = "/api", + tag = "Alerts", + operation_id = "GetDedupSummary", + security( + ("Authorization"= []) + ), + params( + ("org_id" = String, Path, description = "Organization name"), + ), + responses( + (status = 200, description = "Success", body = DedupSummaryResponse), + ) +)] +#[get("/{org_id}/alerts/dedup/summary")] +pub async fn get_dedup_summary( + org_id: web::Path, +) -> Result { + let org_id = org_id.into_inner(); + + // Get all alerts for the organization + let alerts_data = match crate::service::db::alerts::alert::list(&org_id, None, None).await { + Ok(data) => data, + Err(e) => { + log::error!("Failed to list alerts for org {}: {}", org_id, e); + return Ok( + HttpResponse::InternalServerError().json(MetaHttpResponse::error( + actix_web::http::StatusCode::INTERNAL_SERVER_ERROR.as_u16(), + e.to_string(), + )), + ); + } + }; + + let total_alerts = alerts_data.len(); + let alerts_with_dedup: usize = alerts_data + .iter() + .filter(|alert| alert.deduplication.as_ref().is_some_and(|d| d.enabled)) + .count(); + + // Get current pending batches count from in-memory state (enterprise only) + #[cfg(feature = "enterprise")] + let pending_batches = crate::service::alerts::grouping::get_pending_batch_count(&org_id); + + #[cfg(not(feature = "enterprise"))] + let pending_batches = 0; + + // For suppressions/passed counts, we'll use database for now + // In future, could query Prometheus metrics for time-series data + let (suppressions_total, passed_total) = match get_dedup_counts_from_db(&org_id).await { + Ok(counts) => counts, + Err(e) => { + log::warn!("Failed to get dedup counts from database: {}", e); + (0, 0) + } + }; + + let suppression_rate = if suppressions_total + passed_total > 0 { + suppressions_total as f64 / (suppressions_total + passed_total) as f64 + } else { + 0.0 + }; + + let response = DedupSummaryResponse { + total_alerts, + alerts_with_dedup, + suppressions_total, + passed_total, + suppression_rate, + pending_batches, + timestamp: config::utils::time::now_micros(), + }; + + Ok(HttpResponse::Ok().json(response)) +} + +/// Get dedup counts from database +/// Returns (suppressions, passed) counts based on occurrence_count +async fn get_dedup_counts_from_db(_org_id: &str) -> Result<(i64, i64), anyhow::Error> { + #[cfg(feature = "enterprise")] + { + use infra::table::entity::alert_dedup_state; + use sea_orm::{ColumnTrait, EntityTrait, QueryFilter}; + + let db = infra::db::ORM_CLIENT + .get() + .ok_or_else(|| anyhow::anyhow!("Database not initialized"))?; + + // Get all dedup states for this org + let states = alert_dedup_state::Entity::find() + .filter(alert_dedup_state::Column::OrgId.eq(_org_id)) + .all(db) + .await?; + + let mut suppressions = 0i64; + let mut passed = 0i64; + + for state in states { + // If occurrence_count > 1, then (count - 1) were suppressed + if state.occurrence_count > 1 { + suppressions += state.occurrence_count - 1; + } + // Each fingerprint represents 1 passed alert + passed += 1; + } + + Ok((suppressions, passed)) + } + + #[cfg(not(feature = "enterprise"))] + { + Ok((0, 0)) + } +} diff --git a/src/handler/http/request/alerts/deduplication.rs b/src/handler/http/request/alerts/deduplication.rs new file mode 100644 index 0000000000..3e399372d6 --- /dev/null +++ b/src/handler/http/request/alerts/deduplication.rs @@ -0,0 +1,203 @@ +// Copyright 2025 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 . + +//! Alert deduplication API endpoints (Enterprise feature) + +use actix_web::{HttpResponse, delete, get, post, web}; +#[cfg(feature = "enterprise")] +use config::meta::alerts::deduplication::GlobalDeduplicationConfig; + +// ==================== ENTERPRISE IMPLEMENTATIONS ==================== + +/// Get deduplication configuration for an organization +#[cfg(feature = "enterprise")] +#[utoipa::path( + context_path = "/api", + tag = "Alerts", + operation_id = "GetDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + responses( + (status = 200, description = "Success", body = GlobalDeduplicationConfig), + (status = 403, description = "Forbidden - Enterprise feature"), + (status = 404, description = "Not found"), + (status = 500, description = "Internal server error"), + ), +)] +#[get("/{org_id}/alerts/deduplication/config")] +pub async fn get_config(org_id: web::Path) -> Result { + let org_id = org_id.into_inner(); + + match crate::service::alerts::org_config::get_deduplication_config(&org_id).await { + Ok(Some(config)) => Ok(HttpResponse::Ok().json(config)), + Ok(None) => Ok(HttpResponse::Ok().json(GlobalDeduplicationConfig::default_with_presets())), + Err(e) => { + log::error!( + "Error getting deduplication config for org {}: {}", + org_id, + e + ); + Ok(HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("Failed to get deduplication config: {}", e) + }))) + } + } +} + +/// Get deduplication configuration for an organization (OSS - Not Supported) +#[cfg(not(feature = "enterprise"))] +#[utoipa::path( + context_path = "/api", + tag = "Alerts", + operation_id = "GetDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + responses( + (status = 403, description = "Forbidden - Enterprise feature"), + ), +)] +#[get("/{org_id}/alerts/deduplication/config")] +pub async fn get_config(_org_id: web::Path) -> Result { + Ok(HttpResponse::Forbidden().json("Enterprise feature not available")) +} + +/// Set deduplication configuration for an organization +#[cfg(feature = "enterprise")] +#[utoipa::path( + context_path = "/api", + tag = "Alerts", + operation_id = "SetDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + request_body(content = GlobalDeduplicationConfig, description = "Deduplication configuration", content_type = "application/json"), + responses( + (status = 200, description = "Success"), + (status = 400, description = "Bad request"), + (status = 403, description = "Forbidden - Enterprise feature"), + (status = 500, description = "Internal server error"), + ), +)] +#[post("/{org_id}/alerts/deduplication/config")] +pub async fn set_config( + org_id: web::Path, + config: web::Json, +) -> Result { + let org_id = org_id.into_inner(); + let config = config.into_inner(); + + match crate::service::alerts::org_config::set_deduplication_config(&org_id, &config).await { + Ok(()) => Ok(HttpResponse::Ok().json(serde_json::json!({ + "message": "Deduplication config saved successfully" + }))), + Err(e) => { + log::error!( + "Error setting deduplication config for org {}: {}", + org_id, + e + ); + Ok(HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("Failed to set deduplication config: {}", e) + }))) + } + } +} + +/// Set deduplication configuration for an organization (OSS - Not Supported) +#[cfg(not(feature = "enterprise"))] +#[utoipa::path( + context_path = "/api", + tag = "Alerts", + operation_id = "SetDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + responses( + (status = 403, description = "Forbidden - Enterprise feature"), + ), +)] +#[post("/{org_id}/alerts/deduplication/config")] +pub async fn set_config( + _org_id: web::Path, + _config: web::Json, +) -> Result { + Ok(HttpResponse::Forbidden().json("Enterprise feature not available")) +} + +/// Delete deduplication configuration for an organization +#[cfg(feature = "enterprise")] +#[utoipa::path( + delete, + path = "/{org_id}/alerts/deduplication/config", + context_path = "/api", + tag = "Alerts", + operation_id = "DeleteDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + responses( + (status = 200, description = "Success"), + (status = 403, description = "Forbidden - Enterprise feature"), + (status = 500, description = "Internal server error"), + ), +)] +#[delete("/{org_id}/alerts/deduplication/config")] +pub async fn delete_config(org_id: web::Path) -> Result { + let org_id = org_id.into_inner(); + + match crate::service::alerts::org_config::delete_deduplication_config(&org_id).await { + Ok(()) => Ok(HttpResponse::Ok().json(serde_json::json!({ + "message": "Deduplication config deleted successfully" + }))), + Err(e) => { + log::error!( + "Error deleting deduplication config for org {}: {}", + org_id, + e + ); + Ok(HttpResponse::InternalServerError().json(serde_json::json!({ + "error": format!("Failed to delete deduplication config: {}", e) + }))) + } + } +} + +/// Delete deduplication configuration for an organization (OSS - Not Supported) +#[cfg(not(feature = "enterprise"))] +#[utoipa::path( + delete, + path = "/{org_id}/alerts/deduplication/config", + context_path = "/api", + tag = "Alerts", + operation_id = "DeleteDeduplicationConfig", + security(("Authorization" = [])), + params( + ("org_id" = String, Path, description = "Organization ID"), + ), + responses( + (status = 403, description = "Forbidden - Enterprise feature"), + ), +)] +#[delete("/{org_id}/alerts/deduplication/config")] +pub async fn delete_config(_org_id: web::Path) -> Result { + Ok(HttpResponse::Forbidden().json("Enterprise feature not available")) +} diff --git a/src/handler/http/request/alerts/history.rs b/src/handler/http/request/alerts/history.rs index d58db0c1e4..19d4091551 100644 --- a/src/handler/http/request/alerts/history.rs +++ b/src/handler/http/request/alerts/history.rs @@ -83,6 +83,17 @@ pub struct AlertHistoryEntry { pub evaluation_took_in_secs: Option, pub source_node: Option, pub query_took: Option, + // Deduplication information + #[serde(skip_serializing_if = "Option::is_none")] + pub dedup_enabled: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dedup_suppressed: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub dedup_count: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub grouped: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub group_size: Option, } #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] @@ -520,6 +531,18 @@ pub async fn get_alert_history( .and_then(|v| v.as_str()) .map(String::from), query_took: hit.get("query_took").and_then(|v| v.as_i64()), + // Deduplication fields - populated from trigger data if available + dedup_enabled: hit.get("dedup_enabled").and_then(|v| v.as_bool()), + dedup_suppressed: hit.get("dedup_suppressed").and_then(|v| v.as_bool()), + dedup_count: hit + .get("dedup_count") + .and_then(|v| v.as_i64()) + .map(|v| v as i32), + grouped: hit.get("grouped").and_then(|v| v.as_bool()), + group_size: hit + .get("group_size") + .and_then(|v| v.as_i64()) + .map(|v| v as i32), }); } @@ -599,6 +622,11 @@ mod tests { evaluation_took_in_secs: Some(1.5), source_node: Some("node1".to_string()), query_took: Some(100), + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; assert_eq!(entry.alert_name, "test_alert"); @@ -644,6 +672,11 @@ mod tests { evaluation_took_in_secs: None, source_node: None, query_took: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; let response = AlertHistoryResponse { @@ -678,6 +711,11 @@ mod tests { evaluation_took_in_secs: Some(5.5), source_node: Some("node2".to_string()), query_took: Some(500), + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, }; assert_eq!(entry.status, "error"); diff --git a/src/handler/http/request/alerts/mod.rs b/src/handler/http/request/alerts/mod.rs index 2cb825976d..4fa9fc0728 100644 --- a/src/handler/http/request/alerts/mod.rs +++ b/src/handler/http/request/alerts/mod.rs @@ -46,6 +46,8 @@ use crate::{ }, }; +pub mod dedup_stats; +pub mod deduplication; #[allow(deprecated)] pub mod deprecated; pub mod destinations; diff --git a/src/handler/http/router/mod.rs b/src/handler/http/router/mod.rs index 3c114c0e5d..3d5f6f1a5e 100644 --- a/src/handler/http/router/mod.rs +++ b/src/handler/http/router/mod.rs @@ -522,6 +522,7 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) { .service(alerts::trigger_alert) .service(alerts::move_alerts) .service(alerts::history::get_alert_history) + .service(alerts::dedup_stats::get_dedup_summary) .service(alerts::deprecated::save_alert) .service(alerts::deprecated::update_alert) .service(alerts::deprecated::get_alert) @@ -636,6 +637,12 @@ pub fn get_service_routes(svc: &mut web::ServiceConfig) { .service(traces::get_store_stats) .service(patterns::extract_patterns); + #[cfg(feature = "enterprise")] + let service = service + .service(alerts::deduplication::get_config) + .service(alerts::deduplication::set_config) + .service(alerts::deduplication::delete_config); + #[cfg(feature = "cloud")] let service = service .service(organization::org::get_org_invites) diff --git a/src/job/alert_grouping.rs b/src/job/alert_grouping.rs new file mode 100644 index 0000000000..e9ae1f52b1 --- /dev/null +++ b/src/job/alert_grouping.rs @@ -0,0 +1,256 @@ +// Copyright 2025 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 . + +//! Background job to process expired alert batches + +use std::time::Duration; + +use tokio::time; + +// Import trait for alert notification sending +#[cfg(feature = "enterprise")] +use crate::service::alerts::alert::AlertExt; + +/// Background worker that processes expired alert batches +pub async fn process_expired_batches() { + log::info!("[alert_grouping_worker] Starting alert grouping background worker"); + + let mut interval = time::interval(Duration::from_secs(1)); + + loop { + interval.tick().await; + + #[cfg(feature = "enterprise")] + { + let batches = crate::service::alerts::grouping::get_expired_batches(); + + if !batches.is_empty() { + log::debug!( + "[alert_grouping_worker] Processing {} expired batches", + batches.len() + ); + + for batch in batches { + if let Err(e) = send_grouped_notification(batch).await { + log::error!( + "[alert_grouping_worker] Error sending grouped notification: {}", + e + ); + } + } + } + } + } +} + +/// Send a grouped notification for a batch of alerts (public wrapper for sync call) +#[cfg(feature = "enterprise")] +pub async fn send_grouped_notification_sync( + batch: crate::service::alerts::grouping::PendingBatch, +) -> Result<(), anyhow::Error> { + send_grouped_notification(batch).await +} + +/// Send a grouped notification for a batch of alerts +#[cfg(feature = "enterprise")] +async fn send_grouped_notification( + batch: crate::service::alerts::grouping::PendingBatch, +) -> Result<(), anyhow::Error> { + use config::meta::alerts::deduplication::SendStrategy; + + let elapsed_seconds = + (chrono::Utc::now().timestamp_micros() - batch.timer_started_at) / 1_000_000; + + log::info!( + "[alert_grouping_worker] Sending grouped notification for {} alerts with fingerprint {} (waited {}s)", + batch.alerts.len(), + batch.fingerprint, + elapsed_seconds + ); + + // Record wait time metric + config::metrics::ALERT_GROUPING_WAIT_TIME + .with_label_values(&[batch.org_id.as_str()]) + .observe(elapsed_seconds as f64); + + // Get the first alert (primary) and grouping config + let primary_alert = &batch.alerts[0].alert; + let grouping_config = primary_alert + .deduplication + .as_ref() + .and_then(|d| d.grouping.as_ref()) + .ok_or_else(|| anyhow::anyhow!("Grouping config not found"))?; + + let send_strategy = &grouping_config.send_strategy; + + // Collect alert details + let alert_names: Vec = batch.alerts.iter().map(|a| a.alert.name.clone()).collect(); + + let alert_count = batch.alerts.len(); + let suppressed_count = alert_count.saturating_sub(1); + + // Build notification context based on strategy + let notification_context = match send_strategy { + SendStrategy::FirstWithCount => { + // "Alert a60 fired (2 others suppressed: a80, a90)" + if suppressed_count > 0 { + let suppressed_names = alert_names[1..].join(", "); + format!( + "Alert '{}' fired ({} other{} suppressed: {})", + alert_names[0], + suppressed_count, + if suppressed_count == 1 { "" } else { "s" }, + suppressed_names + ) + } else { + format!("Alert '{}' fired", alert_names[0]) + } + } + SendStrategy::Summary => { + // "3 alerts fired: a60, a80, a90" + format!( + "{} alert{} fired: {}", + alert_count, + if alert_count == 1 { "" } else { "s" }, + alert_names.join(", ") + ) + } + SendStrategy::All => { + // Full details of each alert + let mut details = vec![format!("Grouped Alerts ({} total)", alert_count)]; + for (i, batched) in batch.alerts.iter().enumerate() { + details.push(format!( + "\n{}. Alert: '{}'\n Triggered at: {}\n Rows matched: {}", + i + 1, + batched.alert.name, + chrono::DateTime::::from_timestamp( + batched.timestamp / 1_000_000, + 0 + ) + .map(|dt| dt.to_rfc3339()) + .unwrap_or_else(|| "unknown".to_string()), + batched.rows.len() + )); + } + details.join("\n") + } + }; + + log::info!( + "[alert_grouping_worker] Strategy: {:?}, Sending notification with context: {}", + send_strategy, + notification_context.chars().take(200).collect::() // Log first 200 chars + ); + + // Prepare combined trigger data based on strategy + let combined_rows = match send_strategy { + SendStrategy::FirstWithCount | SendStrategy::Summary => { + // For FirstWithCount and Summary, just use the first alert's data + batch.alerts[0].rows.clone() + } + SendStrategy::All => { + // For All, combine all rows from all alerts + let mut all_rows = Vec::new(); + for batched in &batch.alerts { + all_rows.extend(batched.rows.clone()); + } + all_rows + } + }; + + // Use the primary alert for notification and inject grouped context + let mut notification_alert = primary_alert.clone(); + + // Inject grouped context into context_attributes for template access + let mut context_attrs = notification_alert.context_attributes.unwrap_or_default(); + context_attrs.insert("grouped_alerts".to_string(), alert_names.join(", ")); + context_attrs.insert("alert_count".to_string(), alert_count.to_string()); + context_attrs.insert("grouped_summary".to_string(), notification_context.clone()); + context_attrs.insert("is_grouped".to_string(), "true".to_string()); + notification_alert.context_attributes = Some(context_attrs); + + // Get the timestamp for notification + let rows_end_time = batch + .alerts + .iter() + .map(|a| a.timestamp) + .max() + .unwrap_or_else(|| chrono::Utc::now().timestamp_micros()); + + let start_time = batch.alerts.iter().map(|a| a.timestamp).min(); + + let evaluation_timestamp = chrono::Utc::now().timestamp_micros(); + + // Send notification using alert's send_notification method + match notification_alert + .send_notification( + &combined_rows, + rows_end_time, + start_time, + evaluation_timestamp, + ) + .await + { + Ok((success_msg, err_msg)) => { + if !err_msg.is_empty() { + log::error!( + "[alert_grouping_worker] Some destinations failed for grouped notification (fingerprint: {}): {}", + batch.fingerprint, + err_msg + ); + // Record error metric + config::metrics::ALERT_GROUPING_SEND_ERRORS_TOTAL + .with_label_values(&[batch.org_id.as_str(), "partial_failure"]) + .inc(); + return Err(anyhow::anyhow!("Partial failure: {}", err_msg)); + } + + // Record successful send metrics + let strategy_str = format!("{:?}", send_strategy).to_lowercase(); + let reason = if elapsed_seconds >= grouping_config.group_wait_seconds { + "expired" + } else { + "max_size" + }; + + config::metrics::ALERT_GROUPING_NOTIFICATIONS_SENT_TOTAL + .with_label_values(&[batch.org_id.as_str(), strategy_str.as_str(), reason]) + .inc(); + + config::metrics::ALERT_GROUPING_BATCH_SIZE + .with_label_values(&[batch.org_id.as_str(), strategy_str.as_str()]) + .observe(alert_count as f64); + + log::info!( + "[alert_grouping_worker] Successfully sent grouped notification (fingerprint: {}): {}", + batch.fingerprint, + success_msg + ); + Ok(()) + } + Err(e) => { + log::error!( + "[alert_grouping_worker] Failed to send grouped notification (fingerprint: {}): {}", + batch.fingerprint, + e + ); + // Record error metric + config::metrics::ALERT_GROUPING_SEND_ERRORS_TOTAL + .with_label_values(&[batch.org_id.as_str(), "send_failed"]) + .inc(); + Err(anyhow::anyhow!("Send failed: {}", e)) + } + } +} diff --git a/src/job/alert_manager.rs b/src/job/alert_manager.rs index ffc6bef613..7070fdb56e 100644 --- a/src/job/alert_manager.rs +++ b/src/job/alert_manager.rs @@ -160,8 +160,7 @@ async fn cleanup_alert_dedup_state() -> Result<(), anyhow::Error> { // This is conservative - most alerts have shorter time windows let older_than_minutes = 24 * 60; // 24 hours - match service::alerts::deduplication::cleanup_expired_state(db, None, older_than_minutes).await - { + match service::alerts::deduplication::cleanup_expired_state(db, older_than_minutes).await { Ok(deleted_count) => { if deleted_count > 0 { log::info!( diff --git a/src/job/mod.rs b/src/job/mod.rs index 68f5a408e4..12c4802d12 100644 --- a/src/job/mod.rs +++ b/src/job/mod.rs @@ -28,6 +28,8 @@ use crate::{ service::{db, self_reporting, users}, }; +#[cfg(feature = "enterprise")] +pub mod alert_grouping; mod alert_manager; #[cfg(feature = "enterprise")] mod cipher; @@ -308,6 +310,8 @@ pub async fn init() -> Result<(), anyhow::Error> { tokio::task::spawn(async move { metrics::run().await }); let _ = promql::run(); tokio::task::spawn(async move { alert_manager::run().await }); + #[cfg(feature = "enterprise")] + tokio::task::spawn(async move { alert_grouping::process_expired_batches().await }); tokio::task::spawn(async move { file_downloader::run().await }); #[cfg(feature = "enterprise")] tokio::task::spawn(async move { pipeline::run().await }); diff --git a/src/service/alerts/alert.rs b/src/service/alerts/alert.rs index caa30749e3..3d6b1a735e 100644 --- a/src/service/alerts/alert.rs +++ b/src/service/alerts/alert.rs @@ -2523,7 +2523,7 @@ mod tests { let json_str2 = r#"{"level": "warn", "job": "test2"}"#; // Simulate rows_tpl_val containing Value::String with JSON content - let rows_tpl_val = vec![ + let rows_tpl_val = [ Value::String(json_str1.to_string()), Value::String(json_str2.to_string()), ]; @@ -2532,9 +2532,8 @@ mod tests { let mut parsed_values = Vec::new(); for v in rows_tpl_val.iter() { if let Value::String(s) = v { - match serde_json::from_str::(s) { - Ok(parsed) => parsed_values.push(parsed), - Err(_) => {} + if let Ok(parsed) = serde_json::from_str::(s) { + parsed_values.push(parsed) } } else { parsed_values.push(v.clone()); diff --git a/src/service/alerts/deduplication.rs b/src/service/alerts/deduplication.rs index a3400cc69a..bd877b99d5 100644 --- a/src/service/alerts/deduplication.rs +++ b/src/service/alerts/deduplication.rs @@ -19,7 +19,10 @@ //! implementation. It is only compiled when the enterprise feature is enabled. use config::{ - meta::alerts::{alert::Alert, deduplication::DeduplicationConfig}, + meta::alerts::{ + alert::Alert, + deduplication::{DeduplicationConfig, GlobalDeduplicationConfig}, + }, utils::json::{Map, Value}, }; use infra::table::entity::alert_dedup_state; @@ -28,12 +31,16 @@ use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, Qu /// Calculate fingerprint for an alert result row /// /// Delegates to enterprise implementation. +/// With org_config provided, uses semantic dimensions for cross-alert deduplication. pub fn calculate_fingerprint( alert: &Alert, result_row: &Map, config: &DeduplicationConfig, + org_config: Option<&GlobalDeduplicationConfig>, ) -> String { - o2_enterprise::enterprise::alerts::dedup::calculate_fingerprint(alert, result_row, config) + o2_enterprise::enterprise::alerts::dedup::calculate_fingerprint( + alert, result_row, config, org_config, + ) } /// Get or create deduplication state @@ -48,13 +55,12 @@ pub async fn get_dedup_state( /// Parameters for saving deduplication state pub struct DedupStateParams<'a> { + pub org_id: &'a str, pub fingerprint: &'a str, pub alert_id: &'a str, - pub org_id: &'a str, pub first_seen_at: i64, pub last_seen_at: i64, pub occurrence_count: i64, - pub notification_sent: bool, } /// Save or update deduplication state @@ -62,26 +68,24 @@ pub async fn save_dedup_state( db: &DatabaseConnection, params: DedupStateParams<'_>, ) -> Result { - let now = o2_enterprise::enterprise::alerts::dedup::current_timestamp_micros(); - // Try to find existing record if let Some(existing) = get_dedup_state(db, params.fingerprint).await? { // Update existing let mut active: alert_dedup_state::ActiveModel = existing.clone().into(); active.last_seen_at = Set(params.last_seen_at); active.occurrence_count = Set(params.occurrence_count); - active.notification_sent = Set(params.notification_sent); active.update(db).await } else { // Insert new + let now = o2_enterprise::enterprise::alerts::dedup::current_timestamp_micros(); let new_state = alert_dedup_state::ActiveModel { + org_id: Set(params.org_id.to_string()), fingerprint: Set(params.fingerprint.to_string()), alert_id: Set(params.alert_id.to_string()), - org_id: Set(params.org_id.to_string()), first_seen_at: Set(params.first_seen_at), last_seen_at: Set(params.last_seen_at), occurrence_count: Set(params.occurrence_count), - notification_sent: Set(params.notification_sent), + notification_sent: Set(false), created_at: Set(now), }; new_state.insert(db).await @@ -99,38 +103,86 @@ pub fn is_within_window(state: &alert_dedup_state::Model, time_window_minutes: i /// Cleanup old deduplication state records pub async fn cleanup_expired_state( db: &DatabaseConnection, - org_id: Option<&str>, older_than_minutes: i64, ) -> Result { let cutoff_time = o2_enterprise::enterprise::alerts::dedup::current_timestamp_micros() - (older_than_minutes * 60 * 1_000_000); - let mut query = alert_dedup_state::Entity::delete_many() + let query = alert_dedup_state::Entity::delete_many() .filter(alert_dedup_state::Column::LastSeenAt.lt(cutoff_time)); - if let Some(org) = org_id { - query = query.filter(alert_dedup_state::Column::OrgId.eq(org)); - } - let result = query.exec(db).await?; Ok(result.rows_affected) } +/// Check for cross-alert semantic matches (when cross_alert_dedup is enabled) +/// +/// Looks for any fingerprint containing the given semantic dimensions within the time window. +/// Used to suppress alerts that share semantic dimensions with recently fired alerts. +pub async fn find_matching_semantic_fingerprints( + db: &DatabaseConnection, + semantic_dimensions: &std::collections::HashMap, + time_window_minutes: i64, +) -> Result, sea_orm::DbErr> { + let cutoff_time = o2_enterprise::enterprise::alerts::dedup::current_timestamp_micros() + - (time_window_minutes * 60 * 1_000_000); + + // Get all recent fingerprints + let recent_states = alert_dedup_state::Entity::find() + .filter(alert_dedup_state::Column::LastSeenAt.gt(cutoff_time)) + .all(db) + .await?; + + // Filter for matches with overlapping semantic dimensions + // Fingerprint format: "dim1=val1,dim2=val2,..." + let matching_states: Vec<_> = recent_states + .into_iter() + .filter(|state| { + // Parse fingerprint to check for dimension overlap + o2_enterprise::enterprise::alerts::dedup::fingerprint_matches_dimensions( + &state.fingerprint, + semantic_dimensions, + ) + }) + .collect(); + + Ok(matching_states) +} + /// Apply deduplication to alert result rows before sending notifications /// /// This is the main entry point for deduplication logic in the alert execution flow. +/// Fetches org-level config for semantic groups and cross-alert deduplication. pub async fn apply_deduplication( db: &DatabaseConnection, alert: &Alert, result_rows: Vec>, ) -> Result>, sea_orm::DbErr> { - // Check if deduplication is enabled + // Check if per-alert deduplication is enabled let dedup_config = match &alert.deduplication { Some(config) if config.enabled => config, _ => return Ok(result_rows), // Deduplication disabled, return all rows }; - apply_deduplication_impl(db, alert, result_rows, dedup_config).await + // Fetch org-level config for semantic groups + let org_config = match super::org_config::get_deduplication_config(&alert.org_id).await { + Ok(Some(config)) if config.enabled => Some(config), + Ok(Some(_)) => { + // Org config exists but disabled - still allow per-alert dedup + None + } + Ok(None) => None, // No org config + Err(e) => { + log::warn!( + "Failed to fetch org dedup config for {}: {}, proceeding without semantic groups", + alert.org_id, + e + ); + None + } + }; + + apply_deduplication_impl(db, alert, result_rows, dedup_config, org_config.as_ref()).await } /// Enterprise implementation of apply_deduplication @@ -139,6 +191,7 @@ async fn apply_deduplication_impl( alert: &Alert, result_rows: Vec>, dedup_config: &DeduplicationConfig, + org_config: Option<&GlobalDeduplicationConfig>, ) -> Result>, sea_orm::DbErr> { let now = o2_enterprise::enterprise::alerts::dedup::current_timestamp_micros(); let alert_id = alert.get_unique_key(); @@ -153,7 +206,7 @@ async fn apply_deduplication_impl( let mut deduplicated_rows = Vec::new(); for row in result_rows { - let fingerprint = calculate_fingerprint(alert, &row, dedup_config); + let fingerprint = calculate_fingerprint(alert, &row, dedup_config, org_config); // Check if this fingerprint exists and is within time window let should_send = match get_dedup_state(db, &fingerprint).await? { @@ -163,13 +216,12 @@ async fn apply_deduplication_impl( if let Err(e) = save_dedup_state( db, DedupStateParams { + org_id: org_id.as_str(), fingerprint: &fingerprint, alert_id: &alert_id, - org_id, first_seen_at: existing_state.first_seen_at, last_seen_at: now, occurrence_count: existing_state.occurrence_count + 1, - notification_sent: true, }, ) .await @@ -180,6 +232,26 @@ async fn apply_deduplication_impl( e ); } + + // Record suppression metric + let dedup_type = if existing_state.alert_id == alert_id { + "same_alert" + } else { + "cross_alert" + }; + config::metrics::ALERT_DEDUP_SUPPRESSED_TOTAL + .with_label_values(&[org_id.as_str(), &alert.name, dedup_type]) + .inc(); + + log::debug!( + "[dedup] Suppressed alert '{}' for org: {}, fingerprint: {}, occurrence: {}, type: {}", + alert.name, + org_id, + fingerprint, + existing_state.occurrence_count + 1, + dedup_type + ); + false } else { // Outside window - treat as new alert @@ -197,13 +269,12 @@ async fn apply_deduplication_impl( if let Err(e) = save_dedup_state( db, DedupStateParams { + org_id: org_id.as_str(), fingerprint: &fingerprint, alert_id: &alert_id, - org_id, first_seen_at: now, last_seen_at: now, occurrence_count: 1, - notification_sent: true, }, ) .await @@ -215,6 +286,18 @@ async fn apply_deduplication_impl( ); } + // Record passed metric + config::metrics::ALERT_DEDUP_PASSED_TOTAL + .with_label_values(&[org_id.as_str(), &alert.name]) + .inc(); + + log::debug!( + "[dedup] Alert '{}' passed dedup check for org: {}, fingerprint: {}", + alert.name, + org_id, + fingerprint + ); + deduplicated_rows.push(row); } } diff --git a/src/service/alerts/grouping.rs b/src/service/alerts/grouping.rs new file mode 100644 index 0000000000..605ae2fba1 --- /dev/null +++ b/src/service/alerts/grouping.rs @@ -0,0 +1,249 @@ +// Copyright 2025 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 . + +//! Alert grouping and batching for deduplication +//! +//! Implements wait-and-collect logic to batch multiple alerts with the same +//! fingerprint before sending a single grouped notification. + +use std::sync::Arc; + +use chrono::Utc; +use config::{meta::alerts::alert::Alert, utils::json}; +use dashmap::DashMap; +use once_cell::sync::Lazy; + +/// In-memory cache of pending alert batches +/// Key: fingerprint, Value: PendingBatch +static PENDING_BATCHES: Lazy>> = + Lazy::new(|| Arc::new(DashMap::new())); + +/// A batch of alerts waiting to be sent together +#[derive(Clone, Debug)] +pub struct PendingBatch { + pub fingerprint: String, + pub org_id: String, + pub alerts: Vec, + pub timer_started_at: i64, + pub group_wait_seconds: i64, + pub max_group_size: usize, +} + +/// An alert waiting in a batch +#[derive(Clone, Debug)] +pub struct BatchedAlert { + pub alert: Alert, + pub rows: Vec>, + pub timestamp: i64, +} + +impl PendingBatch { + /// Create a new pending batch + pub fn new( + fingerprint: String, + org_id: String, + alert: Alert, + rows: Vec>, + group_wait_seconds: i64, + max_group_size: usize, + ) -> Self { + let now = Utc::now().timestamp_micros(); + Self { + fingerprint, + org_id, + alerts: vec![BatchedAlert { + alert, + rows, + timestamp: now, + }], + timer_started_at: now, + group_wait_seconds, + max_group_size, + } + } + + /// Add an alert to this batch + pub fn add_alert(&mut self, alert: Alert, rows: Vec>) -> bool { + if self.alerts.len() >= self.max_group_size { + return false; // Batch full + } + + let now = Utc::now().timestamp_micros(); + self.alerts.push(BatchedAlert { + alert, + rows, + timestamp: now, + }); + true + } + + /// Check if batch wait time has expired + pub fn is_expired(&self) -> bool { + let now = Utc::now().timestamp_micros(); + let elapsed_seconds = (now - self.timer_started_at) / 1_000_000; + elapsed_seconds >= self.group_wait_seconds + } + + /// Check if batch is at max capacity + pub fn is_full(&self) -> bool { + self.alerts.len() >= self.max_group_size + } +} + +/// Add alert to pending batch or create new batch +/// Returns true if batch is ready to send (expired or full) +pub fn add_to_batch( + fingerprint: String, + org_id: String, + alert: Alert, + rows: Vec>, + group_wait_seconds: i64, + max_group_size: usize, +) -> bool { + let mut batch_ready = false; + let mut is_new_batch = false; + let mut batch_size = 0; + + PENDING_BATCHES + .entry(fingerprint.clone()) + .and_modify(|batch| { + if batch.add_alert(alert.clone(), rows.clone()) { + batch_size = batch.alerts.len(); + log::debug!( + "[grouping] Added alert '{}' to existing batch {} (count: {}/{})", + alert.name, + fingerprint, + batch.alerts.len(), + batch.max_group_size + ); + if batch.is_full() { + batch_ready = true; + log::info!( + "[grouping] Batch {} reached max size ({}), ready to send", + fingerprint, + batch.max_group_size + ); + } + } else { + log::warn!( + "[grouping] Failed to add alert '{}' to batch {} (already full)", + alert.name, + fingerprint + ); + } + }) + .or_insert_with(|| { + is_new_batch = true; + batch_size = 1; + log::info!( + "[grouping] Created new batch for fingerprint {}, alert: '{}', org: {}, wait_seconds: {}, max_size: {}", + fingerprint, + alert.name, + org_id, + group_wait_seconds, + max_group_size + ); + PendingBatch::new( + fingerprint.clone(), + org_id, + alert, + rows, + group_wait_seconds, + max_group_size, + ) + }); + + if is_new_batch { + let batch_count = PENDING_BATCHES.len(); + log::debug!("[grouping] Current pending batches count: {}", batch_count); + + // Update gauge metric for pending batches + // Use org_id from the batch we just inserted + if let Some(batch) = PENDING_BATCHES.get(&fingerprint) { + config::metrics::ALERT_GROUPING_BATCHES_PENDING + .with_label_values(&[batch.org_id.as_str()]) + .set(batch_count as i64); + } + } + + batch_ready +} + +/// Get and remove a batch if it's ready (expired or full) +pub fn get_ready_batch(fingerprint: &str) -> Option { + if let Some(entry) = PENDING_BATCHES.get(fingerprint) + && (entry.is_expired() || entry.is_full()) + { + let batch = PENDING_BATCHES.remove(fingerprint).map(|(_, batch)| batch); + if let Some(ref b) = batch { + log::debug!( + "[grouping] Retrieved ready batch for fingerprint {} ({} alerts)", + fingerprint, + b.alerts.len() + ); + } + return batch; + } + None +} + +/// Get all expired batches +pub fn get_expired_batches() -> Vec { + let mut expired = Vec::new(); + let now = Utc::now().timestamp_micros(); + + PENDING_BATCHES.retain(|fingerprint, batch| { + if batch.is_expired() { + let elapsed_seconds = (now - batch.timer_started_at) / 1_000_000; + log::info!( + "[grouping] Batch {} expired after {}s with {} alerts", + fingerprint, + elapsed_seconds, + batch.alerts.len() + ); + expired.push(batch.clone()); + false // Remove from map + } else { + true // Keep in map + } + }); + + if !expired.is_empty() { + log::debug!( + "[grouping] Found {} expired batches, {} still pending", + expired.len(), + PENDING_BATCHES.len() + ); + } + + expired +} + +/// Get count of pending batches for an organization +pub fn get_pending_batch_count(org_id: &str) -> i64 { + PENDING_BATCHES + .iter() + .filter(|entry| entry.org_id == org_id) + .count() as i64 +} + +/// Clear all pending batches (for testing) +pub fn clear_all_batches() { + let count = PENDING_BATCHES.len(); + PENDING_BATCHES.clear(); + if count > 0 { + log::info!("[grouping] Cleared {} pending batches (testing)", count); + } +} diff --git a/src/service/alerts/mod.rs b/src/service/alerts/mod.rs index 7db3970eb5..92da237cca 100644 --- a/src/service/alerts/mod.rs +++ b/src/service/alerts/mod.rs @@ -48,6 +48,10 @@ pub mod alert; pub mod deduplication; pub mod derived_streams; pub mod destinations; +#[cfg(feature = "enterprise")] +pub mod grouping; +#[cfg(feature = "enterprise")] +pub mod org_config; pub mod scheduler; pub mod templates; diff --git a/src/service/alerts/org_config.rs b/src/service/alerts/org_config.rs new file mode 100644 index 0000000000..f2ce8c965c --- /dev/null +++ b/src/service/alerts/org_config.rs @@ -0,0 +1,81 @@ +// Copyright 2025 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 . + +//! Organization-level alert configuration storage +//! +//! This module provides functions to store and retrieve org-level alert configs +//! using the existing key-value DB interface. + +use config::meta::alerts::deduplication::GlobalDeduplicationConfig; +use infra::db; + +const MODULE: &str = "alert_config"; +const DEDUPLICATION_KEY: &str = "deduplication"; + +/// Get deduplication config for an organization +pub async fn get_deduplication_config( + org_id: &str, +) -> Result, anyhow::Error> { + let key = db::build_key(MODULE, org_id, DEDUPLICATION_KEY, 0); + let db = db::get_db().await; + + match db.get(&key).await { + Ok(bytes) => { + let config: GlobalDeduplicationConfig = serde_json::from_slice(&bytes)?; + Ok(Some(config)) + } + Err(infra::errors::Error::DbError(infra::errors::DbError::KeyNotExists(_))) => Ok(None), + Err(e) => Err(anyhow::anyhow!("Failed to get deduplication config: {}", e)), + } +} + +/// Set deduplication config for an organization +pub async fn set_deduplication_config( + org_id: &str, + config: &GlobalDeduplicationConfig, +) -> Result<(), anyhow::Error> { + let key = db::build_key(MODULE, org_id, DEDUPLICATION_KEY, 0); + let value = serde_json::to_vec(config)?; + let db = db::get_db().await; + + db.put(&key, value.into(), db::NO_NEED_WATCH, None) + .await + .map_err(|e| anyhow::anyhow!("Failed to set deduplication config: {}", e))?; + + Ok(()) +} + +/// Delete deduplication config for an organization +pub async fn delete_deduplication_config(org_id: &str) -> Result<(), anyhow::Error> { + let key = db::build_key(MODULE, org_id, DEDUPLICATION_KEY, 0); + let db = db::get_db().await; + + db.delete_if_exists(&key, false, db::NO_NEED_WATCH) + .await + .map_err(|e| anyhow::anyhow!("Failed to delete deduplication config: {}", e))?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_build_key() { + let key = db::build_key(MODULE, "org123", DEDUPLICATION_KEY, 0); + assert_eq!(key, "/alert_config/org123/deduplication"); + } +} diff --git a/src/service/alerts/scheduler/handlers.rs b/src/service/alerts/scheduler/handlers.rs index c33c52792d..8520b22876 100644 --- a/src/service/alerts/scheduler/handlers.rs +++ b/src/service/alerts/scheduler/handlers.rs @@ -220,12 +220,7 @@ async fn handle_alert_triggers( start_time: now, end_time: now, next_run_at: now, - skipped_alerts_count: None, - delay_in_secs: None, - evaluation_took_in_secs: None, - query_took: None, - success_response: None, - is_partial: None, + ..Default::default() }); return Err(anyhow::anyhow!("Alert not found")); } @@ -269,12 +264,7 @@ async fn handle_alert_triggers( start_time: now, end_time: now, next_run_at: now, - skipped_alerts_count: None, - delay_in_secs: None, - evaluation_took_in_secs: None, - query_took: None, - success_response: None, - is_partial: None, + ..Default::default() }); return Err(anyhow::anyhow!("Error getting alert by id: {}", e)); } @@ -469,15 +459,11 @@ async fn handle_alert_triggers( end_time: *skipped_last_timestamp, retries: trigger.retries, delay_in_secs: Some(Duration::microseconds(delay).num_seconds()), - error: None, - success_response: None, - is_partial: None, - evaluation_took_in_secs: None, source_node: Some(source_node.clone()), - query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), skipped_alerts_count: Some(skipped_alerts_count as i64), + ..Default::default() }); } log::info!( @@ -521,7 +507,7 @@ async fn handle_alert_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, + ..Default::default() }; let evaluation_took = Instant::now(); @@ -643,6 +629,115 @@ async fn handle_alert_triggers( // send notification if let Some(data) = trigger_results.data { + // Check if grouping is enabled BEFORE deduplication (enterprise-only feature) + #[cfg(feature = "enterprise")] + let grouping_enabled = alert + .deduplication + .as_ref() + .and_then(|d| d.grouping.as_ref()) + .map(|g| g.enabled) + .unwrap_or(false); + + #[cfg(not(feature = "enterprise"))] + let grouping_enabled = false; + + if grouping_enabled { + #[cfg(feature = "enterprise")] + { + let grouping_config = alert + .deduplication + .as_ref() + .and_then(|d| d.grouping.as_ref()) + .unwrap(); + + // Calculate fingerprint for grouping + let fingerprint = if let Some(dedup_config) = alert.deduplication.as_ref() { + let org_config = + match crate::service::alerts::org_config::get_deduplication_config( + &new_trigger.org, + ) + .await + { + Ok(Some(config)) => Some(config), + _ => None, + }; + + if let Some(first_row) = data.first() { + crate::service::alerts::deduplication::calculate_fingerprint( + &alert, + first_row, + dedup_config, + org_config.as_ref(), + ) + } else { + alert.get_unique_key() + } + } else { + alert.get_unique_key() + }; + + log::debug!( + "[SCHEDULER trace_id {scheduler_trace_id}] Adding alert to batch, org: {}, alert: {}, fingerprint: {}, rows: {}", + &new_trigger.org, + &alert.name, + fingerprint, + data.len() + ); + + // Add to batch + let batch_ready = crate::service::alerts::grouping::add_to_batch( + fingerprint.clone(), + new_trigger.org.clone(), + alert.clone(), + data.clone(), + grouping_config.group_wait_seconds, + grouping_config.max_group_size, + ); + + if batch_ready { + log::info!( + "[SCHEDULER trace_id {scheduler_trace_id}] Batch {} reached max size, sending immediately", + fingerprint + ); + if let Some(batch) = + crate::service::alerts::grouping::get_ready_batch(&fingerprint) + && let Err(e) = + crate::job::alert_grouping::send_grouped_notification_sync(batch).await + { + log::error!( + "[SCHEDULER trace_id {scheduler_trace_id}] Failed to send grouped notification: {}", + e + ); + } + } else { + log::debug!( + "[SCHEDULER trace_id {scheduler_trace_id}] Alert added to batch, waiting for more alerts or timeout, fingerprint: {}", + fingerprint + ); + } + + // Mark as grouped for history tracking + trigger_data_stream.dedup_enabled = Some(true); + trigger_data_stream.grouped = Some(true); + trigger_data_stream.group_size = Some(if batch_ready { + grouping_config.max_group_size as i32 + } else { + 1 + }); + + // Alert added to batch, don't send individual notification + trigger_data.period_end_time = if should_store_last_end_time { + Some(trigger_results.end_time) + } else { + None + }; + new_trigger.data = json::to_string(&trigger_data).unwrap(); + db::scheduler::update_trigger(new_trigger, true, &query_trace_id).await?; + publish_triggers_usage(trigger_data_stream); + return Ok(()); + } + } + // Apply deduplication if enabled (enterprise-only feature) #[cfg(feature = "enterprise")] let data = if let Some(db) = ORM_CLIENT.get() { @@ -660,6 +755,11 @@ async fn handle_alert_triggers( &new_trigger.org, &new_trigger.module_key ); + + // Mark as suppressed for history tracking + trigger_data_stream.dedup_enabled = Some(true); + trigger_data_stream.dedup_suppressed = Some(true); + // All results were deduplicated, skip notification // Still update the trigger timing trigger_data.period_end_time = if should_store_last_end_time { @@ -737,6 +837,14 @@ async fn handle_alert_triggers( ); trigger_data_stream.start_time = alert_start_time; trigger_data_stream.end_time = alert_end_time; + + // Mark dedup status for history (if dedup was enabled, alert passed through) + if alert.deduplication.as_ref().is_some_and(|d| d.enabled) { + trigger_data_stream.dedup_enabled = Some(true); + trigger_data_stream.dedup_suppressed = Some(false); + } + + // No grouping - send individual notification match alert .send_notification( &data, @@ -1014,7 +1122,7 @@ async fn handle_report_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(Duration::microseconds(time_in_queue).num_milliseconds()), - skipped_alerts_count: None, + ..Default::default() }); return Err(anyhow::anyhow!( "Error getting report: {report_id}, error: {e}" @@ -1148,7 +1256,7 @@ async fn handle_report_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(Duration::microseconds(time_in_queue).num_milliseconds()), - skipped_alerts_count: None, + ..Default::default() }; if trigger.retries >= max_retries { @@ -1311,7 +1419,7 @@ async fn handle_derived_stream_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, + ..Default::default() }; log::error!("[SCHEDULER trace_id {scheduler_trace_id}] {err_msg}"); @@ -1381,7 +1489,7 @@ async fn handle_derived_stream_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, + ..Default::default() }; log::info!("[SCHEDULER trace_id {scheduler_trace_id}] {msg}"); new_trigger_data.reset(); @@ -1420,7 +1528,7 @@ async fn handle_derived_stream_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, + ..Default::default() }; log::error!("[SCHEDULER trace_id {scheduler_trace_id}] {err_msg}"); new_trigger_data.reset(); @@ -1569,7 +1677,7 @@ async fn handle_derived_stream_triggers( query_took: None, scheduler_trace_id: Some(scheduler_trace_id.clone()), time_in_queue_ms: Some(time_in_queue), - skipped_alerts_count: None, + ..Default::default() }; // conditionally modify supposed_to_be_run_at diff --git a/src/service/ingestion/mod.rs b/src/service/ingestion/mod.rs index f64a556533..ac5a0096c9 100644 --- a/src/service/ingestion/mod.rs +++ b/src/service/ingestion/mod.rs @@ -260,7 +260,7 @@ pub async fn evaluate_trigger(triggers: TriggerAlertData) { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, - skipped_alerts_count: None, + ..Default::default() }; match alert.send_notification(val, now, None, now).await { Err(e) => { diff --git a/src/service/self_reporting/queues.rs b/src/service/self_reporting/queues.rs index 7940faeb72..8e63a859b4 100644 --- a/src/service/self_reporting/queues.rs +++ b/src/service/self_reporting/queues.rs @@ -414,6 +414,11 @@ mod tests { query_took: None, scheduler_trace_id: None, time_in_queue_ms: None, + dedup_enabled: None, + dedup_suppressed: None, + dedup_count: None, + grouped: None, + group_size: None, } } diff --git a/src/service/self_reporting/triggers_schema.rs b/src/service/self_reporting/triggers_schema.rs index 4ceb0a1d3e..c667c73851 100644 --- a/src/service/self_reporting/triggers_schema.rs +++ b/src/service/self_reporting/triggers_schema.rs @@ -159,15 +159,15 @@ mod tests { assert!(field_names.contains(&"status".to_string())); assert!(field_names.contains(&"next_run_at".to_string())); - // Verify count matches struct fields (21 total) - assert_eq!(field_names.len(), 21); + // Verify count matches struct fields (26 total: 21 original + 5 dedup/grouping fields) + assert_eq!(field_names.len(), 26); // Verify no duplicate fields let unique_count = field_names .iter() .collect::>() .len(); - assert_eq!(unique_count, 21); + assert_eq!(unique_count, 26); } #[tokio::test] diff --git a/tests/ui-testing/pages/generalPages/homePage.js b/tests/ui-testing/pages/generalPages/homePage.js index 82c77e308b..89e0e96569 100644 --- a/tests/ui-testing/pages/generalPages/homePage.js +++ b/tests/ui-testing/pages/generalPages/homePage.js @@ -104,7 +104,7 @@ export class HomePage { async clickDefaultOrg() { // Click the dropdown to open the list of organizations - await this.page.getByText('arrow_drop_down').click(); + await this.page.getByText('arrow_drop_down').first().click(); // Wait for the dropdown options to be present using the correct selector const optionsSelector = '[data-test="organization-menu-item-label-item-label"]'; @@ -133,7 +133,7 @@ export class HomePage { } // Optionally, close the dropdown again if needed - await this.page.getByText('arrow_drop_down').click(); + await this.page.getByText('arrow_drop_down').first().click(); } } \ No newline at end of file diff --git a/web/src/components/alerts/AlertHistory.vue b/web/src/components/alerts/AlertHistory.vue index 3ef576baa9..5e455bcdfc 100644 --- a/web/src/components/alerts/AlertHistory.vue +++ b/web/src/components/alerts/AlertHistory.vue @@ -220,6 +220,52 @@ along with this program. If not, see . --> + + + + + + +
+ +
+
+ +
+ {{ t('settings.alertCorrelation') }} +
+
+ +
+ + +
+ +
+
+ + { + // Reload organization settings to get updated dedup config + await store.dispatch("getDefaultOrganizationSettings"); + showCorrelationDrawer.value = false; + }; + const exportAlert = async (row: any) => { // Find the alert based on uuid const alertToBeExported = await getAlertById(row.alert_id); @@ -2478,6 +2558,8 @@ export default defineComponent({ showHistoryDrawer, selectedHistoryAlertId, selectedHistoryAlertName, + showCorrelationDrawer, + onCorrelationSettingsSaved, refreshImportedAlerts, folderIdToBeCloned, updateFolderIdToBeCloned, diff --git a/web/src/components/alerts/DedupSummaryCards.vue b/web/src/components/alerts/DedupSummaryCards.vue new file mode 100644 index 0000000000..6feeb14fa1 --- /dev/null +++ b/web/src/components/alerts/DedupSummaryCards.vue @@ -0,0 +1,158 @@ + + + + + + + diff --git a/web/src/components/alerts/OrganizationDeduplicationSettings.vue b/web/src/components/alerts/OrganizationDeduplicationSettings.vue new file mode 100644 index 0000000000..f99e016e34 --- /dev/null +++ b/web/src/components/alerts/OrganizationDeduplicationSettings.vue @@ -0,0 +1,334 @@ + + + + + + + diff --git a/web/src/components/alerts/SemanticFieldGroupsConfig.vue b/web/src/components/alerts/SemanticFieldGroupsConfig.vue new file mode 100644 index 0000000000..a9e2e34ea7 --- /dev/null +++ b/web/src/components/alerts/SemanticFieldGroupsConfig.vue @@ -0,0 +1,398 @@ + + + + + + + diff --git a/web/src/components/alerts/SemanticGroupItem.vue b/web/src/components/alerts/SemanticGroupItem.vue new file mode 100644 index 0000000000..cb60bb61f0 --- /dev/null +++ b/web/src/components/alerts/SemanticGroupItem.vue @@ -0,0 +1,210 @@ + + + + + + + diff --git a/web/src/components/alerts/TagInput.vue b/web/src/components/alerts/TagInput.vue new file mode 100644 index 0000000000..2d49185d0c --- /dev/null +++ b/web/src/components/alerts/TagInput.vue @@ -0,0 +1,194 @@ + + + + + + + diff --git a/web/src/locales/languages/en.json b/web/src/locales/languages/en.json index 76d487f418..17dda9e21a 100644 --- a/web/src/locales/languages/en.json +++ b/web/src/locales/languages/en.json @@ -1246,6 +1246,7 @@ }, "settings": { "header": "Management", + "alertCorrelation": "Alert Deduplication", "userKeyLabel": "User Key", "rumKeyLabel": "RUM Key", "newUserKeyBtnLbl": "Add User Key", diff --git a/web/src/services/alerts.ts b/web/src/services/alerts.ts index a0912b1e53..63a20443e9 100644 --- a/web/src/services/alerts.ts +++ b/web/src/services/alerts.ts @@ -195,6 +195,18 @@ const alerts = { if (query.sort_order) params.append("sort_order", query.sort_order); return http().get(`/api/${org_identifier}/alerts/history?${params}`); }, + getOrganizationDeduplicationConfig: (org_identifier: string) => { + return http().get(`/api/${org_identifier}/alerts/deduplication/config`); + }, + setOrganizationDeduplicationConfig: (org_identifier: string, data: any) => { + return http().post(`/api/${org_identifier}/alerts/deduplication/config`, data); + }, + deleteOrganizationDeduplicationConfig: (org_identifier: string) => { + return http().delete(`/api/${org_identifier}/alerts/deduplication/config`); + }, + get_dedup_summary: (org_identifier: string) => { + return http().get(`/api/${org_identifier}/alerts/dedup/summary`); + }, }; export default alerts;