Elasticsearchによるログの重複排除

SREは毎日、ノイズの多いアプリケーションからの大量のログに追われています。彼の独創的な著作『The Mythical Man Month』の中で、フレデリック・P・ブルックスは、「すべてのプログラマーは楽観主義者である」と述べました。この楽観主義は、ソフトウェアエンジニアが、例外的な障害状況下でアプリケーションが継続的にログを送信するのを停止するための制御を組み込まないという形で現れています。一元化されたログPlatformを持つ大規模な組織では、このイベントの洪水がログPlatformに投入され、かなりのストレージ容量と処理用コンピューティングリソースを消費しています。人材の面では、SREはメッセージの波に飲み込まれ、圧倒されたような気分になり、アラート疲れに苦しんでいます。ちょうどこのような感じです：

][1]

 マイクロサービスや主要なアプリケーションを含むソフトウェアを構築する開発者は、重複したログイベントを送信しないようにし、適切なレベルで正しいログイベントを送信するようにする責任があります。それにもかかわらず、サードパーティソリューションの使用や老朽化したサービスの保守といった状況では、責任あるロギングプラクティスが適用されていることを常に保証できるとは限りません。受信ログイベントからフィールドを整理するに関するこの記事で説明されているように、不要なフィールドを削除したとしても、依然として、大量の重複イベントを格納するという問題が残っています。ここでは、問題のあるサービスからの重複ログを特定する際の課題と、Elastic Beats、Logstash、およびElastic Agentを使用してデータを重複排除する方法について説明します。## What is a duplicate log entry? Before diving into the various ways of preventing these duplicates from making it into your logging platform, we need to understand what a duplicate is. In my prior life as a software engineer, I was responsible for developing and maintaining an ecosystem of vast microservices. Some had considered retry logic that, after some time, would shut the service down gracefully and trigger appropriate alerts. However, not all services are built to gracefully handle these cases. Service misconfiguration can also contribute to event duplication. Inadvertently changing the production log level from WARN to TRACE can lead to more aggressive event volumes that have to be handled by the logging platform.

Elasticsearch automatically generates a unique ID for each document ingested unless a document contains an _id field on ingestion. Therefore, if your service is sending repeated alerts, you run the risk of having the same event stored as multiple documents with different IDs. Another cause can be due to retry mechanisms for the tools used for log collection. A notable example is for Filebeat, where a lost connection or shutdown can cause the retry mechanism of Filebeat to resend an event until the output acknowledges receipt of the event.

ツール概要

このブログでは、4つのElasticツールで利用可能なツールについて検討します：

1. Logstashは、無料のオープンソースETLパイプラインツールであり、Elasticsearchへのインジェストや出力を含む、無数のソース間でのデータの取り込み、変換、出力を可能にします。これらの例は、特定のIDの有無にかかわらずログを生成する際のElasticsearchの異なる動作を示すために使用されます。
2. Beatsは、特定のソースからのイベントをElasticsearchだけでなく、Kafka、Redis、Logstashなどの他の出力先にも取り込みできる軽量シッパーファミリーです。3. 取り込みパイプラインは、Elasticsearchに投入されたドキュメントに対して変換とエンリッチメントを適用できるようにします。これは、別のサービスを実行する必要なしに、Logstashのfilter部分をElasticsearch内で直接実行するようなものです。新しいパイプラインは、Stack Management > 取り込みパイプライン画面内、または _ingest API経由で作成できます。ドキュメントで解説されている通りです。4. Elastic Agentは、ホスト上で実行し、サポートされているさまざまな統合機能を使用して、複数のサービスやインフラストラクチャーからログ、メトリック、セキュリティデータをElasticsearchに送信できる単一のエージェントです。重複の理由にかかわらず、Elasticエコシステムにはいくつかの対処法があります。## IDを指定しないインジェスト デフォルトのアプローチは、すべてのイベントを無視して取り込むことです。ドキュメントでIDが指定されていない場合、Elasticsearchは受信した各ドキュメントに対して新しいIDを自動生成します。こちらのGitHubリポジトリで公開されている、シンプルなExpress HTTPサーバーを使用した簡単な例を見てみましょう。サーバーを実行すると、単一のログメッセージを返す単一のエンドポイントが公開されます：

[1]：https://images.contentstack.io/v3/assets/bltefdd0b53724fa2ce/bltcb726be86c9c8601/6543831208cc0104077cd7d4/gif.gif

Logstashを使用して、60秒ごとにエンドポイント http://locahost:3000/ をポーリングし、その結果をElasticsearchに送信できます。logstash.conf下記のようになります。

