Elasticsearch源码分析六之AllocationDeciders

 public static Collection<AllocationDecider> createAllocationDeciders(Settings settings, ClusterSettings clusterSettings,                                                                         List<ClusterPlugin> clusterPlugins) {        // collect deciders by class so that we can detect duplicates        Map<Class, AllocationDecider> deciders = new LinkedHashMap<>();        addAllocationDecider(deciders, new MaxRetryAllocationDecider());        addAllocationDecider(deciders, new ResizeAllocationDecider());        addAllocationDecider(deciders, new ReplicaAfterPrimaryActiveAllocationDecider());        addAllocationDecider(deciders, new RebalanceOnlyWhenActiveAllocationDecider());        addAllocationDecider(deciders, new ClusterRebalanceAllocationDecider(settings, clusterSettings));        addAllocationDecider(deciders, new ConcurrentRebalanceAllocationDecider(settings, clusterSettings));        addAllocationDecider(deciders, new EnableAllocationDecider(settings, clusterSettings));        addAllocationDecider(deciders, new NodeVersionAllocationDecider());        addAllocationDecider(deciders, new SnapshotInProgressAllocationDecider());        addAllocationDecider(deciders, new RestoreInProgressAllocationDecider());        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));
        clusterPlugins.stream()            .flatMap(p -> p.createAllocationDeciders(settings, clusterSettings).stream())            .forEach(d -> addAllocationDecider(deciders, d));
        return deciders.values();    }

/** Add the given allocation decider to the given deciders collection, erroring if the class name is already used. */    private static void addAllocationDecider(Map<Class, AllocationDecider> deciders, AllocationDecider decider) {        if (deciders.put(decider.getClass(), decider) != null) {            throw new IllegalArgumentException("Cannot specify allocation decider [" + decider.getClass().getName() + "] twice");        }    }

   public final void execute(Task task, Request request, ActionListener<Response> listener) {        ActionRequestValidationException validationException = request.validate();        if (validationException != null) {            listener.onFailure(validationException);            return;        }
        if (task != null && request.getShouldStoreResult()) {            listener = new TaskResultStoringActionListener<>(taskManager, task, listener);        }
        RequestFilterChain<Request, Response> requestFilterChain = new RequestFilterChain<>(this, logger);        requestFilterChain.proceed(task, actionName, request, listener);    }

@Override        public void proceed(Task task, String actionName, Request request, ActionListener<Response> listener) {            int i = index.getAndIncrement();            try {                if (i < this.action.filters.length) {                    this.action.filters[i].apply(task, actionName, request, listener, this);                } else if (i == this.action.filters.length) {                    this.action.doExecute(task, request, listener);                } else {                    listener.onFailure(new IllegalStateException("proceed was called too many times"));                }            } catch(Exception e) {                logger.trace("Error during transport action execution.", e);                listener.onFailure(e);            }        }

    @Override    protected void doExecute(Task task, final Request request, ActionListener<Response> listener) {        new AsyncSingleAction(task, request, listener).start();    }

public void start() {            ClusterState state = clusterService.state();            this.observer                = new ClusterStateObserver(state, clusterService, request.masterNodeTimeout(), logger, threadPool.getThreadContext());            doStart(state);        }

  protected void masterOperation(Task task, Request request, ClusterState state, ActionListener<Response> listener) throws Exception {        masterOperation(request, state, listener);    }

 @Override    protected void masterOperation(final ClusterAllocationExplainRequest request, final ClusterState state,                                   final ActionListener<ClusterAllocationExplainResponse> listener) {        final RoutingNodes routingNodes = state.getRoutingNodes();        final ClusterInfo clusterInfo = clusterInfoService.getClusterInfo();        final RoutingAllocation allocation = new RoutingAllocation(allocationDeciders, routingNodes, state,                clusterInfo, System.nanoTime());
        ShardRouting shardRouting = findShardToExplain(request, allocation);        logger.debug("explaining the allocation for [{}], found shard [{}]", request, shardRouting);
        ClusterAllocationExplanation cae = explainShard(shardRouting, allocation,            request.includeDiskInfo() ? clusterInfo : null, request.includeYesDecisions(), gatewayAllocator, shardAllocator);        listener.onResponse(new ClusterAllocationExplainResponse(cae));    }

