[客户端] <--UDP/TCP/TLS--> [接入网关/SBC] <--gRPC/HTTP2--> [信令集群]
                                                      |
                                                      v (内部RPC/媒体流)
                                              [媒体节点集群: SFU/MCU/转码/录制]
                                                      |
                                                      v
                                              [存储/分析/分发下游]

// 媒体节点进程内嵌入的轻量 Agent 伪代码
type MediaSidecarAgent struct {
    metricsPushGateway *prometheus.Pusher
    xdsClient          *xds.Client
    localNodeID        string
}

func (a *MediaSidecarAgent) Start() {
    // 1. 向 Sidecar 注册本地媒体端口映射（供 Sidecar 健康检查/服务发现）
    a.registerMediaEndpoints()
    
    // 2. 启动指标推送协程（周期 5s）
    go a.pushMediaMetricsLoop()
    
    // 3. 监听 XDS 下发的治理策略（熔断阈值、权重调整）
    go a.watchTrafficPolicy()
}

func (a *MediaSidecarAgent) pushMediaMetricsLoop() {
    for {
        metrics := collectLocalMediaMetrics() // 从共享内存/环形缓冲区读取
        a.metricsPushGateway.Push(metrics)
        time.Sleep(5 * time.Second)
    }
}

+------------------+     +------------------+     +------------------+
|  日志            |     |  指标            |     |  链路追踪        |
|  (Loki/ES)       |     |  (Prometheus/    |     |  (Tempo/Jaeger)  |
|                  |     |   Thanos)        |     |                  |
| - 结构化 JSON    |     | - RED/USE 模型   |     | - W3C TraceContext|
| - 关联 trace_id  |     | - 多维标签       |     | - 媒体流 Span 关联|
+--------+---------+     +--------+---------+     +--------+---------+
         |                        |                        |
         v                        v                        v
+--------+------------------------------------------------+---------+
|                    统一查询与告警平台 (Grafana/自研)                    |
+----------------------------------------------------------------------+

// 信令下发示例
message MediaPlaneConfig {
  string media_trace_id = 1;  // 格式: trace-id-span-id (W3C 兼容)
  repeated MediaNode media_nodes = 2;
}

// SDK 侧 RTP Header Extension 定义 (1 byte header + 4 bytes trace_id)
#define RTP_EXT_TRACE_ID 0xBEDE

# VirtualService 扩展示例 (ISTIO/Envoy 自定义 Filter)
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: sfu-media-routing
spec:
  hosts:
  - sfu.media.svc.cluster.local
  http:
  - match:
    - headers:
        x-media-codec:
          exact: "H264"
    - headers:
        x-media-resolution:
          regex: "^(1080|720)p$"
    route:
    - destination:
        host: sfu-h264-high.media.svc.cluster.local
        subset: high-perf  # 对应高性能节点池
      weight: 80
    - destination:
        host: sfu-h264-high.media.svc.cluster.local
        subset: standard
      weight: 20
    fault:
      delay:
        percentage:
          value: 0.1
        fixedDelay: 50ms  # 混沌工程注入

type MediaCircuitBreaker struct {
    sendQueue      *ringbuffer.RingBuffer
    bandwidthMeter *BandwidthMeter
    state          int32 // 0: CLOSED, 1: HALF_OPEN, 2: OPEN
    mu             sync.Mutex
}

func (cb *MediaCircuitBreaker) TryAcquireStream(req *StreamRequest) error {
    // 1. 快速路径：原子读取状态
    if atomic.LoadInt32(&cb.state) == StateOpen {
        return ErrNodeOverload
    }
    
    // 2. 深度检查（仅在高水位触发）
    if cb.bandwidthMeter.UsageRatio() > 0.85 || cb.sendQueue.Len() > cb.highWatermark {
        cb.mu.Lock()
        defer cb.mu.Unlock()
        if cb.state == StateClosed {
            cb.transitionToOpen()
            go cb.reportOverloadToSignaling() // 异步上报
        }
        return ErrNodeOverload
    }
    return nil
}