Logstashは各イベントをプッシュします。イベントにIDがない場合、Elasticsearchは新しい_idを生成します。各ドキュメントの固有の識別子として機能するフィールド：

この動作はBeats、取り込みパイプライン、およびElastic Agentで一貫しており、追加の構成なしで受信したすべてのイベントを送信します。## BYO-ID 既存のIDを使用して各イベントに固有のIDを指定すると、前のセクションで説明したElasticsearchのID生成ステップをバイパスできます。この属性が既に存在するドキュメントを取り込むと、ElasticsearchはそのIDを持つドキュメントがインデックス内に存在するかどうかを確認し、存在する場合はそのドキュメントを更新します。同じ_idを持つドキュメントが存在するかどうかを確認するためにインデックスを検索する必要があるため、これによってオーバーヘッドが発生します。上記のLogstashの例を拡張すると、LogstashでドキュメントIDの値を指定するには、Elasticsearch出力プラグインで document_id オプションを指定します。これは、イベントをElasticsearchに取り込むために使用されます。

これにより、_idフィールドの値がevent.transaction_idの値に設定されます。今回のケースでは、両方のドキュメントの_idが1であるため、インジェスト時に新しいドキュメントが既存のドキュメントを置き換えることになります。

IDの指定方法はツールによって異なります。これについては、以降のセクションで詳しく説明します。### Beats 多くのログソースで一般的なフォーマットであるJSONドキュメントの場合、イベントにドキュメントのユニークIDとして使用でき、重複エントリを防御できる有用で意味のあるIDが含まれているなら、decode_json_fieldsプロセッサーまたはjson.document_IDを使用します。ドキュメントで推奨されているインプット設定。メッセージ内のJSONフィールドにナチュラルキーが存在する場合は、キーを生成するよりもこのアプローチが推奨されます。両方の設定を以下の例に示します。

取り込みパイプライン

この場合、IDはsetプロセッサを使用して設定できます。copy_fromオプションと組み合わせて、一意のフィールドからElasticsearchの@metadata._idに値を転送します属性:

Elastic Agent

Elastic Agentも同様のアプローチを採用しており、copy_fieldsプロセッサを使用して値を@metadata._idにコピーできます。インテグレーション内の属性：

fail_on_error 設定が true の場合、失敗したプロセッサによって適用された変更が元に戻され、以前の状態に復帰します。一方、ignore_missing が false に設定されている場合、存在しないフィールドを持つドキュメントに対してのみエラーがトリガーされます。## 自動生成ID イベントフィールドのサブセットに対してフィンガープリントなどの手法を使用し、一意のIDを生成します。一連のフィールドをハッシュ化することで一意の値が生成されます。この値が一致すると、Elasticsearchへの取り込み時に元のドキュメントが更新されます。Logstashでの重複処理に関するこちらの便利な記事具体的に概説されているように、フィンガープリントフィルター プラグインを設定して、指定されたハッシュアルゴリズムでIDを生成し、@metadata.fingerprint フィールドに格納することができます：

指定がない場合、デフォルトのハッシュアルゴリズム SHA256 が使用され、|event.start_date|start_date_value|event.data_set|data_set_value|message|message_value| の組み合わせがハッシュ化されます。他の許可されたアルゴリズムオプションを使用する場合は、method オプションを使用して指定できます。その結果、Elasticsearchは生成された _id に一致するドキュメントを更新します：

イベントに単一の有意義な識別フィールドがない場合、ID生成に伴う処理オーバーヘッドや、異なるイベントが同じ生成ハッシュに解決されてしまう衝突の可能性を許容できるのであれば、これが有用な選択肢となる可能性があります。以降のセクションで説明するように、他のツールでも同様の機能を利用できます。### Beats BeatsおよびElastic Agent用のadd_idプロセッサを使用すると、Elasticsearch互換の固有IDを生成できます。デフォルトでは、この値は @metadata._id に格納されます。ElasticsearchドキュメントのIDフィールドであるフィールド。

あるいは、fingerprintプロセッサ|演算子で区切られた、指定されたフィールド名と値のペアを連結したハッシュ値を生成します。

