kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

The cluster coordinator support watch mechanism for notifying `RemoteClientManager` and `ServerStatusService`. (#10192)

This commit is contained in:
Wan Kai 2022-12-22 20:52:55 +08:00 committed by GitHub
parent 3663a4ac5f
commit 995d790b42
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 1230 additions and 274 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
dist-material/release-docs/LICENSE
View File

@ -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

1
docs/en/changes/changes.md
View File

@ -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

2
oap-server/server-cluster-plugin/cluster-consul-plugin/pom.xml
View File

@ -37,7 +37,7 @@
<dependency>
<groupId>com.orbitz.consul</groupId>
<artifactId>consul-client</artifactId>
<version>1.4.2</version>
<version>1.5.3</version>
<exclusions>
<exclusion>
<groupId>com.google.guava</groupId>

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

@ -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;

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

@ -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

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

@ -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<RemoteInstance> queryRemoteNodes() {
List<RemoteInstance> remoteInstances = new ArrayList<>();
try {
initHealthChecker();
HealthClient healthClient = client.healthClient();
// Discover only "passing" nodes
List<ServiceHealth> nodes = healthClient.getHealthyServiceInstances(serviceName).getResponse();
@ -88,33 +95,38 @@ public 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<RemoteInstance> remoteInstances) {
ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
if (healthStatus.isHealth()) {
this.healthChecker.health();
} else {
this.healthChecker.unHealth(healthStatus.getReason());
}
}
@Override
public void start() {
initHealthChecker();
ServiceHealthCache svHealth = ServiceHealthCache.newCache(client.healthClient(), serviceName, true,
QueryOptions.BLANK, 5);
svHealth.addListener(new ConsulEventListener());
svHealth.start();
}
/**
* Notice: If the consul version > v1.10.0, the `consul-client ConsulCache` will throw error response:
* "com.orbitz.consul.ConsulException: Consul cluster has no elected leader" and fails to retrieve data.
* This is a known issue but doesn't release yet, can refer to: https://github.com/rickfast/consul-client/pull/456
*/
class ConsulEventListener implements ConsulCache.Listener<ServiceHealthKey, ServiceHealth> {
@Override
public void notify(final Map<ServiceHealthKey, ServiceHealth> newValues) {
try {
if (newValues.size() > 0) {
List<RemoteInstance> remoteInstances = new ArrayList<>(newValues.size());
newValues.values().forEach(serviceHealth -> {
if (StringUtil.isNotBlank(serviceHealth.getService().getAddress())) {
RemoteInstance remoteInstance = buildRemoteInstance(
serviceHealth.getService().getAddress(), serviceHealth.getService().getPort());
remoteInstances.add(remoteInstance);
}
});
checkHealth(remoteInstances);
notifyWatchers(remoteInstances);
}
} catch (Throwable e) {
healthChecker.unHealth(e);
log.error("Failed to notify and update remote instances.", e);
}
}
}
}

4
oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ClusterModuleConsulProviderTest.java
View File

@ -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);
}
}
}

135
oap-server/server-cluster-plugin/cluster-consul-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/consul/ITClusterModuleConsulProviderFunctionalTest.java
View File

@ -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<RemoteInstance> 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<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertTrue(queryAddress.isSelf());
@ -103,11 +112,16 @@ public class ITClusterModuleConsulProviderFunctionalTest {
Address selfAddress = new Address("127.0.0.2", 1002, true);
RemoteInstance instance = new RemoteInstance(selfAddress);
getClusterRegister(provider).registerRemote(instance);
List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
ClusterCoordinator coordinator = getClusterCoordinator(provider);
ClusterMockWatcher watcher = new ClusterMockWatcher();
coordinator.registerWatcher(watcher);
coordinator.start();
coordinator.registerRemote(instance);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals("127.0.1.2", queryAddress.getHost());
assertEquals(1001, queryAddress.getPort());
@ -119,15 +133,22 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertFalse(queryAddress.isSelf());
@ -138,6 +159,14 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
}
@Test
@ -160,6 +191,14 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
// unregister A
Consul client = Whitebox.getInternalState(providerA, "client");
@ -182,8 +223,9 @@ public 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<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
throws InterruptedException {
return notifiedRemoteNodes(watcher, goals, 20);
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = watcher.getRemoteInstances();
if (instances.size() == goals) {
return instances;
} else {
Thread.sleep(1000);
}
}
while (--cyclic > 0);
return Collections.emptyList();
}
private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
return queryRemoteNodes(provider, goals, 20);
}
@ -287,4 +330,14 @@ public class ITClusterModuleConsulProviderFunctionalTest {
assertTrue(selfExist);
assertTrue(otherExist);
}
class ClusterMockWatcher implements ClusterWatcher {
@Getter
private List<RemoteInstance> remoteInstances = new ArrayList<>();
@Override
public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
this.remoteInstances = remoteInstances;
}
}
}

