模型监控与性能衰减检测：安全视角下的实时系统保障

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
from pyflink.table.expressions import col

# 创建流处理环境
env = StreamExecutionEnvironment.get_execution_environment()
settings = EnvironmentSettings.new_instance().in_streaming_mode().build()
t_env = StreamTableEnvironment.create(env, environment_settings=settings)

# 创建Kafka源表
t_env.execute_sql("""
CREATE TABLE inference_logs (
    timestamp TIMESTAMP(3),
    model_id STRING,
    input_data STRING,
    prediction STRING,
    ground_truth STRING,
    latency DOUBLE
) WITH (
    'connector' = 'kafka',
    'topic' = 'inference-logs',
    'properties.bootstrap.servers' = 'localhost:9092',
    'properties.group.id' = 'model-monitor',
    'format' = 'json',
    'scan.startup.mode' = 'latest-offset'
)
""")

# 创建性能指标计算视图
t_env.execute_sql("""
CREATE VIEW performance_metrics AS
SELECT
    TUMBLE_ROWTIME(timestamp, INTERVAL '1' MINUTE) AS window_time,
    model_id,
    COUNT(*) AS total_samples,
    SUM(CASE WHEN prediction = ground_truth THEN 1 ELSE 0 END) AS correct_predictions,
    SUM(CASE WHEN prediction = 'malicious' AND ground_truth = 'malicious' THEN 1 ELSE 0 END) AS true_positives,
    SUM(CASE WHEN prediction = 'malicious' AND ground_truth = 'benign' THEN 1 ELSE 0 END) AS false_positives,
    SUM(CASE WHEN prediction = 'benign' AND ground_truth = 'malicious' THEN 1 ELSE 0 END) AS false_negatives,
    SUM(CASE WHEN prediction = 'benign' AND ground_truth = 'benign' THEN 1 ELSE 0 END) AS true_negatives,
    AVG(latency) AS avg_latency
FROM inference_logs
GROUP BY TUMBLE(timestamp, INTERVAL '1' MINUTE), model_id
""")

# 计算准确率、召回率、F1值等指标
t_env.execute_sql("""
CREATE TABLE model_metrics (
    window_time TIMESTAMP(3),
    model_id STRING,
    accuracy DOUBLE,
    precision DOUBLE,
    recall DOUBLE,
    f1_score DOUBLE,
    false_positive_rate DOUBLE,
    false_negative_rate DOUBLE,
    avg_latency DOUBLE
) WITH (
    'connector' = 'kafka',
    'topic' = 'model-metrics',
    'properties.bootstrap.servers' = 'localhost:9092',
    'format' = 'json'
)
""")

t_env.execute_sql("""
INSERT INTO model_metrics
SELECT
    window_time,
    model_id,
    CAST(correct_predictions AS DOUBLE) / total_samples AS accuracy,
    CAST(true_positives AS DOUBLE) / (true_positives + false_positives) AS precision,
    CAST(true_positives AS DOUBLE) / (true_positives + false_negatives) AS recall,
    2 * precision * recall / (precision + recall) AS f1_score,
    CAST(false_positives AS DOUBLE) / (false_positives + true_negatives) AS false_positive_rate,
    CAST(false_negatives AS DOUBLE) / (false_negatives + true_positives) AS false_negative_rate,
    avg_latency
FROM performance_metrics
""")

# 执行作业
t_env.execute("Model Monitoring Job")

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

