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

妖狐艹你老母 2022-11-05 00:57 14阅读 0赞

> ### 推荐阅读 ###
> 
> ##### Helm3（K8S 资源对象管理工具）视频教程：[https://edu.csdn.net/course/detail/32506][https_edu.csdn.net_course_detail_32506] #####
> 
> ##### Helm3（K8S 资源对象管理工具）博客专栏：[https://blog.csdn.net/xzk9381/category\_10895812.html][https_blog.csdn.net_xzk9381_category_10895812.html] #####

> [本文原文链接：https://blog.csdn.net/xzk9381/article/details/114324351，转载请注明出处。如有发现文章中的任何问题，欢迎评论区留言。][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][https_blog.csdn.net_xzk9381_article_details_109535450]

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

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

1.  在 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.

1.  在 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]

1.  编写如下 Filebeat 配置文件：

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

1.  分别向 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

1.  启动 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

1.  连接 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_edu.csdn.net_course_detail_32506]: https://edu.csdn.net/course/detail/32506
[https_blog.csdn.net_xzk9381_category_10895812.html]: https://blog.csdn.net/xzk9381/category_10895812.html
[https_blog.csdn.net_xzk9381_article_details_114324351]: https://blog.csdn.net/xzk9381/article/details/114324351
[https_blog.csdn.net_xzk9381_article_details_109535450]: https://blog.csdn.net/xzk9381/article/details/109535450