┌─────────────┐     ┌─────────────┐     ┌─────────────┐     ┌─────────────┐
│  客户端      │     │  信令/网关   │     │  媒体节点    │     │  后台服务    │
│  (Web/iOS/  │     │  (Go/Rust)  │     │  (SFU/MCU)  │     │  (Java/Go)  │
│   Android)  │     │             │     │  (C++/Rust) │     │             │
└──────┬──────┘     └──────┬──────┘     └──────┬──────┘     └──────┬──────┘
       │                   │                   │                   │
       │  OTel SDK         │  OTel SDK         │  OTel SDK         │  OTel SDK
       │  (Web/移动端)     │                   │                   │
       ▼                   ▼                   ▼                   ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                    OpenTelemetry Collector (Sidecar / DaemonSet)        │
│  接收(OTLP/gRPC/HTTP) → 批处理/采样/脱敏 → 导出                         │
└─────────────────────────────────────────────────────────────────────────┘
                                    │
                    ┌───────────────┼───────────────┐
                    ▼               ▼               ▼
              ┌──────────┐   ┌──────────┐   ┌──────────┐
              │ Tempo/   │   │ Prometheus│   │ Loki/    │
              │ Jaeger   │   │ (Metrics) │   │ ES (Logs)│
              └──────────┘   └──────────┘   └──────────┘
                    │               │               │
                    └───────────────┼───────────────┘
                                    ▼
                          ┌──────────────────┐
                          │   Grafana 统一看板 │
                          └──────────────────┘

// Go 信令服务：SDP 交换事件
span.AddEvent("sdp.received", trace.WithAttributes(
    attribute.String("rtc.sdp.type", "offer"),
    attribute.Int("rtc.ice.candidates", len(candidates)),
))
span.AddEvent("sdp.answered", trace.WithAttributes(
    attribute.String("rtc.sdp.type", "answer"),
    attribute.Int64("rtc.latency.ms", time.Since(start).Milliseconds()),
))

// otel-setup.ts
import { WebTracerProvider } from '@opentelemetry/sdk-trace-web';
import { ZoneContextManager } from '@opentelemetry/context-zone';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { registerInstrumentations } from '@opentelemetry/instrumentation';
import { XMLHttpRequestInstrumentation } from '@opentelemetry/instrumentation-xml-http-request';
import { FetchInstrumentation } from '@opentelemetry/instrumentation-fetch';
import { Resource } from '@opentelemetry/resources';
import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions';

const provider = new WebTracerProvider({
  resource: new Resource({
    [SemanticResourceAttributes.SERVICE_NAME]: 'meeting-web-client',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: import.meta.env.MODE,
  }),
  spanProcessor: new BatchSpanProcessor(
    new OTLPTraceExporter({ url: '/otel/v1/traces' }) // 走同域 Collector，避免 CORS
  ),
});

provider.register({ contextManager: new ZoneContextManager() });
registerInstrumentations({
  instrumentations: [
    new XMLHttpRequestInstrumentation({ propagateTraceHeaderCorsUrls: /.*/ }),
    new FetchInstrumentation({ propagateTraceHeaderCorsUrls: /.*/ }),
  ],
});

export const tracer = provider.getTracer('meeting-client');

// meeting.ts
export async function joinMeeting(meetingId: string, token: string) {
  return tracer.startActiveSpan('client.join_meeting', {
    attributes: {
      'meeting.id': meetingId,
      'user.id': getUserId(),
      'device.model': navigator.userAgent,
      'network.type': (navigator as any).connection?.effectiveType ?? 'unknown',
    },
  }, async (span) => {
    try {
      // 1. 信令鉴权
      await tracer.startActiveSpan('signaling.auth', async (s) => {
        const res = await api.auth(token);
        s.setAttribute('auth.result', res.ok ? 'success' : 'fail');
        s.end();
        if (!res.ok) throw new Error('auth failed');
      });

      // 2. SDP 协商
      await negotiate(span); // 透传 traceparent 到 Header

      span.setStatus({ code: SpanStatusCode.OK });
    } catch (e) {
      span.recordException(e);
      span.setStatus({ code: SpanStatusCode.ERROR, message: String(e) });
      throw e;
    } finally {
      span.end();
    }
  });
}

// main.go
import (
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    "go.opentelemetry.io/otel/propagation"
    "go.opentelemetry.io/otel/sdk/resource"
    "go.opentelemetry.io/otel/sdk/trace"
    semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
)

