┌─────────────┐     Register      ┌──────────────────┐
│ Device Plugin │ ───────────────▶ │   Kubelet        │
│  (gRPC)      │ ◀─────────────── │  (Plugin Watcher)│
└─────────────┘   ListAndWatch    └────────┬─────────┘
                                            │ Allocate
                                            ▼
                                    ┌──────────────────┐
                                    │  Container Runtime│
                                    └──────────────────┘

apiVersion: media.example.com/v1alpha1
kind: MediaResource
metadata:
  name: gpu-nvidia-a100-0
spec:
  type: GPU
  vendor: NVIDIA
  model: A100-SXM4-40GB
  capacity:
    memory: "40Gi"
    computeUnits: 108  # SM 数
    nvenc: 5
    nvdec: 5
  topology:
    node: worker-3
    pciAddr: "0000:3b:00.0"
    numaNode: 1
  attributes:
    driverVersion: "535.154.05"
    cudaVersion: "12.2"
    migCapable: true

QueueSort → PreFilter → Filter → PostFilter → PreScore → Score → Reserve → Permit → PreBind → Bind → PostBind
                │                                            │
                ▼                                            ▼
        MediaPreFilter Plugin                          MediaScore Plugin
        - 设备存在性校验                                 - 亲和性打分
        - 拓扑约束检查                                   - 负载均衡权重
        - 配额/租户限额                                  - 碎片化惩罚

func (p *MediaScorePlugin) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, _ := p.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    devices := getAllocatableMediaDevices(nodeInfo) // 从 NodeInfo 扩展字段读取
    
    var score int64
    for _, dev := range devices {
        // 1. 亲和性匹配
        if matchesPodRequirement(dev, pod) {
            score += 50
        }
        // 2. NUMA 局部性
        if dev.Topology.NUMANode == getPreferredNUMANode(pod) {
            score += 20
        }
        // 3. 碎片化惩罚：剩余显存 < 请求显存 * 1.5 视为碎片
        if dev.FreeMemory < podRequestMemory*1.5 {
            score -= 30
        }
        // 4. 负载均衡：已分配设备数越少分越高
        score += int64(100 - len(devices)*5)
    }
    return normalize(score), nil
}

npu-device-plugin/
├── cmd/
│   └── npu-device-plugin/
│       └── main.go
├── pkg/
│   ├── npu/
│   │   ├── discover.go      # 硬件发现逻辑
│   │   ├── health.go        # 健康检查
│   │   └── allocate.go      # 分配逻辑
│   └── kube/
│       └── client.go        # K8s 客户端封装
├── go.mod
├── go.sum
├── Dockerfile
├── Makefile
└── deploy/
    ├── daemonset.yaml
    ├── rbac.yaml
    └── servicemonitor.yaml  # Prometheus 监控

module github.com/example/npu-device-plugin

go 1.22

require (
    github.com/containerd/nri/pkg/api v0.0.0-20240115
    github.com/fsnotify/fsnotify v1.7.0
    github.com/prometheus/client_golang v1.19.0
    google.golang.org/grpc v1.62.0
    google.golang.org/protobuf v1.33.0
    k8s.io/api v0.29.0
    k8s.io/client-go v0.29.0
    k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1 v0.29.0
    k8s.io/klog/v2 v2.120.0
)

// pkg/npu/discover.go
package npu

import (
    "context"
    "fmt"
    "os"
    "path/filepath"
    "strconv"
    "strings"

    "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1"
)

const (
    npuDevicePath = "/dev"
    npuPrefix     = "npu"
    sysfsNPUPath  = "/sys/class/npu"
)

type NPUDevice struct {
    ID           string
    Path         string
    Model        string
    MemoryTotal  uint64 // MB
    MemoryUsed   uint64
    ComputeTOPS  int
    FirmwareVer  string
    Health       v1beta1.Healthy
    NUMAAffinity int
}

