Logstash組件詳解(input、codec、filter、output)

logstash組件詳解

logstash的概念及特點。

概念：logstash是一個數據采集、加工處理以及傳輸(輸出)的工具。
特點：

　　- 所有類型的數據集中處理
　　- 不同模式和格式數據的正常化
　　- 自定義日志格式的迅速擴展
　　- 為自定義數據源輕松添加插件

logstash安裝配置。

①.下載安裝
[root@node1 ~]# wget https://download.elastic.co/logstash/logstash/packages/centos/logstash-7.4.0-1.noarch.rpm
[root@node1 ~]# rpm -ivh logstash-7.4.0-1.noarch.rpm

②.簡單測試
logstash,啟動后輸入"hello,xkops"。
[root@node1 ~]# /opt/logstash/bin/logstash -e 'input{ stdin{} } output{ stdout{} }'
*提示：如果輸出"hello,xkops",則證明logstash啟動成功。

③.以服務啟動方式。
[root@node1 ~]# service logstash start
總結：logstash三種啟動方式，-e sting類型啟動，-f 指定配置文件啟動，服務啟動。

logstash配置語句詳解。

logstash配置文件包含三個配置部分，分別為：input{}、filter{}、output{}。
{} 定義區域，區域內可以定義一個或多個插件，通過插件對數據進行收集，加工處理，輸出。

數據類型：

• 布爾值類型: ssl_enable => true
• 字節類型: bytes => "1MiB"
• 字符串類型: name => "xkops"
• 數值類型: port => 22
• 數組: match => ["datetime","UNIX"]
• 哈希: options => {key1 => "value1",key2 => "value2"}
• 編碼解碼: codec => "json"
• 路徑: file_path => "/tmp/filename"

條件判斷：

  等於: ==
  不等於: !=
  小於: <
  大於: >
  小於等於: <=
  大於等於: >=
  匹配正則: =~
  不匹配正則: !~
  包含: in
  不包含: not in 
  與:	and
  或:	or
  非與: nand
  非或:	xor
  復合表達式: ()
  取反符合: !()

logstash常用插件

類型：

• input類：也就是在input區域內定義使用的插件。
• codec類：定於處理數據格式，如plain,json，json_lines等格式。這個可以定義在input、output區域中。
• filter類：也就是在filter區域內定義使用的插件。
• output類：也就是在output區域內定義使用的插件。
• 各類插件地址：https://github.com/logstash-plugins

input類插件

input類插件，常用的插件：file、tcp、udp、syslog，beats等。

①.file插件：

file插件字段解釋：
codec =>             #可選項，默認是plain，可設置其他編碼方式。

discover_interval => #可選項，logstash多久檢查一下path下有新文件，默認15s。

exclude =>           #可選項，排除path下不想監聽的文件。

sincedb_path =>      #可選項，記錄文件以及文件讀取信息位置的數據文件。~/.sincedb_xxxx

sincedb_write_interval => #可選項，logstash多久寫一次sincedb文件，默認15s.

stat_interval =>          #可選項，logstash多久檢查一次被監聽文件的變化，默認1s。

start_position =>         #可選項，logstash從哪個位置讀取文件數據，默認從尾部，值為：end。初次導入，設置為：beginning。

path =>                   #必選項，配置文件路徑，可定義多個。

tags =>                   #可選項，在數據處理過程中，由具體的插件來添加或者刪除的標記。

type =>                   #可選項，自定義處理時間類型。比如nginxlog。

實例:

