=================================================================================================
                                   Ringi 3D 架构工坊 · 核心全景
                   【企业级万卡统一 AI 算力平台全景指挥中枢与控制拓扑】

     [多租户复杂业务流量入口] (大模型预训练、微调 SFT、在线 Serving、离线评估 Batch)
            │
            ▼
    ┌────────────────────────────────────────────────────────────────────────┐
    │ 统一云原生调度控制平面 (Kubernetes Control Plane + Kueue / Volcano)   │
    │   • 多层级弹性配额队列 (LocalQueue / ClusterQueue / Cohort 动态借用)    │
    │   • 严格 Gang Scheduling (全上或全不上，彻底杜绝多机相互等待死锁)     │
    │   • 拓扑感知调度器 (Topology-Aware: TP=8 严守单机 NVLink 物理拓扑)    │
    └────────────────────────────────────────────────────────────────────────┘
            │                                                │
            ▼ (训练专网: 8x400G 计算导轨)                     ▼ (混部与切分: HAMi 显存硬隔离)
    ┌────────────────────────────────────────┐   ┌────────────────────────────────────────┐
    │  大规模预训练集群 (Pretrain / SFT)     │   │  在线推理与离线评测混部集群 (Serving)   │
    │  • 8000 卡纯物理独占, 消除虚拟化损耗   │   │  • 2000 卡 HAMi 算力时间片与显存截断   │
    │  • 无损 RoCEv2 双轨导轨网 (Rail-Opt)   │   │  • 在线请求突发时，秒级优雅抢占离线任务│
    │  • 3FS / NVMe 四级并行 Checkpoint 流水线│  │  • 动态保障 P99 TTFT < 400ms SLO 契约  │
    └────────────────────────────────────────┘   └────────────────────────────────────────┘
            │                                                │
            └───────────────────────┬────────────────────────┘
                                    ▼
    ┌────────────────────────────────────────────────────────────────────────┐
    │ 智能高可用与自愈闭环守护进程 (Auto-Healing Sentinel & Runbook):        │
    │   500ms 周期嗅探 GPU XID 故障 ──> 自动打污点隔离 ──> 注入热备节点 ──>   │
    │   拉取最近一次分布式 DCP 切片 ──> 算力回滚损耗严格控制在 5 分钟以内！   │
    └────────────────────────────────────────────────────────────────────────┘
=================================================================================================

三亿元集群陷入全面瘫痪的死锁闭环:
┌──────────────────────────────┐          ┌──────────────────────────────┐
│ 任务 A (需 4096 卡, 已占 3500) │ ──等待──> │ 任务 B (需 2048 卡, 已占 1500) │
└──────────────────────────────┘          └──────────────────────────────┘
               ▲                                         │
               │                  等待                   │
               └─────────────────────────────────────────┘
 ▲ 价值数亿元的 8000 张顶级 GPU，竟全在空转死等握手，每小时直接烧掉近十万元电费！

容器内业务进程 (PyTorch / TensorRT)
            │
            ▼ (调用底层 CUDA API)
    ┌────────────────────────────────────────────────────────┐
    │ HAMi 拦截层 (libvgpu.so - 用户态动态 Hook)             │
    │   • 拦截 cuMemAlloc() / cudaMalloc():                  │
    │     检查: (当前已用显存 + 请求显存) <= Pod 硬配额?     │
    │     - 若超过: 立即返回 CUDA_ERROR_OUT_OF_MEMORY 熔断!  │
    │     - 绝不让多余申请触达物理驱动，保护宿主机邻居安全   │
    │                                                        │
    │   • 拦截 cuLaunchKernel():                             │
    │     基于令牌桶（Token Bucket）算法进行时间片节流:       │
    │     根据 Pod 声明的算力百分比 (如 cores: 30%) 限制发射 │
    └────────────────────────────────────────────────────────┘
            │
            ▼ (通过安全校验的指令)
真实物理驱动 (libcuda.so ➔ NVIDIA Driver ➔ 物理 GPU 芯片)

# [RFC-AI-0045] 企业级万卡统一算力调度与 AI 平台系统设计规范

- **提案状态**：APPROVED (已核准生产执行)
- **目标版本**：Production v2.4
- **主导团队**：AI Platform Core Architecture Team
- **责任架构师**：Ringi Lee

## 1. 业务背景与系统目标 (Context & Goals)
### 1.1 核心目标 (Goals)
1. **超大规模调度**：单集群稳定支撑 10,000 张 GPU 统一编排，预训练平均 MFU 保持在 45% 以上。
2. **死锁彻底清零**：在 100% 满载高竞争场景下，作业级 Gang 调度死锁发生率为 0。
3. **多租户混部与弹性借用**：通过 Kueue 实现跨部门多层级配额弹性借用，全集群综合利用率从 45% 提升至 85% 以上。
4. **高可用 SLA**：硬件单点故障自动检测隔离与热备自愈时间（MTTR）< 5 分钟，单次训练中断算力回滚损耗 < 10 分钟。

