[新機能] Amazon OpenSearch Ingestionを試してみた

OpenSearch Ingestionを使ってサーバレスにALBのログをOpenSearch Serviceに連携してみました。
#AWS
#Amazon OpenSearch Service
中川翔太
2023.07.05
この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
こんにちは、札幌オフィスの中川です。
少し前に Amazon OpenSearch Service の新機能としてAmazon OpenSearch Ingestionが発表されておりました。
https://aws.amazon.com/jp/about-aws/whats-new/2023/04/amazon-opensearch-service-ingestion/
この記事では OpenSearch Ingestion の基本機能や動作確認した内容について紹介します。
 OpenSearch IngestionとはOpenSearch Ingestion は、OpenSearch Service ドメインや OpenSearch Serverless コレクションにリアルタイムでログ, メトリック, トレースデータを連携するサーバレスなデータコレクタです。

これまで OpenSearch にログを連携したいときに、Logstash や Fluentd といった収集ツールを動かす EC2 等を用意しなくてはいけないことがありました。OpenSearch Ingestion を使うことで、コンピュートリソースの管理をせず OpenSearch Service にデータ連携をできるようになります。
全体像（公式ドキュメント より抜粋）
OpenSearch Ingestion の機能は OpenSearch プロジェクトの一部としてオープンソースで公開されている Data Prepper によって提供されてるものがベースとなってます。本日時点では Data Prepper のバージョン 2 系がサポート対象です。

Data Prepper との違いとして、Ingestion では Data Prepper で利用可能な一部のプラグインに制約が課せられてます。Ingestion でサポートされているプラグインや制約の詳細については以下のページより確認できます。

Supported plugins and options for Amazon OpenSearch Ingestion pipelines
 構成要素Ingestion では以下のように YAML を定義してパイプラインを設定します。
version: "2"
s3-pipeline:
  source:
    s3:
      notification_type: "sqs"
      codec:
        newline: null
      sqs:
        queue_url: "https://sqs.us-west-2.amazonaws.com/{account-id}/ingestion-queue"
      compression: "none"
      aws:
        region: "us-west-2"
        sts_role_arn: "arn:aws:iam::{account-id}:role/pipeline-role"
  processor:
  - grok:
      match:
        log:
        - "%{COMMONAPACHELOG}"
  - date:
      destination: "@timestamp"
      from_time_received: true
  sink:
  - opensearch:
      hosts:["https://search-domain-endpoint.us-east-1.es.amazonaws.com"]
      index: "index_name"
      aws:
        sts_role_arn: "arn:aws:iam::{account-id}:role/pipeline-role"
        region: "us-east-1"
主な構成要素について紹介します。
バージョン（version:）

バージョンは使用する Data Prepper のメジャーバージョンを指定します。現時点で OpenSearch Ingestion でサポートされている Data Prepper のバージョンは 2.x であるため、version: "2" で問題ないです。
ソース（source: ）

ソースはパイプラインの入力コンポーネントです。HTTP 経由でイベントを受信するか、S3 など外部エンドポイントから読み取るかに応じて設定します。

サポートされたソース、各ソースのオプションについては OpenSearch ドキュメントより確認します。

参考：Pipelines / Sources
プロセッサー（processor: ）

プロセッサーは出力する前にレコードをフィルタリングや変換をすることができる中間処理ユニットです。プロセッサーは複数定義することができ、定義した順序で処理されます。また、プロセッサーを定義しない場合はソースで定義された形式のまま出力されます。

サポートされたプロセッサー、各プロセッサーの詳細については OpenSearch ドキュメントより確認します。

参考：Pipelines /Processors
シンク（sink:）

シンクはパイプラインの出力コンポーネントです。1 つ以上の宛先を定義します。

なお、OpenSearch Ingestion ではシンク対象としては OpenSearch Service と OpenSearch Serverless のみサポートしています。自身でホストした OpenSearch（EC2 や ECS にデプロイした OpenSearch）は対象にできませんので注意します。
 料金OpenSearch Ingestion は <strong>OpenSearch Compute Unit (OCU)</strong> の使用量に応じて課金されます。
OCU とは、データの取り込みや変換に必要なコンピューティングを測定する単位で、1 OCU あたり 2vCPU と 8GB のメモリで構成されています。パイプラインの処理要件とクライアントアプリケーションが生成する負荷に基づいて OCU は増減します。（最小 1 OCU、最大 96 OCU）
東京リージョンの料金は以下のとおりです。常時 1 OCU を消費している場合、月あたり 230USD が発生する見込みです。


