【腾讯云ES】Elasticsearch 分布式架构剖析及扩展性优化

// 当前 master
private final String masterNodeId;
// 本地节点
private final String localNodeId;
// 完整的节点列表
private final ImmutableOpenMap<String, DiscoveryNode> nodes;

// 其它分类型节点列表
private final ImmutableOpenMap<String, DiscoveryNode> dataNodes;
private final ImmutableOpenMap<String, DiscoveryNode> masterNodes;
private final ImmutableOpenMap<String, DiscoveryNode> ingestNodes;

// 集群的动态设定，transient 全量重启后就没有了，persistent 会持久化
private final Settings transientSettings;
private final Settings persistentSettings;

// 索引、模板信息
private final ImmutableOpenMap<String, IndexMetaData> indices;
private final ImmutableOpenMap<String, IndexTemplateMetaData> templates;

// 集群选举管理信息
private final CoordinationMetaData coordinationMetaData;

// raft 选举使用的 term
private final long term;

// 最近提交的选举节点列表，内部是一个 Set<String> nodeIds
private final VotingConfiguration lastCommittedConfiguration;

// 最近接收的选举节点列表
private final VotingConfiguration lastAcceptedConfiguration;

// 用户通过 _cluster/voting_config_exclusions 接口设定的选举排除节点列表
private final Set<VotingConfigExclusion> votingConfigExclusions;

// key 为索引名，IndexRoutingTable 为该索引包含的分片信息
private final ImmutableOpenMap<String, IndexRoutingTable> indicesRouting;

// 索引信息，包含索引名称、uuid 等。
private final Index index;

// 索引的分片列表，key 为分片的编号，例如 2 号分片，3 号分片
private final ImmutableOpenIntMap<IndexShardRoutingTable> shards;

// 当前分片信息，主要包含分片的编号、分片对应的索引信息
final ShardId shardId;

// 当前分片的主分片信息
final ShardRouting primary;

// 当前分片的多个从分片列表
final List<ShardRouting> replicas;

// 当前分片全量的分片列表，包含主、副本
final List<ShardRouting> shards;

// 当前分片已经 started 的活跃分片列表
final List<ShardRouting> activeShards;

// 当前分片已经分配了节点的分片列表
final List<ShardRouting> assignedShards;

private final ShardId shardId;

// 分片所在的当前节点 id
private final String currentNodeId;

// 如果分片正在搬迁，这里为目标节点 id
private final String relocatingNodeId;

// 是否是主分片
private final boolean primary;

// 分片的状态，UNASSIGNED/INITIALIZING/STARTED/RELOCATING
private final ShardRoutingState state;

// 每一个分片分配后都有一个唯一标识
private final AllocationId allocationId;

// 节点到分片的映射，key 为 nodeId，value 为一个包含分片列表的节点信息
private final Map<String, RoutingNode> nodesToShards = new HashMap<>();

// 未分配的分片列表，每次 reroute 从这里获取分片分配
private final UnassignedShards unassignedShards = new UnassignedShards(this);

// 已经分配的分片列表
private final Map<ShardId, List<ShardRouting>> assignedShards = new HashMap<>();

// 当前分片远程恢复的统计数据 incoming/outcoming
private final Map<String, Recoveries> recoveriesPerNode = new HashMap<>();

private final String nodeId;

// 该节点上的分片列表
private final LinkedHashMap<ShardId, ShardRouting> shards; 

// 该节点上的初始化中的分片列表
private final LinkedHashSet<ShardRouting> initializingShards;

// 该节点上的正在搬迁（搬走）的分片列表
private final LinkedHashSet<ShardRouting> relocatingShards;

├── indices
│   ├── 00IUbRWZSzGsN71k0TlPmA
│   │   └── _state
│   │   └── state-8.st
│   ├── 01RTGqCbRe-9PYO55j8zAQ
│   │   └── _state
│   │   └── state-10.st
│   ├── 02DSrGNzRzizuI42Yf6aJg
│   │   └── _state
│   │   └── state-17.st

├── node.lock
└── _state
├── global-4928.st
└── node-0.st

