[Client/App] --> [Sidecar (Envoy Inbound)] --> [Local Service]
       |                                           ^
       |-- Mirror (Clone) --> [Collector/Analyzer] |
                            (Envoy Filter Config)  |

apiVersion: networking.istio.io/v1beta1
kind: EnvoyFilter
metadata:
  name: meeting-signaling-mirror
  namespace: istio-system
spec:
  workloadSelector:
    labels:
      app: video-conference-gateway # 目标业务标签
  configPatches:
  - applyTo: HTTP_FILTER
    match:
      context: SIDECAR_INBOUND # 或 GATEWAY_INBOUND
      listener:
        filterChain:
          filter:
            name: envoy.filters.network.http_connection_manager
            subFilter:
              name: envoy.filters.http.router
    patch:
      operation: INSERT_BEFORE
      value:
        name: envoy.filters.http.lua
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.http.lua.v3.Lua
          inline_code: |
            -- 核心镜像逻辑
            function envoy_on_request(request_handle)
              local headers = request_handle:headers()
              local path = headers:get(":path") or ""
              local content_type = headers:get("content-type") or ""
              local user_agent = headers:get("user-agent") or ""

              -- 1. 特征匹配规则（根据实际协议调整）
              local is_meeting_traffic = false
              -- 匹配 WHIP/WHEP 端点
              if string.find(path, "/whip") or string.find(path, "/whep") then
                is_meeting_traffic = true
              -- 匹配 SDP 协商 Body
              elseif string.find(content_type, "application/sdp") then
                is_meeting_traffic = true
              -- 匹配自定义业务 Header
              elseif headers:get("x-meeting-session-id") then
                is_meeting_traffic = true
              end

              if is_meeting_traffic then
                -- 2. 克隆请求头并发送至镜像集群
                -- 注意：此处需预先定义静态集群 'mirror_collector_cluster'
                local mirror_headers = headers:clone()
                mirror_headers:remove(":authority") -- 避免路由冲突
                mirror_headers:add("x-mirror-source", "envoy-sidecar")
                mirror_headers:add("x-original-path", path)

                -- 异步发送，不阻塞主链路
                request_handle:httpCall(
                  "mirror_collector_cluster", -- 目标集群名称
                  {
                    [":method"] = "POST",
                    [":path"] = "/api/v1/mirror/http",
                    [":authority"] = "collector.internal",
                    ["content-type"] = "application/grpc+proto" -- 或 application/octet-stream
                  },
                  request_handle:body():getBytes(0, 16384), -- 限制镜像 Body 大小防 OOM
                  5000 -- 超时 5s
                )
                -- 记录日志供审计
                request_handle:logInfo("[Mirror] Signaling traffic mirrored: " .. path)
              end
            end

            function envoy_on_response(response_handle)
              -- 响应侧同步镜像关键 Header (如 Session-ID, ICE Candidates)
              local headers = response_handle:headers()
              if headers:get("x-mirrored") == "true" then
                response_handle:httpCall("mirror_collector_cluster", { ... }, "", 2000)
              end
            end

# 仅作演示，生产建议用 Wasm/C++ Filter 实现零拷贝转发
- applyTo: LISTENER
  match:
    listener:
      name: "0.0.0.0_50000" # 媒体端口监听器名
  patch:
    operation: MERGE
    value:
      listener_filters:
      - name: envoy.filters.listener.tap
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.listener.tap.v3.Tap
          match_config:
            # 仅匹配目标采集 Pod IP，避免全量镜像
            destination_ip_range:
              start: "10.200.10.50" # Collector IP
              end: "10.200.10.50"
          output_config:
            sinks:
            - streaming_admin: {} # 输出至 Envoy Admin /tap 接口
            max_buffered_rx_bytes: 1048576
            max_buffered_tx_bytes: 1048576

// 标准化镜像记录示例
{
  "timestamp": "2024-05-20T10:30:00.123Z",
  "source_pod": "video-gw-xxxx",
  "source_ip": "10.244.1.5",
  "dest_ip": "10.244.2.10",
  "protocol": "HTTP/2",
  "method": "POST",
  "path": "/api/v1/meetings/join",
  "status_code": 200,
  "latency_ms": 45,
  "request_size": 1024,
  "response_size": 2048,
  "meeting_id_hash": "sha256(real_id)", // 脱敏 ID
  "user_id_hash": "sha256(real_uid)",
  "sdp_direction": "sendrecv", // 从 Body 解析
  "ice_candidate_count": 4,
  "mirror_node": "envoy-sidecar-istio-proxy"
}

