HDDS-16388. Datanode putBlock Support StorageType

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java

@@ -41,7 +41,9 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
+import org.apache.hadoop.fs.StorageType;
 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;
@@ -186,7 +188,6 @@ public BlockOutputStream(
     replicationIndex = pipeline.getReplicaIndex(pipeline.getClosestNode());
     KeyValue keyValue =
         KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build();
-
     ContainerProtos.DatanodeBlockID.Builder blkIDBuilder =
         ContainerProtos.DatanodeBlockID.newBuilder()
             .setContainerID(blockID.getContainerID())
@@ -195,6 +196,8 @@ public BlockOutputStream(
     if (replicationIndex > 0) {
       blkIDBuilder.setReplicaIndex(replicationIndex);
     }
+    // TODO: Replica to the method parameter
+    blkIDBuilder.setStorageTypeID(StorageTypeUtils.getID(StorageType.DISK));
     this.containerBlockData = BlockData.newBuilder().setBlockID(
         blkIDBuilder.build()).addMetadata(keyValue);
     this.pipeline = pipeline;

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java

@@ -19,6 +19,7 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.Objects;
+import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 
@@ -34,29 +35,42 @@ public class BlockID {
   // BlockID object.
   private final Integer replicaIndex;
 
+  // Represents storage type of Block on a particular Datanode.
+  // Note this variable in the OM side will be null,
+  // because the OmKeyLocationInfo#getProtobuf Use BlockID#getProtobuf to get Protobuf object,
+  // BlockID#getProtobuf does not set storageType value in Protobuf.
+
+  // Currently for java class BlockID, OM and Datanode share the same java class object,
+  // but the protobuf objects HddsProtos.BlockID and ContainerProtos.DatanodeBlockID are
+  // used for OM and Datanode respectively.
+  private StorageType storageType;
+
   public BlockID(long containerID, long localID) {
-    this(containerID, localID, 0, null);
+    this(containerID, localID, 0, null, null);
   }
 
-  private BlockID(long containerID, long localID, long bcsID, Integer repIndex) {
+  private BlockID(long containerID, long localID, long bcsID, Integer repIndex,
+      StorageType storageType) {
     containerBlockID = new ContainerBlockID(containerID, localID);
     blockCommitSequenceId = bcsID;
     this.replicaIndex = repIndex;
+    this.storageType = storageType;
   }
 
   public BlockID(BlockID blockID) {
     this(blockID.getContainerID(), blockID.getLocalID(), blockID.getBlockCommitSequenceId(),
-        blockID.getReplicaIndex());
+        blockID.getReplicaIndex(), blockID.getStorageType());
   }
 
   public BlockID(ContainerBlockID containerBlockID) {
-    this(containerBlockID, 0, null);
+    this(containerBlockID, 0, null, null);
   }
 
-  private BlockID(ContainerBlockID containerBlockID, long bcsId, Integer repIndex) {
+  private BlockID(ContainerBlockID containerBlockID, long bcsId, Integer repIndex, StorageType storageType) {
     this.containerBlockID = containerBlockID;
     blockCommitSequenceId = bcsId;
     this.replicaIndex = repIndex;
+    this.storageType = storageType;
   }
 
   public long getContainerID() {
@@ -84,6 +98,10 @@ public ContainerBlockID getContainerBlockID() {
     return containerBlockID;
   }
 
+  public StorageType getStorageType() {
+    return storageType;
+  }
+
   @Override
   public String toString() {
     StringBuilder sb = new StringBuilder(64);
@@ -94,7 +112,8 @@ public String toString() {
   public void appendTo(StringBuilder sb) {
     containerBlockID.appendTo(sb);
     sb.append(" bcsId: ").append(blockCommitSequenceId)
-        .append(" replicaIndex: ").append(replicaIndex);
+        .append(" replicaIndex: ").append(replicaIndex)
+        .append(" storageType: ").append(storageType);
   }
 
   @JsonIgnore
@@ -103,6 +122,9 @@ public ContainerProtos.DatanodeBlockID getDatanodeBlockIDProtobuf() {
     if (replicaIndex != null) {
       blockID.setReplicaIndex(replicaIndex);
     }
+    if (storageType != null) {
+      blockID.setStorageTypeID(StorageTypeUtils.getID(storageType));
+    }
     return blockID.build();
   }
 
@@ -116,10 +138,15 @@ public ContainerProtos.DatanodeBlockID.Builder getDatanodeBlockIDProtobufBuilder
 
   @JsonIgnore
   public static BlockID getFromProtobuf(ContainerProtos.DatanodeBlockID blockID) {
+    StorageType storageType = null;
+    if (blockID.hasStorageTypeID() && blockID.getStorageTypeID() > 0) {
+      storageType = StorageTypeUtils.getStorageTypeFromID(blockID.getStorageTypeID());
+    }
     return new BlockID(blockID.getContainerID(),
         blockID.getLocalID(),
         blockID.getBlockCommitSequenceId(),
-        blockID.hasReplicaIndex() ? blockID.getReplicaIndex() : null);
+        blockID.hasReplicaIndex() ? blockID.getReplicaIndex() : null,
+        storageType);
   }
 
   @JsonIgnore
@@ -133,7 +160,7 @@ public HddsProtos.BlockID getProtobuf() {
   public static BlockID getFromProtobuf(HddsProtos.BlockID blockID) {
     return new BlockID(
         ContainerBlockID.getFromProtobuf(blockID.getContainerBlockID()),
-        blockID.getBlockCommitSequenceId(), null);
+        blockID.getBlockCommitSequenceId(), null, null);
   }
 
   @Override
@@ -147,12 +174,13 @@ public boolean equals(Object o) {
     BlockID blockID = (BlockID) o;
     return this.getContainerBlockID().equals(blockID.getContainerBlockID())
         && this.getBlockCommitSequenceId() == blockID.getBlockCommitSequenceId()
-        && Objects.equals(this.getReplicaIndex(), blockID.getReplicaIndex());
+        && Objects.equals(this.getReplicaIndex(), blockID.getReplicaIndex())
+        && Objects.equals(this.getStorageType(), blockID.getStorageType());
   }
 
   @Override
   public int hashCode() {
     return Objects.hash(containerBlockID.getContainerID(), containerBlockID.getLocalID(),
-        blockCommitSequenceId, replicaIndex);
+        blockCommitSequenceId, replicaIndex, storageType);
   }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/StorageTypeUtils.java

@@ -103,5 +103,13 @@ public static int getIDFromProtobuf(@Nonnull HddsProtos.StorageTypeProto proto)
       throw new IllegalArgumentException("Illegal Storage Type specified");
     }
   }
+
+  /**
+   * Returns integer representation of FileSystem StorageType.
+   * @return storageType int ID value
+   */
+  public static int getID(StorageType storageType) throws IllegalArgumentException {
+    return getStorageTypeProto(storageType).getNumber();
+  }
 }
 

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java

@@ -636,6 +636,7 @@ public static PutSmallFileResponseProto writeSmallFile(
   public static void createRecoveringContainer(XceiverClientSpi client,
       long containerID, String encodedToken, int replicaIndex)
       throws IOException {
+    // TODO StoragePolicy Support EC
     createContainer(client, containerID, encodedToken,
         ContainerProtos.ContainerDataProto.State.RECOVERING, replicaIndex);
   }

hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/container/ContainerTestHelper.java

@@ -39,7 +39,9 @@
 import java.util.concurrent.ThreadLocalRandom;
 import org.apache.commons.io.IOUtils;
 import org.apache.hadoop.conf.StorageUnit;
+import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.client.StorageTypeUtils;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
@@ -572,15 +574,18 @@ public static ContainerProtos.ContainerCommandRequestProto getFinalizeBlockReque
   }
 
   public static BlockID getTestBlockID(long containerID) {
-    return getTestBlockID(containerID, null);
+    return getTestBlockID(containerID, null, null);
   }
 
-  public static BlockID getTestBlockID(long containerID, Integer replicaIndex) {
+  public static BlockID getTestBlockID(long containerID, Integer replicaIndex, StorageType storageType) {
     DatanodeBlockID.Builder datanodeBlockID = DatanodeBlockID.newBuilder().setContainerID(containerID)
         .setLocalID(UniqueId.next());
     if (replicaIndex != null) {
       datanodeBlockID.setReplicaIndex(replicaIndex);
     }
+    if (storageType != null) {
+      datanodeBlockID.setStorageTypeID(StorageTypeUtils.getID(storageType));
+    }
     return BlockID.getFromProtobuf(datanodeBlockID.build());
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/HddsDispatcher.java

@@ -177,6 +177,7 @@ private boolean canIgnoreException(Result result) {
     case DELETE_ON_OPEN_CONTAINER:
     case UNSUPPORTED_REQUEST:// Blame client for sending unsupported request.
     case MALFORMED_REQUEST:// Blame client for sending malformed request.
+    case INVALID_ARGUMENT:
     case CONTAINER_MISSING:
     case CONTAINER_ALREADY_EXISTS:
       return true;
@@ -194,6 +195,12 @@ public void buildMissingContainerSetAndValidate(
   @Override
   public ContainerCommandResponseProto dispatch(
       ContainerCommandRequestProto msg, DispatcherContext dispatcherContext) {
+    try {
+      HddsUtils.getBlockID(msg);
+    } catch (IllegalArgumentException e) {
+      return ContainerUtils.logAndReturnError(LOG,
+          new StorageContainerException(e.getMessage(), e, Result.INVALID_ARGUMENT), msg);
+    }
     try {
       return dispatcher.processRequest(msg,
           req -> dispatchRequest(msg, dispatcherContext),
@@ -568,6 +575,11 @@ private void validateToken(
   @Override
   public void validateContainerCommand(
       ContainerCommandRequestProto msg) throws StorageContainerException {
+    try {
+      HddsUtils.getBlockID(msg);
+    } catch (IllegalArgumentException e) {
+      throw new StorageContainerException(e.getMessage(), e, Result.INVALID_ARGUMENT);
+    }
     try {
       validateToken(msg);
     } catch (IOException ioe) {

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java

@@ -1188,7 +1188,8 @@ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
         if (r.getResult() != ContainerProtos.Result.SUCCESS
             && r.getResult() != ContainerProtos.Result.CONTAINER_NOT_OPEN
             && r.getResult() != ContainerProtos.Result.CLOSED_CONTAINER_IO
-            && r.getResult() != ContainerProtos.Result.CHUNK_FILE_INCONSISTENCY) {
+            && r.getResult() != ContainerProtos.Result.CHUNK_FILE_INCONSISTENCY
+            && r.getResult() != ContainerProtos.Result.INVALID_ARGUMENT) {
           StorageContainerException sce =
               new StorageContainerException(r.getMessage(), r.getResult());
           LOG.error(

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -692,6 +692,8 @@ ContainerCommandResponseProto handlePutBlock(
       ContainerProtos.BlockData data = request.getPutBlock().getBlockData();
       BlockData blockData = BlockData.getFromProtoBuf(data);
       Objects.requireNonNull(blockData, "blockData == null");
+      Objects.requireNonNull(blockData.getBlockID());
+      BlockUtils.verifyStorageType(kvContainer.getContainerData(), blockData.getBlockID());
 
       boolean endOfBlock = false;
       if (!request.getPutBlock().hasEof() || request.getPutBlock().getEof()) {
@@ -1114,6 +1116,7 @@ ContainerCommandResponseProto handleWriteChunk(
 
       WriteChunkRequestProto writeChunk = request.getWriteChunk();
       BlockID blockID = BlockID.getFromProtobuf(writeChunk.getBlockID());
+      BlockUtils.verifyStorageType(kvContainer.getContainerData(), blockID);
       ContainerProtos.ChunkInfo chunkInfoProto = writeChunk.getChunkData();
 
       ChunkInfo chunkInfo = ChunkInfo.getFromProtoBuf(chunkInfoProto);
@@ -1272,6 +1275,7 @@ ContainerCommandResponseProto handlePutSmallFile(
       BlockData blockData = BlockData.getFromProtoBuf(
           putSmallFileReq.getBlock().getBlockData());
       Objects.requireNonNull(blockData, "blockData == null");
+      BlockUtils.verifyStorageType(kvContainer.getContainerData(), blockData.getBlockID());
 
       ContainerProtos.ChunkInfo chunkInfoProto = putSmallFileReq.getChunkInfo();
       ChunkInfo chunkInfo = ChunkInfo.getFromProtoBuf(chunkInfoProto);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/helpers/BlockUtils.java

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.EXPORT_CONTAINER_METADATA_FAILED;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.IMPORT_CONTAINER_METADATA_FAILED;
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.INVALID_ARGUMENT;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.NO_SUCH_BLOCK;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNABLE_TO_READ_METADATA_DB;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.UNKNOWN_BCSID;
@@ -37,6 +38,7 @@
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.container.common.helpers.BlockData;
+import org.apache.hadoop.ozone.container.common.impl.ContainerData;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.apache.hadoop.ozone.container.common.interfaces.DBHandle;
 import org.apache.hadoop.ozone.container.common.utils.ContainerCache;
@@ -211,6 +213,23 @@ public static BlockData getBlockData(byte[] bytes) throws IOException {
     }
   }
 
+  /**
+   * Verify if request block storage type matches the container storage type.
+   *
+   * @param containerData container data
+   * @param blockID requested block info
+   * @throws StorageContainerException if storage types do not match
+   */
+  public static void verifyStorageType(ContainerData containerData, BlockID blockID)
+      throws StorageContainerException {
+    if (blockID.getStorageType() != null && containerData.getStorageType() != null &&
+        blockID.getStorageType() != containerData.getStorageType()) {
+      throw new StorageContainerException(String.format(
+          "Block storage type %s does not match container storage type %s",
+          blockID.getStorageType(), containerData.getStorageType()), INVALID_ARGUMENT);
+    }
+  }
+
   /**
    * Verify if request block BCSID is supported.
    *

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/BlockManagerImpl.java

@@ -187,6 +187,7 @@ public long persistPutBlock(KeyValueContainer container,
     Objects.requireNonNull(data, "data == null");
     Preconditions.checkState(data.getContainerID() >= 0, "Container Id " +
         "cannot be negative");
+    BlockUtils.verifyStorageType(container.getContainerData(), data.getBlockID());
 
     KeyValueContainerData containerData = container.getContainerData();
 

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/helpers/TestBlockData.java

@@ -131,7 +131,7 @@ public void testToString() {
     final BlockID blockID = new BlockID(5, 123);
     blockID.setBlockCommitSequenceId(42);
     final BlockData subject = new BlockData(blockID);
-    assertEquals("[blockId=conID: 5 locID: 123 bcsId: 42 replicaIndex: null, size=0]",
+    assertEquals("[blockId=conID: 5 locID: 123 bcsId: 42 replicaIndex: null storageType: null, size=0]",
         subject.toString());
   }
 }