Kinesis Data Firehose から Amazon OpenSearch Service へのクロスアカウントストリーミングをセットアップするにはどうすればよいですか?

最終更新日: 2021 年 7 月 23 日

別アカウントにある Amazon OpenSearch Service クラスターに、データを送信する Amazon Kinesis Data Firehose ストリームをセットアップしたいと考えています。異なるアカウント間でデータリソースをストリーミングするにはどうすればよいですか?

簡単な説明

異なるアカウント間でのストリーミングを行うように、Kinesis Data Firehose と、それに依存関係がある Amazon Simple Storage Service (Amazon S3) や Amazon CloudWatch などを設定できます。ストリーミングデータ配信は、きめ細かなアクセス制御 (FGAC) が有効になっているかどうかにかかわらず、パブリックにアクセスできる OpenSearch Service クラスターに対して機能します。この記事では、両方のユースケースについて説明します。

OpenSearch Service クラスターにデータを送信するように Kinesis Data Firehose ストリームを設定するには、以下の手順を実行します。

1.    アカウント A で Amazon S3 バケットを作成します。

2.    アカウント A で CloudWatch ロググループとログストリームを作成します。

3.    アカウント A で Kinesis Data Firehose ロールとポリシーを作成します。

4.    アカウント A で Kinesis Data Firehose ロールがデータをストリーミングする先のアカウント B でパブリックアクセス可能な OpenSearch Service クラスターを作成する

5.    (オプション) きめ細かなアクセス制御 (FGAC) が有効になっている場合は、OpenSearch Dashboards にログインし、ロールマッピングを追加します。

注: OpenSearch Dashboards は Kibana の後継サービスです。

6.    アカウント A で Kinesis Data Firehose ロールの AWS Identity アクセス管理 (IAM) ロールポリシーを更新して、アカウント B にデータを送信します。

7.    アカウント A で Kinesis Data Firehose ストリームを作成します。

8.    OpenSearch Service クラスターへのクロスアカウントストリーミングをテストします。

解決方法

アカウント A で Amazon S3 バケットを作成する

アカウント A で S3 バケットを作成します。Amazon S3 バケットは Amazon リソースネーム (ARN) を生成します。

注意: この Amazon S3 バケットからのレコードを取得および保存するためのアクセス許可を、後ほど Kinesis Data Firehose に付与する際に、この ARN を完全な形で使用します。

アカウント A で CloudWatch ロググループとログストリームを作成する

CloudWatch ロググループを作成するには、次の手順を実行します。

1.    CloudWatch コンソールを開きます。

2.    ナビゲーション ペインで、[ロググループ] をクリックします。

3.    ロググループの作成 を選択します。

4.    [ロググループ名] に名前を入力します。

5.    [ロググループを作成] ボタンをクリックして、新しいロググループを保存します。

6.    新しく作成したロググループを検索し、選択します。このタスクを行うことで、ログストリームが作成可能になったことが確認されます。

次の手順を実行し、Amazon CloudWatch ログストリームを作成します。

1.    [ログストリームの作成] をクリックします。

2.    [ログストリーム名] に名前を入力します。

3.    [ログストリームの作成] をクリックします。このアクションにより、新しく作成したログストリームが保存されます。

重要: CloudWatch ロググループ名と CloudWatch ログストリーム名は、Kinesis Data Firehose ロールポリシーを作成する際に必要となります。

アカウント A で Kinesis Data Firehose ロールとポリシーを作成する

1.    AWS Identity and Access Management (IAM) コンソールを開きます。

2.    Kinesis Data Firehose に以下の操作を許可する IAM ポリシーを作成します。ストリームログを CloudWatch レコードに保存し、 Amazon S3 ストリームデータを OpenSearch Service クラスターに保存します。

例:

{
    "Version": "2012-10-17",
    "Statement": [
        {          
            "Effect": "Allow",    
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
    "Resource": [
                "<Bucket ARN>",
                "<Bucket ARN>/*"    
            ]
        },
        {    
            "Effect": "Allow",
                "Action": [
                "logs:PutLogEvents"
            ],           
                "Resource": [
                "arn:aws:logs:<region>:<account-id>:log-group:/aws/kinesisfirehose/<Firehose Name>:log-stream:*"           
            ]
        }
    ]
}

注: 後で OpenSearch Service クラスターポリシーにストリーミングするためのアクセス許可を追加します。ただし、最初にアカウント B でクラスターを作成する必要があります。

3.    作成したポリシーを保存します。

4.    [ロールの作成] をクリックします。

5.    新しく作成したポリシーを Kinesis Data Firehose ロールに追加します。

アカウント A で Kinesis Data Firehose ロールがデータをストリーミングする先のアカウント B でパブリックアクセス可能な OpenSearch サービスクラスターを作成する

1.    アカウント B でパブリックアクセス可能な OpenSearch Service クラスターを作成します。

