HDFS-12756. Ozone: Add datanodeID to heartbeat responses and container protocol. Contributed by Anu Engineer.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/1a1aec23 Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/1a1aec23 Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/1a1aec23 Branch: refs/heads/HDFS-7240 Commit: 1a1aec23ef0b59321c8ecd4260596e109d1ecb68 Parents: b8297b0 Author: Anu Engineer <[email protected]> Authored: Fri Nov 10 10:26:26 2017 -0800 Committer: Anu Engineer <[email protected]> Committed: Fri Nov 10 10:26:26 2017 -0800 ---------------------------------------------------------------------- .../apache/hadoop/scm/XceiverClientHandler.java | 9 +- .../scm/storage/ContainerProtocolCalls.java | 29 +- .../main/proto/DatanodeContainerProtocol.proto | 1 + .../states/endpoint/HeartbeatEndpointTask.java | 9 +- .../ozone/scm/StorageContainerManager.java | 14 +- .../StorageContainerDatanodeProtocol.proto | 1 + .../apache/hadoop/cblock/TestBufferManager.java | 3 +- .../hadoop/cblock/TestCBlockReadWrite.java | 3 +- .../hadoop/cblock/TestLocalBlockCache.java | 3 +- .../hadoop/ozone/MiniOzoneClassicCluster.java | 556 ++++++++++++++++++ .../apache/hadoop/ozone/MiniOzoneCluster.java | 560 +------------------ .../apache/hadoop/ozone/RatisTestHelper.java | 15 +- .../hadoop/ozone/TestContainerOperations.java | 2 +- .../hadoop/ozone/TestMiniOzoneCluster.java | 8 +- .../ozone/TestStorageContainerManager.java | 24 +- .../TestStorageContainerManagerHelper.java | 4 +- .../ozone/client/rest/TestOzoneRestClient.java | 6 +- .../ozone/client/rpc/TestOzoneRpcClient.java | 19 +- .../ozone/container/ContainerTestHelper.java | 32 +- .../container/metrics/TestContainerMetrics.java | 2 +- .../container/ozoneimpl/TestOzoneContainer.java | 18 +- .../ozoneimpl/TestOzoneContainerRatis.java | 5 +- .../container/ozoneimpl/TestRatisManager.java | 5 +- .../transport/server/TestContainerServer.java | 9 +- .../apache/hadoop/ozone/ksm/TestKSMMetrcis.java | 3 +- .../apache/hadoop/ozone/ksm/TestKSMSQLCli.java | 6 +- .../hadoop/ozone/ksm/TestKeySpaceManager.java | 3 +- .../ksm/TestMultipleContainerReadWrite.java | 3 +- .../hadoop/ozone/ozShell/TestOzoneShell.java | 6 +- .../hadoop/ozone/scm/TestAllocateContainer.java | 3 +- .../hadoop/ozone/scm/TestContainerSQLCli.java | 6 +- .../ozone/scm/TestContainerSmallFile.java | 3 +- .../org/apache/hadoop/ozone/scm/TestSCMCli.java | 6 +- .../apache/hadoop/ozone/scm/TestSCMMXBean.java | 3 +- .../apache/hadoop/ozone/scm/TestSCMMetrics.java | 3 +- .../ozone/scm/TestXceiverClientManager.java | 3 +- .../ozone/scm/TestXceiverClientMetrics.java | 5 +- .../container/TestContainerStateManager.java | 3 +- .../hadoop/ozone/scm/node/TestQueryNode.java | 6 +- .../apache/hadoop/ozone/tools/TestCorona.java | 3 +- .../ozone/web/TestDistributedOzoneVolumes.java | 6 +- .../hadoop/ozone/web/TestLocalOzoneVolumes.java | 6 +- .../ozone/web/TestOzoneRestWithMiniCluster.java | 3 +- .../hadoop/ozone/web/TestOzoneWebAccess.java | 6 +- .../hadoop/ozone/web/client/TestBuckets.java | 6 +- .../hadoop/ozone/web/client/TestKeys.java | 10 +- .../hadoop/ozone/web/client/TestKeysRatis.java | 4 +- .../ozone/web/client/TestOzoneClient.java | 6 +- .../hadoop/ozone/web/client/TestVolume.java | 6 +- .../ozone/web/client/TestVolumeRatis.java | 6 +- .../fs/ozone/TestOzoneFileInterfaces.java | 6 +- .../hadoop/fs/ozone/contract/OzoneContract.java | 6 +- 52 files changed, 783 insertions(+), 690 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientHandler.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientHandler.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientHandler.java index 06b7c5c..1d91b14 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientHandler.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientHandler.java @@ -50,7 +50,6 @@ public class XceiverClientHandler extends private final Pipeline pipeline; private volatile Channel channel; - private XceiverClientMetrics metrics; /** @@ -58,6 +57,7 @@ public class XceiverClientHandler extends */ public XceiverClientHandler(Pipeline pipeline) { super(false); + Preconditions.checkNotNull(pipeline); this.pipeline = pipeline; this.metrics = XceiverClientManager.getXceiverClientMetrics(); } @@ -139,6 +139,13 @@ public class XceiverClientHandler extends if (StringUtils.isEmpty(request.getTraceID())) { throw new IllegalArgumentException("Invalid trace ID"); } + + // Setting the datanode ID in the commands, so that we can distinguish + // commands when the cluster simulator is running. + if(!request.hasDatanodeID()) { + throw new IllegalArgumentException("Invalid Datanode ID"); + } + metrics.incrPendingContainerOpsMetrics(request.getCmdType()); CompletableFuture<ContainerCommandResponseProto> future http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/storage/ContainerProtocolCalls.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/storage/ContainerProtocolCalls.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/storage/ContainerProtocolCalls.java index 0e992a5..1cde67c 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/storage/ContainerProtocolCalls.java +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/storage/ContainerProtocolCalls.java @@ -82,10 +82,12 @@ public final class ContainerProtocolCalls { .newBuilder() .setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setKeyData(containerKeyData); + String id = xceiverClient.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.GetKey) .setTraceID(traceID) + .setDatanodeID(id) .setGetKey(readKeyRequest) .build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); @@ -107,10 +109,12 @@ public final class ContainerProtocolCalls { .newBuilder() .setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setKeyData(containerKeyData); + String id = xceiverClient.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.PutKey) .setTraceID(traceID) + .setDatanodeID(id) .setPutKey(createKeyRequest) .build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); @@ -135,10 +139,12 @@ public final class ContainerProtocolCalls { .setPipeline(xceiverClient.getPipeline().getProtobufMessage()) .setKeyName(key) .setChunkData(chunk); + String id = xceiverClient.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.ReadChunk) .setTraceID(traceID) + .setDatanodeID(id) .setReadChunk(readChunkRequest) .build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); @@ -165,10 +171,12 @@ public final class ContainerProtocolCalls { .setKeyName(key) .setChunkData(chunk) .setData(data); + String id = xceiverClient.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.WriteChunk) .setTraceID(traceID) + .setDatanodeID(id) .setWriteChunk(writeChunkRequest) .build(); ContainerCommandResponseProto response = xceiverClient.sendCommand(request); @@ -212,9 +220,14 @@ public final class ContainerProtocolCalls { .setKey(createKeyRequest).setData(ByteString.copyFrom(data)) .build(); + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = - ContainerCommandRequestProto.newBuilder().setCmdType(Type.PutSmallFile) - .setTraceID(traceID).setPutSmallFile(putSmallFileRequest).build(); + ContainerCommandRequestProto.newBuilder() + .setCmdType(Type.PutSmallFile) + .setTraceID(traceID) + .setDatanodeID(id) + .setPutSmallFile(putSmallFileRequest) + .build(); ContainerCommandResponseProto response = client.sendCommand(request); validateContainerResponse(response); } @@ -236,10 +249,12 @@ public final class ContainerProtocolCalls { createRequest.setPipeline(client.getPipeline().getProtobufMessage()); createRequest.setContainerData(containerData.build()); + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto.Builder request = ContainerCommandRequestProto.newBuilder(); request.setCmdType(ContainerProtos.Type.CreateContainer); request.setCreateContainer(createRequest); + request.setDatanodeID(id); request.setTraceID(traceID); ContainerCommandResponseProto response = client.sendCommand( request.build()); @@ -261,12 +276,13 @@ public final class ContainerProtocolCalls { deleteRequest.setName(client.getPipeline().getContainerName()); deleteRequest.setPipeline(client.getPipeline().getProtobufMessage()); deleteRequest.setForceDelete(force); - + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto.Builder request = ContainerCommandRequestProto.newBuilder(); request.setCmdType(ContainerProtos.Type.DeleteContainer); request.setDeleteContainer(deleteRequest); request.setTraceID(traceID); + request.setDatanodeID(id); ContainerCommandResponseProto response = client.sendCommand(request.build()); validateContainerResponse(response); @@ -285,11 +301,13 @@ public final class ContainerProtocolCalls { ContainerProtos.CloseContainerRequestProto.newBuilder(); closeRequest.setPipeline(client.getPipeline().getProtobufMessage()); + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto.Builder request = ContainerCommandRequestProto.newBuilder(); request.setCmdType(Type.CloseContainer); request.setCloseContainer(closeRequest); request.setTraceID(traceID); + request.setDatanodeID(id); ContainerCommandResponseProto response = client.sendCommand(request.build()); validateContainerResponse(response); @@ -309,11 +327,12 @@ public final class ContainerProtocolCalls { ReadContainerRequestProto.newBuilder(); readRequest.setName(containerName); readRequest.setPipeline(client.getPipeline().getProtobufMessage()); - + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto.Builder request = ContainerCommandRequestProto.newBuilder(); request.setCmdType(Type.ReadContainer); request.setReadContainer(readRequest); + request.setDatanodeID(id); request.setTraceID(traceID); ContainerCommandResponseProto response = client.sendCommand(request.build()); @@ -346,10 +365,12 @@ public final class ContainerProtocolCalls { GetSmallFileRequestProto .newBuilder().setKey(getKey) .build(); + String id = client.getPipeline().getLeader().getDatanodeUuid(); ContainerCommandRequestProto request = ContainerCommandRequestProto .newBuilder() .setCmdType(Type.GetSmallFile) .setTraceID(traceID) + .setDatanodeID(id) .setGetSmallFile(getSmallFileRequest) .build(); ContainerCommandResponseProto response = client.sendCommand(request); http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/DatanodeContainerProtocol.proto ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/DatanodeContainerProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/DatanodeContainerProtocol.proto index 7fb92c9..00a8bf9 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/DatanodeContainerProtocol.proto +++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/proto/DatanodeContainerProtocol.proto @@ -164,6 +164,7 @@ message ContainerCommandRequestProto { optional PutSmallFileRequestProto putSmallFile = 16; optional GetSmallFileRequestProto getSmallFile = 17; optional CloseContainerRequestProto closeContainer = 18; + required string datanodeID = 19; } message ContainerCommandResponseProto { http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java index 49cc3a5..2177a82 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java @@ -103,7 +103,7 @@ public class HeartbeatEndpointTask SCMHeartbeatResponseProto reponse = rpcEndpoint.getEndPoint() .sendHeartbeat(datanodeID, this.context.getNodeReport(), this.context.getContainerReportState()); - processResponse(reponse); + processResponse(reponse, datanodeID); rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now()); rpcEndpoint.zeroMissedCount(); } catch (IOException ex) { @@ -127,9 +127,14 @@ public class HeartbeatEndpointTask * * @param response - SCMHeartbeat response. */ - private void processResponse(SCMHeartbeatResponseProto response) { + private void processResponse(SCMHeartbeatResponseProto response, + final DatanodeID datanodeID) { for (SCMCommandResponseProto commandResponseProto : response .getCommandsList()) { + // Verify the response is indeed for this datanode. + Preconditions.checkState(commandResponseProto.getDatanodeUUID() + .equalsIgnoreCase(datanodeID.getDatanodeUuid().toString()), + "Unexpected datanode ID in the response."); switch (commandResponseProto.getCmdType()) { case sendContainerReport: this.context.addCommand(SendContainerCommand.getFromProtobuf( http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java index 1f86bd0..a9f9360 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/StorageContainerManager.java @@ -226,8 +226,9 @@ public class StorageContainerManager extends ServiceRuntimeInfoImpl StorageContainerManager.initMetrics(); scmStorage = new SCMStorage(conf); - if (scmStorage.getState() != StorageState.INITIALIZED) { - throw new SCMException("SCM not initialized.", + String clusterId = scmStorage.getClusterID(); + if (clusterId == null) { + throw new SCMException("clusterId not found", ResultCodes.SCM_NOT_INITIALIZED); } scmNodeManager = new SCMNodeManager(conf, scmStorage.getClusterID()); @@ -492,11 +493,13 @@ public class StorageContainerManager extends ServiceRuntimeInfoImpl * @throws InvalidProtocolBufferException */ @VisibleForTesting - public SCMCommandResponseProto getCommandResponse(SCMCommand cmd) + public SCMCommandResponseProto getCommandResponse(SCMCommand cmd, + final String datanodID) throws IOException { Type type = cmd.getType(); SCMCommandResponseProto.Builder builder = - SCMCommandResponseProto.newBuilder(); + SCMCommandResponseProto.newBuilder() + .setDatanodeUUID(datanodID); switch (type) { case registeredCommand: return builder.setCmdType(Type.registeredCommand) @@ -881,7 +884,8 @@ public class StorageContainerManager extends ServiceRuntimeInfoImpl getScmNodeManager().sendHeartbeat(datanodeID, nodeReport, reportState); List<SCMCommandResponseProto> cmdResponses = new LinkedList<>(); for (SCMCommand cmd : commands) { - cmdResponses.add(getCommandResponse(cmd)); + cmdResponses.add(getCommandResponse(cmd, datanodeID.getDatanodeUuid() + .toString())); } return SCMHeartbeatResponseProto.newBuilder().addAllCommands(cmdResponses) .build(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/StorageContainerDatanodeProtocol.proto ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/StorageContainerDatanodeProtocol.proto b/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/StorageContainerDatanodeProtocol.proto index 2cbe37e..0fb4bbd 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/StorageContainerDatanodeProtocol.proto +++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/proto/StorageContainerDatanodeProtocol.proto @@ -231,6 +231,7 @@ message SCMCommandResponseProto { optional SendContainerReportProto sendReport = 5; optional SCMReregisterCmdResponseProto reregisterProto = 6; optional SCMDeleteBlocksCmdResponseProto deleteBlocksProto = 7; + required string datanodeUUID = 8; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestBufferManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestBufferManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestBufferManager.java index 4211255..76ee52f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestBufferManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestBufferManager.java @@ -23,6 +23,7 @@ import org.apache.hadoop.cblock.jscsiHelper.CBlockTargetMetrics; import org.apache.hadoop.cblock.jscsiHelper.ContainerCacheFlusher; import org.apache.hadoop.cblock.jscsiHelper.cache.impl.CBlockLocalCache; import org.apache.hadoop.io.IOUtils; +import org.apache.hadoop.ozone.MiniOzoneClassicCluster; import org.apache.hadoop.ozone.MiniOzoneCluster; import org.apache.hadoop.conf.OzoneConfiguration; import org.apache.hadoop.scm.XceiverClientManager; @@ -71,7 +72,7 @@ public class TestBufferManager { config.set(DFS_CBLOCK_DISK_CACHE_PATH_KEY, path); config.setBoolean(DFS_CBLOCK_TRACE_IO, true); config.setBoolean(DFS_CBLOCK_ENABLE_SHORT_CIRCUIT_IO, true); - cluster = new MiniOzoneCluster.Builder(config) + cluster = new MiniOzoneClassicCluster.Builder(config) .numDataNodes(1).setHandlerType("distributed").build(); storageContainerLocationClient = cluster .createStorageContainerLocationClient(); http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestCBlockReadWrite.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestCBlockReadWrite.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestCBlockReadWrite.java index e0389de..f194385 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestCBlockReadWrite.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestCBlockReadWrite.java @@ -25,6 +25,7 @@ import org.apache.hadoop.cblock.jscsiHelper.ContainerCacheFlusher; import org.apache.hadoop.cblock.jscsiHelper.cache.LogicalBlock; import org.apache.hadoop.cblock.jscsiHelper.cache.impl.CBlockLocalCache; import org.apache.hadoop.io.IOUtils; +import org.apache.hadoop.ozone.MiniOzoneClassicCluster; import org.apache.hadoop.ozone.MiniOzoneCluster; import org.apache.hadoop.conf.OzoneConfiguration; import org.apache.hadoop.ozone.OzoneConsts; @@ -77,7 +78,7 @@ public class TestCBlockReadWrite { config.set(DFS_CBLOCK_DISK_CACHE_PATH_KEY, path); config.setBoolean(DFS_CBLOCK_TRACE_IO, true); config.setBoolean(DFS_CBLOCK_ENABLE_SHORT_CIRCUIT_IO, true); - cluster = new MiniOzoneCluster.Builder(config) + cluster = new MiniOzoneClassicCluster.Builder(config) .numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); storageContainerLocationClient = cluster http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestLocalBlockCache.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestLocalBlockCache.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestLocalBlockCache.java index ea7c8dd0..58b94d7 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestLocalBlockCache.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/cblock/TestLocalBlockCache.java @@ -26,6 +26,7 @@ import org.apache.hadoop.cblock.jscsiHelper.ContainerCacheFlusher; import org.apache.hadoop.cblock.jscsiHelper.cache.LogicalBlock; import org.apache.hadoop.cblock.jscsiHelper.cache.impl.CBlockLocalCache; import org.apache.hadoop.io.IOUtils; +import org.apache.hadoop.ozone.MiniOzoneClassicCluster; import org.apache.hadoop.ozone.MiniOzoneCluster; import org.apache.hadoop.conf.OzoneConfiguration; import org.apache.hadoop.ozone.OzoneConsts; @@ -83,7 +84,7 @@ public class TestLocalBlockCache { config.set(DFS_CBLOCK_DISK_CACHE_PATH_KEY, path); config.setBoolean(DFS_CBLOCK_TRACE_IO, true); config.setBoolean(DFS_CBLOCK_ENABLE_SHORT_CIRCUIT_IO, true); - cluster = new MiniOzoneCluster.Builder(config) + cluster = new MiniOzoneClassicCluster.Builder(config) .numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); storageContainerLocationClient = cluster http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneClassicCluster.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneClassicCluster.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneClassicCluster.java new file mode 100644 index 0000000..65295bb --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneClassicCluster.java @@ -0,0 +1,556 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.hadoop.ozone; + +import java.io.File; +import java.util.Optional; +import com.google.common.base.Preconditions; +import org.apache.commons.io.FileUtils; +import org.apache.hadoop.classification.InterfaceAudience; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.conf.OzoneConfiguration; +import org.apache.hadoop.hdfs.MiniDFSCluster; +import org.apache.hadoop.ipc.Client; +import org.apache.hadoop.ipc.RPC; +import org.apache.hadoop.net.NetUtils; +import org.apache.hadoop.ozone.client.rest.OzoneException; +import org.apache.hadoop.ozone.container.common + .statemachine.DatanodeStateMachine; +import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; +import org.apache.hadoop.ozone.ksm.KSMConfigKeys; +import org.apache.hadoop.ozone.ksm.KeySpaceManager; +import org.apache.hadoop.ozone.scm.SCMStorage; +import org.apache.hadoop.ozone.web.client.OzoneRestClient; +import org.apache.hadoop.scm.ScmConfigKeys; +import org.apache.hadoop.scm.protocolPB + .StorageContainerLocationProtocolClientSideTranslatorPB; +import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolPB; +import org.apache.hadoop.ozone.scm.StorageContainerManager; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.test.GenericTestUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.slf4j.event.Level; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.URISyntaxException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Random; +import java.util.UUID; +import java.util.concurrent.TimeoutException; + +import static org.apache.hadoop.ozone.OzoneConfigKeys + .DFS_CONTAINER_IPC_PORT; +import static org.apache.hadoop.ozone.OzoneConfigKeys + .DFS_CONTAINER_IPC_RANDOM_PORT; +import static org.apache.hadoop.ozone.OzoneConfigKeys + .DFS_CONTAINER_RATIS_IPC_PORT; +import static org.apache.hadoop.ozone.OzoneConfigKeys + .DFS_CONTAINER_RATIS_IPC_RANDOM_PORT; + +import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos.NodeState + .HEALTHY; +import static org.junit.Assert.assertFalse; + +/** + * MiniOzoneCluster creates a complete in-process Ozone cluster suitable for + * running tests. The cluster consists of a StorageContainerManager, Namenode + * and multiple DataNodes. This class subclasses {@link MiniDFSCluster} for + * convenient reuse of logic for starting DataNodes. + */ [email protected] +public final class MiniOzoneClassicCluster extends MiniDFSCluster + implements MiniOzoneCluster { + private static final Logger LOG = + LoggerFactory.getLogger(MiniOzoneClassicCluster.class); + private static final String USER_AUTH = "hdfs"; + + private final OzoneConfiguration conf; + private final StorageContainerManager scm; + private final KeySpaceManager ksm; + private final Path tempPath; + + /** + * Creates a new MiniOzoneCluster. + * + * @param builder cluster builder + * @param scm StorageContainerManager, already running + * @throws IOException if there is an I/O error + */ + private MiniOzoneClassicCluster(Builder builder, StorageContainerManager scm, + KeySpaceManager ksm) + throws IOException { + super(builder); + this.conf = builder.conf; + this.scm = scm; + this.ksm = ksm; + tempPath = Paths.get(builder.getPath(), builder.getRunID()); + } + + + @Override + protected void setupDatanodeAddress( + int i, Configuration dnConf, boolean setupHostsFile, + boolean checkDnAddrConf) throws IOException { + super.setupDatanodeAddress(i, dnConf, setupHostsFile, checkDnAddrConf); + setConf(i, dnConf, OzoneConfigKeys.DFS_CONTAINER_RATIS_DATANODE_STORAGE_DIR, + getInstanceStorageDir(i, -1).getCanonicalPath()); + String containerMetaDirs = dnConf.get( + OzoneConfigKeys.OZONE_METADATA_DIRS) + "-dn-" + i; + Path containerMetaDirPath = Paths.get(containerMetaDirs); + setConf(i, dnConf, OzoneConfigKeys.OZONE_METADATA_DIRS, + containerMetaDirs); + Path containerRootPath = + containerMetaDirPath.resolve(OzoneConsts.CONTAINER_ROOT_PREFIX); + Files.createDirectories(containerRootPath); + } + + static void setConf(int i, Configuration conf, String key, String value) { + conf.set(key, value); + LOG.info("dn{}: set {} = {}", i, key, value); + } + + @Override + public void close() { + shutdown(); + try { + FileUtils.deleteDirectory(tempPath.toFile()); + } catch (IOException e) { + String errorMessage = "Cleaning up metadata directories failed." + e; + assertFalse(errorMessage, true); + } + + try { + final String localStorage = + conf.getTrimmed(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT, + OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT); + FileUtils.deleteDirectory(new File(localStorage)); + } catch (IOException e) { + LOG.error("Cleaning up local storage failed", e); + } + } + + @Override + public boolean restartDataNode(int i) throws IOException { + return restartDataNode(i, true); + } + /* + * Restart a particular datanode, wait for it to become active + */ + @Override + public boolean restartDataNode(int i, boolean keepPort) throws IOException { + LOG.info("restarting datanode:{} keepPort:{}", i, keepPort); + if (keepPort) { + DataNodeProperties dnProp = dataNodes.get(i); + OzoneContainer container = + dnProp.getDatanode().getOzoneContainerManager(); + Configuration config = dnProp.getConf(); + int currentPort = container.getContainerServerPort(); + config.setInt(DFS_CONTAINER_IPC_PORT, currentPort); + config.setBoolean(DFS_CONTAINER_IPC_RANDOM_PORT, false); + int ratisPort = container.getRatisContainerServerPort(); + config.setInt(DFS_CONTAINER_RATIS_IPC_PORT, ratisPort); + config.setBoolean(DFS_CONTAINER_RATIS_IPC_RANDOM_PORT, false); + } + boolean status = super.restartDataNode(i, keepPort); + + try { + this.waitActive(); + waitDatanodeOzoneReady(i); + } catch (TimeoutException | InterruptedException e) { + Thread.interrupted(); + } + return status; + } + + @Override + public void shutdown() { + super.shutdown(); + LOG.info("Shutting down the Mini Ozone Cluster"); + + if (ksm != null) { + LOG.info("Shutting down the keySpaceManager"); + ksm.stop(); + ksm.join(); + } + + if (scm != null) { + LOG.info("Shutting down the StorageContainerManager"); + scm.stop(); + scm.join(); + } + } + + @Override + public StorageContainerManager getStorageContainerManager() { + return this.scm; + } + + @Override + public KeySpaceManager getKeySpaceManager() { + return this.ksm; + } + + /** + * Creates an {@link OzoneRestClient} connected to this cluster's REST + * service. Callers take ownership of the client and must close it when done. + * + * @return OzoneRestClient connected to this cluster's REST service + * @throws OzoneException if Ozone encounters an error creating the client + */ + @Override + public OzoneRestClient createOzoneRestClient() throws OzoneException { + Preconditions.checkState(!getDataNodes().isEmpty(), + "Cannot create OzoneRestClient if the cluster has no DataNodes."); + // An Ozone request may originate at any DataNode, so pick one at random. + int dnIndex = new Random().nextInt(getDataNodes().size()); + String uri = String.format("http://127.0.0.1:%d";, + getDataNodes().get(dnIndex).getInfoPort()); + LOG.info("Creating Ozone client to DataNode {} with URI {} and user {}", + dnIndex, uri, USER_AUTH); + try { + return new OzoneRestClient(uri, USER_AUTH); + } catch (URISyntaxException e) { + // We control the REST service URI, so it should never be invalid. + throw new IllegalStateException("Unexpected URISyntaxException", e); + } + } + + /** + * Creates an RPC proxy connected to this cluster's StorageContainerManager + * for accessing container location information. Callers take ownership of + * the proxy and must close it when done. + * + * @return RPC proxy for accessing container location information + * @throws IOException if there is an I/O error + */ + @Override + public StorageContainerLocationProtocolClientSideTranslatorPB + createStorageContainerLocationClient() throws IOException { + long version = RPC.getProtocolVersion( + StorageContainerLocationProtocolPB.class); + InetSocketAddress address = scm.getClientRpcAddress(); + LOG.info( + "Creating StorageContainerLocationProtocol RPC client with address {}", + address); + return new StorageContainerLocationProtocolClientSideTranslatorPB( + RPC.getProxy(StorageContainerLocationProtocolPB.class, version, + address, UserGroupInformation.getCurrentUser(), conf, + NetUtils.getDefaultSocketFactory(conf), + Client.getRpcTimeout(conf))); + } + + /** + * Waits for the Ozone cluster to be ready for processing requests. + */ + @Override + public void waitOzoneReady() throws TimeoutException, InterruptedException { + GenericTestUtils.waitFor(() -> { + final int healthy = scm.getNodeCount(HEALTHY); + final boolean isReady = healthy >= numDataNodes; + LOG.info("{}. Got {} of {} DN Heartbeats.", + isReady? "Cluster is ready" : "Waiting for cluster to be ready", + healthy, numDataNodes); + return isReady; + }, 1000, 60 * 1000); //wait for 1 min. + } + + /** + * Waits for a particular Datanode to be ready for processing ozone requests. + */ + @Override + public void waitDatanodeOzoneReady(int dnIndex) + throws TimeoutException, InterruptedException { + GenericTestUtils.waitFor(() -> { + DatanodeStateMachine.DatanodeStates state = + dataNodes.get(dnIndex).getDatanode().getOzoneStateMachineState(); + final boolean rebootComplete = + (state == DatanodeStateMachine.DatanodeStates.RUNNING); + LOG.info("{} Current state:{}", rebootComplete, state); + return rebootComplete; + }, 1000, 60 * 1000); //wait for 1 min. + } + + /** + * Waits for SCM to be out of Chill Mode. Many tests can be run iff we are out + * of Chill mode. + * + * @throws TimeoutException + * @throws InterruptedException + */ + @Override + public void waitTobeOutOfChillMode() throws TimeoutException, + InterruptedException { + GenericTestUtils.waitFor(() -> { + if (scm.getScmNodeManager().isOutOfChillMode()) { + return true; + } + LOG.info("Waiting for cluster to be ready. No datanodes found"); + return false; + }, 100, 45000); + } + + @Override + public void waitForHeartbeatProcessed() throws TimeoutException, + InterruptedException { + GenericTestUtils.waitFor(() -> + scm.getScmNodeManager().waitForHeartbeatProcessed(), 100, + 4 * 1000); + GenericTestUtils.waitFor(() -> + scm.getScmNodeManager().getStats().getCapacity().get() > 0, 100, + 4 * 1000); + } + + /** + * Builder for configuring the MiniOzoneCluster to run. + */ + public static class Builder + extends MiniDFSCluster.Builder { + + private final OzoneConfiguration conf; + private static final int DEFAULT_HB_SECONDS = 1; + private static final int DEFAULT_PROCESSOR_MS = 100; + private final String path; + private final UUID runID; + private Optional<String> ozoneHandlerType = java.util.Optional.empty(); + private Optional<Boolean> enableTrace = Optional.of(false); + private Optional<Integer> hbSeconds = Optional.empty(); + private Optional<Integer> hbProcessorInterval = Optional.empty(); + private Optional<String> scmMetadataDir = Optional.empty(); + private Boolean ozoneEnabled = true; + private Boolean waitForChillModeFinish = true; + private Boolean randomContainerPort = true; + // Use relative smaller number of handlers for testing + private int numOfKsmHandlers = 20; + private int numOfScmHandlers = 20; + + /** + * Creates a new Builder. + * + * @param conf configuration + */ + public Builder(OzoneConfiguration conf) { + super(conf); + // Mini Ozone cluster will not come up if the port is not true, since + // Ratis will exit if the server port cannot be bound. We can remove this + // hard coding once we fix the Ratis default behaviour. + conf.setBoolean(OzoneConfigKeys.DFS_CONTAINER_RATIS_IPC_RANDOM_PORT, + true); + this.conf = conf; + path = GenericTestUtils.getTempPath( + MiniOzoneClassicCluster.class.getSimpleName() + + UUID.randomUUID().toString()); + runID = UUID.randomUUID(); + } + + public Builder setRandomContainerPort(boolean randomPort) { + this.randomContainerPort = randomPort; + return this; + } + + @Override + public Builder numDataNodes(int val) { + super.numDataNodes(val); + return this; + } + + @Override + public Builder storageCapacities(long[] capacities) { + super.storageCapacities(capacities); + return this; + } + + public Builder setHandlerType(String handler) { + ozoneHandlerType = Optional.of(handler); + return this; + } + + public Builder setTrace(Boolean trace) { + enableTrace = Optional.of(trace); + return this; + } + + public Builder setSCMHBInterval(int seconds) { + hbSeconds = Optional.of(seconds); + return this; + } + + public Builder setSCMHeartbeatProcessingInterval(int milliseconds) { + hbProcessorInterval = Optional.of(milliseconds); + return this; + } + + public Builder setSCMMetadataDir(String scmMetadataDirPath) { + scmMetadataDir = Optional.of(scmMetadataDirPath); + return this; + } + + public Builder disableOzone() { + ozoneEnabled = false; + return this; + } + + public Builder doNotwaitTobeOutofChillMode() { + waitForChillModeFinish = false; + return this; + } + + public Builder setNumOfKSMHandlers(int numOfHandlers) { + numOfKsmHandlers = numOfHandlers; + return this; + } + + public Builder setNumOfSCMHandlers(int numOfHandlers) { + numOfScmHandlers = numOfHandlers; + return this; + } + + public String getPath() { + return path; + } + + public String getRunID() { + return runID.toString(); + } + + @Override + public MiniOzoneClassicCluster build() throws IOException { + + + configureHandler(); + configureTrace(); + configureSCMheartbeat(); + configScmMetadata(); + + conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(ScmConfigKeys.OZONE_SCM_HTTP_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(KSMConfigKeys.OZONE_KSM_ADDRESS_KEY, "127.0.0.1:0"); + conf.set(KSMConfigKeys.OZONE_KSM_HTTP_ADDRESS_KEY, "127.0.0.1:0"); + + // Configure KSM and SCM handlers + conf.setInt(ScmConfigKeys.OZONE_SCM_HANDLER_COUNT_KEY, numOfScmHandlers); + conf.setInt(KSMConfigKeys.OZONE_KSM_HANDLER_COUNT_KEY, numOfKsmHandlers); + + // Use random ports for ozone containers in mini cluster, + // in order to launch multiple container servers per node. + conf.setBoolean(OzoneConfigKeys.DFS_CONTAINER_IPC_RANDOM_PORT, + randomContainerPort); + + SCMStorage scmStorage = new SCMStorage(conf); + scmStorage.initialize(); + StorageContainerManager scm = StorageContainerManager.createSCM( + null, conf); + scm.start(); + + KeySpaceManager ksm = new KeySpaceManager(conf); + ksm.start(); + + String addressString = scm.getDatanodeRpcAddress().getHostString() + + ":" + scm.getDatanodeRpcAddress().getPort(); + conf.setStrings(ScmConfigKeys.OZONE_SCM_NAMES, addressString); + + MiniOzoneClassicCluster cluster = + new MiniOzoneClassicCluster(this, scm, ksm); + try { + cluster.waitOzoneReady(); + if (waitForChillModeFinish) { + cluster.waitTobeOutOfChillMode(); + } + cluster.waitForHeartbeatProcessed(); + } catch (Exception e) { + // A workaround to propagate MiniOzoneCluster failures without + // changing the method signature (which would require cascading + // changes to hundreds of unrelated HDFS tests). + throw new IOException("Failed to start MiniOzoneCluster", e); + } + return cluster; + } + + private void configScmMetadata() throws IOException { + + + if (scmMetadataDir.isPresent()) { + // if user specifies a path in the test, it is assumed that user takes + // care of creating and cleaning up that directory after the tests. + conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, + scmMetadataDir.get()); + return; + } + + // If user has not specified a path, create a UUID for this miniCluster + // and create SCM under that directory. + Path scmPath = Paths.get(path, runID.toString(), "cont-meta"); + Files.createDirectories(scmPath); + Path containerPath = scmPath.resolve(OzoneConsts.CONTAINER_ROOT_PREFIX); + Files.createDirectories(containerPath); + conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, scmPath + .toString()); + + // TODO : Fix this, we need a more generic mechanism to map + // different datanode ID for different datanodes when we have lots of + // datanodes in the cluster. + conf.setStrings(ScmConfigKeys.OZONE_SCM_DATANODE_ID, + scmPath.toString() + "/datanode.id"); + } + + private void configureHandler() { + conf.setBoolean(OzoneConfigKeys.OZONE_ENABLED, this.ozoneEnabled); + if (!ozoneHandlerType.isPresent()) { + throw new IllegalArgumentException( + "The Ozone handler type must be specified."); + } else { + conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY, + ozoneHandlerType.get()); + } + } + + private void configureTrace() { + if (enableTrace.isPresent()) { + conf.setBoolean(OzoneConfigKeys.OZONE_TRACE_ENABLED_KEY, + enableTrace.get()); + GenericTestUtils.setRootLogLevel(Level.TRACE); + } + GenericTestUtils.setRootLogLevel(Level.INFO); + } + + private void configureSCMheartbeat() { + if (hbSeconds.isPresent()) { + conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS, + hbSeconds.get()); + + } else { + conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS, + DEFAULT_HB_SECONDS); + } + + if (hbProcessorInterval.isPresent()) { + conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS, + hbProcessorInterval.get()); + } else { + conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS, + DEFAULT_PROCESSOR_MS); + } + + } + } +} http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java index 71e48db..70444fb 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java @@ -17,564 +17,46 @@ */ package org.apache.hadoop.ozone; -import java.io.File; -import java.util.Optional; -import com.google.common.base.Preconditions; -import org.apache.commons.io.FileUtils; -import org.apache.hadoop.classification.InterfaceAudience; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.conf.OzoneConfiguration; -import org.apache.hadoop.hdfs.MiniDFSCluster; -import org.apache.hadoop.ipc.Client; -import org.apache.hadoop.ipc.ProtobufRpcEngine; -import org.apache.hadoop.ipc.RPC; -import org.apache.hadoop.net.NetUtils; -import org.apache.hadoop.ozone.client.OzoneClientUtils; -import org.apache.hadoop.ozone.container.common - .statemachine.DatanodeStateMachine; -import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; -import org.apache.hadoop.ozone.ksm.KSMConfigKeys; +import org.apache.hadoop.ozone.client.rest.OzoneException; import org.apache.hadoop.ozone.ksm.KeySpaceManager; -import org.apache.hadoop.ozone.ksm.protocolPB - .KeySpaceManagerProtocolClientSideTranslatorPB; -import org.apache.hadoop.ozone.ksm.protocolPB.KeySpaceManagerProtocolPB; -import org.apache.hadoop.ozone.scm.SCMStorage; +import org.apache.hadoop.ozone.scm.StorageContainerManager; import org.apache.hadoop.ozone.web.client.OzoneRestClient; -import org.apache.hadoop.scm.ScmConfigKeys; import org.apache.hadoop.scm.protocolPB .StorageContainerLocationProtocolClientSideTranslatorPB; -import org.apache.hadoop.scm.protocolPB.StorageContainerLocationProtocolPB; -import org.apache.hadoop.ozone.scm.StorageContainerManager; -import org.apache.hadoop.ozone.client.rest.OzoneException; -import org.apache.hadoop.security.UserGroupInformation; -import org.apache.hadoop.test.GenericTestUtils; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; -import org.slf4j.event.Level; import java.io.Closeable; import java.io.IOException; -import java.net.InetSocketAddress; -import java.net.URISyntaxException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.util.Random; -import java.util.UUID; import java.util.concurrent.TimeoutException; -import static org.apache.hadoop.ozone.OzoneConfigKeys - .DFS_CONTAINER_IPC_PORT; -import static org.apache.hadoop.ozone.OzoneConfigKeys - .DFS_CONTAINER_IPC_RANDOM_PORT; -import static org.apache.hadoop.ozone.OzoneConfigKeys - .DFS_CONTAINER_RATIS_IPC_PORT; -import static org.apache.hadoop.ozone.OzoneConfigKeys - .DFS_CONTAINER_RATIS_IPC_RANDOM_PORT; - -import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos.NodeState - .HEALTHY; -import static org.junit.Assert.assertFalse; - /** - * MiniOzoneCluster creates a complete in-process Ozone cluster suitable for - * running tests. The cluster consists of a StorageContainerManager, Namenode - * and multiple DataNodes. This class subclasses {@link MiniDFSCluster} for - * convenient reuse of logic for starting DataNodes. + * Interface used for MiniOzoneClusters. */ [email protected] -public final class MiniOzoneCluster extends MiniDFSCluster - implements Closeable { - private static final Logger LOG = - LoggerFactory.getLogger(MiniOzoneCluster.class); - private static final String USER_AUTH = "hdfs"; - - private final OzoneConfiguration conf; - private final StorageContainerManager scm; - private final KeySpaceManager ksm; - private final Path tempPath; - - /** - * Creates a new MiniOzoneCluster. - * - * @param builder cluster builder - * @param scm StorageContainerManager, already running - * @throws IOException if there is an I/O error - */ - private MiniOzoneCluster(Builder builder, StorageContainerManager scm, - KeySpaceManager ksm) - throws IOException { - super(builder); - this.conf = builder.conf; - this.scm = scm; - this.ksm = ksm; - tempPath = Paths.get(builder.getPath(), builder.getRunID()); - } - - - @Override - protected void setupDatanodeAddress( - int i, Configuration dnConf, boolean setupHostsFile, - boolean checkDnAddrConf) throws IOException { - super.setupDatanodeAddress(i, dnConf, setupHostsFile, checkDnAddrConf); - setConf(i, dnConf, OzoneConfigKeys.DFS_CONTAINER_RATIS_DATANODE_STORAGE_DIR, - getInstanceStorageDir(i, -1).getCanonicalPath()); - String containerMetaDirs = dnConf.get( - OzoneConfigKeys.OZONE_METADATA_DIRS) + "-dn-" + i; - Path containerMetaDirPath = Paths.get(containerMetaDirs); - setConf(i, dnConf, OzoneConfigKeys.OZONE_METADATA_DIRS, - containerMetaDirs); - Path containerRootPath = - containerMetaDirPath.resolve(OzoneConsts.CONTAINER_ROOT_PREFIX); - Files.createDirectories(containerRootPath); - } - - static void setConf(int i, Configuration conf, String key, String value) { - conf.set(key, value); - LOG.info("dn{}: set {} = {}", i, key, value); - } - - @Override - public void close() { - shutdown(); - try { - FileUtils.deleteDirectory(tempPath.toFile()); - } catch (IOException e) { - String errorMessage = "Cleaning up metadata directories failed." + e; - assertFalse(errorMessage, true); - } - - try { - final String localStorage = - conf.getTrimmed(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT, - OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT); - FileUtils.deleteDirectory(new File(localStorage)); - } catch (IOException e) { - LOG.error("Cleaning up local storage failed", e); - } - } - - public boolean restartDataNode(int i) throws IOException { - return restartDataNode(i, true); - } - /* - * Restart a particular datanode, wait for it to become active - */ - public boolean restartDataNode(int i, boolean keepPort) throws IOException { - LOG.info("restarting datanode:{} keepPort:{}", i, keepPort); - if (keepPort) { - DataNodeProperties dnProp = dataNodes.get(i); - OzoneContainer container = - dnProp.getDatanode().getOzoneContainerManager(); - Configuration config = dnProp.getConf(); - int currentPort = container.getContainerServerPort(); - config.setInt(DFS_CONTAINER_IPC_PORT, currentPort); - config.setBoolean(DFS_CONTAINER_IPC_RANDOM_PORT, false); - int ratisPort = container.getRatisContainerServerPort(); - config.setInt(DFS_CONTAINER_RATIS_IPC_PORT, ratisPort); - config.setBoolean(DFS_CONTAINER_RATIS_IPC_RANDOM_PORT, false); - } - boolean status = super.restartDataNode(i, keepPort); - - try { - this.waitActive(); - waitDatanodeOzoneReady(i); - } catch (TimeoutException | InterruptedException e) { - Thread.interrupted(); - } - return status; - } - - @Override - public void shutdown() { - super.shutdown(); - LOG.info("Shutting down the Mini Ozone Cluster"); - - if (ksm != null) { - LOG.info("Shutting down the keySpaceManager"); - ksm.stop(); - ksm.join(); - } - - if (scm != null) { - LOG.info("Shutting down the StorageContainerManager"); - scm.stop(); - scm.join(); - } - } - - public StorageContainerManager getStorageContainerManager() { - return this.scm; - } - - public KeySpaceManager getKeySpaceManager() { - return this.ksm; - } - - /** - * Creates an {@link OzoneRestClient} connected to this cluster's REST - * service. Callers take ownership of the client and must close it when done. - * - * @return OzoneRestClient connected to this cluster's REST service - * @throws OzoneException if Ozone encounters an error creating the client - */ - public OzoneRestClient createOzoneRestClient() throws OzoneException { - Preconditions.checkState(!getDataNodes().isEmpty(), - "Cannot create OzoneRestClient if the cluster has no DataNodes."); - // An Ozone request may originate at any DataNode, so pick one at random. - int dnIndex = new Random().nextInt(getDataNodes().size()); - String uri = String.format("http://127.0.0.1:%d";, - getDataNodes().get(dnIndex).getInfoPort()); - LOG.info("Creating Ozone client to DataNode {} with URI {} and user {}", - dnIndex, uri, USER_AUTH); - try { - return new OzoneRestClient(uri, USER_AUTH); - } catch (URISyntaxException e) { - // We control the REST service URI, so it should never be invalid. - throw new IllegalStateException("Unexpected URISyntaxException", e); - } - } - - /** - * Creates an RPC proxy connected to this cluster's StorageContainerManager - * for accessing container location information. Callers take ownership of - * the proxy and must close it when done. - * - * @return RPC proxy for accessing container location information - * @throws IOException if there is an I/O error - */ - public StorageContainerLocationProtocolClientSideTranslatorPB - createStorageContainerLocationClient() throws IOException { - long version = RPC.getProtocolVersion( - StorageContainerLocationProtocolPB.class); - InetSocketAddress address = scm.getClientRpcAddress(); - LOG.info( - "Creating StorageContainerLocationProtocol RPC client with address {}", - address); - return new StorageContainerLocationProtocolClientSideTranslatorPB( - RPC.getProxy(StorageContainerLocationProtocolPB.class, version, - address, UserGroupInformation.getCurrentUser(), conf, - NetUtils.getDefaultSocketFactory(conf), - Client.getRpcTimeout(conf))); - } - - /** - * Creates an RPC proxy connected to this cluster's KeySpaceManager - * for accessing Key Space Manager information. Callers take ownership of - * the proxy and must close it when done. - * - * @return RPC proxy for accessing Key Space Manager information - * @throws IOException if there is an I/O error - */ - public KeySpaceManagerProtocolClientSideTranslatorPB - createKeySpaceManagerClient() throws IOException { - long ksmVersion = RPC.getProtocolVersion(KeySpaceManagerProtocolPB.class); - InetSocketAddress ksmAddress = OzoneClientUtils - .getKsmAddressForClients(conf); - LOG.info("Creating KeySpaceManager RPC client with address {}", - ksmAddress); - RPC.setProtocolEngine(conf, KeySpaceManagerProtocolPB.class, - ProtobufRpcEngine.class); - return new KeySpaceManagerProtocolClientSideTranslatorPB( - RPC.getProxy(KeySpaceManagerProtocolPB.class, ksmVersion, - ksmAddress, UserGroupInformation.getCurrentUser(), conf, - NetUtils.getDefaultSocketFactory(conf), - Client.getRpcTimeout(conf))); - } - - /** - * Waits for the Ozone cluster to be ready for processing requests. - */ - public void waitOzoneReady() throws TimeoutException, InterruptedException { - GenericTestUtils.waitFor(() -> { - final int healthy = scm.getNodeCount(HEALTHY); - final boolean isReady = healthy >= numDataNodes; - LOG.info("{}. Got {} of {} DN Heartbeats.", - isReady? "Cluster is ready" : "Waiting for cluster to be ready", - healthy, numDataNodes); - return isReady; - }, 1000, 60 * 1000); //wait for 1 min. - } - - /** - * Waits for a particular Datanode to be ready for processing ozone requests. - */ - public void waitDatanodeOzoneReady(int dnIndex) - throws TimeoutException, InterruptedException { - GenericTestUtils.waitFor(() -> { - DatanodeStateMachine.DatanodeStates state = - dataNodes.get(dnIndex).getDatanode().getOzoneStateMachineState(); - final boolean rebootComplete = - (state == DatanodeStateMachine.DatanodeStates.RUNNING); - LOG.info("{} Current state:{}", rebootComplete, state); - return rebootComplete; - }, 1000, 60 * 1000); //wait for 1 min. - } - - /** - * Waits for SCM to be out of Chill Mode. Many tests can be run iff we are out - * of Chill mode. - * - * @throws TimeoutException - * @throws InterruptedException - */ - public void waitTobeOutOfChillMode() throws TimeoutException, - InterruptedException { - GenericTestUtils.waitFor(() -> { - if (scm.getScmNodeManager().isOutOfChillMode()) { - return true; - } - LOG.info("Waiting for cluster to be ready. No datanodes found"); - return false; - }, 100, 45000); - } - - public void waitForHeartbeatProcessed() throws TimeoutException, - InterruptedException { - GenericTestUtils.waitFor(() -> - scm.getScmNodeManager().waitForHeartbeatProcessed(), 100, - 4 * 1000); - GenericTestUtils.waitFor(() -> - scm.getScmNodeManager().getStats().getCapacity().get() > 0, 100, - 4 * 1000); - } - - /** - * Builder for configuring the MiniOzoneCluster to run. - */ - public static class Builder - extends MiniDFSCluster.Builder { - - private final OzoneConfiguration conf; - private static final int DEFAULT_HB_SECONDS = 1; - private static final int DEFAULT_PROCESSOR_MS = 100; - private final String path; - private final UUID runID; - private Optional<String> ozoneHandlerType = java.util.Optional.empty(); - private Optional<Boolean> enableTrace = Optional.of(false); - private Optional<Integer> hbSeconds = Optional.empty(); - private Optional<Integer> hbProcessorInterval = Optional.empty(); - private Optional<String> scmMetadataDir = Optional.empty(); - private Boolean ozoneEnabled = true; - private Boolean waitForChillModeFinish = true; - private Boolean randomContainerPort = true; - // Use relative smaller number of handlers for testing - private int numOfKsmHandlers = 20; - private int numOfScmHandlers = 20; - - /** - * Creates a new Builder. - * - * @param conf configuration - */ - public Builder(OzoneConfiguration conf) { - super(conf); - // Mini Ozone cluster will not come up if the port is not true, since - // Ratis will exit if the server port cannot be bound. We can remove this - // hard coding once we fix the Ratis default behaviour. - conf.setBoolean(OzoneConfigKeys.DFS_CONTAINER_RATIS_IPC_RANDOM_PORT, - true); - this.conf = conf; - path = GenericTestUtils.getTempPath( - MiniOzoneCluster.class.getSimpleName() + - UUID.randomUUID().toString()); - runID = UUID.randomUUID(); - } - - public Builder setRandomContainerPort(boolean randomPort) { - this.randomContainerPort = randomPort; - return this; - } - - @Override - public Builder numDataNodes(int val) { - super.numDataNodes(val); - return this; - } - - @Override - public Builder storageCapacities(long[] capacities) { - super.storageCapacities(capacities); - return this; - } - - public Builder setHandlerType(String handler) { - ozoneHandlerType = Optional.of(handler); - return this; - } - - public Builder setTrace(Boolean trace) { - enableTrace = Optional.of(trace); - return this; - } - - public Builder setSCMHBInterval(int seconds) { - hbSeconds = Optional.of(seconds); - return this; - } - - public Builder setSCMHeartbeatProcessingInterval(int milliseconds) { - hbProcessorInterval = Optional.of(milliseconds); - return this; - } - - public Builder setSCMMetadataDir(String scmMetadataDirPath) { - scmMetadataDir = Optional.of(scmMetadataDirPath); - return this; - } - - public Builder disableOzone() { - ozoneEnabled = false; - return this; - } - - public Builder doNotwaitTobeOutofChillMode() { - waitForChillModeFinish = false; - return this; - } - - public Builder setNumOfKSMHandlers(int numOfHandlers) { - numOfKsmHandlers = numOfHandlers; - return this; - } - - public Builder setNumOfSCMHandlers(int numOfHandlers) { - numOfScmHandlers = numOfHandlers; - return this; - } - - public String getPath() { - return path; - } - - public String getRunID() { - return runID.toString(); - } - - @Override - public MiniOzoneCluster build() throws IOException { - - - configureHandler(); - configureTrace(); - configureSCMheartbeat(); - configScmMetadata(); - configVersionFile(); - - conf.set(ScmConfigKeys.OZONE_SCM_CLIENT_ADDRESS_KEY, "127.0.0.1:0"); - conf.set(ScmConfigKeys.OZONE_SCM_BLOCK_CLIENT_ADDRESS_KEY, "127.0.0.1:0"); - conf.set(ScmConfigKeys.OZONE_SCM_DATANODE_ADDRESS_KEY, "127.0.0.1:0"); - conf.set(ScmConfigKeys.OZONE_SCM_HTTP_ADDRESS_KEY, "127.0.0.1:0"); - conf.set(KSMConfigKeys.OZONE_KSM_ADDRESS_KEY, "127.0.0.1:0"); - conf.set(KSMConfigKeys.OZONE_KSM_HTTP_ADDRESS_KEY, "127.0.0.1:0"); - - // Configure KSM and SCM handlers - conf.setInt(ScmConfigKeys.OZONE_SCM_HANDLER_COUNT_KEY, numOfScmHandlers); - conf.setInt(KSMConfigKeys.OZONE_KSM_HANDLER_COUNT_KEY, numOfKsmHandlers); - - // Use random ports for ozone containers in mini cluster, - // in order to launch multiple container servers per node. - conf.setBoolean(OzoneConfigKeys.DFS_CONTAINER_IPC_RANDOM_PORT, - randomContainerPort); - - StorageContainerManager scm = - StorageContainerManager.createSCM(null, conf); - scm.start(); - - KeySpaceManager ksm = new KeySpaceManager(conf); - ksm.start(); - - String addressString = scm.getDatanodeRpcAddress().getHostString() + - ":" + scm.getDatanodeRpcAddress().getPort(); - conf.setStrings(ScmConfigKeys.OZONE_SCM_NAMES, addressString); - - MiniOzoneCluster cluster = new MiniOzoneCluster(this, scm, ksm); - try { - cluster.waitOzoneReady(); - if (waitForChillModeFinish) { - cluster.waitTobeOutOfChillMode(); - } - cluster.waitForHeartbeatProcessed(); - } catch (Exception e) { - // A workaround to propagate MiniOzoneCluster failures without - // changing the method signature (which would require cascading - // changes to hundreds of unrelated HDFS tests). - throw new IOException("Failed to start MiniOzoneCluster", e); - } - return cluster; - } - - private void configScmMetadata() throws IOException { +public interface MiniOzoneCluster extends AutoCloseable, Closeable { + void close(); + boolean restartDataNode(int i) throws IOException; - if (scmMetadataDir.isPresent()) { - // if user specifies a path in the test, it is assumed that user takes - // care of creating and cleaning up that directory after the tests. - conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, - scmMetadataDir.get()); - return; - } + boolean restartDataNode(int i, boolean keepPort) throws IOException; - // If user has not specified a path, create a UUID for this miniCluster - // and create SCM under that directory. - Path scmPath = Paths.get(path, runID.toString(), "cont-meta"); - Files.createDirectories(scmPath); - Path containerPath = scmPath.resolve(OzoneConsts.CONTAINER_ROOT_PREFIX); - Files.createDirectories(containerPath); - conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, scmPath - .toString()); + void shutdown(); - // TODO : Fix this, we need a more generic mechanism to map - // different datanode ID for different datanodes when we have lots of - // datanodes in the cluster. - conf.setStrings(ScmConfigKeys.OZONE_SCM_DATANODE_ID, - scmPath.toString() + "/datanode.id"); - } + StorageContainerManager getStorageContainerManager(); - private void configVersionFile() throws IOException { - SCMStorage scmStore = new SCMStorage(conf); - scmStore.setClusterId(runID.toString()); - scmStore.initialize(); - } + KeySpaceManager getKeySpaceManager(); - private void configureHandler() { - conf.setBoolean(OzoneConfigKeys.OZONE_ENABLED, this.ozoneEnabled); - if (!ozoneHandlerType.isPresent()) { - throw new IllegalArgumentException( - "The Ozone handler type must be specified."); - } else { - conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY, - ozoneHandlerType.get()); - } - } + OzoneRestClient createOzoneRestClient() throws OzoneException; - private void configureTrace() { - if (enableTrace.isPresent()) { - conf.setBoolean(OzoneConfigKeys.OZONE_TRACE_ENABLED_KEY, - enableTrace.get()); - GenericTestUtils.setRootLogLevel(Level.TRACE); - } - GenericTestUtils.setRootLogLevel(Level.INFO); - } + StorageContainerLocationProtocolClientSideTranslatorPB + createStorageContainerLocationClient() throws IOException; - private void configureSCMheartbeat() { - if (hbSeconds.isPresent()) { - conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS, - hbSeconds.get()); + void waitOzoneReady() throws TimeoutException, InterruptedException; - } else { - conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_INTERVAL_SECONDS, - DEFAULT_HB_SECONDS); - } + void waitDatanodeOzoneReady(int dnIndex) + throws TimeoutException, InterruptedException; - if (hbProcessorInterval.isPresent()) { - conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS, - hbProcessorInterval.get()); - } else { - conf.setInt(ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL_MS, - DEFAULT_PROCESSOR_MS); - } + void waitTobeOutOfChillMode() throws TimeoutException, + InterruptedException; - } - } -} + void waitForHeartbeatProcessed() throws TimeoutException, + InterruptedException; +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java index d7d6ad7..7f79e82 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/RatisTestHelper.java @@ -44,10 +44,10 @@ public interface RatisTestHelper { static final int NUM_DATANODES = 3; private final OzoneConfiguration conf; - private final MiniOzoneCluster cluster; + private final MiniOzoneClassicCluster cluster; /** - * Create a {@link MiniOzoneCluster} for testing by setting + * Create a {@link MiniOzoneClassicCluster} for testing by setting * OZONE_ENABLED = true, * RATIS_ENABLED = true, and * OZONE_HANDLER_TYPE_KEY = "distributed". @@ -61,7 +61,7 @@ public interface RatisTestHelper { return conf; } - public MiniOzoneCluster getCluster() { + public MiniOzoneClassicCluster getCluster() { return cluster; } @@ -95,15 +95,12 @@ public interface RatisTestHelper { + " = " + rpc.name()); } - static MiniOzoneCluster newMiniOzoneCluster( + static MiniOzoneClassicCluster newMiniOzoneCluster( int numDatanodes, OzoneConfiguration conf) throws IOException { - final MiniOzoneCluster cluster = new MiniOzoneCluster.Builder(conf) + final MiniOzoneClassicCluster cluster = + new MiniOzoneClassicCluster.Builder(conf) .numDataNodes(numDatanodes) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); -// cluster.getRatisManager().createPipeline("ratis0", -// cluster.getDataNodes().stream() -// .map(DataNode::getDatanodeId) -// .collect(Collectors.toList())); return cluster; } http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java index 16d5330..26922f6 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestContainerOperations.java @@ -55,7 +55,7 @@ public class TestContainerOperations { ozoneConf = new OzoneConfiguration(); ozoneConf.setClass(ScmConfigKeys.OZONE_SCM_CONTAINER_PLACEMENT_IMPL_KEY, SCMContainerPlacementCapacity.class, ContainerPlacementPolicy.class); - cluster = new MiniOzoneCluster.Builder(ozoneConf).numDataNodes(1) + cluster = new MiniOzoneClassicCluster.Builder(ozoneConf).numDataNodes(1) .storageCapacities(new long[] {datanodeCapacities, datanodeCapacities}) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); StorageContainerLocationProtocolClientSideTranslatorPB client = http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java index a69f475..946b7e0 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestMiniOzoneCluster.java @@ -53,7 +53,7 @@ import static org.junit.Assert.*; */ public class TestMiniOzoneCluster { - private static MiniOzoneCluster cluster; + private static MiniOzoneClassicCluster cluster; private static OzoneConfiguration conf; private final static File TEST_ROOT = TestGenericTestUtils.getTestDir(); @@ -83,8 +83,10 @@ public class TestMiniOzoneCluster { @Test(timeout = 30000) public void testStartMultipleDatanodes() throws Exception { final int numberOfNodes = 3; - cluster = new MiniOzoneCluster.Builder(conf).numDataNodes(numberOfNodes) - .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); + cluster = new MiniOzoneClassicCluster.Builder(conf) + .numDataNodes(numberOfNodes) + .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED) + .build(); // make sure datanode.id file is correct File idPath = new File( http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java index e42065d..e7e36a5 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManager.java @@ -34,7 +34,6 @@ import org.apache.hadoop.ozone.scm.StorageContainerManager; import org.apache.hadoop.ozone.scm.StorageContainerManager.StartupOption; import org.apache.hadoop.ozone.scm.block.DeletedBlockLog; import org.apache.hadoop.ozone.scm.block.SCMBlockDeletingService; -import org.apache.hadoop.ozone.scm.exceptions.SCMException; import org.apache.hadoop.ozone.scm.node.NodeManager; import org.apache.hadoop.scm.XceiverClientManager; import org.apache.hadoop.scm.container.common.helpers.Pipeline; @@ -102,7 +101,7 @@ public class TestStorageContainerManager { OzoneConfiguration ozoneConf, String fakeRemoteUsername, boolean expectPermissionDenied) throws IOException { MiniOzoneCluster cluster = - new MiniOzoneCluster.Builder(ozoneConf).numDataNodes(1) + new MiniOzoneClassicCluster.Builder(ozoneConf).numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); try { @@ -192,8 +191,8 @@ public class TestStorageContainerManager { conf.setInt(ScmConfigKeys.OZONE_SCM_CONTAINER_PROVISION_BATCH_SIZE, numKeys); - MiniOzoneCluster cluster = - new MiniOzoneCluster.Builder(conf).numDataNodes(1) + MiniOzoneClassicCluster cluster = + new MiniOzoneClassicCluster.Builder(conf).numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); try { @@ -267,7 +266,7 @@ public class TestStorageContainerManager { conf.setInt(ScmConfigKeys.OZONE_SCM_CONTAINER_PROVISION_BATCH_SIZE, numKeys); - MiniOzoneCluster cluster = new MiniOzoneCluster.Builder(conf) + MiniOzoneClassicCluster cluster = new MiniOzoneClassicCluster.Builder(conf) .numDataNodes(1).setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED) .build(); @@ -388,7 +387,7 @@ public class TestStorageContainerManager { conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, scmPath.toString()); //This will set the cluster id in the version file MiniOzoneCluster cluster = - new MiniOzoneCluster.Builder(conf).numDataNodes(1) + new MiniOzoneClassicCluster.Builder(conf).numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); StartupOption.INIT.setClusterId("testClusterId"); // This will initialize SCM @@ -397,17 +396,4 @@ public class TestStorageContainerManager { Assert.assertEquals(OzoneConsts.NodeType.SCM, scmStore.getNodeType()); Assert.assertNotEquals("testClusterId", scmStore.getClusterID()); } - - @Test - public void testSCMInitializationFailure() throws IOException { - OzoneConfiguration conf = new OzoneConfiguration(); - final String path = - GenericTestUtils.getTempPath(UUID.randomUUID().toString()); - Path scmPath = Paths.get(path, "scm-meta"); - conf.set(OzoneConfigKeys.OZONE_METADATA_DIRS, scmPath.toString()); - conf.setBoolean(OzoneConfigKeys.OZONE_ENABLED, true); - exception.expect(SCMException.class); - exception.expectMessage("SCM not initialized."); - StorageContainerManager.createSCM(null, conf); - } } http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java index 2123d6f..fdfa9ba 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/TestStorageContainerManagerHelper.java @@ -55,11 +55,11 @@ import java.util.Set; */ public class TestStorageContainerManagerHelper { - private final MiniOzoneCluster cluster; + private final MiniOzoneClassicCluster cluster; private final Configuration conf; private final StorageHandler storageHandler; - public TestStorageContainerManagerHelper(MiniOzoneCluster cluster, + public TestStorageContainerManagerHelper(MiniOzoneClassicCluster cluster, Configuration conf) throws IOException { this.cluster = cluster; this.conf = conf; http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rest/TestOzoneRestClient.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rest/TestOzoneRestClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rest/TestOzoneRestClient.java index 7b46bfc..2bc5818 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rest/TestOzoneRestClient.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rest/TestOzoneRestClient.java @@ -21,7 +21,7 @@ package org.apache.hadoop.ozone.client.rest; import org.apache.hadoop.conf.OzoneConfiguration; import org.apache.hadoop.fs.StorageType; import org.apache.hadoop.hdfs.server.datanode.DataNode; -import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.MiniOzoneClassicCluster; import org.apache.hadoop.ozone.OzoneAcl; import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.OzoneConsts; @@ -58,7 +58,7 @@ public class TestOzoneRestClient { @Rule public ExpectedException thrown = ExpectedException.none(); - private static MiniOzoneCluster cluster = null; + private static MiniOzoneClassicCluster cluster = null; private static OzoneClient ozClient = null; private static ObjectStore store = null; @@ -75,7 +75,7 @@ public class TestOzoneRestClient { OzoneConfiguration conf = new OzoneConfiguration(); conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY, OzoneConsts.OZONE_HANDLER_DISTRIBUTED); - cluster = new MiniOzoneCluster.Builder(conf) + cluster = new MiniOzoneClassicCluster.Builder(conf).numDataNodes(1) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); DataNode datanode = cluster.getDataNodes().get(0); conf.set(OzoneConfigKeys.OZONE_CLIENT_PROTOCOL, http://git-wip-us.apache.org/repos/asf/hadoop/blob/1a1aec23/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java ---------------------------------------------------------------------- diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java index c5ced8c..a78bc3f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java +++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClient.java @@ -20,6 +20,7 @@ package org.apache.hadoop.ozone.client.rpc; import org.apache.commons.lang.RandomStringUtils; import org.apache.hadoop.fs.StorageType; +import org.apache.hadoop.ozone.MiniOzoneClassicCluster; import org.apache.hadoop.ozone.MiniOzoneCluster; import org.apache.hadoop.ozone.OzoneAcl; import org.apache.hadoop.ozone.OzoneConfigKeys; @@ -38,11 +39,10 @@ import org.apache.hadoop.ozone.client.ReplicationType; import org.apache.hadoop.ozone.client.VolumeArgs; import org.apache.hadoop.ozone.client.io.OzoneInputStream; import org.apache.hadoop.ozone.client.io.OzoneOutputStream; +import org.apache.hadoop.ozone.ksm.KeySpaceManager; import org.apache.hadoop.ozone.ksm.helpers.KsmKeyArgs; import org.apache.hadoop.ozone.ksm.helpers.KsmKeyInfo; import org.apache.hadoop.ozone.ksm.helpers.KsmKeyLocationInfo; -import org.apache.hadoop.ozone.ksm.protocolPB. - KeySpaceManagerProtocolClientSideTranslatorPB; import org.apache.hadoop.ozone.protocol.proto.OzoneProtos; import org.apache.hadoop.ozone.client.rest.OzoneException; import org.apache.hadoop.scm.container.common.helpers.Pipeline; @@ -70,11 +70,10 @@ public class TestOzoneRpcClient { @Rule public ExpectedException thrown = ExpectedException.none(); - private static MiniOzoneCluster cluster = null; + private static MiniOzoneClassicCluster cluster = null; private static OzoneClient ozClient = null; private static ObjectStore store = null; - private static KeySpaceManagerProtocolClientSideTranslatorPB - keySpaceManagerClient; + private static KeySpaceManager keySpaceManager; private static StorageContainerLocationProtocolClientSideTranslatorPB storageContainerLocationClient; @@ -91,7 +90,7 @@ public class TestOzoneRpcClient { OzoneConfiguration conf = new OzoneConfiguration(); conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY, OzoneConsts.OZONE_HANDLER_DISTRIBUTED); - cluster = new MiniOzoneCluster.Builder(conf).numDataNodes(5) + cluster = new MiniOzoneClassicCluster.Builder(conf) .setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build(); conf.set("ozone.client.protocol", "org.apache.hadoop.ozone.client.rpc.RpcClient"); @@ -100,7 +99,7 @@ public class TestOzoneRpcClient { store = ozClient.getObjectStore(); storageContainerLocationClient = cluster.createStorageContainerLocationClient(); - keySpaceManagerClient = cluster.createKeySpaceManagerClient(); + keySpaceManager = cluster.getKeySpaceManager(); } @Test @@ -388,7 +387,7 @@ public class TestOzoneRpcClient { OzoneProtos.ReplicationType.valueOf(type.toString()); OzoneProtos.ReplicationFactor replicationFactor = OzoneProtos.ReplicationFactor.valueOf(factor.getValue()); - KsmKeyInfo keyInfo = keySpaceManagerClient.lookupKey(keyArgs); + KsmKeyInfo keyInfo = keySpaceManager.lookupKey(keyArgs); for (KsmKeyLocationInfo info: keyInfo.getKeyLocationList()) { Pipeline pipeline = storageContainerLocationClient.getContainer(info.getContainerName()); @@ -809,10 +808,6 @@ public class TestOzoneRpcClient { storageContainerLocationClient.close(); } - if (keySpaceManagerClient != null) { - keySpaceManagerClient.close(); - } - if (cluster != null) { cluster.shutdown(); } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
