Logstash语法常用案例解析（二）

Logstash语法常用案例解析（二）

摘要

本文主要讲过滤插件，处理了nginx日志的各种例子。

最后一部分继续说插件。

1、过滤插件

格洛克:常规捕获

日期:时间处理

Mutate:数据修改

Geoip:查询分类

JSON:编解码器

grok:解析和构造任何文本。

http://grokdebug.herokuapp.com/patterns#匹配规则，注意空网格，如果空网格不匹配，也会报错

http://grokdebug.herokuapp.com/matching检查，有语法提示

Grok是目前logstash将非结构化日志数据解析为结构化和可查询数据的最佳方式。Logstash内置了120种匹配模式，可以满足大部分需求。

格式:

filter {     grok {         match => { "message" => "grok_pattern" }     } }

注意:
grok_pattern由零个或多个%{SYNTAX:SEMANTIC}组成，其中SYNTAX是表达式的名称，由grok提供。例如，数值表达式的名称是NUMBER，IP地址表达式的名称是IP。语义表示被解析字符的名称，它是由它自己定义的。例如，IP字段的名称可以是client。

简单的例子:

#cat  conf.d/test.conf   input {stdin{}}          #输入方式为标准输入   filter {   grok {                   #grok插件匹配     #patterns_dir => "/path/to/patterns"   #将匹配规则写到指定文件方便管理     match => {     "message" => "%{WORD} %{NUMBER:request_time:float} %{WORD}" }           #WORD匹配字符串，NUMBER匹配数值，支持int,float格式。匹配的值赋给request_time变量   #remove_field => ["message"]  #处理结果删除掉message字段   } } output {                  #输出方式为标准输出   stdout {codec=>rubydebug}        #定义输出格式为rubydebug }

结果:

# ./bin/logstash   -f conf.d/test.conf Logstash startup completed begin 123.456 end   { "message" => "begin 123.456 end",   #remove之后就不显示了。 "@version" => "1", "@timestamp" => "2016-05-09T02:43:47.952Z", "host" => "dev-online", "request_time" => 123.456               #grok匹配中新加的变量 }

NGX日志处理匹配:
由于Nginx日志已经被处理成json数据，所以是key:value的方式，以rubydebug格式打印如下:

所以现在我想过滤掉不想要的字段。

