HDDS-10338. Implement a Client Datanode API to stream a block

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

@@ -113,6 +113,13 @@ public class OzoneClientConfig {
       tags = ConfigTag.CLIENT)
   private long streamBufferMaxSize = 32 * 1024 * 1024;
 
+  @Config(key = "stream.readblock.enable",
+      defaultValue = "false",
+      type = ConfigType.BOOLEAN,
+      description = "Allow ReadBlock to stream all the readChunk in one request.",
+      tags = ConfigTag.CLIENT)
+  private boolean streamReadBlock = false;
+
   @Config(key = "ozone.client.max.retries",
       defaultValue = "5",
       description = "Maximum number of retries by Ozone Client on "
@@ -151,7 +158,7 @@ public class OzoneClientConfig {
       description = "The checksum type [NONE/ CRC32/ CRC32C/ SHA256/ MD5] "
           + "determines which algorithm would be used to compute checksum for "
           + "chunk data. Default checksum type is CRC32.",
-      tags = { ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE })
+      tags = {ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE})
   private String checksumType = ChecksumType.CRC32.name();
 
   @Config(key = "ozone.client.bytes.per.checksum",
@@ -160,7 +167,7 @@ public class OzoneClientConfig {
       description = "Checksum will be computed for every bytes per checksum "
           + "number of bytes and stored sequentially. The minimum value for "
           + "this config is 8KB.",
-      tags = { ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE })
+      tags = {ConfigTag.CLIENT, ConfigTag.CRYPTO_COMPLIANCE})
   private int bytesPerChecksum = 16 * 1024;
 
   @Config(key = "ozone.client.verify.checksum",
@@ -538,6 +545,14 @@ public int getMaxConcurrentWritePerKey() {
     return this.maxConcurrentWritePerKey;
   }
 
+  public boolean isStreamReadBlock() {
+    return streamReadBlock;
+  }
+
+  public void setStreamReadBlock(boolean streamReadBlock) {
+    this.streamReadBlock = streamReadBlock;
+  }
+
   /**
    * Enum for indicating what mode to use when combining chunk and block
    * checksums to define an aggregate FileChecksum. This should be considered

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

@@ -384,25 +384,9 @@ private XceiverClientReply sendCommandWithTraceIDAndRetry(
         });
   }
 
-  private XceiverClientReply sendCommandWithRetry(
-      ContainerCommandRequestProto request, List<Validator> validators)
-      throws IOException {
-    ContainerCommandResponseProto responseProto = null;
-    IOException ioException = null;
-
-    // In case of an exception or an error, we will try to read from the
-    // datanodes in the pipeline in a round-robin fashion.
-    XceiverClientReply reply = new XceiverClientReply(null);
+  private List<DatanodeDetails> sortDatanodes(ContainerCommandRequestProto request) throws IOException {
     List<DatanodeDetails> datanodeList = null;
-
-    DatanodeBlockID blockID = null;
-    if (request.getCmdType() == ContainerProtos.Type.GetBlock) {
-      blockID = request.getGetBlock().getBlockID();
-    } else if  (request.getCmdType() == ContainerProtos.Type.ReadChunk) {
-      blockID = request.getReadChunk().getBlockID();
-    } else if (request.getCmdType() == ContainerProtos.Type.GetSmallFile) {
-      blockID = request.getGetSmallFile().getBlock().getBlockID();
-    }
+    DatanodeBlockID blockID = getRequestBlockID(request);
 
     if (blockID != null) {
       if (request.getCmdType() != ContainerProtos.Type.ReadChunk) {
@@ -440,6 +424,33 @@ private XceiverClientReply sendCommandWithRetry(
     if (!allInService) {
       datanodeList = sortDatanodeByOperationalState(datanodeList);
     }
+    return datanodeList;
+  }
+
+  private static DatanodeBlockID getRequestBlockID(ContainerCommandRequestProto request) {
+    DatanodeBlockID blockID = null;
+    if (request.getCmdType() == ContainerProtos.Type.GetBlock) {
+      blockID = request.getGetBlock().getBlockID();
+    } else if  (request.getCmdType() == ContainerProtos.Type.ReadChunk) {
+      blockID = request.getReadChunk().getBlockID();
+    } else if (request.getCmdType() == ContainerProtos.Type.GetSmallFile) {
+      blockID = request.getGetSmallFile().getBlock().getBlockID();
+    } else if (request.getCmdType() == ContainerProtos.Type.ReadBlock) {
+      blockID = request.getReadBlock().getBlockID();
+    }
+    return blockID;
+  }
+
+  private XceiverClientReply sendCommandWithRetry(
+      ContainerCommandRequestProto request, List<Validator> validators)
+      throws IOException {
+    ContainerCommandResponseProto responseProto = null;
+    IOException ioException = null;
+
+    // In case of an exception or an error, we will try to read from the
+    // datanodes in the pipeline in a round-robin fashion.
+    XceiverClientReply reply = new XceiverClientReply(null);
+    List<DatanodeDetails> datanodeList = sortDatanodes(request);
 
     for (DatanodeDetails dn : datanodeList) {
       try {
@@ -495,7 +506,7 @@ private XceiverClientReply sendCommandWithRetry(
       String message = "Failed to execute command {}";
       if (LOG.isDebugEnabled()) {
         LOG.debug(message + " on the pipeline {}.",
-                processForDebug(request), pipeline);
+            processForDebug(request), pipeline);
       } else {
         LOG.warn(message + " on the pipeline {}.",
                 request.getCmdType(), pipeline);
@@ -504,6 +515,62 @@ private XceiverClientReply sendCommandWithRetry(
     }
   }
 
+  /**
+   * Starts a streaming read operation, intended to read entire blocks from the datanodes. This method expects a
+   * {@link StreamingReaderSpi} to be passed in, which will be used to receive the streamed data from the datanode.
+   * Upon successfully starting the streaming read, a {@link StreamingReadResponse} is set into the pass StreamObserver,
+   * which contains information about the datanode used for the read, and the request observer that can be used to
+   * manage the stream (e.g., to cancel it if needed). A semaphore is acquired to limit the number of concurrent
+   * streaming reads so upon successful return of this method, the caller must ensure to call
+   * {@link #completeStreamRead()} to release the semaphore once the streaming read is complete.
+   * @param request The container command request to initiate the streaming read.
+   * @param streamObserver The observer that will handle the streamed responses.=
+   * @throws IOException
+   * @throws InterruptedException
+   */
+  @Override
+  public void streamRead(ContainerCommandRequestProto request,
+      StreamingReaderSpi streamObserver) throws IOException, InterruptedException {
+    List<DatanodeDetails> datanodeList = sortDatanodes(request);
+    IOException lastException = null;
+    for (DatanodeDetails dn : datanodeList) {
+      try {
+        checkOpen(dn);
+        semaphore.acquire();
+        XceiverClientProtocolServiceStub stub = asyncStubs.get(dn.getID());
+        if (stub == null) {
+          throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
+        }
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Executing command {} on datanode {}", processForDebug(request), dn);
+        }
+        stub.withDeadlineAfter(timeout, TimeUnit.SECONDS)
+            .streamBlock(request, streamObserver);
+        streamObserver.setStreamingDatanode(dn);
+        return;
+      } catch (IOException e) {
+        LOG.error("Failed to start streaming read to DataNode {}", dn, e);
+        semaphore.release();
+        lastException = e;
+      }
+    }
+    if (lastException != null) {
+      throw lastException;
+    } else {
+      throw new IOException("Failed to start streaming read to any available DataNodes");
+    }
+  }
+
+  /**
+   * This method should be called to indicate the end of streaming read. Its primary purpose is to release the
+   * semaphore acquired when starting the streaming read, but is also used to update any metrics or debug logs as
+   * needed.
+   */
+  @Override
+  public void completeStreamRead() {
+    semaphore.release();
+  }
+
   private static List<DatanodeDetails> sortDatanodeByOperationalState(
       List<DatanodeDetails> datanodeList) {
     List<DatanodeDetails> sortedDatanodeList = new ArrayList<>(datanodeList);

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

@@ -17,7 +17,20 @@
 
 package org.apache.hadoop.hdds.scm.storage;
 
+import java.io.IOException;
+import java.time.Instant;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
 import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.OzoneClientConfig;
+import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier;
+import org.apache.hadoop.io.retry.RetryPolicy;
+import org.apache.hadoop.security.token.Token;
+import org.apache.ratis.thirdparty.io.grpc.Status;
 
 /**
  * Abstract class used as an interface for input streams related to Ozone
@@ -26,6 +39,8 @@
 public abstract class BlockExtendedInputStream extends ExtendedInputStream
     implements PartInputStream {
 
+  private static final org.slf4j.Logger LOG = org.slf4j.LoggerFactory.getLogger(BlockExtendedInputStream.class);
+
   public abstract BlockID getBlockID();
 
   @Override
@@ -38,4 +53,86 @@ public long getRemaining() {
 
   @Override
   public abstract long getPos();
+
+  protected Pipeline setPipeline(Pipeline pipeline) throws IOException {
+    if (pipeline == null) {
+      return null;
+    }
+    long replicaIndexes = pipeline.getNodes().stream().mapToInt(pipeline::getReplicaIndex).distinct().count();
+
+    if (replicaIndexes > 1) {
+      throw new IOException(String.format("Pipeline: %s has nodes containing different replica indexes.",
+          pipeline));
+    }
+
+    // irrespective of the container state, we will always read via Standalone protocol.
+    boolean okForRead = pipeline.getType() == HddsProtos.ReplicationType.STAND_ALONE
+            || pipeline.getType() == HddsProtos.ReplicationType.EC;
+    return okForRead ? pipeline : pipeline.copyForRead();
+  }
+
+  protected boolean shouldRetryRead(IOException cause, RetryPolicy retryPolicy, int retries) throws IOException {
+    RetryPolicy.RetryAction retryAction;
+    try {
+      retryAction = retryPolicy.shouldRetry(cause, retries, 0, true);
+    } catch (IOException e) {
+      throw e;
+    } catch (Exception e) {
+      throw new IOException(e);
+    }
+    if (retryAction.action == RetryPolicy.RetryAction.RetryDecision.RETRY) {
+      if (retryAction.delayMillis > 0) {
+        try {
+          LOG.debug("Retry read after {}ms", retryAction.delayMillis);
+          Thread.sleep(retryAction.delayMillis);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+          String msg = "Interrupted: action=" + retryAction.action + ", retry policy=" + retryPolicy;
+          throw new IOException(msg, e);
+        }
+      }
+      return true;
+    }
+    return false;
+  }
+
+  protected RetryPolicy getReadRetryPolicy(OzoneClientConfig config) {
+    return HddsClientUtils.createRetryPolicy(config.getMaxReadRetryCount(),
+        TimeUnit.SECONDS.toMillis(config.getReadRetryInterval()));
+  }
+
+  protected void refreshBlockInfo(IOException cause, BlockID blockID, AtomicReference<Pipeline> pipelineRef,
+      AtomicReference<Token<OzoneBlockTokenIdentifier>> tokenRef, Function<BlockID, BlockLocationInfo> refreshFunction)
+      throws IOException {
+    LOG.info("Attempting to update pipeline and block token for block {} from pipeline {}: {}",
+        blockID, pipelineRef.get().getId(), cause.getMessage());
+    if (refreshFunction != null) {
+      LOG.debug("Re-fetching pipeline and block token for block {}", blockID);
+      BlockLocationInfo blockLocationInfo = refreshFunction.apply(blockID);
+      if (blockLocationInfo == null) {
+        LOG.warn("No new block location info for block {}", blockID);
+      } else {
+        pipelineRef.set(setPipeline(blockLocationInfo.getPipeline()));
+        LOG.info("New pipeline for block {}: {}", blockID, blockLocationInfo.getPipeline());
+
+        tokenRef.set(blockLocationInfo.getToken());
+        if (blockLocationInfo.getToken() != null) {
+          OzoneBlockTokenIdentifier tokenId = new OzoneBlockTokenIdentifier();
+          tokenId.readFromByteArray(tokenRef.get().getIdentifier());
+          LOG.info("A new token is added for block {}. Expiry: {}", blockID,
+              Instant.ofEpochMilli(tokenId.getExpiryDate()));
+        }
+      }
+    } else {
+      throw cause;
+    }
+  }
+
+  /**
+   * Check if this exception is because datanodes are not reachable.
+   */
+  protected boolean isConnectivityIssue(IOException ex) {
+    return Status.fromThrowable(ex).getCode() == Status.UNAVAILABLE.getCode();
+  }
+
 }