Clean Up Unused AWS Resources — EBS Volumes, EIPs, Old AMIs with Cost Report

"""
aws_resource_cleanup.py

Find and optionally delete unused AWS resources to reduce costs.
Run with DRY_RUN=true (default) to see what would be deleted first.
"""

import boto3
import json
import logging
import os
import sys
from datetime import datetime, timezone
from dataclasses import dataclass, field
from collections import defaultdict

logger = logging.getLogger(__name__)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)


@dataclass
class CleanupReport:
    """Accumulates findings and savings estimates across all cleanup operations."""
    # Lists of resources found/deleted
    ebs_volumes:  list = field(default_factory=list)
    eips:         list = field(default_factory=list)
    amis:         list = field(default_factory=list)

    # Cost estimates
    ebs_savings_per_month:  float = 0.0
    eip_savings_per_month:  float = 0.0
    ami_savings_per_month:  float = 0.0

    @property
    def total_savings_per_month(self) -> float:
        return self.ebs_savings_per_month + self.eip_savings_per_month + self.ami_savings_per_month


class AWSResourceCleaner:
    """
    Finds and removes unused AWS resources in a single region.

    All destructive operations are gated by dry_run=True.
    Always run with dry_run=True first to review what will be deleted.
    """

    # Rough pricing — varies by region; adjust if needed
    EBS_PRICE_PER_GB_MONTH = {
        "gp2": 0.10,
        "gp3": 0.08,
        "io1": 0.125,
        "io2": 0.125,
        "st1": 0.045,
        "sc1": 0.015,
        "standard": 0.05,
    }
    EIP_PRICE_PER_HOUR      = 0.005   # When not associated
    SNAPSHOT_PRICE_PER_GB   = 0.05    # Per GB per month

    def __init__(self, region: str = "us-east-1", dry_run: bool = True):
        """
        region:   AWS region to audit (one region at a time).
        dry_run:  When True, reports findings but does NOT delete anything.
                  Set to False only after reviewing the dry-run output.
        """
        self.ec2     = boto3.client("ec2", region_name=region)
        self.region  = region
        self.dry_run = dry_run
        self.report  = CleanupReport()

        if dry_run:
            logger.info("Running in DRY-RUN mode — no resources will be deleted")
        else:
            logger.warning("Running in LIVE mode — resources WILL be deleted!")

    # ══════════════════════════════════════════════════════════════════
    # Part 1: Unattached EBS Volumes
    # ══════════════════════════════════════════════════════════════════
    def clean_unattached_ebs_volumes(self) -> list[dict]:
        """
        Find EBS volumes in 'available' state — these are not attached
        to any instance and accumulating charges.

        Volume lifecycle states:
          creating  → available (not attached) → in-use (attached) → deleting → deleted
          also: error, recovering

        We paginate describe_volumes with Filters so that only
        'available' volumes are returned — no need to filter client-side.

        For each volume we estimate the monthly waste and log it.
        """
        logger.info(f"[EBS] Scanning for unattached volumes in {self.region}")
        found = []

        paginator = self.ec2.get_paginator("describe_volumes")
        for page in paginator.paginate(
            Filters=[{"Name": "status", "Values": ["available"]}]
        ):
            for vol in page["Volumes"]:
                volume_id   = vol["VolumeId"]
                size_gb     = vol["Size"]
                vol_type    = vol["VolumeType"]
                create_time = vol["CreateTime"]
                tags        = {t["Key"]: t["Value"] for t in vol.get("Tags", [])}
                name        = tags.get("Name", "")

                # Calculate days idle (how long since creation with no attachment)
                # Note: real idle time requires checking attachment history, which
                # is not in the EC2 API. We use creation time as a proxy.
                days_old = (datetime.now(timezone.utc) - create_time).days

                # Estimate monthly cost
                price_per_gb = self.EBS_PRICE_PER_GB_MONTH.get(vol_type, 0.08)
                monthly_cost = size_gb * price_per_gb

                entry = {
                    "volume_id":    volume_id,
                    "name":         name,
                    "size_gb":      size_gb,
                    "type":         vol_type,
                    "days_old":     days_old,
                    "monthly_cost": round(monthly_cost, 2),
                    "deleted":      False,
                }

                logger.info(
                    f"[EBS] Found unattached: {volume_id} ({name}) "
                    f"{size_gb} GB {vol_type} — ${monthly_cost:.2f}/mo"
                )

                if not self.dry_run:
                    try:
                        # delete_volume() permanently destroys the EBS volume.
                        # This is IRREVERSIBLE — data is gone.
                        # The volume must be in 'available' state (not attached).
                        self.ec2.delete_volume(VolumeId=volume_id)
                        entry["deleted"] = True
                        logger.info(f"[EBS] Deleted: {volume_id}")
                        self.report.ebs_savings_per_month += monthly_cost
                    except self.ec2.exceptions.ClientError as e:
                        logger.error(f"[EBS] Failed to delete {volume_id}: {e}")
                else:
                    # In dry-run mode, still accumulate savings estimate
                    self.report.ebs_savings_per_month += monthly_cost

                found.append(entry)

        self.report.ebs_volumes = found
        logger.info(
            f"[EBS] Found {len(found)} unattached volumes "
            f"(${self.report.ebs_savings_per_month:.2f}/mo potential savings)"
        )
        return found

    # ══════════════════════════════════════════════════════════════════
    # Part 2: Unassociated Elastic IPs
    # ══════════════════════════════════════════════════════════════════
    def clean_unassociated_eips(self) -> list[dict]:
        """
        Find Elastic IPs not associated with any running instance or
        network interface. AWS charges $0.005/hour for idle EIPs.

        describe_addresses() returns ALL EIPs in the region.
        An EIP is unassociated if it has no AssociationId field.

        Two allocation domains:
          vpc:      EIP allocated for use in a VPC (AllocationId exists)
          standard: Legacy EC2-Classic (almost extinct, treat same way)

        release_address() returns the EIP to the AWS pool.
        You can no longer use that specific IP after this call.
        """
        logger.info(f"[EIP] Scanning for unassociated Elastic IPs in {self.region}")
        found = []

        # describe_addresses() is NOT paginated — returns all at once
        response = self.ec2.describe_addresses()

        for addr in response["Addresses"]:
            # Skip EIPs that are associated (in use)
            if "AssociationId" in addr:
                continue

            allocation_id = addr.get("AllocationId", "")
            public_ip     = addr["PublicIp"]
            tags          = {t["Key"]: t["Value"] for t in addr.get("Tags", [])}
            name          = tags.get("Name", "")

            # Cost: $0.005/hour × 24h × 30.5 days ≈ $3.65/month
            monthly_cost  = self.EIP_PRICE_PER_HOUR * 24 * 30.5

            entry = {
                "allocation_id": allocation_id,
                "public_ip":     public_ip,
                "name":          name,
                "monthly_cost":  round(monthly_cost, 2),
                "released":      False,
            }

            logger.info(
                f"[EIP] Found idle: {public_ip} ({name}) "
                f"— ${monthly_cost:.2f}/mo"
            )

            if not self.dry_run and allocation_id:
                try:
                    # release_address() returns this EIP to the AWS pool.
                    # Use AllocationId for VPC EIPs (not PublicIp).
                    self.ec2.release_address(AllocationId=allocation_id)
                    entry["released"] = True
                    logger.info(f"[EIP] Released: {public_ip} ({allocation_id})")
                    self.report.eip_savings_per_month += monthly_cost
                except self.ec2.exceptions.ClientError as e:
                    logger.error(f"[EIP] Failed to release {public_ip}: {e}")
            else:
                self.report.eip_savings_per_month += monthly_cost

            found.append(entry)

        self.report.eips = found
        logger.info(
            f"[EIP] Found {len(found)} unassociated EIPs "
            f"(${self.report.eip_savings_per_month:.2f}/mo potential savings)"
        )
        return found

    # ══════════════════════════════════════════════════════════════════
    # Part 3: Old AMIs — Keep Only N Most Recent Per Name Prefix
    # ══════════════════════════════════════════════════════════════════
    def clean_old_amis(self, keep_count: int = 3, name_prefix: str = "") -> list[dict]:
        """
        Delete AMIs older than the N most recent for each name prefix group.

        Strategy:
          - Group AMIs by name prefix (e.g., "app-server-*")
          - Sort each group by creation date, newest first
          - Keep the first `keep_count` AMIs
          - Deregister the rest, then delete their backing snapshots

        Two-step deletion:
          1. deregister_image(ImageId)   — removes the AMI registration
             (cannot launch new instances from it, but snapshots still exist)
          2. delete_snapshot(SnapshotId) — deletes the actual EBS snapshot
             (this is what actually frees storage and stops billing)

        Skipping these steps means orphaned snapshots continue to bill you
        even after the AMI is deregistered.

        OwnerIds=["self"] limits results to AMIs owned by THIS account.
        Without this, describe_images() could return AWS marketplace AMIs.
        """
        logger.info(
            f"[AMI] Scanning for old AMIs in {self.region} "
            f"(keep latest {keep_count} per name prefix)"
        )
        found = []

        # Fetch all AMIs owned by this account
        filters = [{"Name": "state", "Values": ["available"]}]
        if name_prefix:
            filters.append({"Name": "name", "Values": [f"{name_prefix}*"]})

        response = self.ec2.describe_images(
            OwnerIds=["self"],
            Filters=filters,
        )
        all_amis = response["Images"]

        # Group AMIs by name prefix (first word before a timestamp/version separator)
        # e.g., "app-server-20250120" → group key "app-server"
        groups: dict[str, list] = defaultdict(list)
        for ami in all_amis:
            ami_name  = ami.get("Name", "")
            # Split on common separators: -, _, space; take first 2 parts as prefix
            parts = ami_name.replace("_", "-").split("-")
            # Use first 2 dash-parts as the grouping key (customise as needed)
            group_key = "-".join(parts[:2]) if len(parts) >= 2 else ami_name
            groups[group_key].append(ami)

        for group_key, amis_in_group in groups.items():
            # Sort newest first by creation date
            # CreationDate format: "2025-01-20T14:30:00.000Z" — lexicographic sort works
            amis_sorted = sorted(
                amis_in_group, key=lambda a: a["CreationDate"], reverse=True
            )
            to_delete = amis_sorted[keep_count:]   # Everything after the N most recent

            for ami in to_delete:
                image_id      = ami["ImageId"]
                ami_name      = ami.get("Name", "")
                creation_date = ami["CreationDate"]

                # Collect snapshot IDs from the AMI's block device mappings
                # Each AMI has one or more EBS snapshots backing its volumes
                snapshot_ids = [
                    bdm["Ebs"]["SnapshotId"]
                    for bdm in ami.get("BlockDeviceMappings", [])
                    if "Ebs" in bdm and "SnapshotId" in bdm["Ebs"]
                ]

                # Estimate snapshot storage size
                total_size_gb = sum(
                    bdm["Ebs"].get("VolumeSize", 0)
                    for bdm in ami.get("BlockDeviceMappings", [])
                    if "Ebs" in bdm
                )
                monthly_cost = total_size_gb * self.SNAPSHOT_PRICE_PER_GB

                entry = {
                    "image_id":     image_id,
                    "name":         ami_name,
                    "group":        group_key,
                    "created":      creation_date,
                    "snapshots":    snapshot_ids,
                    "size_gb":      total_size_gb,
                    "monthly_cost": round(monthly_cost, 2),
                    "deregistered": False,
                    "snapshots_deleted": [],
                }

                logger.info(
                    f"[AMI] Old AMI: {image_id} ({ami_name}) "
                    f"created {creation_date[:10]} "
                    f"— ${monthly_cost:.2f}/mo in snapshots"
                )

                if not self.dry_run:
                    try:
                        # Step 1: Deregister the AMI
                        # After this, you cannot launch new instances from this AMI.
                        # Existing running instances are NOT affected.
                        self.ec2.deregister_image(ImageId=image_id)
                        entry["deregistered"] = True
                        logger.info(f"[AMI] Deregistered: {image_id}")

                        # Step 2: Delete each backing snapshot
                        for snap_id in snapshot_ids:
                            try:
                                self.ec2.delete_snapshot(SnapshotId=snap_id)
                                entry["snapshots_deleted"].append(snap_id)
                                logger.info(f"[AMI] Deleted snapshot: {snap_id}")
                            except self.ec2.exceptions.ClientError as e:
                                logger.warning(
                                    f"[AMI] Could not delete snapshot {snap_id}: {e}"
                                )

                        self.report.ami_savings_per_month += monthly_cost

                    except self.ec2.exceptions.ClientError as e:
                        logger.error(f"[AMI] Failed to deregister {image_id}: {e}")
                else:
                    self.report.ami_savings_per_month += monthly_cost

                found.append(entry)

        self.report.amis = found
        logger.info(
            f"[AMI] Found {len(found)} old AMIs "
            f"(${self.report.ami_savings_per_month:.2f}/mo potential savings)"
        )
        return found

    # ══════════════════════════════════════════════════════════════════
    # Part 4: Generate Cost-Savings Report
    # ══════════════════════════════════════════════════════════════════
    def generate_report(self, output_file: str = "cleanup_report.json") -> dict:
        """
        Write a JSON cost-savings report with all findings.
        The report is machine-readable (can be sent to Slack, stored in S3,
        or imported into a spreadsheet).
        """
        report_data = {
            "generated_at":  datetime.now(timezone.utc).isoformat(),
            "region":        self.region,
            "dry_run":       self.dry_run,
            "summary": {
                "ebs_volumes_found":     len(self.report.ebs_volumes),
                "eips_found":            len(self.report.eips),
                "amis_found":            len(self.report.amis),
                "ebs_savings_per_month": round(self.report.ebs_savings_per_month, 2),
                "eip_savings_per_month": round(self.report.eip_savings_per_month, 2),
                "ami_savings_per_month": round(self.report.ami_savings_per_month, 2),
                "total_savings_per_month": round(self.report.total_savings_per_month, 2),
                "total_savings_per_year":  round(self.report.total_savings_per_month * 12, 2),
            },
            "ebs_volumes": self.report.ebs_volumes,
            "eips":        self.report.eips,
            "amis":        self.report.amis,
        }

        # json.dump with default=str handles datetime objects
        with open(output_file, "w") as f:
            json.dump(report_data, f, indent=2, default=str)

        self._print_summary(report_data["summary"])
        logger.info(f"Full report written to {output_file}")
        return report_data

    def _print_summary(self, summary: dict) -> None:
        """Print a human-readable cost summary to console."""
        mode = "DRY-RUN estimate" if self.dry_run else "Actual savings"
        print(f"\n{'='*60}")
        print(f"AWS RESOURCE CLEANUP REPORT — {self.region.upper()}")
        print(f"Mode: {mode}")
        print(f"{'='*60}")
        print(f"  Unattached EBS volumes: {summary['ebs_volumes_found']:>4}  ${summary['ebs_savings_per_month']:>8.2f}/mo")
        print(f"  Unassociated EIPs:      {summary['eips_found']:>4}  ${summary['eip_savings_per_month']:>8.2f}/mo")
        print(f"  Old AMIs:               {summary['amis_found']:>4}  ${summary['ami_savings_per_month']:>8.2f}/mo")
        print(f"{'─'*60}")
        print(f"  TOTAL POTENTIAL SAVINGS:      ${summary['total_savings_per_month']:>8.2f}/mo")
        print(f"  ANNUALIZED:                   ${summary['total_savings_per_year']:>8.2f}/yr")
        print(f"{'='*60}\n")


