diff --git a/dist-material/release-docs/LICENSE b/dist-material/release-docs/LICENSE index 16880c233d..9d20c83729 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 ba5cd05006..a045c61699 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 16ce31b485..57c48e6b58 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 @@ com.orbitz.consul consul-client - 1.4.2 + 1.5.3 com.google.guava 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 6f2cd9a72e..69612650cd 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.Getter; 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 b596daf197..50a4e6c730 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.ArrayList; 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 class ClusterModuleConsulProvider extends ModuleProvider { 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 777441a6b1..8ba79589e3 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.google.common.base.Strings; 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.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.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 class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery { public List queryRemoteNodes() { List remoteInstances = new ArrayList<>(); try { - initHealthChecker(); HealthClient healthClient = client.healthClient(); // Discover only "passing" nodes List nodes = healthClient.getHealthyServiceInstances(serviceName).getResponse(); @@ -88,33 +95,38 @@ public class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery { 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 class ConsulCoordinator implements ClusterRegister, ClusterNodesQuery { 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 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 { + @Override + public void notify(final Map newValues) { + try { + if (newValues.size() > 0) { + List 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-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 62edc69a1a..70233a4c41 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 class ClusterModuleConsulProviderTest { 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 class ClusterModuleConsulProviderTest { 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 78d3524de7..501a0fb539 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 @@ package org.apache.skywalking.oap.server.cluster.plugin.consul; 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 class ITClusterModuleConsulProviderFunctionalTest { ModuleProvider provider = createProvider(serviceName); Address selfAddress = new Address("127.0.0.1", 1000, true); - RemoteInstance instance = new RemoteInstance(selfAddress); - getClusterRegister(provider).registerRemote(instance); - List remoteInstances = queryRemoteNodes(provider, 1); + RemoteInstance instance = new RemoteInstance(selfAddress); + ClusterCoordinator coordinator = getClusterCoordinator(provider); + ClusterMockWatcher watcher = new ClusterMockWatcher(); + coordinator.registerWatcher(watcher); + coordinator.start(); + coordinator.registerRemote(instance); + + List 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 class ITClusterModuleConsulProviderFunctionalTest { Address selfAddress = new Address("127.0.0.2", 1002, true); RemoteInstance instance = new RemoteInstance(selfAddress); - getClusterRegister(provider).registerRemote(instance); - - List remoteInstances = queryRemoteNodes(provider, 1); + ClusterCoordinator coordinator = getClusterCoordinator(provider); + ClusterMockWatcher watcher = new ClusterMockWatcher(); + coordinator.registerWatcher(watcher); + coordinator.start(); + coordinator.registerRemote(instance); + List 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 class ITClusterModuleConsulProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(providerB, 1); + List 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 class ITClusterModuleConsulProviderFunctionalTest { 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 class ITClusterModuleConsulProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2); validateServiceInstance(addressB, addressA, remoteInstancesOfB); + assertEquals(2, queryRemoteNodes(providerB, 2).size()); } @Test @@ -160,6 +191,14 @@ public class ITClusterModuleConsulProviderFunctionalTest { 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 class ITClusterModuleConsulProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List 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 class ITClusterModuleConsulProviderFunctionalTest { 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 @@ public class ITClusterModuleConsulProviderFunctionalTest { 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 @@ public class ITClusterModuleConsulProviderFunctionalTest { 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 @@ public class ITClusterModuleConsulProviderFunctionalTest { return provider.getService(ClusterNodesQuery.class); } + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals) + throws InterruptedException { + return notifiedRemoteNodes(watcher, goals, 20); + } + + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals, + int cyclic) throws InterruptedException { + do { + List instances = watcher.getRemoteInstances(); + if (instances.size() == goals) { + return instances; + } else { + Thread.sleep(1000); + } + } + while (--cyclic > 0); + return Collections.emptyList(); + } + private List queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException { return queryRemoteNodes(provider, goals, 20); } @@ -287,4 +330,14 @@ public class ITClusterModuleConsulProviderFunctionalTest { assertTrue(selfExist); assertTrue(otherExist); } + + class ClusterMockWatcher implements ClusterWatcher { + @Getter + private List remoteInstances = new ArrayList<>(); + + @Override + public void onClusterNodesChanged(final List remoteInstances) { + this.remoteInstances = remoteInstances; + } + } } 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 c7510a69db..7deab5a25a 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 class ClusterModuleEtcdProvider extends ModuleProvider { 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 cee466a079..382e3e93d9 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.Client; 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.MetricsCreator; 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 class EtcdCoordinator implements ClusterRegister, ClusterNodesQuery { public List queryRemoteNodes() { List remoteInstances = new ArrayList<>(); try { - initHealthChecker(); - final KV kvClient = client.getKVClient(); final GetResponse response = kvClient.get( serviceNameBS, @@ -121,6 +121,9 @@ public class EtcdCoordinator implements ClusterRegister, ClusterNodesQuery { 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 class EtcdCoordinator implements ClusterRegister, ClusterNodesQuery { .port(selfAddress.getPort()) .build(); try { - initHealthChecker(); - final Lease leaseClient = client.getLeaseClient(); final long leaseID = leaseClient.grant(30L).get().getID(); @@ -198,4 +199,41 @@ public class EtcdCoordinator implements ClusterRegister, ClusterNodesQuery { "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-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 aecc4a952e..3fbde15353 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 class ITClusterModuleEtcdProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(provider, 1); + List 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 class ITClusterModuleEtcdProviderFunctionalTest { Address selfAddress = new Address("127.0.0.2", 1000, true); RemoteInstance instance = new RemoteInstance(selfAddress); - getClusterRegister(provider).registerRemote(instance); - - List remoteInstances = queryRemoteNodes(provider, 1); + ClusterCoordinator coordinator = getClusterCoordinator(provider); + ClusterMockWatcher watcher = new ClusterMockWatcher(); + coordinator.registerWatcher(watcher); + coordinator.start(); + coordinator.registerRemote(instance); + List 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 class ITClusterModuleEtcdProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(providerB, 1); + List 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 class ITClusterModuleEtcdProviderFunctionalTest { 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 class ITClusterModuleEtcdProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2); validateServiceInstance(addressB, addressA, remoteInstancesOfB); + assertEquals(2, queryRemoteNodes(providerB, 2).size()); } @Test @@ -151,29 +181,39 @@ public class ITClusterModuleEtcdProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List 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 @@ public class ITClusterModuleEtcdProviderFunctionalTest { 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 queryRemoteNodes(ModuleProvider provider, int goals) - throws InterruptedException { + private List queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException { return queryRemoteNodes(provider, goals, 20); } @@ -241,6 +280,25 @@ public class ITClusterModuleEtcdProviderFunctionalTest { return Collections.emptyList(); } + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals) + throws InterruptedException { + return notifiedRemoteNodes(watcher, goals, 20); + } + + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals, + int cyclic) throws InterruptedException { + do { + List 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 queryResult) { assertEquals(2, queryResult.size()); @@ -260,4 +318,13 @@ public class ITClusterModuleEtcdProviderFunctionalTest { assertTrue(otherExist); } + class ClusterMockWatcher implements ClusterWatcher { + @Getter + private List remoteInstances = new ArrayList<>(); + + @Override + public void onClusterNodesChanged(final List remoteInstances) { + this.remoteInstances = remoteInstances; + } + } } 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 05d3b4798f..7aefc876b6 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 @@ ${project.version} test + + org.apache.skywalking + library-kubernetes-support + ${project.version} + - \ No newline at end of file + 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 5b33c78270..0169ccd282 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 class ClusterModuleKubernetesProvider extends ModuleProvider { 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 60ee79b6f7..46ec16811e 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.HealthCheckMetrics; 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 remoteInstanceMap; + private volatile List 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 queryRemoteNodes() { try { - initHealthChecker(); List pods = NamespacedPodListInformer.INFORMER.listPods().orElseGet(this::selfPod); if (log.isDebugEnabled()) { List 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 class KubernetesCoordinator implements ClusterRegister, ClusterNodesQuery 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 class KubernetesCoordinator implements ClusterRegister, ClusterNodesQuery @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 @@ public class KubernetesCoordinator implements ClusterRegister, ClusterNodesQuery 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 { + + @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 updatedInstances = new ArrayList<>(this.remoteInstanceMap.keySet()); + if (this.latestInstances.size() != updatedInstances.size() || !this.latestInstances.containsAll(updatedInstances)) { + List remoteInstances = new ArrayList<>(this.remoteInstanceMap.values()); + this.latestInstances = updatedInstances; + checkHealth(remoteInstances); + notifyWatchers(remoteInstances); + } + } + + private void checkHealth(List 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 9c3d2edc9a..265a18f219 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 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 eventHandler) throws IOException { + KubernetesClient.setDefault(); + CoreV1Api coreV1Api = new CoreV1Api(); factory = new SharedInformerFactory(executorService); SharedIndexInformer podSharedIndexInformer = factory.sharedIndexInformerFor( @@ -91,6 +85,8 @@ public enum NamespacedPodListInformer { V1Pod.class, V1PodList.class ); + podSharedIndexInformer.addEventHandler(eventHandler); + factory.startAllRegisteredInformers(); podLister = new Lister<>(podSharedIndexInformer.getIndexer()); } 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 ba8d230b6f..87c003cf0d 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.V1ObjectMeta; 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.core.classloader.annotations.PrepareForTest; 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 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 remoteInstances = Whitebox.invokeMethod(coordinator, "queryRemoteNodes"); Assert.assertEquals(5, remoteInstances.size()); @@ -104,17 +154,161 @@ public class KubernetesCoordinatorTest { } - 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 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 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 remoteInstancesOfA = watcherA.getRemoteInstances(); + validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, coordinatorA.queryRemoteNodes().size()); + + List 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 remoteInstancesOfA = watcherA.getRemoteInstances(); + validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, coordinatorA.queryRemoteNodes().size()); + + List 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 mockPodList() { @@ -123,10 +317,39 @@ public class KubernetesCoordinatorTest { 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 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 remoteInstances = new ArrayList<>(); + + @Override + public void onClusterNodesChanged(final List remoteInstances) { + this.remoteInstances = remoteInstances; + } + } } 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 5b685d7925..47019f4f92 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.NamingFactory; 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 class ClusterModuleNacosProvider extends ModuleProvider { 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 b5fa3daf46..f6f9a44eb5 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 class NacosCoordinator implements ClusterRegister, ClusterNodesQuery { public List queryRemoteNodes() { List remoteInstances = new ArrayList<>(); try { - initHealthChecker(); List instances = namingService.selectInstances(config.getServiceName(), true); if (CollectionUtils.isNotEmpty(instances)) { instances.forEach(instance -> { @@ -78,6 +83,11 @@ public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery { healthChecker.unHealth(e); throw new ServiceQueryException(e.getMessage()); } + + if (log.isDebugEnabled()) { + log.debug("Nacos cluster instances:{}", remoteInstances); + } + return remoteInstances; } @@ -86,17 +96,16 @@ public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery { 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 @@ public class NacosCoordinator implements ClusterRegister, ClusterNodesQuery { 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 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 instances = ((NamingEvent) event).getInstances(); + List 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-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 396ad5c521..f575b961a0 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 class ITClusterModuleNacosProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(provider, 1); + List 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 class ITClusterModuleNacosProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(provider, 1); + List 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 class ITClusterModuleNacosProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(providerB, 1); + List 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 class ITClusterModuleNacosProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2); validateServiceInstance(addressB, addressA, remoteInstancesOfB); + assertEquals(2, queryRemoteNodes(providerB, 2).size()); } @Test @@ -159,30 +186,39 @@ public class ITClusterModuleNacosProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List 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 @@ public class ITClusterModuleNacosProviderFunctionalTest { 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 queryRemoteNodes(ModuleProvider provider, int goals) - throws InterruptedException { - int i = 20; + private List queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException { + return queryRemoteNodes(provider, goals, 20); + } + + private List queryRemoteNodes(ModuleProvider provider, int goals, + int cyclic) throws InterruptedException { do { List instances = getClusterNodesQuery(provider).queryRemoteNodes(); if (instances.size() == goals) { @@ -252,7 +291,26 @@ public class ITClusterModuleNacosProviderFunctionalTest { Thread.sleep(1000); } } - while (--i > 0); + while (--cyclic > 0); + return Collections.emptyList(); + } + + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals) + throws InterruptedException { + return notifiedRemoteNodes(watcher, goals, 30); + } + + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals, + int cyclic) throws InterruptedException { + do { + List instances = watcher.getRemoteInstances(); + if (instances.size() == goals) { + return instances; + } else { + Thread.sleep(1000); + } + } + while (--cyclic > 0); return Collections.emptyList(); } @@ -275,4 +333,13 @@ public class ITClusterModuleNacosProviderFunctionalTest { assertTrue(otherExist); } + class ClusterMockWatcher implements ClusterWatcher { + @Getter + private List remoteInstances = new ArrayList<>(); + + @Override + public void onClusterNodesChanged(final List remoteInstances) { + this.remoteInstances = remoteInstances; + } + } } 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 7c4b91c3a9..d0d4b57afc 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.ModuleStartException; 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 class ClusterModuleStandaloneProvider extends ModuleProvider { @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 class ClusterModuleStandaloneProvider extends ModuleProvider { @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 8b160a740e..99c09e0e21 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 class StandaloneManager implements ClusterNodesQuery, ClusterRegister { 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 4acb88c5b1..8e2c5f4471 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.retry.ExponentialBackoffRetry; 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 class ClusterModuleZookeeperProvider extends ModuleProvider { 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 c33da237dd..6bb2161ef6 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.ArrayList; 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 serviceDiscovery) throws Exception { + final ServiceDiscovery 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 thisInstance = ServiceInstance.builder().name(REMOTE_NAME_PATH) .id(UUID.randomUUID() .toString()) @@ -81,8 +84,6 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery .build(); serviceDiscovery.registerService(thisInstance); - - this.selfAddress = remoteInstance.getAddress(); this.healthChecker.health(); } catch (Throwable e) { this.healthChecker.unHealth(e); @@ -94,7 +95,6 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery public List queryRemoteNodes() { List remoteInstances = new ArrayList<>(20); try { - initHealthChecker(); List> serviceInstances = serviceCache.getInstances(); serviceInstances.forEach(serviceInstance -> { RemoteInstance instance = serviceInstance.getPayload(); @@ -115,6 +115,10 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery 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 @@ public class ZookeeperCoordinator implements ClusterRegister, ClusterNodesQuery 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 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-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 5a7c7e7c6f..77d719f9a8 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 class ITClusterModuleZookeeperProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(provider, 1); + List 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 class ITClusterModuleZookeeperProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(provider, 1); + List 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 class ITClusterModuleZookeeperProviderFunctionalTest { 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 remoteInstances = queryRemoteNodes(providerB, 1); + List 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 class ITClusterModuleZookeeperProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2); validateServiceInstance(addressB, addressA, remoteInstancesOfB); + assertEquals(2, queryRemoteNodes(providerB, 2).size()); } @Test @@ -155,33 +185,42 @@ public class ITClusterModuleZookeeperProviderFunctionalTest { 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 remoteInstancesOfA = queryRemoteNodes(providerA, 2); + List remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2); validateServiceInstance(addressA, addressB, remoteInstancesOfA); + assertEquals(2, queryRemoteNodes(providerA, 2).size()); - List remoteInstancesOfB = queryRemoteNodes(providerB, 2); + List remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2); validateServiceInstance(addressB, addressA, remoteInstancesOfB); + assertEquals(2, queryRemoteNodes(providerB, 2).size()); // unregister A - ClusterRegister register = getClusterRegister(providerA); - ServiceDiscovery discoveryA = Whitebox.getInternalState(register, "serviceDiscovery"); + ServiceDiscovery 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 @@ public class ITClusterModuleZookeeperProviderFunctionalTest { 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 @@ public class ITClusterModuleZookeeperProviderFunctionalTest { } private List queryRemoteNodes(ModuleProvider provider, int goals, - int cyclic) throws InterruptedException { + int cyclic) throws InterruptedException { do { List instances = getClusterNodesQuery(provider).queryRemoteNodes(); if (instances.size() == goals) { @@ -243,6 +286,25 @@ public class ITClusterModuleZookeeperProviderFunctionalTest { return Collections.emptyList(); } + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals) + throws InterruptedException { + return notifiedRemoteNodes(watcher, goals, 20); + } + + private List notifiedRemoteNodes(ClusterMockWatcher watcher, int goals, + int cyclic) throws InterruptedException { + do { + List 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 queryResult) { assertEquals(2, queryResult.size()); @@ -260,4 +322,14 @@ public class ITClusterModuleZookeeperProviderFunctionalTest { assertTrue(selfExist); assertTrue(otherExist); } + + class ClusterMockWatcher implements ClusterWatcher { + @Getter + private List remoteInstances = new ArrayList<>(); + + @Override + public void onClusterNodesChanged(final List remoteInstances) { + this.remoteInstances = remoteInstances; + } + } } 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 b32c9489ab..91f411927e 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.annotation.AnnotationScan; 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 class CoreModuleProvider extends ModuleProvider { 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 class CoreModuleProvider extends ModuleProvider { 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 class CoreModuleProvider extends ModuleProvider { .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 class CoreModuleProvider extends ModuleProvider { } 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 0000000000..a78205da38 --- /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 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 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 9e65fcc5b6..feaf4d5c83 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 class ClusterModule extends ModuleDefine { 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 0000000000..dce88afb41 --- /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 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 0000000000..e595a6a421 --- /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 c7ee1129b6..08c64cf33e 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 java.util.stream.Collectors; 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 class RemoteClientManager implements Service { 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 instanceList) { if (gauge == null) { gauge = moduleDefineHolder.find(TelemetryModule.NAME) .provider() @@ -111,40 +126,33 @@ public class RemoteClientManager implements Service { ); } 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 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 @@ public class RemoteClientManager implements Service { * 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 @@ public class RemoteClientManager implements Service { } } + @Override + public void onClusterNodesChanged(List 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 0000000000..49203abb4e --- /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 fac17a1b5d..100cecee21 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 class ServerStatusService implements Service { .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); + } }