# スケーラビリティの3つの次元を分析するフレームワーク
class ScalabilityAnalyzer:
    """システムのスケーラビリティを3つの次元で分析"""
 
    def __init__(self, system_name: str):
        self.system_name = system_name
        self.dimensions = {}
 
    def assess_dimension(self, dimension: str, current: float,
                         target: float, bottleneck: str,
                         strategy: str):
        """各次元のスケーラビリティを評価"""
        self.dimensions[dimension] = {
            "current": current,
            "target": target,
            "ratio": target / current if current > 0 else float('inf'),
            "bottleneck": bottleneck,
            "strategy": strategy,
        }
 
    def report(self) -> str:
        lines = [f"=== {self.system_name} スケーラビリティ分析 ===\n"]
        for dim, data in self.dimensions.items():
            lines.append(f"【{dim}】")
            lines.append(f"  現在: {data['current']:,.0f}")
            lines.append(f"  目標: {data['target']:,.0f}")
            lines.append(f"  倍率: {data['ratio']:.1f}x")
            lines.append(f"  ボトルネック: {data['bottleneck']}")
            lines.append(f"  戦略: {data['strategy']}")
            lines.append("")
        return "\n".join(lines)
 
 
# 使用例: ECサイトのスケーラビリティ分析
analyzer = ScalabilityAnalyzer("ECサイト")
 
# 1. 負荷スケーラビリティ（Load Scalability）
analyzer.assess_dimension(
    "負荷スケーラビリティ",
    current=1000,          # 現在のQPS
    target=50000,          # 目標QPS
    bottleneck="DBの接続数上限（max_connections=500）",
    strategy="Read Replica導入 + コネクションプーリング（PgBouncer）"
)
 
# 2. データスケーラビリティ（Data Scalability）
analyzer.assess_dimension(
    "データスケーラビリティ",
    current=500,           # 現在のデータ量（GB）
    target=50000,          # 目標データ量（GB）
    bottleneck="単一DBのストレージ上限（16TB）とクエリ性能低下",
    strategy="シャーディング（user_id基準）+ コールドデータのS3退避"
)
 
# 3. 地理的スケーラビリティ（Geographic Scalability）
analyzer.assess_dimension(
    "地理的スケーラビリティ",
    current=1,             # 現在のリージョン数
    target=3,              # 目標リージョン数
    bottleneck="単一リージョンでの海外ユーザーのレイテンシ（300ms+）",
    strategy="CDN導入 + マルチリージョンRead Replica + Edge Computing"
)
 
print(analyzer.report())
 
# 出力:
# === ECサイト スケーラビリティ分析 ===
#
# 【負荷スケーラビリティ】
#   現在: 1,000
#   目標: 50,000
#   倍率: 50.0x
#   ボトルネック: DBの接続数上限（max_connections=500）
#   戦略: Read Replica導入 + コネクションプーリング（PgBouncer）
#
# 【データスケーラビリティ】
#   現在: 500
#   目標: 50,000
#   倍率: 100.0x
#   ボトルネック: 単一DBのストレージ上限（16TB）とクエリ性能低下
#   戦略: シャーディング（user_id基準）+ コールドデータのS3退避
#
# 【地理的スケーラビリティ】
#   現在: 1
#   目標: 3
#   倍率: 3.0x
#   ボトルネック: 単一リージョンでの海外ユーザーのレイテンシ（300ms+）
#   戦略: CDN導入 + マルチリージョンRead Replica + Edge Computing

import math
 
def vertical_scaling_cost(base_cost: float, scale_factor: int) -> float:
    """
    垂直スケーリングのコストモデル
 
    なぜ指数的に増加するか:
    1. CPUは周波数を上げるほど消費電力が指数的に増える（P ∝ f^3）
    2. 大容量メモリは製造コストが高い（ECCメモリ、多チャネル）
    3. 高性能マシンは市場が小さく、競争が少ない
    4. ベンダーロックインによるプレミアム価格
 
    実例: AWS EC2インスタンスの価格
    - t3.micro  (2vCPU,  1GB):   $0.0104/h  → 基準
    - t3.xlarge (4vCPU,  16GB):  $0.1664/h  → 16倍のスペック、16倍の価格
    - r5.4xlarge(16vCPU, 128GB): $1.008/h   → 128倍のスペック、97倍の価格
    - x1.32xlarge(128vCPU,1952GB):$13.338/h → 1952倍のスペック、1282倍の価格
    """
    # 実測データに基づく指数: 約1.6乗
    return base_cost * (scale_factor ** 1.6)
 
 
def horizontal_scaling_cost(base_cost: float, scale_factor: int,
                            overhead_pct: float = 10) -> float:
    """
    水平スケーリングのコストモデル
 
    なぜほぼ線形か:
    1. 同じスペックのマシンを追加するため、単価が変わらない
    2. クラウドでは大量購入による割引もある
    3. オーバーヘッドは固定的（LB、管理ノード等）
 
    overhead_pctの内訳:
    - ロードバランサー: 2-3%
    - 管理/モニタリング: 2-3%
    - サービスディスカバリ: 1-2%
    - 通信オーバーヘッド: 2-3%
    """
    overhead = 1 + overhead_pct / 100
    return base_cost * scale_factor * overhead
 
 
# コスト比較シミュレーション
base = 1000  # 基本コスト $1000/月
print("=== スケーリングコスト比較 ===\n")
print(f"{'倍率':>5s}  {'垂直コスト':>12s}  {'水平コスト':>12s}  "
      f"{'差額':>12s}  {'垂直/水平':>10s}")
print("-" * 65)
 
