Capacity Planning skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Capacity Planning"
  short_description: "Know which resource runs out first, and when"
  default_prompt: "$capacity-planning We expect 2 million daily users by next summer, how many API pods and how big a database?"

#!/usr/bin/env python3
"""Back-of-envelope capacity estimate: peak load, instances, concurrency, storage, egress.

Standard library only. Every input is explicit so the assumptions show in the output.

Examples:
  python3 capacity.py --dau 2000000 --requests-per-user 40 --peak-factor 3 \
      --per-instance-rps 450 --latency-ms 80 --zones 3 --survive-zone-loss \
      --monthly-growth 0.05 --months 12 \
      --writes-per-user 5 --bytes-per-write 1200 --retention-days 365 --replication 3 \
      --response-kb 8
  python3 capacity.py --peak-rps 1200 --per-instance-rps 300
"""
from __future__ import annotations

import argparse
import math
import sys

SECONDS_PER_DAY = 86_400


def human_bytes(n: float) -> str:
    for unit in ("B", "KB", "MB", "GB", "TB", "PB"):
        if abs(n) < 1000:
            return f"{n:,.1f} {unit}"
        n /= 1000
    return f"{n:,.1f} EB"


def parse() -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    load = p.add_argument_group("load (give --peak-rps, or --dau with --requests-per-user)")
    load.add_argument("--peak-rps", type=float, help="measured or forecast peak requests/second today")
    load.add_argument("--dau", type=float, help="daily active users today")
    load.add_argument("--requests-per-user", type=float, help="requests per active user per day")
    load.add_argument("--peak-factor", type=float, default=3.0, help="peak / average ratio (default 3)")
    p.add_argument("--monthly-growth", type=float, default=0.0, help="compound growth per month, e.g. 0.05")
    p.add_argument("--months", type=int, default=0, help="planning horizon in months")
    p.add_argument("--per-instance-rps", type=float, required=True,
                   help="max sustainable rps of one instance at SLO latency (from a load test)")
    p.add_argument("--target-util", type=float, default=0.6, help="planned utilisation at peak (default 0.6)")
    p.add_argument("--latency-ms", type=float, help="mean time in system per request, for Little's law")
    p.add_argument("--zones", type=int, default=1, help="availability zones the fleet spreads across")
    p.add_argument("--survive-zone-loss", action="store_true", help="size so peak fits after losing one zone")
    p.add_argument("--writes-per-user", type=float, help="stored records per active user per day")
    p.add_argument("--bytes-per-write", type=float, help="bytes per stored record incl. indexes/overhead")
    p.add_argument("--retention-days", type=float, default=365)
    p.add_argument("--replication", type=float, default=1.0, help="copies kept (replicas, backups)")
    p.add_argument("--response-kb", type=float, help="mean response size in KB, for egress")
    a = p.parse_args()

    if a.peak_rps is None and (a.dau is None or a.requests_per_user is None):
        p.error("give --peak-rps, or both --dau and --requests-per-user")
    for name in ("per_instance_rps", "peak_factor"):
        if getattr(a, name) <= 0:
            p.error(f"--{name.replace('_', '-')} must be > 0")
    if not 0 < a.target_util <= 1:
        p.error("--target-util must be in (0, 1]")
    if a.zones < 1 or (a.survive_zone_loss and a.zones < 2):
        p.error("--survive-zone-loss needs --zones >= 2")
    if a.monthly_growth < 0 or a.months < 0:
        p.error("growth and months must be >= 0")
    return a


def main() -> int:
    a = parse()
    growth = (1 + a.monthly_growth) ** a.months
    rows: list[tuple[str, str]] = []

    if a.peak_rps is not None:
        peak_now = a.peak_rps
        rows.append(("Peak rps today (given)", f"{peak_now:,.0f}"))
    else:
        avg_now = a.dau * a.requests_per_user / SECONDS_PER_DAY
        peak_now = avg_now * a.peak_factor
        rows.append(("Average rps today", f"{avg_now:,.1f}"))
        rows.append((f"Peak rps today (x{a.peak_factor:g})", f"{peak_now:,.0f}"))

    peak = peak_now * growth
    rows.append((f"Peak rps in {a.months} months (x{growth:.2f})", f"{peak:,.0f}"))

    usable = a.per_instance_rps * a.target_util
    needed = max(1, math.ceil(peak / usable))
    rows.append((f"Usable rps per instance ({a.target_util:.0%} of {a.per_instance_rps:g})", f"{usable:,.0f}"))
    rows.append(("Instances to serve peak", str(needed)))

    total = needed
    if a.zones > 1:
        if a.survive_zone_loss:
            total = math.ceil(needed * a.zones / (a.zones - 1))
        total = math.ceil(total / a.zones) * a.zones  # even spread
        label = f"Instances across {a.zones} zones" + (" (survives 1 zone loss)" if a.survive_zone_loss else "")
        rows.append((label, f"{total} ({total // a.zones} per zone)"))
    rows.append(("Utilisation at peak, all healthy", f"{peak / (total * a.per_instance_rps):.0%}"))

    if a.latency_ms:
        conc = peak * a.latency_ms / 1000
        rows.append(("In-flight requests at peak (Little's law)", f"{conc:,.0f}"))
        rows.append(("In-flight per instance", f"{conc / total:,.1f}"))

    if a.dau and a.writes_per_user and a.bytes_per_write:
        daily = a.dau * growth * a.writes_per_user * a.bytes_per_write
        stored = daily * a.retention_days * a.replication
        rows.append(("New data per day (end of horizon)", human_bytes(daily)))
        rows.append((f"Stored at steady state ({a.retention_days:g} d x{a.replication:g} copies)", human_bytes(stored)))
        rows.append(("Write rps from stored records (avg)", f"{a.dau * growth * a.writes_per_user / SECONDS_PER_DAY:,.1f}"))

    if a.response_kb:
        mbps = peak * a.response_kb * 8 / 1000
        rows.append(("Peak egress", f"{mbps:,.0f} Mbit/s"))

    width = max(len(k) for k, _ in rows)
    print("Capacity estimate (assumptions are the inputs; validate per-instance rps by load test)")
    print("-" * (width + 22))
    for k, v in rows:
        print(f"{k:<{width}}  {v}")
    if peak / (total * a.per_instance_rps) > 0.8:
        print("\nWARNING: utilisation above 80% at peak; queueing delay will dominate latency.")
        return 2
    return 0


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