graph LR
    A[Client SDK<br/>Web/iOS/Android] -->|OTLP/HTTP| B(Otel Collector<br/>Gateway)
    C[Gateway/SBC<br/>Go/Rust] -->|OTLP/gRPC| B
    D[Signaling Service<br/>Java/Go] -->|OTLP/gRPC| B
    E[Media Server<br/>C++/Rust<br/>SFU/MCU] -->|OTLP/gRPC/UDP| B
    F[Business Service<br/>Python/Go] -->|OTLP/gRPC| B
    B -->|Processor| G[(Trace Backend<br/>Tempo/Jaeger/Elasticsearch)]
    B -->|Processor| H[(Metrics Backend<br/>Prometheus/VictoriaMetrics)]
    B -->|Processor| I[(Log Backend<br/>Loki/Elasticsearch)]
    G --> J[Grafana Dashboard]
    H --> J
    I --> J

// Go伪代码： mediasoup worker 处理信令时
func handleSignalingMessage(msg SignalMessage, peer *Peer) {
    // 1. 从信令载荷提取 TraceContext
    ctx := otel.GetTextMapPropagator().Extract(context.Background(), propagation.MapCarrier{
        "traceparent": msg.TraceContext,
    })
    
    // 2. 创建 Span，关联会议与传输标识
    ctx, span := tracer.Start(ctx, "media.sfu.handle_offer",
        trace.WithAttributes(
            attribute.String("meeting.id", peer.MeetingID()),
            attribute.String("transport.id", peer.TransportID()),
            attribute.String("user.id", peer.UserID()),
        ),
    )
    defer span.End()
    
    // 3. 业务逻辑...
    // 4. 记录关键事件
    span.AddEvent("ice_connection_state_change", trace.WithAttributes(
        attribute.String("state", "connected"),
    ))
}

processors:
  tailsampling:
    decision_wait: 30s
    num_traces: 50000
    expected_new_traces_per_sec: 2000
    policies:
      - name: error-policy
        type: string_attribute
        string_attribute:
          key: "error.category"
          values: ["*"] # 只要有error属性即命中
      - name: slow-join-policy
        type: latency
        latency:
          threshold_ms: 5000
      - name: large-meeting-policy
        type: string_attribute
        string_attribute:
          key: "meeting.size"
          values: ["large", "webinar"]
      - name: default-low-sample
        type: probabilistic
        probabilistic:
          sampling_percentage: 1.0

// 定期采集 getStats 关联到当前 Span
setInterval(() => {
  const span = trace.getSpan(context.active());
  if (!span || !peerConnection) return;
  
  peerConnection.getStats().then(stats => {
    stats.forEach(report => {
      if (report.type === 'inbound-rtp' && report.kind === 'video') {
        span.setAttribute('webrtc.inbound.fps', report.framesPerSecond);
        span.setAttribute('webrtc.inbound.jitter_ms', report.jitter * 1000);
        span.setAttribute('webrtc.inbound.packets_lost', report.packetsLost);
        span.setAttribute('webrtc.inbound.nack_count', report.nackCount);
      }
      if (report.type === 'candidate-pair' && report.state === 'succeeded') {
        span.setAttribute('network.rtt_ms', report.currentRoundTripTime * 1000);
        span.setAttribute('network.available_outgoing_bitrate', report.availableOutgoingBitrate);
      }
    });
  });
}, 5000); // 5秒采集一次，避免性能开销

// C++ 伪代码：基于 opentelemetry-cpp 的异步非阻塞埋点
class OtelTracer {
    std::shared_ptr<opentelemetry::trace::Tracer> tracer_;
    // 无锁环形缓冲区，生产者(媒体线程) -> 消费者(上报线程)
    moodycamel::ConcurrentQueue<SpanEvent> event_queue_; 
    std::thread reporter_thread_;

public:
    void OnIceStateChange(Transport* t, IceState new_state) {
        // 仅记录状态变更，不记录每次 Check
        if (t->last_ice_state_ == new_state) return;
        t->last_ice_state_ = new_state;
        
        SpanEvent ev;
        ev.name = "ice.state_change";
        ev.attrs = {
            {"meeting.id", t->meeting_id_},
            {"transport.id", t->id_},
            {"ice.state", ToString(new_state)},
            {"ice.role", t->role_ == ICE_ROLE_CONTROLLING ? "controlling" : "controlled"}
        };
        ev.timestamp = std::chrono::system_clock::now();
        event_queue_.enqueue(std::move(ev)); // 非阻塞，极快
    }

