ES分片均衡策略分析与改进

Elasticsearch 的分片均衡是指将索引的分片均匀地分布在集群中的各个节点上，以实现负载均衡和高可用性。在 Elasticsearch 中，索引被分为多个分片，每个分片可以在不同的节点上进行存储和处理。分片均衡的目标是确保每个节点上的分片数量大致相等，避免某些节点负载过重，而其他节点负载较轻的情况。

  WeightFunction(float indexBalance, float shardBalance) {
    float sum = indexBalance + shardBalance;
    if (sum <= 0.0f) {
        throw new IllegalArgumentException("Balance factors must sum to a value > 0 but was: " + sum);
    }
    theta0 = shardBalance / sum;
    theta1 = indexBalance / sum;
    this.indexBalance = indexBalance;
    this.shardBalance = shardBalance;
  }

  float weight(Balancer balancer, ModelNode node, String index) {
    final float weightShard = node.numShards() - balancer.avgShardsPerNode();
    final float weightIndex = node.numShards(index) - balancer.avgShardsPerNode(index);
    return theta0 * weightShard + theta1 * weightIndex;
  }

  // get the delta between the weights of the node we are checking and the node that holds the shard
  float currentDelta = absDelta(nodeWeight, currentWeight);
  // checks if the weight delta is above a certain threshold; if it is not above a certain threshold,
  // then even though the node we are examining has a better weight and may make the cluster balance
  // more even, it doesn't make sense to execute the heavyweight operation of relocating a shard unless
  // the gains make it worth it, as defined by the threshold
  boolean deltaAboveThreshold = lessThan(currentDelta, threshold) == false;

  public Decision canRebalance(ShardRouting shardRouting, RoutingAllocation allocation) {
    if (allocation.ignoreDisable()) {
      return allocation.decision(Decision.YES, NAME, "allocation is explicitly ignoring any disabling of rebalancing");
    }

    Settings indexSettings = allocation.metadata().getIndexSafe(shardRouting.index()).getSettings();
    final Rebalance enable;
    final boolean usedIndexSetting;
    if (INDEX_ROUTING_REBALANCE_ENABLE_SETTING.exists(indexSettings)) {
      enable = INDEX_ROUTING_REBALANCE_ENABLE_SETTING.get(indexSettings);
      usedIndexSetting = true;
    } else {
      enable = this.enableRebalance;
      usedIndexSetting = false;
    }
    switch (enable) {
      case ALL:
        return allocation.decision(Decision.YES, NAME, "all rebalancing is allowed");
      case NONE:
        return allocation.decision(Decision.NO, NAME, "no rebalancing is allowed due to %s", setting(enable, usedIndexSetting));
      case PRIMARIES:
        if (shardRouting.primary()) {
          return allocation.decision(Decision.YES, NAME, "primary rebalancing is allowed");
        } else {
          return allocation.decision(
            Decision.NO,
            NAME,
            "replica rebalancing is forbidden due to %s",
            setting(enable, usedIndexSetting)
          );
        }
      case REPLICAS:
        if (shardRouting.primary() == false) {
          return allocation.decision(Decision.YES, NAME, "replica rebalancing is allowed");
        } else {
          return allocation.decision(
            Decision.NO,
            NAME,
            "primary rebalancing is forbidden due to %s",
            setting(enable, usedIndexSetting)
          );
        }
      default:
        throw new IllegalStateException("Unknown rebalance option");
    }
  }

  public Decision canRebalance(RoutingAllocation allocation) {
    final RoutingNodes routingNodes = allocation.routingNodes();
    switch (type) {
      case INDICES_PRIMARIES_ACTIVE:
        // check if there are unassigned primaries.
        if (routingNodes.hasUnassignedPrimaries()) {
          return NO_UNASSIGNED_PRIMARIES;
        }
        // check if there are initializing primaries that don't have a relocatingNodeId entry.
        if (routingNodes.hasInactivePrimaries()) {
          return NO_INACTIVE_PRIMARIES;
        }
        return YES_ALL_PRIMARIES_ACTIVE;
      case INDICES_ALL_ACTIVE:
        // check if there are unassigned shards.
        if (routingNodes.hasUnassignedShards()) {
          return NO_UNASSIGNED_SHARDS;
        }
        // in case all indices are assigned, are there initializing shards which
        // are not relocating?
        if (routingNodes.hasInactiveShards()) {
          return NO_INACTIVE_SHARDS;
        }
        // fall-through
      default:
        // all shards active from above or type == Type.ALWAYS
        return YES_ALL_SHARDS_ACTIVE;
    }
  }

  public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {

    ... ...

    if (freeDiskPercentage < diskThresholdSettings.getFreeDiskThresholdLow()) {
      // If the shard is a replica or is a non-empty primary, check the low threshold
      if (skipLowThresholdChecks == false) {
        if (logger.isDebugEnabled()) {
          logger.debug(
            "more than the allowed {} used disk threshold ({} used) on node [{}], preventing allocation",
            Strings.format1Decimals(usedDiskThresholdLow, "%"),
            Strings.format1Decimals(usedDiskPercentage, "%"),
            node.nodeId()
          );
        }
        return allocation.decision(
          Decision.NO,
          NAME,
          "the node is above the low watermark cluster setting [%s=%s], using more disk space than the maximum allowed [%s%%], actual free: [%s%%]",
          CLUSTER_ROUTING_ALLOCATION_LOW_DISK_WATERMARK_SETTING.getKey(),
          diskThresholdSettings.getLowWatermarkRaw(),
          usedDiskThresholdLow,
          freeDiskPercentage
        );
      }
    }

   ... ...

    if (freeBytes < diskThresholdSettings.getFreeBytesThresholdHigh().getBytes()) {
      if (logger.isDebugEnabled()) {
        logger.debug(
          "less than the required {} free bytes threshold ({} bytes free) on node {}, shard cannot remain",
          diskThresholdSettings.getFreeBytesThresholdHigh(),
          freeBytes,
          node.nodeId()
        );
      }
      return allocation.decision(
        Decision.NO,
        NAME,
        "the shard cannot remain on this node because it is above the high watermark cluster setting [%s=%s] "
          + "and there is less than the required [%s] free space on node, actual free: [%s]",
        CLUSTER_ROUTING_ALLOCATION_HIGH_DISK_WATERMARK_SETTING.getKey(),
        diskThresholdSettings.getHighWatermarkRaw(),
        diskThresholdSettings.getFreeBytesThresholdHigh(),
        new ByteSizeValue(freeBytes)
      );
    }

   ... ...

  }