AWS請求書に50万円の謎|使ってないリソースを自動検出した2ヶ月の戦い

まず地獄を見たのは請求書だった

先月のAWS請求書を見た瞬間、僕は息を吐いた。月450万円。前月は400万円だったはずだ。増分の50万円が何なのか、チーム誰も答えられない状態になっていた。

うちのチームはマイクロサービス化が進んで、もう3年前のEC2インスタンスがどこで何をしてるのか誰も把握していない。古いプロジェクトのRDS、どこにも使われていないS3バケット、VPCエンドポイントまで。「いつか使う」「念のため」で残ってるリソースが数百個単位である可能性が高かった。

正直、請求書を眺めてるだけじゃ何もわからない。なので、不要リソースを自動で検出して、本当に使ってるのか確認する仕組みを0から作ることにした。2ヶ月かけて実装してみたら、想像以上に効いた。月50万円以上削減できただけじゃなく、チーム全体がリソース意識を持つようになったんだよね。

Cost Anomaly Detection + Lambda自動通知で異常を早期キャッチ

最初にやったのは、異常なコスト増加を検出して即座に通知する仕組みだ。AWSのCost Anomaly Detectionって地味だけど、使い方次第で本当に強い。

# Lambda関数：異常コストをSlackに通知
import json
import boto3
import requests
from datetime import datetime

ce_client = boto3.client('ce')
sns_client = boto3.client('sns')

def lambda_handler(event, context):
    # SNSイベントをパース
    message = json.loads(event['Records'][0]['Sns']['Message'])
    anomaly = message.get('detail', {})
    
    cost = float(anomaly.get('estimatedCharges', 0))[0]
    start_date = anomaly.get('startDate')
    service = anomaly.get('service', 'Unknown')
    threshold = float(anomaly.get('estimatedThreshold', 0))[0]
    
    # 増加額
    increase = cost - threshold
    percent_increase = (increase / threshold * 100) if threshold > 0 else 0
    
    # Slack通知
    slack_message = {
        'text': f':warning: AWS Cost Anomaly Detected',
        'blocks': [
            {
                'type': 'header',
                'text': {'type': 'plain_text', 'text': f'Cost Spike in {service}'}
            },
            {
                'type': 'section',
                'fields': [
                    {'type': 'mrkdwn', 'text': f'*Actual Cost*\n${cost:,.2f}'},
                    {'type': 'mrkdwn', 'text': f'*Expected Cost*\n${threshold:,.2f}'},
                    {'type': 'mrkdwn', 'text': f'*Increase*\n${increase:,.2f} (+{percent_increase:.1f}%)'},
                    {'type': 'mrkdwn', 'text': f'*Period*\n{start_date}'}
                ]
            },
            {
                'type': 'actions',
                'elements': [
                    {
                        'type': 'button',
                        'text': {'type': 'plain_text', 'text': 'Check Cost Explorer'},
                        'url': 'https://console.aws.amazon.com/cost-management/home'
                    }
                ]
            }
        ]
    }
    
    # Slack通知を送信
    webhook_url = os.environ['SLACK_WEBHOOK_URL']
    requests.post(webhook_url, json=slack_message)
    
    return {'statusCode': 200, 'body': json.dumps('Notification sent')}

これ、やってみて気づいたことが2つある。

1つ目は、単純に異常を通知するだけだと、チームはアラート疲れするってこと。何度も何度も「なんじゃこりゃ」って感じのアラートが来て、結局見なくなる。だから、閾値を調整して「本当にヤバい増加」だけを拾うようにした。地味だけど、この調整が実は一番大事なんだ。

2つ目は、「異常」と分かっても、原因を特定するのが地獄ってこと。Cost Explorerで「EC2が増えた」くらいしか分からない。だから、並行して他の仕組みが必要になった。

Compute Optimizer で使えないEC2・RDS を自動発見

AWSのCompute Optimizerは3ヶ月分の利用データを見て「このインスタンス、オーバースペックじゃない？」と教えてくれるやつだ。ただ、その先は手動で確認する人が多い。僕たちは、これを自動で棚卸し情報に変換することにしたんだよね。

# Lambda：Compute Optimizerの推奨を取得して保存
import boto3
import json
from datetime import datetime

compute_optimizer = boto3.client('compute-optimizer')
s3_client = boto3.client('s3')