### 1.2 非目标 (Non-Goals)
- 本平台不负责具体模型代码内部算法与超参数的自动调优。

## 2. 总体架构拓扑 (System Topology)
系统划分为三大物理/逻辑平面：
- **控制与调度中枢**：Kube-Apiserver + Kueue Controller + Topology-Aware Scheduler Plugin；
- **计算网络数据面**：双轨分离架构。计算专网采用 8x400G 无损 RoCEv2（Rail-Optimized），存储/管理采用双 100G 独立上行；
- **四级存储金字塔**：HBM3 (GPU) ➔ 1.5TB 本地 NVMe (Scratch Cache) ➔ 3FS 并行分布式文件系统 ➔ 对象存储 (S3 冷备)。

## 3. 核心子系统详细规范
### 3.1 调度准入准则 (Admission Policy)
- 任何卡数规模 $\ge 8$ 的作业必须配置 `queue.x-k8s.io/gang-scheduling: strict`；
- 同一个 Tensor Parallel（TP）通信组必须调度在同一个物理 Node（基于 Topology Hints 匹配）；
- 跨机作业必须调度在同一个 Leaf 交换机网络导轨组内，严禁跨 Spine 交换机产生不规则链路跳跃。

### 3.2 多租户隔离与 HAMi 切分规则
- 预训练生产任务一律分配完整物理卡（`nvidia.com/gpu: 8`），严禁任何虚拟化拦截；
- 在线推理与开发调试任务统一采用 HAMi 切分，通过资源声明：
  - `hami.io/gpu-mem: 24000` (硬限 24GB 显存)
  - `hami.io/gpu-cores: 50` (硬限 50% 核心时间片)

## 4. 容灾与故障自愈规程 (Fault-Tolerance Runbook)
- 节点 Daemon 捕获 Xid 错误代码 [31, 43, 45, 48, 62, 79, 92] 时，立即触发 P0 级自愈：
  1. 调用 Kubernetes API 将本节点打上污点并踢出调度池；
  2. 调度器从 Hot-Spare 备用池挑选拓扑对齐的节点加入作业；
  3. 作业 Controller 重载最新 Checkpoint 切片并恢复计算；
  4. 向 On-Call 工程师钉钉/飞书群发送硬件返修工单。

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: kueue_cluster_simulator.py
Description: 云原生 AI 调度中枢：Gang Scheduling 防死锁与多租户配额借用仿真器
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

from typing import List, Dict, Any

class Job:
    def __init__(self, name: str, required_gpus: int, tenant: str, duration_steps: int):
        self.name = name
        self.required_gpus = required_gpus
        self.tenant = tenant
        self.duration_steps = duration_steps
        self.allocated_gpus = 0
        self.status = "PENDING" # PENDING, RUNNING, COMPLETED

class NaiveScheduler:
    """传统逐 Pod 贪心分配调度器 (极易引发死锁)"""
    def __init__(self, total_gpus: int = 16):
        self.total_gpus = total_gpus
        self.available_gpus = total_gpus

    def step(self, jobs: List[Job]) -> str:
        for job in jobs:
            if job.status == "PENDING":
                # 贪心分配：有多少给多少
                can_allocate = min(self.available_gpus, job.required_gpus - job.allocated_gpus)
                if can_allocate > 0:
                    job.allocated_gpus += can_allocate
                    self.available_gpus -= can_allocate
                    if job.allocated_gpus == job.required_gpus:
                        job.status = "RUNNING"
        
        # 检查死锁: 资源用光但没有任何作业处于 RUNNING
        if self.available_gpus == 0 and not any(j.status == "RUNNING" for j in jobs if j.status != "COMPLETED"):
            return "🚨 全局死锁爆发！所有可用 GPU 被瓜分，但没有一个作业凑齐完整算力！"
        return "正常运转中"


class GangKueueScheduler:
    """严格 Gang 调度与原子准入控制调度器 (彻底杜绝死锁)"""
    def __init__(self, total_gpus: int = 16):
        self.total_gpus = total_gpus
        self.available_gpus = total_gpus

    def step(self, jobs: List[Job]):
        for job in jobs:
            if job.status == "PENDING":
                # All-or-Nothing 原子判定
                if self.available_gpus >= job.required_gpus:
                    # 一次性完整分配
                    job.allocated_gpus = job.required_gpus
                    self.available_gpus -= job.required_gpus
                    job.status = "RUNNING"
                else:
                    # 凑不齐则 1 张卡都不占，继续挂起排队
                    job.allocated_gpus = 0