├── node.lock
└── _state
├── _3q.cfe
├── _3q.cfs
├── _3q.si
├── _3r.cfe
├── _3r.cfs
├── _3r.si
├── node-0.st
├── segments_6i
└── write.lock

* +------------------------------+-----------------------------+----------------------------------------------+
* | "type" (string field)        | "index_uuid" (string field) | "data" (stored binary field in SMILE format) |
* +------------------------------+-----------------------------+----------------------------------------------+
* | GLOBAL_TYPE_NAME == "global" | (omitted)                   | Global metadata                              |
* | INDEX_TYPE_NAME  == "index"  | Index UUID                  | Index metadata                               |
* +------------------------------+-----------------------------+----------------------------------------------+
*
* Additionally each commit has the following user data:
*
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* |        Key symbol         |       Key literal       |                                     Value                                     |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+
* | CURRENT_TERM_KEY          | "current_term"          | Node's "current" term (≥ last-accepted term and the terms of all sent joins)  |
* | LAST_ACCEPTED_VERSION_KEY | "last_accepted_version" | The cluster state version corresponding with the persisted metadata           |
* | NODE_ID_KEY               | "node_id"               | The (persistent) ID of the node that wrote this metadata                      |
* | NODE_VERSION_KEY          | "node_version"          | The (ID of the) version of the node that wrote this metadata                  |
* +---------------------------+-------------------------+-------------------------------------------------------------------------------+

if (DiscoveryNode.isMasterNode(settings) || DiscoveryNode.canContainData(settings)) {
    ......
    if (DiscoveryNode.isMasterNode(settings)) {
        persistedState = new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState);
    } else {
        persistedState = new AsyncPersistedState(
            settings,
            transportService.getThreadPool(),
            new LucenePersistedState(persistedClusterStateService, currentTerm, clusterState)
        );
    }        
    ......
    this.persistedState.set(persistedState);
} else {
    ......
    persistedState.set(new InMemoryPersistedState(currentTerm, clusterState));
}

enum PublicationTargetState {
    NOT_STARTED,
    FAILED,
    SENT_PUBLISH_REQUEST, // 已发送 publish 请求
    WAITING_FOR_QUORUM,   // 等待大多数节点响应
    SENT_APPLY_COMMIT,    // 已发送 commit 请求
    APPLIED_COMMIT,       // 元数据应用完毕
}

private void onlyCreateIndex(final CreateIndexClusterStateUpdateRequest request, final ActionListener<AcknowledgedResponse> listener) {
    normalizeRequestSetting(request);
    //提交元数据变更任务，索引创建优先级为 URGENT
    submitUnbatchedTask(
        "create-index [" + request.index() + "], cause [" + request.cause() + "]",
        new AckedClusterStateUpdateTask(Priority.URGENT, request, listener) {

            @Override
            public ClusterState execute(ClusterState currentState) throws Exception {
                // 分配分片产生新版本的元数据
                return applyCreateIndexRequest(currentState, request, false);
            }

            @Override
            public void onFailure(Exception e) {
                ......
            }
        }
    );
}

public void submitTask(BatchedTask task, @Nullable TimeValue timeout) throws EsRejectedExecutionException {
    tasksPerBatchingKey.compute(task.batchingKey, (k, existingTasks) -> {
        if (existingTasks == null) {
            existingTasks = Collections.synchronizedSet(new LinkedHashSet<>());
        } else {
            assert assertNoDuplicateTasks(task, existingTasks);
        }
        // 按 executor 汇总 task
        existingTasks.add(task);
        return existingTasks;
    });

    if (timeout != null) {
        // 指定超时时间，不指定默认元数据变更 30s 超时
        threadExecutor.execute(task, timeout, () -> onTimeoutInternal(task, timeout));
    } else {
        threadExecutor.execute(task);
    }
}

protected PrioritizedEsThreadPoolExecutor createThreadPoolExecutor() {
    return EsExecutors.newSinglePrioritizing(
        // masterService#updateTask，我们经常在 jstack 堆栈中看到的 master 节点元数据变更线程
        nodeName + "/" + MASTER_UPDATE_THREAD_NAME,
        daemonThreadFactory(nodeName, MASTER_UPDATE_THREAD_NAME),
        threadPool.getThreadContext(),
        threadPool.scheduler(),
        ......
    );
}

