企业级生产环境中基于事件驱动的可观测性建设与故障闭环

event-obs/
├── README.md           # 项目说明（此处不展示内容，因要求禁止）
├── requirements.txt    # Python 依赖清单
├── config.yaml         # 应用配置文件
├── run.py              # 应用主入口
│
├── core/               # 核心抽象与基础组件
│   ├── __init__.py
│   ├── event.py       # 事件基类与数据模型
│   ├── bus.py         # 事件总线（内存实现）
│   └── topology.py    # 系统拓扑图管理
│
├── engine/             # 处理引擎
│   ├── __init__.py
│   ├── rule_engine.py # 规则引擎
│   ├── rules/         # 规则定义
│   │   ├── __init__.py
│   │   ├── base_rule.py
│   │   ├── error_rate_rule.py
│   │   └── latency_rule.py
│   └── rca_engine.py  # 根因分析引擎
│
├── manager/            # 管理模块
│   ├── __init__.py
│   ├── alert_manager.py   # 告警管理器
│   └── incident_manager.py # 故障管理器
│
├── simulator/          # 数据模拟器
│   ├── __init__.py
│   ├── data_simulator.py
│   └── service_repair_simulator.py
│
├── storage/            # 存储抽象（内存实现）
│   ├── __init__.py
│   └── memory_storage.py
│
└── api/                # 对外API（简易Flask实现）
    ├── __init__.py
    └── app.py

Flask==2.3.2
pyyaml==6.0
wheel==0.40.0

app:
  name: "Event-Obs Demo"
  simulation_interval: 2  # 模拟数据生成间隔(秒)

event_bus:
  max_size: 10000

topology:
  # 模拟的服务依赖拓扑， edges: [from, to]
  services: ["frontend", "cart-service", "product-service", "payment-service", "database"]
  edges:

    - ["frontend", "cart-service"]
    - ["frontend", "product-service"]
    - ["cart-service", "payment-service"]
    - ["payment-service", "database"]
    - ["product-service", "database"]

rules:

  - name: "high_error_rate"
    rule_class: "ErrorRateRule"
    enabled: true
    config:
      threshold: 0.1  # 错误率阈值10%
      window_seconds: 30
      service: "*"    # 对所有服务生效

  - name: "high_latency_p95"
    rule_class: "LatencyRule"
    enabled: true
    config:
      percentile: 95
      threshold_ms: 500
      window_seconds: 60
      service: "payment-service"

storage:
  type: "memory"

import time
import uuid
from enum import Enum
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field

class EventType(str, Enum):
    METRIC = "metric"
    LOG = "log"
    TRACE = "trace"
    ALERT = "alert"
    INCIDENT = "incident"

class ObservabilityEvent(BaseModel):
    """可观测性事件基类"""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    type: EventType
    source: str  # 产生此事件的组件，如 service name
    timestamp: int = Field(default_factory=lambda: int(time.time() * 1000))
    data: Dict[str, Any]
    tags: Dict[str, str] = Field(default_factory=dict)

    class Config:
        use_enum_values = True

class MetricEventData(BaseModel):
    name: str  # e.g., `http_requests_total`, `response_time_ms`
    value: float
    labels: Dict[str, str] = Field(default_factory=dict)

class LogEventData(BaseModel):
    level: str  # ERROR, WARN, INFO
    message: str
    logger: str

# Alert 和 Incident 事件在后续manager中定义

import threading
from queue import Queue, Empty
from typing import Callable, List
from core.event import ObservabilityEvent
import logging

logger = logging.getLogger(__name__)

