将 json 输入到 logstash - 配置问题？答案

【问题标题】：input json to logstash - config issues?将 json 输入到 logstash - 配置问题？
【发布时间】：2014-10-30 23:57:58
【问题描述】：

我有以下 json 输入，我想转储到 logstash（最终在 elasticsearch/kibana 中搜索/仪表板）。

{"vulnerabilities":[
    {"ip":"10.1.1.1","dns":"z.acme.com","vid":"12345"},
    {"ip":"10.1.1.2","dns":"y.acme.com","vid":"12345"},
    {"ip":"10.1.1.3","dns":"x.acme.com","vid":"12345"}
]}

我正在使用以下 logstash 配置

input {
  file {
    path => "/tmp/logdump/*"
    type => "assets"
    codec => "json"
  }
}
output {
  stdout { codec => rubydebug }
  elasticsearch { host => localhost }
}

输出

{
       "message" => "{\"vulnerabilities\":[\r",
      "@version" => "1",
    "@timestamp" => "2014-10-30T23:41:19.788Z",
          "type" => "assets",
          "host" => "av12612sn00-pn9",
          "path" => "/tmp/logdump/stack3.json"
}
{
       "message" => "{\"ip\":\"10.1.1.30\",\"dns\":\"z.acme.com\",\"vid\":\"12345\"},\r",
      "@version" => "1",
    "@timestamp" => "2014-10-30T23:41:19.838Z",
          "type" => "assets",
          "host" => "av12612sn00-pn9",
          "path" => "/tmp/logdump/stack3.json"
}
{
       "message" => "{\"ip\":\"10.1.1.31\",\"dns\":\"y.acme.com\",\"vid\":\"12345\"},\r",
      "@version" => "1",
    "@timestamp" => "2014-10-30T23:41:19.870Z",
          "type" => "shellshock",
          "host" => "av1261wag2sn00-pn9",
          "path" => "/tmp/logdump/stack3.json"
}
{
            "ip" => "10.1.1.32",
           "dns" => "x.acme.com",
           "vid" => "12345",
      "@version" => "1",
    "@timestamp" => "2014-10-30T23:41:19.884Z",
          "type" => "assets",
          "host" => "av12612sn00-pn9",
          "path" => "/tmp/logdump/stack3.json"
}

很明显，logstash 将每一行视为一个事件，它认为{"vulnerabilities":[ 是一个事件，我猜后面 2 个后续节点上的尾随逗号会弄乱解析，最后一个节点看起来是正确的。我如何告诉 logstash 解析漏洞数组中的事件并忽略行尾的逗号？

更新：2014-11-05 按照 Magnus 的建议，我添加了 json 过滤器，它运行良好。但是，如果不在文件输入块中指定 start_position => "beginning"，它将无法正确解析 json 的最后一行。有什么想法为什么不呢？我知道它默认解析自下而上，但预计 mutate/gsub 会顺利处理这个问题？

file {
    path => "/tmp/logdump/*"
    type => "assets"
    start_position => "beginning"
  }
}
filter {
  if [message] =~ /^\[?{"ip":/ {
    mutate {
      gsub => [
        "message", "^\[{", "{",
        "message", "},?\]?$", "}"
      ]
    }
    json {
      source => "message"
      remove_field => ["message"]
    }
  }
}
output {
  stdout { codec => rubydebug }
  elasticsearch { host => localhost }
}

【问题讨论】：

标签： elasticsearch logstash

【解决方案1】：

您可以跳过 json 编解码器并使用多行过滤器将消息连接成单个字符串，您可以将其提供给 json filter.filter {

filter {
  multiline {
    pattern => '^{"vulnerabilities":\['
    negate => true
    what => "previous"
  }
  json {
    source => "message"
  }
}

但是，这会产生以下不需要的结果：

{
            "message" => "<omitted for brevity>",
           "@version" => "1",
         "@timestamp" => "2014-10-31T06:48:15.589Z",
               "host" => "name-of-your-host",
               "tags" => [
        [0] "multiline"
    ],
    "vulnerabilities" => [
        [0] {
             "ip" => "10.1.1.1",
            "dns" => "z.acme.com",
            "vid" => "12345"
        },
        [1] {
             "ip" => "10.1.1.2",
            "dns" => "y.acme.com",
            "vid" => "12345"
        },
        [2] {
             "ip" => "10.1.1.3",
            "dns" => "x.acme.com",
            "vid" => "12345"
        }
    ]
}

除非漏洞数组中有固定数量的元素，否则我认为我们对此无能为力（不借助 ruby 过滤器）。

如何将 json 过滤器应用于看起来像我们想要的行并丢弃其余的行？您的问题并不清楚是否所有日志都看起来像这样，所以这可能不是那么有用。

filter {
  if [message] =~ /^\s+{"ip":/ {
    # Remove trailing commas
    mutate {
      gsub => ["message", ",$", ""]
    }
    json {
      source => "message"
      remove_field => ["message"]
    }
  } else {
    drop {}
  }
}

【讨论】：