6.3-容错与恢复(2)

6.3-容错与恢复(2) — GPU推理优化 高可用性保障

本节导读：继续深入讲解大模型推理服务的高级容错策略和完整的高可用架构实现方案。

步骤2：故障恢复机制

# fault_recovery.py
import time
import subprocess
import logging
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import threading
import psutil
import os
from datetime import datetime

@dataclass
class RecoveryAction:
    """恢复动作"""
    id: str
    name: str
    type: str  # restart, reboot, restore, scale
    component: str
    command: str
    timeout: int
    retry_count: int
    success_criteria: Dict[str, Any]
    status: str  # pending, running, success, failed
    start_time: Optional[float] = None
    end_time: Optional[float] = None
    last_error: Optional[str] = None

class FaultRecovery:
    """故障恢复器"""
    
    def __init__(self):
        self.recovery_actions: List[RecoveryAction] = []
        self.running = True
        self.logger = logging.getLogger(__name__)
        self.init_recovery_templates()
        
    def init_recovery_templates(self):
        templates = [
            {
                'name': 'restart_gpu_service',
                'type': 'restart',
                'component': 'gpu_service',
                'command': 'systemctl restart nvidia-fabricmanager && systemctl restart nvidia-container-runtime',
                'timeout': 60,
                'retry_count': 3,
                'success_criteria': {
                    'gpu_available': True,
                    'service_running': True
                }
            },
            {
                'name': 'restore_model',
                'type': 'restore',
                'component': 'model',
                'command': 'rsync -avz /backup/models/ /models/',
                'timeout': 120,
                'retry_count': 3,
                'success_criteria': {
                    'model_loaded': True,
                    'model_size': 'original'
                }
            }
        ]
        
        for template in templates:
            self.create_recovery_action(template)
            
    def execute_recovery_action(self, action_id: str) -> bool:
        """执行恢复动作"""
        action = next((a for a in self.recovery_actions if a.id == action_id), None)
        if not action:
            return False
            
        action.status = 'running'
        action.start_time = time.time()
        self.logger.info(f"Starting recovery action: {action.name}")
        
        # 执行恢复动作
        retry_count = 0
        last_error = None
        
        while retry_count < action.retry_count:
            try:
                process = subprocess.Popen(
                    action.command,
                    shell=True,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True
                )
                
                stdout, stderr = process.communicate(timeout=action.timeout)
                exit_code = process.returncode
                
                if exit_code == 0:
                    if self.check_success_criteria(action):
                        action.status = 'success'
                        action.end_time = time.time()
                        self.logger.info(f"Recovery action {action.name} completed successfully")
                        return True
                    else:
                        last_error = "Success criteria not met"
                        retry_count += 1
                else:
                    last_error = stderr
                    retry_count += 1
                    
            except subprocess.TimeoutExpired:
                last_error = "Timeout"
                retry_count += 1
            except Exception as e:
                last_error = str(e)
                retry_count += 1
                
        # 所有重试都失败
        action.status = 'failed'
        action.end_time = time.time()
        action.last_error = last_error
        self.logger.error(f"Recovery action {action.name} failed after {retry_count} attempts")
        return False
        
    def check_success_criteria(self, action: RecoveryAction) -> bool:
        """检查成功条件"""
        try:
            criteria = action.success_criteria
            
            if 'gpu_available' in criteria:
                try:
                    import GPUtil
                    gpus = GPUtil.getGPUs()
                    if not gpus:
                        return False
                    for gpu in gpus:
                        if gpu.health != 0:
                            return False
                except Exception:
                    return False
                    
            if 'model_loaded' in criteria:
                try:
                    if not os.path.exists('/models') or not os.listdir('/models'):
                        return False
                except Exception:
                    return False
                    
            return True
            
        except Exception as e:
            self.logger.error(f"Error checking success criteria: {e}")
            return False
            
    def auto_recover(self, fault_type: str, component: str) -> Optional[str]:
        """自动恢复"""
        suitable_actions = [
            action for action in self.recovery_actions
            if action.component == component and action.status in ['pending', 'failed']
        ]
        
        if suitable_actions:
            action = suitable_actions[0]
            success = self.execute_recovery_action(action.id)
            if success:
                return action.id
            else:
                for action in suitable_actions[1:]:
                    if action.status == 'pending':
                        success = self.execute_recovery_action(action.id)
                        if success:
                            return action.id
                            
        return None
        
    def get_recovery_summary(self) -> Dict[str, Any]:
        running_actions = [a for a in self.recovery_actions if a.status == 'running']
        success_actions = [a for a in self.recovery_actions if a.status == 'success']
        failed_actions = [a for a in self.recovery_actions if a.status == 'failed']
        
        return {
            'total_actions': len(self.recovery_actions),
            'running_actions': len(running_actions),
            'success_actions': len(success_actions),
            'failed_actions': len(failed_actions),
            'success_rate': len(success_actions) / len(self.recovery_actions) * 100 if self.recovery_actions else 0,
            'recent_actions': [
                {
                    'id': a.id,
                    'name': a.name,
                    'status': a.status,
                    'component': a.component,
                    'start_time': a.start_time,
                    'end_time': a.end_time,
                    'last_error': a.last_error
                }
                for a in self.recovery_actions[-5:]
            ]
        }

