logstash經常使用插件介紹

前言

在《使用EFK快速搭建安全可靠的日誌服務》一文中，咱們已經大體介紹了分佈式日誌服務的各個組件。可是對於不少實現細節，並未涵蓋到。本文主要介紹logstash用到的插件，以及對插件的離線打包，避免每次手動安裝插件。本文主要分爲4部分：

1. 插件的離線打包
2. input插件
3. filter插件
4. output插件

在使用插件以前，咱們先了解一個概念：事件。Logstash 每讀取一次數據的行爲叫作事件。

1.插件的離線打包

在生產環境中，logstash多是部署到多臺機器上，若是每次logstash的安裝更新都須要手動更新一遍插件成本是很是高的。那咱們能夠對用到的插件進行offline打包。
製做logstash-offline-plugins-x.x.x.zip說明：

1. 須要在一臺已經安裝了logstash的機器上，先給logstash安裝好所須要的插件
2. 而後再導出插件，具體命令：

bin/logstash-plugin update logstash-filter-mutate 
bin/logstash-plugin install logstash-filter-json_encode 
bin/logstash-plugin prepare-offline-pack logstash-filter-mutate logstash-filter-json_encode

參考elastic的官方文檔：https://www.elastic.co/guide/...

在logstash安裝時經過以下命令進行插件包的安裝：

bin/logstash-plugin install file://$(pwd)/logstash-offline-plugins.zip

安裝插件後要及時觀察logstash的打印日誌，避免由於插件版本衝突致使的logstash啓動失敗的狀況。

2.input插件

輸入插件容許一個特定的事件源能夠讀取到 Logstash 管道中，配置在 input {} 中，且能夠設置多個。 在《使用EFK快速搭建安全可靠的日誌服務》一文中咱們是分別將input、filter和output三部分的配置放到了不一樣的配置文件中。input的配置在input.conf文件中。
因爲咱們使用的是EFK，所以須要beats插件。能夠參考官網：https://www.elastic.co/guide/...

input {
    beats {
        port => 5044
        client_inactivity_timeout => 600
        ssl => true
        ssl_certificate_authorities => ["/home/work/certificate/chain-ca.pem"]
        ssl_certificate => "/home/work/certificate/server.crt.pem"
        ssl_key => "/home/work/certificate/server.key.pem"
        ssl_verify_mode => "force_peer"
    }
}

須要注意的是，在6.4版本以上的beats配置中，多了ssl_peer_metadata，能夠拿到證書裏的meta信息，能夠便於咱們後續在logstash上作一些鑑權校驗。
除了beats插件，inputs類型插件還有不少種，主要列舉以下：

• elasticsearch
• file
• log4j
• kafka

更多插件配置介紹參考官網：https://www.elastic.co/guide/...

3.filter插件

filter插件就比較多了。日誌接入以後要進行日誌的規則匹配、過濾等操做，配置filter是必不可少的。最長用到的filter插件有如下幾種：

• grok
• date
• mutate
• geoip
• split
• uuid
• json
• jsonencode

3.1 grok插件

Grok是logstash最主要的過濾插件，grok是經過系統預約義的正則表達式或者經過本身定義正則表達式來匹配日誌中的各個值，安裝了logstash後默認會有一些經常使用軟件的日誌匹配正則，在使用時本身能夠根據本身的日誌格式或者對匹配正則進行調整或者直接調用。若是，本身想在其餘目錄定義正則規則匹配日誌，在使用時須要指定正則的路徑。
對於如下一行日誌：

55.3.244.1 GET /index.html 15824 0.043

使用以下的規則：

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

grok配置示例以下：

filter {
  grok {
    match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
  }
}

那麼這行日誌通過grok以後會獲得如下的field：

client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043

經常使用的配置選項：

• match：用來對字段的模式進行匹配
• patterns_dir：用來指定規則的匹配路勁，若是使用logstash自定義的規則時，不須要寫此參數。Patterns_dir能夠同時制定多個存放過濾規則的目錄。語法格式：patterns_dir => [「/ opt / logstash / patterns」，「/ opt / logstash / extra_patterns」]
• remove_field：若是匹配到某個」日誌字段，則將匹配的這個日誌字段從這條日誌中刪除

