日志分析logstash插件介绍

Logstash是一款轻量级的日志搜集处理框架，能够方便的把分散的、多样化的日志搜集起来，并进行自定义的处理，而后传输到指定的位置，好比某个服务器或者文件。

logstash功能很强大。从 logstash 1.5.0 版本开始，logstash 将全部的插件都独立拆分红 gem 包。这样，每一个插件均可以独立更新，不用等待 logstash 自身作总体更新的时候才能使用了。为了达到这个目标，logstash 配置了专门的 plugins 管理命令

logstash插件安装（本地安装）

logstash处理事件有三个阶段：input ---> filter ---> output。input产生事件，filter 对事件进行修改，output输出到其它地方。

filter是logstash管道中间处理的设备。能够结合条件语句对符合标准的事件进行处理。这里只介绍filter的插件：

你能够经过 bin/plugin list 查看本机如今有多少插件可用。(其实就在 vendor/bundle/jruby/1.9/gems/ 目录下)

插件本地安装的方法

bin/logstash-plugin install logstash-filter-kv
Ignoring ffi-1.9.13 because its extensions are not built.  Try: gem pristine ffi --version 1.9.13
Validating logstash-filter-kv
Installing logstash-filter-kv
Installation successful

更新插件：bin/logstash-plugin update logstash-filter-kv

插件介绍

这里只介绍几种经常使用的

grok: 解析和结构化任何文本（前面单独介绍过就不重复了）

http://irow10.blog.51cto.com/2425361/1828077

geoip: 添加有关IP地址地理位置信息。

geoip这个插件很是重要，并且很经常使用。他能分析访问的ip分析出访问者的地址信息。举例以下：

filter {
    geoip {
        source => "message"
    }
}

message你输入个IP地址，结果以下：

{
       "message" => "183.60.92.253",
      "@version" => "1",
    "@timestamp" => "2016-07-07T10:32:55.610Z",
          "host" => "raochenlindeMacBook-Air.local",
         "geoip" => {
                      "ip" => "183.60.92.253",
           "country_code2" => "CN",
           "country_code3" => "CHN",
            "country_name" => "China",
          "continent_code" => "AS",
             "region_name" => "30",
               "city_name" => "Guangzhou",
                "latitude" => 23.11670000000001,
               "longitude" => 113.25,
                "timezone" => "Asia/Chongqing",
        "real_region_name" => "Guangdong",
                "location" => [
            [0] 113.25,
            [1] 23.11670000000001
        ]
    }
}

实际应用中咱们能够把grok获取的request_ip传给geoip处理。

