第302集云服务器ECS告警系统架构实战：智能监控、预警机制与企业级运维自动化解决方案

class DynamicThreshold:
    def __init__(self, window_size=24, sensitivity=2.0):
        self.window_size = window_size  # 时间窗口大小（小时）
        self.sensitivity = sensitivity  # 敏感度参数

    def calculate_threshold(self, historical_data):
        """计算动态阈值"""
        # 计算历史数据的统计特征
        mean = np.mean(historical_data)
        std = np.std(historical_data)

        # 基于3-sigma原则计算阈值
        upper_threshold = mean + self.sensitivity * std
        lower_threshold = mean - self.sensitivity * std

        return upper_threshold, lower_threshold

    def detect_anomaly(self, current_value, historical_data):
        """异常检测"""
        upper_threshold, lower_threshold = self.calculate_threshold(historical_data)

        if current_value > upper_threshold:
            return "high_anomaly"
        elif current_value < lower_threshold:
            return "low_anomaly"
        else:
            return "normal"

class CompositeAlertRule:
    def __init__(self, rule_id, name, conditions, logic_operator="AND"):
        self.rule_id = rule_id
        self.name = name
        self.conditions = conditions  # 条件列表
        self.logic_operator = logic_operator  # 逻辑操作符

    def evaluate(self, metrics_data):
        """评估复合告警规则"""
        results = []

        for condition in self.conditions:
            metric_name = condition['metric']
            operator = condition['operator']
            threshold = condition['threshold']
            time_window = condition.get('time_window', 300)  # 默认5分钟

            # 获取指标数据
            metric_values = self.get_metric_values(metric_name, time_window)

            # 评估条件
            if operator == '>':
                result = any(value > threshold for value in metric_values)
            elif operator == '<':
                result = any(value < threshold for value in metric_values)
            elif operator == '>=':
                result = any(value >= threshold for value in metric_values)
            elif operator == '<=':
                result = any(value <= threshold for value in metric_values)
            elif operator == '==':
                result = any(value == threshold for value in metric_values)
            elif operator == '!=':
                result = any(value != threshold for value in metric_values)
            else:
                result = False

            results.append(result)

        # 根据逻辑操作符组合结果
        if self.logic_operator == "AND":
            return all(results)
        elif self.logic_operator == "OR":
            return any(results)
        else:
            return False

    def get_metric_values(self, metric_name, time_window):
        """获取指标值"""
        # 从监控系统获取指标数据
        # 这里简化实现，实际需要连接监控数据库
        return []

import numpy as np
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

class IntelligentAlertDetector:
    def __init__(self):
        self.isolation_forest = IsolationForest(contamination=0.1)
        self.scaler = StandardScaler()
        self.is_trained = False

    def train(self, historical_data):
        """训练异常检测模型"""
        # 数据预处理
        scaled_data = self.scaler.fit_transform(historical_data)

        # 训练模型
        self.isolation_forest.fit(scaled_data)
        self.is_trained = True

    def detect_anomaly(self, current_data):
        """检测异常"""
        if not self.is_trained:
            return False

        # 数据预处理
        scaled_data = self.scaler.transform([current_data])

        # 异常检测
        anomaly_score = self.isolation_forest.decision_function(scaled_data)[0]
        is_anomaly = self.isolation_forest.predict(scaled_data)[0] == -1

        return {
            'is_anomaly': is_anomaly,
            'anomaly_score': anomaly_score,
            'confidence': abs(anomaly_score)
        }

class AlertSuppression:
    def __init__(self):
        self.suppression_rules = []
        self.active_suppressions = {}

    def add_suppression_rule(self, rule):
        """添加抑制规则"""
        self.suppression_rules.append(rule)

    def check_suppression(self, alert):
        """检查告警是否需要抑制"""
        for rule in self.suppression_rules:
            if self.match_rule(alert, rule):
                suppression_key = self.generate_suppression_key(rule)

                if suppression_key not in self.active_suppressions:
                    self.active_suppressions[suppression_key] = {
                        'start_time': time.time(),
                        'rule': rule,
                        'suppressed_alerts': []
                    }

                self.active_suppressions[suppression_key]['suppressed_alerts'].append(alert)
                return True

        return False

    def match_rule(self, alert, rule):
        """匹配抑制规则"""
        # 检查告警属性是否匹配规则条件
        for condition in rule['conditions']:
            alert_value = getattr(alert, condition['field'])
            if not self.evaluate_condition(alert_value, condition):
                return False
        return True