def run_scheduler_demo():
    print("=" * 80)
    print(">> 实战 1：集群调度防死锁对决：传统逐卡调度 vs Kueue 严格 Gang 调度")
    print("=" * 80)

    # 场景：集群共有 16 张 GPU，两个大作业 A(需12卡) 和 B(需8卡) 同时提交争抢
    print("\n【场景 A: 传统逐卡调度 (Naive Kube-Scheduler)】:")
    naive_jobs = [Job("Pretrain-A", required_gpus=12, tenant="LLM", duration_steps=5),
                  Job("FineTune-B", required_gpus=8, tenant="Search", duration_steps=3)]
    
    naive_sched = NaiveScheduler(total_gpus=16)
    result_msg = naive_sched.step(naive_jobs)
    
    for j in naive_jobs:
        print(f"  * 作业 {j.name} (需求 {j.required_gpus} 卡) ──> 已抢占: {j.allocated_gpus} 卡 | 状态: {j.status}")
    print(f"  -> 调度诊断: {result_msg}")

    print("\n【场景 B: Kueue 严格 Gang 调度 (All-or-Nothing)】:")
    gang_jobs = [Job("Pretrain-A", required_gpus=12, tenant="LLM", duration_steps=5),
                 Job("FineTune-B", required_gpus=8, tenant="Search", duration_steps=3)]
    
    gang_sched = GangKueueScheduler(total_gpus=16)
    gang_sched.step(gang_jobs)

    for j in gang_jobs:
        print(f"  * 作业 {j.name} (需求 {j.required_gpus} 卡) ──> 已分配: {j.allocated_gpus} 卡 | 状态: {j.status}")
    print(f"  -> 调度诊断: ✅ 彻底避免半死锁！作业 A 完整拿到 12 卡全速计算，作业 B 优雅排队等待！")
    print("=" * 80)


if __name__ == "__main__":
    run_scheduler_demo()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: hami_memory_interceptor.py
Description: HAMi GPU 显存虚拟化与动态 Hook 拦截熔断仿真器
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

from typing import Dict, Tuple

class HAMiMemoryController:
    """
    HAMi 容器显存配额硬拦截控制器
    """
    def __init__(self, physical_gpu_mem_mb: float = 81920.0):
        self.physical_gpu_mem_mb = physical_gpu_mem_mb
        self.used_physical_mb = 0.0
        # 记录每个容器的配额与当前用量 {container_id: {"limit": mb, "used": mb}}
        self.containers: Dict[str, Dict[str, float]] = {}

    def register_container(self, container_id: str, limit_mb: float):
        """容器初始化注册配额"""
        self.containers[container_id] = {"limit": limit_mb, "used": 0.0}

    def hook_cu_mem_alloc(self, container_id: str, request_bytes: int) -> Tuple[bool, str]:
        """
        拦截业务层的 cuMemAlloc() 调用
        """
        request_mb = request_bytes / (1024 * 1024)
        c_info = self.containers.get(container_id)
        if not c_info:
            return False, "未知容器上下文！"

        # 1. 容器级别硬配额检查 (Cgroup-like)
        if c_info["used"] + request_mb > c_info["limit"]:
            return False, f"触发 HAMi 显存配额硬熔断！当前已用 {c_info['used']:.1f}MB + 申请 {request_mb:.1f}MB > 限制 {c_info['limit']:.1f}MB"

        # 2. 物理显存余量检查 (保护宿主机)
        if self.used_physical_mb + request_mb > self.physical_gpu_mem_mb:
            return False, f"物理 GPU 显存耗尽 (OOM)！"

        # 通过安全校验，放行底层申请
        c_info["used"] += request_mb
        self.used_physical_mb += request_mb
        return True, f"成功分配 {request_mb:.1f}MB | 容器当前占用: {c_info['used']:.1f}MB"


