Setting the file. One moment. 10
Setup DevOps Agent
33
AWS Deployment
57.9
IO Optimized Instructions
def ri_adjusted_monthly
— line 419
This file
- Number
- 57.31
- Position
- 31 of 32
- Type
- Python
- Size
- 33 KB
- Lines
- 908
scripts/commitment_pricing_analyzer.py
Python·908 lines·33 KB
15
16from __future__ import annotations
17
18import argparse
19import json
20import re
21import sys
22from dataclasses import dataclass
23
24HOURS_PER_MONTH = 730
25
26# ---------------------------------------------------------------------------
27# Static fallback on-demand prices (us-east-1, Aurora MySQL Standard;
28# Aurora MySQL and PostgreSQL share these instance rates).
29# I/O-Optimized premium applied via multiplier.
30# ---------------------------------------------------------------------------
31IO_OPT_COMPUTE_MULTIPLIER = 1.30
32ACU_PRICE_STANDARD = 0.12 # $/ACU-Hr
33ACU_PRICE_IO_OPTIMIZED = 0.156
34
35_STATIC_INSTANCE_PRICES = {
36 "db.t3.medium": 0.082,
37 "db.t3.large": 0.164,
38 "db.t4g.medium": 0.073,
39 "db.t4g.large": 0.146,
40 "db.r5.large": 0.290,
41 "db.r5.xlarge": 0.580,
42 "db.r5.2xlarge": 1.160,
43 "db.r5.4xlarge": 2.320,
44 "db.r5.8xlarge": 4.640,
45 "db.r5.12xlarge": 6.960,
46 "db.r5.16xlarge": 9.280,
47 "db.r5.24xlarge": 13.920,
48 "db.r6g.large": 0.260,
49 "db.r6g.xlarge": 0.519,
50 "db.r6g.2xlarge": 1.038,
51 "db.r6g.4xlarge": 2.076,
52 "db.r6g.8xlarge": 4.152,
53 "db.r6g.12xlarge": 6.228,
54 "db.r6g.16xlarge": 8.304,
55 "db.r7g.large": 0.276,
56 "db.r7g.xlarge": 0.553,
57 "db.r7g.2xlarge": 1.106,
58 "db.r7g.4xlarge": 2.211,
59 "db.r7g.8xlarge": 4.422,
60 "db.r7g.12xlarge": 6.633,
61 "db.r7g.16xlarge": 8.844,
62 "db.r8g.large": 0.276,
63 "db.r8g.xlarge": 0.552,
64 "db.r8g.2xlarge": 1.104,
65 "db.r8g.4xlarge": 2.208,
66 "db.r8g.8xlarge": 4.416,
67 "db.r8g.12xlarge": 6.624,
68 "db.r8g.16xlarge": 8.832,
69 "db.r8g.24xlarge": 13.248,
70 "db.r8g.48xlarge": 26.496,
71}
72
73_REGION_NAMES = {
74 "us-east-1": "US East (N. Virginia)",
75 "us-east-2": "US East (Ohio)",
76 "us-west-1": "US West (N. California)",
77 "us-west-2": "US West (Oregon)",
78 "eu-west-1": "EU (Ireland)",
79 "eu-west-2": "EU (London)",
80 "eu-central-1": "EU (Frankfurt)",
81 "eu-north-1": "EU (Stockholm)",
82 "ap-southeast-1": "Asia Pacific (Singapore)",
83 "ap-southeast-2": "Asia Pacific (Sydney)",
84 "ap-northeast-1": "Asia Pacific (Tokyo)",
85 "ap-south-1": "Asia Pacific (Mumbai)",
86 "ca-central-1": "Canada (Central)",
87}
88
89# DSP only covers these families
90_DSP_ELIGIBLE_FAMILIES = {"r7g", "r7i", "r8g", "r8gd", "m7g", "m7i", "c7g", "c7i", "x8g"}
91
92_DSP_SIZE_MAP = {
93 "micro": "micro",
94 "small": "small",
95 "medium": "medium",
96 "large": "large",
97 "xl": "xlarge",
98 "2xl": "2xlarge",
99 "4xl": "4xlarge",
100 "8xl": "8xlarge",
101 "12xl": "12xlarge",
102 "16xl": "16xlarge",
103 "24xl": "24xlarge",
104 "48xl": "48xlarge",
105}
106
107
108# ---------------------------------------------------------------------------
109# Data classes
110# ---------------------------------------------------------------------------
111
112
113@dataclass
114class RIOffering:
115 instance_type: str
116 term_years: int
117 payment_option: str # "No Upfront" | "Partial Upfront" | "All Upfront"
118 effective_hourly: float # (upfront / term_hours) + recurring
119 upfront_cost: float
120 recurring_hourly: float
121
122 def monthly_cost(self) -> float:
123 return self.effective_hourly * HOURS_PER_MONTH
124
125
126@dataclass
127class DSPRate:
128 usage_type: str # instance type or "ServerlessV2"
129 term_years: int # always 1 for Aurora DSP
130 payment_option: str
131 rate_per_hour: float
132
133 def monthly_cost(self) -> float:
134 return self.rate_per_hour * HOURS_PER_MONTH
135
136
137# ---------------------------------------------------------------------------
138# Live AWS fetchers
139# ---------------------------------------------------------------------------
140
141
142def _family_from_instance(instance_type: str) -> str:
143 m = re.match(r"db\.([a-z0-9]+)\.", instance_type)
144 return m.group(1) if m else ""
145
146
147def get_on_demand_price(instance_type: str, region: str = "us-east-1") -> float:
148 """Return on-demand hourly price. Tries Pricing API, falls back to static."""
149 try:
150 import boto3
151 except ImportError:
152 return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)
153
154 location = _REGION_NAMES.get(region)
155 if not location:
156 return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)
157
158 try:
159 pricing = boto3.client("pricing", region_name="us-east-1")
160 filters = [
161 {"Type": "TERM_MATCH", "Field": "databaseEngine", "Value": "Aurora MySQL"},
162 {"Type": "TERM_MATCH", "Field": "location", "Value": location},
163 {"Type": "TERM_MATCH", "Field": "instanceType", "Value": instance_type},
164 {"Type": "TERM_MATCH", "Field": "deploymentOption", "Value": "Single-AZ"},
165 {"Type": "TERM_MATCH", "Field": "termType", "Value": "OnDemand"},
166 ]
167 for page in pricing.get_paginator("get_products").paginate(
168 ServiceCode="AmazonRDS", Filters=filters
169 ):
170 for item_json in page["PriceList"]:
171 item = json.loads(item_json) if isinstance(item_json, str) else item_json
172 attrs = item.get("product", {}).get("attributes", {})
173 if "IOOptimized" in attrs.get("usagetype", ""):
174 continue
175 for term in item.get("terms", {}).get("OnDemand", {}).values():
176 for dim in term.get("priceDimensions", {}).values():
177 try:
178 price = float(dim.get("pricePerUnit", {}).get("USD", "0"))
179 except (ValueError, TypeError):
180 continue
181 if price > 0:
182 return price
183 except Exception:
184 pass
185
186 return _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)
187
188
189def fetch_ri_offerings(instance_type: str, region: str) -> list[RIOffering]:
190 """Fetch all RI offerings for an instance type. Returns [] on failure."""
191 try:
192 import boto3
193 except ImportError:
194 return []
195
196 results: list[RIOffering] = []
197 try:
198 rds = boto3.client("rds", region_name=region)
199 for engine in ("aurora-mysql",):
200 try:
201 paginator = rds.get_paginator("describe_reserved_db_instances_offerings")
202 for page in paginator.paginate(
203 DBInstanceClass=instance_type,
204 ProductDescription=engine,
205 MultiAZ=False,
206 ):
207 for offering in page.get("ReservedDBInstancesOfferings", []):
208 inst = offering.get("DBInstanceClass", "")
209 if inst != instance_type:
210 continue
211 duration = offering.get("Duration", 0)
212 term_years = 3 if duration > 94_000_000 else 1
213 payment = offering.get("OfferingType", "")
214 fixed = float(offering.get("FixedPrice", 0.0))
215 recurring_list = offering.get("RecurringCharges", [])
216 recurring_hr = sum(
217 float(rc.get("RecurringChargeAmount", 0.0)) for rc in recurring_list
218 )
219 term_hours = term_years * 365 * 24
220 effective = (fixed / term_hours) + recurring_hr
221 results.append(
222 RIOffering(
223 instance_type=inst,
224 term_years=term_years,
225 payment_option=payment,
226 effective_hourly=round(effective, 6),
227 upfront_cost=round(fixed, 2),
228 recurring_hourly=round(recurring_hr, 6),
229 )
230 )
231 except Exception:
232 continue
233 except Exception:
234 return []
235
236 # Deduplicate (same offering exists for both engines)
237 seen = set()
238 deduped = []
239 for r in results:
240 key = (r.term_years, r.payment_option, round(r.effective_hourly, 6))
241 if key in seen:
242 continue
243 seen.add(key)
244 deduped.append(r)
245 return deduped
246
247
248def fetch_dsp_rates(region: str) -> dict[str, list[DSPRate]]:
249 """Fetch Database Savings Plan rates for Aurora in the region.
250
251 Returns dict mapping usage key (instance type or 'ServerlessV2') to rates.
252 """
253 try:
254 import boto3
255 except ImportError:
256 return {}
257
258 result: dict[str, list[DSPRate]] = {}
259 try:
260 sp = boto3.client("savingsplans", region_name="us-east-1")
261 except Exception:
262 return {}
263
264 for engine in ("Aurora MySQL",):
265 try:
266 rates = []
267 token = None
268 while True:
269 kwargs = {
270 "savingsPlanTypes": ["Database"],
271 "products": ["RDS"],
272 "serviceCodes": ["AmazonRDS"],
273 "filters": [
274 {"name": "region", "values": [region]},
275 {"name": "productDescription", "values": [engine]},
276 ],
277 "maxResults": 1000,
278 }
279 if token:
280 kwargs["nextToken"] = token
281 resp = sp.describe_savings_plans_offering_rates(**kwargs)
282 rates.extend(resp.get("searchResults", []))
283 token = resp.get("nextToken")
284 if not token:
285 break
286
287 for rate_entry in rates:
288 offering = rate_entry.get("savingsPlanOffering", {})
289 dur = offering.get("durationSeconds", 0)
290 term_years = 3 if dur > 94_000_000 else 1
291 payment = offering.get("paymentOption", "")
292 try:
293 rate_val = float(rate_entry.get("rate", "0"))
294 except (ValueError, TypeError):
295 continue
296 if rate_val <= 0:
297 continue
298
299 usage = rate_entry.get("usageType", "")
300 unit = rate_entry.get("unit", "")
301
302 # Skip I/O-Optimized variants for consistency; main pricing uses Standard
303 if "IOOptimized" in usage:
304 continue
305
306 if unit == "ACU-Hr" and "ServerlessV2" in usage:
307 key = "ServerlessV2"
308 else:
309 m = re.match(r"InstanceUsage:db\.(\w+)\.(\w+)", usage)
310 if not m:
311 continue
312 family = m.group(1)
313 short_size = m.group(2)
314 size = _DSP_SIZE_MAP.get(short_size, short_size)
315 key = f"db.{family}.{size}"
316
317 entry = DSPRate(
318 usage_type=key,
319 term_years=term_years,
320 payment_option=payment,
321 rate_per_hour=round(rate_val, 6),
322 )
323 existing = result.get(key, [])
324 if not any(
325 e.term_years == entry.term_years and e.payment_option == entry.payment_option
326 for e in existing
327 ):
328 result.setdefault(key, []).append(entry)
329 except Exception:
330 continue
331
332 return result
333
334
335# ---------------------------------------------------------------------------
336# Best-of selection (lowest effective monthly cost per term/category)
337# ---------------------------------------------------------------------------
338
339
340def best_ri(offerings: list[RIOffering], term_years: int) -> RIOffering | None:
341 candidates = [r for r in offerings if r.term_years == term_years]
342 if not candidates:
343 return None
344 return min(candidates, key=lambda r: r.effective_hourly)
345
346
347def best_dsp(rates: list[DSPRate], term_years: int = 1) -> DSPRate | None:
348 candidates = [r for r in rates if r.term_years == term_years]
349 if not candidates:
350 return None
351 return min(candidates, key=lambda r: r.rate_per_hour)
352
353
354# ---------------------------------------------------------------------------
355# Comparison builder — single workload
356# ---------------------------------------------------------------------------
357
358
359def build_comparison(
360 instance_type: str,
361 num_instances: int,
362 region: str,
363 io_optimized: bool = False,
364 is_serverless: bool = False,
365 avg_acu: float = 0.0,
366 dsp_rates: dict[str, list[DSPRate]] | None = None,
367) -> dict:
368 """Compare on-demand, RI, and DSP for a single workload description."""
369 if dsp_rates is None:
370 dsp_rates = fetch_dsp_rates(region)
371
372 if is_serverless:
373 od_hourly = ACU_PRICE_IO_OPTIMIZED if io_optimized else ACU_PRICE_STANDARD
374 # For Serverless v2, "number of instances" is irrelevant — we price avg ACU continuously
375 units = avg_acu
376 od_monthly = od_hourly * units * HOURS_PER_MONTH
377
378 dsp_entry = best_dsp(dsp_rates.get("ServerlessV2", []))
379 dsp_monthly = dsp_entry.rate_per_hour * units * HOURS_PER_MONTH if dsp_entry else None
380 dsp_savings = (od_monthly - dsp_monthly) if dsp_monthly is not None else None
381
382 return {
383 "workload_type": "serverless_v2",
384 "avg_acu": avg_acu,
385 "io_optimized": io_optimized,
386 "on_demand": {
387 "hourly": od_hourly,
388 "monthly": round(od_monthly, 2),
389 },
390 "ri_1yr": None,
391 "ri_3yr": None,
392 "dsp_1yr": _format_dsp(dsp_entry, units, od_monthly) if dsp_entry else None,
393 "recommendation": _recommend_serverless(dsp_savings, od_monthly),
394 "notes": [
395 "Reserved Instances do not apply to Aurora Serverless v2.",
396 "DSP covers ACU-hours but bills the committed $/hr continuously, "
397 "even during auto-pause. Only commit to the steady baseline ACU.",
398 ],
399 }
400
401 # Provisioned
402 family = _family_from_instance(instance_type)
403 od_hourly = get_on_demand_price(instance_type, region)
404 if io_optimized:
405 od_hourly *= IO_OPT_COMPUTE_MULTIPLIER
406 od_monthly = od_hourly * HOURS_PER_MONTH * num_instances
407
408 ri_offerings = fetch_ri_offerings(instance_type, region)
409 ri_1yr = best_ri(ri_offerings, 1)
410 ri_3yr = best_ri(ri_offerings, 3)
411
412 # I/O-Optimized RI coverage (AWS Compute Optimizer, verified): an I/O-Optimized
413 # instance is FULLY covered by Reserved Instances — no portion is forced to
414 # on-demand — but it "consumes 30% more normalized units per hour than Aurora
415 # Standard", i.e. it draws down RI capacity at 1.30×. So the effective RI cost is
416 # the (Standard-normalized) RI rate × 1.30. Equivalently: buy ~30% more normalized
417 # RI units to cover the same I/O-Optimized fleet.
418 # io-opt RI monthly = ri_rate × 1.30 × hours × N
419 def ri_adjusted_monthly(ri: RIOffering | None) -> float | None:
420 if ri is None:
421 return None
422 units = IO_OPT_COMPUTE_MULTIPLIER if io_optimized else 1.0
423 return ri.effective_hourly * units * HOURS_PER_MONTH * num_instances
424
425 ri_1yr_monthly = ri_adjusted_monthly(ri_1yr)
426 ri_3yr_monthly = ri_adjusted_monthly(ri_3yr)
427
428 dsp_entry = best_dsp(dsp_rates.get(instance_type, []))
429 dsp_monthly = dsp_entry.rate_per_hour * HOURS_PER_MONTH * num_instances if dsp_entry else None
430
431 dsp_eligible = family in _DSP_ELIGIBLE_FAMILIES
432 notes = []
433 if not dsp_eligible:
434 notes.append(
435 f"Database Savings Plans do not cover the {family} family. "
436 f"DSP requires r7g, r8g, r7i, or newer-gen Aurora-compatible families."
437 )
438 if io_optimized:
439 notes.append(
440 "Cluster is I/O-Optimized (30% compute premium). Both RI and DSP cover the "
441 "full I/O-Optimized instance price — I/O-Optimized consumes 30% more "
442 "normalized units per hour, so an RI draws down capacity at 1.30× (buy ~30% "
443 "more normalized RI units to fully cover the fleet); no portion is on-demand."
444 )
445
446 return {
447 "workload_type": "provisioned",
448 "instance_type": instance_type,
449 "num_instances": num_instances,
450 "io_optimized": io_optimized,
451 "on_demand": {
452 "hourly": round(od_hourly, 4),
453 "monthly": round(od_monthly, 2),
454 },
455 "ri_1yr": _format_ri(ri_1yr, ri_1yr_monthly, od_monthly, num_instances),
456 "ri_3yr": _format_ri(ri_3yr, ri_3yr_monthly, od_monthly, num_instances),
457 "dsp_1yr": (_format_dsp(dsp_entry, num_instances, od_monthly) if dsp_entry else None),
458 "recommendation": _recommend_provisioned(
459 od_monthly,
460 ri_1yr_monthly,
461 ri_3yr_monthly,
462 dsp_monthly,
463 dsp_eligible=dsp_eligible,
464 io_optimized=io_optimized,
465 ),
466 "notes": notes,
467 }
468
469
470def _format_ri(
471 ri: RIOffering | None, monthly: float | None, od_monthly: float, n: int
472) -> dict | None:
473 if ri is None or monthly is None:
474 return None
475 savings = od_monthly - monthly
476 pct = (savings / od_monthly * 100) if od_monthly > 0 else 0
477 return {
478 "term_years": ri.term_years,
479 "payment_option": ri.payment_option,
480 "effective_hourly_per_instance": round(ri.effective_hourly, 4),
481 "upfront_total": round(ri.upfront_cost * n, 2),
482 "monthly": round(monthly, 2),
483 "savings_monthly": round(savings, 2),
484 "savings_pct": round(pct, 1),
485 }
486
487
488def _format_dsp(dsp: DSPRate | None, units: float, od_monthly: float) -> dict | None:
489 if dsp is None:
490 return None
491 monthly = dsp.rate_per_hour * HOURS_PER_MONTH * units
492 savings = od_monthly - monthly
493 pct = (savings / od_monthly * 100) if od_monthly > 0 else 0
494 return {
495 "term_years": dsp.term_years,
496 "payment_option": dsp.payment_option,
497 "rate_per_hour": round(dsp.rate_per_hour, 4),
498 "monthly": round(monthly, 2),
499 "savings_monthly": round(savings, 2),
500 "savings_pct": round(pct, 1),
501 }
502
503
504def _recommend_provisioned(
505 od: float,
506 ri_1yr: float | None,
507 ri_3yr: float | None,
508 dsp: float | None,
509 dsp_eligible: bool,
510 io_optimized: bool,
511) -> dict:
512 options = []
513 if ri_1yr is not None:
514 options.append(("1yr RI", ri_1yr))
515 if ri_3yr is not None:
516 options.append(("3yr RI", ri_3yr))
517 if dsp is not None:
518 options.append(("1yr DSP", dsp))
519
520 if not options:
521 return {
522 "best_option": "on_demand",
523 "reason": "No RI or DSP offerings available for this instance type/region.",
524 }
525
526 best_label, best_cost = min(options, key=lambda x: x[1])
527 savings = od - best_cost
528 pct = (savings / od * 100) if od > 0 else 0
529
530 reasons = [
531 f"{best_label} is the lowest-cost option, saving ${savings:.0f}/mo ({pct:.0f}%) vs on-demand."
532 ]
533 if best_label == "3yr RI":
534 reasons.append(
535 "Best fit for steady 24/7 workloads you're confident will stay on this instance family for 3 years."
536 )
537 elif best_label == "1yr DSP":
538 reasons.append(
539 "Offers flexibility — covers any eligible Aurora instance family in the account, including future upgrades."
540 )
541 if io_optimized:
542 reasons.append(
543 "DSP is particularly attractive for I/O-Optimized clusters since it covers the full rate."
544 )
545 elif best_label == "1yr RI":
546 reasons.append(
547 "Shorter commitment than 3yr, useful when instance-family migration is on the horizon."
548 )
549
550 if not dsp_eligible and dsp is None:
551 reasons.append(
552 "DSP is not available for this instance family, so RI is the only commitment option."
553 )
554
555 return {
556 "best_option": best_label,
557 "best_monthly_cost": round(best_cost, 2),
558 "savings_vs_on_demand": round(savings, 2),
559 "savings_pct": round(pct, 1),
560 "reason": " ".join(reasons),
561 }
562
563
564def _recommend_serverless(dsp_savings: float | None, od_monthly: float) -> dict:
565 if dsp_savings is None:
566 return {
567 "best_option": "on_demand",
568 "reason": "No Database Savings Plan rates available for Serverless v2 ACU in this region.",
569 }
570 if dsp_savings <= 0:
571 return {
572 "best_option": "on_demand",
573 "reason": "DSP would not save money at the specified average ACU.",
574 }
575 pct = (dsp_savings / od_monthly * 100) if od_monthly > 0 else 0
576 return {
577 "best_option": "1yr DSP",
578 "savings_vs_on_demand": round(dsp_savings, 2),
579 "savings_pct": round(pct, 1),
580 "reason": (
581 f"1yr DSP saves ${dsp_savings:.0f}/mo ({pct:.0f}%). "
582 "Size the commitment to your steady baseline ACU — DSP bills the committed $/hr "
583 "continuously, even during idle periods."
584 ),
585 }
586
587
588# ---------------------------------------------------------------------------
589# Live cluster analysis
590# ---------------------------------------------------------------------------
591
592
593def _is_empty_cluster(cluster: dict) -> bool:
594 """Skip clusters with no compute to analyze.
595
596 An Aurora cluster with no DB instances has no compute cost, so RI and DSP
597 commitment analysis doesn't apply. Typical cases: the last writer/reader
598 instance was deleted, paused/stopped clusters, or clusters mid-migration.
599 """
600 return len(cluster.get("DBClusterMembers", [])) == 0
601
602
603def analyze_cluster_live(cluster_id: str, region: str) -> dict:
604 import boto3
605
606 rds = boto3.client("rds", region_name=region)
607
608 try:
609 resp = rds.describe_db_clusters(DBClusterIdentifier=cluster_id)
610 except Exception as e:
611 return {"cluster_id": cluster_id, "error": str(e)}
612 clusters = resp.get("DBClusters", [])
613 if not clusters:
614 return {"cluster_id": cluster_id, "error": "cluster not found"}
615 cluster = clusters[0]
616 storage_type = cluster.get("StorageType", "aurora")
617 io_optimized = storage_type == "aurora-iopt1"
618 engine = cluster.get("Engine", "")
619
620 # Guardrail: skip clusters with no DB instances (last instance deleted, paused, mid-migration, etc.)
621 if _is_empty_cluster(cluster):
622 return {
623 "cluster_id": cluster_id,
624 "engine": engine,
625 "engine_version": cluster.get("EngineVersion", ""),
626 "storage_type": storage_type,
627 "skipped": True,
628 "reason": (
629 "Cluster has no DB instances — no compute to price. "
630 "RI and DSP commitment analysis does not apply. "
631 "This typically indicates the last instance was deleted, a paused "
632 "cluster, or a cluster mid-migration."
633 ),
634 }
635
636 # Identify instances
637 member_ids = [m["DBInstanceIdentifier"] for m in cluster.get("DBClusterMembers", [])]
638 type_counts: dict[str, int] = {}
639 serverless_instances = 0
640 for mid in member_ids:
641 try:
642 iresp = rds.describe_db_instances(DBInstanceIdentifier=mid)
643 for inst in iresp.get("DBInstances", []):
644 itype = inst.get("DBInstanceClass", "")
645 if itype == "db.serverless":
646 serverless_instances += 1
647 else:
648 type_counts[itype] = type_counts.get(itype, 0) + 1
649 except Exception:
650 continue
651
652 # Prefetch DSP rates once for this cluster analysis
653 dsp_rates = fetch_dsp_rates(region)
654
655 sub_workloads = []
656 for itype, count in type_counts.items():
657 sub_workloads.append(
658 build_comparison(
659 instance_type=itype,
660 num_instances=count,
661 region=region,
662 io_optimized=io_optimized,
663 dsp_rates=dsp_rates,
664 )
665 )
666 if serverless_instances > 0:
667 # Without observed ACU metrics, we can't price serverless exactly — note it
668 sub_workloads.append(
669 {
670 "workload_type": "serverless_v2",
671 "instance_count": serverless_instances,
672 "note": "Serverless v2 instances detected. Re-run with 'offline --serverless "
673 "--avg-acu <N>' using your observed average ACU (from CloudWatch "
674 "ServerlessDatabaseCapacity metric) for a precise DSP estimate.",
675 "io_optimized": io_optimized,
676 }
677 )
678
679 return {
680 "cluster_id": cluster_id,
681 "engine": engine,
682 "storage_type": storage_type,
683 "io_optimized": io_optimized,
684 "instance_mix": {**type_counts, "serverless": serverless_instances},
685 "workloads": sub_workloads,
686 }
687
688
689def list_clusters(region: str) -> list[str]:
690 import boto3
691
692 rds = boto3.client("rds", region_name=region)
693 names = []
694 for page in rds.get_paginator("describe_db_clusters").paginate():
695 for c in page.get("DBClusters", []):
696 if c.get("Engine", "").startswith("aurora"):
697 names.append(c["DBClusterIdentifier"])
698 return names
699
700
701# ---------------------------------------------------------------------------
702# Output formatting
703# ---------------------------------------------------------------------------
704
705
706def _format_table_single(result: dict) -> str:
707 lines = []
708 lines.append("=" * 72)
709 if result.get("workload_type") == "serverless_v2":
710 lines.append(f"Aurora Serverless v2 Commitment Pricing")
711 lines.append(
712 f" Avg ACU: {result.get('avg_acu', 0):.1f} "
713 f"I/O-Optimized: {result.get('io_optimized', False)}"
714 )
715 else:
716 lines.append(
717 f"Aurora Commitment Pricing — "
718 f"{result.get('num_instances', 1)}× {result.get('instance_type', '?')}"
719 )
720 if result.get("io_optimized"):
721 lines.append(" Storage: I/O-Optimized (30% compute premium applied)")
722 lines.append("=" * 72)
723 od_monthly = result["on_demand"]["monthly"]
724 lines.append("")
725 lines.append(f" {'Option':<28} {'Monthly':>12} {'Savings':>12} {'Upfront':>12} Term")
726 lines.append(" " + "-" * 70)
727 lines.append(f" {'On-Demand':<28} ${od_monthly:>11,.0f} {'—':>12} {'$0':>12} —")
728
729 for key, label, term_hint in (
730 ("ri_1yr", "1yr RI", "1 year"),
731 ("ri_3yr", "3yr RI", "3 years"),
732 ("dsp_1yr", "1yr DSP", "1 year"),
733 ):
734 entry = result.get(key)
735 if not entry:
736 continue
737 payment = entry.get("payment_option", "")
738 display = f"{label} ({payment})" if payment else label
739 monthly = entry.get("monthly", 0)
740 savings = entry.get("savings_monthly", 0)
741 pct = entry.get("savings_pct", 0)
742 upfront = entry.get("upfront_total", 0)
743 savings_str = f"${savings:,.0f} ({pct:.0f}%)" if savings else "—"
744 upfront_str = f"${upfront:,.0f}" if upfront else "$0"
745 lines.append(
746 f" {display:<28} ${monthly:>11,.0f} {savings_str:>12} {upfront_str:>12} {term_hint}"
747 )
748
749 rec = result.get("recommendation", {})
750 lines.append("")
751 lines.append(f" Recommendation: {rec.get('best_option', '?')}")
752 lines.append(f" {rec.get('reason', '')}")
753
754 notes = result.get("notes", [])
755 if notes:
756 lines.append("")
757 for n in notes:
758 lines.append(f" Note: {n}")
759 lines.append("=" * 72)
760 return "\n".join(lines)
761
762
763def _format_cluster(result: dict) -> str:
764 lines = []
765 lines.append(
766 f"Cluster: {result.get('cluster_id', '?')} "
767 f"({result.get('engine', '?')}) "
768 f"storage_type={result.get('storage_type', '?')}"
769 )
770 workloads = result.get("workloads", [])
771 for wl in workloads:
772 if wl.get("workload_type") == "serverless_v2" and "note" in wl:
773 lines.append("")
774 lines.append(f" [Serverless v2 — {wl.get('instance_count', 0)} instance(s)]")
775 lines.append(f" {wl['note']}")
776 continue
777 lines.append("")
778 lines.append(_format_table_single(wl))
779 return "\n".join(lines)
780
781
782def _format_fleet(output: dict) -> str:
783 lines = []
784 lines.append(f"Region: {output['region']}")
785 lines.append("")
786 total_od = 0.0
787 total_best = 0.0
788 for cluster in output["clusters"]:
789 if "error" in cluster:
790 lines.append(f"{cluster['cluster_id']}: ERROR {cluster['error']}")
791 continue
792 for wl in cluster.get("workloads", []):
793 if wl.get("workload_type") != "provisioned":
794 continue
795 od = wl["on_demand"]["monthly"]
796 best = wl.get("recommendation", {}).get("best_monthly_cost", od)
797 total_od += od
798 total_best += best
799 lines.append(f" Fleet monthly on-demand: ${total_od:,.0f}")
800 lines.append(f" With best commitments: ${total_best:,.0f}")
801 savings = total_od - total_best
802 pct = (savings / total_od * 100) if total_od > 0 else 0
803 lines.append(f" Fleet savings opportunity: ${savings:,.0f}/mo ({pct:.0f}%)")
804 lines.append("")
805 for cluster in output["clusters"]:
806 if "error" in cluster:
807 continue
808 lines.append("")
809 lines.append(_format_cluster(cluster))
810 return "\n".join(lines)
811
812
813# ---------------------------------------------------------------------------
814# CLI
815# ---------------------------------------------------------------------------
816
817
818def main():
819 parser = argparse.ArgumentParser(
820 description="Aurora RI & Database Savings Plan estimator (read-only)"
821 )
822 parser.add_argument("--region", default="us-east-1")
823 parser.add_argument("--format", choices=["json", "table"], default="json")
824 parser.add_argument("--cluster", help="Analyze a single cluster by identifier")
825 parser.add_argument(
826 "--all", action="store_true", help="Analyze all Aurora clusters in the region"
827 )
828
829 sub = parser.add_subparsers(dest="mode")
830 off = sub.add_parser("offline", help="Use user-supplied workload description")
831 off.add_argument("--instance", help="Instance type (e.g., db.r7g.2xlarge)")
832 off.add_argument("--num-instances", type=int, default=1)
833 off.add_argument(
834 "--io-optimized", action="store_true", help="Workload uses Aurora I/O-Optimized storage"
835 )
836 off.add_argument(
837 "--serverless", action="store_true", help="Serverless v2 workload — requires --avg-acu"
838 )
839 off.add_argument(
840 "--avg-acu", type=float, default=0.0, help="Average ACU for serverless workload"
841 )
842 off.add_argument("--region", default="us-east-1")
843 off.add_argument("--format", choices=["json", "table"], default="json")
844
845 args = parser.parse_args()
846
847 if args.mode == "offline":
848 if args.serverless:
849 if args.avg_acu <= 0:
850 print("ERROR: --serverless requires --avg-acu > 0", file=sys.stderr)
851 sys.exit(2)
852 result = build_comparison(
853 instance_type="",
854 num_instances=0,
855 region=args.region,
856 io_optimized=args.io_optimized,
857 is_serverless=True,
858 avg_acu=args.avg_acu,
859 )
860 else:
861 if not args.instance:
862 print("ERROR: offline mode requires --instance (or --serverless)", file=sys.stderr)
863 sys.exit(2)
864 result = build_comparison(
865 instance_type=args.instance,
866 num_instances=args.num_instances,
867 region=args.region,
868 io_optimized=args.io_optimized,
869 )
870 if args.format == "json":
871 print(json.dumps(result, indent=2, default=str))
872 else:
873 print(_format_table_single(result))
874 return
875
876 # Live modes require boto3
877 try:
878 import boto3 # noqa: F401
879 except ImportError:
880 print(
881 "ERROR: boto3 required for live AWS analysis. Use the 'offline' subcommand.",
882 file=sys.stderr,
883 )
884 sys.exit(2)
885
886 if args.all:
887 cluster_ids = list_clusters(args.region)
888 results = [analyze_cluster_live(cid, args.region) for cid in cluster_ids]
889 output = {"region": args.region, "cluster_count": len(results), "clusters": results}
890 if args.format == "json":
891 print(json.dumps(output, indent=2, default=str))
892 else:
893 print(_format_fleet(output))
894 return
895
896 if args.cluster:
897 result = analyze_cluster_live(args.cluster, args.region)
898 if args.format == "json":
899 print(json.dumps(result, indent=2, default=str))
900 else:
901 print(_format_cluster(result))
902 return
903
904 parser.print_help()
905
906
907if __name__ == "__main__":
908 main()