月200万円のAWSコスト地獄から脱出｜放置リソースを自動検出する仕組み

月200万円が消える悪夢から始まった

先日チームのAWS請求書を眺めていたら、3時間ぐらい呆然としてた。月200万円が「使ってないリソース」に吸い込まれてたんですよ。正直、その瞬間は本気で泣きかけた。

具体的には、こんなのが放置されてた：

• 開発環境のELBが5個。どれが生きてるかわからん
• RDSスナップショットが山盛り。3年分くらい
• 誰も使ってないS3バケットが8個
• 停止状態のEC2が15台。電源切れてるけど課金は続いてる
• CloudFront配信停止してるのにディストリビューションがハングしたまま

この地獄から脱出するために、3ヶ月かけて自動化の仕組みを作った。チームで共有したら「これ最初からあったらよかったのに」って言われたので、実装パターンをそのまま書きます。

AWS構成と自動検出の基本設計

全体像をこんな感じで設計しました。

graph TB
    subgraph "Data Collection"
        CT[CloudTrail]
        CW[CloudWatch Events]
        CG[Config]
    end
    
    subgraph "Detection Layer"
        LAMBDA[Lambda<br/>Resource Scanner]
        ANALYSIS[Resource Analysis<br/>Cost Estimation]
    end
    
    subgraph "Storage & State"
        DDB[DynamoDB<br/>Unused Resources]
        S3LOG[S3<br/>Audit Logs]
    end
    
    subgraph "Notification & Action"
        SNS[SNS]
        SLACK[Slack<br/>通知]
        CONSOLE[Cost Console<br/>Dashboard]
    end
    
    CT --> LAMBDA
    CW --> LAMBDA
    CG --> LAMBDA
    LAMBDA --> ANALYSIS
    ANALYSIS --> DDB
    ANALYSIS --> S3LOG
    DDB --> SNS
    SNS --> SLACK
    DDB --> CONSOLE
    
    style LAMBDA fill:#FF9900
    style DDB fill:#527FFF
    style SLACK fill:#E01E5A

基本的な流れは、まずCloudTrail・Config・CloudWatch Eventsでリソースの活動を記録する。そこからLambdaが定期的に走って「使われてないリソース」を検出。その結果をDynamoDBに棚卸しして、Slackでチームに通知。監査ログはS3に保存しておく、って感じ。これだけあれば、月1回の自動検出でほぼ全リソースをカバーできるんですよ。

実装パターン：Lambdaで不要リソースを検出する

ここからがメイン。実装コードを公開します。正直、これは完璧ではなくて、チーム運用する中で何度も改善してきたやつです。

1. 最初にやるべき：各リソースタイプの判定ロジック

import boto3
import json
from datetime import datetime, timedelta
from typing import List, Dict

