企業級實戰模塊四：Logstash配置語法詳解

1 Logstash基本語法組成

logstash之所以功能強大和流行，還與其豐富的過濾器插件是分不開的，過濾器提供的並不單單是過濾的功能，還可以對進入過濾器的原始數據進行復雜的邏輯處理，甚至添加獨特的事件到后續流程中。

Logstash配置文件有如下三部分組成，其中input、output部分是必須配置，filter部分是可選配置，而filter就是過濾器插件，可以在這部分實現各種日志過濾功能。

input {
    #輸入插件
}
filter {
    #過濾匹配插件
}
output {
    #輸出插件
}

2 Logstash輸入插件(Input)

2.1 讀取文件(File)

logstash使用一個名為filewatch的ruby gem庫來監聽文件變化,並通過一個叫.sincedb的數據庫文件來記錄被監聽的日志文件的讀取進度（時間戳），這個sincedb數據文件的默認路徑在 <path.data>/plugins/inputs/file下面，文件名類似於.sincedb_452905a167cf4509fd08acb964fdb20c，而<path.data>表示logstash插件存儲目錄，默認是LOGSTASH_HOME/data。

看下面一個事件配置文件：

input {
   file {
       path => ["/var/log/messages"]
       type => "system"
       start_position => "beginning"
   }
}
output {
   stdout{
        codec=>rubydebug    
   }
}

這個配置是監聽並接收本機的/var/log/messages文件內容，start_position表示按時間戳記錄的地方開始讀取，如果沒有時間戳則從頭開始讀取，有點類似cat命令，默認情況下，logstash會從文件的結束位置開始讀取數據，也就是說logstash進程會以類似tail -f命令的形式逐行獲取數據。type用來標記事件類型,通常會在輸入區域通過type標記事件類型。

2.2 標准輸入(Stdin)

stdin是從標准輸入獲取信息，關於stdin的使用，前面已經做過了一些簡單的介紹，這里再看一個稍微復雜一點的例子，下面是一個關於stdin的事件配置文件：

input{
   stdin{
        add_field=>{"key"=>"iivey"}
        tags=>["add1"]
        type=>"test1"
   }
}
output {
   stdout{
        codec=>rubydebug    
   }
}

2.3 讀取 Syslog日志

如何將rsyslog收集到的日志信息發送到logstash中，這里以centos7.5為例，需要做如下兩個步驟的操作：

首先，在需要收集日志的服務器上找到rsyslog的配置文件/etc/rsyslog.conf,添加如下內容：

*.*   @@192.168.5.7:5514

其中，192.168.5.7是logstash服務器的地址。5514是logstash啟動的監聽端口。

接着，重啟rsyslog服務：

systemctl restart rsyslog

然后，在logstash服務器上創建一個事件配置文件，內容如下：

input {
 syslog {
   port => "5514"
 }
}
​
output {
   stdout{
        codec=>rubydebug    
   }
}

2.4 讀取TCP網絡數據

下面的事件配置文件就是通過"LogStash::Inputs::TCP"和"LogStash::Filters::Grok"配合實現syslog功能的例子，這里使用了logstash的TCP/UDP插件讀取網絡數據：

input {
 tcp {
   port => "5514"
 }
}
​
filter {
 grok {
   match => { "message" => "%{SYSLOGLINE}" }
 }
}
​
output {
   stdout{
        codec=>rubydebug
   }
}

其中，5514端口是logstash啟動的tcp監聽端口。注意這里用到了日志過濾"LogStash::Filters::Grok"功能，下面馬上會介紹到。

3 Logstash編碼插件(Codec)

其實我們就已經用過編碼插件codec了，也就是這個rubydebug，它就是一種codec，雖然它一般只會用在stdout插件中，作為配置測試或者調試的工具。

編碼插件(Codec)可以在logstash輸入或輸出時處理不同類型的數據，因此，Logstash不只是一個input-->filter-->output的數據流，而是一個input-->decode-->filter-->encode-->output的數據流。

 Codec支持的編碼格式常見的有plain、json、json_lines等。

3.1 codec插件之plain

plain是一個空的解析器，它可以讓用戶自己指定格式，也就是說輸入是什么格式，輸出就是什么格式。下面是一個包含plain編碼的事件配置文件：

input{
    stdin{
	}
}
output{
    stdout{
        codec => "plain"
	}
}

3.2 codec插件之json、json_lines

如果發送給logstash的數據內容為json格式,可以在input字段加入codec=>json來進行解析，這樣就可以根據具體內容生成字段，方便分析和儲存。如果想讓logstash輸出為json格式，可以在output字段加入codec=>json，下面是一個包含json編碼的事件配置文件：

input {
    stdin {
        }
    }
output {
    stdout {
        codec => json
        }
}

這就是json格式的輸出，可以看出，json每個字段是key:values格式，多個字段之間通過逗號分隔。有時候，如果json文件比較長，需要換行的話，那么就要用json_lines編碼格式了。

