💸

AWS Lambdaで今月の請求額をTeamsに通知させる

2024/04/19に公開

本記事はAWS LambdaでTeamsにAWS使用量を通知する仕組みをPythonを用いて作成した話を紹介します。

きっかけ

会社でAWSを使用することが決まりAWS使用料を把握したいとの声が出ました。
請求額アラートであればBudgetsで閾値を設定してあげればいいですが、日々どのくらい使っているのか通知してほしかったので毎日通知してくれる仕組みを作ることに。

ツール選択

TeamsにAWSから通知を飛ばす場合の選択肢はおそらく次の2つです。
1.AWS Chatbot
2.AWS Lambda

前者は、Teamsのチャンネルリンクを取得して連携
後者はIncoming WebhookのURLを使って通知を飛ばします。

最初は簡単そうかつ便利そうなChatbotを検討していたのですが、チャンネルリンクを承認する権限が私のTeamsアカウントになかったので後者で作成しました。
*手順は後述します。

Incoming Webhook x AWS Lambdaの記事は検索すれば色々ヒットしたのですが、ここから少しハマりました。

ここから私の奮闘記が挟まりますので興味ない方は目次で『通知例』辺りまで飛ばしてください。

我儘

ネットの海に転がっているコードの多くはPythonで書かれていたのですが、
いずれも以下の一文が埋め込まれていたのです。

import requests

requestsモジュールです。

requestsモジュールは、PythonのHTTPクライアントライブラリであり、HTTPリクエストを送信するための使いやすいインターフェースを提供します。このモジュールを使用することで、ウェブページの取得、APIへのアクセス、データの送信など、さまざまなHTTPリクエストを簡潔に行うことができます。
--ChatGPT

このrequestsモジュール、サードパーティ製のためPython3.8以降では標準ライブラリから外れています。
ただ、非常に使いやすく多くの方に活用されています。

私はというと
1.現在AWS LambdaではPython3.8以降しか使えない!
2.モジュールのインポートとかやりたくない!てか、標準ライブラリから外すなら何か代わりがあるんだろ!?

以上の理由からPython3.8以降の標準ライブラリだけでAWS請求額のレポート通知コードを組みました。

ChatGPTを使って

とにかく新しいPythonで動作してほしいのでChatGPTが知ってる最新バージョン(3.10)で組んでもらいました。

通知例

完成したLambda関数の実行例です。

作成方法

1.Teams側の準備
2.Lambda用IAMロールの作成
3.Lambda関数の作成
4.(おまけ)EventBridgeの設定

の4段階でお届けします。

Teams側の準備

通知を送りたいチームにIncoming Webhookを導入します。

*旧Teamsの場合
チームのメニュー「…」から「コネクタ」を選択
『Incoming Webhook』で検索しIncoming Webhookを「追加」してください。

*新Teamsの場合
チームメニュー→チャンネルを管理→コネクタ→編集
検索画面が開きます。

Incoming Webhook URLの作成
Incoming Webhookの構成から設定画面を開きます。
着信名を入力し作成ボタンを押すとURLが生成されますので控えておいてください。
*アイコン画像もカスタマイズできるようです。

設定が完了するとテストメッセージがチームに届きます。

Lambda用IAMロールの作成

今回の仕組みはLambdaでCostExplorerから取得するので、LambdaにCostExplorerの読み取り権限を付与する必要があります。

CostExplorerの読み取り権限ポリシーはないのでそこから作っていきます。

IAM→ポリシー→ポリシーの作成

サービス名:Cost Explorer Service
アクション許可:GetCostAndUsage
[次へ]

ポリシー名:任意の名前
[ポリシーの作成]

続いてこのポリシーを使ってLambda用のIAMロールを作ります。

IAM→ロール→ロールを作成

信頼されたエンティティタイプ:AWSのサービス
ユースケース:Lambda
[次へ]

許可ポリシー:先ほど作成したポリシーを選択
[次へ]

ロール名:任意の名前
[ロールを作成]

Lambda関数の作成

それでは、Lambda関数を作っていきます。

マネジメントコンソールからLambda→関数→[関数の作成]と進みます。

オプション:一から作成
関数名:任意の名前
ランタイム:Python(3.8~3.12で動作確認済みです)
デフォルトの実行ロールの変更
実行ロール:既存のロールを使用する→Lambda用に作ったIAMロールを選択
他はデフォルトのまま
[関数の作成]

