CloudFormationのスタック数とリソース数をSlackに通知する仕組みを作った

CloudFormationには、スタックの最大数が200、1スタック内のリソース最大数が200の制限があります。 デプロイ失敗時に気づいて対処するよりも、日頃から確認しておき、制限が近くなったときに対処できれば安心です。 気づく方法のひとつとして、週1回の頻度でCloudFormationのスタック数とリソース数をSlackに通知してみました。

パラメータストアに通知先を追加

通知先URLを取得し、下記コマンドでSSM（AWS Systems Manager）のパラメータストアに追加します。 URLの先頭にhttps://があるとコマンド実行に失敗するため除去しています。

aws ssm put-parameter \
    --type 'String' \
    --name '/Slack/INCOMING_WEBHOOK_URL/CloudFormationResource' \
    --value 'hooks.slack.com/services/xxxxx/yyyyy/zzzzz'

Slackに通知するサーバーレスアプリを作成する

AWS SAMプロジェクトを作成

下記コマンドでAWS SAMプロジェクトを作成します。

sam init --runtime python3.7 --name NotifyCloudFormationResource

テンプレートファイルを修正

AWS SAMのテンプレートファイルを下記にします。

template.yaml

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: NotifyCloudFormationResource

Parameters:
  NotifySlackUrl:
    Type: AWS::SSM::Parameter::Value<String>

Resources:
  NotifyCloudFormationResourceFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: notify-cloudformation-resource-function
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.7
      Timeout: 30
      Policies:
        - arn:aws:iam::aws:policy/AWSCloudFormationReadOnlyAccess
      Environment:
        Variables:
          NOTIFY_SLACK_URL: !Ref NotifySlackUrl
      Events:
        NotifySlack:
          Type: Schedule
          Properties:
            Schedule: cron(0 0 ? * MON *) # 日本時間で月曜日のAM9時

  NotifyCloudFormationResourceFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub "/aws/lambda/${NotifyCloudFormationResourceFunction}"

Lambdaコードを作成

下記のLambdaコードを作成します。

app.py

import boto3
import json
import os
import requests
from typing import List, Dict

cfn = boto3.client('cloudformation')

NOTIFY_SLACK_URL = os.environ['NOTIFY_SLACK_URL']

def lambda_handler(event, context) -> None:
    # スタック一覧を取得する
    stacks = get_stacks()

    # 各スタックのリソース数を調べる
    result = []
    for stack in stacks:
        stack_name = stack['StackName']
        resources = get_stack_resources(stack_name)
        result.append({
            'StackName': stack_name,
            'ResourceCount': len(resources)
        })

    # 通知用のメッセージを作成する
    message = create_message(stacks, result)

    # メッセージをSlackに通知する
    post_slack(message)

def get_stacks(token: str=None) -> List[Dict]:
    """スタック一覧を取得する"""
    # https://docs.aws.amazon.com/ja_jp/AWSCloudFormation/latest/APIReference/API_ListStacks.html
    option = {
        'StackStatusFilter': [
            'CREATE_COMPLETE',
            'UPDATE_COMPLETE',
            'ROLLBACK_COMPLETE'
        ]
    }

    if token is not None:
        option['NextToken'] = token

    res = cfn.list_stacks(**option)
    stacks = res.get('StackSummaries', [])

    if 'NextToken' in res:
        stacks += get_stacks(res['NextToken'])
    return stacks

def get_stack_resources(stack_name: str, token: str=None) -> List[Dict]:
    """指定したスタックのリソース一覧を取得する"""
    option = {
        'StackName': stack_name
    }

    if token is not None:
        option['NextToken'] = token

    res = cfn.list_stack_resources(**option)
    resources = res.get('StackResourceSummaries', [])

    if 'NextToken' in res:
        resources += get_stack_resources(res['NextToken'])
    return resources

def create_message(stacks: List[Dict], result: List[Dict]) -> str:
    """メッセージを作成する"""
    # リソース数が多い順に並び替えてメッセージを作成する
    message = []
    for item in sorted(result, key=lambda x:x['ResourceCount'], reverse=True):
        stack_name = item['StackName']
        resource_count = item['ResourceCount']
        message.append(f'- {resource_count:3}: {stack_name}')

    message.append('----------------------------')
    message.append(f'total stack: {len(stacks)}')
    return '\n'.join(message)

def post_slack(message: str) -> None:
    """SlackにメッセージをPOSTする"""
    # https://api.slack.com/tools/block-kit-builder
    payload = {
        'blocks': [
            {
                'type': 'section',
                'text': {
                    'type': 'mrkdwn',
                    'text': 'CloudFormation Resource Count.'
                }
            },
            {
                'type': 'context',
                'elements': [
                    {
                        'type': 'mrkdwn',
                        'text': message
                    }
                ]
            }
        ]
    }
 
    # http://requests-docs-ja.readthedocs.io/en/latest/user/quickstart/
    try:
        response = requests.post(f'https://{NOTIFY_SLACK_URL}', data=json.dumps(payload))
    except requests.exceptions.RequestException as e:
        print(e)
        raise
    else:
        print(response.status_code)

なお、hello_world/requirements.txtはそのままでOKです。

requirements.txt

requests

AWSにデプロイする

下記のコマンドでビルド＆デプロイを行います。バケット名やスタック名は適宜変更してください。parameter-overridesにはSSMパラメータストアに追加した際のKeyを指定しています。

sam build

sam package \
    --output-template-file packaged.yaml \
    --s3-bucket your-bucket-name

sam deploy \
    --template-file packaged.yaml \
    --stack-name Notify-CloudFormation-Resource-Stack \
    --capabilities CAPABILITY_NAMED_IAM \
    --no-fail-on-empty-changeset \
    --parameter-overrides \
        NotifySlackUrl=/Slack/INCOMING_WEBHOOK_URL/CloudFormationResource

動作確認する

月曜日のAM9時に通知が来ました！

参考

• CloudFormationのスタック数とリソース数の一覧を表示するスクリプトを作った | Developers.IO