class UnusedResourceDetector:
    def __init__(self):
        self.ec2 = boto3.client('ec2')
        self.rds = boto3.client('rds')
        self.elb = boto3.client('elbv2')
        self.cloudfront = boto3.client('cloudfront')
        self.s3 = boto3.client('s3')
        self.cloudtrail = boto3.client('cloudtrail')
        self.dynamodb = boto3.resource('dynamodb')
        
        # リソース棚卸しテーブル
        self.table = self.dynamodb.Table('unused-resources')
        
        # 判定の閾値（日数）
        self.UNUSED_THRESHOLD_DAYS = 30
    
    def get_last_activity(self, resource_id: str, resource_type: str) -> datetime:
        """CloudTrailからリソースの最終操作時刻を取得"""
        try:
            response = self.cloudtrail.lookup_events(
                LookupAttributes=[
                    {
                        'AttributeKey': 'ResourceName',
                        'AttributeValue': resource_id
                    }
                ],
                MaxResults=1
            )
            
            if response['Events']:
                return response['Events'][0]['EventTime']
            return None
        except Exception as e:
            print(f"Error getting activity for {resource_id}: {e}")
            return None
    
    def detect_unused_ec2(self) -> List[Dict]:
        """30日間CloudWatchメトリクス更新がないEC2を検出"""
        unused = []
        cloudwatch = boto3.client('cloudwatch')
        
        try:
            # 停止状態のEC2をすべてチェック
            response = self.ec2.describe_instances(
                Filters=[
                    {
                        'Name': 'instance-state-name',
                        'Values': ['stopped', 'terminated']
                    }
                ]
            )
            
            for reservation in response['Reservations']:
                for instance in reservation['Instances']:
                    instance_id = instance['InstanceId']
                    state = instance['State']['Name']
                    launch_time = instance['LaunchTime']
                    
                    if state == 'stopped':
                        # CloudWatchメトリクスで最終稼働時刻を確認
                        response = cloudwatch.get_metric_statistics(
                            Namespace='AWS/EC2',
                            MetricName='CPUUtilization',
                            Dimensions=[
                                {
                                    'Name': 'InstanceId',
                                    'Value': instance_id
                                }
                            ],
                            StartTime=datetime.utcnow() - timedelta(days=self.UNUSED_THRESHOLD_DAYS),
                            EndTime=datetime.utcnow(),
                            Period=86400,  # 1日単位
                            Statistics=['Average']
                        )
                        
                        # メトリクスがなければ未使用
                        if not response['Datapoints']:
                            unused.append({
                                'resource_id': instance_id,
                                'resource_type': 'EC2',
                                'state': state,
                                'launch_time': launch_time.isoformat(),
                                'reason': 'No activity in 30 days',
                                'estimated_monthly_cost': instance['InstanceType']
                            })
        
        except Exception as e:
            print(f"Error detecting unused EC2: {e}")
        
        return unused
    
    def detect_unused_rds_snapshots(self) -> List[Dict]:
        """30日以上前のRDSスナップショットを検出"""
        unused = []
        threshold = datetime.utcnow() - timedelta(days=self.UNUSED_THRESHOLD_DAYS)
        
        try:
            response = self.rds.describe_db_snapshots()
            
            for snapshot in response['DBSnapshots']:
                create_time = snapshot['SnapshotCreateTime'].replace(tzinfo=None)
                
                if create_time < threshold:
                    # スナップショットサイズから推定月額を計算
                    size_gb = snapshot['AllocatedStorage']
                    # AWS: EBSスナップショット $0.05/GB/月（東京）
                    monthly_cost = size_gb * 0.05
                    
                    unused.append({
                        'resource_id': snapshot['DBSnapshotIdentifier'],
                        'resource_type': 'RDS-Snapshot',
                        'state': snapshot['Status'],
                        'create_time': create_time.isoformat(),
                        'size_gb': size_gb,
                        'estimated_monthly_cost': monthly_cost
                    })
        
        except Exception as e:
            print(f"Error detecting unused RDS snapshots: {e}")
        
        return unused
    
    def detect_unused_elb(self) -> List[Dict]:
        """接続ゼロのELBを検出"""
        unused = []
        cloudwatch = boto3.client('cloudwatch')
        
        try:
            response = self.elb.describe_load_balancers()
            
            for lb in response['LoadBalancers']:
                lb_name = lb['LoadBalancerName']
                lb_arn = lb['LoadBalancerArn']
                
                # 過去30日間のアクティブな接続をチェック
                cw_response = cloudwatch.get_metric_statistics(
                    Namespace='AWS/ApplicationELB',
                    MetricName='ActiveConnectionCount',
                    Dimensions=[
                        {
                            'Name': 'LoadBalancer',
                            'Value': lb_name
                        }
                    ],
                    StartTime=datetime.utcnow() - timedelta(days=30),
                    EndTime=datetime.utcnow(),
                    Period=86400,
                    Statistics=['Sum']
                )
                
                # 接続数がゼロ継続
                if not cw_response['Datapoints'] or all(dp['Sum'] == 0 for dp in cw_response['Datapoints']):
                    unused.append({
                        'resource_id': lb_arn,
                        'resource_type': 'ALB',
                        'name': lb_name,
                        'reason': 'No active connections in 30 days',
                        'estimated_monthly_cost': 16.20  # 東京リージョンのALB基本料
                    })
        
        except Exception as e:
            print(f"Error detecting unused ELB: {e}")
        
        return unused
    
    def detect_unused_s3_buckets(self) -> List[Dict]:
        """リクエストゼロのS3バケットを検出"""
        unused = []
        cloudwatch = boto3.client('cloudwatch')
        
        try:
            response = self.s3.list_buckets()
            
            for bucket in response['Buckets']:
                bucket_name = bucket['Name']
                
                # 過去30日間のリクエスト数をチェック
                cw_response = cloudwatch.get_metric_statistics(
                    Namespace='AWS/S3',
                    MetricName='NumberOfObjects',
                    Dimensions=[
                        {
                            'Name': 'BucketName',
                            'Value': bucket_name
                        },
                        {
                            'Name': 'StorageType',
                            'Value': 'AllStorageTypes'
                        }
                    ],
                    StartTime=datetime.utcnow() - timedelta(days=30),
                    EndTime=datetime.utcnow(),
                    Period=86400,
                    Statistics=['Average']
                )
                
                if not cw_response['Datapoints']:
                    # 容量を確認
                    size_response = cloudwatch.get_metric_statistics(
                        Namespace='AWS/S3',
                        MetricName='BucketSizeBytes',
                        Dimensions=[
                            {
                                'Name': 'BucketName',
                                'Value': bucket_name
                            },
                            {
                                'Name': 'StorageType',
                                'Value': 'StandardStorage'
                            }
                        ],
                        StartTime=datetime.utcnow() - timedelta(days=1),
                        EndTime=datetime.utcnow(),
                        Period=86400,
                        Statistics=['Average']
                    )
                    
                    size_gb = 0
                    if size_response['Datapoints']:
                        size_gb = size_response['Datapoints'][0]['Average'] / (1024**3)
                    
                    unused.append({
                        'resource_id': bucket_name,
                        'resource_type': 'S3-Bucket',
                        'reason': 'No requests in 30 days',
                        'size_gb': round(size_gb, 2),
                        'estimated_monthly_cost': size_gb * 0.023  # 東京の標準ストレージ料金
                    })
        
        except Exception as e:
            print(f"Error detecting unused S3 buckets: {e}")
        
        return unused
    
    def scan_all(self) -> Dict:
        """全リソースタイプをスキャン"""
        results = {
            'ec2': self.detect_unused_ec2(),
            'rds': self.detect_unused_rds_snapshots(),
            'elb': self.detect_unused_elb(),
            's3': self.detect_unused_s3_buckets(),
            'scan_time': datetime.utcnow().isoformat(),
            'total_count': 0,
            'total_monthly_cost': 0
        }
        
        # 集計
        for resource_type in ['ec2', 'rds', 'elb', 's3']:
            resources = results[resource_type]
            results['total_count'] += len(resources)
            results['total_monthly_cost'] += sum(
                float(r.get('estimated_monthly_cost', 0)) for r in resources
            )
        
        return results
    
    def save_results(self, results: Dict):
        """DynamoDBに結果を保存"""
        try:
            # 既存の推奨リストを取得
            existing = self.table.scan()
            existing_ids = {item['resource_id'] for item in existing.get('Items', [])}
            
            # 新しいリソースを追加
            new_count = 0
            for resource_type in ['ec2', 'rds', 'elb', 's3']:
                for resource in results[resource_type]:
                    if resource['resource_id'] not in existing_ids:
                        resource['discovered_at'] = datetime.utcnow().isoformat()
                        resource['status'] = 'NEW'
                        self.table.put_item(Item=resource)
                        new_count += 1
            
            print(f"Saved {new_count} new unused resources")
        
        except Exception as e:
            print(f"Error saving results: {e}")