class AlertAggregation:
    def __init__(self, aggregation_window=300):
        self.aggregation_window = aggregation_window  # 聚合时间窗口（秒）
        self.alert_groups = {}

    def aggregate_alerts(self, alert):
        """聚合告警"""
        group_key = self.generate_group_key(alert)
        current_time = time.time()

        if group_key not in self.alert_groups:
            self.alert_groups[group_key] = {
                'alerts': [],
                'first_alert_time': current_time,
                'last_alert_time': current_time,
                'count': 0
            }

        group = self.alert_groups[group_key]
        group['alerts'].append(alert)
        group['last_alert_time'] = current_time
        group['count'] += 1

        # 检查是否需要发送聚合告警
        if self.should_send_aggregated_alert(group):
            return self.create_aggregated_alert(group)

        return None

    def should_send_aggregated_alert(self, group):
        """判断是否应该发送聚合告警"""
        time_diff = group['last_alert_time'] - group['first_alert_time']

        # 时间窗口内告警数量达到阈值
        if group['count'] >= 5 and time_diff >= self.aggregation_window:
            return True

        # 告警频率过高
        if group['count'] >= 10:
            return True

        return False

class NotificationChannel:
    def __init__(self, channel_type, config):
        self.channel_type = channel_type
        self.config = config

    def send_notification(self, alert, recipients):
        """发送通知"""
        if self.channel_type == 'email':
            return self.send_email(alert, recipients)
        elif self.channel_type == 'sms':
            return self.send_sms(alert, recipients)
        elif self.channel_type == 'webhook':
            return self.send_webhook(alert, recipients)
        elif self.channel_type == 'dingtalk':
            return self.send_dingtalk(alert, recipients)
        else:
            raise ValueError(f"Unsupported channel type: {self.channel_type}")

    def send_email(self, alert, recipients):
        """发送邮件通知"""
        subject = f"[{alert.severity}] {alert.title}"
        body = self.format_email_body(alert)

        # 使用SMTP发送邮件
        # 实际实现需要配置SMTP服务器
        return True

    def send_sms(self, alert, recipients):
        """发送短信通知"""
        message = f"[{alert.severity}] {alert.title}: {alert.description}"

        # 调用短信服务API
        # 实际实现需要集成短信服务商
        return True

    def send_webhook(self, alert, recipients):
        """发送Webhook通知"""
        payload = {
            'alert_id': alert.alert_id,
            'severity': alert.severity,
            'title': alert.title,
            'description': alert.description,
            'timestamp': alert.timestamp,
            'instance_id': alert.instance_id
        }

        # 发送HTTP POST请求
        # 实际实现需要处理重试、超时等
        return True

    def format_email_body(self, alert):
        """格式化邮件内容"""
        return f"""
        告警详情：
        标题：{alert.title}
        描述：{alert.description}
        级别：{alert.severity}
        时间：{alert.timestamp}
        实例ID：{alert.instance_id}

        请及时处理此告警。
        """

class AutoRepairEngine:
    def __init__(self):
        self.repair_strategies = {}
        self.repair_history = []

    def register_repair_strategy(self, strategy_name, strategy_func):
        """注册修复策略"""
        self.repair_strategies[strategy_name] = strategy_func

    def execute_repair(self, alert):
        """执行自动修复"""
        # 分析告警类型
        repair_strategy = self.analyze_alert(alert)

        if repair_strategy:
            try:
                # 执行修复策略
                result = self.repair_strategies[repair_strategy](alert)

                # 记录修复历史
                self.repair_history.append({
                    'alert_id': alert.alert_id,
                    'strategy': repair_strategy,
                    'result': result,
                    'timestamp': time.time()
                })

                return result
            except Exception as e:
                # 修复失败，记录错误
                self.log_repair_error(alert, repair_strategy, str(e))
                return False

        return False

    def analyze_alert(self, alert):
        """分析告警，确定修复策略"""
        # 基于告警类型和指标确定修复策略
        if 'cpu' in alert.metric_name.lower():
            if alert.value > 90:
                return 'scale_up'
            elif alert.value < 10:
                return 'scale_down'
        elif 'memory' in alert.metric_name.lower():
            if alert.value > 90:
                return 'restart_service'
        elif 'disk' in alert.metric_name.lower():
            if alert.value > 90:
                return 'cleanup_disk'
        elif 'network' in alert.metric_name.lower():
            if alert.value > 1000:  # 延迟过高
                return 'restart_network'

        return None

