A+B 複合 — discount_calculator 条件分岐地獄 Bad→Good（第8章: 早期return × テーブル駆動 × ポリシーパターン）× price-promotion-calculationバッチのGKE Autopilot移行（Argo Workflows並列シャーディング × リソースサイジング × 冪等性）

📋 コピー# tests/test_discount_calculator.py
import pytest
from discount_calculator import (
    AmountTier, CustomerTier, DiscountContext, calculate_discount_rate,
)


class TestCalculateDiscountRate:
    def test_gold_high_campaign_percentage_coupon(self):
        ctx = DiscountContext(
            CustomerTier.GOLD, 60_000, True, True, "percentage", 25, False
        )
        assert calculate_discount_rate(ctx) == pytest.approx(0.28)

    def test_clamped_at_max_discount_rate(self):
        # 複数ボーナスが重なっても上限0.30を超えない回帰テスト
        ctx = DiscountContext(
            CustomerTier.GOLD, 100_000, True, True, "percentage", 50, True
        )
        assert calculate_discount_rate(ctx) == pytest.approx(0.30)

    def test_adding_new_tier_requires_no_if_elif_edit(self):
        # 新tier追加はテーブルへの行追加のみで完結することを示す回帰テスト
        from discount_calculator import _BASE_RATE_TABLE
        assert (CustomerTier.BRONZE, AmountTier.LOW) in _BASE_RATE_TABLE

📋 コピー"""batch/process_shard.py — シャード単位の並列処理 × 冪等な書き込み(修正6) × OTel計装(修正7)"""
from __future__ import annotations

import argparse
import asyncio
from typing import Final

import httpx
from opentelemetry import metrics, trace

from discount_calculator import CustomerTier, DiscountContext, calculate_discount_rate

tracer = trace.get_tracer(__name__)
meter = metrics.get_meter(__name__)
_processed_counter = meter.create_counter("price_promotion.orders_processed")
_failed_counter = meter.create_counter("price_promotion.orders_failed")

# 修正4: order-serviceへの同時接続数を絞りI/O待ちを重ね合わせる(CPUコア数ではなくここが律速)
_MAX_CONCURRENT_REQUESTS: Final[int] = 20


async def process_order(client: httpx.AsyncClient, semaphore: asyncio.Semaphore, order_id: str) -> None:
    """1注文を取得・割引計算・冪等書き込みする。"""
    async with semaphore:
        with tracer.start_as_current_span("process_order") as span:  # 修正7
            span.set_attribute("order_id", order_id)
            try:
                order = (await client.get(f"/orders/{order_id}")).json()
                context = DiscountContext(
                    customer_tier=CustomerTier(order["customer_tier"]),
                    purchase_amount=order["purchase_amount"],
                    is_campaign_member=order["is_campaign_member"],
                    has_coupon=order["has_coupon"],
                    coupon_type=order.get("coupon_type"),
                    order_count=order["order_count"],
                    is_first_purchase=order["is_first_purchase"],
                )
                discount_rate = calculate_discount_rate(context)

                # 修正6: POST新規作成からPATCH upsertへ変更。リトライしても重複レコードを作らない
                await client.patch(
                    f"/orders/{order_id}/discount",
                    json={"discount_rate": discount_rate},
                    headers={"Idempotency-Key": f"price-promo-{order_id}-{order['target_date']}"},
                )
                _processed_counter.add(1)
            except httpx.HTTPError as exc:
                span.record_exception(exc)
                _failed_counter.add(1)
                raise  # ArgoのretryStrategyに処理を委ねる(shard単位リトライ)


async def process_shard(shard_id: int, shard_count: int) -> None:
    """担当シャード分の注文をasyncioで並行処理する。"""
    order_ids = await fetch_shard_order_ids(shard_id, shard_count)  # order_idのhash % shard_countで担当分を取得
    semaphore = asyncio.Semaphore(_MAX_CONCURRENT_REQUESTS)
    timeout = httpx.Timeout(connect=2.0, read=5.0)
    async with httpx.AsyncClient(base_url="https://order-service.internal", timeout=timeout) as client:
        await asyncio.gather(*(process_order(client, semaphore, oid) for oid in order_ids))


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--shard-id", type=int, required=True)
    parser.add_argument("--shard-count", type=int, required=True)
    args = parser.parse_args()
    asyncio.run(process_shard(args.shard_id, args.shard_count))

Bad vs Good 設計比較

確認コマンド

📋 コピー# 1. CronWorkflowが登録されていることを確認
argo cron list -n mops-batch
# Expected: price-promotion-calculation  Schedule=0 0 * * *  Suspended=false

# 2. 直近の実行が10シャード並列で走ったことを確認
argo get price-promotion-calculation-xxxxx -n mops-batch
# Expected: process-shard(0)〜process-shard(9)がすべてSucceeded

# 3. 各Podのリソース使用量がrequests/limitsの範囲内か確認
kubectl top pods -n mops-batch -l workflows.argoproj.io/workflow=price-promotion-calculation-xxxxx
# Expected: 各Podのメモリが256Mi〜512Miの範囲内

# 4. 冪等性: 同一注文への再実行で重複レコードが作られないことを確認
curl -s https://order-service.internal/orders/{order_id}/discount/history | jq 'length'
# Expected: 1 (再実行後も1件のまま)

# 5. ワークフロー全体のdurationがSLO(45分)以内か確認(DataDog)
# ダッシュボード: price_promotion.workflow_duration_seconds