class EventBus:
    """简单内存事件总线，生产-消费者模式"""
    def __init__(self, max_size: int = 10000):
        self.queue = Queue(maxsize=max_size)
        self.subscribers: Dict[str, List[Callable[[ObservabilityEvent], None]]] = {}
        self._lock = threading.RLock()

    def publish(self, event: ObservabilityEvent):
        try:
            self.queue.put_nowait(event)
            logger.debug(f"Event published: {event.type} from {event.source}")
        except Full:
            logger.error("Event bus queue is full, event dropped.")

    def subscribe(self, event_type: str, callback: Callable[[ObservabilityEvent], None]):
        with self._lock:
            if event_type not in self.subscribers:
                self.subscribers[event_type] = []
            self.subscribers[event_type].append(callback)
            logger.info(f"Callback subscribed for event type: {event_type}")

    def start_consuming(self):
        """启动消费线程（简化版，单线程消费）"""
        def _consumer():
            while True:
                try:
                    event = self.queue.get(timeout=1)
                    self._dispatch(event)
                    self.queue.task_done()
                except Empty:
                    continue
                except Exception as e:
                    logger.exception(f"Error consuming event: {e}")

        consumer_thread = threading.Thread(target=_consumer, daemon=True)
        consumer_thread.start()
        logger.info("Event bus consumer started.")

    def _dispatch(self, event: ObservabilityEvent):
        callbacks = self.subscribers.get(event.type, [])
        for callback in callbacks:
            try:
                callback(event)
            except Exception as e:
                logger.exception(f"Error in subscriber callback: {e}")

import networkx as nx
from typing import List, Tuple

class SystemTopology:
    """管理系统服务依赖拓扑图"""
    def __init__(self, services: List[str], edges: List[Tuple[str, str]]):
        self.graph = nx.DiGraph()
        self.graph.add_nodes_from(services)
        self.graph.add_edges_from(edges)

    def get_root_candidates(self, affected_nodes: List[str]) -> List[str]:
        """
        根据受影响节点列表，推测可能的根因节点。
        简单策略：计算受影响节点子图中，入度最小的节点（可能是影响的源头）。
        """
        if not affected_nodes:
            return []
        # 创建受影响节点的诱导子图
        subgraph = self.graph.subgraph(affected_nodes)
        if nx.is_empty(subgraph):
            return affected_nodes

        # 找到子图中入度为0的节点（源头）
        root_candidates = [node for node in subgraph.nodes() if subgraph.in_degree(node) == 0]
        if root_candidates:
            return root_candidates
        # 如果没有入度为0的节点，返回原列表中第一个（降级策略）
        return [affected_nodes[0]]

    def get_downstream_services(self, service: str) -> List[str]:
        """获取指定服务的所有下游依赖（递归）"""
        return list(nx.descendants(self.graph, service))

    def get_upstream_services(self, service: str) -> List[str]:
        """获取指定服务的所有上游调用者（递归）"""
        return list(nx.ancestors(self.graph, service))

    def visualize(self):
        """返回用于可视化的节点和边列表（简化）"""
        nodes = list(self.graph.nodes())
        edges = list(self.graph.edges())
        return nodes, edges

import threading
import time
from collections import defaultdict
from typing import Dict, List, Type
from core.event import ObservabilityEvent, EventType
from engine.rules.base_rule import Rule
import logging

logger = logging.getLogger(__name__)

class RuleEngine:
    """规则引擎，处理Metric事件，生成Alert事件"""
    def __init__(self, event_bus, rules: List[Rule]):
        self.event_bus = event_bus
        self.rules: List[Rule] = rules
        # 用于滑动窗口计算的数据结构 service_name -> [(timestamp, value), ...]
        self.metric_window: Dict[str, List] = defaultdict(list)
        self._lock = threading.RLock()
        self._cleanup_thread = None

    def start(self):
        """启动规则引擎，订阅Metric事件并启动清理线程"""
        self.event_bus.subscribe(EventType.METRIC, self._handle_metric)
        self._start_background_cleanup()
        logger.info("Rule engine started.")

    def _handle_metric(self, event: ObservabilityEvent):
        """处理metric事件"""
        with self._lock:
            key = f"{event.source}:{event.data['name']}"
            self.metric_window[key].append((event.timestamp, event.data['value']))
        # 异步评估所有规则
        self._evaluate_rules(event)

    def _evaluate_rules(self, metric_event: ObservabilityEvent):
        """针对新到的metric，评估所有相关规则"""
        for rule in self.rules:
            if rule.is_match(metric_event):
                if rule.evaluate(self.metric_window):
                    alert_event = rule.generate_alert(metric_event)
                    self.event_bus.publish(alert_event)

    def _start_background_cleanup(self):
        """定期清理过期的metric数据"""
        def _cleanup():
            while True:
                time.sleep(60)
                now_ts = time.time() * 1000
                with self._lock:
                    for key, records in list(self.metric_window.items()):
                        # 移除超过1小时的数据
                        fresh_records = [(ts, v) for ts, v in records if now_ts - ts < 3600*1000]
                        if fresh_records:
                            self.metric_window[key] = fresh_records
                        else:
                            del self.metric_window[key]
        self._cleanup_thread = threading.Thread(target=_cleanup, daemon=True)
        self._cleanup_thread.start()

