consul enhancement (#3963)

* revert to use http ttl check for consul

* clean code

Author: moriadry

dubbo-registry/dubbo-registry-consul/src/main/java/org/apache/dubbo/registry/consul/ConsulRegistry.java

@@ -40,9 +40,12 @@
 import java.util.ArrayList;
 import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.stream.Collectors;
 
 import static java.util.concurrent.Executors.newCachedThreadPool;
@@ -57,31 +60,34 @@ public class ConsulRegistry extends FailbackRegistry {
     private static final String SERVICE_TAG = "dubbo";
     private static final String URL_META_KEY = "url";
     private static final String WATCH_TIMEOUT = "consul-watch-timeout";
-    private static final String CHECK_INTERVAL = "consul-check-interval";
-    private static final String CHECK_TIMEOUT = "consul-check-timeout";
+    private static final String CHECK_PASS_INTERVAL = "consul-check-pass-interval";
     private static final String DEREGISTER_AFTER = "consul-deregister-critical-service-after";
 
     private static final int DEFAULT_PORT = 8500;
     // default watch timeout in millisecond
     private static final int DEFAULT_WATCH_TIMEOUT = 60 * 1000;
-    // default tcp check interval
-    private static final String DEFAULT_CHECK_INTERVAL = "10s";
-    // default tcp check timeout
-    private static final String DEFAULT_CHECK_TIMEOUT = "1s";
+    // default time-to-live in millisecond
+    private static final long DEFAULT_CHECK_PASS_INTERVAL = 16000L;
     // default deregister critical server after
     private static final String DEFAULT_DEREGISTER_TIME = "20s";
 
     private ConsulClient client;
-
+    private long checkPassInterval;
     private ExecutorService notifierExecutor = newCachedThreadPool(
             new NamedThreadFactory("dubbo-consul-notifier", true));
     private ConcurrentMap<URL, ConsulNotifier> notifiers = new ConcurrentHashMap<>();
+    private ScheduledExecutorService ttlConsulCheckExecutor;
+
 
     public ConsulRegistry(URL url) {
         super(url);
         String host = url.getHost();
         int port = url.getPort() != 0 ? url.getPort() : DEFAULT_PORT;
         client = new ConsulClient(host, port);
+        checkPassInterval = url.getParameter(CHECK_PASS_INTERVAL, DEFAULT_CHECK_PASS_INTERVAL);
+        ttlConsulCheckExecutor = Executors.newSingleThreadScheduledExecutor();
+        ttlConsulCheckExecutor.scheduleAtFixedRate(this::checkPass, checkPassInterval / 8,
+                checkPassInterval / 8, TimeUnit.MILLISECONDS);
     }
 
     @Override
@@ -164,7 +170,7 @@ public List<URL> lookup(URL url) {
         }
         try {
             String service = url.getServiceKey();
-            Response<List<HealthService>> result = client.getHealthServices(service, HealthServicesRequest.newBuilder().setTag(SERVICE_TAG).build());
+            Response<List<HealthService>> result = getHealthServices(service, -1, buildWatchTimeout(url));
             if (result == null || result.getValue() == null || result.getValue().isEmpty()) {
                 return new ArrayList<>();
             } else {
@@ -184,6 +190,21 @@ public boolean isAvailable() {
     public void destroy() {
         super.destroy();
         notifierExecutor.shutdown();
+        ttlConsulCheckExecutor.shutdown();
+    }
+
+    private void checkPass() {
+        for (URL url : getRegistered()) {
+            String checkId = buildId(url);
+            try {
+                client.agentCheckPass("service:" + checkId);
+                if (logger.isDebugEnabled()) {
+                    logger.debug("check pass for url: " + url + " with check id: " + checkId);
+                }
+            } catch (Throwable t) {
+                logger.warn("fail to check pass for url: " + url + ", check id is: " + checkId);
+            }
+        }
     }
 
     private Response<List<HealthService>> getHealthServices(String service, long index, int watchTimeout) {
@@ -258,9 +279,7 @@ private String buildId(URL url) {
 
     private NewService.Check buildCheck(URL url) {
         NewService.Check check = new NewService.Check();
-        check.setTcp(url.getAddress());
-        check.setInterval(url.getParameter(CHECK_INTERVAL, DEFAULT_CHECK_INTERVAL));
-        check.setTimeout(url.getParameter(CHECK_TIMEOUT, DEFAULT_CHECK_TIMEOUT));
+        check.setTtl((checkPassInterval / 1000) + "s");
         check.setDeregisterCriticalServiceAfter(url.getParameter(DEREGISTER_AFTER, DEFAULT_DEREGISTER_TIME));
         return check;
     }