一、下载及安装

sudo rm -rf /etc/filebeat      # 配置文件目录
sudo rm -rf /var/lib/filebeat  # 运行数据目录
sudo rm -rf /var/log/filebeat  # 日志目录
sudo rm -rf /usr/share/filebeat # 共享文件

wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.13.4-x86_64.rpm

rpm -iv filebeat-7.13.4-x86_64.rpm

二、配置文件

filebeat.inputs:
- type: filestream
  paths:
    #- /var/log/nginx/*.log
    - /var/log/nginx/gw-hu-tls-new.log
  multiline:
    pattern: '^(?:\d{4}-\d{2}-\d{2}|\d{4}/\d{2}/\d{2}|\{)'
    negate: true
    match: after
  max_bytes: 5000
  ignore_older: 0
  tail_files: true
  fields:
    app: gw-hu-tls-new-nginx
processors:
  - decode_json_fields:
      fields: ["message"]
      target: ""
      overwrite_keys: true
      max_depth: 2
  - drop_fields:
      fields: ["[http_user_agent]","ecs","agent","kubernetes.labels","offset","log.offset","type","input.type","prospector.type","source","stream","log.file.path","log.flags"]
output.kafka:
  hosts: ["10.20.15.234:9092"]
  topic: "gwm-ecs-services"
  compression: snappy
  compression_level: 3
  required_acks: 1
  broker_timeout: 10s
  channel_buffer_size: 1024
  keep_alive: 120
  max_message_bytes: 10485760

参数说明：

      - decode_json_fields:
          fields: ["message"]
          target: ""
          overwrite_keys: true #对于同名的key，会覆盖原有key值
          max_depth: 1   #最大解析深度 默认为1

三、配置文件验证

filebeat test config -c filebeat.yml

四、对应 logstash 配置

input{
         kafka {
           consumer_threads => 1
           bootstrap_servers => "10.19.5.179:9092"
           topics => ["test-ecs-services"]
           group_id => "test-ecs-services"
           auto_offset_reset => "latest"
           type => "test-ecs-services"
         }
}


filter {

     json{
            source => "message"
        }
  
     mutate {
        remove_field => "agent"
        remove_field => "[old][event]"
        remove_field => "[ecs][version]"
        remove_field => "[input][type]"
        remove_field => "[fields][type]"
        remove_field => "[log][log.offset]"
        remove_field => "[log][file][path]"
        remove_field => "[log][offset]"
        remove_field => "tags"
        remove_field => "@version"
     }

      ruby {
          code => "event.set('index_date', event.get('@timestamp').time.localtime + 8*60*60)"
      }
      mutate {
          convert => ["index_date", "string"]
          gsub => ["index_date", "T([\S\s]*?)Z", ""]
          gsub => ["index_date", "-", "."]
      }

}


output {

    if [type] == "test-ecs-services" {
    elasticsearch {
    hosts => ["10.10.5.60:9200"]
    index => "%{[fields][app]}-%{index_date}"
    user => "elastic"
    password => "111111111"
    }
    }

}

五、同一宿主机多路径采集并增加不同的 fields

yaml
   filebeat.inputs:
   - type: log
     paths:
       - /path/to/logfile1.log
     fields:
       app: app1
   - type: log
     paths:
       - /path/to/logfile2.log
     fields:
       app: app2
   - type: log
     paths:
       - /path/to/logfile3.log
     fields:
       app: app3

1