""" Smart Alert Manager for AegisLM Framework Production-ready alert management with validation, cooldown periods, intelligent filtering, and comprehensive notification handling. """ import asyncio import logging from typing import Dict, List, Any, Optional, Callable from datetime import datetime, timedelta from dataclasses import dataclass, field from collections import defaultdict import json logger = logging.getLogger(__name__) @dataclass class AlertRule: """Alert rule configuration.""" name: str condition: str # Metric condition (e.g., "cpu_usage > 80") threshold: float duration: int # seconds before alert triggers severity: str # "low", "medium", "high", "critical" cooldown: int # seconds between alerts enabled: bool = True description: str = "" tags: List[str] = field(default_factory=list) @dataclass class Alert: """Alert information.""" alert_id: str rule_name: str metric_name: str current_value: float threshold: float severity: str message: str triggered_at: datetime acknowledged: bool = False acknowledged_by: Optional[str] = None acknowledged_at: Optional[datetime] = None resolved: bool = False resolved_at: Optional[datetime] = None metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class AlertStatistics: """Alert system statistics.""" total_alerts: int = 0 active_alerts: int = 0 resolved_alerts: int = 0 acknowledged_alerts: int = 0 alerts_by_severity: Dict[str, int] = field(default_factory=lambda: defaultdict(int)) alerts_by_rule: Dict[str, int] = field(default_factory=lambda: defaultdict(int)) average_resolution_time: float = 0.0 false_positive_rate: float = 0.0 class SmartAlertManager: """ Smart alert management with validation and cooldown periods. Provides intelligent alert filtering, cooldown enforcement, false positive reduction, and comprehensive alert lifecycle management. """ def __init__(self, validation_interval: int = 30, max_history: int = 10000): """ Initialize smart alert manager. Args: validation_interval: Interval for alert validation in seconds max_history: Maximum number of alerts to keep in history """ self.validation_interval = validation_interval self.max_history = max_history self.alert_rules: Dict[str, AlertRule] = {} self.active_alerts: Dict[str, Alert] = {} self.alert_history: List[Alert] = [] self.metric_tracker: Dict[str, List[Dict[str, Any]]] = defaultdict(list) self.cooldown_tracker: Dict[str, datetime] = {} self.alert_statistics = AlertStatistics() self._validation_task = None self._notification_handlers: List[Callable] = [] self._suppression_rules: Dict[str, Callable] = {} def add_alert_rule(self, rule: AlertRule): """ Add an alert rule. Args: rule: Alert rule configuration """ self.alert_rules[rule.name] = rule logger.info(f"Added alert rule: {rule.name}") def remove_alert_rule(self, rule_name: str) -> bool: """ Remove an alert rule. Args: rule_name: Name of rule to remove Returns: bool: True if rule was removed """ if rule_name in self.alert_rules: del self.alert_rules[rule_name] logger.info(f"Removed alert rule: {rule_name}") return True return False def add_notification_handler(self, handler: Callable): """ Add notification handler for alerts. Args: handler: Function to handle alert notifications """ self._notification_handlers.append(handler) logger.info(f"Added notification handler: {handler.__name__}") def add_suppression_rule(self, rule_name: str, suppressor: Callable): """ Add suppression rule to prevent false positives. Args: rule_name: Name of rule to suppress suppressor: Function that returns True if alert should be suppressed """ self._suppression_rules[rule_name] = suppressor logger.info(f"Added suppression rule for: {rule_name}") async def start_validation(self, interval: Optional[int] = None): """ Start alert validation task. Args: interval: Override default validation interval """ if self._validation_task is None: validation_interval = interval or self.validation_interval self._validation_task = asyncio.create_task(self._validate_alerts(validation_interval)) logger.info(f"Alert validation started with interval: {validation_interval}s") async def stop_validation(self): """Stop alert validation task.""" if self._validation_task: self._validation_task.cancel() try: await self._validation_task except asyncio.CancelledError: pass self._validation_task = None logger.info("Alert validation stopped") async def process_metric(self, metric_name: str, value: float, timestamp: Optional[datetime] = None): """ Process a metric and check against alert rules. Args: metric_name: Name of the metric value: Metric value timestamp: Optional timestamp for the metric """ timestamp = timestamp or datetime.utcnow() # Store metric for trend analysis self.metric_tracker[metric_name].append({ 'value': value, 'timestamp': timestamp }) # Keep metric history manageable if len(self.metric_tracker[metric_name]) > 1000: self.metric_tracker[metric_name] = self.metric_tracker[metric_name][-500:] # Check against alert rules for rule_name, rule in self.alert_rules.items(): if not rule.enabled: continue if self._should_check_alert(rule, metric_name): await self._evaluate_alert(rule, metric_name, value, timestamp) def _should_check_alert(self, rule: AlertRule, metric_name: str) -> bool: """Check if alert should be evaluated for this metric.""" # Simple matching - could be enhanced with regex or patterns return rule.condition in metric_name or metric_name in rule.condition async def _evaluate_alert(self, rule: AlertRule, metric_name: str, value: float, timestamp: datetime): """Evaluate alert condition and trigger if necessary.""" if not self._condition_met(rule.condition, value, rule.threshold): await self._clear_alert_if_active(rule.name, metric_name) return # Check cooldown period if self._is_in_cooldown(rule.name): return # Check suppression rules if self._should_suppress_alert(rule.name, metric_name, value): return # Start tracking potential alert await self._track_potential_alert(rule, metric_name, value, timestamp) def _condition_met(self, condition: str, value: float, threshold: float) -> bool: """Check if alert condition is met.""" try: # Parse condition (simple implementation) if '>' in condition: return value > threshold elif '<' in condition: return value < threshold elif '=' in condition: return abs(value - threshold) < 0.001 else: return False except Exception as e: logger.error(f"Error evaluating condition '{condition}': {e}") return False def _is_in_cooldown(self, rule_name: str) -> bool: """Check if rule is in cooldown period.""" if rule_name not in self.cooldown_tracker: return False rule = self.alert_rules[rule_name] last_alert = self.cooldown_tracker[rule_name] return (datetime.utcnow() - last_alert).total_seconds() < rule.cooldown def _should_suppress_alert(self, rule_name: str, metric_name: str, value: float) -> bool: """Check if alert should be suppressed.""" if rule_name not in self._suppression_rules: return False try: suppressor = self._suppression_rules[rule_name] return suppressor(metric_name, value, self.metric_tracker[metric_name]) except Exception as e: logger.error(f"Error in suppression rule for {rule_name}: {e}") return False async def _track_potential_alert(self, rule: AlertRule, metric_name: str, value: float, timestamp: datetime): """Track potential alert and trigger if duration threshold met.""" alert_key = f"{rule.name}_{metric_name}" # Check if we already have an active alert if alert_key in self.active_alerts: active_alert = self.active_alerts[alert_key] # Update current value active_alert.current_value = value active_alert.triggered_at = timestamp return # Create new alert alert = Alert( alert_id=str(uuid.uuid4()), rule_name=rule.name, metric_name=metric_name, current_value=value, threshold=rule.threshold, severity=rule.severity, message=self._generate_alert_message(rule, metric_name, value), triggered_at=timestamp, metadata={ 'rule_description': rule.description, 'tags': rule.tags, 'condition': rule.condition } ) # Store as active alert self.active_alerts[alert_key] = alert self.alert_statistics.total_alerts += 1 self.alert_statistics.active_alerts += 1 self.alert_statistics.alerts_by_severity[rule.severity] += 1 self.alert_statistics.alerts_by_rule[rule.name] += 1 # Check if duration threshold is met (immediate for now) if rule.duration <= 0: await self._trigger_alert(alert) async def _trigger_alert(self, alert: Alert): """Trigger an alert notification.""" try: # Send notifications for handler in self._notification_handlers: try: await handler(alert) except Exception as e: logger.error(f"Notification handler failed: {e}") # Set cooldown self.cooldown_tracker[alert.rule_name] = datetime.utcnow() # Add to history self.alert_history.append(alert) if len(self.alert_history) > self.max_history: self.alert_history = self.alert_history[-self.max_history // 2:] logger.warning(f"Alert triggered: {alert.rule_name} - {alert.message}") except Exception as e: logger.error(f"Failed to trigger alert: {e}") async def _clear_alert_if_active(self, rule_name: str, metric_name: str): """Clear alert if condition is no longer met.""" alert_key = f"{rule_name}_{metric_name}" if alert_key in self.active_alerts: alert = self.active_alerts.pop(alert_key) alert.resolved = True alert.resolved_at = datetime.utcnow() # Update statistics self.alert_statistics.active_alerts -= 1 self.alert_statistics.resolved_alerts += 1 # Add to history self.alert_history.append(alert) logger.info(f"Alert resolved: {alert.rule_name} - {alert.message}") def _generate_alert_message(self, rule: AlertRule, metric_name: str, value: float) -> str: """Generate alert message.""" if rule.description: return f"{rule.description}: {metric_name} = {value:.2f} (threshold: {rule.threshold})" else: return f"Alert triggered for {metric_name}: {value:.2f} (threshold: {rule.threshold})" async def acknowledge_alert(self, alert_id: str, acknowledged_by: str) -> bool: """ Acknowledge an alert. Args: alert_id: Alert ID to acknowledge acknowledged_by: User acknowledging the alert Returns: bool: True if alert was acknowledged """ # Check active alerts for alert in self.active_alerts.values(): if alert.alert_id == alert_id: alert.acknowledged = True alert.acknowledged_by = acknowledged_by alert.acknowledged_at = datetime.utcnow() self.alert_statistics.acknowledged_alerts += 1 logger.info(f"Alert acknowledged: {alert_id} by {acknowledged_by}") return True # Check alert history for alert in self.alert_history: if alert.alert_id == alert_id and not alert.resolved: alert.acknowledged = True alert.acknowledged_by = acknowledged_by alert.acknowledged_at = datetime.utcnow() self.alert_statistics.acknowledged_alerts += 1 logger.info(f"Alert acknowledged: {alert_id} by {acknowledged_by}") return True return False async def resolve_alert(self, alert_id: str) -> bool: """ Resolve an alert. Args: alert_id: Alert ID to resolve Returns: bool: True if alert was resolved """ # Check active alerts for alert_key, alert in list(self.active_alerts.items()): if alert.alert_id == alert_id: alert.resolved = True alert.resolved_at = datetime.utcnow() # Remove from active alerts self.active_alerts.pop(alert_key, None) # Update statistics self.alert_statistics.active_alerts -= 1 self.alert_statistics.resolved_alerts += 1 logger.info(f"Alert resolved: {alert_id}") return True return False async def _validate_alerts(self, interval: int): """Periodic validation of active alerts and conditions.""" while True: try: await asyncio.sleep(interval) await self._validate_active_alerts() except asyncio.CancelledError: break except Exception as e: logger.error(f"Alert validation error: {e}") async def _validate_active_alerts(self): """Validate active alerts and update statistics.""" current_time = datetime.utcnow() # Update average resolution time resolved_alerts = [alert for alert in self.alert_history if alert.resolved and alert.resolved_at] if resolved_alerts: total_resolution_time = sum( (alert.resolved_at - alert.triggered_at).total_seconds() for alert in resolved_alerts ) self.alert_statistics.average_resolution_time = total_resolution_time / len(resolved_alerts) # Check for stale alerts stale_threshold = timedelta(hours=24) for alert_key, alert in list(self.active_alerts.items()): if current_time - alert.triggered_at > stale_threshold: logger.warning(f"Stale alert detected: {alert.alert_id}") # Optionally auto-resolve stale alerts # await self.resolve_alert(alert.alert_id) def get_active_alerts(self) -> List[Alert]: """Get all active alerts.""" return list(self.active_alerts.values()) def get_alert_history(self, limit: int = 100, severity: Optional[str] = None) -> List[Alert]: """Get alert history.""" history = self.alert_history.copy() # Filter by severity if specified if severity: history = [alert for alert in history if alert.severity == severity] # Sort by triggered_at descending and limit history.sort(key=lambda x: x.triggered_at, reverse=True) return history[:limit] def get_alert_statistics(self) -> AlertStatistics: """Get alert system statistics.""" # Update active count self.alert_statistics.active_alerts = len(self.active_alerts) return self.alert_statistics def get_rule_statistics(self) -> Dict[str, Dict[str, Any]]: """Get statistics for each alert rule.""" rule_stats = {} for rule_name, rule in self.alert_rules.items(): rule_alerts = [alert for alert in self.alert_history if alert.rule_name == rule_name] rule_stats[rule_name] = { 'total_alerts': len(rule_alerts), 'active_alerts': len([alert for alert in self.active_alerts.values() if alert.rule_name == rule_name]), 'last_triggered': max([alert.triggered_at for alert in rule_alerts]) if rule_alerts else None, 'severity_distribution': defaultdict(int), 'enabled': rule.enabled } # Severity distribution for alert in rule_alerts: rule_stats[rule_name]['severity_distribution'][alert.severity] += 1 return rule_stats async def get_metric_analysis(self, metric_name: str, hours: int = 24) -> Dict[str, Any]: """Get analysis of a specific metric.""" if metric_name not in self.metric_tracker: return {'error': f'Metric {metric_name} not found'} cutoff_time = datetime.utcnow() - timedelta(hours=hours) recent_metrics = [ m for m in self.metric_tracker[metric_name] if m['timestamp'] > cutoff_time ] if not recent_metrics: return {'error': f'No recent data for metric {metric_name}'} values = [m['value'] for m in recent_metrics] return { 'metric_name': metric_name, 'period_hours': hours, 'data_points': len(recent_metrics), 'current_value': values[-1], 'min_value': min(values), 'max_value': max(values), 'avg_value': sum(values) / len(values), 'trend': self._calculate_trend(values), 'alerts_triggered': len([ alert for alert in self.alert_history if alert.metric_name == metric_name and alert.triggered_at > cutoff_time ]) } def _calculate_trend(self, values: List[float]) -> str: """Calculate trend from values.""" if len(values) < 2: return 'insufficient_data' # Simple trend calculation recent_avg = sum(values[-10:]) / min(len(values), 10) older_avg = sum(values[:10]) / min(len(values), 10) if recent_avg > older_avg * 1.1: return 'increasing' elif recent_avg < older_avg * 0.9: return 'decreasing' else: return 'stable' async def reset_statistics(self): """Reset alert statistics.""" self.alert_statistics = AlertStatistics() self.alert_history.clear() self.active_alerts.clear() self.cooldown_tracker.clear() logger.info("Alert statistics reset") # Factory function def create_smart_alert_manager(validation_interval: int = 30, max_history: int = 10000) -> SmartAlertManager: """ Create a smart alert manager instance. Args: validation_interval: Validation interval in seconds max_history: Maximum alert history size Returns: SmartAlertManager: Configured manager """ return SmartAlertManager(validation_interval, max_history)