Create snapshots for EBS volumes to backup EC2 instances

EC2Backup.py

#!/usr/bin/env python3
# coding: utf-8

"""
EC2Backup
A tool for encrypting EC2 volumes
Inspried by https://github.com/jbrt/ec2cryptomatic
"""

import argparse
import logging
import sys
import boto3
from botocore.exceptions import ClientError, EndpointConnectionError

__version__ = '0.0.1'

# Define the global logger
LOGGER = logging.getLogger('ec2-backup')
LOGGER.setLevel(logging.DEBUG)
STREAM_HANDLER = logging.StreamHandler()
STREAM_HANDLER.setLevel(logging.DEBUG)
LOGGER.addHandler(STREAM_HANDLER)

# Constants
MAX_RETRIES = 360
DELAY_RETRY = 60


class EC2Backup:
    """ Backup EBS volumes from an EC2 instance """

    def __init__(self, region: str, instance: str):
        """
        Initialization
        :param region: (str) the AWS region where the instance is
        :param instance: (str) one instance-id
        """
        self._ec2_client = boto3.client('ec2', region_name=region)
        self._ec2_resource = boto3.resource('ec2', region_name=region)
        self._region = region
        self._instance = self._ec2_resource.Instance(id=instance)

        # Volumes
        self._snapshot = None
        self._volume = None

        # Waiters
        self._wait_snapshot = self._ec2_client.get_waiter('snapshot_completed')

        # Waiters retries values
        self._wait_snapshot.config.max_attempts = MAX_RETRIES
        self._wait_snapshot.config.delay = DELAY_RETRY

        # Do some pre-check : instances must exists and be stopped
        self._instance_is_exists()
        self._instance_is_stopped()

    def _instance_is_exists(self) -> None:
        """
        Check if instance exists
        :return: None
        :except: ClientError
        """
        try:
            self._ec2_client.describe_instances(InstanceIds=[self._instance.id])
        except ClientError:
            raise

    def _instance_is_stopped(self) -> None:
        """
        Check if instance is stopped
        :return: None
        :except: TypeError
        """
        if self._instance.state['Name'] != 'stopped':
            raise TypeError('Instance still running ! please stop it.')

    def _take_snapshot(self, device, tags):
        """
        Take the snapshot from the volume
        :param device: EBS device to backup
        :param device: list of tags to added on snapshot
        """
        LOGGER.info(f'-- Take a snapshot for volume {device.id}')
        snapshot = device.create_snapshot(Description=f'backup snap of {device.id}',
            TagSpecifications=[{
                'ResourceType': 'snapshot',
                'Tags': tags
                }])
        self._wait_snapshot.wait(SnapshotIds=[snapshot.id])
        LOGGER.info(f'-- Snapshot {snapshot.id} done')

        return snapshot

    def start_backup(self) -> None:
        """
        Launch backup process
        :return: None
        """

        LOGGER.info(f'\nStart to backup instance {self._instance.id}')

        if self._instance.tags is not None:
            for tag in self._instance.tags:
                if tag['Key'] == 'Name':
                    instance_name = tag['Value']
        else:
            instance_name = 'None'

        for device in self._instance.block_device_mappings:
            if 'Ebs' not in device:
                msg = f'{self._instance.id}: Skip {device["VolumeId"]} not an EBS device'
                LOGGER.warning(msg)
                continue

        for device in self._instance.volumes.all():

            LOGGER.info(f'- Let\'s backup volume {device.id}')

            tags = [{
                'Key': 'InstanceName',
                'Value': instance_name
            },{
                'Key': 'InstanceID',
                'Value': self._instance.id
            },{
                'Key': 'Volume',
                'Value': device.id
            },{
                'Key': 'Device',
                'Value': device.attachments[0]['Device']
            },]

            # take a snapshot from the original device
            self._snapshot = self._take_snapshot(device=device, tags=tags)


def main(args: argparse.Namespace) -> None:
    """
    Main program
    :param args: arguments from CLI
    :return: None
    """

    for instance in args.instances:
        try:
            EC2Backup(region=args.region, instance=instance).start_backup()

        except (EndpointConnectionError, ValueError) as error:
            LOGGER.error(f'Problem with your AWS region ? ({error})')
            sys.exit(1)

        except (ClientError, TypeError) as error:
            LOGGER.error(f'Problem with the instance ({error})')
            continue


def parse_arguments() -> argparse.Namespace:
    """
    Parse arguments from CLI
    :returns: argparse.Namespace
    """
    description = 'EC2Backup - Encrypt EBS volumes from EC2 instances'
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument('-r', '--region', help='AWS Region', required=True)
    parser.add_argument('-i', '--instances', nargs='+',
                        help='Instance to backup', required=True)
    parser.add_argument('-v', '--version', action='version', version=__version__)
    return parser.parse_args()


if __name__ == '__main__':
    main(parse_arguments())