上記の例では、デフォルトのハッシュアルゴリズム sha256 を使用して、|event.start_date|start_date_value|event.data_set|data_set_value|message|message_value| の組み合わせをハッシュ化します。他の許可されたアルゴリズムオプションを使用する場合は、method オプションを使用して指定できます。エラー処理も重要な検討事項であり、ignore_missing オプションがその助けとなります。例：event.start_date がフィールドが特定のドキュメントに存在しない場合、ignore_missing が false に設定されているとエラーが発生します。ignore_missing が明示的に設定されていない場合のデフォルトの実装ですが、値を true に指定してエラーを無視するのが一般的です。### Elastic Agent Beats と同様に、Elastic Agent には一意の ID を生成するために使用できる add_id プロセッサ があります。target_field 属性が指定されていない場合、デフォルトで @metadata._id が使用されます。

あるいは、fingerprintプロセッサーもElastic Agentで利用可能であり、processorsオプションを含む高度な構成セクションを持つあらゆるインテグレーションセグメントに適用できます。プロセッサーのロジックは次のようになります。

Kafka を例にとると統合を例に挙げると、上記のプロセッサスニペットは、Collect logs from Kafka brokers（Kafkaブローカーからのログ収集）の高度な構成セクションにあるプロセッサセグメントに適用できます。

 Beatsと同様に、ハッシュ化される値は、フィールド名とフィールド値を | で区切って連結したものとして構成されます。例：|field1|value1|field2|value2|。ただし、Beatsと同様でLogstashとは異なり、同じエンコードアルゴリズムをサポートしているにもかかわらず、method 値は小文字になっています。### 取り込みパイプライン ここでは、_ingest API を使用して fingerprint プロセッサ を含むパイプラインを作成するサンプルリクエストを示します。以下の構成が当社のBeatsプロセッサと類似している点にご注目ください：

イベントのアグリゲーション

使用するツールがサポートしている場合、共通のフィールドに基づいてイベントをアグリゲーションすることも選択肢の一つです。イベントのアグリゲーションにはトレードオフが伴います。ツールがイベントを即座に出力へ転送するのではなく、アグリゲーションを実行するために複数のイベントをメモリ内に保持する必要があるためです。このため、Elasticエコシステム内でイベントのアグリゲーションをサポートしているツールはLogstashのみです。Logstashでアグリゲーションベースのアプローチを実装するには、aggregateプラグインを使用します。今回のケースでは、重複を区別するための特定の終了イベントが送信される可能性は低いため、以下の例のようにtimeoutを指定してバッチ処理を制御する必要があります。

上記の例では、600秒（10分）後にイベントを送信し、集約イベントであることを示すために error_count 属性と has_multiple_occurrences 属性をイベントに追加します。push_map_as_event_on_timeout オプションを使用すると、タイムアウトのたびにアグリゲーション結果が確実にプッシュされるため、アラートの量を削減できます。データのタイムアウトを決定する際は、ボリュームを考慮し、可能な限り低いタイムアウト値を選択してください。Logstashはタイムアウトが経過して集約イベントがプッシュされるまで、イベントをメモリ内に保持するためです。

Conclusions

Log volume spikes can quickly overwhelm logging platforms and SRE engineers looking to maintain reliable applications. We have discussed several approaches to handling duplicate events using Elastic Beats, Logstash (which are available in this GitHub repository), and Elastic Agent.

When generating IDs via a hashing algorithm using fingerprint processors, or performing aggregates, consider the attributes used carefully to balance preventing a flood and obfuscating legitimate streams pointing to a large-scale problem in your ecosystem. Both approaches have an overhead, either in terms of processing to generate the ID, or memory overhead to store the documents eligible to aggregate.

Selecting an option really depends on the events you consider duplicates and the performance trade-offs. As discussed, when you specify an ID Elasticsearch needs to check for the existence of a document matching that ID before adding the document to the index. This results in a slight delay in ingestion to perform the _id existence check.

Using hashing algorithms to generate the ID adds additional processing time as the ID needs to be generated for each event before it is compared and potentially ingested. Choosing to not specify an ID bypasses this check as Elastic will generate the ID for you, but will result in all events being stored which increases your storage footprint.

Dropping full events is a legitimate practice not covered in this piece. If you want to drop log entries to reduce your volume check out this piece on pruning fields from incoming log events.

If your favorite way to deduplicate events is not listed here, do let us know!

リソース

1. Elastic Beats2. Filebeat | Fingerprintプロセッサー
2. Filebeat | JSONフィールドをデコードするプロセッサー
3. Filebeat | Filebeatの重複排除
4. Logstash6. Logstash | Fingerprintプラグイン
5. Logstash | Aggregateプラグイン
6. Elastic Agent
7. Elastic Agent | Fingerprintプロセッサー
8. 取り込みパイプライン
9. Elasticsearch | 取り込みパイプラインのFingerprintプロセッサー