dify-1.8.1/api/services/alert_service.py

"""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