4 Logstash過濾器插件(Filter)

4.1 Grok 正則捕獲

grok是一個十分強大的logstash filter插件，他可以通過正則解析任意文本，將非結構化日志數據弄成結構化和方便查詢的結構。他是目前logstash 中解析非結構化日志數據最好的方式。

Grok 的語法規則是：

	%{語法: 語義}

“語法”指的就是匹配的模式，例如使用NUMBER模式可以匹配出數字，IP模式則會匹配出127.0.0.1這樣的IP地址：

例如輸入的內容為：

192.168.50.21 [08/Oct/2021:23:24:19 +0800] "GET / HTTP/1.1" 403 5039

那么，%{IP:clientip}匹配模式將獲得的結果為：

clientip: 192.168.50.21

%{HTTPDATE:timestamp}匹配模式將獲得的結果為：

timestamp: 08/Oct/2021:23:24:19 +0800

而%{QS:referrer}匹配模式將獲得的結果為：

referrer: "GET / HTTP/1.1"

下面是一個組合匹配模式，它可以獲取上面輸入的所有內容：

%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}	

通過上面這個組合匹配模式，我們將輸入的內容分成了五個部分，即五個字段，將輸入內容分割為不同的數據字段，這對於日后解析和查詢日志數據非常有用，這正是使用grok的目的。

Logstash默認提供了近200個匹配模式（其實就是定義好的正則表達式）讓我們來使用，可以在logstash安裝目錄下，例如這里是/usr/local/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.1.2/patterns目錄里面查看，基本定義在grok-patterns文件中。

從這些定義好的匹配模式中，可以查到上面使用的四個匹配模式對應的定義規則

匹配模式	正則定義規則
NUMBER	(?:%{BASE10NUM})
HTTPDATE	%{MONTHDAY}/%{MONTH}/%{YEAR}:%{TIME} %{INT}
IP	(?:%{IPV6}|%{IPV4})
QS	%{QUOTEDSTRING}

示例：

