こんにちは!司空です!
今日は AWS Lambda を使ってる人ならぜひ知っておきたい
「構造化ログ(JSONログ)でCloudWatch Logs Insightsを快適に使おう!」
というお話です。
そもそも CloudWatch Logs Insights ってなに?
CloudWatch Logs Insights(以下 Insights)は、CloudWatchに溜まったログを検索・分析するためのクエリツール。
Lambdaのログをちゃんと整理しておくと、めちゃくちゃ便利に使えるんです。
でも普通のログって…
たとえば、こんな感じでログ出してません?
print("処理開始")
print(f"ユーザーID: {user_id}, ステータス: {status}")
CloudWatchでは、これが全部「ただの文字列」として保存されます。
なので、Insightsでフィルタするときにちょっとめんどくさいんです。
これをJSONにするだけで世界が変わる!
実は、JSON形式でログを出すだけで、CloudWatch側が中身を勝手に解析してくれて、各キーに対してクエリが書けるようになるんです。
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info(json.dumps({
"message": "処理開始",
"user_id": "abc123",
"status": "active"
}))
たったこれだけで、CloudWatch Logs Insightsではこんな風にクエリが書けるように👇
fields @timestamp, @message, user_id, status
| filter status = "active"
| sort @timestamp desc
コツ:ログは全部JSONで出そう
Pythonなら json.dumps()
で簡単にJSON化できます。
ポイントは、全部のログをJSONに統一すること。
途中で "Starting process..."
みたいな普通の文字列を混ぜると、Insightsでは取り扱いにくくなるので、可能な限り全部をJSON形式で統一するのがおすすめ!
※日本語が文字化けしてしまう
json.dumps()
を使って Python ログを JSON 形式で出力する際、日本語が文字化けして \uXXXX
のようにエスケープされてしまうことがあります。
これは json.dumps()
のデフォルト設定が ensure_ascii=True
だからです。
json.dumps(..., ensure_ascii=False)
を指定してください
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info(json.dumps({
"message": "処理開始",
"user_id": "abc123",
"status": "active"
}, ensure_ascii=False))
※Lambdaだとboto3, botocoreのログが出てしまう
原因
この二つのライブラリはデフォルト設定でWARNING以上のログを出力されるようになっている。
下記を追加する
logging.getLogger('boto3').setLevel(logging.WARNING)
logging.getLogger('botocore').setLevel(logging.WARNING)
補足:FastAPI や Django でも応用できる?
もちろん!構造化ログはLambdaに限らず、どこでも便利です。
FastAPIやDjangoなら、structlog
や python-json-logger
を使うとスマートに出力できますよ。
まとめ
✅ CloudWatch Logs Insights を使うなら構造化ログがマスト
✅ Lambdaでのログは json.dumps()
で出せばOK
✅ フィルタ・分析・可視化が一気に快適になる
おまけ:テンプレ関数
最後に、Lambda用の構造化ログテンプレを載せておきます👇
import json
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def log_info(**kwargs):
logger.info(json.dumps(kwargs))
def lambda_handler(event, context):
log_info(event_type="start", user_id="abc123", status="active")
# 他の処理
log_info(event_type="end", result="success")
ぜひ、構造化ログで快適なLambdaライフを!