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 36480bfd0..c4abe740d 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 @@ -98,6 +98,12 @@ public class FileBasedSeed this.timestamp = timestamp; } + @Override + public String getUniqueInstanceId() + { + return "none"; + } + @Override public String serialize() throws IOException 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 982ee501e..be6d3fb27 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 @@ -26,7 +26,7 @@ public class FileBasedSeedConstants static final String SEED_FILE_NAME = "seeds.txt"; // ON-YARN seed file name - static final String ON_YARN_SEED_FILE_NAME = "seeds.json"; + static final String ON_YARN_SEED_FILE_NAME = "seeds-resources.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/FileBasedSeedOnYarn.java b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedOnYarn.java new file mode 100644 index 000000000..76145b295 --- /dev/null +++ b/hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedOnYarn.java @@ -0,0 +1,163 @@ +/* + * Copyright (C) 2018-2021. 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.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.hetu.core.common.util.SecureObjectInputStream; +import io.prestosql.spi.seedstore.Seed; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.util.Base64; +import java.util.Objects; + +/** + * FileBasedSeedOnYarn is used for storing seed store on yarn information + * + * @since 2020-03-08 + */ + +public class FileBasedSeedOnYarn + implements Seed +{ + private static final long serialVersionUID = 4L; + + // External URI (for instance, http://ip:8080) + private String location; + // Timestamp for this seed + private long timestamp; + // Hazelcast state store URI (for instance, ip:5701) + private String internalStateStoreUri; + + @JsonCreator + public FileBasedSeedOnYarn( + @JsonProperty("location") String location, + @JsonProperty("timestamp") long timestamp, + @JsonProperty("internal-state-store-uri") String internalStateStoreUri) + { + this.location = location; + this.timestamp = timestamp; + this.internalStateStoreUri = internalStateStoreUri; + } + + /** + * Deserialize FileBasedSeedOnyarn serialized string to FileBasedSeedOnyarn Object + * + * @param serialized FileBasedSeedOnyarn serialized String to be converted + * @return FileBasedSeedOnyarn Object + * @throws IOException if an I/O error occurs while reading string + * @throws ClassNotFoundException if serialized string cannot be deserialized to FileBasedSeedOnyarn Object + */ + public static FileBasedSeedOnYarn deserialize(String serialized) + throws IOException, ClassNotFoundException + { + byte[] datas = Base64.getDecoder().decode(serialized); + try (ObjectInputStream ois = new SecureObjectInputStream(new ByteArrayInputStream(datas), + FileBasedSeedOnYarn.class.getName())) { + FileBasedSeedOnYarn obj = (FileBasedSeedOnYarn) ois.readObject(); + return obj; + } + } + + @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; + } + + @JsonProperty + public String getInternalStateStoreUri() + { + return internalStateStoreUri; + } + + @Override + public String getUniqueInstanceId() + { + return this.internalStateStoreUri; + } + + @JsonProperty + public void setInternalStateStoreUri(String internalStateStoreUri) + { + this.internalStateStoreUri = internalStateStoreUri; + } + + @Override + public String serialize() + throws IOException + { + try (ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ObjectOutputStream oos = new ObjectOutputStream(baos)) { + oos.writeObject(this); + return Base64.getEncoder().encodeToString(baos.toByteArray()); + } + } + + @Override + public boolean equals(Object obj) + { + if (this == obj) { + return true; + } + if (obj == null || getClass() != obj.getClass()) { + return false; + } + FileBasedSeedOnYarn fileBasedSeed = (FileBasedSeedOnYarn) obj; + return location.equals(fileBasedSeed.location) && internalStateStoreUri.equals(fileBasedSeed.internalStateStoreUri); + } + + @Override + public int hashCode() + { + return Objects.hash(location); + } + + @Override + public String toString() + { + return "FileBasedSeedOnyarn{" + + "location='" + location + '\'' + + ", timestamp=" + timestamp + '\'' + + ", internalStateStoreUri='" + internalStateStoreUri + "\'}"; + } +} 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 index f1c083425..09b3b6a2c 100644 --- 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 @@ -52,12 +52,12 @@ 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 static final JsonCodec> LIST_FILE_BASED_SEED_CODEC = listJsonCodec(FileBasedSeedOnYarn.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 + private Path seedFilePath; // seed file full path, = //seeds-resources.json /** * Constructor for file based seed store - on YARN @@ -104,17 +104,17 @@ public class FileBasedSeedStoreOnYarn Lock lock = new FileBasedLock(fs, lockPath); try { lock.lock(); - Set latestSeeds = new HashSet<>(); + Set latestSeeds = new HashSet<>(); // add all new seeds latestSeeds.addAll( seeds.stream() - .filter(s -> (s instanceof FileBasedSeed)) - .map(s -> (FileBasedSeed) s) + .filter(s -> (s instanceof FileBasedSeedOnYarn)) + .map(s -> (FileBasedSeedOnYarn) 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); + List existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json); latestSeeds.addAll( existingSeeds.stream() .filter(s -> !latestSeeds.contains(s)) @@ -166,12 +166,12 @@ public class FileBasedSeedStoreOnYarn 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); + 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) + .filter(s -> (s instanceof FileBasedSeedOnYarn)) + .map(s -> (FileBasedSeedOnYarn) s) .collect(Collectors.toList())); String output = LIST_FILE_BASED_SEED_CODEC.toJson(ImmutableList.copyOf(latestSeeds)); writeToFile(seedFilePath, output, true); @@ -193,11 +193,13 @@ public class FileBasedSeedStoreOnYarn String location = properties.get(Seed.LOCATION_PROPERTY_NAME); String timestamp = properties.get(Seed.TIMESTAMP_PROPERTY_NAME); // add more properties if interfaces change + String internalStateStoreUri = properties.keySet().contains(Seed.INTERNAL_STATE_STORE_URI_PROPERTY_NAME) ? + properties.get(FileBasedSeedOnYarn.INTERNAL_STATE_STORE_URI_PROPERTY_NAME) : "none"; 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)); + return new FileBasedSeedOnYarn(location, Long.parseLong(timestamp), internalStateStoreUri); } @Override 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 d285b500b..75ef53b95 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 @@ -148,6 +148,12 @@ public class TestHazelcastClusterLifecycleListener this.timestamp = timestamp; } + @Override + public String getUniqueInstanceId() + { + return "none"; + } + @Override public String serialize() throws IOException 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 fb50f63fe..2f4011036 100644 --- a/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java +++ b/presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java @@ -15,6 +15,7 @@ package io.prestosql.seedstore; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Lists; import com.google.inject.Inject; @@ -38,6 +39,7 @@ import java.util.Collection; import java.util.Comparator; import java.util.HashMap; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; @@ -328,7 +330,18 @@ public class SeedStoreManager } refreshableSeedsMap.remove(seedLocation); - Optional seedOptional = seedStore.get().stream().filter(s -> s.getLocation().equals(seedLocation)).findFirst(); + Optional seedOptional; + if (subType == SeedStoreSubType.ON_YARN) { + seedOptional = seedStore.get() + .stream() + .filter(s -> { + return s.getLocation().equals(seedLocation) || s.getUniqueInstanceId().equals(seedLocation); + }) + .findFirst(); + } + else { + seedOptional = seedStore.get().stream().filter(s -> s.getLocation().equals(seedLocation)).findFirst(); + } if (seedOptional.isPresent()) { seeds = seedStore.remove(Lists.newArrayList(seedOptional.get())); @@ -365,6 +378,38 @@ public class SeedStoreManager } } + /** + * Update the existing seed at seedLocation with new properties + * @param subType + * @param seedLocation + * @param updatedProperties + */ + public void updateSeed(SeedStoreSubType subType, String seedLocation, Map updatedProperties) + { + SeedStore seedStore = getSeedStore(subType); + if (seedStore == null) { + throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null"); + } + + try { + Collection existingSeeds = seedStore.get(); + List toUpdate = existingSeeds.stream() + .filter(s -> { + return s.getLocation().equals(seedLocation); + }) + .collect(Collectors.toList()); + LOG.debug("SeedStoreManager::updateSeed toUpdate.size() is %s", Integer.toString(toUpdate.size())); + for (Seed s : toUpdate) { + seedStore.remove(ImmutableList.of(s)); + Seed newSeed = seedStore.create(updatedProperties); + addSeed(subType, newSeed); + } + } + catch (IOException e) { + LOG.warn("Update seed %s failed with error: %s", seedLocation, e.getMessage()); + } + } + private Collection addSeed(SeedStoreSubType subType, Seed seed) throws IOException { 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 3e516a234..7d63ad3c8 100755 --- a/presto-main/src/main/java/io/prestosql/server/PrestoServer.java +++ b/presto-main/src/main/java/io/prestosql/server/PrestoServer.java @@ -146,6 +146,12 @@ public class PrestoServer fileSystemClientManager.loadFactoryConfigs(); injector.getInstance(SeedStoreManager.class).loadSeedStore(); + if (injector.getInstance(SeedStoreManager.class).isSeedStoreOnYarnEnabled()) { + addSeedOnYarnInformation( + injector.getInstance(ServerConfig.class), + injector.getInstance(SeedStoreManager.class), + (HetuHttpServerInfo) injector.getInstance(HttpServerInfo.class)); + } launchEmbeddedStateStore(injector.getInstance(HetuConfig.class), injector.getInstance(StateStoreLauncher.class)); injector.getInstance(StateStoreProvider.class).loadStateStore(); injector.getInstance(HetuMetaStoreManager.class).loadHetuMetastore(fileSystemClientManager); // relies on state-store @@ -162,13 +168,6 @@ public class PrestoServer injector.getInstance(EventListenerManager.class).loadConfiguredEventListener(); injector.getInstance(GroupProviderManager.class).loadConfiguredGroupProvider(); - 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()) { HeuristicIndexerManager heuristicIndexerManager = injector.getInstance(HeuristicIndexerManager.class); 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 f34721fb0..3c1420b41 100644 --- a/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java +++ b/presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java @@ -15,6 +15,7 @@ package io.prestosql.statestore; import com.google.common.annotations.VisibleForTesting; +import com.google.common.collect.ImmutableMap; import com.google.inject.Inject; import io.airlift.http.server.HttpServerInfo; import io.airlift.log.Logger; @@ -22,6 +23,7 @@ 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.Seed; import io.prestosql.spi.seedstore.SeedStoreSubType; import io.prestosql.spi.statestore.StateCollection; import io.prestosql.spi.statestore.StateMap; @@ -146,6 +148,7 @@ public class EmbeddedStateStoreLauncher private void launchStateStoreFromSeedStore(Map properties) throws IOException { + URI externalUri = httpServerInfo.getHttpExternalUri() != null ? httpServerInfo.getHttpExternalUri() : httpServerInfo.getHttpsExternalUri(); // Get all seeds Set locations = seedStoreManager.getAllSeeds(SeedStoreSubType.HAZELCAST) .stream() @@ -163,6 +166,14 @@ public class EmbeddedStateStoreLauncher // Add seed to seed store if and only if state store launched successfully seedStoreManager.addSeed(SeedStoreSubType.HAZELCAST, currentLocation, true); } + // Also add this hazelcast state store uri to the on-yarn seedstore + if (seedStoreManager.getSeedStore(SeedStoreSubType.ON_YARN) != null) { + Map seedProperties = ImmutableMap.of( + Seed.LOCATION_PROPERTY_NAME, externalUri.toString(), + Seed.TIMESTAMP_PROPERTY_NAME, String.valueOf(System.currentTimeMillis()), + Seed.INTERNAL_STATE_STORE_URI_PROPERTY_NAME, currentLocation); + seedStoreManager.updateSeed(SeedStoreSubType.ON_YARN, externalUri.toString(), seedProperties); + } } private String checkAndGetAvailablePort(String port) @@ -300,6 +311,7 @@ public class EmbeddedStateStoreLauncher private void handleNodeFailure(Object failureMember) { + LOG.debug("EmbeddedStateStoreLauncher::handleNodeFailure invoked on %s", (String) failureMember); if (hetuConfig.isMultipleCoordinatorEnabled()) { // failureMember format host:port String failureMemberHost = ((String) failureMember).split(":")[0]; @@ -314,6 +326,15 @@ public class EmbeddedStateStoreLauncher LOG.error("Cannot remove failure node %s from seed store: %s", failureMember, e.getMessage()); } } + + if (seedStoreManager.getSeedStore(SeedStoreSubType.ON_YARN) != null) { + try { + seedStoreManager.removeSeed(SeedStoreSubType.ON_YARN, (String) failureMember); + } + catch (Exception e) { + LOG.error("Cannot remove failure node %s from seeds-resources.json: %s", (String) failureMember, e.getMessage()); + } + } } private URI getNodeUri() 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 83573b9a1..8a7a6f0d0 100644 --- a/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java +++ b/presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java @@ -261,6 +261,12 @@ public class TestSeedStoreManager this.timestamp = timestamp; } + @Override + public String getUniqueInstanceId() + { + return "none"; + } + @Override public String serialize() throws IOException 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 d5572fa34..2afaa9d9e 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 @@ -35,6 +35,11 @@ public interface Seed */ String TIMESTAMP_PROPERTY_NAME = "timestamp"; + /** + * Internal state store URI (currently only used in FileBasedSeedOnYarn) + */ + String INTERNAL_STATE_STORE_URI_PROPERTY_NAME = "internal-state-store-uri"; + /** * Get location of seed * @@ -63,6 +68,13 @@ public interface Seed */ void setTimestamp(long timestamp); + /** + * Get attribute of seed + * @param key + * @return value of attribute + */ + String getUniqueInstanceId(); + /** * Serialize seed object to string and return *