混沌工程实验中根因定位的因果推断方法与误报治理

chaos_cause/
├── core/
│   ├── __init__.py
│   ├── causal_graph.py      # 因果图数据模型与操作
│   ├── pc_algorithm.py      # PC算法实现
│   └── fault_injector.py    # 模拟故障注入与指标生成
├── experiment/
│   ├── __init__.py
│   ├── runner.py            # 实验运行器
│   └── data_collector.py    # 模拟数据采集
├── analysis/
│   ├── __init__.py
│   ├── root_cause_analyzer.py # 根因分析器
│   └── false_positive_filter.py # 误报过滤器
├── models/
│   ├── __init__.py
│   └── metrics.py           # 数据模型（指标、服务）
├── config/
│   └── settings.yaml        # 配置文件
├── utils/
│   ├── __init__.py
│   └── visualization.py     # 可视化工具（含Mermaid生成）
├── tests/                   # 单元测试（简要）
├── requirements.txt
└── main.py                  # 主入口

from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Any, Optional
import numpy as np

@dataclass
class Service:
    """微服务模型"""
    name: str
    endpoints: List[str] = field(default_factory=list) # 例如 ["/api/v1/order", "/health"]

@dataclass
class MetricSample:
    """指标采样点"""
    timestamp: datetime
    service_name: str
    metric_name: str  # 如 "latency_ms", "error_rate", "cpu_util"
    value: float

@dataclass
class ExperimentData:
    """一次混沌实验的数据集"""
    experiment_id: str
    baseline_metrics: List[MetricSample] = field(default_factory=list) # 对照组数据
    fault_metrics: List[MetricSample] = field(default_factory=list)    # 实验组（注入故障）数据
    fault_injected: str = ""  # 注入的故障描述，如 "service_payment latency +300ms"
    start_time: Optional[datetime] = None
    end_time: Optional[datetime] = None

import networkx as nx
from typing import Set, List, Tuple
import pandas as pd

class CausalGraph:
    """因果图管理器，基于有向无环图（DAG）"""
    def __init__(self):
        self.graph = nx.DiGraph()

    def add_node(self, node_id: str, node_type: str = "service"):
        """添加节点，节点ID通常为 `service_name.metric_name`"""
        self.graph.add_node(node_id, type=node_type)

    def add_edge(self, from_node: str, to_node: str, weight: float = 1.0, confidence: float = 1.0):
        """添加一条因果边，表示 from_node 影响 to_node"""
        if from_node == to_node:
            return
        self.graph.add_edge(from_node, to_node, weight=weight, confidence=confidence)

    def remove_edge(self, from_node: str, to_node: str):
        """移除边"""
        if self.graph.has_edge(from_node, to_node):
            self.graph.remove_edge(from_node, to_node)

    def get_ancestors(self, node: str) -> Set[str]:
        """获取某个节点的所有祖先节点（潜在原因）"""
        return nx.ancestors(self.graph, node)

    def get_descendants(self, node: str) -> Set[str]:
        """获取某个节点的所有后代节点（影响范围）"""
        return nx.descendants(self.graph, node)

    def infer_root_causes(self, anomalous_nodes: List[str], depth_limit: int = 3) -> List[Tuple[str, float]]:
        """
        基于因果图推断根因。
        简化的启发式算法：寻找 anomalous_nodes 的公共祖先，并根据距离和边置信度打分。
        """
        candidate_scores = {}
        for ano_node in anomalous_nodes:
            ancestors = self.get_ancestors(ano_node)
            # 限制搜索深度，避免找到太遥远的祖先
            shallow_ancestors = set()
            for anc in ancestors:
                try:
                    # 计算最短路径长度作为距离
                    path_len = nx.shortest_path_length(self.graph, anc, ano_node)
                    if path_len <= depth_limit:
                        shallow_ancestors.add(anc)
                except nx.NetworkXNoPath:
                    continue

            for cand in shallow_ancestors:
                # 打分：初始分1.0，每多一个异常节点将其作为祖先，分数增加
                candidate_scores[cand] = candidate_scores.get(cand, 0.0) + 1.0
                # 可选：根据边的置信度加权
                try:
                    path = nx.shortest_path(self.graph, cand, ano_node)
                    for i in range(len(path)-1):
                        edge_data = self.graph.get_edge_data(path[i], path[i+1])
                        candidate_scores[cand] *= edge_data.get('confidence', 1.0)
                except:
                    pass

        # 排序并返回
        sorted_causes = sorted(candidate_scores.items(), key=lambda x: x[1], reverse=True)
        return sorted_causes

    def to_mermaid_format(self) -> str:
        """将因果图转换为Mermaid流程图定义，用于可视化"""
        lines = ["graph TD"]
        for u, v, data in self.graph.edges(data=True):
            weight = data.get('weight', 1.0)
            lines.append(f"    {u}-->{v};")
        return "\n".join(lines)

    def learn_from_data(self, df: pd.DataFrame, independence_test_func, alpha: float = 0.05):
        """
        预留接口：从数据中学习因果结构。
        实际应调用PC算法等。
        """
        # 此方法将委托给 pc_algorithm.py 中的具体实现
        pass

