最近我们搭建了一个新的 Elasticsearch（以下简称 ES）集群，集群中有 3 个 master 节点和 6 个 data 节点，集群使用到的 ES 版本为 7.5.2。其中 master 节点 JVM 的堆内存设置为 2GB，data 节点 JVM 的堆内存设置为 30GB，master 和 data 节点都只用作单一的节点角色（即节点不会同时是 master 和 data 两种角色）。因为 master 不存储数据，所以我们给 master 三台负载的配置都比较低。

master 配置不够导致读写出错

我们在操作 ES 集群的时候使用了 elasticsearch-py 这个 ES 的 Python 操作库，问题在于这个库在操作 ES 的时候会先根据 ES 的/_nodes/_all/http接口获取集群所有节点的 HTTP 接口地址，之后利用这些接口地址对 ES 进行读写。当然 elasticsearch-py 会在这些接口地址间做负载均衡以及错误重试等等操作，但是/_nodes/_all/http接口会同时返回 master、data 以及一些其它角色节点的 HTTP 接口地址，这就导致 elasticsearch-py 在后面操作集群的过程中既会读写 data 节点，也会读写 master 节点。

因为我们 master 节点的配置很低，所以一旦在 master 节点上面进行读写操作，那么 master 节点的压力尤其是内存的压力就会比较大，经常就会出现circuit_breaking_exception的错误。

解决办法就是利用的 elasticsearch-py 的host_info_callback参数来过滤要操作的节点，具体使用方式如下

在创建 ES 连接对象的时候指定回调方法

1
2
3
4
5
6
from elasticsearch import Elasticsearch

es = Elasticsearch(
    host_info_callback=not_master_nodes,  # 指定操作的回调方法
    sniff_on_start=True, sniff_on_connection_fail=True, sniffer_timeout=60
)

之后我们创建not_master_nodes过滤方法如下

1
2
3
4
5
6
7
8
9
def not_master_nodes(node_info, host):
    """
    由于master节点的性能较差，所以过滤掉master节点
    :param node_info:
    :param host:
    :return:
    """
    roles = node_info.get('roles', [])
    return host if 'master' not in roles else None

逻辑非常简单，就是把roles属性中包含了master的节点给剔除即可。

elasticsearch-py 会在 _get_host_info 方法中 调用 host_info_callback 方法，并且在 sniff_hosts 方法中 对不符合要求的节点进行过滤，具体实现逻辑可以参考前面的源码连接。

添加了host_info_callback属性之后，elasticsearch-py 就再也不会操作 master 节点了。这样带来了两个好处

  1. master 节点的压力降低，提升了 master 节点的稳定性
  2. 读写操作也不会因为 master 节点的内存不足而报错了，提升了读写操作的稳定性

我们把上面的代码上线之后，通过nload -u M命令查看三台 master 节点的网速，发现三个节点的平均网速均从0.3 MByte/s降到了0.02 MByte/s，网速下降非常明显。此外通过命令netstat -an | grep 9200也看不到任何与 master 节点的 9200 端口的连接了，说明此时 elasticsearch-py 已经不再连接 master 节点了。

为什么之前从来没有出过这个问题？

我们使用 ES 已经很久了，为什么之前的 1.5.2 版本的集群都很正常，但是到了 7.5.2 版本的集群上就会出现 master 节点内存不足的错误呢？带着这样的疑问我们也查看了一下 1.5.2 版本的 elasticsearch-py 的源码，发现在 1.5.2 的源码中同样会 根据 host_info_callback 方法来过滤节点，区别在于 1.5.2 的源码中在创建 Transport 对象的时候会给 host_info_callback 参数设置一个默认值：host_info_callback=get_host_info。

get_host_info 方法在源码中已经 定义好了，这里摘录如下

1
2
3
4
5
6
7
8
9
def get_host_info(node_info, host):
    attrs = node_info.get('attributes', {})

    # ignore master only nodes
    if (attrs.get('data', 'true') == 'false' and
        attrs.get('client', 'false') == 'false' and
        attrs.get('master', 'true') == 'true'):
        return None
    return host

可以看到在 1.5.2 版本的 elasticsearch-py 中，库本身就已经帮我们过滤掉了纯 master 节点了，也就是说 1.5.2 是不会读写纯粹的 master 节点的，这也难怪为什么我们之前从来没有遇到过这个问题了。至于为什么在后续版本中 elasticsearch-py 把这个特性去掉了，我猜也许是为了避免让库对用户的操作进行过多的干涉吧，因为想不想读 master 节点这种事情本来也应该交给用户来决定而不是库本身擅自决定的。

应用本身的一些配置

如果使用两个版本的 elasticsearch-py

