HDFS-17646. Add Option to limit Balancer prefer highly utilized nodes num in each iteration.

Author: huangzhaobo99

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/Balancer.java

@@ -208,6 +208,8 @@ public class Balancer {
       + "\n\t[-sortTopNodes]"
       + "\tSort datanodes based on the utilization so "
       + "that highly utilized datanodes get scheduled first."
+      + "\n\t[-limitTopNodes <maxNodesNum>]"
+      + "\tUse the '-sortTopNodes' parameter to limit the number of top nodes"
       + "\n\t[-hotBlockTimeInterval]\tprefer to move cold blocks.";
 
   @VisibleForTesting
@@ -227,6 +229,7 @@ public class Balancer {
   private final long maxSizeToMove;
   private final long defaultBlockSize;
   private final boolean sortTopNodes;
+  private final int limitTopNodesNum;
   private final BalancerMetrics metrics;
 
   // all data node lists
@@ -352,6 +355,7 @@ static int getFailedTimesSinceLastSuccessfulBalance() {
     this.sourceNodes = p.getSourceNodes();
     this.runDuringUpgrade = p.getRunDuringUpgrade();
     this.sortTopNodes = p.getSortTopNodes();
+    this.limitTopNodesNum = p.getLimitTopNodesNum();
 
     this.maxSizeToMove = getLongBytes(conf,
         DFSConfigKeys.DFS_BALANCER_MAX_SIZE_TO_MOVE_KEY,
@@ -452,6 +456,7 @@ private long init(List<DatanodeStorageReport> reports) {
       }
     }
 
+    int overUtilizedDiffNum = Math.max(overUtilized.size() - limitTopNodesNum, 0);
     if (sortTopNodes) {
       sortOverUtilized(overUtilizedPercentage);
     }
@@ -461,27 +466,31 @@ private long init(List<DatanodeStorageReport> reports) {
     metrics.setNumOfUnderUtilizedNodes(underUtilized.size());
 
     Preconditions.checkState(dispatcher.getStorageGroupMap().size()
-        == overUtilized.size() + underUtilized.size() + aboveAvgUtilized.size()
-           + belowAvgUtilized.size(),
+            == overUtilizedDiffNum + overUtilized.size() + underUtilized.size()
+            + aboveAvgUtilized.size() + belowAvgUtilized.size(),
         "Mismatched number of storage groups");
 
     // return number of bytes to be moved in order to make the cluster balanced
     return Math.max(overLoadedBytes, underLoadedBytes);
   }
 
   private void sortOverUtilized(Map<Source, Double> overUtilizedPercentage) {
-    Preconditions.checkState(overUtilized instanceof List,
-        "Collection overUtilized is not a List.");
+    Preconditions.checkState(overUtilized instanceof LinkedList,
+        "Collection overUtilized is not a LinkedList.");
 
     LOG.info("Sorting over-utilized nodes by capacity" +
         " to bring down top used datanode capacity faster");
 
-    List<Source> list = (List<Source>) overUtilized;
+    LinkedList<Source> list = (LinkedList<Source>) overUtilized;
     list.sort(
         (Source source1, Source source2) ->
             (Double.compare(overUtilizedPercentage.get(source2),
                 overUtilizedPercentage.get(source1)))
     );
+    int size = overUtilized.size();
+    for (int i = 0; i < size - limitTopNodesNum; i++) {
+      list.removeLast();
+    }
   }
 
   private static long computeMaxSize2Move(final long capacity, final long remaining,
@@ -1071,6 +1080,12 @@ static BalancerParameters parse(String[] args) {
               b.setSortTopNodes(true);
               LOG.info("Balancer will sort nodes by" +
                   " capacity usage percentage to prioritize top used nodes");
+            } else if ("-limitTopNodesNum".equalsIgnoreCase(args[i])) {
+              Preconditions.checkArgument(++i < args.length,
+                  "limitTopNodesNum value is missing: args = " + Arrays.toString(args));
+              int limitNum = Integer.parseInt(args[i]);
+              LOG.info("Using a limitTopNodesNum of {}", limitNum);
+              b.setLimitTopNodesNum(limitNum);
             } else {
               throw new IllegalArgumentException("args = "
                   + Arrays.toString(args));

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/balancer/BalancerParameters.java

@@ -50,6 +50,8 @@ final class BalancerParameters {
 
   private final boolean sortTopNodes;
 
+  private final int limitTopNodesNum;
+
   static final BalancerParameters DEFAULT = new BalancerParameters();
 
   private BalancerParameters() {
@@ -67,6 +69,7 @@ private BalancerParameters(Builder builder) {
     this.runDuringUpgrade = builder.runDuringUpgrade;
     this.runAsService = builder.runAsService;
     this.sortTopNodes = builder.sortTopNodes;
+    this.limitTopNodesNum = builder.limitTopNodesNum;
     this.hotBlockTimeInterval = builder.hotBlockTimeInterval;
   }
 
@@ -110,6 +113,10 @@ boolean getSortTopNodes() {
     return this.sortTopNodes;
   }
 
+  int getLimitTopNodesNum() {
+    return this.limitTopNodesNum;
+  }
+
   long getHotBlockTimeInterval() {
     return this.hotBlockTimeInterval;
   }
@@ -120,12 +127,12 @@ public String toString() {
         + " max idle iteration = %s," + " #excluded nodes = %s,"
         + " #included nodes = %s," + " #source nodes = %s,"
         + " #blockpools = %s," + " run during upgrade = %s,"
-        + " sort top nodes = %s,"
+        + " sort top nodes = %s," + " limit top nodes num = %s"
         + " hot block time interval = %s]",
         Balancer.class.getSimpleName(), getClass().getSimpleName(), policy,
         threshold, maxIdleIteration, excludedNodes.size(),
         includedNodes.size(), sourceNodes.size(), blockpools.size(),
-        runDuringUpgrade, sortTopNodes, hotBlockTimeInterval);
+        runDuringUpgrade, sortTopNodes, limitTopNodesNum, hotBlockTimeInterval);
   }
 
   static class Builder {
@@ -141,6 +148,7 @@ static class Builder {
     private boolean runDuringUpgrade = false;
     private boolean runAsService = false;
     private boolean sortTopNodes = false;
+    private int limitTopNodesNum = Integer.MAX_VALUE;
     private long hotBlockTimeInterval = 0;
 
     Builder() {
@@ -201,6 +209,11 @@ Builder setSortTopNodes(boolean shouldSortTopNodes) {
       return this;
     }
 
+    Builder setLimitTopNodesNum(int limitTopNodesNum) {
+      this.limitTopNodesNum = limitTopNodesNum;
+      return this;
+    }
+
     BalancerParameters build() {
       return new BalancerParameters(this);
     }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/balancer/TestBalancerLongRunningTasks.java

@@ -672,6 +672,97 @@ public void testBalancerWithSortTopNodes() throws Exception {
     assertEquals(900, maxUsage);
   }
 
+  @Test(timeout = 60000)
+  public void testBalancerWithLimitTopNodesNum() throws Exception {
+    final Configuration conf = new HdfsConfiguration();
+    // Init the config (block size to 100)
+    initConf(conf);
+    conf.setInt(DFS_HEARTBEAT_INTERVAL_KEY, 30000);
+
+    final long totalCapacity = 1000L;
+    final int diffBetweenNodes = 50;
+
+    // Set up the nodes with two groups:
+    // 5 over-utilized nodes with 80%, 85%, 90%, 95%, 100% usage
+    // 2 under-utilized nodes with 0%, 5% usage
+    // With sortTopNodes and limitTopNodesNum option, 100% used ones will be chosen
+    final int numOfOverUtilizedDn = 5;
+    final int numOfUnderUtilizedDn = 2;
+    final int totalNumOfDn = numOfOverUtilizedDn + numOfUnderUtilizedDn;
+    final long[] capacityArray = new long[totalNumOfDn];
+    Arrays.fill(capacityArray, totalCapacity);
+
+    try (MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf)
+        .numDataNodes(totalNumOfDn)
+        .simulatedCapacities(capacityArray)
+        .build()) {
+      cluster.setDataNodesDead();
+      List<DataNode> dataNodes = cluster.getDataNodes();
+      // Create top used nodes
+      for (int i = 0; i < numOfOverUtilizedDn; i++) {
+        // Bring one node alive
+        DataNodeTestUtils.triggerHeartbeat(dataNodes.get(i));
+        DataNodeTestUtils.triggerBlockReport(dataNodes.get(i));
+        // Create nodes with: 80%, 85%, 90%, 95%, 100%
+        int nodeCapacity = (int) totalCapacity - diffBetweenNodes * (numOfOverUtilizedDn - i - 1);
+        TestBalancer.createFile(cluster, new Path("test_big" + i), nodeCapacity, (short) 1, 0);
+        cluster.setDataNodesDead();
+      }
+
+      // Create under utilized nodes
+      for (int i = 0; i < numOfUnderUtilizedDn; i++) {
+        // Bring one node alive
+        DataNodeTestUtils.triggerHeartbeat(dataNodes.get(i));
+        DataNodeTestUtils.triggerBlockReport(dataNodes.get(i));
+        // Create nodes with: 5%, 0%
+        int nodeCapacity = diffBetweenNodes * i;
+        TestBalancer.createFile(cluster, new Path("test_small" + i), nodeCapacity, (short) 1, 0);
+        cluster.setDataNodesDead();
+      }
+
+      // Bring all nodes alive
+      cluster.triggerHeartbeats();
+      cluster.triggerBlockReports();
+      cluster.waitFirstBRCompleted(0, 6000);
+
+      final BalancerParameters balancerParameters = Balancer.Cli.parse(new String[] {
+        "-policy", BalancingPolicy.Node.INSTANCE.getName(),
+        "-threshold", "1",
+        "-sortTopNodes",
+        "-limitTopNodesNum", "1"
+      });
+
+      client = NameNodeProxies.createProxy(conf, cluster.getFileSystem(0)
+              .getUri(), ClientProtocol.class)
+              .getProxy();
+
+      // Set max-size-to-move to small number
+      // so only top two nodes will be chosen in one iteration
+      conf.setLong(DFS_BALANCER_MAX_SIZE_TO_MOVE_KEY, 99L);
+      final Collection<URI> namenodes = DFSUtil.getInternalNsRpcUris(conf);
+      List<NameNodeConnector> connectors =
+          NameNodeConnector.newNameNodeConnectors(namenodes, Balancer.class.getSimpleName(),
+              Balancer.BALANCER_ID_PATH, conf, BalancerParameters.DEFAULT.getMaxIdleIteration());
+      final Balancer balancer = new Balancer(connectors.get(0), balancerParameters, conf);
+      Balancer.Result balancerResult = balancer.runOneIteration();
+
+      cluster.triggerDeletionReports();
+      cluster.triggerBlockReports();
+      cluster.triggerHeartbeats();
+
+      DatanodeInfo[] datanodeReport =
+          client.getDatanodeReport(HdfsConstants.DatanodeReportType.ALL);
+      long maxUsage = 0;
+      for (int i = 0; i < totalNumOfDn; i++) {
+        maxUsage = Math.max(maxUsage, datanodeReport[i].getDfsUsed());
+      }
+      // The maxUsage value is 950, only 100% of the nodes will be balanced
+      assertEquals(950, maxUsage);
+      assertTrue("BalancerResult is not as expected. " + balancerResult,
+          (balancerResult.getBytesAlreadyMoved() == 100 && balancerResult.getBlocksMoved() == 1));
+    }
+  }
+
   @Test(timeout = 100000)
   public void testMaxIterationTime() throws Exception {
     final Configuration conf = new HdfsConfiguration();