Lambda 関数を利用して EBS スナップショットの差分サイズを計測し、その値を CloudWatch に出力してみた

import boto3
import os
import json
from datetime import datetime, timezone

def lambda_handler(event, context):
    """
    メイン関数: EBSボリュームの最新スナップショット間の差分サイズを計算し、CloudWatchに送信する

    Parameters:
    event (dict): Lambda呼び出しイベント。volumeIdまたはvolumeIdsパラメータを含むことができる
    context (object): Lambda実行コンテキスト

    Returns:
    dict: 処理結果を含むレスポンス
    """
    # 環境変数からボリュームIDのリストを取得
    volume_ids_str = os.environ.get('VOLUME_IDS', '')
    volume_ids = [vid.strip() for vid in volume_ids_str.split(',') if vid.strip()]

    if not volume_ids:
        # イベントからボリュームIDを取得する試み
        if 'volumeId' in event:
            volume_ids = [event['volumeId']]
        elif 'volumeIds' in event:
            volume_ids = event['volumeIds']
        else:
            raise ValueError("ボリュームIDが指定されていません")

    results = []

    for volume_id in volume_ids:
        try:
            # このボリュームの最新スナップショットペアを分析
            result = analyze_latest_snapshot_pair(volume_id)
            if result:
                results.append(result)
        except Exception as e:
            print(f"ボリューム {volume_id} の処理中にエラー: {str(e)}")

    return {
        'statusCode': 200,
        'body': json.dumps(results, default=str)
    }

def analyze_latest_snapshot_pair(volume_id):
    """
    指定されたボリュームの最新2つのスナップショット間の差分を分析する

    Parameters:
    volume_id (str): 分析対象のEBSボリュームID

    Returns:
    dict: 分析結果を含む辞書、またはスナップショットが不足している場合はNone
    """
    ec2_client = boto3.client('ec2')
    ebs_client = boto3.client('ebs')
    cloudwatch = boto3.client('cloudwatch')

    # このボリュームの全スナップショットを取得
    response = ec2_client.describe_snapshots(
        Filters=[
            {'Name': 'volume-id', 'Values': [volume_id]},
            {'Name': 'status', 'Values': ['completed']}
        ]
    )

    snapshots = response['Snapshots']
    if len(snapshots) < 2:
        print(f"ボリューム {volume_id} には比較可能なスナップショットが不足しています")
        return None

    # 作成日時でソート
    snapshots.sort(key=lambda x: x['StartTime'])

    # 最新の2つを取得
    older_snapshot = snapshots[-2]
    newer_snapshot = snapshots[-1]

    # 差分ブロックをカウント
    total_blocks = count_changed_blocks(
        ebs_client,
        older_snapshot['SnapshotId'],
        newer_snapshot['SnapshotId']
    )

    # 差分サイズを計算（1ブロック = 512 KiB）
    diff_size_kb = total_blocks * 512
    diff_size_mb = diff_size_kb / 1024
    diff_size_gb = diff_size_mb / 1024

    # 時間差を計算（時間単位）
    time_diff = (newer_snapshot['StartTime'] - older_snapshot['StartTime']).total_seconds() / 3600

    # 結果を作成
    result = {
        'volume_id': volume_id,
        'first_snapshot_id': older_snapshot['SnapshotId'],
        'second_snapshot_id': newer_snapshot['SnapshotId'],
        'first_date': older_snapshot['StartTime'],
        'second_date': newer_snapshot['StartTime'],
        'time_difference_hours': time_diff,
        'changed_blocks': total_blocks,
        'diff_size_kb': diff_size_kb,
        'diff_size_mb': diff_size_mb,
        'diff_size_gb': diff_size_gb
    }

    # CloudWatchにメトリクスを送信
    timestamp = datetime.now(timezone.utc)
    cloudwatch.put_metric_data(
        Namespace='EBS/SnapshotDiff',
        MetricData=[
            {
                'MetricName': 'DiffSizeGiB',
                'Value': diff_size_gb,
                'Unit': 'Gigabytes',
                'Dimensions': [
                    {'Name': 'VolumeId', 'Value': volume_id}
                ],
                'Timestamp': timestamp
            }
        ]
    )

    print(f"ボリューム {volume_id} の分析結果: 差分サイズ {diff_size_gb:.2f} GiB")

    return result

def count_changed_blocks(ebs_client, first_snapshot_id, second_snapshot_id):
    """
    2つのスナップショット間で変更されたブロック数をカウントする

    Parameters:
    ebs_client: boto3 EBSクライアント
    first_snapshot_id (str): 比較元のスナップショットID（古い方）
    second_snapshot_id (str): 比較先のスナップショットID（新しい方）

    Returns:
    int: 変更されたブロックの総数
    """
    total_blocks = 0
    next_token = None

    # ページネーション処理を行いながら全ての変更ブロックを取得
    while True:
        params = {
            'FirstSnapshotId': first_snapshot_id,
            'SecondSnapshotId': second_snapshot_id
        }

        if next_token:
            params['NextToken'] = next_token

        try:
            # EBS Direct APIを使用して変更されたブロックを取得
            response = ebs_client.list_changed_blocks(**params)

            # 現在のページの変更ブロック数をカウント
            if 'ChangedBlocks' in response:
                total_blocks += len(response['ChangedBlocks'])

            # 次のページがあるかチェック
            if 'NextToken' in response:
                next_token = response['NextToken']
            else:
                break

        except Exception as e:
            print(f"ブロック取得中にエラー: {str(e)}")
            break

    return total_blocks