def lambda_handler(event, context):
    recommendations = []
    
    # EC2インスタンスの推奨を取得
    ec2_paginator = compute_optimizer.get_paginator('describe_ec2_instance_recommendations')
    for page in ec2_paginator.paginate():
        for rec in page.get('instanceRecommendations', []):
            if rec['finding'] in ['Underutilized', 'Optimized']:
                recommendations.append({
                    'type': 'EC2',
                    'instance_id': rec['instanceId'],
                    'current_type': rec['currentInstanceType'],
                    'recommended_type': rec['recommendationOptions'][0].get('instanceType'),
                    'finding': rec['finding'],
                    'utilization_metrics': rec.get('utilizationMetrics', []),
                    'potential_savings': rec['recommendationOptions'][0].get('savingsOpportunity', {}).get('savingsOpportunityPercentage', 0),
                    'timestamp': datetime.utcnow().isoformat()
                })
    
    # RDSインスタンスの推奨を取得
    rds_paginator = compute_optimizer.get_paginator('describe_db_instance_recommendations')
    for page in rds_paginator.paginate():
        for rec in page.get('dbInstanceRecommendations', []):
            if rec['finding'] in ['Underutilized', 'Optimized']:
                recommendations.append({
                    'type': 'RDS',
                    'db_instance': rec['dbInstanceName'],
                    'current_class': rec['currentDBInstanceClass'],
                    'recommended_class': rec['recommendationOptions'][0].get('dbInstanceClass'),
                    'finding': rec['finding'],
                    'potential_savings': rec['recommendationOptions'][0].get('savingsOpportunity', {}).get('savingsOpportunityPercentage', 0),
                    'timestamp': datetime.utcnow().isoformat()
                })
    
    # S3に保存（タイムシリーズで見られるように）
    bucket_name = 'my-cost-optimization-data'
    key = f"compute-optimizer/{datetime.now().strftime('%Y/%m/%d')}/recommendations.json"
    s3_client.put_object(
        Bucket=bucket_name,
        Key=key,
        Body=json.dumps(recommendations, indent=2),
        ContentType='application/json'
    )
    
    # ダッシュボード向けに集計も作成
    summary = {
        'total_recommendations': len(recommendations),
        'underutilized': len([r for r in recommendations if r['finding'] == 'Underutilized']),
        'optimized': len([r for r in recommendations if r['finding'] == 'Optimized']),
        'total_potential_savings_percent': sum([r['potential_savings'] for r in recommendations]) / len(recommendations) if recommendations else 0,
        'by_type': {
            'EC2': len([r for r in recommendations if r['type'] == 'EC2']),
            'RDS': len([r for r in recommendations if r['type'] == 'RDS'])
        }
    }
    
    print(json.dumps(summary))
    return {'statusCode': 200, 'body': json.dumps(summary)}

実際に走らせてみたら、150個のEC2インスタンスの中で、40個くらいが「本当に必要？」という状態だった。その中には、

• 開発環境で立てっぱなしのc5.2xlarge（実際の使用率5%以下）
• テストプロジェクト用のRDS（3年前のコード）
• ログ保存用のS3（誰も見てない古いログ）

こういった「あるある」が山盛りだった。正直、見つけた時は「え、これ誰が作ったんだ…」って感じでしたね。

Trusted Advisor + ResourceGroupsTaggingAPI で「孤立したリソース」を検出

Compute Optimizerだけじゃ足りない。EC2は使ってるけど、そのセキュリティグループは他の誰も使ってない、みたいなケースもあるからね。それを見つけるために、リソースのタグと関連を追跡するシステムを作ることにしたんだ。

# Lambda：タグを使って孤立したリソースを検出
import boto3
from collections import defaultdict
from datetime import datetime

tagging_client = boto3.client('resourcegroupstaggingapi')
ec2_client = boto3.client('ec2')
rds_client = boto3.client('rds')
s3_client = boto3.client('s3')

