CloudFormation一撃で作るAWS料金通知ツール(Email/Slack/LINE対応)

つくぼし

2024.04.18

こんにちは、つくぼし(tsukuboshi0755)です！

以前以下のブログで、利用しているAWS料金を毎日LINEに通知するツールを構築しました。

上記ブログは様々な方々から大きな反響を頂いた一方で、以下のような課題もありました。

AWS SAMの利用を前提とするため、ローカル開発環境の構築が別途必要
通知間隔として毎日しか指定できない
通知先としてLINEしか指定できない
LINE Access Token等の機密情報をLambdaの環境変数に直接入力しているため、セキュリティに多少不安が残る

そこで今回は以前のコードをさらに改良し、上記の課題を解消しつつ、初心者でも簡単かつ柔軟に構築できるAWS料金通知ツールを作成したので紹介します！

システム概要

アーキテクチャ

今回作成するシステムは以下のような構成になります。

なお後述するEmailAddress/SlackWebhookURL/LineAccessTokenをCloudFormationのパラメータで指定しない場合は、それぞれFor Email/For Slack/For LINEのリソースを作成しないようになっています。

以前のシステムでは抱えていた課題について、以下のように解消しています。

CloudFormationテンプレートのためAWSコンソールから簡単にデプロイでき、ローカル開発環境の構築が不要
EventBridge Schedulerを用いてJST時刻を設定すると共に、通知間隔を1-31日のいずれかで指定可能
Lambdaコードをリファクタリングした事で、通知先としてEmail/Slack/LINEを指定できるようになり、併用も可
機密情報(Slack Webhook URL/LINE Access Token)をSecrets Managerに保存する事で、Lambdaの環境変数に直接入力する必要がなくなり、セキュリティが向上

コスト

本システムでは、主にCost Explorer及びSecrets Managerで料金がかかります。

Email/Slack/LINE全ての通知先に対して毎日メッセージを送付した場合の最大コストは、おおよそ月1.40USD程度になる想定です。

コストの内訳については、以下も併せてご参照ください。

AWS料金通知ツール見積もり

構築手順

前提条件

今回はCloudFormationを用いるため、AWSコンソールにログインできるAWSアカウントを用いるだけで構築可能です。

事前に通知先となる、メールアドレス/Slackアカウント/LINEアカウントのいずれかを用意しておいてください。

Cost Explorerの有効化

通知先に関わらず、もしCost Explorerがまだ有効になっていない場合は、この段階で有効にしておきます。

以下の公式ドキュメントを参考に、コスト管理コンソールより、「Cost Explorerを起動」ボタンをクリックしてください。

SlackのWebhook URL取得

通知先にSlackを利用する場合、SlackのWebhook URLを取得する必要があります。

以下を参考に、Slack Appを作成し、Webhook URLを取得し、メモしておいてください。

LINE NotifyのPersonal Access Token取得

通知先にLINEを利用する場合、LINE NotifyのPersonal Access Tokenを取得する必要があります。

以下を参考に、LINE NotifyのPersonal Access Tokenを取得し、メモしておいてください。

CloudFormationテンプレートのデプロイ

以下のCloudFormationテンプレートをYAML形式で、ローカルに保存してください。

なおLambdaコードを含んでいるため、通常より長めのテンプレートとなっていますのでご了承ください。

テンプレート

template.yaml

Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      - Label:
          default: "Notification Settings"
        Parameters:
          - NotifyDaysInterval
          - EmailAddress
          - SlackWebhookUrl
          - LineAccessToken

Parameters:
  NotifyDaysInterval:
    Type: Number
    Default: 1
    MinValue: 1
    MaxValue: 31
    Description: "Choose the interval of notification. (1-31)"
  EmailAddress:
    Type: String
    Default: ""
    Description: "If you want to notify by Email, set Email Address. If not, leave it blank."
  SlackWebhookUrl:
    Type: String
    Default: ""
    NoEcho: true
    Description: "If you want to notify by Slack, set Slack Webhook URL. If not, leave it blank."
  LineAccessToken:
    Type: String
    Default: ""
    NoEcho: true
    Description: "If you want to notify by LINE, set LINE Notify Access Token. If not, leave it blank."

Conditions:
  OnEmail: !Not [!Equals [!Ref EmailAddress, ""]]
  OnSlack: !Not [!Equals [!Ref SlackWebhookUrl, ""]]
  OnLine: !Not [!Equals [!Ref LineAccessToken, ""]]

