You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ozone.apache.org by sh...@apache.org on 2021/03/04 11:19:57 UTC
[ozone] branch HDDS-2823 updated: HDDS-4807. Add install checkpoint
in SCMStateMachine. (#1936)
This is an automated email from the ASF dual-hosted git repository.
shashikant pushed a commit to branch HDDS-2823
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-2823 by this push:
new 4a312ba HDDS-4807. Add install checkpoint in SCMStateMachine. (#1936)
4a312ba is described below
commit 4a312ba1fbdbf9575aeaf19549a6eca92ab62ffa
Author: bshashikant <sh...@apache.org>
AuthorDate: Thu Mar 4 16:49:40 2021 +0530
HDDS-4807. Add install checkpoint in SCMStateMachine. (#1936)
---
.../hadoop/hdds/scm/ha/SCMHAConfiguration.java | 8 +
.../org/apache/hadoop/hdds/scm/ha/SCMHAUtils.java | 4 +-
.../java/org/apache/hadoop/ozone/OzoneConsts.java | 1 +
.../hadoop/hdds/scm/metadata/SCMMetadataStore.java | 0
.../hadoop/hdds/scm/metadata/package-info.java | 21 ++
.../certificate/authority/CertificateServer.java | 8 +
.../certificate/authority/CertificateStore.java | 8 +-
.../certificate/authority/DefaultCAServer.java | 6 +
.../java/org/apache/hadoop/hdds/utils/HAUtils.java | 203 +++++++++++++
.../x509/certificate/authority/MockCAStore.java | 6 +-
.../hadoop/hdds/scm/block/BlockManagerImpl.java | 2 +-
.../hadoop/hdds/scm/block/DeletedBlockLog.java | 7 +
.../hadoop/hdds/scm/block/DeletedBlockLogImpl.java | 6 +
.../hdds/scm/block/DeletedBlockLogImplV2.java | 12 +-
.../scm/block/DeletedBlockLogStateManager.java | 3 +
.../scm/block/DeletedBlockLogStateManagerImpl.java | 15 +-
.../hdds/scm/block/SCMBlockDeletingService.java | 5 +
.../hdds/scm/container/ContainerManagerImpl.java | 14 +
.../hdds/scm/container/ContainerManagerV2.java | 8 +
.../scm/container/ContainerStateManagerImpl.java | 19 +-
.../scm/container/ContainerStateManagerV2.java | 9 +
.../hdds/scm/container/ReplicationManager.java | 1 +
.../scm/container/states/ContainerStateMap.java | 5 -
.../hadoop/hdds/scm/ha/DBTransactionBuffer.java | 2 +
.../hadoop/hdds/scm/ha/InterSCMGrpcService.java | 23 +-
.../hdds/scm/ha/MockDBTransactionBuffer.java | 5 +
.../hadoop/hdds/scm/ha/MockSCMHAManager.java | 80 +++--
.../hdds/scm/ha/SCMDBCheckpointProvider.java | 2 +-
.../hadoop/hdds/scm/ha/SCMDBTransactionBuffer.java | 41 +--
.../apache/hadoop/hdds/scm/ha/SCMHAManager.java | 15 +
.../hadoop/hdds/scm/ha/SCMHAManagerImpl.java | 199 +++++++++++--
.../apache/hadoop/hdds/scm/ha/SCMRatisServer.java | 2 +
.../hadoop/hdds/scm/ha/SCMRatisServerImpl.java | 5 +
.../org/apache/hadoop/hdds/scm/ha/SCMService.java | 10 +
.../hadoop/hdds/scm/ha/SCMServiceManager.java | 20 ++
.../hadoop/hdds/scm/ha/SCMSnapshotProvider.java | 7 +-
.../apache/hadoop/hdds/scm/ha/SCMStateMachine.java | 58 ++++
.../hdds/scm/metadata/SCMMetadataStoreImpl.java | 21 ++
.../scm/pipeline/BackgroundPipelineCreatorV2.java | 1 +
.../hadoop/hdds/scm/pipeline/PipelineManager.java | 8 +
.../hdds/scm/pipeline/PipelineManagerV2Impl.java | 6 +
.../hdds/scm/pipeline/PipelineStateManager.java | 6 +
.../scm/pipeline/PipelineStateManagerV2Impl.java | 20 +-
.../hdds/scm/pipeline/SCMPipelineManager.java | 6 +
.../hadoop/hdds/scm/pipeline/StateManager.java | 5 +
.../hadoop/hdds/scm/server/SCMCertStore.java | 12 +-
.../hdds/scm/server/StorageContainerManager.java | 6 +-
.../hdds/scm/ha/TestReplicationAnnotation.java | 5 +
.../hadoop/hdds/scm/ha/TestSCMServiceManager.java | 16 +
.../hdds/scm/pipeline/MockPipelineManager.java | 7 +
.../dev-support/findbugsExcludeFile.xml | 8 +
.../hadoop/hdds/scm/TestSCMInstallSnapshot.java | 64 +++-
.../hadoop/ozone/MiniOzoneHAClusterImpl.java | 16 +
.../apache/hadoop/ozone/om/TestOzoneManagerHA.java | 2 +-
.../ozone/scm/TestSCMInstallSnapshotWithHA.java | 328 +++++++++++++++++++++
.../hadoop/ozone/om/OmMetadataManagerImpl.java | 2 +-
.../om/ratis/utils/OzoneManagerRatisUtils.java | 74 +----
57 files changed, 1269 insertions(+), 184 deletions(-)
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAConfiguration.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAConfiguration.java
index ddf008c..52f69cc 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAConfiguration.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAConfiguration.java
@@ -210,6 +210,14 @@ public class SCMHAConfiguration {
this.ratisStorageDir = dir;
}
+ public void setRatisSnapshotDir(String dir) {
+ this.ratisSnapshotDir = dir;
+ }
+
+ public void setRaftLogPurgeGap(int gap) {
+ this.raftLogPurgeGap = gap;
+ }
+
public InetSocketAddress getRatisBindAddress() {
return NetUtils.createSocketAddr(ratisBindHost, ratisBindPort);
}
diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAUtils.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAUtils.java
index 3845f1c..192b784 100644
--- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAUtils.java
+++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAUtils.java
@@ -36,7 +36,7 @@ import java.util.Collection;
import java.util.List;
import static org.apache.hadoop.hdds.HddsConfigKeys.OZONE_METADATA_DIRS;
-import static org.apache.hadoop.ozone.OzoneConsts.OM_RATIS_SNAPSHOT_DIR;
+import static org.apache.hadoop.ozone.OzoneConsts.SCM_RATIS_SNAPSHOT_DIR;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEFAULT_SERVICE_ID;
import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_SERVICE_IDS_KEY;
@@ -104,7 +104,7 @@ public final class SCMHAUtils {
OZONE_METADATA_DIRS);
File metaDirPath = ServerUtils.getOzoneMetaDirPath(conf);
snapshotDir =
- Paths.get(metaDirPath.getPath(), OM_RATIS_SNAPSHOT_DIR).toString();
+ Paths.get(metaDirPath.getPath(), SCM_RATIS_SNAPSHOT_DIR).toString();
}
return snapshotDir;
}
diff --git 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
index 235c631..33114fc 100644
--- 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
@@ -126,6 +126,7 @@ public final class OzoneConsts {
public static final String OM_DB_NAME = "om.db";
public static final String SCM_DB_NAME = "scm.db";
public static final String OM_DB_BACKUP_PREFIX = "om.db.backup.";
+ public static final String SCM_DB_BACKUP_PREFIX = "scm.db.backup.";
public static final String STORAGE_DIR_CHUNKS = "chunks";
public static final String OZONE_DB_CHECKPOINT_REQUEST_FLUSH =
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStore.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStore.java
similarity index 100%
rename from hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStore.java
rename to hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStore.java
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/metadata/package-info.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/metadata/package-info.java
new file mode 100644
index 0000000..3dcf08b
--- /dev/null
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/scm/metadata/package-info.java
@@ -0,0 +1,21 @@
+/*
+ * 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.
+ */
+/**
+ * Metadata specific package utility for SCM.
+ */
+package org.apache.hadoop.hdds.scm.metadata;
\ No newline at end of file
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateServer.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateServer.java
index 1f2093f..4fa60d7 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateServer.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateServer.java
@@ -20,6 +20,7 @@
package org.apache.hadoop.hdds.security.x509.certificate.authority;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
import org.apache.hadoop.hdds.security.x509.SecurityConfig;
import org.apache.hadoop.hdds.security.x509.certificate.authority.CertificateApprover.ApprovalType;
@@ -131,6 +132,13 @@ public interface CertificateServer {
long startSerialId, int count, boolean isRevoked) throws IOException;
/**
+ * Reinitialise the certificate server withe the SCMMetastore during SCM
+ * state reload post install db checkpoint.
+ * @param scmMetadataStore
+ */
+ void reinitialize(SCMMetadataStore scmMetadataStore);
+
+ /**
* Make it explicit what type of CertificateServer we are creating here.
*/
enum CAType {
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateStore.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateStore.java
index e9ce137..5803ce7 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateStore.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/CertificateStore.java
@@ -22,6 +22,7 @@ package org.apache.hadoop.hdds.security.x509.certificate.authority;
import org.bouncycastle.asn1.x509.CRLReason;
import org.bouncycastle.cert.X509CertificateHolder;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import java.io.IOException;
import java.math.BigInteger;
@@ -105,11 +106,16 @@ public interface CertificateStore {
throws IOException;
/**
+ * Reinitialize the certificate server.
+ * @param metadataStore SCMMetaStore.
+ */
+ void reinitialize(SCMMetadataStore metadataStore);
+
+ /**
* Different kind of Certificate stores.
*/
enum CertType {
VALID_CERTS,
REVOKED_CERTS
}
-
}
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/DefaultCAServer.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/DefaultCAServer.java
index 86dffde..440bd4c 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/DefaultCAServer.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/authority/DefaultCAServer.java
@@ -24,6 +24,7 @@ import com.google.common.base.Preconditions;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.validator.routines.DomainValidator;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
import org.apache.hadoop.hdds.security.x509.SecurityConfig;
import org.apache.hadoop.hdds.security.x509.certificate.authority.PKIProfiles.DefaultProfile;
@@ -312,6 +313,11 @@ public class DefaultCAServer implements CertificateServer {
CertificateStore.CertType.VALID_CERTS);
}
+ @Override
+ public void reinitialize(SCMMetadataStore scmMetadataStore) {
+ store.reinitialize(scmMetadataStore);
+ }
+
/**
* Generates a Self Signed CertificateServer. These are the steps in
* generating a Self-Signed CertificateServer.
diff --git a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
index 00a8246..4632b36 100644
--- a/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
+++ b/hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/utils/HAUtils.java
@@ -16,6 +16,7 @@
*/
package org.apache.hadoop.hdds.utils;
+import org.apache.hadoop.hdds.HddsConfigKeys;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.AddSCMRequest;
import org.apache.hadoop.hdds.scm.ScmInfo;
@@ -24,13 +25,33 @@ import org.apache.hadoop.hdds.scm.protocol.ScmBlockLocationProtocol;
import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolClientSideTranslatorPB;
import org.apache.hadoop.hdds.scm.proxy.SCMBlockLocationFailoverProxyProvider;
import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.hdds.utils.db.DBDefinition;
+import org.apache.hadoop.hdds.utils.db.DBColumnFamilyDefinition;
+import org.apache.hadoop.hdds.utils.db.DBStore;
+import org.apache.hadoop.hdds.utils.db.RocksDBConfiguration;
+import org.apache.hadoop.hdds.utils.db.Table;
+import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
+import org.apache.ratis.util.ExitUtils;
+import org.apache.ratis.util.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+
+import static org.apache.hadoop.hdds.server.ServerUtils.getOzoneMetaDirPath;
+import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER;
+import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
/**
* utility class used by SCM and OM for HA.
*/
public final class HAUtils {
+ public static final Logger LOG = LoggerFactory.getLogger(HAUtils.class);
private HAUtils() {
}
@@ -82,4 +103,186 @@ public final class HAUtils {
.createProxy(scmBlockLocationClient, ScmBlockLocationProtocol.class,
conf);
}
+
+ /**
+ * Replace the current DB with the new DB checkpoint.
+ *
+ * @param lastAppliedIndex the last applied index in the current SCM DB.
+ * @param checkpointPath path to the new DB checkpoint
+ * @return location of backup of the original DB
+ * @throws Exception
+ */
+ public static File replaceDBWithCheckpoint(long lastAppliedIndex,
+ File oldDB, Path checkpointPath, String dbPrefix) throws IOException {
+
+ // Take a backup of the current DB
+ String dbBackupName =
+ dbPrefix + lastAppliedIndex + "_" + System
+ .currentTimeMillis();
+ File dbDir = oldDB.getParentFile();
+ File dbBackup = new File(dbDir, dbBackupName);
+
+ try {
+ Files.move(oldDB.toPath(), dbBackup.toPath());
+ } catch (IOException e) {
+ LOG.error("Failed to create a backup of the current DB. Aborting "
+ + "snapshot installation.");
+ throw e;
+ }
+
+ // Move the new DB checkpoint into the metadata dir
+ Path markerFile = new File(dbDir, DB_TRANSIENT_MARKER).toPath();
+ try {
+ // Create a Transient Marker file. This file will be deleted if the
+ // checkpoint DB is successfully moved to the old DB location or if the
+ // old DB backup is reset to its location. If not, then the DB is in
+ // an inconsistent state and this marker file will fail it from
+ // starting up.
+ Files.createFile(markerFile);
+ Files.move(checkpointPath, oldDB.toPath());
+ Files.deleteIfExists(markerFile);
+ } catch (IOException e) {
+ LOG.error("Failed to move downloaded DB checkpoint {} to metadata "
+ + "directory {}. Resetting to original DB.", checkpointPath,
+ oldDB.toPath());
+ try {
+ Files.move(dbBackup.toPath(), oldDB.toPath());
+ Files.deleteIfExists(markerFile);
+ } catch (IOException ex) {
+ String errorMsg = "Failed to reset to original DB. SCM is in an "
+ + "inconsistent state.";
+ ExitUtils.terminate(1, errorMsg, ex, LOG);
+ }
+ throw e;
+ }
+ return dbBackup;
+ }
+
+ /**
+ * Obtain SCMTransactionInfo from Checkpoint.
+ */
+ public static TransactionInfo getTrxnInfoFromCheckpoint(
+ OzoneConfiguration conf, Path dbPath, DBDefinition definition)
+ throws Exception {
+
+ if (dbPath != null) {
+ Path dbDir = dbPath.getParent();
+ Path dbFile = dbPath.getFileName();
+ if (dbDir != null && dbFile != null) {
+ return getTransactionInfoFromDB(conf, dbDir, dbFile.toString(),
+ definition);
+ }
+ }
+
+ throw new IOException("Checkpoint " + dbPath + " does not have proper " +
+ "DB location");
+ }
+
+ /**
+ * Obtain Transaction info from DB.
+ * @param tempConfig
+ * @param dbDir path to DB
+ * @return TransactionInfo
+ * @throws Exception
+ */
+ private static TransactionInfo getTransactionInfoFromDB(
+ OzoneConfiguration tempConfig, Path dbDir, String dbName,
+ DBDefinition definition)
+ throws Exception {
+ DBStore dbStore = loadDB(tempConfig, dbDir.toFile(),
+ dbName, definition);
+
+ // Get the table name with TransactionInfo as the value. The transaction
+ // info table name are different in SCM and SCM.
+
+ // In case, a new table gets added where the value is TransactionInfo, this
+ // logic may not work.
+
+
+ Table<String, TransactionInfo> transactionInfoTable =
+ getTransactionInfoTable(dbStore, definition);
+
+ TransactionInfo transactionInfo =
+ transactionInfoTable.get(TRANSACTION_INFO_KEY);
+ dbStore.close();
+
+ if (transactionInfo == null) {
+ throw new IOException("Failed to read TransactionInfo from DB " +
+ definition.getName() + " at " + dbDir);
+ }
+ return transactionInfo;
+ }
+
+ public static Table<String, TransactionInfo> getTransactionInfoTable(
+ DBStore dbStore, DBDefinition definition) throws IOException {
+ return Arrays.stream(definition.getColumnFamilies())
+ .filter(t -> t.getValueType() == TransactionInfo.class).findFirst()
+ .get().getTable(dbStore);
+ }
+
+ /**
+ * Verify transaction info with provided lastAppliedIndex.
+ *
+ * If transaction info transaction Index is less than or equal to
+ * lastAppliedIndex, return false, else return true.
+ * @param transactionInfo
+ * @param lastAppliedIndex
+ * @param leaderId
+ * @param newDBlocation
+ * @return boolean
+ */
+ public static boolean verifyTransactionInfo(TransactionInfo transactionInfo,
+ long lastAppliedIndex, String leaderId, Path newDBlocation,
+ Logger logger) {
+ if (transactionInfo.getTransactionIndex() <= lastAppliedIndex) {
+ logger.error("Failed to install checkpoint from SCM leader: {}"
+ + ". The last applied index: {} is greater than or equal to the "
+ + "checkpoint's applied index: {}. Deleting the downloaded "
+ + "checkpoint {}", leaderId, lastAppliedIndex,
+ transactionInfo.getTransactionIndex(), newDBlocation);
+ try {
+ FileUtils.deleteFully(newDBlocation);
+ } catch (IOException e) {
+ logger.error("Failed to fully delete the downloaded DB "
+ + "checkpoint {} from SCM leader {}.", newDBlocation, leaderId, e);
+ }
+ return false;
+ }
+ return true;
+ }
+
+ public static DBStore loadDB(OzoneConfiguration configuration, File metaDir,
+ String dbName, DBDefinition definition) throws IOException {
+ RocksDBConfiguration rocksDBConfiguration =
+ configuration.getObject(RocksDBConfiguration.class);
+ DBStoreBuilder dbStoreBuilder =
+ DBStoreBuilder.newBuilder(configuration, rocksDBConfiguration)
+ .setName(dbName).setPath(Paths.get(metaDir.getPath()));
+ // Add column family names and codecs.
+ for (DBColumnFamilyDefinition columnFamily : definition
+ .getColumnFamilies()) {
+
+ dbStoreBuilder.addTable(columnFamily.getName());
+ dbStoreBuilder
+ .addCodec(columnFamily.getKeyType(), columnFamily.getKeyCodec());
+ dbStoreBuilder
+ .addCodec(columnFamily.getValueType(), columnFamily.getValueCodec());
+ }
+ return dbStoreBuilder.build();
+ }
+
+ public static File getMetaDir(DBDefinition definition,
+ OzoneConfiguration configuration) {
+ // Set metadata dirs.
+ File metadataDir = definition.getDBLocation(configuration);
+
+ if (metadataDir == null) {
+ LOG.warn("{} is not configured. We recommend adding this setting. "
+ + "Falling back to {} instead.",
+ definition.getLocationConfigKey(), HddsConfigKeys.
+ OZONE_METADATA_DIRS);
+ metadataDir = getOzoneMetaDirPath(configuration);
+ }
+ return metadataDir;
+ }
}
diff --git a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/MockCAStore.java b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/MockCAStore.java
index cc38584..366b285 100644
--- a/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/MockCAStore.java
+++ b/hadoop-hdds/framework/src/test/java/org/apache/hadoop/hdds/security/x509/certificate/authority/MockCAStore.java
@@ -22,8 +22,7 @@ package org.apache.hadoop.hdds.security.x509.certificate.authority;
import org.bouncycastle.asn1.x509.CRLReason;
import org.bouncycastle.cert.X509CertificateHolder;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeType;
-
-
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import java.io.IOException;
import java.math.BigInteger;
import java.security.cert.X509Certificate;
@@ -72,4 +71,7 @@ public class MockCAStore implements CertificateStore {
throws IOException {
return Collections.emptyList();
}
+
+ @Override
+ public void reinitialize(SCMMetadataStore metadataStore) {}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
index 8f2dd3f..9160800 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/BlockManagerImpl.java
@@ -73,7 +73,7 @@ public class BlockManagerImpl implements BlockManager, BlockmanagerMXBean {
private final long containerSize;
- private final DeletedBlockLog deletedBlockLog;
+ private DeletedBlockLog deletedBlockLog;
private final SCMBlockDeletingService blockDeletingService;
private ObjectName mxBean;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java
index ddcd2d1..4b5ec0a 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLog.java
@@ -22,6 +22,7 @@ import org.apache.hadoop.hdds.protocol.proto
.DeleteBlockTransactionResult;
import org.apache.hadoop.hdds.protocol.proto
.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
+import org.apache.hadoop.hdds.utils.db.Table;
import java.io.Closeable;
import java.io.IOException;
@@ -104,4 +105,10 @@ public interface DeletedBlockLog extends Closeable {
* @throws IOException
*/
int getNumOfValidTransactions() throws IOException;
+
+ /**
+ * Reinitialize the delete log from the db.
+ * @param deletedBlocksTXTable delete transaction table
+ */
+ void reinitialize(Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImpl.java
index aff0be2..28666d9 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImpl.java
@@ -342,6 +342,12 @@ public class DeletedBlockLogImpl
}
}
+ @Override
+ public void reinitialize(
+ Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable) {
+ throw new RuntimeException("Not supported operation.");
+ }
+
/**
* {@inheritDoc}
*
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImplV2.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImplV2.java
index fee81e7..3e38c9b 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImplV2.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogImplV2.java
@@ -85,7 +85,7 @@ public class DeletedBlockLogImplV2
// DeletedBlockLogStateManager.
private final DeletedBlockLogStateManager deletedBlockLogStateManager;
private final SCMContext scmContext;
- private final Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable;
+ private Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable;
public DeletedBlockLogImplV2(ConfigurationSource conf,
ContainerManagerV2 containerManager,
@@ -298,6 +298,16 @@ public class DeletedBlockLogImplV2
}
}
+ @Override
+ public void reinitialize(
+ Table<Long, DeletedBlocksTransaction> deletedTable) {
+ // we don't need handle transactionToDNsCommitMap and
+ // deletedBlockLogStateManager, since they will be cleared
+ // when becoming leader.
+ this.deletedBlocksTXTable = deletedTable;
+ deletedBlockLogStateManager.reinitialize(deletedBlocksTXTable);
+ }
+
/**
* Called in SCMStateMachine#notifyLeaderChanged when current SCM becomes
* leader.
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManager.java
index f152a11..d90e176 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManager.java
@@ -19,6 +19,7 @@ package org.apache.hadoop.hdds.scm.block;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
import org.apache.hadoop.hdds.scm.metadata.Replicate;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.hdds.utils.db.TableIterator;
import org.apache.hadoop.hdds.utils.db.Table.KeyValue;
@@ -42,4 +43,6 @@ public interface DeletedBlockLogStateManager {
KeyValue<Long, DeletedBlocksTransaction>> getReadOnlyIterator();
void onFlush();
+
+ void reinitialize(Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable);
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManagerImpl.java
index 28a7098..a91a685 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/DeletedBlockLogStateManagerImpl.java
@@ -46,7 +46,7 @@ public class DeletedBlockLogStateManagerImpl
public static final Logger LOG =
LoggerFactory.getLogger(DeletedBlockLogStateManagerImpl.class);
- private final Table<Long, DeletedBlocksTransaction> deletedTable;
+ private Table<Long, DeletedBlocksTransaction> deletedTable;
private final DBTransactionBuffer transactionBuffer;
private final int maxRetry;
private final Set<Long> deletingTxIDs;
@@ -206,6 +206,19 @@ public class DeletedBlockLogStateManagerImpl
skippingRetryTxIDs.clear();
}
+ @Override
+ public void reinitialize(
+ Table<Long, DeletedBlocksTransaction> deletedBlocksTXTable) {
+ // Before Reinitialization, flush will be called from Ratis StateMachine.
+ // Just the DeletedDb will be loaded here.
+
+ // We don't need to handle transactionBuffer, deletingTxIDs
+ // and skippingRetryTxIDs here, since onFlush() will be called
+ // before reinitialization. Just update deletedTable here.
+ Preconditions.checkArgument(deletingTxIDs.isEmpty());
+ this.deletedTable = deletedBlocksTXTable;
+ }
+
public static Builder newBuilder() {
return new Builder();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMBlockDeletingService.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMBlockDeletingService.java
index 52aa4c3..edf6ca8 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMBlockDeletingService.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMBlockDeletingService.java
@@ -248,4 +248,9 @@ public class SCMBlockDeletingService extends BackgroundService
public String getServiceName() {
return SCMBlockDeletingService.class.getSimpleName();
}
+
+ @Override
+ public void stop() {
+ throw new RuntimeException("Not supported operation.");
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerImpl.java
index 04e59c2..7109185 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerImpl.java
@@ -122,6 +122,20 @@ public class ContainerManagerImpl implements ContainerManagerV2 {
}
@Override
+ public void reinitialize(Table<ContainerID, ContainerInfo> containerStore)
+ throws IOException {
+ lock.lock();
+ try {
+ containerStateManager.reinitialize(containerStore);
+ } catch (IOException ioe) {
+ LOG.error("Failed to reinitialize containerManager", ioe);
+ throw ioe;
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ @Override
public ContainerInfo getContainer(final ContainerID id)
throws ContainerNotFoundException {
return Optional.ofNullable(containerStateManager
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerV2.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerV2.java
index 54e3b4d..6f0233d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerV2.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerManagerV2.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleEvent;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionException;
/**
@@ -40,6 +41,13 @@ import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionExcepti
public interface ContainerManagerV2 extends Closeable {
// TODO: Rename this to ContainerManager
+ /**
+ * Reinitialize the containerManager with the updated container store.
+ * @param containerStore Container Table
+ * @throws IOException
+ */
+ void reinitialize(Table<ContainerID, ContainerInfo> containerStore)
+ throws IOException;
/**
* Returns the ContainerInfo from the container ID.
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerImpl.java
index c26afef..dedfe8c 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerImpl.java
@@ -95,7 +95,7 @@ public final class ContainerStateManagerImpl
/**
* In-memory representation of Container States.
*/
- private final ContainerStateMap containers;
+ private ContainerStateMap containers;
/**
* Persistent store for Container States.
@@ -118,7 +118,7 @@ public final class ContainerStateManagerImpl
* We use the containers in round-robin fashion for operations like block
* allocation. This map is used for remembering the last used container.
*/
- private final ConcurrentHashMap<ContainerState, ContainerID> lastUsedMap;
+ private ConcurrentHashMap<ContainerState, ContainerID> lastUsedMap;
private final Map<LifeCycleEvent, CheckedConsumer<ContainerInfo, IOException>>
containerStateChangeActions;
@@ -499,6 +499,21 @@ public final class ContainerStateManagerImpl
}
@Override
+ public void reinitialize(
+ Table<ContainerID, ContainerInfo> store) throws IOException {
+ lock.writeLock().lock();
+ try {
+ close();
+ this.containerStore = store;
+ this.containers = new ContainerStateMap();
+ this.lastUsedMap = new ConcurrentHashMap<>();
+ initialize();
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ @Override
public void close() throws IOException {
try {
containerStore.close();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerV2.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerV2.java
index fd5079d..276e21c 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerV2.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ContainerStateManagerV2.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ContainerInfoProto;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.LifeCycleState;
import org.apache.hadoop.hdds.scm.metadata.Replicate;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.hadoop.ozone.common.statemachine.InvalidStateTransitionException;
/**
@@ -174,6 +175,14 @@ public interface ContainerStateManagerV2 {
throws IOException;
/**
+ * Reinitialize the ContainerStateManager with container store.
+ * @param containerStore container table.
+ * @throws IOException
+ */
+ void reinitialize(Table<ContainerID, ContainerInfo> containerStore)
+ throws IOException;
+
+ /**
*
*/
void close() throws IOException;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
index 5a57378..2d757d7 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/ReplicationManager.java
@@ -215,6 +215,7 @@ public class ReplicationManager implements MetricsSource, SCMService {
/**
* Starts Replication Monitor thread.
*/
+ @Override
public synchronized void start() {
if (!isRunning()) {
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
index a76c851..ff2ca66 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/states/ContainerStateMap.java
@@ -120,11 +120,6 @@ public class ContainerStateMap {
factorMap.insert(info.getReplicationFactor(), id);
typeMap.insert(info.getReplicationType(), id);
replicaMap.put(id, ConcurrentHashMap.newKeySet());
-
- // Flush the cache of this container type, will be added later when
- // get container queries are executed.
- flushCache(info);
- LOG.trace("Container {} added to ContainerStateMap.", id);
}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/DBTransactionBuffer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/DBTransactionBuffer.java
index ca76c59..8ec70a0 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/DBTransactionBuffer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/DBTransactionBuffer.java
@@ -41,4 +41,6 @@ public interface DBTransactionBuffer extends Closeable {
void setLatestSnapshot(SnapshotInfo latestSnapshot);
void flush() throws IOException;
+
+ void init() throws IOException;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/InterSCMGrpcService.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/InterSCMGrpcService.java
index 726f566..61d132e 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/InterSCMGrpcService.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/InterSCMGrpcService.java
@@ -18,17 +18,23 @@
package org.apache.hadoop.hdds.scm.ha;
import java.io.IOException;
+import java.util.Arrays;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.protocol.scm.proto.InterSCMProtocolProtos.CopyDBCheckpointRequestProto;
import org.apache.hadoop.hdds.protocol.scm.proto.InterSCMProtocolProtos.CopyDBCheckpointResponseProto;
import org.apache.hadoop.hdds.protocol.scm.proto.InterSCMProtocolServiceGrpc;
+import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
+
/**
* Service to handle Rocks db Checkpointing.
*/
@@ -40,12 +46,13 @@ public class InterSCMGrpcService extends
private static final int BUFFER_SIZE = 1024 * 1024;
- private final String clusterId;
private final SCMDBCheckpointProvider provider;
+ private final StorageContainerManager scm;
+
public InterSCMGrpcService(final StorageContainerManager scm) {
Preconditions.checkNotNull(scm);
- this.clusterId = scm.getClusterId();
+ this.scm = scm;
provider =
new SCMDBCheckpointProvider(scm.getScmMetadataStore().getStore());
}
@@ -54,8 +61,18 @@ public class InterSCMGrpcService extends
public void download(CopyDBCheckpointRequestProto request,
StreamObserver<CopyDBCheckpointResponseProto> responseObserver) {
try {
+ scm.getScmHAManager().getDBTransactionBuffer().flush();
+ Table<String, TransactionInfo> transactionInfoTable =
+ Arrays.stream(new SCMDBDefinition().getColumnFamilies())
+ .filter(t -> t.getValueType() == TransactionInfo.class)
+ .findFirst().get().getTable(scm.getScmMetadataStore().getStore());
+
+ TransactionInfo transactionInfo =
+ transactionInfoTable.get(TRANSACTION_INFO_KEY);
+ Preconditions.checkNotNull(transactionInfo);
SCMGrpcOutputStream outputStream =
- new SCMGrpcOutputStream(responseObserver, clusterId, BUFFER_SIZE);
+ new SCMGrpcOutputStream(responseObserver, scm.getClusterId(),
+ BUFFER_SIZE);
provider.writeDBCheckPointToSream(outputStream, request.getFlush());
} catch (IOException e) {
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockDBTransactionBuffer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockDBTransactionBuffer.java
index 48b1492..e5a67e2 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockDBTransactionBuffer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockDBTransactionBuffer.java
@@ -77,6 +77,11 @@ public class MockDBTransactionBuffer implements DBTransactionBuffer {
}
@Override
+ public void init() throws IOException {
+
+ }
+
+ @Override
public void close() throws IOException {
flush();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockSCMHAManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockSCMHAManager.java
index 76d8c14..4926e71 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockSCMHAManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/MockSCMHAManager.java
@@ -36,6 +36,7 @@ import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.protocol.exceptions.StateMachineException;
+import org.apache.ratis.server.protocol.TermIndex;
// TODO: Move this class to test package after fixing Recon
/**
@@ -116,6 +117,11 @@ public final class MockSCMHAManager implements SCMHAManager {
return false;
}
+ @Override
+ public TermIndex installSnapshotFromLeader(String leaderId) {
+ return null;
+ }
+
private class MockRatisServer implements SCMRatisServer {
private Map<RequestType, Object> handlers =
@@ -127,7 +133,7 @@ public final class MockSCMHAManager implements SCMHAManager {
@Override
public void registerStateMachineHandler(final RequestType handlerType,
- final Object handler) {
+ final Object handler) {
handlers.put(handlerType, handler);
}
@@ -140,59 +146,42 @@ public final class MockSCMHAManager implements SCMHAManager {
if (isLeader()) {
try {
final Message result = process(request);
- reply = RaftClientReply.newBuilder()
- .setClientId(ClientId.randomId())
- .setServerId(raftId)
- .setGroupId(RaftGroupId.emptyGroupId())
- .setCallId(1L)
- .setSuccess(true)
- .setMessage(result)
- .setException(null)
- .setLogIndex(1L)
- .build();
+ reply = RaftClientReply.newBuilder().setClientId(ClientId.randomId())
+ .setServerId(raftId).setGroupId(RaftGroupId.emptyGroupId())
+ .setCallId(1L).setSuccess(true).setMessage(result)
+ .setException(null).setLogIndex(1L).build();
} catch (Exception ex) {
- reply = RaftClientReply.newBuilder()
- .setClientId(ClientId.randomId())
- .setServerId(raftId)
- .setGroupId(RaftGroupId.emptyGroupId())
- .setCallId(1L)
- .setSuccess(false)
- .setMessage(Message.EMPTY)
+ reply = RaftClientReply.newBuilder().setClientId(ClientId.randomId())
+ .setServerId(raftId).setGroupId(RaftGroupId.emptyGroupId())
+ .setCallId(1L).setSuccess(false).setMessage(Message.EMPTY)
.setException(new StateMachineException(raftId, ex))
- .setLogIndex(1L)
- .build();
+ .setLogIndex(1L).build();
}
} else {
- reply = RaftClientReply.newBuilder()
- .setClientId(ClientId.randomId())
- .setServerId(raftId)
- .setGroupId(RaftGroupId.emptyGroupId())
- .setCallId(1L)
- .setSuccess(false)
- .setMessage(Message.EMPTY)
- .setException(triggerNotLeaderException())
- .setLogIndex(1L)
- .build();
+ reply = RaftClientReply.newBuilder().setClientId(ClientId.randomId())
+ .setServerId(raftId).setGroupId(RaftGroupId.emptyGroupId())
+ .setCallId(1L).setSuccess(false).setMessage(Message.EMPTY)
+ .setException(triggerNotLeaderException()).setLogIndex(1L).build();
}
return SCMRatisResponse.decode(reply);
}
- private Message process(final SCMRatisRequest request)
- throws Exception {
+ private Message process(final SCMRatisRequest request) throws Exception {
try {
final Object handler = handlers.get(request.getType());
if (handler == null) {
- throw new IOException("No handler found for request type " +
- request.getType());
+ throw new IOException(
+ "No handler found for request type " + request.getType());
}
final List<Class<?>> argumentTypes = new ArrayList<>();
- for(Object args : request.getArguments()) {
+ for (Object args : request.getArguments()) {
argumentTypes.add(args.getClass());
}
- final Object result = handler.getClass().getMethod(
- request.getOperation(), argumentTypes.toArray(new Class<?>[0]))
+ final Object result = handler.getClass()
+ .getMethod(request.getOperation(),
+ argumentTypes.toArray(new Class<?>[0]))
.invoke(handler, request.getArguments());
return SCMRatisResponse.encode(result);
@@ -215,17 +204,20 @@ public final class MockSCMHAManager implements SCMHAManager {
@Override
public List<String> getRatisRoles() {
- return Arrays.asList(
- "180.3.14.5:9865",
- "180.3.14.21:9865",
- "180.3.14.145:9865");
+ return Arrays
+ .asList("180.3.14.5:9865", "180.3.14.21:9865", "180.3.14.145:9865");
}
@Override
public NotLeaderException triggerNotLeaderException() {
- return new NotLeaderException(RaftGroupMemberId.valueOf(
- RaftPeerId.valueOf("peer"), RaftGroupId.randomId()),
- null, new ArrayList<>());
+ return new NotLeaderException(RaftGroupMemberId
+ .valueOf(RaftPeerId.valueOf("peer"), RaftGroupId.randomId()), null,
+ new ArrayList<>());
+ }
+
+ @Override
+ public SCMStateMachine getSCMStateMachine() {
+ return null;
}
@Override
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBCheckpointProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBCheckpointProvider.java
index d588bb9..07e276f 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBCheckpointProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBCheckpointProvider.java
@@ -43,7 +43,7 @@ public class SCMDBCheckpointProvider {
public void writeDBCheckPointToSream(OutputStream stream, boolean flush)
throws IOException {
- LOG.info("Received request to obtain OM DB checkpoint snapshot");
+ LOG.info("Received request to obtain SCM DB checkpoint snapshot");
if (scmDbStore == null) {
LOG.error("Unable to process checkpointing request. DB Store is null");
return;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBTransactionBuffer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBTransactionBuffer.java
index 35afa5d..748b5e7 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBTransactionBuffer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMDBTransactionBuffer.java
@@ -38,7 +38,7 @@ import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
*/
public class SCMDBTransactionBuffer implements DBTransactionBuffer {
private final StorageContainerManager scm;
- private final SCMMetadataStore metadataStore;
+ private SCMMetadataStore metadataStore;
private BatchOperation currentBatchOperation;
private TransactionInfo latestTrxInfo;
private SnapshotInfo latestSnapshot;
@@ -46,22 +46,7 @@ public class SCMDBTransactionBuffer implements DBTransactionBuffer {
public SCMDBTransactionBuffer(StorageContainerManager scm)
throws IOException {
this.scm = scm;
- this.metadataStore = scm.getScmMetadataStore();
-
- // initialize a batch operation during construction time
- currentBatchOperation = this.metadataStore.getStore().initBatchOperation();
- latestTrxInfo = this.metadataStore.getTransactionInfoTable()
- .get(TRANSACTION_INFO_KEY);
- if (latestTrxInfo == null) {
- // transaction table is empty
- latestTrxInfo =
- TransactionInfo
- .builder()
- .setTransactionIndex(-1)
- .setCurrentTerm(0)
- .build();
- }
- latestSnapshot = latestTrxInfo.toSnapshotInfo();
+ init();
}
@Override
@@ -116,12 +101,32 @@ public class SCMDBTransactionBuffer implements DBTransactionBuffer {
}
@Override
+ public void init() throws IOException {
+ metadataStore = scm.getScmMetadataStore();
+
+ // initialize a batch operation during construction time
+ currentBatchOperation = this.metadataStore.getStore().initBatchOperation();
+ latestTrxInfo = this.metadataStore.getTransactionInfoTable()
+ .get(TRANSACTION_INFO_KEY);
+ if (latestTrxInfo == null) {
+ // transaction table is empty
+ latestTrxInfo =
+ TransactionInfo
+ .builder()
+ .setTransactionIndex(-1)
+ .setCurrentTerm(0)
+ .build();
+ }
+ latestSnapshot = latestTrxInfo.toSnapshotInfo();
+ }
+
+ @Override
public String toString() {
return latestTrxInfo.toString();
}
@Override
public void close() throws IOException {
-
+ flush();
}
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManager.java
index 937f0e3..4745eb1 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMHAManager.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hdds.scm.ha;
import org.apache.hadoop.hdds.scm.AddSCMRequest;
+import org.apache.ratis.server.protocol.TermIndex;
import java.io.IOException;
@@ -51,5 +52,19 @@ public interface SCMHAManager {
*/
void shutdown() throws IOException;
+ /**
+ * Adds the SC M instance to the SCM HA group.
+ * @param request AddSCM request
+ * @return status signying whether the AddSCM request succeeded or not.
+ * @throws IOException
+ */
boolean addSCM(AddSCMRequest request) throws IOException;
+
+ /**
+ * Download the SCM DB checkpoint from leader and reload the SCM state from
+ * it.
+ * @param leaderId leader id.
+ * @return
+ */
+ TermIndex installSnapshotFromLeader(String leaderId);
}
diff --git 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
index b4773ff..c46560d 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
@@ -21,14 +21,25 @@ import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.conf.ConfigurationSource;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.scm.AddSCMRequest;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.utils.HAUtils;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.ozone.OzoneSecurityUtil;
+import org.apache.hadoop.ozone.util.ExitManager;
import org.apache.ratis.server.protocol.TermIndex;
+import org.apache.ratis.util.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.io.File;
import java.io.IOException;
+import java.nio.file.Path;
/**
* SCMHAManagerImpl uses Apache Ratis for HA implementation. We will have 2N+1
@@ -48,6 +59,7 @@ public class SCMHAManagerImpl implements SCMHAManager {
private final SCMDBTransactionBuffer transactionBuffer;
private final SCMSnapshotProvider scmSnapshotProvider;
private final StorageContainerManager scm;
+ private ExitManager exitManager;
// this should ideally be started only in a ratis leader
private final InterSCMGrpcProtocolService grpcServer;
@@ -109,50 +121,37 @@ public class SCMHAManagerImpl implements SCMHAManager {
}
/**
- * Download and install latest checkpoint from leader OM.
+ * Download and install latest checkpoint from leader SCM.
*
- * @param leaderId peerNodeID of the leader OM
+ * @param leaderId peerNodeID of the leader SCM
* @return If checkpoint is installed successfully, return the
* corresponding termIndex. Otherwise, return null.
*/
public TermIndex installSnapshotFromLeader(String leaderId) {
- if(scmSnapshotProvider == null) {
- LOG.error("OM Snapshot Provider is not configured as there are no peer " +
- "nodes.");
+ if (scmSnapshotProvider == null) {
+ LOG.error("SCM Snapshot Provider is not configured as there are no peer "
+ + "nodes.");
return null;
}
- DBCheckpoint omDBCheckpoint = getDBCheckpointFromLeader(leaderId);
+ DBCheckpoint dBCheckpoint = getDBCheckpointFromLeader(leaderId);
LOG.info("Downloaded checkpoint from Leader {} to the location {}",
- leaderId, omDBCheckpoint.getCheckpointLocation());
+ leaderId, dBCheckpoint.getCheckpointLocation());
TermIndex termIndex = null;
try {
- termIndex = installCheckpoint(leaderId, omDBCheckpoint);
+ termIndex = installCheckpoint(leaderId, dBCheckpoint);
} catch (Exception ex) {
- LOG.error("Failed to install snapshot from Leader OM.", ex);
+ LOG.error("Failed to install snapshot from Leader SCM.", ex);
}
return termIndex;
}
/**
- * Install checkpoint. If the checkpoints snapshot index is greater than
- * SCM's last applied transaction index, then re-initialize the OM
- * state via this checkpoint. Before re-initializing OM state, the OM Ratis
- * server should be stopped so that no new transactions can be applied.
- */
- TermIndex installCheckpoint(String leaderId, DBCheckpoint omDBCheckpoint)
- throws Exception {
- // TODO : implement install checkpoint
- return null;
- }
-
-
- /**
- * Download the latest SCM DB checkpoint from the leader OM.
+ * Download the latest SCM DB checkpoint from the leader SCM.
*
- * @param leaderId OMNodeID of the leader OM node.
- * @return latest DB checkpoint from leader OM.
+ * @param leaderId SCMNodeID of the leader SCM node.
+ * @return latest DB checkpoint from leader SCM.
*/
private DBCheckpoint getDBCheckpointFromLeader(String leaderId) {
LOG.info("Downloading checkpoint from leader SCM {} and reloading state " +
@@ -161,11 +160,127 @@ public class SCMHAManagerImpl implements SCMHAManager {
try {
return scmSnapshotProvider.getSCMDBSnapshot(leaderId);
} catch (IOException e) {
- LOG.error("Failed to download checkpoint from OM leader {}", leaderId, e);
+ LOG.error("Failed to download checkpoint from SCM leader {}", leaderId,
+ e);
}
return null;
}
+ /**
+ * Install checkpoint. If the checkpoints snapshot index is greater than
+ * SCM's last applied transaction index, then re-initialize the SCM
+ * state via this checkpoint. Before re-initializing SCM state, the SCM Ratis
+ * server should be stopped so that no new transactions can be applied.
+ */
+ @VisibleForTesting
+ public TermIndex installCheckpoint(String leaderId, DBCheckpoint dbCheckpoint)
+ throws Exception {
+
+ Path checkpointLocation = dbCheckpoint.getCheckpointLocation();
+ TransactionInfo checkpointTrxnInfo = HAUtils
+ .getTrxnInfoFromCheckpoint(OzoneConfiguration.of(conf),
+ checkpointLocation, new SCMDBDefinition());
+
+ LOG.info("Installing checkpoint with SCMTransactionInfo {}",
+ checkpointTrxnInfo);
+
+ return installCheckpoint(leaderId, checkpointLocation, checkpointTrxnInfo);
+ }
+
+ public TermIndex installCheckpoint(String leaderId, Path checkpointLocation,
+ TransactionInfo checkpointTrxnInfo) throws Exception {
+
+ File dbBackup = null;
+ TermIndex termIndex =
+ getRatisServer().getSCMStateMachine().getLastAppliedTermIndex();
+ long term = termIndex.getTerm();
+ long lastAppliedIndex = termIndex.getIndex();
+ // Check if current applied log index is smaller than the downloaded
+ // checkpoint transaction index. If yes, proceed by stopping the ratis
+ // server so that the SCM state can be re-initialized. If no then do not
+ // proceed with installSnapshot.
+ boolean canProceed = HAUtils
+ .verifyTransactionInfo(checkpointTrxnInfo, lastAppliedIndex, leaderId,
+ checkpointLocation, LOG);
+ File oldDBLocation = scm.getScmMetadataStore().getStore().getDbLocation();
+ if (canProceed) {
+ try {
+ // Stop services
+ stopServices();
+
+ // Pause the State Machine so that no new transactions can be applied.
+ // This action also clears the SCM Double Buffer so that if there
+ // are any pending transactions in the buffer, they are discarded.
+
+ } catch (Exception e) {
+ LOG.error("Failed to stop/ pause the services. Cannot proceed with "
+ + "installing the new checkpoint.");
+ startServices();
+ throw e;
+ }
+ try {
+ dbBackup = HAUtils
+ .replaceDBWithCheckpoint(lastAppliedIndex, oldDBLocation,
+ checkpointLocation, OzoneConsts.SCM_DB_BACKUP_PREFIX);
+ term = checkpointTrxnInfo.getTerm();
+ lastAppliedIndex = checkpointTrxnInfo.getTransactionIndex();
+ LOG.info(
+ "Replaced DB with checkpoint from SCM: {}, term: {}, index: {}",
+ leaderId, term, lastAppliedIndex);
+ } catch (Exception e) {
+ LOG.error("Failed to install Snapshot from {} as SCM failed to replace"
+ + " DB with downloaded checkpoint. Reloading old SCM state.", e);
+ }
+ } else {
+ LOG.warn("Cannot proceed with InstallSnapshot as SCM is at TermIndex {} "
+ + "and checkpoint has lower TermIndex {}. Reloading old "
+ + "state of SCM.", termIndex, checkpointTrxnInfo.getTermIndex());
+ }
+
+ // Reload the DB store with the new checkpoint.
+ // Restart (unpause) the state machine and update its last applied index
+ // to the installed checkpoint's snapshot index.
+ try {
+ reloadSCMState();
+ getRatisServer().getSCMStateMachine().unpause(lastAppliedIndex, term);
+ LOG.info("Reloaded SCM state with Term: {} and Index: {}", term,
+ lastAppliedIndex);
+ } catch (Exception ex) {
+ String errorMsg = "Failed to reload SCM state and instantiate services.";
+ exitManager.exitSystem(1, errorMsg, ex, LOG);
+ }
+
+ // Delete the backup DB
+ try {
+ if (dbBackup != null) {
+ FileUtils.deleteFully(dbBackup);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to delete the backup of the original DB {}", dbBackup);
+ }
+
+ if (lastAppliedIndex != checkpointTrxnInfo.getTransactionIndex()) {
+ // Install Snapshot failed and old state was reloaded. Return null to
+ // Ratis to indicate that installation failed.
+ return null;
+ }
+
+ // TODO: We should only return the snpashotIndex to the leader.
+ // Should be fixed after RATIS-586
+ TermIndex newTermIndex = TermIndex.valueOf(term, lastAppliedIndex);
+ return newTermIndex;
+ }
+
+
+ /**
+ * Re-instantiate MetadataManager with new DB checkpoint.
+ * All the classes which use/ store MetadataManager should also be updated
+ * with the new MetadataManager instance.
+ */
+ void reloadSCMState()
+ throws IOException {
+ startServices();
+ }
/**
* {@inheritDoc}
@@ -173,6 +288,7 @@ public class SCMHAManagerImpl implements SCMHAManager {
@Override
public void shutdown() throws IOException {
ratisServer.stop();
+ ratisServer.getSCMStateMachine().stop();
grpcServer.stop();
}
@@ -189,4 +305,35 @@ public class SCMHAManagerImpl implements SCMHAManager {
getRatisServer().getDivision().getGroup().getGroupId());
return getRatisServer().addSCM(request);
}
+
+ void stopServices() throws Exception {
+
+ // just stop the SCMMetaData store. All other background
+ // services will be in pausing state in the follower.
+ scm.getScmMetadataStore().stop();
+ }
+
+ void startServices() throws IOException {
+
+ // TODO: Fix the metrics ??
+ final SCMMetadataStore metadataStore = scm.getScmMetadataStore();
+ metadataStore.start(OzoneConfiguration.of(conf));
+ scm.getPipelineManager().reinitialize(metadataStore.getPipelineTable());
+ scm.getContainerManager().reinitialize(metadataStore.getContainerTable());
+ scm.getScmBlockManager().getDeletedBlockLog().reinitialize(
+ metadataStore.getDeletedBlocksTXTable());
+ if (OzoneSecurityUtil.isSecurityEnabled(conf)) {
+ scm.getCertificateServer().reinitialize(metadataStore);
+ }
+ }
+
+ @VisibleForTesting
+ public void setExitManagerForTesting(ExitManager exitManagerForTesting) {
+ this.exitManager = exitManagerForTesting;
+ }
+
+ @VisibleForTesting
+ public static Logger getLogger() {
+ return LOG;
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServer.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServer.java
index c5a4bae..b351c86 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServer.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServer.java
@@ -53,4 +53,6 @@ public interface SCMRatisServer {
NotLeaderException triggerNotLeaderException();
boolean addSCM(AddSCMRequest request) throws IOException;
+
+ SCMStateMachine getSCMStateMachine();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java
index 4157046..0d61bca 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMRatisServerImpl.java
@@ -152,6 +152,11 @@ public class SCMRatisServerImpl implements SCMRatisServer {
}
@Override
+ public SCMStateMachine getSCMStateMachine() {
+ return stateMachine;
+ }
+
+ @Override
public void registerStateMachineHandler(final RequestType handlerType,
final Object handler) {
stateMachine.registerHandler(handlerType, handler);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMService.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMService.java
index 32194a6..4d7c435 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMService.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMService.java
@@ -62,4 +62,14 @@ public interface SCMService {
NEW_NODE_HANDLER_TRIGGERED,
UNHEALTHY_TO_HEALTHY_NODE_HANDLER_TRIGGERED
}
+
+ /**
+ * starts the SCM service.
+ */
+ void start();
+
+ /**
+ * stops the SCM service.
+ */
+ void stop();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
index 52e3d26..14ece30 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMServiceManager.java
@@ -64,4 +64,24 @@ public final class SCMServiceManager {
service.notifyEventTriggered(event);
}
}
+
+ /**
+ * Starts all running services.
+ */
+ public synchronized void start() {
+ for (SCMService service : services) {
+ LOG.debug("Stopping service:{}.", service.getServiceName());
+ service.start();
+ }
+ }
+
+ /**
+ * Stops all running services.
+ */
+ public synchronized void stop() {
+ for (SCMService service : services) {
+ LOG.debug("Stopping service:{}.", service.getServiceName());
+ service.stop();
+ }
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMSnapshotProvider.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMSnapshotProvider.java
index c0e1db3..093b810 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMSnapshotProvider.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMSnapshotProvider.java
@@ -101,7 +101,7 @@ public class SCMSnapshotProvider {
String snapshotFilePath =
Paths.get(scmSnapshotDir.getAbsolutePath(), snapshotFileName).toFile()
.getAbsolutePath();
- File targetFile = new File(snapshotFileName + ".tar.gz");
+ File targetFile = new File(snapshotFilePath + ".tar.gz");
// the client instance will be initialized only when first install snapshot
// notification from ratis leader will be received.
@@ -123,8 +123,9 @@ public class SCMSnapshotProvider {
FileUtil.unTar(targetFile, untarredDbDir.toFile());
FileUtils.deleteQuietly(targetFile);
- LOG.info("Successfully downloaded latest checkpoint from leader SCM: {}",
- leaderSCMNodeID);
+ LOG.info(
+ "Successfully downloaded latest checkpoint from leader SCM: {} path {}",
+ leaderSCMNodeID, untarredDbDir.toAbsolutePath());
RocksDBCheckpoint scmCheckpoint = new RocksDBCheckpoint(untarredDbDir);
return scmCheckpoint;
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
index c856ae3..f65aeea 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/ha/SCMStateMachine.java
@@ -25,6 +25,8 @@ import java.util.List;
import java.util.Map;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import com.google.common.base.Preconditions;
@@ -33,6 +35,7 @@ import org.apache.hadoop.hdds.scm.block.DeletedBlockLog;
import org.apache.hadoop.hdds.scm.block.DeletedBlockLogImplV2;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
import org.apache.ratis.proto.RaftProtos;
import org.apache.hadoop.hdds.scm.exceptions.SCMException;
import org.apache.hadoop.util.Time;
@@ -46,6 +49,7 @@ import org.apache.ratis.statemachine.impl.BaseStateMachine;
import org.apache.hadoop.hdds.protocol.proto.SCMRatisProtocol.RequestType;
import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
+import org.apache.ratis.util.LifeCycle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -65,6 +69,7 @@ public class SCMStateMachine extends BaseStateMachine {
private final SimpleStateMachineStorage storage =
new SimpleStateMachineStorage();
private final AtomicBoolean isInitialized;
+ private ExecutorService installSnapshotExecutor;
public SCMStateMachine(final StorageContainerManager scm,
final SCMRatisServer ratisServer, DBTransactionBuffer buffer)
@@ -85,6 +90,7 @@ public class SCMStateMachine extends BaseStateMachine {
), SCM_NOT_INITIALIZED);
}
}
+ this.installSnapshotExecutor = HadoopExecutors.newSingleThreadExecutor();
isInitialized = new AtomicBoolean(true);
}
@@ -158,6 +164,29 @@ public class SCMStateMachine extends BaseStateMachine {
scm.getSCMServiceManager().notifyStatusChanged();
}
+ /**
+ * Leader SCM has purged entries from its log. To catch up, SCM must download
+ * the latest checkpoint from the leader SCM and install it.
+ * @param roleInfoProto the leader node information
+ * @param firstTermIndexInLog TermIndex of the first append entry available
+ * in the Leader's log.
+ * @return the last term index included in the installed snapshot.
+ */
+ @Override
+ public CompletableFuture<TermIndex> notifyInstallSnapshotFromLeader(
+ RaftProtos.RoleInfoProto roleInfoProto, TermIndex firstTermIndexInLog) {
+
+ String leaderNodeId = RaftPeerId.valueOf(roleInfoProto.getFollowerInfo()
+ .getLeaderInfo().getId().getId()).toString();
+ LOG.info("Received install snapshot notification from SCM leader: {} with "
+ + "term index: {}", leaderNodeId, firstTermIndexInLog);
+
+ CompletableFuture<TermIndex> future = CompletableFuture.supplyAsync(
+ () -> scm.getScmHAManager().installSnapshotFromLeader(leaderNodeId),
+ installSnapshotExecutor);
+ return future;
+ }
+
@Override
public void notifyLeaderChanged(RaftGroupMemberId groupMemberId,
RaftPeerId newLeaderId) {
@@ -237,4 +266,33 @@ public class SCMStateMachine extends BaseStateMachine {
public void notifyConfigurationChanged(long term, long index,
RaftProtos.RaftConfigurationProto newRaftConfiguration) {
}
+
+ @Override
+ public void pause() {
+ getLifeCycle().transition(LifeCycle.State.PAUSING);
+ getLifeCycle().transition(LifeCycle.State.PAUSED);
+ }
+
+ public void stop() {
+ HadoopExecutors.
+ shutdown(installSnapshotExecutor, LOG, 5, TimeUnit.SECONDS);
+ }
+ /**
+ * Unpause the StateMachine, re-initialize the DoubleBuffer and update the
+ * lastAppliedIndex. This should be done after uploading new state to the
+ * StateMachine.
+ */
+ public void unpause(long newLastAppliedSnaphsotIndex,
+ long newLastAppliedSnapShotTermIndex) {
+ getLifeCycle().startAndTransition(() -> {
+ try {
+ transactionBuffer.init();
+ this.setLastAppliedTermIndex(TermIndex
+ .valueOf(newLastAppliedSnapShotTermIndex,
+ newLastAppliedSnaphsotIndex));
+ } catch (IOException ioe) {
+ LOG.error("Failed to unpause ", ioe);
+ }
+ });
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStoreImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStoreImpl.java
index 903d2b0..54a190b 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStoreImpl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/metadata/SCMMetadataStoreImpl.java
@@ -17,6 +17,7 @@
*/
package org.apache.hadoop.hdds.scm.metadata;
+import java.io.File;
import java.io.IOException;
import java.math.BigInteger;
import java.security.cert.X509Certificate;
@@ -25,6 +26,7 @@ import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.DeletedBlocksTransaction;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.utils.HAUtils;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
@@ -45,7 +47,9 @@ import static org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition.REVOKED_CERTS;
import static org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition.TRANSACTIONINFO;
import static org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition.VALID_CERTS;
import static org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition.VALID_SCM_CERTS;
+import static org.apache.hadoop.ozone.OzoneConsts.DB_TRANSIENT_MARKER;
+import org.apache.ratis.util.ExitUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -95,6 +99,23 @@ public class SCMMetadataStoreImpl implements SCMMetadataStore {
throws IOException {
if (this.store == null) {
+ File metaDir = HAUtils.getMetaDir(new SCMDBDefinition(), configuration);
+ // Check if there is a DB Inconsistent Marker in the metaDir. This
+ // marker indicates that the DB is in an inconsistent state and hence
+ // the OM process should be terminated.
+ File markerFile = new File(metaDir, DB_TRANSIENT_MARKER);
+ if (markerFile.exists()) {
+ LOG.error("File {} marks that SCM DB is in an inconsistent state.",
+ markerFile);
+ // Note - The marker file should be deleted only after fixing the DB.
+ // In an HA setup, this can be done by replacing this DB with a
+ // checkpoint from another SCM.
+ String errorMsg = "Cannot load SCM DB as it is in an inconsistent " +
+ "state.";
+ ExitUtils.terminate(1, errorMsg, LOG);
+ }
+
+
this.store = DBStoreBuilder.createDBStore(config, new SCMDBDefinition());
deletedBlocksTable =
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreatorV2.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreatorV2.java
index 343444d..41d3aa8 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreatorV2.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/BackgroundPipelineCreatorV2.java
@@ -114,6 +114,7 @@ public class BackgroundPipelineCreatorV2 implements SCMService {
/**
* Start RatisPipelineUtilsThread.
*/
+ @Override
public void start() {
if (!running.compareAndSet(false, true)) {
LOG.warn("{} is already started, just ignore.", THREAD_NAME);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
index 2078460..04985d4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java
@@ -28,6 +28,7 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
/**
@@ -122,4 +123,11 @@ public interface PipelineManager extends Closeable, PipelineManagerMXBean {
* @return boolean
*/
boolean getSafeModeStatus();
+
+ /**
+ * Reinitialize the pipelineManager with the lastest pipeline store
+ * during SCM reload.
+ */
+ void reinitialize(Table<PipelineID, Pipeline> pipelineStore)
+ throws IOException;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
index 16bce43..a8ec6e4 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerV2Impl.java
@@ -502,6 +502,12 @@ public final class PipelineManagerV2Impl implements PipelineManager {
}
@Override
+ public void reinitialize(Table<PipelineID, Pipeline> pipelineStore)
+ throws IOException {
+ stateManager.reinitialize(pipelineStore);
+ }
+
+ @Override
public void close() throws IOException {
if (backgroundPipelineCreator != null) {
backgroundPipelineCreator.stop();
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
index 899d877..581477e 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManager.java
@@ -24,6 +24,7 @@ import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
+import org.apache.hadoop.hdds.utils.db.Table;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -183,6 +184,11 @@ public class PipelineStateManager implements StateManager {
}
@Override
+ public void reinitialize(Table<PipelineID, Pipeline> pipelineStore)
+ throws IOException {
+ }
+
+ @Override
public void updatePipelineState(PipelineID id, PipelineState newState)
throws PipelineNotFoundException {
pipelineStateMap.updatePipelineState(id, newState);
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManagerV2Impl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManagerV2Impl.java
index 01f11f5..2130c4d 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManagerV2Impl.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineStateManagerV2Impl.java
@@ -50,7 +50,7 @@ public class PipelineStateManagerV2Impl implements StateManager {
private static final Logger LOG =
LoggerFactory.getLogger(PipelineStateManager.class);
- private final PipelineStateMap pipelineStateMap;
+ private PipelineStateMap pipelineStateMap;
private final NodeManager nodeManager;
private Table<PipelineID, Pipeline> pipelineStore;
private final DBTransactionBuffer transactionBuffer;
@@ -328,8 +328,24 @@ public class PipelineStateManagerV2Impl implements StateManager {
throw new IOException("Not supported.");
}
- // legacy interfaces end
+ @Override
+ public void reinitialize(Table<PipelineID, Pipeline> store)
+ throws IOException {
+ lock.writeLock().lock();
+ try {
+ pipelineStore.close();
+ this.pipelineStateMap = new PipelineStateMap();
+ this.pipelineStore = store;
+ initialize();
+ } catch (Exception ex) {
+ LOG.error("PipelineManager reinitialization close failed", ex);
+ throw new IOException(ex);
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+ // legacy interfaces end
public static Builder newBuilder() {
return new Builder();
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
index faa7bda..b22feab 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/SCMPipelineManager.java
@@ -759,6 +759,12 @@ public class SCMPipelineManager implements
return this.isInSafeMode.get();
}
+ @Override
+ public void reinitialize(Table<PipelineID, Pipeline> store)
+ throws IOException {
+ throw new RuntimeException("Not supported operation.");
+ }
+
public Table<PipelineID, Pipeline> getPipelineStore() {
return pipelineStore;
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/StateManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/StateManager.java
index 3a772e5..e8a3902 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/StateManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/StateManager.java
@@ -21,6 +21,7 @@ import org.apache.hadoop.hdds.protocol.DatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.metadata.Replicate;
+import org.apache.hadoop.hdds.utils.db.Table;
import java.io.IOException;
import java.util.Collection;
@@ -117,4 +118,8 @@ public interface StateManager {
void deactivatePipeline(PipelineID pipelineID)
throws IOException;
+
+ void reinitialize(Table<PipelineID, Pipeline> pipelineStore)
+ throws IOException;
+
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMCertStore.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMCertStore.java
index 4fd9866..8172148 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMCertStore.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/SCMCertStore.java
@@ -59,7 +59,7 @@ import static org.apache.hadoop.hdds.security.x509.certificate.authority.Certifi
public class SCMCertStore implements CertificateStore {
private static final Logger LOG =
LoggerFactory.getLogger(SCMCertStore.class);
- private final SCMMetadataStore scmMetadataStore;
+ private SCMMetadataStore scmMetadataStore;
private final Lock lock;
private AtomicLong crlSequenceId;
@@ -266,4 +266,14 @@ public class SCMCertStore implements CertificateStore {
}
}
}
+
+ /**
+ * Reinitialise the underlying store with SCMMetaStore
+ * during SCM StateMachine reload.
+ * @param metadataStore
+ */
+ @Override
+ public void reinitialize(SCMMetadataStore metadataStore) {
+ this.scmMetadataStore = metadataStore;
+ }
}
diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
index f45f372..404fd76 100644
--- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
+++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/server/StorageContainerManager.java
@@ -557,6 +557,10 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl
certificateServer);
}
+ public CertificateServer getCertificateServer() {
+ return certificateServer;
+ }
+
// For Internal gRPC client from SCM to DN with gRPC TLS
static GrpcTlsConfig createTlsClientConfigForSCM(SecurityConfig conf,
CertificateServer certificateServer) throws IOException {
@@ -1110,7 +1114,7 @@ public final class StorageContainerManager extends ServiceRuntimeInfoImpl
scmSafeModeManager.stop();
}
-
+
/**
* Wait until service has completed shutdown.
*/
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestReplicationAnnotation.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestReplicationAnnotation.java
index a7eb637..1b9e0b5 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestReplicationAnnotation.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestReplicationAnnotation.java
@@ -80,6 +80,11 @@ public class TestReplicationAnnotation {
throws IOException {
return false;
}
+
+ @Override
+ public SCMStateMachine getSCMStateMachine() {
+ return null;
+ }
};
scmhaInvocationHandler = new SCMHAInvocationHandler(
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMServiceManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMServiceManager.java
index 47b4537..786a893 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMServiceManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/ha/TestSCMServiceManager.java
@@ -56,6 +56,14 @@ public class TestSCMServiceManager {
public String getServiceName() {
return "serviceRunWhenLeader";
}
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void stop() {
+ }
};
SCMServiceManager serviceManager = new SCMServiceManager();
@@ -118,6 +126,14 @@ public class TestSCMServiceManager {
public String getServiceName() {
return "serviceRunWhenLeaderAndOutOfSafeMode";
}
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void stop() {
+ }
};
SCMServiceManager serviceManager = new SCMServiceManager();
diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
index c29ae43..d03aa55 100644
--- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
+++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java
@@ -22,6 +22,7 @@ import org.apache.hadoop.hdds.protocol.MockDatanodeDetails;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
import org.apache.hadoop.hdds.scm.container.ContainerID;
+import org.apache.hadoop.hdds.utils.db.Table;
import java.io.IOException;
import java.util.Collection;
@@ -219,6 +220,12 @@ public final class MockPipelineManager implements PipelineManager {
}
@Override
+ public void reinitialize(Table<PipelineID, Pipeline> pipelineStore)
+ throws IOException {
+
+ }
+
+ @Override
public void close() throws IOException {
}
diff --git a/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml b/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
index 0bf7ea4..b089712 100644
--- a/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-ozone/integration-test/dev-support/findbugsExcludeFile.xml
@@ -128,4 +128,12 @@
<Class name="org.apache.hadoop.ozone.client.rpc.TestDeleteWithSlowFollower"/>
<Bug pattern="ST_WRITE_TO_STATIC_FROM_INSTANCE_METHOD" />
</Match>
+ <Match>
+ <Class name="org.apache.hadoop.hdds.scm.TestSCMInstallSnapshot"/>
+ <Bug pattern="NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE" />
+ </Match>
+ <Match>
+ <Class name="org.apache.hadoop.ozone.scm.TestSCMInstallSnapshotWithHA"/>
+ <Bug pattern="NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE" />
+ </Match>
</FindBugsFilter>
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMInstallSnapshot.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMInstallSnapshot.java
index 3768d95..ecf0e13 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMInstallSnapshot.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMInstallSnapshot.java
@@ -19,15 +19,24 @@ package org.apache.hadoop.hdds.scm;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.scm.container.ContainerID;
import org.apache.hadoop.hdds.scm.container.ContainerManagerV2;
import org.apache.hadoop.hdds.scm.ha.SCMHAConfiguration;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManagerImpl;
import org.apache.hadoop.hdds.scm.ha.SCMNodeDetails;
import org.apache.hadoop.hdds.scm.ha.SCMSnapshotProvider;
+import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
import org.apache.hadoop.hdds.scm.pipeline.PipelineManager;
import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
+import org.apache.hadoop.hdds.utils.db.DBStore;
import org.apache.hadoop.ozone.MiniOzoneCluster;
import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.test.GenericTestUtils;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
@@ -55,6 +64,8 @@ public class TestSCMInstallSnapshot {
SCMHAConfiguration scmhaConfiguration = conf.getObject(
SCMHAConfiguration.class);
scmhaConfiguration.setRatisSnapshotThreshold(1L);
+ scmhaConfiguration.setRatisSnapshotDir(
+ GenericTestUtils.getRandomizedTempPath() + "/snapshot");
conf.setFromObject(scmhaConfiguration);
cluster = MiniOzoneCluster
.newBuilder(conf)
@@ -71,7 +82,11 @@ public class TestSCMInstallSnapshot {
}
@Test
- public void testInstallSnapshot() throws Exception {
+ public void testDownloadSnapshot() throws Exception {
+ downloadSnapshot();
+ }
+
+ private DBCheckpoint downloadSnapshot() throws Exception {
StorageContainerManager scm = cluster.getStorageContainerManager();
ContainerManagerV2 containerManager = scm.getContainerManager();
PipelineManager pipelineManager = scm.getPipelineManager();
@@ -91,10 +106,53 @@ public class TestSCMInstallSnapshot {
SCMSnapshotProvider provider =
scm.getScmHAManager().getSCMSnapshotProvider();
provider.setPeerNodesMap(peerMap);
- provider.getSCMDBSnapshot(scmNodeDetails.getNodeId());
+ DBCheckpoint checkpoint =
+ provider.getSCMDBSnapshot(scmNodeDetails.getNodeId());
final File[] files = FileUtil.listFiles(provider.getScmSnapshotDir());
- Assert.assertEquals(1, files.length);
Assert.assertTrue(files[0].getName().startsWith(
OzoneConsts.SCM_DB_NAME + "-" + scmNodeDetails.getNodeId()));
+ return checkpoint;
+ }
+
+ @Test
+ public void testInstallCheckPoint() throws Exception {
+ DBCheckpoint checkpoint = downloadSnapshot();
+ StorageContainerManager scm = cluster.getStorageContainerManager();
+ SCMHAManagerImpl scmhaManager = (SCMHAManagerImpl)scm.getScmHAManager();
+ DBStore db = HAUtils
+ .loadDB(conf, checkpoint.getCheckpointLocation().getParent().toFile(),
+ checkpoint.getCheckpointLocation().getFileName().toString(),
+ new SCMDBDefinition());
+ // Hack the transaction index in the checkpoint so as to ensure the
+ // checkpointed transaction index is higher than when it was downloaded
+ // from.
+ Assert.assertNotNull(db);
+ HAUtils.getTransactionInfoTable(db, new SCMDBDefinition())
+ .put(OzoneConsts.TRANSACTION_INFO_KEY, TransactionInfo.builder()
+ .setCurrentTerm(1).setTransactionIndex(100).build());
+ db.close();
+ ContainerID cid =
+ scm.getContainerManager().getContainers().get(0).containerID();
+ PipelineID pipelineID =
+ scm.getPipelineManager().getPipelines().get(0).getId();
+ scm.getScmMetadataStore().getPipelineTable().delete(pipelineID);
+ scm.getContainerManager().deleteContainer(cid);
+ Assert.assertNull(
+ scm.getScmMetadataStore().getPipelineTable().get(pipelineID));
+ Assert.assertFalse(scm.getContainerManager().containerExist(cid));
+ scmhaManager.installCheckpoint("scm1", checkpoint);
+
+ Assert.assertNotNull(
+ scm.getScmMetadataStore().getPipelineTable().get(pipelineID));
+ Assert.assertNotNull(
+ scm.getScmMetadataStore().getContainerTable().get(cid));
+ Assert.assertTrue(scm.getPipelineManager().containsPipeline(pipelineID));
+ Assert.assertTrue(scm.getContainerManager().containerExist(cid));
+ Assert.assertEquals(100, scm.getScmMetadataStore().
+ getTransactionInfoTable().get(OzoneConsts.TRANSACTION_INFO_KEY)
+ .getTransactionIndex());
+ Assert.assertEquals(100,
+ scm.getScmHAManager().getDBTransactionBuffer().getLatestTrxInfo()
+ .getTermIndex().getIndex());
}
}
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
index 5cf9d9b..cb07a4c 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/MiniOzoneHAClusterImpl.java
@@ -144,6 +144,14 @@ public class MiniOzoneHAClusterImpl extends MiniOzoneClusterImpl {
return omhaService.isServiceActive(omNodeId);
}
+ public boolean isSCMActive(String scmNodeId) {
+ return scmhaService.isServiceActive(scmNodeId);
+ }
+
+ public StorageContainerManager getSCM(String scmNodeId) {
+ return this.scmhaService.getServiceById(scmNodeId);
+ }
+
public OzoneManager getOzoneManager(int index) {
return this.omhaService.getServiceByIndex(index);
}
@@ -183,6 +191,14 @@ public class MiniOzoneHAClusterImpl extends MiniOzoneClusterImpl {
omhaService.startInactiveService(omNodeID, OzoneManager::start);
}
+ /**
+ * Start a previously inactive SCM.
+ */
+ public void startInactiveSCM(String scmNodeId) throws IOException {
+ scmhaService
+ .startInactiveService(scmNodeId, StorageContainerManager::start);
+ }
+
@Override
public void restartOzoneManager() throws IOException {
for (OzoneManager ozoneManager : this.omhaService.getServices()) {
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
index 8cf0f5e..f38b55b 100644
--- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManagerHA.java
@@ -189,7 +189,7 @@ public abstract class TestOzoneManagerHA {
* Create a key in the bucket.
* @return the key name.
*/
- static String createKey(OzoneBucket ozoneBucket) throws IOException {
+ public static String createKey(OzoneBucket ozoneBucket) throws IOException {
String keyName = "key" + RandomStringUtils.randomNumeric(5);
String data = "data" + RandomStringUtils.randomNumeric(5);
OzoneOutputStream ozoneOutputStream = ozoneBucket.createKey(keyName,
diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMInstallSnapshotWithHA.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMInstallSnapshotWithHA.java
new file mode 100644
index 0000000..2ffddd7
--- /dev/null
+++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/scm/TestSCMInstallSnapshotWithHA.java
@@ -0,0 +1,328 @@
+/**
+ * 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.scm;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
+import org.apache.hadoop.hdds.scm.container.ContainerInfo;
+import org.apache.hadoop.hdds.scm.ha.SCMHAConfiguration;
+import org.apache.hadoop.hdds.scm.ha.SCMHAManagerImpl;
+import org.apache.hadoop.hdds.scm.ha.SCMStateMachine;
+import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
+import org.apache.hadoop.hdds.scm.metadata.SCMMetadataStore;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.HAUtils;
+import org.apache.hadoop.hdds.utils.db.DBCheckpoint;
+import org.apache.hadoop.ozone.MiniOzoneCluster;
+import org.apache.hadoop.ozone.MiniOzoneHAClusterImpl;
+import org.apache.hadoop.ozone.OzoneConsts;
+import org.apache.hadoop.hdds.utils.TransactionInfo;
+import org.apache.hadoop.ozone.util.ExitManager;
+import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.ratis.server.protocol.TermIndex;
+
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Assert;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.api.Disabled;
+import org.slf4j.Logger;
+import org.slf4j.event.Level;
+
+/**
+ * Tests the Ratis snapshot feature in SCM.
+ */
+@Timeout(500)
+public class TestSCMInstallSnapshotWithHA {
+
+ private MiniOzoneHAClusterImpl cluster = null;
+ private OzoneConfiguration conf;
+ private String clusterId;
+ private String scmId;
+ private String omServiceId;
+ private String scmServiceId;
+ private int numOfOMs = 1;
+ private int numOfSCMs = 3;
+
+ private static final long SNAPSHOT_THRESHOLD = 5;
+ // private static final int LOG_PURGE_GAP = 5;
+
+ /**
+ * Create a MiniOzoneCluster for testing.
+ *
+ * @throws IOException
+ */
+ @BeforeEach
+ public void init() throws Exception {
+ conf = new OzoneConfiguration();
+ clusterId = UUID.randomUUID().toString();
+ scmId = UUID.randomUUID().toString();
+ omServiceId = "om-service-test1";
+ scmServiceId = "scm-service-test1";
+ SCMHAConfiguration scmhaConfiguration =
+ conf.getObject(SCMHAConfiguration.class);
+ // scmhaConfiguration.setRaftLogPurgeEnabled(true);
+ // scmhaConfiguration.setRaftLogPurgeGap(LOG_PURGE_GAP);
+ scmhaConfiguration.setRatisSnapshotThreshold(SNAPSHOT_THRESHOLD);
+ conf.setFromObject(scmhaConfiguration);
+
+ cluster = (MiniOzoneHAClusterImpl) MiniOzoneCluster.newHABuilder(conf)
+ .setClusterId(clusterId)
+ .setScmId(scmId)
+ .setOMServiceId(omServiceId)
+ .setSCMServiceId(scmServiceId)
+ .setNumOfOzoneManagers(numOfOMs)
+ .setNumOfStorageContainerManagers(numOfSCMs)
+ .setNumOfActiveSCMs(2)
+ .build();
+ cluster.waitForClusterToBeReady();
+ }
+
+ /**
+ * Shutdown MiniDFSCluster.
+ */
+ @AfterEach
+ public void shutdown() {
+ if (cluster != null) {
+ cluster.shutdown();
+ }
+ }
+
+ /**
+ * This test is disabled for now as there seems to be an issue with
+ * Ratis install Snapshot code. In ratis while a new node gets added,
+ * unless and until the node gets added to the voter list, the follower state
+ * is not updated with leader info. So, while an install snapshot notification
+ * is received in the leader, the leader info is not set and hence, out of
+ * ratis transfer using the same leader info doesn't work.
+ *
+ * TODO: Fix this
+ * */
+ @Test
+ @Disabled
+ public void testInstallSnapshot() throws Exception {
+ // Get the leader SCM
+ StorageContainerManager leaderSCM = getLeader(cluster);
+ String leaderNodeId = leaderSCM.getScmNodeDetails().getNodeId();
+ Assert.assertNotNull(leaderSCM);
+ // Find the inactive SCM
+ String followerId = getInactiveSCM(cluster).getScmId();
+
+ StorageContainerManager follower = cluster.getSCM(followerId);
+ // Do some transactions so that the log index increases
+ List<ContainerInfo> containers = writeToIncreaseLogIndex(leaderSCM, 200);
+
+ // Get the latest db checkpoint from the leader SCM.
+ TransactionInfo transactionInfo =
+ leaderSCM.getScmHAManager().getDBTransactionBuffer().getLatestTrxInfo();
+ TermIndex leaderTermIndex =
+ TermIndex.valueOf(transactionInfo.getTerm(),
+ transactionInfo.getTransactionIndex());
+ long leaderSnaphsotIndex = leaderTermIndex.getIndex();
+ long leaderSnapshotTermIndex = leaderTermIndex.getTerm();
+
+ DBCheckpoint leaderDbCheckpoint =
+ leaderSCM.getScmMetadataStore().getStore().getCheckpoint(false);
+
+ // Start the inactive
+ cluster.startInactiveSCM(followerId);
+
+ // The recently started should be lagging behind the leader .
+ long followerLastAppliedIndex =
+ follower.getScmHAManager().getRatisServer().getSCMStateMachine()
+ .getLastAppliedTermIndex().getIndex();
+ assertTrue(
+ followerLastAppliedIndex < leaderSnaphsotIndex);
+
+ SCMHAManagerImpl scmhaManager =
+ (SCMHAManagerImpl) (follower.getScmHAManager());
+ // Install leader 's db checkpoint on the lagging .
+ scmhaManager.installCheckpoint(leaderNodeId, leaderDbCheckpoint);
+
+ SCMStateMachine followerStateMachine =
+ follower.getScmHAManager().getRatisServer().getSCMStateMachine();
+ // After the new checkpoint is installed, the follower
+ // lastAppliedIndex must >= the snapshot index of the checkpoint. It
+ // could be great than snapshot index if there is any conf entry from ratis.
+ followerLastAppliedIndex = followerStateMachine
+ .getLastAppliedTermIndex().getIndex();
+ assertTrue(followerLastAppliedIndex >= leaderSnaphsotIndex);
+ assertTrue(followerStateMachine
+ .getLastAppliedTermIndex().getTerm() >= leaderSnapshotTermIndex);
+
+ // Verify that the follower 's DB contains the transactions which were
+ // made while it was inactive.
+ SCMMetadataStore followerMetaStore = follower.getScmMetadataStore();
+ for (ContainerInfo containerInfo : containers) {
+ Assert.assertNotNull(followerMetaStore.getContainerTable()
+ .get(containerInfo.containerID()));
+ }
+ }
+
+ @Test
+ public void testInstallOldCheckpointFailure() throws Exception {
+ // Get the leader SCM
+ StorageContainerManager leaderSCM = getLeader(cluster);
+ String leaderNodeId = leaderSCM.getScmNodeDetails().getNodeId();
+ String followerId = getInactiveSCM(cluster).getScmId();
+ // Find the inactive SCM
+
+ StorageContainerManager follower = cluster.getSCM(followerId);
+ cluster.startInactiveSCM(followerId);
+ follower.exitSafeMode();
+ DBCheckpoint leaderDbCheckpoint = leaderSCM.getScmMetadataStore().getStore()
+ .getCheckpoint(false);
+
+ SCMStateMachine leaderSM =
+ leaderSCM.getScmHAManager().getRatisServer().getSCMStateMachine();
+ TermIndex lastTermIndex = leaderSM.getLastAppliedTermIndex();
+
+ SCMStateMachine followerSM =
+ follower.getScmHAManager().getRatisServer().getSCMStateMachine();
+ follower.getScmMetadataStore().getTransactionInfoTable().
+ put(OzoneConsts.TRANSACTION_INFO_KEY, TransactionInfo.builder()
+ .setCurrentTerm(lastTermIndex.getTerm())
+ .setTransactionIndex(lastTermIndex.getIndex() + 100).build());
+ // Advance the follower
+ followerSM.notifyTermIndexUpdated(lastTermIndex.getTerm(),
+ lastTermIndex.getIndex() + 100);
+
+ GenericTestUtils.setLogLevel(SCMHAManagerImpl.getLogger(), Level.INFO);
+ GenericTestUtils.LogCapturer logCapture =
+ GenericTestUtils.LogCapturer.captureLogs(SCMHAManagerImpl.getLogger());
+
+ // Install the old checkpoint on the follower . This should fail as the
+ // follower is already ahead of that transactionLogIndex and the
+ // state should be reloaded.
+ TermIndex followerTermIndex = followerSM.getLastAppliedTermIndex();
+ SCMHAManagerImpl scmhaManager =
+ (SCMHAManagerImpl) (follower.getScmHAManager());
+ TermIndex newTermIndex =
+ scmhaManager.installCheckpoint(leaderNodeId, leaderDbCheckpoint);
+
+ String errorMsg = "Reloading old state of SCM";
+ Assert.assertTrue(logCapture.getOutput().contains(errorMsg));
+ Assert.assertNull(" installed checkpoint even though checkpoint " +
+ "logIndex is less than it's lastAppliedIndex", newTermIndex);
+ Assert.assertEquals(followerTermIndex,
+ followerSM.getLastAppliedTermIndex());
+ }
+
+ @Test
+ public void testInstallCorruptedCheckpointFailure() throws Exception {
+ StorageContainerManager leaderSCM = getLeader(cluster);
+ String leaderNodeId = leaderSCM.getScmId();
+ // Find the inactive SCM
+ String followerId = getInactiveSCM(cluster).getScmId();
+ StorageContainerManager follower = cluster.getSCM(followerId);
+ // Do some transactions so that the log index increases
+ writeToIncreaseLogIndex(leaderSCM, 100);
+
+ DBCheckpoint leaderDbCheckpoint = leaderSCM.getScmMetadataStore().getStore()
+ .getCheckpoint(false);
+ Path leaderCheckpointLocation = leaderDbCheckpoint.getCheckpointLocation();
+ TransactionInfo leaderCheckpointTrxnInfo = HAUtils
+ .getTrxnInfoFromCheckpoint(conf, leaderCheckpointLocation,
+ new SCMDBDefinition());
+
+ Assert.assertNotNull(leaderCheckpointLocation);
+ // Corrupt the leader checkpoint and install that on the follower. The
+ // operation should fail and should shutdown.
+ boolean delete = true;
+ for (File file : leaderCheckpointLocation.toFile()
+ .listFiles()) {
+ if (file.getName().contains(".sst")) {
+ if (delete) {
+ file.delete();
+ delete = false;
+ } else {
+ delete = true;
+ }
+ }
+ }
+
+ SCMHAManagerImpl scmhaManager =
+ (SCMHAManagerImpl) (follower.getScmHAManager());
+ GenericTestUtils.setLogLevel(SCMHAManagerImpl.getLogger(), Level.ERROR);
+ GenericTestUtils.LogCapturer logCapture =
+ GenericTestUtils.LogCapturer.captureLogs(SCMHAManagerImpl.getLogger());
+ scmhaManager.setExitManagerForTesting(new DummyExitManager());
+
+ scmhaManager.installCheckpoint(leaderNodeId, leaderCheckpointLocation,
+ leaderCheckpointTrxnInfo);
+
+ Assert.assertTrue(logCapture.getOutput()
+ .contains("Failed to reload SCM state and instantiate services."));
+ }
+
+ private List<ContainerInfo> writeToIncreaseLogIndex(
+ StorageContainerManager scm, long targetLogIndex)
+ throws IOException, InterruptedException {
+ List<ContainerInfo> containers = new ArrayList<>();
+ SCMStateMachine stateMachine =
+ scm.getScmHAManager().getRatisServer().getSCMStateMachine();
+ long logIndex = scm.getScmHAManager().getRatisServer().getSCMStateMachine()
+ .getLastAppliedTermIndex().getIndex();
+ while (logIndex < targetLogIndex) {
+ containers.add(scm.getContainerManager()
+ .allocateContainer(HddsProtos.ReplicationType.RATIS,
+ HddsProtos.ReplicationFactor.THREE,
+ TestSCMInstallSnapshotWithHA.class.getName()));
+ Thread.sleep(100);
+ logIndex = stateMachine.getLastAppliedTermIndex().getIndex();
+ }
+ return containers;
+ }
+
+ private static class DummyExitManager extends ExitManager {
+ @Override
+ public void exitSystem(int status, String message, Throwable throwable,
+ Logger log) {
+ log.error("System Exit: " + message, throwable);
+ }
+ }
+
+
+ static StorageContainerManager getLeader(MiniOzoneHAClusterImpl impl) {
+ for (StorageContainerManager scm : impl.getStorageContainerManagers()) {
+ if (scm.checkLeader()) {
+ return scm;
+ }
+ }
+ return null;
+ }
+
+ static StorageContainerManager getInactiveSCM(MiniOzoneHAClusterImpl impl) {
+ for (StorageContainerManager scm : impl.getStorageContainerManagers()) {
+ if (!impl.isSCMActive(scm.getScmId())) {
+ return scm;
+ }
+ }
+ return null;
+ }
+}
+
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
index 1f713db..d348270 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java
@@ -272,7 +272,7 @@ public class OmMetadataManagerImpl implements OMMetadataManager {
// Check if there is a DB Inconsistent Marker in the metaDir. This
// marker indicates that the DB is in an inconsistent state and hence
- // the OM process should be terminated.
+ // the SCM process should be terminated.
File markerFile = new File(metaDir, DB_TRANSIENT_MARKER);
if (markerFile.exists()) {
LOG.error("File {} marks that OM DB is in an inconsistent state.",
diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
index 52241b8..14657c6 100644
--- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
+++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java
@@ -19,10 +19,9 @@ package org.apache.hadoop.ozone.om.ratis.utils;
import com.google.common.base.Preconditions;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.utils.db.DBStore;
-import org.apache.hadoop.hdds.utils.db.Table;
-import org.apache.hadoop.ozone.om.OmMetadataManagerImpl;
+import org.apache.hadoop.hdds.utils.HAUtils;
import org.apache.hadoop.ozone.om.OzoneManager;
+import org.apache.hadoop.ozone.om.codec.OMDBDefinition;
import org.apache.hadoop.ozone.om.exceptions.OMException;
import org.apache.hadoop.hdds.utils.TransactionInfo;
import org.apache.hadoop.ozone.om.request.bucket.OMBucketCreateRequest;
@@ -69,14 +68,11 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMReque
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneObj.ObjectType;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
-import org.apache.ratis.util.FileUtils;
import org.rocksdb.RocksDBException;
import java.io.IOException;
import java.nio.file.Path;
-import static org.apache.hadoop.ozone.OzoneConsts.TRANSACTION_INFO_KEY;
-import static org.apache.hadoop.ozone.om.OmMetadataManagerImpl.TRANSACTION_INFO_TABLE;
/**
* Utility class used by OzoneManager HA.
@@ -235,45 +231,8 @@ public final class OzoneManagerRatisUtils {
*/
public static TransactionInfo getTrxnInfoFromCheckpoint(
OzoneConfiguration conf, Path dbPath) throws Exception {
-
- if (dbPath != null) {
- Path dbDir = dbPath.getParent();
- Path dbFile = dbPath.getFileName();
- if (dbDir != null && dbFile != null) {
- return getTransactionInfoFromDB(conf, dbDir, dbFile.toString());
- }
- }
-
- throw new IOException("Checkpoint " + dbPath + " does not have proper " +
- "DB location");
- }
-
- /**
- * Obtain Transaction info from DB.
- * @param tempConfig
- * @param dbDir path to DB
- * @return OMTransactionInfo
- * @throws Exception
- */
- private static TransactionInfo getTransactionInfoFromDB(
- OzoneConfiguration tempConfig, Path dbDir, String dbName)
- throws Exception {
- DBStore dbStore = OmMetadataManagerImpl.loadDB(tempConfig, dbDir.toFile(),
- dbName);
-
- Table<String, TransactionInfo> transactionInfoTable =
- dbStore.getTable(TRANSACTION_INFO_TABLE,
- String.class, TransactionInfo.class);
-
- TransactionInfo transactionInfo =
- transactionInfoTable.get(TRANSACTION_INFO_KEY);
- dbStore.close();
-
- if (transactionInfo == null) {
- throw new IOException("Failed to read OMTransactionInfo from DB " +
- dbName + " at " + dbDir);
- }
- return transactionInfo;
+ return HAUtils
+ .getTrxnInfoFromCheckpoint(conf, dbPath, new OMDBDefinition());
}
/**
@@ -287,25 +246,10 @@ public final class OzoneManagerRatisUtils {
* @param newDBlocation
* @return boolean
*/
- public static boolean verifyTransactionInfo(
- TransactionInfo transactionInfo,
- long lastAppliedIndex,
- String leaderId, Path newDBlocation) {
- if (transactionInfo.getTransactionIndex() <= lastAppliedIndex) {
- OzoneManager.LOG.error("Failed to install checkpoint from OM leader: {}" +
- ". The last applied index: {} is greater than or equal to the " +
- "checkpoint's applied index: {}. Deleting the downloaded " +
- "checkpoint {}", leaderId, lastAppliedIndex,
- transactionInfo.getTransactionIndex(), newDBlocation);
- try {
- FileUtils.deleteFully(newDBlocation);
- } catch (IOException e) {
- OzoneManager.LOG.error("Failed to fully delete the downloaded DB " +
- "checkpoint {} from OM leader {}.", newDBlocation, leaderId, e);
- }
- return false;
- }
-
- return true;
+ public static boolean verifyTransactionInfo(TransactionInfo transactionInfo,
+ long lastAppliedIndex, String leaderId, Path newDBlocation) {
+ return HAUtils
+ .verifyTransactionInfo(transactionInfo, lastAppliedIndex, leaderId,
+ newDBlocation, OzoneManager.LOG);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@ozone.apache.org
For additional commands, e-mail: commits-help@ozone.apache.org