import pandas as pd
import numpy as np
from itertools import combinations
from typing import Set, Tuple, List, Dict
import networkx as nx
from core.causal_graph import CausalGraph

def gaussian_ci_test(X: pd.Series, Y: pd.Series, Z: List[pd.Series], alpha: float = 0.05) -> bool:
    """
    基于偏相关的条件独立性检验（高斯假设简化版）。
    返回 True 表示独立 (p-value > alpha)。
    实际生产环境应使用更鲁棒的检验（如基于距离的相关性）。
    """
    from scipy import stats
    if len(Z) == 0:
        # 无条件相关性检验
        corr, p_value = stats.pearsonr(X, Y)
        return p_value > alpha
    else:
        # 构建设计矩阵进行偏相关计算
        data = pd.concat([X, Y] + Z, axis=1)
        # 简化为计算偏相关系数的p值（通过线性回归残差的相关性）
        # 这里是一个高度简化的实现，仅用于演示
        import statsmodels.api as sm
        Z_df = pd.concat(Z, axis=1)
        Z_df = sm.add_constant(Z_df)
        model_x = sm.OLS(X, Z_df).fit()
        model_y = sm.OLS(Y, Z_df).fit()
        res_corr, p_value = stats.pearsonr(model_x.resid, model_y.resid)
        return p_value > alpha

def pc_algorithm(data_df: pd.DataFrame, alpha: float = 0.05, max_condition_size: int = 3) -> CausalGraph:
    """
    PC算法骨架实现（约束性因果发现）。
    输入：DataFrame，每列为一个变量（节点）的时序数据。
    输出：一个部分有向因果图（此处简化为有向图CausalGraph）。
    注意：此为教学简化版，未完整实现V-结构定向等所有步骤。
    """
    nodes = list(data_df.columns)
    n = len(nodes)
    # 1. 初始化完全无向图
    graph = {node: set(nodes) - {node} for node in nodes}
    # 2. 逐步删除边（条件独立性检验）
    condition_size = 0
    while condition_size <= max_condition_size and any(len(adj) > condition_size for adj in graph.values()):
        for X in nodes:
            neighbors = list(graph[X])
            for Y in neighbors:
                # 获取X和Y的邻接集（除去对方）
                adj_set = set(neighbors) - {Y}
                if len(adj_set) >= condition_size:
                    for Z in combinations(adj_set, condition_size):
                        Z_list = [data_df[z] for z in Z]
                        if gaussian_ci_test(data_df[X], data_df[Y], Z_list, alpha):
                            # 独立，删除边
                            graph[X].discard(Y)
                            graph[Y].discard(X)
                            break # 找到任一条件集独立即可删除边，跳出Z循环
        condition_size += 1

    # 3. 定向阶段（简化版，仅定向一部分明显的V-结构 X->Z<-Y）
    # 此处省略完整定向过程，直接构建一个有向图（基于剩余边的方向性假设或后续学习）
    cg = CausalGraph()
    for node in nodes:
        cg.add_node(node, node_type="metric")

    # 简化：将所有剩余无向边随机定向为有向边（实际项目必须实现完整定向规则）
    edges_added = set()
    for X in nodes:
        for Y in graph[X]:
            if (Y, X) not in edges_added:
                cg.add_edge(X, Y, weight=0.5, confidence=0.8) # 示例权重和置信度
                edges_added.add((X, Y))

    return cg

