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;