#!/usr/bin/env python3 """Inventory reclaimable AWS spend: unattached EBS volumes, idle load balancers, public IPv4. Read-only. Produces candidate lists for a human to confirm; it never deletes anything. Two questions it answers, from the July cost attribution: 1. Unattached EBS volumes anywhere in the account -- billed in full, attached to nothing. 2. On the two base control planes, where the money actually goes: load balancers and public IPv4 addresses dominate their cost, not compute. An ALB/NLB whose target groups have no registered or no healthy targets is serving nothing while still billing. Usage: python3 reclaim_candidates.py --profile PingCAP python3 reclaim_candidates.py --profile PingCAP --vpc vpc-068ed861c2dad891c """ import argparse import collections import json import subprocess import sys REGIONS = ["us-west-2", "us-east-1"] # us-east-1 list prices; close enough for ranking candidates. EBS_GB_MONTH = {"gp3": 0.08, "gp2": 0.10, "io1": 0.125, "io2": 0.125, "st1": 0.045, "sc1": 0.015} LB_HOURS_MONTH = 730 * 0.0225 # ALB/NLB hourly PUBLIC_IPV4_MONTH = 730 * 0.005 def aws(profile, region, *args): cmd = ["aws", "--profile", profile, "--output", "json"] if region: cmd += ["--region", region] out = subprocess.run(cmd + list(args), capture_output=True, text=True, timeout=180) if out.returncode != 0: print(f" ! {args[0]} {args[1]} ({region}): {out.stderr.strip()[:100]}", file=sys.stderr) return None return json.loads(out.stdout or "null") def name_of(tags): for t in tags or []: if t.get("Key") == "Name": return t["Value"] return "" def unattached_volumes(profile): rows = [] for region in REGIONS: vols = aws(profile, region, "ec2", "describe-volumes", "--filters", "Name=status,Values=available") or {} for v in vols.get("Volumes", []): rate = EBS_GB_MONTH.get(v["VolumeType"], 0.08) rows.append(dict(region=region, id=v["VolumeId"], size=v["Size"], type=v["VolumeType"], created=str(v.get("CreateTime"))[:10], name=name_of(v.get("Tags")), monthly=v["Size"] * rate)) return sorted(rows, key=lambda r: -r["monthly"]) def load_balancers(profile, vpcs): rows = [] for region in REGIONS: lbs = aws(profile, region, "elbv2", "describe-load-balancers") or {} tgs = aws(profile, region, "elbv2", "describe-target-groups") or {} by_lb = collections.defaultdict(list) for tg in tgs.get("TargetGroups", []): for arn in tg.get("LoadBalancerArns", []): by_lb[arn].append(tg) for lb in lbs.get("LoadBalancers", []): if vpcs and lb.get("VpcId") not in vpcs: continue registered = healthy = 0 for tg in by_lb.get(lb["LoadBalancerArn"], []): h = aws(profile, region, "elbv2", "describe-target-health", "--target-group-arn", tg["TargetGroupArn"]) or {} for t in h.get("TargetHealthDescriptions", []): registered += 1 if t["TargetHealth"]["State"] == "healthy": healthy += 1 rows.append(dict(region=region, name=lb["LoadBalancerName"], vpc=lb.get("VpcId"), type=lb.get("Type"), scheme=lb.get("Scheme"), created=str(lb.get("CreatedTime"))[:10], target_groups=len(by_lb.get(lb["LoadBalancerArn"], [])), registered=registered, healthy=healthy, monthly=LB_HOURS_MONTH)) classic = aws(profile, region, "elb", "describe-load-balancers") or {} for lb in classic.get("LoadBalancerDescriptions", []): if vpcs and lb.get("VPCId") not in vpcs: continue insts = lb.get("Instances") or [] rows.append(dict(region=region, name=lb["LoadBalancerName"], vpc=lb.get("VPCId"), type="classic", scheme=lb.get("Scheme"), created=str(lb.get("CreatedTime"))[:10], target_groups=0, registered=len(insts), healthy=len(insts), monthly=LB_HOURS_MONTH)) return rows def public_ips(profile, vpcs): rows = [] for region in REGIONS: enis = aws(profile, region, "ec2", "describe-network-interfaces") or {} for e in enis.get("NetworkInterfaces", []): assoc = e.get("Association") or {} if not assoc.get("PublicIp"): continue if vpcs and e.get("VpcId") not in vpcs: continue rows.append(dict(region=region, eni=e["NetworkInterfaceId"], vpc=e.get("VpcId"), ip=assoc["PublicIp"], status=e.get("Status"), desc=(e.get("Description") or "")[:58], itype=(e.get("InterfaceType") or ""), monthly=PUBLIC_IPV4_MONTH)) return rows def main(): ap = argparse.ArgumentParser() ap.add_argument("--profile", required=True) ap.add_argument("--vpc", action="append", default=[], help="restrict the LB / public-IP audit to these VPCs (repeatable)") args = ap.parse_args() vpcs = set(args.vpc) vols = unattached_volumes(args.profile) total = sum(v["monthly"] for v in vols) print(f"UNATTACHED EBS VOLUMES -- {len(vols)} volumes, ~${total:,.0f}/mo, " f"{sum(v['size'] for v in vols):,} GB\n") print(f"{'region':<12}{'volume':<24}{'GB':>7}{'type':>7}{'created':>13} {'$/mo':>7} name") for v in vols[:40]: print(f"{v['region']:<12}{v['id']:<24}{v['size']:>7}{v['type']:>7}{v['created']:>13} " f"{v['monthly']:>7,.0f} {v['name'][:34]}") if len(vols) > 40: print(f" ... and {len(vols)-40} more") lbs = load_balancers(args.profile, vpcs) idle = [l for l in lbs if l["healthy"] == 0] print(f"\n\nLOAD BALANCERS{' in ' + ','.join(sorted(vpcs)) if vpcs else ''} -- " f"{len(lbs)} total, {len(idle)} with ZERO healthy targets (~${len(idle)*LB_HOURS_MONTH:,.0f}/mo)\n") print(f"{'name':<44}{'type':<9}{'scheme':<10}{'created':<12}{'tgs':>4}{'reg':>5}{'healthy':>8}") for l in sorted(lbs, key=lambda r: (r["healthy"], r["registered"])): print(f"{l['name'][:42]:<44}{str(l['type']):<9}{str(l['scheme'])[:8]:<10}" f"{l['created']:<12}{l['target_groups']:>4}{l['registered']:>5}{l['healthy']:>8}") ips = public_ips(args.profile, vpcs) print(f"\n\nPUBLIC IPv4{' in ' + ','.join(sorted(vpcs)) if vpcs else ''} -- " f"{len(ips)} addresses, ~${len(ips)*PUBLIC_IPV4_MONTH:,.0f}/mo\n") kinds = collections.Counter(i["itype"] or "normal" for i in ips) for k, n in kinds.most_common(): print(f" {k:<28}{n:>4} ~${n*PUBLIC_IPV4_MONTH:,.0f}/mo") print() for i in sorted(ips, key=lambda r: r["desc"])[:30]: print(f" {i['eni']:<24}{i['ip']:<17}{(i['itype'] or 'normal'):<20}{i['desc']}") if __name__ == "__main__": main()