public enum Priority {
    IMMEDIATE((byte) 0), // 节点加入、脱离等
    URGENT((byte) 1), // 索引、template 创建等
    HIGH((byte) 2),
    NORMAL((byte) 3),
    LOW((byte) 4),
    LANGUID((byte) 5);
}

@Override
public void publish(
    ClusterStatePublicationEvent clusterStatePublicationEvent,
    ActionListener<Void> publishListener,
    AckListener ackListener
) {
    try {
        synchronized (mutex) {
           ......
            try {
                ......
                // 获取目标发布节点
                final DiscoveryNodes publishNodes = publishRequest.getAcceptedState().nodes();
                // leader checker 在数据节点上用于追踪 master，和 master 保持心跳
                leaderChecker.setCurrentNodes(publishNodes);
                // 设置当前 leader 需要追踪的 follower 节点，主要是用于心跳保持
                followersChecker.setCurrentNodes(publishNodes);
                // lag 探测器，如果某个节点超过指定时间默认 90s 没有 publish 成功则踢出
                lagDetector.setTrackedNodes(publishNodes);
                // 启动 publish 任务
                publication.start(followersChecker.getFaultyNodes());
            } finally {
                publicationContext.decRef();
            }
        }
    } catch (Exception e) {
       ......
    }
}

public void start(Set<DiscoveryNode> faultyNodes) {
    logger.trace("publishing {} to {}", publishRequest, publicationTargets);
    publicationTargets.forEach(PublicationTarget::sendPublishRequest);
}
void sendPublishRequest() {
    Publication.this.sendPublishRequest(discoveryNode, publishRequest, new PublishResponseHandler());
}

private PublishWithJoinResponse handleIncomingPublishRequest(BytesTransportRequest request) throws IOException {
    StreamInput in = request.bytes().streamInput();
    try {
        ......
        // 全量或增量标记
        if (in.readBoolean()) {
            final ClusterState incomingState;
            ......
            final PublishWithJoinResponse response = acceptState(incomingState);
            // 这里 lastSeenClusterState 的目的就是为了下次接收增量之后应用 diff
            lastSeenClusterState.set(incomingState);
            return response;
        } else {
            final ClusterState lastSeen = lastSeenClusterState.get();
            if (lastSeen == null) {
                throw new IncompatibleClusterStateVersionException("have no local cluster state");
            } else {
                // 基于 lastSeen 应用增量 diff 产生新的完整的 incomingState
                ClusterState incomingState;
                try {
                    final Diff<ClusterState> diff;
                    // Close stream early to release resources used by the de-compression as early as possible
                    try (StreamInput input = in) {
                        diff = ClusterState.readDiffFrom(input, lastSeen.nodes().getLocalNode());
                    }
                    incomingState = diff.apply(lastSeen); // might throw IncompatibleClusterStateVersionException
                } catch (Exception e) {
                    throw e;
                }
                ......
                final PublishWithJoinResponse response = acceptState(incomingState);
                // 本地变量 CAS 替换成全局
                lastSeenClusterState.compareAndSet(lastSeen, incomingState);
                return response;
            }
        }
    } finally {
        IOUtils.close(in);
    }
}

public PublishResponse handlePublishRequest(PublishRequest publishRequest) {
    final ClusterState clusterState = publishRequest.getAcceptedState();
    if (clusterState.term() != getCurrentTerm()) {
       ......
    }
    if (clusterState.term() == getLastAcceptedTerm() && clusterState.version() <= getLastAcceptedVersion()) {
       ......
    }
    // 更新 lastAcceptedState
    persistedState.setLastAcceptedState(clusterState);
    return new PublishResponse(clusterState.term(), clusterState.version());
}

public Optional<ApplyCommitRequest> handlePublishResponse(DiscoveryNode sourceNode, PublishResponse publishResponse) {
    ......
    publishVotes.addVote(sourceNode);
    if (isPublishQuorum(publishVotes)) {
        return Optional.of(new ApplyCommitRequest(localNode, publishResponse.getTerm(), publishResponse.getVersion()));
    }
    return Optional.empty();
}

