AWS の利用料金の合計を Cost Explorer APIで取得する

この記事は、大国魂（ITブログ） Advent Calendar 2019 の15日目です。

記事概要

AWS の利用料総額を、Slack 通知する Lambda 関数を Python 3 で作成します。
Python/AWS 共に初心者レベルで修行中の身のため、極力単純な内容にしています。
勉強のため軽く見たいけど応用例みたいな記事しかなくて困ったので、そういった方の参考になれば幸いです。

AWS 使用料金について

Cost Explorer API を利用するため、1回の実行当たり 0.01 USD が最低でも発生します。
連続実行などはされないかと思いますがご注意ください。

また、この記事では AWS Cost Explorer 画面の(おそらく)デフォルトで選択されている 非ブレンドコスト ではなく、ブレンドコスト を Cost Explorer API にて取得します。

Slack 設定

Slack 通知するための Incomming Webhook 設定は こちらの記事 がわかりやすいので、詳細は省きます。
Incomming Webhook の URL をテキストエディタなどに控えておけば OK です。

ポリシーの作成

Cost Explorer の読み取り権限が必要なので、その権限がついたポリシーを作成します。
AWS IAM 画面の ポリシーの作成 ボタンを押下して、以下を設定。

• サービス: Cost Explorer Service
• アクション: 読み込みにチェック

(読み取り権限をすべて付与していますが、利用状況によるものの以下でも十分そうです)

"ce:GetReservationUtilization",
"ce:GetDimensionValues",
"ce:GetCostAndUsage",
"ce:GetReservationCoverage",
"ce:GetReservationPurchaseRecommendation",
"ce:GetCostForecast",
"ce:GetTags"

ポリシーの確認 ボタンを押下して、以下を入力。

• 名前: CostExplorerReadOnly
• 説明: CostCheck By lambda

ポリシーを作成して、この作業は完了です。

Lambda の設定

Lambda 関数の作成

Lambda の ダッシュボード or 関数 の、関数の作成 ボタンを押下。
以下の設定をして 関数の作成 ボタンを押下。

• 一から作成 を選択。
• 関数名: cost_check などそれっぽい名前
• ランタイム: Python 3.7

実行ロール設定

作成された Lambda 関数の 実行ロールの部分で、特に設定をしていなければ service-role/cost_check-role-hoge のようなロールが作成＆設定されているかと思います。
このロールに Cost Explorer の読み取りポリシーをアタッチします。

• IAM コンソールで cost_check-role-hoge ロールを表示します のリンクを押下。
• 表示された画面で、ポリシーをアタッチします ボタンを押下。
• 先ほど作成した CostExplorerReadOnly ポリシーを選択して、ポリシーのアタッチを押下。

環境変数

実行ロールの少し上にある 環境変数 にて、以下の2つを設定します。
ドル円換算に利用する1ドル何円かと、事前に設定した Slack の Webhookk URL です。

• JPY: 110
• WEBHOOK_URL: https://hooks.slack.com/services/(以下略)

関数コード

以下となります。
各々の処理でそれほど難しいことはしていないので、ちょっと悩んだ所だけ解説します。

• 費用を取得する cost_check
• Slack で表示するメッセージ作成の create_message
• Slack に通知する post_slack

import json
import urllib.request
import os
import boto3
import math
from datetime import datetime, date, timedelta

def cost_check():
	client = boto3.client('ce')
	today = datetime.today()

	# 合計金額を知りたいので、フィルタは最小限
	response = client.get_cost_and_usage(
	    TimePeriod={
	        'Start': datetime.strftime(today.replace(day=1), '%Y-%m-%d'),
	        'End': datetime.strftime(today, '%Y-%m-%d')
	    },
	    Granularity='MONTHLY',
        Metrics=['BlendedCost'],
	)

	return response['ResultsByTime'][0]['Total']['BlendedCost']['Amount']

def create_message(cost, jpy):
    # cost と jpy から日本円換算。日本円表記なので切り上げで処理。
    cost_jpy = str(math.ceil(float(cost) * int(jpy)))
    message = "当月のAWS使用料金は `$" + cost + "` です。" \
