OpenTelemetryトレース相関実戦：5つのパターンでエンドツーエンド分散トレーシングを構築

OpenTelemetryトレース相関：マイクロサービスオブザーバビリティの中核の紐帯

マイクロサービスアーキテクチャでは、1つのユーザーリクエストが10以上のサービスにまたがる可能性があり、ログは散在、トレースは断絶、ルート原因分析は干し草の山から針を探すような状態。OpenTelemetryトレース相関はW3C TraceContext標準でクロスサービスTrace ID伝播を実現し、Baggageでビジネスコンテキストを伝達、分散トレーシングを「見える」から「相関できる」へアップグレード。2026年、OpenTelemetryはCNCF卒業プロジェクトとなり、W3C TraceContext仕様はすべての主要フレームワークでサポートされています。

本記事では5つのコアパターンから、トレース伝播→クロスサービス相関→Baggage渡し→非同期トレース→マルチシグナル相関のフルパイプライン実戦を解説します。

---

コア概念

概念	説明
Trace	1回のリクエストの完全な呼び出しチェーン
Span	Trace内の単一操作ユニット
TraceContext	W3C標準、traceparent/tracestateヘッダー
Baggage	クロスサービス伝播のビジネスコンテキストキーバリューペア
Propagator	クロスプロセスでトレースコンテキストを伝播するコンポーネント
Span Link	異なるTraceを関連付けるSpan
Sampling	サンプリング戦略、収集量を制御
Collector	OTel収集ゲートウェイ、テレメトリデータの受信と転送

---

問題分析：トレース相関の5つの課題

1. クロスプロトコル伝播：HTTP/gRPC/メッセージキューのコンテキスト伝播方法が異なる
2. 非同期トレース断絶：メッセージキュー、定期タスクによるTrace断絶
3. Baggage乱用：過剰なビジネスデータ伝播によるヘッダー膨張
4. サンプリング損失：低サンプリング率で重要なトレースが破棄
5. マルチシグナル相関：Trace/Metric/Logの3信号相関が困難

---

ステップバイステップ：5つのトレース相関パターン

パターン1：W3C TraceContext伝播

// Node.js - HTTPサービス間トレース伝播
import { NodeSDK } from '@opentelemetry/sdk-node';
import { getNodeAutoInstrumentations } from '@opentelemetry/auto-instrumentations-node';
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-http';
import { Resource } from '@opentelemetry/resources';
import { ATTR_SERVICE_NAME } from '@opentelemetry/semantic-conventions';
import { W3CTraceContextPropagator } from '@opentelemetry/core';

const sdk = new NodeSDK({
  resource: new Resource({
    [ATTR_SERVICE_NAME]: 'order-service',
  }),
  traceExporter: new OTLPTraceExporter({
    url: 'http://collector:4318/v1/traces',
  }),
  instrumentations: [getNodeAutoInstrumentations()],
  textMapPropagator: new W3CTraceContextPropagator(),
});

sdk.start();

# Python - 手動トレースコンテキスト伝播
from opentelemetry import trace, context
from opentelemetry.propagate import inject, extract
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import requests

provider = TracerProvider()
provider.add_span_processor(
    BatchSpanProcessor(OTLPSpanExporter(endpoint='collector:4317'))
)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer('order-service')

def call_payment_service(order_id: str):
    with tracer.start_as_current_span('call-payment-service') as span:
        span.set_attribute('order.id', order_id)

        headers = {}
        inject(headers)

        response = requests.post(
            'http://payment-service/api/charge',
            json={'order_id': order_id},
            headers=headers,
        )
        return response.json()

# 受信側コンテキスト抽出
from flask import Flask, request

app = Flask(__name__)

@app.route('/api/charge', methods=['POST'])
def charge():
    ctx = extract(request.headers)
    token = context.attach(ctx)
    try:
        with tracer.start_as_current_span('process-charge') as span:
            span.set_attribute('payment.amount', 100)
            return {'status': 'ok'}
    finally:
        context.detach(token)

