Logstash學習系列之插件介紹

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標准輸入

可用參數

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

file文件輸入

可用參數

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的訪問日志和錯誤日志

配置文件

[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輸入

TCP

語法格式

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
    }
}