henrybear327 commented on code in PR #10456:
URL: https://github.com/apache/ozone/pull/10456#discussion_r3570282874
##########
hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/LocalOzoneCluster.java:
##########
@@ -227,26 +334,183 @@ private int configureScm(OzoneConfiguration conf,
conf.setInt(OZONE_SCM_GRPC_PORT_KEY, scmGrpcPort);
conf.setStrings(OZONE_SCM_NAMES,
address(config.getHost(), scmDatanodePort));
+ configureScmStorage(conf);
return scmClientPort;
}
+ private void configureScmStorage(OzoneConfiguration conf) throws IOException
{
+ Path scmDir = config.getDataDir().resolve(SCM_DIR_NAME);
+ Path scmMetadataDir = scmDir.resolve(OZONE_METADATA_DIR_NAME);
+ Files.createDirectories(scmDir.resolve(DATA_DIR_NAME));
+
+ conf.setIfUnset(OZONE_SCM_DB_DIRS,
+ scmDir.resolve(DATA_DIR_NAME).toString());
+ conf.setIfUnset(OZONE_SCM_HA_RATIS_STORAGE_DIR,
+ scmDir.resolve(RATIS_DIR_NAME).toString());
+ conf.setIfUnset(OZONE_SCM_HA_RATIS_SNAPSHOT_DIR,
+ scmMetadataDir.resolve(OZONE_RATIS_SNAPSHOT_DIR).toString());
+ conf.setIfUnset(OZONE_HTTP_BASEDIR, scmMetadataDir + SERVER_DIR);
+ }
+
private int configureOm(OzoneConfiguration conf,
PersistedPortState persistedPorts, PortAllocator portAllocator)
throws IOException {
int omRpcPort = reservePort(portAllocator, persistedPorts, OM_RPC_PORT_KEY,
config.getOmPort());
int omHttpPort = reservePort(portAllocator, persistedPorts,
OM_HTTP_PORT_KEY, 0);
+ int omHttpsPort = reservePort(portAllocator, persistedPorts,
+ OM_HTTPS_PORT_KEY, 0);
int omRatisPort = reservePort(portAllocator, persistedPorts,
OM_RATIS_PORT_KEY, 0);
conf.set(OZONE_OM_ADDRESS_KEY, address(config.getHost(), omRpcPort));
conf.set(OZONE_OM_HTTP_ADDRESS_KEY, address(config.getHost(), omHttpPort));
conf.set(OZONE_OM_HTTP_BIND_HOST_KEY, config.getBindHost());
+ conf.set(OZONE_OM_HTTPS_ADDRESS_KEY,
+ address(config.getHost(), omHttpsPort));
+ conf.set(OZONE_OM_HTTPS_BIND_HOST_KEY, config.getBindHost());
conf.setInt(OZONE_OM_RATIS_PORT_KEY, omRatisPort);
+ configureOmStorage(conf);
return omRpcPort;
}
+ private void configureOmStorage(OzoneConfiguration conf) throws IOException {
+ Path omDir = config.getDataDir().resolve(OM_DIR_NAME);
+ Path omMetadataDir = omDir.resolve(OZONE_METADATA_DIR_NAME);
+ Files.createDirectories(omDir.resolve(DATA_DIR_NAME));
+
+ conf.setIfUnset(OZONE_OM_DB_DIRS, omDir.resolve(DATA_DIR_NAME).toString());
+ conf.setIfUnset(OZONE_OM_RATIS_STORAGE_DIR,
+ omDir.resolve(RATIS_DIR_NAME).toString());
+ conf.setIfUnset(OZONE_OM_RATIS_SNAPSHOT_DIR,
+ omMetadataDir.resolve(OZONE_RATIS_SNAPSHOT_DIR).toString());
+ conf.setIfUnset(OZONE_OM_SNAPSHOT_DIFF_DB_DIR,
+ omMetadataDir.toString());
+ conf.setIfUnset(OZONE_HTTP_BASEDIR, omMetadataDir + SERVER_DIR);
+ }
+
+ private void initializeStorage(OzoneConfiguration conf) throws IOException {
+ SCMStorageConfig scmStorage = new SCMStorageConfig(conf);
+ OMStorage omStorage = new OMStorage(conf);
+ String clusterId = resolveClusterId(scmStorage, omStorage);
+ String scmId = initializeScmStorage(conf, scmStorage, clusterId);
+ initializeOmStorage(conf, omStorage, clusterId, scmId);
+ }
+
+ private String resolveClusterId(SCMStorageConfig scmStorage,
+ OMStorage omStorage) throws IOException {
+ String scmClusterId = initializedClusterId(scmStorage);
+ String omClusterId = initializedClusterId(omStorage);
+ if (scmClusterId != null && omClusterId != null
+ && !scmClusterId.equals(omClusterId)) {
+ throw new IOException("Local Ozone SCM cluster ID " + scmClusterId
+ + " does not match OM cluster ID " + omClusterId + ".");
+ }
+ // Reuse an initialized component's cluster ID so local metadata survives
+ // restarting with a partially formatted data directory.
+ if (scmClusterId != null) {
+ return scmClusterId;
+ }
+ if (omClusterId != null) {
+ return omClusterId;
+ }
+ return UUID.randomUUID().toString();
+ }
+
+ private String initializeScmStorage(OzoneConfiguration conf,
+ SCMStorageConfig scmStorage, String clusterId) throws IOException {
+ if (scmStorage.getState() == INITIALIZED) {
+ return scmStorage.getScmId();
+ }
+
+ requireStorageFormatting("SCM");
+ String scmId = UUID.randomUUID().toString();
+ scmStorage.setClusterId(clusterId);
+ scmStorage.setScmId(scmId);
+ scmStorage.setPrimaryScmNodeId(scmId);
+ scmStorage.initialize();
+ scmStorage.setSCMHAFlag(true);
+ scmStorage.persistCurrentState();
+
+ SCMRatisServerImpl.initialize(clusterId, scmId,
+ SCMHANodeDetails.loadSCMHAConfig(conf, scmStorage)
+ .getLocalNodeDetails(), conf);
+ HddsUtils.createDir(SCMHAUtils.getSCMRatisSnapshotDirectory(conf));
+ return scmId;
+ }
+
+ private void initializeOmStorage(OzoneConfiguration conf, OMStorage
omStorage,
+ String clusterId, String scmId) throws IOException {
+ if (omStorage.getState() == INITIALIZED) {
+ if (!clusterId.equals(omStorage.getClusterID())) {
+ throw new IOException("Local Ozone OM cluster ID "
+ + omStorage.getClusterID() + " does not match SCM cluster ID "
+ + clusterId + ".");
+ }
+ return;
+ }
+
+ requireStorageFormatting("OM");
+ omStorage.setClusterId(clusterId);
+ omStorage.setOmId(UUID.randomUUID().toString());
+ if (OzoneSecurityUtil.isSecurityEnabled(conf)) {
+ OzoneManager.initializeSecurity(conf, omStorage, scmId);
+ }
+ omStorage.initialize();
+ }
+
+ private void requireStorageFormatting(String component) throws IOException {
+ if (config.getFormatMode() == LocalOzoneClusterConfig.FormatMode.NEVER) {
+ throw new IOException(component + " storage is not initialized. "
+ + "Format mode NEVER requires existing SCM and OM storage.");
+ }
+ }
+
+ private StorageContainerManager startScm(OzoneConfiguration conf)
+ throws Exception {
+ StorageContainerManager manager = StorageContainerManager.createSCM(conf);
+ manager.start();
+ return manager;
+ }
+
+ private OzoneManager startOm(OzoneConfiguration conf) throws Exception {
+ OzoneManager manager = OzoneManager.createOm(conf);
+ manager.start();
+ return manager;
+ }
+
+ private void waitForScmAndOmReadiness(Duration timeout) throws Exception {
+ long deadlineNanos = System.nanoTime() + timeout.toNanos();
+ while (System.nanoTime() < deadlineNanos) {
+ // Readiness is intentionally scoped to SCM leadership and RPC
reachability
+ // until datanode and full safe-mode readiness are added in later
tickets.
+ if (scm.checkLeader()
+ && isPortOpen(getDisplayHost(), getScmPort())
+ && isPortOpen(getDisplayHost(), getOmPort())) {
Review Comment:
Refactoring. Thanks for pointing this 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]
