elk6.3.1+zookeeper+kafka+filebeat收集dockerswarm容器日志

前面有说过使用 redis 来缓解 elk 的数据接受压力，但是呢，如果 redis 面对突发情况也会承受不住的，这里需要借助两个工具， zookeeper 和 kafka

Zookeeper 主要值借助分布式锁，保证事务的不变，原子性隔离性。。。

Kafka 消息队列，从生产这到 filebeta 再到消费这 logstash 接受到 es 中，起到缓存，减缓压力

来吧开始往上怼了

首先下载 zookeeper 和卡夫卡

wget http: //mirrors.hust.edu.cn/apache/zookeeper/zookeeper-3.4.10/zookeeper-3.4.10.tar.gz

wget http: //mirror.bit.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

这里需要注意我是单台服务器安装的，要添加 hosts

这里 zookeeper 和卡夫卡的安装可以参考文档：

http://www.cnblogs.com/saneri/p/8822116.html

只需要 zookeeper 和 kafka 参考就行了，注意修改 ip 和 hostname

完成后验证：

Zookeeper+Kafka 集群测试

创建 topic ：

显示 topic ：

行了这个成功之后开始配置 filebeat ，这里换是收集 dockerswarm 集群的 tomat 和 nginx 容器的日志

filebeat.prospectors:

- type: log

enabled: true

paths:

- /var/log/docker-nginx/access_json.log

fields:

log_topics: 192.168.9.36-nginx

- type: log

enabled: true

paths:

- /var/log/docker-tomcat/catalina.out

fields:

log_topics: 192.168.9.36-tomcat

#    include_lines: ['ERROR','WARN']

#    exclude_lines: ['DEBUG']

output.kafka:

enabled: true

hosts: ["node1:9092"]

topic: '%{[fields][log_topics]}'

partition.hash:

reachable_only: true

compression: gzip

max_message_bytes: 1000000

required_acks: 1

接下来配置 logstash

input {

kafka{

bootstrap_servers => "node1:9092"

topics => ["192.168.9.36-nginx","192.168.9.36-tomcat"]

codec => "json"

consumer_threads => 1

decorate_events => true

auto_offset_reset => "latest"

}

}

filter{

date{

match=>["logdate","MMM dd HH:mm:ss yyyy"]

target=>"@timestamp"

timezone=>"Asia/Shanghai"

}

ruby{

code =>"event.timestamp.time.localtime+8*60*60"

}

}

output {

if [fields][log_topics] == "192.168.9.36-nginx" {

elasticsearch {

hosts => ["http://192.168.9.142:9200"]

index => "192.168.9.36-nginx-%{[@metadata][version]}-%{+YYYY.MM.dd}"

}

}

if [fields][log_topics] == "192.168.9.36-tomcat" {

elasticsearch {

hosts => ["http://192.168.9.142:9200"]

index => "192.168.9.36-tomcat-%{[@metadata][version]}-%{+YYYY.MM.dd}"

}

}

}

完成后启动，然后测试一下

去查看一下