パターン2：gRPCクロスサービストレース相関

// Go - gRPCインターセプター自動伝播
package main

import (
    "context"
    "google.golang.org/grpc"
    "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"
)

func initTracer() func() {
    exporter, _ := otlptracegrpc.New(context.Background(),
        otlptracegrpc.WithEndpoint("collector:4317"),
        otlptracegrpc.WithInsecure(),
    )
    provider := sdktrace.NewTracerProvider(sdktrace.WithBatcher(exporter))
    otel.SetTracerProvider(provider)
    return func() { provider.Shutdown(context.Background()) }
}

func main() {
    shutdown := initTracer()
    defer shutdown()

    conn, _ := grpc.Dial("payment-service:50051",
        grpc.WithTransportCredentials(insecure.NewCredentials()),
        grpc.WithStatsHandler(otelgrpc.NewClientHandler()),
    )
    defer conn.Close()

    server := grpc.NewServer(
        grpc.StatsHandler(otelgrpc.NewServerHandler()),
    )
}

パターン3：Baggageビジネスコンテキスト渡し

# Baggageの設定
from opentelemetry import baggage, context

def handle_request(request):
    ctx = baggage.set_baggage('user.id', 'user-123')
    ctx = baggage.set_baggage('tenant.id', 'tenant-456', context=ctx)
    ctx = baggage.set_baggage('request.source', 'mobile', context=ctx)

    token = context.attach(ctx)
    try:
        process_order(request)
    finally:
        context.detach(token)

# 下流サービスでBaggageを読み取り
def process_order(request):
    user_id = baggage.get_baggage('user.id')
    tenant_id = baggage.get_baggage('tenant.id')

    with tracer.start_as_current_span('process-order') as span:
        span.set_attribute('user.id', user_id)
        span.set_attribute('tenant.id', tenant_id)

// TypeScript - Baggage制限と検証
import { baggageEntryMetadataFromString } from '@opentelemetry/api';

const MAX_BAGGAGE_ITEMS = 10;
const MAX_BAGGAGE_VALUE_LENGTH = 4096;

class SafeBaggageManager {
  private items: Map<string, string> = new Map();

  set(key: string, value: string): boolean {
    if (this.items.size >= MAX_BAGGAGE_ITEMS) {
      console.warn(`Baggage items exceed limit: ${MAX_BAGGAGE_ITEMS}`);
      return false;
    }
    if (value.length > MAX_BAGGAGE_VALUE_LENGTH) {
      console.warn(`Baggage value too long for key: ${key}`);
      return false;
    }
    this.items.set(key, value);
    return true;
  }

  get(key: string): string | undefined {
    return this.items.get(key);
  }

  toContext(): Record<string, string> {
    const result: Record<string, string> = {};
    this.items.forEach((value, key) => {
      result[key] = value;
    });
    return result;
  }
}

パターン4：非同期メッセージキュートレース相関

# Kafkaプロデューサー - メッセージヘッダーにトレースコンテキストを注入
from opentelemetry import trace, context
from opentelemetry.propagate import inject
from kafka import KafkaProducer
import json

producer = KafkaProducer(bootstrap_servers='kafka:9092')

def publish_order_event(order_id: str, event_type: str):
    with tracer.start_as_current_span('publish-order-event') as span:
        span.set_attribute('messaging.system', 'kafka')
        span.set_attribute('messaging.destination', 'order-events')
        span.set_attribute('messaging.operation', 'publish')

        headers = {}
        inject(headers)

        kafka_headers = [(k, v.encode()) for k, v in headers.items()]

        producer.send(
            'order-events',
            key=order_id.encode(),
            value=json.dumps({
                'order_id': order_id,
                'event_type': event_type,
            }).encode(),
            headers=kafka_headers,
        )