# 1. 检查 EnvoyFilter 状态
istioctl x efx describe pod <pod-name> -n <ns>

# 2. 查看 Sidecar 实际生成的 Listener/Filter Chain
istioctl proxy-config listener <pod-name> -n <ns> --port 15000 -o json | jq '.[] | select(.name | contains("http")) | .filterChains[].filters[] | select(.name=="envoy.filters.http.lua")'

# 3. 确认镜像集群健康
istioctl proxy-config cluster <pod-name> -n <ns> --fqdn mirror_collector_cluster -o json

# 在 Collector 节点抓取镜像入口流量
tcpdump -i eth0 -nn -s 0 -w mirror_verify.pcap port 8080 and host <sidecar_ip>

# Deployment Patch 示例
resources:
  limits:
    cpu: "2000m"
    memory: "2Gi"
  requests:
    cpu: "500m"
    memory: "1Gi"

circuit_breakers:
  thresholds:
  - max_connections: 1000
    max_pending_requests: 500
    max_requests: 2000
    max_retries: 3
    track_remaining: true
outlier_detection:
  consecutive_5xx: 3
  interval: 10s
  base_ejection_time: 30s
  max_ejection_percent: 50

// Cargo.toml 关键依赖
// proxy-wasm = { version = "0.4", features = ["logging"] }
// prost = "0.12"
// sdp = { git = "https://github.com/webrtc-rs/sdp" } // SDP 解析库
// bytes = "1.0"

use proxy_wasm::traits::*;
use proxy_wasm::types::*;
use std::collections::HashMap;

proxy_wasm::main! {{
    proxy_wasm::set_log_level(LogLevel::Trace);
    proxy_wasm::set_http_context(|_, _| -> Box<dyn HttpContext> { Box::new(MirrorContext) });
}}

struct MirrorContext {
    // 缓存会话上下文，避免重复解析
    session_cache: HashMap<String, MeetingMeta>, 
    mirror_cluster: String,
}

impl Context for MirrorContext {}

impl HttpContext for MirrorContext {
    // 请求头阶段：特征识别 + 决策
    fn on_http_request_headers(&mut self, _num_headers: usize, _end_of_stream: bool) -> Action {
        let path = self.get_http_request_header(":path").unwrap_or_default();
        let content_type = self.get_http_request_header("content-type").unwrap_or_default();
        
        // 1. 快速路径：非会议流量直接放行
        if !Self::is_meeting_signal(&path, &content_type) {
            return Action::Continue;
        }

        // 2. 提取关键元数据 (MeetingID, UserID, SDP Type)
        let meeting_id = self.extract_meeting_id(&path, &content_type);
        if let Some(mid) = meeting_id {
            // 缓存元数据供 Body/Response 阶段复用
            self.session_cache.insert(mid.clone(), MeetingMeta::new(mid));
            
            // 3. 异步发起镜像调用 (非阻塞)
            // 注意：Wasm 目前不支持直接 httpCall 发送 Body 流，需配合 Buffer 限制
            self.mirror_headers(&mid, "request");
        }
        Action::Continue
    }

    // 请求 Body 阶段：深度解析 SDP/Protobuf
    fn on_http_request_body(&mut self, body_size: usize, end_of_stream: bool) -> Action {
        if !end_of_stream { return Action::Continue; }
        
        // 零拷贝获取 Body 指针
        let body = self.get_http_request_body(0, body_size).unwrap_or_default();
        if let Ok(sdp) = sdp::Session::parse(&body) {
            // 解析媒体方向、编解码、ICE 参数
            if let Some(meta) = self.session_cache.values_mut().next() {
                meta.parse_sdp(&sdp);
            }
            // 镜像 Body (截断大包)
            self.mirror_body(&body[..body_size.min(16_384)]);
        }
        Action::Continue
    }

    // 响应阶段：补全 ICE Candidate、选中编解码
    fn on_http_response_headers(&mut self, _num_headers: usize, _end_of_stream: bool) -> Action {
        if let Some(meta) = self.session_cache.values_mut().next() {
            meta.response_status = self.get_http_response_header(":status").unwrap_or_default();
            self.mirror_headers(&meta.meeting_id, "response");
        }
        Action::Continue
    }
}

impl MirrorContext {
    fn is_meeting_signal(path: &str, ct: &str) -> bool {
        // 编译期正则优化，或使用 Aho-Corasick 多模匹配
        path.contains("/whip") || path.contains("/whep") 
        || ct.contains("application/sdp") 
        || ct.contains("application/grpc") // gRPC 信令
    }