func DiscoverNPUs(ctx context.Context) ([]*NPUDevice, error) {
    entries, err := os.ReadDir(sysfsNPUPath)
    if err != nil {
        return nil, fmt.Errorf("read sysfs npu: %w", err)
    }

    var devices []*NPUDevice
    for _, entry := range entries {
        if !strings.HasPrefix(entry.Name(), npuPrefix) {
            continue
        }
        devPath := filepath.Join(npuDevicePath, entry.Name())
        if _, err := os.Stat(devPath); os.IsNotExist(err) {
            continue // 设备节点未就绪
        }

        dev := &NPUDevice{
            ID:   entry.Name(),
            Path: devPath,
        }
        // 读取 sysfs 属性
        dev.Model = readSysfsAttr(devPath, "model")
        dev.MemoryTotal = parseUint(readSysfsAttr(devPath, "memory_total"))
        dev.ComputeTOPS = parseInt(readSysfsAttr(devPath, "compute_tops"))
        dev.FirmwareVer = readSysfsAttr(devPath, "firmware_version")
        dev.NUMAAffinity = parseInt(readSysfsAttr(devPath, "numa_node"))
        dev.Health = v1beta1.Healthy

        devices = append(devices, dev)
    }
    return devices, nil
}

// pkg/npu/health.go
func (p *NPUPlugin) ListAndWatch(e *v1beta1.Empty, s v1beta1.DevicePlugin_ListAndWatchServer) error {
    ticker := time.NewTicker(10 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            devs := p.scanAndUpdate()
            resp := &v1beta1.ListAndWatchResponse{Devices: toPluginDevices(devs)}
            if err := s.Send(resp); err != nil {
                klog.ErrorS(err, "Send ListAndWatch response failed")
                return err
            }
        case <-s.Context().Done():
            return s.Context().Err()
        }
    }
}

func (p *NPUPlugin) scanAndWatch() []*NPUDevice {
    devs, err := DiscoverNPUs(context.Background())
    if err != nil {
        klog.ErrorS(err, "Discover NPUs failed")
        return p.devices // 保持上次状态
    }
    // 健康检查：心跳、温度、ECC 错误计数
    for _, d := range devs {
        if !checkNPUHealth(d) {
            d.Health = v1beta1.Unhealthy
        }
    }
    p.mu.Lock()
    p.devices = devs
    p.mu.Unlock()
    return devs
}

// pkg/npu/allocate.go
func (p *NPUPlugin) Allocate(ctx context.Context, reqs *v1beta1.AllocateRequest) (*v1beta1.AllocateResponse, error) {
    responses := make([]*v1beta1.ContainerAllocateResponse, 0, len(reqs.ContainerRequests))

    for _, cr := range reqs.ContainerRequests {
        var mounts []*v1beta1.Mount
        var envs []*v1beta1.EnvVar
        var annotations map[string]string

        for _, devID := range cr.DevicesIDs {
            dev := p.getDevice(devID)
            if dev == nil {
                return nil, fmt.Errorf("device %s not found", devID)
            }

            // 1. 设备节点挂载
            mounts = append(mounts, &v1beta1.Mount{
                ContainerPath: dev.Path,
                HostPath:      dev.Path,
                ReadOnly:      false,
            })

            // 2. 环境变量：供容器内运行时识别
            envs = append(envs,
                &v1beta1.EnvVar{Key: "NPU_DEVICE_ID", Value: dev.ID},
                &v1beta1.EnvVar{Key: "NPU_MEMORY_TOTAL", Value: strconv.FormatUint(dev.MemoryTotal, 10)},
                &v1beta1.EnvVar{Key: "NPU_COMPUTE_TOPS", Value: strconv.Itoa(dev.ComputeTOPS)},
            )

            // 3. 拓扑感知注解：供 CNI/运行时做 CPU 绑核参考
            if annotations == nil {
                annotations = make(map[string]string)
            }
            annotations[fmt.Sprintf("npu.example.com/%s-numa", dev.ID)] = strconv.Itoa(dev.NUMAAffinity)
        }

        responses = append(responses, &v1beta1.ContainerAllocateResponse{
            Mounts:      mounts,
            Envs:        envs,
            Annotations: annotations,
        })
    }
    return &v1beta1.AllocateResponse{ContainerResponses: responses}, nil
}

# deploy/daemonset.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: npu-device-plugin
  namespace: kube-system
