# Lambda を使用した Amazon DocumentDB イベントの処理
Amazon DocumentDB クラスターをイベントソースとして設定することにより、Lambda 関数を使用して [Amazon DocumentDB (MongoDB 互換) 変更ストリーム](https://docs.aws.amazon.com/documentdb/latest/developerguide/change_streams.html)でイベントを処理できます。その後、Amazon DocumentDB クラスターでデータが変更されるたびに Lambda 関数を呼び出すことで、イベント駆動型のワークロードを自動化できます。
**注記**
Lambda では Amazon DocumentDB のバージョン 4.0 および 5.0 のみがサポートされています。バージョン 3.6 はサポートされていません。
また、イベントソースマッピングでは、Lambda はインスタンスベースのクラスターとリージョンレベルのクラスターのみをサポートします。Lambda は、[Elastic クラスター](https://docs.aws.amazon.com/documentdb/latest/developerguide/docdb-using-elastic-clusters.html)または[グローバルクラスター](https://docs.aws.amazon.com/documentdb/latest/developerguide/global-clusters.html)をサポートしていません。この制限は、Lambda を Amazon DocumentDB に接続するクライアントとして使用する場合には適用されません。Lambda はすべてのクラスタータイプに接続して CRUD 操作を実行できます。
Lambda は、Amazon DocumentDB 変更ストリームからのイベントを、到着した順序に沿って処理します。このため、関数は Amazon DocumentDB からの同時呼び出しを一度に 1 つしか処理できません。関数を監視するには、その[同時実行メトリクス](https://docs.aws.amazon.com/lambda/latest/dg/monitoring-concurrency.html)を追跡できます。
**警告**
Lambda イベントソースマッピングは各イベントを少なくとも 1 回処理し、レコードの重複処理が発生する可能性があります。重複するイベントに関連する潜在的な問題を避けるため、関数コードを冪等にすることを強くお勧めします。詳細については、AWS ナレッジセンターの「[Lambda 関数を冪等にするにはどうすればよいですか?](https://repost.aws/knowledge-center/lambda-function-idempotent)」を参照してください。
**Topics**
+ [
## Amazon DocumentDB イベントの例
](#docdb-sample-event)
+ [
## 前提条件とアクセス許可
](#docdb-prereqs)
+ [
## ネットワークセキュリティを設定する
](#docdb-network)
+ [
## Amazon DocumentDB イベントソースマッピングを作成する (コンソール)
](#docdb-configuration)
+ [
## Amazon DocumentDB イベントソースマッピングを作成する (SDK または CLI)
](#docdb-api)
+ [
## ポーリングとストリームの開始位置
](#docdb-stream-polling)
+ [
## Amazon DocumentDB イベントソースのモニタリング
](#docdb-monitoring)
+ [
# チュートリアル: Amazon DocumentDB を用いて AWS Lambda のストリームの使用
](with-documentdb-tutorial.md)
## Amazon DocumentDB イベントの例
```
{
"eventSourceArn": "arn:aws:rds:us-east-1:123456789012:cluster:canaryclusterb2a659a2-qo5tcmqkcl03",
"events": [
{
"event": {
"_id": {
"_data": "0163eeb6e7000000090100000009000041e1"
},
"clusterTime": {
"$timestamp": {
"t": 1676588775,
"i": 9
}
},
"documentKey": {
"_id": {
"$oid": "63eeb6e7d418cd98afb1c1d7"
}
},
"fullDocument": {
"_id": {
"$oid": "63eeb6e7d418cd98afb1c1d7"
},
"anyField": "sampleValue"
},
"ns": {
"db": "test_database",
"coll": "test_collection"
},
"operationType": "insert"
}
}
],
"eventSource": "aws:docdb"
}
```
この例のイベントとその形状の詳細については、MongoDB ドキュメントウェブサイトの「[変更イベント](https://www.mongodb.com/docs/manual/reference/change-events/)」を参照してください。
## 前提条件とアクセス許可
Amazon DocumentDB を Lambda 関数のイベントソースとして使用する前に、次の前提条件に注意してください。必要なもの:
+ **関数と同じ AWS アカウント と AWS リージョン に既存の Amazon DocumentDB クラスターが必要です。**既存のクラスターがない場合は、「*Amazon DocumentDB デベロッパーガイド*」の「[Amazon DocumentDB の開始方法](https://docs.aws.amazon.com/documentdb/latest/developerguide/get-started-guide.html)」のステップに従って作成できます。または、「[チュートリアル: Amazon DocumentDB を用いて AWS Lambda のストリームの使用](with-documentdb-tutorial.md)」の最初の一連のステップに従って、必要な前提条件をすべて備えた Amazon DocumentDB クラスターを作成する方法もあります。
+ **Lambda が、Amazon DocumentDB クラスターに関連付けられている Amazon Virtual Private Cloud (Amazon VPC) リソースにアクセスできるようにします。**詳細については、「[ネットワークセキュリティを設定する](#docdb-network)」を参照してください。
+ **Amazon DocumentDB クラスターで TLS を有効にします。**これはデフォルトの設定です。TLS を無効にすると、Lambda はクラスターと通信できません。
+ **Amazon DocumentDB クラスターで変更ストリームをアクティブ化します。**詳細については、「*Amazon DocumentDB デベロッパーガイド*」の「[Amazon DocumentDB で変更ストリームを使用する](https://docs.aws.amazon.com/documentdb/latest/developerguide/change_streams.html)」を参照してください。
+ **Amazon DocumentDB クラスターにアクセスするための認証情報を Lambda に提供します。**イベントソースを設定するときは、クラスターへのアクセスに必要な認証の詳細 (ユーザー名とパスワード) を含む [AWS Secrets Manager](https://docs.aws.amazon.com/secretsmanager/latest/userguide/intro.html) キーを指定します。セットアップ中にこのキーを指定するには、次のいずれかを行います。
+ セットアップに Lambda コンソールを使用している場合は、**[Secrets Manager キー]** フィールドでこのキーを指定します。
+ セットアップに AWS Command Line Interface (AWS CLI) を使用している場合は、`source-access-configurations` オプションでこのキーを指定します。このオプションは、[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) コマンドまたは [https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html) コマンドのどちらにも含めることができます。例えば、次のようになります。
```
aws lambda create-event-source-mapping \
...
--source-access-configurations '[{"Type":"BASIC_AUTH","URI":"arn:aws:secretsmanager:us-west-2:123456789012:secret:DocDBSecret-AbC4E6"}]' \
...
```
+ **Amazon DocumentDB ストリームに関連するリソースを管理するには、Lambda に許可を付与します。**関数の[実行ロール](lambda-intro-execution-role.md)に次の許可を手動で追加します。
+ [rds:DescribeDBClusters](https://docs.aws.amazon.com/AmazonRDS/latest/APIReference/API_DescribeDBClusters.html)
+ [rds:DescribeDBClusterParameters](https://docs.aws.amazon.com/AmazonRDS/latest/APIReference/API_DescribeDBClusterParameters.html)
+ [rds:DescribeDBSubnetGroups](https://docs.aws.amazon.com/AmazonRDS/latest/APIReference/API_DescribeDBSubnetGroups.html)
+ [ec2:CreateNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_CreateNetworkInterface.html)
+ [ec2:DescribeNetworkInterfaces](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeNetworkInterfaces.html)
+ [ec2:DescribeVpcs](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeVpcs.html)
+ [ec2:DeleteNetworkInterface](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DeleteNetworkInterface.html)
+ [ec2:DescribeSubnets](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSubnets.html)
+ [ec2:DescribeSecurityGroups](https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeSecurityGroups.html)
+ [kms:Decrypt](https://docs.aws.amazon.com/kms/latest/APIReference/API_Decrypt.html)
+ [secretsmanager:GetSecretValue](https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html)
+ **Lambda に送信する Amazon DocumentDB 変更ストリームイベントのサイズは 6 MB 未満にしてください。**Lambda は最大 6 MB のペイロードサイズをサポートします。変更ストリームが 6 MB を超えるイベントを Lambda に送信しようとすると、Lambda はメッセージを削除して `OversizedRecordCount` メトリクスを発行します。Lambda は、ベストエフォートベースですべてのメトリクスを発行します。
**注記**
Lambda 関数の最大タイムアウト制限は通常 15 分ですが、Amazon MSK、自己管理型 Apache Kafka、Amazon DocumentDB、および ActiveMQ と RabbitMQ 向け Amazon MQ のイベントソースマッピングでは、最大タイムアウト制限が 14 分の関数のみがサポートされます。この制約により、イベントソースマッピングは関数エラーと再試行を適切に処理できます。
## ネットワークセキュリティを設定する
イベントソースマッピングを通じて Lambda に Amazon DocumentDB へのフルアクセスを許可するには、クラスターがパブリックエンドポイント (パブリック IP アドレス) を使用するか、クラスターを作成した Amazon VPC へのアクセスを提供する必要があります。
Lambda で Amazon DocumentDB を使用する場合は、関数に Amazon VPC 内のリソースへのアクセスを付与する [AWS PrivateLink VPC エンドポイント](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html)を作成します。
**注記**
イベントポーラーにデフォルト (オンデマンド) モードを使用するイベントソースマッピングを持つ関数には、AWS PrivateLink VPC エンドポイントが必要です。イベントソースマッピングが[プロビジョンドモード](invocation-eventsourcemapping.md#invocation-eventsourcemapping-provisioned-mode)を使用している場合は、AWS PrivateLink VPC エンドポイントを設定する必要はありません。
エンドポイントを作成して、次のリソースへのアクセスを提供します。
+ Lambda — Lambda サービスプリンシパルのエンドポイントを作成します。
+ AWS STS — サービスプリンシパルがユーザーに代わってロールを引き受けるために、AWS STS のエンドポイントを作成します。
+ Secrets Manager — クラスターが Secrets Manager を使用して認証情報を保存する場合は、Secrets Manager のエンドポイントを作成します。
または、Amazon VPC の各パブリックサブネットに NAT ゲートウェイを設定します。詳細については、「[VPC に接続された Lambda 関数にインターネットアクセスを有効にする](configuration-vpc-internet.md)」を参照してください。
Amazon DocumentDB のイベントソースマッピングを作成するとき、Lambda は Amazon VPC 用に設定されたサブネットおよびセキュリティグループに Elastic Network Interface (ENI) が既に存在するかどうかを確認します。Lambda が既存の ENI を検出した場合、再利用しようとします。それ以外の場合、Lambda は新しい ENI を作成し、イベントソースに接続して関数を呼び出します。
**注記**
Lambda 関数は、Lambda サービスが所有する VPC 内で常に実行されます。関数の VPC 設定はイベントソースマッピングに影響しません。Lambda がイベントソースに接続する方法を判定するのは、イベントソースのネットワーク設定のみです。
クラスターを含む Amazon VPC のセキュリティグループを設定します。デフォルトでは、Amazon DocumentDB はポート: `27017` を使用します。
+ インバウンドルール – イベントソースに関連付けられたセキュリティグループに対してデフォルトのブローカーポート上のすべてのトラフィックを許可します。または、自己参照セキュリティグループルールを使用して、同じセキュリティグループ内のインスタンスからのアクセスを許可することもできます。
+ アウトバウンドルール – 関数が AWS サービスと通信する必要がある場合、外部送信先のポート `443` 上のすべてのトラフィックを許可します。または、自己参照セキュリティグループルールを使用して、他の AWS サービスと通信する必要がない場合は、ブローカーへのアクセスを制限することもできます。
+ Amazon VPC エンドポイントのインバウンドルール – Amazon VPC エンドポイントを使用している場合、Amazon VPC エンドポイントに関連付けられたセキュリティグループは、クラスターセキュリティグループからポート `443` でインバウンドトラフィックを許可する必要があります。
クラスターが認証を使用する場合、Secrets Manager エンドポイントのエンドポイントポリシーを制限することもできます。Secrets Manager API を呼び出す場合、Lambda は Lambda サービスプリンシパルではなく、関数ロールを使用します。
**Example VPC エンドポイントポリシー – Secrets Manager エンドポイント**
```
{
"Statement": [
{
"Action": "secretsmanager:GetSecretValue",
"Effect": "Allow",
"Principal": {
"AWS": [
"arn:aws::iam::123456789012:role/my-role"
]
},
"Resource": "arn:aws::secretsmanager:us-west-2:123456789012:secret:my-secret"
}
]
}
```
Amazon VPC エンドポイントを使用する場合、AWS は API コールをルーティングし、エンドポイントの Elastic Network Interface (ENI) を使用して関数を呼び出します。Lambda サービスプリンシパルは、これらの ENI を使用するすべてのロールおよび関数に対して `lambda:InvokeFunction` を呼び出す必要があります。
デフォルトでは、Amazon VPC エンドポイントには、リソースへの広範なアクセスを許可するオープンな IAM ポリシーが適用されています。そのエンドポイントを使用して必要なアクションを実行するためのベストプラクティスは、これらのポリシーを制限することです。イベントソースマッピングが Lambda 関数を呼び出せるようにするには、VPC エンドポイントポリシーで、Lambda サービスプリンシパルが `sts:AssumeRole` および `lambda:InvokeFunction` を呼び出すことを許可する必要があります。組織内で発生する API コールのみを許可するように VPC エンドポイントポリシーを制限すると、イベントソースマッピングが正しく機能しなくなるため、これらのポリシーには `"Resource": "*"` が必要です。
次の VPC エンドポイントポリシーの例では、AWS STS および Lambda エンドポイントに Lambda サービスプリンシパルの必要なアクセスを付与する方法について示しています。
**Example VPC エンドポイントポリシー - AWS STS エンドポイント**
```
{
"Statement": [
{
"Action": "sts:AssumeRole",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
**Example VPC エンドポイントポリシー – Lambda エンドポイント**
```
{
"Statement": [
{
"Action": "lambda:InvokeFunction",
"Effect": "Allow",
"Principal": {
"Service": [
"lambda.amazonaws.com"
]
},
"Resource": "*"
}
]
}
```
## Amazon DocumentDB イベントソースマッピングを作成する (コンソール)
Amazon DocumentDB クラスターの変更ストリームから読み取る Lambda 関数に対して、[イベントソースマッピング](invocation-eventsourcemapping.md)を作成します。このセクションでは、Lambda コンソールからこれを実行する方法を説明します。AWSSDK と AWS CLI 手順については、「[Amazon DocumentDB イベントソースマッピングを作成する (SDK または CLI)](#docdb-api)」を参照してください。
**Amazon DocumentDB イベントソースマッピングを作成するには (コンソール)**
1. Lambda コンソールの[関数ページ](https://console.aws.amazon.com/lambda/home#/functions)を開きます。
1. 関数の名前を選択します。
1. **[関数の概要]** で **[トリガーを追加]** をクリックします。
1. **[トリガーの設定]** のドロップダウンリストから **[DocumentDB]** を選択します。
1. 必要なオプションを設定し、**[追加]** を選択します。
Lambda は、Amazon DocumentDB イベントソースの以下のオプションをサポートしています。
+ **[DocumentDB クラスター]** – Amazon DocumentDB クラスターを選択します。
+ **[トリガーをアクティブ化]** – トリガーを今すぐアクティブ化するかどうかを選択します。このチェックボックスをオンにすると、関数は、イベントソースマッピングの作成時に、指定された Amazon DocumentDB 変更ストリームからのトラフィックの受信を直ちに開始します。チェックボックスをオフにして、テストのために非アクティブ化された状態でイベントソースマッピングを作成することを推奨します。作成後、いつでもイベントソースマッピングをアクティブ化できます。
+ **[データベース名]** – 使用するクラスター内のデータベースの名前を入力します。
+ (オプション) **[コレクション名]** – 使用するデータベース内のコレクションの名前を入力します。コレクションを指定しない場合、Lambda はデータベース内の各コレクションのすべてのイベントをリッスンします。
+ **[バッチサイズ]** - 単一のバッチで取得するメッセージの最大数 (最大 10,000 件) を設定します。デフォルトバッチサイズは 100 です。
+ **[開始位置] –** レコードの読み取りを開始するストリーム内の位置を選択します。
+ **[最新] -** ストリームに追加された新しいレコードのみを処理します。関数は、Lambda がイベントソースの作成を完了した後にのみ、レコードの処理を開始します。これは、イベントソースが正常に作成されるまで、一部のレコードが削除される可能性があることを意味します。
+ **[水平トリム]** - ストリーム内のすべてのレコードを処理します。Lambda は、クラスターのログ保持期間を使用して、イベントの読み取りを開始する場所を決定します。具体的には、Lambda は `current_time - log_retention_duration` から読み取りを開始します。Lambda がすべてのイベントを読み取るには、このタイムスタンプの前に変更ストリームが既にアクティブになっている必要があります。
+ **[タイムスタンプ]** - 特定の時刻以降のレコードを処理します。Lambda がすべてのイベントを適切に読み取るには、指定されたタイムスタンプの前に変更ストリームが既にアクティブになっている必要があります。
+ **[認証]** – クラスター内のブローカーにアクセスするための認証方法を選択します。
+ **[BASIC\$1AUTH]** – 基本認証では、クラスターにアクセスするための認証情報を含む Secrets Manager キーを指定する必要があります。
+ **[Secrets Manager キー]** – Amazon DocumentDB クラスターへのアクセスに必要な認証の詳細 (ユーザー名とパスワード) を含む Secrets Manager キーを選択します。
+ (オプション) **[バッチウィンドウ]** - 関数を呼び出す前にレコードを収集する最大時間を 300 までの秒数で設定します。
+ (オプション) **[ドキュメントの完全な設定]** – ドキュメントの更新オペレーションでは、ストリームに送信するものを選択します。デフォルト値は `Default` です。これは、各変更ストリームイベントについて、行われた変更について記述するデルタのみを Amazon DocumentDB が送信することを意味します。このフィールドの詳細については、MongoDB Javadocs の API ドキュメントの「[FullDocument](https://mongodb.github.io/mongo-java-driver/3.9/javadoc/com/mongodb/client/model/changestream/FullDocument.html#DEFAULT)」を参照してください。
+ **[デフォルト]** – Lambda は、行われた変更について記述する部分的なドキュメントのみを送信します。
+ **[UpdateLookup]** – Lambda は、ドキュメント全体のコピーとともに、変更について記述するデルタを送信します。
## Amazon DocumentDB イベントソースマッピングを作成する (SDK または CLI)
Amazon DocumentDB イベントソースマッピングを [AWS SDK](https://aws.amazon.com/developer/tools/) を使用して作成または管理するには、次の API オペレーションを使用します。
+ [CreateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_CreateEventSourceMapping.html)
+ [ListEventSourceMappings](https://docs.aws.amazon.com/lambda/latest/api/API_ListEventSourceMappings.html)
+ [GetEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_GetEventSourceMapping.html)
+ [UpdateEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_UpdateEventSourceMapping.html)
+ [DeleteEventSourceMapping](https://docs.aws.amazon.com/lambda/latest/api/API_DeleteEventSourceMapping.html)
AWS CLI を使用してイベントソースマッピングを作成するには、[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/create-event-source-mapping.html) コマンドを使用します。以下の例では、このコマンドを使用して、`my-function` という名前の関数を Amazon DocumentDB 変更ストリームにマッピングします。イベントソースは Amazon リソースネーム (ARN) によって指定され、バッチサイズ 500 で、Unix 時間形式のタイムスタンプから始まります。このコマンドでは、Lambda が Amazon DocumentDB への接続に使用する Secrets Manager キーも指定します。さらに、データベースと読み取り元のコレクションを指定する `document-db-event-source-config` パラメーターも含まれています。
```
aws lambda create-event-source-mapping --function-name my-function \
--event-source-arn arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy
--batch-size 500 \
--starting-position AT_TIMESTAMP \
--starting-position-timestamp 1541139109 \
--source-access-configurations '[{"Type":"BASIC_AUTH","URI":"arn:aws:secretsmanager:us-east-1:123456789012:secret:DocDBSecret-BAtjxi"}]' \
--document-db-event-source-config '{"DatabaseName":"test_database", "CollectionName": "test_collection"}' \
```
次のような出力が表示されます。
```
{
"UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
"BatchSize": 500,
"DocumentDBEventSourceConfig": {
"CollectionName": "test_collection",
"DatabaseName": "test_database",
"FullDocument": "Default"
},
"MaximumBatchingWindowInSeconds": 0,
"EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
"LastModified": 1541348195.412,
"LastProcessingResult": "No records processed",
"State": "Creating",
"StateTransitionReason": "User action"
}
```
作成後、[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/update-event-source-mapping.html) コマンドを使用して、Amazon DocumentDB イベントソースに関連する設定を更新できます。次の例では、バッチサイズを 1,000 に更新し、バッチウィンドウを 10 秒に更新します。このコマンドには、`list-event-source-mapping` コマンドまたは Lambda コンソールから取得できるイベントソースマッピングの UUID が必要です。
```
aws lambda update-event-source-mapping --function-name my-function \
--uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b \
--batch-size 1000 \
--batch-window 10
```
このような出力が表示されます。
```
{
"UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
"BatchSize": 500,
"DocumentDBEventSourceConfig": {
"CollectionName": "test_collection",
"DatabaseName": "test_database",
"FullDocument": "Default"
},
"MaximumBatchingWindowInSeconds": 0,
"EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
"LastModified": 1541359182.919,
"LastProcessingResult": "OK",
"State": "Updating",
"StateTransitionReason": "User action"
}
```
Lambda は設定を非同期的に更新するため、プロセスが完了するまでこれらの変更が出力に表示されない場合があります。イベントソースマッピングの現在の設定を表示するには、[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/get-event-source-mapping.html) コマンドを使用します。
```
aws lambda get-event-source-mapping --uuid f89f8514-cdd9-4602-9e1f-01a5b77d449b
```
このような出力が表示されます。
```
{
"UUID": "2b733gdc-8ac3-cdf5-af3a-1827b3b11284",
"DocumentDBEventSourceConfig": {
"CollectionName": "test_collection",
"DatabaseName": "test_database",
"FullDocument": "Default"
},
"BatchSize": 1000,
"MaximumBatchingWindowInSeconds": 10,
"EventSourceArn": "arn:aws:rds:us-west-2:123456789012:cluster:privatecluster7de2-epzcyvu4pjoy",
"FunctionArn": "arn:aws:lambda:us-west-2:123456789012:function:my-function",
"LastModified": 1541359182.919,
"LastProcessingResult": "OK",
"State": "Enabled",
"StateTransitionReason": "User action"
}
```
Amazon DocumentDB イベントソースマッピングを削除するには、[https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/delete-event-source-mapping.html](https://awscli.amazonaws.com/v2/documentation/api/latest/reference/lambda/delete-event-source-mapping.html) コマンドを使用します。
```
aws lambda delete-event-source-mapping \
--uuid 2b733gdc-8ac3-cdf5-af3a-1827b3b11284
```
## ポーリングとストリームの開始位置
イベントソースマッピングの作成時および更新時のストリームのポーリングは、最終的に一貫性があることに注意してください。
+ イベントソースマッピングの作成時、ストリームからのイベントのポーリングが開始されるまでに数分かかる場合があります。
+ イベントソースマッピングの更新時、ストリームからのイベントのポーリングが停止および再開されるまでに数分かかる場合があります。
つまり、`LATEST` をストリームの開始位置として指定すると、イベントソースマッピングの作成または更新中にイベントを見逃す可能性があります。イベントを見逃さないようにするには、ストリームの開始位置を `TRIM_HORIZON` または `AT_TIMESTAMP` として指定します。
## Amazon DocumentDB イベントソースのモニタリング
Amazon DocumentDB イベントソースのモニタリングに役立つよう、関数がレコードのバッチの処理を完了したときに、Lambda は `IteratorAge` メトリクスを発行します。*イテレータの有効期間*とは、最新のイベントのタイムスタンプと現在のタイムスタンプの差のことです。基本的に、`IteratorAge` メトリクスは、バッチで最後に処理されたレコードの古さを示します。関数が新しいイベントを現在処理している場合、イテレーターの有効期間を使用して、レコードが追加されてから関数によって処理されるまでのレイテンシーを推定できます。`IteratorAge` の増加傾向は、関数に問題があることを示している可能性があります。詳細については、「[Lambda での CloudWatch メトリクスの使用](monitoring-metrics.md)」を参照してください。
Amazon DocumentDB の変更ストリームは、イベント間の大きなタイムギャップを処理するようには最適化されていません。Amazon DocumentDB イベントソースが長期間にイベントを受信しない場合、Lambda はイベントソースマッピングを無効にすることがあります。この期間の長さは、クラスターのサイズやその他のワークロードに応じて、数週間から数か月までさまざまです。
Lambda は最大 6 MB のペイロードをサポートします。ただし、Amazon DocumentDB 変更ストリームイベントのサイズは最大 16 MB です。変更ストリームが 6 MB を超える変更ストリームイベントを Lambda に送信しようとすると、Lambda はメッセージを削除して `OversizedRecordCount` メトリクスを発行します。Lambda は、ベストエフォートベースですべてのメトリクスを発行します。
# チュートリアル: Amazon DocumentDB を用いて AWS Lambda のストリームの使用
このチュートリアルでは、Amazon DocumentDB (MongoDB 互換) 変更ストリームからのイベントを処理する基本的な Lambda 関数を作成します。このチュートリアルは、以下の段階を通じて完了します。
+ Amazon DocumentDB クラスターをセットアップして接続し、そのクラスターで変更ストリームをアクティブ化します。
+ Lambda 関数を作成し、Amazon DocumentDB クラスターを関数のイベントソースとして設定します。
+ Amazon DocumentDB データベースにアイテムを挿入して、セットアップをテストします。
## Amazon DocumentDB クラスターの作成
1. [Amazon DocumentDB コンソール](https://console.aws.amazon.com/docdb/home#)を開きます。**[クラスタ]** で **[作成]** を選択します。
1. 次の設定でクラスターを作成します。
+ **[クラスタータイプ]** で **[インスタンスベースのクラスター]** を選択します。これがデフォルトのオプションです。
+ **[クラスター設定]** で、**[エンジンバージョン]** 5.0.0 が選択されていることを確認します。これがデフォルトのオプションです。
+ **[インスタンス設定]** で次の設定を行なってください。
+ **[DB インスタンスクラス]** で、**[メモリ最適化クラス]** を選択します。これがデフォルトのオプションです。
+ **[通常のレプリカインスタンスの数]** には 1 を選択します。
+ **[インスタンスクラス]** にはデフォルトの選択を使用します。
+ **[認証]** でプライマリユーザーのユーザー名を入力し、**[セルフマネージド]** を選択します。パスワードを入力して確認します。
+ 他のデフォルト設定をすべて維持します。
1. **[クラスターを作成]** を選択します。
## Secrets Manager でシークレットを作成する
Amazon DocumentDB がクラスターを作成している間に、データベースの認証情報を保存する AWS Secrets Manager シークレットを作成します。このシークレットは、後の手順で Lambda イベントソースマッピングを作成するときに指定します。
**Secrets Manager でシークレットを保存するには**
1. [[Secrets Manager]](https://console.aws.amazon.com/secretsmanager/home#) コンソールを開き、**[新しいシークレットを保存]** を選択します。
1. **[シークレットのタイプを選択]** で、以下のいずれかのオプションを選択します。
+ **[基本情報]** の下:
+ **[シークレットタイプ]**: Amazon DocumentDB データベース用の認証情報
+ **[認証情報]** で、Amazon DocumentDB クラスターの作成に使用したユーザー名およびパスワードを入力します。
+ **[データベース]**: ご使用の Amazon DocumentDB クラスターを選択します。
+ [**次へ**] を選択します。
1. **[条件]** は、以下のオプションから選択します。
+ **[シークレット名]**: `DocumentDBSecret`
+ [**次へ**] を選択します。
1. [**次へ**] を選択します。
1. [**保存する**] を選択します。
1. コンソールを更新して、`DocumentDBSecret` シークレットが正常に保存されたことを確認します。
**[シークレット ARN]** をメモしておきます。これは、後のステップで必要になります。
## クラスターに接続する
**AWS CloudShell を使用して Amazon DocumentDB クラスターに接続する**
1. Amazon DocumentDB マネジメントコンソールの **[クラスター]** で、作成したクラスターを探します。クラスターの横にあるチェックボックスをクリックして選択します。
1. **[クラスターに接続]** を選択します。CloudShell の **[Run コマンド]** 画面が表示されます。
1. **[新しい環境の名前]** フィールドに、「test」などの一意の名前を入力し、**[作成して実行する]** を選択します。
1. プロンプトが表示されたら、パスワードを入力します。プロンプトが `rs0 [direct: primary] >` に変わったら、Amazon DocumentDB クラスターに正常に接続できています。
## 変更ストリームを有効にする
本チュートリアルでは、Amazon DocumentDB クラスター内にある `docdbdemo` データベースの `products` コレクションへの変更を追跡します。これを行うには、[[変更ストリーム]](https://docs.aws.amazon.com/documentdb/latest/developerguide/change_streams.html) を有効にします。
**クラスター内に新しいデータベースを作成するには**
1. 次のコマンドを実行し、`docdbdemo` と呼ばれる新しいデータベースを作成します。
```
use docdbdemo
```
1. ターミナルウィンドウで、次のコマンドを使用してレコードを `docdbdemo` に挿入します。
```
db.products.insertOne({"hello":"world"})
```
次のような出力が表示されます。
```
{
acknowledged: true,
insertedId: ObjectId('67f85066ca526410fd531d59')
}
```
1. 次に、次のコマンドを使用して、`docdbdemo` データベースの `products` コレクションの変更ストリームを有効にします。
```
db.adminCommand({modifyChangeStreams: 1,
database: "docdbdemo",
collection: "products",
enable: true});
```
次のような出力が表示されます。
```
{ "ok" : 1, "operationTime" : Timestamp(1680126165, 1) }
```
## インターフェイス VPC エンドポイントを作成する
次に、[インターフェイス VPC エンドポイント](https://docs.aws.amazon.com/vpc/latest/privatelink/create-interface-endpoint.html#create-interface-endpoint-aws)を作成して、Lambda と Secrets Manager (後でクラスターアクセス認証情報を保存するために使用) がデフォルト VPC に接続できるようにします。
**インターフェイス VPC エンドポイントを作成するには**
1. [[VPC] コンソール](https://console.aws.amazon.com/vpc/home#)を開きます。左側のメニューの **[仮想プライベートクラウド]** で、**[エンドポイント]** を選択します。
1. **エンドポイントの作成** を選択します。次の構成でエンドポイントを作成します。
+ **[名前タグ]** に「`lambda-default-vpc`」を入力します。
+ **[サービスカテゴリ]** で、[AWS サービス] を選択します。
+ **サービス**には、検索ボックスで「`lambda`」と入力します。フォーマット `com.amazonaws..lambda` のサービスを選択してください。
+ **[VPC]** で、Amazon DocumentDB クラスターがある VPC を選択します。通常、これは[デフォルト VPC](https://docs.aws.amazon.com/vpc/latest/userguide/default-vpc.html) になります。
+ **サブネット** には、各アベイラビリティーゾーンの横にあるボックスをチェックします。それぞれのアベイラビリティゾーンに正しいサブネット ID を選択します。
+ [**IP アドレスの種類**] には [IPv4] を選択します。
+ **[セキュリティグループ]** で、Amazon DocumentDB クラスターが使用するセキュリティグループを選択します。通常、これは `default` セキュリティグループになります。
+ 他のデフォルト設定をすべて維持します。
+ **エンドポイントの作成** を選択します。
1. **[エンドポイントの作成]** を再び選択します。次の構成でエンドポイントを作成します。
+ **[名前タグ]** に「`secretsmanager-default-vpc`」を入力します。
+ **[サービスカテゴリ]** で、[AWS サービス] を選択します。
+ **サービス**には、検索ボックスで「`secretsmanager`」と入力します。フォーマット `com.amazonaws..secretsmanager` のサービスを選択してください。
+ **[VPC]** で、Amazon DocumentDB クラスターがある VPC を選択します。通常、これは[デフォルト VPC](https://docs.aws.amazon.com/vpc/latest/userguide/default-vpc.html) になります。
+ **サブネット** には、各アベイラビリティーゾーンの横にあるボックスをチェックします。それぞれのアベイラビリティゾーンに正しいサブネット ID を選択します。
+ [**IP アドレスの種類**] には [IPv4] を選択します。
+ **[セキュリティグループ]** で、Amazon DocumentDB クラスターが使用するセキュリティグループを選択します。通常、これは `default` セキュリティグループになります。
+ 他のデフォルト設定をすべて維持します。
+ **エンドポイントの作成** を選択します。
これで、このチュートリアルのクラスターセットアップの部分は完了です。
## 実行ロールを作成する
次のステップでは、Lambda 関数を作成します。まず、クラスターにアクセスするためのアクセス許可を関数に付与する実行ロールを作成する必要があります。これを行うには、最初に IAM ポリシーを作成してから、次にこのポリシーを IAM ロールにアタッチします。
**IAM ポリシーを作成するには**
1. IAM コンソールの [[ポリシー] ページ](https://console.aws.amazon.com/iam/home#/policies) を開き、**[ポリシーの作成]** を選択します。
1. **JSON** タブを選択します。次のポリシーでは、ステートメントの最後の行にある Secrets Manager リソース ARN を以前のシークレット ARN で置き換え、ポリシーをエディタにコピーします。
------
#### [ JSON ]
****
```
{
"Version":"2012-10-17",
"Statement": [
{
"Sid": "LambdaESMNetworkingAccess",
"Effect": "Allow",
"Action": [
"ec2:CreateNetworkInterface",
"ec2:DescribeNetworkInterfaces",
"ec2:DescribeVpcs",
"ec2:DeleteNetworkInterface",
"ec2:DescribeSubnets",
"ec2:DescribeSecurityGroups",
"kms:Decrypt"
],
"Resource": "*"
},
{
"Sid": "LambdaDocDBESMAccess",
"Effect": "Allow",
"Action": [
"rds:DescribeDBClusters",
"rds:DescribeDBClusterParameters",
"rds:DescribeDBSubnetGroups"
],
"Resource": "*"
},
{
"Sid": "LambdaDocDBESMGetSecretValueAccess",
"Effect": "Allow",
"Action": [
"secretsmanager:GetSecretValue"
],
"Resource": "arn:aws:secretsmanager:us-east-1:123456789012:secret:DocumentDBSecret"
}
]
}
```
------
1. **[次へ: タグ]**、**[次へ: 確認]** の順に選択します。
1. **[Name]** (名前) に `AWSDocumentDBLambdaPolicy` と入力します。
1. [**ポリシーの作成**] を選択します。
**IAM ロールを作成するには**
1. IAM コンソールの [[ロール] ページ](https://console.aws.amazon.com/iam/home#/roles)を開いて、**[ロールの作成]** を選択します。
1. **[信頼できるエンティティを選択]** には、次のオプションを選択します。
+ **[信頼できるエンティティタイプ]**: AWS サービス
+ **[サービスまたはユースケース]**: Lambda
+ [**次へ**] を選択します。
1. **[アクセス権限の追加]** では、作成したばかりの `AWSDocumentDBLambdaPolicy` ポリシーを選択し、`AWSLambdaBasicExecutionRole` と同様に関数に Amazon CloudWatch Logs への書き込み権限を付与します。
1. [**次へ**] を選択します。
1. **[Role name]** (ロール名) に`AWSDocumentDBLambdaExecutionRole`と入力します。
1. [**ロールの作成**] を選択してください。
## Lambda 関数を作成する
このチュートリアルは Python 3.14 ランタイムを使用しますが、他のランタイム用のサンプルコードのファイルも用意しています。次のボックスでタブを選択すると、関心のあるランタイムのコードが表示されます。
このコードは Amazon DocumentDB イベント入力を受け取り、そこに含まれるメッセージを処理します。
**Lambda 関数を作成するには**
1. Lambda コンソールの [[関数]](https://console.aws.amazon.com/lambda/home#/functions) ページを開きます。
1. [**関数の作成**] を選択してください。
1. **[一から作成]** を選択します。
1. **[基本的な情報]** で、以下を実行します。
1. **[関数名]** に `ProcessDocumentDBRecords` と入力します。
1. **[ランタイム]** には、**[Python 3.14]** を選択します。
1. **[アーキテクチャ]** で **[x86\$164]** を選択します。
1. **[デフォルトの実行ロールの変更]** タブで、次の操作を行います。
1. タブを展開し、**[既存のロールを使用する]** を選択します。
1. 先ほど作成した `AWSDocumentDBLambdaExecutionRole` を選択します。
1. [**関数の作成**] を選択してください。
**関数コードをデプロイするには**
1. 次のボックスで **[Python]** タブを選択し、コードをコピーします。
------
#### [ .NET ]
**SDK for .NET**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
.NET を使用した Lambda での Amazon DocumentDB イベントの消費。
```
using Amazon.Lambda.Core;
using System.Text.Json;
using System;
using System.Collections.Generic;
using System.Text.Json.Serialization;
//Assembly attribute to enable the Lambda function's JSON input to be converted into a .NET class.
[assembly: LambdaSerializer(typeof(Amazon.Lambda.Serialization.SystemTextJson.DefaultLambdaJsonSerializer))]
namespace LambdaDocDb;
public class Function
{
///
/// Lambda function entry point to process Amazon DocumentDB events.
///
/// The Amazon DocumentDB event.
/// The Lambda context object.
/// A string to indicate successful processing.
public string FunctionHandler(Event evnt, ILambdaContext context)
{
foreach (var record in evnt.Events)
{
ProcessDocumentDBEvent(record, context);
}
return "OK";
}
private void ProcessDocumentDBEvent(DocumentDBEventRecord record, ILambdaContext context)
{
var eventData = record.Event;
var operationType = eventData.OperationType;
var databaseName = eventData.Ns.Db;
var collectionName = eventData.Ns.Coll;
var fullDocument = JsonSerializer.Serialize(eventData.FullDocument, new JsonSerializerOptions { WriteIndented = true });
context.Logger.LogLine($"Operation type: {operationType}");
context.Logger.LogLine($"Database: {databaseName}");
context.Logger.LogLine($"Collection: {collectionName}");
context.Logger.LogLine($"Full document:\n{fullDocument}");
}
public class Event
{
[JsonPropertyName("eventSourceArn")]
public string EventSourceArn { get; set; }
[JsonPropertyName("events")]
public List Events { get; set; }
[JsonPropertyName("eventSource")]
public string EventSource { get; set; }
}
public class DocumentDBEventRecord
{
[JsonPropertyName("event")]
public EventData Event { get; set; }
}
public class EventData
{
[JsonPropertyName("_id")]
public IdData Id { get; set; }
[JsonPropertyName("clusterTime")]
public ClusterTime ClusterTime { get; set; }
[JsonPropertyName("documentKey")]
public DocumentKey DocumentKey { get; set; }
[JsonPropertyName("fullDocument")]
public Dictionary FullDocument { get; set; }
[JsonPropertyName("ns")]
public Namespace Ns { get; set; }
[JsonPropertyName("operationType")]
public string OperationType { get; set; }
}
public class IdData
{
[JsonPropertyName("_data")]
public string Data { get; set; }
}
public class ClusterTime
{
[JsonPropertyName("$timestamp")]
public Timestamp Timestamp { get; set; }
}
public class Timestamp
{
[JsonPropertyName("t")]
public long T { get; set; }
[JsonPropertyName("i")]
public int I { get; set; }
}
public class DocumentKey
{
[JsonPropertyName("_id")]
public Id Id { get; set; }
}
public class Id
{
[JsonPropertyName("$oid")]
public string Oid { get; set; }
}
public class Namespace
{
[JsonPropertyName("db")]
public string Db { get; set; }
[JsonPropertyName("coll")]
public string Coll { get; set; }
}
}
```
------
#### [ Go ]
**SDK for Go V2**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Go を使用して Lambda で Amazon DocumentDB イベントの消費。
```
package main
import (
"context"
"encoding/json"
"fmt"
"github.com/aws/aws-lambda-go/lambda"
)
type Event struct {
Events []Record `json:"events"`
}
type Record struct {
Event struct {
OperationType string `json:"operationType"`
NS struct {
DB string `json:"db"`
Coll string `json:"coll"`
} `json:"ns"`
FullDocument interface{} `json:"fullDocument"`
} `json:"event"`
}
func main() {
lambda.Start(handler)
}
func handler(ctx context.Context, event Event) (string, error) {
fmt.Println("Loading function")
for _, record := range event.Events {
logDocumentDBEvent(record)
}
return "OK", nil
}
func logDocumentDBEvent(record Record) {
fmt.Printf("Operation type: %s\n", record.Event.OperationType)
fmt.Printf("db: %s\n", record.Event.NS.DB)
fmt.Printf("collection: %s\n", record.Event.NS.Coll)
docBytes, _ := json.MarshalIndent(record.Event.FullDocument, "", " ")
fmt.Printf("Full document: %s\n", string(docBytes))
}
```
------
#### [ Java ]
**SDK for Java 2.x**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Java を使用した Lambda での Amazon DocumentDB イベントの消費。
```
import java.util.List;
import java.util.Map;
import com.amazonaws.services.lambda.runtime.Context;
import com.amazonaws.services.lambda.runtime.RequestHandler;
public class Example implements RequestHandler, String> {
@SuppressWarnings("unchecked")
@Override
public String handleRequest(Map event, Context context) {
List> events = (List>) event.get("events");
for (Map record : events) {
Map eventData = (Map) record.get("event");
processEventData(eventData);
}
return "OK";
}
@SuppressWarnings("unchecked")
private void processEventData(Map eventData) {
String operationType = (String) eventData.get("operationType");
System.out.println("operationType: %s".formatted(operationType));
Map ns = (Map) eventData.get("ns");
String db = (String) ns.get("db");
System.out.println("db: %s".formatted(db));
String coll = (String) ns.get("coll");
System.out.println("coll: %s".formatted(coll));
Map fullDocument = (Map) eventData.get("fullDocument");
System.out.println("fullDocument: %s".formatted(fullDocument));
}
}
```
------
#### [ JavaScript ]
**SDK for JavaScript (v3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
JavaScript を使用した Lambda での Amazon DocumentDB イベントの消費。
```
console.log('Loading function');
exports.handler = async (event, context) => {
event.events.forEach(record => {
logDocumentDBEvent(record);
});
return 'OK';
};
const logDocumentDBEvent = (record) => {
console.log('Operation type: ' + record.event.operationType);
console.log('db: ' + record.event.ns.db);
console.log('collection: ' + record.event.ns.coll);
console.log('Full document:', JSON.stringify(record.event.fullDocument, null, 2));
};
```
TypeScript を使用して Lambda で Amazon DocumentDB イベントの消費。
```
import { DocumentDBEventRecord, DocumentDBEventSubscriptionContext } from 'aws-lambda';
console.log('Loading function');
export const handler = async (
event: DocumentDBEventSubscriptionContext,
context: any
): Promise => {
event.events.forEach((record: DocumentDBEventRecord) => {
logDocumentDBEvent(record);
});
return 'OK';
};
const logDocumentDBEvent = (record: DocumentDBEventRecord): void => {
console.log('Operation type: ' + record.event.operationType);
console.log('db: ' + record.event.ns.db);
console.log('collection: ' + record.event.ns.coll);
console.log('Full document:', JSON.stringify(record.event.fullDocument, null, 2));
};
```
------
#### [ PHP ]
**SDK for PHP**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
PHP を使用して Lambda で Amazon DocumentDB イベントの消費。
```
logDocumentDBEvent($record['event']);
}
return 'OK';
}
private function logDocumentDBEvent($event): void
{
// Extract information from the event record
$operationType = $event['operationType'] ?? 'Unknown';
$db = $event['ns']['db'] ?? 'Unknown';
$collection = $event['ns']['coll'] ?? 'Unknown';
$fullDocument = $event['fullDocument'] ?? [];
// Log the event details
echo "Operation type: $operationType\n";
echo "Database: $db\n";
echo "Collection: $collection\n";
echo "Full document: " . json_encode($fullDocument, JSON_PRETTY_PRINT) . "\n";
}
}
return new DocumentDBEventHandler();
```
------
#### [ Python ]
**SDK for Python (Boto3)**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Python を使用して Lambda で Amazon DocumentDB イベントの消費。
```
import json
def lambda_handler(event, context):
for record in event.get('events', []):
log_document_db_event(record)
return 'OK'
def log_document_db_event(record):
event_data = record.get('event', {})
operation_type = event_data.get('operationType', 'Unknown')
db = event_data.get('ns', {}).get('db', 'Unknown')
collection = event_data.get('ns', {}).get('coll', 'Unknown')
full_document = event_data.get('fullDocument', {})
print(f"Operation type: {operation_type}")
print(f"db: {db}")
print(f"collection: {collection}")
print("Full document:", json.dumps(full_document, indent=2))
```
------
#### [ Ruby ]
**SDK for Ruby**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Ruby を使用して Lambda で Amazon DocumentDB イベントの消費。
```
require 'json'
def lambda_handler(event:, context:)
event['events'].each do |record|
log_document_db_event(record)
end
'OK'
end
def log_document_db_event(record)
event_data = record['event'] || {}
operation_type = event_data['operationType'] || 'Unknown'
db = event_data.dig('ns', 'db') || 'Unknown'
collection = event_data.dig('ns', 'coll') || 'Unknown'
full_document = event_data['fullDocument'] || {}
puts "Operation type: #{operation_type}"
puts "db: #{db}"
puts "collection: #{collection}"
puts "Full document: #{JSON.pretty_generate(full_document)}"
end
```
------
#### [ Rust ]
**SDK for Rust**
GitHub には、その他のリソースもあります。[サーバーレスサンプル](https://github.com/aws-samples/serverless-snippets/tree/main/integration-docdb-to-lambda)リポジトリで完全な例を見つけて、設定と実行の方法を確認してください。
Rust を使用して Lambda で Amazon DocumentDB イベントの消費。
```
use lambda_runtime::{service_fn, tracing, Error, LambdaEvent};
use aws_lambda_events::{
event::documentdb::{DocumentDbEvent, DocumentDbInnerEvent},
};
// Built with the following dependencies:
//lambda_runtime = "0.11.1"
//serde_json = "1.0"
//tokio = { version = "1", features = ["macros"] }
//tracing = { version = "0.1", features = ["log"] }
//tracing-subscriber = { version = "0.3", default-features = false, features = ["fmt"] }
//aws_lambda_events = "0.15.0"
async fn function_handler(event: LambdaEvent) ->Result<(), Error> {
tracing::info!("Event Source ARN: {:?}", event.payload.event_source_arn);
tracing::info!("Event Source: {:?}", event.payload.event_source);
let records = &event.payload.events;
if records.is_empty() {
tracing::info!("No records found. Exiting.");
return Ok(());
}
for record in records{
log_document_db_event(record);
}
tracing::info!("Document db records processed");
// Prepare the response
Ok(())
}
fn log_document_db_event(record: &DocumentDbInnerEvent)-> Result<(), Error>{
tracing::info!("Change Event: {:?}", record.event);
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
tracing_subscriber::fmt()
.with_max_level(tracing::Level::INFO)
.with_target(false)
.without_time()
.init();
let func = service_fn(function_handler);
lambda_runtime::run(func).await?;
Ok(())
}
```
------
1. Lambda コンソールの **[コードソース]** ペインで、コードをコードエディタに貼り付け、Lambda が作成したコードを置き換えます。
1. **[DEPLOY]** セクションで、**[デプロイ]** を選択して関数のコードを更新します。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/getting-started-tutorial/deploy-console.png)
## Lambda イベントソースマッピングを作成します。
Amazon DocumentDB 変更ストリームを Lambda 関数と関連付けるイベントソースマッピングを作成します。このイベントソースマッピングを作成すると、AWS Lambda はストリームのポーリングをすぐに開始します。
**イベントソースマッピングを作成するには**
1. Lambda コンソールの [[関数]](https://console.aws.amazon.com/lambda/home#/functions) ページを開きます。
1. 先ほど作成した `ProcessDocumentDBRecords` 関数を選択します。
1. **[設定]** タブを選択し、左側のメニューで **[トリガー]** を選択します。
1. **[トリガーを追加]** を選択します。
1. **[トリガー設定]** で、**[Amazon DocumentDB]** をソースとして選択します。
1. イベントソースマッピングには、次の設定制限があります。
+ **[Amazon DocumentDB クラスター]**: 先ほど作成したクラスターを選択します。
+ **[データベース名]**: docdbdemo
+ **[コレクション名]**: 製品
+ **[バッチサイズ]**: 1
+ **[開始位置]**: 最新
+ **[認証]**: BASIC\$1AUTH
+ **[Secrets Manager キー]**: Amazon DocumentDB クラスターのシークレットを選択します。`rds!cluster-12345678-a6f0-52c0-b290-db4aga89274f` のような名前になります。
+ **[バッチウィンドウ]**: 1
+ **[フルドキュメント設定]**: UpdateLookup
1. **[Add]** (追加) を選択します。イベントソースマッピングの作成には数分かかる場合があります。
## 関数をテストする
イベントソースマッピングが **[有効]** 状態になるまで待ちます。これには数分間かかる場合があります。次に、データベースレコードを挿入、更新、削除して、エンドツーエンドのセットアップをテストします。開始する前に:
1. CloudShell 環境で [Amazon DocumentDB クラスターに再接続](#docdb-connect-to-cluster)します。
1. 次のコマンドを実行し、`docdbdemo` データベースを使用できているのを確認します。
```
use docdbdemo
```
### レコードを挿入する
`docdbdemo` データベースの `products` コレクションにレコードを挿入します。
```
db.products.insertOne({"name":"Pencil", "price": 1.00})
```
[CloudWatch Logs をチェック](monitoring-cloudwatchlogs-view.md#monitoring-cloudwatchlogs-console)して、関数がこのイベントを正常に処理したことを確認します。次のようなログエントリが表示されます。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/documentdb-insert-log.png)
### レコードの更新
次のコマンドを使用して挿入したレコードを更新します。
```
db.products.updateOne(
{ "name": "Pencil" },
{ $set: { "price": 0.50 }}
)
```
[CloudWatch Logs をチェック](monitoring-cloudwatchlogs-view.md#monitoring-cloudwatchlogs-console)して、関数がこのイベントを正常に処理したことを確認します。次のようなログエントリが表示されます。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/documentdb-update-log.png)
### レコードを削除する
次のコマンドを使用して更新したレコードを削除します。
```
db.products.deleteOne( { "name": "Pencil" } )
```
[CloudWatch Logs をチェック](monitoring-cloudwatchlogs-view.md#monitoring-cloudwatchlogs-console)して、関数がこのイベントを正常に処理したことを確認します。次のようなログエントリが表示されます。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/documentdb-delete-log.png)
## トラブルシューティング
関数の CloudWatch ログにデータベースイベントが表示されない場合、次の内容を確認してください。
+ Lambda イベントソースマッピング (トリガーとも呼ばれる) が **[有効]** 状態になっていることを確認します。イベントソースマッピングは作成に数分かかる場合があります。
+ イベントソースマッピングが **[有効]** になっても CloudWatch にデータベースイベントが表示されない場合、次の内容を実行してください。
+ イベントソースマッピングの **[データベース名]** が `docdbdemo` に設定されていることを確認します。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/documentdb-trigger.png)
+ イベントソースマッピングの **[最終処理結果]** フィールドに、次のメッセージがあるかどうか確認します。「問題: 接続エラー。VPC は Lambda および STS との接続に加え、認証が必要な場合は Secrets Manager にも接続できる必要があります」 このエラーが表示された場合、[Lambda および Secrets Manager VPC インターフェイスエンドポイントが作成されているか](#docdb-create-interface-vpc-endpoints)、またエンドポイントは Amazon DocumentDB クラスターが使用するのと同じ VPC およびサブネットを使用しているかを確認してください。
![\[\]](http://docs.aws.amazon.com/ja_jp/lambda/latest/dg/images/documentdb-lastprocessingresult.png)
## リソースのクリーンアップ
このチュートリアル用に作成したリソースは、保持しない場合は削除できます。使用しなくなった AWS リソースを削除することで、AWS アカウント アカウントに請求される料金の発生を防ぎます。
**Lambda 関数を削除するには**
1. Lambda コンソールの [[関数]](https://console.aws.amazon.com/lambda/home#/functions) ページを開きます。
1. 作成した関数を選択します。
1. **[アクション]** で、**[削除]** を選択します。
1. テキスト入力フィールドに **confirm** と入力し、**[削除]** を選択します。
**実行ロールを削除する**
1. IAM コンソールの [[ロール]](https://console.aws.amazon.com/iam/home#/roles) ページを開きます。
1. 作成した実行ロールを選択します。
1. **[削除]** を選択します。
1. テキスト入力フィールドにロールの名前を入力し、**[削除]** を選択します。
**VPC エンドポイントを削除するには**
1. [[VPC] コンソール](https://console.aws.amazon.com/vpc/home#)を開きます。左側のメニューの **[仮想プライベートクラウド]** で、**[エンドポイント]** を選択します。
1. 作成したエンドポイントを選択します。
1. **[アクション]**、**[VPC エンドポイントを削除]** の順に選択します。
1. テキスト入力フィールドに **delete** を入力します。
1. **[削除]** を選択します。
**Amazon DocumentDB クラスターを削除するには**
1. [Amazon DocumentDB コンソール](https://console.aws.amazon.com/docdb/home#)を開きます。
1. 本チュートリアル用に作成した Amazon DocumentDB クラスターを選択し、削除保護を無効にします。
1. メインの **[クラスター]** ページで、作成した Amazon DocumentDB クラスターを再度選択します。
1. **[アクション]**、**[削除]** の順に選択します。
1. **[最終クラスタースナップショットの作成]** で **[いいえ]** を選択します。
1. テキスト入力フィールドに **delete** を入力します。
1. **[削除]** を選択します。
**シークレットを Secrets Manager で削除するには**
1. [Secrets Manager コンソール](https://console.aws.amazon.com/secretsmanager/home#)を開きます。
1. このチュートリアルで作成したシークレットを選択します。
1. **[アクション]**、**[シークレットの削除]** を選択します。
1. **[Schedule deletion]** (削除をスケジュールする) を選択します。