使用grok-match匹配日志+output时format调整日志格式

不修改nginx日志格式的情况下，使用logstash filter-grok，匹配nginx日志格式后转为json格式存入kafka。

input {
  file {
    type => "access-log"
    path => ["/var/log/nginx/access.log"]
  }
}

filter {
  if [type] == "access-log" {
    grok {
      match => { "message" => '%{IPV4:remote_ip} \- %{HTTPDUSER:remote_user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status} %{NUMBER:bytes} %{QS:referer} %{QS:useragent} %{USER:gzipratio}'}
    }
  }
}

output {
  if [type] == "access-log" {
    kafka {
      bootstrap_servers  => "ip1:9092,ip2:9092,ip3:9092"
      topic_id => "nginx-access-log"
      codec => plain {
        format => '{"remoteip":"%{remote_ip}","remoteuser":"%{remote_user}","timestamp":"%{timestamp}"},"method":"%{method}","request":"%{request}","httpversion":"HTTP/%{httpversion}","status":"%{status}","bytes":"%{bytes}","referer":%{referer},"useragent":%{useragent},"gzipratio":"%{gzipratio}"}'
      }
    }
  }
}

grok对nginx日志内容进行匹配

logstash内置的grok匹配规则都在此目录下，可以服用已有的正则表达式匹配日志内容:

有很多预置的正则表达式，其名称可能与要匹配的值的命令不通，但这完全不影响你复用它。比如：内置的USER的正则表达式，可以匹配到数字、字母、下划线、中划线……当你想要匹配的值也可以用这个正则来匹配时，完全可以直接复用USER。

logstash-8.6.1/vendor/bundle/jruby/2.6.0/gems/logstash-patterns-core-4.3.4/patterns/legacy/

需要注意的是，match中匹配nginx log内容时，一个符号都不能错，要完全匹配日志内容。否则后续引用匹配到的变量时，变量将无法正确被使用。

grok {

match => { “message” => ‘%{IPV4:remote_ip} \- %{HTTPDUSER:remote_user} \[%{HTTPDATE:timestamp}\] “%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}\” %{NUMBER:status} %{NUMBER:bytes} %{QS:referer} %{QS:useragent} %{USER:gzipratio}’}

}

format将输出格式调整为json格式

可以人为将匹配到的变量，以json的格式输出到kafka。

codec => plain {

format => ‘{“remoteip”:”%{remote_ip}”,”remoteuser”:”%{remote_user}”,”timestamp”:”%{timestamp}”},”method”:”%{method}”,”request”:”%{request}”,”httpversion”:”HTTP/%{httpversion}”,”status”:”%{status}”,”bytes”:”%{bytes}”,”referer”:%{referer},”useragent”:%{useragent},”gzipratio”:”%{gzipratio}”}’

}