HDDS-5517. Support multiple container moves from a source datanode in one balance iteration

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java

@@ -78,18 +78,19 @@ public class ContainerBalancer {
   private List<DatanodeUsageInfo> unBalancedNodes;
   private List<DatanodeUsageInfo> overUtilizedNodes;
   private List<DatanodeUsageInfo> underUtilizedNodes;
-  private List<DatanodeUsageInfo> withinThresholdUtilizedNodes;
   private ContainerBalancerConfiguration config;
   private ContainerBalancerMetrics metrics;
   private long clusterCapacity;
   private long clusterUsed;
   private long clusterRemaining;
   private double clusterAvgUtilisation;
   private double upperLimit;
+  private double lowerLimit;
   private volatile boolean balancerRunning;
   private volatile Thread currentBalancingThread;
   private Lock lock;
   private ContainerBalancerSelectionCriteria selectionCriteria;
+  private SourceDataNodeSelectionCriteria srcDnSelectionCriteria;
   private Map<DatanodeDetails, ContainerMoveSelection> sourceToTargetMap;
   private Map<DatanodeDetails, Long> sizeLeavingNode;
   private Map<DatanodeDetails, Long> sizeEnteringNode;
@@ -127,12 +128,12 @@ public ContainerBalancer(
     this.selectedContainers = new HashSet<>();
     this.overUtilizedNodes = new ArrayList<>();
     this.underUtilizedNodes = new ArrayList<>();
-    this.withinThresholdUtilizedNodes = new ArrayList<>();
     this.unBalancedNodes = new ArrayList<>();
+    this.sizeEnteringNode = new HashMap<>();
 
     this.lock = new ReentrantLock();
-    findTargetStrategy =
-        new FindTargetGreedy(containerManager, placementPolicy);
+    findTargetStrategy = new FindTargetGreedy(
+        containerManager, placementPolicy, sizeEnteringNode);
   }
 
   /**
@@ -251,7 +252,6 @@ private boolean initializeIteration() {
     this.selectedContainers.clear();
     this.overUtilizedNodes.clear();
     this.underUtilizedNodes.clear();
-    this.withinThresholdUtilizedNodes.clear();
     this.unBalancedNodes.clear();
     this.countDatanodesInvolvedPerIteration = 0;
     this.sizeMovedPerIteration = 0;
@@ -262,20 +262,19 @@ private boolean initializeIteration() {
           clusterAvgUtilisation);
     }
 
-    // under utilized nodes have utilization(that is, used / capacity) less
-    // than lower limit
-    double lowerLimit = clusterAvgUtilisation - threshold;
-
     // over utilized nodes have utilization(that is, used / capacity) greater
     // than upper limit
     this.upperLimit = clusterAvgUtilisation + threshold;
+    // under utilized nodes have utilization(that is, used / capacity) less
+    // than lower limit
+    this.lowerLimit = clusterAvgUtilisation - threshold;
 
     if (LOG.isDebugEnabled()) {
       LOG.debug("Lower limit for utilization is {} and Upper limit for " +
           "utilization is {}", lowerLimit, upperLimit);
     }
 
-    long overUtilizedBytes = 0L, underUtilizedBytes = 0L;
+    long totalOverUtilizedBytes = 0L, totalUnderUtilizedBytes = 0L;
     // find over and under utilized nodes
     for (DatanodeUsageInfo datanodeUsageInfo : datanodeUsageInfos) {
       if (!isBalancerRunning()) {
@@ -291,7 +290,7 @@ private boolean initializeIteration() {
             datanodeUsageInfo.getScmNodeStat().getRemaining().get(),
             utilization);
       }
-      if (utilization > upperLimit) {
+      if (Double.compare(utilization, upperLimit) > 0) {
         overUtilizedNodes.add(datanodeUsageInfo);
         metrics.incrementDatanodesNumToBalance(1);
 
@@ -300,27 +299,28 @@ private boolean initializeIteration() {
             ratioToPercent(utilization)));
 
         // amount of bytes greater than upper limit in this node
-        overUtilizedBytes += ratioToBytes(
+        Long overUtilizedBytes = ratioToBytes(
             datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
             utilization) - ratioToBytes(
             datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
             upperLimit);
-      } else if (utilization < lowerLimit) {
+        totalOverUtilizedBytes += overUtilizedBytes;
+      } else if (Double.compare(utilization, lowerLimit) < 0) {
         underUtilizedNodes.add(datanodeUsageInfo);
         metrics.incrementDatanodesNumToBalance(1);
 
         // amount of bytes lesser than lower limit in this node
-        underUtilizedBytes += ratioToBytes(
+        Long underUtilizedBytes = ratioToBytes(
             datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
             lowerLimit) - ratioToBytes(
             datanodeUsageInfo.getScmNodeStat().getCapacity().get(),
             utilization);
-      } else {
-        withinThresholdUtilizedNodes.add(datanodeUsageInfo);
+        totalUnderUtilizedBytes += underUtilizedBytes;
       }
     }
     metrics.setDataSizeToBalanceGB(
-        Math.max(overUtilizedBytes, underUtilizedBytes) / OzoneConsts.GB);
+        Math.max(totalOverUtilizedBytes, totalUnderUtilizedBytes) /
+            OzoneConsts.GB);
     Collections.reverse(underUtilizedNodes);
 
     unBalancedNodes = new ArrayList<>(
@@ -339,43 +339,43 @@ private boolean initializeIteration() {
 
     selectionCriteria = new ContainerBalancerSelectionCriteria(config,
         nodeManager, replicationManager, containerManager);
-    sourceToTargetMap = new HashMap<>(overUtilizedNodes.size() +
-        withinThresholdUtilizedNodes.size());
+    sourceToTargetMap = new HashMap<>(overUtilizedNodes.size());
 
     // initialize maps to track how much size is leaving and entering datanodes
-    sizeLeavingNode = new HashMap<>(overUtilizedNodes.size() +
-        withinThresholdUtilizedNodes.size());
+    sizeLeavingNode = new HashMap<>(overUtilizedNodes.size());
     overUtilizedNodes.forEach(datanodeUsageInfo -> sizeLeavingNode
         .put(datanodeUsageInfo.getDatanodeDetails(), 0L));
-    withinThresholdUtilizedNodes.forEach(datanodeUsageInfo -> sizeLeavingNode
-        .put(datanodeUsageInfo.getDatanodeDetails(), 0L));
 
-    sizeEnteringNode = new HashMap<>(underUtilizedNodes.size() +
-        withinThresholdUtilizedNodes.size());
+    sizeEnteringNode.clear();
     underUtilizedNodes.forEach(datanodeUsageInfo -> sizeEnteringNode
         .put(datanodeUsageInfo.getDatanodeDetails(), 0L));
-    withinThresholdUtilizedNodes.forEach(datanodeUsageInfo -> sizeEnteringNode
-        .put(datanodeUsageInfo.getDatanodeDetails(), 0L));
 
+    srcDnSelectionCriteria = new
+        SourceDataNodeSelectionCriteria(overUtilizedNodes, sizeLeavingNode);
     return true;
   }
 
   private IterationResult doIteration() {
     // note that potential and selected targets are updated in the following
     // loop
-    List<DatanodeDetails> potentialTargets = getPotentialTargets();
+    List<DatanodeUsageInfo> potentialTargets = getPotentialTargets();
     Set<DatanodeDetails> selectedTargets =
         new HashSet<>(potentialTargets.size());
     moveSelectionToFutureMap = new HashMap<>(unBalancedNodes.size());
     boolean isMoveGenerated = false;
 
     try {
       // match each overUtilized node with a target
-      for (DatanodeUsageInfo datanodeUsageInfo : overUtilizedNodes) {
+      while (true) {
+        DatanodeDetails source =
+            srcDnSelectionCriteria.getNextCandidateSourceDataNode();
+        if (source == null) {
+          break;
+        }
         if (!isBalancerRunning()) {
           return IterationResult.ITERATION_INTERRUPTED;
         }
-        DatanodeDetails source = datanodeUsageInfo.getDatanodeDetails();
+
         IterationResult result = checkConditionsForBalancing();
         if (result != null) {
           return result;
@@ -387,53 +387,23 @@ private IterationResult doIteration() {
           isMoveGenerated = true;
           LOG.info("ContainerBalancer is trying to move container {} from " +
                   "source datanode {} to target datanode {}",
-              moveSelection.getContainerID().toString(), source.getUuidString(),
+              moveSelection.getContainerID().toString(),
+              source.getUuidString(),
               moveSelection.getTargetNode().getUuidString());
 
           if (moveContainer(source, moveSelection)) {
             // consider move successful for now, and update selection criteria
             potentialTargets = updateTargetsAndSelectionCriteria(
                 potentialTargets, selectedTargets, moveSelection, source);
           }
-        }
-      }
-
-      // if not all underUtilized nodes have been selected, try to match
-      // withinThresholdUtilized nodes with underUtilized nodes
-      if (selectedTargets.size() < underUtilizedNodes.size()) {
-        potentialTargets.removeAll(selectedTargets);
-        Collections.reverse(withinThresholdUtilizedNodes);
-
-        for (DatanodeUsageInfo datanodeUsageInfo :
-            withinThresholdUtilizedNodes) {
-          if (!balancerRunning) {
-            return IterationResult.ITERATION_INTERRUPTED;
-          }
-          DatanodeDetails source = datanodeUsageInfo.getDatanodeDetails();
-          IterationResult result = checkConditionsForBalancing();
-          if (result != null) {
-            return result;
-          }
-
-          ContainerMoveSelection moveSelection =
-              matchSourceWithTarget(source, potentialTargets);
-          if (moveSelection != null) {
-            isMoveGenerated = true;
-            LOG.info("ContainerBalancer is trying to move container {} from " +
-                    "source datanode {} to target datanode {}",
-                moveSelection.getContainerID().toString(),
-                source.getUuidString(),
-                moveSelection.getTargetNode().getUuidString());
 
-            if (moveContainer(source, moveSelection)) {
-              // consider move successful for now, and update selection criteria
-              potentialTargets =
-                  updateTargetsAndSelectionCriteria(potentialTargets,
-                      selectedTargets, moveSelection, source);
-            }
-          }
+        } else {
+          // can not find any target for this source
+          srcDnSelectionCriteria.removeCandidateSourceDataNode(source);
         }
+
       }
+
       if (!isMoveGenerated) {
         //no move option is generated, so the cluster can not be
         //balanced any more, just stop iteration and exit
@@ -507,7 +477,7 @@ private void checkIterationMoveResults(Set<DatanodeDetails> selectedTargets) {
    * @return ContainerMoveSelection containing the selected target and container
    */
   private ContainerMoveSelection matchSourceWithTarget(
-      DatanodeDetails source, Collection<DatanodeDetails> potentialTargets) {
+      DatanodeDetails source, List<DatanodeUsageInfo> potentialTargets) {
     NavigableSet<ContainerID> candidateContainers =
         selectionCriteria.getCandidateContainers(source);
 
@@ -518,6 +488,23 @@ private ContainerMoveSelection matchSourceWithTarget(
       }
       return null;
     }
+
+    //if the utilization of the source data node becomes lower than lowerLimit
+    //after the container is moved out , then the container can not be
+    // a candidate one, and we should remove it from the candidateContainers.
+    candidateContainers.removeIf(c -> {
+      ContainerInfo cInfo;
+      try {
+        cInfo = containerManager.getContainer(c);
+      } catch (ContainerNotFoundException e) {
+        LOG.warn("Could not find container {} when " +
+            "be matched with a move target", c);
+        //remove this not found container
+        return true;
+      }
+      return !canSizeLeaveSource(source, cInfo.getUsedBytes());
+    });
+
     if (LOG.isDebugEnabled()) {
       LOG.debug("ContainerBalancer is finding suitable target for source " +
           "datanode {}", source.getUuidString());
@@ -629,8 +616,8 @@ private boolean moveContainer(DatanodeDetails source,
    * @param source           the source datanode
    * @return List of updated potential targets
    */
-  private List<DatanodeDetails> updateTargetsAndSelectionCriteria(
-      Collection<DatanodeDetails> potentialTargets,
+  private List<DatanodeUsageInfo> updateTargetsAndSelectionCriteria(
+      Collection<DatanodeUsageInfo> potentialTargets,
       Set<DatanodeDetails> selectedTargets,
       ContainerMoveSelection moveSelection, DatanodeDetails source) {
     // count source if it has not been involved in move earlier
@@ -649,7 +636,7 @@ private List<DatanodeDetails> updateTargetsAndSelectionCriteria(
     selectionCriteria.setSelectedContainers(selectedContainers);
 
     return potentialTargets.stream()
-        .filter(node -> sizeEnteringNode.get(node) <
+        .filter(node -> sizeEnteringNode.get(node.getDatanodeDetails()) <
             config.getMaxSizeEnteringTarget()).collect(Collectors.toList());
   }
 
@@ -708,8 +695,33 @@ boolean canSizeEnterTarget(DatanodeDetails target, long size) {
       //2 current usage of target datanode plus sizeEnteringAfterMove
       // is smaller than or equal to upperLimit
       return sizeEnteringAfterMove <= config.getMaxSizeEnteringTarget() &&
-           nodeManager.getUsageInfo(target)
-               .calculateUtilization(sizeEnteringAfterMove) <= upperLimit;
+           Double.compare(nodeManager.getUsageInfo(target)
+               .calculateUtilization(sizeEnteringAfterMove), upperLimit) <= 0;
+    }
+    return false;
+  }
+
+  /**
+   * Checks if specified size can leave a specified target datanode
+   * according to {@link ContainerBalancerConfiguration}
+   * "size.entering.target.max".
+   *
+   * @param source target datanode in which size is entering
+   * @param size   size in bytes
+   * @return true if size can leave, else false
+   */
+  boolean canSizeLeaveSource(DatanodeDetails source, long size) {
+    if (sizeLeavingNode.containsKey(source)) {
+      long sizeLeavingAfterMove = sizeLeavingNode.get(source) + size;
+      //size can be moved out of source datanode only when the following
+      //two condition are met.
+      //1 sizeLeavingAfterMove does not succeed the configured
+      // MaxSizeLeavingTarget
+      //2 after subtracting sizeLeavingAfterMove, the usage is bigger
+      // than or equal to lowerLimit
+      return sizeLeavingAfterMove <= config.getMaxSizeLeavingSource() &&
+          Double.compare(nodeManager.getUsageInfo(source)
+              .calculateUtilization(-sizeLeavingAfterMove), lowerLimit) >= 0;
     }
     return false;
   }
@@ -718,17 +730,10 @@ boolean canSizeEnterTarget(DatanodeDetails target, long size) {
    * Get potential targets for container move. Potential targets are under
    * utilized and within threshold utilized nodes.
    *
-   * @return A list of potential target DatanodeDetails.
+   * @return A list of potential target DatanodeUsageInfo.
    */
-  private List<DatanodeDetails> getPotentialTargets() {
-    List<DatanodeDetails> potentialTargets = new ArrayList<>(
-        underUtilizedNodes.size() + withinThresholdUtilizedNodes.size());
-
-    underUtilizedNodes
-        .forEach(node -> potentialTargets.add(node.getDatanodeDetails()));
-    withinThresholdUtilizedNodes
-        .forEach(node -> potentialTargets.add(node.getDatanodeDetails()));
-    return potentialTargets;
+  private List<DatanodeUsageInfo> getPotentialTargets() {
+    return underUtilizedNodes;
   }
 
   /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancerConfiguration.java

@@ -24,9 +24,7 @@
 import org.apache.hadoop.hdds.conf.ConfigTag;
 import org.apache.hadoop.hdds.conf.ConfigType;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.fs.DUFactory;
-import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.ContainerID;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.slf4j.Logger;
@@ -75,17 +73,15 @@ public final class ContainerBalancerConfiguration {
       defaultValue = "", tags = {ConfigTag.BALANCER}, description = "The " +
       "maximum size that can enter a target datanode in each " +
       "iteration while balancing. This is the sum of data from multiple " +
-      "sources. The default value is greater than the configured" +
-      " (or default) ozone.scm.container.size by 1GB.")
-  private long maxSizeEnteringTarget;
+      "sources. by default, we do not limit this value")
+  private long maxSizeEnteringTarget = Long.MAX_VALUE;
 
   @Config(key = "size.leaving.source.max", type = ConfigType.SIZE,
       defaultValue = "", tags = {ConfigTag.BALANCER}, description = "The " +
       "maximum size that can leave a source datanode in each " +
       "iteration while balancing. This is the sum of data moving to multiple " +
-      "targets. The default value is greater than the configured" +
-      " (or default) ozone.scm.container.size by 1GB.")
-  private long maxSizeLeavingSource;
+      "targets. by default, we do not limit this value")
+  private long maxSizeLeavingSource = Long.MAX_VALUE;;
 
   @Config(key = "idle.iterations", type = ConfigType.INT,
       defaultValue = "10", tags = {ConfigTag.BALANCER},
@@ -121,15 +117,6 @@ public ContainerBalancerConfiguration(OzoneConfiguration config) {
         "OzoneConfiguration should not be null.");
     this.ozoneConfiguration = config;
 
-    // maxSizeEnteringTarget and maxSizeLeavingSource should by default be
-    // greater than container size
-    long size = (long) ozoneConfiguration.getStorageSize(
-        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE,
-        ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE_DEFAULT, StorageUnit.GB) +
-        OzoneConsts.GB;
-    maxSizeEnteringTarget = size;
-    maxSizeLeavingSource = size;
-
     // balancing interval should be greater than DUFactory refresh period
     duConf = ozoneConfiguration.getObject(DUFactory.Conf.class);
     balancingInterval = duConf.getRefreshPeriod().toMillis() +