# 滚动升级核心实现
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()

# 混沌工程框架实现
import time
import threading
from typing import Dict, List, Any, Optional, Callable, Set
from datetime import datetime, timedelta
import random
import json

class ChaosExperiment:
    """混沌实验"""
    
    def __init__(self, experiment_id: str, name: str, description: str):
        self.experiment_id = experiment_id
        self.name = name
        self.description = description
        self.status = "pending"  # pending, running, completed, failed
        self.start_time: Optional[datetime] = None
        self.end_time: Optional[datetime] = None
        self.configuration: Dict[str, Any] = {}
        self.targets: List[str] = []
        self.actions: List[Dict[str, Any]] = []
        self.verifications: List[Dict[str, Any]] = []
        self.results: Dict[str, Any] = {}

class ChaosPrinciple:
    """混沌工程原则"""
    
    def __init__(self):
        self.principles = [
            "构建假设：围绕稳态行为建立假设",
            "多样化现实：多样化真实世界的事件",
            "在生产环境中运行实验",
            "自动化实验以推动流程",
            "最小化影响：将负面影响降到最低"
        ]
    
    def get_principles(self) -> List[str]:
        """获取混沌工程原则"""
        return self.principles.copy()
    
    def validate_experiment(self, experiment: ChaosExperiment) -> bool:
        """验证实验是否符合混沌工程原则"""
        violations = []
        
        # 检查是否建立了假设
        if not experiment.configuration.get("hypothesis"):
            violations.append("实验缺少明确的假设")
        
        # 检查是否在生产环境中（模拟检查）
        if not experiment.configuration.get("production_environment", False):
            violations.append("实验未在生产环境中运行")
        
        # 检查是否有最小化影响的措施
        if not experiment.configuration.get("blast_radius_limitation"):
            violations.append("实验缺少影响范围限制")
        
        if violations:
            print(f"实验 {experiment.experiment_id} 违反混沌工程原则:")
            for violation in violations:
                print(f"  - {violation}")
            return False
        
        print(f"实验 {experiment.experiment_id} 符合混沌工程原则")
        return True

class ChaosHypothesis:
    """混沌假设"""
    
    def __init__(self, name: str, description: str):
        self.name = name
        self.description = description
        self.metrics: List[Dict[str, Any]] = []
        self.tolerance: float = 0.95  # 95%的容忍度
    
    def add_metric(self, metric_name: str, expected_value: Any, 
                   tolerance: float = 0.95):
        """添加指标"""
        self.metrics.append({
            "name": metric_name,
            "expected": expected_value,
            "tolerance": tolerance
        })
    
    def validate(self, actual_metrics: Dict[str, Any]) -> Dict[str, Any]:
        """验证假设"""
        results = {
            "passed": True,
            "metric_results": [],
            "overall_score": 1.0
        }
        
        passed_count = 0
        total_count = len(self.metrics)
        
        for metric in self.metrics:
            metric_name = metric["name"]
            expected = metric["expected"]
            tolerance = metric["tolerance"]
            actual = actual_metrics.get(metric_name)
            
            # 简化的验证逻辑
            if actual is not None:
                # 对于数值类型指标
                if isinstance(expected, (int, float)) and isinstance(actual, (int, float)):
                    diff_ratio = abs(actual - expected) / expected if expected != 0 else 0
                    passed = diff_ratio <= (1 - tolerance)
                # 对于布尔类型指标
                elif isinstance(expected, bool) and isinstance(actual, bool):
                    passed = expected == actual
                # 对于其他类型
                else:
                    passed = expected == actual
            else:
                passed = False
            
            metric_result = {
                "metric": metric_name,
                "expected": expected,
                "actual": actual,
                "passed": passed
            }
            results["metric_results"].append(metric_result)
            
            if passed:
                passed_count += 1
        
        results["overall_score"] = passed_count / total_count if total_count > 0 else 1.0
        results["passed"] = results["overall_score"] >= self.tolerance
        
        return results

# 使用示例
def demonstrate_chaos_principles():
    """演示混沌工程原则"""
    # 创建原则验证器
    principle = ChaosPrinciple()
    
    # 显示原则
    print("混沌工程原则:")
    for i, p in enumerate(principle.get_principles(), 1):
        print(f"  {i}. {p}")
    
    # 创建实验
    experiment = ChaosExperiment(
        "exp-001",
        "网络延迟测试",
        "测试网络延迟对存储系统性能的影响"
    )
    
    # 配置实验
    experiment.configuration = {
        "hypothesis": "网络延迟增加50%不会导致系统不可用",
        "production_environment": True,
        "blast_radius_limitation": "仅影响单个数据中心"
    }
    
    # 验证实验
    principle.validate_experiment(experiment)
    
    # 创建假设
    hypothesis = ChaosHypothesis(
        "性能稳定性假设",
        "系统在故障条件下仍能保持95%的正常性能"
    )
    hypothesis.add_metric("read_latency_ms", 50, 0.90)
    hypothesis.add_metric("write_latency_ms", 100, 0.90)
    hypothesis.add_metric("system_availability", True, 0.95)
    
    # 验证假设（模拟实际指标）
    actual_metrics = {
        "read_latency_ms": 55,  # 略高于预期
        "write_latency_ms": 95,  # 略低于预期
        "system_availability": True
    }
    
    validation_result = hypothesis.validate(actual_metrics)
    print(f"\n假设验证结果:")
    print(f"  总体通过: {validation_result['passed']}")
    print(f"  总体得分: {validation_result['overall_score']:.2%}")
    print(f"  指标详情:")
    for result in validation_result["metric_results"]:
        status = "通过" if result["passed"] else "失败"
        print(f"    {result['metric']}: {result['actual']} (期望: {result['expected']}) [{status}]")