我们的应用现在在同时读写 1.5.2 和 7.5.2 这两个版本的集群，因为使用两个版本的 elasticsearch-py 会导致包冲突，我们的解决办法是把这两个版本的 elasticsearch-py 的源码直接复制到我们应用的源码中，两个版本的源码分别放在应用中的 elasticsearch1 和 elasticsearch7 这两个模块的文件夹中，之后想要调用时直接使用import elasticsearch1和import elasticsearch7导入模块即可。

如何配置 seed hosts

我们现在已经知道了 elasticsearch-py 操作 ES 分为两个步骤

  1. 通过配置的 seed hosts 访问 ES 集群，根据 ES 提供的接口获取到 ES 集群所有节点的 HTTP 接口，之后根据需要剔除掉一些节点（这一步在 1.5.2 和 7.5.2 中有所差异），最终得到一个符合我们需要的 ES 集群的节点 HTTP 接口列表（在 elasticsearch-py 中这一步操作对应的方法叫做sniff_hosts）
  2. 通过第一步拿到的节点列表来对 ES 集群进行真正的读写操作

由此我们可以知道我们配置的 seed hosts 并不是一定会作为真正的读写节点的，真正读写的节点会在第一步操作中通过接口获取并进行判断得到的。所以我们现在在设置 seed hosts 时会把 seed hosts 设置为所有的 master 节点地址，这样的好处在于 master 节点基本上不会更换，而 data 节点可能会频繁的变更（例如更换硬盘、增加配置等等），使用 master 节点作为 seed hosts 就保证了可以在 data 节点变更时不再需要修改配置。

总结

其实本文核心过程就是这几部分

  1. 设置 seed hosts
  2. 以 seed hosts 为基础，根据 sniff_hosts 方法拿到集群的全部节点
  3. 对拿到的全部节点进行过滤，过滤之后剩下的就是我们想要的节点
  4. 通过过滤之后的节点对集群进行真正的操作

参考

https://discuss.elastic.co/t/how-to-only-query-on-data-nodes-by-elasticsearch-py/249293
https://github.com/elastic/elasticsearch-py/issues/1378

2020-09-24 补充

后来经过研究发现其实在 elasticsearch-py 的 7.5.2 版本的源码中也是定义了 get_host_info 方法的，摘录如下

1
2
3
4
5
def get_host_info(node_info, host):
    # ignore master only nodes
    if node_info.get("roles", []) == ["master"]:
        return None
    return host

上面的逻辑表示，只要一个节点的 roles 是["master"]那么默认就会被剔除。只是我调用 ES 的接口/_nodes/_all/http查看了一下 7.5.2 集群的配置，发现我的 master 节点的 roles 是["master", "ml"]，还附带了一个ml的角色，因此不能匹配上面的条件，导致我的 master 被保留了下来。

知道了原理之后，我们就知道了解决办法了。除了像上面重写过滤方法之外，我们也可以把 master 节点的ml的角色设置为 false，根据 ES 文档，我们只需要在 master 节点中添加如下设置即可

node.ml: false

2020-11-03 补充

通过总结我们可以得到一个最佳实践

对于集群本身，我们先创建三个 master 节点，它们的配置如下

discovery.seed_hosts: [{{ alias.MasterIPList }}]
cluster.initial_master_nodes: [{{ alias.MasterIPList }}]

集群在刚刚启动的时候 master 节点的 seed_hosts 和 initial_master_nodes 都需要设置为 master 节点的地址列表。在集群启动完毕之后，我们可以添加 data 节点或者 ingest 节点等等其它的节点，它们只需要把 seed_hosts 设置为当前的 master 节点即可

discovery.seed_hosts: [{{ alias.MasterIPList }}]

而对于客户端，同样我们在设置 ES 地址的时候也只需要设置 master 节点的地址就可以了

1
2
3
{
    "elasticsearch_hosts": [{{ alias.MasterIPList }}]
}

es 的 Python 客户端会使用 master 的节点地址作为种子，通过种子获取了 data 节点的地址之后再在 data 节点上面进行真正的读写操作。