class RepairStrategies:
    @staticmethod
    def scale_up(alert):
        """扩容策略"""
        # 增加ECS实例数量或规格
        # 实际实现需要调用云服务API
        return True

    @staticmethod
    def scale_down(alert):
        """缩容策略"""
        # 减少ECS实例数量或规格
        return True

    @staticmethod
    def restart_service(alert):
        """重启服务策略"""
        # 重启相关服务
        return True

    @staticmethod
    def cleanup_disk(alert):
        """磁盘清理策略"""
        # 清理临时文件、日志文件等
        return True

    @staticmethod
    def restart_network(alert):
        """网络重启策略"""
        # 重启网络服务
        return True

class FailoverManager:
    def __init__(self):
        self.failover_policies = {}
        self.active_failovers = {}

    def register_failover_policy(self, service_name, policy):
        """注册故障转移策略"""
        self.failover_policies[service_name] = policy

    def execute_failover(self, alert):
        """执行故障转移"""
        service_name = self.extract_service_name(alert)

        if service_name in self.failover_policies:
            policy = self.failover_policies[service_name]

            # 检查是否已经执行过故障转移
            failover_key = f"{service_name}_{alert.instance_id}"
            if failover_key in self.active_failovers:
                return False

            # 执行故障转移
            result = self.perform_failover(service_name, policy, alert)

            if result:
                self.active_failovers[failover_key] = {
                    'timestamp': time.time(),
                    'policy': policy,
                    'alert': alert
                }

            return result

        return False

    def perform_failover(self, service_name, policy, alert):
        """执行具体的故障转移操作"""
        if policy['type'] == 'load_balancer':
            # 从负载均衡器中移除故障实例
            return self.remove_from_load_balancer(alert.instance_id)
        elif policy['type'] == 'dns':
            # 修改DNS记录，指向备用实例
            return self.update_dns_record(service_name, alert.instance_id)
        elif policy['type'] == 'container':
            # 重启容器或迁移到其他节点
            return self.restart_container(alert.instance_id)

        return False

    def remove_from_load_balancer(self, instance_id):
        """从负载均衡器中移除实例"""
        # 调用负载均衡器API移除实例
        return True

    def update_dns_record(self, service_name, instance_id):
        """更新DNS记录"""
        # 更新DNS记录指向备用实例
        return True

    def restart_container(self, instance_id):
        """重启容器"""
        # 重启容器服务
        return True

class QueryOptimizer:
    def __init__(self):
        self.query_cache = {}
        self.cache_ttl = 300  # 缓存TTL（秒）

    def optimize_query(self, query):
        """优化查询"""
        # 查询重写
        optimized_query = self.rewrite_query(query)

        # 添加索引提示
        optimized_query = self.add_index_hints(optimized_query)

        # 查询缓存
        cache_key = self.generate_cache_key(optimized_query)
        if cache_key in self.query_cache:
            cached_result = self.query_cache[cache_key]
            if time.time() - cached_result['timestamp'] < self.cache_ttl:
                return cached_result['data']

        # 执行查询
        result = self.execute_query(optimized_query)

        # 缓存结果
        self.query_cache[cache_key] = {
            'data': result,
            'timestamp': time.time()
        }

        return result

    def rewrite_query(self, query):
        """查询重写"""
        # 移除不必要的字段
        # 优化JOIN顺序
        # 添加LIMIT限制
        return query

    def add_index_hints(self, query):
        """添加索引提示"""
        # 根据查询条件添加合适的索引提示
        return query