Grok過濾正則規則能夠本身根據本身的日誌格式自行編寫，在編寫grok過濾規則時容易出錯，此時可使用grokdebug網站對本身的日誌及編寫的正則規則進行調試，grokdebug網址爲（https://grokdebug.herokuapp.com/）

3.2 date插件

date用於解析字段中的日期，而後使用該日期或時間戳做爲事件的logstash時間戳。
對於date插件的介紹，在https://segmentfault.com/a/11...（感受是官網https://www.elastic.co/guide/... 的翻譯版本）中有很是詳細的介紹。我來列舉一個本身在配置過程當中遇到的示例：
對於給定的時間格式：

2019-01-10T18:11:28.699+08:00

須要進行配置的pattern應該以下：

date {
    match => ["[@metadata][log_timestamp]", "yyyy-MM-dd'T'HH:mm:ss.SSSZZ"]
    timezone => "Asia/Shanghai"
}

須要注意的是，對於filter.conf配置是能夠熱加載的，若是時間格式配置錯誤，會致使logstash進程掛掉，對於問題的排查，能夠自行查詢logstash的日誌。

3.3 mutate插件

mutate 插件能夠在字段上執行變換，包括重命名、刪除、替換和修改。這個插件至關經常使用。
好比：
你已經根據 Grok 表達式將 Tomcat 日誌的內容放到各個字段中，想把狀態碼、字節大小或是響應時間，轉換成整型；
你已經根據正則表達式將日誌內容放到各個字段中，可是字段的值，大小寫都有，這對於 Elasticsearch 的全文檢索來講，顯然用處不大，那麼能夠用該插件，將字段內容所有轉換成小寫。
示例以下：

filter {
    mutate {
        split => ["hostname", "."]
        add_field => { "shortHostname" => "%{hostname[0]}" }
    }

    mutate {
        rename => ["shortHostname", "hostname" ]
    }
}

在筆者以前寫的《使用EFK快速搭建安全可靠的日誌服務》一文中，主要使用mutate插件來add_field, remove_field, convert，gsub等操做。其中covert能夠將字符串類型轉換爲int/long/float等類型，便於進行聚合等操做。
例如：

mutate {
  convert => {
        "averageSliceQueryTime" => "integer"
        "maxSliceQueryTime" => "integer"
  }

關於更多filter類型插件能夠參考：https://www.elastic.co/guide/...

4.output插件

和前面介紹的input插件同樣，output插件也是最基礎最簡單的輸出插件。在《使用EFK快速搭建安全可靠的日誌服務》一文中咱們使用的elasticsearch做爲最終的輸出存儲，所以output插件使用的是elasticsearch。對於grok解析失敗的文件會寫入到本地文件中，所以用了file插件，經過分支進行判斷。示例以下：

output {
    if "_grokparsefailure" in [tags] or "_jsonparsefailure" in [tags] {
        file {
            path => "/home/work/logstash/failure_output/failure-%{+YYYY-MM-dd}.log"
        }
    } else {
        elasticsearch {
            hosts => ["https://es-host1:9920","https://es-host2:9920","https://es-host3:9920"]
            index => "logstash-%{[@metadata][index_name]}-%{+YYYY.MM.dd}"
            document_id => "%{[@metadata][document_id]}"
            ssl => true
            ssl_certificate_verification => true
            truststore => "/home/work/certificate/truststore.jks"
            truststore_password => "adofkoe"
            user => "logstash"
            password => "dafodmfkamkefadfg"
        }
    }
}

除了elasticsearch，還支持如下輸出：

• email
• file
• influxdb
• mongodb
• redis

更多能夠參考官網：https://www.elastic.co/guide/...

總結

logstash的各類插件極大豐富了其功能。在實際使用中，咱們能夠根據本身的實際需求進行相關配置，來構建本身的日誌服務系統。筆者此處主要是對本身使用過程當中所用到內容的總結。遇到的不少實際問題，能夠自行查詢官網進行解決。