特定の保持期間よりも長く保持されている EBS スナップショットを棚卸ししてみた

#EBS
#AWS
たかくに
2025.02.15
こんにちは！クラウド事業本部コンサルティング部のたかくに（@takakuni_）です。
みなさん、 EBS のスナップショットを、きちんと棚卸できていますでしょうか。
1 日 1 回バックアップを取得し、X世代を保持するなど、、ライフサイクルを定め運用できていれば OK です。
まれに、ライフサイクルが定まっておらず、バックアップを意図せず保持し続けていた環境を見かけます。EBS スナップショットの保管料金には 2 つの層があり、保持し続けていると毎月地味に痛いです。
スタンダード層	USD 0.05/1 か月あたりの GB
アーカイブ層	USD 0.0125/1 か月あたりの GB
https://aws.amazon.com/jp/ebs/pricing/
そこで今回は特定期間以上に保持し続けられているスナップショットを洗い出してみたいと思います。
 作成したコード今回は定期点検の観点で Lambda で動かす想定としたため、boto3 で実装してみました。全文は以下になります。
main.py
import os
import boto3
import botocore
import logging
from datetime import datetime, timezone
from botocore.exceptions import ClientError

# ロガーの設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 環境変数からのデフォルト値設定
SNAPSHOT_AGE_THRESHOLD_DAYS = int(os.environ.get('SNAPSHOT_AGE_THRESHOLD_DAYS', 30))
MAX_SNAPSHOTS_TO_RETRIEVE = int(os.environ.get('MAX_SNAPSHOTS_TO_RETRIEVE', 100))

def identify_old_ebs_snapshots(
    client,
    age_threshold_days=SNAPSHOT_AGE_THRESHOLD_DAYS
):
    """
    古いEBSスナップショットを特定する関数
    
    Args:
        client (boto3.client): EC2クライアント
        age_threshold_days (int): スナップショットの経過日数のしきい値

    Returns:
        stale_snapshots: 古いスナップショットの情報のリスト
    """
    stale_snapshots = []
    current_time = datetime.now(timezone.utc)

    try:
        paginator = client.get_paginator('describe_snapshots')
        page_iterator = paginator.paginate(OwnerIds=['self'])
        
        for page in page_iterator:
            for snapshot in page['Snapshots']:
                snapshot_id = snapshot['SnapshotId']
                snapshot_start_time = snapshot['StartTime']
                
                # タグの処理
                snapshot_tags = {}
                if 'Tags' in snapshot:
                    snapshot_tags = {tag['Key']: tag['Value'] for tag in snapshot['Tags']}
                
                # 経過日数の計算
                snapshot_age = current_time - snapshot_start_time
                
                if snapshot_age.days >= age_threshold_days:
                    logger.info(f"Found old snapshot: {snapshot_id} ({snapshot_age.days} days old)")
                    stale_snapshots.append({
                        'snapshot_id': snapshot_id,
                        'snapshot_age_days': snapshot_age.days,
                        'snapshot_tags': snapshot_tags,
                        'start_time': snapshot_start_time.isoformat(),
                        'size_gb': snapshot.get('VolumeSize', 0),
                        'encrypted': snapshot.get('Encrypted', False)
                    })
        return stale_snapshots
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code in ['AccessDenied', 'AllAccessDisabled']:
            logger.warning(f"Access denied to retrieve snapshots: {str(e)}")
            return None
        logger.error(f"AWS API error to retrieve snapshots: {str(e)}")
        return None
    except Exception as e:
        logger.error(f"Error in identify_old_ebs_snapshots: {str(e)}")
        raise