class DataAggregator:
    def __init__(self):
        self.aggregation_cache = {}

    def pre_aggregate_data(self, time_window, metrics):
        """预聚合数据"""
        # 按时间窗口预聚合指标数据
        # 减少实时查询的计算量
        pass

    def get_aggregated_data(self, time_range, metrics, aggregation_type):
        """获取聚合数据"""
        cache_key = f"{time_range}_{metrics}_{aggregation_type}"

        if cache_key in self.aggregation_cache:
            return self.aggregation_cache[cache_key]

        # 计算聚合数据
        aggregated_data = self.calculate_aggregation(time_range, metrics, aggregation_type)

        # 缓存结果
        self.aggregation_cache[cache_key] = aggregated_data

        return aggregated_data

class SystemMonitor:
    def __init__(self):
        self.metrics_collector = MetricsCollector()
        self.performance_analyzer = PerformanceAnalyzer()

    def monitor_system_performance(self):
        """监控系统性能"""
        # 收集系统指标
        cpu_usage = self.metrics_collector.get_cpu_usage()
        memory_usage = self.metrics_collector.get_memory_usage()
        disk_usage = self.metrics_collector.get_disk_usage()
        network_usage = self.metrics_collector.get_network_usage()

        # 分析性能瓶颈
        bottlenecks = self.performance_analyzer.analyze_bottlenecks({
            'cpu': cpu_usage,
            'memory': memory_usage,
            'disk': disk_usage,
            'network': network_usage
        })

        # 生成优化建议
        recommendations = self.generate_recommendations(bottlenecks)

        return {
            'metrics': {
                'cpu': cpu_usage,
                'memory': memory_usage,
                'disk': disk_usage,
                'network': network_usage
            },
            'bottlenecks': bottlenecks,
            'recommendations': recommendations
        }

    def generate_recommendations(self, bottlenecks):
        """生成优化建议"""
        recommendations = []

        for bottleneck in bottlenecks:
            if bottleneck['type'] == 'cpu':
                recommendations.append({
                    'type': 'cpu_optimization',
                    'description': 'CPU使用率过高，建议优化算法或增加CPU资源',
                    'priority': 'high'
                })
            elif bottleneck['type'] == 'memory':
                recommendations.append({
                    'type': 'memory_optimization',
                    'description': '内存使用率过高，建议优化内存使用或增加内存',
                    'priority': 'high'
                })
            elif bottleneck['type'] == 'disk':
                recommendations.append({
                    'type': 'disk_optimization',
                    'description': '磁盘使用率过高，建议清理数据或扩容磁盘',
                    'priority': 'medium'
                })
            elif bottleneck['type'] == 'network':
                recommendations.append({
                    'type': 'network_optimization',
                    'description': '网络延迟过高，建议优化网络配置',
                    'priority': 'medium'
                })

        return recommendations

class LogManager:
    def __init__(self):
        self.log_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
        self.log_formatters = {
            'standard': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            'detailed': '%(asctime)s - %(name)s - %(levelname)s - %(filename)s:%(lineno)d - %(message)s'
        }

    def setup_logging(self, log_config):
        """设置日志配置"""
        import logging
        import logging.handlers

        # 创建日志记录器
        logger = logging.getLogger('alert_system')
        logger.setLevel(getattr(logging, log_config['level']))

        # 创建文件处理器
        file_handler = logging.handlers.RotatingFileHandler(
            log_config['file_path'],
            maxBytes=log_config['max_bytes'],
            backupCount=log_config['backup_count']
        )

        # 创建控制台处理器
        console_handler = logging.StreamHandler()

        # 设置格式器
        formatter = logging.Formatter(self.log_formatters[log_config['format']])
        file_handler.setFormatter(formatter)
        console_handler.setFormatter(formatter)

        # 添加处理器
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)

        return logger

class MetricsCollector:
    def __init__(self):
        self.metrics = {}

    def collect_system_metrics(self):
        """收集系统指标"""
        import psutil

        metrics = {
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'network_io': psutil.net_io_counters(),
            'process_count': len(psutil.pids()),
            'load_average': psutil.getloadavg() if hasattr(psutil, 'getloadavg') else None
        }

        return metrics

    def collect_application_metrics(self):
        """收集应用指标"""
        metrics = {
            'active_alerts': self.count_active_alerts(),
            'processed_alerts': self.count_processed_alerts(),
            'notification_sent': self.count_notifications_sent(),
            'auto_repair_executed': self.count_auto_repairs(),
            'response_time': self.get_average_response_time()
        }

        return metrics