EventBridgeを使用してAWS上で発生したイベントをTeamsに通知する方法

本記事では、EventBridgeとLambdaを使用して、AWS上で発生したイベントをTeamsのチャネルに通知する方法について解説いたします。

この方法を使用すれば、監視したいイベントの即時検知を行うことができ、インシデントへの対応速度の向上にも繋がります。

EventBridgeとは
実際にやってみよう
おわりに

EventBridgeとは

EventBridgeとは、AWSリソースやアプリケーションのイベントを使用して、イベント駆動型のアプリケーションを構築することができるサービスとなります。

今回は、EventBridgeを使用して「IAMロールの作成」イベントをLambdaに連携し、イベントの内容を整形してTeamsに通知する仕組みを作成していこうと思います。

実際にやってみよう

今回は、以下の手順でEventBridgeルールの作成とLambda関数の作成を実施致します。

受信用Teamsチャネルの準備
Lambda関数の作成
EventBridgeルールの作成

今回構築するシステムの構成図は以下の通りです。

「IAMロール作成」イベントの内容をLambdaに連携し、イベントの内容から通知のメッセージに組み込む値を取得し、Teamsに通知する構成とします。

受信用Teamsチャネルの準備

通知を受信するTeamsのチャネルで設定を行います。事前に専用のTeamsのチャネルを作成してから以下の手順を実施して下さい。

（１）通知を受信するTeamsチャネルの右上のメニューから「コネクタ」を選択します。

（２）「Incoming-Webhook」を検索し「構成」を選択します。

（３）任意の名前を入力し「作成」を選択します。

（４）WebhookURLをコピーし「完了」を選択します。コピーしたWebhookURLは控えておいて下さい。

Lambda関数の作成

今回は、バージニア北部のリージョンにLambda関数を作成します。

（１）「関数の作成」から「関数名」と「ランタイム」を設定して作成を行います。今回はランタイムに「Python 3.10」を設定します。

（２）作成したLambda関数のコードに以下のコードを貼り付けます。

今回は、必要最低限の処理のみを記述したコードを使用します。コード内のwebhookurlには「受信用Teamsチャネルの準備」手順でコピーしたURLを貼り付けてください。

import json
import urllib.request

webhookurl = 'コピーしたURL'

def lambda_handler(event, context):
    EventName = event['detail']['eventName']
    RoleName = event['detail']['requestParameters']['roleName']
    User = event['detail']['userIdentity']['arn']
    data = {
        "@type": "MessageCard",
        "@context": "http://schema.org/extensions",
        "themeColor": "d76100",
        "summary": "IAM Role Create Notification",
        "sections": [{
            "activityTitle": "IAMロール作成通知",
            "activityImage": "https://img.icons8.com/color/50/000000/amazon-web-services.png",
            "facts": [{
                "name": "イベント名",
                "value": EventName
            },{
                "name": "作成者",
                "value": User
            },{
                "name": "ロール名",
                "value": RoleName
            }]
            }]
        }
    headers = {'Content-Type': 'application/json'}
    request = urllib.request.Request(webhookurl,json.dumps(data).encode('utf-8'), headers)
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")