AWS CDK(ver.2) pythonを使って、Lambdaのエラー内容をメールで通知してみた

今回の記事では、下の構成図のように、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で作成する方法について紹介しました。エラー内容の通知は運用監視時には必要な項目と思いますので、参考になれば幸いです。