背景说明

在此之前日志收集方式采用的logstash安装在各个业务服务器上收集日志直接推到ES的方式（如下图），此模式存在几个弊端：

1、logstash解析消息格式并调用es接口，消耗资源比较多，直接表现为高峰时期CPU飙升至60%，内存占用25%；

2、logstash直接调用ES进行存储，完全依赖ES服务的存储处理速度，如果ES处理太慢或者宕机，可能导致业务服务因资源耗用处理缓慢，甚至出现宕机等雪崩状况；

3、高峰时期日志已经出现部分服务接入后消费缓慢等现象，最长消费延时达到2个小时以上，而且只是接入了部分服务。

根据以上问题主要考虑3点：

1、降低日志收集对业务服务资源的影响；

2、解耦业务服务器和ES直接影响关系；

3、加快日志收集到ES的速度，减少日志延时展示时间。

改进方案如下图所示：

1、应用服务器的日志收集采用资源消耗极低的filebeat，将日志收集后直接推送到kafka集群；

2、kafka进行异步存储，且不受ES存储索引压力影响，push操作响应时间接近于零；

3、kafka日志消息堆积不影响业务服务器，ES端logstash可以根据状况消费kafka消息；

4、应用服务器端logstash日志收集到压力一直维持在相对平稳水平，如果kafka消息堆积量较大，根据需要增加ES和消费节点即可。

日志收集整体服务部署架构如下图所示：

1、新增3台kafka服务器安装kafka套件+logstash消费端；

2、停用业务服务的logstash，安装filebeat配置将日志收集后直接推送至kafka集群；

3、配置logstash消费各个服务的topic，将日志消费至ES集群。

首先配置接入的是channel和sms服务，数据接入展示正常，kibana后台查看到的数据展示也正常，高峰时间日志延迟在20s，都属于可以接受范围。

然后开心的接入了api服务，在下午3点左右逐步进入业务高峰时期，kafka日志开始出现堆积，ES内查看到的消费延时逐步从10分钟逐步延长到5小时，后面kafka堆积lag日志达6亿条，延时大20小时+。

由此开启了本次的花式躺坑历程 ........

logstash配置坑

ES更新版本速度很快，而且各版本间配置都有不一样，相应的套件（如：kibana/logstash/sdk等）也都需要使用对应版本的，我们使用的ES7.0+版本。

出现延迟后我们查看了ES集群的压力状况，除个别节点的CPU到达80%左右，其他节点均低于10%。

1、提升logstash的消费速度，查看网络上各大神攻略 / 最佳实践等（绝大部分是logstash2.0等版本的配置），根据攻略配置后服务完全没法启动或者无效果；

2、增加logstash消费端节点数量，从3个节点增加至6个节点，高峰观察后并无任何改善；

3、重新整理查看官方对应版本资料和网络信息，Get一个关键点 – logstash从6.0开始引入了 Multiple Pipelines 方案降低n多个节点维护痛点；

4、将配置改成 Multiple Pipelines 方式，启动方式也和默认不一样（这个官方没有说明，配置不生效折腾了比较久），再次高峰观察后也没改善。

网络最佳实践坑

根据网络大神们对logstash的最佳生产实践经验，只需要调整这些参数就好，总结logstash优化里几个关键性参数：

1、kafka参数： consumer_threads ----- input里配置消费者线程数，最佳为与kafka partitions数一致

2、kafka参数： max_poll_records ---- input里配置最大获取记录条数，理论上当然是越大越好

3、pipeline参数： pipeline.workers ---- worker数量，最佳与CPU核心数相同

4、pipeline参数： pipeline.batch.size ---- 批次提交处理的数据量

5、pipeline参数：pipeline.batch.delay ---- 批次提交延时等待数据收集齐的时长

最初不理解各参数的实际功效，根据综合大神推荐对应配置如下：

consumer_threads: 16
max_poll_records: 10000
---
pipeline.workers: 8
pipeline.batch.size: 10000
pipeline.batch.delay: 5

发现消费速度有较大提升，但是消费速度仍然跟不上生产速度；

consumer_threads: 16
max_poll_records: 50000
---
pipeline.workers: 8
pipeline.batch.size: 10000
pipeline.batch.delay: 5

随后出现poll() timeout异常，大意为单批次处理时间超时，也就是等待时间内批次处理时间过长；

