タグ: 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)

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