    fn mirror_headers(&self, meeting_id: &str, phase: &str) {
        let mut headers = vec![
            (":method", "POST"),
            (":path", "/api/v1/mirror/wasm"),
            (":authority", "collector.internal"),
            ("x-mirror-phase", phase),
            ("x-meeting-id-hash", &sha256_hash(meeting_id)), // 脱敏
            ("x-envoy-wasm", "true"),
        ];
        // 发送至镜像集群 (需在 Envoy 配置中定义 wasm_vm 的 cluster 权限)
        self.http_call(
            "mirror_collector_cluster", 
            &headers, 
            None, 
            None, 
            2000, 
            |_| {} // 回调忽略结果，保证非阻塞
        ).ok();
    }
}

# 1. 编译产物上传 OSS/Artifactory
# wasm_file: target/wasm32-wasi/release/meeting_mirror.wasm

# 2. Istio WasmPlugin 资源 (支持滚动更新)
apiVersion: extensions.istio.io/v1alpha1
kind: WasmPlugin
metadata:
  name: meeting-mirror-wasm
  namespace: istio-system
spec:
  selector:
    matchLabels:
      app: video-conference-gateway
  url: oci://registry.internal/wasm/meeting-mirror:v1.2.3 # OCI 镜像分发
  sha256: "a1b2c3d4..." # 校验和，防篡改
  vmConfig:
    vmId: "meeting-mirror-vm" # 共享 VM 实例，节省内存
    runtime: "v8" # 或 wasmtime/wamr
    allowPrecompiled: true
  pluginConfig:
    mirror_cluster: "mirror_collector_cluster"
    max_body_size: 16384
    # 动态规则下发，无需重新编译 Wasm
    match_rules: 
      - path_prefix: "/api/v1/meeting"
        headers:
          "x-biz-type": "conference"
      - content_type: "application/sdp"
  phase: AUTHZ # 在 AUTHZ 阶段运行，早于 Router
  priority: 100

# Envoy Bootstrap / Istio Gateway 扩展
static_resources:
  listeners:
  - name: quic_https
    address:
      socket_address:
        address: 0.0.0.0
        port_value: 443
    listener_filters:
    - name: envoy.filters.listener.tls_inspector
    filter_chains:
    - filter_chain_match:
        transport_protocol: "quic"
      filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          codec_type: HTTP3
          http3_protocol_options: {}
          # 复用 HTTP Filter 链 (含 Wasm 镜像插件)
          http_filters:
          - name: envoy.filters.http.wasm # Wasm 镜像插件
            typed_config: ...
          - name: envoy.filters.http.router

enum GrpcStreamState {
    Init,           // 等待 Headers
    Streaming { meeting_id: String, msg_count: u64 }, // 流式传输中
    Closed,
}

impl HttpContext for MirrorContext {
    fn on_http_request_headers(&mut self, _, end_of_stream) -> Action {
        if self.is_grpc_bidi_stream() {
            self.state = GrpcStreamState::Streaming { 
                meeting_id: self.extract_grpc_metadata("meeting-id"), 
                msg_count: 0 
            };
            // 镜像初始元数据
            self.mirror_grpc_initial_metadata();
        }
        Action::Continue
    }

    fn on_http_request_body(&mut self, body_size: usize, end_of_stream: bool) -> Action {
        if let GrpcStreamState::Streaming { meeting_id, msg_count } = &mut self.state {
            // gRPC Frame 解帧: Length-Prefixed Message (1B Compressed + 4B Length + Payload)
            let frames = self.parse_grpc_frames(&body);
            for frame in frames {
                *msg_count += 1;
                // 采样镜像：高频心跳包 1% 采样，关键控制帧 100% 镜像
                if self.should_mirror_frame(&frame, *msg_count) {
                    self.mirror_grpc_frame(&meeting_id, &frame);
                }
            }
        }
        if end_of_stream { self.state = GrpcStreamState::Closed; }
        Action::Continue
    }
}

// 统一镜像元数据 Protobuf 定义 (mirror_metadata.proto)
message MirrorEnvelope {
  // 全局唯一追踪 ID (W3C Traceparent 兼容)
  string trace_id = 1;        // 格式: version-trace-id-parent-id-flags
  // 业务会议 ID (脱敏 Hash)
  string meeting_id_hash = 2; 
  // 来源拓扑定位
  TopologyInfo topology = 3;
  // 原始协议载荷
  bytes payload = 4;
  // 采样决策标记
  SamplingDecision sampling = 5;
}