Resources:
  NABTopicToEmail:
    Type: AWS::SNS::Topic
    Condition : OnEmail
    Properties:
      TopicName: !Sub ${AWS::StackName}-nab-topic
      Subscription:
        - Endpoint: !Ref EmailAddress
          Protocol: email

  NABSecretForSlack:
    Type: AWS::SecretsManager::Secret
    Condition : OnSlack
    Properties:
      Description: "Slack Webhook URL"
      SecretString: !Sub '{"info": "${SlackWebhookUrl}"}'
      Name: !Sub /${AWS::StackName}-nab-secret/slack

  NABSecretForLine:
    Type: AWS::SecretsManager::Secret
    Condition : OnLine
    Properties:
      Description: "LINE Access Token"
      SecretString: !Sub '{"info": "${LineAccessToken}"}'
      Name: !Sub /${AWS::StackName}-nab-secret/line

  NABFunctionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub ${AWS::StackName}-nab-function-role
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: "sts:AssumeRole"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

  NABCEAccessPolicy:
    Type: AWS::IAM::ManagedPolicy
    Properties:
      ManagedPolicyName: !Sub ${AWS::StackName}-nab-ce-access-policy
      Roles:
        - !Ref NABFunctionRole
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "ce:GetCostAndUsage"
            Resource: "*"

  NABEmailPolicy:
    Type: AWS::IAM::ManagedPolicy
    Condition: OnEmail
    Properties:
      ManagedPolicyName: !Sub ${AWS::StackName}-nab-email-policy
      Roles:
        - !Ref NABFunctionRole
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "sns:publish"
            Resource: !Ref NABTopicToEmail

  NABSlackPolicy:
    Type: AWS::IAM::ManagedPolicy
    Condition: OnSlack
    Properties:
      ManagedPolicyName: !Sub ${AWS::StackName}-nab-slack-policy
      Roles:
      - !Ref NABFunctionRole
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "secretsmanager:GetSecretValue"
            Resource: !Ref NABSecretForSlack

  NABLinePolicy:
    Type: AWS::IAM::ManagedPolicy
    Condition: OnLine
    Properties:
      ManagedPolicyName: !Sub ${AWS::StackName}-nab-line-policy
      Roles:
      - !Ref NABFunctionRole
      PolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Action:
              - "secretsmanager:GetSecretValue"
            Resource: !Ref NABSecretForLine

  NABFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub ${AWS::StackName}-nab-function
      Handler: index.lambda_handler
      Runtime: python3.12
      Role: !GetAtt NABFunctionRole.Arn
      Timeout: 60
      LoggingConfig:
        LogFormat: JSON
        ApplicationLogLevel: INFO
        SystemLogLevel: INFO
      Layers:
        - arn:aws:lambda:ap-northeast-1:133490724326:layer:AWS-Parameters-and-Secrets-Lambda-Extension:4
      Environment:
        Variables:
          EMAIL_TOPIC_ARN: !If
            - OnEmail
            - !Ref NABTopicToEmail
            - !Ref AWS::NoValue
          SLACK_SECRET_NAME: !If
            - OnSlack
            - !Sub /${AWS::StackName}-nab-secret/slack
            - !Ref AWS::NoValue
          LINE_SECRET_NAME: !If
            - OnLine
            - !Sub /${AWS::StackName}-nab-secret/line
            - !Ref AWS::NoValue
      Code:
        ZipFile: |
          import json
          import logging
          import os
          from datetime import date, datetime, timedelta
          from typing import Any, Dict, MutableMapping, Optional, Tuple
          from urllib import parse, request

          import boto3

          logger = logging.getLogger()

          ce = boto3.client("ce", region_name="us-east-1")


          # Lambdaのエントリーポイント
          def lambda_handler(event: Dict[str, Any], context: Any) -> None:
              # 合計とサービス毎の請求額を取得する
              total_billing = get_total_billing()
              service_billings = get_service_billings()

              # 投稿用のメッセージを作成する
              (title, detail) = create_message(total_billing, service_billings)

              try:
                  email_topic_arn = os.environ.get("EMAIL_TOPIC_ARN")
                  slack_secret_name = os.environ.get("SLACK_SECRET_NAME")
                  line_secret_name = os.environ.get("LINE_SECRET_NAME")

                  # メール用トピックが設定されている場合は、メール用トピックにメッセージを送信する
                  if email_topic_arn:
                      sns = boto3.client("sns")
                      sns.publish(
                          TopicArn=email_topic_arn,
                          Subject=title,
                          Message=detail,
                      )

                  # SlackのWebhook URLが設定されている場合は、Slackにメッセージを投稿する
                  if slack_secret_name:
                      webhook_url = get_secret(slack_secret_name, "info")
                      payload = {
                          "text": title,
                          "blocks": [
                              {"type": "header", "text": {"type": "plain_text", "text": title}},
                              {"type": "section", "text": {"type": "plain_text", "text": detail}}
                          ],
                      }
                      data = json.dumps(payload).encode()
                      headers = {"Content-Type": "application/json"}

                      send_request(webhook_url, data, headers)

                  # LINEのアクセストークンが設定されている場合は、LINEにメッセージを投稿する
                  if line_secret_name:
                      access_token = get_secret(line_secret_name, "info")
                      webhook_url = "https://notify-api.line.me/api/notify"
                      payload = {"message": f"{title}\n\n{detail}"}
                      data = parse.urlencode(payload).encode("utf-8")
                      headers = {"Authorization": "Bearer %s" % access_token}

                      send_request(webhook_url, data, headers)

                  # いずれの送信先も設定されていない場合はエラーを出力する
                  if not email_topic_arn and not slack_secret_name and not line_secret_name:
                      logger.error(
                          "No destination to post message. Please set environment variables."
                      )

              except Exception as e:
                  logger.exception("Exception occurred: %s", e)
                  raise e


          # 合計の請求額を取得する関数
          def get_total_billing() -> dict:
              (start_date, end_date) = get_total_cost_date_range()

              response = ce.get_cost_and_usage(
                  TimePeriod={"Start": start_date, "End": end_date},
                  Granularity="MONTHLY",
                  Metrics=["AmortizedCost"],
              )
              return {
                  "start": response["ResultsByTime"][0]["TimePeriod"]["Start"],
                  "end": response["ResultsByTime"][0]["TimePeriod"]["End"],
                  "billing": response["ResultsByTime"][0]["Total"]["AmortizedCost"]["Amount"],
              }


          # サービス毎の請求額を取得する関数
          def get_service_billings() -> list:
              (start_date, end_date) = get_total_cost_date_range()

              response = ce.get_cost_and_usage(
                  TimePeriod={"Start": start_date, "End": end_date},
                  Granularity="MONTHLY",
                  Metrics=["AmortizedCost"],
                  GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
              )

              billings = []

              for item in response["ResultsByTime"][0]["Groups"]:
                  billings.append(
                      {
                          "service_name": item["Keys"][0],
                          "billing": item["Metrics"]["AmortizedCost"]["Amount"],
                      }
                  )
              return billings


          # メッセージを作成する関数
          def create_message(total_billing: dict, service_billings: list) -> Tuple[str, str]:
              start = datetime.strptime(total_billing["start"], "%Y-%m-%d").strftime("%m/%d")

              # Endの日付は結果に含まないため、表示上は前日にしておく
              end_today = datetime.strptime(total_billing["end"], "%Y-%m-%d")
              end_yesterday = (end_today - timedelta(days=1)).strftime("%m/%d")

              total = round(float(total_billing["billing"]), 2)

              title = f"AWS Billing Notification ({start}～{end_yesterday}) : {total:.2f} USD"

              details = []
              for item in service_billings:
                  service_name = item["service_name"]
                  billing = round(float(item["billing"]), 2)

                  if billing == 0.0:
                      # 請求無し（0.0 USD）の場合は、内訳を表示しない
                      continue
                  details.append(f"・{service_name}: {billing:.2f} USD")

              # 全サービスの請求無し（0.0 USD）の場合は以下メッセージを追加
              if not details:
                  details.append("No charge this period at present.")

              return title, "\n".join(details)


          # 請求額の期間を取得する関数
          def get_total_cost_date_range() -> Tuple[str, str]:
              start_date = date.today().replace(day=1).isoformat()
              end_date = date.today().isoformat()

              # get_cost_and_usage()のstartとendに同じ日付は指定不可のため、
              # 「今日が1日」なら、「先月1日から今月1日（今日）」までの範囲にする
              if start_date == end_date:
                  end_of_month = datetime.strptime(start_date, "%Y-%m-%d") + timedelta(days=-1)
                  begin_of_month = end_of_month.replace(day=1)
                  return begin_of_month.date().isoformat(), end_date
              return start_date, end_date


          # シークレットマネージャからシークレットを取得する関数
          def get_secret(secret_name: Optional[str], secret_key: str) -> Any:
              # シークレット名を取得
              if secret_name is None:
                  raise ValueError("Secret name must not be None")
              secrets_extension_endpoint = (
                  "http://localhost:2773/secretsmanager/get?secretId=" + secret_name
              )

              # ヘッダーにAWSセッショントークンを設定
              aws_session_token = os.environ.get("AWS_SESSION_TOKEN")
              if aws_session_token is None:
                  raise ValueError("aws sessuib token must not be None")
              headers = {"X-Aws-Parameters-Secrets-Token": aws_session_token}

              # シークレットマネージャからシークレットを取得
              secrets_extension_req = request.Request(secrets_extension_endpoint, headers=headers)
              with request.urlopen(secrets_extension_req) as response:
                  secret_config = response.read()
              secret_json = json.loads(secret_config)["SecretString"]
              secret_value = json.loads(secret_json)[secret_key]
              return secret_value


          # HTTPリクエストを送信する関数
          def send_request(url: str, data: bytes, headers: MutableMapping[str, str]) -> None:
              req = request.Request(url, data=data, headers=headers, method="POST")
              with request.urlopen(req) as response:
                  print(response.status)

  NABFunctionScheduler:
    Type: AWS::Scheduler::Schedule
    Properties:
      Name: !Sub ${AWS::StackName}-nab-function-scheduler
      Description: "Start Notify AWS Billing Function"
      ScheduleExpression: !Sub cron(0 9 */${NotifyDaysInterval} * ? *)
      ScheduleExpressionTimezone: "Asia/Tokyo"
      FlexibleTimeWindow:
        Mode: "OFF"
      State: ENABLED
      Target:
        Arn: !GetAtt NABFunction.Arn
        RoleArn: !GetAtt NABFunctionSchedulerRole.Arn

  NABFunctionSchedulerRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub ${AWS::StackName}-nab-function-scheduler-role
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: scheduler.amazonaws.com
            Action: "sts:AssumeRole"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaRole

