日志分析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