!1469 Gossip protocol implementation for failure detection

Merge pull request !1469 from i-robot/pull347

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
 

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<String, Long> gossipTales = new HashMap<>();
+    private final Set<GossipMonitoringTask> tasks = new HashSet<>();
+    private final Duration gossipValidityPeriod;
+    private final ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, daemonThreadsNamed("gossip-failure-detector"));
+    private final JsonCodec<WhomToGossipInfo> 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<WhomToGossipInfo> 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<WhomToGossipInfo> whomToGossipInfoCodec)
+    {
+        super(selector, httpClient, failureDetectorConfig, nodeInfo, internalCommunicationConfig);
+        this.monitoringServiceUpdateInterval = monitoringServiceUpdateInterval;
+        this.gossipValidityPeriod = cnGossipCollateInterval;
+        this.whomToGossipInfoCodec = whomToGossipInfoCodec;
+        this.gossipGroupSize = gossipGroupSize;
+    }
+
+    public JsonCodec<WhomToGossipInfo> 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<URI> getSortedOnlineServiceDescriptors(Set<ServiceDescriptor> online)
+    {
+        List<URI> uris = new ArrayList<>();
+        List<ServiceDescriptor> onlineServices = new ArrayList<>(online);
+        Collections.sort(onlineServices, new Comparator<ServiceDescriptor>() {
+            @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<ServiceDescriptor> 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<Map.Entry<String, Long>> 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<String> getAliveNodes()
+        {
+            Set<String> 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<String> 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<ServiceDescriptor> online = getOnlineServiceDescriptors();
+            int[] startAndEnd = assignLocalGossipGroups(this.idx, online.size());
+            List<URI> 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<Object, Exception>() {
+                @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<Object, Exception>()
+                        {
+                            @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());
+            }
+        }
+    }
+}

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();
+    }
+}

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;
+    }
+}

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<MonitoringTask> 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<ServiceDescriptor, Long> waitingTasks = getTasksTimestamp();
+        return waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong();
+    }
+
     @Override
     public void waitForServiceStateRefresh()
     {
-        Map<ServiceDescriptor, Long> waitingTasks = getTasksTimestamp();
+        long currentTime = getCurrentTime();
-        long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong();
         updateMonitoredServices();
         // remove expired tasks
         synchronized (tasks) {
@@ -281,17 +299,7 @@ public class HeartbeatFailureDetector
             disableOfflineTasks(onlineIds);
 
             // 3. create tasks for new services
-            Set<ServiceDescriptor> newServices = online.stream()
+            createTasksForNewServices(online);
-                    .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));
-                }
-            }
 
             // 4. enable all online tasks (existing plus newly created)
             tasks.values().stream()
@@ -300,7 +308,30 @@ public class HeartbeatFailureDetector
         }
     }
 
-    private Set<ServiceDescriptor> getOnlineServiceDescriptors()
+    private Set<ServiceDescriptor> getNewServices(Set<ServiceDescriptor> online)
+    {
+        return online.stream()
+                .filter(service -> !tasks.keySet().contains(service.getId()))
+                .collect(toImmutableSet());
+    }
+
+    protected void createTasksForNewServices(Set<ServiceDescriptor> online)
+    {
+        Set<ServiceDescriptor> 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<ServiceDescriptor> 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<Object, Exception>()
                 {
                     @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) {

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> 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> 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<URI> 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();
+    }
+}

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<WhomToGossipInfo> whomToGossipInfoCodec;
+
+    @Inject
+    public WorkerGossipFailureDetector(@ServiceType("presto") ServiceSelector selector,
+                                       @ForFailureDetector HttpClient httpClient,
+                                       FailureDetectorConfig failureDetectorConfig,
+                                       NodeInfo nodeInfo,
+                                       InternalCommunicationConfig internalCommunicationConfig,
+                                       GossipProtocolConfig config,
+                                       JsonCodec<WhomToGossipInfo> 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<WhomToGossipInfo> 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<ServiceDescriptor> getOnlineServiceDescriptors()
+    {
+        Set<ServiceDescriptor> 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;
+        }
+    }
+}

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();
+    }
+}

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());
+        if (gossip) {
-        jaxrsBinder(binder).bind(NodeResource.class);
+            binder.install(new CoordinatorGossipFailureDetectorModule());
-        jaxrsBinder(binder).bind(WorkerResource.class);
+            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

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);
+    }
+}

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();
+    }
+}

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)
     {

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<String> 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);
+    }
 }

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 -> {

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.
+        // failure detector
-        binder.bind(FailureDetector.class).to(NoOpFailureDetector.class).in(Scopes.SINGLETON);
+        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) -> {

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<TestServiceDescriptor> 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<TestServiceDescriptor> onlineServices = new ArrayList<>(online);
+
+            Collections.sort(onlineServices, new Comparator<TestServiceDescriptor>() {
+                @Override
+                public int compare(TestServiceDescriptor t2, TestServiceDescriptor t1)
+                {
+                    return t1.getNodeId().compareTo(t1.getNodeId());
+                }
+            });
+            List<URI> uris = new ArrayList<>();
+            onlineServices.stream().forEach(o -> uris.add(o.getUri()));
+
+            List<URI> subUris = uris.subList(START, END + 1);
+            assertEquals(subUris.size(), 2);
+        }
+        catch (URISyntaxException e) {
+            System.out.println(e.getMessage());
+        }
+    }
+}

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<WhomToGossipInfo> 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<URI> 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]);
+        }
+    }
+}

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<ServiceDescriptor, Long> waitingTasks = new HashMap<>();
+        long currentTime = waitingTasks.values().stream().mapToLong(t -> t).max().getAsLong();
+    }
+}

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