AWSのマネジメントコンソールからLambdaを組んで、AWSの利用料を毎日Slackに通知してみた

AWSの毎日の請求額をSlackに通知してくれるアプリを作成しました。元の記事ではSAMから作成していましたので、今回は別アプローチをとって、マネコンからぽちぽちしてアプリを作成します。

おのやん

2023.08.19

はじめに

こんにちは！おのやんです

みなさん、AWSの月額の請求額を知りたくないですか？私はめちゃくちゃ知りたいです。

DevelopersIOには、「AWSサービス毎の請求額を毎日Slackに通知してみた」というブログがあります。こちらは先輩が執筆してくださいました。

こちらのブログでは、AWS サーバーレスアプリケーションモデル(SAM)を用いてLambdaなどを構築していました。

私はこちらの記事を参考にして、AWS マネジメントコンソールからブラウザぽちぽちしてLambdaを構築しました。

ということで、今回は元の記事とは別のアプローチをとって、AWSの利用料を毎日Slackに通知してみます

前提

SlackチャンネルへのIncoming Webhookアプリ追加方法は、元記事と同じです。そのため、Webhook URLは取得できているものとします。

今回は、SAMで構築されていた部分をマネコン上で組んでみるという内容で進めていきます。

IAMロール作成

元の記事によると、Cost Explorer用のIAMロールがなかったので作成する必要があります。ということで、まずはIAMロールを作成しましょう。

IAMの画面から、「ロールを作成」を選択します。

今回はIAMロールをLambdaにアタッチしたいので、Lambdaを選択します。

現時点では、まだポリシーが作成されていません。なので、「ポリシーを作成」からポリシー作成画面へと進みましょう。

検索ボックスで「Logs」などと検索すると、CloudWatch Logsがサジェストされると思います。ここで、CloudWatch Logsをクリックします。

すると、CloudWatch LogsのAPI一覧が検索できるようになります。ここで、CreateLogGroup、CreateLogStream、PutLogEventsを選択します。

選択した状態だと、下の方にリソースARNを指定する項目があります。こちらの「このアカウント内のいずれか」にチェックを入れます。

これが完了したら、「許可をさらに追加」を押して,,,

Cost Explorerを検索・選択して...

Cost Explorerのポリシーも追加しておきましょう。

なお、ここではARNのチェックボックスを選択する必要はありません。

ポリシー名は、元の記事を同じくNotifySlackToBillingLambdaPolicyにしました。こちらからポリシーを作成しましょう。

ちゃんと作成できていますね！

こちらを、今度はIAMロールに付与していきます。

先ほどの許可ポリシー選択画面にて、NotifySlackToBillingLambdaPolicyを検索してチェックし、「次へ」を押します。

ロール名は、元記事と同じくBillingIamRoleにしておきます。名前を入力したら次に進みます。

無事にIAMロールが作成できました。こちらのIAMロールを、これから作成するLambdaにアタッチしていきます。

Lambda作成

続いで、LambdaでSlackに通知を送信する関数を作成していきます。

今回は、元記事のコードを少しだけ変更したこちらのコードを使っていきます。

import os
import boto3
import json
import requests
from datetime import datetime, timedelta, date


SLACK_WEBHOOK_URL = os.environ['SLACK_WEBHOOK_URL']


def lambda_handler(event, context) -> None:
    client = boto3.client('ce', region_name='us-east-1')

    # 合計とサービス毎の請求額を取得する
    total_billing = get_total_billing(client)
    service_billings = get_service_billings(client)

    # Slack用のメッセージを作成して投げる
    (title, detail) = get_message(total_billing, service_billings)
    post_slack(title, detail)


def get_total_billing(client) -> dict:
    (start_date, end_date) = get_total_cost_date_range()

    # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ce.html#CostExplorer.Client.get_cost_and_usage
    response = client.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(client) -> list:
    (start_date, end_date) = get_total_cost_date_range()

    # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ce.html#CostExplorer.Client.get_cost_and_usage
    response = client.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 get_message(total_billing: dict, service_billings: list) -> (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'{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') return title, '\n'.join(details) def post_slack(title: str, detail: str) -> None:
    # https://api.slack.com/incoming-webhooks
    # https://api.slack.com/docs/message-formatting
    # https://api.slack.com/docs/messages/builder
    payload = {
        'attachments': [
            {
                'color': '#36a64f',
                'pretext': title,
                'text': detail
            }
        ]
    }

    # http://requests-docs-ja.readthedocs.io/en/latest/user/quickstart/
    try:
        response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(payload))
    except requests.exceptions.RequestException as e:
        print(e)
    else:
        print(response.status_code)


def get_total_cost_date_range() -> (str, str):
    start_date = get_begin_of_month()
    end_date = get_today()

    # 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_begin_of_month() -> str:
    return date.today().replace(day=1).isoformat()


def get_prev_day(prev: int) -> str:
    return (date.today() - timedelta(days=prev)).isoformat()


def get_today() -> str:
    return date.today().isoformat()

Lambdaの画面にて、新規にLambdaを作成します。