CloudFrontへのリクエスト状況をLineへ通知する

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "GetCloudWatchData",
      "Effect": "Allow",
      "Action": ["cloudwatch:GetMetricStatistics", "cloudwatch:ListMetrics"],
      "Resource": "*"
    },
    {
      "Sid": "GetSSMParameter",
      "Effect": "Allow",
      "Action": ["kms:Decrypt", "ssm:GetParameter"],
      "Resource": [
        "arn:aws:kms:ap-northeast-1:アカウントID:key/alias/aws/ssm",
        "arn:aws:ssm:us-east-1:アカウントID:parameter/*"
      ]
    }
  ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:us-east-1:アカウントID:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:us-east-1:アカウントID:log-group:/aws/lambda/関数名:*"
            ]
        }
    ]
}

import boto3
import json
import requests
from datetime import datetime, timedelta, timezone

# SSMクライアントの作成
ssm = boto3.client('ssm')
cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')

# SSM Parameter StoreからLINEアクセストークンとユーザーIDを取得
LINE_ACCESS_TOKEN = ssm.get_parameter(Name='LINE_ACCESS_TOKEN', WithDecryption=True)['Parameter']['Value']
LINE_USER_ID = ssm.get_parameter(Name='LINE_USER_ID', WithDecryption=True)['Parameter']['Value']

# SSM Parameter StoreからCloudFrontディストリビューションIDを取得
DISTRIBUTION_ID = ssm.get_parameter(Name='DISTRIBUTION_ID', WithDecryption=True)['Parameter']['Value']

def lambda_handler(event, context):
    # 現在の時間と24時間前の時間を取得
    end_time = datetime.now(timezone.utc)
    start_time = end_time - timedelta(days=1)

    # 取得したいメトリクスの名前をリストで指定
    metrics = ['Requests', '4xxErrorRate', '5xxErrorRate']

    # 統計データに応じたStatisticsの値
    statistics = {
        'Requests': 'Sum',
        '4xxErrorRate': 'Average',
        '5xxErrorRate': 'Average'
    }

    results = {}
    for metric_name in metrics:
        try:
            # CloudWatchメトリクスの統計データを取得
            response = cloudwatch.get_metric_statistics(
                Namespace='AWS/CloudFront',
                MetricName=metric_name,
                Dimensions=[
                    {'Name': 'DistributionId', 'Value': DISTRIBUTION_ID},
                    {'Name': 'Region', 'Value': 'Global'}
                ],
                StartTime=start_time,
                EndTime=end_time,
                Period=3600,
                Statistics=[statistics[metric_name]]
            )
            # 結果を辞書に格納
            if 'Datapoints' in response:
                sorted_datapoints = sorted(response['Datapoints'], key=lambda x: x['Timestamp'])
                results[metric_name] = [dp_to_json(dp) for dp in sorted_datapoints]
            else:
                results[metric_name] = []

        except Exception as e:
            results[metric_name] = f"Error: {str(e)}"

    message = construct_message(results)
    try:
        # LINEにメッセージを送信
        send_line_message(LINE_ACCESS_TOKEN, LINE_USER_ID, message)
        return {'statusCode': 200, 'body': json.dumps('Message sent to LINE successfully')}
    except Exception as e:
        return {'statusCode': 500, 'body': json.dumps(f'Failed to send message to LINE: {str(e)}')}

# タイムスタンプをJSTに変換する関数
def dp_to_json(dp):
    jst_time = dp['Timestamp'].astimezone(timezone(timedelta(hours=+9))).strftime('%Y-%m-%d %H:%M:%S')
    return {
        'Timestamp': jst_time,
        'Sum': dp.get('Sum', dp.get('Average', ''))
    }

# LINEに送信するメッセージ
def construct_message(results):
    message = "CloudFrontの統計データを取得しました:\n"
    for metric_name, datapoints in results.items():
        if isinstance(datapoints, list) and datapoints:
            message += f"{metric_name}:\n"
            for dp in datapoints:
                message += f"Timestamp: {dp['Timestamp']}, Sum: {dp['Sum']}\n"
        else:
            message += f"{metric_name}: データなし\n"
        message += "\n"
    return message

# LINEにメッセージを送信する関数
def send_line_message(token, user_id, message):
    url = 'https://api.line.me/v2/bot/message/push'
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {token}'
    }
    data = {
        'to': user_id,
        'messages': [{'type': 'text', 'text': message}]
    }
    response = requests.post(url, headers=headers, data=json.dumps(data))
    response.raise_for_status()

$ lsb_release -d
Description:	Ubuntu 22.04.3 LTS
$ python -V
Python 3.10.12
$ pip --version
pip 24.1.2 from /home/user/.pyenv/versions/3.10.12/lib/python3.10/site-packages/pip (python 3.10)

$ mkdir ./lambda_work ; cd $_
$ pip install -t ./python requests
$ zip -r requests.zip python/