filebeat以及logstash配置文件说明解释

filebeat配置文件讲解

cat filebeat.yml
filebeat.inputs:
- input_type: log
  paths:
    # 收集日志的路径
    - /var/log/pods/*/*/*.log
  fields:
    # 设置字段
    log_topic: ${TOPIC_ID}
  tail_files: true
  # 删除文件状态，该clean_inactive设置必须大于以ignore_older + scan_frequency（简隔多久扫描一次日志文件，默认十秒）
  clean_inactive: 48h   
  # 超过该时间之前更新的文件filebeta不采集，默认关闭（ignore_older为大于close_inactive）
  ignore_older: 24h
  # harvester读取文件最后一行后，1分钟后日志文件没有变动。关闭文件句柄  
  close_inactive: 1m    

output.kafka:
  # 获取集群元数据的 Kafka 代理地址列表
  hosts: ["10.0.7.53:9092", "10.0.7.54:9092", "10.0.7.55:9092"]
  # 此配置使用自定义字段 fields.log_topic来为每个事件设置主题
  topic: '%{[fields.log_topic]}'
  partition.round_robin:
    # reachable_only设置为true，事件将仅发布到可用分区
    reachable_only: true
  #  ACK 可靠性级别。0=无响应，1=等待本地提交，-1=等待所有副本提交。默认值为 1
  required_acks: 1
  # 设置输出压缩编解码器。必须是none、snappy和lz4之一gzip。默认值为gzip。
  compression: gzip
  # JSON 编码消息的最大允许大小。更大的消息将被丢弃。默认值为 1000000（字节
  max_message_bytes: 1000000
# filebeat只收级错误日志  
logging.level: error

logstash配置文件讲解

input {
  kafka {
    bootstrap_servers => "10.0.7.53:9092,10.0.7.54:9092,10.0.7.55:9092"
    topics => ["test-clavaplus-consume"] 
    # 将收集过的json日志解析成json格式
    codec => json {
        charset => "UTF-8"   # 允许中文搜索
    }
  }
}

filter {
  json {
     # 将message字段的内容赋值给source字段
    source => "message"
  }
  mutate {    
    # 删除多余没用的字段
    remove_field => ["message","ecs","agent","timestamp","type","@version","log"]
  }
}


output {
    stdout {
        #控制台调式专用格式
        codec => rubydebug
    }
  if [fields][log_topic] == "test-clavaplus-consume"  {
    elasticsearch {
      hosts => ["10.0.7.46:9200","10.0.7.47:9200","10.0.7.48:9200"]
      index => "test-clavaplus-consume-%{+YYYY.MM.dd}"
    }}
}