from typing import Dict, List
from core.event import ObservabilityEvent, EventType, MetricEventData
from engine.rules.base_rule import Rule
import time

class ErrorRateRule(Rule):
    def __init__(self, name: str, config: Dict):
        super().__init__(name, config)
        self.threshold = config.get('threshold', 0.1)
        self.window_seconds = config.get('window_seconds', 30)
        self.target_service = config.get('service', '*')

    def is_match(self, event: ObservabilityEvent) -> bool:
        if event.type != EventType.METRIC:
            return False
        if self.target_service != '*' and event.source != self.target_service:
            return False
        return event.data.get('name') == 'http_requests_total'

    def evaluate(self, metric_window: Dict[str, List]) -> bool:
        now = time.time() * 1000
        window_start = now - (self.window_seconds * 1000)
        total_requests = 0
        error_requests = 0

        # 遍历所有相关metric key
        for key, records in metric_window.items():
            service, metric_name = key.split(':')
            if self.target_service not in ('*', service) or metric_name != 'http_requests_total':
                continue
            # 统计时间窗口内的请求
            for ts, value in records:
                if ts >= window_start:
                    total_requests += value
                    # 假设错误请求数存储在同源的另一个metric中，这里简化处理：
                    # 实际应从 `service:http_errors_total` 中获取
                    # 为演示，我们随机化一个错误率触发点，实际项目需按真实数据计算
                    pass

        # 演示逻辑：如果找到了对应服务的请求，使用一个模拟的错误率
        # 真实场景需要查询 `http_errors_total` 的metric数据
        if total_requests > 10: # 有一定基数才计算
            # 模拟触发：假设错误率是15%，大于阈值10%
            simulated_error_rate = 0.15
            return simulated_error_rate > self.threshold
        return False

    def generate_alert(self, metric_event: ObservabilityEvent) -> ObservabilityEvent:
        from core.event import ObservabilityEvent, EventType
        return ObservabilityEvent(
            type=EventType.ALERT,
            source=f"rule:{self.name}",
            data={
                "title": f"High Error Rate Detected on {metric_event.source}",
                "description": f"Error rate exceeds {self.threshold*100}% in last {self.window_seconds}s.",
                "severity": "critical",
                "service": metric_event.source,
                "triggering_metric": metric_event.data
            },
            tags={"rule": self.name, "type": "error_rate"}
        )

from typing import Dict, List
from core.event import ObservabilityEvent, EventType
from core.topology import SystemTopology
import logging

logger = logging.getLogger(__name__)

