From d0513cc6ce1b3b784d79a85b4b901b2df3369129 Mon Sep 17 00:00:00 2001 From: Faras Mohan Dewal Date: Sat, 7 Aug 2021 05:21:29 -0600 Subject: [PATCH] Feature OpenLookeng ON YARN Signed-off-by: fmdewal --- .../TestHetuMetastoreGlobalCache.java | 3 +- hetu-seed-store/pom.xml | 8 + .../seedstore/filebased/FileBasedSeed.java | 71 ++--- .../filebased/FileBasedSeedConstants.java | 5 +- .../filebased/FileBasedSeedStore.java | 2 +- .../filebased/FileBasedSeedStoreFactory.java | 10 +- .../filebased/FileBasedSeedStoreOnYarn.java | 217 ++++++++++++++++ .../filebased/TestFileBasedSeedStore.java | 9 +- hetu-state-store/pom.xml | 6 +- ...TestHazelcastClusterLifecycleListener.java | 25 +- .../TestHazelcastStateStoreFactory.java | 2 +- .../discovery/HetuServiceInventory.java | 67 +++-- .../httpserver/HetuHttpServerInfo.java | 222 ++++++++++++++++ .../httpserver/HetuHttpServerModule.java | 31 +++ .../prestosql/seedstore/SeedStoreManager.java | 244 ++++++++++++------ .../io/prestosql/server/PrestoServer.java | 41 ++- .../EmbeddedStateStoreLauncher.java | 43 ++- .../statestore/LocalStateStoreProvider.java | 3 +- .../discovery/TestHetuServiceInventory.java | 2 + .../TestDynamicFilterSourceOperator.java | 3 +- .../seedstore/TestSeedStoreManager.java | 58 +++-- .../TestStateStoreLauncherAndProvider.java | 9 +- .../java/io/prestosql/spi/seedstore/Seed.java | 14 + .../spi/seedstore/SeedStoreFactory.java | 2 +- .../spi/seedstore/SeedStoreSubType.java | 21 ++ .../DistributedQueryRunnerWithStateStore.java | 5 +- 26 files changed, 931 insertions(+), 192 deletions(-) create mode 100644 hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java create mode 100644 presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerInfo.java create mode 100644 presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerModule.java create mode 100644 presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreSubType.java diff --git a/hetu-metastore/src/test/java/io/hetu/core/metastore/TestHetuMetastoreGlobalCache.java b/hetu-metastore/src/test/java/io/hetu/core/metastore/TestHetuMetastoreGlobalCache.java index cb35113bc..3e16e7c9d 100644 --- a/hetu-metastore/src/test/java/io/hetu/core/metastore/TestHetuMetastoreGlobalCache.java +++ b/hetu-metastore/src/test/java/io/hetu/core/metastore/TestHetuMetastoreGlobalCache.java @@ -40,6 +40,7 @@ import io.prestosql.spi.metastore.model.TableEntity; import io.prestosql.spi.metastore.model.TableEntityType; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateStore; import io.prestosql.spi.statestore.StateStoreBootstrapper; import io.prestosql.spi.statestore.StateStoreFactory; @@ -134,7 +135,7 @@ public class TestHetuMetastoreGlobalCache seeds.add(mockSeed); SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class); - when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore); + when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore); when(mockSeed.getLocation()).thenReturn(LOCALHOST + ":" + PORT3); when(mockSeedStore.get()).thenReturn(seeds); diff --git a/hetu-seed-store/pom.xml b/hetu-seed-store/pom.xml index 107b8044f..05e0e12f9 100644 --- a/hetu-seed-store/pom.xml +++ b/hetu-seed-store/pom.xml @@ -42,6 +42,14 @@ ${project.version} test + + io.airlift + json + + + com.fasterxml.jackson.core + jackson-annotations + diff --git a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java index 6d4fd5805..7c249682e 100644 --- a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java @@ -15,6 +15,8 @@ package io.hetu.core.seedstore.filebased; +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; import io.hetu.core.common.util.SecureObjectInputStream; import io.prestosql.spi.seedstore.Seed; @@ -37,10 +39,13 @@ public class FileBasedSeed { private static final long serialVersionUID = 4L; - private final String location; - private final long timestamp; + private String location; + private long timestamp; - private FileBasedSeed(String location, long timestamp) + @JsonCreator + public FileBasedSeed( + @JsonProperty("location") String location, + @JsonProperty("timestamp") long timestamp) { this.location = location; this.timestamp = timestamp; @@ -66,17 +71,33 @@ public class FileBasedSeed } @Override + @JsonProperty public String getLocation() { return location; } @Override + @JsonProperty + public void setLocation(String location) + { + this.location = location; + } + + @Override + @JsonProperty public long getTimestamp() { return timestamp; } + @Override + @JsonProperty + public void setTimestamp(long timestamp) + { + this.timestamp = timestamp; + } + @Override public String serialize() throws IOException @@ -114,48 +135,4 @@ public class FileBasedSeed + "location='" + location + '\'' + ", timestamp=" + timestamp + '}'; } - - /** - * FileBasedSeedBuilder - * - * @since 2020-03-08 - */ - - public static class FileBasedSeedBuilder - { - private String location; - private long timestamp; - - /** - * constructor of FileBasedSeedBuilder - * - * @param location location(eg ip) of seed - */ - public FileBasedSeedBuilder(String location) - { - this.location = location; - } - - /** - * Set timestamp for FileBasedSeedBuilder - * - * @param timestamp timestamp of seed - * @return FileBasedSeedBuilder - */ - public FileBasedSeedBuilder setTimestamp(long timestamp) - { - this.timestamp = timestamp; - return this; - } - - /** - * build FileBasedSeed Object - * - * @return FileBasedSeed - */ - public FileBasedSeed build() - { - return new FileBasedSeed(location, timestamp); - } - } } diff --git a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java index 346d18462..7493f69fd 100644 --- a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java @@ -22,9 +22,12 @@ package io.hetu.core.seedstore.filebased; */ public class FileBasedSeedConstants { - // seed file name + // Hazelcast seed file name static final String SEED_FILE_NAME = "seeds.txt"; + // ON-YARN seed file name + static final String ON_YARN_SEED_FILE_NAME = "seeds.json"; + // dir config properties name static final String SEED_STORE_FILESYSTEM_DIR = "seed-store.filesystem.seed-dir"; diff --git a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStore.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStore.java index 2fd0b595f..6fbde7b35 100644 --- a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStore.java +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStore.java @@ -157,7 +157,7 @@ public class FileBasedSeedStore throw new NullPointerException("Cannot create filebased seed since location is null or empty"); } - return new FileBasedSeed.FileBasedSeedBuilder(location).setTimestamp(Long.parseLong(timestamp)).build(); + return new FileBasedSeed(location, Long.parseLong(timestamp)); } @Override diff --git a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreFactory.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreFactory.java index ff02c9364..2ba05f650 100644 --- a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreFactory.java +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreFactory.java @@ -18,6 +18,7 @@ package io.hetu.core.seedstore.filebased; import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.seedstore.SeedStore; import io.prestosql.spi.seedstore.SeedStoreFactory; +import io.prestosql.spi.seedstore.SeedStoreSubType; import java.util.Map; @@ -32,9 +33,14 @@ public class FileBasedSeedStoreFactory private static final String FACTORY_TYPE = "filebased"; @Override - public SeedStore create(String name, HetuFileSystemClient fs, Map config) + public SeedStore create(String name, SeedStoreSubType subType, HetuFileSystemClient fs, Map config) { - return new FileBasedSeedStore(name, fs, config); + if (subType == SeedStoreSubType.ON_YARN) { + return new FileBasedSeedStoreOnYarn(name, fs, config); + } + else { + return new FileBasedSeedStore(name, fs, config); + } } @Override diff --git a/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java new file mode 100644 index 000000000..425b26ae7 --- /dev/null +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java @@ -0,0 +1,217 @@ +/* + * Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. + * Licensed 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 io.hetu.core.seedstore.filebased; + +import com.google.common.collect.ImmutableList; +import io.airlift.json.JsonCodec; +import io.airlift.log.Logger; +import io.hetu.core.common.util.SecurePathWhiteList; +import io.prestosql.spi.filesystem.FileBasedLock; +import io.prestosql.spi.filesystem.HetuFileSystemClient; +import io.prestosql.spi.seedstore.Seed; +import io.prestosql.spi.seedstore.SeedStore; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.io.UncheckedIOException; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.Collection; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.locks.Lock; +import java.util.stream.Collectors; + +import static com.google.common.base.Preconditions.checkArgument; +import static io.airlift.json.JsonCodec.listJsonCodec; +import static java.nio.file.StandardOpenOption.CREATE_NEW; + +/** + * FileBasedOnYarnSeedStore + * + * @since 2021-06-23 + */ +public class FileBasedSeedStoreOnYarn + implements SeedStore +{ + private static final Logger LOG = Logger.get(FileBasedSeedStoreOnYarn.class); + private static final JsonCodec> LIST_FILE_BASED_SEED_CODEC = listJsonCodec(FileBasedSeed.class); + private HetuFileSystemClient fs; + private Map config; + private String name; // seed store name represents the sub-folder where a seed file is located + private Path seedFileFolder; // seed file folder + private Path seedFilePath; // seed file full path, = //seeds.json + + /** + * Constructor for file based seed store - on YARN + * + * @param name seed store name + * @param config seed store config + */ + public FileBasedSeedStoreOnYarn(String name, HetuFileSystemClient fs, Map config) + { + this.name = name; + this.fs = fs; + this.config = config; + + seedFileFolder = Paths.get(config.getOrDefault(FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR, FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR_DEFAULT_VALUE).trim()); + try { + checkArgument(!seedFileFolder.toString().contains("../"), + "SeedStore directory path must be absolute and at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); + checkArgument(SecurePathWhiteList.isSecurePath(seedFileFolder.toString()), + "SeedStore directory path must be at user workspace " + SecurePathWhiteList.getSecurePathWhiteList().toString()); + } + catch (IOException e) { + throw new IllegalArgumentException("Failed to get secure path list.", e); + } + seedFilePath = seedFileFolder.resolve(name).resolve(FileBasedSeedConstants.ON_YARN_SEED_FILE_NAME); + } + + @Override + public Set add(Collection seeds) + throws IOException + { + LOG.debug("FileBasedOnYarnSeedStore::add() invoked."); + Lock lock = new FileBasedLock(fs, seedFileFolder.resolve(name)); + try { + lock.lock(); + Set latestSeeds = new HashSet<>(); + // add all new seeds + latestSeeds.addAll( + seeds.stream() + .filter(s -> (s instanceof FileBasedSeed)) + .map(s -> (FileBasedSeed) s) + .collect(Collectors.toList())); + // load existing seeds and filter out repeated seed compared to new seeds + if (fs.exists(seedFilePath)) { + String json = loadFromFile(seedFilePath); + List existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json); + latestSeeds.addAll( + existingSeeds.stream() + .filter(s -> !latestSeeds.contains(s)) + .collect(Collectors.toList())); + } + String output = LIST_FILE_BASED_SEED_CODEC.toJson(ImmutableList.copyOf(latestSeeds)); + writeToFile(seedFilePath, output, true); + return new HashSet<>(latestSeeds); + } + catch (UncheckedIOException e) { + throw new IOException(e); + } + finally { + lock.unlock(); + } + } + + @Override + public Set get() + throws IOException + { + LOG.debug("FileBasedOnYarnSeedStore::get() invoked."); + try { + Set outputs = new HashSet<>(); + if (fs.exists(seedFilePath)) { + String json = loadFromFile(seedFilePath); + outputs.addAll(LIST_FILE_BASED_SEED_CODEC.fromJson(json)); + } + return outputs; + } + catch (UncheckedIOException e) { + throw new IOException(e); + } + } + + @Override + public Set remove(Collection seeds) + throws IOException + { + LOG.debug("FileBasedOnYarnSeedStore::remove() invoked."); + Lock lock = new FileBasedLock(fs, seedFileFolder.resolve(name)); + try { + lock.lock(); + Set outputs = new HashSet<>(); + if (fs.exists(seedFilePath)) { + String json = loadFromFile(seedFilePath); + List existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json); + Set latestSeeds = new HashSet<>(existingSeeds); + latestSeeds.removeAll( + seeds.stream() + .filter(s -> (s instanceof FileBasedSeed)) + .map(s -> (FileBasedSeed) s) + .collect(Collectors.toList())); + String output = LIST_FILE_BASED_SEED_CODEC.toJson(ImmutableList.copyOf(latestSeeds)); + writeToFile(seedFilePath, output, true); + outputs.addAll(latestSeeds); + } + return outputs; + } + catch (UncheckedIOException e) { + throw new IOException(e); + } + finally { + lock.unlock(); + } + } + + @Override + public Seed create(Map properties) + { + String location = properties.get(Seed.LOCATION_PROPERTY_NAME); + String timestamp = properties.get(Seed.TIMESTAMP_PROPERTY_NAME); + // add more properties if interfaces change + if (location == null || location.isEmpty()) { + throw new NullPointerException("Cannot create file-based seed since location is null or empty"); + } + + return new FileBasedSeed(location, Long.parseLong(timestamp)); + } + + @Override + public String getName() + { + return name; + } + + @Override + public void setName(String name) + { + checkArgument(name.matches("[\\p{Alnum}_\\-]+"), "Invalid cluster name"); + this.name = name; + this.seedFilePath = seedFileFolder.resolve(name).resolve(FileBasedSeedConstants.SEED_FILE_NAME); + } + + private void writeToFile(Path file, String content, boolean overwrite) + throws IOException + { + try (OutputStream os = (overwrite) ? fs.newOutputStream(file) : fs.newOutputStream(file, CREATE_NEW)) { + os.write(content.getBytes()); + } + } + + private String loadFromFile(Path file) + throws IOException + { + StringBuilder content = new StringBuilder(0); + try (BufferedReader br = new BufferedReader(new InputStreamReader(fs.newInputStream(file)))) { + br.lines().forEach(content::append); + } + return content.toString(); + } +} diff --git a/hetu-seed-store/src/test/java/io/hetu/core/seedstore/filebased/TestFileBasedSeedStore.java b/hetu-seed-store/src/test/java/io/hetu/core/seedstore/filebased/TestFileBasedSeedStore.java index 4835442c3..20b661519 100644 --- a/hetu-seed-store/src/test/java/io/hetu/core/seedstore/filebased/TestFileBasedSeedStore.java +++ b/hetu-seed-store/src/test/java/io/hetu/core/seedstore/filebased/TestFileBasedSeedStore.java @@ -18,6 +18,7 @@ package io.hetu.core.seedstore.filebased; import io.hetu.core.filesystem.HetuLocalFileSystemClient; import io.hetu.core.filesystem.LocalConfig; import io.prestosql.spi.seedstore.Seed; +import io.prestosql.spi.seedstore.SeedStoreSubType; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -56,7 +57,7 @@ public class TestFileBasedSeedStore Map config = new HashMap<>(0); config.put(FileBasedSeedConstants.SEED_STORE_FILESYSTEM_DIR, rootDir); filebasedSeedStoreFactory = new FileBasedSeedStoreFactory(); - seedStore = (FileBasedSeedStore) filebasedSeedStoreFactory.create("filebased", + seedStore = (FileBasedSeedStore) filebasedSeedStoreFactory.create("filebased", SeedStoreSubType.HAZELCAST, new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get(rootDir)), config); seedStore.setName(clusterName); } @@ -85,8 +86,8 @@ public class TestFileBasedSeedStore String ip2 = "10.0.0.2"; final long timestamp2 = 2000L; final int resultSize = 2; - Seed seed1 = new FileBasedSeed.FileBasedSeedBuilder(ip1).setTimestamp(timestamp1).build(); - Seed seed2 = new FileBasedSeed.FileBasedSeedBuilder(ip2).setTimestamp(timestamp2).build(); + Seed seed1 = new FileBasedSeed(ip1, timestamp1); + Seed seed2 = new FileBasedSeed(ip2, timestamp2); // add operation seedStore.add(Lists.newArrayList(seed1, seed2)); @@ -110,7 +111,7 @@ public class TestFileBasedSeedStore // overwrite operation final long updateTimestamp2 = 3000L; - Seed seed2Update = new FileBasedSeed.FileBasedSeedBuilder(ip2).setTimestamp(updateTimestamp2).build(); + Seed seed2Update = new FileBasedSeed(ip2, updateTimestamp2); seedStore.add(Lists.newArrayList(seed2Update)); results = seedStore.get(); assertEquals(results.size(), 1); diff --git a/hetu-state-store/pom.xml b/hetu-state-store/pom.xml index 01607e4a4..72ed81ee3 100644 --- a/hetu-state-store/pom.xml +++ b/hetu-state-store/pom.xml @@ -77,6 +77,10 @@ com.fasterxml.jackson.datatype jackson-datatype-jdk8 + + com.fasterxml.jackson.core + jackson-annotations + @@ -113,4 +117,4 @@ - \ No newline at end of file + diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java index f8251db5a..a471bf17b 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java @@ -14,6 +14,7 @@ */ package io.hetu.core.statestore.hazelcast; +import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.collect.ImmutableSet; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; @@ -106,7 +107,7 @@ public class TestHazelcastClusterLifecycleListener private static final long serialVersionUID = 4L; String location; - long timeStamp; + long timestamp; /** * Constructor for the mock seed @@ -116,21 +117,37 @@ public class TestHazelcastClusterLifecycleListener MockSeed(String location) { this.location = location; - timeStamp = 0L; + timestamp = 0L; } @Override + @JsonProperty public String getLocation() { return location; } @Override + @JsonProperty + public void setLocation(String location) + { + this.location = location; + } + + @Override + @JsonProperty public long getTimestamp() { return 0L; } + @Override + @JsonProperty + public void setTimestamp(long timestamp) + { + this.timestamp = timestamp; + } + @Override public String serialize() throws IOException @@ -138,9 +155,9 @@ public class TestHazelcastClusterLifecycleListener return "MOCK SEED. SHOULD NOT SERIALIZE."; } - public void setTimeStamp(long timeStamp) + public void setTimeStamp(long timestamp) { - this.timeStamp = timeStamp; + this.timestamp = timestamp; } } diff --git a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java index 30251c155..02bdc2d1a 100644 --- a/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java +++ b/hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastStateStoreFactory.java @@ -78,7 +78,7 @@ public class TestHazelcastStateStoreFactory properties.put(DISCOVERY_MODE_CONFIG_NAME, DISCOVERY_MODE_TCPIP); SeedStore seedStore = mock(SeedStore.class); - Seed seed = new FileBasedSeed.FileBasedSeedBuilder(MEMBER_ADDRESS).build(); + Seed seed = new FileBasedSeed(MEMBER_ADDRESS, 0); when(seedStore.get()).thenReturn(ImmutableList.of(seed)); setupHazelcastInstance(); diff --git a/presto-main/src/main/java/io/prestosql/discovery/HetuServiceInventory.java b/presto-main/src/main/java/io/prestosql/discovery/HetuServiceInventory.java index c666aaef6..7dd7a2968 100644 --- a/presto-main/src/main/java/io/prestosql/discovery/HetuServiceInventory.java +++ b/presto-main/src/main/java/io/prestosql/discovery/HetuServiceInventory.java @@ -26,8 +26,10 @@ import io.airlift.http.client.HttpClient; import io.airlift.json.JsonCodec; import io.airlift.log.Logger; import io.airlift.node.NodeInfo; +import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.server.InternalCommunicationConfig; import io.prestosql.spi.PrestoException; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateMap; import io.prestosql.spi.statestore.StateStore; import io.prestosql.statestore.StateStoreConstants; @@ -49,11 +51,13 @@ public class HetuServiceInventory private final HetuConfig hetuConfig; private final String nodeId; private final StateStoreProvider stateStoreProvider; + private final SeedStoreManager seedStoreManager; private final InternalCommunicationConfig internalCommunicationConfig; private AtomicBoolean serverUp = new AtomicBoolean(true); @Inject public HetuServiceInventory(HetuConfig hetuConfig, + SeedStoreManager seedStoreManager, StateStoreProvider stateStoreProvider, InternalCommunicationConfig internalCommunicationConfig, ServiceInventoryConfig config, @@ -64,6 +68,7 @@ public class HetuServiceInventory super(config, nodeInfo, serviceDescriptorsCodec, httpClient); this.nodeId = nodeInfo.getNodeId(); this.hetuConfig = hetuConfig; + this.seedStoreManager = seedStoreManager; this.stateStoreProvider = stateStoreProvider; this.internalCommunicationConfig = internalCommunicationConfig; } @@ -71,28 +76,58 @@ public class HetuServiceInventory @Override public Iterable getServiceDescriptors(String type) { - if ("discovery".equals(type) && hetuConfig.isMultipleCoordinatorEnabled()) { - try { - StateStore stateStore = stateStoreProvider.getStateStore(); + if ("discovery".equals(type)) { + if (hetuConfig.isMultipleCoordinatorEnabled()) { + try { + StateStore stateStore = stateStoreProvider.getStateStore(); - if (stateStore == null) { - throw new PrestoException(GENERIC_INTERNAL_ERROR, "State store has not been loaded yet"); - } + if (stateStore == null) { + throw new PrestoException(GENERIC_INTERNAL_ERROR, "State store has not been loaded yet"); + } - Map.Entry entry = ((StateMap) stateStore.getStateCollection(StateStoreConstants.DISCOVERY_SERVICE_COLLECTION_NAME)) - .getAll().entrySet().stream().findFirst().get(); + Map.Entry entry = ((StateMap) stateStore.getStateCollection(StateStoreConstants.DISCOVERY_SERVICE_COLLECTION_NAME)) + .getAll().entrySet().stream().findFirst().get(); - ImmutableMap.Builder properties = new ImmutableMap.Builder(); - if (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired()) { - properties.put("https", "https://" + entry.getKey() + ":" + entry.getValue()); + ImmutableMap.Builder properties = new ImmutableMap.Builder(); + if (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired()) { + properties.put("https", "https://" + entry.getKey() + ":" + entry.getValue()); + } + else { + properties.put("http", "http://" + entry.getKey() + ":" + entry.getValue()); + } + return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build())); } - else { - properties.put("http", "http://" + entry.getKey() + ":" + entry.getValue()); + catch (Exception e) { + logServerError("Select service from state store failed: " + e); + // do not return the service descriptors loaded from the configuration file, it might register this node to other cluster. + // return an empty list here. + return ImmutableList.of(); } - return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build())); } - catch (Exception e) { - logServerError("Select service from state store failed:" + e); + else if (seedStoreManager != null && seedStoreManager.isSeedStoreOnYarnEnabled()) { + try { + boolean httpsRequired = (internalCommunicationConfig != null && internalCommunicationConfig.isHttpsRequired()); + String location = seedStoreManager.getLatestSeedLocation(SeedStoreSubType.ON_YARN, httpsRequired); + + if (location == null || location.isEmpty()) { + throw new PrestoException(GENERIC_INTERNAL_ERROR, "Seed store has not been initialized yet"); + } + + ImmutableMap.Builder properties = new ImmutableMap.Builder(); + if (httpsRequired) { + properties.put("https", location); + } + else { + properties.put("http", location); + } + return ImmutableList.of(new ServiceDescriptor(UUID.randomUUID(), nodeId, "discovery", null, null, ServiceState.RUNNING, properties.build())); + } + catch (Exception e) { + logServerError("Select service from seed store failed: " + e); + // do not return the service descriptors loaded from the configuration file, it might register this node to other cluster. + // return an empty list here. + return ImmutableList.of(); + } } } return super.getServiceDescriptors(type); diff --git a/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerInfo.java b/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerInfo.java new file mode 100644 index 000000000..46817cc4f --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerInfo.java @@ -0,0 +1,222 @@ +/* + * Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. + * Licensed 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 io.prestosql.httpserver; + +import com.google.common.annotations.VisibleForTesting; +import io.airlift.http.server.HttpServerConfig; +import io.airlift.http.server.HttpServerInfo; +import io.airlift.log.Logger; +import io.airlift.node.NodeInfo; +import io.prestosql.spi.PrestoException; +import io.prestosql.spi.StandardErrorCode; + +import javax.inject.Inject; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.channels.ServerSocketChannel; + +public class HetuHttpServerInfo + extends HttpServerInfo +{ + private static final Logger LOG = Logger.get(HetuHttpServerInfo.class); + private URI httpUri; + private URI httpExternalUri; + private URI httpsUri; + private URI httpsExternalUri; + private URI adminUri; + private URI adminExternalUri; + private ServerSocketChannel httpChannel; + private ServerSocketChannel httpsChannel; + private ServerSocketChannel adminChannel; + + private static final int MIN_PORT_NUMBER = 1100; + private static final int MAX_PORT_NUMBER = 65535; + + @Inject + public HetuHttpServerInfo(HttpServerConfig config, NodeInfo nodeInfo) + { + super(new HttpServerConfig().setHttpEnabled(false).setHttpsEnabled(false).setAdminEnabled(false), nodeInfo); + // http channel + if (config.isHttpEnabled()) { + this.httpChannel = createChannel(config.getHttpPort(), config.getHttpAcceptQueueSize()); + if (config.getHttpPort() != port(this.httpChannel)) { + config.setHttpPort(port(this.httpChannel)); + } + this.httpUri = buildUri("http", nodeInfo.getInternalAddress(), port(this.httpChannel)); + this.httpExternalUri = buildUri("http", nodeInfo.getExternalAddress(), this.httpUri.getPort()); + LOG.info(String.format("HetuHttpServerInfo http channel bind to port %s", port(this.httpChannel))); + } + else { + httpChannel = null; + httpUri = null; + httpExternalUri = null; + } + setBaseChannel(HttpServerInfo.class, "httpChannel", httpChannel); + // https channel + if (config.isHttpsEnabled()) { + this.httpsChannel = createChannel(config.getHttpsPort(), config.getHttpAcceptQueueSize()); + if (config.getHttpsPort() != port(this.httpsChannel)) { + config.setHttpsPort(port(this.httpsChannel)); + } + this.httpsUri = buildUri("https", nodeInfo.getInternalAddress(), port(this.httpsChannel)); + this.httpsExternalUri = buildUri("https", nodeInfo.getExternalAddress(), this.httpsUri.getPort()); + LOG.info(String.format("HetuHttpServerInfo https channel bind to port %s", port(this.httpChannel))); + } + else { + httpsChannel = null; + httpsUri = null; + httpsExternalUri = null; + } + setBaseChannel(HttpServerInfo.class, "httpsChannel", httpsChannel); + // admin channel + if (config.isAdminEnabled()) { + this.adminChannel = createChannel(config.getAdminPort(), config.getHttpAcceptQueueSize()); + if (config.isHttpsEnabled()) { + this.adminUri = buildUri("https", nodeInfo.getInternalAddress(), port(this.adminChannel)); + this.adminExternalUri = buildUri("https", nodeInfo.getExternalAddress(), this.adminUri.getPort()); + } + else { + this.adminUri = buildUri("http", nodeInfo.getInternalAddress(), port(this.adminChannel)); + this.adminExternalUri = buildUri("http", nodeInfo.getExternalAddress(), this.adminUri.getPort()); + } + if (config.getAdminPort() != port(this.adminChannel)) { + config.setAdminPort(port(this.adminChannel)); + } + } + else { + adminChannel = null; + adminUri = null; + adminExternalUri = null; + } + setBaseChannel(HttpServerInfo.class, "adminChannel", adminChannel); + } + + private void setBaseChannel(Class clazz, String name, Object value) + { + Field channel = null; + try { + channel = clazz.getDeclaredField(name); + } + catch (NoSuchFieldException e) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, "No such channel found"); + } + boolean oldVal = channel.isAccessible(); + try { + channel.setAccessible(true); + channel.set(this, value); + } + catch (IllegalAccessException e) { + throw new PrestoException(StandardErrorCode.GENERIC_INTERNAL_ERROR, "Channel not accessible"); + } + finally { + channel.setAccessible(oldVal); + } + } + + public URI getHttpUri() + { + return this.httpUri; + } + + public URI getHttpExternalUri() + { + return this.httpExternalUri; + } + + public URI getHttpsUri() + { + return this.httpsUri; + } + + public URI getHttpsExternalUri() + { + return this.httpsExternalUri; + } + + public URI getAdminUri() + { + return this.adminUri; + } + + public URI getAdminExternalUri() + { + return this.adminExternalUri; + } + + ServerSocketChannel getHttpChannel() + { + return this.httpChannel; + } + + ServerSocketChannel getHttpsChannel() + { + return this.httpsChannel; + } + + ServerSocketChannel getAdminChannel() + { + return this.adminChannel; + } + + private static URI buildUri(String scheme, String host, int port) + { + try { + return new URI(scheme, (String) null, host, port, (String) null, (String) null, (String) null); + } + catch (URISyntaxException e) { + throw new IllegalArgumentException(e); + } + } + + @VisibleForTesting + static int port(ServerSocketChannel channel) + { + try { + return ((InetSocketAddress) channel.getLocalAddress()).getPort(); + } + catch (IOException e) { + throw new UncheckedIOException(e); + } + } + + private static ServerSocketChannel createChannel(int port, int acceptQueueSize) + { + ServerSocketChannel channel = null; + int nextPort = port; + while (true) { + try { + channel = ServerSocketChannel.open(); + channel.socket().setReuseAddress(true); + channel.socket().bind(new InetSocketAddress(nextPort), acceptQueueSize); + return channel; + } + catch (IOException e) { + // cannot bind to this port + nextPort++; + + if (nextPort < MIN_PORT_NUMBER || nextPort > MAX_PORT_NUMBER) { + LOG.warn("Failed to bind port withing acceptable limit"); + nextPort = MIN_PORT_NUMBER; + } + } + } + } +} diff --git a/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerModule.java b/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerModule.java new file mode 100644 index 000000000..b56fbc9c4 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/httpserver/HetuHttpServerModule.java @@ -0,0 +1,31 @@ +/* + * Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. + * Licensed 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 io.prestosql.httpserver; + +import com.google.inject.Binder; +import com.google.inject.Module; +import com.google.inject.Scopes; +import io.airlift.http.server.HttpServerInfo; + +public class HetuHttpServerModule + implements Module +{ + @Override + public void configure(Binder binder) + { + binder.bind(HttpServerInfo.class).to(HetuHttpServerInfo.class).in(Scopes.SINGLETON); + } +} diff --git a/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java b/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java index fb79ab664..c55437901 100644 --- a/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java +++ b/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java @@ -26,13 +26,16 @@ import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; import io.prestosql.spi.seedstore.SeedStoreFactory; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.statestore.StateStoreConstants; import java.io.File; import java.io.IOException; import java.nio.file.Paths; import java.security.InvalidParameterException; +import java.util.ArrayList; import java.util.Collection; +import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; import java.util.Map; @@ -49,6 +52,7 @@ import static io.prestosql.spi.StandardErrorCode.SEED_STORE_FAILURE; import static java.lang.String.format; import static java.util.Objects.requireNonNull; import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor; +import static java.util.stream.Collectors.toCollection; /** * SeedStoreManager manages the lifecycle of SeedStores @@ -58,31 +62,43 @@ import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor; public class SeedStoreManager { private static final Logger LOG = Logger.get(SeedStoreManager.class); - // properties name + + // configuration files private static final File SEED_STORE_CONFIGURATION = new File("etc/seed-store.properties"); private static final File STATE_STORE_CONFIGURATION = new File("etc/state-store.properties"); + // properties name + private static final String SEED_STORE_ON_YARN_PROPERTY_NAME = "seed-store.on-yarn"; private static final String SEED_STORE_TYPE_PROPERTY_NAME = "seed-store.type"; + private static final String SEED_STORE_CLUSTER_PROPERTY_NAME = "seed-store.cluster"; private static final String SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME = "seed-store.seed.heartbeat"; private static final String SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME = "seed-store.seed.heartbeat.timeout"; private static final String SEED_STORE_FILESYSTEM_PROFILE = "seed-store.filesystem.profile"; // properties default value + private static final boolean SEED_STORE_ON_YARN_ENABLED_DEFAULT_VALUE = false; private static final String SEED_STORE_TYPE_DEFAULT_VALUE = "filebased"; - private static final String SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE = "10000"; // 10 seconds - private static final String SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE = "60000"; // 60 seconds + private static final String SEED_STORE_CLUSTER_DEFAULT_VALUE = "olk_default"; + private static final long SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE = 10000; // 10 seconds + private static final long SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE = 60000; // 60 seconds private static final String SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE = "hdfs-config-default"; private static final int SEED_RETRY_TIMES = 5; private static final long SEED_RETRY_INTERVAL = 500L; private final Map seedStoreFactories = new ConcurrentHashMap<>(); private final FileSystemClientManager fileSystemClientManager; - private ScheduledExecutorService seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher")); - private SeedStore seedStore; - private String seedStoreType; - private String filesystemProfile; private HetuFileSystemClient fileSystemClient; + private ScheduledExecutorService seedRefreshExecutor = newSingleThreadScheduledExecutor(daemonThreadsNamed("SeedRefresher")); + + private String seedStoreType; + private String clusterName; + private String filesystemProfile; + private boolean isSeedStoreOnYarnEnabled; private long seedHeartBeat; private long seedHeartBeatTimeout; - private boolean isSeedStoreEnabled; + private boolean isSeedStoreHazelcastEnabled; + + private SeedStore seedStoreOnYarn; + private SeedStore seedStoreHazelcast; + private ConcurrentHashMap refreshableSeedsMap = new ConcurrentHashMap<>(); @Inject @@ -111,26 +127,90 @@ public class SeedStoreManager public void loadSeedStore() throws IOException { - // load configuration - Map config = loadConfiguration(STATE_STORE_CONFIGURATION, SEED_STORE_CONFIGURATION); + // initialize variables + isSeedStoreOnYarnEnabled = SEED_STORE_ON_YARN_ENABLED_DEFAULT_VALUE; + seedStoreType = SEED_STORE_TYPE_DEFAULT_VALUE; + clusterName = SEED_STORE_CLUSTER_DEFAULT_VALUE; + filesystemProfile = SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE; + seedHeartBeat = SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE; + seedHeartBeatTimeout = SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE; - if (isSeedStoreEnabled) { - LOG.info("-- Loading seed store --"); - // create seed store - SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType); - checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory); - try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) { - fileSystemClient = fileSystemClientManager.getFileSystemClient(filesystemProfile, Paths.get("/")); - seedStore = seedStoreFactory.create(seedStoreType, - fileSystemClient, - ImmutableMap.copyOf(config)); + Map config = new HashMap<>(); + // load seed store configuration + if (SEED_STORE_CONFIGURATION.exists()) { + Map seedStoreProperties = new HashMap<>(loadPropertiesFrom(SEED_STORE_CONFIGURATION.getPath())); + String propertyValue = seedStoreProperties.get(SEED_STORE_ON_YARN_PROPERTY_NAME); + if (propertyValue != null) { + isSeedStoreOnYarnEnabled = Boolean.parseBoolean(propertyValue); } - try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStore.getClass().getClassLoader())) { - // start seed refresher - seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeeds(), 0, seedHeartBeat, TimeUnit.MILLISECONDS); + seedStoreType = seedStoreProperties.getOrDefault(SEED_STORE_TYPE_PROPERTY_NAME, seedStoreType); + clusterName = seedStoreProperties.getOrDefault(SEED_STORE_CLUSTER_PROPERTY_NAME, clusterName); + filesystemProfile = seedStoreProperties.getOrDefault(SEED_STORE_FILESYSTEM_PROFILE, filesystemProfile); + propertyValue = seedStoreProperties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME); + if (propertyValue != null) { + seedHeartBeat = Long.parseLong(propertyValue); + } + propertyValue = seedStoreProperties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME); + if (propertyValue != null) { + seedHeartBeatTimeout = Long.parseLong(propertyValue); + } + if (seedHeartBeat > seedHeartBeatTimeout) { + throw new InvalidParameterException(format("The value of %s cannot be greater than the value of %s in the property file", + SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME, SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME)); + } + config.putAll(seedStoreProperties); + if (isSeedStoreOnYarnEnabled) { + // create seed store + SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType); + checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory); + if (fileSystemClient == null) { + fileSystemClient = fileSystemClientManager.getFileSystemClient(filesystemProfile, Paths.get("/")); + } + LOG.info("-- Loading seed store on-yarn --"); + try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) { + seedStoreOnYarn = seedStoreFactory.create(clusterName, + SeedStoreSubType.ON_YARN, + fileSystemClient, + ImmutableMap.copyOf(config)); + } } - LOG.info("-- Loaded seed store %s --", seedStoreType); } + // load state store config if exist + if (STATE_STORE_CONFIGURATION.exists()) { + Map stateStoreProperties = new HashMap<>(loadPropertiesFrom(STATE_STORE_CONFIGURATION.getPath())); + // for now, seed store is started only if tcp-ip mode enabled and tcp-ip.seeds is not set + String discoveryMode = stateStoreProperties.get(StateStoreConstants.DISCOVERY_MODE_PROPERTY_NAME); + isSeedStoreHazelcastEnabled = discoveryMode != null && discoveryMode.equals(StateStoreConstants.DISCOVERY_MODE_TCPIP) + && stateStoreProperties.get(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_SEEDS) == null; + if (isSeedStoreHazelcastEnabled) { + // create seed store + SeedStoreFactory seedStoreFactory = seedStoreFactories.get(seedStoreType); + checkState(seedStoreFactory != null, "SeedStoreFactory %s is not registered", seedStoreFactory); + if (fileSystemClient == null) { + String stateStoreFilesystemProfile = stateStoreProperties.getOrDefault(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_PROFILE, filesystemProfile); + fileSystemClient = fileSystemClientManager.getFileSystemClient(stateStoreFilesystemProfile, Paths.get("/")); + } + LOG.info("-- Loading seed store hazelcast --"); + try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreFactory.getClass().getClassLoader())) { + seedStoreHazelcast = seedStoreFactory.create(clusterName, + SeedStoreSubType.HAZELCAST, + fileSystemClient, + ImmutableMap.copyOf(config)); + } + try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(seedStoreHazelcast.getClass().getClassLoader())) { + // start seed refresher + seedRefreshExecutor.scheduleWithFixedDelay(() -> refreshSeeds(SeedStoreSubType.HAZELCAST), 0, seedHeartBeat, TimeUnit.MILLISECONDS); + } + } + } + } + + /** + * Check if seed-store on-YARN is enabled + */ + public boolean isSeedStoreOnYarnEnabled() + { + return isSeedStoreOnYarnEnabled; } /** @@ -148,9 +228,10 @@ public class SeedStoreManager * @return a collection of seeds in the seed store * @throws IOException */ - public Collection getAllSeeds() + public Collection getAllSeeds(SeedStoreSubType subType) throws IOException { + SeedStore seedStore = getSeedStore(subType); if (seedStore == null) { throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } @@ -160,6 +241,43 @@ public class SeedStoreManager return seeds; } + /** + * Get all seeds from seed store + * + * @return the latest seed location in the seed store - https or http + * @throws IOException + */ + public String getLatestSeedLocation(SeedStoreSubType subType, boolean httpsRequired) + throws IOException + { + SeedStore seedStore = getSeedStore(subType); + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } + + Collection seeds = seedStore.get(); + if (seeds.isEmpty()) { + return null; + } + ArrayList list = seeds.stream() + .filter(seed -> { + if (httpsRequired) { + return seed.getLocation().contains("https:"); + } + else { + return seed.getLocation().contains("http:"); + } + }) + .sorted(Comparator.comparing(Seed::getTimestamp)) + .collect(toCollection(ArrayList::new)); + if (list.isEmpty()) { + return null; + } + else { + return list.get(list.size() - 1).getLocation(); + } + } + /** * Add seed to seed store. If refreshable is enabled, seed will be refreshed periodically * @@ -168,11 +286,12 @@ public class SeedStoreManager * @throws IOException */ - public Collection addSeed(String seedLocation, boolean refreshable) + public Collection addSeed(SeedStoreSubType subType, String seedLocation, boolean refreshable) throws IOException { Collection seeds = new HashSet<>(); + SeedStore seedStore = getSeedStore(subType); if (seedStore == null) { throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } @@ -180,7 +299,7 @@ public class SeedStoreManager Seed seed = seedStore.create(ImmutableMap.of( Seed.LOCATION_PROPERTY_NAME, seedLocation, Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis()))); - seeds = addSeed(seed); + seeds = addSeed(subType, seed); if (refreshable) { refreshableSeedsMap.put(seedLocation, seed); @@ -198,11 +317,12 @@ public class SeedStoreManager * @throws IOException */ - public Collection removeSeed(String seedLocation) + public Collection removeSeed(SeedStoreSubType subType, String seedLocation) throws IOException { Collection seeds = new HashSet<>(); + SeedStore seedStore = getSeedStore(subType); if (seedStore == null) { throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } @@ -223,9 +343,10 @@ public class SeedStoreManager * * @throws IOException */ - public void clearExpiredSeeds() + public void clearExpiredSeeds(SeedStoreSubType subType) throws IOException { + SeedStore seedStore = getSeedStore(subType); if (seedStore == null) { throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); } @@ -244,13 +365,17 @@ public class SeedStoreManager } } - private Collection addSeed(Seed seed) + private Collection addSeed(SeedStoreSubType subType, Seed seed) throws IOException { int retryTimes = 0; long retryInterval = 0L; Collection seeds = null; + SeedStore seedStore = getSeedStore(subType); + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } do { try { TimeUnit.MILLISECONDS.sleep(retryInterval); @@ -280,9 +405,14 @@ public class SeedStoreManager * * Set the local SeedStore * */ - public void setSeedStore(SeedStore seedStore) + public void setSeedStore(SeedStoreSubType subType, SeedStore seedStore) { - this.seedStore = seedStore; + if (subType == SeedStoreSubType.ON_YARN) { + seedStoreOnYarn = seedStore; + } + else { + seedStoreHazelcast = seedStore; + } } /** @@ -290,13 +420,18 @@ public class SeedStoreManager * * @return loaded SeedStore or null if it's not loaded */ - public SeedStore getSeedStore() + public SeedStore getSeedStore(SeedStoreSubType subType) { + SeedStore seedStore = (subType == SeedStoreSubType.ON_YARN) ? seedStoreOnYarn : seedStoreHazelcast; return seedStore; } - private void refreshSeeds() + private void refreshSeeds(SeedStoreSubType subType) { + SeedStore seedStore = getSeedStore(subType); + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } for (Map.Entry entry : refreshableSeedsMap.entrySet()) { long newTime = System.currentTimeMillis(); LOG.debug("seed=%s refresh with oldTimestamp=%s and newTimestamp=%s", entry.getKey(), entry.getValue().getTimestamp(), newTime); @@ -314,45 +449,4 @@ public class SeedStoreManager } } } - - private Map loadConfiguration(File stateStoreConfig, File seedStoreConfig) - throws IOException - { - Map properties = new HashMap<>(); - - // initialize variables - seedStoreType = SEED_STORE_TYPE_DEFAULT_VALUE; - filesystemProfile = SEED_STORE_FILESYSTEM_PROFILE_DEFAULT_VALUE; - seedHeartBeat = Long.parseLong(SEED_STORE_SEED_HEARTBEAT_DEFAULT_VALUE); - seedHeartBeatTimeout = Long.parseLong(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_DEFAULT_VALUE); - - // load state store config if exist - if (stateStoreConfig.exists()) { - Map stateStoreProperties = new HashMap<>(loadPropertiesFrom(stateStoreConfig.getPath())); - filesystemProfile = stateStoreProperties.getOrDefault(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_PROFILE, filesystemProfile); - // for now, seed store is started only if tcp-ip mode enabled and tcp-ip.seeds is not set - String discoveryMode = stateStoreProperties.get(StateStoreConstants.DISCOVERY_MODE_PROPERTY_NAME); - isSeedStoreEnabled = discoveryMode != null && discoveryMode.equals(StateStoreConstants.DISCOVERY_MODE_TCPIP) - && stateStoreProperties.get(StateStoreConstants.HAZELCAST_DISCOVERY_TCPIP_SEEDS) == null; - } - - // load seed store config if exist - if (seedStoreConfig.exists()) { - Map seedStoreProperties = new HashMap<>(loadPropertiesFrom(seedStoreConfig.getPath())); - properties.putAll(seedStoreProperties); - seedStoreType = properties.getOrDefault(SEED_STORE_TYPE_PROPERTY_NAME, seedStoreType); - filesystemProfile = properties.getOrDefault(SEED_STORE_FILESYSTEM_PROFILE, filesystemProfile); - if (properties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME) != null) { - seedHeartBeat = Long.parseLong(properties.get(SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME)); - } - if (properties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME) != null) { - seedHeartBeatTimeout = Long.parseLong(properties.get(SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME)); - } - if (seedHeartBeat > seedHeartBeatTimeout) { - throw new InvalidParameterException(format("The value of %s cannot be greater than the value of %s in the property file", - SEED_STORE_SEED_HEARTBEAT_PROPERTY_NAME, SEED_STORE_SEED_HEARTBEAT_TIMEOUT_PROPERTY_NAME)); - } - } - return properties; - } } diff --git a/presto-main/src/main/java/io/prestosql/server/PrestoServer.java b/presto-main/src/main/java/io/prestosql/server/PrestoServer.java index c6f4e9ac9..4f517b9bd 100755 --- a/presto-main/src/main/java/io/prestosql/server/PrestoServer.java +++ b/presto-main/src/main/java/io/prestosql/server/PrestoServer.java @@ -22,6 +22,7 @@ import io.airlift.discovery.client.Announcer; import io.airlift.discovery.client.DiscoveryModule; import io.airlift.event.client.JsonEventModule; import io.airlift.event.client.http.HttpEventModule; +import io.airlift.http.server.HttpServerInfo; import io.airlift.http.server.HttpServerModule; import io.airlift.jaxrs.JaxrsModule; import io.airlift.jmx.JmxHttpModule; @@ -43,6 +44,8 @@ import io.prestosql.execution.scheduler.NodeSchedulerConfig; import io.prestosql.execution.warnings.WarningCollectorModule; import io.prestosql.filesystem.FileSystemClientManager; import io.prestosql.heuristicindex.HeuristicIndexerManager; +import io.prestosql.httpserver.HetuHttpServerInfo; +import io.prestosql.httpserver.HetuHttpServerModule; import io.prestosql.jmx.HetuJmxModule; import io.prestosql.metadata.StaticCatalogStore; import io.prestosql.metadata.StaticFunctionNamespaceStore; @@ -55,6 +58,7 @@ import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.server.security.PasswordAuthenticatorManager; import io.prestosql.server.security.ServerSecurityModule; import io.prestosql.snapshot.SnapshotUtils; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.sql.parser.SqlParserOptions; import io.prestosql.statestore.StateStoreLauncher; import io.prestosql.statestore.StateStoreProvider; @@ -106,7 +110,7 @@ public class PrestoServer modules.add( new NodeModule(), Modules.override(new DiscoveryModule()).with(new HetuDiscoveryModule()), - new HttpServerModule(), + Modules.override(new HttpServerModule()).with(new HetuHttpServerModule()), new JsonModule(), new SmileModule(), new JaxrsModule(), @@ -156,6 +160,13 @@ public class PrestoServer injector.getInstance(PasswordAuthenticatorManager.class).loadPasswordAuthenticator(); injector.getInstance(EventListenerManager.class).loadConfiguredEventListener(); + if (injector.getInstance(SeedStoreManager.class).isSeedStoreOnYarnEnabled()) { + addSeedOnYarnInformation( + injector.getInstance(ServerConfig.class), + injector.getInstance(SeedStoreManager.class), + (HetuHttpServerInfo) injector.getInstance(HttpServerInfo.class)); + } + // preload index (on coordinator only) if (injector.getInstance(ServerConfig.class).isCoordinator()) { injector.getInstance(HeuristicIndexerManager.class).preloadIndex(); @@ -187,6 +198,34 @@ public class PrestoServer return ImmutableList.of(); } + private static void addSeedOnYarnInformation(ServerConfig serverConfig, + SeedStoreManager seedStoreManager, + HetuHttpServerInfo httpServerInfo) + { + if (serverConfig == null || seedStoreManager == null || httpServerInfo == null) { + return; + } + if (!serverConfig.isCoordinator()) { + return; + } + String httpUri; + if (httpServerInfo.getHttpExternalUri() != null) { + httpUri = httpServerInfo.getHttpExternalUri().toString(); + } + else if (httpServerInfo.getHttpsExternalUri() != null) { + httpUri = httpServerInfo.getHttpsExternalUri().toString(); + } + else { + return; + } + try { + seedStoreManager.addSeed(SeedStoreSubType.ON_YARN, httpUri, false); + } + catch (IOException e) { + return; + } + } + private static void launchEmbeddedStateStore(HetuConfig config, StateStoreLauncher launcher) throws Exception { diff --git a/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java b/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java index 2bbf7a437..9dbf5729f 100644 --- a/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java +++ b/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java @@ -22,12 +22,17 @@ import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.server.InternalCommunicationConfig; import io.prestosql.spi.PrestoException; import io.prestosql.spi.filesystem.FileBasedLock; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateCollection; import io.prestosql.spi.statestore.StateMap; import io.prestosql.spi.statestore.StateStore; import io.prestosql.spi.statestore.StateStoreBootstrapper; import io.prestosql.utils.HetuConfig; +import javax.net.ServerSocketFactory; +import javax.net.ssl.SSLServerSocket; +import javax.net.ssl.SSLServerSocketFactory; + import java.io.File; import java.io.IOException; import java.net.URI; @@ -111,11 +116,11 @@ public class EmbeddedStateStoreLauncher if (staticSeeds.size() > 0) { launchStateStore(staticSeeds, properties); } - else if (seedStoreManager.getSeedStore() != null) { + else if (seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST) != null) { // Set seed store name - seedStoreManager.getSeedStore().setName(properties.get(STATE_STORE_CLUSTER_PROPERTY_NAME)); + seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST).setName(properties.get(STATE_STORE_CLUSTER_PROPERTY_NAME)); // Clear expired seeds - seedStoreManager.clearExpiredSeeds(); + seedStoreManager.clearExpiredSeeds(SeedStoreSubType.HAZELCAST); // Use lock to control synchronization of state store launch among all coordinators Lock launcherLock = new FileBasedLock(seedStoreManager.getFileSystemClient(), Paths.get(LAUNCHER_LOCK_FILE_PATH)); try { @@ -142,18 +147,42 @@ public class EmbeddedStateStoreLauncher private void launchStateStoreFromSeedStore(Map properties) throws IOException { // Get all seeds - Set locations = seedStoreManager.getAllSeeds() + Set locations = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST) .stream() .map(x -> x.getLocation()) .collect(Collectors.toSet()); String launcherPort = getStateStoreLauncherPort(properties); requireNonNull(launcherPort, "The launcher port is null"); + // Detect port conflict and get the next usable one + launcherPort = checkAndGetAvailablePort(launcherPort); + properties.put(HAZELCAST_DISCOVERY_PORT_PROPERTY_NAME, launcherPort); // Launch state store String currentLocation = getNodeUri().getHost() + ":" + launcherPort; locations.add(currentLocation); if (launchStateStore(locations, properties) != null) { // Add seed to seed store if and only if state store launched successfully - seedStoreManager.addSeed(currentLocation, true); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, currentLocation, true); + } + } + + private String checkAndGetAvailablePort(String port) + { + int nextPort = Integer.parseInt(port); + while (!isPortAvailable(nextPort)) { + nextPort++; + } + + return String.valueOf(nextPort); + } + + private static boolean isPortAvailable(int port) + { + ServerSocketFactory sslServerSocketFactory = SSLServerSocketFactory.getDefault(); + try (SSLServerSocket socket = (SSLServerSocket) sslServerSocketFactory.createServerSocket(port)) { + return true; + } + catch (IOException e) { + return false; } } @@ -277,9 +306,9 @@ public class EmbeddedStateStoreLauncher registerDiscoveryService(failureMemberHost); } - if (seedStoreManager.getSeedStore() != null) { + if (seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST) != null) { try { - seedStoreManager.removeSeed((String) failureMember); + seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, (String) failureMember); } catch (Exception e) { LOG.error("Cannot remove failure node %s from seed store: %s", failureMember, e.getMessage()); diff --git a/presto-main/src/main/java/io/prestosql/statestore/LocalStateStoreProvider.java b/presto-main/src/main/java/io/prestosql/statestore/LocalStateStoreProvider.java index ace2db59e..677b5cc51 100644 --- a/presto-main/src/main/java/io/prestosql/statestore/LocalStateStoreProvider.java +++ b/presto-main/src/main/java/io/prestosql/statestore/LocalStateStoreProvider.java @@ -21,6 +21,7 @@ import io.prestosql.metastore.MetaStoreConstants; import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.spi.PrestoException; import io.prestosql.spi.classloader.ThreadContextClassLoader; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateCollection; import io.prestosql.spi.statestore.StateStore; import io.prestosql.spi.statestore.StateStoreFactory; @@ -100,7 +101,7 @@ public class LocalStateStoreProvider stateStoreName = DEFAULT_STATE_STORE_NAME; } // Create state stores defined in config - stateStore = stateStoreFactory.create(stateStoreName, seedStoreManager.getSeedStore(), ImmutableMap.copyOf(properties)); + stateStore = stateStoreFactory.create(stateStoreName, seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST), ImmutableMap.copyOf(properties)); stateStore.registerClusterFailureHandler(this::handleClusterDisconnection); stateStore.init(); } diff --git a/presto-main/src/test/java/io/prestosql/discovery/TestHetuServiceInventory.java b/presto-main/src/test/java/io/prestosql/discovery/TestHetuServiceInventory.java index 16e45fdc0..243716742 100644 --- a/presto-main/src/test/java/io/prestosql/discovery/TestHetuServiceInventory.java +++ b/presto-main/src/test/java/io/prestosql/discovery/TestHetuServiceInventory.java @@ -59,6 +59,7 @@ public class TestHetuServiceInventory hetuConfig.setMultipleCoordinatorEnabled(true); HetuServiceInventory inventory = new HetuServiceInventory(hetuConfig, + null, createMockStateStoreProvider(), internalCommunicationConfig, new ServiceInventoryConfig(), @@ -86,6 +87,7 @@ public class TestHetuServiceInventory hetuConfig.setMultipleCoordinatorEnabled(true); HetuServiceInventory inventory = new HetuServiceInventory(hetuConfig, + null, createMockStateStoreProvider(), internalCommunicationConfig, new ServiceInventoryConfig(), diff --git a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java index c232901ba..f16485947 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestDynamicFilterSourceOperator.java @@ -35,6 +35,7 @@ import io.prestosql.spi.plan.PlanNodeId; import io.prestosql.spi.plan.Symbol; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateSet; import io.prestosql.spi.statestore.StateStore; import io.prestosql.spi.statestore.StateStoreBootstrapper; @@ -140,7 +141,7 @@ public class TestDynamicFilterSourceOperator seeds.add(mockSeed); SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class); - when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore); + when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore); when(mockSeed.getLocation()).thenReturn("127.0.0.1:6991"); when(mockSeedStore.get()).thenReturn(seeds); diff --git a/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java b/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java index 6ca818885..c629c8ade 100644 --- a/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java +++ b/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java @@ -14,12 +14,14 @@ */ package io.prestosql.seedstore; +import com.fasterxml.jackson.annotation.JsonProperty; import com.google.common.collect.ImmutableSet; import io.prestosql.filesystem.FileSystemClientManager; import io.prestosql.spi.filesystem.HetuFileSystemClient; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; import io.prestosql.spi.seedstore.SeedStoreFactory; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.testing.assertions.Assert; import org.testng.annotations.BeforeMethod; import org.testng.annotations.BeforeTest; @@ -90,6 +92,7 @@ public class TestSeedStoreManager SeedStoreFactory mockSeedStoreFactory = mock(SeedStoreFactory.class); when(mockSeedStoreFactory.getName()).thenReturn("filebased"); when(mockSeedStoreFactory.create(any(String.class), + SeedStoreSubType.HAZELCAST, any(HetuFileSystemClient.class), any(Map.class))).thenReturn(mockSeedStore); seedStoreManager.addSeedStoreFactory(mockSeedStoreFactory); @@ -100,7 +103,7 @@ public class TestSeedStoreManager throws IOException { seedStoreManager.loadSeedStore(); - SeedStore returned = seedStoreManager.getSeedStore(); + SeedStore returned = seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST); } @Test @@ -110,9 +113,9 @@ public class TestSeedStoreManager String location1 = "location1"; String location2 = "location2"; seedStoreManager.loadSeedStore(); - seedStoreManager.addSeed(location1, false); - seedStoreManager.addSeed(location2, true); - Collection result = seedStoreManager.getAllSeeds(); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true); + Collection result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST); Assert.assertEquals(result.size(), 2); Assert.assertTrue(result.stream().anyMatch(s -> s.getLocation().equals(location1))); Assert.assertTrue(result.stream().anyMatch(s -> s.getLocation().equals(location2))); @@ -121,7 +124,7 @@ public class TestSeedStoreManager long timestamp2Old = result.stream().filter(s -> s.getLocation().equals(location2)).findAny().get().getTimestamp(); //wait 2 seconds, seed2 will be updated with new timestamp Thread.sleep(2000); - result = seedStoreManager.getAllSeeds(); + result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST); long timestamp1New = result.stream().filter(s -> s.getLocation().equals(location1)).findAny().get().getTimestamp(); long timestamp2New = result.stream().filter(s -> s.getLocation().equals(location2)).findAny().get().getTimestamp(); Assert.assertTrue(timestamp1Old == timestamp1New); @@ -136,12 +139,12 @@ public class TestSeedStoreManager String location2 = "location2"; String location3 = "location3"; seedStoreManager.loadSeedStore(); - seedStoreManager.addSeed(location1, false); - seedStoreManager.addSeed(location2, true); - Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2); - seedStoreManager.removeSeed(location1); - seedStoreManager.removeSeed(location3); - Collection result = seedStoreManager.getAllSeeds(); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true); + Assert.assertEquals(seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST).size(), 2); + seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, location1); + seedStoreManager.removeSeed(SeedStoreSubType.HAZELCAST, location3); + Collection result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST); Assert.assertEquals(result.size(), 1); Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent()); } @@ -153,13 +156,13 @@ public class TestSeedStoreManager String location1 = "location1"; String location2 = "location2"; seedStoreManager.loadSeedStore(); - seedStoreManager.addSeed(location1, false); - seedStoreManager.addSeed(location2, true); - Assert.assertEquals(seedStoreManager.getAllSeeds().size(), 2); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location1, false); + seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, location2, true); + Assert.assertEquals(seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST).size(), 2); //wait 5 seconds, location1 expired since refreshable is not enabled Thread.sleep(5000); - seedStoreManager.clearExpiredSeeds(); - Collection result = seedStoreManager.getAllSeeds(); + seedStoreManager.clearExpiredSeeds(SeedStoreSubType.HAZELCAST); + Collection result = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST); Assert.assertEquals(result.size(), 1); Assert.assertFalse(result.stream().filter(s -> s.getLocation().equals(location1)).findAny().isPresent()); } @@ -188,7 +191,7 @@ public class TestSeedStoreManager }}; createConfigFile("etc/state-store.properties", stateStoreConfigs); seedStoreManager.loadSeedStore(); - Assert.assertNull(seedStoreManager.getSeedStore()); + Assert.assertNull(seedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)); // reset config files prepareConfigFiles(); } @@ -231,17 +234,33 @@ public class TestSeedStoreManager } @Override + @JsonProperty public String getLocation() { return location; } @Override + @JsonProperty + public void setLocation(String location) + { + this.location = location; + } + + @Override + @JsonProperty public long getTimestamp() { return timestamp; } + @Override + @JsonProperty + public void setTimestamp(long timestamp) + { + this.timestamp = timestamp; + } + @Override public String serialize() throws IOException @@ -249,11 +268,6 @@ public class TestSeedStoreManager return "MOCK SEED. SHOULD NOT SERIALIZE."; } - public void setTimestamp(long timestamp) - { - this.timestamp = timestamp; - } - @Override public boolean equals(Object obj) { diff --git a/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java b/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java index 378c8196f..1ee2aafb1 100644 --- a/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java +++ b/presto-main/src/test/java/io/prestosql/statestore/TestStateStoreLauncherAndProvider.java @@ -26,6 +26,7 @@ import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.server.InternalCommunicationConfig; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateStore; import io.prestosql.spi.statestore.StateStoreBootstrapper; import io.prestosql.spi.statestore.StateStoreFactory; @@ -109,7 +110,7 @@ public class TestStateStoreLauncherAndProvider seeds.add(mockSeed); SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class); - when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore); + when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore); when(mockSeed.getLocation()).thenReturn(LOCALHOST + ":" + PORT3); when(mockSeedStore.get()).thenReturn(seeds); @@ -136,8 +137,8 @@ public class TestStateStoreLauncherAndProvider when(mockSeedStore.get()).thenReturn(seeds); SeedStoreManager mockSeedStoreManager = mock(SeedStoreManager.class); - when(mockSeedStoreManager.getSeedStore()).thenReturn(mockSeedStore); - when(mockSeedStoreManager.addSeed(LOCALHOST, true)).thenReturn(seeds); + when(mockSeedStoreManager.getSeedStore(SeedStoreSubType.HAZELCAST)).thenReturn(mockSeedStore); + when(mockSeedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, LOCALHOST, true)).thenReturn(seeds); when(mockSeedStoreManager.getFileSystemClient()).thenReturn(new HetuLocalFileSystemClient(new LocalConfig(new Properties()), Paths.get("/"))); InternalCommunicationConfig mockInternalCommunicationConfig = mock(InternalCommunicationConfig.class); @@ -157,7 +158,7 @@ public class TestStateStoreLauncherAndProvider // mock "remove" second instance from cluster (delete from seed store) seeds.remove(mockSeed2); when(mockSeed1.getLocation()).thenReturn(LOCALHOST + ":" + PORT1); - when(mockSeedStoreManager.addSeed(LOCALHOST, true)).thenReturn(seeds); + when(mockSeedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, LOCALHOST, true)).thenReturn(seeds); ((HazelcastStateStore) second).shutdown(); // Allow the first node to handle failure diff --git a/presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java b/presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java index 6feb388bb..507a5ee18 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java +++ b/presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java @@ -42,6 +42,13 @@ public interface Seed */ String getLocation(); + /** + * St location of seed + * + * @return void + */ + void setLocation(String location); + /** * Get timestamp of seed * @@ -49,6 +56,13 @@ public interface Seed */ long getTimestamp(); + /** + * Set timestamp of seed + * + * @return void + */ + void setTimestamp(long timestamp); + /** * Serialize seed object to string and return * diff --git a/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreFactory.java b/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreFactory.java index 8c9c8bb0d..b1571a230 100644 --- a/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreFactory.java +++ b/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreFactory.java @@ -40,5 +40,5 @@ public interface SeedStoreFactory * @param config seed store configurations * @return created seed store */ - SeedStore create(String name, HetuFileSystemClient fs, Map config); + SeedStore create(String name, SeedStoreSubType subType, HetuFileSystemClient fs, Map config); } diff --git a/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreSubType.java b/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreSubType.java new file mode 100644 index 000000000..8bcadf2af --- /dev/null +++ b/presto-spi/src/main/java/io/prestosql/spi/seedstore/SeedStoreSubType.java @@ -0,0 +1,21 @@ +/* + * Copyright (C) 2018-2020. Huawei Technologies Co., Ltd. All rights reserved. + * Licensed 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 io.prestosql.spi.seedstore; + +public enum SeedStoreSubType { + DEFAULT, + ON_YARN, + HAZELCAST +} diff --git a/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java b/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java index 376bf0d1f..2526ecd7d 100644 --- a/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java +++ b/presto-tests/src/main/java/io/prestosql/tests/DistributedQueryRunnerWithStateStore.java @@ -28,6 +28,7 @@ import io.prestosql.seedstore.SeedStoreManager; import io.prestosql.server.testing.TestingPrestoServer; import io.prestosql.spi.seedstore.Seed; import io.prestosql.spi.seedstore.SeedStore; +import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateStore; import io.prestosql.sql.parser.SqlParserOptions; import io.prestosql.statestore.EmbeddedStateStoreLauncher; @@ -107,7 +108,7 @@ public class DistributedQueryRunnerWithStateStore launcher.launchStateStore(); StateStoreProvider provider = server.getInstance(Key.get(StateStoreProvider.class)); - Seed seed = new FileBasedSeed.FileBasedSeedBuilder("127.0.0.1:" + port).build(); + Seed seed = new FileBasedSeed("127.0.0.1:" + port, 0); SeedStore seedStore = new SeedStore() { @Override @@ -148,7 +149,7 @@ public class DistributedQueryRunnerWithStateStore { } }; - server.getInstance(Key.get(SeedStoreManager.class)).setSeedStore(seedStore); + server.getInstance(Key.get(SeedStoreManager.class)).setSeedStore(SeedStoreSubType.HAZELCAST, seedStore); if (provider instanceof LocalStateStoreProvider) { Map stateStoreProperties = new HashMap<>(); stateStoreProperties.putIfAbsent(HazelcastConstants.DISCOVERY_MODE_CONFIG_NAME, HazelcastConstants.DISCOVERY_MODE_TCPIP);