Capacity Planning skill for Claude Code

#!/usr/bin/env python3
"""Project traffic, instances, connections, workers, storage and cost from a JSON model.

Usage:
    python3 scripts/capacity_model.py model.json [--example]

--example prints a complete sample model to start from. Every section except
"traffic" and "service" is optional.

Model fields (units in names):
  traffic:  dau, requests_per_user_per_day, peak_to_average, monthly_growth_pct
            (or avg_rps instead of dau and requests_per_user_per_day)
  service:  rps_per_instance_at_slo (measured by load test at the latency SLO),
            target_utilization (0-1), min_instances, zones, survive_zone_loss,
            instance_cost_month
  database: max_connections, reserved_connections, pool_per_instance,
            write_ratio, max_writes_per_sec (measured), max_write_utilization
  queue:    jobs_per_request, seconds_per_job, worker_utilization, worker_cost_month
  storage:  current_gb, rows_per_day_now, bytes_per_row, index_overhead,
            replication_factor, retention_days, volume_gb
  horizons_months: e.g. [0, 6, 12, 24]

Exit codes: 0 = all horizons fit, 1 = at least one limit exceeded, 2 = bad input.
"""
import json
import math
import sys

EXAMPLE = {
    "traffic": {"dau": 1_200_000, "requests_per_user_per_day": 60, "peak_to_average": 3.0, "monthly_growth_pct": 5},
    "service": {"rps_per_instance_at_slo": 400, "target_utilization": 0.6, "min_instances": 3,
                "zones": 3, "survive_zone_loss": True, "instance_cost_month": 150},
    "database": {"max_connections": 800, "reserved_connections": 50, "pool_per_instance": 10,
                 "write_ratio": 0.12, "max_writes_per_sec": 6000, "max_write_utilization": 0.7},
    "queue": {"jobs_per_request": 0.05, "seconds_per_job": 0.5, "worker_utilization": 0.7, "worker_cost_month": 60},
    "storage": {"current_gb": 900, "rows_per_day_now": 4_000_000, "bytes_per_row": 700, "index_overhead": 0.6,
                "replication_factor": 2, "retention_days": 730, "volume_gb": 4000},
    "horizons_months": [0, 6, 12, 24],
}


def need(section, key, kind=float):
    if key not in section:
        raise KeyError(key)
    return kind(section[key])


def project(model, months):
    t, s = model["traffic"], model["service"]
    growth = (1 + float(t.get("monthly_growth_pct", 0)) / 100) ** months
    if "avg_rps" in t:
        avg_rps = float(t["avg_rps"]) * growth
    else:
        avg_rps = need(t, "dau") * need(t, "requests_per_user_per_day") / 86_400 * growth
    peak_rps = avg_rps * need(t, "peak_to_average")

    per_instance = need(s, "rps_per_instance_at_slo") * need(s, "target_utilization")
    base = math.ceil(peak_rps / per_instance)
    zones = int(s.get("zones", 1))
    if s.get("survive_zone_loss") and zones > 1:
        base = math.ceil(base * zones / (zones - 1))
        base = math.ceil(base / zones) * zones           # spread evenly across zones
    instances = max(base, int(s.get("min_instances", 1)))
    row = {"months": months, "avg_rps": avg_rps, "peak_rps": peak_rps, "instances": instances,
           "cost": instances * float(s.get("instance_cost_month", 0)), "issues": []}

    db = model.get("database")
    if db:
        conns = instances * int(db.get("pool_per_instance", 10))
        limit = int(db.get("max_connections", 100)) - int(db.get("reserved_connections", 0))
        row["db_conns"] = conns
        if conns > limit:
            row["issues"].append(f"DB connections {conns} > usable {limit}: add a pooler (PgBouncer) or shrink pools")
        if "max_writes_per_sec" in db:
            writes = peak_rps * float(db.get("write_ratio", 0))
            cap = float(db["max_writes_per_sec"]) * float(db.get("max_write_utilization", 0.7))
            row["db_writes"] = writes
            if writes > cap:
                row["issues"].append(f"peak writes {writes:,.0f}/s > {cap:,.0f}/s allowed on primary")

    q = model.get("queue")
    if q:
        busy = peak_rps * float(q["jobs_per_request"]) * float(q["seconds_per_job"])
        workers = max(1, math.ceil(busy / float(q.get("worker_utilization", 0.7))))
        row["workers"] = workers
        row["cost"] += workers * float(q.get("worker_cost_month", 0))

    st = model.get("storage")
    if st:
        factor = (1 + float(st.get("index_overhead", 0))) * float(st.get("replication_factor", 1))
        rate = (1 + float(t.get("monthly_growth_pct", 0)) / 100)
        gb = float(st.get("current_gb", 0))
        for m in range(months):                          # accumulate month by month at that month's volume
            gb += st["rows_per_day_now"] * rate ** m * 30 * st["bytes_per_row"] * factor / 1e9
        if "retention_days" in st:
            steady = st["rows_per_day_now"] * rate ** months * st["retention_days"] * st["bytes_per_row"] * factor / 1e9
            gb = min(gb, max(steady, float(st.get("current_gb", 0))))
        row["storage_gb"] = gb
        vol = st.get("volume_gb")
        if vol and gb > 0.8 * float(vol):
            row["issues"].append(f"storage {gb:,.0f} GB > 80% of {float(vol):,.0f} GB volume")
    return row


def main(argv):
    if "--example" in argv:
        print(json.dumps(EXAMPLE, indent=2))
        return 0
    if len(argv) < 2:
        print(__doc__)
        return 2
    try:
        with open(argv[1], encoding="utf-8") as fh:
            model = json.load(fh)
        horizons = model.get("horizons_months", [0, 6, 12])
        rows = [project(model, int(h)) for h in horizons]
    except (OSError, json.JSONDecodeError) as exc:
        print(f"error: cannot read model: {exc}", file=sys.stderr)
        return 2
    except KeyError as exc:
        print(f"error: model is missing required field {exc}", file=sys.stderr)
        return 2

    cols = ["months", "avg_rps", "peak_rps", "instances", "db_conns", "db_writes", "workers", "storage_gb", "cost"]
    cols = [c for c in cols if any(c in r for r in rows)]
    print("  ".join(f"{c:>11}" for c in cols))
    for r in rows:
        cells = []
        for c in cols:
            v = r.get(c, "")
            cells.append(f"{v:>11,.0f}" if isinstance(v, (int, float)) else f"{v:>11}")
        print("  ".join(cells))

    s = model["service"]
    print(f"\nassumes {float(s['rps_per_instance_at_slo']):,.0f} rps/instance at SLO, "
          f"{float(s['target_utilization']):.0%} target utilization"
          + (", N+1 zone headroom" if s.get("survive_zone_loss") else "")
          + "; cost column is USD/month for app instances plus workers")

    failed = False
    for r in rows:
        for issue in r["issues"]:
            failed = True
            print(f"LIMIT  month {r['months']}: {issue}")
    if failed:
        print("\nAt least one horizon exceeds a limit; plan the change before that month (minus lead time).")
        return 1
    print("\nOK: all horizons fit within the stated limits.")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))