def find_orphaned_resources():
    """
    タグが付いているが、実際には使われていないリソースを検出
    """
    orphaned = []
    
    # 開発環境タグがあるが、3ヶ月アクセスされてないリソース
    resources = tagging_client.get_resources(
        TagFilters=[
            {'Key': 'Environment', 'Values': ['dev', 'staging']}
        ],
        ResourceTypeFilters=['ec2:instance', 'rds:db', 's3']
    )
    
    for resource in resources.get('ResourceTagMappingList', []):
        arn = resource['ResourceARN']
        resource_type = arn.split(':')[5].split('/')[0]
        
        # CloudTrailで最後のアクセス日時を確認
        last_access = get_last_cloudtrail_access(arn)
        days_since_access = (datetime.now() - last_access).days if last_access else 999
        
        if days_since_access > 90:  # 3ヶ月以上アクセスなし
            orphaned.append({
                'arn': arn,
                'type': resource_type,
                'last_access': last_access.isoformat() if last_access else 'Never',
                'days_unused': days_since_access,
                'tags': resource.get('Tags', {})
            })
    
    return orphaned

def get_last_cloudtrail_access(resource_arn):
    """
    CloudTrailからリソースの最後のアクセス時刻を取得
    """
    cloudtrail = boto3.client('cloudtrail')
    
    try:
        events = cloudtrail.lookup_events(
            LookupAttributes=[
                {'AttributeKey': 'ResourceName', 'AttributeValue': resource_arn.split('/')[-1]}
            ],
            MaxResults=1
        )
        if events.get('Events'):
            return events['Events'][0]['EventTime']
    except:
        pass
    
    return None

# 定期的に実行（週1回くらい）
orphaned_resources = find_orphaned_resources()
print(f"Found {len(orphaned_resources)} orphaned resources")
for r in orphaned_resources:
    print(f"  {r['type']}: {r['arn']} - Last access: {r['last_access']}")

これが意外と強力だった。タグをちゃんと付けてる環境なら、3ヶ月以上アクセスされてないリソースはほぼ100%削除対象だ。本当に使ってるなら、何らかのアクセスは必ず残るからね。

自動棚卸しダッシュボード＋自動削除フロー

ここまでで検出ができるようになったら、あとは見える化とアクション。

# QuickSightダッシュボード用のデータセット構築
import boto3
import pandas as pd
from datetime import datetime, timedelta

s3 = boto3.client('s3')
athena = boto3.client('athena')

def create_inventory_dataset():
    """
    不要リソース棚卸し用のデータセットを作成
    """
    
    # 1. Compute Optimizerの推奨
    # 2. CloudTrailのアクセスログ
    # 3. リソースのタグ情報
    # 4. コスト情報
    # を統合する形ですね
    
    query = """
    WITH compute_opt AS (
        SELECT 
            'EC2' as resource_type,
            instance_id as resource_id,
            finding,
            potential_savings,
            from_iso8601_timestamp(timestamp) as last_checked
        FROM compute_optimizer_data
        WHERE potential_savings > 10  -- 10%以上の削減機会
    ),
    unused_resources AS (
        SELECT 
            resource_type,
            resource_id,
            'Unused' as finding,
            100 as potential_savings,
            last_access
        FROM resource_inventory
        WHERE days_since_last_access > 90
    ),
    combined AS (
        SELECT * FROM compute_opt
        UNION ALL
        SELECT * FROM unused_resources
    )
    SELECT 
        resource_type,
        resource_id,
        finding,
        potential_savings,
        COUNT(*) as occurrences,
        MAX(last_checked) as last_updated
    FROM combined
    GROUP BY resource_type, resource_id, finding, potential_savings
    ORDER BY potential_savings DESC
    """
    
    response = athena.start_query_execution(
        QueryString=query,
        QueryExecutionContext={'Database': 'cost_optimization'},
        ResultConfiguration={'OutputLocation': 's3://my-athena-results/'}
    )
    
    return response['QueryExecutionId']

ダッシュボードには、以下みたいな情報を表示するようにしてます：

• 削減機会ランキング（上位100個）
• リソースタイプ別の内訳
• 3ヶ月の削減額トレンド
• リスク分類（確実に削除できる vs 要確認）

これによって、チーム全体に「今、何が捨てられるのか」が見えるようになった。地味だけど、これが実は一番重要なんだよね。

AWS構成図：自動棚卸しシステム全体

