Refactor soft constraints to simply the algorithm and fix potential issues. (#520)

* Refactor soft constraints to simply the algorithm and fix potential issues.

1. Check for zero weight so as to avoid unnecessary calculations.
2. Simply the soft constraint interfaces and implementations. Avoid duplicate code.
3. Adjust partition movements constraint logic to reduce the chance of moving partition when the baseline and best possible assignment diverge.
4. Estimate utilization in addition to the other usage estimation. The estimation will be used as a base when calculating the capacity usage score. This is to ensure the algorithm treats different clusters with different overall usage in the same way.
5. Fix the issue that high utilization calculation does not consider the current proposed replica usage.
6. Use Sigmoid to calculate usage-based soft constraints score. This enhances the assignment result of the algorithm.
7. Adjust the related test cases.

Author: jiajunwang

helix-core/pom.xml

@@ -37,7 +37,7 @@ under the License.
       org.I0Itec.zkclient*,
       org.apache.commons.cli*;version="[1.2,2)",
       org.apache.commons.io*;version="[1.4,2)",
-      org.apache.commons.math*;version="[2.1,3)",
+      org.apache.commons.math*;version="[2.1,4)",
       org.apache.jute*;resolution:=optional,
       org.apache.zookeeper.server.persistence*;resolution:=optional,
       org.apache.zookeeper.server.util*;resolution:=optional,
@@ -139,6 +139,11 @@ under the License.
       <artifactId>commons-math</artifactId>
       <version>2.1</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-math3</artifactId>
+      <version>3.6.1</version>
+    </dependency>
     <dependency>
       <groupId>commons-codec</groupId>
       <artifactId>commons-codec</artifactId>

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithm.java

@@ -109,19 +109,22 @@ private Optional<AssignableNode> getNodeWithHighestPoints(AssignableReplica repl
       return Optional.empty();
     }
 
-    Function<AssignableNode, Float> calculatePoints =
+    Function<AssignableNode, Double> calculatePoints =
         (candidateNode) -> getAssignmentNormalizedScore(candidateNode, replica, clusterContext);
 
     return candidateNodes.stream().max(Comparator.comparing(calculatePoints));
   }
 
-  private float getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
+  private double getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
-    float sum = 0;
+    double sum = 0;
     for (Map.Entry<SoftConstraint, Float> softConstraintEntry : _softConstraints.entrySet()) {
       SoftConstraint softConstraint = softConstraintEntry.getKey();
       float weight = softConstraintEntry.getValue();
-      sum += weight * softConstraint.getAssignmentNormalizedScore(node, replica, clusterContext);
+      if (weight != 0) {
+        // Skip calculating zero weighted constraints.
+        sum += weight * softConstraint.getAssignmentNormalizedScore(node, replica, clusterContext);
+      }
     }
     return sum;
   }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ConstraintBasedAlgorithmFactory.java

@@ -24,31 +24,25 @@
 import java.util.Map;
 import java.util.Properties;
 
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Maps;
 import org.apache.helix.HelixManagerProperties;
 import org.apache.helix.SystemPropertyKeys;
 import org.apache.helix.controller.rebalancer.waged.RebalanceAlgorithm;
 import org.apache.helix.model.ClusterConfig;
 
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Maps;
-
 /**
  * The factory class to create an instance of {@link ConstraintBasedAlgorithm}
  */
 public class ConstraintBasedAlgorithmFactory {
-  // Evenness constraints tend to score within a smaller range.
-  // In order to let their scores cause enough difference in the final evaluation result, we need to
-  // enlarge the overall weight of the evenness constraints compared with the movement constraint.
-  // TODO: Tune or make the following factor configurable.
-  private static final int EVENNESS_PREFERENCE_NORMALIZE_FACTOR = 50;
   private static final Map<String, Float> MODEL = new HashMap<String, Float>() {
     {
       // The default setting
-      put(PartitionMovementConstraint.class.getSimpleName(), 1f);
-      put(InstancePartitionsCountConstraint.class.getSimpleName(), 0.3f);
-      put(ResourcePartitionAntiAffinityConstraint.class.getSimpleName(), 0.1f);
-      put(ResourceTopStateAntiAffinityConstraint.class.getSimpleName(), 0.1f);
-      put(MaxCapacityUsageInstanceConstraint.class.getSimpleName(), 0.5f);
+      put(PartitionMovementConstraint.class.getSimpleName(), 2f);
+      put(InstancePartitionsCountConstraint.class.getSimpleName(), 1f);
+      put(ResourcePartitionAntiAffinityConstraint.class.getSimpleName(), 1f);
+      put(ResourceTopStateAntiAffinityConstraint.class.getSimpleName(), 3f);
+      put(MaxCapacityUsageInstanceConstraint.class.getSimpleName(), 5f);
     }
   };
 
@@ -68,21 +62,19 @@ public static RebalanceAlgorithm getInstance(
             new ValidGroupTagConstraint(), new SamePartitionOnInstanceConstraint());
 
     int evennessPreference =
-        preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.EVENNESS, 1)
-            * EVENNESS_PREFERENCE_NORMALIZE_FACTOR;
+        preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.EVENNESS, 1);
     int movementPreference =
         preferences.getOrDefault(ClusterConfig.GlobalRebalancePreferenceKey.LESS_MOVEMENT, 1);