# Kafkaコンシューマー - トレースコンテキストを抽出
from kafka import KafkaConsumer
from opentelemetry.propagate import extract

consumer = KafkaConsumer(
    'order-events',
    bootstrap_servers='kafka:9092',
)

for message in consumer:
    headers = {k: v.decode() for k, v in message.headers}
    ctx = extract(headers)
    token = context.attach(ctx)
    try:
        with tracer.start_as_current_span('process-order-event') as span:
            span.set_attribute('messaging.system', 'kafka')
            span.set_attribute('messaging.operation', 'process')
            span.set_attribute('messaging.kafka.consumer_group', 'order-processor')

            data = json.loads(message.value.decode())
            handle_event(data)
    finally:
        context.detach(token)

パターン5：Trace/Metric/Logマルチシグナル相関

# 統一Trace IDのログ注入
import logging
from opentelemetry import trace

class TraceFormatter(logging.Formatter):
    def format(self, record):
        span = trace.get_current_span()
        if span.is_recording():
            ctx = span.get_span_context()
            record.trace_id = format(ctx.trace_id, '032x')
            record.span_id = format(ctx.span_id, '016x')
        else:
            record.trace_id = '0' * 32
            record.span_id = '0' * 16
        return super().format(record)

logger = logging.getLogger('order-service')
handler = logging.StreamHandler()
handler.setFormatter(TraceFormatter(
    '%(asctime)s [trace_id=%(trace_id)s span_id=%(span_id)s] %(levelname)s %(message)s'
))
logger.addHandler(handler)

# MetricとTraceの相関
from opentelemetry import metrics

meter = metrics.get_meter('order-service')
order_duration = meter.create_histogram(
    'order.processing.duration',
    unit='ms',
)

def record_order_metric(duration_ms: float):
    span = trace.get_current_span()
    ctx = span.get_span_context()
    order_duration.record(
        duration_ms,
        attributes={
            'trace_id': format(ctx.trace_id, '032x'),
            'service.name': 'order-service',
        },
    )

# Grafana Tempo + Loki + Prometheus相関設定
# tempo-datasource.yaml
apiVersion: 1
datasources:
  - name: Tempo
    type: tempo
    url: http://tempo:3200
    jsonData:
      tracesToMetrics:
        datasourceUid: prometheus
        tags:
          - service.name
        queries:
          - name: 'Request Rate'
            query: 'sum(rate(http_server_request_duration_seconds_count{service="$service"}[5m]))'
      tracesToLogs:
        datasourceUid: loki
        tags: ['service.name']
        filterByTraceID: true
        filterBySpanID: true
      nodeGraph:
        enabled: true

---

よくある落とし穴

落とし穴1：Propagator未設定

# ❌ 間違い：Propagator未設定、クロスサービスTrace IDが伝播されない
# デフォルトでNonePropagatorが使用される可能性

# ✅ 正しい：W3C TraceContextを明示的に設定
from opentelemetry.propagate import set_global_textmap
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

set_global_textmap(TraceContextTextMapPropagator())

落とし穴2：非同期タスクでコンテキスト消失

# ❌ 間違い：非同期タスクでコンテキストが消失
import asyncio

async def process():
    await asyncio.sleep(1)  # コンテキスト消失

# ✅ 正しい：コンテキストを手動で渡す
from opentelemetry import context

async def process():
    ctx = context.get_current()
    await asyncio.sleep(1)
    token = context.attach(ctx)
    try:
        do_work()
    finally:
        context.detach(token)

落とし穴3：Baggageで機密情報を伝播

# ❌ 間違い：Baggageで機密データを伝播
baggage.set_baggage('user.email', 'secret@example.com')

# ✅ 正しい：IDのみ渡し、機密データはデータベースから検索
baggage.set_baggage('user.id', 'user-123')

落とし穴4：サンプリングで重要なトレースが破棄

# ❌ 間違い：固定低サンプリング率
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
sampler = TraceIdRatioBased(0.01)  # 1%サンプリング