spec:
  selector:
    matchLabels:
      app: npu-device-plugin
  template:
    metadata:
      labels:
        app: npu-device-plugin
    spec:
      serviceAccountName: npu-device-plugin
      priorityClassName: system-node-critical
      tolerations:
      - operator: Exists
      volumes:
      - name: device-plugin
        hostPath:
          path: /var/lib/kubelet/device-plugins
          type: DirectoryOrCreate
      - name: npu-devices
        hostPath:
          path: /dev
      - name: sysfs
        hostPath:
          path: /sys/class/npu
      containers:
      - name: npu-device-plugin
        image: registry.example.com/npu-device-plugin:v1.2.0
        imagePullPolicy: IfNotPresent
        securityContext:
          privileged: true  # 需要访问 /dev、sysfs
        volumeMounts:
        - name: device-plugin
          mountPath: /var/lib/kubelet/device-plugins
        - name: npu-devices
          mountPath: /dev
        - name: sysfs
          mountPath: /sys/class/npu
        env:
        - name: NODE_NAME
          valueFrom:
            fieldRef:
              fieldPath: spec.nodeName
        ports:
        - containerPort: 9090  # Prometheus metrics
        livenessProbe:
          httpGet:
            path: /healthz
            port: 9090
          initialDelaySeconds: 10
          periodSeconds: 30

# media-transcode-job.yaml
apiVersion: batch/v1
kind: Job
metadata:
  name: transcode-4k-h265-{{.TaskID}}
  labels:
    app: media-transcode
    workload-type: batch
spec:
  template:
    spec:
      restartPolicy: OnFailure
      affinity:
        nodeAffinity:
          requiredDuringSchedulingIgnoredDuringExecution:
            nodeSelectorTerms:
            - matchExpressions:
              - key: media.example.com/gpu-vendor
                operator: In
                values: ["nvidia"]
      containers:
      - name: ffmpeg
        image: registry.example.com/ffmpeg-nvidia:6.1-cuda12
        resources:
          limits:
            nvidia.com/gpu: 1
            media.example.com/nvenc: "1"
            media.example.com/gpu-memory: "8Gi"
          requests:
            nvidia.com/gpu: 1
            media.example.com/nvenc: "1"
            media.example.com/gpu-memory: "8Gi"
        env:
        - name: NVIDIA_VISIBLE_DEVICES
          value: "all"
        - name: NVIDIA_DRIVER_CAPABILITIES
          value: "compute,video,utility"
        command: ["/bin/bash", "-c"]
        args:
        - |
          ffmpeg -hwaccel cuda -i input.ts -c:v hevc_nvenc -preset p4 -b:v 15M output.mp4

# ai-inference-hpa.yaml
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: video-analytics-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: video-analytics
  minReplicas: 2
  maxReplicas: 32
  metrics:
  - type: Pods
    pods:
      metric:
        name: npu_inference_queue_length
      target:
        type: AverageValue
        averageValue: "10"  # 每 Pod 积压 10 帧触发扩容
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 30
      policies:
      - type: Percent
        value: 100
        periodSeconds: 15
    scaleDown:
      stabilizationWindowSeconds: 300
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

flowchart TD
    A[Device Plugin 心跳检测异常] --> B{连续 3 次 Unhealthy}
    B -->|是| C[标记设备 Unschedulable]
    C --> D[驱逐该设备上 Pod<br/>(配合 Taint/Toleration)]
    D --> E[发送告警至运维平台]
    E --> F[人工/自动化诊断]
    F --> G{硬件可恢复?}
    G -->|是| H[重新 Register 设备]
    G -->|否| I[申请硬件更换/下线节点]

