big-data/hadoop
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

HDFS-12598. Ozone: Fix 3 node ratis replication in Ozone. Contributed by Mukul Kumar Singh

Browse Source
This commit is contained in:
Tsz-Wo Nicholas Sze 2017-10-08 22:53:01 +08:00 committed by Owen O'Malley
parent 3504af9411
commit e3b51d9074
13 changed files with 135 additions and 67 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/XceiverClientRatis.java
View File

@ -77,6 +77,7 @@ public void createPipeline(String clusterId, List<DatanodeID> datanodes)
throws IOException {
final RaftPeer[] newPeers = datanodes.stream().map(RatisHelper::toRaftPeer)
.toArray(RaftPeer[]::new);
LOG.debug("initializing pipeline:{} with nodes:{}", clusterId, newPeers);
reinitialize(datanodes, newPeers);
}

1
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/container/common/helpers/ContainerInfo.java
View File

@ -48,6 +48,7 @@ public class ContainerInfo {
public ContainerInfo(ContainerInfo container) {
this.pipeline = container.getPipeline();
this.state = container.getState();
this.containerName = container.getContainerName();
this.stateEnterTime = container.getStateEnterTime();
this.owner = container.getOwner();
}

3
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/scm/container/common/helpers/Pipeline.java
View File

@ -299,6 +299,9 @@ public String toString() {
if (getType() != null) {
b.append(" type:").append(getType().toString());
}
if (getFactor() != null) {
b.append(" factor:").append(getFactor().toString());
}
if (getLifeCycleState() != null) {
b.append(" State:").append(getLifeCycleState().toString());
}

14
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
View File

@ -21,6 +21,7 @@
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
import org.apache.hadoop.ozone.container.common.transport.server
@ -31,7 +32,7 @@
import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.grpc.GrpcConfigKeys;
import org.apache.ratis.netty.NettyConfigKeys;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.RatisHelper;
import org.apache.ratis.rpc.RpcType;
import org.apache.ratis.rpc.SupportedRpcType;
import org.apache.ratis.server.RaftServer;
@ -56,14 +57,13 @@ public final class XceiverServerRatis implements XceiverServerSpi {
private final int port;
private final RaftServer server;
private XceiverServerRatis(
String id, int port, String storageDir,
private XceiverServerRatis(DatanodeID id, int port, String storageDir,
ContainerDispatcher dispatcher, RpcType rpcType) throws IOException {
Objects.requireNonNull(id, "id == null");
this.port = port;
this.server = RaftServer.newBuilder()
.setServerId(RaftPeerId.valueOf(id))
.setServerId(RatisHelper.toRaftPeerId(id))
.setPeers(Collections.emptyList())
.setProperties(newRaftProperties(rpcType, port, storageDir))
.setStateMachine(new ContainerStateMachine(dispatcher))
@ -85,7 +85,7 @@ static RaftProperties newRaftProperties(
return properties;
}
public static XceiverServerRatis newXceiverServerRatis(String datanodeID,
public static XceiverServerRatis newXceiverServerRatis(DatanodeID datanodeID,
Configuration ozoneConf, ContainerDispatcher dispatcher)
throws IOException {
final String ratisDir = File.separator + "ratis";
@ -125,12 +125,14 @@ public static XceiverServerRatis newXceiverServerRatis(String datanodeID,
// probably running under MiniOzoneCluster. Ratis locks the storage
// directories, so we need to pass different local directory for each
// local instance. So we map ratis directories under datanode ID.
storageDir = storageDir.concat(File.separator + datanodeID);
storageDir =
storageDir.concat(File.separator + datanodeID.getDatanodeUuid());
} catch (IOException e) {
LOG.error("Unable find a random free port for the server, "
+ "fallback to use default port {}", localPort, e);
}
}
datanodeID.setRatisPort(localPort);
return new XceiverServerRatis(datanodeID, localPort, storageDir,
dispatcher, rpc);
}

4
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
View File

@ -116,8 +116,8 @@ public OzoneContainer(DatanodeID datanodeID, Configuration ozoneConfig) throws
server = new XceiverServerSpi[]{
new XceiverServer(this.ozoneConfig, this.dispatcher),
XceiverServerRatis.newXceiverServerRatis(datanodeID
.getDatanodeUuid().toString(), ozoneConfig, dispatcher)
XceiverServerRatis
.newXceiverServerRatis(datanodeID, ozoneConfig, dispatcher)
};
}

2
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/container/ContainerStateManager.java
View File

@ -274,7 +274,7 @@ public ContainerInfo allocateContainer(PipelineSelector selector, OzoneProtos
writeLock.lock();
try {
ContainerKey key = new ContainerKey(owner, type, replicationFactor,
info.getState());
blockInfo.getState());
PriorityQueue<BlockContainerInfo> queue = containers.get(key);
Preconditions.checkNotNull(queue);
queue.add(blockInfo);

3
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/pipelines/PipelineSelector.java
View File

@ -68,7 +68,8 @@ public PipelineSelector(NodeManager nodeManager, Configuration conf) {
new StandaloneManagerImpl(this.nodeManager, placementPolicy,
containerSize);
this.ratisManager =
new RatisManagerImpl(this.nodeManager, placementPolicy, containerSize);
new RatisManagerImpl(this.nodeManager, placementPolicy, containerSize,
conf);
}
/**

33
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/scm/pipelines/ratis/RatisManagerImpl.java
View File

@ -17,12 +17,15 @@
package org.apache.hadoop.ozone.scm.pipelines.ratis;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.ozone.protocol.proto.OzoneProtos;
import org.apache.hadoop.ozone.scm.container.placement.algorithms.ContainerPlacementPolicy;
import org.apache.hadoop.ozone.scm.container.placement.algorithms
.ContainerPlacementPolicy;
import org.apache.hadoop.ozone.scm.node.NodeManager;
import org.apache.hadoop.ozone.scm.pipelines.PipelineManager;
import org.apache.hadoop.ozone.scm.pipelines.PipelineSelector;
import org.apache.hadoop.scm.XceiverClientRatis;
import org.apache.hadoop.scm.container.common.helpers.Pipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -35,8 +38,10 @@
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos.LifeCycleState.ALLOCATED;
import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos.LifeCycleState.OPEN;
import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos
.LifeCycleState.ALLOCATED;
import static org.apache.hadoop.ozone.protocol.proto.OzoneProtos
.LifeCycleState.OPEN;
/**
@ -54,6 +59,7 @@ public class RatisManagerImpl implements PipelineManager {
private final List<Pipeline> activePipelines;
private final AtomicInteger pipelineIndex;
private static final String PREFIX = "Ratis-";
private final Configuration conf;
/**
* Constructs a Ratis Pipeline Manager.
@ -61,13 +67,14 @@ public class RatisManagerImpl implements PipelineManager {
* @param nodeManager
*/
public RatisManagerImpl(NodeManager nodeManager,
ContainerPlacementPolicy placementPolicy, long size) {
ContainerPlacementPolicy placementPolicy, long size, Configuration conf) {
this.nodeManager = nodeManager;
this.placementPolicy = placementPolicy;
this.containerSize = size;
ratisMembers = new HashSet<>();
activePipelines = new LinkedList<>();
pipelineIndex = new AtomicInteger(0);
this.conf = conf;
}
/**
@ -85,7 +92,7 @@ public RatisManagerImpl(NodeManager nodeManager,
*/
@Override
public synchronized Pipeline getPipeline(String containerName,
OzoneProtos.ReplicationFactor replicationFactor) {
OzoneProtos.ReplicationFactor replicationFactor) throws IOException {
/**
* In the ratis world, we have a very simple policy.
*
@ -106,7 +113,13 @@ public synchronized Pipeline getPipeline(String containerName,
Preconditions.checkState(newNodes.size() ==
getReplicationCount(replicationFactor), "Replication factor " +
"does not match the expected node count.");
pipeline = allocateRatisPipeline(newNodes, containerName);
pipeline =
allocateRatisPipeline(newNodes, containerName, replicationFactor);
try (XceiverClientRatis client =
XceiverClientRatis.newXceiverClientRatis(pipeline, conf)) {
client
.createPipeline(pipeline.getPipelineName(), pipeline.getMachines());
}
} else {
pipeline = findOpenPipeline();
}
@ -151,7 +164,8 @@ Pipeline findOpenPipeline() {
* @param containerName - container Name
* @return - Pipeline.
*/
Pipeline allocateRatisPipeline(List<DatanodeID> nodes, String containerName) {
Pipeline allocateRatisPipeline(List<DatanodeID> nodes, String containerName,
OzoneProtos.ReplicationFactor factor) {
Preconditions.checkNotNull(nodes);
Pipeline pipeline = PipelineSelector.newPipelineFromNodes(nodes);
if (pipeline != null) {
@ -160,6 +174,7 @@ Pipeline allocateRatisPipeline(List<DatanodeID> nodes, String containerName) {
UUID.randomUUID().toString().substring(PREFIX.length());
pipeline.setType(OzoneProtos.ReplicationType.RATIS);
pipeline.setLifeCycleState(ALLOCATED);
pipeline.setFactor(factor);
pipeline.setPipelineName(pipelineName);
pipeline.setContainerName(containerName);
LOG.info("Creating new ratis pipeline: {}", pipeline.toString());
@ -192,8 +207,12 @@ private List<DatanodeID> allocatePipelineNodes(OzoneProtos.ReplicationFactor
//TODO: Add Raft State to the Nodes, so we can query and skip nodes from
// data from datanode instead of maintaining a set.
for (DatanodeID datanode : datanodes) {
Preconditions.checkNotNull(datanode);
if (!ratisMembers.contains(datanode)) {
newNodesList.add(datanode);
// once a datanode has been added to a pipeline, exclude it from
// further allocations
ratisMembers.add(datanode);
if (newNodesList.size() == count) {
LOG.info("Allocating a new pipeline of size: {}", count);
return newNodesList;

56
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/ozone/tools/Corona.java
View File

@ -132,7 +132,8 @@ public final class Corona extends Configured implements Tool {
private String numOfKeys;
private String jsonDir;
private boolean useRatis;
private int replicationFactor = 0;
private OzoneProtos.ReplicationType type;
private OzoneProtos.ReplicationFactor factor;
private int keySize;
private byte[] keyValue = null;
@ -357,9 +358,24 @@ private void parseOptions(CommandLine cmdLine) {
useRatis = cmdLine.hasOption(RATIS);
//To-do if replication factor is not mentioned throw an exception
replicationFactor =
useRatis ? Integer.parseInt(cmdLine.getOptionValue(RATIS)) : 0;
type = OzoneProtos.ReplicationType.STAND_ALONE;
factor = OzoneProtos.ReplicationFactor.ONE;
if (useRatis) {
type = OzoneProtos.ReplicationType.RATIS;
int replicationFactor = Integer.parseInt(cmdLine.getOptionValue(RATIS));
switch (replicationFactor) {
case 1:
factor = OzoneProtos.ReplicationFactor.ONE;
break;
case 3:
factor = OzoneProtos.ReplicationFactor.THREE;
break;
default:
throw new IllegalArgumentException("Illegal replication factor:"
+ replicationFactor);
}
}
}
private void usage() {
@ -464,10 +480,13 @@ private void printStats(PrintStream out) {
out.println();
out.println("***************************************************");
out.println("Status: " + (exception ? "Failed" : "Success"));
out.println("Git Base Revision: " + VersionInfo.getRevision());
out.println("Number of Volumes created: " + numberOfVolumesCreated);
out.println("Number of Buckets created: " + numberOfBucketsCreated);
out.println("Number of Keys added: " + numberOfKeysAdded);
out.println("Ratis replication factor: " + factor.name());
out.println("Ratis replication type: " + type.name());
out.println("Time spent in volume creation: " + prettyTotalVolumeTime);
out.println("Time spent in bucket creation: " + prettyTotalBucketTime);
out.println("Time spent in key creation: " + prettyTotalKeyCreationTime);
@ -658,17 +677,6 @@ private class OfflineProcessor implements Runnable {
@Override
public void run() {
OzoneProtos.ReplicationType type = OzoneProtos.ReplicationType
.STAND_ALONE;
OzoneProtos.ReplicationFactor factor = OzoneProtos.ReplicationFactor.ONE;
if (useRatis) {
type = OzoneProtos.ReplicationType.RATIS;
factor = replicationFactor != 0 ?
OzoneProtos.ReplicationFactor.valueOf(replicationFactor) :
OzoneProtos.ReplicationFactor.THREE;
}
Long threadKeyWriteTime = 0L;
for (int j = 0; j < totalBuckets; j++) {
String bucketName = "bucket-" + j + "-" +
@ -735,6 +743,7 @@ public void run() {
private final class CoronaJobInfo {
private String status;
private String gitBaseRevision;
private String jobStartTime;
private String numOfVolumes;
@ -752,6 +761,8 @@ private final class CoronaJobInfo {
private String averageKeyWriteTime;
private String dataWritten;
private String execTime;
private String replicationFactor;
private String replicationType;
private int keySize;
@ -761,6 +772,7 @@ private final class CoronaJobInfo {
private String totalThroughputPerSecond;
private CoronaJobInfo() {
this.status = exception ? "Failed" : "Success";
this.numOfVolumes = Corona.this.numOfVolumes;
this.numOfBuckets = Corona.this.numOfBuckets;
this.numOfKeys = Corona.this.numOfKeys;
@ -768,6 +780,8 @@ private CoronaJobInfo() {
this.keySize = Corona.this.keySize;
this.mode = Corona.this.mode;
this.jobStartTime = Time.formatTime(Corona.this.jobStartTime);
this.replicationFactor = Corona.this.factor.name();
this.replicationType = Corona.this.type.name();
long totalBytes =
Long.parseLong(numOfVolumes) * Long.parseLong(numOfBuckets) * Long
@ -928,6 +942,18 @@ public String getExecTime() {
return execTime;
}
public String getReplicationFactor() {
return replicationFactor;
}
public String getReplicationType() {
return replicationType;
}
public String getStatus() {
return status;
}
public int getKeySize() {
return keySize;
}

1
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/MiniOzoneCluster.java
View File

@ -155,6 +155,7 @@ public boolean restartDataNode(int i) throws IOException {
* 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 =

4
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/container/transport/server/TestContainerServer.java
View File

@ -53,7 +53,6 @@
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.function.BiConsumer;
import static org.apache.ratis.rpc.SupportedRpcType.GRPC;
@ -124,8 +123,7 @@ static XceiverServerRatis newXceiverServerRatis(
conf.set(OzoneConfigKeys.DFS_CONTAINER_RATIS_DATANODE_STORAGE_DIR, dir);
final ContainerDispatcher dispatcher = new TestContainerDispatcher();
return XceiverServerRatis.newXceiverServerRatis(UUID.randomUUID()
.toString(), conf, dispatcher);
return XceiverServerRatis.newXceiverServerRatis(dn, conf, dispatcher);
}
static void initXceiverServerRatis(

22
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/tools/TestCorona.java
View File

@ -56,7 +56,8 @@ public static void init() throws Exception {
conf.set(OzoneConfigKeys.OZONE_HANDLER_TYPE_KEY,
OzoneConsts.OZONE_HANDLER_DISTRIBUTED);
cluster = new MiniOzoneCluster.Builder(conf)
.setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build();
.setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED)
.numDataNodes(5).build();
}
/**
@ -115,4 +116,23 @@ public void validateWriteTest() throws Exception {
System.setOut(originalStream);
}
@Test
public void ratisTest() throws Exception {
List<String> args = new ArrayList<>();
args.add("-numOfVolumes");
args.add("1");
args.add("-numOfBuckets");
args.add("1");
args.add("-numOfKeys");
args.add("10");
args.add("-ratis");
args.add("3");
Corona corona = new Corona(conf);
int res = ToolRunner.run(conf, corona,
args.toArray(new String[0]));
Assert.assertEquals(1, corona.getNumberOfVolumesCreated());
Assert.assertEquals(1, corona.getNumberOfBucketsCreated());
Assert.assertEquals(10, corona.getNumberOfKeysAdded());
Assert.assertEquals(0, res);
}
}

58
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/ozone/web/client/TestKeysRatis.java
View File

@ -18,53 +18,48 @@
package org.apache.hadoop.ozone.web.client;
import org.apache.commons.lang.RandomStringUtils;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.OzoneConfigKeys;
import org.apache.hadoop.conf.OzoneConfiguration;
import org.apache.hadoop.ozone.OzoneConsts;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.hadoop.ozone.RatisTestHelper;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Rule;
import org.junit.Test;
import org.junit.Ignore;
import org.junit.rules.Timeout;
import static org.apache.hadoop.ozone.web.client.TestKeys.PutHelper;
import static org.apache.hadoop.ozone.web.client.TestKeys.getMultiPartKey;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestGetKeyInfo;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestPutAndDeleteKey;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestPutAndGetKey;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestPutAndGetKeyWithDnRestart;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestPutAndListKey;
import static org.apache.hadoop.ozone.web.client.TestKeys.runTestPutKey;
import static org.apache.hadoop.ozone.web.client
.TestKeys.PutHelper;
import static org.apache.hadoop.ozone.web.client
.TestKeys.getMultiPartKey;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestGetKeyInfo;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestPutAndDeleteKey;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestPutAndGetKey;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestPutAndGetKeyWithDnRestart;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestPutAndListKey;
import static org.apache.hadoop.ozone.web.client
.TestKeys.runTestPutKey;
/** The same as {@link TestKeys} except that this test is Ratis enabled. */
public class TestKeysRatis {
@Rule
public Timeout testTimeout = new Timeout(300000);
private static RatisTestHelper.RatisTestSuite suite;
private static MiniOzoneCluster ozoneCluster = null;
static private String path;
private static OzoneRestClient ozoneRestClient = null;
@BeforeClass
public static void init() throws Exception {
OzoneConfiguration conf = new OzoneConfiguration();
path = GenericTestUtils.getTempPath(TestKeys.class.getSimpleName());
path += conf.getTrimmed(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT,
OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT_DEFAULT);
conf.set(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT, path);
Logger.getLogger("log4j.logger.org.apache.http").setLevel(Level.DEBUG);
ozoneCluster = new MiniOzoneCluster.Builder(conf)
.setHandlerType(OzoneConsts.OZONE_HANDLER_DISTRIBUTED).build();
DataNode dataNode = ozoneCluster.getDataNodes().get(0);
final int port = dataNode.getInfoPort();
ozoneRestClient = new OzoneRestClient(
String.format("http://localhost:%d", port));
suite = new RatisTestHelper.RatisTestSuite(TestBucketsRatis.class);
path = suite.getConf().get(OzoneConfigKeys.OZONE_LOCALSTORAGE_ROOT);
ozoneCluster = suite.getCluster();
ozoneRestClient = suite.newOzoneRestClient();
}
/**
@ -72,8 +67,8 @@ public static void init() throws Exception {
*/
@AfterClass
public static void shutdown() {
if (ozoneCluster != null) {
ozoneCluster.shutdown();
if (suite != null) {
suite.close();
}
}
@ -86,6 +81,7 @@ public void testPutKey() throws Exception {
getMultiPartKey(delimiter)));
}
@Ignore("disabling for now, datanodes restart with ratis is buggy")
@Test
public void testPutAndGetKeyWithDnRestart() throws Exception {
runTestPutAndGetKeyWithDnRestart(