Add internalComHost and internalComPort config to Nacos. (#5192)

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

@@ -92,6 +92,12 @@ cluster:
   # other configurations
 ```
 
+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 host 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.
+
 ## Nacos
 Set the **cluster/selector** to **nacos** in the yml to enable.
 
@@ -99,4 +105,10 @@ Set the **cluster/selector** to **nacos** in the yml to enable.
 cluster:
   selector: ${SW_CLUSTER:nacos}
   # other configurations
-```
\ No newline at end of file
+```
+
+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 host 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.
\ No newline at end of file

docs/en/setup/backend/configuration-vocabulary.md

@@ -63,6 +63,8 @@ core|default|role|Option values, `Mixed/Receiver/Aggregator`. **Receiver** mode
 | - | Nacos| serviceName| Service name used for SkyWalking cluster. |SW_SERVICE_NAME|SkyWalking_OAP_Cluster|
 | - | - | hostPort| hosts and ports used of Nacos cluster.| SW_CLUSTER_NACOS_HOST_PORT|localhost:8848|
 | - | - | namespace| Namespace used by SkyWalking node coordination.| SW_CLUSTER_NACOS_NAMESPACE|public|
+| - | - | internalComHost| The hostname registered in the Nacos for the internal communication of OAP cluster.| - | -|
+| - | - | internalComPort| The port registered in the Nacos for the internal communication of OAP cluster.| - | -1|
 | storage|elasticsearch| - | ElasticSearch 6 storage implementation | - | - |
 | - | - | nameSpace | Prefix of indexes created and used by SkyWalking. | SW_NAMESPACE | - |
 | - | - | clusterNodes | ElasticSearch cluster nodes for client connection.| SW_STORAGE_ES_CLUSTER_NODES |localhost|

oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosConfig.java

@@ -32,4 +32,10 @@ public class ClusterModuleNacosConfig extends ModuleConfig {
     @Setter
     @Getter
     private String namespace = "public";
+    @Setter
+    @Getter
+    private String internalComHost;
+    @Setter
+    @Getter
+    private int internalComPort = -1;
 }

oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinator.java

@@ -23,6 +23,8 @@ import com.alibaba.nacos.api.naming.NamingService;
 import com.alibaba.nacos.api.naming.pojo.Instance;
 import java.util.ArrayList;
 import java.util.List;
+
+import com.google.common.base.Strings;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
@@ -65,6 +67,9 @@ public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery {
 
     @Override
     public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
+        if (needUsingInternalAddr()) {
+            remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
+        }
         String host = remoteInstance.getAddress().getHost();
         int port = remoteInstance.getAddress().getPort();
         try {
@@ -75,4 +80,8 @@ public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery {
         this.selfAddress = remoteInstance.getAddress();
         TelemetryRelatedContext.INSTANCE.setId(selfAddress.toString());
     }
+
+    private boolean needUsingInternalAddr() {
+        return !Strings.isNullOrEmpty(config.getInternalComHost()) && config.getInternalComPort() > 0;
+    }
 }

oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ITClusterModuleNacosProviderFunctionalTest.java

@@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.cluster.plugin.nacos;
 import com.alibaba.nacos.api.naming.NamingService;
 import java.util.Collections;
 import java.util.List;
+
 import org.apache.skywalking.apm.util.StringUtil;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -62,6 +63,25 @@ public class ITClusterModuleNacosProviderFunctionalTest {
         assertTrue(queryAddress.isSelf());
     }
 
+    @Test
+    public void registerRemoteOfInternal() throws Exception {
+        final String serviceName = "register_remote_internal";
+        ModuleProvider provider = createProvider(serviceName, "127.0.1.2", 1000);
+
+        Address selfAddress = new Address("127.0.0.2", 1000, true);
+        RemoteInstance instance = new RemoteInstance(selfAddress);
+        getClusterRegister(provider).registerRemote(instance);
+
+        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+
+        ClusterModuleNacosConfig config = (ClusterModuleNacosConfig) provider.createConfigBeanIfAbsent();
+        assertEquals(1, remoteInstances.size());
+        Address queryAddress = remoteInstances.get(0).getAddress();
+        assertEquals(config.getInternalComHost(), queryAddress.getHost());
+        assertEquals(config.getInternalComPort(), queryAddress.getPort());
+        assertTrue(queryAddress.isSelf());
+    }
+
     @Test
     public void registerRemoteOfReceiver() throws Exception {
         final String serviceName = "register_remote_receiver";
@@ -151,6 +171,29 @@ public class ITClusterModuleNacosProviderFunctionalTest {
         return provider;
     }
 
+    private ClusterModuleNacosProvider createProvider(String serviceName, String internalComHost,
+                                                       int internalComPort) throws Exception {
+        ClusterModuleNacosProvider provider = new ClusterModuleNacosProvider();
+
+        ClusterModuleNacosConfig config = (ClusterModuleNacosConfig) provider.createConfigBeanIfAbsent();
+
+        config.setHostPort(nacosAddress);
+        config.setServiceName(serviceName);
+
+        if (!StringUtil.isEmpty(internalComHost)) {
+            config.setInternalComHost(internalComHost);
+        }
+
+        if (internalComPort > 0) {
+            config.setInternalComPort(internalComPort);
+        }
+
+        provider.prepare();
+        provider.start();
+        provider.notifyAfterCompleted();
+        return provider;
+    }
+
     private ClusterRegister getClusterRegister(ModuleProvider provider) {
         return provider.getService(ClusterRegister.class);
     }

oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinatorTest.java

@@ -45,6 +45,8 @@ public class NacosCoordinatorTest {
     private Address remoteAddress = new Address("10.0.0.1", 1000, false);
     private Address selfRemoteAddress = new Address("10.0.0.2", 1001, true);
 
+    private Address internalAddress = new Address("10.0.0.3", 1002, false);
+
     private static final String SERVICE_NAME = "test-service";
 
     @Before
@@ -93,6 +95,13 @@ public class NacosCoordinatorTest {
         registerRemote(selfRemoteAddress);
     }
 
+    @Test
+    public void registerRemoteUsingInternal() throws NacosException {
+        nacosConfig.setInternalComHost(internalAddress.getHost());
+        nacosConfig.setInternalComPort(internalAddress.getPort());
+        registerRemote(internalAddress);
+    }
+
     private void validate(Address originArress, RemoteInstance instance) {
         Address instanceAddress = instance.getAddress();
         assertEquals(originArress.getHost(), instanceAddress.getHost());