┌─────────────────────────────────────────────────────────────┐
│                    统一控制平面 (Control Plane)              │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │ 会议调度器   │  │ 媒体节点    │  │ 资源画像    │          │
│  │ (Meeting    │  │ 控制器      │  │ 服务        │          │
│  │  Scheduler) │  │ (MediaNode  │  │ (Resource   │          │
│  │             │  │  Controller)│  │  Profiler) │          │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘          │
└─────────┼────────────────┼────────────────┼──────────────────┘
          │                │                │
          ▼                ▼                ▼
┌─────────────────────────────────────────────────────────────┐
│                      数据面                                   │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐          │
│  │  媒体节点池  │  │  网络拓扑   │  │  硬件加速   │          │
│  │ (Media Node │  │  感知层     │  │  资源池     │          │
│  │  Pool)      │  │ (Topology   │  │ (GPU/NPU,   │          │
│  │             │  │  Awareness) │  │  SR-IOV)    │          │
│  └─────────────┘  └─────────────┘  └─────────────┘          │
└─────────────────────────────────────────────────────────────┘

# 伪代码：基于时间序列的预测扩容逻辑
def predictive_scale(meeting_schedule: List[Meeting], 
                     historical_load: TimeSeries) -> ScalePlan:
    # 1. 聚合未来 30 分钟预期并发会议数
    predicted_concurrent = sum(
        m.expected_participants * m.join_probability 
        for m in meeting_schedule 
        if m.start_time <= now + 30min
    )
    
    # 2. 引入历史峰值修正系数（考虑突发入会）
    correction_factor = historical_load.peak_ratio(last_7d)
    target_capacity = predicted_concurrent * correction_factor * 1.2  # 20% 缓冲
    
    # 3. 换算所需媒体节点数（单节点承载上限配置化）
    required_nodes = ceil(target_capacity / NODE_MAX_CONCURRENT)
    
    # 4. 生成扩容计划，下发至 MediaNodePool
    return ScalePlan(target_replicas=required_nodes, 
                     priority="predictive",
                     ttl=45min)

# MediaNodePool HPA 配置示例
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: media-node-hpa
spec:
  scaleTargetRef:
    apiVersion: media.example.com/v1alpha1
    kind: MediaNodePool
    name: media-node-pool
  minReplicas: 10
  maxReplicas: 500
  metrics:
  - type: Pods
    pods:
      metric:
        name: media_node_load_score  # 综合负载评分 0-100
      target:
        type: AverageValue
        averageValue: "65"  # 目标负载 65%
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 30
      policies:
      - type: Percent
        value: 50
        periodSeconds: 30
      - type: Pods
        value: 20
        periodSeconds: 30
      selectPolicy: Max
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容冷却 5 分钟防抖
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60

Load_Score = 0.4 * CPU_Util + 0.25 * Memory_Util + 0.2 * Bandwidth_Util 
           + 0.1 * Encode_Queue_Depth + 0.05 * Packet_Loss_Rate

// 核心过滤逻辑
func (p *MediaSchedulerPlugin) Filter(ctx context.Context, 
    state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
    
    // 1. 硬件加速能力匹配（GPU/NPU/VPU）
    if !p.matchHardwareAcceleration(pod, nodeInfo) {
        return framework.NewStatus(framework.Unschedulable, 
            "insufficient hardware acceleration")
    }
    
    // 2. 网络拓扑约束：同可用区优先，跨交换机惩罚
    if !p.checkTopologyConstraint(pod, nodeInfo) {
        return framework.NewStatus(framework.Unschedulable, 
            "topology constraint violated")
    }
    
    // 3. 端口资源预留（媒体节点需大量 UDP 端口）
    if !p.checkPortAvailability(nodeInfo, REQUIRED_PORT_RANGE) {
        return framework.NewStatus(framework.Unschedulable, 
            "insufficient UDP port range")
    }
    
    return framework.NewStatus(framework.Success, "")
}

Final_Score = Σ(Weight_i * Normalize(Score_i)) + Random_Tiebreaker(0-1)

# 会议 CRD 中的调度策略定义
apiVersion: meeting.example.com/v1alpha1
kind: Meeting
metadata:
  name: quarterly-all-hands