func (cb *MediaCircuitBreaker) transitionToOpen() {
    atomic.StoreInt32(&cb.state, StateOpen)
    // 启动后台恢复检测协程
    go func() {
        ticker := time.NewTicker(5 * time.Second)
        defer ticker.Stop()
        for range ticker.C {
            if cb.bandwidthMeter.UsageRatio() < 0.7 && cb.sendQueue.Len() < cb.lowWatermark {
                atomic.StoreInt32(&cb.state, StateClosed)
                cb.reportRecoveryToSignaling()
                return
            }
        }
    }()
}

graph LR
    A[指标异常/告警触发] --> B{自动化诊断引擎}
    B -->|带宽水位高| C[触发 HPA 扩容/调度器驱逐低优先级 Pod]
    B -->|丢包率高| D[下发路由规则 切换备用链路/节点]
    B -->|版本指标劣化| E[自动回滚金丝雀版本]
    B -->|证书即将过期| F[触发 Cert-Manager 续签并热更]
    C & D & E & F --> G[生成变更单/复盘报告]
    G --> H[知识库沉淀]

# Envoy Listener 配置：QUIC 早期数据与连接迁移感知
listener:
  name: media_quic_ingress
  address:
    socket_address:
      address: 0.0.0.0
      port_value: 443
  udp_listener_config:
    quic_options:
      # 关键：开启连接迁移支持，避免客户端切网（WiFi->5G）导致 Sidecar 断流
      enable_migration: true
      # 早期数据（0-RTT）风险控制：媒体流幂等性低，建议关闭或仅允许信令帧
      allow_0rtt: false
  filter_chains:
  - filter_chain_match:
      transport_protocol: "QUIC"
    filters:
    - name: envoy.filters.network.udp_proxy
      typed_config:
        "@type": type.googleapis.com/envoy.extensions.filters.network.udp_proxy.v3.UdpProxyConfig
        # 核心：路由决策基于 QUIC Initial Packet 中的 SNI/DCID，不解密 Payload
        route_table_config:
          routes:
          - match:
              # 自定义 Matcher：解析 QUIC Client Hello 中的 SNI 字段
              sni: "sfu.media.internal"
            cluster: sfu_media_cluster

sequenceDiagram
    participant Client as 客户端 SDK
    participant Sidecar as Sidecar (旁路)
    participant SFU as 媒体节点
    participant Control as 信令/控制面
    
    Client->>SFU: 加密媒体流 (SFrame)
    SFU->>Sidecar: 明文 RTCP Sender/Receiver Report (SR/RR)
    Note right of Sidecar: 旁路抓包/内核探针采集网络层 QoS
    Client->>Control: 信令通道上报 "Media Telemetry" (加密负载下的编码器内部状态)
    Note right of Control: 关键帧间隔、QP值、编码耗时、CPU占用
    Control->>Sidecar: 下发聚合指标标签 (conference_id, user_id)
    Sidecar->>TSDB: 合并网络层+编码层指标

+-----------------------------------------------------------+
|                    全局控制平面 (Global CP)                |
|  - 统一服务注册 (Multi-Cluster Service Registry)           |
|  - 全局流量策略 (Global Traffic Policy / GTP)              |
|  - 证书根信任链分发 (Root CA -> Intermediate CA per Cloud) |
+---------------------------+-------------------------------+
                            | gRPC/XDS (mTLS)
        +-------------------+-------------------+
        |                   |                   |
+-------v------+    +-------v------+    +-------v------+
|  公有云网格   |    |  私有IDC网格  |    |  边缘POP网格  |
|  (Regional CP)|    |  (Regional CP)|    |  (Lightweight)|
|  - Sidecar   |    |  - Sidecar   |    |  - eBPF Agent |
|  - Ingress GW|    |  - Ingress GW|    |  - Local LB   |
+--------------+    +--------------+    +--------------+

# Gateway API (SIG-Networking 标准) 定义媒体入口
kind: Gateway
apiVersion: gateway.networking.k8s.io/v1beta1
metadata:
  name: media-edge-gateway
  annotations:
    # 关键：声明媒体流量类，触发专用数据面调度
    media.mesh.io/traffic-class: "realtime-udp"
spec:
  gatewayClassName: global-media-gateway
  listeners:
  - name: rtp-udp
    port: 30000
    protocol: UDP
    allowedRoutes:
      kinds:
      - kind: UDPRoute
        namespace:
          from: All
