ElastiCacheのバックアップについての備忘

ElastiCacheのバックアップ方法

イベントについて

また、バックアップに起因した処理を行うためにCloudTrailのイベント履歴を確認したのだが、両者には以下の違いがあった 手動バックアップ イベント名「CreateSnapshot」がCloudTrailに履歴として表示される 自動バックアップ イベントがCloudTrailに履歴として表示されることはない

python boto3について

python boto3でスナップショットの情報を取得すると以下のようになる (スナップショットを作成日でソートしたかったため調査、レスポンスとして「CacheClusterCreateTime」が得られるため可能だった)

リクエスト構文

response = client.describe_snapshots(
    ReplicationGroupId='string',
    CacheClusterId='string',
    SnapshotName='string',
    SnapshotSource='string',
    Marker='string',
    MaxRecords=123,
    ShowNodeGroupConfig=True|False
)

レスポンス構文

{
    'Marker': 'string',
    'Snapshots': [
        {
            'SnapshotName': 'string',
            'ReplicationGroupId': 'string',
            'ReplicationGroupDescription': 'string',
            'CacheClusterId': 'string',
            'SnapshotStatus': 'string',
            'SnapshotSource': 'string',
            'CacheNodeType': 'string',
            'Engine': 'string',
            'EngineVersion': 'string',
            'NumCacheNodes': 123,
            'PreferredAvailabilityZone': 'string',
            'PreferredOutpostArn': 'string',
            'CacheClusterCreateTime': datetime(2015, 1, 1),
            'PreferredMaintenanceWindow': 'string',
            'TopicArn': 'string',
            'Port': 123,
            'CacheParameterGroupName': 'string',
            'CacheSubnetGroupName': 'string',
            'VpcId': 'string',
            'AutoMinorVersionUpgrade': True|False,
            'SnapshotRetentionLimit': 123,
            'SnapshotWindow': 'string',
            'NumNodeGroups': 123,
            'AutomaticFailover': 'enabled'|'disabled'|'enabling'|'disabling',
            'NodeSnapshots': [
                {
                    'CacheClusterId': 'string',
                    'NodeGroupId': 'string',
                    'CacheNodeId': 'string',
                    'NodeGroupConfiguration': {
                        'NodeGroupId': 'string',
                        'Slots': 'string',
                        'ReplicaCount': 123,
                        'PrimaryAvailabilityZone': 'string',
                        'ReplicaAvailabilityZones': [
                            'string',
                        ],
                        'PrimaryOutpostArn': 'string',
                        'ReplicaOutpostArns': [
                            'string',
                        ]
                    },
                    'CacheSize': 'string',
                    'CacheNodeCreateTime': datetime(2015, 1, 1),
                    'SnapshotCreateTime': datetime(2015, 1, 1)
                },
            ],
            'KmsKeyId': 'string',
            'ARN': 'string',
            'DataTiering': 'enabled'|'disabled'
        },
    ]
}

※AWS Boto3 1.34.9 documentationから引用 AWS Boto3 1.34.9 documentation

補足

ElastiCacheダッシュボードの「バックアップ」に格納されたスナップショットのうち、作成日が最新で、当日であるもののコピーをS3に格納するLambda

import boto3
import datetime
import os
import re
import random

client = boto3.client('elasticache')

def lambda_handler(event, context):
    DIFF_JST_FROM_UTC = 9
    todays = datetime.datetime.today().date()
    source_name = None

    try:
        response = client.describe_snapshots(
            ShowNodeGroupConfig=True
        )

    except Exception as e:
        error_message = f"エラーが発生しました: {str(e)}"
        print(error_message)
        raise Exception(error_message)

    if 'Snapshots' in response and response['Snapshots']:
        snapshots = sorted(response['Snapshots'], key=lambda x: x['CacheClusterCreateTime'])
        latest = snapshots[-1]

        if latest['SnapshotStatus'] == 'available':        
            snapshot_date = latest['CacheClusterCreateTime'].date()
            if snapshot_date == todays:
                source_name = latest['SnapshotName']

        try:
            if source_name is not None: 
                response = client.copy_snapshot(
                    SourceSnapshotName=source_name, 
                    TargetSnapshotName=source_name + str(random.randint(100000,999999)),
                    TargetBucket=os.getenv("TARGETBUCKET")
                )
                print('スナップショット:'+source_name+'をS3にコピーしました')

        except Exception as e:
            print(source_name)
            print("2")
            error_message = f"エラーが発生しました: {str(e)}"
            print(error_message)
            raise Exception(error_message)

    else:
        error_message = 'source_nameが見つかりません'
        raise Exception(error_message)

環境変数 キー：TARGETBUCKET 値：任意の格納先S3バケット名