import random
import numpy as np
from datetime import datetime, timedelta
from typing import List
from models.metrics import MetricSample, Service

class FaultInjector:
    """模拟故障注入与指标数据生成器"""
    def __init__(self, services: List[Service]):
        self.services = {s.name: s for s in services}
        # 定义健康的基线指标范围
        self.baseline_ranges = {
            "latency_ms": (10.0, 100.0),   # 健康延迟
            "error_rate": (0.0, 0.01),     # 健康错误率
            "cpu_util": (20.0, 60.0)       # 健康CPU利用率
        }
        # 定义服务间的简易调用关系（用于模拟故障传播）
        self.call_graph = {
            "web_gateway": ["service_order", "service_payment"],
            "service_order": ["service_inventory", "service_payment"],
            "service_payment": [],
            "service_inventory": []
        }

    def generate_baseline_metrics(self, duration_minutes: int = 5, sample_interval_s: int = 10) -> List[MetricSample]:
        """生成健康状态下的基线指标"""
        samples = []
        end_time = datetime.now()
        start_time = end_time - timedelta(minutes=duration_minutes)
        current = start_time
        while current <= end_time:
            for svc_name, svc in self.services.items():
                for metric, (low, high) in self.baseline_ranges.items():
                    # 添加一些随机波动
                    value = low + (high - low) * random.random() * 0.3  # 波动30%
                    value = max(low, min(high, value)) # 确保在范围内
                    samples.append(MetricSample(
                        timestamp=current,
                        service_name=svc_name,
                        metric_name=metric,
                        value=value
                    ))
            current += timedelta(seconds=sample_interval_s)
        return samples

    def inject_fault_and_generate_metrics(self, fault_spec: dict, duration_minutes: int = 5,
                                          sample_interval_s: int = 10) -> List[MetricSample]:
        """
        根据故障规约注入故障并生成受影响的指标数据。
        fault_spec 示例: {'target_service': 'service_payment', 'fault_type': 'latency_increase', 'magnitude': 300}
        """
        fault_service = fault_spec['target_service']
        fault_type = fault_spec['fault_type']
        magnitude = fault_spec.get('magnitude', 100)

        baseline_samples = self.generate_baseline_metrics(duration_minutes, sample_interval_s)
        fault_samples = []

        # 模拟故障影响：直接影响的指标
        for sample in baseline_samples:
            new_sample = MetricSample(
                timestamp=sample.timestamp,
                service_name=sample.service_name,
                metric_name=sample.metric_name,
                value=sample.value
            )
            if sample.service_name == fault_service:
                if fault_type == 'latency_increase' and sample.metric_name == 'latency_ms':
                    new_sample.value += magnitude
                elif fault_type == 'error_rate_increase' and sample.metric_name == 'error_rate':
                    new_sample.value = min(1.0, sample.value + magnitude / 100.0) # magnitude 作为百分比点
                elif fault_type == 'cpu_spike' and sample.metric_name == 'cpu_util':
                    new_sample.value = min(100.0, sample.value + magnitude)

            # 模拟故障沿调用链传播（简化：影响下游服务的延迟和错误率）
            if sample.service_name in self.call_graph.get(fault_service, []):
                propagation_factor = 0.5  # 传播衰减因子
                if sample.metric_name == 'latency_ms':
                    new_sample.value += magnitude * propagation_factor * random.uniform(0.8, 1.2)
                elif sample.metric_name == 'error_rate':
                    new_sample.value = min(1.0, sample.value + (magnitude / 100.0) * propagation_factor * random.uniform(0.5, 1.0))

            fault_samples.append(new_sample)

        return fault_samples

import yaml
import json
from datetime import datetime
from typing import Dict, Any
from core.fault_injector import FaultInjector
from models.metrics import ExperimentData, Service
from experiment.data_collector import DataCollector
from analysis.root_cause_analyzer import RootCauseAnalyzer