spec:
  schedulingPolicy:
    # 强制反亲和：同会议媒体节点分散至不同可用区
    antiAffinity:
      topologyKey: topology.kubernetes.io/zone
      required: true
    # 专属节点池：绑定高性能 GPU 节点池
    nodePoolSelector:
      matchLabels:
        media-tier: "premium-gpu"
    # 资源保障：预留带宽、编解码槽位
    resourceReservation:
      bandwidthGbps: 10
      encodeSlots: 200
    # 容灾策略：主备节点 1:1 热备
    haPolicy:
      mode: "active-standby"
      standbyReplicas: 1

# SriovNetworkNodePolicy 示例
apiVersion: sriovnetwork.openshift.io/v1
kind: SriovNetworkNodePolicy
metadata:
  name: media-sriov-policy
spec:
  deviceType: netdevice
  nicSelector:
    vendor: "8086"
    deviceID: "158b"  # Intel E810
    pfNames: ["ens1f0", "ens1f1"]
  numVfs: 16
  priority: 10
  resourceName: "intel.com/media_sriov"
  mtu: 9000

resources:
  limits:
    intel.com/media_sriov: "2"
    nvidia.com/gpu: "1"
  requests:
    intel.com/media_sriov: "2"
    nvidia.com/gpu: "1"

// 优雅缩容控制器核心逻辑
func (c *GracefulScaleDownController) drainNode(ctx context.Context, nodeName string) error {
    // 1. 标记节点不可调度
    c.cordonNode(nodeName)
    
    // 2. 等待现有会话自然结束 或 主动迁移
    deadline := time.Now().Add(MAX_DRAIN_DURATION) // 10 分钟
    
    for time.Now().Before(deadline) {
        sessions := c.getActiveSessions(nodeName)
        if len(sessions) == 0 {
            break
        }
        
        // 仅迁移长会话（>30分钟），短会话等待自然结束
        for _, s := range sessions {
            if s.Duration > 30*time.Minute {
                c.migrateSession(ctx, s, c.selectTargetNode(s))
            }
        }
        time.Sleep(30 * time.Second)
    }
    
    // 3. 强制终止残留会话（极少数），触发客户端重连
    c.forceTerminateRemaining(nodeName)
    
    // 4. 删除节点
    return c.deleteNode(nodeName)
}

graph TD
    A[入会失败率飙升] --> B{关联分析}
    B --> C[新节点就绪延迟高]
    B --> D[某可用区丢包率异常]
    B --> E[GPU 显存泄漏导致编码失败]
    C --> F[镜像拉取超时/预热池不足]
    D --> G[交换机端口拥塞/光模块故障]
    E --> H[媒体引擎版本回归Bug]

graph TB
    subgraph Control_Plane[控制平面 共享]
        API[K8s APIServer]
        SVC[Service Mesh Control]
    end
    
    subgraph Tenant_A[租户 A 网络平面]
        GW_A[Ingress Gateway A]
        NS_A[Namespace: tenant-a]
        CNI_A[Cilium Network Policy A]
        MEDIA_A[Media Node Pool A]
    end
    
    subgraph Tenant_B[租户 B 网络平面]
        GW_B[Ingress Gateway B]
        NS_B[Namespace: tenant-b]
        CNI_B[Cilium Network Policy B]
        MEDIA_B[Media Node Pool B]
    end
    
    subgraph Shared_Infra[共享基础设施]
        LB[Cloud Load Balancer]
        SRV[SR-IOV PF / Switch]
    end
    
    LB --> GW_A & GW_B
    GW_A --> MEDIA_A
    GW_B --> MEDIA_B
    SRV -.->|VF 直通| MEDIA_A
    SRV -.->|VF 直通| MEDIA_B

# 审计代理 Sidecar 注入策略
apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
  name: media-audit-injector
webhooks:
- name: audit.media.example.com
  rules:
  - operations: ["CREATE"]
    apiGroups: [""]
    apiVersions: ["v1"]
    resources: ["pods"]
    scope: "Namespaced"
  clientConfig:
    service:
      name: audit-injector-svc
      namespace: media-system
      path: /mutate
  namespaceSelector:
    matchExpressions:
    - key: media.audit/enabled
      operator: In
      values: ["true"]

