diff --git a/.github/workflows/post-commit.yml b/.github/workflows/post-commit.yml index 1aaf27c6da23..43b52423c1d0 100644 --- a/.github/workflows/post-commit.yml +++ b/.github/workflows/post-commit.yml @@ -19,8 +19,7 @@ on: schedule: - cron: 30 0,12 * * * jobs: - build: - name: compile + compile: runs-on: ubuntu-18.04 strategy: matrix: @@ -49,7 +48,14 @@ jobs: with: java-version: ${{ matrix.java }} - name: Run a full build - run: hadoop-ozone/dev-support/checks/build.sh + run: hadoop-ozone/dev-support/checks/build.sh -Pcoverage -Pdist + - name: Store binaries for tests + uses: actions/upload-artifact@v2 + if: matrix.java == '8' + with: + name: ozone-bin + path: hadoop-ozone/dist/target/hadoop-ozone*.tar.gz + retention-days: 1 bats: runs-on: ubuntu-18.04 steps: @@ -167,6 +173,7 @@ jobs: path: target/findbugs continue-on-error: true acceptance: + needs: compile runs-on: ubuntu-18.04 strategy: matrix: @@ -178,37 +185,20 @@ jobs: steps: - name: Checkout project uses: actions/checkout@v2 - - name: Cache for maven dependencies - uses: actions/cache@v2 - with: - path: ~/.m2/repository - key: maven-repo-${{ hashFiles('**/pom.xml') }} - restore-keys: maven-repo- - - name: Cache for npm dependencies - uses: actions/cache@v2 with: - path: | - ~/.pnpm-store - **/node_modules - key: ${{ runner.os }}-pnpm-${{ hashFiles('**/pnpm-lock.yaml') }} - restore-keys: | - ${{ runner.os }}-pnpm- - - name: Checkout to /mnt/ozone + path: ozone + - name: Move Ozone to /mnt run: | sudo chmod 777 /mnt - git clone 'https://github.com/${{ github.repository }}.git' /mnt/ozone - cd /mnt/ozone - if [[ '${{ github.event_name }}' == 'pull_request' ]]; then - git fetch --verbose origin '${{ github.ref }}' - else - git fetch --verbose origin '${{ github.sha }}' - fi - git checkout FETCH_HEAD - git reset --hard - - name: Run a full build + mv ozone /mnt/ + - name: Download compiled Ozone binaries + uses: actions/download-artifact@v2 + with: + name: ozone-bin + - name: Untar binaries run: | - cd /mnt/ozone - hadoop-ozone/dev-support/checks/build.sh -Pcoverage + mkdir -p /mnt/ozone/hadoop-ozone/dist/target + tar xzvf hadoop-ozone*.tar.gz -C /mnt/ozone/hadoop-ozone/dist/target - name: Install robotframework run: sudo pip install robotframework - name: Execute tests @@ -284,7 +274,7 @@ jobs: run: ./hadoop-ozone/dev-support/checks/coverage.sh - name: Upload coverage to Sonar uses: ./.github/buildenv - if: github.repository == 'apache/hadoop-ozone' && github.event_name != 'pull_request' + if: github.repository == 'apache/ozone' && github.event_name != 'pull_request' with: args: ./hadoop-ozone/dev-support/checks/sonar.sh env: @@ -292,7 +282,7 @@ jobs: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - name: Upload coverage to Codecov uses: codecov/codecov-action@v1 - if: github.repository == 'apache/hadoop-ozone' && github.event_name != 'pull_request' + if: github.repository == 'apache/ozone' && github.event_name != 'pull_request' with: file: ./target/coverage/all.xml name: codecov-umbrella @@ -304,37 +294,25 @@ jobs: path: target/coverage continue-on-error: true kubernetes: + needs: compile runs-on: ubuntu-18.04 steps: - name: Checkout project uses: actions/checkout@v2 - - name: Cache for maven dependencies - uses: actions/cache@v2 - with: - path: ~/.m2/repository - key: maven-repo-${{ hashFiles('**/pom.xml') }} - restore-keys: maven-repo- - - name: Cache for npm dependencies - uses: actions/cache@v2 with: - path: | - ~/.pnpm-store - **/node_modules - key: ${{ runner.os }}-pnpm-${{ hashFiles('**/pnpm-lock.yaml') }} - restore-keys: | - ${{ runner.os }}-pnpm- - - name: Checkout to /mnt/ozone + path: ozone + - name: Move Ozone to /mnt run: | sudo chmod 777 /mnt - git clone 'https://github.com/${{ github.repository }}.git' /mnt/ozone - cd /mnt/ozone - if [[ '${{ github.event_name }}' == 'pull_request' ]]; then - git fetch --verbose origin '${{ github.ref }}' - else - git fetch --verbose origin '${{ github.sha }}' - fi - git checkout FETCH_HEAD - git reset --hard + mv ozone /mnt/ + - name: Download compiled Ozone binaries + uses: actions/download-artifact@v2 + with: + name: ozone-bin + - name: Untar binaries + run: | + mkdir -p /mnt/ozone/hadoop-ozone/dist/target + tar xzvf hadoop-ozone*.tar.gz -C /mnt/ozone/hadoop-ozone/dist/target - name: Install robotframework run: sudo pip install robotframework - name: Install k3s @@ -350,10 +328,6 @@ jobs: wget https://github.com/elek/flekszible/releases/download/v1.8.1/flekszible_1.8.1_Linux_x86_64.tar.gz -O - | tar -zx chmod +x flekszible sudo mv flekszible /usr/bin/flekszible - - name: Run a full build - run: | - cd /mnt/ozone - hadoop-ozone/dev-support/checks/build.sh -Pcoverage - name: Execute tests run: | cd /mnt/ozone/hadoop-ozone/dist/target/ozone-* && sudo mkdir .aws && sudo chmod 777 .aws && sudo chown 1000 .aws diff --git a/HISTORY.md b/HISTORY.md index 233471cac8f3..5069238d4eea 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -30,12 +30,12 @@ As a summary: * Ozone: provides Object Store semantics with the help of HDDS * CBlock: provides mountable volumes with the help of the HDDS layer (based on iScsi protocol) -In the beginning of the year 2017 a new podling project was started inside [Apache Incubator](http://incubator.apache.org/): [Apache Ratis](https://ratis.apache.org/). Ratis is an embeddable RAFT protcol implementation it is which became the corner stone of consensus inside both Ozone and HDDS projects. (Started to [be used](https://issues.apache.org/jira/browse/HDFS-11519) by Ozone in March of 2017) +In the beginning of the year 2017 a new podling project was started inside [Apache Incubator](http://incubator.apache.org/): [Apache Ratis](https://ratis.apache.org/). Ratis is an embeddable RAFT protocol implementation it is which became the corner stone of consensus inside both Ozone and HDDS projects. (Started to [be used](https://issues.apache.org/jira/browse/HDFS-11519) by Ozone in March of 2017) In the October of 2017 a [discussion](https://lists.apache.org/thread.html/3b5b65ce428f88299e6cb4c5d745ec65917490be9e417d361cc08d7e@%3Chdfs-dev.hadoop.apache.org%3E) has been started on hdfs-dev mailing list to merge the existing functionality to the Apache Hadoop trunk. After a long debate Owen O'Malley [suggested a consensus](https://lists.apache.org/thread.html/c85e5263dcc0ca1d13cbbe3bcfb53236784a39111b8c353f60582eb4@%3Chdfs-dev.hadoop.apache.org%3E) to merge it to the trunk but use separated release cycle: > * HDSL become a subproject of Hadoop. - > * HDSL will release separately from Hadoop. Hadoop releases will notcontain HDSL and vice versa. + > * HDSL will release separately from Hadoop. Hadoop releases will not contain HDSL and vice versa. > * HDSL will get its own jira instance so that the release tags stay separate. > * On trunk (as opposed to release branches) HDSL will be a separate module in Hadoop's source tree. This will enable the HDSL to work on their trunk and the Hadoop trunk without making releases for every change. > * Hadoop's trunk will only build HDSL if a non-default profile is enabled. When Hadoop creates a release branch, the RM will delete the HDSL module from the branch. diff --git a/README.md b/README.md index b4a40bfcd863..2df0f8be0fb0 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ Ozone is a top level project under the [Apache Software Foundation](https://apac * Chat: You can find the #ozone channel on the official ASF slack. Invite link is [here](http://s.apache.org/slack-invite). * There are Open [Weekly calls](https://cwiki.apache.org/confluence/display/HADOOP/Ozone+Community+Calls) where you can ask anything about Ozone. * Past meeting notes are also available from the wiki. + * Reporting security issues: Please consult with [SECURITY.md](./SECURITY.md) about reporting security vulnerabilities and issues. ## Download diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 000000000000..8d8a42b48d36 --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,23 @@ +# Security Policy + +## Supported Versions + +The first stable release of Apache Ozone is 1.0, the previous alpha and beta releases are not supported by the community. + +| Version | Supported | +| ------------- | ------------------ | +| 0.3.0 (alpha) | :x: | +| 0.4.0 (alpha) | :x: | +| 0.4.1 (alpha) | :x: | +| 0.5.0 (beta) | :x: | +| 1.0 | :white_check_mark: | + +## Reporting a Vulnerability + +To report any found security issues or vulnerabilities, please send a mail to security@ozone.apache.org, so that they may be investigated and fixed before the vulnerabilities is published. + +This email address is a private mailing list for discussion of potential security vulnerabilities issues. + +This mailing list is **NOT** for end-user questions and discussion on security. Please use the dev@ozone.apache.org list for such issues. + +In order to post to the list, it is **NOT** necessary to first subscribe to it. diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java index 3578bd6ea1e2..8cb97c1d2070 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java @@ -26,6 +26,7 @@ import java.util.List; import java.util.function.Function; +import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.Seekable; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; @@ -33,7 +34,6 @@ import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.GetBlockResponseProto; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.XceiverClientFactory; -import org.apache.hadoop.hdds.scm.XceiverClientManager; import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; @@ -50,7 +50,8 @@ * This class encapsulates all state management for iterating * through the sequence of chunks through {@link ChunkInputStream}. */ -public class BlockInputStream extends InputStream implements Seekable { +public class BlockInputStream extends InputStream + implements Seekable, CanUnbuffer { private static final Logger LOG = LoggerFactory.getLogger(BlockInputStream.class); @@ -95,25 +96,25 @@ public class BlockInputStream extends InputStream implements Seekable { // can be reset if a new position is seeked. private int chunkIndexOfPrevPosition; - private Function refreshPipelineFunction; + private final Function refreshPipelineFunction; public BlockInputStream(BlockID blockId, long blockLen, Pipeline pipeline, Token token, boolean verifyChecksum, - XceiverClientFactory xceiverClientFctry, + XceiverClientFactory xceiverClientFactory, Function refreshPipelineFunction) { this.blockID = blockId; this.length = blockLen; this.pipeline = pipeline; this.token = token; this.verifyChecksum = verifyChecksum; - this.xceiverClientFactory = xceiverClientFctry; + this.xceiverClientFactory = xceiverClientFactory; this.refreshPipelineFunction = refreshPipelineFunction; } public BlockInputStream(BlockID blockId, long blockLen, Pipeline pipeline, Token token, boolean verifyChecksum, - XceiverClientManager xceiverClientFactory + XceiverClientFactory xceiverClientFactory ) { this(blockId, blockLen, pipeline, token, verifyChecksum, xceiverClientFactory, null); @@ -129,22 +130,12 @@ public synchronized void initialize() throws IOException { return; } - List chunks = null; + List chunks; try { chunks = getChunkInfos(); } catch (ContainerNotFoundException ioEx) { - LOG.error("Unable to read block information from pipeline."); - if (refreshPipelineFunction != null) { - LOG.debug("Re-fetching pipeline for block {}", blockID); - Pipeline newPipeline = refreshPipelineFunction.apply(blockID); - if (newPipeline == null || newPipeline.equals(pipeline)) { - throw ioEx; - } else { - LOG.debug("New pipeline got for block {}", blockID); - this.pipeline = newPipeline; - chunks = getChunkInfos(); - } - } + refreshPipeline(ioEx); + chunks = getChunkInfos(); } if (chunks != null && !chunks.isEmpty()) { @@ -171,6 +162,24 @@ public synchronized void initialize() throws IOException { } } + private void refreshPipeline(IOException cause) throws IOException { + LOG.info("Unable to read information for block {} from pipeline {}: {}", + blockID, pipeline.getId(), cause.getMessage()); + if (refreshPipelineFunction != null) { + LOG.debug("Re-fetching pipeline for block {}", blockID); + Pipeline newPipeline = refreshPipelineFunction.apply(blockID); + if (newPipeline == null || newPipeline.equals(pipeline)) { + LOG.warn("No new pipeline for block {}", blockID); + throw cause; + } else { + LOG.debug("New pipeline got for block {}", blockID); + this.pipeline = newPipeline; + } + } else { + throw cause; + } + } + /** * Send RPC call to get the block info from the container. * @return List of chunks in this block. @@ -182,7 +191,7 @@ protected List getChunkInfos() throws IOException { pipeline = Pipeline.newBuilder(pipeline) .setType(HddsProtos.ReplicationType.STAND_ALONE).build(); } - xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline); + acquireClient(); boolean success = false; List chunks; try { @@ -207,17 +216,25 @@ protected List getChunkInfos() throws IOException { return chunks; } + protected void acquireClient() throws IOException { + xceiverClient = xceiverClientFactory.acquireClientForReadData(pipeline); + } + /** * Append another ChunkInputStream to the end of the list. Note that the * ChunkInputStream is only created here. The chunk will be read from the * Datanode only when a read operation is performed on for that chunk. */ protected synchronized void addStream(ChunkInfo chunkInfo) { - chunkStreams.add(new ChunkInputStream(chunkInfo, blockID, - xceiverClient, verifyChecksum, token)); + chunkStreams.add(createChunkInputStream(chunkInfo)); } - public synchronized long getRemaining() throws IOException { + protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) { + return new ChunkInputStream(chunkInfo, blockID, + xceiverClientFactory, () -> pipeline, verifyChecksum, token); + } + + public synchronized long getRemaining() { return length - getPos(); } @@ -266,7 +283,13 @@ public synchronized int read(byte[] b, int off, int len) throws IOException { // Get the current chunkStream and read data from it ChunkInputStream current = chunkStreams.get(chunkIndex); int numBytesToRead = Math.min(len, (int)current.getRemaining()); - int numBytesRead = current.read(b, off, numBytesToRead); + int numBytesRead; + try { + numBytesRead = current.read(b, off, numBytesToRead); + } catch (IOException e) { + handleReadError(e); + continue; + } if (numBytesRead != numBytesToRead) { // This implies that there is either data loss or corruption in the @@ -356,7 +379,7 @@ public synchronized void seek(long pos) throws IOException { } @Override - public synchronized long getPos() throws IOException { + public synchronized long getPos() { if (length == 0) { return 0; } @@ -376,9 +399,13 @@ public boolean seekToNewSource(long targetPos) throws IOException { @Override public synchronized void close() { + releaseClient(); + xceiverClientFactory = null; + } + + private void releaseClient() { if (xceiverClientFactory != null && xceiverClient != null) { xceiverClientFactory.releaseClient(xceiverClient, false); - xceiverClientFactory = null; xceiverClient = null; } } @@ -393,7 +420,7 @@ public synchronized void resetPosition() { * @throws IOException if stream is closed */ protected synchronized void checkOpen() throws IOException { - if (xceiverClient == null) { + if (xceiverClientFactory == null) { throw new IOException("BlockInputStream has been closed."); } } @@ -416,8 +443,32 @@ synchronized long getBlockPosition() { return blockPosition; } - @VisibleForTesting - synchronized List getChunkStreams() { - return chunkStreams; + @Override + public void unbuffer() { + storePosition(); + releaseClient(); + + final List inputStreams = this.chunkStreams; + if (inputStreams != null) { + for (ChunkInputStream is : inputStreams) { + is.unbuffer(); + } + } + } + + private synchronized void storePosition() { + blockPosition = getPos(); + } + + private void handleReadError(IOException cause) throws IOException { + releaseClient(); + final List inputStreams = this.chunkStreams; + if (inputStreams != null) { + for (ChunkInputStream is : inputStreams) { + is.releaseClient(); + } + } + + refreshPipeline(cause); } } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java index cfb3a21f62be..9c034538c572 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java @@ -20,14 +20,17 @@ import com.google.common.annotations.VisibleForTesting; import com.google.common.base.Preconditions; +import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.Seekable; import org.apache.hadoop.hdds.client.BlockID; 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.ChunkInfo; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; import org.apache.hadoop.hdds.scm.XceiverClientSpi; import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.ozone.common.Checksum; import org.apache.hadoop.ozone.common.ChecksumData; import org.apache.hadoop.ozone.common.OzoneChecksumException; @@ -40,18 +43,22 @@ import java.io.InputStream; import java.nio.ByteBuffer; import java.util.List; +import java.util.function.Supplier; /** * An {@link InputStream} called from BlockInputStream to read a chunk from the * container. Each chunk may contain multiple underlying {@link ByteBuffer} * instances. */ -public class ChunkInputStream extends InputStream implements Seekable { +public class ChunkInputStream extends InputStream + implements Seekable, CanUnbuffer { private ChunkInfo chunkInfo; private final long length; private final BlockID blockID; + private final XceiverClientFactory xceiverClientFactory; private XceiverClientSpi xceiverClient; + private final Supplier pipelineSupplier; private boolean verifyChecksum; private boolean allocated = false; // Buffer to store the chunk data read from the DN container @@ -69,9 +76,8 @@ public class ChunkInputStream extends InputStream implements Seekable { // Position of the ChunkInputStream is maintained by this variable (if a // seek is performed. This position is w.r.t to the chunk only and not the - // block or key. This variable is set only if either the buffers are not - // yet allocated or the if the allocated buffers do not cover the seeked - // position. Once the chunk is read, this variable is reset. + // block or key. This variable is also set before attempting a read to enable + // retry. Once the chunk is read, this variable is reset. private long chunkPosition = -1; private final Token token; @@ -79,17 +85,19 @@ public class ChunkInputStream extends InputStream implements Seekable { private static final int EOF = -1; ChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, - XceiverClientSpi xceiverClient, boolean verifyChecksum, - Token token) { + XceiverClientFactory xceiverClientFactory, + Supplier pipelineSupplier, + boolean verifyChecksum, Token token) { this.chunkInfo = chunkInfo; this.length = chunkInfo.getLen(); this.blockID = blockId; - this.xceiverClient = xceiverClient; + this.xceiverClientFactory = xceiverClientFactory; + this.pipelineSupplier = pipelineSupplier; this.verifyChecksum = verifyChecksum; this.token = token; } - public synchronized long getRemaining() throws IOException { + public synchronized long getRemaining() { return length - getPos(); } @@ -98,7 +106,7 @@ public synchronized long getRemaining() throws IOException { */ @Override public synchronized int read() throws IOException { - checkOpen(); + acquireClient(); int available = prepareRead(1); int dataout = EOF; @@ -143,7 +151,7 @@ public synchronized int read(byte[] b, int off, int len) throws IOException { if (len == 0) { return 0; } - checkOpen(); + acquireClient(); int total = 0; while (len > 0) { int available = prepareRead(len); @@ -196,7 +204,7 @@ public synchronized void seek(long pos) throws IOException { } @Override - public synchronized long getPos() throws IOException { + public synchronized long getPos() { if (chunkPosition >= 0) { return chunkPosition; } @@ -219,19 +227,23 @@ public boolean seekToNewSource(long targetPos) throws IOException { @Override public synchronized void close() { - if (xceiverClient != null) { + releaseClient(); + } + + protected synchronized void releaseClient() { + if (xceiverClientFactory != null && xceiverClient != null) { + xceiverClientFactory.releaseClient(xceiverClient, false); xceiverClient = null; } } /** - * Checks if the stream is open. If not, throw an exception. - * - * @throws IOException if stream is closed + * Acquire new client if previous one was released. */ - protected synchronized void checkOpen() throws IOException { - if (xceiverClient == null) { - throw new IOException("BlockInputStream has been closed."); + protected synchronized void acquireClient() throws IOException { + if (xceiverClientFactory != null && xceiverClient == null) { + xceiverClient = xceiverClientFactory.acquireClientForReadData( + pipelineSupplier.get()); } } @@ -292,6 +304,11 @@ private synchronized void readChunkFromContainer(int len) throws IOException { startByteIndex = bufferOffset + bufferLength; } + // bufferOffset and bufferLength are updated below, but if read fails + // and is retried, we need the previous position. Position is reset after + // successful read in adjustBufferPosition() + storePosition(); + if (verifyChecksum) { // Update the bufferOffset and bufferLength as per the checksum // boundary requirement. @@ -437,7 +454,8 @@ private void adjustBufferPosition(long bufferPosition) { /** * Check if the buffers have been allocated data and false otherwise. */ - private boolean buffersAllocated() { + @VisibleForTesting + protected boolean buffersAllocated() { return buffers != null && !buffers.isEmpty(); } @@ -538,6 +556,10 @@ void resetPosition() { this.chunkPosition = -1; } + private void storePosition() { + chunkPosition = getPos(); + } + String getChunkName() { return chunkInfo.getChunkName(); } @@ -550,4 +572,11 @@ protected long getLength() { protected long getChunkPosition() { return chunkPosition; } + + @Override + public synchronized void unbuffer() { + storePosition(); + releaseBuffers(); + releaseClient(); + } } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStream.java index 5db722acaa41..1c7968b13446 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStream.java @@ -24,7 +24,7 @@ import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; -import org.apache.hadoop.hdds.scm.XceiverClientManager; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.security.token.OzoneBlockTokenIdentifier; import org.apache.hadoop.security.token.Token; @@ -34,9 +34,9 @@ */ class DummyBlockInputStream extends BlockInputStream { - private List chunks; + private final List chunks; - private Map chunkDataMap; + private final Map chunkDataMap; @SuppressWarnings("parameternumber") DummyBlockInputStream( @@ -45,23 +45,7 @@ class DummyBlockInputStream extends BlockInputStream { Pipeline pipeline, Token token, boolean verifyChecksum, - XceiverClientManager xceiverClientManager, - List chunkList, - Map chunkMap) { - super(blockId, blockLen, pipeline, token, verifyChecksum, - xceiverClientManager); - this.chunks = chunkList; - this.chunkDataMap = chunkMap; - } - - @SuppressWarnings("parameternumber") - DummyBlockInputStream( - BlockID blockId, - long blockLen, - Pipeline pipeline, - Token token, - boolean verifyChecksum, - XceiverClientManager xceiverClientManager, + XceiverClientFactory xceiverClientManager, Function refreshFunction, List chunkList, Map chunks) { @@ -78,11 +62,10 @@ protected List getChunkInfos() throws IOException { } @Override - protected void addStream(ChunkInfo chunkInfo) { - TestChunkInputStream testChunkInputStream = new TestChunkInputStream(); - getChunkStreams().add(new DummyChunkInputStream(testChunkInputStream, + protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) { + return new DummyChunkInputStream( chunkInfo, null, null, false, - chunkDataMap.get(chunkInfo.getChunkName()).clone())); + chunkDataMap.get(chunkInfo.getChunkName()).clone(), null); } @Override diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStreamWithRetry.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStreamWithRetry.java index 1686ed45b859..51ba2c6eddce 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStreamWithRetry.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyBlockInputStreamWithRetry.java @@ -26,7 +26,7 @@ import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; -import org.apache.hadoop.hdds.scm.XceiverClientManager; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; import org.apache.hadoop.hdds.scm.container.ContainerNotFoundException; import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.pipeline.PipelineID; @@ -49,7 +49,7 @@ final class DummyBlockInputStreamWithRetry Pipeline pipeline, Token token, boolean verifyChecksum, - XceiverClientManager xceiverClientManager, + XceiverClientFactory xceiverClientManager, List chunkList, Map chunkMap, AtomicBoolean isRerfreshed) { diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java index e654d119f75d..15f7edab6fcf 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/DummyChunkInputStream.java @@ -22,8 +22,9 @@ import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; -import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; /** @@ -31,18 +32,18 @@ */ public class DummyChunkInputStream extends ChunkInputStream { - private byte[] chunkData; + private final byte[] chunkData; // Stores the read chunk data in each readChunk call - private List readByteBuffers = new ArrayList<>(); + private final List readByteBuffers = new ArrayList<>(); - public DummyChunkInputStream(TestChunkInputStream testChunkInputStream, - ChunkInfo chunkInfo, + public DummyChunkInputStream(ChunkInfo chunkInfo, BlockID blockId, - XceiverClientSpi xceiverClient, + XceiverClientFactory xceiverClientFactory, boolean verifyChecksum, - byte[] data) { - super(chunkInfo, blockId, xceiverClient, verifyChecksum, null); + byte[] data, Pipeline pipeline) { + super(chunkInfo, blockId, xceiverClientFactory, () -> pipeline, + verifyChecksum, null); this.chunkData = data; } @@ -56,10 +57,15 @@ protected ByteString readChunk(ChunkInfo readChunkInfo) { } @Override - protected void checkOpen() { + protected void acquireClient() { // No action needed } + @Override + protected void releaseClient() { + // no-op + } + public List getReadByteBuffers() { return readByteBuffers; } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java index 3f5e12a8e16b..e1ede9771970 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockInputStream.java @@ -23,25 +23,42 @@ import org.apache.hadoop.hdds.client.ContainerBlockID; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException; +import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.ozone.common.Checksum; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; import java.io.EOFException; +import java.io.IOException; import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Random; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Function; +import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND; import static org.apache.hadoop.hdds.scm.storage.TestChunkInputStream.generateRandomData; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * Tests for {@link BlockInputStream}'s functionality. */ +@RunWith(MockitoJUnitRunner.class) public class TestBlockInputStream { private static final int CHUNK_SIZE = 100; @@ -52,7 +69,9 @@ public class TestBlockInputStream { private int blockSize; private List chunks; private Map chunkDataMap; - private AtomicBoolean isRefreshed = new AtomicBoolean(); + + @Mock + private Function refreshPipeline; @Before public void setup() throws Exception { @@ -61,7 +80,7 @@ public void setup() throws Exception { createChunkList(5); blockStream = new DummyBlockInputStream(blockID, blockSize, null, null, - false, null, chunks, chunkDataMap); + false, null, refreshPipeline, chunks, chunkDataMap); } /** @@ -199,9 +218,11 @@ public void testSeekAndRead() throws Exception { @Test public void testRefreshPipelineFunction() throws Exception { BlockID blockID = new BlockID(new ContainerBlockID(1, 1)); + AtomicBoolean isRefreshed = new AtomicBoolean(); createChunkList(5); BlockInputStream blockInputStreamWithRetry = - new DummyBlockInputStreamWithRetry(blockID, blockSize, null, null, + new DummyBlockInputStreamWithRetry(blockID, blockSize, + MockPipeline.createSingleNodePipeline(), null, false, null, chunks, chunkDataMap, isRefreshed); Assert.assertFalse(isRefreshed.get()); @@ -210,4 +231,89 @@ public void testRefreshPipelineFunction() throws Exception { blockInputStreamWithRetry.read(b, 0, 200); Assert.assertTrue(isRefreshed.get()); } + + @Test + public void testRefreshOnReadFailure() throws Exception { + // GIVEN + BlockID blockID = new BlockID(new ContainerBlockID(1, 1)); + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + Pipeline newPipeline = MockPipeline.createSingleNodePipeline(); + + final int len = 200; + final ChunkInputStream stream = mock(ChunkInputStream.class); + when(stream.read(any(), anyInt(), anyInt())) + .thenThrow(new StorageContainerException("test", CONTAINER_NOT_FOUND)) + .thenReturn(len); + when(stream.getRemaining()) + .thenReturn((long) len); + + when(refreshPipeline.apply(blockID)) + .thenReturn(newPipeline); + + BlockInputStream subject = new DummyBlockInputStream(blockID, blockSize, + pipeline, null, false, null, refreshPipeline, chunks, null) { + @Override + protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) { + return stream; + } + }; + subject.initialize(); + + // WHEN + byte[] b = new byte[len]; + int bytesRead = subject.read(b, 0, len); + + // THEN + Assert.assertEquals(len, bytesRead); + verify(refreshPipeline).apply(blockID); + } + + @Test + public void testRefreshOnReadFailureAfterUnbuffer() throws Exception { + // GIVEN + BlockID blockID = new BlockID(new ContainerBlockID(1, 1)); + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + Pipeline newPipeline = MockPipeline.createSingleNodePipeline(); + XceiverClientFactory clientFactory = mock(XceiverClientFactory.class); + XceiverClientSpi client = mock(XceiverClientSpi.class); + when(clientFactory.acquireClientForReadData(pipeline)) + .thenReturn(client); + + final int len = 200; + final ChunkInputStream stream = mock(ChunkInputStream.class); + when(stream.read(any(), anyInt(), anyInt())) + .thenThrow(new StorageContainerException("test", CONTAINER_NOT_FOUND)) + .thenReturn(len); + when(stream.getRemaining()) + .thenReturn((long) len); + + when(refreshPipeline.apply(blockID)) + .thenReturn(newPipeline); + + BlockInputStream subject = new BlockInputStream(blockID, blockSize, + pipeline, null, false, clientFactory, refreshPipeline) { + @Override + protected List getChunkInfos() throws IOException { + acquireClient(); + return chunks; + } + + @Override + protected ChunkInputStream createChunkInputStream(ChunkInfo chunkInfo) { + return stream; + } + }; + subject.initialize(); + subject.unbuffer(); + + // WHEN + byte[] b = new byte[len]; + int bytesRead = subject.read(b, 0, len); + + // THEN + Assert.assertEquals(len, bytesRead); + verify(refreshPipeline).apply(blockID); + verify(clientFactory).acquireClientForReadData(pipeline); + verify(clientFactory).releaseClient(client, false); + } } diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java index eea8e1f14ef4..cb110b182d85 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestChunkInputStream.java @@ -20,16 +20,26 @@ import java.io.EOFException; import java.util.Random; +import java.util.concurrent.atomic.AtomicReference; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChecksumType; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo; +import org.apache.hadoop.hdds.scm.XceiverClientFactory; +import org.apache.hadoop.hdds.scm.XceiverClientSpi; +import org.apache.hadoop.hdds.scm.pipeline.MockPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.ozone.common.Checksum; import org.apache.hadoop.test.GenericTestUtils; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + /** * Tests for {@link ChunkInputStream}'s functionality. */ @@ -59,8 +69,8 @@ public void setup() throws Exception { chunkData, 0, CHUNK_SIZE).getProtoBufMessage()) .build(); - chunkStream = - new DummyChunkInputStream(this, chunkInfo, null, null, true, chunkData); + chunkStream = new DummyChunkInputStream(chunkInfo, null, null, true, + chunkData, null); } static byte[] generateRandomData(int length) { @@ -174,4 +184,50 @@ public void testSeekAndRead() throws Exception { chunkStream.read(b2, 0, 20); matchWithInputData(b2, 70, 20); } -} \ No newline at end of file + + @Test + public void testUnbuffer() throws Exception { + byte[] b1 = new byte[20]; + chunkStream.read(b1, 0, 20); + matchWithInputData(b1, 0, 20); + + chunkStream.unbuffer(); + + Assert.assertFalse(chunkStream.buffersAllocated()); + + // Next read should start from the position of the last read + 1 i.e. 20 + byte[] b2 = new byte[20]; + chunkStream.read(b2, 0, 20); + matchWithInputData(b2, 20, 20); + } + + @Test + public void connectsToNewPipeline() throws Exception { + // GIVEN + Pipeline pipeline = MockPipeline.createSingleNodePipeline(); + Pipeline newPipeline = MockPipeline.createSingleNodePipeline(); + XceiverClientFactory clientFactory = mock(XceiverClientFactory.class); + XceiverClientSpi client = mock(XceiverClientSpi.class); + when(clientFactory.acquireClientForReadData(pipeline)) + .thenReturn(client); + + AtomicReference pipelineRef = new AtomicReference<>(pipeline); + + ChunkInputStream subject = new ChunkInputStream(chunkInfo, null, + clientFactory, pipelineRef::get, false, null) { + @Override + protected ByteString readChunk(ChunkInfo readChunkInfo) { + return ByteString.copyFrom(chunkData); + } + }; + + // WHEN + subject.unbuffer(); + pipelineRef.set(newPipeline); + int b = subject.read(); + + // THEN + Assert.assertNotEquals(-1, b); + verify(clientFactory).acquireClientForReadData(newPipeline); + } +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneQuota.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneQuota.java index b778e0397cbc..56a87e83d686 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneQuota.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/OzoneQuota.java @@ -43,13 +43,25 @@ public final class OzoneQuota { public static final String OZONE_QUOTA_TB = "TB"; /** Quota Units.*/ - public enum Units {UNDEFINED, BYTES, KB, MB, GB, TB} + public enum Units {BYTES, KB, MB, GB, TB} // Quota to decide how many buckets can be created. - private long quotaInCounts; + private long quotaInNamespace; // Quota to decide how many storage space will be used in bytes. private long quotaInBytes; private RawQuotaInBytes rawQuotaInBytes; + // Data class of Quota. + private static QuotaList quotaList; + + /** Setting QuotaList parameters from large to small. */ + static { + quotaList = new QuotaList(); + quotaList.addQuotaList(OZONE_QUOTA_TB, Units.TB, TB); + quotaList.addQuotaList(OZONE_QUOTA_GB, Units.GB, GB); + quotaList.addQuotaList(OZONE_QUOTA_MB, Units.MB, MB); + quotaList.addQuotaList(OZONE_QUOTA_KB, Units.KB, KB); + quotaList.addQuotaList(OZONE_QUOTA_BYTES, Units.BYTES, 1L); + } /** * Used to convert user input values into bytes such as: 1MB-> 1048576. @@ -72,24 +84,17 @@ public long getSize() { } /** - * Returns size in Bytes or -1 if there is no Quota. + * Returns size in Bytes or negative num if there is no Quota. */ public long sizeInBytes() { - switch (this.unit) { - case BYTES: - return this.getSize(); - case KB: - return this.getSize() * KB; - case MB: - return this.getSize() * MB; - case GB: - return this.getSize() * GB; - case TB: - return this.getSize() * TB; - case UNDEFINED: - default: - return -1; + long sQuota = -1L; + for (Units quota : quotaList.getUnitQuotaArray()) { + if (quota == this.unit) { + sQuota = quotaList.getQuotaSize(quota); + break; + } } + return this.getSize() * sQuota; } @Override @@ -120,11 +125,11 @@ public Units getUnit() { /** * Constructor for Ozone Quota. * - * @param quotaInCounts Volume quota in counts + * @param quotaInNamespace Volume quota in counts * @param rawQuotaInBytes RawQuotaInBytes value */ - private OzoneQuota(long quotaInCounts, RawQuotaInBytes rawQuotaInBytes) { - this.quotaInCounts = quotaInCounts; + private OzoneQuota(long quotaInNamespace, RawQuotaInBytes rawQuotaInBytes) { + this.quotaInNamespace = quotaInNamespace; this.rawQuotaInBytes = rawQuotaInBytes; this.quotaInBytes = rawQuotaInBytes.sizeInBytes(); } @@ -144,12 +149,12 @@ public static String formatQuota(OzoneQuota quota) { * Quota Object. * * @param quotaInBytes Volume quota in bytes - * @param quotaInCounts Volume quota in counts + * @param quotaInNamespace Volume quota in counts * * @return OzoneQuota object */ public static OzoneQuota parseQuota(String quotaInBytes, - long quotaInCounts) { + long quotaInNamespace) { if (Strings.isNullOrEmpty(quotaInBytes)) { throw new IllegalArgumentException( @@ -164,46 +169,22 @@ public static OzoneQuota parseQuota(String quotaInBytes, long quotaMultiplyExact = 0; try { - if (uppercase.endsWith(OZONE_QUOTA_KB)) { - size = uppercase - .substring(0, uppercase.length() - OZONE_QUOTA_KB.length()); - currUnit = Units.KB; - quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), KB); - } - - if (uppercase.endsWith(OZONE_QUOTA_MB)) { - size = uppercase - .substring(0, uppercase.length() - OZONE_QUOTA_MB.length()); - currUnit = Units.MB; - quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), MB); - } - - if (uppercase.endsWith(OZONE_QUOTA_GB)) { - size = uppercase - .substring(0, uppercase.length() - OZONE_QUOTA_GB.length()); - currUnit = Units.GB; - quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), GB); - } - - if (uppercase.endsWith(OZONE_QUOTA_TB)) { - size = uppercase - .substring(0, uppercase.length() - OZONE_QUOTA_TB.length()); - currUnit = Units.TB; - quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), TB); - } - - if (uppercase.endsWith(OZONE_QUOTA_BYTES)) { - size = uppercase - .substring(0, uppercase.length() - OZONE_QUOTA_BYTES.length()); - currUnit = Units.BYTES; - quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), 1L); + for (String quota : quotaList.getOzoneQuotaArray()) { + if (uppercase.endsWith((quota))) { + size = uppercase + .substring(0, uppercase.length() - quota.length()); + currUnit = quotaList.getUnits(quota); + quotaMultiplyExact = Math.multiplyExact(Long.parseLong(size), + quotaList.getQuotaSize(currUnit)); + break; + } } nSize = Long.parseLong(size); } catch (NumberFormatException e) { throw new IllegalArgumentException("Invalid values for quota, to ensure" + - " that the Quota format is legal(supported values are BYTES, KB, " + - "MB, GB and TB)."); - } catch (ArithmeticException e) { + " that the Quota format is legal(supported values are BYTES, " + + " KB, MB, GB and TB)."); + } catch (ArithmeticException e) { LOG.debug("long overflow:\n{}", quotaMultiplyExact); throw new IllegalArgumentException("Invalid values for quota, the quota" + " value cannot be greater than Long.MAX_VALUE BYTES"); @@ -213,7 +194,7 @@ public static OzoneQuota parseQuota(String quotaInBytes, throw new IllegalArgumentException("Quota cannot be negative."); } - return new OzoneQuota(quotaInCounts, + return new OzoneQuota(quotaInNamespace, new RawQuotaInBytes(currUnit, nSize)); } @@ -222,35 +203,25 @@ public static OzoneQuota parseQuota(String quotaInBytes, * Returns OzoneQuota corresponding to size in bytes. * * @param quotaInBytes in bytes to be converted - * @param quotaInCounts in counts to be converted + * @param quotaInNamespace in counts to be converted * * @return OzoneQuota object */ public static OzoneQuota getOzoneQuota(long quotaInBytes, - long quotaInCounts) { - long size; - Units unit; - if (quotaInBytes % TB == 0) { - size = quotaInBytes / TB; - unit = Units.TB; - } else if (quotaInBytes % GB == 0) { - size = quotaInBytes / GB; - unit = Units.GB; - } else if (quotaInBytes % MB == 0) { - size = quotaInBytes / MB; - unit = Units.MB; - } else if (quotaInBytes % KB == 0) { - size = quotaInBytes / KB; - unit = Units.KB; - } else { - size = quotaInBytes; - unit = Units.BYTES; + long quotaInNamespace) { + long size = 1L; + Units unit = Units.BYTES; + for (Long quota : quotaList.getSizeQuotaArray()) { + if (quotaInBytes % quota == 0) { + size = quotaInBytes / quota; + unit = quotaList.getQuotaUnit(quota); + } } - return new OzoneQuota(quotaInCounts, new RawQuotaInBytes(unit, size)); + return new OzoneQuota(quotaInNamespace, new RawQuotaInBytes(unit, size)); } - public long getQuotaInCounts() { - return quotaInCounts; + public long getQuotaInNamespace() { + return quotaInNamespace; } public long getQuotaInBytes() { @@ -260,6 +231,6 @@ public long getQuotaInBytes() { @Override public String toString() { return "Space Bytes Quota: " + rawQuotaInBytes.toString() + "\n" + - "Counts Quota: " + quotaInCounts; + "Counts Quota: " + quotaInNamespace; } } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/QuotaList.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/QuotaList.java new file mode 100644 index 000000000000..205cca1100c6 --- /dev/null +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/QuotaList.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.hdds.client; + +import java.util.ArrayList; + +/** + *This class contains arraylist for storage constant used in OzoneQuota. + */ +public class QuotaList { + private ArrayList ozoneQuota; + private ArrayList unitQuota; + private ArrayList sizeQuota; + + public QuotaList(){ + ozoneQuota = new ArrayList(); + unitQuota = new ArrayList(); + sizeQuota = new ArrayList(); + } + + public void addQuotaList(String oQuota, OzoneQuota.Units uQuota, Long sQuota){ + ozoneQuota.add(oQuota); + unitQuota.add(uQuota); + sizeQuota.add(sQuota); + } + + public ArrayList getOzoneQuotaArray() { + return this.ozoneQuota; + } + + public ArrayList getSizeQuotaArray() { + return this.sizeQuota; + } + + public ArrayList getUnitQuotaArray() { + return this.unitQuota; + } + + public OzoneQuota.Units getUnits(String oQuota){ + return unitQuota.get(ozoneQuota.indexOf(oQuota)); + } + + public Long getQuotaSize(OzoneQuota.Units uQuota){ + return sizeQuota.get(unitQuota.indexOf(uQuota)); + } + + public OzoneQuota.Units getQuotaUnit(Long sQuota){ + return unitQuota.get(sizeQuota.indexOf(sQuota)); + } + +} diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java index 9cfe0f671d22..4b26cf14cb55 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/conf/OzoneConfiguration.java @@ -41,6 +41,9 @@ import org.apache.hadoop.hdds.utils.LegacyHadoopConfigurationSource; import com.google.common.base.Preconditions; +import org.apache.ratis.server.RaftServerConfigKeys; + +import static org.apache.hadoop.hdds.ratis.RatisHelper.HDDS_DATANODE_RATIS_PREFIX_KEY; /** * Configuration for ozone. @@ -49,6 +52,8 @@ public class OzoneConfiguration extends Configuration implements MutableConfigurationSource { static { + addDeprecatedKeys(); + activate(); } @@ -287,4 +292,13 @@ public Map getPropsWithPrefix(String confPrefix) { } return configMap; } + + private static void addDeprecatedKeys(){ + Configuration.addDeprecations(new DeprecationDelta[]{ + new DeprecationDelta(HDDS_DATANODE_RATIS_PREFIX_KEY + "." + + RaftServerConfigKeys.PREFIX + "." + "rpcslowness.timeout", + HDDS_DATANODE_RATIS_PREFIX_KEY + "." + + RaftServerConfigKeys.PREFIX + "." + "rpc.slowness.timeout") + }); + } } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java index 1a42f3a5698e..3c1c6ecf3a1a 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java @@ -57,6 +57,8 @@ public class DatanodeDetails extends NodeImpl implements private long setupTime; private String revision; private String buildDate; + private HddsProtos.NodeOperationalState persistedOpState; + private long persistedOpStateExpiryEpochSec = 0; /** * Constructs DatanodeDetails instance. DatanodeDetails.Builder is used @@ -71,11 +73,16 @@ public class DatanodeDetails extends NodeImpl implements * @param setupTime the setup time of DataNode * @param revision DataNodes's revision * @param buildDate DataNodes's build timestamp + * @param persistedOpState Operational State stored on DN. + * @param persistedOpStateExpiryEpochSec Seconds after the epoch the stored + * state should expire. */ @SuppressWarnings("parameternumber") private DatanodeDetails(UUID uuid, String ipAddress, String hostName, String networkLocation, List ports, String certSerialId, - String version, long setupTime, String revision, String buildDate) { + String version, long setupTime, String revision, String buildDate, + HddsProtos.NodeOperationalState persistedOpState, + long persistedOpStateExpiryEpochSec) { super(hostName, networkLocation, NetConstants.NODE_COST_DEFAULT); this.uuid = uuid; this.uuidString = uuid.toString(); @@ -87,6 +94,8 @@ private DatanodeDetails(UUID uuid, String ipAddress, String hostName, this.setupTime = setupTime; this.revision = revision; this.buildDate = buildDate; + this.persistedOpState = persistedOpState; + this.persistedOpStateExpiryEpochSec = persistedOpStateExpiryEpochSec; } public DatanodeDetails(DatanodeDetails datanodeDetails) { @@ -103,6 +112,9 @@ public DatanodeDetails(DatanodeDetails datanodeDetails) { this.setupTime = datanodeDetails.setupTime; this.revision = datanodeDetails.revision; this.buildDate = datanodeDetails.buildDate; + this.persistedOpState = datanodeDetails.getPersistedOpState(); + this.persistedOpStateExpiryEpochSec = + datanodeDetails.getPersistedOpStateExpiryEpochSec(); } /** @@ -180,6 +192,46 @@ public List getPorts() { return ports; } + /** + * Return the persistedOpState. If the stored value is null, return the + * default value of IN_SERVICE. + * + * @return The OperationalState persisted on the datanode. + */ + public HddsProtos.NodeOperationalState getPersistedOpState() { + if (persistedOpState == null) { + return HddsProtos.NodeOperationalState.IN_SERVICE; + } else { + return persistedOpState; + } + } + + /** + * Set the persistedOpState for this instance. + * + * @param state The new operational state. + */ + public void setPersistedOpState(HddsProtos.NodeOperationalState state) { + this.persistedOpState = state; + } + + /** + * Get the persistedOpStateExpiryEpochSec for the instance. + * @return Seconds from the epoch when the operational state should expire. + */ + public long getPersistedOpStateExpiryEpochSec() { + return persistedOpStateExpiryEpochSec; + } + + /** + * Set persistedOpStateExpiryEpochSec. + * @param expiry The number of second after the epoch the operational state + * should expire. + */ + public void setPersistedOpStateExpiryEpochSec(long expiry) { + this.persistedOpStateExpiryEpochSec = expiry; + } + /** * Given the name returns port number, null if the asked port is not found. * @@ -231,6 +283,13 @@ public static DatanodeDetails getFromProtoBuf( if (datanodeDetailsProto.hasNetworkLocation()) { builder.setNetworkLocation(datanodeDetailsProto.getNetworkLocation()); } + if (datanodeDetailsProto.hasPersistedOpState()) { + builder.setPersistedOpState(datanodeDetailsProto.getPersistedOpState()); + } + if (datanodeDetailsProto.hasPersistedOpStateExpiry()) { + builder.setPersistedOpStateExpiry( + datanodeDetailsProto.getPersistedOpStateExpiry()); + } return builder.build(); } @@ -294,6 +353,10 @@ public HddsProtos.DatanodeDetailsProto getProtoBufMessage() { if (!Strings.isNullOrEmpty(getNetworkLocation())) { builder.setNetworkLocation(getNetworkLocation()); } + if (persistedOpState != null) { + builder.setPersistedOpState(persistedOpState); + } + builder.setPersistedOpStateExpiry(persistedOpStateExpiryEpochSec); for (Port port : ports) { builder.addPorts(HddsProtos.Port.newBuilder() @@ -342,6 +405,8 @@ public String toString() { ", networkLocation: " + getNetworkLocation() + ", certSerialId: " + certSerialId + + ", persistedOpState: " + persistedOpState + + ", persistedOpStateExpiryEpochSec: " + persistedOpStateExpiryEpochSec + "}"; } @@ -385,6 +450,8 @@ public static final class Builder { private long setupTime; private String revision; private String buildDate; + private HddsProtos.NodeOperationalState persistedOpState; + private long persistedOpStateExpiryEpochSec = 0; /** * Default private constructor. To create Builder instance use @@ -412,6 +479,9 @@ public Builder setDatanodeDetails(DatanodeDetails details) { this.setupTime = details.getSetupTime(); this.revision = details.getRevision(); this.buildDate = details.getBuildDate(); + this.persistedOpState = details.getPersistedOpState(); + this.persistedOpStateExpiryEpochSec = + details.getPersistedOpStateExpiryEpochSec(); return this; } @@ -542,6 +612,31 @@ public Builder setSetupTime(long time) { return this; } + /* + * Adds persistedOpState. + * + * @param state The operational state persisted on the datanode + * + * @return DatanodeDetails.Builder + */ + public Builder setPersistedOpState(HddsProtos.NodeOperationalState state){ + this.persistedOpState = state; + return this; + } + + /* + * Adds persistedOpStateExpiryEpochSec. + * + * @param expiry The seconds after the epoch the operational state should + * expire. + * + * @return DatanodeDetails.Builder + */ + public Builder setPersistedOpStateExpiry(long expiry){ + this.persistedOpStateExpiryEpochSec = expiry; + return this; + } + /** * Builds and returns DatanodeDetails instance. * @@ -553,8 +648,8 @@ public DatanodeDetails build() { networkLocation = NetConstants.DEFAULT_RACK; } DatanodeDetails dn = new DatanodeDetails(id, ipAddress, hostName, - networkLocation, ports, certSerialId, - version, setupTime, revision, buildDate); + networkLocation, ports, certSerialId, version, setupTime, revision, + buildDate, persistedOpState, persistedOpStateExpiryEpochSec); if (networkName != null) { dn.setNetworkName(networkName); } diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java index 0e16968def56..74f89c8f79c6 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java @@ -239,7 +239,7 @@ public final class ScmConfigKeys { public static final String OZONE_SCM_HEARTBEAT_RPC_TIMEOUT = "ozone.scm.heartbeat.rpc-timeout"; public static final String OZONE_SCM_HEARTBEAT_RPC_TIMEOUT_DEFAULT = - "1s"; + "5s"; public static final String OZONE_SCM_HEARTBEAT_RPC_RETRY_COUNT = "ozone.scm.heartbeat.rpc-retry-count"; @@ -451,6 +451,11 @@ public final class ScmConfigKeys { OZONE_SCM_RATIS_SERVER_ROLE_CHECK_INTERVAL_DEFAULT = TimeDuration.valueOf(15, TimeUnit.SECONDS); + public static final String OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL = + "ozone.scm.datanode.admin.monitor.interval"; + public static final String OZONE_SCM_DATANODE_ADMIN_MONITOR_INTERVAL_DEFAULT = + "30s"; + /** * Never constructed. */ diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java index 7c3c94cb7ae1..b8d73e8695f8 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/client/ScmClient.java @@ -144,15 +144,52 @@ ContainerWithPipeline createContainer(HddsProtos.ReplicationType type, String owner) throws IOException; /** - * Returns a set of Nodes that meet a query criteria. - * @param nodeStatuses - Criteria that we want the node to have. + * Returns a set of Nodes that meet a query criteria. Passing null for opState + * or nodeState acts like a wild card, returning all nodes in that state. + * @param opState - Operational State of the node, eg IN_SERVICE, + * DECOMMISSIONED, etc + * @param nodeState - Health of the nodeCriteria that we want the node to + * have, eg HEALTHY, STALE etc * @param queryScope - Query scope - Cluster or pool. * @param poolName - if it is pool, a pool name is required. * @return A set of nodes that meet the requested criteria. * @throws IOException */ - List queryNode(HddsProtos.NodeState nodeStatuses, - HddsProtos.QueryScope queryScope, String poolName) throws IOException; + List queryNode(HddsProtos.NodeOperationalState opState, + HddsProtos.NodeState nodeState, HddsProtos.QueryScope queryScope, + String poolName) throws IOException; + + /** + * Allows a list of hosts to be decommissioned. The hosts are identified + * by their hostname and optionally port in the format foo.com:port. + * @param hosts A list of hostnames, optionally with port + * @throws IOException + */ + void decommissionNodes(List hosts) throws IOException; + + /** + * Allows a list of hosts in maintenance or decommission states to be placed + * back in service. The hosts are identified by their hostname and optionally + * port in the format foo.com:port. + * @param hosts A list of hostnames, optionally with port + * @throws IOException + */ + void recommissionNodes(List hosts) throws IOException; + + /** + * Place the list of datanodes into maintenance mode. If a non-zero endDtm + * is passed, the hosts will automatically exit maintenance mode after the + * given time has passed. Passing an end time of zero means the hosts will + * remain in maintenance indefinitely. + * The hosts are identified by their hostname and optionally port in the + * format foo.com:port. + * @param hosts A list of hostnames, optionally with port + * @param endHours The number of hours from now which maintenance will end or + * zero if maintenance must be manually ended. + * @throws IOException + */ + void startMaintenanceNodes(List hosts, int endHours) + throws IOException; /** * Creates a specified replication pipeline. diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java index 1d58c97a9e36..1b63f5014242 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocol.java @@ -119,12 +119,22 @@ List listContainer(long startContainerID, int count) void deleteContainer(long containerID) throws IOException; /** - * Queries a list of Node Statuses. - * @param state + * Queries a list of Node Statuses. Passing a null for either opState or + * state acts like a wildcard returning all nodes in that state. + * @param opState The node operational state + * @param state The node health * @return List of Datanodes. */ - List queryNode(HddsProtos.NodeState state, - HddsProtos.QueryScope queryScope, String poolName) throws IOException; + List queryNode(HddsProtos.NodeOperationalState opState, + HddsProtos.NodeState state, HddsProtos.QueryScope queryScope, + String poolName) throws IOException; + + void decommissionNodes(List nodes) throws IOException; + + void recommissionNodes(List nodes) throws IOException; + + void startMaintenanceNodes(List nodes, int endInHours) + throws IOException; /** * Close a container. diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java index 03da6dd9dab6..368d03c2d994 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java @@ -23,6 +23,8 @@ import org.apache.ratis.thirdparty.io.grpc.Context; import org.apache.ratis.thirdparty.io.grpc.Metadata; +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.util.regex.Pattern; import static org.apache.ratis.thirdparty.io.grpc.Metadata.ASCII_STRING_MARSHALLER; @@ -268,8 +270,9 @@ private OzoneConsts() { public static final String SRC_KEY = "srcKey"; public static final String DST_KEY = "dstKey"; public static final String USED_BYTES = "usedBytes"; + public static final String USED_NAMESPACE = "usedNamespace"; public static final String QUOTA_IN_BYTES = "quotaInBytes"; - public static final String QUOTA_IN_COUNTS = "quotaInCounts"; + public static final String QUOTA_IN_NAMESPACE = "quotaInNamespace"; public static final String OBJECT_ID = "objectID"; public static final String UPDATE_ID = "updateID"; public static final String CLIENT_ID = "clientID"; @@ -351,7 +354,7 @@ private OzoneConsts() { public static final String GDPR_FLAG = "gdprEnabled"; public static final String GDPR_ALGORITHM_NAME = "AES"; public static final int GDPR_DEFAULT_RANDOM_SECRET_LENGTH = 16; - public static final String GDPR_CHARSET = "UTF-8"; + public static final Charset GDPR_CHARSET = StandardCharsets.UTF_8; public static final String GDPR_LENGTH = "length"; public static final String GDPR_SECRET = "secret"; public static final String GDPR_ALGORITHM = "algorithm"; diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java index 5e52b4050acc..68ae49b26a27 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/lease/LeaseManager.java @@ -17,9 +17,6 @@ package org.apache.hadoop.ozone.lease; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - import java.util.List; import java.util.Map; import java.util.concurrent.Callable; @@ -28,6 +25,8 @@ import java.util.concurrent.Executors; import static org.apache.hadoop.ozone.lease.Lease.messageForResource; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * LeaseManager is someone who can provide you leases based on your @@ -46,6 +45,7 @@ public class LeaseManager { private final String name; private final long defaultTimeout; + private final Object monitor = new Object(); private Map> activeLeases; private LeaseMonitor leaseMonitor; private Thread leaseMonitorThread; @@ -115,12 +115,14 @@ public synchronized Lease acquire(T resource, long timeout) if (LOG.isDebugEnabled()) { LOG.debug("Acquiring lease on {} for {} milliseconds", resource, timeout); } - if(activeLeases.containsKey(resource)) { + if (activeLeases.containsKey(resource)) { throw new LeaseAlreadyExistException(messageForResource(resource)); } Lease lease = new Lease<>(resource, timeout); activeLeases.put(resource, lease); - leaseMonitorThread.interrupt(); + synchronized (monitor) { + monitor.notifyAll(); + } return lease; } @@ -135,7 +137,7 @@ public synchronized Lease acquire(T resource, long timeout) public Lease get(T resource) throws LeaseNotFoundException { checkStatus(); Lease lease = activeLeases.get(resource); - if(lease != null) { + if (lease != null) { return lease; } throw new LeaseNotFoundException(messageForResource(resource)); @@ -156,7 +158,7 @@ public synchronized void release(T resource) LOG.debug("Releasing lease on {}", resource); } Lease lease = activeLeases.remove(resource); - if(lease == null) { + if (lease == null) { throw new LeaseNotFoundException(messageForResource(resource)); } lease.invalidate(); @@ -171,11 +173,13 @@ public void shutdown() { checkStatus(); LOG.debug("Shutting down LeaseManager service"); leaseMonitor.disable(); - leaseMonitorThread.interrupt(); - for(T resource : activeLeases.keySet()) { + synchronized (monitor) { + monitor.notifyAll(); + } + for (T resource : activeLeases.keySet()) { try { release(resource); - } catch(LeaseNotFoundException ex) { + } catch (LeaseNotFoundException ex) { //Ignore the exception, someone might have released the lease } } @@ -187,7 +191,7 @@ public void shutdown() { * running. */ private void checkStatus() { - if(!isRunning) { + if (!isRunning) { throw new LeaseManagerNotRunningException("LeaseManager not running."); } } @@ -198,8 +202,8 @@ private void checkStatus() { */ private final class LeaseMonitor implements Runnable { - private volatile boolean monitor = true; private final ExecutorService executorService; + private volatile boolean running = true; private LeaseMonitor() { this.executorService = Executors.newCachedThreadPool(); @@ -207,7 +211,7 @@ private LeaseMonitor() { @Override public void run() { - while (monitor) { + while (running) { LOG.debug("{}-LeaseMonitor: checking for lease expiry", name); long sleepTime = Long.MAX_VALUE; @@ -230,12 +234,12 @@ public void run() { } try { - if(!Thread.interrupted()) { - Thread.sleep(sleepTime); + synchronized (monitor) { + monitor.wait(sleepTime); } } catch (InterruptedException e) { // This means a new lease is added to activeLeases. - LOG.error("Execution was interrupted ", e); + LOG.warn("Lease manager is interrupted. Shutting down...", e); Thread.currentThread().interrupt(); } } @@ -246,7 +250,7 @@ public void run() { * will stop lease monitor. */ public void disable() { - monitor = false; + running = false; } } diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml index 76d959eefb0d..74005a0b0306 100644 --- a/hadoop-hdds/common/src/main/resources/ozone-default.xml +++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml @@ -943,7 +943,7 @@ ozone.scm.heartbeat.rpc-timeout - 1s + 5s OZONE, MANAGEMENT Timeout value for the RPC from Datanode to SCM. @@ -1643,17 +1643,6 @@ - - ozone.om.ratis.server.role.check.interval - 15s - OZONE, OM, RATIS, MANAGEMENT - The interval between OM leader performing a role - check on its ratis server. Ratis server informs OM if it - loses the leader role. The scheduled check is an secondary - check to ensure that the leader role is updated periodically - . - - ozone.om.ratis.snapshot.dir @@ -2515,6 +2504,17 @@ The number of Recon Tasks that are waiting on updates from OM. + + ozone.scm.datanode.admin.monitor.interval + 30s + SCM + + This sets how frequently the datanode admin monitor runs to check for + nodes added to the admin workflow or removed from it. The progress + of decommissioning and entering maintenance nodes is also checked to see + if they have completed. + + ozone.client.list.trash.keys.max 1000 diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/protocol/MockDatanodeDetails.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/protocol/MockDatanodeDetails.java index 06a1bf0f8678..41ae6ec1f723 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/protocol/MockDatanodeDetails.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/protocol/MockDatanodeDetails.java @@ -17,6 +17,8 @@ */ package org.apache.hadoop.hdds.protocol; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; + import java.io.IOException; import java.net.ServerSocket; import java.util.Random; @@ -101,6 +103,8 @@ public static DatanodeDetails createDatanodeDetails(String uuid, .addPort(ratisPort) .addPort(restPort) .setNetworkLocation(networkLocation) + .setPersistedOpState(HddsProtos.NodeOperationalState.IN_SERVICE) + .setPersistedOpStateExpiry(0) .build(); } diff --git a/hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigFileAppender.java b/hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigFileAppender.java index 9f1c08712b43..4256ac81fdaf 100644 --- a/hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigFileAppender.java +++ b/hadoop-hdds/config/src/main/java/org/apache/hadoop/hdds/conf/ConfigFileAppender.java @@ -28,6 +28,7 @@ import javax.xml.transform.stream.StreamResult; import java.io.InputStream; import java.io.Writer; +import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.stream.Collectors; @@ -117,7 +118,8 @@ public void write(Writer writer) { factory.setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true); Transformer transformer = factory.newTransformer(); - transformer.setOutputProperty(OutputKeys.ENCODING, "UTF-8"); + transformer.setOutputProperty(OutputKeys.ENCODING, + StandardCharsets.UTF_8.name()); transformer.setOutputProperty(OutputKeys.INDENT, "yes"); transformer.setOutputProperty("{http://xml.apache.org/xslt}indent-amount", "2"); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DatanodeIdYaml.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DatanodeIdYaml.java index 44a12c27c79a..3b14641d57a1 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DatanodeIdYaml.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/helpers/DatanodeIdYaml.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.io.OutputStreamWriter; import java.io.Writer; +import java.nio.charset.StandardCharsets; import java.util.LinkedHashMap; import java.util.Map; import java.util.UUID; @@ -30,6 +31,7 @@ import org.apache.commons.collections.CollectionUtils; import org.apache.commons.collections.MapUtils; import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.yaml.snakeyaml.DumperOptions; import org.yaml.snakeyaml.Yaml; @@ -57,7 +59,7 @@ public static void createDatanodeIdFile(DatanodeDetails datanodeDetails, Yaml yaml = new Yaml(options); try (Writer writer = new OutputStreamWriter( - new FileOutputStream(path), "UTF-8")) { + new FileOutputStream(path), StandardCharsets.UTF_8)) { yaml.dump(getDatanodeDetailsYaml(datanodeDetails), writer); } } @@ -83,6 +85,12 @@ public static DatanodeDetails readDatanodeIdFile(File path) .setIpAddress(datanodeDetailsYaml.getIpAddress()) .setHostName(datanodeDetailsYaml.getHostName()) .setCertSerialId(datanodeDetailsYaml.getCertSerialId()); + if (datanodeDetailsYaml.getPersistedOpState() != null) { + builder.setPersistedOpState(HddsProtos.NodeOperationalState.valueOf( + datanodeDetailsYaml.getPersistedOpState())); + } + builder.setPersistedOpStateExpiry( + datanodeDetailsYaml.getPersistedOpStateExpiryEpochSec()); if (!MapUtils.isEmpty(datanodeDetailsYaml.getPortDetails())) { for (Map.Entry portEntry : @@ -106,6 +114,8 @@ public static class DatanodeDetailsYaml { private String ipAddress; private String hostName; private String certSerialId; + private String persistedOpState; + private long persistedOpStateExpiryEpochSec = 0; private Map portDetails; public DatanodeDetailsYaml() { @@ -114,11 +124,15 @@ public DatanodeDetailsYaml() { private DatanodeDetailsYaml(String uuid, String ipAddress, String hostName, String certSerialId, + String persistedOpState, + long persistedOpStateExpiryEpochSec, Map portDetails) { this.uuid = uuid; this.ipAddress = ipAddress; this.hostName = hostName; this.certSerialId = certSerialId; + this.persistedOpState = persistedOpState; + this.persistedOpStateExpiryEpochSec = persistedOpStateExpiryEpochSec; this.portDetails = portDetails; } @@ -138,6 +152,14 @@ public String getCertSerialId() { return certSerialId; } + public String getPersistedOpState() { + return persistedOpState; + } + + public long getPersistedOpStateExpiryEpochSec() { + return persistedOpStateExpiryEpochSec; + } + public Map getPortDetails() { return portDetails; } @@ -158,6 +180,14 @@ public void setCertSerialId(String certSerialId) { this.certSerialId = certSerialId; } + public void setPersistedOpState(String persistedOpState) { + this.persistedOpState = persistedOpState; + } + + public void setPersistedOpStateExpiryEpochSec(long opStateExpiryEpochSec) { + this.persistedOpStateExpiryEpochSec = opStateExpiryEpochSec; + } + public void setPortDetails(Map portDetails) { this.portDetails = portDetails; } @@ -173,11 +203,17 @@ private static DatanodeDetailsYaml getDatanodeDetailsYaml( } } + String persistedOpString = null; + if (datanodeDetails.getPersistedOpState() != null) { + persistedOpString = datanodeDetails.getPersistedOpState().name(); + } return new DatanodeDetailsYaml( datanodeDetails.getUuid().toString(), datanodeDetails.getIpAddress(), datanodeDetails.getHostName(), datanodeDetails.getCertSerialId(), + persistedOpString, + datanodeDetails.getPersistedOpStateExpiryEpochSec(), portDetails); } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java index ba34a29e02bd..19cc1e2138bc 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerData.java @@ -291,7 +291,7 @@ public synchronized boolean isOpen() { * @return - boolean */ public synchronized boolean isValid() { - return !(ContainerDataProto.State.INVALID == state); + return ContainerDataProto.State.INVALID != state; } /** diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java index 74cbbc0af846..757d7e856d63 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/impl/ContainerDataYaml.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.container.common.impl; -import java.beans.IntrospectionException; import java.io.ByteArrayInputStream; import java.io.File; import java.io.FileInputStream; @@ -201,8 +200,7 @@ public static Yaml getYamlForContainerType(ContainerType containerType) */ private static class ContainerDataRepresenter extends Representer { @Override - protected Set getProperties(Class type) - throws IntrospectionException { + protected Set getProperties(Class type) { Set set = super.getProperties(type); Set filtered = new TreeSet(); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportPublisher.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportPublisher.java index 685a1d993f92..5d181eca10d7 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportPublisher.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/report/ReportPublisher.java @@ -69,7 +69,7 @@ public ConfigurationSource getConf() { public void run() { publishReport(); if (!executor.isShutdown() && - !(context.getState() == DatanodeStates.SHUTDOWN)) { + (context.getState() != DatanodeStates.SHUTDOWN)) { executor.schedule(this, getReportFrequency(), TimeUnit.MILLISECONDS); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java index 26bc853b0fe9..a7cb2d3de7e3 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java @@ -43,6 +43,7 @@ import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.DeleteBlocksCommandHandler; import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.DeleteContainerCommandHandler; import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.ReplicateContainerCommandHandler; +import org.apache.hadoop.ozone.container.common.statemachine.commandhandler.SetNodeOperationalStateCommandHandler; import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker; import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.container.replication.ContainerReplicator; @@ -146,6 +147,7 @@ public DatanodeStateMachine(DatanodeDetails datanodeDetails, dnConf.getContainerDeleteThreads())) .addHandler(new ClosePipelineCommandHandler()) .addHandler(new CreatePipelineCommandHandler(conf)) + .addHandler(new SetNodeOperationalStateCommandHandler(conf)) .setConnectionManager(connectionManager) .setContainer(container) .setContext(context) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java index f39755ffe8fc..b8387b378665 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/StateContext.java @@ -34,15 +34,23 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import java.util.function.Consumer; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.Sets; +import com.google.protobuf.Descriptors.Descriptor; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus.Status; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineAction; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.ozone.container.common.states.DatanodeState; import org.apache.hadoop.ozone.container.common.states.datanode.InitDatanodeState; @@ -65,6 +73,27 @@ * Current Context of State Machine. */ public class StateContext { + + @VisibleForTesting + final static String CONTAINER_REPORTS_PROTO_NAME = + ContainerReportsProto.getDescriptor().getFullName(); + @VisibleForTesting + final static String NODE_REPORT_PROTO_NAME = + NodeReportProto.getDescriptor().getFullName(); + @VisibleForTesting + final static String PIPELINE_REPORTS_PROTO_NAME = + PipelineReportsProto.getDescriptor().getFullName(); + @VisibleForTesting + final static String COMMAND_STATUS_REPORTS_PROTO_NAME = + CommandStatusReportsProto.getDescriptor().getFullName(); + @VisibleForTesting + final static String INCREMENTAL_CONTAINER_REPORT_PROTO_NAME = + IncrementalContainerReportProto.getDescriptor().getFullName(); + // Accepted types of reports that can be queued to incrementalReportsQueue + private final static Set ACCEPTED_INCREMENTAL_REPORT_TYPE_SET = + Sets.newHashSet(COMMAND_STATUS_REPORTS_PROTO_NAME, + INCREMENTAL_CONTAINER_REPORT_PROTO_NAME); + static final Logger LOG = LoggerFactory.getLogger(StateContext.class); private final Queue commandQueue; @@ -74,7 +103,13 @@ public class StateContext { private final AtomicLong stateExecutionCount; private final ConfigurationSource conf; private final Set endpoints; - private final Map> reports; + // Only the latest full report of each type is kept + private final AtomicReference containerReports; + private final AtomicReference nodeReport; + private final AtomicReference pipelineReports; + // Incremental reports are queued in the map below + private final Map> + incrementalReportsQueue; private final Map> containerActions; private final Map> pipelineActions; private DatanodeStateMachine.DatanodeStates state; @@ -116,7 +151,10 @@ public StateContext(ConfigurationSource conf, this.parent = parent; commandQueue = new LinkedList<>(); cmdStatusMap = new ConcurrentHashMap<>(); - reports = new HashMap<>(); + incrementalReportsQueue = new HashMap<>(); + containerReports = new AtomicReference<>(); + nodeReport = new AtomicReference<>(); + pipelineReports = new AtomicReference<>(); endpoints = new HashSet<>(); containerActions = new HashMap<>(); pipelineActions = new HashMap<>(); @@ -204,17 +242,34 @@ void setShutdownGracefully() { public boolean getShutdownOnError() { return shutdownOnError; } + /** * Adds the report to report queue. * * @param report report to be added */ public void addReport(GeneratedMessage report) { - if (report != null) { - synchronized (reports) { - for (InetSocketAddress endpoint : endpoints) { - reports.get(endpoint).add(report); + if (report == null) { + return; + } + final Descriptor descriptor = report.getDescriptorForType(); + Preconditions.checkState(descriptor != null); + final String reportType = descriptor.getFullName(); + Preconditions.checkState(reportType != null); + for (InetSocketAddress endpoint : endpoints) { + if (reportType.equals(CONTAINER_REPORTS_PROTO_NAME)) { + containerReports.set(report); + } else if (reportType.equals(NODE_REPORT_PROTO_NAME)) { + nodeReport.set(report); + } else if (reportType.equals(PIPELINE_REPORTS_PROTO_NAME)) { + pipelineReports.set(report); + } else if (ACCEPTED_INCREMENTAL_REPORT_TYPE_SET.contains(reportType)) { + synchronized (incrementalReportsQueue) { + incrementalReportsQueue.get(endpoint).add(report); } + } else { + throw new IllegalArgumentException( + "Unidentified report message type: " + reportType); } } } @@ -228,9 +283,24 @@ public void addReport(GeneratedMessage report) { */ public void putBackReports(List reportsToPutBack, InetSocketAddress endpoint) { - synchronized (reports) { - if (reports.containsKey(endpoint)){ - reports.get(endpoint).addAll(0, reportsToPutBack); + if (LOG.isDebugEnabled()) { + LOG.debug("endpoint: {}, size of reportsToPutBack: {}", + endpoint, reportsToPutBack.size()); + } + // We don't expect too much reports to be put back + for (GeneratedMessage report : reportsToPutBack) { + final Descriptor descriptor = report.getDescriptorForType(); + Preconditions.checkState(descriptor != null); + final String reportType = descriptor.getFullName(); + Preconditions.checkState(reportType != null); + if (!ACCEPTED_INCREMENTAL_REPORT_TYPE_SET.contains(reportType)) { + throw new IllegalArgumentException( + "Unaccepted report message type: " + reportType); + } + } + synchronized (incrementalReportsQueue) { + if (incrementalReportsQueue.containsKey(endpoint)){ + incrementalReportsQueue.get(endpoint).addAll(0, reportsToPutBack); } } } @@ -246,6 +316,22 @@ public List getAllAvailableReports( return getReports(endpoint, Integer.MAX_VALUE); } + List getIncrementalReports( + InetSocketAddress endpoint, int maxLimit) { + List reportsToReturn = new LinkedList<>(); + synchronized (incrementalReportsQueue) { + List reportsForEndpoint = + incrementalReportsQueue.get(endpoint); + if (reportsForEndpoint != null) { + List tempList = reportsForEndpoint.subList( + 0, min(reportsForEndpoint.size(), maxLimit)); + reportsToReturn.addAll(tempList); + tempList.clear(); + } + } + return reportsToReturn; + } + /** * Returns available reports from the report queue with a max limit on * list size, or empty list if the queue is empty. @@ -254,15 +340,19 @@ public List getAllAvailableReports( */ public List getReports(InetSocketAddress endpoint, int maxLimit) { - List reportsToReturn = new LinkedList<>(); - synchronized (reports) { - List reportsForEndpoint = reports.get(endpoint); - if (reportsForEndpoint != null) { - List tempList = reportsForEndpoint.subList( - 0, min(reportsForEndpoint.size(), maxLimit)); - reportsToReturn.addAll(tempList); - tempList.clear(); - } + List reportsToReturn = + getIncrementalReports(endpoint, maxLimit); + GeneratedMessage report = containerReports.get(); + if (report != null) { + reportsToReturn.add(report); + } + report = nodeReport.get(); + if (report != null) { + reportsToReturn.add(report); + } + report = pipelineReports.get(); + if (report != null) { + reportsToReturn.add(report); } return reportsToReturn; } @@ -672,7 +762,22 @@ public void addEndpoint(InetSocketAddress endpoint) { this.endpoints.add(endpoint); this.containerActions.put(endpoint, new LinkedList<>()); this.pipelineActions.put(endpoint, new LinkedList<>()); - this.reports.put(endpoint, new LinkedList<>()); + this.incrementalReportsQueue.put(endpoint, new LinkedList<>()); } } + + @VisibleForTesting + public GeneratedMessage getContainerReports() { + return containerReports.get(); + } + + @VisibleForTesting + public GeneratedMessage getNodeReport() { + return nodeReport.get(); + } + + @VisibleForTesting + public GeneratedMessage getPipelineReports() { + return pipelineReports.get(); + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/SetNodeOperationalStateCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/SetNodeOperationalStateCommandHandler.java new file mode 100644 index 000000000000..4a46d5fb8965 --- /dev/null +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/SetNodeOperationalStateCommandHandler.java @@ -0,0 +1,157 @@ +/** + * 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.statemachine.commandhandler; + + +import com.google.common.base.Preconditions; +import org.apache.hadoop.hdds.conf.ConfigurationSource; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; +import org.apache.hadoop.hdds.scm.ScmConfigKeys; +import org.apache.hadoop.hdds.utils.HddsServerUtil; +import org.apache.hadoop.ozone.container.common.helpers.ContainerUtils; +import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager; +import org.apache.hadoop.ozone.container.common.statemachine.StateContext; +import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; +import org.apache.hadoop.ozone.protocol.commands.SCMCommand; +import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand; +import org.apache.hadoop.util.Time; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import org.apache.hadoop.hdds.protocol.proto. + StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type; + +import java.io.File; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + + +/** + * Handle the SetNodeOperationalStateCommand sent from SCM to the datanode + * to persist the current operational state. + */ +public class SetNodeOperationalStateCommandHandler implements CommandHandler { + + private static final Logger LOG = + LoggerFactory.getLogger(SetNodeOperationalStateCommandHandler.class); + private final ConfigurationSource conf; + private final AtomicInteger invocationCount = new AtomicInteger(0); + private final AtomicLong totalTime = new AtomicLong(0); + + /** + * Set Node State command handler. + * + * @param conf - Configuration for the datanode. + */ + public SetNodeOperationalStateCommandHandler(ConfigurationSource conf) { + this.conf = conf; + } + + /** + * Handles a given SCM command. + * + * @param command - SCM Command + * @param container - Ozone Container. + * @param context - Current Context. + * @param connectionManager - The SCMs that we are talking to. + */ + @Override + public void handle(SCMCommand command, OzoneContainer container, + StateContext context, SCMConnectionManager connectionManager) { + long startTime = Time.monotonicNow(); + invocationCount.incrementAndGet(); + StorageContainerDatanodeProtocolProtos.SetNodeOperationalStateCommandProto + setNodeCmdProto = null; + + if (command.getType() != Type.setNodeOperationalStateCommand) { + LOG.warn("Skipping handling command, expected command " + + "type {} but found {}", + Type.setNodeOperationalStateCommand, command.getType()); + return; + } + SetNodeOperationalStateCommand setNodeCmd = + (SetNodeOperationalStateCommand) command; + setNodeCmdProto = setNodeCmd.getProto(); + DatanodeDetails dni = context.getParent().getDatanodeDetails(); + dni.setPersistedOpState(setNodeCmdProto.getNodeOperationalState()); + dni.setPersistedOpStateExpiryEpochSec( + setNodeCmd.getStateExpiryEpochSeconds()); + try { + persistDatanodeDetails(dni); + } catch (IOException ioe) { + LOG.error("Failed to persist the datanode state", ioe); + // TODO - this should probably be raised, but it will break the command + // handler interface. + } + totalTime.addAndGet(Time.monotonicNow() - startTime); + } + + // TODO - this duplicates code in HddsDatanodeService and InitDatanodeState + // Need to refactor. + private void persistDatanodeDetails(DatanodeDetails dnDetails) + throws IOException { + String idFilePath = HddsServerUtil.getDatanodeIdFilePath(conf); + if (idFilePath == null || idFilePath.isEmpty()) { + LOG.error("A valid path is needed for config setting {}", + ScmConfigKeys.OZONE_SCM_DATANODE_ID_DIR); + throw new IllegalArgumentException( + ScmConfigKeys.OZONE_SCM_DATANODE_ID_DIR + + " must be defined. See" + + " https://wiki.apache.org/hadoop/Ozone#Configuration" + + " for details on configuring Ozone."); + } + + Preconditions.checkNotNull(idFilePath); + File idFile = new File(idFilePath); + ContainerUtils.writeDatanodeDetailsTo(dnDetails, idFile); + } + + /** + * Returns the command type that this command handler handles. + * + * @return Type + */ + @Override + public StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type + getCommandType() { + return Type.setNodeOperationalStateCommand; + } + + /** + * Returns number of times this handler has been invoked. + * + * @return int + */ + @Override + public int getInvocationCount() { + return invocationCount.intValue(); + } + + /** + * Returns the average time this function takes to run. + * + * @return long + */ + @Override + public long getAverageRunTime() { + final int invocations = invocationCount.get(); + return invocations == 0 ? + 0 : totalTime.get() / invocations; + } +} \ No newline at end of file diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index eac7b37e3383..f656d84c41f7 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -52,6 +52,7 @@ import org.apache.hadoop.ozone.protocol.commands.DeleteContainerCommand; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; +import org.apache.hadoop.ozone.protocol.commands.SetNodeOperationalStateCommand; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -141,15 +142,15 @@ public EndpointStateMachine.EndPointStates call() throws Exception { if (LOG.isDebugEnabled()) { LOG.debug("Sending heartbeat message :: {}", request.toString()); } - SCMHeartbeatResponseProto reponse = rpcEndpoint.getEndPoint() + SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint() .sendHeartbeat(request); - processResponse(reponse, datanodeDetailsProto); + processResponse(response, datanodeDetailsProto); rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now()); rpcEndpoint.zeroMissedCount(); } catch (IOException ex) { + Preconditions.checkState(requestBuilder != null); // put back the reports which failed to be sent putBackReports(requestBuilder); - rpcEndpoint.logIfNeeded(ex); } finally { rpcEndpoint.unlock(); @@ -160,12 +161,9 @@ public EndpointStateMachine.EndPointStates call() throws Exception { // TODO: Make it generic. private void putBackReports(SCMHeartbeatRequestProto.Builder requestBuilder) { List reports = new LinkedList<>(); - if (requestBuilder.hasContainerReport()) { - reports.add(requestBuilder.getContainerReport()); - } - if (requestBuilder.hasNodeReport()) { - reports.add(requestBuilder.getNodeReport()); - } + // We only put back CommandStatusReports and IncrementalContainerReport + // because those are incremental. Container/Node/PipelineReport are + // accumulative so we can keep only the latest of each. if (requestBuilder.getCommandStatusReportsCount() != 0) { reports.addAll(requestBuilder.getCommandStatusReportsList()); } @@ -193,6 +191,7 @@ private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) { } else { requestBuilder.setField(descriptor, report); } + break; } } } @@ -349,6 +348,17 @@ private void processResponse(SCMHeartbeatResponseProto response, } this.context.addCommand(closePipelineCommand); break; + case setNodeOperationalStateCommand: + SetNodeOperationalStateCommand setNodeOperationalStateCommand = + SetNodeOperationalStateCommand.getFromProtobuf( + commandResponseProto.getSetNodeOperationalStateCommandProto()); + if (LOG.isDebugEnabled()) { + LOG.debug("Received SCM set operational state command. State: {} " + + "Expiry: {}", setNodeOperationalStateCommand.getOpState(), + setNodeOperationalStateCommand.getStateExpiryEpochSeconds()); + } + this.context.addCommand(setNodeOperationalStateCommand); + break; default: throw new IllegalArgumentException("Unknown response : " + commandResponseProto.getCommandType().name()); 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 1a87ce55e26d..a79f09b93299 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 @@ -763,7 +763,8 @@ public CompletableFuture applyTransaction(TransactionContext trx) { } }, getCommandExecutor(requestProto)); future.thenApply(r -> { - if (trx.getServerRole() == RaftPeerRole.LEADER) { + if (trx.getServerRole() == RaftPeerRole.LEADER + && trx.getStateMachineContext() != null) { long startTime = (long) trx.getStateMachineContext(); metrics.incPipelineLatency(cmdType, Time.monotonicNowNanos() - startTime); @@ -811,6 +812,12 @@ public CompletableFuture applyTransaction(TransactionContext trx) { } return applyTransactionFuture; }).whenComplete((r, t) -> { + if (t != null) { + stateMachineHealthy.set(false); + LOG.error("gid {} : ApplyTransaction failed. cmd {} logIndex " + + "{} exception {}", gid, requestProto.getCmdType(), + index, t); + } applyTransactionSemaphore.release(); metrics.recordApplyTransactionCompletion( Time.monotonicNowNanos() - applyTxnStartTime); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java index cb281f062236..1e9e64749a09 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisor.java @@ -25,11 +25,11 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicLong; -import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.apache.hadoop.ozone.container.common.impl.ContainerSet; import org.apache.hadoop.ozone.container.replication.ReplicationTask.Status; import com.google.common.annotations.VisibleForTesting; +import com.google.common.util.concurrent.ThreadFactoryBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -58,7 +58,8 @@ public class ReplicationSupervisor { @VisibleForTesting ReplicationSupervisor( ContainerSet containerSet, ContainerReplicator replicator, - ExecutorService executor) { + ExecutorService executor + ) { this.containerSet = containerSet; this.replicator = replicator; this.containersInFlight = ConcurrentHashMap.newKeySet(); @@ -67,9 +68,10 @@ public class ReplicationSupervisor { public ReplicationSupervisor( ContainerSet containerSet, - ContainerReplicator replicator, int poolSize) { + ContainerReplicator replicator, int poolSize + ) { this(containerSet, replicator, new ThreadPoolExecutor( - 0, poolSize, 60, TimeUnit.SECONDS, + poolSize, poolSize, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new ThreadFactoryBuilder().setDaemon(true) .setNameFormat("ContainerReplicationThread-%d") @@ -85,6 +87,12 @@ public void addTask(ReplicationTask task) { } } + @VisibleForTesting + public void shutdownAfterFinish() throws InterruptedException { + executor.shutdown(); + executor.awaitTermination(1L, TimeUnit.DAYS); + } + public void stop() { try { executor.shutdown(); @@ -100,6 +108,7 @@ public void stop() { /** * Get the number of containers currently being downloaded * or scheduled for download. + * * @return Count of in-flight replications. */ @VisibleForTesting diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java index c1445bde58ca..5d8a86bc930a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/SimpleContainerDownloader.java @@ -18,7 +18,6 @@ package org.apache.hadoop.ozone.container.replication; -import java.io.IOException; import java.nio.file.Path; import java.nio.file.Paths; import java.security.cert.X509Certificate; @@ -26,7 +25,6 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.function.Function; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -35,6 +33,7 @@ import org.apache.hadoop.ozone.OzoneConfigKeys; import com.google.common.annotations.VisibleForTesting; +import org.jetbrains.annotations.NotNull; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,8 +53,10 @@ public class SimpleContainerDownloader implements ContainerDownloader { private final SecurityConfig securityConfig; private final X509Certificate caCert; - public SimpleContainerDownloader(ConfigurationSource conf, - X509Certificate caCert) { + public SimpleContainerDownloader( + ConfigurationSource conf, + X509Certificate caCert + ) { String workDirString = conf.get(OzoneConfigKeys.OZONE_CONTAINER_COPY_WORKDIR); @@ -71,38 +72,31 @@ public SimpleContainerDownloader(ConfigurationSource conf, } @Override - public CompletableFuture getContainerDataFromReplicas(long containerId, - List sourceDatanodes) { + public CompletableFuture getContainerDataFromReplicas( + long containerId, + List sourceDatanodes + ) { CompletableFuture result = null; - //There is a chance for the download is successful but import is failed, - //due to data corruption. We need a random selected datanode to have a - //chance to succeed next time. - final ArrayList shuffledDatanodes = - new ArrayList<>(sourceDatanodes); - Collections.shuffle(shuffledDatanodes); + final List shuffledDatanodes = + shuffleDatanodes(sourceDatanodes); for (DatanodeDetails datanode : shuffledDatanodes) { try { if (result == null) { result = downloadContainer(containerId, datanode); } else { - result = result.thenApply(CompletableFuture::completedFuture) - .exceptionally(t -> { - LOG.error("Error on replicating container: " + containerId, t); - try { - GrpcReplicationClient grpcReplicationClient = - new GrpcReplicationClient(datanode.getIpAddress(), - datanode.getPort(Name.STANDALONE).getValue(), - workingDirectory, securityConfig, caCert); - return grpcReplicationClient.download(containerId); - } catch (IOException e) { - LOG.error("Error on replicating container: " + containerId, - t); - return null; - } - }).thenCompose(Function.identity()); + result = result.exceptionally(t -> { + LOG.error("Error on replicating container: " + containerId, t); + try { + return downloadContainer(containerId, datanode).join(); + } catch (Exception e) { + LOG.error("Error on replicating container: " + containerId, + e); + return null; + } + }); } } catch (Exception ex) { LOG.error(String.format( @@ -114,18 +108,42 @@ public CompletableFuture getContainerDataFromReplicas(long containerId, } + //There is a chance for the download is successful but import is failed, + //due to data corruption. We need a random selected datanode to have a + //chance to succeed next time. + @NotNull + protected List shuffleDatanodes( + List sourceDatanodes + ) { + + final ArrayList shuffledDatanodes = + new ArrayList<>(sourceDatanodes); + + Collections.shuffle(shuffledDatanodes); + + return shuffledDatanodes; + } + @VisibleForTesting protected CompletableFuture downloadContainer( long containerId, DatanodeDetails datanode ) throws Exception { CompletableFuture result; - try (GrpcReplicationClient grpcReplicationClient = + GrpcReplicationClient grpcReplicationClient = new GrpcReplicationClient(datanode.getIpAddress(), datanode.getPort(Name.STANDALONE).getValue(), - workingDirectory, securityConfig, caCert)) { - result = grpcReplicationClient.download(containerId); - } + workingDirectory, securityConfig, caCert); + result = grpcReplicationClient.download(containerId) + .thenApply(r -> { + try { + grpcReplicationClient.close(); + } catch (Exception e) { + LOG.error("Couldn't close Grpc replication client", e); + } + return r; + }); + return result; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/SetNodeOperationalStateCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/SetNodeOperationalStateCommand.java new file mode 100644 index 000000000000..3ff79498b182 --- /dev/null +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/SetNodeOperationalStateCommand.java @@ -0,0 +1,89 @@ +/** + * 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.protocol.commands; + +import com.google.common.base.Preconditions; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.hdds.protocol.proto + .StorageContainerDatanodeProtocolProtos.SCMCommandProto; +import org.apache.hadoop.hdds.protocol.proto + .StorageContainerDatanodeProtocolProtos.SetNodeOperationalStateCommandProto; + +/** + * A command used to persist the current node operational state on the datanode. + */ +public class SetNodeOperationalStateCommand + extends SCMCommand { + + private final HddsProtos.NodeOperationalState opState; + private long stateExpiryEpochSeconds; + + /** + * Ctor that creates a SetNodeOperationalStateCommand. + * + * @param id - Command ID. Something a time stamp would suffice. + * @param state - OperationalState that want the node to be set into. + * @param stateExpiryEpochSeconds The epoch time when the state should + * expire, or zero for the state to remain + * indefinitely. + */ + public SetNodeOperationalStateCommand(long id, + HddsProtos.NodeOperationalState state, long stateExpiryEpochSeconds) { + super(id); + this.opState = state; + this.stateExpiryEpochSeconds = stateExpiryEpochSeconds; + } + + /** + * Returns the type of this command. + * + * @return Type - This is setNodeOperationalStateCommand. + */ + @Override + public SCMCommandProto.Type getType() { + return SCMCommandProto.Type.setNodeOperationalStateCommand; + } + + /** + * Gets the protobuf message of this object. + * + * @return A protobuf message. + */ + @Override + public SetNodeOperationalStateCommandProto getProto() { + return SetNodeOperationalStateCommandProto.newBuilder() + .setCmdId(getId()) + .setNodeOperationalState(opState) + .setStateExpiryEpochSeconds(stateExpiryEpochSeconds).build(); + } + + public HddsProtos.NodeOperationalState getOpState() { + return opState; + } + + public long getStateExpiryEpochSeconds() { + return stateExpiryEpochSeconds; + } + + public static SetNodeOperationalStateCommand getFromProtobuf( + SetNodeOperationalStateCommandProto cmdProto) { + Preconditions.checkNotNull(cmdProto); + return new SetNodeOperationalStateCommand(cmdProto.getCmdId(), + cmdProto.getNodeOperationalState(), + cmdProto.getStateExpiryEpochSeconds()); + } +} \ No newline at end of file diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index d3032c3211f5..e9c39d3ee026 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -23,11 +23,19 @@ import static org.apache.hadoop.test.GenericTestUtils.waitFor; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.Executors; import java.util.concurrent.ExecutorService; @@ -37,6 +45,7 @@ import java.util.concurrent.atomic.AtomicInteger; import com.google.common.util.concurrent.ThreadFactoryBuilder; +import com.google.protobuf.Descriptors.Descriptor; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineAction; @@ -53,6 +62,271 @@ */ public class TestStateContext { + /** + * Only accepted types of reports can be put back to the report queue. + */ + @Test + public void testPutBackReports() { + OzoneConfiguration conf = new OzoneConfiguration(); + DatanodeStateMachine datanodeStateMachineMock = + mock(DatanodeStateMachine.class); + + StateContext ctx = new StateContext(conf, DatanodeStates.getInitState(), + datanodeStateMachineMock); + InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001); + ctx.addEndpoint(scm1); + InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001); + ctx.addEndpoint(scm2); + + Map expectedReportCount = new HashMap<>(); + + // Case 1: Put back an incremental report + + ctx.putBackReports(Collections.singletonList(newMockReport( + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME)), scm1); + // scm2 report queue should be empty + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + // Check scm1 queue + expectedReportCount.put( + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + // getReports dequeues incremental reports + expectedReportCount.clear(); + + ctx.putBackReports(Collections.singletonList(newMockReport( + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME)), scm2); + // scm1 report queue should be empty + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + // Check scm2 queue + expectedReportCount.put( + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + // getReports dequeues incremental reports + expectedReportCount.clear(); + + // Case 2: Attempt to put back a full report + + try { + ctx.putBackReports(Collections.singletonList( + newMockReport(StateContext.CONTAINER_REPORTS_PROTO_NAME)), scm1); + fail("Should throw exception when putting back unaccepted reports!"); + } catch (IllegalArgumentException ignored) { + } + try { + ctx.putBackReports(Collections.singletonList( + newMockReport(StateContext.NODE_REPORT_PROTO_NAME)), scm2); + fail("Should throw exception when putting back unaccepted reports!"); + } catch (IllegalArgumentException ignored) { + } + try { + ctx.putBackReports(Collections.singletonList( + newMockReport(StateContext.PIPELINE_REPORTS_PROTO_NAME)), scm1); + fail("Should throw exception when putting back unaccepted reports!"); + } catch (IllegalArgumentException ignored) { + } + + // Case 3: Put back mixed types of incremental reports + + ctx.putBackReports(Arrays.asList( + newMockReport(StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME), + newMockReport(StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME), + newMockReport(StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME), + newMockReport(StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME), + newMockReport(StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME) + ), scm1); + // scm2 report queue should be empty + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + // Check scm1 queue + expectedReportCount.put( + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME, 2); + expectedReportCount.put( + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 3); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + // getReports dequeues incremental reports + expectedReportCount.clear(); + + // Case 4: Attempt to put back mixed types of full reports + + try { + ctx.putBackReports(Arrays.asList( + newMockReport(StateContext.CONTAINER_REPORTS_PROTO_NAME), + newMockReport(StateContext.NODE_REPORT_PROTO_NAME), + newMockReport(StateContext.PIPELINE_REPORTS_PROTO_NAME) + ), scm1); + fail("Should throw exception when putting back unaccepted reports!"); + } catch (IllegalArgumentException ignored) { + } + + // Case 5: Attempt to put back mixed full and incremental reports + + try { + ctx.putBackReports(Arrays.asList( + newMockReport(StateContext.CONTAINER_REPORTS_PROTO_NAME), + newMockReport(StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME), + newMockReport(StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME) + ), scm2); + fail("Should throw exception when putting back unaccepted reports!"); + } catch (IllegalArgumentException ignored) { + } + } + + @Test + public void testReportQueueWithAddReports() { + OzoneConfiguration conf = new OzoneConfiguration(); + DatanodeStateMachine datanodeStateMachineMock = + mock(DatanodeStateMachine.class); + + StateContext ctx = new StateContext(conf, DatanodeStates.getInitState(), + datanodeStateMachineMock); + InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001); + ctx.addEndpoint(scm1); + InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001); + ctx.addEndpoint(scm2); + // Check initial state + assertEquals(0, ctx.getAllAvailableReports(scm1).size()); + assertEquals(0, ctx.getAllAvailableReports(scm2).size()); + + Map expectedReportCount = new HashMap<>(); + + // Add a bunch of ContainerReports + batchAddReports(ctx, StateContext.CONTAINER_REPORTS_PROTO_NAME, 128); + // Should only keep the latest one + expectedReportCount.put(StateContext.CONTAINER_REPORTS_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + + // Add a bunch of NodeReport + batchAddReports(ctx, StateContext.NODE_REPORT_PROTO_NAME, 128); + // Should only keep the latest one + expectedReportCount.put(StateContext.NODE_REPORT_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + + // Add a bunch of PipelineReports + batchAddReports(ctx, StateContext.PIPELINE_REPORTS_PROTO_NAME, 128); + // Should only keep the latest one + expectedReportCount.put(StateContext.PIPELINE_REPORTS_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + + // Add a bunch of PipelineReports + batchAddReports(ctx, StateContext.PIPELINE_REPORTS_PROTO_NAME, 128); + // Should only keep the latest one + expectedReportCount.put(StateContext.PIPELINE_REPORTS_PROTO_NAME, 1); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + + // Add a bunch of CommandStatusReports + batchAddReports(ctx, + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME, 128); + expectedReportCount.put( + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME, 128); + // Should keep all of them + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + // getReports dequeues incremental reports + expectedReportCount.remove( + StateContext.COMMAND_STATUS_REPORTS_PROTO_NAME); + + // Add a bunch of IncrementalContainerReport + batchAddReports(ctx, + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 128); + // Should keep all of them + expectedReportCount.put( + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME, 128); + checkReportCount(ctx.getAllAvailableReports(scm1), expectedReportCount); + checkReportCount(ctx.getAllAvailableReports(scm2), expectedReportCount); + // getReports dequeues incremental reports + expectedReportCount.remove( + StateContext.INCREMENTAL_CONTAINER_REPORT_PROTO_NAME); + } + + void batchAddReports(StateContext ctx, String reportName, int count) { + for (int i = 0; i < count; i++) { + ctx.addReport(newMockReport(reportName)); + } + } + + void checkReportCount(List reports, + Map expectedReportCount) { + Map reportCount = new HashMap<>(); + for (GeneratedMessage report : reports) { + final String reportName = report.getDescriptorForType().getFullName(); + reportCount.put(reportName, reportCount.getOrDefault(reportName, 0) + 1); + } + // Verify + assertEquals(expectedReportCount, reportCount); + } + + /** + * Check if Container, Node and Pipeline report APIs work as expected. + */ + @Test + public void testContainerNodePipelineReportAPIs() { + OzoneConfiguration conf = new OzoneConfiguration(); + DatanodeStateMachine datanodeStateMachineMock = + mock(DatanodeStateMachine.class); + + // ContainerReports + StateContext context1 = newStateContext(conf, datanodeStateMachineMock); + assertNull(context1.getContainerReports()); + assertNull(context1.getNodeReport()); + assertNull(context1.getPipelineReports()); + GeneratedMessage containerReports = + newMockReport(StateContext.CONTAINER_REPORTS_PROTO_NAME); + context1.addReport(containerReports); + + assertNotNull(context1.getContainerReports()); + assertEquals(StateContext.CONTAINER_REPORTS_PROTO_NAME, + context1.getContainerReports().getDescriptorForType().getFullName()); + assertNull(context1.getNodeReport()); + assertNull(context1.getPipelineReports()); + + // NodeReport + StateContext context2 = newStateContext(conf, datanodeStateMachineMock); + GeneratedMessage nodeReport = + newMockReport(StateContext.NODE_REPORT_PROTO_NAME); + context2.addReport(nodeReport); + + assertNull(context2.getContainerReports()); + assertNotNull(context2.getNodeReport()); + assertEquals(StateContext.NODE_REPORT_PROTO_NAME, + context2.getNodeReport().getDescriptorForType().getFullName()); + assertNull(context2.getPipelineReports()); + + // PipelineReports + StateContext context3 = newStateContext(conf, datanodeStateMachineMock); + GeneratedMessage pipelineReports = + newMockReport(StateContext.PIPELINE_REPORTS_PROTO_NAME); + context3.addReport(pipelineReports); + + assertNull(context3.getContainerReports()); + assertNull(context3.getNodeReport()); + assertNotNull(context3.getPipelineReports()); + assertEquals(StateContext.PIPELINE_REPORTS_PROTO_NAME, + context3.getPipelineReports().getDescriptorForType().getFullName()); + } + + private StateContext newStateContext(OzoneConfiguration conf, + DatanodeStateMachine datanodeStateMachineMock) { + StateContext stateContext = new StateContext(conf, + DatanodeStates.getInitState(), datanodeStateMachineMock); + InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001); + stateContext.addEndpoint(scm1); + InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001); + stateContext.addEndpoint(scm2); + return stateContext; + } + + private GeneratedMessage newMockReport(String messageType) { + GeneratedMessage pipelineReports = mock(GeneratedMessage.class); + when(pipelineReports.getDescriptorForType()).thenReturn( + mock(Descriptor.class)); + when(pipelineReports.getDescriptorForType().getFullName()).thenReturn( + messageType); + return pipelineReports; + } + @Test public void testReportAPIs() { OzoneConfiguration conf = new OzoneConfiguration(); @@ -64,8 +338,14 @@ public void testReportAPIs() { InetSocketAddress scm1 = new InetSocketAddress("scm1", 9001); InetSocketAddress scm2 = new InetSocketAddress("scm2", 9001); - // Try to add report with endpoint. Should not be stored. - stateContext.addReport(mock(GeneratedMessage.class)); + GeneratedMessage generatedMessage = mock(GeneratedMessage.class); + when(generatedMessage.getDescriptorForType()).thenReturn( + mock(Descriptor.class)); + when(generatedMessage.getDescriptorForType().getFullName()).thenReturn( + "hadoop.hdds.CommandStatusReportsProto"); + + // Try to add report with zero endpoint. Should not be stored. + stateContext.addReport(generatedMessage); assertTrue(stateContext.getAllAvailableReports(scm1).isEmpty()); // Add 2 scm endpoints. @@ -73,7 +353,7 @@ public void testReportAPIs() { stateContext.addEndpoint(scm2); // Add report. Should be added to all endpoints. - stateContext.addReport(mock(GeneratedMessage.class)); + stateContext.addReport(generatedMessage); List allAvailableReports = stateContext.getAllAvailableReports(scm1); assertEquals(1, allAvailableReports.size()); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCreatePipelineCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCreatePipelineCommandHandler.java index febd1c3bd0df..d23f1c49bf13 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCreatePipelineCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCreatePipelineCommandHandler.java @@ -44,6 +44,7 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.Mockito; +import org.mockito.stubbing.Answer; import org.powermock.api.mockito.PowerMockito; import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.modules.junit4.PowerMockRunner; @@ -79,7 +80,10 @@ public void setup() throws Exception { Mockito.when(raftClient.getGroupManagementApi( Mockito.any(RaftPeerId.class))).thenReturn(raftClientGroupManager); PowerMockito.mockStatic(RaftClient.class); - PowerMockito.when(RaftClient.newBuilder()).thenReturn(builder); + // Work around for mockito bug: + // https://github.com/powermock/powermock/issues/992 + PowerMockito.when(RaftClient.newBuilder()).thenAnswer( + (Answer) invocation -> builder); } private RaftClient.Builder mockRaftClientBuilder() { diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java index bee77c7916e5..d248ac1059f5 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestTarContainerPacker.java @@ -23,7 +23,7 @@ import java.io.FileOutputStream; import java.io.FileWriter; import java.io.IOException; -import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; @@ -55,7 +55,6 @@ import org.junit.runner.RunWith; import org.junit.runners.Parameterized; -import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.commons.compress.compressors.CompressorStreamFactory.GZIP; /** @@ -187,7 +186,7 @@ public void pack() throws IOException, CompressorException { //read the container descriptor only try (FileInputStream input = new FileInputStream(targetFile.toFile())) { String containerYaml = new String(packer.unpackContainerDescriptor(input), - Charset.forName(UTF_8.name())); + StandardCharsets.UTF_8); Assert.assertEquals(TEST_DESCRIPTOR_FILE_CONTENT, containerYaml); } @@ -203,7 +202,7 @@ public void pack() throws IOException, CompressorException { try (FileInputStream input = new FileInputStream(targetFile.toFile())) { descriptor = new String(packer.unpackContainerData(destinationContainer, input), - Charset.forName(UTF_8.name())); + StandardCharsets.UTF_8); } assertExampleMetadataDbIsGood( @@ -359,7 +358,7 @@ private void assertExampleMetadataDbIsGood(Path dbPath, String filename) try (FileInputStream testFile = new FileInputStream(dbFile.toFile())) { List strings = IOUtils - .readLines(testFile, Charset.forName(UTF_8.name())); + .readLines(testFile, StandardCharsets.UTF_8); Assert.assertEquals(1, strings.size()); Assert.assertEquals(TEST_DB_FILE_CONTENT, strings.get(0)); } @@ -377,7 +376,7 @@ private void assertExampleChunkFileIsGood(Path chunkPath, String filename) try (FileInputStream testFile = new FileInputStream(chunkFile.toFile())) { List strings = IOUtils - .readLines(testFile, Charset.forName(UTF_8.name())); + .readLines(testFile, StandardCharsets.UTF_8); Assert.assertEquals(1, strings.size()); Assert.assertEquals(TEST_CHUNK_FILE_CONTENT, strings.get(0)); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorScheduling.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorScheduling.java new file mode 100644 index 000000000000..2c517cb123fc --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorScheduling.java @@ -0,0 +1,125 @@ +/* + * 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.replication; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.UUID; + +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.ozone.container.common.impl.ContainerSet; + +import org.junit.Assert; +import org.junit.Test; + +/** + * Helper to check scheduling efficiency. + *