class ExperimentRunner:
    """混沌实验运行器"""
    def __init__(self, config_path: str = "config/settings.yaml"):
        with open(config_path, 'r') as f:
            self.config = yaml.safe_load(f)
        services = [Service(name=svc['name'], endpoints=svc.get('endpoints', []))
                    for svc in self.config['services']]
        self.fault_injector = FaultInjector(services)
        self.data_collector = DataCollector()
        self.analyzer = RootCauseAnalyzer()

    def run_experiment(self, fault_spec: Dict[str, Any]) -> ExperimentData:
        """执行单次混沌实验"""
        print(f"开始实验，注入故障: {json.dumps(fault_spec)}")
        exp_id = f"exp_{datetime.now().strftime('%Y%m%d_%H%M%S')}"

        # 1. 采集基线数据（在实际系统中，可能是在故障注入前的一段时间）
        print("采集基线数据...")
        baseline_data = self.fault_injector.generate_baseline_metrics(duration_minutes=2)

        # 2. 注入故障并采集实验数据
        print("注入故障并采集实验数据...")
        fault_data = self.fault_injector.inject_fault_and_generate_metrics(fault_spec, duration_minutes=3)

        # 3. 封装实验数据
        exp_data = ExperimentData(
            experiment_id=exp_id,
            baseline_metrics=baseline_data,
            fault_metrics=fault_data,
            fault_injected=json.dumps(fault_spec),
            start_time=datetime.now()  # 简化
        )

        # 4. 可选：保存数据
        self.data_collector.save_experiment_data(exp_data)

        print(f"实验 {exp_id} 完成。")
        return exp_data

    def analyze_experiment(self, exp_data: ExperimentData) -> Dict[str, Any]:
        """分析实验数据，定位根因"""
        print(f"分析实验 {exp_data.experiment_id} ...")
        result = self.analyzer.analyze(exp_data)
        return result

import pandas as pd
import numpy as np
from typing import Dict, Any, List, Tuple
from models.metrics import ExperimentData
from core.causal_graph import CausalGraph
from core.pc_algorithm import pc_algorithm
from analysis.false_positive_filter import FalsePositiveFilter

class RootCauseAnalyzer:
    """根因分析器，整合因果推断与误报过滤"""
    def __init__(self, use_predefined_graph: bool = True):
        self.causal_graph = CausalGraph()
        self.fp_filter = FalsePositiveFilter()
        if use_predefined_graph:
            self._init_predefined_graph()

    def _init_predefined_graph(self):
        """初始化一个预设的简单因果图（基于领域知识）"""
        nodes = [
            "web_gateway.latency_ms", "web_gateway.error_rate",
            "service_order.latency_ms", "service_order.error_rate",
            "service_payment.latency_ms", "service_payment.error_rate",
            "service_inventory.latency_ms", "service_inventory.error_rate"
        ]
        for node in nodes:
            self.causal_graph.add_node(node)
        # 预设一些调用关系导致的因果边（支付影响订单，订单影响网关等）
        edges = [
            ("service_payment.latency_ms", "service_order.latency_ms"),
            ("service_payment.error_rate", "service_order.error_rate"),
            ("service_order.latency_ms", "web_gateway.latency_ms"),
            ("service_order.error_rate", "web_gateway.error_rate"),
            ("service_inventory.latency_ms", "service_order.latency_ms"),
        ]
        for u, v in edges:
            self.causal_graph.add_edge(u, v, weight=0.9, confidence=0.8)

    def _detect_anomalies(self, baseline_df: pd.DataFrame, fault_df: pd.DataFrame, threshold: float = 2.0) -> List[str]:
        """简单的异常检测：基于均值和标准差"""
        anomalous_nodes = []
        for col in baseline_df.columns:
            base_mean = baseline_df[col].mean()
            base_std = baseline_df[col].std()
            fault_mean = fault_df[col].mean()
            if base_std > 0 and abs(fault_mean - base_mean) > threshold * base_std:
                anomalous_nodes.append(col)
        return anomalous_nodes

    def analyze(self, exp_data: ExperimentData) -> Dict[str, Any]:
        """执行完整的分析流程"""
        # 1. 数据预处理：转换为DataFrame
        def to_dataframe(metrics):
            # 简化：假设我们只分析特定指标，且按时间和服务聚合
            data = {}
            for m in metrics:
                key = f"{m.service_name}.{m.metric_name}"
                if key not in data:
                    data[key] = []
                data[key].append(m.value)
            # 对齐长度（简化处理）
            max_len = max(len(v) for v in data.values())
            for k in data:
                if len(data[k]) < max_len:
                    data[k] += [np.nan] * (max_len - len(data[k]))
            return pd.DataFrame(data)

        baseline_df = to_dataframe(exp_data.baseline_metrics)
        fault_df = to_dataframe(exp_data.fault_metrics)

        # 2. （可选）从本次实验数据中学习因果图（增量学习）
        # combined_df = pd.concat([baseline_df, fault_df]).reset_index(drop=True)
        # learned_graph = pc_algorithm(combined_df, alpha=0.1, max_condition_size=2)
        # self._merge_graph(learned_graph) # 实现合并逻辑

        # 3. 检测异常节点
        anomalous_nodes = self._detect_anomalies(baseline_df, fault_df, threshold=1.5)
        print(f"检测到异常节点: {anomalous_nodes}")

        # 4. 基于因果图推断根因候选
        raw_causes = self.causal_graph.infer_root_causes(anomalous_nodes, depth_limit=3)
        print(f"原始根因推断结果: {raw_causes}")

        # 5. 误报过滤
        filtered_causes = self.fp_filter.filter(
            raw_causes,
            baseline_metrics=baseline_df.mean().to_dict(),
            fault_metrics=fault_df.mean().to_dict()
        )

        # 6. 准备结果
        result = {
            'experiment_id': exp_data.experiment_id,
            'fault_injected': exp_data.fault_injected,
            'anomalous_metrics': anomalous_nodes,
            'raw_root_causes': [{'node': n, 'score': s} for n, s in raw_causes],
            'filtered_root_causes': filtered_causes,
            'causal_graph_mermaid': self.causal_graph.to_mermaid_format()
        }
        return result