def lambda_handler(event, context):
    """Lambdaハンドラー"""
    detector = UnusedResourceDetector()
    results = detector.scan_all()
    detector.save_results(results)
    
    return {
        'statusCode': 200,
        'body': json.dumps(results, default=str)
    }

正直に言うと、最初はCloudTrailだけに頼ってて、とんでもなく遅かった。30分かかるとか。その後CloudWatchメトリクスに切り替えたから劇的に改善した。今は10分以内に全スキャンが完了してる。

Slack通知で可視化した

レポートをDynamoDBに貯めるだけじゃなく、チームに見せる仕組みが必須なんですよ。Slack通知を実装したら、月1回のスキャンで「あ、こんなリソースあったんだ」って気づくようになった。それまでは完全に埋もれてた。

import slack_sdk
from slack_sdk.errors import SlackApiError

class SlackNotifier:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self.client = slack_sdk.WebClient(token=webhook_url.split('/')[-1])
    
    def notify_unused_resources(self, results: Dict):
        """DynamoDB内の未対応リソースをSlackで通知"""
        dynamodb = boto3.resource('dynamodb')
        table = dynamodb.Table('unused-resources')
        
        # ステータスが NEW のものを取得
        response = table.scan(
            FilterExpression='#status = :status',
            ExpressionAttributeNames={'#status': 'status'},
            ExpressionAttributeValues={':status': 'NEW'}
        )
        
        items = response.get('Items', [])
        if not items:
            return
        
        # リソースタイプ別に集計
        by_type = {}
        total_cost = 0
        
        for item in items:
            rtype = item['resource_type']
            cost = float(item.get('estimated_monthly_cost', 0))
            total_cost += cost
            
            if rtype not in by_type:
                by_type[rtype] = []
            by_type[rtype].append(item)
        
        # Slackメッセージを構築
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "⚠️ 不要リソース月次レポート"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*検出リソース数:* {len(items)} 個\n*推定月額コスト:* ¥{total_cost:,.0f}"
                }
            }
        ]
        
        # リソースタイプごとの詳細
        for rtype, resources in by_type.items():
            type_cost = sum(float(r.get('estimated_monthly_cost', 0)) for r in resources)
            
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*{rtype}* ({len(resources)} 個 / ¥{type_cost:,.0f}/月)"
                }
            })
            
            # 最大3件まで表示
            for resource in resources[:3]:
                resource_id = resource.get('resource_id', 'Unknown')
                name = resource.get('name', resource_id)
                reason = resource.get('reason', '不明')
                cost = resource.get('estimated_monthly_cost', 0)
                
                blocks.append({
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"• `{name}` - {reason}\n  推定コスト: ¥{cost:,.0f}/月"
                    }
                })
            
            if len(resources) > 3:
                blocks.append({
                    "type": "context",
                    "elements": [
                        {
                            "type": "mrkdwn",
                            "text": f"他 {len(resources) - 3} 件"
                        }
                    ]
                })
        
        blocks.append({
            "type": "divider"
        })
        
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "<https://your-dashboard-url|詳細はダッシュボードで確認>"
            }
        })
        
        try:
            self.client.chat_postMessage(
                channel='#aws-cost-alerts',
                blocks=blocks
            )
        except SlackApiError as e:
            print(f"Error posting to Slack: {e}")

