營銷助手appseo網(wǎng)絡(luò)營銷的技術(shù)
logstash插件
輸入插件(input)
Input:輸入插件。
Input plugins | Logstash Reference [8.11] | Elastic
-
所有輸入插件都支持的配置選項(xiàng)
Setting | Input type | Required | Default | Description |
---|---|---|---|---|
add_field | hash | No | {} | 添加一個字段到一個事件 |
codec | codec | No | plain | 用于輸入數(shù)據(jù)的編解碼器 |
enable_metric | boolean | No | true | |
id | string | No | 添加一個ID插件配置,如果沒有指定ID,則Logstash將生成一個ID。強(qiáng)烈建議配置此ID,當(dāng)兩個或多個相同類型的插件時,這個非常有用的。例如,有兩個文件輸入,添加命名標(biāo)識有助于監(jiān)視 | |
tags | array | No | 添加任意數(shù)量的標(biāo)簽,有助于后期處理 | |
type | string | No | 為輸入處理的所有事件添加一個字段,自已隨便定義,比如linux系統(tǒng)日志,定義為syslog |
stdin
-
標(biāo)準(zhǔn)輸入
# cat /etc/logstash/config.d/stdtest.conf input {stdin { ?} } filter { ? } output {stdout {} } ?
file
-
從文件中讀取內(nèi)容
File input plugin | Logstash Reference [8.11] | Elastic
Setting | Input type | Required | Default | Description |
---|---|---|---|---|
close_older | number | No | 3600 | 單位秒,打開文件多長時間關(guān)閉 |
delimiter | string | No | \n | 每行分隔符 |
discover_interval | number | No | 15 | 單位秒,多長時間檢查一次path選項(xiàng)是否有新文件 |
exclude | array | No | 排除監(jiān)聽的文件,跟path一樣,支持通配符 | |
max_open_files | number | No | 打開文件最大數(shù)量 | |
path | array | YES | 輸入文件的路徑,可以使用通配符 例如/var/log/*/.log,則會遞歸搜索 | |
sincedb_path | string | No | sincedb數(shù)據(jù)庫文件的路徑,用于記錄被監(jiān)控的日志文件當(dāng)前位置 | |
sincedb_write_interval | number | No | 15 | 單位秒,被監(jiān)控日志文件當(dāng)前位置寫入數(shù)據(jù)庫的頻率 |
start_position | string, one of ["beginning", "end"] | No | end | 指定從什么位置開始讀取文件:開頭或結(jié)尾。默認(rèn)從結(jié)尾開始,如果要想導(dǎo)入舊數(shù)據(jù),將其設(shè)置為begin。如果sincedb記錄了此文件位置,那么此選項(xiàng)不起作用 |
stat_interval | number | No | 1 | 單位秒,統(tǒng)計文件的頻率,判斷是否被修改。增加此值會減少系統(tǒng)調(diào)用次數(shù)。 |
# cat /etc/logstash/conf.d/filetest.conf input {file {path => "/var/log/messages"} } filter { } output {stdout {} }
TCP
-
通過TCP套接字讀取事件,即接收數(shù)據(jù)。與標(biāo)準(zhǔn)輸入和文件輸入一樣,每個事件都被定位一行文本。
# cat /etc/logstash/conf.d/tcptest.conf input {tcp {port => 12345} } filter { ? } output {stdout{} }
在其他主機(jī)上安裝nc工具,對logstash發(fā)送信息,即可被讀取出來。
[root@vm4 ~]# yum -y install nc [root@vm4 ~]# nc 10.1.1.13 12345 haha ? ? 在vm3上驗(yàn)證查看 {"@version" => "1","@timestamp" => 2019-07-02T15:28:00.301Z,"port" => 33778,"type" => "nc","message" => "haha","host" => "vm4.cluster.com" }
Beats
-
從Elastic Beats框架接收事件
logstash配置文件 ? # cat /etc/logstash/conf.d/filebeattest.conf input {beats {port => 5044host => "0.0.0.0"} }filter {} ? output {stdout { } }
filebeat配置文件 ? filebeat.prospectors:- type: logpaths:- /var/log/messagestags: ["system-log","123"]fields:level: debug ? output.logstash:hosts: ['127.0.0.1:5044'] ?
過濾插件(filter)
參考: Filter plugins | Logstash Reference [8.11] | Elastic
Filter:過濾,將日志格式化。
有豐富的過濾插件:
-
Grok正則捕獲
-
date時間處理
-
JSON編解碼
-
數(shù)據(jù)修改Mutate
-
geoip等。
所有的過濾器插件都支持以下配置選項(xiàng):
Setting | Input type | Required | Default | Description |
---|---|---|---|---|
add_field | hash | No | {} | 如果過濾成功,添加任何field到這個事件。例如:add_field => [ "foo_%{somefield}", "Hello world, from %{host}" ],如果這個事件有一個字段somefiled,它的值是hello,那么我們會增加一個字段foo_hello,字段值則用%{host}代替。 |
add_tag | array | No | [] | 過濾成功會增加一個任意的標(biāo)簽到事件例如:add_tag => [ "foo_%{somefield}" ] |
enable_metric | boolean | No | true | |
id | string | No | ||
periodic_flush | boolean | No | false | 定期調(diào)用過濾器刷新方法 |
remove_field | array | No | [] | 過濾成功從該事件中移除任意filed。例:remove_field => [ "foo_%{somefield}" ] |
remove_tag | array | No | [] | 過濾成功從該事件中移除任意標(biāo)簽,例如:remove_tag => [ "foo_%{somefield}" ] |
json(關(guān)注)
-
JSON解析過濾器,接收一個JSON的字段,將其展開為Logstash事件中的實(shí)際數(shù)據(jù)結(jié)構(gòu)。
示例: 將原信息轉(zhuǎn)成一個大字段,key-value做成大字段中的小字段
# cat /etc/logstash/conf.d/jsontest.conf input {stdin {} } ? filter {json {source => "message"target => "content"} } ? output {stdout {} } ? ? 對標(biāo)準(zhǔn)輸入的內(nèi)容進(jìn)行json格式輸出 把輸出內(nèi)容定向到target指定的content ? [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/jsontest.conf
輸入測試數(shù)據(jù) ? {"ip":"10.1.1.1","hostname":"vm3.cluster.com"} ? 輸出測試數(shù)據(jù) ? {"content" => {"hostname" => "vm3.cluster.com","ip" => "10.1.1.1"},"@timestamp" => 2019-07-02T11:57:36.398Z,"@version" => "1","host" => "vm3.cluster.com","message" => "{\"ip\":\"10.1.1.1\",\"hostname\":\"vm3.cluster.com\"}" }
示例: 直接將原信息轉(zhuǎn)成各個字段
# cat /etc/logstash/conf.d/jsontest.conf input {stdin {} } ? filter {json {source => "message"} } ? output {stdout {} } ? ? ? [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/jsontest.conf
? 輸入測試數(shù)據(jù) ? {"ip":"10.1.1.1","hostname":"vm3.cluster.com"} ? 輸出測試數(shù)據(jù) ? ? {"port" => 39442,"@version" => "1","@timestamp" => 2019-09-19T09:07:03.800Z,"hostname" => "vm3.cluster.com","host" => "vm4.cluster.com","ip" => "10.1.1.1","message" => "{\"ip\":\"10.1.1.1\",\"hostname\":\"vm3.cluster.com\"}" } ?
kv
-
自動解析為key=value。
-
也可以任意字符串分割數(shù)據(jù)。
-
field_split 一串字符,指定分隔符分析鍵值對
URL查詢字符串拆分參數(shù)示例 # cat /etc/logstash/conf.d/kvtest.conf input {stdin {} } ? filter {kv {field_split => "&?"} } ? output {stdout { ?} } ? 文件中的列以&或?進(jìn)行分隔 ? 執(zhí)行 [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/kvtest.conf
輸入數(shù)據(jù) address=www.abc.com?pid=123&user=abc ? 輸出數(shù)據(jù) {"user" => "abc","@timestamp" => 2019-07-02T12:05:23.143Z,"host" => "vm3.cluster.com","@version" => "1","message" => "address=www.abc.com?pid=123&abc=user","address" => "www.abc.com","pid" => "123" }
使用正則也可以匹配 ? [root@vm3 bin]# cat /etc/logstash/conf.d/kvtest.conf input {stdin {} } ? filter {kv {field_split_pattern => ":+"} } ? output {stdout { ?} }
grok(關(guān)注)
-
grok是將非結(jié)構(gòu)化數(shù)據(jù)解析為結(jié)構(gòu)化
-
這個工具非常適于系統(tǒng)日志,mysql日志,其他Web服務(wù)器日志以及通常人類無法編寫任何日志的格式。
-
默認(rèn)情況下,Logstash附帶約120個模式。也可以添加自己的模式(patterns_dir)
-
模式后面對應(yīng)正則表達(dá)式
-
查看模式地址:https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns
-
包含字段如下
Setting | Input type | Required | Default | Description |
---|---|---|---|---|
break_on_match | boolean | No | true | |
keep_empty_captures | No | false | 如果true將空保留為事件字段 | |
match | hash | No | {} | 一個hash匹配字段=>值 |
named_captures_only | boolean | No | true | 如果true,只存儲 |
overwrite | array | No | [] | 覆蓋已存在的字段的值 |
pattern_definitions | No | {} | ||
patterns_dir | array | No | [] | 自定義模式 |
patterns_files_glob | string | No | * | Glob模式,用于匹配patterns_dir指定目錄中的模式文件 |
tag_on_failure | array | No | _grokparsefailure | tags沒有匹配成功時,將值附加到字段 |
tag_on_timeout | string | No | _groktimeout | 如果Grok正則表達(dá)式超時,則應(yīng)用標(biāo)記 |
timeout_millis | number | 30000 | 正則表達(dá)式超時時間 |
grok模式語法
格式:%{SYNTAX:SEMANTIC}
-
SYNTAX 模式名稱
-
SEMANTIC 匹配文本的標(biāo)識符
例如:%{NUMBER:duration} %{IP:client}
# vim /etc/logstash/conf.d/groktest.conf input {stdin {} } ? ? filter {grok {match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"}} } ? ? output {stdout { ?} } ? ? 虛構(gòu)http請求日志抽出有用的字段 55.3.244.1 GET /index.html 15824 0.043 ? ? 輸出結(jié)果 {"client" => "55.3.244.1","duration" => "0.043","message" => "55.3.244.1 GET /index.html 15824 0.043","method" => "GET","bytes" => "15824","@version" => "1","@timestamp" => 2019-07-03T12:24:47.596Z,"host" => "vm3.cluster.com","request" => "/index.html" } ?
自定義模式
如果默認(rèn)模式中沒有匹配的,可以自己寫正則表達(dá)式。
# vim /opt/patterns ID [0-9]{3,5} ? 配置文件中應(yīng)包含如下內(nèi)容 filter {grok {patterns_dir =>"/opt/patterns"match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{ID:id}" ? ? ? ? ? ? ?}} } ? ? 完整文件內(nèi)容 [root@vm3 ~]# cat /etc/logstash/conf.d/groktest.conf input {stdin {} } ? filter {grok {patterns_dir =>"/opt/patterns"match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} %{ID:id}"}} } ? output {stdout {} } ? #執(zhí)行 [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/groktest.conf ? 輸入測試數(shù)據(jù) 55.3.244.1 GET /index.html 15824 0.043 6666 ? 輸出測試數(shù)據(jù) {"client" => "55.3.244.1","host" => "vm3.cluster.com","request" => "/index.html","@timestamp" => 2019-07-02T12:34:11.906Z,"bytes" => "15824","method" => "GET","message" => "55.3.244.1 GET /index.html 15824 0.043 15BF7F3ABB","@version" => "1","id" => "666","duration" => "0.043" }
geoip(關(guān)注)
-
開源IP地址庫
-
GeoLite2 Free Geolocation Data | MaxMind Developer Portal
下載IP地址庫 [root@vm3 ~]# wget https://geolite.maxmind.com/download/geoip/database/GeoLite2-City.tar.gz ? [root@vm3 ~]# tar xf GeoLite2-City.tar.gz ? [root@vm3 ~]# cp GeoLite2-City_20190625/GeoLite2-City.mmdb /opt
# cat /etc/logstash/conf.d/geoiptest.conf ? ? ? ? ? ? ? ? ? ? ? ? ? input {stdin {} } ? filter {grok {match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"}}geoip {source => "client"database => "/opt/GeoLite2-City.mmdb"} } ? ? output {stdout { ?} } ? ? ? 執(zhí)行 [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/geoiptest.conf ? ? 輸入測試數(shù)據(jù) ? 202.106.0.20 GET /index.html 123 0.331 ? 輸出結(jié)果 ? {"method" => "GET","client" => "202.106.0.20","bytes" => "123","request" => "/index.html","geoip" => {"country_code2" => "CN","country_name" => "China","region_code" => "BJ","longitude" => 116.3883,"latitude" => 39.9289,"timezone" => "Asia/Shanghai","location" => {"lon" => 116.3883,"lat" => 39.9289},"country_code3" => "CN","ip" => "202.106.0.20","continent_code" => "AS","region_name" => "Beijing"},"duration" => "0.331","host" => "vm3.cluster.com","message" => "202.106.0.20 GET /index.html 123 0.331","@timestamp" => 2019-07-02T12:15:29.384Z,"@version" => "1" } ? ?
[root@vm3 bin]# cat /etc/logstash/conf.d/geoiptest2.conf input {stdin {} } ? filter {grok {match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" ?}}geoip {source => "client"database => "/opt/GeoLite2-City.mmdb"target => "geoip"fields => ["city_name", "country_code2", "country_name","region_name"]} } ? ? output {stdout {codec => rubydebug} } ? ? ? 執(zhí)行 [root@vm3 bin]# ./logstash --path.settings /etc/logstash -r -f /etc/logstash/conf.d/geoiptest2.conf ? ? 輸入測試數(shù)據(jù) ? 110.226.4.6 GET /home.html 518 0.247 ? 輸出結(jié)果 ? {"host" => "vm3.cluster.com","duration" => "0.247","request" => "/home.html","@version" => "1","client" => "110.226.4.6","message" => "110.226.4.6 GET /home.html 518 0.247","method" => "GET","bytes" => "518","@timestamp" => 2019-07-02T12:22:22.458Z,"geoip" => {"country_name" => "India","country_code2" => "IN"} } ?
輸出插件(output)
Output:輸出,輸出目標(biāo)可以是Stdout、ES、Redis、File、TCP等。
ES
Setting | Input type | Required | Default | Description |
---|---|---|---|---|
hosts | URL | No | ||
index | string | No | logstash-%{+YYYY.MM.dd} | 將事件寫入索引。默認(rèn)按日期劃分。 |
user | string | No | ES集群用戶 | |
password | password | No | ES集群密碼 |
input {file {path => ["/var/log/messages"]type => "system"tags => ["syslog","test"]start_position => "beginning"}file {path => ["/var/log/audit/audit.log"]type => "system"tags => ["auth","test"]start_position => "beginning"} } ? ? filter { ? } ? output {if [type] == "system" {if [tags][0] == "syslog" {elasticsearch {hosts ?=> ["http://es1:9200","http://es2:9200","http://es3:9200"]index ?=> "logstash-system-syslog-%{+YYYY.MM.dd}"}stdout { codec=> rubydebug }}else if [tags][0] == "auth" {elasticsearch {hosts ?=> ["http://es1:9200","http://es2:9200","http://es3:9200"]index ?=> "logstash-system-auth-%{+YYYY.MM.dd}"}stdout { codec=> rubydebug }}} } ?