2
oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ClusterModuleEtcdProvider.java
View File

@ -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);
}

52
oap-server/server-cluster-plugin/cluster-etcd-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/EtcdCoordinator.java
View File

@ -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<RemoteInstance> queryRemoteNodes() {
List<RemoteInstance> remoteInstances = new ArrayList<>();
try {
initHealthChecker();
final KV kvClient = client.getKVClient();
final GetResponse response = kvClient.get(
serviceNameBS,
@ -121,6 +121,9 @@ public 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() {
}
}
}

113
oap-server/server-cluster-plugin/cluster-etcd-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/etcd/ITClusterModuleEtcdProviderFunctionalTest.java
View File

@ -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<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertTrue(queryAddress.isSelf());
@ -94,11 +102,16 @@ public class ITClusterModuleEtcdProviderFunctionalTest {
Address selfAddress = new Address("127.0.0.2", 1000, true);
RemoteInstance instance = new RemoteInstance(selfAddress);
getClusterRegister(provider).registerRemote(instance);
List<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
ClusterCoordinator coordinator = getClusterCoordinator(provider);
ClusterMockWatcher watcher = new ClusterMockWatcher();
coordinator.registerWatcher(watcher);
coordinator.start();
coordinator.registerRemote(instance);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals("127.0.1.2", queryAddress.getHost());
assertEquals(1000, queryAddress.getPort());
@ -110,15 +123,22 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertFalse(queryAddress.isSelf());
@ -129,6 +149,14 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
}
@Test
@ -151,29 +181,39 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
// unregister A
Client client = Whitebox.getInternalState(getClusterRegister(providerA), "client");
Client client = Whitebox.getInternalState(coordinatorA, "client");
client.close();
// only B
remoteInstancesOfB = queryRemoteNodes(providerB, 1, 120);
remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1, 120);
assertEquals(1, remoteInstancesOfB.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address address = remoteInstancesOfB.get(0).getAddress();
assertEquals(address, addressB);
assertTrue(addressB.isSelf());
@ -214,16 +254,15 @@ 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<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals)
throws InterruptedException {
private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
return queryRemoteNodes(provider, goals, 20);
}
@ -241,6 +280,25 @@ public class ITClusterModuleEtcdProviderFunctionalTest {
return Collections.emptyList();
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
throws InterruptedException {
return notifiedRemoteNodes(watcher, goals, 20);
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = watcher.getRemoteInstances();
if (instances.size() == goals) {
return instances;
} else {
Thread.sleep(1000);
}
}
while (--cyclic > 0);
return Collections.emptyList();
}
private void validateServiceInstance(Address selfAddress, Address otherAddress,
List<RemoteInstance> queryResult) {
assertEquals(2, queryResult.size());
@ -260,4 +318,13 @@ public class ITClusterModuleEtcdProviderFunctionalTest {
assertTrue(otherExist);
}
class ClusterMockWatcher implements ClusterWatcher {
@Getter
private List<RemoteInstance> remoteInstances = new ArrayList<>();
@Override
public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
this.remoteInstances = remoteInstances;
}
}
}

7
oap-server/server-cluster-plugin/cluster-kubernetes-plugin/pom.xml
View File

@ -44,5 +44,10 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>library-kubernetes-support</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>
</project>

3
oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/ClusterModuleKubernetesProvider.java
View File

@ -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

111
oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinator.java
View File

