記事公開時点（2025年8月末頃）では Streams の機能は Tech Preview かつ Elastic Cloud Serverless - Elatsic for Observability のみで提供している機能です。詳細はドキュメントをご確認ください。

Streams とは

Streams は Data streams を扱いやすくする事を目的とした機能です。フィールド抽出ロジックの作成や保持期間の編集が Kibana の画面から簡単に操作できるようになります。

Elasticsearch にはインテグレーションという、データソース毎に定義されたフィールド情報や、抽出ロジックを活用してデータを収集する仕組みはありました。必要に応じてインテグレーションをカスタマイズしながら、Elasticsearch に情報を収集する事が一般的でした。インテグレーションのないデータソースは、イチからフィールド抽出ロジックを作成する必要がありました。

Streams では生成AI を活用してフィールド抽出ロジックを作成する事で、インテグレーションのないデータソースも簡単に取り込めるようになります。

試した環境

Elasitc Cloud Serverless の Elastic for Observability のプロジェクト
生成AI は EIS を活用
Ubuntu 24.04 の VM上にリリースした nginx

今回は少しカスタマイズされた nginx のログを取り込んでみました。デフォルトのフォーマットでは表示しないパラメーターを出力するようにしています。ログの取り込みは Elastic Agent の Custom logs (Filestream) を利用して、生ログを message フィールドに取り込んでいます。

nginx.conf

log_format custom_access_log '$remote_addr - $remote_user [$time_local] '
                             '"$request" $status $body_bytes_sent '
                             '"$http_referer" "$http_user_agent" "$gzip_ratio" '
                             'rt="$request_time" uct="$upstream_connect_time" '
                             'uht="$upstream_header_time" urt="$upstream_response_time"';

発行されている nginx のログはこんな感じです。

access.log

118.156.252.109 - - [25/Aug/2025:02:02:01 +0000] "GET / HTTP/1.1" 200 615 "-" "curl/8.7.1" "-" rt="0.000" uct="-" uht="-" urt="-"
118.156.252.109 - - [25/Aug/2025:02:02:01 +0000] "GET /index.html HTTP/1.1" 404 162 "-" "curl/8.7.1" "-" rt="0.000" uct="-" uht="-" urt="-"
118.156.252.109 - - [25/Aug/2025:02:02:01 +0000] "GET /hoge.html HTTP/1.1" 404 162 "-" "curl/8.7.1" "-" rt="0.000" uct="-" uht="-" urt="-"
...

Streams の操作

Elastic for Observability のプロジェクトにログインし、左側のナビゲーション・メニューにある、Streams をクリックします。

Data streams の一覧が表示されるので、取り込んでいるログが入っている Data stream を選択します。（今回は logs-filestream.nginx-default）

直近の Data stream の取り込みであったり、保持期間の情報が表示されています。
ここで Edit data retention をクリックすると、保持日数が設定できます（Hot-Warm-Cold のようなデータティアの移動は未対応）

上部の Processing をクリックすると、画面左側に処理するプロセッサを定義する画面、右側にはサンプルとして利用するデータが一覧化されています。+ Add processor をクリックします。

プロセッサの作成するウィンドウが表示されます。デフォルトでは、GROK プロセッサが選択されています。Generate pattern というボタンをクリックすると、サンプルデータを元に、生成AIを活用した GROK パターンの作成が開始されます。

生成AIの作成した GROK パターンが表示されます。今回は HTTP のアクセスログとして認識されたようです。（何度か試した結果、 Apache のログとして認識されたりもしました）

Accept をクリックすると、パース結果が視覚的に表示されます。Create processor をクリックすると、そのプロセッサは保存されます。

実際にプロセッサが適応された結果もプレビューできます。問題がなければ Save changes をクリックする事で、設定を適応できます。

プロセッサの設定後、 Discovery からログを確認してみると、フィールド抽出された状態のドキュメントが確認できました。

感じたこと

これまでは Elasticsearch 内でカスタムログを処理する場合、インジェストパイプラインを設定する必要がありました。GROK パターンもなんらかの方法で用意する必要がありました。そして、適応するにはインデックスの設定を変更する必要があります。
保持期間の設定を行う場合も ILM で削除するポリシーを作成し、インデックスにあてる必要があります。
それらの操作を Kibana の画面ひとつで抽出処理やポリシーの作成まで適応できれば、運用も楽になりそうです。

執筆時点（2025年8月）では Serverless の Observability のみでしか利用できないため、今後どこまで使えるようになるのか気になるところです。

参考までに、試すにあたって AWS Singapore リージョンにプロジェクトをデプロイして、ログは数千行程度取り込み、Elastic でマネージしている生成AIモデルに対しては数十回?程度 GROKパターンの生成リクエストを叩いたと思うのですが、1.1ECU くらいかかったようです。当然、長期的に運用する場合は他のコストもかかってきますので注意は必要です。（実際のコストは使い方次第で変わりますので、参考程度にお考えください）料金体系が気になる方は Elastic Observability Serverless のコストのページをこちらからご確認ください。

最後に

Elastic はデータの取り込みと活用をスムーズにするための機能が増えています。

Elasticsearch の構造を理解しないと使いこなせなかったインジェストパイプラインや ILM の仕組みをよりシンプルな操作で設定できるようになれば、上級者でなくても Elasticsearch を効率的に使えるようになると思います。

試してみたいという方は Elastic Cloud Serverless のトライアルも可能ですので、ぜひお試しください。