# 外部入口：云厂商LB或MetalLB，TCP/UDP透传
apiVersion: v1
kind: Service
metadata:
  name: media-gateway-external
  annotations:
    service.beta.kubernetes.io/aws-load-balancer-type: "nlb"
    service.beta.kubernetes.io/aws-load-balancer-cross-zone-load-balancing-enabled: "true"
spec:
  type: LoadBalancer
  ports:
  - name: signal-ws
    port: 443
    targetPort: 8443
    protocol: TCP
  - name: media-udp
    port: 10000
    targetPort: 10000
    protocol: UDP
  selector:
    app: media-gateway

---
# 内部媒体节点发现：Headless Service，配合EndpointSlice实现就绪感知
apiVersion: v1
kind: Service
metadata:
  name: media-node-internal
spec:
  clusterIP: None
  ports:
  - name: media
    port: 50000
    protocol: UDP
  selector:
    app: media-node

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: media-node-hpa
  namespace: rtc-prod
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: media-node
  minReplicas: 6          # 兜底最小副本数，覆盖日常低峰
  maxReplicas: 200        # 单集群上限，受限于Node数量与IP池
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 600   # 缩容冷却10分钟，防抖
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60
      - type: Pods
        value: 2
        periodSeconds: 60
      selectPolicy: Min
    scaleUp:
      stabilizationWindowSeconds: 30    # 扩容响应30秒
      policies:
      - type: Percent
        value: 50
        periodSeconds: 15
      - type: Pods
        value: 10
        periodSeconds: 15
      selectPolicy: Max
  metrics:
  - type: Pods
    pods:
      metric:
        name: media_node_active_sessions
      target:
        type: AverageValue
        averageValue: "600"   # 单Pod目标会议数
  - type: Pods
    pods:
      metric:
        name: media_node_bandwidth_usage_mbps
      target:
        type: AverageValue
        averageValue: "600Mi" # 单Pod目标带宽
  - type: External
    external:
      metric:
        name: signaling_gateway_queue_depth
        selector:
          matchLabels:
            queue: session_establish
      target:
        type: Value
        value: "50"

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: media-node-predictive-scaler
  namespace: rtc-prod
spec:
  scaleTargetRef:
    name: media-node
  pollingInterval: 60
  cooldownPeriod: 600
  minReplicaCount: 6
  maxReplicaCount: 200
  advanced:
    predictiveScaling:
      mode: "PredictAndScale"
      algorithm: "prophet"
      historyLength: "30d"
      seasonality:
        - name: "daily"
          period: "24h"
        - name: "weekly"
          period: "168h"

# 节点组标签与污点，专属承载媒体节点
apiVersion: v1
kind: Node
metadata:
  labels:
    workload-type: media-node
    node-pool: media-pool-ondemand
  annotations:
    cluster-autoscaler.kubernetes.io/scale-down-disabled: "false"
spec:
  taints:
  - key: "dedicated"
    value: "media-node"
    effect: "NoSchedule"

func (s *MediaServer) ReadyCheck() error {
    // 1. 基础资源阈值
    if s.stats.CPUUsagePercent > 85 || s.stats.MemUsagePercent > 80 {
        return fmt.Errorf("resource saturation: cpu=%.1f%%, mem=%.1f%%", 
            s.stats.CPUUsagePercent, s.stats.MemUsagePercent)
    }
    // 2. 关键依赖连通性
    if !s.redisPool.IsHealthy() || !s.signalGWClient.IsHealthy() {
        return errors.New("downstream dependency unhealthy")
    }
    // 3. 业务能力自检：能否分配新端口/会话
    if !s.portAllocator.HasAvailable(100) || s.sessionCount >= s.maxSessions {
        return errors.New("capacity exhausted")
    }
    // 4. 网络质量自检：近期丢包率
    if s.stats.RecentPacketLossRate > 0.02 {
        return fmt.Errorf("high packet loss: %.2f%%", s.stats.RecentPacketLossRate*100)
    }
    return nil
}

# Deployment 关键配置
spec:
  template:
    spec:
      terminationGracePeriodSeconds: 120  # 留足2分钟迁移窗口
      preStop:
        exec:
          command: ["/bin/sh", "-c", "/usr/local/bin/graceful-shutdown.sh"]

#!/bin/bash
set -e

# 1. 标记Pod为"Draining"，停止接收新会话调度
curl -X POST localhost:8080/admin/drain

# 2. 等待现有会话自然结束或主动迁移（配合信令网关）
#    信令网关收到Draining事件后，对新加入用户调度至其他节点；
#    对在会用户发送RE-INVITE/UPDATE引导媒体流切换至目标节点
timeout 90s /usr/local/bin/wait-sessions-drain --max-wait=90 --check-interval=5

# 3. 强制清理残留资源（端口、共享内存、FD）
/usr/local/bin/force-cleanup

# 4. 通知服务注册中心下线
curl -X DELETE "http://service-registry:8080/v1/instances/$(hostname)"

exit 0

