本記事は、BIPROGY / ユニアデックス社内AWSコミュニティ「BIPROGY AWS SPARK」の定期投稿企画第10回目の記事です。
他の定期投稿企画の記事は、インデックスページをご覧ください
目次
1. はじめに
最近業務とは別で、個人のAWS環境でハンズオンや検証をしているのですが、情けないことに、発生してしまう事象が1つあります。それは「リソースの削除し忘れ」です。
これは完全に自分の経験談ですが、検証を短時間で回していると、作ることに集中して片付けが後回しになり、数日後に請求画面やカード明細を見て「あの検証環境、あのリソース、削除したと思ったのにまだ残っていたのか」と気づくことがありました。
特に検証用途では、
- EIPを確保したまま関連付けを外して放置
- 一時的に作ったEBSボリュームの消し忘れ
- 停止中EC2に紐づくEBS課金の見落とし
- NAT Gatewayの消し忘れによる時間課金
に加えて、次のような「検証あるある」も起きることがあると思います。
- ALB/NLBを消したつもりで、ターゲットグループや関連リソースが残る
- RDSを停止しただけで安心し、スナップショットやバックアップ課金を見落とす
- ECRにプッシュした古いイメージが溜まり続ける
- CloudWatch Logsの保持期間を無期限のままにして、ログ保存コストが増える
- OpenSearch/ElastiCacheなどを検証後に縮退・削除し忘れる
挙げたのは一例ですが、特に上記のような状態が起きやすく、気づくのはカード明細や請求タイミングになってから、ということも少なくありません。
「気をつける」だけでは再発しやすいと思い、今回は検出と通知の仕組み自体を作ってみることにしました。
具体的には、まず課金インパクトが大きく、かつ検知しやすいリソースをピックアップし、Lambdaを利用しスキャンします。さらに、その結果をBedrock（Claude）で要約して、SNS経由でメール通知するところまでを、検証用途として簡易的に組んでみました。
この記事では、その実装と動作確認までを簡単にまとめます。
2. やりたいこと
今回のゴールは、次の3点です。
- 未使用/アイドルな課金リソースを定期的に検知する
- 検知結果を人が読みやすい形に整形する
- 通知まで自動化して、放置を早期に防ぐ
対象にしたのは、次の4種類です。
- 未関連付けEIP
- 未アタッチEBS（State: available）
- 停止中EC2（付随EBS課金の可視化）
- アイドルNAT Gateway（CloudWatchメトリクスで判定）
本当は他にも検知したい項目はありますが、まずは「検証で置き去りになりやすく、金額インパクトが読みやすいもの」をピックアップし、仕組みとして構築してみました。
なお、今回検知ロジックを構築するうえで意識したポイントが1つあります。
それは、「存在している」だけでは検知させないことです。
というのも、例えばNAT Gatewayは、存在していても本番で利用中なら消してはいけません。そこで今回はCloudWatchの BytesOutToDestination を見て、直近N日でほぼ通信がないものをアイドル判定するようにしています。
3. 全体アーキテクチャ
構成はシンプルにしてみました。
利用する各サービスおよび、役割は以下になります。
- Lambda: リソース棚卸し + レポート生成 + 通知
- Bedrock（Claude）: 検知結果を読みやすい文章へ整形
- SNS Topic: メール通知
下図の①〜④が、この後に示す処理フローの①〜④に対応しています。
凡例: 🟦 処理フロー（①〜④） 🟨 データ取得・ログ出力
処理フローは以下です（番号は上図と対応）。
-
① Lambdaの呼び出し:
aws lambda invokeでLambdaを呼び出します（本記事の検証手順（5章）ではdemo_setup.shから実行しています） - ② Lambdaの実行（リソース棚卸・判定）: LambdaがリソースをスキャンしてEIP/EBS/EC2/NAT Gatewayの4種類を判定する。一覧は EC2 の Describe API で取得し、NAT Gatewayのアイドル判定だけは CloudWatchメトリクスの通信量も参照する（いずれも図では「データ取得・ログ出力」色のノード）
- ③ 文面生成（Bedrock）: 検知結果をBedrock（Claude）に渡して、要点・優先度付きのメール本文を生成する
- ④ メール通知（SNS Topic）: 生成した本文をSNS経由でメール通知する
なお、②の実行中に出力される処理ログ（print 出力）は CloudWatch Logs に保存されます（記事 5.4 で確認します）。上図でも、②から CloudWatch Logs への出力として示しています。
また、Bedrock呼び出しが失敗しても通知が止まらないように、Lambda側にフォールバック文面（素の検知一覧）も実装しています。
4. 実装の全体像と検知ロジック
この章は、検知のロジックや実装の意図が追えるように、実際のコードを抜粋して説明します。
4.1 デプロイ設定（変数と主要リソース）
この節では、この検証に必要なリソースをデプロイするTerraformコードのうち、主要な部分を紹介します。ここで作成されるリソースは、大きく分けて次の4つです。
- 通知先となる SNSトピック（とメール購読）
- 棚卸しを実行する Lambda関数
- そのLambdaに与える IAM権限（ロール／ポリシー）
- Lambdaの実行ログを保存する CloudWatch Logs ロググループ
まずは可変パラメータ（変数）です。リージョンや通知先メールなど、環境に合わせて変えたい値をここで定義します。
variable "region" {
default = "ap-northeast-1"
}
variable "notification_email" {
description = "通知先メールアドレス（SNSサブスクリプション）"
type = string
}
variable "nat_idle_days" {
default = 7
}
variable "bedrock_model_id" {
default = "jp.anthropic.claude-haiku-4-5-20251001-v1:0"
}
通知先のSNSトピックと、そのメール購読（サブスクリプション）は、以下のように作成しました。
resource "aws_sns_topic" "eip_alert" {
name = "${var.project_name}-alert"
}
resource "aws_sns_topic_subscription" "email" {
topic_arn = aws_sns_topic.eip_alert.arn
protocol = "email"
endpoint = var.notification_email
}
棚卸しを実行するLambda関数の本体です（実際の lambda.tf から、要点となる部分だけを抜粋しています）。環境変数を使って、通知先のSNSトピックARN・利用するBedrockモデル・NAT判定日数をLambdaへ渡しています。
resource "aws_lambda_function" "eip_checker" {
function_name = var.project_name
runtime = "python3.12"
handler = "index.handler"
timeout = 60
environment {
variables = {
SNS_TOPIC_ARN = aws_sns_topic.eip_alert.arn
BEDROCK_MODEL_ID = var.bedrock_model_id
NAT_IDLE_DAYS = tostring(var.nat_idle_days)
}
}
}
続いて、そのLambdaに与えるIAM権限（ポリシー）です。リソースの参照系API・CloudWatch・Bedrock・SNSに対して、必要最小限の権限だけを付与しています。
data "aws_iam_policy_document" "lambda" {
statement {
actions = [
"ec2:DescribeAddresses",
"ec2:DescribeVolumes",
"ec2:DescribeInstances",
"ec2:DescribeNatGateways",
]
resources = ["*"]
}
statement {
actions = ["cloudwatch:GetMetricStatistics"]
resources = ["*"]
}
statement {
actions = ["bedrock:InvokeModel"]
resources = [
"arn:aws:bedrock:*::foundation-model/*",
"arn:aws:bedrock:*:${data.aws_caller_identity.current.account_id}:inference-profile/*",
]
}
statement {
actions = ["sns:Publish"]
resources = [aws_sns_topic.eip_alert.arn]
}
}
ここでのポイントは、Lambdaへ削除系権限を渡さないことです。Lambdaは「棚卸しと通知」に限定しています。
なお、ポリシー内で参照している data.aws_caller_identity.current.account_id（実行中のAWSアカウントIDの取得）は、別ファイルの main.tf で data "aws_caller_identity" "current" {} として宣言しています（本節はコードの抜粋のため、この宣言は省略しています）。
最後に、Lambdaの実行ログを保存する CloudWatch Logs のロググループです。Lambdaの print() などの出力は、ランタイムの仕様で自動的に CloudWatch Logs へ送られます。ここでは保持期間を7日に設定して明示的に作成し、ログが無期限にたまってコスト増につながらないようにしています（このログは 5.4 で確認します）。
resource "aws_cloudwatch_log_group" "lambda" {
name = "/aws/lambda/${var.project_name}"
retention_in_days = 7
}
4.2 Lambda（index.py）の中身と判定基準
まず、環境変数と概算単価を読み込みます。ここで通知先・モデル・NAT判定日数が決まります。
TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]
MODEL_ID = os.environ["BEDROCK_MODEL_ID"]
NAT_IDLE_DAYS = int(os.environ.get("NAT_IDLE_DAYS", "7"))
EIP_MONTHLY = 3.6
EBS_GB_MONTHLY = 0.096
NAT_MONTHLY = 45.0
NAT Gatewayだけは、存在チェックではなくトラフィック実績で判定します。
def _is_nat_idle(nat_id):
end = datetime.now(timezone.utc)
start = end - timedelta(days=NAT_IDLE_DAYS)
resp = cw.get_metric_statistics(
Namespace="AWS/NATGateway",
MetricName="BytesOutToDestination",
Dimensions=[{"Name": "NatGatewayId", "Value": nat_id}],
StartTime=start,
EndTime=end,
Period=86400,
Statistics=["Sum"],
)
total_bytes = sum(d["Sum"] for d in resp.get("Datapoints", []))
return total_bytes < 1_000_000
4種類の検知ロジックは scan() に集約されています。
def scan():
findings = []
# 1) 未関連付けEIP
for a in ec2.describe_addresses().get("Addresses", []):
if not a.get("AssociationId"):
findings.append(_finding("EIP", a.get("AllocationId"), "未関連付け", EIP_MONTHLY))
# 2) 未アタッチEBS
volumes = ec2.describe_volumes().get("Volumes", [])
for v in volumes:
if v.get("State") == "available":
size = v.get("Size", 0)
findings.append(_finding("EBS", v.get("VolumeId"), "未アタッチ", round(size * EBS_GB_MONTHLY, 2)))
# 3) 停止中EC2
resp = ec2.describe_instances(
Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}]
)
# （実装では、停止EC2ごとの付随EBS容量を算出して概算コストを付ける）
# 4) アイドルNAT Gateway
nats = ec2.describe_nat_gateways(
Filter=[{"Name": "state", "Values": ["available"]}]
).get("NatGateways", [])
for ng in nats:
if _is_nat_idle(ng.get("NatGatewayId")):
findings.append(_finding("NATGateway", ng.get("NatGatewayId"), "低トラフィック", NAT_MONTHLY))
return findings
実行の最終段（handler）では、次の順で処理します。
-
scan()の検知結果が 0件なら、通知せずに終了する - 検知ありなら、通知メールの本文を用意する。通常は Bedrock（Claude）で生成し、失敗したときだけ、「handler関数」によってフォールバックの素の一覧に切り替える。成功した場合は、4.3節内の「generate_report関数」でメール本文を作成します。
- 用意した本文を SNS で通知する（Bedrock 生成・フォールバックのどちらの場合でも、通知そのものは必ず行う）
def handler(event, context):
findings = scan()
total_cost = round(sum(f.get("monthly_cost_usd") or 0 for f in findings), 2)
if not findings:
return {"count": 0}
try:
report = generate_report(findings, total_cost)
except Exception:
report = _fallback_text(findings, total_cost)
sns.publish(
TopicArn=TOPIC_ARN,
Subject=f"[AWS棚卸AI] 未使用リソース{len(findings)}件 / 推定月額 約${total_cost:.2f}",
Message=report,
)
return {"count": len(findings), "monthly_cost_usd": total_cost}
4.3 Bedrockの役割
「AIが未使用リソースを見つけている」と思われがちですが、そうではありません。未使用かどうかの判定は、すべて4.2で見たLambdaのコードがAPIの値に基づいて行っています。生成AI（BedrockのClaude）が担当するのは、その判定結果を受け取って、人が読みやすいメール本文に整形する部分のみになります。
役割を整理すると、次のように分担しています。
- 判定（何が未使用か）: Lambdaのコードが担当。判定ルールを明確に記載
- 文章化（どう伝えるか）: 生成AIが担当。件数・コスト・推奨アクションを、そのまま送れるメール文面にまとめる
次のコードが、その「文章化」を行っている部分です。検知結果と推定コストをプロンプトに渡し、生成された本文を受け取っています。
def generate_report(findings, total_cost):
prompt = (
"あなたはAWSコスト最適化アシスタントです...～"
)
resp = bedrock.invoke_model(modelId=MODEL_ID, body=json.dumps(body))
payload = json.loads(resp["body"].read())
return payload["content"][0]["text"]
このように役割を分けることで、「判定を正確に実施」「通知を読みやすく」の両方を成り立たせています。もし判定まで生成AIに任せると、結果が毎回ぶれて棚卸しの信頼性が下がってしまうため、あえて判定はコード側に寄せています。
ここまでは要点ごとに抜粋してきましたが、実際のLambda関数の全文も載せておきます。抜粋で触れた _is_nat_idle / scan / generate_report / _fallback_text / handler が、どう組み合わさっているかを確認できます。
index.py（Lambda関数全文）
import json
import os
from datetime import datetime, timedelta, timezone
import boto3
ec2 = boto3.client("ec2")
cw = boto3.client("cloudwatch")
sns = boto3.client("sns")
bedrock = boto3.client("bedrock-runtime")
TOPIC_ARN = os.environ["SNS_TOPIC_ARN"]
MODEL_ID = os.environ["BEDROCK_MODEL_ID"]
NAT_IDLE_DAYS = int(os.environ.get("NAT_IDLE_DAYS", "7"))
# ap-northeast-1 のおおよその単価（USD）。厳密値ではなく棚卸しの目安。
EIP_MONTHLY = 3.6 # 未関連付けEIP: 約$0.005/h
EBS_GB_MONTHLY = 0.096 # gp3: 約$0.096/GiB-月
NAT_MONTHLY = 45.0 # NAT Gateway: 約$0.062/h（トラフィック別）
def _finding(rtype, rid, detail, monthly_cost):
return {
"type": rtype,
"id": rid,
"detail": detail,
"monthly_cost_usd": monthly_cost,
}
def _is_nat_idle(nat_id):
"""直近 NAT_IDLE_DAYS 日間の送出トラフィックがほぼ0ならアイドルと判定。"""
end = datetime.now(timezone.utc)
start = end - timedelta(days=NAT_IDLE_DAYS)
resp = cw.get_metric_statistics(
Namespace="AWS/NATGateway",
MetricName="BytesOutToDestination",
Dimensions=[{"Name": "NatGatewayId", "Value": nat_id}],
StartTime=start,
EndTime=end,
Period=86400,
Statistics=["Sum"],
)
total_bytes = sum(d["Sum"] for d in resp.get("Datapoints", []))
return total_bytes < 1_000_000 # 1MB未満 = 実質未使用
def scan():
"""複数種類の未使用リソースを横断スキャンして findings のリストを返す。"""
findings = []
# ボリューム情報は使い回す（available判定 と 停止EC2の付随EBSコスト算出）
volumes = ec2.describe_volumes().get("Volumes", [])
gb_by_instance = {}
for v in volumes:
for att in v.get("Attachments", []):
iid = att.get("InstanceId")
if iid:
gb_by_instance[iid] = gb_by_instance.get(iid, 0) + v.get("Size", 0)
# 1. 未関連付けEIP
for a in ec2.describe_addresses().get("Addresses", []):
if not a.get("AssociationId"):
findings.append(_finding(
"EIP", a.get("AllocationId"),
f"PublicIp {a.get('PublicIp')} / どのリソースにも未関連付け",
EIP_MONTHLY,
))
# 2. 未アタッチ(available)なEBSボリューム
for v in volumes:
if v.get("State") == "available":
size = v.get("Size", 0)
findings.append(_finding(
"EBS", v.get("VolumeId"),
f"{size}GiB / {v.get('VolumeType')} / 未アタッチ(available)",
round(size * EBS_GB_MONTHLY, 2),
))
# 3. 停止中EC2（コンピュートは無料だが付随EBSは課金継続）
resp = ec2.describe_instances(
Filters=[{"Name": "instance-state-name", "Values": ["stopped"]}]
)
for r in resp.get("Reservations", []):
for i in r.get("Instances", []):
iid = i.get("InstanceId")
gb = gb_by_instance.get(iid, 0)
findings.append(_finding(
"EC2(stopped)", iid,
f"type={i.get('InstanceType')} / 停止中・付随EBS {gb}GiB が課金継続",
round(gb * EBS_GB_MONTHLY, 2),
))
# 4. アイドルなNAT Gateway（CloudWatchのトラフィックメトリクスで判定）
nats = ec2.describe_nat_gateways(
Filter=[{"Name": "state", "Values": ["available"]}]
).get("NatGateways", [])
for ng in nats:
ngid = ng.get("NatGatewayId")
if _is_nat_idle(ngid):
findings.append(_finding(
"NATGateway", ngid,
f"直近{NAT_IDLE_DAYS}日間の送出トラフィックがほぼ0 / アイドル",
NAT_MONTHLY,
))
return findings
def generate_report(findings, total_cost):
"""Bedrock(Claude)で棚卸レポート(メール本文)を生成する。"""
prompt = (
"あなたはAWSコスト最適化アシスタントです。以下の『未使用リソース棚卸スキャン結果』を、"
"AWS棚卸担当者向けに日本語のメール本文としてまとめてください。\n\n"
"要件:\n"
f"- 冒頭に全体サマリ（検出件数と推定月額合計 約${total_cost:.2f}）\n"
"- リソース種別ごとの内訳と、それぞれの推奨アクション（開放/削除/起動要否の確認など）\n"
"- コスト影響の大きい順に対応優先度を提示\n"
"- 箇条書き中心、そのままメールで送れる体裁。前置きや後書きの雑談は不要\n\n"
"スキャン結果(JSON):\n"
f"{json.dumps(findings, ensure_ascii=False, indent=2)}"
)
body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 1500,
"messages": [{"role": "user", "content": prompt}],
}
resp = bedrock.invoke_model(modelId=MODEL_ID, body=json.dumps(body))
payload = json.loads(resp["body"].read())
return payload["content"][0]["text"]
def _fallback_text(findings, total_cost):
"""Bedrock呼び出しに失敗した場合の素のテキスト（デモを止めないための保険）。"""
lines = [
f"- [{f['type']}] {f['id']} : {f['detail']}（推定 ${f['monthly_cost_usd']}/月）"
for f in findings
]
return (
f"未使用リソースを{len(findings)}件検出しました（推定月額合計 約${total_cost:.2f}）。\n\n"
+ "\n".join(lines)
)
def handler(event, context):
findings = scan()
total_cost = round(sum(f.get("monthly_cost_usd") or 0 for f in findings), 2)
print(f"findings={len(findings)} total_monthly_usd={total_cost}")
if not findings:
print("未使用リソースなし。通知をスキップします。")
return {"count": 0}
try:
report = generate_report(findings, total_cost)
except Exception as e: # モデル未有効化などでも通知自体は出す
print(f"Bedrock生成に失敗、素の一覧で通知します: {e}")
report = _fallback_text(findings, total_cost)
sns.publish(
TopicArn=TOPIC_ARN,
Subject=f"[AWS棚卸AI] 未使用リソース{len(findings)}件 / 推定月額 約${total_cost:.2f}",
Message=report,
)
print(f"SNSへ通知しました。count={len(findings)}")
return {"count": len(findings), "monthly_cost_usd": total_cost}
5. 検証手順と確認結果
この章では、実際に手を動かして「4種類の未使用リソースを検知できること」を確認します。
検証の全体の流れは次の3ステップで、上から順に実行します。
- デプロイ（5.1）: TerraformでLambda・SNSなどを作成する
-
ダミー作成＋Lambda実行（5.2）:
demo_setup.shを実行して検知対象をわざと作り、続けてLambdaを実行する - 結果の確認（5.3・5.4）: Lambdaの戻り値とCloudWatch Logsを見て、検知できたかを判定する
事前に必要なもの: AWS CLI と Terraform が使える環境、および認証済みのデプロイ先AWSアカウント、Bedrockから使用するモデルのアクセス有効化
5.1 検証前に実行したコマンド
まず、検証対象となるLambda・SNSなどのAWSリソースを、Terraformでまとめて作成（デプロイ）します。実行するのは次の4つのコマンドで、それぞれ次のことをしています。
-
cp terraform.tfvars.example terraform.tfvars- 今回の検証では、設定値の記入例として
terraform.tfvars.exampleというファイルを用意しました。これをコピーして、自分用の設定ファイルterraform.tfvarsを作成します - 作成したら
terraform.tfvarsを開き、通知先メールアドレス（notification_email）を自分のアドレスに書き換えます。この値が、後で通知メールの届く先になります
- 今回の検証では、設定値の記入例として
-
terraform init… プロバイダなどの初期化 -
terraform plan… 作成されるリソースの差分を確認 -
terraform apply… 実際にAWSへリソースを作成
コピー元となる terraform.tfvars.example の中身は次のとおりです。notification_email が通知先メールの設定になります。
# このファイルを terraform.tfvars にコピーして値を埋めてください
notification_email = "you@example.com"
# 以下は任意で上書きできる項目の例です（コメントのままなら variables.tf の既定値が使われます）
# region = "ap-northeast-1"
# project_name = "unused-resource-ai"
# bedrock_model_id = "jp.anthropic.claude-haiku-4-5-20251001-v1:0"
# nat_idle_days = 7
# で始まる行は「任意で上書きできる項目の例」です。コメントを外さなければ、4.1 の variables.tf の既定値がそのまま使われます（例: bedrock_model_id の既定は jp.anthropic.claude-haiku-4-5-20251001-v1:0）。
これをコピーし、notification_email を自分のアドレスに書き換えたものが、実際に使う terraform.tfvars です。今回は通知先メールだけ変更すれば動きます。
notification_email = "xxxx@example.com" # ← 自分のメールアドレスに変更
なぜコピーするのか？
terraform.tfvars は .gitignore で Git 管理対象外にしており、実際のメールアドレスなど環境ごとの値をリポジトリに残さないようにしています。そのぶん「何を設定すべきか」が分からなくなるため、記入例として terraform.tfvars.example（こちらはコミットする）を用意し、これをコピーして自分の値を埋める、という流れにしています。notification_email は必須項目（デフォルト値なし）なので、この設定を記載しています。
以上をふまえた4コマンドが次のとおりです。
cp terraform.tfvars.example terraform.tfvars
terraform init
terraform plan
terraform apply
terraform apply を実行すると、しばらくして設定した宛先にAWSから「SNSの購読確認メール」が届きます。メール本文中の承認リンクをクリックしてください。この承認をしないと、後でLambdaが通知を送っても、メールが手元に届きません。
5.2 ダミーリソースの作成とLambda実行（demo_setup.sh）
5.1でリソースのデプロイが終わったら、検証用スクリプト demo_setup.sh を実行します。このスクリプトは、①〜④の検知対象リソースをわざと作成し、続けてLambdaを実行するところまでを1本で行います。
実行方法は、スクリプトを叩くだけです。
./demo_setup.sh
default VPC が無い環境では、使用するサブネットを指定して実行します。
SUBNET_ID=subnet-xxxx ./demo_setup.sh
以下は、このスクリプトの中身のうち「検知対象を作成している部分」の抜粋です（①〜④は2章で挙げた検知対象に対応しています）。
# ① 未関連付けEIP
EIP_ALLOC_1="$(aws ec2 allocate-address ... --query 'AllocationId' --output text)"
# ② 未アタッチEBS（1GiB）
VOL_ID="$(aws ec2 create-volume ... --size 1 --volume-type gp3 --query 'VolumeId' --output text)"
# ③ EC2を起動して停止（停止中EC2 + 付随EBS課金の再現）
INSTANCE_ID="$(aws ec2 run-instances ... --query 'Instances[0].InstanceId' --output text)"
aws ec2 wait instance-running --instance-ids "$INSTANCE_ID"
aws ec2 stop-instances --instance-ids "$INSTANCE_ID"
aws ec2 wait instance-stopped --instance-ids "$INSTANCE_ID"
# ④ NAT Gatewayを作成（作りたてで低トラフィック状態）
EIP_ALLOC_2="$(aws ec2 allocate-address ... --query 'AllocationId' --output text)"
NAT_ID="$(aws ec2 create-nat-gateway ... --query 'NatGateway.NatGatewayId' --output text)"
aws ec2 wait nat-gateway-available --nat-gateway-ids "$NAT_ID"
さらにこのスクリプト内の最後で、Lambdaを実行しています。
FUNC="$(terraform output -raw lambda_function_name)"
aws lambda invoke --function-name "$FUNC" out.json >/dev/null
cat out.json
aws lambda invoke を実行するとLambda関数（index.py）が動き、その戻り値が out.json というファイルに保存されます。最後の cat out.json でその中身を表示しています。この戻り値の見方は、次の5.3で説明します。
以下に、demo_setup.sh の全文を載せました。サブネットの自動選択や、作成したリソースIDを .demo_resources.env に記録する処理（デモ用リソースであるEIPのIDなどを失念しない為）なども含まれています。
demo_setup.sh（全文）
#!/usr/bin/env bash
#
# demo_setup.sh — ①〜④の検知対象ダミーリソースを作成し、Lambdaを手動実行して検知を確認する。
#
# ① 未関連付けEIP : allocate-address（未関連付けのまま）
# ② 未アタッチEBS : create-volume 1GiB gp3（available）
# ③ 停止中EC2 : run-instances t2.micro → stop（付随EBSが課金継続）
# ④ アイドルNAT Gateway : create-nat-gateway（作りたて＝トラフィック0で即アイドル判定）
#
# 作成したリソースIDは .demo_resources.env に記録し、demo_teardown.sh が読んで確実に削除する。
# コストを抑えるため、確認後は速やかに demo_teardown.sh を実行すること。
#
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
STATE_FILE="${SCRIPT_DIR}/.demo_resources.env"
REGION="${AWS_REGION:-${AWS_DEFAULT_REGION:-ap-northeast-1}}"
MARKER="unused-resource-ai-demo"
TAGSPEC_TMPL="Tags=[{Key=DemoMarker,Value=${MARKER}},{Key=Name,Value=${MARKER}}]"
echo "== リージョン: ${REGION} =="
# --- subnet / VPC / AZ を特定 -------------------------------------------------
# SUBNET_ID が環境変数で指定されていればそれを使う（デフォルトVPCが無い環境向け）。
# 未指定なら default VPC の先頭サブネットを自動選択する。
if [[ -n "${SUBNET_ID:-}" ]]; then
echo "SUBNET_ID=${SUBNET_ID}（指定値を使用）"
else
VPC_ID="$(aws ec2 describe-vpcs --region "$REGION" \
--filters Name=isDefault,Values=true \
--query 'Vpcs[0].VpcId' --output text)"
if [[ "$VPC_ID" == "None" || -z "$VPC_ID" ]]; then
echo "!! default VPC が見つかりません。既存サブネットを SUBNET_ID で指定して再実行してください。" >&2
echo " 候補: aws ec2 describe-subnets --region ${REGION} \\" >&2
echo " --query 'Subnets[].{SubnetId:SubnetId,VpcId:VpcId,AZ:AvailabilityZone}' --output table" >&2
echo " 例: SUBNET_ID=subnet-xxxx ./demo_setup.sh" >&2
exit 1
fi
SUBNET_ID="$(aws ec2 describe-subnets --region "$REGION" \
--filters Name=vpc-id,Values="$VPC_ID" \
--query 'Subnets[0].SubnetId' --output text)"
fi
# サブネットから VPC と AZ を確定（指定・自動どちらの場合も整合させる）
read -r VPC_ID AZ < <(aws ec2 describe-subnets --region "$REGION" \
--subnet-ids "$SUBNET_ID" \
--query 'Subnets[0].[VpcId,AvailabilityZone]' --output text)
echo "VPC=${VPC_ID} SUBNET=${SUBNET_ID} AZ=${AZ}"
# --- ① 未関連付けEIP ----------------------------------------------------------
echo "-- ① EIP を作成 --"
EIP_ALLOC_1="$(aws ec2 allocate-address --region "$REGION" --domain vpc \
--tag-specifications "ResourceType=elastic-ip,${TAGSPEC_TMPL}" \
--query 'AllocationId' --output text)"
echo " EIP AllocationId=${EIP_ALLOC_1}"
# --- ② 未アタッチEBS ----------------------------------------------------------
echo "-- ② EBS(1GiB gp3) を作成 --"
VOL_ID="$(aws ec2 create-volume --region "$REGION" \
--availability-zone "$AZ" --size 1 --volume-type gp3 \
--tag-specifications "ResourceType=volume,${TAGSPEC_TMPL}" \
--query 'VolumeId' --output text)"
echo " VolumeId=${VOL_ID}"
# --- ③ 停止中EC2 --------------------------------------------------------------
echo "-- ③ EC2(t2.micro) を起動 → 停止 --"
AMI_ID="$(aws ssm get-parameters --region "$REGION" \
--names /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64 \
--query 'Parameters[0].Value' --output text)"
INSTANCE_ID="$(aws ec2 run-instances --region "$REGION" \
--image-id "$AMI_ID" --instance-type t2.micro --count 1 \
--subnet-id "$SUBNET_ID" \
--tag-specifications "ResourceType=instance,${TAGSPEC_TMPL}" \
--query 'Instances[0].InstanceId' --output text)"
echo " InstanceId=${INSTANCE_ID} (running待ち...)"
aws ec2 wait instance-running --region "$REGION" --instance-ids "$INSTANCE_ID"
aws ec2 stop-instances --region "$REGION" --instance-ids "$INSTANCE_ID" >/dev/null
echo " stopped待ち..."
aws ec2 wait instance-stopped --region "$REGION" --instance-ids "$INSTANCE_ID"
echo " 停止完了"
# --- ④ アイドルNAT Gateway ----------------------------------------------------
echo "-- ④ NAT Gateway を作成（作りたて＝トラフィック0で即アイドル判定）--"
EIP_ALLOC_2="$(aws ec2 allocate-address --region "$REGION" --domain vpc \
--tag-specifications "ResourceType=elastic-ip,${TAGSPEC_TMPL}" \
--query 'AllocationId' --output text)"
NAT_ID="$(aws ec2 create-nat-gateway --region "$REGION" \
--subnet-id "$SUBNET_ID" --allocation-id "$EIP_ALLOC_2" \
--tag-specifications "ResourceType=natgateway,${TAGSPEC_TMPL}" \
--query 'NatGateway.NatGatewayId' --output text)"
echo " NatGatewayId=${NAT_ID} (available待ち... 1〜2分)"
aws ec2 wait nat-gateway-available --region "$REGION" --nat-gateway-ids "$NAT_ID"
echo " NAT GW available"
# --- 状態ファイルに記録 -------------------------------------------------------
cat > "$STATE_FILE" <<EOF
REGION=${REGION}
EIP_ALLOC_1=${EIP_ALLOC_1}
VOL_ID=${VOL_ID}
INSTANCE_ID=${INSTANCE_ID}
NAT_ID=${NAT_ID}
EIP_ALLOC_2=${EIP_ALLOC_2}
EOF
echo "== 作成したリソースIDを ${STATE_FILE} に記録しました =="
# --- Lambda を手動実行して検知確認 -------------------------------------------
echo "-- Lambda を手動実行して検知を確認 --"
if FUNC="$(cd "$SCRIPT_DIR" && terraform output -raw lambda_function_name 2>/dev/null)" && [[ -n "$FUNC" ]]; then
OUT="${SCRIPT_DIR}/out.json"
aws lambda invoke --region "$REGION" --function-name "$FUNC" "$OUT" >/dev/null
echo " Lambda戻り値: $(cat "$OUT")"
echo " => count が 4 であれば①〜④すべて検知。メール本文とログもご確認ください。"
echo " ログ: aws logs tail \"\$(terraform output -raw log_group_name)\" --since 5m --region ${REGION}"
else
echo "!! terraform output からLambda関数名を取得できませんでした（未applyの可能性）。"
echo " apply済みなら手動で: aws lambda invoke --function-name <name> out.json --region ${REGION}"
fi
echo ""
5.3 どの値を見て「成功」と判断したか
5.2の最後で cat out.json を実行すると、Lambdaが返した結果（戻り値）が表示されます。これは自分で入力する値ではなく、Lambdaの実行結果として出力されるJSONで、次のような内容になります。
{"count": 4, "monthly_cost_usd": 48.7}
このJSONを見て、次の基準で「成功」と判断します。
count == 4-
monthly_cost_usdが0より大きい
count == 4 なら、今回意図的に作った次の4種類が全て拾えていると判断できます。
- 未関連付けEIP
- 未アタッチEBS
- 停止中EC2
- アイドルNAT Gateway
5.4 CloudWatch Logsで見た確認ポイント
戻り値（out.json）だけでなく、Lambdaが出力したログでも中身を確認できます。次のコマンドを自分で実行すると、直近5分ぶんのログが表示されます。
aws logs tail "$(terraform output -raw log_group_name)" --since 5m
表示されたログで、次の3点を確認します。
-
findings=... total_monthly_usd=...が出ている - 検知0件時は
未使用リソースなし。通知をスキップします。で終わる - Bedrock呼び出し失敗時はフォールバック通知へ切り替わる
この3点で、「検知」「整形」「通知」のどこで問題が起きたか切り分けを行うことができます。
5.5 この検証で確認できたこと
今回の手動検証で、次を確認できました。
- 4種類の検知ロジックが意図どおり動く
- 検知結果をBedrockで読みやすく整形して通知できる
- Bedrock失敗時もフォールバックで通知経路は維持される
6. 実際の通知メール比較（モデル生成とフォールバック）
ここでは、同じ検知結果でも「モデル生成のメール」と「フォールバック時のメール」で、受け取る印象がどう違うかを比較します。
6.1 モデル生成時のメール（例）
以下に、実際に受信したモデル生成メールのスクリーンショットを掲載します。
モデルにより、優先度付けやリソース、コスト、推奨アクション等をわかりやすくまとめてくれます。
＊実際は他のリソースについてもメール下部に記載されてます。
6.2 フォールバック時のメール（例）
ちなみにフォールバックの文面はこのようになります。Bedrock呼び出しに失敗しても、最低限の検知結果は受け取れるようになっています。
7. まとめ
個人検証環境の課金漏れ対策として、
- Lambdaで決定的な検知を行い
- Bedrockで人が読みやすいレポート化を行い
- SNSで通知する
という一連の流れを作ることで、削除し忘れの早期発見がしやすくなりました。
今回の実装で特に良かった点は、次の2つと考えています。
- 検知ロジックはコードで明確化し、再現性を持たせたこと
- AIは説明・優先度付けに限定し、障害時はフォールバックで止めないこと
一方で、今回の記事で示した検証は、あくまで動作確認を優先したため一部手動です。
具体的には demo_setup.sh の実行や検証タイミングの調整など、オペレーションとしては手を入れる前提になっています。
実運用を考えるなら、ここは「削除漏れに気づくまでの時間を短くする」方向で自動化を進めるのが理想かと思います。
- EventBridgeで定期的にLambdaを起動し、毎日/毎週の棚卸しを自動実行する
- 通知先や判定閾値などの設定をコード管理し、環境差分を扱いやすくする
- 必要に応じて承認フローを挟み、承認後に自動削除までつなぐ
この形にすることで、「気づいた人がたまたま片付ける」運用から、
「放置しづらい仕組みで継続的に回す」運用へ移行できます。
今後の拡張としては、
- 検知対象の追加（ALB、RDS、古いSnapshotなど）
- マルチリージョン対応
- 承認フロー付きの自動削除（Step Functions連携）
あたりが実用面で効果が高いと考えています。
とはいえ、「まずは削除漏れを検知し、通知で気づけるようにする」という第一歩としては、十分使える構成にできたかなと思います。是非参考にしていただけると嬉しいです！
BIPROGYグループの技術への取り組み
We Are Hiring!