1 FileBeat

Filebeat 是使用 Golang 實(shí)現(xiàn)的輕量型日志采集器，也是 Elasticsearch stack 里面的一員。本質(zhì)上是一個(gè) agent ，可以安裝在各個(gè)節(jié)點(diǎn)上，根據(jù)配置讀取對(duì)應(yīng)位置的日志，并上報(bào)到相應(yīng)的地方去。

1.1 FileBeat 安裝與使用

從 官網(wǎng) 下載對(duì)應(yīng)的版本，我這里的 ElasticSearch 版本號(hào)是 6.4.3，所以下載 FileBeat 的版本也是 6.4.3。
下載后解壓：

cd /home/software
tar -zxvf filebeat-6.4.3-linux-x86_64.tar.gz -C /usr/local/
cd /usr/local
mv filebeat-6.4.3-linux-x86_64/ filebeat-6.4.3

配置 Filebeat，可以參考 filebeat.full.yml 中的配置

vim /usr/local/filebeat-6.4.3/filebeat.yml

內(nèi)容如下：

###################### Filebeat Configuration Example #########################
filebeat.prospectors:- input_type: logpaths:## 定義了日志文件路徑，可以采用模糊匹配模式，如*.log- /workspaces/logs/logCollector/app-collector.log#定義寫入 ES 時(shí)的 _type 值document_type: "app-log"multiline:#pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表達(dá)式（匹配以 2017-11-15 08:04:23:889 時(shí)間格式開頭的字符串）pattern: '^\['                              # 指定匹配的表達(dá)式（匹配以 [ 開頭的字符串）negate: true                                # 是否需要匹配到match: after                                # 不匹配的行，合并到上一行的末尾max_lines: 2000                             # 最大的行數(shù)timeout: 2s                                 # 如果在規(guī)定時(shí)間沒有新的日志事件就不等待后面的日志fields: ## topic 對(duì)應(yīng)的消息字段或自定義增加的字段logbiz: collectorlogtopic: app-log-collector   ## 按服務(wù)劃分用作kafka topic，會(huì)在logstash filter 過濾數(shù)據(jù)時(shí)候作為 判斷參數(shù) [fields][logtopic]evn: dev- input_type: logpaths:## 定義了日志文件路徑，可以采用模糊匹配模式，如*.log- /workspaces/logs/logCollector/error-collector.log#定義寫入 ES 時(shí)的 _type 值document_type: "error-log"multiline:#pattern: '^\s*(\d{4}|\d{2})\-(\d{2}|[a-zA-Z]{3})\-(\d{2}|\d{4})'   # 指定匹配的表達(dá)式（匹配以 2017-11-15 08:04:23:889 時(shí)間格式開頭的字符串）pattern: '^\['                              # 指定匹配的表達(dá)式（匹配以 [ 開頭的字符串）negate: true                                # 是否匹配到match: after                                # 不匹配的行，合并到上一行的末尾max_lines: 2000                             # 最大的行數(shù)timeout: 2s                                 # 如果在規(guī)定時(shí)間沒有新的日志事件就不等待后面的日志，直接進(jìn)行推送操作fields: ## topic 對(duì)應(yīng)的消息字段或自定義增加的字段logbiz: collectorlogtopic: error-log-collector   ## 按服務(wù)劃分用作kafka topicevn: devoutput.kafka: ## filebeat 支持多種輸出，支持向 kafka，logstash，elasticsearch 輸出數(shù)據(jù)，此處設(shè)置數(shù)據(jù)輸出到 kafka。enabled: true ## 啟動(dòng)這個(gè)模塊hosts: ["192.168.212.128:9092"] ## 地址topic: '%{[fields.logtopic]}'  ## 主題（使用動(dòng)態(tài)變量）partition.hash:  ## kafka 分區(qū) hash 規(guī)則reachable_only: truecompression: gzip  ## 數(shù)據(jù)壓縮max_message_bytes: 1000000  ## 最大容量required_acks: 1  ## 是否需要 ack
logging.to_files: true

檢查配置是否正確：

./filebeat -c filebeat.yml -configtest

啟動(dòng)filebeat：

/usr/local/filebeat-6.4.3/filebeat &

注：需要啟動(dòng) kafka

2 Logstash 日志過濾

在 《Elasticsearch入門筆記（Logstash數(shù)據(jù)同步）》 這邊文章中已經(jīng)介紹過任何安裝配置 Logstash 了，不過那時(shí)輸入的數(shù)據(jù)源是來自于 MySQL，此時(shí)輸入的數(shù)據(jù)源是 Kafka。
在 /usr/local/logstash-6.4.3 目錄下新建一個(gè) script 用于存放對(duì)接 Kafka 的配置文件。
以下是該目錄下創(chuàng)建的 logstash-script.conf 文件：

