个人博客

# 滚动升级核心实现
import time
import threading
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime, timedelta
import random

class ServiceInstance:
    """服务实例"""
    
    def __init__(self, instance_id: str, service_name: str, version: str):
        self.instance_id = instance_id
        self.service_name = service_name
        self.version = version
        self.status = "running"  # running, upgrading, stopped, error
        self.health = "healthy"  # healthy, warning, error
        self.last_heartbeat = datetime.now()
        self.startup_time = datetime.now()

class RollingUpgradeOrchestrator:
    """滚动升级编排器"""
    
    def __init__(self, health_check_interval: int = 30, 
                 batch_interval: int = 10):
        self.instances: Dict[str, ServiceInstance] = {}
        self.upgrade_queue: List[str] = []
        self.upgrading = False
        self.health_check_interval = health_check_interval
        self.batch_interval = batch_interval
        self.health_check_callback: Optional[Callable[[ServiceInstance], bool]] = None
        self.pre_upgrade_hook: Optional[Callable[[ServiceInstance], bool]] = None
        self.post_upgrade_hook: Optional[Callable[[ServiceInstance], bool]] = None
        self.on_upgrade_complete: Optional[Callable[[], None]] = None
        self.upgrade_log: List[Dict[str, Any]] = []
    
    def add_instance(self, instance: ServiceInstance):
        """添加服务实例"""
        self.instances[instance.instance_id] = instance
    
    def set_health_check_callback(self, callback: Callable[[ServiceInstance], bool]):
        """设置健康检查回调"""
        self.health_check_callback = callback
    
    def set_pre_upgrade_hook(self, callback: Callable[[ServiceInstance], bool]):
        """设置升级前钩子"""
        self.pre_upgrade_hook = callback
    
    def set_post_upgrade_hook(self, callback: Callable[[ServiceInstance], bool]):
        """设置升级后钩子"""
        self.post_upgrade_hook = callback
    
    def set_upgrade_complete_callback(self, callback: Callable[[], None]):
        """设置升级完成回调"""
        self.on_upgrade_complete = callback
    
    def start_rolling_upgrade(self, target_version: str, 
                           batch_size: int = 1, 
                           health_check_timeout: int = 300,
                           rollback_on_failure: bool = True) -> bool:
        """开始滚动升级"""
        if self.upgrading:
            print("升级已在进行中")
            return False
        
        # 记录升级开始
        self._log_upgrade_event("start", {
            "target_version": target_version,
            "batch_size": batch_size,
            "instance_count": len(self.instances)
        })
        
        self.upgrading = True
        self.upgrade_queue = list(self.instances.keys())
        
        print(f"开始滚动升级到版本 {target_version}")
        print(f"总实例数: {len(self.upgrade_queue)}, 批量大小: {batch_size}")
        
        # 分批升级
        batch_number = 1
        while self.upgrade_queue and self.upgrading:
            batch = self.upgrade_queue[:batch_size]
            self.upgrade_queue = self.upgrade_queue[batch_size:]
            
            print(f"升级批次 {batch_number}: {[inst_id for inst_id in batch]}")
            
            # 升级批次中的实例
            if not self._upgrade_batch(batch, target_version, health_check_timeout, rollback_on_failure):
                print("批次升级失败")
                if rollback_on_failure:
                    print("执行回滚操作...")
                    self._rollback_failed_upgrade(target_version)
                self.upgrading = False
                self._log_upgrade_event("failure", {
                    "batch_number": batch_number,
                    "failed_instances": batch
                })
                return False
            
            # 记录批次完成
            self._log_upgrade_event("batch_complete", {
                "batch_number": batch_number,
                "instances": batch
            })
            
            # 批次间等待
            if self.upgrade_queue:
                print(f"等待 {self.batch_interval} 秒后开始下一批次...")
                time.sleep(self.batch_interval)
            
            batch_number += 1
        
        self.upgrading = False
        print("滚动升级完成")
        
        # 记录升级完成
        self._log_upgrade_event("complete", {
            "target_version": target_version
        })
        
        if self.on_upgrade_complete:
            self.on_upgrade_complete()
        
        return True
    
    def _upgrade_batch(self, batch: List[str], target_version: str, 
                      health_check_timeout: int, rollback_on_failure: bool) -> bool:
        """升级一个批次"""
        # 1. 执行升级前钩子
        for instance_id in batch:
            if instance_id in self.instances:
                instance = self.instances[instance_id]
                if self.pre_upgrade_hook:
                    try:
                        if not self.pre_upgrade_hook(instance):
                            print(f"实例 {instance_id} 升级前检查失败")
                            return False
                    except Exception as e:
                        print(f"执行实例 {instance_id} 升级前钩子时出错: {e}")
                        return False
        
        # 2. 停止实例
        for instance_id in batch:
            if instance_id in self.instances:
                instance = self.instances[instance_id]
                print(f"停止实例 {instance_id}")
                if not self._stop_instance(instance):
                    print(f"停止实例 {instance_id} 失败")
                    return False
        
        # 3. 升级实例
        for instance_id in batch:
            if instance_id in self.instances:
                instance = self.instances[instance_id]
                print(f"升级实例 {instance_id} 到版本 {target_version}")
                if not self._upgrade_instance(instance, target_version):
                    print(f"升级实例 {instance_id} 失败")
                    return False
        
        # 4. 启动实例
        for instance_id in batch:
            if instance_id in self.instances:
                instance = self.instances[instance_id]
                print(f"启动实例 {instance_id}")
                if not self._start_instance(instance):
                    print(f"启动实例 {instance_id} 失败")
                    return False
        
        # 5. 健康检查
        if not self._health_check_batch(batch, health_check_timeout):
            print(f"批次 {[inst_id for inst_id in batch]} 健康检查失败")
            return False
        
        # 6. 执行升级后钩子
        for instance_id in batch:
            if instance_id in self.instances:
                instance = self.instances[instance_id]
                if self.post_upgrade_hook:
                    try:
                        if not self.post_upgrade_hook(instance):
                            print(f"实例 {instance_id} 升级后检查失败")
                            return False
                    except Exception as e:
                        print(f"执行实例 {instance_id} 升级后钩子时出错: {e}")
                        return False
        
        print(f"批次 {[inst_id for inst_id in batch]} 升级成功")
        return True
    
    def _stop_instance(self, instance: ServiceInstance) -> bool:
        """停止实例"""
        # 模拟停止过程
        instance.status = "stopped"
        time.sleep(random.uniform(1, 3))  # 模拟停止时间
        return True
    
    def _upgrade_instance(self, instance: ServiceInstance, target_version: str) -> bool:
        """升级实例"""
        # 模拟升级过程
        instance.status = "upgrading"
        time.sleep(random.uniform(2, 5))  # 模拟升级时间
        instance.version = target_version
        return True
    
    def _start_instance(self, instance: ServiceInstance) -> bool:
        """启动实例"""
        # 模拟启动过程
        instance.status = "running"
        instance.startup_time = datetime.now()
        time.sleep(random.uniform(1, 3))  # 模拟启动时间
        return True
    
    def _health_check_batch(self, batch: List[str], timeout: int) -> bool:
        """批量健康检查"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            all_healthy = True
            for instance_id in batch:
                if instance_id in self.instances:
                    instance = self.instances[instance_id]
                    if self.health_check_callback:
                        try:
                            is_healthy = self.health_check_callback(instance)
                            instance.health = "healthy" if is_healthy else "error"
                            if not is_healthy:
                                all_healthy = False
                        except Exception as e:
                            print(f"健康检查实例 {instance_id} 时出错: {e}")
                            instance.health = "error"
                            all_healthy = False
                    instance.last_heartbeat = datetime.now()
            
            if all_healthy:
                print(f"批次 {[inst_id for inst_id in batch]} 健康检查通过")
                return True
            
            print("健康检查未通过，等待重试...")
            time.sleep(self.health_check_interval)
        
        print(f"批次 {[inst_id for inst_id in batch]} 健康检查超时")
        return False
    
    def _rollback_failed_upgrade(self, target_version: str):
        """回滚失败的升级"""
        print("执行升级回滚...")
        # 在实际实现中，这里会执行回滚逻辑
        # 简化实现，只打印信息
        self._log_upgrade_event("rollback", {
            "target_version": target_version
        })
    
    def _log_upgrade_event(self, event_type: str, details: Dict[str, Any]):
        """记录升级事件"""
        log_entry = {
            "timestamp": datetime.now(),
            "event_type": event_type,
            "details": details
        }
        self.upgrade_log.append(log_entry)
    
    def get_upgrade_status(self) -> Dict[str, Any]:
        """获取升级状态"""
        total_instances = len(self.instances)
        upgraded_instances = sum(1 for inst in self.instances.values() 
                               if inst.status == "running")
        
        return {
            "upgrading": self.upgrading,
            "total_instances": total_instances,
            "upgraded_instances": upgraded_instances,
            "progress": upgraded_instances / total_instances if total_instances > 0 else 0,
            "remaining_instances": len(self.upgrade_queue),
            "upgrade_log": self.upgrade_log[-10:]  # 最近10条日志
        }

# 健康检查模拟函数
def simulate_health_check(instance: ServiceInstance) -> bool:
    """模拟健康检查"""
    # 模拟95%的成功率
    return random.random() < 0.95

# 升级钩子函数
def pre_upgrade_check(instance: ServiceInstance) -> bool:
    """升级前检查"""
    print(f"执行实例 {instance.instance_id} 升级前检查")
    # 模拟检查过程
    time.sleep(0.5)
    # 模拟99%的成功率
    return random.random() < 0.99

def post_upgrade_check(instance: ServiceInstance) -> bool:
    """升级后检查"""
    print(f"执行实例 {instance.instance_id} 升级后检查")
    # 模拟检查过程
    time.sleep(0.5)
    # 模拟98%的成功率
    return random.random() < 0.98

# 升级完成回调
def on_upgrade_complete():
    """升级完成回调"""
    print("所有实例升级完成，系统已更新到新版本")

# 使用示例
def demonstrate_rolling_upgrade():
    """演示滚动升级"""
    # 创建升级编排器
    orchestrator = RollingUpgradeOrchestrator(health_check_interval=10, batch_interval=5)
    
    # 设置回调函数
    orchestrator.set_health_check_callback(simulate_health_check)
    orchestrator.set_pre_upgrade_hook(pre_upgrade_check)
    orchestrator.set_post_upgrade_hook(post_upgrade_check)
    orchestrator.set_upgrade_complete_callback(on_upgrade_complete)
    
    # 添加服务实例
    instances = [
        ServiceInstance("meta-001", "metadata-service", "1.0.0"),
        ServiceInstance("meta-002", "metadata-service", "1.0.0"),
        ServiceInstance("meta-003", "metadata-service", "1.0.0"),
        ServiceInstance("data-001", "data-service", "1.0.0"),
        ServiceInstance("data-002", "data-service", "1.0.0"),
        ServiceInstance("data-003", "data-service", "1.0.0"),
    ]
    
    for instance in instances:
        orchestrator.add_instance(instance)
    
    # 开始滚动升级
    orchestrator.start_rolling_upgrade("2.0.0", batch_size=2, health_check_timeout=120)
    
    # 显示升级状态
    status = orchestrator.get_upgrade_status()
    print(f"升级状态: {status}")

# 运行演示
# demonstrate_rolling_upgrade()

# 扩容流程实现
import time
import threading
from typing import List, Dict, Any, Optional, Callable
from datetime import datetime, timedelta
import random

class ClusterNode:
    """集群节点"""
    
    def __init__(self, node_id: str, node_type: str, capacity_gb: int):
        self.node_id = node_id
        self.node_type = node_type  # metadata, data, mixed
        self.capacity_gb = capacity_gb
        self.used_gb = 0
        self.status = "active"  # active, joining, leaving, inactive
        self.last_heartbeat = datetime.now()
        self.data_replicas = {}  # data_id -> replica_info

class DataPlacementManager:
    """数据放置管理器"""
    
    def __init__(self):
        self.nodes: Dict[str, ClusterNode] = {}
        self.data_objects: Dict[str, Dict[str, Any]] = {}  # data_id -> info
        self.placement_policy = "consistent_hashing"  # or "replication", "erasure_coding"
    
    def add_node(self, node: ClusterNode):
        """添加节点"""
        self.nodes[node.node_id] = node
        print(f"节点 {node.node_id} 已添加到集群")
    
    def remove_node(self, node_id: str) -> bool:
        """移除节点"""
        if node_id not in self.nodes:
            print(f"节点 {node_id} 不存在")
            return False
        
        node = self.nodes[node_id]
        if node.status != "active":
            print(f"节点 {node_id} 状态不正确: {node.status}")
            return False
        
        # 标记节点为离开状态
        node.status = "leaving"
        print(f"节点 {node_id} 标记为离开状态")
        
        # 迁移数据
        self._migrate_data_from_node(node_id)
        
        # 从集群中移除节点
        del self.nodes[node_id]
        print(f"节点 {node_id} 已从集群中移除")
        return True
    
    def scale_up(self, new_nodes: List[ClusterNode], 
                rebalance_timeout: int = 300) -> bool:
        """扩容集群"""
        print(f"开始扩容，新增 {len(new_nodes)} 个节点")
        
        # 1. 添加新节点
        for node in new_nodes:
            self.add_node(node)
            node.status = "joining"
        
        # 2. 等待节点准备就绪
        time.sleep(5)  # 模拟节点启动时间
        
        # 3. 将新节点标记为活动状态
        for node in new_nodes:
            node.status = "active"
        
        # 4. 重新平衡数据
        print("开始数据重新平衡...")
        self._rebalance_cluster(rebalance_timeout)
        
        print("集群扩容完成")
        return True
    
    def scale_down(self, node_ids: List[str], 
                  rebalance_timeout: int = 300) -> bool:
        """缩容集群"""
        print(f"开始缩容，移除 {len(node_ids)} 个节点: {node_ids}")
        
        # 1. 验证节点
        for node_id in node_ids:
            if node_id not in self.nodes:
                print(f"节点 {node_id} 不存在")
                return False
        
        # 2. 逐个移除节点
        for node_id in node_ids:
            self.remove_node(node_id)
        
        # 3. 重新平衡数据
        print("开始数据重新平衡...")
        self._rebalance_cluster(rebalance_timeout)
        
        print("集群缩容完成")
        return True
    
    def _migrate_data_from_node(self, node_id: str):
        """从节点迁移数据"""
        if node_id not in self.nodes:
            return
        
        node = self.nodes[node_id]
        print(f"开始从节点 {node_id} 迁移数据，共 {len(node.data_replicas)} 个副本")
        
        # 迁移每个数据副本
        migrated_count = 0
        for data_id, replica_info in list(node.data_replicas.items()):
            # 寻找目标节点
            target_node = self._find_target_node_for_data(data_id, exclude_node=node_id)
            if target_node:
                # 迁移数据
                self._migrate_data_replica(data_id, node_id, target_node.node_id)
                # 更新元数据
                del node.data_replicas[data_id]
                migrated_count += 1
            else:
                print(f"无法为数据 {data_id} 找到迁移目标节点")
        
        print(f"从节点 {node_id} 成功迁移 {migrated_count} 个数据副本")
    
    def _find_target_node_for_data(self, data_id: str, exclude_node: str) -> Optional[ClusterNode]:
        """为数据寻找目标节点"""
        # 简化实现：选择使用率最低的活动节点
        active_nodes = [node for node in self.nodes.values() 
                       if node.status == "active" and node.node_id != exclude_node]
        
        if not active_nodes:
            return None
        
        # 选择使用率最低的节点
        target_node = min(active_nodes, key=lambda n: n.used_gb / n.capacity_gb)
        
        # 检查容量是否足够
        data_size = self.data_objects.get(data_id, {}).get("size_gb", 1)
        if target_node.used_gb + data_size <= target_node.capacity_gb:
            return target_node
        
        return None
    
    def _migrate_data_replica(self, data_id: str, source_node_id: str, target_node_id: str):
        """迁移数据副本"""
        print(f"迁移数据 {data_id} 从节点 {source_node_id} 到节点 {target_node_id}")
        
        # 模拟数据迁移过程
        time.sleep(0.1)  # 模拟迁移时间
        
        # 更新目标节点的元数据
        if target_node_id in self.nodes:
            target_node = self.nodes[target_node_id]
            target_node.data_replicas[data_id] = {
                "source": source_node_id,
                "migrated_at": datetime.now()
            }
        
        # 更新数据对象的元数据
        if data_id in self.data_objects:
            if "replicas" not in self.data_objects[data_id]:
                self.data_objects[data_id]["replicas"] = []
            self.data_objects[data_id]["replicas"].append(target_node_id)
    
    def _rebalance_cluster(self, timeout: int):
        """重新平衡集群"""
        start_time = time.time()
        
        # 计算集群使用率
        total_capacity = sum(node.capacity_gb for node in self.nodes.values() 
                           if node.status == "active")
        total_used = sum(node.used_gb for node in self.nodes.values() 
                        if node.status == "active")
        
        if total_capacity == 0:
            print("集群容量为0，无法重新平衡")
            return
        
        target_utilization = total_used / total_capacity
        print(f"目标使用率: {target_utilization:.2%}")
        
        # 简化实现：这里只打印信息，实际实现会进行数据迁移
        print("执行数据重新平衡...")
        time.sleep(2)  # 模拟重新平衡时间
        
        print("集群重新平衡完成")
    
    def get_cluster_status(self) -> Dict[str, Any]:
        """获取集群状态"""
        active_nodes = [node for node in self.nodes.values() if node.status == "active"]
        
        total_capacity = sum(node.capacity_gb for node in active_nodes)
        total_used = sum(node.used_gb for node in active_nodes)
        
        return {
            "total_nodes": len(self.nodes),
            "active_nodes": len(active_nodes),
            "total_capacity_gb": total_capacity,
            "total_used_gb": total_used,
            "utilization": total_used / total_capacity if total_capacity > 0 else 0,
            "nodes": [
                {
                    "node_id": node.node_id,
                    "status": node.status,
                    "capacity_gb": node.capacity_gb,
                    "used_gb": node.used_gb,
                    "utilization": node.used_gb / node.capacity_gb if node.capacity_gb > 0 else 0
                }
                for node in self.nodes.values()
            ]
        }

class AutoScaler:
    """自动扩缩容器"""
    
    def __init__(self, placement_manager: DataPlacementManager):
        self.placement_manager = placement_manager
        self.scaling_policies: Dict[str, Dict[str, Any]] = {}
        self.monitoring = False
        self.monitor_thread: Optional[threading.Thread] = None
    
    def set_scaling_policy(self, policy_name: str, policy_config: Dict[str, Any]):
        """设置扩缩容策略"""
        self.scaling_policies[policy_name] = policy_config
        print(f"设置扩缩容策略: {policy_name}")
    
    def start_monitoring(self):
        """开始监控"""
        if self.monitoring:
            return
        
        self.monitoring = True
        self.monitor_thread = threading.Thread(target=self._monitor_cluster)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
        print("自动扩缩容监控已启动")
    
    def stop_monitoring(self):
        """停止监控"""
        self.monitoring = False
        if self.monitor_thread:
            self.monitor_thread.join(timeout=5)
        print("自动扩缩容监控已停止")
    
    def _monitor_cluster(self):
        """监控集群"""
        while self.monitoring:
            try:
                # 检查是否需要扩缩容
                self._check_scaling_needed()
                time.sleep(30)  # 每30秒检查一次
            except Exception as e:
                print(f"监控集群时出错: {e}")
    
    def _check_scaling_needed(self):
        """检查是否需要扩缩容"""
        cluster_status = self.placement_manager.get_cluster_status()
        
        # 检查扩容策略
        if "scale_up" in self.scaling_policies:
            scale_up_policy = self.scaling_policies["scale_up"]
            threshold = scale_up_policy.get("utilization_threshold", 0.8)
            
            if cluster_status["utilization"] > threshold:
                print(f"集群使用率 {cluster_status['utilization']:.2%} 超过扩容阈值 {threshold:.2%}")
                self._trigger_scale_up(scale_up_policy)
        
        # 检查缩容策略
        if "scale_down" in self.scaling_policies:
            scale_down_policy = self.scaling_policies["scale_down"]
            threshold = scale_down_policy.get("utilization_threshold", 0.3)
            
            if cluster_status["utilization"] < threshold:
                print(f"集群使用率 {cluster_status['utilization']:.2%} 低于缩容阈值 {threshold:.2%}")
                self._trigger_scale_down(scale_down_policy)
    
    def _trigger_scale_up(self, policy: Dict[str, Any]):
        """触发扩容"""
        # 简化实现，实际应根据策略创建新节点
        print("触发扩容操作")
        # 这里应该调用实际的节点创建和添加逻辑
    
    def _trigger_scale_down(self, policy: Dict[str, Any]):
        """触发缩容"""
        # 简化实现，实际应根据策略选择要移除的节点
        print("触发缩容操作")
        # 这里应该调用实际的节点移除逻辑

# 使用示例
def demonstrate_cluster_scaling():
    """演示集群扩缩容"""
    # 创建数据放置管理器
    placement_manager = DataPlacementManager()
    
    # 添加初始节点
    initial_nodes = [
        ClusterNode("node-001", "data", 1000),
        ClusterNode("node-002", "data", 1000),
        ClusterNode("node-003", "data", 1000),
    ]
    
    for node in initial_nodes:
        node.used_gb = 600  # 模拟已使用60%的容量
        placement_manager.add_node(node)
    
    # 显示初始状态
    status = placement_manager.get_cluster_status()
    print("初始集群状态:")
    print(f"  活动节点数: {status['active_nodes']}")
    print(f"  总容量: {status['total_capacity_gb']} GB")
    print(f"  已使用: {status['total_used_gb']} GB")
    print(f"  使用率: {status['utilization']:.2%}")
    
    # 扩容集群
    new_nodes = [
        ClusterNode("node-004", "data", 1000),
        ClusterNode("node-005", "data", 1000),
    ]
    
    placement_manager.scale_up(new_nodes, rebalance_timeout=120)
    
    # 显示扩容后状态
    status = placement_manager.get_cluster_status()
    print("\n扩容后集群状态:")
    print(f"  活动节点数: {status['active_nodes']}")
    print(f"  总容量: {status['total_capacity_gb']} GB")
    print(f"  已使用: {status['total_used_gb']} GB")
    print(f"  使用率: {status['utilization']:.2%}")
    
    # 缩容集群
    placement_manager.scale_down(["node-005"], rebalance_timeout=120)
    
    # 显示缩容后状态
    status = placement_manager.get_cluster_status()
    print("\n缩容后集群状态:")
    print(f"  活动节点数: {status['active_nodes']}")
    print(f"  总容量: {status['total_capacity_gb']} GB")
    print(f"  已使用: {status['total_used_gb']} GB")
    print(f"  使用率: {status['utilization']:.2%}")

# 运行演示
# demonstrate_cluster_scaling()

# 磁盘故障检测实现
import time
import threading
from typing import Dict, List, Any, Optional, Callable
from datetime import datetime, timedelta
import hashlib
import random

class DiskHealthChecker:
    """磁盘健康检查器"""
    
    def __init__(self, check_interval: int = 30):
        self.check_interval = check_interval
        self.disks: Dict[str, Dict[str, Any]] = {}
        self.health_callbacks: List[Callable[[str, str], None]] = []
        self.check_thread: Optional[threading.Thread] = None
        self.checking = False
    
    def add_disk(self, disk_id: str, path: str, capacity_gb: int):
        """添加磁盘"""
        self.disks[disk_id] = {
            "path": path,
            "capacity_gb": capacity_gb,
            "status": "healthy",
            "last_check": None,
            "error_count": 0,
            "last_error": None
        }
        print(f"磁盘 {disk_id} 已添加: {path}")
    
    def add_health_callback(self, callback: Callable[[str, str], None]):
        """添加健康状态回调"""
        self.health_callbacks.append(callback)
    
    def start_health_check(self):
        """开始健康检查"""
        if self.checking:
            return
        
        self.checking = True
        self.check_thread = threading.Thread(target=self._health_check_loop)
        self.check_thread.daemon = True
        self.check_thread.start()
        print("磁盘健康检查已启动")
    
    def stop_health_check(self):
        """停止健康检查"""
        self.checking = False
        if self.check_thread:
            self.check_thread.join(timeout=5)
        print("磁盘健康检查已停止")
    
    def _health_check_loop(self):
        """健康检查循环"""
        while self.checking:
            try:
                self._perform_health_check()
                time.sleep(self.check_interval)
            except Exception as e:
                print(f"健康检查时出错: {e}")
    
    def _perform_health_check(self):
        """执行健康检查"""
        for disk_id, disk_info in self.disks.items():
            old_status = disk_info["status"]
            new_status = self._check_disk_health(disk_id, disk_info)
            
            # 更新状态
            disk_info["status"] = new_status
            disk_info["last_check"] = datetime.now()
            
            # 如果状态发生变化，调用回调函数
            if old_status != new_status:
                print(f"磁盘 {disk_id} 状态变化: {old_status} -> {new_status}")
                for callback in self.health_callbacks:
                    try:
                        callback(disk_id, new_status)
                    except Exception as e:
                        print(f"调用健康回调时出错: {e}")
    
    def _check_disk_health(self, disk_id: str, disk_info: Dict[str, Any]) -> str:
        """检查磁盘健康状态"""
        # 模拟磁盘健康检查
        # 实际实现中可能包括：
        # 1. SMART状态检查
        # 2. 读写测试
        # 3. 坏块检测
        # 4. 温度监控等
        
        # 模拟有一定概率出现故障
        if random.random() < 0.01:  # 1%概率出现故障
            disk_info["error_count"] += 1
            disk_info["last_error"] = datetime.now()
            return "failed"
        
        # 检查错误计数
        if disk_info["error_count"] > 5:
            return "degraded"
        
        return "healthy"
    
    def get_disk_status(self, disk_id: str) -> Dict[str, Any]:
        """获取磁盘状态"""
        if disk_id not in self.disks:
            return {"error": "磁盘不存在"}
        return self.disks[disk_id].copy()
    
    def get_all_disk_status(self) -> Dict[str, Dict[str, Any]]:
        """获取所有磁盘状态"""
        return {disk_id: info.copy() for disk_id, info in self.disks.items()}

class DiskFailureHandler:
    """磁盘故障处理器"""
    
    def __init__(self, health_checker: DiskHealthChecker):
        self.health_checker = health_checker
        self.failure_callbacks: List[Callable[[str, Dict[str, Any]], None]] = []
        
        # 注册健康状态回调
        self.health_checker.add_health_callback(self._on_disk_health_change)
    
    def add_failure_callback(self, callback: Callable[[str, Dict[str, Any]], None]):
        """添加故障回调"""
        self.failure_callbacks.append(callback)
    
    def _on_disk_health_change(self, disk_id: str, status: str):
        """磁盘健康状态变化处理"""
        if status == "failed":
            self._handle_disk_failure(disk_id)
        elif status == "degraded":
            self._handle_disk_degradation(disk_id)
    
    def _handle_disk_failure(self, disk_id: str):
        """处理磁盘故障"""
        disk_info = self.health_checker.get_disk_status(disk_id)
        if "error" in disk_info:
            return
        
        print(f"处理磁盘 {disk_id} 故障")
        
        # 执行故障处理流程
        failure_info = {
            "disk_id": disk_id,
            "failure_time": datetime.now(),
            "error_count": disk_info["error_count"],
            "last_error": disk_info["last_error"],
            "recovery_actions": []
        }
        
        # 1. 标记磁盘为不可用
        failure_info["recovery_actions"].append("标记磁盘为不可用")
        print(f"  1. 标记磁盘 {disk_id} 为不可用")
        
        # 2. 触发数据重建
        failure_info["recovery_actions"].append("触发数据重建流程")
        print(f"  2. 触发数据重建流程")
        self._trigger_data_rebuild(disk_id)
        
        # 3. 发送告警
        failure_info["recovery_actions"].append("发送故障告警")
        print(f"  3. 发送故障告警")
        self._send_failure_alert(disk_id, failure_info)
        
        # 4. 记录故障日志
        failure_info["recovery_actions"].append("记录故障日志")
        print(f"  4. 记录故障日志")
        
        # 调用故障回调
        for callback in self.failure_callbacks:
            try:
                callback(disk_id, failure_info)
            except Exception as e:
                print(f"调用故障回调时出错: {e}")
    
    def _handle_disk_degradation(self, disk_id: str):
        """处理磁盘降级"""
        print(f"磁盘 {disk_id} 状态降级，建议更换")
        # 发送降级告警
        self._send_degradation_alert(disk_id)
    
    def _trigger_data_rebuild(self, disk_id: str):
        """触发数据重建"""
        # 在实际实现中，这里会触发数据重建流程
        # 包括从其他副本或纠删码中恢复数据
        print(f"    触发磁盘 {disk_id} 上数据的重建流程")
        time.sleep(0.1)  # 模拟重建时间
    
    def _send_failure_alert(self, disk_id: str, failure_info: Dict[str, Any]):
        """发送故障告警"""
        print(f"    发送磁盘 {disk_id} 故障告警")
        # 实际实现中会发送告警到监控系统
    
    def _send_degradation_alert(self, disk_id: str):
        """发送降级告警"""
        print(f"    发送磁盘 {disk_id} 降级告警")

# 使用示例
def on_disk_failure(disk_id: str, failure_info: Dict[str, Any]):
    """磁盘故障回调"""
    print(f"收到磁盘 {disk_id} 故障通知")
    print(f"  故障时间: {failure_info['failure_time']}")
    print(f"  恢复动作: {failure_info['recovery_actions']}")

def demonstrate_disk_failure_handling():
    """演示磁盘故障处理"""
    # 创建健康检查器
    health_checker = DiskHealthChecker(check_interval=10)
    
    # 添加磁盘
    disks = [
        ("disk-001", "/data/disk1", 1000),
        ("disk-002", "/data/disk2", 1000),
        ("disk-003", "/data/disk3", 1000),
    ]
    
    for disk_id, path, capacity in disks:
        health_checker.add_disk(disk_id, path, capacity)
    
    # 创建故障处理器
    failure_handler = DiskFailureHandler(health_checker)
    failure_handler.add_failure_callback(on_disk_failure)
    
    # 启动健康检查
    health_checker.start_health_check()
    
    # 运行一段时间观察
    print("开始磁盘健康监控...")
    time.sleep(60)
    
    # 停止健康检查
    health_checker.stop_health_check()
    
    # 显示最终状态
    print("\n最终磁盘状态:")
    all_status = health_checker.get_all_disk_status()
    for disk_id, status in all_status.items():
        print(f"  {disk_id}: {status['status']} (错误计数: {status['error_count']})")

# 运行演示
# demonstrate_disk_failure_handling()