2.    OpenSearch Service のドメイン ARN を記録します。後の手順では ARN が必要になります。

3.    クラスターのセキュリティ設定を設定します。

重要: アカウント A で Kinesis Data Firehose ロールが OpenSearch Service クラスターにストリーミングできるようにするには、OpenSearch Service のセキュリティ設定を設定する必要があります。

セキュリティ設定を設定するには、次の手順を実行します。

1.    OpenSearch Service で、[アクセスポリシー] に移動します。

2.    JSON 定義のアクセスポリシーを選択します。ポリシーには、次の許可が必要です。

{
  "Version": "2012-10-17",
  "Statement": [    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "*"
      },
      "Action": "es:*",
      "Resource": "<ES Domain ARN in Account B>/*",
      "Condition": {
        "IpAddress": {
          "aws:SourceIp": "<Your IP Address for OpenSearch Dashboards access>"
        }
      }
    },
    {
      "Effect": "Allow",
      "Principal": {
        "AWS": "<Firehose Role ARN in Account A>"
      },
      "Action": [
        "es:ESHttpPost",
        "es:ESHttpPut"
      ],
      "Resource": [
        "<ES Domain ARN in Account B>",
        "<ES Domain ARN in Account B>/*"
      ]
    },
    {    
      "Effect": "Allow",
      "Principal": {
        "AWS": "<Firehose Role ARN in Account A>"
      },
      "Action": "es:ESHttpGet",
      "Resource": [
        "<ES Domain ARN in Account B>/_all/_settings",
        "<ES Domain ARN in Account B>/_cluster/stats",
        "<ES Domain ARN in Account B>/index-name*/_mapping/type-name",
        "<ES Domain ARN in Account B>/roletest*/_mapping/roletest",
        "<ES Domain ARN in Account B>/_nodes",
        "<ES Domain ARN in Account B>/_nodes/stats",
        "<ES Domain ARN in Account B>/_nodes/*/stats",
        "<ES Domain ARN in Account B>/_stats",
        "<ES Domain ARN in Account B>/index-name*/_stats",
        "<ES Domain ARN in Account B>/roletest*/_stats"
      ]
    }
  ]
}

OpenSearch Service ポリシーにおける許可の詳細については、「OpenSearch Service 送信先へのクロスアカウント間の配信」を参照してください。

3.    (オプション) クラスターできめ細かなアクセス制御 (FGAC) が有効になっている場合は、OpenSearch Dashboards にログインし、ロールマッピングを追加します。ロールマッピングにより、Kinesis Data Firehose ロールが OpenSearch Service にリクエストを送信できるようになります。

(オプション) きめ細かなアクセス制御 (FGAC) が有効になっている場合は、OpenSearch Dashboards にログインし、ロールマッピングを追加する

OpenSearch Service クラスターできめ細かな制御が有効になっている場合は、OpenSearch Dashboards にログインし、Kinesis Data Firehose ロールにロールマッピングを追加する必要があります。ロールマッピングは、OpenSearch Service クラスターにストリーミングするための Kinesis Data Firehose ロールアクセスを提供します。

OpenSearch Dashboards にログインしてロールマッピングを追加するには、次の手順を実行します。

1.    [ダッシュボード] を開きます。

2.    [セキュリティ] タブを選択します。

3.    ロールを選択します。

4.    [all_access] ロールを選択します。

5.    [マッピングされたユーザー] タブを選択します。

6.    [Manage mapping] を選択します。

7.    [バックエンドロール] セクションで、Kinesis Data Firehose ロールを入力します。

8.    [マッピング] を選択します。

アカウント A で Kinesis Data Firehose ロールの IAM ロールポリシーを更新して、アカウント B にデータを送信する

アカウント A の Kinesis Data Firehose ロールからアカウント B の OpenSearch Service クラスターにデータを送信するには、次のようにして Kinesis Data Firehose ポリシーを更新します。

{
    "Version": "2012-10-17",
    "Statement": [
        {          
            "Effect": "Allow",    
            "Action": [
                "s3:AbortMultipartUpload",
                "s3:GetBucketLocation",
                "s3:GetObject",
                "s3:ListBucket",
                "s3:ListBucketMultipartUploads",
                "s3:PutObject",
                "s3:PutObjectAcl"
            ],
            "Resource": [
                "<Bucket ARN>",
                "<Bucket ARN>/*"    
            ]
        },
        {    
            "Effect": "Allow",
                "Action": [
                "logs:PutLogEvents"
            ],           
                "Resource": [
                "arn:aws:logs:<region>:<account-id>:log-group:/aws/kinesisfirehose/<Firehose Name>:log-stream:*"
            ]
        },    
         {
            "Effect": "Allow",
            "Action": [
                "es:ESHttpPost",
                "es:ESHttpPut",
                "es:DescribeDomain",    
                "es:DescribeDomains",
                "es:DescribeDomainConfig"
            ],
            "Resource": [
                "<Domain ARN in Account B>",
                "<Domain ARN in Account B>/*"
            ]
        },
        {        
            "Effect": "Allow",
            "Action": [
                "es:ESHttpGet"
            ],
            "Resource": [
                "<Domain ARN in Account B>/_all/_settings",
                "<Domain ARN in Account B>/_cluster/stats",
                "<Domain ARN in Account B>/index-name*/_mapping/superstore",
                "<Domain ARN in Account B>/_nodes",
                "<Domain ARN in Account B>/_nodes/stats",
                "<Domain ARN in Account B>/_nodes/*/stats",
                "<Domain ARN in Account B>/_stats",
                "<Domain ARN in Account B>/index-name*/_stats"
            ]
        }
    ]
}

