A+B 複合 — 継承より委譲 × Protocol × ポリシーパターン（通知ディスパッチャー Bad→Good Ch3/Ch6-7/Ch8/Ch10）× GKE Autopilot Pub/Sub Pull Worker グレースフルシャットダウン × requests適正化 × liveness/readinessProbe × KEDA gcp-pubsub バックログスケーリング × PodDisruptionBudget

2026-07-18 (Day 101) 土曜複合問題 ★★★★☆ Python 3.12 / Protocol / StrEnum / frozen dataclass / 委譲 GKE Autopilot / SIGTERM / KEDA gcp-pubsub / PodDisruptionBudget

概要

🧩

継承ではなく Protocol × 委譲で送信手段を差し替え可能にする

BaseNotifier の継承ツリーは送信手段が違うだけなのに状態とリトライロジックまで共有させていた。Notifier を Protocol（構造的部分型）にし、リトライは RetryPolicy という別オブジェクトに委譲することで、クラス間の結合を「メソッドシグネチャの一致」だけに絞れる。

📇

レジストリパターンで if/elif 分岐を排除（OCP）

NOTIFIER_FACTORIES: dict[NotificationChannel, Callable] に置き換えることで、新チャネル追加時に dispatch_campaign_notifications 自体は1行も変更不要になる。StrEnum と組み合わせてマジックストリングも排除する。

🛑

SIGTERM ハンドラーで in-flight メッセージを drain してから終了

Kubernetes の Pod 終了は SIGTERM → 猶予期間 → SIGKILL の2段階。アプリが SIGTERM を無視すると処理中メッセージが未ACKのまま強制終了され、Pub/Sub の再配信で重複送信が起きる。フラグを立てて新規pullを止め、処理中分だけ完了させる。

📈

KEDA でPub/Subのbacklogを直接見てスケール

CPUベースのHPAは「今の負荷」にしか反応できない。KEDAの gcp-pubsub トリガー（SubscriptionSize）はキューの未配信メッセージ数を先読みしてスケールアウトでき、フラッシュセールのような急増にも配信SLAを守れる。

問題 A: コーディング — 継承より委譲 × Protocol × ポリシーパターン（通知ディスパッチャー Bad→Good）

以下の「悪いコード」は、ECサイト MOps チームのキャンペーン通知（メール/SMS/プッシュ）を送信する BaseNotifier 継承ツリーです。問題点を全て洗い出し、継承より委譲・Protocol による構造的部分型・ポリシーパターン（リトライ）・完全コンストラクタ・専用例外階層を使って Bad→Good にリファクタリングしてください。

制約・前提条件

  • Python 3.12+（StrEnum は enum.StrEnum）
  • Notifier を Protocol として定義し、各 Notifier は継承ではなく構造的部分型で差し替え可能にすること
  • リトライは BaseNotifier に継承させず、RetryPolicy という別クラスに委譲すること（Ch6-7: 継承より委譲）
  • NotificationChannel を StrEnum で定義し、if/elif 分岐をレジストリ（dict）+ Protocol 注入で解消すること
  • SMS の文字数制限超過は暗黙トリミングせず、専用例外 MessageTooLongError を送出すること
  • 各 Notifier は @dataclass(frozen=True, slots=True) の完全コンストラクタで依存性を注入すること（new での内部生成禁止）
  • 例外を握りつぶさず、logging で構造化して記録すること
  • Google スタイル docstring・インラインコメント・名前付き定数を含めること
期待する回答形式: 問題点の列挙（番号付き）+ 改善後コード + 実行例（input→output）+ 適用した設計パターン名と書籍対応章

悪いコード (Before) — カテゴリ A

このコードには 7つの設計上の問題 が隠れています。見つけてみてください。
bad_notifier.py — 継承で状態共有・NotImplementedError・if/elif分岐・暗黙トリミング
class BaseNotifier:
    def __init__(self, campaign_id, retry_count=3):
        self.campaign_id = campaign_id
        self.retry_count = retry_count
        self.sent_log = []  # 問題1: 状態を親クラスが持ち、全サブクラスに強制継承

    def send(self, message):
        raise NotImplementedError  # 問題2: 抽象メソッドをNotImplementedErrorで表現

    def send_with_retry(self, message):
        for i in range(self.retry_count):
            try:
                self.send(message)
                self.sent_log.append(message)
                return True
            except Exception as e:
                print(f"retry {i}: {e}")  # 問題7: 例外を握りつぶし、ログレベルも不明
        return False


