A+B 複合 — Protocol + Registry で if-elif 根絶（Ch5）× Native Sidecar + OTel Tail Sampling

📋 コピー"""notifier.py — 通知サービス（Email / Slack / SMS）の Protocol ベース実装。

良いコード・悪いコードで学ぶ設計入門（改訂新版）
  - Ch2: 型の活用（StrEnum, 型ヒント）
  - Ch5: 条件分岐の削減（Protocol + レジストリ dict）
  - Ch6: 単一責任（各 Notifier クラスが1チャネルの責務のみ持つ）
"""
from __future__ import annotations

import logging
from dataclasses import dataclass
from enum import StrEnum
from typing import Final, Protocol, runtime_checkable

import requests

logger = logging.getLogger(__name__)

# ── 定数（API URL を名前付き定数で管理）──────────────────────────────────
EMAIL_API_URL: Final[str] = "https://email-api/send"
SLACK_API_URL: Final[str] = "https://slack-api/chat.postMessage"
SMS_API_URL:   Final[str] = "https://sms-api/send"


# ── 列挙型（Ch2: 型の活用 — タイポをコンパイル時に検出）────────────────────
class Channel(StrEnum):
    """通知チャネル識別子。文字列互換でありつつ型安全。"""
    EMAIL = "email"
    SLACK = "slack"
    SMS   = "sms"
    # 新チャネル追加はここに1行足すだけ


# ── カスタム例外 ─────────────────────────────────────────────────────────
class NotificationError(RuntimeError):
    """通知送信に失敗した場合の例外。呼び出し元でキャッチ可能にする。"""


# ── Notifier Protocol（Ch5: if-elif を dict に置き換えるための共通型）────────
@runtime_checkable
class Notifier(Protocol):
    """通知チャネルの共通インターフェース。

    新チャネルを追加する際はこのプロトコルを実装したクラスを作成し
    NOTIFIER_REGISTRY に登録するだけでよい。send_notification() は変更不要（OCP）。
    """
    def send(self, destination: str, message: str) -> None:
        """通知を送信する。

        Args:
            destination: 送信先識別子（メールアドレス / チャンネルID / 電話番号）。
            message: 送信するメッセージ本文。

        Raises:
            NotificationError: 送信に失敗した場合。
        """
        ...


# ── 各チャネルの実装（dataclass で設定を値オブジェクト化）──────────────────
@dataclass(slots=True)  # slots=True でメモリ効率化 + 属性追加ミスを型検査で防ぐ
class EmailNotifier:
    """Email 通知チャネル。"""
    api_url: str  # 設定を dataclass フィールドに閉じ込める

    def send(self, destination: str, message: str) -> None:
        """Email を送信する。"""
        try:
            resp = requests.post(
                self.api_url,
                json={"to": destination, "body": message},
                timeout=10,  # タイムアウトを明示
            )
            resp.raise_for_status()  # 4xx/5xx で例外を送出（Fail-Fast）
        except requests.RequestException as exc:
            # 内部例外を NotificationError でラップして呼び出し元に伝播
            raise NotificationError(
                f"Email send failed to {destination}: {exc}"
            ) from exc
        logger.info("Email sent to %s", destination)


@dataclass(slots=True)
class SlackNotifier:
    """Slack 通知チャネル。"""
    api_url: str

    def send(self, destination: str, message: str) -> None:
        """Slack メッセージを送信する。"""
        try:
            resp = requests.post(
                self.api_url,
                json={"channel": destination, "text": message},
                timeout=10,
            )
            resp.raise_for_status()
        except requests.RequestException as exc:
            raise NotificationError(
                f"Slack send failed to {destination}: {exc}"
            ) from exc
        logger.info("Slack message sent to %s", destination)