    void OnFirstKeyFrameReceived(Transport* t, uint32_t ssrc, CodecType codec) {
        SpanEvent ev;
        ev.name = "media.first_keyframe";
        ev.attrs = {
            {"meeting.id", t->meeting_id_},
            {"rtp.ssrc", std::to_string(ssrc)},
            {"codec.type", CodecToString(codec)},
            {"latency.ms", (Now() - t->created_at_).count()}
        };
        event_queue_.enqueue(std::move(ev));
    }
};

// Rust 伪代码：使用 opentelemetry_metrics 推送高基数指标
use opentelemetry::metrics::{Counter, Histogram, UpDownCounter};
use opentelemetry::KeyValue;

struct MediaMetrics {
    rtx_sent: Counter<u64>,
    nack_received: Counter<u64>,
    jitter_ms: Histogram<f64>,
    available_bitrate: UpDownCounter<i64>,
}

impl MediaMetrics {
    fn new(meter: &opentelemetry::metrics::Meter) -> Self {
        Self {
            rtx_sent: meter.u64_counter("media.rtx.packets_sent")
                .with_description("RTX retransmitted packets")
                .init(),
            jitter_ms: meter.f64_histogram("media.jitter_ms")
                .with_boundaries(vec![1.0, 5.0, 10.0, 30.0, 50.0, 100.0])
                .init(),
            // ...
        }
    }

    // 在媒体线程直接调用，极低开销（原子操作）
    fn record_jitter(&self, jitter: f64, attrs: &[KeyValue]) {
        self.jitter_ms.record(jitter, attrs);
    }
}

# otelcol-contrib 配置片段：合规处理管道
processors:
  # 1. 字段重命名/标准化
  attributes/standardize:
    actions:
      - key: "http.client_ip"
        action: rename
        new_key: "client.ip.original" # 标记为原始IP，后续处理
      - key: "enduser.id"
        action: rename
        new_key: "user.id"

  # 2. 敏感字段红名单拦截 (L1 级)
  attributes/redact_pii:
    actions:
      - key: "client.ip.original"
        action: delete # 直接删除，或用 hash 替换
      - key: "user.real_name"
        action: hash # 单向哈希，保留关联性但不可逆
        hash_function: "sha256"
      - key: "meeting.topic"
        action: delete # 会议主题可能包含敏感词，默认不采集
      - key: "http.request.header.authorization"
        action: delete
      - pattern: "credit_card|password|token|secret"
        action: delete

  # 3. 资源属性注入 (集群/租户/环境标识)
  resource:
    attributes:
      - key: "deployment.environment"
        value: "prod"
        action: "upsert"
      - key: "k8s.cluster.name"
        from_attribute: "k8s.cluster.uid"
        action: "insert"

  # 4. 批量压缩与内存保护
  batch:
    timeout: 10s
    send_batch_max_size: 8192
    send_batch_size: 2048
  memory_limiter:
    check_interval: 1s
    limit_mib: 1500 # 限制 Collector 占用内存，防止 OOM Kill
    spike_limit_mib: 512

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, attributes/standardize, attributes/redact_pii, resource, batch, tailsampling]
      exporters: [tempo, logging/debug]
    metrics:
      receivers: [otlp, prometheus]
      processors: [memory_limiter, attributes/redact_pii, resource, batch]
      exporters: [prometheusremotewrite]

graph TB
    subgraph Edge["边缘采集层 (DaemonSet - 每节点 1 个)"]
        E1[Collector Agent<br/>Port: 4317/4318<br/>职责: 接收、批量、压缩、首轮脱敏、本地缓冲]
    end
    subgraph Core["中心网关层 (Deployment - HPA 扩缩容)"]
        C1[Collector Gateway x N<br/>Port: 4317<br/>职责: 尾部采样、二次脱敏、路由分发、熔断降级]
    end
    subgraph Backend["后端存储"]
        B1[(Tempo/ES)]
        B2[(Prometheus/VM)]
        B3[(Loki)]
    end

    App[业务 Pod] -->|OTLP/gRPC<br/>Localhost| E1
    E1 -->|OTLP/gRPC<br/>LoadBalancer| C1
    C1 --> B1
    C1 --> B2
    C1 --> B3