def lambda_handler(event, context):
    """
    Lambda関数のメインハンドラー
    
    Args:
        event: Lambda イベントデータ
        context: Lambda コンテキスト
    
    Returns:
        stale_snapshots_results: 古いスナップショットの情報のリスト
    """
    try:
        # EC2クライアントの作成
        client = boto3.client(
            'ec2',
        )

        # 古いスナップショットの特定
        stale_snapshots_results = identify_old_ebs_snapshots(
            client, 
            age_threshold_days=SNAPSHOT_AGE_THRESHOLD_DAYS
        )

        # 古い順にソート
        stale_snapshots_results.sort(key=lambda x: x['snapshot_age_days'], reverse=True)
        logger.info(f"Found {len(stale_snapshots_results)} stale snapshots")

        # 最大数まで結果を返却
        return {
            'stale_snapshots': stale_snapshots_results[:MAX_SNAPSHOTS_TO_RETRIEVE]
        }
    except ValueError as e:
        logger.error(f"Validation error: {str(e)}")
        raise
    except botocore.exceptions.BotoCoreError as e:
        logger.error(f"AWS API error: {str(e)}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        raise
    finally:
        # セキュリティのため認証情報を削除
        if 'credential' in locals():
            del credential
 paginator の実装EBS スナップショットがアカウント内にどれほどあってもいいように、 paginator を利用して describe_snapshots の処理を実装しました。
paginator を使えば、自分でページめくりの実装をしなくて済むため、非常に少ないコードで全件取得できます。 大変便利ですし、学びになりました。
main.py
def identify_old_ebs_snapshots(
    client,
    age_threshold_days=SNAPSHOT_AGE_THRESHOLD_DAYS
):
    """
    古いEBSスナップショットを特定する関数
    
    Args:
        client (boto3.client): EC2クライアント
        age_threshold_days (int): スナップショットの経過日数のしきい値

    Returns:
        stale_snapshots: 古いスナップショットの情報のリスト
    """
    stale_snapshots = []
    current_time = datetime.now(timezone.utc)

    try:
        paginator = client.get_paginator('describe_snapshots')
        page_iterator = paginator.paginate(OwnerIds=['self'])
        
        for page in page_iterator:
            for snapshot in page['Snapshots']:
                snapshot_id = snapshot['SnapshotId']
                snapshot_start_time = snapshot['StartTime']
                
                # タグの処理
                snapshot_tags = {}
                if 'Tags' in snapshot:
                    snapshot_tags = {tag['Key']: tag['Value'] for tag in snapshot['Tags']}
                
                # 経過日数の計算
                snapshot_age = current_time - snapshot_start_time
                
                if snapshot_age.days >= age_threshold_days:
                    logger.info(f"Found old snapshot: {snapshot_id} ({snapshot_age.days} days old)")
                    stale_snapshots.append({
                        'snapshot_id': snapshot_id,
                        'snapshot_age_days': snapshot_age.days,
                        'snapshot_tags': snapshot_tags,
                        'start_time': snapshot_start_time.isoformat(),
                        'size_gb': snapshot.get('VolumeSize', 0),
                        'encrypted': snapshot.get('Encrypted', False)
                    })
        return stale_snapshots
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code in ['AccessDenied', 'AllAccessDisabled']:
            logger.warning(f"Access denied to retrieve snapshots: {str(e)}")
            return None
        logger.error(f"AWS API error to retrieve snapshots: {str(e)}")
        return None
    except Exception as e:
        logger.error(f"Error in identify_old_ebs_snapshots: {str(e)}")
        raise
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/get_paginator.html
 保持期間の計算describe_snapshots のレスポンスを見ると、 StartTime や CompletionTime などは、datetime 型で返ってくることが確認できます。
{
    'NextToken': 'string',
    'Snapshots': [
        {
            'OwnerAlias': 'string',
            'OutpostArn': 'string',
            'Tags': [
                {
                    'Key': 'string',
                    'Value': 'string'
                },
            ],
            'StorageTier': 'archive'|'standard',
            'RestoreExpiryTime': datetime(2015, 1, 1),
            'SseType': 'sse-ebs'|'sse-kms'|'none',
            'AvailabilityZone': 'string',
            'TransferType': 'time-based'|'standard',
            'CompletionDurationMinutes': 123,
            'CompletionTime': datetime(2015, 1, 1),
            'FullSnapshotSizeInBytes': 123,
            'SnapshotId': 'string',
            'VolumeId': 'string',
            'State': 'pending'|'completed'|'error'|'recoverable'|'recovering',
            'StateMessage': 'string',
            'StartTime': datetime(2015, 1, 1),
            'Progress': 'string',
            'OwnerId': 'string',
            'Description': 'string',
            'VolumeSize': 123,
            'Encrypted': True|False,
            'KmsKeyId': 'string',
            'DataEncryptionKeyId': 'string'
        },
    ]
}
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ec2/client/describe_snapshots.html
この結果を利用して、snapshot_age = current_time - snapshot_start_time のような形で経過日数を計算しました。保持期間の内容を変更しやすいよう環境変数から変更できるようにもしてみました。
main.py
# 環境変数からのデフォルト値設定
SNAPSHOT_AGE_THRESHOLD_DAYS = int(os.environ.get('SNAPSHOT_AGE_THRESHOLD_DAYS', 30))
MAX_SNAPSHOTS_TO_RETRIEVE = int(os.environ.get('MAX_SNAPSHOTS_TO_RETRIEVE', 100))

def identify_old_ebs_snapshots(
    client,
    age_threshold_days=SNAPSHOT_AGE_THRESHOLD_DAYS
):
    """
    古いEBSスナップショットを特定する関数
    
    Args:
        client (boto3.client): EC2クライアント
        age_threshold_days (int): スナップショットの経過日数のしきい値

    Returns:
        stale_snapshots: 古いスナップショットの情報のリスト
    """
    stale_snapshots = []
    current_time = datetime.now(timezone.utc)

    try:
        paginator = client.get_paginator('describe_snapshots')
        page_iterator = paginator.paginate(OwnerIds=['self'])
        
        for page in page_iterator:
            for snapshot in page['Snapshots']:
                snapshot_id = snapshot['SnapshotId']
                snapshot_start_time = snapshot['StartTime']
                
                # タグの処理
                snapshot_tags = {}
                if 'Tags' in snapshot:
                    snapshot_tags = {tag['Key']: tag['Value'] for tag in snapshot['Tags']}
                
                # 経過日数の計算
                snapshot_age = current_time - snapshot_start_time
                
                if snapshot_age.days >= age_threshold_days:
                    logger.info(f"Found old snapshot: {snapshot_id} ({snapshot_age.days} days old)")
                    stale_snapshots.append({
                        'snapshot_id': snapshot_id,
                        'snapshot_age_days': snapshot_age.days,
                        'snapshot_tags': snapshot_tags,
                        'start_time': snapshot_start_time.isoformat(),
                        'size_gb': snapshot.get('VolumeSize', 0),
                        'encrypted': snapshot.get('Encrypted', False)
                    })
        return stale_snapshots
    except ClientError as e:
        error_code = e.response['Error']['Code']
        if error_code in ['AccessDenied', 'AllAccessDisabled']:
            logger.warning(f"Access denied to retrieve snapshots: {str(e)}")
            return None
        logger.error(f"AWS API error to retrieve snapshots: {str(e)}")
        return None
    except Exception as e:
        logger.error(f"Error in identify_old_ebs_snapshots: {str(e)}")
        raise
 やってみた実際に実行してみました。389 日と 295 日と長い日数保持し続けられているスナップショットが出てきました。
{
  "stale_snapshots": [
    {
      "snapshot_id": "snap-XXXXXXXXXXXXXXXXX",
      "snapshot_age_days": 389,
      "snapshot_tags": {
        "hoge": "hoge-server-1"
      },
      "start_time": "2024-01-23T10:44:29.968000+00:00", 
      "size_gb": 200,
      "encrypted": true
    },
    {
      "snapshot_id": "snap-YYYYYYYYYYYYYYYYY",
      "snapshot_age_days": 295,
      "snapshot_tags": {},
      "start_time": "2024-04-26T01:21:40.106000+00:00",
      "size_gb": 8,
      "encrypted": false
    }
  ]
}
 まとめ以上、「特定の保持期間よりも長く保持されている EBS スナップショットを棚卸ししてみた」でした。
今回のような、優先度が低いもののやっておいた方が、良いことってたくさんあると思うのですが、ぜひ大掃除期間を設けてコスト最適化にも取り組んでいただけると幸いです。
このブログがどなたかの参考になれば幸いです。クラウド事業本部コンサルティング部のたかくに（@takakuni_）でした！