class SmsNotifier(BaseNotifier):
    def __init__(self, campaign_id, sms_client, retry_count=5):  # 問題3: サブクラスごとに既定値バラバラ
        super().__init__(campaign_id, retry_count)
        self.sms_client = sms_client
        self.max_length = 70

    def send(self, message):
        if len(message) > self.max_length:
            message = message[:self.max_length]  # 問題4: 暗黙のトリミング
        self.sms_client.send_sms(message)


class PushNotifier(BaseNotifier):
    def send(self, message):
        client = PushGatewayClient()  # 問題5: 依存を内部でnew、DI無視
        client.push(message)


def dispatch_campaign_notifications(campaign_id, channel, message, smtp_client=None, sms_client=None):
    if channel == "email":                                    # 問題6: if/elif分岐
        notifier = EmailNotifier(campaign_id, smtp_client)
    elif channel == "sms":
        notifier = SmsNotifier(campaign_id, sms_client)
    elif channel == "push":
        notifier = PushNotifier(campaign_id)
    else:
        return False
    return notifier.send_with_retry(message)
問題点サマリー（7点）
1BaseNotifier が状態(sent_log)を強制継承（Ch6-7） — 送信手段が違うだけのサブクラスに親クラスの内部状態を背負わせている。継承より委譲に置き換える
2抽象メソッドをNotImplementedErrorで表現（Ch6-7） — 実装漏れが実行時まで検出できない。Protocolで静的検査可能にする
3retry_countの既定値がサブクラスごとにバラバラ（Ch3） — ポリシーが暗黙的に発散。RetryPolicyに一元化する
4SMS文字数超過の暗黙トリミング（Ch10） — 呼び出し側が気づけない欠落。専用例外でフェイルファストにする
5PushNotifierが依存を内部でnew（Ch3） — テスト時にモック注入できない。完全コンストラクタで受け取る
6if/elifによるチャネル分岐（Ch6-7/Ch8） — 新チャネル追加のたびに関数修正が必要。レジストリで解消する
7例外の握りつぶし（Ch10） — 種別を区別せずprintするだけ。専用例外階層+loggingに置き換える

ヒント A（段階的開示）

ヒント1 — 方向性
BaseNotifier の継承ツリーは「送信方法が違うだけ」なのに、状態（sent_log）とリトライロジックまで親クラスから強制的に引き継がせている。継承は「is-a」関係にのみ使い、送信手段は Protocol（構造的部分型）で表現する。リトライは RetryPolicy という別オブジェクトに委譲し、notifier.send() を呼び出す側に回る。チャネル分岐は if/elif ではなく dict[NotificationChannel, Notifier] のレジストリに置き換えれば、新チャネル追加時に既存コードを1行も変更しなくて済む（OCP: 開放閉鎖原則）。
ヒント2 — アプローチ
  • class NotificationChannel(StrEnum): EMAIL = "email"; SMS = "sms"; PUSH = "push"
  • class Notifier(Protocol): def send(self, message: str) -> None: ...
  • @dataclass(frozen=True, slots=True) class EmailNotifier / SmsNotifier / PushNotifier はそれぞれ依存クライアントをコンストラクタで受け取るのみ
  • SmsNotifier.send は文字数超過時に raise MessageTooLongError(...)（暗黙トリミング禁止）
  • @dataclass(frozen=True, slots=True) class RetryPolicy: max_attempts: int = 3 に execute(self, notifier, message) -> bool を実装し、try/except NotificationError で logging.warning を出す
  • NOTIFIER_FACTORIES: dict[NotificationChannel, Callable[..., Notifier]] のレジストリを用意し、dispatch_campaign_notifications はレジストリ検索のみに単純化する
ヒント3 — コードの骨格 
from __future__ import annotations
from dataclasses import dataclass
from enum import StrEnum
from typing import Protocol, Callable, Final
import logging

logger = logging.getLogger(__name__)
MAX_SMS_LENGTH: Final[int] = 70