# K8s HPA 配置示例 (需安装 Prometheus Adapter)
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: otelcol-gateway-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: otelcol-gateway
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: Pods
    pods:
      metric:
        name: otelcol_receiver_accepted_spans_per_second # 每秒接收 Span 数
      target:
        type: AverageValue
        averageValue: "5000" # 每个实例处理 5k spans/s 扩容
  - type: Pods
    pods:
      metric:
        name: otelcol_exporter_queue_capacity # 导出队列积压率
      target:
        type: AverageValue
        averageValue: "0.7" # 队列超过 70% 扩容
  behavior:
    scaleDown:
      stabilizationWindowSeconds: 300 # 缩容冷却 5 分钟，防抖
      policies:
      - type: Percent
        value: 10
        periodSeconds: 60
    scaleUp:
      stabilizationWindowSeconds: 30
      policies:
      - type: Percent
        value: 50
        periodSeconds: 30
      - type: Pods
        value: 4
        periodSeconds: 30
      selectPolicy: Max

# Collector 导出器配置：熔断策略
exporters:
  tempo:
    endpoint: "tempo-distributor:4317"
    tls:
      insecure: true
    # 关键熔断配置
    sending_queue:
      enabled: true
      num_consumers: 10
      queue_size: 10000 # 内存队列上限
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 30s
      max_elapsed_time: 120s # 2分钟内重试失败，触发熔断
    # 熔断后行为：丢弃数据而非阻塞
    # 需配合 processor 的 memory_limiter 使用

processors:
  attributes/hash_meeting_id:
    actions:
      - key: "meeting.id"
        action: hash
        hash_function: "sha256"
        # 保留前 16 字符作为伪 ID，平衡碰撞率与存储
        # 同时保留原始 ID 到另一个仅用于映射的属性 (可选，极小流量采样保留)
      - key: "meeting.id"
        action: rename
        new_key: "meeting.id.hashed"

-- 离线任务：每小时从信令数据库同步最新会议映射
CREATE TABLE meeting_id_map (
    hashed_id String,      -- SHA256(meeting_id) 前16位
    original_id String,    -- 真实 meeting_id
    tenant_id String,
    start_time DateTime,
    end_time DateTime
) ENGINE = MergeTree()
PARTITION BY toYYYYMMDD(start_time)
ORDER BY (tenant_id, hashed_id);

// Go 示例：使用 otel/sdk/trace/tracetest 验证 Span 产出
func TestSignalingInvite_SpanAttributes(t *testing.T) {
    // 1. 初始化 In-Memory Exporter
    exp := tracetest.NewInMemoryExporter()
    tp := trace.NewTracerProvider(trace.WithBatcher(exp))
    defer tp.Shutdown(context.Background())
    
    // 2. 执行业务逻辑
    ctx := context.Background()
    svc := NewSignalingService(tp.Tracer("signaling"))
    err := svc.Invite(ctx, "meeting_123", "user_456")
    require.NoError(t, err)
    
    // 3. 断言 Span 存在性与属性完整性
    spans := exp.GetSpans()
    require.Len(t, spans, 1)
    span := spans[0]
    
    assert.Equal(t, "signaling.invite.process", span.Name)
    assert.Equal(t, "meeting_123", span.Attributes()["meeting.id"].AsString())
    assert.Equal(t, "user_456", span.Attributes()["user.id"].AsString())
    assert.Equal(t, "invite", span.Attributes()["signaling.action"].AsString())
    
    // 4. 合规断言：严禁包含敏感字段
    forbidden := []string{"user.phone", "client.ip", "user.real_name"}
    for _, f := range forbidden {
        assert.NotContains(t, span.Attributes(), f, "合规违规：检测到敏感字段 %s", f)
    }
}

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