┌────────────────────────────────────────────────────────────┐
│              Federation Control Plane (Karmada)             │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────────┐  │
│  │ Federation   │  │ Multi-Cluster│  │ Global Resource  │  │
│  │ Scheduler    │  │ Service      │  │ Interpreter      │  │
│  │ (Media Aware)│  │ Discovery    │  │ (MediaNodePool)  │  │
│  └──────┬───────┘  └──────┬───────┘  └────────┬─────────┘  │
└─────────┼─────────────────┼────────────────────┼────────────┘
          │                 │                    │
    ┌─────▼─────┐     ┌─────▼─────┐        ┌────▼────┐
    │ Cluster 1 │     │ Cluster 2 │  ...   │ Cluster N│
    │ (IDC)     │     │ (AWS)     │        │ (Azure)  │
    │ Member    │     │ Member    │        │ Member   │
    └───────────┘     └───────────┘        └──────────┘

// 联邦调度评分扩展
func (f *FederationMediaScheduler) ScoreCluster(ctx context.Context, 
    meeting *Meeting, clusters []*ClusterInfo) (map[string]int64, error) {
    
    scores := make(map[string]int64)
    for _, c := range clusters {
        var score int64
        
        // 1. 网络质量权重 35%：基于历史探测数据（延迟、丢包、带宽）
        score += int64(35 * (1 - c.NetworkQualityIndex)) // 0-1, 越小越好
        
        // 2. 成本权重 25%：实时 Spot/预留实例价格 + 传输费用
        score += int64(25 * (1 - c.NormalizedCost))
        
        // 3. 合规权重 20%：数据驻留要求（GDPR、数据不出境）
        if meeting.Compliance.RegionLock != "" && !c.MatchRegion(meeting.Compliance.RegionLock) {
            score = -1000 // 硬性过滤
            continue
        }
        score += 20
        
        // 4. 容量权重 15%：可分配媒体节点数 / 会议需求
        score += int64(15 * c.AvailableCapacityRatio)
        
        // 5. 碳强度权重 5%：绿色调度（见第五章）
        score += int64(5 * (1 - c.CarbonIntensityIndex))
        
        scores[c.Name] = score
    }
    return scores, nil
}

# 多集群媒体服务导出
apiVersion: policy.karmada.io/v1alpha1
kind: ClusterPropagationPolicy
metadata:
  name: media-node-svc-propagation
spec:
  resourceSelectors:
  - apiVersion: v1
    kind: Service
    name: media-node-svc
    namespace: media-system
  placement:
    clusterAffinity:
      clusterNames: ["idc-shanghai", "aws-singapore", "azure-tokyo"]
---
# 全局 VIP 通过云厂商 Global Accelerator (GA) 实现就近接入
# 客户端解析 media.global.example.com -> GA Anycast IP -> 就近集群 Ingress

# Dockerfile 关键配置
# 1. 启用 HugePages (2MB/1GB)
# 2. 设置 GOMEMLIMIT 留足系统内存
# 3. 预分配内存池
ENV GOMEMLIMIT=12GiB 
    GOGC=50 
    GODEBUG=madvdontneed=1

# 启动命令
CMD ["/media-engine", 
     "--hugepage-dir=/mnt/hugepages", 
     "--packet-pool-size=500000", 
     "--frame-pool-size=100000"]

resources:
  limits:
    memory: "16Gi"
    hugepages-2Mi: "4Gi"   # 预留 2MB 大页
    hugepages-1Gi: "2Gi"   # 预留 1GB 大页 (需节点预留)
  requests:
    memory: "16Gi"
    hugepages-2Mi: "4Gi"

传统路径: NIC -> Kernel (skb) -> Userspace (App) -> Kernel -> NIC
XDP 路径: NIC -> XDP Program (eBPF) -> AF_XDP Socket -> Userspace (App) -> AF_XDP -> NIC