# NPD 自定义检查器配置
apiVersion: v1
kind: ConfigMap
metadata:
  name: npd-config
  namespace: kube-system
data:
  custom-plugin-monitor.json: |
    {
      "plugins": [
        {
          "name": "media-node-health",
          "type": "exec",
          "interval": "30s",
          "command": "/usr/local/bin/media-node-health-check.sh",
          "source": "node-problem-detector"
        }
      ]
    }

#!/bin/bash
# 返回非0即上报Node Condition: MediaNodeDegraded

# 1. 网卡队列丢包检测
for iface in eth0 eth1; do
    rx_drop=$(ethtool -S $iface | grep -E 'rx_missed|rx_fifo' | awk '{sum+=$2} END {print sum}')
    if [ "$rx_drop" -gt 1000 ]; then
        echo "NIC $iface high packet drops: $rx_drop"
        exit 1
    fi
done

# 2. 系统负载与内核异常
load1=$(cut -d' ' -f1 /proc/loadavg)
cpu_cores=$(nproc)
if (( $(echo "$load1 > $cpu_cores * 2" | bc -l) )); then
    echo "System load critical: $load1"
    exit 1
fi

# 3. 关键内核参数漂移检测
for param in net.core.somaxconn net.ipv4.udp_mem; do
    current=$(sysctl -n $param)
    expected=$(cat /etc/sysctl.d/99-media-node.conf | grep $param | cut -d= -f2)
    if [ "$current" != "$expected" ]; then
        echo "Kernel param drift: $param=$current (expected $expected)"
        exit 1
    fi
done

exit 0

// 熔断器状态机：Closed -> HalfOpen -> Open
type CircuitBreaker struct {
    failureThreshold int           // 连续失败次数阈值
    successThreshold int           // 半开状态成功次数阈值
    timeout          time.Duration // 开启状态持续时间
    // ...
}

func (cb *CircuitBreaker) Execute(req Request) (Response, error) {
    switch cb.state {
    case Open:
        if time.Since(cb.lastFailure) > cb.timeout {
            cb.toHalfOpen()
        } else {
            return nil, ErrCircuitOpen // 快速失败，保护下游
        }
    case HalfOpen:
        resp, err := cb.tryRequest(req)
        if err == nil {
            cb.onSuccess()
        } else {
            cb.onFailure()
        }
        return resp, err
    case Closed:
        return cb.tryRequest(req)
    }
}

用户APP -> 信令网关 -> 媒体节点A -> 媒体节点B -> 录制服务
    |           |            |            |            |
  Span       Span         Span         Span         Span
(trace-id)  (trace-id)   (trace-id)   (trace-id)   (trace-id)

span.SetAttributes(
    attribute.String("rtc.session_id", sessionID),
    attribute.String("rtc.user_id", userID),
    attribute.String("rtc.media_node", podName),
    attribute.Int("rtc.bitrate_kbps", bitrate),
    attribute.Float64("rtc.packet_loss", lossRate),
    attribute.String("rtc.codec", "H264/VP8/OPUS"),
)

graph TD
    User[用户终端] --> DNS[智能DNS/HTTPDNS]
    DNS -->|解析最优入口IP| GW[区域网关集群]
    GW -->|信令交互| Signal[信令网关]
    Signal -->|调度决策| Scheduler[全局调度器]
    Scheduler -->|查询多集群健康度| MC[Multi-Cluster Controller]
    MC -->|聚合指标| Prom[Thanos/Cortex 全局监控]
    Scheduler -->|返回媒体节点IP| Signal
    Signal --> User

func ScoreCluster(cluster *ClusterStatus, req *ScheduleRequest) float64 {
    // 权重可通过 ConfigMap 热更新
    w := Weights{Latency: 0.4, Load: 0.3, Health: 0.2, Cost: 0.1}
    
    latencyScore := 1.0 - math.Min(cluster.AvgRTTMs/200.0, 1.0) // 200ms 归一化
    loadScore := 1.0 - cluster.ResourceUtilization               // 资源利用率越低越好
    healthScore := cluster.HealthyNodeRatio                      // 就绪节点占比
    costScore := 1.0 - cluster.UnitCostRatio                     // 单位成本（Spot/预留实例优先）
    
    return w.Latency*latencyScore + w.Load*loadScore + w.Health*healthScore + w.Cost*costScore
}

# 边缘节点自治配置：云端下发 ConfigMap，边缘缓存至本地 SQLite/Etcd
apiVersion: apps.edge.kubernetes.io/v1alpha1
kind: UnitedDeployment
metadata:
  name: edge-media-node
spec:
  topology:
    pools:
    - name: pop-shanghai
      nodeSelectorTerm:
        matchExpressions:
        - key: edge-type
          operator: In
          values: ["pop-media"]
      replicas: 3
      # 关键：定义本地自治策略
      autonomy:
        enabled: true
        configSyncMode: "Incremental" # 增量同步，弱网友好
        maxOfflineDuration: "24h"     # 离线 24h 仍维持业务运行

