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)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。