@ -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<String, RemoteInstance> remoteInstanceMap;
private volatile List<String> latestInstances;
public KubernetesCoordinator(final ModuleDefineHolder manager,
final ClusterModuleKubernetesConfig config) {
this.uid = new UidEnvSupplier(config.getUidEnvName()).get();
this.manager = manager;
this.config = config;
this.remoteInstanceMap = new ConcurrentHashMap<>(20);
this.latestInstances = new ArrayList<>(20);
}
@Override
public List<RemoteInstance> queryRemoteNodes() {
try {
initHealthChecker();
List<V1Pod> pods = NamespacedPodListInformer.INFORMER.listPods().orElseGet(this::selfPod);
if (log.isDebugEnabled()) {
List<String> uidList = pods
.stream()
.map(item -> item.getMetadata().getUid())
.collect(Collectors.toList());
log.debug("[kubernetes cluster pods uid list]:{}", uidList.toString());
log.debug("[kubernetes cluster pods uid list]:{}", uidList);
}
if (port == -1) {
port = manager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort();
@ -80,6 +95,10 @@ public 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<V1Pod> {
@Override
public void onAdd(final V1Pod obj) {
updateRemoteInstances(obj, ADDED);
}
@Override
public void onUpdate(final V1Pod oldObj, final V1Pod newObj) {
updateRemoteInstances(newObj, MODIFIED);
}
@Override
public void onDelete(final V1Pod obj, final boolean deletedFinalStateUnknown) {
updateRemoteInstances(obj, DELETED);
}
}
/**
* When a remote instance up/off line, will receive multi event according to the pod status.
* To avoid notify the watchers too frequency, here use a `remoteInstanceMap` to cache them.
* Only notify watchers once when the instances changed.
*/
private void updateRemoteInstances(V1Pod pod, EventType event) {
try {
initHealthChecker();
if (StringUtil.isNotBlank(pod.getStatus().getPodIP())) {
if (port == -1) {
port = manager.find(CoreModule.NAME).provider().getService(ConfigService.class).getGRPCPort();
}
RemoteInstance remoteInstance = new RemoteInstance(
new Address(pod.getStatus().getPodIP(), this.port, pod.getMetadata().getUid().equals(uid)));
switch (event) {
case ADDED:
case MODIFIED:
if ("Running".equalsIgnoreCase(pod.getStatus().getPhase())) {
this.remoteInstanceMap.put(remoteInstance.getAddress().toString(), remoteInstance);
}
break;
case DELETED:
this.remoteInstanceMap.remove(remoteInstance.getAddress().toString());
break;
default:
return;
}
updateRemoteInstances();
}
} catch (Throwable e) {
healthChecker.unHealth(e);
log.error("Failed to notify RemoteInstances update.", e);
}
}
private void updateRemoteInstances() {
List<String> updatedInstances = new ArrayList<>(this.remoteInstanceMap.keySet());
if (this.latestInstances.size() != updatedInstances.size() || !this.latestInstances.containsAll(updatedInstances)) {
List<RemoteInstance> remoteInstances = new ArrayList<>(this.remoteInstanceMap.values());
this.latestInstances = updatedInstances;
checkHealth(remoteInstances);
notifyWatchers(remoteInstances);
}
}
private void checkHealth(List<RemoteInstance> remoteInstances) {
ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
if (healthStatus.isHealth()) {
this.healthChecker.health();
} else {
this.healthChecker.unHealth(healthStatus.getReason());
}
}
}

22
oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/NamespacedPodListInformer.java
View File

@ -18,25 +18,22 @@
package org.apache.skywalking.oap.server.cluster.plugin.kubernetes;
import io.kubernetes.client.informer.ResourceEventHandler;
import io.kubernetes.client.informer.SharedIndexInformer;
import io.kubernetes.client.informer.SharedInformerFactory;
import io.kubernetes.client.informer.cache.Lister;
import io.kubernetes.client.openapi.ApiClient;
import io.kubernetes.client.openapi.Configuration;
import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.openapi.models.V1Pod;
import io.kubernetes.client.openapi.models.V1PodList;
import io.kubernetes.client.util.Config;
import java.io.IOException;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.library.kubernetes.KubernetesClient;
import static java.util.Objects.isNull;
@Slf4j
@ -65,21 +62,18 @@ public enum NamespacedPodListInformer {
}));
}
public synchronized void init(ClusterModuleKubernetesConfig podConfig) {
public synchronized void init(ClusterModuleKubernetesConfig podConfig, ResourceEventHandler<V1Pod> eventHandler) {
try {
doStartPodInformer(podConfig);
doStartPodInformer(podConfig, eventHandler);
} catch (IOException e) {
log.error("cannot connect with api server in kubernetes", e);
}
}
private void doStartPodInformer(ClusterModuleKubernetesConfig podConfig) throws IOException {
ApiClient apiClient = Config.defaultClient();
apiClient.setHttpClient(apiClient.getHttpClient().newBuilder().readTimeout(0, TimeUnit.SECONDS).build());
Configuration.setDefaultApiClient(apiClient);
CoreV1Api coreV1Api = new CoreV1Api(apiClient);
private void doStartPodInformer(ClusterModuleKubernetesConfig podConfig, ResourceEventHandler<V1Pod> eventHandler) throws IOException {
KubernetesClient.setDefault();
CoreV1Api coreV1Api = new CoreV1Api();
factory = new SharedInformerFactory(executorService);
SharedIndexInformer<V1Pod> podSharedIndexInformer = factory.sharedIndexInformerFor(
@ -91,6 +85,8 @@ public enum NamespacedPodListInformer {
V1Pod.class, V1PodList.class
);
podSharedIndexInformer.addEventHandler(eventHandler);
factory.startAllRegisteredInformers();
podLister = new Lister<>(podSharedIndexInformer.getIndexer());
}

287
oap-server/server-cluster-plugin/cluster-kubernetes-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/kubernetes/KubernetesCoordinatorTest.java
View File

@ -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<RemoteInstance> remoteInstances = Whitebox.invokeMethod(coordinator, "queryRemoteNodes");
Assert.assertEquals(1, remoteInstances.size());
Assert.assertEquals(selfAddress, remoteInstances.get(0).getAddress());
Assert.assertEquals(addressA, remoteInstances.get(0).getAddress());
}
@Test
public void queryRemoteNodesWhenInformerWork() throws Exception {
ModuleProvider provider = createProvider(SELF_UID + "0");
KubernetesCoordinator coordinator = getClusterCoordinator(provider);
coordinator.start();
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinator, SELF_UID + "0");
PowerMockito.doReturn(Optional.of(mockPodList())).when(NamespacedPodListInformer.INFORMER).listPods();
List<RemoteInstance> remoteInstances = Whitebox.invokeMethod(coordinator, "queryRemoteNodes");
Assert.assertEquals(5, remoteInstances.size());
@ -104,17 +154,161 @@ public 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<RemoteInstance> remoteInstances = watcher.getRemoteInstances();
assertEquals(1, remoteInstances.size());
assertEquals(1, coordinatorA.queryRemoteNodes().size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(addressA, queryAddress);
assertTrue(queryAddress.isSelf());
}
@Test
public void registerRemoteOfReceiver() throws Exception {
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
ClusterMockWatcher watcherB = new ClusterMockWatcher();
coordinatorB.registerWatcher(watcherB);
PowerMockito.doReturn(Optional.of(Collections.singletonList(podA)))
.when(NamespacedPodListInformer.INFORMER)
.listPods();
RemoteInstance instance = new RemoteInstance(addressA);
coordinatorA.registerRemote(instance);
KubernetesCoordinator.K8sResourceEventHandler listener = coordinatorB.new K8sResourceEventHandler();
listener.onAdd(podA);
// Receiver
List<RemoteInstance> remoteInstances = watcherB.getRemoteInstances();
assertEquals(1, remoteInstances.size());
assertEquals(1, coordinatorB.queryRemoteNodes().size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(addressA, queryAddress);
assertFalse(queryAddress.isSelf());
}
@Test
public void registerRemoteOfCluster() throws Exception {
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
ClusterMockWatcher watcherA = new ClusterMockWatcher();
coordinatorA.registerWatcher(watcherA);
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
ClusterMockWatcher watcherB = new ClusterMockWatcher();
coordinatorB.registerWatcher(watcherB);
PowerMockito.doReturn(Optional.of(Arrays.asList(podA, podB)))
.when(NamespacedPodListInformer.INFORMER)
.listPods();
RemoteInstance instanceA = new RemoteInstance(addressA);
RemoteInstance instanceB = new RemoteInstance(addressB);
coordinatorA.registerRemote(instanceA);
coordinatorB.registerRemote(instanceB);
KubernetesCoordinator.K8sResourceEventHandler listenerA = coordinatorA.new K8sResourceEventHandler();
listenerA.onAdd(podA);
listenerA.onAdd(podB);
KubernetesCoordinator.K8sResourceEventHandler listenerB = coordinatorB.new K8sResourceEventHandler();
listenerB.onAdd(podA);
listenerB.onAdd(podB);
List<RemoteInstance> remoteInstancesOfA = watcherA.getRemoteInstances();
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, coordinatorA.queryRemoteNodes().size());
List<RemoteInstance> remoteInstancesOfB = watcherB.getRemoteInstances();
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, coordinatorB.queryRemoteNodes().size());
}
@Test
public void deregisterRemoteOfCluster() throws Exception {
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorA, SELF_UID);
ClusterMockWatcher watcherA = new ClusterMockWatcher();
coordinatorA.registerWatcher(watcherA);
MemberModifier.field(KubernetesCoordinator.class, "uid").set(coordinatorB, REMOTE_UID);
ClusterMockWatcher watcherB = new ClusterMockWatcher();
coordinatorB.registerWatcher(watcherB);
PowerMockito.doReturn(Optional.of(Arrays.asList(podA, podB)))
.when(NamespacedPodListInformer.INFORMER)
.listPods();
RemoteInstance instanceA = new RemoteInstance(addressA);
RemoteInstance instanceB = new RemoteInstance(addressB);
coordinatorA.registerRemote(instanceA);
coordinatorB.registerRemote(instanceB);
KubernetesCoordinator.K8sResourceEventHandler listenerA = coordinatorA.new K8sResourceEventHandler();
listenerA.onAdd(podA);
listenerA.onAdd(podB);
KubernetesCoordinator.K8sResourceEventHandler listenerB = coordinatorB.new K8sResourceEventHandler();
listenerB.onAdd(podA);
listenerB.onAdd(podB);
List<RemoteInstance> remoteInstancesOfA = watcherA.getRemoteInstances();
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, coordinatorA.queryRemoteNodes().size());
List<RemoteInstance> remoteInstancesOfB = watcherB.getRemoteInstances();
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, coordinatorB.queryRemoteNodes().size());
// deregister A
listenerB.onDelete(podA, false);
PowerMockito.doReturn(Optional.of(Collections.singletonList(podB)))
.when(NamespacedPodListInformer.INFORMER)
.listPods();
// only B
remoteInstancesOfB = watcherB.getRemoteInstances();
assertEquals(1, remoteInstancesOfB.size());
assertEquals(1, coordinatorB.queryRemoteNodes().size());
Address address = remoteInstancesOfB.get(0).getAddress();
assertEquals(addressB, address);
assertTrue(address.isSelf());
}
private ClusterModuleKubernetesProvider createProvider(String uidEnvName)
throws ModuleStartException {
ClusterModuleKubernetesProvider provider = new ClusterModuleKubernetesProvider();
ClusterModuleKubernetesConfig config = new ClusterModuleKubernetesConfig();
provider.newConfigCreator().onInitialized(config);
config.setNamespace("default");
config.setLabelSelector("app=oap");
config.setUidEnvName(uidEnvName);
provider.setManager(moduleManager);
provider.prepare();
provider.start();
provider.notifyAfterCompleted();
return provider;
}
private KubernetesCoordinator getClusterCoordinator(ModuleProvider provider) {
return (KubernetesCoordinator) provider.getService(ClusterCoordinator.class);
}
private V1Pod mockPod(String uid, String ip) {
V1Pod v1Pod = new V1Pod();
v1Pod.setMetadata(new V1ObjectMeta());
v1Pod.setStatus(new V1PodStatus());
v1Pod.getStatus().setPhase("Running");
v1Pod.getMetadata().setUid(uid);
v1Pod.getStatus().setPodIP(ip);
return v1Pod;
}
private List<V1Pod> mockPodList() {
@ -123,10 +317,39 @@ 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<RemoteInstance> queryResult) {
assertEquals(2, queryResult.size());
boolean selfExist = false, otherExist = false;
for (RemoteInstance instance : queryResult) {
Address queryAddress = instance.getAddress();
if (queryAddress.equals(selfAddress) && queryAddress.isSelf()) {
selfExist = true;
} else if (queryAddress.equals(otherAddress) && !queryAddress.isSelf()) {
otherExist = true;
}
}
assertTrue(selfExist);
assertTrue(otherExist);
}
class ClusterMockWatcher implements ClusterWatcher {
@Getter
private List<RemoteInstance> remoteInstances = new ArrayList<>();
@Override
public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
this.remoteInstances = remoteInstances;
}
}
}

