HDDS-16722. Serve ReadBlock over Ratis read-only data streams on the datanode

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

@@ -0,0 +1,67 @@
+/*
+ * 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.ozone.container.common.transport.server.ratis;
+
+import java.io.IOException;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
+import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage;
+import org.apache.hadoop.ozone.container.common.interfaces.Container;
+import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
+import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController;
+import org.apache.ratis.protocol.RaftClientRequest;
+import org.apache.ratis.server.api.DataStreamApi;
+
+/**
+ * Serves ReadBlock on a Ratis read-only data stream, without a Raft group, for a container that is not open or closing.
+ * SCM gives the read pipeline of such a container a random ID, and no Raft group has this ID.
+ * The container no longer changes through Raft, so this datanode can serve the read from its replica.
+ */
+final class ClosedContainerReadResolver implements DataStreamApi.Resolver {
+  private final ContainerDispatcher dispatcher;
+  private final ContainerController containerController;
+  private final String datanodeUuid;
+
+  ClosedContainerReadResolver(ContainerDispatcher dispatcher, ContainerController containerController,
+      DatanodeDetails datanode) {
+    this.dispatcher = dispatcher;
+    this.containerController = containerController;
+    this.datanodeUuid = datanode.getUuidString();
+  }
+
+  /**
+   * @return the API to serve a ReadBlock of a container on this datanode that is not open or closing. Otherwise, null:
+   *     Ratis then serves the request with its Raft group.
+   */
+  @Override
+  public DataStreamApi resolve(RaftClientRequest request) throws IOException {
+    final ContainerCommandRequestProto requestProto = ContainerCommandRequestMessage.toProto(
+        request.getMessage().getContent(), request.getRaftGroupId());
+    if (requestProto.getCmdType() != Type.ReadBlock
+        || requestProto.hasDatanodeUuid() && !datanodeUuid.equals(requestProto.getDatanodeUuid())) {
+      return null;
+    }
+
+    final Container<?> container = containerController.getContainer(requestProto.getContainerID());
+    if (container == null || container.getContainerData().isOpen() || container.getContainerData().isClosing()) {
+      return null;
+    }
+    return (message, stream) -> ContainerStateMachine.streamReadBlock(dispatcher, requestProto, stream);
+  }
+}

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

@@ -29,6 +29,7 @@
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.nio.ByteBuffer;
+import java.nio.channels.WritableByteChannel;
 import java.nio.file.Files;
 import java.util.Arrays;
 import java.util.Collection;
@@ -53,8 +54,11 @@
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.client.BlockID;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.conf.StorageUnit;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
@@ -72,6 +76,7 @@
 import org.apache.hadoop.hdds.scm.utils.ClientCommandsUtils;
 import org.apache.hadoop.hdds.utils.Cache;
 import org.apache.hadoop.hdds.utils.ResourceCache;
+import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel;
 import org.apache.hadoop.ozone.HddsDatanodeService;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.common.utils.BufferUtils;
@@ -109,6 +114,9 @@
 import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.ratis.thirdparty.com.google.protobuf.TextFormat;
 import org.apache.ratis.thirdparty.com.google.protobuf.UnsafeByteOperations;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.apache.ratis.util.FileUtils;
 import org.apache.ratis.util.JavaUtils;
 import org.apache.ratis.util.LifeCycle;
