内容简介：Logstash语法常用案例解析（二）
摘要
此篇主要讲Filter插件，已经对nginx 日志的各种处理实例
接着上篇继续说插件
1，Filter插件
-
Grok：正则捕获
-
Date：时间处理
-
Mutate：数据修改
-
Geoip：查询归类
-
JSON：编解码
Grok： 解析和结构化任何文本。
http://grokdebug.herokuapp.com/patterns# 匹配规则,注意空格，如果空格不匹配也会报错
http://grokdebug.herokuapp.com/ 匹配检查，而且有语法提示
Grok 目前是logstash最好的方式对非结构化日志数据解析成结构化和可查询化。logstash内置了120个匹配模式，满足大部分需求。
格式：
filter {
grok {
match => { "message" => "grok_pattern" }
}
}
注 ：
grok_pattern由零个或多个%{SYNTAX:SEMANTIC}组成，其中SYNTAX是表达式的名字，是由grok提供的，例如数字表达式的名字是NUMBER，IP地址表达式的名字是IP。SEMANTIC表示解析出来的这个字符的名字，由自己定义，例如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匹配中新加的变量
}
Nginx 日志处理匹配：
因为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} 访问请求，一般都加"",匹配时也加一下。 WORD匹配GET,POST
method %{URIPATHPARAM:request} URIPATHPARAM匹配请求的uri
HTTP %{NUMBER:httpversion}" NUMBER匹配数字，并赋值给httpversion http协议版本
$status %{NUMBER:status} NUMBER匹配数字,并赋值给status,作为返回状态
$body_bytes_sent %{NUMBER:response} 内容大小
"$http_referer" "%{QS:referrer}" 匹配请求refer
"$http_user_agent" "%{QS:agent}" 匹配亲切agent
"$http_x_forwarded_for" "%{QS:xforwardedfor}" 匹配xfw
$upstream_addr %{IPV4:upstream}:%{POSINT:port}
$scheme %{WORD:scheme} 匹配http or https
eg:nginx日志格式：
log_format access '$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.html HTTP/1.1" 500 15534 "http://test.103.100xhs.com/" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/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的key 就是geoip 中source所指定的值。
完整的例子：
结果：
注意：geoip 插件的 "source" 字段可以是任一处理后的字段，比如 "client_ip"，但是字段内容却需要
小心！geoip 库内只存有公共网络上的 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"
}
Date事件处理
注意：因为在稍后的 outputs/elasticsearch 中常用的 %{+YYYY.MM.dd} 这种写法必须读取 @timestamp 数据，所以一定不要直接删掉这个字段保留自己的字段，而是应该用 filters/date 转换后删除自己的字段！
filter {
grok {
match => ["message", "%{HTTPDATE:logdate}"]
}
date {
match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]
}
}
注意：时区偏移量只需要用一个字母 Z 即可。
Mutate数据修改
1,类型转换
可以设置的转换类型包括："integer"，"float" 和 "string"。示例如下
filter {
mutate {
convert => ["request_time", "float"]
}
}
注意：mutate 除了转换简单的字符值，还支持对数组类型的字段进行转换，即将 ["1","2"] 转换成 [1,2]。但不支持对哈希类型的字段做类似处理。有这方面需求的可以采用稍后讲述的 filters/ruby 插件完成。
2,字符串处理
gsub 仅对字符串类型字段有效
gsub => ["urlparams", "[\\?#]", "_"]
split
split => ["message", "|"]
随意输入一串以|分割的字符，比如 "123|321|adfd|dfjld*=123"，可以看到如下输出：
join 仅对数组类型字段有效
我们在之前已经用 split 割切的基础再 join 回去。配置改成：
join => ["message", ","]
merge合并两个数组或者哈希字段。依然在之前 split 的基础上继续：
merge => ["message", "message"]
rename 重命名某个字段，如果目的字段已经存在，会被覆盖掉：
rename => ["syslog_host", "host"]
update 更新某个字段的内容。如果字段不存在，不会新建。
replace 作用和 update 类似，但是当字段不存在的时候，它会起到 add_field 参数一样的效果，自动添加新的字段。
Codec编码插件
json: 直接输入预定义好的 JSON 数据，这样就可以省略掉 filter/grok 配置！
path => "/var/log/nginx/access.log_json""
codec => "json"
Multiline:合并多行数据
stdin {
codec => multiline {
pattern => "^\["
negate => true
what => "previous"
}
}
终端输入：以 the end 结束，换行无法结束。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持 码农网
猜你喜欢:
- Protobuf语法全解析
- Nginx Location 指令语法解析
- Smali 语法解析——Hello World
- 使用 Calcite 做 SQL 语法解析
- [译] Python 之父的解析器（七）：PEG 解析器的元语法
- php – 解析错误：语法错误,意外’未设置'(T_UNSET)
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
Programming Python
Mark Lutz / O'Reilly Media / 2006-8-30 / USD 59.99
Already the industry standard for Python users, "Programming Python" from O'Reilly just got even better. This third edition has been updated to reflect current best practices and the abundance of chan......一起来看看 《Programming Python》 这本书的介绍吧!
