filebeat-kafka日志收集

https://www.cnblogs.com/mathli/p/10087825.html

首先在kafka上創建topic，這里是 

servicelog

filebeat.yml配置

filebeat.inputs:
- type: log
  paths:
    - /opt/logs/*/error.log
    - /opt/logs/*/info.log
    - /opt/*/logs/*.log
  fields:
    log_topic: 'servicelog' 
  include_lines: ["^java","^org","^com","ERROR",".*:.*:*:.*"]
  #exclude_lines: ["^*at"]
  multiline.pattern: '^[[:space:]]+(at|\.{3}) \b|.*Exception:.*'
  multiline.negate: false
  multiline.match: after
  tail_files: true
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml

processors:
  - add_host_metadata: ~
output.kafka:
  enabled: true
  hosts: ["192.168.10.11:9092"]
  topic: 'servicelog'
  partition.round_robin:
    reachable_only: true
  worker: 2
  required_acks: 1
  #compression: gzip
  max_message_bytes: 10000000
#logging.level: debug
name: test1

　　

啟動filebeat，然后測試往log文件里寫匹配的字符，並在kafka上一個這個topic的消費者，就能看到filebeat生產者拋過來的信息

./kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic servicelog  --from-beginning 
{"@timestamp":"2020-04-18T06:01:21.419Z","@metadata":{"beat":"filebeat","type":"doc","version":"6.6.2","topic":"servicelog"},"log":{"file":{"path":"/opt/logs/app01/error.log"}},"input":{"type":"log"},"beat":{"name":"test1","hostname":"docker1","version":"6.6.2"},"prospector":{"type":"log"},"fields":{"log_topic":"servicelog"},"host":{"name":"docker1","architecture":"x86_64","os":{"platform":"centos","version":"7 (Core)","family":"redhat","name":"CentOS Linux","codename":"Core"},"id":"ebcb2eeb0fb64fdb9a0be26f2e81cd50","containerized":true},"message":"ERROR","source":"/opt/logs/app01/error.log","offset":0}