From 809668ebc13361d78d369fe3f0f53aba6a1ee16f Mon Sep 17 00:00:00 2001 From: peng-yongsheng <8082209@qq.com> Date: Sat, 28 Oct 2017 15:25:01 +0800 Subject: [PATCH] Add zookeeper implement of cluster module. --- .../collector/cluster/ClusterException.java | 35 ++++ .../cluster/ClusterModuleListener.java | 52 ++++++ .../cluster/ClusterNodeExistException.java | 33 ++++ .../apm/collector/cluster/DataMonitor.java | 39 ++++ .../collector/cluster/ModuleRegistration.java | 55 ++++++ .../service/ModuleListenerService.java | 29 +++ .../service/ModuleRegisterService.java | 2 + .../service/RedisModuleRegisterService.java | 4 + .../StandaloneModuleRegisterService.java | 4 + .../ClusterModuleZookeeperProvider.java | 29 ++- .../zookeeper/ClusterZKDataMonitor.java | 166 ++++++++++++++++++ .../ZookeeperModuleListenerService.java | 39 ++++ .../ZookeeperModuleRegisterService.java | 13 ++ apm-collector/apm-collector-cluster/pom.xml | 5 + 14 files changed, 502 insertions(+), 3 deletions(-) create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterException.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleListener.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterNodeExistException.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/DataMonitor.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ModuleRegistration.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleListenerService.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKDataMonitor.java create mode 100644 apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleListenerService.java diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterException.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterException.java new file mode 100644 index 000000000..cee9eba36 --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterException.java @@ -0,0 +1,35 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster; + +import org.skywalking.apm.collector.core.CollectorException; + +/** + * @author peng-yongsheng + */ +public abstract class ClusterException extends CollectorException { + + public ClusterException(String message) { + super(message); + } + + public ClusterException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleListener.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleListener.java new file mode 100644 index 000000000..543245d8e --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleListener.java @@ -0,0 +1,52 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster; + +import java.util.HashSet; +import java.util.Set; + +/** + * @author peng-yongsheng + */ +public abstract class ClusterModuleListener { + + private Set addresses; + + public ClusterModuleListener() { + addresses = new HashSet<>(); + } + + public abstract String path(); + + public final void addAddress(String address) { + addresses.add(address); + } + + public final void removeAddress(String address) { + addresses.remove(address); + } + + public final Set getAddresses() { + return addresses; + } + + public abstract void serverJoinNotify(String serverAddress); + + public abstract void serverQuitNotify(String serverAddress); +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterNodeExistException.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterNodeExistException.java new file mode 100644 index 000000000..506b0fbe3 --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ClusterNodeExistException.java @@ -0,0 +1,33 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster; + +/** + * @author peng-yongsheng + */ +public class ClusterNodeExistException extends ClusterException { + + public ClusterNodeExistException(String message) { + super(message); + } + + public ClusterNodeExistException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/DataMonitor.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/DataMonitor.java new file mode 100644 index 000000000..bb961ad24 --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/DataMonitor.java @@ -0,0 +1,39 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster; + +import org.skywalking.apm.collector.client.Client; +import org.skywalking.apm.collector.client.ClientException; + +/** + * @author peng-yongsheng + */ +public interface DataMonitor { + void setClient(Client client); + + void addListener(ClusterModuleListener listener) throws ClientException; + + void register(String path, ModuleRegistration registration) throws ClientException; + + ClusterModuleListener getListener(String path); + + void createPath(String path) throws ClientException; + + void setData(String path, String value) throws ClientException; +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ModuleRegistration.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ModuleRegistration.java new file mode 100644 index 000000000..2aed56281 --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/ModuleRegistration.java @@ -0,0 +1,55 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster; + +/** + * @author peng-yongsheng + */ +public abstract class ModuleRegistration { + + public abstract Value buildValue(); + + public static class Value { + private final String host; + private final int port; + private final String contextPath; + + public Value(String host, int port, String contextPath) { + this.host = host; + this.port = port; + this.contextPath = contextPath; + } + + public String getHost() { + return host; + } + + public int getPort() { + return port; + } + + public String getHostPort() { + return host + ":" + port; + } + + public String getContextPath() { + return contextPath; + } + } +} \ No newline at end of file diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleListenerService.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleListenerService.java new file mode 100644 index 000000000..f5a908aaf --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleListenerService.java @@ -0,0 +1,29 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster.service; + +import org.skywalking.apm.collector.cluster.ClusterModuleListener; +import org.skywalking.apm.collector.core.module.Service; + +/** + * @author peng-yongsheng + */ +public interface ModuleListenerService extends Service { + void addListener(ClusterModuleListener listener); +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleRegisterService.java b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleRegisterService.java index e03c01144..4566204e0 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleRegisterService.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-define/src/main/java/org/skywalking/apm/collector/cluster/service/ModuleRegisterService.java @@ -18,10 +18,12 @@ package org.skywalking.apm.collector.cluster.service; +import org.skywalking.apm.collector.cluster.ModuleRegistration; import org.skywalking.apm.collector.core.module.Service; /** * @author peng-yongsheng */ public interface ModuleRegisterService extends Service { + void register(String moduleName, String providerName, ModuleRegistration registration); } diff --git a/apm-collector/apm-collector-cluster/collector-cluster-redis-provider/src/main/java/org/skywalking/apm/collector/cluster/redis/service/RedisModuleRegisterService.java b/apm-collector/apm-collector-cluster/collector-cluster-redis-provider/src/main/java/org/skywalking/apm/collector/cluster/redis/service/RedisModuleRegisterService.java index 80a0ffb12..c067d8d3d 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-redis-provider/src/main/java/org/skywalking/apm/collector/cluster/redis/service/RedisModuleRegisterService.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-redis-provider/src/main/java/org/skywalking/apm/collector/cluster/redis/service/RedisModuleRegisterService.java @@ -24,4 +24,8 @@ import org.skywalking.apm.collector.cluster.service.ModuleRegisterService; * @author peng-yongsheng */ public class RedisModuleRegisterService implements ModuleRegisterService { + + @Override public void register(String moduleName, String providerName, String address, String others) { + + } } diff --git a/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/service/StandaloneModuleRegisterService.java b/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/service/StandaloneModuleRegisterService.java index 2a43d850f..0538d51c7 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/service/StandaloneModuleRegisterService.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-standalone-provider/src/main/java/org/skywalking/apm/collector/cluster/standalone/service/StandaloneModuleRegisterService.java @@ -24,4 +24,8 @@ import org.skywalking.apm.collector.cluster.service.ModuleRegisterService; * @author peng-yongsheng */ public class StandaloneModuleRegisterService implements ModuleRegisterService { + + @Override public void register(String moduleName, String providerName, String address, String others) { + + } } diff --git a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterModuleZookeeperProvider.java b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterModuleZookeeperProvider.java index 62638a592..f8411bb9c 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterModuleZookeeperProvider.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterModuleZookeeperProvider.java @@ -19,10 +19,16 @@ package org.skywalking.apm.collector.cluster.zookeeper; import java.util.Properties; +import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient; import org.skywalking.apm.collector.cluster.ClusterModule; +import org.skywalking.apm.collector.cluster.service.ModuleListenerService; import org.skywalking.apm.collector.cluster.service.ModuleRegisterService; +import org.skywalking.apm.collector.cluster.service.ModuleRegistrationGetService; +import org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleListenerService; import org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleRegisterService; import org.skywalking.apm.collector.cluster.zookeeper.service.ZookeeperModuleRegistrationGetService; +import org.skywalking.apm.collector.core.CollectorException; +import org.skywalking.apm.collector.core.UnexpectedException; import org.skywalking.apm.collector.core.module.Module; import org.skywalking.apm.collector.core.module.ModuleProvider; import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; @@ -32,6 +38,11 @@ import org.skywalking.apm.collector.core.module.ServiceNotProvidedException; */ public class ClusterModuleZookeeperProvider extends ModuleProvider { + private static final String HOST_PORT = "hostPort"; + private static final String SESSION_TIMEOUT = "sessionTimeout"; + + private ClusterZKDataMonitor dataMonitor; + @Override public String name() { return "zookeeper"; } @@ -41,12 +52,24 @@ public class ClusterModuleZookeeperProvider extends ModuleProvider { } @Override public void prepare(Properties config) throws ServiceNotProvidedException { - this.registerServiceImplementation(ModuleRegisterService.class, new ZookeeperModuleRegisterService()); - this.registerServiceImplementation(ModuleRegisterService.class, new ZookeeperModuleRegistrationGetService()); + dataMonitor = new ClusterZKDataMonitor(); + + final String hostPort = config.getProperty(HOST_PORT); + final String sessionTimeout = config.getProperty(SESSION_TIMEOUT); + ZookeeperClient zookeeperClient = new ZookeeperClient(hostPort, Integer.valueOf(sessionTimeout), dataMonitor); + dataMonitor.setClient(zookeeperClient); + + this.registerServiceImplementation(ModuleListenerService.class, new ZookeeperModuleListenerService(dataMonitor)); + this.registerServiceImplementation(ModuleRegisterService.class, new ZookeeperModuleRegisterService(dataMonitor)); + this.registerServiceImplementation(ModuleRegistrationGetService.class, new ZookeeperModuleRegistrationGetService()); } @Override public void start(Properties config) throws ServiceNotProvidedException { - + try { + dataMonitor.start(); + } catch (CollectorException e) { + throw new UnexpectedException(e.getMessage()); + } } @Override public void notifyAfterCompleted() throws ServiceNotProvidedException { diff --git a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKDataMonitor.java b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKDataMonitor.java new file mode 100644 index 000000000..6ad398ebe --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKDataMonitor.java @@ -0,0 +1,166 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster.zookeeper; + +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.zookeeper.CreateMode; +import org.apache.zookeeper.WatchedEvent; +import org.apache.zookeeper.Watcher; +import org.apache.zookeeper.ZooDefs; +import org.apache.zookeeper.data.Stat; +import org.skywalking.apm.collector.client.Client; +import org.skywalking.apm.collector.client.ClientException; +import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient; +import org.skywalking.apm.collector.client.zookeeper.ZookeeperClientException; +import org.skywalking.apm.collector.client.zookeeper.util.PathUtils; +import org.skywalking.apm.collector.cluster.ClusterModuleListener; +import org.skywalking.apm.collector.cluster.ClusterNodeExistException; +import org.skywalking.apm.collector.cluster.DataMonitor; +import org.skywalking.apm.collector.cluster.ModuleRegistration; +import org.skywalking.apm.collector.core.CollectorException; +import org.skywalking.apm.collector.core.util.CollectionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author peng-yongsheng + */ +public class ClusterZKDataMonitor implements DataMonitor, Watcher { + + private final Logger logger = LoggerFactory.getLogger(ClusterZKDataMonitor.class); + + private ZookeeperClient client; + + private Map listeners; + private Map registrations; + + public ClusterZKDataMonitor() { + listeners = new LinkedHashMap<>(); + registrations = new LinkedHashMap<>(); + } + + @Override public synchronized void process(WatchedEvent event) { + logger.info("changed path {}, event type: {}", event.getPath(), event.getType().name()); + if (listeners.containsKey(event.getPath())) { + List paths; + try { + paths = client.getChildren(event.getPath(), true); + ClusterModuleListener listener = listeners.get(event.getPath()); + Set remoteNodes = new HashSet<>(); + Set notifiedNodes = listener.getAddresses(); + if (CollectionUtils.isNotEmpty(paths)) { + for (String serverPath : paths) { + Stat stat = new Stat(); + byte[] data = client.getData(event.getPath() + "/" + serverPath, true, stat); + String dataStr = new String(data); + String addressValue = serverPath + dataStr; + remoteNodes.add(addressValue); + if (!notifiedNodes.contains(addressValue)) { + logger.info("path children has been created, path: {}, data: {}", event.getPath() + "/" + serverPath, dataStr); + listener.addAddress(addressValue); + listener.serverJoinNotify(addressValue); + } + } + } + + String[] notifiedNodeArray = notifiedNodes.toArray(new String[notifiedNodes.size()]); + for (int i = notifiedNodeArray.length - 1; i >= 0; i--) { + String address = notifiedNodeArray[i]; + if (remoteNodes.isEmpty() || !remoteNodes.contains(address)) { + logger.info("path children has been remove, path and data: {}", event.getPath() + "/" + address); + listener.removeAddress(address); + listener.serverQuitNotify(address); + } + } + } catch (ZookeeperClientException e) { + logger.error(e.getMessage(), e); + } + } + } + + @Override public void setClient(Client client) { + this.client = (ZookeeperClient)client; + } + + public void start() throws CollectorException { + Iterator> entryIterator = registrations.entrySet().iterator(); + while (entryIterator.hasNext()) { + Map.Entry next = entryIterator.next(); + createPath(next.getKey()); + + ModuleRegistration.Value value = next.getValue().buildValue(); + String contextPath = value.getContextPath() == null ? "" : value.getContextPath(); + + client.getChildren(next.getKey(), true); + String serverPath = next.getKey() + "/" + value.getHostPort(); + + Stat stat = client.exists(serverPath, false); + if (stat != null) { + client.delete(serverPath, stat.getVersion()); + } + stat = client.exists(serverPath, false); + if (stat == null) { + setData(serverPath, contextPath); + } else { + client.delete(serverPath, stat.getVersion()); + throw new ClusterNodeExistException("current address: " + value.getHostPort() + " has been registered, check the host and port configuration or wait a moment."); + } + } + } + + @Override public void addListener(ClusterModuleListener listener) { + String path = PathUtils.convertKey2Path(listener.path()); + logger.info("listener path: {}", path); + listeners.put(path, listener); + } + + @Override public void register(String path, ModuleRegistration registration) { + registrations.put(path, registration); + } + + @Override public ClusterModuleListener getListener(String path) { + path = PathUtils.convertKey2Path(path); + return listeners.get(path); + } + + @Override public void createPath(String path) throws ClientException { + String[] paths = path.replaceFirst("/", "").split("/"); + + StringBuilder pathBuilder = new StringBuilder(); + for (String subPath : paths) { + pathBuilder.append("/").append(subPath); + if (client.exists(pathBuilder.toString(), false) == null) { + client.create(pathBuilder.toString(), null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT); + } + } + } + + @Override public void setData(String path, String value) throws ClientException { + if (client.exists(path, false) == null) { + client.create(path, value.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); + } else { + client.setData(path, value.getBytes(), -1); + } + } +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleListenerService.java b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleListenerService.java new file mode 100644 index 000000000..8d745474f --- /dev/null +++ b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleListenerService.java @@ -0,0 +1,39 @@ +/* + * Copyright 2017, OpenSkywalking Organization All rights reserved. + * + * Licensed 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. + * + * Project repository: https://github.com/OpenSkywalking/skywalking + */ + +package org.skywalking.apm.collector.cluster.zookeeper.service; + +import org.skywalking.apm.collector.cluster.ClusterModuleListener; +import org.skywalking.apm.collector.cluster.service.ModuleListenerService; +import org.skywalking.apm.collector.cluster.zookeeper.ClusterZKDataMonitor; + +/** + * @author peng-yongsheng + */ +public class ZookeeperModuleListenerService implements ModuleListenerService { + + private final ClusterZKDataMonitor dataMonitor; + + public ZookeeperModuleListenerService(ClusterZKDataMonitor dataMonitor) { + this.dataMonitor = dataMonitor; + } + + @Override public void addListener(ClusterModuleListener listener) { + dataMonitor.addListener(listener); + } +} diff --git a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleRegisterService.java b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleRegisterService.java index 2ef55e46e..7b2c08cf0 100644 --- a/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleRegisterService.java +++ b/apm-collector/apm-collector-cluster/collector-cluster-zookeeper-provider/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/service/ZookeeperModuleRegisterService.java @@ -18,10 +18,23 @@ package org.skywalking.apm.collector.cluster.zookeeper.service; +import org.skywalking.apm.collector.cluster.ModuleRegistration; import org.skywalking.apm.collector.cluster.service.ModuleRegisterService; +import org.skywalking.apm.collector.cluster.zookeeper.ClusterZKDataMonitor; /** * @author peng-yongsheng */ public class ZookeeperModuleRegisterService implements ModuleRegisterService { + + private final ClusterZKDataMonitor dataMonitor; + + public ZookeeperModuleRegisterService(ClusterZKDataMonitor dataMonitor) { + this.dataMonitor = dataMonitor; + } + + @Override public void register(String moduleName, String providerName, ModuleRegistration registration) { + String path = "/" + moduleName + "/" + providerName; + dataMonitor.register(path, registration); + } } diff --git a/apm-collector/apm-collector-cluster/pom.xml b/apm-collector/apm-collector-cluster/pom.xml index 24c360ace..f0b2e2935 100644 --- a/apm-collector/apm-collector-cluster/pom.xml +++ b/apm-collector/apm-collector-cluster/pom.xml @@ -24,5 +24,10 @@ apm-collector-core ${project.version} + + org.skywalking + client-component + ${project.version} + \ No newline at end of file