これはなに?

ElasticCacheのイベント通知をSlackに通知する奴

どうして通知するんですか?

他のプロジェクトでDB周りのアラートが発報して確認したらイベント通知にメンテで再起動されていた。

イベント情報を検知できるようにイベント通知をしたくなった。

構成

シンプルにElasticCache→SNS→Lambda→Slack

ElasticCacheのイベント通知先をSNSのトピックにして、サブスクライバーであるLambdaに対して投げつける。ログはCloudWatch Logsに投げる。Lambda FunctionはPythonでSlack通知させました。

ここの一通りの設定を見ていきます。

f:id:oza__shu:20201113191816p:plain

Lambdaで使用するので、Webhook URLを作成

名前と通知先とかわいい画像を指定して作成してください。

SNS通知先の設定

これはElasticCacheの設定でSNSのarnを渡すだけ。

KMSのkeyを作成

keyを作成したら使うのにaliasも必要になるので準備すればOK

SNSでTopicとサブスクライバーの設定

Topicとサブスクリプションを作成します。

Topicは作成したら以下のポリシーを当てるぐらい

statement {
    sid    = "LambdaPublish"
    effect = "Allow"
    principals {
      type        = "AWS"
      identifiers = ["*"]
    }
    actions = [
      "SNS:GetTopicAttributes",
      "SNS:Publish"
    ]
    resources = [
      "arn:aws:sns:ap-northeast-1:${data.aws_caller_identity.self.account_id}:topic",
    ]
  }
}

サブスクリプションはLambdaに送るので送り先のARNをプロトコルでLambdaを指定すればOK.

メッセージ送信をテストで実行してSNSとLambda間で疎通が取れるかテスト可能です。

次はLambdaを用意していきます。

Lambdaに必要なIAMポリシー

Lambda用のRoleが必要なので作成します。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Action": "sts:AssumeRole",
      "Principal": {
        "Service": "lambda.amazonaws.com"
      },
      "Effect": "Allow"
    }
  ]
}

Webhook URLはKMSで暗号化して渡すので、解読させるためにKMSのDecryptの権限ポリシーが必要です。

{
    "Version": "2012-10-17",
    "Statement": [
      {
          "Effect": "Allow",
          "Action": "kms:Decrypt",
          "Resource": "arn:aws:kms:ap-northeast-1:${data.aws_caller_identity.self.account_id}:key/${aws_kms_alias.kms_alias.target_key_id}"
      }
    ]
 }

作成したロールに上のと下2つのポリシーをアタッチすればOK

CloudWatchReadOnlyAccess
AWSLambdaBasicExecutionRole

SlackのhookのURLをKMSで暗号化

Terraformではできなかったので、手動で暗号化をして、それをTerraformでLambdaの関数に読ませました。

暗号化したURLは次の手順で作成

転送時の暗号化に使用するヘルパーの有効化のチェックボックスをチェック
保管時に暗号化する AWS KMS キーの選択で、作った暗号化キーを選択
kmsEncryptedHookUrlのvalueに、SlackのWebhookのURLを入れる
暗号化ボタンを押下
暗号化完了

暗号化したらLambdaの kmsEncryptedHookUrl 環境変数に渡します。

Lambdaでイベント通知の絞り込み

イベント内容で絞り込みしないと毎日のsnapshotのイベントとかで検知してしまうので、

それでイベント通知は以下の2つに絞り込みました。

ElastiCache:FailoverComplete
ElastiCache:CacheNodeReplaceComplete

スクリプトは以下のように暗号化したslackのURLとチャンネル名を渡して、

イベント通知に一致したらslack通知する内容です。

import boto3
import json
import logging
import os

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

# The base-64 encoded, encrypted key (CiphertextBlob) stored in the kmsEncryptedHookUrl environment variable
ENCRYPTED_HOOK_URL = os.environ['kmsEncryptedHookUrl']
# The Slack channel to send a message to stored in the slackChannel environment variable
SLACK_CHANNEL = os.environ['slackChannel']

HOOK_URL = "https://hooks.slack.com" + boto3.client('kms').decrypt(
    CiphertextBlob=b64decode(ENCRYPTED_HOOK_URL),
    EncryptionContext={
        'LambdaFunctionName': os.environ['AWS_LAMBDA_FUNCTION_NAME']}
)['Plaintext'].decode('utf-8')

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

NOTIFICATION_EVENT_TYPE = [
    'ElastiCache:FailoverComplete',
    'ElastiCache:CacheNodeReplaceComplete',
]

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

    # イベントタイプがNOTIFICATION_EVENT_TYPEに含まれない場合は処理を終了
    event_set = set(message.keys())
    notification_event_type_set = set(NOTIFICATION_EVENT_TYPE)
    event_type = event_set & notification_event_type_set
    if not event_type:
        return

    logger.info("Message: " + str(message))

    slack_message = {
        'channel': SLACK_CHANNEL,
        'text': "%s ElastiCache Notification Message: %s" % (event_time, message)
    }

    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)