Amazon Aurora Mysql

scripts/commitment_pricing_analyzer.py

"""Aurora Reserved Instance & Database Savings Plan estimator.

Read-only tool that fetches live RI and DSP rates from AWS and projects
monthly cost under each commitment option for a cluster, fleet, or
user-supplied workload. No purchase APIs are ever called.

Usage:
    python commitment_pricing_analyzer.py --cluster my-cluster --region us-east-1
    python commitment_pricing_analyzer.py --all --region us-east-1
    python commitment_pricing_analyzer.py offline \\
        --instance db.r7g.2xlarge --num-instances 2 --region us-east-1
    python commitment_pricing_analyzer.py offline \\
        --serverless --avg-acu 8 --region us-east-1
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from dataclasses import dataclass

HOURS_PER_MONTH = 730

# ---------------------------------------------------------------------------
# Static fallback on-demand prices (us-east-1, Aurora MySQL Standard;
# Aurora MySQL and PostgreSQL share these instance rates).
# I/O-Optimized premium applied via multiplier.
# ---------------------------------------------------------------------------
IO_OPT_COMPUTE_MULTIPLIER = 1.30
ACU_PRICE_STANDARD = 0.12  # $/ACU-Hr
ACU_PRICE_IO_OPTIMIZED = 0.156

_STATIC_INSTANCE_PRICES = {
    "db.t3.medium": 0.082,
    "db.t3.large": 0.164,
    "db.t4g.medium": 0.073,
    "db.t4g.large": 0.146,
    "db.r5.large": 0.290,
    "db.r5.xlarge": 0.580,
    "db.r5.2xlarge": 1.160,
    "db.r5.4xlarge": 2.320,
    "db.r5.8xlarge": 4.640,
    "db.r5.12xlarge": 6.960,
    "db.r5.16xlarge": 9.280,
    "db.r5.24xlarge": 13.920,
    "db.r6g.large": 0.260,
    "db.r6g.xlarge": 0.519,
    "db.r6g.2xlarge": 1.038,
    "db.r6g.4xlarge": 2.076,
    "db.r6g.8xlarge": 4.152,
    "db.r6g.12xlarge": 6.228,
    "db.r6g.16xlarge": 8.304,
    "db.r7g.large": 0.276,
    "db.r7g.xlarge": 0.553,
    "db.r7g.2xlarge": 1.106,
    "db.r7g.4xlarge": 2.211,
    "db.r7g.8xlarge": 4.422,
    "db.r7g.12xlarge": 6.633,
    "db.r7g.16xlarge": 8.844,
    "db.r8g.large": 0.276,
    "db.r8g.xlarge": 0.552,
    "db.r8g.2xlarge": 1.104,
    "db.r8g.4xlarge": 2.208,
    "db.r8g.8xlarge": 4.416,
    "db.r8g.12xlarge": 6.624,
    "db.r8g.16xlarge": 8.832,
    "db.r8g.24xlarge": 13.248,
    "db.r8g.48xlarge": 26.496,
}

_REGION_NAMES = {
    "us-east-1": "US East (N. Virginia)",
    "us-east-2": "US East (Ohio)",
    "us-west-1": "US West (N. California)",
    "us-west-2": "US West (Oregon)",
    "eu-west-1": "EU (Ireland)",
    "eu-west-2": "EU (London)",
    "eu-central-1": "EU (Frankfurt)",
    "eu-north-1": "EU (Stockholm)",
    "ap-southeast-1": "Asia Pacific (Singapore)",
    "ap-southeast-2": "Asia Pacific (Sydney)",
    "ap-northeast-1": "Asia Pacific (Tokyo)",
    "ap-south-1": "Asia Pacific (Mumbai)",
    "ca-central-1": "Canada (Central)",
}

# DSP only covers these families
_DSP_ELIGIBLE_FAMILIES = {"r7g", "r7i", "r8g", "r8gd", "m7g", "m7i", "c7g", "c7i", "x8g"}

_DSP_SIZE_MAP = {
    "micro": "micro",
    "small": "small",
    "medium": "medium",
    "large": "large",
    "xl": "xlarge",
    "2xl": "2xlarge",
    "4xl": "4xlarge",
    "8xl": "8xlarge",
    "12xl": "12xlarge",
    "16xl": "16xlarge",
    "24xl": "24xlarge",
    "48xl": "48xlarge",
}


# ---------------------------------------------------------------------------
# Data classes
# ---------------------------------------------------------------------------


@dataclass
class RIOffering:
    instance_type: str
    term_years: int
    payment_option: str  # "No Upfront" | "Partial Upfront" | "All Upfront"
    effective_hourly: float  # (upfront / term_hours) + recurring
    upfront_cost: float
    recurring_hourly: float

    def monthly_cost(self) -> float:
        return self.effective_hourly * HOURS_PER_MONTH


@dataclass
class DSPRate:
    usage_type: str  # instance type or "ServerlessV2"
    term_years: int  # always 1 for Aurora DSP
    payment_option: str
    rate_per_hour: float

    def monthly_cost(self) -> float:
        return self.rate_per_hour * HOURS_PER_MONTH


# ---------------------------------------------------------------------------
# Live AWS fetchers
# ---------------------------------------------------------------------------


def _family_from_instance(instance_type: str) -> str:
    m = re.match(r"db\.([a-z0-9]+)\.", instance_type)
    return m.group(1) if m else ""


def get_on_demand_price(instance_type: str, region: str = "us-east-1") -> float:
    """Return on-demand hourly price. Tries Pricing API, falls back to static."""
    try:
        import boto3
    except ImportError:
        return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)

    location = _REGION_NAMES.get(region)
    if not location:
        return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)

    try:
        pricing = boto3.client("pricing", region_name="us-east-1")
        filters = [
            {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"},
            {"Type": "TERM_MATCH", "Field": "location", "Value": location},
            {"Type": "TERM_MATCH", "Field": "instanceType", "Value": instance_type},
            {"Type": "TERM_MATCH", "Field": "deploymentOption", "Value": "Single-AZ"},
            {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"},
        ]
        for page in pricing.get_paginator("get_products").paginate(
            ServiceCode="AmazonRDS", Filters=filters
        ):
            for item_json in page["PriceList"]:
                item = json.loads(item_json) if isinstance(item_json, str) else item_json
                attrs = item.get("product", {}).get("attributes", {})
                if "IOOptimized" in attrs.get("usagetype", ""):
                    continue
                for term in item.get("terms", {}).get("OnDemand", {}).values():
                    for dim in term.get("priceDimensions", {}).values():
                        try:
                            price = float(dim.get("pricePerUnit", {}).get("USD", "0"))
                        except (ValueError, TypeError):
                            continue
                        if price > 0:
                            return price
    except Exception:
        pass

    return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)


def fetch_ri_offerings(instance_type: str, region: str) -> list[RIOffering]:
    """Fetch all RI offerings for an instance type. Returns [] on failure."""
    try:
        import boto3
    except ImportError:
        return []

    results: list[RIOffering] = []
    try:
        rds = boto3.client("rds", region_name=region)
        for engine in ("aurora-mysql",):
            try:
                paginator = rds.get_paginator("describe_reserved_db_instances_offerings")
                for page in paginator.paginate(
                    DBInstanceClass=instance_type,
                    ProductDescription=engine,
                    MultiAZ=False,
                ):
                    for offering in page.get("ReservedDBInstancesOfferings", []):
                        inst = offering.get("DBInstanceClass", "")
                        if inst != instance_type:
                            continue
                        duration = offering.get("Duration", 0)
                        term_years = 3 if duration > 94_000_000 else 1
                        payment = offering.get("OfferingType", "")
                        fixed = float(offering.get("FixedPrice", 0.0))
                        recurring_list = offering.get("RecurringCharges", [])
                        recurring_hr = sum(
                            float(rc.get("RecurringChargeAmount", 0.0)) for rc in recurring_list
                        )
                        term_hours = term_years * 365 * 24
                        effective = (fixed / term_hours) + recurring_hr
                        results.append(
                            RIOffering(
                                instance_type=inst,
                                term_years=term_years,
                                payment_option=payment,
                                effective_hourly=round(effective, 6),
                                upfront_cost=round(fixed, 2),
                                recurring_hourly=round(recurring_hr, 6),
                            )
                        )
            except Exception:
                continue
    except Exception:
        return []

    # Deduplicate (same offering exists for both engines)
    seen = set()
    deduped = []
    for r in results:
        key = (r.term_years, r.payment_option, round(r.effective_hourly, 6))
        if key in seen:
            continue
        seen.add(key)
        deduped.append(r)
    return deduped


def fetch_dsp_rates(region: str) -> dict[str, list[DSPRate]]:
    """Fetch Database Savings Plan rates for Aurora in the region.

    Returns dict mapping usage key (instance type or 'ServerlessV2') to rates.
    """
    try:
        import boto3
    except ImportError:
        return {}

    result: dict[str, list[DSPRate]] = {}
    try:
        sp = boto3.client("savingsplans", region_name="us-east-1")
    except Exception:
        return {}

    for engine in ("Aurora MySQL",):
        try:
            rates = []
            token = None
            while True:
                kwargs = {
                    "savingsPlanTypes": ["Database"],
                    "products": ["RDS"],
                    "serviceCodes": ["AmazonRDS"],
                    "filters": [
                        {"name": "region", "values": [region]},
                        {"name": "productDescription", "values": [engine]},
                    ],
                    "maxResults": 1000,
                }
                if token:
                    kwargs["nextToken"] = token
                resp = sp.describe_savings_plans_offering_rates(**kwargs)
                rates.extend(resp.get("searchResults", []))
                token = resp.get("nextToken")
                if not token:
                    break

            for rate_entry in rates:
                offering = rate_entry.get("savingsPlanOffering", {})
                dur = offering.get("durationSeconds", 0)
                term_years = 3 if dur > 94_000_000 else 1
                payment = offering.get("paymentOption", "")
                try:
                    rate_val = float(rate_entry.get("rate", "0"))
                except (ValueError, TypeError):
                    continue
                if rate_val <= 0:
                    continue

                usage = rate_entry.get("usageType", "")
                unit = rate_entry.get("unit", "")

                # Skip I/O-Optimized variants for consistency; main pricing uses Standard
                if "IOOptimized" in usage:
                    continue

                if unit == "ACU-Hr" and "ServerlessV2" in usage:
                    key = "ServerlessV2"
                else:
                    m = re.match(r"InstanceUsage:db\.(\w+)\.(\w+)", usage)
                    if not m:
                        continue
                    family = m.group(1)
                    short_size = m.group(2)
                    size = _DSP_SIZE_MAP.get(short_size, short_size)
                    key = f"db.{family}.{size}"

                entry = DSPRate(
                    usage_type=key,
                    term_years=term_years,
                    payment_option=payment,
                    rate_per_hour=round(rate_val, 6),
                )
                existing = result.get(key, [])
                if not any(
                    e.term_years == entry.term_years and e.payment_option == entry.payment_option
                    for e in existing
                ):
                    result.setdefault(key, []).append(entry)
        except Exception:
            continue

    return result


# ---------------------------------------------------------------------------
# Best-of selection (lowest effective monthly cost per term/category)
# ---------------------------------------------------------------------------


def best_ri(offerings: list[RIOffering], term_years: int) -> RIOffering | None:
    candidates = [r for r in offerings if r.term_years == term_years]
    if not candidates:
        return None
    return min(candidates, key=lambda r: r.effective_hourly)


def best_dsp(rates: list[DSPRate], term_years: int = 1) -> DSPRate | None:
    candidates = [r for r in rates if r.term_years == term_years]
    if not candidates:
        return None
    return min(candidates, key=lambda r: r.rate_per_hour)


# ---------------------------------------------------------------------------
# Comparison builder — single workload
# ---------------------------------------------------------------------------


def build_comparison(
    instance_type: str,
    num_instances: int,
    region: str,
    io_optimized: bool = False,
    is_serverless: bool = False,
    avg_acu: float = 0.0,
    dsp_rates: dict[str, list[DSPRate]] | None = None,
) -> dict:
    """Compare on-demand, RI, and DSP for a single workload description."""
    if dsp_rates is None:
        dsp_rates = fetch_dsp_rates(region)

    if is_serverless:
        od_hourly = ACU_PRICE_IO_OPTIMIZED if io_optimized else ACU_PRICE_STANDARD
        # For Serverless v2, "number of instances" is irrelevant — we price avg ACU continuously
        units = avg_acu
        od_monthly = od_hourly * units * HOURS_PER_MONTH

        dsp_entry = best_dsp(dsp_rates.get("ServerlessV2", []))
        dsp_monthly = dsp_entry.rate_per_hour * units * HOURS_PER_MONTH if dsp_entry else None
        dsp_savings = (od_monthly - dsp_monthly) if dsp_monthly is not None else None

        return {
            "workload_type": "serverless_v2",
            "avg_acu": avg_acu,
            "io_optimized": io_optimized,
            "on_demand": {
                "hourly": od_hourly,
                "monthly": round(od_monthly, 2),
            },
            "ri_1yr": None,
            "ri_3yr": None,
            "dsp_1yr": _format_dsp(dsp_entry, units, od_monthly) if dsp_entry else None,
            "recommendation": _recommend_serverless(dsp_savings, od_monthly),
            "notes": [
                "Reserved Instances do not apply to Aurora Serverless v2.",
                "DSP covers ACU-hours but bills the committed $/hr continuously, "
                "even during auto-pause. Only commit to the steady baseline ACU.",
            ],
        }

    # Provisioned
    family = _family_from_instance(instance_type)
    od_hourly = get_on_demand_price(instance_type, region)
    if io_optimized:
        od_hourly *= IO_OPT_COMPUTE_MULTIPLIER
    od_monthly = od_hourly * HOURS_PER_MONTH * num_instances

    ri_offerings = fetch_ri_offerings(instance_type, region)
    ri_1yr = best_ri(ri_offerings, 1)
    ri_3yr = best_ri(ri_offerings, 3)

    # I/O-Optimized RI coverage (AWS Compute Optimizer, verified): an I/O-Optimized
    # instance is FULLY covered by Reserved Instances — no portion is forced to
    # on-demand — but it "consumes 30% more normalized units per hour than Aurora
    # Standard", i.e. it draws down RI capacity at 1.30×. So the effective RI cost is
    # the (Standard-normalized) RI rate × 1.30. Equivalently: buy ~30% more normalized
    # RI units to cover the same I/O-Optimized fleet.
    #   io-opt RI monthly = ri_rate × 1.30 × hours × N
    def ri_adjusted_monthly(ri: RIOffering | None) -> float | None:
        if ri is None:
            return None
        units = IO_OPT_COMPUTE_MULTIPLIER if io_optimized else 1.0
        return ri.effective_hourly * units * HOURS_PER_MONTH * num_instances

    ri_1yr_monthly = ri_adjusted_monthly(ri_1yr)
    ri_3yr_monthly = ri_adjusted_monthly(ri_3yr)

    dsp_entry = best_dsp(dsp_rates.get(instance_type, []))
    dsp_monthly = dsp_entry.rate_per_hour * HOURS_PER_MONTH * num_instances if dsp_entry else None

    dsp_eligible = family in _DSP_ELIGIBLE_FAMILIES
    notes = []
    if not dsp_eligible:
        notes.append(
            f"Database Savings Plans do not cover the {family} family. "
            f"DSP requires r7g, r8g, r7i, or newer-gen Aurora-compatible families."
        )
    if io_optimized:
        notes.append(
            "Cluster is I/O-Optimized (30% compute premium). Both RI and DSP cover the "
            "full I/O-Optimized instance price — I/O-Optimized consumes 30% more "
            "normalized units per hour, so an RI draws down capacity at 1.30× (buy ~30% "
            "more normalized RI units to fully cover the fleet); no portion is on-demand."
        )

    return {
        "workload_type": "provisioned",
        "instance_type": instance_type,
        "num_instances": num_instances,
        "io_optimized": io_optimized,
        "on_demand": {
            "hourly": round(od_hourly, 4),
            "monthly": round(od_monthly, 2),
        },
        "ri_1yr": _format_ri(ri_1yr, ri_1yr_monthly, od_monthly, num_instances),
        "ri_3yr": _format_ri(ri_3yr, ri_3yr_monthly, od_monthly, num_instances),
        "dsp_1yr": (_format_dsp(dsp_entry, num_instances, od_monthly) if dsp_entry else None),
        "recommendation": _recommend_provisioned(
            od_monthly,
            ri_1yr_monthly,
            ri_3yr_monthly,
            dsp_monthly,
            dsp_eligible=dsp_eligible,
            io_optimized=io_optimized,
        ),
        "notes": notes,
    }


def _format_ri(
    ri: RIOffering | None, monthly: float | None, od_monthly: float, n: int
) -> dict | None:
    if ri is None or monthly is None:
        return None
    savings = od_monthly - monthly
    pct = (savings / od_monthly * 100) if od_monthly > 0 else 0
    return {
        "term_years": ri.term_years,
        "payment_option": ri.payment_option,
        "effective_hourly_per_instance": round(ri.effective_hourly, 4),
        "upfront_total": round(ri.upfront_cost * n, 2),
        "monthly": round(monthly, 2),
        "savings_monthly": round(savings, 2),
        "savings_pct": round(pct, 1),
    }


def _format_dsp(dsp: DSPRate | None, units: float, od_monthly: float) -> dict | None:
    if dsp is None:
        return None
    monthly = dsp.rate_per_hour * HOURS_PER_MONTH * units
    savings = od_monthly - monthly
    pct = (savings / od_monthly * 100) if od_monthly > 0 else 0
    return {
        "term_years": dsp.term_years,
        "payment_option": dsp.payment_option,
        "rate_per_hour": round(dsp.rate_per_hour, 4),
        "monthly": round(monthly, 2),
        "savings_monthly": round(savings, 2),
        "savings_pct": round(pct, 1),
    }


def _recommend_provisioned(
    od: float,
    ri_1yr: float | None,
    ri_3yr: float | None,
    dsp: float | None,
    dsp_eligible: bool,
    io_optimized: bool,
) -> dict:
    options = []
    if ri_1yr is not None:
        options.append(("1yr RI", ri_1yr))
    if ri_3yr is not None:
        options.append(("3yr RI", ri_3yr))
    if dsp is not None:
        options.append(("1yr DSP", dsp))

    if not options:
        return {
            "best_option": "on_demand",
            "reason": "No RI or DSP offerings available for this instance type/region.",
        }

    best_label, best_cost = min(options, key=lambda x: x[1])
    savings = od - best_cost
    pct = (savings / od * 100) if od > 0 else 0

    reasons = [
        f"{best_label} is the lowest-cost option, saving ${savings:.0f}/mo ({pct:.0f}%) vs on-demand."
    ]
    if best_label == "3yr RI":
        reasons.append(
            "Best fit for steady 24/7 workloads you're confident will stay on this instance family for 3 years."
        )
    elif best_label == "1yr DSP":
        reasons.append(
            "Offers flexibility — covers any eligible Aurora instance family in the account, including future upgrades."
        )
        if io_optimized:
            reasons.append(
                "DSP is particularly attractive for I/O-Optimized clusters since it covers the full rate."
            )
    elif best_label == "1yr RI":
        reasons.append(
            "Shorter commitment than 3yr, useful when instance-family migration is on the horizon."
        )

    if not dsp_eligible and dsp is None:
        reasons.append(
            "DSP is not available for this instance family, so RI is the only commitment option."
        )

    return {
        "best_option": best_label,
        "best_monthly_cost": round(best_cost, 2),
        "savings_vs_on_demand": round(savings, 2),
        "savings_pct": round(pct, 1),
        "reason": " ".join(reasons),
    }


def _recommend_serverless(dsp_savings: float | None, od_monthly: float) -> dict:
    if dsp_savings is None:
        return {
            "best_option": "on_demand",
            "reason": "No Database Savings Plan rates available for Serverless v2 ACU in this region.",
        }
    if dsp_savings <= 0:
        return {
            "best_option": "on_demand",
            "reason": "DSP would not save money at the specified average ACU.",
        }
    pct = (dsp_savings / od_monthly * 100) if od_monthly > 0 else 0
    return {
        "best_option": "1yr DSP",
        "savings_vs_on_demand": round(dsp_savings, 2),
        "savings_pct": round(pct, 1),
        "reason": (
            f"1yr DSP saves ${dsp_savings:.0f}/mo ({pct:.0f}%). "
            "Size the commitment to your steady baseline ACU — DSP bills the committed $/hr "
            "continuously, even during idle periods."
        ),
    }


# ---------------------------------------------------------------------------
# Live cluster analysis
# ---------------------------------------------------------------------------


def _is_empty_cluster(cluster: dict) -> bool:
    """Skip clusters with no compute to analyze.

    An Aurora cluster with no DB instances has no compute cost, so RI and DSP
    commitment analysis doesn't apply. Typical cases: the last writer/reader
    instance was deleted, paused/stopped clusters, or clusters mid-migration.
    """
    return len(cluster.get("DBClusterMembers", [])) == 0


def analyze_cluster_live(cluster_id: str, region: str) -> dict:
    import boto3

    rds = boto3.client("rds", region_name=region)

    try:
        resp = rds.describe_db_clusters(DBClusterIdentifier=cluster_id)
    except Exception as e:
        return {"cluster_id": cluster_id, "error": str(e)}
    clusters = resp.get("DBClusters", [])
    if not clusters:
        return {"cluster_id": cluster_id, "error": "cluster not found"}
    cluster = clusters[0]
    storage_type = cluster.get("StorageType", "aurora")
    io_optimized = storage_type == "aurora-iopt1"
    engine = cluster.get("Engine", "")

    # Guardrail: skip clusters with no DB instances (last instance deleted, paused, mid-migration, etc.)
    if _is_empty_cluster(cluster):
        return {
            "cluster_id": cluster_id,
            "engine": engine,
            "engine_version": cluster.get("EngineVersion", ""),
            "storage_type": storage_type,
            "skipped": True,
            "reason": (
                "Cluster has no DB instances — no compute to price. "
                "RI and DSP commitment analysis does not apply. "
                "This typically indicates the last instance was deleted, a paused "
                "cluster, or a cluster mid-migration."
            ),
        }

    # Identify instances
    member_ids = [m["DBInstanceIdentifier"] for m in cluster.get("DBClusterMembers", [])]
    type_counts: dict[str, int] = {}
    serverless_instances = 0
    for mid in member_ids:
        try:
            iresp = rds.describe_db_instances(DBInstanceIdentifier=mid)
            for inst in iresp.get("DBInstances", []):
                itype = inst.get("DBInstanceClass", "")
                if itype == "db.serverless":
                    serverless_instances += 1
                else:
                    type_counts[itype] = type_counts.get(itype, 0) + 1
        except Exception:
            continue

    # Prefetch DSP rates once for this cluster analysis
    dsp_rates = fetch_dsp_rates(region)

    sub_workloads = []
    for itype, count in type_counts.items():
        sub_workloads.append(
            build_comparison(
                instance_type=itype,
                num_instances=count,
                region=region,
                io_optimized=io_optimized,
                dsp_rates=dsp_rates,
            )
        )
    if serverless_instances > 0:
        # Without observed ACU metrics, we can't price serverless exactly — note it
        sub_workloads.append(
            {
                "workload_type": "serverless_v2",
                "instance_count": serverless_instances,
                "note": "Serverless v2 instances detected. Re-run with 'offline --serverless "
                "--avg-acu <N>' using your observed average ACU (from CloudWatch "
                "ServerlessDatabaseCapacity metric) for a precise DSP estimate.",
                "io_optimized": io_optimized,
            }
        )

    return {
        "cluster_id": cluster_id,
        "engine": engine,
        "storage_type": storage_type,
        "io_optimized": io_optimized,
        "instance_mix": {**type_counts, "serverless": serverless_instances},
        "workloads": sub_workloads,
    }


def list_clusters(region: str) -> list[str]:
    import boto3

    rds = boto3.client("rds", region_name=region)
    names = []
    for page in rds.get_paginator("describe_db_clusters").paginate():
        for c in page.get("DBClusters", []):
            if c.get("Engine", "").startswith("aurora"):
                names.append(c["DBClusterIdentifier"])
    return names


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------


def _format_table_single(result: dict) -> str:
    lines = []
    lines.append("=" * 72)
    if result.get("workload_type") == "serverless_v2":
        lines.append(f"Aurora Serverless v2 Commitment Pricing")
        lines.append(
            f"  Avg ACU: {result.get('avg_acu', 0):.1f}  "
            f"I/O-Optimized: {result.get('io_optimized', False)}"
        )
    else:
        lines.append(
            f"Aurora Commitment Pricing — "
            f"{result.get('num_instances', 1)}× {result.get('instance_type', '?')}"
        )
        if result.get("io_optimized"):
            lines.append("  Storage: I/O-Optimized (30% compute premium applied)")
    lines.append("=" * 72)
    od_monthly = result["on_demand"]["monthly"]
    lines.append("")
    lines.append(f"  {'Option':<28} {'Monthly':>12} {'Savings':>12} {'Upfront':>12}  Term")
    lines.append("  " + "-" * 70)
    lines.append(f"  {'On-Demand':<28} ${od_monthly:>11,.0f} {'—':>12} {'$0':>12}  —")

    for key, label, term_hint in (
        ("ri_1yr", "1yr RI", "1 year"),
        ("ri_3yr", "3yr RI", "3 years"),
        ("dsp_1yr", "1yr DSP", "1 year"),
    ):
        entry = result.get(key)
        if not entry:
            continue
        payment = entry.get("payment_option", "")
        display = f"{label} ({payment})" if payment else label
        monthly = entry.get("monthly", 0)
        savings = entry.get("savings_monthly", 0)
        pct = entry.get("savings_pct", 0)
        upfront = entry.get("upfront_total", 0)
        savings_str = f"${savings:,.0f} ({pct:.0f}%)" if savings else "—"
        upfront_str = f"${upfront:,.0f}" if upfront else "$0"
        lines.append(
            f"  {display:<28} ${monthly:>11,.0f} {savings_str:>12} {upfront_str:>12}  {term_hint}"
        )

    rec = result.get("recommendation", {})
    lines.append("")
    lines.append(f"  Recommendation: {rec.get('best_option', '?')}")
    lines.append(f"  {rec.get('reason', '')}")

    notes = result.get("notes", [])
    if notes:
        lines.append("")
        for n in notes:
            lines.append(f"  Note: {n}")
    lines.append("=" * 72)
    return "\n".join(lines)


def _format_cluster(result: dict) -> str:
    lines = []
    lines.append(
        f"Cluster: {result.get('cluster_id', '?')}  "
        f"({result.get('engine', '?')})  "
        f"storage_type={result.get('storage_type', '?')}"
    )
    workloads = result.get("workloads", [])
    for wl in workloads:
        if wl.get("workload_type") == "serverless_v2" and "note" in wl:
            lines.append("")
            lines.append(f"  [Serverless v2 — {wl.get('instance_count', 0)} instance(s)]")
            lines.append(f"  {wl['note']}")
            continue
        lines.append("")
        lines.append(_format_table_single(wl))
    return "\n".join(lines)


def _format_fleet(output: dict) -> str:
    lines = []
    lines.append(f"Region: {output['region']}")
    lines.append("")
    total_od = 0.0
    total_best = 0.0
    for cluster in output["clusters"]:
        if "error" in cluster:
            lines.append(f"{cluster['cluster_id']}: ERROR {cluster['error']}")
            continue
        for wl in cluster.get("workloads", []):
            if wl.get("workload_type") != "provisioned":
                continue
            od = wl["on_demand"]["monthly"]
            best = wl.get("recommendation", {}).get("best_monthly_cost", od)
            total_od += od
            total_best += best
    lines.append(f"  Fleet monthly on-demand:    ${total_od:,.0f}")
    lines.append(f"  With best commitments:      ${total_best:,.0f}")
    savings = total_od - total_best
    pct = (savings / total_od * 100) if total_od > 0 else 0
    lines.append(f"  Fleet savings opportunity:  ${savings:,.0f}/mo ({pct:.0f}%)")
    lines.append("")
    for cluster in output["clusters"]:
        if "error" in cluster:
            continue
        lines.append("")
        lines.append(_format_cluster(cluster))
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def main():
    parser = argparse.ArgumentParser(
        description="Aurora RI & Database Savings Plan estimator (read-only)"
    )
    parser.add_argument("--region", default="us-east-1")
    parser.add_argument("--format", choices=["json", "table"], default="json")
    parser.add_argument("--cluster", help="Analyze a single cluster by identifier")
    parser.add_argument(
        "--all", action="store_true", help="Analyze all Aurora clusters in the region"
    )

    sub = parser.add_subparsers(dest="mode")
    off = sub.add_parser("offline", help="Use user-supplied workload description")
    off.add_argument("--instance", help="Instance type (e.g., db.r7g.2xlarge)")
    off.add_argument("--num-instances", type=int, default=1)
    off.add_argument(
        "--io-optimized", action="store_true", help="Workload uses Aurora I/O-Optimized storage"
    )
    off.add_argument(
        "--serverless", action="store_true", help="Serverless v2 workload — requires --avg-acu"
    )
    off.add_argument(
        "--avg-acu", type=float, default=0.0, help="Average ACU for serverless workload"
    )
    off.add_argument("--region", default="us-east-1")
    off.add_argument("--format", choices=["json", "table"], default="json")

    args = parser.parse_args()

    if args.mode == "offline":
        if args.serverless:
            if args.avg_acu <= 0:
                print("ERROR: --serverless requires --avg-acu > 0", file=sys.stderr)
                sys.exit(2)
            result = build_comparison(
                instance_type="",
                num_instances=0,
                region=args.region,
                io_optimized=args.io_optimized,
                is_serverless=True,
                avg_acu=args.avg_acu,
            )
        else:
            if not args.instance:
                print("ERROR: offline mode requires --instance (or --serverless)", file=sys.stderr)
                sys.exit(2)
            result = build_comparison(
                instance_type=args.instance,
                num_instances=args.num_instances,
                region=args.region,
                io_optimized=args.io_optimized,
            )
        if args.format == "json":
            print(json.dumps(result, indent=2, default=str))
        else:
            print(_format_table_single(result))
        return

    # Live modes require boto3
    try:
        import boto3  # noqa: F401
    except ImportError:
        print(
            "ERROR: boto3 required for live AWS analysis. Use the 'offline' subcommand.",
            file=sys.stderr,
        )
        sys.exit(2)

    if args.all:
        cluster_ids = list_clusters(args.region)
        results = [analyze_cluster_live(cid, args.region) for cid in cluster_ids]
        output = {"region": args.region, "cluster_count": len(results), "clusters": results}
        if args.format == "json":
            print(json.dumps(output, indent=2, default=str))
        else:
            print(_format_fleet(output))
        return

    if args.cluster:
        result = analyze_cluster_live(args.cluster, args.region)
        if args.format == "json":
            print(json.dumps(result, indent=2, default=str))
        else:
            print(_format_cluster(result))
        return

    parser.print_help()


if __name__ == "__main__":
    main()