安装 filebeat 7.17.10 数据采集至kafka

filebeat  

安装 7.17.10

创建yum源文件

[elastic-7.x]
name=Elastic repository for 7.x packages  
baseurl=https://artifacts.elastic.co/packages/7.x/yum  
gpgcheck=0  
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch  
enabled=1  
autorefresh=1  
type=rpm-md  

yum安装

yum install filebeat  

启动并设置开机启动

service filebeat start

systemctl enable filebeat  

filebeat主配置文件

vim /etc/filebeat/filebeat.yml

filebeat.inputs:  
max_procs: 1                     #限制filebeat的进程数量,其实就是内核数,避免过多抢占业务资源  
# logging.level: debug           # 开启debug模式,用于排查错误
queue.mem.events: 256            # 存储于内存队列的事件数,排队发送 (默认4096)  
queue.mem.flush.min_events: 128  # 小于 queue.mem.events ,增加此值可提高吞吐量 (默认值2048)  
filebeat.config.inputs:  
  enabled: true #开启子配置文件查看,在这个目录里新增子配置文件会自动加载,不需要重启filebeat
  path: ${path.config}/inputs.d/*.yml
  reload.enabled: true
  reload.period: 10s
filebeat.config.modules:  
  path: ${path.config}/modules.d/*.yml # filebeat内置模块
  reload.enabled: false
  reload.period: 10000s
setup.template.settings:  
  index.number_of_shards: 3
setup.template.name: "logcall"  
setup.template.pattern: "*" #这里直接匹配全部应用  
setup.template.overwrite: true  
setup.template.enabled: true  
#自定义es的索引需要把ilm设置为false
setup.ilm.enabled: false  
output.kafka:  
  enabled: true
  hosts: ["192.168.240.121:9092"]
  compression: gzip      # 压缩
  keep_alive: 10s
  max_message_bytes: 1000000  # Event最大字节数。默认1000000。应小于等于kafka broker message.max.bytes值
  required_acks: 1  # kafka ack等级
  worker: 1  # kafka output的最大并发数
  bulk_max_size: 2048    # 单次发往kafka的最大事件数
  logging.to_files: true   # 输出所有日志到file,默认true, 达到日志文件大小限制时,日志文件会自动限制替换
  timeout: 5
  topic: "%{[fields.project_name]}" # 存档到kafka的topic
  codec.format:
    string: '%{[message]}'  # 只要日志数据,对filebeat所携带的元数据一律不要
#=============================== other ===============================
close_older: 30m         # 如果文件在某个时间段内没有发生过更新,则关闭监控的文件handle。默认1h  
force_close_files: false # 这个选项关闭一个文件,当文件名称的变化。只在window建议为true  
#没有新日志采集后多长时间关闭文件句柄,默认5分钟,设置成1分钟,加快文件句柄关闭
close_inactive: 1m  
#传输了3h后荏没有传输完成的话就强行关闭文件句柄,这个配置项是解决以上案例问题的key point
close_timeout: 3h  
#这个配置项也应该配置上,默认值是0表示不清理,不清理的意思是采集过的文件描述在registry文件里永不清理,在运行一段时间后,registry会变大,可能会带来问题
clean_inactive: 72h  
#设置了clean_inactive后就需要设置ignore_older,且要保证ignore_older < clean_inactive
ignore_older: 70h

重启即可生效

service filebeat restart  

子配置文件,配置子配置文件不需要重启filebeat

mkdir /etc/filebeat/inputs.d vim /etc/filebeat/inputs.d/xm2-game-log

- type: log
  enabled: true # 开启type:log
  encoding: utf-8  #文件编码,注意win系统项目文件编码必须填写utf-16-bom
  paths:
    - /data/gamelog/*.log
  overwrite_keys: true  #
  max_bytes: 20480 # 最大传输字节
  fields:
    project_name: xm2-game-log  # 项目标签
  scan_frequency: 5s  #文件扫描间隔
  tail_files: true  #直接从末尾开始收集(如果是不想收集老旧文件,设置为true)

创建测试数据

vim /data/gamelog/test.log

{
  "logtime": "2023-01-01 10:00:00",
  "ip": "127.0.0.1",
  "game_id": "xm2",
  "package_id": "",
  "platform": "117",
  "group": "11700003",
  "account": "local_ben",
  "role_id": "96772055629825",
  "role_name": "S56.Demi-Leigh",
  "log_type": "RoleRegisterLog",
  "role_career": ""
}

去kafka中创建一个消费队列查看是否收到消息