最近一直在重构新公司的日志传输系统，因为业务量较大，目前的单日日志条数已经超过了30亿条，而且业务有明显的高峰期。所以日志收集传输系统，必须要满足明显的波峰性能要求。方案还是以Filebeat + Logstash为主，Logstash直接入kafka， Filebeat从磁盘读取文本文件(json格式)。

优化完成后，单filebeat + 单logstash可以处理 30000条/秒的日志. 单filebeat + 多logstash可以处理 40000条/秒的日志.

日志大小为1.2kbyte.

环境信息：

2台服务器， 32core， 200GB内存， SSD硬盘， 千兆网络内网互联

Filebeat 6.5.2

/usr/bin/filebeat -c /data/filebeat.yml

Logstash 6.5.2

/usr/share/logstash/bin/logstash -f /etc/logstash/logstash-file.conf --path.settings /etc/logstash --path.data /data/tmp/logstash/data --path.logs /data/tmp/logstash/logs -w 8

优化后的配置文件信息如下:

filebeat.yml

关键参数：

scan_frequency

harvester_buffer_size

queue.mem.*

filebeat.inputs:
- type: log

  # Change to true to enable this input configuration.
  enabled: true
  encoding: utf-8

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /data/tmp/data/testlogstash3.log
    #- c:\programdata\elasticsearch\logs\*

# Filebeat以多快的频率去prospector指定的目录下面检测文件更新比如是否有新增文件如果设置为0s则Filebeat会尽可能快地感知更新占用的CPU会变高。默认是10s。
  scan_frequency: 1s
  # 如果设置为true, Filebeat从文件尾开始监控文件新增内容把新增的每一行文件作为一个事件依次发送而不是从文件开始处重新发送所有内容。
  tail_files: false
  harvester_buffer_size: 104857600

# backoff选项指定Filebeat如何积极地抓取新文件进行更新。默认1s. backoff选项定义Filebeat在达到EOF之后再次检查文件之间等待的时间
  backoff: 1s
# 在达到EOF之后再次检查文件之前Filebeat等待的最长时间
  max_backoff: 10s

  close_inactive: 6h
  clean_inactive: 72h
  ignore_older: 70h
  close_timeout: 6h
  fields:
    log_name: newdata

- type: log

  # Change to true to enable this input configuration.
  enabled: true
  encoding: utf-8

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
    - /data/topicdata/*.hequandata

# Filebeat以多快的频率去prospector指定的目录下面检测文件更新比如是否有新增文件如果设置为0s则Filebeat会尽可能快地感知更新占用的CPU会变高。默认是10s。
  scan_frequency: 1s
  # 如果设置为true, Filebeat从文件尾开始监控文件新增内容把新增的每一行文件作为一个事件依次发送而不是从文件开始处重新发送所有内容。
  tail_files: false

# backoff选项指定Filebeat如何积极地抓取新文件进行更新。默认1s. backoff选项定义Filebeat在达到EOF之后再次检查文件之间等待的时间
  backoff: 1s
# 在达到EOF之后再次检查文件之前Filebeat等待的最长时间
  max_backoff: 10s

  close_inactive: 6h
  clean_inactive: 72h
  ignore_older: 70h
  close_timeout: 6h
  fields:
    log_name: olddata



#============================= Filebeat modules ===============================

filebeat.config.modules:
  # Glob pattern for configuration loading
  path: ${path.config}/modules.d/*.yml

  # Set to true to enable config reloading
  reload.enabled: false

#==================== Elasticsearch template setting ==========================

setup.template.settings:
  index.number_of_shards: 3
  #index.codec: best_compression
  #_source.enabled: false

#================================ General =====================================
filebeat.registry_file: /data/tmp/registry-logstash3.index

max_procs: 8

path.home: /etc/filebeat
path.config: /etc/filebeat
path.data: /data/filebeat/data
path.logs: /data/filebeat/logs

queue.mem.events: 409600
queue.mem.flush.min_events: 1024
queue.mem.flush.timeout: 2s


#================================ Outputs =====================================
output.logstash:
  enable: true
  hosts: ['10.28.3.8:5046','10.28.3.8:5047','10.28.3.8:5048','10.28.3.8:5049']
  loadbalance: true
  worker: 2
  bulk_max_size: 50000
  compression_level: 0


#================================ Logging =====================================

# Minimum log level. One of debug, info, warning, or error. The default log level is info
logging.level: info
logging.to_files: true
logging.files:
  path: /data/tmp
  name: filebeat-logstash3.log
  keepfiles: 7
  rotateeverybytes: 10485760
  interval: 24h
  permissions: 0644

logstash.conf配置文件

input {
    beats {
        port => 5046
    }
}

filter {
    grok {
        match => {
            "message" => "(?<kafkaname>\S+?)##########(?<kafkadata>.*)"
        }
        overwrite => ["message"]
    }
}

output {
    if[fields][log_name] == "newdata" or [fields][log_name] == "olddata" {
        file {
            path => "/data/tmp/logstash/logstash-kafkadata5046.log"
                codec => line {
                    format => "%{+YYYY-MM-dd HH:mm:ss}##########%{kafkadata}"
                }
        }
    }
}

logstash.yml配置文件

pipeline.workers: 15
pipeline.output.workers: 15
pipeline.batch.size: 5000
pipeline.batch.delay: 10

优化后的结果(单Filebeat + 单Logstash）：

单条文本大小为 1.2kbyte

优化后的结果(单Filebeat + 4个Logstash)：

单条文本大小为 1.2kbyte大小.