for factor in [1, 2, 4, 8, 16, 32, 64, 128]:
    v_cost = vertical_scaling_cost(base, factor)
    h_cost = horizontal_scaling_cost(base, factor)
    ratio = v_cost / h_cost if h_cost > 0 else 0
    print(f"  x{factor:<3d}  ${v_cost:>10,.0f}  ${h_cost:>10,.0f}  "
          f"${v_cost - h_cost:>10,.0f}  {ratio:>8.1f}x")
 
# 出力:
# === スケーリングコスト比較 ===
#
# 倍率    垂直コスト    水平コスト          差額    垂直/水平
# -----------------------------------------------------------------
#   x1       $1,000       $1,100         $-100       0.9x
#   x2       $3,031       $2,200          $831       1.4x
#   x4       $9,190       $4,400        $4,790       2.1x
#   x8      $27,858       $8,800       $19,058       3.2x
#   x16     $84,449      $17,600       $66,849       4.8x
#   x32    $256,000      $35,200      $220,800       7.3x
#   x64    $776,247      $70,400      $705,847      11.0x
#   x128  $2,353,487     $140,800    $2,212,687      16.7x

# クラウドプロバイダー別の最大インスタンスサイズ（2024年時点の概算）
cloud_max_instances = {
    "AWS": {
        "max_vcpu": 448,      # u-24tb1.metal
        "max_memory_gb": 24576,  # 24 TB
        "max_local_storage_tb": 60,
        "cost_per_hour": 218.40,
        "instance_type": "u-24tb1.metal",
        "use_case": "SAP HANA等のインメモリDB",
    },
    "GCP": {
        "max_vcpu": 416,      # m3-megamem-128
        "max_memory_gb": 8192,
        "max_local_storage_tb": 9,
        "cost_per_hour": 125.75,
        "instance_type": "m3-megamem-128",
        "use_case": "大規模OLAP",
    },
    "Azure": {
        "max_vcpu": 416,      # M416ms_v2
        "max_memory_gb": 11400,
        "max_local_storage_tb": 14,
        "cost_per_hour": 110.22,
        "instance_type": "M416ms_v2",
        "use_case": "SAP HANA",
    },
}
 
print("=== クラウド最大インスタンスサイズ ===\n")
for provider, spec in cloud_max_instances.items():
    print(f"  {provider}:")
    print(f"    インスタンス: {spec['instance_type']}")
    print(f"    vCPU: {spec['max_vcpu']}")
    print(f"    メモリ: {spec['max_memory_gb']:,} GB ({spec['max_memory_gb']/1024:.0f} TB)")
    print(f"    ストレージ: {spec['max_local_storage_tb']} TB")
    print(f"    コスト: ${spec['cost_per_hour']:.2f}/時間 "
          f"(${spec['cost_per_hour']*730:,.0f}/月)")
    print(f"    用途: {spec['use_case']}")
    print()
 
# 教訓:
# - 最大インスタンスでも vCPU 448、メモリ 24TB が上限
# - 月額 $160,000 以上のコストがかかる
# - これ以上のスケーリングには水平展開が必須

from fastapi import FastAPI, Depends, Header, HTTPException
from pydantic import BaseModel
import redis
import json
import uuid
import time
from typing import Optional
 
app = FastAPI()
 
# 共有ストア（Redis）への接続
# 全てのアプリサーバーが同じRedisクラスタを参照する
redis_client = redis.Redis(
    host="redis-cluster.internal",
    port=6379,
    decode_responses=True,
    # コネクションプールでパフォーマンス確保
    connection_pool=redis.ConnectionPool(max_connections=50)
)
 
 
# ========================================
# アンチパターン: ステートフルなAPI
# ========================================
 
# NG: メモリ内にセッションを保持
sessions_local = {}  # これはサーバーごとに異なる!
user_cache_local = {}  # Server 1 のキャッシュは Server 2 にはない
 
@app.get("/bad/profile")
def bad_get_profile(session_id: str):
    """
    アンチパターン: ローカルメモリにセッション保存
 
    問題点:
    1. Server 1 で作成したセッションは Server 2 では無効
    2. サーバー再起動でセッション全消失
    3. LBのスティッキーセッションが必要になり、負荷分散が不均等に
    4. Auto Scalingでサーバーが増減するたびにセッションが失われる
    """
    user = sessions_local.get(session_id)
    if not user:
        return {"error": "session not found"}
    return user
 
 
# ========================================
# ベストプラクティス: ステートレスなAPI
# ========================================
 
class SessionManager:
    """
    ステートレスなセッション管理
 
    設計原則:
    - サーバーは状態を持たない（Shared-Nothing Architecture）
    - 全ての状態は外部ストア（Redis）に保存
    - セッションIDのみをクライアントに返す
    - TTL（有効期限）で自動クリーンアップ
    """
 
    def __init__(self, redis_client: redis.Redis,
                 session_ttl: int = 3600):
        self.redis = redis_client
        self.session_ttl = session_ttl  # デフォルト1時間
 
    def create_session(self, user_id: str, user_data: dict) -> str:
        """新しいセッションを作成"""
        session_id = str(uuid.uuid4())
        session_data = {
            "user_id": user_id,
            "data": json.dumps(user_data),
            "created_at": time.time(),
        }
        # Redisに保存（TTL付き）
        self.redis.hset(f"session:{session_id}", mapping=session_data)
        self.redis.expire(f"session:{session_id}", self.session_ttl)
        return session_id
 
    def get_session(self, session_id: str) -> Optional[dict]:
        """セッションを取得（どのサーバーからでも同じ結果）"""
        data = self.redis.hgetall(f"session:{session_id}")
        if not data:
            return None
        # アクセスのたびにTTLを延長（スライディングウィンドウ）
        self.redis.expire(f"session:{session_id}", self.session_ttl)
        return {
            "user_id": data["user_id"],
            "data": json.loads(data["data"]),
            "created_at": float(data["created_at"]),
        }
 
    def delete_session(self, session_id: str) -> bool:
        """セッションを削除（ログアウト時）"""
        return self.redis.delete(f"session:{session_id}") > 0
 
 