// xdp_media_redirect.bpf.c
SEC("xdp")
int xdp_media_redirect(struct xdp_md *ctx) {
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;
    
    // 1. 解析 UDP + RTP 头，提取 SSRC
    struct ethhdr *eth = data;
    if (data + sizeof(*eth) > data_end) return XDP_PASS;
    // ... IPv4/UDP/RTP 解析 ...
    
    // 2. 查找 SSRC -> 队列映射 (BPF Map: LRU_HASH)
    __u32 queue_id = 0;
    struct ssrc_map_val *val = bpf_map_lookup_elem(&ssrc_queue_map, &ssrc);
    if (val) queue_id = val->queue_id;
    
    // 3. 重定向至 AF_XDP Socket (零拷贝)
    return bpf_redirect_map(&xsks_map, queue_id, XDP_DROP);
}

# 统一中断信号采集器
class SpotInterruptionWatcher:
    def __init__(self):
        self.providers = {
            "aws": AWSSpotWatcher(),      # 监听 Instance Metadata Service / EventBridge
            "azure": AzureSpotWatcher(),  # 监听 Scheduled Events
            "aliyun": AliyunSpotWatcher(),# 监听实例元数据
            "gcp": GCPSpotWatcher()       # 监听 Metadata Server
        }
    
    async def watch(self, node_name: str, provider: str) -> InterruptionSignal:
        # 标准化输出：提前 2 分钟 / 30 秒 / 即将中断
        signal = await self.providers[provider].get_signal(node_name)
        return InterruptionSignal(
            node=node_name,
            severity=signal.severity,  # WARNING(2min) / CRITICAL(30s) / TERMINATING
            deadline=signal.deadline,
            reason=signal.reason
        )

func (m *SessionMigrator) MigrateOnSpotInterruption(ctx context.Context, signal InterruptionSignal) {
    sessions := m.getSessionsOnNode(signal.Node)
    
    // 按会话价值排序：付费会议 > 大型会议 > 普通会议
    sort.Slice(sessions, func(i, j int) bool {
        return sessions[i].Priority > sessions[j].Priority
    })
    
    for _, s := range sessions {
        // 选目标节点：同 AZ 优先、GPU 亲和、负载最低
        target := m.scheduler.SelectTargetNode(s, signal.Node)
        
        // 异步发起 SDP 重协商
        go m.initiateHandover(ctx, s, target, signal.Severity)
        
        // 限流：每秒最多迁移 50 个会话，避免雪崩
        m.rateLimiter.Wait(ctx)
    }
}

# 节点碳标签 (由外部碳数据服务定时注入)
apiVersion: v1
kind: Node
metadata:
  name: gpu-node-001
  labels:
    carbon.intensity.gco2eq/kwh: "320"  # 当前电网碳强度 gCO2eq/kWh
    carbon.region: "cn-shanghai"
    carbon.source: "grid-realtime-api"
  annotations:
    carbon.last-updated: "2024-01-15T10:30:00Z"

// 碳感知评分插件
func (c *CarbonAwarePlugin) Score(ctx context.Context, state *framework.CycleState, 
    pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    
    nodeInfo, _ := c.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    carbonLabel := nodeInfo.Node().Labels["carbon.intensity.gco2eq/kwh"]
    carbonIntensity, _ := strconv.Atoi(carbonLabel)
    
    // 归一化：全球电网碳强度范围 50-800 gCO2eq/kWh
    // 评分越高 = 碳排越低 = 越优先调度
    normalizedScore := 100 - int64((carbonIntensity-50)*100/(800-50))
    
    // 业务优先级加权：免费会议权重 1.0，付费会议权重 0.3 (体验优先)
    priorityWeight := getMeetingPriorityWeight(pod)
    
    finalScore := normalizedScore * priorityWeight
    return finalScore, nil
}

# 碳感知预测性扩容示例
def carbon_aware_predictive_scale(meeting_forecast: List[Meeting]) -> ScalePlan:
    # 1. 获取未来 24h 各区域碳强度预测曲线
    carbon_forecast = carbon_api.get_forecast(regions=["cn-shanghai", "us-west", "eu-central"], hours=24)
    
    # 2. 识别 "绿色窗口期" (碳强度 < 200 gCO2eq/kWh)
    green_windows = find_green_windows(carbon_forecast)
    
    # 3. 可延迟任务 (录制转码、字幕生成、归档) 调度至绿色窗口
    for task in delayable_tasks:
        best_window = min(green_windows, key=lambda w: w.carbon_intensity)
        task.schedule_at = best_window.start_time
        task.target_region = best_window.region
    
    # 4. 实时会议：优先调度至当前最低碳区域 (满足延迟 < 100ms 前提)
    for meeting in realtime_meetings:
        meeting.preferred_region = select_lowest_carbon_region(meeting.latency_budget)
    
    return generate_scale_plan(realtime_meetings, delayable_tasks)