Outputs:
  FunctionArn:
    Description: "Lambda Function ARN"
    Value: !GetAtt NABFunction.Arn
  SchedulerArn:
    Description: "Scheduler ARN"
    Value: !GetAtt NABFunctionScheduler.Arn

テンプレートを保存したら、CloudFormationコンソールでスタックを作成していきます。

CloudFormationコンソールでのスタックの作成方法については、以下の公式ドキュメントを参考にしてください。

東京リージョン(ap-northeast-1)のCloudFormationコンソールでスタックの作成を実施し、保存したテンプレートをアップロードします。

以下のスタックの詳細画面では、スタック名及びパラメータを指定する必要があります。

スタック名は任意の名前を入力してください。

続いてパラメータについては、通知間隔と通知先の設定を行います。

まず通知間隔の日数として、NotifyDaysIntervalに1から31までのいずれかで指定してください。

次に通知先によって、各々のパラメータに対して以下のように指定してください。

Emailの場合
- EmailAddressに、対象のメールアドレスを入力してください。
Slackの場合
- SlackWebhookUrlに、先ほど取得したSlackのWebhook URLを入力してください。
LINEの場合
- LineAccessTokenに、先ほど取得したLINE NotifyのPersonal Access Tokenを入力してください。

なおEmailAddress、SlackWebhookUrl、LineAccessTokenに何かしらの文字列が入力されると、対応するリソースがデプロイされるテンプレートになっています。