[root@node1 conf.d]# cat /tmp/file{1,2}.log
This is test file-plugin in file1.log
This is test file-plugin in file2.log
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat file.conf
input{
    file{
        start_position => "beginning"
        path => ["/tmp/file1.log","/tmp/file2.log"]
        type => 'filelog'
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f file.conf -t
Configuration OK
[root@node1 conf.d]# /opt/logstash/bin/logstash -f file.conf 
Settings: Default pipeline workers: 1
Pipeline main started
2019-07-16T02:50:07.410Z node1 This is test file-plugin in file1.log
2019-07-16T02:50:07.428Z node1 This is test file-plugin in file2.log
*提示：能夠輸出內容則證明file插件正常工作。

②.tcp插件:

tcp插件字段解釋：
add_field =>              #可選項，默認{}。
codec =>                  #可選項，默認plain。
data_timeout =>           #可選項，默認-1。
host =>                   #可選項，默認0.0.0.0。
mode =>                   #可選項，值為["server","client"]之一，默認為server。
port =>                   #必選，端口。
ssl_cacert =>             #可選項，定義相關路徑。
ssl_cert =>               #可選項，定義相關路徑。
ssl_enable =>             #可選項，默認為false。
ssl_key =>                #可選項，定義相關路徑。
ssl_key_passphrase =>     #可選項，默認nil
ssl_verify =>             #可選項，默認false。
tags =>                   #可選項
type =>                   #可選項

實例：

[root@node1 conf.d]# cat /tmp/tcp.log 
this is test tcp-plugin in tcp.log
send tcplog data
output tcplog data
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat tcp.conf
input{
    tcp{
        host => "127.0.0.1"
        port => 8888
        type => "tcplog"
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f tcp.conf
# 此時開啟另個終端，使用nc開啟一個tcp端口8888，並將數據推送到8888端口。
[root@node1 conf.d]# nc 127.0.0.1 8888 < /tmp/tcp.log 
# 提示：前一個終端如果出現數據，則證明tcp插件工作正常。

③udp插件：

udp插件字段解釋：
add_field =>         #可選項，默認{}。
host =>              #可選項，默認0.0.0.0。
queue_size =>        #默認2000
tags =>              #可選項
type =>              #可選項
workers =>           #默認為2

實例：

[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat udp.conf
input{
    udp{
        host => "127.0.0.1"
        port => 9999
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f udp.conf
# 打開另一終端執行如下腳本，並輸入內容："hello,udplog"。
[root@node1 conf.d]# cat /tmp/udpclient.py 

#/usr/bin/env python
import socket

host = "127.0.0.1"
port = 9999
file_input = raw_input("Please input udp log: ")
s = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
s.sendto(file_input,(host,port))

*提示：如果前一個終端收到日志，則證明udp插件工作正常。

④.syslog插件：

實例：

[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat syslog.conf
input{
    syslog{
        host => "127.0.0.1"
        port => 518
        type => "syslog"
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# echo '*.* @@127.0.0.1:518' >> /etc/rsyslog.conf
[root@node1 conf.d]# /etc/init.d/rsyslog restart
關閉系統日志記錄器： [確定]
啟動系統日志記錄器： [確定]
[root@node1 conf.d]# /opt/logstash/bin/logstash -f syslog.conf 
使用logger命令向系統寫入日志：
[root@node1 conf.d]# logger

# 提示：此處隨便輸入內容，查看前一個終端是否會有內容輸出，如果輸出則證明syslog插件工作正常。

codec類插件

codec類插件，常用的插件：plain、json、json_lines、rubydebug、multiline等。
①.plain插件：

實例：
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat plain.conf
input{
    stdin{
        codec => "plain"
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f plain.conf 
輸入信息查看輸出。

②.json插件：

實例：
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat json.conf
input{
    stdin{}
}
output{
    stdout{
        codec => "json"
    }
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f json.conf 
輸入信息查看輸出。

③.json_lines插件：(json文本過長時使用)

實例：
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat jsonlines.conf
input{
    tcp{
        host => "127.0.0.1"
        port => 8888
        codec => "json_lines"
    }
}
output{
    stdout{}
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f jsonlines.conf 
啟動一個新的終端，執行如下命令。
[root@node1 conf.d]# cat /tmp/jsonlines.txt 
You run a price alerting platform which allows price-savvy customers to specify a rule like "I am interested in buying a specific electronic gadget and I want to be notified if the price of gadget falls below $X from any vendor within the next month". In this case you can scrape vendor prices, push them into Elasticsearch and use its reverse-search (Percolator) capability to match price movements against customer queries and eventually push the alerts out to the customer once matches are found.
[root@node1 conf.d]# nc 127.0.0.1 8888 < /tmp/jsonlines.txt 
# 提示：觀察前一個終端的輸出，如果正常輸出，則json_lines插件工作正常。

④.rubedebug插件:

實例：

[root@node1 conf.d]# cat rubydebug.conf
input{
    stdin{
        codec => "json"
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f rubydebug.conf
輸入json串查看輸出效果。
json串：{"name":"xkops","age":"25"}

⑤.multiline插件：(處理錯誤日志)

multiline插件字段：
charset =>           #字符編碼，可選
max_bytes =>         #bytes類型，設置最大的字節數，可選
max_lines =>         #number類型，設置最大的行數，默認是500行，可選
multiline_tag =>     #string類型，設置一個事件標簽，默認是"multiline" ，可選
pattern =>           #string 類型，設置匹配的正則表達式 ，必選 
patterns_dir =>      #array類型，可以設置多個正則表達式，可選
negate =>            #boolean類型，設置正向匹配還是反向匹配，默認是false，可選
what =>              #設置未匹配的內容是向前合並還是向后合並，previous, next 兩個值選擇，必選

錯誤日志：
[16-07-2016 22:54:01] PHP warning: unknown exception in /xxx/test/index.php:99
111111111111111111
222222222222222222
[16-07-2016 23:19:43] PHP warning: unknown exception in /xxx/test/index.php:93
[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat codecmultilines.conf

input{
    stdin{
        codec => multiline{
        pattern => "^\["
        negate => true
        what => "previous"
        }
    }
}
output{
    stdout{}
}

[root@node1 conf.d]# /opt/logstash/bin/logstash -f codecmultilines.conf 
# 提示：輸入上述錯誤日志查看輸出。

filter類插件

filter插件，常用的filter插件：json、grok等。

①.json插件：

add_field =>     #hash（可選項），默認{}
add_tag =>       #array（可選項），默認[]
remove_field =>  #array（可選項），默認[]
remove_tag =>    #array（可選項），默認[]
source =>        #string（必選項）
target =>        #string（可選項）

實例：

[root@node1 conf.d]# pwd
/etc/logstash/conf.d
[root@node1 conf.d]# cat filterjson.conf

input{
    stdin{}
}
filter{
    json{
        source => "message"
        #target => "content"
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

[root@node1 conf.d]# /opt/logstash/bin/logstash -f filterjson.conf 
輸入兩個串，查看輸出。
{"name":"xkops","age":"25"}
name xkops

②.grok插件

grok插件解析各種非結構化的日志數據插件。
grok有豐富的patterns,查看方式：

cat /opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns/grok-patterns
https://github.com/logstash-plugins/logstash-patterns-core/blob/master/patterns/grok-patterns

實例：

輸入的元數據內容：
20.3.1.3 GET /xkops/index.html 8838 1.323
[root@node1 conf.d]# pwd
/etc/logstash/conf.d

[root@node1 conf.d]# cat filtergrok.conf
input{
    stdin{}
}
filter{
    grok{
        match => ["message","%{IP:ip} %{WORD:method} %{URIPATH:uri} %{NUMBER:bytes} %{NUMBER:duration}"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}
[root@node1 conf.d]# /opt/logstash/bin/logstash -f filtergrok.conf
# 提示：輸入上述元數據查看輸出。

提示：grok在線工具使用：https://grokdebug.herokuapp.com/

③.kv插件

kv插件解析處理key-value鍵值對數據

output類插件

output插件：

• ①.file插件
• ②.tcp/udp插件
• ③.redis/kfaka
• ④.elasticsearch

附錄：
redis配置：

input{
    redis{
        host => 'redis-server'
        port => '6379'
        data_type => 'list'
        key => 'lb'
        codec => 'json'
    }
}