input {   file {     path => "/var/log/nginx/access.log"     type => "json"     codec => "json"     start_position => "beginning"   } } filter {   grok {     match => {     "@timestamp" =>"%{WORD}"        #先把不想要的字段匹配出来     "type" => "%{WORD}"     }   remove_field => ["@timestamp","type"]  ＃再移除字段 } } output {   stdout {     codec=>rubydebug   } }

运行结果:

Nginx日志json格式:

log_format     json        '{"@timestamp":"$time_iso8601",'                             '"@version":"1",'                             '"host":"$server_addr",'                             '"client":"$remote_addr",'                             '"size":$body_bytes_sent,'                             '"responsetime":$request_time,'                             '"domain":"$host",'                             '"url":"$request",'                             '"refer":"$http_referer",'                             '"agent":"$http_user_agent",'                             '"status":"$status"}'; access_log /var/log/nginx/access.log json;

nginx配置文件中常用的常规匹配参数

nginx日志格式匹配项目注释

$remote_addr%{IPORHOST:clientip}

$remote_user%{notspace:remote_user}

[$time_local]\[%{httpdate:timestamp}\]"[]"需要是特殊字符，需要转义。

"$request""%{WORD:method}访问请求通常添加""，匹配时也会添加""。匹配单词GET，POST

方法%{uripathparam:request}uripathparam与请求的uri匹配。

http%{number:httpversion}"number与该数字匹配，并被分配给httpversionhttp协议版本。

$status%{NUMBER:status}NUMBER与该号码匹配，并被分配给status作为退货状态

$body_bytes_sent%{number:response}内容大小

$http_referer“%{QS:referer}”与请求refer匹配。

$http_user_agent“%{QS:agent}”与友好代理匹配。

“%{QS:xfforwardefor}”的$http_x_forwarded_for与xfw匹配。

$upstream_addr%{IPv4:upstream}:%{POSINT:port}

$scheme%{WORD:scheme}匹配http或https

例如:nginx日志格式:

log_formataccess'$remote_addr-$remote_user[$time_local]"$request"$status$body_bytes_sent"$http_referer""$http_user_agent"'；

日志实例:

"192.168.1.22---[20/Apr/2016:16:28:14+0800]"GET/ask/232323.htmlHTTP/1.1"50015534"HTTP://test.103.100xhs.com/""Mozilla/5.0(WindowsNT6.1；WOW64)AppleWebKit/537.36(KHTML，像壁虎一样)Chrome/45.0.2454.101Safari/537.36""

匹配规则

"%{IPORHOST:clientip}-%{notspace:remote_user}\[%{HTTPdate:timestamp}\]"%{WORD:method}%{URIPATHPARAM:request}HTTP/%{NUMBER:HTTPversion}"%{NUMBER:status}%{NUMBER:response}%{QS:referrer}%{QS:agent}"

Geoip地址查询:

GeoIP是最常见的免费IP地址分类查询库。GeoIP库可以根据IP地址提供相应的区域信息，包括国家、省、市、经纬度等。这是非常有用的可视化地图和区域统计。

input {     stdin{}     }   filter {   geoip {       source => "message"     #source必须为公网ip  否则geoip不会显示数据    #       fields => ["city_name","country_code2","country_name","latitude","longitude"]           }      #geoip输出的内容比较多，可以指定输出的列       } output {   stdout{       codec=>rubydebug         }   }

查找对应IP的键是geoip中source指定的值。

完整示例:

结果:

注意:geoip插件的“source”字段可以是任何处理过的字段，比如“client_ip”，但是字段的内容需要
慎重！只有公共网络上的IP信息存储在IP数据库中。如果没有找到查询结果，将直接返回null。

JSON:

input {stdin{}}   filter {     json {       source => "message"              #必选项     } } output {   stdout{     codec=>rubydebug   } }

结果:

{"name":"wd","age":"15"} { "message" => "{\"name\":\"wd\",\"age\":\"15\"}", "@version" => "1", "@timestamp" => "2016-05-09T06:32:13.546Z",     #加一个时间戳的好处是方便kibana导入 "host" => "dev-online", "name" => "wd", "age" => "15" }

日期事件处理

注:因为%{+YYYY。后期outputs/elasticsearch常用的MM.dd}，要读取@timestamp数据，一定不能直接删除这个字段来保留自己的字段，要用filters/date转换后删除自己的字段！

filter {   grok {     match => ["message", "%{HTTPDATE:logdate}"]   }   date {     match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]   } }

注意:时区偏移量只需要一个字母z。

更改数据修改

1、类型转换

可以设置的转换类型包括:“整数”、“浮点”和“字符串”。例子如下

filter {   mutate {     convert => ["request_time", "float"]   } }

注意:mutate不仅转换简单的字符值，还支持数组类型的字段转换，即["1"，"2"]转换为[1，2]。但是，不支持对哈希类型字段的类似处理。那些需要这样做的人可以通过使用稍后描述的filters/ruby插件来完成。

2、字符串处理

Gsub仅对字符串类型字段有效。

gsub=>["urlparams"，"[\？#]","_"]

使分离

split=>["消息"，"|"]

随意输入一串用|分隔的字符，如“123|321|adfd|dfjld*=123”，可以看到如下输出:

联接仅对数组类型字段有效。

我们以前使用过分割切割的基础，然后再连接回去。将配置更改为:

join=>["消息"，"，"]

合并两个数组或哈希字段。在上一次拆分的基础上继续:

merge=>["消息"，"消息"]

Rename重命名字段，如果目标字段已经存在，它将被覆盖:

rename=>["系统日志主机"，"主机"]

更新字段的内容。如果该字段不存在，则不会创建它。

replace的作用和update类似，但是当字段不存在时，会和add_field参数作用相同，自动添加一个新字段。

编解码器编码插件

Json:直接输入预定义的JSON数据，这样可以省略filter/grok的配置！

path=>"/var/log/nginx/access.log_json""

编解码器=>“json”

多行:合并多行数据

stdin {   codec => multiline {     pattern => "^\["     negate => true     what => "previous"   } }

终端:以结尾结束，换行不能结束。