class AlertManager:
    """告警管理器，聚合Alert事件，生成Incident（故障）事件"""
    def __init__(self, event_bus, topology: SystemTopology, storage):
        self.event_bus = event_bus
        self.topology = topology
        self.storage = storage
        # 活跃的告警分组: incident_id -> List[alert_event]
        self.active_incidents: Dict[str, List] = {}
        self.incident_timeout = 300  # 5分钟内无新告警，则认为故障结束

    def start(self):
        self.event_bus.subscribe(EventType.ALERT, self._handle_alert)
        logger.info("Alert manager started.")

    def _handle_alert(self, alert_event: ObservabilityEvent):
        logger.warning(f"Alert received: {alert_event.data.get('title')}")
        # 1. 关联：查找是否有相关的、未关闭的故障
        incident_id = self._correlate_with_existing_incident(alert_event)
        if incident_id:
            # 追加告警到现有故障
            self.active_incidents[incident_id].append(alert_event)
            self._update_incident(incident_id)
        else:
            # 2. 创建新的故障
            incident_id = self._create_new_incident([alert_event])
        # 3. 存储或更新
        self.storage.save_alert(alert_event)
        self.storage.update_incident_alerts(incident_id, self.active_incidents.get(incident_id, []))

    def _correlate_with_existing_incident(self, alert_event) -> str:
        """简单关联策略：相同服务或上游/下游服务的未关闭故障"""
        alert_service = alert_event.data.get('service')
        for inc_id, alerts in self.active_incidents.items():
            # 检查此故障是否已超时（已恢复）
            # 此处简化，实际应有更复杂的状态和超时管理
            for a in alerts:
                existing_service = a.data.get('service')
                if (alert_service == existing_service or
                    alert_service in self.topology.get_upstream_services(existing_service) or
                    alert_service in self.topology.get_downstream_services(existing_service)):
                    return inc_id
        return ""

    def _create_new_incident(self, initial_alerts: List[ObservabilityEvent]):
        from core.event import ObservabilityEvent, EventType
        import uuid
        incident_id = f"inc_{uuid.uuid4().hex[:8]}"
        self.active_incidents[incident_id] = initial_alerts

        # 生成故障事件
        affected_services = list(set([a.data.get('service') for a in initial_alerts]))
        incident_event = ObservabilityEvent(
            type=EventType.INCIDENT,
            source="alert_manager",
            data={
                "incident_id": incident_id,
                "title": f"Incident: Multiple alerts on {', '.join(affected_services)}",
                "status": "open",
                "severity": "high",
                "affected_services": affected_services,
                "created_at": int(time.time() * 1000)
            }
        )
        self.event_bus.publish(incident_event)
        logger.critical(f"New incident created: {incident_id}")
        return incident_id

    def _update_incident(self, incident_id: str):
        # 更新故障严重性、状态等，此处简化
        pass

from core.event import ObservabilityEvent, EventType
from core.topology import SystemTopology
import logging

logger = logging.getLogger(__name__)

class RCAEngine:
    """根因分析引擎，处理Incident事件，尝试定位根因"""
    def __init__(self, event_bus, topology: SystemTopology, storage):
        self.event_bus = event_bus
        self.topology = topology
        self.storage = storage

    def start(self):
        self.event_bus.subscribe(EventType.INCIDENT, self._handle_incident)
        logger.info("RCA engine started.")

    def _handle_incident(self, incident_event: ObservabilityEvent):
        incident_id = incident_event.data.get('incident_id')
        affected_services = incident_event.data.get('affected_services', [])
        logger.info(f"Starting RCA for incident {incident_id}, affected: {affected_services}")

        # 核心分析逻辑：基于拓扑图推测根因
        root_candidates = self.topology.get_root_candidates(affected_services)
        analysis_result = {
            "incident_id": incident_id,
            "root_candidate_services": root_candidates,
            "confidence": 0.7,  # 模拟置信度
            "analysis_time": int(time.time() * 1000),
            "logic": "Topology-based root cause inference (in-degree zero nodes in subgraph)."
        }
        # 存储分析结果
        self.storage.save_rca_result(incident_id, analysis_result)
        # 生成诊断报告事件（可触发自动修复或通知）
        self._generate_diagnosis(incident_id, analysis_result)

    def _generate_diagnosis(self, incident_id: str, rca_result: dict):
        diagnosis = {
            "incident_id": incident_id,
            "recommended_actions": [],
            "summary": f"Root cause likely in service(s): {', '.join(rca_result['root_candidate_services'])}."
        }
        # 根据根因服务生成建议动作
        for svc in rca_result['root_candidate_services']:
            diagnosis["recommended_actions"].append(
                f"Restart or check logs of service `{svc}`."
            )
            diagnosis["recommended_actions"].append(
                f"Scale out service `{svc}` if resource metrics indicate high load."
            )
        self.storage.save_diagnosis(incident_id, diagnosis)
        # 发布诊断事件，可供"修复执行器"订阅
        diagnosis_event = ObservabilityEvent(
            type=EventType.ALERT,  # 复用ALERT类型，也可定义新类型
            source="rca_engine",
            data={
                "title": f"Diagnosis for Incident {incident_id}",
                "diagnosis": diagnosis,
                "severity": "medium"
            },
            tags={"type": "diagnosis", "incident_id": incident_id}
        )
        self.event_bus.publish(diagnosis_event)

