You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by sa...@apache.org on 2020/01/08 03:58:03 UTC
[hadoop-ozone] branch HDDS-1564 updated (53afa84 -> cc0060c)
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a change to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git.
omit 53afa84 Resolve rebase conflict.
omit 09bb190 HDDS-1574 Average out pipeline allocation on datanodes and add metrcs/test (#291)
omit f842691 HDDS-2650 Fix createPipeline CLI and make it message based. (#370)
omit 7604c3c Revert "HDDS-2650 Fix createPipeline CLI. (#340)"
omit 8c07782 HDDS-2035 Implement datanode level CLI to reveal pipeline relation. (#348)
omit baf58d5 HDDS-2650 Fix createPipeline CLI. (#340)
omit f27605c Rebase Fix
omit 5525b98 HDDS-1572 Implement a Pipeline scrubber to clean up non-OPEN pipeline. (#237)
omit da464bc HDDS-1569 Support creating multiple pipelines with same datanode. Contributed by Li Cheng.
omit c3f450e HDDS-2089: Add createPipeline CLI. (#1418)
omit 3d2c2e4 HDDS-1571. Create an interface for pipeline placement policy to support network topologies. (#1395)
omit 7c752f4 HDDS-1577. Add default pipeline placement policy implementation. (#1366)
add 2f9db71 HDDS-2788. Fix the wrong source link in docs (#388)
add c658374 HDDS-2705. Use Ozone specific LICENSE and NOTICE in the root of the git repo (#335)
add 092b1eb HDDS-2787. Use try-with-resources or close resource in finally block (#387)
add 529438a HDDS-2805. Fix sudo reset the environment variables about proxy (#392)
add 42bd150 HDDS-2812. Fix low version wget cannot resolve the proxy of https (#395)
add 87f69cb HDDS-2504. Handle InterruptedException properly (#386)
add e5559f7 HDDS-2552. Sonar: Save and reuse Random object in GenesisUtil (#404)
add 28e370e HDDS-2830. Remove unnecessary sleep in acceptance test. (#409)
add 8942888 HDDS-2806. Remove the LongCodec in the metadata package, and redundancy addCodec for Long class. (#391)
add 41910d7 HDDS-2679. Ratis ring creation might be failed with async pipeline creation. (#363)
add 7fb46e1 HDDS-2784. Read to ByteBuffer uses wrong offset (#406)
add c2c2c26 HDDS-2820. OM Ratis dir creation may fail (#403)
add f2b0651 HDDS-2813. Provide a ByteArrayCodec to CodecRegistry. (#394)
add f1a8f8f HDDS-2773. Configure Goldmark renderer
add 5f6ce75 HDDS-2814. Fail to connect s3g in docker container with network proxy
add edc9349 HDDS-2737. Insight point should provide service type to get correct config
add 3f92632 HDDS-2661. Use field based Config annotation instead of method based
add 28cefc6 HDDS-2785. Move ozonefs and tools minicluster tests to integration-test
add 9372779 HDDS-2690. Improve the command usage of audit parser tool
add 46bfd43 HDDS-2829. Deduplicate KDC docker image definition (#407)
add 7a99050 HDDS-2719. Rename S3Token to S3AuthInfo (#349)
add f57ef24 HDDS-2712. _index.md translated to Chinese (#365)
add 573ce21 HDDS-2815. Fix shell description for --start parameter of listing buckets (#402)
add 6baa1a3 HDDS-2484 New Metrics for Trash Key Lists and Fails (#408)
add e297cf9 HDDS-2826. Add initial UI of Dashboard for Recon
add 37a01da HDDS-2835. update ozone to latest ratis. (#412)
add cbf2a53 HDDS-2721. Allow wrapping list of ByteBuffers with ChunkBuffer (#378)
add cacf3d1 HDDS-2651 Make startId parameter non-mandatory while listing containers through shell command (#283)
new f1d994e HDDS-1577. Add default pipeline placement policy implementation. (#1366)
new 7fe5881 HDDS-1571. Create an interface for pipeline placement policy to support network topologies. (#1395)
new f9fe003 HDDS-2089: Add createPipeline CLI. (#1418)
new f8c5eb0 HDDS-1569 Support creating multiple pipelines with same datanode. Contributed by Li Cheng.
new a6c1f2a HDDS-1572 Implement a Pipeline scrubber to clean up non-OPEN pipeline. (#237)
new 667a0c0 Rebase Fix
new a2c65a5 HDDS-2650 Fix createPipeline CLI. (#340)
new 3796154 HDDS-2035 Implement datanode level CLI to reveal pipeline relation. (#348)
new 00f0f81 Revert "HDDS-2650 Fix createPipeline CLI. (#340)"
new fced9c1 HDDS-2650 Fix createPipeline CLI and make it message based. (#370)
new acc9640 HDDS-1574 Average out pipeline allocation on datanodes and add metrcs/test (#291)
new cc0060c Resolve rebase conflict.
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (53afa84)
\
N -- N -- N refs/heads/HDDS-1564 (cc0060c)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
LICENSE.txt | 53 +-
NOTICE.txt | 5 +-
.../hadoop/hdds/scm/XceiverClientManager.java | 60 +-
.../hadoop/hdds/scm/storage/CommitWatcher.java | 23 +-
.../hadoop/hdds/conf/OzoneConfiguration.java | 103 +-
.../org/apache/hadoop/hdds/ratis/RatisHelper.java | 13 +
.../java/org/apache/hadoop/hdds/scm/ScmConfig.java | 30 +-
.../apache/hadoop/hdds/scm/XceiverClientSpi.java | 23 +-
.../hadoop/hdds/security/x509/SecurityConfig.java | 8 +-
.../hadoop/hdds/utils/BackgroundService.java | 10 +-
.../org/apache/hadoop/hdds/utils/Scheduler.java | 31 +-
...rNotFoundException.java => ByteArrayCodec.java} | 21 +-
.../apache/hadoop/hdds/utils/db/CodecRegistry.java | 1 +
.../hadoop/hdds/utils/db/RocksDBConfiguration.java | 35 +-
.../org/apache/hadoop/ozone/OzoneConfigKeys.java | 4 +-
.../apache/hadoop/ozone/common/ChunkBuffer.java | 8 +
.../common/ChunkBufferImplWithByteBuffer.java | 7 +
.../common/ChunkBufferImplWithByteBufferList.java | 215 ++
.../ozone/common/IncrementalChunkBuffer.java | 5 +
.../hadoop/hdds/conf/SimpleConfiguration.java | 45 +-
.../hdds/conf/SimpleConfigurationParent.java} | 22 +-
.../hadoop/hdds/conf/TestOzoneConfiguration.java | 9 +
.../hdds/utils/db/TestTypedRDBTableStore.java | 20 +-
.../hadoop/ozone/common/TestChunkBuffer.java | 61 +-
.../TestChunkBufferImplWithByteBufferList.java | 80 +
.../java/org/apache/hadoop/hdds/conf/Config.java | 2 +-
.../hadoop/hdds/conf/ConfigFileGenerator.java | 77 +-
.../hadoop/hdds/conf/ConfigurationExample.java | 26 +-
.../conf/ConfigurationExampleGrandParent.java} | 18 +-
.../hdds/conf/ConfigurationExampleParent.java} | 19 +-
.../hadoop/hdds/conf/TestConfigFileGenerator.java | 60 +
hadoop-hdds/container-service/pom.xml | 16 +
.../common/statemachine/DatanodeConfiguration.java | 65 +-
.../common/statemachine/DatanodeStateMachine.java | 2 +-
.../CreatePipelineCommandHandler.java | 46 +-
.../server/ratis/RatisServerConfiguration.java | 4 +-
.../transport/server/ratis/XceiverServerRatis.java | 11 +-
.../ozone/container/keyvalue/KeyValueHandler.java | 5 +-
.../ozoneimpl/ContainerScrubberConfiguration.java | 58 +-
.../TestCreatePipelineCommandHandler.java | 159 ++
hadoop-hdds/docs/config.yaml | 4 +
hadoop-hdds/docs/content/_index.md | 2 +-
hadoop-hdds/docs/content/_index.zh.md | 18 +-
hadoop-hdds/docs/content/beyond/Containers.md | 4 +-
hadoop-hdds/docs/content/concept/Overview.md | 2 +-
.../Datanodes.zh.md => interface/JavaApi.zh.md} | 8 +-
.../{concept/_index.zh.md => interface/S3.zh.md} | 11 +-
hadoop-hdds/docs/content/security/SecureOzone.md | 6 +-
hadoop-hdds/docs/content/{ => shell}/_index.zh.md | 10 +-
.../themes/ozonedoc/layouts/partials/navbar.html | 2 +-
.../org/apache/hadoop/hdds/server/ServerUtils.java | 2 +-
.../hdds/scm/container/ReplicationManager.java | 24 +-
.../apache/hadoop/hdds/scm/metadata/LongCodec.java | 45 -
.../hdds/scm/metadata/SCMMetadataStoreRDBImpl.java | 1 -
.../hdds/scm/pipeline/RatisPipelineProvider.java | 11 +-
.../hdds/scm/server/SCMHTTPServerConfig.java | 21 +-
.../hdds/scm/cli/container/ListSubcommand.java | 4 +-
.../main/java/org/apache/hadoop/ozone/OmUtils.java | 2 +-
.../org/apache/hadoop/ozone/audit/OMAction.java | 1 +
.../OzoneDelegationTokenSecretManager.java | 10 +-
.../ozone/security/OzoneTokenIdentifier.java | 8 +-
.../hadoop/ozone/security/acl/OzoneAclConfig.java | 25 +-
.../src/main/proto/OzoneManagerProtocol.proto | 2 +-
.../java/org/apache/hadoop/ozone/TestOmUtils.java | 35 +
.../org/apache/hadoop/ozone/csi/CsiServer.java | 57 +-
hadoop-ozone/dev-support/checks/integration.sh | 2 +-
hadoop-ozone/dev-support/checks/unit.sh | 2 +-
.../dist/src/main/assemblies/ozone-src.xml | 4 +-
.../docker-image/docker-krb5/Dockerfile-krb5 | 2 +-
.../docker-image/docker-krb5/README.md | 0
.../docker-image/docker-krb5/kadm5.acl | 0
.../docker-image/docker-krb5/krb5.conf | 0
.../docker-image/docker-krb5/launcher.sh | 0
.../dist/src/main/compose/ozone-hdfs/docker-config | 2 +
.../main/compose/ozone-mr/hadoop27/docker-config | 2 +
.../src/main/compose/ozone-mr/hadoop27/test.sh | 4 +-
.../main/compose/ozone-mr/hadoop31/docker-config | 2 +
.../src/main/compose/ozone-mr/hadoop31/test.sh | 2 +-
.../main/compose/ozone-mr/hadoop32/docker-config | 2 +
.../src/main/compose/ozone-om-ha-s3/docker-config | 2 +
.../src/main/compose/ozone-om-ha/docker-config | 2 +
.../src/main/compose/ozone-topology/docker-config | 2 +
.../dist/src/main/compose/ozone/docker-config | 2 +
.../src/main/compose/ozoneblockade/docker-config | 2 +
.../src/main/compose/ozones3-haproxy/docker-config | 2 +
.../src/main/compose/ozonescripts/docker-config | 2 +
.../compose/ozonesecure-mr/docker-compose.yaml | 5 +-
.../src/main/compose/ozonesecure-mr/docker-config | 2 +
.../docker-image/docker-krb5/kadm5.acl | 20 -
.../docker-image/docker-krb5/krb5.conf | 41 -
.../docker-image/docker-krb5/launcher.sh | 25 -
.../main/compose/ozonesecure/docker-compose.yaml | 5 +-
.../src/main/compose/ozonesecure/docker-config | 2 +
.../docker-image/docker-krb5/Dockerfile-krb5 | 34 -
.../ozonesecure/docker-image/docker-krb5/README.md | 34 -
hadoop-ozone/dist/src/main/compose/testlib.sh | 11 +-
hadoop-ozone/dist/src/main/license/src/LICENSE.txt | 239 --
hadoop-ozone/dist/src/main/license/src/NOTICE.txt | 33 -
.../dist/src/main/smoketest/s3/commonawslib.robot | 4 +-
.../ozone/insight/ConfigurationSubCommand.java | 10 +-
hadoop-ozone/integration-test/pom.xml | 42 +
.../hadoop/fs/ozone/TestOzoneFSInputStream.java | 0
.../hadoop/fs/ozone/TestOzoneFileInterfaces.java | 0
.../hadoop/fs/ozone/TestOzoneFileSystem.java | 0
.../apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java | 0
.../hadoop/fs/ozone/TestOzoneFsRenameDir.java | 0
.../ozone/contract/ITestOzoneContractCreate.java | 0
.../ozone/contract/ITestOzoneContractDelete.java | 0
.../ozone/contract/ITestOzoneContractDistCp.java | 0
.../contract/ITestOzoneContractGetFileStatus.java | 0
.../fs/ozone/contract/ITestOzoneContractMkdir.java | 0
.../fs/ozone/contract/ITestOzoneContractOpen.java | 0
.../ozone/contract/ITestOzoneContractRename.java | 0
.../ozone/contract/ITestOzoneContractRootDir.java | 0
.../fs/ozone/contract/ITestOzoneContractSeek.java | 0
.../hadoop/fs/ozone/contract/OzoneContract.java | 0
.../org/apache/hadoop/fs/ozone/package-info.java | 0
.../hadoop/ozone/TestOzoneConfigurationFields.java | 2 +-
.../hadoop/ozone/freon/TestDataValidate.java | 0
.../freon/TestDataValidateWithDummyContainers.java | 0
.../TestDataValidateWithSafeByteOperations.java | 0
.../TestDataValidateWithUnsafeByteOperations.java | 0
.../freon/TestFreonWithDatanodeFastRestart.java | 0
.../ozone/freon/TestFreonWithDatanodeRestart.java | 0
.../ozone/freon/TestFreonWithPipelineDestroy.java | 0
.../hadoop/ozone/freon/TestRandomKeyGenerator.java | 0
.../hadoop/ozone/fsck/TestContainerMapper.java | 0
.../org/apache/hadoop/ozone/om/TestOmMetrics.java | 16 +-
.../org/apache/hadoop/ozone/om/TestOmSQLCli.java | 0
.../src/test/resources/contract/ozone.xml | 0
.../java/org/apache/hadoop/ozone/om/OMMetrics.java | 20 +
.../org/apache/hadoop/ozone/om/OzoneManager.java | 8 +-
.../web/ozShell/bucket/ListBucketHandler.java | 3 +-
.../TestOzoneDelegationTokenSecretManager.java | 10 +-
hadoop-ozone/ozonefs/pom.xml | 47 -
.../fs/ozone/BasicOzoneClientAdapterImpl.java | 4 +-
.../hadoop/fs/ozone/OzoneClientAdapterFactory.java | 1 -
.../apache/hadoop/fs/ozone/OzoneFSInputStream.java | 3 +-
.../ozone/contract/TestOzoneFSInputStreamUnit.java | 101 +
.../recon/spi/impl/ReconContainerDBProvider.java | 2 -
.../webapps/recon/ozone-recon-web/LICENSE | 2649 +++++++++++++-------
.../webapps/recon/ozone-recon-web/README.md | 8 +
.../webapps/recon/ozone-recon-web/api/db.json | 54 +
.../webapps/recon/ozone-recon-web/api/routes.json | 3 +
.../webapps/recon/ozone-recon-web/package.json | 15 +-
.../recon/ozone-recon-web/public/index.html | 12 +-
.../webapps/recon/ozone-recon-web/src/App.tsx | 64 +-
.../src/components/Breadcrumbs/Breadcrumbs.tsx | 12 +-
.../src/components/NavBar/NavBar.tsx | 28 +-
.../components/OverviewCard/OverviewCard.less} | 33 +-
.../src/components/OverviewCard/OverviewCard.tsx | 94 +
.../src/constants/breadcrumbs.constants.tsx | 5 +-
.../webapps/recon/ozone-recon-web/src/routes.tsx | 18 +-
.../src/views/Dashboard/Dashboard.tsx | 32 -
.../Datanodes.tsx} | 7 +-
.../src/views/NotFound/NotFound.tsx | 2 +-
.../src/views/Overview/Overview.less | 4 +-
.../src/views/Overview/Overview.tsx | 111 +
.../webapps/recon/ozone-recon-web/yarn.lock | 1083 ++++++--
.../apache/hadoop/ozone/s3/AWSV4AuthParser.java | 8 +-
.../hadoop/ozone/s3/OzoneClientProducer.java | 12 +-
.../hadoop/ozone/s3/exception/S3ErrorTable.java | 4 +-
hadoop-ozone/tools/pom.xml | 6 -
.../hadoop/ozone/audit/parser/AuditParser.java | 7 +-
.../audit/parser/handler/LoadCommandHandler.java | 4 +-
.../audit/parser/handler/QueryCommandHandler.java | 4 +-
.../parser/handler/TemplateCommandHandler.java | 9 +-
.../hadoop/ozone/freon/DatanodeChunkGenerator.java | 68 +-
.../hadoop/ozone/freon/RandomKeyGenerator.java | 58 +-
.../apache/hadoop/ozone/genesis/GenesisUtil.java | 22 +-
pom.xml | 2 +-
171 files changed, 4601 insertions(+), 2374 deletions(-)
copy hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/utils/db/{SequenceNumberNotFoundException.java => ByteArrayCodec.java} (71%)
create mode 100644 hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBufferList.java
copy hadoop-hdds/{config/src/main/java/org/apache/hadoop/hdds/conf/ConfigType.java => common/src/test/java/org/apache/hadoop/hdds/conf/SimpleConfigurationParent.java} (73%)
create mode 100644 hadoop-hdds/common/src/test/java/org/apache/hadoop/ozone/common/TestChunkBufferImplWithByteBufferList.java
copy hadoop-hdds/config/src/{main/java/org/apache/hadoop/hdds/conf/ConfigurationException.java => test/java/org/apache/hadoop/hdds/conf/ConfigurationExampleGrandParent.java} (69%)
copy hadoop-hdds/config/src/{main/java/org/apache/hadoop/hdds/conf/ConfigurationException.java => test/java/org/apache/hadoop/hdds/conf/ConfigurationExampleParent.java} (66%)
create mode 100644 hadoop-hdds/config/src/test/java/org/apache/hadoop/hdds/conf/TestConfigFileGenerator.java
create mode 100644 hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCreatePipelineCommandHandler.java
copy hadoop-hdds/docs/content/{concept/Datanodes.zh.md => interface/JavaApi.zh.md} (76%)
copy hadoop-hdds/docs/content/{concept/_index.zh.md => interface/S3.zh.md} (78%)
copy hadoop-hdds/docs/content/{ => shell}/_index.zh.md (90%)
delete mode 100644 hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/LongCodec.java
rename hadoop-ozone/dist/src/main/compose/{ozonesecure-mr => common}/docker-image/docker-krb5/Dockerfile-krb5 (96%)
rename hadoop-ozone/dist/src/main/compose/{ozonesecure-mr => common}/docker-image/docker-krb5/README.md (100%)
rename hadoop-ozone/dist/src/main/compose/{ozonesecure => common}/docker-image/docker-krb5/kadm5.acl (100%)
rename hadoop-ozone/dist/src/main/compose/{ozonesecure => common}/docker-image/docker-krb5/krb5.conf (100%)
rename hadoop-ozone/dist/src/main/compose/{ozonesecure => common}/docker-image/docker-krb5/launcher.sh (100%)
delete mode 100644 hadoop-ozone/dist/src/main/compose/ozonesecure-mr/docker-image/docker-krb5/kadm5.acl
delete mode 100644 hadoop-ozone/dist/src/main/compose/ozonesecure-mr/docker-image/docker-krb5/krb5.conf
delete mode 100644 hadoop-ozone/dist/src/main/compose/ozonesecure-mr/docker-image/docker-krb5/launcher.sh
delete mode 100644 hadoop-ozone/dist/src/main/compose/ozonesecure/docker-image/docker-krb5/Dockerfile-krb5
delete mode 100644 hadoop-ozone/dist/src/main/compose/ozonesecure/docker-image/docker-krb5/README.md
delete mode 100644 hadoop-ozone/dist/src/main/license/src/LICENSE.txt
delete mode 100644 hadoop-ozone/dist/src/main/license/src/NOTICE.txt
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFSInputStream.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileInterfaces.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystem.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsRenameDir.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractCreate.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractDelete.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractDistCp.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractGetFileStatus.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractMkdir.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractOpen.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractRename.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractRootDir.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/ITestOzoneContractSeek.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/contract/OzoneContract.java (100%)
copy hadoop-ozone/{ozonefs => integration-test}/src/test/java/org/apache/hadoop/fs/ozone/package-info.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidate.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidateWithDummyContainers.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidateWithSafeByteOperations.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidateWithUnsafeByteOperations.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeFastRestart.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeRestart.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithPipelineDestroy.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/freon/TestRandomKeyGenerator.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/fsck/TestContainerMapper.java (100%)
rename hadoop-ozone/{tools => integration-test}/src/test/java/org/apache/hadoop/ozone/om/TestOmSQLCli.java (100%)
rename hadoop-ozone/{ozonefs => integration-test}/src/test/resources/contract/ozone.xml (100%)
create mode 100644 hadoop-ozone/ozonefs/src/test/java/org/apache/hadoop/fs/ozone/contract/TestOzoneFSInputStreamUnit.java
create mode 100644 hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/api/db.json
create mode 100644 hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/api/routes.json
copy hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/{config-overrides.js => src/components/OverviewCard/OverviewCard.less} (70%)
create mode 100644 hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/src/components/OverviewCard/OverviewCard.tsx
delete mode 100644 hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/src/views/Dashboard/Dashboard.tsx
rename hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/src/views/{ContainerBrowser/ContainerBrowser.tsx => Datanodes/Datanodes.tsx} (89%)
copy hadoop-hdds/container-service/src/main/java/org/apache/hadoop/hdds/scm/package-info.java => hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/src/views/Overview/Overview.less (95%)
create mode 100644 hadoop-ozone/recon/src/main/resources/webapps/recon/ozone-recon-web/src/views/Overview/Overview.tsx
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 06/12: Rebase Fix
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit 667a0c0f1ea8f534c8a619c68a30980af5e81852
Author: Sammi Chen <sa...@apache.org>
AuthorDate: Tue Dec 3 16:32:08 2019 +0800
Rebase Fix
---
.../apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java | 6 +++---
.../org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java | 6 ++++--
2 files changed, 7 insertions(+), 5 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
index 7513cad..ff52470 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
@@ -36,9 +36,9 @@ public class MockRatisPipelineProvider extends RatisPipelineProvider {
private boolean autoOpenPipeline;
public MockRatisPipelineProvider(NodeManager nodeManager,
- PipelineStateManager stateManager,
- Configuration conf, boolean autoOpen) {
- super(nodeManager, stateManager, conf, null);
+ PipelineStateManager stateManager, Configuration conf,
+ EventPublisher eventPublisher, boolean autoOpen) {
+ super(nodeManager, stateManager, conf, eventPublisher);
autoOpenPipeline = autoOpen;
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
index 2df851d..491e289 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
@@ -383,10 +383,12 @@ public class TestSCMPipelineManager {
OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT, -1,
TimeUnit.MILLISECONDS);
+ EventQueue eventQueue = new EventQueue();
final SCMPipelineManager pipelineManager =
- new SCMPipelineManager(conf, nodeManager, new EventQueue(), null);
+ new SCMPipelineManager(conf, nodeManager, eventQueue);
final PipelineProvider ratisProvider = new MockRatisPipelineProvider(
- nodeManager, pipelineManager.getStateManager(), conf, false);
+ nodeManager, pipelineManager.getStateManager(), conf, eventQueue,
+ false);
pipelineManager.setPipelineProvider(HddsProtos.ReplicationType.RATIS,
ratisProvider);
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 03/12: HDDS-2089: Add createPipeline CLI. (#1418)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit f9fe0032325a5531446e28f8386c87c6be59daa0
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Fri Sep 13 07:01:16 2019 +0800
HDDS-2089: Add createPipeline CLI. (#1418)
(cherry picked from commit 326b5acd4a63fe46821919322867f5daff30750c)
---
.../org/apache/hadoop/ozone/audit/SCMAction.java | 1 +
.../hdds/scm/pipeline/SimplePipelineProvider.java | 2 +-
.../hdds/scm/server/SCMClientProtocolServer.java | 8 +--
.../scm/cli/pipeline/CreatePipelineSubcommand.java | 71 ++++++++++++++++++++++
.../hdds/scm/cli/pipeline/PipelineCommands.java | 1 +
5 files changed, 78 insertions(+), 5 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/audit/SCMAction.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/audit/SCMAction.java
index c3e9440..fada2d8 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/audit/SCMAction.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/audit/SCMAction.java
@@ -31,6 +31,7 @@ public enum SCMAction implements AuditAction {
GET_CONTAINER,
GET_CONTAINER_WITH_PIPELINE,
LIST_CONTAINER,
+ CREATE_PIPELINE,
LIST_PIPELINE,
CLOSE_PIPELINE,
ACTIVATE_PIPELINE,
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
index 00cb7ae..a772a97 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SimplePipelineProvider.java
@@ -48,7 +48,7 @@ public class SimplePipelineProvider implements PipelineProvider {
String e = String
.format("Cannot create pipeline of factor %d using %d nodes.",
factor.getNumber(), dns.size());
- throw new IOException(e);
+ throw new InsufficientDatanodesException(e);
}
Collections.shuffle(dns);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
index 4058c0a..61b1791 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMClientProtocolServer.java
@@ -395,10 +395,10 @@ public class SCMClientProtocolServer implements
public Pipeline createReplicationPipeline(HddsProtos.ReplicationType type,
HddsProtos.ReplicationFactor factor, HddsProtos.NodePool nodePool)
throws IOException {
- // TODO: will be addressed in future patch.
- // This is needed only for debugging purposes to make sure cluster is
- // working correctly.
- return null;
+ Pipeline result = scm.getPipelineManager().createPipeline(type, factor);
+ AUDIT.logWriteSuccess(
+ buildAuditMessageForSuccess(SCMAction.CREATE_PIPELINE, null));
+ return result;
}
@Override
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
new file mode 100644
index 0000000..edeb786
--- /dev/null
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
@@ -0,0 +1,71 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.cli.pipeline;
+
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.cli.SCMCLI;
+import org.apache.hadoop.hdds.scm.client.ScmClient;
+import picocli.CommandLine;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Handler of createPipeline command.
+ */
+@CommandLine.Command(
+ name = "createPipeline",
+ description = "create pipeline",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class)
+public class CreatePipelineSubcommand implements Callable<Void> {
+ @CommandLine.ParentCommand
+ private SCMCLI parent;
+
+ @CommandLine.Option(
+ names = {"-t", "--replicationType"},
+ description = "Replication type (STAND_ALONE, RATIS)",
+ defaultValue = "STAND_ALONE"
+ )
+ private HddsProtos.ReplicationType type
+ = HddsProtos.ReplicationType.STAND_ALONE;
+
+ @CommandLine.Option(
+ names = {"-f", "--replicationFactor"},
+ description = "Replication factor (ONE, THREE)",
+ defaultValue = "ONE"
+ )
+ private HddsProtos.ReplicationFactor factor
+ = HddsProtos.ReplicationFactor.ONE;
+
+ @Override
+ public Void call() throws Exception {
+ if (type == HddsProtos.ReplicationType.CHAINED) {
+ throw new IllegalArgumentException(type.name()
+ + " is not supported yet.");
+ }
+ try (ScmClient scmClient = parent.createScmClient()) {
+ scmClient.createReplicationPipeline(
+ type,
+ factor,
+ HddsProtos.NodePool.getDefaultInstance());
+ return null;
+ }
+ }
+}
\ No newline at end of file
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/PipelineCommands.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/PipelineCommands.java
index 948a51a..0bdbc19 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/PipelineCommands.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/PipelineCommands.java
@@ -37,6 +37,7 @@ import java.util.concurrent.Callable;
ListPipelinesSubcommand.class,
ActivatePipelineSubcommand.class,
DeactivatePipelineSubcommand.class,
+ CreatePipelineSubcommand.class,
ClosePipelineSubcommand.class
})
public class PipelineCommands implements Callable<Void> {
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 12/12: Resolve rebase conflict.
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit cc0060c01685ed5497332bd6fbe1cb65910ec5eb
Author: Li Cheng <ti...@tencent.com>
AuthorDate: Mon Dec 23 15:52:37 2019 +0800
Resolve rebase conflict.
---
.../hdds/scm/pipeline/TestPipelineDatanodesIntersection.java | 2 +-
.../hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java | 2 +-
.../hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java | 11 +++++------
3 files changed, 7 insertions(+), 8 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
index 45f85ef..87e8cf4 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
@@ -79,7 +79,7 @@ public class TestPipelineDatanodesIntersection {
NodeManager nodeManager= new MockNodeManager(true, nodeCount);
conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, nodeHeaviness);
conf.setBoolean(OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE, false);
- PipelineStateManager stateManager = new PipelineStateManager(conf);
+ PipelineStateManager stateManager = new PipelineStateManager();
PipelineProvider provider = new MockRatisPipelineProvider(nodeManager,
stateManager, conf);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
index 1e34039..2fe67f9 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
@@ -52,7 +52,7 @@ public class TestPipelinePlacementPolicy {
conf = new OzoneConfiguration();
conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 5);
placementPolicy = new PipelinePlacementPolicy(
- nodeManager, new PipelineStateManager(conf), conf);
+ nodeManager, new PipelineStateManager(), conf);
}
@Test
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
index 46fd8c8..5623359 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
@@ -39,8 +39,6 @@ import java.util.UUID;
import java.util.stream.Collectors;
import static org.apache.commons.collections.CollectionUtils.intersection;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
@@ -57,12 +55,14 @@ public class TestRatisPipelineProvider {
private PipelineProvider provider;
private PipelineStateManager stateManager;
private OzoneConfiguration conf;
+ private int maxPipelinePerNode = 2;
@Before
public void init() throws Exception {
nodeManager = new MockNodeManager(true, 10);
- OzoneConfiguration conf = new OzoneConfiguration();
- conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
+ conf = new OzoneConfiguration();
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
+ maxPipelinePerNode);
stateManager = new PipelineStateManager();
provider = new MockRatisPipelineProvider(nodeManager,
stateManager, conf);
@@ -194,8 +194,7 @@ public class TestRatisPipelineProvider {
// Use up first 3 DNs for an open pipeline.
List<DatanodeDetails> dns = healthyNodes.subList(0, 3);
- for (int i = 0; i < conf.getInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
- OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT); i++) {
+ for (int i = 0; i < maxPipelinePerNode; i++) {
// Saturate pipeline counts on all the 1st 3 DNs.
addPipeline(dns, factor, Pipeline.PipelineState.OPEN, REPLICATION_TYPE);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 08/12: HDDS-2035 Implement datanode level CLI to
reveal pipeline relation. (#348)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit 37961549ab45a7626c0bc6d29a442aa7049bcce8
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Wed Dec 18 19:05:52 2019 +0800
HDDS-2035 Implement datanode level CLI to reveal pipeline relation. (#348)
---
.../org/apache/hadoop/hdds/scm/cli/SCMCLI.java | 2 +
.../hdds/scm/cli/datanode/DatanodeCommands.java | 52 +++++++++
.../hdds/scm/cli/datanode/ListInfoSubcommand.java | 124 +++++++++++++++++++++
.../hadoop/hdds/scm/cli/datanode/package-info.java | 22 ++++
4 files changed, 200 insertions(+)
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/SCMCLI.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/SCMCLI.java
index 8c0fb03..20a35a8 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/SCMCLI.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/SCMCLI.java
@@ -29,6 +29,7 @@ import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.cli.container.ContainerCommands;
+import org.apache.hadoop.hdds.scm.cli.datanode.DatanodeCommands;
import org.apache.hadoop.hdds.scm.cli.pipeline.PipelineCommands;
import org.apache.hadoop.hdds.scm.client.ContainerOperationClient;
import org.apache.hadoop.hdds.scm.client.ScmClient;
@@ -59,6 +60,7 @@ import picocli.CommandLine.Option;
SafeModeCommands.class,
ContainerCommands.class,
PipelineCommands.class,
+ DatanodeCommands.class,
TopologySubcommand.class,
ReplicationManagerCommands.class
},
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/DatanodeCommands.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/DatanodeCommands.java
new file mode 100644
index 0000000..94763d3
--- /dev/null
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/DatanodeCommands.java
@@ -0,0 +1,52 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.cli.datanode;
+
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.cli.MissingSubcommandException;
+import org.apache.hadoop.hdds.scm.cli.SCMCLI;
+import picocli.CommandLine;
+
+import java.util.concurrent.Callable;
+
+/**
+ * Subcommand for datanode related operations.
+ */
+@CommandLine.Command(
+ name = "datanode",
+ description = "Datanode specific operations",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class,
+ subcommands = {
+ ListInfoSubcommand.class
+ })
+public class DatanodeCommands implements Callable<Void> {
+
+ @CommandLine.ParentCommand
+ private SCMCLI parent;
+
+ public SCMCLI getParent() {
+ return parent;
+ }
+
+ @Override
+ public Void call() throws Exception {
+ throw new MissingSubcommandException(
+ this.parent.getCmd().getSubcommands().get("datanode"));
+ }
+}
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/ListInfoSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/ListInfoSubcommand.java
new file mode 100644
index 0000000..dcd8402
--- /dev/null
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/ListInfoSubcommand.java
@@ -0,0 +1,124 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.cli.datanode;
+
+import org.apache.hadoop.hdds.cli.HddsVersionProvider;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.client.ScmClient;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import picocli.CommandLine;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Handler of list datanodes info command.
+ */
+@CommandLine.Command(
+ name = "list",
+ description = "List info of datanodes",
+ mixinStandardHelpOptions = true,
+ versionProvider = HddsVersionProvider.class)
+public class ListInfoSubcommand implements Callable<Void> {
+
+ @CommandLine.ParentCommand
+ private DatanodeCommands parent;
+
+ @CommandLine.Option(names = {"--ip"},
+ description = "Show info by ip address.",
+ defaultValue = "",
+ required = false)
+ private String ipaddress;
+
+ @CommandLine.Option(names = {"--id"},
+ description = "Show info by datanode UUID.",
+ defaultValue = "",
+ required = false)
+ private String uuid;
+
+ private List<Pipeline> pipelines;
+
+
+ @Override
+ public Void call() throws Exception {
+ try (ScmClient scmClient = parent.getParent().createScmClient()) {
+ pipelines = scmClient.listPipelines();
+ if (isNullOrEmpty(ipaddress) && isNullOrEmpty(uuid)) {
+ getAllNodes(scmClient).stream().forEach(p -> printDatanodeInfo(p));
+ } else {
+ Stream<DatanodeDetails> allNodes = getAllNodes(scmClient).stream();
+ if (!isNullOrEmpty(ipaddress)) {
+ allNodes = allNodes.filter(p -> p.getIpAddress()
+ .compareToIgnoreCase(ipaddress) == 0);
+ }
+ if (!isNullOrEmpty(uuid)) {
+ allNodes = allNodes.filter(p -> p.getUuid().toString().equals(uuid));
+ }
+ allNodes.forEach(p -> printDatanodeInfo(p));
+ }
+ return null;
+ }
+ }
+
+ private List<DatanodeDetails> getAllNodes(ScmClient scmClient)
+ throws IOException {
+ List<HddsProtos.Node> nodes = scmClient.queryNode(
+ HddsProtos.NodeState.HEALTHY, HddsProtos.QueryScope.CLUSTER, "");
+
+ return nodes.stream()
+ .map(p -> DatanodeDetails.getFromProtoBuf(p.getNodeID()))
+ .collect(Collectors.toList());
+ }
+
+ private void printDatanodeInfo(DatanodeDetails datanode) {
+ StringBuilder pipelineListInfo = new StringBuilder();
+ int relatedPipelineNum = 0;
+ if (!pipelines.isEmpty()) {
+ List<Pipeline> relatedPipelines = pipelines.stream().filter(
+ p -> p.getNodes().contains(datanode)).collect(Collectors.toList());
+ if (relatedPipelines.isEmpty()) {
+ pipelineListInfo.append("No related pipelines" +
+ " or the node is not in Healthy state.");
+ } else {
+ relatedPipelineNum = relatedPipelines.size();
+ relatedPipelines.stream().forEach(
+ p -> pipelineListInfo.append(p.getId().getId().toString())
+ .append("/").append(p.getFactor().toString()).append("/")
+ .append(p.getType().toString()).append("/")
+ .append(p.getPipelineState().toString()).append("/")
+ .append(datanode.getUuid().equals(p.getLeaderId()) ?
+ "Leader" : "Follower")
+ .append(System.getProperty("line.separator")));
+ }
+ } else {
+ pipelineListInfo.append("No pipelines in cluster.");
+ }
+ System.out.println("Datanode: " + datanode.getUuid().toString() +
+ " (" + datanode.getIpAddress() + "/"
+ + datanode.getHostName() + "/" + relatedPipelineNum +
+ " pipelines) \n" + "Related pipelines: \n" + pipelineListInfo);
+ }
+
+ protected static boolean isNullOrEmpty(String str) {
+ return ((str == null) || str.trim().isEmpty());
+ }
+}
\ No newline at end of file
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/package-info.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/package-info.java
new file mode 100644
index 0000000..f4c45cf
--- /dev/null
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/datanode/package-info.java
@@ -0,0 +1,22 @@
+/**
+ * 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.
+ */
+
+/**
+ * Contains all of the datanode related scm commands.
+ */
+package org.apache.hadoop.hdds.scm.cli.datanode;
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 09/12: Revert "HDDS-2650 Fix createPipeline CLI.
(#340)"
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit 00f0f8123a5bfacdeb4c490ce0c498150dc13fa7
Author: Sammi Chen <sa...@apache.org>
AuthorDate: Wed Dec 18 19:47:56 2019 +0800
Revert "HDDS-2650 Fix createPipeline CLI. (#340)"
This reverts commit 7c7171082e344d51b5ef9473413db487e5267ac0.
---
.../hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java | 7 ++++---
1 file changed, 4 insertions(+), 3 deletions(-)
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
index 58a1778..edeb786 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hdds.scm.cli.pipeline;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.cli.SCMCLI;
import org.apache.hadoop.hdds.scm.client.ScmClient;
import picocli.CommandLine;
@@ -29,13 +30,13 @@ import java.util.concurrent.Callable;
* Handler of createPipeline command.
*/
@CommandLine.Command(
- name = "create",
+ name = "createPipeline",
description = "create pipeline",
mixinStandardHelpOptions = true,
versionProvider = HddsVersionProvider.class)
public class CreatePipelineSubcommand implements Callable<Void> {
@CommandLine.ParentCommand
- private PipelineCommands parent;
+ private SCMCLI parent;
@CommandLine.Option(
names = {"-t", "--replicationType"},
@@ -59,7 +60,7 @@ public class CreatePipelineSubcommand implements Callable<Void> {
throw new IllegalArgumentException(type.name()
+ " is not supported yet.");
}
- try (ScmClient scmClient = parent.getParent().createScmClient()) {
+ try (ScmClient scmClient = parent.createScmClient()) {
scmClient.createReplicationPipeline(
type,
factor,
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 01/12: HDDS-1577. Add default pipeline placement
policy implementation. (#1366)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit f1d994e93a47c4c35e872053e0fe7bc3ee90ade0
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Thu Sep 5 11:51:40 2019 +0800
HDDS-1577. Add default pipeline placement policy implementation. (#1366)
(cherry picked from commit b640a5f6d53830aee4b9c2a7d17bf57c987962cd)
---
.../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 5 +
.../common/src/main/resources/ozone-default.xml | 7 +
.../apache/hadoop/hdds/scm/node/NodeManager.java | 14 +
.../hadoop/hdds/scm/node/NodeStateManager.java | 9 +
.../hadoop/hdds/scm/node/SCMNodeManager.java | 19 ++
.../hdds/scm/node/states/Node2ObjectsMap.java | 4 +-
.../hdds/scm/node/states/Node2PipelineMap.java | 12 +-
.../hdds/scm/pipeline/PipelinePlacementPolicy.java | 338 +++++++++++++++++++++
.../hadoop/hdds/scm/container/MockNodeManager.java | 36 ++-
.../scm/pipeline/TestPipelinePlacementPolicy.java | 197 ++++++++++++
.../testutils/ReplicationNodeManagerMock.java | 16 +
11 files changed, 654 insertions(+), 3 deletions(-)
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 00d326e..68a2091 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
@@ -320,6 +320,11 @@ public final class ScmConfigKeys {
public static final String OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT =
"ozone.scm.pipeline.owner.container.count";
public static final int OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT_DEFAULT = 3;
+ // Pipeline placement policy:
+ // the max number of pipelines can a single datanode be engaged in.
+ public static final String OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT =
+ "ozone.scm.datanode.max.pipeline.engagement";
+ public static final int OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT = 5;
public static final String
OZONE_SCM_KEY_VALUE_CONTAINER_DELETION_CHOOSING_POLICY =
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index c83a5bf..315101a 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -838,6 +838,13 @@
</description>
</property>
<property>
+ <name>ozone.scm.datanode.max.pipeline.engagement</name>
+ <value>5</value>
+ <tag>OZONE, SCM, PIPELINE</tag>
+ <description>Max number of pipelines per datanode can be engaged in.
+ </description>
+ </property>
+ <property>
<name>ozone.scm.container.size</name>
<value>5GB</value>
<tag>OZONE, PERFORMANCE, MANAGEMENT</tag>
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
index fd8bb87..37562fe 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hdds.scm.node;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
@@ -118,6 +119,13 @@ public interface NodeManager extends StorageContainerNodeProtocol,
Set<PipelineID> getPipelines(DatanodeDetails datanodeDetails);
/**
+ * Get the count of pipelines a datanodes is associated with.
+ * @param datanodeDetails DatanodeDetails
+ * @return The number of pipelines
+ */
+ int getPipelinesCount(DatanodeDetails datanodeDetails);
+
+ /**
* Add pipeline information in the NodeManager.
* @param pipeline - Pipeline to be added
*/
@@ -199,4 +207,10 @@ public interface NodeManager extends StorageContainerNodeProtocol,
* @return the given datanode, or empty list if none found
*/
List<DatanodeDetails> getNodesByAddress(String address);
+
+ /**
+ * Get cluster map as in network topology for this node manager.
+ * @return cluster map
+ */
+ NetworkTopology getClusterNetworkTopologyMap();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
index 954cb0e..9d2a9f2 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeStateManager.java
@@ -284,6 +284,15 @@ public class NodeStateManager implements Runnable, Closeable {
}
/**
+ * Get the count of pipelines associated to single datanode.
+ * @param datanodeDetails single datanode
+ * @return number of pipelines associated with it
+ */
+ public int getPipelinesCount(DatanodeDetails datanodeDetails) {
+ return node2PipelineMap.getPipelinesCount(datanodeDetails.getUuid());
+ }
+
+ /**
* Get information about the node.
*
* @param datanodeDetails DatanodeDetails
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
index f077e72..66cca46 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java
@@ -510,6 +510,16 @@ public class SCMNodeManager implements NodeManager {
}
/**
+ * Get the count of pipelines a datanodes is associated with.
+ * @param datanodeDetails DatanodeDetails
+ * @return The number of pipelines
+ */
+ @Override
+ public int getPipelinesCount(DatanodeDetails datanodeDetails) {
+ return nodeStateManager.getPipelinesCount(datanodeDetails);
+ }
+
+ /**
* Add pipeline information in the NodeManager.
*
* @param pipeline - Pipeline to be added
@@ -643,6 +653,15 @@ public class SCMNodeManager implements NodeManager {
return results;
}
+ /**
+ * Get cluster map as in network topology for this node manager.
+ * @return cluster map
+ */
+ @Override
+ public NetworkTopology getClusterNetworkTopologyMap() {
+ return clusterMap;
+ }
+
private String nodeResolve(String hostname) {
List<String> hosts = new ArrayList<>(1);
hosts.add(hostname);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2ObjectsMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2ObjectsMap.java
index 37525b0..57a377d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2ObjectsMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2ObjectsMap.java
@@ -67,6 +67,7 @@ public class Node2ObjectsMap<T> {
* @param datanodeID -- Datanode UUID
* @param containerIDs - List of ContainerIDs.
*/
+ @VisibleForTesting
public void insertNewDatanode(UUID datanodeID, Set<T> containerIDs)
throws SCMException {
Preconditions.checkNotNull(containerIDs);
@@ -83,7 +84,8 @@ public class Node2ObjectsMap<T> {
*
* @param datanodeID - Datanode ID.
*/
- void removeDatanode(UUID datanodeID) {
+ @VisibleForTesting
+ public void removeDatanode(UUID datanodeID) {
Preconditions.checkNotNull(datanodeID);
dn2ObjectMap.computeIfPresent(datanodeID, (k, v) -> null);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
index f8633f9..714188d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
@@ -42,7 +42,7 @@ public class Node2PipelineMap extends Node2ObjectsMap<PipelineID> {
}
/**
- * Returns null if there no pipelines associated with this datanode ID.
+ * Returns null if there are no pipelines associated with this datanode ID.
*
* @param datanode - UUID
* @return Set of pipelines or Null.
@@ -52,6 +52,16 @@ public class Node2PipelineMap extends Node2ObjectsMap<PipelineID> {
}
/**
+ * Return 0 if there are no pipelines associated with this datanode ID.
+ * @param datanode - UUID
+ * @return Number of pipelines or 0.
+ */
+ public int getPipelinesCount(UUID datanode) {
+ Set<PipelineID> pipelines = getObjects(datanode);
+ return pipelines == null ? 0 : pipelines.size();
+ }
+
+ /**
* Adds a pipeline entry to a given dataNode in the map.
*
* @param pipeline Pipeline to be added
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
new file mode 100644
index 0000000..cb9954d
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
@@ -0,0 +1,338 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMCommonPolicy;
+import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
+import org.apache.hadoop.hdds.scm.net.Node;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Pipeline placement policy that choose datanodes based on load balancing
+ * and network topology to supply pipeline creation.
+ * <p>
+ * 1. get a list of healthy nodes
+ * 2. filter out nodes that are not too heavily engaged in other pipelines
+ * 3. Choose an anchor node among the viable nodes.
+ * 4. Choose other nodes around the anchor node based on network topology
+ */
+public final class PipelinePlacementPolicy extends SCMCommonPolicy {
+ @VisibleForTesting
+ static final Logger LOG =
+ LoggerFactory.getLogger(PipelinePlacementPolicy.class);
+ private final NodeManager nodeManager;
+ private final Configuration conf;
+ private final int heavyNodeCriteria;
+
+ /**
+ * Constructs a pipeline placement with considering network topology,
+ * load balancing and rack awareness.
+ *
+ * @param nodeManager Node Manager
+ * @param conf Configuration
+ */
+ public PipelinePlacementPolicy(
+ final NodeManager nodeManager, final Configuration conf) {
+ super(nodeManager, conf);
+ this.nodeManager = nodeManager;
+ this.conf = conf;
+ heavyNodeCriteria = conf.getInt(
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT);
+ }
+
+ /**
+ * Returns true if this node meets the criteria.
+ *
+ * @param datanodeDetails DatanodeDetails
+ * @return true if we have enough space.
+ */
+ @VisibleForTesting
+ boolean meetCriteria(DatanodeDetails datanodeDetails, long heavyNodeLimit) {
+ return (nodeManager.getPipelinesCount(datanodeDetails) <= heavyNodeLimit);
+ }
+
+ /**
+ * Filter out viable nodes based on
+ * 1. nodes that are healthy
+ * 2. nodes that are not too heavily engaged in other pipelines
+ *
+ * @param excludedNodes - excluded nodes
+ * @param nodesRequired - number of datanodes required.
+ * @return a list of viable nodes
+ * @throws SCMException when viable nodes are not enough in numbers
+ */
+ List<DatanodeDetails> filterViableNodes(
+ List<DatanodeDetails> excludedNodes, int nodesRequired)
+ throws SCMException {
+ // get nodes in HEALTHY state
+ List<DatanodeDetails> healthyNodes =
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
+ if (excludedNodes != null) {
+ healthyNodes.removeAll(excludedNodes);
+ }
+ String msg;
+ if (healthyNodes.size() == 0) {
+ msg = "No healthy node found to allocate pipeline.";
+ LOG.error(msg);
+ throw new SCMException(msg, SCMException.ResultCodes
+ .FAILED_TO_FIND_HEALTHY_NODES);
+ }
+
+ if (healthyNodes.size() < nodesRequired) {
+ msg = String.format("Not enough healthy nodes to allocate pipeline. %d "
+ + " datanodes required. Found %d",
+ nodesRequired, healthyNodes.size());
+ LOG.error(msg);
+ throw new SCMException(msg,
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+
+ // filter nodes that meet the size and pipeline engagement criteria.
+ // Pipeline placement doesn't take node space left into account.
+ List<DatanodeDetails> healthyList = healthyNodes.stream().filter(d ->
+ meetCriteria(d, heavyNodeCriteria)).collect(Collectors.toList());
+
+ if (healthyList.size() < nodesRequired) {
+ msg = String.format("Unable to find enough nodes that meet " +
+ "the criteria that cannot engage in more than %d pipelines." +
+ " Nodes required: %d Found: %d",
+ heavyNodeCriteria, nodesRequired, healthyList.size());
+ LOG.error(msg);
+ throw new SCMException(msg,
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+ return healthyList;
+ }
+
+ /**
+ * Pipeline placement choose datanodes to join the pipeline.
+ *
+ * @param excludedNodes - excluded nodes
+ * @param favoredNodes - list of nodes preferred.
+ * @param nodesRequired - number of datanodes required.
+ * @param sizeRequired - size required for the container or block.
+ * @return a list of chosen datanodeDetails
+ * @throws SCMException when chosen nodes are not enough in numbers
+ */
+ @Override
+ public List<DatanodeDetails> chooseDatanodes(
+ List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes,
+ int nodesRequired, final long sizeRequired) throws SCMException {
+ // get a list of viable nodes based on criteria
+ List<DatanodeDetails> healthyNodes =
+ filterViableNodes(excludedNodes, nodesRequired);
+
+ List<DatanodeDetails> results = new ArrayList<>();
+
+ // Randomly picks nodes when all nodes are equal.
+ // This happens when network topology is absent or
+ // all nodes are on the same rack.
+ if (checkAllNodesAreEqual(nodeManager.getClusterNetworkTopologyMap())) {
+ LOG.info("All nodes are considered equal. Now randomly pick nodes. " +
+ "Required nodes: {}", nodesRequired);
+ results = super.getResultSet(nodesRequired, healthyNodes);
+ if (results.size() < nodesRequired) {
+ LOG.error("Unable to find the required number of healthy nodes that " +
+ "meet the criteria. Required nodes: {}, Found nodes: {}",
+ nodesRequired, results.size());
+ throw new SCMException("Unable to find required number of nodes.",
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+ return results;
+ }
+
+ // Since nodes are widely distributed, the results should be selected
+ // base on distance in topology, rack awareness and load balancing.
+ List<DatanodeDetails> exclude = new ArrayList<>();
+ exclude.addAll(excludedNodes);
+ // First choose an anchor nodes randomly
+ DatanodeDetails anchor = chooseNode(healthyNodes);
+ if (anchor == null) {
+ LOG.error("Unable to find the first healthy nodes that " +
+ "meet the criteria. Required nodes: {}, Found nodes: {}",
+ nodesRequired, results.size());
+ throw new SCMException("Unable to find required number of nodes.",
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+
+ results.add(anchor);
+ exclude.add(anchor);
+ nodesRequired--;
+
+ // Choose the second node on different racks from anchor.
+ DatanodeDetails nodeOnDifferentRack = chooseNodeBasedOnRackAwareness(
+ healthyNodes, excludedNodes,
+ nodeManager.getClusterNetworkTopologyMap(), anchor);
+ if (nodeOnDifferentRack == null) {
+ LOG.error("Unable to find nodes on different racks that " +
+ "meet the criteria. Required nodes: {}, Found nodes: {}",
+ nodesRequired, results.size());
+ throw new SCMException("Unable to find required number of nodes.",
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+
+ results.add(nodeOnDifferentRack);
+ exclude.add(nodeOnDifferentRack);
+ nodesRequired--;
+
+ // Then choose nodes close to anchor based on network topology
+ for (int x = 0; x < nodesRequired; x++) {
+ // invoke the choose function defined in the derived classes.
+ DatanodeDetails pick = chooseNodeFromNetworkTopology(
+ nodeManager.getClusterNetworkTopologyMap(), anchor, exclude);
+ if (pick != null) {
+ results.add(pick);
+ // exclude the picked node for next time
+ exclude.add(pick);
+ }
+ }
+
+ if (results.size() < nodesRequired) {
+ LOG.error("Unable to find the required number of healthy nodes that " +
+ "meet the criteria. Required nodes: {}, Found nodes: {}",
+ nodesRequired, results.size());
+ throw new SCMException("Unable to find required number of nodes.",
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+ return results;
+ }
+
+ /**
+ * Find a node from the healthy list and return it after removing it from the
+ * list that we are operating on.
+ *
+ * @param healthyNodes - Set of healthy nodes we can choose from.
+ * @return chosen datanodDetails
+ */
+ @Override
+ public DatanodeDetails chooseNode(
+ List<DatanodeDetails> healthyNodes) {
+ int firstNodeNdx = getRand().nextInt(healthyNodes.size());
+ int secondNodeNdx = getRand().nextInt(healthyNodes.size());
+
+ DatanodeDetails datanodeDetails;
+ // There is a possibility that both numbers will be same.
+ // if that is so, we just return the node.
+ if (firstNodeNdx == secondNodeNdx) {
+ datanodeDetails = healthyNodes.get(firstNodeNdx);
+ } else {
+ DatanodeDetails firstNodeDetails = healthyNodes.get(firstNodeNdx);
+ DatanodeDetails secondNodeDetails = healthyNodes.get(secondNodeNdx);
+ SCMNodeMetric firstNodeMetric =
+ nodeManager.getNodeStat(firstNodeDetails);
+ SCMNodeMetric secondNodeMetric =
+ nodeManager.getNodeStat(secondNodeDetails);
+ datanodeDetails = firstNodeMetric.isGreater(secondNodeMetric.get())
+ ? firstNodeDetails : secondNodeDetails;
+ }
+ // the pick is decided and it should be removed from candidates.
+ healthyNodes.remove(datanodeDetails);
+ return datanodeDetails;
+ }
+
+ /**
+ * Choose node on different racks as anchor is on based on rack awareness.
+ * If a node on different racks cannot be found, then return a random node.
+ * @param healthyNodes healthy nodes
+ * @param excludedNodes excluded nodes
+ * @param networkTopology network topology
+ * @param anchor anchor node
+ * @return a node on different rack
+ */
+ @VisibleForTesting
+ protected DatanodeDetails chooseNodeBasedOnRackAwareness(
+ List<DatanodeDetails> healthyNodes, List<DatanodeDetails> excludedNodes,
+ NetworkTopology networkTopology, DatanodeDetails anchor) {
+ Preconditions.checkArgument(networkTopology != null);
+ if (checkAllNodesAreEqual(networkTopology)) {
+ return null;
+ }
+
+ for (DatanodeDetails node : healthyNodes) {
+ if (excludedNodes.contains(node)
+ || networkTopology.isSameParent(anchor, node)) {
+ continue;
+ } else {
+ // the pick is decided and it should be removed from candidates.
+ healthyNodes.remove(node);
+ return node;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Check if all nodes are equal in topology.
+ * They are equal when network topology is absent or there are on
+ * the same rack.
+ * @param topology network topology
+ * @return true when all nodes are equal
+ */
+ private boolean checkAllNodesAreEqual(NetworkTopology topology) {
+ if (topology == null) {
+ return true;
+ }
+ return (topology.getNumOfNodes(topology.getMaxLevel() - 1) == 1);
+ }
+
+ /**
+ * Choose node based on network topology.
+ * @param networkTopology network topology
+ * @param anchor anchor datanode to start with
+ * @param excludedNodes excluded datanodes
+ * @return chosen datanode
+ */
+ @VisibleForTesting
+ protected DatanodeDetails chooseNodeFromNetworkTopology(
+ NetworkTopology networkTopology, DatanodeDetails anchor,
+ List<DatanodeDetails> excludedNodes) {
+ Preconditions.checkArgument(networkTopology != null);
+
+ Collection<Node> excluded = new ArrayList<>();
+ if (excludedNodes != null && excludedNodes.size() != 0) {
+ excluded.addAll(excludedNodes);
+ }
+ excluded.add(anchor);
+
+ Node pick = networkTopology.chooseRandom(
+ anchor.getNetworkLocation(), excluded);
+ DatanodeDetails pickedNode = (DatanodeDetails) pick;
+ // exclude the picked node for next time
+ if (excludedNodes != null) {
+ excludedNodes.add(pickedNode);
+ }
+ return pickedNode;
+ }
+}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
index 06dc675..bca4189 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
@@ -17,10 +17,12 @@
package org.apache.hadoop.hdds.scm.container;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
+import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.scm.net.NetConstants;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
+import org.apache.hadoop.hdds.scm.net.NetworkTopologyImpl;
import org.apache.hadoop.hdds.scm.net.Node;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
@@ -86,7 +88,7 @@ public class MockNodeManager implements NodeManager {
private final SCMNodeStat aggregateStat;
private boolean safemode;
private final Map<UUID, List<SCMCommand>> commandMap;
- private final Node2PipelineMap node2PipelineMap;
+ private Node2PipelineMap node2PipelineMap;
private final Node2ContainerMap node2ContainerMap;
private NetworkTopology clusterMap;
private ConcurrentMap<String, Set<String>> dnsToUuidMap;
@@ -100,6 +102,7 @@ public class MockNodeManager implements NodeManager {
this.node2ContainerMap = new Node2ContainerMap();
this.dnsToUuidMap = new ConcurrentHashMap<>();
aggregateStat = new SCMNodeStat();
+ clusterMap = new NetworkTopologyImpl(new Configuration());
if (initializeFakeNodes) {
for (int x = 0; x < nodeCount; x++) {
DatanodeDetails dd = MockDatanodeDetails.randomDatanodeDetails();
@@ -251,6 +254,16 @@ public class MockNodeManager implements NodeManager {
}
/**
+ * Get the count of pipelines a datanodes is associated with.
+ * @param datanodeDetails DatanodeDetails
+ * @return The number of pipelines
+ */
+ @Override
+ public int getPipelinesCount(DatanodeDetails datanodeDetails) {
+ return node2PipelineMap.getPipelinesCount(datanodeDetails.getUuid());
+ }
+
+ /**
* Add pipeline information in the NodeManager.
* @param pipeline - Pipeline to be added
*/
@@ -260,6 +273,22 @@ public class MockNodeManager implements NodeManager {
}
/**
+ * Get the entire Node2PipelineMap.
+ * @return Node2PipelineMap
+ */
+ public Node2PipelineMap getNode2PipelineMap() {
+ return node2PipelineMap;
+ }
+
+ /**
+ * Set the Node2PipelineMap.
+ * @param node2PipelineMap Node2PipelineMap
+ */
+ public void setNode2PipelineMap(Node2PipelineMap node2PipelineMap) {
+ this.node2PipelineMap = node2PipelineMap;
+ }
+
+ /**
* Remove a pipeline information from the NodeManager.
* @param pipeline - Pipeline to be removed
*/
@@ -517,6 +546,11 @@ public class MockNodeManager implements NodeManager {
return results;
}
+ @Override
+ public NetworkTopology getClusterNetworkTopologyMap() {
+ return clusterMap;
+ }
+
public void setNetworkTopology(NetworkTopology topology) {
this.clusterMap = topology;
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
new file mode 100644
index 0000000..2e0d0b1
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
@@ -0,0 +1,197 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.container.MockNodeManager;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.net.*;
+import org.apache.hadoop.hdds.scm.node.states.Node2PipelineMap;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * Test for PipelinePlacementPolicy.
+ */
+public class TestPipelinePlacementPolicy {
+ private MockNodeManager nodeManager;
+ private PipelinePlacementPolicy placementPolicy;
+ private static final int PIPELINE_PLACEMENT_MAX_NODES_COUNT = 10;
+
+ @Before
+ public void init() throws Exception {
+ nodeManager = new MockNodeManager(true,
+ PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ placementPolicy =
+ new PipelinePlacementPolicy(nodeManager, new OzoneConfiguration());
+ }
+
+ @Test
+ public void testChooseNodeBasedOnNetworkTopology() {
+ List<DatanodeDetails> healthyNodes =
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
+ DatanodeDetails anchor = placementPolicy.chooseNode(healthyNodes);
+ // anchor should be removed from healthyNodes after being chosen.
+ Assert.assertFalse(healthyNodes.contains(anchor));
+
+ List<DatanodeDetails> excludedNodes =
+ new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT);
+ DatanodeDetails nextNode = placementPolicy.chooseNodeFromNetworkTopology(
+ nodeManager.getClusterNetworkTopologyMap(), anchor, excludedNodes);
+ // excludedNodes should contain nextNode after being chosen.
+ Assert.assertTrue(excludedNodes.contains(nextNode));
+ // nextNode should not be the same as anchor.
+ Assert.assertTrue(anchor.getUuid() != nextNode.getUuid());
+ }
+
+ @Test
+ public void testChooseNodeBasedOnRackAwareness() {
+ List<DatanodeDetails> healthyNodes = overWriteLocationInNodes(
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY));
+ DatanodeDetails anchor = placementPolicy.chooseNode(healthyNodes);
+ NetworkTopology topologyWithDifRacks =
+ createNetworkTopologyOnDifRacks();
+ DatanodeDetails nextNode = placementPolicy.chooseNodeBasedOnRackAwareness(
+ healthyNodes, new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
+ topologyWithDifRacks, anchor);
+ Assert.assertFalse(topologyWithDifRacks.isSameParent(anchor, nextNode));
+ }
+
+ private final static Node[] NODES = new NodeImpl[] {
+ new NodeImpl("h1", "/r1", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h2", "/r1", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h3", "/r1", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h4", "/r1", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h5", "/r2", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h6", "/r2", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h7", "/r2", NetConstants.NODE_COST_DEFAULT),
+ new NodeImpl("h8", "/r2", NetConstants.NODE_COST_DEFAULT),
+ };
+
+
+ private NetworkTopology createNetworkTopologyOnDifRacks() {
+ NetworkTopology topology = new NetworkTopologyImpl(new Configuration());
+ for (Node n : NODES) {
+ topology.add(n);
+ }
+ return topology;
+ }
+
+ private List<DatanodeDetails> overWriteLocationInNodes(
+ List<DatanodeDetails> datanodes) {
+ List<DatanodeDetails> results = new ArrayList<>(datanodes.size());
+ for (int i = 0; i < datanodes.size(); i++) {
+ DatanodeDetails datanode = datanodes.get(i);
+ DatanodeDetails result = DatanodeDetails.newBuilder()
+ .setUuid(datanode.getUuidString())
+ .setHostName(datanode.getHostName())
+ .setIpAddress(datanode.getIpAddress())
+ .addPort(datanode.getPort(DatanodeDetails.Port.Name.STANDALONE))
+ .addPort(datanode.getPort(DatanodeDetails.Port.Name.RATIS))
+ .addPort(datanode.getPort(DatanodeDetails.Port.Name.REST))
+ .setNetworkLocation(NODES[i].getNetworkLocation()).build();
+ results.add(result);
+ }
+ return results;
+ }
+
+ @Test
+ public void testHeavyNodeShouldBeExcluded() throws SCMException{
+ List<DatanodeDetails> healthyNodes =
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
+ int nodesRequired = healthyNodes.size()/2;
+ // only minority of healthy NODES are heavily engaged in pipelines.
+ int minorityHeavy = healthyNodes.size()/2 - 1;
+ List<DatanodeDetails> pickedNodes1 = placementPolicy.chooseDatanodes(
+ new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
+ new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
+ nodesRequired, 0);
+ // modify node to pipeline mapping.
+ insertHeavyNodesIntoNodeManager(healthyNodes, minorityHeavy);
+ // NODES should be sufficient.
+ Assert.assertEquals(nodesRequired, pickedNodes1.size());
+ // make sure pipeline placement policy won't select duplicated NODES.
+ Assert.assertTrue(checkDuplicateNodesUUID(pickedNodes1));
+
+ // majority of healthy NODES are heavily engaged in pipelines.
+ int majorityHeavy = healthyNodes.size()/2 + 2;
+ insertHeavyNodesIntoNodeManager(healthyNodes, majorityHeavy);
+ boolean thrown = false;
+ List<DatanodeDetails> pickedNodes2 = null;
+ try {
+ pickedNodes2 = placementPolicy.chooseDatanodes(
+ new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
+ new ArrayList<>(PIPELINE_PLACEMENT_MAX_NODES_COUNT),
+ nodesRequired, 0);
+ } catch (SCMException e) {
+ Assert.assertFalse(thrown);
+ thrown = true;
+ }
+ // NODES should NOT be sufficient and exception should be thrown.
+ Assert.assertNull(pickedNodes2);
+ Assert.assertTrue(thrown);
+ }
+
+ private boolean checkDuplicateNodesUUID(List<DatanodeDetails> nodes) {
+ HashSet<UUID> uuids = nodes.stream().
+ map(DatanodeDetails::getUuid).
+ collect(Collectors.toCollection(HashSet::new));
+ return uuids.size() == nodes.size();
+ }
+
+ private Set<PipelineID> mockPipelineIDs(int count) {
+ Set<PipelineID> pipelineIDs = new HashSet<>(count);
+ for (int i = 0; i < count; i++) {
+ pipelineIDs.add(PipelineID.randomId());
+ }
+ return pipelineIDs;
+ }
+
+ private void insertHeavyNodesIntoNodeManager(
+ List<DatanodeDetails> nodes, int heavyNodeCount) throws SCMException{
+ if (nodes == null) {
+ throw new SCMException("",
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+
+ int considerHeavyCount =
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT + 1;
+
+ Node2PipelineMap mockMap = new Node2PipelineMap();
+ for (DatanodeDetails node : nodes) {
+ // mock heavy node
+ if (heavyNodeCount > 0) {
+ mockMap.insertNewDatanode(
+ node.getUuid(), mockPipelineIDs(considerHeavyCount));
+ heavyNodeCount--;
+ } else {
+ mockMap.insertNewDatanode(node.getUuid(), mockPipelineIDs(1));
+ }
+ }
+ nodeManager.setNode2PipelineMap(mockMap);
+ }
+}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
index 0ecff3f..7e8ec52 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
@@ -20,6 +20,7 @@ import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
@@ -166,6 +167,16 @@ public class ReplicationNodeManagerMock implements NodeManager {
}
/**
+ * Get the count of pipelines a datanodes is associated with.
+ * @param dnId DatanodeDetails
+ * @return The number of pipelines
+ */
+ @Override
+ public int getPipelinesCount(DatanodeDetails dnId) {
+ throw new UnsupportedOperationException("Not yet implemented");
+ }
+
+ /**
* Add pipeline information in the NodeManager.
* @param pipeline - Pipeline to be added
*/
@@ -327,4 +338,9 @@ public class ReplicationNodeManagerMock implements NodeManager {
public List<DatanodeDetails> getNodesByAddress(String address) {
return new LinkedList<>();
}
+
+ @Override
+ public NetworkTopology getClusterNetworkTopologyMap() {
+ return null;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 04/12: HDDS-1569 Support creating multiple pipelines
with same datanode. Contributed by Li Cheng.
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit f8c5eb0795082571558699d24648cf3c5ae75816
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Tue Oct 29 12:46:00 2019 +0800
HDDS-1569 Support creating multiple pipelines with same datanode. Contributed by Li Cheng.
This closes #28
---
.../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 10 +-
.../common/src/main/resources/ozone-default.xml | 15 ++-
.../hadoop/hdds/scm/block/BlockManagerImpl.java | 5 +
.../ContainerPlacementPolicyFactory.java | 8 +-
.../hdds/scm/node/states/Node2PipelineMap.java | 2 +-
.../scm/pipeline/BackgroundPipelineCreator.java | 1 +
.../hdds/scm/pipeline/PipelinePlacementPolicy.java | 89 +++++++++----
.../hadoop/hdds/scm/pipeline/PipelineStateMap.java | 5 +-
.../hdds/scm/pipeline/RatisPipelineProvider.java | 137 ++++++++++++---------
.../hdds/scm/pipeline/RatisPipelineUtils.java | 103 ++++++++++++++++
.../hdds/scm/pipeline/SCMPipelineManager.java | 13 +-
.../hdds/scm/pipeline/SCMPipelineMetrics.java | 8 ++
.../scm/safemode/HealthyPipelineSafeModeRule.java | 13 +-
.../hadoop/hdds/scm/node/TestDeadNodeHandler.java | 3 +
.../scm/pipeline/TestPipelinePlacementPolicy.java | 15 ++-
.../hdds/scm/pipeline/TestSCMPipelineManager.java | 17 +--
.../apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java | 3 +
.../hdds/scm/pipeline/TestPipelineClose.java | 1 +
.../TestRatisPipelineCreateAndDestroy.java | 24 +++-
.../scm/pipeline/TestRatisPipelineProvider.java | 112 +++++++----------
.../hadoop/hdds/scm/pipeline/TestSCMRestart.java | 5 +-
.../safemode/TestSCMSafeModeWithPipelineRules.java | 3 +
.../org/apache/hadoop/ozone/MiniOzoneCluster.java | 12 ++
.../apache/hadoop/ozone/MiniOzoneClusterImpl.java | 6 +-
.../ozone/client/rpc/Test2WayCommitInRatis.java | 1 +
.../ozone/client/rpc/TestBlockOutputStream.java | 1 +
.../rpc/TestBlockOutputStreamWithFailures.java | 7 +-
.../hadoop/ozone/client/rpc/TestCommitWatcher.java | 1 +
.../rpc/TestContainerReplicationEndToEnd.java | 5 +-
.../client/rpc/TestContainerStateMachine.java | 5 +-
.../client/rpc/TestDeleteWithSlowFollower.java | 12 +-
.../client/rpc/TestFailureHandlingByClient.java | 4 +-
.../client/rpc/TestHybridPipelineOnDatanode.java | 3 +-
.../ozone/client/rpc/TestKeyInputStream.java | 1 +
.../rpc/TestMultiBlockWritesWithDnFailures.java | 8 +-
.../rpc/TestOzoneClientRetriesOnException.java | 1 +
.../client/rpc/TestOzoneRpcClientAbstract.java | 1 +
.../ozone/client/rpc/TestWatchForCommit.java | 3 +
.../TestCloseContainerByPipeline.java | 5 +
.../hadoop/ozone/freon/TestDataValidate.java | 2 +-
.../ozone/freon/TestFreonWithPipelineDestroy.java | 1 +
.../TestSCMContainerPlacementPolicyMetrics.java | 1 +
.../hadoop/ozone/scm/node/TestQueryNode.java | 3 +
43 files changed, 473 insertions(+), 202 deletions(-)
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 68a2091..7e8c435 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
@@ -324,7 +324,15 @@ public final class ScmConfigKeys {
// the max number of pipelines can a single datanode be engaged in.
public static final String OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT =
"ozone.scm.datanode.max.pipeline.engagement";
- public static final int OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT = 5;
+ // Setting to zero by default means this limit doesn't take effect.
+ public static final int OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT = 0;
+
+ // Upper limit for how many pipelines can be created.
+ // Only for test purpose now.
+ public static final String OZONE_SCM_PIPELINE_NUMBER_LIMIT =
+ "ozone.scm.pipeline.number.limit";
+ // Setting to zero by default means this limit doesn't take effect.
+ public static final int OZONE_SCM_PIPELINE_NUMBER_LIMIT_DEFAULT = 0;
public static final String
OZONE_SCM_KEY_VALUE_CONTAINER_DELETION_CHOOSING_POLICY =
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index 9555d08..3093292 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -840,10 +840,19 @@
</description>
</property>
<property>
- <name>ozone.scm.datanode.max.pipeline.engagement</name>
- <value>5</value>
+ <name>ozone.scm.datanode.max.pipeline.engagement</name>
+ <value>0</value>
+ <tag>OZONE, SCM, PIPELINE</tag>
+ <description>Max number of pipelines per datanode can be engaged in.
+ </description>
+ </property>
+ <property>
+ <name>ozone.scm.pipeline.number.limit</name>
+ <value>0</value>
<tag>OZONE, SCM, PIPELINE</tag>
- <description>Max number of pipelines per datanode can be engaged in.
+ <description>Upper limit for how many pipelines can be OPEN in SCM.
+ 0 as default means there is no limit. Otherwise, the number is the limit
+ of max amount of pipelines which are OPEN.
</description>
</property>
<property>
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
index b7a7525..cdc3878 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
@@ -197,8 +197,13 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
// TODO: #CLUTIL Remove creation logic when all replication types and
// factors are handled by pipeline creator
pipeline = pipelineManager.createPipeline(type, factor);
+
// wait until pipeline is ready
pipelineManager.waitPipelineReady(pipeline.getId(), 0);
+ } catch (SCMException se) {
+ LOG.warn("Pipeline creation failed for type:{} factor:{}. " +
+ "Datanodes may be used up.", type, factor, se);
+ break;
} catch (IOException e) {
LOG.warn("Pipeline creation failed for type:{} factor:{}. Retrying " +
"get pipelines call once.", type, factor, e);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
index adaeb87..74431f9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
@@ -43,10 +43,10 @@ public final class ContainerPlacementPolicyFactory {
}
- public static PlacementPolicy getPolicy(Configuration conf,
- final NodeManager nodeManager, NetworkTopology clusterMap,
- final boolean fallback, SCMContainerPlacementMetrics metrics)
- throws SCMException{
+ public static PlacementPolicy getPolicy(
+ Configuration conf, final NodeManager nodeManager,
+ NetworkTopology clusterMap, final boolean fallback,
+ SCMContainerPlacementMetrics metrics) throws SCMException{
final Class<? extends PlacementPolicy> placementClass = conf
.getClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
OZONE_SCM_CONTAINER_PLACEMENT_IMPL_DEFAULT,
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
index 714188d..18809ed 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/states/Node2PipelineMap.java
@@ -80,7 +80,7 @@ public class Node2PipelineMap extends Node2ObjectsMap<PipelineID> {
dn2ObjectMap.computeIfPresent(dnId,
(k, v) -> {
v.remove(pipeline.getId());
- return v;
+ return v.isEmpty() ? null : v;
});
}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
index 3006987..b663f2a 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
@@ -115,6 +115,7 @@ class BackgroundPipelineCreator {
if (scheduler.isClosed()) {
break;
}
+
pipelineManager.createPipeline(type, factor);
} catch (IOException ioe) {
break;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
index 1983ed6..23eb574 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
@@ -36,6 +36,7 @@ import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -52,6 +53,7 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
static final Logger LOG =
LoggerFactory.getLogger(PipelinePlacementPolicy.class);
private final NodeManager nodeManager;
+ private final PipelineStateManager stateManager;
private final Configuration conf;
private final int heavyNodeCriteria;
@@ -59,15 +61,17 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
* Constructs a pipeline placement with considering network topology,
* load balancing and rack awareness.
*
- * @param nodeManager Node Manager
+ * @param nodeManager NodeManager
+ * @param stateManager PipelineStateManager
* @param conf Configuration
*/
- public PipelinePlacementPolicy(
- final NodeManager nodeManager, final Configuration conf) {
+ public PipelinePlacementPolicy(final NodeManager nodeManager,
+ final PipelineStateManager stateManager, final Configuration conf) {
super(nodeManager, conf);
this.nodeManager = nodeManager;
this.conf = conf;
- heavyNodeCriteria = conf.getInt(
+ this.stateManager = stateManager;
+ this.heavyNodeCriteria = conf.getInt(
ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT);
}
@@ -76,11 +80,46 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
* Returns true if this node meets the criteria.
*
* @param datanodeDetails DatanodeDetails
+ * @param nodesRequired nodes required count
* @return true if we have enough space.
*/
@VisibleForTesting
- boolean meetCriteria(DatanodeDetails datanodeDetails, long heavyNodeLimit) {
- return (nodeManager.getPipelinesCount(datanodeDetails) <= heavyNodeLimit);
+ boolean meetCriteria(DatanodeDetails datanodeDetails, int nodesRequired) {
+ if (heavyNodeCriteria == 0) {
+ // no limit applied.
+ return true;
+ }
+ // Datanodes from pipeline in some states can also be considered available
+ // for pipeline allocation. Thus the number of these pipeline shall be
+ // deducted from total heaviness calculation.
+ int pipelineNumDeductable = 0;
+ Set<PipelineID> pipelines = nodeManager.getPipelines(datanodeDetails);
+ for (PipelineID pid : pipelines) {
+ Pipeline pipeline;
+ try {
+ pipeline = stateManager.getPipeline(pid);
+ } catch (PipelineNotFoundException e) {
+ LOG.error("Pipeline not found in pipeline state manager during" +
+ " pipeline creation. PipelineID: " + pid +
+ " exception: " + e.getMessage());
+ continue;
+ }
+ if (pipeline != null &&
+ pipeline.getFactor().getNumber() == nodesRequired &&
+ pipeline.getType() == HddsProtos.ReplicationType.RATIS &&
+ pipeline.getPipelineState() == Pipeline.PipelineState.CLOSED) {
+ pipelineNumDeductable++;
+ }
+ }
+ boolean meet = (nodeManager.getPipelinesCount(datanodeDetails)
+ - pipelineNumDeductable) < heavyNodeCriteria;
+ if (!meet) {
+ LOG.info("Pipeline Placement: can't place more pipeline on heavy " +
+ "datanode： " + datanodeDetails.getUuid().toString() + " Heaviness: " +
+ nodeManager.getPipelinesCount(datanodeDetails) + " limit: " +
+ heavyNodeCriteria);
+ }
+ return meet;
}
/**
@@ -102,18 +141,19 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
if (excludedNodes != null) {
healthyNodes.removeAll(excludedNodes);
}
+ int initialHealthyNodesCount = healthyNodes.size();
String msg;
- if (healthyNodes.size() == 0) {
+ if (initialHealthyNodesCount == 0) {
msg = "No healthy node found to allocate pipeline.";
LOG.error(msg);
throw new SCMException(msg, SCMException.ResultCodes
.FAILED_TO_FIND_HEALTHY_NODES);
}
- if (healthyNodes.size() < nodesRequired) {
+ if (initialHealthyNodesCount < nodesRequired) {
msg = String.format("Not enough healthy nodes to allocate pipeline. %d "
+ " datanodes required. Found %d",
- nodesRequired, healthyNodes.size());
+ nodesRequired, initialHealthyNodesCount);
LOG.error(msg);
throw new SCMException(msg,
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
@@ -121,14 +161,17 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
// filter nodes that meet the size and pipeline engagement criteria.
// Pipeline placement doesn't take node space left into account.
- List<DatanodeDetails> healthyList = healthyNodes.stream().filter(d ->
- meetCriteria(d, heavyNodeCriteria)).collect(Collectors.toList());
+ List<DatanodeDetails> healthyList = healthyNodes.stream()
+ .filter(d -> meetCriteria(d, nodesRequired)).limit(nodesRequired)
+ .collect(Collectors.toList());
if (healthyList.size() < nodesRequired) {
msg = String.format("Unable to find enough nodes that meet " +
"the criteria that cannot engage in more than %d pipelines." +
- " Nodes required: %d Found: %d",
- heavyNodeCriteria, nodesRequired, healthyList.size());
+ " Nodes required: %d Found: %d, healthy nodes count in " +
+ "NodeManager: %d.",
+ heavyNodeCriteria, nodesRequired, healthyList.size(),
+ initialHealthyNodesCount);
LOG.error(msg);
throw new SCMException(msg,
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
@@ -154,13 +197,11 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
// and make sure excludedNodes are excluded from list.
List<DatanodeDetails> healthyNodes =
filterViableNodes(excludedNodes, nodesRequired);
-
- // Randomly picks nodes when all nodes are equal.
+
+ // Randomly picks nodes when all nodes are equal or factor is ONE.
// This happens when network topology is absent or
// all nodes are on the same rack.
if (checkAllNodesAreEqual(nodeManager.getClusterNetworkTopologyMap())) {
- LOG.info("All nodes are considered equal. Now randomly pick nodes. " +
- "Required nodes: {}", nodesRequired);
return super.getResultSet(nodesRequired, healthyNodes);
} else {
// Since topology and rack awareness are available, picks nodes
@@ -188,8 +229,8 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
// First choose an anchor nodes randomly
DatanodeDetails anchor = chooseNode(healthyNodes);
if (anchor == null) {
- LOG.error("Unable to find the first healthy nodes that " +
- "meet the criteria. Required nodes: {}, Found nodes: {}",
+ LOG.error("Pipeline Placement: Unable to find the first healthy nodes " +
+ "that meet the criteria. Required nodes: {}, Found nodes: {}",
nodesRequired, results.size());
throw new SCMException("Unable to find required number of nodes.",
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
@@ -204,8 +245,8 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
healthyNodes, exclude,
nodeManager.getClusterNetworkTopologyMap(), anchor);
if (nodeOnDifferentRack == null) {
- LOG.error("Unable to find nodes on different racks that " +
- "meet the criteria. Required nodes: {}, Found nodes: {}",
+ LOG.error("Pipeline Placement: Unable to find nodes on different racks " +
+ " that meet the criteria. Required nodes: {}, Found nodes: {}",
nodesRequired, results.size());
throw new SCMException("Unable to find required number of nodes.",
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
@@ -228,9 +269,9 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
}
if (results.size() < nodesRequired) {
- LOG.error("Unable to find the required number of healthy nodes that " +
- "meet the criteria. Required nodes: {}, Found nodes: {}",
- nodesRequired, results.size());
+ LOG.error("Pipeline Placement: Unable to find the required number of " +
+ "healthy nodes that meet the criteria. Required nodes: {}, " +
+ "Found nodes: {}", nodesRequired, results.size());
throw new SCMException("Unable to find required number of nodes.",
SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateMap.java
index 443378c..8e0f32d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateMap.java
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -52,8 +53,8 @@ class PipelineStateMap {
PipelineStateMap() {
// TODO: Use TreeMap for range operations?
- pipelineMap = new HashMap<>();
- pipeline2container = new HashMap<>();
+ pipelineMap = new ConcurrentHashMap<>();
+ pipeline2container = new ConcurrentHashMap<>();
query2OpenPipelines = new HashMap<>();
initializeQueryMap();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
index dacc4ca..23b02ed 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
@@ -20,14 +20,12 @@ package org.apache.hadoop.hdds.scm.pipeline;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
-import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms
- .SCMContainerPlacementRandom;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
-import org.apache.hadoop.hdds.scm.PlacementPolicy;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
import org.apache.hadoop.hdds.server.events.EventPublisher;
@@ -38,8 +36,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
-import java.lang.reflect.Constructor;
-import java.lang.reflect.InvocationTargetException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
@@ -60,6 +56,9 @@ public class RatisPipelineProvider implements PipelineProvider {
private final PipelineStateManager stateManager;
private final Configuration conf;
private final EventPublisher eventPublisher;
+ private final PipelinePlacementPolicy placementPolicy;
+ private int pipelineNumberLimit;
+ private int maxPipelinePerDatanode;
// Set parallelism at 3, as now in Ratis we create 1 and 3 node pipelines.
private final int parallelismForPool = 3;
@@ -82,65 +81,93 @@ public class RatisPipelineProvider implements PipelineProvider {
this.stateManager = stateManager;
this.conf = conf;
this.eventPublisher = eventPublisher;
+ this.placementPolicy =
+ new PipelinePlacementPolicy(nodeManager, stateManager, conf);
+ this.pipelineNumberLimit = conf.getInt(
+ ScmConfigKeys.OZONE_SCM_PIPELINE_NUMBER_LIMIT,
+ ScmConfigKeys.OZONE_SCM_PIPELINE_NUMBER_LIMIT_DEFAULT);
+ this.maxPipelinePerDatanode = conf.getInt(
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT);
}
+ private List<DatanodeDetails> pickNodesNeverUsed(ReplicationFactor factor)
+ throws SCMException {
+ Set<DatanodeDetails> dnsUsed = new HashSet<>();
+ stateManager.getPipelines(ReplicationType.RATIS, factor)
+ .stream().filter(
+ p -> p.getPipelineState().equals(PipelineState.OPEN) ||
+ p.getPipelineState().equals(PipelineState.DORMANT) ||
+ p.getPipelineState().equals(PipelineState.ALLOCATED))
+ .forEach(p -> dnsUsed.addAll(p.getNodes()));
- /**
- * Create pluggable container placement policy implementation instance.
- *
- * @param nodeManager - SCM node manager.
- * @param conf - configuration.
- * @return SCM container placement policy implementation instance.
- */
- @SuppressWarnings("unchecked")
- // TODO: should we rename PlacementPolicy to PipelinePlacementPolicy?
- private static PlacementPolicy createContainerPlacementPolicy(
- final NodeManager nodeManager, final Configuration conf) {
- Class<? extends PlacementPolicy> implClass =
- (Class<? extends PlacementPolicy>) conf.getClass(
- ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementRandom.class);
+ // Get list of healthy nodes
+ List<DatanodeDetails> dns = nodeManager
+ .getNodes(HddsProtos.NodeState.HEALTHY)
+ .parallelStream()
+ .filter(dn -> !dnsUsed.contains(dn))
+ .limit(factor.getNumber())
+ .collect(Collectors.toList());
+ if (dns.size() < factor.getNumber()) {
+ String e = String
+ .format("Cannot create pipeline of factor %d using %d nodes." +
+ " Used %d nodes. Healthy nodes %d", factor.getNumber(),
+ dns.size(), dnsUsed.size(),
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY).size());
+ throw new SCMException(e,
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
+ return dns;
+ }
- try {
- Constructor<? extends PlacementPolicy> ctor =
- implClass.getDeclaredConstructor(NodeManager.class,
- Configuration.class);
- return ctor.newInstance(nodeManager, conf);
- } catch (RuntimeException e) {
- throw e;
- } catch (InvocationTargetException e) {
- throw new RuntimeException(implClass.getName()
- + " could not be constructed.", e.getCause());
- } catch (Exception e) {
-// LOG.error("Unhandled exception occurred, Placement policy will not " +
-// "be functional.");
- throw new IllegalArgumentException("Unable to load " +
- "PlacementPolicy", e);
+ private boolean exceedPipelineNumberLimit(ReplicationFactor factor) {
+ if (factor != ReplicationFactor.THREE) {
+ // Only put limits for Factor THREE pipelines.
+ return false;
+ }
+ // Per datanode limit
+ if (maxPipelinePerDatanode > 0) {
+ return (stateManager.getPipelines(ReplicationType.RATIS, factor).size() -
+ stateManager.getPipelines(ReplicationType.RATIS, factor,
+ Pipeline.PipelineState.CLOSED).size()) > maxPipelinePerDatanode *
+ nodeManager.getNodeCount(HddsProtos.NodeState.HEALTHY) /
+ factor.getNumber();
+ }
+
+ // Global limit
+ if (pipelineNumberLimit > 0) {
+ return (stateManager.getPipelines(ReplicationType.RATIS,
+ ReplicationFactor.THREE).size() - stateManager.getPipelines(
+ ReplicationType.RATIS, ReplicationFactor.THREE,
+ Pipeline.PipelineState.CLOSED).size()) >
+ (pipelineNumberLimit - stateManager.getPipelines(
+ ReplicationType.RATIS, ReplicationFactor.ONE).size());
}
+
+ return false;
}
@Override
public Pipeline create(ReplicationFactor factor) throws IOException {
- // Get set of datanodes already used for ratis pipeline
- Set<DatanodeDetails> dnsUsed = new HashSet<>();
- stateManager.getPipelines(ReplicationType.RATIS, factor).stream().filter(
- p -> p.getPipelineState().equals(PipelineState.OPEN) ||
- p.getPipelineState().equals(PipelineState.DORMANT) ||
- p.getPipelineState().equals(PipelineState.ALLOCATED))
- .forEach(p -> dnsUsed.addAll(p.getNodes()));
+ if (exceedPipelineNumberLimit(factor)) {
+ throw new SCMException("Ratis pipeline number meets the limit: " +
+ pipelineNumberLimit + " factor : " +
+ factor.getNumber(),
+ SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
+ }
- // Get list of healthy nodes
- List<DatanodeDetails> dns =
- nodeManager.getNodes(NodeState.HEALTHY)
- .parallelStream()
- .filter(dn -> !dnsUsed.contains(dn))
- .limit(factor.getNumber())
- .collect(Collectors.toList());
- if (dns.size() < factor.getNumber()) {
- String e = String
- .format("Cannot create pipeline of factor %d using %d nodes.",
- factor.getNumber(), dns.size());
- throw new InsufficientDatanodesException(e);
+ List<DatanodeDetails> dns;
+
+ switch(factor) {
+ case ONE:
+ dns = pickNodesNeverUsed(ReplicationFactor.ONE);
+ break;
+ case THREE:
+ dns = placementPolicy.chooseDatanodes(null,
+ null, factor.getNumber(), 0);
+ break;
+ default:
+ throw new IllegalStateException("Unknown factor: " + factor.name());
}
Pipeline pipeline = Pipeline.newBuilder()
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java
new file mode 100644
index 0000000..04393a1
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
+import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
+import org.apache.hadoop.hdds.ratis.RatisHelper;
+import org.apache.ratis.client.RaftClient;
+import org.apache.ratis.grpc.GrpcTlsConfig;
+import org.apache.ratis.protocol.RaftGroup;
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.RaftPeer;
+import org.apache.ratis.retry.RetryPolicy;
+import org.apache.ratis.rpc.SupportedRpcType;
+import org.apache.ratis.util.TimeDuration;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Utility class for Ratis pipelines. Contains methods to create and destroy
+ * ratis pipelines.
+ */
+public final class RatisPipelineUtils {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(RatisPipelineUtils.class);
+
+ private RatisPipelineUtils() {
+ }
+ /**
+ * Removes pipeline from SCM. Sends ratis command to destroy pipeline on all
+ * the datanodes.
+ *
+ * @param pipeline - Pipeline to be destroyed
+ * @param ozoneConf - Ozone configuration
+ * @param grpcTlsConfig
+ * @throws IOException
+ */
+ public static void destroyPipeline(Pipeline pipeline, Configuration ozoneConf,
+ GrpcTlsConfig grpcTlsConfig) {
+ final RaftGroup group = RatisHelper.newRaftGroup(pipeline);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("destroying pipeline:{} with {}", pipeline.getId(), group);
+ }
+ for (DatanodeDetails dn : pipeline.getNodes()) {
+ try {
+ destroyPipeline(dn, pipeline.getId(), ozoneConf, grpcTlsConfig);
+ } catch (IOException e) {
+ LOG.warn("Pipeline destroy failed for pipeline={} dn={} exception={}",
+ pipeline.getId(), dn, e.getMessage());
+ }
+ }
+ }
+
+ /**
+ * Sends ratis command to destroy pipeline on the given datanode.
+ *
+ * @param dn - Datanode on which pipeline needs to be destroyed
+ * @param pipelineID - ID of pipeline to be destroyed
+ * @param ozoneConf - Ozone configuration
+ * @param grpcTlsConfig - grpc tls configuration
+ * @throws IOException
+ */
+ static void destroyPipeline(DatanodeDetails dn, PipelineID pipelineID,
+ Configuration ozoneConf, GrpcTlsConfig grpcTlsConfig) throws IOException {
+ final String rpcType = ozoneConf
+ .get(ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_KEY,
+ ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_DEFAULT);
+ final RetryPolicy retryPolicy = RatisHelper.createRetryPolicy(ozoneConf);
+ final RaftPeer p = RatisHelper.toRaftPeer(dn);
+ final int maxOutstandingRequests =
+ HddsClientUtils.getMaxOutstandingRequests(ozoneConf);
+ final TimeDuration requestTimeout =
+ RatisHelper.getClientRequestTimeout(ozoneConf);
+ try(RaftClient client = RatisHelper
+ .newRaftClient(SupportedRpcType.valueOfIgnoreCase(rpcType), p,
+ retryPolicy, maxOutstandingRequests, grpcTlsConfig,
+ requestTimeout)) {
+ client.groupRemove(RaftGroupId.valueOf(pipelineID.getId()),
+ true, p.getId());
+ }
+ }
+}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
index dc13804..1b2c476 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
@@ -55,10 +55,6 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import static org.apache.hadoop.hdds.scm
- .ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_DEFAULT;
-import static org.apache.hadoop.hdds.scm
- .ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_MB;
import static org.apache.hadoop.ozone.OzoneConsts.SCM_PIPELINE_DB;
/**
@@ -98,8 +94,8 @@ public class SCMPipelineManager implements PipelineManager {
scheduler = new Scheduler("RatisPipelineUtilsThread", false, 1);
this.backgroundPipelineCreator =
new BackgroundPipelineCreator(this, scheduler, conf);
- int cacheSize = conf.getInt(OZONE_SCM_DB_CACHE_SIZE_MB,
- OZONE_SCM_DB_CACHE_SIZE_DEFAULT);
+ int cacheSize = conf.getInt(ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_MB,
+ ScmConfigKeys.OZONE_SCM_DB_CACHE_SIZE_DEFAULT);
final File metaDir = ServerUtils.getScmDbDir(conf);
final File pipelineDBPath = new File(metaDir, SCM_PIPELINE_DB);
this.pipelineStore =
@@ -167,10 +163,9 @@ public class SCMPipelineManager implements PipelineManager {
metrics.createPerPipelineMetrics(pipeline);
}
return pipeline;
- } catch (InsufficientDatanodesException idEx) {
- throw idEx;
} catch (IOException ex) {
metrics.incNumPipelineCreationFailed();
+ LOG.error("Pipeline creation failed.", ex);
throw ex;
} finally {
lock.writeLock().unlock();
@@ -179,7 +174,7 @@ public class SCMPipelineManager implements PipelineManager {
@Override
public Pipeline createPipeline(ReplicationType type, ReplicationFactor factor,
- List<DatanodeDetails> nodes) {
+ List<DatanodeDetails> nodes) {
// This will mostly be used to create dummy pipeline for SimplePipelines.
// We don't update the metrics for SimplePipelines.
lock.writeLock().lock();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
index 40a6f29..8c348ed 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
@@ -135,6 +135,14 @@ public final class SCMPipelineMetrics implements MetricsSource {
}
/**
+ * Get the number of pipeline created.
+ * @return number of pipeline
+ */
+ long getNumPipelineCreated() {
+ return numPipelineCreated.value();
+ }
+
+ /**
* Increments number of failed pipeline creation count.
*/
void incNumPipelineCreationFailed() {
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/HealthyPipelineSafeModeRule.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/HealthyPipelineSafeModeRule.java
index 33936d5..1a03c34 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/HealthyPipelineSafeModeRule.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/safemode/HealthyPipelineSafeModeRule.java
@@ -17,11 +17,14 @@
*/
package org.apache.hadoop.hdds.scm.safemode;
+import java.util.HashSet;
+import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.server.events.EventQueue;
@@ -46,6 +49,8 @@ public class HealthyPipelineSafeModeRule
private int healthyPipelineThresholdCount;
private int currentHealthyPipelineCount = 0;
private final double healthyPipelinesPercent;
+ private final Set<PipelineID> processedPipelineIDs =
+ new HashSet<>();
HealthyPipelineSafeModeRule(String ruleName, EventQueue eventQueue,
PipelineManager pipelineManager,
@@ -117,8 +122,11 @@ public class HealthyPipelineSafeModeRule
Preconditions.checkNotNull(pipeline);
if (pipeline.getType() == HddsProtos.ReplicationType.RATIS &&
pipeline.getFactor() == HddsProtos.ReplicationFactor.THREE) {
- getSafeModeMetrics().incCurrentHealthyPipelinesCount();
- currentHealthyPipelineCount++;
+ if (!processedPipelineIDs.contains(pipeline.getId())) {
+ getSafeModeMetrics().incCurrentHealthyPipelinesCount();
+ currentHealthyPipelineCount++;
+ processedPipelineIDs.add(pipeline.getId());
+ }
}
if (scmInSafeMode()) {
@@ -131,6 +139,7 @@ public class HealthyPipelineSafeModeRule
@Override
protected void cleanup() {
+ processedPipelineIDs.clear();
}
@VisibleForTesting
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestDeadNodeHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestDeadNodeHandler.java
index 594ea5c..977038e 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestDeadNodeHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestDeadNodeHandler.java
@@ -66,6 +66,8 @@ import org.junit.Ignore;
import org.junit.Test;
import org.mockito.Mockito;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+
/**
* Test DeadNodeHandler.
*/
@@ -87,6 +89,7 @@ public class TestDeadNodeHandler {
storageDir = GenericTestUtils.getTempPath(
TestDeadNodeHandler.class.getSimpleName() + UUID.randomUUID());
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, storageDir);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 0);
eventQueue = new EventQueue();
scm = HddsTestUtils.getScm(conf);
nodeManager = (SCMNodeManager) scm.getScmNodeManager();
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
index 2e0d0b1..1e34039 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelinePlacementPolicy.java
@@ -34,11 +34,14 @@ import org.junit.Test;
import java.util.*;
import java.util.stream.Collectors;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+
/**
* Test for PipelinePlacementPolicy.
*/
public class TestPipelinePlacementPolicy {
private MockNodeManager nodeManager;
+ private OzoneConfiguration conf;
private PipelinePlacementPolicy placementPolicy;
private static final int PIPELINE_PLACEMENT_MAX_NODES_COUNT = 10;
@@ -46,8 +49,10 @@ public class TestPipelinePlacementPolicy {
public void init() throws Exception {
nodeManager = new MockNodeManager(true,
PIPELINE_PLACEMENT_MAX_NODES_COUNT);
- placementPolicy =
- new PipelinePlacementPolicy(nodeManager, new OzoneConfiguration());
+ conf = new OzoneConfiguration();
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 5);
+ placementPolicy = new PipelinePlacementPolicy(
+ nodeManager, new PipelineStateManager(conf), conf);
}
@Test
@@ -123,7 +128,7 @@ public class TestPipelinePlacementPolicy {
public void testHeavyNodeShouldBeExcluded() throws SCMException{
List<DatanodeDetails> healthyNodes =
nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
- int nodesRequired = healthyNodes.size()/2;
+ int nodesRequired = HddsProtos.ReplicationFactor.THREE.getNumber();
// only minority of healthy NODES are heavily engaged in pipelines.
int minorityHeavy = healthyNodes.size()/2 - 1;
List<DatanodeDetails> pickedNodes1 = placementPolicy.chooseDatanodes(
@@ -179,7 +184,9 @@ public class TestPipelinePlacementPolicy {
}
int considerHeavyCount =
- ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT + 1;
+ conf.getInt(
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
+ ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT) + 1;
Node2PipelineMap mockMap = new Node2PipelineMap();
for (DatanodeDetails node : nodes) {
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
index 81723e1..08f5185 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.pipeline;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
@@ -34,12 +35,13 @@ import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.TestUtils;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.MockNodeManager;
-import org.apache.hadoop.hdds.scm.safemode.SCMSafeModeManager;
import org.apache.hadoop.hdds.scm.server.SCMDatanodeHeartbeatDispatcher
.PipelineReportFromDatanode;
-import org.apache.hadoop.hdds.scm.TestUtils;
import org.apache.hadoop.hdds.server.events.EventQueue;
import org.apache.hadoop.metrics2.MetricsRecordBuilder;
import org.apache.hadoop.test.GenericTestUtils;
@@ -59,6 +61,7 @@ public class TestSCMPipelineManager {
@Before
public void setUp() throws Exception {
conf = new OzoneConfiguration();
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 1);
testDir = GenericTestUtils
.getTestDir(TestSCMPipelineManager.class.getSimpleName());
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS, testDir.getAbsolutePath());
@@ -253,10 +256,10 @@ public class TestSCMPipelineManager {
pipelineManager.createPipeline(HddsProtos.ReplicationType.RATIS,
HddsProtos.ReplicationFactor.THREE);
Assert.fail();
- } catch (InsufficientDatanodesException idEx) {
- Assert.assertEquals(
- "Cannot create pipeline of factor 3 using 1 nodes.",
- idEx.getMessage());
+ } catch (SCMException ioe) {
+ // pipeline creation failed this time.
+ Assert.assertEquals(SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE,
+ ioe.getResult());
}
metrics = getMetrics(
@@ -266,7 +269,7 @@ public class TestSCMPipelineManager {
numPipelineCreateFailed = getLongCounter(
"NumPipelineCreationFailed", metrics);
- Assert.assertTrue(numPipelineCreateFailed == 0);
+ Assert.assertTrue(numPipelineCreateFailed == 1);
// clean up
pipelineManager.close();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java
index 23d7833..acc4031 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFsHAURLs.java
@@ -97,6 +97,7 @@ public class TestOzoneFsHAURLs {
conf.setTimeDuration(
OMConfigKeys.OZONE_OM_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_KEY,
LEADER_ELECTION_TIMEOUT, TimeUnit.MILLISECONDS);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 3);
OMStorage omStore = new OMStorage(conf);
omStore.setClusterId(clusterId);
@@ -106,6 +107,8 @@ public class TestOzoneFsHAURLs {
// Start the cluster
cluster = MiniOzoneCluster.newHABuilder(conf)
+ .setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setClusterId(clusterId)
.setScmId(scmId)
.setOMServiceId(omServiceId)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineClose.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineClose.java
index 21fa7bd..aba9cae 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineClose.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineClose.java
@@ -170,6 +170,7 @@ public class TestPipelineClose {
pipelineActionHandler
.onMessage(pipelineActionsFromDatanode, new EventQueue());
Thread.sleep(5000);
+
OzoneContainer ozoneContainer =
cluster.getHddsDatanodes().get(0).getDatanodeStateMachine()
.getContainer();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineCreateAndDestroy.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineCreateAndDestroy.java
index 2f852f2..0874f8b 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineCreateAndDestroy.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineCreateAndDestroy.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hdds.scm.pipeline;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.ozone.HddsDatanodeService;
import org.apache.hadoop.ozone.MiniOzoneCluster;
@@ -35,6 +36,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
/**
@@ -49,9 +51,12 @@ public class TestRatisPipelineCreateAndDestroy {
public void init(int numDatanodes) throws Exception {
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS,
GenericTestUtils.getRandomizedTempPath());
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
+
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(numDatanodes)
- .setHbInterval(1000)
+ .setTotalPipelineNumLimit(numDatanodes + numDatanodes/3)
+ .setHbInterval(2000)
.setHbProcessorInterval(1000)
.build();
cluster.waitForClusterToBeReady();
@@ -132,7 +137,9 @@ public class TestRatisPipelineCreateAndDestroy {
} catch (IOException ioe) {
// As now all datanodes are shutdown, they move to stale state, there
// will be no sufficient datanodes to create the pipeline.
- Assert.assertTrue(ioe instanceof InsufficientDatanodesException);
+ Assert.assertTrue(ioe instanceof SCMException);
+ Assert.assertEquals(SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE,
+ ((SCMException) ioe).getResult());
}
// make sure pipelines is destroyed
@@ -145,9 +152,14 @@ public class TestRatisPipelineCreateAndDestroy {
for (Pipeline pipeline : pipelines) {
pipelineManager.finalizeAndDestroyPipeline(pipeline, false);
}
- // make sure pipelines is created after node start
- pipelineManager.triggerPipelineCreation();
- waitForPipelines(1);
+
+ if (cluster.getStorageContainerManager()
+ .getScmNodeManager().getNodeCount(HddsProtos.NodeState.HEALTHY) >=
+ HddsProtos.ReplicationFactor.THREE.getNumber()) {
+ // make sure pipelines is created after node start
+ pipelineManager.triggerPipelineCreation();
+ waitForPipelines(1);
+ }
}
private void waitForPipelines(int numPipelines)
@@ -155,6 +167,6 @@ public class TestRatisPipelineCreateAndDestroy {
GenericTestUtils.waitFor(() -> pipelineManager
.getPipelines(HddsProtos.ReplicationType.RATIS,
HddsProtos.ReplicationFactor.THREE, Pipeline.PipelineState.OPEN)
- .size() == numPipelines, 100, 40000);
+ .size() >= numPipelines, 100, 40000);
}
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
index 7786195..ce34ba6 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
@@ -18,7 +18,7 @@
package org.apache.hadoop.hdds.scm.pipeline;
-import org.apache.commons.collections.CollectionUtils;
+import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
@@ -33,6 +33,9 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys
+ .OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+
/**
* Test for RatisPipelineProvider.
*/
@@ -45,15 +48,19 @@ public class TestRatisPipelineProvider {
@Before
public void init() throws Exception {
nodeManager = new MockNodeManager(true, 10);
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setBoolean(HddsConfigKeys.HDDS_SCM_SAFEMODE_PIPELINE_CREATION, false);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 1);
stateManager = new PipelineStateManager();
provider = new MockRatisPipelineProvider(nodeManager,
- stateManager, new OzoneConfiguration());
+ stateManager, conf);
}
private void createPipelineAndAssertions(
HddsProtos.ReplicationFactor factor) throws IOException {
Pipeline pipeline = provider.create(factor);
stateManager.addPipeline(pipeline);
+ nodeManager.addPipeline(pipeline);
Assert.assertEquals(pipeline.getType(), HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipeline.getFactor(), factor);
Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
@@ -61,10 +68,7 @@ public class TestRatisPipelineProvider {
Assert.assertEquals(pipeline.getNodes().size(), factor.getNumber());
Pipeline pipeline1 = provider.create(factor);
stateManager.addPipeline(pipeline1);
- // New pipeline should not overlap with the previous created pipeline
- Assert.assertTrue(
- CollectionUtils.intersection(pipeline.getNodes(), pipeline1.getNodes())
- .isEmpty());
+ nodeManager.addPipeline(pipeline1);
Assert.assertEquals(pipeline1.getType(), HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipeline1.getFactor(), factor);
Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
@@ -77,6 +81,7 @@ public class TestRatisPipelineProvider {
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
Pipeline pipeline = provider.create(factor);
stateManager.addPipeline(pipeline);
+ nodeManager.addPipeline(pipeline);
Assert.assertEquals(pipeline.getType(), HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipeline.getFactor(), factor);
Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
@@ -86,11 +91,7 @@ public class TestRatisPipelineProvider {
factor = HddsProtos.ReplicationFactor.ONE;
Pipeline pipeline1 = provider.create(factor);
stateManager.addPipeline(pipeline1);
- // New pipeline should overlap with the previous created pipeline,
- // and one datanode should overlap between the two types.
- Assert.assertEquals(
- CollectionUtils.intersection(pipeline.getNodes(),
- pipeline1.getNodes()).size(), 1);
+ nodeManager.addPipeline(pipeline1);
Assert.assertEquals(pipeline1.getType(), HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipeline1.getFactor(), factor);
Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
@@ -138,54 +139,39 @@ public class TestRatisPipelineProvider {
@Test
public void testCreatePipelinesDnExclude() throws IOException {
-
- // We need 9 Healthy DNs in MockNodeManager.
- NodeManager mockNodeManager = new MockNodeManager(true, 12);
- PipelineStateManager stateManagerMock =
- new PipelineStateManager();
- PipelineProvider providerMock = new MockRatisPipelineProvider(
- mockNodeManager, stateManagerMock, new OzoneConfiguration());
-
- // Use up first 3 DNs for an open pipeline.
- List<DatanodeDetails> openPiplineDns = mockNodeManager.getAllNodes()
- .subList(0, 3);
+ List<DatanodeDetails> allHealthyNodes =
+ nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
+ int totalHealthyNodesCount = allHealthyNodes.size();
+
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
- Pipeline openPipeline = Pipeline.newBuilder()
- .setType(HddsProtos.ReplicationType.RATIS)
- .setFactor(factor)
- .setNodes(openPiplineDns)
- .setState(Pipeline.PipelineState.OPEN)
- .setId(PipelineID.randomId())
- .build();
-
- stateManagerMock.addPipeline(openPipeline);
-
- // Use up next 3 DNs also for an open pipeline.
- List<DatanodeDetails> moreOpenPiplineDns = mockNodeManager.getAllNodes()
- .subList(3, 6);
- Pipeline anotherOpenPipeline = Pipeline.newBuilder()
- .setType(HddsProtos.ReplicationType.RATIS)
- .setFactor(factor)
- .setNodes(moreOpenPiplineDns)
- .setState(Pipeline.PipelineState.OPEN)
- .setId(PipelineID.randomId())
- .build();
- stateManagerMock.addPipeline(anotherOpenPipeline);
-
- // Use up next 3 DNs also for a closed pipeline.
- List<DatanodeDetails> closedPiplineDns = mockNodeManager.getAllNodes()
- .subList(6, 9);
- Pipeline anotherClosedPipeline = Pipeline.newBuilder()
- .setType(HddsProtos.ReplicationType.RATIS)
- .setFactor(factor)
- .setNodes(closedPiplineDns)
- .setState(Pipeline.PipelineState.CLOSED)
- .setId(PipelineID.randomId())
- .build();
- stateManagerMock.addPipeline(anotherClosedPipeline);
-
- Pipeline pipeline = providerMock.create(factor);
+ List<DatanodeDetails> closePipelineDns = new ArrayList<>();
+ for (int i = 0; i < totalHealthyNodesCount/3; i++) {
+ List<DatanodeDetails> pipelineDns = allHealthyNodes
+ .subList(3 * i, 3 * (i + 1));
+
+ Pipeline.PipelineState state;
+ if (i % 2 == 0) {
+ state = Pipeline.PipelineState.OPEN;
+ } else {
+ state = Pipeline.PipelineState.CLOSED;
+ closePipelineDns.addAll(pipelineDns);
+ }
+
+ Pipeline openPipeline = Pipeline.newBuilder()
+ .setType(HddsProtos.ReplicationType.RATIS)
+ .setFactor(factor)
+ .setNodes(pipelineDns)
+ .setState(state)
+ .setId(PipelineID.randomId())
+ .build();
+
+
+ stateManager.addPipeline(openPipeline);
+ nodeManager.addPipeline(openPipeline);
+ }
+
+ Pipeline pipeline = provider.create(factor);
Assert.assertEquals(pipeline.getType(), HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipeline.getFactor(), factor);
Assert.assertEquals(Pipeline.PipelineState.ALLOCATED,
@@ -193,15 +179,9 @@ public class TestRatisPipelineProvider {
Assert.assertEquals(pipeline.getNodes().size(), factor.getNumber());
List<DatanodeDetails> pipelineNodes = pipeline.getNodes();
- // Pipline nodes cannot be from open pipelines.
- Assert.assertTrue(
- pipelineNodes.parallelStream().filter(dn ->
- (openPiplineDns.contains(dn) || moreOpenPiplineDns.contains(dn)))
- .count() == 0);
-
- // Since we have only 9 Healthy DNs, at least 1 pipeline node should have
- // been from the closed pipeline DN list.
+ // Since we have only 10 DNs, at least 1 pipeline node should have been
+ // from the closed pipeline DN list.
Assert.assertTrue(pipelineNodes.parallelStream().filter(
- closedPiplineDns::contains).count() > 0);
+ closePipelineDns::contains).count() > 0);
}
}
\ No newline at end of file
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMRestart.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMRestart.java
index 459a67a..baeee6a 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMRestart.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMRestart.java
@@ -57,8 +57,11 @@ public class TestSCMRestart {
@BeforeClass
public static void init() throws Exception {
conf = new OzoneConfiguration();
+ int numOfNodes = 4;
cluster = MiniOzoneCluster.newBuilder(conf)
- .setNumDatanodes(4)
+ .setNumDatanodes(numOfNodes)
+ // allow only one FACTOR THREE pipeline.
+ .setTotalPipelineNumLimit(numOfNodes + 1)
.setHbInterval(1000)
.setHbProcessorInterval(1000)
.build();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeWithPipelineRules.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeWithPipelineRules.java
index 7cfd555..09c633d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeWithPipelineRules.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSCMSafeModeWithPipelineRules.java
@@ -38,6 +38,7 @@ import org.junit.rules.TemporaryFolder;
import java.util.List;
import java.util.concurrent.TimeoutException;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
import static org.junit.Assert.fail;
/**
@@ -62,6 +63,8 @@ public class TestSCMSafeModeWithPipelineRules {
true);
conf.set(HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, "10s");
conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL, "10s");
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 50);
+
clusterBuilder = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(numDatanodes)
.setHbInterval(1000)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
index 0102246..1070888 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
@@ -238,6 +238,7 @@ public interface MiniOzoneCluster {
protected static final int DEFAULT_HB_INTERVAL_MS = 1000;
protected static final int DEFAULT_HB_PROCESSOR_INTERVAL_MS = 100;
protected static final int ACTIVE_OMS_NOT_SET = -1;
+ protected static final int DEFAULT_PIPELIME_LIMIT = 3;
protected final OzoneConfiguration conf;
protected String path;
@@ -265,6 +266,7 @@ public interface MiniOzoneCluster {
protected int numOfDatanodes = 3;
protected boolean startDataNodes = true;
protected CertificateClient certClient;
+ protected int pipelineNumLimit = DEFAULT_PIPELIME_LIMIT;
protected Builder(OzoneConfiguration conf) {
this.conf = conf;
@@ -352,6 +354,16 @@ public interface MiniOzoneCluster {
}
/**
+ * Sets the total number of pipelines to create.
+ * @param val number of pipelines
+ * @return MiniOzoneCluster.Builder
+ */
+ public Builder setTotalPipelineNumLimit(int val) {
+ pipelineNumLimit = val;
+ return this;
+ }
+
+ /**
* Sets the number of HeartBeat Interval of Datanodes, the value should be
* in MilliSeconds.
*
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
index ec53755..b0b5135 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneClusterImpl.java
@@ -513,6 +513,10 @@ public class MiniOzoneClusterImpl implements MiniOzoneCluster {
streamBufferMaxSize.get(), streamBufferSizeUnit.get());
conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, blockSize.get(),
streamBufferSizeUnit.get());
+ // MiniOzoneCluster should have global pipeline upper limit.
+ conf.setInt(ScmConfigKeys.OZONE_SCM_PIPELINE_NUMBER_LIMIT,
+ pipelineNumLimit == DEFAULT_PIPELIME_LIMIT ?
+ 2 * numOfDatanodes : pipelineNumLimit);
configureTrace();
}
@@ -524,7 +528,7 @@ public class MiniOzoneClusterImpl implements MiniOzoneCluster {
* Creates a new StorageContainerManager instance.
*
* @return {@link StorageContainerManager}
- *
+ *Wa
* @throws IOException
*/
protected StorageContainerManager createSCM()
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/Test2WayCommitInRatis.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/Test2WayCommitInRatis.java
index fd2cea3..1eef382 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/Test2WayCommitInRatis.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/Test2WayCommitInRatis.java
@@ -81,6 +81,7 @@ public class Test2WayCommitInRatis {
conf.setQuietMode(false);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStream.java
index 2b41012..2584499 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStream.java
@@ -85,6 +85,7 @@ public class TestBlockOutputStream {
StorageUnit.MB);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStreamWithFailures.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStreamWithFailures.java
index 3cbe06d..a4c0ef4 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStreamWithFailures.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockOutputStreamWithFailures.java
@@ -21,6 +21,7 @@ import org.apache.hadoop.hdds.client.ReplicationFactor;
import org.apache.hadoop.hdds.client.ReplicationType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.XceiverClientManager;
import org.apache.hadoop.hdds.scm.XceiverClientMetrics;
import org.apache.hadoop.hdds.scm.XceiverClientRatis;
@@ -90,9 +91,11 @@ public class TestBlockOutputStreamWithFailures {
conf.setQuietMode(false);
conf.setStorageSize(OzoneConfigKeys.OZONE_SCM_BLOCK_SIZE, 4,
StorageUnit.MB);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 3);
+
cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(7)
- .setBlockSize(blockSize).setChunkSize(chunkSize)
- .setStreamBufferFlushSize(flushSize)
+ .setTotalPipelineNumLimit(10).setBlockSize(blockSize)
+ .setChunkSize(chunkSize).setStreamBufferFlushSize(flushSize)
.setStreamBufferMaxSize(maxFlushSize)
.setStreamBufferSizeUnit(StorageUnit.BYTES).build();
cluster.waitForClusterToBeReady();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCommitWatcher.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCommitWatcher.java
index 8089ac3..eb541b2 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCommitWatcher.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestCommitWatcher.java
@@ -96,6 +96,7 @@ public class TestCommitWatcher {
StorageUnit.MB);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerReplicationEndToEnd.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerReplicationEndToEnd.java
index bc27926..1f2a6b5 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerReplicationEndToEnd.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerReplicationEndToEnd.java
@@ -56,6 +56,7 @@ import java.util.function.Predicate;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_DESTROY_TIMEOUT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
/**
* Tests delete key operation with a slow follower in the datanode
@@ -99,10 +100,12 @@ public class TestContainerReplicationEndToEnd {
1000, TimeUnit.SECONDS);
conf.setLong("hdds.scm.replication.thread.interval",
containerReportInterval);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
conf.setQuietMode(false);
cluster =
- MiniOzoneCluster.newBuilder(conf).setNumDatanodes(4).setHbInterval(200)
+ MiniOzoneCluster.newBuilder(conf).setNumDatanodes(4)
+ .setTotalPipelineNumLimit(6).setHbInterval(200)
.build();
cluster.waitForClusterToBeReady();
cluster.getStorageContainerManager().getReplicationManager().start();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachine.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachine.java
index 2e862bf..ebf9b65 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachine.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachine.java
@@ -52,8 +52,7 @@ import java.util.concurrent.TimeUnit;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_BLOCK_TOKEN_ENABLED;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_COMMAND_STATUS_REPORT_INTERVAL;
import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.*;
/**
* Tests the containerStateMachine failure handling.
@@ -82,7 +81,7 @@ public class TestContainerStateMachine {
baseDir.mkdirs();
conf.setBoolean(HDDS_BLOCK_TOKEN_ENABLED, true);
- // conf.setBoolean(OZONE_SECURITY_ENABLED_KEY, true);
+ // conf.setBoolean(OZONE_SECURITY_ENABLED_KEY, true);
conf.setTimeDuration(HDDS_CONTAINER_REPORT_INTERVAL, 200,
TimeUnit.MILLISECONDS);
conf.setTimeDuration(HDDS_COMMAND_STATUS_REPORT_INTERVAL, 200,
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestDeleteWithSlowFollower.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestDeleteWithSlowFollower.java
index 30c2624..644469e 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestDeleteWithSlowFollower.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestDeleteWithSlowFollower.java
@@ -80,6 +80,7 @@ public class TestDeleteWithSlowFollower {
private static String bucketName;
private static String path;
private static XceiverClientManager xceiverClientManager;
+ private static final int FACTOR_THREE_PIPELINE_COUNT = 1;
/**
* Create a MiniDFSCluster for testing.
@@ -111,10 +112,13 @@ public class TestDeleteWithSlowFollower {
1000, TimeUnit.SECONDS);
conf.setTimeDuration(OzoneConfigKeys.OZONE_BLOCK_DELETING_SERVICE_INTERVAL,
1, TimeUnit.SECONDS);
-
conf.setQuietMode(false);
- cluster =
- MiniOzoneCluster.newBuilder(conf).setNumDatanodes(3).setHbInterval(100)
+ int numOfDatanodes = 3;
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setNumDatanodes(numOfDatanodes)
+ .setTotalPipelineNumLimit(
+ numOfDatanodes + FACTOR_THREE_PIPELINE_COUNT)
+ .setHbInterval(100)
.build();
cluster.waitForClusterToBeReady();
//the easiest way to create an open container is creating a key
@@ -176,7 +180,7 @@ public class TestDeleteWithSlowFollower {
cluster.getStorageContainerManager().getPipelineManager()
.getPipelines(HddsProtos.ReplicationType.RATIS,
HddsProtos.ReplicationFactor.THREE);
- Assert.assertTrue(pipelineList.size() == 1);
+ Assert.assertTrue(pipelineList.size() >= FACTOR_THREE_PIPELINE_COUNT);
Pipeline pipeline = pipelineList.get(0);
for (HddsDatanodeService dn : cluster.getHddsDatanodes()) {
if (ContainerTestHelper.isRatisFollower(dn, pipeline)) {
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
index a7f5960..d86ae51 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestFailureHandlingByClient.java
@@ -23,6 +23,7 @@ import org.apache.hadoop.hdds.client.ReplicationType;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
@@ -98,6 +99,7 @@ public class TestFailureHandlingByClient {
1, TimeUnit.SECONDS);
conf.setBoolean(
OzoneConfigKeys.OZONE_NETWORK_TOPOLOGY_AWARE_READ_KEY, true);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
conf.setQuietMode(false);
conf.setClass(NET_TOPOLOGY_NODE_SWITCH_MAPPING_IMPL_KEY,
@@ -106,7 +108,7 @@ public class TestFailureHandlingByClient {
Collections.singleton(HddsUtils.getHostName(conf))).get(0),
"/rack1");
cluster = MiniOzoneCluster.newBuilder(conf)
- .setNumDatanodes(10).build();
+ .setNumDatanodes(10).setTotalPipelineNumLimit(15).build();
cluster.waitForClusterToBeReady();
//the easiest way to create an open container is creating a key
client = OzoneClientFactory.getClient(conf);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestHybridPipelineOnDatanode.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestHybridPipelineOnDatanode.java
index 47a716e..75af061 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestHybridPipelineOnDatanode.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestHybridPipelineOnDatanode.java
@@ -67,7 +67,8 @@ public class TestHybridPipelineOnDatanode {
@BeforeClass
public static void init() throws Exception {
conf = new OzoneConfiguration();
- cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(3).build();
+ cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(3)
+ .setTotalPipelineNumLimit(5).build();
cluster.waitForClusterToBeReady();
//the easiest way to create an open container is creating a key
client = OzoneClientFactory.getClient(conf);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestKeyInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestKeyInputStream.java
index d834350..505b46c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestKeyInputStream.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestKeyInputStream.java
@@ -83,6 +83,7 @@ public class TestKeyInputStream {
StorageUnit.MB);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(3)
+ .setTotalPipelineNumLimit(5)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestMultiBlockWritesWithDnFailures.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestMultiBlockWritesWithDnFailures.java
index 5717f58..b60fe3f 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestMultiBlockWritesWithDnFailures.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestMultiBlockWritesWithDnFailures.java
@@ -48,8 +48,7 @@ import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.*;
/**
* Tests MultiBlock Writes with Dn failures by Ozone Client.
@@ -88,10 +87,13 @@ public class TestMultiBlockWritesWithDnFailures {
conf.setTimeDuration(
OzoneConfigKeys.DFS_RATIS_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_KEY,
1, TimeUnit.SECONDS);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
conf.setQuietMode(false);
cluster = MiniOzoneCluster.newBuilder(conf)
- .setNumDatanodes(datanodes).build();
+ .setNumDatanodes(datanodes)
+ .setTotalPipelineNumLimit(0)
+ .build();
cluster.waitForClusterToBeReady();
//the easiest way to create an open container is creating a key
client = OzoneClientFactory.getClient(conf);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientRetriesOnException.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientRetriesOnException.java
index 6758f4f..7f12edd 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientRetriesOnException.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneClientRetriesOnException.java
@@ -92,6 +92,7 @@ public class TestOzoneClientRetriesOnException {
conf.setQuietMode(false);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
index da89103..8abb49d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java
@@ -165,6 +165,7 @@ public abstract class TestOzoneRpcClientAbstract {
static void startCluster(OzoneConfiguration conf) throws Exception {
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(3)
+ .setTotalPipelineNumLimit(10)
.setScmId(scmId)
.build();
cluster.waitForClusterToBeReady();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestWatchForCommit.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestWatchForCommit.java
index 5808655..511315f 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestWatchForCommit.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestWatchForCommit.java
@@ -57,6 +57,7 @@ import java.util.concurrent.TimeoutException;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.HDDS_SCM_WATCHER_TIMEOUT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
/**
* This class verifies the watchForCommit Handling by xceiverClient.
@@ -93,10 +94,12 @@ public class TestWatchForCommit {
conf.setTimeDuration(
OzoneConfigKeys.DFS_RATIS_CLIENT_REQUEST_RETRY_INTERVAL_KEY,
1, TimeUnit.SECONDS);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 5);
conf.setQuietMode(false);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(7)
+ .setTotalPipelineNumLimit(10)
.setBlockSize(blockSize)
.setChunkSize(chunkSize)
.setStreamBufferFlushSize(flushSize)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCloseContainerByPipeline.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCloseContainerByPipeline.java
index b676e1c..3cafb7d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCloseContainerByPipeline.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestCloseContainerByPipeline.java
@@ -52,6 +52,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.concurrent.TimeoutException;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+
/**
* Test container closing.
*/
@@ -73,8 +75,11 @@ public class TestCloseContainerByPipeline {
public static void init() throws Exception {
conf = new OzoneConfiguration();
conf.set(ScmConfigKeys.OZONE_SCM_PIPELINE_OWNER_CONTAINER_COUNT, "1");
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
+
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(10)
+ .setTotalPipelineNumLimit(15)
.build();
cluster.waitForClusterToBeReady();
//the easiest way to create an open container is creating a key
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidate.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidate.java
index fdcb822..3e1c826 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidate.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestDataValidate.java
@@ -42,7 +42,7 @@ public abstract class TestDataValidate {
static void startCluster(OzoneConfiguration conf) throws Exception {
conf.set(OzoneConfigKeys.OZONE_CLIENT_WATCH_REQUEST_TIMEOUT, "5000ms");
cluster = MiniOzoneCluster.newBuilder(conf)
- .setNumDatanodes(5).build();
+ .setNumDatanodes(5).setTotalPipelineNumLimit(8).build();
cluster.waitForClusterToBeReady();
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithPipelineDestroy.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithPipelineDestroy.java
index 13ecab6..bd30d4e 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithPipelineDestroy.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithPipelineDestroy.java
@@ -53,6 +53,7 @@ public class TestFreonWithPipelineDestroy {
.setHbProcessorInterval(1000)
.setHbInterval(1000)
.setNumDatanodes(3)
+ .setTotalPipelineNumLimit(8)
.build();
cluster.waitForClusterToBeReady();
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMContainerPlacementPolicyMetrics.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMContainerPlacementPolicyMetrics.java
index 536d807..e677544 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMContainerPlacementPolicyMetrics.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMContainerPlacementPolicyMetrics.java
@@ -82,6 +82,7 @@ public class TestSCMContainerPlacementPolicyMetrics {
"/rack1");
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(4)
+ .setTotalPipelineNumLimit(10)
.build();
cluster.waitForClusterToBeReady();
metrics = getMetrics(SCMContainerPlacementMetrics.class.getSimpleName());
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/node/TestQueryNode.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/node/TestQueryNode.java
index ecc2b3e..a882dcd 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/node/TestQueryNode.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/node/TestQueryNode.java
@@ -16,6 +16,7 @@
*/
package org.apache.hadoop.ozone.scm.node;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -76,9 +77,11 @@ public class TestQueryNode {
conf.setTimeDuration(HDDS_NODE_REPORT_INTERVAL, 1, SECONDS);
conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 3, SECONDS);
conf.setTimeDuration(OZONE_SCM_DEADNODE_INTERVAL, 6, SECONDS);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 3);
cluster = MiniOzoneCluster.newBuilder(conf)
.setNumDatanodes(numOfDatanodes)
+ .setTotalPipelineNumLimit(numOfDatanodes + numOfDatanodes/2)
.build();
cluster.waitForClusterToBeReady();
scmClient = new ContainerOperationClient(conf);
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 11/12: HDDS-1574 Average out pipeline allocation on
datanodes and add metrcs/test (#291)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit acc964015d1daac9e594c49868fe24d85bf8fbf5
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Thu Dec 19 06:00:35 2019 +0800
HDDS-1574 Average out pipeline allocation on datanodes and add metrcs/test (#291)
---
.../apache/hadoop/hdds/scm/pipeline/Pipeline.java | 19 +++
.../hdds/scm/pipeline/PipelinePlacementPolicy.java | 3 +-
.../hdds/scm/pipeline/PipelineStateManager.java | 7 ++
.../hdds/scm/pipeline/RatisPipelineProvider.java | 8 ++
.../hdds/scm/pipeline/RatisPipelineUtils.java | 38 +++++-
.../hdds/scm/pipeline/SCMPipelineManager.java | 26 +++++
.../hdds/scm/pipeline/SCMPipelineMetrics.java | 9 ++
.../hadoop/hdds/scm/container/MockNodeManager.java | 14 ++-
.../container/TestCloseContainerEventHandler.java | 7 +-
.../scm/pipeline/MockRatisPipelineProvider.java | 4 +
.../TestPipelineDatanodesIntersection.java | 129 +++++++++++++++++++++
.../scm/pipeline/TestRatisPipelineProvider.java | 73 ++++++++++--
.../hdds/scm/pipeline/TestSCMPipelineManager.java | 15 ++-
.../testutils/ReplicationNodeManagerMock.java | 4 +-
14 files changed, 338 insertions(+), 18 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
index 295156d..1dc2373 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
@@ -58,6 +58,8 @@ public final class Pipeline {
private UUID leaderId;
// Timestamp for pipeline upon creation
private Long creationTimestamp;
+ // Only valid for Ratis THREE pipeline. No need persist.
+ private int nodeIdsHash;
/**
* The immutable properties of pipeline object is used in
@@ -73,6 +75,7 @@ public final class Pipeline {
this.state = state;
this.nodeStatus = nodeStatus;
this.creationTimestamp = System.currentTimeMillis();
+ this.nodeIdsHash = 0;
}
/**
@@ -129,6 +132,14 @@ public final class Pipeline {
this.creationTimestamp = creationTimestamp;
}
+ public int getNodeIdsHash() {
+ return nodeIdsHash;
+ }
+
+ void setNodeIdsHash(int nodeIdsHash) {
+ this.nodeIdsHash = nodeIdsHash;
+ }
+
/**
* Return the pipeline leader's UUID.
*
@@ -347,6 +358,7 @@ public final class Pipeline {
private List<DatanodeDetails> nodesInOrder = null;
private UUID leaderId = null;
private Long creationTimestamp = null;
+ private int nodeIdsHash = 0;
public Builder() {}
@@ -359,6 +371,7 @@ public final class Pipeline {
this.nodesInOrder = pipeline.nodesInOrder.get();
this.leaderId = pipeline.getLeaderId();
this.creationTimestamp = pipeline.getCreationTimestamp();
+ this.nodeIdsHash = 0;
}
public Builder setId(PipelineID id1) {
@@ -397,6 +410,11 @@ public final class Pipeline {
return this;
}
+ public Builder setNodeIdsHash(int nodeIdsHash1) {
+ this.nodeIdsHash = nodeIdsHash1;
+ return this;
+ }
+
public Pipeline build() {
Preconditions.checkNotNull(id);
Preconditions.checkNotNull(type);
@@ -405,6 +423,7 @@ public final class Pipeline {
Preconditions.checkNotNull(nodeStatus);
Pipeline pipeline = new Pipeline(id, type, factor, state, nodeStatus);
pipeline.setLeaderId(leaderId);
+ pipeline.setNodeIdsHash(nodeIdsHash);
// overwrite with original creationTimestamp
if (creationTimestamp != null) {
pipeline.setCreationTimestamp(creationTimestamp);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
index 23eb574..bc65d14 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
@@ -162,7 +162,7 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
// filter nodes that meet the size and pipeline engagement criteria.
// Pipeline placement doesn't take node space left into account.
List<DatanodeDetails> healthyList = healthyNodes.stream()
- .filter(d -> meetCriteria(d, nodesRequired)).limit(nodesRequired)
+ .filter(d -> meetCriteria(d, nodesRequired))
.collect(Collectors.toList());
if (healthyList.size() < nodesRequired) {
@@ -308,6 +308,7 @@ public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
}
// the pick is decided and it should be removed from candidates.
healthyNodes.remove(datanodeDetails);
+
return datanodeDetails;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
index 1842a8d..81287a5 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
@@ -132,6 +132,13 @@ class PipelineStateManager {
pipeline = pipelineStateMap
.updatePipelineState(pipelineId, PipelineState.OPEN);
}
+ // Amend nodeIdsHash if needed.
+ if (pipeline.getType() == ReplicationType.RATIS &&
+ pipeline.getFactor() == ReplicationFactor.THREE &&
+ pipeline.getNodeIdsHash() == 0) {
+ pipeline.setNodeIdsHash(RatisPipelineUtils
+ .encodeNodeIdsOfFactorThreePipeline(pipeline.getNodes()));
+ }
return pipeline;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
index 23b02ed..9585907 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
@@ -157,6 +157,7 @@ public class RatisPipelineProvider implements PipelineProvider {
}
List<DatanodeDetails> dns;
+ int nodeIdHash = 0;
switch(factor) {
case ONE:
@@ -165,6 +166,7 @@ public class RatisPipelineProvider implements PipelineProvider {
case THREE:
dns = placementPolicy.chooseDatanodes(null,
null, factor.getNumber(), 0);
+ nodeIdHash = RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(dns);
break;
default:
throw new IllegalStateException("Unknown factor: " + factor.name());
@@ -176,6 +178,7 @@ public class RatisPipelineProvider implements PipelineProvider {
.setType(ReplicationType.RATIS)
.setFactor(factor)
.setNodes(dns)
+ .setNodeIdsHash(nodeIdHash)
.build();
// Send command to datanodes to create pipeline
@@ -196,12 +199,17 @@ public class RatisPipelineProvider implements PipelineProvider {
@Override
public Pipeline create(ReplicationFactor factor,
List<DatanodeDetails> nodes) {
+ int nodeIdHash = 0;
+ if (factor == ReplicationFactor.THREE) {
+ nodeIdHash = RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(nodes);
+ }
return Pipeline.newBuilder()
.setId(PipelineID.randomId())
.setState(PipelineState.ALLOCATED)
.setType(ReplicationType.RATIS)
.setFactor(factor)
.setNodes(nodes)
+ .setNodeIdsHash(nodeIdHash)
.build();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java
index 04393a1..366eabc 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineUtils.java
@@ -18,9 +18,12 @@
package org.apache.hadoop.hdds.scm.pipeline;
import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.client.HddsClientUtils;
import org.apache.hadoop.hdds.ratis.RatisHelper;
@@ -35,7 +38,6 @@ import org.apache.ratis.util.TimeDuration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-
/**
* Utility class for Ratis pipelines. Contains methods to create and destroy
* ratis pipelines.
@@ -100,4 +102,38 @@ public final class RatisPipelineUtils {
true, p.getId());
}
}
+
+ static int encodeNodeIdsOfFactorThreePipeline(List<DatanodeDetails> nodes) {
+ if (nodes.size() != HddsProtos.ReplicationFactor.THREE.getNumber()) {
+ return 0;
+ }
+ return nodes.get(0).getUuid().hashCode() ^
+ nodes.get(1).getUuid().hashCode() ^
+ nodes.get(2).getUuid().hashCode();
+ }
+
+ /**
+ * Return first existed pipeline which share the same set of datanodes
+ * with the input pipeline.
+ * @param stateManager PipelineStateManager
+ * @param pipeline input pipeline
+ * @return first matched pipeline
+ */
+ static Pipeline checkPipelineContainSameDatanodes(
+ PipelineStateManager stateManager, Pipeline pipeline) {
+ List<Pipeline> matchedPipelines = stateManager.getPipelines(
+ HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE)
+ .stream().filter(p -> !p.getId().equals(pipeline.getId()) &&
+ (// For all OPEN or ALLOCATED pipelines
+ p.getPipelineState() == Pipeline.PipelineState.OPEN ||
+ p.getPipelineState() == Pipeline.PipelineState.ALLOCATED) &&
+ p.getNodeIdsHash() == pipeline.getNodeIdsHash())
+ .collect(Collectors.toList());
+ if (matchedPipelines.size() == 0) {
+ return null;
+ } else {
+ return matchedPipelines.stream().findFirst().get();
+ }
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
index 4346dea..58fce89 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
@@ -128,6 +128,19 @@ public class SCMPipelineManager implements PipelineManager {
pipelineFactory.setProvider(replicationType, provider);
}
+ private int computeNodeIdHash(Pipeline pipeline) {
+ if (pipeline.getType() != ReplicationType.RATIS) {
+ return 0;
+ }
+
+ if (pipeline.getFactor() != ReplicationFactor.THREE) {
+ return 0;
+ }
+
+ return RatisPipelineUtils.
+ encodeNodeIdsOfFactorThreePipeline(pipeline.getNodes());
+ }
+
private void initializePipelineState() throws IOException {
if (pipelineStore.isEmpty()) {
LOG.info("No pipeline exists in current db");
@@ -143,6 +156,7 @@ public class SCMPipelineManager implements PipelineManager {
Pipeline pipeline = Pipeline.getFromProtobuf(pipelineBuilder.setState(
HddsProtos.PipelineState.PIPELINE_ALLOCATED).build());
Preconditions.checkNotNull(pipeline);
+ pipeline.setNodeIdsHash(computeNodeIdHash(pipeline));
stateManager.addPipeline(pipeline);
nodeManager.addPipeline(pipeline);
}
@@ -163,6 +177,18 @@ public class SCMPipelineManager implements PipelineManager {
metrics.incNumPipelineCreated();
metrics.createPerPipelineMetrics(pipeline);
}
+ Pipeline overlapPipeline = RatisPipelineUtils
+ .checkPipelineContainSameDatanodes(stateManager, pipeline);
+ if (overlapPipeline != null) {
+ metrics.incNumPipelineContainSameDatanodes();
+ //TODO remove until pipeline allocation is proved equally distributed.
+ LOG.info("Pipeline: " + pipeline.getId().toString() +
+ " contains same datanodes as previous pipeline: " +
+ overlapPipeline.getId().toString() + " nodeIds: " +
+ pipeline.getNodes().get(0).getUuid().toString() +
+ ", " + pipeline.getNodes().get(1).getUuid().toString() +
+ ", " + pipeline.getNodes().get(2).getUuid().toString());
+ }
return pipeline;
} catch (IOException ex) {
metrics.incNumPipelineCreationFailed();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
index 8c348ed..1cf8d3a 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineMetrics.java
@@ -54,6 +54,7 @@ public final class SCMPipelineMetrics implements MetricsSource {
private @Metric MutableCounterLong numPipelineDestroyFailed;
private @Metric MutableCounterLong numPipelineReportProcessed;
private @Metric MutableCounterLong numPipelineReportProcessingFailed;
+ private @Metric MutableCounterLong numPipelineContainSameDatanodes;
private Map<PipelineID, MutableCounterLong> numBlocksAllocated;
/** Private constructor. */
@@ -92,6 +93,7 @@ public final class SCMPipelineMetrics implements MetricsSource {
numPipelineDestroyFailed.snapshot(recordBuilder, true);
numPipelineReportProcessed.snapshot(recordBuilder, true);
numPipelineReportProcessingFailed.snapshot(recordBuilder, true);
+ numPipelineContainSameDatanodes.snapshot(recordBuilder, true);
numBlocksAllocated
.forEach((pid, metric) -> metric.snapshot(recordBuilder, true));
}
@@ -176,4 +178,11 @@ public final class SCMPipelineMetrics implements MetricsSource {
void incNumPipelineReportProcessingFailed() {
numPipelineReportProcessingFailed.incr();
}
+
+ /**
+ * Increments number of pipeline who contains same set of datanodes.
+ */
+ void incNumPipelineContainSameDatanodes() {
+ numPipelineContainSameDatanodes.incr();
+ }
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
index bca4189..cbeef7f 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/MockNodeManager.java
@@ -17,7 +17,7 @@
package org.apache.hadoop.hdds.scm.container;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
-import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
import org.apache.hadoop.hdds.scm.net.NetConstants;
@@ -93,7 +93,8 @@ public class MockNodeManager implements NodeManager {
private NetworkTopology clusterMap;
private ConcurrentMap<String, Set<String>> dnsToUuidMap;
- public MockNodeManager(boolean initializeFakeNodes, int nodeCount) {
+ public MockNodeManager(NetworkTopologyImpl clusterMap,
+ boolean initializeFakeNodes, int nodeCount) {
this.healthyNodes = new LinkedList<>();
this.staleNodes = new LinkedList<>();
this.deadNodes = new LinkedList<>();
@@ -101,8 +102,8 @@ public class MockNodeManager implements NodeManager {
this.node2PipelineMap = new Node2PipelineMap();
this.node2ContainerMap = new Node2ContainerMap();
this.dnsToUuidMap = new ConcurrentHashMap<>();
- aggregateStat = new SCMNodeStat();
- clusterMap = new NetworkTopologyImpl(new Configuration());
+ this.aggregateStat = new SCMNodeStat();
+ this.clusterMap = clusterMap;
if (initializeFakeNodes) {
for (int x = 0; x < nodeCount; x++) {
DatanodeDetails dd = MockDatanodeDetails.randomDatanodeDetails();
@@ -114,6 +115,11 @@ public class MockNodeManager implements NodeManager {
this.commandMap = new HashMap<>();
}
+ public MockNodeManager(boolean initializeFakeNodes, int nodeCount) {
+ this(new NetworkTopologyImpl(new OzoneConfiguration()),
+ initializeFakeNodes, nodeCount);
+ }
+
/**
* Invoked from ctor to create some node Metrics.
*
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestCloseContainerEventHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestCloseContainerEventHandler.java
index 612bf5d..f35bfe2 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestCloseContainerEventHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestCloseContainerEventHandler.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.TestUtils;
import org.apache.hadoop.hdds.scm.pipeline.MockRatisPipelineProvider;
import org.apache.hadoop.hdds.scm.pipeline.PipelineProvider;
@@ -67,13 +68,14 @@ public class TestCloseContainerEventHandler {
.getTestDir(TestCloseContainerEventHandler.class.getSimpleName());
configuration
.set(HddsConfigKeys.OZONE_METADATA_DIRS, testDir.getAbsolutePath());
+ configuration.setInt(ScmConfigKeys.OZONE_SCM_PIPELINE_NUMBER_LIMIT, 16);
nodeManager = new MockNodeManager(true, 10);
eventQueue = new EventQueue();
pipelineManager =
new SCMPipelineManager(configuration, nodeManager, eventQueue);
PipelineProvider mockRatisProvider =
new MockRatisPipelineProvider(nodeManager,
- pipelineManager.getStateManager(), configuration);
+ pipelineManager.getStateManager(), configuration, eventQueue);
pipelineManager.setPipelineProvider(HddsProtos.ReplicationType.RATIS,
mockRatisProvider);
containerManager = new SCMContainerManager(configuration, pipelineManager);
@@ -91,6 +93,9 @@ public class TestCloseContainerEventHandler {
if (containerManager != null) {
containerManager.close();
}
+ if (pipelineManager != null) {
+ pipelineManager.close();
+ }
FileUtil.fullyDelete(testDir);
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
index ff52470..3eb146a 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
@@ -73,6 +73,8 @@ public class MockRatisPipelineProvider extends RatisPipelineProvider {
.setType(initialPipeline.getType())
.setFactor(factor)
.setNodes(initialPipeline.getNodes())
+ .setNodeIdsHash(RatisPipelineUtils
+ .encodeNodeIdsOfFactorThreePipeline(initialPipeline.getNodes()))
.build();
}
}
@@ -91,6 +93,8 @@ public class MockRatisPipelineProvider extends RatisPipelineProvider {
.setType(HddsProtos.ReplicationType.RATIS)
.setFactor(factor)
.setNodes(nodes)
+ .setNodeIdsHash(RatisPipelineUtils
+ .encodeNodeIdsOfFactorThreePipeline(nodes))
.build();
}
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
new file mode 100644
index 0000000..45f85ef
--- /dev/null
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineDatanodesIntersection.java
@@ -0,0 +1,129 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.pipeline;
+
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.container.MockNodeManager;
+import org.apache.hadoop.hdds.scm.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE;
+
+/**
+ * Test for pipeline datanodes intersection.
+ */
+@RunWith(Parameterized.class)
+public class TestPipelineDatanodesIntersection {
+ private static final Logger LOG = LoggerFactory
+ .getLogger(TestPipelineDatanodesIntersection.class.getName());
+
+ private int nodeCount;
+ private int nodeHeaviness;
+ private OzoneConfiguration conf;
+ private boolean end;
+
+ @Before
+ public void initialize() {
+ conf = new OzoneConfiguration();
+ end = false;
+ }
+
+ public TestPipelineDatanodesIntersection(int nodeCount, int nodeHeaviness) {
+ this.nodeCount = nodeCount;
+ this.nodeHeaviness = nodeHeaviness;
+ }
+
+ @Parameterized.Parameters
+ public static Collection inputParams() {
+ return Arrays.asList(new Object[][] {
+ {4, 5},
+ {10, 5},
+ {20, 5},
+ {50, 5},
+ {100, 5},
+ {100, 10}
+ });
+ }
+
+ @Test
+ public void testPipelineDatanodesIntersection() {
+ NodeManager nodeManager= new MockNodeManager(true, nodeCount);
+ conf.setInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, nodeHeaviness);
+ conf.setBoolean(OZONE_SCM_PIPELINE_AUTO_CREATE_FACTOR_ONE, false);
+ PipelineStateManager stateManager = new PipelineStateManager(conf);
+ PipelineProvider provider = new MockRatisPipelineProvider(nodeManager,
+ stateManager, conf);
+
+ int healthyNodeCount = nodeManager
+ .getNodeCount(HddsProtos.NodeState.HEALTHY);
+ int intersectionCount = 0;
+ int createdPipelineCount = 0;
+ while (!end && createdPipelineCount <= healthyNodeCount * nodeHeaviness) {
+ try {
+ Pipeline pipeline = provider.create(HddsProtos.ReplicationFactor.THREE);
+ stateManager.addPipeline(pipeline);
+ nodeManager.addPipeline(pipeline);
+ Pipeline overlapPipeline = RatisPipelineUtils
+ .checkPipelineContainSameDatanodes(stateManager, pipeline);
+ if (overlapPipeline != null){
+ intersectionCount++;
+ LOG.info("This pipeline: " + pipeline.getId().toString() +
+ " overlaps with previous pipeline: " + overlapPipeline.getId() +
+ ". They share same set of datanodes as: " +
+ pipeline.getNodesInOrder().get(0).getUuid() + "/" +
+ pipeline.getNodesInOrder().get(1).getUuid() + "/" +
+ pipeline.getNodesInOrder().get(2).getUuid() + " and " +
+ overlapPipeline.getNodesInOrder().get(0).getUuid() + "/" +
+ overlapPipeline.getNodesInOrder().get(1).getUuid() + "/" +
+ overlapPipeline.getNodesInOrder().get(2).getUuid() +
+ " is the same.");
+ }
+ createdPipelineCount++;
+ } catch(SCMException e) {
+ end = true;
+ } catch (IOException e) {
+ end = true;
+ // Should not throw regular IOException.
+ Assert.fail();
+ }
+ }
+
+ end = false;
+
+ LOG.info("Among total " +
+ stateManager.getPipelines(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE).size() + " created pipelines" +
+ " with " + healthyNodeCount + " healthy datanodes and " +
+ nodeHeaviness + " as node heaviness, " +
+ intersectionCount + " pipelines has same set of datanodes.");
+ }
+}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
index 66991e4..46fd8c8 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
@@ -25,6 +25,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.MockNodeManager;
import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.junit.Assert;
import org.junit.Assume;
import org.junit.Before;
import org.junit.Test;
@@ -34,9 +35,14 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.UUID;
+import java.util.stream.Collectors;
import static org.apache.commons.collections.CollectionUtils.intersection;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertTrue;
/**
@@ -50,12 +56,13 @@ public class TestRatisPipelineProvider {
private NodeManager nodeManager;
private PipelineProvider provider;
private PipelineStateManager stateManager;
+ private OzoneConfiguration conf;
@Before
public void init() throws Exception {
nodeManager = new MockNodeManager(true, 10);
OzoneConfiguration conf = new OzoneConfiguration();
- conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 1);
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 2);
stateManager = new PipelineStateManager();
provider = new MockRatisPipelineProvider(nodeManager,
stateManager, conf);
@@ -75,8 +82,12 @@ public class TestRatisPipelineProvider {
// New pipeline should not overlap with the previous created pipeline
assertTrue(
intersection(pipeline.getNodes(), pipeline1.getNodes())
- .isEmpty());
+ .size() < factor.getNumber());
+ if (pipeline.getFactor() == HddsProtos.ReplicationFactor.THREE) {
+ assertNotEquals(pipeline.getNodeIdsHash(), pipeline1.getNodeIdsHash());
+ }
stateManager.addPipeline(pipeline1);
+ nodeManager.addPipeline(pipeline1);
}
@Test
@@ -92,10 +103,9 @@ public class TestRatisPipelineProvider {
assertPipelineProperties(pipeline1, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
stateManager.addPipeline(pipeline1);
- // New pipeline should overlap with the previous created pipeline,
- // and one datanode should overlap between the two types.
- assertEquals(1,
- intersection(pipeline.getNodes(), pipeline1.getNodes()).size());
+ // With enough pipeline quote on datanodes, they should not share
+ // the same set of datanodes.
+ assertNotEquals(pipeline.getNodeIdsHash(), pipeline1.getNodeIdsHash());
}
@Test
@@ -131,6 +141,49 @@ public class TestRatisPipelineProvider {
}
@Test
+ public void testComputeNodeIdsHash() {
+ int total = HddsProtos.ReplicationFactor.THREE.getNumber();
+ List<DatanodeDetails> nodes1 = new ArrayList<>();
+ for (int i = 0; i < total; i++) {
+ nodes1.add(MockDatanodeDetails.createDatanodeDetails(
+ UUID.fromString("00000-11000-00000-00000-0000" + (i + 1))));
+ }
+
+ Assert.assertEquals(total, nodes1.size());
+ Assert.assertNotEquals(0,
+ RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(nodes1));
+
+ List<DatanodeDetails> nodes2 = new ArrayList<>();
+ for (int i = 0; i < total; i++) {
+ nodes2.add(MockDatanodeDetails.createDatanodeDetails(
+ UUID.fromString("00000-11000-00000-00000-0000" + (total - i))));
+ }
+ Assert.assertEquals(total, nodes2.size());
+ Assert.assertNotEquals(0,
+ RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(nodes2));
+
+ Assert.assertEquals(
+ RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(nodes1),
+ RatisPipelineUtils.encodeNodeIdsOfFactorThreePipeline(nodes2));
+ }
+
+ @Test
+ public void testCreateFactorTHREEPipelineWithSameDatanodes() {
+ List<DatanodeDetails> healthyNodes = nodeManager
+ .getNodes(HddsProtos.NodeState.HEALTHY).stream()
+ .limit(3).collect(Collectors.toList());
+
+ Pipeline pipeline1 = provider.create(
+ HddsProtos.ReplicationFactor.THREE, healthyNodes);
+ Pipeline pipeline2 = provider.create(
+ HddsProtos.ReplicationFactor.THREE, healthyNodes);
+
+ Assert.assertTrue(pipeline1.getNodes().parallelStream()
+ .allMatch(pipeline2.getNodes()::contains));
+ Assert.assertEquals(pipeline1.getNodeIdsHash(), pipeline2.getNodeIdsHash());
+ }
+
+ @Test
public void testCreatePipelinesDnExclude() throws IOException {
List<DatanodeDetails> healthyNodes =
nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
@@ -141,7 +194,11 @@ public class TestRatisPipelineProvider {
// Use up first 3 DNs for an open pipeline.
List<DatanodeDetails> dns = healthyNodes.subList(0, 3);
- addPipeline(dns, factor, Pipeline.PipelineState.OPEN, REPLICATION_TYPE);
+ for (int i = 0; i < conf.getInt(OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT,
+ OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT_DEFAULT); i++) {
+ // Saturate pipeline counts on all the 1st 3 DNs.
+ addPipeline(dns, factor, Pipeline.PipelineState.OPEN, REPLICATION_TYPE);
+ }
Set<DatanodeDetails> membersOfOpenPipelines = new HashSet<>(dns);
// Use up next 3 DNs for a closed pipeline.
@@ -160,7 +217,7 @@ public class TestRatisPipelineProvider {
List<DatanodeDetails> nodes = pipeline.getNodes();
assertTrue(
- "nodes of new pipeline cannot be from open pipelines",
+ "nodes of new pipeline cannot be all from open pipelines",
nodes.stream().noneMatch(membersOfOpenPipelines::contains));
assertTrue(
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
index 491e289..e6bf7a0 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.pipeline;
+import static org.apache.commons.collections.CollectionUtils.intersection;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT;
import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
@@ -30,6 +31,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
@@ -89,8 +91,10 @@ public class TestSCMPipelineManager {
pipelineManager.getStateManager(), conf);
pipelineManager.setPipelineProvider(HddsProtos.ReplicationType.RATIS,
mockRatisProvider);
+ int pipelineNum = 5;
+
Set<Pipeline> pipelines = new HashSet<>();
- for (int i = 0; i < 5; i++) {
+ for (int i = 0; i < pipelineNum; i++) {
Pipeline pipeline = pipelineManager
.createPipeline(HddsProtos.ReplicationType.RATIS,
HddsProtos.ReplicationFactor.THREE);
@@ -112,6 +116,15 @@ public class TestSCMPipelineManager {
List<Pipeline> pipelineList =
pipelineManager.getPipelines(HddsProtos.ReplicationType.RATIS);
Assert.assertEquals(pipelines, new HashSet<>(pipelineList));
+ // All NodeIdsHash from original pipeline list
+ List<Integer> originalPipelineHash = pipelineList.stream()
+ .map(Pipeline::getNodeIdsHash).collect(Collectors.toList());
+ // All NodeIdsHash from reloaded pipeline list
+ List<Integer> reloadedPipelineHash = pipelines.stream()
+ .map(Pipeline::getNodeIdsHash).collect(Collectors.toList());
+ // Original NodeIdsHash list should contain same items from reloaded one.
+ Assert.assertEquals(pipelineNum,
+ intersection(originalPipelineHash, reloadedPipelineHash).size());
// clean up
for (Pipeline pipeline : pipelines) {
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
index 7e8ec52..0698443 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/ozone/container/testutils/ReplicationNodeManagerMock.java
@@ -168,11 +168,11 @@ public class ReplicationNodeManagerMock implements NodeManager {
/**
* Get the count of pipelines a datanodes is associated with.
- * @param dnId DatanodeDetails
+ * @param dn DatanodeDetails
* @return The number of pipelines
*/
@Override
- public int getPipelinesCount(DatanodeDetails dnId) {
+ public int getPipelinesCount(DatanodeDetails dn) {
throw new UnsupportedOperationException("Not yet implemented");
}
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 07/12: HDDS-2650 Fix createPipeline CLI. (#340)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit a2c65a5fd287a193ba9a4be627931b6ce3a45a2d
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Thu Dec 12 14:11:00 2019 +0800
HDDS-2650 Fix createPipeline CLI. (#340)
---
.../hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java | 7 +++----
1 file changed, 3 insertions(+), 4 deletions(-)
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
index edeb786..58a1778 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdds.scm.cli.pipeline;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.scm.cli.SCMCLI;
import org.apache.hadoop.hdds.scm.client.ScmClient;
import picocli.CommandLine;
@@ -30,13 +29,13 @@ import java.util.concurrent.Callable;
* Handler of createPipeline command.
*/
@CommandLine.Command(
- name = "createPipeline",
+ name = "create",
description = "create pipeline",
mixinStandardHelpOptions = true,
versionProvider = HddsVersionProvider.class)
public class CreatePipelineSubcommand implements Callable<Void> {
@CommandLine.ParentCommand
- private SCMCLI parent;
+ private PipelineCommands parent;
@CommandLine.Option(
names = {"-t", "--replicationType"},
@@ -60,7 +59,7 @@ public class CreatePipelineSubcommand implements Callable<Void> {
throw new IllegalArgumentException(type.name()
+ " is not supported yet.");
}
- try (ScmClient scmClient = parent.createScmClient()) {
+ try (ScmClient scmClient = parent.getParent().createScmClient()) {
scmClient.createReplicationPipeline(
type,
factor,
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 10/12: HDDS-2650 Fix createPipeline CLI and make it
message based. (#370)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit fced9c1ac024bf18e60af8723760449958b54073
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Wed Dec 18 22:27:04 2019 +0800
HDDS-2650 Fix createPipeline CLI and make it message based. (#370)
---
...inerLocationProtocolServerSideTranslatorPB.java | 26 ++++++++++++++++++++++
.../scm/cli/pipeline/CreatePipelineSubcommand.java | 7 +++---
2 files changed, 29 insertions(+), 4 deletions(-)
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
index f48fad1..e587463 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/StorageContainerLocationProtocolServerSideTranslatorPB.java
@@ -23,6 +23,7 @@ import java.util.List;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.PipelineResponseProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ActivatePipelineRequestProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ActivatePipelineResponseProto;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.ClosePipelineRequestProto;
@@ -70,6 +71,9 @@ import com.google.protobuf.ServiceException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.PipelineResponseProto.Error.errorPipelineAlreadyExists;
+import static org.apache.hadoop.hdds.protocol.proto.StorageContainerLocationProtocolProtos.PipelineResponseProto.Error.success;
+
/**
* This class is the server-side translator that forwards requests received on
* {@link StorageContainerLocationProtocolPB} to the
@@ -158,6 +162,12 @@ public final class StorageContainerLocationProtocolServerSideTranslatorPB
.setScmCloseContainerResponse(closeContainer(
request.getScmCloseContainerRequest()))
.build();
+ case AllocatePipeline:
+ return ScmContainerLocationResponse.newBuilder()
+ .setCmdType(request.getCmdType())
+ .setStatus(Status.OK)
+ .setPipelineResponse(allocatePipeline(request.getPipelineRequest()))
+ .build();
case ListPipelines:
return ScmContainerLocationResponse.newBuilder()
.setCmdType(request.getCmdType())
@@ -318,6 +328,22 @@ public final class StorageContainerLocationProtocolServerSideTranslatorPB
return SCMCloseContainerResponseProto.newBuilder().build();
}
+ public PipelineResponseProto allocatePipeline(
+ StorageContainerLocationProtocolProtos.PipelineRequestProto request)
+ throws IOException {
+ Pipeline pipeline = impl.createReplicationPipeline(
+ request.getReplicationType(), request.getReplicationFactor(),
+ HddsProtos.NodePool.getDefaultInstance());
+ if (pipeline == null) {
+ return PipelineResponseProto.newBuilder()
+ .setErrorCode(errorPipelineAlreadyExists).build();
+ }
+ PipelineResponseProto response = PipelineResponseProto.newBuilder()
+ .setErrorCode(success)
+ .setPipeline(pipeline.getProtobufMessage()).build();
+ return response;
+ }
+
public ListPipelineResponseProto listPipelines(
ListPipelineRequestProto request)
throws IOException {
diff --git a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
index edeb786..58a1778 100644
--- a/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
+++ b/hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/pipeline/CreatePipelineSubcommand.java
@@ -20,7 +20,6 @@ package org.apache.hadoop.hdds.scm.cli.pipeline;
import org.apache.hadoop.hdds.cli.HddsVersionProvider;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.scm.cli.SCMCLI;
import org.apache.hadoop.hdds.scm.client.ScmClient;
import picocli.CommandLine;
@@ -30,13 +29,13 @@ import java.util.concurrent.Callable;
* Handler of createPipeline command.
*/
@CommandLine.Command(
- name = "createPipeline",
+ name = "create",
description = "create pipeline",
mixinStandardHelpOptions = true,
versionProvider = HddsVersionProvider.class)
public class CreatePipelineSubcommand implements Callable<Void> {
@CommandLine.ParentCommand
- private SCMCLI parent;
+ private PipelineCommands parent;
@CommandLine.Option(
names = {"-t", "--replicationType"},
@@ -60,7 +59,7 @@ public class CreatePipelineSubcommand implements Callable<Void> {
throw new IllegalArgumentException(type.name()
+ " is not supported yet.");
}
- try (ScmClient scmClient = parent.createScmClient()) {
+ try (ScmClient scmClient = parent.getParent().createScmClient()) {
scmClient.createReplicationPipeline(
type,
factor,
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 02/12: HDDS-1571. Create an interface for pipeline
placement policy to support network topologies. (#1395)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit 7fe5881281e3ec9eca43cd16b92dd673481e15f4
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Tue Sep 10 20:15:51 2019 +0800
HDDS-1571. Create an interface for pipeline placement policy to support network topologies. (#1395)
(cherry picked from commit 753fc6703a39154ed6013e44dbae572391748906)
---
...erPlacementPolicy.java => PlacementPolicy.java} | 12 +++----
.../placement/algorithms/package-info.java | 21 -----------
.../common/src/main/resources/ozone-default.xml | 6 ++--
...onPolicy.java => SCMCommonPlacementPolicy.java} | 23 ++++++------
.../hdds/scm/container/ReplicationManager.java | 13 +++----
.../ContainerPlacementPolicyFactory.java | 18 +++++-----
.../algorithms/SCMContainerPlacementCapacity.java | 4 ++-
.../algorithms/SCMContainerPlacementRackAware.java | 12 +++----
.../algorithms/SCMContainerPlacementRandom.java | 6 ++--
.../hdds/scm/pipeline/PipelinePlacementPolicy.java | 42 +++++++++++++---------
.../hdds/scm/pipeline/RatisPipelineProvider.java | 15 ++++----
.../hdds/scm/server/StorageContainerManager.java | 4 +--
.../hdds/scm/container/TestReplicationManager.java | 7 ++--
.../algorithms/TestContainerPlacementFactory.java | 7 ++--
.../hdds/scm/node/TestContainerPlacement.java | 5 ++-
.../hdds/scm/safemode/TestSafeModeHandler.java | 5 ++-
.../hadoop/ozone/TestContainerOperations.java | 7 ++--
.../TestContainerStateMachineIdempotency.java | 5 ++-
.../hadoop/ozone/dn/scrubber/TestDataScrubber.java | 4 +--
.../hadoop/ozone/scm/TestContainerSmallFile.java | 4 +--
.../scm/TestGetCommittedBlockLengthAndPutKey.java | 5 ++-
21 files changed, 106 insertions(+), 119 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicy.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
similarity index 80%
rename from hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicy.java
rename to hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
index 52ce796..f6a0e8b 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicy.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/PlacementPolicy.java
@@ -15,7 +15,7 @@
* the License.
*/
-package org.apache.hadoop.hdds.scm.container.placement.algorithms;
+package org.apache.hadoop.hdds.scm;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -23,14 +23,14 @@ import java.io.IOException;
import java.util.List;
/**
- * A ContainerPlacementPolicy support choosing datanodes to build replication
- * pipeline with specified constraints.
+ * A PlacementPolicy support choosing datanodes to build
+ * pipelines or containers with specified constraints.
*/
-public interface ContainerPlacementPolicy {
+public interface PlacementPolicy {
/**
- * Given the replication factor and size required, return set of datanodes
- * that satisfy the nodes and size requirement.
+ * Given an initial set of datanodes and the size required,
+ * return set of datanodes that satisfy the nodes and size requirement.
*
* @param excludedNodes - list of nodes to be excluded.
* @param favoredNodes - list of nodes preferred.
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/package-info.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/package-info.java
deleted file mode 100644
index dac4752..0000000
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/package-info.java
+++ /dev/null
@@ -1,21 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.hadoop.hdds.scm.container.placement.algorithms;
-/**
- Contains container placement policy interface definition.
- **/
\ No newline at end of file
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index 315101a..9555d08 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -825,9 +825,11 @@
</value>
<tag>OZONE, MANAGEMENT</tag>
<description>
- The full name of class which implements org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy.
+ The full name of class which implements
+ org.apache.hadoop.hdds.scm.PlacementPolicy.
The class decides which datanode will be used to host the container replica. If not set,
- org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementRandom will be used as default value.
+ org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementRandom will be used as default
+ value.
</description>
</property>
<property>
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMCommonPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java
similarity index 90%
rename from hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMCommonPolicy.java
rename to hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java
index 77cdd83..25457f7 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMCommonPolicy.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/SCMCommonPlacementPolicy.java
@@ -15,7 +15,7 @@
* the License.
*/
-package org.apache.hadoop.hdds.scm.container.placement.algorithms;
+package org.apache.hadoop.hdds.scm;
import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
@@ -33,25 +33,25 @@ import java.util.Random;
import java.util.stream.Collectors;
/**
- * SCM CommonPolicy implements a set of invariants which are common
- * for all container placement policies, acts as the repository of helper
+ * This policy implements a set of invariants which are common
+ * for all basic placement policies, acts as the repository of helper
* functions which are common to placement policies.
*/
-public abstract class SCMCommonPolicy implements ContainerPlacementPolicy {
+public abstract class SCMCommonPlacementPolicy implements PlacementPolicy {
@VisibleForTesting
static final Logger LOG =
- LoggerFactory.getLogger(SCMCommonPolicy.class);
+ LoggerFactory.getLogger(SCMCommonPlacementPolicy.class);
private final NodeManager nodeManager;
private final Random rand;
private final Configuration conf;
/**
- * Constructs SCM Common Policy Class.
+ * Constructor.
*
* @param nodeManager NodeManager
* @param conf Configuration class.
*/
- public SCMCommonPolicy(NodeManager nodeManager, Configuration conf) {
+ public SCMCommonPlacementPolicy(NodeManager nodeManager, Configuration conf) {
this.nodeManager = nodeManager;
this.rand = new Random();
this.conf = conf;
@@ -85,7 +85,7 @@ public abstract class SCMCommonPolicy implements ContainerPlacementPolicy {
}
/**
- * Given the replication factor and size required, return set of datanodes
+ * Given size required, return set of datanodes
* that satisfy the nodes and size requirement.
* <p>
* Here are some invariants of container placement.
@@ -149,7 +149,7 @@ public abstract class SCMCommonPolicy implements ContainerPlacementPolicy {
* @param datanodeDetails DatanodeDetails
* @return true if we have enough space.
*/
- boolean hasEnoughSpace(DatanodeDetails datanodeDetails,
+ public boolean hasEnoughSpace(DatanodeDetails datanodeDetails,
long sizeRequired) {
SCMNodeMetric nodeMetric = nodeManager.getNodeStat(datanodeDetails);
return (nodeMetric != null) && (nodeMetric.get() != null)
@@ -164,7 +164,7 @@ public abstract class SCMCommonPolicy implements ContainerPlacementPolicy {
* @param nodesRequired - Nodes Required
* @param healthyNodes - List of Nodes in the result set.
* @return List of Datanodes that can be used for placement.
- * @throws SCMException
+ * @throws SCMException SCMException
*/
public List<DatanodeDetails> getResultSet(
int nodesRequired, List<DatanodeDetails> healthyNodes)
@@ -190,8 +190,7 @@ public abstract class SCMCommonPolicy implements ContainerPlacementPolicy {
/**
* Choose a datanode according to the policy, this function is implemented
- * by the actual policy class. For example, PlacementCapacity or
- * PlacementRandom.
+ * by the actual policy class.
*
* @param healthyNodes - Set of healthy nodes we can choose from.
* @return DatanodeDetails
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
index bfa411d..509b8c2 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
@@ -38,8 +38,9 @@ import org.apache.hadoop.hdds.conf.ConfigGroup;
import org.apache.hadoop.hdds.conf.ConfigType;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
-import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.protocol.proto
+ .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto.State;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.server.events.EventPublisher;
import org.apache.hadoop.metrics2.MetricsCollector;
@@ -85,7 +86,7 @@ public class ReplicationManager implements MetricsSource {
* PlacementPolicy which is used to identify where a container
* should be replicated.
*/
- private final ContainerPlacementPolicy containerPlacement;
+ private final PlacementPolicy containerPlacement;
/**
* EventPublisher to fire Replicate and Delete container events.
@@ -131,12 +132,12 @@ public class ReplicationManager implements MetricsSource {
*
* @param conf OzoneConfiguration
* @param containerManager ContainerManager
- * @param containerPlacement ContainerPlacementPolicy
+ * @param containerPlacement PlacementPolicy
* @param eventPublisher EventPublisher
*/
public ReplicationManager(final ReplicationManagerConfiguration conf,
final ContainerManager containerManager,
- final ContainerPlacementPolicy containerPlacement,
+ final PlacementPolicy containerPlacement,
final EventPublisher eventPublisher,
final LockManager<ContainerID> lockManager) {
this.containerManager = containerManager;
@@ -474,7 +475,7 @@ public class ReplicationManager implements MetricsSource {
/**
* If the given container is under replicated, identify a new set of
- * datanode(s) to replicate the container using ContainerPlacementPolicy
+ * datanode(s) to replicate the container using PlacementPolicy
* and send replicate container command to the identified datanode(s).
*
* @param container ContainerInfo
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
index 18ec2c3..adaeb87 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/ContainerPlacementPolicyFactory.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hdds.scm.container.placement.algorithms;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
@@ -34,22 +35,23 @@ public final class ContainerPlacementPolicyFactory {
private static final Logger LOG =
LoggerFactory.getLogger(ContainerPlacementPolicyFactory.class);
- private static final Class<? extends ContainerPlacementPolicy>
+ private static final Class<? extends PlacementPolicy>
OZONE_SCM_CONTAINER_PLACEMENT_IMPL_DEFAULT =
SCMContainerPlacementRandom.class;
private ContainerPlacementPolicyFactory() {
}
- public static ContainerPlacementPolicy getPolicy(Configuration conf,
- final NodeManager nodeManager, NetworkTopology clusterMap,
- final boolean fallback, SCMContainerPlacementMetrics metrics)
- throws SCMException{
- final Class<? extends ContainerPlacementPolicy> placementClass = conf
+
+ public static PlacementPolicy getPolicy(Configuration conf,
+ final NodeManager nodeManager, NetworkTopology clusterMap,
+ final boolean fallback, SCMContainerPlacementMetrics metrics)
+ throws SCMException{
+ final Class<? extends PlacementPolicy> placementClass = conf
.getClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
OZONE_SCM_CONTAINER_PLACEMENT_IMPL_DEFAULT,
- ContainerPlacementPolicy.class);
- Constructor<? extends ContainerPlacementPolicy> constructor;
+ PlacementPolicy.class);
+ Constructor<? extends PlacementPolicy> constructor;
try {
constructor = placementClass.getDeclaredConstructor(NodeManager.class,
Configuration.class, NetworkTopology.class, boolean.class,
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
index 85d281c..1909344 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementCapacity.java
@@ -21,6 +21,7 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
@@ -65,7 +66,8 @@ import org.slf4j.LoggerFactory;
* little or no work and the cluster will achieve a balanced distribution
* over time.
*/
-public final class SCMContainerPlacementCapacity extends SCMCommonPolicy {
+public final class SCMContainerPlacementCapacity
+ extends SCMCommonPlacementPolicy {
@VisibleForTesting
static final Logger LOG =
LoggerFactory.getLogger(SCMContainerPlacementCapacity.class);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
index 6d49459..8933fe9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRackAware.java
@@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.net.NetConstants;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
@@ -45,7 +46,8 @@ import java.util.List;
* recommend to use this if the network topology has more layers.
* <p>
*/
-public final class SCMContainerPlacementRackAware extends SCMCommonPolicy {
+public final class SCMContainerPlacementRackAware
+ extends SCMCommonPlacementPolicy {
@VisibleForTesting
static final Logger LOG =
LoggerFactory.getLogger(SCMContainerPlacementRackAware.class);
@@ -271,11 +273,9 @@ public final class SCMContainerPlacementRackAware extends SCMCommonPolicy {
throw new SCMException("No satisfied datanode to meet the" +
" excludedNodes and affinityNode constrains.", null);
}
- if (hasEnoughSpace((DatanodeDetails)node, sizeRequired)) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Datanode {} is chosen for container. Required size is {}",
- node.toString(), sizeRequired);
- }
+ if (super.hasEnoughSpace((DatanodeDetails)node, sizeRequired)) {
+ LOG.debug("Datanode {} is chosen. Required size is {}",
+ node.toString(), sizeRequired);
metrics.incrDatanodeChooseSuccessCount();
if (isFallbacked) {
metrics.incrDatanodeChooseFallbackCount();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
index 6b1a5c8..ce5d10d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/SCMContainerPlacementRandom.java
@@ -19,6 +19,8 @@ package org.apache.hadoop.hdds.scm.container.placement.algorithms;
import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
+import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
import org.apache.hadoop.hdds.scm.node.NodeManager;
@@ -37,8 +39,8 @@ import java.util.List;
* Balancer will need to support containers as a feature before this class
* can be practically used.
*/
-public final class SCMContainerPlacementRandom extends SCMCommonPolicy
- implements ContainerPlacementPolicy {
+public final class SCMContainerPlacementRandom extends SCMCommonPlacementPolicy
+ implements PlacementPolicy {
@VisibleForTesting
static final Logger LOG =
LoggerFactory.getLogger(SCMContainerPlacementRandom.class);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
index cb9954d..1983ed6 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelinePlacementPolicy.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMCommonPolicy;
+import org.apache.hadoop.hdds.scm.SCMCommonPlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.scm.net.NetworkTopology;
@@ -47,7 +47,7 @@ import java.util.stream.Collectors;
* 3. Choose an anchor node among the viable nodes.
* 4. Choose other nodes around the anchor node based on network topology
*/
-public final class PipelinePlacementPolicy extends SCMCommonPolicy {
+public final class PipelinePlacementPolicy extends SCMCommonPlacementPolicy {
@VisibleForTesting
static final Logger LOG =
LoggerFactory.getLogger(PipelinePlacementPolicy.class);
@@ -150,33 +150,41 @@ public final class PipelinePlacementPolicy extends SCMCommonPolicy {
public List<DatanodeDetails> chooseDatanodes(
List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes,
int nodesRequired, final long sizeRequired) throws SCMException {
- // get a list of viable nodes based on criteria
+ // Get a list of viable nodes based on criteria
+ // and make sure excludedNodes are excluded from list.
List<DatanodeDetails> healthyNodes =
filterViableNodes(excludedNodes, nodesRequired);
-
- List<DatanodeDetails> results = new ArrayList<>();
-
+
// Randomly picks nodes when all nodes are equal.
// This happens when network topology is absent or
// all nodes are on the same rack.
if (checkAllNodesAreEqual(nodeManager.getClusterNetworkTopologyMap())) {
LOG.info("All nodes are considered equal. Now randomly pick nodes. " +
"Required nodes: {}", nodesRequired);
- results = super.getResultSet(nodesRequired, healthyNodes);
- if (results.size() < nodesRequired) {
- LOG.error("Unable to find the required number of healthy nodes that " +
- "meet the criteria. Required nodes: {}, Found nodes: {}",
- nodesRequired, results.size());
- throw new SCMException("Unable to find required number of nodes.",
- SCMException.ResultCodes.FAILED_TO_FIND_SUITABLE_NODE);
- }
- return results;
+ return super.getResultSet(nodesRequired, healthyNodes);
+ } else {
+ // Since topology and rack awareness are available, picks nodes
+ // based on them.
+ return this.getResultSet(nodesRequired, healthyNodes);
}
+ }
+ /**
+ * Get result set based on the pipeline placement algorithm which considers
+ * network topology and rack awareness.
+ * @param nodesRequired - Nodes Required
+ * @param healthyNodes - List of Nodes in the result set.
+ * @return a list of datanodes
+ * @throws SCMException SCMException
+ */
+ @Override
+ public List<DatanodeDetails> getResultSet(
+ int nodesRequired, List<DatanodeDetails> healthyNodes)
+ throws SCMException {
+ List <DatanodeDetails> results = new ArrayList<>(nodesRequired);
// Since nodes are widely distributed, the results should be selected
// base on distance in topology, rack awareness and load balancing.
List<DatanodeDetails> exclude = new ArrayList<>();
- exclude.addAll(excludedNodes);
// First choose an anchor nodes randomly
DatanodeDetails anchor = chooseNode(healthyNodes);
if (anchor == null) {
@@ -193,7 +201,7 @@ public final class PipelinePlacementPolicy extends SCMCommonPolicy {
// Choose the second node on different racks from anchor.
DatanodeDetails nodeOnDifferentRack = chooseNodeBasedOnRackAwareness(
- healthyNodes, excludedNodes,
+ healthyNodes, exclude,
nodeManager.getClusterNetworkTopologyMap(), anchor);
if (nodeOnDifferentRack == null) {
LOG.error("Unable to find nodes on different racks that " +
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
index 6b93192..dacc4ca 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/RatisPipelineProvider.java
@@ -25,10 +25,9 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.placement.algorithms
- .ContainerPlacementPolicy;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms
.SCMContainerPlacementRandom;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
import org.apache.hadoop.hdds.server.events.EventPublisher;
@@ -94,16 +93,16 @@ public class RatisPipelineProvider implements PipelineProvider {
* @return SCM container placement policy implementation instance.
*/
@SuppressWarnings("unchecked")
- // TODO: should we rename ContainerPlacementPolicy to PipelinePlacementPolicy?
- private static ContainerPlacementPolicy createContainerPlacementPolicy(
+ // TODO: should we rename PlacementPolicy to PipelinePlacementPolicy?
+ private static PlacementPolicy createContainerPlacementPolicy(
final NodeManager nodeManager, final Configuration conf) {
- Class<? extends ContainerPlacementPolicy> implClass =
- (Class<? extends ContainerPlacementPolicy>) conf.getClass(
+ Class<? extends PlacementPolicy> implClass =
+ (Class<? extends PlacementPolicy>) conf.getClass(
ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
SCMContainerPlacementRandom.class);
try {
- Constructor<? extends ContainerPlacementPolicy> ctor =
+ Constructor<? extends PlacementPolicy> ctor =
implClass.getDeclaredConstructor(NodeManager.class,
Configuration.class);
return ctor.newInstance(nodeManager, conf);
@@ -116,7 +115,7 @@ public class RatisPipelineProvider implements PipelineProvider {
// LOG.error("Unhandled exception occurred, Placement policy will not " +
// "be functional.");
throw new IllegalArgumentException("Unable to load " +
- "ContainerPlacementPolicy", e);
+ "PlacementPolicy", e);
}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
index 7e76e6a..65c56df 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
@@ -58,7 +58,7 @@ import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.hdds.scm.container.ContainerReportHandler;
import org.apache.hadoop.hdds.scm.container.IncrementalContainerReportHandler;
import org.apache.hadoop.hdds.scm.container.SCMContainerManager;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.metrics.ContainerStat;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMMetrics;
import org.apache.hadoop.hdds.scm.container.ReplicationManager;
@@ -392,7 +392,7 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl
}
placementMetrics = SCMContainerPlacementMetrics.create();
- ContainerPlacementPolicy containerPlacementPolicy =
+ PlacementPolicy containerPlacementPolicy =
ContainerPlacementPolicyFactory.getPolicy(conf, scmNodeManager,
clusterMap, true, placementMetrics);
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManager.java
index dadb309..87d7655 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/TestReplicationManager.java
@@ -27,8 +27,7 @@ import org.apache.hadoop.hdds.protocol.proto
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
import org.apache.hadoop.hdds.scm.container.ReplicationManager.ReplicationManagerConfiguration;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms
- .ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.hdds.server.events.EventHandler;
@@ -66,7 +65,7 @@ public class TestReplicationManager {
private ReplicationManager replicationManager;
private ContainerStateManager containerStateManager;
- private ContainerPlacementPolicy containerPlacementPolicy;
+ private PlacementPolicy containerPlacementPolicy;
private EventQueue eventQueue;
private DatanodeCommandHandler datanodeCommandHandler;
@@ -93,7 +92,7 @@ public class TestReplicationManager {
.thenAnswer(invocation -> containerStateManager
.getContainerReplicas((ContainerID)invocation.getArguments()[0]));
- containerPlacementPolicy = Mockito.mock(ContainerPlacementPolicy.class);
+ containerPlacementPolicy = Mockito.mock(PlacementPolicy.class);
Mockito.when(containerPlacementPolicy.chooseDatanodes(
Mockito.anyListOf(DatanodeDetails.class),
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java
index b685ba9..a454de2 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/placement/algorithms/TestContainerPlacementFactory.java
@@ -21,6 +21,7 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeState;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.placement.metrics.SCMNodeMetric;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
@@ -99,7 +100,7 @@ public class TestContainerPlacementFactory {
when(nodeManager.getNodeStat(datanodes.get(4)))
.thenReturn(new SCMNodeMetric(storageCapacity, 70L, 30L));
- ContainerPlacementPolicy policy = ContainerPlacementPolicyFactory
+ PlacementPolicy policy = ContainerPlacementPolicyFactory
.getPolicy(conf, nodeManager, cluster, true,
SCMContainerPlacementMetrics.create());
@@ -117,7 +118,7 @@ public class TestContainerPlacementFactory {
@Test
public void testDefaultPolicy() throws IOException {
- ContainerPlacementPolicy policy = ContainerPlacementPolicyFactory
+ PlacementPolicy policy = ContainerPlacementPolicyFactory
.getPolicy(conf, null, null, true, null);
Assert.assertSame(SCMContainerPlacementRandom.class, policy.getClass());
}
@@ -125,7 +126,7 @@ public class TestContainerPlacementFactory {
/**
* A dummy container placement implementation for test.
*/
- public static class DummyImpl implements ContainerPlacementPolicy {
+ public static class DummyImpl implements PlacementPolicy {
@Override
public List<DatanodeDetails> chooseDatanodes(
List<DatanodeDetails> excludedNodes, List<DatanodeDetails> favoredNodes,
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestContainerPlacement.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestContainerPlacement.java
index 3905823..8808546 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestContainerPlacement.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestContainerPlacement.java
@@ -27,8 +27,7 @@ import org.apache.hadoop.hdds.scm.TestUtils;
import org.apache.hadoop.hdds.scm.XceiverClientManager;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.SCMContainerManager;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms
- .ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms
.SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
@@ -135,7 +134,7 @@ public class TestContainerPlacement {
conf.set(HddsConfigKeys.OZONE_METADATA_DIRS,
testDir.getAbsolutePath());
conf.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
SCMNodeManager nodeManager = createNodeManager(conf);
SCMContainerManager containerManager =
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeHandler.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeHandler.java
index 5572e9a..4ad3456 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeHandler.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/safemode/TestSafeModeHandler.java
@@ -25,8 +25,7 @@ import org.apache.hadoop.hdds.scm.block.BlockManagerImpl;
import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.hdds.scm.container.ReplicationManager;
import org.apache.hadoop.hdds.scm.container.ReplicationManager.ReplicationManagerConfiguration;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms
- .ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.events.SCMEvents;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import org.apache.hadoop.hdds.scm.pipeline.SCMPipelineManager;
@@ -70,7 +69,7 @@ public class TestSafeModeHandler {
.thenReturn(new HashSet<>());
replicationManager = new ReplicationManager(
new ReplicationManagerConfiguration(),
- containerManager, Mockito.mock(ContainerPlacementPolicy.class),
+ containerManager, Mockito.mock(PlacementPolicy.class),
eventQueue, new LockManager(configuration));
scmPipelineManager = Mockito.mock(SCMPipelineManager.class);
blockManager = Mockito.mock(BlockManagerImpl.class);
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java
index eadb520..17c090d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java
@@ -17,11 +17,10 @@
*/
package org.apache.hadoop.ozone;
-import org.apache.hadoop.conf.StorageUnit;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.client.ContainerOperationClient;
@@ -30,7 +29,6 @@ import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CONTAINER_SIZE;
import static org.junit.Assert.assertEquals;
/**
@@ -47,8 +45,7 @@ public class TestContainerOperations {
public static void setup() throws Exception {
ozoneConf = new OzoneConfiguration();
ozoneConf.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
- ozoneConf.setStorageSize(OZONE_SCM_CONTAINER_SIZE, 5, StorageUnit.GB);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
cluster = MiniOzoneCluster.newBuilder(ozoneConf).setNumDatanodes(3).build();
storageClient = new ContainerOperationClient(ozoneConf);
cluster.waitForClusterToBeReady();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java
index 548f9b6..b0b3fbf 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/TestContainerStateMachineIdempotency.java
@@ -29,8 +29,7 @@ import org.apache.hadoop.hdds.scm.XceiverClientManager;
import org.apache.hadoop.hdds.scm.XceiverClientSpi;
import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.
- ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.
SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.scm.protocolPB.
@@ -59,7 +58,7 @@ public class TestContainerStateMachineIdempotency {
public static void init() throws Exception {
ozoneConfig = new OzoneConfiguration();
ozoneConfig.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
cluster =
MiniOzoneCluster.newBuilder(ozoneConfig).setNumDatanodes(3).build();
cluster.waitForClusterToBeReady();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scrubber/TestDataScrubber.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scrubber/TestDataScrubber.java
index b18dc77..c0b6730 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scrubber/TestDataScrubber.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/dn/scrubber/TestDataScrubber.java
@@ -30,7 +30,7 @@ import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
import org.apache.hadoop.hdds.scm.container.ContainerManager;
import org.apache.hadoop.hdds.scm.container.ContainerReplica;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.ozone.HddsDatanodeService;
@@ -84,7 +84,7 @@ public class TestDataScrubber {
ozoneConfig = new OzoneConfiguration();
ozoneConfig.set(HddsConfigKeys.HDDS_CONTAINER_REPORT_INTERVAL, "1s");
ozoneConfig.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
ozoneConfig.setBoolean(HddsConfigKeys.HDDS_SCM_SAFEMODE_PIPELINE_CREATION,
false);
cluster = MiniOzoneCluster.newBuilder(ozoneConfig).setNumDatanodes(1)
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestContainerSmallFile.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestContainerSmallFile.java
index 9d187ff..3b6da67 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestContainerSmallFile.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestContainerSmallFile.java
@@ -24,7 +24,7 @@ import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.protocolPB
@@ -60,7 +60,7 @@ public class TestContainerSmallFile {
public static void init() throws Exception {
ozoneConfig = new OzoneConfiguration();
ozoneConfig.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
cluster = MiniOzoneCluster.newBuilder(ozoneConfig).setNumDatanodes(3)
.build();
cluster.waitForClusterToBeReady();
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestGetCommittedBlockLengthAndPutKey.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestGetCommittedBlockLengthAndPutKey.java
index 2f8c755..93659e0 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestGetCommittedBlockLengthAndPutKey.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestGetCommittedBlockLengthAndPutKey.java
@@ -31,8 +31,7 @@ import org.apache.hadoop.hdds.scm.container.common.helpers.
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.container.common.helpers.
StorageContainerException;
-import org.apache.hadoop.hdds.scm.container.placement.algorithms.
- ContainerPlacementPolicy;
+import org.apache.hadoop.hdds.scm.PlacementPolicy;
import org.apache.hadoop.hdds.scm.container.placement.algorithms.
SCMContainerPlacementCapacity;
import org.apache.hadoop.hdds.scm.protocolPB.
@@ -63,7 +62,7 @@ public class TestGetCommittedBlockLengthAndPutKey {
public static void init() throws Exception {
ozoneConfig = new OzoneConfiguration();
ozoneConfig.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY,
- SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class);
+ SCMContainerPlacementCapacity.class, PlacementPolicy.class);
cluster =
MiniOzoneCluster.newBuilder(ozoneConfig).setNumDatanodes(3).build();
cluster.waitForClusterToBeReady();
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org
[hadoop-ozone] 05/12: HDDS-1572 Implement a Pipeline scrubber to
clean up non-OPEN pipeline. (#237)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sammichen pushed a commit to branch HDDS-1564
in repository https://gitbox.apache.org/repos/asf/hadoop-ozone.git
commit a6c1f2a919fa4b97af4f343de6aa967cf540656f
Author: Li Cheng <bl...@gmail.com>
AuthorDate: Wed Nov 27 20:18:50 2019 +0800
HDDS-1572 Implement a Pipeline scrubber to clean up non-OPEN pipeline. (#237)
---
.../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 7 ++++
.../apache/hadoop/hdds/scm/pipeline/Pipeline.java | 31 +++++++++++++++-
hadoop-hdds/common/src/main/proto/hdds.proto | 1 +
.../common/src/main/resources/ozone-default.xml | 12 ++++++
.../scm/pipeline/BackgroundPipelineCreator.java | 8 +++-
.../hadoop/hdds/scm/pipeline/PipelineManager.java | 3 ++
.../hdds/scm/pipeline/SCMPipelineManager.java | 26 +++++++++++++
.../scm/pipeline/MockRatisPipelineProvider.java | 28 ++++++++++++++
.../scm/pipeline/TestRatisPipelineProvider.java | 10 ++++-
.../hdds/scm/pipeline/TestSCMPipelineManager.java | 43 +++++++++++++++++++++-
.../scm/pipeline/TestRatisPipelineProvider.java | 2 +-
11 files changed, 166 insertions(+), 5 deletions(-)
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 7e8c435..845d86f 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
@@ -338,6 +338,13 @@ public final class ScmConfigKeys {
OZONE_SCM_KEY_VALUE_CONTAINER_DELETION_CHOOSING_POLICY =
"ozone.scm.keyvalue.container.deletion-choosing.policy";
+ // Max timeout for pipeline to stay at ALLOCATED state before scrubbed.
+ public static final String OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT =
+ "ozone.scm.pipeline.allocated.timeout";
+
+ public static final String OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT_DEFAULT =
+ "5m";
+
public static final String OZONE_SCM_CONTAINER_CREATION_LEASE_TIMEOUT =
"ozone.scm.container.creation.lease.timeout";
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
index 594fcf7..295156d 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
@@ -56,6 +56,8 @@ public final class Pipeline {
private ThreadLocal<List<DatanodeDetails>> nodesInOrder = new ThreadLocal<>();
// Current reported Leader for the pipeline
private UUID leaderId;
+ // Timestamp for pipeline upon creation
+ private Long creationTimestamp;
/**
* The immutable properties of pipeline object is used in
@@ -70,6 +72,7 @@ public final class Pipeline {
this.factor = factor;
this.state = state;
this.nodeStatus = nodeStatus;
+ this.creationTimestamp = System.currentTimeMillis();
}
/**
@@ -109,6 +112,24 @@ public final class Pipeline {
}
/**
+ * Return the creation time of pipeline.
+ *
+ * @return Creation Timestamp
+ */
+ public Long getCreationTimestamp() {
+ return creationTimestamp;
+ }
+
+ /**
+ * Set the creation timestamp. Only for protobuf now.
+ *
+ * @param creationTimestamp
+ */
+ void setCreationTimestamp(Long creationTimestamp) {
+ this.creationTimestamp = creationTimestamp;
+ }
+
+ /**
* Return the pipeline leader's UUID.
*
* @return DatanodeDetails.UUID.
@@ -221,6 +242,7 @@ public final class Pipeline {
.setFactor(factor)
.setState(PipelineState.getProtobuf(state))
.setLeaderID(leaderId != null ? leaderId.toString() : "")
+ .setCreationTimeStamp(creationTimestamp)
.addAllMembers(nodeStatus.keySet().stream()
.map(DatanodeDetails::getProtoBufMessage)
.collect(Collectors.toList()));
@@ -299,7 +321,8 @@ public final class Pipeline {
b.append(", Factor:").append(getFactor());
b.append(", State:").append(getPipelineState());
b.append(", leaderId:").append(getLeaderId());
- b.append(" ]");
+ b.append(", CreationTimestamp").append(getCreationTimestamp());
+ b.append("]");
return b.toString();
}
@@ -323,6 +346,7 @@ public final class Pipeline {
private List<Integer> nodeOrder = null;
private List<DatanodeDetails> nodesInOrder = null;
private UUID leaderId = null;
+ private Long creationTimestamp = null;
public Builder() {}
@@ -334,6 +358,7 @@ public final class Pipeline {
this.nodeStatus = pipeline.nodeStatus;
this.nodesInOrder = pipeline.nodesInOrder.get();
this.leaderId = pipeline.getLeaderId();
+ this.creationTimestamp = pipeline.getCreationTimestamp();
}
public Builder setId(PipelineID id1) {
@@ -380,6 +405,10 @@ public final class Pipeline {
Preconditions.checkNotNull(nodeStatus);
Pipeline pipeline = new Pipeline(id, type, factor, state, nodeStatus);
pipeline.setLeaderId(leaderId);
+ // overwrite with original creationTimestamp
+ if (creationTimestamp != null) {
+ pipeline.setCreationTimestamp(creationTimestamp);
+ }
if (nodeOrder != null && !nodeOrder.isEmpty()) {
// This branch is for build from ProtoBuf
diff --git a/hadoop-hdds/common/src/main/proto/hdds.proto b/hadoop-hdds/common/src/main/proto/hdds.proto
index 39a01dc..b313604 100644
--- a/hadoop-hdds/common/src/main/proto/hdds.proto
+++ b/hadoop-hdds/common/src/main/proto/hdds.proto
@@ -75,6 +75,7 @@ message Pipeline {
required PipelineID id = 5;
optional string leaderID = 6;
repeated uint32 memberOrders = 7;
+ optional uint64 creationTimeStamp = 8;
}
message KeyValue {
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index 3093292..653d530 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -856,6 +856,18 @@
</description>
</property>
<property>
+ <name>ozone.scm.pipeline.allocated.timeout</name>
+ <value>5m</value>
+ <tag>OZONE, SCM, PIPELINE</tag>
+ <description>
+ Timeout for every pipeline to stay in ALLOCATED stage. When pipeline is created,
+ it should be at OPEN stage once pipeline report is successfully received by SCM.
+ If a pipeline stays at ALLOCATED for too long, it should be scrubbed so that new
+ pipeline can be created. This timeout is for how long pipeline can stay at ALLOCATED
+ stage until it gets scrubbed.
+ </description>
+ </property>
+ <property>
<name>ozone.scm.container.size</name>
<value>5GB</value>
<tag>OZONE, PERFORMANCE, MANAGEMENT</tag>
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
index b663f2a..8e4ec6a 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreator.java
@@ -110,12 +110,18 @@ class BackgroundPipelineCreator {
// Skip this iteration for creating pipeline
continue;
}
+
+ try {
+ pipelineManager.scrubPipeline(type, factor);
+ } catch (IOException e) {
+ LOG.error("Error while scrubbing pipelines {}", e);
+ }
+
while (true) {
try {
if (scheduler.isClosed()) {
break;
}
-
pipelineManager.createPipeline(type, factor);
} catch (IOException ioe) {
break;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
index 779008f..44432d9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
@@ -75,6 +75,9 @@ public interface PipelineManager extends Closeable, PipelineManagerMXBean {
void finalizeAndDestroyPipeline(Pipeline pipeline, boolean onTimeout)
throws IOException;
+ void scrubPipeline(ReplicationType type, ReplicationFactor factor)
+ throws IOException;
+
void startPipelineCreator();
void triggerPipelineCreation();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
index 1b2c476..4346dea 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
@@ -54,6 +54,7 @@ import java.util.Collection;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.stream.Collectors;
import static org.apache.hadoop.ozone.OzoneConsts.SCM_PIPELINE_DB;
@@ -341,6 +342,31 @@ public class SCMPipelineManager implements PipelineManager {
}
@Override
+ public void scrubPipeline(ReplicationType type, ReplicationFactor factor)
+ throws IOException{
+ if (type != ReplicationType.RATIS || factor != ReplicationFactor.THREE) {
+ // Only srub pipeline for RATIS THREE pipeline
+ return;
+ }
+ Long currentTime = System.currentTimeMillis();
+ Long pipelineScrubTimeoutInMills = conf.getTimeDuration(
+ ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT,
+ ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT_DEFAULT,
+ TimeUnit.MILLISECONDS);
+ List<Pipeline> needToSrubPipelines = stateManager.getPipelines(type, factor,
+ Pipeline.PipelineState.ALLOCATED).stream()
+ .filter(p -> (currentTime - p.getCreationTimestamp()
+ >= pipelineScrubTimeoutInMills))
+ .collect(Collectors.toList());
+ for (Pipeline p : needToSrubPipelines) {
+ LOG.info("srubbing pipeline: id: " + p.getId().toString() +
+ " since it stays at ALLOCATED stage for " +
+ (currentTime - p.getCreationTimestamp())/60000 + " mins.");
+ finalizeAndDestroyPipeline(p, false);
+ }
+ }
+
+ @Override
public Map<String, Integer> getPipelineInfo() {
final Map<String, Integer> pipelineInfo = new HashMap<>();
for (Pipeline.PipelineState state : Pipeline.PipelineState.values()) {
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
index 25b0adc..7513cad 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockRatisPipelineProvider.java
@@ -33,6 +33,15 @@ import java.util.List;
*/
public class MockRatisPipelineProvider extends RatisPipelineProvider {
+ private boolean autoOpenPipeline;
+
+ public MockRatisPipelineProvider(NodeManager nodeManager,
+ PipelineStateManager stateManager,
+ Configuration conf, boolean autoOpen) {
+ super(nodeManager, stateManager, conf, null);
+ autoOpenPipeline = autoOpen;
+ }
+
public MockRatisPipelineProvider(NodeManager nodeManager,
PipelineStateManager stateManager,
Configuration conf) {
@@ -43,6 +52,7 @@ public class MockRatisPipelineProvider extends RatisPipelineProvider {
PipelineStateManager stateManager, Configuration conf,
EventPublisher eventPublisher) {
super(nodeManager, stateManager, conf, eventPublisher);
+ autoOpenPipeline = true;
}
protected void initializePipeline(Pipeline pipeline) throws IOException {
@@ -50,6 +60,24 @@ public class MockRatisPipelineProvider extends RatisPipelineProvider {
}
@Override
+ public Pipeline create(HddsProtos.ReplicationFactor factor)
+ throws IOException {
+ if (autoOpenPipeline) {
+ return super.create(factor);
+ } else {
+ Pipeline initialPipeline = super.create(factor);
+ return Pipeline.newBuilder()
+ .setId(initialPipeline.getId())
+ // overwrite pipeline state to main ALLOCATED
+ .setState(Pipeline.PipelineState.ALLOCATED)
+ .setType(initialPipeline.getType())
+ .setFactor(factor)
+ .setNodes(initialPipeline.getNodes())
+ .build();
+ }
+ }
+
+ @Override
public void shutdown() {
// Do nothing.
}
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
index f5e3f84..66991e4 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
@@ -22,6 +22,7 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.ScmConfigKeys;
import org.apache.hadoop.hdds.scm.container.MockNodeManager;
import org.apache.hadoop.hdds.scm.node.NodeManager;
import org.junit.Assume;
@@ -53,9 +54,11 @@ public class TestRatisPipelineProvider {
@Before
public void init() throws Exception {
nodeManager = new MockNodeManager(true, 10);
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setInt(ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT, 1);
stateManager = new PipelineStateManager();
provider = new MockRatisPipelineProvider(nodeManager,
- stateManager, new OzoneConfiguration());
+ stateManager, conf);
}
private void createPipelineAndAssertions(
@@ -64,6 +67,7 @@ public class TestRatisPipelineProvider {
assertPipelineProperties(pipeline, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
stateManager.addPipeline(pipeline);
+ nodeManager.addPipeline(pipeline);
Pipeline pipeline1 = provider.create(factor);
assertPipelineProperties(pipeline1, factor, REPLICATION_TYPE,
@@ -149,6 +153,9 @@ public class TestRatisPipelineProvider {
Pipeline pipeline = provider.create(factor);
assertPipelineProperties(pipeline, factor, REPLICATION_TYPE,
Pipeline.PipelineState.ALLOCATED);
+ nodeManager.addPipeline(pipeline);
+ stateManager.addPipeline(pipeline);
+
List<DatanodeDetails> nodes = pipeline.getNodes();
@@ -184,5 +191,6 @@ public class TestRatisPipelineProvider {
.build();
stateManager.addPipeline(openPipeline);
+ nodeManager.addPipeline(openPipeline);
}
}
\ No newline at end of file
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
index 08f5185..2df851d 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestSCMPipelineManager.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hdds.scm.pipeline;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_DATANODE_MAX_PIPELINE_ENGAGEMENT;
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT;
import static org.apache.hadoop.test.MetricsAsserts.getLongCounter;
import static org.apache.hadoop.test.MetricsAsserts.getMetrics;
@@ -28,6 +29,7 @@ import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
@@ -270,7 +272,7 @@ public class TestSCMPipelineManager {
numPipelineCreateFailed = getLongCounter(
"NumPipelineCreationFailed", metrics);
Assert.assertTrue(numPipelineCreateFailed == 1);
-
+
// clean up
pipelineManager.close();
}
@@ -374,6 +376,45 @@ public class TestSCMPipelineManager {
pipelineManager.close();
}
+ @Test
+ public void testScrubPipeline() throws IOException {
+ // No timeout for pipeline scrubber.
+ conf.setTimeDuration(
+ OZONE_SCM_PIPELINE_ALLOCATED_TIMEOUT, -1,
+ TimeUnit.MILLISECONDS);
+
+ final SCMPipelineManager pipelineManager =
+ new SCMPipelineManager(conf, nodeManager, new EventQueue(), null);
+ final PipelineProvider ratisProvider = new MockRatisPipelineProvider(
+ nodeManager, pipelineManager.getStateManager(), conf, false);
+
+ pipelineManager.setPipelineProvider(HddsProtos.ReplicationType.RATIS,
+ ratisProvider);
+
+ Pipeline pipeline = pipelineManager
+ .createPipeline(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE);
+ // At this point, pipeline is not at OPEN stage.
+ Assert.assertEquals(pipeline.getPipelineState(),
+ Pipeline.PipelineState.ALLOCATED);
+
+ // pipeline should be seen in pipelineManager as ALLOCATED.
+ Assert.assertTrue(pipelineManager
+ .getPipelines(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE,
+ Pipeline.PipelineState.ALLOCATED).contains(pipeline));
+ pipelineManager.scrubPipeline(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE);
+
+ // pipeline should be scrubbed.
+ Assert.assertFalse(pipelineManager
+ .getPipelines(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE,
+ Pipeline.PipelineState.ALLOCATED).contains(pipeline));
+
+ pipelineManager.close();
+ }
+
private void sendPipelineReport(DatanodeDetails dn,
Pipeline pipeline, PipelineReportHandler pipelineReportHandler,
boolean isLeader, EventQueue eventQueue) {
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
index ce34ba6..418e86d 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestRatisPipelineProvider.java
@@ -142,7 +142,7 @@ public class TestRatisPipelineProvider {
List<DatanodeDetails> allHealthyNodes =
nodeManager.getNodes(HddsProtos.NodeState.HEALTHY);
int totalHealthyNodesCount = allHealthyNodes.size();
-
+
HddsProtos.ReplicationFactor factor = HddsProtos.ReplicationFactor.THREE;
List<DatanodeDetails> closePipelineDns = new ArrayList<>();
---------------------------------------------------------------------
To unsubscribe, e-mail: ozone-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: ozone-commits-help@hadoop.apache.org