┌─────────────────────────────────────────────────────────────────┐
│                    CI/CD Pipeline                                │
├─────────────┬─────────────┬─────────────┬─────────────────────┤
│  Layer 1    │  Layer 2    │  Layer 3    │  Layer 4            │
│  单元/集成  │  压测/混沌  │  金丝雀     │  全量发布            │
│  测试       │  验证       │  发布       │  (蓝绿/滚动)         │
├─────────────┼─────────────┼─────────────┼─────────────────────┤
│ - 代码扫描  │ - 10k 并发  │ - 1% 节点   │ - 分批次 10%/30%/60%│
│ - 模糊测试  │ - 弱网模拟  │ - 影子流量  │ - 关键指标自动回滚  │
│ - 协议一致性│ - 故障注入  │ - 核心租户  │ - 事后复盘报告      │
└─────────────┴─────────────┴─────────────┴─────────────────────┘

# VirtualService 影子流量配置
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: media-node-vs
spec:
  hosts:
  - media-node-svc.media-system.svc.cluster.local
  http:
  - route:
    - destination:
        host: media-node-svc
        subset: stable
      weight: 100
    mirror:
      host: media-node-svc
      subset: canary
    mirrorPercentage:
      value: 1.0

# Argo Rollouts AnalysisTemplate
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: media-node-canary-analysis
spec:
  args:
  - name: canary-hash
  metrics:
  - name: join-success-rate
    interval: 30s
    count: 20
    successCondition: result[0] >= 0.995
    failureLimit: 3
    provider:
      prometheus:
        address: http://prometheus.monitoring.svc
        query: |
          sum(rate(meeting_join_success_total{version="{{args.canary-hash}}"}[1m])) 
          / 
          sum(rate(meeting_join_total{version="{{args.canary-hash}}"}[1m]))
  - name: p99_latency
    successCondition: result[0] <= 300
    provider:
      prometheus:
        query: histogram_quantile(0.99, rate(media_signaling_latency_bucket{version="{{args.canary-hash}}"}[1m]))

// client_telemetry.proto
message NetworkQualityReport {
  string meeting_id = 1;
  string user_id = 2;
  string client_ip = 3;
  int64 timestamp_ms = 4;
  
  // 网络指标
  double rtt_ms = 5;
  double jitter_ms = 6;
  double packet_loss_ratio = 7;      // 0.0 - 1.0
  double available_bandwidth_kbps = 8;
  
  // 体验指标
  double mos_score = 9;              // 1.0 - 5.0
  int32 freeze_count = 10;           // 卡顿次数
  int32 total_freeze_duration_ms = 11;
  
  // 设备能力
  string device_model = 12;
  string cpu_arch = 13;
  bool hardware_decoder_support = 14;
}

sequenceDiagram
    participant Client as 客户端 SDK
    participant Gateway as 信令网关
    participant Scheduler as 会议调度器
    participant MediaNode as 媒体节点
    
    Client->>Gateway: 入会请求 + 设备指纹 + 预测带宽
    Gateway->>Scheduler: 调度决策请求 (含客户端画像)
    Scheduler->>Scheduler: 结合节点负载、网络拓扑、客户端位置计算最优节点
    Scheduler-->>Gateway: 目标媒体节点 IP + Token
    Gateway-->>Client: 媒体节点连接信息
    Client->>MediaNode: ICE/DTLS 建连 + 媒体流协商
    
    loop 每 5 秒
        Client->>Gateway: NetworkQualityReport (gRPC 流)
        Gateway->>Scheduler: 实时网络质量流
        Scheduler->>Scheduler: 评分模型更新节点权重
        alt 检测到持续弱网 (丢包>10% 或 RTT>300ms)
            Scheduler->>MediaNode: 触发降码率/切换编码器/迁移建议
            MediaNode-->>Client: REMB / PLI / FIR / SDP Re-negotiation
        end
    end