個人でAWSアカウントを持っているとAWS利用額についてはとても気になります。
AWSには請求アラームを使って設定したしきい値を超えたらアラーム通知する機能もありますが、
日々どのくらい増えているのか把握しておきたいので毎日利用額をSlackへ通知する仕組みを作ってみました
構成
AWS Lambdaを使ってAWS利用料を取得しWebhookでSlackへ通知を行うシンプルな構成です
EventBridgeで毎日 9:00に定期的な実行する
Slackの設定
① Slackに通知用チャンネルを作成する
チャンネル名はわかりやすければ何でも良いです
② Webhook URLを発行する
Webhookの発行手順はこちらを参考にしてください
Lambda関数
Lambda関数の作成
以下の内容で関数を作成
関数名: 任意の名前
ランタイム: Python 3.10
アーキテクチャ: x86_64
実行ロール: 基本的な Lambda アクセス権限で新しいロールを作成
Lambda関数の設定
タイムアウト: 30秒
実行ロール
Lambda関数が実行に使われれるロールにCloudWatchReadOnlyAccessを追加する
環境変数
環境変数に以下を追加する
slackChannel: Slackの設定で用意したチャンネル名
slackPostURL: Slackの設定で用意したWebhookURL
Lambdaレイヤー
外部ライブラリにrequestsを使用しています
このままでは外部ライブラリを読み込めずエラーになるのでレイヤーを使って使用できるようにします
自作でも良いのですが少し手間を省きたいので有志の方が用意されたレイヤーを使用します
ここに記載されているrequestのARN情報を指定しレイヤーを追加してください
Lambda関数のコード
Lambdaコードはこちら
app.py
#!/usr/bin/env python
# encoding: utf-8
import json
import datetime
import requests
import boto3
import os
import logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def get_parameter_store_value(parameter_name):
"""Parameter Storeから値を取得する"""
try:
ssm = boto3.client('ssm')
response = ssm.get_parameter(Name=parameter_name, WithDecryption=True)
return response['Parameter']['Value']
except Exception as e:
logger.error(f"Error getting parameter {parameter_name}: {e}")
raise
SLACK_POST_URL = get_parameter_store_value('/eslamate/slack/post-url')
SLACK_CHANNEL = get_parameter_store_value('/eslamate/slack/channel')
def get_account_info():
"""AWSアカウント情報を取得する"""
try:
sts = boto3.client('sts')
account_id = sts.get_caller_identity()['Account']
# アカウントエイリアスを取得（設定されている場合）
try:
iam = boto3.client('iam')
aliases = iam.list_account_aliases()['AccountAliases']
account_name = aliases[0] if aliases else None
except Exception as e:
logger.warning(f"Could not get account alias: {e}")
account_name = None
return account_id, account_name
except Exception as e:
logger.error(f"Error getting account info: {e}")
return None, None
def get_billing_data():
"""AWS Billingデータを取得する"""
try:
cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')
# 過去2日間のデータを取得（データの遅延を考慮）
end_time = datetime.datetime.utcnow()
start_time = end_time - datetime.timedelta(days=2)
response = cloudwatch.get_metric_statistics(
Namespace='AWS/Billing',
MetricName='EstimatedCharges',
Dimensions=[
{
'Name': 'Currency',
'Value': 'USD'
}
],
StartTime=start_time,
EndTime=end_time,
Period=86400,
Statistics=['Maximum']
)
logger.info(f"CloudWatch response: {response}")
datapoints = response.get('Datapoints', [])
if not datapoints:
logger.warning("No billing data available yet")
return None, None
# データポイントを時間順にソート（最新のデータを取得）
sorted_datapoints = sorted(datapoints, key=lambda x: x['Timestamp'], reverse=True)
latest_data = sorted_datapoints[0]
cost = latest_data['Maximum']
date = latest_data['Timestamp'].strftime('%Y年%m月%d日')
return cost, date
except Exception as e:
logger.error(f"Error getting billing data: {e}")
return None, None
def build_message(cost, date, account_id, account_name):
"""Slackメッセージを構築する"""
# アカウント情報の表示文字列を作成
if account_name:
account_info = f"{account_name} ({account_id})"
else:
account_info = account_id if account_id else "不明"
if cost is None:
# データがない場合のメッセージ
# 利用料のしきい値は任意で設定してください私は$10を目安にしています
title = f"AWS利用料金通知 - {account_info}"
text = "AWS利用料金のデータがまだ利用できません。"
color = "#808080" # gray
else:
if float(cost) >= 10.0:
color = "#ff0000" # red
elif float(cost) > 8.0:
color = "warning" # yellow
else:
color = "good" # green
title = f"AWS利用料金通知 - {account_info}"
text = f"{date}までのAWS利用料は、${cost:.2f}です。"
attachment = {
"title": title,
"text": text,
"color": color,
"fields": [
{
"title": "アカウント",
"value": account_info,
"short": True
},
{
"title": "利用料金",
"value": f"${cost:.2f}" if cost is not None else "データなし",
"short": True
}
],
"footer": "AWS Billing",
"footer_icon": "https://a0.awsstatic.com/libra-css/images/logos/aws_logo_smile_1200x630.png",
"ts": int(datetime.datetime.now().timestamp())
}
return attachment
def lambda_handler(event, context):
"""Lambda関数のメインハンドラー"""
try:
# アカウント情報を取得
account_id, account_name = get_account_info()
# Billingデータを取得
cost, date = get_billing_data()
# メッセージを構築
content = build_message(cost, date, account_id, account_name)
# Slackメッセージを作成
slack_message = {
'channel': SLACK_CHANNEL,
'attachments': [content],
}
# Slackに送信
headers = {
'Content-Type': 'application/json'
}
response = requests.post(
SLACK_POST_URL,
data=json.dumps(slack_message),
headers=headers
)
if response.status_code != 200:
logger.error(f"Slack API returned error: {response.status_code} - {response.text}")
return {
'statusCode': 500,
'body': json.dumps('Failed to post message to Slack')
}
logger.info(f"Message posted to {slack_message['channel']}")
return {
'statusCode': 200,
'body': json.dumps('Successfully posted billing info to Slack')
}
except requests.exceptions.RequestException as e:
logger.error(f"Request failed: {e}")
return {
'statusCode': 500,
'body': json.dumps(f'Request failed: {str(e)}')
}
except Exception as e:
logger.error(f"Unexpected error: {e}")
return {
'statusCode': 500,
'body': json.dumps(f'Unexpected error: {str(e)}')
}
動作確認
Lambdaをテスト実行して通知がされていればOK！
定期実行
① Amazon EventBridgeのスケジュールを選択
② 実行する時間をcron形式で入力
毎日9:00に実行をする設定にしています
③ ターゲットにLambdaを先ほど作成したLambda関数名を選びます
④ 設定はデフォルトの状態のままで次へをおしてスケジュールの作成は完了です
さいごに
少し手間はかかりますが比較的簡単に通知できるので日々の利用額が把握できて安心です
アラーム通知だけだと少し不安と思われる方が是非試してみてください