func initTracer() *trace.TracerProvider {
    exp, _ := otlptracegrpc.New(context.Background(),
        otlptracegrpc.WithInsecure(),
        otlptracegrpc.WithEndpoint("otel-collector:4317"),
    )
    tp := trace.NewTracerProvider(
        trace.WithBatcher(exp),
        trace.WithResource(resource.NewWithAttributes(
            semconv.SchemaURL,
            semconv.ServiceName("meeting-signaling"),
            semconv.DeploymentEnvironment("prod"),
        )),
    )
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, propagation.Baggage{},
    ))
    return tp
}

// 中间件自动提取/注入 traceparent
func TraceMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        ctx := otel.GetTextMapPropagator().Extract(r.Context(), propagation.HeaderCarrier(r.Header))
        tr := otel.Tracer("meeting-signaling")
        ctx, span := tr.Start(ctx, "http.request",
            trace.WithAttributes(
                semconv.HTTPMethodKey.String(r.Method),
                semconv.HTTPTargetKey.String(r.URL.Path),
            ),
        )
        defer span.End()
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}

// otel_init.cpp
#include <opentelemetry/exporters/otlp/otlp_grpc_exporter.h>
#include <opentelemetry/sdk/trace/batch_span_processor.h>
#include <opentelemetry/sdk/trace/tracer_provider.h>

std::shared_ptr<opentelemetry::trace::TracerProvider> InitTracer() {
    auto exporter = std::unique_ptr<opentelemetry::exporter::otlp::OtlpGrpcExporter>(
        new opentelemetry::exporter::otlp::OtlpGrpcExporter(
            opentelemetry::exporter::otlp::OtlpGrpcExporterOptions{
                .endpoint = "otel-collector:4317",
                .use_ssl_credentials = false,
            }
        )
    );
    auto processor = std::unique_ptr<opentelemetry::sdk::trace::BatchSpanProcessor>(
        new opentelemetry::sdk::trace::BatchSpanProcessor(
            std::move(exporter),
            opentelemetry::sdk::trace::BatchSpanProcessorOptions{
                .max_queue_size = 8192,
                .schedule_delay_millis = 500,
                .max_export_batch_size = 1024,
            }
        )
    );
    auto provider = std::shared_ptr<opentelemetry::sdk::trace::TracerProvider>(
        new opentelemetry::sdk::trace::TracerProvider(std::move(processor))
    );
    opentelemetry::trace::Provider::SetTracerProvider(provider);
    return provider;
}

void SFUSession::ForwardPacket(const RtpPacket& pkt, const Participant& to) {
    auto tracer = opentelemetry::trace::Provider::GetTracerProvider()->GetTracer("sfu");
    auto span = tracer->StartSpan("sfu.forward",
        { {"rtc.stream.id", to.stream_id},
          {"rtc.target.participant_id", to.id},
          {"rtc.packet.ssrc", pkt.ssrc},
          {"rtc.packet.seq", pkt.seq_num} });
    // 记录丢包率（每秒聚合一次写 Event）
    if (should_report_loss_) {
        span->AddEvent("packet.loss", {
            {"rtc.loss_rate", current_loss_rate_},
            {"rtc.rtt_ms", current_rtt_}
        });
    }
    span->End();
}

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317
      http:
        endpoint: 0.0.0.0:4318

processors:
  batch:
    timeout: 1s
    send_batch_max_size: 1024
  tail_sampling:
    decision_wait: 30s
    num_traces: 50000
    expected_new_traces_per_sec: 20000
    policies:
      - name: errors
        type: status_code
        status_code: {status_codes: [ERROR]}
      - name: high_latency
        type: latency
        latency: {threshold_ms: 500}
      - name: probabilistic
        type: probabilistic
        probabilistic: {sampling_percentage: 1.0}

exporters:
  tempo:
    endpoint: tempo:4317
    tls:
      insecure: true
  prometheus:
    endpoint: 0.0.0.0:8889

service:
  pipelines:
    traces:
      receivers: [otlp]
      processors: [batch, tail_sampling]
      exporters: [tempo]
    metrics:
      receivers: [otlp]
      processors: [batch]
      exporters: [prometheus]

groups:
- name: meeting-rtc-alerts
  rules:
  - alert: MeetingJoinFailureRateHigh
    expr: |
      sum(rate(span_status_code{span_name="client.join_meeting",status_code="ERROR"}[5m]))
      /
      sum(rate(span_status_code{span_name="client.join_meeting"}[5m])) > 0.02
    for: 3m
    labels:
      severity: critical
    annotations:
      summary: "入会失败率超过 2%"
      runbook_url: "https://wiki.example.com/runbook/meeting-join-failure"

  - alert: EndToEndLatencyP99High
    expr: |
      histogram_quantile(0.99,
        sum(rate(trace_span_duration_seconds_bucket{span_name=~"client.join_meeting|sfu.forward"}[5m])) by (le)
      ) > 2
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "端到端 P99 延迟超过 2s"

