AWSのコストや利用状況を定期的にメールとLINEに通知してもらう

はじめに

こんにちは、やくもです。
最近AIネタが多かったのですが、今回は珍しくちょっと離れた話題
定期的にAWSのコストが知りたいなあと思いサクッと作りました。

今回やったこと

構成図

構成は以下のようになっています（めっちゃシンプル）
EventBridgeによりコスト状況を取得するLambdaを定期的に実行し、SNSを利用してメールに配信しています。
Bedrockを使っても良かったのですが、機械的に数字を取得するだけでいいのでわざわざ呼び出すという選択はしませんでした。

• EventBridge：Lambdaを定期駆動（毎週）
• Lambda：AWS コストエクスプローラー APIを使用して、習慣のコスト状況を取得し、メール文を作成
• SNS：メールに配信

AWS Cost Explorer APIの利用

また、今回コストの取得にはCost Explorer APIを利用しました。

resp = ce.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
    )

詳しい解説や例は以下のドキュメントやリファレンスをご覧ください。

配信イメージ

実際に届くメールは以下のようになります。
毎週このレポートを確認して、自身のコストの利用状況を確認できるようになります。
私は今までBudgetを利用したアラートを設定していましたが、1万円超えないと通知来なかったのでありがたいです。

SNSトピックの作成

では、ここから具体的な実装手順に入っていきます。
まずSNSの設定を行います。

トピックの作成

適当な名前を入れてトピックを作成します。

サブスクリプションの作成

次にサブスクリプションの作成をします。
今回はメールに通知を飛ばしたいので、「Eメール」を選択し、メアドを入力する。

その後入力したメアド宛に以下のように、「このサブスクリプション許可する？」といった通知が来るので、リンクを踏んで許可する。

許可後はブラウザが開きこんな画面になる

私はこの時迷惑メールフォルダに振られていたので、一応迷惑メールフォルダも確認してみるといいかもしれないです。
これを許可した後は、通常の受信フォルダに来るようになりました。

Lambda関数の作成

次にメインの処理を行うラムダ関数を作成します。

ラムダ関数の記述

実際に書いたコードは以下のようになります。
AWS コストは上でも触れましたが、コストエクスプローラーAPIを利用して取得しています。

lambda_function.py

import os
import boto3
from datetime import datetime, timedelta, timezone
from decimal import Decimal

ce = boto3.client("ce")
sns = boto3.client("sns")

SNS_TOPIC_ARN = os.environ["SNS_TOPIC_ARN"].strip()
JST = timezone(timedelta(hours=9))


def _fmt_money(amount: str) -> str:
    try:
        v = Decimal(amount)
    except Exception:
        return amount
    return f"{v.quantize(Decimal('0.01'))}"


def _last_week_range_jst(anchor: datetime):
    """
    先週(月曜00:00)〜今週(月曜00:00) を返す（Cost ExplorerはEndが排他的）
    """
    this_monday = (anchor - timedelta(days=anchor.weekday())).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    last_monday = this_monday - timedelta(days=7)
    start = last_monday.date().isoformat()
    end = this_monday.date().isoformat()
    return start, end


def lambda_handler(event, context):
    now_jst = datetime.now(JST)
    start, end = _last_week_range_jst(now_jst)

    resp = ce.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="DAILY",
        Metrics=["UnblendedCost"],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
    )

    service_totals = {}
    currency = "USD"

    for day in resp.get("ResultsByTime", []):
        for g in day.get("Groups", []):
            service = g["Keys"][0]
            amount = g["Metrics"]["UnblendedCost"]["Amount"]
            currency = g["Metrics"]["UnblendedCost"]["Unit"]
            service_totals[service] = service_totals.get(service, Decimal("0")) + Decimal(amount)

    grand_total = sum(service_totals.values(), Decimal("0"))

    TOP_N = 15
    sorted_items = sorted(service_totals.items(), key=lambda x: x[1], reverse=True)
    top = sorted_items[:TOP_N]
    rest = sorted_items[TOP_N:]
    others_total = sum((v for _, v in rest), Decimal("0"))

    # ===== 日本語メール本文（ここが関数内に入っているのが重要）=====
    subject = f"[AWS] 週間コストレポート {start} 〜 {end}"

    lines = []
    lines.append("AWS 週間コストレポート")
    lines.append("")
    lines.append(f"対象期間：{start} 〜 {end}（※終了日は含まれません）")
    lines.append("")
    lines.append("■ 合計コスト")
    lines.append(f"  {_fmt_money(str(grand_total))} {currency}")
    lines.append("")
    lines.append("■ サービス別内訳（上位）")

    for svc, val in top:
        lines.append(f"- {svc}: {_fmt_money(str(val))} {currency}")

    if rest:
        lines.append(f"- その他: {_fmt_money(str(others_total))} {currency}")

    lines.append("")
    lines.append("※ Cost Explorer の UnblendedCost を使用しています")

    message = "\n".join(lines)

    sns.publish(
        TopicArn=SNS_TOPIC_ARN,
        Subject=subject[:100],
        Message=message,
    )

    return {
        "status": "ok",
        "start": start,
        "end": end,
        "total": str(grand_total),
        "currency": currency,
    }