---
kind: UDPRoute
apiVersion: gateway.networking.k8s.io/v1alpha1
metadata:
  name: sfu-media-route
spec:
  parentRefs:
  - name: media-edge-gateway
  rules:
  - backendRefs:
    - name: sfu-service
      namespace: media-plane
      weight: 80
    - name: sfu-service-backup
      namespace: media-plane-dr
      weight: 20
    # 扩展：基于客户端 GeoIP 的就近调度策略
    filters:
    - type: ExtensionRef
      extensionRef:
        group: media.mesh.io
        kind: GeoProximityFilter
        name: nearest-pop

// XDP 程序：媒体流 UDP 包快速路径转发 + 指标采集
SEC("xdp/media_forward")
int xdp_media_forward(struct xdp_md *ctx) {
    void *data = (void *)(long)ctx->data;
    void *data_end = (void *)(long)ctx->data_end;
    struct ethhdr *eth = data;
    
    // 1. 协议解析 (Ethernet -> IP -> UDP)
    if (parse_udp_headers(eth, data_end, &udp_hdr, &payload) < 0)
        return XDP_PASS; // 非媒体流放行内核协议栈

    // 2. 从 Map 查找目标后端 (Key: 5元组 / DCID)
    struct backend_info *be = bpf_map_lookup_elem(&backend_map, &flow_key);
    if (!be) return XDP_DROP; // 无后端策略，丢弃

    // 3. 修改目标 IP/MAC (FIB 转发或直接重写)
    bpf_l3_csum_replace(...); // 校验和增量更新
    bpf_l4_csum_replace(...);
    memcpy(eth->h_dest, be->dst_mac, ETH_ALEN);

    // 4. 关键指标原子更新 (无锁)
    __sync_fetch_and_add(&be->stats.rx_packets, 1);
    __sync_fetch_and_add(&be->stats.rx_bytes, payload_len);
    
    // 5. 重定向到目标网卡/队列
    return bpf_redirect(be->ifindex, 0);
}

# 成本评估模型 (每分钟滚动计算)
class MediaCostModel:
    def __init__(self, cloud_pricing_api, bandwidth_tier_config):
        self.pricing = cloud_pricing_api
        self.bw_tiers = bandwidth_tier_config  # 阶梯价格

    def calculate_egress_cost(self, src_region, dst_region, bitrate_kbps, duration_sec):
        # 1. 跨区域/跨云出口带宽单价 (元/GB)
        unit_price = self.pricing.get_inter_region_price(src_region, dst_region)
        
        # 2. 考虑带宽包/预留实例抵扣
        effective_price = self.apply_commitment_discount(unit_price, src_region)
        
        # 3. 计算单流成本
        gb = (bitrate_kbps / 8 / 1024 / 1024) * duration_sec
        return gb * effective_price

    def calculate_compute_cost(self, node_type, codec, resolution, duration_sec):
        # 映射编码配置到 CPU 核心占用
        cpu_cores = self.estimate_cpu_usage(codec, resolution) 
        spot_discount = self.pricing.get_spot_discount(node_type)
        return cpu_cores * self.pricing.get_cpu_hourly_price(node_type) * (duration_sec/3600) * (1-spot_discount)

// 伪代码：成本感知权重计算
func (r *MediaRouter) CalculateWeight(endpoint *Endpoint, req *MediaRequest) float64 {
    baseWeight := endpoint.HealthScore * endpoint.CapacityWeight
    
    // 成本因子 (0.0 ~ 1.0，越低越省钱)
    costFactor := 1.0
    
    // 1. 同可用区优先 (零跨AZ费)
    if endpoint.Zone == req.ClientZone {
        costFactor *= 0.1
    } 
    // 2. 同地域跨AZ (低费用)
    else if endpoint.Region == req.ClientRegion {
        costFactor *= 0.5
    }
    // 3. 跨云/跨地域 (高费用)
    else {
        costFactor *= 1.0
    }
    
    // 4. 抢占式实例/边缘节点额外折扣
    if endpoint.InstanceType == "Spot" || endpoint.NodeType == "Edge" {
        costFactor *= 0.7
    }
    
    // 5. 带宽包抵扣余量
    if endpoint.BandwidthCommitmentRemaining > req.EstimatedBitrate {
        costFactor *= 0.3
    }
    
    // 最终权重 = 基础权重 * (1 - 成本敏感度系数 * costFactor)
    // 成本敏感度系数可动态配置：大促/高峰期降低敏感度保体验，平峰期提高敏感度省成本
    return baseWeight * (1 - r.costSensitivity * costFactor)
}