# 使用示例
if __name__ == "__main__":
    recovery = FaultRecovery()
    
    try:
        # 模拟故障并自动恢复
        print("Simulating GPU service fault...")
        recovery_action = recovery.auto_recover('hardware', 'gpu_service')
        print(f"Recovery action: {recovery_action}")
        
        # 输出恢复摘要
        summary = recovery.get_recovery_summary()
        print("Recovery summary:", summary)
        
    except KeyboardInterrupt:
        print("Shutting down recovery system...")
        recovery.running = False
​

步骤3：故障预防机制

# fault_prevention.py
import time
import subprocess
import logging
from dataclasses import dataclass
from typing import List, Dict, Any, Optional
import threading
import psutil
import os
from datetime import datetime

@dataclass
class PreventiveAction:
    """预防动作"""
    id: str
    name: str
    type: str
    component: str
    schedule: str
    command: str
    description: str
    status: str
    success_count: int = 0
    failure_count: int = 0

class FaultPrevention:
    """故障预防系统"""
    
    def __init__(self):
        self.preventive_actions: List[PreventiveAction] = []
        self.running = True
        self.logger = logging.getLogger(__name__)
        self.init_preventive_templates()
        self.scheduler_thread = threading.Thread(target=self.scheduler_loop)
        self.scheduler_thread.daemon = True
        self.scheduler_thread.start()
        
    def init_preventive_templates(self):
        templates = [
            {
                'name': 'gpu_memory_cleanup',
                'type': 'maintenance',
                'component': 'gpu',
                'schedule': '0 */6 * * *',
                'command': 'nvidia-smi -cc --gpu_reset && sync',
                'description': 'GPU内存清理和重置'
            },
            {
                'name': 'model_backup',
                'type': 'backup',
                'component': 'model',
                'schedule': '0 0 * * *',
                'command': 'rsync -avz /models/ /backup/models/',
                'description': '模型数据备份'
            },
            {
                'name': 'service_restart',
                'type': 'maintenance',
                'component': 'inference_service',
                'schedule': '0 */4 * * *',
                'command': 'systemctl restart inference-service',
                'description': '推理服务定时重启'
            }
        ]
        
        for template in templates:
            self.create_preventive_action(template)
            
    def create_preventive_action(self, template: Dict[str, Any]) -> PreventiveAction:
        action_id = f"{template['name']}_{int(time.time())}"
        preventive_action = PreventiveAction(
            id=action_id,
            name=template['name'],
            type=template['type'],
            component=template['component'],
            schedule=template['schedule'],
            command=template['command'],
            description=template['description'],
            status='scheduled'
        )
        
        self.preventive_actions.append(preventive_action)
        return preventive_action
        
    def scheduler_loop(self):
        """调度器循环"""
        while self.running:
            try:
                current_time = time.time()
                
                for action in self.preventive_actions:
                    if (action.schedule and current_time % 3600 < 60 and 
                        action.status in ['scheduled', 'failed']):
                        
                        action.status = 'running'
                        exec_thread = threading.Thread(
                            target=self.execute_preventive_action,
                            args=(action,)
                        )
                        exec_thread.daemon = True
                        exec_thread.start()
                        
                time.sleep(60)
                
            except Exception as e:
                self.logger.error(f"Scheduler error: {e}")
                time.sleep(300)
                
    def execute_preventive_action(self, action: PreventiveAction):
        """执行预防动作"""
        try:
            self.logger.info(f"Starting preventive action: {action.name}")
            
            process = subprocess.Popen(
                action.command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True
            )
            
            stdout, stderr = process.communicate(timeout=300)
            exit_code = process.returncode
            
            if exit_code == 0:
                action.status = 'success'
                action.success_count += 1
                self.logger.info(f"Preventive action {action.name} completed successfully")
            else:
                action.status = 'failed'
                action.failure_count += 1
                self.logger.error(f"Preventive action {action.name} failed: {stderr}")
                
        except subprocess.TimeoutExpired:
            action.status = 'failed'
            action.failure_count += 1
            self.logger.error(f"Preventive action {action.name} timed out")
            
        except Exception as e:
            action.status = 'failed'
            action.failure_count += 1
            self.logger.error(f"Preventive action {action.name} error: {e}")
            
    def get_preventive_summary(self) -> Dict[str, Any]:
        running_actions = [a for a in self.preventive_actions if a.status == 'running']
        success_actions = [a for a in self.preventive_actions if a.status == 'success']
        failed_actions = [a for a in self.preventive_actions if a.status == 'failed']
        
        return {
            'total_actions': len(self.preventive_actions),
            'running_actions': len(running_actions),
            'success_actions': len(success_actions),
            'failed_actions': len(failed_actions),
            'success_rate': len(success_actions) / len(self.preventive_actions) * 100 if self.preventive_actions else 0,
            'actions_by_type': {
                action_type: len([a for a in self.preventive_actions if a.type == action_type])
                for action_type in ['maintenance', 'optimization', 'backup', 'health_check']
            }
        }

