This runtime intentionally stops at filesystem and configuration - * preparation. Starting SCM, OM, datanodes, and S3 Gateway belongs to later - * HDDS-15084/HDDS-15086 work.
+ *Datanodes, S3 Gateway, Recon, and end-to-end key writes are added by + * later local runtime tickets.
*/ public final class LocalOzoneCluster implements LocalOzoneRuntime { @@ -78,6 +106,11 @@ public final class LocalOzoneCluster implements LocalOzoneRuntime { static final String PORTS_STATE_FILE_NAME = "ports.properties"; private static final String METADATA_DIR_NAME = "metadata"; + private static final String SCM_DIR_NAME = "scm"; + private static final String OM_DIR_NAME = "om"; + private static final String OZONE_METADATA_DIR_NAME = "ozone-metadata"; + private static final String DATA_DIR_NAME = "data"; + private static final String RATIS_DIR_NAME = "ratis"; private static final String SCM_CLIENT_PORT_KEY = "scm.client"; private static final String SCM_BLOCK_PORT_KEY = "scm.block"; private static final String SCM_DATANODE_PORT_KEY = "scm.datanode"; @@ -88,7 +121,11 @@ public final class LocalOzoneCluster implements LocalOzoneRuntime { private static final String SCM_GRPC_PORT_KEY = "scm.grpc"; private static final String OM_RPC_PORT_KEY = "om.rpc"; private static final String OM_HTTP_PORT_KEY = "om.http"; + private static final String OM_HTTPS_PORT_KEY = "om.https"; private static final String OM_RATIS_PORT_KEY = "om.ratis"; + private static final int LOCAL_RATIS_RPC_TIMEOUT_SECONDS = 1; + private static final long SCM_CLIENT_MAX_RETRY_TIMEOUT_MILLIS = 30_000; + private static final long READINESS_POLL_INTERVAL_MILLIS = 500; private static final int MAX_PORT = 65_535; @@ -103,13 +140,19 @@ public final class LocalOzoneCluster implements LocalOzoneRuntime { SCM_GRPC_PORT_KEY, OM_RPC_PORT_KEY, OM_HTTP_PORT_KEY, + OM_HTTPS_PORT_KEY, OM_RATIS_PORT_KEY }; private final LocalOzoneClusterConfig config; private final OzoneConfiguration seedConfiguration; + private boolean closed; private PreparedConfiguration preparedConfiguration; + private StorageContainerManager scm; + private OzoneManager om; + private boolean previousMetricsMiniClusterMode; + private boolean metricsMiniClusterModeEnabled; public LocalOzoneCluster(LocalOzoneClusterConfig config, OzoneConfiguration seedConfiguration) { @@ -119,8 +162,30 @@ public LocalOzoneCluster(LocalOzoneClusterConfig config, } @Override - public void start() throws IOException { - prepareConfiguration(); + public void start() throws Exception { + if (closed) { + throw new IOException("Local Ozone cluster is already closed."); + } + if (scm != null && om != null) { + return; + } + if (scm != null || om != null) { + throw new IOException("Local Ozone cluster is partially started."); + } + + try { + enableSameJvmMetricsMode(); + PreparedConfiguration prepared = prepareConfiguration(); + initializeStorage(prepared.getConfiguration()); + startScm(prepared.getConfiguration()); + startOm(prepared.getConfiguration()); + waitForScmAndOmReadiness(config.getStartupTimeout()); + } catch (Exception ex) { + // Roll back without latching closed: the caller's close() still owns the + // ephemeral data dir lifecycle. + stopServices(); + throw ex; + } } PreparedConfiguration prepareConfiguration() throws IOException { @@ -151,14 +216,12 @@ public String getDisplayHost() { @Override public int getScmPort() { - return preparedConfiguration == null ? config.getScmPort() - : preparedConfiguration.getScmPort(); + return scm.getClientRpcAddress().getPort(); } @Override public int getOmPort() { - return preparedConfiguration == null ? config.getOmPort() - : preparedConfiguration.getOmPort(); + return om.getOmRpcServerAddr().getPort(); } @Override @@ -173,14 +236,57 @@ public String getS3Endpoint() { @Override public void close() throws IOException { - // Ephemeral mode owns the data directory lifecycle for short-lived runs. - if (config.isEphemeral()) { - deleteDirectory(config.getDataDir()); + if (closed) { + return; + } + closed = true; + + try { + stopServices(); + } finally { + // Ephemeral mode owns the data directory lifecycle for short-lived runs. + if (config.isEphemeral()) { + try { + deleteDirectory(config.getDataDir()); + } catch (IOException ex) { + LOG.warn("Failed to delete local Ozone data dir {} during shutdown.", + config.getDataDir(), ex); + } + } + } + } + + private void stopServices() { + try { + // Shutdown is best-effort so one failed service cannot leak the other. + IOUtils.closeQuietly(this::stopOm, this::stopScm); + } finally { + restoreSameJvmMetricsMode(); + } + } + + private void stopOm() { + OzoneManager service = om; + om = null; + if (service != null && service.stop()) { + service.join(); + } + } + + private void stopScm() { + StorageContainerManager service = scm; + scm = null; + if (service != null) { + service.stop(); + service.join(); } } private void configureLocalDefaults(OzoneConfiguration conf) { conf.set(OZONE_METADATA_DIRS, metadataDir().toString()); + // SCM and OM share one configuration and JVM, so the Jetty base dir is a + // single service-neutral location; Jetty keeps per-context temp dirs. + conf.setIfUnset(OZONE_HTTP_BASEDIR, metadataDir() + SERVER_DIR); conf.set(OZONE_REPLICATION, ReplicationFactor.ONE.name()); conf.set(OZONE_REPLICATION_TYPE, ReplicationType.STAND_ALONE.name()); conf.set(OZONE_SERVER_DEFAULT_REPLICATION_KEY, ReplicationFactor.ONE.name()); @@ -190,6 +296,15 @@ private void configureLocalDefaults(OzoneConfiguration conf) { conf.setBoolean(HDDS_SCM_SAFEMODE_PIPELINE_CREATION, false); conf.setInt(HDDS_SCM_SAFEMODE_MIN_DATANODE, Math.max(1, config.getDatanodes())); + conf.setTimeDuration(OZONE_OM_RATIS_MINIMUM_TIMEOUT_KEY, + LOCAL_RATIS_RPC_TIMEOUT_SECONDS, TimeUnit.SECONDS); + conf.set(HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT, "3s"); + conf.setIfUnset(OZONE_SCM_HA_RATIS_SERVER_RPC_FIRST_ELECTION_TIMEOUT, + "1s"); + + SCMClientConfig scmClientConfig = conf.getObject(SCMClientConfig.class); + scmClientConfig.setMaxRetryTimeout(SCM_CLIENT_MAX_RETRY_TIMEOUT_MILLIS); + conf.setFromObject(scmClientConfig); } private int configureScm(OzoneConfiguration conf, @@ -234,9 +349,23 @@ 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()); + } + private int configureOm(OzoneConfiguration conf, PersistedPortState persistedPorts, PortAllocator portAllocator) throws IOException { @@ -244,16 +373,147 @@ private int configureOm(OzoneConfiguration conf, 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()); + } + + 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) { + requireStorageFormatting("SCM"); + } + // scmInit formats storage before Ratis so a crash between the two is + // recoverable, and re-initializes Ratis for a half-formatted directory. + if (!StorageContainerManager.scmInit(conf, clusterId)) { + throw new IOException("SCM initialization failed for local Ozone; see the SCM log for the cause."); + } + return new SCMStorageConfig(conf).getScmId(); + } + + 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 void startScm(OzoneConfiguration conf) throws Exception { + // Assign the field before start() so a failed start can still be rolled + // back by stopServices(). + scm = StorageContainerManager.createSCM(conf); + scm.start(); + } + + private void startOm(OzoneConfiguration conf) throws Exception { + om = OzoneManager.createOm(conf); + om.start(); + } + + private void waitForScmAndOmReadiness(Duration timeout) throws Exception { + long deadlineNanos = System.nanoTime() + timeout.toNanos(); + while (true) { + // Readiness is intentionally scoped to SCM and OM leadership until + // datanode and full safe-mode readiness are added in later tickets. + if (scm.checkLeader() && om.isLeaderReady()) { + return; + } + if (System.nanoTime() >= deadlineNanos) { + throw new TimeoutException("Timed out waiting " + timeout + + " for local SCM and OM leadership."); + } + Thread.sleep(READINESS_POLL_INTERVAL_MILLIS); + } + } + + private void enableSameJvmMetricsMode() { + if (!metricsMiniClusterModeEnabled) { + previousMetricsMiniClusterMode = DefaultMetricsSystem.inMiniClusterMode(); + DefaultMetricsSystem.setMiniClusterMode(true); + metricsMiniClusterModeEnabled = true; + } + } + + private void restoreSameJvmMetricsMode() { + if (metricsMiniClusterModeEnabled) { + DefaultMetricsSystem.setMiniClusterMode(previousMetricsMiniClusterMode); + metricsMiniClusterModeEnabled = false; + } + } + private void prepareStorageLayout() throws IOException { Path dataDir = config.getDataDir(); if (Files.exists(dataDir) && !Files.isDirectory(dataDir)) { @@ -334,7 +594,11 @@ private static String address(String host, int port) { return host + ":" + port; } - private static void deleteDirectory(Path directory) { + private static String initializedClusterId(Storage storage) { + return storage.getState() == INITIALIZED ? storage.getClusterID() : null; + } + + private static void deleteDirectory(Path directory) throws IOException { if (!Files.exists(directory)) { return; } @@ -344,8 +608,6 @@ private static void deleteDirectory(Path directory) { for (Path path : deleteOrder) { Files.deleteIfExists(path); } - } catch (IOException ex) { - LOG.error("Failed to delete local Ozone directory {}", directory, ex); } } diff --git a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneCluster.java b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneCluster.java index c62c108ce712..8ecc3a9db982 100644 --- a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneCluster.java +++ b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneCluster.java @@ -59,18 +59,19 @@ class TestLocalOzoneCluster { private Path tempDir; @Test - void startPreparesConfiguration() throws Exception { + void prepareConfigurationExposesPreparedPorts() throws Exception { Path dataDir = tempDir.resolve("local-ozone"); LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(dataDir).build(); try (LocalOzoneCluster cluster = newCluster(config)) { - cluster.start(); + LocalOzoneCluster.PreparedConfiguration prepared = + cluster.prepareConfiguration(); assertTrue(Files.isDirectory(metadataDir(dataDir))); assertTrue(Files.isRegularFile(portStateFile(dataDir))); - assertTrue(cluster.getScmPort() > 0); - assertTrue(cluster.getOmPort() > 0); + assertTrue(prepared.getScmPort() > 0); + assertTrue(prepared.getOmPort() > 0); assertEquals(-1, cluster.getS3gPort()); assertEquals("", cluster.getS3Endpoint()); }