// XDP 程序片段：早期丢包检测与 FEC 解码卸载
SEC("xdp_media_fec")
int xdp_fec_decode(struct xdp_md *ctx) {
    void *data = (void *)(long)ctx->data;
    struct rtp_hdr *rtp = data;
    // 1. 识别媒体流 (UDP + 特定端口范围 + RTP 版本号)
    // 2. 检查序列号间隙 -> 触发 FEC 恢复逻辑 (Reed-Solomon / RaptorQ)
    // 3. 恢复成功修改 seq_num 重新注入协议栈，应用层无感知
    // 4. 统计丢包率、恢复率上报用户态 Ring Buffer
    return XDP_PASS;
}

# 媒体节点 Pod 请求 NPU 资源跑降噪模型
apiVersion: v1
kind: Pod
metadata:
  name: media-node-ai
  annotations:
    # 指定模型版本，支持灰度发布
    ai.model.version: "denoise-v2.1.0"
spec:
  containers:
  - name: media-server
    resources:
      limits:
        cpu: "16"
        memory: "32Gi"
        cambricon.com/mlu: "1"     # 寒武纪 NPU
        # nvidia.com/gpu: "1"      # 或 NVIDIA GPU
    env:
    - name: MODEL_PATH
      value: "/models/denoise/$(AI_MODEL_VERSION)"
    volumeMounts:
    - name: model-store
      mountPath: /models
  volumes:
  - name: model-store
    persistentVolumeClaim:
      claimName: ai-model-pvc  # 模型由 InitContainer 从模型仓库拉取

graph LR
    A[线上流量采样] --> B[数据标注平台]
    B --> C[训练流水线 Kubeflow/MLflow]
    C --> D[模型评估/回归测试]
    D --> E[模型仓库 Harbor/MLflow]
    E --> F[金丝雀发布 Argo Rollouts]
    F --> G[影子流量验证]
    G --> H[全量推送]
    H --> A

# 使用 CSI Secrets Store Driver + Vault/KMS Provider，避免 Secret 落盘
apiVersion: secrets-store.csi.x-k8s.io/v1
kind: SecretProviderClass
metadata:
  name: media-tls-certs
spec:
  provider: vault
  parameters:
    vaultAddress: "https://vault.internal"
    roleName: "media-node-role"
    objects: |
      - objectName: "tls/serving-cert"
        objectType: "secret"
        objectVersion: ""
    # 自动轮转同步
    rotationPollInterval: "300s"

# CiliumNetworkPolicy: 仅允许信令网关访问媒体节点信令端口
apiVersion: cilium.io/v2
kind: CiliumNetworkPolicy
metadata:
  name: media-node-ingress
spec:
  endpointSelector:
    matchLabels:
      app: media-node
  ingress:
  - fromEndpoints:
    - matchLabels:
        app: signaling-gateway
    toPorts:
    - ports:
      - port: "8443"
        protocol: TCP
      rules:
        http:
        - method: "POST"
          path: "/api/v1/sessions/*"
  - fromEndpoints:
    - matchLabels:
        app: media-node  # 节点间媒体流转发
    toPorts:
    - ports:
      - port: "50000-60000"
        protocol: UDP

# NodePool 配置：基础容量用预留实例/节省计划，弹性容量用 Spot/抢占式实例
apiVersion: karpenter.sh/v1beta1
kind: NodePool
metadata:
  name: media-node-mixed
spec:
  template:
    spec:
      requirements:
      - key: "karpenter.k8s.aws/instance-category"
        operator: In
        values: ["c", "m", "r"]  # 计算/内存/高性能计算型
      - key: "karpenter.k8s.aws/instance-generation"
        operator: In
        values: ["2", "3"]       # 最新两代 CPU
      # 关键：容忍 Spot 中断，配合 PodDisruptionBudget 优雅驱逐
      tolerations:
      - key: "spot-instance"
        operator: "Exists"
        effect: "NoSchedule"
  disruption:
    consolidationPolicy: WhenEmptyOrUnderutilized
    consolidateAfter: 30s
    # Spot 实例收到中断通知 2 分钟前触发驱逐
    expirationSeconds: 7200 

apiVersion: autoscaling.k8s.io/v1
kind: VerticalPodAutoscaler
metadata:
  name: media-node-vpa
spec:
  targetRef:
    apiVersion: "apps/v1"
    kind: Deployment
    name: media-node
  updatePolicy:
    updateMode: "Auto" # 或 "Recreate" 配合滚动更新
  resourcePolicy:
    containerPolicies:
    - containerName: media-server
      # 设置上下界，防止 VPA 推荐值过大/过小
      minAllowed:
        cpu: "2"
        memory: "4Gi"
      maxAllowed:
        cpu: "16"
        memory: "32Gi"
      # 关键：基于 P99 使用量 + 20% 安全边际推荐，而非 Peak
      controlledResources: ["cpu", "memory"]
      controlledValues: RequestsOnly # 仅调整 Request，Limit 由 QoS 策略固定