# 运行演示
# demonstrate_chaos_principles()

# 版本兼容性设计示例
from typing import Dict, Any, List, Optional
from enum import Enum
import json

class CompatibilityLevel(Enum):
    """兼容性级别"""
    FULL = "full"        # 完全兼容
    BACKWARD = "backward"  # 向后兼容
    FORWARD = "forward"   # 向前兼容
    NONE = "none"         # 不兼容

class Version:
    """版本信息"""
    
    def __init__(self, major: int, minor: int, patch: int, build: str = ""):
        self.major = major
        self.minor = minor
        self.patch = patch
        self.build = build
    
    def __str__(self):
        version_str = f"{self.major}.{self.minor}.{self.patch}"
        if self.build:
            version_str += f"-{self.build}"
        return version_str
    
    def __lt__(self, other):
        if self.major != other.major:
            return self.major < other.major
        if self.minor != other.minor:
            return self.minor < other.minor
        return self.patch < other.patch
    
    def __eq__(self, other):
        return (self.major == other.major and 
                self.minor == other.minor and 
                self.patch == other.patch)

class CompatibilityMatrix:
    """兼容性矩阵"""
    
    def __init__(self):
        self.matrix: Dict[str, Dict[str, CompatibilityLevel]] = {}
    
    def add_compatibility(self, version1: str, version2: str, 
                         level: CompatibilityLevel):
        """添加兼容性信息"""
        if version1 not in self.matrix:
            self.matrix[version1] = {}
        self.matrix[version1][version2] = level
        
        # 确保矩阵对称性（对于双向兼容性）
        if level == CompatibilityLevel.FULL:
            if version2 not in self.matrix:
                self.matrix[version2] = {}
            self.matrix[version2][version1] = level
    
    def check_compatibility(self, version1: str, version2: str) -> CompatibilityLevel:
        """检查两个版本的兼容性"""
        if version1 in self.matrix and version2 in self.matrix[version1]:
            return self.matrix[version1][version2]
        return CompatibilityLevel.NONE

class UpgradePlanner:
    """升级规划器"""
    
    def __init__(self, compatibility_matrix: CompatibilityMatrix):
        self.compatibility_matrix = compatibility_matrix
        self.current_versions: Dict[str, str] = {}  # component -> version
    
    def set_current_version(self, component: str, version: str):
        """设置当前版本"""
        self.current_versions[component] = version
    
    def plan_upgrade_path(self, target_version: str, 
                         components: List[str]) -> List[Dict[str, Any]]:
        """规划升级路径"""
        upgrade_steps = []
        
        for component in components:
            current_version = self.current_versions.get(component, "0.0.0")
            
            # 检查兼容性
            compatibility = self.compatibility_matrix.check_compatibility(
                current_version, target_version)
            
            if compatibility == CompatibilityLevel.NONE:
                # 需要中间版本
                intermediate_steps = self._find_intermediate_versions(
                    current_version, target_version)
                upgrade_steps.extend(intermediate_steps)
            else:
                # 可以直接升级
                upgrade_steps.append({
                    "component": component,
                    "from_version": current_version,
                    "to_version": target_version,
                    "compatibility": compatibility.value,
                    "direct_upgrade": True
                })
        
        return upgrade_steps
    
    def _find_intermediate_versions(self, from_version: str, 
                                  to_version: str) -> List[Dict[str, Any]]:
        """查找中间版本（简化实现）"""
        # 在实际实现中，这里会查询兼容性矩阵找到合适的中间版本
        # 这里简化为直接返回一个示例步骤
        return [{
            "component": "example",
            "from_version": from_version,
            "to_version": to_version,
            "compatibility": "需要中间版本",
            "direct_upgrade": False,
            "notes": "需要先升级到兼容的中间版本"
        }]

# 使用示例
def demonstrate_version_compatibility():
    """演示版本兼容性设计"""
    # 创建兼容性矩阵
    compat_matrix = CompatibilityMatrix()
    
    # 定义版本兼容性
    compat_matrix.add_compatibility("1.0.0", "1.1.0", CompatibilityLevel.FULL)
    compat_matrix.add_compatibility("1.1.0", "1.2.0", CompatibilityLevel.FULL)
    compat_matrix.add_compatibility("1.0.0", "1.2.0", CompatibilityLevel.BACKWARD)
    compat_matrix.add_compatibility("2.0.0", "2.1.0", CompatibilityLevel.FULL)
    
    # 创建升级规划器
    planner = UpgradePlanner(compat_matrix)
    
    # 设置当前版本
    planner.set_current_version("metadata-service", "1.0.0")
    planner.set_current_version("data-service", "1.1.0")
    planner.set_current_version("client-sdk", "1.0.0")
    
    # 规划升级到2.0.0
    upgrade_path = planner.plan_upgrade_path("2.0.0", 
                                           ["metadata-service", "data-service", "client-sdk"])
    
    print("升级路径规划:")
    for step in upgrade_path:
        print(f"  组件 {step['component']}: {step['from_version']} -> {step['to_version']}")
        print(f"    兼容性: {step['compatibility']}")
        if not step['direct_upgrade']:
            print(f"    注意: {step.get('notes', '需要特殊处理')}")

# 运行演示
# demonstrate_version_compatibility()