session_mgr = SessionManager(redis_client)
 
 
@app.get("/good/profile")
def good_get_profile(x_session_id: str = Header(...)):
    """
    ベストプラクティス: Redisにセッション保存
 
    どのサーバーにリクエストが来ても同じ結果を返す。
    LBのアルゴリズムに関係なく正常動作。
    """
    session = session_mgr.get_session(x_session_id)
    if not session:
        raise HTTPException(status_code=401, detail="Invalid session")
    return {"user_id": session["user_id"], "profile": session["data"]}
 
 
class LoginRequest(BaseModel):
    username: str
    password: str
 
@app.post("/auth/login")
def login(req: LoginRequest):
    """ログイン: セッションを作成してIDを返す"""
    # 実際にはDBでユーザー認証を行う
    user_data = {"name": req.username, "role": "user"}
    session_id = session_mgr.create_session(req.username, user_data)
    return {"session_id": session_id, "expires_in": 3600}
 
 
@app.post("/auth/logout")
def logout(x_session_id: str = Header(...)):
    """ログアウト: セッションを削除"""
    session_mgr.delete_session(x_session_id)
    return {"success": True}
 
# 出力例（curlで実行した場合）:
# $ curl -X POST http://localhost:8000/auth/login \
#   -H "Content-Type: application/json" \
#   -d '{"username": "gaku", "password": "secret"}'
# {"session_id": "a1b2c3d4-...", "expires_in": 3600}
#
# $ curl http://localhost:8000/good/profile \
#   -H "X-Session-Id: a1b2c3d4-..."
# {"user_id": "gaku", "profile": {"name": "gaku", "role": "user"}}

# Twelve-Factor App の12の原則とスケーラビリティとの関係
 
twelve_factors = [
    ("I.   コードベース", "1つのコードベースからN個のデプロイ",
     "全インスタンスが同じコードを実行"),
    ("II.  依存関係", "明示的に宣言し分離する",
     "requirements.txt / package.json で完結"),
    ("III. 設定", "環境変数に格納",
     "環境ごとの差異を環境変数で吸収。コードにDB接続先をハードコードしない"),
    ("IV.  バックエンド", "付帯サービスとしてアタッチ",
     "DB/Redis/MQは外部サービス。URLで接続先を切り替え可能"),
    ("V.   ビルド/リリース/実行", "3段階を厳密に分離",
     "ビルド成果物は不変。同じイメージを全サーバーにデプロイ"),
    ("VI.  プロセス", "ステートレスなプロセスとして実行 ★最重要★",
     "メモリやファイルシステムに状態を持たない。共有ストアを使う"),
    ("VII. ポートバインディング", "ポートバインディングでサービス公開",
     "自己完結型。リバースプロキシなしで直接ポートをリッスン"),
    ("VIII.並行性", "プロセスモデルによるスケールアウト",
     "水平スケーリング。プロセスを増やして負荷分散"),
    ("IX.  廃棄容易性", "高速な起動とグレースフルシャットダウン",
     "Auto Scalingでの迅速な追加/削除を可能にする"),
    ("X.   開発/本番一致", "開発、ステージング、本番をできるだけ一致させる",
     "Docker等で環境差異を最小化"),
    ("XI.  ログ", "イベントストリームとして扱う",
     "stdoutに出力。集約は外部ツール（Fluentd/CloudWatch）で"),
    ("XII. 管理プロセス", "ワンオフの管理タスクとして実行",
     "マイグレーション等は通常プロセスと同じ環境で実行"),
]
 
print("=== Twelve-Factor App とスケーラビリティ ===\n")
for factor, description, scalability_note in twelve_factors:
    print(f"  {factor}")
    print(f"    内容: {description}")
    print(f"    スケーラビリティ: {scalability_note}")
    print()
 
# 出力:
# === Twelve-Factor App とスケーラビリティ ===
#
#   I.   コードベース
#     内容: 1つのコードベースからN個のデプロイ
#     スケーラビリティ: 全インスタンスが同じコードを実行
#
#   II.  依存関係
#     内容: 明示的に宣言し分離する
#     スケーラビリティ: requirements.txt / package.json で完結
#   ...

import hashlib
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class Task:
    id: str
    payload: dict
    priority: int = 0
 
class ConsistentHashRing:
    """
    コンシステントハッシングによるタスク分配
 
    なぜ単純なモジュロ演算（hash % N）ではダメなのか:
    - ワーカーをN=3→4に増やすと、ほぼ全てのタスクの割り当て先が変わる
    - キャッシュがあった場合、全てのキャッシュが無効になる
 
    コンシステントハッシングでは:
    - ワーカーをN=3→4に増やしても、約25%のタスクしか移動しない
    - K/N のタスクのみ再割り当て（K=タスク数、N=新ワーカー数）
    """
 
    def __init__(self, replicas: int = 150):
        self.replicas = replicas  # 各ノードの仮想ノード数
        self.ring = {}            # hash -> node_id
        self.sorted_keys = []     # ソート済みハッシュ値
 
    def _hash(self, key: str) -> int:
        return int(hashlib.md5(key.encode()).hexdigest(), 16)
 
    def add_node(self, node_id: str):
        """ノードをリングに追加"""
        for i in range(self.replicas):
            virtual_key = f"{node_id}:{i}"
            h = self._hash(virtual_key)
            self.ring[h] = node_id
            self.sorted_keys.append(h)
        self.sorted_keys.sort()
 
    def remove_node(self, node_id: str):
        """ノードをリングから削除"""
        for i in range(self.replicas):
            virtual_key = f"{node_id}:{i}"
            h = self._hash(virtual_key)
            del self.ring[h]
            self.sorted_keys.remove(h)
 
    def get_node(self, key: str) -> str:
        """キーに対応するノードを取得"""
        if not self.ring:
            raise ValueError("No nodes in the ring")
 
        h = self._hash(key)
        # ハッシュ値以上の最小のキーを見つける（時計回り）
        for ring_key in self.sorted_keys:
            if ring_key >= h:
                return self.ring[ring_key]
        # 末尾に到達したら先頭に戻る
        return self.ring[self.sorted_keys[0]]
 
 
