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)