From 640440d5810a6e764da971a06bb2914b82266442 Mon Sep 17 00:00:00 2001 From: pengys5 <8082209@qq.com> Date: Tue, 18 Jul 2017 14:49:47 +0800 Subject: [PATCH] =?UTF-8?q?1.=20Add=20stream=20module:=20The=20stream=20ba?= =?UTF-8?q?se=20framework=202.=20Add=20context=20help=20for=20every=20modu?= =?UTF-8?q?le=20to=20find=20another=20module=E2=80=99s=20context.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../apm-collector-agentstream/pom.xml | 7 ++- .../agentstream/AgentStreamModuleDefine.java | 6 ++- .../AgentStreamModuleException.java | 2 +- .../AgentStreamModuleGroupDefine.java | 26 ++++++++++ .../AgentStreamModuleInstaller.java | 51 +++++++++++++++++++ .../grpc/AgentStreamGRPCConfig.java | 2 +- .../grpc/AgentStreamGRPCConfigParser.java | 2 +- .../grpc/AgentStreamGRPCModuleDefine.java | 10 ++-- .../AgentStreamGRPCModuleRegistration.java | 2 +- .../jetty/AgentStreamJettyConfig.java | 2 +- .../jetty/AgentStreamJettyConfigParser.java | 2 +- .../jetty/AgentStreamJettyModuleDefine.java | 10 ++-- .../AgentStreamJettyModuleRegistration.java | 2 +- .../resources/META-INF/defines/group.define | 1 + .../resources/META-INF/defines/module.define | 4 +- .../collector/boot/CollectorBootStartUp.java | 1 - .../apm/collector/boot}/CollectorStarter.java | 27 ++++------ apm-collector/apm-collector-client/pom.xml | 10 ++++ .../cluster/ClusterModuleGroupDefine.java | 17 +++++-- .../cluster/ClusterModuleInstaller.java | 11 +++- .../redis/ClusterRedisModuleDefine.java | 14 ++--- .../ClusterStandaloneModuleDefine.java | 14 ++--- .../zookeeper/ClusterZKModuleDefine.java | 14 ++--- apm-collector/apm-collector-core/pom.xml | 5 -- .../core/cluster/ClusterModuleContext.java | 29 +++++++++-- .../core/cluster/ClusterModuleDefine.java | 4 +- .../framework/CollectorContextHelper.java | 25 +++++++++ .../apm/collector/core/framework/Context.java | 11 +++- .../collector/core/framework/Executor.java | 8 +++ .../core/module/ModuleConfigLoader.java | 9 +++- .../collector/core/module/ModuleDefine.java | 2 +- .../core/module/ModuleDefineLoader.java | 2 +- .../collector/core/module/ModuleGroup.java | 8 --- .../core/module/ModuleGroupDefine.java | 6 ++- .../core/module/ModuleInstallMode.java | 8 --- .../core/module/ModuleInstallerAdapter.java | 25 --------- .../apm/collector/core/queue/Creator.java | 10 ---- .../collector/core/queue/QueueCreator.java | 8 +++ .../collector/core/queue/QueueExecutor.java | 9 ++++ .../core/queue/QueueModuleContext.java | 8 --- .../collector/core/util/ResourceUtils.java | 11 ++-- .../apm/collector/core/worker/Role.java | 13 ----- .../main/resources/application-default.yml} | 0 .../collector/queue/QueueModuleContext.java | 23 +++++++++ .../collector}/queue/QueueModuleDefine.java | 2 +- .../collector/queue/QueueModuleException.java | 16 ++++++ .../queue/QueueModuleGroupDefine.java | 25 +++++++++ .../collector/queue/QueueModuleInstaller.java | 23 +++++++++ .../queue/datacarrier/DataCarrierCreator.java | 15 ------ .../datacarrier/DataCarrierQueueCreator.java | 15 ++++++ .../QueueDataCarrierModuleDefine.java | 13 ++--- .../disruptor/DisruptorEventHandler.java | 23 ++++----- ...reator.java => DisruptorQueueCreator.java} | 10 ++-- .../disruptor/QueueDisruptorModuleDefine.java | 13 ++--- .../resources/META-INF/defines/group.define | 1 + .../collector/remote/RemoteModuleContext.java | 12 +++++ .../collector}/remote/RemoteModuleDefine.java | 2 +- .../remote/RemoteModuleGroupDefine.java | 25 +++++++++ .../remote/RemoteModuleInstaller.java | 17 +++++++ .../remote/grpc/RemoteGRPCModuleDefine.java | 8 +-- .../resources/META-INF/defines/group.define | 1 + .../resources/META-INF/defines/module.define | 1 + apm-collector/apm-collector-stream/pom.xml | 27 ++++++++++ .../stream}/AbstractClusterWorker.java | 2 +- .../AbstractClusterWorkerProvider.java | 2 +- .../stream}/AbstractLocalAsyncWorker.java | 2 +- .../AbstractLocalAsyncWorkerProvider.java | 10 ++-- .../stream}/AbstractLocalSyncWorker.java | 2 +- .../AbstractLocalSyncWorkerProvider.java | 2 +- .../stream}/AbstractLocalWorker.java | 2 +- .../stream}/AbstractLocalWorkerProvider.java | 2 +- .../apm/collector/stream}/AbstractWorker.java | 6 ++- .../stream}/AbstractWorkerProvider.java | 2 +- .../stream}/ClusterWorkerContext.java | 2 +- .../collector/stream}/ClusterWorkerRef.java | 2 +- .../stream}/ClusterWorkerRefCounter.java | 2 +- .../apm/collector/stream}/Context.java | 2 +- .../stream}/LocalAsyncWorkerRef.java | 2 +- .../collector/stream}/LocalSyncWorkerRef.java | 2 +- .../collector/stream}/LocalWorkerContext.java | 2 +- .../apm/collector/stream}/LookUp.java | 2 +- .../apm/collector/stream}/Provider.java | 2 +- .../stream}/ProviderNotFoundException.java | 2 +- .../skywalking/apm/collector/stream/Role.java | 13 +++++ .../stream}/UsedRoleNameException.java | 2 +- .../apm/collector/stream}/WorkerContext.java | 2 +- .../collector/stream}/WorkerException.java | 2 +- .../stream}/WorkerInvokeException.java | 2 +- .../stream}/WorkerModuleInstaller.java | 2 +- .../stream}/WorkerNotFoundException.java | 2 +- .../apm/collector/stream}/WorkerRef.java | 2 +- .../apm/collector/stream}/WorkerRefs.java | 4 +- .../stream}/selector/AbstractHashMessage.java | 2 +- .../stream}/selector/HashCodeSelector.java | 8 +-- .../stream}/selector/RollingSelector.java | 6 +-- .../stream}/selector/WorkerSelector.java | 6 +-- apm-collector/pom.xml | 13 ++++- 97 files changed, 578 insertions(+), 250 deletions(-) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core => apm-collector-agentstream/src/main/java/org/skywalking/apm/collector}/agentstream/AgentStreamModuleDefine.java (80%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core => apm-collector-agentstream/src/main/java/org/skywalking/apm/collector}/agentstream/AgentStreamModuleException.java (86%) create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleGroupDefine.java create mode 100644 apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleInstaller.java rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/grpc/AgentStreamGRPCConfig.java (66%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/grpc/AgentStreamGRPCConfigParser.java (93%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/grpc/AgentStreamGRPCModuleDefine.java (74%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/grpc/AgentStreamGRPCModuleRegistration.java (83%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/jetty/AgentStreamJettyConfig.java (72%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/jetty/AgentStreamJettyConfigParser.java (94%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/jetty/AgentStreamJettyModuleDefine.java (75%) rename apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/{agent/stream/server => agentstream}/jetty/AgentStreamJettyModuleRegistration.java (88%) create mode 100644 apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/group.define rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework => apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot}/CollectorStarter.java (57%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core => apm-collector-cluster/src/main/java/org/skywalking/apm/collector}/cluster/ClusterModuleInstaller.java (77%) create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorContextHelper.java create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Executor.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroup.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallMode.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallerAdapter.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/Creator.java create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueCreator.java create mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueExecutor.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleContext.java delete mode 100644 apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Role.java rename apm-collector/{apm-collector-boot/src/main/resources/application.yml => apm-collector-core/src/main/resources/application-default.yml} (100%) create mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleContext.java rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core => apm-collector-queue/src/main/java/org/skywalking/apm/collector}/queue/QueueModuleDefine.java (95%) create mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleException.java create mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleGroupDefine.java create mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleInstaller.java delete mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierCreator.java create mode 100644 apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierQueueCreator.java rename apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/{DisruptorCreator.java => DisruptorQueueCreator.java} (77%) create mode 100644 apm-collector/apm-collector-queue/src/main/resources/META-INF/defines/group.define create mode 100644 apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleContext.java rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core => apm-collector-remote/src/main/java/org/skywalking/apm/collector}/remote/RemoteModuleDefine.java (75%) create mode 100644 apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleGroupDefine.java create mode 100644 apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleInstaller.java create mode 100644 apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/group.define create mode 100644 apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/module.define create mode 100644 apm-collector/apm-collector-stream/pom.xml rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractClusterWorker.java (97%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractClusterWorkerProvider.java (96%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalAsyncWorker.java (97%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalAsyncWorkerProvider.java (55%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalSyncWorker.java (96%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalSyncWorkerProvider.java (92%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalWorker.java (84%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractLocalWorkerProvider.java (74%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractWorker.java (82%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/AbstractWorkerProvider.java (95%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/ClusterWorkerContext.java (96%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/ClusterWorkerRef.java (89%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/ClusterWorkerRefCounter.java (91%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/Context.java (85%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/LocalAsyncWorkerRef.java (90%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/LocalSyncWorkerRef.java (92%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/LocalWorkerContext.java (87%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/LookUp.java (79%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/Provider.java (73%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/ProviderNotFoundException.java (74%) create mode 100644 apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Role.java rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/UsedRoleNameException.java (73%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerContext.java (96%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerException.java (87%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerInvokeException.java (88%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerModuleInstaller.java (95%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerNotFoundException.java (74%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerRef.java (85%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/WorkerRefs.java (89%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/selector/AbstractHashMessage.java (89%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/selector/HashCodeSelector.java (83%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/selector/RollingSelector.java (82%) rename apm-collector/{apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker => apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream}/selector/WorkerSelector.java (75%) diff --git a/apm-collector/apm-collector-agentstream/pom.xml b/apm-collector/apm-collector-agentstream/pom.xml index 8ef444a8a..8e268c685 100644 --- a/apm-collector/apm-collector-agentstream/pom.xml +++ b/apm-collector/apm-collector-agentstream/pom.xml @@ -15,7 +15,12 @@ org.skywalking - apm-collector-core + apm-collector-stream + ${project.version} + + + org.skywalking + apm-collector-cluster ${project.version} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleDefine.java similarity index 80% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleDefine.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleDefine.java index 97e377ba6..239e57424 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleDefine.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleDefine.java @@ -1,11 +1,13 @@ -package org.skywalking.apm.collector.core.agentstream; +package org.skywalking.apm.collector.agentstream; import java.util.Map; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; import org.skywalking.apm.collector.core.client.Client; import org.skywalking.apm.collector.core.client.ClientException; import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer; import org.skywalking.apm.collector.core.cluster.ClusterModuleContext; import org.skywalking.apm.collector.core.config.ConfigParseException; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.core.framework.DataInitializer; import org.skywalking.apm.collector.core.framework.DefineException; import org.skywalking.apm.collector.core.module.ModuleDefine; @@ -24,7 +26,7 @@ public abstract class AgentStreamModuleDefine extends ModuleDefine { server.initialize(); String key = ClusterDataInitializer.BASE_CATALOG + "." + name(); - ClusterModuleContext.WRITER.write(key, registration().buildValue()); + ((ClusterModuleContext)CollectorContextHelper.INSTANCE.getContext(ClusterModuleGroupDefine.GROUP_NAME)).getWriter().write(key, registration().buildValue()); } catch (ConfigParseException | ServerException e) { throw new AgentStreamModuleException(e.getMessage(), e); } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleException.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleException.java similarity index 86% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleException.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleException.java index f46602bbd..e51b3f6f6 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/agentstream/AgentStreamModuleException.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.agentstream; +package org.skywalking.apm.collector.agentstream; import org.skywalking.apm.collector.core.module.ModuleException; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleGroupDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleGroupDefine.java new file mode 100644 index 000000000..ea730c5d6 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleGroupDefine.java @@ -0,0 +1,26 @@ +package org.skywalking.apm.collector.agentstream; + +import org.skywalking.apm.collector.core.cluster.ClusterModuleContext; +import org.skywalking.apm.collector.core.framework.Context; +import org.skywalking.apm.collector.core.module.ModuleGroupDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; + +/** + * @author pengys5 + */ +public class AgentStreamModuleGroupDefine implements ModuleGroupDefine { + + public static final String GROUP_NAME = "agent_stream"; + + @Override public String name() { + return GROUP_NAME; + } + + @Override public Context groupContext() { + return new ClusterModuleContext(GROUP_NAME); + } + + @Override public ModuleInstaller moduleInstaller() { + return new AgentStreamModuleInstaller(); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleInstaller.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleInstaller.java new file mode 100644 index 000000000..df6a5f168 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleInstaller.java @@ -0,0 +1,51 @@ +package org.skywalking.apm.collector.agentstream; + +import java.util.Iterator; +import java.util.Map; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; +import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.cluster.ClusterModuleContext; +import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; +import org.skywalking.apm.collector.core.framework.DefineException; +import org.skywalking.apm.collector.core.module.ModuleDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; +import org.skywalking.apm.collector.core.util.CollectionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class AgentStreamModuleInstaller implements ModuleInstaller { + + private final Logger logger = LoggerFactory.getLogger(AgentStreamModuleInstaller.class); + + @Override public void install(Map moduleConfig, + Map moduleDefineMap) throws DefineException, ClientException { + logger.info("beginning cluster module install"); + + ModuleDefine moduleDefine = null; + if (CollectionUtils.isEmpty(moduleConfig)) { + logger.info("could not configure cluster module, use the default"); + Iterator> moduleDefineEntry = moduleDefineMap.entrySet().iterator(); + while (moduleDefineEntry.hasNext()) { + moduleDefine = moduleDefineEntry.next().getValue(); + if (moduleDefine.defaultModule()) { + logger.info("module {} initialize", moduleDefine.getClass().getName()); + moduleDefine.initialize(null); + break; + } + } + } else { + Map.Entry clusterConfigEntry = moduleConfig.entrySet().iterator().next(); + moduleDefine = moduleDefineMap.get(clusterConfigEntry.getKey()); + moduleDefine.initialize(clusterConfigEntry.getValue()); + } + + ClusterModuleContext context = new ClusterModuleContext(ClusterModuleGroupDefine.GROUP_NAME); + context.setWriter(((ClusterModuleDefine)moduleDefine).registrationWriter()); + + CollectorContextHelper.INSTANCE.putContext(context); + } +} diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfig.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfig.java similarity index 66% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfig.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfig.java index 950775c50..d04feda9f 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfig.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfig.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.grpc; +package org.skywalking.apm.collector.agentstream.grpc; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfigParser.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfigParser.java similarity index 93% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfigParser.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfigParser.java index 05fd728a7..35b9e288f 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCConfigParser.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCConfigParser.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.grpc; +package org.skywalking.apm.collector.agentstream.grpc; import java.util.Map; import org.skywalking.apm.collector.core.config.ConfigParseException; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java similarity index 74% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleDefine.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java index 5efeb7a58..b41b062d5 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleDefine.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java @@ -1,8 +1,8 @@ -package org.skywalking.apm.collector.agent.stream.server.grpc; +package org.skywalking.apm.collector.agentstream.grpc; -import org.skywalking.apm.collector.core.agentstream.AgentStreamModuleDefine; +import org.skywalking.apm.collector.agentstream.AgentStreamModuleDefine; +import org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; import org.skywalking.apm.collector.core.module.ModuleRegistration; import org.skywalking.apm.collector.core.server.Server; import org.skywalking.apm.collector.server.grpc.GRPCServer; @@ -12,8 +12,8 @@ import org.skywalking.apm.collector.server.grpc.GRPCServer; */ public class AgentStreamGRPCModuleDefine extends AgentStreamModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.AgentStream; + @Override protected String group() { + return AgentStreamModuleGroupDefine.GROUP_NAME; } @Override public String name() { diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleRegistration.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleRegistration.java similarity index 83% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleRegistration.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleRegistration.java index 347b29b4e..800ebba24 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/grpc/AgentStreamGRPCModuleRegistration.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleRegistration.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.grpc; +package org.skywalking.apm.collector.agentstream.grpc; import org.skywalking.apm.collector.core.module.ModuleRegistration; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfig.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfig.java similarity index 72% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfig.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfig.java index 240d5ef1e..5ed3e24c5 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfig.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfig.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.jetty; +package org.skywalking.apm.collector.agentstream.jetty; /** * @author pengys5 diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfigParser.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfigParser.java similarity index 94% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfigParser.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfigParser.java index d50a463c2..11afc270b 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyConfigParser.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyConfigParser.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.jetty; +package org.skywalking.apm.collector.agentstream.jetty; import java.util.Map; import org.skywalking.apm.collector.core.config.ConfigParseException; diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleDefine.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleDefine.java similarity index 75% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleDefine.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleDefine.java index b23de9696..ac14352cb 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleDefine.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleDefine.java @@ -1,8 +1,8 @@ -package org.skywalking.apm.collector.agent.stream.server.jetty; +package org.skywalking.apm.collector.agentstream.jetty; -import org.skywalking.apm.collector.core.agentstream.AgentStreamModuleDefine; +import org.skywalking.apm.collector.agentstream.AgentStreamModuleDefine; +import org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; import org.skywalking.apm.collector.core.module.ModuleRegistration; import org.skywalking.apm.collector.core.server.Server; import org.skywalking.apm.collector.server.jetty.JettyServer; @@ -12,8 +12,8 @@ import org.skywalking.apm.collector.server.jetty.JettyServer; */ public class AgentStreamJettyModuleDefine extends AgentStreamModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.AgentStream; + @Override protected String group() { + return AgentStreamModuleGroupDefine.GROUP_NAME; } @Override public String name() { diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleRegistration.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleRegistration.java similarity index 88% rename from apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleRegistration.java rename to apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleRegistration.java index 82ac5aa62..3968fb8f7 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agent/stream/server/jetty/AgentStreamJettyModuleRegistration.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleRegistration.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.agent.stream.server.jetty; +package org.skywalking.apm.collector.agentstream.jetty; import com.google.gson.JsonObject; import org.skywalking.apm.collector.core.module.ModuleRegistration; diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/group.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/group.define new file mode 100644 index 000000000..2c200d484 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/group.define @@ -0,0 +1 @@ +org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/module.define b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/module.define index 6891cf0e2..7103f4219 100644 --- a/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/module.define +++ b/apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/module.define @@ -1,2 +1,2 @@ -org.skywalking.apm.collector.agent.stream.server.grpc.AgentStreamGRPCModuleDefine -org.skywalking.apm.collector.agent.stream.server.jetty.AgentStreamJettyModuleDefine \ No newline at end of file +org.skywalking.apm.collector.agentstream.grpc.AgentStreamGRPCModuleDefine +org.skywalking.apm.collector.agentstream.jetty.AgentStreamJettyModuleDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorBootStartUp.java b/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorBootStartUp.java index 46acc6cc5..ef2dc90df 100644 --- a/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorBootStartUp.java +++ b/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorBootStartUp.java @@ -2,7 +2,6 @@ package org.skywalking.apm.collector.boot; import org.skywalking.apm.collector.core.client.ClientException; import org.skywalking.apm.collector.core.config.ConfigException; -import org.skywalking.apm.collector.core.framework.CollectorStarter; import org.skywalking.apm.collector.core.framework.DefineException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorStarter.java b/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorStarter.java similarity index 57% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorStarter.java rename to apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorStarter.java index 44534172d..a5d55fd25 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorStarter.java +++ b/apm-collector/apm-collector-boot/src/main/java/org/skywalking/apm/collector/boot/CollectorStarter.java @@ -1,15 +1,16 @@ -package org.skywalking.apm.collector.core.framework; +package org.skywalking.apm.collector.boot; import java.util.Map; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; import org.skywalking.apm.collector.core.client.ClientException; import org.skywalking.apm.collector.core.config.ConfigException; +import org.skywalking.apm.collector.core.framework.DefineException; +import org.skywalking.apm.collector.core.framework.Starter; import org.skywalking.apm.collector.core.module.ModuleConfigLoader; import org.skywalking.apm.collector.core.module.ModuleDefine; import org.skywalking.apm.collector.core.module.ModuleDefineLoader; -import org.skywalking.apm.collector.core.module.ModuleGroup; import org.skywalking.apm.collector.core.module.ModuleGroupDefine; import org.skywalking.apm.collector.core.module.ModuleGroupDefineLoader; -import org.skywalking.apm.collector.core.module.ModuleInstallerAdapter; import org.skywalking.apm.collector.core.remote.SerializedDefineLoader; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -22,29 +23,23 @@ public class CollectorStarter implements Starter { private final Logger logger = LoggerFactory.getLogger(CollectorStarter.class); @Override public void start() throws ConfigException, DefineException, ClientException { - Context context = new Context(); ModuleConfigLoader configLoader = new ModuleConfigLoader(); Map configuration = configLoader.load(); SerializedDefineLoader serializedDefineLoader = new SerializedDefineLoader(); serializedDefineLoader.load(); - ModuleDefineLoader defineLoader = new ModuleDefineLoader(); - Map> moduleDefineMap = defineLoader.load(); - ModuleGroupDefineLoader groupDefineLoader = new ModuleGroupDefineLoader(); Map moduleGroupDefineMap = groupDefineLoader.load(); - ModuleInstallerAdapter moduleInstallerAdapter = new ModuleInstallerAdapter(ModuleGroup.Cluster); - moduleInstallerAdapter.install(configuration.get(ModuleGroup.Cluster.name().toLowerCase()), moduleDefineMap.get(ModuleGroup.Cluster.name().toLowerCase())); + ModuleDefineLoader defineLoader = new ModuleDefineLoader(); + Map> moduleDefineMap = defineLoader.load(); - ModuleGroup[] moduleGroups = ModuleGroup.values(); - for (ModuleGroup moduleGroup : moduleGroups) { - if (!ModuleGroup.Cluster.equals(moduleGroup)) { - moduleInstallerAdapter = new ModuleInstallerAdapter(moduleGroup); - logger.info("module group {}, configuration {}", moduleGroup.name().toLowerCase(), configuration.get(moduleGroup.name().toLowerCase())); - moduleInstallerAdapter.install(configuration.get(moduleGroup.name().toLowerCase()), moduleDefineMap.get(moduleGroup.name().toLowerCase())); - } + moduleGroupDefineMap.get(ClusterModuleGroupDefine.GROUP_NAME).moduleInstaller().install(configuration.get(ClusterModuleGroupDefine.GROUP_NAME), moduleDefineMap.get(ClusterModuleGroupDefine.GROUP_NAME)); + moduleGroupDefineMap.remove(ClusterModuleGroupDefine.GROUP_NAME); + + for (ModuleGroupDefine moduleGroupDefine : moduleGroupDefineMap.values()) { + moduleGroupDefine.moduleInstaller().install(configuration.get(moduleGroupDefine.name()), moduleDefineMap.get(moduleGroupDefine.name())); } } } diff --git a/apm-collector/apm-collector-client/pom.xml b/apm-collector/apm-collector-client/pom.xml index 1de50be53..ae71e6f14 100644 --- a/apm-collector/apm-collector-client/pom.xml +++ b/apm-collector/apm-collector-client/pom.xml @@ -32,6 +32,16 @@ org.apache.zookeeper zookeeper 3.4.10 + + + slf4j-api + org.slf4j + + + slf4j-log4j12 + org.slf4j + + \ No newline at end of file diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleGroupDefine.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleGroupDefine.java index 4f5246442..d51023d8a 100644 --- a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleGroupDefine.java +++ b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleGroupDefine.java @@ -1,17 +1,26 @@ package org.skywalking.apm.collector.cluster; +import org.skywalking.apm.collector.core.cluster.ClusterModuleContext; +import org.skywalking.apm.collector.core.framework.Context; import org.skywalking.apm.collector.core.module.ModuleGroupDefine; -import org.skywalking.apm.collector.core.module.ModuleInstallMode; +import org.skywalking.apm.collector.core.module.ModuleInstaller; /** * @author pengys5 */ public class ClusterModuleGroupDefine implements ModuleGroupDefine { + + public static final String GROUP_NAME = "cluster"; + @Override public String name() { - return "cluster"; + return GROUP_NAME; } - @Override public ModuleInstallMode mode() { - return ModuleInstallMode.Single; + @Override public Context groupContext() { + return new ClusterModuleContext(GROUP_NAME); + } + + @Override public ModuleInstaller moduleInstaller() { + return new ClusterModuleInstaller(); } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleInstaller.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleInstaller.java similarity index 77% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleInstaller.java rename to apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleInstaller.java index d632bace9..666366244 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleInstaller.java +++ b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/ClusterModuleInstaller.java @@ -1,8 +1,11 @@ -package org.skywalking.apm.collector.core.cluster; +package org.skywalking.apm.collector.cluster; import java.util.Iterator; import java.util.Map; import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.cluster.ClusterModuleContext; +import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.core.framework.DefineException; import org.skywalking.apm.collector.core.module.ModuleDefine; import org.skywalking.apm.collector.core.module.ModuleInstaller; @@ -38,6 +41,10 @@ public class ClusterModuleInstaller implements ModuleInstaller { moduleDefine = moduleDefineMap.get(clusterConfigEntry.getKey()); moduleDefine.initialize(clusterConfigEntry.getValue()); } - ClusterModuleContext.WRITER = ((ClusterModuleDefine)moduleDefine).registrationWriter(); + + ClusterModuleContext context = new ClusterModuleContext(ClusterModuleGroupDefine.GROUP_NAME); + context.setWriter(((ClusterModuleDefine)moduleDefine).registrationWriter()); + + CollectorContextHelper.INSTANCE.putContext(context); } } diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleDefine.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleDefine.java index cc7e9c71f..625d894fa 100644 --- a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleDefine.java +++ b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/redis/ClusterRedisModuleDefine.java @@ -1,25 +1,27 @@ package org.skywalking.apm.collector.cluster.redis; import org.skywalking.apm.collector.client.redis.RedisClient; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; import org.skywalking.apm.collector.core.client.Client; import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter; import org.skywalking.apm.collector.core.framework.DataInitializer; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; /** * @author pengys5 */ public class ClusterRedisModuleDefine extends ClusterModuleDefine { - @Override public ModuleGroup group() { - return ModuleGroup.Cluster; + public static final String MODULE_NAME = "redis"; + + @Override public String group() { + return ClusterModuleGroupDefine.GROUP_NAME; } @Override public String name() { - return "redis"; + return MODULE_NAME; } @Override public boolean defaultModule() { @@ -38,11 +40,11 @@ public class ClusterRedisModuleDefine extends ClusterModuleDefine { return new ClusterRedisDataInitializer(); } - @Override protected ClusterModuleRegistrationWriter registrationWriter() { + @Override public ClusterModuleRegistrationWriter registrationWriter() { return new ClusterRedisModuleRegistrationWriter(getClient()); } - @Override protected ClusterModuleRegistrationReader registrationReader() { + @Override public ClusterModuleRegistrationReader registrationReader() { return null; } } diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java index d7802e216..583439d7c 100644 --- a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java +++ b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/standalone/ClusterStandaloneModuleDefine.java @@ -1,25 +1,27 @@ package org.skywalking.apm.collector.cluster.standalone; import org.skywalking.apm.collector.client.h2.H2Client; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; import org.skywalking.apm.collector.core.client.Client; import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter; import org.skywalking.apm.collector.core.framework.DataInitializer; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; /** * @author pengys5 */ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine { - @Override public ModuleGroup group() { - return ModuleGroup.Cluster; + public static final String MODULE_NAME = "standalone"; + + @Override public String group() { + return ClusterModuleGroupDefine.GROUP_NAME; } @Override public String name() { - return "standalone"; + return MODULE_NAME; } @Override public boolean defaultModule() { @@ -38,11 +40,11 @@ public class ClusterStandaloneModuleDefine extends ClusterModuleDefine { return new ClusterStandaloneDataInitializer(); } - @Override protected ClusterModuleRegistrationWriter registrationWriter() { + @Override public ClusterModuleRegistrationWriter registrationWriter() { return new ClusterStandaloneModuleRegistrationWriter(getClient()); } - @Override protected ClusterModuleRegistrationReader registrationReader() { + @Override public ClusterModuleRegistrationReader registrationReader() { return null; } } diff --git a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleDefine.java b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleDefine.java index b555ac458..f9d570601 100644 --- a/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleDefine.java +++ b/apm-collector/apm-collector-cluster/src/main/java/org/skywalking/apm/collector/cluster/zookeeper/ClusterZKModuleDefine.java @@ -1,25 +1,27 @@ package org.skywalking.apm.collector.cluster.zookeeper; import org.skywalking.apm.collector.client.zookeeper.ZookeeperClient; +import org.skywalking.apm.collector.cluster.ClusterModuleGroupDefine; import org.skywalking.apm.collector.core.client.Client; import org.skywalking.apm.collector.core.cluster.ClusterDataInitializer; import org.skywalking.apm.collector.core.cluster.ClusterModuleDefine; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationReader; import org.skywalking.apm.collector.core.cluster.ClusterModuleRegistrationWriter; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; /** * @author pengys5 */ public class ClusterZKModuleDefine extends ClusterModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.Cluster; + public static final String MODULE_NAME = "zookeeper"; + + @Override protected String group() { + return ClusterModuleGroupDefine.GROUP_NAME; } @Override public String name() { - return "zookeeper"; + return MODULE_NAME; } @Override public boolean defaultModule() { @@ -38,11 +40,11 @@ public class ClusterZKModuleDefine extends ClusterModuleDefine { return new ClusterZKDataInitializer(); } - @Override protected ClusterModuleRegistrationWriter registrationWriter() { + @Override public ClusterModuleRegistrationWriter registrationWriter() { return new ClusterZKModuleRegistrationWriter(getClient()); } - @Override protected ClusterModuleRegistrationReader registrationReader() { + @Override public ClusterModuleRegistrationReader registrationReader() { return null; } } diff --git a/apm-collector/apm-collector-core/pom.xml b/apm-collector/apm-collector-core/pom.xml index 5c21a41b7..8c3d9f630 100644 --- a/apm-collector/apm-collector-core/pom.xml +++ b/apm-collector/apm-collector-core/pom.xml @@ -18,11 +18,6 @@ snakeyaml 1.18 - - ch.qos.logback - logback-classic - 1.2.3 - com.google.code.gson gson diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleContext.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleContext.java index 41a6a2c52..a104e24ea 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleContext.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleContext.java @@ -1,10 +1,33 @@ package org.skywalking.apm.collector.core.cluster; +import org.skywalking.apm.collector.core.framework.Context; + /** * @author pengys5 */ -public class ClusterModuleContext { - public static ClusterModuleRegistrationWriter WRITER; +public class ClusterModuleContext extends Context { - public static ClusterModuleRegistrationReader READER; + public ClusterModuleContext(String groupName) { + super(groupName); + } + + private ClusterModuleRegistrationWriter writer; + + private ClusterModuleRegistrationReader reader; + + public ClusterModuleRegistrationWriter getWriter() { + return writer; + } + + public void setWriter(ClusterModuleRegistrationWriter writer) { + this.writer = writer; + } + + public ClusterModuleRegistrationReader getReader() { + return reader; + } + + public void setReader(ClusterModuleRegistrationReader reader) { + this.reader = reader; + } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleDefine.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleDefine.java index 42e4016c3..1d788d8ec 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleDefine.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/cluster/ClusterModuleDefine.java @@ -38,7 +38,7 @@ public abstract class ClusterModuleDefine extends ModuleDefine { throw new UnsupportedOperationException("Cluster module do not need module registration."); } - protected abstract ClusterModuleRegistrationWriter registrationWriter(); + public abstract ClusterModuleRegistrationWriter registrationWriter(); - protected abstract ClusterModuleRegistrationReader registrationReader(); + public abstract ClusterModuleRegistrationReader registrationReader(); } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorContextHelper.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorContextHelper.java new file mode 100644 index 000000000..b0d6e96cf --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/CollectorContextHelper.java @@ -0,0 +1,25 @@ +package org.skywalking.apm.collector.core.framework; + +import java.util.LinkedHashMap; +import java.util.Map; + +/** + * @author pengys5 + */ +public enum CollectorContextHelper { + INSTANCE; + + private Map contexts = new LinkedHashMap(); + + public Context getContext(String moduleGroupName) { + return contexts.get(moduleGroupName); + } + + public void putContext(Context context) { + if (contexts.containsKey(context.getGroupName())) { + throw new UnsupportedOperationException("This module context was put, do not allow put a new one"); + } else { + contexts.put(context.getGroupName(), context); + } + } +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Context.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Context.java index db34bb29a..9be638427 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Context.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Context.java @@ -3,5 +3,14 @@ package org.skywalking.apm.collector.core.framework; /** * @author pengys5 */ -public class Context { +public abstract class Context { + private final String groupName; + + public Context(String groupName) { + this.groupName = groupName; + } + + public final String getGroupName() { + return groupName; + } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Executor.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Executor.java new file mode 100644 index 000000000..b09717518 --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/framework/Executor.java @@ -0,0 +1,8 @@ +package org.skywalking.apm.collector.core.framework; + +/** + * @author pengys5 + */ +public interface Executor { + void execute(Object message); +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleConfigLoader.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleConfigLoader.java index 55c8a0e78..8e89ef152 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleConfigLoader.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleConfigLoader.java @@ -1,6 +1,7 @@ package org.skywalking.apm.collector.core.module; import java.io.FileNotFoundException; +import java.io.FileReader; import java.util.Map; import org.skywalking.apm.collector.core.config.ConfigLoader; import org.skywalking.apm.collector.core.util.ResourceUtils; @@ -18,7 +19,13 @@ public class ModuleConfigLoader implements ConfigLoader> { @Override public Map load() throws ModuleConfigLoaderException { Yaml yaml = new Yaml(); try { - return (Map)yaml.load(ResourceUtils.read("application.yml")); + try { + FileReader applicationFileReader = ResourceUtils.read("application.yml"); + return (Map)yaml.load(applicationFileReader); + } catch (FileNotFoundException e) { + logger.info("Could not found application.yml file, use default"); + return (Map)yaml.load(ResourceUtils.read("application-default.yml")); + } } catch (FileNotFoundException e) { throw new ModuleConfigLoaderException(e.getMessage(), e); } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefine.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefine.java index 1ead094ea..eeb3f56e3 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefine.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefine.java @@ -10,7 +10,7 @@ import org.skywalking.apm.collector.core.server.Server; */ public abstract class ModuleDefine implements Define { - protected abstract ModuleGroup group(); + protected abstract String group(); public abstract boolean defaultModule(); diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefineLoader.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefineLoader.java index 407bc13ae..67a3adb60 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefineLoader.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleDefineLoader.java @@ -24,7 +24,7 @@ public class ModuleDefineLoader implements Loader()); } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroup.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroup.java deleted file mode 100644 index 31d9e4aed..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroup.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.skywalking.apm.collector.core.module; - -/** - * @author pengys5 - */ -public enum ModuleGroup { - Cluster, Queue, AgentStream -} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroupDefine.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroupDefine.java index 8bca7831e..50783f9d3 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroupDefine.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleGroupDefine.java @@ -1,10 +1,14 @@ package org.skywalking.apm.collector.core.module; +import org.skywalking.apm.collector.core.framework.Context; + /** * @author pengys5 */ public interface ModuleGroupDefine { String name(); - ModuleInstallMode mode(); + Context groupContext(); + + ModuleInstaller moduleInstaller(); } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallMode.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallMode.java deleted file mode 100644 index 3b2976797..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallMode.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.skywalking.apm.collector.core.module; - -/** - * @author pengys5 - */ -public enum ModuleInstallMode { - Single, Multiple -} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallerAdapter.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallerAdapter.java deleted file mode 100644 index 48c6c2d19..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/module/ModuleInstallerAdapter.java +++ /dev/null @@ -1,25 +0,0 @@ -package org.skywalking.apm.collector.core.module; - -import java.util.Map; -import org.skywalking.apm.collector.core.client.ClientException; -import org.skywalking.apm.collector.core.cluster.ClusterModuleInstaller; -import org.skywalking.apm.collector.core.framework.DefineException; - -/** - * @author pengys5 - */ -public class ModuleInstallerAdapter implements ModuleInstaller { - - private ModuleInstaller moduleInstaller; - - public ModuleInstallerAdapter(ModuleGroup moduleGroup) { - if (ModuleGroup.Cluster.equals(moduleGroup)) { - moduleInstaller = new ClusterModuleInstaller(); - } - } - - @Override public void install(Map moduleConfig, - Map moduleDefineMap) throws DefineException, ClientException { - moduleInstaller.install(moduleConfig, moduleDefineMap); - } -} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/Creator.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/Creator.java deleted file mode 100644 index c2da5c469..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/Creator.java +++ /dev/null @@ -1,10 +0,0 @@ -package org.skywalking.apm.collector.core.queue; - -import org.skywalking.apm.collector.core.worker.AbstractLocalAsyncWorker; - -/** - * @author pengys5 - */ -public interface Creator { - QueueEventHandler create(int queueSize, AbstractLocalAsyncWorker localAsyncWorker); -} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueCreator.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueCreator.java new file mode 100644 index 000000000..d404f23a6 --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueCreator.java @@ -0,0 +1,8 @@ +package org.skywalking.apm.collector.core.queue; + +/** + * @author pengys5 + */ +public interface QueueCreator { + QueueEventHandler create(int queueSize, QueueExecutor executor); +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueExecutor.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueExecutor.java new file mode 100644 index 000000000..1576caf58 --- /dev/null +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueExecutor.java @@ -0,0 +1,9 @@ +package org.skywalking.apm.collector.core.queue; + +import org.skywalking.apm.collector.core.framework.Executor; + +/** + * @author pengys5 + */ +public interface QueueExecutor extends Executor { +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleContext.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleContext.java deleted file mode 100644 index 7825ca638..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleContext.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.skywalking.apm.collector.core.queue; - -/** - * @author pengys5 - */ -public class QueueModuleContext { - public static Creator CREATOR; -} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/util/ResourceUtils.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/util/ResourceUtils.java index 31daeec6f..257f85dd2 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/util/ResourceUtils.java +++ b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/util/ResourceUtils.java @@ -1,16 +1,21 @@ package org.skywalking.apm.collector.core.util; +import java.io.File; import java.io.FileNotFoundException; import java.io.FileReader; +import java.net.URL; /** * @author pengys5 */ public class ResourceUtils { - private static final String PATH = ResourceUtils.class.getResource("/").getPath(); - public static FileReader read(String fileName) throws FileNotFoundException { - return new FileReader(PATH + fileName); + URL url = ResourceUtils.class.getClassLoader().getResource(fileName); + if (url == null) { + throw new FileNotFoundException("file not found: " + fileName); + } + File file = new File(ResourceUtils.class.getClassLoader().getResource(fileName).getFile()); + return new FileReader(file); } } diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Role.java b/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Role.java deleted file mode 100644 index 8c59ff0ac..000000000 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Role.java +++ /dev/null @@ -1,13 +0,0 @@ -package org.skywalking.apm.collector.core.worker; - -import org.skywalking.apm.collector.core.worker.selector.WorkerSelector; - -/** - * @author pengys5 - */ -public interface Role { - - String roleName(); - - WorkerSelector workerSelector(); -} diff --git a/apm-collector/apm-collector-boot/src/main/resources/application.yml b/apm-collector/apm-collector-core/src/main/resources/application-default.yml similarity index 100% rename from apm-collector/apm-collector-boot/src/main/resources/application.yml rename to apm-collector/apm-collector-core/src/main/resources/application-default.yml diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleContext.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleContext.java new file mode 100644 index 000000000..cf9bf00cd --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleContext.java @@ -0,0 +1,23 @@ +package org.skywalking.apm.collector.queue; + +import org.skywalking.apm.collector.core.framework.Context; +import org.skywalking.apm.collector.core.queue.QueueCreator; + +/** + * @author pengys5 + */ +public class QueueModuleContext extends Context { + private QueueCreator queueCreator; + + public QueueModuleContext(String groupName) { + super(groupName); + } + + public QueueCreator getQueueCreator() { + return queueCreator; + } + + public void setQueueCreator(QueueCreator queueCreator) { + this.queueCreator = queueCreator; + } +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleDefine.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleDefine.java similarity index 95% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleDefine.java rename to apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleDefine.java index 9b6bd447f..087792f22 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/queue/QueueModuleDefine.java +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleDefine.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.queue; +package org.skywalking.apm.collector.queue; import org.skywalking.apm.collector.core.client.Client; import org.skywalking.apm.collector.core.framework.DataInitializer; diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleException.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleException.java new file mode 100644 index 000000000..b22b24089 --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleException.java @@ -0,0 +1,16 @@ +package org.skywalking.apm.collector.queue; + +import org.skywalking.apm.collector.core.module.ModuleException; + +/** + * @author pengys5 + */ +public class QueueModuleException extends ModuleException { + public QueueModuleException(String message) { + super(message); + } + + public QueueModuleException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleGroupDefine.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleGroupDefine.java new file mode 100644 index 000000000..9b40b7c2c --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleGroupDefine.java @@ -0,0 +1,25 @@ +package org.skywalking.apm.collector.queue; + +import org.skywalking.apm.collector.core.framework.Context; +import org.skywalking.apm.collector.core.module.ModuleGroupDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; + +/** + * @author pengys5 + */ +public class QueueModuleGroupDefine implements ModuleGroupDefine { + + public static final String GROUP_NAME = "queue"; + + @Override public String name() { + return GROUP_NAME; + } + + @Override public Context groupContext() { + return new QueueModuleContext(GROUP_NAME); + } + + @Override public ModuleInstaller moduleInstaller() { + return new QueueModuleInstaller(); + } +} diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleInstaller.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleInstaller.java new file mode 100644 index 000000000..6b4353efd --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/QueueModuleInstaller.java @@ -0,0 +1,23 @@ +package org.skywalking.apm.collector.queue; + +import java.util.Map; +import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.framework.DefineException; +import org.skywalking.apm.collector.core.module.ModuleDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class QueueModuleInstaller implements ModuleInstaller { + + private final Logger logger = LoggerFactory.getLogger(QueueModuleInstaller.class); + + @Override public void install(Map moduleConfig, + Map moduleDefineMap) throws DefineException, ClientException { + logger.info("beginning cluster module install"); + + } +} diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierCreator.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierCreator.java deleted file mode 100644 index 2402b83f4..000000000 --- a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierCreator.java +++ /dev/null @@ -1,15 +0,0 @@ -package org.skywalking.apm.collector.queue.datacarrier; - -import org.skywalking.apm.collector.core.queue.Creator; -import org.skywalking.apm.collector.core.queue.QueueEventHandler; -import org.skywalking.apm.collector.core.worker.AbstractLocalAsyncWorker; - -/** - * @author pengys5 - */ -public class DataCarrierCreator implements Creator { - - @Override public QueueEventHandler create(int queueSize, AbstractLocalAsyncWorker localAsyncWorker) { - return null; - } -} diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierQueueCreator.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierQueueCreator.java new file mode 100644 index 000000000..e8593d6fc --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/DataCarrierQueueCreator.java @@ -0,0 +1,15 @@ +package org.skywalking.apm.collector.queue.datacarrier; + +import org.skywalking.apm.collector.core.queue.QueueCreator; +import org.skywalking.apm.collector.core.queue.QueueEventHandler; +import org.skywalking.apm.collector.core.queue.QueueExecutor; + +/** + * @author pengys5 + */ +public class DataCarrierQueueCreator implements QueueCreator { + + @Override public QueueEventHandler create(int queueSize, QueueExecutor executor) { + return null; + } +} diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/QueueDataCarrierModuleDefine.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/QueueDataCarrierModuleDefine.java index 572557711..a5880c1b7 100644 --- a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/QueueDataCarrierModuleDefine.java +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/datacarrier/QueueDataCarrierModuleDefine.java @@ -2,18 +2,19 @@ package org.skywalking.apm.collector.queue.datacarrier; import java.util.Map; import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.core.framework.DefineException; -import org.skywalking.apm.collector.core.module.ModuleGroup; -import org.skywalking.apm.collector.core.queue.QueueModuleContext; -import org.skywalking.apm.collector.core.queue.QueueModuleDefine; +import org.skywalking.apm.collector.queue.QueueModuleContext; +import org.skywalking.apm.collector.queue.QueueModuleDefine; +import org.skywalking.apm.collector.queue.QueueModuleGroupDefine; /** * @author pengys5 */ public class QueueDataCarrierModuleDefine extends QueueModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.Queue; + @Override protected String group() { + return QueueModuleGroupDefine.GROUP_NAME; } @Override public String name() { @@ -25,6 +26,6 @@ public class QueueDataCarrierModuleDefine extends QueueModuleDefine { } @Override public final void initialize(Map config) throws DefineException, ClientException { - QueueModuleContext.CREATOR = new DataCarrierCreator(); + ((QueueModuleContext)CollectorContextHelper.INSTANCE.getContext(group())).setQueueCreator(new DataCarrierQueueCreator()); } } diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorEventHandler.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorEventHandler.java index 34090fa08..f25396531 100644 --- a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorEventHandler.java +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorEventHandler.java @@ -5,8 +5,7 @@ import com.lmax.disruptor.RingBuffer; import org.skywalking.apm.collector.core.queue.EndOfBatchCommand; import org.skywalking.apm.collector.core.queue.MessageHolder; import org.skywalking.apm.collector.core.queue.QueueEventHandler; -import org.skywalking.apm.collector.core.worker.AbstractLocalAsyncWorker; -import org.skywalking.apm.collector.core.worker.WorkerException; +import org.skywalking.apm.collector.core.queue.QueueExecutor; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,11 +17,11 @@ public class DisruptorEventHandler implements EventHandler, Queue private final Logger logger = LoggerFactory.getLogger(DisruptorEventHandler.class); private RingBuffer ringBuffer; - private AbstractLocalAsyncWorker asyncWorker; + private QueueExecutor executor; - DisruptorEventHandler(RingBuffer ringBuffer, AbstractLocalAsyncWorker asyncWorker) { + DisruptorEventHandler(RingBuffer ringBuffer, QueueExecutor executor) { this.ringBuffer = ringBuffer; - this.asyncWorker = asyncWorker; + this.executor = executor; } /** @@ -34,16 +33,12 @@ public class DisruptorEventHandler implements EventHandler, Queue * @param endOfBatch flag to indicate if this is the last event in a batch from the {@link RingBuffer} */ public void onEvent(MessageHolder event, long sequence, boolean endOfBatch) { - try { - Object message = event.getMessage(); - event.reset(); + Object message = event.getMessage(); + event.reset(); - asyncWorker.allocateJob(message); - if (endOfBatch) { - asyncWorker.allocateJob(new EndOfBatchCommand()); - } - } catch (WorkerException e) { - logger.error(e.getMessage(), e); + executor.execute(message); + if (endOfBatch) { + executor.execute(new EndOfBatchCommand()); } } diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorCreator.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorQueueCreator.java similarity index 77% rename from apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorCreator.java rename to apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorQueueCreator.java index 8be095b26..df027d0e6 100644 --- a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorCreator.java +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/DisruptorQueueCreator.java @@ -2,18 +2,18 @@ package org.skywalking.apm.collector.queue.disruptor; import com.lmax.disruptor.RingBuffer; import com.lmax.disruptor.dsl.Disruptor; -import org.skywalking.apm.collector.core.queue.Creator; import org.skywalking.apm.collector.core.queue.DaemonThreadFactory; import org.skywalking.apm.collector.core.queue.MessageHolder; +import org.skywalking.apm.collector.core.queue.QueueCreator; import org.skywalking.apm.collector.core.queue.QueueEventHandler; -import org.skywalking.apm.collector.core.worker.AbstractLocalAsyncWorker; +import org.skywalking.apm.collector.core.queue.QueueExecutor; /** * @author pengys5 */ -public class DisruptorCreator implements Creator { +public class DisruptorQueueCreator implements QueueCreator { - public QueueEventHandler create(int queueSize, AbstractLocalAsyncWorker localAsyncWorker) { + @Override public QueueEventHandler create(int queueSize, QueueExecutor executor) { // Specify the size of the ring buffer, must be power of 2. if (!((((queueSize - 1) & queueSize) == 0) && queueSize != 0)) { throw new IllegalArgumentException("queue size must be power of 2"); @@ -23,7 +23,7 @@ public class DisruptorCreator implements Creator { Disruptor disruptor = new Disruptor(MessageHolderFactory.INSTANCE, queueSize, DaemonThreadFactory.INSTANCE); RingBuffer ringBuffer = disruptor.getRingBuffer(); - DisruptorEventHandler eventHandler = new DisruptorEventHandler(ringBuffer, localAsyncWorker); + DisruptorEventHandler eventHandler = new DisruptorEventHandler(ringBuffer, executor); // Connect the handler disruptor.handleEventsWith(eventHandler); diff --git a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/QueueDisruptorModuleDefine.java b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/QueueDisruptorModuleDefine.java index cae0f1325..f550dd4e1 100644 --- a/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/QueueDisruptorModuleDefine.java +++ b/apm-collector/apm-collector-queue/src/main/java/org/skywalking/apm/collector/queue/disruptor/QueueDisruptorModuleDefine.java @@ -2,18 +2,19 @@ package org.skywalking.apm.collector.queue.disruptor; import java.util.Map; import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; import org.skywalking.apm.collector.core.framework.DefineException; -import org.skywalking.apm.collector.core.module.ModuleGroup; -import org.skywalking.apm.collector.core.queue.QueueModuleContext; -import org.skywalking.apm.collector.core.queue.QueueModuleDefine; +import org.skywalking.apm.collector.queue.QueueModuleContext; +import org.skywalking.apm.collector.queue.QueueModuleDefine; +import org.skywalking.apm.collector.queue.QueueModuleGroupDefine; /** * @author pengys5 */ public class QueueDisruptorModuleDefine extends QueueModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.Queue; + @Override protected String group() { + return QueueModuleGroupDefine.GROUP_NAME; } @Override public String name() { @@ -25,6 +26,6 @@ public class QueueDisruptorModuleDefine extends QueueModuleDefine { } @Override public final void initialize(Map config) throws DefineException, ClientException { - QueueModuleContext.CREATOR = new DisruptorCreator(); + ((QueueModuleContext)CollectorContextHelper.INSTANCE.getContext(group())).setQueueCreator(new DisruptorQueueCreator()); } } diff --git a/apm-collector/apm-collector-queue/src/main/resources/META-INF/defines/group.define b/apm-collector/apm-collector-queue/src/main/resources/META-INF/defines/group.define new file mode 100644 index 000000000..761c2b7e0 --- /dev/null +++ b/apm-collector/apm-collector-queue/src/main/resources/META-INF/defines/group.define @@ -0,0 +1 @@ +org.skywalking.apm.collector.queue.QueueModuleGroupDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleContext.java b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleContext.java new file mode 100644 index 000000000..f5bb5b805 --- /dev/null +++ b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleContext.java @@ -0,0 +1,12 @@ +package org.skywalking.apm.collector.remote; + +import org.skywalking.apm.collector.core.framework.Context; + +/** + * @author pengys5 + */ +public class RemoteModuleContext extends Context { + public RemoteModuleContext(String groupName) { + super(groupName); + } +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/remote/RemoteModuleDefine.java b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleDefine.java similarity index 75% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/remote/RemoteModuleDefine.java rename to apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleDefine.java index b8e57b5a7..ad6875615 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/remote/RemoteModuleDefine.java +++ b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleDefine.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.remote; +package org.skywalking.apm.collector.remote; import org.skywalking.apm.collector.core.module.ModuleDefine; diff --git a/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleGroupDefine.java b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleGroupDefine.java new file mode 100644 index 000000000..2271155de --- /dev/null +++ b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleGroupDefine.java @@ -0,0 +1,25 @@ +package org.skywalking.apm.collector.remote; + +import org.skywalking.apm.collector.core.framework.Context; +import org.skywalking.apm.collector.core.module.ModuleGroupDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; + +/** + * @author pengys5 + */ +public class RemoteModuleGroupDefine implements ModuleGroupDefine { + + public static final String GROUP_NAME = "remote"; + + @Override public String name() { + return GROUP_NAME; + } + + @Override public Context groupContext() { + return new RemoteModuleContext(GROUP_NAME); + } + + @Override public ModuleInstaller moduleInstaller() { + return new RemoteModuleInstaller(); + } +} diff --git a/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleInstaller.java b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleInstaller.java new file mode 100644 index 000000000..470aa6392 --- /dev/null +++ b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/RemoteModuleInstaller.java @@ -0,0 +1,17 @@ +package org.skywalking.apm.collector.remote; + +import java.util.Map; +import org.skywalking.apm.collector.core.client.ClientException; +import org.skywalking.apm.collector.core.framework.DefineException; +import org.skywalking.apm.collector.core.module.ModuleDefine; +import org.skywalking.apm.collector.core.module.ModuleInstaller; + +/** + * @author pengys5 + */ +public class RemoteModuleInstaller implements ModuleInstaller { + @Override public void install(Map moduleConfig, + Map moduleDefineMap) throws DefineException, ClientException { + + } +} diff --git a/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteGRPCModuleDefine.java b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteGRPCModuleDefine.java index 30b74e893..8711e6e07 100644 --- a/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteGRPCModuleDefine.java +++ b/apm-collector/apm-collector-remote/src/main/java/org/skywalking/apm/collector/remote/grpc/RemoteGRPCModuleDefine.java @@ -6,17 +6,17 @@ import org.skywalking.apm.collector.core.client.ClientException; import org.skywalking.apm.collector.core.framework.DataInitializer; import org.skywalking.apm.collector.core.framework.DefineException; import org.skywalking.apm.collector.core.module.ModuleConfigParser; -import org.skywalking.apm.collector.core.module.ModuleGroup; import org.skywalking.apm.collector.core.module.ModuleRegistration; -import org.skywalking.apm.collector.core.remote.RemoteModuleDefine; import org.skywalking.apm.collector.core.server.Server; +import org.skywalking.apm.collector.remote.RemoteModuleDefine; +import org.skywalking.apm.collector.remote.RemoteModuleGroupDefine; /** * @author pengys5 */ public class RemoteGRPCModuleDefine extends RemoteModuleDefine { - @Override protected ModuleGroup group() { - return ModuleGroup.Queue; + @Override protected String group() { + return RemoteModuleGroupDefine.GROUP_NAME; } @Override public boolean defaultModule() { diff --git a/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/group.define b/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/group.define new file mode 100644 index 000000000..141478aab --- /dev/null +++ b/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/group.define @@ -0,0 +1 @@ +org.skywalking.apm.collector.remote.RemoteModuleGroupDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/module.define b/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/module.define new file mode 100644 index 000000000..02abab5e6 --- /dev/null +++ b/apm-collector/apm-collector-remote/src/main/resources/META-INF/defines/module.define @@ -0,0 +1 @@ +org.skywalking.apm.collector.remote.grpc.RemoteGRPCModuleDefine \ No newline at end of file diff --git a/apm-collector/apm-collector-stream/pom.xml b/apm-collector/apm-collector-stream/pom.xml new file mode 100644 index 000000000..e9be3a610 --- /dev/null +++ b/apm-collector/apm-collector-stream/pom.xml @@ -0,0 +1,27 @@ + + + + apm-collector + org.skywalking + 3.2-2017 + + 4.0.0 + + apm-collector-stream + jar + + + + org.skywalking + apm-collector-core + ${project.version} + + + org.skywalking + apm-collector-queue + ${project.version} + + + \ No newline at end of file diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorker.java similarity index 97% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorker.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorker.java index 56cfc50f4..cc2991b4d 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorker.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * The AbstractClusterWorker implementations represent workers, diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorkerProvider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorkerProvider.java similarity index 96% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorkerProvider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorkerProvider.java index 519689182..3e1fd1337 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractClusterWorkerProvider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractClusterWorkerProvider.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * The AbstractClusterWorkerProvider implementations represent providers, diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorker.java similarity index 97% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorker.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorker.java index bba13892b..01ce9b8c7 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorker.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * The AbstractLocalAsyncWorker implementations represent workers, diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorkerProvider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorkerProvider.java similarity index 55% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorkerProvider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorkerProvider.java index 22b3e9a58..adb9e6aa8 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalAsyncWorkerProvider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalAsyncWorkerProvider.java @@ -1,7 +1,10 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; +import org.skywalking.apm.collector.core.framework.CollectorContextHelper; +import org.skywalking.apm.collector.core.queue.QueueCreator; import org.skywalking.apm.collector.core.queue.QueueEventHandler; -import org.skywalking.apm.collector.core.queue.QueueModuleContext; +import org.skywalking.apm.collector.queue.QueueModuleContext; +import org.skywalking.apm.collector.queue.QueueModuleGroupDefine; /** * @author pengys5 @@ -15,7 +18,8 @@ public abstract class AbstractLocalAsyncWorkerProviderAbstractLocalSyncWorker defines workers who receive data from jvm inside call and response in real diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalSyncWorkerProvider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalSyncWorkerProvider.java similarity index 92% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalSyncWorkerProvider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalSyncWorkerProvider.java index 056b724f6..30ab8713c 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalSyncWorkerProvider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalSyncWorkerProvider.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorker.java similarity index 84% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorker.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorker.java index f370cdac2..e94b6119e 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorker.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorkerProvider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorkerProvider.java similarity index 74% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorkerProvider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorkerProvider.java index 1f7116be4..b0e155c93 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractLocalWorkerProvider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractLocalWorkerProvider.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorker.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorker.java similarity index 82% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorker.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorker.java index cdfc839c7..78105eaf5 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorker.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorker.java @@ -1,9 +1,11 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; + +import org.skywalking.apm.collector.core.framework.Executor; /** * @author pengys5 */ -public abstract class AbstractWorker { +public abstract class AbstractWorker implements Executor { private final LocalWorkerContext selfContext; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorkerProvider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorkerProvider.java similarity index 95% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorkerProvider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorkerProvider.java index bc87c19e4..4e50c2d5c 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/AbstractWorkerProvider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/AbstractWorkerProvider.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerContext.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerContext.java similarity index 96% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerContext.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerContext.java index ebd29da62..d049ec6c0 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerContext.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerContext.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRef.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRef.java similarity index 89% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRef.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRef.java index dc2243f3a..45f13d081 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRef.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRef.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRefCounter.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRefCounter.java similarity index 91% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRefCounter.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRefCounter.java index c82d3953d..85bd2773c 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ClusterWorkerRefCounter.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ClusterWorkerRefCounter.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Context.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Context.java similarity index 85% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Context.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Context.java index 7fdbab6a0..9790d4084 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Context.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Context.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalAsyncWorkerRef.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalAsyncWorkerRef.java similarity index 90% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalAsyncWorkerRef.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalAsyncWorkerRef.java index 84a3000fb..286fde0f6 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalAsyncWorkerRef.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalAsyncWorkerRef.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import org.skywalking.apm.collector.core.queue.QueueEventHandler; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalSyncWorkerRef.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalSyncWorkerRef.java similarity index 92% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalSyncWorkerRef.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalSyncWorkerRef.java index 9eb502dc8..e40cfe0f3 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalSyncWorkerRef.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalSyncWorkerRef.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalWorkerContext.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalWorkerContext.java similarity index 87% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalWorkerContext.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalWorkerContext.java index 3d70bbb56..a5e642f4d 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LocalWorkerContext.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LocalWorkerContext.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LookUp.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LookUp.java similarity index 79% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LookUp.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LookUp.java index 4a94fc2f8..802de88e2 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/LookUp.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/LookUp.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Provider.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Provider.java similarity index 73% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Provider.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Provider.java index d9ccf9769..e81a561f9 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/Provider.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Provider.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ProviderNotFoundException.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ProviderNotFoundException.java similarity index 74% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ProviderNotFoundException.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ProviderNotFoundException.java index 0f44a778f..f6c8f3a72 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/ProviderNotFoundException.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/ProviderNotFoundException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; public class ProviderNotFoundException extends Exception { public ProviderNotFoundException(String message) { diff --git a/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Role.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Role.java new file mode 100644 index 000000000..23df1c250 --- /dev/null +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/Role.java @@ -0,0 +1,13 @@ +package org.skywalking.apm.collector.stream; + +import org.skywalking.apm.collector.stream.selector.WorkerSelector; + +/** + * @author pengys5 + */ +public interface Role { + + String roleName(); + + WorkerSelector workerSelector(); +} diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/UsedRoleNameException.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/UsedRoleNameException.java similarity index 73% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/UsedRoleNameException.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/UsedRoleNameException.java index df2629a23..13cd45d70 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/UsedRoleNameException.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/UsedRoleNameException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; public class UsedRoleNameException extends Exception { public UsedRoleNameException(String message) { diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerContext.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerContext.java similarity index 96% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerContext.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerContext.java index 78ff5e231..2bef9d0dd 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerContext.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerContext.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import java.util.ArrayList; import java.util.List; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerException.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerException.java similarity index 87% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerException.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerException.java index b0bd12310..1729e9449 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerException.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * Defines a general exception a worker can throw when it diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerInvokeException.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerInvokeException.java similarity index 88% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerInvokeException.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerInvokeException.java index 03f5de06f..a93a80a37 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerInvokeException.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerInvokeException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * This exception is raised when worker fails to process job during "call" or "ask" diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerModuleInstaller.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerModuleInstaller.java similarity index 95% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerModuleInstaller.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerModuleInstaller.java index 0b86aa90c..5dd6095ca 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerModuleInstaller.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerModuleInstaller.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import java.util.Map; import org.skywalking.apm.collector.core.client.ClientException; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerNotFoundException.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerNotFoundException.java similarity index 74% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerNotFoundException.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerNotFoundException.java index 6c6fc9b2c..2f454e061 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerNotFoundException.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerNotFoundException.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; public class WorkerNotFoundException extends WorkerException { public WorkerNotFoundException(String message) { diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRef.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRef.java similarity index 85% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRef.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRef.java index 28268d0b7..195950c01 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRef.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRef.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; /** * @author pengys5 diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRefs.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRefs.java similarity index 89% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRefs.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRefs.java index 8708bec21..80cc6fdd9 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/WorkerRefs.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/WorkerRefs.java @@ -1,7 +1,7 @@ -package org.skywalking.apm.collector.core.worker; +package org.skywalking.apm.collector.stream; import java.util.List; -import org.skywalking.apm.collector.core.worker.selector.WorkerSelector; +import org.skywalking.apm.collector.stream.selector.WorkerSelector; import org.slf4j.Logger; import org.slf4j.LoggerFactory; diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/AbstractHashMessage.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/AbstractHashMessage.java similarity index 89% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/AbstractHashMessage.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/AbstractHashMessage.java index 7d13191d9..ef8aedbba 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/AbstractHashMessage.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/AbstractHashMessage.java @@ -1,4 +1,4 @@ -package org.skywalking.apm.collector.core.worker.selector; +package org.skywalking.apm.collector.stream.selector; /** * The AbstractHashMessage implementations represent aggregate message, diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/HashCodeSelector.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/HashCodeSelector.java similarity index 83% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/HashCodeSelector.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/HashCodeSelector.java index fea6323be..e1fdc6cf2 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/HashCodeSelector.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/HashCodeSelector.java @@ -1,7 +1,7 @@ -package org.skywalking.apm.collector.core.worker.selector; +package org.skywalking.apm.collector.stream.selector; import java.util.List; -import org.skywalking.apm.collector.core.worker.WorkerRef; +import org.skywalking.apm.collector.stream.WorkerRef; /** * The HashCodeSelector is a simple implementation of {@link WorkerSelector}. It choose {@link WorkerRef} @@ -17,13 +17,13 @@ public class HashCodeSelector implements WorkerSelector { * Use message hashcode to select {@link WorkerRef}. * * @param members given {@link WorkerRef} list, which size is greater than 0; - * @param message the {@link org.skywalking.apm.collector.core.worker.AbstractWorker} is going to send. + * @param message the {@link org.skywalking.apm.collector.stream.AbstractWorker} is going to send. * @return the selected {@link WorkerRef} */ @Override public WorkerRef select(List members, Object message) { if (message instanceof AbstractHashMessage) { - AbstractHashMessage hashMessage = (AbstractHashMessage) message; + AbstractHashMessage hashMessage = (AbstractHashMessage)message; int size = members.size(); int selectIndex = Math.abs(hashMessage.getHashCode()) % size; return members.get(selectIndex); diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/RollingSelector.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/RollingSelector.java similarity index 82% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/RollingSelector.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/RollingSelector.java index 5bb133477..c68fb3319 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/RollingSelector.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/RollingSelector.java @@ -1,7 +1,7 @@ -package org.skywalking.apm.collector.core.worker.selector; +package org.skywalking.apm.collector.stream.selector; import java.util.List; -import org.skywalking.apm.collector.core.worker.WorkerRef; +import org.skywalking.apm.collector.stream.WorkerRef; /** * The RollingSelector is a simple implementation of {@link WorkerSelector}. @@ -18,7 +18,7 @@ public class RollingSelector implements WorkerSelector { * Use round-robin to select {@link WorkerRef}. * * @param members given {@link WorkerRef} list, which size is greater than 0; - * @param message message the {@link org.skywalking.apm.collector.core.worker.AbstractWorker} is going to send. + * @param message message the {@link org.skywalking.apm.collector.stream.AbstractWorker} is going to send. * @return the selected {@link WorkerRef} */ @Override diff --git a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/WorkerSelector.java b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/WorkerSelector.java similarity index 75% rename from apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/WorkerSelector.java rename to apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/WorkerSelector.java index 6013ac22e..45a217003 100644 --- a/apm-collector/apm-collector-core/src/main/java/org/skywalking/apm/collector/core/worker/selector/WorkerSelector.java +++ b/apm-collector/apm-collector-stream/src/main/java/org/skywalking/apm/collector/stream/selector/WorkerSelector.java @@ -1,7 +1,7 @@ -package org.skywalking.apm.collector.core.worker.selector; +package org.skywalking.apm.collector.stream.selector; import java.util.List; -import org.skywalking.apm.collector.core.worker.WorkerRef; +import org.skywalking.apm.collector.stream.WorkerRef; /** * The WorkerSelector should be implemented by any class whose instances @@ -18,7 +18,7 @@ public interface WorkerSelector { * select a {@link WorkerRef} from a {@link WorkerRef} list. * * @param members given {@link WorkerRef} list, which size is greater than 0; - * @param message the {@link org.skywalking.apm.collector.core.worker.AbstractWorker} is going to send. + * @param message the {@link org.skywalking.apm.collector.stream.AbstractWorker} is going to send. * @return the selected {@link WorkerRef} */ T select(List members, Object message); diff --git a/apm-collector/pom.xml b/apm-collector/pom.xml index d7b882735..9fd1fbfb1 100644 --- a/apm-collector/pom.xml +++ b/apm-collector/pom.xml @@ -14,6 +14,7 @@ apm-collector-ui apm-collector-boot apm-collector-remote + apm-collector-stream apm @@ -24,10 +25,18 @@ pom - 2.4.17 - 2.8.1 + + org.slf4j + slf4j-api + 1.7.25 + + + ch.qos.logback + logback-classic + 1.2.3 +