# デモ: ワーカー追加時のタスク再配置率を計測
def measure_redistribution():
    """ワーカー追加時のタスク再配置率を測定"""
 
    # 1000個のタスクを用意
    tasks = [f"task-{i}" for i in range(1000)]
 
    # 3ワーカーで初期配置
    ring_before = ConsistentHashRing(replicas=150)
    for worker in ["worker-1", "worker-2", "worker-3"]:
        ring_before.add_node(worker)
 
    assignments_before = {task: ring_before.get_node(task) for task in tasks}
 
    # 4ワーカーに増やす
    ring_after = ConsistentHashRing(replicas=150)
    for worker in ["worker-1", "worker-2", "worker-3", "worker-4"]:
        ring_after.add_node(worker)
 
    assignments_after = {task: ring_after.get_node(task) for task in tasks}
 
    # 再配置されたタスクをカウント
    moved = sum(1 for task in tasks
                if assignments_before[task] != assignments_after[task])
 
    print(f"=== コンシステントハッシング: ワーカー追加の影響 ===")
    print(f"  タスク数: {len(tasks)}")
    print(f"  ワーカー: 3 → 4")
    print(f"  再配置タスク: {moved} ({moved/len(tasks)*100:.1f}%)")
    print(f"  理論値: {1/4*100:.1f}% (1/N)")
    print()
 
    # 比較: 単純なモジュロ演算
    moved_modulo = sum(
        1 for task in tasks
        if (int(hashlib.md5(task.encode()).hexdigest(), 16) % 3 !=
            int(hashlib.md5(task.encode()).hexdigest(), 16) % 4)
    )
    print(f"  参考: モジュロ演算での再配置: {moved_modulo} "
          f"({moved_modulo/len(tasks)*100:.1f}%)")
 
measure_redistribution()
 
# 出力:
# === コンシステントハッシング: ワーカー追加の影響 ===
#   タスク数: 1000
#   ワーカー: 3 → 4
#   再配置タスク: 約250 (約25.0%)
#   理論値: 25.0% (1/N)
#
#   参考: モジュロ演算での再配置: 約750 (約75.0%)

from enum import Enum
from dataclasses import dataclass
from typing import List, Optional
import time
 
class ScalingPolicyType(Enum):
    TARGET_TRACKING = "target_tracking"     # 目標追跡型
    STEP_SCALING = "step_scaling"           # ステップ型
    SCHEDULED = "scheduled"                 # スケジュール型
    PREDICTIVE = "predictive"              # 予測型
 
 
@dataclass
class ScalingConfig:
    """Auto Scaling の設定"""
    min_capacity: int = 2
    max_capacity: int = 20
    scale_out_cooldown: int = 180   # 秒（スケールアウト後の待機時間）
    scale_in_cooldown: int = 300    # 秒（スケールイン後の待機時間）
 
 
class TargetTrackingPolicy:
    """
    目標追跡型ポリシー（最も一般的）
 
    仕組み:
    - 特定のメトリクスを目標値に維持するように自動的にインスタンス数を調整
    - PID制御に似た仕組みで、オーバーシュートを防ぐ
 
    メリット:
    - 設定がシンプル（目標値を1つ指定するだけ）
    - AWSが自動的にCloudWatchアラームを作成・管理
 
    デメリット:
    - 単一メトリクスでしか判断できない
    - 突発的な負荷変動への反応が遅い
    """
 
    def __init__(self, target_value: float, metric_name: str = "CPUUtilization"):
        self.target_value = target_value
        self.metric_name = metric_name
 
    def calculate_desired(self, current_instances: int,
                          current_metric: float) -> int:
        """
        必要インスタンス数を計算
 
        公式: desired = ceil(current_instances * (current_metric / target_metric))
        """
        import math
        desired = math.ceil(
            current_instances * (current_metric / self.target_value)
        )
        return desired
 
 
class StepScalingPolicy:
    """
    ステップ型ポリシー
 
    仕組み:
    - メトリクスの値に応じて段階的にインスタンスを追加/削除
    - より細かい制御が可能
 
    メリット:
    - 負荷の大きさに応じた段階的な応答
    - 複数のステップを定義可能
 
    デメリット:
    - 設定が複雑
    - ステップ間の境界値の設定にチューニングが必要
    """
 
    def __init__(self):
        self.scale_out_steps = []  # (threshold, adjustment)
        self.scale_in_steps = []
 
    def add_scale_out_step(self, threshold: float, adjustment: int):
        """スケールアウトのステップを追加"""
        self.scale_out_steps.append((threshold, adjustment))
        self.scale_out_steps.sort(key=lambda x: x[0])
 
    def add_scale_in_step(self, threshold: float, adjustment: int):
        """スケールインのステップを追加"""
        self.scale_in_steps.append((threshold, adjustment))
        self.scale_in_steps.sort(key=lambda x: x[0], reverse=True)
 
    def calculate_adjustment(self, current_metric: float) -> int:
        """メトリクスに基づく調整数を計算"""
        # スケールアウト
        for threshold, adjustment in reversed(self.scale_out_steps):
            if current_metric >= threshold:
                return adjustment
 
        # スケールイン
        for threshold, adjustment in self.scale_in_steps:
            if current_metric <= threshold:
                return adjustment  # 負の値
 
        return 0  # 変更なし
 
 
