故障自愈与自动恢复机制：从手动重启到无人值守-尧图网络科技

故障自愈与自动恢复机制：从手动重启到无人值守

一、凌晨3点的PagerDuty告警

"Judge Worker 集群中 3 个 Pod 挂了，评测队列积压 1200+。"

这是实习第三个月遇上的真实场景。从接到告警到手动重启 Pod，花了 18 分钟——其中有 15 分钟在"找问题"，只有 3 分钟在"修问题"。如果系统能自动发现异常、自动隔离故障节点、自动恢复服务，用户甚至不会感知到中间发生了故障。

flowchart TB A[健康检查] --> B{检测到异常?} B -->|否| A B -->|是| C[隔离故障节点] C --> D{故障类型?} D -->|进程崩溃| E[K8s 自动重启 Pod] D -->|内存泄漏| F[自动扩容替换节点] D -->|依赖服务不可用| G[激活降级策略] D -->|磁盘满| H[自动清理日志] E --> I[等待健康检查通过] F --> I G --> I H --> I I --> J{恢复成功?} J -->|是| K[恢复正常流量] J -->|否| L{重试次数 < N?} L -->|是| M[等待 backoff 后重试] M --> C L -->|否| N[升级为人工介入告警] style C fill:#ffc style K fill:#cfc style N fill:#fcc

二、故障自愈的三个层次

L1 进程级自愈：Pod 崩溃时，Kubernetes 的 liveness probe 检测到进程退出，自动重启。最基础的自愈，但无法处理"进程活着但行为异常"的情况。

L2 服务级自愈：基于健康探测指标（响应延迟、错误率、内存使用率）判断服务是否健康。不健康时自动替换节点、调整权重。

L3 业务级自愈：根据业务指标（评测任务成功率、用户满意度）判断异常。触发业务层面的降级（如关闭非核心功能、启动备用链路）。

三、故障自愈系统的实现

""" 故障自愈管理器 三层自愈策略 + 断路器 + 自动扩容 """ import time import threading import random from typing import Dict, List, Callable, Optional from dataclasses import dataclass, field from enum import Enum class NodeStatus(Enum): HEALTHY = "healthy" DEGRADED = "degraded" UNHEALTHY = "unhealthy" ISOLATED = "isolated" @dataclass class ServiceNode: """服务节点""" node_id: str status: NodeStatus = NodeStatus.HEALTHY health_score: float = 1.0 error_count: int = 0 last_heartbeat: float = field(default_factory=time.time) failure_count: int = 0 restart_count: int = 0 @dataclass class SelfHealingConfig: """自愈配置""" max_health_check_failures: int = 3 health_check_interval_sec: float = 5.0 recovery_timeout_sec: float = 60.0 max_restart_attempts: int = 3 backoff_multiplier: float = 2.0 auto_scale_threshold: float = 0.5 class SelfHealingManager: """故障自愈管理器""" def __init__(self, config: SelfHealingConfig, health_checker: Callable[[str], bool], restart_handler: Callable[[str], bool], scale_handler: Callable[[int], bool]): self.config = config self.health_checker = health_checker self.restart_handler = restart_handler self.scale_handler = scale_handler self.nodes: Dict[str, ServiceNode] = {} self._stop_flag = threading.Event() self._healing_thread: Optional[threading.Thread] = None def register_node(self, node_id: str): """注册节点""" self.nodes[node_id] = ServiceNode(node_id=node_id) def start(self): """启动自愈环路""" self._healing_thread = threading.Thread( target=self._healing_loop, daemon=True ) self._healing_thread.start() def _healing_loop(self): """自愈主循环""" while not self._stop_flag.is_set(): for node_id, node in list(self.nodes.items()): if node.status == NodeStatus.ISOLATED: self._attempt_recovery(node_id) else: self._health_check(node_id) # 检查是否需要自动扩容 self._check_auto_scale() time.sleep(self.config.health_check_interval_sec) def _health_check(self, node_id: str): """执行健康检查""" node = self.nodes[node_id] is_healthy = self.health_checker(node_id) node.last_heartbeat = time.time() if not is_healthy: node.failure_count += 1 if node.failure_count >= self.config.max_health_check_failures: self._isolate_node(node_id) else: node.failure_count = 0 node.error_count = 0 def _isolate_node(self, node_id: str): """隔离故障节点""" node = self.nodes[node_id] node.status = NodeStatus.ISOLATED node.health_score = 0.0 print(f"[自愈] 节点 {node_id} 已隔离（连续失败 " f"{node.failure_count} 次）") # 立即尝试恢复 threading.Thread( target=self._attempt_recovery, args=(node_id,) ).start() def _attempt_recovery(self, node_id: str): """尝试恢复节点""" node = self.nodes[node_id] attempt = 0 while attempt < self.config.max_restart_attempts: delay = (self.config.backoff_multiplier ** attempt * self.config.health_check_interval_sec) time.sleep(delay) success = self.restart_handler(node_id) if success: # 等待节点就绪 time.sleep(self.config.health_check_interval_sec * 2) if self.health_checker(node_id): node.status = NodeStatus.HEALTHY node.failure_count = 0 node.health_score = 0.8 node.restart_count += 1 print(f"[自愈] 节点 {node_id} 恢复成功") return attempt += 1 # 无法恢复：触发告警 print(f"[自愈] 节点 {node_id} 恢复失败，需人工介入") node.status = NodeStatus.UNHEALTHY def _check_auto_scale(self): """检查自动扩容""" healthy_count = sum( 1 for n in self.nodes.values() if n.status == NodeStatus.HEALTHY ) total = len(self.nodes) if total == 0: return healthy_ratio = healthy_count / total if healthy_ratio < self.config.auto_scale_threshold: shortage = max(1, total - healthy_count) print(f"[自愈] 触发自动扩容：需要 {shortage} 个新节点") self.scale_handler(shortage) def stop(self): """停止自愈""" self._stop_flag.set() if self._healing_thread: self._healing_thread.join(timeout=10) if __name__ == "__main__": def mock_health_check(node_id: str) -> bool: return random.random() > 0.15 # 85% 健康率 def mock_restart(node_id: str) -> bool: return random.random() > 0.3 # 70% 重启成功率 def mock_scale(count: int) -> bool: print(f"扩容 {count} 个节点") return True config = SelfHealingConfig( max_health_check_failures=3, health_check_interval_sec=2, max_restart_attempts=3, ) manager = SelfHealingManager( config, mock_health_check, mock_restart, mock_scale ) for i in range(5): manager.register_node(f"worker-{i:02d}") manager.start() time.sleep(30) manager.stop() print("\n节点状态汇总:") for nid, node in manager.nodes.items(): print(f" {nid}: {node.status.value} (重启{node.restart_count}次)")