From be048d42362ccfc079a551e81bfca00ab658aa82 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=B4=E6=99=9F=20Wu=20Sheng?= Date: Wed, 4 Mar 2020 10:17:04 +0800 Subject: [PATCH] Support change detection mechanism in DCS (#4438) * Add UUID to DCS --- .../api/ConfigWatcherRegister.java | 60 +++++++++++-------- .../api/ConfigWatcherRegisterTest.java | 5 +- .../apollo/ApolloConfigWatcherRegister.java | 5 +- .../ConsulConfigurationWatcherRegister.java | 4 +- .../etcd/EtcdConfigWatcherRegister.java | 4 +- .../etcd/EtcdConfigWatcherRegisterTest.java | 2 +- .../nacos/NacosConfigWatcherRegister.java | 4 +- .../nacos/NacosConfigWatcherRegisterTest.java | 2 +- .../ZookeeperConfigWatcherRegister.java | 5 +- .../MockZookeeperConfigWatcherRegister.java | 5 +- ...ookeeperConfigWatcherRegisterTestCase.java | 2 +- .../grpc/GRPCConfigWatcherRegister.java | 28 ++++++--- .../main/proto/configuration-service.proto | 5 ++ .../analysis/ApdexThresholdConfigTest.java | 5 +- 14 files changed, 85 insertions(+), 51 deletions(-) diff --git a/oap-server/server-configuration/configuration-api/src/main/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegister.java b/oap-server/server-configuration/configuration-api/src/main/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegister.java index 84609702d..c4a436de3 100644 --- a/oap-server/server-configuration/configuration-api/src/main/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-api/src/main/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegister.java @@ -20,6 +20,7 @@ package org.apache.skywalking.oap.server.configuration.api; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; @@ -67,41 +68,52 @@ public abstract class ConfigWatcherRegister implements DynamicConfigurationServi logger.info("Current configurations after the bootstrap sync." + LINE_SEPARATOR + register.toString()); Executors.newSingleThreadScheduledExecutor() - .scheduleAtFixedRate(new RunnableWithExceptionProtection(this::configSync, t -> logger.error("Sync config center error.", t)), syncPeriod, syncPeriod, TimeUnit.SECONDS); + .scheduleAtFixedRate( + new RunnableWithExceptionProtection( + this::configSync, + t -> logger.error("Sync config center error.", t) + ), syncPeriod, syncPeriod, TimeUnit.SECONDS); } void configSync() { - ConfigTable configTable = readConfig(register.keys()); + Optional configTable = readConfig(register.keys()); - configTable.getItems().forEach(item -> { - String itemName = item.getName(); - WatcherHolder holder = register.get(itemName); - if (holder != null) { - ConfigChangeWatcher watcher = holder.getWatcher(); - String newItemValue = item.getValue(); - if (newItemValue == null) { - if (watcher.value() != null) { - // Notify watcher, the new value is null with delete event type. - watcher.notify(new ConfigChangeWatcher.ConfigChangeEvent(null, ConfigChangeWatcher.EventType.DELETE)); + // Config table would be null if no change detected from the implementation. + configTable.ifPresent(config -> { + config.getItems().forEach(item -> { + String itemName = item.getName(); + WatcherHolder holder = register.get(itemName); + if (holder != null) { + ConfigChangeWatcher watcher = holder.getWatcher(); + String newItemValue = item.getValue(); + if (newItemValue == null) { + if (watcher.value() != null) { + // Notify watcher, the new value is null with delete event type. + watcher.notify( + new ConfigChangeWatcher.ConfigChangeEvent(null, ConfigChangeWatcher.EventType.DELETE)); + } else { + // Don't need to notify, stay in null. + } } else { - // Don't need to notify, stay in null. + if (!newItemValue.equals(watcher.value())) { + watcher.notify(new ConfigChangeWatcher.ConfigChangeEvent( + newItemValue, + ConfigChangeWatcher.EventType.MODIFY + )); + } else { + // Don't need to notify, stay in the same config value. + } } } else { - if (!newItemValue.equals(watcher.value())) { - watcher.notify(new ConfigChangeWatcher.ConfigChangeEvent(newItemValue, ConfigChangeWatcher.EventType.MODIFY)); - } else { - // Don't need to notify, stay in the same config value. - } + logger.warn("Config {} from configuration center, doesn't match any watcher, ignore.", itemName); } - } else { - logger.warn("Config {} from configuration center, doesn't match any watcher, ignore.", itemName); - } - }); + }); - logger.trace("Current configurations after the sync." + LINE_SEPARATOR + register.toString()); + logger.trace("Current configurations after the sync." + LINE_SEPARATOR + register.toString()); + }); } - public abstract ConfigTable readConfig(Set keys); + public abstract Optional readConfig(Set keys); public class Register { private Map register = new HashMap<>(); diff --git a/oap-server/server-configuration/configuration-api/src/test/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegisterTest.java b/oap-server/server-configuration/configuration-api/src/test/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegisterTest.java index 702015145..944ab3c2c 100644 --- a/oap-server/server-configuration/configuration-api/src/test/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegisterTest.java +++ b/oap-server/server-configuration/configuration-api/src/test/java/org/apache/skywalking/oap/server/configuration/api/ConfigWatcherRegisterTest.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.configuration.api; +import java.util.Optional; import java.util.Set; import org.apache.skywalking.oap.server.library.module.ModuleConfig; import org.apache.skywalking.oap.server.library.module.ModuleDefine; @@ -88,14 +89,14 @@ public class ConfigWatcherRegisterTest { public static class MockConfigWatcherRegister extends ConfigWatcherRegister { @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { ConfigTable.ConfigItem item1 = new ConfigTable.ConfigItem("module.provider.prop1", "abc"); ConfigTable.ConfigItem item2 = new ConfigTable.ConfigItem("MockModule.provider.prop2", "abc2"); ConfigTable table = new ConfigTable(); table.add(item1); table.add(item2); - return table; + return Optional.of(table); } } diff --git a/oap-server/server-configuration/configuration-apollo/src/main/java/org/apache/skywalking/oap/server/configuration/apollo/ApolloConfigWatcherRegister.java b/oap-server/server-configuration/configuration-apollo/src/main/java/org/apache/skywalking/oap/server/configuration/apollo/ApolloConfigWatcherRegister.java index 6600a6969..60d9319c3 100644 --- a/oap-server/server-configuration/configuration-apollo/src/main/java/org/apache/skywalking/oap/server/configuration/apollo/ApolloConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-apollo/src/main/java/org/apache/skywalking/oap/server/configuration/apollo/ApolloConfigWatcherRegister.java @@ -21,6 +21,7 @@ package org.apache.skywalking.oap.server.configuration.apollo; import com.ctrip.framework.apollo.Config; import com.ctrip.framework.apollo.ConfigService; import com.google.common.base.Strings; +import java.util.Optional; import java.util.Set; import org.apache.skywalking.oap.server.configuration.api.ConfigTable; import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister; @@ -53,7 +54,7 @@ public class ApolloConfigWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { final ConfigTable configTable = new ConfigTable(); for (final String name : keys) { @@ -61,6 +62,6 @@ public class ApolloConfigWatcherRegister extends ConfigWatcherRegister { configTable.add(new ConfigTable.ConfigItem(name, value)); } - return configTable; + return Optional.of(configTable); } } diff --git a/oap-server/server-configuration/configuration-consul/src/main/java/org/apache/skywalking/oap/server/configuration/consul/ConsulConfigurationWatcherRegister.java b/oap-server/server-configuration/configuration-consul/src/main/java/org/apache/skywalking/oap/server/configuration/consul/ConsulConfigurationWatcherRegister.java index 5df153178..73c3d5592 100644 --- a/oap-server/server-configuration/configuration-consul/src/main/java/org/apache/skywalking/oap/server/configuration/consul/ConsulConfigurationWatcherRegister.java +++ b/oap-server/server-configuration/configuration-consul/src/main/java/org/apache/skywalking/oap/server/configuration/consul/ConsulConfigurationWatcherRegister.java @@ -76,7 +76,7 @@ public class ConsulConfigurationWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { removeUninterestedKeys(keys); registerKeyListeners(keys); @@ -91,7 +91,7 @@ public class ConsulConfigurationWatcherRegister extends ConfigWatcherRegister { } }); - return table; + return Optional.of(table); } private void registerKeyListeners(final Set keys) { diff --git a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegister.java b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegister.java index 1d701581f..aedb37504 100644 --- a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegister.java @@ -65,7 +65,7 @@ public class EtcdConfigWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { removeUninterestedKeys(keys); registerKeyListeners(keys); final ConfigTable table = new ConfigTable(); @@ -81,7 +81,7 @@ public class EtcdConfigWatcherRegister extends ConfigWatcherRegister { } } - return table; + return Optional.of(table); } private void registerKeyListeners(final Set keys) { diff --git a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegisterTest.java b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegisterTest.java index 09ead678f..4f1f47698 100644 --- a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegisterTest.java +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegisterTest.java @@ -121,7 +121,7 @@ public class EtcdConfigWatcherRegisterTest { when(node1.getKey()).thenReturn("/skywalking/testKey"); when(node1.getValue()).thenReturn("testVal"); - final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(testKey1, testKey2)); + final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(testKey1, testKey2)).get(); assertEquals(2, configTable.getItems().size()); Map kvs = new HashMap<>(); diff --git a/oap-server/server-configuration/configuration-nacos/src/main/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegister.java b/oap-server/server-configuration/configuration-nacos/src/main/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegister.java index 3a3f2cb01..0101cdf51 100644 --- a/oap-server/server-configuration/configuration-nacos/src/main/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-nacos/src/main/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegister.java @@ -60,7 +60,7 @@ public class NacosConfigWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { removeUninterestedKeys(keys); registerKeyListeners(keys); @@ -77,7 +77,7 @@ public class NacosConfigWatcherRegister extends ConfigWatcherRegister { } } - return table; + return Optional.of(table); } private void registerKeyListeners(final Set keys) { diff --git a/oap-server/server-configuration/configuration-nacos/src/test/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegisterTest.java b/oap-server/server-configuration/configuration-nacos/src/test/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegisterTest.java index f4ae9687c..671ae2a52 100644 --- a/oap-server/server-configuration/configuration-nacos/src/test/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegisterTest.java +++ b/oap-server/server-configuration/configuration-nacos/src/test/java/org/apache/skywalking/oap/server/configuration/nacos/NacosConfigWatcherRegisterTest.java @@ -52,7 +52,7 @@ public class NacosConfigWatcherRegisterTest { Whitebox.setInternalState(mockRegister, "configService", mockConfigService); - final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(testKey1, testKey2)); + final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(testKey1, testKey2)).get(); assertEquals(2, configTable.getItems().size()); Map kvs = new HashMap<>(); diff --git a/oap-server/server-configuration/configuration-zookeeper/src/main/java/org/apache/skywalking/oap/server/configuration/zookeeper/ZookeeperConfigWatcherRegister.java b/oap-server/server-configuration/configuration-zookeeper/src/main/java/org/apache/skywalking/oap/server/configuration/zookeeper/ZookeeperConfigWatcherRegister.java index b0ca07919..b8f0d1cb2 100644 --- a/oap-server/server-configuration/configuration-zookeeper/src/main/java/org/apache/skywalking/oap/server/configuration/zookeeper/ZookeeperConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-zookeeper/src/main/java/org/apache/skywalking/oap/server/configuration/zookeeper/ZookeeperConfigWatcherRegister.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.configuration.zookeeper; +import java.util.Optional; import java.util.Set; import org.apache.curator.RetryPolicy; import org.apache.curator.framework.CuratorFramework; @@ -43,12 +44,12 @@ public class ZookeeperConfigWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { ConfigTable table = new ConfigTable(); keys.forEach(s -> { ChildData data = this.childrenCache.getCurrentData(this.prefix + s); table.add(new ConfigTable.ConfigItem(s, data == null ? null : new String(data.getData()))); }); - return table; + return Optional.of(table); } } diff --git a/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/MockZookeeperConfigWatcherRegister.java b/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/MockZookeeperConfigWatcherRegister.java index 4b424af10..3cf419787 100644 --- a/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/MockZookeeperConfigWatcherRegister.java +++ b/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/MockZookeeperConfigWatcherRegister.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.configuration.zookeeper.ut; +import java.util.Optional; import java.util.Set; import org.apache.curator.framework.recipes.cache.ChildData; import org.apache.curator.framework.recipes.cache.PathChildrenCache; @@ -35,12 +36,12 @@ public class MockZookeeperConfigWatcherRegister extends ConfigWatcherRegister { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { ConfigTable table = new ConfigTable(); keys.forEach(s -> { ChildData data = this.childrenCache.getCurrentData(this.prefix + s); table.add(new ConfigTable.ConfigItem(s, data == null ? null : new String(data.getData()))); }); - return table; + return Optional.of(table); } } \ No newline at end of file diff --git a/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/ZookeeperConfigWatcherRegisterTestCase.java b/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/ZookeeperConfigWatcherRegisterTestCase.java index 1d8cf2b40..e4e08fb7d 100644 --- a/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/ZookeeperConfigWatcherRegisterTestCase.java +++ b/oap-server/server-configuration/configuration-zookeeper/src/test/java/org/apache/skywalking/oap/server/configuration/zookeeper/ut/ZookeeperConfigWatcherRegisterTestCase.java @@ -48,7 +48,7 @@ public class ZookeeperConfigWatcherRegisterTestCase { Whitebox.setInternalState(mockRegister, "childrenCache", mockPathChildrenCache); - final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(key)); + final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(key)).get(); assertEquals(1, configTable.getItems().size()); assertEquals(key, configTable.getItems().get(0).getName()); diff --git a/oap-server/server-configuration/grpc-configuration-sync/src/main/java/org/apache/skywalking/oap/server/configuration/grpc/GRPCConfigWatcherRegister.java b/oap-server/server-configuration/grpc-configuration-sync/src/main/java/org/apache/skywalking/oap/server/configuration/grpc/GRPCConfigWatcherRegister.java index dd92b2429..1332e0b7a 100644 --- a/oap-server/server-configuration/grpc-configuration-sync/src/main/java/org/apache/skywalking/oap/server/configuration/grpc/GRPCConfigWatcherRegister.java +++ b/oap-server/server-configuration/grpc-configuration-sync/src/main/java/org/apache/skywalking/oap/server/configuration/grpc/GRPCConfigWatcherRegister.java @@ -19,6 +19,8 @@ package org.apache.skywalking.oap.server.configuration.grpc; import io.grpc.netty.NettyChannelBuilder; +import java.util.Objects; +import java.util.Optional; import java.util.Set; import org.apache.skywalking.oap.server.configuration.api.ConfigTable; import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister; @@ -33,22 +35,32 @@ public class GRPCConfigWatcherRegister extends ConfigWatcherRegister { private RemoteEndpointSettings settings; private ConfigurationServiceGrpc.ConfigurationServiceBlockingStub stub; + private String uuid = null; public GRPCConfigWatcherRegister(RemoteEndpointSettings settings) { super(settings.getPeriod()); this.settings = settings; - stub = ConfigurationServiceGrpc.newBlockingStub(NettyChannelBuilder.forAddress(settings.getHost(), settings.getPort()) - .usePlaintext() - .build()); + stub = ConfigurationServiceGrpc.newBlockingStub( + NettyChannelBuilder.forAddress(settings.getHost(), settings.getPort()) + .usePlaintext() + .build()); } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { ConfigTable table = new ConfigTable(); try { - ConfigurationResponse response = stub.call(ConfigurationRequest.newBuilder() - .setClusterName(settings.getClusterName()) - .build()); + ConfigurationRequest.Builder builder = ConfigurationRequest.newBuilder() + .setClusterName(settings.getClusterName()); + if (uuid != null) { + builder.setUuid(uuid); + } + ConfigurationResponse response = stub.call(builder.build()); + String responseUuid = response.getUuid(); + if (Objects.equals(uuid, responseUuid)) { + // If UUID matched, the config table is expected as empty. + return Optional.empty(); + } response.getConfigTableList().forEach(config -> { final String name = config.getName(); if (keys.contains(name)) { @@ -58,6 +70,6 @@ public class GRPCConfigWatcherRegister extends ConfigWatcherRegister { } catch (Exception e) { logger.error("Remote config center [" + settings + "] is not available.", e); } - return table; + return Optional.of(table); } } diff --git a/oap-server/server-configuration/grpc-configuration-sync/src/main/proto/configuration-service.proto b/oap-server/server-configuration/grpc-configuration-sync/src/main/proto/configuration-service.proto index be15aef1d..8c0b4ef9d 100644 --- a/oap-server/server-configuration/grpc-configuration-sync/src/main/proto/configuration-service.proto +++ b/oap-server/server-configuration/grpc-configuration-sync/src/main/proto/configuration-service.proto @@ -31,6 +31,8 @@ message ConfigurationRequest { // in case the remote configuration center implementation support // configuration management for multiple clusters. string clusterName = 1; + // The config UUID response from the config server side. + string uuid = 2; } message ConfigurationResponse { @@ -42,6 +44,9 @@ message ConfigurationResponse { // If the config center wants to set the value to NULL or empty, // must set the name with empty value explicitly. repeated Config configTable = 1; + // UUID is literal string represents the content of the config table. + // If config table is unchanged, then could response the same uuid, and config table is not required. + string uuid = 2; } message Config { diff --git a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/ApdexThresholdConfigTest.java b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/ApdexThresholdConfigTest.java index 8489e8bb8..4ba736f32 100644 --- a/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/ApdexThresholdConfigTest.java +++ b/oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/ApdexThresholdConfigTest.java @@ -18,6 +18,7 @@ package org.apache.skywalking.oap.server.core.analysis; +import java.util.Optional; import java.util.Set; import org.apache.skywalking.oap.server.configuration.api.ConfigTable; import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister; @@ -68,10 +69,10 @@ public class ApdexThresholdConfigTest { } @Override - public ConfigTable readConfig(Set keys) { + public Optional readConfig(Set keys) { ConfigTable table = new ConfigTable(); table.add(new ConfigTable.ConfigItem("core.default.apdexThreshold", "default: 1000 \nfoo: 200")); - return table; + return Optional.of(table); } } } \ No newline at end of file