コード

import boto3
import datetime
import json
import urllib.request

def lambda_handler(event, context):
    # AWS Cost Explorer APIのクライアントを作成
    ce_client = boto3.client('ce', region_name='us-east-1')  # 使用するリージョンを適宜変更
    
    # 現在の年月を取得
    current_date = datetime.datetime.utcnow()
    current_year = current_date.year
    current_month = current_date.month
    
    # 前月の開始日と終了日を計算
    start_date = datetime.datetime(current_year, current_month, 1)
    end_date = start_date.replace(day=1, month=start_date.month+1) - datetime.timedelta(days=1)
    
    # Cost Explorer APIでサービスごとの請求額を取得
    response = ce_client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date.strftime('%Y-%m-%d'),
            'End': end_date.strftime('%Y-%m-%d')
        },
        Granularity='MONTHLY',
        Metrics=['BlendedCost'],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'SERVICE'
            }
        ]
    )
    
    # サービスごとの請求額を取得
    service_costs = {}
    total_cost = 0.0
    for group in response['ResultsByTime'][0]['Groups']:
        service = group['Keys'][0]
        cost = group['Metrics']['BlendedCost']['Amount']
        if float(cost) > 0:  # 請求額が0でない場合のみ追加
            service_costs[service] = cost
            total_cost += float(cost)
    
    # Teamsに通知するメッセージを作成
    message = {
        "text": "AWSの今月の請求額の内訳です:<br />"
    }
    for service, cost in service_costs.items():
        message["text"] += f"{service}: ${cost}<br />"
    
    # 合計請求額をメッセージに追加
    message["text"] += f"<br />合計請求額: ${total_cost}"
    
    # TeamsのIncoming WebhookのURL
    webhook_url = "YOUR_TEAMS_INCOMING_WEBHOOK_URL"
    
    # TeamsにHTTPリクエストを送信して通知
    data = json.dumps(message).encode('utf-8')
    headers = {'Content-Type': 'application/json'}
    req = urllib.request.Request(webhook_url, data=data, headers=headers, method='POST')
    try:
        with urllib.request.urlopen(req) as response:
            response_data = response.read().decode('utf-8')
            print(response_data)
    except Exception as e:
        print(f"Error: {e}")
    
    return {
        'statusCode': 200,
        'body': "Notification sent successfully"
    }

以下の内容を自身の環境に合わせて入力してください。
region_name='us-east-1':請求額を確認したいリージョン
#TeamsのIncoming WebhookのURL:Teamsの準備で控えたURLを""内に入力
webhook_url = "YOUR_TEAMS_INCOMING_WEBHOOK_URL"

実際に通知が届くかテストしてみましょう。

テスト手順
*保存していない変更がある場合はまずDeployしてください。
Test
→テストイベントの設定が開くのでイベント名を入力し他はデフォルトのままで[保存]
→Testボタンを押すとコードが実行されます

IAMロールを後から設定する
関数作成の段階でロールを準備していなかった場合は以下の手順で設定できます。

関数作成ページの「設定」タブから「アクセス権限」を選択
「編集」ボタンからIAMロールを変更できます。

(おまけ)EventBridgeの設定

完成した関数を定期的に動かしたい場合はEventBridgeでルールを活用します。

CloudWatchのルール or Amazon EventBridgeのルール
[ルールを作成]

名前:任意の文字列
ルールタイプ:スケジュール
他はデフォルトのまま
[EventBridge Scheduler で続行]

頻度:定期的なスケジュール
スケジュールの種類:cronベースのスケジュール
cron式の書き方はこちらを参照してください。
https://docs.aws.amazon.com/ja_jp/eventbridge/latest/userguide/eb-cron-expressions.html

フレックスタイムウィンドウ:オフ
他はデフォルトのままで大丈夫です。
[次へ]

ターゲットの詳細:AWS Lambda Invoke
Invoke Lambda関数:今回作成したLambda関数
[次へ]

スケジュール完了後のアクション:NONE
他はデフォルトのままで大丈夫です。
[次へ]

最後に内容を確認して
[スケジュールを作成]

おわり

以上で定期的にAWS請求額をTeamsに通知してくれるLambda関数の完成です。

お疲れ様でした。

Discussion