【AWS CloudWatch Logs】EC2からPythonで実務想定のテストログを生成/送信

はじめに

個人学習の目的を含めているため、不要な手順や無駄な工程があるかもしれません。
Pythonの実行環境としてEC2を利用していますが、お好きな環境で利用可能です。

本記事の目的

Pythonを使ってランダムなIPアドレスを含むログメッセージを生成し、
それをAWSのCloudWatch Logsに送信する。
（PythonとAWS CLIの個人学習）

↓実際に生成したログ↓
メッセージにランダムなIP例）IP=186.139.154.218とランダムなstatus含む。

CloudWatchから確認したログのサンプル（3件）

2024-11-16T12:17:23.252+09:00
Sample log entry at 2024-11-16T03:17:23.252405: status=INFO, message='This is a test log', IP=186.139.154.218

2024-11-16T12:17:24.354+09:00
Sample log entry at 2024-11-16T03:17:24.354606: status=WARN, message='This is a test log', IP=131.127.200.53


2024-11-16T12:17:25.377+09:00
Sample log entry at 2024-11-16T03:17:25.377620: status=WARN, message='This is a test log', IP=174.92.5.202

Pythonのスクリプトだけ興味ある方は、
本記事の「3.Pythonスクリプトの作成」を確認ください。

動機

「CloudWatch Logs Insightsを使って通信ログの分析を行いたい！」
実際のログがあれば良いのですが、個人の環境だと用意するのが難しいため、
ランダムなIPアドレスを含むテストログを生成し、CloudWatch Logsに送信します。

↓今回実際に作成したログをCloudWatch Logs Insightsのクエリを実行して試してみた結果はこちら

概要図

EC2にPythonのスクリプトを作成し、スクリプトで生成したログをCloudWatchに送信します。

手順

1.AWS環境の準備

1-1.CloudWatch Logsへのアクセス権限を持つIAMロール作成

今回はEC2からスクリプトでCloudwatchのロググループとログストリームの作成と、生成したログイベントを投げるのでそれ用のIAMロールです。

• ロール名：EC2-cloudwatch-full （なんでも良い）
• 信頼されたエンティティ：AWSサービス＞EC2
• 許可ポリシー：CloudWatchFullAccess
  （最低限、以下があれば良いかも）
  "logs:PutLogEvents"
  "logs:CreateLogGroup"
  "logs:CreateLogStream"

1-2.EC2インスタンス作成

Pythonスクリプト実行用のインスタンスです。

• インスタンスタイプ：t2.micro
• パブリックIP：自動的に割り当て
• OS：Amazon Linux
• IAMロール：EC2-cloudwatch-full
  （作成済みのCloudWatch Logsへのアクセス権限）
• セキュリティグループ：インバウンドルール>自身のIPのSSHのみを許可

2.EC2へ必要ライブラリインストール

2-1.EC2へSSH接続

TeraTermで作成したEC2へSSH接続
（EC2 Instance Connectなど、好きな方法で可）

2-2.AWS CLIのセットアップ

システムパッケージの最新化

$ sudo yum update -y

AWS CLIのインストール
Complete!になれば完了

$ sudo yum install awscli -y

以下の通りバージョンが確認できればOK

$ aws --version
aws-cli/2.15.30 Python/3.9.16 Linux/6.1.115-126.197.amzn2023.x86_64 source/x86_64.amzn.2023 prompt/off

aws configureにて必要な情報を設定
今回は東京リージョンを使用

$ aws configure
AWS Access Key ID [****************LFTU]:
AWS Secret Access Key [****************HEkq]:
Default region name [ap-northeast-1]:
Default output format [None]:
$

2-3.Python環境の準備

Python3とそのパッケージ管理ツールpipをインストール
※pipはboto3をインストールするため必要

$ sudo yum install python3 -y
Complete!

$ python3 --versionpython3 --version
Python 3.9.16

$ sudo yum install python3-pip -y
Complete!

$ pip3 --versionpip3 --version
pip 21.3.1 from /usr/lib/python3.9/site-packages/pip (python 3.9)

Boto3をインストール
SuccessfullyになればOK

$ pip3 install boto3
Successfully installed boto3-1.35.63 botocore-1.35.63 s3transfer-0.10.3