課金対象
料金


OpenSearch Compute Unit (OCU)
0.326USD 1 OCU 1 時間あたり

パイプラインごとに OCU の最小数と最大数を設定してコストを抑えることができます。また、使用しない間においてはパイプラインを一時停止することで OCU は完全に消費されなくすることもできます。
参考：Amazon OpenSearch Service の料金
 リージョンOpenSearch Ingestion は OpenSearch Service が利用可能なすべてのリージョンで利用できます。
 やってみた以下の構成で S3 に取り込んだ ALB のログを OpenSearch に収集してみます。
ログが出力されると S3 イベント通知機能で SQS キューにメッセージを送ります。パイプラインはメッセージを受信すると、S3 からオブジェクトを取得して定義された処理をします。

S3 がソースの場合はパイプラインからオブジェクトを取得するため、パイプラインに設定する IAM ロールには S3 と SQS の両方の権限を付与します。
パイプラインは OpenSearch のドメイン同様に VPC 内に構築することもできますが、本記事ではパブリック環境で構築しています。

設定は以下の流れで進めます。
SQS キューを作成
S3 バケットでイベント通知を設定
OpenSearch ドメインを作成
パイプラインの IAM ロールを作成
OpenSearch アクセスポリシーを更新
OpenSearch パイプラインを作成
前提として ALB から S3 のログ出力までは設定済みであるものとして、ここでは割愛します。
 1. SQSキューを作成S3 イベント通知を受ける SQS キューを作成します。（キュー名：ingestion-queue）

標準キューでデフォルト設定のまま作成します。アクセスポリシーではログが保存されたバケットからメッセージを送れるように設定します。
アクセスポリシーの例
{
  "Version": "2012-10-17",
  "Statement":[
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "s3.amazonaws.com"
      },
      "Action": "SQS:SendMessage",
      "Resource": "arn:aws:sqs:ap-northeast-1:12345678XXXX:ingestion-queue",
      "Condition": {
        "StringEquals": {
          "aws:SourceAccount": "12345678XXXX"
        },
        "ArnLike": {
          "aws:SourceArn": "arn:aws:s3:*:*:XXX-log-bucket-12345678XXXX"
        }
      }
    }
 ]
}
 2. S3バケットでイベント通知を設定ログを保存したバケットでイベント通知を設定します。

対象とするイベントは「すべてのオブジェクト作成イベント」、送信先タイプとしては 1.で作成した SQS を選択します。
イベント通知の設定例

 3. OpenSearchドメインを作成OpenSearch として特別な機能は利用しません。検証であるため、チュートリアルと同じく以下の要件を満たすような構成で用意します。
OpenSearch バージョン 2.5
パブリックアクセスを利用
きめ細かいアクセス制御を利用しない
ドメインのアクセス制御の設定は一旦飛ばします。（以降のステップで設定します）
OpneSearch ドメインの設定例

 4. パイプラインのIAMロールを作成OpenSearch パイプラインが使用する IAM ロールを作成します。（ロール名：OpenSearchPipelineRole）

ロールのポリシーでは以下を設定します。
ログ保存した S3 に対する s3:Get:Object
イベント通知先の S3 に対する sqs:DeleteMessage, sqs:ReceiveMessage
OpenSearch へ HTTP リクエストを投げれるように es:ESHttp*, es:DescribeDomain
IAM ロールのポリシー例
{
    "Version": "2012-10-17",
    "Statement":[
        {
            "Effect": "Allow",
            "Action": "s3:GetObject",
            "Resource": "arn:aws:s3:::XXX-log-bucket-12345678XXXX/*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "sqs:DeleteMessage",
                "sqs:ReceiveMessage"
            ],
            "Resource": "arn:aws:sqs:ap-northeast-1:12345678XXXX:ingestion-queue"
        },
        {
            "Effect": "Allow",
            "Action": "es:DescribeDomain",
            "Resource": "arn:aws:es:ap-northeast-1:12345678XXXX:domain/logs-cluster"
        },
        {
            "Effect": "Allow",
            "Action": "es:ESHttp*",
            "Resource": "arn:aws:es:ap-northeast-1:12345678XXXX:domain/logs-cluster/*"
        }
   ]
}
信頼ポリシーの Principal では  osis-pipelines.amazonaws.com を設定します。
{
    "Version": "2012-10-17",
    "Statement":[
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "osis-pipelines.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
   ]
}
 5. OpenSearchアクセスポリシーを更新OpenSearch のアクセスポリシーを以下の内容で更新します。
