AWS EBS – スナップショットの自動取得＆世代管理

概要

AWS EBSのスナップショットの「自動取得」と「指定の世代までの管理」を実施できる手順について記載しています。

手順

• タグの設定
• 関数の作成
• テスト・確認
• 補足

タグの設定

1.インスタンスに「名前」と「世代管理用」のタグを設定します。

1-1.[ EC2 > インスタンス > インスタンス ] を表示します。

1-2.『タグ』タブを表示し【 タグの追加/編集 】をクリックします。

1-3.表示された『タグの追加/編集』画面で【 タグの作成 】をクリックすると行が追加されます。

1-4.適当な値を入力して【 保存 】をクリックします。

※「Backup-Generation」キーで指定している値が「世代管理」数となります。

関数の作成

1.Lambdaで関数を作成します。

1-1.[ Lambda ]を表示します。

1-2.【 関数の作成 】をクリックします。

2.設計図の選択します。

2-1.[ lambda-canary ]を検索します。

2-2.表示された[ lambda-canary ]をクリックします。

3.「基本的な情報」を設定します。

3-1.『名前』を入力し、『ロール』から [ カスタムロールの作成 ] を選択します。

3-2.『ロールの作成』タブが作成されます。

3-3.『ロール名』に値を入力し、ポリシードキュメントの【 編集 】をクリックします。

3-4.ダイアログが表示されます。【 OK 】をクリックします。

3-5.ポリシードキュメントが編集可能となったら以下のコードに差し替え後、【 許可 】をクリックします。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:*:*:*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ec2:DescribeInstances",
                "ec2:DescribeSnapshots",
                "ec2:CreateSnapshot",
                "ec2:DeleteSnapshot"
            ],
            "Resource": [
                "*"
            ]
        }
    ]
}

『Lambda』画面に戻り、既存のロールに先ほど作成したロールが適用されます。

4.「cloudwatch-events」を設定します。

4-1.以下の値を設定します。

※スケジュール式：日本時間 00:00 に実行

5.「Lambda関数コード」を設定します。

※環境変数をすべて削除します。

6.関数を作成します。

【 関数の作成 】をクリックします。

7.関数コードを設定します。

7-1.「設定」タブを選択し、関数コードを以下のコードと差し替えて【 保存 】をクリックします。

import boto3
import collections
import time
from botocore.client import ClientError

ec2 = boto3.client('ec2')


def lambda_handler(event, context):
    descriptions = create_snapshots()
    delete_old_snapshots(descriptions)


def create_snapshots():
    instances = get_instances(['Backup-Generation'])

    descriptions = {}

    for i in instances:
        tags = {t['Key']: t['Value'] for t in i['Tags']}
        generation = int(tags.get('Backup-Generation', 0))

        if generation < 1:
            continue

        for b in i['BlockDeviceMappings']:
            if b.get('Ebs') is None:
                continue

            volume_id = b['Ebs']['VolumeId']
            description = volume_id if tags.get('Name') is '' else '%s(%s)' % (volume_id, tags['Name'])
            description = 'Auto Snapshot ' + description

            snapshot = _create_snapshot(volume_id, description)
            print 'create snapshot %s(%s)' % (snapshot['SnapshotId'], description)

            descriptions[description] = generation

    return descriptions


def get_instances(tag_names):
    reservations = ec2.describe_instances(
        Filters=[
            {
                'Name': 'tag-key',
                'Values': tag_names
            }
        ]
    )['Reservations']

    return sum([
        [i for i in r['Instances']]
        for r in reservations
    ], [])


def delete_old_snapshots(descriptions):
    snapshots_descriptions = get_snapshots_descriptions(descriptions.keys())

    for description, snapshots in snapshots_descriptions.items():
        delete_count = len(snapshots) - descriptions[description]

        if delete_count <= 0:
            continue

        snapshots.sort(key=lambda x: x['StartTime'])

        old_snapshots = snapshots[0:delete_count]

        for s in old_snapshots:
            _delete_snapshot(s['SnapshotId'])
            print 'delete snapshot %s(%s)' % (s['SnapshotId'], s['Description'])


def get_snapshots_descriptions(descriptions):
    snapshots = ec2.describe_snapshots(
        Filters=[
            {
                'Name': 'description',
                'Values': descriptions,
            }
        ]
    )['Snapshots']

    groups = collections.defaultdict(lambda: [])
    {groups[s['Description']].append(s) for s in snapshots}

    return groups


def _create_snapshot(id, description):
    for i in range(1, 3):
        try:
            return ec2.create_snapshot(VolumeId=id, Description=description)
        except ClientError as e:
            print str(e)
        time.sleep(1)
    raise Exception('cannot create snapshot ' + description)


def _delete_snapshot(id):
    for i in range(1, 3):
        try:
            return ec2.delete_snapshot(SnapshotId=id)
        except ClientError as e:
            print str(e)
        time.sleep(1)
    raise Exception('cannot delete snapshot ' + id)

テスト・確認

1.テストを実施します。

1-1.「テストイベントの設定」を選択します。

1-2.テストイベントを設定します。

1-3.「イベントテンプレート」リストから [ Schedule Event ] を選択します。

1-4.イベント名を入力して【 作成 】をクリックします。

※コードはそのまま

1-5. 作成したイベントを選択してテスト をクリックします。

※『 実行結果 』が【 成功 】と表示される事を確認します。

2.スナップショットの確認します。

2-1.スナップショットが作成されている事が確認できます。

2-2.世代管理の確認します。

テストを複数回実行し、タグ「Backup-Generation」で設定した値のみ最新のスナップショットが保存されている事が確認できます。

補足

CloudWatchでアラームを設定する事で、スナップショットの作成に失敗した場合にメールを送信することができます。

AWS CloudWatch - スナップショットの作成失敗時にアラート送信