[客户端 SDK] → [Sidecar/DaemonSet Collector] → [Central Collector Cluster] → [存储后端] → [Grafana 看板]
       ↑                                              ↑
[网关/Ingress 自动埋点]                      [尾部采样/脱敏/聚合]

// Go 示例：注入会议维度属性
span.SetAttributes(
    attribute.String("vc.meeting.id", meetingID),
    attribute.String("vc.room.type", roomType),        // "small"/"large"/"webinar"
    attribute.Int("vc.participant.count", participantCount),
    attribute.Bool("vc.is.recording", isRecording),
    attribute.String("vc.codec.video", "H264/VP8/AV1"),
)

// 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 { DocumentLoadInstrumentation } from '@opentelemetry/instrumentation-document-load';
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]: 'vc-web-client',
    [SemanticResourceAttributes.SERVICE_VERSION]: '1.2.3',
    [SemanticResourceAttributes.DEPLOYMENT_ENVIRONMENT]: 'production',
  }),
});

const exporter = new OTLPTraceExporter({
  url: 'https://otel-collector.example.com/v1/traces', // 指向 Sidecar/网关
  headers: { 'Authorization': 'Bearer <token>' },
});

provider.addSpanProcessor(new BatchSpanProcessor(exporter));
provider.register({ contextManager: new ZoneContextManager() });

// 自动采集页面加载、XHR、Fetch
registerInstrumentations({
  instrumentations: [
    new DocumentLoadInstrumentation(),
    new XmlHttpRequestInstrumentation({ propagateTraceHeaderCorsUrls: /.*/ }),
    new FetchInstrumentation({ propagateTraceHeaderCorsUrls: /.*/ }),
  ],
});

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

// meeting-service.ts
import { tracer, SpanStatusCode } from './otel-setup';

export async function joinMeeting(meetingId: string, userId: string) {
  return tracer.startActiveSpan('meeting.client.join', {
    attributes: {
      'vc.meeting.id': meetingId,
      'vc.user.id': userId,
      'vc.network.type': navigator.connection?.effectiveType || 'unknown',
    },
  }, async (span) => {
    try {
      const resp = await signalingClient.join(meetingId, userId);
      span.setAttribute('vc.session.id', resp.sessionId);
      span.setStatus({ code: SpanStatusCode.OK });
      return resp;
    } catch (err) {
      span.recordException(err);
      span.setStatus({ code: SpanStatusCode.ERROR, message: err.message });
      throw err;
    } finally {
      span.end();
    }
  });
}

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

func initTracer() *sdktrace.TracerProvider {
    exporter, _ := otlptracegrpc.New(context.Background(),
        otlptracegrpc.WithEndpoint("otel-collector:4317"),
        otlptracegrpc.WithInsecure(),
    )
    res, _ := resource.New(context.Background(),
        resource.WithAttributes(
            semconv.ServiceName("signaling-service"),
            semconv.ServiceVersion("v2.1.0"),
            semconv.DeploymentEnvironment("prod"),
        ),
    )
    tp := sdktrace.NewTracerProvider(
        sdktrace.WithBatcher(exporter),
        sdktrace.WithResource(res),
        sdktrace.WithSampler(sdktrace.ParentBased(sdktrace.TraceIDRatioBased(0.01))), // 头部 1%
    )
    otel.SetTracerProvider(tp)
    otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
        propagation.TraceContext{}, propagation.Baggage{},
    ))
    return tp
}

// 信令处理器中提取上下文
func HandleOffer(ctx context.Context, req *OfferRequest) (*OfferResponse, error) {
    // 从 HTTP Header / gRPC Metadata 自动提取 Trace Context
    ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.HeaderCarrier(req.Headers))
    
    ctx, span := tracer.Start(ctx, "signaling.server.handle_offer",
        trace.WithAttributes(
            attribute.String("vc.meeting.id", req.MeetingID),
            attribute.String("vc.session.id", req.SessionID),
            attribute.String("vc.sdp.type", req.SDP.Type),
        ),
    )
    defer span.End()

    // 业务逻辑...
    if err := validateSDP(req.SDP); err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, err.Error())
        return nil, err
    }
    return &OfferResponse{Answer: generateAnswer()}, nil
}

# otel-collector-config.yaml
processors:
  attributes/redact:
    actions:
      - key: "net.peer.ip"
        action: "hash"          # SHA256 脱敏，保留关联性
      - key: "vc.user.id"
        action: "hash"
      - key: "http.request.header.authorization"
        action: "delete"        # 直接删除 Token
      - key: "vc.recording.path"
        action: "replace"
        pattern: "(.*)"
        replacement: "[REDACTED]"