graph TB
    subgraph Detection["リソース検出層"]
        CA["Cost Anomaly<br/>Detection"]
        CO["Compute<br/>Optimizer"]
        CW["CloudWatch Metrics<br/>& CloudTrail"]
        TA["Trusted Advisor"]
    end
    
    subgraph Processing["処理層"]
        Lambda1["集約Lambda<br/>Compute Optimizer"]
        Lambda2["検出Lambda<br/>孤立リソース"]
        Lambda3["異常検知<br/>Cost Anomaly"]
    end
    
    subgraph Storage["データ層"]
        S3Inventory["S3 Inventory<br/>Data Lake"]
        RDSMetadata["RDS<br/>メタデータDB"]
    end
    
    subgraph Analytics["分析・可視化層"]
        Athena["Athena<br/>クエリ"]
        QuickSight["QuickSight<br/>ダッシュボード"]
    end
    
    subgraph Action["アクション層"]
        ApprovalSNS["SNS Topics<br/>承認ワークフロー"]
        AutoDelete["Lambda<br/>自動削除"]
        Slack["Slack通知"]
    end
    
    CA --> Lambda3
    CO --> Lambda1
    CW --> Lambda2
    TA --> Lambda2
    
    Lambda1 --> S3Inventory
    Lambda2 --> S3Inventory
    Lambda3 --> RDSMetadata
    
    S3Inventory --> Athena
    RDSMetadata --> Athena
    Athena --> QuickSight
    Athena --> ApprovalSNS
    
    ApprovalSNS --> AutoDelete
    ApprovalSNS --> Slack
    AutoDelete --> Slack
    
    style Detection fill:#e1f5ff
    style Processing fill:#f3e5f5
    style Storage fill:#fff3e0
    style Analytics fill:#f1f8e9
    style Action fill:#fee

正直、最初は「自動削除は怖い」って感じだったけど、段階的に実装した結果、今は以下のフローで動いてますよ：

1. Week 1: 候補リソースを検出・DynamoDBに記録
2. Week 2: 「この間に本当に使われてない？」を確認（CloudTrail再確認）
3. Week 3: Slackで承認メッセージを投稿（24時間の猶予期間）
4. Week 4: 承認なければ自動削除実行

このプロセス中に、何度も「あ、これ実は使ってた！」って発見がある。だから、無条件自動削除じゃなく、人間による最終確認＋タイムゾーン配慮が大事なんだ。時差がある場合、夜中に削除が走ると修復できないからね。

実装して気づいた、思ったより地味な落とし穴

正直に言うと、最初は「自動化すれば解決」と思ってた。でも、実装して2ヶ月走らせてみたら、いろいろありました。

1. 依存関係が複雑すぎる

EC2を削除できると思ったら、そのセキュリティグループはALBが使ってた。RDSを削除できると思ったら、Lambdaがシークレットマネージャー経由で接続してた。結局、リソース間の関連を全部マッピングする作業が必要になった。

# リソース依存関係の追跡
def build_dependency_graph():
    dependency = defaultdict(set)
    
    # EC2が参照してるセキュリティグループ
    for instance in ec2.describe_instances()['Reservations']:
        for inst in instance['Instances']:
            for sg in inst['SecurityGroups']:
                dependency[f"sg-{sg['GroupId']}"].add(f"instance-{inst['InstanceId']}")
    
    # RDSが使ってるサブネットグループ
    for db in rds.describe_db_instances()['DBInstances']:
        for sg in db['VpcSecurityGroups']:
            dependency[f"sg-{sg['VpcSecurityGroupId']}"].add(f"rds-{db['DBInstanceIdentifier']}")
    
    # IAM Roleの依存
    for role in iam.list_roles()['Roles']:
        for inline_policy in iam.list_role_policies(RoleName=role['RoleName'])['PolicyNames']:
            # ポリシーがどのリソースを参照してるか抽出
            policy = iam.get_role_policy(RoleName=role['RoleName'], PolicyName=inline_policy)
            # ... リソースアーン解析
    
    return dependency

2. CloudTrailのアクセス検出が遅い

「3ヶ月アクセスなし」の判定に、CloudTrailを見てるんだけど、これが結構な遅延があって地獄なんですよ。特に、キャッシュ系のS3へのアクセスは CloudFrontを経由してるから、CloudTrailには見えない。結果、「実は使ってるのに削除候補に上がってた」が何度か起きた。

今は、より詳細なメトリクスを使う方に転換してます：

