使用 logstash 处理多个 json 事件答案

【问题标题】：Processing multiple json events using logstash使用 logstash 处理多个 json 事件
【发布时间】：2018-09-26 10:56:39
【问题描述】：

我有以下配置：来自应用程序实例的日志使用 filebeat 和 logstash 转发到弹性搜索。

   Apps
+--------+
| +--------+
| | +--------+     +----------+    +----------+
| | |   +--- |     |          |    |          |
+ | |   |file| --> | logstash | -->| elastic  |
  + |   |beat|     |   (1)    |    | search   |
    +--------+     +----------+    +----------+
                         |               |
             (not avail) X               | (query)
                         V               |
                   +----------+          V
                   |          |      +------+
                   | logstash |<-----| Json |
                   |   (2)    |      | file |
                   +----------+      +------+

我想测试 logstash-2 中的日志处理，但我目前无法实现从 logstash-1 的转发。所以我尝试了以下方法：查询 elasticsearch 并检索文档的 _source 字段，我得到了一些这样的 json 文档：

{
 "@timestamp": <timestamp>,
 "@version": "1",
 "requestMethod": "PUT",
 "requestUri": "/api/endopoint",
 "servername": "myserver" 
 ....  many other fields
}
{
 "@timestamp": <timestamp>,
 "@version": "1",

}
... many other json objects

我的问题是，如何使用 logstash 从 elasticsearch 查询中处理这些 json 文档？

我尝试使用多行编解码器和 json 过滤器处理它们，但无法使其工作：这是一个尝试：

input {
  file {
    path => "events.json"
    sincedb_path => "/dev/null"
    start_position => beginning
    codec => multiline {
       pattern => "^\}"    #end of each json object
       what => "previous"
    }
  }
}

filter {
  json {
    source => "event"
  }
 }

 output {
  stdout{}
}

【问题讨论】：

标签： json logstash

【解决方案1】：

经过一些额外的研究，我意识到多行编解码器配置是错误的。我已经修复了，现在我在消息字段上有了整个事件。

input {
  file {
    path => "events.json"
    sincedb_path => "/dev/null"
    start_position => beginning
    codec => multiline {
        pattern => "^\}"    #end of each json object
        negate => true
        what => "next"
    }
  }
}

filter {
  json {
    source => "message"
  }

  mutate {
    remove_field => ["message"]
  } 
} 

output {
 stdout{}
}

【讨论】：