前言
前文解決了鏈路的基本部署和配置調(diào)整，見專題 集中式日志分析平臺(tái) - ELK Stack弃理。在實(shí)際的生產(chǎn)過程中，我們需要把多個(gè)不同來源的日志按照不同的 index 寫入 ES，這樣方便針對不同的數(shù)據(jù)進(jìn)行匯總分析。本文介紹如何根據(jù) type 和 document_type 進(jìn)行日志分離治筒，達(dá)到我們所期望的效果，整個(gè)配置步驟我們是從數(shù)據(jù)流的產(chǎn)生到傳輸?shù)铰涞剡@個(gè)角度去闡述舷蒲。
本文針對的是整合了 Kafka 作為緩沖消息隊(duì)列的架構(gòu)：
filebeat -> kafka -> logstash -> elasticsearch -> kibana
Filebeat 端配置
Filebeat 官方文檔中描述了一個(gè)配置耸袜，可以對數(shù)據(jù)流按照 input_type 級(jí)別的打標(biāo)，通過 document_type 關(guān)鍵字申明：
filebeat.prospectors:
- input_type: log
paths:
- /home/admin/soft/filebeat-5.2.1-linux-x86_64/logs/filebeat
document_type: fb_log
- input_type: log
paths:
- /home/admin/soft/elasticsearch-5.2.1/logs/elasticsearch.log
document_type: es_log
比如上面的配置表述牲平，prospectors 具有 2 個(gè)采集輸入堤框，分別是采集 filebeat 日志和 elasticsearch 日志，filebeat 日志打標(biāo)為 fb_log欠拾，elasticsearch 日志打標(biāo)為 es_log胰锌。
然后，我們的輸出是 kafka藐窄，我們可以通過宏去通配推送數(shù)據(jù)到不同的 kafka topic ：
output.kafka:
hosts: ["172.16.134.3:9092"]
topic: '%{[type]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
這樣配置资昧，我們的 filebeat 日志和 elasticsearch 日志會(huì)推送到不同的 kafka topic，分別是 fb_log 和 es_log荆忍，配置完成后記得重啟格带。
Logstash 端配置
和 filebeat 對應(yīng)的撤缴， logstash 的 input / filter / output 配置段，也有類似的打標(biāo)方式：
input {
kafka {
bootstrap_servers => "${BOOTSTRAP_SERVERS}"
consumer_threads => 3
topics => "fb_log"
type => "fb_log"
}
kafka {
bootstrap_servers => "${BOOTSTRAP_SERVERS}"
consumer_threads => 3
topics => "es_log"
type => "es_log"
}
}
output {
if [type] == "fb_log" {
elasticsearch {
hosts => "${HOSTS}"
manage_template => false
index => "fb_log-%{+YYYY.MM.dd}"
document_type => "fb_log"
}
}
if [type] == "es_log" {
elasticsearch {
hosts => "${HOSTS}"
manage_template => false
index => "es_log-%{+YYYY.MM.dd}"
document_type => "es_log"
}
}
}
配置完成后重啟叽唱。
和之前的文檔類似屈呕，我們在 Kibana Discover 中，需要添加剝離后的 Index棺亭，比如按照 fb_log-* 去搜索虎眨。
小結(jié)
本文簡述了，在 ELK 生產(chǎn)環(huán)境中镶摘，如何進(jìn)行按照日志來源區(qū)別的 ES index 配置嗽桩。