# ── Entry point ───────────────────────────────────────────────────
def main() -> int:
    region  = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
    dry_run = os.environ.get("DRY_RUN", "true").lower() != "false"

    cleaner = AWSResourceCleaner(region=region, dry_run=dry_run)

    # Run all three cleanup operations
    cleaner.clean_unattached_ebs_volumes()
    cleaner.clean_unassociated_eips()
    cleaner.clean_old_amis(
        keep_count=3,
        name_prefix=os.environ.get("AMI_PREFIX", ""),  # e.g., "app-server"
    )

    # Write the consolidated report
    report = cleaner.generate_report("cleanup_report.json")

    # Return non-zero if live run deleted resources (useful for CI audit jobs)
    if not dry_run and report["summary"]["ebs_volumes_found"] > 0:
        return 0   # Deletions happened — success
    return 0


if __name__ == "__main__":
    sys.exit(main())

# Step 1: Always dry-run first — see what WOULD be deleted
DRY_RUN=true AWS_DEFAULT_REGION=ap-south-1 python aws_resource_cleanup.py

# Step 2: Review cleanup_report.json
cat cleanup_report.json | python -m json.tool | grep -E '"name"|"size_gb"|"monthly_cost"'

# Step 3: If the report looks correct, run for real
DRY_RUN=false AWS_DEFAULT_REGION=ap-south-1 python aws_resource_cleanup.py