+ * This unit test is not enabled (doesn't start with Test) but can be used + * to validate changes manually. + */ +public class ReplicationSupervisorScheduling { + + private final Random random = new Random(); + + @Test + public void test() throws InterruptedException { + List datanodes = new ArrayList<>(); + datanodes.add(MockDatanodeDetails.randomDatanodeDetails()); + datanodes.add(MockDatanodeDetails.randomDatanodeDetails()); + + //locks representing the limited resource of remote and local disks + + //datanode -> disk -> lock object (remote resources) + Map> volumeLocks = new HashMap<>(); + + //disk -> lock (local resources) + Map destinationLocks = new HashMap<>(); + + //init the locks + for (DatanodeDetails datanode : datanodes) { + volumeLocks.put(datanode.getUuid(), new HashMap<>()); + for (int i = 0; i < 10; i++) { + volumeLocks.get(datanode.getUuid()).put(i, new Object()); + } + } + + for (int i = 0; i < 10; i++) { + destinationLocks.put(i, new Object()); + } + + ContainerSet cs = new ContainerSet(); + + ReplicationSupervisor rs = new ReplicationSupervisor(cs, + + //simplified executor emulating the current sequential download + + //import. + task -> { + + //download, limited by the number of source datanodes + final DatanodeDetails sourceDatanode = + task.getSources().get(random.nextInt(task.getSources().size())); + + final Map volumes = + volumeLocks.get(sourceDatanode.getUuid()); + synchronized (volumes.get(random.nextInt(volumes.size()))) { + System.out.println("Downloading " + task.getContainerId() + " from " + + sourceDatanode.getUuid()); + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + ex.printStackTrace(); + } + } + + //import, limited by the destination datanode + final int volumeIndex = random.nextInt(destinationLocks.size()); + synchronized (destinationLocks.get(volumeIndex)) { + System.out.println( + "Importing " + task.getContainerId() + " to disk " + + volumeIndex); + + try { + Thread.sleep(1000); + } catch (InterruptedException ex) { + ex.printStackTrace(); + } + } + + }, 10); + + final long start = System.currentTimeMillis(); + + //schedule 100 container replication + for (int i = 0; i < 100; i++) { + List sources = new ArrayList<>(); + sources.add(datanodes.get(random.nextInt(datanodes.size()))); + rs.addTask(new ReplicationTask(i, sources)); + } + rs.shutdownAfterFinish(); + final long executionTime = System.currentTimeMillis() - start; + System.out.println(executionTime); + Assert.assertTrue("Execution was too slow : " + executionTime + " ms", + executionTime < 100_000); + } + +} diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java index 3f0f5a9faaeb..f29b1579198a 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestSimpleContainerDownloader.java @@ -21,10 +21,13 @@ import java.nio.file.Path; import java.nio.file.Paths; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; +import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; @@ -32,17 +35,74 @@ import org.junit.Assert; import org.junit.Test; -/** - * Test container downloader. +/* + * Test SimpleContainerDownloader. */ public class TestSimpleContainerDownloader { private static final String SUCCESS_PATH = "downloaded"; + @Test + public void testGetContainerDataFromReplicasHappyPath() throws Exception { + + //GIVEN + List datanodes = createDatanodes(); + + SimpleContainerDownloader downloader = + createDownloaderWithPredefinedFailures(true); + + //WHEN + final Path result = + downloader.getContainerDataFromReplicas(1L, datanodes) + .get(1L, TimeUnit.SECONDS); + + //THEN + Assert.assertEquals(datanodes.get(0).getUuidString(), result.toString()); + } + + @Test + public void testGetContainerDataFromReplicasDirectFailure() + throws Exception { + + //GIVEN + List datanodes = createDatanodes(); + + SimpleContainerDownloader downloader = + createDownloaderWithPredefinedFailures(true, datanodes.get(0)); + + //WHEN + final Path result = + downloader.getContainerDataFromReplicas(1L, datanodes) + .get(1L, TimeUnit.SECONDS); + + //THEN + //first datanode is failed, second worked + Assert.assertEquals(datanodes.get(1).getUuidString(), result.toString()); + } + + @Test + public void testGetContainerDataFromReplicasAsyncFailure() throws Exception { + + //GIVEN + List datanodes = createDatanodes(); + + SimpleContainerDownloader downloader = + createDownloaderWithPredefinedFailures(false, datanodes.get(0)); + + //WHEN + final Path result = + downloader.getContainerDataFromReplicas(1L, datanodes) + .get(1L, TimeUnit.SECONDS); + + //THEN + //first datanode is failed, second worked + Assert.assertEquals(datanodes.get(1).getUuidString(), result.toString()); + } + /** * Test if different datanode is used for each download attempt. */ - @Test(timeout = 1000L) + @Test(timeout = 10_000L) public void testRandomSelection() throws ExecutionException, InterruptedException { @@ -62,24 +122,79 @@ protected CompletableFuture downloadContainer( } }; - //WHEN executed, THEN at least once the second datanode should be returned. + //WHEN executed, THEN at least once the second datanode should be + //returned. for (int i = 0; i < 10000; i++) { - Path path = downloader.getContainerDataFromReplicas(1L, datanodes).get(); + Path path = + downloader.getContainerDataFromReplicas(1L, datanodes).get(); if (path.toString().equals(datanodes.get(1).getUuidString())) { return; } } - //there is 1/2^10_000 chance for false positive, which is practically 0. + //there is 1/3^10_000 chance for false positive, which is practically 0. Assert.fail( "Datanodes are selected 10000 times but second datanode was never " + "used."); } + /** + * Creates downloader which fails with datanodes in the arguments. + * + * @param directException if false the exception will be wrapped in the + * returning future. + */ + private SimpleContainerDownloader createDownloaderWithPredefinedFailures( + boolean directException, + DatanodeDetails... failedDatanodes + ) { + + ConfigurationSource conf = new OzoneConfiguration(); + + final List datanodes = + Arrays.asList(failedDatanodes); + + return new SimpleContainerDownloader(conf, null) { + + //for retry testing we use predictable list of datanodes. + @Override + protected List shuffleDatanodes( + List sourceDatanodes + ) { + //turn off randomization + return sourceDatanodes; + } + + @Override + protected CompletableFuture downloadContainer( + long containerId, + DatanodeDetails datanode + ) throws Exception { + + if (datanodes.contains(datanode)) { + if (directException) { + throw new RuntimeException("Unavailable datanode"); + } else { + return CompletableFuture.supplyAsync(() -> { + throw new RuntimeException("Unavailable datanode"); + }); + } + } else { + + //path includes the dn id to make it possible to assert. + return CompletableFuture.completedFuture( + Paths.get(datanode.getUuidString())); + } + + } + }; + } + private List createDatanodes() { List datanodes = new ArrayList<>(); datanodes.add(MockDatanodeDetails.randomDatanodeDetails()); datanodes.add(MockDatanodeDetails.randomDatanodeDetails()); + datanodes.add(MockDatanodeDetails.randomDatanodeDetails()); return datanodes; } } \ No newline at end of file diff --git a/hadoop-hdds/container-service/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/hadoop-hdds/container-service/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker new file mode 100644 index 000000000000..3c9e1c8a6971 --- /dev/null +++ b/hadoop-hdds/container-service/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker @@ -0,0 +1,16 @@ +# 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. +mock-maker-inline \ No newline at end of file diff --git a/hadoop-hdds/docs/README.md b/hadoop-hdds/docs/README.md index 8d5cdb714fd7..c5c91677016a 100644 --- a/hadoop-hdds/docs/README.md +++ b/hadoop-hdds/docs/README.md @@ -14,7 +14,7 @@ See the License for the specific language governing permissions and limitations under the License. --> -# Hadoop Ozone/HDDS docs +# Apache Ozone/HDDS docs This subproject contains the inline documentation for Ozone/HDDS components. diff --git a/hadoop-hdds/docs/content/_index.md b/hadoop-hdds/docs/content/_index.md index 7890f6f16580..be0b30303e97 100644 --- a/hadoop-hdds/docs/content/_index.md +++ b/hadoop-hdds/docs/content/_index.md @@ -21,7 +21,7 @@ weight: -10 limitations under the License. --> -# Apache Hadoop Ozone +# Apache Ozone {{

}} diff --git a/hadoop-hdds/docs/content/_index.zh.md b/hadoop-hdds/docs/content/_index.zh.md index 689490be11ad..57011d10261c 100644 --- a/hadoop-hdds/docs/content/_index.zh.md +++ b/hadoop-hdds/docs/content/_index.zh.md @@ -20,7 +20,7 @@ weight: -10 limitations under the License. --> -# Apache Hadoop Ozone +# Apache Ozone {{
}} @@ -29,7 +29,7 @@ Ozone 不仅能存储数十亿个不同大小的对象，还支持在容器化 Apache Spark、Hive 和 YARN 等应用无需任何修改即可使用 Ozone。Ozone 提供了 [Java API]({{< ref "JavaApi.zh.md" >}})、[S3 接口]({{< ref "S3.zh.md" >}})和命令行接口，极大地方便了 Ozone - 在不同应用场景下的的使用。 + 在不同应用场景下的使用。 Ozone 的管理由卷、桶和键组成： diff --git a/hadoop-hdds/docs/content/recipe/Prometheus.md b/hadoop-hdds/docs/content/recipe/Prometheus.md index f63b46e8b03e..9c852e0b80bf 100644 --- a/hadoop-hdds/docs/content/recipe/Prometheus.md +++ b/hadoop-hdds/docs/content/recipe/Prometheus.md @@ -46,9 +46,9 @@ _Note_: for Docker compose based pseudo cluster put the \ * Restart the Ozone Manager and Storage Container Manager and check the prometheus endpoints: - * http://scm:9874/prom + * http://scm:9876/prom - * http://ozoneManager:9876/prom + * http://ozoneManager:9874/prom * Create a prometheus.yaml configuration with the previous endpoints: @@ -93,4 +93,4 @@ The ozone distribution contains a ready-to-use, dockerized environment to try ou cd compose/ozone export COMPOSE_FILE=docker-compose.yaml:monitoring.yaml docker-compose up -d -``` \ No newline at end of file +``` diff --git a/hadoop-hdds/docs/content/recipe/Prometheus.zh.md b/hadoop-hdds/docs/content/recipe/Prometheus.zh.md index 069b3403660a..bb64edc89885 100644 --- a/hadoop-hdds/docs/content/recipe/Prometheus.zh.md +++ b/hadoop-hdds/docs/content/recipe/Prometheus.zh.md @@ -44,9 +44,9 @@ _注意_: 对于基于 docker-compose 方式的伪集群，在 `docker-config` * 重启 OM 和 SCM，检查端点： - * http://scm:9874/prom + * http://scm:9876/prom - * http://ozoneManager:9876/prom + * http://ozoneManager:9874/prom * 根据这两个端点，创建 prometheus.yaml 配置文件： @@ -91,4 +91,4 @@ Ozone 发行包中包含了一个即开即用的容器化环境来试用 Ozone cd compose/ozone export COMPOSE_FILE=docker-compose.yaml:monitoring.yaml docker-compose up -d -``` \ No newline at end of file +``` diff --git a/hadoop-hdds/docs/content/security/SecurityWithRanger.md b/hadoop-hdds/docs/content/security/SecurityWithRanger.md index 7daaf8199e18..ee86a115930b 100644 --- a/hadoop-hdds/docs/content/security/SecurityWithRanger.md +++ b/hadoop-hdds/docs/content/security/SecurityWithRanger.md @@ -27,8 +27,9 @@ icon: user Apache Ranger™ is a framework to enable, monitor and manage comprehensive data -security across the Hadoop platform. Any version of Apache Ranger which is greater -than 1.20 is aware of Ozone, and can manage an Ozone cluster. +security across the Hadoop platform. Apache Ranger has supported Ozone authentication +since version 2.0. However, due to some bugs in 2.0, Apache Ranger +2.1 and later versions are recommended. To use Apache Ranger, you must have Apache Ranger installed in your Hadoop @@ -44,3 +45,19 @@ Property|Value --------|------------------------------------------------------------ ozone.acl.enabled | true ozone.acl.authorizer.class| org.apache.ranger.authorization.ozone.authorizer.RangerOzoneAuthorizer + +The Ranger permissions corresponding to the Ozone operations are as follows: + +| operation&permission | Volume permission | Bucket permission | Key permission | +| :--- | :--- | :--- | :--- | +| Create volume | CREATE | | | +| List volume | LIST | | | +| Get volume Info | READ | | | +| Delete volume | DELETE | | | +| Create bucket | READ | CREATE | | +| List bucket | LIST, READ | | | +| Get bucket info | READ | READ | | +| Delete bucket | READ | DELETE | | +| List key | READ | LIST, READ | | +| Write key | READ | READ | CREATE, WRITE | +| Read key | READ | READ | READ | diff --git a/hadoop-hdds/docs/content/security/SecurityWithRanger.zh.md b/hadoop-hdds/docs/content/security/SecurityWithRanger.zh.md index 4d40a17d0188..e7ff33e9268f 100644 --- a/hadoop-hdds/docs/content/security/SecurityWithRanger.zh.md +++ b/hadoop-hdds/docs/content/security/SecurityWithRanger.zh.md @@ -26,7 +26,7 @@ icon: user --> -Apache Ranger™ 是一个用于管理和监控 Hadoop 平台复杂数据权限的框架。版本大于 1.20 的 Apache Ranger 都可以用于管理 Ozone 集群。 +Apache Ranger™ 是一个用于管理和监控 Hadoop 平台复杂数据权限的框架。Apache Ranger 从2.0版本开始支持Ozone鉴权。但由于在2.0中存在一些bug，因此我们更推荐使用Apache Ranger 2.1及以后版本。 你需要先在你的 Hadoop 集群上安装 Apache Ranger，安装指南可以参考 [Apache Ranger 官网](https://ranger.apache.org/index.html). @@ -36,3 +36,19 @@ Apache Ranger™ 是一个用于管理和监控 Hadoop 平台复杂数据权限 --------|------------------------------------------------------------ ozone.acl.enabled | true ozone.acl.authorizer.class| org.apache.ranger.authorization.ozone.authorizer.RangerOzoneAuthorizer + +Ozone各类操作对应Ranger权限如下： + +| operation&permission | Volume permission | Bucket permission | Key permission | +| :--- | :--- | :--- | :--- | +| Create volume | CREATE | | | +| List volume | LIST | | | +| Get volume Info | READ | | | +| Delete volume | DELETE | | | +| Create bucket | READ | CREATE | | +| List bucket | LIST, READ | | | +| Get bucket info | READ | READ | | +| Delete bucket | READ | DELETE | | +| List key | READ | LIST, READ | | +| Write key | READ | READ | CREATE, WRITE | +| Read key | READ | READ | READ | \ No newline at end of file diff --git a/hadoop-hdds/docs/pom.xml b/hadoop-hdds/docs/pom.xml index 404b6c2a253b..3a6aea0aa6c0 100644 --- a/hadoop-hdds/docs/pom.xml +++ b/hadoop-hdds/docs/pom.xml @@ -24,8 +24,8 @@ https://maven.apache.org/xsd/maven-4.0.0.xsd"> hadoop-hdds-docs 1.1.0-SNAPSHOT - Apache Hadoop HDDS/Ozone Documentation - Apache Hadoop HDDS/Ozone Documentation + Apache Ozone/HDDS Documentation + Apache Ozone/HDDS Documentation jar diff --git a/hadoop-hdds/docs/static/ozone-logo-monochrome.svg b/hadoop-hdds/docs/static/ozone-logo-monochrome.svg index cd046a00b7cf..89cc166afe79 100644 --- a/hadoop-hdds/docs/static/ozone-logo-monochrome.svg +++ b/hadoop-hdds/docs/static/ozone-logo-monochrome.svg @@ -28,7 +28,7 @@ sodipodi:docname="ozone_bolt.svg" inkscape:version="0.92.4 (5da689c313, 2019-01-14)"> Apache Hadoop Ozone Logo + id="title39">Apache Ozone Logo @@ -37,7 +37,7 @@ image/svg+xml - Apache Hadoop Ozone Logo + Apache Ozone Logo diff --git a/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/header.html b/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/header.html index a4e24c951330..8f475b61f8d1 100644 --- a/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/header.html +++ b/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/header.html @@ -21,9 +21,9 @@ - + - Documentation for Apache Hadoop Ozone + Documentation for Apache Ozone diff --git a/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/navbar.html b/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/navbar.html index f942e4a1d25d..d4c9f2e994de 100644 --- a/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/navbar.html +++ b/hadoop-hdds/docs/themes/ozonedoc/layouts/partials/navbar.html @@ -27,9 +27,9 @@ - Hadoop Ozone + Apache Ozone