官方文檔logstash2.3 document：
https://www.elastic.co/guide/en/logstash/current/index.html
Logstash 是有管道輸送能力的開(kāi)源數據收集引擎。它可以動(dòng)態(tài)地從分散的數據源收集數據，并且標準化數據輸送到你選擇的目的地。它是一款日志而不僅限于日志的搜集處理框架，將分散多樣的數據搜集自定義處理并輸出到指定位置。
Logstash使用管道方式進(jìn)行日志的搜集處理和輸出。有點(diǎn)類(lèi)似*NIX系統的管道命令xxx | ccc | ddd，xxx執行完了會(huì )執行ccc，然后執行ddd。
在logstash中，包括了三個(gè)階段:
輸入input -->處理filter（不是必須的）--> 輸出output
每個(gè)階段都由很多的插件配合工作，比如file、elasticsearch、redis等等。
每個(gè)階段也可以指定多種方式，比如輸出既可以輸出到elasticsearch中，也可以指定到stdout在控制臺打印。
由于這種插件式的組織方式，使得logstash變得易于擴展和定制。
Logstash運行僅僅依賴(lài)java運行環(huán)境(jre)
若logstash的安裝目錄在${logstashHome}，進(jìn)入安裝目錄可在控制臺輸入命令：bin/logstash -e 'input { stdin { } } output { stdout {} }'
我們現在可以在命令行下輸入一些字符，然后我們將看到logstash的輸出內容：
hello world
2013-11-21T01:22:14.405+0000 0.0.0.0 hello world
使用-e參數在命令行中指定配置是很一種方式，不過(guò)如果需要配置更多設置則需要很長(cháng)的內容。這種情況，我們首先創(chuàng )建一個(gè)配置文件，并且指定logstash使用這個(gè)配置文件。
標準配置為文件含有input{} filter{}和output{}三部分，如配置一個(gè)配置文件sampl.conf
input {
file {
path => "/path/to/logstash-tutorial.log"
start_position => beginning
ignore_older => 0
}}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}}
output{
elasticsearch {hosts=>localhost:9020}
Stdout{Codec=>”rubydebug” }
}
/path/to/是配置文件在你的文件系統的具體位置
logstash-tutorial.log為Apache日志內容如下：
83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] "GET /presentations/logstash-monitorama-2013/images/kibana-search.png
HTTP/1.1" 200 203023 "http://semicomplete.com/presentations/logstash-monitorama-2013/" "Mozilla/5.0 (Macintosh; Intel
Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"
在命令行輸入：
bin/logstash -f first-pipeline.conf --configtest查證配置文件有無(wú)錯誤
然后運行logstash調用配置文件
bin/logstash -f first-pipeline.conf
在命令行有輸出JSON格式：
{"clientip" : "83.149.9.216",
"ident" : ,
"auth" : ,
"timestamp" : "04/Jan/2015:05:13:42 +0000",
"verb" : "GET",
"request" :"/presentations/logstash-monitorama-2013/images/kibana-search.png",
"httpversion" : "HTTP/1.1",
"response" : "200",
"bytes" : "203023",
"referrer" :"http://semicomplete.com/presentations/logstash-monitorama-2013/",
"agent" : "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36"}
命令行參數
Logstash 提供了一個(gè) shell腳本叫 logstash方便快速運行。它支持一下參數：
-e
意即執行。我們在 "Hello World" 的時(shí)候已經(jīng)用過(guò)這個(gè)參數了。事實(shí)上你可以不寫(xiě)任何具體配置，直接運行bin/logstash -e '' 達到相同效果。這個(gè)參數的默認值是下面這樣：
input {
stdin { }
}
output {
stdout { }
}
--config 或 -f
意即文件。真實(shí)運用中，我們會(huì )寫(xiě)很長(cháng)的配置，甚至可能超過(guò) shell 所能支持的1024 個(gè)字符長(cháng)度。所以我們必把配置固化到文件里，然后通過(guò)bin/logstash -f agent.conf 這樣的形式來(lái)運行。
此外，logstash 還提供一個(gè)方便我們規劃和書(shū)寫(xiě)配置的小功能。你可以直接用bin/logstash -f /etc/logstash.d/ 來(lái)運行。logstash會(huì )自動(dòng)讀取 /etc/logstash.d/目錄下所有的文本文件，然后在自己內存里拼接成一個(gè)完整的大配置文件，再去執行。
--configtest 或 -t
意即測試。用來(lái)測試 Logstash 讀取到的配置文件語(yǔ)法是否能正常解析。Logstash配置語(yǔ)法是用 grammar.treetop定義的。尤其是使用了上一條提到的讀取目錄方式的讀者，尤其要提前測試。
--log 或 -l
意即日志。Logstash 默認輸出日志到標準錯誤。生產(chǎn)環(huán)境下你可以通過(guò)bin/logstash -l logs/logstash.log 命令來(lái)統一存儲日志。
--filterworkers或 -w
意即工作線(xiàn)程。Logstash 會(huì )運行多個(gè)線(xiàn)程。你可以用bin/logstash -w 5 這樣的方式強制Logstash 為過(guò)濾插件運行5 個(gè)線(xiàn)程。
注意：Logstash目前還不支持輸入插件的多線(xiàn)程。而輸出插件的多線(xiàn)程需要在配置內部設置，這個(gè)命令行參數只是用來(lái)設置過(guò)濾插件的！
提示：Logstash 目前不支持對過(guò)濾器線(xiàn)程的監測管理。如果filterworker 掛掉，Logstash會(huì )處于一個(gè)無(wú) filter的僵死狀態(tài)。這種情況在使用 filter/ruby自己寫(xiě)代碼時(shí)非常需要注意，很容易碰上 NoMethodError: undefined method '*' for nil:NilClass錯誤。需要妥善處理，提前判斷。
--pluginpath或 -P
可以寫(xiě)自己的插件，然后用 bin/logstash --pluginpath /path/to/own/plugins 加載它們。
--verbose
輸出一定的調試日志。
小貼士：如果你使用的 Logstash 版本低于1.3.0，你只能用bin/logstash -v 來(lái)代替。
--debug
輸出更多的調試日志。
小貼士：如果你使用的 Logstash 版本低于1.3.0，你只能用bin/logstash -vv 來(lái)代替
CTRL-D 退出運行中的logstashshell
創(chuàng )建配置文件，在配置文件的各個(gè)部分（input，filter，output）制定并配置你要使用的插件
Logstash 支持少量的數據值類(lèi)型：
bool
debug => true
string
host => "hostname"
number
port => 514
array
match => ["datetime", "UNIX", "ISO8601"]
hash
options => {
key1 => "value1",
key2 => "value2"
}
ü
1. Filed references ,sprintf format, conditionals 作用在filter和output階段
Filed reference 是filed嵌套，如下面事件
{ "agent": "Mozilla/5.0 (compatible; MSIE 9.0)",
"ip": "192.168.24.44",
"request": "/index.html"
"response": {
"status": 200,
"bytes": 52353 },
"ua": {
"os": "Windows 7" }
}
你可以定義[ua][os]來(lái)引用os字段
Sprintf format
（1） 根據appache日志狀態(tài)碼計數
output {
statsd {
increment => "apache.%{[response][status]}"
} }
（2）output{ file { path => "/var/log/%{type}.%{+yyyy.MM.dd.HH}" } }
Conditional
在某種特定條件下過(guò)濾，輸出event（if ,else if,else可以嵌套）
comparison operators比較操作:
equality: ==, !=, <, >, <=, >=
regexp: =~, !~ (checks a pattern on the right against a string value on the left)
inclusion: in, not in
boolean operators布爾操作：
and, or, nand, xor
一元操作:
!
僅讓grok成功的字段索引到elasticsearch中
output {
if "_grokparsefailure" not in[tags] {
elasticsearch { ... } }
}
2. @matedata字段
在output階段不作為event的一部分輸出，可以很好的作為條件判斷，或是利用字段引用（filed reference）
和sprintf format 來(lái)擴展或是創(chuàng )建字段
如果要顯示輸出@matedata可設置stdout { codec => rubydebug { metadata => true } }
Similarly, you can use conditionals to direct events to particular outputs. For example, you could:
l alert nagios of any apache events with status 5xx
l record any 4xx status to Elasticsearch
l record all status code hits via statsd
To tell nagios about any http event that has a 5xx status code, you first need to check the value of the typefield. If it’s apache, then you can check to see if the status field contains a 5xx error. If it is, send it to nagios. If it isn’t a 5xx error, check to see if the status field contains a 4xx error. If so, send it to Elasticsearch. Finally, send all apache status codes to statsd no matter what the status field contains:
output {
if [type] == "apache" {
if [status] =~ /^5\d\d/ {
nagios { ... }
} else if [status] =~ /^4\d\d/ {
elasticsearch { ... }
}
statsd { increment => "apache.%{status}" }
}}
l Input-file-plugin
Path=>”/var/log/*” 作為輸入的文件的完全路徑
Excude =>”*.gz” 不用匹配的文件，非=完全路徑
ignore_older => 86400 (s)時(shí)間間隔最后被修改的被忽略
max_open_files => number
add_field=> {} hash值，eg：add_field => {“addfield”=>”content”}
close =>”plain”
delimiter=> “\n” 默認值，定界符
discover_interval => 15（s） 每隔多久在文件路徑模式下發(fā)現新的文件 (interval 時(shí)間間隔)
sincedb_path 默認$HOME/.sincedb ,將被寫(xiě)入硬盤(pán)的sincedb_database文件的路徑，默認是匹配$HOME/.sincedb*的路徑
sincedb_write_interval =>15（s）多久將被監控日志文件的當前位置寫(xiě)一個(gè)since database一次
start_position => “end” 還可”beginning” 從哪開(kāi)始讀取文件，僅當文件是新建的或是從前沒(méi)見(jiàn)過(guò)的，否則失效
start_interval => 設置多久看一次文件是否修改，有新的日志行，
tags =>”arrary” 添加時(shí)間tag可任意多的
type => “string”為input控制的所有事件添加一個(gè)type的字段， 在filter操作會(huì )用到該字段
l Input-stdin-plugin
是 logstash 里最簡(jiǎn)單和基礎的插件
l Input-jdbc-plugin
schedule：設置監聽(tīng)間隔?？梢栽O置每隔多久監聽(tīng)一次什么的。具體參考官方文
statement_filepath： 執行的sql文件路徑+名稱(chēng)
數據庫的表的一條記錄為一個(gè)事件
下面配置文件監控的是遠程數據庫，注意的是，MySQL數據庫配置是要設為允許遠程訪(fǎng)問(wèn)
l Filter-grok-plugin
解析任意文本并且結構化他們。grok目前是logstash中最好的解析非結構化日志并且結構化他們的工具。這個(gè)工具非常適合syslog、apache log、mysql log之類(lèi)的人們可讀日志的解析
其配置參數如下：
add_field => ... # hash (optional), default: {}
add_tag => ... # array (optional), default: []
break_on_match => ... # boolean (optional), default: true
設為true時(shí)grok的第一個(gè)成功匹配將結束fliter操作，若想完成所有的匹配應該設為false
drop_if_match => ... # boolean (optional), default: false
keep_empty_captures => ... # boolean (optional), default: false
match => ... # hash (optional), default: {}
named_captures_only => ... # boolean (optional), default: true存儲時(shí)間有命名的捕獲
overwrite => ... # array (optional), default: []
允許字段重寫(xiě)，字段已經(jīng)存在用grok匹配的信息重寫(xiě)該字段
grok{
match=>{"message"=>"%{SYSLOGBASE"}%{DATA:message}}
Overwrite=>[“message”] #重寫(xiě)message字段
}
patterns_dir => ... # array (optional), default: []
remove_field => ... # array (optional), default: []
remove_tag => ... # array (optional), default: []
tag_on_failure => ... # array (optional), default: ["_grokparsefailure"]表示沒(méi)有成功匹配
l Filter-date-plugin
從字段中解析日期，并將這個(gè)日期date或timestamp作為事件的@timestamp
這個(gè)字段在整理事件和回填舊數據是特別有用
add-field =>{}
remove-field =>[]
add-tag =>[]
remove-tag =>[]
local=>””
match=>[“fieldname”,”format” “format”]
Format 有多種匹配格式的可能：MMM dd YYYY HH:mm:ss
MMM d YYYY HH:mm:ss
dd/MMM/yyyy:HH:mm:ss Z
ISO8601 (如：2011-04-19T03:44:01.1032)
Periodic_flush=>false
Tag_on_failure=>[“_datepardefailure”]沒(méi)有成功匹配是添加字段，此處為默認值
Target=>“@timestamp”將匹配的timestamp字段放在指定的字段 默認是@timestamp
l Filter-syslog-plugin
Syslog對于Logstash是一個(gè)很長(cháng)用的配置，并且它有很好的表現(協(xié)議格式符合RFC3164)。Syslog實(shí)際上是UNIX的一個(gè)網(wǎng)絡(luò )日志 標準，由客戶(hù)端發(fā)送日志數據到本地文件或者日志服務(wù)器。在這個(gè)例子中，你根本不用建立syslog實(shí)例;我們通過(guò)命令行就可以實(shí)現一個(gè)syslog服務(wù)， 通過(guò)這個(gè)例子你將會(huì )看到發(fā)生什么。
首先，讓我們創(chuàng )建一個(gè)簡(jiǎn)單的配置文件來(lái)實(shí)現logstash+syslog，文件名是 logstash-syslog.conf
input {
tcp {
port => 5000
type => syslog
}
udp {
port => 5000
type => syslog
}
}
filter {
if [type] == "syslog" {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
syslog_pri { }
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
}
}
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
執行logstash：
bin/logstash -f logstash-syslog.conf
通常，需要一個(gè)客戶(hù)端鏈接到Logstash服務(wù)器上的5000端口然后發(fā)送日志數據。在這個(gè)簡(jiǎn)單的演示中我們簡(jiǎn)單的使用telnet鏈接到 logstash服務(wù)器發(fā)送日志數據(與之前例子中我們在命令行標準輸入狀態(tài)下發(fā)送日志數據類(lèi)似)。首先我們打開(kāi)一個(gè)新的shell窗口，然后輸入下面的 命令：
telnet localhost 5000
你可以復制粘貼下面的樣例信息(當然也可以使用其他字符，不過(guò)這樣可能會(huì )被grok filter不能正確的解析)：
Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154]
Dec 23 14:42:56 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' denied
Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)
Dec 22 18:28:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'.
之后你可以在你之前運行Logstash的窗口中看到輸出結果，信息被處理和解析！
{
"message" => "Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
"@timestamp" => "2013-12-23T22:30:01.000Z",
"@version" => "1",
"type" => "syslog",
"host" => "0:0:0:0:0:0:0:1:52617",
"syslog_timestamp" => "Dec 23 14:30:01",
"syslog_hostname" => "louis",
"syslog_program" => "CRON",
"syslog_pid" => "619",
"syslog_message" => "(www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
"received_at" => "2013-12-23 22:49:22 UTC",
"received_from" => "0:0:0:0:0:0:0:1:52617",
"syslog_severity_code" => 5,
"syslog_facility_code" => 1,
"syslog_facility" => "user-level",
"syslog_severity" => "notice"
}
l Filter-mutate-plugin
替換重命名移除事件中的字段
add_field(hash) ,add_tag(array) ,remove_field(array),remove_tag(array)
convert=>{} 改變字段類(lèi)型，若為array所有將轉換
eg：convert=> { “field_name” =>”integer”}
gsub => { “fieldname”,”/”,”_” } 將fielname字段值中所有”/”替換成 “_”
“fieldname2”,”[\\?#-]”,”.” } 將fielname2字段值中所有”\?#-”替換成 “_”
Join=> {} 將數組用分隔符連接合并起來(lái)
“fieldname”=>”,”
strip=>[] 去除字段首尾的空白格
Lowercase=> [“fieldname”] 將字符串轉換后才能小寫(xiě) uppercase =>[]
merge => { } 合并兩個(gè)字段，array+string或string+string
remove => [] 移除一個(gè)或多個(gè)字段， 新版本可能不可用
remove_field => {} 如果過(guò)濾器成功，從事件移除任意字段，可%{fieldname}
remove_tag =>[]
rename 重命名一個(gè)或多個(gè)字段，hash rename=> {“HOSTIP”=>”client_ip”}
replace => { } 給字段賦新值，update=> { } 用一個(gè)新值更新字段
split=> { “fieldname”=>””},通過(guò)一個(gè)分隔字符將字段分離成一個(gè)array，僅string
注： mutate在有json插件的情況下要放在json之后，可修改json轉換后結構data中的字段
l Filter-geoip-plugin
查詢(xún)IP地址，從中得到物理地址信息并添加到日志中，（該插件配置必須在grok配置后面）
指定包含要查詢(xún)的IP的字段的名稱(chēng)（name）
Geoip{
Source=>”clientip” ->包含ip或者主機名，如果是array只第一個(gè)可用
}
可直接分析一個(gè)ip的經(jīng)緯度，歸屬地址信息
target=>”geoip”將解析得到的字段信息存放到指定字段中
database=>”pathofGeoipdatabase” logstash使用的geoip數據庫，默認的是GeoLiteCity數據庫
l Filter-KV-plugin
幫助自動(dòng)解析message，包含有foo=bar形式的自動(dòng)解析為foo：bar字段
適合像postfix，iptables和其他趨向key=value語(yǔ)法的日志
remove-field =>[]
add-tag =>[]
remove-tag =>[]
tagret=>””
periodic-flush=flase/true
allow_duplicate_values=>“true”允許重復key/value對兒
eg：當有兩個(gè)“from”=me“from”=m設為false 只允許唯一一個(gè)
default-keys=>{}向事件中添加健值對哈希表，以防source中不包含這些應有的字段
exclude-keys=>[]去除事件中不必要的健值對，填key
field-split=>“”字符串中的一個(gè)字符，用以劃分出可解析的key-value
eg：pin=123&d=123&e=foo則field-split=>“&”
include-keys=>[]保留事件中需要的鍵值，去除剩余的
include-brackets=>“true”值不包含括號（有去除）
prefix=>“”在所有key前加一個(gè)字符串
trim=>“”
trimkey=>“”去除key=value中的字符 如
value -split=>“=”
上述操作僅對原event中field有效對default-keys添加的字段無(wú)效。
l Filter-multiline -plugin
不是線(xiàn)程安全的 不可解決多流信息
mulyiline{
type=>
pattern=>“”匹配模式
negate=>“false”true=>不匹配的進(jìn)行what
what=>“previous/next”
}
“^\s”空格開(kāi)頭“\\$”反斜線(xiàn)符
source=>“message”
allow-dulplicate=>true 允許重復值在event中
pattern-dir=>array logstash自帶的匹配模式
NUMBER\d+
l Fliter-ruby-plugin
執行ruby代碼
init=>“”在logstash啟動(dòng)時(shí)執行的任何代碼
code=>“”為每個(gè)事件執行的代碼要有一個(gè)指代事件 本身的event變量可用
add-field =>{}
remove-field =>[]
add-tag =>[]
remove-tag =>[]
periodic-flush=>“false”
l Filter-json-plugin
是一個(gè)轉換過(guò)濾插件，對一個(gè)包含json字格式數據的字段，可擴展成一個(gè)數據結構
add-field =>{}
remove-field =>[]
add-tag =>[]
remove-tag =>[]
source=>“message”指定要轉換的字段（是Json格式） 默認是message字段
tagret=>”” 指定轉換后數據的字段
輸出插件統一具有一個(gè)參數是 workers。Logstash為輸出做了多線(xiàn)程的準備。
與input-file不同 這里可用sprintf format格式自定義輸出到文件路徑
path=>’’/path/to/configure file/’
message_format=>”%{message}”
默認參 是按json形式輸出整個(gè)event的數據，設為%{message}是保存按照日志的原始格式保存（前提是filter未修改刪除message字段）
Worker=>1 output 的worker數目
gzip=>false 如果設為true，亂碼輸出
create_if_delete=>true 輸出file如果不存在被刪除 就自動(dòng)新建一個(gè)
Codec=>”json_lines”編碼格式 可設為rubydebug
l Output-stdout-plugin
標準輸出插件，直接將處理的數據輸出到命令行，是最基礎和簡(jiǎn)單的輸出插件
Codec=>”json_lines”編碼格式 可設為rubydebug
單就 outputs/stdout 插件來(lái)說(shuō)，其最重要和常見(jiàn)的用途就是調試。所以在不太有效的時(shí)候，加上命令行參數-vv 運行，查看更多詳細調試信息。
l Output-elasticsearch-plugin
將數據輸出到elasticsearch存放，方便搜索分析處理
hosts=>[host:port]
這部分會(huì )專(zhuān)門(mén)發(fā)一篇博客，最近正在學(xué)習
聯(lián)系客服