consumer_threads: 16
max_poll_records: 50000
---
pipeline.workers: 8
pipeline.batch.size: 20000
pipeline.batch.delay: 5

观察ES监控发现index处理延时变长，ES服务CPU飙升至99%，logstash服务CPU飙升至800%，日志偶尔出现poll() timeout异常。

也设置过max_poll_records: 100000，pipeline.batch.size: 50000,但是效果都不理想；

重新整理思路后查阅各类文档，大致确认kafka最佳poll应该在10000～50000，es存储数据的pipeline.batch最佳应该是3000～10000；

根据以上思路，经过多次调试发现当前的服务器集群资源状况下最佳配置如下：

consumer_threads: 16
max_poll_records: 20000
---
pipeline.workers: 8
pipeline.batch.size: 5000
pipeline.batch.delay: 10

但无论设置为3个节点或6个节点，在ES的监控tps上始终没有较大改善，至此可以认为logstash消费没有问题，瓶颈可能在kafka消费时存在poll量限制或ES处理压力问题。

ES配置版本坑

根据上面推测最大可能堵塞点为消费方存储时间消耗太长，ES的主节点最高TPS只有18000/s，开始尝试优化ES效果的配置。

这块有两个考虑优化点：

1、基础配置优化

2、索引分片数优化

从网络大神提供的最佳实践里找到一些优化配置点，进行了如下配置：

index.store.type: niofs 读写文件方式
index.cache.field.type: soft 缓存类型
 
index.merge.scheduler.max_thread_count: 8 索引merge最大线程数
index.translog.flush_threshold_size:600MB 刷新translog文件阀值
 
 
bootstrap.mlockall: true 禁用swap
 
threadpool.index.type: fixed 写索引线程池类型
threadpool.index.size: 64 线程池大小（建议2~3倍cpu数）
threadpool.index.queue_size: 1000 队列大小
 
 
threadpool.search.size: 64 搜索线程池大小
threadpool.search.type: fixed 搜索线程池类型
threadpool.search.queue_size: 1000 队列大小
 
 
threadpool.get.type: fixed 取数据线程池类型
threadpool.get.size: 32 取数据线程池大小
threadpool.get.queue_size: 1000 队列大小
 
 
threadpool.bulk.type: fixed 批量请求线程池类型
threadpool.bulk.size: 32 批量请求线程池大小
threadpool.bulk.queue_size: 1000 队列大小
 
 
threadpool.flush.type: fixed 刷磁盘线程池类型
threadpool.flush.size: 32 刷磁盘线程池大小
threadpool.flush.queue_size: 1000 队列大小
 
 
indices.store.throttle.type: merge
indices.store.throttle.type: none 写磁盘类型
indices.store.throttle.max_bytes_per_sec:500mb 写磁盘最大带宽
 
cluster.routing.allocation.node_initial_primaries_recoveries:8 并发恢复分片数
cluster.routing.allocation.node_concurrent_recoveries:2 同时recovery并发数

配置完成后开心的尝试启动，服务启动失败，为爆出哪个配置问题，然后开始配置排除、报错、排除、报错、、、

发现很多配置是不能用的，index相关的配置不能在elasticsearch.yml配置，threadpool全部报错。。。

根据提示和重新查找发现，7.0版本里对线程池配置进行了全面调整，threadpool得改成 thread_pool，threadpool.index、threadpool.bulk等配置被弃用等等。

最终只优化调整了如下配置：

threadpool.write.size: 9 //因CPU核心数等问题 提示只支持<=9
threadpool.write.queue_size: 1000
 
 
cluster.routing.allocation.node_initial_primaries_recoveries:8 //并发恢复分片数
cluster.routing.allocation.node_concurrent_recoveries:2 //同时recovery并发数

全部调整完成后，根据监控反馈看并没实际变化，只能进行下一步，索引分片优化。

考虑这一点的有两个原因：

1、当前日志存储数据量分布不均衡

2、数据存储节点的压力不均衡

其实在上面的基础配置时也是考虑在配置文件里可以有相关配置可以调整（因为没弄过），但是实际发现并没有找到配置点。

尝试根据提示先关闭目标index，在执行优化分片命令，直接提示报错了，final不允许修改。

PUT xxxxx-2019.12.16/_settings
{
  "index.refresh_interval" : "30s",
  "index.number_of_shards" : "18",
  "index.number_of_replicas" : "0"
}

