OpenSearch Ingestion を使って同一 OpenSrearch Serverless 内で reindex を行う

タダです.

OpenSearch Serverless で reindex をやろうとした時にサポートされている API の中になかったため,OpenSearch Ingstion を試してみました.この記事で設定を備忘録としてまとめます.なお,既に OpenSearch Serverless は構築済みのものを対象とし,設定は以下のブログを参考にしています.

aws.amazon.com

1. 専用IAM ロールとポリシーの作成

OpenSearch Ingestion が使用する IAM ロールとポリシーを作ります.権限としては以下の設定を行います.

{
    "Version": "2012-10-17",
    "Statement": [{
            "Action": [
                "aoss:BatchGetCollection",
                "aoss:APIAccessAll"
            ],
            "Effect": "Allow",
            "Resource": "arn:aws:aoss:{region}:{account-id}:collection/{collection-id}"
        },
        {
            "Action": [
                "aoss:CreateSecurityPolicy",
                "aoss:GetSecurityPolicy",
                "aoss:UpdateSecurityPolicy"
            ],
            "Effect": "Allow",
            "Resource": "*"
        }
    ]
}

加えて IAM ロールの信頼関係は次の設定です.初めて osis-pipelines.amazonaws.com を使いました.

{
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": {
            "Service": "osis-pipelines.amazonaws.com"
        },
        "Action": "sts:AssumeRole",
        "Condition": {
            "StringEquals": {
                "aws:SourceAccount": "{account-id}"
            },
            "ArnLike": {
                "aws:SourceArn": "arn:aws:osis:{region}:{account-id}:pipeline/*"
            }
        }
    }]
}

2. データアクセスポリシーの更新

既存の OpenSearch Serverless コレクションのデータアクセスポリシーを以下のように更新します.

{
    "Rules": [{
        "Resource": [
            "index/{collection-name}/*"
        ],
        "Permission": [
            "aoss:CreateIndex",
            "aoss:UpdateIndex",
            "aoss:DescribeIndex",
            "aoss:ReadDocument",
            "aoss:WriteDocument"
        ],
        "ResourceType": "index"
    }],
    "Principal": [
        "{1で作ったIAM ロール ARN }"
    ]
}

3. OpenSearch Ingestion pipeline 用のセキュリティグループの作成と OpenSearch Serverless のセキュリティグループを開放

OpenSearch Ingestion pipeline 用のセキュリティグループとしてアウトバウンドで HTTPS の通信を許可します.宛先は既存の OpenSearch Serverless の VPC エンドポイントで使用しているセキュリティグループを指定し,インバウンドでも HTTPS を許可します.

4. OpenSearch Ingestion pipeline を設定

次に OpenSearch Ingestion pipeline の設定を行っていきます.OpenSearchDataMigrationPipeline が OpenSearch Serverless の reindex を行うためのテンプレートなので,これを使用して定義します.

version: "2"
opensearch-migration-pipeline:
  source:
    opensearch:
      acknowledgments: true
      hosts: [ "既存 OpenSearch endpoint" ]
      indices:
        include:
         - index_name_regex: "reindex 元の index名"
      aws:
        region: "ap-northeast-1"
        sts_role_arn: "1  で作成した IAM ロール ARN"
        serverless: true
        serverless_options:
          network_policy_name: "既存の OpenSearch Serverelss で使用しているネットワークポリシー名"
  sink:
    - opensearch:
        hosts: [ "既存 OpenSearch endpoint" ]
        aws:
          sts_role_arn: "1  で作成した IAM ロール ARN"
          region: "ap-northeast-1"
          serverless: true
          serverless_options:
            network_policy_name: "既存の OpenSearch Serverelss で使用しているネットワークポリシー名"
        index: "reindex 先の index名"

上記の設定を行ったら自動で OpenSearch Ingestion pipeline が動作し始めます.この時に OpenSearch Ingestion pipeline が自動で VPC エンドポイントが作られ, Created by Data Prepper というタイトルで既存のネットワークポリシーが更新されます.データ量に応じて reindex が行われますが,完了すると次のログが出力されました.

2025-01-31T09:04:08.189 [pool-12-thread-1] INFO org.opensearch.dataprepper.plugins.source.opensearch.worker.NoSearchContextWorker - Completed processing for index: 'reindex 元の index名'

2025-01-31T09:05:16.177 [acknowledgement-callback-1] INFO org.opensearch.dataprepper.plugins.source.opensearch.worker.WorkerCommonUtils - Received acknowledgment of completion from sink for index reindex 元の index名

以下はサンプルになりますが,OpenSearch のダッシュボードでも確認してインデックスのドキュメント数が一致していたので期待通りになっていそうでした.

GET _cat/indices
  reindex 元の index 名 hogehoge   1111111   1111  11.1mb 11.1mb
  reindex 先の index 名 fugfuga       1111111     0 11.1mb 11.1mb

まとめ

OpenSearch Ingestion を使って同一 OpenSrearch Serverless 内で reindex を行った時の設定をまとめました.