feat:support polaris server nearby backup LB. (#615)

Author: SkyeBeFreeman

polaris-common/polaris-config-default/src/main/resources/conf/default-config.yml

@@ -61,6 +61,8 @@ global:
     #描述：默认服务端埋点接入地址
     addresses:
       - 127.0.0.1:8091
+    #描述：默认服务端地址负载均衡策略
+    lbPolicy: roundRobin
     #描述: 访问server的连接协议，SDK会根据协议名称会加载对应的插件
     protocol: grpc
     #描述: 发起连接后的连接超时时间

polaris-common/polaris-config/src/main/java/com/tencent/polaris/api/config/consumer/LoadBalanceConfig.java

@@ -43,6 +43,16 @@ public interface LoadBalanceConfig extends PluginConfig, Verifier {
      */
     String LOAD_BALANCE_WEIGHTED_ROUND_ROBIN = "weightedRoundRobin";
 
+    /**
+     * 轮询负载均衡插件名
+     */
+    String LOAD_BALANCE_ROUND_ROBIN = "roundRobin";
+
+    /**
+     * 就近主备负载均衡插件名
+     */
+    String LOAD_BALANCE_NEARBY_BACKUP = "nearbyBackup";
+
     /**
      * 负载均衡类型
      *

polaris-common/polaris-config/src/main/java/com/tencent/polaris/api/config/global/ServerConnectorConfig.java

@@ -19,6 +19,7 @@
 
 import com.tencent.polaris.api.config.plugin.PluginConfig;
 import com.tencent.polaris.api.config.verify.Verifier;
+
 import java.util.List;
 import java.util.Map;
 
@@ -36,6 +37,13 @@ public interface ServerConnectorConfig extends PluginConfig, Verifier {
      */
     List<String> getAddresses();
 
+    /**
+     * 远端server地址负载均衡策略
+     *
+     * @return 负载均衡策略
+     */
+    String getLbPolicy();
+
     /**
      * 与server对接的协议，默认GRPC
      *

polaris-common/polaris-config/src/main/java/com/tencent/polaris/factory/config/global/ServerConnectorConfigImpl.java

@@ -19,12 +19,16 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
+import com.tencent.polaris.api.config.consumer.LoadBalanceConfig;
 import com.tencent.polaris.api.config.global.ServerConnectorConfig;
 import com.tencent.polaris.api.config.plugin.DefaultPlugins;
 import com.tencent.polaris.api.utils.CollectionUtils;
+import com.tencent.polaris.api.utils.StringUtils;
 import com.tencent.polaris.factory.config.plugin.PluginConfigImpl;
 import com.tencent.polaris.factory.util.ConfigUtils;
 import com.tencent.polaris.factory.util.TimeStrJsonDeserializer;
+
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -42,6 +46,8 @@ public class ServerConnectorConfigImpl extends PluginConfigImpl implements Serve
     @JsonProperty
     private List<String> addresses;
     @JsonProperty
+    private String lbPolicy = LoadBalanceConfig.LOAD_BALANCE_ROUND_ROBIN;
+    @JsonProperty
     private String protocol;
     @JsonProperty
     @JsonDeserialize(using = TimeStrJsonDeserializer.class)
@@ -78,6 +84,15 @@ public void setAddresses(List<String> addresses) {
         this.addresses = addresses;
     }
 
+    @Override
+    public String getLbPolicy() {
+        return lbPolicy;
+    }
+
+    public void setLbPolicy(String lbPolicy) {
+        this.lbPolicy = lbPolicy;
+    }
+
     @Override
     public String getProtocol() {
         return protocol;
@@ -199,6 +214,10 @@ public void verify() {
                 throw new IllegalArgumentException(String.format("address [%s] of [%s] is invalid", address, protocol));
             }
         }
+        List<String> targetLbPolicyList = new ArrayList<>();
+        targetLbPolicyList.add(LoadBalanceConfig.LOAD_BALANCE_ROUND_ROBIN);
+        targetLbPolicyList.add(LoadBalanceConfig.LOAD_BALANCE_NEARBY_BACKUP);
+        ConfigUtils.validateIn(lbPolicy, targetLbPolicyList, "serverConnector.lbPolicy");
         ConfigUtils.validateString(id, "serverConnector.id");
         if (DefaultPlugins.SERVER_CONNECTOR_GRPC.equals(protocol)) {
             ConfigUtils.validateString(protocol, "serverConnector.protocol");
@@ -218,6 +237,9 @@ public void setDefault(Object defaultObject) {
             if (null == addresses) {
                 setAddresses(serverConnectorConfig.getAddresses());
             }
+            if (StringUtils.isBlank(lbPolicy)) {
+                setLbPolicy(LoadBalanceConfig.LOAD_BALANCE_ROUND_ROBIN);
+            }
             if (null == protocol) {
                 setProtocol(serverConnectorConfig.getProtocol());
             }

polaris-common/polaris-config/src/main/java/com/tencent/polaris/factory/util/ConfigUtils.java

@@ -18,7 +18,10 @@
 package com.tencent.polaris.factory.util;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.tencent.polaris.api.utils.CollectionUtils;
 import com.tencent.polaris.api.utils.StringUtils;
+
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -114,4 +117,16 @@ public static void validateTrue(Boolean value, String name) {
             throw new IllegalArgumentException(name + " must not be false");
         }
     }
+
+    public static void validateIn(String value, List<String> targetList, String name) {
+        if (StringUtils.isBlank(value)) {
+            throw new IllegalArgumentException(name + " must not be empty");
+        }
+        if (CollectionUtils.isEmpty(targetList)) {
+            throw new IllegalArgumentException("target list must not be empty");
+        }
+        if (!targetList.contains(value)) {
+            throw new IllegalArgumentException(name + " must be one of " + targetList);
+        }
+    }
 }

polaris-common/polaris-model/src/main/java/com/tencent/polaris/api/utils/IPAddressUtils.java

@@ -20,9 +20,10 @@
 import com.tencent.polaris.logging.LoggerFactory;
 import org.slf4j.Logger;
 
-import java.net.Inet6Address;
-import java.net.InetAddress;
-import java.net.UnknownHostException;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.*;
 
 /**
  * @author Haotian Zhang
@@ -73,4 +74,61 @@ public static boolean checkIpv6Host(String host) {
         }
         return false;
     }
+
+    public static boolean detect(String host, int port, int timeout) {
+        if (!ping(host)) {
+            return connect(host, port, timeout);
+        }
+        return true;
+    }
+
+    public static boolean ping(String host) {
+        try {
+            // 根据操作系统构造不同的ping命令
+            String command;
+            if (System.getProperty("os.name").startsWith("Windows")) {
+                command = "ping -n 1 " + host;
+            } else {
+                command = "ping -c 1 " + host;
+            }
+            // 执行ping命令
+            Process process = Runtime.getRuntime().exec(command);
+            // 读取命令输出
+            BufferedReader reader = new BufferedReader(
+                    new InputStreamReader(process.getInputStream()));
+            String line;
+            while ((line = reader.readLine()) != null) {
+                LOG.trace(line);
+            }
+            // 等待命令执行完成
+            int exitCode = process.waitFor();
+            // 0表示成功，其他值表示失败
+            if (exitCode != 0) {
+                LOG.warn("ping {} with exit code: {}", host, exitCode);
+            } else {
+                LOG.debug("ping {} successfully with exit code: {}", host, exitCode);
+            }
+            return exitCode == 0;
+        } catch (IOException | InterruptedException e) {
+            LOG.warn("ping {} failed: {}", host, e.getMessage());
+        }
+        return false;
+    }
+
+    public static boolean connect(String host, int port, int timeout) {
+        try (Socket socket = new Socket()) {
+            // 设置SO_REUSEADDR为false可以避免重用TIME_WAIT状态的端口
+            socket.setReuseAddress(false);
+            // 设置TCP_NODELAY禁用Nagle算法
+            socket.setTcpNoDelay(true);
+            // 绑定随机端口
+            socket.bind(new InetSocketAddress(0));
+            socket.connect(new InetSocketAddress(host, port), timeout); // 1秒超时
+            LOG.debug("connect {}:{} successfully", host, port);
+            return true;
+        } catch (IOException e) {
+            LOG.warn("connect {}:{} failed: {}", host, port, e.getMessage());
+        }
+        return false;
+    }
 }

polaris-common/polaris-model/src/main/java/com/tencent/polaris/client/pojo/Node.java

@@ -37,6 +37,11 @@ public Node(String host, int port) {
         this.port = port;
     }
 
+    public Node(Node another) {
+        this.host = another.host;
+        this.port = another.port;
+    }
+
     public String getHost() {
         return host;
     }

polaris-common/polaris-model/src/test/java/com/tencent/polaris/api/utils/IPAddressUtilsTest.java

@@ -104,4 +104,9 @@ public void testCheckIpv6Host_WithIpv6MappedIpv4() {
         // 测试IPv6映射的IPv4地址
         assertThat(IPAddressUtils.checkIpv6Host("::ffff:192.168.1.1")).isFalse();
     }
+
+    @Test
+    public void testPing() {
+        assertThat(IPAddressUtils.ping("localhost")).isTrue();
+    }
 }

polaris-common/polaris-model/src/test/java/com/tencent/polaris/client/pojo/NodeTest.java

@@ -0,0 +1,122 @@
+/*
+ * Tencent is pleased to support the open source community by making polaris-java available.
+ *
+ * Copyright (C) 2021 THL A29 Limited, a Tencent company. All rights reserved.
+ *
+ * Licensed under the BSD 3-Clause License (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://opensource.org/licenses/BSD-3-Clause
+ *
+ * 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.tencent.polaris.client.pojo;
+
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Test for {@link Node}.
+ *
+ * @author Haotian Zhang
+ */
+public class NodeTest {
+
+    @Test
+    public void testConstructorAndGetters() {
+        // 测试构造器和getter方法
+        Node node = new Node("localhost", 8080);
+
+        assertThat(node)
+                .extracting(Node::getHost, Node::getPort)
+                .containsExactly("localhost", 8080);
+    }
+
+    @Test
+    public void testCopyConstructor() {
+        // 测试拷贝构造器
+        Node original = new Node("localhost", 8081);
+        Node copy = new Node(original);
+
+        assertThat(copy)
+                .usingRecursiveComparison()
+                .isEqualTo(original);
+    }
+
+    @Test
+    public void testGetHostPort() {
+        // 测试getHostPort方法
+        Node node = new Node("example.com", 443);
+
+        assertThat(node.getHostPort())
+                .isEqualTo("example.com:443")
+                .contains("example.com")
+                .endsWith("443");
+    }
+
+    @Test
+    public void testEqualsAndHashCode() {
+        // 测试equals和hashCode方法
+        Node node1 = new Node("host1", 80);
+        Node node2 = new Node("host1", 80);
+        Node node3 = new Node("host2", 80);
+
+        assertThat(node1)
+                .isEqualTo(node2)
+                .hasSameHashCodeAs(node2)
+                .isNotEqualTo(node3)
+                .isNotEqualTo(null)
+                .isNotEqualTo("string");
+    }
+
+    @Test
+    public void testToString() {
+        // 测试toString方法
+        Node node = new Node("testhost", 9999);
+
+        assertThat(node.toString())
+                .isEqualTo("Node{host='testhost', port=9999}")
+                .contains("testhost")
+                .containsPattern("port=\\d+");
+    }
+
+    @Test
+    public void testIsAnyAddress() {
+        // 测试isAnyAddress方法
+        Node localNode = new Node("0.0.0.0", 8080);
+        Node nonLocalNode = new Node("8.8.8.8", 53);
+        Node invalidNode = new Node("invalid.host.name", 1234);
+
+        assertThat(localNode.isAnyAddress()).isTrue();
+        assertThat(nonLocalNode.isAnyAddress()).isFalse();
+        assertThat(invalidNode.isAnyAddress()).isFalse();
+    }
+
+    @Test
+    public void testIsAnyAddressWithNullHost() {
+        // 测试null主机名的情况
+        Node nullHostNode = new Node(null, 8080);
+
+        assertThat(nullHostNode.isAnyAddress()).isFalse();
+    }
+
+    @Test
+    public void testEdgeCases() {
+        // 测试边界情况
+        Node minPortNode = new Node("host", 0);
+        Node maxPortNode = new Node("host", 65535);
+        Node emptyHostNode = new Node("", 8080);
+
+        assertThat(minPortNode.getPort()).isZero();
+
+        assertThat(maxPortNode.getPort()).isEqualTo(65535);
+
+        assertThat(emptyHostNode.getHost()).isEmpty();
+    }
+}