diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java index dba10525d2bd..1a952d8b8cf5 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java @@ -130,6 +130,22 @@ public class OzoneClientConfig { tags = ConfigTag.CLIENT) private boolean streamReadBlock = false; + @Config(key = "ozone.client.ratis.stream.readblock.enable", + defaultValue = "false", + type = ConfigType.BOOLEAN, + description = "Allow ReadBlock to use Ratis data stream read-only requests.", + tags = ConfigTag.CLIENT) + private boolean ratisStreamReadBlock = false; + + @Config(key = "ozone.client.ratis.stream.read.window-size", + defaultValue = "268435456", + type = ConfigType.LONG, + tags = {ConfigTag.CLIENT}, + description = "Target request window for sequential ReadBlock requests " + + "over Ratis data stream. Larger windows reduce data stream setup " + + "overhead for large reads.") + private long ratisStreamReadWindowSize = 256L << 20; + @Config(key = "ozone.client.max.retries", defaultValue = "5", description = "Maximum number of retries by Ozone Client on " @@ -382,6 +398,13 @@ public void validate() { streamReadPreReadSize = 32L << 20; // 32MB } + if (ratisStreamReadWindowSize < 0) { + LOG.warn("Invalid ozone.client.ratis.stream.read.window-size = {}. " + + "Resetting to default 256MB.", + ratisStreamReadWindowSize); + ratisStreamReadWindowSize = 256L << 20; // 256MB + } + // Ensure response data size is positive. if (streamReadResponseDataSize <= 0) { LOG.warn("Invalid ozone.client.stream.read.response-data-size = {}. " + @@ -622,6 +645,14 @@ public void setStreamReadBlock(boolean streamReadBlock) { this.streamReadBlock = streamReadBlock; } + public boolean isRatisStreamReadBlock() { + return ratisStreamReadBlock; + } + + public void setRatisStreamReadBlock(boolean ratisStreamReadBlock) { + this.ratisStreamReadBlock = ratisStreamReadBlock; + } + public long getStreamReadPreReadSize() { return streamReadPreReadSize; } @@ -630,6 +661,10 @@ public int getStreamReadResponseDataSize() { return streamReadResponseDataSize; } + public long getRatisStreamReadWindowSize() { + return ratisStreamReadWindowSize; + } + public Duration getStreamReadTimeout() { return streamReadTimeout; } @@ -642,6 +677,10 @@ public void setStreamReadResponseDataSize(int streamReadResponseDataSize) { this.streamReadResponseDataSize = streamReadResponseDataSize; } + public void setRatisStreamReadWindowSize(long ratisStreamReadWindowSize) { + this.ratisStreamReadWindowSize = ratisStreamReadWindowSize; + } + public void setStreamReadTimeout(Duration streamReadTimeout) { this.streamReadTimeout = streamReadTimeout; } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index 221a48be828d..d08bff8766b6 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -66,7 +66,9 @@ public MultipartInputStream(String keyName, this.key = keyName; this.partStreams = Collections.unmodifiableList(inputStreams); - this.isStreamBlockInputStream = !inputStreams.isEmpty() && inputStreams.get(0) instanceof StreamBlockInputStream; + this.isStreamBlockInputStream = !inputStreams.isEmpty() + && (inputStreams.get(0) instanceof StreamBlockInputStream + || inputStreams.get(0) instanceof RatisDataStreamBlockInputStream); // Calculate and update the partOffsets this.partOffsets = new long[inputStreams.size()]; @@ -75,7 +77,9 @@ public MultipartInputStream(String keyName, for (PartInputStream partInputStream : inputStreams) { this.partOffsets[i++] = streamLength; if (isStreamBlockInputStream) { - Preconditions.assertInstanceOf(partInputStream, StreamBlockInputStream.class); + Preconditions.assertTrue(partInputStream instanceof StreamBlockInputStream + || partInputStream instanceof RatisDataStreamBlockInputStream, + () -> "Unexpected stream block input stream " + partInputStream); } streamLength += partInputStream.getLength(); } @@ -199,7 +203,12 @@ public boolean readFully(long position, ByteBuffer buffer) throws IOException { read(new ByteBufferReader(buffer) { @Override int readImpl(InputStream inputStream) throws IOException { - return Preconditions.assertInstanceOf(inputStream, StreamBlockInputStream.class) + if (inputStream instanceof StreamBlockInputStream) { + return ((StreamBlockInputStream) inputStream) + .readFully(getBuffer(), false); + } + return Preconditions.assertInstanceOf(inputStream, + RatisDataStreamBlockInputStream.class) .readFully(getBuffer(), false); } }); diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisDataStreamBlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisDataStreamBlockInputStream.java new file mode 100644 index 000000000000..1c8b58510053 --- /dev/null +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/RatisDataStreamBlockInputStream.java @@ -0,0 +1,476 @@ +/* + * 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.scm.storage; + +import com.google.common.annotations.VisibleForTesting; +import java.io.EOFException; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.time.Duration; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicReference; +import org.apache.commons.lang3.NotImplementedException; +import org.apache.hadoop.hdds.client.BlockID; +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.ContainerCommandResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; +import org.apache.hadoop.hdds.scm.XceiverClientRatis; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; +import org.apache.hadoop.hdds.tracing.TracingUtil; +import org.apache.hadoop.ozone.common.Checksum; +import org.apache.hadoop.ozone.common.ChecksumData; +import org.apache.hadoop.security.token.Token; +import org.apache.ratis.client.api.DataStreamInput; +import org.apache.ratis.client.impl.ClientProtoUtils; +import org.apache.ratis.proto.RaftProtos.DataStreamPacketHeaderProto.Type; +import org.apache.ratis.protocol.DataStreamReply; +import org.apache.ratis.protocol.RaftClientReply; +import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException; +import org.apache.ratis.util.Preconditions; +import org.apache.ratis.util.ReferenceCountedObject; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Reads RATIS blocks exclusively through the Ratis data stream read-only API. + */ +public class RatisDataStreamBlockInputStream extends BlockExtendedInputStream { + private static final Logger LOG = + LoggerFactory.getLogger(RatisDataStreamBlockInputStream.class); + private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0); + private static final int RATIS_READ_BLOCK_STREAM_HEADER_BYTES = + Integer.BYTES; + + private final BlockID blockID; + private final long blockLength; + private final AtomicReference pipelineRef = new AtomicReference<>(); + private final AtomicReference> tokenRef = + new AtomicReference<>(); + private final XceiverClientFactory xceiverClientFactory; + private final boolean verifyChecksum; + private final long preReadSize; + private final long readWindowSize; + private final int responseDataSize; + private final Duration readTimeout; + + private XceiverClientRatis xceiverClient; + private DataStreamInput streamInput; + private ByteBuffer buffer = EMPTY_BUFFER; + private ReferenceCountedObject retainedDataReply; + private long position; + private boolean streamDataSeen; + private boolean largeReadWindowEligible; + private boolean closed; + + public RatisDataStreamBlockInputStream(BlockID blockID, long length, + Pipeline pipeline, Token token, + XceiverClientFactory xceiverClientFactory, + OzoneClientConfig config) throws IOException { + this.blockID = Objects.requireNonNull(blockID, "blockID == null"); + this.blockLength = length; + pipelineRef.set(setRatisPipeline(pipeline)); + tokenRef.set(token); + this.xceiverClientFactory = Objects.requireNonNull(xceiverClientFactory, + "xceiverClientFactory == null"); + Objects.requireNonNull(config, "config == null"); + this.verifyChecksum = config.isChecksumVerify(); + this.preReadSize = config.getStreamReadPreReadSize(); + this.readWindowSize = config.getRatisStreamReadWindowSize(); + this.responseDataSize = config.getStreamReadResponseDataSize(); + this.readTimeout = config.getStreamReadTimeout(); + } + + @Override + public BlockID getBlockID() { + return blockID; + } + + @Override + public long getLength() { + return blockLength; + } + + @Override + public long getPos() { + return position; + } + + @Override + public synchronized int read(byte[] b, int off, int len) throws IOException { + Objects.requireNonNull(b, "b == null"); + if (off < 0 || len < 0 || len > b.length - off) { + throw new IndexOutOfBoundsException(); + } + return read(ByteBuffer.wrap(b, off, len)); + } + + @Override + public synchronized int read(ByteBuffer targetBuf) throws IOException { + return readFully(targetBuf, true); + } + + public synchronized int readFully(ByteBuffer targetBuf, boolean preRead) + throws IOException { + checkOpen(); + if (!targetBuf.hasRemaining()) { + return 0; + } + int read = 0; + while (targetBuf.hasRemaining() && position < blockLength) { + if (!buffer.hasRemaining()) { + releaseRetainedDataReply(); + buffer = readBlock(targetBuf.remaining(), preRead); + } + if (!buffer.hasRemaining()) { + break; + } + + final int toCopy = Math.min(buffer.remaining(), targetBuf.remaining()); + final ByteBuffer tmp = buffer.duplicate(); + tmp.limit(tmp.position() + toCopy); + targetBuf.put(tmp); + buffer.position(tmp.position()); + position += toCopy; + read += toCopy; + if (!buffer.hasRemaining()) { + releaseRetainedDataReply(); + } + } + return read > 0 ? read : EOF; + } + + @Override + protected int readWithStrategy(ByteReaderStrategy strategy) { + throw new NotImplementedException("readWithStrategy is not implemented."); + } + + @Override + public synchronized void seek(long pos) throws IOException { + checkOpen(); + if (pos < 0) { + throw new IOException("Cannot seek to negative offset"); + } + if (pos > blockLength) { + throw new EOFException("Failed to seek to position " + pos + + " > block length = " + blockLength); + } + if (pos != position) { + closeStream(); + position = pos; + discardBufferedData(); + largeReadWindowEligible = false; + } + } + + @Override + public synchronized boolean seekToNewSource(long targetPos) + throws IOException { + return false; + } + + @Override + public synchronized void unbuffer() { + discardBufferedData(); + releaseClient(false); + } + + @Override + public synchronized void close() { + closed = true; + discardBufferedData(); + closeStream(); + releaseClient(false); + } + + private ByteBuffer readBlock(int length, boolean preRead) throws IOException { + while (position < blockLength) { + if (streamInput == null) { + streamDataSeen = false; + streamInput = openStream(length, preRead); + } + + final ReferenceCountedObject ref = readReply(); + final DataStreamReply reply = ref.get(); + if (reply.getType() == Type.STREAM_DATA) { + streamDataSeen = true; + final ByteBuffer data = readDataReply(ref); + if (data.hasRemaining()) { + return data; + } + } else if (reply.getType() == Type.STREAM_HEADER) { + handleTerminalReply(ref); + if (!streamDataSeen && position < blockLength) { + throw new EOFException("ReadBlock stream returned no data for " + + blockID + " at position " + position); + } + largeReadWindowEligible = streamDataSeen; + closeStream(); + } else { + try { + throw new IOException("Unexpected data stream reply type " + + reply.getType() + " for " + blockID); + } finally { + ref.release(); + } + } + } + return ByteBuffer.allocate(0); + } + + private DataStreamInput openStream(int length, boolean preRead) + throws IOException { + final long readLength = computeReadLength(blockLength, position, length, + preRead, largeReadWindowEligible, preReadSize, readWindowSize); + final ContainerCommandRequestProto request = + ContainerProtocolCalls.buildReadBlockCommandProto(blockID, position, + readLength, responseDataSize, tokenRef.get(), pipelineRef.get()); + acquireClient(); + final ContainerCommandRequestMessage message = + ContainerCommandRequestMessage.toMessage(request, + TracingUtil.exportCurrentSpan()); + return xceiverClient.getDataStreamApi() + .streamReadOnly(message.getContent().asReadOnlyByteBuffer()); + } + + private ReferenceCountedObject readReply() throws IOException { + try { + return streamInput.readAsync().get(readTimeout.toMillis(), + TimeUnit.MILLISECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException("Interrupted Ratis read-only data stream request", + e); + } catch (ExecutionException e) { + releaseClient(true); + final Throwable cause = e.getCause(); + if (cause instanceof IOException) { + throw (IOException) cause; + } + throw new IOException("Failed Ratis read-only data stream request", + cause != null ? cause : e); + } catch (TimeoutException e) { + releaseClient(true); + throw new IOException("Timed out waiting for Ratis read-only data " + + "stream reply", e); + } + } + + private ByteBuffer readDataReply(ReferenceCountedObject ref) + throws IOException { + final DataStreamReply reply = ref.get(); + boolean releaseReply = true; + try { + final ReadBlockData readBlockData = parseReadBlockData(reply); + final ContainerCommandResponseProto response = readBlockData.response; + ContainerProtocolCalls.validateContainerResponse(response); + if (!response.hasReadBlock()) { + throw new IOException("Missing ReadBlock response: " + response); + } + + final ReadBlockResponseProto readBlock = response.getReadBlock(); + final ByteBuffer dataBuffer = readBlockData.data != null ? + readBlockData.data : readBlock.getData().asReadOnlyByteBuffer(); + if (verifyChecksum) { + final ChecksumData checksumData = + ChecksumData.getFromProtoBuf(readBlock.getChecksumData()); + Checksum.verifyChecksum(dataBuffer.duplicate(), checksumData, 0); + } + + final long blockOffset = readBlock.getOffset(); + if (position < blockOffset) { + throw new IOException("ReadBlock response is ahead of requested " + + "position " + position + ", response offset " + blockOffset); + } + dataBuffer.position(Math.toIntExact( + Math.min(position - blockOffset, dataBuffer.limit()))); + if (dataBuffer.hasRemaining()) { + retainedDataReply = ref; + releaseReply = false; + } + return dataBuffer; + } finally { + if (releaseReply) { + ref.release(); + } + } + } + + private void handleTerminalReply(ReferenceCountedObject ref) + throws IOException { + try { + final RaftClientReply raftReply = + ClientProtoUtils.getRaftClientReply(ref.get()); + if (!raftReply.isSuccess()) { + throw new IOException("Failed Ratis read-only data stream request", + raftReply.getException()); + } + } finally { + ref.release(); + } + } + + private ReadBlockData parseReadBlockData(DataStreamReply reply) + throws IOException { + try { + return parseReadBlockData(reply.nioBuffer()); + } catch (InvalidProtocolBufferException e) { + releaseClient(true); + throw new IOException("Failed to parse ReadBlock response", e); + } + } + + private ReadBlockData parseReadBlockData(ByteBuffer replyBuffer) + throws InvalidProtocolBufferException { + final ByteBuffer duplicate = replyBuffer.duplicate(); + if (duplicate.remaining() < RATIS_READ_BLOCK_STREAM_HEADER_BYTES) { + throw new InvalidProtocolBufferException( + "Missing Ratis ReadBlock metadata length"); + } + final int metadataLength = duplicate.getInt(duplicate.position()); + if (metadataLength < 0 + || metadataLength > duplicate.remaining() + - RATIS_READ_BLOCK_STREAM_HEADER_BYTES) { + throw new InvalidProtocolBufferException( + "Invalid Ratis ReadBlock metadata length " + metadataLength); + } + duplicate.position( + duplicate.position() + RATIS_READ_BLOCK_STREAM_HEADER_BYTES); + final ByteBuffer metadata = duplicate.slice(); + metadata.limit(metadataLength); + duplicate.position(duplicate.position() + metadataLength); + final ByteBuffer data = duplicate.slice(); + return new ReadBlockData( + ContainerCommandResponseProto.parseFrom(metadata), data); + } + + private static final class ReadBlockData { + private final ContainerCommandResponseProto response; + private final ByteBuffer data; + + private ReadBlockData(ContainerCommandResponseProto response, + ByteBuffer data) { + this.response = response; + this.data = data; + } + } + + private synchronized void acquireClient() throws IOException { + checkOpen(); + if (xceiverClient == null) { + final XceiverClientSpi client = + xceiverClientFactory.acquireClientForReadData(pipelineRef.get()); + if (!(client instanceof XceiverClientRatis)) { + xceiverClientFactory.releaseClientForReadData(client, false); + throw new IOException("Unexpected client class: " + + client.getClass().getName() + ", " + pipelineRef.get()); + } + xceiverClient = (XceiverClientRatis) client; + } + } + + private synchronized void releaseClient(boolean invalidateClient) { + discardBufferedData(); + largeReadWindowEligible = false; + if (xceiverClient != null) { + closeStream(); + xceiverClientFactory.releaseClientForReadData(xceiverClient, + invalidateClient); + xceiverClient = null; + } + } + + private synchronized void closeStream() { + if (streamInput != null) { + try { + streamInput.close(); + } catch (IOException e) { + LOG.debug("Failed to close Ratis read-only stream for {}", blockID, e); + } finally { + streamInput = null; + streamDataSeen = false; + } + } + } + + private void discardBufferedData() { + buffer = EMPTY_BUFFER; + releaseRetainedDataReply(); + } + + private void releaseRetainedDataReply() { + if (retainedDataReply != null) { + retainedDataReply.release(); + retainedDataReply = null; + } + } + + private static long addWithSaturation(long left, long right) { + final long result = left + right; + return result < 0 ? Long.MAX_VALUE : result; + } + + private static long computeReadLength(long blockLength, long position, + int length, boolean preRead, boolean largeReadWindowEligible, + long preReadSize, long readWindowSize) { + final long requestedLength = Math.max(1L, (long) length); + long wantedLength = requestedLength; + if (preRead && largeReadWindowEligible) { + wantedLength = Math.max(addWithSaturation(requestedLength, preReadSize), + readWindowSize); + } + return Math.min(blockLength - position, wantedLength); + } + + @VisibleForTesting + static long computeReadLengthForTesting(long blockLength, long position, + int length, boolean preRead, boolean largeReadWindowEligible, + long preReadSize, long readWindowSize) { + return computeReadLength(blockLength, position, length, preRead, + largeReadWindowEligible, preReadSize, readWindowSize); + } + + private void checkOpen() throws IOException { + if (closed) { + throw new IOException("Stream is closed for block " + blockID); + } + } + + private static Pipeline setRatisPipeline(Pipeline pipeline) throws IOException { + Objects.requireNonNull(pipeline, "pipeline == null"); + Preconditions.assertTrue( + pipeline.getType() == HddsProtos.ReplicationType.RATIS, + () -> "Expected RATIS pipeline but got " + pipeline); + for (DatanodeDetails dn : pipeline.getNodes()) { + if (!dn.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)) { + throw new IOException("RATIS_DATASTREAM port is missing for datanode " + + dn + " in pipeline " + pipeline.getId()); + } + } + return pipeline; + } +} diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java index 02527129d5a4..f37e2b87756e 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/BlockInputStreamFactoryImpl.java @@ -17,6 +17,7 @@ package org.apache.hadoop.ozone.client.io; +import static org.apache.hadoop.hdds.DatanodeVersion.RATIS_DATASTREAM_READ_BLOCK_SUPPORT; import static org.apache.hadoop.hdds.DatanodeVersion.STREAM_BLOCK_SUPPORT; import java.io.IOException; @@ -24,6 +25,7 @@ import java.util.concurrent.Executors; import java.util.function.Function; import java.util.function.Supplier; +import org.apache.hadoop.hdds.DatanodeVersion; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; @@ -35,6 +37,7 @@ import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream; import org.apache.hadoop.hdds.scm.storage.BlockInputStream; import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo; +import org.apache.hadoop.hdds.scm.storage.RatisDataStreamBlockInputStream; import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.io.ByteBufferPool; @@ -89,7 +92,15 @@ public BlockExtendedInputStream create(ReplicationConfig repConfig, return new ECBlockInputStreamProxy((ECReplicationConfig)repConfig, blockInfo, xceiverFactory, refreshFunction, ecBlockStreamFactory, config); - } else if (config.isStreamReadBlock() && allDataNodesSupportStreamBlock(pipeline)) { + } else if (config.isRatisStreamReadBlock() + && repConfig.getReplicationType().equals(HddsProtos.ReplicationType.RATIS) + && allDataNodesSupport(pipeline, + RATIS_DATASTREAM_READ_BLOCK_SUPPORT)) { + return new RatisDataStreamBlockInputStream(blockInfo.getBlockID(), + blockInfo.getLength(), pipeline, token, xceiverFactory, + config); + } else if (config.isStreamReadBlock() + && allDataNodesSupport(pipeline, STREAM_BLOCK_SUPPORT)) { return new StreamBlockInputStream(blockInfo.getBlockID(), blockInfo.getLength(), pipeline, token, xceiverFactory, refreshFunction, config); } else { @@ -99,11 +110,12 @@ public BlockExtendedInputStream create(ReplicationConfig repConfig, } } - private boolean allDataNodesSupportStreamBlock(Pipeline pipeline) { + private boolean allDataNodesSupport(Pipeline pipeline, + DatanodeVersion requiredVersion) { // return true only if all DataNodes in the pipeline are on a version - // that supports for reading a block by streaming chunks.. + // that supports the requested read path. for (DatanodeDetails dn : pipeline.getNodes()) { - if (dn.getCurrentVersion() < STREAM_BLOCK_SUPPORT.toProtoValue()) { + if (dn.getCurrentVersion() < requiredVersion.toProtoValue()) { return false; } } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestRatisDataStreamBlockInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestRatisDataStreamBlockInputStream.java new file mode 100644 index 000000000000..081fe2123f35 --- /dev/null +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestRatisDataStreamBlockInputStream.java @@ -0,0 +1,51 @@ +/* + * 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.scm.storage; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import org.junit.jupiter.api.Test; + +/** + * Tests for {@link RatisDataStreamBlockInputStream}. + */ +class TestRatisDataStreamBlockInputStream { + private static final long ONE_GB = 1L << 30; + private static final long READ_WINDOW = 256L << 20; + private static final long PRE_READ = 32L << 20; + + @Test + void readLengthUsesLargeWindowOnlyAfterSequentialStream() { + final int smallRead = 4 << 10; + + assertEquals(smallRead, + RatisDataStreamBlockInputStream.computeReadLengthForTesting( + ONE_GB, 0, smallRead, true, false, PRE_READ, READ_WINDOW)); + assertEquals(READ_WINDOW, + RatisDataStreamBlockInputStream.computeReadLengthForTesting( + ONE_GB, PRE_READ + smallRead, smallRead, true, true, PRE_READ, + READ_WINDOW)); + assertEquals(smallRead, + RatisDataStreamBlockInputStream.computeReadLengthForTesting( + ONE_GB, 0, smallRead, false, true, PRE_READ, READ_WINDOW)); + assertEquals(1024, + RatisDataStreamBlockInputStream.computeReadLengthForTesting( + ONE_GB, ONE_GB - 1024, smallRead, true, true, PRE_READ, + READ_WINDOW)); + } +} diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestBlockInputStreamFactoryImpl.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestBlockInputStreamFactoryImpl.java index 2fd81fc91dc0..ee8cb178a674 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestBlockInputStreamFactoryImpl.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/ozone/client/io/TestBlockInputStreamFactoryImpl.java @@ -17,6 +17,8 @@ package org.apache.hadoop.ozone.client.io; +import static org.apache.hadoop.hdds.DatanodeVersion.RATIS_DATASTREAM_READ_BLOCK_SUPPORT; +import static org.apache.hadoop.hdds.DatanodeVersion.STREAM_BLOCK_SUPPORT; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.mockito.ArgumentMatchers.any; @@ -34,11 +36,13 @@ import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream; import org.apache.hadoop.hdds.scm.storage.BlockInputStream; import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo; +import org.apache.hadoop.hdds.scm.storage.RatisDataStreamBlockInputStream; import org.apache.hadoop.hdds.scm.storage.StreamBlockInputStream; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; @@ -80,6 +84,56 @@ public void testNonECGivesBlockInputStream(boolean streamReadBlockEnabled) throw assertEquals(stream.getLength(), blockInfo.getLength()); } + @Test + public void testRatisStreamReadBlockWhenAllDatanodesSupport() + throws IOException { + BlockInputStreamFactory factory = new BlockInputStreamFactoryImpl(); + ReplicationConfig repConfig = + RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE); + BlockLocationInfo blockInfo = createKeyLocationInfo(repConfig, 3, + 1024 * 1024 * 10); + blockInfo.getPipeline().getNodes().forEach(dn -> dn.setCurrentVersion( + RATIS_DATASTREAM_READ_BLOCK_SUPPORT.toProtoValue())); + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setRatisStreamReadBlock(true); + + BlockExtendedInputStream stream = factory.create(repConfig, blockInfo, + blockInfo.getPipeline(), blockInfo.getToken(), + Mockito.mock(XceiverClientFactory.class), null, clientConfig); + + assertInstanceOf(RatisDataStreamBlockInputStream.class, stream); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + public void testRatisStreamReadBlockUsesCompatibilityPathForOlderDatanode( + boolean streamReadBlockEnabled) throws IOException { + BlockInputStreamFactory factory = new BlockInputStreamFactoryImpl(); + ReplicationConfig repConfig = + RatisReplicationConfig.getInstance(HddsProtos.ReplicationFactor.THREE); + BlockLocationInfo blockInfo = createKeyLocationInfo(repConfig, 3, + 1024 * 1024 * 10); + blockInfo.getPipeline().getNodes().get(0) + .setCurrentVersion(STREAM_BLOCK_SUPPORT.toProtoValue()); + Pipeline pipeline = Mockito.spy(blockInfo.getPipeline()); + blockInfo.setPipeline(pipeline); + Mockito.when(pipeline.getReplicaIndex(any(DatanodeDetails.class))) + .thenReturn(1); + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setRatisStreamReadBlock(true); + clientConfig.setStreamReadBlock(streamReadBlockEnabled); + + BlockExtendedInputStream stream = factory.create(repConfig, blockInfo, + blockInfo.getPipeline(), blockInfo.getToken(), null, null, + clientConfig); + + if (streamReadBlockEnabled) { + assertInstanceOf(StreamBlockInputStream.class, stream); + } else { + assertInstanceOf(BlockInputStream.class, stream); + } + } + @Test public void testECGivesECBlockInputStream() throws IOException { BlockInputStreamFactory factory = new BlockInputStreamFactoryImpl(); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java index 2717e8eb3d9d..388dced66e0e 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/DatanodeVersion.java @@ -35,6 +35,9 @@ public enum DatanodeVersion implements ComponentVersion { "a PutBlock request"), STREAM_BLOCK_SUPPORT(3, "This version has support for reading a block by streaming chunks."), + RATIS_DATASTREAM_READ_BLOCK_SUPPORT(4, + "This version supports Ratis DataStream ReadBlock, including " + + "group-independent reads for closed containers."), FUTURE_VERSION(-1, "Used internally in the client when the server side is " + " newer and an unknown server version has arrived to the client."); diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/ContainerCommandResponseBuilders.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/ContainerCommandResponseBuilders.java index c60f3d7449a0..66c13b64ff09 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/ContainerCommandResponseBuilders.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocolPB/ContainerCommandResponseBuilders.java @@ -337,10 +337,16 @@ public static ContainerCommandResponseProto getReadChunkResponse( public static ContainerCommandResponseProto getReadBlockResponse( ContainerCommandRequestProto request, ChecksumData checksumData, ByteBuffer data, long offset) { + return getReadBlockResponse(request, checksumData, + ByteString.copyFrom(data), offset); + } + public static ContainerCommandResponseProto getReadBlockResponse( + ContainerCommandRequestProto request, ChecksumData checksumData, + ByteString data, long offset) { ContainerProtos.ReadBlockResponseProto response = ContainerProtos.ReadBlockResponseProto.newBuilder() .setChecksumData(checksumData.getProtoBufMessage()) - .setData(ByteString.copyFrom(data)) + .setData(data) .setOffset(offset) .build(); 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 249e58df94d4..8d551cb38390 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 @@ -70,6 +70,7 @@ import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; import org.apache.hadoop.ozone.container.common.interfaces.Container; import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.interfaces.Handler; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext; @@ -77,8 +78,8 @@ import org.apache.hadoop.ozone.container.ozoneimpl.ContainerScanError; import org.apache.hadoop.ozone.container.ozoneimpl.DataScanResult; import org.apache.hadoop.util.Time; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.statemachine.StateMachine; -import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.apache.ratis.util.UncheckedAutoCloseable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -823,7 +824,7 @@ public StateMachine.DataChannel getStreamDataChannel( @Override public void streamDataReadOnly(ContainerCommandRequestProto msg, - StreamObserver streamObserver, + DataStreamObserver streamObserver, RandomAccessFileChannel blockFile, DispatcherContext dispatcherContext) { Objects.requireNonNull(msg, "msg == null"); Type cmdType = msg.getCmdType(); @@ -875,20 +876,22 @@ public void streamDataReadOnly(ContainerCommandRequestProto msg, containerSet.scanContainer(containerID, "ReadBlock failed " + responseProto.getResult()); audit(action, eventType, msg, dispatcherContext, AuditEventStatus.FAILURE, new Exception(responseProto.getMessage())); - streamObserver.onNext(responseProto); + streamObserver.onNext(new ReadBlockResponse(responseProto, null)); } perf.appendOpLatencyMs(oPLatencyMS); performanceAudit(action, msg, dispatcherContext, perf, oPLatencyMS); } catch (StorageContainerException sce) { audit(action, eventType, msg, dispatcherContext, AuditEventStatus.FAILURE, sce); - streamObserver.onNext(ContainerUtils.logAndReturnError(LOG, sce, msg)); + streamObserver.onNext(new ReadBlockResponse( + ContainerUtils.logAndReturnError(LOG, sce, msg), null)); } catch (IOException ioe) { final String s = ContainerProtos.Result.BLOCK_TOKEN_VERIFICATION_FAILED + " for " + dispatcherContext + ": " + ioe.getMessage(); final StorageContainerException sce = new StorageContainerException( s, ioe, ContainerProtos.Result.BLOCK_TOKEN_VERIFICATION_FAILED); - streamObserver.onNext(ContainerUtils.logAndReturnError(LOG, sce, msg)); + streamObserver.onNext(new ReadBlockResponse( + ContainerUtils.logAndReturnError(LOG, sce, msg), null)); } finally { span.end(); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerDispatcher.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerDispatcher.java index 2aba8253cefe..7b6756007c7f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerDispatcher.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/ContainerDispatcher.java @@ -17,6 +17,7 @@ package org.apache.hadoop.ozone.container.common.interfaces; +import java.nio.ByteBuffer; import java.util.Map; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; @@ -24,8 +25,8 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.statemachine.StateMachine; -import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; /** * Dispatcher acts as the bridge between the transport layer and @@ -36,6 +37,26 @@ * The reply from the request is dispatched to the client. */ public interface ContainerDispatcher { + /** A ReadBlock response with its data kept outside the protobuf metadata. */ + final class ReadBlockResponse { + private final ContainerCommandResponseProto response; + private final ByteBuffer data; + + public ReadBlockResponse(ContainerCommandResponseProto response, + ByteBuffer data) { + this.response = response; + this.data = data; + } + + public ContainerCommandResponseProto getResponse() { + return response; + } + + public ByteBuffer getData() { + return data; + } + } + /** * Dispatches commands to container layer. * @param msg - Command Request @@ -97,7 +118,7 @@ default StateMachine.DataChannel getStreamDataChannel( */ default void streamDataReadOnly( ContainerCommandRequestProto msg, - StreamObserver streamObserver, + DataStreamObserver streamObserver, RandomAccessFileChannel blockFile, DispatcherContext dispatcherContext) { throw new UnsupportedOperationException("streamDataReadOnly not supported."); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java index fec625f89a08..68256c2859e5 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Handler.java @@ -39,13 +39,14 @@ import org.apache.hadoop.ozone.container.common.helpers.ContainerMetrics; import org.apache.hadoop.ozone.container.common.impl.ContainerData; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.report.IncrementalReportSender; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext; import org.apache.hadoop.ozone.container.common.volume.VolumeSet; import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler; import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.statemachine.StateMachine; -import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; /** * Dispatcher sends ContainerCommandRequests to Handler. Each Container Type @@ -282,6 +283,6 @@ public abstract void copyContainer( public abstract ContainerCommandResponseProto readBlock( ContainerCommandRequestProto msg, Container container, RandomAccessFileChannel blockFile, - StreamObserver streamObserver); + DataStreamObserver streamObserver); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java index 1516a382688f..891dfe50ac3c 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/GrpcXceiverService.java @@ -26,8 +26,11 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.XceiverClientProtocolServiceGrpc; import org.apache.hadoop.hdds.utils.io.RandomAccessFileChannel; import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.transport.server.ratis.DispatcherContext; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.grpc.util.ZeroCopyMessageMarshaller; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.apache.ratis.thirdparty.com.google.protobuf.MessageLite; import org.apache.ratis.thirdparty.io.grpc.MethodDescriptor; import org.apache.ratis.thirdparty.io.grpc.ServerCallHandler; @@ -117,7 +120,28 @@ public void onNext(ContainerCommandRequestProto request) { try { if (request.getCmdType() == Type.ReadBlock) { - dispatcher.streamDataReadOnly(request, responseObserver, blockFile, context); + dispatcher.streamDataReadOnly(request, + new DataStreamObserver() { + @Override + public void onNext(ReadBlockResponse response) { + responseObserver.onNext(response.getData() == null + ? response.getResponse() + : response.getResponse().toBuilder() + .setReadBlock(response.getResponse().getReadBlock().toBuilder() + .setData(ByteString.copyFrom(response.getData()))) + .build()); + } + + @Override + public void onCompleted() { + responseObserver.onCompleted(); + } + + @Override + public void onError(Throwable throwable) { + responseObserver.onError(throwable); + } + }, blockFile, context); } else { final ContainerCommandResponseProto resp = dispatcher.dispatch(request, context); responseObserver.onNext(resp); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java new file mode 100644 index 000000000000..b342e644914f --- /dev/null +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ClosedContainerReadResolver.java @@ -0,0 +1,79 @@ +/* + * 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 static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerDataProto.State.CLOSED; + +import java.io.IOException; +import java.nio.channels.WritableByteChannel; +import java.util.concurrent.CompletionException; +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.Message; +import org.apache.ratis.protocol.RaftClientRequest; +import org.apache.ratis.server.DataStreamReadResolver; +import org.apache.ratis.statemachine.StateMachine; + +/** Resolves group-independent reads against immutable local containers. */ +final class ClosedContainerReadResolver implements DataStreamReadResolver { + 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(); + } + + @Override + public StateMachine.DataApi 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.getContainerState() != CLOSED) { + return null; + } + + return new StateMachine.DataApi() { + @Override + public void query(Message ignored, WritableByteChannel stream) { + try { + ContainerStateMachine.streamReadBlock( + dispatcher, requestProto, stream); + } catch (IOException e) { + throw new CompletionException(e); + } + } + }; + } +} diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index 3de4110a01b3..892034320b37 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -28,7 +28,10 @@ import java.io.IOException; 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.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.HashMap; @@ -52,6 +55,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import org.apache.hadoop.hdds.HddsUtils; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -60,6 +64,7 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Container2BCSIDMapProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadBlockResponseProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkRequestProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type; @@ -70,15 +75,18 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; 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; import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration; import org.apache.hadoop.ozone.container.keyvalue.impl.KeyValueStreamDataChannel; import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController; import org.apache.hadoop.util.Time; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.proto.RaftProtos; import org.apache.ratis.proto.RaftProtos.LogEntryProto; import org.apache.ratis.proto.RaftProtos.RaftPeerRole; @@ -139,6 +147,8 @@ */ public class ContainerStateMachine extends BaseStateMachine { static final Logger LOG = LoggerFactory.getLogger(ContainerStateMachine.class); + private static final int RATIS_READ_BLOCK_STREAM_HEADER_BYTES = + Integer.BYTES; private final SimpleStateMachineStorage storage = new SimpleStateMachineStorage(); private final ContainerDispatcher dispatcher; @@ -838,6 +848,10 @@ public CompletableFuture query(Message request) { metrics.incNumQueryStateMachineOps(); final ContainerCommandRequestProto requestProto = message2ContainerCommandRequestProto(request); + if (requestProto.getCmdType() == Type.ReadBlock) { + return CompletableFuture.completedFuture( + queryReadBlock(requestProto)::toByteString); + } return CompletableFuture.completedFuture( dispatchCommand(requestProto, null)::toByteString); } catch (IOException e) { @@ -846,6 +860,222 @@ public CompletableFuture query(Message request) { } } + @Override + public void query(Message request, WritableByteChannel stream) { + try { + metrics.incNumQueryStateMachineOps(); + final ContainerCommandRequestProto requestProto = + message2ContainerCommandRequestProto(request); + if (requestProto.getCmdType() == Type.ReadBlock) { + streamReadBlock(dispatcher, requestProto, stream); + } else { + writeAndClose(stream, dispatchCommand(requestProto, null)); + } + } catch (IOException e) { + metrics.incNumQueryStateMachineFails(); + throw new CompletionException(e); + } + } + + static void streamReadBlock( + ContainerDispatcher dispatcher, + ContainerCommandRequestProto requestProto, + WritableByteChannel stream) throws IOException { + final AtomicReference error = new AtomicReference<>(); + final AtomicBoolean responseSeen = new AtomicBoolean(false); + final DataStreamObserver observer = + new DataStreamObserver() { + @Override + public void onNext(ReadBlockResponse response) { + responseSeen.set(true); + try { + writeReadBlockStreamResponse(stream, response.getResponse(), + response.getData()); + } catch (IOException e) { + error.compareAndSet(null, e); + throw new CompletionException(e); + } catch (RuntimeException e) { + error.compareAndSet(null, e); + throw e; + } + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + } + + @Override + public void onCompleted() { + } + + }; + + try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) { + dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null); + } catch (CompletionException e) { + if (e.getCause() instanceof IOException) { + throw (IOException) e.getCause(); + } + throw new IOException("Failed to stream ReadBlock " + requestProto, + e.getCause() != null ? e.getCause() : e); + } + + if (error.get() != null) { + throw new IOException("Failed to stream ReadBlock " + requestProto, + error.get()); + } + if (!responseSeen.get()) { + throw new IOException("ReadBlock stream returned no responses: " + + requestProto); + } + + stream.close(); + } + + private static void writeAndClose( + WritableByteChannel stream, ContainerCommandResponseProto response) + throws IOException { + try { + writeFully(stream, response.toByteString().asReadOnlyByteBuffer()); + } finally { + stream.close(); + } + } + + private static void writeReadBlockStreamResponse( + WritableByteChannel stream, ContainerCommandResponseProto response, + ByteBuffer data) throws IOException { + writeFully(stream, encodeReadBlockStreamResponse(response, data)); + } + + private static void writeFully(WritableByteChannel stream, ByteBuffer buffer) + throws IOException { + while (buffer.hasRemaining()) { + final int position = buffer.position(); + final int remaining = buffer.remaining(); + final int written = stream.write(buffer); + if (written < 0) { + throw new IOException("EOF reached while writing ReadBlock stream response"); + } + if (written == 0) { + throw new IOException("Zero bytes written to ReadBlock stream response channel"); + } + final int advanced = buffer.position() - position; + if (advanced == 0) { + if (written > remaining) { + throw new IOException("ReadBlock stream response write exceeded remaining bytes: written=" + + written + ", remaining=" + remaining); + } + buffer.position(position + written); + } else if (advanced != written) { + throw new IOException("Inconsistent ReadBlock stream response write: written=" + + written + ", advanced=" + advanced); + } + } + } + + private static ByteBuffer encodeReadBlockStreamResponse( + ContainerCommandResponseProto response, ByteBuffer dataBuffer) { + final ByteBuffer data; + final ContainerCommandResponseProto metadata; + if (response.hasReadBlock()) { + final ReadBlockResponseProto readBlock = response.getReadBlock(); + data = dataBuffer != null ? dataBuffer.asReadOnlyBuffer() + : readBlock.getData().asReadOnlyByteBuffer(); + metadata = response.toBuilder() + .setReadBlock(readBlock.toBuilder().setData(ByteString.EMPTY)) + .build(); + } else { + data = ByteBuffer.allocate(0); + metadata = response; + } + + final ByteBuffer metadataBuffer = + metadata.toByteString().asReadOnlyByteBuffer(); + final ByteBuffer header = + ByteBuffer.allocate(RATIS_READ_BLOCK_STREAM_HEADER_BYTES); + header.putInt(metadataBuffer.remaining()); + header.flip(); + final ByteBuffer frame = ByteBuffer.allocate( + header.remaining() + metadataBuffer.remaining() + data.remaining()); + frame.put(header); + frame.put(metadataBuffer); + frame.put(data); + frame.flip(); + return frame.asReadOnlyBuffer(); + } + + private ContainerCommandResponseProto queryReadBlock( + ContainerCommandRequestProto requestProto) throws IOException { + final List responses = new ArrayList<>(); + final AtomicReference error = new AtomicReference<>(); + final DataStreamObserver observer = + new DataStreamObserver() { + @Override + public void onNext(ReadBlockResponse response) { + responses.add(response.getData() == null ? response.getResponse() + : response.getResponse().toBuilder() + .setReadBlock(response.getResponse().getReadBlock().toBuilder() + .setData(ByteString.copyFrom(response.getData()))) + .build()); + } + + @Override + public void onError(Throwable throwable) { + error.set(throwable); + } + + @Override + public void onCompleted() { + } + }; + + try (RandomAccessFileChannel blockFile = new RandomAccessFileChannel()) { + dispatcher.streamDataReadOnly(requestProto, observer, blockFile, null); + } + + if (error.get() != null) { + throw new IOException("Failed to query ReadBlock " + requestProto, + error.get()); + } + if (responses.isEmpty()) { + throw new IOException("ReadBlock query returned no responses: " + + requestProto); + } + return mergeReadBlockResponses(responses); + } + + private static ContainerCommandResponseProto mergeReadBlockResponses( + List responses) { + final ContainerCommandResponseProto first = responses.get(0); + if (responses.size() == 1 + || first.getResult() != ContainerProtos.Result.SUCCESS + || !first.hasReadBlock()) { + return first; + } + + ByteString data = ByteString.EMPTY; + final ReadBlockResponseProto firstReadBlock = first.getReadBlock(); + final ContainerProtos.ChecksumData.Builder checksum = + firstReadBlock.getChecksumData().toBuilder().clearChecksums(); + for (ContainerCommandResponseProto response : responses) { + if (response.getResult() != ContainerProtos.Result.SUCCESS + || !response.hasReadBlock()) { + return response; + } + final ReadBlockResponseProto readBlock = response.getReadBlock(); + data = data.concat(readBlock.getData()); + checksum.addAllChecksums(readBlock.getChecksumData().getChecksumsList()); + } + + return first.toBuilder() + .setReadBlock(firstReadBlock.toBuilder() + .setData(data) + .setChecksumData(checksum)) + .build(); + } + private ByteString readStateMachineData( ContainerCommandRequestProto requestProto, long term, long index) throws IOException { diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java index 6eadec2d6d36..5550122b8e2f 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java @@ -102,6 +102,7 @@ import org.apache.ratis.protocol.exceptions.StateMachineException; import org.apache.ratis.rpc.RpcType; import org.apache.ratis.rpc.SupportedRpcType; +import org.apache.ratis.server.DataStreamReadResolver; import org.apache.ratis.server.DataStreamServerRpc; import org.apache.ratis.server.RaftServer; import org.apache.ratis.server.RaftServerConfigKeys; @@ -525,6 +526,12 @@ public static XceiverServerRatis newXceiverServerRatis(HddsDatanodeService hddsD CertificateClient caClient, StateContext context) throws IOException { Parameters parameters = createTlsParameters( new SecurityConfig(ozoneConf), caClient); + if (parameters == null) { + parameters = new Parameters(); + } + ClosedContainerReadResolver resolver = new ClosedContainerReadResolver(dispatcher, + containerController, datanodeDetails); + DataStreamReadResolver.set(parameters, resolver); return new XceiverServerRatis(hddsDatanodeService, datanodeDetails, dispatcher, containerController, context, ozoneConf, parameters); 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 1b3f399b46fa..be9e790c6241 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 @@ -148,6 +148,7 @@ import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.interfaces.BlockIterator; import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.interfaces.DBHandle; import org.apache.hadoop.ozone.container.common.interfaces.Handler; import org.apache.hadoop.ozone.container.common.interfaces.ScanResult; @@ -170,9 +171,9 @@ import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures; import org.apache.hadoop.security.token.Token; import org.apache.hadoop.util.Time; +import org.apache.ratis.datastream.DataStreamObserver; import org.apache.ratis.statemachine.StateMachine; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; -import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -2102,7 +2103,7 @@ boolean deleteUnreferencedFile(File file) { public ContainerCommandResponseProto readBlock( ContainerCommandRequestProto request, Container kvContainer, RandomAccessFileChannel blockFile, - StreamObserver streamObserver) { + DataStreamObserver streamObserver) { if (kvContainer.getContainerData().getLayoutVersion() != FILE_PER_BLOCK) { return ContainerUtils.logAndReturnError(LOG, @@ -2142,7 +2143,7 @@ public ContainerCommandResponseProto readBlock( } private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFileChannel blockFile, - Container kvContainer, StreamObserver streamObserver, boolean verifyChecksum) + Container kvContainer, DataStreamObserver streamObserver, boolean verifyChecksum) throws IOException { final ReadBlockRequestProto readBlock = request.getReadBlock(); int responseDataSize = readBlock.getResponseDataSize(); @@ -2179,7 +2180,7 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil final long offsetAlignment = readBlock.getOffset() % bytesPerChecksum; long adjustedOffset = readBlock.getOffset() - offsetAlignment; - final ByteBuffer buffer = ByteBuffer.allocate(responseDataSize); + ByteBuffer buffer = ByteBuffer.allocate(responseDataSize); blockFile.position(adjustedOffset); long totalDataLength = 0; int numResponses = 0; @@ -2207,13 +2208,16 @@ private long readBlockImpl(ContainerCommandRequestProto request, RandomAccessFil Checksum.verifyChecksum(buffer.duplicate(), checksumData, 0); } } - final ContainerCommandResponseProto response = getReadBlockResponse( - request, checksumData, buffer, adjustedOffset); - final int dataLength = response.getReadBlock().getData().size(); + final ContainerCommandResponseProto response; + final int dataLength; + response = getReadBlockResponse( + request, checksumData, ByteString.EMPTY, adjustedOffset); + dataLength = readLength; + streamObserver.onNext(new ReadBlockResponse( + response, buffer.asReadOnlyBuffer())); LOG.debug("server onNext response {}: dataLength={}, numChecksums={}", numResponses, dataLength, response.getReadBlock().getChecksumData().getChecksumsList().size()); - streamObserver.onNext(response); - buffer.clear(); + buffer = ByteBuffer.allocate(responseDataSize); adjustedOffset += readLength; totalDataLength += dataLength; diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java new file mode 100644 index 000000000000..0418067078c3 --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/TestClosedContainerReadResolver.java @@ -0,0 +1,114 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.UUID; +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.ContainerDataProto.State; +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.protocol.RaftGroupId; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; + +class TestClosedContainerReadResolver { + private static final long CONTAINER_ID = 1L; + + private ContainerController containerController; + private Container container; + private ClosedContainerReadResolver resolver; + private String datanodeUuid; + + @BeforeEach + void setUp() { + datanodeUuid = UUID.randomUUID().toString(); + final DatanodeDetails datanode = mock(DatanodeDetails.class); + when(datanode.getUuidString()).thenReturn(datanodeUuid); + containerController = mock(ContainerController.class); + container = mock(Container.class); + when(containerController.getContainer(CONTAINER_ID)).thenReturn(container); + resolver = new ClosedContainerReadResolver( + mock(ContainerDispatcher.class), containerController, datanode); + } + + @Test + void resolvesReadBlockForClosedContainer() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNotNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @ParameterizedTest + @EnumSource(names = {"OPEN", "CLOSING", "QUASI_CLOSED", "UNHEALTHY", + "INVALID", "DELETED", "RECOVERING"}) + void declinesReadBlockForContainerInOtherState(State state) + throws IOException { + when(container.getContainerState()).thenReturn(state); + + assertNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @Test + void declinesReadBlockForMissingContainer() throws IOException { + when(containerController.getContainer(CONTAINER_ID)).thenReturn(null); + + assertNull(resolver.resolve(newRequest(Type.ReadBlock, datanodeUuid))); + } + + @Test + void declinesOtherCommandForClosedContainer() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNull(resolver.resolve(newRequest(Type.GetBlock, datanodeUuid))); + } + + @Test + void declinesRequestForAnotherDatanode() throws IOException { + when(container.getContainerState()).thenReturn(State.CLOSED); + + assertNull(resolver.resolve( + newRequest(Type.ReadBlock, UUID.randomUUID().toString()))); + } + + private static RaftClientRequest newRequest(Type type, String targetUuid) { + final ContainerCommandRequestProto proto = + ContainerCommandRequestProto.newBuilder() + .setCmdType(type) + .setContainerID(CONTAINER_ID) + .setDatanodeUuid(targetUuid) + .build(); + final RaftClientRequest request = mock(RaftClientRequest.class); + when(request.getMessage()).thenReturn( + ContainerCommandRequestMessage.toMessage(proto, null)); + when(request.getRaftGroupId()).thenReturn(RaftGroupId.randomId()); + return request; + } +} 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 f77d6fec2cbc..1117b93fb5d9 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 @@ -102,6 +102,7 @@ import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.common.impl.HddsDispatcher; import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher.ReadBlockResponse; import org.apache.hadoop.ozone.container.common.interfaces.Handler; import org.apache.hadoop.ozone.container.common.report.IncrementalReportSender; import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration; @@ -119,7 +120,7 @@ import org.apache.hadoop.util.Time; import org.apache.ozone.test.GenericTestUtils; import org.apache.ozone.test.GenericTestUtils.LogCapturer; -import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver; +import org.apache.ratis.datastream.DataStreamObserver; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -1091,11 +1092,12 @@ public void testReadBlockMetrics() throws Exception { final AtomicInteger responseCount = new AtomicInteger(0); - StreamObserver streamObserver = - new StreamObserver() { + DataStreamObserver streamObserver = + new DataStreamObserver() { @Override - public void onNext(ContainerCommandResponseProto response) { - assertEquals(ContainerProtos.Result.SUCCESS, response.getResult()); + public void onNext(ReadBlockResponse response) { + assertEquals(ContainerProtos.Result.SUCCESS, + response.getResponse().getResult()); responseCount.incrementAndGet(); } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestRatisDataStreamReadBlock.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestRatisDataStreamReadBlock.java new file mode 100644 index 000000000000..2d38faa7b95d --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestRatisDataStreamReadBlock.java @@ -0,0 +1,223 @@ +/* + * 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.client.rpc.read; + +import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.ONE; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.ByteBuffer; +import java.security.MessageDigest; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.hdds.StringUtils; +import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.scm.OzoneClientConfig; +import org.apache.hadoop.hdds.scm.container.ContainerID; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream; +import org.apache.hadoop.hdds.scm.storage.RatisDataStreamBlockInputStream; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.OzoneClientFactory; +import org.apache.hadoop.ozone.client.io.BlockDataStreamOutputEntry; +import org.apache.hadoop.ozone.client.io.KeyInputStream; +import org.apache.hadoop.ozone.client.io.OzoneDataStreamOutput; +import org.apache.hadoop.ozone.container.TestHelper; +import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi; +import org.apache.hadoop.ozone.om.TestBucket; +import org.apache.ratis.util.SizeInBytes; +import org.junit.jupiter.api.Test; + +/** + * Tests reading a RATIS block through the Ratis data stream transport. + */ +public class TestRatisDataStreamReadBlock { + private static final SizeInBytes KEY_SIZE = SizeInBytes.valueOf("16M"); + private static final SizeInBytes BUFFER_SIZE = SizeInBytes.valueOf("1M"); + + @Test + void readLargeBlockWithRatisDataStream() throws Exception { + try (MiniOzoneCluster cluster = TestStreamRead.newCluster(16 << 10)) { + cluster.waitForClusterToBeReady(); + + final OzoneConfiguration conf = cluster.getConf(); + final OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setStreamReadBlock(false); + clientConfig.setRatisStreamReadBlock(true); + conf.setFromObject(clientConfig); + + try (OzoneClient client = OzoneClientFactory.getRpcClient(conf)) { + final TestBucket testBucket = TestBucket.newBuilder(client).build(); + final String keyName = "ratis-datastream-read-block"; + final String expectedMd5 = createKey(testBucket.delegate(), keyName); + + try (KeyInputStream in = testBucket.getKeyInputStream(keyName)) { + final List streams = + assertRatisStreams(in); + assertEquals(0, streams.get(0).read(ByteBuffer.allocate(0))); + assertEquals(expectedMd5, readMd5(in)); + } + } + } + } + + @Test + void readClosedContainerAfterRatisGroupRemovalWithRatisDataStream() + throws Exception { + try (MiniOzoneCluster cluster = TestStreamRead.newCluster(16 << 10)) { + cluster.waitForClusterToBeReady(); + + final OzoneConfiguration conf = cluster.getConf(); + final OzoneClientConfig clientConfig = + conf.getObject(OzoneClientConfig.class); + clientConfig.setStreamReadBlock(false); + clientConfig.setRatisStreamReadBlock(true); + conf.setFromObject(clientConfig); + + try (OzoneClient client = OzoneClientFactory.getRpcClient(conf)) { + final TestBucket testBucket = TestBucket.newBuilder(client).build(); + final String keyName = "ratis-datastream-read-closed-container"; + final String expectedMd5 = createKeyAndCloseContainersAndRemoveGroups( + cluster, testBucket.delegate(), keyName); + + try (KeyInputStream in = testBucket.getKeyInputStream(keyName)) { + assertRatisStreams(in); + assertEquals(expectedMd5, readMd5(in)); + } + } + } + } + + private static List assertRatisStreams( + KeyInputStream in) { + assertTrue(in.isStreamBlockInputStream()); + final List streams = new ArrayList<>(); + for (BlockExtendedInputStream stream : in.getPartStreams()) { + streams.add(assertInstanceOf(RatisDataStreamBlockInputStream.class, + stream, "Unexpected stream classes: " + in.getPartStreams())); + } + assertTrue(!streams.isEmpty()); + return streams; + } + + private static String createKey(OzoneBucket bucket, String keyName) throws Exception { + try (OutputStream out = bucket.createStreamKey(keyName, KEY_SIZE.getSize(), + RatisReplicationConfig.getInstance(ONE), Collections.emptyMap())) { + return writeKey(out); + } + } + + private static String createKeyAndCloseContainersAndRemoveGroups( + MiniOzoneCluster cluster, OzoneBucket bucket, String keyName) + throws Exception { + final String expectedMd5; + final List containerIds; + try (OzoneDataStreamOutput out = bucket.createStreamKey(keyName, + KEY_SIZE.getSize(), RatisReplicationConfig.getInstance(ONE), + Collections.emptyMap())) { + expectedMd5 = writeKey(out); + containerIds = getContainerIds(out); + } + final List pipelines = getPipelines(cluster, containerIds); + TestHelper.waitForContainerClose(cluster, containerIds.toArray(new Long[0])); + removeRatisGroups(cluster, pipelines); + return expectedMd5; + } + + private static List getPipelines(MiniOzoneCluster cluster, + List containerIds) throws Exception { + final List pipelines = new ArrayList<>(); + for (long containerId : containerIds) { + final Pipeline pipeline = cluster.getStorageContainerManager() + .getPipelineManager().getPipeline(cluster.getStorageContainerManager() + .getContainerManager().getContainer( + ContainerID.valueOf(containerId)).getPipelineID()); + if (!pipelines.contains(pipeline)) { + pipelines.add(pipeline); + } + } + return pipelines; + } + + private static void removeRatisGroups(MiniOzoneCluster cluster, + List pipelines) throws Exception { + for (Pipeline pipeline : pipelines) { + for (DatanodeDetails dn : pipeline.getNodes()) { + final XceiverServerSpi server = cluster.getHddsDatanodes() + .get(cluster.getHddsDatanodeIndex(dn)).getDatanodeStateMachine() + .getContainer().getWriteChannel(); + if (server.isExist(pipeline.getId().getProtobuf())) { + server.removeGroup(pipeline.getId().getProtobuf()); + } + assertFalse(server.isExist(pipeline.getId().getProtobuf())); + } + } + } + + private static List getContainerIds(OzoneDataStreamOutput out) { + final List containerIds = new ArrayList<>(); + for (BlockDataStreamOutputEntry entry : + out.getKeyDataStreamOutput().getStreamEntries()) { + final long containerId = entry.getBlockID().getContainerID(); + if (!containerIds.contains(containerId)) { + containerIds.add(containerId); + } + } + assertTrue(!containerIds.isEmpty()); + return containerIds; + } + + private static String writeKey(OutputStream out) throws Exception { + final MessageDigest md5 = MessageDigest.getInstance("MD5"); + final byte[] buffer = new byte[BUFFER_SIZE.getSizeInt()]; + for (int i = 0; i < buffer.length; i++) { + buffer[i] = (byte) (i & 0xff); + } + + for (long pos = 0; pos < KEY_SIZE.getSize();) { + final int writeSize = + Math.toIntExact(Math.min(buffer.length, KEY_SIZE.getSize() - pos)); + out.write(buffer, 0, writeSize); + md5.update(buffer, 0, writeSize); + pos += writeSize; + } + return StringUtils.bytes2Hex(md5.digest()); + } + + private static String readMd5(InputStream in) throws Exception { + final MessageDigest md5 = MessageDigest.getInstance("MD5"); + final byte[] buffer = new byte[BUFFER_SIZE.getSizeInt()]; + for (;;) { + final int read = in.read(buffer); + if (read < 0) { + break; + } + md5.update(buffer, 0, read); + } + return StringUtils.bytes2Hex(md5.digest()); + } +} diff --git a/pom.xml b/pom.xml index b63164afa94a..84a31f492799 100644 --- a/pom.xml +++ b/pom.xml @@ -188,7 +188,7 @@ 4.1.130.Final 3.25.8 1.0.11 - 3.2.1 + 3.3.0-SNAPSHOT 1.7 0.10.2 1.2.26