1.前言
Filebeat是一个轻量级的开源日志数据收集器，用于将日志和文件数据从不同的来源发送到各种目的地，如Elasticsearch、Logstash和Kafka等。它是Elastic Stack（之前称为ELK Stack）的一部分，用于实时数据传输和集中式日志管理，Filebeat的主要作用是实时监控指定的日志文件或位置，以及持续不断地将新数据发送到目标位置。它可以解析各种格式的日志数据，并将其转发到配置的输出目标
在标准架构中，filebeat通常用来部署在每个需要收集日志的主机上，收集主机中的日志，然后统一发送到logstash中但在日志量大的架构中，需要加入消息队列，可以有效避免日志的丢失和避免日志量的暴涨导致logstash崩溃
2.filebeat的使用
filebeat收集日志输出到elasticsearch
vi /opt/filebeat/filebeat/filebeat.yml
filebeat.inputs:
- type: log #定义类型为logenabled: true #启用该配置paths: #日志的路径配置- /var/log/nginx/access.logfields:log_type: nginx_access #自定义名称，在有多个日志路径的情况下，可以将日志通过自定义的名称字段区分开来，分别输入不同的elasticsearch索引中，后面kafka也会用到tail_files: truejson.keys_under_root: true #输入的日志为json格式时需要配置本项与下一项参数json.overwrite_keys: true
# ======================= Elasticsearch template setting =======================
setup.template.settings:index.number_of_shards: 1 #配置elasticsearch主分片数量，但是配置了好像无效，还是默认的一个主分片
setup.template.name: "nginx" #这个配置项用于设置 Elasticsearch index template(索引模板) 的名称，这里设置为nginx
setup.template.pattern: "nginx-*" #这个配置项用于设置 Elasticsearch index pattern（索引模式），这里设置为 "nginx-*"，表示所有以 "nginx-" 开头的 index 都会使用这个 template
setup.template.overwrite: false #这个配置项用于设置是否覆盖已经存在的 template。这里设置为 false，表示如果已经存在同名的 template，则不会被覆盖
setup.template.json.enabled: true #这个配置项用于设置是否启用 JSON 格式的 template。这里设置为 true，表示使用 JSON 格式的 template
setup.template.enabled: false #这个配置项用于设置是否启用 Elasticsearch index template。这里设置为 false，表示不使用 Elasticsearch index template(索引模板)
setup.ilm.enabled: false #这个配置项用于设置是否启用 Index Lifecycle Management（ILM）。这里设置为 false，表示不使用 ILM
# ---------------------------- Elasticsearch Output ----------------------------
output.elasticsearch:hosts: ["10.1.60.114:9200","10.1.60.115:9200","10.1.60.118:9200"] #配置elasticsearch地址index: "%{[fields.log_type]}-%{+yyyy.MM.dd}" #配置elasticsearch索引名称，此处使用了自定义的名称变量和时间变量作为索引名称，这样就能实现再有多个不同日志的情况下，自动创建多个不同的索引，还能根据时间自动切割每日的日志，关于多个日志的配置，下面kafka配置会展示filebeat收集日志输出到logstash
vi /opt/filebeat/filebeat/filebeat.yml
filebeat.inputs: #此处配置，收集了两个不同的日志
- type: logenabled: truepaths:- /var/log/nginx/access.logfields:log_type: nginx-access-log #自定义名称，在有多个日志路径的情况下，可以将日志通过自定义的名称字段区分开来，分别输入不同的elasticsearch索引中，后面kafka也会用到tail_files: true
- type: logenabled: truepaths:- /var/log/nginx/error.logfields:log_type: nginx-error-log #自定义名称，在有多个日志路径的情况下，可以将日志通过自定义的名称字段区分开来，分别输入不同的elasticsearch索引中，后面kafka也会用到tail_files: true# ------------------------------ Logstash Output -------------------------------
output.logstash:hosts: ["10.1.60.115:5044"] #配置logstash地址，此端口需要根据logstash配置的填写logstash配置可以参考：logstash收集日志到elasticsearch_Apex Predator的博客-CSDN博客
filebeat收集日志输出到kafka
vi /opt/filebeat/filebeat/filebeat.yml
filebeat.inputs:
- type: logenabled: truepaths:- /var/log/nginx/access.logfields:log_type: nginx-access-log #自定义名称，在有多个日志路径的情况下，可以将日志通过自定义的名称字段区分开来，分别输入不同的elasticsearch索引中，后面kafka也会用到tail_files: true
- type: logenabled: truepaths:- /var/log/nginx/error.logfields:log_type: nginx-error-log #自定义名称，在有多个日志路径的情况下，可以将日志通过自定义的名称字段区分开来，分别输入不同的elasticsearch索引中，后面kafka也会用到tail_files: true# ------------------------------ kafka Output -------------------------------
#filebeat配置文件里默认是没有kafka的配置的，需要自己添加
output.kafka:enabled: true #开启output到kafkahosts: ["10.1.60.112:9092","10.1.60.114:9092","10.1.60.115:9092"] #配置kafka集群地址和端口topic: '%{[fields][topic]}' #主题名称，通过以上自定义的字段变量分别自动写入对应的消息队列中，若是没有创建好的队列，则会自动创建，但是创建的队列只有一个分区且没有副本，建议提前根据名称在kafka创建好分区
后续的调用参考logstash：logstash收集日志到elasticsearch_Apex Predator的博客-CSDN博客
