使用logstash中的grok解析多行JSON

Question

I've got a JSON of the format:

{
    "SOURCE":"Source A",
    "Model":"ModelABC",
    "Qty":"3"
}

我正在尝试使用logstash解析这个JSON。基本上，我希望logstash的输出是一个键值对列表，以便我可以使用kibana进行分析。我认为这可以直接完成。通过大量阅读，我了解到必须使用grok插件（我仍然不确定json插件是用来做什么的）。但是我无法获取包含所有字段的事件。我会得到多个事件（每个JSON属性一个事件）。就像这样：

{
       "message" => "  \"SOURCE\": \"Source A\",",
      "@version" => "1",
    "@timestamp" => "2014-08-31T01:26:23.432Z",
          "type" => "my-json",
          "tags" => [
        [0] "tag-json"
    ],
          "host" => "myserver.example.com",
          "path" => "/opt/mount/ELK/json/mytestjson.json"
}
{
       "message" => "  \"Model\": \"ModelABC\",",
      "@version" => "1",
    "@timestamp" => "2014-08-31T01:26:23.438Z",
          "type" => "my-json",
          "tags" => [
        [0] "tag-json"
    ],
          "host" => "myserver.example.com",
          "path" => "/opt/mount/ELK/json/mytestjson.json"
}
{
       "message" => "  \"Qty\": \"3\",",
      "@version" => "1",
    "@timestamp" => "2014-08-31T01:26:23.438Z",
          "type" => "my-json",
          "tags" => [
        [0] "tag-json"
    ],
          "host" => "myserver.example.com",
          "path" => "/opt/mount/ELK/json/mytestjson.json"
}

我应该使用multi-line编解码器还是json_lines编解码器？如果需要，我应该如何操作？我需要自己编写grok模式吗？还是有通用的JSON解析方法可以为我提供具有键值对的一个事件，就像上面的一个事件？我找不到任何能够解决我的问题的文档。非常感谢你的帮助。下面是我的配置文件：

input
{
        file
        {
                type => "my-json"
                path => ["/opt/mount/ELK/json/mytestjson.json"]
                codec => json
                tags => "tag-json"
        }
}

filter
{
   if [type] == "my-json"
   {
        date { locale => "en"  match => [ "RECEIVE-TIMESTAMP", "yyyy-mm-dd HH:mm:ss" ] }
   }
}

output
{
        elasticsearch
        {
                host => localhost
        }
        stdout { codec => rubydebug }
}

- Joseph

Answer 1

我认为我找到了解决我的问题的有效方法。我不确定它是否是一个干净的解决方案，但它有助于解析上述类型的多行JSON。

input 
{   
    file 
    {
        codec => multiline
        {
            pattern => '^\{'
            negate => true
            what => previous                
        }
        path => ["/opt/mount/ELK/json/*.json"]
        start_position => "beginning"
        sincedb_path => "/dev/null"
        exclude => "*.gz"
    }
}

filter 
{
    mutate
    {
        replace => [ "message", "%{message}}" ]
        gsub => [ 'message','\n','']
    }
    if [message] =~ /^{.*}$/ 
    {
        json { source => message }
    }

}

output 
{ 
    stdout { codec => rubydebug }
}

我的多行编解码器无法处理最后一个括号，因此它不会以JSON格式出现在json { source => message }中。因此需要使用mutate过滤器：

replace => [ "message", "%{message}}" ]

那么请加上缺失的括号。以及

。

gsub => [ 'message','\n','']

去掉引入的\n字符。最终得到的是一行JSON，可以通过json { source => message }读取。

如果有更简单的方法将原始的多行JSON转换为一行JSON，请发布POST，因为我感觉上述方法不太干净。

- Joseph

Answer 2

您需要使用多行编解码器。

input {
  file {
    codec => multiline {
        pattern => '^{'
        negate => true
        what => previous
    }
    path => ['/opt/mount/ELK/json/mytestjson.json']
  }
}
filter {
  json {
    source => message
    remove_field => message
  }
}

你会遇到的问题与文件中的最后一个事件有关。只有在文件中再出现另一个事件时，它才会显示出来（所以基本上你会失去一个文件中的最后一个事件）——在旋转文件之前，你可以附加一个单独的 { 来处理这种情况。

- Alcanzar

---

谢谢Alcanzar，不过我遇到了JSON解析失败的问题： [0] "_jsonparsefailure" 尝试将模式改为 pattern => '^{' 但结果还是一样。而且我的文件每个文件中只有一个JSON，即只有一个{或}字符。每个文件都是一个事件（1个文件= 1个JSON = 1个事件）。 - Joseph

你可能需要在文件输入中添加 start_position => beginning 以确保它从记录的开头开始... 还有你的文件中是否还有其他内容？（您可以删除过滤器，只需添加 output { stdout {} } 来查看它正在收集什么以传递给 json 过滤器） - Alcanzar

我注意到我的生产JSON确实有额外的“{”和“}” :( 所以我的JSON实际上是这样的：{ "SOURCE":"Source A", "Model":"ModelABC", "Qty":"3" "DESC": "{"New prod-125"}" } (抱歉在评论中无法解析) 而且我不能更改这些JSON。我们从另一个来源接收它们，我需要按原样使用。 - Joseph

在对其进行json处理之前，您需要“修复”消息。例如，您可以使用mutate过滤器和gsub => [ 'message','\"','']。如果您需要更复杂的内容，则可以使用ruby代码过滤器。 - Alcanzar

我认为问题在于将我的多行JSON（由大括号限定）缩减为一行，然后我就可以应用过滤器：if [message] =~ /^{.*}$/ {json { source => message } }。如何将我的多行JSON缩减为一行？我不是Ruby专家，所以我无法做到这一点。有什么建议吗？奇怪的是我找不到其他需要解析多行JSON的人。 - Joseph