You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by su...@apache.org on 2018/10/24 05:52:50 UTC
[30/50] [abbrv] hadoop git commit: HDDS-638. Enable ratis snapshots
for HDDS datanodes. Contributed by Mukul Kumar Singh.
HDDS-638. Enable ratis snapshots for HDDS datanodes. Contributed by Mukul Kumar Singh.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/82919a1e
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/82919a1e
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/82919a1e
Branch: refs/heads/HDFS-12943
Commit: 82919a1e7af2ec22993f273e9ee64512a688c99b
Parents: 52cb766
Author: Mukul Kumar Singh <ms...@apache.org>
Authored: Mon Oct 22 19:59:35 2018 +0530
Committer: Mukul Kumar Singh <ms...@apache.org>
Committed: Mon Oct 22 19:59:35 2018 +0530
----------------------------------------------------------------------
.../apache/hadoop/hdds/scm/ScmConfigKeys.java | 4 +
.../apache/hadoop/ozone/OzoneConfigKeys.java | 4 +
.../common/src/main/resources/ozone-default.xml | 8 ++
.../server/ratis/ContainerStateMachine.java | 85 +++++++++++--
.../server/ratis/XceiverServerRatis.java | 14 ++-
.../container/ozoneimpl/OzoneContainer.java | 5 +
.../freon/TestFreonWithDatanodeRestart.java | 122 +++++++++++++++++++
7 files changed, 233 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
index 5910eda..f95b748 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ScmConfigKeys.java
@@ -104,6 +104,10 @@ public final class ScmConfigKeys {
DFS_RATIS_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_DEFAULT =
TimeDuration.valueOf(1, TimeUnit.SECONDS);
+ public static final String DFS_RATIS_SNAPSHOT_THRESHOLD_KEY =
+ "dfs.ratis.snapshot.threshold";
+ public static final long DFS_RATIS_SNAPSHOT_THRESHOLD_DEFAULT = 10000;
+
public static final String DFS_RATIS_SERVER_FAILURE_DURATION_KEY =
"dfs.ratis.server.failure.duration";
public static final TimeDuration
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
index 09d798f..c931dcf 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java
@@ -266,6 +266,10 @@ public final class OzoneConfigKeys {
public static final TimeDuration
DFS_RATIS_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_DEFAULT =
ScmConfigKeys.DFS_RATIS_LEADER_ELECTION_MINIMUM_TIMEOUT_DURATION_DEFAULT;
+ public static final String DFS_RATIS_SNAPSHOT_THRESHOLD_KEY =
+ ScmConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_KEY;
+ public static final long DFS_RATIS_SNAPSHOT_THRESHOLD_DEFAULT =
+ ScmConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_DEFAULT;
public static final String DFS_RATIS_SERVER_FAILURE_DURATION_KEY =
ScmConfigKeys.DFS_RATIS_SERVER_FAILURE_DURATION_KEY;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/common/src/main/resources/ozone-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-hdds/common/src/main/resources/ozone-default.xml b/hadoop-hdds/common/src/main/resources/ozone-default.xml
index b2d47c9..237f8d8 100644
--- a/hadoop-hdds/common/src/main/resources/ozone-default.xml
+++ b/hadoop-hdds/common/src/main/resources/ozone-default.xml
@@ -119,6 +119,14 @@
</description>
</property>
<property>
+ <name>dfs.ratis.snapshot.threshold</name>
+ <value>10000</value>
+ <tag>OZONE, RATIS</tag>
+ <description>Number of transactions after which a ratis snapshot should be
+ taken.
+ </description>
+ </property>
+ <property>
<name>dfs.container.ratis.num.write.chunk.threads</name>
<value>60</value>
<tag>OZONE, RATIS, PERFORMANCE</tag>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
index 7b7be91..fa9fbf3 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java
@@ -25,6 +25,9 @@ import org.apache.hadoop.ozone.container.common.helpers.BlockData;
import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.impl.RaftServerConstants;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.statemachine.impl.SingleFileSnapshotInfo;
import org.apache.ratis.thirdparty.com.google.protobuf
.InvalidProtocolBufferException;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
@@ -55,8 +58,10 @@ import org.apache.ratis.statemachine.impl.TransactionContextImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
@@ -115,6 +120,7 @@ public class ContainerStateMachine extends BaseStateMachine {
createContainerFutureMap;
private ExecutorService[] executors;
private final int numExecutors;
+ private final Map<Long, Long> containerCommandCompletionMap;
/**
* CSM metrics.
*/
@@ -131,6 +137,7 @@ public class ContainerStateMachine extends BaseStateMachine {
this.createContainerFutureMap = new ConcurrentHashMap<>();
this.numExecutors = numOfExecutors;
executors = new ExecutorService[numExecutors];
+ containerCommandCompletionMap = new ConcurrentHashMap<>();
for (int i = 0; i < numExecutors; i++) {
executors[i] = Executors.newSingleThreadExecutor();
}
@@ -151,10 +158,47 @@ public class ContainerStateMachine extends BaseStateMachine {
throws IOException {
super.initialize(server, id, raftStorage);
storage.init(raftStorage);
- // TODO handle snapshots
- // TODO: Add a flag that tells you that initialize has been called.
- // Check with Ratis if this feature is done in Ratis.
+ loadSnapshot(storage.getLatestSnapshot());
+ }
+
+ private long loadSnapshot(SingleFileSnapshotInfo snapshot) {
+ if (snapshot == null) {
+ TermIndex empty = TermIndex.newTermIndex(0, 0);
+ LOG.info("The snapshot info is null." +
+ "Setting the last applied index to:" + empty);
+ setLastAppliedTermIndex(empty);
+ return RaftServerConstants.INVALID_LOG_INDEX;
+ }
+
+ final TermIndex last =
+ SimpleStateMachineStorage.getTermIndexFromSnapshotFile(
+ snapshot.getFile().getPath().toFile());
+ LOG.info("Setting the last applied index to " + last);
+ setLastAppliedTermIndex(last);
+ return last.getIndex();
+ }
+
+ @Override
+ public long takeSnapshot() throws IOException {
+ TermIndex ti = getLastAppliedTermIndex();
+ LOG.info("Taking snapshot at termIndex:" + ti);
+ if (ti != null) {
+ final File snapshotFile =
+ storage.getSnapshotFile(ti.getTerm(), ti.getIndex());
+ LOG.info("Taking a snapshot to file {}", snapshotFile);
+ try {
+ //TODO: For now, just create the file to save the term index,
+ //persist open container info to snapshot later.
+ snapshotFile.createNewFile();
+ } catch(IOException ioe) {
+ LOG.warn("Failed to write snapshot file \"" + snapshotFile
+ + "\", last applied index=" + ti);
+ throw ioe;
+ }
+ return ti.getIndex();
+ }
+ return -1;
}
@Override
@@ -353,10 +397,9 @@ public class ContainerStateMachine extends BaseStateMachine {
public CompletableFuture<Void> flushStateMachineData(long index) {
List<CompletableFuture<Message>> futureList =
writeChunkFutureMap.entrySet().stream().filter(x -> x.getKey() <= index)
- .map(x -> x.getValue()).collect(Collectors.toList());
- CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
+ .map(Map.Entry::getValue).collect(Collectors.toList());
+ return CompletableFuture.allOf(
futureList.toArray(new CompletableFuture[futureList.size()]));
- return combinedFuture;
}
/*
* This api is used by the leader while appending logs to the follower
@@ -394,11 +437,28 @@ public class ContainerStateMachine extends BaseStateMachine {
}
}
+ private void updateLastApplied() {
+ Long appliedTerm = null;
+ long appliedIndex = -1;
+ for(long i = getLastAppliedTermIndex().getIndex() + 1;; i++) {
+ final Long removed = containerCommandCompletionMap.remove(i);
+ if (removed == null) {
+ break;
+ }
+ appliedTerm = removed;
+ appliedIndex = i;
+ }
+ if (appliedTerm != null) {
+ updateLastAppliedTermIndex(appliedIndex, appliedTerm);
+ }
+ }
+
/*
* ApplyTransaction calls in Ratis are sequential.
*/
@Override
public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
+ long index = trx.getLogEntry().getIndex();
try {
metrics.incNumApplyTransactionsOps();
ContainerCommandRequestProto requestProto =
@@ -418,7 +478,7 @@ public class ContainerStateMachine extends BaseStateMachine {
blockDataProto.getBlockID());
return completeExceptionally(ioe);
}
- blockData.setBlockCommitSequenceId(trx.getLogEntry().getIndex());
+ blockData.setBlockCommitSequenceId(index);
final ContainerProtos.PutBlockRequestProto putBlockRequestProto =
ContainerProtos.PutBlockRequestProto
.newBuilder(requestProto.getPutBlock())
@@ -440,6 +500,14 @@ public class ContainerStateMachine extends BaseStateMachine {
future.thenApply(
r -> createContainerFutureMap.remove(containerID).complete(null));
}
+
+ future.thenAccept(m -> {
+ final Long previous =
+ containerCommandCompletionMap
+ .put(index, trx.getLogEntry().getTerm());
+ Preconditions.checkState(previous == null);
+ updateLastApplied();
+ });
return future;
} catch (IOException e) {
metrics.incNumApplyTransactionsFails();
@@ -466,7 +534,8 @@ public class ContainerStateMachine extends BaseStateMachine {
@Override
public void close() throws IOException {
- for (int i = 0; i < numExecutors; i++){
+ takeSnapshot();
+ for (int i = 0; i < numExecutors; i++) {
executors[i].shutdown();
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
index 0cd407a..f0c2845 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/XceiverServerRatis.java
@@ -130,6 +130,10 @@ public final class XceiverServerRatis implements XceiverServerSpi {
.build();
}
+ @VisibleForTesting
+ public ContainerStateMachine getStateMachine() {
+ return stateMachine;
+ }
private RaftProperties newRaftProperties(Configuration conf) {
final RaftProperties properties = new RaftProperties();
@@ -254,6 +258,15 @@ public final class XceiverServerRatis implements XceiverServerSpi {
} else if (rpc == SupportedRpcType.NETTY) {
NettyConfigKeys.Server.setPort(properties, port);
}
+
+ long snapshotThreshold =
+ conf.getLong(OzoneConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_KEY,
+ OzoneConfigKeys.DFS_RATIS_SNAPSHOT_THRESHOLD_DEFAULT);
+ RaftServerConfigKeys.Snapshot.
+ setAutoTriggerEnabled(properties, true);
+ RaftServerConfigKeys.Snapshot.
+ setAutoTriggerThreshold(properties, snapshotThreshold);
+
return properties;
}
@@ -298,7 +311,6 @@ public final class XceiverServerRatis implements XceiverServerSpi {
public void stop() {
try {
chunkExecutor.shutdown();
- stateMachine.close();
server.close();
} catch (IOException e) {
throw new RuntimeException(e);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
----------------------------------------------------------------------
diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
index 503112d..9fac3cb 100644
--- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
+++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java
@@ -230,6 +230,11 @@ public class OzoneContainer {
return this.hddsDispatcher;
}
+ @VisibleForTesting
+ public XceiverServerSpi getServer(ReplicationType replicationType) {
+ return servers.get(replicationType);
+ }
+
public VolumeSet getVolumeSet() {
return volumeSet;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/82919a1e/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeRestart.java
----------------------------------------------------------------------
diff --git a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeRestart.java b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeRestart.java
new file mode 100644
index 0000000..a1c50b6
--- /dev/null
+++ b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/freon/TestFreonWithDatanodeRestart.java
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.freon;
+
+import org.apache.hadoop.hdds.client.ReplicationFactor;
+import org.apache.hadoop.hdds.client.ReplicationType;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi;
+import org.apache.hadoop.ozone.container.common.transport.server.ratis.ContainerStateMachine;
+import org.apache.hadoop.ozone.container.common.transport.server.ratis.XceiverServerRatis;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
+import org.apache.ratis.statemachine.impl.SingleFileSnapshotInfo;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL;
+
+/**
+ * Tests Freon with Datanode restarts.
+ */
+public class TestFreonWithDatanodeRestart {
+
+ private static MiniOzoneCluster cluster;
+ private static OzoneConfiguration conf;
+
+ /**
+ * Create a MiniDFSCluster for testing.
+ * <p>
+ * Ozone is made active by setting OZONE_ENABLED = true
+ *
+ */
+ @BeforeClass
+ public static void init() throws Exception {
+ conf = new OzoneConfiguration();
+ conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 5, TimeUnit.SECONDS);
+ cluster = MiniOzoneCluster.newBuilder(conf)
+ .setHbProcessorInterval(1000)
+ .setHbInterval(1000)
+ .setNumDatanodes(3)
+ .build();
+ cluster.waitForClusterToBeReady();
+ }
+
+ /**
+ * Shutdown MiniDFSCluster.
+ */
+ @AfterClass
+ public static void shutdown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ @Test
+ public void testRestart() throws Exception {
+ RandomKeyGenerator randomKeyGenerator =
+ new RandomKeyGenerator((OzoneConfiguration) cluster.getConf());
+ randomKeyGenerator.setNumOfVolumes(1);
+ randomKeyGenerator.setNumOfBuckets(1);
+ randomKeyGenerator.setNumOfKeys(1);
+ randomKeyGenerator.setType(ReplicationType.RATIS);
+ randomKeyGenerator.setFactor(ReplicationFactor.THREE);
+ randomKeyGenerator.setKeySize(20971520);
+ randomKeyGenerator.setValidateWrites(true);
+ randomKeyGenerator.call();
+ Assert.assertEquals(1, randomKeyGenerator.getNumberOfVolumesCreated());
+ Assert.assertEquals(1, randomKeyGenerator.getNumberOfBucketsCreated());
+ Assert.assertEquals(1, randomKeyGenerator.getNumberOfKeysAdded());
+ Assert.assertEquals(0, randomKeyGenerator.getUnsuccessfulValidationCount());
+
+ ContainerStateMachine sm = getStateMachine();
+ TermIndex termIndexBeforeRestart = sm.getLastAppliedTermIndex();
+ cluster.restartHddsDatanode(0);
+ sm = getStateMachine();
+ SimpleStateMachineStorage storage =
+ (SimpleStateMachineStorage)sm.getStateMachineStorage();
+ SingleFileSnapshotInfo snapshotInfo = storage.getLatestSnapshot();
+ TermIndex termInSnapshot = snapshotInfo.getTermIndex();
+ String expectedSnapFile =
+ storage.getSnapshotFile(termIndexBeforeRestart.getTerm(),
+ termIndexBeforeRestart.getIndex()).getAbsolutePath();
+ Assert.assertEquals(snapshotInfo.getFile().getPath().toString(),
+ expectedSnapFile);
+ Assert.assertEquals(termInSnapshot, termIndexBeforeRestart);
+
+ // After restart the term index might have progressed to apply pending
+ // transactions.
+ TermIndex termIndexAfterRestart = sm.getLastAppliedTermIndex();
+ Assert.assertTrue(termIndexAfterRestart.getIndex() >=
+ termIndexBeforeRestart.getIndex());
+ }
+
+ private ContainerStateMachine getStateMachine() {
+ XceiverServerSpi server =
+ cluster.getHddsDatanodes().get(0).getDatanodeStateMachine().
+ getContainer().getServer(HddsProtos.ReplicationType.RATIS);
+ return ((XceiverServerRatis)server).getStateMachine();
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org