Zenn
DELTAテックブログPublicationへの投稿
👀

EventBridgeでイベントを検知してSlackに通知したい

Shogo-B
2025/02/07に公開
AWS
S3
SRE
SQS
EventBridge
tech

はじめに

こんにちは、馬場です!

今回の記事では、EventBridgeで任意のAWSイベントが実行されたことをトリガーにSlack通知をする方法について記事を書いていきます。

通知機能の要素は以下の通りとなります。

  • EventBridgeルール
  • SQS(FIFO)
  • Lambda(Python)
  • (S3)

背景

AWSのアカウントを複数管理していて
アカウントの作成自体は任意のタイミング/メンバーがやって構わないが、管理用のRoleなどは統制の都合上同じものを使って欲しい
という課題の解決方法は色々ありますが、この通知を行いたいと思ったものについてはCloudFormationのテンプレートをS3に保存し、それを使ったCloudFormationスタック作成ページのURLを作成して共有し、簡単かつ迅速に運用してもらえる状態にしていました。

当初はテンプレートを使ったら管理用にAWSアカウントも教えてね　で運用していたのですが、抜け漏れが多く、使われる側の立場から検知したい！　となったのが発端でした。

概要


今回解説する構成では以下のことを実現します。

  1. S3 で GetObject イベントが発生
  2. EventBridge ルールがイベントを検知
  3. 入力トランスフォーマーを使って SQS にメッセージを送信
  4. Lambda がメッセージを受け取りSlackにメッセージを送信

前提としてCloudTrailを有効にし、Slackのwebhookを取得しておく必要があります。

構成

この通知機能をCloudformationに起こしたので、それを使いながらポイントを解説していきます。
当初手で作った後にcloudformationのテンプレートを作ったので、割愛されているリソースがあるかもしれません。
記事のテンプレートの完成形を載せておきます。

EventBridge ルール

単純にGetObjectだけで探すと”CloudFormationのテンプレートとして利用された”というイベントだけを検知できないので、 sourceIPAddress が cloudformation.amazonaws.com であるという条件を追加します。
また、デフォルトのメッセージだと情報過多になってしまうので、入力トランスフォーマーを使って必要最低限のSlack送信用メッセージを用意しておきます。
イベントバスは適当にdefaultとしています。

  EventsRule:
    Type: "AWS::Events::Rule"
    Properties:
      Name: "S3GetObjectNotiferRule"
      EventBusName: !Ref EventBusName
      State: "ENABLED"
      EventPattern:
        detail-type: ["AWS API Call via CloudTrail"]
        source: ["aws.s3"]
        detail:
          eventSource: ["s3.amazonaws.com"]
          eventName: ["GetObject"]
          sourceIPAddress: ["cloudformation.amazonaws.com"]
      Targets:
        - Id: "SQSQueueTarget"
          Arn: !GetAtt SQSQueue.Arn
          SqsParameters:
            MessageGroupId: "EventBridge"
          InputTransformer:
            InputPathsMap:
              Account: "$.detail.userIdentity.accountId"
              templateFile: "$.detail.requestParameters.key"
            InputTemplate: |
              {
                "version" : "1.0",
                "source" : "custom",
                "content" : {
                  "textType" : "client-markdown",
                  "title" : ":bell: IAM Role が作成されました。 ",
                  "description" : "<templateFile>を元にRoleが作成されました。",
                  "nextSteps" : [
                    "作成元アカウントは以下の通りです。",
                    "AWS Account: <Account>"
                  ]
                }
              }

SQS

当初は普通のSQSキューを使っていましたが、Slack通知が重複して飛んできていました。
調べてみたところCloudFormationから公開しているテンプレートを呼び出すときにどうしても2回GetObjectが呼ばれるようなので、FIFOキューの重複削除機能を活用して鬱陶しくない通知を目指します。
FIFOキューの場合、キュー名が”.fifo”で終わる必要がある点にも注意が必要です。

  SQSQueue:
    Type: "AWS::SQS::Queue"
    Properties:
      QueueName: !Ref QueueName
      ReceiveMessageWaitTimeSeconds: 0
      SqsManagedSseEnabled: true
      FifoThroughputLimit: "perQueue"
      DelaySeconds: 0
      FifoQueue: true
      MessageRetentionPeriod: 300
      MaximumMessageSize: 262144
      DeduplicationScope: "messageGroup"
      VisibilityTimeout: 30
      ContentBasedDeduplication: true

  SQSQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues: [!Ref SQSQueue]
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: "sqs:SendMessage"
            Resource: !GetAtt SQSQueue.Arn

  SNSTopic:
    Type: "AWS::SNS::Topic"
    Properties:
      TopicName: !Ref QueueName
      FifoTopic: true
      ContentBasedDeduplication: true
      Subscription:
      - Endpoint: !GetAtt SQSQueue.Arn
        Protocol: "sqs"