附上我们目前使用的 elasticsearch.yml 和 jvm.options 配置文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
# ======================== Elasticsearch Configuration =========================
#
# NOTE: Elasticsearch comes with reasonable defaults for most settings.
#       Before you set out to tweak and tune the configuration, make sure you
#       understand what are you trying to accomplish and the consequences.
#
# The primary way of configuring a node is via this file. This template lists
# the most important settings you may want to configure for a production cluster.
#
# Please consult the documentation for further information on configuration options:
# https://www.elastic.co/guide/en/elasticsearch/reference/index.html
#
# ---------------------------------- Cluster -----------------------------------
#
# Use a descriptive name for your cluster:
#
cluster.name: {{ attrs.t_clustername }}
#
# ------------------------------------ Node ------------------------------------
#
# Use a descriptive name for the node:
#
node.name: {{ cHost.name }}
#
# Add custom attributes to the node:
#
# node.attr.rack: r1
#
node.master: {{ attrs.t_is_master }}
#
node.data: {{ attrs.t_is_data }}
#
node.ingest: {{ attrs.t_is_ingest }}
#
node.ml: false
#
# ----------------------------------- Paths ------------------------------------
#
# Path to directory where to store the data (separate multiple locations by comma):
#
path.data: /home/elasticsearch/data
#
# Path to log files:
#
path.logs: /home/elasticsearch/elasticsearch/logs
#
# ----------------------------------- Memory -----------------------------------
#
# Lock the memory on startup:
#
#bootstrap.memory_lock: true
bootstrap.system_call_filter: false
#
# Make sure that the heap size is set to about half the memory available
# on the system and that the owner of the process is allowed to use this
# limit.
#
# Elasticsearch performs poorly when the system is swapping the memory.
#
# ---------------------------------- Network -----------------------------------
#
# Set the bind address to a specific IP (IPv4 or IPv6):
#
network.host: {{ cHost.ip }}
#
# Set a custom port for HTTP:
#
http.port: {{ attrs.t_server_port }}
# 
# For more information, consult the network module documentation.
#
# --------------------------------- Discovery ----------------------------------
#
# Pass an initial list of hosts to perform discovery when this node is started:
# The default list of hosts is ["127.0.0.1", "[::1]"]
#
discovery.seed_hosts: [{{ alias.MasterIPList }}]
#
# Bootstrap the cluster using an initial set of master-eligible nodes:
#
# cluster.initial_master_nodes: [{{ alias.MasterIPList }}]
#
# For more information, consult the discovery and cluster formation module documentation.
#
# ---------------------------------- Gateway -----------------------------------
#
# Block initial recovery after a full cluster restart until N nodes are started:
#
#gateway.recover_after_nodes: 3
#
# For more information, consult the gateway module documentation.
#
# ---------------------------------- Various -----------------------------------
#
# Require explicit names when deleting indices:
#
#action.destructive_requires_name: true

path.repo: ["/home/elasticsearch/ES_backup/7-5-2"]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
## JVM configuration

################################################################
## IMPORTANT: JVM heap size
################################################################
##
## You should always set the min and max JVM heap
## size to the same value. For example, to set
## the heap to 4 GB, set:
##
## -Xms4g
## -Xmx4g
##
## See https://www.elastic.co/guide/en/elasticsearch/reference/current/heap-size.html
## for more information
##
################################################################

# Xms represents the initial size of total heap space
# Xmx represents the maximum size of total heap space

-Xms{{attrs.t_heap_size}}
-Xmx{{attrs.t_heap_size}}

################################################################
## Expert settings
################################################################
##
## All settings below this section are considered
## expert settings. Don't tamper with them unless
## you understand what you are doing
##
################################################################

## GC configuration
-XX:+UseConcMarkSweepGC
-XX:CMSInitiatingOccupancyFraction=75
-XX:+UseCMSInitiatingOccupancyOnly

## G1GC Configuration
# NOTE: G1GC is only supported on JDK version 10 or later.
# To use G1GC uncomment the lines below.
# 10-:-XX:-UseConcMarkSweepGC
# 10-:-XX:-UseCMSInitiatingOccupancyOnly
# 10-:-XX:+UseG1GC
# 10-:-XX:G1ReservePercent=25
# 10-:-XX:InitiatingHeapOccupancyPercent=30

## JVM temporary directory
-Djava.io.tmpdir=${ES_TMPDIR}

## heap dumps

# generate a heap dump when an allocation from the Java heap fails
# heap dumps are created in the working directory of the JVM
-XX:+HeapDumpOnOutOfMemoryError

# specify an alternative path for heap dumps; ensure the directory exists and
# has sufficient space
-XX:HeapDumpPath=data

# specify an alternative path for JVM fatal error logs
-XX:ErrorFile=logs/hs_err_pid%p.log

## JDK 8 GC logging
8:-XX:+PrintGCDetails
8:-XX:+PrintGCDateStamps
8:-XX:+PrintTenuringDistribution
8:-XX:+PrintGCApplicationStoppedTime
8:-Xloggc:logs/gc.log
8:-XX:+UseGCLogFileRotation
8:-XX:NumberOfGCLogFiles=32
8:-XX:GCLogFileSize=64m

# JDK 9+ GC logging
9-:-Xlog:gc*:file=logs/gc.log:t,tags,level:filecount=32,filesize=64m