使用Logstash的正则匹配日志格式

牛刀小试

在ELK里经常需要写正则来匹配日志里的的具体信息，这样可以在kibana上更加直观的观看，grok就是Logstash最重要的插件。你可以在grok里预定义好命名正则表达式，在稍后(grok参数或者其他正则表达式里)引用它。

grok的格式是：%{SYNTAX:SEMANTIC}，如果需要转义就要加上\。具体的grok匹配规则可以在logstash查看grok-patterns这个文件，如图：

SYNTAX代表匹配值的类型，例如，0.11可以NUMBER类型所匹配，10.222.22.25可以使用IP匹配。
SEMANTIC表示存储该值的一个变量声明，它会存储在elasticsearch当中方便kibana做字段搜索和统计，你可以将一个IP定义为客户端IP地址client_ip_address，eg:%{IP:client_ip_address}，所匹配到的值就会存储到client_ip_address这个字段里边，类似数据库的列名。

而检测grok正则的网站：http://grokdebug.herokuapp.com/ 。

举个例子，如果日志里是：

1	100.97.73.142 - - [15/Feb/2019:16:54:24 +0800]

可以看到里面有IP、日期、时间、时区，那么对比刚才的grok-patterns文件，就知道要匹配IP这个字段，就是用IP (?:%{IPV6}|%{IPV4})，IP即包含了IPV6也有IPV4，那么具体的匹配就是%{IP:client}，client是自己定义名称。同理，[15/Feb/2019这部分是日期，可以使用如下的配置规则：

如法炮制，匹配结果就是\[%{MONTHDAY:day}/%{MONTH:month}/%{YEAR:year}。

拿到检测网站试一下结果：

上面那个例子比较简单，比如匹配下面这样的一个日志：

[2019-02-01T08:59:59.124] [INFO] default - [115.63.121.10 GET /wap/_nuxt/vendor.61a8f274bce0acb0de6d.js 200 97s][https://www.lechange.com/wap/node/goodDetail/264 HTTP/1.1 Mozilla/5.0 (Linux; Android 8.1; V1818A Build/OPM1.171019.026; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044409 Mobile Safari/537.36]

那么写法就是：

20%{YEAR:year}-%{MONTHNUM:month}-%{MONTHDAY:day}T%{HOUR:hour}:?%{MINUTE:minutes}(?::?%{SECOND:second})] \[%{LOGLEVEL:level}\] default - \[%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:status} %{NUMBER:duration}s\]\[%{DATA:data}\]

效果如图:

但是要注意!在网站上的匹配可能是OK的，但是在logstash的grok里是不可以有-、'、"等这样的字符出现，比如下面这个日志：

100.97.73.232 - - [15/Feb/2019:16:54:24 +0800] "GET /public/app/site/statics/favicon.ico HTTP/1.1" 200 4286 "https://www.lechange.com/wap/" "Mozilla/5.0 (Linux; Android 7.1.1; OPPO R11 Build/NMF26X; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/57.0.2987.132 MQQBrowser/6.2 TBS/044409 Mobile Safari/537.36 Imou"

虽然下面的规则是可以正确匹配的：

%{IP:client} - - \[%{MONTHDAY:day}/%{MONTH:month}/%{YEAR:year}:%{HOUR:hour}:%{MINUTE:minutes}:%{SECOND:second} \+0800] \"%{WORD:method} %{URIPATH:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:status} %{NUMBER:bytes} "%{DATA:website}" "%{DATA:data}"

但是当你把这个配置复制到logstash.conf里的时候，启动logstash就会有参数不合法的报错：

1	[2019-02-18T17:33:52,060][FATAL][logstash.runner ] The given configuration is invalid. Reason: Expected one of #, {, -, ", ', } at line 32, column 218 (byte 1226) 。

应该改成这样：

%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-)" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QUOTEDSTRING:domain} %{QUOTEDSTRING:data}

一般来说对于字符串，有双引号包含的用QS，没有的用DATA类型，如%{DATA:request_body}。

grok配置到logstash

假设这个服务的日志输入到kafka里的topic叫lcshop-log，那么对应的logstash如下：

input {
        kafka{
                bootstrap_servers=>"172.31.0.84:9092"	#这里写的是kafka的ip和端口
                topics=>["lcshop-log","lcshop-errorlog"]	#这里是对应的topic
                decorate_events=>"true"
                codec=>plain
        }       
}   

filter {
	if [@metadata][kafka][topic] == "lcshop-log" {
      mutate {
         add_field => {"[@metadata][index]" => "lcshop-log-%{+YYYY-MM}"}
      }  
      grok {	
         match => { "message" => "%{IPORHOST:client_ip} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:http_version})?|-)\" %{NUMBER:response} (?:%{NUMBER:bytes}|-) %{QUOTEDSTRING:domain} %{QUOTEDSTRING:data}"}
         remove_field => ["message"]
		#这里对双引号增加了转义符
      }
   } else if [@metadata][kafka][topic] == "lcshop-errorlog" {
      mutate {
         add_field => {"[@metadata][index]" => "lcshop-errorlog-%{+YYYY-MM}"}
      }
   }
}

output {
    stdout {
               codec => rubydebug		#日志输出，这个看情况开启，生成日志量非常可观！
       }
       elasticsearch {
               hosts=>["172.31.0.76:9200"]		#这里是es的ip和端口
               index=>"%{[@metadata][index]}"		#这里对不同的topic分配不同的index
       }
}

然后去kibana里查看结果,然而会发现这是失败的。

为什么明明在grok的测试网站里通过了，但是在实际的kibana里却失败？这是因为在kibana里会自动给双引号添加一个转义符！正式因为多了这个转义符，所以整段正则都不匹配了，自然grok无法生效了。

这个时候需要改成这样：

1	%{IPORHOST:clientip} - %{NOTSPACE:remote_user} \[%{HTTPDATE:timestamp}\] \\"%{DATA:request}\\" %{NUMBER:response} %{NUMBER:bytes} %{DATA:data} \\"%{DATA:detail}\\"

重启logstash之后，再去kibana里看结果：