processors:
  spanmetrics:
    metrics_exporter: prometheus
    latency_histogram_buckets: [50, 100, 200, 500, 1000, 2000] # ms
    dimensions:
      - name: "vc.meeting.id"
      - name: "vc.track.kind"
      - name: "vc.codec.video"
  filter/drop_high_cardinality:
    spans:
      include:
        match_type: "regexp"
        attributes:
          - key: "vc.meeting.id"
            value: ".*"   # 仅保留带会议 ID 的 Span

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"
          values: ["true"]
      - name: "slow-policy"
        type: "latency"
        latency:
          threshold_ms: 500
      - name: "large-meeting-policy"
        type: "numeric_attribute"
        numeric_attribute:
          key: "vc.participant.count"
          min_value: 50
      - name: "fallback-policy"
        type: "probabilistic"
        probabilistic:
          sampling_percentage: 0.1

groups:
- name: vc-observability
  rules:
  - alert: MeetingJoinLatencyP95High
    expr: histogram_quantile(0.95, sum(rate(vc_meeting_join_duration_seconds_bucket[5m])) by (le)) > 3
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "会议加入 P95 延迟超过 3 秒"
      description: "当前 P95={{ $value }}s，建议检查信令服务与 SFU 扩容状态"

  - alert: SFUPacketLossRateHigh
    expr: sum(rate(sfu_packet_lost_total[1m])) by (instance) / sum(rate(sfu_packet_total[1m])) by (instance) > 0.05
    for: 1m
    labels:
      severity: critical
    annotations:
      summary: "SFU 实例 {{ $labels.instance }} 丢包率超 5%"
      description: "可能存在带宽拥塞或编码器异常，请排查媒体节点网络与 CPU"

  - alert: RecordingTaskQueueBacklog
    expr: recording_task_queue_length > 100
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "录制任务积压超过 100 个"
      description: "录制工作节点可能不足或存储写入慢，建议扩容或排查存储 IO"

// client-qos-reporter.ts
import { metrics, DiagConsoleLogger, DiagLogLevel, context } from '@opentelemetry/api';
import { MeterProvider, PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics';
import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-http';

// 1. 初始化 Metric Provider (复用 Trace 的 Resource)
const meterProvider = new MeterProvider({
  resource: traceProvider.resource,
});
metrics.setGlobalMeterProvider(meterProvider);

const meter = meterProvider.getMeter('vc-client-qos', '1.0.0');

// 2. 定义核心指标仪器
const rttHistogram = meter.createHistogram('vc.webrtc.rtt', { unit: 'ms', description: 'WebRTC RTT' });
const packetLossGauge = meter.createObservableGauge('vc.webrtc.packet_loss_rate', { description: 'Packet Loss Rate' });
const jitterHistogram = meter.createHistogram('vc.webrtc.jitter', { unit: 'ms' });
const frameRateGauge = meter.createObservableGauge('vc.webrtc.frame_rate', { unit: 'fps' });
const resolutionGauge = meter.createObservableGauge('vc.webrtc.resolution', { unit: 'pixels' });

// 3. 绑定当前会议上下文 (从 Trace Context 获取 meeting_id)
let currentMeetingId: string | null = null;
export function bindMeetingContext(meetingId: string) { currentMeetingId = meetingId; }

// 4. 定期采集 getStats 并上报 (建议 10s/次)
setInterval(async () => {
  if (!currentMeetingId || !pc) return;
  const stats = await pc.getStats();
  let totalLost = 0, totalPackets = 0, sumRtt = 0, rttCount = 0;
  let maxJitter = 0, currentFps = 0, currentWidth = 0, currentHeight = 0;

  stats.forEach(report => {
    if (report.type === 'candidate-pair' && report.nominated) {
      sumRtt += (report.currentRoundTripTime || 0) * 1000; // s -> ms
      rttCount++;
    }
    if (report.type === 'inbound-rtp' && report.kind === 'video') {
      totalLost += report.packetsLost || 0;
      totalPackets += report.packetsReceived || 0;
      maxJitter = Math.max(maxJitter, (report.jitter || 0) * 1000);
      currentFps = report.framesPerSecond || 0;
      currentWidth = report.frameWidth || 0;
      currentHeight = report.frameHeight || 0;
    }
  });

  const attrs = { 'vc.meeting.id': currentMeetingId, 'vc.user.id': getUserId() };
  
  if (rttCount > 0) rttHistogram.record(sumRtt / rttCount, attrs);
  if (totalPackets > 0) packetLossGauge.addCallback((obs) => obs.observe(totalLost / totalPackets, attrs));
  if (maxJitter > 0) jitterHistogram.record(maxJitter, attrs);
  if (currentFps > 0) frameRateGauge.addCallback((obs) => obs.observe(currentFps, attrs));
  if (currentWidth > 0) resolutionGauge.addCallback((obs) => obs.observe(currentWidth * currentHeight, attrs));
}, 10_000);

// 信令网关/服务端：统一拦截器注入与提取
func TraceContextInterceptor() grpc.UnaryServerInterceptor {
    return func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
        // 1. 从 Metadata 提取
        md, _ := metadata.FromIncomingContext(ctx)
        ctx = otel.GetTextMapPropagator().Extract(ctx, propagation.MapCarrier(md))
        
        // 2. 业务属性注入 Baggage (供下游自动获取)
        if meetingID := getMeetingIDFromReq(req); meetingID != "" {
            ctx = baggage.ContextWithValues(ctx,
                attribute.String("vc.meeting.id", meetingID),
                attribute.String("vc.participant.id", getParticipantID(req)),
            )
        }
        return handler(ctx, req)
    }
}

