"""Aurora Serverless v2 ACU Calculator. Estimates ACU sizing, costs, and generates provisioned-vs-serverless comparisons. Pricing & instance data: pulls live from AWS Pricing API + EC2/RDS APIs when boto3 credentials are available, falls back to static defaults (us-east-1, static fallback) when offline or credentials are missing. Usage: python acu_calculator.py --help python acu_calculator.py estimate --instance db.r6g.xlarge --cpu-p95 35 --cpu-max 72 --storage 500 python acu_calculator.py estimate --region eu-west-1 --instance db.r6g.2xlarge --cpu-p95 20 --cpu-max 55 --connections 200 --storage 1000 --working-set 12 """ import argparse import json import math import re import sys from typing import Any # --------------------------------------------------------------------------- # Static fallback data (us-east-1) # Used when AWS APIs are unavailable (no credentials, offline, API errors). # --------------------------------------------------------------------------- _STATIC_ACU_PRICE_STANDARD = 0.12 # $/ACU-Hr _STATIC_ACU_PRICE_IO_OPTIMIZED = 0.156 # $/ACU-Hr (30% premium) _STATIC_STORAGE_STANDARD_PER_GIB = 0.10 # $/GiB-month _STATIC_STORAGE_IO_OPT_PER_GIB = 0.225 # $/GiB-month _STATIC_LAST_UPDATED = "2026-03-28" _STATIC_REGION = "us-east-1" # Static instance specs: {name: (vcpus, memory_gib, price_per_hour)} _STATIC_INSTANCE_SPECS = { "db.t3.medium": (2, 4, 0.082), "db.t3.large": (2, 8, 0.164), "db.t4g.medium": (2, 4, 0.073), "db.t4g.large": (2, 8, 0.146), "db.r5.large": (2, 16, 0.290), "db.r5.xlarge": (4, 32, 0.580), "db.r5.2xlarge": (8, 64, 1.160), "db.r5.4xlarge": (16, 128, 2.320), "db.r5.8xlarge": (32, 256, 4.640), "db.r5.12xlarge": (48, 384, 6.960), "db.r5.16xlarge": (64, 512, 9.280), "db.r5.24xlarge": (96, 768, 13.920), "db.r6g.large": (2, 16, 0.260), "db.r6g.xlarge": (4, 32, 0.519), "db.r6g.2xlarge": (8, 64, 1.038), "db.r6g.4xlarge": (16, 128, 2.076), "db.r6g.8xlarge": (32, 256, 4.152), "db.r6g.12xlarge": (48, 384, 6.228), "db.r6g.16xlarge": (64, 512, 8.304), "db.r7g.large": (2, 16, 0.276), "db.r7g.xlarge": (4, 32, 0.553), "db.r7g.2xlarge": (8, 64, 1.106), "db.r7g.4xlarge": (16, 128, 2.211), "db.r7g.8xlarge": (32, 256, 4.422), "db.r7g.12xlarge": (48, 384, 6.633), "db.r7g.16xlarge": (64, 512, 8.844), "db.r8g.large": (2, 16, 0.276), "db.r8g.xlarge": (4, 32, 0.552), "db.r8g.2xlarge": (8, 64, 1.104), "db.r8g.4xlarge": (16, 128, 2.208), "db.r8g.8xlarge": (32, 256, 4.416), "db.r8g.12xlarge": (48, 384, 6.624), "db.r8g.16xlarge": (64, 512, 8.832), "db.r8g.24xlarge": (96, 768, 13.248), "db.r8g.48xlarge": (192, 1536, 26.496), } # --------------------------------------------------------------------------- # Constants (non-pricing, do not vary by region) # --------------------------------------------------------------------------- # Aurora bills storage on actual usage per GiB-month with dynamic resizing — # there is no fixed minimum billed storage. (No MIN_STORAGE_GIB floor.) HOURS_PER_MONTH = 730 ACU_MIN = 0.5 ACU_MAX = 256 IO_OPT_COMPUTE_MULTIPLIER = 1.30 # I/O-Optimized compute premium GIB_PER_ACU = 2.0 # Each ACU provides ~2 GiB of memory # Instance family -> ACU ratio ACU_FAMILY_RATIO = {"r": 4, "m": 2, "t": 2, "c": 1, "x": 4} # AWS region code -> Pricing API "location" name _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-west-3": "EU (Paris)", "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-northeast-2": "Asia Pacific (Seoul)", "ap-south-1": "Asia Pacific (Mumbai)", "ca-central-1": "Canada (Central)", "sa-east-1": "South America (Sao Paulo)", } # --------------------------------------------------------------------------- # Active pricing & catalog (mutable — overwritten by refresh_pricing()) # --------------------------------------------------------------------------- ACU_PRICE_STANDARD = _STATIC_ACU_PRICE_STANDARD ACU_PRICE_IO_OPTIMIZED = _STATIC_ACU_PRICE_IO_OPTIMIZED STORAGE_STANDARD_PER_GIB = _STATIC_STORAGE_STANDARD_PER_GIB STORAGE_IO_OPT_PER_GIB = _STATIC_STORAGE_IO_OPT_PER_GIB INSTANCE_SPECS = dict(_STATIC_INSTANCE_SPECS) # Tracks where the active data came from _pricing_source: dict[str, Any] = { "source": "static_fallback", "region": _STATIC_REGION, "last_updated": _STATIC_LAST_UPDATED, "details": "Built-in us-east-1 defaults", } # --------------------------------------------------------------------------- # Live AWS API fetchers # --------------------------------------------------------------------------- def _fetch_instance_pricing(region: str) -> dict[str, float]: """Fetch on-demand hourly pricing for Aurora instances via the Pricing API. The Pricing API is only available in us-east-1 and ap-south-1, but returns pricing for any region. Queries aurora-mysql (covers all instance types). Returns dict: instance_type -> price_per_hour. """ import boto3 location = _REGION_NAMES.get(region) if not location: raise ValueError( f"Unknown region '{region}'. Supported: {', '.join(sorted(_REGION_NAMES))}" ) 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": "deploymentOption", "Value": "Single-AZ"}, {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"}, ] prices = {} paginator = pricing.get_paginator("get_products") for page in paginator.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", {}) instance_type = attrs.get("instanceType", "") if not instance_type.startswith("db."): continue # Skip I/O-Optimized SKUs if "IOOptimized" in attrs.get("usagetype", ""): continue terms = item.get("terms", {}).get("OnDemand", {}) for term in terms.values(): for dim in term.get("priceDimensions", {}).values(): try: price = float(dim.get("pricePerUnit", {}).get("USD", "0")) except (ValueError, TypeError): continue if price > 0: prices[instance_type] = price return prices def _fetch_instance_pricing_bulk(region: str) -> dict[str, float]: """Fetch on-demand Aurora MySQL pricing from the public AWS Bulk Pricing CSV. No IAM credentials required — this is a publicly accessible HTTPS endpoint. Used as a fallback when the Pricing API is not accessible (AccessDeniedException). Returns dict: instance_type -> price_per_hour (Aurora Standard only). """ import csv import io import urllib.request url = ( f"https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonRDS/" f"current/{region}/index.csv" ) req = urllib.request.Request(url, headers={"Accept-Encoding": "identity"}) with urllib.request.urlopen(req, timeout=30) as resp: # CSV has metadata rows before the header; find the header row raw = resp.read().decode("utf-8") # AWS pricing CSVs start with metadata lines (FormatVersion, Disclaimer, etc.) # before the actual column header. Find the header row (starts with "SKU"). lines = raw.splitlines() header_idx = 0 for i, line in enumerate(lines): if line.startswith('"SKU"') or line.startswith("SKU"): header_idx = i break reader = csv.DictReader(io.StringIO("\n".join(lines[header_idx:]))) prices = {} for row in reader: if row.get("Database Engine") != "Aurora MySQL": continue if row.get("Deployment Option") != "Single-AZ": continue # On-demand hourly rates only. The CSV also carries Reserved rows # (including fixed-fee Quantity rows with values like 2530), which would # otherwise overwrite the real hourly price via last-write-wins. if row.get("TermType") != "OnDemand" or row.get("Unit") != "Hrs": continue instance_type = row.get("Instance Type", "") if not instance_type.startswith("db."): continue # Skip I/O-Optimized SKUs. The column is "usageType" (camelCase) in the # AWS RDS bulk CSV; I/O-Optimized is encoded there as InstanceUsageIOOptimized:*. usage_type = row.get("usageType", "") if "IOOptimized" in usage_type: continue try: price = float(row.get("PricePerUnit", "0")) except (ValueError, TypeError): continue if price > 0: prices[instance_type] = price return prices def _fetch_acu_and_storage_pricing(region: str) -> dict: """Fetch ACU and storage pricing from the Pricing API. Returns dict with acu_standard, acu_io_optimized, storage_standard, storage_io_optimized keys. """ import boto3 location = _REGION_NAMES.get(region) if not location: raise ValueError(f"Unknown region '{region}'") pricing = boto3.client("pricing", region_name="us-east-1") result = {} # ACU pricing acu_filters = [ {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"}, {"Type": "TERM_MATCH", "Field": "location", "Value": location}, {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"}, ] paginator = pricing.get_paginator("get_products") for page in paginator.paginate(ServiceCode="AmazonRDS", Filters=acu_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", {}) usagetype = attrs.get("usagetype", "") if "ServerlessV2Usage" in usagetype and "IOOptimized" not in usagetype: terms = item.get("terms", {}).get("OnDemand", {}) for term in terms.values(): for dim in term.get("priceDimensions", {}).values(): price = float(dim.get("pricePerUnit", {}).get("USD", "0")) if price > 0: result["acu_standard"] = price result["acu_io_optimized"] = round(price * IO_OPT_COMPUTE_MULTIPLIER, 4) # Storage pricing storage_filters = [ {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"}, {"Type": "TERM_MATCH", "Field": "location", "Value": location}, {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"}, {"Type": "TERM_MATCH", "Field": "productFamily", "Value": "Database Storage"}, ] for page in paginator.paginate(ServiceCode="AmazonRDS", Filters=storage_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", {}) usagetype = attrs.get("usagetype", "") terms = item.get("terms", {}).get("OnDemand", {}) for term in terms.values(): for dim in term.get("priceDimensions", {}).values(): price = float(dim.get("pricePerUnit", {}).get("USD", "0")) if price <= 0: continue if "IOOptimized" in usagetype: result["storage_io_optimized"] = price elif "Aurora:StorageUsage" in usagetype: result["storage_standard"] = price return result def _fetch_instance_catalog(region: str) -> dict[str, tuple]: """Discover Aurora instance types via RDS + EC2 APIs. Returns dict: instance_type -> (vcpus, memory_gib, 0.0). Price is 0.0 here; caller merges with pricing data. """ import boto3 rds = boto3.client("rds", region_name=region) instances = {} for engine in ("aurora-mysql",): try: paginator = rds.get_paginator("describe_orderable_db_instance_options") for page in paginator.paginate(Engine=engine): for opt in page.get("OrderableDBInstanceOptions", []): name = opt.get("DBInstanceClass", "") if name.startswith("db.") and name != "db.serverless": instances[name] = {"vcpus": 0, "memory_gib": 0.0} except Exception: pass if not instances: return {} # Get vCPU/memory from EC2 describe_instance_types ec2 = boto3.client("ec2", region_name=region) ec2_names = [name.replace("db.", "", 1) for name in instances] for i in range(0, len(ec2_names), 100): batch = ec2_names[i : i + 100] try: resp = ec2.describe_instance_types(InstanceTypes=batch) for it in resp.get("InstanceTypes", []): db_name = "db." + it["InstanceType"] if db_name in instances: instances[db_name]["vcpus"] = it.get("VCpuInfo", {}).get("DefaultVCpus", 0) instances[db_name]["memory_gib"] = ( it.get("MemoryInfo", {}).get("SizeInMiB", 0) / 1024 ) except Exception: # Try one by one for families EC2 doesn't know about for ec2_name in batch: try: resp = ec2.describe_instance_types(InstanceTypes=[ec2_name]) for it in resp.get("InstanceTypes", []): db_name = "db." + it["InstanceType"] if db_name in instances: instances[db_name]["vcpus"] = it.get("VCpuInfo", {}).get( "DefaultVCpus", 0 ) instances[db_name]["memory_gib"] = ( it.get("MemoryInfo", {}).get("SizeInMiB", 0) / 1024 ) except Exception: pass # Convert to tuple format, drop entries missing vCPU/memory catalog = {} for name, spec in instances.items(): if spec["vcpus"] > 0 and spec["memory_gib"] > 0: catalog[name] = (spec["vcpus"], spec["memory_gib"], 0.0) return catalog def refresh_pricing(region: str = "us-east-1") -> dict: """Refresh all pricing and instance data from AWS APIs. Tries live APIs first. On any failure, falls back to static defaults and reports what succeeded and what didn't. Returns a summary dict describing the pricing source. """ global ACU_PRICE_STANDARD, ACU_PRICE_IO_OPTIMIZED global STORAGE_STANDARD_PER_GIB, STORAGE_IO_OPT_PER_GIB global INSTANCE_SPECS, _pricing_source errors = [] live_instances = 0 live_prices = 0 live_acu = False # 1. Try to fetch the instance catalog (RDS + EC2) api_catalog = {} try: api_catalog = _fetch_instance_catalog(region) live_instances = len(api_catalog) except Exception as e: errors.append(f"Instance catalog: {e}") # 2. Try to fetch instance pricing (Pricing API, then public bulk CSV fallback) instance_prices = {} try: instance_prices = _fetch_instance_pricing(region) live_prices = len(instance_prices) except Exception as e: errors.append(f"Instance pricing (API): {e}") # Fallback: public bulk pricing CSV (no IAM credentials needed) try: instance_prices = _fetch_instance_pricing_bulk(region) live_prices = len(instance_prices) if live_prices > 0: errors[-1] += " [recovered via bulk pricing CSV]" except Exception as e2: errors.append(f"Instance pricing (bulk CSV): {e2}") # 3. Try to fetch ACU + storage pricing try: acu_data = _fetch_acu_and_storage_pricing(region) if "acu_standard" in acu_data: ACU_PRICE_STANDARD = acu_data["acu_standard"] ACU_PRICE_IO_OPTIMIZED = acu_data.get( "acu_io_optimized", round(acu_data["acu_standard"] * IO_OPT_COMPUTE_MULTIPLIER, 4), ) live_acu = True if "storage_standard" in acu_data: STORAGE_STANDARD_PER_GIB = acu_data["storage_standard"] if "storage_io_optimized" in acu_data: STORAGE_IO_OPT_PER_GIB = acu_data["storage_io_optimized"] except Exception as e: errors.append(f"ACU/storage pricing: {e}") # 4. Merge: start with static, overlay API catalog, overlay prices merged = dict(_STATIC_INSTANCE_SPECS) for name, (vcpus, mem, _) in api_catalog.items(): price = instance_prices.get(name, 0.0) # If API didn't return a price, keep static price if we have one if price == 0.0 and name in _STATIC_INSTANCE_SPECS: price = _STATIC_INSTANCE_SPECS[name][2] merged[name] = (vcpus, mem, price) # For instances in static but not in API catalog, update price if available for name in _STATIC_INSTANCE_SPECS: if name not in api_catalog and name in instance_prices: v, m, _ = _STATIC_INSTANCE_SPECS[name] merged[name] = (v, m, instance_prices[name]) INSTANCE_SPECS = merged # Determine source description if not errors: source = "live" details = ( f"Live AWS APIs ({region}): {live_instances} instance types, " f"{live_prices} prices, ACU=${ACU_PRICE_STANDARD}/hr" ) elif live_prices > 0 or live_acu: source = "partial_live" details = ( f"Partial live data ({region}): {live_instances} instances, " f"{live_prices} prices. Gaps filled from static defaults. " f"Errors: {'; '.join(errors)}" ) else: # Full fallback source = "static_fallback" INSTANCE_SPECS = dict(_STATIC_INSTANCE_SPECS) ACU_PRICE_STANDARD = _STATIC_ACU_PRICE_STANDARD ACU_PRICE_IO_OPTIMIZED = _STATIC_ACU_PRICE_IO_OPTIMIZED STORAGE_STANDARD_PER_GIB = _STATIC_STORAGE_STANDARD_PER_GIB STORAGE_IO_OPT_PER_GIB = _STATIC_STORAGE_IO_OPT_PER_GIB details = ( f"Static fallback (us-east-1, {_STATIC_LAST_UPDATED}). " f"Live fetch failed: {'; '.join(errors)}" ) _pricing_source = { "source": source, "region": region, "last_updated": _STATIC_LAST_UPDATED if source == "static_fallback" else "now", "details": details, "instance_count": len(INSTANCE_SPECS), "acu_price_standard": ACU_PRICE_STANDARD, "storage_price_standard": STORAGE_STANDARD_PER_GIB, } if errors: _pricing_source["errors"] = errors return _pricing_source def get_pricing_source() -> dict: """Return metadata about the active pricing data source.""" return dict(_pricing_source) # --------------------------------------------------------------------------- # Core calculation functions # --------------------------------------------------------------------------- def round_up_to_half(value: float) -> float: """Round up to nearest 0.5 ACU.""" return math.ceil(value * 2) / 2 def family_ratio(instance_type: str) -> int: """Get ACU ratio for an instance family.""" m = re.match(r"db\.([a-z])", instance_type) if m: return ACU_FAMILY_RATIO.get(m.group(1), 4) return 4 def get_instance_specs(instance_type: str) -> tuple: """Get (vcpus, memory_gib, price_per_hour) for an instance type.""" if instance_type in INSTANCE_SPECS: return INSTANCE_SPECS[instance_type] raise ValueError( f"Unknown instance type: {instance_type}. " f"Supported: {', '.join(sorted(INSTANCE_SPECS.keys()))}" ) def estimate_acu( cpu_p95: float, cpu_max: float, vcpus: int, instance_type: str, cpu_avg: float = 0, ) -> dict: """Estimate ACU needed for a workload. Returns dict with typical ACU, min/max recommendations, and breakdown. """ ratio = family_ratio(instance_type) # Typical ACU: weighted 95/5 blend weighted_cpu = (cpu_p95 * 0.95 + cpu_max * 0.05) / 100 raw_typical = weighted_cpu * vcpus * ratio typical_acu = round_up_to_half(raw_typical) typical_acu = max(ACU_MIN, min(typical_acu, ACU_MAX)) # Peak ACU raw_peak = (cpu_max / 100) * vcpus * ratio peak_acu = round_up_to_half(raw_peak) # Average ACU (for min recommendation) if cpu_avg > 0: avg_acu = round_up_to_half((cpu_avg / 100) * vcpus * ratio) else: # Estimate average as 60% of P95 when not provided avg_acu = round_up_to_half((cpu_p95 * 0.6 / 100) * vcpus * ratio) exceeds_capacity = raw_typical > ACU_MAX or raw_peak > ACU_MAX return { "typical_acu": typical_acu, "peak_acu": peak_acu, "avg_acu": avg_acu, "raw_typical": round(raw_typical, 2), "raw_peak": round(raw_peak, 2), "family_ratio": ratio, "exceeds_capacity": exceeds_capacity, } def recommend_min_max( acu_result: dict, connections: int = 0, working_set_gib: float = 0, ) -> dict: """Recommend min/max ACU settings.""" avg_acu = acu_result["avg_acu"] typical_acu = acu_result["typical_acu"] peak_acu = acu_result["peak_acu"] # Connection floor conn_mem_gib = connections * 10 / 1024 # ~10 MB per connection average conn_acu = round_up_to_half(conn_mem_gib / GIB_PER_ACU) # Memory floor (advisory — working set) mem_acu = round_up_to_half(working_set_gib / GIB_PER_ACU) if working_set_gib > 0 else 0 # Min: based on avg CPU + connection floor (uncapped first, so we can detect # a baseline that already exceeds serverless limits). raw_min = max(ACU_MIN, avg_acu, conn_acu) # Max: peak + 30% headroom, at least 1.5x typical recommended_max = round_up_to_half(peak_acu * 1.3) recommended_max = max(recommended_max, round_up_to_half(typical_acu * 1.5)) recommended_max = min(recommended_max, ACU_MAX) # The workload's baseline doesn't fit a single serverless instance when the # uncapped min exceeds the ACU ceiling or the (capped) max — flag it, mirroring # estimate_acu's exceeds_capacity. exceeds_capacity = raw_min > ACU_MAX or raw_min > recommended_max # Cap min at ACU_MAX and enforce the invariant: min must never exceed max. recommended_min = min(raw_min, ACU_MAX, recommended_max) # Memory advisory memory_advisory = None if mem_acu > recommended_min: memory_advisory = ( f"Working set needs {mem_acu} ACU ({working_set_gib:.1f} GiB / " f"{GIB_PER_ACU} GiB per ACU). Your min ACU ({recommended_min}) is below this. " f"Setting min to {mem_acu} ACU keeps the working set cached and avoids " f"cold-cache I/O penalties on scale-up. Trade-off: higher baseline cost." ) return { "recommended_min": recommended_min, "recommended_max": recommended_max, "connection_floor_acu": conn_acu, "memory_floor_acu": mem_acu, "memory_advisory": memory_advisory, "exceeds_capacity": exceeds_capacity, } def calculate_costs( typical_acu: float, min_acu: float, max_acu: float, storage_gib: float, provisioned_instance: str, num_provisioned_instances: int = 1, exceeds_capacity: bool = False, ) -> dict: """Calculate and compare serverless vs provisioned costs.""" _, _, price_per_hour = get_instance_specs(provisioned_instance) # Provisioned cost prov_compute = price_per_hour * HOURS_PER_MONTH * num_provisioned_instances prov_storage = storage_gib * STORAGE_STANDARD_PER_GIB prov_total = prov_compute + prov_storage # Serverless cost (typical steady-state) sv_compute = typical_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH sv_storage = storage_gib * STORAGE_STANDARD_PER_GIB sv_total = sv_compute + sv_storage # Serverless cost range sv_low = min_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH + sv_storage sv_high = max_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH + sv_storage # Savings savings = prov_total - sv_total savings_pct = (savings / prov_total * 100) if prov_total > 0 else 0 # Recommendation logic if exceeds_capacity: recommendation = "not_recommended" reason = ( f"Workload's baseline/peak demand exceeds the {ACU_MAX:.0f} ACU serverless " f"maximum. Stay with provisioned or split across multiple serverless clusters." ) elif savings_pct > 10 and sv_high <= prov_total * 2: recommendation = "recommended" reason = ( f"Serverless saves ${savings:.0f}/mo ({savings_pct:.0f}%) vs provisioned. " f"Cost range: ${sv_low:.0f}–${sv_high:.0f}/mo." ) elif savings_pct > 10: recommendation = "consider" reason = ( f"Typical cost is lower (${sv_total:.0f} vs ${prov_total:.0f}/mo), but " f"peak cost could reach ${sv_high:.0f}/mo. Variable workloads benefit; " f"sustained peaks may not." ) elif savings_pct > -5: recommendation = "consider" reason = ( f"Similar cost (${sv_total:.0f} vs ${prov_total:.0f}/mo). Choose serverless " f"for auto-scaling and zero management overhead." ) elif savings_pct > -30: recommendation = "more_expensive" reason = ( f"Serverless costs ${abs(savings):.0f}/mo more than provisioned " f"(${sv_total:.0f} vs ${prov_total:.0f}/mo)." ) else: recommendation = "not_recommended" reason = ( f"Serverless at ${sv_total:.0f}/mo is {abs(savings_pct):.0f}% more expensive " f"than provisioned at ${prov_total:.0f}/mo. Sustained workloads are cheaper " f"on provisioned instances." ) return { "provisioned": { "instance_type": provisioned_instance, "num_instances": num_provisioned_instances, "compute_monthly": round(prov_compute, 2), "storage_monthly": round(prov_storage, 2), "total_monthly": round(prov_total, 2), }, "serverless": { "typical_acu": typical_acu, "compute_monthly": round(sv_compute, 2), "storage_monthly": round(sv_storage, 2), "total_monthly": round(sv_total, 2), "cost_range": { "low": round(sv_low, 2), "typical": round(sv_total, 2), "high": round(sv_high, 2), }, }, "savings_monthly": round(savings, 2), "savings_pct": round(savings_pct, 1), "recommendation": recommendation, "reason": reason, } def format_table(result: dict) -> str: """Format result as a readable text table.""" lines = [] source = result.get("pricing_source", _pricing_source) tag = source.get("source", "static_fallback").replace("_", " ").title() lines.append("=" * 65) lines.append(" Aurora Serverless v2 — ACU Estimate & Cost Comparison") lines.append(f" Pricing: {tag} ({source.get('region', '?')})") lines.append("=" * 65) # ACU settings acu = result["acu_settings"] lines.append("") lines.append(" ACU Configuration") lines.append(" " + "-" * 45) lines.append(f" Recommended Min ACU: {acu['recommended_min']:.1f}") lines.append(f" Recommended Max ACU: {acu['recommended_max']:.1f}") lines.append(f" Typical ACU: {acu['typical_acu']:.1f}") lines.append(f" Peak ACU: {acu['peak_acu']:.1f}") if acu.get("connection_floor_acu", 0) > 0: lines.append(f" Connection floor: {acu['connection_floor_acu']:.1f} ACU") if acu.get("memory_floor_acu", 0) > 0: lines.append(f" Memory floor: {acu['memory_floor_acu']:.1f} ACU (advisory)") if acu.get("memory_advisory"): lines.append(f" NOTE: {acu['memory_advisory']}") # Cost comparison costs = result["cost_comparison"] prov = costs["provisioned"] sv = costs["serverless"] lines.append("") lines.append(" Monthly Cost Comparison") lines.append(" " + "-" * 45) lines.append(f" {'':30s} {'Provisioned':>14s} {'Serverless':>14s}") lines.append( f" {'Compute':30s} {'$'+str(prov['compute_monthly']):>14s} {'$'+str(sv['compute_monthly']):>14s}" ) lines.append( f" {'Storage':30s} {'$'+str(prov['storage_monthly']):>14s} {'$'+str(sv['storage_monthly']):>14s}" ) lines.append( f" {'Total':30s} {'$'+str(prov['total_monthly']):>14s} {'$'+str(sv['total_monthly']):>14s}" ) lines.append("") lines.append( f" Serverless cost range: ${sv['cost_range']['low']:.0f} – ${sv['cost_range']['high']:.0f}/mo" ) lines.append(f" Savings: ${costs['savings_monthly']:.0f}/mo ({costs['savings_pct']:.0f}%)") # Recommendation lines.append("") lines.append(f" Recommendation: {costs['recommendation'].upper()}") lines.append(f" {costs['reason']}") lines.append("") lines.append("=" * 65) return "\n".join(lines) def run_estimate(args) -> dict: """Run full estimation from CLI arguments.""" vcpus, memory_gib, price = get_instance_specs(args.instance) acu_result = estimate_acu( cpu_p95=args.cpu_p95, cpu_max=args.cpu_max, vcpus=vcpus, instance_type=args.instance, cpu_avg=args.cpu_avg, ) min_max = recommend_min_max( acu_result, connections=args.connections, working_set_gib=args.working_set, ) # Workload overflows serverless if EITHER signal trips: estimate_acu's # typical/peak check, or recommend_min_max's baseline-min check. exceeds_capacity = acu_result["exceeds_capacity"] or min_max["exceeds_capacity"] costs = calculate_costs( typical_acu=acu_result["typical_acu"], min_acu=min_max["recommended_min"], max_acu=min_max["recommended_max"], storage_gib=args.storage, provisioned_instance=args.instance, num_provisioned_instances=args.num_instances, exceeds_capacity=exceeds_capacity, ) return { "input": { "instance_type": args.instance, "vcpus": vcpus, "memory_gib": memory_gib, "cpu_p95": args.cpu_p95, "cpu_max": args.cpu_max, "cpu_avg": args.cpu_avg, "connections": args.connections, "working_set_gib": args.working_set, "storage_gib": args.storage, "num_instances": args.num_instances, }, "acu_settings": { "recommended_min": min_max["recommended_min"], "recommended_max": min_max["recommended_max"], "typical_acu": acu_result["typical_acu"], "peak_acu": acu_result["peak_acu"], "avg_acu": acu_result["avg_acu"], "connection_floor_acu": min_max["connection_floor_acu"], "memory_floor_acu": min_max["memory_floor_acu"], "memory_advisory": min_max["memory_advisory"], "exceeds_capacity": exceeds_capacity, }, "cost_comparison": costs, "pricing_source": get_pricing_source(), } def main(): # Shared flags accepted both before the subcommand and after it (so e.g. # `... estimate --region X --offline` and `... --region X estimate` both work). # Defaults are SUPPRESSed here so a subparser copy does NOT re-apply its own # default and clobber a value the user passed before the subcommand; the real # defaults are resolved once, after parsing, below. common = argparse.ArgumentParser(add_help=False) common.add_argument( "--region", default=argparse.SUPPRESS, help="AWS region for pricing (default: us-east-1). " "Live pricing requires boto3 + AWS credentials.", ) common.add_argument( "--offline", action="store_true", default=argparse.SUPPRESS, help="Skip live API calls, use static fallback data only.", ) parser = argparse.ArgumentParser( description="Aurora Serverless v2 ACU Calculator", parents=[common] ) sub = parser.add_subparsers(dest="command") # estimate command est = sub.add_parser("estimate", parents=[common], help="Estimate ACU sizing and compare costs") est.add_argument( "--instance", required=True, help="Current provisioned instance type (e.g., db.r6g.xlarge)" ) est.add_argument("--cpu-p95", type=float, required=True, help="P95 CPU utilization (0-100)") est.add_argument("--cpu-max", type=float, required=True, help="Maximum CPU utilization (0-100)") est.add_argument( "--cpu-avg", type=float, default=0, help="Average CPU utilization (0-100), estimated if omitted", ) est.add_argument("--connections", type=int, default=0, help="Peak connection count") est.add_argument("--working-set", type=float, default=0, help="Working set size in GiB") est.add_argument("--storage", type=float, default=10, help="Storage in GiB") est.add_argument( "--num-instances", type=int, default=1, help="Number of provisioned instances (for cost comparison)", ) est.add_argument("--format", choices=["json", "table"], default="json", help="Output format") # list-instances command sub.add_parser( "list-instances", parents=[common], help="List supported instance types with specs and pricing", ) # pricing-source command sub.add_parser( "pricing-source", parents=[common], help="Show where pricing data is coming from" ) args = parser.parse_args() # Resolve shared-flag defaults once (they were SUPPRESSed on both the main and # subparsers so neither position clobbers the other; an explicit flag in either # position lands in the namespace, otherwise we apply the default here). if not hasattr(args, "region"): args.region = "us-east-1" if not hasattr(args, "offline"): args.offline = False # Refresh pricing (live or offline) if not args.offline: source = refresh_pricing(region=args.region) if args.command != "pricing-source": # Brief status line to stderr so it doesn't pollute JSON output tag = ( "LIVE" if source["source"] == "live" else ("PARTIAL" if source["source"] == "partial_live" else "STATIC") ) print( f"[Pricing: {tag} — {source['region']}, " f"{source['instance_count']} instances, " f"ACU=${source['acu_price_standard']}/hr]", file=sys.stderr, ) if args.command == "estimate": result = run_estimate(args) if args.format == "table": print(format_table(result)) else: print(json.dumps(result, indent=2)) elif args.command == "list-instances": source = get_pricing_source() print(f"Pricing source: {source['source']} ({source['region']})") print(f"{'Instance Type':<25s} {'vCPUs':>6s} {'Memory':>8s} {'$/hr':>8s} {'$/mo':>10s}") print("-" * 62) for name in sorted(INSTANCE_SPECS.keys()): v, m, p = INSTANCE_SPECS[name] print(f"{name:<25s} {v:>6d} {m:>6.0f} GiB {p:>8.3f} {p*730:>10.2f}") print(f"\nTotal: {len(INSTANCE_SPECS)} instance types") elif args.command == "pricing-source": print(json.dumps(get_pricing_source(), indent=2)) else: parser.print_help() if __name__ == "__main__": main()