// 客户端 -> 控制面 (单向上报，无需 ACK)
message ClientDiagnosticReport {
  string trace_id = 1;
  int64 timestamp_ms = 2;
  // 网络层指标
  NetworkSnapshot network = 3;
  // 编解码器内部状态
  EncoderState video_encoder = 4;
  EncoderState audio_encoder = 5;
  // 关键事件
  repeated KeyEvent events = 6;
}

message NetworkSnapshot {
  double rtt_ms = 1;
  double packet_loss_rate = 2; // 最近 1s
  double bandwidth_estimate_kbps = 3;
  bool ecn_ce_marked = 4;      // 显式拥塞通知
  string local_interface_type = 5; // wifi/cellular/ethernet
  int32 signal_strength_dbm = 6;   // 无线信号强度
}

message KeyEvent {
  enum Type {
    KEY_FRAME_LOSS = 0;
    DTLS_HANDSHAKE_FAIL = 1;
    ICE_NOMINATION_FAIL = 2;
    CPU_THROTTLING = 3;
  }
  Type type = 1;
  int64 timestamp_ms = 2;
  map<string, string> context = 3; // 如 { "codec": "H264", "layer": "spatial_2" }
}

# 伪代码：多模态 Embedding 生成管线
class MultiModalEmbedder:
    def embed_incident(self, incident_id: str) -> np.ndarray:
        # 1. 时序指标 -> PatchTST / TimesNet Embedding (捕捉异常形态)
        metric_emb = self.ts_encoder.encode(
            fetch_prometheus_range(incident_id, window="30m")
        )
        
        # 2. 日志/Trace -> LogBERT / BERT-Log Embedding (捕捉错误语义)
        log_emb = self.log_encoder.encode(
            fetch_loki_logs(incident_id, window="15m")
        )
        
        # 3. 拓扑变更 -> GraphSAGE Embedding (捕捉依赖影响面)
        topo_emb = self.graph_encoder.encode(
            fetch_service_graph_snapshot(incident_id)
        )
        
        # 4. 发布变更 -> 文本 Embedding (捕捉变更风险)
        change_emb = self.text_encoder.encode(
            fetch_git_changes(incident_id, window="2h")
        )
        
        # 5. 加权融合 (权重可通过历史标注数据学习)
        return np.concatenate([
            metric_emb * 0.4,
            log_emb * 0.3,
            topo_emb * 0.2,
            change_emb * 0.1
        ])

kind: BackendTrafficPolicy
apiVersion: gateway.networking.k8s.io/v1alpha1
metadata:
  name: sfu-disaster-recovery
spec:
  targetRef:
    group: gateway.networking.k8s.io
    kind: HTTPRoute
    name: sfu-media-route
  # 故障检测：主动探测 + 被动健康检查
  healthCheck:
    active:
      interval: 2s
      timeout: 1s
      # 自定义 UDP 探活载荷 (STUN Binding Request)
      customPayload: "000100002112a442..." 
      healthyThreshold: 2
      unhealthyThreshold: 3
    passive:
      # 结合 Sidecar 上报的媒体指标判定
      customMetrics:
      - name: media.packet_loss_rate
        threshold: "0.05" # 5%
      - name: media.mos_score
        threshold: "3.0"
  # 故障转移策略
  failover:
    # 优先同城双AZ
    - locality: "same-region-different-zone"
      weight: 100
    # 次选异地灾备 (延迟容忍度配置)
    - locality: "different-region"
      weight: 10
      # 仅当同城全挂时启用，且客户端需支持高延迟模式
      activationCondition: "all-primary-unhealthy"
  # 熔断保护：防止故障节点恢复瞬间流量洪峰冲垮
  circuitBreaker:
    maxConnections: 10000
    maxPendingRequests: 500
    splitExternalLocalOriginErrors: true

