故障自愈与自动恢复机制从手动重启到无人值守一、凌晨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_factorytime.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_idnode_id) def start(self): 启动自愈环路 self._healing_thread threading.Thread( targetself._healing_loop, daemonTrue ) 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( targetself._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(timeout10) 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_failures3, health_check_interval_sec2, max_restart_attempts3, ) manager SelfHealingManager( config, mock_health_check, mock_restart, mock_scale ) for i in range(5): manager.register_node(fworker-{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}次))四、自愈的边界与陷阱误判导致正常节点被隔离健康检查过于敏感会把偶尔抖动当作永久故障。需要 3 次连续失败 指数退避来避免。自愈风暴所有节点同时故障时自动扩容可能无限创建新节点无底洞。需要设置最大节点数作为安全帽。有状态服务的自愈数据库节点不能简单重启替换涉及数据一致性。对于有状态服务自愈仅限于切主/从库切换。五、总结自愈不是万能药能处理进程崩溃、内存泄漏但不能修复代码 bug。健康检查要审慎连续失败 指数退避比单次失败触发更可靠。自愈层次要分明L1 重启、L2 替换、L3 降级分层处理不同严重程度的故障。一定要设上限最大重启次数、最大节点数、最大恢复时间防止自愈本身成为故障源。本文实现的三层自愈管理器包含健康检查、节点隔离、指数退避恢复和自动扩容核心的 SelfHealingManager 可作为 K8s liveness probe 的上层自愈策略参考。
