HDDS-15059. Shift streaming write sortDatanodes logic to OM

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java

@@ -17,7 +17,6 @@
 
 package org.apache.hadoop.hdds.scm.client;
 
-import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OM_NETWORK_TOPOLOGY_REFRESH_DURATION;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_OM_NETWORK_TOPOLOGY_REFRESH_DURATION_DEFAULT;
@@ -60,8 +59,7 @@ public ScmTopologyClient(
   }
 
   public NetworkTopology getClusterMap() {
-    return requireNonNull(cache.get(),
-        "ScmBlockLocationClient must have been initialized already.");
+    return cache.get();
   }
 
   public void start(ConfigurationSource conf) throws IOException {

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java

@@ -184,6 +184,18 @@ public class OmConfig extends ReconfigurableConfig {
   )
   private long followerReadLocalLeaseTimeMs;
 
+  @Config(key = "ozone.om.block.write.sort.datanodes.enabled",
+      defaultValue = "false",
+      type = ConfigType.BOOLEAN,
+      tags = {ConfigTag.OM, ConfigTag.PERFORMANCE},
+      description = "If true, OM sorts the streaming-write pipeline (nearest " +
+          "datanode first) locally using its cached cluster topology, instead " +
+          "of asking SCM to sort on every allocateBlock. Defaults to false so " +
+          "SCM performs the sort. Enable this to offload the sort from SCM " +
+          "when multiple OM services share a single SCM service."
+  )
+  private boolean sortDatanodesForWriteEnabled;
+
   public long getRatisBasedFinalizationTimeout() {
     return ratisBasedFinalizationTimeout;
   }
@@ -224,6 +236,10 @@ public void setAllowLeaderSkipLinearizableRead(boolean newValue) {
     allowLeaderSkipLinearizableRead = newValue;
   }
 
+  public boolean isSortDatanodesForWriteEnabled() {
+    return sortDatanodesForWriteEnabled;
+  }
+
   public boolean isFollowerReadLocalLeaseEnabled() {
     return followerReadLocalLeaseEnabled;
   }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java

@@ -22,6 +22,7 @@
 import static org.apache.hadoop.hdds.scm.net.NetConstants.ROOT_LEVEL;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.mockito.Mockito.mock;
 
 import com.google.common.collect.ImmutableMap;
@@ -176,6 +177,57 @@ private static void assertRackOrder(String rack, List<? extends DatanodeDetails>
     }
   }
 