# AuthorizationPolicy 扩展：基于数据分级的流量隔离
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: media-data-residency
spec:
  selector:
    matchLabels:
      app: sfu
  action: ALLOW
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/signaling/sa/signaling"]
    to:
    - operation:
        ports: ["30000-32000"]
    # 关键约束：仅允许同合规域流量
    when:
    - key: request.headers[x-compliance-domain]
      values: ["CN-FINANCE", "CN-GOV", "EU-GDPR"] # 必须匹配目标节点域标签
    - key: destination.labels[topology.compliance-domain]
      values: ["CN-FINANCE", "CN-GOV", "EU-GDPR"]

resources:
  requests:
    cpu: "200m"        # 非代理媒体流，仅控制面通信
    memory: "128Mi"
  limits:
    cpu: "500m"        # 防止抢占媒体进程 CPU
    memory: "256Mi"
# 关键环境变量
env:
- name: ENVOY_CONCURRENCY
  value: "2"           # 绑定 2 核，避免上下文切换
- name: ISTIO_META_CPUS
  value: "2"
- name: PILOT_ENABLE_PROTOCOL_SNIFFING_FOR_INBOUND
  value: "false"       # 关闭协议嗅探，媒体端口非标准协议
- name: ISTIO_META_DNS_CAPTURE
  value: "false"       # 关闭 DNS 劫持，媒体节点直连 IP

# /etc/sysctl.d/99-media-mesh.conf
# UDP 缓冲区扩大 (应对突发流量)
net.core.rmem_max = 67108864
net.core.wmem_max = 67108864
net.ipv4.udp_rmem_min = 16384
net.ipv4.udp_wmem_min = 16384

# 连接跟踪表扩容 (高并发 NAT 场景)
net.netfilter.nf_conntrack_max = 2000000
net.netfilter.nf_conntrack_udp_timeout = 30
net.netfilter.nf_conntrack_udp_timeout_stream = 120

# BBR 拥塞控制 (QUIC/媒体流友好)
net.core.default_qdisc = fq
net.ipv4.tcp_congestion_control = bbr

# 时间戳与快速回收 (NAT 环境必开)
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_timestamps = 1

groups:
- name: media-plane-critical
  rules:
  # 1. 入会失败率飙升 (P0)
  - alert: MediaJoinFailureRateHigh
    expr: |
      sum(rate(media_conference_join_failed_total[2m])) by (cluster, region)
      /
      sum(rate(media_conference_join_attempt_total[2m])) by (cluster, region)
      > 0.01
    for: 1m
    labels:
      severity: critical
      team: rtc-platform
    annotations:
      summary: "入会失败率 > 1% ({{ $labels.region }})"
      runbook_url: "https://wiki.internal/runbook/media-join-failure"

  # 2. 单节点带宽水位逼近物理上限 (P0)
  - alert: MediaNodeBandwidthSaturation
    expr: |
      (media_node_bandwidth_usage_bytes / media_node_bandwidth_limit_bytes) > 0.85
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "节点 {{ $labels.pod }} 带宽水位 85%"
      action: "触发 HPA 扩容 / 调度器驱逐低优先级 Pod"

  # 3. 跨云专线丢包/延迟异常 (P1)
  - alert: CrossCloudLinkDegraded
    expr: |
      histogram_quantile(0.99, rate(media_cross_cloud_rtt_seconds_bucket[5m])) > 0.15
      or
      rate(media_cross_cloud_packet_loss_total[5m]) > 0.005
    for: 3m
    labels:
      severity: warning
    annotations:
      summary: "跨云链路 {{ $labels.src_cloud }}->{{ $labels.dst_cloud }} 质量劣化"
      action: "网格路由自动切换备用链路 / 联系网络团队排查专线"

  # 4. 证书即将过期 (P1, 防止突发中断)
  - alert: MediaMTLSCertExpiringSoon
    expr: |
      (istio_agent_cert_expiry_timestamp - time()) < 86400 * 7
    for: 1h
    labels:
      severity: warning
    annotations:
      summary: "节点 {{ $labels.workload }} mTLS 证书 7 天后过期"
      action: "检查 Cert-Manager / SDS 同步状态"