在 logstash 中使用 grok 解析多行 JSON

Question

我有一个 JSON 格式：

{
    "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"
}

我应该使用多行编解码器还是 json_lines 编解码器？如果是这样，我该怎么做？我是否需要编写自己的 grok 模式，或者是否有一些通用的 JSON 可以给我一个带有键的事件：我为上面的一个事件获得的值对？我找不到任何说明这一点的文档。任何帮助，将不胜感激。我的conf文件如下所示：

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 }
}

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 }。因此变异过滤器：

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

这增加了缺少的大括号。和

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

删除引入的\n 字符。最后，我有一个可以被json { source => message }读取的单行JSON

如果有一种更简洁/更简单的方法可以将原始多行 JSON 转换为单行 JSON，请执行 POST，因为我觉得上面的内容不太干净。

Answer 2

您需要使用multiline 编解码器。

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

您将遇到的问题与文件中的最后一个事件有关。在文件中有另一个事件之前它不会显示（所以基本上你会丢失文件中的最后一个事件）——你可以在文件被旋转之前附加一个 { 到文件以处理这种情况.