Logstash學習系列之插件介紹

時間 2019-12-20

原文原文鏈接

Logstash插件獲取方式

插件獲取地址：
https://github.com/logstash-plugins 

在線安裝:
/plugin install  logstash-input-jdbc

升級插件：
/plugin update logstash-input-jdbc 
卸載插件：
/plugin uninstall  logstash-input-jdbc

淘寶源地址： https://ruby.taobao.org

Logstash插件分類

inputs 輸入
codecs 解碼
filters 過濾
outputs 輸出

Logstash_input插件

stdin標準輸入html

可用參數linux

參數               Input Type                Required        Default Value
add_field            hash                      No                 {}
codec                codec                     NO                 "line"
tags                 array                     NO             
type                 string                    NO

file文件輸入nginx

可用參數git

input {
    file {
        #file插件的參數
        codec=>...                        #可選項                codec,         默認是plain,可經過這個參數設置編碼方式
        discover_interval=>...            #可選項                number,        logstash沒隔多久去檢查一次被監聽的path下是否有新文件,默認值是15秒
        exclude=>...                      #可選項                array,         不想被監聽的文件能夠排除出去,這裏跟path同樣支持glob展開
        sincedb_path=>...                 #可選項                string,        若是你不想用默認的$HOME/.sincedb,能夠經過這個配置定義sincedb文件到其餘位置    
        stat_interval...                  #可選項                number,        logstash沒隔多久檢查一次被監聽文件狀態(是否有更新),默認是1秒
        start_position...                 #可選項                string,        logstash從什麼位置開始讀取文件數據,默認是結束位置,也就是說logstash進程會以相似tial -f的形式運行.若是你是要導入一個完整的文件,把這個設定能夠改爲"beginning",logstash進程就會從文件開頭讀取,有點相似cat
        path=>...                         #必選項                array,         處理的文件的路徑,能夠定義多個路徑
        tags=>...                         #可選項                array,         在數據處理的過程當中,由具體的插件來添加或者刪除的標記
        type=>...                         #可選項                string,        自定義將要處理事件類型,能夠本身隨便定義,好比處理的是linux的系統日誌,能夠定義爲"syslog"
    }
}

實例1:過濾nginx的訪問日誌和錯誤日誌github

配置文件正則表達式

[root@ELK-STACK logstash-1.5.5]# cat conf/test1.conf
input {
    file {
        path => ["/var/log/nginx/access.log"]
        type => "accesslog"
        start_position => "beginning"
        }
    file {
        path => ["/var/log/nginx/error.log"]
        type => "errorlog"
        start_position => "beginning"
        }
    }
output {
    stdout{}
    }

測試數據庫

[root@ELK-STACK logstash-1.5.5]# ./bin/logstash -f conf/test1.conf
Logstash startup completed
2017-02-23T08:00:25.757Z 0.0.0.0 2017/02/23 12:14:01 [error] 2145#0: *47 open() "/usr/share/nginx/html/zabbix" failed (2: No such file or directory), client: 172.16.2.64, server: _, request: "GET /zabbix HTTP/1.1", host: "172.16.1.225"
2017-02-23T08:00:25.757Z 0.0.0.0 172.16.2.64 - - [23/Feb/2017:12:14:01 +0800] "GET /zabbix HTTP/1.1" 404 3650 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/56.0.2924.87 Safari/537.36" "-"
2017-02-23T08:01:09.785Z 0.0.0.0 172.16.2.64 - - [23/Feb/2017:16:01:09 +0800] "GET / HTTP/1.1" 200 3700 "-" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12) AppleWebKit/602.1.50 (KHTML, like Gecko) Version/10.0 Safari/602.1.50" "-"

TCP/UDP輸入json

TCPvim

語法格式ruby

input {
    tcp {
        #tcp插件的參數                           類型            默認值
        add_field=>...            #可選項        hash           {}
        codec=>...                #可選項                       plain
        data_timeout=>...        #可選項         Number         1
        host=>...                #可選項                        0.0.0.0                
        mode=>...                #可選項                       值是["server","client"]其中之一,默認是server    
        port=>...                #必填項         number        端口號,須要和另外一端匹配
        ssl_cacert=>...            #可選項
        ssl_cert=>...            #可選項
        ssl_enable...            #可選項
        ssk_key=>...            #可選項
        ssl_key_passphrase=>...    #可選項
        ssl_verify...            #可選項
        tags=>...                #可選項         array         
        type=>...                #可選項         string         
    }
}

UDP

語法格式

input {
    udp {
        #可用參數  
                                                                 #默認值
        add_field=>...            #hash類型            #可選項      默認{}
        host=>...                #string類型           #可選項      默認0.0.0.0 
        port=>...                #number類型           #必填項      端口號,須要和另外一端匹配
        tags=>...                #array類型            #必填項
        type=>...                #string類型           #可選項
        workers=>...            #number類型            #默認2
    }
}

Logstash_codec插件

介紹: codec:編碼,解碼(json,msgpack,edn)

放置位置: input{}和output()字段裏的任何插件裏(好比說input裏file插件裏 input{file{codec=>json}})

codec插件之plain

介紹:plain是一個空的解析器，它可讓用戶本身指定格式

codec之json

介紹

若是輸入到logstash的內容是json格式,能夠在input字段加入codec=>json來進行解析
若是想讓logstash輸出爲json格式,能夠在output字段加入codec=>json

語法格式: codec=> json

實例1:輸出json化

配置文件

[root@ELK-STACK logstash-1.5.5]# cat conf/json.conf
input {
    stdin {
        }
    }
output {
    stdout {
        codec => json
        }
    }

