【Lambda】CloudWatch の通知を Lambda で Slack に飛ばしてみる【Slack】

担当している案件で CloudWatch からの通知を Slackに飛ばす必要があったので、Lambda で実装してみます。

1.事前準備

■ 通知を飛ばす Slack に Incoming WebHooks を追加しておく
https://slack.com/services/new/incoming-webhook

■ 必要なポリシーを付与した IAM ロールを作成しておく
・CloudWatchReadOnlyAccess
・AWSLambdaBasicExecutionRole

■ CloudWatch + SNS の通知設定

2.Lambda設定

Lambda > 関数 > 関数の作成 > 一から作成

関数名:<>
ランタイム:python 3.7
実行ロールの選択: 事前準備で作成したIAMロール

> トリガーを追加

トリガーの設定:SNS
SNS トピック:「事前準備で作成したSNSトピック」
トリガーの有効化:有効

関数コード

import boto3
import json
import logging
import os
from base64 import b64decode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
# 通知を飛ばすチャンネルを定義
SLACK_CHANNEL = "#xxxxxx"
# WEB_HOOKURLを定義
HOOK_URL = "https://hooks.slack.com/services/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
logger.info("Event: " + str(event))
message = json.loads(event['Records'][0]['Sns']['Message'])
logger.info("Message: " + str(message))
alarm_name = message['AlarmName']
#old_state = message['OldStateValue']
new_state = message['NewStateValue']
reason = message['NewStateReason']
slack_message = {
'channel': SLACK_CHANNEL,
'text': "%s state is now %s: %s" % (alarm_name, new_state, reason)
}
req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8'))
try:
response = urlopen(req)
response.read()
logger.info("Message posted to %s", slack_message['channel'])
except HTTPError as e:
logger.error("Request failed: %d %s", e.code, e.reason)
except URLError as e:
logger.error("Server connection failed: %s", e.reason)
import boto3 import json import logging import os from base64 import b64decode from urllib.request import Request, urlopen from urllib.error import URLError, HTTPError # 通知を飛ばすチャンネルを定義 SLACK_CHANNEL = "#xxxxxx" # WEB_HOOKURLを定義 HOOK_URL = "https://hooks.slack.com/services/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx" logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event, context): logger.info("Event: " + str(event)) message = json.loads(event['Records'][0]['Sns']['Message']) logger.info("Message: " + str(message)) alarm_name = message['AlarmName'] #old_state = message['OldStateValue'] new_state = message['NewStateValue'] reason = message['NewStateReason'] slack_message = { 'channel': SLACK_CHANNEL, 'text': "%s state is now %s: %s" % (alarm_name, new_state, reason) } req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8')) try: response = urlopen(req) response.read() logger.info("Message posted to %s", slack_message['channel']) except HTTPError as e: logger.error("Request failed: %d %s", e.code, e.reason) except URLError as e: logger.error("Server connection failed: %s", e.reason)
import boto3
import json
import logging
import os

from base64 import b64decode
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError


# 通知を飛ばすチャンネルを定義
SLACK_CHANNEL = "#xxxxxx"

# WEB_HOOKURLを定義
HOOK_URL = "https://hooks.slack.com/services/xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def lambda_handler(event, context):
    logger.info("Event: " + str(event))
    message = json.loads(event['Records'][0]['Sns']['Message'])
    logger.info("Message: " + str(message))

    alarm_name = message['AlarmName']
    #old_state = message['OldStateValue']
    new_state = message['NewStateValue']
    reason = message['NewStateReason']

    slack_message = {
        'channel': SLACK_CHANNEL,
        'text': "%s state is now %s: %s" % (alarm_name, new_state, reason)
    }

    req = Request(HOOK_URL, json.dumps(slack_message).encode('utf-8'))
    try:
        response = urlopen(req)
        response.read()
        logger.info("Message posted to %s", slack_message['channel'])
    except HTTPError as e:
        logger.error("Request failed: %d %s", e.code, e.reason)
    except URLError as e:
        logger.error("Server connection failed: %s", e.reason)

⇒「SLACK_CHANNEL」「HOOK_URL」に通知を飛ばすチャンネル名とWebHookURLを定義してください。