-    float evennessRatio = (float) evennessPreference / (evennessPreference + movementPreference);
-    float movementRatio = (float) movementPreference / (evennessPreference + movementPreference);
 
-    List<SoftConstraint> softConstraints = ImmutableList.of(new PartitionMovementConstraint(),
-        new InstancePartitionsCountConstraint(), new ResourcePartitionAntiAffinityConstraint(),
-        new ResourceTopStateAntiAffinityConstraint(), new MaxCapacityUsageInstanceConstraint());
+    List<SoftConstraint> softConstraints = ImmutableList
+        .of(new PartitionMovementConstraint(), new InstancePartitionsCountConstraint(),
+            new ResourcePartitionAntiAffinityConstraint(),
+            new ResourceTopStateAntiAffinityConstraint(), new MaxCapacityUsageInstanceConstraint());
     Map<SoftConstraint, Float> softConstraintsWithWeight = Maps.toMap(softConstraints, key -> {
       String name = key.getClass().getSimpleName();
       float weight = MODEL.get(name);
-      return name.equals(PartitionMovementConstraint.class.getSimpleName()) ? movementRatio * weight
-          : evennessRatio * weight;
+      return name.equals(PartitionMovementConstraint.class.getSimpleName()) ?
+          movementPreference * weight : evennessPreference * weight;
     });
 
     return new ConstraintBasedAlgorithm(hardConstraints, softConstraintsWithWeight);

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/InstancePartitionsCountConstraint.java

@@ -29,20 +29,13 @@
  * Discourage the assignment if the instance's occupancy rate is above average
  * The normalized score will be within [0, 1]
  */