そのため、使用しない通知先については、パラメータを空欄のままにしておくようご注意ください。

パラメータの内容に問題がなければ、スタックをデプロイします。

Emailのサブスクライブ

通知先にEmailを利用する場合、SNSトピックに対するサブスクライブを実施する必要があります。

スタックをデプロイ後、指定したメールアドレスに以下のようなメールが届いているので、"Confirm Subscription"をクリックしてサブスクライブを実施してください。

通知テスト

最後に、デプロイしたシステムがきちんと動くかテストします。

以下の通りLambdaコンソールにアクセスし、デプロイしたLambda関数である<Stack Name>-nab-functionを選択した後、テストタブを選択し"テスト"ボタンを押します。

その後、通知先として設定したEmail/Slack/LINEに、以下のような通知が届く事を確認できればOKです！

Emailの場合

Slackの場合

LINEの場合

あとは指定時刻(今回の場合はAM9:00)まで待機し、同じようなメッセージが届けばSchedulerの動作も問題ありません！

補足

従来の構築方法であるAWS SAMを利用した方法も、以下のリポジトリにて公開しています。

Lambdaコードを自分好みにカスタマイズしたい場合はCloudFormationよりSAMの方が便利な場合も多いので、必要に応じてご参照ください。

最後に

今回は簡単かつ柔軟に構築できるAWS料金通知ツールを紹介しました。

CloudFormationで構築する事で、特別な準備が必要なく、AWS初心者でも簡単に構築できるようになっています。

また通知間隔や通知先を自由に設定できるため、自分に合った通知システムを柔軟に構築できます。

普段からAWSを利用する方は、想定外の高額利用を防ぐために、ぜひ構築を検討してみてください。

以上、つくぼし(tsukuboshi0755)でした！

CloudFormation一撃で作るAWS料金通知ツール(Email/Slack/LINE対応)