Lambda

ソースコードは適当にS3バケットを作成し、zip化して配置しておきます。
料金もお安く済ませたいので、アーキテクチャは arm64 にしておきます。
Slackの通知はテストタイミングと実運用で投稿先を分けたかったので、環境変数化しています。

ソースコードはこんな感じです

import json
import os
import urllib.request

SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL")

def lambda_handler(event, context):
    for record in event['Records']:
        message_body = record['body']
        
        try:
            message_content = json.loads(message_body)
            print(f"Parsed Message Content: {message_content}")
            
            version = message_content.get('version')
            source = message_content.get('source')
            content = message_content.get('content', {})
            
            # Slackメッセージをフォーマット
            slack_message = {
                "text": (
                    f":bell: *{content.get('title', '').strip()}*\n\n"
                    f"{content.get('description', '').strip()}\n\n"
                    f"*Next Steps:*\n• " + "\n• ".join(content.get('nextSteps', []))
                )
            }

            data = json.dumps(slack_message).encode("utf-8")
            req = urllib.request.Request(
                SLACK_WEBHOOK_URL, 
                data=data, 
                headers={'Content-Type': 'application/json'}
            )
            with urllib.request.urlopen(req) as response:
                response_body = response.read()
                print(f"Response from Slack: {response_body}")

        except json.JSONDecodeError as e:
            print(f"Failed to parse message: {message_body}")
            print(f"Error: {e}")

    return {
        'statusCode': 200,
        'body': json.dumps('Messages processed successfully')
    }

cloudformationのテンプレートはこんな感じです

  sendSlackNotificationOfCFnUsage:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: "sendSlackNotificationOfCFnUsage"
      Description: "send message to Slack from SQS"
      MemorySize: 128
      Timeout: 10
      Handler: "lambda_function.lambda_handler"
      Runtime: "python3.12"
      Architectures: ["arm64"]
      Environment:
        Variables:
          SLACK_WEBHOOK_URL: !Ref SlackWebhookURL
      Code:
        S3Bucket: !Ref BucketName
        S3Key: !Ref CodeZipFileName
      Role: !GetAtt sendSlackNotificationRole.Arn

  sendSlackNotificationRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: sendSlackNotificationOfCFnUsageRole
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: { Service: "lambda.amazonaws.com" }
            Action: "sts:AssumeRole"
      Policies:
        - PolicyName: LambdaSQSPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - sqs:DeleteMessage
                  - sqs:GetQueueUrl
                  - sqs:ReceiveMessage
                  - sqs:GetQueueAttributes
                  - sqs:List*
                Resource: "*"

  LambdaEventSourceMapping:
    Type: "AWS::Lambda::EventSourceMapping"
    Properties:
      BatchSize: 10
      FunctionName: !GetAtt sendSlackNotificationOfCFnUsage.Arn
      Enabled: true
      EventSourceArn: !GetAtt SQSQueue.Arn

CFnテンプレートの全体

Parameters:
  SlackWebhookURL:
    Description: target slack webhook URL.
    Type: String
  QueueName:
    Description: Queue name. XXX.fifo
    Type: String
  EventBusName:
    Description: EventBus name.
    Type: String
    Default: "default"
  BucketName:
    Description: BucketName name.
    Type: String
  CodeZipFileName:
    Description: CodeZipFileName name.
    Type: String


