diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java index d595bd6e0958..7abd2c9c0161 100644 --- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/client/ScmTopologyClient.java +++ b/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 { diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java index b8a60f9bcdb1..c981afd414d8 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/OmConfig.java +++ b/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; } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java index cfce524537fc..a206276ce8fd 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestOMSortDatanodes.java +++ b/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 } } + @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 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 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 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 nodes = new ArrayList<>(); + nodes.add(randomDatanodeDetails()); + nodes.addAll(nodeManager.getAllNodes()); + + List 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 nodes = nodeManager.getAllNodes(); + List 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 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, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java index fdf4172c71b2..320536dab2e4 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManager.java +++ b/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 sortDatanodesForWrite( + List 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(); } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java index 08b6d6abbf18..658f78f68d62 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/KeyManagerImpl.java +++ b/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 keyInfos) { List 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 keyInfos) { @VisibleForTesting public List sortDatanodes(List 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 sortDatanodesForWrite( + List 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 sortByClusterMapDistance( + NetworkTopology clusterMap, Node client, + List nodes) { + final List topologyNodes = new ArrayList<>(nodes.size()); + final Map 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 sorted = + clusterMap.sortByDistanceCost(client, topologyNodes, topologyNodes.size()); + final List result = new ArrayList<>(sorted.size()); + for (Node node : sorted) { + result.add(nodeByPath.get(node.getNetworkFullPath())); + } + return result; } private Node getClientNode(String clientMachine, - List nodes) { - List matchingNodes = new ArrayList<>(); - boolean useHostname = ozoneManager.getConfiguration().getBoolean( - HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME, - HddsConfigKeys.HDDS_DATANODE_USE_DN_HOSTNAME_DEFAULT); + List 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, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java index e142e4cfa715..3f965ba025e6 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OMPerformanceMetrics.java +++ b/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); } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java index 1910d92e9690..13198fbcbc30 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java +++ b/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(); } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java index d12b8fa05257..4e2253315b8b 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java @@ -57,11 +57,14 @@ import org.apache.hadoop.hdds.client.ContainerBlockID; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.BlockTokenSecretProto.AccessModeProto; import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock; import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; import org.apache.hadoop.hdds.scm.exceptions.SCMException; +import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.hdds.utils.db.cache.CacheKey; import org.apache.hadoop.hdds.utils.db.cache.CacheValue; @@ -69,6 +72,7 @@ import org.apache.hadoop.ozone.OmUtils; import org.apache.hadoop.ozone.OzoneAcl; import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.OMMetadataManager; import org.apache.hadoop.ozone.om.OmConfig; import org.apache.hadoop.ozone.om.OzoneManager; @@ -189,15 +193,38 @@ protected List allocateBlock( UserInfo userInfo, OzoneManager ozoneManager) throws IOException { final long scmBlockSize = ozoneManager.getScmBlockSize(); + final KeyManager keyManager = ozoneManager.getKeyManager(); int dataGroupSize = replicationConfig instanceof ECReplicationConfig ? ((ECReplicationConfig) replicationConfig).getData() : 1; final int numBlocks = (int) Math.min(ozoneManager.getPreallocateBlocksMax(), (requestedSize - 1) / (scmBlockSize * dataGroupSize) + 1); - String clientMachine = ""; - if (shouldSortDatanodes) { - clientMachine = userInfo.getRemoteAddress(); + final String scmClientMachine; + final String omClientMachine; + // Sorted order cached by datanode set so blocks whose pipelines share the + // same datanodes are sorted once (mirrors the read path's caching). Keyed by + // the UUID set so it is order-insensitive and dedups across pipelines. + final Map, List> sortedByNodes; + final String remoteAddress = userInfo.getRemoteAddress(); + final NetworkTopology clusterMap = shouldSortDatanodes + && keyManager.isSortDatanodesForWriteEnabled() + ? ozoneManager.getClusterMapAllowNull() : null; + if (!shouldSortDatanodes) { + scmClientMachine = ""; + omClientMachine = ""; + sortedByNodes = null; + } else if (clusterMap != null && !remoteAddress.isEmpty()) { + // Sort in OM: SCM skips sorting (empty machine), OM sorts by remoteAddress. + scmClientMachine = ""; + omClientMachine = remoteAddress; + sortedByNodes = new HashMap<>(); + } else { + // Sort in SCM (or keep order when remoteAddress is empty, since SCM skips + // sorting for an empty client machine). + scmClientMachine = remoteAddress; + omClientMachine = ""; + sortedByNodes = null; } List locationInfos = new ArrayList<>(numBlocks); @@ -205,7 +232,7 @@ protected List allocateBlock( final List allocatedBlocks; try { allocatedBlocks = ozoneManager.getScmClient().getBlockClient().allocateBlock( - scmBlockSize, numBlocks, replicationConfig, ozoneManager.getOMServiceId(), excludeList, clientMachine); + scmBlockSize, numBlocks, replicationConfig, ozoneManager.getOMServiceId(), excludeList, scmClientMachine); } catch (SCMException ex) { ozoneManager.getMetrics().incNumBlockAllocateCallFails(); if (ex.getResult() == SCMException.ResultCodes.SAFE_MODE_EXCEPTION) { @@ -216,11 +243,30 @@ protected List allocateBlock( } for (AllocatedBlock allocatedBlock : allocatedBlocks) { BlockID blockID = new BlockID(allocatedBlock.getBlockID()); + Pipeline pipeline = allocatedBlock.getPipeline(); + if (sortedByNodes != null) { + final List nodes = pipeline.getNodes(); + final Set uuidSet = nodes.stream() + .map(DatanodeDetails::getUuidString).collect(Collectors.toSet()); + List sorted = sortedByNodes.get(uuidSet); + if (sorted == null) { + sorted = keyManager.sortDatanodesForWrite(nodes, omClientMachine, clusterMap); + // Cache only a freshly sorted order, not an input list returned + // unchanged when the client is unresolved: that order is per-pipeline + // and must not be reused for another pipeline with the same node set. + if (sorted != nodes) { + sortedByNodes.put(uuidSet, sorted); + } + } + if (!Objects.equals(sorted, pipeline.getNodesInOrder())) { + pipeline = pipeline.copyWithNodesInOrder(sorted); + } + } OmKeyLocationInfo.Builder builder = new OmKeyLocationInfo.Builder() .setBlockID(blockID) .setLength(scmBlockSize) .setOffset(0) - .setPipeline(allocatedBlock.getPipeline()); + .setPipeline(pipeline); if (ozoneManager.isGrpcBlockTokenEnabled()) { final Token token = ozoneManager.getBlockTokenSecretManager().generateToken( remoteUser, blockID, READ_WRITE, scmBlockSize); diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java index 8f720deaf1bc..a584fd7b1b25 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/key/TestOMAllocateBlockRequest.java @@ -20,13 +20,38 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import jakarta.annotation.Nonnull; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.UUID; +import org.apache.hadoop.hdds.client.ContainerBlockID; import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.client.StandaloneReplicationConfig; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor; +import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock; +import org.apache.hadoop.hdds.scm.container.common.helpers.ExcludeList; +import org.apache.hadoop.hdds.scm.net.NetworkTopology; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.KeyManager; import org.apache.hadoop.ozone.om.helpers.BucketLayout; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; @@ -36,7 +61,9 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.AllocateBlockRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.KeyArgs; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; /** * Tests OMAllocateBlockRequest class. @@ -225,6 +252,242 @@ protected OMRequest doPreExecute(OMRequest originalOMRequest) return modifiedOmRequest; } + @Test + public void testAllocateBlockSendsClientMachineToScmWhenFlagOff() throws Exception { + // Flag off (default): OM must NOT sort; SCM receives the real client address + // so it performs the sort. + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(false); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + } + + @Test + public void testAllocateBlockDoesNotSendClientMachineToScm() throws Exception { + // OM now sorts the write pipeline locally, so SCM must receive an empty + // clientMachine even when the client requests sorted datanodes. + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("", clientMachine.getValue()); + } + + @Test + public void testAllocateBlockFallsBackToScmWhenTopologyUnavailable() throws Exception { + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequestWithSort("1.2.3.4")); + request.preExecute(ozoneManager); + + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + verify(scmBlockLocationProtocol).allocateBlock(anyLong(), anyInt(), any(), + any(), any(), clientMachine.capture()); + assertEquals("1.2.3.4", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + } + + @Test + public void testAllocateBlockSortsSharedPipelineOnce() throws Exception { + // Two blocks on the same 3-node pipeline must be sorted once, and the + // sorted order must land in every block's pipeline. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .build(); + AllocatedBlock.Builder blockBuilder = + new AllocatedBlock.Builder().setPipeline(pipeline); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), anyString())).thenAnswer(inv -> { + int num = inv.getArgument(1); + List blocks = new ArrayList<>(num); + for (int i = 0; i < num; i++) { + blockBuilder.setContainerBlockID( + new ContainerBlockID(CONTAINER_ID + i, LOCAL_ID + i)); + blocks.add(blockBuilder.build()); + } + return blocks; + }); + + List sortedOrder = new ArrayList<>(nodes); + Collections.reverse(sortedOrder); + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + .thenAnswer(inv -> sortedOrder); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + // requestedSize spans two scmBlockSize blocks on the same pipeline. + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), 2 * scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + // Sorted once for the shared pipeline... + verify(mockKeyManager, times(1)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + // ...and the sorted order is applied to every block's pipeline. + assertEquals(2, locations.size()); + for (OmKeyLocationInfo location : locations) { + assertEquals(sortedOrder, location.getPipeline().getNodesInOrder()); + } + } + + @Test + public void testAllocateBlockKeepsPerPipelineOrderWhenSortSkipped() throws Exception { + // Two pipelines share the same datanode set but in a different order. When + // the sort is skipped (sortDatanodesForWrite returns the input unchanged), + // each pipeline must keep its own order: the unsorted result must not be + // cached under the node set and reused for the other pipeline. + DatanodeDetails a = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails b = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails c = MockDatanodeDetails.randomDatanodeDetails(); + List nodes1 = Arrays.asList(a, b, c); + List nodes2 = Arrays.asList(c, b, a); + Pipeline pipeline1 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes1) + .build(); + Pipeline pipeline2 = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes2) + .build(); + AllocatedBlock block1 = new AllocatedBlock.Builder().setPipeline(pipeline1) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + AllocatedBlock block2 = new AllocatedBlock.Builder().setPipeline(pipeline2) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID + 1, LOCAL_ID + 1)).build(); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), anyString())) + .thenReturn(Arrays.asList(block1, block2)); + + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + // Skip the sort: return the input list instance unchanged. + when(mockKeyManager.sortDatanodesForWrite(any(), any(), any())) + .thenAnswer(inv -> inv.getArgument(0)); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), 2 * scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("1.2.3.4").build(), ozoneManager); + + assertEquals(2, locations.size()); + // Each pipeline keeps its own order; the skipped-sort result is not shared. + assertEquals(nodes1, locations.get(0).getPipeline().getNodesInOrder()); + assertEquals(nodes2, locations.get(1).getPipeline().getNodesInOrder()); + // Sorted per pipeline, since the unsorted result is not cached. + verify(mockKeyManager, times(2)).sortDatanodesForWrite(any(), eq("1.2.3.4"), any()); + } + + @Test + public void testAllocateBlockKeepsOrderWhenRemoteAddressEmpty() throws Exception { + // Sort enabled and topology available, but the client has no remote address: + // OM must not sort, SCM receives an empty clientMachine, and the pipeline + // order is preserved. + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + Pipeline pipeline = Pipeline.newBuilder() + .setState(Pipeline.PipelineState.OPEN) + .setId(PipelineID.randomId()) + .setReplicationConfig( + StandaloneReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .build(); + AllocatedBlock block = new AllocatedBlock.Builder().setPipeline(pipeline) + .setContainerBlockID(new ContainerBlockID(CONTAINER_ID, LOCAL_ID)).build(); + ArgumentCaptor clientMachine = ArgumentCaptor.forClass(String.class); + when(scmBlockLocationProtocol.allocateBlock(anyLong(), anyInt(), any(), + anyString(), any(ExcludeList.class), clientMachine.capture())) + .thenReturn(Collections.singletonList(block)); + + KeyManager mockKeyManager = mock(KeyManager.class); + when(mockKeyManager.isSortDatanodesForWriteEnabled()).thenReturn(true); + when(ozoneManager.getKeyManager()).thenReturn(mockKeyManager); + when(ozoneManager.getClusterMapAllowNull()).thenReturn(mock(NetworkTopology.class)); + + OMAllocateBlockRequest request = + getOmAllocateBlockRequest(createAllocateBlockRequest()); + List locations = request.allocateBlock(replicationConfig, + new ExcludeList(), scmBlockSize, true, + UserInfo.newBuilder().setRemoteAddress("").build(), ozoneManager); + + assertEquals("", clientMachine.getValue()); + verify(mockKeyManager, never()).sortDatanodesForWrite(any(), anyString(), any()); + assertEquals(1, locations.size()); + // Assert the write order (nodesInOrder), which copyWithNodesInOrder would + // have changed had OM sorted; it must stay as the original pipeline order. + assertEquals(nodes, locations.get(0).getPipeline().getNodesInOrder()); + } + + @Test + public void sortDatanodesForWriteRequiresClientMachine() { + List nodes = Arrays.asList( + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails(), + MockDatanodeDetails.randomDatanodeDetails()); + assertThrows(IllegalArgumentException.class, + () -> keyManager.sortDatanodesForWrite(nodes, "", mock(NetworkTopology.class))); + } + + // Like createAllocateBlockRequest, but sets sortDatanodes and a UserInfo remote + // address for OM-side sort. + private OMRequest createAllocateBlockRequestWithSort(String clientAddress) { + KeyArgs keyArgs = KeyArgs.newBuilder() + .setVolumeName(volumeName).setBucketName(bucketName).setKeyName(keyName) + .setFactor(((RatisReplicationConfig) replicationConfig).getReplicationFactor()) + .setType(replicationConfig.getReplicationType()) + .setSortDatanodes(true) + .build(); + AllocateBlockRequest allocateBlockRequest = AllocateBlockRequest.newBuilder() + .setClientID(clientID).setKeyArgs(keyArgs).build(); + return OMRequest.newBuilder() + .setCmdType(OzoneManagerProtocolProtos.Type.AllocateBlock) + .setClientId(UUID.randomUUID().toString()) + .setUserInfo(UserInfo.newBuilder().setRemoteAddress(clientAddress).build()) + .setAllocateBlockRequest(allocateBlockRequest).build(); + } + protected OMRequest createAllocateBlockRequest() { KeyArgs keyArgs = KeyArgs.newBuilder()