+ "\n日本円換算で `約" + cost_jpy + "円`となります"

    return message

def post_slack(webhook_url, message):
    
    # ユーザ名、アイコン含み Slack 投稿内容を設定
    send_data = {
        "username": "Slack 通知時に表示される名前",
        "icon_emoji": ":moneybag:",
        "text": message,
    }
    #print(send_data)
    send_text = "payload=" + json.dumps(send_data)
    request = urllib.request.Request(
        webhook_url, 
        data=send_text.encode("utf-8"), 
        method="POST"
    )
    with urllib.request.urlopen(request) as response:
        response_body = response.read().decode("utf-8")

def lambda_handler(event, context):
    # 環境変数取得
    webhook_url = os.environ['WEBHOOK_URL']
    jpy = os.environ['JPY']
    
    # 当月費用を取得
    cost = cost_check()
    
    # Slack 通知するメッセージ生成
    message = create_message(cost,jpy)
    
    # Slack 通知
    post_slack(webhook_url,message)

ちょっと悩んだところ

cost_check 内の get_cost_and_usage のレスポンス処理です。
return の前で print(response) などしてレスポンスの中身を見ると、以下のような形となっています。

{
	'ResultsByTime': 
	[
		{
			'TimePeriod': 
			{
				'Start': '2019-12-01', 
				'End': '2019-12-15'
			}, 
			'Total': 
			{
				'BlendedCost': 
				{
					'Amount': '0.2763438289', 
					'Unit': 'USD'
				}
			}, 
			'Groups': [], 
			'Estimated': True
		}
	], 
	'ResponseMetadata': (略)
}

JSON 形式のレスポンスから、ResultsByTime の Total の BlendedCost の Amount を取得する！
ということで、当初は
response['ResultsByTime']['Total']['BlendedCost']['Amount']
なんて指定をしていてうまくいかず。
JSONを分解してみて [0] が必要なことに気づきました。。。
response['ResultsByTime'][0]['Total']['BlendedCost']['Amount']

テスト

画面上部の テスト ボタンを押下した後、イベントテンプレート Hello World のテストが表示されているかと思います。
以下の設定をして、作成 ボタンを押下します。

• イベント名: costchecktest
• テストの中身: {} のみ

画面上部の 保存 ボタンで内容を保存した後、テスト ボタンを押下して、テスト実行します。

テスト完了後、画面上部に実行結果が表示されます。
成功 ではない場合、表示されているログから修正箇所を探り、修正します。

実行トリガの設定

実行結果のすぐ下、Designer の トリガーを追加 ボタンを押下します。
CloudWatch Events を選択した後、新規ルールの作成を選択します。
今回は毎月1日から5日おき(1,6,11...,31日)の日本時間 10:00 に Lambda 関数が呼び出されるようなルールを作成します。
具体的には以下の内容を設定します。

• ルール名: 5Day_1000JST など、わかりやすい名前
• ルールの説明: execute 1,6,11...,31 1000JST など、説明っぽいもの。
• ルールタイプ: スケジュール式
• スケジュール式: cron(0 1 */5 * ? *)
• トリガーの有効化: チェックを入れる

上記設定をして 追加 ボタンを押せば、あとは CloudWatch Events から呼び出されて Lambda 関数が自動実行される日を待つだけです。

曜日部分を * ではなく ? にするのが若干ハマりどころですが、公式ドキュメント を読んでおけば迷いはないはずです(迷った)
また、Cron 式は UTC にて実行されるため、JST で実行するにはマイナス9時間する必要があります(当初はしてなくて思いもよらない時間に実行されてびっくりした)

終わりに

Python 自体の辞書やリストの扱いにも不慣れで、JSON を扱うにも一苦労でした。
色々細かなものを作成して、このあたりの処理については慣れていきたいですね。

油断していたら日付ギリギリぐらいになってしまいました。
来週は余裕をもって投稿したいと思います。