// pkg/udm/types.go - 统一设备模型核心定义
type UnifiedDevice struct {
    // 标识维度
    UID            string            `json:"uid"`             // 集群全局唯一: <vendor>-<node>-<pci/bdf>
    Vendor         string            `json:"vendor"`          // nvidia | amd | intel | huawei | cambricon | hygon
    Type           DeviceType        `json:"type"`            // GPU | NPU | VPU | FPGA | CPU
    Model          string            `json:"model"`           // A100 | MI300 | Gaudi2 | Ascend910 | MLU590
    
    // 算力量化维度（统一单位便于调度器打分）
    ComputePower   ComputeSpec       `json:"computePower"`    // 统一换算为 "TFLOPS-FP16" 或 "TOPS-INT8"
    Memory         MemorySpec        `json:"memory"`          // 总量/可分配/带宽
    CodecEngine    CodecCapability   `json:"codecEngine"`     // 编解码硬件单元抽象
    
    // 拓扑与亲和性
    Topology       TopologyInfo      `json:"topology"`        // NUMA/PCIE/Switch/Link
    Firmware       FirmwareInfo      `json:"firmware"`        // 版本/签名/安全启动状态
    
    // 运行时能力标签（供调度器匹配）
    Capabilities   []string          `json:"capabilities"`    // ["cuda-12.2", "nvenc-5", "vulkan", "rocm-6.0", "cann-8.0"]
    
    // 扩展字段（厂商私有透传）
    VendorExtensions map[string]string `json:"vendorExtensions,omitempty"`
}

type CodecCapability struct {
    // 统一抽象编解码能力，避免上层感知厂商差异
    Encoders map[CodecFormat]int `json:"encoders"` // format -> 实例数
    Decoders map[CodecFormat]int `json:"decoders"`
    MaxResolution string         `json:"maxResolution"` // "8192x4320"
    MaxThroughput string         `json:"maxThroughput"` // "480 1080p@30fps streams"
}

┌─────────────────────────────────────────────────────────────┐
│                    Unified Device Manager                    │
│  (Watch MediaResource CRD → Build UnifiedDevice Cache)       │
└─────────────────────────────────────────────────────────────┘
                              ▲
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
    ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
    │ NVIDIA Adapter  │ │  Ascend Adapter │ │  Cambricon Adp  │
    │ (Wrapper)       │ │ (Wrapper)       │ │ (Wrapper)       │
    └────────┬────────┘ └────────┬────────┘ └────────┬────────┘
             │                  │                  │
    ┌────────▼────────┐ ┌───────▼────────┐ ┌───────▼────────┐
    │ NVIDIA Device   │ │ Ascend Device  │ │ Cambricon Dev  │
    │ Plugin (Official)│ │ Plugin (Official)│ │ Plugin (Custom)│
    └─────────────────┘ └─────────────────┘ └─────────────────┘

// pkg/adapter/interface.go
type VendorAdapter interface {
    // 将厂商原始设备信息转换为统一模型
    Translate(rawDev interface{}) (*udm.UnifiedDevice, error)
    
    // 校验驱动/固件/容器运行时兼容性矩阵
    ValidateCompatibility(nodeEnv *NodeEnvironment) *CompatibilityReport
    
    // 生成厂商特定的容器运行时配置
    GenerateRuntimeSpec(dev *udm.UnifiedDevice, req *AllocateRequest) (*RuntimeSpec, error)
    
    // 健康检查扩展点
    DeepHealthCheck(dev *udm.UnifiedDevice) *HealthDetail
}

// 注册中心单例
var Registry = make(map[string]VendorAdapter)

func RegisterAdapter(vendor string, adapter VendorAdapter) {
    Registry[vendor] = adapter
}

// pkg/adapter/nvidia/adapter.go
func (a *NvidiaAdapter) GenerateRuntimeSpec(dev *udm.UnifiedDevice, req *AllocateRequest) (*RuntimeSpec, error) {
    // 1. 解析请求中的编解码实例数需求
    nvencReq := req.Annotations["media.example.com/nvenc-count"]
    nvdecReq := req.Annotations["media.example.com/nvdec-count"]
    
    // 2. 构造 NVIDIA Container Toolkit 兼容的 CDI Spec 或 env/mounts
    spec := &RuntimeSpec{
        Env: []EnvVar{
            {Key: "NVIDIA_VISIBLE_DEVICES", Value: dev.UID},
            {Key: "NVIDIA_DRIVER_CAPABILITIES", Value: "compute,video,utility,graphics"},
        },
        Mounts: []Mount{
            {Source: "/usr/lib/x86_64-linux-gnu/libnvidia-encode.so", Target: "/usr/lib/x86_64-linux-gnu/libnvidia-encode.so", ReadOnly: true},
            {Source: "/usr/lib/x86_64-linux-gnu/libnvidia-decode.so", Target: "/usr/lib/x86_64-linux-gnu/libnvidia-decode.so", ReadOnly: true},
        },
        DeviceNodes: []DeviceNode{
            {Path: fmt.Sprintf("/dev/nvidia%d", dev.PCIBusID), Permissions: "rwm"},
            {Path: fmt.Sprintf("/dev/nvidiactl"), Permissions: "rwm"},
            {Path: fmt.Sprintf("/dev/nvidia-modeset"), Permissions: "rwm"},
        },
    }
    
    // 3. 若启用 MIG，注入 MIG UUID
    if dev.VendorExtensions["mig.enabled"] == "true" {
        spec.Env = append(spec.Env, EnvVar{Key: "NVIDIA_VISIBLE_DEVICES", Value: dev.VendorExtensions["mig.uuid"]})
    }
    return spec, nil
}

