前言
前文解决了链路的基本部署和配置调整，见专题 集中式日志分析平台 - ELK Stack。在实际的生产过程中，我们需要把多个不同来源的日志按照不同的 index 写入 ES，这样方便针对不同的数据进行汇总分析。本文介绍如何根据 type 和 document_type 进行日志分离，达到我们所期望的效果，整个配置步骤我们是从数据流的产生到传输到落地这个角度去阐述。
本文针对的是整合了 Kafka 作为缓冲消息队列的架构：
filebeat -> kafka -> logstash -> elasticsearch -> kibana
Filebeat 端配置
Filebeat 官方文档中描述了一个配置，可以对数据流按照 input_type 级别的打标，通过 document_type 关键字申明：
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 个采集输入，分别是采集 filebeat 日志和 elasticsearch 日志，filebeat 日志打标为 fb_log，elasticsearch 日志打标为 es_log。
然后，我们的输出是 kafka，我们可以通过宏去通配推送数据到不同的 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 日志会推送到不同的 kafka topic，分别是 fb_log 和 es_log，配置完成后记得重启。
Logstash 端配置
和 filebeat 对应的， logstash 的 input / filter / output 配置段，也有类似的打标方式：
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-* 去搜索。
小结
本文简述了，在 ELK 生产环境中，如何进行按照日志来源区别的 ES index 配置。
