《Elasticsearch 源码解析与优化实战》第12章：allocation模型分析

1.index 增删
2.node 增删
3.replica数量改变
4.手工 reroute
5.集群重启

public static Collection<AllocationDecider> createAllocationDeciders(...) {
    Map<Class, AllocationDecider> deciders = new LinkedHashMap<>();
    addAllocationDecider (deciders, new MaxRetryAllocationDecider(settings));
    addAllocationDecider (deciders, new ResizeAllocationDecider(settings));
    addAllocationDecider (deciders, new ReplicaAfterPrimaryActiveAllocation-Decider(settings));
    addAllocationDecider (deciders, new RebalanceOnlyWhenActiveAllocation-Decider(settings));
    addAllocationDecider (deciders, new ClusterRebalanceAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new ConcurrentRebalanceAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new EnableAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new NodeVersionAllocationDecider(settings));
    addAllocationDecider (deciders, new SnapshotInProgressAllocationDecider(settings));
    addAllocationDecider (deciders, new RestoreInProgressAllocationDecider(settings));
    addAllocationDecider (deciders, new FilterAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new SameShardAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new DiskThresholdDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new ThrottlingAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new ShardsLimitAllocationDecider(settings, clusterSettings));
    addAllocationDecider (deciders, new AwarenessAllocationDecider(settings, clusterSettings));
    
    return deciders.values();
}

ALWAYS、YES、NO、THROTTLE

SameShardAllocationDecider
避免主副分片分配到同一个节点

AwarenessAllocationDecider
感知分配器，感知服务器、机架等，尽量分散存储shard
有两种参数用于调整:
    cluster.routing.allocation.awareness.attributes: rack_id
    cluster.routing.allocation.awareness.attributes: zone

ShardsLimitAllocationDecider
同一个节点上允许存在的同一个index的shard数目

ThrottlingAllocationDecider
recovery阶段的限速配置，包括: 
    cluster.routing.allocation.node_concurrent_recoveries
    cluster.routing.allocation.node_initial_primaries_recoveries
    cluster.routing.allocation.node_concurrent_incoming_recoveries
    cluster.routing.allocation.node_concurrent_outgoing_recoveries

ConcurrentRebalanceAllocationDecider
rebalance并发控制，可以通过下面的参数配置:
    cluster.routing.allocation.cluster_concurrent_rebalance

DiskThresholdDecider
根据磁盘空间进行决策的分配器

RebalanceOnlyWhenActiveAllocationDecider
所有shard都处在active状态下，才可以执行rebalance操作

FilterAllocationDecider
可以调整的参数如下，可以通过接口动态设置:
    index.routing.allocation.require.* [ 必须] ]
    index.routing.allocation.include.* [ 允许 ]
    index.routing.allocation.exclude.* [ 排除 ]
    cluster.routing.allocation.require.*
    cluster.routing.allocation.include.*
    cluster.routing.allocation.exclude.*
    配置的目标为节点IP或节点名等。cluster 级别设置会覆盖index级别设置。

ReplicaAfterPrimaryActiveAllocationDecider
保证只会在主分片分配完毕后才开始分配分片副本。

ClusterRebalanceAllocationDecider
通过集群中active的shard状态来决定是否可以执行rebalance，通过下面的配置控制，可以动态生效:
    cluster.routing.allocation.allow_rebalance

private void reroute (RoutingAllocation allocation) {
    if(allocation.routingNodes().unassigned().size() > 0) {
        removeDelayMarkers (allocation) ;
        //gateway分配器
        gatewayAllocator.allocateUnassigned (allocation);
    }
    //分片均衡分配器
    shardsAllocator.allocate(allocation);
}

primaryShardAllocator.allocateUnassigned (allocation) ;
replicaShardAllocator.processExistingRecoveries (allocation) ;
replicaShardAllocator.allocateUnassigned (allocation) ;

public void allocateUnassigned (RoutingAllocation allocation) {
    //遍历未分配的分片
    while (unassignedIterator.hasNext() ) {
        final ShardRouting shard = unassignedIterator.next() ;
        //调用主分片分配器的makeAllocationDecision进行决策
        final AllocateUnassignedDecision allocateUnassignedDecision = makeAllocationDecision (shard, allocation，logger);
        //根据决策器结果决定是否初始化分片
        if (allocateUnas signedDecision.getAllocationDecision() == AllocationDecision.YES) {
            unassignedIterator. initialize(...);
        } else {
            unassignedIterator . removeAndIgnore(. ..) ;
        }
    }
}

public ShardRouting initializeShard(. ..) {
    //初始化一个ShardRouting
    ShardRouting initializedShard = unassignedShard.initialize(nodeId, existingAllocationId, expectedSize);
    //添加到目的节点的shard列表
    node (nodeId).add(initializedShard);
    addRecovery (initializedShard);
    //添加到assignedshards列表
    assignedShardsAdd (initial i zedShard);
    //设置状态已更新
    routingChangesObserver.shardInitialized (unassignedShard, initializedShard);
    return initializedShard;
}

