filebeat主要用于收集和转发日志。filebeat监视指定的日志文件和位置，收集日志事件，并将它们转发到es或logstash进行索引。
安装
官网：https://www.elastic.co/guide/en/beats/filebeat/current/filebeat-installation.html
# curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-6.5.2-x86_64.rpm
# sudo rpm -vi filebeat-6.5.2-x86_64.rpm
说明
Filebeat的工作原理：启动Filebeat时，它会启动一个或多个inputs，这些inputs将查找指定的log的路径。对于查找到的每个日志，Filebeat将启动一个harvester。每个harvester读取单个日志的新内容，并将新日志数据发送到libbeat，libbeat聚合事件并将聚合数据发送到配置的output。
我们采用的是 filebeat -> kafka ->logstash -> es。
配置文件
filebeat 配置文件比较简单, 只需要配置一个filebeat.yml
input
#============== Filebeat prospectors ===========
filebeat.inputs: # 6.3以前是 filebeat.prospectors:
- type: log # input类型,默认为log，6.0以前配置是 - input_type: log
paths:
- /var/log/java/test-java.log
- /var/log/java/*.log
- /var/log/*/*.log
fields:
log_topic: java_log # 附加的可选字段,以向output添加额外的信息。output里面可以使用这个变量
multiline.pattern: '^\[' #多行合并 同logstash的multiline用法相似
multiline.negate: true
multiline.match: after - type: log # 一个配置文件里面可以同时收集多个日志
enabled: true
paths:
- /var/log/nginx/*.log
fields:
my_log_topic: nginx_log
常用配置
type: 指定input类型，有log、stdin、redis、tcp、syslog等
fields： 可以自定义添加字段，到output中去，后续可以使用这些字段，例如：
fields:
log_topic: mysql_slow
idc: szjf
tail_files: 默认为false。配置为true时，filebeat将从新文件的最后位置开始读取,而不是从开头读取新文件， 注意：如果配合日志轮循使用，新文件的第一行将被跳过。
此选项适用于Filebeat尚未处理的文件。如果先前运行了Filebeat并且文件的状态已经保留，tail_files则不会应用。
第一次运行Filebeat时，可以使用 tail_files: true 来避免索引旧的日志行。第一次运行后，建议禁用此选项。
registry file
filebeat会将自己处理日志文件的进度信息写入到registry文件中，以保证filebeat在重启之后能够接着处理未处理过的数据，而无需从头开始。
如果要让filebeat从头开始读文件，需要停止filebeat，然后删除registry file:
systemctl stop filebeat ;rm -fr /var/lib/filebeat/registry/* ; systemctl start filebaet
registry文件里字段的解释：
- source： 日志文件的路径
- offset：已经采集的日志的字节数;已经采集到日志的哪个字节位置
- inode： 日志文件的inode号
- device： 日志所在的磁盘编号
- timestamp： 日志最后一次发生变化的时间戳
- ttl： 采集失效时间，-1表示只要日志存在，就一直采集该日志
multiline: https://www.elastic.co/guide/en/beats/filebeat/current/multiline-examples.html#multiline 处理多行合并
output
配置输出的位置，比如kafka，logstash，es，redis，file 等
#================================ Outputs =====================================
#-------------------------Kafka output ------------------------------
output.kafka:
hosts: ["kafka01:9092","kafka02:9092","kafka03:9092]
topic: '%{[fields.my_log_topic]}'
partition.round_robin:
reachable_only: false
version: 0.10.0.0
required_acks: 1
compression: none
max_message_bytes: 1000000
#output.logstash:
# hosts: ["localhost:5044"]
还可以有一些其他配置，比如processor处理器配置（不是必需的）
#====================Preocessors=====================
processors:
- drop_fields: #删除字段，不再kibana里面展示，默认情况kibana里面会自动展示这些beat字段
fields: ["beat.hostname","beat.version","beat.name","prospector.type","input.type"] #也可以在logstash里面通过remove_field 删除字段
#mutate{
# remove_field => ["[beat][hostname]","[beat][version]","input","prospector"]
# }
配置完filebeat.yml，启动filebeat 即可。# /etc/init.d/filebeat start
使用filebeat遇到的问题
手动在kafka创建topic后，再重新启动filebeat， 问题解决。
./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 6 --topic nginx_log
经检查发现是因为kafka配置文件里面配置了 auto.create.topics.enable = false。auto.create.topics.enable 参数可以用来配置Kafka Server是否自动创建topic。