こんにちは、札幌オフィスの中川です。
少し前にAmazon OpenSearch Serviceの新機能としてAmazon OpenSearch 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は OpenSearch Compute Unit (OCU) の使用量に応じて課金されます。
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のイベントを連携する方法が紹介されていました。
今後は他のAWSサービスとの連携も増え、ますます活用しやすくなることを期待したいです。
0
