[#4687] fixed service center isolated address cannot be restored problem (#4688)

Author: chengyouling

clients/http-client-common/src/main/java/org/apache/servicecomb/http/client/utils/ServiceCombServiceAvailableUtils.java

@@ -55,7 +55,7 @@ public static void checkAddressAvailable(AbstractAddressManager addressManager,
     }
   }
 
-  private static boolean telnetCheckAddress(String address) {
+  public static boolean telnetCheckAddress(String address) {
     URI ipPort = parseIpPortFromURI(address);
     if (ipPort == null) {
       return false;
@@ -71,7 +71,8 @@ private static boolean telnetCheckAddress(String address) {
 
   private static URI parseIpPortFromURI(String address) {
     try {
-      return new URI(address);
+      String newAddress = address.startsWith("http") ? address : "https://" + address;
+      return new URI(newAddress);
     } catch (URISyntaxException e) {
       LOGGER.error("build uri error with address [{}].", address);
       return null;

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/client/IpPortManager.java

@@ -88,4 +88,16 @@ public IpPort getAvailableAddress() {
   public void recordState(String address) {
     addressManger.recordFailState(address);
   }
+
+  public void recordSuccessState(String address) {
+    addressManger.recordSuccessState(address);
+  }
+
+  public List<String> getIsolationAddresses() {
+    return addressManger.getIsolationAddresses();
+  }
+
+  public IpPort transformIpPort(String address) {
+    return addressManger.transformIpPort(address);
+  }
 }

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/client/ServiceRegistryClient.java

@@ -181,4 +181,9 @@ MicroserviceInstances findServiceInstances(String consumerId, String appId, Stri
    * @return whether this operation success
    */
   boolean updateMicroserviceInstanceStatus(String microserviceId, String instanceId, MicroserviceInstanceStatus status);
+
+  /**
+   * Check service center isolation addresses
+   */
+  void checkIsolationAddressAvailable();
 }

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/client/http/ServiceRegistryClientImpl.java

@@ -35,6 +35,7 @@
 import org.apache.servicecomb.foundation.common.net.IpPort;
 import org.apache.servicecomb.foundation.common.utils.JsonUtils;
 import org.apache.servicecomb.foundation.vertx.AsyncResultCallback;
+import org.apache.servicecomb.http.client.utils.ServiceCombServiceAvailableUtils;
 import org.apache.servicecomb.registry.api.event.MicroserviceInstanceChangedEvent;
 import org.apache.servicecomb.registry.api.registry.FindInstancesResponse;
 import org.apache.servicecomb.registry.api.registry.Framework;
@@ -90,6 +91,8 @@ public final class ServiceRegistryClientImpl implements ServiceRegistryClient {
 
   private static final String ERR_SCHEMA_NOT_EXISTS = "400016";
 
+  private static final String ADDRESS_CHECK_PATH = "/v4/default/registry/health/readiness";
+
   private final IpPortManager ipPortManager;
 
   // key是本进程的微服务id和服务管理中心的id
@@ -275,11 +278,13 @@ private Handler<RestResponse> syncHandlerForInstances(CountDownLatch countDownLa
               switch (response.statusCode()) {
                 case 304:
                   mInstances.setNeedRefresh(false);
+                  ipPortManager.recordSuccessState(requestContext.getIpPort().toString());
                   break;
                 case 200:
                   mInstances
                       .setInstancesResponse(JsonUtils.readValue(bodyBuffer.getBytes(), FindInstancesResponse.class));
                   mInstances.setNeedRefresh(true);
+                  ipPortManager.recordSuccessState(requestContext.getIpPort().toString());
                   break;
                 case 400: {
                   @SuppressWarnings("unchecked")
@@ -289,6 +294,7 @@ private Handler<RestResponse> syncHandlerForInstances(CountDownLatch countDownLa
                     mInstances.setNeedRefresh(false);
                   }
                   LOGGER.warn("failed to findInstances: " + bodyBuffer);
+                  ipPortManager.recordState(requestContext.getIpPort().toString());
                 }
                 break;
                 default:
@@ -965,6 +971,47 @@ public boolean updateMicroserviceInstanceStatus(String microserviceId, String in
     return false;
   }
 
+  @Override
+  public void checkIsolationAddressAvailable() {
+    Holder<GetAllServicesResponse> holder = new Holder<>();
+    List<String> isolationAddresses = ipPortManager.getIsolationAddresses();
+    for (String address : isolationAddresses) {
+      CountDownLatch countDownLatch = new CountDownLatch(1);
+      restClientUtil.get(ipPortManager.transformIpPort(address),
+          ADDRESS_CHECK_PATH,
+          new RequestParam(),
+          addressSyncHandler(countDownLatch, GetAllServicesResponse.class, holder));
+      try {
+        countDownLatch.await();
+        if (holder.statusCode == 200) {
+          ipPortManager.recordSuccessState(address);
+          return;
+        }
+        if (holder.statusCode == 404 && ServiceCombServiceAvailableUtils.telnetCheckAddress(address)) {
+          ipPortManager.recordSuccessState(address);
+          return;
+        }
+      } catch (Exception e) {
+        LOGGER.error("check service center isolation address [{}] failed", address);
+      }
+    }
+  }
+
+  public <T> Handler<RestResponse> addressSyncHandler(CountDownLatch countDownLatch, Class<T> cls,
+      Holder<T> holder) {
+    return restResponse -> {
+      HttpClientResponse response = restResponse.getResponse();
+      if (response == null) {
+        countDownLatch.countDown();
+        holder.setStatusCode(500);
+        return;
+      }
+      holder.setStatusCode(response.statusCode());
+      sendUnAuthorizedEvent(response);
+      countDownLatch.countDown();
+    };
+  }
+
   @Subscribe
   public void onMicroserviceHeartbeatTask(MicroserviceInstanceHeartbeatTask event) {
     if (HeartbeatResult.SUCCESS.equals(event.getHeartbeatResult())) {

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/refresh/ServiceRegistryAddressManager.java

@@ -45,7 +45,7 @@ protected String normalizeUri(String endpoint) {
     return new URIEndpointObject(endpoint).toString();
   }
 
-  private IpPort transformIpPort(String address) {
+  public IpPort transformIpPort(String address) {
     URI uri = URI.create(URI_PREFIX + address);
     return new IpPort(uri.getHost(), uri.getPort());
   }

service-registry/registry-service-center/src/main/java/org/apache/servicecomb/serviceregistry/task/MicroserviceInstanceRegisterTask.java

@@ -16,6 +16,10 @@
  */
 package org.apache.servicecomb.serviceregistry.task;
 
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.servicecomb.registry.RegistrationManager;
 import org.apache.servicecomb.registry.api.registry.Microservice;
@@ -36,13 +40,18 @@ public class MicroserviceInstanceRegisterTask extends AbstractRegisterTask {
 
   private final MicroserviceInstance microserviceInstance;
 
+  private final ScheduledExecutorService addrCheckExecutor;
+
+  private boolean isAddrCheckInit = false;
+
   public MicroserviceInstanceRegisterTask(EventBus eventBus, ServiceRegistryConfig serviceRegistryConfig,
       ServiceRegistryClient srClient,
       Microservice microservice) {
     super(eventBus, srClient, microservice);
 
     this.serviceRegistryConfig = serviceRegistryConfig;
     this.microserviceInstance = microservice.getInstance();
+    this.addrCheckExecutor = Executors.newScheduledThreadPool(1, (t) -> new Thread(t, "sc-addr-check"));
   }
 
   @Subscribe
@@ -90,6 +99,18 @@ protected boolean doRegister() {
         microserviceInstance.getEndpoints(),
         microserviceInstance.getHealthCheck().getTTL());
 
+    if (!isAddrCheckInit) {
+      addrCheckExecutor.scheduleWithFixedDelay(new CheckAddressTask(), 0,
+          serviceRegistryConfig.getHeartbeatInterval(), TimeUnit.SECONDS);
+      isAddrCheckInit = true;
+    }
     return true;
   }
+
+  class CheckAddressTask implements Runnable {
+    @Override
+    public void run() {
+      srClient.checkIsolationAddressAvailable();
+    }
+  }
 }

service-registry/registry-service-center/src/test/java/org/apache/servicecomb/serviceregistry/client/LocalServiceRegistryClientImpl.java

@@ -458,4 +458,9 @@ public boolean updateMicroserviceInstanceStatus(String microserviceId, String in
     microserviceInstance.setStatus(status);
     return true;
   }
+
+  @Override
+  public void checkIsolationAddressAvailable() {
+    // NOT SUPPORT
+  }
 }