# 兼容性矩阵 ConfigMap 示例
apiVersion: v1
kind: ConfigMap
metadata:
  name: hardware-compatibility-matrix
  namespace: kube-system
data:
  matrix.yaml: |
    nvidia:
      - driver: "535.x"
        cuda: "12.2"
        containerToolkit: "1.14.x"
        k8s: ["1.27", "1.28", "1.29"]
        plugin: "v0.14.x"
        kernel: ["5.15", "6.1", "6.5"]
      - driver: "550.x"
        cuda: "12.4"
        containerToolkit: "1.15.x"
        k8s: ["1.28", "1.29", "1.30"]
        plugin: "v0.15.x"
        kernel: ["6.1", "6.5", "6.8"]
    huawei-ascend:
      - driver: "24.0.rc1"
        cann: "8.0.RC1"
        k8s: ["1.28"]
        plugin: "v1.0.0"
        kernel: ["6.1"]

# Node 资源上报示例 (通过 Device Plugin 扩展资源上报)
apiVersion: v1
kind: Node
metadata:
  name: worker-gpu-01
  annotations:
    # 物理 GPU 级别
    media.example.com/gpu-a100-0: '{"pci":"0000:3b:00.0","mig":"disabled"}'
    # 编解码器实例级别 (可调度资源)
    media.example.com/nvenc: "5"
    media.example.com/nvdec: "5"
    media.example.com/jpeg: "1"
status:
  capacity:
    nvidia.com/gpu: "1"
    media.example.com/nvenc: "5"
    media.example.com/nvdec: "5"
  allocatable:
    nvidia.com/gpu: "1"
    media.example.com/nvenc: "5"
    media.example.com/nvdec: "5"

resources:
  limits:
    media.example.com/nvenc: "1"   # 独占 1 个编码器实例
    media.example.com/nvdec: "1"   # 独占 1 个解码器实例
    media.example.com/gpu-memory: "2Gi" # 显存配额
  requests:
    media.example.com/nvenc: "1"
    media.example.com/nvdec: "1"
    media.example.com/gpu-memory: "2Gi"

// pkg/npu/allocate.go - 编解码器实例分配逻辑
func (p *NPUPlugin) allocateCodecInstances(req *v1beta1.ContainerRequest) ([]*CodecInstance, error) {
    // 1. 从 Node 级共享内存/Etcd 读取当前可用实例位图
    bitmap := p.getCodecBitmap(req.DeviceType) // nvenc/nvdec/jpeg
    
    // 2. 优先分配 NUMA 局部的实例
    preferredNUMA := getPreferredNUMA(req)
    instances := bitmap.FindAndAlloc(req.Count, preferredNUMA)
    if len(instances) < req.Count {
        return nil, fmt.Errorf("insufficient codec instances: need %d, got %d", req.Count, len(instances))
    }
    
    // 3. 生成环境变量指定实例 ID (FFmpeg/VA-API 通过索引选择)
    envs := []*v1beta1.EnvVar{
        {Key: "NVENC_INSTANCE_IDS", Value: strings.Join(instances.IDs, ",")},
        {Key: "NVDEC_INSTANCE_IDS", Value: strings.Join(instances.IDs, ",")},
    }
    return instances, nil
}

