elasticsearch - Logstash:突变过滤器不起作用

Question

我有以下过滤器

filter {
    grok {
    break_on_match => false
        match => { 'message' => '\[(?<log_time>\d{0,2}\/\d{0,2}\/\d{2} \d{2}:\d{2}:\d{2}:\d{3} [A-Z]{3})\]%{SPACE}%{BASE16NUM}%{SPACE}%{WORD:system_stat}%{GREEDYDATA}\]%{SPACE}%{LOGLEVEL}%{SPACE}(?<log_method>[a-zA-Z\.]+)%{SPACE}-%{SPACE}%{GREEDYDATA:log_message}%{SPACE}@%{SPACE}%{IP:app_host}:%{INT:app_port};%{SPACE}%{GREEDYDATA}Host:%{IPORHOST:host_name}:%{POSINT:host_port}' }
    match => { 'message' => '\[(?<log_time>\d{0,2}\/\d{0,2}\/\d{2} \d{2}:\d{2}:\d{2}:\d{3} [A-Z]{3})\]'}
    }
    kv{
      field_split => "\n;"
      value_split => "=:" 
      trimkey => "<>\[\],;\n"
      trim => "<>\[\],;\n"
    }
    date{
    match => [ "log_time","MM/dd/YY HH:mm:ss:SSS z" ]
    target => "log_time"
    locale => "en"
    }

    mutate {

      convert => { 
              "line_number" => "integer" 
              "app_port" => "integer" 
              "host_port" => "integer" 
              "et" => "integer" 

      }

      #remove_field => [ "message" ]

    }

    mutate {

    rename => {

              "et" => "execution_time"
              "URI" => "uri"
              "Method" => "method"

      }
    }

}

我可以从grok和kv过滤器中获得结果，但是任何mutate过滤器都无法工作。是因为有kv滤波器吗？

编辑:用途

我的问题是我的日志包含异构日志记录。例如

[9/13/16 15:01:18:301 EDT] 89798797 SystemErr                                                     jbhsdbhbdv [vjnwnvurnuvuv] INFO djsbbdyebycbe - Filter.doFilter(..) took 0 ms.

[9/13/16 15:01:18:302 EDT] 4353453443 SystemErr                                                    sdgegrebrb [dbebtrntn] INFO sverbrebtnnrb - [SECURITY AUDIT] Received request from: "null" @ wrvrbtbtbtf:000222; Headers=Host:vervreertherg:1111
Connection:keep-alive
User-Agent:Mozilla/5.0
Accept:text/css,*/*;q=0.1
Referer:https:kokokfuwnvuwnev/ikvdwninirnv/inwengi
Accept-Encoding:gzip
Accept-Language:en-US,en;q=0.8
; Body=; Method=GET; URI=dasd/wgomnwiregnm/iwenviewn; et=10ms; SC=200

我只关心捕获每个记录开头的时间戳和其他一些字段(如果存在)。我想要 Method， et， Host， loglevel和 URI。如果这些字段不存在，我仍然希望捕获 loglevel和 message记录的事件。

是否建议使用相同的logstash流程捕获此类事件？我应该运行两个logstash进程吗？问题是，除了我确实要捕获的几个字段之外，我事先不知道日志的结构。

多行配置

path => ["path to log"]
start_position => "beginning" 
ignore_older => 0 
sincedb_path => "/dev/null"
codec => multiline {
pattern => "^\[\d{0,2}\/\d{0,2}\/\d{2} \d{2}:\d{2}:\d{2}:\d{3} [A-Z]{3}\]"
negate => "true"
what => "previous"

Answer 1

可能是因为最初的grok期间没有创建某些字段(line_number，et，URI，Method)。例如，我看到您定义了“log_method”，但是在mutate-> rename中，您引用了“Method”。输入块中是否存在json编解码器或其他应用程序，以添加这些额外字段？

如果您发布示例日志，我可以使用您的过滤器对其进行测试，并为您提供更多帮助。 :)

编辑:
我看到您发送的日志有多行。您是否在输入上使用多行过滤器？您也可以共享您的输入块吗？

您绝对不需要运行两个Logstash进程。一个Logstash可以处理多种日志格式。您可以使用条件，尝试/捕获或通过添加“？”将字段标记为可选字段后。

更多编辑:
我得到的输出表明您的mutate过滤器有效:

"execution_time" => 10,
"uri" => "dasd/wgomnwiregnm/iwenviewn",
"method" => "GET"

一旦我将 trimkey => "<>\[\],;\n"更改为 trimkey => "<>\[\],;( )?\n"。我注意到这些字段(et，Method)都以空格作为前缀。

注意:我正在使用以下多行过滤器进行测试，如果您使用的过滤器不同，则会影响结果。让我知道是否有帮助。

codec => multiline {
  pattern => "\n"
  negate => true
  what => previous
}