This implementation manages the lifecycle of datanodes (and eventually + * SCM, OM, S3G) running within a single JVM process.
+ */ +public final class LocalOzoneCluster implements LocalOzoneRuntime { + + private static final Logger LOG = + LoggerFactory.getLogger(LocalOzoneCluster.class); + + private static final String[] NO_ARGS = new String[0]; + private static final String PORTS_STATE_FILE = "ports.properties"; + + private final LocalOzoneClusterConfig config; + private final OzoneConfiguration seedConfiguration; + private final AtomicBoolean closed = new AtomicBoolean(); + private final ListThis immutable configuration holds all settings needed to start + * a local Ozone cluster including datanode count, network settings, + * and storage options.
+ */ +public final class LocalOzoneClusterConfig { + + /** Default advertised host for local services. */ + public static final String DEFAULT_HOST = "127.0.0.1"; + + /** Default bind host for local services. */ + public static final String DEFAULT_BIND_HOST = "0.0.0.0"; + + /** Default number of datanodes to start. */ + public static final int DEFAULT_DATANODES = 1; + + /** Default timeout waiting for cluster to become ready. */ + public static final Duration DEFAULT_STARTUP_TIMEOUT = Duration.ofMinutes(2); + + private final Path dataDir; + private final FormatMode formatMode; + private final int datanodes; + private final boolean ephemeral; + private final String host; + private final String bindHost; + private final Duration startupTimeout; + + private LocalOzoneClusterConfig(Builder builder) { + this.dataDir = Objects.requireNonNull(builder.dataDir, "dataDir"); + this.formatMode = Objects.requireNonNull(builder.formatMode, "formatMode"); + this.datanodes = builder.datanodes; + this.ephemeral = builder.ephemeral; + this.host = Objects.requireNonNull(builder.host, "host"); + this.bindHost = Objects.requireNonNull(builder.bindHost, "bindHost"); + this.startupTimeout = Objects.requireNonNull(builder.startupTimeout, + "startupTimeout"); + } + + /** + * Returns the root data directory for the local cluster. + */ + public Path getDataDir() { + return dataDir; + } + + /** + * Returns the storage format mode. + */ + public FormatMode getFormatMode() { + return formatMode; + } + + /** + * Returns the number of datanodes to start. + */ + public int getDatanodes() { + return datanodes; + } + + /** + * Returns whether the data directory should be deleted on shutdown. + */ + public boolean isEphemeral() { + return ephemeral; + } + + /** + * Returns the advertised host for service addresses. + */ + public String getHost() { + return host; + } + + /** + * Returns the bind host for service listeners. + */ + public String getBindHost() { + return bindHost; + } + + /** + * Returns the timeout for waiting for cluster readiness. + */ + public Duration getStartupTimeout() { + return startupTimeout; + } + + /** + * Creates a new builder with the specified data directory. + * + * @param dataDir the root data directory for the local cluster + * @return a new builder instance + */ + public static Builder builder(Path dataDir) { + return new Builder(dataDir); + } + + /** + * Storage initialization mode for the local runtime. + */ + public enum FormatMode { + /** Format storage only if not already initialized. */ + IF_NEEDED, + /** Always format storage, destroying existing data. */ + ALWAYS, + /** Never format; fail if storage is not initialized. */ + NEVER; + + /** + * Parses a format mode from string representation. + * + * @param value the string value (e.g., "if-needed", "always", "never") + * @return the corresponding FormatMode + * @throws IllegalArgumentException if the value is not recognized + */ + public static FormatMode fromString(String value) { + return valueOf(value.trim().toUpperCase(Locale.ROOT).replace('-', '_')); + } + } + + /** + * Builder for {@link LocalOzoneClusterConfig}. + */ + public static final class Builder { + + private final Path dataDir; + private FormatMode formatMode = FormatMode.IF_NEEDED; + private int datanodes = DEFAULT_DATANODES; + private boolean ephemeral; + private String host = DEFAULT_HOST; + private String bindHost = DEFAULT_BIND_HOST; + private Duration startupTimeout = DEFAULT_STARTUP_TIMEOUT; + + private Builder(Path dataDir) { + this.dataDir = dataDir.toAbsolutePath().normalize(); + } + + /** + * Sets the storage format mode. + */ + public Builder setFormatMode(FormatMode value) { + this.formatMode = value; + return this; + } + + /** + * Sets the number of datanodes to start. + * + * @param value the datanode count, must be at least 1 + * @throws IllegalArgumentException if value is less than 1 + */ + public Builder setDatanodes(int value) { + if (value < 1) { + throw new IllegalArgumentException( + "Datanode count must be at least 1, got: " + value); + } + this.datanodes = value; + return this; + } + + /** + * Sets whether the data directory should be deleted on shutdown. + */ + public Builder setEphemeral(boolean value) { + this.ephemeral = value; + return this; + } + + /** + * Sets the advertised host for service addresses. + */ + public Builder setHost(String value) { + this.host = value; + return this; + } + + /** + * Sets the bind host for service listeners. + */ + public Builder setBindHost(String value) { + this.bindHost = value; + return this; + } + + /** + * Sets the timeout for waiting for cluster readiness. + */ + public Builder setStartupTimeout(Duration value) { + this.startupTimeout = value; + return this; + } + + /** + * Builds the configuration. + * + * @return an immutable LocalOzoneClusterConfig instance + */ + public LocalOzoneClusterConfig build() { + return new LocalOzoneClusterConfig(this); + } + } +} diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PersistedPorts.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PersistedPorts.java new file mode 100644 index 000000000000..8e14cd5e2597 --- /dev/null +++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PersistedPorts.java @@ -0,0 +1,92 @@ +/* + * 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. + */ + +package org.apache.hadoop.ozone.local; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Properties; + +/** + * Persists allocated ports to enable stable endpoints across restarts. + * + *When a local Ozone cluster starts, it allocates ephemeral ports for + * its services. This class saves those ports to a properties file so that + * subsequent restarts can reuse the same ports, providing stable endpoints + * for clients.
+ */ +final class PersistedPorts { + + private final Path path; + private final Properties properties = new Properties(); + + private PersistedPorts(Path path) { + this.path = path; + } + + /** + * Loads persisted ports from the specified file. + * + * @param path the path to the ports properties file + * @return a PersistedPorts instance, empty if file doesn't exist + * @throws IOException if reading the file fails + */ + static PersistedPorts load(Path path) throws IOException { + PersistedPorts persistedPorts = new PersistedPorts(path); + if (Files.exists(path)) { + try (InputStream input = Files.newInputStream(path)) { + persistedPorts.properties.load(input); + } + } + return persistedPorts; + } + + /** + * Gets a previously persisted port value. + * + * @param key the port identifier (e.g., "dn.0.container.ipc") + * @return the port number, or 0 if not persisted + */ + int get(String key) { + String value = properties.getProperty(key); + return value == null ? 0 : Integer.parseInt(value); + } + + /** + * Sets a port value to be persisted. + * + * @param key the port identifier + * @param port the port number + */ + void set(String key, int port) { + properties.setProperty(key, Integer.toString(port)); + } + + /** + * Saves all port values to the file. + * + * @throws IOException if writing the file fails + */ + void store() throws IOException { + try (OutputStream output = Files.newOutputStream(path)) { + properties.store(output, "Local Ozone reserved ports"); + } + } +} diff --git a/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PortAllocator.java b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PortAllocator.java new file mode 100644 index 000000000000..822e6602e074 --- /dev/null +++ b/hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/local/PortAllocator.java @@ -0,0 +1,72 @@ +/* + * 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. + */ + +package org.apache.hadoop.ozone.local; + +import java.io.IOException; +import java.net.ServerSocket; +import java.util.HashSet; +import java.util.Set; + +/** + * Allocates unique ports for local Ozone services. + * + *This allocator ensures that each service gets a unique port, + * either by using a preferred port or by finding a free ephemeral port. + * It tracks all reserved ports to prevent conflicts when multiple + * services are started in the same JVM.
+ */ +final class PortAllocator { + + private final SetThese tests focus on configuration generation without starting + * actual services, which require SCM.
+ */ +class TestLocalOzoneCluster { + + @TempDir + private Path tempDir; + + @Test + void prepareBaseConfigurationSetsReplicationDefaults() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + OzoneConfiguration baseConf = cluster.prepareBaseConfiguration(); + + assertEquals(ReplicationFactor.ONE.name(), + baseConf.get(OZONE_REPLICATION)); + assertEquals(ReplicationType.STAND_ALONE.name(), + baseConf.get(OZONE_REPLICATION_TYPE)); + assertFalse(baseConf.getBoolean(HDDS_CONTAINER_RATIS_ENABLED_KEY, true)); + } + + @Test + void prepareBaseConfigurationCreatesMetadataDir() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + OzoneConfiguration baseConf = cluster.prepareBaseConfiguration(); + + String metadataDir = baseConf.get(OZONE_METADATA_DIRS); + assertTrue(Files.exists(Paths.get(metadataDir)), + "Metadata directory should be created"); + assertTrue(metadataDir.contains("metadata"), + "Metadata dir path should contain 'metadata'"); + } + + @Test + void getDisplayHostReturnsConfiguredHost() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .setHost("192.168.1.100") + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + assertEquals("192.168.1.100", cluster.getDisplayHost()); + } + + @Test + void getDisplayHostReturnsLocalhostForBindAll() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .setHost("0.0.0.0") + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + assertEquals("127.0.0.1", cluster.getDisplayHost()); + } + + @Test + void getDatanodeCountReturnsZeroBeforeStart() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .setDatanodes(3) + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + assertEquals(0, cluster.getDatanodeCount(), + "Should have zero datanodes before start"); + } + + @Test + void closeIsIdempotent() throws Exception { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .build(); + + LocalOzoneCluster cluster = new LocalOzoneCluster(config, + new OzoneConfiguration()); + + // Multiple closes should not throw + cluster.close(); + cluster.close(); + cluster.close(); + } +} diff --git a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneClusterConfig.java b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneClusterConfig.java new file mode 100644 index 000000000000..f8103a10df03 --- /dev/null +++ b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestLocalOzoneClusterConfig.java @@ -0,0 +1,113 @@ +/* + * 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. + */ + +package org.apache.hadoop.ozone.local; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import org.apache.hadoop.ozone.local.LocalOzoneClusterConfig.FormatMode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Unit tests for {@link LocalOzoneClusterConfig}. + */ +class TestLocalOzoneClusterConfig { + + @TempDir + private Path tempDir; + + @Test + void builderSetsDefaults() { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .build(); + + assertEquals(tempDir.toAbsolutePath().normalize(), config.getDataDir()); + assertEquals(FormatMode.IF_NEEDED, config.getFormatMode()); + assertEquals(LocalOzoneClusterConfig.DEFAULT_DATANODES, + config.getDatanodes()); + assertEquals(LocalOzoneClusterConfig.DEFAULT_HOST, config.getHost()); + assertEquals(LocalOzoneClusterConfig.DEFAULT_BIND_HOST, + config.getBindHost()); + assertEquals(LocalOzoneClusterConfig.DEFAULT_STARTUP_TIMEOUT, + config.getStartupTimeout()); + assertFalse(config.isEphemeral()); + } + + @Test + void builderAcceptsCustomValues() { + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(tempDir) + .setFormatMode(FormatMode.ALWAYS) + .setDatanodes(3) + .setHost("192.168.1.100") + .setBindHost("192.168.1.100") + .setStartupTimeout(Duration.ofMinutes(5)) + .setEphemeral(true) + .build(); + + assertEquals(FormatMode.ALWAYS, config.getFormatMode()); + assertEquals(3, config.getDatanodes()); + assertEquals("192.168.1.100", config.getHost()); + assertEquals("192.168.1.100", config.getBindHost()); + assertEquals(Duration.ofMinutes(5), config.getStartupTimeout()); + assertTrue(config.isEphemeral()); + } + + @Test + void builderRejectsInvalidDatanodeCount() { + LocalOzoneClusterConfig.Builder builder = + LocalOzoneClusterConfig.builder(tempDir); + + assertThrows(IllegalArgumentException.class, + () -> builder.setDatanodes(0)); + assertThrows(IllegalArgumentException.class, + () -> builder.setDatanodes(-1)); + } + + @Test + void dataDirIsNormalized() { + Path unnormalized = Paths.get(tempDir.toString(), "subdir", "..", "data"); + LocalOzoneClusterConfig config = LocalOzoneClusterConfig.builder(unnormalized) + .build(); + + Path expected = tempDir.resolve("data").toAbsolutePath().normalize(); + assertEquals(expected, config.getDataDir()); + } + + @Test + void formatModeFromStringParsesValidValues() { + assertEquals(FormatMode.IF_NEEDED, FormatMode.fromString("if-needed")); + assertEquals(FormatMode.IF_NEEDED, FormatMode.fromString("IF_NEEDED")); + assertEquals(FormatMode.IF_NEEDED, FormatMode.fromString("If-Needed")); + assertEquals(FormatMode.ALWAYS, FormatMode.fromString("always")); + assertEquals(FormatMode.NEVER, FormatMode.fromString("never")); + } + + @Test + void formatModeFromStringRejectsInvalidValues() { + assertThrows(IllegalArgumentException.class, + () -> FormatMode.fromString("invalid")); + assertThrows(IllegalArgumentException.class, + () -> FormatMode.fromString("")); + } +} diff --git a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPersistedPorts.java b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPersistedPorts.java new file mode 100644 index 000000000000..ee00128bbc26 --- /dev/null +++ b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPersistedPorts.java @@ -0,0 +1,116 @@ +/* + * 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. + */ + +package org.apache.hadoop.ozone.local; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Unit tests for {@link PersistedPorts}. + */ +class TestPersistedPorts { + + @TempDir + private Path tempDir; + + @Test + void loadFromNonExistentFileReturnsEmpty() throws IOException { + Path portsFile = tempDir.resolve("ports.properties"); + PersistedPorts ports = PersistedPorts.load(portsFile); + + assertEquals(0, ports.get("nonexistent.key")); + } + + @Test + void setAndGetReturnsValue() throws IOException { + Path portsFile = tempDir.resolve("ports.properties"); + PersistedPorts ports = PersistedPorts.load(portsFile); + + ports.set("dn.0.http", 9878); + assertEquals(9878, ports.get("dn.0.http")); + } + + @Test + void storeAndLoadPreservesValues() throws IOException { + Path portsFile = tempDir.resolve("ports.properties"); + + // Store some ports + PersistedPorts ports1 = PersistedPorts.load(portsFile); + ports1.set("dn.0.http", 9878); + ports1.set("dn.0.client", 9879); + ports1.set("dn.1.http", 9880); + ports1.store(); + + assertTrue(Files.exists(portsFile), "Ports file should be created"); + + // Load and verify + PersistedPorts ports2 = PersistedPorts.load(portsFile); + assertEquals(9878, ports2.get("dn.0.http")); + assertEquals(9879, ports2.get("dn.0.client")); + assertEquals(9880, ports2.get("dn.1.http")); + assertEquals(0, ports2.get("nonexistent")); + } + + @Test + void storeOverwritesExistingFile() throws IOException { + Path portsFile = tempDir.resolve("ports.properties"); + + // First store + PersistedPorts ports1 = PersistedPorts.load(portsFile); + ports1.set("key1", 1111); + ports1.store(); + + // Second store with different value + PersistedPorts ports2 = PersistedPorts.load(portsFile); + ports2.set("key1", 2222); + ports2.set("key2", 3333); + ports2.store(); + + // Verify final state + PersistedPorts ports3 = PersistedPorts.load(portsFile); + assertEquals(2222, ports3.get("key1")); + assertEquals(3333, ports3.get("key2")); + } + + @Test + void loadPreservesExistingValuesWhenAddingNew() throws IOException { + Path portsFile = tempDir.resolve("ports.properties"); + + // Store initial value + PersistedPorts ports1 = PersistedPorts.load(portsFile); + ports1.set("existing", 1111); + ports1.store(); + + // Load, add new, store + PersistedPorts ports2 = PersistedPorts.load(portsFile); + assertEquals(1111, ports2.get("existing")); + ports2.set("new", 2222); + ports2.store(); + + // Verify both exist + PersistedPorts ports3 = PersistedPorts.load(portsFile); + assertEquals(1111, ports3.get("existing")); + assertEquals(2222, ports3.get("new")); + } +} diff --git a/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPortAllocator.java b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPortAllocator.java new file mode 100644 index 000000000000..e1f85a5c8e60 --- /dev/null +++ b/hadoop-ozone/tools/src/test/java/org/apache/hadoop/ozone/local/TestPortAllocator.java @@ -0,0 +1,99 @@ +/* + * 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. + */ + +package org.apache.hadoop.ozone.local; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.util.HashSet; +import java.util.Set; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link PortAllocator}. + */ +class TestPortAllocator { + + @Test + void reserveWithPreferredPortReturnsPreferred() throws IOException { + PortAllocator allocator = new PortAllocator(); + int port = allocator.reserve(9878); + assertEquals(9878, port); + } + + @Test + void reserveWithZeroAllocatesEphemeralPort() throws IOException { + PortAllocator allocator = new PortAllocator(); + int port = allocator.reserve(0); + assertTrue(port > 0, "Should allocate a valid port"); + assertTrue(port <= 65535, "Port should be in valid range"); + } + + @Test + void reserveAllocatesUniqueEphemeralPorts() throws IOException { + PortAllocator allocator = new PortAllocator(); + Set