elk笔记10--filebeat使用

Posted 2022-08-29 昕光xg

elk笔记10--filebeat使用

• ​​1 filebeat 介绍​​
• ​​2 filebeat 使用案例​​

• ​​2.1 软件安装​​
• ​​2.2 采集数据到 kafka​​
• ​​2.3 采集数据到 es​​

• ​​3 使用技巧​​

• ​​3.1 filebeat 将日志按照类别发送到不同 kafka topic​​
• ​​3.2 filebeat将日志按照类别发送到不同es index​​

• ​​4 说明​​

1 filebeat 介绍

Filebeat 是一个用于转发和集中化日志数据的轻量级工具，它一般作为agent安装在服务器上。它监控日志文件和文件夹，收集日志事件，并转发到logstash、ES或者Kafka。

Filebeat的流程图如下，当filebeat运行后，其根据指定的log日志位置，启动一个或者多个input。 对于每一个定位跟踪的日志，filebeat都为启动一个独特的harvester(收割机), harvester会读取一条新日志内容，并将其发送到libbeat， libbeat收集一些列的事件，并将收集的日志发送到对应的output。

2 filebeat 使用案例

2.1 软件安装

1. 下载filebeat bin文件，解压即可用
   curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.8.0-linux-x86_64.tar.gz
   tar xzvf filebeat-7.8.0-linux-x86_64.tar.gz
2. 通过deb包安装
   curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.8.0-amd64.deb
   sudo dpkg -i filebeat-7.8.0-amd64.deb

2.2 采集数据到 kafka

1. 配置yml

# 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

1. 创建对应topic：
   bin/kafka-topics.sh --create --zookeeper 10.120.75.102:2181 --replication-factor 1 --partitions 1 --topic syslog001
   Created topic syslog001.
2. 创建对应rollover索引

PUT /%3Csyslog001-%7Bnow%2Fd%7D-000001%3E

  "aliases": 
  "syslog001_write": 
  

1. 启动filebeat：
   ./filebeat [-e] -c filebeat.yml &
   -e 可以省略
2. kafka查看是否收到数据：
   $ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 [–from-beginning] --topic syslog001
   Jul 5 20:45:01 xg kernel: [5108113.014561] sadc[7456]: segfault at 1 ip 00007f7837f2051f sp 00007ffd072c92e8 error 4 in libc-2.24.so[7f7837e96000+195000]
   [–from-beginning]无该选项则消费最新的数据，有该选项则从头开始消费。
3. 查看数据是否到es
   若配置了kafka采集数据的话，可以直接在es中查看数据是否到es，具体配置方法见：​​​elk笔记5–logstash使用​​ GET syslog001_write/_count 或者_serach 查看具体数据

2.3 采集数据到 es

1. 配置yml

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-*"

1. 创建对应rollover索引

PUT /%3Cdmesg001-%7Bnow%2Fd%7D-000001%3E

  "aliases": 
  "dmesg001_write": 
  

1. 启动filebeat：
   ./filebeat [-e] -c filebeat.yml &
2. 查看数据是否写到es
   GET dmesg001_write/_count 或者_serach 查看具体数据
   由于数据直接写到es，因此不需要第三方采集，可以直接在es中查看数据

3 使用技巧

3.1 filebeat 将日志按照类别发送到不同 kafka topic

有时候需要使用一个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

3.2 filebeat将日志按照类别发送到不同es index

对于多个不同类别的日志，若希望采集到不同的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

4 说明

1. 测试软件: elasticsearch版本为7.2.1， filebeat为7.8.0
2. 参考文献:​​elastic 官网–filebeat/current/index​​