class NotificationChannel(StrEnum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"

class Notifier(Protocol):
    def send(self, message: str) -> None: ...

@dataclass(frozen=True, slots=True)
class RetryPolicy:
    max_attempts: int = 3
    def execute(self, notifier: Notifier, message: str) -> bool:
        ...  # try/except NotificationError, logging.warning

問題点分析 — カテゴリ A

#問題点分類改善方法
1BaseNotifierが状態を強制継承関心の分離 Ch6-7継承より委譲、状態は共有しない
2NotImplementedErrorで抽象メソッド表現構造的部分型 Ch6-7Protocolで静的検査可能にする
3retry_countの既定値がサブクラスごとに発散完全コンストラクタ Ch3RetryPolicyに一元化
4SMS文字数超過の暗黙トリミング設計の悪魔 Ch10MessageTooLongErrorでフェイルファスト
5PushNotifierが依存を内部でnew完全コンストラクタ Ch3コンストラクタで依存を注入
6if/elifによるチャネル分岐ポリシーパターン Ch8NOTIFIER_FACTORIESレジストリ
7例外の握りつぶし設計の悪魔 Ch10専用例外階層 + logging

模範解答 A

Before — 継承で状態共有・if/elif分岐・暗黙トリミング
class BaseNotifier:
    def __init__(self, campaign_id, retry_count=3):
        self.sent_log = []          # 強制継承される状態
    def send(self, message):
        raise NotImplementedError    # 実行時まで検出不可
    def send_with_retry(self, message):
        for i in range(self.retry_count):
            try:
                self.send(message)
                return True
            except Exception as e:
                print(f"retry {i}: {e}")  # 握りつぶし
        return False

class SmsNotifier(BaseNotifier):
    def send(self, message):
        if len(message) > self.max_length:
            message = message[:self.max_length]  # 暗黙トリミング
        self.sms_client.send_sms(message)

def dispatch_campaign_notifications(campaign_id, channel, message, ...):
    if channel == "email":       # if/elif分岐
        notifier = EmailNotifier(...)
    elif channel == "sms":
        notifier = SmsNotifier(...)
    ...
After — Protocol × 委譲（RetryPolicy）× レジストリ × 専用例外
"""notifier.py — 継承より委譲 × Protocol × ポリシーパターン（Ch3/Ch6-7/Ch8/Ch10）"""
from __future__ import annotations

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

logger = logging.getLogger(__name__)
MAX_SMS_LENGTH: Final[int] = 70  # 名前付き定数


class NotificationChannel(StrEnum):
    EMAIL = "email"
    SMS = "sms"
    PUSH = "push"


class NotificationError(Exception):
    """通知送信に関する例外の基底クラス。"""


class MessageTooLongError(NotificationError):
    """SMSメッセージが規定の文字数を超えた場合の例外。"""
    def __init__(self, channel: NotificationChannel, length: int, limit: int) -> None:
        super().__init__(f"{channel} message too long: {length} > {limit}")
        self.channel = channel


class Notifier(Protocol):
    """送信手段のプロトコル（継承ではなく構造的部分型）。"""
    def send(self, message: str) -> None: ...


@dataclass(frozen=True, slots=True)
class EmailNotifier:
    smtp_client: object

    def send(self, message: str) -> None:
        try:
            self.smtp_client.send_mail(message)
        except Exception as exc:
            raise NotificationError(f"email send failed: {exc}") from exc


@dataclass(frozen=True, slots=True)
class SmsNotifier:
    sms_client: object

    def send(self, message: str) -> None:
        if len(message) > MAX_SMS_LENGTH:
            raise MessageTooLongError(NotificationChannel.SMS, len(message), MAX_SMS_LENGTH)
        try:
            self.sms_client.send_sms(message)
        except Exception as exc:
            raise NotificationError(f"sms send failed: {exc}") from exc


@dataclass(frozen=True, slots=True)
class PushNotifier:
    push_client: object  # コンストラクタでDI（内部でnewしない）

    def send(self, message: str) -> None:
        try:
            self.push_client.push(message)
        except Exception as exc:
            raise NotificationError(f"push send failed: {exc}") from exc


@dataclass(frozen=True, slots=True)
class RetryPolicy:
    """リトライは継承ではなく委譲するポリシーオブジェクト（Ch6-7）。"""
    max_attempts: int = 3

    def execute(self, notifier: Notifier, message: str) -> bool:
        for attempt in range(1, self.max_attempts + 1):
            try:
                notifier.send(message)
                return True
            except NotificationError:
                logger.warning("notification failed", extra={"attempt": attempt})
        return False


# チャネル→Notifierファクトリのレジストリ（if/elif分岐を排除、OCP、Ch6-7/Ch8）
NOTIFIER_FACTORIES: Final[dict[NotificationChannel, Callable[..., Notifier]]] = {
    NotificationChannel.EMAIL: EmailNotifier,
    NotificationChannel.SMS: SmsNotifier,
    NotificationChannel.PUSH: PushNotifier,
}


def dispatch_campaign_notifications(
    channel: NotificationChannel, message: str, client: object,
    retry_policy: RetryPolicy | None = None,
) -> bool:
    """チャネルに応じたNotifierを生成し、リトライポリシー付きで送信する。"""
    factory = NOTIFIER_FACTORIES[channel]  # 未知チャネルはKeyErrorでフェイルラウド
    notifier = factory(client)
    policy = retry_policy or RetryPolicy()
    return policy.execute(notifier, message)
class FakeSmtpClient:
    def send_mail(self, message: str) -> None:
        print(f"[email] {message}")

class FakeSmsClient:
    def send_sms(self, message: str) -> None:
        print(f"[sms] {message}")

# 正常系
dispatch_campaign_notifications(NotificationChannel.EMAIL, "夏セール開始！", FakeSmtpClient())
# → [email] 夏セール開始！
# → True

# SMS文字数超過はフェイルファスト（暗黙トリミングしない）
dispatch_campaign_notifications(NotificationChannel.SMS, "あ" * 100, FakeSmsClient())
# → MessageTooLongError: sms message too long: 100 > 70

# リトライポリシーを差し替え（テストでは1回のみに短縮）
dispatch_campaign_notifications(
    NotificationChannel.SMS, "在庫復活のお知らせ", FakeSmsClient(),
    retry_policy=RetryPolicy(max_attempts=1),
)
# → [sms] 在庫復活のお知らせ
# → True
ポイント適用した設計原則/パターン書籍対応章
Notifier を継承ではなく Protocol で表現継承より委譲・構造的部分型Ch6-7
RetryPolicy に試行ロジックを分離委譲・単一責任Ch6-7
各 Notifier がコンストラクタで依存を受け取る完全コンストラクタ・DICh3
NOTIFIER_FACTORIES レジストリで分岐排除ポリシーパターン・OCPCh8
MessageTooLongError で暗黙トリミング禁止フェイルファスト・専用例外Ch10
logging.warning で例外を握りつぶさない例外の握りつぶし禁止Ch10
 
# tests/test_notifier.py
import pytest
from notifier import (
    NotificationChannel, NotificationError, MessageTooLongError,
    EmailNotifier, SmsNotifier, RetryPolicy, dispatch_campaign_notifications,
)


class FakeSmtpClient:
    def __init__(self): self.sent = []
    def send_mail(self, message): self.sent.append(message)


class FlakySmsClient:
    """1回目は失敗、2回目で成功するフェイク（リトライ検証用）。"""
    def __init__(self): self.calls = 0
    def send_sms(self, message):
        self.calls += 1
        if self.calls < 2:
            raise ConnectionError("timeout")


class TestDispatch:
    def test_email_success(self):
        client = FakeSmtpClient()
        assert dispatch_campaign_notifications(NotificationChannel.EMAIL, "hi", client)
        assert client.sent == ["hi"]

    def test_sms_too_long_raises_without_trimming(self):
        with pytest.raises(MessageTooLongError):
            dispatch_campaign_notifications(NotificationChannel.SMS, "x" * 100, FlakySmsClient())

    def test_retry_recovers_from_transient_failure(self):
        client = FlakySmsClient()
        ok = dispatch_campaign_notifications(
            NotificationChannel.SMS, "short", client, retry_policy=RetryPolicy(max_attempts=3)
        )
        assert ok is True
        assert client.calls == 2

    def test_unknown_channel_raises_keyerror(self):
        with pytest.raises(KeyError):
            dispatch_campaign_notifications("line", "hi", None)  # type: ignore[arg-type]


class TestRetryPolicy:
    def test_gives_up_after_max_attempts(self):
        class AlwaysFails:
            def send(self, message): raise NotificationError("boom")
        assert RetryPolicy(max_attempts=2).execute(AlwaysFails(), "x") is False

問題 B: インフラ — GKE Autopilot Pub/Sub Pull Worker のグレースフルシャットダウン × KEDA バックログスケーリング

問題Aの通知ディスパッチャーは、GKE Autopilot 上で Pub/Sub の Pull Worker（notify-worker）として常駐稼働しています。現状の Deployment には以下の課題があります。

現状の課題:
  • Python 側が while True: pull → process → ack のループを直接回しており、SIGTERM ハンドラーが無い → Pod 削除時に処理中メッセージが未 ACK のまま強制終了され、Pub/Sub の再配信で通知が重複送信される
  • resources.requests が cpu: "2" / memory: "4Gi" で固定されているが、実測ピークはその1/3程度 → GKE Autopilot は requests に対して課金されるため無駄なコストが発生している
  • livenessProbe/readinessProbe が設定されておらず、内部でデッドロックした Pod が検知されず Ready のまま残り続ける
  • HPA が CPU 使用率のみをトリガーにしており、Pub/Sub の未配信メッセージ数（backlog）に応じてスケールしない → フラッシュセール開始時の通知急増でスケールアウトが遅れ、配信SLA（5分以内）を超過する
  • PodDisruptionBudget が未設定 → GKE のノードアップグレード時に全 Pod が同時に退避され、配信が数分間完全停止する

要件

#要件
1Python 側で SIGTERM を捕捉し、新規メッセージの pull を止めて処理中のメッセージの ACK 完了を待ってから終了すること（グレースフルシャットダウン）
2terminationGracePeriodSeconds をグレースフルシャットダウンの猶予に合わせて延長すること
3resources.requests/limits を実測ベースに適正化し、Autopilot の Guaranteed QoS（requests == limits）にすること
4livenessProbe/readinessProbe を実装し、pull ループの心拍で死活監視すること
5KEDA ScaledObject（gcp-pubsub トリガー）で未配信メッセージ数に応じてスケールすること
6PodDisruptionBudget で minAvailable を設定し、ノードメンテナンス時も配信を継続すること
期待する回答形式: Python（グレースフルシャットダウン実装）+ Kubernetes マニフェスト（Deployment/PDB/KEDA ScaledObject）+ Bad vs Good 比較表 + 確認コマンド

ヒント B（段階的開示）

ヒント1 — 方向性
Kubernetes の Pod 終了は SIGTERM → terminationGracePeriodSeconds 経過後に SIGKILL という2段階。アプリが SIGTERM を無視すると、猶予期間を待たずに強制終了されたのと同じ結果になる（処理中メッセージの喪失・重複配信）。HPA の CPU ベーススケーリングは「今の負荷」にしか反応できないため、Pub/Sub のようなキューベースワークロードは KEDA でキューの深さ（backlog）を直接見るのが定石。Autopilot は Pod の requests に対して秒単位で課金されるため、過大な requests はそのままコスト増に直結する。
ヒント2 — アプローチ
  • signal.signal(signal.SIGTERM, handler) でシャットダウンフラグ（threading.Event）を立てるだけにし、実際のループ脱出はメインループ側の条件分岐で行う（シグナルハンドラー内で重い処理をしない）
  • terminationGracePeriodSeconds: 90 のように、1メッセージの最大処理時間 + マージンを確保
  • resources: { requests: {cpu: "250m", memory: "512Mi"}, limits: {cpu: "250m", memory: "512Mi"} }（Guaranteed QoS）
  • livenessProbe/readinessProbe は軽量な HTTP ヘルスチェックサーバーを別スレッドで立て、心拍のタイムスタンプが一定時間以内かを判定する
  • KEDA ScaledObject の triggers[].type: gcp-pubsub、metadata.subscriptionName + metadata.mode: SubscriptionSize + metadata.value
  • PodDisruptionBudget.spec.minAvailable
ヒント3 — リソースの骨格
グレースフルシャットダウン + probe
spec:
  terminationGracePeriodSeconds: 90
  containers:
    - name: worker
      resources:
        requests: { cpu: "250m", memory: "512Mi" }
        limits:   { cpu: "250m", memory: "512Mi" }
      livenessProbe:
        httpGet: { path: /healthz, port: 8080 }
      readinessProbe:
        httpGet: { path: /healthz, port: 8080 }
KEDA ScaledObject
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
spec:
  scaleTargetRef:
    name: notify-worker
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: gcp-pubsub
      metadata:
        subscriptionName: "notify-campaign-sub"
        mode: "SubscriptionSize"
        value: "50"

アーキテクチャ図 — グレースフルシャットダウン × KEDA バックログスケーリング

Pub/Sub topic: notify-campaign-topic sub: notify-campaign-sub num_undelivered_messages (backlog) KEDA ScaledObject trigger: gcp-pubsub mode: SubscriptionSize, value: 50 min=1 / max=20 / cooldown=60s HPA（KEDAが自動生成） external metric = backlog 修正5: CPUのみに依存しない 先読みでスケールアウト TriggerAuthentication podIdentity.provider: gcp Workload Identity 経由 静的認証情報なし GKE Autopilot — namespace: mops Deployment: notify-worker（replica 1〜20、KEDAが制御） アプリコンテナ 修正1: SIGTERMでpull停止フラグ in-flightをdrainしてからexit 修正3: requests=limits=250m/512Mi （Guaranteed QoS） 修正4: /healthz で心拍監視 liveness/readinessProbe notifier.py（問題Aの成果物） Pod ライフサイクル Before: SIGTERM未処理 →即終了・未ACK→重複配信 修正2: terminationGracePeriodSeconds=90 SIGTERM→drain→正常終了 90s超過分のみSIGKILL （想定外の長時間処理のみ） PodDisruptionBudget 修正6: minAvailable: 1 ノードアップグレード時も 常時1台は保護 Before: PDBなし 全Pod同時退避→配信停止 /healthz 別スレッドHTTPサーバー last_heartbeat監視 30秒超過で503 Before: probeなし デッドロックPod放置 通知先: SMTP / SMS Gateway / Push Gateway（問題Aの Notifier 経由） RetryPolicy.execute で最大N回試行、失敗は logging.warning で記録 ✓ SIGTERM受信でpull停止・in-flight完了を待って終了（重複配信防止） ✓ Guaranteed QoS（requests=limits）でAutopilotの過剰課金を排除 ✓ /healthz によるliveness/readinessProbeでデッドロックPodを排除 ✓ KEDA gcp-pubsub でbacklog連動スケール（フラッシュセール急増に追従） ✓ PodDisruptionBudgetでノードメンテナンス時も配信継続 replicaを増減 pull / ack

模範解答 B

"""notify_worker.py — Pub/Sub Pull Worker（グレースフルシャットダウン対応）"""
from __future__ import annotations

import logging
import signal
import threading
import time
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, HTTPServer
from typing import Final

from google.cloud import pubsub_v1

logger = logging.getLogger(__name__)

# ヘルスチェックが「生きている」と判定する心拍の許容遅延（秒）
HEARTBEAT_TIMEOUT_SECONDS: Final[int] = 30
PULL_MAX_MESSAGES: Final[int] = 10
PULL_TIMEOUT_SECONDS: Final[int] = 5


@dataclass(slots=True)
class WorkerState:
    """ワーカーのシャットダウン要求・心拍を保持する共有状態。"""
    shutdown_requested: threading.Event = field(default_factory=threading.Event)
    last_heartbeat: float = field(default_factory=time.time)


def _install_signal_handler(state: WorkerState) -> None:
    """SIGTERM受信時は新規pullを止めるフラグを立てるだけにする（重い処理はしない）。"""
    def _on_sigterm(signum, frame) -> None:
        logger.info("SIGTERM received: stop pulling new messages, draining in-flight work")
        state.shutdown_requested.set()

    signal.signal(signal.SIGTERM, _on_sigterm)


class _HealthzHandler(BaseHTTPRequestHandler):
    """liveness/readiness 用の軽量ヘルスチェックサーバー。"""
    state: WorkerState  # クラス属性として外部から注入

    def do_GET(self) -> None:
        healthy = (time.time() - self.state.last_heartbeat) < HEARTBEAT_TIMEOUT_SECONDS
        self.send_response(200 if healthy else 503)
        self.end_headers()

    def log_message(self, format: str, *args) -> None:
        pass  # アクセスログを標準出力に垂れ流さない


def _start_healthz_server(state: WorkerState, port: int = 8080) -> None:
    _HealthzHandler.state = state
    server = HTTPServer(("0.0.0.0", port), _HealthzHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()


def run(subscription_path: str, dispatch, retry_policy) -> None:
    """グレースフルシャットダウン対応の pull ループ本体。

    Args:
        subscription_path: Pub/Sub サブスクリプションのフルパス。
        dispatch: 1メッセージを処理する呼び出し可能オブジェクト。
        retry_policy: RetryPolicy インスタンス。
    """
    state = WorkerState()
    _install_signal_handler(state)
    _start_healthz_server(state)

    subscriber = pubsub_v1.SubscriberClient()
    while not state.shutdown_requested.is_set():
        response = subscriber.pull(
            request={"subscription": subscription_path, "max_messages": PULL_MAX_MESSAGES},
            timeout=PULL_TIMEOUT_SECONDS,
        )
        for msg in response.received_messages:
            # シャットダウン要求後も、受信済みメッセージは最後まで処理してACKする
            ok = dispatch(msg.message.data.decode(), retry_policy)
            if ok:
                subscriber.acknowledge(
                    request={"subscription": subscription_path, "ack_ids": [msg.ack_id]}
                )
        state.last_heartbeat = time.time()

    logger.info("graceful shutdown complete: all in-flight messages drained")
# deployment.yaml — グレースフルシャットダウン + 適正サイジング + probe
apiVersion: apps/v1
kind: Deployment
metadata:
  name: notify-worker
  namespace: mops
spec:
  replicas: 1  # 初期値のみ。以降はKEDAが制御する
  selector:
    matchLabels: { app: notify-worker }
  template:
    metadata:
      labels: { app: notify-worker }
    spec:
      serviceAccountName: notify-worker
      terminationGracePeriodSeconds: 90  # 修正2: 1メッセージ最大処理時間+マージン
      containers:
        - name: worker
          image: asia-northeast1-docker.pkg.dev/PROJECT_ID/mops/notify-worker@sha256:abc123
          resources:                      # 修正3: 実測ベースの適正サイジング（Guaranteed QoS）
            requests: { cpu: "250m", memory: "512Mi" }
            limits:   { cpu: "250m", memory: "512Mi" }
          readinessProbe:                 # 修正4
            httpGet: { path: /healthz, port: 8080 }
            periodSeconds: 10
            failureThreshold: 2
          livenessProbe:                  # 修正4
            httpGet: { path: /healthz, port: 8080 }
            periodSeconds: 15
            failureThreshold: 3
          securityContext:
            runAsNonRoot: true
            allowPrivilegeEscalation: false
            capabilities: { drop: ["ALL"] }
---
# pdb.yaml — 修正6: ノードメンテナンス時の全滅を防ぐ
apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: notify-worker-pdb
  namespace: mops
spec:
  minAvailable: 1
  selector:
    matchLabels: { app: notify-worker }
---
# scaledobject.yaml — 修正5: Pub/Sub backlogベースのスケーリング（KEDA）
apiVersion: keda.sh/v1alpha1
kind: TriggerAuthentication
metadata:
  name: keda-trigger-auth-gcp
  namespace: mops
spec:
  podIdentity:
    provider: gcp  # Workload Identity 経由（静的認証情報なし）
---
apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: notify-worker-scaledobject
  namespace: mops
spec:
  scaleTargetRef:
    name: notify-worker
  minReplicaCount: 1
  maxReplicaCount: 20
  cooldownPeriod: 60          # スケールイン時の過剰反応を抑制
  triggers:
    - type: gcp-pubsub
      authenticationRef:
        name: keda-trigger-auth-gcp
      metadata:
        subscriptionName: "notify-campaign-sub"
        mode: "SubscriptionSize"   # 未配信メッセージ数を直接見る
        value: "50"                 # Pod 1台あたり許容backlog 50件を目安にスケール

Bad vs Good 設計比較

観点Bad（現状）Good（改善後）
シャットダウンSIGTERM未処理 → 処理中メッセージ喪失・重複配信SIGTERMでpull停止フラグ、in-flightをdrainしてから終了
Pod終了猶予デフォルト30秒（不足）terminationGracePeriodSeconds: 90
リソースrequests過大（cpu=2/memory=4Gi固定）、Autopilotで課金増実測ベース cpu=250m/memory=512Mi かつ Guaranteed QoS
死活監視probeなし。デッドロックPodが放置されるliveness/readinessProbeで心拍監視
スケーリングHPA CPUベースのみ、backlogに反応しないKEDA gcp-pubsubトリガーでbacklog連動スケール
可用性PDBなし、ノードメンテで全滅しうるPodDisruptionBudget minAvailable=1

確認コマンド

# 1. グレースフルシャットダウンの疎通確認（Pod削除→ログでdrain完了を確認）
kubectl delete pod -n mops -l app=notify-worker --wait=false
kubectl logs -n mops -l app=notify-worker --since=2m | grep "graceful shutdown complete"
# Expected: "graceful shutdown complete: all in-flight messages drained" が出力されること

# 2. Guaranteed QoS の確認
kubectl get pod -n mops -l app=notify-worker -o jsonpath='{.items[0].status.qosClass}'
# Expected: Guaranteed

# 3. probe設定の確認
kubectl get deployment notify-worker -n mops -o jsonpath='{.spec.template.spec.containers[0].livenessProbe}'
# Expected: httpGet.path=/healthz が設定されていること

# 4. KEDA ScaledObject の状態確認
kubectl get scaledobject notify-worker-scaledobject -n mops
kubectl describe scaledobject notify-worker-scaledobject -n mops | grep -A3 "Triggers Types"
# Expected: READY=True, ACTIVE=True（backlogがしきい値を超えたら）

# 5. backlog連動スケールの負荷確認（擬似的にメッセージを大量publish）
gcloud pubsub topics publish notify-campaign-topic --message="load-test" --count=2000
kubectl get hpa -n mops -w
# Expected: サブスクリプションのbacklog増加に伴いreplicaが段階的に増加すること

# 6. PDBの確認
kubectl get pdb notify-worker-pdb -n mops
# Expected: MIN AVAILABLE=1, ALLOWED DISRUPTIONS が0にならないこと（常時1台は保護される）

ポイント解説

カテゴリ A

1 継承は「is-a」を検証できる場合のみ使う（Ch6-7）
EmailNotifier/SmsNotifier/PushNotifier は「送信手段が違うだけ」で、共通の親クラスの内部状態を共有する必然性がない。継承ツリーが深くなるほど、親クラスの変更が予期しないサブクラスに波及するリスクが増える。Protocol による構造的部分型に置き換えることで、クラス間の結合を「メソッドシグネチャの一致」だけに絞れる。
2 リトライを委譲することでテスト容易性が上がる（Ch6-7）
RetryPolicy が独立したオブジェクトになったことで、RetryPolicy(max_attempts=1) のようにテスト時だけ試行回数を変えられる。継承ベースの実装ではサブクラスごとに retry_count を上書きする必要があり、一貫したポリシー変更ができなかった。
3 レジストリパターンでOCPを満たす（Ch8）
NOTIFIER_FACTORIES に新チャネル（例: LINE通知）を追加する場合、辞書に1行足すだけで済み、dispatch_campaign_notifications 自体は変更不要になる。if/elif の連鎖は行数が伸びるほど見落としが増える。

カテゴリ B

4 SIGTERMハンドラーは「重い処理をしない」のが鉄則
シグナルハンドラー内で pull/ack のような I/O を行うとデッドロックのリスクがあるため、フラグを立てるだけに留め、実際のドレイン処理はメインループの条件分岐に任せる。terminationGracePeriodSeconds はこのドレインが完了しうる時間を見積もって設定する。
5 Autopilotのコストは「実測に基づくrequests適正化」がそのまま効く
Autopilot は requests に対して秒単位課金されるため、CPU/メモリを過大に確保すると常にその分課金され続ける。limits を requests と同値にする Guaranteed QoS は、リソース保証と課金の予測可能性を両立できる。
6 KEDAのbacklogベーススケーリングはHPAのCPUベースを補完する
CPU使用率は「今処理している量」にしか反応しないが、Pub/Subのbacklogは「これから処理すべき量」を先読みできる。SubscriptionSize モードでスケールアウトを早め、通知SLA（5分以内）の超過を防ぐ。

実務への応用

  • NOTIFIER_FACTORIES のようなレジストリパターンは、MOps の配信チャネル追加（LINE通知、Slack通知など）が頻発する箇所全般に転用できる: 新チャネル追加のPRが「辞書に1エントリ追加するだけ」になっていれば、レビューコストも下がる
  • グレースフルシャットダウンの実装は、Pub/Sub Pull Worker だけでなく Argo Workflows のロングランニングステップや Cloud Run の常駐処理にも同じ設計思想が適用できる: 「SIGTERMを受けたら新規受付を止め、処理中のものだけ完了させる」というパターンは汎用的
  • KEDAのbacklogベーススケーリングは、通知配信だけでなく画像リサイズ・注文確定メール送信など、あらゆるPub/Sub Pull Workerに横展開できる標準パターン: SubscriptionSizeの閾値（本問では50）はPod1台あたりの処理スループットから逆算して調整する
  • Autopilotのrequests適正化はコスト削減の即効性が高い: kubectl top podやDatadogのリソース使用率メトリクスで実測値を確認し、requestsを実測ピークの1.2〜1.5倍程度に絞ることで、機能を落とさずにコストを圧縮できる

今日のまとめ

カテゴリAでは、送信手段ごとに継承で状態とリトライロジックを共有していた BaseNotifier を、Protocol による構造的部分型と RetryPolicy への委譲に置き換えることで、継承の強い結合を解消し、NOTIFIER_FACTORIES レジストリで if/elif 分岐をOCP準拠の設計に改めた（Ch3/Ch6-7/Ch8/Ch10）。

カテゴリBでは、SIGTERM を無視していた Pub/Sub Pull Worker にグレースフルシャットダウンを実装し、Autopilotのrequests適正化でコストを、liveness/readinessProbeで死活監視を、KEDAのbacklogベーススケーリングでSLA遵守を、それぞれ構造的に改善した。どちらも共通するのは「暗黙のうちに発生していた副作用（継承の強制・強制終了による喪失）を、明示的な委譲・明示的なライフサイクル制御に置き換える」という設計思想である。

次のステップ

  • 発展問題: RetryPolicy に指数バックオフ+jitterを組み込み、NotificationError のサブタイプ（一時的エラー vs 永続的エラー）によってリトライ可否を分岐させる（2026-07-12の弱点補強Circuit Breakerと組み合わせる）
  • 発展問題: KEDAの ScaledObject に加えて、cooldownPeriod とスケールイン時のグレースフルシャットダウンの相互作用（スケールイン対象Podの選定とin-flightメッセージの関係）を検証する
  • 参考: 「良いコード・悪いコードで学ぶ設計入門」Ch3（カプセル化）/ Ch6-7（関心の分離）/ Ch8（条件分岐）/ Ch10（設計の悪魔）、Kubernetes Pod Lifecycle（terminationGracePeriodSeconds）、KEDA gcp-pubsub scaler ドキュメント、GKE Autopilot 課金モデル

自己評価（あとで記入）

自分の回答

気づき・メモ