【AWS】AWSサービスごとの請求額をSlackへ通知する仕組みの構築

どうも、AWS勉強中のとがみんです。

前回の記事でAWSの合計利用金額を集計しSlackへ通知しました。
この記事ではAWSサービスごとの利用金額を取得しSlackへ通知する仕組みをつくっていきます。

Contents

今回作成するもの
作業手順概要
作業手順詳細
- ソースコードの追加修正
- 動作確認
まとめ
参考

今回作成するもの

構成は前回のものとは変わらず、サービス毎に金額を取得し、Slackへ通知されるようにソースコードを修正していき、下記のような通知がSlackに通知されるようにします。

構成は下記記事と同じで、今回はソースコードを追記＆改修します。

作業手順概要

前回作成したソースコードにサービスごとに金額が出力されるコードを追加修正
動作確認

作業手順詳細

ソースコードの追加修正

ソースコード全体に関しては、下記のGitHubリポジトリの「feature/each_service」に挙げています。

処理をハンドリングする関数

def handler(event, context) -> None:

    # CostExplorer
    client = boto3.client('ce', region_name='us-east-1')

    # 対象月にかかったAWSの合計金額の算出
    total_billing = get_total_billing(client)

    # 対象月にかかったAWSのサービス毎の利用金額の算出
    service_billings = get_service_billings(client)

    # メッセージの作成
    (title,detail) = get_message(total_billing, service_billings)

    # Slackへの通知
    post_slack(title, detail)

def handler(event, context) -> None:

# CostExplorer

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

# 対象月にかかったAWSの合計金額の算出

total_billing = get_total_billing(client)

# 対象月にかかったAWSのサービス毎の利用金額の算出

service_billings = get_service_billings(client)

# メッセージの作成

(title,detail) = get_message(total_billing, service_billings)

# Slackへの通知

post_slack(title, detail)

全体の流れとしては、CostExplorerのインスタンス化、対象月の合計を取得する関数の呼び出し、AWSサービス毎の利用金額を取得する関数の呼び出し、合計金額と、サービス毎の利用金額のデータを元に、Slackへ通知するメッセージを作成する関数の呼び出し、作成したメッセージをSlackへ通知する関数の呼び出しを行なっています。

サービス毎の金額を取得する関数

# サービス毎の利用金額を取得
def get_service_billings(client) -> list:
    (start_date, end_date) = get_monthly_cost_date_range()

    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_service_billings(client) -> list:

(start_date, end_date) = get_monthly_cost_date_range()

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

GroupByの項目を活用することで、AWSのサービス毎の金額を取得しています。

Slackへ通知するメッセージの作成

# Slackへ通知するメッセージの作成
def get_message(total_billing: dict, service_billings: list) -> (str, str):
    title = total_billing['start'] + '〜' + total_billing['end'] + 'のAWS利用費用の合計金額は' + str(round(float(total_billing['billing']), 2)) + 'です。'
    details = []
    for item in service_billings:
        service_name = item['service_name']
        billing = round(float(item['billing']), 2)
        if billing == 0.0:
            continue
        details.append(f'{service_name}：{billing}$')
    return title, '\n'.join(details)

# Slackへ通知するメッセージの作成

def get_message(total_billing: dict, service_billings: list) -> (str, str):

title = total_billing['start'] + '〜' + total_billing['end'] + 'のAWS利用費用の合計金額は' + str(round(float(total_billing['billing']), 2)) + 'です。'

details = []

for item in service_billings:

service_name = item['service_name']

billing = round(float(item['billing']), 2)

if billing == 0.0:

continue

details.append(f'{service_name}：{billing}$')

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

Slackへ通知するメッセージを作成しています。

Slackへ通知する関数の作成

# Slackへ通知する関数
def post_slack(title: str, detail: str) -> None:
    # Slack通知内容の作成
    payload = {
        'attachments': [
            {
                'color': '#36a64f',
                'pretext': title,
                'text': detail
            }
        ]
    }
    # Slackへの通知
    try:
        response = requests.post(SLACK_WEBHOOK_URL, data=json.dumps(payload))
    except requests.exceptions.RequestException as e:
        print(e)
    else:
        print(response.status_code)