ELK技術棧之-Logstash詳解

ELK技術棧之-Logstash詳解

 

前言

在第九章節中,咱們已經安裝好Logstash組件了,而且啓動實例測試它的數據輸入和輸出,可是用的是最簡單的控制檯標準輸入和標準輸出,那這節咱們就來深刻的學習Logstash的詳細使用。php

經常使用啓動參數

咱們在上一節中演示了啓動Logstash的實例,其中咱們啓動的時候給Logstash腳本傳入了-e的參數,但實際上,Logstash的啓動參數有不少,咱們來看一下各個啓動參數的做用:html

  • -e #當即啓動實例,例如:./logstash -e "input {stdin {}} output {stdout {}}"
  • -f #指定啓動實例的配置文件,例如:./logstash -f config/test.conf
  • -t #測試配置文件的正確性,例如:./logstash -f config/test.conf -t
  • -l #指定日誌文件名稱,例如:./logstash -f config/test.conf -l logs/test.log
  • -w #指定filter線程數量,不指定默認是5,例如:./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

file輸入插件

讀取文件插件主要用來抓取文件的數據變化信息,以此做爲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輸入插件

該插件可使用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正則插件

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。
    二、使用grok過濾插件處理該文件,match爲匹配Event中的message,message就是該文件的一行數據,好比55.3.244.1 GET /index.html 15824 0.043 message-id:BEF25A72965
    三、匹配message的內容使用%{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時間處理插件

在講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插件

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.mmdbGeoLite2-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"
        }
    }
}
  • 示例解析
    因爲geoip使用的ip數據庫不能匹配私網地址,因此在使用geoip插件前,先判斷一下ip地址是否爲私網ip,若是不是,這使用geoip插件。source爲指定一個ip地址,爲其查詢歸屬地等信息,remote_ip爲Event中存儲ip地址的自定義字段,database爲IP地址數據庫所在的位置,不指定的話,使用geoip插件內置的GeoLite2-City默認數據庫。
  • 經常使用參數(空 = 同上)
參數名稱 數據類型 默認值 描述
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輸出插件做爲例子,來學習輸出插件的使用,其餘過濾插件使用起來大同小異,你們自行擴展。

elasticsearch輸出插件

用於將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 網絡超時時間

redis輸出插件

用於將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的實戰案例,綜合運用咱們這章所學的內容。

相關文章
相關標籤/搜索