Fix graceful shutdown notify failure

Signed-off-by: lilai <lilai23@foxmail.com>
sermant-io · Jul 31, 2024 · 3af06e3 · 3af06e3
1 parent 71acc27
commit 3af06e3
Show file tree

Hide file tree

Showing 7 changed files with 40 additions and 10 deletions.
diff --git a/.github/actions/scenarios/spring/graceful/action.yml b/.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.

diff --git a/...dubbo-test/dubbo-integration-test/src/test/java/io/sermant/integration/lane/LaneTest.java b/...dubbo-test/dubbo-integration-test/src/test/java/io/sermant/integration/lane/LaneTest.java
@@ -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);

diff --git a/...t/spring-integration-test/src/test/java/io/sermant/integration/graceful/GracefulTest.java b/...t/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);

diff --git a/...ervice-registry/registry-common/src/main/java/io/sermant/registry/config/GraceConfig.java b/...ervice-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 expiration time of 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
      *

diff --git a/...gistry/registry-common/src/main/java/io/sermant/registry/config/grace/GraceConstants.java b/...gistry/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 expiration time for the cache upstream address
+     */
+    public static final long WAIT_NOTIFY_TIME = 20L;
+
     /**
      * Maximum port
      */

diff --git a/...cloud-registry-plugin/src/main/java/io/sermant/registry/entity/GraceShutdownBehavior.java b/...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) {

diff --git a/...oud-registry-service/src/main/java/io/sermant/registry/service/impl/GraceServiceImpl.java b/...oud-registry-service/src/main/java/io/sermant/registry/service/impl/GraceServiceImpl.java
@@ -27,6 +27,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;
@@ -57,6 +58,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 +73,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 +86,14 @@ 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;
+            }
+        }
     }
 
     /**