本站所有文章均为原创,如对您有帮助,恳请帮忙点击任何一处广告

ELK Stack插件——Logstash filter的常用插件

发布:TangLu2017-4-25 18:04分类: ELK Stack 标签: ELK ElasticSearch logstash kibana grok ELKStack

Logstash的filter插件可以将传输过来的数据进行过滤解析并完成格式转换,是最为重要和复杂的一个插件。它可以对收集到的日志做多种处理,比如解析数据、删除字段、切割、格式化等。比如收集到的Nginx日志中包含了客户端IP、访问时间、请求页面、返回状态码、客户端浏览器等信息,如果不做任何处理的话这些内容在Kibana中全部显示在message这一个字段中,不利于阅读,使用filter插件中的grok功能则可以每个字段拆分开来


filter插件又包含了多个小插件,下面是一些常用插件的介绍:

1、date插件

用于日期转换,将日期类型的字符串生成为一个新的字段,用以替代默认的@timestamp字段。主要解决日志本身产生的时间与Kibana收集时间不一致的问题,比如有个历史日志是2020年1月产生的,如果不做配置的话Kibana收集到日志会是当前时间。

filter {
  date{
    match => [ "logdate", "MMM dd yyyy HH:mm:ss"]  #定义一个logdate字段,字段格式为后面引号内的内容。当logstash遇到这样格式的内容就会放入到logdate字段中
    target => "@timestamp"  #用logdatae覆盖默认的timestamp字段
    timezone => "Asia/Shanghai"
  }
}


2、json插件

可以将指定字段内容转为json格式

filter{
  json{
    source => "message"  #要处理的字段名
    target => "msg_json"  #解析后储存到哪个字段,该字段和message同级
  }
}


3、grok插件

将日志中符合正则规则的字段重新定义为另一个字段,实现格式化输出。Logstash已经预定义了很多规则,规则文件存放在logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/。通过一些在线grokdebug网站,可以快速的通过字符串内容查找出已经定义好的正则表达式,如下:

grok_debug1.jpg


grok插件配置说明,以一个简单的日志为例,可以看到这个日志有客户端IP、请求方法、请求页面等字段

55.3.244.1 GET /index.html 15824 0.043


如果想把不同字段做分类的话,就可以用grok来进行定义。Logstash已经事先写好了很多规则,只需引用:

%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}

大写的单词是预定义的正则表达式,可以在patterns目录看到其具体匹配的内容。如果匹配的话,就将该字段名称重新定义为冒号后面的小写的单词。如果正则表达式不匹配的话会出_grokparsefailure的错误。用上面那条访问日志的客户端IP来说明,它将在Kibana里被定义为client字段并单独显示出来,而不是全部塞在了message里


使用grok插件处理apache日志示例:

input{
  file {
    path => "/usr/local/apache/logs/access_log"
    type => "apache_access.log"
    start_position => "beginning"
  }
}

filter {
  if [type] == "apache_access.log" {
    grok {
      match => {"message" => "%{COMBINEDAPACHELOG}" }
      }
  }
}

output{
  if [type] == "apache_access.log" {
    elasticsearch {
      hosts => ["192.168.44.129:9200"]
      index=> "apache_access-%{+YYYY-MM}"
    }
  }
}


使用grok插件处理nginx日志的示例,这是我公司线上环境的配置,将Nginx日志按需要进行了字段划分,并且去掉了部分不需要的字段:

input {
  file {
    path => [ "/tmp/test.txt" ]
  }
}

filter {
  grok {
    patterns_dir => ["./patterns"]
    match => {"message" => "%{IPORHOST:remote_addr} - %{USER:remote_user} %{TIMESTAMP_ISO8601:time} %{IPORHOST:site} \"(?:%{WORD:method} %{URIPATH:url}(|%{URIPARAMC:query})(?: HTTP/%{NUMBER:htt
pversion})?|%{DATA})\" %{NUMBER:status} (?:%{NUMBER:body_bytes_sent}|-) %{QS:http_referer} %{QS:http_user_agent} %{QS:x_forwarded_for} (((%{URIHOST:http_user_agent})(, |- )?|(-, |-)?)+) (((
%{NUMBER:upstream_status})(, |- )?|(-, |-)?)+) (((%{BASE16FLOAT:upstream_response_time})(, |- )?|(-, |-)?)+) (%{BASE16FLOAT:request_time}) %{WORD:scheme} %{IPORHOST:real_ip}"}
    remove_field => ["message","@version"]
    }
  }

output {
  elasticsearch {
    hosts => ["192.168.145.85:9200"]
    index => "test-%{+YYYY.MM.dd}"
  }
}

定义好grok后可以通过Kibana来查看日志格式,如图:左边选择了三个字段,然后中间已经把这三个字段显示出来了,说明日志已经自动切割好了,非常直观

grok2.png


在Kibana搜索中还有一些小技巧,在Dev Tools工具的Grok Debugger工具中可以对grok规则进行测试,如果正确匹配上的话能看到相关数据;也可以按需要过滤某个字段,比如需要查看response返回为404的日志,只需要在搜索框里输入response:404即可,如图

grok3.png


4、dissect插件

基于分割符来切割字段,避免grok消耗CPU的问题,性能是grok的三倍左右。但是需要每行日志格式相似,并有明显的分割符,下图就是把一条日志进行了切割,+号代表追加到某个字段中

diss.png



5、mutate插件该插件也是使用很频繁的插件,可以对字段内容做处理,比如使用rename进行字段重命名、使用remove_field删除字段、使用convert进行数据类型转换等,转换后的数据一般用于Kibana聚合出图,所以当遇到Kibana出图时缺少自己需要的字段时可以考虑是否数据类型不对

filter {
  mutate { 
    remove_field => [ "message" ]  #删除message字段,多个字段用逗号分隔
    convert => {
      "userid" => "integer"  #将userid字符串格式转换为int整数型
      "Action" => "strint"
    }
  }
}


6、geoip插件,配置后会在Lostash的输出中新增地理位置的相关信息

filter {
  geoip {
    source => "ip"  #source指定需要处理的字段,这里就是将IP字段中的数据进行分析
    fields => ["location","country_name","city_name"] #对geoip输出的N多字段过滤,只看自己需要的 
  }
}


温馨提示如有转载或引用以上内容之必要,敬请将本文链接作为出处标注,谢谢合作!
et_highlighter51
版权所有:《Linux运维技术学习站点
文章标题:《ELK Stack插件——Logstash filter的常用插件
除非注明,文章均为 《Linux运维技术学习站点》 原创
转载请注明本文短网址:http://www.linuxe.cn/post-313.html  [生成短网址]

已有 0/8891 人参与

发表评论:

欢迎分享Linux运维技术学习站点

欢迎使用手机扫描访问本站,还可以关注微信哦~