┌──────────────┐     DMA-BUF / GPUDirect RDMA      ┌──────────────┐
│  NIC (DPDK/   │ ◀──────────────────────────────▶  │  GPU/NPU     │
│  XDP/AF_XDP)  │      Zero-Copy Buffer Pool        │  (Decode/    │
└──────┬───────┘                                    │   Infer/     │
       │                                            │   Encode)    │
       │  V4L2 / VA-API / CUDA Interop              └──────┬───────┘
       ▼                                                   │
┌──────────────┐     Shared DMA-BUF FD (dmabuf-heap)      │
│  Userspace   │ ◀────────────────────────────────────────┘
│  Pipeline    │
│ (FFmpeg/     │
│  GStreamer)  │
└──────────────┘

securityContext:
  capabilities:
    add: ["SYS_ADMIN", "SYS_RESOURCE", "IPC_LOCK", "CAP_DAC_OVERRIDE"] # DMA-BUF 需要 SYS_ADMIN 创建 heap
  privileged: false # 严禁特权容器

// pkg/npu/allocate.go - 新增 DMA-BUF Heap 挂载
func (p *NPUPlugin) getDMABUFMounts() []*v1beta1.Mount {
    return []*v1beta1.Mount{
        {ContainerPath: "/dev/dma_heap/system", HostPath: "/dev/dma_heap/system", ReadOnly: false},
        {ContainerPath: "/dev/dma_heap/linux,cma", HostPath: "/dev/dma_heap/linux,cma", ReadOnly: false},
        {ContainerPath: "/dev/ion", HostPath: "/dev/ion", ReadOnly: false}, // 兼容旧内核
    }
}

annotations:
  media.example.com/zero-copy: "true"
  media.example.com/dmabuf-heap: "system,cma"  # 指定使用的 heap
  media.example.com/gpudirect-rdma: "true"     # 启用 GPUDirect RDMA

# Dockerfile.media-runtime
FROM nvidia/cuda:12.4-devel-ubuntu22.04
# 安装 V4L2, GDRCopy, DMA-BUF 测试工具
RUN apt-get update && apt-get install -y 
    libv4l2rds0 v4l-utils 
    libdrm-dev libegl1-mesa-dev libgbm-dev 
    && git clone https://github.com/NVIDIA/gdrcopy.git /tmp/gdrcopy 
    && cd /tmp/gdrcopy && make prefix=/usr install 
    && rm -rf /tmp/gdrcopy
# 预装 FFmpeg/GStreamer 支持 DMA-BUF 版本
COPY --from=builder /usr/local/bin/ffmpeg /usr/local/bin/

# 租户资源配额
apiVersion: v1
kind: ResourceQuota
metadata:
  name: tenant-a-quota
  namespace: tenant-a
spec:
  hard:
    requests.nvidia.com/gpu: "10"
    requests.media.example.com/nvenc: "20"
    requests.media.example.com/gpu-memory: "80Gi"
    limits.media.example.com/gpu-memory: "100Gi"
    # 优先级类配额
    priorityclasses.high: "5"
    priorityclasses.medium: "20"
---
# PriorityClass 定义 (集群级)
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: media-realtime-high
value: 1000000
globalDefault: false
preemptionPolicy: PreemptLowerPriority
description: "Real-time transcoding / Live streaming - Preempt batch inference"
---
apiVersion: scheduling.k8s.io/v1
kind: PriorityClass
metadata:
  name: media-batch-low
value: 10000
globalDefault: false
preemptionPolicy: Never
description: "Offline batch transcoding / Model training"

# kube-scheduler config
profiles:
- pluginConfig:
  - name: DefaultPreemption
    args:
      apiVersion: kubescheduler.config.k8s.io/v1
      kind: DefaultPreemptionArgs
      minCandidateNodesPercentage: 10
      # 关键：仅允许高优先级抢占低优先级，且仅抢占同租户或 BestEffort Pod
      # 需配合自定义 Preemption Plugin 实现租户感知抢占

        ┌─────────────────────┐
        │  E2E 硬件在环 (HIL)  │  ← 真实 GPU/NPU/VPU 集群、真实流量、混沌注入
        │  (每周/发布前)       │
        ├─────────────────────┤
        │  集成测试 (Device    │  ← Kind/K3s + 模拟设备 / 共享测试集群 1-2 卡
        │   Plugin + Scheduler)│
        ├─────────────────────┤
        │  单元测试 (Mock     │  ← Go Test + gomock, 覆盖率 > 85%
        │   Kubelet/Device)   │
        └─────────────────────┘

