diff --git a/apm-dist/release-docs/LICENSE b/apm-dist/release-docs/LICENSE index bddc04a56..95e0c0889 100644 --- a/apm-dist/release-docs/LICENSE +++ b/apm-dist/release-docs/LICENSE @@ -325,6 +325,7 @@ The text of each license is the standard Apache 2.0 license. Ctripcorp: apollo 1.4.0: https://github.com/ctripcorp/apollo Apache 2.0 etcd4j 2.17.0: https://github.com/jurmous/etcd4j Apache 2.0 javaassist 3.25.0-GA: https://github.com/jboss-javassist/javassist Apache 2.0 + jackson-module-afterburner 2.9.5: https://github.com/FasterXML/jackson-modules-base, Apache 2.0 ======================================================================== MIT licenses diff --git a/oap-server/pom.xml b/oap-server/pom.xml index b2e218696..8cd33bc4f 100644 --- a/oap-server/pom.xml +++ b/oap-server/pom.xml @@ -82,10 +82,14 @@ 2.12.0 2.17.0 v3.2.3 + 4.1.27.Final + 2.9.5 4.7.1 2.3.28 3.25.0-GA + 3.5 + @@ -378,11 +382,75 @@ nacos-client ${nacos.version} + org.mousio etcd4j + + + netty-codec-dns + io.netty + + + + netty-codec-dns + io.netty + + + + netty-codec-http + io.netty + + + + netty-handler + io.netty + + + + netty-resolver-dns + io.netty + + + + com.fasterxml.jackson.module + jackson-module-afterburner + + ${etcd4j.version} + + + io.netty + netty-handler + ${netty.version} + + + + io.netty + netty-resolver-dns + ${netty.version} + + + + io.netty + netty-codec-dns + ${netty.version} + + + + io.netty + netty-codec-http + ${netty.version} + + + + com.fasterxml.jackson.module + jackson-module-afterburner + ${jackson-module-afterburner.version} + + + org.apache.curator curator-x-discovery diff --git a/oap-server/server-cluster-plugin/cluster-etcd-plugin/pom.xml b/oap-server/server-cluster-plugin/cluster-etcd-plugin/pom.xml index d2315eccb..7673108f0 100644 --- a/oap-server/server-cluster-plugin/cluster-etcd-plugin/pom.xml +++ b/oap-server/server-cluster-plugin/cluster-etcd-plugin/pom.xml @@ -36,60 +36,34 @@ ${project.version} + + io.netty + netty-codec-dns + + + + io.netty + netty-codec-http + + io.netty netty-handler - 4.1.27.Final io.netty netty-resolver-dns - 4.1.27.Final - org.mousio etcd4j - - - netty-codec-dns - io.netty - - - - netty-codec-dns - io.netty - - - - netty-codec-http - io.netty - - - - netty-handler - io.netty - - - - netty-resolver-dns - io.netty - - - - com.fasterxml.jackson.module - jackson-module-afterburner - - - com.fasterxml.jackson.module jackson-module-afterburner - 2.9.5 diff --git a/oap-server/server-configuration/configuration-etcd/pom.xml b/oap-server/server-configuration/configuration-etcd/pom.xml new file mode 100644 index 000000000..4ee2145b3 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/pom.xml @@ -0,0 +1,177 @@ + + + + + server-configuration + org.apache.skywalking + 6.3.0-SNAPSHOT + + 4.0.0 + + configuration-etcd + + + + org.apache.skywalking + configuration-api + ${project.version} + + + + io.netty + netty-codec-dns + + + + io.netty + netty-codec-http + + + + io.netty + netty-handler + + + + io.netty + netty-resolver-dns + + + + org.mousio + etcd4j + + + + com.fasterxml.jackson.module + jackson-module-afterburner + + + + org.yaml + snakeyaml + + + + + + CI-with-IT + + + + io.fabric8 + docker-maven-plugin + + all + default + true + IfNotPresent + + + + start + pre-integration-test + + start + + + + + quayio/coreos-etcd:${etcd.version} + etcd-client-integration-test + + + etcd.port:2379 + + + + + + + + /usr/local/bin/etcd + --advertise-client-urls=http://0.0.0.0:2379 + --listen-client-urls=http://0.0.0.0:2379 + + + + + + + + + remove-it-etcd + post-integration-test + + stop + + + + + + + org.codehaus.gmaven + gmaven-plugin + 1.5 + + + add-default-properties + initialize + + execute + + + 2.0 + + project.properties.setProperty('etcd.host', 'localhost') + + log.info("Etcd host is " + project.properties['etcd.host']) + + + + + + + + org.apache.maven.plugins + maven-failsafe-plugin + + + + ${etcd.host} + + + ${etcd.port} + + + + + + + integration-test + verify + + + + + + + + + diff --git a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigException.java b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigException.java new file mode 100644 index 000000000..e2ef9c895 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigException.java @@ -0,0 +1,33 @@ +/* + * 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.configuration.etcd; + +/** + * exception type throw by Etcd Configuration. + * + * @author Alan Lau + */ +public class EtcdConfigException extends RuntimeException { + + + public EtcdConfigException(String message, Throwable cause) { + super(message, cause); + } + +} 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 new file mode 100644 index 000000000..17899ea1f --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegister.java @@ -0,0 +1,162 @@ +/* + * 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.configuration.etcd; + +import java.net.URI; +import java.util.HashSet; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import mousio.client.promises.ResponsePromise; +import mousio.etcd4j.EtcdClient; +import mousio.etcd4j.promises.EtcdResponsePromise; +import mousio.etcd4j.responses.EtcdErrorCode; +import mousio.etcd4j.responses.EtcdException; +import mousio.etcd4j.responses.EtcdKeysResponse; +import org.apache.skywalking.oap.server.configuration.api.ConfigTable; +import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author Alan Lau + */ +public class EtcdConfigWatcherRegister extends ConfigWatcherRegister { + + private final static Logger logger = LoggerFactory.getLogger(EtcdConfigWatcherRegister.class); + + /** + * server settings for Etcd configuration + */ + private EtcdServerSettings settings; + + /** + * etcd client. + */ + private final EtcdClient client; + + private final Map listenersByKey; + + private final Map> configItemKeyedByName; + + private final Map> responsePromiseByKey; + + public EtcdConfigWatcherRegister(EtcdServerSettings settings) { + super(settings.getPeriod()); + this.settings = settings; + this.configItemKeyedByName = new ConcurrentHashMap<>(); + this.client = new EtcdClient(EtcdUtils.parse(settings).toArray(new URI[] {})); + this.listenersByKey = new ConcurrentHashMap<>(); + responsePromiseByKey = new ConcurrentHashMap<>(); + } + + @Override public ConfigTable readConfig(Set keys) { + removeUninterestedKeys(keys); + registerKeyListeners(keys); + final ConfigTable table = new ConfigTable(); + + for (Map.Entry> entry : configItemKeyedByName.entrySet()) { + final String key = entry.getKey(); + final Optional value = entry.getValue(); + + if (value.isPresent()) { + table.add(new ConfigTable.ConfigItem(key, value.get())); + } else { + table.add(new ConfigTable.ConfigItem(key, null)); + } + } + + return table; + } + + private void registerKeyListeners(final Set keys) { + for (final String key : keys) { + String dataId = "/" + settings.getGroup() + "/" + key; + if (listenersByKey.containsKey(dataId)) { + continue; + } + + listenersByKey.putIfAbsent(dataId, p -> { + onDataValueChanged(p, dataId); + }); + + try { + EtcdResponsePromise responsePromise = client.get(dataId).waitForChange().send(); + responsePromise.addListener(listenersByKey.get(dataId)); + responsePromiseByKey.putIfAbsent(dataId, responsePromise); + + // the key is newly added, read the config for the first time + EtcdResponsePromise promise = client.get(dataId).send(); + onDataValueChanged(promise, dataId); + } catch (Exception e) { + throw new EtcdConfigException("wait for etcd value change fail", e); + } + } + } + + private void removeUninterestedKeys(final Set interestedKeys) { + final Set uninterestedKeys = new HashSet<>(listenersByKey.keySet()); + uninterestedKeys.removeAll(interestedKeys); + + uninterestedKeys.forEach(k -> { + final ResponsePromise.IsSimplePromiseResponseHandler listener = listenersByKey.remove(k); + if (listener != null) { + responsePromiseByKey.remove(k).removeListener(listener); + } + }); + } + + private void onDataValueChanged(ResponsePromise promise, String dataId) { + String key = getRealKey(dataId, settings.getGroup()); + try { + EtcdKeysResponse.EtcdNode node = promise.get().getNode(); + String value = node.getValue(); + if (logger.isInfoEnabled()) { + logger.info("Etcd config changed: {}: {}", key, node.getValue()); + } + + configItemKeyedByName.put(key, Optional.ofNullable(value)); + } catch (Exception e) { + if (e instanceof EtcdException) { + if (EtcdErrorCode.KeyNotFound == ((EtcdException)e).errorCode) { + configItemKeyedByName.put(key, Optional.empty()); + return; + } + } + throw new EtcdConfigException("wait for value changed fail", e); + } + } + + /** + * get real key in etcd cluster which is removed "/${group}" from the key retrive from etcd. + * + * @param key + * @param group + * @return + */ + private String getRealKey(String key, String group) { + int index = key.indexOf(group); + if (index <= 0) { + throw new RuntimeException("the group doesn't match"); + } + String realKey = key.substring(index + group.length() + 1); + return realKey; + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationProvider.java b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationProvider.java new file mode 100644 index 000000000..f63314c32 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationProvider.java @@ -0,0 +1,67 @@ +/* + * 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.configuration.etcd; + +import com.google.common.base.Strings; +import org.apache.skywalking.oap.server.configuration.api.AbstractConfigurationProvider; +import org.apache.skywalking.oap.server.configuration.api.ConfigWatcherRegister; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; +import org.apache.skywalking.oap.server.library.module.ModuleStartException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Get Configuration from etcd. + * + * @author Alan Lau + */ +public class EtcdConfigurationProvider extends AbstractConfigurationProvider { + + private final static Logger logger = LoggerFactory.getLogger(EtcdConfigurationProvider.class); + + private EtcdServerSettings settings; + + public EtcdConfigurationProvider() { + settings = new EtcdServerSettings(); + } + + @Override protected ConfigWatcherRegister initConfigReader() throws ModuleStartException { + logger.info("settings: {}", settings); + if (Strings.isNullOrEmpty(settings.getServerAddr())) { + throw new ModuleStartException("Etcd serverAddr cannot be null or empty."); + } + if (Strings.isNullOrEmpty(settings.getGroup())) { + throw new ModuleStartException("Etcd group cannot be null or empty."); + } + + try { + return new EtcdConfigWatcherRegister(settings); + } catch (Exception e) { + throw new ModuleStartException(e.getMessage(), e); + } + } + + @Override public String name() { + return "etcd"; + } + + @Override public ModuleConfig createConfigBeanIfAbsent() { + return settings; + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdServerSettings.java b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdServerSettings.java new file mode 100644 index 000000000..689382444 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdServerSettings.java @@ -0,0 +1,52 @@ +/* + * 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.configuration.etcd; + +import lombok.Getter; +import lombok.Setter; +import lombok.ToString; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; + +/** + * entity wrapps the etcd cluster configuration. + * + * @author Alan Lau + */ +@ToString +@Getter +@Setter +public class EtcdServerSettings extends ModuleConfig { + + private String clusterName = "default"; + /** + * etcd cluster address, like "10.10.10.1:2379, 10.10.10.2:2379,10.10.10.3.2379". + */ + private String serverAddr; + + /** + * directory for configuration + */ + private String group; + + /** + * sec for interval refresh config data. + */ + private int period = 60; + +} diff --git a/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdUtils.java b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdUtils.java new file mode 100644 index 000000000..71834d7aa --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdUtils.java @@ -0,0 +1,72 @@ +/* + * 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.configuration.etcd; + +import java.net.URI; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.skywalking.oap.server.library.util.Address; +import org.apache.skywalking.oap.server.library.util.ConnectUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * a util for etcd serverAddr parse. + * + * @author Alan Lau + */ +public class EtcdUtils { + + private final static Logger logger = LoggerFactory.getLogger(EtcdUtils.class); + + public EtcdUtils() { + } + + public static List parse(EtcdServerSettings settings) { + List uris = new ArrayList<>(); + try { + logger.info("etcd settings is {}", settings); + List
addressList = ConnectUtils.parse(settings.getServerAddr()); + for (Address address : addressList) { + uris.add(new URI("http", null, address.getHost(), address.getPort(), null, null, null)); + } + } catch (Exception e) { + throw new EtcdConfigException(e.getMessage(), e); + } + + return uris; + } + + public static List parseProp(Properties properties) { + List uris = new ArrayList<>(); + try { + logger.info("etcd server addr is {}", properties); + List
addressList = ConnectUtils.parse(properties.getProperty("serverAddr")); + for (Address address : addressList) { + uris.add(new URI("http", null, address.getHost(), address.getPort(), null, null, null)); + } + } catch (Exception e) { + throw new EtcdConfigException(e.getMessage(), e); + } + + return uris; + } + +} diff --git a/oap-server/server-configuration/configuration-etcd/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-configuration/configuration-etcd/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider new file mode 100644 index 000000000..a23863923 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/main/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -0,0 +1,19 @@ +# +# 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. +# +# + +org.apache.skywalking.oap.server.configuration.etcd.EtcdConfigurationProvider \ No newline at end of file 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 new file mode 100644 index 000000000..2d4fcf6f9 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigWatcherRegisterTest.java @@ -0,0 +1,132 @@ +/* + * 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.configuration.etcd; + +import com.google.common.collect.Sets; +import java.net.URI; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import mousio.client.promises.ResponsePromise; +import mousio.etcd4j.EtcdClient; +import mousio.etcd4j.promises.EtcdResponsePromise; +import mousio.etcd4j.requests.EtcdKeyGetRequest; +import mousio.etcd4j.responses.EtcdKeysResponse; +import org.apache.skywalking.oap.server.configuration.api.ConfigTable; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.powermock.api.mockito.PowerMockito; +import org.powermock.core.classloader.annotations.PowerMockIgnore; +import org.powermock.core.classloader.annotations.PrepareForTest; +import org.powermock.modules.junit4.PowerMockRunner; +import org.powermock.reflect.Whitebox; + +import static junit.framework.TestCase.assertEquals; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.spy; +import static org.powermock.api.mockito.PowerMockito.mock; +import static org.powermock.api.mockito.PowerMockito.mockStatic; +import static org.powermock.api.mockito.PowerMockito.when; +import static org.powermock.api.mockito.PowerMockito.whenNew; + +/** + * @author Alan Lau + */ +@RunWith(PowerMockRunner.class) +@PrepareForTest({EtcdKeysResponse.class, EtcdUtils.class, EtcdClient.class, URI.class}) +@PowerMockIgnore({"javax.management.*"}) +public class EtcdConfigWatcherRegisterTest { + + @Before + @Test + public void shouldReadConfigs() throws Exception { + final String group = "skywalking"; + final String testKey1 = "receiver-trace.default.slowDBAccessThreshold"; + final String testVal1 = "test"; + final String testKey2 = "testKey"; + final String testVal2 = "testVal"; + + final EtcdServerSettings mockSettings = mock(EtcdServerSettings.class); + when(mockSettings.getGroup()).thenReturn(group); + mockStatic(EtcdUtils.class); + + List uris = mock(List.class); + when(EtcdUtils.parse(any())).thenReturn(uris); + + final EtcdClient client = PowerMockito.mock(EtcdClient.class); + whenNew(EtcdClient.class).withAnyArguments().thenReturn(client); + + String port = System.getProperty("etcd.port"); + URI uri = new URI("http://localhost:" + port); + List urisArray = spy(ArrayList.class); + urisArray.add(uri); + URI[] array = urisArray.toArray(new URI[] {}); + when(uris.toArray(new URI[] {})).thenReturn(array); + + final EtcdConfigWatcherRegister mockRegister = spy(new EtcdConfigWatcherRegister(mockSettings)); + + Whitebox.setInternalState(mockRegister, "client", client); + Whitebox.setInternalState(mockRegister, "settings", mockSettings); + + final EtcdKeysResponse response = PowerMockito.mock(EtcdKeysResponse.class); + final EtcdKeysResponse response1 = PowerMockito.mock(EtcdKeysResponse.class); + + final EtcdKeyGetRequest request = PowerMockito.mock(EtcdKeyGetRequest.class); + + when(client.get("/skywalking/receiver-trace.default.slowDBAccessThreshold")).thenReturn(request); + when(request.waitForChange()).thenReturn(request); + + final EtcdResponsePromise promise = mock(EtcdResponsePromise.class); + final ResponsePromise responseResponsePromise = mock(ResponsePromise.class); + when(request.send()).thenReturn(promise); + when(promise.get()).thenReturn(response); + when(responseResponsePromise.get()).thenReturn(response); + + final EtcdKeysResponse.EtcdNode node = mock(EtcdKeysResponse.EtcdNode.class); + when(response.getNode()).thenReturn(node); + when(node.getKey()).thenReturn("/skywalking/receiver-trace.default.slowDBAccessThreshold"); + when(node.getValue()).thenReturn("test"); + + final EtcdKeyGetRequest request1 = mock(EtcdKeyGetRequest.class); + when(client.get("/skywalking/testKey")).thenReturn(request1); + when(request1.waitForChange()).thenReturn(request1); + final EtcdResponsePromise promise1 = mock(EtcdResponsePromise.class); + final ResponsePromise responseResponsePromise1 = mock(ResponsePromise.class); + when(request1.send()).thenReturn(promise1); + when(promise1.get()).thenReturn(response1); + when(responseResponsePromise1.get()).thenReturn(response1); + + final EtcdKeysResponse.EtcdNode node1 = mock(EtcdKeysResponse.EtcdNode.class); + when(response1.getNode()).thenReturn(node1); + when(node1.getKey()).thenReturn("/skywalking/testKey"); + when(node1.getValue()).thenReturn("testVal"); + + final ConfigTable configTable = mockRegister.readConfig(Sets.newHashSet(testKey1, testKey2)); + + assertEquals(2, configTable.getItems().size()); + Map kvs = new HashMap<>(); + for (ConfigTable.ConfigItem item : configTable.getItems()) { + kvs.put(item.getName(), item.getValue()); + } + assertEquals(testVal1, kvs.get(testKey1)); + assertEquals(testVal2, kvs.get(testKey2)); + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestModule.java b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestModule.java new file mode 100644 index 000000000..c03fae40c --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestModule.java @@ -0,0 +1,37 @@ +/* + * 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.configuration.etcd; + +import org.apache.skywalking.oap.server.library.module.ModuleDefine; + +/** + * @author Alan Lau + */ +public class EtcdConfigurationTestModule extends ModuleDefine { + + public static final String NAME = "test-module"; + + public EtcdConfigurationTestModule() { + super(NAME); + } + + @Override public Class[] services() { + return new Class[0]; + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestProvider.java b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestProvider.java new file mode 100644 index 000000000..93d54969c --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/EtcdConfigurationTestProvider.java @@ -0,0 +1,89 @@ +/* + * 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.configuration.etcd; + +import org.apache.skywalking.oap.server.configuration.api.ConfigChangeWatcher; +import org.apache.skywalking.oap.server.configuration.api.ConfigurationModule; +import org.apache.skywalking.oap.server.configuration.api.DynamicConfigurationService; +import org.apache.skywalking.oap.server.library.module.ModuleConfig; +import org.apache.skywalking.oap.server.library.module.ModuleDefine; +import org.apache.skywalking.oap.server.library.module.ModuleProvider; +import org.apache.skywalking.oap.server.library.module.ModuleStartException; +import org.apache.skywalking.oap.server.library.module.ServiceNotProvidedException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author Alan Lau + */ +public class EtcdConfigurationTestProvider extends ModuleProvider { + + private final static Logger LOGGER = LoggerFactory.getLogger(EtcdConfigurationTestProvider.class); + + ConfigChangeWatcher watcher; + + @Override public String name() { + return "default"; + } + + @Override public Class module() { + return EtcdConfigurationTestModule.class; + } + + @Override public ModuleConfig createConfigBeanIfAbsent() { + return new ModuleConfig() { + }; + } + + @Override public void prepare() throws ServiceNotProvidedException, ModuleStartException { + watcher = new ConfigChangeWatcher(EtcdConfigurationTestModule.NAME, this, "testKey") { + private volatile String testValue; + + @Override + public void notify(ConfigChangeWatcher.ConfigChangeEvent value) { + LOGGER.info("ConfigChangeWatcher.ConfigChangeEvent: {}", value); + if (EventType.DELETE.equals(value.getEventType())) { + testValue = null; + } else { + testValue = value.getNewValue(); + } + } + + @Override + public String value() { + return testValue; + } + }; + } + + @Override public void start() throws ServiceNotProvidedException, ModuleStartException { + getManager().find(ConfigurationModule.NAME) + .provider() + .getService(DynamicConfigurationService.class) + .registerConfigChangeWatcher(watcher); + } + + @Override public void notifyAfterCompleted() throws ServiceNotProvidedException, ModuleStartException { + + } + + @Override public String[] requiredModules() { + return new String[0]; + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/ITEtcdConfigurationTest.java b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/ITEtcdConfigurationTest.java new file mode 100644 index 000000000..c157652bc --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/ITEtcdConfigurationTest.java @@ -0,0 +1,153 @@ +/* + * 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.configuration.etcd; + +import java.io.FileNotFoundException; +import java.io.Reader; +import java.net.URI; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import mousio.etcd4j.EtcdClient; +import mousio.etcd4j.promises.EtcdResponsePromise; +import mousio.etcd4j.responses.EtcdKeysResponse; +import org.apache.skywalking.apm.util.PropertyPlaceholderHelper; +import org.apache.skywalking.oap.server.library.module.ApplicationConfiguration; +import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.util.CollectionUtils; +import org.apache.skywalking.oap.server.library.util.ResourceUtils; +import org.junit.Before; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.yaml.snakeyaml.Yaml; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * @author Alan Lau + */ +public class ITEtcdConfigurationTest { + + private static final Logger logger = LoggerFactory.getLogger(ITEtcdConfigurationTest.class); + + private final Yaml yaml = new Yaml(); + + private EtcdServerSettings settings; + + private EtcdConfigurationTestProvider provider; + + private EtcdClient client; + + @Before + public void setUp() throws Exception { + final ApplicationConfiguration applicationConfiguration = new ApplicationConfiguration(); + loadConfig(applicationConfiguration); + + final ModuleManager moduleManager = new ModuleManager(); + moduleManager.init(applicationConfiguration); + + final String etcdHost = System.getProperty("etcd.host"); + final String etcdPort = System.getProperty("etcd.port"); + logger.info("etcdHost: {}, etcdPort: {}", etcdHost, etcdPort); + Properties properties = new Properties(); + properties.setProperty("serverAddr", etcdHost + ":" + etcdPort); + + List uris = EtcdUtils.parseProp(properties); + client = new EtcdClient(uris.toArray(new URI[] {})); + + provider = + (EtcdConfigurationTestProvider)moduleManager + .find(EtcdConfigurationTestModule.NAME) + .provider(); + + assertNotNull(provider); + } + + @Test(timeout = 20000) + public void shouldReadUpdated() throws Exception { + assertNull(provider.watcher.value()); + + assertTrue(publishConfig("test-module.default.testKey", "skywalking", "500")); + + for (String v = provider.watcher.value(); v == null; v = provider.watcher.value()) { + logger.info("value is : {}", provider.watcher.value()); + } + + assertEquals("500", provider.watcher.value()); + + assertTrue(removeConfig("test-module.default.testKey", "skywalking")); + + for (String v = provider.watcher.value(); v != null; v = provider.watcher.value()) { + } + + assertNull(provider.watcher.value()); + } + + @SuppressWarnings("unchecked") + private void loadConfig(ApplicationConfiguration configuration) throws FileNotFoundException { + Reader applicationReader = ResourceUtils.read("application.yml"); + Map>> moduleConfig = yaml.loadAs(applicationReader, Map.class); + if (CollectionUtils.isNotEmpty(moduleConfig)) { + moduleConfig.forEach((moduleName, providerConfig) -> { + if (providerConfig.size() > 0) { + ApplicationConfiguration.ModuleConfiguration moduleConfiguration = configuration.addModule(moduleName); + providerConfig.forEach((name, propertiesConfig) -> { + Properties properties = new Properties(); + if (propertiesConfig != null) { + propertiesConfig.forEach((key, value) -> { + properties.put(key, value); + final Object replaceValue = yaml.load(PropertyPlaceholderHelper.INSTANCE + .replacePlaceholders(value + "", properties)); + if (replaceValue != null) { + properties.replace(key, replaceValue); + } + }); + } + moduleConfiguration.addProviderConfiguration(name, properties); + }); + } + }); + } + } + + private boolean publishConfig(String key, String group, String value) { + try { + client.putDir(group).send().get(); + EtcdResponsePromise promise = client.put(generateKey(key, group), value).send(); + promise.get(); + return true; + } catch (Exception e) { + return false; + } + } + + private boolean removeConfig(String key, String group) throws Exception { + client.delete(generateKey(key, group)).send().get(); + return true; + } + + private String generateKey(String key, String group) { + return new StringBuilder("/").append(group).append("/").append(key).toString(); + } + +} diff --git a/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/TestEtcdUtils.java b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/TestEtcdUtils.java new file mode 100644 index 000000000..0977fabd9 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/java/org/apache/skywalking/oap/server/configuration/etcd/TestEtcdUtils.java @@ -0,0 +1,64 @@ +/* + * 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.configuration.etcd; + +import java.net.URI; +import java.util.List; +import java.util.Properties; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +/** + * @author Alan Lau + */ +public class TestEtcdUtils { + + private EtcdServerSettings settings; + + private Properties properties; + + @Before + public void setUp() { + settings = new EtcdServerSettings(); + settings.setServerAddr("localhost:2379"); + properties = new Properties(); + properties.setProperty("serverAddr", "localhost:2379"); + } + + @Test + public void testParse() { + List list = EtcdUtils.parse(settings); + Assert.assertEquals(1, list.size()); + URI uri = list.get(0); + Assert.assertEquals("http", uri.getScheme()); + Assert.assertEquals("localhost", uri.getHost()); + Assert.assertEquals(2379, uri.getPort()); + } + + @Test + public void testProp() { + List list = EtcdUtils.parseProp(properties); + Assert.assertEquals(1, list.size()); + URI uri = list.get(0); + Assert.assertEquals("http", uri.getScheme()); + Assert.assertEquals("localhost", uri.getHost()); + Assert.assertEquals(2379, uri.getPort()); + } +} diff --git a/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine b/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine new file mode 100644 index 000000000..cbbb641b9 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleDefine @@ -0,0 +1,20 @@ +# +# 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. +# +# + +org.apache.skywalking.oap.server.configuration.api.ConfigurationModule +org.apache.skywalking.oap.server.configuration.etcd.EtcdConfigurationTestModule diff --git a/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider b/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider new file mode 100644 index 000000000..0afae5f59 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/resources/META-INF/services/org.apache.skywalking.oap.server.library.module.ModuleProvider @@ -0,0 +1,19 @@ +# +# 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. +# +# + +org.apache.skywalking.oap.server.configuration.etcd.EtcdConfigurationTestProvider diff --git a/oap-server/server-configuration/configuration-etcd/src/test/resources/application.yml b/oap-server/server-configuration/configuration-etcd/src/test/resources/application.yml new file mode 100644 index 000000000..be3009c89 --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/resources/application.yml @@ -0,0 +1,33 @@ +# 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. + +test-module: + default: + testKey: 300 + + +configuration: + etcd: + # Etcd Server Host + serverAddr: ${etcd.host}:${etcd.port} + # Etcd Server Port + port: ${etcd.port} + # Etcd Configuration Group + group: 'skywalking' + # Unit seconds, sync period. Default fetch every 60 seconds. + period: 1 + # the name of current cluster, set the name if you want to upstream system known. + clusterName: "default" diff --git a/oap-server/server-configuration/configuration-etcd/src/test/resources/log4j2.xml b/oap-server/server-configuration/configuration-etcd/src/test/resources/log4j2.xml new file mode 100644 index 000000000..c9eec4f6e --- /dev/null +++ b/oap-server/server-configuration/configuration-etcd/src/test/resources/log4j2.xml @@ -0,0 +1,31 @@ + + + + + + + + + + + + + + + diff --git a/oap-server/server-configuration/pom.xml b/oap-server/server-configuration/pom.xml index 5d89c4e6e..be0196f2e 100644 --- a/oap-server/server-configuration/pom.xml +++ b/oap-server/server-configuration/pom.xml @@ -33,6 +33,7 @@ configuration-apollo configuration-nacos configuration-zookeeper + configuration-etcd diff --git a/oap-server/server-starter/pom.xml b/oap-server/server-starter/pom.xml index cd9b471f7..5eb911e0e 100644 --- a/oap-server/server-starter/pom.xml +++ b/oap-server/server-starter/pom.xml @@ -203,6 +203,11 @@ configuration-zookeeper ${project.version} + + org.apache.skywalking + configuration-etcd + ${project.version} + skywalking-oap diff --git a/oap-server/server-starter/src/main/assembly/application.yml b/oap-server/server-starter/src/main/assembly/application.yml index e45666ccf..50d7ecf05 100644 --- a/oap-server/server-starter/src/main/assembly/application.yml +++ b/oap-server/server-starter/src/main/assembly/application.yml @@ -149,6 +149,11 @@ configuration: # #Retry Policy # baseSleepTimeMs: 1000 # initial amount of time to wait between retries # maxRetries: 3 # max number of times to retry +# etcd: +# period : 60 # Unit seconds, sync period. Default fetch every 60 seconds. +# group : 'skywalking' +# serverAddr: localhost:2379 +# clusterName: "default" #exporter: # grpc: # targetHost: ${SW_EXPORTER_GRPC_HOST:127.0.0.1} diff --git a/oap-server/server-starter/src/main/resources/application.yml b/oap-server/server-starter/src/main/resources/application.yml index 85a5b8bc3..bb4fc85cb 100644 --- a/oap-server/server-starter/src/main/resources/application.yml +++ b/oap-server/server-starter/src/main/resources/application.yml @@ -168,6 +168,11 @@ configuration: # #Retry Policy # baseSleepTimeMs: 1000 # initial amount of time to wait between retries # maxRetries: 3 # max number of times to retry +# etcd: +# period : 60 # Unit seconds, sync period. Default fetch every 60 seconds. +# group : 'skywalking' +# serverAddr: localhost:2379 +# clusterName: "default" #exporter: # grpc: