今回の記事では、下の構成図のように、Lambdaがエラーを発生してからユーザにメールで通知するまでの一連のリソースをCDK Ver.2 pythonをつかって作成していきます。
全コードはGithubを参照して下さい。
Githubはここ
前提条件
- AWS CDK v.2 がインストールされていること
- Python 3.x がインストールされていること
- AWSアカウントがあり、AWS CLIが設定されていること
※Cloud9を使うとこの辺りがPassできるため、Cloud9を使って今回の記事の内容は作成しています。
構築手順
1.CDKアプリの初期化
先ずはCDKアプリの初期化を行います。
$ mkdir cdk-lambda-rds-proxy
$ cd cdk-lambda-rds-proxy
$ cdk init --language python
2. 必要なパッケージをインストール
ここでは、CDKアプリを初期化した際に作成された.venvを有効化しています。
$ source .venv/bin/activate
$ pip install -r requirements.txt
3. スタックの作成
import os
from aws_cdk import (
Stack,
aws_lambda as lambda_,
aws_logs as logs,
aws_logs_destinations as logs_destinations,
aws_sns as sns,
aws_sns_subscriptions as subscriptions,
)
from constructs import Construct
class CdkLambdaErrorNotificationStack(Stack):
def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
super().__init__(scope, construct_id, **kwargs)
# エラーを引き起こすLambda関数を作成
error_trigger_function = lambda_.Function(
self,
"ErrorTriggerFunction",
runtime=lambda_.Runtime.PYTHON_3_9,
handler="lambda_function.lambda_handler",
code=lambda_.Code.from_asset(
os.path.join('lambda', 'error_trigger_function')
),
)
# メッセ―ジを出すSNS Topicを作成
topic = sns.Topic(self, 'MyTopic')
# Subscription追加
subscription = topic.add_subscription(
subscriptions.EmailSubscription('メールアドレス')
)
# エラーメッセージを出すLambda関数を作成
publish_message_function = lambda_.Function(
self,
'PublishMessageFunction',
runtime=lambda_.Runtime.PYTHON_3_9,
handler='lambda_function.lambda_handler',
code=lambda_.Code.from_asset(
os.path.join('lambda', 'publish_message_function')
),
environment={
'TOPIC_ARN': topic.topic_arn
},
)
# lambdaに権限を付与
topic.grant_publish(publish_message_function)
# LogGroupのSubscription Filterを追加
error_trigger_function.log_group.add_subscription_filter(
'ErrorTriggerFunctionSubscriptionFilter',
destination=logs_destinations.LambdaDestination(
publish_message_function
),
filter_pattern=logs.FilterPattern.any_term('ERROR'),
)
4. Lambda関数の作成
先ずは、エラーを引き起こす関数を作成します。
lambda/error_trigger_functionフォルダを作成し、その中にlambda_function.pyを作成して、以下の内容を記述します。以下のコードは、単純にExceptionを引き起こすだけとなっています。
def lambda_handler(event, context):
raise Exception('Intentional Error')
次に、Subscription FilterによってERRORが起きた際に呼出されるLambda関数を作成します。
lambda/publish_message_functionフォルダを作成し、その中にlambda_function.pyを作成して、以下の内容を記述します。以下のコードは、受け取ったEvent情報をデコードして、SNSでメール送信するまでを実行しています。途中、メールの内容が見やすくなるように、Event情報をメール用に成型しています。
import json
import boto3
import os
import gzip
import base64
def lambda_handler(event, context):
sns_client = boto3.client('sns')
topic_arn = os.environ['TOPIC_ARN']
# デコード
decoded_event = base64.b64decode(event['awslogs']['data'])
# バイナリに圧縮されているため展開
decoded_event = json.loads(gzip.decompress(decoded_event))
# SNS通知
# 件名
subject = f'Error in {decoded_event["logGroup"]}'
# 本文
# 見やすいように成形
message = f'Account:\t{decoded_event["owner"]}\nLogGroup:\t{decoded_event["logGroup"]}\nLogStream:\t{decoded_event["logStream"]}\n'
for i, log_event in enumerate(decoded_event['logEvents']):
message += f'\nLogEvent-{i + 1}\n'
for key, value in log_event.items():
message += f'\t{key}:\t{value}\n'
# メール送信
response = sns_client.publish(
TopicArn = topic_arn,
Subject = subject,
Message = message
)
return {
'statusCode': 200
}
CDKアプリをデプロイ
Stackの準備が完了したため、デプロイします。
$ cdk deploy
デプロイが完了すると、Lambdaのエラーがメール通知されるようになります。
下は通知されたメールの内容です。
まとめ
この記事では、Lambdaのエラー内容をメールで通知する構成をCDKで作成する方法について紹介しました。エラー内容の通知は運用監視時には必要な項目と思いますので、参考になれば幸いです。