Setting the file. One moment. 10
Setup DevOps Agent
33
AWS Deployment
57.9
IO Optimized Instructions
613
def calculate_costs
— line 613
This file
- Number
- 57.30
- Position
- 30 of 32
- Type
- Python
- Size
- 36 KB
- Lines
- 942
scripts/acu_calculator.py
Python·942 lines·36 KB
16import json
17import math
18import re
19import sys
20from typing import Any
21
22# ---------------------------------------------------------------------------
23# Static fallback data (us-east-1)
24# Used when AWS APIs are unavailable (no credentials, offline, API errors).
25# ---------------------------------------------------------------------------
26_STATIC_ACU_PRICE_STANDARD = 0.12 # $/ACU-Hr
27_STATIC_ACU_PRICE_IO_OPTIMIZED = 0.156 # $/ACU-Hr (30% premium)
28_STATIC_STORAGE_STANDARD_PER_GIB = 0.10 # $/GiB-month
29_STATIC_STORAGE_IO_OPT_PER_GIB = 0.225 # $/GiB-month
30_STATIC_LAST_UPDATED = "2026-03-28"
31_STATIC_REGION = "us-east-1"
32
33# Static instance specs: {name: (vcpus, memory_gib, price_per_hour)}
34_STATIC_INSTANCE_SPECS = {
35 "db.t3.medium": (2, 4, 0.082),
36 "db.t3.large": (2, 8, 0.164),
37 "db.t4g.medium": (2, 4, 0.073),
38 "db.t4g.large": (2, 8, 0.146),
39 "db.r5.large": (2, 16, 0.290),
40 "db.r5.xlarge": (4, 32, 0.580),
41 "db.r5.2xlarge": (8, 64, 1.160),
42 "db.r5.4xlarge": (16, 128, 2.320),
43 "db.r5.8xlarge": (32, 256, 4.640),
44 "db.r5.12xlarge": (48, 384, 6.960),
45 "db.r5.16xlarge": (64, 512, 9.280),
46 "db.r5.24xlarge": (96, 768, 13.920),
47 "db.r6g.large": (2, 16, 0.260),
48 "db.r6g.xlarge": (4, 32, 0.519),
49 "db.r6g.2xlarge": (8, 64, 1.038),
50 "db.r6g.4xlarge": (16, 128, 2.076),
51 "db.r6g.8xlarge": (32, 256, 4.152),
52 "db.r6g.12xlarge": (48, 384, 6.228),
53 "db.r6g.16xlarge": (64, 512, 8.304),
54 "db.r7g.large": (2, 16, 0.276),
55 "db.r7g.xlarge": (4, 32, 0.553),
56 "db.r7g.2xlarge": (8, 64, 1.106),
57 "db.r7g.4xlarge": (16, 128, 2.211),
58 "db.r7g.8xlarge": (32, 256, 4.422),
59 "db.r7g.12xlarge": (48, 384, 6.633),
60 "db.r7g.16xlarge": (64, 512, 8.844),
61 "db.r8g.large": (2, 16, 0.276),
62 "db.r8g.xlarge": (4, 32, 0.552),
63 "db.r8g.2xlarge": (8, 64, 1.104),
64 "db.r8g.4xlarge": (16, 128, 2.208),
65 "db.r8g.8xlarge": (32, 256, 4.416),
66 "db.r8g.12xlarge": (48, 384, 6.624),
67 "db.r8g.16xlarge": (64, 512, 8.832),
68 "db.r8g.24xlarge": (96, 768, 13.248),
69 "db.r8g.48xlarge": (192, 1536, 26.496),
70}
71
72# ---------------------------------------------------------------------------
73# Constants (non-pricing, do not vary by region)
74# ---------------------------------------------------------------------------
75# Aurora bills storage on actual usage per GiB-month with dynamic resizing —
76# there is no fixed minimum billed storage. (No MIN_STORAGE_GIB floor.)
77HOURS_PER_MONTH = 730
78ACU_MIN = 0.5
79ACU_MAX = 256
80IO_OPT_COMPUTE_MULTIPLIER = 1.30 # I/O-Optimized compute premium
81GIB_PER_ACU = 2.0 # Each ACU provides ~2 GiB of memory
82
83# Instance family -> ACU ratio
84ACU_FAMILY_RATIO = {"r": 4, "m": 2, "t": 2, "c": 1, "x": 4}
85
86# AWS region code -> Pricing API "location" name
87_REGION_NAMES = {
88 "us-east-1": "US East (N. Virginia)",
89 "us-east-2": "US East (Ohio)",
90 "us-west-1": "US West (N. California)",
91 "us-west-2": "US West (Oregon)",
92 "eu-west-1": "EU (Ireland)",
93 "eu-west-2": "EU (London)",
94 "eu-west-3": "EU (Paris)",
95 "eu-central-1": "EU (Frankfurt)",
96 "eu-north-1": "EU (Stockholm)",
97 "ap-southeast-1": "Asia Pacific (Singapore)",
98 "ap-southeast-2": "Asia Pacific (Sydney)",
99 "ap-northeast-1": "Asia Pacific (Tokyo)",
100 "ap-northeast-2": "Asia Pacific (Seoul)",
101 "ap-south-1": "Asia Pacific (Mumbai)",
102 "ca-central-1": "Canada (Central)",
103 "sa-east-1": "South America (Sao Paulo)",
104}
105
106# ---------------------------------------------------------------------------
107# Active pricing & catalog (mutable — overwritten by refresh_pricing())
108# ---------------------------------------------------------------------------
109ACU_PRICE_STANDARD = _STATIC_ACU_PRICE_STANDARD
110ACU_PRICE_IO_OPTIMIZED = _STATIC_ACU_PRICE_IO_OPTIMIZED
111STORAGE_STANDARD_PER_GIB = _STATIC_STORAGE_STANDARD_PER_GIB
112STORAGE_IO_OPT_PER_GIB = _STATIC_STORAGE_IO_OPT_PER_GIB
113INSTANCE_SPECS = dict(_STATIC_INSTANCE_SPECS)
114
115# Tracks where the active data came from
116_pricing_source: dict[str, Any] = {
117 "source": "static_fallback",
118 "region": _STATIC_REGION,
119 "last_updated": _STATIC_LAST_UPDATED,
120 "details": "Built-in us-east-1 defaults",
121}
122
123
124# ---------------------------------------------------------------------------
125# Live AWS API fetchers
126# ---------------------------------------------------------------------------
127
128
129def _fetch_instance_pricing(region: str) -> dict[str, float]:
130 """Fetch on-demand hourly pricing for Aurora instances via the Pricing API.
131
132 The Pricing API is only available in us-east-1 and ap-south-1, but returns
133 pricing for any region. Queries aurora-mysql (covers all instance types).
134
135 Returns dict: instance_type -> price_per_hour.
136 """
137 import boto3
138
139 location = _REGION_NAMES.get(region)
140 if not location:
141 raise ValueError(
142 f"Unknown region '{region}'. Supported: {', '.join(sorted(_REGION_NAMES))}"
143 )
144
145 pricing = boto3.client("pricing", region_name="us-east-1")
146 filters = [
147 {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"},
148 {"Type": "TERM_MATCH", "Field": "location", "Value": location},
149 {"Type": "TERM_MATCH", "Field": "deploymentOption", "Value": "Single-AZ"},
150 {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"},
151 ]
152
153 prices = {}
154 paginator = pricing.get_paginator("get_products")
155 for page in paginator.paginate(ServiceCode="AmazonRDS", Filters=filters):
156 for item_json in page["PriceList"]:
157 item = json.loads(item_json) if isinstance(item_json, str) else item_json
158 attrs = item.get("product", {}).get("attributes", {})
159 instance_type = attrs.get("instanceType", "")
160 if not instance_type.startswith("db."):
161 continue
162 # Skip I/O-Optimized SKUs
163 if "IOOptimized" in attrs.get("usagetype", ""):
164 continue
165 terms = item.get("terms", {}).get("OnDemand", {})
166 for term in terms.values():
167 for dim in term.get("priceDimensions", {}).values():
168 try:
169 price = float(dim.get("pricePerUnit", {}).get("USD", "0"))
170 except (ValueError, TypeError):
171 continue
172 if price > 0:
173 prices[instance_type] = price
174
175 return prices
176
177
178def _fetch_instance_pricing_bulk(region: str) -> dict[str, float]:
179 """Fetch on-demand Aurora MySQL pricing from the public AWS Bulk Pricing CSV.
180
181 No IAM credentials required — this is a publicly accessible HTTPS endpoint.
182 Used as a fallback when the Pricing API is not accessible (AccessDeniedException).
183
184 Returns dict: instance_type -> price_per_hour (Aurora Standard only).
185 """
186 import csv
187 import io
188 import urllib.request
189
190 url = (
191 f"https://pricing.us-east-1.amazonaws.com/offers/v1.0/aws/AmazonRDS/"
192 f"current/{region}/index.csv"
193 )
194 req = urllib.request.Request(url, headers={"Accept-Encoding": "identity"})
195 with urllib.request.urlopen(req, timeout=30) as resp:
196 # CSV has metadata rows before the header; find the header row
197 raw = resp.read().decode("utf-8")
198
199 # AWS pricing CSVs start with metadata lines (FormatVersion, Disclaimer, etc.)
200 # before the actual column header. Find the header row (starts with "SKU").
201 lines = raw.splitlines()
202 header_idx = 0
203 for i, line in enumerate(lines):
204 if line.startswith('"SKU"') or line.startswith("SKU"):
205 header_idx = i
206 break
207 reader = csv.DictReader(io.StringIO("\n".join(lines[header_idx:])))
208 prices = {}
209 for row in reader:
210 if row.get("Database Engine") != "Aurora MySQL":
211 continue
212 if row.get("Deployment Option") != "Single-AZ":
213 continue
214 # On-demand hourly rates only. The CSV also carries Reserved rows
215 # (including fixed-fee Quantity rows with values like 2530), which would
216 # otherwise overwrite the real hourly price via last-write-wins.
217 if row.get("TermType") != "OnDemand" or row.get("Unit") != "Hrs":
218 continue
219 instance_type = row.get("Instance Type", "")
220 if not instance_type.startswith("db."):
221 continue
222 # Skip I/O-Optimized SKUs. The column is "usageType" (camelCase) in the
223 # AWS RDS bulk CSV; I/O-Optimized is encoded there as InstanceUsageIOOptimized:*.
224 usage_type = row.get("usageType", "")
225 if "IOOptimized" in usage_type:
226 continue
227 try:
228 price = float(row.get("PricePerUnit", "0"))
229 except (ValueError, TypeError):
230 continue
231 if price > 0:
232 prices[instance_type] = price
233
234 return prices
235
236
237def _fetch_acu_and_storage_pricing(region: str) -> dict:
238 """Fetch ACU and storage pricing from the Pricing API.
239
240 Returns dict with acu_standard, acu_io_optimized, storage_standard,
241 storage_io_optimized keys.
242 """
243 import boto3
244
245 location = _REGION_NAMES.get(region)
246 if not location:
247 raise ValueError(f"Unknown region '{region}'")
248
249 pricing = boto3.client("pricing", region_name="us-east-1")
250 result = {}
251
252 # ACU pricing
253 acu_filters = [
254 {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"},
255 {"Type": "TERM_MATCH", "Field": "location", "Value": location},
256 {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"},
257 ]
258 paginator = pricing.get_paginator("get_products")
259 for page in paginator.paginate(ServiceCode="AmazonRDS", Filters=acu_filters):
260 for item_json in page["PriceList"]:
261 item = json.loads(item_json) if isinstance(item_json, str) else item_json
262 attrs = item.get("product", {}).get("attributes", {})
263 usagetype = attrs.get("usagetype", "")
264 if "ServerlessV2Usage" in usagetype and "IOOptimized" not in usagetype:
265 terms = item.get("terms", {}).get("OnDemand", {})
266 for term in terms.values():
267 for dim in term.get("priceDimensions", {}).values():
268 price = float(dim.get("pricePerUnit", {}).get("USD", "0"))
269 if price > 0:
270 result["acu_standard"] = price
271 result["acu_io_optimized"] = round(price * IO_OPT_COMPUTE_MULTIPLIER, 4)
272
273 # Storage pricing
274 storage_filters = [
275 {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"},
276 {"Type": "TERM_MATCH", "Field": "location", "Value": location},
277 {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"},
278 {"Type": "TERM_MATCH", "Field": "productFamily", "Value": "Database Storage"},
279 ]
280 for page in paginator.paginate(ServiceCode="AmazonRDS", Filters=storage_filters):
281 for item_json in page["PriceList"]:
282 item = json.loads(item_json) if isinstance(item_json, str) else item_json
283 attrs = item.get("product", {}).get("attributes", {})
284 usagetype = attrs.get("usagetype", "")
285 terms = item.get("terms", {}).get("OnDemand", {})
286 for term in terms.values():
287 for dim in term.get("priceDimensions", {}).values():
288 price = float(dim.get("pricePerUnit", {}).get("USD", "0"))
289 if price <= 0:
290 continue
291 if "IOOptimized" in usagetype:
292 result["storage_io_optimized"] = price
293 elif "Aurora:StorageUsage" in usagetype:
294 result["storage_standard"] = price
295
296 return result
297
298
299def _fetch_instance_catalog(region: str) -> dict[str, tuple]:
300 """Discover Aurora instance types via RDS + EC2 APIs.
301
302 Returns dict: instance_type -> (vcpus, memory_gib, 0.0).
303 Price is 0.0 here; caller merges with pricing data.
304 """
305 import boto3
306
307 rds = boto3.client("rds", region_name=region)
308 instances = {}
309
310 for engine in ("aurora-mysql",):
311 try:
312 paginator = rds.get_paginator("describe_orderable_db_instance_options")
313 for page in paginator.paginate(Engine=engine):
314 for opt in page.get("OrderableDBInstanceOptions", []):
315 name = opt.get("DBInstanceClass", "")
316 if name.startswith("db.") and name != "db.serverless":
317 instances[name] = {"vcpus": 0, "memory_gib": 0.0}
318 except Exception:
319 pass
320
321 if not instances:
322 return {}
323
324 # Get vCPU/memory from EC2 describe_instance_types
325 ec2 = boto3.client("ec2", region_name=region)
326 ec2_names = [name.replace("db.", "", 1) for name in instances]
327
328 for i in range(0, len(ec2_names), 100):
329 batch = ec2_names[i : i + 100]
330 try:
331 resp = ec2.describe_instance_types(InstanceTypes=batch)
332 for it in resp.get("InstanceTypes", []):
333 db_name = "db." + it["InstanceType"]
334 if db_name in instances:
335 instances[db_name]["vcpus"] = it.get("VCpuInfo", {}).get("DefaultVCpus", 0)
336 instances[db_name]["memory_gib"] = (
337 it.get("MemoryInfo", {}).get("SizeInMiB", 0) / 1024
338 )
339 except Exception:
340 # Try one by one for families EC2 doesn't know about
341 for ec2_name in batch:
342 try:
343 resp = ec2.describe_instance_types(InstanceTypes=[ec2_name])
344 for it in resp.get("InstanceTypes", []):
345 db_name = "db." + it["InstanceType"]
346 if db_name in instances:
347 instances[db_name]["vcpus"] = it.get("VCpuInfo", {}).get(
348 "DefaultVCpus", 0
349 )
350 instances[db_name]["memory_gib"] = (
351 it.get("MemoryInfo", {}).get("SizeInMiB", 0) / 1024
352 )
353 except Exception:
354 pass
355
356 # Convert to tuple format, drop entries missing vCPU/memory
357 catalog = {}
358 for name, spec in instances.items():
359 if spec["vcpus"] > 0 and spec["memory_gib"] > 0:
360 catalog[name] = (spec["vcpus"], spec["memory_gib"], 0.0)
361
362 return catalog
363
364
365def refresh_pricing(region: str = "us-east-1") -> dict:
366 """Refresh all pricing and instance data from AWS APIs.
367
368 Tries live APIs first. On any failure, falls back to static defaults
369 and reports what succeeded and what didn't.
370
371 Returns a summary dict describing the pricing source.
372 """
373 global ACU_PRICE_STANDARD, ACU_PRICE_IO_OPTIMIZED
374 global STORAGE_STANDARD_PER_GIB, STORAGE_IO_OPT_PER_GIB
375 global INSTANCE_SPECS, _pricing_source
376
377 errors = []
378 live_instances = 0
379 live_prices = 0
380 live_acu = False
381
382 # 1. Try to fetch the instance catalog (RDS + EC2)
383 api_catalog = {}
384 try:
385 api_catalog = _fetch_instance_catalog(region)
386 live_instances = len(api_catalog)
387 except Exception as e:
388 errors.append(f"Instance catalog: {e}")
389
390 # 2. Try to fetch instance pricing (Pricing API, then public bulk CSV fallback)
391 instance_prices = {}
392 try:
393 instance_prices = _fetch_instance_pricing(region)
394 live_prices = len(instance_prices)
395 except Exception as e:
396 errors.append(f"Instance pricing (API): {e}")
397 # Fallback: public bulk pricing CSV (no IAM credentials needed)
398 try:
399 instance_prices = _fetch_instance_pricing_bulk(region)
400 live_prices = len(instance_prices)
401 if live_prices > 0:
402 errors[-1] += " [recovered via bulk pricing CSV]"
403 except Exception as e2:
404 errors.append(f"Instance pricing (bulk CSV): {e2}")
405
406 # 3. Try to fetch ACU + storage pricing
407 try:
408 acu_data = _fetch_acu_and_storage_pricing(region)
409 if "acu_standard" in acu_data:
410 ACU_PRICE_STANDARD = acu_data["acu_standard"]
411 ACU_PRICE_IO_OPTIMIZED = acu_data.get(
412 "acu_io_optimized",
413 round(acu_data["acu_standard"] * IO_OPT_COMPUTE_MULTIPLIER, 4),
414 )
415 live_acu = True
416 if "storage_standard" in acu_data:
417 STORAGE_STANDARD_PER_GIB = acu_data["storage_standard"]
418 if "storage_io_optimized" in acu_data:
419 STORAGE_IO_OPT_PER_GIB = acu_data["storage_io_optimized"]
420 except Exception as e:
421 errors.append(f"ACU/storage pricing: {e}")
422
423 # 4. Merge: start with static, overlay API catalog, overlay prices
424 merged = dict(_STATIC_INSTANCE_SPECS)
425
426 for name, (vcpus, mem, _) in api_catalog.items():
427 price = instance_prices.get(name, 0.0)
428 # If API didn't return a price, keep static price if we have one
429 if price == 0.0 and name in _STATIC_INSTANCE_SPECS:
430 price = _STATIC_INSTANCE_SPECS[name][2]
431 merged[name] = (vcpus, mem, price)
432
433 # For instances in static but not in API catalog, update price if available
434 for name in _STATIC_INSTANCE_SPECS:
435 if name not in api_catalog and name in instance_prices:
436 v, m, _ = _STATIC_INSTANCE_SPECS[name]
437 merged[name] = (v, m, instance_prices[name])
438
439 INSTANCE_SPECS = merged
440
441 # Determine source description
442 if not errors:
443 source = "live"
444 details = (
445 f"Live AWS APIs ({region}): {live_instances} instance types, "
446 f"{live_prices} prices, ACU=${ACU_PRICE_STANDARD}/hr"
447 )
448 elif live_prices > 0 or live_acu:
449 source = "partial_live"
450 details = (
451 f"Partial live data ({region}): {live_instances} instances, "
452 f"{live_prices} prices. Gaps filled from static defaults. "
453 f"Errors: {'; '.join(errors)}"
454 )
455 else:
456 # Full fallback
457 source = "static_fallback"
458 INSTANCE_SPECS = dict(_STATIC_INSTANCE_SPECS)
459 ACU_PRICE_STANDARD = _STATIC_ACU_PRICE_STANDARD
460 ACU_PRICE_IO_OPTIMIZED = _STATIC_ACU_PRICE_IO_OPTIMIZED
461 STORAGE_STANDARD_PER_GIB = _STATIC_STORAGE_STANDARD_PER_GIB
462 STORAGE_IO_OPT_PER_GIB = _STATIC_STORAGE_IO_OPT_PER_GIB
463 details = (
464 f"Static fallback (us-east-1, {_STATIC_LAST_UPDATED}). "
465 f"Live fetch failed: {'; '.join(errors)}"
466 )
467
468 _pricing_source = {
469 "source": source,
470 "region": region,
471 "last_updated": _STATIC_LAST_UPDATED if source == "static_fallback" else "now",
472 "details": details,
473 "instance_count": len(INSTANCE_SPECS),
474 "acu_price_standard": ACU_PRICE_STANDARD,
475 "storage_price_standard": STORAGE_STANDARD_PER_GIB,
476 }
477 if errors:
478 _pricing_source["errors"] = errors
479
480 return _pricing_source
481
482
483def get_pricing_source() -> dict:
484 """Return metadata about the active pricing data source."""
485 return dict(_pricing_source)
486
487
488# ---------------------------------------------------------------------------
489# Core calculation functions
490# ---------------------------------------------------------------------------
491
492
493def round_up_to_half(value: float) -> float:
494 """Round up to nearest 0.5 ACU."""
495 return math.ceil(value * 2) / 2
496
497
498def family_ratio(instance_type: str) -> int:
499 """Get ACU ratio for an instance family."""
500 m = re.match(r"db\.([a-z])", instance_type)
501 if m:
502 return ACU_FAMILY_RATIO.get(m.group(1), 4)
503 return 4
504
505
506def get_instance_specs(instance_type: str) -> tuple:
507 """Get (vcpus, memory_gib, price_per_hour) for an instance type."""
508 if instance_type in INSTANCE_SPECS:
509 return INSTANCE_SPECS[instance_type]
510 raise ValueError(
511 f"Unknown instance type: {instance_type}. "
512 f"Supported: {', '.join(sorted(INSTANCE_SPECS.keys()))}"
513 )
514
515
516def estimate_acu(
517 cpu_p95: float,
518 cpu_max: float,
519 vcpus: int,
520 instance_type: str,
521 cpu_avg: float = 0,
522) -> dict:
523 """Estimate ACU needed for a workload.
524
525 Returns dict with typical ACU, min/max recommendations, and breakdown.
526 """
527 ratio = family_ratio(instance_type)
528
529 # Typical ACU: weighted 95/5 blend
530 weighted_cpu = (cpu_p95 * 0.95 + cpu_max * 0.05) / 100
531 raw_typical = weighted_cpu * vcpus * ratio
532 typical_acu = round_up_to_half(raw_typical)
533 typical_acu = max(ACU_MIN, min(typical_acu, ACU_MAX))
534
535 # Peak ACU
536 raw_peak = (cpu_max / 100) * vcpus * ratio
537 peak_acu = round_up_to_half(raw_peak)
538
539 # Average ACU (for min recommendation)
540 if cpu_avg > 0:
541 avg_acu = round_up_to_half((cpu_avg / 100) * vcpus * ratio)
542 else:
543 # Estimate average as 60% of P95 when not provided
544 avg_acu = round_up_to_half((cpu_p95 * 0.6 / 100) * vcpus * ratio)
545
546 exceeds_capacity = raw_typical > ACU_MAX or raw_peak > ACU_MAX
547
548 return {
549 "typical_acu": typical_acu,
550 "peak_acu": peak_acu,
551 "avg_acu": avg_acu,
552 "raw_typical": round(raw_typical, 2),
553 "raw_peak": round(raw_peak, 2),
554 "family_ratio": ratio,
555 "exceeds_capacity": exceeds_capacity,
556 }
557
558
559def recommend_min_max(
560 acu_result: dict,
561 connections: int = 0,
562 working_set_gib: float = 0,
563) -> dict:
564 """Recommend min/max ACU settings."""
565 avg_acu = acu_result["avg_acu"]
566 typical_acu = acu_result["typical_acu"]
567 peak_acu = acu_result["peak_acu"]
568
569 # Connection floor
570 conn_mem_gib = connections * 10 / 1024 # ~10 MB per connection average
571 conn_acu = round_up_to_half(conn_mem_gib / GIB_PER_ACU)
572
573 # Memory floor (advisory — working set)
574 mem_acu = round_up_to_half(working_set_gib / GIB_PER_ACU) if working_set_gib > 0 else 0
575
576 # Min: based on avg CPU + connection floor (uncapped first, so we can detect
577 # a baseline that already exceeds serverless limits).
578 raw_min = max(ACU_MIN, avg_acu, conn_acu)
579
580 # Max: peak + 30% headroom, at least 1.5x typical
581 recommended_max = round_up_to_half(peak_acu * 1.3)
582 recommended_max = max(recommended_max, round_up_to_half(typical_acu * 1.5))
583 recommended_max = min(recommended_max, ACU_MAX)
584
585 # The workload's baseline doesn't fit a single serverless instance when the
586 # uncapped min exceeds the ACU ceiling or the (capped) max — flag it, mirroring
587 # estimate_acu's exceeds_capacity.
588 exceeds_capacity = raw_min > ACU_MAX or raw_min > recommended_max
589
590 # Cap min at ACU_MAX and enforce the invariant: min must never exceed max.
591 recommended_min = min(raw_min, ACU_MAX, recommended_max)
592
593 # Memory advisory
594 memory_advisory = None
595 if mem_acu > recommended_min:
596 memory_advisory = (
597 f"Working set needs {mem_acu} ACU ({working_set_gib:.1f} GiB / "
598 f"{GIB_PER_ACU} GiB per ACU). Your min ACU ({recommended_min}) is below this. "
599 f"Setting min to {mem_acu} ACU keeps the working set cached and avoids "
600 f"cold-cache I/O penalties on scale-up. Trade-off: higher baseline cost."
601 )
602
603 return {
604 "recommended_min": recommended_min,
605 "recommended_max": recommended_max,
606 "connection_floor_acu": conn_acu,
607 "memory_floor_acu": mem_acu,
608 "memory_advisory": memory_advisory,
609 "exceeds_capacity": exceeds_capacity,
610 }
611
612
613def calculate_costs(
614 typical_acu: float,
615 min_acu: float,
616 max_acu: float,
617 storage_gib: float,
618 provisioned_instance: str,
619 num_provisioned_instances: int = 1,
620 exceeds_capacity: bool = False,
621) -> dict:
622 """Calculate and compare serverless vs provisioned costs."""
623 _, _, price_per_hour = get_instance_specs(provisioned_instance)
624
625 # Provisioned cost
626 prov_compute = price_per_hour * HOURS_PER_MONTH * num_provisioned_instances
627 prov_storage = storage_gib * STORAGE_STANDARD_PER_GIB
628 prov_total = prov_compute + prov_storage
629
630 # Serverless cost (typical steady-state)
631 sv_compute = typical_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH
632 sv_storage = storage_gib * STORAGE_STANDARD_PER_GIB
633 sv_total = sv_compute + sv_storage
634
635 # Serverless cost range
636 sv_low = min_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH + sv_storage
637 sv_high = max_acu * ACU_PRICE_STANDARD * HOURS_PER_MONTH + sv_storage
638
639 # Savings
640 savings = prov_total - sv_total
641 savings_pct = (savings / prov_total * 100) if prov_total > 0 else 0
642
643 # Recommendation logic
644 if exceeds_capacity:
645 recommendation = "not_recommended"
646 reason = (
647 f"Workload's baseline/peak demand exceeds the {ACU_MAX:.0f} ACU serverless "
648 f"maximum. Stay with provisioned or split across multiple serverless clusters."
649 )
650 elif savings_pct > 10 and sv_high <= prov_total * 2:
651 recommendation = "recommended"
652 reason = (
653 f"Serverless saves ${savings:.0f}/mo ({savings_pct:.0f}%) vs provisioned. "
654 f"Cost range: ${sv_low:.0f}–${sv_high:.0f}/mo."
655 )
656 elif savings_pct > 10:
657 recommendation = "consider"
658 reason = (
659 f"Typical cost is lower (${sv_total:.0f} vs ${prov_total:.0f}/mo), but "
660 f"peak cost could reach ${sv_high:.0f}/mo. Variable workloads benefit; "
661 f"sustained peaks may not."
662 )
663 elif savings_pct > -5:
664 recommendation = "consider"
665 reason = (
666 f"Similar cost (${sv_total:.0f} vs ${prov_total:.0f}/mo). Choose serverless "
667 f"for auto-scaling and zero management overhead."
668 )
669 elif savings_pct > -30:
670 recommendation = "more_expensive"
671 reason = (
672 f"Serverless costs ${abs(savings):.0f}/mo more than provisioned "
673 f"(${sv_total:.0f} vs ${prov_total:.0f}/mo)."
674 )
675 else:
676 recommendation = "not_recommended"
677 reason = (
678 f"Serverless at ${sv_total:.0f}/mo is {abs(savings_pct):.0f}% more expensive "
679 f"than provisioned at ${prov_total:.0f}/mo. Sustained workloads are cheaper "
680 f"on provisioned instances."
681 )
682
683 return {
684 "provisioned": {
685 "instance_type": provisioned_instance,
686 "num_instances": num_provisioned_instances,
687 "compute_monthly": round(prov_compute, 2),
688 "storage_monthly": round(prov_storage, 2),
689 "total_monthly": round(prov_total, 2),
690 },
691 "serverless": {
692 "typical_acu": typical_acu,
693 "compute_monthly": round(sv_compute, 2),
694 "storage_monthly": round(sv_storage, 2),
695 "total_monthly": round(sv_total, 2),
696 "cost_range": {
697 "low": round(sv_low, 2),
698 "typical": round(sv_total, 2),
699 "high": round(sv_high, 2),
700 },
701 },
702 "savings_monthly": round(savings, 2),
703 "savings_pct": round(savings_pct, 1),
704 "recommendation": recommendation,
705 "reason": reason,
706 }
707
708
709def format_table(result: dict) -> str:
710 """Format result as a readable text table."""
711 lines = []
712 source = result.get("pricing_source", _pricing_source)
713 tag = source.get("source", "static_fallback").replace("_", " ").title()
714 lines.append("=" * 65)
715 lines.append(" Aurora Serverless v2 — ACU Estimate & Cost Comparison")
716 lines.append(f" Pricing: {tag} ({source.get('region', '?')})")
717 lines.append("=" * 65)
718
719 # ACU settings
720 acu = result["acu_settings"]
721 lines.append("")
722 lines.append(" ACU Configuration")
723 lines.append(" " + "-" * 45)
724 lines.append(f" Recommended Min ACU: {acu['recommended_min']:.1f}")
725 lines.append(f" Recommended Max ACU: {acu['recommended_max']:.1f}")
726 lines.append(f" Typical ACU: {acu['typical_acu']:.1f}")
727 lines.append(f" Peak ACU: {acu['peak_acu']:.1f}")
728 if acu.get("connection_floor_acu", 0) > 0:
729 lines.append(f" Connection floor: {acu['connection_floor_acu']:.1f} ACU")
730 if acu.get("memory_floor_acu", 0) > 0:
731 lines.append(f" Memory floor: {acu['memory_floor_acu']:.1f} ACU (advisory)")
732 if acu.get("memory_advisory"):
733 lines.append(f" NOTE: {acu['memory_advisory']}")
734
735 # Cost comparison
736 costs = result["cost_comparison"]
737 prov = costs["provisioned"]
738 sv = costs["serverless"]
739 lines.append("")
740 lines.append(" Monthly Cost Comparison")
741 lines.append(" " + "-" * 45)
742 lines.append(f" {'':30s} {'Provisioned':>14s} {'Serverless':>14s}")
743 lines.append(
744 f" {'Compute':30s} {'$'+str(prov['compute_monthly']):>14s} {'$'+str(sv['compute_monthly']):>14s}"
745 )
746 lines.append(
747 f" {'Storage':30s} {'$'+str(prov['storage_monthly']):>14s} {'$'+str(sv['storage_monthly']):>14s}"
748 )
749 lines.append(
750 f" {'Total':30s} {'$'+str(prov['total_monthly']):>14s} {'$'+str(sv['total_monthly']):>14s}"
751 )
752 lines.append("")
753 lines.append(
754 f" Serverless cost range: ${sv['cost_range']['low']:.0f} – ${sv['cost_range']['high']:.0f}/mo"
755 )
756 lines.append(f" Savings: ${costs['savings_monthly']:.0f}/mo ({costs['savings_pct']:.0f}%)")
757
758 # Recommendation
759 lines.append("")
760 lines.append(f" Recommendation: {costs['recommendation'].upper()}")
761 lines.append(f" {costs['reason']}")
762 lines.append("")
763 lines.append("=" * 65)
764
765 return "\n".join(lines)
766
767
768def run_estimate(args) -> dict:
769 """Run full estimation from CLI arguments."""
770 vcpus, memory_gib, price = get_instance_specs(args.instance)
771
772 acu_result = estimate_acu(
773 cpu_p95=args.cpu_p95,
774 cpu_max=args.cpu_max,
775 vcpus=vcpus,
776 instance_type=args.instance,
777 cpu_avg=args.cpu_avg,
778 )
779
780 min_max = recommend_min_max(
781 acu_result,
782 connections=args.connections,
783 working_set_gib=args.working_set,
784 )
785
786 # Workload overflows serverless if EITHER signal trips: estimate_acu's
787 # typical/peak check, or recommend_min_max's baseline-min check.
788 exceeds_capacity = acu_result["exceeds_capacity"] or min_max["exceeds_capacity"]
789
790 costs = calculate_costs(
791 typical_acu=acu_result["typical_acu"],
792 min_acu=min_max["recommended_min"],
793 max_acu=min_max["recommended_max"],
794 storage_gib=args.storage,
795 provisioned_instance=args.instance,
796 num_provisioned_instances=args.num_instances,
797 exceeds_capacity=exceeds_capacity,
798 )
799
800 return {
801 "input": {
802 "instance_type": args.instance,
803 "vcpus": vcpus,
804 "memory_gib": memory_gib,
805 "cpu_p95": args.cpu_p95,
806 "cpu_max": args.cpu_max,
807 "cpu_avg": args.cpu_avg,
808 "connections": args.connections,
809 "working_set_gib": args.working_set,
810 "storage_gib": args.storage,
811 "num_instances": args.num_instances,
812 },
813 "acu_settings": {
814 "recommended_min": min_max["recommended_min"],
815 "recommended_max": min_max["recommended_max"],
816 "typical_acu": acu_result["typical_acu"],
817 "peak_acu": acu_result["peak_acu"],
818 "avg_acu": acu_result["avg_acu"],
819 "connection_floor_acu": min_max["connection_floor_acu"],
820 "memory_floor_acu": min_max["memory_floor_acu"],
821 "memory_advisory": min_max["memory_advisory"],
822 "exceeds_capacity": exceeds_capacity,
823 },
824 "cost_comparison": costs,
825 "pricing_source": get_pricing_source(),
826 }
827
828
829def main():
830 # Shared flags accepted both before the subcommand and after it (so e.g.
831 # `... estimate --region X --offline` and `... --region X estimate` both work).
832 # Defaults are SUPPRESSed here so a subparser copy does NOT re-apply its own
833 # default and clobber a value the user passed before the subcommand; the real
834 # defaults are resolved once, after parsing, below.
835 common = argparse.ArgumentParser(add_help=False)
836 common.add_argument(
837 "--region",
838 default=argparse.SUPPRESS,
839 help="AWS region for pricing (default: us-east-1). "
840 "Live pricing requires boto3 + AWS credentials.",
841 )
842 common.add_argument(
843 "--offline",
844 action="store_true",
845 default=argparse.SUPPRESS,
846 help="Skip live API calls, use static fallback data only.",
847 )
848
849 parser = argparse.ArgumentParser(
850 description="Aurora Serverless v2 ACU Calculator", parents=[common]
851 )
852 sub = parser.add_subparsers(dest="command")
853
854 # estimate command
855 est = sub.add_parser("estimate", parents=[common], help="Estimate ACU sizing and compare costs")
856 est.add_argument(
857 "--instance", required=True, help="Current provisioned instance type (e.g., db.r6g.xlarge)"
858 )
859 est.add_argument("--cpu-p95", type=float, required=True, help="P95 CPU utilization (0-100)")
860 est.add_argument("--cpu-max", type=float, required=True, help="Maximum CPU utilization (0-100)")
861 est.add_argument(
862 "--cpu-avg",
863 type=float,
864 default=0,
865 help="Average CPU utilization (0-100), estimated if omitted",
866 )
867 est.add_argument("--connections", type=int, default=0, help="Peak connection count")
868 est.add_argument("--working-set", type=float, default=0, help="Working set size in GiB")
869 est.add_argument("--storage", type=float, default=10, help="Storage in GiB")
870 est.add_argument(
871 "--num-instances",
872 type=int,
873 default=1,
874 help="Number of provisioned instances (for cost comparison)",
875 )
876 est.add_argument("--format", choices=["json", "table"], default="json", help="Output format")
877
878 # list-instances command
879 sub.add_parser(
880 "list-instances",
881 parents=[common],
882 help="List supported instance types with specs and pricing",
883 )
884
885 # pricing-source command
886 sub.add_parser(
887 "pricing-source", parents=[common], help="Show where pricing data is coming from"
888 )
889
890 args = parser.parse_args()
891
892 # Resolve shared-flag defaults once (they were SUPPRESSed on both the main and
893 # subparsers so neither position clobbers the other; an explicit flag in either
894 # position lands in the namespace, otherwise we apply the default here).
895 if not hasattr(args, "region"):
896 args.region = "us-east-1"
897 if not hasattr(args, "offline"):
898 args.offline = False
899
900 # Refresh pricing (live or offline)
901 if not args.offline:
902 source = refresh_pricing(region=args.region)
903 if args.command != "pricing-source":
904 # Brief status line to stderr so it doesn't pollute JSON output
905 tag = (
906 "LIVE"
907 if source["source"] == "live"
908 else ("PARTIAL" if source["source"] == "partial_live" else "STATIC")
909 )
910 print(
911 f"[Pricing: {tag} — {source['region']}, "
912 f"{source['instance_count']} instances, "
913 f"ACU=${source['acu_price_standard']}/hr]",
914 file=sys.stderr,
915 )
916
917 if args.command == "estimate":
918 result = run_estimate(args)
919 if args.format == "table":
920 print(format_table(result))
921 else:
922 print(json.dumps(result, indent=2))
923
924 elif args.command == "list-instances":
925 source = get_pricing_source()
926 print(f"Pricing source: {source['source']} ({source['region']})")
927 print(f"{'Instance Type':<25s} {'vCPUs':>6s} {'Memory':>8s} {'$/hr':>8s} {'$/mo':>10s}")
928 print("-" * 62)
929 for name in sorted(INSTANCE_SPECS.keys()):
930 v, m, p = INSTANCE_SPECS[name]
931 print(f"{name:<25s} {v:>6d} {m:>6.0f} GiB {p:>8.3f} {p*730:>10.2f}")
932 print(f"\nTotal: {len(INSTANCE_SPECS)} instance types")
933
934 elif args.command == "pricing-source":
935 print(json.dumps(get_pricing_source(), indent=2))
936
937 else:
938 parser.print_help()
939
940
941if __name__ == "__main__":
942 main()