こんにちは、つくぼし(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程度になる想定です。
コストの内訳については、以下も併せてご参照ください。
構築手順
前提条件
今回は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)でした!