import random
import threading
import time
from core.event import ObservabilityEvent, EventType, MetricEventData, LogEventData
import logging

logger = logging.getLogger(__name__)

class DataSimulator:
    """模拟生产环境生成可观测性数据"""
    def __init__(self, event_bus, services: list, interval: int = 2):
        self.event_bus = event_bus
        self.services = services
        self.interval = interval
        self.running = False

    def start(self):
        self.running = True
        thread = threading.Thread(target=self._run_simulation, daemon=True)
        thread.start()
        logger.info(f"Data simulator started for services: {self.services}")

    def stop(self):
        self.running = False

    def _run_simulation(self):
        while self.running:
            for service in self.services:
                self._generate_metric(service)
                # 偶尔生成错误日志
                if random.random() < 0.05:
                    self._generate_error_log(service)
            time.sleep(self.interval)

    def _generate_metric(self, service: str):
        # 模拟请求总数
        request_count = random.randint(10, 100)
        metric_event = ObservabilityEvent(
            type=EventType.METRIC,
            source=service,
            data={
                "name": "http_requests_total",
                "value": request_count,
                "labels": {"method": "GET", "path": "/api/v1/items"}
            }
        )
        self.event_bus.publish(metric_event)
        # 模拟响应时间 P95
        latency = random.uniform(50, 600) # 50ms 到 600ms
        latency_event = ObservabilityEvent(
            type=EventType.METRIC,
            source=service,
            data={"name": "response_time_p95_ms", "value": latency}
        )
        self.event_bus.publish(latency_event)

    def _generate_error_log(self, service: str):
        log_event = ObservabilityEvent(
            type=EventType.LOG,
            source=service,
            data={
                "level": "ERROR",
                "message": f"Something went wrong in {service}! Connection timeout.",
                "logger": f"{service}.network"
            }
        )
        self.event_bus.publish(log_event)

from typing import Dict, List, Optional
from core.event import ObservabilityEvent

class MemoryStorage:
    """内存存储，用于演示。生产环境应替换为数据库。"""
    def __init__(self):
        self.alerts: List[ObservabilityEvent] = []
        self.incidents: Dict[str, Dict] = {}  # incident_id -> incident metadata
        self.rca_results: Dict[str, Dict] = {}
        self.diagnoses: Dict[str, Dict] = {}

    def save_alert(self, alert: ObservabilityEvent):
        self.alerts.append(alert)

    def update_incident_alerts(self, incident_id: str, alerts: List[ObservabilityEvent]):
        if incident_id not in self.incidents:
            self.incidents[incident_id] = {"alerts": [], "status": "open"}
        self.incidents[incident_id]["alerts"] = [a.dict() for a in alerts]

    def save_rca_result(self, incident_id: str, result: dict):
        self.rca_results[incident_id] = result

    def save_diagnosis(self, incident_id: str, diagnosis: dict):
        self.diagnoses[incident_id] = diagnosis

    def get_incident(self, incident_id: str) -> Optional[Dict]:
        return self.incidents.get(incident_id)

    def get_all_incidents(self) -> Dict[str, Dict]:
        return self.incidents

from flask import Flask, jsonify, request
import json
from storage.memory_storage import MemoryStorage
from core.topology import SystemTopology

# 假设storage和topology是全局单例，由主程序注入
# 这里为演示，我们创建模拟实例
app = Flask(__name__)
storage = MemoryStorage()  # 应替换为从主app传入
topology = SystemTopology([], [])  # 应替换为从主app传入

@app.route('/api/v1/incidents', methods=['GET'])
def get_incidents():
    incidents = storage.get_all_incidents()
    return jsonify({"incidents": incidents})

