2.1 輸入插件在"hello World" 示例中,我們已經(jīng)見到并介紹了Logstash 的運(yùn)行流程和配置的基礎(chǔ)語法。請記住一個(gè)原則: Logstash 配置一定要有一個(gè)input和一個(gè)output在演示過程中,如果沒有寫明input,默認(rèn)就會使用 logstash-input-stdin同理,沒有寫明的output 就是logstash-output-stdout2.1.1 標(biāo)準(zhǔn)輸入[elk@Vsftp logstash]$ cat stdin.conf input {  stdin {  add_field =>{"key11"=>"value22"}  codec=>"plain"  tags=>["add"]  type=>"std" }}output {     stdout {                        codec => rubydebug                } }[elk@Vsftp logstash]$ logstash -f stdin.conf Settings: Default pipeline workers: 4Pipeline main startedabc123{       "message" => "abc123",      "@version" => "1",    "@timestamp" => "2017-02-08T02:14:53.476Z",          "type" => "std",         "key11" => "value22",          "tags" => [        [0] "add"    ],          "host" => "Vsftp"}[elk@Vsftp logstash]$ cat stdin.conf input {  stdin {  add_field =>{"key11"=>"value22"}  codec=>"plain"  tags=>["add","xxyy"]  type=>"std" }}output {     stdout {                        codec => rubydebug                } }[elk@Vsftp logstash]$ logstash -f stdin.conf Settings: Default pipeline workers: 4Pipeline main startedthis is scan{       "message" => "this is scan",      "@version" => "1",    "@timestamp" => "2017-02-08T02:15:39.183Z",          "type" => "std",         "key11" => "value22",          "tags" => [        [0] "add",        [1] "xxyy"    ],          "host" => "Vsftp"}根據(jù)tags 判斷:[elk@Vsftp logstash]$ cat stdin.conf input {  stdin {  add_field =>{"key11"=>"value22"}  codec=>"plain"  tags=>["add","xxyy"]  type=>"std" }}output {  if "tttt" in [tags]{   stdout {    codec=>rubydebug{}     }  }   else if "add" in [tags]{   stdout {    codec=>json     }  }    }[elk@Vsftp logstash]$ logstash -f stdin.conf Settings: Default pipeline workers: 4Pipeline main startedyyyyyjjjj{"message":"yyyyyjjjj","@version":"1","@timestamp":"2017-02-08T02:20:42.833Z","type":"std","key11":"value22","tags":["add","xxyy"],"host":"Vsftp"}2.1.2  文件輸入:logstash 使用一個(gè)名叫FileWatch的Ruby Gem庫來監(jiān)聽文件變化。這個(gè)庫支持glob展開文件路徑,而且會記錄一個(gè)叫.sincedb的數(shù)據(jù)庫文件來跟蹤被監(jiān)聽日志文件的當(dāng)前讀取位置[elk@Vsftp logstash]$ cat log.conf input {  file {   path =>["/var/log/*.log","/var/log/mm"]   type=>"system"  start_position =>"beginning"}}output {        stdout {                        codec => rubydebug                } }[elk@Vsftp logstash]$ logstash -f log.conf Settings: Default pipeline workers: 4Pipeline main started{       "message" => "11111111111",      "@version" => "1",LogStash::Inputs::File 只是在進(jìn)程運(yùn)行的注冊階段初始化一個(gè)FileWatch對象。所以它不能支持類型fluentd 那樣的path=>"2.1.3 TCP 輸入 未來你可能會用Redis 服務(wù)器或者其他的消息隊(duì)列系統(tǒng)來作為Logstash Broker的角色。不過Logstash 其實(shí)也有自己的TCP/UDP 插件,在臨時(shí)任務(wù)的時(shí)候,也算能用,尤其是測試環(huán)境。[elk@Vsftp logstash]$ cat tcp.conf input { tcp {   port =>8888   mode=>"server"  ssl_enable =>false }}output {        stdout {                        codec => rubydebug                }}[elk@Vsftp logstash]$ logstash -f tcp.conf Settings: Default pipeline workers: 4Pipeline main started{       "message" => "9999999999",      "@version" => "1",    "@timestamp" => "2017-02-08T03:02:43.746Z",          "host" => "127.0.0.1",          "port" => 47187}{       "message" => "000000000",      "@version" => "1",    "@timestamp" => "2017-02-08T03:02:43.747Z",          "host" => "127.0.0.1",          "port" => 47187}Vsftp:/var/log#  nc 127.0.0.1 8888 < mmVsftp:/var/log# cat mm99999999990000000002.1.4 syslog輸入: syslog 可能是運(yùn)維領(lǐng)域最流行的數(shù)據(jù)傳輸協(xié)議了,當(dāng)你想從設(shè)備上收集系統(tǒng)日志的時(shí)候,syslog 應(yīng)該會是你第一選擇。尤其是網(wǎng)絡(luò)設(shè)備介紹 如何把Logstash 配置成一個(gè)syslog 服務(wù)器來接收數(shù)據(jù)。2.2 編解碼配置:Codec 是Logstash 從1.3.0 開始引入的概念(Codec 來自Coder/decoder 兩個(gè)單詞的首字母縮寫)我們在第一個(gè)"Hello World" 用例就已經(jīng)用過Codec了 rubydebug就是一種Codec 雖然它一般只會在stdout 插件中,作為配置測試或者調(diào)試的工具2.2.1 JSON 編解碼:2.2.2  多行事件編碼有些時(shí)候,應(yīng)用程序調(diào)試日志會包含非常豐富的內(nèi)容,為一個(gè)事件打印出很多行內(nèi)容。這種日志通常都很難通過命令行解析的方式做分析而Logstash 正為此準(zhǔn)備好了 codec/multiline 插件！當(dāng)然,multiline插件也可以用于其他類似的堆棧信息Vsftp:/home/elk/logstash# cat multi.conf input {  stdin {  codec=>multiline {  pattern =>"^/["  negate =>true what=>"PRevious"  }}}output {        stdout {                        codec => rubydebug                }}[elk@Vsftp logstash]$ logstash -f multi.conf Settings: Default pipeline workers: 4Pipeline main started[Aug/02/03 11:45:27] aaaabbbbcccc[Aug/02/03 11:45:27]  998877{    "@timestamp" => "2017-02-08T05:27:07.442Z",       "message" => "[Aug/02/03 11:45:27] aaaa/nbbbb/ncccc",      "@version" => "1",          "tags" => [        [0] "multiline"    ],          "host" => "Vsftp"}其實(shí)這個(gè)插件的原理很簡單,就是把當(dāng)前行的數(shù)據(jù)添加到前面一行后面,直到新進(jìn)的當(dāng)前行匹配^/[正則為止。2.3 過濾器配置:2.3.1 date時(shí)間處理之前章節(jié)已經(jīng)提過,logstash-filter-date插件可以用來轉(zhuǎn)換你的日志記錄中的時(shí)間字符串,變成LogStash::Timestamp 對象,然后轉(zhuǎn)存到@timestamp 字段里因?yàn)樵谏院蟮膌ogstash-outout-elasticsearch 中常用的%{+YYYY.MM.dd}這種寫法必須讀取@timestamp 數(shù)據(jù)%{TIMESTAMP_ISO8601:time}  匹配如下時(shí)間格式:2011-04-18 08:20:112011-04-18 08:20:11,108[elk@Vsftp logstash]$ logstash -f stdin.conf Settings: Default pipeline workers: 4Pipeline main startedaaaaabbbbbccccc{"message":"aaaaabbbbbccccc","@version":"1","@timestamp":"2017-02-08T05:44:44.165Z","type":"std","key11":"value22","tags":["add","xxyy"],"host":"Vsftp"}