Filebeat 将多个日志输出到不同的 Kafka Topic

推荐阅读

Helm3（K8S 资源对象管理工具）视频教程：https://edu.csdn.net/course/detail/32506

Helm3（K8S 资源对象管理工具）博客专栏：https://blog.csdn.net/xzk9381/category_10895812.html

本文原文链接：https://blog.csdn.net/xzk9381/article/details/114324351，转载请注明出处。如有发现文章中的任何问题，欢迎评论区留言。

平时在物理机上使用 Filebeat 收集日志并输出到 Kafka 中时，会编写多个 filebeat 配置文件然后启动多个 filebeat 进程来收集不同路径下的日志并推送到不同的 Topic。那么如果将所有的日志路径都写到一个 filebeat 配置文件中，那么就需要根据不同的日志来设置 Topic 了。

其实 logstash 也可以实现这个功能。但是此处只演示在 Filebeat 上实现。步骤和讲解如下：

如果希望将不同的日志收集到不同的索引中，可以参考我的另一篇文章：https://blog.csdn.net/xzk9381/article/details/109535450

例如现在有如下三个日志文件，需要输出到不同的 Topic：

access.log  ---->  Topic：topic-for-access-log
error.log   ---->  Topic：topic-for-error-log
info.log    ---->  Topic：topic-for-info-log

在 Kafka 中创建三个 Topic：

[@localhost ~]# for type in access error info; do /opt/kafka_cluster/kafka/bin/kafka-topics.sh --create --topic topic-for-${type}-log --zookeeper 10.16.12.204:2181 --partitions 1 --replication-factor 1; done
Created topic topic-for-access-log.
Created topic topic-for-error-log.
Created topic topic-for-info-log.

在 zookeeper 中查看是否已经存在相应的 Topic：

[@localhost ~]# /opt/kafka_cluster/zookeeper/bin/zkCli.sh
Connecting to localhost:2181
[zk: localhost:2181(CONNECTED) 0] ls /brokers/topics
[__consumer_offsets, topic-for-access-log, topic-for-error-log, topic-for-info-log]

编写如下 Filebeat 配置文件：

filebeat.idle_timeout: 2s
filebeat.name: filebeat-shiper
filebeat.spool_zie: 50000filebeat.inputs:                      # 从这里开始定义每个日志的路径、类型、收集方式等信息
- type: log                           # 指定收集的类型为 logpaths:- /opt/log_test/access.log         # 设置 access.log 的路径fields:                             # 设置一个 fields，用于标记这个日志 topic: topic-for-access-log       # 为 fields 设置一个关键字 topic，值为 kafka 中已经设置好的 topic 名称enabled: truebackoff: 1sbackoff_factor: 2close_inactive: 1hencoding: plainharvester_buffer_size: 262144max_backoff: 10smax_bytes: 10485760scan_frequency: 10stail_lines: true
- type: logpaths:- /opt/log_test/error.log          # 设置 error.log 的路径fields:                             # 设置一个 fields，用于标记这个日志 topic: topic-for-error-log        # 为 fields 设置一个关键字 topic，值为 kafka 中已经设置好的 topic 名称enabled: truebackoff: 1sbackoff_factor: 2close_inactive: 1hencoding: plainharvester_buffer_size: 262144max_backoff: 10smax_bytes: 10485760scan_frequency: 10stail_lines: true
- type: logpaths:- /opt/log_test/info.log          # 设置 info.log 的路径fields:                            # 设置一个 fields，用于标记这个日志 topic: topic-for-info-log        # 为 fields 设置一个关键字 topic，值为 kafka 中已经设置好的 topic 名称enabled: truebackoff: 1sbackoff_factor: 2close_inactive: 1hencoding: plainharvester_buffer_size: 262144max_backoff: 10smax_bytes: 10485760scan_frequency: 10stail_lines: trueoutput.kafka:                       # 指定输出到 Kafkabulk_flush_frequency: 0bulk_max_size: 2048codec.format:string: '%{[message]}'compression: gzipcompression_level: 4hosts:- 10.16.12.204:9092               # 指定 Kafka 的地址，如果是集群则写集群的地址max_message_bytes: 10485760partition.round_robin:reachable_only: truerequired_acks: 1topic: '%{[fields.topic]}'        # 根据每个日志设置的 fields.topic 来输出到不同的 topicworkers: 4setup.ilm.enabled: false

分别向 access、error、info 日志文件中写入如下内容：