# ✅ 正しい：属性ベースのスマートサンプリング
from opentelemetry.sdk.trace.sampling import ParentBased, TraceIdRatioBased
from opentelemetry.sdk.trace.sampling import Sampler, SamplingResult

class ErrorAwareSampler(Sampler):
    def should_sample(self, parent_context, trace_id, name, kind, attributes, links):
        if attributes.get('http.status_code', 200) >= 400:
            return SamplingResult(RECORD_AND_SAMPLE, attributes, links)
        return TraceIdRatioBased(0.1).should_sample(
            parent_context, trace_id, name, kind, attributes, links
        )

落とし穴5：Span Link未使用

# ❌ 間違い：メッセージ消費時に全く新しいTraceを作成、プロデューサーとの相関が消失

# ✅ 正しい：Span Linkでプロデューサーとコンシューマーを相関
from opentelemetry.trace import Link

def process_message(message):
    producer_ctx = extract(message.headers)
    producer_span = trace.get_current_span(producer_ctx)
    link = Link(producer_span.get_span_context())

    with tracer.start_as_current_span(
        'process-message',
        links=[link],
    ) as span:
        handle(message)

---

エラートラブルシューティング

#	エラーメッセージ	原因	解決方法
1	Trace ID not propagated	Propagator未設定	W3CTraceContextPropagatorを設定
2	Context lost in async	非同期タスクでコンテキスト消失	コンテキストを手動でattach/detach
3	Baggage header too large	Baggage項目が多すぎる	項目数と値の長さを制限
4	Span not exported	Exporterの設定エラー	Collectorアドレスとプロトコルを確認
5	Duplicate spans	Providerの重複登録	グローバルで1回だけ初期化すること
6	gRPC trace broken	インターセプター未追加	otelgrpcインターセプターを使用
7	Kafka trace broken	ヘッダーの注入/抽出なし	メッセージヘッダーでコンテキストを伝播
8	Sampling drops errors	サンプリング率が低すぎる	エラー認識サンプラーを使用
9	Log-Trace correlation failed	ログにTrace IDが未注入	TraceFormatterを使用
10	Collector connection refused	Collectorが未起動	Collectorサービスとポートを確認

---

高度な最適化

1. Tail-Based Sampling：完全なTrace結果に基づいてサンプリングを決定、エラートレースを保持
2. Trace→Metricエクスポート：SpanからREDメトリクス（Rate/Error/Duration）を自動生成
3. 自動相関ルール：Grafana TempoがLokiログとPrometheusメトリクスを自動相関
4. Span Metrics Connector：Collector内蔵のSpan→Metric変換
5. アダプティブサンプリング：トラフィックパターンに基づいてサンプリング率を動的調整

---

比較分析

次元	OpenTelemetry	Jaeger	Zipkin	SkyWalking
標準化	W3C/CNCF	独自	独自	独自
多言語	11+	6+	4+	8+
マルチシグナル	Trace+Metric+Log	Trace	Trace	Trace+Metric
Baggage	✅	❌	❌	❌
サンプリング戦略	豊富	基本	基本	豊富
エコシステム統合	最も広い	広い	中	中

---

まとめ：OpenTelemetryトレース相関はマイクロサービスオブザーバビリティの中核の紐帯です。W3C TraceContext標準伝播+Baggageビジネスコンテキスト+Span Linkクロストレース相関の三位一体により、分散トレーシングが「単一トレース可視」から「フルトレース相関」へアップグレード。2026年のOTelの成熟によりマルチシグナル（Trace/Metric/Log）相関が標準機能となり、Tail-Based Samplingとアダプティブサンプリングが本番環境の重要な最適化となります。

---

オンラインツール推奨

• JSONフォーマッター：/ja/json/format
• Hash計算：/ja/encode/hash
• cURL→コード変換：/ja/dev/curl-to-code