Support to register different host/port at cluster coordinator (#2244)

* Support to register different host/port at cluster coordinator. Only for Zookeeper and Consul

* Provide document for internalComHost and internalComPort

* Reset port to default.

docs/en/setup/backend/backend-cluster.md

@@ -30,6 +30,22 @@ cluster:
 - `hostPort` is the list of zookeeper servers. Format is `IP1:PORT1,IP2:PORT2,...,IPn:PORTn`
 - `hostPort`, `baseSleepTimeMs` and `maxRetries` are settings of Zookeeper curator client.
 
+In some cases, oap default gRPC host and port in core are not suitable for internal communication among the oap nodes.
+The following setting are provided to set the hot and port manually, based on your own LAN env.
+- internalComHost, the host registered and other oap node use this to communicate with current node.
+- internalComPort, the port registered and other oap node use this to communicate with current node.
+
+```yaml
+zookeeper:
+  nameSpace: ${SW_NAMESPACE:""}
+  hostPort: ${SW_CLUSTER_ZK_HOST_PORT:localhost:2181}
+  #Retry Policy
+  baseSleepTimeMs: ${SW_CLUSTER_ZK_SLEEP_TIME:1000} # initial amount of time to wait between retries
+  maxRetries: ${SW_CLUSTER_ZK_MAX_RETRIES:3} # max number of times to retry
+  internalComHost: 172.10.4.10
+  internalComPort: 11800
+``` 
+
 
 ## Kubernetes
 Require backend cluster are deployed inside kubernetes, guides are in [Deploy in kubernetes](backend-k8s.md).
@@ -56,3 +72,9 @@ cluster:
     # Consul cluster nodes, example: 10.0.0.1:8500,10.0.0.2:8500,10.0.0.3:8500
     hostPort: ${SW_CLUSTER_CONSUL_HOST_PORT:localhost:8500}
 ```
+
+Same as Zookeeper coordinator,
+in some cases, oap default gRPC host and port in core are not suitable for internal communication among the oap nodes.
+The following setting are provided to set the hot and port manually, based on your own LAN env.
+- internalComHost, the host registered and other oap node use this to communicate with current node.
+- internalComPort, the port registered and other oap node use this to communicate with current node.

oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulConfig.java

@@ -27,4 +27,6 @@ import org.apache.skywalking.oap.server.library.module.ModuleConfig;
 class ClusterModuleConsulConfig extends ModuleConfig {
     @Setter @Getter private String serviceName;
     @Setter @Getter private String hostPort;
+    @Setter @Getter private String internalComHost;
+    @Setter @Getter private int internalComPort = -1;
 }

oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProvider.java

@@ -80,7 +80,7 @@ public class ClusterModuleConsulProvider extends ModuleProvider {
             throw new ModuleStartException(e.getMessage(), e);
         }
 
-        ConsulCoordinator coordinator = new ConsulCoordinator(client, config.getServiceName());
+        ConsulCoordinator coordinator = new ConsulCoordinator(config, client);
         this.registerServiceImplementation(ClusterRegister.class, coordinator);
         this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
     }

oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java

@@ -35,11 +35,13 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery {
 
     private final Consul client;
     private final String serviceName;
+    private final ClusterModuleConsulConfig config;
     private volatile Address selfAddress;
 
-    public ConsulCoordinator(Consul client, String serviceName) {
+    public ConsulCoordinator(ClusterModuleConsulConfig config, Consul client) {
+        this.config = config;
         this.client = client;
-        this.serviceName = serviceName;
+        this.serviceName = config.getServiceName();
     }
 
     @Override public List<RemoteInstance> queryRemoteNodes() {
@@ -66,6 +68,10 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery {
     }
 
     @Override public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
+        if (needUsingInternalAddr()) {
+            remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
+        }
+
         AgentClient agentClient = client.agentClient();
 
         this.selfAddress = remoteInstance.getAddress();
@@ -81,4 +87,8 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery {
 
         agentClient.register(registration);
     }
+
+    private boolean needUsingInternalAddr() {
+        return !Strings.isNullOrEmpty(config.getInternalComHost()) && config.getInternalComPort() > 0;
+    }
 }

oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperConfig.java

@@ -31,6 +31,8 @@ class ClusterModuleZookeeperConfig extends ModuleConfig {
     private String hostPort;
     private int baseSleepTimeMs;
     private int maxRetries;
+    @Setter @Getter private String internalComHost;
+    @Setter @Getter private int internalComPort = -1;
 
     public String getHostPort() {
         return Strings.isNullOrEmpty(hostPort) ? "localhost:2181" : hostPort;

oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperProvider.java

@@ -79,7 +79,7 @@ public class ClusterModuleZookeeperProvider extends ModuleProvider {
             throw new ModuleStartException(e.getMessage(), e);
         }
 
-        ZookeeperCoordinator coordinator = new ZookeeperCoordinator(serviceDiscovery);
+        ZookeeperCoordinator coordinator = new ZookeeperCoordinator(config, serviceDiscovery);
         this.registerServiceImplementation(ClusterRegister.class, coordinator);
         this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
     }

oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.zookeeper;
 
+import com.google.common.base.Strings;
 import java.util.*;
 import org.apache.curator.x.discovery.*;
 import org.apache.skywalking.oap.server.core.cluster.*;
@@ -31,17 +32,22 @@ import org.slf4j.*;
 public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery {
     private static final Logger logger = LoggerFactory.getLogger(ZookeeperCoordinator.class);
 
+    private final ClusterModuleZookeeperConfig config;
     private final ServiceDiscovery<RemoteInstance> serviceDiscovery;
     private volatile ServiceCache<RemoteInstance> serviceCache;
     private volatile Address selfAddress;
 
-    ZookeeperCoordinator(ServiceDiscovery<RemoteInstance> serviceDiscovery) {
+    ZookeeperCoordinator(ClusterModuleZookeeperConfig config, ServiceDiscovery<RemoteInstance> serviceDiscovery) {
+        this.config = config;
         this.serviceDiscovery = serviceDiscovery;
     }
 
     @Override public synchronized void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
         try {
             String remoteNamePath = "remote";
+            if (needUsingInternalAddr()) {
+                remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
+            }
 
             ServiceInstance<RemoteInstance> thisInstance = ServiceInstance.<RemoteInstance>builder()
                 .name(remoteNamePath)
@@ -83,4 +89,8 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery
         }
         return remoteInstanceDetails;
     }
+
+    private boolean needUsingInternalAddr() {
+        return !Strings.isNullOrEmpty(config.getInternalComHost()) && config.getInternalComPort() > 0;
+    }
 }