@app.route('/api/v1/topology', methods=['GET'])
def get_topology():
    nodes, edges = topology.visualize()
    return jsonify({"nodes": nodes, "edges": edges})

@app.route('/api/v1/diagnosis/<incident_id>', methods=['GET'])
def get_diagnosis(incident_id: str):
    diagnosis = storage.diagnoses.get(incident_id, {})
    return jsonify(diagnosis)

@app.route('/api/v1/simulate/error_injection', methods=['POST'])
def inject_error():
    """模拟故障注入，手动触发一个服务的高错误率"""
    data = request.json
    service = data.get('service', 'payment-service')
    # 这里可以调用模拟器的方法，临时修改其行为，使其产生错误
    # 为简化，我们仅返回一个消息
    return jsonify({"message": f"Error injection requested for {service}. Simulator behavior would be altered."})

if __name__ == '__main__':
    # 注意：实际运行时不直接启动，由主程序run.py集成
    app.run(debug=True, port=5000)

import yaml
import logging
from core.bus import EventBus
from core.topology import SystemTopology
from engine.rule_engine import RuleEngine
from engine.rca_engine import RCAEngine
from manager.alert_manager import AlertManager
from simulator.data_simulator import DataSimulator
from storage.memory_storage import MemoryStorage
from api.app import app as api_app
import threading

# 配置日志
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

def load_rules_from_config(config):
    """动态加载规则类"""
    rules = []
    rule_configs = config.get('rules', [])
    for rc in rule_configs:
        if not rc.get('enabled', True):
            continue
        rule_class_name = rc['rule_class']
        try:
            # 简单映射，实际项目应使用更动态的导入
            if rule_class_name == 'ErrorRateRule':
                from engine.rules.error_rate_rule import ErrorRateRule
                rule_class = ErrorRateRule
            elif rule_class_name == 'LatencyRule':
                from engine.rules.latency_rule import LatencyRule
                rule_class = LatencyRule
            else:
                logger.warning(f"Unknown rule class: {rule_class_name}")
                continue
            rule_instance = rule_class(rc['name'], rc.get('config', {}))
            rules.append(rule_instance)
            logger.info(f"Rule loaded: {rc['name']}")
        except Exception as e:
            logger.exception(f"Failed to load rule {rc['name']}: {e}")
    return rules

def main():
    # 1. 加载配置
    with open('config.yaml', 'r') as f:
        config = yaml.safe_load(f)
    app_config = config['app']
    topo_config = config['topology']

    # 2. 初始化核心组件
    event_bus = EventBus(max_size=config['event_bus']['max_size'])
    topology = SystemTopology(topo_config['services'], topo_config['edges'])
    storage = MemoryStorage()

    # 3. 初始化引擎与管理器
    rules = load_rules_from_config(config)
    rule_engine = RuleEngine(event_bus, rules)
    alert_manager = AlertManager(event_bus, topology, storage)
    rca_engine = RCAEngine(event_bus, topology, storage)

    # 4. 初始化模拟器
    simulator = DataSimulator(event_bus, topo_config['services'], app_config['simulation_interval'])

    # 5. 启动所有组件
    event_bus.start_consuming()
    rule_engine.start()
    alert_manager.start()
    rca_engine.start()
    simulator.start()

    logger.info("====== Event-Obs System Started ======")
    logger.info(f"Simulating services: {topo_config['services']}")
    logger.info("Press Ctrl+C to stop.")

    # 6. 启动API服务（在独立线程）
    api_thread = threading.Thread(target=lambda: api_app.run(
        host='0.0.0.0', port=5000, debug=False, use_reloader=False), daemon=True)
    api_thread.start()
    logger.info("API server started at http://localhost:5000")

    try:
        # 主线程保持运行
        while True:
            import time
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("Shutting down...")
        simulator.stop()

if __name__ == "__main__":
    main()

pip install -r requirements.txt

python run.py

curl -X POST http://localhost:5000/api/v1/simulate/error_injection \
         -H "Content-Type: application/json" \
         -d '{"service": "payment-service"}'