# .gitlab-ci.yml / Jenkinsfile 片段
stages:
  - unit-test
  - integration-test
  - hil-test          # Hardware-in-the-Loop
  - performance-baseline
  - canary-deploy

hil-test:
  stage: hil-test
  tags: [gpu-a100, npu-ascend, vpu-vcn]  # 标签调度到物理测试节点
  variables:
    TEST_CLUSTER_KUBECONFIG: /secrets/test-cluster.kubeconfig
    MEDIA_TEST_ASSETS_BUCKET: s3://media-test-assets/4k-h265-hevc/
  script:
    - |
      # 1. 部署待测 Plugin (Canary 版本)
      helm upgrade --install npu-plugin ./charts/npu-device-plugin 
        --namespace kube-system 
        --set image.tag=$CI_COMMIT_SHA 
        --set logLevel=debug
    - |
      # 2. 运行媒体压测套件
      python3 -m pytest tests/hil/ 
        --tb=short 
        -k "transcode or inference" 
        --junitxml=hil-report.xml 
        --benchmark-json=benchmark.json 
        --hypothesis-profile=ci
    - |
      # 3. 混沌工程注入 (网络分区、设备热拔插模拟、驱动崩溃)
      chaosctl create experiment 
        --target=npu-device-plugin 
        --action=pod-kill 
        --namespace=kube-system 
        --duration=30s
    - |
      # 4. 校验核心指标
      python3 scripts/validate_sla.py 
        --benchmark benchmark.json 
        --baseline s3://media-baselines/a100-transcode-v1.2.json 
        --threshold-latency-p99=1.2 
        --threshold-throughput-drop=0.05
  artifacts:
    reports:
      junit: hil-report.xml
    expire_in: 30 days
  allow_failure: false
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
    - if: $CI_COMMIT_TAG

{
  "metadata": {
    "hardware": "NVIDIA A100-SXM4-40GB",
    "driver": "550.90.07",
    "cuda": "12.4",
    "plugin_version": "v1.3.0-rc2",
    "kernel": "6.5.0-1019-azure",
    "timestamp": "2024-05-20T10:00:00Z"
  },
  "benchmarks": {
    "transcode_4k_hevc_nvenc": {
      "throughput_fps": 480,
      "latency_p50_ms": 12.3,
      "latency_p99_ms": 28.7,
      "gpu_util_avg": 0.78,
      "nvenc_util_avg": 0.92,
      "memory_used_gb": 6.4
    },
    "inference_yolov8_int8_batch4": {
      "throughput_fps": 1250,
      "latency_p99_ms": 8.2,
      "npu_util_avg": 0.85
    }
  },
  "thresholds": {
    "latency_p99_regression": 1.15,
    "throughput_regression": 0.95,
    "error_rate_max": 0.001
  }
}

# scripts/validate_sla.py
def check_regression(current: dict, baseline: dict, thresholds: dict) -> tuple[bool, list]:
    errors = []
    for test_name, cur_metrics in current.items():
        base_metrics = baseline.get(test_name)
        if not base_metrics:
            continue
        
        # 吞吐回归
        thr_ratio = cur_metrics['throughput_fps'] / base_metrics['throughput_fps']
        if thr_ratio < thresholds['throughput_regression']:
            errors.append(f"{test_name}: Throughput regression {thr_ratio:.2%} < {thresholds['throughput_regression']:.0%}")
        
        # 延迟恶化
        lat_ratio = cur_metrics['latency_p99_ms'] / base_metrics['latency_p99_ms']
        if lat_ratio > thresholds['latency_p99_regression']:
            errors.append(f"{test_name}: Latency P99 regression {lat_ratio:.2%} > {thresholds['latency_p99_regression']:.0%}")
        
        # 错误率
        if cur_metrics.get('error_rate', 0) > thresholds['error_rate_max']:
            errors.append(f"{test_name}: Error rate {cur_metrics['error_rate']:.4f} > {thresholds['error_rate_max']:.4f}")
    
    return len(errors) == 0, errors

