jffwang
/
hetu-core
0
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

!1254 Remove seeds from on-yarn seedstore file 

Merge pull request !1254 from lilianyuan_c78e/master
This commit is contained in:
i-robot 2021-12-01 19:44:06 +00:00 committed by Gitee
parent 5cde9e95a9 c1262c18e8
commit 1235ff63ef
10 changed files with 280 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeed.java
View File

@ -98,6 +98,12 @@ public class FileBasedSeed
this.timestamp = timestamp;
}
@Override
public String getUniqueInstanceId()
{
return "none";
}
@Override
public String serialize()
throws IOException

2
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedConstants.java
View File

@ -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";

163
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedOnYarn.java Normal file
View File

@ -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 + "\'}";
}
}

24
hetu-seed-store/src/main/java/io/hetu/core/seedstore/filebased/FileBasedSeedStoreOnYarn.java
View File

@ -52,12 +52,12 @@ public class FileBasedSeedStoreOnYarn
implements SeedStore
{
private static final Logger LOG = Logger.get(FileBasedSeedStoreOnYarn.class);
private static final JsonCodec<List<FileBasedSeed>> LIST_FILE_BASED_SEED_CODEC = listJsonCodec(FileBasedSeed.class);
private static final JsonCodec<List<FileBasedSeedOnYarn>> LIST_FILE_BASED_SEED_CODEC = listJsonCodec(FileBasedSeedOnYarn.class);
private HetuFileSystemClient fs;
private Map<String, String> 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, = <seedFileFolder>/<name>/seeds.json
private Path seedFilePath; // seed file full path, = <seedFileFolder>/<name>/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<FileBasedSeed> latestSeeds = new HashSet<>();
Set<FileBasedSeedOnYarn> 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<FileBasedSeed> existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json);
List<FileBasedSeedOnYarn> 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<Seed> outputs = new HashSet<>();
if (fs.exists(seedFilePath)) {
String json = loadFromFile(seedFilePath);
List<FileBasedSeed> existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json);
Set<FileBasedSeed> latestSeeds = new HashSet<>(existingSeeds);
List<FileBasedSeedOnYarn> existingSeeds = LIST_FILE_BASED_SEED_CODEC.fromJson(json);
Set<FileBasedSeedOnYarn> 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

6
hetu-state-store/src/test/java/io/hetu/core/statestore/hazelcast/TestHazelcastClusterLifecycleListener.java
View File

@ -148,6 +148,12 @@ public class TestHazelcastClusterLifecycleListener
this.timestamp = timestamp;
}
@Override
public String getUniqueInstanceId()
{
return "none";
}
@Override
public String serialize()
throws IOException

47
presto-main/src/main/java/io/prestosql/seedstore/SeedStoreManager.java
View File

@ -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<Seed> seedOptional = seedStore.get().stream().filter(s -> s.getLocation().equals(seedLocation)).findFirst();
Optional<Seed> 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<String, String> updatedProperties)
{
SeedStore seedStore = getSeedStore(subType);
if (seedStore == null) {
throw new PrestoException(SEED_STORE_FAILURE, "Seed store is null");
}
try {
Collection<Seed> existingSeeds = seedStore.get();
List<Seed> 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<Seed> addSeed(SeedStoreSubType subType, Seed seed)
throws IOException
{

13
presto-main/src/main/java/io/prestosql/server/PrestoServer.java
View File

@ -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);

21
presto-main/src/main/java/io/prestosql/statestore/EmbeddedStateStoreLauncher.java
View File

@ -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<String, String> properties) throws IOException
{
URI externalUri = httpServerInfo.getHttpExternalUri() != null ? httpServerInfo.getHttpExternalUri() : httpServerInfo.getHttpsExternalUri();
// Get all seeds
Set<String> 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<String, String> 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()

6
presto-main/src/test/java/io/prestosql/seedstore/TestSeedStoreManager.java
View File

@ -261,6 +261,12 @@ public class TestSeedStoreManager
this.timestamp = timestamp;
}
@Override
public String getUniqueInstanceId()
{
return "none";
}
@Override
public String serialize()
throws IOException

12
presto-spi/src/main/java/io/prestosql/spi/seedstore/Seed.java
View File

@ -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
*