はじめに
こんにちは。
Penetrator プロダクト開発部で主にインフラを担当している遠藤です。
みなさん元気に Lambda Life を送っていますか?
Lambda 関数の運用では、適切なログ出力が欠かせません。
しかし、標準の logging ライブラリだけでは、JSON 形式の構造化ログを出力するための設定が煩雑です。
また、Lambda コンテキスト情報 (関数名, リクエスト ID, コールドスタート状態等) を毎回手動でログに含めるのは手間がかかり、実装の一貫性を保つのも困難です。
そんな課題を解決してくれるのが、Powertools for AWS Lambda の Logger 機能です。
今回はこの機能について、実際の使用イメージと共に紹介します。
なお、今後も Powertools for AWS Lambda の各種機能について継続的にブログを書いていく予定なので、興味のある方は楽しみにしていただけると嬉しいです。
※ 過去のブログ記事はコチラ👇
Powertools for AWS Lambda とは
Powertools for AWS Lambda は、サーバーレスベストプラクティスの実装と開発者の生産性向上を支援する開発者ツールキットです。
段階的に導入できる柔軟な設計により、必要な機能から順次採用することができます。
Python, TypeScript, Java, .NET で提供されており、AWS Well-Architected Serverless Lens に基づく推奨事項を簡単に実装できます。
Logger 機能
Logger 機能 は、JSON 形式で構造化されたログ出力を提供する、Lambda に最適化されたロギングユーティリティです。
主な特徴
- Lambda コンテキストの自動キャプチャ: Lambda 関数のコンテキスト情報 (関数名, リクエスト ID, コールドスタート状態など) を自動的にログに含める
- 構造化 JSON ログ: すべてのログを JSON 形式で出力し、CloudWatch Logs での検索性・分析性を向上
- Lambda イベントのロギング: デバッグ時に受信イベントを自動的にログに記録
- ログサンプリング: 全リクエストの一定割合のみ DEBUG レベルでログを出力し、コストと詳細情報のバランスを取る
- 動的なキー追加: 実行時に任意のキー・バリューをログに追加
- ログバッファリング: 特定のリクエストや呼び出しに対してログをバッファリングし、エラー時に自動フラッシュまたは手動でフラッシュ
実践: Lambda 関数の構築・実装
ここからは、実際に Lambda 関数を構築・実装するイメージを紹介します。
※ プロジェクトの完全なコードは コチラ。
| 使用ツール | バージョン |
|---|---|
| aws-cdk-cli | 2.1021.0 |
| aws-cdk-lib | 2.199.0 |
| aws-cli | 2.13.32 |
| node | 22.13.1 |
| python | 3.11.8 |
| uv | 0.6.14 |
| jq | 1.7 |
0. プロジェクトの初期化
# プロジェクトディレクトリの作成
mkdir logger && \
cd logger
# AWS CDK プロジェクトの初期化
cdk init app --language python
# プロジェクトのセットアップ
source .venv/bin/activate && \
uv init && \
uv add --group infra -r requirements.txt && \
uv add --dev -r requirements-dev.txt && \
mkdir lambda && \
touch lambda/function.py && \
rm requirements.txt requirements-dev.txt main.py
主なプロジェクト構成は以下のようになります:
logger/
├── logger/
│ └── logger_stack.py # AWS CDK スタック定義
├── lambda/ # Lambda 関数のソースコード
│ └── function.py # Logger 機能使用版
├── app.py # AWS CDK アプリケーションのエントリーポイント
├── cdk.json # AWS CDK 設定ファイル
├── pyproject.toml # 依存関係管理ファイル
└── uv.lock # 依存関係管理ファイル
1. 依存関係の設定
# Powertools for AWS Lambda の追加
uv add --group lambda aws-lambda-powertools
2. Lambda 関数の実装
import json
from typing import Any
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.typing import LambdaContext
# Logger のインスタンス化 (サンプリング設定付き)
logger = Logger(
service="payment",
sampling_rate=0.1, # 10% の確率で DEBUG レベルのログを出力
)
@logger.inject_lambda_context(
log_event=True, # イベントのロギングを有効化 (開発環境のみ推奨)
clear_state=True, # 各呼び出し後にカスタムキーをクリア
)
def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
"""Logger 機能使用"""
# ユーザー情報の取得
user_id = event.get("body", {}).get("user_id")
amount = event.get("body", {}).get("amount", 0)
# 永続的なキーの追加 (このリクエスト中のすべてのログに含まれる)
logger.append_keys(user_id=user_id)
logger.info("Starting payment process")
# DEBUG レベルのログ (サンプリングにより 10% の確率でのみ出力される)
logger.debug(
"Payment validation details",
extra={
"validation_rules": ["amount_positive", "user_exists"],
"amount": amount,
},
)
# 処理ステップごとのログ
logger.info("Validating payment request")
# メタデータ付きログ出力 (extra パラメータ)
logger.info(
"Payment details validated", extra={"amount": amount, "currency": "JPY"}
)
try:
# 決済処理のシミュレーション
if amount <= 0:
logger.warning("Invalid payment amount", extra={"amount": amount})
return {"statusCode": 400, "body": json.dumps({"error": "Invalid amount"})}
# 追加のコンテキスト情報をログに含める
payment_id = f"PAY-{user_id}-{amount}"
logger.append_keys(payment_id=payment_id)
logger.info("Payment processed")
logger.debug("Payment processing completed successfully")
return {
"statusCode": 200,
"body": json.dumps(
{"message": "Payment successful", "payment_id": payment_id}
),
}
except ValueError as e:
# 特定の例外のハンドリング
logger.exception("Validation error occurred")
return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
except Exception:
# 予期しない例外
logger.exception("Unexpected error during payment processing")
return {
"statusCode": 500,
"body": json.dumps({"error": "Internal server error"}),
}
3. AWS CDK スタックの定義
from aws_cdk import CfnOutput, Duration, Stack
from aws_cdk import aws_lambda as _lambda
from constructs import Construct
class LoggerStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# Lambda Layer: Powertools for AWS Lambda (Python)
layer = _lambda.LayerVersion.from_layer_version_arn(
self,
"PowertoolsLayer",
layer_version_arn="arn:aws:lambda:ap-northeast-1:017000801446:layer:AWSLambdaPowertoolsPythonV3-python311-x86_64:20"
)
# Lambda 関数
function = _lambda.Function(
self,
"Function",
runtime=_lambda.Runtime.PYTHON_3_11,
handler="function.lambda_handler",
code=_lambda.Code.from_asset("lambda"),
layers=[layer],
timeout=Duration.seconds(10),
environment={
"POWERTOOLS_SERVICE_NAME": "payment",
"POWERTOOLS_LOG_LEVEL": "INFO",
"POWERTOOLS_LOGGER_SAMPLE_RATE": "0.1", # 10% サンプリング
"LOG_LEVEL": "INFO", # 後方互換性のため
}
)
# 出力
CfnOutput(
self,
"FunctionName",
value=function.function_name,
export_name="FunctionName",
)
4. デプロイ・動作確認
# デプロイの実行
cdk watch
# テスト用のペイロード作成
cat > test_payload.json << 'EOF'
{
"body": {
"user_id": "user456",
"amount": 10000
},
"requestContext": {
"requestId": "test-request-123"
}
}
EOF
# Lambda 関数のテスト
aws lambda invoke \
--function-name <関数名> \
--cli-binary-format raw-in-base64-out \
--payload file://test_payload.json \
response.json && \
cat response.json | jq
出力されるログの例:
{
"level": "INFO",
"location": "lambda_handler:23",
"message": "Starting payment process",
"timestamp": "2025-10-26 12:34:56,789+0000",
"service": "payment",
"sampling_rate": 0.1,
"cold_start": true,
"function_name": "logger-FunctionAfter-XXXXX",
"function_memory_size": 128,
"function_arn": "arn:aws:lambda:ap-northeast-1:123456789012:function:logger-FunctionAfter-XXXXX",
"function_request_id": "899856cb-83d1-40d7-8611-9e78f15f32f4",
"correlation_id": "test-request-123",
"user_id": "user456",
"xray_trace_id": "1-5759e988-bd862e3fe1be46a994272793"
}
5. リソースのクリーンアップ
cdk destroy -f
Logger 機能を使用するメリット
1. 構造化されたログで検索性・分析性が向上
JSON 形式でログを出力するため、CloudWatch Logs Insights で簡単にクエリを実行できます。
2. Lambda コンテキスト情報の自動キャプチャ
@logger.inject_lambda_context デコレーターを使用するだけで、以下の情報が自動的にログに含まれます:
- 関数名, メモリサイズ, ARN
- リクエスト ID
- コールドスタート状態
- X-Ray トレース ID (X-Ray が有効な場合)
これにより、手動でコンテキスト情報を追加する手間が省けます。
3. コストと詳細情報のバランスを取るサンプリング機能
通常は INFO レベルでログを出力し、全リクエストの 10% だけ DEBUG レベルで詳細なログを出力することで、ログのコストを抑えつつ、トラブルシューティングに必要な情報を確保できます。
4. 例外の統一的なロギング
logger.exception() メソッドを使用することで、例外名とスタックトレースが構造化された形式でログに記録されます。これにより、エラーの種類ごとに集計・分析が容易になります。
5. Correlation ID による分散トレーシング
API Gateway や ALB などのイベントソースから Correlation ID を自動的に抽出し、ログに含めることができます。これにより、複数の Lambda 関数にまたがるリクエストの追跡が容易になります。
まとめ
本記事では、Powertools for AWS Lambda (Python) の Logger 機能について紹介しました。
特に、本番環境での運用を考えると、構造化ログと適切なログレベル管理は必須と言えます。Logger 機能を活用することで、これらのベストプラクティスを簡単に実装できます。
Powertools for AWS Lambda に興味のある方、Lambda 関数の運用方法に課題を感じている方の参考になれば嬉しいです。
今後も Powertools for AWS Lambda の各種機能についてブログを書いていく予定ですので、お楽しみに。
参考資料
最後に
株式会社 WHERE (旧: 株式会社 Penetrator) は、シリーズ A ラウンドにおいて総額 5.5 億円の資金調達を実施し、不動産テック業界における更なる成長を目指して採用活動を一層強化しています。
エンジニア, デザイナー, カスタマーサクセス, BizDev, 営業, マーケティングなど、事業拡大を支える多様なポジションで共に挑戦していただける方を待っています!!
▽ 会社のカルチャーを知りたい方はこちら
https://www.wantedly.com/companies/company_9924832
▽ 募集職種を知りたい方はこちら
https://www.wantedly.com/companies/company_9924832/projects
Discussion