CUDA编程模型：CUDA内存模型

"""On-call轮值调度系统 功能：轮值表生成、排班算法、告警升级逻辑""" import jsonfrom datetime import datetime, timedelta, datefrom typing import List, Dict, Optional, Tuplefrom enum import Enumfrom collections import defaultdict  class AlertLevel(Enum):    """告警级别定义"""    P0 = {"level": 0, "name": "紧急", "response_minutes": 5, "channels": ["phone", "sms", "slack"]}    P1 = {"level": 1, "name": "高优先级", "response_minutes": 30, "channels": ["phone", "slack"]}    P2 = {"level": 2, "name": "中优先级", "response_minutes": 240, "channels": ["slack", "email"]}    P3 = {"level": 3, "name": "低优先级", "response_minutes": 1440, "channels": ["email"]}  class Engineer:    """值班工程师"""    def __init__(self, name: str, tier: int, skills: List[str], timezone: str = "Asia/Shanghai"):        self.name = name        self.tier = tier        self.skills = skills        self.timezone = timezone        self.shift_count = 0        self.last_shift_end: Optional[date] = None     def __repr__(self):        return f"Engineer({self.name}, Tier={self.tier})"  class EscalationPolicy:    """告警升级策略"""    def __init__(self):        self.policies = {            AlertLevel.P0: {"escalation_minutes": 5, "max_escalations": 3},            AlertLevel.P1: {"escalation_minutes": 15, "max_escalations": 2},            AlertLevel.P2: {"escalation_minutes": 60, "max_escalations": 1},            AlertLevel.P3: {"escalation_minutes": 480, "max_escalations": 0},        }     def get_escalation_chain(self, level: AlertLevel, on_call_engineers: List[Engineer]) -> List[Engineer]:        """获取告警升级链路"""        policy = self.policies[level]        chain = []        current_tier = 1        for _ in range(policy["max_escalations"] + 1):            tier_engineers = [e for e in on_call_engineers if e.tier == current_tier]            chain.extend(tier_engineers)            current_tier += 1        return chain  class OnCallScheduler:    """On-call轮值调度器"""     def __init__(self, engineers: List[Engineer]):        self.engineers = engineers        self.shifts: Dict[str, List[Dict]] = defaultdict(list)     def generate_schedule(        self,        start_date: date,        days: int = 30,        shift_days: int = 7,        primary_count: int = 1,        secondary_count: int = 1    ) -> List[Dict]:        """        生成轮值排班表         使用加权轮转算法，综合考虑工程师的负载均衡和技能匹配         Args:            start_date: 排班开始日期            days: 排班周期天数            shift_days: 每班轮值天数            primary_count: 主值班人数            secondary_count: 备用值班人数         Returns:            排班表列表        """        schedule = []        current_date = start_date        engineer_index = 0         while current_date < start_date + timedelta(days=days):            shift_end = current_date + timedelta(days=shift_days)             # 选择主值班工程师（加权轮转，优先选择负载低的工程师）            primary = self._select_engineers(                count=primary_count,                exclude_names=set(),                start_index=engineer_index            )             # 选择备用值班工程师            primary_names = {e.name for e in primary}            secondary = self._select_engineers(                count=secondary_count,                exclude_names=primary_names,                start_index=engineer_index + primary_count            )             shift = {                "shift_id": f"SHIFT-{current_date.strftime('%Y%m%d')}",                "start_date": current_date.isoformat(),                "end_date": shift_end.isoformat(),                "primary": [{"name": e.name, "tier": e.tier} for e in primary],                "secondary": [{"name": e.name, "tier": e.tier} for e in secondary],            }             schedule.append(shift)             # 更新工程师状态            for e in primary + secondary:                e.shift_count += 1                e.last_shift_end = shift_end             engineer_index = (engineer_index + primary_count + secondary_count) % len(self.engineers)            current_date = shift_end         self.shifts["schedule"] = schedule        return schedule     def _select_engineers(        self,        count: int,        exclude_names: set,        start_index: int    ) -> List[Engineer]:        """选择值班工程师，基于负载均衡"""        available = [e for e in self.engineers if e.name not in exclude_names]         if not available:            return []         # 按值班次数升序排列（负载均衡）        available.sort(key=lambda e: (e.shift_count, start_index))         selected = []        idx = 0        while len(selected) < count and idx < len(available):            selected.append(available[idx])            idx += 1         return selected     def get_shift_stats(self) -> Dict:        """获取值班统计"""        stats = []        for e in self.engineers:            stats.append({                "name": e.name,                "tier": e.tier,                "shift_count": e.shift_count,                "last_shift_end": str(e.last_shift_end) if e.last_shift_end else None,            })        return {"engineers": stats, "total_shifts": len(self.shifts.get("schedule", []))}  class AlertManager:    """告警管理器"""     def __init__(self, scheduler: OnCallScheduler, escalation_policy: EscalationPolicy):        self.scheduler = scheduler        self.escalation_policy = escalation_policy        self.alerts: List[Dict] = []        self.alert_counter = 0     def trigger_alert(        self,        title: str,        description: str,        level: AlertLevel,        source: str    ) -> Dict:        """触发告警并执行升级逻辑"""        self.alert_counter += 1        alert_id = f"ALERT-{self.alert_counter:04d}"         # 获取当前值班工程师        current_shift = self._get_current_shift()        if not current_shift:            return {"error": "No active shift found", "alert_id": alert_id}         # 获取升级链路        on_call_engineers = self._get_on_call_engineers(current_shift)        escalation_chain = self.escalation_policy.get_escalation_chain(level, on_call_engineers)         alert = {            "alert_id": alert_id,            "title": title,            "description": description,            "level": level.value,            "source": source,            "created_at": datetime.now().isoformat(),            "status": "triggered",            "current_shift": current_shift["shift_id"],            "escalation_chain": [                {"name": e.name, "tier": e.tier} for e in escalation_chain            ],            "escalation_level": 0,            "acknowledged_by": None,            "resolved_at": None,        }         self.alerts.append(alert)        return alert     def acknowledge_alert(self, alert_id: str, engineer_name: str) -> Dict:        """确认告警"""        for alert in self.alerts:            if alert["alert_id"] == alert_id:                alert["status"] = "acknowledged"                alert["acknowledged_by"] = engineer_name                alert["acknowledged_at"] = datetime.now().isoformat()                return alert        return {"error": f"Alert {alert_id} not found"}     def escalate_alert(self, alert_id: str) -> Dict:        """升级告警到下一级"""        for alert in self.alerts:            if alert["alert_id"] == alert_id:                current_level = alert["escalation_level"]                max_level = len(alert["escalation_chain"]) - 1                 if current_level >= max_level:                    alert["status"] = "max_escalated"                    return {"warning": "Max escalation reached", "alert": alert}                 alert["escalation_level"] = current_level + 1                alert["status"] = "escalated"                return alert        return {"error": f"Alert {alert_id} not found"}     def resolve_alert(self, alert_id: str, resolution: str) -> Dict:        """解决告警"""        for alert in self.alerts:            if alert["alert_id"] == alert_id:                alert["status"] = "resolved"                alert["resolution"] = resolution                alert["resolved_at"] = datetime.now().isoformat()                return alert        return {"error": f"Alert {alert_id} not found"}     def _get_current_shift(self) -> Optional[Dict]:        """获取当前值班班次"""        schedule = self.scheduler.shifts.get("schedule", [])        today = date.today()        for shift in schedule:            start = date.fromisoformat(shift["start_date"])            end = date.fromisoformat(shift["end_date"])            if start <= today < end:                return shift        return None     def _get_on_call_engineers(self, shift: Dict) -> List[Engineer]:        """获取当前值班的所有工程师"""        names = set()        names.update(e["name"] for e in shift["primary"])        names.update(e["name"] for e in shift["secondary"])        return [e for e in self.scheduler.engineers if e.name in names]     def get_alert_stats(self) -> Dict:        """获取告警统计"""        stats = defaultdict(lambda: {"total": 0, "resolved": 0, "acknowledged": 0, "escalated": 0})        for alert in self.alerts:            level = alert["level"]["name"]            stats[level]["total"] += 1            if alert["status"] == "resolved":                stats[level]["resolved"] += 1            elif alert["status"] == "acknowledged":                stats[level]["acknowledged"] += 1            elif "escalated" in alert["status"]:                stats[level]["escalated"] += 1        return dict(stats)  # ==================== 演示用法 ==================== if __name__ == "__main__":    # 创建工程师团队    engineers = [        Engineer("张伟", tier=1, skills=["kubernetes", "docker", "linux"]),        Engineer("李娜", tier=1, skills=["python", "tensorflow", "prometheus"]),        Engineer("王强", tier=2, skills=["gpu", "nvidia", "cuda", "nccl"]),        Engineer("赵敏", tier=2, skills=["kafka", "spark", "hadoop"]),        Engineer("陈磊", tier=3, skills=["architecture", "mlops", "distributed"]),        Engineer("刘洋", tier=1, skills=["grafana", "alertmanager", "pagerduty"]),    ]     # 初始化调度器和告警管理器    scheduler = OnCallScheduler(engineers)    escalation_policy = EscalationPolicy()    alert_manager = AlertManager(scheduler, escalation_policy)     # 生成30天排班表    schedule = scheduler.generate_schedule(        start_date=date(2026, 8, 1),        days=30,        shift_days=7,        primary_count=1,        secondary_count=1    )     print("=" * 60)    print("ON-CALL轮值排班表")    print("=" * 60)    for shift in schedule:        primary_names = ", ".join(p["name"] for p in shift["primary"])        secondary_names = ", ".join(p["name"] for p in shift["secondary"])        print(f"\n{shift['shift_id']}: {shift['start_date']} ~ {shift['end_date']}")        print(f"  主值班: {primary_names}")        print(f"  备用值班: {secondary_names}")     # 模拟告警场景    print("\n" + "=" * 60)    print("告警模拟")    print("=" * 60)     # 模拟P0告警    alert1 = alert_manager.trigger_alert(        title="GPU集群故障 - 训练任务中断",        description="NCCL通信超时，3个节点无法连接",        level=AlertLevel.P0,        source="prometheus-gpu-monitor"    )    print(f"\n[P0告警] {alert1['alert_id']}: {alert1['title']}")    print(f"  升级链路: {[e['name'] for e in alert1['escalation_chain']]}")     # 确认告警    ack = alert_manager.acknowledge_alert(alert1["alert_id"], "张伟")    print(f"  确认人: {ack.get('acknowledged_by', 'N/A')}")     # 升级告警    esc = alert_manager.escalate_alert(alert1["alert_id"])    print(f"  升级后状态: {esc.get('status', 'N/A')}")     # 解决告警    resolved = alert_manager.resolve_alert(        alert1["alert_id"],        "重启NCCL服务，恢复GPU节点通信，训练任务已恢复"    )    print(f"  最终状态: {resolved.get('status', 'N/A')}")     # 打印统计    print("\n" + "=" * 60)    print("值班统计")    print("=" * 60)    stats = scheduler.get_shift_stats()    for eng in stats["engineers"]:        print(f"  {eng['name']} (Tier-{eng['tier']}): {eng['shift_count']} 次值班")     print("\n" + "=" * 60)    print("告警统计")    print("=" * 60)    alert_stats = alert_manager.get_alert_stats()    for level, stat in alert_stats.items():        print(f"  {level}: 总计{stat['total']}, 已解决{stat['resolved']}, "              f"已确认{stat['acknowledged']}, 已升级{stat['escalated']}")