-class InstancePartitionsCountConstraint extends SoftConstraint {
-  private static final float MAX_SCORE = 1f;
-  private static final float MIN_SCORE = 0f;
-
-  InstancePartitionsCountConstraint() {
-    super(MAX_SCORE, MIN_SCORE);
-  }
+class InstancePartitionsCountConstraint extends UsageSoftConstraint {
 
   @Override
-  protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
-    float doubleEstimatedMaxPartitionCount = 2 * clusterContext.getEstimatedMaxPartitionCount();
-    float currentPartitionCount = node.getAssignedReplicaCount();
-    return Math.max((doubleEstimatedMaxPartitionCount - currentPartitionCount)
-        / doubleEstimatedMaxPartitionCount, 0);
+    int estimatedMaxPartitionCount = clusterContext.getEstimatedMaxPartitionCount();
+    int currentPartitionCount = node.getAssignedReplicaCount();
+    return computeUtilizationScore(estimatedMaxPartitionCount, currentPartitionCount);
   }
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/MaxCapacityUsageInstanceConstraint.java

@@ -30,18 +30,13 @@
  * that it is that much less desirable to assign anything on the given node.
  * It is a greedy approach since it evaluates only on the most used capacity key.
  */
-class MaxCapacityUsageInstanceConstraint extends SoftConstraint {
-  private static final float MIN_SCORE = 0;
-  private static final float MAX_SCORE = 1;
-
-  MaxCapacityUsageInstanceConstraint() {
-    super(MAX_SCORE, MIN_SCORE);
-  }
+class MaxCapacityUsageInstanceConstraint extends UsageSoftConstraint {
 
   @Override
-  protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
-    float maxCapacityUsage = node.getHighestCapacityUtilization();
-    return 1.0f - maxCapacityUsage / 2.0f;
+    float estimatedMaxUtilization = clusterContext.getEstimatedMaxUtilization();
+    float projectedHighestUtilization = node.getProjectedHighestUtilization(replica.getCapacity());
+    return computeUtilizationScore(estimatedMaxUtilization, projectedHighestUtilization);
   }
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/PartitionMovementConstraint.java

@@ -38,39 +38,33 @@
  * evaluated score will become lower.
  */
 class PartitionMovementConstraint extends SoftConstraint {
-  private static final float MAX_SCORE = 1f;
-  private static final float MIN_SCORE = 0f;
+  private static final double MAX_SCORE = 1f;
+  private static final double MIN_SCORE = 0f;
   //TODO: these factors will be tuned based on user's preference
   // This factor indicates the default score that is evaluated if only partition allocation matches
   // (states are different).
-  private static final float ALLOCATION_MATCH_FACTOR = 0.5f;
-  // This factor indicates the contribution of the Baseline assignment matching to the final score.
-  private static final float BASELINE_MATCH_FACTOR = 0.25f;
+  private static final double ALLOCATION_MATCH_FACTOR = 0.5;
 
   PartitionMovementConstraint() {
     super(MAX_SCORE, MIN_SCORE);
   }
 
   @Override
-  protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
-    Map<String, String> bestPossibleStateMap =
+    // Prioritize the previous Best Possible assignment
+    Map<String, String> bestPossibleAssignment =
         getStateMap(replica, clusterContext.getBestPossibleAssignment());
-    Map<String, String> baselineStateMap =
+    if (!bestPossibleAssignment.isEmpty()) {
+      return calculateAssignmentScale(node, replica, bestPossibleAssignment);
+    }
+    // else, compare the baseline only if the best possible assignment does not contain the replica
+    Map<String, String> baselineAssignment =
         getStateMap(replica, clusterContext.getBaselineAssignment());
-
-    // Prioritize the matching of the previous Best Possible assignment.
-    float scale = calculateAssignmentScale(node, replica, bestPossibleStateMap);
-    // If the baseline is also provided, adjust the final score accordingly.
-    scale = scale * (1 - BASELINE_MATCH_FACTOR)
-        + calculateAssignmentScale(node, replica, baselineStateMap) * BASELINE_MATCH_FACTOR;
-
-    return scale;
-  }
-
-  @Override
-  NormalizeFunction getNormalizeFunction() {
-    return score -> score * (getMaxScore() - getMinScore()) + getMinScore();
+    if (!baselineAssignment.isEmpty()) {
+      return calculateAssignmentScale(node, replica, baselineAssignment);
+    }
+    return 0;
   }
 
   private Map<String, String> getStateMap(AssignableReplica replica,
@@ -83,14 +77,20 @@ private Map<String, String> getStateMap(AssignableReplica replica,
     return assignment.get(resourceName).getReplicaMap(new Partition(partitionName));
   }
 
-  private float calculateAssignmentScale(AssignableNode node, AssignableReplica replica,
+  private double calculateAssignmentScale(AssignableNode node, AssignableReplica replica,
       Map<String, String> instanceToStateMap) {
     String instanceName = node.getInstanceName();
     if (!instanceToStateMap.containsKey(instanceName)) {
       return 0;
     } else {
-      return (instanceToStateMap.get(instanceName).equals(replica.getReplicaState()) ? 1
-          : ALLOCATION_MATCH_FACTOR);
+      return (instanceToStateMap.get(instanceName).equals(replica.getReplicaState()) ? 1 :
+          ALLOCATION_MATCH_FACTOR);
     }
   }
+
+  @Override
+  protected NormalizeFunction getNormalizeFunction() {
+    // PartitionMovementConstraint already scale the score properly.
+    return (score) -> score;
+  }
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourcePartitionAntiAffinityConstraint.java

@@ -29,24 +29,15 @@
  * the same resource be assigned to the same node to minimize the impact of node failure scenarios.
  * The score is higher the fewer the partitions are on the node belonging to the same resource.
  */
-class ResourcePartitionAntiAffinityConstraint extends SoftConstraint {
-  private static final float MAX_SCORE = 1f;
-  private static final float MIN_SCORE = 0f;
-
-  ResourcePartitionAntiAffinityConstraint() {
-    super(MAX_SCORE, MIN_SCORE);
-  }
-
+class ResourcePartitionAntiAffinityConstraint extends UsageSoftConstraint {
   @Override
-  protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
     String resource = replica.getResourceName();
     int curPartitionCountForResource = node.getAssignedPartitionsByResource(resource).size();
-    int doubleMaxPartitionCountForResource =
-        2 * clusterContext.getEstimatedMaxPartitionByResource(resource);
-    // The score measures the twice the max allowed count versus current counts
-    // The returned value is a measurement of remaining quota ratio, in the case of exceeding allowed counts, return 0
-    return Math.max(((float) doubleMaxPartitionCountForResource - curPartitionCountForResource)
-        / doubleMaxPartitionCountForResource, 0);
+    int estimatedMaxPartitionCountForResource =
+        clusterContext.getEstimatedMaxPartitionByResource(resource);
+    return computeUtilizationScore(estimatedMaxPartitionCountForResource,
+        curPartitionCountForResource);
   }
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/ResourceTopStateAntiAffinityConstraint.java

@@ -28,26 +28,17 @@
  * The higher number the number of top state partitions assigned to the instance, the lower the
  * score, vice versa.
  */
-class ResourceTopStateAntiAffinityConstraint extends SoftConstraint {
-  private static final float MAX_SCORE = 1f;
-  private static final float MIN_SCORE = 0f;
-
-  ResourceTopStateAntiAffinityConstraint() {
-    super(MAX_SCORE, MIN_SCORE);
-  }
-
+class ResourceTopStateAntiAffinityConstraint extends UsageSoftConstraint {
   @Override
-  protected float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
     if (!replica.isReplicaTopState()) {
-      return (getMaxScore() + getMinScore()) / 2.0f;
+      // For non top state replica, this constraint is not applicable.
+      // So return zero on any assignable node candidate.
+      return 0;
     }
-
     int curTopPartitionCountForResource = node.getAssignedTopStatePartitionsCount();
-    int doubleMaxTopStateCount = 2 * clusterContext.getEstimatedMaxTopStateCount();
-
-    return Math.max(
-        ((float) doubleMaxTopStateCount - curTopPartitionCountForResource) / doubleMaxTopStateCount,
-        0);
+    int estimatedMaxTopStateCount = clusterContext.getEstimatedMaxTopStateCount();
+    return computeUtilizationScore(estimatedMaxTopStateCount, curTopPartitionCountForResource);
   }
 }

helix-core/src/main/java/org/apache/helix/controller/rebalancer/waged/constraints/SoftConstraint.java

@@ -30,8 +30,8 @@
  * The lower score the score, the worse the assignment; Intuitively, the assignment is penalized.
  */
 abstract class SoftConstraint {
-  private float _maxScore = 1000f;
-  private float _minScore = -1000f;
+  private final double _maxScore;
+  private final double _minScore;
 
   interface NormalizeFunction {
     /**
@@ -40,47 +40,41 @@ interface NormalizeFunction {
      * @param originScore The origin score
      * @return The normalized value between (0, 1)
      */
-    float scale(float originScore);
-  }
-
-  /**
-   * Default constructor, uses default min/max scores
-   */
-  SoftConstraint() {
+    double scale(double originScore);
   }
 
   /**
    * Child class customize the min/max score on its own
    * @param maxScore The max score
    * @param minScore The min score
    */
-  SoftConstraint(float maxScore, float minScore) {
+  SoftConstraint(double maxScore, double minScore) {
     _maxScore = maxScore;
     _minScore = minScore;
   }
 
-  float getMaxScore() {
+  protected double getMaxScore() {
     return _maxScore;
   }
 
-  float getMinScore() {
+  protected double getMinScore() {
     return _minScore;
   }
 
   /**
    * Evaluate and give a score for an potential assignment partition -> instance
    * Child class only needs to care about how the score is implemented
-   * @return The score of the assignment in float value
+   * @return The score of the assignment in double value
    */
-  protected abstract float getAssignmentScore(AssignableNode node, AssignableReplica replica,
+  protected abstract double getAssignmentScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext);
 
   /**
    * Evaluate and give a score for an potential assignment partition -> instance
    * It's the only exposed method to the caller
    * @return The score is normalized to be within MinScore and MaxScore
    */
-  float getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
+  double getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replica,
       ClusterContext clusterContext) {
     return getNormalizeFunction().scale(getAssignmentScore(node, replica, clusterContext));
   }
@@ -90,7 +84,7 @@ float getAssignmentNormalizedScore(AssignableNode node, AssignableReplica replic
    * Child class could override the method and customize the method on its own
    * @return The MinMaxScaler instance by default
    */
-  NormalizeFunction getNormalizeFunction() {
+  protected NormalizeFunction getNormalizeFunction() {
     return (score) -> (score - getMinScore()) / (getMaxScore() - getMinScore());
   }
 }