def run_hami_demo():
    print("=" * 80)
    print(">> 实战 2：HAMi 显存虚拟化：多租户动态 Hook 拦截与安全隔离实测")
    print("=" * 80)

    # 模拟在一张 A100 (80GB) 上混部两个在线推理 Pod
    controller = HAMiMemoryController(physical_gpu_mem_mb=81920.0)
    controller.register_container("Serving-Pod-1", limit_mb=24576.0) # 24GB 配额
    controller.register_container("Serving-Pod-2", limit_mb=24576.0) # 24GB 配额

    print("[阶段 1: 正常加载模型权重 (各申请 14GB)]")
    ok1, msg1 = controller.hook_cu_mem_alloc("Serving-Pod-1", 14 * 1024 * 1024 * 1024)
    ok2, msg2 = controller.hook_cu_mem_alloc("Serving-Pod-2", 14 * 1024 * 1024 * 1024)
    print(f"  - Serving-Pod-1: {'✅' if ok1 else '❌'} {msg1}")
    print(f"  - Serving-Pod-2: {'✅' if ok2 else '❌'} {msg2}")

    print("\n[阶段 2: Pod-1 突发内存泄露，试图恶意申请超额 15GB 显存]")
    ok_leak, msg_leak = controller.hook_cu_mem_alloc("Serving-Pod-1", 15 * 1024 * 1024 * 1024)
    print(f"  - Serving-Pod-1 越界申请: {'✅' if ok_leak else '🚨 成功拦截'} {msg_leak}")

    print("\n[阶段 3: 检查被拦截后相邻 Pod-2 的安全状态]")
    ok_safe, msg_safe = controller.hook_cu_mem_alloc("Serving-Pod-2", 2 * 1024 * 1024 * 1024)
    print(f"  - Serving-Pod-2 正常业务请求: {'✅ 安全放行' if ok_safe else '❌'} {msg_safe}")
    print("=" * 80)


if __name__ == "__main__":
    run_hami_demo()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: cluster_auto_healing_daemon.py
Description: 生产级万卡集群 XID 故障嗅探与热备自愈控制器
Author: Ringi (AI Infra Architect)
Standards: Full-Output Enforcement, Zero-Omission, Production-Ready
"""

from typing import List, Dict, Any

class ClusterAutoHealingDaemon:
    """
    万卡集群高可用故障自愈调度引擎
    """
    def __init__(self):
        # 记录集群节点状态
        self.active_nodes = ["node-01", "node-02", "node-03", "node-04"]
        self.hot_spare_nodes = ["spare-node-01", "spare-node-02"]
        self.tainted_nodes: List[str] = []
        
        # 致命硬件 Xid 错误代码黑名单
        self.critical_xid_codes = {
            31: "GPU 内存页面错误 (Memory Page Fault)",
            48: "不可纠正的 Double-bit ECC 校验翻转",
            79: "GPU 掉卡 (Fallen off the Bus)",
            92: "高频过热硬件降频告警 (Thermal Throttle)"
        }

    def inspect_node_telemetry(self, node_id: str, reported_xid: int) -> Dict[str, Any]:
        """
        嗅探节点健康上报
        """
        if reported_xid in self.critical_xid_codes:
            diag_reason = self.critical_xid_codes[reported_xid]
            # 1. 触发节点驱逐与隔离 (Cordon & Taint)
            self.tainted_nodes.append(node_id)
            if node_id in self.active_nodes:
                self.active_nodes.remove(node_id)

            # 2. 从热备节点池唤醒新节点置换
            if self.hot_spare_nodes:
                new_spare = self.hot_spare_nodes.pop(0)
                self.active_nodes.append(new_spare)
                action_plan = f"隔离故障机 [{node_id}] 并打上 NoSchedule 污点，热调配 [{new_spare}] 替换入网并重载 Checkpoint！"
            else:
                action_plan = f"隔离故障机 [{node_id}]，但热备池枯竭，作业被迫进入挂起等待！"

            return {
                "healthy": False,
                "node_id": node_id,
                "xid": reported_xid,
                "reason": diag_reason,
                "action": action_plan
            }
        return {"healthy": True, "node_id": node_id, "reason": "节点运行健康"}


def run_healing_demo():
    print("=" * 80)
    print(">> 实战 3：万卡超算平台毫秒级 XID 故障嗅探与热备自愈恢复测试")
    print("=" * 80)

    daemon = ClusterAutoHealingDaemon()
    print(f"初始活跃节点: {daemon.active_nodes} | 热备池节点: {daemon.hot_spare_nodes}")

    # 模拟 node-02 突发 XID 79 (掉卡)
    print("\n[突发事件]: node-02 发生 PCIe 链路重置，抛出 XID 79 致命硬件错误！")
    event = daemon.inspect_node_telemetry(node_id="node-02", reported_xid=79)

    print(f"  - 诊断结论: [🚨 致命告警] {event['reason']} (Xid={event['xid']})")
    print(f"  - 平台自愈行动: {event['action']}")
    print(f"\n自愈后活跃节点: {daemon.active_nodes} (成功踢出故障卡，注入 spare-node-01！)")
    print(f"被隔离污点节点: {daemon.tainted_nodes}")
    print("=" * 80)


if __name__ == "__main__":
    run_healing_demo()

万卡调度莫乱投，原子准入消死囚。
部门配额可借用，主人回眸优雅收。
拓扑感知机内守，双轨导轨畅无忧。
显存拦截硬如铁，虚实隔离不发愁。
故障探针毫秒嗅，热备自愈立神舟！