lambda_function.py

"""
AWS Lambda function to create daily EBS snapshots and prune expired ones.

Iterates over all configured regions and creates a snapshot of every EBS
volume attached to an instance carrying the configured backup tag. Each
snapshot is tagged with a 'DeleteOn' date; snapshots whose 'DeleteOn' date
matches today are deleted.
"""

import logging
import os
from datetime import date, timedelta

import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

BACKUP_TAG = os.environ.get("EC2_INSTANCE_TAG", "Backup")
RETENTION_DAYS = int(os.environ.get("RETENTION_DAYS", "10"))
REGIONS = [r.strip() for r in os.environ.get("REGIONS", "eu-central-1").split(",") if r.strip()]


def find_instance_name(instance: dict) -> str:
    """
    Return the value of the 'Name' tag of an instance or an empty string.
    """

    for tag in instance.get("Tags", []):
        if tag["Key"] == "Name":
            return tag["Value"]
    return ""


def find_backup_instances(client) -> list[dict]:
    """
    Return all instances carrying the configured backup tag.
    """

    instances = []
    paginator = client.get_paginator("describe_instances")
    for page in paginator.paginate(Filters=[{"Name": "tag-key", "Values": [BACKUP_TAG]}]):
        for reservation in page["Reservations"]:
            instances.extend(reservation["Instances"])
    return instances


def create_snapshots(client, instances: list[dict], today: date) -> int:
    """
    Create a snapshot of every EBS volume of the given instances.

    Returns the number of snapshots successfully created.
    """

    delete_date = today + timedelta(days=RETENTION_DAYS)
    created = 0

    for instance in instances:
        instance_id = instance["InstanceId"]
        instance_name = find_instance_name(instance)

        for device in instance["BlockDeviceMappings"]:
            if not device.get("Ebs"):
                # skip non EBS volumes
                continue

            vol_id = device["Ebs"]["VolumeId"]
            device_name = device["DeviceName"]
            logger.info("Found EBS volume %s on instance %s", vol_id, instance_id)

            try:
                client.create_snapshot(
                    Description=f"Snapshot of {device_name} from {instance_id} ({instance_name})",
                    VolumeId=vol_id,
                    TagSpecifications=[
                        {
                            "ResourceType": "snapshot",
                            "Tags": [
                                {"Key": "DeleteOn", "Value": delete_date.strftime("%Y-%m-%d")},
                                {"Key": "Name", "Value": f"{vol_id}_{today.strftime('%Y-%m-%d')}"},
                                {"Key": "DeviceName", "Value": device_name},
                            ],
                        },
                    ],
                )
                created += 1
            except ClientError as exc:
                logger.error("Could not create snapshot of volume %s: %s", vol_id, exc)

    return created


def delete_expired_snapshots(client, today: date) -> int:
    """
    Delete all snapshots owned by this account whose 'DeleteOn' tag matches today.

    Returns the number of snapshots successfully deleted.
    """

    filters = [
        {"Name": "tag-key", "Values": ["DeleteOn"]},
        {"Name": "tag-value", "Values": [today.strftime("%Y-%m-%d")]},
    ]
    deleted = 0

    paginator = client.get_paginator("describe_snapshots")
    for page in paginator.paginate(OwnerIds=["self"], Filters=filters):
        for snapshot in page["Snapshots"]:
            snapshot_id = snapshot["SnapshotId"]
            try:
                client.delete_snapshot(SnapshotId=snapshot_id)
                logger.info("Deleted snapshot %s", snapshot_id)
                deleted += 1
            except ClientError as exc:
                logger.error("Could not delete snapshot %s: %s", snapshot_id, exc)

    return deleted


def lambda_handler(event, context):  # pylint: disable=unused-argument
    """
    Entrypoint for AWS Lambda.
    """

    today = date.today()

    for region in REGIONS:
        logger.info("Processing region %s", region)
        client = boto3.client("ec2", region_name=region)

        instances = find_backup_instances(client)
        created = create_snapshots(client, instances, today)
        deleted = delete_expired_snapshots(client, today)

        logger.info("Region %s: created %d snapshot(s), deleted %d expired snapshot(s)", region, created, deleted)