HDDS-5602. make it configurable to choose the nearest one as the target in the candidates according to networkTopology

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/FindTargetGreedy.java → hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/AbstractFindTargetGreedy.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.hdds.scm.container.balancer;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
 import org.apache.hadoop.hdds.scm.PlacementPolicy;
@@ -29,70 +30,66 @@
 import org.apache.hadoop.hdds.scm.node.DatanodeUsageInfo;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
 import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.TreeSet;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
 /**
- * Find a target giving preference to more under-utilized nodes.
+ * Find a target for a source datanode with greedy strategy.
  */
-public class FindTargetGreedy implements FindTargetStrategy {
-  private static final Logger LOG =
-      LoggerFactory.getLogger(FindTargetGreedy.class);
-
+public abstract class AbstractFindTargetGreedy implements FindTargetStrategy {
+  private Logger logger;
   private ContainerManager containerManager;
   private PlacementPolicy placementPolicy;
   private Map<DatanodeDetails, Long> sizeEnteringNode;
   private NodeManager nodeManager;
   private ContainerBalancerConfiguration config;
   private Double upperLimit;
-  private TreeSet<DatanodeUsageInfo> potentialTargets;
+  private Collection<DatanodeUsageInfo> potentialTargets;
 
-  public FindTargetGreedy(
+  protected AbstractFindTargetGreedy(
       ContainerManager containerManager,
       PlacementPolicy placementPolicy,
       NodeManager nodeManager) {
     sizeEnteringNode = new HashMap<>();
     this.containerManager = containerManager;
     this.placementPolicy = placementPolicy;
     this.nodeManager = nodeManager;
+  }
 
-    potentialTargets = new TreeSet<>((a, b) -> {
-      double currentUsageOfA = a.calculateUtilization(
-          sizeEnteringNode.get(a.getDatanodeDetails()));
-      double currentUsageOfB = b.calculateUtilization(
-          sizeEnteringNode.get(b.getDatanodeDetails()));
-      int ret = Double.compare(currentUsageOfA, currentUsageOfB);
-      if (ret != 0) {
-        return ret;
-      }
-      UUID uuidA = a.getDatanodeDetails().getUuid();
-      UUID uuidB = b.getDatanodeDetails().getUuid();
-      return uuidA.compareTo(uuidB);
-    });
+  protected void setLogger(Logger log) {
+    logger = log;
+  }
+
+  protected void setPotentialTargets(Collection<DatanodeUsageInfo> pt) {
+    potentialTargets = pt;
   }
 
   private void setUpperLimit(Double upperLimit){
     this.upperLimit = upperLimit;
   }
 
-  private void setPotentialTargets(
-      List<DatanodeUsageInfo> potentialTargetDataNodes) {
-    sizeEnteringNode.clear();
-    potentialTargetDataNodes.forEach(
-        p -> sizeEnteringNode.put(p.getDatanodeDetails(), 0L));
-    potentialTargets.clear();
-    potentialTargets.addAll(potentialTargetDataNodes);
+  protected int compareByUsage(DatanodeUsageInfo a, DatanodeUsageInfo b) {
+    double currentUsageOfA = a.calculateUtilization(
+        sizeEnteringNode.get(a.getDatanodeDetails()));
+    double currentUsageOfB = b.calculateUtilization(
+        sizeEnteringNode.get(b.getDatanodeDetails()));
+    int ret = Double.compare(currentUsageOfA, currentUsageOfB);
+    if (ret != 0) {
+      return ret;
+    }
+    UUID uuidA = a.getDatanodeDetails().getUuid();
+    UUID uuidB = b.getDatanodeDetails().getUuid();
+    return uuidA.compareTo(uuidB);
   }
 
   private void setConfiguration(ContainerBalancerConfiguration conf) {
-    this.config = conf;
+    config = conf;
   }
 
   /**
@@ -109,6 +106,7 @@ private void setConfiguration(ContainerBalancerConfiguration conf) {
   @Override
   public ContainerMoveSelection findTargetForContainerMove(
       DatanodeDetails source, Set<ContainerID> candidateContainers) {
+    sortTargetForSource(source);
     for (DatanodeUsageInfo targetInfo : potentialTargets) {
       DatanodeDetails target = targetInfo.getDatanodeDetails();
       for (ContainerID container : candidateContainers) {
@@ -118,7 +116,7 @@ public ContainerMoveSelection findTargetForContainerMove(
           replicas = containerManager.getContainerReplicas(container);
           containerInfo = containerManager.getContainer(container);
         } catch (ContainerNotFoundException e) {
-          LOG.warn("Could not get Container {} from Container Manager for " +
+          logger.warn("Could not get Container {} from Container Manager for " +
               "obtaining replicas in Container Balancer.", container, e);
           continue;
         }
@@ -132,8 +130,8 @@ public ContainerMoveSelection findTargetForContainerMove(
         }
       }
     }
-    LOG.info("Container Balancer could not find a target for source datanode " +
-        "{}", source.getUuidString());
+    logger.info("Container Balancer could not find a target for " +
+        "source datanode {}", source.getUuidString());
     return null;
   }
 
@@ -153,7 +151,7 @@ private boolean containerMoveSatisfiesPlacementPolicy(
     try {
       containerInfo = containerManager.getContainer(containerID);
     } catch (ContainerNotFoundException e) {
-      LOG.warn("Could not get Container {} from Container Manager while " +
+      logger.warn("Could not get Container {} from Container Manager while " +
           "checking if container move satisfies placement policy in " +
           "Container Balancer.", containerID.toString(), e);
       return false;
@@ -212,7 +210,7 @@ public void increaseSizeEntering(DatanodeDetails target, long size) {
       }
       return;
     }
-    LOG.warn("Cannot find {} in the candidates target nodes",
+    logger.warn("Cannot find {} in the candidates target nodes",
         target.getUuid());
   }
 
@@ -225,6 +223,23 @@ public void reInitialize(List<DatanodeUsageInfo> potentialDataNodes,
                            Double upLimit) {
     setConfiguration(conf);
     setUpperLimit(upLimit);
-    setPotentialTargets(potentialDataNodes);
+    sizeEnteringNode.clear();
+    potentialDataNodes.forEach(
+        p -> sizeEnteringNode.put(p.getDatanodeDetails(), 0L));
+    potentialTargets.clear();
+    potentialTargets.addAll(potentialDataNodes);
+  }
+
+  @VisibleForTesting
+  public Collection<DatanodeUsageInfo> getPotentialTargets() {
+    return potentialTargets;
   }
+
+  /**
+   * sort potentialTargets for specified source datanode according to
+   * network topology if enabled.
+   * @param source the specified source datanode
+   */
+  @VisibleForTesting
+  public abstract void sortTargetForSource(DatanodeDetails source);
 }

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

@@ -30,6 +30,7 @@
 import org.apache.hadoop.hdds.scm.container.ReplicationManager;
 import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeStat;
 import org.apache.hadoop.hdds.scm.ha.SCMContext;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
 import org.apache.hadoop.hdds.scm.node.DatanodeUsageInfo;
 import org.apache.hadoop.hdds.scm.node.NodeManager;
 import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
@@ -85,6 +86,8 @@ public class ContainerBalancer {
   private long clusterUsed;
   private long clusterRemaining;
   private double clusterAvgUtilisation;
+  private PlacementPolicy placementPolicy;
+  private NetworkTopology networkTopology;
   private double upperLimit;
   private double lowerLimit;
   private volatile boolean balancerRunning;
@@ -115,6 +118,7 @@ public ContainerBalancer(
       ReplicationManager replicationManager,
       OzoneConfiguration ozoneConfiguration,
       final SCMContext scmContext,
+      NetworkTopology networkTopology,
       PlacementPolicy placementPolicy) {
     this.nodeManager = nodeManager;
     this.containerManager = containerManager;
@@ -129,10 +133,10 @@ public ContainerBalancer(
     this.underUtilizedNodes = new ArrayList<>();
     this.withinThresholdUtilizedNodes = new ArrayList<>();
     this.unBalancedNodes = new ArrayList<>();
+    this.placementPolicy = placementPolicy;
+    this.networkTopology = networkTopology;
 
     this.lock = new ReentrantLock();
-    findTargetStrategy = new FindTargetGreedy(
-        containerManager, placementPolicy, nodeManager);
     findSourceStrategy = new FindSourceGreedy(nodeManager);
   }
 
@@ -179,6 +183,13 @@ private void balance() {
     this.maxDatanodesRatioToInvolvePerIteration =
         config.getMaxDatanodesRatioToInvolvePerIteration();
     this.maxSizeToMovePerIteration = config.getMaxSizeToMovePerIteration();
+    if (config.getNetworkTopologyEnable()) {
+      findTargetStrategy = new FindTargetGreedyByNetworkTopology(
+          containerManager, placementPolicy, nodeManager, networkTopology);
+    } else {
+      findTargetStrategy = new FindTargetGreedyByUsageInfo(containerManager,
+          placementPolicy, nodeManager);
+    }
     for (int i = 0; i < idleIteration && balancerRunning; i++) {
       // stop balancing if iteration is not initialized
       if (!initializeIteration()) {
@@ -367,11 +378,6 @@ private IterationResult doIteration() {
     try {
       // match each overUtilized node with a target
       while (true) {
-        DatanodeDetails source =
-            findSourceStrategy.getNextCandidateSourceDataNode();
-        if (source == null) {
-          break;
-        }
         if (!isBalancerRunning()) {
           return IterationResult.ITERATION_INTERRUPTED;
         }
@@ -381,6 +387,12 @@ private IterationResult doIteration() {
           return result;
         }
 
+        DatanodeDetails source =
+            findSourceStrategy.getNextCandidateSourceDataNode();
+        if (source == null) {
+          break;
+        }
+
         ContainerMoveSelection moveSelection = matchSourceWithTarget(source);
         if (moveSelection != null) {
           isMoveGenerated = true;

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

@@ -118,12 +118,19 @@ public final class ContainerBalancerConfiguration {
   private String includeNodes = "";
 
   @Config(key = "exclude.datanodes", type = ConfigType.STRING, defaultValue =
-      "", tags = ConfigTag.BALANCER, description = "A list of Datanode " +
+      "", tags = {ConfigTag.BALANCER}, description = "A list of Datanode " +
       "hostnames or ip addresses separated by commas. The Datanodes specified" +
       " in this list are excluded from balancing. This configuration is empty" +
       " by default.")
   private String excludeNodes = "";
 
+  @Config(key = "move.networkTopology.enable", type = ConfigType.BOOLEAN,
+      defaultValue = "false", tags = {ConfigTag.BALANCER},
+      description = "whether to take network topology into account when " +
+          "selecting a target for a source. " +
+          "This configuration is false by default.")
+  private boolean networkTopologyEnable = false;
+
   private DUFactory.Conf duConf;
 
   /**
@@ -196,6 +203,24 @@ public void setIdleIteration(int count) {
     this.idleIterations = count;
   }
 
+  /**
+   * Get the NetworkTopologyEnable value for Container Balancer.
+   *
+   * @return the boolean value of networkTopologyEnable
+   */
+  public Boolean getNetworkTopologyEnable() {
+    return networkTopologyEnable;
+  }
+
+  /**
+   * Set the NetworkTopologyEnable value for Container Balancer.
+   *
+   * @param enable the boolean value to be set to networkTopologyEnable
+   */
+  public void setNetworkTopologyEnable(Boolean enable) {
+    networkTopologyEnable = enable;
+  }
+
   /**
    * Gets the ratio of maximum number of datanodes that will be involved in
    * balancing by Container Balancer in one iteration to the total number of

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

@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.container.balancer;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
+import org.apache.hadoop.hdds.scm.container.ContainerManager;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
+import org.apache.hadoop.hdds.scm.node.DatanodeUsageInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.slf4j.LoggerFactory;
+
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * an implementation of FindTargetGreedy, which will always select the
+ * target with the shortest distance according to network topology
+ * distance to the give source datanode.
+ */
+public class FindTargetGreedyByNetworkTopology
+    extends AbstractFindTargetGreedy {
+
+  private NetworkTopology networkTopology;
+  private List potentialTargets;
+
+  public FindTargetGreedyByNetworkTopology(
+      ContainerManager containerManager,
+      PlacementPolicy placementPolicy,
+      NodeManager nodeManager,
+      NetworkTopology networkTopology) {
+    super(containerManager, placementPolicy, nodeManager);
+    setLogger(LoggerFactory.getLogger(FindTargetGreedyByNetworkTopology.class));
+    potentialTargets = new LinkedList<>();
+    setPotentialTargets(potentialTargets);
+    this.networkTopology = networkTopology;
+  }
+
+  /**
+   * sort potentialTargets for specified source datanode according to
+   * network topology.
+   * @param source the specified source datanode
+   */
+  @VisibleForTesting
+  public void sortTargetForSource(DatanodeDetails source) {
+    Collections.sort(potentialTargets,
+        (DatanodeUsageInfo da, DatanodeUsageInfo db) -> {
+        DatanodeDetails a = da.getDatanodeDetails();
+        DatanodeDetails b = db.getDatanodeDetails();
+        // sort by network topology first
+        int distanceToA = networkTopology.getDistanceCost(source, a);
+        int distanceToB = networkTopology.getDistanceCost(source, b);
+        if (distanceToA != distanceToB) {
+          return distanceToA - distanceToB;
+        }
+        // if distance to source is equal , sort by usage
+        return compareByUsage(da, db);
+      });
+  }
+}