可观测性3.0：AI驱动的智能监控与故障预测

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# AI异常检测

class AnomalyDetector:
    """异常检测器"""

    def __init__(self, model_type='isolation_forest'):
        self.model = self._load_model(model_type)
        self.baseline = None

    def train_baseline(self, metrics_data):
        """训练基线"""

        from sklearn.ensemble import IsolationForest

        self.model = IsolationForest(
            contamination=0.1,
            random_state=42
        )

        self.model.fit(metrics_data)

        # 计算基线统计
        self.baseline = {
            'mean': metrics_data.mean(axis=0),
            'std': metrics_data.std(axis=0),
            'threshold': metrics_data.quantile(0.95)
        }

    def detect(self, current_metrics):
        """检测异常"""

        # 1. 统计异常检测
        z_score = (current_metrics - self.baseline['mean']) / self.baseline['std']
        statistical_anomalies = z_score > 3

        # 2. ML异常检测
        anomaly_score = self.model.score_samples([current_metrics])[0]
        ml_anomaly = anomaly_score < -0.5

        # 3. 组合判断
        return {
            'is_anomaly': statistical_anomalies or ml_anomaly,
            'statistical_score': z_score.max(),
            'ml_score': anomaly_score,
            'severity': self._calculate_severity(z_score.max(), anomaly_score)
        }

    def _calculate_severity(self, stat_score, ml_score):
        """计算严重程度"""

        if stat_score > 5 or ml_score < -0.8:
            return 'CRITICAL'
        elif stat_score > 3 or ml_score < -0.6:
            return 'HIGH'
        elif stat_score > 2 or ml_score < -0.4:
            return 'MEDIUM'
        else:
            return 'LOW'

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

# AI根因分析

class RootCauseAnalyzer:
    """根因分析器"""

    def __init__(self):
        self.graph = self._build_dependency_graph()
        self.symptom_patterns = {}

    def analyze(self, incident):
        """分析故障根因"""

        # 1. 收集相关指标
        metrics = self._collect_metrics(incident)

        # 2. 识别异常指标
        anomalies = self._identify_anomalies(metrics)

        # 3. 图分析 - 找传播路径
        causal_path = self._find_causal_path(anomalies)

        # 4. 历史匹配 - 找相似案例
        similar_incidents = self._find_similar_incidents(incident)

        # 5. 生成根因假设
        hypotheses = self._generate_hypotheses(
            anomalies,
            causal_path,
            similar_incidents
        )

        # 6. 验证假设
        verified_causes = self._verify_hypotheses(hypotheses)

        return verified_causes

    def _build_dependency_graph(self):
        """构建依赖关系图"""

        import networkx as nx

        graph = nx.DiGraph()

        # 添加节点
        components = [
            'Load Balancer',
            'Web Server',
            'Application Server',
            'Database',
            'Cache',
            'External API'
        ]

        for component in components:
            graph.add_node(component)

        # 添加依赖关系
        dependencies = [
            ('Load Balancer', 'Web Server'),
            ('Web Server', 'Application Server'),
            ('Application Server', 'Cache'),
            ('Application Server', 'Database'),
            ('Application Server', 'External API')
        ]

        graph.add_edges_from(dependencies)

        return graph

    def _find_causal_path(self, anomalies):
        """寻找因果路径"""

        import networkx as nx

        # 找到异常节点
        anomaly_nodes = [a['component'] for a in anomalies]

        # 从每个异常节点向上游追溯
        paths = []
        for node in anomaly_nodes:
            # 找到所有影响该节点的祖先节点
            ancestors = nx.ancestors(self.graph, node)
            paths.append(list(ancestors))

        return paths

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

# 容量预测

