refactor: switch diskNormalized strategy to be more tune-able

docs/configuration/index.md

@@ -707,8 +707,8 @@ These Coordinator static configurations can be defined in the `coordinator/runti
 |`druid.coordinator.period`|The run period for the Coordinator. The Coordinator operates by maintaining the current state of the world in memory and periodically looking at the set of "used" segments and segments being served to make decisions about whether any changes need to be made to the data topology. This property sets the delay between each of these runs.|`PT60S`|
 |`druid.coordinator.startDelay`|The operation of the Coordinator works on the assumption that it has an up-to-date view of the state of the world when it runs, the current ZooKeeper interaction code, however, is written in a way that doesn’t allow the Coordinator to know for a fact that it’s done loading the current state of the world. This delay is a hack to give it enough time to believe that it has all the data.|`PT300S`|
 |`druid.coordinator.load.timeout`|The timeout duration for when the Coordinator assigns a segment to a Historical service.|`PT15M`|
-|`druid.coordinator.balancer.strategy`|The [balancing strategy](../design/coordinator.md#balancing-segments-in-a-tier) used by the Coordinator to distribute segments among the Historical servers in a tier. The `cost` strategy distributes segments by minimizing a cost function, `diskNormalized` divides these costs by the projected available disk headroom of each server and `random` distributes segments randomly.|`cost`|
-|`druid.coordinator.balancer.diskNormalized.moveCostSavingsThreshold`|Only used when `druid.coordinator.balancer.strategy` is `diskNormalized`. Minimum fractional cost reduction required before a segment is moved off a server that already holds it. A value of `0.05` requires the destination to be at least 5% cheaper than the source, which prevents oscillation between servers with similar projected headroom. Must be in `[0.0, 1.0)`; `0.0` disables the anti-oscillation discount.|`0.05`|
+|`druid.coordinator.balancer.strategy`|The [balancing strategy](../design/coordinator.md#balancing-segments-in-a-tier) used by the Coordinator to distribute segments among the Historical servers in a tier. The `cost` strategy distributes segments by minimizing a cost function, `diskNormalized` uses normal cost within a projected disk utilization threshold and exponentially penalizes candidates outside that threshold, and `random` distributes segments randomly.|`cost`|
+|`druid.coordinator.balancer.diskNormalized.utilizationThreshold`|Only used when `druid.coordinator.balancer.strategy` is `diskNormalized`. Projected disk utilization difference that the strategy ignores before applying the exponential utilization penalty. A value of `0.05` means candidates within 5 percentage points of the least-used candidate are ordered by normal cost, while candidates outside that band are penalized exponentially by the excess utilization difference. Must be in `(0.0, 1.0)`.|`0.05`|
 |`druid.coordinator.loadqueuepeon.http.repeatDelay`|The start and repeat delay (in milliseconds) for the load queue peon, which manages the load/drop queue of segments for any server.|1 minute|
 |`druid.coordinator.loadqueuepeon.http.batchSize`|Number of segment load/drop requests to batch in one HTTP request. Note that it must be smaller than or equal to the `druid.segmentCache.numLoadingThreads` config on Historical service. If this value is not configured, the coordinator uses the value of the `numLoadingThreads` for the respective server. | `druid.segmentCache.numLoadingThreads` |
 |`druid.coordinator.asOverlord.enabled`|Boolean value for whether this Coordinator service should act like an Overlord as well. This configuration allows users to simplify a Druid cluster by not having to deploy any standalone Overlord services. If set to true, then Overlord console is available at `http://coordinator-host:port/console.html` and be sure to set `druid.coordinator.asOverlord.overlordService` also.|false|

docs/design/coordinator.md

@@ -88,7 +88,7 @@ But in a tier with several Historicals (or a low replication factor), segment re
 Thus, the Coordinator constantly monitors the set of segments present on each Historical in a tier and employs one of the following strategies to identify segments that may be moved from one Historical to another to retain balance.
 
 - `cost` (default): For a given segment in a tier, this strategy picks the server with the minimum "cost" of placing that segment. The cost is a function of the data interval of the segment and the data intervals of all the segments already present on the candidate server. In essence, this strategy tries to avoid placing segments with adjacent or overlapping data intervals on the same server. This is based on the premise that adjacent-interval segments are more likely to be used together in a query and placing them on the same server may lead to skewed CPU usages of Historicals.