class AutoScaler:
    """Auto Scalerのシミュレーター"""
 
    def __init__(self, config: ScalingConfig,
                 policy: TargetTrackingPolicy):
        self.config = config
        self.policy = policy
        self.current_instances = config.min_capacity
        self.last_scale_time = 0
        self.history = []
 
    def evaluate(self, current_metric: float, timestamp: float) -> dict:
        """現在のメトリクスに基づいてスケーリング判断"""
        desired = self.policy.calculate_desired(
            self.current_instances, current_metric)
 
        # min/maxの範囲に制限
        desired = max(self.config.min_capacity,
                      min(self.config.max_capacity, desired))
 
        # クールダウン期間のチェック
        elapsed = timestamp - self.last_scale_time
        if desired > self.current_instances:
            cooldown = self.config.scale_out_cooldown
            action = "SCALE_OUT"
        elif desired < self.current_instances:
            cooldown = self.config.scale_in_cooldown
            action = "SCALE_IN"
        else:
            action = "NO_CHANGE"
            cooldown = 0
 
        can_scale = elapsed >= cooldown or self.last_scale_time == 0
 
        result = {
            "timestamp": timestamp,
            "metric": current_metric,
            "current": self.current_instances,
            "desired": desired,
            "action": action,
            "can_scale": can_scale,
            "cooldown_remaining": max(0, cooldown - elapsed),
        }
 
        if can_scale and desired != self.current_instances:
            self.current_instances = desired
            self.last_scale_time = timestamp
 
        self.history.append(result)
        return result
 
 
# シミュレーション
config = ScalingConfig(min_capacity=2, max_capacity=20)
policy = TargetTrackingPolicy(target_value=70, metric_name="CPUUtilization")
scaler = AutoScaler(config, policy)
 
# 1日の負荷パターン（CPU使用率%）
# 深夜低負荷 → 朝方増加 → 昼ピーク → 夕方ピーク → 夜間低下
load_pattern = [
    (0, 20), (3, 15), (6, 30), (9, 65), (10, 80),
    (11, 90), (12, 95), (13, 85), (14, 75), (15, 70),
    (16, 60), (17, 80), (18, 90), (19, 85), (20, 70),
    (21, 50), (22, 35), (23, 25),
]
 
print("=== Auto Scaling シミュレーション (1日) ===\n")
print(f"{'時刻':>5s}  {'CPU%':>5s}  {'現在':>4s}  {'目標':>4s}  "
      f"{'アクション':>10s}  {'状態'}")
print("-" * 60)
 
for hour, cpu in load_pattern:
    timestamp = hour * 3600
    result = scaler.evaluate(cpu, timestamp)
    status = ""
    if not result["can_scale"] and result["action"] != "NO_CHANGE":
        status = f"(CD: {result['cooldown_remaining']:.0f}s)"
    print(f"  {hour:02d}:00  {cpu:>4d}%  {result['current']:>4d}  "
          f"{result['desired']:>4d}  {result['action']:>10s}  {status}")
 
# 出力:
# === Auto Scaling シミュレーション (1日) ===
#
#  時刻   CPU%   現在   目標   アクション   状態
# ------------------------------------------------------------
#  00:00    20%     2     2    NO_CHANGE
#  03:00    15%     2     2    NO_CHANGE
#  06:00    30%     2     2    NO_CHANGE
#  09:00    65%     2     2    NO_CHANGE
#  10:00    80%     2     3    SCALE_OUT
#  11:00    90%     3     4    SCALE_OUT
#  12:00    95%     4     6    SCALE_OUT
#  13:00    85%     6     8    SCALE_OUT
#  14:00    75%     8     9    SCALE_OUT
#  15:00    70%     9     9    NO_CHANGE
#  16:00    60%     9     8    SCALE_IN
#  17:00    80%     8    10    SCALE_OUT
#  18:00    90%    10    13    SCALE_OUT
#  19:00    85%    13    16    SCALE_OUT
#  20:00    70%    16    16    NO_CHANGE
#  21:00    50%    16    12    SCALE_IN
#  22:00    35%    12     6    SCALE_IN
#  23:00    25%     6     3    SCALE_IN

# AWS CDK (Python) でのAuto Scaling設定例
# 実際にAWSにデプロイ可能なコード
 
