Filebeat日志采集与ELK+Kafka缓冲架构部署

一、Filebeat采集Elasticsearch集群日志1.1 部署下载安装filebeat[rootserver1 ~]# rpm -ivh filebeat-7.6.1-x86_64.rpm[rootserver1 ~]# cd /etc/filebeat/[rootserver1 filebeat]# cd modules.d/将elasticsearch允许打开启用Elasticsearch内置模块[rootserver1 modules.d]# filebeat modules enable elasticsearchEnabled elasticsearch进入这个文件后可以看到文件上方的链接复制在浏览器访问会有指导文件可以按照指导文件[rootserver1 modules.d]# vim elasticsearch.ymlElasticsearch module | Filebeat Reference [7.6] | Elastic修改elasticsearch.yml模块配置打开var.paths模块并写上指导文件中的内容记得有两个缩进格式很重要[rootserver1 modules.d]# cd ..[rootserver1 filebeat]# lsfields.yml filebeat.reference.yml filebeat.yml modules.d修改文件配置输出目标使集群日志能够输出一般统一节点[rootserver1 filebeat]# vim filebeat.yml[rootserver1 modules.d]# systemctl enable --now filebeat指定采集到的日志发送到ES集群开启安全认证时必须填写elastic账号密码否则无法写入索引。安全认证在上一篇博客记得xpack认证时的密码配置好一台后将文件拷贝给其他两个节点将配置好的模块文件远程拷贝至另外两台 ES 服务器之后在 server2、server3 启动 filebeat实现三台节点日志全部采集。[rootserver1 modules.d]# scp elasticsearch.yml server2:/etc/filebeat/modules.d/[rootserver1 modules.d]# scp elasticsearch.yml server3:/etc/filebeat/modules.d此时访问浏览器http://192.168.159.155:5601可以看到有日志1.2 问题有一台节点老是下线将以下文件中的打开锁定ES进程内存不让操作系统把内存交换到swap磁盘swap会严重拖慢ES性能加剧节点失联。并将内存改为512m然后三台节点都要一样[rootserver3 elasticsearch]# vim /etc/elasticsearch/elasticsearch.yml[rootserver3 elasticsearch]# vim /etc/elasticsearch/jvm.options然后记得一个一个修改重新启动否则很有可能起不来修改好后集群状态稳定二、ELKKafka日志缓冲Kafka是分布式高吞吐消息中间件生产者向主题发送消息消息存储在分区并支持副本保障数据安全消费者以消费组形式订阅拉取数据依靠偏移量记录消费位置。它主要用于系统异步解耦、流量削峰、采集日志与流转实时数据流支持消息持久化与历史数据回溯适合海量流式数据传输常搭配大数据实时计算组件使用。基本架构Filebeat采集→ Kafka缓冲→ Logstash处理→ Elasticsearch存储→ Kibana可视化2.1 部署kafka新加三台虚拟机server6、server7、server8做kafka三台虚拟机都是一样的操作安装JDKKafka运行依赖JVM[rootserver6 ~]# rpm -ivh jdk-17.0.17_linux-x64_bin.rpm[rootserver6 ~]# java -version解压部署 Kafka[rootserver6 ~]# tar zxf kafka_2.12-3.9.1.tgz -C /opt[rootserver6 ~]# mv /opt/kafka_2.12-3.9.1 /opt/kafka[rootserver6 ~]# cd /opt/kafka/config/kraft[rootserver6 kraft]# cp server.properties{,.bak} # 备份原始配置[rootserver6 kraft]# mkdir /opt/kafka/logs # 创建Kafka数据存储目录批量分发安装包到另外两台节点server7、server8 执行相同解压、创建目录操作。[rootserver6 ~]# scp jdk-17.0.17_linux-x64_bin.rpm server7:[rootserver6 ~]# scp jdk-17.0.17_linux-x64_bin.rpm server8:[rootserver6 ~]# scp kafka_2.12-3.9.1.tgz server7:[rootserver6 ~]# scp kafka_2.12-3.9.1.tgz server8:修改kraft模式配置文件server.propertiesrootserver6 ~]# cd /opt/[rootserver6 opt]# cd kafka/[rootserver6 kafka]# cd config/[rootserver6 config]# cd kraft/[rootserver6 kraft]# vim server.propertiesserver7、server8 修改node.id、advertised.listeners内IP其余配置保持一致。process.rolesbroker,controller# 当前节点同时承担数据节点broker集群控制器controllerKRaft模式node.id156# 集群内节点唯一IDserver7改为157server8改为158controller.quorum.voters156192.168.159.156:9093,157192.168.159.157:9093,158192.168.159.158:9093# 集群控制器投票节点列表参与选主listenersPLAINTEXT://:9092,CONTROLLER://:9093# 本机监听端口inter.broker.listener.namePLAINTEXTadvertised.listenersPLAINTEXT://192.168.159.156:9092,CONTROLLER://192.168.159.156:9093# 对外广播访问地址其他机器通过此地址连接每台机器修改为本机IPcontroller.listener.namesCONTROLLERlistener.security.protocol.mapCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLnum.network.threads3num.io.threads8socket.send.buffer.bytes102400socket.receive.buffer.bytes102400socket.request.max.bytes104857600log.dirs/opt/kafka/logs # 消息持久化存储目录num.partitions3 # Topic默认分区数num.recovery.threads.per.data.dir1offsets.topic.replication.factor3 # 偏移量主题副本数集群3台设3保证高可用transaction.state.log.replication.factor3transaction.state.log.min.isr1log.retention.hours168 # 消息默认保留7天log.segment.bytes1073741824log.retention.check.interval.ms300000生成集群唯一ID仅在一台节点执行[rootserver6 kraft]# KAFKA_CLUSTER_ID$(/opt/kafka/bin/kafka-storage.sh random-uuid)[rootserver6 kraft]# echo ${KAFKA_CLUSTER_ID}输出一串UUID三台节点格式化数据都使用同一个ClusterID。格式化Kafka数据目录三台全部执行只需要格式化一次重复格式化会清空所有数据。三台节点格式化数据都使用同一个ClusterID[rootserver6 kraft]# /opt/kafka/bin/kafka-storage.sh format -t 1GROlYMiRkWVzED10qJ0tw -c /opt/kafka/config/kraft/server.properties后台启动Kafka集群三台一次启动[rootserver6 kraft]# /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/kraft/server.properties[rootserver6 kraft]# jps #验证进程创建测试主题testtopics[rootserver6 kraft]# /opt/kafka/bin/kafka-topics.sh --create --topic testtopics --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3[rootserver7 kraft]# /opt/kafka/bin/kafka-topics.sh --list --bootstrap-server localhost:9092功能测试生产者输入内容消费者正常接收Kafka 集群通信正常。[rootserver6 kraft]# /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic testtopics[rootserver8 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testtopics --from-beginning[rootserver7 kraft]# /opt/kafka/bin/kafka-topics.sh --bootstrap-server server7:9092,server8:9092,server9:9092 --topic testtopics --describe2.2 配置Filebeat日志输出到Kafka修改filebeat.yml[rootserver5 filebeat]# vim filebeat.ymlhostskafka集群地址列表topic日志发送到 kafka 的主题apache-logsrequired_acks:1broker写入本地日志即返回兼顾性能与可靠性compression:gzip开启压缩减少网络流量。[rootserver5 filebeat]# filebeat test config -c /etc/filebeat/filebeat.yml[rootserver5 filebeat]# filebeat test output -c /etc/filebeat/filebeat.yml检查配置语法测试连通kafka重启生效启动Kafka消费者消费apache-logs可以收到Filebeat推送的Apache访问日志[rootserver8 kraft]# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic apache-logs --from-beginning2.3 配置Logstash[rootserver5 ~]# cd /etc/logstash/[rootserver5 logstash]# cd conf.d/[rootserver5 conf.d]# vim kafka.conf记得把打印输出打开否则压测完后filebeat就输出不了信息input连接Kafka集群订阅apache-logs主题作为日志来源filter grok使用内置正则HTTPD_COMBINEDLOG自动拆分Apache日志把文本拆分为clientip、status、request、bytes等结构化字段output将清洗完成的数据写入Elasticsearch索引按日期自动生成apachelog-年月日stdout {} 仅调试使用压测时注释避免大量控制台输出占用资源。前台启动Logstash测试[rootserver5 conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/kafka.conf然后进行压测有值并且可以看到[rootserver3 ~]# ab -c1 -n 50 http://192.168.159.155/index.html[rootserver1 ~]# ab -c1 -n 100 http://192.168.159.155/index.html能够看到有正常输出访问kibana也能正常看到数据访问http://192.168.159.154:9100可以查看到索引apachelog-2026.08.13三、总结3.1 filebeat采集日志总结1.Filebeat启动失败大概率YAML缩进格式错误2. Kibana无日志检查Filebeat状态、ES端口连通性、账号密码3. ES节点反复掉线优先排查JVM堆内存大小、主机内存、Swap使用情况。3.2 ELKkafka缓冲总结1.KRaft模式必须使用同一个ClusterID格式化所有节点ID不一致集群无法组建2. server.properties内advertised.listeners的IP填写错误外部无法连接Kafka3. Filebeat测试output失败检查9092端口、kafka进程状态4. Logstash收不到数据核对topic名称、consumer-group调试打开stdout观察输出