主配置文件，采集到的信息放到kafka

cat > filebeat.yml <<EOF
#=========================== Filebeat inputs =============================
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: 10s

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

setup.kibana:
  host: "172.17.0.22:5601"

output.kafka:
  enabled: true
  # hosts: ["172.17.0.20:9092","172.17.0.21:9092","172.17.0.22:9092"] # 内网kafka
  hosts: ["192.168.0.20:9012","192.168.0.21:9012","192.168.0.22:9012"] # 外网kafka
  username: "producer"
  password: "abcd1234"
  compression: gzip      # 压缩
  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的最大事件数
  reachable_only: true  # 以轮询的方式生产数据到kafka
  logging.to_files: true   # 输出所有日志到file，默认true， 达到日志文件大小限制时，日志文件会自动限制替换
  timeout: 5
  topic: "elk-%{[fields.project_name]}" # 存档到kafka的topic

#=============================== 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
EOF

子配置文件

cat > /etc/filebeat/inputs.d/nginx.yml <<EOF
- type: log
  enable: true # 开启type：log
  encoding: utf-8  #文件编码,注意win系统项目文件编码必须填写utf-16-bom
  paths:
    - /data/software/nginx/logs/*.log   # nginx日志目录
  json.keys_under_root: true # json格式开启
  overwrite_keys: true # 若启用此设置，则解码的 JSON 对象中的值将覆盖 Filebeat 通常添加的字段(类型，源，偏移等)以防发生冲突
  max_bytes: 20480 # 最大传输字节
  fields:
    project_name: nginx  # 项目标签
    ip_addr: "172.17.0.22" # 做服务器的标识用的

  scan_frequency: 5s  #文件扫描间隔
  tail_files: true  #直接从末尾开始收集(如果是不想收集老旧文件,设置为true)
EOF

一、对采集的信息不做处理，就保持主配置文件和子配置文件不做修改 

二、对采集的信息做处理，对一些filebeat所携带的元数据进行删除

在主配置文件做修改，添加一些信息，使用Processors(处理器)过滤和增强(加)数据

cat > filebeat.yml <<EOF
#=========================== Filebeat inputs =============================
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: 10s

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

setup.kibana:
  host: "172.17.0.22:5601"

output.kafka:
  enabled: true
  # hosts: ["172.17.0.20:9092","172.17.0.21:9092","172.17.0.22:9092"] # 内网kafka
  hosts: ["192.168.0.20:9012","192.168.0.21:9012","192.168.0.22:9012"] # 外网kafka
  username: "producer"
  password: "abcd1234"
  compression: gzip      # 压缩
  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的最大事件数
  reachable_only: true  # 以轮询的方式生产数据到kafka
  logging.to_files: true   # 输出所有日志到file，默认true， 达到日志文件大小限制时，日志文件会自动限制替换
  timeout: 5
  topic: "elk-%{[fields.project_name]}" # 存档到kafka的topic

#=============================== other ===============================

processors:
- drop_fields:  # 删除不太需要的字段，减少浪费存储空间
    fields: ["log.offset", "input", "agent.type", "agent.ephemeral_id", "agent.id", "agent.version", "agent.name", "ecs", "host"]

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
EOF

三、只要日志数据，对filebeat所携带的元数据一律不要

在主配置文件做修改，添加一些信息，增加 自定义输出内容(codec.format) 参数配置

cat > filebeat.yml <<EOF
#=========================== Filebeat inputs =============================
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: 10s

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

setup.kibana:
  host: "172.17.0.22:5601"

output.kafka:
  enabled: true
  # hosts: ["172.17.0.20:9092","172.17.0.21:9092","172.17.0.22:9092"] # 内网kafka
  hosts: ["192.168.0.20:9012","192.168.0.21:9012","192.168.0.22:9012"] # 外网kafka
  username: "producer"
  password: "abcd1234"
  compression: gzip      # 压缩
  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的最大事件数
  reachable_only: true  # 以轮询的方式生产数据到kafka
  logging.to_files: true   # 输出所有日志到file，默认true， 达到日志文件大小限制时，日志文件会自动限制替换
  timeout: 5
  topic: "elk-%{[fields.project_name]}" # 存档到kafka的topic
  codec.format:
    string: '%{[message]}'

#=============================== 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
EOF

子配置文件 ，子配置文件不能开启json格式

cat > /etc/filebeat/inputs.d/nginx.yml <<EOF
- type: log
  enable: true # 开启type：log
  encoding: utf-8  #文件编码,注意win系统项目文件编码必须填写utf-16-bom
  paths:
    - /data/software/nginx/logs/*.log   # nginx日志目录
  # json.keys_under_root: true # json格式开启
  # overwrite_keys: true # 若启用此设置，则解码的 JSON 对象中的值将覆盖 Filebeat 通常添加的字段(类型，源，偏移等)以防发生冲突
  max_bytes: 20480 # 最大传输字节
  fields:
    project_name: nginx  # 项目标签
    ip_addr: "172.17.0.22" # 做服务器的标识用的

  scan_frequency: 5s  #文件扫描间隔
  tail_files: true  #直接从末尾开始收集(如果是不想收集老旧文件,设置为true)
EOF