Merge pull request #1590 from lilai23/grace_shutdown_fix

Fix graceful shutdown notify failure

.github/actions/scenarios/spring/graceful/action.yml

@@ -37,6 +37,7 @@ runs:
         grace.rule.enableGraceShutdown: true
         grace.rule.enableOfflineNotify: true
         grace.rule.warmUpTime: 600
+        grace.rule.upstreamAddressExpiredTime: 600
         servicecomb.service.enableSpringRegister: true
         servicecomb.service.preferIpAddress: true
       # graceful-rest-provider service port 8443 do not change, it special for springCloud Edgware.SR2 test ssl feature.

sermant-integration-tests/dubbo-test/dubbo-integration-test/src/test/java/io/sermant/integration/lane/LaneTest.java

@@ -61,7 +61,7 @@ public LaneTest() {
     }
 
     @Test
-    public void testDubbo() {
+    public void testDubbo() throws InterruptedException {
         if (isExecuteSpringTest){
             testBySpring("Dubbo");
         }
@@ -121,15 +121,15 @@ public void testDubbo() {
     }
 
     @Test
-    public void testFeign() {
+    public void testFeign() throws InterruptedException {
         if (isExecuteSpringTest){
             testBySpring("Feign");
             testByDubbo("Feign");
         }
     }
 
     @Test
-    public void testRest() {
+    public void testRest() throws InterruptedException {
         if (isExecuteSpringTest){
             testBySpring("Rest");
             testByDubbo("Rest");
@@ -200,7 +200,8 @@ private void testBySpring(String path) {
      *
      * @param path 路径
      */
-    private void testByDubbo(String path) {
+    private void testByDubbo(String path) throws InterruptedException {
+        Thread.sleep(10000);
         // 正常染色
         HttpHeaders headers = new HttpHeaders();
         HttpEntity<String> entity = new HttpEntity<>(null, headers);

sermant-integration-tests/spring-test/spring-integration-test/src/test/java/io/sermant/integration/graceful/GracefulTest.java

@@ -66,7 +66,7 @@ public void testGracefulUp() {
         final Map<String, Integer> statisticMap = new HashMap<>();
         for (int i = 0; i < 4; i++) {
             try {
-                for(int j = 0; j < UP_REQUEST_COUNT; j++) {
+                for (int j = 0; j < UP_REQUEST_COUNT; j++) {
                     statistic(statisticMap);
                 }
                 Thread.sleep(10000);
@@ -98,15 +98,14 @@ private boolean isTargetTest(String type) {
     /**
      * 测试优雅下线
      */
+    @Test
     public void testGracefulDown() {
         if (!isTargetTest("down")) {
             return;
         }
         try {
             for (int i = 0; i < DOWN_REQUEST_COUNT; i++) {
-                String port = RequestUtils.get(buildUrl("testGraceful"), Collections.emptyMap(),
-                        String.class);
-                System.out.println("port: " + port);
+                RequestUtils.get(buildUrl("testGraceful"), Collections.emptyMap(), String.class);
             }
         } catch (Exception exception) {
             LOGGER.error(exception.getMessage(), exception);

sermant-plugins/sermant-service-registry/registry-common/src/main/java/io/sermant/registry/config/GraceConfig.java

@@ -114,6 +114,11 @@ public class GraceConfig implements PluginConfig, Cloneable {
      */
     private long upstreamAddressExpiredTime = GraceConstants.UPSTREAM_ADDRESS_DEFAULT_EXPIRED_TIME;
 
+    /**
+     * Cache the wait time of notifying upstream addresses
+     */
+    private long waitNotifyTime = GraceConstants.WAIT_NOTIFY_TIME;
+
     /**
      * Correct the relevant switch attributes according to the aggregation switch,
      * and turn on all functions of elegant online and offline with one click
@@ -262,6 +267,14 @@ public void setUpstreamAddressExpiredTime(long upstreamAddressExpiredTime) {
         this.upstreamAddressExpiredTime = upstreamAddressExpiredTime;
     }
 
+    public long getWaitNotifyTime() {
+        return waitNotifyTime;
+    }
+
+    public void setWaitNotifyTime(long waitNotifyTime) {
+        this.waitNotifyTime = waitNotifyTime;
+    }
+
     /**
      * Check whether the preheating parameter is valid
      *

sermant-plugins/sermant-service-registry/registry-common/src/main/java/io/sermant/registry/config/grace/GraceConstants.java

@@ -156,6 +156,11 @@ public class GraceConstants {
      */
     public static final long UPSTREAM_ADDRESS_DEFAULT_EXPIRED_TIME = 60L;
 
+    /**
+     * The default wait time of notifying upstream addresses
+     */
+    public static final long WAIT_NOTIFY_TIME = 20L;
+
     /**
      * Maximum port
      */

sermant-plugins/sermant-service-registry/spring-cloud-registry-plugin/src/main/java/io/sermant/registry/entity/GraceShutdownBehavior.java

@@ -56,6 +56,8 @@ public void run() {
     }
 
     private void graceShutDown() {
+        // wait notify consumer complete
+        CommonUtils.sleep(graceConfig.getWaitNotifyTime() * ConfigConstants.SEC_DELTA);
         long shutdownWaitTime = graceConfig.getShutdownWaitTime() * ConfigConstants.SEC_DELTA;
         final long shutdownCheckTimeUnit = graceConfig.getShutdownCheckTimeUnit() * ConfigConstants.SEC_DELTA;
         while (GraceContext.INSTANCE.getGraceShutDownManager().getRequestCount() > 0 && shutdownWaitTime > 0) {

sermant-plugins/sermant-service-registry/spring-cloud-registry-service/src/main/java/io/sermant/registry/service/impl/GraceServiceImpl.java

@@ -18,6 +18,7 @@
 
 import com.alibaba.fastjson.JSONObject;
 
+import io.sermant.core.common.LoggerFactory;
 import io.sermant.core.plugin.config.PluginConfigManager;
 import io.sermant.core.plugin.service.PluginServiceManager;
 import io.sermant.core.utils.ReflectUtils;
@@ -27,6 +28,7 @@
 import io.sermant.registry.context.RegisterContext;
 import io.sermant.registry.context.RegisterContext.ClientInfo;
 import io.sermant.registry.service.cache.AddressCache;
+import io.sermant.registry.service.utils.HttpClientResult;
 import io.sermant.registry.service.utils.HttpClientUtils;
 import io.sermant.registry.services.GraceService;
 import io.sermant.registry.services.RegisterCenterService;
@@ -39,6 +41,8 @@
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.logging.Level;
+import java.util.logging.Logger;
 
 /**
  * Offline notification service
@@ -47,6 +51,8 @@
  * @since 2022-05-26
  */
 public class GraceServiceImpl implements GraceService {
+    private static final Logger LOGGER = LoggerFactory.getLogger();
+
     private static final Executor EXECUTOR = Executors.newFixedThreadPool(10);
 
     private static final AtomicBoolean SHUTDOWN = new AtomicBoolean();
@@ -57,6 +63,8 @@ public class GraceServiceImpl implements GraceService {
 
     private static final String REQUEST_BODY = JSONObject.toJSONString(new Object());
 
+    private static final int RETRY_TIME = 3;
+
     /**
      * Offline notifications
      */
@@ -70,10 +78,10 @@ public void shutdown() {
             ClientInfo clientInfo = RegisterContext.INSTANCE.getClientInfo();
             Map<String, Collection<String>> header = new HashMap<>();
             header.put(GraceConstants.MARK_SHUTDOWN_SERVICE_NAME,
-                Collections.singletonList(clientInfo.getServiceName()));
+                    Collections.singletonList(clientInfo.getServiceName()));
             header.put(GraceConstants.MARK_SHUTDOWN_SERVICE_ENDPOINT,
-                Arrays.asList(clientInfo.getIp() + ":" + clientInfo.getPort(),
-                    clientInfo.getHost() + ":" + clientInfo.getPort()));
+                    Arrays.asList(clientInfo.getIp() + ":" + clientInfo.getPort(),
+                            clientInfo.getHost() + ":" + clientInfo.getPort()));
             AddressCache.INSTANCE.getAddressSet().forEach(address -> notifyToGraceHttpServer(address, header));
         }
     }
@@ -83,8 +91,15 @@ private void notifyToGraceHttpServer(String address, Map<String, Collection<Stri
     }
 
     private void execute(String address, Map<String, Collection<String>> header) {
-        HttpClientUtils.INSTANCE.doPost(GRACE_HTTP_SERVER_PROTOCOL + address + GraceConstants.GRACE_NOTIFY_URL_PATH,
-                REQUEST_BODY, header);
+        for (int i = 0; i < RETRY_TIME; i++) {
+            HttpClientResult result = HttpClientUtils.INSTANCE.doPost(
+                    GRACE_HTTP_SERVER_PROTOCOL + address + GraceConstants.GRACE_NOTIFY_URL_PATH,
+                    REQUEST_BODY, header);
+            if (result.getCode() == GraceConstants.GRACE_HTTP_SUCCESS_CODE) {
+                break;
+            }
+            LOGGER.log(Level.WARNING, "Failed to notify before shutdown, address: {0}", address);
+        }
     }
 
     /**