# otel-collector-config.yaml 扩展：使用 k8s 下发 ConfigMap 实现热更新
processors:
  tailsampling:
    policies:
      - name: "adaptive-large-meeting"
        type: "and"
        and:
          sub_policy_cfg:
            - name: "large-meeting-filter"
              type: "numeric_attribute"
              numeric_attribute:
                key: "vc.participant.count"
                min_value: 200
            - name: "adaptive-rate"
              type: "probabilistic"
              probabilistic:
                # 采样率由外部 Controller 根据集群负载动态下发 (0.1% ~ 10%)
                sampling_percentage: ${env:ADAPTIVE_SAMPLING_RATE:-1.0}

# sampling-controller 每 30s 运行一次
def adjust_sampling_rate():
    collector_cpu = get_prom_metric('otelcol_process_cpu_seconds_total')
    queue_size = get_prom_metric('otelcol_receiver_queue_size')
    error_rate = get_prom_metric('otelcol_exporter_send_failed_spans')
    
    base_rate = 1.0  # 默认 1%
    if collector_cpu > 0.7 or queue_size > 10000:
        base_rate = max(0.1, base_rate * 0.5)  # 降级
    elif collector_cpu < 0.3 and error_rate == 0:
        base_rate = min(10.0, base_rate * 1.2) # 升级
    
    # 核心业务保护：付费大客户、直播间强制全采
    vip_meetings = get_vip_meeting_ids_from_redis()
    push_configmap({"ADAPTIVE_SAMPLING_RATE": base_rate, "VIP_MEETINGS": vip_meetings})

# tempo.yaml
compactor:
  compaction:
    block_retention: 72h # 热数据 3 天
  retention: 72h
storage:
  trace:
    backend: s3
    s3:
      bucket: tempo-traces
    # 利用 Tempo 的 Native TSDB (v2.0+) 自动构建倒排索引
    bloom_filter_false_positive: 0.05
  # 定期将超过 7 天的 Block 通过 Compactor 重写为无索引 Parquet 迁移至冷存储桶

processors:
  # 1. 字段级加密 (需自定义 Processor 或使用 OTTL 转换 + 外部 KMS 调用)
  transform/encrypt_pii:
    trace_statements:
      - context: span
        statements:
          - set(attributes["vc.user.id_encrypted"], encrypt(attributes["vc.user.id"], "kms-key-arn")) where attributes["vc.user.id"] != nil
          - delete_key(attributes["vc.user.id"])
  
  # 2. 敏感词过滤 (防止业务代码误埋点写入 Token/密码)
  filter/sensitive_keywords:
    spans:
      exclude:
        match_type: regexp
        attributes:
          - key: ".*"
            value: "(?i)(password|token|secret|key|authorization|cookie)"

  # 3. 审计日志生成 (记录谁在何时查询了什么敏感 Trace)
  # 需配合后端存储 (Tempo/Jaeger) 的 Query Frontend 审计插件实现

# vc-semconv.yaml (片段)
attributes:
  - id: vc.meeting.id
    type: string
    requirement_level: required
    stability: stable
    brief: "会议唯一标识符，格式 UUID v4"
    examples: ["a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8"]
  - id: vc.room.type
    type: string
    requirement_level: required
    allowed_values: ["small", "large", "webinar", "breakout"]
    stability: stable
  - id: vc.codec.video
    type: string
    requirement_level: recommended
    allowed_values: ["H264", "VP8", "VP9", "AV1", "H265"]

