弱点補強 — OpenTelemetry 1.x 計装 Bad→Good 7点（TracerProvider × BatchSpanProcessor × record_exception × set_status × structlog trace_id 相関 × Propagator W3C / Baggage × MeterProvider + ヒストグラム）

from __future__ import annotations

import os
from typing import Final

from opentelemetry import metrics, trace
from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import OTLPMetricExporter
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.propagate import set_global_textmap
from opentelemetry.propagators.composite import CompositePropagator
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor  # ← SimpleSpanProcessor は本番 NG
from opentelemetry.trace import set_tracer_provider
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.baggage.propagation import BaggagePropagator

SERVICE_NAME_VALUE: Final[str] = "campaign-dispatcher"
METRIC_DISPATCHED:  Final[str] = "campaigns.dispatched"
METRIC_DURATION_MS: Final[str] = "campaigns.dispatch_duration_ms"

def configure_otel(service_name: str = SERVICE_NAME_VALUE) -> None:
    """TracerProvider / MeterProvider / Propagator を初期化する（起動時一度）."""
    endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
    resource = Resource(attributes={
        SERVICE_NAME: service_name,
        "service.version": "2.1.0",
        "deployment.environment": os.environ.get("ENV", "development"),
    })

    # ── TracerProvider ─────────────────────────────────────────────
    tracer_provider = TracerProvider(resource=resource)
    tracer_provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, insecure=True))
    )
    set_tracer_provider(tracer_provider)  # ← これより後に get_tracer() を呼ぶ

    # ── MeterProvider ──────────────────────────────────────────────
    metrics.set_meter_provider(MeterProvider(
        resource=resource,
        metric_readers=[PeriodicExportingMetricReader(
            OTLPMetricExporter(endpoint=endpoint, insecure=True),
            export_interval_millis=30_000,  # 30 秒ごとに push
        )],
    ))

    # ── Propagator（W3C TraceContext + Baggage）────────────────────
    # Pub/Sub → Cloud Run の traceparent ヘッダーを引き継ぐ
    set_global_textmap(CompositePropagator([
        TraceContextTextMapPropagator(),
        BaggagePropagator(),
    ]))

import logging
import structlog

def _add_otel_context(logger, method_name, event_dict: dict) -> dict:
    """structlog プロセッサ: 現在の trace_id / span_id を付与."""
    span = trace.get_current_span()
    ctx = span.get_span_context()
    if ctx.is_valid:
        event_dict["trace_id"] = format(ctx.trace_id, "032x")  # Loki ↔ Tempo リンク用
        event_dict["span_id"]  = format(ctx.span_id, "016x")
    return event_dict

def configure_structlog() -> None:
    structlog.configure(
        processors=[
            structlog.contextvars.merge_contextvars,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            _add_otel_context,       # ← trace_id / span_id を自動注入
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),  # JSON 出力 → Loki でフィルタ可能
        ],
        wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
        logger_factory=structlog.PrintLoggerFactory(),
    )

from opentelemetry.trace.status import Status, StatusCode

# ── Provider 初期化後にモジュール変数として取得 ──────────────────────────
_tracer  = trace.get_tracer(__name__)
_meter   = metrics.get_meter(__name__)
_log     = structlog.get_logger(__name__)

# カウンタ: success / error 別に集計
_dispatched_counter = _meter.create_counter(
    name=METRIC_DISPATCHED,
    description="キャンペーン配信数",
    unit="1",
)
# ヒストグラム: p50/p95/p99 レイテンシ
_duration_histogram = _meter.create_histogram(
    name=METRIC_DURATION_MS,
    description="配信所要時間",
    unit="ms",
)

def dispatch_campaign(campaign_id: str, user_ids: list[str]) -> dict[str, int]:
    """キャンペーンを指定ユーザーにディスパッチする."""
    import time
    start_ms = time.monotonic() * 1000

    with _tracer.start_as_current_span("campaign.dispatch") as span:
        # ② contextual attribute: Tempo で campaign_id 検索可能になる
        span.set_attribute("campaign.id", campaign_id)
        span.set_attribute("campaign.target_count", len(user_ids))

        success, errors = 0, []

        for uid in user_ids:
            # ③ 子 span で per-user レイテンシ + エラー特定
            with _tracer.start_as_current_span(
                "campaign.send_push",
                attributes={"user.id": uid, "campaign.id": campaign_id},
            ) as child:
                try:
                    _send_push(uid, campaign_id)
                    success += 1
                    child.set_attribute("send.status", "success")
                except Exception as exc:
                    errors.append(uid)
                    child.record_exception(exc)                          # スタックトレースを event 記録
                    child.set_status(Status(StatusCode.ERROR, str(exc))) # ⑤ Tempo でエラー赤表示
                    _log.warning("push_send_failed", uid=uid, campaign_id=campaign_id, error=str(exc))

        duration_ms = time.monotonic() * 1000 - start_ms
        error_count = len(errors)

        span.set_attribute("campaign.success_count", success)
        span.set_attribute("campaign.error_count", error_count)
        if error_count:
            span.set_status(Status(StatusCode.ERROR, f"{error_count} users failed"))

        # ⑥ メトリクス記録
        _dispatched_counter.add(success,      {"campaign_id": campaign_id, "status": "success"})
        _dispatched_counter.add(error_count,  {"campaign_id": campaign_id, "status": "error"})
        _duration_histogram.record(duration_ms, {"campaign_id": campaign_id})

        # ④ 構造化ログ（trace_id が _add_otel_context で自動付与）
        _log.info("campaign_dispatched",
                  campaign_id=campaign_id, success=success,
                  total=len(user_ids), error_count=error_count,
                  duration_ms=round(duration_ms, 2))

        return {"success": success, "total": len(user_ids), "error": error_count}

# ── エントリーポイント ────────────────────────────────────────────────────
if __name__ == "__main__":
    configure_otel()
    configure_structlog()
    result = dispatch_campaign("camp_2026_summer_sale", ["u001", "u002", "u003"])
    print(result)
    # → {"success": 3, "total": 3, "error": 0}