"""
from aws_cdk import (
    Stack,
    aws_ec2 as ec2,
    aws_autoscaling as autoscaling,
    aws_elasticloadbalancingv2 as elbv2,
    Duration,
)
from constructs import Construct
 
class ScalableWebServiceStack(Stack):
    def __init__(self, scope: Construct, id: str, **kwargs):
        super().__init__(scope, id, **kwargs)
 
        # VPC
        vpc = ec2.Vpc(self, "VPC", max_azs=3)
 
        # Auto Scaling Group
        asg = autoscaling.AutoScalingGroup(
            self, "ASG",
            vpc=vpc,
            instance_type=ec2.InstanceType("t3.medium"),
            machine_image=ec2.AmazonLinuxImage(
                generation=ec2.AmazonLinuxGeneration.AMAZON_LINUX_2
            ),
            min_capacity=2,        # 最小2台
            max_capacity=20,       # 最大20台
            desired_capacity=4,    # 初期4台
            health_check=autoscaling.HealthCheck.elb(
                grace=Duration.minutes(5)
            ),
        )
 
        # 目標追跡スケーリングポリシー (CPU)
        asg.scale_on_cpu_utilization(
            "CpuScaling",
            target_utilization_percent=70,
            cooldown=Duration.minutes(3),
            estimated_instance_warmup=Duration.minutes(5),
        )
 
        # 目標追跡スケーリングポリシー (リクエスト数)
        asg.scale_on_request_count(
            "RequestScaling",
            target_requests_per_minute=1000,
            # 各インスタンスあたり1000 req/min を目標
        )
 
        # ステップスケーリングポリシー (メモリ)
        asg.scale_on_metric(
            "MemoryScaling",
            metric=asg.metric("MemoryUtilization"),
            scaling_steps=[
                autoscaling.ScalingInterval(upper=30, change=-2),   # 30%以下: 2台削減
                autoscaling.ScalingInterval(lower=30, upper=70, change=0),  # 維持
                autoscaling.ScalingInterval(lower=70, upper=85, change=2),  # +2台
                autoscaling.ScalingInterval(lower=85, change=4),    # 85%以上: +4台
            ],
        )
 
        # スケジュールスケーリング（セール対応）
        asg.scale_on_schedule(
            "SaleStart",
            schedule=autoscaling.Schedule.cron(
                hour="8", minute="50"  # セール10分前
            ),
            min_capacity=10,
            max_capacity=50,
        )
        asg.scale_on_schedule(
            "SaleEnd",
            schedule=autoscaling.Schedule.cron(
                hour="23", minute="0"
            ),
            min_capacity=2,
            max_capacity=20,
        )
 
        # ALB
        lb = elbv2.ApplicationLoadBalancer(
            self, "ALB",
            vpc=vpc,
            internet_facing=True,
        )
 
        listener = lb.add_listener("Listener", port=443)
        listener.add_targets("AppTarget",
            port=8080,
            targets=[asg],
            health_check=elbv2.HealthCheck(
                path="/health",
                interval=Duration.seconds(30),
                healthy_threshold_count=2,
                unhealthy_threshold_count=3,
            ),
        )
"""
 
# 上記は実際のAWS CDKコードだが、ここではシミュレーションを示す
print("=== Auto Scaling 設定サマリー ===")
print()
print("  ポリシー1: CPU目標追跡 (70%)")
print("  ポリシー2: リクエスト目標追跡 (1000 req/min per instance)")
print("  ポリシー3: メモリステップスケーリング")
print("    - < 30%: -2台")
print("    - 30-70%: 維持")
print("    - 70-85%: +2台")
print("    - > 85%: +4台")
print("  ポリシー4: スケジュール（セール時 min=10, max=50）")
print()
print("  ※ 複数ポリシーが同時発火した場合、最も多いインスタンス数が採用される")

import asyncio
import time
from dataclasses import dataclass, field
from typing import List
import statistics
 
