[AWS] DynamoDBとOpenSearch Serverlessを連携する

## 事前準備
1. DynamoDBテーブルの作成

    - オンデマンドのテーブルを作成しておきます。テーブル名を`opensearch-sample`パーティションキー名は`pk`としました
    - DynamoDBのインテグレーションというメニューを開き、そこからポイントインタイムリカバリ (PITR) と DynamoDB ストリーム　を有効にします
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/491627/e02cfc4a-4b72-e51c-336d-b81fc5ec4ef6.png)

1. S3バケットの作成

    - テーブルの内容をExportするためのS3バケットを用意しておきます。バケット名を`opensearch-sample-20240409`としました

## 本作業

1. IAM Policyの作成
    [公式ドキュメント](https://docs.aws.amazon.com/opensearch-service/latest/developerguide/configure-client-ddb.html)を参考にPolicyを作ります。アカウントID、DynamoDBテーブル名およびS3バケット名を先ほど作成したものに差し替えてください。ポリシー名は`opensearch-dynamodb-pipeline`としました
    
    ```JSON
    {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Sid": "allowRunExportJob",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:DescribeTable",
                    "dynamodb:DescribeContinuousBackups",
                    "dynamodb:ExportTableToPointInTime"
                ],
                "Resource": [
                    "arn:aws:dynamodb:us-east-1:{account-id}:table/opensearch-sample"
                ]
            },
            {
                "Sid": "allowCheckExportjob",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:DescribeExport"
                ],
                "Resource": [
                    "arn:aws:dynamodb:us-east-1:{account-id}:table/opensearch-sample/export/*"
                ]
            },
            {
                "Sid": "allowReadFromStream",
                "Effect": "Allow",
                "Action": [
                    "dynamodb:DescribeStream",
                    "dynamodb:GetRecords",
                    "dynamodb:GetShardIterator"
                ],
                "Resource": [
                    "arn:aws:dynamodb:us-east-1:{account-id}:table/opensearch-sample/stream/*"
                ]
            },
            {
                "Sid": "allowReadAndWriteToS3ForExport",
                "Effect": "Allow",
                "Action": [
                    "s3:GetObject",
                    "s3:AbortMultipartUpload",
                    "s3:PutObject",
                    "s3:PutObjectAcl"
                ],
                "Resource": [
                    "arn:aws:s3:::opensearch-sample-20240409/export/*"
                ]
            }
        ]
    }
    ```

1. IAM Roleの作成
    - 信頼されたエンティティからカスタム信頼ポリシーを選択し、以下の内容をコピペします
        ```JSON
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Principal": {
                        "Service": "osis-pipelines.amazonaws.com"
                    },
                    "Action": "sts:AssumeRole"
                }
            ]
        }
        ```
    - 先ほど作成したポリシー`opensearch-dynamodb-pipeline`をアタッチしてロールを作成します。ロール名は`opensearch-serverless-role`としました

1. OpenSearch Serverless コレクションの作成
    - コレクションの作成
        コレクション名は`opensearch-serverless`としました
        ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/491627/2f499cc8-16b7-294e-1244-0d2c3de45938.png)


    - IAM Policyの作成
    [公式ドキュメント](https://docs.aws.amazon.com/ja_jp/opensearch-service/latest/developerguide/pipeline-collection-access.html)を参考にPipelineが後で作成するOpenSearch ServerlessコレクションにアクセスするためのPolicyを作成します。ポリシー名は`opensearch-access`としました
        ```JSON
        {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Action": [
                        "aoss:BatchGetCollection",
                        "aoss:APIAccessAll"
                    ],
                    "Effect": "Allow",
                    "Resource": "arn:aws:aoss:ap-northeast-1:{your-account-id}:collection/*"
                },
                {
                    "Action": [
                        "aoss:CreateSecurityPolicy",
                        "aoss:GetSecurityPolicy",
                        "aoss:UpdateSecurityPolicy"
                    ],
                    "Effect": "Allow",
                    "Resource": "*",
                    "Condition": {
                        "StringEquals": {
                            "aoss:collection": "opensearch-serverless-sample"
                        }
                    }
                }
            ]
        }
        ```

    - IAMロール`opensearch-dynamodb-pipeline`へのポリシー`opensearch-access`のアタッチ

1. データアクセスポリシーの設定
    データアクセスポリシーが自動作成されるため、プリンシパルに今回作成したIAMロールを追加する必要があります
    データアクセスポリシーにopensearch-serverless-roleを追加して保存してください


1. Pipelineの構築
    AWS-DynamoDBChangeDataCapturePipelineをブループリントに組み立てていきます
    ```yaml
    version: "2"
    dynamodb-pipeline:
      source:
        dynamodb:
          acknowledgments: true
          tables:
            - table_arn: "arn:aws:dynamodb:ap-northeast-1:{account-id}:table/opensearch-sample"
              stream:
                start_position: "LATEST"
              export:
                s3_bucket: "opensearch-sample-20240409"
                s3_region: "ap-northeast-1"
                s3_prefix: "export/"
          aws:
             sts_role_arn: "arn:aws:iam::{account-id}:role/opensearch-serverless-role"
             region: "ap-northeast-1"
      sink:
        - opensearch:
            # 先ほど作成したOpenSearch Serverless コレクションのEndopoint
            hosts: [ "<<https://search-mydomain-1a2a3a4a5a6a7a8a9a0a9a8a7a.us-east-1.es.amazonaws.com>>" ]
            index: "table-index"
            index_type: custom
            document_id: "${getMetadata(\"primary_key\")}"
            action: "${getMetadata(\"opensearch_action\")}"
            document_version: "${getMetadata(\"document_version\")}"
            document_version_type: "external"
            aws:
              sts_role_arn: "arn:aws:iam::{account-id}:role/opensearch-serverless-role"
              region: "ap-northeast-1"
              # サーバーレスの場合の設定
              serverless: true
    
    ```

1. 作成完了まで待ちます
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/491627/7985eaa5-61a2-3c1e-8471-b0400e2b75e7.png)

1. Indexができています
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/491627/f34699e8-09aa-76cc-e25d-446674070b07.png)


## ハマりポイント
- IAMの権限設定が難しいです。とくにS3の出力先フォルダ（Prefix）名、リージョンを確認する。
- aoss:BatchGetCollectionはcollectionのIDを指定してもだめなので、*でリソースを限定しない
- パイプラインの設定を変えた後反映されない場合、一度停止してから再開する（時間はかかる）
- Dashboardにアクセスできない場合、ネットワークポリシーからダッシュボードアクセスに許可を与える
- CloudWatchに出力されるログが読みにくいが、どのリソースでエラーがでているかはわかるので、都度IAMまわり見直す