protected ClusterState reroute (final ClusterState clusterState, String reason, boolean debug) {
    reroute (allocation);
    if (allocation.routingNodesChanged() == false) {
        return clusterState;
    }
    return buildResultAndLogHealthChange(clusterState, allocation, reason);
}

void asyncFetch (final DiscoveryNode[] nodes， long fetchingRound) {
    //nodes为节点列表
    action.list(shardId, nodes, new ActionListener<BaseNodesResponse<T>>() {
        //收到成功的回复
        public void onResponse (BaseNodesResponse<T> response) {
            processAsyncFetch (response .getNodes()，response. failures() , fetchingRound);
        }
        //收到失败的回复
        public void onFailure (Exception e) {
            List<FailedNodeException> failures = new ArrayList<>(nodes.length);
            for (final Di scoveryNode node: nodes) {
                failures.add(. ..);
            }
            processAsyncFetch(null, failures, fetchingRound);
        }
    });
}

internal:gateway/local/started_ shards

TransportNodesAction.AsyncAction#start

gateway.TransportNodesListGatewayStartedShards#nodeOperation

AsyncShardFetch#processAsyncFetch

reroute(shardId, "post_ response");
//接着调用-->
routingService.reroute ("async_ shard_ fetch");

//根据shard id从集群状态获取同步ID列表
final Set<String> inSyncAllocationIds = indexMetaData.inSyncAllocationIds(unassignedShard.id());
//构建匹配同步ID的目标节点列表
final NodeShardsResult nodeShardsResult = buildNodeShardsResult(unassignedShard, snapshotRestore, allocation.getIgnoreNodes(unassignedShard.shardId()), inSyncAllocationIds, shardState, logger);
//对上一步得到的列表中的节点依次执行decider,决策结果
NodesToAllocate nodesToAllocate = buildNodesToAllocate(allocation, nodeShardsResult.orderedAllocationCandidates, unassignedShard, false);

public Decision canAllocate (ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
    //遍历全部decider
    for(AllocationDecider allocationDecider : allocations) {
        Decision decision = allocationDecider.canAllocate(shardRouting, node, allocation);
        if (decision == Decision.NO) {
            if (!allocation. debugDecision()) {
                return decision;
            } else {
                ret.add (decision);
        } else if (decision != Decision.ALWAYS && (allocation.getDebugMode() != EXCLUDE_ YES_ DECISIONS || decision.type() != Decision.Type.YES)) {
            ret.add (decision);
        }
    return ret;
}

public Decision canForceAllocatePrimary (ShardRouting shardRouting, RoutingNode node, RoutingAl location allocation) {
    Decision decision = canAllocate (shardRouting, node, allocation);
    if (decision.type() == Type.NO) {
        // On a NO decision, by default, we allow force allocating the primary.
        return allocation. decision (Decision.YES, decision.label(),
            "primary shard [%s] allowed to force allocate on node [%s]",
            shardRouting. shardId() , node.nodeId());
    } else {
        // On a THROTTLE/ YES decision, we use the same decision instead of forcing allocation
        return decision;
    }
}

ReplicaShardAllocator#makeAllocationDecision

index.unassigned.node_left.delayed_timeout

cluster.routing.allocation.balance.index 
cluster.routing.allocation.balance.shard 
cluster.routing.allocation.balance.threshold

public void allocate (RoutingAllocation allocation) {
    final Balancer balancer = new Balancer (logger, allocation, weightFunction, threshold);
    balancer.allocateUnassigned();
    balancer.moveShards();
    balancer.balance();
}

private void createShard (DiscoveryNodes nodes, RoutingTable routingTable, ShardRouting shardRouting, ClusterState state) {
    if (shardRouting.recoverySource().getType() == Type.PEER) {
        sourceNode = findSourceNodeForPeerRecovery (logger, routingTable, nodes, shardRouting);
        try {
            RecoveryState recoveryState = new RecoveryState(shardRouting, nodes.getLocalNode(), sourceNode);
            indicesService. createShard(...);
        } catch (Exception e) {
            failAndRemoveShard ( shardRouting, true, "failed to create shard", e, state);
        }
    }
}

public IndexShard createShard(. . .) throws IOException {
    IndexService indexService = indexService(shardRouting. index());
    IndexShard indexShard = indexService.createShard (shardRouting, globalCheckpointSyncer);
    indexShard.addShardFailureCallback(onShardFailure);
    indexShard.startRecover (recoveryState, recoveryTargetService, recoveryListener, repositoriesService,
        (type, mapping) -> {
            try (
                //concrete index - no name clash, it uses uuid
                client.admin().indices().preparePutMapping().setConcreteIndex(shardRouting.index()) 
                    .setType (type)
                    .setSource(mapping.source().string(), XContentType.JSON)
                    .get () ;
            } catch ( IOException ex) {
                throw new ElasticsearchException("failed to stringify mapping source", exx);
        ), this);
    return indexShard;
}