8
oap-server/server-cluster-plugin/cluster-nacos-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/nacos/ClusterModuleNacosProvider.java
View File

@ -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

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

@ -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<RemoteInstance> queryRemoteNodes() {
List<RemoteInstance> remoteInstances = new ArrayList<>();
try {
initHealthChecker();
List<Instance> instances = namingService.selectInstances(config.getServiceName(), true);
if (CollectionUtils.isNotEmpty(instances)) {
instances.forEach(instance -> {
@ -78,6 +83,11 @@ public 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<RemoteInstance> remoteInstances) {
ClusterHealthStatus healthStatus = OAPNodeChecker.isHealth(remoteInstances);
if (healthStatus.isHealth()) {
this.healthChecker.health();
} else {
this.healthChecker.unHealth(healthStatus.getReason());
}
}
@Override
public void start() throws ModuleStartException {
initHealthChecker();
try {
namingService.subscribe(config.getServiceName(), new NacosEventListener());
} catch (NacosException e) {
throw new ModuleStartException("Failed to start cluster coordinator.", e);
}
}
class NacosEventListener implements EventListener {
@Override
public void onEvent(final Event event) {
try {
List<Instance> instances = ((NamingEvent) event).getInstances();
List<RemoteInstance> remoteInstances = new ArrayList<>(instances.size());
instances.forEach(instance -> {
RemoteInstance remoteInstance = buildRemoteInstance(instance);
if (instance.isHealthy() && instance.isEnabled()) {
remoteInstances.add(remoteInstance);
}
});
checkHealth(remoteInstances);
notifyWatchers(remoteInstances);
} catch (Throwable e) {
healthChecker.unHealth(e);
log.error("Failed to notify and update remote instances.", e);
}
}
}
}

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

@ -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<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertTrue(queryAddress.isSelf());
@ -102,11 +110,16 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals("127.0.1.2", queryAddress.getHost());
assertEquals(1000, queryAddress.getPort());
@ -118,15 +131,21 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertFalse(queryAddress.isSelf());
@ -137,21 +156,29 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
}
@Test
@ -159,30 +186,39 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
// deregister A
ClusterRegister register = getClusterRegister(providerA);
NamingService namingServiceA = Whitebox.getInternalState(register, "namingService");
NamingService namingServiceA = Whitebox.getInternalState(coordinatorA, "namingService");
namingServiceA.deregisterInstance(serviceName, addressA.getHost(), addressA.getPort());
// only B
remoteInstancesOfB = queryRemoteNodes(providerB, 1);
remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstancesOfB.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address address = remoteInstancesOfB.get(0).getAddress();
assertEquals(addressB, address);
assertTrue(address.isSelf());
@ -233,17 +269,20 @@ 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<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals)
throws InterruptedException {
int i = 20;
private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals) throws InterruptedException {
return queryRemoteNodes(provider, goals, 20);
}
private List<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals,
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = getClusterNodesQuery(provider).queryRemoteNodes();
if (instances.size() == goals) {
@ -252,7 +291,26 @@ public class ITClusterModuleNacosProviderFunctionalTest {
Thread.sleep(1000);
}
}
while (--i > 0);
while (--cyclic > 0);
return Collections.emptyList();
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
throws InterruptedException {
return notifiedRemoteNodes(watcher, goals, 30);
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = watcher.getRemoteInstances();
if (instances.size() == goals) {
return instances;
} else {
Thread.sleep(1000);
}
}
while (--cyclic > 0);
return Collections.emptyList();
}
@ -275,4 +333,13 @@ public class ITClusterModuleNacosProviderFunctionalTest {
assertTrue(otherExist);
}
class ClusterMockWatcher implements ClusterWatcher {
@Getter
private List<RemoteInstance> remoteInstances = new ArrayList<>();
@Override
public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
this.remoteInstances = remoteInstances;
}
}
}