echo "this is access log" > /opt/log_test/access.log
echo "this is error log" > /opt/log_test/error.log
echo "this is info log" > /opt/log_test/info.log

启动 Filebeat：

/opt/filebeat-7.3.0/filebeat run -c /opt/filebeat-7.3.0/conf/test.yaml -httpprof 0.0.0.0:15502 -path.logs /opt/filebeat-7.3.0/logs/filebeat_15502 -e

连接 Kafka 集群消费日志：

[@k8s-master1 ~]# /opt/kafka_cluster/kafka/bin/kafka-console-consumer.sh --topic topic-for-access-log --bootstrap-server 10.16.12.204:9092 --from-beginning
this is access log[@k8s-master1 ~]# /opt/kafka_cluster/kafka/bin/kafka-console-consumer.sh --topic topic-for-error-log --bootstrap-server 10.16.12.204:9092 --from-beginning
this is error log[@k8s-master1 ~]# /opt/kafka_cluster/kafka/bin/kafka-console-consumer.sh --topic topic-for-info-log --bootstrap-server 10.16.12.204:9092 --from-beginning
this is info log

可以看到不同路径的日志已经收集到 Kafka 不同的 Topic 中。

本文原文链接：https://blog.csdn.net/xzk9381/article/details/114324351，转载请注明出处。如有发现文章中的任何问题，欢迎评论区留言。

Filebeat 将多个日志输出到不同的 Kafka Topic相关推荐

ES Filebeat 使用 Pipeline 处理日志中的 @timestamp
推荐阅读 Helm3(K8S 资源对象管理工具)视频教程:https://edu.csdn.net/course/detail/32506 Helm3(K8S 资源对象管理工具)博客专栏:https: ...
ELK 使用filebeat替代Logstash收集日志
使用beats采集日志之前也介绍过beats是ELK体系中新增的一个工具,它属于一个轻量的日志采集器,以上我们使用的日志采集工具是logstash,但是logstash占用的资源比较大,没有beat ...
filebeat+logstash收集错误日志发送邮件提醒
filebeat+logstash收集错误日志发送邮件提醒典型ELK应用架构因为只收集错误日志并且数据量并不是非常大所以简化流程使用filebeat+logstash发送异常日志软件版本说 ...
CC00290.CloudKubernetes——|KuberNetes运维.V12|——|EFLK架构.v07|验证日志输出采集是否正常|
一.验证日志输出采集是否正常 ### --- 进入Filebeat-pod查看是否正常启动 ~~~ 查看Filebeat容器[root@k8s-master01 filebeat]# kubectl ...
Python之向日志输出中添加上下文信息
除了传递给日志记录函数的参数(如msg)外,有时候我们还想在日志输出中包含一些额外的上下文信息.比如,在一个网络应用中,可能希望在日志中记录客户端的特定信息,如:远程客户端的IP地址和用户名.这里我们 ...
python日志输出到屏幕，python日志写入文件
python日志输出到屏幕,python日志写入文件日志日志是跟踪软件运行时所发生的事件的一种方法.软件开发者在代码中调用日志函数,表明发生了特定的事件.事件由描述性消息描述,该描述性消息可以可选 ...
基于单例模式的日志输出(C++)
原文:https://blog.csdn.net/giser_mxd/article/details/75557410 在软件运行中,系统一般会采用一个持久化的日志系统来记录运行情况.本题你需要实现一 ...
Srping Boot日志输出（转）
说明:其实经过研究,在最新版本的Spring Boot中默认使用的是logback进行日志输出,其余的都没有引入.但是网上的教程说只要按照下面的文件列表引入对应的配置文件就会进行输出,这个没有实践过, ...
iOS根据Debug和Release状态的变化来屏蔽日志输出
今天在这里分享一个很实用的小技巧. 我们平时在开发应用的时候,经常会用到NSLog来调试我们的程序,而随着项目越来越大,这些用于调试的日志输出就会变得很难管理. 我们在发布正式版的时候一定要屏蔽掉所有 ...

Filebeat 将多个日志输出到不同的 Kafka Topic

推荐阅读

Helm3（K8S 资源对象管理工具）视频教程：https://edu.csdn.net/course/detail/32506

Helm3（K8S 资源对象管理工具）博客专栏：https://blog.csdn.net/xzk9381/category_10895812.html

Filebeat 将多个日志输出到不同的 Kafka Topic相关推荐

最新文章

热门文章