from typing import List, Tuple, Dict, Any

class FalsePositiveFilter:
    """误报过滤器"""
    def __init__(self):
        # 静态规则：指标波动必须超过绝对阈值才认为是有效根因
        self.abs_thresholds = {
            "latency_ms": 50.0,   # 延迟增加超过50ms才考虑
            "error_rate": 0.05,   # 错误率增加超过5个百分点才考虑
            "cpu_util": 10.0      # CPU利用率增加超过10%才考虑
        }
        # 模拟一个简单的历史误报模式库（实际可从数据库加载）
        self.fp_patterns = [
            {"node_pattern": ".*latency_ms", "condition": "delta < 20 && baseline_value < 30"},
        ]

    def filter(self, raw_causes: List[Tuple[str, float]],
               baseline_metrics: Dict[str, float],
               fault_metrics: Dict[str, float]) -> List[Dict[str, Any]]:
        """
        过滤误报。
        raw_causes: [(node_id, score), ...]
        返回过滤后的列表，每个元素为包含节点、分数和是否被过滤标志的字典。
        """
        filtered = []
        for node, score in raw_causes:
            # 规则1: 检查绝对变化阈值
            should_keep = self._check_absolute_threshold(node, baseline_metrics, fault_metrics)
            # 规则2: （示例）可以添加基于历史模式匹配的规则
            if should_keep and self._matches_fp_pattern(node, baseline_metrics, fault_metrics):
                should_keep = False

            if should_keep:
                filtered.append({
                    'node': node,
                    'score': score,
                    'filtered_out': False
                })
            else:
                filtered.append({
                    'node': node,
                    'score': score,
                    'filtered_out': True,
                    'reason': 'Below absolute threshold or matches FP pattern'
                })
        # 返回排序后的结果（根据分数，仅显示未过滤的在前）
        return sorted([c for c in filtered if not c['filtered_out']], key=lambda x: x['score'], reverse=True)

    def _check_absolute_threshold(self, node: str, baseline: Dict, fault: Dict) -> bool:
        """检查指标变化是否超过绝对阈值"""
        for metric, threshold in self.abs_thresholds.items():
            if node.endswith(f".{metric}"):
                base_val = baseline.get(node, 0)
                fault_val = fault.get(node, 0)
                delta = fault_val - base_val
                if delta < threshold:
                    return False
        return True

    def _matches_fp_pattern(self, node: str, baseline: Dict, fault: Dict) -> bool:
        """简单模式匹配（示例）"""
        for pattern in self.fp_patterns:
            # 简化实现：仅检查节点名称模式
            import re
            if re.match(pattern["node_pattern"], node):
                # 这里可以解析 condition 并求值，为简化直接返回True示例匹配
                return True  # 示例：匹配到了，认为是误报模式
        return False