7
oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/ClusterModuleStandaloneProvider.java
View File

@ -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

14
oap-server/server-cluster-plugin/cluster-standalone-plugin/src/main/java/org/apache/skywalking/oap/server/cluster/plugin/standalone/StandaloneManager.java
View File

@ -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() {
}
}

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

@ -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

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

@ -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<RemoteInstance> serviceDiscovery) throws Exception {
final ServiceDiscovery<RemoteInstance> serviceDiscovery) {
this.manager = manager;
this.config = config;
this.serviceDiscovery = serviceDiscovery;
this.serviceCache = serviceDiscovery.serviceCacheBuilder().name(REMOTE_NAME_PATH).build();
this.serviceCache.start();
}
@Override
public synchronized void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
public void registerRemote(RemoteInstance remoteInstance) throws ServiceRegisterException {
try {
initHealthChecker();
if (needUsingInternalAddr()) {
remoteInstance = new RemoteInstance(new Address(config.getInternalComHost(), config.getInternalComPort(), true));
}
this.selfAddress = remoteInstance.getAddress();
ServiceInstance<RemoteInstance> thisInstance = ServiceInstance.<RemoteInstance>builder().name(REMOTE_NAME_PATH)
.id(UUID.randomUUID()
.toString())
@ -81,8 +84,6 @@ public 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<RemoteInstance> queryRemoteNodes() {
List<RemoteInstance> remoteInstances = new ArrayList<>(20);
try {
initHealthChecker();
List<ServiceInstance<RemoteInstance>> serviceInstances = serviceCache.getInstances();
serviceInstances.forEach(serviceInstance -> {
RemoteInstance instance = serviceInstance.getPayload();
@ -115,6 +115,10 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes();
notifyWatchers(remoteInstances);
} catch (Throwable e) {
healthChecker.unHealth(e);
log.error("Failed to notify and update remote instances", e);
}
}
@Override
public void stateChanged(final CuratorFramework client, final ConnectionState newState) {
if (log.isDebugEnabled()) {
log.debug("Zookeeper ConnectionState changed, state: {}", newState.name());
}
}
}
}

122
oap-server/server-cluster-plugin/cluster-zookeeper-plugin/src/test/java/org/apache/skywalking/oap/server/cluster/plugin/zookeeper/ITClusterModuleZookeeperProviderFunctionalTest.java
View File

@ -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<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertTrue(queryAddress.isSelf());
@ -98,9 +107,16 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(provider, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcher, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(provider, 1).size());
assertEquals(1, remoteInstances.size());
Address queryAddress = remoteInstances.get(0).getAddress();
@ -114,15 +130,21 @@ public 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<RemoteInstance> remoteInstances = queryRemoteNodes(providerB, 1);
List<RemoteInstance> remoteInstances = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstances.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address queryAddress = remoteInstances.get(0).getAddress();
assertEquals(selfAddress, queryAddress);
assertFalse(queryAddress.isSelf());
@ -133,21 +155,29 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
}
@Test
@ -155,33 +185,42 @@ public 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<RemoteInstance> remoteInstancesOfA = queryRemoteNodes(providerA, 2);
List<RemoteInstance> remoteInstancesOfA = notifiedRemoteNodes(watcherA, 2);
validateServiceInstance(addressA, addressB, remoteInstancesOfA);
assertEquals(2, queryRemoteNodes(providerA, 2).size());
List<RemoteInstance> remoteInstancesOfB = queryRemoteNodes(providerB, 2);
List<RemoteInstance> remoteInstancesOfB = notifiedRemoteNodes(watcherB, 2);
validateServiceInstance(addressB, addressA, remoteInstancesOfB);
assertEquals(2, queryRemoteNodes(providerB, 2).size());
// unregister A
ClusterRegister register = getClusterRegister(providerA);
ServiceDiscovery<RemoteInstance> discoveryA = Whitebox.getInternalState(register, "serviceDiscovery");
ServiceDiscovery<RemoteInstance> discoveryA = Whitebox.getInternalState(providerA, "serviceDiscovery");
discoveryA.close();
// only B
remoteInstancesOfB = queryRemoteNodes(providerB, 1, 120);
remoteInstancesOfB = notifiedRemoteNodes(watcherB, 1);
assertEquals(1, remoteInstancesOfB.size());
assertEquals(1, queryRemoteNodes(providerB, 1).size());
Address address = remoteInstancesOfB.get(0).getAddress();
assertEquals(address, addressB);
assertTrue(addressB.isSelf());
assertEquals(addressB, address);
assertTrue(address.isSelf());
}
private ClusterModuleZookeeperProvider createProvider(String namespace) throws Exception {
@ -221,6 +260,10 @@ 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<RemoteInstance> queryRemoteNodes(ModuleProvider provider, int goals,
int cyclic) throws InterruptedException {
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = getClusterNodesQuery(provider).queryRemoteNodes();
if (instances.size() == goals) {
@ -243,6 +286,25 @@ public class ITClusterModuleZookeeperProviderFunctionalTest {
return Collections.emptyList();
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals)
throws InterruptedException {
return notifiedRemoteNodes(watcher, goals, 20);
}
private List<RemoteInstance> notifiedRemoteNodes(ClusterMockWatcher watcher, int goals,
int cyclic) throws InterruptedException {
do {
List<RemoteInstance> instances = watcher.getRemoteInstances();
if (instances.size() == goals) {
return instances;
} else {
Thread.sleep(1000);
}
}
while (--cyclic > 0);
return Collections.emptyList();
}
private void validateServiceInstance(Address selfAddress, Address otherAddress, List<RemoteInstance> queryResult) {
assertEquals(2, queryResult.size());
@ -260,4 +322,14 @@ public class ITClusterModuleZookeeperProviderFunctionalTest {
assertTrue(selfExist);
assertTrue(otherExist);
}
class ClusterMockWatcher implements ClusterWatcher {
@Getter
private List<RemoteInstance> remoteInstances = new ArrayList<>();
@Override
public void onClusterNodesChanged(final List<RemoteInstance> remoteInstances) {
this.remoteInstances = remoteInstances;
}
}
}

17
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/CoreModuleProvider.java
View File

@ -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()) {

47
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterCoordinator.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.cluster;
import java.util.ArrayList;
import java.util.List;
import lombok.extern.slf4j.Slf4j;
import org.apache.skywalking.oap.server.library.module.ModuleStartException;
@Slf4j
public abstract class ClusterCoordinator implements ClusterRegister, ClusterNodesQuery, ClusterWatcherRegister {
private final List<ClusterWatcher> clusterWatchers = new ArrayList<>();
/**
* Initialize the required resources, such as healthy checker and listener.
*/
public abstract void start() throws ModuleStartException;
@Override
public void registerWatcher(final ClusterWatcher watcher) {
this.clusterWatchers.add(watcher);
}
protected void notifyWatchers(List<RemoteInstance> remoteInstances) {
if (log.isDebugEnabled()) {
log.debug("Notify watchers and update cluster instances:{}", remoteInstances.toString());
}
this.clusterWatchers.forEach(
clusterWatcher -> clusterWatcher.onClusterNodesChanged(remoteInstances));
}
}

3
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterModule.java
View File

@ -32,7 +32,8 @@ public class ClusterModule extends ModuleDefine {
public Class[] services() {
return new Class[] {
ClusterRegister.class,
ClusterNodesQuery.class
ClusterNodesQuery.class,
ClusterCoordinator.class
};
}
}

25
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcher.java Normal file
View File

@ -0,0 +1,25 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.oap.server.core.cluster;
import java.util.List;
public interface ClusterWatcher {
void onClusterNodesChanged(List<RemoteInstance> remoteInstances);
}

25
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/cluster/ClusterWatcherRegister.java Normal file
View File

@ -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);
}

71
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/remote/client/RemoteClientManager.java
View File

@ -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<RemoteInstance> 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<RemoteInstance> instanceList = clusterNodesQuery.queryRemoteNodes();
instanceList = distinct(instanceList);
Collections.sort(instanceList);
gauge.setValue(instanceList.size());
if (LOGGER.isDebugEnabled()) {
instanceList.forEach(instance -> LOGGER.debug("Cluster instance: {}", instance.toString()));
if (log.isDebugEnabled()) {
instanceList.forEach(instance -> log.debug("Cluster instance: {}", instance.toString()));
}
if (!compare(instanceList)) {
if (LOGGER.isDebugEnabled()) {
LOGGER.debug("ReBuilding remote clients.");
if (log.isDebugEnabled()) {
log.debug("ReBuilding remote clients.");
}
reBuildRemoteClients(instanceList);
moduleDefineHolder.find(CoreModule.NAME)
.provider()
.getService(ServerStatusService.class)
.reBalancedCluster(System.currentTimeMillis());
}
printRemoteClientList();
} catch (Throwable t) {
LOGGER.error(t.getMessage(), t);
log.error(t.getMessage(), t);
}
}
@ -152,10 +160,10 @@ 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<RemoteInstance> remoteInstances) {
refresh(remoteInstances);
}
enum Action {
Close, Unchanged, Create
}

32
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ClusterStatus.java Normal file
View File

@ -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;
}

10
oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/status/ServerStatusService.java
View File

@ -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);
}
}