@Override    public ShardAllocationDecision decideShardAllocation(final ShardRouting shard, final RoutingAllocation allocation) {        Balancer balancer = new Balancer(logger, allocation, weightFunction, threshold);        AllocateUnassignedDecision allocateUnassignedDecision = AllocateUnassignedDecision.NOT_TAKEN;        MoveDecision moveDecision = MoveDecision.NOT_TAKEN;        if (shard.unassigned()) {            allocateUnassignedDecision = balancer.decideAllocateUnassigned(shard, Sets.newHashSet());        } else {            moveDecision = balancer.decideMove(shard);            if (moveDecision.isDecisionTaken() && moveDecision.canRemain()) {                MoveDecision rebalanceDecision = balancer.decideRebalance(shard);                moveDecision = rebalanceDecision.withRemainDecision(moveDecision.getCanRemainDecision());            }        }        return new ShardAllocationDecision(allocateUnassignedDecision, moveDecision);    }

 @Override    public Decision canRebalance(ShardRouting shardRouting, RoutingAllocation allocation) {        Decision.Multi ret = new Decision.Multi();        for (AllocationDecider allocationDecider : allocations) {            Decision decision = allocationDecider.canRebalance(shardRouting, allocation);            // short track if a NO is returned.            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 static final String NAME = "cluster_rebalance";    private static final String CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE = "cluster.routing.allocation.allow_rebalance";    public static final Setting<ClusterRebalanceType> CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE_SETTING =        new Setting<>(CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE, ClusterRebalanceType.INDICES_ALL_ACTIVE.toString(),            ClusterRebalanceType::parseString, Property.Dynamic, Property.NodeScope);

 public enum ClusterRebalanceType {        /**         * Re-balancing is allowed once a shard replication group is active         */        ALWAYS,        /**         * Re-balancing is allowed only once all primary shards on all indices are active.         */        INDICES_PRIMARIES_ACTIVE,        /**         * Re-balancing is allowed only once all shards on all indices are active.         */        INDICES_ALL_ACTIVE;

public ClusterRebalanceAllocationDecider(Settings settings, ClusterSettings clusterSettings) {        type = CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE_SETTING.get(settings);        logger.debug("using [{}] with [{}]", CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE, type);        clusterSettings.addSettingsUpdateConsumer(CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE_SETTING, this::setType);    }

 private final List<SettingUpdater<?>> settingUpdaters = new CopyOnWriteArrayList<>();

private void setType(ClusterRebalanceType type) {        this.type = type;    }

 @Override    public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {        // 节点为主节点而且没有分配过        if (shardRouting.primary() && shardRouting.unassigned()) {            // 进行节点的初始化            assert initializingShard(shardRouting, node.nodeId()).recoverySource().getType() != RecoverySource.Type.PEER;            // primary is unassigned, means we are going to do recovery from store, snapshot or local shards            // count *just the primaries* currently doing recovery on the node and check against primariesInitialRecoveries            // 统计在进行recovery的主节点数            int primariesInRecovery = 0;            // 遍历当前节点中的分片数            for (ShardRouting shard : node) {                // when a primary shard is INITIALIZING, it can be because of *initial recovery* or *relocation from another node*                // we only count initial recoveries here, so we need to make sure that relocating node is null                // 一个主节点的初始化，可能是因为initial recovery参数或者relocation from another node                // 我们仅仅在这里统计在进行recovery的主分片，所以我们需要确保正在进行relocating的节点为null                if (shard.initializing() && shard.primary() && shard.relocatingNodeId() == null) {                    primariesInRecovery++;                }            }            // 当处于恢复中的主分片数量大于初始化加载的cluster.routing.allocation.node_initial_primaries_recoveries配置的数量时            if (primariesInRecovery >= primariesInitialRecoveries) {                // TODO: Should index creation not be throttled for primary shards?                // 相当于达到阀值了                return allocation.decision(THROTTLE, NAME,                    "reached the limit of ongoing initial primary recoveries [%d], cluster setting [%s=%d]",                    primariesInRecovery, CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES_SETTING.getKey(),                    primariesInitialRecoveries);            } else {                return allocation.decision(YES, NAME, "below primary recovery limit of [%d]", primariesInitialRecoveries);            }        } else {            // Peer recovery            assert initializingShard(shardRouting, node.nodeId()).recoverySource().getType() == RecoverySource.Type.PEER;
            // Allocating a shard to this node will increase the incoming recoveries            // 当前节点每多分配一个分片就会增加incoming recoveries的数量            int currentInRecoveries = allocation.routingNodes().getIncomingRecoveries(node.nodeId());            // 当前正在恢复中的节点数量大于当前配置的限制值时            if (currentInRecoveries >= concurrentIncomingRecoveries) {                return allocation.decision(THROTTLE, NAME,                    "reached the limit of incoming shard recoveries [%d], cluster setting [%s=%d] (can also be set via [%s])",                    currentInRecoveries, CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_INCOMING_RECOVERIES_SETTING.getKey(),                    concurrentIncomingRecoveries,                    CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES_SETTING.getKey());            } else {                // search for corresponding recovery source (= primary shard) and check number of outgoing recoveries on that node                // 搜索相应的恢复源(也就是主节点) 并且检查该节点上传出的恢复数                ShardRouting primaryShard = allocation.routingNodes().activePrimary(shardRouting.shardId());                if (primaryShard == null) {                    return allocation.decision(Decision.NO, NAME, "primary shard for this replica is not yet active");                }                int primaryNodeOutRecoveries = allocation.routingNodes().getOutgoingRecoveries(primaryShard.currentNodeId());                if (primaryNodeOutRecoveries >= concurrentOutgoingRecoveries) {                    return allocation.decision(THROTTLE, NAME,                        "reached the limit of outgoing shard recoveries [%d] on the node [%s] which holds the primary, " +                        "cluster setting [%s=%d] (can also be set via [%s])",                        primaryNodeOutRecoveries, primaryShard.currentNodeId(),                        CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_OUTGOING_RECOVERIES_SETTING.getKey(),                        concurrentOutgoingRecoveries,                        CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES_SETTING.getKey());                } else {                    return allocation.decision(YES, NAME, "below shard recovery limit of outgoing: [%d < %d] incoming: [%d < %d]",                        primaryNodeOutRecoveries,                        concurrentOutgoingRecoveries,                        currentInRecoveries,                        concurrentIncomingRecoveries);                }            }        }    }

curl -XPUT localhost:9200/test/_settings -d '{    "index.routing.allocation.include.tag" : "value1,value2"}'  

curl -XPUT localhost:9200/test/_settings -d '{    "index.routing.allocation.exclude.tag" : "value3"}'  

node.group1: group1_value1node.group2: group2_value4

curl -XPUT localhost:9200/test/_settings -d '{    "index.routing.allocation.include.group1" : "xxx"    "index.routing.allocation.include.group2" : "yyy",    "index.routing.allocation.exclude.group3" : "zzz",}'  

curl -XPUT localhost:9200/_cluster/settings -d '{    "transient" : {        "cluster.routing.allocation.exclude._ip" : "10.0.0.1"    }}'  

 /**     * 是否开启磁盘分配域值校验     */    public static final Setting<Boolean> CLUSTER_ROUTING_ALLOCATION_DISK_THRESHOLD_ENABLED_SETTING =        Setting.boolSetting("cluster.routing.allocation.disk.threshold_enabled", true,            Setting.Property.Dynamic, Setting.Property.NodeScope);    /**     * 磁盘分配的低水平线（下限）     */    public static final Setting<String> CLUSTER_ROUTING_ALLOCATION_LOW_DISK_WATERMARK_SETTING =        new Setting<>("cluster.routing.allocation.disk.watermark.low", "85%",            (s) -> validWatermarkSetting(s, "cluster.routing.allocation.disk.watermark.low"),            new LowDiskWatermarkValidator(),            Setting.Property.Dynamic, Setting.Property.NodeScope);    /**     * 磁盘分配的高水平线（上限）     */    public static final Setting<String> CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING =        new Setting<>("cluster.routing.allocation.disk.watermark.high", "90%",            (s) -> validWatermarkSetting(s, "cluster.routing.allocation.disk.watermark.high"),            new HighDiskWatermarkValidator(),            Setting.Property.Dynamic, Setting.Property.NodeScope);    /**     * 比高水平线还高一些的值     */    public static final Setting<String> CLUSTER_ROUTING_ALLOCATION_DISK_FLOOD_STAGE_WATERMARK_SETTING =        new Setting<>("cluster.routing.allocation.disk.watermark.flood_stage", "95%",            (s) -> validWatermarkSetting(s, "cluster.routing.allocation.disk.watermark.flood_stage"),            new FloodStageValidator(),            Setting.Property.Dynamic, Setting.Property.NodeScope);    /**     * 磁盘不足时是否迁移     */    public static final Setting<Boolean> CLUSTER_ROUTING_ALLOCATION_INCLUDE_RELOCATIONS_SETTING =        Setting.boolSetting("cluster.routing.allocation.disk.include_relocations", true,            Setting.Property.Dynamic, Setting.Property.NodeScope, Setting.Property.Deprecated);    /**     * 磁盘没法分配到重新磁盘分配路由的时间间隔     */    public static final Setting<TimeValue> CLUSTER_ROUTING_ALLOCATION_REROUTE_INTERVAL_SETTING =        Setting.positiveTimeSetting("cluster.routing.allocation.disk.reroute_interval", TimeValue.timeValueSeconds(60),            Setting.Property.Dynamic, Setting.Property.NodeScope);

/**     * Controls the maximum number of shards per index on a single Elasticsearch     * node. Negative values are interpreted as unlimited.     * index级别的”index.routing.allocation.total_shards_per_node”，表示这个index每个node的总共允许存在多少个shard，默认值是-1表示无穷多个     */    public static final Setting<Integer> INDEX_TOTAL_SHARDS_PER_NODE_SETTING =        Setting.intSetting("index.routing.allocation.total_shards_per_node", -1, -1,            Property.Dynamic, Property.IndexScope);
    /**     * Controls the maximum number of shards per node on a global level.     * Negative values are interpreted as unlimited.     * cluster级别”cluster.routing.allocation.total_shards_per_node”，表示集群范围内每个Node允许存在有多少个shard。默认值是-1表示无穷多个     */    public static final Setting<Integer> CLUSTER_TOTAL_SHARDS_PER_NODE_SETTING =        Setting.intSetting("cluster.routing.allocation.total_shards_per_node", -1,  -1,            Property.Dynamic, Property.NodeScope);

node.rack_id: rack_one

cluster.routing.allocation.awareness.attributes: rack_id

cluster.routing.allocation.awareness.attributes: rack_id,zone

1. cluster.routing.allocation.awareness.force.zone.values: zone1,zone22. cluster.routing.allocation.awareness.attributes: zone