具有跨多行任务 ID 的 Logstash 聚合插件答案

【问题标题】：Logstash aggregate plugin with task id across multiple line具有跨多行任务 ID 的 Logstash 聚合插件
【发布时间】：2019-04-27 14:05:42
【问题描述】：

我正在尝试根据进程 ID 聚合事件。进程 ID 不是按顺序排列的，并且跨越多行。下面是一个这样的示例 -

2019 Apr 23 14:01:44:870 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526635 Main Process has started^M
2019 Apr 23 14:01:44:870 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526636 Main Process has started^M
2019 Apr 23 14:01:44:878 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526636 Ending ODS query Process to create enterprise message^M
2019 Apr 23 14:01:44:882 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526635 Send Message Process has finished at  :  1556028104882 for flight 1206 departing on 2019-04-24 from EWR to CLT of type Leg Level Message^M
2019 Apr 23 14:01:44:882 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526635 Send Message Process has finished at  :  1556028104882 for flight 1206 departing on 2019-04-24 from EWR to CLT of type Flight Level Message^M
2019 Apr 23 14:01:44:882 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526636 Send Message Process has begun at  :  1556028104882 for flight 1196 departing on 2019-04-24 from CUN to ORD of type Leg Level Message^M
2019 Apr 23 14:01:44:883 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526636 Send Message Process has begun at  :  1556028104882 for flight 1196 departing on 2019-04-24 from CUN to ORD of type Flight Level Message^M
2019 Apr 23 14:01:44:882 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526635 Main Process is ending for flight 1206 departing on 2019-04-24^M
2019 Apr 23 14:01:44:882 GMT +0000 BW.EnterpriseFlifoFormatter-EnterpriseFlifoFormatter USER [BW-User] - Interface Name : EFF: Process ID :674526636 Main Process is ending for flight 1206 departing on 2019-04-24^M

所以进程ID不是按顺序排列的，我们如何使用logstash聚合过滤器将属于同一进程ID的所有消息关联并作为单个事件发送

【问题讨论】：

感谢您的提问，我发现了这个插件elastic.co/guide/en/logstash/current/… 它应该适合您！我在下面留下我的答案并进行编辑。

标签： logstash logstash-grok

【解决方案1】：

我建议您检查此plugin，它跟踪一对开始/结束事件并使用它们的时间戳来计算它们之间的经过时间，因此根据您需要的 id，它可以识别“开始”事件并等待其相关的“结束”事件。这是使用它的简单example。

调整此插件以执行您需要的操作可能很有用，因为它已经处理超时、等待基于特定字段的事件等。

但您也需要注意以下几点：

知道什么时候停止，什么时候需要累积事件？
可以在内存中累积大量事件吗？
什么时候超时？

如果您要摄取大量数据，在我看来，Logstash 似乎不太适合这种需求。考虑使用其他工具聚合事件，可能是一个 apache spark 日常工作，它使用弹性搜索作为后端聚合您的事件？

编辑：我迅速检查了插件的source，我的建议基本上就是将您需要的所有信息添加到“统一事件”中。

假设您需要从所有事件中获取“my_field”的内容。你可以更新第 167 行：

return add_elapsed_info(event, elapsed, unique_id, start_event.get("@timestamp"), start_event.get("my_field"))

最后编辑：这个插件应该适合你 https://www.elastic.co/guide/en/logstash/current/plugins-filters-aggregate.html

【讨论】：