┌────────────────────────────────────────────────────────────────┐
│                     Resource Manager (Koordinator / Volcano)    │
│  - 细粒度 Resource Quota (CPU/内存/显存/编解码器实例/NPU TOPS)  │
│  - 动态资源超售模型 (基于历史利用率预测)                        │
└────────────────────────────────────────────────────────────────┘
                              │
        ┌─────────────────────┼─────────────────────┐
        ▼                     ▼                     ▼
┌───────────────┐     ┌───────────────┐     ┌───────────────┐
│  Online Tier  │     │  Batch Tier   │     │  BestEffort   │
│  (LS/RTMP/    │     │  (Transcode/  │     │  (Model Train │
│   WebRTC)     │     │   Inference)  │     │   /Data Prep) │
├───────────────┤     ├───────────────┤     ├───────────────┤
│ QoS: Guaranteed│     │ QoS: Burstable│     │ QoS: BestEffort│
│ Priority: High │     │ Priority: Med │     │ Priority: Low │
│ CPU: 独占绑核   │     │ CPU: 共享+限额 │     │ CPU: 碎片填充  │
│ Mem: 独占+预留  │     │ Mem: 限额     │     │ Mem: 限额     │
│ GPU: MIG/独占   │     │ GPU: 时间片/  │     │ GPU: 闲时抢占  │
│                │     │   MPS 共享    │     │               │
└───────────────┘     └───────────────┘     └───────────────┘

# koordinator-colocation-profile
apiVersion: config.koordinator.sh/v1alpha1
kind: ColocationProfile
metadata:
  name: media-colocation-profile
spec:
  resourceQoSStrategy:
    # CPU 超售：在线独占物理核，离线使用超线程/空闲核
    cpu:
      enable: true
      # 在线 Pod 请求 = Limit，离线 Pod 可申请 (NodeCapacity - OnlineRequest) * OverSellRatio
      overSellRatio: 1.5 
      # 核绑定策略
      cpusetPolicy: "static-burst" 
    
    # 内存 超售：基于 Working Set 预测
    memory:
      enable: true
      overSellRatio: 1.2
      # 触发回收阈值
      evictionThreshold: "85%"
    
    # GPU/NPU 显存：严禁超售 (OOM 即崩溃)，但允许算力时间片共享
    gpuMemory:
      enable: false
    gpuCore:
      enable: true
      # MPS 共享模式下的时间片权重
      timeSliceWeight: 
        online: 70
        batch: 30

// pkg/qos/guardian.go - 在线业务 SLA 守护进程
func (g *QoSGuardian) Run(ctx context.Context) {
    ticker := time.NewTicker(10 * time.Second)
    for {
        select {
        case <-ticker.C:
            g.checkAndEnforce()
        case <-ctx.Done():
            return
        }
    }
}

func (g *QoSGuardian) checkAndEnforce() {
    onlinePods := g.lister.Pods("").List(labels.SelectorFromSet(labels.Set{"tier": "online"}))
    for _, pod := range onlinePods {
        // 1. 采集实时指标 (从 Prometheus/CAdvisor/Device Plugin)
        metrics := g.metricsFetcher.GetPodMetrics(pod)
        
        // 2. 判定 SLA 违规
        if metrics.LatencyP99 > pod.Annotations["sla/latency-p99-ms"] {
            g.logger.Warn("SLA breach detected", "pod", pod.Name, "latency", metrics.LatencyP99)
            
            // 3. 熔断动作：驱逐同节点 BestEffort/Batch Pod
            victims := g.selectVictims(pod.Spec.NodeName, []string{"batch", "besteffort"})
            for _, v := range victims {
                g.evictor.Evict(v, "QoSGuardian:OnlineSLAbreach")
                g.recorder.Eventf(pod, "Warning", "PreemptedForQoS", "Preempted %s to protect %s", v.Name, pod.Name)
            }
            
            // 4. 触发扩容/调度器重调度
            g.scaler.TriggerScaleUp(pod.Labels["app"])
        }
    }
}