#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.10" # dependencies = ["boto3>=1.34"] # /// """Inspect Reserved Instance / Savings Plan coverage as compact JSON. A cost step with flat usage-hours and no CloudTrail config change is the classic signature of a *coverage lapse* — a Reserved Instance (or Savings Plan) expired and the same resources started billing on-demand. This script lists reservations across the common reservable services AND the account's Savings Plan inventory, and flags any that expired (RI) or retired (SP) near a date of interest, so that hypothesis can be confirmed or ruled out without guesswork. Both the reservation describes (RDS/EC2/ElastiCache/OpenSearch/Redshift) and the Savings Plans inventory (`describe_savings_plans`) are **free**, so both run unconditionally on every call. The --savings-plans flag is separate: it additionally pulls Savings Plans *utilization/coverage* from Cost Explorer (each CE call costs ~$0.01) to judge whether buying a Savings Plan would recover the lost discount. Note: that CE view reports coverage %, but only the (free, always-on) SP inventory tells you *which* plan retired — when no RI expired near the spike, look there for a `retired` Compute/EC2 plan whose `commitment_per_hr` ≈ the daily on-demand $ jump. Usage: # What reservations/Savings Plans expired around the spike start? uv run scripts/reservation_coverage.py --near 2026-05-26 --region eu-central-1 # Narrow to one service uv run scripts/reservation_coverage.py --near 2026-05-26 --service rds --region eu-central-1 # Also assess Savings Plan utilization/coverage over the anomaly window uv run scripts/reservation_coverage.py --near 2026-05-26 --region eu-central-1 \ --savings-plans --start 2026-05-26 --end 2026-06-11 """ import argparse import json from datetime import datetime, timedelta, timezone import boto3 # Services whose reservations we can describe directly (free, no Cost Explorer cost). SERVICES = ["rds", "ec2", "elasticache", "opensearch", "redshift"] # How close to --near a reservation's expiry must be to get flagged. NEAR_WINDOW_DAYS_DEFAULT = 10 def parse_date(s: str) -> datetime: return datetime.strptime(s, "%Y-%m-%d").replace(tzinfo=timezone.utc) def _iso(dt) -> str | None: return dt.isoformat() if dt else None def _expiry(start, duration_seconds): if start is None or duration_seconds is None: return None return start + timedelta(seconds=int(duration_seconds)) def list_rds(session, region): c = session.client("rds", region_name=region) out = [] for r in c.describe_reserved_db_instances().get("ReservedDBInstances", []): start = r.get("StartTime") out.append({ "service": "rds", "id": r.get("ReservedDBInstanceId"), "type": r.get("DBInstanceClass"), "count": r.get("DBInstanceCount"), "state": r.get("State"), "engine": r.get("ProductDescription"), "start": _iso(start), "expiry": _iso(_expiry(start, r.get("Duration"))), }) return out def list_ec2(session, region): c = session.client("ec2", region_name=region) out = [] for r in c.describe_reserved_instances().get("ReservedInstances", []): out.append({ "service": "ec2", "id": r.get("ReservedInstancesId"), "type": r.get("InstanceType"), "count": r.get("InstanceCount"), "state": r.get("State"), "engine": r.get("ProductDescription"), "start": _iso(r.get("Start")), "expiry": _iso(r.get("End")), # EC2 gives End directly }) return out def list_elasticache(session, region): c = session.client("elasticache", region_name=region) out = [] for r in c.describe_reserved_cache_nodes().get("ReservedCacheNodes", []): start = r.get("StartTime") out.append({ "service": "elasticache", "id": r.get("ReservedCacheNodeId"), "type": r.get("CacheNodeType"), "count": r.get("CacheNodeCount"), "state": r.get("State"), "engine": r.get("ProductDescription"), "start": _iso(start), "expiry": _iso(_expiry(start, r.get("Duration"))), }) return out def list_opensearch(session, region): c = session.client("opensearch", region_name=region) out = [] for r in c.describe_reserved_instances().get("ReservedInstances", []): start = r.get("StartTime") out.append({ "service": "opensearch", "id": r.get("ReservedInstanceId"), "type": r.get("InstanceType"), "count": r.get("InstanceCount"), "state": r.get("State"), "engine": "OpenSearch", "start": _iso(start), "expiry": _iso(_expiry(start, r.get("Duration"))), }) return out def list_redshift(session, region): c = session.client("redshift", region_name=region) out = [] for r in c.describe_reserved_nodes().get("ReservedNodes", []): start = r.get("StartTime") out.append({ "service": "redshift", "id": r.get("ReservedNodeId"), "type": r.get("NodeType"), "count": r.get("NodeCount"), "state": r.get("State"), "engine": "Redshift", "start": _iso(start), "expiry": _iso(_expiry(start, r.get("Duration"))), }) return out LISTERS = { "rds": list_rds, "ec2": list_ec2, "elasticache": list_elasticache, "opensearch": list_opensearch, "redshift": list_redshift, } def _parse_iso(s): """Parse an ISO8601 string, tolerating a trailing 'Z' (the SP API returns it).""" if not s: return None if isinstance(s, datetime): return s return datetime.fromisoformat(str(s).replace("Z", "+00:00")) def list_savings_plans(session): """List the account's Savings Plan inventory (all states). Free — not Cost Explorer. Savings Plans are a global resource; the client lives in us-east-1 regardless of --region. A *retired* Compute/EC2 plan whose `end` is ~spike-day−1 and whose `commitment_per_hr` ≈ the daily on-demand $ jump is the confirmed root cause of a coverage-lapse cost step — the thing the Cost Explorer coverage % can't name. """ c = session.client("savingsplans", region_name="us-east-1") out = [] token = None while True: kwargs = { # All states, so a lapsed plan (which is `retired`) is caught, plus # queued/failed renewals that could explain a gap. "states": [ "active", "retired", "queued", "queued-deleted", "payment-failed", "pending-return", "returned", ], } if token: kwargs["nextToken"] = token resp = c.describe_savings_plans(**kwargs) for sp in resp.get("savingsPlans", []): out.append({ "service": "savings-plan", "id": sp.get("savingsPlanId"), "type": sp.get("savingsPlanType"), # Compute / EC2Instance / SageMaker "instance_family": sp.get("ec2InstanceFamily"), "commitment_per_hr": sp.get("commitment"), "state": sp.get("state"), "start": sp.get("start"), "expiry": sp.get("end"), # already ISO8601; normalized key for flagging "region": sp.get("region"), }) token = resp.get("nextToken") if not token: break return out def savings_plans_view(session, start, end): """Cost Explorer Savings Plans utilization + coverage. ~2 CE calls (~$0.02).""" ce = session.client("ce") # Cost Explorer is global; lives in us-east-1 period = {"Start": start.strftime("%Y-%m-%d"), "End": (end + timedelta(days=1)).strftime("%Y-%m-%d")} util = ce.get_savings_plans_utilization(TimePeriod=period) agg = util.get("Total", {}) cov = ce.get_savings_plans_coverage(TimePeriod=period, Granularity="MONTHLY") cov_totals = [c.get("Coverage", {}) for c in cov.get("SavingsPlansCoverages", [])] return { "period": period, "utilization": { "used_commitment_usd": (agg.get("Utilization") or {}).get("UsedCommitment"), "unused_commitment_usd": (agg.get("Utilization") or {}).get("UnusedCommitment"), "utilization_pct": (agg.get("Utilization") or {}).get("UtilizationPercentage"), "net_savings_usd": (agg.get("Savings") or {}).get("NetSavings"), }, "coverage": [ { "on_demand_cost_usd": c.get("OnDemandCost"), "covered_commitment_usd": c.get("SpendCoveredBySavingsPlans"), "coverage_pct": c.get("CoveragePercentage"), } for c in cov_totals ], "note": ( "Low coverage_pct with high on_demand_cost_usd means a Savings Plan " "(or RI) would recover discount on that on-demand spend. Match the " "commitment to the steady-state on-demand run-rate, not the peak." ), } def main() -> None: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--near", type=parse_date, help="date of interest (usually the spike start); reservations " "expiring within --near-window days of it are flagged") parser.add_argument("--near-window", type=int, default=NEAR_WINDOW_DAYS_DEFAULT, help=f"days around --near to flag expiries (default: {NEAR_WINDOW_DAYS_DEFAULT})") parser.add_argument("--service", choices=SERVICES, help="restrict to one reservable service (default: all)") parser.add_argument("--region", help="region to query (default: session/default region)") parser.add_argument("--savings-plans", action="store_true", help="also fetch Savings Plans utilization/coverage (uses Cost Explorer, ~$0.02)") parser.add_argument("--start", type=parse_date, help="Savings Plans window start (with --savings-plans)") parser.add_argument("--end", type=parse_date, help="Savings Plans window end, inclusive (with --savings-plans)") parser.add_argument("--profile", help="AWS profile name") args = parser.parse_args() session = boto3.Session(profile_name=args.profile, region_name=args.region) services = [args.service] if args.service else SERVICES reservations = [] errors = {} for svc in services: try: reservations.extend(LISTERS[svc](session, args.region)) except Exception as e: # service not used in this account/region, or no perms errors[svc] = str(e).splitlines()[0][:200] # Savings Plan inventory is global and free — always list it (like the # reservation describes), so an expired SP is never missed for lack of a flag. savings_plans = [] try: savings_plans = list_savings_plans(session) except Exception as e: # no Savings Plans read permission, etc. errors["savings-plans"] = str(e).splitlines()[0][:200] def flag_near(items): """Flag items whose `expiry` falls within ±near-window of --near.""" lo = args.near - timedelta(days=args.near_window) hi = args.near + timedelta(days=args.near_window) out = [] for it in items: exp = _parse_iso(it.get("expiry")) if exp is None: continue if lo <= exp <= hi: out.append(dict(it, expired=exp <= args.near, days_from_near=(exp - args.near).days)) out.sort(key=lambda it: it["expiry"]) return out # Flag reservations and Savings Plans expiring/retiring near the date of interest. flagged = [] sp_flagged = [] if args.near: flagged = flag_near(reservations) sp_flagged = flag_near(savings_plans) active = [r for r in reservations if r.get("state") == "active"] active.sort(key=lambda r: r.get("expiry") or "") active_sps = [s for s in savings_plans if s.get("state") == "active"] active_sps.sort(key=lambda s: s.get("expiry") or "") result = { "region_queried": args.region or session.region_name, "services_checked": services, "reservations_found": len(reservations), "active_reservations": active, "savings_plans_found": len(savings_plans), "active_savings_plans": active_sps, } if args.near: result["near"] = args.near.strftime("%Y-%m-%d") result["near_window_days"] = args.near_window result["expiring_near_date"] = flagged result["lapse_hint"] = ( f"{len(flagged)} reservation(s) expire within {args.near_window} days of " f"{result['near']}. Any 'expired: true' entry whose type matches the spiking " "usage type is a likely coverage lapse — confirm the spiking instance class " "equals the reservation type." ) if flagged else ( f"No reservations expire within {args.near_window} days of {result['near']}. " "RI coverage lapse is unlikely; check savings_plans_expiring_near, then " "look elsewhere (new capacity, rate change)." ) result["savings_plans_expiring_near"] = sp_flagged retired_near = [s for s in sp_flagged if s.get("state") == "retired"] if retired_near: top = retired_near[0] result["savings_plan_lapse_hint"] = ( f"{len(retired_near)} Savings Plan(s) retired within {args.near_window} " f"days of {result['near']}. {top.get('type')} SP {top.get('id')} " f"(${top.get('commitment_per_hr')}/hr) retired " f"{(top.get('expiry') or '')[:10]} — if its commitment_per_hr ≈ the daily " "on-demand $ jump for the spiking compute, this lapse is the root cause; " "repurchase a comparable SP to restore coverage." ) else: result["savings_plan_lapse_hint"] = ( f"No Savings Plans retired within {args.near_window} days of " f"{result['near']}." ) if errors: result["unavailable_services"] = errors if args.savings_plans: if not (args.start and args.end): result["savings_plans_error"] = "--savings-plans requires --start and --end" else: try: result["savings_plans"] = savings_plans_view(session, args.start, args.end) except Exception as e: result["savings_plans_error"] = str(e).splitlines()[0][:200] print(json.dumps(result, indent=1, default=str)) if __name__ == "__main__": main()