EBS Snapshot Manager — Auto Backup & Retention Cleanup

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "ec2:DescribeInstances",
      "ec2:CreateSnapshot",
      "ec2:DescribeSnapshots",
      "ec2:DeleteSnapshot"
    ],
    "Resource": "*"
  }]
}

import boto3
from datetime import datetime, timezone, timedelta
import logging

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


class EBSSnapshotManager:
    def __init__(self, region="us-east-1", retention_days=30):
        """
        boto3.client("ec2") creates a low-level EC2 service client.
        region_name targets the correct AWS region.
        retention_days controls how long snapshots are kept.
        """
        self.ec2 = boto3.client("ec2", region_name=region)
        self.retention_days = retention_days

    # ── Step 1: Discover volumes ─────────────────────────────────────
    def get_running_instance_volumes(self) -> list[dict]:
        """
        get_paginator("describe_instances") returns a paginator object.
        Paginators automatically handle the NextToken loop so you never
        miss instances when you have more than 1,000 results.

        paginator.paginate() yields one page dict at a time.
        Each page["Reservations"] is a list of Reservation objects.
        Each Reservation["Instances"] is a list of Instance objects.

        BlockDeviceMappings lists every EBS volume attached to the instance.
        Each mapping has "DeviceName" (e.g., /dev/xvda) and "Ebs.VolumeId".
        """
        volumes = []
        paginator = self.ec2.get_paginator("describe_instances")

        for page in paginator.paginate(
            Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
        ):
            for reservation in page["Reservations"]:
                for instance in reservation["Instances"]:
                    instance_id = instance["InstanceId"]

                    # Extract the Name tag — default to instance ID if absent
                    instance_name = next(
                        (tag["Value"] for tag in instance.get("Tags", [])
                         if tag["Key"] == "Name"),
                        instance_id,
                    )

                    for mapping in instance.get("BlockDeviceMappings", []):
                        volumes.append({
                            "volume_id":     mapping["Ebs"]["VolumeId"],
                            "instance_id":   instance_id,
                            "instance_name": instance_name,
                            "device":        mapping["DeviceName"],  # e.g. /dev/xvda
                        })
        return volumes

    # ── Step 2: Create snapshots ─────────────────────────────────────
    def create_snapshots(self, volumes: list[dict]) -> list[str]:
        """
        create_snapshot() initiates an async snapshot.
        The snapshot state starts as "pending" and transitions to "completed".

        TagSpecifications lets you tag the snapshot at creation time
        (atomic — avoids a separate create_tags call).

        ResourceType: "snapshot" tells AWS these tags belong to the snapshot,
        not the source volume.
        """
        created = []
        timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d-%H%M")

        for vol in volumes:
            try:
                response = self.ec2.create_snapshot(
                    VolumeId=vol["volume_id"],
                    Description=(
                        f"Auto-backup {vol['instance_name']} "
                        f"{vol['device']} {timestamp}"
                    ),
                    TagSpecifications=[{
                        "ResourceType": "snapshot",
                        "Tags": [
                            {"Key": "Name",       "Value": f"auto-snap-{vol['instance_name']}-{timestamp}"},
                            {"Key": "AutoBackup", "Value": "true"},   # Used by cleanup filter
                            {"Key": "InstanceId", "Value": vol["instance_id"]},
                            {"Key": "CreatedAt",  "Value": timestamp},
                        ],
                    }],
                )
                snapshot_id = response["SnapshotId"]
                created.append(snapshot_id)
                logger.info(
                    f"Created snapshot {snapshot_id} for volume "
                    f"{vol['volume_id']} ({vol['instance_name']} {vol['device']})"
                )
            except Exception as e:
                logger.error(f"Failed to snapshot {vol['volume_id']}: {e}")

        return created

    # ── Step 3: Delete old snapshots ─────────────────────────────────
    def delete_old_snapshots(self) -> list[str]:
        """
        describe_snapshots with OwnerIds=["self"] only returns YOUR snapshots
        (not public ones — important to avoid accidental deletions).

        The tag filter Filters=[{"Name":"tag:AutoBackup","Values":["true"]}]
        ensures we only touch snapshots this script created, never manual ones.

        snapshot["StartTime"] is timezone-aware (UTC) so we compare against
        datetime.now(timezone.utc) — mixing naive/aware datetimes raises TypeError.
        """
        cutoff = datetime.now(timezone.utc) - timedelta(days=self.retention_days)
        deleted = []
        paginator = self.ec2.get_paginator("describe_snapshots")

        for page in paginator.paginate(
            Filters=[{"Name": "tag:AutoBackup", "Values": ["true"]}],
            OwnerIds=["self"],
        ):
            for snapshot in page["Snapshots"]:
                if snapshot["StartTime"] < cutoff:
                    try:
                        self.ec2.delete_snapshot(SnapshotId=snapshot["SnapshotId"])
                        deleted.append(snapshot["SnapshotId"])
                        age_days = (datetime.now(timezone.utc) - snapshot["StartTime"]).days
                        logger.info(
                            f"Deleted snapshot {snapshot['SnapshotId']} "
                            f"(age: {age_days} days)"
                        )
                    except Exception as e:
                        # Snapshot may be in use by an AMI — skip gracefully
                        logger.error(
                            f"Could not delete {snapshot['SnapshotId']}: {e}"
                        )

        return deleted

    # ── Orchestrator ──────────────────────────────────────────────────
    def run(self) -> dict:
        logger.info("Starting EBS snapshot backup cycle...")
        volumes = self.get_running_instance_volumes()
        logger.info(f"Found {len(volumes)} volumes across running instances")

        created = self.create_snapshots(volumes)
        deleted = self.delete_old_snapshots()

        result = {
            "volumes_found":      len(volumes),
            "snapshots_created":  len(created),
            "snapshots_deleted":  len(deleted),
            "created_ids":        created,
            "deleted_ids":        deleted,
        }
        logger.info(f"Backup cycle complete: {result}")
        return result


