ELK + Filebeat + Kafka 分布式日志管理平台搭建

写在前面：2020年面试必备的Java后端进阶面试题总结了一份复习指南在Github上，内容详细，图文并茂，有需要学习的朋友可以Star一下！
GitHub地址：https://github.com/abel-max/Java-Study-Note/tree/master

1 工作流程

在这之前，我写了三篇文章关于日志系统平台的搭建，我这边现简单列出这几种的工作流程

1.1 ELK

Docker整合ELK实现日志收集

1.2 ELFK

docker 安装ELFK 实现日志统计

1.3 架构演进

• ELK缺点：ELK架构，并且Spring Boot应用使用 logstash-logback-encoder 直接发送给 Logstash，缺点就是Logstash是重量级日志收集server，占用cpu资源高且内存占用比较高
• ELFK缺点：一定程度上解决了ELK中Logstash的不足，但是由于Beats 收集的每秒数据量越来越大，Logstash 可能无法承载这么大量日志的处理

1.4 日志新贵ELK + Filebeat + Kafka

随着 Beats 收集的每秒数据量越来越大，Logstash 可能无法承载这么大量日志的处理。虽然说，可以增加 Logstash 节点数量，提高每秒数据的处理速度，但是仍需考虑可能 Elasticsearch 无法承载这么大量的日志的写入。此时，我们可以考虑 引入消息队列 ，进行缓存：

• Beats 收集数据，写入数据到消息队列中。
• Logstash 从消息队列中，读取数据，写入 Elasticsearch 中

如下就是其工作流程

2. ELK + Filebeat + Kafka 分布式日志管理平台搭建

2.1 ELFK的搭建

docker 安装ELFK 实现日志统计

2.1.1 Filebeat变动

由于我们架构演变，在filebeat中原来由传输到logstash改变为发送到kafka，我们这边filebeat.yml改动的部分为：

filebeat.inputs:
- type: logenabled: truepaths:- /var/logs/springboot/sparksys-authorization.log # 配置我们要读取的 Spring Boot 应用的日志fields:#定义日志来源，添加了自定义字段log_source: authorization
- type: logenabled: truepaths:- /var/logs/springboot/sparksys-gateway.logfields:log_source: gateway
- type: logenabled: truepaths:- /var/logs/springboot/sparksys-file.logfields:log_source: file
- type: logenabled: truepaths:- /var/logs/springboot/sparksys-oauth.logfields:log_source: oauth
#================================ Outputs =====================================
#-------------------------- Elasticsearch output ------------------------------
#output.elasticsearch:# Array of hosts to connect to.# hosts: ["192.168.3.3:9200"]#----------------------------- Logstash output --------------------------------
#output.logstash:# The Logstash hosts
#  hosts: ["logstash:5044"]#----------------------------- kafka output --------------------------------
output.kafka:enabled: truehosts: ["192.168.3.3:9092"]topic: sparksys-log

添加kafka输出的配置，将logstash输出配置注释掉。hosts表示kafka的ip和端口号，topic表示filebeat将数据输出到topic为sparksys-log的主题下，此处也根据自己情况修改

2.1.2 Logstash变动

logstash.conf配置input由原来的输入源beat改为kafka

input {kafka {codec => "json"topics => ["sparksys-log"]bootstrap_servers => "192.168.3.3:9092"auto_offset_reset => "latest"group_id => "logstash-g1"}
}output {elasticsearch {hosts => "es:9200"index => "filebeat_%{[fields][log_source]}-%{+YYYY.MM.dd}"}
}

上述配置说明如下:

• topics后面的sparksys-log表示从kafka中topic为sparksys-log的主题中获取数据，此处的配置根据自己的具体情况去配置。
• bootstrap_servers表示配置kafka的ip与端口。

到此，ELFK的变动部分结束，接下来就是kafka的搭建

2.2 kafka搭建

2.2.1 新建docker-compose.yaml

配置如下：

version: '3'
services:zookeeper:image: zookeeper:latestcontainer_name: zookeepervolumes:- /Users/zhouxinlei/docker/kafka/zookeeper/data:/data- /Users/zhouxinlei/docker/kafka/zookeeper/datalog:/datalogports:- 2181:2181restart: alwayskafka:image: wurstmeister/kafkacontainer_name: kafkavolumes: - /Users/zhouxinlei/docker/kafka/data:/kafkaports:- 9092:9092environment:KAFKA_ADVERTISED_HOST_NAME: 192.168.3.3KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181KAFKA_ADVERTISED_PORT: 9092KAFKA_LOG_RETENTION_HOURS: 120KAFKA_MESSAGE_MAX_BYTES: 10000000KAFKA_REPLICA_FETCH_MAX_BYTES: 10000000KAFKA_GROUP_MAX_SESSION_TIMEOUT_MS: 60000KAFKA_NUM_PARTITIONS: 3KAFKA_DELETE_RETENTION_MS: 1000restart: alwayskafka-manager:  image: kafkamanager/kafka-managercontainer_name: kafka-managerenvironment:ZK_HOSTS: 192.168.3.3ports:  - 9001:9000restart: always

2.2.3 创建并启动kafka容器

docker-compose up -d

2.2.2 访问 http://192.168.3.3:9001

• 进入kafka-manager web页面新建cluster

• 列表展示

• 进入kafka01

• 新建topic

到此kafka的简单使用完成

3 ELK + Filebeat + Kafka 分布式日志管理平台使用测试

• Filebeat发送日志到kafka

• lofstash 消费kafka消息，输入日志到es中

• kabana查看日志

3 总结

1. 在部署的过程中可能会遇到各种情况，此时根据日志说明都可以百度处理（如部署的过程中不能分配内存的问题）。
2. 如果完成后如果数据显示不了，可以先到根据工作流程到各个节点查询数据是否存储和传输成功。如查询filebeat是否成功把数据传输到了kafka，可以进入kafka容器当中使用kafka中如下命令查询：

bin/kafka-console-consumer.sh –zookeeper localhost:2181 –topic 
sparksys-log –from-beginning

查看日志filebeat中的数据是否正常在kafka中存储。

docker logs -f --tail=200 filebeat

3. 该平台的搭建是比较简便的方式，大家可以更加灵活以及动态的配置该平台。