@dataclass
class LoadTestResult:
    """負荷テストの結果"""
    total_requests: int = 0
    success: int = 0
    errors: int = 0
    latencies: List[float] = field(default_factory=list)
    elapsed: float = 0.0
 
    @property
    def throughput(self) -> float:
        return self.success / self.elapsed if self.elapsed > 0 else 0
 
    @property
    def p50(self) -> float:
        if not self.latencies:
            return 0
        sorted_lat = sorted(self.latencies)
        return sorted_lat[len(sorted_lat) // 2]
 
    @property
    def p95(self) -> float:
        if not self.latencies:
            return 0
        sorted_lat = sorted(self.latencies)
        return sorted_lat[int(len(sorted_lat) * 0.95)]
 
    @property
    def p99(self) -> float:
        if not self.latencies:
            return 0
        sorted_lat = sorted(self.latencies)
        return sorted_lat[int(len(sorted_lat) * 0.99)]
 
    @property
    def error_rate(self) -> float:
        return self.errors / self.total_requests * 100 if self.total_requests > 0 else 0
 
 
async def load_test(url: str, total_requests: int,
                    concurrency: int) -> LoadTestResult:
    """
    非同期HTTP負荷テスト
 
    使い方:
    asyncio.run(load_test("http://localhost:8080/api/health", 10000, 100))
 
    パラメータ:
    - url: テスト対象のエンドポイント
    - total_requests: 総リクエスト数
    - concurrency: 同時実行数（並行度）
    """
    import aiohttp
 
    semaphore = asyncio.Semaphore(concurrency)
    result = LoadTestResult(total_requests=total_requests)
 
    async def make_request(session):
        async with semaphore:
            start = time.monotonic()
            try:
                async with session.get(url) as resp:
                    await resp.text()
                    latency = (time.monotonic() - start) * 1000  # ms
                    result.latencies.append(latency)
                    if resp.status == 200:
                        result.success += 1
                    else:
                        result.errors += 1
            except Exception:
                result.errors += 1
 
    async with aiohttp.ClientSession() as session:
        tasks = [make_request(session) for _ in range(total_requests)]
        start_time = time.monotonic()
        await asyncio.gather(*tasks)
        result.elapsed = time.monotonic() - start_time
 
    return result
 
 
def print_load_test_report(result: LoadTestResult, concurrency: int):
    """負荷テスト結果のレポートを出力"""
    print("=== 負荷テスト結果 ===\n")
    print(f"  リクエスト総数: {result.total_requests:,}")
    print(f"  同時実行数:     {concurrency}")
    print(f"  経過時間:       {result.elapsed:.2f}秒")
    print(f"  成功:           {result.success:,}")
    print(f"  エラー:         {result.errors:,} ({result.error_rate:.1f}%)")
    print(f"  スループット:   {result.throughput:,.0f} req/s")
    print(f"  レイテンシ P50: {result.p50:.1f}ms")
    print(f"  レイテンシ P95: {result.p95:.1f}ms")
    print(f"  レイテンシ P99: {result.p99:.1f}ms")
 
    # SLO判定
    print(f"\n  === SLO判定 ===")
    slo_latency = 200  # ms
    slo_error_rate = 1  # %
    slo_throughput = 1000  # req/s
 
    latency_ok = result.p99 < slo_latency
    error_ok = result.error_rate < slo_error_rate
    throughput_ok = result.throughput > slo_throughput
 
    print(f"  レイテンシ P99 < {slo_latency}ms: "
          f"{'PASS' if latency_ok else 'FAIL'} ({result.p99:.1f}ms)")
    print(f"  エラー率 < {slo_error_rate}%: "
          f"{'PASS' if error_ok else 'FAIL'} ({result.error_rate:.1f}%)")
    print(f"  スループット > {slo_throughput} req/s: "
          f"{'PASS' if throughput_ok else 'FAIL'} ({result.throughput:,.0f} req/s)")
 
# 使用方法:
# result = asyncio.run(load_test("http://localhost:8080/api/health", 10000, 100))
# print_load_test_report(result, 100)
 
# サンプル出力:
# === 負荷テスト結果 ===
#
#   リクエスト総数: 10,000
#   同時実行数:     100
#   経過時間:       8.53秒
#   成功:           9,950
#   エラー:         50 (0.5%)
#   スループット:   1,166 req/s
#   レイテンシ P50: 45.2ms
#   レイテンシ P95: 123.7ms
#   レイテンシ P99: 198.3ms
#
#   === SLO判定 ===
#   レイテンシ P99 < 200ms: PASS (198.3ms)
#   エラー率 < 1%: PASS (0.5%)
#   スループット > 1000 req/s: PASS (1,166 req/s)

# シャーディング戦略のシミュレーション
import hashlib
 
class ShardingStrategy:
    """シャーディング戦略の比較"""
 
    @staticmethod
    def range_based(user_id: int, shard_count: int) -> int:
        """
        レンジベースシャーディング
 
        仕組み: user_id の範囲でシャードを決定
        例: user 1-1M → Shard 0, user 1M-2M → Shard 1, ...
 
        長所: 範囲クエリが効率的
        短所: ホットスポット（新規ユーザーが集中するシャード）
        """
        users_per_shard = 1_000_000  # 100万ユーザー/シャード
        return min(user_id // users_per_shard, shard_count - 1)
 
    @staticmethod
    def hash_based(user_id: int, shard_count: int) -> int:
        """
        ハッシュベースシャーディング
 
        仕組み: user_id のハッシュ値でシャードを決定
        公式: shard = hash(user_id) % shard_count
 
        長所: データが均等に分散
        短所: 範囲クエリが全シャードに散る、リシャーディングが大変
        """
        h = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
        return h % shard_count
 
    @staticmethod
    def directory_based(user_id: int, directory: dict) -> int:
        """
        ディレクトリベースシャーディング
 
        仕組み: ルックアップテーブル（ディレクトリ）でシャードを決定
        例: user_id → shard_id のマッピングをDBに保存
 
        長所: 柔軟なリバランシングが可能
        短所: ディレクトリがSPOFになる、追加のルックアップレイテンシ
        """
        return directory.get(user_id, 0)
 
 
# シャード分布のシミュレーション
def simulate_shard_distribution(strategy_name: str, strategy_func,
                                user_count: int, shard_count: int):
    """各シャーディング戦略のデータ分布を検証"""
    distribution = [0] * shard_count
 
    for user_id in range(user_count):
        shard = strategy_func(user_id, shard_count)
        distribution[shard] += 1
 
    avg = user_count / shard_count
    max_dev = max(abs(d - avg) / avg * 100 for d in distribution)
    min_count = min(distribution)
    max_count = max(distribution)
 
    print(f"\n=== {strategy_name} (ユーザー: {user_count:,}, シャード: {shard_count}) ===")
    for i, count in enumerate(distribution):
        bar = "#" * int(count / user_count * 100)
        deviation = (count - avg) / avg * 100
        print(f"  Shard {i}: {count:>8,} ({deviation:>+6.1f}%)  {bar}")
    print(f"  最大偏差: {max_dev:.1f}% (理想は0%)")
 
# レンジベース
simulate_shard_distribution(
    "レンジベース", ShardingStrategy.range_based, 3_500_000, 4)
 
# ハッシュベース
simulate_shard_distribution(
    "ハッシュベース", ShardingStrategy.hash_based, 3_500_000, 4)
 
# 出力:
# === レンジベース (ユーザー: 3,500,000, シャード: 4) ===
#   Shard 0: 1,000,000 (+14.3%)  #############################
#   Shard 1: 1,000,000 (+14.3%)  #############################
#   Shard 2: 1,000,000 (+14.3%)  #############################
#   Shard 3:   500,000 (-42.9%)  ##############
#   最大偏差: 42.9% (理想は0%)
#
# === ハッシュベース (ユーザー: 3,500,000, シャード: 4) ===
#   Shard 0:   875,123 (+0.0%)   #########################
#   Shard 1:   874,892 (-0.0%)   #########################
#   Shard 2:   875,034 (+0.0%)   #########################
#   Shard 3:   874,951 (-0.0%)   #########################
#   最大偏差: 0.0% (理想は0%)

# --- NG例 ---
 
class BadUserService:
    """アンチパターン: ステートフルなサービスの水平スケール"""
 
    def __init__(self):
        self.cache = {}  # インスタンスローカルのキャッシュ
        self.session_store = {}  # ローカルセッション
 
    def get_user(self, user_id: str) -> dict:
        """
        問題点:
        - Server 1 のキャッシュにあるデータは Server 2 では使えない
        - スティッキーセッションに依存するとLBが制限される
        - サーバー追加時にキャッシュヒット率が低下
        - サーバー障害時にキャッシュとセッションが全て消失
        """
        if user_id in self.cache:
            return self.cache[user_id]  # Server 1にしかない!
        user = {"id": user_id, "name": f"User {user_id}"}
        self.cache[user_id] = user
        return user
 
 
# --- OK例 ---
 
class GoodUserService:
    """ベストプラクティス: ステートレスなサービス"""
 
    def __init__(self, redis_client, db_client):
        self.redis = redis_client   # 共有キャッシュ
        self.db = db_client         # 共有DB
 
    def get_user(self, user_id: str) -> dict:
        """
        どのサーバーにリクエストが来ても同じ結果を返す
 
        キャッシュ戦略: Cache-Aside (Lazy Loading)
        1. Redisにキャッシュがあればそれを返す
        2. なければDBから取得してRedisにキャッシュ
        3. TTL付きで自動的に期限切れ
        """
        # 1. 共有キャッシュをチェック
        cached = self.redis.get(f"user:{user_id}")
        if cached:
            return json.loads(cached)
 
        # 2. DBから取得
        user = self.db.query(f"SELECT * FROM users WHERE id = %s", user_id)
        if not user:
            return None
 
        # 3. キャッシュに保存（TTL 5分）
        self.redis.setex(f"user:{user_id}", 300, json.dumps(user))
        return user

# 演習1: スケーリング戦略の選択
 
scenarios = [
    {
        "name": "社内チャットアプリ",
        "dau": 500,
        "peak_qps": 50,
        "data_growth": "月10GB",
        "budget": "$200/月",
        "team": "エンジニア1名（兼任）",
    },
    {
        "name": "ECサイト（セール時に10倍）",
        "dau": 100_000,
        "peak_qps": 5_000,
        "data_growth": "月100GB",
        "budget": "$5,000/月",
        "team": "エンジニア5名",
        "note": "年2回の大型セールで通常の10倍のトラフィック",
    },
    {
        "name": "動画配信サービス",
        "dau": 10_000_000,
        "peak_qps": 500_000,
        "data_growth": "月50TB",
        "budget": "$500,000/月",
        "team": "エンジニア50名",
        "note": "グローバル展開、4K動画対応",
    },
]
 
# TODO: 各シナリオに対して以下を回答
# 1. 推奨アーキテクチャ（図で表現）
# 2. スケーリング方針（垂直/水平、自動/手動）
# 3. データ層の構成
# 4. コスト配分の概算

# 演習2: Auto Scaling ポリシー設計
 
workload = {
    "service": "ニュースアプリAPI",
    "normal_qps": 2000,
    "peak_qps": 20000,       # 重大ニュース発生時
    "morning_spike": 8000,   # 朝の通勤時間帯
    "evening_spike": 6000,   # 夕方の通勤時間帯
    "latency_slo": "P99 < 200ms",
    "error_rate_slo": "< 0.1%",
    "instance_capacity": 500, # 1インスタンスあたりの処理能力 (req/s)
}
 
# TODO: 以下を設計
# 1. 通常時のスケーリングポリシー（目標追跡型）
# 2. スケジュールベースのスケーリング（朝夕の通勤時間帯）
# 3. 突発的な負荷（重大ニュース）への対策
# 4. クールダウン期間の設定と理由
# 5. 最小/最大インスタンス数の決定根拠

# 演習3: シャーディング設計
 
system_spec = {
    "dau": 10_000_000,
    "messages_per_user_per_day": 50,
    "message_size_bytes": 500,
    "retention_years": 5,
    "read_pattern": "ユーザーの最新メッセージを時系列で表示",
    "write_pattern": "ユーザーが送信、グループメンバーに配信",
    "query_pattern": [
        "特定ユーザーの最新100件取得（最も頻繁）",
        "特定会話の全メッセージ取得",
        "全文検索（低頻度）",
    ],
}
 
# TODO: 以下を設計・回答
# 1. シャーディングキーの選定（なぜそのキーか、代替案との比較）
# 2. シャード数の決定（データ量ベースの計算）
# 3. シャーディング方式の選択（ハッシュ/レンジ/ディレクトリ）
# 4. クロスシャードクエリへの対応策
# 5. リシャーディング計画（将来のデータ増加への備え）

def amdahls_law(parallel_fraction: float, num_processors: int) -> float:
    """
    アムダールの法則でスケーリング効率を計算
 
    parallel_fraction: 並列化可能な部分の割合 (0.0 - 1.0)
    num_processors: プロセッサ数（スケーリング倍率）
    """
    serial_fraction = 1 - parallel_fraction
    speedup = 1 / (serial_fraction + parallel_fraction / num_processors)
    return speedup
 
# 並列化率による理論的スピードアップ上限
print("=== アムダールの法則: 並列化率による上限 ===\n")
print(f"{'並列化率':>8s}  {'2台':>6s}  {'4台':>6s}  {'8台':>6s}  "
      f"{'16台':>6s}  {'64台':>6s}  {'理論上限':>8s}")
print("-" * 60)
 
for pct in [0.5, 0.75, 0.9, 0.95, 0.99]:
    speedups = [amdahls_law(pct, n) for n in [2, 4, 8, 16, 64]]
    theoretical_max = 1 / (1 - pct)
    print(f"  {pct*100:>5.0f}%  " +
          "  ".join(f"{s:>5.1f}x" for s in speedups) +
          f"  {theoretical_max:>6.1f}x")
 
# 出力:
# === アムダールの法則: 並列化率による上限 ===
#
# 並列化率    2台    4台    8台   16台   64台  理論上限
# ------------------------------------------------------------
#    50%   1.3x   1.6x   1.8x   1.9x   2.0x    2.0x
#    75%   1.6x   2.3x   2.9x   3.4x   3.8x    4.0x
#    90%   1.8x   3.1x   4.7x   6.4x   8.8x   10.0x
#    95%   1.9x   3.5x   5.9x   9.1x  14.8x   20.0x
#    99%   2.0x   3.9x   7.5x  13.9x  39.3x  100.0x
 
# 教訓:
# - 並列化率が90%でも、64台に増やしてスピードアップは8.8倍止まり
# - 10%のシリアル部分が全体のボトルネックになる
# - スケーリングの前に、シリアル部分（DB、ロック、I/O）を最小化すべき