-- `diskNormalized`: A derivative of the `cost` strategy that divides the cost of placing a segment on a server by the server's projected available disk headroom. The projected usage ratio is `(diskUsed + segmentSizeIfNotAlreadyProjected) / maxSize`, so the disk-adjusted cost is `cost / max(EPSILON, 1 - projectedUsageRatio)`. This strongly penalizes servers that would be nearly full after placement and drives disk utilization to equalize across the tier, which is useful when historicals within a tier hold segments of widely varying sizes. To prevent oscillation when servers have similar headroom, a segment that is already placed on a server receives a cost discount; a move only fires when the destination saves at least `druid.coordinator.balancer.diskNormalized.moveCostSavingsThreshold` (default `0.05`, i.e. 5%) of the source's cost.
+- `diskNormalized`: A derivative of the `cost` strategy that applies normal cost when candidates are within a configured projected disk utilization threshold, and exponentially penalizes candidates outside that threshold. The projected usage ratio is `(diskUsed + segmentSizeIfNotAlreadyProjected) / maxSize`. A candidate whose projected usage ratio is within `druid.coordinator.balancer.diskNormalized.utilizationThreshold` (default `0.05`, i.e. 5 percentage points) of the least-used candidate is ordered by normal cost. A candidate outside that band is multiplied by `exp((utilizationGap - utilizationThreshold) / utilizationThreshold)`, where `utilizationGap` is the difference between the candidate's projected usage ratio and the least-used candidate's projected usage ratio. This drives disk utilization toward the configured band while avoiding moves based only on small utilization differences, which helps prevent oscillation when servers have similar utilization.
 - `random`: Distributes segments randomly across servers. This is an experimental strategy and is not recommended for a production cluster.
 
 All of the above strategies prioritize moving segments from the Historical with the least available disk space.

server/src/main/java/org/apache/druid/server/coordinator/balancer/CostBalancerStrategy.java

@@ -47,6 +47,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.ToDoubleFunction;
 import java.util.stream.Collectors;
 
 public class CostBalancerStrategy implements BalancerStrategy
@@ -317,6 +318,20 @@ protected double computePlacementCost(DataSegment proposalSegment, ServerHolder
     return cost;
   }
 
+  /**
+   * Creates a function that computes the cost of placing the given segment on a
+   * server. Subclasses may override this method when the cost for one server
+   * depends on the full candidate set.
+   */
+  protected ToDoubleFunction<ServerHolder> makePlacementCostFunction(
+      final DataSegment proposalSegment,
+      final List<ServerHolder> serverHolders,
+      final SegmentAction action
+  )
+  {
+    return server -> computePlacementCost(proposalSegment, server);
+  }
+
   /**
    * Orders the servers by increasing cost for placing the given segment.
    */
@@ -328,10 +343,12 @@ private List<ServerHolder> orderServersByPlacementCost(
   {
     final Stopwatch computeTime = Stopwatch.createStarted();
     final List<ListenableFuture<Pair<Double, ServerHolder>>> futures = new ArrayList<>();
+    final ToDoubleFunction<ServerHolder> placementCostFunction =
+        makePlacementCostFunction(segment, serverHolders, action);
     for (ServerHolder server : serverHolders) {
       futures.add(
           exec.submit(
-              () -> Pair.of(computePlacementCost(segment, server), server)
+              () -> Pair.of(placementCostFunction.applyAsDouble(server), server)
           )
       );
     }
@@ -405,4 +422,3 @@ private static Interval getCostComputeInterval(DataSegment segment)
   }
 
 }
-

server/src/main/java/org/apache/druid/server/coordinator/balancer/DiskNormalizedCostBalancerStrategy.java

@@ -22,86 +22,110 @@
 import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.ListeningExecutorService;
 import org.apache.druid.server.coordinator.ServerHolder;
+import org.apache.druid.server.coordinator.loading.SegmentAction;
 import org.apache.druid.timeline.DataSegment;
 
