如何使用logstash过滤器处理多行日志条目？

Question

背景:

我有一个自定义生成的日志文件,具有以下模式:

[2014-03-02 17:34:20] - 127.0.0.1|ERROR| E:\xampp\htdocs\test.php|123|subject|The error message goes here ; array (
  'create' => 
  array (
    'key1' => 'value1',
    'key2' => 'value2',
    'key3' => 'value3'
  ),
)
[2014-03-02 17:34:20] - 127.0.0.1|DEBUG| flush_multi_line

第二个条目[2014-03-02 17:34:20] - 127.0.0.1|DEBUG| flush_multi_line是一个虚拟行,只是为了让logstash知道多行事件结束,稍后将删除此行.

我的配置文件如下:

input {
  stdin{}
}

filter{
  multiline{
      pattern => "^\["
      what => "previous"
      negate=> true
  }
  grok{
    match => ['message',"\[.+\] - %{IP:ip}\|%{LOGLEVEL:loglevel}"]
  }

  if [loglevel] == "DEBUG"{ # the event flush  line
    drop{}
  }else if [loglevel] == "ERROR"  { # the first line of multievent
    grok{
      match => ['message',".+\|.+\| %{PATH:file}\|%{NUMBER:line}\|%{WORD:tag}\|%{GREEDYDATA:content}"] 
    }
  }else{ # its a new line (from the multi line event)
    mutate{
      replace => ["content", "%{content} %{message}"] # Supposing each new line will override the message field
    }
  }  
}

output {
  stdout{ debug=>true }
}

内容字段的输出是:The error message goes here ; array (

问题:

我的问题是我想将多行的其余部分存储到内容字段:

The error message goes here ; array (
  'create' => 
  array (
    'key1' => 'value1',
    'key2' => 'value2',
    'key3' => 'value3'
  ),
)

所以我可以稍后删除消息字段.

该@message字段包含整个多事件,所以我尝试了发生变异过滤,用替换上的功能,但我只是无法得到它的工作:(.

我不了解Multiline过滤器的工作方式,如果有人可以对此有所了解,那将非常感激.

谢谢,

阿卜杜.

Answer 1

我浏览了源代码,发现:

• 该多过滤器将取消所有正在发生的事件认为是跟进未决事件,然后追加该行原来的消息字段,这意味着任何过滤器后,多过滤器将不会适用于这种情况
• 唯一会通过过滤器的事件是一个被认为是新事件的事件(以[在我的情况下] 开头的事情)

这是工作代码:

input {
   stdin{}
}  

filter{
      if "|ERROR|" in [message]{ #if this is the 1st message in many lines message
      grok{
        match => ['message',"\[.+\] - %{IP:ip}\|%{LOGLEVEL:loglevel}\| %{PATH:file}\|%{NUMBER:line}\|%{WORD:tag}\|%{GREEDYDATA:content}"]
      }

      mutate {
        replace => [ "message", "%{content}" ] #replace the message field with the content field ( so it auto append later in it )
        remove_field => ["content"] # we no longer need this field
      }
    }

    multiline{ #Nothing will pass this filter unless it is a new event ( new [2014-03-02 1.... )
        pattern => "^\["
        what => "previous"
        negate=> true
    }

    if "|DEBUG| flush_multi_line" in [message]{
      drop{} # We don't need the dummy line so drop it
    }
}

output {
  stdout{ debug=>true }
}

干杯,

阿卜杜

Answer 2

本期中提到了grok和多行处理https://logstash.jira.com/browse/LOGSTASH-509

只需在你的grok正则表达式前添加"(？m)",你就不需要变异了.问题示例:

pattern => "(?m)<%{POSINT:syslog_pri}>(?:%{SPACE})%{GREEDYDATA:message_remainder}"

Answer 3

多行过滤器会在消息中添加"\n".例如:

"[2014-03-02 17:34:20] - 127.0.0.1|ERROR| E:\\xampp\\htdocs\\test.php|123|subject|The error message goes here ; array (\n  'create' => \n  array (\n    'key1' => 'value1',\n    'key2' => 'value2',\n    'key3' => 'value3'\n  ),\n)"

但是,grok过滤器无法解析"\n".因此,您需要将\n替换为另一个字符,即空格.

mutate {
    gsub => ['message', "\n", " "]
}

然后,grok模式可以解析消息.例如:

 "content" => "The error message goes here ; array (   'create' =>    array (     'key1' => 'value1',     'key2' => 'value2',     'key3' => 'value3'   ), )"