環境変数のセット

SNS_TOPIC_ARNといった環境変数にSNSのトピックのARNを環境変数として格納しています。

EventBridgeスケジュールの作成

毎週駆動させたいので、EventBridgeのスケジュールを作成します。
私は一旦毎週日曜朝8時に接待しています。

cron式について

cron式は初見だとちょっとクセがありますが、分解すると以下のような構成になっています。

項目	値
分	0
時間	8
日付	?
月	*
曜日	SUN
年	*

また、私の環境だとEventBridgeの起動するタイムゾーンがデフォルトだとUTCになっていたので、忘れずにAsia/Tokyoに変更しておきましょう。
でないと9時間ズレて起動してしまいます。

メールの確認

ここまでできれば定期的にコストレポートがメールに送信されるかと思います。
全体の流れを動作確認したい場合は、EventBridgeのスケジュールを一時的に5分ごとにするなどしてもいいかと思います。

LINEにも送信したい

せっかくなので、LINEにも送信してみようと思います。

公式アカウント作成

アカウント作成などの諸々の準備は以前もやっているのでよければご覧ください。

Lambda関数の作成

今のメールに送信しているレポートだと、LINEに送るには少し長いので要点に絞ったものを送ろうと思います。
LINEに送信する際、以下のようなLambda関数を作成します。

コード全文はこちら

lambda_function.py

import os
import json
import calendar
import urllib.request
from datetime import datetime, timedelta, timezone
from decimal import Decimal, ROUND_HALF_UP
import boto3

# ===== AWS Clients =====
ce = boto3.client("ce")

# ===== Timezone =====
JST = timezone(timedelta(hours=9))

# ===== Environment variables =====
LINE_CHANNEL_ACCESS_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"].strip()
LINE_TO = os.environ["LINE_TO"].strip()  # Uxxxx / Cxxxx / Rxxxx
COST_METRIC = os.environ.get("COST_METRIC", "UnblendedCost").strip()  # UnblendedCost / AmortizedCost etc.


# ---------- helpers ----------
def _q2(amount: Decimal) -> str:
    return str(amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))


def _week_to_date_range_jst(now: datetime):
    """
    今週(月曜00:00 JST) 〜 明日(00:00 JST)  (Endは排他的)
    """
    this_monday = (now - timedelta(days=now.weekday())).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    tomorrow_0 = (now + timedelta(days=1)).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    return this_monday.date().isoformat(), tomorrow_0.date().isoformat()


def _last_week_range_jst(now: datetime):
    """
    前週(月曜00:00 JST) 〜 今週(月曜00:00 JST) (Endは排他的)
    """
    this_monday = (now - timedelta(days=now.weekday())).replace(
        hour=0, minute=0, second=0, microsecond=0
    )
    last_monday = this_monday - timedelta(days=7)
    return last_monday.date().isoformat(), this_monday.date().isoformat()


def _elapsed_days_in_week(now: datetime) -> int:
    # 月曜=0 ... 日曜=6 -> 1..7
    return now.weekday() + 1


def _fetch_cost_by_service(start: str, end: str):
    """
    Cost Explorerから期間内のサービス別コストを集計して返す
    """
    resp = ce.get_cost_and_usage(
        TimePeriod={"Start": start, "End": end},
        Granularity="DAILY",
        Metrics=[COST_METRIC],
        GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
    )

    service_totals: dict[str, Decimal] = {}
    currency = "USD"

    for day in resp.get("ResultsByTime", []):
        for g in day.get("Groups", []):
            service = g["Keys"][0]
            metric = g["Metrics"][COST_METRIC]
            currency = metric["Unit"]
            service_totals[service] = service_totals.get(service, Decimal("0")) + Decimal(
                metric["Amount"]
            )

    total = sum(service_totals.values(), Decimal("0"))
    return total, currency, service_totals