@dataclass(slots=True)
class SmsNotifier:
    """SMS 通知チャネル。"""
    api_url: str

    def send(self, destination: str, message: str) -> None:
        """SMS を送信する。"""
        try:
            resp = requests.post(
                self.api_url,
                json={"to": destination, "body": message},
                timeout=10,
            )
            resp.raise_for_status()
        except requests.RequestException as exc:
            raise NotificationError(
                f"SMS send failed to {destination}: {exc}"
            ) from exc
        logger.info("SMS sent to %s", destination)


# ── レジストリ（if-elif チェーンを dict に置き換え）────────────────────────
# 新チャネルを追加する場合: クラスを作成してここに1行追加するだけ
NOTIFIER_REGISTRY: dict[Channel, Notifier] = {
    Channel.EMAIL: EmailNotifier(api_url=EMAIL_API_URL),
    Channel.SLACK: SlackNotifier(api_url=SLACK_API_URL),
    Channel.SMS:   SmsNotifier(api_url=SMS_API_URL),
}


def get_notifier(channel: Channel) -> Notifier:
    """チャネルに対応する Notifier を返す。

    Args:
        channel: 通知チャネル識別子。

    Returns:
        対応する Notifier インスタンス。

    Raises:
        ValueError: 未登録のチャネルが指定された場合（利用可能チャネル一覧を含むメッセージ）。
    """
    notifier = NOTIFIER_REGISTRY.get(channel)
    if notifier is None:
        available = ", ".join(c.value for c in NOTIFIER_REGISTRY)
        raise ValueError(
            f"Unknown channel: {channel!r}. Available: {available}"
        )
    return notifier


def send_notification(channel: Channel, destination: str, message: str) -> None:
    """指定チャネルへ通知を送信するエントリーポイント。

    新しいチャネルを追加する際、この関数を変更する必要はない（OCP）。

    Args:
        channel: 通知チャネル識別子。
        destination: 送信先識別子（チャネルにより意味が異なる）。
        message: 送信メッセージ本文。

    Raises:
        ValueError: 未登録のチャネルが指定された場合。
        NotificationError: 送信に失敗した場合。

    Example:
        >>> send_notification(Channel.EMAIL, "user@example.com", "注文が確定しました")
        >>> send_notification(Channel.SLACK, "#orders", "新規注文 #ORD-123 が到着しました")
    """
    notifier = get_notifier(channel)
    notifier.send(destination, message)

📋 コピー# otel-collector-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: otel-collector-config
  namespace: production
data:
  otel-config.yaml: |
    receivers:
      otlp:
        protocols:
          grpc:
            # Pod 内通信のみ受け付ける（外部からの接続不可）
            endpoint: localhost:4317

    processors:
      # ── Tail Sampling（エラー or 200ms 超のみ通す）──────────────
      tail_sampling:
        decision_wait: 10s      # スパンが揃うまで最大10秒待機してから判断
        num_traces: 50000       # メモリに保持するトレース数の上限
        policies:
          - name: errors-policy
            type: status_code
            status_code:
              status_codes: [ERROR]   # エラートレースは全件転送
          - name: slow-traces-policy
            type: latency
            latency:
              threshold_ms: 200       # 200ms 超のトレースは転送
          - name: health-check-drop
            type: string_attribute
            string_attribute:
              key: http.url
              values: ["/healthz", "/readyz"]
              invert_match: true      # /healthz, /readyz 以外を通す（ヘルスチェックを除外）

      # ── バッチ処理（DataDog への送信を効率化）────────────────────
      batch:
        send_batch_size: 1000
        timeout: 5s

    exporters:
      otlp:
        # ${DATADOG_OTLP_ENDPOINT} は Pod の env から注入（本番/ステージ切り替え）
        endpoint: "${DATADOG_OTLP_ENDPOINT}"
        headers:
          DD-API-KEY: "${DD_API_KEY}"
        retry_on_failure:
          enabled: true
          max_elapsed_time: 120s

    service:
      pipelines:
        traces:
          receivers:  [otlp]
          processors: [tail_sampling, batch]   # tail_sampling → batch の順が重要
          exporters:  [otlp]