hanishakoneru commented on a change in pull request #948: HDDS-1649. On installSnapshot notification from OM leader, download checkpoint and reload OM state URL: https://github.com/apache/hadoop/pull/948#discussion_r305113075
########## File path: hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java ########## @@ -3150,6 +3169,195 @@ public boolean setAcl(OzoneObj obj, List<OzoneAcl> acls) throws IOException { } } + /** + * Download and install latest checkpoint from leader OM. + * If the download checkpoints snapshot index is greater than this OM'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. + * @param leaderId peerNodeID of the leader OM + * @return If checkpoint is installed, return the corresponding termIndex. + * Otherwise, return null. + */ + public TermIndex installSnapshot(String leaderId) { + if (omSnapshotProvider == null) { + LOG.error("OM Snapshot Provider is not configured as there are no peer " + + "nodes."); + return null; + } + + DBCheckpoint omDBcheckpoint = getDBCheckpointFromLeader(leaderId); + + // Check if current ratis log index is smaller than the downloaded + // snapshot index. If yes, proceed by stopping the ratis server so that + // the OM state can be re-initialized. If no, then do not proceed with + // installSnapshot. + long lastAppliedIndex = omRatisServer.getStateMachineLastAppliedIndex(); + long checkpointSnapshotIndex = omDBcheckpoint.getRatisSnapshotIndex(); + if (checkpointSnapshotIndex <= lastAppliedIndex) { + LOG.error("Failed to install checkpoint from OM leader: {}. The last " + + "applied index: {} is greater than or equal to the checkpoint's " + + "snapshot index: {}", leaderId, lastAppliedIndex, + checkpointSnapshotIndex); + return null; + } + + // Pause the State Machine so that no new transactions can be applied. + // This action also clears the OM Double Buffer so that if there are any + // pending transactions in the buffer, they are discarded. + // TODO: The Ratis server should also be paused here. This is required + // because a leader election might happen while the snapshot + // installation is in progress and the new leader might start sending + // append log entries to the ratis server. + omRatisServer.getOmStateMachine().pause(); + + try { + replaceOMDBWithCheckpoint(lastAppliedIndex, omDBcheckpoint); + } catch (Exception e) { + LOG.error("OM DB checkpoint replacement with new downloaded checkpoint " + + "failed.", e); + return null; + } + + // Reload the OM 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 { + reloadOMState(); + omRatisServer.getOmStateMachine().unpause(checkpointSnapshotIndex); + } catch (IOException e) { + LOG.error("Failed to reload OM state with new DB checkpoint.", e); + return null; + } + + // TODO: We should only return the snpashotIndex to the leader. + // Should be fixed after RATIS-586 + TermIndex newTermIndex = TermIndex.newTermIndex(0, + checkpointSnapshotIndex); + + return newTermIndex; + } + + /** + * Download the latest OM DB checkpoint from the leader OM. + * @param leaderId OMNodeID of the leader OM node. + * @return latest DB checkpoint from leader OM. + */ + private DBCheckpoint getDBCheckpointFromLeader(String leaderId) { + LOG.info("Downloading checkpoint from leader OM {} and reloading state " + + "from the checkpoint.", leaderId); + + try { + return omSnapshotProvider.getOzoneManagerDBSnapshot(leaderId); + } catch (IOException e) { + LOG.error("Failed to download checkpoint from OM leader {}", leaderId, e); + } + return null; + } + + /** + * Replace the current OM DB with the new DB checkpoint. + * @param lastAppliedIndex the last applied index in the current OM DB. + * @param omDBcheckpoint the new DB checkpoint + * @throws Exception + */ + void replaceOMDBWithCheckpoint(long lastAppliedIndex, + DBCheckpoint omDBcheckpoint) throws Exception { + // Stop the DB first + DBStore store = metadataManager.getStore(); + store.close(); + + // Take a backup of the current DB + File db = store.getDbLocation(); + String dbBackupName = OzoneConsts.OM_DB_BACKUP_PREFIX + + lastAppliedIndex + "_" + System.currentTimeMillis(); + File dbBackup = new File(db.getParentFile(), dbBackupName); + + try { + Files.move(db.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 om metadata dir + Path checkpointPath = omDBcheckpoint.getCheckpointLocation(); + try { + Files.move(checkpointPath, db.toPath()); + } catch (IOException e) { + LOG.error("Failed to move downloaded DB checkpoint {} to metadata " + + "directory {}",checkpointPath, db.toPath()); + throw e; + } + } + + /** + * Re-instantiate MetadataManager with new DB checkpoint. + * All the classes which use/ store MetadataManager should also be updated + * with the new MetadataManager instance. + */ + void reloadOMState() throws IOException { + + metadataManager = new OmMetadataManagerImpl(configuration); + + metadataManager.start(configuration); + + // Set metrics and start metrics back ground thread + metrics.setNumVolumes(metadataManager.countRowsInTable(metadataManager + .getVolumeTable())); + metrics.setNumBuckets(metadataManager.countRowsInTable(metadataManager + .getBucketTable())); + + // Delete the omMetrics file if it exists + Files.deleteIfExists(getMetricsStorageFile().toPath()); + + // Re-initialize metadataManager dependent implementations + volumeManager = new VolumeManagerImpl(metadataManager, configuration); + bucketManager = new BucketManagerImpl(metadataManager, getKmsProvider(), + isRatisEnabled); + s3BucketManager = new S3BucketManagerImpl(configuration, metadataManager, + volumeManager, bucketManager); + if (secConfig.isSecurityEnabled()) { + s3SecretManager = new S3SecretManagerImpl(configuration, metadataManager); + } + + keyManager = new KeyManagerImpl( + new ScmClient(scmBlockClient, scmContainerClient), metadataManager, + configuration, omStorage.getOmId(), blockTokenMgr, getKmsProvider(), + prefixManager); + keyManager.start(configuration); + prefixManager = new PrefixManagerImpl(metadataManager); + } + + /** + * Startup options. + */ + public enum StartupOption { + INIT("--init"), + HELP("--help"), + REGULAR("--regular"); + Review comment: This is by mistake. Not sure how this came here. Removing it. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: common-issues-unsubscr...@hadoop.apache.org For additional commands, e-mail: common-issues-h...@hadoop.apache.org