"""Alert service for managing alert rules and checking thresholds using Redis.""" import json import logging import uuid from datetime import datetime, timedelta, timezone from typing import Any, Optional import sqlalchemy as sa from extensions.ext_database import db from extensions.ext_redis import redis_client, redis_fallback from models.model import App from services.notification_service import NotificationService logger = logging.getLogger(__name__) class AlertService: """Service for managing application alert rules stored in Redis.""" # Redis Key prefixes RULE_PREFIX = "alert:rule" # Alert rule data RULE_LIST_PREFIX = "alert:rules" # List of rule IDs per app LOG_PREFIX = "alert:log" # Alert logs LOG_LIST_PREFIX = "alert:logs" # List of log IDs per rule ENABLED_RULES_PREFIX = "alert:enabled" # Set of enabled rule IDs CONFIG_PREFIX = "alert:config" # Notification channel config per app # Data retention time (90 days) RETENTION_DAYS = 90 RETENTION_SECONDS = RETENTION_DAYS * 24 * 60 * 60 # Maximum log entries per rule MAX_LOG_ENTRIES = 1000 @classmethod def _get_rule_key(cls, rule_id: str) -> str: """Get Redis key for rule data.""" return f"{cls.RULE_PREFIX}:{rule_id}" @classmethod def _get_rule_list_key(cls, app_id: str) -> str: """Get Redis key for app's rule list.""" return f"{cls.RULE_LIST_PREFIX}:{app_id}" @classmethod def _get_log_key(cls, log_id: str) -> str: """Get Redis key for log data.""" return f"{cls.LOG_PREFIX}:{log_id}" @classmethod def _get_config_key(cls, app_id: str) -> str: """Get Redis key for notification config.""" return f"{cls.CONFIG_PREFIX}:{app_id}" @classmethod def _get_log_list_key(cls, rule_id: str) -> str: """Get Redis key for rule's log list.""" return f"{cls.LOG_LIST_PREFIX}:{rule_id}" @classmethod @redis_fallback(default_return=[]) def get_alert_rules(cls, app_id: str) -> list[dict[str, Any]]: """Get all alert rules for an app. Args: app_id: Application ID Returns: List of alert rule dictionaries """ rule_list_key = cls._get_rule_list_key(app_id) rule_ids = redis_client.smembers(rule_list_key) rules = [] for rule_id_bytes in rule_ids: rule_id = rule_id_bytes.decode("utf-8") if isinstance(rule_id_bytes, bytes) else rule_id_bytes rule = cls.get_alert_rule(rule_id) if rule: rules.append(rule) return rules @classmethod @redis_fallback(default_return=[]) def get_all_enabled_rules(cls) -> list[dict[str, Any]]: """Get all enabled alert rules across all apps. Returns: List of enabled alert rule dictionaries """ enabled_rule_ids = redis_client.smembers(cls.ENABLED_RULES_PREFIX) rules = [] for rule_id_bytes in enabled_rule_ids: rule_id = rule_id_bytes.decode("utf-8") if isinstance(rule_id_bytes, bytes) else rule_id_bytes rule = cls.get_alert_rule(rule_id) if rule and rule.get("is_enabled"): rules.append(rule) return rules @classmethod @redis_fallback(default_return=None) def get_alert_rule(cls, rule_id: str) -> Optional[dict[str, Any]]: """Get a single alert rule. Args: rule_id: Alert rule ID Returns: Alert rule dictionary or None if not found """ rule_key = cls._get_rule_key(rule_id) rule_data = redis_client.hgetall(rule_key) if not rule_data: return None # Decode and parse rule data rule = {} for key, value in rule_data.items(): key_str = key.decode("utf-8") if isinstance(key, bytes) else key value_str = value.decode("utf-8") if isinstance(value, bytes) else value # Parse JSON fields if key_str in ["notification_channels"]: try: rule[key_str] = json.loads(value_str) if value_str else {} except json.JSONDecodeError: rule[key_str] = {} elif key_str in ["is_enabled"]: rule[key_str] = value_str.lower() == "true" elif key_str in ["threshold_value", "check_interval_minutes", "consecutive_triggers", "trigger_count", "consecutive_trigger_count"]: try: rule[key_str] = float(value_str) if "." in value_str else int(value_str) except (ValueError, AttributeError): rule[key_str] = 0 else: rule[key_str] = value_str return rule @classmethod @redis_fallback(default_return=None) def create_alert_rule(cls, app_id: str, data: dict[str, Any], created_by: str) -> Optional[dict[str, Any]]: """Create a new alert rule. Args: app_id: Application ID data: Alert rule data created_by: User ID who created the rule Returns: Created alert rule dictionary """ # Get app to validate and get tenant_id app = db.session.query(App).filter(App.id == app_id).first() if not app: raise ValueError(f"App {app_id} not found") # Generate rule ID rule_id = str(uuid.uuid4()) # Build rule data now = datetime.now(timezone.utc).isoformat() rule = { "id": rule_id, "tenant_id": app.tenant_id, "app_id": app_id, "name": data["name"], "description": data.get("description", ""), "is_enabled": data.get("is_enabled", True), "metric_type": data["metric_type"], "operator": data["operator"], "threshold_value": data["threshold_value"], "check_interval_minutes": data.get("check_interval_minutes", 5), "consecutive_triggers": data.get("consecutive_triggers", 2), "notification_channels": json.dumps(data["notification_channels"]), "created_by": created_by, "created_at": now, "updated_at": now, "last_triggered_at": "", "trigger_count": 0, "consecutive_trigger_count": 0, } # Store rule data rule_key = cls._get_rule_key(rule_id) redis_client.hset( rule_key, mapping={k: str(v) if not isinstance(v, str) else v for k, v in rule.items()}, ) redis_client.expire(rule_key, cls.RETENTION_SECONDS) # Add to app's rule list rule_list_key = cls._get_rule_list_key(app_id) redis_client.sadd(rule_list_key, rule_id) redis_client.expire(rule_list_key, cls.RETENTION_SECONDS) # Add to enabled rules set if enabled if rule["is_enabled"]: redis_client.sadd(cls.ENABLED_RULES_PREFIX, rule_id) redis_client.expire(cls.ENABLED_RULES_PREFIX, cls.RETENTION_SECONDS) logger.info(f"Created alert rule {rule_id} for app {app_id}") # Return parsed rule return cls.get_alert_rule(rule_id) @classmethod @redis_fallback(default_return=None) def update_alert_rule(cls, rule_id: str, data: dict[str, Any]) -> Optional[dict[str, Any]]: """Update an existing alert rule. Args: rule_id: Alert rule ID data: Updated alert rule data Returns: Updated alert rule dictionary """ rule = cls.get_alert_rule(rule_id) if not rule: raise ValueError(f"Alert rule {rule_id} not found") # Update fields rule_key = cls._get_rule_key(rule_id) updates = {} if "name" in data: updates["name"] = data["name"] if "description" in data: updates["description"] = data["description"] if "is_enabled" in data: updates["is_enabled"] = str(data["is_enabled"]) # Update enabled rules set if data["is_enabled"]: redis_client.sadd(cls.ENABLED_RULES_PREFIX, rule_id) else: redis_client.srem(cls.ENABLED_RULES_PREFIX, rule_id) if "metric_type" in data: updates["metric_type"] = data["metric_type"] if "operator" in data: updates["operator"] = data["operator"] if "threshold_value" in data: updates["threshold_value"] = str(data["threshold_value"]) if "check_interval_minutes" in data: updates["check_interval_minutes"] = str(data["check_interval_minutes"]) if "consecutive_triggers" in data: updates["consecutive_triggers"] = str(data["consecutive_triggers"]) if "notification_channels" in data: updates["notification_channels"] = json.dumps(data["notification_channels"]) updates["updated_at"] = datetime.now(timezone.utc).isoformat() # Apply updates if updates: redis_client.hset(rule_key, mapping=updates) redis_client.expire(rule_key, cls.RETENTION_SECONDS) logger.info(f"Updated alert rule {rule_id}") return cls.get_alert_rule(rule_id) @classmethod @redis_fallback(default_return=None) def delete_alert_rule(cls, rule_id: str) -> None: """Delete an alert rule. Args: rule_id: Alert rule ID """ rule = cls.get_alert_rule(rule_id) if not rule: raise ValueError(f"Alert rule {rule_id} not found") # Remove from app's rule list rule_list_key = cls._get_rule_list_key(rule["app_id"]) redis_client.srem(rule_list_key, rule_id) # Remove from enabled rules set redis_client.srem(cls.ENABLED_RULES_PREFIX, rule_id) # Delete rule data rule_key = cls._get_rule_key(rule_id) redis_client.delete(rule_key) # Delete associated logs log_list_key = cls._get_log_list_key(rule_id) log_ids = redis_client.smembers(log_list_key) for log_id_bytes in log_ids: log_id = log_id_bytes.decode("utf-8") if isinstance(log_id_bytes, bytes) else log_id_bytes log_key = cls._get_log_key(log_id) redis_client.delete(log_key) redis_client.delete(log_list_key) logger.info(f"Deleted alert rule {rule_id}") @staticmethod def check_threshold(value: float | None, operator: str, threshold: float) -> bool: """Check if a value exceeds a threshold. Args: value: Current metric value (None means no data) operator: Comparison operator ('>', '<', '>=', '<=', '=') threshold: Threshold value Returns: True if threshold is exceeded, False otherwise """ if value is None: return False if operator == ">": return value > threshold elif operator == "<": return value < threshold elif operator == ">=": return value >= threshold elif operator == "<=": return value <= threshold elif operator == "=": return value == threshold else: logger.warning(f"Unknown operator: {operator}") return False @staticmethod def get_metric_value(app_id: str, metric_type: str, time_range_minutes: int = 5) -> float | None: """Get current metric value for an app. Args: app_id: Application ID metric_type: Type of metric ('error_rate', 'response_time_p95', 'response_time_p99', 'avg_response_time') time_range_minutes: Time range to calculate metric (default 5 minutes) Returns: Metric value or None if no data """ try: # Calculate time range end_time = datetime.now(timezone.utc) start_time = end_time - timedelta(minutes=time_range_minutes) # Build base query from models.model import Message base_query = db.session.query(Message).filter( Message.app_id == app_id, Message.created_at >= start_time, Message.created_at <= end_time ) if metric_type == "error_rate": # Calculate error rate total_count = base_query.count() if total_count == 0: return None error_count = base_query.filter( sa.or_( Message.status == "error", Message.status == "stopped", ) ).count() return (error_count / total_count) * 100.0 elif metric_type == "avg_response_time": # Calculate average response time from message_metadata JSON # Since message_metadata is stored as Text, we need to parse it messages = base_query.all() if not messages: return None latencies = [] for msg in messages: metadata = msg.message_metadata_dict if metadata and "latency" in metadata: try: latencies.append(float(metadata["latency"])) except (ValueError, TypeError): continue if not latencies: return None # Return average latency in milliseconds return sum(latencies) / len(latencies) * 1000 elif metric_type == "response_time_p95": # Calculate P95 response time from message_metadata JSON messages = base_query.all() if not messages: return None latencies = [] for msg in messages: metadata = msg.message_metadata_dict if metadata and "latency" in metadata: try: latencies.append(float(metadata["latency"]) * 1000) # Convert to ms except (ValueError, TypeError): continue if not latencies: return None # Calculate P95 latencies.sort() p95_index = int(len(latencies) * 0.95) return latencies[p95_index] if p95_index < len(latencies) else latencies[-1] elif metric_type == "response_time_p99": # Calculate P99 response time from message_metadata JSON messages = base_query.all() if not messages: return None latencies = [] for msg in messages: metadata = msg.message_metadata_dict if metadata and "latency" in metadata: try: latencies.append(float(metadata["latency"]) * 1000) # Convert to ms except (ValueError, TypeError): continue if not latencies: return None # Calculate P99 latencies.sort() p99_index = int(len(latencies) * 0.99) return latencies[p99_index] if p99_index < len(latencies) else latencies[-1] else: logger.warning(f"Unknown metric type: {metric_type}") return None except Exception: logger.exception(f"Error getting metric value for app {app_id}, metric {metric_type}") return None @classmethod @redis_fallback(default_return=None) def trigger_alert(cls, rule: dict[str, Any], metric_value: float) -> None: """Trigger an alert and send notifications. Args: rule: Alert rule dictionary metric_value: Current value of the metric """ try: # Get app name app = db.session.query(App).filter(App.id == rule["app_id"]).first() app_name = app.name if app else "Unknown App" # Send notifications notification_results = NotificationService.send_alert_notification( channels=rule["notification_channels"], app_name=app_name, metric_name=rule["metric_type"], metric_value=metric_value, threshold_value=rule["threshold_value"], operator=rule["operator"], ) # Create alert log log_id = str(uuid.uuid4()) now = datetime.now(timezone.utc).isoformat() log_data = { "id": log_id, "alert_rule_id": rule["id"], "metric_value": metric_value, "threshold_value": rule["threshold_value"], "notification_status": json.dumps(notification_results), "error_message": "", "triggered_at": now, } # Store log data log_key = cls._get_log_key(log_id) redis_client.hset( log_key, mapping={k: str(v) if not isinstance(v, str) else v for k, v in log_data.items()}, ) redis_client.expire(log_key, cls.RETENTION_SECONDS) # Add to rule's log list (sorted set by timestamp) log_list_key = cls._get_log_list_key(rule["id"]) timestamp = datetime.now(timezone.utc).timestamp() redis_client.zadd(log_list_key, {log_id: timestamp}) redis_client.expire(log_list_key, cls.RETENTION_SECONDS) # Cleanup old logs cls._cleanup_old_logs(rule["id"]) # Update rule statistics rule_key = cls._get_rule_key(rule["id"]) redis_client.hset( rule_key, mapping={ "last_triggered_at": now, "trigger_count": str(int(rule.get("trigger_count", 0)) + 1), }, ) redis_client.expire(rule_key, cls.RETENTION_SECONDS) logger.info( f"Alert triggered for rule {rule['id']}, app {rule['app_id']}, " f"metric {rule['metric_type']}={metric_value}, threshold {rule['operator']} {rule['threshold_value']}" ) except Exception as e: logger.exception(f"Error triggering alert for rule {rule['id']}") # Try to log the error try: log_id = str(uuid.uuid4()) now = datetime.now(timezone.utc).isoformat() log_data = { "id": log_id, "alert_rule_id": rule["id"], "metric_value": metric_value, "threshold_value": rule["threshold_value"], "notification_status": json.dumps({"error": "failed"}), "error_message": str(e), "triggered_at": now, } log_key = cls._get_log_key(log_id) redis_client.hset( log_key, mapping={k: str(v) if not isinstance(v, str) else v for k, v in log_data.items()}, ) redis_client.expire(log_key, cls.RETENTION_SECONDS) except Exception: logger.exception("Failed to log alert error") @classmethod @redis_fallback(default_return=[]) def get_alert_logs(cls, rule_id: str, limit: int = 100) -> list[dict[str, Any]]: """Get alert logs for a rule. Args: rule_id: Alert rule ID limit: Maximum number of logs to return Returns: List of alert log dictionaries """ log_list_key = cls._get_log_list_key(rule_id) # Get log IDs sorted by timestamp (descending) log_ids = redis_client.zrevrange(log_list_key, 0, limit - 1) logs = [] for log_id_bytes in log_ids: log_id = log_id_bytes.decode("utf-8") if isinstance(log_id_bytes, bytes) else log_id_bytes log_key = cls._get_log_key(log_id) log_data = redis_client.hgetall(log_key) if not log_data: continue # Decode and parse log data log = {} for key, value in log_data.items(): key_str = key.decode("utf-8") if isinstance(key, bytes) else key value_str = value.decode("utf-8") if isinstance(value, bytes) else value if key_str == "notification_status": try: log[key_str] = json.loads(value_str) if value_str else {} except json.JSONDecodeError: log[key_str] = {} elif key_str in ["metric_value", "threshold_value"]: try: log[key_str] = float(value_str) except (ValueError, AttributeError): log[key_str] = 0.0 else: log[key_str] = value_str logs.append(log) return logs @classmethod @redis_fallback(default_return=None) def _cleanup_old_logs(cls, rule_id: str) -> None: """Clean up old logs for a rule, keeping only the most recent MAX_LOG_ENTRIES. Args: rule_id: Alert rule ID """ log_list_key = cls._get_log_list_key(rule_id) # Get count of logs log_count = redis_client.zcard(log_list_key) if log_count > cls.MAX_LOG_ENTRIES: # Remove oldest logs logs_to_remove = log_count - cls.MAX_LOG_ENTRIES # Get oldest log IDs old_log_ids = redis_client.zrange(log_list_key, 0, logs_to_remove - 1) # Delete old log entries for log_id_bytes in old_log_ids: log_id = log_id_bytes.decode("utf-8") if isinstance(log_id_bytes, bytes) else log_id_bytes log_key = cls._get_log_key(log_id) redis_client.delete(log_key) # Remove from sorted set redis_client.zremrangebyrank(log_list_key, 0, logs_to_remove - 1) @classmethod @redis_fallback(default_return={}) def get_notification_config(cls, app_id: str) -> dict: """Get notification channel configuration for an app. Args: app_id: Application ID Returns: Dictionary with channel configurations """ config_key = cls._get_config_key(app_id) config_data = redis_client.get(config_key) if not config_data: return {"email": {}, "dingtalk": {}, "wecom": {}} config_str = config_data.decode("utf-8") if isinstance(config_data, bytes) else config_data return json.loads(config_str) @classmethod @redis_fallback(default_return={}) def save_notification_config(cls, app_id: str, config: dict) -> dict: """Save notification channel configuration for an app. Args: app_id: Application ID config: Configuration dictionary with channels Returns: Saved configuration Raises: ValueError: If config contains invalid channels """ config_key = cls._get_config_key(app_id) # Validate config structure valid_channels = ["email", "dingtalk", "wecom"] for channel in config.keys(): if channel not in valid_channels: raise ValueError(f"Invalid channel: {channel}") # Save config with 90 days expiration redis_client.setex( config_key, cls.RETENTION_SECONDS, json.dumps(config), ) return config