filter {
  if [type] == "apache" {
    grok {
      patterns_dir => "/usr/local/logstash-2.3.4/ownpatterns/patterns"
      match => {
                "message" => "%{APACHE_LOG}"
                }
    remove_field => ["message"]
    }
    geoip {
        source => "request_ip"
    }
  }

在logstash分析完数据到output阶段输出到其它地方时，数据中就有访问者的地理信息。

date：用来转换你的日志记录中的时间字符串

date 插件是日期插件，这个插件，经常使用而重要。

该插件必须是用 date 包裹，以下所示：

date {

}

可用的配置选项以下表所示：

add_field 

add_tag

locale

match  匹配日志格式

periodic_flush 按时间间隔调用

remove_field

remove_tag

tag_on_failure 若是标签匹配失败，则默认为_grokparsefailure

target  把 match 的时间字段保存到指定字段。若为指定，默认更新到 @timestamp。

timezone

备注：

add_field、remove_field、add_tag、remove_tag 是全部 Logstash 插件都有。

tag 做用是，当你对字段处理期间，还指望进行后续处理，就先做个标记。Logstash 有个内置 tags 数组，包含了期间产生的 tag，不管是 Logstash 本身产生的，仍是你添加的，好比，你用 grok 解析日志，可是错了，那么 Logstash 本身就会本身添加一个 _grokparsefailure 的 tag。这样，你在 output 时，能够对解析失败的日志不作任何处理；

field 做用是，对字段的操做。

举例：

filter {
  if [type] == "apache" {
    grok {
      patterns_dir => "/usr/local/logstash-2.3.4/ownpatterns/patterns"
      match => {
                "message" => "%{APACHE_LOG}"
                }
    remove_field => ["message"]
    }
   date {
      match => [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
        }
  }
}

apache日志中的时间戳是：[19/Jul/2016:16:28:52 +0800] 。match的时间格式要和日志中的匹配对应。若是你的时间字段可能有多个格式，则可指定多个可能的日期格式：

match => [ "timestamp", "MMM dd YYY HH:mm:ss", "MMM  d YYY HH:mm:ss", "ISO8601" ]

apache日志的grok表达式：

APACHE_LOG %{IPORHOST:addre} %{USER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] \"%{WORD:http_method} %{NOTSPACE:request} HTTP/%{NUMBER:httpversion}\" %{NUMBER:status} (?:%{NUMBER:bytes}|-) \"(?:%{URI:http_referer}|-)\" \"%{GREEDYDATA:User_Agent}\"

在apache日志中已经有[%{HTTPDATE:timestamp}\]，为何还要通过date插件再处理下呢？

咱们将访问时间做为logstash的时间戳，有了这个，咱们就能够以时间为区分，查看分析某段时间的请求是怎样的，若是没有匹配到这个时间的话，logstash将以当前时间做为该条记录的时间戳。因此须要再filter里面定义时间戳的格式。若是不用 date 插件，那么 Logstash 将处理时间做为时间戳。时间戳字段是 Logstash 本身添加的内置字段 @timestamp，在ES中关于时间的相关查询，必须使用该字段，你固然也能够修改该字段的值。

备注：@timestamp 比咱们晚了 8 个小时，在kibana显示也是如此。如何能让日志收集时间能和@timestamp一致呢？在date插件中添加以下字段：timezone =>"Asia/Chongqing"

useragent：用来处理分析访问者使用的浏览器及操做系统

在apache日志中会发现有这么一段日志：Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/537.36

咱们在用grok分割数据时也是使用%{GREEDYDATA:User_Agent}。

备注：GREEDYDATA这个grok表达式是匹配任何类型的数据。  GREEDYDATA .*

经过这条数据，即便咱们显示出来意义也不大，但咱们能够经过useragent挖掘它的信息

filter {
  if [type] == "apache" {
    grok {
      patterns_dir => "/usr/local/logstash-2.3.4/ownpatterns/patterns"
      match => {
                "message" => "%{APACHE_LOG}"
                }
    remove_field => ["message"]
    }
   useragent {
                source => "User_Agent"
                target => "ua"
        }
   date {
      match => [ "timestamp", "dd/MMM/YYYY:HH:mm:ss Z" ]
        }
  }
}

显示结果：

从上图咱们能够看到访问者浏览器及操做系统的信息。比那一大串信息更加有意义。

mutate：它提供了丰富的基础类型数据处理能力。包括类型转换，字符串处理和字段处理等。

能够设置的转换类型包括："integer"，"float" 和 "string"。

可用的配置选项以下表所示：

add_field 添加新的字段

add_tag 添加新的标签

convert 数据类型转换

gsub 字符串替换。用正则表达式和字符串都行

join 用分隔符链接数组. 若是字段不是数组，那什么都不作

lowercase 把字符串转换成小写

merge 合并两个数组或散列字段

periodic_flush 按时间间隔调用

remove_field 移除字段

remove_tag 移除标识

rename 重命名一个或多个字段

replace 用一个新的值替换掉指定字段的值

split 用分隔符或字符分割一个字符串。只能应用在字符串上

strip 去掉字段首尾的空格

update 更新字段的值。若是该字段不存在，则什么都不作

uppercase 把字符串转换成大写

简单优化数据

logstash采集数据加上date，geoip，useragent等插件会使咱们获取的信息更加详细，可是也更加臃肿。全部咱们要踢掉一些没有意义的数据，简化传输给elasticsearch的数据。

remove_field能很好的完成这个任务。上面也有用到。

remove_field => ["message"]

在grok中咱们已经发message分红了不少段小数据，若是在把message传输给elasticsearch就重复了。

固然在传输的小数据中也有不少咱们用不到或者毫无心义。咱们就可使用remove_field来清除。

 mutate{
                remove_field => ["Syslog_Timestamp"]
                remove_field => ["message"]

参考：https://zengjice.gitbooks.io/logstash-best-practice-cn/content/filter/mutate.html

drop: 彻底丢弃事件，如debug事件

filter {
  if [loglevel] == "debug" {
    drop { }
  }
}

参考：https://www.elastic.co/guide/en/logstash/current/plugins-filters-drop.html