class ModelMonitor:
    def __init__(self, contamination=0.1):
        self.contamination = contamination
        self.scaler = StandardScaler()
        self.model = IsolationForest(contamination=contamination, random_state=42)
        self.is_trained = False
        self.baseline_metrics = None
    
    def train(self, historical_metrics):
        """
        使用历史指标训练异常检测模型
        historical_metrics: 历史性能指标，包含accuracy, precision, recall, f1_score等
        """
        # 标准化数据
        scaled_data = self.scaler.fit_transform(historical_metrics)
        # 训练隔离森林模型
        self.model.fit(scaled_data)
        # 计算基线指标
        self.baseline_metrics = historical_metrics.mean().to_dict()
        self.is_trained = True
        print("异常检测模型训练完成")
    
    def detect_anomaly(self, current_metrics):
        """
        检测当前指标是否异常
        current_metrics: 当前性能指标
        返回: 异常得分，-1表示异常，1表示正常
        """
        if not self.is_trained:
            raise ValueError("模型未训练")
        
        # 标准化当前指标
        scaled_metrics = self.scaler.transform([current_metrics])
        # 预测异常得分
        anomaly_score = self.model.predict(scaled_metrics)[0]
        return anomaly_score
    
    def analyze_root_cause(self, current_metrics, data_stats=None, system_metrics=None):
        """
        分析性能衰减的根因
        current_metrics: 当前性能指标
        data_stats: 数据统计信息，如数据分布、特征分布等
        system_metrics: 系统指标，如CPU使用率、内存使用率等
        返回: 根因分析结果
        """
        root_causes = []
        
        # 比较当前指标与基线
        for metric, value in current_metrics.items():
            baseline = self.baseline_metrics[metric]
            # 计算变化百分比
            change_pct = (value - baseline) / baseline * 100
            
            # 检查是否超过阈值
            if abs(change_pct) > 10:  # 10%变化阈值
                root_causes.append({
                    'metric': metric,
                    'current_value': value,
                    'baseline_value': baseline,
                    'change_pct': change_pct,
                    'possible_cause': self._get_possible_cause(metric, change_pct)
                })
        
        # 结合数据统计信息分析
        if data_stats:
            # 检查数据分布变化
            if 'distribution_drift' in data_stats and data_stats['distribution_drift'] > 0.5:
                root_causes.append({
                    'metric': 'data_distribution',
                    'current_value': data_stats['distribution_drift'],
                    'baseline_value': 0,
                    'change_pct': float('inf'),
                    'possible_cause': '数据分布发生显著变化'
                })
        
        # 结合系统指标分析
        if system_metrics:
            # 检查系统资源使用率
            if 'cpu_usage' in system_metrics and system_metrics['cpu_usage'] > 80:
                root_causes.append({
                    'metric': 'cpu_usage',
                    'current_value': system_metrics['cpu_usage'],
                    'baseline_value': 50,
                    'change_pct': (system_metrics['cpu_usage'] - 50) / 50 * 100,
                    'possible_cause': 'CPU使用率过高，影响模型推理性能'
                })
        
        return root_causes
    
    def _get_possible_cause(self, metric, change_pct):
        """
        根据指标类型和变化方向，获取可能的原因
        """
        cause_map = {
            'accuracy': {
                'decrease': '模型泛化能力下降，可能是数据分布漂移或模型老化',
                'increase': '模型性能提升，可能是数据质量提高或模型优化'
            },
            'precision': {
                'decrease': '误报增加，可能是攻击者使用了新型攻击手段',
                'increase': '误报减少，模型对正常样本的识别能力提升'
            },
            'recall': {
                'decrease': '漏报增加，可能是新型攻击未被模型识别',
                'increase': '漏报减少，模型对攻击样本的识别能力提升'
            },
            'f1_score': {
                'decrease': '模型整体性能下降，需要进一步分析',
                'increase': '模型整体性能提升'
            },
            'false_positive_rate': {
                'increase': '误报率增加，可能影响业务正常运行',
                'decrease': '误报率降低，模型准确性提升'
            },
            'false_negative_rate': {
                'increase': '漏报率增加，可能导致安全事件',
                'decrease': '漏报率降低，模型安全性提升'
            },
            'avg_latency': {
                'increase': '推理延迟增加，可能是系统资源不足或模型复杂度增加',
                'decrease': '推理延迟降低，系统性能优化'
            }
        }
        
        direction = 'decrease' if change_pct < 0 else 'increase'
        return cause_map.get(metric, {}).get(direction, '未知原因')

# 示例用法
if __name__ == "__main__":
    # 模拟历史指标数据
    historical_data = pd.DataFrame({
        'accuracy': [0.98, 0.97, 0.98, 0.975, 0.98],
        'precision': [0.95, 0.94, 0.95, 0.945, 0.95],
        'recall': [0.96, 0.95, 0.96, 0.955, 0.96],
        'f1_score': [0.955, 0.945, 0.955, 0.95, 0.955],
        'false_positive_rate': [0.02, 0.025, 0.02, 0.023, 0.02],
        'false_negative_rate': [0.01, 0.015, 0.01, 0.012, 0.01],
        'avg_latency': [50, 52, 51, 53, 50]
    })
    
    # 创建监控实例
    monitor = ModelMonitor()
    # 训练模型
    monitor.train(historical_data)
    
    # 模拟当前指标数据（异常情况）
    current_metrics = [
        0.90,  # accuracy 下降
        0.85,  # precision 下降
        0.80,  # recall 下降
        0.82,  # f1_score 下降
        0.08,  # false_positive_rate 上升
        0.05,  # false_negative_rate 上升
        65     # avg_latency 上升
    ]
    
    # 检测异常
    anomaly_score = monitor.detect_anomaly(current_metrics)
    print(f"异常得分: {anomaly_score}")
    
    # 分析根因
    data_stats = {'distribution_drift': 0.7}  # 数据分布发生明显漂移
    system_metrics = {'cpu_usage': 85}  # CPU使用率过高
    
    root_causes = monitor.analyze_root_cause(current_metrics, data_stats, system_metrics)
    print("根因分析结果:")
    for cause in root_causes:
        print(f"- 指标: {cause['metric']}, 当前值: {cause['current_value']}, 基线值: {cause['baseline_value']}, 变化率: {cause['change_pct']:.2f}%, 可能原因: {cause['possible_cause']}")

# 安装必要的依赖
pip install pyflink pandas numpy scikit-learn prometheus-client grafana-api

# 启动Kafka
bin/kafka-server-start.sh config/server.properties

# 启动Flink
./bin/start-cluster.sh

# 启动Prometheus
./prometheus --config.file=prometheus.yml

# 启动Grafana
grafana-server

{
  "dashboard": {
    "title": "模型监控面板",
    "panels": [
      {
        "type": "graph",
        "title": "准确率变化趋势",
        "targets": [
          {
            "expr": "model_accuracy{model_id='security_model'}",
            "interval": "",
            "legendFormat": "准确率",
            "refId": "A"
          }
        ]
      },
      {
        "type": "graph",
        "title": "召回率变化趋势",
        "targets": [
          {
            "expr": "model_recall{model_id='security_model'}",
            "interval": "",
            "legendFormat": "召回率",
            "refId": "B"
          }
        ]
      },
      {
        "type": "graph",
        "title": "F1值变化趋势",
        "targets": [
          {
            "expr": "model_f1_score{model_id='security_model'}",
            "interval": "",
            "legendFormat": "F1值",
            "refId": "C"
          }
        ]
      },
      {
        "type": "graph",
        "title": "平均延迟变化趋势",
        "targets": [
          {
            "expr": "model_avg_latency{model_id='security_model'}",
            "interval": "",
            "legendFormat": "平均延迟",
            "refId": "D"
          }
        ]
      }
    ]
  }
}