只能用设置索引模板的方式，创建好模板，删除索引，再次创建出来的索引 — 成功 shards 18，replicas 0，interval 30。

PUT _template/xxxx
{
    "order" : 0,
    "index_patterns" : [
      "xxxx-*"
    ],
    "settings" : {
      "index" : {
        "refresh_interval" : "30s",
        "unassigned" : {
          "node_left" : {
            "delayed_timeout" : "5m"
          }
        },
        "number_of_shards" : "18",
        "translog" : {
          "flush_threshold_size" : "600mb",
          "durability" : "async"
        },
        "number_of_replicas" : "0"
      }
    }
  }

查看ES监控，发现ES的主节点索引的TPS上升到了35000+/s，kafka消费lag速度明显加快了，再查看ES各节点CPU负载基本都保持着40～60%。

至此，表示这次的优化有明显的提升，接下来尝试配置前几天积压的数据并行进行消费，压测服务的稳定性和承压状况。

压测1.5小时后，结果如下：

1、ES主节点索引TPS在41000～46000/s，稳定在42000/s；

2、logstash消费时的cpu压力在300%左右；

3、ES的各个节点的CPU压力在35～75%之间；

4、kafka消费速度值40000～80000条/s

如此表现基本可以确定logstash消费已经稳定输出切达到相对最佳状况，再结合线上真实消费观察，线上服务最高压力消费ES主节点索引TPS在33000/s，kafka数据完全没有lag积压。

Filter速度坑

网络上很多实践里说logstash的filter是最耗性能的，有大神测试过不加filter json source => "message" 的速度达到 15000/s，加了filter速度只有7000～10000/s，但实际尝试时发现绝大部分情况下的使用对logstash的要求是没有这么极端的，而且我们的业务还是需要source里的字段的。

其他扩展

1、查看各类资料时候比较了使用logstash直传和filebeat方式的数据发现，每条数据增加了很多filebeat特性的字段（近20个），而且每天的数据存储量上浮了30%左右，去除这些字段肯定是能够提升速度和降低存储成本的，于是mutate{ remove_field => ["host"] ...} 加上了，实验证明效果非常明显，1亿条数据只有86G左右（原来的120G+）；

2、consumer_threads设置为kafka partitions数一致可能是最佳实践，但并非必须且相对也耗费一定的资源，可以适当调小；

3、在Multiple Pipelines模式下可以根据需要设置对应的wokers数量，不一定要和CPU核心数一样；

4、logstash建议使用Multiple Pipelines模式进行消费配置，这样可以方便管理及资源分配，同时各Pipeline之间不会因为单业务堵塞而受影响。

根据真实线上服务的配置最终将logstash调整成了如下配置:

xxx.conf

input {
  kafka{
    bootstrap_servers => "10.10.5.xx:9092,10.10.5.xx:19092,10.10.13.xx:9092,10.10.13.xx:19092,10.10.12.98:9092,10.10.12.xx:19092"
    topics => "xxx-filebeat-topic"
    group_id => "consumer-xxx-filebeat"
    client_id => "kafka_client_xxxxxx"
    consumer_threads => 4
    decorate_events => false
    codec => "json"
    auto_offset_reset => "earliest"
    max_poll_records => "20000"
    auto_commit_interval_ms => "1000"
    type => "api"
  }
}
 
filter {
  json {
    source => "message"
  }
 
  mutate{
    remove_field => ["host"]
    remove_field => ["agent"]
    remove_field => ["ecs"]
    remove_field => ["tags"]
    remove_field => ["cloud"]
    remove_field => ["input"]
    remove_field => ["fields"]
  }
}
 
output {
  stdout { codec => json }
  elasticsearch {
    hosts => ["http://xxxxx.xx.amazonaws.com:80"]
    index => "%{projectName}-%{+yyyy.MM.dd}"
  }
}

pipelines.yml

- pipeline.id: "pipe-xxx"
  path.config: "/etc/logstash/conf.d/xxx.conf"
  pipeline.workers: 2
  pipeline.batch.size: 5000
  pipeline.batch.delay: 10
- pipeline.id: "pipe-xxxxxx"
  path.config: "/etc/logstash/conf.d/xxxxxx.conf"
  pipeline.workers: 1
  pipeline.batch.size: 5000
  pipeline.batch.delay: 10