input{
    stdin{}
}
filter{
    grok{
        match => ["message","%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

輸入內容：

192.168.50.21 [08/Oct/2021:23:24:19 +0800] "GET / HTTP/1.1" 403 5039

4.2 時間處理(Date)

date插件是對於排序事件和回填舊數據尤其重要，它可以用來轉換日志記錄中的時間字段，變成LogStash::Timestamp對象，然后轉存到@timestamp字段里，這在之前已經做過簡單的介紹。

下面是date插件的一個配置示例：

input{
    stdin{}
}
filter {
    grok {
        match => ["message", "%{HTTPDATE:timestamp}"]
    }
    date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
}
output{
    stdout{
        codec => "rubydebug"
    }
}

時間字段	字母	表示含義
年	yyyy	表示全年號碼。 例如：2021
年	yy	表示兩位數年份。 例如：2021年即為21
月	M	表示1位數字月份，例如：1月份為數字1，12月份為數字12
月	MM	表示兩位數月份，例如：1月份為數字01，12月份為數字12
月	MMM	表示縮短的月份文本，例如：1月份為Jan，12月份為Dec
月	MMMM	表示全月文本，例如：1月份為January，12月份為December
日	d	表示1位數字的幾號，例如8表示某月8號
日	dd	表示2位數字的幾號，例如08表示某月8號
時	H	表示1位數字的小時，例如1表示凌晨1點
時	HH	表示2位數字的小時，例如01表示凌晨1點
分	m	表示1位數字的分鍾，例如5表示某點5分
分	mm	表示2位數字的分鍾，例如05表示某點5分
秒	s	表示1位數字的秒，例如6表示某點某分6秒
秒	ss	表示2位數字的秒，例如06表示某點某分6秒
時區	Z	表示時區偏移，結構為HHmm，例如：+0800
時區	ZZ	表示時區偏移，結構為HH:mm，例如：+08:00
時區	ZZZ	表示時區身份，例如Asia/Shanghai

4.3 數據修改(Mutate)

1）正則表達式替換匹配字段

gsub可以通過正則表達式替換字段中匹配到的值，只對字符串字段有效，下面是一個關於mutate插件中gsub的示例（僅列出filter部分）：

filter {
    mutate {
        gsub => ["filed_name_1", "/" , "_"]
    }
}

這個示例表示將filed_name_1字段中所有"/"字符替換為"_"。

2）分隔符分割字符串為數組

split可以通過指定的分隔符分割字段中的字符串為數組，下面是一個關於mutate插件中split的示例（僅列出filter部分）：

filter {
    mutate {
        split => ["filed_name_2", "|"]
    }
}

這個示例表示將filed_name_2字段以"|"為區間分隔為數組。

3）重命名字段

rename可以實現重命名某個字段的功能，下面是一個關於mutate插件中rename的示例（僅列出filter部分）：

filter {
    mutate {
        rename => { "old_field" => "new_field" }
    }
}

這個示例表示將字段old_field重命名為new_field。

4）刪除字段

remove_field可以實現刪除某個字段的功能，下面是一個關於mutate插件中remove_field的示例（僅列出filter部分）：

filter {
    mutate {
        remove_field  =>  ["timestamp"]
    }
}

這個示例表示將字段timestamp刪除。

5）綜合示例

input {
    stdin {}
}
filter {
    grok {
        match => { "message" => "%{IP:clientip}\ \[%{HTTPDATE:timestamp}\]\ %{QS:referrer}\ %{NUMBER:response}\ %{NUMBER:bytes}" }
        remove_field => [ "message" ]
   }
date {
        match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
    }
mutate {
           rename => { "response" => "response_new" }
           convert => [ "response","float" ]
           gsub => ["referrer","\"",""]
           remove_field => ["timestamp"]
           split => ["clientip", "."]
        }
}
output {
    stdout {
        codec => "rubydebug"
    }
}

4.4 GeoIP 地址查詢歸類

GeoIP是最常見的免費IP地址歸類查詢庫，當然也有收費版可以使用。GeoIP庫可以根據IP 地址提供對應的地域信息，包括國別，省市，經緯度等，此插件對於可視化地圖和區域統計非常有用。

下面是一個關於GeoIP插件的簡單示例（僅列出filter部分）：

filter {
    geoip {
        source => "ip_field"
    }
}

其中，ip_field字段是輸出IP地址的一個字段。

4.5 filter插件綜合應用實例

下面給出一個業務系統輸出的日志格式，由於業務系統輸出的日志格式無法更改，因此就需要我們通過logstash的filter過濾功能以及grok插件來獲取需要的數據格式，此業務系統輸出的日志內容以及原始格式如下：

2021-10-09T0:57:42+08:00|~|123.87.240.97|~|Mozilla/5.0 (iPhone; CPU iPhone OS 11_2_2 like Mac OS X) AppleWebKit/604.4.7 Version/11.0 Mobile/15C202 Safari/604.1|~|http://m.sina.cn/cm/ads_ck_wap.html|~|1460709836200|~|DF0184266887D0E

可以看出，這段日志都是以“|~|”為區間進行分隔的，那么剛好我們就以“|~|”為區間分隔符，將這段日志內容分割為6個字段。這里通過grok插件進行正則匹配組合就能完成這個功能。

完整的grok正則匹配組合語句如下：

%{TIMESTAMP_ISO8601:localtime}\|\~\|%{IPORHOST:clientip}\|\~\|(%{GREEDYDATA:http_user_agent})\|\~\|(%{DATA:http_referer})\|\~\|%{GREEDYDATA:mediaid}\|\~\|%{GREEDYDATA:osid}

5 Logstash輸出插件（output）

5.1 輸出到標准輸出(stdout)

stdout與之前介紹過的stdin插件一樣，它是最基礎和簡單的輸出插件，下面是一個配置實例：

output {
    stdout {
        codec => rubydebug
    }
}

stdout插件，主要的功能和用途就是用於調試，這個插件，在前面已經多次使用過。這里不再過多介紹。

5.2 保存為文件（file）

file插件可以將輸出保存到一個文件中，配置實例如下：

output {
    file {
        path => "/data/log3/%{+yyyy-MM-dd}/%{host}_%{+HH}.log"
    }

上面這個配置中，使用了變量匹配，用於自動匹配時間和主機名，這在實際使用中很有幫助。

5.3 輸出到elasticsearch

Logstash將過濾、分析好的數據輸出到elasticsearch中進行存儲和查詢，是最經常使用的方法。下面是一個配置實例：

    output {
    elasticsearch {
        host => ["192.168.5.8:9200","192.168.5.9:9200","192.168.5.10:9200"]
        index => "logstash-%{+YYYY.MM.dd}"
        manage_template => false
        template_name => "template-web_access_log"
    }
}

上面配置中每個配置項含義如下：

• host：是一個數組類型的值，后面跟的值是elasticsearch節點的地址與端口，默認端口是9200。可添加多個地址。

• index：寫入elasticsearch的索引的名稱，這里可以使用變量。Logstash提供了%{+YYYY.MM.dd}這種寫法。在語法解析的時候，看到以+ 號開頭的，就會自動認為后面是時間格式，嘗試用時間格式來解析后續字符串。這種以天為單位分割的寫法，可以很容易的刪除老的數據或者搜索指定時間范圍內的數據。此外，注意索引名中不能有大寫字母。

• manage_template:用來設置是否開啟logstash自動管理模板功能，如果設置為false將關閉自動管理模板功能。如果我們自定義了模板，那么應該設置為false。

• template_name:這個配置項用來設置在Elasticsearch中模板的名稱。

•