services:

  - name: "web_gateway"
    endpoints: ["/api/*"]

  - name: "service_order"
    endpoints: ["/order", "/cart"]

  - name: "service_payment"
    endpoints: ["/pay", "/wallet"]

  - name: "service_inventory"
    endpoints: ["/stock", "/item"]

experiment:
  baseline_duration_min: 2
  fault_duration_min: 3
  sample_interval_s: 10

analysis:
  anomaly_threshold_sigma: 1.5
  pc_algorithm_alpha: 0.1
  max_condition_set_size: 2
  use_predefined_graph: true

import sys
sys.path.insert(0, '.')

from experiment.runner import ExperimentRunner

def main():
    # 初始化实验运行器
    runner = ExperimentRunner()

    # 定义要注入的故障（模拟支付服务延迟增加）
    fault_spec = {
        "target_service": "service_payment",
        "fault_type": "latency_increase",
        "magnitude": 300  # 增加300ms
    }

    # 运行实验
    exp_data = runner.run_experiment(fault_spec)

    # 分析实验
    result = runner.analyze_experiment(exp_data)

    # 打印结果
    print("\n" + "="*50)
    print("根因分析报告")
    print("="*50)
    print(f"实验ID: {result['experiment_id']}")
    print(f"注入故障: {result['fault_injected']}")
    print(f"检测到异常指标: {result['anomalous_metrics']}")
    print("\n--- 原始根因推断 ---")
    for rc in result['raw_root_causes'][:5]:  # 显示前5个
        print(f"  {rc['node']}: {rc['score']:.2f}")
    print("\n--- 误报过滤后根因 ---")
    if result['filtered_root_causes']:
        for rc in result['filtered_root_causes']:
            print(f"  {rc['node']}: {rc['score']:.2f}")
    else:
        print("  (无)")
    print("\n因果图Mermaid定义已生成，可用于可视化。")

    # 可选：保存结果或触发告警
    # ...

if __name__ == "__main__":
    main()

# 创建项目目录
mkdir chaos_cause
cd chaos_cause

# 创建虚拟环境（Python 3.8+）
python3 -m venv venv
source venv/bin/activate  # Linux/macOS
# venv\Scripts\activate  # Windows

# 安装依赖
pip install numpy pandas networkx scipy statsmodels pyyaml

python main.py

开始实验，注入故障: {"target_service": "service_payment", "fault_type": "latency_increase", "magnitude": 300}
采集基线数据...
注入故障并采集实验数据...
实验 exp_20231027_143022 完成。
分析实验 exp_20231027_143022 ...
检测到异常节点: ['service_payment.latency_ms', 'service_order.latency_ms', 'web_gateway.latency_ms']
原始根因推断结果: [('service_payment.latency_ms', 2.0), ('service_inventory.latency_ms', 1.0), ...]
==================================================
根因分析报告
==================================================
实验ID: exp_20231027_143022
注入故障: {"target_service": "service_payment", "fault_type": "latency_increase", "magnitude": 300}
检测到异常指标: ['service_payment.latency_ms', 'service_order.latency_ms', 'web_gateway.latency_ms']

--- 原始根因推断 ---
  service_payment.latency_ms: 2.00
  service_inventory.latency_ms: 1.00
  ...

--- 误报过滤后根因 ---
  service_payment.latency_ms: 2.00
  ...

因果图Mermaid定义已生成，可用于可视化。

import sys
sys.path.insert(0, '..')
from core.causal_graph import CausalGraph

def test_infer_root_causes():
    cg = CausalGraph()
    nodes = ['A', 'B', 'C', 'D']
    for n in nodes:
        cg.add_node(n)
    cg.add_edge('A', 'B')
    cg.add_edge('B', 'C')
    cg.add_edge('D', 'B')
    # 假设 C 是异常的
    causes = cg.infer_root_causes(['C'], depth_limit=2)
    print(causes)
    # 期望 A 和 D 可能是根因，因为它们是C的祖先
    cause_nodes = [c[0] for c in causes]
    assert 'A' in cause_nodes or 'D' in cause_nodes
    print("test_infer_root_causes passed.")

if __name__ == '__main__':
    test_infer_root_causes()

python tests/test_causal_graph.py