logstash截取指定字符
由于项目原因有些日志打印出来之后，会在kibana中显示很不友好而且加载ES的时候也特别的慢，所以我想有没有办法可以让日志在kibana中展示的比较友好一点呢，于是找来很多相关的资料，种感觉有点差异，然后自己摸索的一点出来
在网上看到有很多种截取方式 有在filebeat中做过滤的 ，有在logstash中过滤的，这里简单的说说logstash中的一些
我们用grok来做一些过滤，grok支持正则表达式.
正则表达式的验证.
| Item | Comment |
|---|---|
| %{USER:user} | 以 USER 模式进行正则匹配，结果放在user中 |
| [[^]]+] | 以 [ 开头 以]结尾，内容是由一个或多个不是 ] 的字符填充而成 |
| %{NUMBER: id:int} | 以 NUMBER 模式进行正则匹配，为整数型，结果放在id中 |
| \n | 匹配换行符 |
| %{NUMBER:query_time:float} | 以 NUMBER 模式进行正则匹配，为浮点型，结果放在query_time中 |
| (?:use\s+%{USER:usedatabase};\s*\n)? | 这个匹配可能有，也可能无，如果有，就是以use开头，若干空字符，以 USER 模式进行正则匹配，结果放在usedatabase中，然后紧接着 ; ，后面是0个或多个空字符，然后是换行，注意：如果有是整体有，如果无，是整体无 |
| \b | 代表字单词边界不占位置，只用来指示位置 |
| .* | 尽可能多的任意匹配 |
| (?< query>(?< action>\w+)\b.*) | 整体匹配，存到query中，以一个或多个字符开头组成的单词，结果存到action中 |
| (?:\n#\s+Time)? | 内容可能有，也可能无，如果有，是接在一个换行之后，以 # 开头，隔着一个或多个空字符，然后是Time |
| .*$ | 任意匹配直到结尾 |
grok正则表达式：(?<temMsg>(.*)(?=Report)/?) 获取Report之前的字符
grok正则表达式：(?<temMsg>(?=Report)(.*)/?) 获取Report之后的字符
grok{
match => {
#截取<Report>之前的字符作为temMsg字段的值
"message" => "(?<temMsg>(.*)(?=Report)/?)"
}
}
这个是截取特定的字符集日志，要日志中包含了【Report】关键字
（注：表达式中(?=Report)中的等于【=】符号如果换成【<=】这表示就不包含本身了，例如(?<temMsg>(.*)(?=Report)/?)可以写成(?<temMsg>(.*)(?<=Report)/?)这样输出的结果就不包含Report了，同理下面的一样）
grok正则表达式：(?<temMsg>(?<=report).*?(?=msg)) 截取report和msg之间的值 不包含report和msg本身
grok正则表达式：(?<temMsg>(report).*?(?=msg)) 截取 包含report但不包含msg
grok正则表达式：(?<temMsg>(?<=report).*?(msg))截取 不包含report但包含msg
grok正则表达式：(?<temMsg>(report).*?(msg|request))输出以report开头,以msg或者以request结尾的所有包含头尾信息
grok正则表达式：(?<temMsg>(report).*?(?=(msg|request)))输出以report开头,以msg或者以request结尾的不包含头尾信息
grok{
match => {
#截取<Report>之后的和<msg>之前的值作为temMsg字段的值
"message" => "(?<temMsg>(?<=report).*?(?=msg))"
}
}
这个是截取特定的字符集日志，要日志中包含了【report和msg和request】关键字
之间的表达式只要替换一下就可以使用了
（注过个表达式中出现异常，在单个的字符串中可以将小括号【()】去掉，例如：(report).*?(?=msg) 可以写成report.*?(?=msg)）
grok正则表达式：(?<MYELF>([\s\S]{500}))
grok{
match => {
#截取日志500个字符 作为MYELF的值
"message" => "(?<MYELF>([\s\S]{500}))"
}
}
对有所日志截取500个字符，可以加入if()做为判断条件，根据自身项目来
grok正则表达式：%{LOGLEVEL:level}
grok {
#这个patterns_dir大家都应该正对 单独写表达式的地方
#patterns_dir => "/usr/local/nlp/logstash-6.0.1/config/patterns"
match => [
"message","%{LOGLEVEL:level}"
]
}
这个比较简单 就不多说了
结合上面的 这个是对level级别的日志做判断 如果日志中含有DEBUG的，就drop掉
if [level] == "DEBUG" {
drop { }
}
这个其实和上面差不多，加了一个【~】表示对单条的前后日志做匹配
if[message]=~"ASPECT"{
drop { }
}
这个是说对temMsg赋值的所有的日志从新命名打印message
mutate {
#重命名字段temMsg为message
rename => {"temMsg" => "message"}
}
#logstash过滤器切割
filter {
if [type] == "simple" {
mutate{
split => ["message","|"] #按 | 进行split切割message
add_field => {
"requestId" => "%{[message][0]}"
}
add_field => {
"timeCost" => "%{[message][1]}"
}
add_field => {
"responseStatus" => "%{[message][2]}"
}
add_field => {
"channelCode" => "%{[message][3]}"
}
add_field => {
"transCode" => "%{[message][4]}"
}
}
mutate {
convert => ["timeCost", "integer"] #修改timeCost字段类型为整型
}
} else if [type] == "detail" {
grok{
match => {
#将message里面 TJParam后面的内容，分隔并新增为ES字段和值
"message" => ".*TJParam %{PROG:requestId} %{PROG:channelCode} %{PROG:transCode}"
}
}
grok{
match => {
#截取TJParam之前的字符作为temMsg字段的值
"message" => "(?<temMsg>(.*)(?=TJParam)/?)"
#删除字段message
remove_field => ["message"]
}
}
mutate {
#重命名字段temMsg为message
rename => {"temMsg" => "message"}
}
}
}
最后来一个完整的
input {
redis {
data_type => "list"
host => "localhost1"
port => "5100"
key => "nlp_log_file"
db => 0
threads => 1 #线程数量
codec => "json"
}
redis {
data_type => "list"
host => "localhost2"
port => "5101"
key => "nlp_log_file"
db => 0
threads => 1 #线程数量
codec => "json"
}
}
filter {
grok {
#patterns_dir => "/usr/local/nlp/logstash-6.0.1/config/patterns"
match => [
"message","%{LOGLEVEL:level}"
]
}
grok{
match => {
#截取<ReportPdf>之前的字符作为temMsg字段的值
"message" => "(?<temMsg>(.*)(?=<ReportPdf>)/?)"
}
}
mutate {
#重命名字段temMsg为message
rename => {"temMsg" => "message"}
}
if [level] == "DEBUG" {
drop { }
}
if[message]=~"ASPECT"{
drop { }
}
#获取日志文件带RAWT关键字的
if[message]=~"[RAW]"{
grok{
match => {
#截取带RAW关键字的日志500个字符 作为MYELF的值
"message" => "(?<MYELF>([\s\S]{500}))"
}
}
mutate {
rename => {"MYELF" => "message"} #重命名字段MYELF为message
}
}
}
output {
elasticsearch {
hosts => ["localhost:9200"]
index => "logstash-%{+YYYY.MM.dd}"
action => "index"
template_overwrite => true
#user => "elastic"
#password => "admins-1"
}
stdout{codec => dots}
}
