在第九章節中,咱們已經安裝好Logstash組件了,而且啓動實例測試它的數據輸入和輸出,可是用的是最簡單的控制檯標準輸入和標準輸出,那這節咱們就來深刻的學習Logstash的詳細使用。php
咱們在上一節中演示了啓動Logstash的實例,其中咱們啓動的時候給Logstash腳本傳入了-e的參數,但實際上,Logstash的啓動參數有不少,咱們來看一下各個啓動參數的做用:html
./logstash -e "input {stdin {}} output {stdout {}}"
./logstash -f config/test.conf
./logstash -f config/test.conf -t
./logstash -f config/test.conf -l logs/test.log
./logstash-f config/test.conf -w 8
咱們剛剛知道,啓動參數能夠指定一個配置文件,那麼接下來就有必要來了解一下配置文件的結構:
Logstash經過{}來定義區域,區域內能夠定義插件,一個區域內能夠定義多個插件,以下:java
input { #標準輸入源插件 stdin { } #普通文件源插件 file { path => ["/var/log/*.log", "/var/log/message"] .... } ...... } filter { #grok過濾插件 grok { match => ["message", "%{HTTPDATE:logdate}"] ..... } #date過濾插件 date { match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"] ..... } ..... } output { stdout { } elasticsearch { hosts => ["127.0.0.1:9200"] .... } ..... }
咱們先大概瞭解一下配置文件的結構,接下來咱們再詳細看這些插件的配置。node
Logstash配置文件支持的數據類型有:
一、Boolean,例如:ssl_enable => true
二、Number,例如:port => 33
三、String,例如:name => 「Hello world」
四、hash,例如:options => {key1 => "value1", key2 => "value2"}
五、array,例如:match => ["datetime", "UNIX", "ISO8601"]mysql
Logstash數據流中的數據被稱之爲Event對象,Event以JSON結構構成,Event的屬性被稱之爲字段,若是你想在配置文件中引用這些字段,只須要把字段的名字寫在中括號[]
裏就好了,如[type]
,對於嵌套字段每層字段名稱都寫在[]
裏就能夠了,好比:[tags][type]
;除此以外,對於Logstash的arrag類型支持下標與倒序下表,如:[tags][type][0]
和[tags][type][-1]
如下的內容就是一個Event對象:git
{ "message" => "hello logstash", "@version" => "1", "@timestamp" => 2018-08-13T17:32:01.122Z, "host" => "localhost.localdomain" }
Logstash支持下面的操做符:
一、==(等於), !=(不等於), <(小於), >(大於), <=(小於等於), >=(大於等於)
二、=~(匹配正則), !~(不匹配正則)
三、in(包含), not in(不包含)
四、and(與), or(或), nand(非與), xor(非或)
五、()(複合表達式), !()(對複合表達式結果取反)
例如如下的條件判斷:web
if "_grokparsefailure" not in [tags] { } else if [status] !~ /^2\d\d/ or ( [url] == "/noc.gif" nand [geoip][city] != "beijing" ) { } else { }
Logstash支持引用系統環境變量,環境變量不存在時能夠設置默認值,例如:正則表達式
export TCP_PORT=12345 input { tcp { port => "${TCP_PORT:54321}" } }
在第九章中,咱們已經使用是標準輸入,以鍵盤的輸入數據做爲Logstash數據源,但實際上咱們也知道,Logstash的數據源有不少,每種數據源都有相應的配置,在Logstash中,這些數據源的相應配置稱爲插件,咱們經常使用的輸入插件有:file、jdbc、redis、tcp、syslog,這些輸入插件會監聽數據源的數據,若是新增數據,將數據封裝成Event進程處理或者傳遞,更多的輸入插件你們能夠看Logstash官網,接下來咱們以file和jdbc兩個輸入插件做爲例子,來學習輸入插件的使用,其餘輸入插件使用起來大同小異,你們自行擴展。redis
讀取文件插件主要用來抓取文件的數據變化信息,以此做爲Logstash的數據源。sql
input{ file { path => ["/var/log/*.log", "/var/log/message"] type => "system" start_position => "beginning" } } output{ stdout{} }
參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
path | array | 無 | 用於匹配被監控的文件,如"/var/logs/*.log"或者 "/var/log/message",必須使用絕對路徑 |
type | string | 無 | Event的type字段,若是採用elasticsearch作store,在默認狀況下將做爲elasticsearch的type |
sincedb_path | string | 「$HOME/.sincedb*」 | 文件讀取記錄,必須指定一個文件而不是目錄,文件中保存沒個被監控的文件等當前inode和byteoffset |
sincedb_write_interval | number | 15 | 間隔多少秒寫一次sincedb文件 |
start_position | string | "end" | 值爲「beginning」和「end」,從文件等開頭仍是結尾讀取文件內容,默認是結尾,若是須要導入文件中的老數據,能夠設置爲「beginning」,該選項只在第一次啓動logstash時有效,若是文件已經存在於sincedb的記錄內,則此配置無效 |
stat_interval | number | 1 | 間隔多少秒檢查一下文件是否被修改,加大此參數將下降系統負載,可是增長了發現新日誌的間隔時間 |
close_older | number | 3600 | 設置文件多少秒內沒有更新就關掉對文件的監聽 |
codec | string | 「plain」 | 輸入數據以後對數據進行解碼 |
add_field | hash | {} | 用於向Event中添加字段 |
delimiter | string | 「\n」 | 文件內容的行分隔符,默認按照換行進行Event對象封裝,也就是說一行數據就一個Event對象 |
discover_interval | number | 15 | 間隔多少秒查看一下path匹配的路徑下是否有新文件產生 |
exclude | array | 無 | path匹配的文件中指定例外,如:path => 「/var/log/「;exclude =>」*.gz」 |
id | string | 無 | 區分兩個相同類型的插件,好比兩個filter |
ignore_older | number | 無 | 忽略歷史修改,若是設置3600秒,logstash只會發現一小時內被修改過的文件,一小時以前修改的文件的變化不會被讀取,若是再次修改該文件,全部的變化都會被讀取,默認被禁用 |
tags | array | 無 | 能夠在Event中增長標籤,以便於在後續的處理流程中使用 |
該插件可使用jdbc把關係型數據庫的數據做爲Logstash的數據源
input { jdbc { jdbc_driver_library => "/opt/logstash/mysql-connector-java-5.1.36-bin.jar" jdbc_driver_class => "com.mysql.jdbc.Driver" jdbc_connection_string => "jdbc:mysql://localhost:3306/mydb" jdbc_user => "mysql" jdbc_password => "123456" parameters => { "favorite_artist" => "Beethoven" } schedule => "* * * * *" statement => "SELECT * from songs where artist = :favorite_artist" } } output{ stdout{} }
參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
jdbc_driver_library | string | 無 | jdbc鏈接驅動的jar包位置 |
jdbc_driver_class | string | 無 | jdbc驅動類 |
jdbc_connection_string | string | 無 | 數據庫的鏈接地址 |
jdbc_user | string | 無 | 數據庫的用戶名 |
jdbc_password | string | 無 | 數據庫的密碼 |
jdbc_paging_enabled | boolean | false | 開啓分頁 |
jdbc_page_size | number | 100000 | 每頁查詢多少條數據 |
statement | string | 無 | 執行的SQL查詢語句 |
parameters | hash | {} | 設置SQL查詢語句的參數 |
use_column_value | boolean | false | 是否須要在程序中使用查詢出來的列 |
tracking_column | string | 無 | 須要在程序中使用哪一個查詢出來的列 |
clean_run | boolean | false | 是否應該保留先前的運行狀態 |
colums_charset | hsah | {} | 特定列的字符編碼。此選項將覆蓋指定列的原來charset配置 |
schedule | string | 無 | 定時執行數據讀取的任務,使用cornexpression表達式:分、時、天、月、年,所有爲*默認含義爲每分鐘執行任務,該cornexpression表達式於Linux系統的crontab使用方式同樣,若是不會使用cornexpression表達式,能夠查看crontab相關文檔。不設置的話,只執行一次數據的讀取 |
lowercase_column_names | boolean | true | 是否須要把查詢出來的列名轉換成小寫,也就是說即便列名是駝峯的,也會轉成所有都是小寫的,默認爲轉小寫,若是不須要,則設置爲false |
record_last_run | boolean | true | 是否記錄數據庫中最後一條數據的位置 |
last_run_metadata_path | string | 無 | 記錄數據庫中最後一條數據的位置信息存放路徑 |
add_field | |||
codec | |||
id | |||
tags | |||
type |
豐富的過濾器插件的是 logstash威力如此強大的重要因素,過濾器插件主要處理流經當前Logstash的事件信息,能夠添加字段、移除字段、轉換字段類型,經過正則表達式切分數據等,也能夠根據條件判斷來進行不一樣的數據處理方式。咱們經常使用的過濾插件有:grok、date、geoip、mutate、json、Split、ruby,更多的過濾插件你們能夠看Logstash官網,接下來咱們以grok、date和geoip這3個過濾插件做爲例子,來學習過濾插件的使用,其餘過濾插件使用起來大同小異,你們自行擴展。
grok正則捕獲是Logstash中將非結構化數據解析成結構化數據以便於查詢的最好工具,很是適合解析system log,web log, database log等任意的 log文件。
內置正則表達式調用
grok提供100多個經常使用正則表達式可供使用,這100多個正則表達式定義在logstash/vendor/bundle/jruby/x.x/gems/logstash-patterns-core-xxx/patterns/grok-patterns
文件中,想要靈活的匹配各類數據,那麼必須查看該文件,大概瞭解grok提供了什麼內置的正則表達式。調用它們的語法以下:%{SYNTAX:SEMANTIC}
SYNTAX:表示內置的正則表達式的名稱
SEMANTIC:表示在Event中建立該字段名來存儲匹配到的值
例如:輸入的數據內容爲「[debug] 127.0.0.1 - test log content」,咱們想提取127.0.0.1這個IP地址,那麼可使用如下語法來匹配:%{IP:client}
,將得到「client: 127.0.0.1」的結果,該結果將成爲Event的一個新的字段和字段值;若是你在捕獲數據時想進行數據類型轉換可使用%{NUMBER:num:int}這種語法,默認狀況下,全部的返回結果都是string類型,當前Logstash所支持的轉換類型僅有「int」和「float」;
自定義表達式調用
與預約義表達式相同,你也能夠將自定義的表達式配置到Logstash中,而後就能夠像於定義的表達式同樣使用;如下是操做步驟說明:
一、在Logstash根目錄下建立文件夾「patterns」,在「patterns」文件夾中建立文件「extra」(文件名稱無所謂,可本身選擇有意義的文件名稱);
二、在文件「extra」中添加表達式,格式:patternName regexp,名稱與表達式之間用空格隔開便可,例如:POSTFIX_QUEUEID [0-9A-F]{10,11}
三、使用自定義的表達式時須要在grok插件配置「patterns_dir」屬性,屬性值爲extra文件所在的目錄。
配置示例
日誌文件http.log每行內容爲:55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965
grok表達式:表達式:%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}
配置文件內容:
input { file { path => "/var/log/http.log" } } filter { grok { patterns_dir => ["./patterns"] match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} message\-id:%{POSTFIX_QUEUEID:queueid}"} } } output{ stdout{} }
輸出結果:
client: 55.3.244.1 method: GET request: /index.html bytes: 15824 duration: 0.043 queueid: BEF25A72965
/var/log/http.log
文件每一行的格式爲55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965
,到時候會把每一行數據封裝成一個Event。55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965
。%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration} message\-id:%{POSTFIX_QUEUEID:queueid}
表達式,把一條message拆成client、method、request、bytes、duration、queueid這6個字段,並添加到Event中。%{POSTFIX_QUEUEID:queueid}
爲自定義表達式的調用,其餘5個是grok內置的表達式調用,若是須要使用自定義表達式,則須要在grok插件配置patterns_dir
屬性,屬性值數據類型爲array,就是自定義表達式定義的文件所在的目錄。\
來轉義,好比-
,""
,[]
這些特殊字符,因此咱們上面實例中的message-id
數據在用表達式匹配的時候是使用了message\-id
去匹配了。參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
match | array | {} | 設置pattern數組 |
patterns_dir | array | [] | 指定自定義的pattern文件存放目錄,Logstash在啓動時會讀取文件夾內全部文件,但前提是這些文件的內容是按照grok語法語法來定義的表達式變量 |
patterns_files_glob | string | "*" | 用於匹配patterns_dir中的文件 |
add_field | |||
add_tag | |||
id | |||
break_on_match | boolean | true | match字段存在多個pattern時,當第一個匹配成功後結束後面的匹配,若是想匹配全部的pattern,將此參數設置爲false |
keep_empty_captures | boolean | false | 若是爲true,捕獲失敗的字段獎設置爲空值 |
clean_run | boolean | false | 是否應該保留先前的運行狀態 |
colums_charset | hsah | {} | 特定列的字符編碼。此選項將覆蓋指定列的原來charset配置 |
overwrite | array | [] | 覆蓋字段內容: match=> { 「message」 => 「%{SYSLOGBASE} %{DATA:message}」 } overwrite=> [ 「message」 ] |
periodic_flush | boolean | false | 按期調用filter的flush方法 |
remove_field | array | [] | 從Event中刪除指定字段: remove_field=> [ 「fieldname」 ] |
remove_tag | array | [] | 刪除「tags」中的值: remove_tag=> [ 「tagname」 ] |
tag_on_failure | array | [「_grokparsefailure」] | 當沒有匹配成功時,將此array添加到「tags」字段內 |
tag_on_timeout | string | 「_groktimeout」 | 當匹配超時時,將此字符串內容添加到「tags」字段內 |
timeout_millis | number | 30000 | 設置單個match到超時時間,單位:毫秒,若是設置爲0,則不啓用超時設置 |
在講date插件的使用前,咱們須要先講解一下Logstash的時間記錄方式。在Logstash產生了一個Event對象的時候,會給該Event設置一個時間,字段爲「@timestamp」,同時,咱們的日誌內容通常也會有時間,可是這兩個時間是不同的,由於日誌內容的時間是該日誌打印出來的時間,而「@timestamp」字段的時間是input插件接收到了一條數據並建立Event的時間,全部通常來講的話「@timestamp」的時間要比日誌內容的時間晚一點,由於Logstash監控數據變化,數據輸入,建立Event致使的時間延遲。這兩個時間均可以使用,具體要根據本身的需求來定。
可是無論「@timestamp」字段的時間仍是日誌內容中的時間,其時間格式通常都不是咱們想要的,因此咱們就須要使用date插件,把時間格式轉成咱們想要的格式,好比:好比將Apr 17 09:32:01(MMM dd HH:mm:ss)轉換爲04-17 09:32:01 (MM-dd HH:mm:ss)
。
filter { date { match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z"] } }
timestamp
是自定義的Event字段,用於存放經過grok解析到的日誌內容中的時間,dd/MMM/YYYY:HH:mm:ss Z
是日誌內容中的時間格式,好比:16/Sep/2018:00:42:38 +0800,匹配到了以後,date插件會默認把該時間轉成本地格式的時間,而且覆蓋掉Event爲咱們建立的@timestamp
字段中的時間。參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
add_field | |||
add_tag | |||
periodic_flush | |||
id | |||
remove_field | |||
remove_tag | |||
remove_tag | |||
tag_on_failure | |||
match | array | [] | 時間字段匹配,可自定多種格式,直到匹配到或者匹配結束,格式:[ field,formats… ],如:match=> [ 「logdate」, 「MMM dd yyyy HH:mm:ss」, 「MMM d yyyy HH:mm:ss」, 「ISO8601」 ] |
target | string | 「@timestamp」 | 指定match匹配而且轉換爲date類型的存儲位置(字段),默認覆蓋到「@timestamp」 |
timezone | string | 無 | 指定時間格式化的時區 |
geoip插件是用於根據IP地址來肯定該IP的歸屬地,默認的數據來源於Maxmind公司GeoLite2(https://dev.maxmind.com/geoip/geoip2/geolite2/)數據庫,該數據庫內嵌在geoip插件中,存儲位置爲:logstash/vendor/bundle/jruby/x.x/gems/logstash-filter-geoip-x.x-java/vendor
目錄中,數據庫文件名分別爲GeoLite2-City.mmdb
和GeoLite2-ASN.mmdb
。
從Maxmind的描述 ----「GeoLite2數據庫是免費的IP地理位置數據庫,可與MaxMind收費的GeoIP2數據庫相媲美,但沒有GeoIP2準確」。 有關更多詳細信息,請參閱GeoIP Lite2許可證。
Maxmind的商業數據庫GeoIP2(https://www.maxmind.com/en/geoip2-databases)也支持geoip插件。簡單來講就是兩個數據庫一個免費的一個商業的,免費的不如商業的地理位置精準。
若是您須要使用內嵌的GeoLite2之外的數據庫,則能夠直接從Maxmind網站下載數據庫,並使用數據庫選項指定其位置(下面會講到若是使用數據選項)。
filter { if [remote_ip] !~ "^127\.|^192\.168\.|^172\.1[6-9]\.|^172\.2[0-9]\.|^172\.3[01]\.|^10\." { geoip { source => "remote_ip" database => "/usr/share/GeoIP/GeoLite2-Country.mmdb" } } }
參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
add_field | |||
add_tag | |||
periodic_flush | |||
id | |||
remove_field | |||
remove_tag | |||
remove_tag | |||
source | string | 無 | 要經過geoip插件查詢的IP地址所在的字段 |
tag_on_failure | array | [「_geoip_lookup_failure」] | 若是ip地址查詢不到地理位置信息,這在標籤中添加該值 |
cache_size | number | 1000 | 因爲geoip查詢很耗時間,因此默認狀況下,查詢過一次的ip地址會緩存起來,目前geoip沒有內存數據驅逐策略,設置的緩存用完了就不能在緩存新的數據了,因此緩存須要設置一個合理值,過小查詢性能增長不明顯,太大就很佔用服務器的內存 |
target | string | 「geoip」 | 把IP地址匹配到的信息放到該字段中 |
database | string | 無 | 使用的Maxmind數據庫文件的路徑。 若是不指定,則默認數據庫是GeoLite2-City。 GeoLite2-City,GeoLite2-Country,GeoLite2-ASN是Maxmind支持的免費數據庫。 GeoIP2-City,GeoIP2-ISP,GeoIP2-Country是Maxmind支持的商業數據庫。 |
default_database_type | string | "City" | 該字段的值只能接受"City"和"ASN",GeoLite2數據庫下細分爲GeoLite2-City,GeoLite2-ASN和GeoLite2-Country,geoip插件內嵌了GeoLite2-City和GeoLite2-ASN,默認使用GeoLite2-City,若是須要使用GeoLite2-ASN,則需設置該字段並設置爲ASN |
fields | array | 無 | 要包含在Event中的geoip字段的信息(每一個數據庫的信息都不同)。 默認狀況下,全部信息都會列出。對於內置的GeoLite2-City數據庫,可使用如下信息:city_name, continent_code, country_code2, country_code3, country_name, dma_code, ip, latitude, longitude, postal_code, region_name 和 timezone |
通過以上的學習,咱們已經學習了Logstash三大主件中的其中兩個了,分別是input和filter,那如今咱們就來學習最後一個組件:output。
每一個數據流通過input和filter後,最終要由output輸出,以上的內容咱們都是使用最簡單的標準輸出:stdout,把數據輸出到顯示器,但實際上stdout只是Logstash的其中一個輸出插件而已,它的經常使用輸出插件有:Elasticsearch,Redis,File,TCP等等,更多的輸出插件你們能夠看Logstash官網,接下來咱們以Elasticsearch和Redis輸出插件做爲例子,來學習輸出插件的使用,其餘過濾插件使用起來大同小異,你們自行擴展。
用於將Event信息寫入到Elasticsearch中,官方推薦插件,ELK技術棧必備插件。
output { elasticsearch { hosts => ["127.0.0.1:9200"] index => "logstash-%{type}-%{+YYYY.MM.dd}" document_type => "%{type}" } }
參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
add_field | |||
add_tag | |||
periodic_flush | |||
flush_size | |||
id | |||
remove_field | |||
remove_tag | |||
codec | |||
enable_metric | |||
hosts | array | 無 | elasticsearch服務器集羣所在的地址 |
index | string | [無] | 文檔索引庫名稱 |
document_type | string | [無] | 文檔類型,不指定的話,類型名爲「doc」 |
document_id | string | [無] | 文檔id,不指定的話,由elasticsearch自動生成 |
user | string | [無] | elasticsearch用戶名 |
password | string | [無] | elasticsearch集羣訪問密碼 |
parent | string | 「nil」 | 爲文檔子節點指定父節點的id |
pool_max | number | 1000 | elasticsearch最大鏈接數 |
pool_max_per_route | number | 100 | 每一個「endpoint」的最大鏈接數 |
proxy | string | 無 | 代理URL |
template | string | 無 | 設置自定義的文檔映射模版存放路徑 |
template_name | string | 無 | 設置使用的默版名稱 |
template_overwrite | boolean | false | 是否始終覆蓋現有模版 |
manage_template | boolean | true | 是否啓用elasticsearch模版,Logstash自帶一個模版,可是隻有名稱匹配「logstash-*」的索引纔會應用該默版 |
parameters | hash | 無 | 添加到elasticsearch URL後面的參數鍵值對 |
timeout | number | 60 | 網絡超時時間 |
用於將Event寫入Redis中進行緩存,因爲Redis數據庫是先把數據存在內存的,因此效率會很是高,是一個經常使用的logstash輸出插件
output { redis { host => ["127.0.0.1"] port => 6379 data_type => "list" key => "logstash-list" } }
參數名稱 | 數據類型 | 默認值 | 描述 |
---|---|---|---|
add_field | |||
add_tag | |||
periodic_flush | |||
flush_size | |||
id | |||
remove_field | |||
remove_tag | |||
codec | |||
enable_metric | |||
hosts | array | ["127.0.0.1"] | redis服務列表,若是配置多個,將隨機選擇一個,若是當前的redis服務不可用,將選擇下一個 |
port | number | 6379 | 文檔索引庫名稱 |
db | number | 0 | 使用的redis數據庫編號,默認使用0號數據庫 |
password | string | 無 | redis的密碼 |
data_type | string | 無 | 存儲在redis中的數據類型,只能使用「list」和「channel」這兩個值。若是使用「list」,將採用「RPUSH」操做,若是是「channel」,將採用「PUBLISH」操做 |
key | string | 無 | 給redis設置一個key,來存儲收集到的數據庫 |
batch | boolean | false | 是否啓用redis的batch模式,僅在data_type=」list」時有效 |
batch_events | number | 50 | batch大小,batch達到此大小時執行「RPUSH」 |
那到這裏,咱們已經介紹了Logstash配置文件的語法和經常使用插件的配置方式了,這些經常使用的插件都是使用頻率很是高的,全部咱們後面須要來作一個Logstash的實戰案例,綜合運用咱們這章所學的內容。