AWS Lambdaを利用してAWS利用料金をSlackのチャンネルへ通知する

はじめに

社内で利用しているAWSアカウントの利用料金について、RDSやEC2の停止忘れ等で
うっかり料金が嵩んでいないか検知するために、利用料金をSkackに通知する仕組みを作成しました。

実装方式

EventBridgeをトリガーとしてLambdaを実行することでCost Explorerから利用料金を取得、
メッセージを成形してWebhookを使ってSlackへ投稿します。

取得する利用料金はLambda実行日の2日前の1日分の料金と月次の累積料金としました。
Lambda実行日を基準とした料金ではなく、2日前の値を取得している点については、
以下の公式ドキュメントやAWS利用料金の確定についての参考記事の記載を考慮して、
なるべく確定済みの料金を取得するために2日前としています。

すべてのコストには、前日までの使用量が反映されます。例えば、今日の日付が 12 月 2 日だとすると、データには 12 月 1 日までの使用状況が反映されます。

現在の請求期間では、データは請求アプリケーションのアップストリームデータに依存し、一部のデータが 24 時間より後に更新される場合があります。

環境詳細

以下の環境設定で実装しました。

• Lambdaランタイム: Python 3.13
• その他Lambda設定:デフォルト
• Lambda利用IAMロール権限: 後述

EventBridge(毎日実行)の設定やSlack側の設定については省略

IAMポリシー

CloudWatchLogsへのログ出力権限と、利用料金取得のためce:GetCostAndUsageを許可します。

Lambda用ポリシー

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogGroup",
                "logs:CreateLogStream",
                "logs:PutLogEvents",
                "ce:GetCostAndUsage"
            ],
            "Resource": "*"
        }
    ]
}

Lambdaコード

以下のコードを作成しました。

import boto3
import datetime
import json
import logging
import os
import urllib.request
from botocore.exceptions import ClientError

# --- 環境変数 ---
SLACK_WEBHOOK_URL = os.environ["SLACK_WEBHOOK_URL"]

# --- ロガー設定 ---
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# --- boto3設定　---
cost_explorer_client = boto3.client("ce")

# --- タイムゾーンをJSTに設定　---
JST = datetime.timezone(datetime.timedelta(hours=9))

def get_two_days_ago_cost(client):
    # Cost Explorer APIを使用して、Lambda実行日の2日前のコストを取得

    # APIアクセスのため、取得期間の算出
    today_jst = datetime.datetime.now(JST)
    two_days_ago_jst = today_jst - datetime.timedelta(days=2)
    start_date = two_days_ago_jst.strftime("%Y-%m-%d")
    end_date = today_jst.strftime("%Y-%m-%d") # Endは期間に含まれないため、当日の日付を指定

    logger.info(f"Fetching cost data for the period: {start_date}")

    try:
        # Cost Explorer APIからコスト取得
        response = client.get_cost_and_usage(
            TimePeriod={
                "Start": start_date,
                "End": end_date
            },
            Granularity="DAILY",
            Metrics=["UnblendedCost"]
        )

        # APIレスポンスからコスト情報を抽出
        cost_data = response["ResultsByTime"][0]["Total"]["UnblendedCost"]
        amount = float(cost_data["Amount"])
        unit = cost_data["Unit"]

        logger.info(f"2 days ago cost: {amount} {unit}")

        return {"date": start_date, "amount": amount, "unit": unit}

    except (ClientError, IndexError) as e:
        logger.error(f"Failed to get 2 days ago cost from Cost Explorer: {e}")
        return None

def get_month_to_date_cost(client):
    # Cost Explorer APIを使用して、Lambda実行日の2日前の属する月の1日から、Lambda実行日の2日前までの累積コストを取得
    today_jst = datetime.datetime.now(JST)

    # 期間の終了日を「2日前」に設定
    end_date_dt = today_jst - datetime.timedelta(days=2)
    end_date = end_date_dt.strftime("%Y-%m-%d")

    # 期間の開始日を「終了日の属する月の1日」に設定
    start_date = end_date_dt.strftime("%Y-%m-01")

    logger.info(f"Fetching month-to-date cost from {start_date} to {end_date}")

    try:
        # Cost Explorer APIからコスト取得
        response = client.get_cost_and_usage(
            TimePeriod={
                "Start": start_date,
                "End": end_date
            },
            Granularity="MONTHLY",
            Metrics=["UnblendedCost"]
        )

        # APIレスポンスからコスト情報を抽出
        cost_data = response["ResultsByTime"][0]["Total"]["UnblendedCost"]
        amount = float(cost_data["Amount"])
        unit = cost_data["Unit"]

        logger.info(f"Month-to-date cost: {amount} {unit}")

        return {"start_date": start_date, "end_date": end_date, "amount": amount, "unit": unit}

    except (ClientError, IndexError) as e:
        logger.error(f"Failed to get month-to-date cost from Cost Explorer: {e}")
        return None