実装してから「あ、こういう通知があると良いな」って細かい要望がいっぱい出た。チーム運用2ヶ月目でこのあたりの細部が固まった感じですね。

実装して気づいた落とし穴

1. CloudTrailだけじゃ不足

EC2をずっと停止してる場合、CloudTrailに記録が出ないことが多い。最終的にCloudWatchメトリクスのない期間を見るほうが正確だった。私たちも手探りで何度か失敗してます。

2. 誤検出が月50件くらいある

例えば「テスト用ELBを月1回だけ起動」みたいなのが引っかかる。最初、無条件に削除推奨にしてたら、インフラチーム怒られた。結果、「3ヶ月検出し続けたら推奨」って仕組みに変えた。地味に重要な改良です。

3. コスト計算が若干ズレる

特にECS・Lambdaは複雑で、EC2やRDSのほうが推定値の精度が高い。今は「目安」として扱ってる。正確さよりも「ざっくり把握する」が目的だから、これで十分かな。

実装の価値：月200万円削減の現実

この仕組みを導入してから、3ヶ月で月200万円削減できた。内訳はこんな感じです：

リソースタイプ	削減額
RDSスナップショット削除	月45万円
停止EC2削除	月38万円
未使用ELB削除	月32万円
古いS3バケット削除	月28万円
その他小さいの	月57万円
合計	月200万円

xychart-beta
    title "AWS月間コスト推移（3ヶ月）"
    x-axis [実装前, 1ヶ月後, 2ヶ月後, 3ヶ月後]
    y-axis "月額コスト（万円）" 0 --> 600
    line [520, 450, 350, 320]

最初は「手動で毎月削除しろ」って運用だったけど、自動化したら人力ゼロになった。これは本当に大きい。毎月レビューのミーティング時間だけで月3時間くらい浮いてますよ。

2026年時点で大事だなと思うこと

正直まだ完璧じゃない部分がある。Compute Optimizerの推奨との組み合わせとか、Reserved Instanceとの整合性とか。でもこれぐらいの自動化があると、月1回のスキャンで「あ、こんなの残ってた」って気づける。手動なら絶対見落とす。

特にマルチアカウント環境だと効く。うちは8個のアカウント運用してるんだけど、一括でスキャンできるのはマジで助かる。Cross Account Roleを作ってLambdaで全部舐めてる。

まとめ

月200万円の無駄に気づかないままでは終わりたくなかった。Lambda + CloudWatchメトリクスで自動検出する仕組みを作ったら、3ヶ月で劇的に改善した。

実装するときのポイントをまとめるなら：

• CloudTrailだけじゃ不足 → メトリクス・Stateの複合判定が必須
• 誤検出が多い → 複数期間の検出で確度を上げる仕組みに
• Slack通知で可視化 → チームが自然と削除判断を始める
• 3ヶ月で月200万円削減 → 投資対効果で言えば、これが一番効いた施策

これから導入する人は、最初はEC2 + RDSだけ対応で十分。そのあとS3・ELB・CloudFrontって増やしていく方が失敗しない。完璧を目指さずに「動いてる」状態を作ってから、細部を詰めるのが現実的ですよ。