# Limit AMI cleanup to a specific name prefix
AMI_PREFIX=app-server DRY_RUN=true python aws_resource_cleanup.py

@dataclass
class CleanupReport:
    ebs_volumes: list = field(default_factory=list)
    eips:        list = field(default_factory=list)
    amis:        list = field(default_factory=list)
    ebs_savings_per_month: float = 0.0

    @property
    def total_savings_per_month(self) -> float:
        return self.ebs_savings_per_month + self.eip_savings_per_month + self.ami_savings_per_month

def __init__(self, region: str = "us-east-1", dry_run: bool = True):
    self.ec2     = boto3.client("ec2", region_name=region)
    self.region  = region
    self.dry_run = dry_run
    self.report  = CleanupReport()
    if dry_run:
        logger.info("Running in DRY-RUN mode — no resources will be deleted")
    else:
        logger.warning("Running in LIVE mode — resources WILL be deleted!")

paginator = self.ec2.get_paginator("describe_volumes")
for page in paginator.paginate(
    Filters=[{"Name": "status", "Values": ["available"]}]
):
    for vol in page["Volumes"]:

days_old = (datetime.now(timezone.utc) - create_time).days
price_per_gb = self.EBS_PRICE_PER_GB_MONTH.get(vol_type, 0.08)
monthly_cost = size_gb * price_per_gb

