sadanand48 commented on code in PR #10696:
URL: https://github.com/apache/ozone/pull/10696#discussion_r3577030864
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -402,115 +371,76 @@ public void stop() throws Exception {
reconTaskController.stop();
omMetadataManager.stop();
scheduler.shutdownNow();
- tarExtractor.stop();
+ reconSnapshotProvider.close();
metrics.unRegister();
reconSyncMetrics.unRegister();
connectionFactory.destroy();
}
+ private boolean isOmSpnegoEnabled() {
+ return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
+ .equals("kerberos");
+ }
+
/**
- * Find the OM leader's address to get the snapshot from.
+ * Return the current OM leader's {@link ServiceInfo} from the OM service
+ * list. The retained transfer checkpoint is local to the leader, so Recon
+ * always downloads from (and resumes against) the leader.
*/
- @VisibleForTesting
- public String getOzoneManagerSnapshotUrl() throws IOException {
- String omLeaderUrl = omDBSnapshotUrl;
- List<org.apache.hadoop.ozone.om.helpers.ServiceInfo> serviceList =
- ozoneManagerClient.getServiceList();
- HttpConfig.Policy policy = HttpConfig.getHttpPolicy(configuration);
- if (!serviceList.isEmpty()) {
- for (org.apache.hadoop.ozone.om.helpers.ServiceInfo info : serviceList) {
- if (info.getNodeType().equals(HddsProtos.NodeType.OM) &&
- info.getOmRoleInfo().hasServerRole() &&
- info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
- omLeaderUrl = (policy.isHttpsEnabled() ?
- "https://"; + info.getServiceAddress(Type.HTTPS) :
- "http://"; + info.getServiceAddress(Type.HTTP)) +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+ private ServiceInfo getLeaderServiceInfo() {
+ try {
+ List<ServiceInfo> serviceList = ozoneManagerClient.getServiceList();
+ for (ServiceInfo info : serviceList) {
+ if (info.getNodeType().equals(HddsProtos.NodeType.OM)
+ && info.getOmRoleInfo().hasServerRole()
+ && info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
+ return info;
}
}
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to fetch OM service list", e);
}
- return omLeaderUrl;
+ throw new IllegalStateException("No OM leader found in the OM service
list.");
}
- private boolean isOmSpnegoEnabled() {
- return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
- .equals("kerberos");
+ private String getLeaderNodeId() {
+ return getLeaderServiceInfo().getOmRoleInfo().getNodeId();
}
/**
- * Method to obtain current OM DB Snapshot.
- * @return DBCheckpoint instance.
+ * Obtain the current OM DB snapshot using the same OM-follower bootstrap
+ * mechanism (chunked, resumable {@code POST /v2/dbCheckpoint} with hard-link
+ * dedup on the leader). The returned {@link DBCheckpoint} points at
+ * a stable {@code om.snapshot.db_<ts>} directory ready to be promoted by
+ * {@link #updateReconOmDBWithNewSnapshot()}; returns {@code null} on
failure,
+ * leaving the current active DB untouched.
+ *
+ * @return DBCheckpoint instance, or {@code null} on failure.
*/
@VisibleForTesting
public DBCheckpoint getOzoneManagerDBSnapshot() {
- String snapshotFileName = RECON_OM_SNAPSHOT_DB + "_" +
System.currentTimeMillis();
- Path untarredDbDir = Paths.get(omSnapshotDBParentDir.getAbsolutePath(),
snapshotFileName);
-
- // Before fetching full snapshot again and create a new OM DB snapshot
directory, check and delete
- // any existing OM DB snapshot directories under recon om db dir location
and delete all such
- // om db snapshot dirs including the last known om db snapshot dir
returned by reconUtils.getLastKnownDB
- File lastKnownDB = reconUtils.getLastKnownDB(omSnapshotDBParentDir,
RECON_OM_SNAPSHOT_DB);
Review Comment:
We can still keep these 2 code blocks i.e deleting lastKnownDB and leftover
staging dirs as it can happen b/w switching v1 and v2 , there leftover
artifacts from v1. I think keeping this should be safe
##########
hadoop-ozone/recon/src/main/java/org/apache/hadoop/ozone/recon/spi/impl/OzoneManagerServiceProviderImpl.java:
##########
@@ -402,115 +371,76 @@ public void stop() throws Exception {
reconTaskController.stop();
omMetadataManager.stop();
scheduler.shutdownNow();
- tarExtractor.stop();
+ reconSnapshotProvider.close();
metrics.unRegister();
reconSyncMetrics.unRegister();
connectionFactory.destroy();
}
+ private boolean isOmSpnegoEnabled() {
+ return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
+ .equals("kerberos");
+ }
+
/**
- * Find the OM leader's address to get the snapshot from.
+ * Return the current OM leader's {@link ServiceInfo} from the OM service
+ * list. The retained transfer checkpoint is local to the leader, so Recon
+ * always downloads from (and resumes against) the leader.
*/
- @VisibleForTesting
- public String getOzoneManagerSnapshotUrl() throws IOException {
- String omLeaderUrl = omDBSnapshotUrl;
- List<org.apache.hadoop.ozone.om.helpers.ServiceInfo> serviceList =
- ozoneManagerClient.getServiceList();
- HttpConfig.Policy policy = HttpConfig.getHttpPolicy(configuration);
- if (!serviceList.isEmpty()) {
- for (org.apache.hadoop.ozone.om.helpers.ServiceInfo info : serviceList) {
- if (info.getNodeType().equals(HddsProtos.NodeType.OM) &&
- info.getOmRoleInfo().hasServerRole() &&
- info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
- omLeaderUrl = (policy.isHttpsEnabled() ?
- "https://"; + info.getServiceAddress(Type.HTTPS) :
- "http://"; + info.getServiceAddress(Type.HTTP)) +
- OZONE_DB_CHECKPOINT_HTTP_ENDPOINT;
+ private ServiceInfo getLeaderServiceInfo() {
+ try {
+ List<ServiceInfo> serviceList = ozoneManagerClient.getServiceList();
+ for (ServiceInfo info : serviceList) {
+ if (info.getNodeType().equals(HddsProtos.NodeType.OM)
+ && info.getOmRoleInfo().hasServerRole()
+ && info.getOmRoleInfo().getServerRole().equals(LEADER.name())) {
+ return info;
}
}
+ } catch (IOException e) {
+ throw new UncheckedIOException("Failed to fetch OM service list", e);
}
- return omLeaderUrl;
+ throw new IllegalStateException("No OM leader found in the OM service
list.");
}
- private boolean isOmSpnegoEnabled() {
- return configuration.get(OZONE_OM_HTTP_AUTH_TYPE, "simple")
- .equals("kerberos");
+ private String getLeaderNodeId() {
+ return getLeaderServiceInfo().getOmRoleInfo().getNodeId();
}
/**
- * Method to obtain current OM DB Snapshot.
- * @return DBCheckpoint instance.
+ * Obtain the current OM DB snapshot using the same OM-follower bootstrap
+ * mechanism (chunked, resumable {@code POST /v2/dbCheckpoint} with hard-link
+ * dedup on the leader). The returned {@link DBCheckpoint} points at
+ * a stable {@code om.snapshot.db_<ts>} directory ready to be promoted by
+ * {@link #updateReconOmDBWithNewSnapshot()}; returns {@code null} on
failure,
+ * leaving the current active DB untouched.
+ *
+ * @return DBCheckpoint instance, or {@code null} on failure.
*/
@VisibleForTesting
public DBCheckpoint getOzoneManagerDBSnapshot() {
- String snapshotFileName = RECON_OM_SNAPSHOT_DB + "_" +
System.currentTimeMillis();
- Path untarredDbDir = Paths.get(omSnapshotDBParentDir.getAbsolutePath(),
snapshotFileName);
-
- // Before fetching full snapshot again and create a new OM DB snapshot
directory, check and delete
- // any existing OM DB snapshot directories under recon om db dir location
and delete all such
- // om db snapshot dirs including the last known om db snapshot dir
returned by reconUtils.getLastKnownDB
- File lastKnownDB = reconUtils.getLastKnownDB(omSnapshotDBParentDir,
RECON_OM_SNAPSHOT_DB);
- if (lastKnownDB != null) {
- boolean existingOmSnapshotDBDeleted =
FileUtils.deleteQuietly(lastKnownDB);
- if (existingOmSnapshotDBDeleted) {
- LOG.info("Successfully deleted existing OM DB snapshot directory: {}",
- lastKnownDB.getAbsolutePath());
- } else {
- LOG.warn("Failed to delete existing OM DB snapshot directory: {}",
- lastKnownDB.getAbsolutePath());
- }
- }
-
- // Now below cleanup operation will even remove any left over staging dirs
in recon om db dir location which
- // may be left due to any previous partial extraction of tar entries and
during copy sst files process by
- // tarExtractor.extractTar
- File[] leftOverStagingDirs = omSnapshotDBParentDir.listFiles(f ->
f.getName().startsWith(STAGING));
- if (leftOverStagingDirs != null) {
- for (File stagingDir : leftOverStagingDirs) {
- LOG.warn("Cleaning up leftover staging folder from failed extraction:
{}", stagingDir.getAbsolutePath());
- boolean stagingDirDeleted = FileUtils.deleteQuietly(stagingDir);
- if (stagingDirDeleted) {
- LOG.info("Successfully deleted leftover staging folder: {}",
stagingDir.getAbsolutePath());
- } else {
- LOG.warn("Failed to delete leftover staging folder: {}",
stagingDir.getAbsolutePath());
- }
- }
- }
-
try {
- SecurityUtil.doAsLoginUser(() -> {
- try (InputStream inputStream = reconUtils.makeHttpCall(
- connectionFactory, getOzoneManagerSnapshotUrl(),
isOmSpnegoEnabled()).getInputStream()) {
- tarExtractor.extractTar(inputStream, untarredDbDir);
- } catch (IOException | InterruptedException e) {
- reconContext.updateHealthStatus(new AtomicBoolean(false));
-
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
- throw new RuntimeException("Error while extracting OM DB Snapshot
TAR.", e);
- }
- return null;
- });
- // Validate extracted files
- File[] sstFiles = untarredDbDir.toFile().listFiles((dir, name) ->
name.endsWith(".sst"));
- if (sstFiles != null && sstFiles.length > 0) {
- LOG.info("Number of SST files found in the OM snapshot directory: {} -
{}", untarredDbDir, sstFiles.length);
- }
-
- List<String> sstFileNames = Arrays.stream(sstFiles)
- .map(File::getName)
- .collect(Collectors.toList());
- LOG.debug("Valid SST files found: {}", sstFileNames);
-
- // Currently, OM DB type is not configurable. Hence, defaulting to
- // RocksDB.
+ String leaderNodeId = getLeaderNodeId();
+ DBCheckpoint checkpoint =
+ reconSnapshotProvider.downloadDBSnapshotFromLeader(leaderNodeId);
reconContext.updateHealthStatus(new AtomicBoolean(true));
-
reconContext.getErrors().remove(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
- return new RocksDBCheckpoint(untarredDbDir);
- } catch (IOException e) {
+ reconContext.getErrors()
+ .remove(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ return checkpoint;
+ } catch (IOException | RuntimeException e) {
LOG.error("Unable to obtain Ozone Manager DB Snapshot.", e);
reconContext.updateHealthStatus(new AtomicBoolean(false));
-
reconContext.updateErrors(ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ reconContext.updateErrors(
+ ReconContext.ErrorCode.GET_OM_DB_SNAPSHOT_FAILED);
+ // Reset the candidate dir so the next attempt starts from a clean state.
+ try {
+ reconSnapshotProvider.init();
Review Comment:
We don't support chunking/batched transfers from active om.db today so this
should not be a problem but in case we did this would be a problem as failure
in a single batch would delete the whole candidate dir. Just want to call it
out.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