class CapacityPredictor:
    """容量预测器"""

    def __init__(self):
        self.models = {
            'cpu': self._create_model('cpu'),
            'memory': self._create_model('memory'),
            'storage': self._create_model('storage')
        }

    def _create_model(self, metric_type):
        """创建预测模型"""

        from sklearn.linear_model import LinearRegression

        return LinearRegression()

    def train(self, historical_data):
        """训练预测模型"""

        for metric_type in ['cpu', 'memory', 'storage']:
            model = self.models[metric_type]

            # 特征：时间、历史值
            X = historical_data[metric_type]['features']
            y = historical_data[metric_type]['values']

            model.fit(X, y)

    def predict_capacity(self, horizon_hours=24):
        """预测未来容量需求"""

        predictions = {}

        for metric_type, model in self.models.items():
            # 获取最新数据点
            latest_data = self._get_latest_data(metric_type)

            # 预测
            X_pred = self._create_prediction_features(latest_data, horizon_hours)
            y_pred = model.predict(X_pred)

            predictions[metric_type] = {
                'predictions': y_pred,
                'upper_bound': y_pred * 1.2,  # +20%
                'lower_bound': y_pred * 0.8,  # -20%
            }

        return predictions

    def recommend_scaling(self, predictions):
        """推荐扩容策略"""

        recommendations = []

        for metric_type, pred in predictions.items():
            max_value = pred['predictions'].max()
            current_capacity = self._get_current_capacity(metric_type)

            if max_value > current_capacity * 0.8:
                recommendations.append({
                    'metric': metric_type,
                    'action': 'scale_up',
                    'target': int(max_value * 1.2),
                    'reason': f'{metric_type} usage expected to reach {max_value:.2f}%'
                })

        return recommendations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

# 故障预测

class FailurePredictor:
    """故障预测器"""

    def __init__(self):
        self.models = {
            'hardware': self._create_hardware_model(),
            'software': self._create_software_model(),
            'network': self._create_network_model()
        }

    def predict_failure_probability(self, system_metrics):
        """预测故障概率"""

        probabilities = {}

        for component, metrics in system_metrics.items():
            # 提取特征
            features = self._extract_features(metrics)

            # 预测
            prob = self.models[component].predict_proba([features])[0][1]

            probabilities[component] = {
                'probability': prob,
                'risk_level': self._classify_risk(prob),
                'confidence': self._calculate_confidence(metrics)
            }

        return probabilities

    def _classify_risk(self, probability):
        """分类风险等级"""

        if probability > 0.8:
            return 'CRITICAL'
        elif probability > 0.5:
            return 'HIGH'
        elif probability > 0.2:
            return 'MEDIUM'
        else:
            return 'LOW'

    def suggest_preventive_action(self, prediction):
        """建议预防措施"""

        actions = []

        if prediction['probability'] > 0.5:
            if prediction['risk_level'] == 'HIGH':
                actions.append({
                    'action': 'immediate_investigation',
                    'priority': 'HIGH',
                    'description': '立即调查组件状态'
                })

                actions.append({
                    'action': 'prepare_failover',
                    'priority': 'HIGH',
                    'description': '准备故障转移方案'
                })

            elif prediction['risk_level'] == 'MEDIUM':
                actions.append({
                    'action': 'schedule_maintenance',
                    'priority': 'MEDIUM',
                    'description': '安排预防性维护'
                })

        return actions

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96

# 自动化响应系统

class AutoRemediation:
    """自动修复系统"""

    def __init__(self):
        self.playbooks = {}
        self.execution_history = []

    def register_playbook(self, playbook):
        """注册自动化手册"""

        self.playbooks[playbook['id']] = playbook

    async def execute_remediation(self, incident):
        """执行自动修复"""

        # 匹配修复手册
        playbook = self._match_playbook(incident)

        if not playbook:
            return {
                'success': False,
                'reason': 'No matching playbook'
            }

        # 执行修复步骤
        results = []

        for step in playbook['steps']:
            try:
                result = await self._execute_step(step, incident)

                results.append({
                    'step': step['name'],
                    'success': result['success'],
                    'output': result.get('output', '')
                })

                if not result['success'] and step.get('critical', False):
                    # 关键步骤失败，停止执行
                    break

            except Exception as e:
                results.append({
                    'step': step['name'],
                    'success': False,
                    'error': str(e)
                })
                break

        # 记录执行历史
        self.execution_history.append({
            'incident': incident,
            'playbook': playbook['id'],
            'results': results,
            'timestamp': datetime.now()
        })

        return {
            'success': all(r['success'] for r in results),
            'results': results
        }

    def _match_playbook(self, incident):
        """匹配修复手册"""

        for playbook in self.playbooks.values():
            if self._incident_matches_playbook(incident, playbook):
                return playbook

        return None

    def _incident_matches_playbook(self, incident, playbook):
        """检查事故是否匹配手册"""

        # 检查触发条件
        conditions = playbook.get('triggers', [])

        for condition in conditions:
            if condition['type'] == 'metric_threshold':
                metric_value = incident['metrics'].get(condition['metric'])

                if metric_value:
                    if condition['operator'] == '>':
                        if metric_value > condition['threshold']:
                            continue
                    elif condition['operator'] == '<':
                        if metric_value < condition['threshold']:
                            continue

            # 条件不满足
            return False

        # 所有条件都满足
        return True