[ISSUE alibaba#5169] Fix instance beat run only by responsible server. (

alibaba#5179)

* [ISSUE alibaba#5169] DISTRO filter support `beat` parameter

For old version clients.

* [ISSUE alibaba#5169] Only responsible server need to check instance heartbeats

naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/InstanceBeatCheckResponsibleInterceptor.java

@@ -0,0 +1,40 @@
+/*
+ * Copyright (c) 1999-2021 Alibaba Group Holding Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.alibaba.nacos.naming.healthcheck.heartbeat;
+
+import com.alibaba.nacos.naming.core.DistroMapper;
+import com.alibaba.nacos.sys.utils.ApplicationUtils;
+
+/**
+ * Instance responsibility check interceptor.
+ *
+ * @author gengtuo.ygt
+ * on 2021/3/24
+ */
+public class InstanceBeatCheckResponsibleInterceptor extends AbstractBeatCheckInterceptor {
+
+    @Override
+    public boolean intercept(InstanceBeatCheckTask object) {
+        return !ApplicationUtils.getBean(DistroMapper.class).responsible(object.getClient().getResponsibleId());
+    }
+
+    @Override
+    public int order() {
+        return Integer.MIN_VALUE + 2;
+    }
+
+}

naming/src/main/java/com/alibaba/nacos/naming/healthcheck/heartbeat/InstanceBeatCheckTask.java

@@ -17,7 +17,7 @@
 package com.alibaba.nacos.naming.healthcheck.heartbeat;
 
 import com.alibaba.nacos.common.spi.NacosServiceLoader;
-import com.alibaba.nacos.naming.core.v2.client.Client;
+import com.alibaba.nacos.naming.core.v2.client.impl.IpPortBasedClient;
 import com.alibaba.nacos.naming.core.v2.pojo.HealthCheckInstancePublishInfo;
 import com.alibaba.nacos.naming.core.v2.pojo.Service;
 import com.alibaba.nacos.naming.interceptor.Interceptable;
@@ -34,7 +34,7 @@ public class InstanceBeatCheckTask implements Interceptable {
 
     private static final List<InstanceBeatChecker> CHECKERS = new LinkedList<>();
 
-    private final Client client;
+    private final IpPortBasedClient client;
 
     private final Service service;
 
@@ -46,7 +46,7 @@ public class InstanceBeatCheckTask implements Interceptable {
         CHECKERS.addAll(NacosServiceLoader.load(InstanceBeatChecker.class));
     }
 
-    public InstanceBeatCheckTask(Client client, Service service, HealthCheckInstancePublishInfo instancePublishInfo) {
+    public InstanceBeatCheckTask(IpPortBasedClient client, Service service, HealthCheckInstancePublishInfo instancePublishInfo) {
         this.client = client;
         this.service = service;
         this.instancePublishInfo = instancePublishInfo;
@@ -63,7 +63,7 @@ public void passIntercept() {
     public void afterIntercept() {
     }
 
-    public Client getClient() {
+    public IpPortBasedClient getClient() {
         return client;
     }
 

naming/src/main/java/com/alibaba/nacos/naming/web/DistroIpPortTagGenerator.java

@@ -16,9 +16,12 @@
 
 package com.alibaba.nacos.naming.web;
 
+import com.alibaba.nacos.api.exception.runtime.NacosDeserializationException;
 import com.alibaba.nacos.common.utils.IPUtil;
+import com.alibaba.nacos.common.utils.JacksonUtils;
 import com.alibaba.nacos.common.utils.StringUtils;
 import com.alibaba.nacos.core.utils.ReuseHttpServletRequest;
+import com.alibaba.nacos.naming.healthcheck.RsInfo;
 
 /**
  * Distro IP and port tag generator.
@@ -27,6 +30,8 @@
  */
 public class DistroIpPortTagGenerator implements DistroTagGenerator {
 
+    private static final String PARAMETER_BEAT = "beat";
+
     private static final String PARAMETER_IP = "ip";
 
     private static final String PARAMETER_PORT = "port";
@@ -35,6 +40,18 @@ public class DistroIpPortTagGenerator implements DistroTagGenerator {
     public String getResponsibleTag(ReuseHttpServletRequest request) {
         String ip = request.getParameter(PARAMETER_IP);
         String port = request.getParameter(PARAMETER_PORT);
+        if (StringUtils.isBlank(ip)) {
+            // some old version clients using beat parameter
+            String beatStr = request.getParameter(PARAMETER_BEAT);
+            if (StringUtils.isNotBlank(beatStr)) {
+                try {
+                    RsInfo rsInfo = JacksonUtils.toObj(beatStr, RsInfo.class);
+                    ip = rsInfo.getIp();
+                    port = String.valueOf(rsInfo.getPort());
+                } catch (NacosDeserializationException ignored) {
+                }
+            }
+        }
         if (StringUtils.isNotBlank(ip)) {
             ip = ip.trim();
         }

naming/src/main/resources/META-INF/services/com.alibaba.nacos.naming.healthcheck.heartbeat.AbstractBeatCheckInterceptor

@@ -16,3 +16,4 @@
 
 com.alibaba.nacos.naming.healthcheck.heartbeat.ServiceEnableBeatCheckInterceptor
 com.alibaba.nacos.naming.healthcheck.heartbeat.InstanceEnableBeatCheckInterceptor
+com.alibaba.nacos.naming.healthcheck.heartbeat.InstanceBeatCheckResponsibleInterceptor