# ── Lambda entry point ────────────────────────────────────────────
def lambda_handler(event, context):
    """Deploy as Lambda + EventBridge cron(0 1 * * ? *) to run at 1 AM UTC daily."""
    manager = EBSSnapshotManager(retention_days=30)
    return manager.run()


# ── Local / CLI entry point ───────────────────────────────────────
if __name__ == "__main__":
    manager = EBSSnapshotManager(region="ap-south-1", retention_days=30)
    result = manager.run()
    print(result)

# 1. Zip the script
zip ebs_backup.zip ebs_snapshot_manager.py

# 2. Create Lambda function
aws lambda create-function \
  --function-name EBSSnapshotManager \
  --runtime python3.12 \
  --role arn:aws:iam::123456789012:role/LambdaEC2BackupRole \
  --handler ebs_snapshot_manager.lambda_handler \
  --zip-file fileb://ebs_backup.zip \
  --timeout 300

# 3. Add daily EventBridge trigger (1 AM UTC)
aws events put-rule \
  --name DailyEBSBackup \
  --schedule-expression "cron(0 1 * * ? *)" \
  --state ENABLED

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

def __init__(self, region="us-east-1", retention_days=30):
    self.ec2 = boto3.client("ec2", region_name=region)
    self.retention_days = retention_days

paginator = self.ec2.get_paginator("describe_instances")

for page in paginator.paginate(
    Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
):

for reservation in page["Reservations"]:
    for instance in reservation["Instances"]:
        instance_id = instance["InstanceId"]

instance_name = next(
    (tag["Value"] for tag in instance.get("Tags", []) if tag["Key"] == "Name"),
    instance_id,
)

for mapping in instance.get("BlockDeviceMappings", []):
    volumes.append({
        "volume_id": mapping["Ebs"]["VolumeId"],
        "device":    mapping["DeviceName"],
    })

response = self.ec2.create_snapshot(
    VolumeId=vol["volume_id"],
    Description=f"Auto-backup {vol['instance_name']} {vol['device']} {timestamp}",
    TagSpecifications=[{
        "ResourceType": "snapshot",
        "Tags": [
            {"Key": "AutoBackup", "Value": "true"},
            ...
        ],
    }],
)

cutoff = datetime.now(timezone.utc) - timedelta(days=self.retention_days)

paginator = self.ec2.get_paginator("describe_snapshots")
for page in paginator.paginate(
    Filters=[{"Name": "tag:AutoBackup", "Values": ["true"]}],
    OwnerIds=["self"],
):

if snapshot["StartTime"] < cutoff:
    self.ec2.delete_snapshot(SnapshotId=snapshot["SnapshotId"])

volumes = self.get_running_instance_volumes()
created = self.create_snapshots(volumes)
deleted = self.delete_old_snapshots()

def lambda_handler(event, context):
    manager = EBSSnapshotManager(retention_days=30)
    return manager.run()

if __name__ == "__main__":
    manager = EBSSnapshotManager(region="ap-south-1", retention_days=30)