HDDS-15093. Make RackScatter intra-rack placement capacity-aware

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackScatter.java

@@ -33,6 +33,7 @@
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
 import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
+import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.net.NetworkTopology;
 import org.apache.hadoop.hdds.scm.net.Node;
@@ -445,7 +446,7 @@ private Node chooseNode(String scope, List<Node> excludedNodes,
       }
       Node node = null;
       try {
-        node = networkTopology.chooseRandom(scope, excludedNodes);
+        node = chooseLessUtilizedNode(scope, excludedNodes);
       } catch (Exception e) {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Error while choosing Node: Scope: {}, Excluded Nodes: " +
@@ -481,6 +482,39 @@ private Node chooseNode(String scope, List<Node> excludedNodes,
     }
   }
 
+  /**
+   * Pick two distinct candidate nodes within the rack and return the one with
+   * lower space utilization, so a nearly-full datanode is not chosen as often
+   * as an emptier peer in the same rack.
+   *
+   * @param scope - the rack we are searching nodes under
+   * @param excludedNodes - list of the datanodes to exclude. Can be null.
+   * @return the chosen datanode, or null if none is available.
+   */
+  private Node chooseLessUtilizedNode(String scope, List<Node> excludedNodes) {
+    Node first = networkTopology.chooseRandom(scope, excludedNodes);
+    if (first == null) {
+      return null;
+    }
+    // Exclude the first pick so the second candidate is a distinct node.
+    // Otherwise a small rack often draws the same node twice and the capacity
+    // comparison below is skipped.
+    List<Node> secondExcludedNodes = new ArrayList<>(excludedNodes);
+    secondExcludedNodes.add(first);
+    Node second = networkTopology.chooseRandom(scope, secondExcludedNodes);
+    if (second == null) {
+      return first;
+    }
+    SCMNodeMetric firstMetric =
+        getNodeManager().getNodeStat((DatanodeDetails) first);
+    SCMNodeMetric secondMetric =
+        getNodeManager().getNodeStat((DatanodeDetails) second);
+    if (firstMetric == null || secondMetric == null) {
+      return first;
+    }
+    return firstMetric.isGreater(secondMetric.get()) ? second : first;
+  }
+
   /**
    * For EC placement policy, desired rack count would be equal to the num of
    * Replicas.

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestSCMContainerPlacementRackScatter.java

@@ -40,8 +40,10 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -56,6 +58,7 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.StorageReportProto;
 import org.apache.hadoop.hdds.scm.ContainerPlacementStatus;
 import org.apache.hadoop.hdds.scm.HddsTestUtils;
+import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.net.NetConstants;
 import org.apache.hadoop.hdds.scm.net.NetworkTopology;
@@ -254,6 +257,22 @@ private void createMocksAndUpdateStorageReports(int datanodeCount) {
       DatanodeInfo di = invocation.getArgument(0);
       return di.getStorageReports().stream().anyMatch(r -> r.getRemaining() > 1L);
     });
+    when(nodeManager.getNodeStat(any(DatanodeDetails.class))).thenAnswer(invocation -> {
+      DatanodeDetails dd = invocation.getArgument(0);
+      DatanodeInfo di = dnInfos.stream()
+          .filter(d -> d.getID().equals(dd.getID()))
+          .findFirst().orElse(null);
+      if (di == null) {
+        return null;
+      }
+      long capacity = di.getStorageReports().stream()
+          .mapToLong(StorageReportProto::getCapacity).sum();
+      long used = di.getStorageReports().stream()
+          .mapToLong(StorageReportProto::getScmUsed).sum();
+      long remaining = di.getStorageReports().stream()
+          .mapToLong(StorageReportProto::getRemaining).sum();
+      return new SCMNodeMetric(capacity, used, remaining, 0, remaining, 0);
+    });
 
     // create placement policy instances
     policy = new SCMContainerPlacementRackScatter(
@@ -900,6 +919,34 @@ public void testAllNodesOnRackExcludedReducesRackCount2()
     assertEquals(1, chosenNodes.size());
   }
 
+  /**
+   * Within a single rack holding an emptier and a fuller datanode, the intra
+   * rack selection should prefer the less utilized node instead of choosing
+   * uniformly at random.
+   */
+  @Test
+  public void chooseNodeWithinRackPrefersLessUtilized() throws SCMException {
+    // Single rack with two datanodes.
+    setup(2, 2);
+    updateStorageInDatanode(0, 10, 90);   // emptier node
+    updateStorageInDatanode(1, 90, 10);   // fuller node
+
+    Map<DatanodeDetails, Integer> selectedCount = new HashMap<>();
+    selectedCount.put(datanodes.get(0), 0);
+    selectedCount.put(datanodes.get(1), 0);
+
+    for (int i = 0; i < 1000; i++) {
+      List<DatanodeDetails> chosen = policy.chooseDatanodes(
+          new ArrayList<>(), new ArrayList<>(), null, 1, 0, 0);
+      assertEquals(1, chosen.size());
+      DatanodeDetails dn = chosen.get(0);
+      selectedCount.put(dn, selectedCount.get(dn) + 1);
+    }
+
+    assertThat(selectedCount.get(datanodes.get(0)))
+        .isGreaterThan(selectedCount.get(datanodes.get(1)));
+  }
+
   private int getRackSize(List<DatanodeDetails>... datanodeDetails) {
     Set<Node> racks = new HashSet<>();
     for (List<DatanodeDetails> list : datanodeDetails) {