Logstash：将两个日志合并为一个输出文档

Question

我已将 syslog 设置为使用以下过滤器将日志发送到 logstash：

output {
  elasticsearch 
  { hosts => ["localhost:9200"]
  document_id => "%{job_id}"   
}

}
filter {
    grok {
        overwrite => ["message"]
    }
    json {
     source => "message"
    }
}

我的一个应用程序的典型消息将具有初始状态和 job_id:

{"job_id": "xyz782", state: "processing", job_type: "something"}

几分钟后，另一个日志将具有相同的 log_id、不同的状态和处理时间：

{"job_id": "xyz782", state:"failed", processing_time: 12.345}

这些字段已正确加载，但创建了两个文档。我想要的是只为初始日志创建一个文档，第二个日志而不是更新第一个日志，这意味着更新的文档将具有以下字段：

{"job_id": "xyz782", state: "failed", job_type: "something", processing_time: 12.345}

正如您在我的 logstash conf 输出中看到的，我使用 job_id 作为文档 ID，但是，第二条消息似乎替换了第一条消息中的字段，但也删除了中的所有字段第一条消息不在第二条消息中，例如，第一条消息中存在的 job_type 字段不会出现在最终文档中。这可能与 json 两次来自同一字段 "message" 这一事实有关。是否有另一种方法可以将两个日志消息合并到 logstash 中的一个文档中？

Answer 1

您可以使用 aggregate 过滤器来执行此操作。聚合过滤器支持基于公共字段值将多个日志行聚合到一个事件中。在您的情况下，公共字段将是 job_id 字段。

然后我们需要另一个字段来检测应该聚合的第一个事件与第二个事件。在您的情况下，这将是 state 字段。

因此，您只需向现有的 Logstash 配置添加另一个过滤器，如下所示：

filter {
    ...your other filters

    if [state] == "processing" {
        aggregate {
            task_id => "%{job_id}"
        }
    } else if [state] == "failed" {
        aggregate {
            task_id => "%{job_id}"
            end_of_task => true
            timeout => 120
        }
    }
}

您可以根据作业的时长运行.

自由调整 timeout（以秒为单位）

Logstash：将两个日志合并为一个输出文档

Logstash: Merge two logs into one output document

elasticsearch

logstash