大规模分布式系统下根因分析引擎的延迟与吞吐权衡及可观测性集成

rca_engine/
├── engine/
│   ├── __init__.py
│   ├── causal_graph.py          # 因果图定义与构建
│   ├── anomaly_detector.py      # 异常检测
│   ├── root_cause_analyzer.py   # 根因定位（随机游走）
│   ├── throughput_controller.py # 吞吐控制（动态采样）
│   └── latency_optimizer.py     # 延迟优化（超时与批处理）
├── observability/
│   ├── __init__.py
│   ├── tracer.py                # OpenTelemetry tracer配置
│   └── metrics.py               # 自定义指标
├── config/
│   └── settings.py              # 全局配置
├── tests/
│   ├── test_causal_graph.py
│   └── test_engine_integration.py
├── main.py                      # 入口：模拟事件流
├── requirements.txt

import os

class Settings:
    # 因果图配置
    CAUSAL_GRAPH_MAX_NODES = 100
    
    # 异常检测配置
    ANOMALY_WINDOW_SIZE = 60      # 60秒滑动窗口
    ANOMALY_ZSCORE_THRESHOLD = 3.0
    
    # 根因分析配置
    RCA_TOP_K = 5
    RCA_MAX_ITERATIONS = 50
    RCA_DAMPING_FACTOR = 0.85
    
    # 吞吐控制
    BASE_SAMPLE_RATE = 1.0       # 初始采样率（1.0 = 100%）
    QUEUE_SIZE_THRESHOLD = 1000  # 队列积压触发降采样
    MIN_SAMPLE_RATE = 0.1
    
    # 延迟优化
    RCA_TIMEOUT_MS = 500         # 推理超时
    BATCH_MAX_SIZE = 20          # 批处理最大事件数
    BATCH_WINDOW_MS = 100        # 批处理等待时间
    
    # 可观测性
    OTEL_SERVICE_NAME = "rca-engine"
    OTEL_EXPORTER_OTLP_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4318")

settings = Settings()

import networkx as nx
from typing import Dict, List, Tuple
from config.settings import settings

class CausalGraph:
    """基于服务调用关系的因果图（有向无环图）"""
    
    def __init__(self):
        self._graph = nx.DiGraph()
        self._causal_strength: Dict[Tuple[str, str], float] = {}  # (source, target) -> strength
        
    def add_dependency(self, source: str, target: str, strength: float = 1.0):
        """添加依赖边：source调用target，strength表示因果强度（基于调用频次或错误传播）"""
        self._graph.add_edge(source, target)
        self._causal_strength[(source, target)] = strength
        
    def get_ancestors(self, node: str) -> List[str]:
        """获取所有祖先节点（可能导致异常的根因候选）"""
        ancestors = set()
        for n in self._graph.nodes:
            if nx.has_path(self._graph, n, node):
                ancestors.add(n)
        ancestors.discard(node)
        return list(ancestors)
    
    def causal_influence(self, source: str, target: str) -> float:
        """计算source到target的累计因果影响（沿所有路径）"""
        if source == target:
            return 1.0
        try:
            paths = list(nx.all_simple_paths(self._graph, source, target, cutoff=5))
        except nx.NetworkXNoPath:
            return 0.0
        total = 0.0
        for path in paths:
            influence = 1.0
            for i in range(len(path)-1):
                influence *= self._causal_strength.get((path[i], path[i+1]), 1.0)
            total += influence
        return total
    
    @property
    def nodes(self):
        return list(self._graph.nodes)

import numpy as np
from collections import deque
from typing import Dict
from config.settings import settings

class AnomalyDetector:
    """基于滑动窗口Z-Score的异常检测"""
    
    def __init__(self):
        self._windows: Dict[str, deque] = {}  # node -> deque of recent values
    
    def update(self, node: str, value: float) -> bool:
        """更新节点指标，若异常返回True"""
        if node not in self._windows:
            self._windows[node] = deque(maxlen=settings.ANOMALY_WINDOW_SIZE)
        window = self._windows[node]
        window.append(value)
        if len(window) < 10:  # 预热期
            return False
        mean = np.mean(window)
        std = np.std(window)
        if std == 0:
            return False
        zscore = (value - mean) / std
        return abs(zscore) > settings.ANOMALY_ZSCORE_THRESHOLD

from typing import List, Dict, Tuple
import random
from config.settings import settings
from engine.causal_graph import CausalGraph