processors:
  # 为关键指标自动注入 Exemplar
  spanmetrics:
    metrics_exporter: prometheus
    latency_histogram_buckets: [50, 100, 200, 500, 1000, 2000, 5000] # ms
    dimensions:
      - name: meeting.id
        default: "unknown"
      - name: rtc.stream.kind
        default: "unknown"
  # 关键：开启 Exemplar 支持
  batch:
    send_batch_max_size: 1024
exporters:
  prometheus:
    endpoint: "0.0.0.0:8889"
    enable_open_metrics: true
    # 需配合 Prometheus 启用 --enable-feature=exemplar-storage

// Go 统一日志封装（zap + otelzap）
import "go.opentelemetry.io/contrib/bridges/otelzap"

func NewLogger() *zap.Logger {
    core := otelzap.NewCore("meeting-signaling", zapcore.InfoLevel)
    return zap.New(core, zap.AddCaller(), zap.AddStacktrace(zap.ErrorLevel))
}

// 业务代码无感知
logger.Info("ICE candidate gathered",
    zap.String("candidate_type", "srflx"),
    zap.String("network_path", "wifi->4g"),
) // 自动带上 trace_id/span_id

# loki-config.yaml
limits_config:
  # 仅对高基数字段做流式索引，避免写入放大
  structured_metadata:
    - trace_id
    - span_id
    - meeting.id
    - service.name

# otel-collector-config.yaml (扩展)
processors:
  # 1. 字段级脱敏/哈希/删除
  attributes/pii:
    actions:
      - key: "user.id"
        action: hash # SHA256 单向哈希，保留关联性不泄露原值
      - key: "user.ip"
        action: mask # 保留前缀 192.168.xxx.xxx
        mask: "xxx.xxx"
      - key: "device.idfa"
        action: delete # 严禁上报
      - key: "meeting.title"
        action: delete # 业务敏感内容不入 Trace
      - key: "recording.file_path"
        action: replace
        replace_pattern: "(.*)"
        replace_with: "[REDACTED]"

  # 2. 正则脱敏 Body/Event 内容（如 SDP 中的 IP）
  transform/sdp_scrub:
    trace_statements:
      - context: span
        statements:
          - set(attributes["rtc.sdp.raw"], replace_all_patterns(attributes["rtc.sdp.raw"], "(?m)^c=IN IP4 \d+\.\d+\.\d+\.\d+$", "c=IN IP4 [REDACTED]"))
          - set(attributes["rtc.sdp.raw"], replace_all_patterns(attributes["rtc.sdp.raw"], "a=candidate:.*? \d+\.\d+\.\d+\.\d+", "a=candidate:... [REDACTED]"))

  # 3. 租户数据隔离（SaaS 场景）
  resource/tenant_isolation:
    attributes:
      - key: "tenant.id"
        action: upsert
        from_attribute: "http.request.header.x-tenant-id" # 网关层注入
    # 后续导出器按 tenant.id 路由到不同 Tempo 实例/存储桶

service:
  pipelines:
    traces:
      processors: [memory_limiter, batch, attributes/pii, transform/sdp_scrub, resource/tenant_isolation, tail_sampling]
      exporters: [tempo]

// otel-worker.ts (运行在 Web Worker)
import { WebTracerProvider, BatchSpanProcessor } from '@opentelemetry/sdk-trace-web';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { parentPort } from 'worker_threads'; // 或使用 Comlink 通信

const provider = new WebTracerProvider({...});
provider.addSpanProcessor(new BatchSpanProcessor(
  new OTLPTraceExporter({ url: '/otel/v1/traces' }),
  { scheduledDelayMillis: 2000, maxExportBatchSize: 512 } // 聚合更久，省电
));

// 接收主线程发来的 Span 数据
parentPort?.on('message', (spanData) => {
  // 在 Worker 侧重建 Span 并导出
  const span = tracer.startSpan(spanData.name, { attributes: spanData.attrs });
  span.end(spanData.endTime);
});

// main-thread.ts
worker.postMessage({
  name: 'client.ice_gathering',
  attrs: { 'ice.state': 'gathering', 'ice.candidates': 3 },
  endTime: performance.now()
});

