From a3446685c26924f1c2f362af69ebb3e252eb429b Mon Sep 17 00:00:00 2001 From: Ahana Date: Thu, 19 May 2022 11:14:19 +0530 Subject: [PATCH] codecheck Gossip Protocol for Failure Detection --- hetu-docs/en/admin/properties.md | 62 +++ .../CoordinatorGossipFailureDetector.java | 358 ++++++++++++++++++ ...oordinatorGossipFailureDetectorModule.java | 50 +++ .../failuredetector/GossipProtocolConfig.java | 86 +++++ .../HeartbeatFailureDetector.java | 88 +++-- .../failuredetector/WhomToGossipInfo.java | 101 +++++ .../WorkerGossipFailureDetector.java | 141 +++++++ .../WorkerGossipFailureDetectorModule.java | 50 +++ .../prestosql/server/CoordinatorModule.java | 26 +- .../prestosql/server/GossipNodeResource.java | 31 ++ .../server/GossipStatusResource.java | 57 +++ .../io/prestosql/server/NodeResource.java | 5 + .../io/prestosql/server/ServerConfig.java | 24 ++ .../io/prestosql/server/ServerMainModule.java | 6 +- .../io/prestosql/server/WorkerModule.java | 27 +- .../TestCoordinatorGossipFailureDetector.java | 168 ++++++++ .../failuredetector/TestWhomToGossipInfo.java | 181 +++++++++ .../TestWorkerGossipFailureDetector.java | 48 +++ .../io/prestosql/server/TestServerConfig.java | 9 +- 19 files changed, 1483 insertions(+), 35 deletions(-) create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetector.java create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetectorModule.java create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/GossipProtocolConfig.java create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/WhomToGossipInfo.java create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetector.java create mode 100644 presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetectorModule.java create mode 100644 presto-main/src/main/java/io/prestosql/server/GossipNodeResource.java create mode 100644 presto-main/src/main/java/io/prestosql/server/GossipStatusResource.java create mode 100644 presto-main/src/test/java/io/prestosql/failuredetector/TestCoordinatorGossipFailureDetector.java create mode 100644 presto-main/src/test/java/io/prestosql/failuredetector/TestWhomToGossipInfo.java create mode 100644 presto-main/src/test/java/io/prestosql/failuredetector/TestWorkerGossipFailureDetector.java diff --git a/hetu-docs/en/admin/properties.md b/hetu-docs/en/admin/properties.md index 5b925ebaa..54fd876c7 100644 --- a/hetu-docs/en/admin/properties.md +++ b/hetu-docs/en/admin/properties.md @@ -377,6 +377,8 @@ Exchanges transfer data between openLooKeng nodes for different stages of a quer ## Failure Recovery handling Properties +### Failure Retry Policies + ### `failure.recovery.retry.profile` > - **Type:** `String` @@ -420,6 +422,66 @@ Exchanges transfer data between openLooKeng nodes for different stages of a quer > This property is used only for `max-retry` based failure detection profiles. > The minimum value for this parameter is 100. +### Gossip Protocol Configurations for Failure Detection + +### `failure-detection-protocol` + +>- **Type:** String +>- **Default value:** `heartbeat` +> +> This property defines the type of failure detector in use. Default configuration is `heartbeat` failure detector. +> Gossip protocol can be enabled by specifying this parameter in `config.properties` file, with the value `gossip`. +> All nodes (i.e. coordinator as well as workers) in a cluster should have this property specified in their respective `etc/config.properties` file. + +### `failure-detector.heartbeat-interval` + +>- **Type:** Duration +>- **Default value:** `500ms` (500 miliseconds) +> +> This is one of the existing configuration properties, which is used by the gossip protocol. +> This is the interval of gossip between two nodes in the cluster. +> In gossip protocol, two workers are expected to gossip with higher frequency than the coordinator and a worker. +> In `config.properties` for the coordinator, this property can be set with a reasonably higher value, such as `5s` (5 seconds). +> In workers, this property can be left to use the default value. +> +### `failure-detector.gossip.worker-gossip-probe-interval` +> +> - **Type:** Duration +>- **Default value:** `5s` (5 seconds) +> +> Gossip protocol uses monitoring tasks (same as the heartbeat failure detector) to keep tab on the other nodes. +> This property specifies the interval of refreshing the monitoring tasks to trigger worker to worker gossip. +> This property, if needed to be configured with any other value than the default, should be specified only for the worker nodes. +> This parameter should have higher value than `failure-detector.heartbeat-interval`. +> +### `failure-detector.gossip.coordinator-gossip-probe-interval` +> +> - **Type:** Duration +>- **Default value:** `5s` (5 seconds) +> +> Gossip protocol uses monitoring tasks (same as the heartbeat failure detector) to keep tab on the other nodes. +> This property specifies the interval of refreshing the monitoring tasks to trigger coordinator to worker gossip. +> This property, if needed to be configured with any other value than the default, should be specified only for the coordinator. +> This parameter should have higher value than `failure-detector.heartbeat-interval` and `failure-detector.gossip.worker-gossip-probe-interval`. +> +### `failure-detector.gossip.coordinator-gossip-collate-interval` +> +> - **Type:** Duration +>- **Default value:** `5s` (2 seconds) +> +> This property specifies the interval in which the coordinator collates all the gossips it obtained from all the workers. +> This property has to be specified only for the coordinator. +> This parameter should have higher value than `failure-detector.heartbeat-interval`. +> +### `failure-detector.gossip.group-size` +> +> - **Type:** Integer +>- **Default value:** `Integer.MAX_VALUE` +> +> A worker should gossip with how many other workers in the cluster, is defined by this parameter. +> Any value higher than the cluster-size (i.e. the number of workers) implies all-to-all gossip. +> To keep the network overhead low, this value should be reasonably low for a big cluster (e.g. 10 for a cluster size of 100). +> On each refresh of the worker-monitoring tasks at the coordinator, the coordinator defines the list of worker URIs of size `failure-detector.gossip.group-size` to trigger worker-to-worker gossip. ## Task Properties diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetector.java b/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetector.java new file mode 100644 index 000000000..92211373c --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetector.java @@ -0,0 +1,358 @@ +/* + * 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.prestosql.failuredetector; + +import com.google.common.io.ByteStreams; +import io.airlift.discovery.client.ServiceDescriptor; +import io.airlift.discovery.client.ServiceSelector; +import io.airlift.discovery.client.ServiceType; +import io.airlift.http.client.HttpClient; +import io.airlift.http.client.Request; +import io.airlift.http.client.Response; +import io.airlift.http.client.ResponseHandler; +import io.airlift.http.client.StaticBodyGenerator; +import io.airlift.json.JsonCodec; +import io.airlift.log.Logger; +import io.airlift.node.NodeInfo; +import io.airlift.units.Duration; +import io.prestosql.server.GossipStatusResource; +import io.prestosql.server.InternalCommunicationConfig; +import io.prestosql.spi.HostAddress; + +import javax.annotation.PostConstruct; +import javax.annotation.concurrent.ThreadSafe; +import javax.inject.Inject; + +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.BitSet; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import static io.airlift.concurrent.Threads.daemonThreadsNamed; +import static io.airlift.http.client.HttpUriBuilder.uriBuilderFrom; +import static io.airlift.http.client.JsonBodyGenerator.jsonBodyGenerator; +import static io.airlift.http.client.Request.Builder.prepareGet; +import static io.airlift.http.client.Request.Builder.preparePost; +import static io.prestosql.failuredetector.FailureDetector.State.ALIVE; +import static io.prestosql.protocol.RequestHelpers.setContentTypeHeaders; +import static io.prestosql.spi.HostAddress.fromUri; + +public class CoordinatorGossipFailureDetector + extends HeartbeatFailureDetector + implements FailureDetector +{ + private static final Logger log = Logger.get(CoordinatorGossipFailureDetector.class); + private static int uniq; // initialization to zero by default for int + private final Map gossipTales = new HashMap<>(); + private final Set tasks = new HashSet<>(); + private final Duration gossipValidityPeriod; + private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, daemonThreadsNamed("gossip-failure-detector")); + private final JsonCodec whomToGossipInfoCodec; + private final int gossipGroupSize; + private static final int START = 0; + private static final int END = 1; + + @Inject + public CoordinatorGossipFailureDetector(@ServiceType("presto") ServiceSelector selector, + @ForFailureDetector HttpClient httpClient, + FailureDetectorConfig failureDetectorConfig, + NodeInfo nodeInfo, + InternalCommunicationConfig internalCommunicationConfig, + GossipProtocolConfig config, + JsonCodec whomToGossipInfoCodec) + { + this(selector, httpClient, failureDetectorConfig, nodeInfo, + internalCommunicationConfig, config.getCnWkUpdateGossipMonitorServiceInterval(), + config.getCnGossipCollateInterval(), config.getGossipGroupSize(), whomToGossipInfoCodec); + } + + private CoordinatorGossipFailureDetector(@ServiceType("presto") ServiceSelector selector, + @ForFailureDetector HttpClient httpClient, + FailureDetectorConfig failureDetectorConfig, + NodeInfo nodeInfo, + InternalCommunicationConfig internalCommunicationConfig, + Duration monitoringServiceUpdateInterval, + Duration cnGossipCollateInterval, + int gossipGroupSize, + JsonCodec whomToGossipInfoCodec) + { + super(selector, httpClient, failureDetectorConfig, nodeInfo, internalCommunicationConfig); + this.monitoringServiceUpdateInterval = monitoringServiceUpdateInterval; + this.gossipValidityPeriod = cnGossipCollateInterval; + this.whomToGossipInfoCodec = whomToGossipInfoCodec; + this.gossipGroupSize = gossipGroupSize; + } + + public JsonCodec getWhomToGossipInfoCodec() + { + return whomToGossipInfoCodec; + } + + @Override + protected MonitoringTask createNewTask(ServiceDescriptor service, URI uri) + { + log.debug("creating new gossip monitoring task for uri " + uri); + GossipMonitoringTask task = new GossipMonitoringTask(service, uri, uniq++); + tasks.add(task); + return task; + } + + public List getSortedOnlineServiceDescriptors(Set online) + { + List uris = new ArrayList<>(); + List onlineServices = new ArrayList<>(online); + Collections.sort(onlineServices, new Comparator() { + @Override + public int compare(ServiceDescriptor t2, ServiceDescriptor t1) + { + return t1.getNodeId().compareTo(t1.getNodeId()); + } + }); + onlineServices.stream().forEach(o -> uris.add(getHttpUri(o))); + return uris; + } + + private int[] assignLocalGossipGroups(int idx, int clusterSize) + { + int[] startAndEndIdx = new int[2]; + if (this.gossipGroupSize >= clusterSize) { + startAndEndIdx[END] = clusterSize; // START is already 0 by initialization + } + else { + int start = idx * this.gossipGroupSize % clusterSize; + int end = (start + this.gossipGroupSize - 1) % clusterSize; + if (start < end) { + startAndEndIdx[START] = start; + startAndEndIdx[END] = end + 1; + } + else { + startAndEndIdx[START] = end; + startAndEndIdx[END] = start + 1; + } + } + log.debug("In " + idx + " rotation, start idx " + + startAndEndIdx[START] + ", end idx " + + startAndEndIdx[END] + " for total " + + clusterSize + " workers"); + return startAndEndIdx; + } + + private static BitSet readResponseBitSet(Response response) + { + try { + byte[] bytes = ByteStreams.toByteArray(response.getInputStream()); + return BitSet.valueOf(new byte[] {bytes[0]}); + } + catch (IOException e) { + log.error("error reading response input stream"); + } + return new BitSet(); + } + + @PostConstruct + @Override + public void start() + { + initGossipTales(); + super.start(); + + executor.scheduleWithFixedDelay(new Runnable() + { + @Override + public void run() + { + try { + forgetGossip(); + } + catch (Throwable e) { + log.warn(e, "Error removing stale gossip entries"); + } + } + }, 0, ((Double) gossipValidityPeriod.getValue()).longValue(), gossipValidityPeriod.getUnit()); + } + + private void initGossipTales() + { + Set online = getOnlineServiceDescriptors(); + online.forEach(o -> gossipTales.put(o.getNodeId(), 0L)); + } + + @Override + public State getState(HostAddress hostAddress) + { + State state = super.getState(hostAddress); + MonitoringTask task = tasks.stream() + .filter(t -> hostAddress.equals(fromUri(t.getUri()))).findFirst().orElse(null); + if (state != ALIVE && task != null + && gossipTales.containsKey(task.getService().getNodeId())) { + log.debug("node cannot be connected, but gossip is, it is alive!"); + state = ALIVE; + } + return state; + } + + /** + * The following method removes stale gossip info. + * Gossip validity period (default is 2 seconds). + * Any entry older than 2 seconds are removed. + * If no gossip has been received for a node for more than 2 seconds, should be considered GONE/UNRESPONSIVE + */ + private synchronized void forgetGossip() + { + Set> staleGossip = gossipTales.entrySet().stream() + .filter(e -> (System.nanoTime() - e.getValue() > gossipValidityPeriod.convertTo(TimeUnit.NANOSECONDS).getValue())) + .collect(Collectors.toSet()); + log.debug("Stale gossip count " + staleGossip.size()); + staleGossip.forEach(s -> gossipTales.remove(s.getKey())); + } + + private StaticBodyGenerator createBodyGenerator(WhomToGossipInfo info) + { + return jsonBodyGenerator(whomToGossipInfoCodec, info); + } + + @ThreadSafe + private class GossipMonitoringTask + extends HeartbeatFailureDetector.MonitoringTask + { + private BitSet gossip; + private WhomToGossipInfo whomToGossipInfo; + private final int idx; + + public GossipMonitoringTask(ServiceDescriptor service, URI uri, int idx) + { + super(service, uri); + this.idx = idx; + } + + private Set getAliveNodes() + { + Set alive = new HashSet<>(); + for (int i = 0; i < whomToGossipInfo.getUriList().size(); i++) { + if (gossip.get(i)) { + URI uri = whomToGossipInfo.getUriList().get(i); + alive.add(tasks.stream().filter(t -> t.getUri().equals(uri)).findFirst().get().getService().getNodeId()); + } + } + log.debug(alive.size() + "other workers alive"); + return alive; + } + + /** + * this method updates the gossip table. + * Each entry is identified by the node id, and its value is a timestamp, when CN got to know it alive. + * Ideally, each node should send their local timestamp. Due to payload concerns, it was not approached. + */ + private synchronized void doGossip() + { + Set alive = getAliveNodes(); + long currentTime = System.nanoTime(); // this is CN system time. So, later enties always have larger timestamp + alive.forEach(a -> gossipTales.put(a, currentTime)); + gossipTales.put(getService().getNodeId(), currentTime); // put itself + } + + private void postInitGossipUriList() + throws Exception + { + Set online = getOnlineServiceDescriptors(); + int[] startAndEnd = assignLocalGossipGroups(this.idx, online.size()); + List uris = getSortedOnlineServiceDescriptors(online).subList(startAndEnd[START], startAndEnd[END]); + + this.whomToGossipInfo = new WhomToGossipInfo(); + for (URI uri : uris) { + if (!uri.equals(this.getUri())) { + this.whomToGossipInfo.add(uri); + log.debug("add " + uri + " to gossip for worker " + getUri()); + } + } + + log.debug("CN ping to Wk " + getUri() + "to start gossip "); + URI reqUri = uriBuilderFrom(getUri()).appendPath(GossipStatusResource.GOSSIP_STATUS).build(); + Request request = setContentTypeHeaders(false, preparePost()).setUri(reqUri) + .setBodyGenerator(createBodyGenerator(this.whomToGossipInfo)).build(); + getHttpClient().execute(request, new ResponseHandler() { + @Override + public Object handleException(Request request, Exception exception) + { + log.debug("initial ping exception"); + return null; + } + + @Override + public Object handle(Request request, Response response) + { + return null; + } + }); + } + + @Override + public synchronized void enable() + { + try { + postInitGossipUriList(); + } + catch (Exception e) { + log.error("error while init gossip"); + } + super.enable(); + } + + @Override + protected void ping() + { + try { + getStats().recordStart(); + log.debug("ping ... " + getUri()); + URI wUri = uriBuilderFrom(getUri()).appendPath(GossipStatusResource.GOSSIP_STATUS).build(); + Request request = prepareGet().setUri(wUri).build(); + + getHttpClient().executeAsync(request, + new ResponseHandler() + { + @Override + public Exception handleException(Request request, Exception exception) + { + getStats().recordFailure(exception); + log.warn("gossip ping got exception"); + return null; + } + + @Override + public Object handle(Request request, Response response) + { + getStats().recordSuccess(); + gossip = readResponseBitSet(response); + doGossip(); + return null; + } + }); + } + catch (Exception e) { + log.warn(e, "Error scheduling request for %s", getUri()); + } + } + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetectorModule.java b/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetectorModule.java new file mode 100644 index 000000000..3175a0bea --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/CoordinatorGossipFailureDetectorModule.java @@ -0,0 +1,50 @@ +/* + * 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.prestosql.failuredetector; + +import com.google.inject.Binder; +import com.google.inject.Module; +import com.google.inject.Scopes; +import org.weakref.jmx.guice.ExportBinder; + +import static io.airlift.configuration.ConfigBinder.configBinder; +import static io.airlift.http.client.HttpClientBinder.httpClientBinder; +import static io.airlift.json.JsonCodecBinder.jsonCodecBinder; + +public class CoordinatorGossipFailureDetectorModule + implements Module +{ + @Override + public void configure(Binder binder) + { + httpClientBinder(binder) + .bindHttpClient("coordinator-gossip-failure-detector", ForFailureDetector.class) + .withTracing(); + + configBinder(binder).bindConfig(FailureDetectorConfig.class); + configBinder(binder).bindConfig(GossipProtocolConfig.class); + jsonCodecBinder(binder).bindJsonCodec(WhomToGossipInfo.class); + + binder.bind(CoordinatorGossipFailureDetector.class).in(Scopes.SINGLETON); + + binder.bind(FailureDetector.class) + .to(CoordinatorGossipFailureDetector.class) + .in(Scopes.SINGLETON); + + ExportBinder.newExporter(binder) + .export(CoordinatorGossipFailureDetector.class) + .withGeneratedName(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/GossipProtocolConfig.java b/presto-main/src/main/java/io/prestosql/failuredetector/GossipProtocolConfig.java new file mode 100644 index 000000000..3238ce3eb --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/GossipProtocolConfig.java @@ -0,0 +1,86 @@ +/* + * 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.prestosql.failuredetector; + +import io.airlift.configuration.Config; +import io.airlift.units.Duration; + +import java.util.concurrent.TimeUnit; + +public class GossipProtocolConfig +{ + public static final String GOSSIP_CONFIG_PREFIX = "failure-detector.gossip."; + public static final String GOSSIP_GROUP_SIZE = GOSSIP_CONFIG_PREFIX + "group-size"; + public static final String CN_GOSSIP_PROBE_INTERVAL = GOSSIP_CONFIG_PREFIX + "coordinator-gossip-probe-interval"; + public static final String WK_GOSSIP_PROBE_INTERVAL = GOSSIP_CONFIG_PREFIX + "worker-gossip-probe-interval"; + public static final String CN_GOSSIP_COLLATE_INTERVAL = GOSSIP_CONFIG_PREFIX + "coordinator-gossip-collate-interval"; + + private static final int CN_GOSSIP_PROBE_INTERVAL_SEC = 5; + private static final int WK_GOSSIP_PROBE_INTERVAL_SEC = 5; + private static final int CN_GOSSIP_COLLATE_INTERVAL_SEC = 2; + + private Duration cnWkUpdateGossipMonitorServiceInterval = new Duration(CN_GOSSIP_PROBE_INTERVAL_SEC, TimeUnit.SECONDS); + private Duration wkWkUpdateGossipMonitorServiceInterval = new Duration(WK_GOSSIP_PROBE_INTERVAL_SEC, TimeUnit.SECONDS); + private Duration cnGossipCollateInterval = new Duration(CN_GOSSIP_COLLATE_INTERVAL_SEC, TimeUnit.SECONDS); + private int gossipGroupSize = Integer.MAX_VALUE; // any number more than the cluster size means all to all node gossip + + @Config(GOSSIP_GROUP_SIZE) + public GossipProtocolConfig setGossipGroupSize(int s) + { + this.gossipGroupSize = s; + return this; + } + + public int getGossipGroupSize() + { + return this.gossipGroupSize; + } + + @Config(CN_GOSSIP_PROBE_INTERVAL) + public GossipProtocolConfig setCnWkUpdateGossipMonitorServiceInterval(Duration sec) + { + this.cnWkUpdateGossipMonitorServiceInterval = sec; + return this; + } + + public Duration getCnWkUpdateGossipMonitorServiceInterval() + { + return this.cnWkUpdateGossipMonitorServiceInterval; + } + + @Config(WK_GOSSIP_PROBE_INTERVAL) + public GossipProtocolConfig setWkWkUpdateGossipMonitorServiceInterval(Duration sec) + { + this.wkWkUpdateGossipMonitorServiceInterval = sec; + return this; + } + + public Duration getWkWkUpdateGossipMonitorServiceInterval() + { + return this.wkWkUpdateGossipMonitorServiceInterval; + } + + @Config(CN_GOSSIP_COLLATE_INTERVAL) + public GossipProtocolConfig setCnGossipCollateInterval(Duration sec) + { + this.cnGossipCollateInterval = sec; + return this; + } + + public Duration getCnGossipCollateInterval() + { + return this.cnGossipCollateInterval; + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java b/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java index 9c8201a38..0ee93544c 100644 --- a/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java +++ b/presto-main/src/main/java/io/prestosql/failuredetector/HeartbeatFailureDetector.java @@ -80,6 +80,8 @@ public class HeartbeatFailureDetector { private static final Logger log = Logger.get(HeartbeatFailureDetector.class); + protected Duration monitoringServiceUpdateInterval = new Duration(5, TimeUnit.SECONDS); + private final ServiceSelector selector; private final HttpClient httpClient; private final NodeInfo nodeInfo; @@ -145,10 +147,15 @@ public class HeartbeatFailureDetector log.warn(e, "Error updating services"); } } - }, 0, 5, TimeUnit.SECONDS); + }, 0, monitoringServiceUpdateInterval.toMillis(), TimeUnit.MILLISECONDS); } } + protected HttpClient getHttpClient() + { + return this.httpClient; + } + @PreDestroy public void shutdown() { @@ -171,6 +178,12 @@ public class HeartbeatFailureDetector .collect(toImmutableSet()); } + public Set getAliveNodes() + { + waitForServiceStateRefresh(); + return tasks.values().stream().filter(t -> !getFailed().contains(t)).collect(Collectors.toSet()); + } + @Override public State getState(HostAddress hostAddress) { @@ -228,11 +241,16 @@ public class HeartbeatFailureDetector return builder.build(); } + protected long getCurrentTime() + { + Map waitingTasks = getTasksTimestamp(); + return waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong(); + } + @Override public void waitForServiceStateRefresh() { - Map waitingTasks = getTasksTimestamp(); - long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong(); + long currentTime = getCurrentTime(); updateMonitoredServices(); // remove expired tasks synchronized (tasks) { @@ -281,17 +299,7 @@ public class HeartbeatFailureDetector disableOfflineTasks(onlineIds); // 3. create tasks for new services - Set newServices = online.stream() - .filter(service -> !tasks.keySet().contains(service.getId())) - .collect(toImmutableSet()); - - for (ServiceDescriptor service : newServices) { - URI uri = getHttpUri(service); - - if (uri != null) { - tasks.put(service.getId(), new MonitoringTask(service, uri)); - } - } + createTasksForNewServices(online); // 4. enable all online tasks (existing plus newly created) tasks.values().stream() @@ -300,7 +308,30 @@ public class HeartbeatFailureDetector } } - private Set getOnlineServiceDescriptors() + private Set getNewServices(Set online) + { + return online.stream() + .filter(service -> !tasks.keySet().contains(service.getId())) + .collect(toImmutableSet()); + } + + protected void createTasksForNewServices(Set online) + { + Set newServices = getNewServices(online); + for (ServiceDescriptor service : newServices) { + URI uri = getHttpUri(service); + if (uri != null) { + tasks.put(service.getId(), createNewTask(service, uri)); + } + } + } + + protected MonitoringTask createNewTask(ServiceDescriptor service, URI uri) + { + return new MonitoringTask(service, uri); + } + + protected Set getOnlineServiceDescriptors() { return selector.selectAllServices().stream() .filter(descriptor -> !nodeInfo.getNodeId().equals(descriptor.getNodeId())) @@ -332,7 +363,7 @@ public class HeartbeatFailureDetector tasks.keySet().removeAll(expiredIds); } - private URI getHttpUri(ServiceDescriptor descriptor) + protected URI getHttpUri(ServiceDescriptor descriptor) { String url = descriptor.getProperties().get(httpsRequired ? "https" : "http"); if (url != null) { @@ -347,28 +378,33 @@ public class HeartbeatFailureDetector } @ThreadSafe - private class MonitoringTask + protected class MonitoringTask { private final ServiceDescriptor service; private final URI uri; private final Stats stats; @GuardedBy("this") - private ScheduledFuture future; + protected ScheduledFuture future; @GuardedBy("this") - private Long disabledTimestamp; + protected Long disabledTimestamp; @GuardedBy("this") - private Long successTransitionTimestamp; + protected Long successTransitionTimestamp; @GuardedBy("this") - private long lastCompleteTimestamp; + protected long lastCompleteTimestamp; @GuardedBy("this") - private double lastFailureCount; + protected double lastFailureCount; - private MonitoringTask(ServiceDescriptor service, URI uri) + public URI getUri() + { + return this.uri; + } + + public MonitoringTask(ServiceDescriptor service, URI uri) { this.uri = uri; this.service = service; @@ -433,10 +469,11 @@ public class HeartbeatFailureDetector return lastCompleteTimestamp; } - private void ping() + protected void ping() { try { stats.recordStart(); + log.debug("pinging ..." + uri); httpClient.executeAsync(prepareHead().setUri(uri).build(), new ResponseHandler() { @Override @@ -444,7 +481,6 @@ public class HeartbeatFailureDetector { // ignore error stats.recordFailure(exception); - // TODO: this will technically cause an NPE in httpClient, but it's not triggered because // we never call get() on the response future. This behavior needs to be fixed in airlift return null; @@ -463,7 +499,7 @@ public class HeartbeatFailureDetector } } - private synchronized void updateState() + protected synchronized void updateState() { // is this an over/under transition? if (stats.getRecentFailureRatio() > failureRatioThreshold) { diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/WhomToGossipInfo.java b/presto-main/src/main/java/io/prestosql/failuredetector/WhomToGossipInfo.java new file mode 100644 index 000000000..73fd72d3f --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/WhomToGossipInfo.java @@ -0,0 +1,101 @@ +/* + * 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.prestosql.failuredetector; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.airlift.log.Logger; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; + +import static com.google.common.base.MoreObjects.toStringHelper; +import static java.util.Objects.requireNonNull; + +public class WhomToGossipInfo +{ + private static final Logger log = Logger.get(WhomToGossipInfo.class); + private List uri = new ArrayList<>(); + + @JsonCreator + public WhomToGossipInfo() + { + } + + @JsonCreator + public WhomToGossipInfo(String list) + { + requireNonNull(list, "list is null"); + String[] uris = list.split(","); + Arrays.stream(uris).filter(u -> !u.isEmpty()).forEach(u -> { + try { + log.debug("adding uri: " + uri); + add(new URI(u)); + } + catch (URISyntaxException e) { + log.error("failed to create object"); + } + }); + } + + @JsonCreator + public WhomToGossipInfo(@JsonProperty("uri") URI uri) + { + requireNonNull(uri, "uri is null"); + this.uri.add(uri); + } + + @JsonCreator + public WhomToGossipInfo(@JsonProperty("uri") List uri) + { + requireNonNull(uri, "uri is null"); + this.uri.addAll(uri); + } + + public void add(URI uri) + { + requireNonNull(uri, "uri is null"); + this.uri.add(uri); + } + + public List getUriList() + { + return this.uri; + } + + @JsonProperty + public String getUri() + { + StringBuilder sb = new StringBuilder(); + for (URI u : this.uri) { + sb.append(u.toString()).append(","); + } + String s = sb.toString(); + String concatUri = (s.length() > 0) ? s.substring(0, s.length() - 1) : ""; + log.debug("uri: " + concatUri); + return concatUri; + } + + @Override + public String toString() + { + return toStringHelper(this) + .add("uri", getUri()) + .toString(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetector.java b/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetector.java new file mode 100644 index 000000000..88afab5c6 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetector.java @@ -0,0 +1,141 @@ +/* + * 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.prestosql.failuredetector; + +import io.airlift.discovery.client.ServiceDescriptor; +import io.airlift.discovery.client.ServiceSelector; +import io.airlift.discovery.client.ServiceType; +import io.airlift.http.client.HttpClient; +import io.airlift.json.JsonCodec; +import io.airlift.log.Logger; +import io.airlift.node.NodeInfo; +import io.airlift.units.Duration; +import io.prestosql.server.InternalCommunicationConfig; + +import javax.annotation.PostConstruct; +import javax.annotation.concurrent.ThreadSafe; +import javax.inject.Inject; + +import java.net.URI; +import java.util.BitSet; +import java.util.Set; +import java.util.stream.Collectors; + +public class WorkerGossipFailureDetector + extends HeartbeatFailureDetector + implements FailureDetector +{ + private static final Logger log = Logger.get(WorkerGossipFailureDetector.class); + private WhomToGossipInfo whomToGossipInfo; + private final JsonCodec whomToGossipInfoCodec; + + @Inject + public WorkerGossipFailureDetector(@ServiceType("presto") ServiceSelector selector, + @ForFailureDetector HttpClient httpClient, + FailureDetectorConfig failureDetectorConfig, + NodeInfo nodeInfo, + InternalCommunicationConfig internalCommunicationConfig, + GossipProtocolConfig config, + JsonCodec whomToGossipInfoCodec) + { + this(selector, httpClient, failureDetectorConfig, nodeInfo, internalCommunicationConfig, config.getWkWkUpdateGossipMonitorServiceInterval(), whomToGossipInfoCodec); + } + + private WorkerGossipFailureDetector(@ServiceType("presto") ServiceSelector selector, + @ForFailureDetector HttpClient httpClient, + FailureDetectorConfig failureDetectorConfig, + NodeInfo nodeInfo, + InternalCommunicationConfig internalCommunicationConfig, + Duration wkWkPingInterval, + JsonCodec whomToGossipInfoCodec) + { + super(selector, httpClient, failureDetectorConfig, nodeInfo, internalCommunicationConfig); + this.monitoringServiceUpdateInterval = wkWkPingInterval; + this.whomToGossipInfoCodec = whomToGossipInfoCodec; + } + + @Override + protected MonitoringTask createNewTask(ServiceDescriptor service, URI uri) + { + log.debug("creating new gossip monitoring task for uri " + uri); + int id = 0; + for (URI u : this.whomToGossipInfo.getUriList()) { + if (u.equals(uri)) { + break; + } + id++; + } + GossipMonitoringTask task = new GossipMonitoringTask(service, uri, id); + return task; + } + + @PostConstruct + @Override + public void start() + { + log.debug("Post construct start: do nothing."); + } + + public void initWhomToGossipList(byte[] whomToGossipInfo) + { + this.whomToGossipInfo = this.whomToGossipInfoCodec.fromJson(whomToGossipInfo); + log.debug("created object " + this.whomToGossipInfo.toString()); + super.start(); + } + + @Override + protected Set getOnlineServiceDescriptors() + { + Set online = super.getOnlineServiceDescriptors().stream() + .filter(s -> this.whomToGossipInfo.getUriList() + .contains(getHttpUri(s))).collect(Collectors.toSet()); + return online; + } + + @Override + protected long getCurrentTime() + { + return System.nanoTime(); + } + + public BitSet getAliveNodesBitmap() + { + if (this.whomToGossipInfo != null) { + BitSet bitset = new BitSet(this.whomToGossipInfo.getUriList().size()); + getAliveNodes().stream().forEach(t -> bitset.set(((GossipMonitoringTask) t).getId())); + log.debug("sending bitset: " + bitset); + return bitset; + } + return new BitSet(); // no info + } + + @ThreadSafe + private class GossipMonitoringTask + extends HeartbeatFailureDetector.MonitoringTask + { + private int id; + + public GossipMonitoringTask(ServiceDescriptor service, URI uri, int id) + { + super(service, uri); + this.id = id; + } + + public int getId() + { + return this.id; + } + } +} diff --git a/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetectorModule.java b/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetectorModule.java new file mode 100644 index 000000000..40bb5903f --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/failuredetector/WorkerGossipFailureDetectorModule.java @@ -0,0 +1,50 @@ +/* + * 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.prestosql.failuredetector; + +import com.google.inject.Binder; +import com.google.inject.Module; +import com.google.inject.Scopes; +import org.weakref.jmx.guice.ExportBinder; + +import static io.airlift.configuration.ConfigBinder.configBinder; +import static io.airlift.http.client.HttpClientBinder.httpClientBinder; +import static io.airlift.json.JsonCodecBinder.jsonCodecBinder; + +public class WorkerGossipFailureDetectorModule + implements Module +{ + @Override + public void configure(Binder binder) + { + httpClientBinder(binder) + .bindHttpClient("worker-gossip-failure-detector", ForFailureDetector.class) + .withTracing(); + + configBinder(binder).bindConfig(FailureDetectorConfig.class); + configBinder(binder).bindConfig(GossipProtocolConfig.class); + jsonCodecBinder(binder).bindJsonCodec(WhomToGossipInfo.class); + + binder.bind(WorkerGossipFailureDetector.class).in(Scopes.SINGLETON); + + binder.bind(FailureDetector.class) + .to(WorkerGossipFailureDetector.class) + .in(Scopes.SINGLETON); + + ExportBinder.newExporter(binder) + .export(WorkerGossipFailureDetector.class) + .withGeneratedName(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/server/CoordinatorModule.java b/presto-main/src/main/java/io/prestosql/server/CoordinatorModule.java index 80dc6f592..a1a385eb4 100644 --- a/presto-main/src/main/java/io/prestosql/server/CoordinatorModule.java +++ b/presto-main/src/main/java/io/prestosql/server/CoordinatorModule.java @@ -100,6 +100,7 @@ import io.prestosql.execution.scheduler.AllAtOnceExecutionPolicy; import io.prestosql.execution.scheduler.ExecutionPolicy; import io.prestosql.execution.scheduler.PhasedExecutionPolicy; import io.prestosql.execution.scheduler.SplitSchedulerStats; +import io.prestosql.failuredetector.CoordinatorGossipFailureDetectorModule; import io.prestosql.failuredetector.FailureDetectorModule; import io.prestosql.memory.ClusterMemoryManager; import io.prestosql.memory.ForMemoryManager; @@ -195,6 +196,18 @@ import static org.weakref.jmx.guice.ExportBinder.newExporter; public class CoordinatorModule extends AbstractConfigurationAwareModule { + private boolean gossip; + + public CoordinatorModule(boolean gossip) + { + this.gossip = gossip; + } + + public CoordinatorModule() + { + this(false); + } + @Override protected void setup(Binder binder) { @@ -231,9 +244,16 @@ public class CoordinatorModule jaxrsBinder(binder).bind(WebUiResource.class); // failure detector - binder.install(new FailureDetectorModule()); - jaxrsBinder(binder).bind(NodeResource.class); - jaxrsBinder(binder).bind(WorkerResource.class); + if (gossip) { + binder.install(new CoordinatorGossipFailureDetectorModule()); + jaxrsBinder(binder).bind(GossipNodeResource.class); + jaxrsBinder(binder).bind(WorkerResource.class); + } + else { + binder.install(new FailureDetectorModule()); + jaxrsBinder(binder).bind(NodeResource.class); + jaxrsBinder(binder).bind(WorkerResource.class); + } httpClientBinder(binder).bindHttpClient("workerInfo", ForWorkerInfo.class); // query monitor diff --git a/presto-main/src/main/java/io/prestosql/server/GossipNodeResource.java b/presto-main/src/main/java/io/prestosql/server/GossipNodeResource.java new file mode 100644 index 000000000..ae2f24133 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/server/GossipNodeResource.java @@ -0,0 +1,31 @@ +/* + * 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.prestosql.server; + +import io.prestosql.failuredetector.CoordinatorGossipFailureDetector; + +import javax.inject.Inject; +import javax.ws.rs.Path; + +@Path("/v1/node") +public class GossipNodeResource + extends NodeResource +{ + @Inject + public GossipNodeResource(CoordinatorGossipFailureDetector failureDetector) + { + super(failureDetector); + } +} diff --git a/presto-main/src/main/java/io/prestosql/server/GossipStatusResource.java b/presto-main/src/main/java/io/prestosql/server/GossipStatusResource.java new file mode 100644 index 000000000..9b58cc556 --- /dev/null +++ b/presto-main/src/main/java/io/prestosql/server/GossipStatusResource.java @@ -0,0 +1,57 @@ +/* + * 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.prestosql.server; + +import io.airlift.log.Logger; +import io.prestosql.failuredetector.WorkerGossipFailureDetector; + +import javax.inject.Inject; +import javax.ws.rs.GET; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.core.Response; + +import static io.prestosql.server.GossipStatusResource.GOSSIP_STATUS; +import static java.util.Objects.requireNonNull; + +@Path(GOSSIP_STATUS) +public class GossipStatusResource +{ + private final WorkerGossipFailureDetector failureDetector; + public static final String GOSSIP_STATUS = "/gossip/status"; + private static final Logger log = Logger.get(GossipStatusResource.class); + + @Inject + public GossipStatusResource(WorkerGossipFailureDetector failureDetector) + { + this.failureDetector = failureDetector; + } + + @GET + public Response getAliveBitmap() + { + byte[] bitset = this.failureDetector.getAliveNodesBitmap().toByteArray(); + return Response.status(Response.Status.OK).entity(bitset).build(); + } + + @POST + public Response postWorkersToGossip(byte[] whomToGossipInfo) + { + requireNonNull(whomToGossipInfo, "whomToGossipInfo is null"); + this.failureDetector.initWhomToGossipList(whomToGossipInfo); + log.debug("init gossip at worker.."); + return Response.ok().build(); + } +} diff --git a/presto-main/src/main/java/io/prestosql/server/NodeResource.java b/presto-main/src/main/java/io/prestosql/server/NodeResource.java index eb629a53e..b17c7c9b7 100644 --- a/presto-main/src/main/java/io/prestosql/server/NodeResource.java +++ b/presto-main/src/main/java/io/prestosql/server/NodeResource.java @@ -29,6 +29,11 @@ public class NodeResource { private final HeartbeatFailureDetector failureDetector; + protected HeartbeatFailureDetector getFailureDetector() + { + return this.failureDetector; + } + @Inject public NodeResource(HeartbeatFailureDetector failureDetector) { diff --git a/presto-main/src/main/java/io/prestosql/server/ServerConfig.java b/presto-main/src/main/java/io/prestosql/server/ServerConfig.java index 423b4d567..be097a7b1 100644 --- a/presto-main/src/main/java/io/prestosql/server/ServerConfig.java +++ b/presto-main/src/main/java/io/prestosql/server/ServerConfig.java @@ -40,6 +40,10 @@ public class ServerConfig private boolean enhancedErrorReporting = true; private Duration httpClientIdleTimeout = new Duration(30, SECONDS); private Duration httpClientRequestTimeout = new Duration(10, SECONDS); + + public static final String GOSSIP = "gossip"; + public static final String HEARTBEAT = "heartbeat"; + private String failureDetectionProtocol = HEARTBEAT; // Main coordinator TODO: remove this when main coordinator election is implemented private final Set admins = new HashSet<>(); @@ -54,6 +58,20 @@ public class ServerConfig return ImmutableSet.copyOf(admins); } + @Config("failure-detection-protocol") + public ServerConfig setFailureDetectionProtocol(String protocol) + { + if (GOSSIP.equalsIgnoreCase(protocol) || HEARTBEAT.equalsIgnoreCase(protocol)) { + failureDetectionProtocol = protocol; + } + return this; + } + + public String getFailureDetectionProtocol() + { + return failureDetectionProtocol; + } + @Config("openlookeng.admins") public ServerConfig setAdmins(String adminsString) { @@ -156,4 +174,10 @@ public class ServerConfig { return httpClientRequestTimeout; } + + public static boolean isGossip(ServerConfig serverConfig) + { + String protocol = serverConfig.getFailureDetectionProtocol(); + return ServerConfig.GOSSIP.equalsIgnoreCase(protocol); + } } diff --git a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java index c5769d78b..9be050d53 100644 --- a/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java +++ b/presto-main/src/main/java/io/prestosql/server/ServerMainModule.java @@ -250,11 +250,13 @@ public class ServerMainModule { ServerConfig serverConfig = buildConfigObject(ServerConfig.class); + boolean gossip = ServerConfig.isGossip(serverConfig); + if (serverConfig.isCoordinator()) { - install(new CoordinatorModule()); + install(new CoordinatorModule(gossip)); } else { - install(new WorkerModule()); + install(new WorkerModule(gossip)); } configBinder(binder).bindConfigDefaults(HttpServerConfig.class, httpServerConfig -> { diff --git a/presto-main/src/main/java/io/prestosql/server/WorkerModule.java b/presto-main/src/main/java/io/prestosql/server/WorkerModule.java index 74b03ba7b..67d145d3a 100644 --- a/presto-main/src/main/java/io/prestosql/server/WorkerModule.java +++ b/presto-main/src/main/java/io/prestosql/server/WorkerModule.java @@ -22,6 +22,7 @@ import io.prestosql.execution.resourcegroups.NoOpResourceGroupManager; import io.prestosql.execution.resourcegroups.ResourceGroupManager; import io.prestosql.failuredetector.FailureDetector; import io.prestosql.failuredetector.NoOpFailureDetector; +import io.prestosql.failuredetector.WorkerGossipFailureDetectorModule; import io.prestosql.queryeditorui.QueryEditorConfig; import io.prestosql.server.security.NoOpWebUIAuthenticator; import io.prestosql.server.security.WebUIAuthenticator; @@ -34,10 +35,24 @@ import javax.inject.Singleton; import static com.google.common.reflect.Reflection.newProxy; import static io.airlift.configuration.ConfigBinder.configBinder; +import static io.airlift.http.client.HttpClientBinder.httpClientBinder; +import static io.airlift.jaxrs.JaxrsBinder.jaxrsBinder; public class WorkerModule implements Module { + private boolean gossip; + + public WorkerModule(boolean gossip) + { + this.gossip = gossip; + } + + public WorkerModule() + { + this(false); + } + @Override public void configure(Binder binder) { @@ -51,8 +66,16 @@ public class WorkerModule // Install no-op transaction manager on workers, since only coordinators manage transactions. binder.bind(TransactionManager.class).to(NoOpTransactionManager.class).in(Scopes.SINGLETON); - // Install no-op failure detector on workers, since only coordinators need global node selection. - binder.bind(FailureDetector.class).to(NoOpFailureDetector.class).in(Scopes.SINGLETON); + // failure detector + if (gossip) { + binder.install(new WorkerGossipFailureDetectorModule()); + jaxrsBinder(binder).bind(GossipStatusResource.class); + httpClientBinder(binder).bindHttpClient("workerInfo", ForWorkerInfo.class); + } + else { + // Install no-op failure detector on workers, since only coordinators need global node selection. + binder.bind(FailureDetector.class).to(NoOpFailureDetector.class).in(Scopes.SINGLETON); + } // HACK: this binding is needed by SystemConnectorModule, but will only be used on the coordinator binder.bind(QueryManager.class).toInstance(newProxy(QueryManager.class, (proxy, method, args) -> { diff --git a/presto-main/src/test/java/io/prestosql/failuredetector/TestCoordinatorGossipFailureDetector.java b/presto-main/src/test/java/io/prestosql/failuredetector/TestCoordinatorGossipFailureDetector.java new file mode 100644 index 000000000..0703d081f --- /dev/null +++ b/presto-main/src/test/java/io/prestosql/failuredetector/TestCoordinatorGossipFailureDetector.java @@ -0,0 +1,168 @@ +/* + * 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.prestosql.failuredetector; + +import io.airlift.units.Duration; +import org.testng.annotations.Test; + +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static java.lang.Thread.sleep; +import static org.testng.Assert.assertEquals; + +public class TestCoordinatorGossipFailureDetector +{ + private static final int START = 0; + private static final int END = 1; + private int gossipGroupSize; + + private void setGossipGroupSize(int x) + { + this.gossipGroupSize = x; + } + + private int[] assignLocalGossipGroups(int idx, int clusterSize) + { + int[] startAndEndIdx = new int[2]; + if (this.gossipGroupSize >= clusterSize) { + startAndEndIdx[END] = clusterSize; // START is already 0 by initialization + } + else { + int start = idx * this.gossipGroupSize % clusterSize; + int end = (start + this.gossipGroupSize - 1) % clusterSize; + + if (start < end) { + startAndEndIdx[START] = start; + startAndEndIdx[END] = end + 1; + } + else { + startAndEndIdx[START] = end; + startAndEndIdx[END] = start + 1; + } + } + return startAndEndIdx; + } + + @Test + public void testAssignLocalGossipGroups() + { + int clusterSize = 10; // 10 workers 0-9 + setGossipGroupSize(5); + for (int i = 0; i < clusterSize; i++) { + int[] idx1 = assignLocalGossipGroups(i, clusterSize); + if (i % 2 == 0) { + assertEquals(idx1[START], 0); + assertEquals(idx1[END], 5); + } + else { + assertEquals(idx1[START], 5); + assertEquals(idx1[END], 10); + } + } + + clusterSize = 20; // 10 workers 0-9 + setGossipGroupSize(5); + + int[] idx1 = assignLocalGossipGroups(101, clusterSize); + assertEquals(idx1[START], 5); + assertEquals(idx1[END], 10); + + setGossipGroupSize(1000); + idx1 = assignLocalGossipGroups(101, clusterSize); + assertEquals(idx1[START], 0); + assertEquals(idx1[END], 20); + } + + @Test + public void testTimeDiff() + { + Duration duration = new Duration(11, TimeUnit.SECONDS); + Duration nanos = duration.convertTo(TimeUnit.NANOSECONDS); + long now = System.nanoTime(); + try { + sleep(1); + } + catch (InterruptedException e) { + System.out.println(e.getMessage()); + } + long now1 = System.nanoTime(); + long nowdiff = now1 - now; + System.out.println(nowdiff); + if (nowdiff > nanos.getValue()) { + System.out.println("yes"); + } + else { + System.out.println("no"); + } + } + + public static class TestServiceDescriptor + { + private String nodeId; + + private URI uri; + + public TestServiceDescriptor(String nodeId, URI uri) + { + this.nodeId = nodeId; + this.uri = uri; + } + + public String getNodeId() + { + return this.nodeId; + } + + public URI getUri() + { + return this.uri; + } + } + + @Test + public void testSubList() + { + Set online = new HashSet<>(); + try { + online.add(new TestServiceDescriptor("fff1", new URI("http://10.1.1.12:8081"))); + online.add(new TestServiceDescriptor("fff2", new URI("http://10.1.1.12:8082"))); + List onlineServices = new ArrayList<>(online); + + Collections.sort(onlineServices, new Comparator() { + @Override + public int compare(TestServiceDescriptor t2, TestServiceDescriptor t1) + { + return t1.getNodeId().compareTo(t1.getNodeId()); + } + }); + List uris = new ArrayList<>(); + onlineServices.stream().forEach(o -> uris.add(o.getUri())); + + List subUris = uris.subList(START, END + 1); + assertEquals(subUris.size(), 2); + } + catch (URISyntaxException e) { + System.out.println(e.getMessage()); + } + } +} diff --git a/presto-main/src/test/java/io/prestosql/failuredetector/TestWhomToGossipInfo.java b/presto-main/src/test/java/io/prestosql/failuredetector/TestWhomToGossipInfo.java new file mode 100644 index 000000000..1b3c55d4f --- /dev/null +++ b/presto-main/src/test/java/io/prestosql/failuredetector/TestWhomToGossipInfo.java @@ -0,0 +1,181 @@ +/* + * 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.prestosql.failuredetector; + +import com.google.inject.Binder; +import com.google.inject.Injector; +import com.google.inject.Key; +import com.google.inject.Module; +import io.airlift.bootstrap.Bootstrap; +import io.airlift.discovery.client.ServiceSelector; +import io.airlift.discovery.client.testing.TestingDiscoveryModule; +import io.airlift.http.server.testing.TestingHttpServerModule; +import io.airlift.jaxrs.JaxrsModule; +import io.airlift.jmx.testing.TestingJmxModule; +import io.airlift.json.JsonCodec; +import io.airlift.json.JsonModule; +import io.airlift.node.testing.TestingNodeModule; +import io.airlift.tracetoken.TraceTokenModule; +import io.prestosql.execution.QueryManagerConfig; +import io.prestosql.server.InternalCommunicationConfig; +import org.testng.annotations.Test; + +import java.net.URI; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.BitSet; +import java.util.List; + +import static io.airlift.configuration.ConfigBinder.configBinder; +import static io.airlift.discovery.client.DiscoveryBinder.discoveryBinder; +import static io.airlift.discovery.client.ServiceTypes.serviceType; +import static io.airlift.jaxrs.JaxrsBinder.jaxrsBinder; +import static io.airlift.json.JsonCodecBinder.jsonCodecBinder; +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertTrue; + +public class TestWhomToGossipInfo +{ + private void testCreateInfo(JsonCodec codec) + { + try { + WhomToGossipInfo w = new WhomToGossipInfo(""); + String ws = "WhomToGossipInfo{uri=}"; + assertEquals(ws, w.toString()); + assertEquals(w.getUriList().size(), 0); + + WhomToGossipInfo info = new WhomToGossipInfo(new URI("http://10.18.18.225:8080")); + info.add(new URI("http://10.18.18.125:8080")); + + String json1 = "WhomToGossipInfo{uri=http://10.18.18.225:8080,http://10.18.18.125:8080}"; + assertEquals(json1, info.toString()); + + String json = codec.toJson(info); + System.out.println(json); + byte[] j = codec.toJsonBytes(info); + WhomToGossipInfo info2 = codec.fromJson(j); + assertEquals(info2.toString(), info.toString()); + + info.add(new URI("http://10.0.0.1:8080")); + + String infoString = info.toString(); + assertEquals(infoString, "WhomToGossipInfo{uri=http://10.18.18.225:8080,http://10.18.18.125:8080,http://10.0.0.1:8080}"); + + List uris = new ArrayList<>(); + uris.add(new URI("https://10.10.10.10:9090")); + uris.add(new URI("https://10.10.11.10:9090")); + uris.add(new URI("https://10.10.12.10:9090")); + + WhomToGossipInfo newInfo = new WhomToGossipInfo(uris); + assertEquals(newInfo.toString(), "WhomToGossipInfo{uri=https://10.10.10.10:9090,https://10.10.11.10:9090,https://10.10.12.10:9090}"); + json = codec.toJson(newInfo); + assertEquals(json, "{\"uri\":\"https://10.10.10.10:9090,https://10.10.11.10:9090,https://10.10.12.10:9090\"}"); + } + catch (URISyntaxException e) { + System.out.println(e.getMessage()); + } + } + + @Test + public void testInitWorkerInfoJson() + throws Exception + { + Bootstrap app = new Bootstrap( + new TestingNodeModule(), + new TestingJmxModule(), + new TestingDiscoveryModule(), + new TestingHttpServerModule(), + new TraceTokenModule(), + new JsonModule(), + new JaxrsModule(), + new CoordinatorGossipFailureDetectorModule(), + new Module() + { + @Override + public void configure(Binder binder) + { + configBinder(binder).bindConfig(InternalCommunicationConfig.class); + configBinder(binder).bindConfig(QueryManagerConfig.class); + discoveryBinder(binder).bindSelector("presto"); + discoveryBinder(binder).bindHttpAnnouncement("presto"); + jsonCodecBinder(binder).bindJsonCodec(WhomToGossipInfo.class); + + // Jersey with jetty 9 requires at least one resource + // todo add a dummy resource to airlift jaxrs in this case + jaxrsBinder(binder).bind(TestHeartbeatFailureDetector.FooResource.class); + } + }); + + Injector injector = app + .strictConfig() + .doNotInitializeLogging() + .quiet() + .initialize(); + + ServiceSelector selector = injector.getInstance(Key.get(ServiceSelector.class, serviceType("presto"))); + assertEquals(selector.selectAllServices().size(), 1); + + CoordinatorGossipFailureDetector detector = injector.getInstance(CoordinatorGossipFailureDetector.class); + detector.updateMonitoredServices(); + + testCreateInfo(detector.getWhomToGossipInfoCodec()); + + assertEquals(detector.getTotalCount(), 0); + assertEquals(detector.getActiveCount(), 0); + assertEquals(detector.getFailedCount(), 0); + assertTrue(detector.getFailed().isEmpty()); + } + + @Test + public void testEqualURI() + { + try { + URI uri1 = new URI("http://10.1.1.10:8080"); + URI uri2 = new URI("http://10.1.1.10:8080"); + + assertTrue(uri1.equals(uri2)); + } + catch (URISyntaxException e) { + System.out.println(e.getMessage()); + } + } + + @Test + public void bitsetTest() + { + BitSet b = new BitSet(5); + b.set(1); + b.set(3); + byte[] bytes = b.toByteArray(); + assertEquals(bytes.length, 1); + assertEquals(bytes[0], 10); + + b.set(0); + bytes = b.toByteArray(); + assertEquals(bytes.length, 1); + assertEquals(bytes[0], 11); + } + + @Test + public void serializationTest() + { + String s = "https://10.10.10.10:9090,https://10.10.11.10:9090,https://10.10.12.10:9090"; + byte[] bytes = s.getBytes(StandardCharsets.UTF_8); + for (int i = 0; i < bytes.length; i++) { + System.out.print(bytes[i]); + } + } +} diff --git a/presto-main/src/test/java/io/prestosql/failuredetector/TestWorkerGossipFailureDetector.java b/presto-main/src/test/java/io/prestosql/failuredetector/TestWorkerGossipFailureDetector.java new file mode 100644 index 000000000..03f457253 --- /dev/null +++ b/presto-main/src/test/java/io/prestosql/failuredetector/TestWorkerGossipFailureDetector.java @@ -0,0 +1,48 @@ +/* + * 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.prestosql.failuredetector; + +import io.airlift.discovery.client.ServiceDescriptor; +import org.testng.annotations.Test; + +import java.util.HashMap; +import java.util.Map; +import java.util.NoSuchElementException; + +import static org.testng.Assert.assertTrue; + +public class TestWorkerGossipFailureDetector +{ + @Test + public void testForEmptyWaitingTasks() + { + boolean pass = false; + try { + testEmptyServiceStateRefresh(); + } + catch (NoSuchElementException e) { + pass = true; + } + finally { + assertTrue(pass); + } + } + + private void testEmptyServiceStateRefresh() + { + Map waitingTasks = new HashMap<>(); + long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong(); + } +} diff --git a/presto-main/src/test/java/io/prestosql/server/TestServerConfig.java b/presto-main/src/test/java/io/prestosql/server/TestServerConfig.java index 0c0ab056b..b4d36773e 100644 --- a/presto-main/src/test/java/io/prestosql/server/TestServerConfig.java +++ b/presto-main/src/test/java/io/prestosql/server/TestServerConfig.java @@ -22,6 +22,8 @@ import java.util.Map; import static io.airlift.configuration.testing.ConfigAssertions.assertFullMapping; import static io.airlift.configuration.testing.ConfigAssertions.assertRecordedDefaults; +import static io.prestosql.server.ServerConfig.GOSSIP; +import static io.prestosql.server.ServerConfig.HEARTBEAT; import static java.util.concurrent.TimeUnit.HOURS; import static java.util.concurrent.TimeUnit.MINUTES; import static java.util.concurrent.TimeUnit.SECONDS; @@ -39,7 +41,8 @@ public class TestServerConfig .setGracePeriod(new Duration(2, MINUTES)) .setEnhancedErrorReporting(true) .setHttpClientIdleTimeout(new Duration(30, SECONDS)) - .setHttpClientRequestTimeout(new Duration(10, SECONDS))); + .setHttpClientRequestTimeout(new Duration(10, SECONDS)) + .setFailureDetectionProtocol(HEARTBEAT)); } @Test @@ -54,6 +57,7 @@ public class TestServerConfig .put("sql.parser.enhanced-error-reporting", "false") .put("http.client.idle-timeout", "5h") .put("http.client.request-timeout", "30m") + .put("failure-detection-protocol", GOSSIP) .build(); ServerConfig expected = new ServerConfig() @@ -64,7 +68,8 @@ public class TestServerConfig .setGracePeriod(new Duration(5, MINUTES)) .setEnhancedErrorReporting(false) .setHttpClientIdleTimeout(new Duration(5, HOURS)) - .setHttpClientRequestTimeout(new Duration(30, MINUTES)); + .setHttpClientRequestTimeout(new Duration(30, MINUTES)) + .setFailureDetectionProtocol(GOSSIP); assertFullMapping(properties, expected); }