class RootCauseAnalyzer:
    """基于随机游走的根因定位"""
    
    def __init__(self, causal_graph: CausalGraph):
        self.graph = causal_graph
    
    def locate(self, anomaly_nodes: List[str]) -> Dict[str, float]:
        """
        对每个异常节点，定位根因候选并评分
        返回: {候选节点: 评分}
        """
        scores = {}
        for anomaly in anomaly_nodes:
            ancestors = self.graph.get_ancestors(anomaly)
            if not ancestors:
                continue
            # 随机游走模拟
            visit_counts = {node: 0 for node in ancestors}
            for _ in range(settings.RCA_MAX_ITERATIONS):
                current = random.choice(ancestors)
                # 沿因果边向下游传播
                while random.random() < settings.RCA_DAMPING_FACTOR:
                    successors = list(self.graph._graph.successors(current))
                    if not successors:
                        break
                    # 按因果强度加权选择
                    weights = [self.graph._causal_strength.get((current, s), 1.0) for s in successors]
                    total = sum(weights)
                    probs = [w/total for w in weights]
                    current = random.choices(successors, weights=probs, k=1)[0]
                    if current in ancestors:
                        visit_counts[current] += 1
                    if current == anomaly:
                        break
            # 归一化评分
            max_visits = max(visit_counts.values()) if visit_counts else 1
            for node, count in visit_counts.items():
                scores[node] = scores.get(node, 0) + count / max_visits
        # 归一化到[0,1]
        if scores:
            max_score = max(scores.values())
            scores = {k: v/max_score for k, v in scores.items()}
        return dict(sorted(scores.items(), key=lambda x: x[1], reverse=True)[:settings.RCA_TOP_K])

import asyncio
from typing import Callable, Awaitable
from config.settings import settings

class ThroughputController:
    """
    动态采样控制器：基于队列长度调整采样率，平衡延迟与吞吐
    """
    
    def __init__(self, queue: asyncio.Queue):
        self._queue = queue
        self._sample_rate = settings.BASE_SAMPLE_RATE
    
    async def should_sample(self) -> bool:
        """判断当前事件是否应该被处理（采样决策）"""
        qsize = self._queue.qsize()
        # 队列积压越大，采样率越低
        if qsize > settings.QUEUE_SIZE_THRESHOLD:
            self._sample_rate = max(
                settings.MIN_SAMPLE_RATE,
                settings.BASE_SAMPLE_RATE * (settings.QUEUE_SIZE_THRESHOLD / qsize)
            )
        else:
            self._sample_rate = settings.BASE_SAMPLE_RATE
        return random.random() < self._sample_rate  # 需要import random
    
    @property
    def current_sample_rate(self) -> float:
        return self._sample_rate

import asyncio
from typing import List, Any, Dict
from config.settings import settings

class LatencyOptimizer:
    """
    延迟优化：批量聚合+超时熔断
    将多个异常事件合并为一次因果推理，并限制单次推理时间
    """
    
    def __init__(self, rca_callback: Callable[[List[str]], Awaitable[Dict]]):
        self._callback = rca_callback
        self._pending_anomalies: List[str] = []
        self._lock = asyncio.Lock()
        self._batch_event = asyncio.Event()
        self._running = False
    
    async def add_anomaly(self, node: str):
        """将异常节点加入批处理队列"""
        async with self._lock:
            self._pending_anomalies.append(node)
            if len(self._pending_anomalies) >= settings.BATCH_MAX_SIZE:
                self._batch_event.set()
            else:
                # 定时触发（BATCH_WINDOW_MS）
                asyncio.create_task(self._delayed_trigger())
    
    async def _delayed_trigger(self):
        await asyncio.sleep(settings.BATCH_WINDOW_MS / 1000.0)
        self._batch_event.set()
    
    async def _batch_processor(self):
        """后台循环：等待batch事件并执行推理"""
        while True:
            await self._batch_event.wait()
            self._batch_event.clear()
            async with self._lock:
                batch = self._pending_anomalies.copy()
                self._pending_anomalies.clear()
            if batch:
                # 执行RCA，带超时
                try:
                    result = await asyncio.wait_for(
                        self._callback(batch),
                        timeout=settings.RCA_TIMEOUT_MS / 1000.0
                    )
                    # 处理结果（例如发送告警）
                    print(f"RCA result for batch {batch}: {result}")
                except asyncio.TimeoutError:
                    print(f"RCA timed out for batch {batch}, using fallback")
                    # 超时时，使用简单启发式：返回最早异常节点
                    result = {batch[0]: 1.0}
                    print(f"Fallback result: {result}")
    
    def start(self):
        self._running = True
        asyncio.create_task(self._batch_processor())

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from config.settings import settings

def init_tracer():
    provider = TracerProvider()
    exporter = OTLPSpanExporter(endpoint=settings.OTEL_EXPORTER_OTLP_ENDPOINT)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return trace.get_tracer(__name__)

from opentelemetry import metrics
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from config.settings import settings