def build_slack_payload(two_days_ago_cost_info, month_to_date_cost_info):
    # Slackに投稿するメッセージを作成
    # どちらかの情報が取得できていない場合はエラーメッセージを返す
    if two_days_ago_cost_info is None or month_to_date_cost_info is None:
        return { "text": "AWSコスト情報の取得に失敗しました。" }

    # 2日前コストの情報
    two_days_ago_date_str = two_days_ago_cost_info["date"]
    two_days_ago_amount = two_days_ago_cost_info["amount"]
    two_days_ago_unit = two_days_ago_cost_info["unit"]

    # 月次累積コストの情報
    month_to_date_start_date_str = month_to_date_cost_info["start_date"]
    month_to_date_end_date_str = month_to_date_cost_info["end_date"]
    month_to_date_amount = month_to_date_cost_info["amount"]
    month_to_date_unit = month_to_date_cost_info["unit"]

    # メッセージのfieldsを作成
    fields = [
        {
            "title": f"{two_days_ago_date_str} のご利用料金",
            "value": f":heavy_dollar_sign: {two_days_ago_amount:,.2f} {two_days_ago_unit}",
            "short": False
        },
        {
            "title": f"{month_to_date_start_date_str} ~ {month_to_date_end_date_str} の累積料金",
            "value": f":heavy_dollar_sign: {month_to_date_amount:,.2f} {month_to_date_unit}",
            "short": False
        }
    ]

    payload = {
        "attachments": [
            {
                "title": "AWSコストレポート",
                "fields": fields,
                "footer": "AWS Cost Explorer"
            }
        ]
    }
    return payload

def post_to_slack(payload):
    # Slackへのpost実行
    try:
        req = urllib.request.Request(
            SLACK_WEBHOOK_URL,
            data=json.dumps(payload).encode("utf-8"),
            headers={"Content-Type": "application/json"},
            method="POST"
        )
        
        with urllib.request.urlopen(req) as response:
            if response.getcode() != 200:
                raise Exception(f"Slack API returned non-200 status code: {response.getcode()}")
        logger.info("Message posted to Slack successfully using urllib.")
    except Exception as e:
        logger.error(f"Failed to post to Slack: {e}")
        raise

def lambda_handler(event, context):
    try:
        # 2日前のコストを取得
        two_days_ago_cost_info = get_two_days_ago_cost(cost_explorer_client)

        # 月次累積コストを取得
        month_to_date_cost_info = get_month_to_date_cost(cost_explorer_client)

        # どちらかのコストが取得できない場合は通知をスキップ
        if two_days_ago_cost_info is None or month_to_date_cost_info is None:
            logger.info("One of the cost data sets is not available. Skipping notification.")
            return {"statusCode": 200, "body": json.dumps("No cost data to report.")}

        # Slackに投稿
        payload = build_slack_payload(two_days_ago_cost_info, month_to_date_cost_info)
        post_to_slack(payload)

        return {"statusCode": 200, "body": json.dumps("Successfully posted AWS cost to Slack.")}

    except Exception as e:
        logger.error(f"An unexpected error occurred: {e}", exc_info=True)
        return {"statusCode": 500, "body": json.dumps("An error occurred.")}

実際のSlack投稿

以下のような形でSlackに投稿されます。

料金通知専用のチャンネルを作成してそこに毎日自動投稿されるように設定すれば、
日々の料金をAWSマネジメントコンソールにログインすることなく確認することができます。
平常時と比較して数日間利用料金が高い状態が続いているようなら
落とし忘れ・消し忘れに気づくトリガーとして使えます。

以上。

参考記事