标签 elk 下的文章

es报错 es_rejected_execution_exception[status: 429]

描述

使用 go-mysql-elasticsearch 同步 mysql 数据到 elasticsearch 时, 由于量大出现一个错误, 如下:

time="2017-11-21T18:20:35+08:00" level=error msg="update index: prod_db_room, type: room, id: 556314, status: 429, error: {\"type\":\"es_rejected_execution_exception\",\"reason\":\"rejected execution of org.elasticsearch.transport.TransportService$4@16211f2a on EsThreadPoolExecutor[bulk, queue capacity = 50, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@6f50e0ac[Running, pool size = 8, active threads = 8, queued tasks = 50, completed tasks = 302613024]]\"}"

time="2017-11-21T18:20:35+08:00" level=error msg="update index: prod_db_vdoid, type: vdoid, id: 6926660, status: 429, error: {\"type\":\"es_rejected_execution_exception\",\"reason\":\"rejected execution of org.elasticsearch.transport.TransportService$4@50d7ff2 on EsThreadPoolExecutor[bulk, queue capacity = 50, org.elasticsearch.common.util.concurrent.EsThreadPoolExecutor@6f50e0ac[Running, pool size = 8, active threads = 8, queued tasks = 50, completed tasks = 302613024]]\"}"

分析

从报错信息初步判断是并发量大, 可用的8个线程和50个长度的队列不够用了, 处理不过来

解决

参考官方文档, 注意我这是 2.4 版本的文档, 各位看官可以根据自己的 es 版本在右侧选择对应版本的文档, 在目录中依次点击Modules - Thread pool 看你使用的版本说明, 其实这块基本通用, 不同版本变化不大

在 elasticsearch.yml 文件末尾添加如下配置后重启es即可:

threadpool.bulk.type: fixed
threadpool.bulk.size: 8
threadpool.bulk.queue_size: 1000

其中:

  • type 是要配置的线程池类型
    • bulk 批量操作, 也就是我们上面报错里提示的
    • index 用于索引/删除操作
    • search 计数/搜索/建议操作
    • get 获取操作
    • snapshot 用于快照/恢复操作
    • refresh 用于刷新操作
  • size 线程数量, 一般设置为 cpu 核数
  • queue_size 等待线程处理的队列容量

其它

增加线程和队列容量是一种解决办法, 另外节点数量和分片的分布也是影响原因
以批量操作为例, 队列默认配置是50个容量, 如果3个节点都是8核, 那批量操作的并发最大是 50 * 3 * 8 = 1200, 如果分片都在同一台机器上, 那可能只有 400
所以, 增加节点数和均匀分布分片也很重要

基于ELK的日志统计系统实践

总结下近期的日志系统

目前的架构如图:
请输入图片描述

部分说明

Logstash

主要用来做数据收集和传输, 至于为什么选择它, 很尴尬的说因为早期版本只有 elk, 所以也没怎么特别针对它优化。目前我们使用了三台 Logstash 节点, 一台用于收集数据(主要是UDP)传输到 Kafka 中, 这里要提醒一下, Logstash 的性能并不是很理想, 后面也会考虑其他替代方式, 如 hangout Flume 等。

Kafka

用 Kafka 做缓存, 主要是基于两个考虑。一是穷, 没那么大内存, Kafka 是存储硬盘的, 效率也挺好。二是 Kafka 支持多重消费。其它的自动支持分布式, 对日志友好, 部署简单等等。

Elasticsearch

这是重头戏, 在这过程中也摔过很多跟头, 过程就不多说了, 直接说我们现在做的一些策略。

  • 分集群。针对不同业务, 不同场景, 分多个集群。比如用于离线数据分析的集群, 心跳数据集群以及给业务端用于实时搜索的集群, 保证一个集群异常后互不影响, 也方便针对性做集群扩展。
  • 冷热分离。这个我有专门写了篇博文来介绍( 传送门 ), 这里就不再赘述。
  • 关闭索引。是的,你没有看错。尤其是日志数据, 一般短期内的数据还会被用于数据统计分析, 历史数据再被翻阅的可能性比较小, 由于倒排词典的索引常驻内存,无法 GC, 尤其随着日积月累索引越来越多, 内存捉襟见肘(主要是穷), 动辄 gc 几十秒, 告警个不停, 或者直接 gg 了, 索引我们做了定时关闭历史索引, 只保留近期一段时间的索引开启着, 如果需要查询历史数据再单独开启。
  • 其它基于配置的一些小优化, 如 slowlog, fielddata.cache 啊, 网上都有。

Kibana, Grafana

这俩都是纯做数据展示的, 初期仅有 Kibana, 针对不同集群分别配置多个 Kibana, 主要用于日志查询, 开发调试跟踪, 这里推荐一个 Kibana 的插件 Sense, 用于调试 Elasticsearch 特别好使, 支持代码提示补全, 个人觉得比 Head 里的查询好使。又多了一个 Grafana 主要也是两方面考虑, 一是 Kibana 不支持权限分配管理, 尤其是 Dashboard 做数据维度展示的时候, 任何一个浏览者都可以随意改动, 用我们运营童鞋的话来说"不敢乱点", 第二个考虑就是因为老板不喜欢, "用不惯", 囧~~~。 Grafana 支持权限分配管理, 数据源支持也很丰富, 不同 Elasticsearch 集群的数据在同一个页面展示, 尤其后期我们可能也会考虑 influxdb, 可以更好的结合 Grafana。

Elasticsearch Monitor

主要用于监控 Elasticsearch 节点健康, 很简单的逻辑, 直接去访问所有节点ip:9200, 1s 不返回结果就立即告警。

告警

先说下为什么不选用 Elastalert, 一是我们想和 Elasticsearch 接偶, 不想在它身上绑太多挂件。目前比较简单, 是直接访问 Kafka, 根据日志里定义的级别, 自己写告警规则和通知, 更灵活一点, 后面可能会考虑平台化。

Spark Streaming

实时分析, 基于 Elasticsearch 的分析已经不能满足某些业务的需求, 这一块还在试水, 后面再来分析。

Mysql to Elasticsearch

主要是基于 Elasticsearch 的全文搜索和分词功能, 同步 MySQL 的数据到 Elasticsearch 中, 给业务端提供搜索服务。由于我们数据库是在云上, 服务商不给 binlog, 早期选用的是 Elasticsearch-jdbc (注意对应elasticsearch版本), 后来同步的表增加了之后, 这玩意儿莫名就会把 Elasticsaerch 整挂了(原因没查到), 现在使用的是 Logstash 的同步插件。这里主要的问题是更新数据和删除数据, 查询出来的语句设置好唯一主键对应 Elasticsarch 里的 _id, 后面根据数据表里的时间字段增量塞到 Elasticsearch中时, _id字段相同的数据会覆盖, 达到更新的目的。 删除主要是靠软删除, 目前还没啥好办法。 最后值得一提的是, MySQL 同步 Elasticsearch 需要注意数据的敏感性, 指定需要同步的字段, 不必要的字段就别同步了, 你懂得。

API 搜索

看各编程语言的插件了, 我使用的 elasticsearch-php 这货, 先在 Kibana 的 Sense 插件里手工调试好后, 再移植到代码里写 DSL 语句。