# 伪代码：流量特征采集关键字段
traffic_profile = {
    "接入层": {
        "并发连接数分布": "P50/P95/P99/峰值",
        "新建连接速率(CPS)": "秒级/分钟级突发曲线",
        "协议版本占比": "WebRTC/SDK/原生客户端版本矩阵"
    },
    "信令层": {
        "会话建立耗时分布": "Offer/Answer/ICE 交互 RTT",
        "重连/重入频次": "弱网/切网场景下的重试模型",
        "消息类型热度": "Join/Leave/Mute/Layout/Control"
    },
    "媒体层": {
        "发布/订阅流比": "1:N / M:N 拓扑分布",
        "码率/分辨率/帧率分布": "自适应码控档位统计",
        "丢包/抖动/RTT 联合分布": "弱网特征三维建模"
    },
    "业务层": {
        "会议规模分布": "2人/10人/100人/1000人+ 占比",
        "会议时长分布": "短会/长会/全天会混合模型",
        "用户行为序列": "进会→开麦→共享→离会 马尔可夫链"
    }
}

# 流量模型配置示例（YAML 化便于版本管理）
traffic_model:
  conference_scale:
    type: "mixture_gaussian"
    components:
      - weight: 0.65; mean: 8; std: 3      # 小会
      - weight: 0.25; mean: 45; std: 20    # 中会
      - weight: 0.10; mean: 320; std: 150  # 大会
  user_behavior:
    type: "markov_chain"
    states: ["join", "publish", "subscribe", "mute", "layout_change", "leave"]
    transition_matrix: [[...]]
  network_impairment:
    type: "copula_marginal"
    marginals:
      rtt: "lognormal(40ms, 1.2)"
      loss: "beta(0.5, 20)"
      jitter: "weibull(15ms, 1.5)"
    correlation: "clayton(theta=2.3)"

// 伪代码：ICE 状态机核心逻辑
type ICEAgent struct {
    gatheringState  GatheringState  // New → Gathering → Complete
    connectionState ConnectionState // New → Checking → Connected → Failed
    candidatePairs  []CandidatePair // 本地/远端候选对
    nominatedPair   *CandidatePair  // 选定对
    controlling     bool            // 控制方/被控方
}

func (a *ICEAgent) Run(ctx context.Context) {
    for {
        select {
        case <-ctx.Done(): return
        case event := <-a.eventChan:
            a.handleEvent(event)  // 绑定请求/响应、提名、保活
        case <-a.ticker.C:
            a.periodicCheck()     // 定期连通性检查
        }
    }
}

# 生产级持续画像命令示例
# CPU 热点（含内核态）
perf record -g -a --call-graph dwarf -o perf.data -- sleep 60

# 内存分配热点
go tool pprof -alloc_space http://localhost:6060/debug/pprof/heap

# 锁竞争分析
go tool pprof -mutex http://localhost:6060/debug/pprof/mutex

# 协程泄漏排查
go tool pprof -goroutines http://localhost:6060/debug/pprof/goroutine

# 生成差分火焰图
flamegraph.pl --diff baseline.folded bottleneck.folded > diff.svg

// eBPF 伪代码：测量 media_server.on_rtp_packet 处理延迟
UPROBE(media_server, on_rtp_packet) {
    start[tid] = nsecs;
}
URETPROBE(media_server, on_rtp_packet) {
    latency = nsecs - start[tid];
    hist.increment(log2(latency/1000)); // 微秒级桶
    delete start[tid];
}

graph LR
    A[选择流量模型版本] --> B[配置压测规模/时长/弱网策略]
    B --> C[提交任务 → 平台自动扩容模拟器集群]
    C --> D[实时观测四层指标大盘]
    D --> E{是否触发告警/瓶颈}
    E -- 是 --> F[一键下载火焰图/Trace/PCAP]
    F --> G[研发定位修复 → 提交回归单]
    E -- 否 --> H[生成压测报告 → 归档基线]
    G --> C