OpenSearch Service クラスターへの Kinesis Data Firehose データ送信の詳細については、「Amazon OpenSearch サービスの送信先へのアクセス権を Kinesis Data Firehose に付与する」を参照してください。

アカウント A で Kinesis Data Firehose ストリームを作成する

OpenSearch Service クラスターへのクロスアカウントアクセスを用いて Kinesis Data Firehose ストリームを作成するには、AWS Command Line Interface (AWS CLI) を使用して設定します

AWS CLI が最新であることを確認してください。

aws --version

注: AWS CLI コマンドの実行時にエラーが発生した場合は、AWS CLI の最新バージョンを使用していることを確認してください

AWS CLI が更新されたら、次の内容で input.json という名前のファイルを作成します。

{
    "DeliveryStreamName": "<Firehose Name>",
    "DeliveryStreamType": "DirectPut",
    "ElasticsearchDestinationConfiguration": {
        "RoleARN": "",
        "ClusterEndpoint": "",
        "IndexName": "local",
        "TypeName": "TypeName",
        "IndexRotationPeriod": "OneDay",
        "BufferingHints": {
            "IntervalInSeconds": 60,
            "SizeInMBs": 50
        },
        "RetryOptions": {
            "DurationInSeconds": 60
        },
        "S3BackupMode": "FailedDocumentsOnly",
        "S3Configuration": {
            "RoleARN": "",
            "BucketARN": "",
            "Prefix": "",
            "BufferingHints": {
                "SizeInMBs": 128,
                "IntervalInSeconds": 128
            },
            "CompressionFormat": "UNCOMPRESSED",
            "CloudWatchLoggingOptions": {
                "Enabled": true,
                "LogGroupName": "/aws/kinesisfirehose/<Firehose Name>",
                "LogStreamName": "S3Delivery"
            }
        },
        "CloudWatchLoggingOptions": {
            "Enabled": true,
            "LogGroupName": "/aws/kinesisfirehose/<Firehose Name>",
            "LogStreamName": "ElasticsearchDelivery"
        }
    }
}

[ClusterEndpoint] 属性フィールドに、エンドポイントの値が正しく入力されていることを確認します。

注意: Type は Elasticsearch バージョン 7.x で非推奨となっています。Elasticsearch バージョン 7.x では、input.json ファイルから TypeName 属性を削除する必要があります。

その後、 input.json ファイルが置かれているのと同じディレクトリで、次の CLI コマンドを実行します。

aws firehose create-delivery-stream --cli-input-json file://input.json

このコマンド構文は、アカウント B の OpenSearch Service クラスターへの送信先を使用して、アカウント A で Kinesis Data Firehose ストリームを作成します。

OpenSearch Service クラスターへのクロスアカウントストリーミングをテストする

Kinesis Data Generator を使用して、アカウント A で Kinesis Data Firehose ストリームにレコードをストリーミングします。

Kinesis Data Generator (KDG) は 毎秒大量のレコードを生成します。この生産性レベルにより、レコード構造のマッピングを正しく検出するのに十分なデータポイントが OpenSearch Service に送られます。

Kinesis Data Generator で使用されるテンプレート構造は次のとおりです。

{
    "device_id": {{random.number(5)}},
    "device_owner": "{{name.firstName}}  {{name.lastName}}",
    "temperature": {{random.number(
        {
            "min":10,
            "max":150
        }
    )}},
    "timestamp": "{{date.now("DD/MMM/YYYY:HH:mm:ss Z")}}"
}

クロスアカウントストリーミングが成功したかどうかを確認するには、クラスターの [インデックス] タブの下のインデックスエントリを確認します。現在の日付で接頭辞「local」を使用してインデックス名があるかどうかをチェックします。OpenSearch Dashboards にレコードが存在するかどうかをチェックすることもできます。

注意:OpenSearch Service が正しいマッピングを判断するには、数分かかります。

Amazon OpenSearch Service は、Amazon Elasticsearch Service の後継サービスです。