// tracing_test.go
func TestJoinMeeting_SpanAttributes(t *testing.T) {
    // 1. 启动 In-Memory Exporter
    exp := otlptrace.NewInMemoryExporter()
    tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exp))
    defer tp.Shutdown(context.Background())
    
    // 2. 执行业务逻辑
    svc := NewMeetingService(tp.Tracer("test"))
    _, err := svc.JoinMeeting(context.Background(), "meeting-123", "user-456")
    require.NoError(t, err)
    
    // 3. 断言 Span
    spans := exp.GetSpans()
    require.Len(t, spans, 1)
    span := spans[0]
    
    // 契约校验
    assert.Equal(t, "meeting.client.join", span.Name)
    assert.Equal(t, "meeting-123", span.Attributes["vc.meeting.id"])
    assert.Contains(t, []string{"small", "large", "webinar"}, span.Attributes["vc.room.type"])
    assert.NotEmpty(t, span.Attributes["vc.network.type"])
    
    // 语义约定校验
    assert.Equal(t, "meeting-service", span.Resource.Attributes()[semconv.ServiceNameKey])
}

# trace_validator.py (Kubernetes Job/CronJob)
from opentelemetry.proto.trace.v1.trace_pb2 import ResourceSpans
import rules_engine

RULES = [
    {"name": "meeting_join_must_have_meeting_id", "query": "span.name == 'meeting.client.join' && !has(attributes['vc.meeting.id'])", "severity": "ERROR"},
    {"name": "error_span_must_have_exception", "query": "span.status.code == ERROR && !has(attributes['exception.type'])", "severity": "WARN"},
    {"name": "sfu_span_must_have_codec", "query": "span.name starts_with 'sfu.' && !has(attributes['vc.codec.video'])", "severity": "WARN"},
]

def validate_batch(batch: ResourceSpans):
    violations = []
    for rs in batch.resource_spans:
        for ss in rs.scope_spans:
            for span in ss.spans:
                ctx = build_context(span, rs.resource)
                for rule in RULES:
                    if rules_engine.eval(rule["query"], ctx):
                        violations.append({"rule": rule["name"], "trace_id": span.trace_id.hex(), "span_id": span.span_id.hex()})
    if violations:
        send_alert_to_lark(violations) # 发送飞书/钉钉告警给对应服务 Owner

# Istio Telemetry API (v1beta1) 配置
apiVersion: telemetry.istio.io/v1alpha1
kind: Telemetry
metadata:
  name: vc-mesh-telemetry
  namespace: istio-system
spec:
  selector:
    matchLabels:
      app: signaling-gateway # 仅对网关生效，避免 Sidecar 资源争抢
  tracing:
  - providers:
    - name: otel-collector
    randomSamplingPercentage: 1.0 # 网关层 1% 头部采样
    customTags:
    - header:
        name: "x-vc-meeting-id"
        tagName: "vc.meeting.id"
    - header:
        name: "x-vc-user-id"
        tagName: "vc.user.id"
  metrics:
  - providers:
    - name: prometheus
    overrides:
    - match:
        metric: REQUEST_DURATION
      tagOverrides:
        vc_meeting_id:
          value: "request.header.x-vc-meeting-id"

[SFU Edge] --> [Local OTel Collector (Sidecar/DaemonSet)] --> [本地 SQLite/Buffer] --> [定时/网络恢复时批量上传 Central Collector]

# edge-collector-config.yaml
exporters:
  otlp/central:
    endpoint: "central-collector.svc:4317"
    retry_on_failure:
      enabled: true
      initial_interval: 5s
      max_interval: 300s
      max_elapsed_time: 3600s # 1 小时内持续重试
  # 本地兜底存储 (防止中央长时间不可用导致内存 OOM)
  file/traces:
    path: /var/lib/otel/traces-%Y%m%d.json
    rotation:
      max_megabytes: 500
      max_days: 3

processors:
  batch:
    timeout: 30s
    send_batch_max_size: 1024
  memory_limiter:
    check_interval: 1s
    limit_mib: 512 # 边缘节点资源受限，严格限制内存
    spike_limit_mib: 128

extensions:
  storage/sqlite: # 官方 storage 扩展 (v0.90+)
    path: /var/lib/otel/queue.db
  # 配合 persistent_queue 实现断点续传
service:
  extensions: [storage/sqlite]
  pipelines:
    traces:
      receivers: [otlp]
      processors: [memory_limiter, batch, tailsampling]
      exporters: [otlp/central, file/traces] # 双写：实时发 + 本地落盘

ROI = (故障前平均 MTTR - 接入后平均 MTTR) × 单分钟故障损失 × 年故障次数 / 年可观测性总成本