如何使用logstash过滤器处理多行日志条目? [英] How to process multiline log entry with logstash filter?

问题描述

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

I have a custom generated log file that has the following pattern :

[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知道多行事件已结束，此行稍后会被删除.

The second entry [2014-03-02 17:34:20] - 127.0.0.1|DEBUG| flush_multi_line Is a dummy line, just to let logstash know that the multi line event is over, this line is dropped later on.

我的配置文件如下:

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 (

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

My problem is that I want to store the rest of the multiline to content field :

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

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

So i can remove the message field later.

@message 字段包含整个多行事件，因此我尝试了 mutate过滤器，并使用了 replace 功能，但是我我只是无法使其工作:(.

The @message field contains the whole multiline event so I tried the mutate filter, with the replace function on that, but I'm just unable to get it working :( .

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

I don't understand the Multiline filter's way of working, if someone could shed some light on this, it would be really appreciated.

谢谢

阿卜杜.

推荐答案

我遍历了源代码，发现:

I went through the source code and found out that :

• 多行过滤器将取消所有被认为是待处理事件的后续事件，然后将该行添加到原始消息字段中，表示任何在这种情况下，多行过滤器之后的过滤器将不适用
• 唯一会通过过滤器的事件是被认为是新事件的事件(在我的情况下，该事件以 [开头")

• The multiline filter will cancel all the events that are considered to be a follow up of a pending event, then append that line to the original message field, meaning any filters that are after the multiline filter won't apply in this case
• The only event that will ever pass the filter, is one that is considered to be a new one ( something that start with [ in my case )

这是工作代码:

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

干杯

阿布杜

这篇关于如何使用logstash过滤器处理多行日志条目?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！