def _forecast_week_end(wtd_total: Decimal, elapsed_days: int) -> Decimal:
    """
    週末予想：現時点の平均/日 × 7日（シンプル予測）
    """
    if elapsed_days <= 0:
        return wtd_total
    return (wtd_total / Decimal(elapsed_days)) * Decimal(7)


def _build_line_report(
    now: datetime,
    start: str,
    end: str,
    wtd_total: Decimal,
    prev_week_total: Decimal,
    forecast: Decimal,
    currency: str,
    service_totals: dict[str, Decimal],
):
    top3 = sorted(service_totals.items(), key=lambda x: x[1], reverse=True)[:3]

    lines = []
    lines.append("📊 今週のAWSコスト利用料")
    lines.append("")

    # 今週合計
    lines.append(f"💰 現在の合計: {_q2(wtd_total)} {currency}")

    # 前週比（前週が0なら出さない）
    if prev_week_total == 0:
        lines.append("📈 前週比: データ不足")
    else:
        diff = wtd_total - prev_week_total
        pct = (diff / prev_week_total) * Decimal(100)
        sign = "+" if diff >= 0 else ""
        lines.append(
            f"📈 前週比: {sign}{_q2(diff)} {currency} "
            f"({sign}{_q2(pct)}%)"
        )

    lines.append("")
    lines.append("🏷 上位3サービス:")
    for svc, val in top3:
        lines.append(f"- {svc}: {_q2(val)} {currency}")

    lines.append("")
    lines.append(f"🤖 週末予想: {_q2(forecast)} {currency}")

    return "\n".join(lines)


def _line_push(text: str):
    """
    LINE Messaging API Push
    """
    url = "https://api.line.me/v2/bot/message/push"
    body = {"to": LINE_TO, "messages": [{"type": "text", "text": text}]}
    data = json.dumps(body, ensure_ascii=False).encode("utf-8")

    req = urllib.request.Request(
        url,
        data=data,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
        },
    )
    with urllib.request.urlopen(req, timeout=10) as res:
        return res.status, res.read().decode("utf-8")


# ---------- Lambda entry ----------
def lambda_handler(event, context):
    now = datetime.now(JST)

    # 今週（途中）: 今週月曜〜明日0時
    start, end = _week_to_date_range_jst(now)
    wtd_total, currency, service_totals = _fetch_cost_by_service(start, end)

    # 前週（確定）: 前週月曜〜今週月曜
    pw_start, pw_end = _last_week_range_jst(now)
    prev_week_total, _, _ = _fetch_cost_by_service(pw_start, pw_end)

    # 週末予想
    elapsed_days = _elapsed_days_in_week(now)
    forecast = _forecast_week_end(wtd_total, elapsed_days)

    # LINE文面
    msg = _build_line_report(
        now=now,
        start=start,
        end=end,
        wtd_total=wtd_total,
        prev_week_total=prev_week_total,
        forecast=forecast,
        currency=currency,
        service_totals=service_totals,
    )

    # 送信
    status, resp = _line_push(msg)

    return {
        "status": "ok",
        "line_status": status,
        "line_resp": resp[:500], 
        "period_this_week": {"start": start, "end": end, "total": str(wtd_total), "currency": currency},
        "period_prev_week": {"start": pw_start, "end": pw_end, "total": str(prev_week_total)},
        "forecast_week_end": str(forecast),
    }

環境変数の確認

環境変数は二種類
チャンネルアクセストークン（LINE_CHANNEL_ACCESS_TOKEN）と、ユーザーID（LINE_TO）を使用します。

動作確認

実際に動かすとこんな感じで

• 現在のコスト
• 前週比での比較
• コストが大きいサービスなんかを教えてくれます

あれ、Neptuneでコスト発生してる...笑
検証用に一瞬だけ立ち上げていたのですが、相変わらず金食い虫ですね..

さいごに

ここまででAWSのコスト状況を教えてくれる構成を作ってみました。
メールとLINE２パターンに分けて通知先を作ってあります。
今のままだとコストしか知れないので、現状からどういったアクションをするといいのか的な提案もしてくれるといいのかなと思ったけど、それはまた今度。

参考