// Android (OpenTelemetry Android + Room DB)
class OfflineSpanExporter(private val db: SpanDao) : SpanExporter {
    override fun `export`(spans: Collection<SpanData>): CompletableResultCode {
        // 1. 判断网络/电量
        if (!NetworkUtil.isWifi(context) || BatteryUtil.isLowPowerMode(context)) {
            db.insertAll(spans.map { toEntity(it) }) // 存本地 SQLite
            return CompletableResultCode.ofSuccess()
        }
        // 2. 优先上报本地积压
        db.getUnsynced(500).forEach { realExporter.export(it) }
        db.markSynced()
        // 3. 上报当前
        realExporter.export(spans)
        return CompletableResultCode.ofSuccess()
    }
}

CREATE TABLE rtc_packet_events (
    ts DateTime64(3),           -- 精确到毫秒
    trace_id String,            -- 关联 Trace
    ssrc UInt32,
    seq UInt16,
    direction Enum('in','out'),
    payload_size UInt16,
    rtt_ms UInt16,
    jitter_ms Float32,
    loss_rate Float32,          -- 滑动窗口 1s
    codec String,
    layer UInt8,                -- Simulcast 层
    cpu_usage Float32           -- 采样时进程 CPU
) ENGINE = MergeTree()
PARTITION BY toDate(ts)
ORDER BY (trace_id, ssrc, ts)
TTL ts + INTERVAL 7 DAY DELETE; -- 热数据 7 天，冷数据下沉 S3

# test_otel_instrumentation.py (pytest)
import pytest
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor, ConsoleSpanExporter
from myapp.signaling import handle_join_request

@pytest.fixture
def tracer_provider():
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(ConsoleSpanExporter()))
    return provider

def test_join_meeting_span_attributes(tracer_provider):
    with tracer_provider.get_tracer("test").start_as_current_span("test"):
        handle_join_request(meeting_id="m-123", user_id="u-456")
    
    # 从内存导出器拿 Span 断言
    spans = memory_exporter.get_finished_spans()
    join_span = next(s for s in spans if s.name == "client.join_meeting")
    
    # 强制约束：核心属性必须存在
    assert join_span.attributes.get("meeting.id") == "m-123"
    assert join_span.attributes.get("user.id") == "u-456"
    assert "device.model" in join_span.attributes
    assert join_span.status.status_code == StatusCode.OK

# .github/workflows/otel-perf.yml
- name: Compare Trace Latency Baseline
  run: |
    python scripts/compare_baseline.py 
      --current ${{ github.sha }} 
      --baseline main 
      --metric trace_span_duration_seconds_bucket 
      --filter 'span_name="sfu.forward"' 
      --threshold-p99 1.2 # P99 不得恶化 20%

# Collector 多租户处理器链
processors:
  # 1. 识别租户
  routing/tenant:
    table:
      - attribute: "tenant.id"
        exporter: "tempo-tenant-${value}" # 动态路由

  # 2. 租户级自适应采样（保护大租户不挤占小租户）
  tail_sampling/per_tenant:
    policy_cfg:
      - name: tenant_policy
        type: string_attribute
        string_attribute:
          key: "tenant.id"
          values:
            "enterprise-a": {sampling_percentage: 10} # 大客户全采
            "default": {sampling_percentage: 1}       # 长尾客户 1%

# tempo.yaml
compactor:
  compaction_window: 24h
  block_retention: 7d # 本地块保留
  retention: 90d      # S3 总保留
  # 开启下采样：将原始 Trace 聚合为 Service Graph Metrics
  downsampling:
    enabled: true
    interval: 1h

# 1. Collector 资源限制 (K8s Deployment)
resources:
  limits:
    cpu: "2000m"
    memory: "4Gi"
  requests:
    cpu: "500m"
    memory: "1Gi"
# 必配 memory_limiter processor
processors:
  memory_limiter:
    check_interval: 1s
    limit_mib: 3500 # 留 500Mi 给 Go runtime
    spike_limit_mib: 512

# 2. Tempo 索引优化
# - 仅索引低基数: service.name, span.name, status.code, deployment.environment
# - 高基数 (meeting.id, user.id, trace.id) 仅存储在 Block/Parquet，不建倒排索引
# - 启用 Bloom Filter: bloom_filter.enabled: true

# 3. 客户端采样率动态下发 (Remote Config)
# Firebase Remote Config / 自研配置中心
# key: otel_sampling_rate_web
# value: {"default": 0.01, "error": 1.0, "vip_users": 0.5}

# 4. 关键告警静默窗口 (避免发版风暴)
# PrometheusRule: inhibition_rules 抑制 "DeploymentRolling" 期间的 "HighLatency" 告警