# CloudWatchメトリクスで実アクセスを見る
def check_actual_usage(resource_type, resource_id):
    cloudwatch = boto3.client('cloudwatch')
    
    if resource_type == 'S3':
        # S3のアクセス数（CloudWatchメトリクス）
        response = cloudwatch.get_metric_statistics(
            Namespace='AWS/S3',
            MetricName='NumberOfObjects',
            Dimensions=[{'Name': 'BucketName', 'Value': resource_id}],
            StartTime=datetime.now() - timedelta(days=90),
            EndTime=datetime.now(),
            Period=86400,  # 日次
            Statistics=['Average']
        )
        return len([d for d in response['Datapoints'] if d['Average'] > 0]) > 0
    
    elif resource_type == 'EC2':
        # CPUメトリクスで本当に動いてるか確認
        response = cloudwatch.get_metric_statistics(
            Namespace='AWS/EC2',
            MetricName='CPUUtilization',
            Dimensions=[{'Name': 'InstanceId', 'Value': resource_id}],
            StartTime=datetime.now() - timedelta(days=90),
            EndTime=datetime.now(),
            Period=3600,  # 時間単位
            Statistics=['Average']
        )
        # 2%以上の利用があれば「使ってる」とみなす
        return any(d['Average'] > 2 for d in response['Datapoints'])

3. タグなしリソースが本当に多い

開発初期とか、個人で検証してた時代のリソースは、タグが全く付いていない。だから、タグベースの分類は大前提として成り立たないって学びました。苦い思い出ですね。

今は、リソース作成時にLambdaで自動的にタグを付ける仕組みを入れました：

# EventBridge経由でEC2/RDS作成時に自動タグ付け
def auto_tag_resources(event, context):
    detail = event['detail']
    resource_id = detail['responseElements'].get('instanceId') or detail['responseElements'].get('dBInstanceIdentifier')
    
    if not resource_id:
        return
    
    ec2 = boto3.client('ec2')
    
    default_tags = {
        'CreatedBy': detail.get('userIdentity', {}).get('principalId'),
        'CreatedDate': detail['eventTime'],
        'CostCenter': 'engineering',  # デフォルト
        'Environment': 'dev',  # デフォルト
        'AutoTagged': 'true'
    }
    
    try:
        ec2.create_tags(Resources=[resource_id], Tags=[{'Key': k, 'Value': str(v)} for k, v in default_tags.items()])
    except Exception as e:
        print(f"Failed to tag {resource_id}: {e}")

数字で見える削減効果

xychart-beta
    title 月間AWSコスト推移（2ヶ月の棚卸し実施前後）
    x-axis [実施前, Week1, Week2, Week3, Week4, 現在]
    y-axis "月間コスト ($)" 0 --> 500000
    line [450000, 435000, 420000, 385000, 350000, 380000]

見ての通り、削減額は段階的なんだ：

• Week 1-2: 小さなリソース（dev環境の孤立したEC2など）削除で月10万円削減
• Week 2-3: RDS削除（バックアップ含む）で月25万円削減
• Week 3-4: ボリュームディスク最適化で月15万円削減

現在は月50万円削減で落ち着いてます。「もっと削減できるのでは」という声もありますが、正直なところ、リスクとのバランスを考えると、ここが現実的かなと感じてます。運用負荷も増えるし、下手に削除して本番障害になるのが一番怖いからね。

まとめ

不要リソース棚卸しを自動化して月50万円削減できたけど、本当に重要だったのは以下5つ：

1. Cost Anomaly Detection + Lambda通知: 異常を即座に検知する基盤
2. Compute Optimizer の定期実行: 利用パターンベースの提案がめちゃ精度高い
3. CloudTrail + CloudWatch メトリクス: 本当のアクセス有無を複層的に確認する
4. 自動化の前に人間による承認ステップ: 過信は危険、段階的アプローチが大事
5. リソースタグの統一: 検出精度の9割はタグで決まる

次のステップとしては、IAM権限の自動棚卸しと古いVPCの統廃合を考えてます。IAMロールなんて本当に誰が使ってるか分からないので、これができれば相当強いはず。地味な作業だけど、地味だからこそ金が眠ってるんだろうなって思う。

皆さんのチームは、今どくらいの不要リソース抱えてますか？コメントで教えてもらえたら嬉しいです。