+  @Test
+  public void sortDatanodesForWriteSortsRpcDeserializedPipeline() {
+    // Pipeline nodes arrive from SCM over RPC as deserialized DatanodeDetails
+    // with no topology linkage; the write sort must resolve them to OM's cluster
+    // map, otherwise every node is equidistant (MAX) and the order is random.
+    List<DatanodeDetails> rpcNodes = new ArrayList<>();
+    for (DatanodeDetails dn : nodeManager.getAllNodes()) {
+      rpcNodes.add(DatanodeDetails.getFromProtoBuf(dn.getProtoBufMessage()));
+    }
+    for (DatanodeDetails dn : nodeManager.getAllNodes()) {
+      // The client address is normally an IP, but the sort must resolve a client
+      // by either IP or hostname, so cover both.
+      List<? extends DatanodeDetails> byIp =
+          keyManager.sortDatanodesForWrite(rpcNodes, dn.getIpAddress(), om.getClusterMap());
+      assertEquals(dn, byIp.get(0),
+          "Source node should be sorted first for writes (IP client)");
+      assertRackOrder(dn.getNetworkLocation(), byIp);
+
+      List<? extends DatanodeDetails> byHostname =
+          keyManager.sortDatanodesForWrite(rpcNodes, dn.getHostName(), om.getClusterMap());
+      assertEquals(dn, byHostname.get(0),
+          "Source node should be sorted first for writes (hostname client)");
+      assertRackOrder(dn.getNetworkLocation(), byHostname);
+    }
+  }
+
+  @Test
+  public void sortDatanodesForWriteKeepsOrderForStaleTopology() {
+    List<DatanodeDetails> nodes = new ArrayList<>();
+    nodes.add(randomDatanodeDetails());
+    nodes.addAll(nodeManager.getAllNodes());
+
+    List<? extends DatanodeDetails> sorted =
+        keyManager.sortDatanodesForWrite(nodes, "edge0", om.getClusterMap());
+
+    assertSame(nodes, sorted,
+        "Pipeline order should be preserved when a node is missing from the OM topology");
+  }
+
+  @Test
+  public void sortDatanodesForWriteKeepsOrderWhenClientUnresolved() {
+    List<? extends DatanodeDetails> nodes = nodeManager.getAllNodes();
+    List<DatanodeDetails> original = new ArrayList<>(nodes);
+    // A client that resolves to no known rack must NOT trigger a shuffle.
+    String unresolved = nodes.get(0).getIpAddress() + "X";
+    List<? extends DatanodeDetails> result =
+        keyManager.sortDatanodesForWrite(nodes, unresolved, om.getClusterMap());
+    assertEquals(original, result,
+        "Write pipeline order must be preserved when client is unresolved");
+  }
+
   private String nodeAddress(DatanodeDetails dn) {
     boolean useHostname = config.getBoolean(
         HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java

@@ -23,6 +23,8 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
 import org.apache.hadoop.hdds.utils.BackgroundService;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.hdds.utils.db.TableIterator;
@@ -363,4 +365,27 @@ DeleteKeysResult getPendingDeletionSubFiles(long volumeId, long bucketId, OmKeyI
    * @return BackgroundService
    */
   CompactionService getCompactionService();
+
+  /**
+   * Sort the datanodes of a write pipeline by network-topology distance to the
+   * client, using OM's locally cached cluster map. Unlike the read-path sort,
+   * the original order is preserved when the client cannot be resolved, because
+   * the first node is used as the streaming-write primary.
+   *
+   * @param nodes the pipeline nodes to sort
+   * @param clientMachine client address (IP or hostname)
+   * @param clusterMap OM's cached cluster map used to resolve topology distance
+   * @return nodes sorted nearest-first, or the original {@code nodes} list
+   *     instance unchanged when sorting is skipped (client unresolved or stale
+   *     topology); callers may use reference equality to detect a skipped sort
+   */
+  List<? extends DatanodeDetails> sortDatanodesForWrite(
+      List<? extends DatanodeDetails> nodes, String clientMachine, NetworkTopology clusterMap);
+
+  /**
+   * @return true if OM should sort the streaming-write pipeline locally
+   *     ({@code ozone.om.block.write.sort.datanodes.enabled}); false to leave
+   *     the sort to SCM.
+   */
+  boolean isSortDatanodesForWriteEnabled();
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java

@@ -115,7 +115,6 @@
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersion;
 import org.apache.hadoop.fs.FileEncryptionInfo;
-import org.apache.hadoop.hdds.HddsConfigKeys;
 import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -124,6 +123,7 @@
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
 import org.apache.hadoop.hdds.scm.net.InnerNode;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
 import org.apache.hadoop.hdds.scm.net.Node;
 import org.apache.hadoop.hdds.scm.net.NodeImpl;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -2186,7 +2186,12 @@ private void sortDatanodes(String clientMachine, List<OmKeyInfo> keyInfos) {
           List<? extends DatanodeDetails> sortedNodes = sortedPipelines.get(uuidSet);
           if (sortedNodes == null) {
             sortedNodes = sortDatanodes(nodes, clientMachine);
-            if (sortedNodes != null) {
+            // Cache only a freshly sorted order, not an input list returned
+            // unchanged when no sort happens: that order is per-pipeline and must
+            // not be reused for another pipeline with the same node set. The read
+            // sort always returns a new list, so this never skips caching here; it
+            // keeps the pattern identical to the write path.
+            if (sortedNodes != null && sortedNodes != nodes) {
               sortedPipelines.put(uuidSet, sortedNodes);
             }
           } else if (LOG.isDebugEnabled()) {
@@ -2204,32 +2209,87 @@ private void sortDatanodes(String clientMachine, List<OmKeyInfo> keyInfos) {
   @VisibleForTesting
   public List<? extends DatanodeDetails> sortDatanodes(List<? extends DatanodeDetails> nodes,
                                              String clientMachine) {
-    final Node client = getClientNode(clientMachine, nodes);
-    return ozoneManager.getClusterMap()
-        .sortByDistanceCost(client, nodes, nodes.size());
+    final NetworkTopology clusterMap = ozoneManager.getClusterMap();
+    final Node client = getClientNode(clientMachine, nodes, clusterMap);
+    return clusterMap.sortByDistanceCost(client, nodes, nodes.size());
+  }
+
+  @Override
+  public List<? extends DatanodeDetails> sortDatanodesForWrite(
+      List<? extends DatanodeDetails> nodes, String clientMachine, NetworkTopology clusterMap) {
+    Preconditions.checkArgument(!StringUtils.isEmpty(clientMachine),
+        "clientMachine is empty");
+    Preconditions.checkNotNull(clusterMap, "clusterMap is null");
+    return captureLatencyNs(
+        metrics.getAllocateBlockSortDatanodesLatencyNs(), () -> {
+          final Node client = getClientNode(clientMachine, nodes, clusterMap);
+          if (client == null) {
+            // Preserve pipeline order for writes: the first node is the write
+            // primary, so do not shuffle when the client cannot be resolved.
+            return nodes;
+          }
+          return sortByClusterMapDistance(clusterMap, client, nodes);
+        });
+  }
+
+  @Override
+  public boolean isSortDatanodesForWriteEnabled() {
+    return ozoneManager.getConfig().isSortDatanodesForWriteEnabled();
+  }
+
+  /**
+   * Sort a pipeline's nodes by topology distance to the client. The nodes come
+   * from SCM over RPC, so they are deserialized {@link DatanodeDetails} with no
+   * parent/level: the topology treats them as unknown (distance
+   * {@link Integer#MAX_VALUE}) and the order comes out random. Look each node
+   * up in OM's cluster map to get the topology-linked instance, sort those,
+   * then map the order back to the original nodes.
+   */
+  private List<? extends DatanodeDetails> sortByClusterMapDistance(
+      NetworkTopology clusterMap, Node client,
+      List<? extends DatanodeDetails> nodes) {
+    final List<Node> topologyNodes = new ArrayList<>(nodes.size());
+    final Map<String, DatanodeDetails> nodeByPath = new HashMap<>();
+    for (DatanodeDetails node : nodes) {
+      final Node resolved = clusterMap.getNode(node.getNetworkFullPath());
+      if (resolved == null) {
+        return nodes;
+      }
+      topologyNodes.add(resolved);
+      nodeByPath.put(resolved.getNetworkFullPath(), node);
+    }
+    final List<Node> sorted =
+        clusterMap.sortByDistanceCost(client, topologyNodes, topologyNodes.size());
+    final List<DatanodeDetails> result = new ArrayList<>(sorted.size());
+    for (Node node : sorted) {
+      result.add(nodeByPath.get(node.getNetworkFullPath()));
+    }
+    return result;
   }
 
   private Node getClientNode(String clientMachine,
-                             List<? extends DatanodeDetails> nodes) {
-    List<DatanodeDetails> matchingNodes = new ArrayList<>();
-    boolean useHostname = ozoneManager.getConfiguration().getBoolean(
-        HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME,
-        HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT);
+      List<? extends DatanodeDetails> nodes, NetworkTopology clusterMap) {
     for (DatanodeDetails node : nodes) {
-      if ((useHostname ? node.getHostName() : node.getIpAddress()).equals(
-          clientMachine)) {
-        matchingNodes.add(node);
+      // Match by either IP or hostname, like SCM's getNodesByAddress. clientMachine
+      // may be a hostname on the read path; the streaming-write remoteAddress is
+      // typically an IP. Matching both covers use.datanode.hostname either way.
+      if (clientMachine.equals(node.getIpAddress())
+          || clientMachine.equals(node.getHostName())) {
+        // The pipeline nodes are RPC-deserialized and not linked into OM's
+        // cluster map; prefer the map's instance so distance can be computed.
+        final Node resolved = clusterMap.getNode(node.getNetworkFullPath());
+        return resolved != null ? resolved : node;
       }
     }
-    return !matchingNodes.isEmpty() ? matchingNodes.get(0) :
-        getOtherNode(clientMachine);
+    return getOtherNode(clientMachine, clusterMap);
   }
 
-  private Node getOtherNode(String clientMachine) {
+  private Node getOtherNode(String clientMachine,
+                            NetworkTopology clusterMap) {
     try {
       String clientLocation = resolveNodeLocation(clientMachine);
       if (clientLocation != null) {
-        Node rack = ozoneManager.getClusterMap().getNode(clientLocation);
+        Node rack = clusterMap.getNode(clientLocation);
         if (rack instanceof InnerNode) {
           return new NodeImpl(clientMachine, clientLocation,
               (InnerNode) rack, rack.getLevel() + 1,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java

@@ -67,6 +67,9 @@ public class OMPerformanceMetrics {
   @Metric(about = "Sort datanodes latency in getKeyInfo")
   private MutableRate getKeyInfoSortDatanodesLatencyNs;
 
+  @Metric(about = "Sort datanodes latency in allocateBlock (streaming write)")
+  private MutableRate allocateBlockSortDatanodesLatencyNs;
+
   @Metric(about = "resolveBucketLink latency in getKeyInfo")
   private MutableRate getKeyInfoResolveBucketLatencyNs;
 
@@ -246,6 +249,10 @@ MutableRate getGetKeyInfoSortDatanodesLatencyNs() {
     return getKeyInfoSortDatanodesLatencyNs;
   }
 
+  MutableRate getAllocateBlockSortDatanodesLatencyNs() {
+    return allocateBlockSortDatanodesLatencyNs;
+  }
+
   public void setForceContainerCacheRefresh(boolean value) {
     forceContainerCacheRefresh.add(value ? 1L : 0L);
   }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.om;
 
 import static java.nio.charset.StandardCharsets.UTF_8;
+import static java.util.Objects.requireNonNull;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_DEFAULT;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED;
@@ -1342,6 +1343,11 @@ public void setScmTopologyClient(
   }
 
   public NetworkTopology getClusterMap() {
+    return requireNonNull(getClusterMapAllowNull(),
+        "OM topology cache has not been initialized yet.");
+  }
+
+  public NetworkTopology getClusterMapAllowNull() {
     return scmTopologyClient.getClusterMap();
   }