很多人都接触过 Elasticsearch。但是，如果要用它来存储“超大型”日志呢？如果还要确保在多个数据中心中任何一个发生故障时都能顺利恢复，又该怎么办？应该选择什么样的架构？又会遇到哪些陷阱？
在 Odnoklassniki，我们决定使用 Elasticsearch 来解决我们的日志管理问题，现在我们正在分享我们使用 Habr 的经验，涵盖其架构和陷阱。
我是 Petr Zaitsev，Odnoklassniki 的系统管理员。在此之前，我也是一名管理员，负责 Manticore Search、Sphinx Search 和 Elasticsearch 的相关工作。如果以后有其他搜索引擎出现，我可能也会参与其中。此外，我还以志愿者的身份为多个开源项目做贡献。
加入 Odnoklassniki 时，我在面试中贸然提到我会使用 Elasticsearch。在掌握了基本操作并完成了一些简单的任务后，我被委以重任，负责改造当时现有的日志管理系统。
需求
系统需求表述如下：
- 我们选择了 Graylog 作为前端。公司之前已经使用过这款产品；程序员和测试人员都很熟悉它，而且它用起来很方便。
- 数据量：平均每秒 50 万至 80 万条消息，但如果出现故障，流量将不受限制，可达每秒 2 万至 3 万行。
- 在与客户讨论搜索查询处理速度要求后，我们意识到此类系统的典型使用模式如下：人们搜索过去两天的应用程序日志，并且不希望等待超过一秒钟才能返回结果。
- 管理员们坚持认为，该系统在必要时应该易于扩展，而无需用户深入了解其工作原理。
- 因此，这些系统唯一需要定期维护的任务就是更换一些硬件。
- 此外，Odnoklassniki 拥有优良的技术传统：我们推出的任何服务都必须能够经受住数据中心故障（突然的、计划外的、绝对任何时间发生的故障）。
该项目实施过程中的最后一个要求让我们付出了最大的代价，稍后我会详细说明。
星期三
我们在四个数据中心运营，但 Elasticsearch 数据节点只能位于其中三个数据中心（由于一些非技术原因）。
这四个数据中心包含大约 18 个不同的日志源——硬件、容器和虚拟机。
一个重要特性：集群是在容器中启动的。 不是在物理机器上，而是在 容器保证使用两个核心，相当于 2.0Ghz v4，如果剩余的核心空闲，则可以选择使用它们。
换句话说：
拓扑结构
最初，我设想的解决方案大致如下：
- Graylog 域的 A 记录背后有 3-4 个 VIP，该记录是日志发送到的地址。
- 每个 VIP 都是 LVS 平衡器。
- 之后，日志被发送到 Graylog 电池，部分数据为 GELF 格式，部分数据为 syslog 格式。
- 然后所有这些数据都会批量写入一组 Elasticsearch 协调器。
- 它们反过来向相关的数据节点发送写入和读取请求。
术语
或许并非每个人都能详细理解这些术语，所以我想再详细解释一下。
Elasticsearch 有几种节点类型：主节点、协调器节点和数据节点。还有另外两种类型用于各种日志转换和连接不同的集群，但我们只使用了上面列出的这些。
总音量
对集群中存在的所有节点进行 Ping，维护最新的集群映射并将其分发到各个节点，处理事件逻辑，并执行各种类型的集群范围的清理工作。
协调员
它只执行一项任务：接收来自客户端的读取或写入请求并路由这些流量。如果请求是写入操作，它很可能会询问主节点将请求放入相关索引的哪个分片中，然后将请求转发出去。
数据节点
存储数据，执行来自外部的搜索查询，并对位于其上的分片执行操作。
灰色日志
Graylog 就像是 ELK 技术栈中 Kibana 和 Logstash 的融合体。它同时集成了用户界面和日志处理管道。底层，Graylog 使用 Kafka 和 Zookeeper，从而实现了 Graylog 集群的内聚性。Graylog 可以在 Elasticsearch 不可用时缓存日志（Kafka），重试失败的读写请求，并根据自定义规则对日志进行分组和标记。与 Logstash 类似，Graylog 也能够在写入 Elasticsearch 之前修改字符串。
Graylog 还内置了服务发现功能，允许您从单个可用的 Elasticsearch 节点检索整个集群映射，并按特定标签进行筛选，从而允许您将查询定向到特定容器。
外观上看起来大概是这样的：
这是特定实例的屏幕截图。在这里，我们根据搜索查询构建直方图并显示相关行。
指数
回到系统架构方面，我想更详细地谈谈我们是如何构建索引模型的，以便一切能够正常运行。
在上图中，这是最低层：Elasticsearch 数据节点。
索引是一个由 Elasticsearch 分片组成的大型虚拟实体。每个分片本身就是一个 Lucene 索引。而每个 Lucene 索引又由一个或多个分片组成。
在设计过程中，我们假设为了满足大量数据的读取速度要求，我们需要将这些数据均匀地分布在各个数据节点上。
这意味着每个索引（带副本）的分片数必须严格等于数据节点数。首先，这是为了确保复制因子为 2（意味着我们可以承受一半集群的损失）。其次，这是为了确保读写请求至少由集群的一半节点处理。
我们最初确定的储存时间为 30 天。
分片分布可以用下图表示：
整个深灰色矩形区域是索引。左侧的红色方块是主分片，即索引中的第一个分片。蓝色方块是副本分片。它们位于不同的数据中心。
当我们添加另一个分片时，它会迁移到第三个数据中心。最终，我们得到一种结构，即使某个数据中心发生故障，也不会丢失数据一致性：
我们将索引轮换周期（即创建新索引并删除最旧的索引）设置为 48 小时（基于索引使用模式：最近 48 小时的搜索频率最高）。
此索引轮换周期是由于以下原因造成的：
当搜索查询到达特定数据节点时，如果单个分片的大小与节点的堆内存大小相当，那么从性能角度来看，查询单个分片更为高效。这样可以将索引的“热点”部分保留在堆内存中，从而实现快速访问。如果存在多个“热点”部分，则索引搜索性能会下降。
当节点开始在单个分片上执行搜索查询时，它会分配与物理机上超线程核心数相等的线程数。如果搜索查询跨越多个分片，线程数将成比例增加。这会对搜索性能和新数据的索引产生负面影响。
为了确保所需的搜索延迟，我们决定使用固态硬盘 (SSD)。为了快速处理查询，运行这些容器的机器至少需要 56 个核心。选择 56 这个数字是因为它比较合理，决定了 Elasticsearch 在运行期间将生成的线程数。在 Elasticsearch 中，许多线程池参数直接取决于可用核心数，而可用核心数又直接影响集群中所需的节点数，遵循“核心数越少，节点数越多”的原则。
因此，我们发现平均每个分片的大小约为 20 GB，每个索引有 360 个分片。所以，如果我们每 48 小时轮换一次分片，则需要 15 次。每个索引包含两天的数据。
数据书写和读取方案
我们来弄清楚这个系统是如何记录数据的。
假设我们从 Graylog 收到一个发送到协调器的查询。例如，我们想要索引 2 到 3 行数据。
协调器收到 Graylog 的请求后，向主节点询问：“在索引请求中，我们明确指定了索引，但没有指定要写入哪个分片。”
主节点响应：“将此信息写入分片号 71”，然后该信息直接发送到相关的数据节点，主分片号 71 就位于该节点上。
之后，事务日志会被复制到位于另一个数据中心的副本分片。
Graylog 向协调器发送搜索查询。协调器将其路由到索引，Elasticsearch 使用轮询机制将查询分发到主分片和副本分片。
这180个节点的响应速度并不均衡。在它们响应的同时，协调器会收集速度更快的数据节点已经“输出”的信息。之后，当所有信息都到达或请求超时后，协调器会将所有信息直接发送给客户端。
整个系统处理过去 48 小时的搜索查询平均耗时 300-400 毫秒，不包括以通配符开头的查询。
Elasticsearch：Java 设置
为了让所有功能都按照我们最初设想的方式运行，我们花了很长时间调试集群中的各种问题。
发现的第一组问题与 Elasticsearch 中 Java 的默认配置方式有关。
第一个问题
我们发现大量报告显示，当后台作业在我们的 Lucene 层运行时，Lucene 段合并操作失败。日志显示这是一个 OutOfMemoryError 错误。遥测数据显示堆内存是空闲的，但尚不清楚该操作失败的原因。
事实证明，Lucene 索引合并操作发生在堆外。而且容器的资源消耗受到严格限制。只有堆内存能够容纳所有操作（堆大小大致等于 RAM），如果某些堆外操作由于某种原因超出剩余的约 500MB 内存限制，就会因内存分配错误而失败。
解决方法很简单：我们增加了容器可用的内存量，然后就忘了我们一开始就遇到过这样的问题。
问题二
集群启动后大约 4-5 天，我们注意到数据节点开始周期性地从集群中掉线，然后每隔 10-20 秒重新加入集群。
深入研究后发现，Elasticsearch 的堆外内存实际上不受控制。当我们为容器分配更多内存时，可以直接用各种数据填充缓冲区池，而这些数据只有在 Elasticsearch 执行显式垃圾回收后才会被清除。
在某些情况下，此操作耗时相当长，在此期间，集群会将节点标记为已退出。这个问题已有详细描述。 .
解决方案是这样的：我们限制了 Java 使用大量非堆内存进行这些操作的能力。我们将其上限设为 16 GB（-XX:MaxDirectMemorySize=16g），确保显式垃圾回收器被更频繁地调用，并且完成速度显著加快，从而消除集群不稳定。
问题三
如果你认为“节点在最意想不到的时刻离开集群”的问题就此结束了，那就大错特错了。
在配置索引工作时，我们选择了 mmapfs，以便： 在高度分段的新分片上，这是一个相当严重的错误。因为在使用 mmapfs 时，文件会被映射到 RAM 中，然后我们直接操作映射后的文件。正因如此，当垃圾回收器尝试停止应用程序中的线程时，需要很长时间才能到达安全点。在此过程中，应用程序会停止响应主节点关于其是否存活的查询。因此，主节点会认为该节点已从集群中移除。5-10 秒后，垃圾回收器再次运行，节点恢复运行，重新加入集群，并开始初始化分片。这一切都非常像我们“应有的生产环境”，根本不适合任何严肃的应用场景。
为了解决这个问题，我们首先切换到了标准的 niofs 文件系统，然后在从 Elastic 5 迁移到 Elastic 6 时，我们尝试了 hybridfs 文件系统，这个问题在 hybridfs 中没有重现。您可以在这里阅读更多关于存储类型的信息。 .
问题四
后来我们又遇到了一个非常有趣的问题，我们花了创纪录的时间去研究它。我们花了两个月到三个月的时间试图找出答案，因为它的规律完全不明朗。
有时我们的垃圾回收器会在午餐时间前后进入完全垃圾回收模式，然后就再也没有回来。记录垃圾回收延迟时，情况是这样的：一切都很顺利，然后突然——一切都出错了。
起初，我们以为是某个恶意用户运行了某种查询，导致协调器无法正常工作。我们花了很长时间记录查询日志，试图弄清楚到底发生了什么。
我们发现，当用户发出大型查询并命中特定的 Elasticsearch 协调器时，有些节点的响应时间比其他节点要长。
在协调器等待所有节点响应期间，它会累积已响应节点发送的结果。对于垃圾回收器而言，这意味着堆内存使用模式变化非常快。而我们之前使用的垃圾回收器无法处理这种情况。
我们发现唯一能改变集群这种行为的方法是迁移到 JDK 13 并使用 Shenandoah 垃圾回收器。这样就解决了问题，我们的协调器不再崩溃了。
Java 问题到此结束，带宽问题开始了。
Elasticsearch 的“Berrys”：吞吐量
吞吐量问题意味着我们的集群虽然稳定，但在索引文档数量达到峰值和进行操作期间性能不足。
我遇到的第一个症状是：在生产环境中发生某种“爆炸”时，突然生成大量日志，导致 Graylog 中频繁出现 es_rejected_execution 索引错误。
这是因为默认情况下，单个数据节点上的 thread_pool.write.queue 只能缓存 200 个请求，直到 Elasticsearch 能够处理索引请求并将信息写入磁盘上的分片。 关于这个参数的说明非常少，只规定了最大线程数和默认大小。
当然，我们开始尝试调整这个值，并发现：在我们特定的设置中，最多 300 个请求可以很好地缓存，而更高的值则有可能导致我们再次陷入完全垃圾回收。
此外，由于这些消息是一次性请求到达的多批消息，因此需要对 Graylog 进行调整，使其不再频繁地以小批量写入，而是以大批量写入，或者如果批次未满则每三秒写入一次。这意味着写入 Elasticsearch 的信息需要五秒钟才能可用，而不是两秒钟（这完全可以接受），同时也减少了写入大量信息所需的重试次数。
在某个地方发生故障并疯狂报告故障时，这一点尤其重要，以免 Elastic 服务器被彻底淹没，一段时间后，Graylog 节点因缓冲区堵塞而无法运行。
此外，当我们在生产环境中遇到这些爆炸事件时，我们收到了程序员和测试人员的抱怨：当他们真正需要这些日志的时候，日志却迟迟没有提供给他们。
我们开始调查此事。一方面，很明显，搜索查询和索引请求本质上都是在同一台物理机上处理的，因此，无论如何都会出现一些性能下降。
但 Elasticsearch 6 版本引入了一种算法，可以部分规避这个问题。该算法并非采用轮询方式随机分配请求到相关的数据节点（因为承载主分片的索引容器可能非常繁忙，无法快速响应），而是将请求路由到负载较低的、承载副本分片的容器，从而显著加快响应速度。换句话说，我们最终采用了 `use_adaptive_replica_selection: true` 参数。
阅读的画面开始变成这样：
改用此算法后，在需要写入大量日志时，查询速度得到了显著提升。
最后，主要问题是如何无痛地拆除数据中心。
在与某个数据中心失去连接后，我们希望集群立即实现以下功能：
- 如果当前主节点位于发生故障的数据中心，它将被重新选举并作为角色转移到另一个数据中心的另一个节点。
- 主节点会迅速将所有不可用的节点从集群中移除。
- 根据剩余的数据，他会明白：在丢失的数据中心里我们有这样的主分片，他会迅速在剩余的数据中心里提升相应的副本分片，我们将继续索引数据。
- 因此，集群的读写吞吐量会逐渐下降，但总体而言，一切都会正常运行，尽管速度较慢，但运行稳定。
结果，我们想要的是这样的东西：
我们得到了以下结果：
这是怎么发生的？
数据中心崩溃后，我们的主服务器成了瓶颈。
为什么呢？
问题在于，主节点有一个 TaskBatcher，它负责将特定的任务和事件分发到整个集群。任何节点退出、任何分片从副本提升为主节点、任何在某处创建分片的任务——所有这些都会首先发送到 TaskBatcher，在那里它们会按顺序在单个线程中处理。
当一个数据中心宕机时，幸存数据中心中的所有数据节点都认为有义务通知主服务器：“我们丢失了某某分片和某某数据节点。”
与此同时，幸存的数据节点将所有这些信息发送给当前主节点，并试图等待其确认接收。但它们未能等到，因为主节点接收任务的速度远超其响应速度。节点超时并重复发送请求，而主节点甚至不再尝试响应，而是完全专注于按优先级对请求进行排序。
终端显示，数据节点不断向主节点发送大量数据，直到主节点触发完全垃圾回收。之后，主节点的角色会转移到其他节点，而这些节点也会经历同样的情况，最终导致集群彻底崩溃。
我们进行了测量，在 6.4.0 版本修复此问题之前，我们只需同时从 360 个数据节点中移除 10 个，就足以使集群完全瘫痪。
它看起来大概是这样的：
6.4.0 版本修复了这个棘手的 bug 之后，数据节点不再导致主节点崩溃。但这并没有让系统变得更“智能”。具体来说，当我们启动 2 个、3 个或 10 个（除 1 个以外的任何数量）数据节点时，主节点会收到一条初始消息，表明节点 A 已退出，然后它会尝试通知节点 B、节点 C 和节点 D。
目前，唯一应对这种情况的方法是设置一个超时时间，即尝试通知某人某些信息的时间大约为 20-30 秒，从而控制数据中心从集群中移除的速度。
原则上，这符合项目最初设定的最终产品要求，但从纯粹的科学角度来看，这是一个漏洞。顺便一提，开发人员已在 7.2 版本中成功修复了这个问题。
此外，当某个数据节点发生故障时，分发有关该节点故障的信息比告诉整个集群哪些主分片位于该节点上更重要（以便将另一个数据中心中的副本分片提升为主分片，并将信息写入其中）。
因此，一旦所有数据都已停止运行，退出的数据节点不会立即被标记为过期。因此，我们必须等到所有发送到退出数据节点的 ping 请求都超时后，集群才会开始发出信号，表明需要在某个位置继续进行数据记录。您可以在这里阅读更多相关信息。 .
因此，目前我们的数据中心停机维护操作在高峰时段大约需要5分钟。对于这样一台庞大而笨重的机器来说，这是一个相当不错的成绩。
因此，我们做出了以下决定：
- 我们有 360 个数据节点，配备 700 GB 的磁盘。
- 60 个协调器用于在这些数据节点之间路由流量。
- 我们之前从 6.4.0 版本遗留下来的 40 个主服务器——为了在数据中心关闭的情况下生存下来，我们已经做好了失去几台机器的心理准备，以确保即使在最糟糕的情况下，我们也能拥有足够的主服务器。
- 任何试图在单个容器上合并角色的尝试最终都会导致节点在负载下崩溃。
- 整个集群使用 31 GB 的堆大小：所有减少大小的尝试都导致带有前导通配符的繁重搜索查询要么导致某些节点崩溃，要么导致 Elasticsearch 本身发生熔断。
- 此外，为了确保搜索性能，我们尽量减少集群中的对象数量，以便在主节点的瓶颈处处理尽可能少的事件。
最后，关于监测
为确保所有功能按预期运行，我们会监控以下各项：
- 每个数据节点都会向我们的云平台报告自身的存在以及包含的分片数量。当我们关闭某个节点时，集群会在 2-3 秒内报告说，在 A 数据中心，我们关闭了节点 2、3 和 4——这意味着在其他数据中心，我们绝对不能关闭只剩下一个分片的节点。
- 了解主节点的行为模式后，我们会密切关注待处理任务的数量。因为即使只有一个任务卡住，如果不能及时超时，在紧急情况下也可能导致诸如副本分片无法提升为主分片等操作，从而阻碍索引的进行。
- 我们还密切关注垃圾回收器的延迟，因为我们过去在优化过程中曾遇到过这方面的重大困难。
- 按线程进行拒绝，以便提前了解瓶颈所在。
- 当然，还有一些标准指标，例如堆内存、RAM 和 I/O。
构建监控系统时，务必考虑 Elasticsearch 中的线程池特性。 它描述了搜索和索引的配置选项和默认值，但完全没有提及 thread_pool.management。这些线程负责处理诸如 _cat/shards 之类的查询，以及其他类似的查询，这些查询便于编写监控脚本。集群规模越大，单位时间内执行的此类查询就越多，而上述 thread_pool.management 不仅在官方文档中没有提及，而且默认限制为五个线程，很快就会被五个线程耗尽，导致监控脚本无法正常工作。
总之，我们成功了！我们为程序员和开发人员提供了一个工具，该工具几乎可以在任何情况下快速可靠地提供有关生产环境中正在发生的事情的信息。
是的，事实证明这相当困难，但尽管如此，我们还是设法将我们的愿望融入到现有产品中，而无需进行修补或重写。
来源： habr.com