# 伪代码：HMM 弱网轨迹生成器核心逻辑
class NetworkTraceGenerator:
    def __init__(self, model_path: str):
        self.hmm = joblib.load(model_path)  # 状态: Good, Congested, Lossy, Handover
        self.param_sampler = {
            "Good":      lambda: {"rtt": lognorm(30, 0.3), "loss": beta(0.1, 50), "jitter": weibull(5, 2)},
            "Congested": lambda: {"rtt": lognorm(200, 0.8), "loss": beta(2, 20),  "jitter": weibull(50, 1.5)},
            "Lossy":     lambda: {"rtt": lognorm(80, 0.5), "loss": beta(10, 10), "jitter": weibull(30, 1.2)},
            "Handover":  lambda: {"rtt": spike(500, 2000), "loss": burst(0.3, 5), "jitter": spike(100, 300)}
        }

    def generate_trace(self, duration_sec: int, granularity_ms: int = 100) -> List[Dict]:
        states = self.hmm.sample(n_samples=duration_sec * 1000 // granularity_ms)
        trace = []
        for s in states:
            params = self.param_sampler[s]()
            params["state"] = s
            trace.append(params)
        return trace

graph TD
    A[原始指标流] --> B(规则引擎: 硬性阈值判定)
    A --> C(统计特征工程: 滑动窗口统计量)
    C --> D[LightGBM / XGBoost 回归模型]
    A --> E(关键帧指纹序列)
    E --> F[MobileNetV3-Tiny + GRU 微模型]
    B --> G[综合质量评分 Q-Score 0-100]
    D --> G
    F --> G
    G --> H{阈值判定}
    H -- 通过 --> I[压测通过]
    H -- 失败 --> J[自动切片定位 + 告警]

// 伪代码：Spot 实例回收前 120s 预警触发的 Checkpoint 逻辑
func (s *Simulator) HandleSpotInterruptionNotice() {
    // 1. 停止接收新连接，标记 Draining
    s.SetState(StateDraining)
    
    // 2. 异步快照核心状态到对象存储 (分片上传)
    snapshot := &Snapshot{
        Connections: s.dumpActiveConnections(), // 仅元数据: SSRC, CryptoSuite, SeqNum, JitterBufState
        TrafficModel: s.currentModelVersion,
        Timestamp: time.Now(),
    }
    go s.uploadSnapshot(snapshot) // 目标 < 30s 完成
    
    // 3. 通知调度器在新节点恢复
    s.schedulerClient.TriggerRecovery(snapshot.ID, s.preferredZone)
}

-- 核心宽表字段设计
CREATE TABLE pressure_test_cost_fact (
    task_id          STRING,
    git_commit       STRING,
    traffic_model_v  STRING,
    simulator_image  STRING,
    cluster_type     ENUM('onprem', 'ri', 'spot'),
    node_hours       DECIMAL(10,2),
    network_gbps_h   DECIMAL(10,2),
    storage_gb_h     DECIMAL(10,2),
    total_cost_cny   DECIMAL(12,2),
    cost_per_10k_ccu DECIMAL(10,4),  -- 单位成本：每万并发连接成本
    quality_score    DECIMAL(5,2),   -- NR-VQA 评分
    status           ENUM('success', 'partial', 'failed')
) PARTITIONED BY (dt STRING);

sequenceDiagram
    participant User as 研发工程师
    participant Copilot as RCA Copilot
    participant KB as 知识库
    participant Obs as 可观测平台
    participant Code as 代码仓库

    User->>Copilot: "本次压测 Task-20240520-001 卡顿率飙升至 5%，请分析"
    Copilot->>Obs: 调用 Tool: fetch_metrics(task_id, "stutter_rate", "p99_latency", "cpu", "retransmit")
    Copilot->>Obs: 调用 Tool: fetch_traces(task_id, error_only=true, span_filter="media_server")
    Copilot->>Obs: 调用 Tool: fetch_profiles(task_id, type="cpu,mutex,block")
    Copilot->>KB: 检索相似历史故障、相关代码变更、架构约束
    Copilot->>Copilot: **推理链路构建** (CoT): 现象 -> 假设 -> 证据链 -> 根因定位
    Copilot->>Code: 调用 Tool: get_code_context(file="media_server/bandwidth_estimator.cc", lines="120-250")
    Copilot-->>User: **输出结构化 RCA 报告** (含证据引用、代码定位、修复建议、验证步骤)
    User->>Copilot: "帮我生成修复 PR 描述和单测用例"
    Copilot->>Code: 生成 Diff + Test Case -> 创建 Draft PR

# 压测任务 CRD 示例 (压测即代码)
apiVersion: pressuretest.io/v1alpha1
kind: PressureTest
metadata:
  name: weekly-full-scale-2024-w21
spec:
  trafficModelRef: "traffic-model/v4.2@sha256:abc..."
  simulatorImage: "registry.io/simulator:v3.1.0-rc.5"
  targetClusters:
    - name: "prod-shanghai-01"
      role: "media-primary"
      replicaHint: 120  # 模拟器副本数
    - name: "prod-beijing-01"
      role: "media-dr"
      replicaHint: 30
    - name: "cloud-aws-sgp"
      role: "media-oversea"
      replicaHint: 20
      networkProfile: "cross-border-high-jitter"  # 引用预定义网络损伤配置
  duration: "30m"
  rampUp: "5m"
  chaos:
    - type: "pod-kill"
      target: "media-server"
      probability: 0.05
      interval: "2m"
  successCriteria:
    - metric: "global_stutter_rate"
      threshold: "< 1.5%"
    - metric: "p99_ttff_ms"
      threshold: "< 2000"
    - metric: "cost_per_10k_ccu"
      threshold: "< 15.0 CNY"
  notification:
    feishu_webhook: "https://open.feishu.cn/..."
    on_failure_only: false

// Neo4j Cypher 示例：核心 Schema
(:Task {id, name, start_time, status, cost})
(:TrafficModel {version, params_hash, description})
(:SimulatorBuild {image_tag, git_commit, compile_flags})
(:Cluster {name, region, provider, spec, network_topo})
(:MetricSnapshot {task_id, timestamp, metric_name, p50, p95, p99, max})
(:Bottleneck {id, type: "CPU/LOCK/NET/MEM/ALGO", location, severity, root_cause_summary})
(:CodeChange {commit_hash, file, function, author, pr_link})
(:FixAction {id, description, pr_link, verified_task_id})

// 关系
(:Task)-[:USES_MODEL]->(:TrafficModel)
(:Task)-[:RUNS_ON]->(:Cluster)
(:Task)-[:USES_BUILD]->(:SimulatorBuild)
(:Task)-[:PRODUCES]->(:MetricSnapshot)
(:Task)-[:IDENTIFIED]->(:Bottleneck)
(:Bottleneck)-[:ROOT_CAUSE_IN]->(:CodeChange)
(:Bottleneck)-[:FIXED_BY]->(:FixAction)
(:FixAction)-[:VERIFIED_IN]->(:Task)