Elasticsearch 5.0の目玉機能の一つである、Ingest Nodeのセッションに参加しました。
Ingest Nodeとは、Logstashが持つ加工機能をElasticsearchに移植したものです。
セッションは少し小さめの部屋で行われたのですが、立ち見まで出る満席っぷり。
かなり注目の機能ですし、実際、インパクトのあるデモも見られました。今日イチのセッションですね。
Ingest Nodeのユースケースとして2種類紹介されました。
- FilebeatからLogstash経由せずに、直接Elasticsearchにデータを送り込む
- Reindex APIと組み合わせて使う
前者は、これまでFilebeat（転送） → Logstash（加工） → Elasticsearch（保存）としていたところを、
Filebeat（転送） → Elasticsearch（加工＆保存）とできるようなる、というパターンです。
また後者は、Elasticsearchに投入済みのデータを取り出しながら加工し、別のindexに投入するというパターンです。
いずれもありそうなシチュエーションですね。
なおElasticsearch 5.0では、デフォルトでIngest Nodeが有効になるように設定されているとのことです。
Ingest Nodeで行われる個別の処理はプロセッサと呼ばれ、それらを組み合わせたものをパイプラインと呼びます。
パイプラインの作成、更新、取得、削除はREST API経由で行います。
PUT _ingest/pipeline/pipeline-name
GET _ingest/pipeline/pipeline-name
GET _ingest/pipeline/*
DELETE _ingest/pipeline/pipeline-name
PUTするJSONはこんな感じ。
{
"description": "mysql pipeline",
"processors": [
{
"grok": {
"field": "message",
"pattern": "..."
}
},
{
"remove": {
"field": "message"
}
}
]
}Logstashの設定は独自の文法で行うものでしたが、Ingest Nodeのパイプラインは普通にJSONで記述するようですね。
このパイプライン用に利用できるプロセッサは、mutate、grok、geoipなど、LogstashにあるFilterはだいたいあります。
というかむしろIngest NodeのプロセッサはLogstashのFilterと相互に互換性があるそうです。こういうものって得てして別物になりがちなので、これは嬉しいところですね。
エラーが発生した時の処理は、プロセッサの on_failure で記述できます。
次のJSONは、bytesという名前のフィールドを数値（integer）に変換し損ねた場合には、値を0にするという例です。
{
"convert": {
"field": "bytes",
"type": "integer",
"on_failure": [
{
"set": {
"field": "bytes",
"value": 0
}
}
]
}
}なるほど、違和感ないですね。
またテストのためにSimulate APIが用意されており、実際にindexを行わずに、ドキュメントを試しに投入するテストを行えます。
最初からこういうものが用意されているのは、ありがたいところ。
Simulate APIは、こんな形をしています。
POST _ingest/pipeline/pipeline-name/_simulate
POST _ingest/pipeline/_simulate
POST _ingest/pipeline/_simulate?verbose
verboseオプションをつけることで、最終的な結果だけでなく、パイプラインの途中の処理結果を確認することができるようになります。
パフォーマンスは？
パフォーマンスは、平たく言えば、いつものマジックワード「depends on」です。プロセッサの内容、プロセッサの数、追加/削除するフィールドの数、などなど。
ベンチマーク結果も示されていましたが、あまりピンと来ませんでした。
ただどうあれ、このような加工処理がindexに影響を与えることは間違いありません。それでパフォーマンス問題が起きるようなことを想定して、
通常のノードとは別に、Ingest専用のノード（dedicated ingest nodes）を作るというパターンも紹介されていました。
masterノードからingest nodeに行って来いする形ですね。
デモがやばい。
パイプラインを実際に作成するデモも行われました。って、このデモが想像以上でした。
Elasticsearch 5.0 + Kibana 4.4の組み合わせでしょうか、Kibana上のindicesのsettingsに「Tail a File」というオプションが追加されており、
ここからパイプラインを作成したり、ファイルを流し込んだりできるようでした。
デモでは、実際にTail a Fileが行われました。まずはテキストエリアに、読み込み対象となるファイルのサンプルを適当に貼り付けます。
そうすると、パイプラインの処理結果としてどういうJSONが得られるかというサンプルが表示されます。
この例ではまだ何のプロセッサも設定していないため message というフィールドに、貼り付けたログの1行が表示されるだけになっています。
その後、プロセッサを追加してパターンなどを設定すると、どんどんOutputの形が変わっていきます。
この写真の例では（すごく見えにくいですが）最終結果のOutputが、source、message、contentの3フィールドが出力されることや、プロセッサの結果として、sourceとcontentが取得できることが分かります。
実際にファイルのサンプルを使いながら、トライ＆エラーで設定を変更できるのは、本当にありがたい機能です。Logstashにはこの機能がないために、
ちょっと修正してはLogstashプロセスを再起動して、試しにデータを流し込んでみて、また修正してはプロセスを再起動して・・・という作業を繰り返さざるを得ませんでした。
（Logstash 5.0では設定が自動でリロードされるようになり、少しは楽になりますけども）
このパイプラインの作成の手軽さを考えると、もうLogstashなんて使っていられないな・・・と思うほどです。
まとめと、今後？
という感じで、Ingest Nodeをとても使いたくなるセッションでした。むしろこれから先、Logstashがどういう位置づけになるのか、微妙だと思います。
まず、小さな規模であれば、Logstashを完全に省いて、Filebeat（や他のBeats）とElasticsearchだけで運用することができるようになるのは間違いないでしょう。
問題は、規模が大きくなってきた時。今の標準的な構成としてはKafkaを挟んだ形になります。
Filebeat → Logstash（加工） → Kafka（キュー） → Logstash（転送） → Elasticsearch（保存）
※それとは別に、Kafka or Logstash からファイルサーバに転送
これが、FilebeatがKafka出力に対応し、ElasticsearchがIngest Nodeを持つことになると、こうなります？
Filebeat → Kafka（キュー） → Logstash（転送） → Elasticsearch（加工＆保存）
※それとは別に、Kafka or Logstash からファイルサーバに転送
Logstashは転送の役割だけで、むしろKafkaからElasticsearchが直接繋がれば、Logstashは完全に要らなくなるのに・・・と思ってしまいます。
この辺りがどうなるのか、今後の動向も含めて数ヶ月ぐらいはウォッチしたいと思います。
いやー、Ingest Node、便利ですよ！
