Repository: apache/skywalking (language: python)

Issue:
[Feature] Support listeners for the `RemoteClients` change.
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar feature requirement.


### Description

   In the cluster mode, OAP instances send metrics to each other according to the `remote-hash code` to do L2 aggregation and alarm calculations for balancing​. Some of the calculations leverage the previously received metrics in the memory cache and incoming metrics. So when the OAP cluster scales,  the metrics might lose their cache and impact the calculations.

   To avoid this situation, I think we should support listeners for the `RemoteClients` change, then the stream workers can be notified and reload previous metrics from storage for calculation.


### Use case

1. Provide listeners' interface for notification and a registration mechanism.
2. When the `RemoteClients` changes, notice all listeners.

### Related issues

https://github.com/apache/skywalking/issues/10051

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

[Bug] Cluster coordinator should be responsive to instance up/down
### Search before asking

- [X] I had searched in the [issues](https://github.com/apache/skywalking/issues?q=is%3Aissue) and found no similar issues.


### Apache SkyWalking Component

OAP server (apache/skywalking)

### What happened

Currently the cluster coordinator is using polling strategy to fetch the OAP instances, and the interval is 5s:

https://github.com/apache/skywalking/blob/34cfafe398e80ca1a7299e1243827937c1a691dd/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java#L96

There is a common case when some of the OAP instances restarted/shutdown, the living ones will still try to connect to those dead instances in the interval, causing error logs like this:

```
2022-10-18 07:33:28,297 - org.apache.skywalking.oap.server.core.remote.client.GRPCRemoteClient -31025 [grpc-default-executor-0] ERROR [] - UNAVAILABLE: io exception
io.grpc.StatusRuntimeException: UNAVAILABLE: io exception
	at io.grpc.Status.asRuntimeException(Status.java:535) ~[grpc-api-1.49.0.jar:1.49.0]
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:487) [grpc-stub-1.49.0.jar:1.49.0]
	at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:563) [grpc-core-1.49.0.jar:1.49.0]
	at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:70) [grpc-core-1.49.0.jar:1.49.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:744) [grpc-core-1.49.0.jar:1.49.0]
	at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:723) [grpc-core-1.49.0.jar:1.49.0]
	at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) [grpc-core-1.49.0.jar:1.49.0]
	at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) [grpc-core-1.49.0.jar:1.49.0]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?]
	at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: finishConnect(..) failed: Connection refused: /10.92.1.254:31800
Caused by: java.net.ConnectException: finishConnect(..) failed: Connection refused
	at io.netty.channel.unix.Errors.newConnectException0(Errors.java:155) ~[netty-transport-native-unix-common-4.1.81.Final-linux-x86_64.jar:4.1.81.Final]
	at io.netty.channel.unix.Errors.handleConnectErrno(Errors.java:128) ~[netty-transport-native-unix-common-4.1.81.Final-linux-x86_64.jar:4.1.81.Final]
	at io.netty.channel.unix.Socket.finishConnect(Socket.java:359) ~[netty-transport-native-unix-common-4.1.81.Final-linux-x86_64.jar:4.1.81.Final]
	at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.doFinishConnect(AbstractEpollChannel.java:710) ~[netty-transport-classes-epoll-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.finishConnect(AbstractEpollChannel.java:687) ~[netty-transport-classes-epoll-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.channel.epoll.AbstractEpollChannel$AbstractEpollUnsafe.epollOutReady(AbstractEpollChannel.java:567) ~[netty-transport-classes-epoll-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.channel.epoll.EpollEventLoop.processReady(EpollEventLoop.java:489) ~[netty-transport-classes-epoll-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:397) ~[netty-transport-classes-epoll-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) ~[netty-common-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) ~[netty-common-4.1.81.Final.jar:4.1.81.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.81.Final.jar:4.1.81.Final]
	... 1 more
```

For most service-discover service like Kubernetes they should have a watcher/listener mode that clients can register to the changes of instances and get notified right after the instances' states changed, it's also easier to wrap the polling mechanism and expose as a listener mechanism for those that doesn't support listener mode.

### What you expected to happen

The OAP should be more responsive to the changes of the cluster instances states. Reducing unnecessary errors.

### How to reproduce

Start OAP in Kubernetes cluster mode, set the replicas to more than 1, after the OAP is ready, restart one of the OAP instance, and observe the logs in the other living OAP instances, there should be error logs as posted above.

### Anything else

_No response_

### Are you willing to submit PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)



Discussion / hints:
This is duplicate of https://github.com/apache/skywalking/issues/9814 and I think @fgksgf is already working on this
Let's see how @fgksgf would do. I think we need more work after this mechanism is ready.
Is there an ETA for review?
> Let's see how @fgksgf would do. I think we need more work after this mechanism is ready.
> Is there an ETA for review?

Before https://github.com/apache/skywalking/issues/10051 this issue is a "nice to have" enhancement so we don't set an ETA, but if it impacts the metrics in https://github.com/apache/skywalking/issues/10051, we could prioritize this.
#10051 doesn't rely on this as a hard requirement. Alerting enhancement requires this more.
I think #9814  is not exactly the same. #9814 is trying to fix the delay between the  OAP fetched `RemoteClients` and the real instance status.  But this issue wants to resolve the other OAP module services doesn't know the `RemoteClients` is changed.
> I think #9814  is not exactly the same. #9814 is trying to fix the delay between the  OAP fetched `RemoteClients` and the real instance status.  But this issue wants to resolve the other OAP module services doesn't know the `RemoteClients` is changed.

If you read carefully they are exactly the same essentially. 9814 is to make the coordinator module know remote clients are changed. This can be shared to other modules and let them know the remote clients changed too. 
@fgksgf @wankai123 @kezhenxu94 Please ref to https://github.com/apache/skywalking/pull/10083, when we build a notification callback mechanism, especially for cluster status, we could build this callback in the core module through new **ServerStatusService**.
Assigning to @fgksgf as well since I discussed with him long time ago
Just contacted @fgksgf and he thinks he has very limited time to work on this so I'm closing this in favor of https://github.com/apache/skywalking/issues/10076 

Fix patch:
diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE
index 16880c233d84..9d20c837294d 100644
--- a/dist-material/release-docs/LICENSE
+++ b/dist-material/release-docs/LICENSE
@@ -222,9 +222,9 @@ The text of each license is the standard Apache 2.0 license.
     https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-core/2.13.3 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind/2.13.4.1 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind/2.13.3 Apache-2.0
-    https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-guava/2.9.10 Apache-2.0
+    https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-guava/2.12.0 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-guava/2.9.5 Apache-2.0
-    https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-jdk8/2.9.10 Apache-2.0
+    https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-jdk8/2.12.0 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-jdk8/2.13.4 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.datatype/jackson-datatype-jdk8/2.9.5 Apache-2.0
     https://mvnrepository.com/artifact/com.fasterxml.jackson.module/jackson-module-kotlin/2.13.4 Apache-2.0
@@ -248,16 +248,16 @@ The text of each license is the standard Apache 2.0 license.
     https://mvnrepository.com/artifact/com.linecorp.armeria/armeria-graphql/1.18.0 Apache-2.0
     https://mvnrepository.com/artifact/com.linecorp.armeria/armeria-graphql-protocol/1.18.0 Apache-2.0
     https://mvnrepository.com/artifact/com.linecorp.armeria/armeria-protobuf/1.18.0 Apache-2.0
-    https://mvnrepository.com/artifact/com.orbitz.consul/consul-client/1.4.2 Apache-2.0
+    https://mvnrepository.com/artifact/com.orbitz.consul/consul-client/1.5.3 Apache-2.0
     https://mvnrepository.com/artifact/com.orbitz.consul/consul-client/1.2.6 Apache-2.0
     https://mvnrepository.com/artifact/com.squareup.okhttp3/logging-interceptor/4.9.2 Apache-2.0
     https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp/3.14.9 Apache-2.0
-    https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp/3.12.2 Apache-2.0
+    https://mvnrepository.com/artifact/com.squareup.okhttp3/okhttp/4.9.0 Apache-2.0
     https://mvnrepository.com/artifact/com.squareup.okio/okio/1.17.2 Apache-2.0
-    https://mvnrepository.com/artifact/com.squareup.okio/okio/1.15.0 Apache-2.0
-    https://mvnrepository.com/artifact/com.squareup.retrofit2/converter-jackson/2.5.0 Apache-2.0
+    https://mvnrepository.com/artifact/com.squareup.okio/okio/2.8.0 Apache-2.0
+    https://mvnrepository.com/artifact/com.squareup.retrofit2/converter-jackson/2.9.0 Apache-2.0
     https://mvnrepository.com/artifact/com.squareup.retrofit2/converter-jackson/2.3.0 Apache-2.0
-    https://mvnrepository.com/artifact/com.squareup.retrofit2/retrofit/2.5.0 Apache-2.0
+    https://mvnrepository.com/artifact/com.squareup.retrofit2/retrofit/2.9.0 Apache-2.0
     https://mvnrepository.com/artifact/com.squareup.retrofit2/retrofit/2.3.0 Apache-2.0
     https://mvnrepository.com/artifact/com.zaxxer/HikariCP/3.1.0 Apache-2.0
     https://mvnrepository.com/artifact/commons-beanutils/commons-beanutils/1.9.4 Apache-2.0
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
index ba5cd050067a..a045c616996e 100644
--- a/docs/en/changes/changes.md
+++ b/docs/en/changes/changes.md
@@ -55,6 +55,7 @@
 * Register ValueColumn as Tag for Record in BanyanDB storage plugin.
 * Bump up Netty to 4.1.86.
 * Remove unnecessary additional columns when storage is in logical sharding mode.
+* The cluster coordinator support watch mechanism for notifying `RemoteClientManager` and `ServerStatusService`.
 * Fix ServiceMeshServiceDispatcher overwrite ServiceDispatcher debug file when open SW_OAL_ENGINE_DEBUG.
 
 #### UI
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/pom.xml b/oap-server/server-cluster-plugin/cluster-consul-plugin/pom.xml
index 16ce31b48564..57c48e6b5854 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/pom.xml
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/pom.xml
@@ -37,7 +37,7 @@
         <dependency>
             <groupId>com.orbitz.consul</groupId>
             <artifactId>consul-client</artifactId>
-            <version>1.4.2</version>
+            <version>1.5.3</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.google.guava</groupId>
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulConfig.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulConfig.java
index 6f2cd9a72e9f..69612650cd91 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulConfig.java
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulConfig.java
@@ -22,7 +22,7 @@
 import lombok.Setter;
 import org.apache.skywalking.oap.server.library.module.ModuleConfig;
 
-class ClusterModuleConsulConfig extends ModuleConfig {
+public class ClusterModuleConsulConfig extends ModuleConfig {
     @Setter
     @Getter
     private String serviceName;
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProvider.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProvider.java
index b596daf197c0..50a4e6c7306f 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProvider.java
@@ -25,6 +25,7 @@
 import java.util.List;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -97,6 +98,7 @@ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
         ConsulCoordinator coordinator = new ConsulCoordinator(getManager(), config, client);
         this.registerServiceImplementation(ClusterRegister.class, coordinator);
         this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
+        this.registerServiceImplementation(ClusterCoordinator.class, coordinator);
     }
 
     @Override
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java
index 777441a6b182..8ba79589e3b8 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ConsulCoordinator.java
@@ -22,15 +22,19 @@
 import com.orbitz.consul.AgentClient;
 import com.orbitz.consul.Consul;
 import com.orbitz.consul.HealthClient;
+import com.orbitz.consul.cache.ConsulCache;
+import com.orbitz.consul.cache.ServiceHealthCache;
+import com.orbitz.consul.cache.ServiceHealthKey;
 import com.orbitz.consul.model.agent.ImmutableRegistration;
 import com.orbitz.consul.model.agent.Registration;
 import com.orbitz.consul.model.health.ServiceHealth;
+import com.orbitz.consul.option.QueryOptions;
 import java.util.ArrayList;
 import java.util.List;
-
+import java.util.Map;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterHealthStatus;
-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.OAPNodeChecker;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.cluster.ServiceQueryException;
@@ -38,12 +42,14 @@
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
 import org.apache.skywalking.oap.server.library.util.CollectionUtils;
+import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
-public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery {
+@Slf4j
+public class ConsulCoordinator extends ClusterCoordinator {
 
     private final ModuleDefineHolder manager;
     private final Consul client;
@@ -52,7 +58,9 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery {
     private volatile Address selfAddress;
     private HealthCheckMetrics healthChecker;
 
-    public ConsulCoordinator(final ModuleDefineHolder manager, final ClusterModuleConsulConfig config, final Consul client) {
+    public ConsulCoordinator(final ModuleDefineHolder manager,
+                             final ClusterModuleConsulConfig config,
+                             final Consul client) {
         this.manager = manager;
         this.config = config;
         this.client = client;
@@ -63,7 +71,6 @@ public ConsulCoordinator(final ModuleDefineHolder manager, final ClusterModuleCo
     public List<RemoteInstance> queryRemoteNodes() {
         List<RemoteInstance> remoteInstances = new ArrayList<>();
         try {
-            initHealthChecker();
             HealthClient healthClient = client.healthClient();
             // Discover only "passing" nodes
             List<ServiceHealth> nodes = healthClient.getHealthyServiceInstances(serviceName).getResponse();
@@ -88,33 +95,38 @@ public List<RemoteInstance> queryRemoteNodes() {
             healthChecker.unHealth(e);
             throw new ServiceQueryException(e.getMessage());
         }
+        if (log.isDebugEnabled()) {
+            remoteInstances.forEach(instance -> log.debug("Cosule cluster instance: {}", instance));
+        }
         return remoteInstances;
     }
 
     @Override
     public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
         if (needUsingInternalAddr()) {
-            remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
+            remoteInstance = new RemoteInstance(
+                new Address(config.getInternalComHost(), config.getInternalComPort(), true));
         }
+        this.selfAddress = remoteInstance.getAddress();
         try {
-            initHealthChecker();
             AgentClient agentClient = client.agentClient();
-
-            this.selfAddress = remoteInstance.getAddress();
-
             Registration registration = ImmutableRegistration.builder()
-                    .id(remoteInstance.getAddress().toString())
-                    .name(serviceName)
-                    .address(remoteInstance.getAddress().getHost())
-                    .port(remoteInstance.getAddress().getPort())
-                    .check(Registration.RegCheck.grpc(remoteInstance.getAddress()
-                            .getHost() + ":" + remoteInstance
-                            .getAddress()
-                            .getPort(), 5)) // registers with a TTL of 5 seconds
-                    .build();
+                                                             .id(remoteInstance.getAddress().toString())
+                                                             .name(serviceName)
+                                                             .address(remoteInstance.getAddress().getHost())
+                                                             .port(remoteInstance.getAddress().getPort())
+                                                             .check(Registration.RegCheck.grpc(
+                                                                 remoteInstance.getAddress()
+                                                                               .getHost() + ":" + remoteInstance
+                                                                     .getAddress()
+                                                                     .getPort(),
+                                                                 5
+                                                             )) // registers with a TTL of 5 seconds
+                                                             .build();
 
             agentClient.register(registration);
             healthChecker.health();
+
         } catch (Throwable e) {
             healthChecker.unHealth(e);
             throw new ServiceRegisterException(e.getMessage());
@@ -123,12 +135,69 @@ public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegister
 
     private void initHealthChecker() {
         if (healthChecker == null) {
-            MetricsCreator metricCreator = manager.find(TelemetryModule.NAME).provider().getService(MetricsCreator.class);
-            healthChecker = metricCreator.createHealthCheckerGauge("cluster_consul", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
+            MetricsCreator metricCreator = manager.find(TelemetryModule.NAME)
+                                                  .provider()
+                                                  .getService(MetricsCreator.class);
+            healthChecker = metricCreator.createHealthCheckerGauge(
+                "cluster_consul", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
         }
     }
 
     private boolean needUsingInternalAddr() {
         return !Strings.isNullOrEmpty(config.getInternalComHost()) && config.getInternalComPort() > 0;
     }
+
+    private RemoteInstance buildRemoteInstance(String host, int port) {
+        Address address = new Address(host, port, false);
+        if (address.equals(selfAddress)) {
+            address.setSelf(true);
+        }
+        return new RemoteInstance(address);
+    }
+
+    private void checkHealth(List<RemoteInstance> remoteInstances) {
+        ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
+        if (healthStatus.isHealth()) {
+            this.healthChecker.health();
+        } else {
+            this.healthChecker.unHealth(healthStatus.getReason());
+        }
+    }
+
+    @Override
+    public void start() {
+        initHealthChecker();
+        ServiceHealthCache svHealth = ServiceHealthCache.newCache(client.healthClient(), serviceName, true,
+                                                                  QueryOptions.BLANK, 5);
+        svHealth.addListener(new ConsulEventListener());
+        svHealth.start();
+    }
+
+    /**
+     * Notice: If the consul version > v1.10.0, the `consul-client ConsulCache` will throw error response:
+     * "com.orbitz.consul.ConsulException: Consul cluster has no elected leader" and fails to retrieve data.
+     * This is a known issue but doesn't release yet, can refer to: https://github.com/rickfast/consul-client/pull/456
+     */
+    class ConsulEventListener implements ConsulCache.Listener<ServiceHealthKey, ServiceHealth> {
+        @Override
+        public void notify(final Map<ServiceHealthKey, ServiceHealth> newValues) {
+            try {
+                if (newValues.size() > 0) {
+                    List<RemoteInstance> remoteInstances = new ArrayList<>(newValues.size());
+                    newValues.values().forEach(serviceHealth -> {
+                        if (StringUtil.isNotBlank(serviceHealth.getService().getAddress())) {
+                            RemoteInstance remoteInstance = buildRemoteInstance(
+                                serviceHealth.getService().getAddress(), serviceHealth.getService().getPort());
+                            remoteInstances.add(remoteInstance);
+                        }
+                    });
+                    checkHealth(remoteInstances);
+                    notifyWatchers(remoteInstances);
+                }
+            } catch (Throwable e) {
+                healthChecker.unHealth(e);
+                log.error("Failed to notify and update remote instances.", e);
+            }
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ClusterModuleEtcdProvider.java b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ClusterModuleEtcdProvider.java
index c7510a69db5e..7deab5a25a90 100644
--- a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ClusterModuleEtcdProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ClusterModuleEtcdProvider.java
@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.cluster.plugin.etcd;
 
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -65,6 +66,7 @@ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
             EtcdCoordinator coordinator = new EtcdCoordinator(getManager(), config);
             this.registerServiceImplementation(ClusterRegister.class, coordinator);
             this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
+            this.registerServiceImplementation(ClusterCoordinator.class, coordinator);
         } catch (Exception e) {
             throw new ModuleStartException("Failed to start ETCD coordinator.", e);
         }
diff --git a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/EtcdCoordinator.java b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/EtcdCoordinator.java
index cee466a07996..382e3e93d9f0 100644
--- a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/EtcdCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/EtcdCoordinator.java
@@ -25,19 +25,21 @@
 import io.etcd.jetcd.ClientBuilder;
 import io.etcd.jetcd.KV;
 import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Watch;
 import io.etcd.jetcd.kv.GetResponse;
 import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
 import io.etcd.jetcd.options.GetOption;
 import io.etcd.jetcd.options.PutOption;
+import io.etcd.jetcd.options.WatchOption;
+import io.etcd.jetcd.watch.WatchResponse;
 import io.grpc.stub.StreamObserver;
 import java.nio.charset.Charset;
 import java.util.ArrayList;
 import java.util.List;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.core.cluster.ClusterHealthStatus;
-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.OAPNodeChecker;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.cluster.ServiceRegisterException;
@@ -50,7 +52,7 @@
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
 @Slf4j
-public class EtcdCoordinator implements ClusterRegister, ClusterNodesQuery {
+public class EtcdCoordinator extends ClusterCoordinator {
     private static final Gson GSON = new Gson().newBuilder().create();
     private final ModuleDefineHolder manager;
     private final ClusterModuleEtcdConfig config;
@@ -91,8 +93,6 @@ public EtcdCoordinator(final ModuleDefineHolder manager,
     public List<RemoteInstance> queryRemoteNodes() {
         List<RemoteInstance> remoteInstances = new ArrayList<>();
         try {
-            initHealthChecker();
-
             final KV kvClient = client.getKVClient();
             final GetResponse response = kvClient.get(
                 serviceNameBS,
@@ -121,6 +121,9 @@ public List<RemoteInstance> queryRemoteNodes() {
             healthChecker.unHealth(e);
             throw new RuntimeException(e);
         }
+        if (log.isDebugEnabled()) {
+            remoteInstances.forEach(instance -> log.debug("Etcd cluster instance: {}", instance));
+        }
         return remoteInstances;
     }
 
@@ -137,8 +140,6 @@ public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegister
                                                                 .port(selfAddress.getPort())
                                                                 .build();
         try {
-            initHealthChecker();
-
             final Lease leaseClient = client.getLeaseClient();
             final long leaseID = leaseClient.grant(30L).get().getID();
 
@@ -198,4 +199,41 @@ private void initHealthChecker() {
                 "cluster_etcd", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
         }
     }
+
+    @Override
+    public void start() {
+        initHealthChecker();
+        this.client.getWatchClient().watch(
+            serviceNameBS, WatchOption.newBuilder().withPrefix(serviceNameBS).build(), new EtcdEventListener());
+    }
+
+    class EtcdEventListener implements Watch.Listener {
+        @Override
+        public void onNext(final WatchResponse response) {
+            response.getEvents().forEach(event -> {
+                switch (event.getEventType()) {
+                    case DELETE:
+                    case PUT:
+                        if (log.isDebugEnabled()) {
+                            String key = event.getKeyValue().getKey().toString(Charset.defaultCharset());
+                            log.debug("{}: key = {}}", event.getEventType().name(), key);
+                        }
+                        notifyWatchers(queryRemoteNodes());
+                        break;
+                    default:
+                        break;
+                }
+            });
+        }
+
+        @Override
+        public void onError(final Throwable throwable) {
+            log.error("Failed to notify RemoteInstances update.", throwable);
+            healthChecker.unHealth(throwable);
+        }
+
+        @Override
+        public void onCompleted() {
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/pom.xml b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/pom.xml
index 05d3b4798f7e..7aefc876b6b5 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/pom.xml
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/pom.xml
@@ -44,5 +44,10 @@
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>library-kubernetes-support</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
-</project>
\ No newline at end of file
+</project>
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/ClusterModuleKubernetesProvider.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/ClusterModuleKubernetesProvider.java
index 5b33c78270e1..0169ccd282de 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/ClusterModuleKubernetesProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/ClusterModuleKubernetesProvider.java
@@ -19,6 +19,7 @@
 package org.apache.skywalking.oap.server.cluster.plugin.kubernetes;
 
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -64,11 +65,11 @@ public void prepare() throws ServiceNotProvidedException {
         coordinator = new KubernetesCoordinator(getManager(), config);
         this.registerServiceImplementation(ClusterRegister.class, coordinator);
         this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
+        this.registerServiceImplementation(ClusterCoordinator.class, coordinator);
     }
 
     @Override
     public void start() {
-        NamespacedPodListInformer.INFORMER.init(config);
     }
 
     @Override
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
index 60ee79b6f7f5..46ec16811ed4 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
@@ -18,19 +18,25 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.kubernetes;
 
+import io.kubernetes.client.informer.EventType;
+import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodStatus;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
+import org.apache.skywalking.oap.server.core.cluster.ClusterHealthStatus;
+import org.apache.skywalking.oap.server.core.cluster.OAPNodeChecker;
+import org.apache.skywalking.oap.server.core.cluster.ServiceQueryException;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.core.CoreModule;
-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;
-import org.apache.skywalking.oap.server.core.cluster.ServiceQueryException;
 import org.apache.skywalking.oap.server.core.cluster.ServiceRegisterException;
 import org.apache.skywalking.oap.server.core.config.ConfigService;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
@@ -40,35 +46,44 @@
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
+import static io.kubernetes.client.informer.EventType.ADDED;
+import static io.kubernetes.client.informer.EventType.DELETED;
+import static io.kubernetes.client.informer.EventType.MODIFIED;
+
 /**
  * Read collector pod info from api-server of kubernetes, then using all containerIp list to construct the list of
  * {@link RemoteInstance}.
  */
 @Slf4j
-public class KubernetesCoordinator implements ClusterRegister, ClusterNodesQuery {
+public class KubernetesCoordinator extends ClusterCoordinator {
 
     private final ModuleDefineHolder manager;
     private final String uid;
     private volatile int port = -1;
     private HealthCheckMetrics healthChecker;
+    private ClusterModuleKubernetesConfig config;
+    private final Map<String, RemoteInstance> remoteInstanceMap;
+    private volatile List<String> latestInstances;
 
     public KubernetesCoordinator(final ModuleDefineHolder manager,
                                  final ClusterModuleKubernetesConfig config) {
         this.uid = new UidEnvSupplier(config.getUidEnvName()).get();
         this.manager = manager;
+        this.config = config;
+        this.remoteInstanceMap = new ConcurrentHashMap<>(20);
+        this.latestInstances = new ArrayList<>(20);
     }
 
     @Override
     public List<RemoteInstance> queryRemoteNodes() {
         try {
-            initHealthChecker();
             List<V1Pod> pods = NamespacedPodListInformer.INFORMER.listPods().orElseGet(this::selfPod);
             if (log.isDebugEnabled()) {
                 List<String> uidList = pods
                     .stream()
                     .map(item -> item.getMetadata().getUid())
                     .collect(Collectors.toList());
-                log.debug("[kubernetes cluster pods uid list]:{}", uidList.toString());
+                log.debug("[kubernetes cluster pods uid list]:{}", uidList);
             }
             if (port == -1) {
                 port = manager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort();
@@ -80,6 +95,10 @@ public List<RemoteInstance> queryRemoteNodes() {
                         new Address(pod.getStatus().getPodIP(), port, pod.getMetadata().getUid().equals(uid))))
                     .collect(Collectors.toList());
             healthChecker.health();
+            this.latestInstances = remoteInstances.stream().map(it -> it.getAddress().toString()).collect(Collectors.toList());
+            if (log.isDebugEnabled()) {
+                remoteInstances.forEach(instance -> log.debug("kubernetes cluster instance: {}", instance));
+            }
             return remoteInstances;
         } catch (Throwable e) {
             healthChecker.unHealth(e);
@@ -90,7 +109,6 @@ public List<RemoteInstance> queryRemoteNodes() {
     @Override
     public void registerRemote(final RemoteInstance remoteInstance) throws ServiceRegisterException {
         try {
-            initHealthChecker();
             this.port = remoteInstance.getAddress().getPort();
             healthChecker.health();
         } catch (Throwable e) {
@@ -117,4 +135,83 @@ private List<V1Pod> selfPod() {
         v1Pod.getStatus().setPodIP("127.0.0.1");
         return Collections.singletonList(v1Pod);
     }
+
+    @Override
+    public void start() {
+        initHealthChecker();
+        NamespacedPodListInformer.INFORMER.init(config, new K8sResourceEventHandler());
+    }
+
+    class K8sResourceEventHandler implements ResourceEventHandler<V1Pod> {
+
+        @Override
+        public void onAdd(final V1Pod obj) {
+            updateRemoteInstances(obj, ADDED);
+        }
+
+        @Override
+        public void onUpdate(final V1Pod oldObj, final V1Pod newObj) {
+            updateRemoteInstances(newObj, MODIFIED);
+        }
+
+        @Override
+        public void onDelete(final V1Pod obj, final boolean deletedFinalStateUnknown) {
+            updateRemoteInstances(obj, DELETED);
+        }
+    }
+
+    /**
+     * When a remote instance up/off line, will receive multi event according to the pod status.
+     * To avoid notify the watchers too frequency, here use a `remoteInstanceMap` to cache them.
+     * Only notify watchers once when the instances changed.
+     */
+    private void updateRemoteInstances(V1Pod pod, EventType event) {
+        try {
+            initHealthChecker();
+            if (StringUtil.isNotBlank(pod.getStatus().getPodIP())) {
+                if (port == -1) {
+                    port = manager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort();
+                }
+
+                RemoteInstance remoteInstance = new RemoteInstance(
+                    new Address(pod.getStatus().getPodIP(), this.port, pod.getMetadata().getUid().equals(uid)));
+                switch (event) {
+                    case ADDED:
+                    case MODIFIED:
+                        if ("Running".equalsIgnoreCase(pod.getStatus().getPhase())) {
+                            this.remoteInstanceMap.put(remoteInstance.getAddress().toString(), remoteInstance);
+                        }
+                        break;
+                    case DELETED:
+                        this.remoteInstanceMap.remove(remoteInstance.getAddress().toString());
+                        break;
+                    default:
+                        return;
+                }
+                updateRemoteInstances();
+            }
+        } catch (Throwable e) {
+            healthChecker.unHealth(e);
+            log.error("Failed to notify RemoteInstances update.", e);
+        }
+    }
+
+    private void updateRemoteInstances() {
+        List<String> updatedInstances = new ArrayList<>(this.remoteInstanceMap.keySet());
+        if (this.latestInstances.size() != updatedInstances.size() || !this.latestInstances.containsAll(updatedInstances)) {
+            List<RemoteInstance> remoteInstances = new ArrayList<>(this.remoteInstanceMap.values());
+            this.latestInstances = updatedInstances;
+            checkHealth(remoteInstances);
+            notifyWatchers(remoteInstances);
+        }
+    }
+
+    private void checkHealth(List<RemoteInstance> remoteInstances) {
+        ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
+        if (healthStatus.isHealth()) {
+            this.healthChecker.health();
+        } else {
+            this.healthChecker.unHealth(healthStatus.getReason());
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/NamespacedPodListInformer.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/NamespacedPodListInformer.java
index 9c3d2edc9ac1..265a18f219ce 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/NamespacedPodListInformer.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/NamespacedPodListInformer.java
@@ -18,25 +18,22 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.kubernetes;
 
+import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.informer.SharedIndexInformer;
 import io.kubernetes.client.informer.SharedInformerFactory;
 import io.kubernetes.client.informer.cache.Lister;
-import io.kubernetes.client.openapi.ApiClient;
-import io.kubernetes.client.openapi.Configuration;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodList;
-import io.kubernetes.client.util.Config;
 import java.io.IOException;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
-
+import org.apache.skywalking.library.kubernetes.KubernetesClient;
 import static java.util.Objects.isNull;
 
 @Slf4j
@@ -65,21 +62,18 @@ public enum NamespacedPodListInformer {
         }));
     }
 
-    public synchronized void init(ClusterModuleKubernetesConfig podConfig) {
+    public synchronized void init(ClusterModuleKubernetesConfig podConfig, ResourceEventHandler<V1Pod> eventHandler) {
 
         try {
-            doStartPodInformer(podConfig);
+            doStartPodInformer(podConfig, eventHandler);
         } catch (IOException e) {
             log.error("cannot connect with api server in kubernetes", e);
         }
     }
 
-    private void doStartPodInformer(ClusterModuleKubernetesConfig podConfig) throws IOException {
-
-        ApiClient apiClient = Config.defaultClient();
-        apiClient.setHttpClient(apiClient.getHttpClient().newBuilder().readTimeout(0, TimeUnit.SECONDS).build());
-        Configuration.setDefaultApiClient(apiClient);
-        CoreV1Api coreV1Api = new CoreV1Api(apiClient);
+    private void doStartPodInformer(ClusterModuleKubernetesConfig podConfig, ResourceEventHandler<V1Pod> eventHandler) throws IOException {
+        KubernetesClient.setDefault();
+        CoreV1Api coreV1Api = new CoreV1Api();
         factory = new SharedInformerFactory(executorService);
 
         SharedIndexInformer<V1Pod> podSharedIndexInformer = factory.sharedIndexInformerFor(
@@ -91,6 +85,8 @@ private void doStartPodInformer(ClusterModuleKubernetesConfig podConfig) throws
             V1Pod.class, V1PodList.class
         );
 
+        podSharedIndexInformer.addEventHandler(eventHandler);
+
         factory.startAllRegisteredInformers();
         podLister = new Lister<>(podSharedIndexInformer.getIndexer());
     }
diff --git a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosProvider.java b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosProvider.java
index 5b685d7925de..47019f4f9256 100644
--- a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosProvider.java
@@ -23,6 +23,7 @@
 import com.alibaba.nacos.api.naming.NamingService;
 import java.util.Properties;
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -83,12 +84,13 @@ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
                 properties.put(PropertyKeyConst.SECRET_KEY, config.getSecretKey());
             }
             namingService = NamingFactory.createNamingService(properties);
+            NacosCoordinator coordinator = new NacosCoordinator(getManager(), namingService, config);
+            this.registerServiceImplementation(ClusterRegister.class, coordinator);
+            this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
+            this.registerServiceImplementation(ClusterCoordinator.class, coordinator);
         } catch (Exception e) {
             throw new ModuleStartException(e.getMessage(), e);
         }
-        NacosCoordinator coordinator = new NacosCoordinator(getManager(), namingService, config);
-        this.registerServiceImplementation(ClusterRegister.class, coordinator);
-        this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
     }
 
     @Override
diff --git a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinator.java b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinator.java
index b5fa3daf4620..f6f9a44eb59d 100644
--- a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/NacosCoordinator.java
@@ -18,28 +18,34 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.nacos;
 
+import com.alibaba.nacos.api.exception.NacosException;
 import com.alibaba.nacos.api.naming.NamingService;
+import com.alibaba.nacos.api.naming.listener.Event;
+import com.alibaba.nacos.api.naming.listener.EventListener;
+import com.alibaba.nacos.api.naming.listener.NamingEvent;
 import com.alibaba.nacos.api.naming.pojo.Instance;
 import java.util.ArrayList;
 import java.util.List;
 
 import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterHealthStatus;
-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.OAPNodeChecker;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.cluster.ServiceQueryException;
 import org.apache.skywalking.oap.server.core.cluster.ServiceRegisterException;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
-public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery {
+@Slf4j
+public class NacosCoordinator extends ClusterCoordinator {
 
     private final ModuleDefineHolder manager;
     private final NamingService namingService;
@@ -57,7 +63,6 @@ public NacosCoordinator(final ModuleDefineHolder manager, final NamingService na
     public List<RemoteInstance> queryRemoteNodes() {
         List<RemoteInstance> remoteInstances = new ArrayList<>();
         try {
-            initHealthChecker();
             List<Instance> instances = namingService.selectInstances(config.getServiceName(), true);
             if (CollectionUtils.isNotEmpty(instances)) {
                 instances.forEach(instance -> {
@@ -78,6 +83,11 @@ public List<RemoteInstance> queryRemoteNodes() {
             healthChecker.unHealth(e);
             throw new ServiceQueryException(e.getMessage());
         }
+
+        if (log.isDebugEnabled()) {
+            log.debug("Nacos cluster instances:{}", remoteInstances);
+        }
+
         return remoteInstances;
     }
 
@@ -86,17 +96,16 @@ public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegister
         if (needUsingInternalAddr()) {
             remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
         }
+        this.selfAddress = remoteInstance.getAddress();
         String host = remoteInstance.getAddress().getHost();
         int port = remoteInstance.getAddress().getPort();
         try {
-            initHealthChecker();
             namingService.registerInstance(config.getServiceName(), host, port);
             healthChecker.health();
         } catch (Throwable e) {
             healthChecker.unHealth(e);
             throw new ServiceRegisterException(e.getMessage());
         }
-        this.selfAddress = remoteInstance.getAddress();
     }
 
     private boolean needUsingInternalAddr() {
@@ -109,4 +118,52 @@ private void initHealthChecker() {
             healthChecker = metricCreator.createHealthCheckerGauge("cluster_nacos", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
         }
     }
+
+    private RemoteInstance buildRemoteInstance(Instance instance) {
+        Address address = new Address(instance.getIp(), instance.getPort(), false);
+        if (address.equals(selfAddress)) {
+            address.setSelf(true);
+        }
+        return new RemoteInstance(address);
+    }
+
+    private void checkHealth(List<RemoteInstance> remoteInstances) {
+        ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
+        if (healthStatus.isHealth()) {
+            this.healthChecker.health();
+        } else {
+            this.healthChecker.unHealth(healthStatus.getReason());
+        }
+    }
+
+    @Override
+    public void start() throws ModuleStartException {
+        initHealthChecker();
+        try {
+            namingService.subscribe(config.getServiceName(), new NacosEventListener());
+        } catch (NacosException e) {
+            throw new ModuleStartException("Failed to start cluster coordinator.", e);
+        }
+    }
+
+    class NacosEventListener implements EventListener {
+        @Override
+        public void onEvent(final Event event) {
+            try {
+                List<Instance> instances = ((NamingEvent) event).getInstances();
+                List<RemoteInstance> remoteInstances = new ArrayList<>(instances.size());
+                instances.forEach(instance -> {
+                    RemoteInstance remoteInstance = buildRemoteInstance(instance);
+                    if (instance.isHealthy() && instance.isEnabled()) {
+                        remoteInstances.add(remoteInstance);
+                    }
+                });
+                checkHealth(remoteInstances);
+                notifyWatchers(remoteInstances);
+            } catch (Throwable e) {
+                healthChecker.unHealth(e);
+                log.error("Failed to notify and update remote instances.", e);
+            }
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/ClusterModuleStandaloneProvider.java b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/ClusterModuleStandaloneProvider.java
index 7c4b91c3a9c1..d0d4b57afce7 100644
--- a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/ClusterModuleStandaloneProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/ClusterModuleStandaloneProvider.java
@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.standalone;
 
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -26,6 +27,8 @@
 import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException;
 
 public class ClusterModuleStandaloneProvider extends ModuleProvider {
+    private StandaloneManager standaloneManager;
+
     public ClusterModuleStandaloneProvider() {
         super();
     }
@@ -47,9 +50,10 @@ public ConfigCreator newConfigCreator() {
 
     @Override
     public void prepare() throws ServiceNotProvidedException {
-        StandaloneManager standaloneManager = new StandaloneManager();
+        standaloneManager = new StandaloneManager();
         this.registerServiceImplementation(ClusterRegister.class, standaloneManager);
         this.registerServiceImplementation(ClusterNodesQuery.class, standaloneManager);
+        this.registerServiceImplementation(ClusterCoordinator.class, standaloneManager);
     }
 
     @Override
@@ -58,6 +62,7 @@ public void start() throws ModuleStartException {
 
     @Override
     public void notifyAfterCompleted() {
+        standaloneManager.notifyWatchers();
     }
 
     @Override
diff --git a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java
index 8b160a740e69..99c09e0e213a 100644
--- a/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java
+++ b/oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java
@@ -19,15 +19,15 @@
 package org.apache.skywalking.oap.server.cluster.plugin.standalone;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
-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.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 
 /**
  * A cluster manager simulator. Work in memory only. Also return the current instance.
  */
-public class StandaloneManager implements ClusterNodesQuery, ClusterRegister {
+public class StandaloneManager extends ClusterCoordinator {
 
     private volatile RemoteInstance remoteInstance;
 
@@ -46,4 +46,12 @@ public List<RemoteInstance> queryRemoteNodes() {
         remoteList.add(remoteInstance);
         return remoteList;
     }
+
+    public void notifyWatchers() {
+        notifyWatchers(Collections.singletonList(remoteInstance));
+    }
+
+    @Override
+    public void start() {
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperProvider.java b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperProvider.java
index 4acb88c5b189..8e2c5f4471b4 100644
--- a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperProvider.java
+++ b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ClusterModuleZookeeperProvider.java
@@ -30,6 +30,7 @@
 import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
 import org.apache.skywalking.oap.server.core.CoreModule;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -147,6 +148,7 @@ public List<ACL> getAclForPath(String s) {
 
         this.registerServiceImplementation(ClusterRegister.class, coordinator);
         this.registerServiceImplementation(ClusterNodesQuery.class, coordinator);
+        this.registerServiceImplementation(ClusterCoordinator.class, coordinator);
     }
 
     @Override
diff --git a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java
index c33da237dd09..6bb2161ef6df 100644
--- a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java
+++ b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ZookeeperCoordinator.java
@@ -23,24 +23,29 @@
 import java.util.List;
 import java.util.UUID;
 
+import lombok.extern.slf4j.Slf4j;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.state.ConnectionState;
 import org.apache.curator.x.discovery.ServiceCache;
 import org.apache.curator.x.discovery.ServiceDiscovery;
 import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.details.ServiceCacheListener;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterHealthStatus;
-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.OAPNodeChecker;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.cluster.ServiceQueryException;
 import org.apache.skywalking.oap.server.core.cluster.ServiceRegisterException;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
-public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery {
+@Slf4j
+public class ZookeeperCoordinator extends ClusterCoordinator {
 
     private static final String REMOTE_NAME_PATH = "remote";
 
@@ -52,22 +57,20 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery
     private HealthCheckMetrics healthChecker;
 
     ZookeeperCoordinator(final ModuleDefineHolder manager, final ClusterModuleZookeeperConfig config,
-                         final ServiceDiscovery<RemoteInstance> serviceDiscovery) throws Exception {
+                         final ServiceDiscovery<RemoteInstance> serviceDiscovery) {
         this.manager = manager;
         this.config = config;
         this.serviceDiscovery = serviceDiscovery;
         this.serviceCache = serviceDiscovery.serviceCacheBuilder().name(REMOTE_NAME_PATH).build();
-        this.serviceCache.start();
     }
 
     @Override
-    public synchronized void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
+    public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
         try {
-            initHealthChecker();
             if (needUsingInternalAddr()) {
                 remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
             }
-
+            this.selfAddress = remoteInstance.getAddress();
             ServiceInstance<RemoteInstance> thisInstance = ServiceInstance.<RemoteInstance>builder().name(REMOTE_NAME_PATH)
                                                                                                     .id(UUID.randomUUID()
                                                                                                             .toString())
@@ -81,8 +84,6 @@ public synchronized void registerRemote(RemoteInstance remoteInstance) throws Se
                                                                                                     .build();
 
             serviceDiscovery.registerService(thisInstance);
-
-            this.selfAddress = remoteInstance.getAddress();
             this.healthChecker.health();
         } catch (Throwable e) {
             this.healthChecker.unHealth(e);
@@ -94,7 +95,6 @@ public synchronized void registerRemote(RemoteInstance remoteInstance) throws Se
     public List<RemoteInstance> queryRemoteNodes() {
         List<RemoteInstance> remoteInstances = new ArrayList<>(20);
         try {
-            initHealthChecker();
             List<ServiceInstance<RemoteInstance>> serviceInstances = serviceCache.getInstances();
             serviceInstances.forEach(serviceInstance -> {
                 RemoteInstance instance = serviceInstance.getPayload();
@@ -115,6 +115,10 @@ public List<RemoteInstance> queryRemoteNodes() {
             this.healthChecker.unHealth(e);
             throw new ServiceQueryException(e.getMessage());
         }
+
+        if (log.isDebugEnabled()) {
+            remoteInstances.forEach(instance -> log.debug("Zookeeper cluster instance: {}", instance));
+        }
         return remoteInstances;
     }
 
@@ -128,4 +132,35 @@ private void initHealthChecker() {
             healthChecker = metricCreator.createHealthCheckerGauge("cluster_zookeeper", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE);
         }
     }
+
+    @Override
+    public void start() throws ModuleStartException {
+        try {
+            initHealthChecker();
+            this.serviceCache.start();
+            serviceCache.addListener(new ZookeeperEventListener());
+        } catch (Exception e) {
+            throw new ModuleStartException("Failed to start cluster coordinator.", e);
+        }
+    }
+
+    class ZookeeperEventListener implements ServiceCacheListener {
+        @Override
+        public void cacheChanged() {
+            try {
+                List<RemoteInstance> remoteInstances = queryRemoteNodes();
+                notifyWatchers(remoteInstances);
+            } catch (Throwable e) {
+                healthChecker.unHealth(e);
+                log.error("Failed to notify and update remote instances", e);
+            }
+        }
+
+        @Override
+        public void stateChanged(final CuratorFramework client, final ConnectionState newState) {
+            if (log.isDebugEnabled()) {
+                log.debug("Zookeeper ConnectionState changed, state: {}", newState.name());
+            }
+        }
+    }
 }
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
index b32c9489ab3d..91f411927e4d 100755
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
@@ -34,8 +34,8 @@
 import org.apache.skywalking.oap.server.core.cache.CacheUpdateTimer;
 import org.apache.skywalking.oap.server.core.cache.NetworkAddressAliasCache;
 import org.apache.skywalking.oap.server.core.cache.ProfileTaskCache;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
-import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
 import org.apache.skywalking.oap.server.core.cluster.OAPNodeChecker;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.command.CommandService;
@@ -341,7 +341,6 @@ public void prepare() throws ServiceNotProvidedException, ModuleStartException {
     public void start() throws ModuleStartException {
         grpcServer.addHandler(new RemoteServiceHandler(getManager()));
         grpcServer.addHandler(new HealthCheckServiceHandler());
-        remoteClientManager.start();
 
         // Disable OAL script has higher priority
         oalEngineLoaderService.load(DisableOALDefine.INSTANCE);
@@ -355,6 +354,12 @@ public void start() throws ModuleStartException {
 
         Address gRPCServerInstanceAddress = new Address(moduleConfig.getGRPCHost(), moduleConfig.getGRPCPort(), true);
         TelemetryRelatedContext.INSTANCE.setId(gRPCServerInstanceAddress.toString());
+        ClusterCoordinator coordinator = this.getManager()
+                                             .find(ClusterModule.NAME)
+                                             .provider()
+                                             .getService(ClusterCoordinator.class);
+        coordinator.registerWatcher(remoteClientManager);
+        coordinator.start();
         if (CoreModuleConfig.Role.Mixed.name()
                                        .equalsIgnoreCase(
                                            moduleConfig.getRole())
@@ -362,11 +367,7 @@ public void start() throws ModuleStartException {
                                                .equalsIgnoreCase(
                                                    moduleConfig.getRole())) {
             RemoteInstance gRPCServerInstance = new RemoteInstance(gRPCServerInstanceAddress);
-            this.getManager()
-                .find(ClusterModule.NAME)
-                .provider()
-                .getService(ClusterRegister.class)
-                .registerRemote(gRPCServerInstance);
+            coordinator.registerRemote(gRPCServerInstance);
         }
 
         OAPNodeChecker.setROLE(CoreModuleConfig.Role.fromName(moduleConfig.getRole()));
@@ -391,7 +392,7 @@ public void notifyAfterCompleted() throws ModuleStartException {
         } catch (ServerException e) {
             throw new ModuleStartException(e.getMessage(), e);
         }
-
+        remoteClientManager.start();
         PersistenceTimer.INSTANCE.start(getManager(), moduleConfig);
 
         if (moduleConfig.isEnableDataKeeperExecutor()) {
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterCoordinator.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterCoordinator.java
new file mode 100644
index 000000000000..a78205da38b2
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterCoordinator.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.oap.server.core.cluster;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
+
+@Slf4j
+public abstract class ClusterCoordinator implements ClusterRegister, ClusterNodesQuery, ClusterWatcherRegister {
+    private final List<ClusterWatcher> clusterWatchers = new ArrayList<>();
+
+    /**
+     * Initialize the required resources, such as healthy checker and listener.
+     */
+    public abstract void start() throws ModuleStartException;
+
+    @Override
+    public void registerWatcher(final ClusterWatcher watcher) {
+        this.clusterWatchers.add(watcher);
+    }
+
+    protected void notifyWatchers(List<RemoteInstance> remoteInstances) {
+        if (log.isDebugEnabled()) {
+            log.debug("Notify watchers and update cluster instances:{}", remoteInstances.toString());
+        }
+        this.clusterWatchers.forEach(
+            clusterWatcher -> clusterWatcher.onClusterNodesChanged(remoteInstances));
+    }
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterModule.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterModule.java
index 9e65fcc5b626..feaf4d5c8358 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterModule.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterModule.java
@@ -32,7 +32,8 @@ public ClusterModule() {
     public Class[] services() {
         return new Class[] {
             ClusterRegister.class,
-            ClusterNodesQuery.class
+            ClusterNodesQuery.class,
+            ClusterCoordinator.class
         };
     }
 }
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcher.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcher.java
new file mode 100644
index 000000000000..dce88afb4128
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcher.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.oap.server.core.cluster;
+
+import java.util.List;
+
+public interface ClusterWatcher {
+    void onClusterNodesChanged(List<RemoteInstance> remoteInstances);
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcherRegister.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcherRegister.java
new file mode 100644
index 000000000000..e595a6a42102
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcherRegister.java
@@ -0,0 +1,25 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.oap.server.core.cluster;
+
+import org.apache.skywalking.oap.server.library.module.Service;
+
+public interface ClusterWatcherRegister extends Service {
+    void registerWatcher(ClusterWatcher watcher);
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java
index c7ee1129b6f6..08c64cf33e5e 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java
@@ -35,27 +35,28 @@
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.skywalking.oap.server.core.CoreModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterModule;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
+import org.apache.skywalking.oap.server.core.cluster.ClusterWatcher;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
+import org.apache.skywalking.oap.server.core.status.ServerStatusService;
 import org.apache.skywalking.oap.server.library.module.ModuleDefineHolder;
 import org.apache.skywalking.oap.server.library.module.Service;
 import org.apache.skywalking.oap.server.library.server.grpc.ssl.DynamicSslContext;
+import org.apache.skywalking.oap.server.library.util.RunnableWithExceptionProtection;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
 import org.apache.skywalking.oap.server.telemetry.api.GaugeMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * This class manages the connections between OAP servers. There is a task schedule that will automatically query a
  * server list from the cluster module. Such as Zookeeper cluster module or Kubernetes cluster module.
  */
-public class RemoteClientManager implements Service {
-
-    private static final Logger LOGGER = LoggerFactory.getLogger(RemoteClientManager.class);
-
+@Slf4j
+public class RemoteClientManager implements Service, ClusterWatcher {
     private final ModuleDefineHolder moduleDefineHolder;
     private DynamicSslContext sslContext;
     private ClusterNodesQuery clusterNodesQuery;
@@ -93,14 +94,28 @@ public RemoteClientManager(final ModuleDefineHolder moduleDefineHolder, final in
 
     public void start() {
         Optional.ofNullable(sslContext).ifPresent(DynamicSslContext::start);
-        Executors.newSingleThreadScheduledExecutor().scheduleWithFixedDelay(this::refresh, 1, 5, TimeUnit.SECONDS);
+        Executors.newSingleThreadScheduledExecutor()
+                 .scheduleWithFixedDelay(new RunnableWithExceptionProtection(this::refresh, t -> log.error(
+                     "Scheduled refresh Remote Clients failure.", t)), 1, 10, TimeUnit.SECONDS);
     }
 
     /**
-     * Query OAP server list from the cluster module and create a new connection for the new node. Make the OAP server
-     * orderly because of each of the server will send stream data to each other by hash code.
+     * Refresh the remote clients by query OAP server list from the cluster module.
      */
     void refresh() {
+        if (Objects.isNull(clusterNodesQuery)) {
+            this.clusterNodesQuery = moduleDefineHolder.find(ClusterModule.NAME)
+                                                       .provider()
+                                                       .getService(ClusterNodesQuery.class);
+        }
+
+        this.refresh(clusterNodesQuery.queryRemoteNodes());
+    }
+
+    /**
+     * Refresh the remote clients according to OAP server list. Make the OAP server orderly because of each of the server will send stream data to each other by hash code.
+     */
+    synchronized void refresh(List<RemoteInstance> instanceList) {
         if (gauge == null) {
             gauge = moduleDefineHolder.find(TelemetryModule.NAME)
                                       .provider()
@@ -111,40 +126,33 @@ void refresh() {
                                       );
         }
         try {
-            if (Objects.isNull(clusterNodesQuery)) {
-                synchronized (RemoteClientManager.class) {
-                    if (Objects.isNull(clusterNodesQuery)) {
-                        this.clusterNodesQuery = moduleDefineHolder.find(ClusterModule.NAME)
-                                                                   .provider()
-                                                                   .getService(ClusterNodesQuery.class);
-                    }
-                }
+            if (log.isDebugEnabled()) {
+                log.debug("Refresh remote nodes collection.");
             }
 
-            if (LOGGER.isDebugEnabled()) {
-                LOGGER.debug("Refresh remote nodes collection.");
-            }
-
-            List<RemoteInstance> instanceList = clusterNodesQuery.queryRemoteNodes();
             instanceList = distinct(instanceList);
             Collections.sort(instanceList);
 
             gauge.setValue(instanceList.size());
 
-            if (LOGGER.isDebugEnabled()) {
-                instanceList.forEach(instance -> LOGGER.debug("Cluster instance: {}", instance.toString()));
+            if (log.isDebugEnabled()) {
+                instanceList.forEach(instance -> log.debug("Cluster instance: {}", instance.toString()));
             }
 
             if (!compare(instanceList)) {
-                if (LOGGER.isDebugEnabled()) {
-                    LOGGER.debug("ReBuilding remote clients.");
+                if (log.isDebugEnabled()) {
+                    log.debug("ReBuilding remote clients.");
                 }
                 reBuildRemoteClients(instanceList);
+                moduleDefineHolder.find(CoreModule.NAME)
+                       .provider()
+                       .getService(ServerStatusService.class)
+                       .reBalancedCluster(System.currentTimeMillis());
             }
 
             printRemoteClientList();
         } catch (Throwable t) {
-            LOGGER.error(t.getMessage(), t);
+            log.error(t.getMessage(), t);
         }
     }
 
@@ -152,10 +160,10 @@ void refresh() {
      * Print the client list into log for confirm how many clients built.
      */
     private void printRemoteClientList() {
-        if (LOGGER.isDebugEnabled()) {
+        if (log.isDebugEnabled()) {
             StringBuilder addresses = new StringBuilder();
             this.usingClients.forEach(client -> addresses.append(client.getAddress().toString()).append(","));
-            LOGGER.debug("Remote client list: {}", addresses);
+            log.debug("Remote client list: {}", addresses);
         }
     }
 
@@ -266,6 +274,11 @@ private boolean compare(List<RemoteInstance> remoteInstances) {
         }
     }
 
+    @Override
+    public void onClusterNodesChanged(List<RemoteInstance> remoteInstances) {
+        refresh(remoteInstances);
+    }
+
     enum Action {
         Close, Unchanged, Create
     }
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ClusterStatus.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ClusterStatus.java
new file mode 100644
index 000000000000..49203abb4e6b
--- /dev/null
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ClusterStatus.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.skywalking.oap.server.core.status;
+
+import lombok.AccessLevel;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Booting status indicate whether the current server starts successfully.
+ */
+@Getter
+@Setter(AccessLevel.PACKAGE)
+public class ClusterStatus {
+    private long reBalanceTime = 0;
+}
diff --git a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ServerStatusService.java b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ServerStatusService.java
index fac17a1b5d9a..100cecee2144 100644
--- a/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ServerStatusService.java
+++ b/oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ServerStatusService.java
@@ -38,6 +38,8 @@ public class ServerStatusService implements Service {
     private final ModuleManager manager;
     @Getter
     private BootingStatus bootingStatus = new BootingStatus();
+    @Getter
+    private ClusterStatus clusterStatus = new ClusterStatus();
 
     public void bootedNow(long uptime) {
         bootingStatus.setBooted(true);
@@ -50,4 +52,12 @@ public void bootedNow(long uptime) {
                .setValue(uptime / 1000d);
     }
 
+    public void reBalancedCluster(long uptime) {
+        clusterStatus.setReBalanceTime(uptime);
+        manager.find(TelemetryModule.NAME)
+               .provider()
+               .getService(MetricsCreator.class)
+               .createGauge("cluster_rebalance_time", "oap cluster rebalance time after scale", MetricsTag.EMPTY_KEY, MetricsTag.EMPTY_VALUE)
+               .setValue(uptime / 1000d);
+    }
 }


Test patch:
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProviderTest.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProviderTest.java
index 62edc69a1af4..70233a4c4154 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProviderTest.java
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProviderTest.java
@@ -138,7 +138,7 @@ public void prepareSingle() throws Exception {
 
         PowerMockito.mockStatic(Consul.class);
         when(Consul.builder()).thenReturn(builder);
-        when(builder.withConnectTimeoutMillis(anyLong())).thenCallRealMethod();
+        when(builder.withConnectTimeoutMillis(anyLong())).thenReturn(builder);
 
         when(builder.withHostAndPort(any())).thenReturn(builder);
 
@@ -166,4 +166,4 @@ public void requiredModules() {
         String[] modules = provider.requiredModules();
         assertArrayEquals(new String[] {CoreModule.NAME}, modules);
     }
-}
\ No newline at end of file
+}
diff --git a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ITClusterModuleConsulProviderFunctionalTest.java b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ITClusterModuleConsulProviderFunctionalTest.java
index 78d3524de740..501a0fb539f8 100644
--- a/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ITClusterModuleConsulProviderFunctionalTest.java
+++ b/oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ITClusterModuleConsulProviderFunctionalTest.java
@@ -21,10 +21,12 @@
 import com.google.common.base.Strings;
 import com.orbitz.consul.AgentClient;
 import com.orbitz.consul.Consul;
-import com.orbitz.consul.model.agent.ImmutableRegistration;
-import com.orbitz.consul.model.agent.Registration;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
+import org.apache.skywalking.oap.server.core.cluster.ClusterWatcher;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 import org.apache.skywalking.oap.server.core.cluster.ClusterNodesQuery;
 import org.apache.skywalking.oap.server.core.cluster.ClusterRegister;
@@ -86,11 +88,18 @@ public void registerRemote() throws Exception {
         ModuleProvider provider = createProvider(serviceName);
 
         Address selfAddress = new Address("127.0.0.1", 1000, true);
+
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(provider, 1).size());
+
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertTrue(queryAddress.isSelf());
@@ -103,11 +112,16 @@ public void registerRemoteOfInternal() throws Exception {
 
         Address selfAddress = new Address("127.0.0.2", 1002, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
-
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(provider, 1).size());
+
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals("127.0.1.2", queryAddress.getHost());
         assertEquals(1001, queryAddress.getPort());
@@ -119,15 +133,22 @@ public void registerRemoteOfReceiver() throws Exception {
         final String serviceName = "register_remote_receiver";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
 
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         // Mixed or Aggregator
         Address selfAddress = new Address("127.0.0.3", 1003, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(providerA).registerRemote(instance);
+        coordinatorA.start();
+        coordinatorA.registerRemote(instance);
 
         // Receiver
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(providerB, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertFalse(queryAddress.isSelf());
@@ -138,6 +159,14 @@ public void registerRemoteOfCluster() throws Exception {
         final String serviceName = "register_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.4", 1004, true);
         Address addressB = new Address("127.0.0.5", 1005, true);
@@ -145,14 +174,16 @@ public void registerRemoteOfCluster() throws Exception {
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2,  queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2,  queryRemoteNodes(providerB, 2).size());
     }
 
     @Test
@@ -160,6 +191,14 @@ public void unregisterRemoteOfCluster() throws Exception {
         final String serviceName = "unregister_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.6", 1006, true);
         Address addressB = new Address("127.0.0.7", 1007, true);
@@ -167,14 +206,16 @@ public void unregisterRemoteOfCluster() throws Exception {
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2,  queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2,  queryRemoteNodes(providerB, 2).size());
 
         // unregister A
         Consul client = Whitebox.getInternalState(providerA, "client");
@@ -182,8 +223,9 @@ public void unregisterRemoteOfCluster() throws Exception {
         agentClient.deregister(instanceA.getAddress().toString());
 
         // only B
-        remoteInstancesOfB = queryRemoteNodes(providerB, 1, 120);
+        remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1, 120);
         assertEquals(1, remoteInstancesOfB.size());
+        assertEquals(1,  queryRemoteNodes(providerB, 1).size());
         Address address = remoteInstancesOfB.get(0).getAddress();
         assertEquals(address, addressB);
         assertTrue(addressB.isSelf());
@@ -215,28 +257,6 @@ private ClusterModuleConsulProvider createProvider(String serviceName, String in
         provider.start();
         provider.notifyAfterCompleted();
 
-        ConsulCoordinator consulCoordinator = (ConsulCoordinator) provider.getService(ClusterRegister.class);
-
-        // ignore health check
-        ClusterRegister register = remoteInstance -> {
-            if (needUsingInternalAddr(config)) {
-                remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
-            }
-
-            Consul client = Whitebox.getInternalState(consulCoordinator, "client");
-            AgentClient agentClient = client.agentClient();
-            Whitebox.setInternalState(consulCoordinator, "selfAddress", remoteInstance.getAddress());
-            Registration registration = ImmutableRegistration.builder()
-                                                             .id(remoteInstance.getAddress().toString())
-                                                             .name(serviceName)
-                                                             .address(remoteInstance.getAddress().getHost())
-                                                             .port(remoteInstance.getAddress().getPort())
-                                                             .build();
-
-            agentClient.register(registration);
-        };
-
-        provider.registerServiceImplementation(ClusterRegister.class, register);
         return provider;
     }
 
@@ -244,6 +264,10 @@ private boolean needUsingInternalAddr(ClusterModuleConsulConfig config) {
         return !Strings.isNullOrEmpty(config.getInternalComHost()) && config.getInternalComPort() > 0;
     }
 
+    private ClusterCoordinator getClusterCoordinator(ModuleProvider provider) {
+        return provider.getService(ClusterCoordinator.class);
+    }
+
     private ClusterRegister getClusterRegister(ModuleProvider provider) {
         return provider.getService(ClusterRegister.class);
     }
@@ -252,6 +276,25 @@ private ClusterNodesQuery getClusterNodesQuery(ModuleProvider provider) {
         return provider.getService(ClusterNodesQuery.class);
     }
 
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
+        throws InterruptedException {
+        return notifiedRemoteNodes(watcher, goals, 20);
+    }
+
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
+                                                     int cyclic) throws InterruptedException {
+        do {
+            List<RemoteInstance> instances = watcher.getRemoteInstances();
+            if (instances.size() == goals) {
+                return instances;
+            } else {
+                Thread.sleep(1000);
+            }
+        }
+        while (--cyclic > 0);
+        return Collections.emptyList();
+    }
+
     private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
         return queryRemoteNodes(provider, goals, 20);
     }
@@ -287,4 +330,14 @@ private void validateServiceInstance(Address selfAddress, Address otherAddress,
         assertTrue(selfExist);
         assertTrue(otherExist);
     }
+
+    class ClusterMockWatcher implements ClusterWatcher {
+        @Getter
+        private List<RemoteInstance> remoteInstances = new ArrayList<>();
+
+        @Override
+        public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
+            this.remoteInstances = remoteInstances;
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ITClusterModuleEtcdProviderFunctionalTest.java b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ITClusterModuleEtcdProviderFunctionalTest.java
index aecc4a952e91..3fbde15353d1 100644
--- a/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ITClusterModuleEtcdProviderFunctionalTest.java
+++ b/oap-server/server-cluster-plugin/cluster-etcd-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ITClusterModuleEtcdProviderFunctionalTest.java
@@ -19,11 +19,14 @@
 package org.apache.skywalking.oap.server.cluster.plugin.etcd;
 
 import io.etcd.jetcd.Client;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
+import org.apache.skywalking.oap.server.core.cluster.ClusterWatcher;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 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;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
@@ -78,10 +81,15 @@ public void registerRemote() throws Exception {
 
         Address selfAddress = new Address("127.0.0.1", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(provider, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertTrue(queryAddress.isSelf());
@@ -94,11 +102,16 @@ public void registerRemoteOfInternal() throws Exception {
 
         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);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(provider, 1).size());
+
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals("127.0.1.2", queryAddress.getHost());
         assertEquals(1000, queryAddress.getPort());
@@ -110,15 +123,22 @@ public void registerRemoteOfReceiver() throws Exception {
         final String serviceName = "register_remote_receiver";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
 
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         // Mixed or Aggregator
         Address selfAddress = new Address("127.0.0.3", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(providerA).registerRemote(instance);
+        coordinatorA.start();
+        coordinatorA.registerRemote(instance);
 
         // Receiver
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1,  queryRemoteNodes(providerB, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertFalse(queryAddress.isSelf());
@@ -129,6 +149,14 @@ public void registerRemoteOfCluster() throws Exception {
         final String serviceName = "register_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.4", 1000, true);
         Address addressB = new Address("127.0.0.5", 1000, true);
@@ -136,14 +164,16 @@ public void registerRemoteOfCluster() throws Exception {
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2,  queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2,  queryRemoteNodes(providerB, 2).size());
     }
 
     @Test
@@ -151,29 +181,39 @@ public void unregisterRemoteOfCluster() throws Exception {
         final String serviceName = "unregister_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
-
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         Address addressA = new Address("127.0.0.4", 1000, true);
         Address addressB = new Address("127.0.0.5", 1000, true);
 
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2,  queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2,  queryRemoteNodes(providerB, 2).size());
 
         // unregister A
-        Client client = Whitebox.getInternalState(getClusterRegister(providerA), "client");
+        Client client = Whitebox.getInternalState(coordinatorA, "client");
         client.close();
 
         // only B
-        remoteInstancesOfB = queryRemoteNodes(providerB, 1, 120);
+        remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1, 120);
         assertEquals(1, remoteInstancesOfB.size());
+        assertEquals(1,  queryRemoteNodes(providerB, 1).size());
         Address address = remoteInstancesOfB.get(0).getAddress();
         assertEquals(address, addressB);
         assertTrue(addressB.isSelf());
@@ -214,16 +254,15 @@ private ClusterModuleEtcdProvider createProvider(String serviceName, String inte
         return provider;
     }
 
-    private ClusterRegister getClusterRegister(ModuleProvider provider) {
-        return provider.getService(ClusterRegister.class);
+    private ClusterCoordinator getClusterCoordinator(ModuleProvider provider) {
+        return provider.getService(ClusterCoordinator.class);
     }
 
     private ClusterNodesQuery getClusterNodesQuery(ModuleProvider provider) {
         return provider.getService(ClusterNodesQuery.class);
     }
 
-    private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals)
-        throws InterruptedException {
+    private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
         return queryRemoteNodes(provider, goals, 20);
     }
 
@@ -241,6 +280,25 @@ private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals
         return Collections.emptyList();
     }
 
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
+        throws InterruptedException {
+        return notifiedRemoteNodes(watcher, goals, 20);
+    }
+
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
+                                                  int cyclic) throws InterruptedException {
+        do {
+            List<RemoteInstance> instances = watcher.getRemoteInstances();
+            if (instances.size() == goals) {
+                return instances;
+            } else {
+                Thread.sleep(1000);
+            }
+        }
+        while (--cyclic > 0);
+        return Collections.emptyList();
+    }
+
     private void validateServiceInstance(Address selfAddress, Address otherAddress,
                                          List<RemoteInstance> queryResult) {
         assertEquals(2, queryResult.size());
@@ -260,4 +318,13 @@ private void validateServiceInstance(Address selfAddress, Address otherAddress,
         assertTrue(otherExist);
     }
 
+    class ClusterMockWatcher implements ClusterWatcher {
+        @Getter
+        private List<RemoteInstance> remoteInstances = new ArrayList<>();
+
+        @Override
+        public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
+            this.remoteInstances = remoteInstances;
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
index ba8d230b6fa4..87c003cf0dcb 100644
--- a/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
+++ b/oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
@@ -22,25 +22,33 @@
 import io.kubernetes.client.openapi.models.V1Pod;
 import io.kubernetes.client.openapi.models.V1PodStatus;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.stream.Collectors;
-
+import lombok.Getter;
 import org.apache.skywalking.oap.server.core.CoreModule;
-import org.apache.skywalking.oap.server.core.CoreModuleConfig;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
+import org.apache.skywalking.oap.server.core.cluster.ClusterWatcher;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.config.ConfigService;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
+import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.module.ModuleProvider;
+import org.apache.skywalking.oap.server.library.module.ModuleProviderHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleServiceHolder;
+import org.apache.skywalking.oap.server.library.module.ModuleStartException;
 import org.apache.skywalking.oap.server.telemetry.TelemetryModule;
-import org.apache.skywalking.oap.server.telemetry.api.HealthCheckMetrics;
 import org.apache.skywalking.oap.server.telemetry.api.MetricsCreator;
 import org.apache.skywalking.oap.server.telemetry.none.MetricsCreatorNoop;
-import org.apache.skywalking.oap.server.testing.module.ModuleDefineTesting;
-import org.apache.skywalking.oap.server.testing.module.ModuleManagerTesting;
+import org.apache.skywalking.oap.server.telemetry.none.NoneTelemetryProvider;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.powermock.api.mockito.PowerMockito;
 import org.powermock.api.support.membermodification.MemberModifier;
 import org.powermock.core.classloader.annotations.PowerMockIgnore;
@@ -48,47 +56,89 @@
 import org.powermock.modules.junit4.PowerMockRunner;
 import org.powermock.reflect.Whitebox;
 
-import static org.mockito.Mockito.doNothing;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.powermock.api.mockito.PowerMockito.when;
 
 @RunWith(PowerMockRunner.class)
-@PowerMockIgnore({"com.sun.org.apache.xerces.*", "javax.xml.*", "org.xml.*", "javax.management.*", "org.w3c.*"})
+@PowerMockIgnore({
+    "com.sun.org.apache.xerces.*",
+    "javax.xml.*",
+    "org.xml.*",
+    "javax.management.*",
+    "org.w3c.*"
+})
 @PrepareForTest({NamespacedPodListInformer.class})
 public class KubernetesCoordinatorTest {
 
-    private KubernetesCoordinator coordinator;
-    private HealthCheckMetrics healthChecker = mock(HealthCheckMetrics.class);
-
     public static final String LOCAL_HOST = "127.0.0.1";
-    public static final Integer GRPC_PORT = 8454;
-    public static final Integer SELF_UID = 12345;
+    public static final String REMOTE_HOST = "127.0.0.2";
+    public static final Integer GRPC_PORT = 11800;
+    public static final String SELF_UID = "self";
+    public static final String REMOTE_UID = "remote";
 
-    private Address selfAddress;
+    @Mock
+    private ModuleManager moduleManager;
+    @Mock
+    private NoneTelemetryProvider telemetryProvider;
     private NamespacedPodListInformer informer;
+    private ModuleProvider providerA;
+    private ModuleProvider providerB;
+    private Address addressA;
+    private Address addressB;
+    private KubernetesCoordinator coordinatorA;
+    private KubernetesCoordinator coordinatorB;
+    private V1Pod podA;
+    private V1Pod podB;
 
     @Before
-    public void prepare() throws IllegalAccessException {
-        coordinator = new KubernetesCoordinator(getManager(), new ClusterModuleKubernetesConfig());
-        Whitebox.setInternalState(coordinator, "healthChecker", healthChecker);
-        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinator, String.valueOf(SELF_UID));
-        selfAddress = new Address(LOCAL_HOST, GRPC_PORT, true);
+    public void prepare() throws ModuleStartException {
+
+        Mockito.when(telemetryProvider.getService(MetricsCreator.class))
+               .thenReturn(new MetricsCreatorNoop());
+        TelemetryModule telemetryModule = Mockito.spy(TelemetryModule.class);
+        Whitebox.setInternalState(telemetryModule, "loadedProvider", telemetryProvider);
         informer = PowerMockito.mock(NamespacedPodListInformer.class);
         Whitebox.setInternalState(NamespacedPodListInformer.class, "INFORMER", informer);
-        doNothing().when(healthChecker).health();
+        Mockito.when(moduleManager.find(TelemetryModule.NAME)).thenReturn(telemetryModule);
+        when(moduleManager.find(CoreModule.NAME)).thenReturn(mock(ModuleProviderHolder.class));
+        when(moduleManager.find(CoreModule.NAME).provider()).thenReturn(mock(ModuleServiceHolder.class));
+        when(moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class)).thenReturn(
+            mock(ConfigService.class));
+        when(moduleManager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort()).thenReturn(
+            GRPC_PORT);
+
+        providerA = createProvider(SELF_UID);
+        providerB = createProvider(REMOTE_UID);
+        addressA = new Address(LOCAL_HOST, GRPC_PORT, true);
+        addressB = new Address(REMOTE_HOST, GRPC_PORT, true);
+        podA = mockPod(SELF_UID, LOCAL_HOST);
+        podB = mockPod(REMOTE_UID, REMOTE_HOST);
+        coordinatorA = getClusterCoordinator(providerA);
+        coordinatorB = getClusterCoordinator(providerB);
+        coordinatorA.start();
+        coordinatorB.start();
     }
 
     @Test
     public void queryRemoteNodesWhenInformerNotwork() throws Exception {
+        KubernetesCoordinator coordinator = getClusterCoordinator(providerA);
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
         PowerMockito.doReturn(Optional.empty()).when(NamespacedPodListInformer.INFORMER).listPods();
         List<RemoteInstance> remoteInstances = Whitebox.invokeMethod(coordinator, "queryRemoteNodes");
         Assert.assertEquals(1, remoteInstances.size());
-        Assert.assertEquals(selfAddress, remoteInstances.get(0).getAddress());
+        Assert.assertEquals(addressA, remoteInstances.get(0).getAddress());
 
     }
 
     @Test
     public void queryRemoteNodesWhenInformerWork() throws Exception {
+        ModuleProvider provider = createProvider(SELF_UID + "0");
+        KubernetesCoordinator coordinator = getClusterCoordinator(provider);
+        coordinator.start();
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinator, SELF_UID + "0");
         PowerMockito.doReturn(Optional.of(mockPodList())).when(NamespacedPodListInformer.INFORMER).listPods();
         List<RemoteInstance> remoteInstances = Whitebox.invokeMethod(coordinator, "queryRemoteNodes");
         Assert.assertEquals(5, remoteInstances.size());
@@ -104,17 +154,161 @@ public void queryRemoteNodesWhenInformerWork() throws Exception {
 
     }
 
-    private ModuleManagerTesting getManager() {
-        ModuleManagerTesting moduleManagerTesting = new ModuleManagerTesting();
-        ModuleDefineTesting coreModuleDefine = new ModuleDefineTesting();
-        moduleManagerTesting.put(CoreModule.NAME, coreModuleDefine);
-        CoreModuleConfig config = PowerMockito.mock(CoreModuleConfig.class);
-        when(config.getGRPCHost()).thenReturn(LOCAL_HOST);
-        when(config.getGRPCPort()).thenReturn(GRPC_PORT);
-        moduleManagerTesting.put(TelemetryModule.NAME, coreModuleDefine);
-        coreModuleDefine.provider().registerServiceImplementation(ConfigService.class, new ConfigService(config));
-        coreModuleDefine.provider().registerServiceImplementation(MetricsCreator.class, new MetricsCreatorNoop());
-        return moduleManagerTesting;
+    @Test
+    public void registerRemote() throws Exception {
+        RemoteInstance instance = new RemoteInstance(addressA);
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
+        PowerMockito.doReturn(Optional.of(Collections.singletonList(podA)))
+                    .when(NamespacedPodListInformer.INFORMER)
+                    .listPods();
+
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcher);
+        coordinatorA.registerRemote(instance);
+        KubernetesCoordinator.K8sResourceEventHandler listener = coordinatorA.new K8sResourceEventHandler();
+        listener.onAdd(podA);
+
+        List<RemoteInstance> remoteInstances = watcher.getRemoteInstances();
+        assertEquals(1, remoteInstances.size());
+        assertEquals(1, coordinatorA.queryRemoteNodes().size());
+        Address queryAddress = remoteInstances.get(0).getAddress();
+        assertEquals(addressA, queryAddress);
+        assertTrue(queryAddress.isSelf());
+    }
+
+    @Test
+    public void registerRemoteOfReceiver() throws Exception {
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+
+        PowerMockito.doReturn(Optional.of(Collections.singletonList(podA)))
+                    .when(NamespacedPodListInformer.INFORMER)
+                    .listPods();
+        RemoteInstance instance = new RemoteInstance(addressA);
+        coordinatorA.registerRemote(instance);
+        KubernetesCoordinator.K8sResourceEventHandler listener = coordinatorB.new K8sResourceEventHandler();
+        listener.onAdd(podA);
+
+        // Receiver
+        List<RemoteInstance> remoteInstances = watcherB.getRemoteInstances();
+        assertEquals(1, remoteInstances.size());
+        assertEquals(1, coordinatorB.queryRemoteNodes().size());
+        Address queryAddress = remoteInstances.get(0).getAddress();
+        assertEquals(addressA, queryAddress);
+        assertFalse(queryAddress.isSelf());
+    }
+
+    @Test
+    public void registerRemoteOfCluster() throws Exception {
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+
+        PowerMockito.doReturn(Optional.of(Arrays.asList(podA, podB)))
+                    .when(NamespacedPodListInformer.INFORMER)
+                    .listPods();
+        RemoteInstance instanceA = new RemoteInstance(addressA);
+        RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
+
+        KubernetesCoordinator.K8sResourceEventHandler listenerA = coordinatorA.new K8sResourceEventHandler();
+        listenerA.onAdd(podA);
+        listenerA.onAdd(podB);
+        KubernetesCoordinator.K8sResourceEventHandler listenerB = coordinatorB.new K8sResourceEventHandler();
+        listenerB.onAdd(podA);
+        listenerB.onAdd(podB);
+
+        List<RemoteInstance> remoteInstancesOfA = watcherA.getRemoteInstances();
+        validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, coordinatorA.queryRemoteNodes().size());
+
+        List<RemoteInstance> remoteInstancesOfB = watcherB.getRemoteInstances();
+        validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, coordinatorB.queryRemoteNodes().size());
+    }
+
+    @Test
+    public void deregisterRemoteOfCluster() throws Exception {
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+
+        MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+
+        PowerMockito.doReturn(Optional.of(Arrays.asList(podA, podB)))
+                    .when(NamespacedPodListInformer.INFORMER)
+                    .listPods();
+        RemoteInstance instanceA = new RemoteInstance(addressA);
+        RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
+
+        KubernetesCoordinator.K8sResourceEventHandler listenerA = coordinatorA.new K8sResourceEventHandler();
+        listenerA.onAdd(podA);
+        listenerA.onAdd(podB);
+        KubernetesCoordinator.K8sResourceEventHandler listenerB = coordinatorB.new K8sResourceEventHandler();
+        listenerB.onAdd(podA);
+        listenerB.onAdd(podB);
+
+        List<RemoteInstance> remoteInstancesOfA = watcherA.getRemoteInstances();
+        validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, coordinatorA.queryRemoteNodes().size());
+
+        List<RemoteInstance> remoteInstancesOfB = watcherB.getRemoteInstances();
+        validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, coordinatorB.queryRemoteNodes().size());
+
+        // deregister A
+        listenerB.onDelete(podA, false);
+        PowerMockito.doReturn(Optional.of(Collections.singletonList(podB)))
+                    .when(NamespacedPodListInformer.INFORMER)
+                    .listPods();
+        // only B
+        remoteInstancesOfB = watcherB.getRemoteInstances();
+        assertEquals(1, remoteInstancesOfB.size());
+        assertEquals(1, coordinatorB.queryRemoteNodes().size());
+
+        Address address = remoteInstancesOfB.get(0).getAddress();
+        assertEquals(addressB, address);
+        assertTrue(address.isSelf());
+    }
+
+    private ClusterModuleKubernetesProvider createProvider(String uidEnvName)
+        throws ModuleStartException {
+        ClusterModuleKubernetesProvider provider = new ClusterModuleKubernetesProvider();
+
+        ClusterModuleKubernetesConfig config = new ClusterModuleKubernetesConfig();
+        provider.newConfigCreator().onInitialized(config);
+        config.setNamespace("default");
+        config.setLabelSelector("app=oap");
+        config.setUidEnvName(uidEnvName);
+
+        provider.setManager(moduleManager);
+        provider.prepare();
+        provider.start();
+        provider.notifyAfterCompleted();
+        return provider;
+    }
+
+    private KubernetesCoordinator getClusterCoordinator(ModuleProvider provider) {
+        return (KubernetesCoordinator) provider.getService(ClusterCoordinator.class);
+    }
+
+    private V1Pod mockPod(String uid, String ip) {
+        V1Pod v1Pod = new V1Pod();
+        v1Pod.setMetadata(new V1ObjectMeta());
+        v1Pod.setStatus(new V1PodStatus());
+        v1Pod.getStatus().setPhase("Running");
+        v1Pod.getMetadata().setUid(uid);
+        v1Pod.getStatus().setPodIP(ip);
+        return v1Pod;
     }
 
     private List<V1Pod> mockPodList() {
@@ -123,10 +317,39 @@ private List<V1Pod> mockPodList() {
             V1Pod v1Pod = new V1Pod();
             v1Pod.setMetadata(new V1ObjectMeta());
             v1Pod.setStatus(new V1PodStatus());
-            v1Pod.getMetadata().setUid(String.valueOf(SELF_UID + i));
+            v1Pod.getMetadata().setUid(SELF_UID + i);
             v1Pod.getStatus().setPodIP(LOCAL_HOST);
             pods.add(v1Pod);
         }
         return pods;
     }
+
+    private void validateServiceInstance(Address selfAddress, Address otherAddress,
+                                         List<RemoteInstance> queryResult) {
+        assertEquals(2, queryResult.size());
+
+        boolean selfExist = false, otherExist = false;
+
+        for (RemoteInstance instance : queryResult) {
+            Address queryAddress = instance.getAddress();
+            if (queryAddress.equals(selfAddress) && queryAddress.isSelf()) {
+                selfExist = true;
+            } else if (queryAddress.equals(otherAddress) && !queryAddress.isSelf()) {
+                otherExist = true;
+            }
+        }
+
+        assertTrue(selfExist);
+        assertTrue(otherExist);
+    }
+
+    class ClusterMockWatcher implements ClusterWatcher {
+        @Getter
+        private List<RemoteInstance> remoteInstances = new ArrayList<>();
+
+        @Override
+        public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
+            this.remoteInstances = remoteInstances;
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ITClusterModuleNacosProviderFunctionalTest.java b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ITClusterModuleNacosProviderFunctionalTest.java
index 396ad5c52199..f575b961a0d5 100644
--- a/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ITClusterModuleNacosProviderFunctionalTest.java
+++ b/oap-server/server-cluster-plugin/cluster-nacos-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ITClusterModuleNacosProviderFunctionalTest.java
@@ -19,11 +19,14 @@
 package org.apache.skywalking.oap.server.cluster.plugin.nacos;
 
 import com.alibaba.nacos.api.naming.NamingService;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import lombok.Getter;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
+import org.apache.skywalking.oap.server.core.cluster.ClusterWatcher;
 import org.apache.skywalking.oap.server.library.util.StringUtil;
 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;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
@@ -86,10 +89,15 @@ public void registerRemote() throws Exception {
 
         Address selfAddress = new Address("127.0.0.1", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(provider, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertTrue(queryAddress.isSelf());
@@ -102,11 +110,16 @@ public void registerRemoteOfInternal() throws Exception {
 
         Address selfAddress = new Address("127.0.0.2", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
 
         assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(provider, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals("127.0.1.2", queryAddress.getHost());
         assertEquals(1000, queryAddress.getPort());
@@ -118,15 +131,21 @@ public void registerRemoteOfReceiver() throws Exception {
         final String serviceName = "register_remote_receiver";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
-
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         // Mixed or Aggregator
         Address selfAddress = new Address("127.0.0.3", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(providerA).registerRemote(instance);
+        coordinatorA.start();
+        coordinatorA.registerRemote(instance);
 
         // Receiver
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(providerB, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertFalse(queryAddress.isSelf());
@@ -137,21 +156,29 @@ public void registerRemoteOfCluster() throws Exception {
         final String serviceName = "register_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
-
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         Address addressA = new Address("127.0.0.4", 1000, true);
         Address addressB = new Address("127.0.0.5", 1000, true);
 
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
-
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, queryRemoteNodes(providerB, 2).size());
     }
 
     @Test
@@ -159,30 +186,39 @@ public void deregisterRemoteOfCluster() throws Exception {
         final String serviceName = "deregister_remote_cluster";
         ModuleProvider providerA = createProvider(serviceName);
         ModuleProvider providerB = createProvider(serviceName);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.6", 1000, true);
         Address addressB = new Address("127.0.0.7", 1000, true);
-
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
-
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, queryRemoteNodes(providerB, 2).size());
 
         // deregister A
-        ClusterRegister register = getClusterRegister(providerA);
-        NamingService namingServiceA = Whitebox.getInternalState(register, "namingService");
+        NamingService namingServiceA = Whitebox.getInternalState(coordinatorA, "namingService");
         namingServiceA.deregisterInstance(serviceName, addressA.getHost(), addressA.getPort());
 
         // only B
-        remoteInstancesOfB = queryRemoteNodes(providerB, 1);
+        remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstancesOfB.size());
+        assertEquals(1, queryRemoteNodes(providerB, 1).size());
+
         Address address = remoteInstancesOfB.get(0).getAddress();
         assertEquals(addressB, address);
         assertTrue(address.isSelf());
@@ -233,17 +269,20 @@ private ClusterModuleNacosProvider createProvider(String serviceName, String int
         return provider;
     }
 
-    private ClusterRegister getClusterRegister(ModuleProvider provider) {
-        return provider.getService(ClusterRegister.class);
+    private ClusterCoordinator getClusterCoordinator(ModuleProvider provider) {
+        return provider.getService(ClusterCoordinator.class);
     }
 
     private ClusterNodesQuery getClusterNodesQuery(ModuleProvider provider) {
         return provider.getService(ClusterNodesQuery.class);
     }
 
-    private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals)
-        throws InterruptedException {
-        int i = 20;
+    private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
+        return queryRemoteNodes(provider, goals, 20);
+    }
+
+    private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals,
+                                                  int cyclic) throws InterruptedException {
         do {
             List<RemoteInstance> instances = getClusterNodesQuery(provider).queryRemoteNodes();
             if (instances.size() == goals) {
@@ -252,7 +291,26 @@ private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals
                 Thread.sleep(1000);
             }
         }
-        while (--i > 0);
+        while (--cyclic > 0);
+        return Collections.emptyList();
+    }
+
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
+        throws InterruptedException {
+        return notifiedRemoteNodes(watcher, goals, 30);
+    }
+
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
+                                                     int cyclic) throws InterruptedException {
+        do {
+            List<RemoteInstance> instances = watcher.getRemoteInstances();
+            if (instances.size() == goals) {
+                return instances;
+            } else {
+                Thread.sleep(1000);
+            }
+        }
+        while (--cyclic > 0);
         return Collections.emptyList();
     }
 
@@ -275,4 +333,13 @@ private void validateServiceInstance(Address selfAddress, Address otherAddress,
         assertTrue(otherExist);
     }
 
+    class ClusterMockWatcher implements ClusterWatcher {
+        @Getter
+        private List<RemoteInstance> remoteInstances = new ArrayList<>();
+
+        @Override
+        public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
+            this.remoteInstances = remoteInstances;
+        }
+    }
 }
diff --git a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ITClusterModuleZookeeperProviderFunctionalTest.java b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ITClusterModuleZookeeperProviderFunctionalTest.java
index 5a7c7e7c6f5f..77d719f9a8b8 100644
--- a/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ITClusterModuleZookeeperProviderFunctionalTest.java
+++ b/oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ITClusterModuleZookeeperProviderFunctionalTest.java
@@ -18,11 +18,15 @@
 
 package org.apache.skywalking.oap.server.cluster.plugin.zookeeper;
 
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import lombok.Getter;
 import org.apache.curator.x.discovery.ServiceDiscovery;
+import org.apache.skywalking.oap.server.core.cluster.ClusterCoordinator;
 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.ClusterWatcher;
 import org.apache.skywalking.oap.server.core.cluster.RemoteInstance;
 import org.apache.skywalking.oap.server.core.remote.client.Address;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
@@ -82,10 +86,15 @@ public void registerRemote() throws Exception {
 
         Address selfAddress = new Address("127.0.0.1", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(provider, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertTrue(queryAddress.isSelf());
@@ -98,9 +107,16 @@ public void registerRemoteOfInternal() throws Exception {
 
         Address selfAddress = new Address("127.0.0.2", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(provider).registerRemote(instance);
+        ClusterCoordinator coordinator = getClusterCoordinator(provider);
+        ClusterMockWatcher watcher = new ClusterMockWatcher();
+        coordinator.registerWatcher(watcher);
+        coordinator.start();
+        coordinator.registerRemote(instance);
 
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
+
+        assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(provider, 1).size());
 
         assertEquals(1, remoteInstances.size());
         Address queryAddress = remoteInstances.get(0).getAddress();
@@ -114,15 +130,21 @@ public void registerRemoteOfReceiver() throws Exception {
         final String namespace = "register_remote_receiver";
         ModuleProvider providerA = createProvider(namespace);
         ModuleProvider providerB = createProvider(namespace);
-
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
         // Mixed or Aggregator
         Address selfAddress = new Address("127.0.0.3", 1000, true);
         RemoteInstance instance = new RemoteInstance(selfAddress);
-        getClusterRegister(providerA).registerRemote(instance);
+        coordinatorA.start();
+        coordinatorA.registerRemote(instance);
 
         // Receiver
-        List<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
+        List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstances.size());
+        assertEquals(1, queryRemoteNodes(providerB, 1).size());
         Address queryAddress = remoteInstances.get(0).getAddress();
         assertEquals(selfAddress, queryAddress);
         assertFalse(queryAddress.isSelf());
@@ -133,21 +155,29 @@ public void registerRemoteOfCluster() throws Exception {
         final String namespace = "register_remote_cluster";
         ModuleProvider providerA = createProvider(namespace);
         ModuleProvider providerB = createProvider(namespace);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.4", 1000, true);
         Address addressB = new Address("127.0.0.5", 1000, true);
-
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
-
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, queryRemoteNodes(providerB, 2).size());
     }
 
     @Test
@@ -155,33 +185,42 @@ public void unregisterRemoteOfCluster() throws Exception {
         final String namespace = "unregister_remote_cluster";
         ModuleProvider providerA = createProvider(namespace);
         ModuleProvider providerB = createProvider(namespace);
+        ClusterCoordinator coordinatorA = getClusterCoordinator(providerA);
+        ClusterMockWatcher watcherA = new ClusterMockWatcher();
+        coordinatorA.registerWatcher(watcherA);
+        coordinatorA.start();
+        ClusterCoordinator coordinatorB = getClusterCoordinator(providerB);
+        ClusterMockWatcher watcherB = new ClusterMockWatcher();
+        coordinatorB.registerWatcher(watcherB);
+        coordinatorB.start();
 
         Address addressA = new Address("127.0.0.4", 1000, true);
         Address addressB = new Address("127.0.0.5", 1000, true);
-
         RemoteInstance instanceA = new RemoteInstance(addressA);
         RemoteInstance instanceB = new RemoteInstance(addressB);
+        coordinatorA.registerRemote(instanceA);
+        coordinatorB.registerRemote(instanceB);
 
-        getClusterRegister(providerA).registerRemote(instanceA);
-        getClusterRegister(providerB).registerRemote(instanceB);
-
-        List<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
+        List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
         validateServiceInstance(addressA, addressB, remoteInstancesOfA);
+        assertEquals(2, queryRemoteNodes(providerA, 2).size());
 
-        List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
+        List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
         validateServiceInstance(addressB, addressA, remoteInstancesOfB);
+        assertEquals(2, queryRemoteNodes(providerB, 2).size());
 
         // unregister A
-        ClusterRegister register = getClusterRegister(providerA);
-        ServiceDiscovery<RemoteInstance> discoveryA = Whitebox.getInternalState(register, "serviceDiscovery");
+        ServiceDiscovery<RemoteInstance> discoveryA = Whitebox.getInternalState(providerA, "serviceDiscovery");
         discoveryA.close();
 
         // only B
-        remoteInstancesOfB = queryRemoteNodes(providerB, 1, 120);
+        remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1);
         assertEquals(1, remoteInstancesOfB.size());
+        assertEquals(1, queryRemoteNodes(providerB, 1).size());
+
         Address address = remoteInstancesOfB.get(0).getAddress();
-        assertEquals(address, addressB);
-        assertTrue(addressB.isSelf());
+        assertEquals(addressB, address);
+        assertTrue(address.isSelf());
     }
 
     private ClusterModuleZookeeperProvider createProvider(String namespace) throws Exception {
@@ -221,6 +260,10 @@ private ClusterRegister getClusterRegister(ModuleProvider provider) {
         return provider.getService(ClusterRegister.class);
     }
 
+    private ClusterCoordinator getClusterCoordinator(ModuleProvider provider) {
+        return provider.getService(ClusterCoordinator.class);
+    }
+
     private ClusterNodesQuery getClusterNodesQuery(ModuleProvider provider) {
         return provider.getService(ClusterNodesQuery.class);
     }
@@ -230,7 +273,7 @@ private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals
     }
 
     private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals,
-        int cyclic) throws InterruptedException {
+                                                  int cyclic) throws InterruptedException {
         do {
             List<RemoteInstance> instances = getClusterNodesQuery(provider).queryRemoteNodes();
             if (instances.size() == goals) {
@@ -243,6 +286,25 @@ private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals
         return Collections.emptyList();
     }
 
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
+        throws InterruptedException {
+        return notifiedRemoteNodes(watcher, goals, 20);
+    }
+
+    private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
+                                                     int cyclic) throws InterruptedException {
+        do {
+            List<RemoteInstance> instances = watcher.getRemoteInstances();
+            if (instances.size() == goals) {
+                return instances;
+            } else {
+                Thread.sleep(1000);
+            }
+        }
+        while (--cyclic > 0);
+        return Collections.emptyList();
+    }
+
     private void validateServiceInstance(Address selfAddress, Address otherAddress, List<RemoteInstance> queryResult) {
         assertEquals(2, queryResult.size());
 
@@ -260,4 +322,14 @@ private void validateServiceInstance(Address selfAddress, Address otherAddress,
         assertTrue(selfExist);
         assertTrue(otherExist);
     }
+
+    class ClusterMockWatcher implements ClusterWatcher {
+        @Getter
+        private List<RemoteInstance> remoteInstances = new ArrayList<>();
+
+        @Override
+        public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
+            this.remoteInstances = remoteInstances;
+        }
+    }
 }