@@ -124,6 +132,7 @@
  * <p>
  * The container requests can be divided into readonly request, WriteChunk request and other write requests.
  * - Read only requests (see {@link HddsUtils#isReadOnly}) are handled by {@link #query(Message)}.
+ * - ReadBlock requests on a Ratis read-only data stream are handled by {@link #transferTo}.
  * - WriteChunk request contains user data
  * - Other write request does not contain user data.
  * <p>
@@ -892,6 +901,97 @@ public CompletableFuture<Message> query(Message request) {
     }
   }
 
+  /** Serves a Ratis read-only data stream. Only ReadBlock is supported, see {@link #streamReadBlock}. */
+  @Override
+  public long transferTo(Message request, WritableByteChannel stream) throws IOException {
+    try {
+      metrics.incNumQueryStateMachineOps();
+      final ContainerCommandRequestProto requestProto = message2ContainerCommandRequestProto(request);
+      if (requestProto.getCmdType() != Type.ReadBlock) {
+        throw new IOException("Unsupported command for a read-only data stream: " + requestProto.getCmdType());
+      }
+      return streamReadBlock(dispatcher, requestProto, stream);
+    } catch (IOException e) {
+      metrics.incNumQueryStateMachineFails();
+      throw e;
+    }
+  }
+
+  /**
+   * Reads a block and writes each ReadBlock response to the stream as one reply:
+   * {@code [int: metadata length][metadata: the response without its data][data]}.
+   * Then closes the stream, which sends the terminal reply.
+   * If the read fails, this method throws without closing the stream. Ratis sends the failure as the terminal reply.
+   *
+   * @return the number of bytes written to the stream
+   */
+  static long streamReadBlock(ContainerDispatcher dispatcher, ContainerCommandRequestProto request,
+      WritableByteChannel stream) throws IOException {
+    final AtomicLong bytesWritten = new AtomicLong();
+    final AtomicReference<Throwable> error = new AtomicReference<>();
+    final StreamObserver<ContainerCommandResponseProto> observer = new StreamObserver<ContainerCommandResponseProto>() {
+      @Override
+      public void onNext(ContainerCommandResponseProto response) {
+        try {
+          bytesWritten.addAndGet(writeReadBlockReply(stream, response));
+        } catch (IOException e) {
+          error.set(e);
+          throw Status.CANCELLED.withDescription("Failed to write a ReadBlock reply").withCause(e).asRuntimeException();
+        }
+      }
+
+      @Override
+      public void onError(Throwable t) {
+        error.compareAndSet(null, t);
+      }
+
+      @Override
+      public void onCompleted() {
+      }
+    };
+
+    try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+      dispatcher.streamDataReadOnly(request, observer, blockFile, null);
+    } catch (StatusRuntimeException e) {
+      error.compareAndSet(null, e);
+    }
+    if (error.get() != null) {
+      throw new IOException("Failed to read block " + BlockID.getFromProtobuf(request.getReadBlock().getBlockID()),
+          error.get());
+    }
+    stream.close();
+    return bytesWritten.get();
+  }
+
+  private static int writeReadBlockReply(WritableByteChannel stream, ContainerCommandResponseProto response)
+      throws IOException {
+    final ByteString data;
+    final ContainerCommandResponseProto metadata;
+    if (response.hasReadBlock()) {
+      data = response.getReadBlock().getData();
+      metadata = response.toBuilder()
+          .setReadBlock(response.getReadBlock().toBuilder().setData(ByteString.EMPTY))
+          .build();
+    } else {
+      data = ByteString.EMPTY;
+      metadata = response;
+    }
+    final ByteString metadataBytes = metadata.toByteString();
+    final ByteBuffer reply = ByteBuffer.allocate(Integer.BYTES + metadataBytes.size() + data.size());
+    reply.putInt(metadataBytes.size());
+    metadataBytes.copyTo(reply);
+    data.copyTo(reply);
+    reply.flip();
+
+    // The client parses each reply on its own, so the reply must be sent in one write.
+    final int length = reply.remaining();
+    final int written = stream.write(reply);
+    if (written != length) {
+      throw new IOException("Failed to write a ReadBlock reply: only " + written + " of " + length + " bytes written");
+    }
+    return length;
+  }
+
   private ByteString readStateMachineData(
       ContainerCommandRequestProto requestProto, long term, long index)
       throws IOException {

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

@@ -525,6 +525,8 @@ public static XceiverServerRatis newXceiverServerRatis(HddsDatanodeService hddsD
       CertificateClient caClient, StateContext context) throws IOException {
     Parameters parameters = createTlsParameters(
         new SecurityConfig(ozoneConf), caClient);
+    RaftServerConfigKeys.DataStream.setServerApiResolver(parameters,
+        new ClosedContainerReadResolver(dispatcher, containerController, datanodeDetails));
 
     return new XceiverServerRatis(hddsDatanodeService, datanodeDetails, dispatcher,
         containerController, context, ozoneConf, parameters);
@@ -548,7 +550,7 @@ private static Parameters createTlsParameters(SecurityConfig conf,
       return RatisHelper.setServerTlsConf(serverConfig, clientConfig);
     }
 
-    return null;
+    return new Parameters();
   }
 
   @Override

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

@@ -175,6 +175,7 @@
 import org.apache.ratis.statemachine.StateMachine;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
 import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
 import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.apache.ratis.util.function.CheckedConsumer;
 import org.slf4j.Logger;
@@ -2316,6 +2317,8 @@ public ContainerCommandResponseProto readBlock(
       final StorageContainerException sce = new StorageContainerException(
           "Failed to readBlock " + request.getReadBlock(), ioe, IO_EXCEPTION);
       responseProto = ContainerUtils.logAndReturnError(LOG, sce, request);
+    } catch (StatusRuntimeException e) {
+      throw e;
     } catch (Exception e) {
       final StorageContainerException sce = new StorageContainerException(
           "Failed to readBlock " + request.getReadBlock(), e, CONTAINER_INTERNAL_ERROR);

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/impl/TestHddsDispatcher.java

@@ -28,9 +28,11 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.anyString;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -43,8 +45,10 @@
 import java.io.IOException;
 import java.nio.file.Path;
 import java.time.Duration;
+import java.util.ArrayList;
 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;
@@ -68,6 +72,7 @@
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.security.token.TokenVerifier;
+import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.common.Checksum;
 import org.apache.hadoop.ozone.common.ChecksumData;
@@ -96,12 +101,18 @@
 import org.apache.hadoop.ozone.container.keyvalue.ContainerLayoutTestInfo;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainer;
 import org.apache.hadoop.ozone.container.keyvalue.KeyValueContainerData;
+import org.apache.hadoop.ozone.container.ozoneimpl.OnDemandContainerScanner;
 import org.apache.hadoop.security.token.Token;
 import org.apache.ozone.test.GenericTestUtils.LogCapturer;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
+import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -639,6 +650,73 @@ public void testDuplicateWriteChunkAndPutBlockRequest() throws  IOException {
     }
   }
 
+  /**
+   * A ReadBlock whose stream is cancelled, for example because the client went away, is not a container problem, so
+   * the datanode must not scan the container. Any other failure while reading still makes it scan.
+   */
+  @ParameterizedTest
+  @ValueSource(booleans = {true, false})
+  public void testReadBlockScansContainerOnlyIfTheStreamIsNotCancelled(boolean cancelled) throws Exception {
+    String testDirPath = testDir.getPath();
+    try {
+      OzoneConfiguration conf = new OzoneConfiguration();
+      conf.set(HDDS_DATANODE_DIR_KEY, testDirPath);
+      conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, testDirPath);
+      DatanodeDetails dd = randomDatanodeDetails();
+      ContainerSet containerSet = newContainerSet();
+      OnDemandContainerScanner scanner = mock(OnDemandContainerScanner.class);
+      containerSet.registerOnDemandScanner(scanner);
+      HddsDispatcher hddsDispatcher = createDispatcher(dd, UUID.randomUUID(), conf, null, containerSet);
+      ContainerCommandRequestProto writeChunkRequest = getWriteChunkRequest(dd.getUuidString(), 1L, 1L);
+      assertEquals(ContainerProtos.Result.SUCCESS, hddsDispatcher.dispatch(writeChunkRequest, null).getResult());
+      assertEquals(ContainerProtos.Result.SUCCESS,
+          hddsDispatcher.dispatch(ContainerTestHelper.getPutBlockRequest(writeChunkRequest), null).getResult());
+      ContainerCommandRequestProto readBlockRequest = ContainerCommandRequestProto.newBuilder()
+          .setCmdType(ContainerProtos.Type.ReadBlock)
+          .setContainerID(writeChunkRequest.getContainerID())
+          .setDatanodeUuid(dd.getUuidString())
+          .setReadBlock(ContainerProtos.ReadBlockRequestProto.newBuilder()
+              .setBlockID(writeChunkRequest.getWriteChunk().getBlockID())
+              .setOffset(0)
+              .setLength(writeChunkRequest.getWriteChunk().getChunkData().getLen()))
+          .build();
+
+      // The stream fails on the data response, and takes the error response that the dispatcher may send after it.
+      RuntimeException failure = cancelled ? Status.CANCELLED.asRuntimeException() : new IllegalStateException();
+      List<ContainerCommandResponseProto> errorResponses = new ArrayList<>();
+      StreamObserver<ContainerCommandResponseProto> stream = new StreamObserver<ContainerCommandResponseProto>() {
+        @Override
+        public void onNext(ContainerCommandResponseProto response) {
+          if (response.hasReadBlock()) {
+            throw failure;
+          }
+          errorResponses.add(response);
+        }
+
+        @Override
+        public void onError(Throwable t) {
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+      };
+      try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) {
+        if (cancelled) {
+          assertSame(failure, assertThrows(StatusRuntimeException.class,
+              () -> hddsDispatcher.streamDataReadOnly(readBlockRequest, stream, blockFile, null)));
+        } else {
+          hddsDispatcher.streamDataReadOnly(readBlockRequest, stream, blockFile, null);
+        }
+      }
+
+      verify(scanner, times(cancelled ? 0 : 1)).scanContainer(any(), anyString());
+      assertEquals(cancelled ? 0 : 1, errorResponses.size());
+    } finally {
+      ContainerMetrics.remove();
+    }
+  }
+
   /**
    * Creates HddsDispatcher instance with given infos.
    * @param dd datanode detail info.
@@ -654,7 +732,11 @@ static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId,
 
   static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId,
       OzoneConfiguration conf, TokenVerifier tokenVerifier) throws IOException {
-    ContainerSet containerSet = newContainerSet();
+    return createDispatcher(dd, scmId, conf, tokenVerifier, newContainerSet());
+  }
+
+  static HddsDispatcher createDispatcher(DatanodeDetails dd, UUID scmId, OzoneConfiguration conf,
+      TokenVerifier tokenVerifier, ContainerSet containerSet) throws IOException {
     MutableVolumeSet volumeSet = new MutableVolumeSet(dd.getUuidString(), conf, null,
         StorageVolume.VolumeType.DATA_VOLUME, null);
     volumeSet.getVolumesList().stream().forEach(v -> {