使用 Elasticsearch 进行日志去重

SRE 每天都会被来自嘈杂应用程序的海量日志所淹没。在他那部开创性的著作 《人月神话》 中，Frederick P. Brooks 曾说：“所有的程序员都是乐观主义者。”这种乐观情绪体现在软件工程师没有设置控制措施，以阻止其应用程序在异常故障情况下发送连续日志。在拥有集中式日志 Platform 的大型组织中，这些海量事件正被采集到日志 Platform 中，并占用了相当大的存储空间和处理计算资源。在人员方面，这会让 SRE 感到不知所措并遭受告警疲劳，因为他们被消息浪潮所淹没，就像这样：

 构建包括微服务和关键应用程序在内的软件的开发人员，有责任确保他们不会发送重复的日志事件，并确保他们发送的是正确级别且正确的日志事件。尽管如此，使用第三方解决方案或维护老旧服务等情况意味着我们无法始终保证已采取负责任的日志记录实践。即使我们删除了 这篇关于从传入日志事件中修剪字段的文章 中所涵盖的不必要字段，我们在存储大量重复事件方面仍然存在问题。在此，我们将讨论识别来自有问题服务的重复日志所面临的挑战，以及如何使用 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.

工具概述

在本篇博客中，我们将研究四种 Elastic 工具中可用的工具：

1. Logstash 是一款免费且开放的 ETL 管道工具，允许您在无数来源之间摄取、转换和输出数据，包括摄取到 Elasticsearch 以及从 Elasticsearch 输出。这些示例将用于展示 Elasticsearch 在生成带有和不带有特定 ID 的日志时表现出的不同行为。
2. Beats是一系列轻量型采集器，使我们能够将来自特定来源的事件不仅摄取到 Elasticsearch，还可以采集到其他输出，包括 Kafka、Redis 或 Logstash。3. 摄取管道允许对采集到 Elasticsearch 中的文档进行转换和扩充。这就像直接在 Elasticsearch 中运行 Logstash 的 filter 部分，而无需运行其他服务。可以通过 Stack Management（堆栈管理）> 摄取管道（摄取管道） 屏幕或通过 _ingest API 创建新管道，具体请参阅文档。4. Elastic Agent 是一个可以在您的主机上执行的单一代理，它使用各种受支持的集成，将来自多个服务和基础架构的日志、指标和安全数据发送到 Elasticsearch。无论重复的原因是什么，在 Elastic 生态系统中都有几种可能的处理方案。## 未指定 ID 的摄取 默认方法是忽略并摄取所有事件。当文档未指定 ID 时，Elasticsearch 将为接收到的每个文档自动生成一个新 ID。让我们看一个简单的示例，该示例可在 此 GitHub 存储库 中找到，它使用了一个 简单的 Express HTTP 服务器。服务器在运行时会公开一个终端，返回一条日志消息：

使用 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 示例，可以通过在 Elasticsearch 输出插件中指定 document_id 选项来在 Logstash 中指定文档 ID 的值，该选项将用于将事件摄取到 Elasticsearch 中：

这会将 _id 字段的值设置为 event.transaction_id 的值。在我们的案例中，这意味着新文档将在摄取时替换现有文档，因为两个文档的 _id 均为 1：

ID 的指定方式因工具而异，后续部分将对此进行更深入的讨论。### Beats 对于 JSON 文档（这是许多日志源的常见格式），如果您的事件确实有一个有用且有意义的 ID，可用作文档的唯一 ID 并防止条目重复，则可以使用 decode_json_fields 处理器或 json.document_ID输入设置 如文档中所建议。当我们的消息中的 JSON 字段内存在自然键时，此方法优于生成密钥。以下示例显示了这两项设置：

摄取管道

在这种情况下，可以使用 set 处理器 来设置 ID结合 copy_from 选项，将值从您的唯一字段传输到 Elasticsearch 的 @metadata._id属性:

Elastic Agent

Elastic Agent 采用类似的方法，您可以使用 copy_fields 处理器将值复制到 @metadata._id集成中的属性：

当 fail_on_error 设置为 true 时，将通过还原失败处理器所应用的更改来恢复到之前的状态。同时，当 ignore_missing 设置为 false 时，仅会对包含不存在字段的文档触发失败。## 自动生成的 ID 使用诸如指纹识别等技术对事件字段的子集生成唯一 ID。通过对一组字段进行哈希处理，可以生成一个唯一值；当该值匹配时，将在 Elasticsearch 摄取时更新原始文档。正如这篇关于使用 Logstash 处理重复文档的实用文章所述具体概述了如何配置 fingerprint filter plugin（指纹筛选插件），以使用指定的哈希算法生成 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 也有一个 add_id 处理器，可用于生成唯一 ID；如果未指定 target_field 属性，则默认为 @metadata._id：

或者，fingerprint 处理器 也可在 Elastic Agent 中使用，并可应用于任何包含高级配置部分（包括 processors 选项）的集成段。处理器逻辑如下所示：

以 Kafka 为例集成作为示例，上述处理器代码片段可应用于“从 Kafka 代理采集日志”的高级配置部分的处理器段中：

就像 Beats 一样，被哈希处理的值是由字段名称和字段值通过 | 分隔并连接而成的。例如 |field1|value1|field2|value2|。然而，就像在 Beats 中一样，且与 Logstash 不同，尽管支持相同的编码算法，但 method 值仍为小写。### 摄取管道 在此，我们将展示使用 _摄取 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 处理器