Resources:
  SQSQueue:
    Type: "AWS::SQS::Queue"
    Properties:
      QueueName: !Ref QueueName
      ReceiveMessageWaitTimeSeconds: 0
      SqsManagedSseEnabled: true
      FifoThroughputLimit: "perQueue"
      DelaySeconds: 0
      FifoQueue: true
      MessageRetentionPeriod: 300
      MaximumMessageSize: 262144
      DeduplicationScope: "messageGroup"
      VisibilityTimeout: 30
      ContentBasedDeduplication: true

  SQSQueuePolicy:
    Type: AWS::SQS::QueuePolicy
    Properties:
      Queues: [!Ref SQSQueue]
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: "sqs:SendMessage"
            Resource: !GetAtt SQSQueue.Arn

  SNSTopic:
    Type: "AWS::SNS::Topic"
    Properties:
      TopicName: !Ref QueueName
      FifoTopic: true
      ContentBasedDeduplication: true
      Subscription:
      - Endpoint: !GetAtt SQSQueue.Arn
        Protocol: "sqs"


  sendSlackNotificationOfCFnUsage:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: "sendSlackNotificationOfCFnUsage"
      Description: "send message to Slack from SQS"
      MemorySize: 128
      Timeout: 10
      Handler: "lambda_function.lambda_handler"
      Runtime: "python3.12"
      Architectures: ["arm64"]
      Environment:
        Variables:
          SLACK_WEBHOOK_URL: !Ref SlackWebhookURL
      Code:
        S3Bucket: !Ref BucketName
        S3Key: !Ref CodeZipFileName
      Role: !GetAtt sendSlackNotificationRole.Arn

  sendSlackNotificationRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: sendSlackNotificationOfCFnUsageRole
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: { Service: "lambda.amazonaws.com" }
            Action: "sts:AssumeRole"
      Policies:
        - PolicyName: LambdaSQSPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - sqs:DeleteMessage
                  - sqs:GetQueueUrl
                  - sqs:ReceiveMessage
                  - sqs:GetQueueAttributes
                  - sqs:List*
                Resource: "*"

  LambdaEventSourceMapping:
    Type: "AWS::Lambda::EventSourceMapping"
    Properties:
      BatchSize: 10
      FunctionName: !GetAtt sendSlackNotificationOfCFnUsage.Arn
      Enabled: true
      EventSourceArn: !GetAtt SQSQueue.Arn


  EventsRule:
    Type: "AWS::Events::Rule"
    Properties:
      Name: "S3GetObjectNotiferRule"
      EventBusName: !Ref EventBusName
      State: "ENABLED"
      EventPattern:
        detail-type: ["AWS API Call via CloudTrail"]
        source: ["aws.s3"]
        detail:
          eventSource: ["s3.amazonaws.com"]
          eventName: ["GetObject"]
          sourceIPAddress: ["cloudformation.amazonaws.com"]
      Targets:
        - Id: "SQSQueueTarget"
          Arn: !GetAtt SQSQueue.Arn
          SqsParameters:
            MessageGroupId: "EventBridge"
          InputTransformer:
            InputPathsMap:
              Account: "$.detail.userIdentity.accountId"
              templateFile: "$.detail.requestParameters.key"
            InputTemplate: |
              {
                "version" : "1.0",
                "source" : "custom",
                "content" : {
                  "textType" : "client-markdown",
                  "title" : ":bell: IAM Role が作成されました。 ",
                  "description" : "<templateFile>を元にRoleが作成されました。",
                  "nextSteps" : [
                    "作成元アカウントは以下の通りです。",
                    "AWS Account: <Account>"
                  ]
                }
              }

まとめ

今回紹介したものを参考にしていただくとCloudFormationからS3に配置しているスタックテンプレートを使ってリソースを作成した際に、EventBridgeでイベントを検知してSlackに通知を送ることができるようになります。
似たような検知機能が欲しくなった際に、この記事が誰かの助けになれば幸いです。

Tips: IaC ジェネレーター

今回、記事を作成するにあたって手で作ったものをIaC化しようとしたところ "IaCジェネレーター" というものがあることを知りました。
https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/UserGuide/generate-IaC.html

実際に使ってみたところ、テンプレート名と削除ポリシー、置換ポリシーを設定しスキャンを実行し

スキャンが終わったら（ちょっと時間かかります）テンプレート化したいリソースを選ぶだけで簡単にCloudFormationのテンプレートを作成することができました。

かなり楽に既存のリソースのIaC化を進めていけそうであり、今回試していませんが既存のスタックへの取り込みもできそうだったので可能性を感じる機能がひそかにリリースされていました。

残念なポイントとしては

  • デフォルト値の設定なども全て細かに出力される
  • 対応していないAWSサービス、リソースがある
    上記があるため、テンプレートを作成した後にパースしてあげた方がメンテナンス性は上がるかなといったところです。
    ただ、既存のもののCloudFormationテンプレートを作成するのはかなり腰が重たいものなので、下書きを作ってくれるだけでも素晴らしい機能だと感じました。
DELTAテックブログ
DELTAテックブログPublication

DELTAは、ベンチャースタートアップなどの成長企業向けに、ソフトウェア・プロダクト開発や技術選定支援などの技術支援事業を提供しています。 また、成長企業向けに様々な事業支援を提供するSEVENRICH GROUPに属して、グループ内の事業会社や、出資先の支援もおこなっています。

Discussion