Setting the file. One moment. RDS Commitment Pricing Analyzer
70
Creating Amazon Aurora Db Cluster With Instances
93
Routing Traffic With Route53 And CloudFront
Resilience Program Design
Creating API Gateway Stage
def _fmt
— line 327
This file
- Number
- 74.18
- Position
- 18 of 18
- Type
- Python
- Size
- 16 KB
- Lines
- 467
scripts/rds_commitment_pricing_analyzer.py
Python·467 lines·16 KB
from
__future__
import
annotations
16
17import argparse
18import json
19import re
20import sys
21from dataclasses import dataclass
22
23HOURS_PER_MONTH = 730
24
25ENGINE_PRODUCT_MAP = {
26 "mysql": "MySQL",
27 "mariadb": "MariaDB",
28 "postgres": "PostgreSQL",
29}
30
31_STATIC_INSTANCE_PRICES = {
32 "db.t3.medium": 0.068,
33 "db.t3.large": 0.136,
34 "db.t4g.medium": 0.065,
35 "db.t4g.large": 0.129,
36 "db.m5.large": 0.171,
37 "db.m5.xlarge": 0.342,
38 "db.m5.2xlarge": 0.684,
39 "db.m6g.large": 0.154,
40 "db.m6g.xlarge": 0.308,
41 "db.m6g.2xlarge": 0.616,
42 "db.m7g.large": 0.162,
43 "db.m7g.xlarge": 0.324,
44 "db.m7g.2xlarge": 0.648,
45 "db.r5.large": 0.240,
46 "db.r5.xlarge": 0.480,
47 "db.r5.2xlarge": 0.960,
48 "db.r5.4xlarge": 1.920,
49 "db.r5.8xlarge": 3.840,
50 "db.r6g.large": 0.218,
51 "db.r6g.xlarge": 0.435,
52 "db.r6g.2xlarge": 0.870,
53 "db.r6g.4xlarge": 1.740,
54 "db.r6g.8xlarge": 3.480,
55 "db.r7g.large": 0.228,
56 "db.r7g.xlarge": 0.456,
57 "db.r7g.2xlarge": 0.912,
58 "db.r7g.4xlarge": 1.824,
59 "db.r7g.8xlarge": 3.648,
60 "db.r8g.large": 0.240,
61 "db.r8g.xlarge": 0.480,
62 "db.r8g.2xlarge": 0.960,
63 "db.r8g.4xlarge": 1.920,
64 "db.r8g.8xlarge": 3.840,
65}
66
67MULTI_AZ_MULTIPLIER = 2.0
68_DSP_ELIGIBLE_FAMILIES = {"r7g", "r7i", "r8g", "r8gd", "m7g", "m7i", "c7g", "c7i", "x8g"}
69
70_REGION_NAMES = {
71 "us-east-1": "US East (N. Virginia)",
72 "us-east-2": "US East (Ohio)",
73 "us-west-1": "US West (N. California)",
74 "us-west-2": "US West (Oregon)",
75 "eu-west-1": "EU (Ireland)",
76 "eu-central-1": "EU (Frankfurt)",
77 "ap-southeast-1": "Asia Pacific (Singapore)",
78 "ap-northeast-1": "Asia Pacific (Tokyo)",
79 "ap-south-1": "Asia Pacific (Mumbai)",
80}
81
82
83@dataclass
84class RIOffering:
85 instance_type: str
86 term_years: int
87 payment_option: str
88 effective_hourly: float
89 upfront_cost: float
90 recurring_hourly: float
91 multi_az: bool = False
92
93 def monthly_cost(self) -> float:
94 return self.effective_hourly * HOURS_PER_MONTH
95
96
97@dataclass
98class DSPRate:
99 usage_type: str
100 term_years: int
101 payment_option: str
102 rate_per_hour: float
103
104 def monthly_cost(self) -> float:
105 return self.rate_per_hour * HOURS_PER_MONTH
106
107
108def _family_from_instance(instance_type: str) -> str:
109 m = re.match(r"db\.([a-z0-9]+)\.", instance_type)
110 return m.group(1) if m else ""
111
112
113def get_on_demand_price(
114 instance_type: str, region: str = "us-east-1", multi_az: bool = False
115) -> float:
116 price = _STATIC_INSTANCE_PRICES.get(instance_type, 0.0)
117 if region != "us-east-1" and price > 0:
118 # Static prices reflect us-east-1 only; actual price in other regions may differ
119 import warnings
120
121 warnings.warn(
122 f"On-demand price for {instance_type} is based on us-east-1. "
123 f"Actual price in {region} may differ."
124 )
125 if multi_az:
126 price *= MULTI_AZ_MULTIPLIER
127 return price
128
129
130def fetch_ri_offerings(
131 instance_type: str, engine: str, region: str, multi_az: bool = False
132) -> list[RIOffering]:
133 try:
134 import boto3
135 except ImportError:
136 return []
137
138 product_desc = ENGINE_PRODUCT_MAP.get(engine, "MySQL")
139 results: list[RIOffering] = []
140 try:
141 rds = boto3.client("rds", region_name=region)
142 paginator = rds.get_paginator("describe_reserved_db_instances_offerings")
143 for page in paginator.paginate(
144 DBInstanceClass=instance_type,
145 ProductDescription=product_desc,
146 MultiAZ=multi_az,
147 ):
148 for offering in page.get("ReservedDBInstancesOfferings", []):
149 inst = offering.get("DBInstanceClass", "")
150 if inst != instance_type:
151 continue
152 duration = offering.get("Duration", 0)
153 term_years = 3 if duration > 94_000_000 else 1
154 payment = offering.get("OfferingType", "")
155 fixed = float(offering.get("FixedPrice", 0.0))
156 recurring_list = offering.get("RecurringCharges", [])
157 recurring_hr = sum(
158 float(rc.get("RecurringChargeAmount", 0.0)) for rc in recurring_list
159 )
160 term_hours = term_years * 365 * 24
161 effective = (fixed / term_hours) + recurring_hr
162 results.append(
163 RIOffering(
164 instance_type=inst,
165 term_years=term_years,
166 payment_option=payment,
167 effective_hourly=round(effective, 6),
168 upfront_cost=round(fixed, 2),
169 recurring_hourly=round(recurring_hr, 6),
170 multi_az=multi_az,
171 )
172 )
173 except Exception:
174 return []
175
176 seen = set()
177 deduped = []
178 for r in results:
179 key = (r.term_years, r.payment_option, round(r.effective_hourly, 6))
180 if key in seen:
181 continue
182 seen.add(key)
183 deduped.append(r)
184 return deduped
185
186
187def fetch_dsp_rates(engine: str, region: str) -> dict[str, list[DSPRate]]:
188 try:
189 import boto3
190 except ImportError:
191 return {}
192
193 result: dict[str, list[DSPRate]] = {}
194 product_desc = ENGINE_PRODUCT_MAP.get(engine, "MySQL")
195 try:
196 sp = boto3.client("savingsplans", region_name="us-east-1")
197 rates = []
198 token = None
199 while True:
200 kwargs = {
201 "savingsPlanTypes": ["Database"],
202 "products": ["RDS"],
203 "serviceCodes": ["AmazonRDS"],
204 "filters": [
205 {"name": "region", "values": [region]},
206 {"name": "productDescription", "values": [product_desc]},
207 ],
208 "maxResults": 1000,
209 }
210 if token:
211 kwargs["nextToken"] = token
212 resp = sp.describe_savings_plans_offering_rates(**kwargs)
213 rates.extend(resp.get("searchResults", []))
214 token = resp.get("nextToken")
215 if not token:
216 break
217
218 for rate_entry in rates:
219 offering = rate_entry.get("savingsPlanOffering", {})
220 dur = offering.get("durationSeconds", 0)
221 term_years = 3 if dur > 94_000_000 else 1
222 payment = offering.get("paymentOption", "")
223 try:
224 rate_val = float(rate_entry.get("rate", "0"))
225 except (ValueError, TypeError):
226 continue
227 if rate_val <= 0:
228 continue
229 usage = rate_entry.get("usageType", "")
230 # Usage types may carry a region prefix (e.g. "USE2-InstanceUsage:db.r7g.2xlarge"),
231 # so search anywhere in the string rather than anchoring at the start.
232 m = re.search(r"InstanceUsage:db\.(\w+)\.(\w+)", usage)
233 if not m:
234 continue
235 family = m.group(1)
236 size = m.group(2)
237 key = f"db.{family}.{size}"
238 entry = DSPRate(
239 usage_type=key,
240 term_years=term_years,
241 payment_option=payment,
242 rate_per_hour=round(rate_val, 6),
243 )
244 result.setdefault(key, []).append(entry)
245 except Exception:
246 pass
247 return result
248
249
250def best_ri(offerings: list[RIOffering], term_years: int) -> RIOffering | None:
251 candidates = [r for r in offerings if r.term_years == term_years]
252 if not candidates:
253 return None
254 return min(candidates, key=lambda r: r.effective_hourly)
255
256
257def best_dsp(rates: list[DSPRate], term_years: int = 1) -> DSPRate | None:
258 candidates = [r for r in rates if r.term_years == term_years]
259 if not candidates:
260 return None
261 return min(candidates, key=lambda r: r.rate_per_hour)
262
263
264def build_comparison(
265 instance_type: str,
266 engine: str,
267 num_instances: int,
268 region: str,
269 multi_az: bool = False,
270 dsp_rates: dict[str, list[DSPRate]] | None = None,
271) -> dict:
272 if dsp_rates is None:
273 dsp_rates = fetch_dsp_rates(engine, region)
274
275 family = _family_from_instance(instance_type)
276 od_hourly = get_on_demand_price(instance_type, region, multi_az)
277 od_monthly = od_hourly * HOURS_PER_MONTH * num_instances
278
279 ri_offerings = fetch_ri_offerings(instance_type, engine, region, multi_az)
280 ri_1yr = best_ri(ri_offerings, 1)
281 ri_3yr = best_ri(ri_offerings, 3)
282
283 ri_1yr_monthly = ri_1yr.effective_hourly * HOURS_PER_MONTH * num_instances if ri_1yr else None
284 ri_3yr_monthly = ri_3yr.effective_hourly * HOURS_PER_MONTH * num_instances if ri_3yr else None
285
286 dsp_entry_1yr = best_dsp(dsp_rates.get(instance_type, []), 1)
287 dsp_entry_3yr = best_dsp(dsp_rates.get(instance_type, []), 3)
288 # Multi-AZ consumes 2x the compute hours the savings plan must cover, mirroring
289 # the on-demand and RI Multi-AZ handling above. Without this, DSP savings are overstated.
290 az_multiplier = MULTI_AZ_MULTIPLIER if multi_az else 1.0
291 dsp_1yr_monthly = (
292 dsp_entry_1yr.rate_per_hour * HOURS_PER_MONTH * num_instances * az_multiplier
293 if dsp_entry_1yr
294 else None
295 )
296 dsp_3yr_monthly = (
297 dsp_entry_3yr.rate_per_hour * HOURS_PER_MONTH * num_instances * az_multiplier
298 if dsp_entry_3yr
299 else None
300 )
301
302 dsp_eligible = family in _DSP_ELIGIBLE_FAMILIES
303 notes = []
304 if od_hourly == 0:
305 notes.append(
306 f"No static on-demand price is bundled for {instance_type}, so the offline "
307 f"baseline is $0 and savings cannot be computed. Run against a live instance "
308 f"(no 'offline' subcommand) or supply pricing to get accurate figures."
309 )
310 if not dsp_eligible:
311 notes.append(
312 f"Database Savings Plans do not cover the {family} family. "
313 f"Eligible families: {', '.join(sorted(_DSP_ELIGIBLE_FAMILIES))}."
314 )
315 if multi_az:
316 notes.append(
317 "Multi-AZ pricing applied. Multi-AZ RIs are separate offerings from Single-AZ. "
318 "Ensure you purchase the correct deployment type."
319 )
320 if region != "us-east-1" and od_hourly > 0:
321 notes.append(
322 f"On-demand baseline for {instance_type} uses us-east-1 static pricing; "
323 f"actual {region} pricing may differ by 10-20%, so savings percentages are approximate. "
324 f"Provide live pricing or run in us-east-1 for exact figures."
325 )
326
327 def _fmt(ri, monthly, od):
328 if ri is None or monthly is None:
329 return None
330 savings = od - monthly
331 pct = (savings / od * 100) if od > 0 else 0
332 return {
333 "term_years": ri.term_years,
334 "payment_option": ri.payment_option,
335 "effective_hourly_per_instance": round(ri.effective_hourly, 4),
336 "upfront_total": round(ri.upfront_cost * num_instances, 2),
337 "monthly": round(monthly, 2),
338 "savings_monthly": round(savings, 2),
339 "savings_pct": round(pct, 1),
340 }
341
342 def _fmt_dsp(dsp, monthly, od):
343 if dsp is None or monthly is None:
344 return None
345 savings = od - monthly
346 pct = (savings / od * 100) if od > 0 else 0
347 return {
348 "term_years": dsp.term_years,
349 "payment_option": dsp.payment_option,
350 "rate_per_hour": round(dsp.rate_per_hour, 4),
351 "monthly": round(monthly, 2),
352 "savings_monthly": round(savings, 2),
353 "savings_pct": round(pct, 1),
354 }
355
356 options = []
357 if ri_1yr_monthly is not None:
358 options.append(("1yr RI", ri_1yr_monthly))
359 if ri_3yr_monthly is not None:
360 options.append(("3yr RI", ri_3yr_monthly))
361 if dsp_1yr_monthly is not None:
362 options.append(("1yr DSP", dsp_1yr_monthly))
363 if dsp_3yr_monthly is not None:
364 options.append(("3yr DSP", dsp_3yr_monthly))
365
366 if options:
367 best_label, best_cost = min(options, key=lambda x: x[1])
368 savings = od_monthly - best_cost
369 pct = (savings / od_monthly * 100) if od_monthly > 0 else 0
370 recommendation = {
371 "best_option": best_label,
372 "best_monthly_cost": round(best_cost, 2),
373 "savings_vs_on_demand": round(savings, 2),
374 "savings_pct": round(pct, 1),
375 }
376 else:
377 recommendation = {"best_option": "on_demand", "reason": "No RI or DSP offerings found."}
378
379 return {
380 "engine": engine,
381 "instance_type": instance_type,
382 "num_instances": num_instances,
383 "multi_az": multi_az,
384 "on_demand": {"hourly": round(od_hourly, 4), "monthly": round(od_monthly, 2)},
385 "ri_1yr": _fmt(ri_1yr, ri_1yr_monthly, od_monthly),
386 "ri_3yr": _fmt(ri_3yr, ri_3yr_monthly, od_monthly),
387 "dsp_1yr": _fmt_dsp(dsp_entry_1yr, dsp_1yr_monthly, od_monthly),
388 "dsp_3yr": _fmt_dsp(dsp_entry_3yr, dsp_3yr_monthly, od_monthly),
389 "recommendation": recommendation,
390 "notes": notes,
391 }
392
393
394def analyze_instance_live(instance_id: str, region: str) -> dict:
395 import boto3
396
397 rds = boto3.client("rds", region_name=region)
398 try:
399 resp = rds.describe_db_instances(DBInstanceIdentifier=instance_id)
400 except Exception as e:
401 return {"instance_id": instance_id, "error": str(e)}
402 instances = resp.get("DBInstances", [])
403 if not instances:
404 return {"instance_id": instance_id, "error": "instance not found"}
405 inst = instances[0]
406 engine = inst.get("Engine", "")
407 instance_type = inst.get("DBInstanceClass", "")
408 multi_az = inst.get("MultiAZ", False)
409 replicas = inst.get("ReadReplicaDBInstanceIdentifiers", [])
410
411 result = build_comparison(
412 instance_type=instance_type,
413 engine=engine,
414 num_instances=1,
415 region=region,
416 multi_az=multi_az,
417 )
418 result["instance_id"] = instance_id
419 result["engine_version"] = inst.get("EngineVersion", "")
420 if replicas:
421 result["notes"].append(
422 f"Instance has {len(replicas)} read replica(s). "
423 "Consider separate RI/DSP for each replica (Single-AZ pricing)."
424 )
425 return result
426
427
428def main():
429 parser = argparse.ArgumentParser(
430 description="RDS RI & Database Savings Plan estimator (read-only)"
431 )
432 parser.add_argument("--region", default="us-east-1")
433 parser.add_argument("--format", choices=["json"], default="json")
434 parser.add_argument("--instance", help="Analyze a single RDS instance by identifier")
435
436 sub = parser.add_subparsers(dest="mode")
437 off = sub.add_parser("offline", help="Use user-supplied workload description")
438 off.add_argument("--instance-type", required=True, help="e.g., db.r7g.2xlarge")
439 off.add_argument("--engine", required=True, choices=["mysql", "mariadb", "postgres"])
440 off.add_argument("--num-instances", type=int, default=1)
441 off.add_argument("--multi-az", action="store_true")
442 # --region and --format are defined on the main parser above; do NOT redefine them
443 # here, or the subparser's default silently overrides a value passed before 'offline'.
444
445 args = parser.parse_args()
446
447 if args.mode == "offline":
448 result = build_comparison(
449 instance_type=args.instance_type,
450 engine=args.engine,
451 num_instances=args.num_instances,
452 region=args.region,
453 multi_az=args.multi_az,
454 )
455 print(json.dumps(result, indent=2, default=str))
456 return
457
458 if args.instance:
459 result = analyze_instance_live(args.instance, args.region)
460 print(json.dumps(result, indent=2, default=str))
461 return
462
463 parser.print_help()
464
465
466if __name__ == "__main__":
467 main()