OpenTelemetry による分散トレーシング実装 ​

# requirements.txt
opentelemetry-api==1.21.0
opentelemetry-sdk==1.21.0
opentelemetry-instrumentation-fastapi==0.42b0
opentelemetry-instrumentation-requests==0.42b0
opentelemetry-instrumentation-sqlalchemy==0.42b0
opentelemetry-exporter-jaeger-thrift==1.21.0
opentelemetry-propagator-b3==1.21.0

# tracing.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
from opentelemetry.sdk.resources import Resource

def configure_tracing(service_name: str, jaeger_endpoint: str):
    """OpenTelemetry トレーシングの設定"""
    
    # リソース情報の設定
    resource = Resource.create({
        "service.name": service_name,
        "service.version": "1.0.0",
        "deployment.environment": "production"
    })
    
    # TraceProvider の設定
    trace.set_tracer_provider(TracerProvider(resource=resource))
    tracer = trace.get_tracer(__name__)
    
    # Jaeger エクスポーターの設定
    jaeger_exporter = JaegerExporter(
        agent_host_name="localhost",
        agent_port=6831,
    )
    
    # スパンプロセッサーの追加
    span_processor = BatchSpanProcessor(jaeger_exporter)
    trace.get_tracer_provider().add_span_processor(span_processor)
    
    # 自動計装の有効化
    FastAPIInstrumentor.instrument()
    RequestsInstrumentor.instrument()
    SQLAlchemyInstrumentor.instrument()
    
    return tracer

# main.py
from fastapi import FastAPI, Depends
from opentelemetry import trace
from opentelemetry.trace import Status, StatusCode
import requests
import asyncio
from tracing import configure_tracing

# トレーシングの初期化
tracer = configure_tracing("user-service", "http://localhost:14268")

app = FastAPI(title="User Service")

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    """ユーザー情報を取得"""
    
    # 手動でスパンを作成
    with tracer.start_as_current_span("get_user_operation") as span:
        try:
            # スパンに属性を追加
            span.set_attribute("user.id", user_id)
            span.set_attribute("service.operation", "get_user")
            
            # データベースからユーザー情報を取得
            user_data = await fetch_user_from_db(user_id)
            
            if not user_data:
                span.set_status(Status(StatusCode.ERROR, "User not found"))
                raise HTTPException(status_code=404, detail="User not found")
            
            # 外部サービスから追加情報を取得
            preferences = await fetch_user_preferences(user_id)
            
            span.set_attribute("user.preferences_count", len(preferences))
            span.set_status(Status(StatusCode.OK))
            
            return {
                "user": user_data,
                "preferences": preferences
            }
            
        except Exception as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            raise

async def fetch_user_from_db(user_id: int):
    """データベースからユーザー情報を取得"""
    
    with tracer.start_as_current_span("db_query_user") as span:
        span.set_attribute("db.operation", "SELECT")
        span.set_attribute("db.table", "users")
        span.set_attribute("db.user_id", user_id)
        
        # 模擬的なデータベースクエリ
        await asyncio.sleep(0.1)  # DB応答時間をシミュレート
        
        return {
            "id": user_id,
            "name": f"User {user_id}",
            "email": f"user{user_id}@example.com"
        }

async def fetch_user_preferences(user_id: int):
    """外部サービスからユーザー設定を取得"""
    
    with tracer.start_as_current_span("external_api_call") as span:
        span.set_attribute("http.method", "GET")
        span.set_attribute("http.url", f"http://preferences-service/users/{user_id}/preferences")
        
        try:
            # 外部API呼び出し（自動計装により自動的にトレース）
            response = requests.get(
                f"http://preferences-service/api/users/{user_id}/preferences",
                timeout=5
            )
            
            span.set_attribute("http.status_code", response.status_code)
            
            if response.status_code == 200:
                return response.json()
            else:
                span.set_status(Status(StatusCode.ERROR, f"HTTP {response.status_code}"))
                return []
                
        except requests.RequestException as e:
            span.record_exception(e)
            span.set_status(Status(StatusCode.ERROR, str(e)))
            return []

# context_propagation.py
from opentelemetry import trace
from opentelemetry.propagate import inject, extract
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
import requests

def make_traced_request(url: str, method: str = "GET", **kwargs):
    """トレースコンテキストを伝播するHTTPリクエスト"""
    
    # 現在のトレースコンテキストを取得
    headers = kwargs.get("headers", {})
    
    # コンテキストをヘッダーに注入
    inject(headers)
    kwargs["headers"] = headers
    
    with tracer.start_as_current_span(f"http_{method.lower()}") as span:
        span.set_attribute("http.method", method)
        span.set_attribute("http.url", url)
        
        response = requests.request(method, url, **kwargs)
        span.set_attribute("http.status_code", response.status_code)
        
        return response

def extract_trace_context(headers: dict):
    """受信したリクエストからトレースコンテキストを抽出"""
    
    # ヘッダーからコンテキストを抽出
    context = extract(headers)
    
    # 抽出したコンテキストを現在のスパンに設定
    token = trace.set_span_in_context(trace.get_current_span(), context)
    
    return token

from opentelemetry.sdk.trace.sampling import (
    TraceIdRatioBased, 
    ParentBased, 
    Decision
)

# 本番環境では適切なサンプリング率を設定
sampler = ParentBased(
    root=TraceIdRatioBased(rate=0.1)  # 10% のトレースをサンプリング
)

trace.set_tracer_provider(
    TracerProvider(
        resource=resource,
        sampler=sampler
    )
)

import time
from opentelemetry.metrics import get_meter

meter = get_meter(__name__)

# カスタムメトリクスの定義
request_duration = meter.create_histogram(
    name="http_request_duration_seconds",
    description="HTTP request duration",
    unit="s"
)

request_counter = meter.create_counter(
    name="http_requests_total",
    description="Total HTTP requests"
)

def monitor_request_performance(func):
    """リクエストパフォーマンスを監視するデコレーター"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        
        try:
            result = func(*args, **kwargs)
            status = "success"
            return result
        except Exception as e:
            status = "error"
            raise
        finally:
            duration = time.time() - start_time
            
            # メトリクスを記録
            request_duration.record(
                duration,
                {"status": status, "endpoint": func.__name__}
            )
            request_counter.add(
                1,
                {"status": status, "endpoint": func.__name__}
            )
    
    return wrapper