LambdaとNameタグで特定システムのコストを監視してみた

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "CostExplorerAccess",
            "Effect": "Allow",
            "Action": [
                "ce:GetCostAndUsage"
            ],
            "Resource": "*"
        },
        {
            "Sid": "SNSPublish",
            "Effect": "Allow",
            "Action": [
                "sns:Publish"
            ],
            "Resource": [
                "arn:aws:sns:ap-northeast-1:123456789012:your-sns-cost-alert-daily-topic",
                "arn:aws:sns:ap-northeast-1:123456789012:your-sns-cost-alert-monthly-topic"
            ]
        }
    ]
}

"""
AWS Lambda: Cost Explorer 予算超過アラート通知

【概要】
Cost Explorerを使用して、特定のNameタグ(プレフィックス)を持つリソースのコストを取得し、
日次および月次の予算を超過していた場合にAmazon SNS経由で通知を行います。
月次予算の通知については、外部データベース(DynamoDB等)を使わずに「当月1回のみ」
通知されるように、「昨日までのコスト」と「一昨日までのコスト」を比較して閾値判定を行っています。

【環境変数】
- PREFIX                : 検索対象とするNameタグのプレフィックス（例: "my-project-"）
- DAILY_BUDGET          : 日次コストの予算閾値（数値 例: 10.0）
- MONTHLY_BUDGET        : 月次コストの予算閾値。カンマ区切りで複数指定可能（例: "100, 500, 1000"）
- DAILY_SNS_TOPIC_ARN   : 日次予算超過時の通知先SNSトピックARN
- MONTHLY_SNS_TOPIC_ARN : 月次予算超過時の通知先SNSトピックARN
"""

import boto3
import os
from datetime import datetime, timedelta
from botocore.config import Config
from botocore.exceptions import ClientError
import time
import random

# ===== 設定 =====
PREFIX = os.environ['PREFIX']
DAILY_BUDGET = float(os.environ['DAILY_BUDGET'])

# 環境変数の "100, 200, 500" のようなカンマ区切り文字列をリスト化し、数値の昇順に並べ替える
MONTHLY_BUDGETS = sorted([float(x.strip()) for x in os.environ['MONTHLY_BUDGET'].split(',')])

DAILY_SNS_TOPIC_ARN = os.environ['DAILY_SNS_TOPIC_ARN']
MONTHLY_SNS_TOPIC_ARN = os.environ['MONTHLY_SNS_TOPIC_ARN']

config = Config(
    retries={
        'max_attempts': 10,
        'mode': 'adaptive'
    }
)

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

# ===== Cost Explorer呼び出し =====
def call_ce_with_retry(params):
    for attempt in range(10):
        try:
            return ce.get_cost_and_usage(**params)
        except ClientError as e:
            code = e.response['Error']['Code']
            if code in ['ThrottlingException', 'TooManyRequestsException']:
                sleep_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Throttle detected. Retry in {sleep_time:.2f}s")
                time.sleep(sleep_time)
            else:
                raise
    raise Exception("Max retries exceeded")

# ===== コスト取得 =====
def get_cost(start, end):
    total = 0.0
    next_token = None

    while True:
        params = {
            'TimePeriod': {
                'Start': start,
                'End': end
            },
            'Granularity': 'DAILY',
            'Metrics': ['UnblendedCost'],
            'GroupBy': [
                {
                    'Type': 'TAG',
                    'Key': 'Name'
                }
            ]
        }

        if next_token:
            params['NextPageToken'] = next_token

        response = call_ce_with_retry(params)

        for result in response['ResultsByTime']:
            for group in result['Groups']:
                raw_key = group['Keys'][0] if group['Keys'] else ''

                # Name$xxxx → xxxx に変換
                tag_value = raw_key.split('$', 1)[-1] if '$' in raw_key else raw_key

                if tag_value.startswith(PREFIX):
                    amount = float(group['Metrics']['UnblendedCost']['Amount'])
                    total += amount

        next_token = response.get('NextPageToken')
        if not next_token:
            break

    return total

# ===== SNS通知 =====
def send_alert(topic_arn, subject, message):
    sns.publish(
        TopicArn=topic_arn,
        Subject=subject,
        Message=message
    )

# ===== Lambda本体 =====
def lambda_handler(event, context):
    today = datetime.utcnow().date()
    yesterday = today - timedelta(days=1)
    
    # ===== 日次（前日分） =====
    daily_cost = get_cost(str(yesterday), str(today))

    # ===== 月次 =====
    first_day = today.replace(day=1)
    monthly_cost_now = 0.0
    monthly_cost_before = 0.0
    
    # Cost Explorerは仕様上 Start と End を同じ日付にできないため、日付によって取得処理を分ける
    if today.day == 1:
        # 当月1日はまだ当月分のデータがないためスキップ
        pass
    elif today.day == 2:
        # 2日の場合は比較対象(前日までの月間累計)を0とする
        monthly_cost_now = get_cost(str(first_day), str(today))
        monthly_cost_before = 0.0
    else:
        # 3日以降は「昨日までの当月コスト」と「一昨日までの当月コスト」を取得する
        monthly_cost_now = get_cost(str(first_day), str(today))
        monthly_cost_before = get_cost(str(first_day), str(yesterday))

    print("=== Cost Report ===")
    print(f"Daily cost: {daily_cost:.2f}")
    print(f"Monthly cost (as of yesterday): {monthly_cost_now:.2f}")
    print("===================")

    # ===== 日次アラート判定 =====
    if daily_cost > DAILY_BUDGET:
        print(f"[Daily] 日次予算({DAILY_BUDGET:.2f})を超過しています。通知を送信します。")
        message = (
            f"[Daily Budget Exceeded]\n"
            f"Cost: {daily_cost:.2f}\n"
            f"予算: {DAILY_BUDGET:.2f}\n"
        )
        send_alert(DAILY_SNS_TOPIC_ARN, "Daily Cost Alert", message)
    else:
        print(f"[Daily] 日次コストは予算({DAILY_BUDGET:.2f})内です。")

    # ===== 月次アラート判定 =====
    exceeded_budgets = []
    
    # 複数設定された月次予算の閾値それぞれに対して状況を判定し、ログ出力する
    print("\n=== Monthly Budget Check ===")
    for budget in MONTHLY_BUDGETS:
        if monthly_cost_now > budget:
            if monthly_cost_before <= budget:
                print(f"[Monthly] 予算 {budget:.2f} を新規に超過しました（通知対象）")
                exceeded_budgets.append(budget)
            else:
                print(f"[Monthly] 予算 {budget:.2f} を超過していますが、既に通知済みのためスキップします")
        else:
            print(f"[Monthly] コストは予算 {budget:.2f} を超過していません")
    print("============================")

    # 1つ以上の閾値を新しく超えていた場合のみ通知を送信
    if exceeded_budgets:
        # 複数の閾値を同時に超えた場合は、一番高い閾値を基準にする
        max_exceeded_budget = max(exceeded_budgets)
        
        message = (
            f"[Monthly Budget Exceeded]\n"
            f"Cost: {monthly_cost_now:.2f}\n"
            f"超過した予算: {max_exceeded_budget:.2f}\n"
            f"(設定されている全予算: {', '.join(map(str, MONTHLY_BUDGETS))})\n"
        )
        
        send_alert(MONTHLY_SNS_TOPIC_ARN, "Monthly Cost Alert", message)

    return {
        "daily_cost": daily_cost,
        "monthly_cost": monthly_cost_now,
        "exceeded_budgets": exceeded_budgets
    }