Filebeat的架構分析、配置解釋與示例
寫在前邊的話
在看filebeat之前我們先來看下Beats，Beats 平臺是 Elastic.co 從 packetbeat 發展出來的資料收集器系統。beat 收集器可以直接寫入 Elasticsearch，也可以傳輸給 Logstash。其中抽象出來的 libbeat，提供了統一的資料傳送方法，輸入配置解析，日誌記錄框架等功能。也就是說，所有的 beat 工具，在配置上，除了 input 以外，在output、filter、shipper、logging、run-options 上的配置規則都是完全一致的
而這裡的filebeat就是beats 的一員，目前beat可以傳送資料給Elasticsearch，Logstash，File，Console四個目的地址。filebeat 是基於原先 logstash-forwarder 的原始碼改造出來的。換句話說：filebeat 就是新版的 logstash-forwarder，也會是 ELK Stack 在 shipper 端的第一選擇。
Filebeat的架構設計
當我們安裝完filebeat之後，我們可以在filebeat的安裝目錄下看到兩個檔案
- filebeat.template.json （輸出的檔案格式，在filebeat的template中指定，當服務啟動時，會被載入）
- filebeat.yml（所有的配置都在該檔案下進行）
整體架構理解：
上邊我們也說了filebeat是用來收集日誌的，那麼在filebeat.yml中會配置指定的監聽檔案，也就是上圖中的一個個log，這個log的目錄是在prospectors中設定，在看配置檔案的時候便可以很明白的看出來，對於prospectors定位每個日誌檔案，Filebeat啟動harvester。每個harvester讀取新的內容一個日誌檔案，新的日誌資料傳送到spooler（後臺處理程式），它彙集的事件和聚合資料傳送到你已經配置了Filebeat輸出。
環境準備
要開始使用自己的Filebeat設定，安裝和配置這些相關產品：
- Elasticsearch儲存和索引資料。
- Kibana為UI。
- Logstash（可選）將資料插入到Elasticsearch。
部署filebeat
deb：
rpm：
mac：
win：
- 下載windows zip檔案 點選下載.
- 解壓檔案到 C:\Program Files.
- 重新命名為 Filebeat.
- 開啟PowerShell提示符作為管理員（右鍵單擊PowerShell的圖示，並選擇以管理員身份執行）。如果您執行的是Windows XP，則可能需要下載並安裝PowerShell
-
執行以下命令來安裝Filebeat作為Windows服務
cd ‘C:\Program Files\Filebeat’
C:\Program Files\Filebeat> .\install-service-filebeat.ps1
在啟動filebeat服務之前，需要先修改配置檔案，接下來我們看下配置檔案
配置解析
上邊我麼也說了FileBeat的四種輸出方式為輸出到Elasticsearch，logstash，file和console，下面我們具體看下示例
PS：
- 這裡說的是需要修改的配置檔案，沒有提的就是不需要修改
- 每次修改完配置檔案都需要重啟filebeat服務
- 這裡不要追究時間的問題，小主是測試，主要是為了方便記錄
這裡主要是自定義監聽檔案的路勁，我設定的是/opt/elk/log/*.log
然後在filebeat.yml中的prospectors路徑設定如下(該配置為以下四種方式通用)
paths:
- /opt/elk/log/*.log
1：output of Elasticsearch
filebeat.yml中output的配置將除了es之外註釋掉
output
elasticsearch:
hosts: ["192.168.197.128:9200"] #es的ip地址和埠，如果有多個，中間用逗號分隔
worker: 1 #對應es的個數
index: "filebeat" #索引根名稱
template:
name: "filebeat" #模板名字和對應的json檔案
path: "filebeat.template.json"
max_retries: 3 #傳送到特定logstash的最大嘗試次數。如果達到該次數仍不成功，事件將被丟棄。預設是3，值0表示禁用重試。值小於0將無限重試知道事件已經發布。
bulk_max_size: 20000 #單個elasticsearch批量API索引請求的最大事件數。預設是50
timeout: 90 #elasticsearch請求超時事件。預設90秒
flush_interval: 5 #新事件兩個批量API索引請求之間需要等待的秒數。如果bulk_max_size在該值之前到達，額外的批量索引請求生效。往log檔案中追加日誌
echo “123456789” >> test1.log
這個時候我們看一下效果： 
2：output of logstash
filebeat.yml中output的配置將除了logstash之外註釋掉
output:
logstash:
hosts: ["192.168.197.130:5044"]
worker: 2
loadbalance: true
index: filebeat
這裡 worker 的含義，是 beat 連到每個 host 的執行緒數。
在 loadbalance 開啟的情況下，意味著有 4 個worker 輪訓傳送資料則對應的logstash配置，編寫一個配置檔案
sudo vim filebeat_logstash_out.conf
beat 寫入 Logstash 時，會配合 Logstash-1.5 後新增的 metadata 特性。將 beat 名和 type 名 記錄在 metadata 裡。所以對應的 Logstash 配置應該是這樣：
input {
beats {
port => 5044
}
}
output {
elasticsearch {
hosts => ["http://192.168.197.128:9200"]
index => "%{[@metadata][beat]}-%{+YYYY.MM.dd}"
document_type => "%{[@metadata][type]}"
}
stdout{
codec=>rubydebug
}
}啟動conf
bin/logstash -f config/filebeat_logstash_out.conf
往log檔案中追加日誌
echo “2222333344445555” >> test1.log
前往檢視效果： 
3：output of File
filebeat.yml中output的配置將除了file之外註釋掉
output：
file:
path: "/opt/elk/log/filebeat_output_file"
filename: filebeat
rotate_every_kb: 10000
number_of_files: 7往log檔案中追加日誌：
echo “this is filebeat output of file” >> test1.log
檢視效果： 
4：output of Console
filebeat.yml中output的配置將除了cosnsole之外註釋掉
output：
console:
pretty: true重啟啟動服務：
sudo filebeat -e -c /etc/filebeat/filebeat.yml
往log檔案中追加日誌：
echo “this is filebeat output of console” >> test1.log
前往服務啟動視窗檢視效果： 
ELK+Filebeat的Demo
1：在/opt/elk/log目錄下有三個檔案分別是test1.log，test2.log，test3.log**
2：通過Python指令碼往三個檔案中追加內容，內容格式如下：
| name | age | address | company |
| aa | 23 | beijing | alibaba |
Python的指令碼內容如下：
#-*-coding:utf-8-*-
name_list = ['aa','bb','cc','dd','ee']
addr_list = ['beijing','shanghai','guangzhou']
company_list =['baidu','tengxun','alibb']
import random
import time
def get_name():
return name_list[random.randint(0,4)]
def get_age():
return random.randint(20,25)
def get_addr():
return addr_list[random.randint(0,2)]
def get_comp():
return company_list[random.randint(0,2)]
if __name__=="__main__":
while True:
print("%s %s %s %s"%(get_name(),get_age(),get_addr(),get_comp()))
str_line = get_name()+" "+str(get_age())+" "+get_addr()+" "+get_comp()
import os
os.system("echo %s >> test1.log" % str_line)
time.sleep(1)
3：filebeat 的配置檔案使用output=logstash
filebeat.yml
output:
logstash:
hosts: ["192.168.197.130:5044"]
worker: 2
loadbalance: true
index: filebeat
4：filebeat_logstash_out.conf
input {
beats {
port => 5044
}
}
filter{
if [type] == 'log'{
grok{
match=>{
"message"=>"%{WORD:username} %{WORD:age} %{WORD:address} %{WORD:company}"
}
}
}
}
output {
elasticsearch {
hosts => ["http://192.168.197.128:9200"]
index => "%{[@metadata][beat]}-%{+YYYY.MM}"
document_type => "%{[@metadata][type]}"
}
stdout{
codec=>rubydebug
}
}5：啟動服務
- 啟動python指令碼
- 啟動conf配置檔案
6：web檢視結果