message TopologyInfo {
  string cloud_provider = 1;   // aws, aliyun, onprem
  string region = 2;           // cn-hangzhou, us-east-1
  string cluster_name = 3;     // k8s cluster id
  string namespace = 4;
  string workload_name = 5;    // deployment/daemonset name
  string pod_ip = 6;
  // 网络路径标识
  string ingress_gateway = 7;  // 入口网关 ID
  string egress_gateway = 8;   // 出口网关 ID
}

fn sanitize_sdp(&self, sdp: &mut sdp::Session) {
    // 1. 替换连接地址 (c=) 为网段掩码
    for conn in &mut sdp.connections {
        conn.address = mask_ip_to_cidr(&conn.address, 24); // 192.168.1.5 -> 192.168.1.0/24
    }
    // 2. 处理候选地址 (a=candidate)
    for attr in &mut sdp.attributes {
        if attr.key == "candidate" {
            attr.value = sanitize_candidate(&attr.value); // 保留 foundation/component/protocol/priority/type, 脱敏 IP/端口
        }
        // 3. 移除指纹/密钥 (a=fingerprint, a=crypto, a=setup:actpass 等)
        if matches!(attr.key.as_str(), "fingerprint" | "crypto" | "setup" | "mid" /* 可选 */) {
            attr.value = "[REDACTED]".to_string();
        }
    }
}

infra-config-repo/
├── envoy-filters/
│   ├── base/                    # 基础通用配置 (集群定义、公共 Lua/Wasm 引用)
│   │   ├── mirror-cluster.yaml
│   │   └── wasm-vm-config.yaml
│   ├── overlays/
│   │   ├── prod/                # 生产环境：严格采样、全脱敏
│   │   │   ├── kustomization.yaml
│   │   │   └── meeting-mirror-patch.yaml  # patches: replica=2, sampling=0.01
│   │   ├── staging/             # 预发环境：全量镜像、保留 Body 用于回放
│   │   │   └── meeting-mirror-patch.yaml  # patches: sampling=1.0, body_limit=64k
│   │   └── canary/              # 灰度环境：新版 Wasm 验证
│   └── modules/
│       └── meeting-mirror-wasm/ # Wasm 源码子模块 (Cargo workspace)
│           ├── Cargo.toml
│           └── src/

# .gitlab-ci.yml / Jenkinsfile 片段
stages:
  - test
  - chaos
  - deploy

chaos_verification:
  stage: chaos
  script:
    - kubectl apply -f chaos/experiments/mirror-link-failure.yaml
    - sleep 300 # 观测窗口
    - python scripts/check_slo.py --target meeting-gateway --threshold 99.99
    - kubectl delete -f chaos/experiments/mirror-link-failure.yaml
  rules:
    - if: $CI_COMMIT_BRANCH == "main"

# 伪代码：运行在 Collector 或控制平面的采样决策服务
class AdaptiveSampler:
    def __init__(self, budget_bps: int, min_rate=0.001, max_rate=1.0):
        self.budget = budget_bps
        self.current_rate = 0.01 # 默认 1%
        self.meeting_states = {} # meeting_id -> {score, rate, start_time}

    def on_meeting_start(self, meeting_id: str, metadata: dict):
        # 基线采样率
        rate = self.current_rate
        # 画像加权：大型会议/跨国/重要客户 提权
        if metadata.get("participant_count", 0) > 50: rate *= 5
        if metadata.get("is_cross_region"): rate *= 3
        if metadata.get("vip_user"): rate = 1.0
        self.meeting_states[meeting_id] = {"rate": min(rate, 1.0), "score": 0}

    def on_metrics_update(self, meeting_id: str, metrics: dict):
        # 实时异常评分 (0-100)
        # PLR>5% +30, RTT>300ms +20, Freeze>3次 +25, Codec Fallback +15
        score = self.calc_anomaly_score(metrics)
        self.meeting_states[meeting_id]["score"] = score
        
        # 动态调整：异常会议升级至全量
        if score > 60 and self.meeting_states[meeting_id]["rate"] < 1.0:
            self.meeting_states[meeting_id]["rate"] = 1.0
            self.push_config_update(meeting_id, 1.0) # 下发 xds/wasm 配置热更新

    def global_budget_control(self):
        # 定期 (10s) 重新分配预算
        total_estimated_bps = sum(
            self.estimate_bps(m) * s["rate"] 
            for m, s in self.meeting_states.items()
        )
        if total_estimated_bps > self.budget:
            # 按优先级 (score) 削减低分会议采样率
            self.redistribute_rates_by_priority()

    def should_mirror(self, meeting_id: str) -> bool:
        rate = self.meeting_states.get(meeting_id, {}).get("rate", self.current_rate)
        return random.random() < rate