if not self.dry_run:
    self.ec2.delete_volume(VolumeId=volume_id)
    entry["deleted"] = True
    self.report.ebs_savings_per_month += monthly_cost
else:
    self.report.ebs_savings_per_month += monthly_cost

response = self.ec2.describe_addresses()
for addr in response["Addresses"]:
    if "AssociationId" in addr:
        continue

monthly_cost = self.EIP_PRICE_PER_HOUR * 24 * 30.5
self.ec2.release_address(AllocationId=allocation_id)

response = self.ec2.describe_images(OwnerIds=["self"], Filters=filters)
all_amis = response["Images"]

groups: dict[str, list] = defaultdict(list)
for ami in all_amis:
    parts = ami_name.replace("_", "-").split("-")
    group_key = "-".join(parts[:2]) if len(parts) >= 2 else ami_name
    groups[group_key].append(ami)

amis_sorted = sorted(amis_in_group, key=lambda a: a["CreationDate"], reverse=True)
to_delete = amis_sorted[keep_count:]

snapshot_ids = [
    bdm["Ebs"]["SnapshotId"]
    for bdm in ami.get("BlockDeviceMappings", [])
    if "Ebs" in bdm and "SnapshotId" in bdm["Ebs"]
]
self.ec2.deregister_image(ImageId=image_id)
for snap_id in snapshot_ids:
    self.ec2.delete_snapshot(SnapshotId=snap_id)

report_data = {
    "generated_at": datetime.now(timezone.utc).isoformat(),
    ...
    "total_savings_per_year": round(self.report.total_savings_per_month * 12, 2),
}
with open(output_file, "w") as f:
    json.dump(report_data, f, indent=2, default=str)