## multiline 插件也可以用于其他類似的堆棧式信息，比如 linux 的內(nèi)核日志。
input {kafka {topics_pattern => "app-log-.*"  ## kafka 主題 topicbootstrap_servers => "192.168.212.128:9092"  ## kafka 地址codec => json  ## 數(shù)據(jù)格式consumer_threads => 1  ## 增加consumer的并行消費(fèi)線程數(shù)（數(shù)值可以設(shè)置為 kafka 的分片數(shù)）decorate_events => truegroup_id => "app-log-group" ## kafka 組別}kafka {topics_pattern => "error-log-.*"  ## kafka 主題 topicbootstrap_servers => "192.168.212.128:9092" ## kafka 地址codec => json  ## 數(shù)據(jù)格式consumer_threads => 1  ## 增加consumer的并行消費(fèi)線程數(shù)（數(shù)值可以設(shè)置為 kafka 的分片數(shù)）decorate_events => truegroup_id => "error-log-group" ## kafka 組別}
}filter {## 時(shí)區(qū)轉(zhuǎn)換，這里使用 ruby 語言，因?yàn)?logstash 本身是東八區(qū)的，這個(gè)時(shí)區(qū)比北京時(shí)間慢8小時(shí)，所以這里采用 ruby 語言設(shè)置為北京時(shí)區(qū)ruby {code => "event.set('index_time',event.timestamp.time.localtime.strftime('%Y.%m.%d'))"}## [fields][logtopic] 這個(gè)是從 FileBeat 定義傳入 Kafka 的if "app-log" in [fields][logtopic]{grok {## 表達(dá)式,這里對(duì)應(yīng)的是Springboot輸出的日志格式match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]}}## [fields][logtopic] 這個(gè)是從 FileBeat 定義傳入 Kafka 的if "error-log" in [fields][logtopic]{grok {## 表達(dá)式match => ["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]}}
}## 測(cè)試輸出到控制臺(tái)：
## 命令行輸入 ./logstash -f /usr/local/logstash-6.4.3/script/logstash-script.conf  --verbose --debug
output {stdout { codec => rubydebug }
}## elasticsearch：
output {if "app-log" in [fields][logtopic]{## es插件elasticsearch {# es服務(wù)地址hosts => ["192.168.212.128:9200"]## 索引名，%{index_time} 是由上面配置的 ruby 腳本定義的日期時(shí)間，即每天生成一個(gè)索引index => "app-log-%{[fields][logbiz]}-%{index_time}"# 是否嗅探集群ip：一般設(shè)置true# 只需要知道一臺(tái) elasticsearch 的地址，就可以訪問這一臺(tái)對(duì)應(yīng)的整個(gè) elasticsearch 集群sniffing => true# logstash默認(rèn)自帶一個(gè)mapping模板，進(jìn)行模板覆蓋template_overwrite => true}}if "error-log" in [fields][logtopic]{elasticsearch {hosts => ["192.168.212.128:9200"]index => "error-log-%{[fields][logbiz]}-%{index_time}"sniffing => truetemplate_overwrite => true} }
}

查看一下其中的過濾規(guī)則，這需要結(jié)合 《日志收集筆記（架構(gòu)設(shè)計(jì)、Log4j2項(xiàng)目初始化、Lombok）》 文章中的定義日志輸出格式一起看：

["message", "\[%{NOTSPACE:currentDateTime}\] \[%{NOTSPACE:level}\] \[%{NOTSPACE:thread-id}\] \[%{NOTSPACE:class}\] \[%{DATA:hostName}\] \[%{DATA:ip}\] \[%{DATA:applicationName}\] \[%{DATA:location}\] \[%{DATA:messageInfo}\] ## (\'\'|%{QUOTEDSTRING:throwable})"]

• "message"：logstash 固定的格式，統(tǒng)一叫傳入的數(shù)據(jù)為 message
• \[%{NOTSPACE:currentDateTime}\]：匹配 [ 為開頭，] 為結(jié)尾，NOTSPACE 表示不能有空格，賦值變量名為 currentDateTime
• \[%{NOTSPACE:level}\]：匹配 [ 為開頭，] 為結(jié)尾，NOTSPACE 表示不能有空格，賦值變量名為 level，日志級(jí)別
• \[%{NOTSPACE:thread-id}\]：匹配 [ 為開頭，] 為結(jié)尾，NOTSPACE 表示不能有空格，賦值變量名為 thread-id，線程ID
• \[%{NOTSPACE:class}\]：匹配 [ 為開頭，] 為結(jié)尾，NOTSPACE 表示不能有空格，賦值變量名為 class，創(chuàng)建對(duì)應(yīng) logger 實(shí)例傳入的 class
• \[%{DATA:hostName}\]：匹配 [ 為開頭，] 為結(jié)尾，DATA 表示數(shù)據(jù)，可為空，賦值變量名為 level，當(dāng)前應(yīng)用主機(jī)名稱
• \[%{DATA:ip}\]：匹配 [ 為開頭，] 為結(jié)尾，DATA 表示數(shù)據(jù)，可為空，賦值變量名為 level，當(dāng)前應(yīng)用的 IP
• \[%{DATA:applicationName}\]：匹配 [ 為開頭，] 為結(jié)尾，DATA 表示數(shù)據(jù)，可為空，賦值變量名為 level，當(dāng)前應(yīng)用的 applicationName
• \[%{DATA:location}\]：匹配 [ 為開頭，] 為結(jié)尾，DATA 表示數(shù)據(jù)，可為空，賦值變量名為 location
• \[%{DATA:messageInfo}\]：匹配 [ 為開頭，] 為結(jié)尾，DATA 表示數(shù)據(jù)，可為空，賦值變量名為 messageInfo，日志輸出的自定義內(nèi)容
• (\'\'|%{QUOTEDSTRING:throwable})：兩個(gè) ' 單引號(hào)之間的 | 表示，之間可為空，不為空就是 throwable 異常信息

啟動(dòng) logstash：

/usr/local/logstash-6.4.3/bin/logstash -f /usr/local/logstash-6.4.3/script/logstash-script.conf
## 如果測(cè)試時(shí)，想要控制臺(tái)輸出日志，輸入以下命令
/usr/local/logstash-6.4.3/bin/logstash -f /usr/local/logstash-6.4.3/script/logstash-script.conf --verbose --debug