CloudTrailでS3オブジェクトにタグ付けをした実行元を特定する方法
S3オブジェクトのタグ付け料金高くないか？
こんにちは！AWS事業本部のおつまみです。
先日あるアカウントでコスト削減できる箇所を調査していたところ、S3オブジェクトのタグ付け料金が異様に高いことに気づきました。
TagStorage-TagHrsは時間毎にレポートされる、バケット内のすべてのオブジェクトのタグの合計数です。
参考：Amazon S3 の AWS 請求および使用状況レポートを理解する - Amazon Simple Storage Service
10,000個のタグあたり$0.01かかります。
料金明細から1ヶ月で約1.6億個のタグ付けがされており、毎月$164のコストがかかっていることがわかりました。
アカウント運用者から、「タグがついているオブジェクトを削除しても一時的にコストは下がるがまた上がっている。何が原因でタグ付いているのか調査する方法を教えてほしい」と要望がありました。
そこで、今回はCloudTrailでS3オブジェクトにタグ付けをした実行元を特定する方法をご紹介します。
この記事で解決できること
- S3オブジェクトのタグ付けを実行したIAMロールやリソースの特定
- タグ付けの実行時刻の確認
3行まとめ
- CloudTrailのデータイベントを有効化することで、S3オブジェクトのタグ付け実行元を特定できる
- 既存の証跡では大量のログをスキャンすることになるため、新しい証跡を作成することを推奨
- Athenaで
PutObjectTaggingイベントを検索することで、タグ付けの実行元や時刻、内容を確認できる
構成図
今回やりたいことを図に示しました。
何らかがS3オブジェクトに対して、タグ付けを実行しており、そのタグ付け元を特定することがゴールです。
前提作業
コストが増加していたアカウントでは作業ができなかったため、弊社検証環境内にて検証を実施しました。検証環境では以下の準備を行っています。（図の灰色箇所）
- 検証用S3バケットの作成
- オブジェクトへタグ付けするリソース
- 今回はLambdaでオブジェクト追加とタグ付けを行うよう設定しました。
リソースは事前にAWS SAMでデプロイしました。
使用したテンプレートはこちらです。
template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: A simple AWS Lambda function that adds an object to S3 and tags it.
Resources:
MyS3Bucket:
Type: AWS::S3::Bucket
Properties:
BucketName: my-sample-bucket-123456 # バケット名はユニークに設定
MyLambdaFunction:
Type: AWS::Serverless::Function
Properties:
Handler: lambda_function.lambda_handler
Runtime: python3.13
CodeUri: ./ # コードがあるディレクトリ
Environment:
BUCKET_NAME: !Ref MyS3Bucket
Policies:
- S3FullAccess # S3へのフルアクセス権限を付与
- AWSLambdaBasicExecutionRole # CloudWatch Logsへのアクセス権限を付与
Timeout: 15 # タイムアウトを15秒に設定
Outputs:
LambdaFunction:
Description: "Lambda Function ARN"
Value: !GetAtt MyLambdaFunction.Arn
S3Bucket:
Description: "S3 Bucket Name"
Value: !Ref MyS3Bucket
lambda_function.py
import json
import boto3
import os
import datetime
def lambda_handler(event, context):
# S3クライアントの作成
s3_client = boto3.client('s3')
# 環境変数からバケット名を取得
bucket_name = os.environ['BUCKET_NAME']
# 現在のタイムスタンプを使用してオブジェクトキーを生成
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
object_key = f"sample_object_{timestamp}.txt" # 任意のファイル名形式
# オブジェクトの内容
object_content = "This is a sample object content."
# S3にオブジェクトを追加
try:
s3_client.put_object(Bucket=bucket_name, Key=object_key, Body=object_content)
print(f"Object {object_key} added to bucket {bucket_name}.")
except Exception as e:
print(f"Error adding object to S3: {e}")
return {
'statusCode': 500,
'body': json.dumps('Error adding object to S3')
}
# タグ付けのためのタグリスト
tags = {
'TagSet': [
{
'Key': 'Environment',
'Value': 'Production'
},
{
'Key': 'Project',
'Value': 'SampleProject'
}
]
}
# オブジェクトにタグを付ける
try:
s3_client.put_object_tagging(Bucket=bucket_name, Key=object_key, Tagging=tags)
print(f"Tags added to object {object_key} in bucket {bucket_name}.")
except Exception as e:
print(f"Error tagging object in S3: {e}")
return {
'statusCode': 500,
'body': json.dumps('Error tagging object in S3')
}
return {
'statusCode': 200,
'body': json.dumps(f'Object {object_key} added and tagged successfully')
}
この状態から、調査する方法を紹介します。
やってみた
S3 オブジェクトレベルのログを閲覧するために必要な作業
S3 オブジェクトレベルのログを閲覧するには下記2つの作業が必要です。
- CloudTrail で S3 バケットのデータイベント記録用の証跡を作成
- 記録した S3 オブジェクトレベルのログを Athenaで実行するための設定・クエリの実行
デフォルトでは、 CloudTrail は S3 バケットレベルのログを記録しますが、 S3 オブジェクトレベルのログは記録しない為、 別途設定を有効化する必要があります 。
また、S3 オブジェクトレベルのログはCloudTrailコンソールのイベント履歴から確認できず、S3に配信されたログを確認する必要があります。そのため、今回はAthenaを用いて該当ログを確認します。
1.CloudTrail で S3 バケットのデータイベント記録用の証跡作成
既存の証跡を利用すると、Athenaでクエリ実行時に大量のログをスキャンすることになり、コストが増加しやすい傾向になるため、新しい証跡を作成します。
- CloudTrailコンソールで[証跡の作成]を選択します。
- 証跡名の入力や新しいストレージを作成し、「次へ」を選択します。
- 「データイベント」のみを選択し、データイベントは下記を選択・入力し、保存します。
- データイベントタイプ：
S3 - ログセレクターテンプレート:
カスタム - 高度なイベントセレクター
- フィールド：
eventName - オペレーター：
次と等しい: - Value:
PutObjectTagging
- フィールド：
- データイベントタイプ：
- 正しく設定されていることを確認し、[証跡の作成]を選択します。
(調査時不要)S3バケットへのオブジェクトの追加・タグ付け
今回は検証環境を用意したので、Lambdaを実行させて、S3バケットへのオブジェクトの追加・タグ付けを実施します。
調査される方はこの手順をスキップしてください。
- 事前に作成したLambda関数を 「テスト」で実行し、成功することを確認します。
- オブジェクトが追加され、そのオブジェクトにタグがついていることを確認します。
aws s3api get-object-tagging --bucket <バケット名> --key <オブジェクト名>
{
"TagSet": [
{
"Key": "タグ名1",
"Value": "タグ値1"
},
{
"Key": "タグ名2",
"Value": "タグ値2"
}
]
}
これで、CloudTrailログにデータイベントとして、PutObjectTaggingのイベントが追加されます。
2. 記録した S3 オブジェクトレベルのログを Athenaで確認
前述した通り、CloudTrail のイベント履歴の画面では、オブジェクトレベルのログが表示できません。そのため、Athenaで確認します。
Athena用のテーブルを作成
- CloudTrail の [イベント履歴] の画面で、 [Athenaテーブル作成] を選択します。
- [ストレージの場所] でバケット名を選択し、[テーブルを作成] を選択します。
-
Athenaコンソールの[クエリエディタ]に遷移します。
-
画面右上の [設定] で[クエリの結果の場所]が設定されているか確認します。入力されてない場合は Athena のクエリ実行結果用の S3 バケットを別途作成して指定します。適切な保存場所でない場合、クエリ実行時に以下のエラーが出ます。
Athenaでクエリ実行
- 新規クエリを作成し、以下を入力します。
eventtimeは調査したい日時に適宜変更してください。
SELECT
eventtime,
eventname,
useridentity.type as user_type,
useridentity.arn as role_arn,
useridentity.sessionContext.sessionIssuer.userName as role_name,
useridentity.principalId as principal_id,
requestparameters
FROM <作成したテーブル名>
WHERE
eventtime >= '2025-02-17T05:57:49Z'
and eventName = 'PutObjectTagging'
AND eventtype = 'AwsApiCall';
- [クエリの実行] を選択します。結果、
PutObjectTaggingのログが確認できました。
クエリ結果で出力されるフィールドは以下の通りです。
- eventtime: タグ付けが実行された時刻
- user_type: 実行者の種類（多くの場合「AssumedRole」）
- role_arn: 使用されたロールのARN
- role_name: ロール名
- principal_id: セッションID
- requestparameters: タグ付けの詳細情報
role_arnやprincipal_idからS3オブジェクトのタグ付けをした実行元（今回でいうLambda関数）を特定することができました！
調査コストについて
この方法で実行元を特定する際のコストは、主にCloudTrailのデータイベントとAthenaのクエリ実行に関わるものです。
CloudTrailのコスト
- データイベント：$0.10/100,000イベント
- 今回はPutObjectTaggingイベントのみに絞っているため、最小限に抑えられます。
Athenaのクエリコスト
- $5.00/スキャンしたデータ1TB
- 新規証跡を作成し、以下の条件で絞り込むことで、スキャンデータ量を最小化できます
- 期間を1時間に限定
- イベント名をPutObjectTaggingに限定
- 特定のバケットに限定
例えば、1日の調査で1,000イベントのデータイベントログ（$0.001）と、100MBのAthenaクエリスキャン（$0.0005）を実行した場合、合計で約$0.0015/日のコストとなります。
これは、今回問題となっていたS3タグ付けの月額コストと比較すると、非常に小さい金額であり、コスト最適化のための調査として十分に実施する価値があると言えます。
なお調査後は今回作成したデータイベントの証跡は削除することを推奨します。
さいごに
今回は、CloudTrailでS3オブジェクトにタグ付けをした実行元を特定する方法をご紹介しました。
タグ付けのコストは意外と見落としがちですが、大量のオブジェクトに対して不要なタグ付けが行われると、予期せぬコスト増加につながります。
CloudTrailログを確認し、意図しないタグ付けが行われていないか調査することを推奨します。
最後までお読みいただきありがとうございました！
以上、おつまみ（＠AWS11077）でした！