Boto3（AWS SDK for Python）とは？
PythonからAWS（Amazon Web Services）を操作するための公式ライブラリ。

パッケージを最新化して完了

$ sudo yum update -y

3.Pythonスクリプトの作成

ここが今回の肝というか本体です。

3-1.スクリプトファイルの作成

今回はファイル名を「upload_logs.py」で作成しています。

$ vi upload_logs.py

viでスクリプトの内容をコピペすればOK
基本的に東京リージョンであればこのまま使えます。

（以下の項目は好きなように、まずは10回とかで試すと良いかも）
ロググループ名：log_group_name = 'PythonGenerateTestLogGroup'
ログストリーム名：log_stream_name = 'PythonGenerateTestLogStream'
ログの生成回数（100回）for _ in range(100): # 任意の回数ログを生成

スクリプトの内容

import boto3
import datetime
import time
import random

# CloudWatch Logs のクライアントを作成
client = boto3.client('logs', region_name='ap-northeast-1')

# ロググループとログストリームの名前を指定
log_group_name = 'PythonGenerateTestLogGroup'
log_stream_name = 'PythonGenerateTestLogStream'

# ロググループが存在しない場合は作成
try:
    client.create_log_group(logGroupName=log_group_name)
except client.exceptions.ResourceAlreadyExistsException:
    pass

# ログストリームが存在しない場合は作成
try:
    client.create_log_stream(logGroupName=log_group_name, logStreamName=log_stream_name)
except client.exceptions.ResourceAlreadyExistsException:
    pass

# ランダムなIPアドレスを生成
def generate_random_ip():
    return '.'.join(str(random.randint(1, 255)) for _ in range(4))

# ログイベントの生成と送信
def generate_log_message():
    timestamp = datetime.datetime.now().isoformat()
    ip_address = generate_random_ip()  # ランダムIPアドレスの生成
    # statusはランダムでINFO,WARN,ERRORから
    message = f"Sample log entry at {timestamp}: status={random.choice(['INFO', 'WARN', 'ERROR'])}, message='This is a test log', IP={ip_address}"
    return message

# ログイベント送信の繰り返し
sequence_token = None
for _ in range(100):  # 任意の回数ログを生成
    log_event = {
        'logGroupName': log_group_name,
        'logStreamName': log_stream_name,
        'logEvents': [
            {
                'timestamp': int(time.time() * 1000),
                'message': generate_log_message()
            }
        ]
    }
    if sequence_token:
        log_event['sequenceToken'] = sequence_token

    response = client.put_log_events(
        logGroupName=log_group_name,
        logStreamName=log_stream_name,
        logEvents=log_event['logEvents']
    )
    sequence_token = response['nextSequenceToken']
    time.sleep(1)  # 少し待機

print("ログがCloudWatch Logsにアップロードされました！")

スクリプトファイル作成後、lsやcatで内容を確認してください。

４.Pythonスクリプトの実行（CloudWatchへテストログ送信）

さぁ！準備は整いました！

まず、既存でロググループが無いことを確認します。
※既存で同じ名前のロググループ名があっても問題ないが、
今回はスクリプト内で無ければロググループとログストリームを作成するため

EC2で作成したスクリプトを実行します。
ログがCloudWatch Logsにアップロードされました！が表示されると完了です。

$ python3 upload_logs.py 

~実行完了まで待つ(これは出力されません)~

ログがCloudWatch Logsにアップロードされました！

５.AWSコンソール（CloudWatch Logs）で確認

正常完了した場合、指定したロググループが作成されています。

ログストリームも作成されています。

statusとランダム生成したIPも意図した挙動になっています！

件数確認のためCloudWatch Logs Insightsでクエリを実行して確認します。
件数も指定した100件になっていました！（必要に応じて条件を選択してください）

ここまでで、目的の「Pythonを使ってランダムなIPアドレスを含むログメッセージを生成し、
それをAWSのCloudWatch Logsにアップロードする。」をクリアです。

おわりに

AWSの管理コンソール（GUI）で基本操作をしていたので、この機会にAWS CLIやPythonについて少し触れる良い機会でした。
もともと今回の動機がCloudWatch Logs Insightsでクエリを触ろうと思っていたので、今回生成したログで遊んでみたいと思います。

↓今回実際に作成したログをCloudWatch Logs Insightsのクエリを実行して試してみた結果はこちら