+import java.util.List;
+import java.util.function.ToDoubleFunction;
+
 /**
- * A {@link BalancerStrategy} which normalizes the cost of placing a segment on a
- * server as calculated by {@link CostBalancerStrategy} by dividing by the
- * server's projected available disk headroom.
+ * A {@link BalancerStrategy} which penalizes the cost of placing a segment on a
+ * server when the server's projected disk utilization is higher than the least
+ * utilized candidate by more than the configured utilization threshold.
  * <pre>
- * normalizedCost = cost / max(EPSILON, 1 - projectedUsageRatio)
- *     where projectedUsageRatio = (diskUsed + segmentSizeIfNotAlreadyProjected) / totalDiskSpace
+ * adjustedCost = cost, when utilizationGap <= utilizationThreshold
+ * adjustedCost = cost * exp((utilizationGap - utilizationThreshold) / utilizationThreshold), otherwise
+ *     where utilizationGap = projectedUsageRatio - minProjectedUsageRatio
+ *     and projectedUsageRatio = (diskUsed + segmentSizeIfNotAlreadyProjected) / totalDiskSpace
  * </pre>
- * The denominator diverges as a server approaches full, so disk fullness has
- * more weight over the placement decision when servers are nearly full,
- * regardless of asymmetries in the locality cost. {@link #EPSILON} is a small
- * numerical floor on the divisor to guard against division by zero (or by
- * negative values during in-flight loads).
- * <p>
- * To prevent oscillation when servers have similar headroom, any server that
- * is already projected to hold the segment (the source on a move, or a currently
- * serving node on a drop) receives a cost discount equal to
- * {@link DiskNormalizedCostBalancerStrategyConfig.DEFAULT_MOVE_COST_SAVINGS_THRESHOLD}. A move therefore fires only when
- * the destination saves at least this fraction of the source's cost. The default
- * is configurable via
- * {@code druid.coordinator.balancer.diskNormalized.moveCostSavingsThreshold}.
+ * Servers inside the threshold band use the normal cost strategy, which avoids
+ * using small utilization differences as a reason to move segments back and
+ * forth. Servers outside the band are penalized exponentially by the amount
+ * they exceed it.
  */
 public class DiskNormalizedCostBalancerStrategy extends CostBalancerStrategy
 {
-  /**
-   * Numerical floor on the headroom divisor to prevent division by zero or by
-   * negative values when {@code usageRatio >= 1.0} (possible for over-allocated
-   * servers or during in-flight loads).
-   */
-  static final double EPSILON = 1e-6;
-
-  private final double sourceCostMultiplier;
+  private final double utilizationThreshold;
 
   public DiskNormalizedCostBalancerStrategy(ListeningExecutorService exec)
   {
-    this(exec, DiskNormalizedCostBalancerStrategyConfig.DEFAULT_MOVE_COST_SAVINGS_THRESHOLD);
+    this(exec, DiskNormalizedCostBalancerStrategyConfig.DEFAULT_UTILIZATION_THRESHOLD);
   }
 
-  public DiskNormalizedCostBalancerStrategy(ListeningExecutorService exec, double moveCostSavingsThreshold)
+  public DiskNormalizedCostBalancerStrategy(ListeningExecutorService exec, double utilizationThreshold)
   {
     super(exec);
     Preconditions.checkArgument(
-        moveCostSavingsThreshold >= 0.0 && moveCostSavingsThreshold < 1.0,
-        "moveCostSavingsThreshold[%s] must be in [0.0, 1.0)",
-        moveCostSavingsThreshold
+        utilizationThreshold > 0.0 && utilizationThreshold < 1.0,
+        "utilizationThreshold[%s] must be in (0.0, 1.0)",
+        utilizationThreshold
     );
-    this.sourceCostMultiplier = 1.0 - moveCostSavingsThreshold;
+    this.utilizationThreshold = utilizationThreshold;
   }
 
   @Override
-  protected double computePlacementCost(
+  protected ToDoubleFunction<ServerHolder> makePlacementCostFunction(
       final DataSegment proposalSegment,
-      final ServerHolder server
+      final List<ServerHolder> serverHolders,
+      final SegmentAction action
   )
   {
-    double cost = super.computePlacementCost(proposalSegment, server);
+    final double minProjectedUsageRatio = serverHolders.stream()
+                                                       .filter(server -> isCandidateForUtilizationBaseline(
+                                                           proposalSegment,
+                                                           server,
+                                                           action
+                                                       ))
+                                                       .mapToDouble(server -> computeProjectedUsageRatio(
+                                                           proposalSegment,
+                                                           server
+                                                       ))
+                                                       .filter(Double::isFinite)
+                                                       .min()
+                                                       .orElse(0.0);
 
-    if (cost == Double.POSITIVE_INFINITY) {
-      return cost;
-    }
+    return server -> {
+      final double cost = super.computePlacementCost(proposalSegment, server);
+
+      if (cost == Double.POSITIVE_INFINITY) {
+        return cost;
+      }
+
+      final double projectedUsageRatio = computeProjectedUsageRatio(proposalSegment, server);
+      if (!Double.isFinite(projectedUsageRatio)) {
+        return cost;
+      }
 
-    // A server with non-positive maxSize cannot hold anything and will be
-    // rejected by canLoadSegment; return the raw cost to avoid NaN propagation.
+      final double utilizationGap = projectedUsageRatio - minProjectedUsageRatio;
+      if (utilizationGap <= utilizationThreshold) {
+        return cost;
+      }
+
+      return cost * Math.exp((utilizationGap - utilizationThreshold) / utilizationThreshold);
+    };
+  }
+
+  private static boolean isCandidateForUtilizationBaseline(
+      final DataSegment proposalSegment,
+      final ServerHolder server,
+      final SegmentAction action
+  )
+  {
+    return action != SegmentAction.LOAD || server.canLoadSegment(proposalSegment);
+  }
+
+  private static double computeProjectedUsageRatio(
+      final DataSegment proposalSegment,
+      final ServerHolder server
+  )
+  {
     final long maxSize = server.getMaxSize();
     if (maxSize <= 0) {
-      return cost;
+      return Double.POSITIVE_INFINITY;
     }
 
     final boolean alreadyProjected = server.isProjectedSegment(proposalSegment);
     final long projectedSizeUsed = server.getSizeUsed() + (alreadyProjected ? 0 : proposalSegment.getSize());
-    final double usageRatio = (double) projectedSizeUsed / maxSize;
-    final double headroom = Math.max(EPSILON, 1.0 - usageRatio);
-    double normalizedCost = cost / headroom;
-
-    if (alreadyProjected) {
-      normalizedCost *= sourceCostMultiplier;
-    }
-
-    return normalizedCost;
+    return (double) projectedSizeUsed / maxSize;
   }
 }

server/src/main/java/org/apache/druid/server/coordinator/balancer/DiskNormalizedCostBalancerStrategyConfig.java

@@ -35,21 +35,21 @@
 public class DiskNormalizedCostBalancerStrategyConfig
 {
   /**
-   * Default minimum fractional cost reduction required before a segment will
-   * be moved off a server that is already projected to hold it. A value of
-   * {@code 0.05} means the destination must be at least 5% cheaper than the
-   * source for the move to happen.
+   * Default projected disk utilization difference that the strategy ignores
+   * before applying the exponential utilization penalty. A value of
+   * {@code 0.05} means servers within 5 percentage points of the least-used
+   * candidate are ordered by the normal cost strategy.
    */
-  static final double DEFAULT_MOVE_COST_SAVINGS_THRESHOLD = 0.05;
+  static final double DEFAULT_UTILIZATION_THRESHOLD = 0.05;
 
   /**
-   * Minimum fractional cost reduction required to move a segment off a server
-   * that is already projected to hold it. For example, a value of {@code 0.05} means the
-   * destination must be at least 5% cheaper than the source before a move
-   * fires.
+   * Projected disk utilization difference that the strategy ignores before
+   * applying the exponential utilization penalty. For example, a value of
+   * {@code 0.05} means servers within 5 percentage points of the least-used
+   * candidate are ordered by the normal cost strategy.
    */
   @JsonProperty
-  private final double moveCostSavingsThreshold;
+  private final double utilizationThreshold;
 
   public DiskNormalizedCostBalancerStrategyConfig()
   {
@@ -58,20 +58,20 @@ public DiskNormalizedCostBalancerStrategyConfig()
 
   @JsonCreator
   public DiskNormalizedCostBalancerStrategyConfig(
-      @JsonProperty("moveCostSavingsThreshold") @Nullable Double moveCostSavingsThreshold
+      @JsonProperty("utilizationThreshold") @Nullable Double utilizationThreshold
   )
   {
-    this.moveCostSavingsThreshold = Configs.valueOrDefault(moveCostSavingsThreshold, DEFAULT_MOVE_COST_SAVINGS_THRESHOLD);
+    this.utilizationThreshold = Configs.valueOrDefault(utilizationThreshold, DEFAULT_UTILIZATION_THRESHOLD);
 
     Preconditions.checkArgument(
-        this.moveCostSavingsThreshold >= 0.0 && this.moveCostSavingsThreshold < 1.0,
-        "'druid.coordinator.balancer.diskNormalized.moveCostSavingsThreshold'[%s] must be in [0.0, 1.0)",
-        this.moveCostSavingsThreshold
+        this.utilizationThreshold > 0.0 && this.utilizationThreshold < 1.0,
+        "'druid.coordinator.balancer.diskNormalized.utilizationThreshold'[%s] must be in (0.0, 1.0)",
+        this.utilizationThreshold
     );
   }
 
-  public double getMoveCostSavingsThreshold()
+  public double getUtilizationThreshold()
   {
-    return moveCostSavingsThreshold;
+    return utilizationThreshold;
   }
 }

server/src/main/java/org/apache/druid/server/coordinator/balancer/DiskNormalizedCostBalancerStrategyFactory.java

@@ -44,7 +44,7 @@ public BalancerStrategy createBalancerStrategy(int numBalancerThreads)
   {
     return new DiskNormalizedCostBalancerStrategy(
         getOrCreateBalancerExecutor(numBalancerThreads),
-        config.getMoveCostSavingsThreshold()
+        config.getUtilizationThreshold()
     );
   }
 }