Add connection health checking to static discovery strategy for WAN [HZ-2888]

hazelcast/src/main/java/com/hazelcast/internal/metrics/MetricDescriptorConstants.java

@@ -699,6 +699,9 @@ public final class MetricDescriptorConstants {
     public static final String WAN_METRIC_ACK_DELAY_CURRENT_MILLIS = "ackDelayCurrentMillis";
     public static final String WAN_METRIC_ACK_DELAY_LAST_START = "ackDelayLastStart";
     public static final String WAN_METRIC_ACK_DELAY_LAST_END = "ackDelayLastEnd";
+    public static final String WAN_METRIC_CONNECTION_HEALTH = "connectionHealth";
+    public static final String WAN_DISCRIMINATOR_CONNECTION_ADDRESS = "address";
+    public static final String WAN_TAG_DISCOVERY_STRATEGY = "discoveryStrategy";
     public static final String WAN_QUEUE_FILL_PERCENT = "queueFillPercent";
     // ===[/WAN]========================================================
 

hazelcast/src/main/java/com/hazelcast/spi/discovery/DiscoveryStrategy.java

@@ -16,11 +16,15 @@
 
 package com.hazelcast.spi.discovery;
 
+import com.hazelcast.cluster.Address;
 import com.hazelcast.cluster.Member;
+import com.hazelcast.spi.annotation.PrivateApi;
 import com.hazelcast.spi.partitiongroup.PartitionGroupStrategy;
 
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * The <code>DiscoveryStrategy</code> itself is the actual implementation to discover
@@ -112,4 +116,29 @@ default PartitionGroupStrategy getPartitionGroupStrategy() {
      * @since 3.7
      */
     Map<String, String> discoverLocalMetadata();
+
+    /**
+     * Marks the passed {@link Address} as unhealthy, which prevents it from being offered as a
+     * viable endpoint in some {@link DiscoveryStrategy} implementations, usually prompting
+     * this endpoint to be periodically probed for liveliness. If not supported by the underlying
+     * implementation, then this call does nothing.
+     *
+     * @param address the address to mark as unhealthy
+     * @since 5.4
+     */
+    @PrivateApi
+    default void markEndpointAsUnhealthy(Address address) { }
+
+    /**
+     * Fetches a set of {@link Address} marked as unhealthy by the underlying implementation.
+     * If not supported by the underlying implementation, then this call returns an empty set.
+     *
+     * @return set of {@link Address} which are currently marked as unhealthy if supported by the
+     *  underlying implementation, otherwise an empty set.
+     * @since 5.4
+     */
+    @PrivateApi
+    default Set<Address> getUnhealthyEndpoints() {
+        return Collections.emptySet();
+    }
 }

hazelcast/src/main/java/com/hazelcast/spi/discovery/impl/DefaultDiscoveryService.java

@@ -16,6 +16,7 @@
 
 package com.hazelcast.spi.discovery.impl;
 
+import com.hazelcast.cluster.Address;
 import com.hazelcast.config.DiscoveryConfig;
 import com.hazelcast.config.DiscoveryStrategyConfig;
 import com.hazelcast.config.InvalidConfigurationException;
@@ -41,8 +42,7 @@
 
 import static com.hazelcast.internal.util.CollectionUtil.nullToEmpty;
 
-public class DefaultDiscoveryService
-        implements DiscoveryService {
+public class DefaultDiscoveryService implements DiscoveryService {
 
     private static final String SERVICE_LOADER_TAG = DiscoveryStrategyFactory.class.getCanonicalName();
 
@@ -98,6 +98,28 @@ public void destroy() {
         }
     }
 
+    @Override
+    public void markEndpointAsUnhealthy(Address address) {
+        for (DiscoveryStrategy discoveryStrategy : discoveryStrategies) {
+            discoveryStrategy.markEndpointAsUnhealthy(address);
+        }
+    }
+
+    @Override
+    public Set<Address> getUnhealthyEndpoints() {
+        Set<Address> combinedAddresses = null;
+        for (DiscoveryStrategy strategy : discoveryStrategies) {
+            Set<Address> strategyAddresses = strategy.getUnhealthyEndpoints();
+            if (!strategyAddresses.isEmpty()) {
+                if (combinedAddresses == null) {
+                    combinedAddresses = new HashSet<>(strategyAddresses.size());
+                }
+                combinedAddresses.addAll(strategyAddresses);
+            }
+        }
+        return combinedAddresses == null ? Collections.emptySet() : combinedAddresses;
+    }
+
     public Iterable<DiscoveryStrategy> getDiscoveryStrategies() {
         return discoveryStrategies;
     }

hazelcast/src/main/java/com/hazelcast/spi/discovery/impl/PredefinedDiscoveryService.java

@@ -16,11 +16,13 @@
 
 package com.hazelcast.spi.discovery.impl;
 
+import com.hazelcast.cluster.Address;
 import com.hazelcast.spi.discovery.DiscoveryNode;
 import com.hazelcast.spi.discovery.DiscoveryStrategy;
 import com.hazelcast.spi.discovery.integration.DiscoveryService;
 
 import java.util.Map;
+import java.util.Set;
 
 import static com.hazelcast.internal.util.Preconditions.checkNotNull;
 
@@ -49,8 +51,22 @@ public Map<String, String> discoverLocalMetadata() {
         return strategy.discoverLocalMetadata();
     }
 
+    @Override
+    public void markEndpointAsUnhealthy(Address address) {
+        strategy.markEndpointAsUnhealthy(address);
+    }
+
+    @Override
+    public Set<Address> getUnhealthyEndpoints() {
+        return strategy.getUnhealthyEndpoints();
+    }
+
     @Override
     public void destroy() {
         strategy.destroy();
     }
+
+    public DiscoveryStrategy getStrategy() {
+        return strategy;
+    }
 }

hazelcast/src/main/java/com/hazelcast/spi/discovery/integration/DiscoveryService.java

@@ -16,12 +16,16 @@
 
 package com.hazelcast.spi.discovery.integration;
 
+import com.hazelcast.cluster.Address;
 import com.hazelcast.cluster.Member;
+import com.hazelcast.spi.annotation.PrivateApi;
 import com.hazelcast.spi.discovery.DiscoveryNode;
 import com.hazelcast.spi.discovery.DiscoveryStrategy;
 import com.hazelcast.spi.discovery.NodeFilter;
 
+import java.util.Collections;
 import java.util.Map;
+import java.util.Set;
 
 /**
  * The <code>DiscoveryService</code> interface defines the basic entry point
@@ -83,4 +87,33 @@ public interface DiscoveryService {
      * @since 3.7
      */
     Map<String, String> discoverLocalMetadata();
+
+    /**
+     * Marks the passed {@link Address} as unhealthy, which prevents it from being offered as a
+     * viable endpoint in some {@link DiscoveryStrategy} implementations, usually prompting
+     * this endpoint to be periodically probed for liveliness. If not supported by the underlying
+     * implementation, then this call does nothing.
+     *
+     * @see DiscoveryStrategy#markEndpointAsUnhealthy(Address)
+     *
+     * @param address the address to mark as unhealthy
+     * @since 5.4
+     */
+    @PrivateApi
+    default void markEndpointAsUnhealthy(Address address) { }
+
+    /**
+     * Fetches a set of {@link Address} marked as unhealthy by the underlying {@link DiscoveryStrategy}.
+     * If not supported by the underlying implementation, then this call returns an empty set.
+     *
+     * @see DiscoveryStrategy#getUnhealthyEndpoints()
+     *
+     * @return set of {@link Address} which are currently marked as unhealthy if supported by the
+     *  underlying implementation, otherwise an empty set.
+     * @since 5.4
+     */
+    @PrivateApi
+    default Set<Address> getUnhealthyEndpoints() {
+        return Collections.emptySet();
+    }
 }