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