OpenSearch パイプラインの IAM ロールから許可
（必要に応じて）OpenSearch ダッシュボードへの IP 制限
OpenSeach アクセスポリシー例
{
    "Version": "2012-10-17",
    "Statement":[
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "arn:aws:iam::12345678XXXX:role/OpenSearchPipelineRole"
            },
            "Action": "es:*",
            "Resource": "arn:aws:es:ap-northeast-1:12345678XXXX:domain/logs-cluster/*"
        },
        {
            "Effect": "Allow",
            "Principal": {
                "AWS": "*"
            },
            "Action": "es:*",
            "Resource": "arn:aws:es:ap-northeast-1:12345678XXXX:domain/logs-cluster/*",
            "Condition": {
                "IpAddress": {
                    "aws:SourceIp":[
                        "XX.XX.XX.XX/32"
                   ]
                }
            }
        }
   ]
}
 6. OpenSearchパイプラインを作成最後にパイプラインを作成します。コンソールの左メニューから[パイプライン]を選択し、[パイプラインを作成]をクリックします。
パイプライン名を入力します。キャパシティはデフォルトのままにします。
パイプラインの構成では最初からユースケースに応じたブループリントを利用できます。ここでは「AWS-AlbAccessLogS3Pipeline」を選択し、更新していきます。
ブループリントを更新してペーストします。ブループリントから更新した箇所は以下のとおりです。
source:
sqs_url:  で作成した SQS の URL に変更
sts_role_arn: を作成したパイプライン用 IAM ロールの ARN に変更

sink: 
hosts: を OpenSearch ドメインのエンドポイントに変更
sts_role_arn: を作成したパイプライン用 IAM ロールの ARN に変更
index: を alb に変更

パイプライン構成の設定例
version: "2"
alb-access-log-pipeline:
  source:
    s3:
      acknowledgments: true
      notification_type: "sqs"
      compression: gzip
      codec:
        newline:
      sqs:
        queue_url: "https://sqs.ap-northeast-1.amazonaws.com/12345678XXXX/ingestion-queue"
        maximum_messages: 10
        visibility_timeout: "60s"
      aws:
        region: "ap-northeast-1"
        sts_role_arn: "arn:aws:iam::12345678XXXX:role/OpenSearchPipelineRole"
  processor:
    - grok:
        match:
          message:["%{ALB_ACCESS_LOG}"]
    - date:
        destination: "@timestamp"
        from_time_received: true
    - delete_entries:
        with_keys:["message", "s3"]
  sink:
    - opensearch:
        hosts:["https://search-logs-cluster-cuvzdzhpjodk5zkoqeghyqi77y.ap-northeast-1.es.amazonaws.com"]
        aws:
          sts_role_arn: "arn:aws:iam::12345678XXXX:role/OpenSearchPipelineRole"
          region: "ap-northeast-1"
        index: "alb"
設定ファイルを貼り付けたら[検証を実行]をすることで、IAM ロールの妥当性など確認できます。
ネットワーク設定は[パブリックアクセス]を選択し、その他はデフォルトのまま[次へ]をクリックします。
確認画面で[パイプラインを作成]をクリックします。

数分するとパイプラインのステータスが[アクティブ]になります。以上で設定は完了です。
 動作確認動作確認します。

適当に ALB へアクセスしてアクセスログを生成しておきます。
OpenSearch 上でインデックスパターンを設定をします。今回は alb で設定しています。
左メニューから[Discover]を選択すると、ログが正常に取り込まれていることを確認できました。
 さいごにOpenSearch Service にデータ連携できる機能として OpenSearch Ingestion が登場したので試してみました。

これまで 3rd パーティのログ収集ツールでデータ連携しようとすると、自分たちでコンピュートリソースを管理しなくてはならず、M/W や OS のセキュリティパッチやバグ対応など運用でコストがかかっていました。OpenSearch Ingestion はサーバレスなので運用面でのコストが減り、また定義した範囲内でキャパシティは自動スケールするので使いやすいです。一般的なユースケースを想定したブループリントも多く用意されており、使い始めやすさも感じました。
実は先日の下記ブログで Ingestion のパイプラインを使って、Securiy Lake のイベントを連携する方法が紹介されていました。
https://dev.classmethod.jp/articles/opensearch-ingestion-security-lake/
今後は他の AWS サービスとの連携も増え、ますます活用しやすくなることを期待したいです。
 参考Amazon OpenSearch Ingestion
Top strategies for high volume tracing with Amazon OpenSearch Ingestion
Sending data to Amazon OpenSearch Ingestion pipelines