Logstash的简单介绍
logstash的簡介
logstash的概念:是一款開源的數(shù)據(jù)收集引擎,具有實時管道處理能力。
logstash具有200多個插件,可以接受各種各樣的數(shù)據(jù)(如日志、網(wǎng)絡(luò)請求、關(guān)系型數(shù)據(jù)庫、傳感器或物聯(lián)網(wǎng)等等)
Logstash工作過程:
Logstash 就像管道符一樣,讀取輸入數(shù)據(jù),然后處理過濾數(shù)據(jù),最后輸出到目的地(數(shù)據(jù)在線程之間以事件的形式流傳)
logstash數(shù)據(jù)處理過程包括三個部分:input、filter、output
input和output部分可以使用codecs對數(shù)據(jù)格式進(jìn)行處理
Input,Filter,Output和codec:都是以插件形式存在
用戶可以通過設(shè)置pipeline配置文件,設(shè)置符合需求的input、filter、outer、codecs插件實現(xiàn)對指定數(shù)據(jù)的采集、處理和輸出功能
logstash的三大部分的介紹:
input:從數(shù)據(jù)源獲取數(shù)據(jù)。不同的數(shù)據(jù)源使用不同的插件。常用的插件有:file、jdbc、Redis、syslog、beats 、http等
filter:用于處理數(shù)據(jù)。對從數(shù)據(jù)源獲取到的數(shù)據(jù)按照需求進(jìn)行處理(如:解析數(shù)據(jù)、刪除字段、類型轉(zhuǎn)換等)。常用的組件有:date、grok、dessect、mutate、json、geoip、ruby等
output:用于將數(shù)據(jù)輸出到目的地。不同的目的地使用不同的插件。常用的插件有:elaticsearch、file、graphite、statsd等
codec:用于對數(shù)據(jù)進(jìn)行編碼解碼。不是一個單獨(dú)的流程,是用于input和output部分對數(shù)據(jù)進(jìn)行編解碼作用。常見的組件有:json、multiline等
logstash不是一個input-filter-output的數(shù)據(jù)流,而是一個 input | decode | filter | encode | output 的數(shù)據(jù)流。
logstash將數(shù)據(jù)轉(zhuǎn)換為事件時候,會給事件添加一些額外的信息。下面介紹幾個常見的額為信息:
@timestamp:用來標(biāo)記事件的發(fā)生時間
host:標(biāo)記事件發(fā)生地址
type:標(biāo)記事件的唯一類型(input和output部分都可以配置多個不同的插件,每個插件可以用type來唯一標(biāo)記這個插件,可以實現(xiàn)對不同的插件進(jìn)行不同的處理)
tags:標(biāo)記事件的某方面屬性。這是一個數(shù)組,一個事件可以有多個標(biāo)簽
Logstash中的數(shù)據(jù)類型:
bool:use_column_value => true
string:jdbc_driver_class => "com.mysql.jdbc.Driver"
number:jdbc_page_size => 50000
array:hosts =>["192.168.57.100:9200","192.168.57.101:9200","192.168.57.102:9200"]
hash:options =>{key1 =>value1,key2 =>value2}
logastah中的邏輯運(yùn)算符:
相等關(guān)系:==、!=、<、>、<=、>=
正則關(guān)系:=~(匹配正則)、!~(不匹配正則)
包含關(guān)系:in、not in
布爾操作:and(與)、or(或)、nand(非與)、xor(非或)
一元運(yùn)算符:!(取反)、()(復(fù)合表達(dá)式)、!() (對復(fù)合表達(dá)式結(jié)果取反)
三大模塊插件的簡單介紹
input
標(biāo)準(zhǔn)輸入插件:表示從控制臺輸入
input {stdin { }}
file插件:表示從文件讀取數(shù)據(jù)(如日志文件)
input {
    file {
        # 要導(dǎo)入的文件的位置,
        path => "/var/lib/mysql/slow.log"            
        # 從文件開始的位置開始讀,end表示從結(jié)尾開始讀
        start_position => "beginning"
        # 多久之內(nèi)沒修改過的文件不讀取,0為無限制,單位為秒
        ignore_older => 0  
        # 記錄文件上次讀取位置,輸出到null表示每次都從文件首行開始解析
        sincedb_path => "/dev/null"
        # type字段,可表明導(dǎo)入的日志類型
        type => "mysql-slow"
    }
}
filter
Filter模塊是logstash功能強(qiáng)大的主要原因,可以對Logstash Event進(jìn)行豐富的處理(比如解析數(shù)據(jù)、刪除字段、類型轉(zhuǎn)換等等)
常見的Filter插件如下:
| 插件名稱 | 說明 | 
| date | 日期解析 | 
| grok | 正則匹配解析 | 
| dessect | 分隔符解析 | 
| mutate | 對字段做處理,比如刪除、重命名、替換 | 
| json | 按照J(rèn)son格式解析字段內(nèi)容到指定字段中 | 
| geoip | 將ip地址進(jìn)行解析獲取地域信息(包括國別,省市,經(jīng)緯度等) | 
| ruby | 利用ruby代碼來動態(tài)修改logstash event | 
Date插件:可以將日期字符串解析為日期類型,然后替換@timestamp字段或者指定成其他字段(如果不指定為@timestamp字段,會使用系統(tǒng)的時間作為@timestamp的值)
將input中的event_date字段的字符串解析日期格式,將解析后的的數(shù)據(jù)作為@timestamp的值;event_date的日期字符串格式需要和后面的yyyy-MM-dd HH:mm:ss對應(yīng)
如:2020/12/31 12:23:32 對應(yīng)的格式為match =>["event_date","yyyy/MM/dd HH:mm:ss"]
#將event_date字符串解析為日期類型,并將解析后的值作為@timestamp的值
filter { date{ match =>["event_date","yyyy-MM-dd HH:mm:ss"] target =>"@timestamp" } }
Grok插件:基于正則解析數(shù)據(jù)
grok是filter最重要的插件,grok使用正則表達(dá)式來生成grok語法,將input的數(shù)據(jù)解析成需要的數(shù)據(jù)格式
grok中常用patterns的路徑:logstash_filevendorundlejrubyx.xgemslogstash-patterns-core-x.x.xpatternsgrok-patterns
grok語法:%{SYNTAX:SEMANTIC}
SYNTAX:grok pattern的名稱
SEMANTIC:賦值字段名稱(將匹配到的值賦值給SEMANTIC這個字段)
如:%{NUMBER:field}:可以匹配數(shù)值類型,但是grok匹配出的內(nèi)容都是字符串類型,可以通過在最后指定為int或者float來強(qiáng)轉(zhuǎn)類型:%{NUMBER:field:int}
實例:控制臺輸入:hello 123456 666666
filter{
    grok{
        match => {
            "message" => "hello %{NUMBER:num_file:int} %{NUMBER:number_file}"
        }
    }
}
自定義正則表達(dá)式:
例子:將匹配到的值賦值給num_file這個字段:(?<num_file>[0-9]{6})
控制臺輸入:hello 123456 666666
filter{
    grok{
        match => {
            "message" => "hello (?<num_file>[0-9]{6}) %{NUMBER:number_file}"
        }
    }
}
自定義grok pattern:通過pattern_definitions參數(shù),以鍵值對的方式定義pattern名稱和內(nèi)容(也可以通過pattern_dir參數(shù),以文件的形式讀取pattern)
例子:定義模板的名稱為pattern_test,模板的內(nèi)容是正則匹配8位的數(shù)字或者字母;并且使用了自定義的pattern
控制臺輸入:hello 1234abcd
filter{
    grok{
        match => {
            "message" => "%{pattern_test:number_file}"
        }
        pattern_definitions => {
            "pattern_test" => "hello [a-z0-9]{8}"
        }
    }
}
Dissect插件:基于分隔符原理解析數(shù)據(jù)
解決grok解析時消耗過多cpu資源的問題
語法簡單,能處理的場景比較有限
它只能處理格式相似,且有分隔符的字符串
語法如下:
%{}里面是字段;字段名稱可以自己定義
兩個%{}之間是分隔符
例子:把日期和時間解析到同一個字段datetime_flie
控制臺輸入:Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool
filter{
    dissect{
        mapping => {
            "message" => "%{datetime_flie} %{+datetime_flie} %{+datetime_flie} %{local_host} %{sys_file}[%{pid}]: %{message_flle}"
        }
    }
}
Mutate插件:是使用最頻繁的插件,可以對字段進(jìn)行各種操作(比如重命名、刪除、替換、更新等)
convert:類型轉(zhuǎn)換
gsub字符串替換
split、join、merge字符串切割、數(shù)組合并為字符串、數(shù)組合并為數(shù)組
rename字段重命名
update、replace字段內(nèi)容更新或替換。它們都可以更新字段的內(nèi)容,區(qū)別在于update只在字段存在時生效,而replace在字段不存在時會執(zhí)行新增字段的操作
remove_field刪除字段
例子:對數(shù)據(jù)進(jìn)行處理分割、添加字段和刪除字段
控制臺輸入:hello Weiking 20200810 man
filter{
    mutate{
        split => ["message", " "]
        add_field => {
            "field_name" => "%{[message][0]}"
            "name" => "%{[message][1]}"
            "date" => "%{[message][2]}"
            "sex" => "%{[message][3]}"
        }
        remove_field => "sex"
    }
}
Json插件:將指定字段的json格式的字符串進(jìn)行解析,如果不指定target,會把解析出來的json數(shù)據(jù)直接放到根級別,可以通過target將解析后的json數(shù)據(jù)放到指定的字段中
例子:將json格式的字符串解析放在trans_all字段中
控制臺輸入:{"trans_jnls_no":"0","card_no":"623061571015020279","terminal_no":"zhanglq","atm_trans_types_type":"","customer_no":"106913609","cash_mark":"212"}
filter{
    json{
        source => "message"
        target => "trans_all"
        remove_field => "message"
    }
}
Geoip插件:基于Geoip庫解析IP數(shù)據(jù)獲取地域信息
GeoIP庫:可以根據(jù)IP地址提供對應(yīng)的地域信息(包括國別,省市,經(jīng)緯度等)
例子:解析IP地址獲取地域信息
控制臺輸入:124.160.75.205
filter{
    geoip{
        source => "message"
        remove_field => "message"
    }
}    
output
標(biāo)準(zhǔn)輸出插件:多用于調(diào)試,從控制臺輸出
output {
    stdout {
        codec => rubydebug
    }
}
elasticsearch插件:將數(shù)據(jù)輸入到elaticsearch中
output {
    elasticsearch {
        #es的ip和端口
        hosts => ["http://192.168.57.100:9200"] 
        #集群:hosts => ["10.0.xx.xx:9200", "10.0.xx.xx:9200", "10.0.xx.xx:9200"]        
        #ES索引名稱(自己定義的)
        index => "mdl_001_202008"
        #文檔類型
        document_type => "MdL_001"
        #設(shè)置數(shù)據(jù)的id為數(shù)據(jù)庫中的字段
        document_id => "%{risk_event_no}-%{rule_id}"
    }
}
logstash配置實例
logstash的配置的文件:input和output都可以配置多個不同的插件,filter可以針對input里面的每個數(shù)據(jù)源做不一樣的過濾,通過各自定義的type來匹配
配置實例:
input{
    #從控制臺輸入數(shù)據(jù)
    stdin{ 
        type => "stdin"
    }  
    #讀取文件中的數(shù)據(jù)
    file {
        # 要導(dǎo)入的文件的位置,
        path => "/opt/data/data.json"            
        # 從文件開始的位置開始讀,end表示從結(jié)尾開始讀
        start_position => "beginning"
        # 多久之內(nèi)沒修改過的文件不讀取,0為無限制,單位為秒
        ignore_older => 0  
        # 記錄文件上次讀取位置,輸出到null表示每次都從文件首行開始解析
        sincedb_path => "/dev/null"
        # type字段,可表明導(dǎo)入的日志類型
        type => "file"
    }
}
filter{
        if[type] == "stdin"{
            mutate{
                rename => ["message","timestamp"]
            }
        }
        if[type] == "file"{
            json{
                source => "message"
            }
        }
}
output {
        if[type] == "file"{
          elasticsearch {
                #es的ip和端口
                hosts => ["http://192.168.57.100:9200"] 
                #集群:hosts => ["10.0.xx.xx:9200", "10.0.xx.xx:9200", "10.0.xx.xx:9200"]        
                #ES索引名稱(自己定義的)
                index => "mdl_001_20200812"
                #文檔類型
                document_type => "MdL_012"
                #設(shè)置數(shù)據(jù)的id為數(shù)據(jù)庫中的字段
                document_id => "%{risk_event_no}"
            }
        }
        if[type] == "stdin"{
            stdout{
                codec => rubydebug
            }
        }
} 
                            總結(jié)
以上是生活随笔為你收集整理的Logstash的简单介绍的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
 
                            
                        