# 使用示例
if __name__ == "__main__":
    prevention = FaultPrevention()
    
    try:
        while True:
            time.sleep(60)
            summary = prevention.get_preventive_summary()
            print(f"Prevention running: {summary['running_actions']}, success: {summary['success_actions']}")
            
    except KeyboardInterrupt:
        print("Shutting down prevention system...")
        prevention.running = False
​

高可用架构部署

# docker-compose.yml
version: '3.8'

services:
  fault-detector:
    image: python:3.9-slim
    volumes:
      - ./fault_detector.py:/app/fault_detector.py
      - ./fault_recovery.py:/app/fault_recovery.py
      - ./fault_prevention.py:/app/fault_prevention.py
    command: python fault_detector.py
    restart: unless-stopped

  inference-service:
    image: nvidia/cuda:11.8.0-base
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    volumes:
      - ./models:/models
      - ./config:/config
    restart: unless-stopped

  load-balancer:
    image: nginx:alpine
    ports:
      - "80:80"
    depends_on:
      - inference-service
    restart: unless-stopped
​

完整系统监控

# monitoring_system.py
import time
import threading
from fault_detector import FaultDetector
from fault_recovery import FaultRecovery
from fault_prevention import FaultPrevention

class MonitoringSystem:
    """监控系统"""
    
    def __init__(self):
        self.fault_detector = FaultDetector()
        self.fault_recovery = FaultRecovery()
        self.fault_prevention = FaultPrevention()
        self.running = True
        
    def monitor_loop(self):
        """监控循环"""
        while self.running:
            try:
                # 获取各组件状态
                fault_summary = self.fault_detector.get_fault_summary()
                recovery_summary = self.fault_recovery.get_recovery_summary()
                prevention_summary = self.fault_prevention.get_preventive_summary()
                
                # 输出状态
                print(f"=== 系统状态监控 ===")
                print(f"活跃故障数: {fault_summary['active_faults']}")
                print(f"恢复成功率: {recovery_summary['success_rate']:.1f}%")
                print(f"预防成功率: {prevention_summary['success_rate']:.1f}%")
                
                # 自动恢复
                if fault_summary['active_faults'] > 0:
                    for fault_type in ['hardware', 'software']:
                        recovery_action = self.fault_recovery.auto_recover(fault_type, 'gpu_service')
                        if recovery_action:
                            print(f"自动恢复动作执行: {recovery_action}")
                
                time.sleep(60)
                
            except Exception as e:
                print(f"监控错误: {e}")
                
    def start(self):
        """启动监控"""
        monitor_thread = threading.Thread(target=self.monitor_loop)
        monitor_thread.daemon = True
        monitor_thread.start()
        
    def stop(self):
        """停止监控"""
        self.running = False
        self.fault_detector.running = False
        self.fault_recovery.running = False
        self.fault_prevention.running = False

# 使用示例
if __name__ == "__main__":
    system = MonitoringSystem()
    system.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        system.stop()
        print("监控系统已停止")
​

常见问题 FAQ

Q1：GPU过热故障如何处理？

A：GPU过热处理方案：

1. 检查散热风扇运转状态
2. 清理GPU散热器灰尘
3. 降低GPU功耗限制
4. 改善机房环境温度
5. 使用温度监控工具定期检查

Q2：推理服务崩溃后如何快速恢复？

A：服务崩溃恢复策略：

1. 配置自动重启脚本
2. 设置容器自动重启
3. 建立健康检查机制
4. 实现多实例部署
5. 建立服务降级机制

Q3：如何预防内存泄漏问题？

A：内存泄漏预防措施：

1. 定期清理GPU内存
2. 监控内存使用趋势
3. 设置内存限制阈值
4. 实现内存泄漏检测
5. 定期重启服务

本节小结

通过本节的学习，读者掌握了完整的大模型推理服务高可用架构设计。从故障检测到故障恢复，再到故障预防，构建了完整的多层次容错体系，大幅提升了推理服务的稳定性和可靠性。

关键词：容错与恢复, 高可用性, 故障检测, 故障恢复, 故障预防, GPU推理优化, 实战教程
难度：进阶
预计阅读：30分钟