observability_gate:
  stage: observability-gate
  image: corp/otel-cli:latest
  script:
    # 1. 校验二进制/镜像依赖的探针版本是否符合兼容性矩阵
    - otel-cli check-version --service=$SERVICE_NAME --matrix=//infra/otel/compatibility.yaml
    
    # 2. 启动临时 Collector + Tempo 模拟生产管道
    - docker-compose -f docker-compose.otel-test.yml up -d
    
    # 3. 运行 E2E 测试流量
    - go test -tags=e2e ./... -run TestFullTracePipeline
    
    # 4. 校验输出
    - otel-cli verify-trace --service=$SERVICE_NAME --min-spans=5 --required-attrs="meeting.id,user.id,error.category"
    
    # 5. 合规扫描
    - otel-cli scan-pii --config=//infra/otel/pii-rules.yaml
  allow_failure: false

# 伪代码：简化版 RCA 评分逻辑
def calculate_suspicion_score(trace: Trace, baseline: Baseline) -> List[SuspectSpan]:
    suspects = []
    for span in trace.spans:
        score = 0.0
        # 1. 自身耗时异常权重 40%
        if span.duration > baseline.p99(span.name) * 3:
            score += 0.4 * min(1.0, span.duration / baseline.p99(span.name))
        
        # 2. 错误标记权重 30%
        if span.status.code == StatusCode.ERROR:
            score += 0.3
            # 细分错误类型
            if "timeout" in span.attributes.get("error.category", ""):
                score += 0.1
        
        # 3. 下游依赖异常传导权重 20%
        # 如果子 Span 高疑似，父 Span 疑似度上升
        child_scores = [s.score for s in suspects if s.parent_span_id == span.span_id]
        if child_scores:
            score += 0.2 * max(child_scores)
            
        # 4. 资源饱和度关联权重 10%
        # 查询该 Span 所在 Pod 当时的 CPU/Network 压力
        pod_metrics = query_pod_metrics(span.resource["k8s.pod.name"], span.start_time)
        if pod_metrics.cpu_usage > 80 or pod_metrics.net_drop > 0:
            score += 0.1
            
        if score > 0.5: # 阈值可调
            suspects.append(SuspectSpan(span.span_id, span.name, score, span.attributes))
            
    return sorted(suspects, key=lambda x: x.score, reverse=True)

graph LR
    App[Tenant A Pod] -->|Header: X-Tenant-ID: t_100| GW[Collector Gateway]
    App2[Tenant B Pod] -->|Header: X-Tenant-ID: t_200| GW
    GW -->|Processor: routing/tenant| TA[(Tempo: tenant=t_100)]
    GW -->|Processor: routing/tenant| TB[(Tempo: tenant=t_200)]
    GW -->|Metrics: relabel| PM[Prometheus: tenant_label]

processors:
  routing/tenant:
    table:
      - statement: 'attributes["tenant.id"] == "t_100"'
        pipelines: [traces/t100, metrics/t100]
      - statement: 'attributes["tenant.id"] == "t_200"'
        pipelines: [traces/t200, metrics/t200]
      - statement: 'true' # 默认/未识别租户
        pipelines: [traces/default, metrics/default]

  # 租户级采样率差异化 (VIP 客户全量，免费版 0.1%)
  tailsampling/tenant_policy:
    policies:
      - name: vip_tenant
        type: string_attribute
        string_attribute:
          key: "tenant.tier"
          values: ["vip", "enterprise"]
        # VIP 走低采样率策略或全量
      - name: free_tenant
        type: probabilistic
        probabilistic:
          sampling_percentage: 0.1

# 每天凌晨 02:00 自动执行核心链路混沌演练
apiVersion: chaos-mesh.org/v1alpha1
kind: Schedule
metadata:
  name: nightly-observability-drill
spec:
  schedule: "0 2 * * *"
  type: "workflow"
  workflowSpec:
    entry: "otel-verification"
    templates:
      - name: "otel-verification"
        templateType: "serial"
        children:
          - templateRef: "inject-network-latency"
          - templateRef: "verify-trace-integrity" # 调用 API 校验 Trace 完整性
          - templateRef: "verify-alert-fired"     # 校验 Alertmanager 收到告警
          - templateRef: "recover-network"