def init_metrics():
    exporter = OTLPMetricExporter(endpoint=settings.OTEL_EXPORTER_OTLP_ENDPOINT)
    reader = PeriodicExportingMetricReader(exporter, export_interval_millis=5000)
    provider = MeterProvider(metric_readers=[reader])
    metrics.set_meter_provider(provider)
    meter = metrics.get_meter(__name__)
    
    # 创建自定义指标
    rca_latency = meter.create_histogram(
        "rca.latency.ms",
        description="Root cause analysis latency in milliseconds",
        unit="ms"
    )
    throughput_rate = meter.create_counter(
        "rca.events.processed",
        description="Number of events processed per second",
        unit="1"
    )
    sample_rate_gauge = meter.create_observable_gauge(
        "rca.sample_rate",
        description="Current dynamic sample rate",
        callbacks=[]  # 需要外部更新
    )
    return meter, rca_latency, throughput_rate, sample_rate_gauge

import asyncio
import random
from config.settings import settings
from engine.causal_graph import CausalGraph
from engine.anomaly_detector import AnomalyDetector
from engine.root_cause_analyzer import RootCauseAnalyzer
from engine.throughput_controller import ThroughputController
from engine.latency_optimizer import LatencyOptimizer
from observability.tracer import init_tracer
from observability.metrics import init_metrics

# 异步包装RCA调用
async def rca_callback(anomaly_batch):
    tracer = init_tracer()
    with tracer.start_as_current_span("rca_batch") as span:
        span.set_attribute("batch_size", len(anomaly_batch))
        result = analyzer.locate(anomaly_batch)
        span.set_attribute("top_candidate", list(result.keys())[0] if result else "none")
    return result

async def simulate_metrics_event(causal_graph, detector, tracer, meter_histogram):
    """模拟一个指标事件：随机选择一个节点，产生随机值，可能触发异常"""
    node = random.choice(causal_graph.nodes)
    value = random.gauss(100, 15)  # 正常分布
    is_anomaly = detector.update(node, value)
    if is_anomaly:
        # 通过吞吐控制决定是否处理此异常
        if await throughput_ctrl.should_sample():
            # 记录指标：处理的事件数
            meter_histogram.add(1)  # 实际上历史图需要绑定attributes，此处简化
            # 交给延迟优化器（批处理）
            await latency_opt.add_anomaly(node)

async def main():
    # 初始化可观测性
    tracer = init_tracer()
    meter, rca_latency_hist, throughput_counter, _ = init_metrics()
    
    # 构建因果图（示例依赖）
    graph = CausalGraph()
    services = ["frontend", "auth", "cart", "payment", "notification", "db"]
    for s in services:
        pass
    graph.add_dependency("frontend", "auth", strength=0.5)
    graph.add_dependency("frontend", "cart", strength=0.8)
    graph.add_dependency("cart", "payment", strength=0.9)
    graph.add_dependency("cart", "notification", strength=0.3)
    graph.add_dependency("payment", "db", strength=0.95)
    graph.add_dependency("auth", "db", strength=0.2)
    
    # 初始化引擎组件
    detector = AnomalyDetector()
    global analyzer
    analyzer = RootCauseAnalyzer(graph)
    
    # 吞吐控制
    event_queue = asyncio.Queue()
    global throughput_ctrl
    throughput_ctrl = ThroughputController(event_queue)
    
    # 延迟优化
    global latency_opt
    latency_opt = LatencyOptimizer(rca_callback)
    latency_opt.start()
    
    # 模拟持续指标事件流（每秒10个事件）
    async def event_stream():
        while True:
            await simulate_metrics_event(graph, detector, tracer, throughput_counter)
            await asyncio.sleep(0.1)  # 100ms per event -> 10 events/s
    
    # 运行3分钟示例
    stream_task = asyncio.create_task(event_stream())
    await asyncio.sleep(180)
    stream_task.cancel()
    print("Demo finished.")

if __name__ == "__main__":
    asyncio.run(main())

opentelemetry-api==1.24.0
opentelemetry-sdk==1.24.0
opentelemetry-exporter-otlp-proto-http==1.24.0
networkx==3.2.1
numpy==1.26.4

# 创建虚拟环境
python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate

# 安装依赖
pip install -r requirements.txt

# 运行主程序（需要配置OpenTelemetry Collector endpoint，若没有，设置OTEL_EXPORTER_OTLP_ENDPOINT为空可运行）
export OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4318   # 可选
python main.py

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

def test_causal_influence():
    graph = CausalGraph()
    graph.add_dependency("A", "B")
    graph.add_dependency("B", "C")
    assert graph.causal_influence("A", "C") == 1.0
    assert graph.causal_influence("C", "A") == 0.0

def test_ancestors():
    graph = CausalGraph()
    graph.add_dependency("A", "B")
    graph.add_dependency("B", "C")
    graph.add_dependency("D", "C")
    ancestors = graph.get_ancestors("C")
    assert "A" in ancestors
    assert "B" in ancestors
    assert "D" in ancestors
    assert "C" not in ancestors

可运行 `pytest tests/` 执行测试。