@Override
protected void sendApplyCommit(
    DiscoveryNode destination,
    ApplyCommitRequest applyCommit,
    ActionListener<Empty> responseActionListener
) {
    transportService.sendRequest(
        destination,
        COMMIT_STATE_ACTION_NAME,
        applyCommit,
        COMMIT_STATE_REQUEST_OPTIONS,
        new ActionListenerResponseHandler<>(wrapWithMutex(responseActionListener), in -> Empty.INSTANCE, Names.CLUSTER_COORDINATION)
    );
}

private void handleApplyCommit(ApplyCommitRequest applyCommitRequest, ActionListener<Void> applyListener) {
    synchronized (mutex) {
        // 将我们之前接收到的 acceptedState 的元数据改为 commit 状态
        coordinationState.get().handleCommit(applyCommitRequest);
        ......
        if (applyCommitRequest.getSourceNode().equals(getLocalNode())) {
            // master node applies the committed state at the end of the publication process, not here.
            applyListener.onResponse(null);
        } else {
            // 元数据接收完毕，进入应用环节
            clusterApplier.onNewClusterState(applyCommitRequest.toString(), () -> applierState, applyListener.map(r -> {
                onClusterStateApplied();
                return r;
            }));
        }
    }
}

private void applyChanges(ClusterState previousClusterState, ClusterState newClusterState, String source, Recorder stopWatch) {
    ......
    logger.debug("apply cluster state with version {}", newClusterState.version());
    // 进行元数据应用，会按优先级分为 low/normal/high 来应用，例如创建索引属于 high
    callClusterStateAppliers(clusterChangedEvent, stopWatch);

    logger.debug("set locally applied cluster state to version {}", newClusterState.version());
    // 更新本地最新 commited 的 clusterState
    state.set(newClusterState);

    // 这里一般是 master 节点发起元数据 commit 结束后，再回调相应的 listener
    callClusterStateListeners(clusterChangedEvent, stopWatch);
}

@Override
public synchronized void applyClusterState(final ClusterChangedEvent event) {
    // 最新 commit 的元数据
    final ClusterState state = event.state();
    ......
    // 删除索引、分片，清理磁盘分片目录
    deleteIndices(event); // also deletes shards of deleted indices

    // 删除索引、分片，只是清理内存对象，主要是针对 Close/Open 操作
    removeIndicesAndShards(event); // also removes shards of removed indices

    // 更新索引 settings、mapping 等
    updateIndices(event); // can also fail shards, but these are then guaranteed to be in failedShardsCache

    // 创建索引、分片
    createIndicesAndUpdateShards(state);
}

private void createIndicesAndUpdateShards(final ClusterState state) {
    DiscoveryNodes nodes = state.nodes();
    // 通过节点到分片的映射，获取属于该节点的索引分片信息
    RoutingNode localRoutingNode = state.getRoutingNodes().node(nodes.getLocalNodeId());
  
    // 对比接收到的元数据和本节点内存的索引列表，找出新增的需要创建的索引
    final Map<Index, List<ShardRouting>> indicesToCreate = new HashMap<>();
    for (ShardRouting shardRouting : localRoutingNode) {
        ShardId shardId = shardRouting.shardId();
        if (failedShardsCache.containsKey(shardId) == false) {
            final Index index = shardRouting.index();
            final var indexService = indicesService.indexService(index);
            if (indexService == null) {
                // 不存在的索引就是需要创建的
                indicesToCreate.computeIfAbsent(index, k -> new ArrayList<>()).add(shardRouting);
            } else {
                // 索引存在看看有没需要创建或更新的分片
                createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
            }
        }
    }

    // 过滤出来的待创建索引列表，遍历依次创建
    for (Map.Entry<Index, List<ShardRouting>> entry : indicesToCreate.entrySet()) {
        final Index index = entry.getKey();
        final IndexMetadata indexMetadata = state.metadata().index(index);
        logger.debug("[{}] creating index", index);

        try {
            // 创建索引
            indicesService.createIndex(indexMetadata, buildInIndexListener, true);
        } catch (Exception e) {
            ......
        }
        // 创建索引对应的本地分片
        for (ShardRouting shardRouting : entry.getValue()) {
            createOrUpdateShard(state, nodes, routingTable, shardRouting, indexService);
        }
    }
}