- 公開日
- 最終更新日
【Lambda】S3レプリケーション失敗通知
この記事を共有する
目次
はじめに
こんにちは。 今回は、Amazon S3のレプリケーション失敗を検知し、Microsoft Teamsへ即座にアラート通知をする仕組みを構築する方法をご紹介します。
背景・課題
現在のS3ライフサイクルでは、レプリケーションステータス が FAILED の間、ライフサイクルがオブジェクトの有効期限切れや移行をブロックするような仕様となっています。レプリケーションが成功し、ステータスが COMPLETED に変わると、ライフサイクルはそれらのオブジェクトの処理を再開します。
もしS3レプリケーションが失敗したオブジェクトに気付かない場合、そのオブジェクトは FAILED のまま残り続け、想定外のストレージ料金が発生し続ける可能性があります。
S3レプリケーションが失敗した際に検知できる仕組みを構築してみました！
前提条件・構成
今回の検証環境と前提条件は以下の通りです。
通知フロー
- Amazon S3: イベント通知（Replication Failed）を発火
- AWS Lambda: イベントのフィルタリングとTeamsへの送信
- Microsoft Teams: Workflows (Webhook) で通知を受信
前提
- 通知対象のS3バケットと、S3レプリケーション設定が既に存在すること
- S3レプリケーションのモニタリング設定(CloudWatch)が有効であること
- Microsoft Teamsのチャネルに対する設定権限があること
手順1：Teams Workflowsによる通知先設定
まずは、Teams側で通知を受け取るためのWebhook URLを発行します。
- Teamsの左メニュー「アプリ」から「Workflows」を開きます。
- 「＋ 作成（新しいフローを作成）」をクリックします。
- テンプレート検索で「Webhook アラートをチャネルに送信する」を選択します。
- 以下の基本情報を設定します。
(例)
フロー名： S3ReplicationFailed通知
実行するチーム・チャネル： [チーム名] > S3のレプリケーション通知連絡用 - 設定完了後、発行された Webhook URL を控えておきます。
手順2：Lambda関数の作成
次に、通知ロジックを担うLambda関数を作成します。
1. 関数の作成
AWSマネジメントコンソール（東京リージョン）のLambda画面から「関数の作成」を行います。
- 関数名: Lambda-S3ReplicationFailed-Teams
- ランタイム: Python 3.14
- 実行ロール: 基本的な Lambda アクセス権限で新しいロールを作成
2. コードのデプロイ
作成後、「コード」タブにて以下のPythonコードを貼り付け、「Deploy」をクリックします。
import json
import os
import urllib.request
from datetime import datetime, timezone, timedelta
TEAMS_WEBHOOK_URL = os.environ.get("TEAMS_WORKFLOW_WEBHOOK_URL")
PROJECT_NAME = os.environ.get("PROJECT_NAME", "未設定")
# 通知対象外のイベント
IGNORE_EVENT_PREFIXES = (
"ObjectRemoved:",
)
def lambda_handler(event, context):
print("Received event:")
print(json.dumps(event, indent=2))
for record in event.get("Records", []):
event_name = record.get("eventName", "")
s3 = record.get("s3", {})
bucket = s3.get("bucket", {}).get("name", "UnknownBucket")
obj = s3.get("object", {})
object_key = obj.get("key", "UnknownObject")
# 削除系イベントは無視
if event_name.startswith(IGNORE_EVENT_PREFIXES):
print(f"Skip delete event: {event_name}")
continue
# DeleteMarker の場合も無視
if obj.get("deleteMarker") is True:
print("Skip delete marker event")
continue
event_time = record.get("eventTime")
# JST 変換
if event_time:
dt = datetime.fromisoformat(event_time.replace("Z", "+00:00"))
jst_time = dt.astimezone(timezone(timedelta(hours=9)))
timestamp = jst_time.strftime("%Y-%m-%d %H:%M:%S")
else:
timestamp = "N/A"
# Adaptive Card payload
payload = {
"type": "message",
"attachments": [
{
"contentType": "application/vnd.microsoft.card.adaptive",
"content": {
"$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
"type": "AdaptiveCard",
"version": "1.5",
"body": [
{
"type": "TextBlock",
"size": "Large",
"weight": "Bolder",
"color": "Attention",
"text": "S3 Replication Failed"
},
{
"type": "TextBlock",
"text": "S3 レプリケーションに失敗しました",
"wrap": True
},
{
"type": "FactSet",
"facts": [
{"title": "案件名", "value": PROJECT_NAME},
{"title": "Bucket", "value": bucket},
{"title": "Object", "value": object_key},
{"title": "Event", "value": event_name},
{"title": "Time (JST)", "value": timestamp}
]
}
]
}
}
]
}
try:
req = urllib.request.Request(
TEAMS_WEBHOOK_URL,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json"}
)
with urllib.request.urlopen(req) as res:
print("Teams response:", res.read().decode())
except Exception as e:
print("Error sending to Teams:", str(e))
return {
"statusCode": 200,
"body": "Teams notification processed"
}
3. Lambda 環境変数の設定
「設定」タブ > 「環境変数」より、以下の変数を追加します。
- PROJECT_NAME： 通知カードに表示される案件名
- TEAMS_WORKFLOW_WEBHOOK_URL： https://XXXX…（手順1で発行したWEBHOOK URL）
手順3：S3イベント通知の設定
S3バケット側で、レプリケーション失敗時にLambdaを起動する設定を行います。
- 対象のS3バケット（test-bucket）を開き、「プロパティ」タブへ移動します。
- 「イベント通知」の「イベント通知を作成」をクリックします。
- 以下のように設定します。
- イベント通知名: S3ReplicationFailed
- イベントタイプ: 「オブジェクトのレプリケートに失敗しました (s3:Replication:OperationFailedReplication)」にチェック
- 送信先: Lambda 関数
- Lambda 関数: 作成した Lambda-S3ReplicationFailed-Teams を選択
設定後、Lambda側のトリガー設定にもS3が追加されていることを確認します。
手順4：レプリケーションメトリクスと通知を有効化
Amazon S3レプリケーションルールの編集から以下を設定します。
- 変更前：レプリケーションメトリクスと通知を無効化
- 変更後：レプリケーションメトリクスと通知を有効化
手順5：検証・結果
実際にS3レプリケーションを失敗させて通知を確認します。
今回は意図的に権限のないIAMロールを割り当てることで失敗を誘発します。
■ 準備
対象のS3バケット（test-bucket）のレプリケーションルールを確認し、現在設定されているIAMロール名を控えます。
1. IAMロールの変更
- レプリケーションルールの設定を編集し、IAMロールをS3への権限がないロール（例: AWSServiceRoleForECS など）に一時的に変更します。
2. テストファイルのアップロード
- S3バケットへテスト用ファイル（test.txt）をアップロードします。
3. バッチオペレーションによるレプリケーション実行
- 「新しい IAM ロールを作成」を選択してレプリケーションジョブを作成し、実行します。
- 本来の権限がないため、このジョブは失敗します。
4. 通知の確認
- Teamsを確認し、「S3 Replication Failed」のカード通知が届いていることを確認します。
テスト後の後片付け
検証が終わったら、忘れずに設定を元に戻します。
- レプリケーションルールのIAMロールを、控えておいた正しいIAMロールに戻します。
- S3上のテスト用ファイル（test.txt）を削除します。
- 作成したLambda関数を削除
まとめ
今回は、S3レプリケーション失敗を検知してTeamsへ通知する仕組みを構築しました。
S3レプリケーション失敗検知できる仕組みですので、ぜひ活用してみてください。
この記事は私が書きました
橋本 歩
記事一覧AWSと犬が好きです。
