システム設計/インフラ — OpenTelemetry 分散トレーシング × DataDog 相関

📋 コピー# batch_processor.py（改善後）
"""販促クーポンバッチ処理モジュール。

OpenTelemetry による分散トレーシングと構造化ログ（DataDog 相関対応）を実装。
"""
from __future__ import annotations

import os

import requests
import structlog
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.propagate import inject
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import get_current_span

OTEL_ENDPOINT = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://datadog-agent:4317")
COUPON_SERVICE_URL = os.getenv("COUPON_SERVICE_URL", "https://coupon-service.example.com")
REQUEST_TIMEOUT_SEC = 5
SERVICE_NAME = "batch-processor"


def _setup_tracer_provider() -> TracerProvider:
    resource = Resource.create({"service.name": SERVICE_NAME})
    exporter = OTLPSpanExporter(endpoint=OTEL_ENDPOINT, insecure=True)
    provider = TracerProvider(resource=resource)
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


_provider = _setup_tracer_provider()
_tracer = trace.get_tracer(SERVICE_NAME)


def _get_logger() -> structlog.BoundLogger:
    structlog.configure(
        processors=[
            structlog.processors.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.JSONRenderer(),
        ]
    )
    return structlog.get_logger()


logger = _get_logger()


def _current_trace_id() -> str:
    """DataDog 形式の trace_id（32桁16進数）を返す。"""
    ctx = get_current_span().get_span_context()
    if not ctx.is_valid:
        return "0" * 32
    return format(ctx.trace_id, "032x")


def _current_span_id() -> str:
    ctx = get_current_span().get_span_context()
    if not ctx.is_valid:
        return "0" * 16
    return format(ctx.span_id, "016x")


def _post_with_trace(url: str, payload: dict) -> requests.Response:
    """W3C Trace Context ヘッダーを注入して HTTP POST を送信する。"""
    headers: dict[str, str] = {"Content-Type": "application/json"}
    inject(headers)  # traceparent / tracestate を自動追加
    resp = requests.post(url, json=payload, headers=headers, timeout=REQUEST_TIMEOUT_SEC)
    resp.raise_for_status()
    return resp


def process_coupon_batch(coupon_ids: list[str]) -> dict[str, str]:
    results: dict[str, str] = {}

    with _tracer.start_as_current_span("process_coupon_batch") as batch_span:
        logger.info(
            "batch_started",
            dd={"trace_id": _current_trace_id(), "span_id": _current_span_id()},
            total=len(coupon_ids),
        )

        for coupon_id in coupon_ids:
            with _tracer.start_as_current_span(
                "apply_coupon",
                attributes={"coupon.id": coupon_id},
            ) as coupon_span:
                try:
                    _post_with_trace(f"{COUPON_SERVICE_URL}/apply", {"coupon_id": coupon_id})
                    results[coupon_id] = "ok"
                    coupon_span.set_attribute("coupon.result", "ok")
                    logger.info(
                        "coupon_applied",
                        dd={"trace_id": _current_trace_id(), "span_id": _current_span_id()},
                        coupon_id=coupon_id,
                    )
                except requests.HTTPError as exc:
                    results[coupon_id] = "error"
                    coupon_span.set_status(trace.StatusCode.ERROR, str(exc))
                    coupon_span.record_exception(exc)
                    logger.error(
                        "coupon_apply_failed",
                        dd={"trace_id": _current_trace_id(), "span_id": _current_span_id()},
                        coupon_id=coupon_id,
                    )
                except requests.Timeout:
                    results[coupon_id] = "error"
                    coupon_span.set_status(trace.StatusCode.ERROR, "timeout")
                    logger.error("coupon_apply_timeout", coupon_id=coupon_id)

        success_count = sum(1 for v in results.values() if v == "ok")
        batch_span.set_attribute("batch.success_count", success_count)
        batch_span.set_attribute("batch.error_count", len(coupon_ids) - success_count)

    return results