You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by na...@apache.org on 2021/01/06 05:17:13 UTC
[ozone] branch HDDS-2823 updated (c75d5b3 -> b0d79e7)
This is an automated email from the ASF dual-hosted git repository.
nanda pushed a change to branch HDDS-2823
in repository https://gitbox.apache.org/repos/asf/ozone.git.
discard c75d5b3 Merge branch 'master' into HDDS-2823
omit 13e02bc HDDS-4637. Improve the log infomation when CSM take snapshot (#1749)
omit 29335c5 HDDS-4644. Block token verification failed: no READ permission for WriteChunk (#1758)
omit 888d555 HDDS-4640. Intermittent failure in MapReduce test due to existing output file (#1753)
omit c3f6b4b HDDS-4558. Support Ozone block token with access mode check. (#1672)
omit d044afb HDDS-4507. Add SCM CA CLI to query certificate. (#1662)
new b0d79e7 Merge branch 'master' into HDDS-2823
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (c75d5b3)
\
N -- N -- N refs/heads/HDDS-2823 (b0d79e7)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../server/ratis/ContainerStateMachine.java | 4 +-
.../hadoop/hdds/protocol/SCMSecurityProtocol.java | 15 ---
.../SCMSecurityProtocolClientSideTranslatorPB.java | 27 ------
.../hdds/security/token/BlockTokenVerifier.java | 31 +------
.../certificate/authority/CertificateServer.java | 12 ---
.../certificate/authority/CertificateStore.java | 16 ----
.../certificate/authority/DefaultCAServer.java | 19 ----
.../x509/certificate/authority/MockCAStore.java | 11 ---
.../src/main/proto/ScmServerSecurityProtocol.proto | 26 ------
.../hadoop/hdds/scm/ha/SCMHAManagerImpl.java | 1 +
.../SCMSecurityProtocolServerSideTranslatorPB.java | 25 -----
.../hadoop/hdds/scm/server/SCMCertStore.java | 42 ---------
.../hdds/scm/server/SCMSecurityProtocolServer.java | 32 +------
.../org/apache/hadoop/hdds/scm/cli/ScmOption.java | 13 ---
.../hadoop/hdds/scm/cli/cert/CertCommands.java | 61 ------------
.../hadoop/hdds/scm/cli/cert/InfoSubcommand.java | 73 ---------------
.../hadoop/hdds/scm/cli/cert/ListSubcommand.java | 102 ---------------------
.../hdds/scm/cli/cert/ScmCertSubcommand.java | 44 ---------
.../hadoop/hdds/scm/cli/cert/package-info.java | 22 -----
.../dist/src/main/smoketest/mapreduce.robot | 2 +-
.../src/main/smoketest/security/admin-cert.robot | 44 ---------
.../org/apache/hadoop/ozone/om/KeyManagerImpl.java | 43 ++++-----
.../hadoop/ozone/om/request/key/OMKeyRequest.java | 17 +++-
.../security/OzoneBlockTokenSecretManager.java | 8 +-
.../security/TestOzoneBlockTokenSecretManager.java | 38 --------
25 files changed, 49 insertions(+), 679 deletions(-)
delete mode 100644 hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/cert/CertCommands.java
delete mode 100644 hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/cert/InfoSubcommand.java
delete mode 100644 hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/cert/ListSubcommand.java
delete mode 100644 hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/cert/ScmCertSubcommand.java
delete mode 100644 hadoop-hdds/tools/src/main/java/org/apache/hadoop/hdds/scm/cli/cert/package-info.java
delete mode 100644 hadoop-ozone/dist/src/main/smoketest/security/admin-cert.robot
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org
[ozone] 01/01: Merge branch 'master' into HDDS-2823
Posted by na...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
nanda pushed a commit to branch HDDS-2823
in repository https://gitbox.apache.org/repos/asf/ozone.git
commit b0d79e7a3b6da0abc17cf57eceb068f08628ead0
Merge: b32c2bb 4128813
Author: Nandakumar <na...@apache.org>
AuthorDate: Wed Jan 6 10:46:35 2021 +0530
Merge branch 'master' into HDDS-2823
.../apache/hadoop/hdds/scm/XceiverClientRatis.java | 2 +-
.../hadoop/hdds/conf/OzoneConfiguration.java | 3 +
.../java/org/apache/hadoop/hdds/scm/ScmConfig.java | 2 +-
.../org/apache/hadoop/hdds/scm/ScmConfigKeys.java | 2 +-
.../java/org/apache/hadoop/ozone/OzoneConsts.java | 5 +
.../common/src/main/resources/ozone-default.xml | 6 +-
.../container/common/helpers/ContainerUtils.java | 14 +-
.../server/ratis/ContainerStateMachine.java | 9 +-
.../transport/server/ratis/XceiverServerRatis.java | 16 +-
.../ozone/container/ContainerTestHelper.java | 29 +-
hadoop-hdds/docs/config.yaml | 2 +
.../docs/content/concept/OzoneManager.zh.md | 4 +-
hadoop-hdds/docs/content/concept/Recon.zh.md | 116 +++++
.../content/concept/StorageContainerManager.zh.md | 6 +-
hadoop-hdds/docs/content/feature/Quota.md | 66 ++-
hadoop-hdds/docs/content/feature/Quota.zh.md | 62 ++-
hadoop-hdds/docs/content/feature/Recon.zh.md | 23 +-
hadoop-hdds/docs/content/interface/ReconApi.zh.md | 502 +++++++++++++++++++++
.../docs/content/security/SecuringOzoneHTTP.md | 12 +-
hadoop-hdds/docs/dev-support/bin/generate-site.sh | 9 +-
.../themes/ozonedoc/layouts/_default/baseof.html | 2 +
.../themes/ozonedoc/layouts/_default/section.html | 3 +
.../themes/ozonedoc/layouts/_default/single.html | 3 +
.../docs/themes/ozonedoc/layouts/index.html | 26 +-
.../themes/ozonedoc/layouts/partials/footer.html | 9 +
.../docs/themes/ozonedoc/static/css/ozonedoc.css | 30 +-
.../org/apache/hadoop/hdds/utils/db/DBStore.java | 14 +-
.../org/apache/hadoop/hdds/utils/db/RDBStore.java | 6 +-
.../apache/hadoop/hdds/utils/db/TypedTable.java | 24 +-
.../{TableCacheImpl.java => FullTableCache.java} | 143 +++---
...{TableCacheImpl.java => PartialTableCache.java} | 122 ++---
.../hadoop/hdds/utils/db/cache/TableCache.java | 30 +-
...TestTableCacheImpl.java => TestTableCache.java} | 144 ++++--
.../hadoop/hdds/scm/ha/SCMHAManagerImpl.java | 28 +-
.../hdds/scm/server/SCMContainerMetrics.java | 11 +-
.../scm/server/ratis/SCMRatisSnapshotInfo.java | 2 +-
.../hdds/scm/server/ratis/SCMStateMachine.java | 4 +-
.../hdds/scm/server/TestSCMContainerMetrics.java | 2 +
.../apache/hadoop/ozone/client/OzoneBucket.java | 16 +-
.../apache/hadoop/ozone/client/OzoneVolume.java | 4 +-
.../apache/hadoop/ozone/client/rpc/RpcClient.java | 2 +
.../java/org/apache/hadoop}/ozone/OFSPath.java | 7 +-
.../apache/hadoop/ozone/conf/OMClientConfig.java | 19 +-
.../org/apache/hadoop/ozone/om/OMConfigKeys.java | 2 +-
.../hadoop/ozone/om/exceptions/OMException.java | 4 +-
.../hadoop/ozone/om/helpers/OmBucketInfo.java | 28 +-
.../hadoop/ozone/om/helpers/OmOzoneAclMap.java | 97 ++--
.../hadoop/ozone/om/helpers/TestOmOzoneAclMap.java | 56 +++
hadoop-ozone/datanode/pom.xml | 5 -
.../dist/src/main/compose/ozone-mr/test.sh | 1 +
hadoop-ozone/dist/src/main/compose/upgrade/test.sh | 6 +-
.../dist/src/main/dockerlibexec/transformation.py | 8 +-
.../src/main/smoketest/basic/ozone-shell-lib.robot | 12 +-
.../hadoop/fs/ozone/TestOzoneFileSystem.java | 59 ++-
.../fs/ozone/TestOzoneFileSystemMissingParent.java | 127 ++++++
.../hadoop/fs/ozone/TestRootedOzoneFileSystem.java | 1 +
.../rpc/TestOzoneClientRetriesOnException.java | 32 +-
.../client/rpc/TestOzoneRpcClientAbstract.java | 77 +++-
.../apache/hadoop/ozone/container/TestHelper.java | 24 +-
.../ozone/freon/TestHadoopDirTreeGenerator.java | 4 +-
.../ozone/freon/TestHadoopNestedDirGenerator.java | 4 +-
.../ozone/freon/TestOzoneClientKeyGenerator.java | 4 +-
.../hadoop/ozone/om/TestOMRatisSnapshots.java | 2 +-
.../ozone/om/parser/TestOMRatisLogParser.java | 14 +-
.../hadoop/ozone/shell/TestOzoneShellHA.java | 2 +-
.../src/main/proto/OmClientProtocol.proto | 9 +-
.../hadoop/ozone/om/ratis/OMTransactionInfo.java | 2 +-
.../hadoop/ozone/om/OmMetadataManagerImpl.java | 9 +-
.../org/apache/hadoop/ozone/om/OzoneManager.java | 33 +-
.../apache/hadoop/ozone/om/TrashPolicyOzone.java | 62 ++-
.../hadoop/ozone/om/codec/OMDBDefinition.java | 49 +-
.../hadoop/ozone/om/ratis/OMRatisSnapshotInfo.java | 2 +-
.../ozone/om/ratis/OzoneManagerDoubleBuffer.java | 34 +-
.../ozone/om/ratis/OzoneManagerRatisServer.java | 25 +-
.../ozone/om/ratis/OzoneManagerStateMachine.java | 4 +-
.../om/request/bucket/OMBucketCreateRequest.java | 8 +-
.../request/bucket/OMBucketSetPropertyRequest.java | 7 +
.../ozone/om/request/file/OMFileCreateRequest.java | 3 +
.../ozone/om/request/key/OMKeyCommitRequest.java | 9 +
.../ozone/om/request/key/OMKeyCreateRequest.java | 3 +
.../ozone/om/request/key/OMKeyDeleteRequest.java | 1 +
.../hadoop/ozone/om/request/key/OMKeyRequest.java | 19 +
.../ozone/om/request/key/OMKeysDeleteRequest.java | 1 +
.../apache/hadoop/ozone/om/TestOMDBDefinition.java | 74 +++
.../om/ratis/TestOzoneManagerRatisServer.java | 4 +-
.../hadoop/fs/ozone/BasicOzoneFileSystem.java | 22 +-
.../ozone/BasicRootedOzoneClientAdapterImpl.java | 1 +
.../fs/ozone/BasicRootedOzoneFileSystem.java | 23 +-
.../org/apache/hadoop/fs/ozone/TestOFSPath.java | 1 +
.../org/apache/hadoop/ozone/debug/DBScanner.java | 6 +-
.../hadoop/ozone/shell/ClearSpaceQuotaOptions.java | 8 +-
.../ozone/shell/bucket/ClearQuotaHandler.java | 4 +-
.../hadoop/ozone/shell/bucket/SetQuotaHandler.java | 6 +-
.../ozone/shell/volume/ClearQuotaHandler.java | 4 +-
.../hadoop/ozone/shell/volume/SetQuotaHandler.java | 6 +-
pom.xml | 6 +-
96 files changed, 1966 insertions(+), 589 deletions(-)
diff --cc hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
index 368d03c,d2d6e35..6ddf5c5
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java
@@@ -382,13 -382,11 +382,18 @@@ public final class OzoneConsts
public static final String CONTAINER_DB_TYPE_ROCKSDB = "RocksDB";
public static final String CONTAINER_DB_TYPE_LEVELDB = "LevelDB";
+ // SCM HA
+ public static final String SCM_SERVICE_ID_DEFAULT = "scmServiceIdDefault";
+
+ // SCM Ratis snapshot file to store the last applied index
+ public static final String SCM_RATIS_SNAPSHOT_INDEX = "scmRatisSnapshotIndex";
+
+ public static final String SCM_RATIS_SNAPSHOT_TERM = "scmRatisSnapshotTerm";
// An on-disk transient marker file used when replacing DB with checkpoint
public static final String DB_TRANSIENT_MARKER = "dbInconsistentMarker";
+
+
+ public static final String OM_RATIS_SNAPSHOT_DIR = "snapshot";
+
+ public static final long DEFAULT_OM_UPDATE_ID = -1L;
}
diff --cc hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
index ae91fc2,0000000..bac6336
mode 100644,000000..100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManagerImpl.java
@@@ -1,109 -1,0 +1,93 @@@
+/*
+ * 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>
+ * <p>http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * <p>Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.ha;
+
- import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.conf.ConfigurationSource;
+import org.apache.ratis.proto.RaftProtos;
- import org.apache.ratis.server.RaftServer;
- import org.apache.ratis.server.impl.RaftServerImpl;
- import org.apache.ratis.server.impl.RaftServerProxy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Optional;
+
+/**
+ * SCMHAManagerImpl uses Apache Ratis for HA implementation. We will have 2N+1
+ * node Ratis ring. The Ratis ring will have one Leader node and 2N follower
+ * nodes.
+ *
+ * TODO
+ *
+ */
+public class SCMHAManagerImpl implements SCMHAManager {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(SCMHAManagerImpl.class);
+
+ private final SCMRatisServer ratisServer;
+ private final ConfigurationSource conf;
+
+ /**
+ * Creates SCMHAManager instance.
+ */
+ public SCMHAManagerImpl(final ConfigurationSource conf) throws IOException {
+ this.conf = conf;
+ this.ratisServer = new SCMRatisServerImpl(
+ conf.getObject(SCMHAConfiguration.class), conf);
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void start() throws IOException {
+ ratisServer.start();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public Optional<Long> isLeader() {
+ if (!SCMHAUtils.isSCMHAEnabled(conf)) {
+ // When SCM HA is not enabled, the current SCM is always the leader.
+ return Optional.of((long)0);
+ }
- RaftServer server = ratisServer.getDivision().getRaftServer();
- Preconditions.checkState(server instanceof RaftServerProxy);
- try {
- // SCM only has one raft group.
- RaftServerImpl serverImpl = ((RaftServerProxy) server)
- .getImpl(ratisServer.getDivision().getGroup().getGroupId());
- if (serverImpl != null) {
- // TODO: getRoleInfoProto() will be exposed from Division later.
- RaftProtos.RoleInfoProto roleInfoProto = serverImpl.getRoleInfoProto();
- return roleInfoProto.hasLeaderInfo()
- ? Optional.of(roleInfoProto.getLeaderInfo().getTerm())
- : Optional.empty();
- }
- } catch (IOException ioe) {
- LOG.error("Fail to get RaftServer impl and therefore it's not clear " +
- "whether it's leader. ", ioe);
- }
- return Optional.empty();
++ RaftProtos.RoleInfoProto roleInfoProto
++ = ratisServer.getDivision().getInfo().getRoleInfoProto();
++
++ return roleInfoProto.hasLeaderInfo()
++ ? Optional.of(roleInfoProto.getLeaderInfo().getTerm())
++ : Optional.empty();
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public SCMRatisServer getRatisServer() {
+ return ratisServer;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ public void shutdown() throws IOException {
+ ratisServer.stop();
+ }
+}
diff --cc hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMRatisSnapshotInfo.java
index 11b3234,0000000..ffee99c
mode 100644,000000..100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMRatisSnapshotInfo.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMRatisSnapshotInfo.java
@@@ -1,179 -1,0 +1,179 @@@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with this
+ * work for additional information regarding copyright ownership. The ASF
+ * licenses this file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.ratis;
+
+import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.server.storage.FileInfo;
+import org.apache.ratis.statemachine.SnapshotInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.DumperOptions;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStreamWriter;
+import java.io.Writer;
+import java.util.List;
+
+import static org.apache.hadoop.ozone.OzoneConsts.SCM_RATIS_SNAPSHOT_INDEX;
+
+/**
+ * This class captures the snapshotIndex and term of the latest snapshot in
+ * the SCM.
+ * Ratis server loads the snapshotInfo during startup and updates the
+ * lastApplied index to this snapshotIndex. OM SnapshotInfo does not contain
+ * any files. It is used only to store/ update the last applied index and term.
+ */
+public class SCMRatisSnapshotInfo implements SnapshotInfo {
+
+ static final Logger LOG = LoggerFactory.getLogger(SCMRatisSnapshotInfo.class);
+
+ private volatile long term = 0;
+ private volatile long snapshotIndex = -1;
+
+ private final File ratisSnapshotFile;
+
+ public SCMRatisSnapshotInfo(File ratisDir) throws IOException {
+ ratisSnapshotFile = new File(ratisDir, SCM_RATIS_SNAPSHOT_INDEX);
+ loadRatisSnapshotIndex();
+ }
+
+ public void updateTerm(long newTerm) {
+ term = newTerm;
+ }
+
+ private void updateTermIndex(long newTerm, long newIndex) {
+ this.term = newTerm;
+ this.snapshotIndex = newIndex;
+ }
+
+ /**
+ * Load the snapshot index and term from the snapshot file on disk,
+ * if it exists.
+ * @throws IOException
+ */
+ private void loadRatisSnapshotIndex() throws IOException {
+ if (ratisSnapshotFile.exists()) {
+ RatisSnapshotYaml ratisSnapshotYaml = readRatisSnapshotYaml();
+ updateTermIndex(ratisSnapshotYaml.term, ratisSnapshotYaml.snapshotIndex);
+ }
+ }
+
+ /**
+ * Read and parse the snapshot yaml file.
+ */
+ private RatisSnapshotYaml readRatisSnapshotYaml() throws IOException {
+ try (FileInputStream inputFileStream = new FileInputStream(
+ ratisSnapshotFile)) {
+ Yaml yaml = new Yaml();
+ try {
+ return yaml.loadAs(inputFileStream, RatisSnapshotYaml.class);
+ } catch (Exception e) {
+ throw new IOException("Unable to parse RatisSnapshot yaml file.", e);
+ }
+ }
+ }
+
+ /**
+ * Update and persist the snapshot index and term to disk.
+ * @param lastAppliedTermIndex new snapshot index to be persisted to disk.
+ * @throws IOException
+ */
+ public void saveRatisSnapshotToDisk(TermIndex lastAppliedTermIndex)
+ throws IOException {
+ updateTermIndex(lastAppliedTermIndex.getTerm(),
+ lastAppliedTermIndex.getIndex());
+ writeRatisSnapshotYaml();
+ LOG.info("Saved Ratis Snapshot on the SCM with snapshotIndex {}",
+ lastAppliedTermIndex);
+ }
+
+ /**
+ * Write snapshot details to disk in yaml format.
+ */
+ private void writeRatisSnapshotYaml() throws IOException {
+ DumperOptions options = new DumperOptions();
+ options.setPrettyFlow(true);
+ options.setDefaultFlowStyle(DumperOptions.FlowStyle.FLOW);
+ Yaml yaml = new Yaml(options);
+
+ RatisSnapshotYaml ratisSnapshotYaml = new RatisSnapshotYaml(term,
+ snapshotIndex);
+
+ try (Writer writer = new OutputStreamWriter(
+ new FileOutputStream(ratisSnapshotFile), "UTF-8")) {
+ yaml.dump(ratisSnapshotYaml, writer);
+ }
+ }
+
+ @Override
+ public TermIndex getTermIndex() {
- return TermIndex.newTermIndex(term, snapshotIndex);
++ return TermIndex.valueOf(term, snapshotIndex);
+ }
+
+ @Override
+ public long getTerm() {
+ return term;
+ }
+
+ @Override
+ public long getIndex() {
+ return snapshotIndex;
+ }
+
+ @Override
+ public List<FileInfo> getFiles() {
+ return null;
+ }
+
+ /**
+ * Ratis Snapshot details to be written to the yaml file.
+ */
+ public static class RatisSnapshotYaml {
+ private long term;
+ private long snapshotIndex;
+
+ public RatisSnapshotYaml() {
+ // Needed for snake-yaml introspection.
+ }
+
+ RatisSnapshotYaml(long term, long snapshotIndex) {
+ this.term = term;
+ this.snapshotIndex = snapshotIndex;
+ }
+
+ public void setTerm(long term) {
+ this.term = term;
+ }
+
+ public long getTerm() {
+ return this.term;
+ }
+
+ public void setSnapshotIndex(long index) {
+ this.snapshotIndex = index;
+ }
+
+ public long getSnapshotIndex() {
+ return this.snapshotIndex;
+ }
+ }
+}
diff --cc hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMStateMachine.java
index 1bc71e4,0000000..ada80c4
mode 100644,000000..100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMStateMachine.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/ratis/SCMStateMachine.java
@@@ -1,241 -1,0 +1,241 @@@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.server.ratis;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
+import org.apache.ratis.proto.RaftProtos;
+import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftClientRequest;
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.server.storage.RaftStorage;
+import org.apache.ratis.statemachine.SnapshotInfo;
+import org.apache.ratis.statemachine.TransactionContext;
+import org.apache.ratis.statemachine.impl.BaseStateMachine;
+import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
+import org.apache.ratis.util.LifeCycle;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Class for SCM StateMachine.
+ */
+public class SCMStateMachine extends BaseStateMachine {
+ static final Logger LOG =
+ LoggerFactory.getLogger(SCMStateMachine.class);
+ private final SimpleStateMachineStorage storage =
+ new SimpleStateMachineStorage();
+ private final SCMRatisServer scmRatisServer;
+ private final StorageContainerManager scm;
+ private RaftGroupId raftGroupId;
+ private final SCMRatisSnapshotInfo snapshotInfo;
+ private final ExecutorService executorService;
+ private final ExecutorService installSnapshotExecutor;
+
+ // Map which contains index and term for the ratis transactions which are
+ // stateMachine entries which are recived through applyTransaction.
+ private ConcurrentMap<Long, Long> applyTransactionMap =
+ new ConcurrentSkipListMap<>();
+
+ /**
+ * Create a SCM state machine.
+ */
+ public SCMStateMachine(SCMRatisServer ratisServer) {
+ this.scmRatisServer = ratisServer;
+ this.scm = ratisServer.getSCM();
+
+ // TODO: remove the whole file later
+ this.snapshotInfo = null;
+ updateLastAppliedIndexWithSnaphsotIndex();
+
+ ThreadFactory build = new ThreadFactoryBuilder().setDaemon(true)
+ .setNameFormat("SCM StateMachine ApplyTransaction Thread - %d").build();
+ this.executorService = HadoopExecutors.newSingleThreadExecutor(build);
+ this.installSnapshotExecutor = HadoopExecutors.newSingleThreadExecutor();
+ }
+
+ /**
+ * Initializes the State Machine with the given server, group and storage.
+ */
+ @Override
+ public void initialize(RaftServer server, RaftGroupId id,
+ RaftStorage raftStorage) throws IOException {
+ getLifeCycle().startAndTransition(() -> {
+ super.initialize(server, id, raftStorage);
+ this.raftGroupId = id;
+ storage.init(raftStorage);
+ });
+ }
+
+ /**
+ * Pre-execute the update request into state machine.
+ */
+ @Override
+ public TransactionContext startTransaction(
+ RaftClientRequest raftClientRequest) {
+ return TransactionContext.newBuilder()
+ .setClientRequest(raftClientRequest)
+ .setStateMachine(this)
+ .setServerRole(RaftProtos.RaftPeerRole.LEADER)
+ .setLogData(raftClientRequest.getMessage().getContent())
+ .build();
+ }
+
+ /**
+ * Apply a committed log entry to state machine.
+ */
+ @Override
+ public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
+ CompletableFuture<Message> ratisFuture =
+ new CompletableFuture<>();
+ //TODO execute SCMRequest and process SCMResponse
+ return ratisFuture;
+ }
+
+ /**
+ * Query state machine.
+ */
+ @Override
+ public CompletableFuture<Message> query(Message request) {
+ //TODO make handler respond to the query request.
+ return CompletableFuture.completedFuture(request);
+ }
+
+ /**
+ * Pause state machine.
+ */
+ @Override
+ public void pause() {
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
+ }
+
+ /**
+ * Unpause state machine and update the lastAppliedIndex.
+ * Following after uploading new state to state machine.
+ */
+ public void unpause(long newLastAppliedSnaphsotIndex,
+ long newLastAppliedSnapShotTermIndex) {
+ getLifeCycle().startAndTransition(() -> {
- this.setLastAppliedTermIndex(TermIndex.newTermIndex(
++ this.setLastAppliedTermIndex(TermIndex.valueOf(
+ newLastAppliedSnapShotTermIndex, newLastAppliedSnaphsotIndex));
+ });
+ }
+
+ /**
+ * Take SCM snapshot and write index to file.
+ * @return actual index or 0 if error.
+ */
+ @Override
+ public long takeSnapshot() throws IOException {
+ LOG.info("Saving Ratis snapshot on the SCM.");
+ if (scm != null) {
+ // TODO: remove the whole file later
+ return 0;
+ }
+ return 0;
+ }
+
+ /**
+ * Get latest SCM snapshot.
+ */
+ @Override
+ public SnapshotInfo getLatestSnapshot() {
+ return snapshotInfo;
+ }
+
+ private synchronized void updateLastApplied() {
+ Long appliedTerm = null;
+ long appliedIndex = -1;
+ for(long i = getLastAppliedTermIndex().getIndex() + 1;; i++) {
+ final Long removed = applyTransactionMap.remove(i);
+ if (removed == null) {
+ break;
+ }
+ appliedTerm = removed;
+ appliedIndex = i;
+ }
+ if (appliedTerm != null) {
+ updateLastAppliedTermIndex(appliedTerm, appliedIndex);
+ }
+ }
+
+ /**
+ * Called to notify state machine about indexes which are processed
+ * internally by Raft Server, this currently happens when conf entries are
+ * processed in raft Server. This keep state machine to keep a track of index
+ * updates.
+ */
+ public void notifyIndexUpdate(long currentTerm, long index) {
+ applyTransactionMap.put(index, currentTerm);
+ updateLastApplied();
+ snapshotInfo.updateTerm(currentTerm);
+ }
+
+ /**
+ * Notifies the state machine that the raft peer is no longer leader.
+ */
+ @Override
+ public void notifyNotLeader(Collection<TransactionContext> pendingEntries) {
+ scmRatisServer.updateServerRole();
+ }
+
+ /**
+ * Transfer from log entry to string.
+ */
+ @Override
+ public String toStateMachineLogEntryString(
+ RaftProtos.StateMachineLogEntryProto proto) {
+ //TODO implement transfer from proto to SCMRequest body.
+ return null;
+ }
+
+ /**
+ * Update lastAppliedIndex term in snapshot info.
+ */
+ public void updateLastAppliedIndexWithSnaphsotIndex() {
- setLastAppliedTermIndex(TermIndex.newTermIndex(snapshotInfo.getTerm(),
++ setLastAppliedTermIndex(TermIndex.valueOf(snapshotInfo.getTerm(),
+ snapshotInfo.getIndex()));
+ LOG.info("LastAppliedIndex set from SnapShotInfo {}",
+ getLastAppliedTermIndex());
+ }
+
+ @VisibleForTesting
+ void addApplyTransactionTermIndex(long term, long index) {
+ applyTransactionMap.put(index, term);
+ }
+
+ public void stop() {
+ HadoopExecutors.shutdown(executorService, LOG, 5, TimeUnit.SECONDS);
+ HadoopExecutors.shutdown(installSnapshotExecutor, LOG, 5, TimeUnit.SECONDS);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org