d_root_cause.py

import numpy as np
import random
import time
import logging
from collections import defaultdict, Counter
from typing import Dict, List, Tuple, Optional
import networkx as nx
import pandas as pd

from b_graph_build import CausalGraphConstructor


class RandomWalkRootCauseLocalizer:
    def __init__(self, rho: float = 0.5, restart_prob: float = 0.15):
        """
        初始化根因定位器

        Args:
            rho: 控制反向步骤影响的参数 [0,1]
            restart_prob: RWR的重启概率
        """
        self.rho = rho
        self.restart_prob = restart_prob
        self.transfer_probs = {}
        self.alias_tables = {}

        # 设置日志记录器
        self.logger = logging.getLogger(f"{__name__}.{self.__class__.__name__}")
        if not self.logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.INFO)

    def localize_root_causes(self,
                             graph: nx.DiGraph,
                             anomaly_node_id: str,
                             num_walks: int = 1000,
                             walk_length: int = 50) -> Dict[str, List[Tuple[str, int]]]:
        """
        执行根因定位的完整流程

        Args:
            graph: 带有异常分数的因果图
            anomaly_node_id: 异常节点ID
            num_walks: 随机游走次数
            walk_length: 每次游走的最大步数

        Returns:
            按节点类型分组的根因排序结果
        """
        total_start_time = time.perf_counter()

        self.logger.info(f"开始根因定位分析 - 异常节点: {anomaly_node_id}")
        self.logger.info(f"图规模: {graph.number_of_nodes()}个节点, {graph.number_of_edges()}条边")
        self.logger.info(f"游走参数: {num_walks}次游走, 每次最大{walk_length}步")

        # Stage 1: 计算转移概率
        stage1_start = time.perf_counter()
        self._calculate_transfer_probabilities(graph, anomaly_node_id)
        stage1_time = time.perf_counter() - stage1_start
        self.logger.info(f"Stage 1 - 转移概率计算完成: {stage1_time:.4f}秒")

        # Stage 2: 执行随机游走
        stage2_start = time.perf_counter()
        visit_counts = self._execute_random_walks(graph, anomaly_node_id, num_walks, walk_length)
        stage2_time = time.perf_counter() - stage2_start
        self.logger.info(f"Stage 2 - 随机游走完成: {stage2_time:.4f}秒")

        # Stage 3: 根因排序
        stage3_start = time.perf_counter()
        ranked_results = self._rank_root_causes(graph, visit_counts)
        stage3_time = time.perf_counter() - stage3_start
        self.logger.info(f"Stage 3 - 根因排序完成: {stage3_time:.4f}秒")

        total_time = time.perf_counter() - total_start_time

        # 输出详细统计信息
        self.logger.info("=== 根因定位性能统计 ===")
        self.logger.info(f"总耗时: {total_time:.4f}秒")
        self.logger.info(f"  - 转移概率计算: {stage1_time:.4f}秒 ({stage1_time / total_time * 100:.1f}%)")
        self.logger.info(f"  - 随机游走执行: {stage2_time:.4f}秒 ({stage2_time / total_time * 100:.1f}%)")
        self.logger.info(f"  - 结果排序处理: {stage3_time:.4f}秒 ({stage3_time / total_time * 100:.1f}%)")

        # 输出结果统计
        total_visited_nodes = sum(len(nodes) for nodes in ranked_results.values())
        self.logger.info(f"访问节点统计: 总计{total_visited_nodes}个节点")
        for node_type, nodes in ranked_results.items():
            if nodes:
                self.logger.info(f"  - {node_type}: {len(nodes)}个节点, 最高访问次数: {nodes[0][1]}")

        return ranked_results

    def _calculate_transfer_probabilities(self, graph: nx.DiGraph, anomaly_node_id: str):
        """Stage 1: 计算转移概率"""
        start_time = time.perf_counter()
        self.transfer_probs = {}

        nodes_count = graph.number_of_nodes()
        edges_with_scores = 0

        # 统计有异常分数的边数量
        for _, _, data in graph.edges(data=True):
            if 'anomaly_score' in data:
                edges_with_scores += 1

        self.logger.info(f"开始计算{nodes_count}个节点的转移概率，{edges_with_scores}条边有异常分数")

        prob_calc_time = 0
        alias_build_time = 0

        for i, node in enumerate(graph.nodes()):
            node_start = time.perf_counter()

            self.transfer_probs[node] = {}
            node_probs = {}

            # 1) Forward step: 从当前节点到其后继节点
            total_forward_prob = 0.0
            successors = list(graph.successors(node))

            for successor in successors:
                edge_data = graph[node][successor]
                anomaly_score = edge_data.get('anomaly_score', 0.0)
                if anomaly_score > 0:
                    node_probs[successor] = anomaly_score
                    total_forward_prob += anomaly_score

            # 2) Backward step: 从前驱节点返回
            predecessors = list(graph.predecessors(node))
            for predecessor in predecessors:
                if predecessor not in node_probs:  # 避免重复添加
                    edge_data = graph[predecessor][node]
                    anomaly_score = edge_data.get('anomaly_score', 0.0)
                    if anomaly_score > 0:
                        backward_prob = self.rho * anomaly_score
                        if backward_prob > 0:
                            node_probs[predecessor] = backward_prob
                            total_forward_prob += backward_prob

            # 3) Self step: 计算停留在当前节点的概率
            max_upstream_score = 0.0
            max_downstream_score = 0.0

            # 计算上游最大异常分数
            for predecessor in predecessors:
                edge_data = graph[predecessor][node]
                score = edge_data.get('anomaly_score', 0.0)
                max_upstream_score = max(max_upstream_score, score)

            # 计算下游最大异常分数
            for successor in successors:
                edge_data = graph[node][successor]
                score = edge_data.get('anomaly_score', 0.0)
                max_downstream_score = max(max_downstream_score, score)

            # 自停留概率
            self_prob = max(0, max_upstream_score - max_downstream_score)
            if self_prob > 0:
                node_probs[node] = self_prob
                total_forward_prob += self_prob

            # 归一化概率
            if total_forward_prob > 0:
                for target in node_probs:
                    node_probs[target] = node_probs[target] / total_forward_prob

            self.transfer_probs[node] = node_probs

            prob_calc_time += time.perf_counter() - node_start

            # 构建Alias Table以加速采样
            alias_start = time.perf_counter()
            self._build_alias_table(node, node_probs)
            alias_build_time += time.perf_counter() - alias_start

            # 定期输出进度
            if (i + 1) % max(1, nodes_count // 10) == 0:
                progress = (i + 1) / nodes_count * 100
                elapsed = time.perf_counter() - start_time
                self.logger.debug(f"转移概率计算进度: {progress:.1f}% ({i + 1}/{nodes_count}), 耗时: {elapsed:.2f}秒")

        total_time = time.perf_counter() - start_time
        self.logger.info(f"转移概率计算详细耗时:")
        self.logger.info(f"  - 概率计算: {prob_calc_time:.4f}秒 ({prob_calc_time / total_time * 100:.1f}%)")
        self.logger.info(f"  - Alias表构建: {alias_build_time:.4f}秒 ({alias_build_time / total_time * 100:.1f}%)")

    def _build_alias_table(self, node: str, probs: Dict[str, float]):
        """构建Alias Table用于O(1)时间复杂度的采样"""
        if not probs:
            self.alias_tables[node] = ([], [], [])
            return

        nodes = list(probs.keys())
        probabilities = list(probs.values())
        n = len(nodes)

        if n == 0:
            self.alias_tables[node] = ([], [], [])
            return

        # 初始化
        scaled_probs = [p * n for p in probabilities]
        alias = [-1] * n
        prob = [0.0] * n

        # 分离小于1和大于等于1的概率
        small = []
        large = []

        for i, p in enumerate(scaled_probs):
            if p < 1.0:
                small.append(i)
            else:
                large.append(i)

        # 构建alias table
        while small and large:
            l = small.pop()
            g = large.pop()

            prob[l] = scaled_probs[l]
            alias[l] = g

            scaled_probs[g] = scaled_probs[g] + scaled_probs[l] - 1.0

            if scaled_probs[g] < 1.0:
                small.append(g)
            else:
                large.append(g)

        while large:
            prob[large.pop()] = 1.0

        while small:
            prob[small.pop()] = 1.0

        self.alias_tables[node] = (nodes, prob, alias)

    def _sample_next_node(self, current_node: str) -> Optional[str]:
        """使用Alias Method采样下一个节点"""
        if current_node not in self.alias_tables:
            return None

        nodes, prob, alias = self.alias_tables[current_node]

        if not nodes:
            return None

        n = len(nodes)
        i = random.randint(0, n - 1)

        if random.random() < prob[i]:
            return nodes[i]
        else:
            return nodes[alias[i]] if alias[i] != -1 else nodes[i]

    def _execute_random_walks(self,
                              graph: nx.DiGraph,
                              anomaly_node_id: str,
                              num_walks: int,
                              walk_length: int) -> Dict[str, int]:
        """Stage 2: 执行随机游走"""
        start_time = time.perf_counter()
        visit_counts = defaultdict(int)

        self.logger.info(f"开始执行{num_walks}次随机游走，每次最大{walk_length}步")

        total_steps = 0
        total_restarts = 0
        sampling_time = 0

        # 批量处理进度报告
        progress_interval = max(1, num_walks // 20)

        for walk_id in range(num_walks):
            current_node = anomaly_node_id
            walk_steps = 0

            for step in range(walk_length):
                visit_counts[current_node] += 1
                total_steps += 1
                walk_steps += 1

                # RWR重启判断
                if random.random() < self.restart_prob:
                    current_node = anomaly_node_id
                    total_restarts += 1
                    continue

                # 采样下一个节点
                sampling_start = time.perf_counter()
                next_node = self._sample_next_node(current_node)
                sampling_time += time.perf_counter() - sampling_start

                if next_node is None:
                    break  # 无法继续游走
                else:
                    current_node = next_node

            # 定期输出进度
            if (walk_id + 1) % progress_interval == 0:
                progress = (walk_id + 1) / num_walks * 100
                elapsed = time.perf_counter() - start_time
                avg_steps_per_walk = total_steps / (walk_id + 1)
                self.logger.debug(f"随机游走进度: {progress:.1f}% ({walk_id + 1}/{num_walks}), "
                                  f"平均每次游走{avg_steps_per_walk:.1f}步, 耗时: {elapsed:.2f}秒")

        total_time = time.perf_counter() - start_time
        avg_steps_per_walk = total_steps / num_walks
        restart_rate = total_restarts / total_steps * 100

        self.logger.info(f"随机游走执行完成:")
        self.logger.info(f"  - 总步数: {total_steps:,} (平均每次游走{avg_steps_per_walk:.1f}步)")
        self.logger.info(f"  - 重启次数: {total_restarts:,} (重启率{restart_rate:.1f}%)")
        self.logger.info(f"  - 节点采样耗时: {sampling_time:.4f}秒 ({sampling_time / total_time * 100:.1f}%)")
        self.logger.info(f"  - 访问的不同节点数: {len(visit_counts)}")

        return dict(visit_counts)

    def _rank_root_causes(self,
                          graph: nx.DiGraph,
                          visit_counts: Dict[str, int]) -> Dict[str, List[Tuple[str, int]]]:
        """Stage 3: 根据节点类型分组并排序根因"""
        start_time = time.perf_counter()

        grouped_results = {
            'service': [],
            'host': [],
            'metric': [],
            'fault': []
        }

        # 按节点类型分组
        type_counts = defaultdict(int)
        unknown_type_nodes = []

        for node_id, count in visit_counts.items():
            if node_id not in graph.nodes():
                unknown_type_nodes.append(node_id)
                continue

            node_data = graph.nodes[node_id]
            node_type = node_data.get('node_type', 'unknown')
            type_counts[node_type] += 1

            if node_type in grouped_results:
                grouped_results[node_type].append((node_id, count))

        # 对每组按访问次数降序排列
        sort_start = time.perf_counter()
        for node_type in grouped_results:
            grouped_results[node_type].sort(key=lambda x: x[1], reverse=True)
        sort_time = time.perf_counter() - sort_start

        total_time = time.perf_counter() - start_time

        self.logger.info(f"根因排序完成:")
        self.logger.info(f"  - 节点类型分组: {dict(type_counts)}")
        if unknown_type_nodes:
            self.logger.warning(f"  - 未知类型节点数: {len(unknown_type_nodes)}")
        self.logger.info(f"  - 排序耗时: {sort_time:.4f}秒")
        self.logger.info(f"  - 总耗时: {total_time:.4f}秒")

        return grouped_results

    def get_propagation_path(self,
                             graph: nx.DiGraph,
                             ranked_results: Dict[str, List[Tuple[str, int]]],
                             top_k: int = 5) -> List[Tuple[str, str, float]]:
        """获取异常传播路径"""
        start_time = time.perf_counter()

        propagation_paths = []

        # 获取前k个最可能的根因节点
        all_candidates = []
        for node_type, nodes in ranked_results.items():
            all_candidates.extend(nodes[:top_k])

        # 按访问次数排序
        all_candidates.sort(key=lambda x: x[1], reverse=True)
        top_candidates = [node_id for node_id, _ in all_candidates[:top_k]]

        # 构建传播路径
        path_count = 0
        for i, source in enumerate(top_candidates):
            for j, target in enumerate(top_candidates):
                if i != j and graph.has_edge(source, target):
                    edge_data = graph[source][target]
                    anomaly_score = edge_data.get('anomaly_score', 0.0)
                    propagation_paths.append((source, target, anomaly_score))
                    path_count += 1

        # 按异常分数排序
        propagation_paths.sort(key=lambda x: x[2], reverse=True)

        elapsed_time = time.perf_counter() - start_time
        self.logger.info(f"传播路径分析完成: 发现{path_count}条路径, 耗时: {elapsed_time:.4f}秒")

        return propagation_paths

    def generate_summary_report(self,
                                ranked_results: Dict[str, List[Tuple[str, int]]],
                                top_k: int = 10) -> Dict:
        """生成根因分析摘要报告"""
        start_time = time.perf_counter()

        summary = {
            'timestamp': pd.Timestamp.now().isoformat(),
            'top_root_causes': {},
            'statistics': {}
        }

        # 获取每种类型的前k个根因
        for node_type, candidates in ranked_results.items():
            if candidates:
                summary['top_root_causes'][node_type] = candidates[:top_k]
                summary['statistics'][f'{node_type}_count'] = len(candidates)
            else:
                summary['top_root_causes'][node_type] = []
                summary['statistics'][f'{node_type}_count'] = 0

        # 计算总体统计
        total_candidates = sum(len(candidates) for candidates in ranked_results.values())
        summary['statistics']['total_candidates'] = total_candidates

        # 找出访问次数最高的节点作为最可能的根因
        all_candidates = []
        for candidates in ranked_results.values():
            all_candidates.extend(candidates)

        if all_candidates:
            most_likely_root_cause = max(all_candidates, key=lambda x: x[1])
            summary['most_likely_root_cause'] = {
                'node_id': most_likely_root_cause[0],
                'visit_count': most_likely_root_cause[1]
            }

        elapsed_time = time.perf_counter() - start_time
        self.logger.debug(f"摘要报告生成完成: 耗时{elapsed_time:.4f}秒")

        return summary


# 为CausalGraphConstructor类添加根因定位方法
def add_root_cause_localization_methods():
    """为CausalGraphConstructor类添加根因定位方法"""

    def localize_root_causes(self,
                             anomaly_node_id: str,
                             num_walks: int = 1000,
                             walk_length: int = 50,
                             rho: float = 0.5,
                             restart_prob: float = 0.15) -> Dict[str, List[Tuple[str, int]]]:
        """执行根因定位"""
        localizer = RandomWalkRootCauseLocalizer(rho, restart_prob)
        return localizer.localize_root_causes(
            self.graph, anomaly_node_id, num_walks, walk_length
        )

    def get_propagation_analysis(self,
                                 anomaly_node_id: str,
                                 num_walks: int = 1000,
                                 walk_length: int = 50,
                                 top_k: int = 5) -> Dict:
        """获取完整的传播分析结果"""
        analysis_start = time.perf_counter()

        logger = logging.getLogger(f"{__name__}.CausalGraphConstructor")
        logger.info(f"开始传播分析: 异常节点={anomaly_node_id}")

        localizer = RandomWalkRootCauseLocalizer()

        # 执行根因定位
        localization_start = time.perf_counter()
        ranked_results = localizer.localize_root_causes(
            self.graph, anomaly_node_id, num_walks, walk_length
        )
        localization_time = time.perf_counter() - localization_start

        # 只关注服务层的结果
        service_results = ranked_results.get('service', [])
        service_ranking = [node_id for node_id, _ in service_results]

        # 获取传播路径
        path_start = time.perf_counter()
        propagation_paths = localizer.get_propagation_path(
            self.graph, ranked_results, top_k
        )
        path_time = time.perf_counter() - path_start

        # 生成摘要报告
        report_start = time.perf_counter()
        summary_report = localizer.generate_summary_report(ranked_results, top_k)
        report_time = time.perf_counter() - report_start

        # 添加服务层专用信息
        summary_report['service_root_cause_ranking'] = service_ranking
        if service_results:
            summary_report['most_likely_root_cause'] = service_results[0][0]

        total_time = time.perf_counter() - analysis_start

        logger.info(f"传播分析完成: 总耗时{total_time:.4f}秒")
        logger.info(f"  - 根因定位: {localization_time:.4f}秒 ({localization_time / total_time * 100:.1f}%)")
        logger.info(f"  - 传播路径: {path_time:.4f}秒 ({path_time / total_time * 100:.1f}%)")
        logger.info(f"  - 摘要报告: {report_time:.4f}秒 ({report_time / total_time * 100:.1f}%)")
        logger.info(f"服务层根因候选数: {len(service_ranking)}")

        return {
            'ranked_results': ranked_results,
            'propagation_paths': propagation_paths,
            'summary_report': summary_report,
            'service_ranking': service_ranking,
            'performance_metrics': {
                'total_time': total_time,
                'localization_time': localization_time,
                'path_analysis_time': path_time,
                'report_generation_time': report_time
            }
        }

    # 动态添加方法到类
    CausalGraphConstructor.localize_root_causes = localize_root_causes
    CausalGraphConstructor.get_propagation_analysis = get_propagation_analysis


# 调用函数添加方法
add_root_cause_localization_methods()