Filebeat 是一个用于转发和集中化日志数据的轻量级工具，它一般作为agent安装在服务器上。它监控日志文件和文件夹，收集日志事件，并转发到logstash、ES或者Kafka。
Filebeat的流程图如下，当filebeat运行后，其根据指定的log日志位置，启动一个或者多个input。 对于每一个定位跟踪的日志，filebeat都为启动一个独特的harvester(收割机), harvester会读取一条新日志内容，并将其发送到libbeat， libbeat收集一些列的事件，并将收集的日志发送到对应的output。
# https://www.elastic.co/guide/en/beats/filebeat/index.html
默认向elasticsearch写数据，需要将elasticsearch注释掉。
- type: log
enabled: true #此处必须为true否则不生效
paths:
- /var/log/syslog
output.kafka:
topic: 'syslog001'
hosts: ["10.120.75.102:9092", "10.120.75.103:9092", "10.120.75.107:9092"]
enabled: true
codec.format:
string: '%{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
PUT /%3Csyslog001-%7Bnow%2Fd%7D-000001%3E
{
"aliases": {
"syslog001_write": {}
}
}
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/dmesg
output.elasticsearch:
# Array of hosts to connect to.
hosts: ["http://10.120.75.102:9201","http://10.120.75.103:9201","http://10.120.75.107:9201"]
indices:
- index: "dmesg001_write"
# setup 必有，否则会出错
setup.template.name: "*" # 这里也可以更换为实际莫个模板
setup.template.pattern: "dmesg001-*"
PUT /%3Cdmesg001-%7Bnow%2Fd%7D-000001%3E
{
"aliases": {
"dmesg001_write": {}
}
}
有时候需要使用一个filebeat将不同类别的日志发送到不同的topic中，此时可以在inputs中添加fields: log_topic 字段，字段值设置为不同的topic名称即可，然后在 output中 通过topic: '%{[fields.log_topic]}'设置其对应topic，此时启动filebeat后，日志将发送到不通的topic中，以下分别发送到dmesg001和syslog001两个topic中。
filebeat.inputs:
- type: log
# Change to true to enable this input configuration.
enabled: true
# Paths that should be crawled and fetched. Glob based paths.
paths:
- /var/log/syslog
#- c:\programdata\elasticsearch\logs\*
fields:
log_topic: syslog001
- type: log
enabled: true
paths:
- /var/log/dmesg
fields:
log_topic: dmesg001
output.kafka:
# topic: 'syslog001'
topic: '%{[fields.log_topic]}'
hosts: ["10.120.75.102:9092", "10.120.75.103:9092", "10.120.75.107:9022"]
enabled: true
codec.format:
string: '%{[message]}'
partition.round_robin:
reachable_only: false
required_acks: 1
compression: gzip
max_message_bytes: 1000000
对于多个不同类别的日志，若希望采集到不同的index中，则可以分别为每类日志创建一个- type: log， 并设置对应的fields.type，后面在output中通过fields.type 来判断属于哪个索引。
具体配置如下：
filebeat.inputs:
- type: log
enabled: true
paths:
- /var/log/dmesg
fields:
type: 'dmesg001'
- type: log
enabled: true
paths:
- /var/log/syslog
fields:
type: 'syslog001'
output.elasticsearch:
hosts: ["http://localhost:9200"]
indices:
- index: "syslog001_write"
when.contains:
fields:
type: 'syslog001'
- index: "dmesg001_write"
when.contains:
fields:
type: 'dmesg001'
setup.template.name: "base-settings"
setup.template.pattern: "*"
最终, 写入kibana的字段会多一个fields.type，syslog001-* 的全部为syslog001，dmesg001-* 的全部为dmesg001