測試

[root@ELK-STACK logstash-1.5.5]# ./bin/logstash -f conf/json.conf
Logstash startup completed             #等待logstash啓動成功       
hello.world                            #輸入一個字符串的hello.world
{"message":"hello.world","@version":"1","@timestamp":"2017-02-24T09:01:11.425Z","host":"0.0.0.0"}  #能夠看到輸入是以json化的格式輸出的

codec之json_lines

介紹: 若是你的json文件比較長,須要換行的話,那麼久得用到json_lines了

codec之rubydebug

介紹: rubydebug將採用Ruby Awsome Print庫來解析日誌

實例1:經過rubydebug輸出鍵值對

配置文件

 cat conf/rubydebug.conf
input {
    stdin {
        codec => json       #輸入爲json格式,因此用codec=>json來解析
        }

    }



output {
    stdout {
        codec =>rubydebug  #讓輸出的格式爲rubydebug模式,就是把鍵值對輸出
        }
    }

測試

[root@ELK-STACK logstash-1.5.5]# ./bin/logstash -f conf/rubydebug.conf
Logstash startup completed
{"bookname":"elk stack"}             #這是輸入的鍵值對
{                                        
      "bookname" => "elk stack",    #這是以rubydebug的模式輸出鍵值對
      "@version" => "1",            #這是以rubydebug的模式輸出鍵值對
    "@timestamp" => "2017-02-24T09:29:50.777Z",
          "host" => "0.0.0.0"
}
{"bookname":"elk stack","price":29}           #這是輸入的鍵值對
{
      "bookname" => "elk stack",               #這是以rubydebug的模式輸出鍵值對
         "price" => 29,                        #這是以rubydebug的模式輸出鍵值對
      "@version" => "1",
    "@timestamp" => "2017-02-24T09:31:08.224Z",
          "host" => "0.0.0.0"
}

codec之multiline

介紹: 有些時候，應用程序調試日誌會包含很是豐富的內容，爲一個事件打印出不少行內容。這種日誌一般都很難經過命令行解析的方式作分析;而 logstash 正爲此準備好了 codec/multiline 插件

語法格式

input {
    stdin {
        codec =>multiline {
            charset=>...          #可選                      字符編碼         
            max_bytes=>...        #可選     bytes類型            設置最大的字節數
            max_lines=>...        #可選     number類型           設置最大的行數,默認是500行
            multiline_tag...      #可選     string類型           設置一個事件標籤,默認是multiline
            pattern=>...          #必選     string類型           設置匹配的正則表達式
            patterns_dir=>...     #可選     array類型           能夠設置多個正則表達式
            negate=>...           #可選     boolean類型         設置true是向前匹配,設置false向後匹配,默認是FALSE
            what=>...             #必選                        設置未匹配的內容是向前合併仍是前後合併,previous,next兩個值選擇
        }
    }

}

Logstash_filter插件

json filter

介紹:若是數據格式是json,那麼能夠經過它把數據解析成你想要的數據結構

語法格式

filter {
    json {
        add_field=>...            #可選項         #hash          添加屬性,默認{}
        add_tag=>...              #可選項         #array          添加標識,默認{}
        remove_field=>...         #可選項         #array         刪除屬性,默認{}
        remove_tag=>...            #可選項         #array         刪除標識,默認{}
        source=>...                #必選項         #string       指定來源數據
        target=>...                #可選項         #string        定義將要解析的目標字段
    }
}

實例1:解析json數據

配置文件

vim conf/filter_json.conf
 input {
     stdin {
         }
     }
 filter {
     json {
         source => "message"
         target => "content"
         }
     }
 output {
         stdout {
             codec=>rubydebug
             }
     }

測試

[root@ELK-STACK logstash-1.5.5]# ./bin/logstash -f conf/filter_json.conf
Logstash startup completed
{"name":"zhai","age":12}       #輸入被解析的值
{
       "message" => "{\"name\":\"zhai\",\"age\":12}",
      "@version" => "1",
    "@timestamp" => "2017-02-27T08:31:13.193Z",
          "host" => "0.0.0.0",
       "content" => {
        "name" => "zhai",   #這裏就是json字段裏的name
         "age" => 12        #這裏就是json字段裏的age
    }
}

grok filter

介紹

grok是目前logstash裏最好的一種解析各類非結構化的日誌數據的工具
grok能夠過濾日誌中你想要的字段
官方patterns地址:https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns
測試grok匹配規則的網址:grokdebug.herokuapp.com

可用參數

filter {
    grok {
        match=>...        #可選項        寫匹配規則
    }
}

kv filter

介紹: 經過指定一個分隔符,截取key,value

Logstash_output插件

介紹: logstash output就是如何讓數據通過logstash的解析和處理後,把結果輸出

輸出到file

可用參數

path => "/root/access_result"                 #結果輸出到文件裏
path => "/root/%{+YYYY.MM.dd}-%{host}-access" #結果輸出到文件裏,以時間命名文件
message_format => "%{ip}"                      #只輸出filter過濾出來的指定字段

輸出到elasticsearch

elasticsearch就是數據庫
把logstash解析過濾的結果輸出到elasticsearch裏

實例1:輸出到elasticsearch

output {
    elasticsearch {
        host => "172.16.1.225"             #elasticsearch的地址,或者cluster=>"ClusterName"
        protocol =>"http"            
        index=>"test_output-%{type}-%{+YYYY.MM.dd}"    #在elasticsearch裏建立索引,索引名稱設置,type就是document_type的值,test_output-nginx-2017-02-28
        document_type=>"nginx"
        worker=>5
    }
}