This is an automated email from the ASF dual-hosted git repository. devmadhuu pushed a commit to branch HDDS-11233 in repository https://gitbox.apache.org/repos/asf/ozone.git
commit e1a5d78b0a4f9fd306fb78e7b99f4fd975b4782b Author: XiChen <[email protected]> AuthorDate: Fri May 29 00:17:46 2026 +0800 HDDS-15241. Datanode Support create Container use specific StorageType (#10362) --- .../hadoop/hdds/scm/storage/BlockOutputStream.java | 5 +- .../hadoop/hdds/client/StorageTypeUtils.java | 102 ++++++++ .../hdds/scm/storage/ContainerProtocolCalls.java | 62 +++-- .../ozone/container/common/impl/ContainerData.java | 1 + .../container/common/impl/HddsDispatcher.java | 23 +- .../container/common/interfaces/Container.java | 14 +- .../common/interfaces/VolumeChoosingPolicy.java | 21 +- .../volume/AbstractStorageTypeChoosingPolicy.java | 6 - .../container/keyvalue/KeyValueContainer.java | 23 +- .../ozone/container/keyvalue/KeyValueHandler.java | 18 +- .../container/replication/ContainerImporter.java | 4 +- .../ozone/container/common/ContainerTestUtils.java | 3 +- .../container/common/TestBlockDeletingService.java | 3 +- .../TestSchemaTwoBackwardsCompatibility.java | 3 +- ...stStaleRecoveringContainerScrubbingService.java | 6 +- .../common/impl/TestContainerPersistence.java | 3 +- .../container/common/impl/TestHddsDispatcher.java | 259 +++++++++++++++++++-- .../volume/TestCapacityVolumeChoosingPolicy.java | 7 +- .../volume/TestRoundRobinVolumeChoosingPolicy.java | 15 +- .../common/volume/TestVolumeSetDiskChecks.java | 5 +- .../diskbalancer/TestDiskBalancerTask.java | 5 +- .../keyvalue/TestKeyValueBlockIterator.java | 3 +- .../container/keyvalue/TestKeyValueContainer.java | 48 ++-- .../TestKeyValueContainerIntegrityChecks.java | 3 +- .../TestKeyValueContainerMarkUnhealthy.java | 8 +- .../container/keyvalue/TestKeyValueHandler.java | 5 +- .../keyvalue/impl/AbstractTestChunkManager.java | 6 +- .../keyvalue/impl/TestBlockManagerImpl.java | 6 +- .../container/ozoneimpl/TestContainerReader.java | 22 +- .../container/ozoneimpl/TestOzoneContainer.java | 5 +- .../replication/TestContainerImporter.java | 15 ++ .../replication/TestGrpcReplicationService.java | 3 +- .../src/main/proto/DatanodeClientProtocol.proto | 6 + .../schemaupgrade/TestUpgradeContainerSchema.java | 6 +- .../hadoop/hdds/scm/TestContainerSmallFile.java | 9 +- .../hadoop/ozone/freon/ChunkManagerDiskWrite.java | 3 +- .../containergenerator/GeneratorDatanode.java | 3 +- 37 files changed, 561 insertions(+), 178 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java index b960e753744..77a667259a7 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java @@ -963,8 +963,11 @@ private CompletableFuture<PutBlockResult> writeChunkToContainer( byteBufferList = null; } + // TODO: Pass the requested storage type from allocation/storage policy + // once that context is wired through BlockOutputStream. Null preserves + // the current any-volume behavior. asyncReply = writeChunkAsync(xceiverClient, chunkInfo, - blockID.get(), data, tokenString, replicationIndex, blockData, close, containerAutoCreate()); + blockID.get(), data, tokenString, replicationIndex, blockData, close, null, containerAutoCreate()); CompletableFuture<ContainerCommandResponseProto> respFuture = asyncReply.getResponse(); validateFuture = respFuture.thenApplyAsync(e -> { diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java new file mode 100644 index 00000000000..0956581ff87 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java @@ -0,0 +1,102 @@ +/* + * 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 + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * 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.client; + +import jakarta.annotation.Nonnull; +import org.apache.hadoop.fs.StorageType; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos.StorageTypeProto; + +/** + * Utility class for converting between Hadoop's {@link StorageType} and + * Ozone's protobuf representation {@link HddsProtos.StorageTypeProto}. + */ +public final class StorageTypeUtils { + private StorageTypeUtils() { + } + + public static HddsProtos.StorageTypeProto getStorageTypeProto(@Nonnull StorageType type) + throws IllegalArgumentException { + switch (type) { + case SSD: + return HddsProtos.StorageTypeProto.SSD; + case DISK: + return HddsProtos.StorageTypeProto.DISK; + case ARCHIVE: + return HddsProtos.StorageTypeProto.ARCHIVE; + case PROVIDED: + return HddsProtos.StorageTypeProto.PROVIDED; + case RAM_DISK: + return HddsProtos.StorageTypeProto.RAM_DISK; + default: + throw new IllegalArgumentException("Illegal Storage Type specified"); + } + } + + public static StorageType getFromProtobuf(@Nonnull HddsProtos.StorageTypeProto proto) throws + IllegalArgumentException { + switch (proto) { + case SSD: + return StorageType.SSD; + case DISK: + return StorageType.DISK; + case ARCHIVE: + return StorageType.ARCHIVE; + case PROVIDED: + return StorageType.PROVIDED; + case RAM_DISK: + return StorageType.RAM_DISK; + default: + throw new IllegalArgumentException("Illegal Storage Type specified"); + } + } + + /** + * Returns Filesystem StorageType enum value corresponding to the int ID value. + * + * @param storageTypeID StorageType int ID value + * @return StorageType + */ + public static StorageType getStorageTypeFromID(int storageTypeID) throws + IllegalArgumentException { + if (StorageTypeProto.forNumber(storageTypeID) == null) { + throw new IllegalArgumentException("Illegal storageTypeID " + storageTypeID); + } + return getFromProtobuf(StorageTypeProto.forNumber(storageTypeID)); + } + + /** + * Returns integer representation of protobuf StorageType. + * + * @return storageType int ID value + */ + public static int getIDFromProtobuf(@Nonnull HddsProtos.StorageTypeProto proto) throws + IllegalArgumentException { + switch (proto) { + case SSD: + case DISK: + case ARCHIVE: + case PROVIDED: + case RAM_DISK: + return proto.getNumber(); + default: + throw new IllegalArgumentException("Illegal Storage Type specified"); + } + } +} + diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java index d2898c395d9..2f86f878f36 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java @@ -33,6 +33,7 @@ import java.util.function.Function; import org.apache.hadoop.hdds.annotation.InterfaceStability; import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.client.StorageTypeUtils; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData; @@ -62,6 +63,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadContainerResponseProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.WriteChunkRequestProto; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.XceiverClientReply; import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.XceiverClientSpi.Validator; @@ -490,16 +492,19 @@ static long getLen(ReadChunkResponseProto response) { * @param blockID ID of the block * @param data the data of the chunk to write * @param tokenString serialized block token + * @param storageType - the type of storage that is required, if the storageType + * is null, any storageType will be considered. * @throws IOException if there is an I/O error while performing the call */ @SuppressWarnings("parameternumber") public static XceiverClientReply writeChunkAsync( XceiverClientSpi xceiverClient, ChunkInfo chunk, BlockID blockID, ByteString data, String tokenString, - int replicationIndex, BlockData blockData, boolean close) + int replicationIndex, BlockData blockData, boolean close, + HddsProtos.StorageTypeProto storageType) throws IOException, ExecutionException, InterruptedException { return writeChunkAsync(xceiverClient, chunk, blockID, data, tokenString, - replicationIndex, blockData, close, true); + replicationIndex, blockData, close, storageType, true); } @SuppressWarnings("parameternumber") @@ -507,17 +512,14 @@ public static XceiverClientReply writeChunkAsync( XceiverClientSpi xceiverClient, ChunkInfo chunk, BlockID blockID, ByteString data, String tokenString, int replicationIndex, BlockData blockData, boolean close, - boolean containerAutoCreate) + HddsProtos.StorageTypeProto storageType, boolean containerAutoCreate) throws IOException, ExecutionException, InterruptedException { + DatanodeBlockID datanodeBlockID = getDatanodeBlockID( + blockID, replicationIndex, storageType); WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto.newBuilder() - .setBlockID(DatanodeBlockID.newBuilder() - .setContainerID(blockID.getContainerID()) - .setLocalID(blockID.getLocalID()) - .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId()) - .setReplicaIndex(replicationIndex) - .build()) + .setBlockID(datanodeBlockID) .setChunkData(chunk) .setData(data); if (blockData != null) { @@ -551,18 +553,23 @@ public static XceiverClientReply writeChunkAsync( * using a single RPC. This API is designed to be used for files which are * smaller than 1 MB. * - * @param client - client that communicates with the container. - * @param blockID - ID of the block - * @param data - Data to be written into the container. - * @param token a token for this block (may be null) + * @param client - client that communicates with the container. + * @param blockID - ID of the block + * @param data - Data to be written into the container. + * @param token a token for this block (may be null) + * @param storageType - the type of storage that is required, if the storageType + * is null, any storageType will be considered. * @return container protocol writeSmallFile response */ public static PutSmallFileResponseProto writeSmallFile( XceiverClientSpi client, BlockID blockID, byte[] data, - Token<OzoneBlockTokenIdentifier> token) throws IOException { + Token<OzoneBlockTokenIdentifier> token, + HddsProtos.StorageTypeProto storageType) throws IOException { + DatanodeBlockID datanodeBlockID = getDatanodeBlockID( + blockID, null, storageType); BlockData containerBlockData = - BlockData.newBuilder().setBlockID(blockID.getDatanodeBlockIDProtobuf()) + BlockData.newBuilder().setBlockID(datanodeBlockID) .build(); PutBlockRequestProto.Builder createBlockRequest = PutBlockRequestProto.newBuilder() @@ -583,10 +590,10 @@ public static PutSmallFileResponseProto writeSmallFile( .setChecksumData(checksumData.getProtoBufMessage()) .build(); - PutSmallFileRequestProto putSmallFileRequest = + PutSmallFileRequestProto.Builder putSmallFileBuilder = PutSmallFileRequestProto.newBuilder().setChunkInfo(chunk) - .setBlock(createBlockRequest).setData(ByteString.copyFrom(data)) - .build(); + .setBlock(createBlockRequest) + .setData(ByteString.copyFrom(data)); String id = client.getPipeline().getFirstNode().getUuidString(); ContainerCommandRequestProto.Builder builder = @@ -594,7 +601,7 @@ public static PutSmallFileResponseProto writeSmallFile( .setCmdType(Type.PutSmallFile) .setContainerID(blockID.getContainerID()) .setDatanodeUuid(id) - .setPutSmallFile(putSmallFileRequest); + .setPutSmallFile(putSmallFileBuilder); if (token != null) { builder.setEncodedToken(token.encodeToUrlString()); } @@ -647,6 +654,7 @@ public static void createContainer(XceiverClientSpi client, throws IOException { ContainerProtos.CreateContainerRequestProto.Builder createRequest = ContainerProtos.CreateContainerRequestProto.newBuilder(); + // TODO StoragePolicy Support createContainer Command createRequest .setContainerType(ContainerProtos.ContainerType.KeyValueContainer); if (state != null) { @@ -1027,4 +1035,20 @@ static DatanodeBlockID getDatanodeBlockID(BlockID blockID, DatanodeDetails datan } return b.build(); } + + private static DatanodeBlockID getDatanodeBlockID(BlockID blockID, + Integer replicationIndex, HddsProtos.StorageTypeProto storageType) { + DatanodeBlockID.Builder blockIDBuilder = DatanodeBlockID.newBuilder() + .setContainerID(blockID.getContainerID()) + .setLocalID(blockID.getLocalID()) + .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId()); + if (replicationIndex != null) { + blockIDBuilder.setReplicaIndex(replicationIndex); + } + if (storageType != null) { + blockIDBuilder.setStorageTypeID( + StorageTypeUtils.getIDFromProtobuf(storageType)); + } + return blockIDBuilder.build(); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java index efd6557bf3b..ff12124ad88 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java @@ -292,6 +292,7 @@ public void setStorageType(StorageType type) { storageType = type; } + @Nullable public StorageType getStorageType() { return storageType; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java index 1564e3ddbf5..f1122a2debf 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java @@ -45,6 +45,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; @@ -510,16 +511,28 @@ ContainerCommandResponseProto createContainer( createRequest.setContainerType(containerType); if (containerRequest.hasWriteChunk()) { - createRequest.setReplicaIndex( - containerRequest.getWriteChunk().getBlockID().getReplicaIndex()); + DatanodeBlockID blockID = containerRequest.getWriteChunk().getBlockID(); + createRequest.setReplicaIndex(blockID.getReplicaIndex()); + if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) { + createRequest.setStorageTypeID(blockID.getStorageTypeID()); + } } if (containerRequest.hasPutBlock()) { - createRequest.setReplicaIndex( - containerRequest.getPutBlock().getBlockData().getBlockID() - .getReplicaIndex()); + DatanodeBlockID blockID = containerRequest.getPutBlock().getBlockData().getBlockID(); + createRequest.setReplicaIndex(blockID.getReplicaIndex()); + if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) { + createRequest.setStorageTypeID(blockID.getStorageTypeID()); + } } + if (containerRequest.hasPutSmallFile()) { + // PutSmallFile Not support EC yet + DatanodeBlockID blockID = containerRequest.getPutSmallFile().getBlock().getBlockData().getBlockID(); + if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) { + createRequest.setStorageTypeID(blockID.getStorageTypeID()); + } + } ContainerCommandRequestProto.Builder requestBuilder = ContainerCommandRequestProto.newBuilder() .setCmdType(Type.CreateContainer) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java index ddb7de4140b..07d5db708c3 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java @@ -40,23 +40,13 @@ */ public interface Container<CONTAINERDATA extends ContainerData> { - /** - * Creates a container. - * - * @throws StorageContainerException - */ - void create(VolumeSet volumeSet, VolumeChoosingPolicy volumeChoosingPolicy, - String scmId) throws StorageContainerException; - /** * Creates a container and optionally constrains the destination volume to a * specific storage type. */ - default void create(VolumeSet volumeSet, + void create(VolumeSet volumeSet, VolumeChoosingPolicy volumeChoosingPolicy, String scmId, - StorageType storageType) throws StorageContainerException { - create(volumeSet, volumeChoosingPolicy, scmId); - } + StorageType storageType) throws StorageContainerException; /** * Deletes the container. diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java index 63d9dccc74b..0e80787510b 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/VolumeChoosingPolicy.java @@ -17,6 +17,7 @@ package org.apache.hadoop.ozone.container.common.interfaces; +import jakarta.annotation.Nullable; import java.io.IOException; import java.util.List; import org.apache.hadoop.fs.StorageType; @@ -29,22 +30,6 @@ @InterfaceAudience.Private public interface VolumeChoosingPolicy { - /** - * Choose a volume to place a container, - * given a list of volumes and the max container size sought for storage. - * - * The implementations of this interface must be thread-safe. - * - * @param volumes - a list of available volumes. - * @param maxContainerSize - the maximum size of the container for which a - * volume is sought. - * @return the chosen volume. - * @throws IOException when disks are unavailable or are full. - */ - @Deprecated - HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize) - throws IOException; - /** * Choose a volume to place a container, optionally constraining the choice * to a specific storage type. @@ -57,6 +42,6 @@ HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize) * @return the chosen volume. * @throws IOException when disks are unavailable or are full. */ - HddsVolume chooseVolume(List<HddsVolume> volumes, - long maxContainerSize, StorageType storageType) throws IOException; + HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize, + @Nullable StorageType storageType) throws IOException; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java index d73ab885e5f..5a013892760 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/AbstractStorageTypeChoosingPolicy.java @@ -31,12 +31,6 @@ public abstract class AbstractStorageTypeChoosingPolicy @Override @Deprecated - public HddsVolume chooseVolume(List<HddsVolume> volumes, - long maxContainerSize) throws IOException { - return chooseVolume(volumes, maxContainerSize, null); - } - - @Override public HddsVolume chooseVolume(List<HddsVolume> volumes, long maxContainerSize, StorageType storageType) throws IOException { List<HddsVolume> candidates = volumes; diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java index 460c957e5c5..e12b7d79447 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java @@ -146,13 +146,6 @@ public void setCheckChunksFilePath(boolean bCheckChunksDirFilePath) { this.bCheckChunksFilePath = bCheckChunksDirFilePath; } - @Override - @Deprecated - public void create(VolumeSet volumeSet, VolumeChoosingPolicy - volumeChoosingPolicy, String clusterId) throws StorageContainerException { - create(volumeSet, volumeChoosingPolicy, clusterId, null); - } - @Override public void create(VolumeSet volumeSet, VolumeChoosingPolicy volumeChoosingPolicy, String clusterId, StorageType storageType) @@ -172,11 +165,7 @@ public void create(VolumeSet volumeSet, VolumeChoosingPolicy HddsVolume containerVolume; String hddsVolumeDir; try { - // TODO Use the `chooseVolume` that supports `storageType` - containerVolume = storageType == null - ? volumeChoosingPolicy.chooseVolume(volumes, maxSize) - : volumeChoosingPolicy.chooseVolume( - volumes, maxSize, storageType); + containerVolume = volumeChoosingPolicy.chooseVolume(volumes, maxSize, storageType); hddsVolumeDir = containerVolume.getHddsRootDir().toString(); // Set volume before getContainerDBFile(), because we may need the // volume to deduce the db file. @@ -186,10 +175,12 @@ public void create(VolumeSet volumeSet, VolumeChoosingPolicy containerData.setCommittedSpace(true); } catch (DiskOutOfSpaceException ex) { throw new StorageContainerException("Container creation failed, " + - "due to disk out of space", ex, DISK_OUT_OF_SPACE); + "due to disk out of space" + storageTypeMessage(storageType), + ex, DISK_OUT_OF_SPACE); } catch (IOException ex) { throw new StorageContainerException( - "Container creation failed. " + ex.getMessage(), ex, + "Container creation failed" + storageTypeMessage(storageType) + + ". " + ex.getMessage(), ex, CONTAINER_INTERNAL_ERROR); } @@ -353,6 +344,10 @@ private void updateContainerFile(File containerFile) writeToContainerFile(containerFile, false); } + private static String storageTypeMessage(StorageType storageType) { + return storageType == null ? "" : " on StorageType: " + storageType; + } + @Override public void delete() throws StorageContainerException { try { diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index a664ae26a58..8a0f54fb78a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -98,8 +98,10 @@ import java.util.function.Function; import java.util.stream.Collectors; import org.apache.hadoop.fs.FileUtil; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsUtils; import org.apache.hadoop.hdds.client.BlockID; +import org.apache.hadoop.hdds.client.StorageTypeUtils; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.conf.StorageUnit; @@ -483,6 +485,18 @@ ContainerCommandResponseProto handleCreateContainer( } newContainerData.setReplicaIndex(request.getCreateContainer() .getReplicaIndex()); + StorageType storageType = null; + if (request.getCreateContainer().hasStorageTypeID()) { + try { + storageType = StorageTypeUtils.getStorageTypeFromID( + request.getCreateContainer().getStorageTypeID()); + } catch (IllegalArgumentException ex) { + return ContainerUtils.logAndReturnError(LOG, + new StorageContainerException(ex.getMessage(), ex, + INVALID_ARGUMENT), request); + } + } + newContainerData.setStorageType(storageType); // TODO: Add support to add metadataList to ContainerData. Add metadata // to container during creation. @@ -494,7 +508,7 @@ ContainerCommandResponseProto handleCreateContainer( containerIdLock.lock(); try { if (containerSet.getContainer(containerID) == null) { - newContainer.create(volumeSet, volumeChoosingPolicy, clusterId); + newContainer.create(volumeSet, volumeChoosingPolicy, clusterId, storageType); if (RECOVERING == newContainer.getContainerState()) { created = containerSet.addContainerByOverwriteMissingContainer(newContainer); } else { @@ -529,6 +543,7 @@ private void populateContainerPathFields(KeyValueContainer container, HddsVolume hddsVolume) throws IOException { volumeSet.readLock(); try { + // TODO StoragePolicy Check whether need to adapt storageType String idDir = VersionedDatanodeFeatures.ScmHA.chooseContainerPathID( hddsVolume, clusterId); container.populatePathFields(idDir, hddsVolume); @@ -555,7 +570,6 @@ ContainerCommandResponseProto handleReadContainer( request, containerData.getProtoBufMessage()); } - /** * Handles Update Container Request. If successful, the container metadata * is updated. diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java index 7b42006b229..682e119c553 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerImporter.java @@ -149,9 +149,11 @@ private static void deleteFileQuietely(Path tarFilePath) { HddsVolume chooseNextVolume(long spaceToReserve) throws IOException { // Choose volume that can hold both container in tmp and dest directory LOG.debug("Choosing volume to reserve space : {}", spaceToReserve); + // TODO: Use the target container storage type once replication/import + // requests carry it. Null preserves the existing any-volume behavior. return volumeChoosingPolicy.chooseVolume( StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), - spaceToReserve); + spaceToReserve, null); } public static Path getUntarDirectory(HddsVolume hddsVolume) diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java index 8231b2b3c7f..d1eee74c83b 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ContainerTestUtils.java @@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -385,7 +386,7 @@ public static KeyValueContainer addContainerToVolumeDir( KeyValueContainer container = new KeyValueContainer(keyValueContainerData, conf); - container.create(volume.getVolumeSet(), volumeChoosingPolicy, clusterId); + container.create(volume.getVolumeSet(), volumeChoosingPolicy, clusterId, StorageType.DISK); container.close(); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java index 4f885961b27..996226c92bf 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestBlockDeletingService.java @@ -58,6 +58,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -189,7 +190,7 @@ private KeyValueContainerData createToDeleteBlocks(ContainerSet containerSet, data.setSchemaVersion(schemaVersion); KeyValueContainer container = new KeyValueContainer(data, conf); container.create(volumeSet, - new RoundRobinVolumeChoosingPolicy(), scmId); + new RoundRobinVolumeChoosingPolicy(), scmId, StorageType.DISK); containerSet.addContainer(container); data = (KeyValueContainerData) containerSet.getContainer( containerID).getContainerData(); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java index 1bdac646b73..a1b32ac15bd 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestSchemaTwoBackwardsCompatibility.java @@ -37,6 +37,7 @@ import java.util.List; import java.util.UUID; import java.util.concurrent.TimeoutException; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -272,7 +273,7 @@ private KeyValueContainer createTestContainer() throws IOException { cData.setSchemaVersion(OzoneConsts.SCHEMA_V2); KeyValueContainer container = new KeyValueContainer(cData, conf); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - clusterID); + clusterID, StorageType.DISK); // populate with some blocks // metadata will be updated here, too diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java index e2881bd6f7e..6a8da4f236c 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/TestStaleRecoveringContainerScrubbingService.java @@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -44,6 +45,7 @@ import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -107,7 +109,7 @@ private void init() throws IOException { volumeSet = mock(MutableVolumeSet.class); volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume); } @@ -137,7 +139,7 @@ private List<Long> createTestContainers( new KeyValueContainer(recoveringContainerData, conf); recoveringKeyValueContainer.create( - volumeSet, volumeChoosingPolicy, clusterID); + volumeSet, volumeChoosingPolicy, clusterID, StorageType.DISK); containerSet.addContainer(recoveringKeyValueContainer); createdIds.add((long) containerIdNum); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java index 0498d849961..c2c2f1b7e89 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestContainerPersistence.java @@ -52,6 +52,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import org.apache.commons.io.FileUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -205,7 +206,7 @@ private KeyValueContainer addContainer(ContainerSet cSet, long cID) KeyValueContainer container = new KeyValueContainer(data, conf); commitBytesBefore = StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()).get(0).getCommittedBytes(); - container.create(volumeSet, volumeChoosingPolicy, SCM_ID); + container.create(volumeSet, volumeChoosingPolicy, SCM_ID, StorageType.DISK); cSet.addContainer(container); commitBytesAfter = container.getContainerData() diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java index d60ca220ece..3019b07b242 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java @@ -19,6 +19,7 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.hadoop.hdds.fs.MockSpaceUsagePersistence.inMemory; +import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.DISK_OUT_OF_SPACE; import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_DATANODE_DIR_KEY; import static org.apache.hadoop.hdds.scm.protocolPB.ContainerCommandResponseBuilders.getContainerCommandResponse; import static org.apache.hadoop.ozone.container.common.ContainerTestUtils.COMMIT_STAGE; @@ -41,10 +42,13 @@ import com.google.common.collect.Maps; import java.io.File; import java.io.IOException; +import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; +import java.util.Arrays; import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; @@ -52,6 +56,7 @@ import org.apache.commons.codec.digest.DigestUtils; import org.apache.commons.lang3.RandomUtils; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory; @@ -65,7 +70,9 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.WriteChunkRequestProto; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.hdds.security.token.TokenVerifier; import org.apache.hadoop.ozone.OzoneConfigKeys; @@ -168,9 +175,9 @@ public void testContainerCloseActionWhenFull( StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()) .forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile())); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - scmId.toString()); + scmId.toString(), StorageType.DISK); container2.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - scmId.toString()); + scmId.toString(), StorageType.DISK); containerSet.addContainer(container); containerSet.addContainer(container2); ContainerMetrics metrics = ContainerMetrics.create(conf); @@ -186,7 +193,7 @@ public void testContainerCloseActionWhenFull( conf, containerSet, volumeSet, handlers, context, metrics, null); hddsDispatcher.setClusterId(scmId.toString()); ContainerCommandResponseProto responseOne = hddsDispatcher - .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null); + .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null), null); assertEquals(ContainerProtos.Result.SUCCESS, responseOne.getResult()); verify(context, times(0)) @@ -197,9 +204,9 @@ public void testContainerCloseActionWhenFull( containerData2.getStatistics().setBlockBytesForTesting(Double.valueOf( StorageUnit.MB.toBytes(950)).longValue()); ContainerCommandResponseProto responseTwo = hddsDispatcher - .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L), null); + .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L, null), null); ContainerCommandResponseProto responseThree = hddsDispatcher - .dispatch(getWriteChunkRequest(dd.getUuidString(), 2L, 1L), null); + .dispatch(getWriteChunkRequest(dd.getUuidString(), 2L, 1L, null), null); assertEquals(ContainerProtos.Result.SUCCESS, responseTwo.getResult()); assertEquals(ContainerProtos.Result.SUCCESS, responseThree.getResult()); @@ -212,7 +219,7 @@ public void testContainerCloseActionWhenFull( // if we write again to container 1, the container action should get added but heartbeat should not get triggered // again because of throttling - hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 3L), null); + hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 3L, null), null); verify(context, times(3)).addContainerActionIfAbsent(any(ContainerAction.class)); verify(stateMachine, times(2)).triggerHeartbeat(); // was called twice before } finally { @@ -327,7 +334,7 @@ public void testContainerCloseActionWhenVolumeFull( StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()) .forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile())); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - scmId.toString()); + scmId.toString(), StorageType.DISK); containerSet.addContainer(container); ContainerMetrics metrics = ContainerMetrics.create(conf); Map<ContainerType, Handler> handlers = Maps.newHashMap(); @@ -344,15 +351,15 @@ public void testContainerCloseActionWhenVolumeFull( containerData.getVolume().incrementUsedSpace(60); usedSpace.addAndGet(60); ContainerCommandResponseProto response = hddsDispatcher - .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L), null); - assertEquals(ContainerProtos.Result.DISK_OUT_OF_SPACE, response.getResult()); + .dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null), null); + assertEquals(DISK_OUT_OF_SPACE, response.getResult()); verify(context, times(1)) .addContainerActionIfAbsent(any(ContainerAction.class)); // verify that immediate heartbeat is triggered verify(stateMachine, times(1)).triggerHeartbeat(); // the volume has reached the min free space boundary but this time the heartbeat should not be triggered because // of throttling - hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L), null); + hddsDispatcher.dispatch(getWriteChunkRequest(dd.getUuidString(), 1L, 2L, null), null); verify(context, times(2)).addContainerActionIfAbsent(any(ContainerAction.class)); verify(stateMachine, times(1)).triggerHeartbeat(); // was called once before @@ -367,8 +374,8 @@ public void testContainerCloseActionWhenVolumeFull( StorageContainerException scException = assertThrows(StorageContainerException.class, () -> container2.create(volumeSet, - new RoundRobinVolumeChoosingPolicy(), scmId.toString())); - assertEquals("Container creation failed, due to disk out of space", + new RoundRobinVolumeChoosingPolicy(), scmId.toString(), StorageType.DISK)); + assertEquals("Container creation failed, due to disk out of space on StorageType: DISK", scException.getMessage()); } finally { volumeSet.shutdown(); @@ -387,7 +394,7 @@ public void testCreateContainerWithWriteChunk() throws IOException { DatanodeDetails dd = randomDatanodeDetails(); HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf); ContainerCommandRequestProto writeChunkRequest = - getWriteChunkRequest(dd.getUuidString(), 1L, 1L); + getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null); // send read chunk request and make sure container does not exist ContainerCommandResponseProto response = hddsDispatcher.dispatch(getReadChunkRequest(writeChunkRequest), null); @@ -429,6 +436,151 @@ public void testCreateContainerWithWriteChunk() throws IOException { } } + @Test + public void testCreateContainerWithWriteChunkWithStorageType() throws IOException { + int volumeNum = 2; + // Create two volume one is SSD volume another is DISK volume + StringBuilder hddsDirs = new StringBuilder(); + File ssdVolume = Files.createTempDirectory(tempDir, "ssd").toFile(); + File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile(); + hddsDirs.append("[SSD]").append(ssdVolume).append(',').append("[DISK]").append(diskVolume); + UUID scmId = UUID.randomUUID(); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, hddsDirs.toString()); + DatanodeDetails dd = randomDatanodeDetails(); + HddsDispatcher dispatcher = createDispatcher(dd, scmId, conf); + long containerID = 1L; + long localId = 1L; + + ContainerCommandRequestProto writeSSDChunkReq1 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, HddsProtos.StorageTypeProto.SSD); + ContainerCommandRequestProto writeSSDChunkReq2 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, HddsProtos.StorageTypeProto.SSD); + ContainerCommandRequestProto writeDISKChunkReq1 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, HddsProtos.StorageTypeProto.DISK); + ContainerCommandRequestProto writeDISKChunkReq2 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, HddsProtos.StorageTypeProto.DISK); + + assertContainerDoNotExist(dispatcher, writeSSDChunkReq1); + assertContainerDoNotExist(dispatcher, writeSSDChunkReq2); + assertContainerDoNotExist(dispatcher, writeDISKChunkReq1); + assertContainerDoNotExist(dispatcher, writeDISKChunkReq2); + + assertContainerCreateAtSpecificVolume(dispatcher, writeSSDChunkReq1, Collections.singletonList(ssdVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeSSDChunkReq2, Collections.singletonList(ssdVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeDISKChunkReq1, Collections.singletonList(diskVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeDISKChunkReq2, Collections.singletonList(diskVolume)); + + // Except the DISK and SSD volume, cannot create Container on other StorageType Volume, + // because there is no other StorageType Volume + ContainerCommandRequestProto writeToRAMDISKChunkRequest = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, HddsProtos.StorageTypeProto.RAM_DISK); + assertContainerDoNotExist(dispatcher, writeToRAMDISKChunkRequest); + ContainerCommandResponseProto response = dispatcher.dispatch(writeToRAMDISKChunkRequest, null); + assertEquals(DISK_OUT_OF_SPACE, response.getResult()); + + // If set the StorageType to null, then the Container can be created on any StorageType Volume + ContainerCommandRequestProto writeAnyStorageTypeChunkRequest1 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, null); + ContainerCommandRequestProto writeAnyStorageTypeChunkRequest2 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, null); + ContainerCommandRequestProto writeAnyStorageTypeChunkRequest3 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, null); + ContainerCommandRequestProto writeAnyStorageTypeChunkRequest4 = + getWriteChunkRequest(dd.getUuidString(), containerID++, localId++, null); + + assertContainerCreateAtSpecificVolume(dispatcher, writeAnyStorageTypeChunkRequest1, + Arrays.asList(ssdVolume, diskVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeAnyStorageTypeChunkRequest2, + Arrays.asList(ssdVolume, diskVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeAnyStorageTypeChunkRequest3, + Arrays.asList(ssdVolume, diskVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, writeAnyStorageTypeChunkRequest4, + Arrays.asList(ssdVolume, diskVolume)); + } + + @Test + public void testCreateContainerWithPutSmallFileWithStorageType() throws IOException { + StringBuilder hddsDirs = new StringBuilder(); + File ssdVolume = Files.createTempDirectory(tempDir, "ssd").toFile(); + File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile(); + hddsDirs.append("[SSD]").append(ssdVolume).append(',').append("[DISK]").append(diskVolume); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, hddsDirs.toString()); + HddsDispatcher dispatcher = createDispatcher(randomDatanodeDetails(), UUID.randomUUID(), conf); + long containerID = 1L; + long localId = 1L; + + ContainerCommandRequestProto putSmallFileToSSD = + newPutSmallFile(containerID++, localId++, HddsProtos.StorageTypeProto.SSD); + ContainerCommandRequestProto putSmallFileToDISK = + newPutSmallFile(containerID++, localId++, HddsProtos.StorageTypeProto.DISK); + ContainerCommandRequestProto putSmallFileToRAMDISK = + newPutSmallFile(containerID++, localId++, HddsProtos.StorageTypeProto.RAM_DISK); + + assertContainerDoNotExist(dispatcher, putSmallFileToSSD); + assertContainerDoNotExist(dispatcher, putSmallFileToDISK); + assertContainerDoNotExist(dispatcher, putSmallFileToRAMDISK); + + assertContainerCreateAtSpecificVolume(dispatcher, putSmallFileToSSD, + Collections.singletonList(ssdVolume)); + assertContainerCreateAtSpecificVolume(dispatcher, putSmallFileToDISK, + Collections.singletonList(diskVolume)); + + ContainerCommandResponseProto response = dispatcher.dispatch(putSmallFileToRAMDISK, null); + assertEquals(DISK_OUT_OF_SPACE, response.getResult()); + } + + @Test + public void testCreateContainerRejectsInvalidStorageType() throws IOException { + File diskVolume = Files.createTempDirectory(tempDir, "disk").toFile(); + OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(ScmConfigKeys.HDDS_DATANODE_DIR_KEY, diskVolume.getAbsolutePath()); + DatanodeDetails dd = randomDatanodeDetails(); + HddsDispatcher dispatcher = createDispatcher(dd, UUID.randomUUID(), conf); + + ContainerCommandRequestProto writeChunkRequest = + getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null); + WriteChunkRequestProto writeChunk = writeChunkRequest.getWriteChunk(); + ContainerCommandRequestProto requestWithInvalidStorageType = + writeChunkRequest.toBuilder() + .setWriteChunk(writeChunk.toBuilder() + .setBlockID(writeChunk.getBlockID().toBuilder() + .setStorageTypeID(999))) + .build(); + + ContainerCommandResponseProto response = + dispatcher.dispatch(requestWithInvalidStorageType, null); + assertEquals(ContainerProtos.Result.INVALID_ARGUMENT, response.getResult()); + } + + private void assertContainerDoNotExist(HddsDispatcher hddsDispatcher, + ContainerCommandRequestProto request) { + ContainerCommandResponseProto response = + hddsDispatcher.dispatch(getReadContainerRequest(request), null); + assertEquals(response.getResult(), + ContainerProtos.Result.CONTAINER_NOT_FOUND); + } + + private void assertContainerCreateAtSpecificVolume(HddsDispatcher hddsDispatcher, + ContainerCommandRequestProto request, List<File> exceptionVolumePaths) { + + // Send data command without sending create container. + ContainerCommandResponseProto response = hddsDispatcher.dispatch(request, null); + // Container should be created as part of the data command. + assertEquals(ContainerProtos.Result.SUCCESS, response.getResult()); + + response = hddsDispatcher.dispatch(getReadContainerRequest(request), null); + assertEquals(ContainerProtos.Result.SUCCESS, response.getResult()); + + // Check if any of the exceptionVolumePaths match + String containerPath = response.getReadContainer().getContainerData().getContainerPath(); + boolean isMatched = exceptionVolumePaths.stream() + .anyMatch(path -> containerPath.startsWith(path.getAbsolutePath())); + + assertTrue(isMatched); + } + @Test public void testContainerNotFoundWithCommitChunk() throws IOException { String testDirPath = testDir.getPath(); @@ -440,7 +592,7 @@ public void testContainerNotFoundWithCommitChunk() throws IOException { DatanodeDetails dd = randomDatanodeDetails(); HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf); ContainerCommandRequestProto writeChunkRequest = - getWriteChunkRequest(dd.getUuidString(), 1L, 1L); + getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null); // send read chunk request and make sure container does not exist ContainerCommandResponseProto response = @@ -474,12 +626,12 @@ public void testWriteChunkWithCreateContainerFailure() throws IOException { DatanodeDetails dd = randomDatanodeDetails(); HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf); ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest( - dd.getUuidString(), 1L, 1L); + dd.getUuidString(), 1L, 1L, null); HddsDispatcher mockDispatcher = spy(hddsDispatcher); ContainerCommandResponseProto.Builder builder = getContainerCommandResponse(writeChunkRequest, - ContainerProtos.Result.DISK_OUT_OF_SPACE, ""); + DISK_OUT_OF_SPACE, ""); // Return DISK_OUT_OF_SPACE response when writing chunk // with container creation. doReturn(builder.build()).when(mockDispatcher) @@ -510,7 +662,7 @@ public void testCreateContainerWhenAlreadyExistsDoesNotMarkUnhealthy() throws IO // Create container via WriteChunk ContainerCommandRequestProto writeChunkRequest = - getWriteChunkRequest(dd.getUuidString(), 1L, 1L); + getWriteChunkRequest(dd.getUuidString(), 1L, 1L, null); ContainerCommandResponseProto initialResponse = hddsDispatcher.dispatch(writeChunkRequest, null); assertEquals(ContainerProtos.Result.SUCCESS, initialResponse.getResult()); @@ -592,7 +744,7 @@ public void testDuplicateWriteChunkAndPutBlockRequest() throws IOException { DatanodeDetails dd = randomDatanodeDetails(); HddsDispatcher hddsDispatcher = createDispatcher(dd, scmId, conf); ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest( - dd.getUuidString(), 1L, 1L); + dd.getUuidString(), 1L, 1L, null); //Send same WriteChunkRequest ContainerCommandResponseProto response; hddsDispatcher.dispatch(writeChunkRequest, null); @@ -700,7 +852,7 @@ private static DatanodeDetails randomDatanodeDetails() { } private ContainerCommandRequestProto getWriteChunkRequest( - String datanodeId, Long containerId, Long localId) { + String datanodeId, Long containerId, Long localId, HddsProtos.StorageTypeProto storageType) { ByteString data = ByteString.copyFrom( UUID.randomUUID().toString().getBytes(UTF_8)); @@ -714,10 +866,16 @@ private ContainerCommandRequestProto getWriteChunkRequest( .setChecksumData(Checksum.getNoChecksumDataProto()) .build(); + ContainerProtos.DatanodeBlockID.Builder blockID = + new BlockID(containerId, localId).getDatanodeBlockIDProtobufBuilder(); + // TODO: Pass the real storage type from the write path once that BlockID support StorageType + if (storageType != null) { + blockID.setStorageTypeID(storageType.getNumber()); + } + WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto .newBuilder() - .setBlockID(new BlockID(containerId, localId) - .getDatanodeBlockIDProtobuf()) + .setBlockID(blockID) .setChunkData(chunk) .setData(data); @@ -808,15 +966,50 @@ private ContainerCommandRequestProto getWriteChunkRequest0( } static ContainerCommandRequestProto newPutSmallFile(Long containerId, Long localId) { + return newPutSmallFile(containerId, localId, null); + } + + static ContainerCommandRequestProto newPutBlock(Long containerId, Long localId, + HddsProtos.StorageTypeProto storageType) { + BlockID blockID = new BlockID(containerId, localId); + ContainerProtos.DatanodeBlockID.Builder datanodeBlockID = + blockID.getDatanodeBlockIDProtobuf().toBuilder(); + if (storageType != null) { + datanodeBlockID.setStorageTypeID(storageType.getNumber()); + } + ContainerProtos.BlockData.Builder blockData = + ContainerProtos.BlockData.newBuilder() + .setBlockID(datanodeBlockID); + return ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.PutBlock) + .setContainerID(containerId) + .setDatanodeUuid(UUID.randomUUID().toString()) + .setPutBlock(ContainerProtos.PutBlockRequestProto.newBuilder() + .setBlockData(blockData)) + .build(); + } + + static ContainerCommandRequestProto newPutSmallFile(Long containerId, Long localId, + HddsProtos.StorageTypeProto storageType) { ByteString chunkData = ByteString.copyFrom(RandomUtils.secure().randomBytes(32)); - return newPutSmallFile(new BlockID(containerId, localId), chunkData); + return newPutSmallFile(new BlockID(containerId, localId), chunkData, storageType); } static ContainerCommandRequestProto newPutSmallFile( BlockID blockID, ByteString data) { + return newPutSmallFile(blockID, data, null); + } + + static ContainerCommandRequestProto newPutSmallFile( + BlockID blockID, ByteString data, HddsProtos.StorageTypeProto storageType) { + ContainerProtos.DatanodeBlockID.Builder datanodeBlockID = + blockID.getDatanodeBlockIDProtobuf().toBuilder(); + if (storageType != null) { + datanodeBlockID.setStorageTypeID(storageType.getNumber()); + } final ContainerProtos.BlockData.Builder blockData = ContainerProtos.BlockData.newBuilder() - .setBlockID(blockID.getDatanodeBlockIDProtobuf()); + .setBlockID(datanodeBlockID); final ContainerProtos.PutBlockRequestProto.Builder putBlockRequest = ContainerProtos.PutBlockRequestProto.newBuilder() .setBlockData(blockData); @@ -869,6 +1062,24 @@ private ContainerCommandRequestProto getReadChunkRequest( .build(); } + /** + * Creates container read chunk request using input container write chunk + * request. + * + * @param writeChunkRequest - Input container write chunk request + * @return container read chunk request + */ + private ContainerCommandRequestProto getReadContainerRequest( + ContainerCommandRequestProto request) { + return ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.ReadContainer) + .setContainerID(request.getContainerID()) + .setTraceID(request.getTraceID()) + .setDatanodeUuid(request.getDatanodeUuid()) + .setReadContainer(ContainerProtos.ReadContainerRequestProto.newBuilder()) + .build(); + } + @Test public void testValidateToken() throws Exception { try { @@ -899,7 +1110,7 @@ public void verify(Token<?> token, }; final ContainerCommandRequestProto request = getWriteChunkRequest( - dd.getUuidString(), 1L, 1L); + dd.getUuidString(), 1L, 1L, HddsProtos.StorageTypeProto.DISK); final HddsDispatcher dispatcher = createDispatcher( dd, scmId, conf, tokenVerifier); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java index e9d6f8739cf..1530727867c 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestCapacityVolumeChoosingPolicy.java @@ -29,6 +29,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory; import org.apache.hadoop.hdds.fs.MockSpaceUsageSource; @@ -113,7 +114,7 @@ public void testCapacityVolumeChoosingPolicy() throws Exception { // Test 1000 rounds of volume choosing for (int i = 0; i < 1000; i++) { - HddsVolume volume = policy.chooseVolume(volumes, 0); + HddsVolume volume = policy.chooseVolume(volumes, 0, StorageType.DISK); chooseCount.put(volume, chooseCount.get(volume) + 1); } @@ -197,7 +198,7 @@ public void testFreeSpaceRatioIsClampedToZeroWhenOverCommitted() throws Exceptio @Test public void throwsDiskOutOfSpaceIfRequestMoreThanAvailable() { Exception e = assertThrows(DiskOutOfSpaceException.class, - () -> policy.chooseVolume(volumes, 500)); + () -> policy.chooseVolume(volumes, 500, StorageType.DISK)); String msg = e.getMessage(); assertThat(msg) @@ -228,7 +229,7 @@ public void testVolumeCommittedSpace() throws Exception { volumes.forEach(vol -> initialCommittedSpace.put(vol, vol.getCommittedBytes())); - HddsVolume selectedVolume = policy.chooseVolume(volumes, 50); + HddsVolume selectedVolume = policy.chooseVolume(volumes, 50, StorageType.DISK); assertEquals(initialCommittedSpace.get(selectedVolume) + 50, selectedVolume.getCommittedBytes()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java index 91bc9caa2c2..c83953bc151 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestRoundRobinVolumeChoosingPolicy.java @@ -28,6 +28,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory; import org.apache.hadoop.hdds.fs.MockSpaceUsageSource; @@ -96,21 +97,21 @@ public void testRRVolumeChoosingPolicy() throws Exception { assertEquals(200L, hddsVolume2.getCurrentUsage().getAvailable()); // Test two rounds of round-robin choosing - assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0)); - assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0)); - assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0)); - assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0)); + assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0, StorageType.DISK)); + assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0, StorageType.DISK)); + assertEquals(hddsVolume1, policy.chooseVolume(volumes, 0, StorageType.DISK)); + assertEquals(hddsVolume2, policy.chooseVolume(volumes, 0, StorageType.DISK)); // The first volume has only 100L space, so the policy should // choose the second one in case we ask for more. assertEquals(hddsVolume2, - policy.chooseVolume(volumes, 120)); + policy.chooseVolume(volumes, 120, StorageType.DISK)); } @Test public void throwsDiskOutOfSpaceIfRequestMoreThanAvailable() { Exception e = assertThrows(DiskOutOfSpaceException.class, - () -> policy.chooseVolume(volumes, 300)); + () -> policy.chooseVolume(volumes, 300, StorageType.DISK)); String msg = e.getMessage(); assertThat(msg).contains("No volumes have enough space for a new container. " + @@ -123,7 +124,7 @@ public void testVolumeCommittedSpace() throws Exception { volumes.forEach(vol -> initialCommittedSpace.put(vol, vol.getCommittedBytes())); - HddsVolume selectedVolume = policy.chooseVolume(volumes, 50); + HddsVolume selectedVolume = policy.chooseVolume(volumes, 50, StorageType.DISK); assertEquals(initialCommittedSpace.get(selectedVolume) + 50, selectedVolume.getCommittedBytes()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java index c2c6ec822ce..29984a56eff 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/volume/TestVolumeSetDiskChecks.java @@ -41,6 +41,7 @@ import java.util.Set; import java.util.UUID; import org.apache.commons.io.FileUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -301,14 +302,14 @@ public void testVolumeFailure() throws IOException { StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()) .forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile())); container.create(volumeSet, - new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString()); + new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString(), StorageType.DISK); conSet.addContainer(container); KeyValueContainer container1 = new KeyValueContainer(data1, conf); StorageVolumeUtil.getHddsVolumesList(volumeSet1.getVolumesList()) .forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile())); container1.create(volumeSet1, - new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString()); + new RoundRobinVolumeChoosingPolicy(), UUID.randomUUID().toString(), StorageType.DISK); conSet.addContainer(container1); DatanodeStateMachine datanodeStateMachineMock = mock(DatanodeStateMachine.class); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java index c858ba18618..5096db8279a 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/diskbalancer/TestDiskBalancerTask.java @@ -55,6 +55,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import org.apache.commons.io.FileUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.fs.MockSpaceUsageCheckFactory; @@ -881,9 +882,9 @@ private KeyValueContainer createContainer(long containerId, HddsVolume vol, Stat KeyValueContainer container = new KeyValueContainer(containerData, conf); VolumeChoosingPolicy policy = mock(VolumeChoosingPolicy.class); - when(policy.chooseVolume(any(List.class), any(Long.class))) + when(policy.chooseVolume(any(List.class), any(Long.class), any(StorageType.class))) .thenReturn(vol); - container.create((VolumeSet) volumeSet, policy, scmId); + container.create((VolumeSet) volumeSet, policy, scmId, StorageType.DISK); containerSet.addContainer(container); // Manually update volume usage for test purposes diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java index 4d24f199581..d2cc55e55b4 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueBlockIterator.java @@ -35,6 +35,7 @@ import java.util.NoSuchElementException; import java.util.UUID; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -116,7 +117,7 @@ public void setup() throws Exception { // Init the container. KeyValueContainer container = new KeyValueContainer(containerData, conf); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - clusterID); + clusterID, StorageType.DISK); db = BlockUtils.getDB(containerData, conf); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java index 23018153756..8fffd1b3ebe 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainer.java @@ -165,7 +165,7 @@ private void init(ContainerTestVersionInfo versionInfo) throws Exception { .map(v -> (StorageVolume) v) .collect(Collectors.toList())); when(volumeChoosingPolicy - .chooseVolume(anyList(), anyLong())).thenAnswer( + .chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))).thenAnswer( invocation -> { List<HddsVolume> volumes = invocation.getArgument(0); return volumes.get(0); @@ -462,7 +462,7 @@ public void testContainerImportExport(ContainerTestVersionInfo versionInfo) KeyValueContainer container = new KeyValueContainer(containerData, CONF); HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume( - StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1); + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1, StorageType.DISK); container.populatePathFields(scmId, containerVolume); try (InputStream fis = Files.newInputStream(folderToExport.toPath())) { @@ -508,7 +508,7 @@ public void testContainerImportExport(ContainerTestVersionInfo versionInfo) container = new KeyValueContainer(containerData, CONF); containerVolume = volumeChoosingPolicy.chooseVolume( - StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1); + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1, StorageType.DISK); container.populatePathFields(scmId, containerVolume); KeyValueContainer finalContainer1 = container; assertThrows(IOException.class, () -> { @@ -603,7 +603,7 @@ private void checkContainerFilesPresent(KeyValueContainerData data, * Create the container on disk. */ private void createContainer() throws StorageContainerException { - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); keyValueContainerData = keyValueContainer.getContainerData(); } @@ -701,9 +701,9 @@ public void concurrentExport(ContainerTestVersionInfo versionInfo) public void testDuplicateContainer(ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); StorageContainerException exception = assertThrows(StorageContainerException.class, () -> - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId)); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK)); assertEquals(ContainerProtos.Result.CONTAINER_ALREADY_EXISTS, exception.getResult()); assertThat(exception).hasMessage("Container creation failed because ContainerFile already exists"); } @@ -713,13 +713,13 @@ public void testDiskFullExceptionCreateContainer( ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); reset(volumeChoosingPolicy); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenThrow(DiskChecker.DiskOutOfSpaceException.class); StorageContainerException exception = assertThrows(StorageContainerException.class, () -> - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId)); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK)); assertEquals(ContainerProtos.Result.DISK_OUT_OF_SPACE, exception.getResult()); - assertThat(exception).hasMessage("Container creation failed, due to disk out of space"); + assertThat(exception).hasMessage("Container creation failed, due to disk out of space on StorageType: DISK"); } @ContainerTestVersionInfo.ContainerTest @@ -729,7 +729,7 @@ public void testDeleteContainer(ContainerTestVersionInfo versionInfo) closeContainer(); keyValueContainer = new KeyValueContainer( keyValueContainerData, CONF); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); KeyValueContainerUtil.removeContainer( keyValueContainer.getContainerData(), CONF); keyValueContainer.delete(); @@ -756,7 +756,7 @@ public void testDeleteContainer(ContainerTestVersionInfo versionInfo) public void testCloseContainer(ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); keyValueContainer.close(); keyValueContainerData = keyValueContainer @@ -778,7 +778,7 @@ public void testCloseContainer(ContainerTestVersionInfo versionInfo) public void testReportOfUnhealthyContainer( ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); assertNotNull(keyValueContainer.getContainerReport()); keyValueContainer.markContainerUnhealthy(); File containerFile = keyValueContainer.getContainerFile(); @@ -822,7 +822,7 @@ public void testMarkUnhealthyWithMissingMetadataDir(ContainerTestVersionInfo ver public void testUpdateContainer(ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); Map<String, String> metadata = new HashMap<>(); metadata.put(OzoneConsts.VOLUME, OzoneConsts.OZONE); metadata.put(OzoneConsts.OWNER, OzoneConsts.OZONE_SIMPLE_HDFS_USER); @@ -851,7 +851,7 @@ public void testUpdateContainerUnsupportedRequest( StorageContainerException exception = assertThrows(StorageContainerException.class, () -> { keyValueContainer = new KeyValueContainer(keyValueContainerData, CONF); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); Map<String, String> metadata = new HashMap<>(); metadata.put(OzoneConsts.VOLUME, OzoneConsts.OZONE); keyValueContainer.update(metadata, false); @@ -868,7 +868,7 @@ public void testContainerRocksDB(ContainerTestVersionInfo versionInfo) closeContainer(); keyValueContainer = new KeyValueContainer( keyValueContainerData, CONF); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); try (DBHandle db = BlockUtils.getDB(keyValueContainerData, CONF)) { RDBStore store = (RDBStore) db.getStore().getStore(); @@ -922,7 +922,7 @@ public void testDBProfileAffectsDBOptions( ContainerTestVersionInfo versionInfo) throws Exception { init(versionInfo); // Create Container 1 - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); DatanodeDBProfile outProfile1; try (DBHandle db1 = @@ -943,7 +943,7 @@ public void testDBProfileAffectsDBOptions( (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(), datanodeId.toString()); keyValueContainer = new KeyValueContainer(keyValueContainerData, otherConf); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); DatanodeDBProfile outProfile2; try (DBHandle db2 = @@ -1023,7 +1023,7 @@ void testAutoCompactionSmallSstFile( List<File> exportFiles = new ArrayList<>(); for (HddsVolume volume: volumeList) { reset(volumeChoosingPolicy); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(volume); for (int index = 0; index < count; index++, containerId++) { // Create new container @@ -1031,7 +1031,7 @@ void testAutoCompactionSmallSstFile( (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(), datanodeId.toString()); container = new KeyValueContainer(containerData, CONF); - container.create(volumeSet, volumeChoosingPolicy, scmId); + container.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); containerData = container.getContainerData(); containerData.setState(ContainerProtos.ContainerDataProto.State.CLOSED); populate(container, numberOfKeysToWrite); @@ -1138,7 +1138,7 @@ public void testIsEmptyContainerStateWhileImport( KeyValueContainer container = new KeyValueContainer(containerData, CONF); HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume( - StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1); + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1, StorageType.DISK); container.populatePathFields(scmId, containerVolume); try (InputStream fis = Files.newInputStream(folderToExport.toPath())) { @@ -1186,7 +1186,7 @@ public void testIsEmptyContainerStateWhileImportWithoutBlock( KeyValueContainer container = new KeyValueContainer(containerData, CONF); HddsVolume containerVolume = volumeChoosingPolicy.chooseVolume( - StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1); + StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()), 1, StorageType.DISK); container.populatePathFields(scmId, containerVolume); try (InputStream fis = Files.newInputStream(folderToExport.toPath())) { @@ -1241,7 +1241,7 @@ private void testMixedSchemaImport(String dir, ContainerTestHelper.CONTAINER_MAX_SIZE, UUID.randomUUID().toString(), UUID.randomUUID().toString()); KeyValueContainer container = new KeyValueContainer(data, conf); - container.create(volumeSet, volumeChoosingPolicy, scmId); + container.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); long pendingDeleteBlockCount = 20; try (DBHandle meta = BlockUtils.getDB(data, conf)) { Table<String, Long> metadataTable = meta.getStore().getMetadataTable(); @@ -1303,10 +1303,10 @@ public void testContainerCreationCommitSpaceReserve( keyValueContainer = new KeyValueContainer(keyValueContainerData, CONF); keyValueContainer = spy(keyValueContainer); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); // verify that - verify(volumeChoosingPolicy).chooseVolume(anyList(), anyLong()); // this would reserve commit space + verify(volumeChoosingPolicy).chooseVolume(anyList(), anyLong(), eq(StorageType.DISK)); assertTrue(keyValueContainerData.isCommittedSpace()); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java index adc4b35ba1a..a1b44f4a44c 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerIntegrityChecks.java @@ -30,6 +30,7 @@ import java.util.List; import java.util.UUID; import org.apache.commons.lang3.RandomStringUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -132,7 +133,7 @@ protected KeyValueContainer createContainerWithBlocks(long containerId, UUID.randomUUID().toString(), UUID.randomUUID().toString()); KeyValueContainer container = new KeyValueContainer(containerData, conf); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - clusterID); + clusterID, StorageType.DISK); try (DBHandle metadataStore = BlockUtils.getDB(containerData, conf)) { assertNotNull(containerData.getChunksPath()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java index 3606f80d13f..77aa70b8fd7 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerMarkUnhealthy.java @@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -32,6 +33,7 @@ import java.nio.file.Path; import java.util.UUID; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.ozone.container.common.impl.ContainerDataYaml; @@ -79,7 +81,7 @@ public void setup() throws Exception { volumeSet = mock(MutableVolumeSet.class); volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume); keyValueContainerData = new KeyValueContainerData(1L, @@ -145,7 +147,7 @@ public void testMarkClosedContainerAsUnhealthy(ContainerLayoutVersion layoutVers initTestData(layoutVersion); // We need to create the container so the compact-on-close operation // does not NPE. - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); keyValueContainer.close(); keyValueContainer.markContainerUnhealthy(); assertThat(keyValueContainerData.getState()).isEqualTo(UNHEALTHY); @@ -159,7 +161,7 @@ public void testMarkQuasiClosedContainerAsUnhealthy(ContainerLayoutVersion layou initTestData(layoutVersion); // We need to create the container so the sync-on-quasi-close operation // does not NPE. - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); keyValueContainer.quasiClose(); keyValueContainer.markContainerUnhealthy(); assertThat(keyValueContainerData.getState()).isEqualTo(UNHEALTHY); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java index 2670f1585a2..7119d3178cc 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueHandler.java @@ -72,6 +72,7 @@ import org.apache.commons.io.FileUtils; import org.apache.hadoop.conf.StorageUnit; import org.apache.hadoop.fs.FileUtil; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -917,7 +918,7 @@ public void testICRsOnContainerClose(ContainerLayoutVersion containerLayoutVersi // markContainerForClose - OPEN -> CLOSING (should send immediate ICR) containerData.setState(ContainerProtos.ContainerDataProto.State.OPEN); KeyValueContainer container = new KeyValueContainer(containerData, conf); - container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID); + container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID, StorageType.DISK); containerSet.addContainer(container); kvHandler.markContainerForClose(container); @@ -1064,7 +1065,7 @@ public void testReadBlockMetrics() throws Exception { (long) StorageUnit.GB.toBytes(1), UUID.randomUUID().toString(), DATANODE_UUID); KeyValueContainer container = new KeyValueContainer(containerData, conf); - container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID); + container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), CLUSTER_ID, StorageType.DISK); containerSet.addContainer(container); BlockID blockID = ContainerTestHelper.getTestBlockID(containerID); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java index 6995b75f530..1e09faea8d9 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/AbstractTestChunkManager.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assumptions.abort; import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -35,6 +36,7 @@ import java.nio.ByteBuffer; import java.util.UUID; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.ozone.container.common.helpers.ChunkInfo; @@ -93,7 +95,7 @@ public final void setUp(@TempDir File confDir) throws Exception { RoundRobinVolumeChoosingPolicy volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume); keyValueContainerData = new KeyValueContainerData(1L, @@ -104,7 +106,7 @@ public final void setUp(@TempDir File confDir) throws Exception { keyValueContainer = new KeyValueContainer(keyValueContainerData, config); keyValueContainer.create(volumeSet, volumeChoosingPolicy, - UUID.randomUUID().toString()); + UUID.randomUUID().toString(), StorageType.DISK); header = "my header".getBytes(UTF_8); byte[] bytes = "testing write chunks".getBytes(UTF_8); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java index eb58ac3ce84..43d012c068c 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestBlockManagerImpl.java @@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -35,6 +36,7 @@ import java.util.List; import java.util.UUID; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -95,7 +97,7 @@ private void initialize() throws Exception { VolumeSet volumeSet = mock(MutableVolumeSet.class); RoundRobinVolumeChoosingPolicy volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume); KeyValueContainerData keyValueContainerData = new KeyValueContainerData(1L, @@ -106,7 +108,7 @@ private void initialize() throws Exception { keyValueContainer = new KeyValueContainer( keyValueContainerData, config); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); // Creating BlockData BlockID blockID = new BlockID(1L, 1L); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java index 234fda2cd00..4b796376fd0 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestContainerReader.java @@ -33,6 +33,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -45,6 +46,7 @@ import java.util.List; import java.util.UUID; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -130,7 +132,7 @@ private void setup(ContainerTestVersionInfo versionInfo) throws Exception { volumeSet = mock(MutableVolumeSet.class); volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume); for (int i = 0; i < 2; i++) { @@ -141,7 +143,7 @@ private void setup(ContainerTestVersionInfo versionInfo) throws Exception { KeyValueContainer keyValueContainer = new KeyValueContainer(keyValueContainerData, conf); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK); List<Long> blkNames; @@ -299,7 +301,7 @@ public void testContainerReader(ContainerTestVersionInfo versionInfo) new KeyValueContainer(recoveringContainerData, conf); recoveringKeyValueContainer.create( - volumeSet, volumeChoosingPolicy, clusterId); + volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK); thread = new Thread(containerReader); thread.start(); @@ -359,7 +361,7 @@ public void testContainerReaderWithLoadException( StorageVolumeUtil.checkVolume(hddsVolume1, clusterId, clusterId, conf, null, null); volumeChoosingPolicy1 = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume1); int containerCount = 3; @@ -370,7 +372,7 @@ public void testContainerReaderWithLoadException( datanodeId.toString()); KeyValueContainer keyValueContainer = new KeyValueContainer(keyValueContainerData, conf); - keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId); + keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId, StorageType.DISK); if (i == 0) { // rename first container directory name @@ -417,7 +419,7 @@ public void testContainerReaderWithInvalidDbPath( StorageVolumeUtil.checkVolume(hddsVolume1, clusterId, clusterId, conf, null, null); volumeChoosingPolicy1 = mock(RoundRobinVolumeChoosingPolicy.class); - when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy1.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenReturn(hddsVolume1); List<File> dbPathList = new ArrayList<>(); @@ -429,7 +431,7 @@ public void testContainerReaderWithInvalidDbPath( datanodeId.toString()); KeyValueContainer keyValueContainer = new KeyValueContainer(keyValueContainerData, conf); - keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId); + keyValueContainer.create(volumeSet1, volumeChoosingPolicy1, clusterId, StorageType.DISK); dbPathList.add(keyValueContainerData.getDbFile()); } ContainerCache.getInstance(conf).shutdownCache(); @@ -642,7 +644,7 @@ private KeyValueContainer createContainerWithId(int id, VolumeSet volSet, KeyValueContainer keyValueContainer = new KeyValueContainer(keyValueContainerData, conf); - keyValueContainer.create(volSet, policy, clusterId); + keyValueContainer.create(volSet, policy, clusterId, StorageType.DISK); List<Long> blkNames; if (id % 2 == 0) { @@ -680,7 +682,7 @@ public void testMarkedDeletedContainerCleared( KeyValueContainer kvContainer = new KeyValueContainer(containerData, conf); kvContainer.create( - volumeSet, volumeChoosingPolicy, clusterId); + volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK); long baseCount = 0; if (containerData.hasSchema(OzoneConsts.SCHEMA_V3)) { // add db entry for the container ID 101 for V3 @@ -849,7 +851,7 @@ private KeyValueContainer createContainer(long containerId) throws Exception { (long) StorageUnit.GB.toBytes(5), UUID.randomUUID().toString(), datanodeId.toString()); containerData.setState(ContainerProtos.ContainerDataProto.State.CLOSED); KeyValueContainer container = new KeyValueContainer(containerData, conf); - container.create(volumeSet, volumeChoosingPolicy, clusterId); + container.create(volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK); return container; } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java index 91c3f8ed58c..ecac604cde6 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/ozoneimpl/TestOzoneContainer.java @@ -41,6 +41,7 @@ import java.util.concurrent.ConcurrentSkipListSet; import org.apache.commons.io.FileUtils; import org.apache.hadoop.conf.StorageUnit; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -168,7 +169,7 @@ public void testBuildContainerMap(ContainerTestVersionInfo versionInfo) containerDatas.add(keyValueContainerData); keyValueContainer = new KeyValueContainer( keyValueContainerData, conf); - keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK); myVolume = keyValueContainer.getContainerData().getVolume(); // Track container in mock volume @@ -286,7 +287,7 @@ public void testContainerCreateDiskFull(ContainerTestVersionInfo versionInfo) StorageContainerException e = assertThrows( StorageContainerException.class, () -> keyValueContainer. - create(volumeSet, volumeChoosingPolicy, clusterId) + create(volumeSet, volumeChoosingPolicy, clusterId, StorageType.DISK) ); assertEquals(DISK_OUT_OF_SPACE, e.getResult()); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java index e594dad3e58..553fb267882 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerImporter.java @@ -23,6 +23,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.anyLong; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.doNothing; @@ -234,6 +236,19 @@ public void testImportContainerResetsLastScanTime() throws Exception { assertEquals(Optional.empty(), containerData.lastDataScanTime()); } + @Test + public void testChooseNextVolumeStorageType() throws Exception { + VolumeChoosingPolicy policy = mock(VolumeChoosingPolicy.class); + HddsVolume expectedVolume = mock(HddsVolume.class); + long spaceToReserve = 100L; + when(policy.chooseVolume(anyList(), anyLong(), isNull())) + .thenReturn(expectedVolume); + ContainerImporter importer = new ContainerImporter(conf, containerSet, + controllerMock, volumeSet, policy); + + assertEquals(expectedVolume, importer.chooseNextVolume(spaceToReserve)); + } + private File containerTarFile(long id, ContainerData data) throws IOException { File yamlFile = new File(tempDir, "container.yaml"); ContainerDataYaml.createContainerFile(data, yamlFile); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java index 079c41fcbec..4d90a8328be 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java @@ -37,6 +37,7 @@ import java.util.Collections; import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -134,7 +135,7 @@ public void init() throws Exception { StorageVolumeUtil.getHddsVolumesList(volumeSet.getVolumesList()) .forEach(hddsVolume -> hddsVolume.setDbParentDir(tempDir.toFile())); container.create(volumeSet, new RoundRobinVolumeChoosingPolicy(), - "test-replication"); + "test-replication", StorageType.DISK); containerSet.addContainer(container); container.close(); diff --git a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto index 72d50c975a5..9989dddc2bf 100644 --- a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto +++ b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto @@ -153,6 +153,8 @@ message DatanodeBlockID { optional uint64 blockCommitSequenceId = 3 [default = 0]; optional int32 replicaIndex = 4; + optional int32 storageTypeID = 5; + } message KeyValue { @@ -283,6 +285,10 @@ message CreateContainerRequestProto { optional ContainerType containerType = 3 [default = KeyValueContainer]; optional int32 replicaIndex = 4; optional ContainerDataProto.State state = 5; + + // If we do not set storageTypeID, it means that we allow Containers to be + // created on Volumes of any StorageType. + optional int32 storageTypeID = 6; } message CreateContainerResponseProto { diff --git a/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java b/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java index f8983e23a9e..e51e50ffe3b 100644 --- a/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java +++ b/hadoop-ozone/cli-repair/src/test/java/org/apache/hadoop/ozone/repair/datanode/schemaupgrade/TestUpgradeContainerSchema.java @@ -36,6 +36,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.Mockito.anyList; import static org.mockito.Mockito.anyLong; +import static org.mockito.Mockito.eq; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -53,6 +54,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; import java.util.stream.Stream; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.cli.GenericCli; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -159,7 +161,7 @@ private void initDatanode(HDDSLayoutFeature layoutFeature) throws IOException { volumeChoosingPolicy = mock(RoundRobinVolumeChoosingPolicy.class); final AtomicInteger loopCount = new AtomicInteger(0); - when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong())) + when(volumeChoosingPolicy.chooseVolume(anyList(), anyLong(), eq(StorageType.DISK))) .thenAnswer(invocation -> { final int ii = loopCount.getAndIncrement() % volumes.size(); return volumes.get(ii); @@ -318,7 +320,7 @@ private void putChunksInBlock( data.setSchemaVersion(SCHEMA_V2); KeyValueContainer container = new KeyValueContainer(data, conf); - container.create(volumeSet, volumeChoosingPolicy, SCM_ID); + container.create(volumeSet, volumeChoosingPolicy, SCM_ID, StorageType.DISK); containerSet.addContainer(container); data = (KeyValueContainerData) containerSet.getContainer(containerId) diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java index 8140ea3eb45..c624b091f0f 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestContainerSmallFile.java @@ -81,7 +81,7 @@ public void testAllocateWrite() throws Exception { BlockID blockID = ContainerTestHelper.getTestBlockID( container.getContainerInfo().getContainerID()); ContainerProtocolCalls.writeSmallFile(client, blockID, - "data123".getBytes(UTF_8), null); + "data123".getBytes(UTF_8), null, null); ContainerProtos.GetSmallFileResponseProto response = ContainerProtocolCalls.readSmallFile(client, blockID, null); String readData = response.getData().getDataBuffers().getBuffersList() @@ -124,7 +124,7 @@ public void testInvalidContainerRead() throws Exception { BlockID blockID = ContainerTestHelper.getTestBlockID( container.getContainerInfo().getContainerID()); ContainerProtocolCalls.writeSmallFile(client, blockID, - "data123".getBytes(UTF_8), null); + "data123".getBytes(UTF_8), null, null); assertThrowsExactly(StorageContainerException.class, () -> ContainerProtocolCalls.readSmallFile(client, @@ -149,7 +149,8 @@ public void testReadWriteWithBCSId() throws Exception { container.getContainerInfo().getContainerID()); ContainerProtos.PutSmallFileResponseProto responseProto = ContainerProtocolCalls - .writeSmallFile(client, blockID1, "data123".getBytes(UTF_8), null); + .writeSmallFile(client, blockID1, "data123".getBytes(UTF_8), null, + null); long bcsId = responseProto.getCommittedBlockLength().getBlockID() .getBlockCommitSequenceId(); @@ -164,7 +165,7 @@ public void testReadWriteWithBCSId() throws Exception { BlockID blockID2 = ContainerTestHelper .getTestBlockID(container.getContainerInfo().getContainerID()); ContainerProtocolCalls - .writeSmallFile(client, blockID2, "data123".getBytes(UTF_8), null); + .writeSmallFile(client, blockID2, "data123".getBytes(UTF_8), null, null); blockID1.setBlockCommitSequenceId(bcsId + 1); //read a file with higher bcsId than the committed bcsId for the block diff --git a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java index b7ca060419d..885ff965520 100644 --- a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java +++ b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/ChunkManagerDiskWrite.java @@ -28,6 +28,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.lang3.RandomStringUtils; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.cli.HddsVersionProvider; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -130,7 +131,7 @@ public Void call() throws Exception { KeyValueContainer keyValueContainer = new KeyValueContainer(keyValueContainerData, ozoneConfiguration); - keyValueContainer.create(volumeSet, volumeChoicePolicy, "scmid"); + keyValueContainer.create(volumeSet, volumeChoicePolicy, "scmid", StorageType.DISK); containersPerThread.put(i, keyValueContainer); } diff --git a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java index 10f47fe2090..239026ddc52 100644 --- a/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java +++ b/hadoop-ozone/vapor/src/main/java/org/apache/hadoop/ozone/freon/containergenerator/GeneratorDatanode.java @@ -34,6 +34,7 @@ import java.util.SplittableRandom; import java.util.concurrent.Callable; import java.util.stream.Stream; +import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdds.cli.HddsVersionProvider; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -329,7 +330,7 @@ private KeyValueContainer createContainer(long containerId) new KeyValueContainer(keyValueContainerData, config); try { - keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId); + keyValueContainer.create(volumeSet, volumeChoosingPolicy, scmId, StorageType.DISK); } catch (StorageContainerException ex) { throw new RuntimeException(ex); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
