Fixed #419, make agent jam, agent register as agent stream sub maven module.

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/AgentJVMModuleContext.java

-package org.skywalking.apm.collector.agentjvm;
-
-import org.skywalking.apm.collector.core.framework.Context;
-
-/**
- * @author pengys5
- */
-public class AgentJVMModuleContext extends Context {
-    public AgentJVMModuleContext(String groupName) {
-        super(groupName);
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/AgentJVMModuleDefine.java

-package org.skywalking.apm.collector.agentjvm;
-
-import org.skywalking.apm.collector.core.client.Client;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListenerDefine;
-import org.skywalking.apm.collector.core.module.ModuleDefine;
-
-/**
- * @author pengys5
- */
-public abstract class AgentJVMModuleDefine extends ModuleDefine implements ClusterDataListenerDefine {
-
-    @Override protected final Client createClient() {
-        throw new UnsupportedOperationException("");
-    }
-
-    @Override protected void initializeOtherContext() {
-
-    }
-
-    @Override public final boolean defaultModule() {
-        return true;
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/AgentJVMModuleException.java

-package org.skywalking.apm.collector.agentjvm;
-
-import org.skywalking.apm.collector.core.module.ModuleException;
-
-/**
- * @author pengys5
- */
-public class AgentJVMModuleException extends ModuleException {
-    public AgentJVMModuleException(String message) {
-        super(message);
-    }
-
-    public AgentJVMModuleException(String message, Throwable cause) {
-        super(message, cause);
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/AgentJVMModuleGroupDefine.java

-package org.skywalking.apm.collector.agentjvm;
-
-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 AgentJVMModuleGroupDefine implements ModuleGroupDefine {
-
-    public static final String GROUP_NAME = "agent_jvm";
-
-    private final AgentJVMCommonModuleInstaller installer;
-
-    public AgentJVMModuleGroupDefine() {
-        installer = new AgentJVMCommonModuleInstaller();
-    }
-
-    @Override public String name() {
-        return GROUP_NAME;
-    }
-
-    @Override public Context groupContext() {
-        return new AgentJVMModuleContext(GROUP_NAME);
-    }
-
-    @Override public ModuleInstaller moduleInstaller() {
-        return installer;
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/AgentJVMGRPCConfig.java

-package org.skywalking.apm.collector.agentjvm.grpc;
-
-/**
- * @author pengys5
- */
-public class AgentJVMGRPCConfig {
-    public static String HOST;
-    public static int PORT;
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/AgentJVMGRPCConfigParser.java

-package org.skywalking.apm.collector.agentjvm.grpc;
-
-import java.util.Map;
-import org.skywalking.apm.collector.core.config.ConfigParseException;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.util.ObjectUtils;
-import org.skywalking.apm.collector.core.util.StringUtils;
-
-/**
- * @author pengys5
- */
-public class AgentJVMGRPCConfigParser implements ModuleConfigParser {
-
-    private static final String HOST = "host";
-    private static final String PORT = "port";
-
-    @Override public void parse(Map config) throws ConfigParseException {
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(HOST))) {
-            AgentJVMGRPCConfig.HOST = "localhost";
-        } else {
-            AgentJVMGRPCConfig.HOST = (String)config.get(HOST);
-        }
-
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(PORT))) {
-            AgentJVMGRPCConfig.PORT = 11800;
-        } else {
-            AgentJVMGRPCConfig.PORT = (Integer)config.get(PORT);
-        }
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/AgentJVMGRPCDataListener.java

-package org.skywalking.apm.collector.agentjvm.grpc;
-
-import org.skywalking.apm.collector.agentjvm.AgentJVMModuleGroupDefine;
-import org.skywalking.apm.collector.cluster.ClusterModuleDefine;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-
-/**
- * @author pengys5
- */
-public class AgentJVMGRPCDataListener extends ClusterDataListener {
-
-    public static final String PATH = ClusterModuleDefine.BASE_CATALOG + "." + AgentJVMModuleGroupDefine.GROUP_NAME + "." + AgentJVMGRPCModuleDefine.MODULE_NAME;
-
-    @Override public String path() {
-        return PATH;
-    }
-
-    @Override public void serverJoinNotify(String serverAddress) {
-
-    }
-
-    @Override public void serverQuitNotify(String serverAddress) {
-
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/AgentJVMGRPCModuleDefine.java

-package org.skywalking.apm.collector.agentjvm.grpc;
-
-import java.util.LinkedList;
-import java.util.List;
-import org.skywalking.apm.collector.agentjvm.AgentJVMModuleDefine;
-import org.skywalking.apm.collector.agentjvm.AgentJVMModuleGroupDefine;
-import org.skywalking.apm.collector.agentjvm.grpc.handler.JVMMetricsServiceHandler;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-import org.skywalking.apm.collector.core.framework.Handler;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-import org.skywalking.apm.collector.core.server.Server;
-import org.skywalking.apm.collector.server.grpc.GRPCServer;
-
-/**
- * @author pengys5
- */
-public class AgentJVMGRPCModuleDefine extends AgentJVMModuleDefine {
-
-    public static final String MODULE_NAME = "grpc";
-
-    @Override protected String group() {
-        return AgentJVMModuleGroupDefine.GROUP_NAME;
-    }
-
-    @Override public String name() {
-        return MODULE_NAME;
-    }
-
-    @Override protected ModuleConfigParser configParser() {
-        return new AgentJVMGRPCConfigParser();
-    }
-
-    @Override protected Server server() {
-        return new GRPCServer(AgentJVMGRPCConfig.HOST, AgentJVMGRPCConfig.PORT);
-    }
-
-    @Override protected ModuleRegistration registration() {
-        return new AgentJVMGRPCModuleRegistration();
-    }
-
-    @Override public ClusterDataListener listener() {
-        return new AgentJVMGRPCDataListener();
-    }
-
-    @Override public List<Handler> handlerList() {
-        List<Handler> handlers = new LinkedList<>();
-        handlers.add(new JVMMetricsServiceHandler());
-        return handlers;
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/grpc/AgentJVMGRPCModuleRegistration.java

-package org.skywalking.apm.collector.agentjvm.grpc;
-
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-
-/**
- * @author pengys5
- */
-public class AgentJVMGRPCModuleRegistration extends ModuleRegistration {
-
-    @Override public Value buildValue() {
-        return new Value(AgentJVMGRPCConfig.HOST, AgentJVMGRPCConfig.PORT, null);
-    }
-}

apm-collector/apm-collector-agentjvm/src/main/resources/META-INF/defines/group.define

-org.skywalking.apm.collector.agentjvm.AgentJVMModuleGroupDefine
\ No newline at end of file

apm-collector/apm-collector-agentjvm/src/main/resources/META-INF/defines/module.define

-org.skywalking.apm.collector.agentjvm.grpc.AgentJVMGRPCModuleDefine
\ No newline at end of file

apm-collector/apm-collector-agentregister/pom.xml

@@ -15,22 +15,22 @@
     <dependencies>
         <dependency>
             <groupId>org.skywalking</groupId>
-            <artifactId>apm-collector-cluster</artifactId>
+            <artifactId>apm-collector-stream</artifactId>
             <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.skywalking</groupId>
-            <artifactId>apm-collector-server</artifactId>
+            <artifactId>apm-collector-cluster</artifactId>
             <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.skywalking</groupId>
-            <artifactId>apm-network</artifactId>
+            <artifactId>apm-collector-server</artifactId>
             <version>${project.version}</version>
         </dependency>
         <dependency>
             <groupId>org.skywalking</groupId>
-            <artifactId>apm-collector-agentstream</artifactId>
+            <artifactId>apm-network</artifactId>
             <version>${project.version}</version>
         </dependency>
     </dependencies>

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/AgentRegisterCommonModuleInstaller.java

-package org.skywalking.apm.collector.agentregister;
-
-import java.util.List;
-import org.skywalking.apm.collector.core.framework.Context;
-import org.skywalking.apm.collector.core.module.MultipleCommonModuleInstaller;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterCommonModuleInstaller extends MultipleCommonModuleInstaller {
-
-    @Override public String groupName() {
-        return AgentRegisterModuleGroupDefine.GROUP_NAME;
-    }
-
-    @Override public Context moduleContext() {
-        return new AgentRegisterModuleContext(groupName());
-    }
-
-    @Override public List<String> dependenceModules() {
-        return null;
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/AgentRegisterModuleContext.java

-package org.skywalking.apm.collector.agentregister;
-
-import org.skywalking.apm.collector.core.framework.Context;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterModuleContext extends Context {
-    public AgentRegisterModuleContext(String groupName) {
-        super(groupName);
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/AgentRegisterModuleDefine.java

-package org.skywalking.apm.collector.agentregister;
-
-import org.skywalking.apm.collector.core.client.Client;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListenerDefine;
-import org.skywalking.apm.collector.core.module.ModuleDefine;
-
-/**
- * @author pengys5
- */
-public abstract class AgentRegisterModuleDefine extends ModuleDefine implements ClusterDataListenerDefine {
-
-    @Override protected void initializeOtherContext() {
-
-    }
-
-    @Override protected final Client createClient() {
-        throw new UnsupportedOperationException("");
-    }
-
-    @Override public final boolean defaultModule() {
-        return true;
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/AgentRegisterModuleException.java

-package org.skywalking.apm.collector.agentregister;
-
-import org.skywalking.apm.collector.core.module.ModuleException;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterModuleException extends ModuleException {
-    public AgentRegisterModuleException(String message) {
-        super(message);
-    }
-
-    public AgentRegisterModuleException(String message, Throwable cause) {
-        super(message, cause);
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/AgentRegisterModuleGroupDefine.java

-package org.skywalking.apm.collector.agentregister;
-
-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 AgentRegisterModuleGroupDefine implements ModuleGroupDefine {
-
-    public static final String GROUP_NAME = "agent_register";
-
-    private final AgentRegisterCommonModuleInstaller installer;
-
-    public AgentRegisterModuleGroupDefine() {
-        installer = new AgentRegisterCommonModuleInstaller();
-    }
-
-    @Override public String name() {
-        return GROUP_NAME;
-    }
-
-    @Override public Context groupContext() {
-        return new AgentRegisterModuleContext(GROUP_NAME);
-    }
-
-    @Override public ModuleInstaller moduleInstaller() {
-        return installer;
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/application/ApplicationIDService.java

 package org.skywalking.apm.collector.agentregister.application;
 
-import org.skywalking.apm.collector.agentstream.worker.cache.ApplicationCache;
-import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
-import org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterRemoteWorker;
+import org.skywalking.apm.collector.agentregister.worker.application.ApplicationRegisterRemoteWorker;
+import org.skywalking.apm.collector.agentregister.worker.cache.ApplicationCache;
 import org.skywalking.apm.collector.core.framework.CollectorContextHelper;
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
 import org.skywalking.apm.collector.stream.StreamModuleContext;
 import org.skywalking.apm.collector.stream.StreamModuleGroupDefine;
 import org.skywalking.apm.collector.stream.worker.WorkerInvokeException;

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/grpc/AgentRegisterGRPCConfig.java

-package org.skywalking.apm.collector.agentregister.grpc;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterGRPCConfig {
-    public static String HOST;
-    public static int PORT;
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/grpc/AgentRegisterGRPCConfigParser.java

-package org.skywalking.apm.collector.agentregister.grpc;
-
-import java.util.Map;
-import org.skywalking.apm.collector.core.config.ConfigParseException;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.util.ObjectUtils;
-import org.skywalking.apm.collector.core.util.StringUtils;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterGRPCConfigParser implements ModuleConfigParser {
-
-    private static final String HOST = "host";
-    private static final String PORT = "port";
-
-    @Override public void parse(Map config) throws ConfigParseException {
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(HOST))) {
-            AgentRegisterGRPCConfig.HOST = "localhost";
-        } else {
-            AgentRegisterGRPCConfig.HOST = (String)config.get(HOST);
-        }
-
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(PORT))) {
-            AgentRegisterGRPCConfig.PORT = 11800;
-        } else {
-            AgentRegisterGRPCConfig.PORT = (Integer)config.get(PORT);
-        }
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/grpc/AgentRegisterGRPCDataListener.java

-package org.skywalking.apm.collector.agentregister.grpc;
-
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleGroupDefine;
-import org.skywalking.apm.collector.cluster.ClusterModuleDefine;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterGRPCDataListener extends ClusterDataListener {
-
-    public static final String PATH = ClusterModuleDefine.BASE_CATALOG + "." + AgentRegisterModuleGroupDefine.GROUP_NAME + "." + AgentRegisterGRPCModuleDefine.MODULE_NAME;
-
-    @Override public String path() {
-        return PATH;
-    }
-
-    @Override public void serverJoinNotify(String serverAddress) {
-
-    }
-
-    @Override public void serverQuitNotify(String serverAddress) {
-
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/grpc/AgentRegisterGRPCModuleDefine.java

-package org.skywalking.apm.collector.agentregister.grpc;
-
-import java.util.LinkedList;
-import java.util.List;
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleDefine;
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleGroupDefine;
-import org.skywalking.apm.collector.agentregister.grpc.handler.ApplicationRegisterServiceHandler;
-import org.skywalking.apm.collector.agentregister.grpc.handler.InstanceDiscoveryServiceHandler;
-import org.skywalking.apm.collector.agentregister.grpc.handler.ServiceNameDiscoveryServiceHandler;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-import org.skywalking.apm.collector.core.framework.Handler;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-import org.skywalking.apm.collector.core.server.Server;
-import org.skywalking.apm.collector.server.grpc.GRPCServer;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterGRPCModuleDefine extends AgentRegisterModuleDefine {
-
-    public static final String MODULE_NAME = "grpc";
-
-    @Override protected String group() {
-        return AgentRegisterModuleGroupDefine.GROUP_NAME;
-    }
-
-    @Override public String name() {
-        return MODULE_NAME;
-    }
-
-    @Override protected ModuleConfigParser configParser() {
-        return new AgentRegisterGRPCConfigParser();
-    }
-
-    @Override protected Server server() {
-        return new GRPCServer(AgentRegisterGRPCConfig.HOST, AgentRegisterGRPCConfig.PORT);
-    }
-
-    @Override protected ModuleRegistration registration() {
-        return new AgentRegisterGRPCModuleRegistration();
-    }
-
-    @Override public ClusterDataListener listener() {
-        return new AgentRegisterGRPCDataListener();
-    }
-
-    @Override public List<Handler> handlerList() {
-        List<Handler> handlers = new LinkedList<>();
-        handlers.add(new ApplicationRegisterServiceHandler());
-        handlers.add(new InstanceDiscoveryServiceHandler());
-        handlers.add(new ServiceNameDiscoveryServiceHandler());
-        return handlers;
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/grpc/AgentRegisterGRPCModuleRegistration.java

-package org.skywalking.apm.collector.agentregister.grpc;
-
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterGRPCModuleRegistration extends ModuleRegistration {
-
-    @Override public Value buildValue() {
-        return new Value(AgentRegisterGRPCConfig.HOST, AgentRegisterGRPCConfig.PORT, null);
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/instance/InstanceIDService.java

 package org.skywalking.apm.collector.agentregister.instance;
 
-import org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterRemoteWorker;
-import org.skywalking.apm.collector.agentstream.worker.register.instance.dao.IInstanceDAO;
+import org.skywalking.apm.collector.agentregister.worker.instance.InstanceRegisterRemoteWorker;
+import org.skywalking.apm.collector.agentregister.worker.instance.dao.IInstanceDAO;
 import org.skywalking.apm.collector.core.framework.CollectorContextHelper;
 import org.skywalking.apm.collector.storage.dao.DAOContainer;
 import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/AgentRegisterJettyConfig.java

-package org.skywalking.apm.collector.agentregister.jetty;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterJettyConfig {
-    public static String HOST;
-    public static int PORT;
-    public static String CONTEXT_PATH;
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/AgentRegisterJettyConfigParser.java

-package org.skywalking.apm.collector.agentregister.jetty;
-
-import java.util.Map;
-import org.skywalking.apm.collector.core.config.ConfigParseException;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.util.ObjectUtils;
-import org.skywalking.apm.collector.core.util.StringUtils;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterJettyConfigParser implements ModuleConfigParser {
-
-    private static final String HOST = "host";
-    private static final String PORT = "port";
-    public static final String CONTEXT_PATH = "contextPath";
-
-    @Override public void parse(Map config) throws ConfigParseException {
-        AgentRegisterJettyConfig.CONTEXT_PATH = "/";
-
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(HOST))) {
-            AgentRegisterJettyConfig.HOST = "localhost";
-        } else {
-            AgentRegisterJettyConfig.HOST = (String)config.get(HOST);
-        }
-
-        if (ObjectUtils.isEmpty(config) || StringUtils.isEmpty(config.get(PORT))) {
-            AgentRegisterJettyConfig.PORT = 12800;
-        } else {
-            AgentRegisterJettyConfig.PORT = (Integer)config.get(PORT);
-        }
-        if (ObjectUtils.isNotEmpty(config) && StringUtils.isNotEmpty(config.get(CONTEXT_PATH))) {
-            AgentRegisterJettyConfig.CONTEXT_PATH = (String)config.get(CONTEXT_PATH);
-        }
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/AgentRegisterJettyDataListener.java

-package org.skywalking.apm.collector.agentregister.jetty;
-
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleGroupDefine;
-import org.skywalking.apm.collector.cluster.ClusterModuleDefine;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterJettyDataListener extends ClusterDataListener {
-
-    public static final String PATH = ClusterModuleDefine.BASE_CATALOG + "." + AgentRegisterModuleGroupDefine.GROUP_NAME + "." + AgentRegisterJettyModuleDefine.MODULE_NAME;
-
-    @Override public String path() {
-        return PATH;
-    }
-
-    @Override public void serverJoinNotify(String serverAddress) {
-
-    }
-
-    @Override public void serverQuitNotify(String serverAddress) {
-
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/AgentRegisterJettyModuleDefine.java

-package org.skywalking.apm.collector.agentregister.jetty;
-
-import java.util.LinkedList;
-import java.util.List;
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleDefine;
-import org.skywalking.apm.collector.agentregister.AgentRegisterModuleGroupDefine;
-import org.skywalking.apm.collector.agentregister.jetty.handler.ApplicationRegisterServletHandler;
-import org.skywalking.apm.collector.agentregister.jetty.handler.InstanceDiscoveryServletHandler;
-import org.skywalking.apm.collector.core.cluster.ClusterDataListener;
-import org.skywalking.apm.collector.core.framework.Handler;
-import org.skywalking.apm.collector.core.module.ModuleConfigParser;
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-import org.skywalking.apm.collector.core.server.Server;
-import org.skywalking.apm.collector.server.jetty.JettyServer;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterJettyModuleDefine extends AgentRegisterModuleDefine {
-
-    public static final String MODULE_NAME = "jetty";
-
-    @Override protected String group() {
-        return AgentRegisterModuleGroupDefine.GROUP_NAME;
-    }
-
-    @Override public String name() {
-        return MODULE_NAME;
-    }
-
-    @Override protected ModuleConfigParser configParser() {
-        return new AgentRegisterJettyConfigParser();
-    }
-
-    @Override protected Server server() {
-        return new JettyServer(AgentRegisterJettyConfig.HOST, AgentRegisterJettyConfig.PORT, AgentRegisterJettyConfig.CONTEXT_PATH);
-    }
-
-    @Override protected ModuleRegistration registration() {
-        return new AgentRegisterJettyModuleRegistration();
-    }
-
-    @Override public ClusterDataListener listener() {
-        return new AgentRegisterJettyDataListener();
-    }
-
-    @Override public List<Handler> handlerList() {
-        List<Handler> handlers = new LinkedList<>();
-        handlers.add(new ApplicationRegisterServletHandler());
-        handlers.add(new InstanceDiscoveryServletHandler());
-        return handlers;
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/AgentRegisterJettyModuleRegistration.java

-package org.skywalking.apm.collector.agentregister.jetty;
-
-import org.skywalking.apm.collector.core.module.ModuleRegistration;
-
-/**
- * @author pengys5
- */
-public class AgentRegisterJettyModuleRegistration extends ModuleRegistration {
-
-    @Override public Value buildValue() {
-        return new Value(AgentRegisterJettyConfig.HOST, AgentRegisterJettyConfig.PORT, AgentRegisterJettyConfig.CONTEXT_PATH);
-    }
-}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/jetty/handler/ServiceNameDiscoveryServiceHandler.java

+package org.skywalking.apm.collector.agentregister.jetty.handler;
+
+import com.google.gson.Gson;
+import com.google.gson.JsonArray;
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import java.io.IOException;
+import javax.servlet.http.HttpServletRequest;
+import org.skywalking.apm.collector.agentregister.servicename.ServiceNameService;
+import org.skywalking.apm.collector.server.jetty.ArgumentsParseException;
+import org.skywalking.apm.collector.server.jetty.JettyHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameDiscoveryServiceHandler extends JettyHandler {
+
+    private final Logger logger = LoggerFactory.getLogger(ServiceNameDiscoveryServiceHandler.class);
+
+    private ServiceNameService serviceNameService = new ServiceNameService();
+    private Gson gson = new Gson();
+
+    @Override public String pathSpec() {
+        return "/servicename/discovery";
+    }
+
+    private static final String APPLICATION_ID = "ai";
+    private static final String SERVICE_NAME = "sn";
+    private static final String SERVICE_ID = "si";
+    private static final String ELEMENT = "el";
+
+    @Override protected JsonElement doGet(HttpServletRequest req) throws ArgumentsParseException {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override protected JsonElement doPost(HttpServletRequest req) throws ArgumentsParseException {
+        JsonArray responseArray = new JsonArray();
+        try {
+            JsonArray services = gson.fromJson(req.getReader(), JsonArray.class);
+            for (JsonElement service : services) {
+                int applicationId = service.getAsJsonObject().get(APPLICATION_ID).getAsInt();
+                String serviceName = service.getAsJsonObject().get(SERVICE_NAME).getAsString();
+
+                int serviceId = serviceNameService.getOrCreate(applicationId, serviceName);
+                if (serviceId != 0) {
+                    JsonObject responseJson = new JsonObject();
+                    responseJson.addProperty(SERVICE_ID, serviceId);
+                    responseJson.add(ELEMENT, service);
+                    responseArray.add(responseJson);
+                }
+            }
+        } catch (IOException e) {
+            logger.error(e.getMessage(), e);
+        }
+        return responseArray;
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/servicename/ServiceNameService.java

 package org.skywalking.apm.collector.agentregister.servicename;
 
 import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
-import org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterRemoteWorker;
-import org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.IServiceNameDAO;
+import org.skywalking.apm.collector.agentregister.worker.servicename.ServiceNameRegisterRemoteWorker;
+import org.skywalking.apm.collector.agentregister.worker.servicename.dao.IServiceNameDAO;
 import org.skywalking.apm.collector.core.framework.CollectorContextHelper;
 import org.skywalking.apm.collector.storage.dao.DAOContainer;
 import org.skywalking.apm.collector.stream.StreamModuleContext;

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/IdAutoIncrement.java

+package org.skywalking.apm.collector.agentregister.worker;
+
+/**
+ * @author pengys5
+ */
+public enum IdAutoIncrement {
+    INSTANCE;
+
+    public int increment(int min, int max) {
+        int instanceId;
+        if (min == max) {
+            instanceId = -1;
+        } else if (min + max == 0) {
+            instanceId = max + 1;
+        } else if (min + max > 0) {
+            instanceId = min - 1;
+        } else if (max < 0) {
+            instanceId = 1;
+        } else {
+            instanceId = max + 1;
+        }
+        return instanceId;
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/ApplicationEsTableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.application;
+
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine;
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine;
+import org.skywalking.apm.collector.storage.define.register.ApplicationTable;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationEsTableDefine extends ElasticSearchTableDefine {
+
+    public ApplicationEsTableDefine() {
+        super(ApplicationTable.TABLE);
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+
+    @Override public int numberOfShards() {
+        return 2;
+    }
+
+    @Override public int numberOfReplicas() {
+        return 0;
+    }
+
+    @Override public void initialize() {
+        addColumn(new ElasticSearchColumnDefine(ApplicationTable.COLUMN_APPLICATION_CODE, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(ApplicationTable.COLUMN_APPLICATION_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/ApplicationH2TableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.application;
+
+import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine;
+import org.skywalking.apm.collector.storage.h2.define.H2TableDefine;
+import org.skywalking.apm.collector.storage.define.register.ApplicationTable;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationH2TableDefine extends H2TableDefine {
+
+    public ApplicationH2TableDefine() {
+        super(ApplicationTable.TABLE);
+    }
+
+    @Override public void initialize() {
+        addColumn(new H2ColumnDefine(ApplicationTable.COLUMN_APPLICATION_CODE, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(ApplicationTable.COLUMN_APPLICATION_ID, H2ColumnDefine.Type.Int.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/ApplicationRegisterRemoteWorker.java

+package org.skywalking.apm.collector.agentregister.worker.application;
+
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationRegisterRemoteWorker extends AbstractRemoteWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(ApplicationRegisterRemoteWorker.class);
+
+    protected ApplicationRegisterRemoteWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        ApplicationDataDefine.Application application = (ApplicationDataDefine.Application)message;
+        logger.debug("application code: {}", application.getApplicationCode());
+        getClusterContext().lookup(ApplicationRegisterSerialWorker.WorkerRole.INSTANCE).tell(application);
+    }
+
+    public static class Factory extends AbstractRemoteWorkerProvider<ApplicationRegisterRemoteWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public ApplicationRegisterRemoteWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new ApplicationRegisterRemoteWorker(role(), clusterContext);
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return ApplicationRegisterRemoteWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new ApplicationDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/ApplicationRegisterSerialWorker.java

+package org.skywalking.apm.collector.agentregister.worker.application;
+
+import org.skywalking.apm.collector.core.util.Const;
+import org.skywalking.apm.collector.agentregister.worker.IdAutoIncrement;
+import org.skywalking.apm.collector.agentregister.worker.application.dao.IApplicationDAO;
+import org.skywalking.apm.collector.storage.dao.DAOContainer;
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationRegisterSerialWorker extends AbstractLocalAsyncWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(ApplicationRegisterSerialWorker.class);
+
+    public ApplicationRegisterSerialWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+        super.preStart();
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        if (message instanceof ApplicationDataDefine.Application) {
+            ApplicationDataDefine.Application application = (ApplicationDataDefine.Application)message;
+            logger.debug("register application, application code: {}", application.getApplicationCode());
+
+            IApplicationDAO dao = (IApplicationDAO)DAOContainer.INSTANCE.get(IApplicationDAO.class.getName());
+            int applicationId = dao.getApplicationId(application.getApplicationCode());
+            if (applicationId == 0) {
+                int min = dao.getMinApplicationId();
+                if (min == 0) {
+                    ApplicationDataDefine.Application userApplication = new ApplicationDataDefine.Application(String.valueOf(Const.USER_ID), Const.USER_CODE, Const.USER_ID);
+                    dao.save(userApplication);
+
+                    application.setApplicationId(-1);
+                    application.setId("-1");
+                } else {
+                    int max = dao.getMaxApplicationId();
+                    applicationId = IdAutoIncrement.INSTANCE.increment(min, max);
+                    application.setApplicationId(applicationId);
+                    application.setId(String.valueOf(applicationId));
+                }
+                dao.save(application);
+            }
+        }
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationRegisterSerialWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public ApplicationRegisterSerialWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new ApplicationRegisterSerialWorker(role(), clusterContext);
+        }
+
+        @Override public int queueSize() {
+            return 256;
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return ApplicationRegisterSerialWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new ApplicationDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/dao/ApplicationEsDAO.java

+package org.skywalking.apm.collector.agentregister.worker.application.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.storage.define.register.ApplicationTable;
+import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationEsDAO extends EsDAO implements IApplicationDAO {
+
+    private final Logger logger = LoggerFactory.getLogger(ApplicationEsDAO.class);
+
+    @Override public int getApplicationId(String applicationCode) {
+        ElasticSearchClient client = getClient();
+
+        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(ApplicationTable.TABLE);
+        searchRequestBuilder.setTypes("type");
+        searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH);
+        searchRequestBuilder.setQuery(QueryBuilders.termQuery(ApplicationTable.COLUMN_APPLICATION_CODE, applicationCode));
+        searchRequestBuilder.setSize(1);
+
+        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
+        if (searchResponse.getHits().totalHits > 0) {
+            SearchHit searchHit = searchResponse.getHits().iterator().next();
+            int applicationId = (int)searchHit.getSource().get(ApplicationTable.COLUMN_APPLICATION_ID);
+            return applicationId;
+        }
+        return 0;
+    }
+
+    @Override public int getMaxApplicationId() {
+        return getMaxId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override public int getMinApplicationId() {
+        return getMinId(ApplicationTable.TABLE, ApplicationTable.COLUMN_APPLICATION_ID);
+    }
+
+    @Override public void save(ApplicationDataDefine.Application application) {
+        logger.debug("save application register info, application id: {}, application code: {}", application.getApplicationId(), application.getApplicationCode());
+        ElasticSearchClient client = getClient();
+        Map<String, Object> source = new HashMap();
+        source.put(ApplicationTable.COLUMN_APPLICATION_CODE, application.getApplicationCode());
+        source.put(ApplicationTable.COLUMN_APPLICATION_ID, application.getApplicationId());
+
+        IndexResponse response = client.prepareIndex(ApplicationTable.TABLE, application.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+        logger.debug("save application register info, application id: {}, application code: {}, status: {}", application.getApplicationId(), application.getApplicationCode(), response.status().name());
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/dao/ApplicationH2DAO.java

+package org.skywalking.apm.collector.agentregister.worker.application.dao;
+
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.client.h2.H2Client;
+import org.skywalking.apm.collector.storage.h2.dao.H2DAO;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationH2DAO extends H2DAO implements IApplicationDAO {
+
+    @Override public int getApplicationId(String applicationCode) {
+        H2Client client = getClient();
+        return 100;
+    }
+
+    @Override public int getMaxApplicationId() {
+        return 0;
+    }
+
+    @Override public int getMinApplicationId() {
+        return 0;
+    }
+
+    @Override public void save(ApplicationDataDefine.Application application) {
+
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/application/dao/IApplicationDAO.java

+package org.skywalking.apm.collector.agentregister.worker.application.dao;
+
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+
+/**
+ * @author pengys5
+ */
+public interface IApplicationDAO {
+    int getApplicationId(String applicationCode);
+
+    int getMaxApplicationId();
+
+    int getMinApplicationId();
+
+    void save(ApplicationDataDefine.Application application);
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/cache/ApplicationCache.java

+package org.skywalking.apm.collector.agentregister.worker.cache;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import org.skywalking.apm.collector.agentregister.worker.application.dao.IApplicationDAO;
+import org.skywalking.apm.collector.storage.dao.DAOContainer;
+
+/**
+ * @author pengys5
+ */
+public class ApplicationCache {
+
+    private static Cache<String, Integer> CACHE = CacheBuilder.newBuilder().initialCapacity(100).maximumSize(1000).build();
+
+    public static int get(String applicationCode) {
+        int applicationId = 0;
+        try {
+            applicationId = CACHE.get(applicationCode, () -> {
+                IApplicationDAO dao = (IApplicationDAO)DAOContainer.INSTANCE.get(IApplicationDAO.class.getName());
+                return dao.getApplicationId(applicationCode);
+            });
+        } catch (Throwable e) {
+            return applicationId;
+        }
+
+        if (applicationId == 0) {
+            IApplicationDAO dao = (IApplicationDAO)DAOContainer.INSTANCE.get(IApplicationDAO.class.getName());
+            applicationId = dao.getApplicationId(applicationCode);
+            if (applicationId != 0) {
+                CACHE.put(applicationCode, applicationId);
+            }
+        }
+        return applicationId;
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/InstanceEsTableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.instance;
+
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine;
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine;
+import org.skywalking.apm.collector.storage.define.register.InstanceTable;
+
+/**
+ * @author pengys5
+ */
+public class InstanceEsTableDefine extends ElasticSearchTableDefine {
+
+    public InstanceEsTableDefine() {
+        super(InstanceTable.TABLE);
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+
+    @Override public int numberOfShards() {
+        return 2;
+    }
+
+    @Override public int numberOfReplicas() {
+        return 0;
+    }
+
+    @Override public void initialize() {
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_APPLICATION_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_AGENT_UUID, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_REGISTER_TIME, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_HEARTBEAT_TIME, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceTable.COLUMN_OS_INFO, ElasticSearchColumnDefine.Type.Keyword.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/InstanceH2TableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.instance;
+
+import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine;
+import org.skywalking.apm.collector.storage.h2.define.H2TableDefine;
+import org.skywalking.apm.collector.storage.define.register.InstanceTable;
+
+/**
+ * @author pengys5
+ */
+public class InstanceH2TableDefine extends H2TableDefine {
+
+    public InstanceH2TableDefine() {
+        super(InstanceTable.TABLE);
+    }
+
+    @Override public void initialize() {
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_APPLICATION_ID, H2ColumnDefine.Type.Int.name()));
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_AGENT_UUID, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_REGISTER_TIME, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_INSTANCE_ID, H2ColumnDefine.Type.Int.name()));
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_HEARTBEAT_TIME, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceTable.COLUMN_OS_INFO, H2ColumnDefine.Type.Varchar.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/InstanceRegisterRemoteWorker.java

+package org.skywalking.apm.collector.agentregister.worker.instance;
+
+import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class InstanceRegisterRemoteWorker extends AbstractRemoteWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceRegisterRemoteWorker.class);
+
+    protected InstanceRegisterRemoteWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        InstanceDataDefine.Instance instance = (InstanceDataDefine.Instance)message;
+        logger.debug("application id: {}, agentUUID: {}, register time: {}", instance.getApplicationId(), instance.getAgentUUID(), instance.getRegisterTime());
+        getClusterContext().lookup(InstanceRegisterSerialWorker.WorkerRole.INSTANCE).tell(instance);
+    }
+
+    public static class Factory extends AbstractRemoteWorkerProvider<InstanceRegisterRemoteWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public InstanceRegisterRemoteWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new InstanceRegisterRemoteWorker(role(), clusterContext);
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return InstanceRegisterRemoteWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new InstanceDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/InstanceRegisterSerialWorker.java

+package org.skywalking.apm.collector.agentregister.worker.instance;
+
+import org.skywalking.apm.collector.agentregister.worker.instance.dao.IInstanceDAO;
+import org.skywalking.apm.collector.storage.dao.DAOContainer;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.storage.define.register.ApplicationDataDefine;
+import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class InstanceRegisterSerialWorker extends AbstractLocalAsyncWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceRegisterSerialWorker.class);
+
+    public InstanceRegisterSerialWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+        super.preStart();
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        if (message instanceof InstanceDataDefine.Instance) {
+            InstanceDataDefine.Instance instance = (InstanceDataDefine.Instance)message;
+            logger.debug("register instance, application id: {}, agentUUID: {}", instance.getApplicationId(), instance.getAgentUUID());
+
+            IInstanceDAO dao = (IInstanceDAO)DAOContainer.INSTANCE.get(IInstanceDAO.class.getName());
+            int instanceId = dao.getInstanceId(instance.getApplicationId(), instance.getAgentUUID());
+            if (instanceId == 0) {
+//                int min = dao.getMinInstanceId();
+//                if (min == 0) {
+//                    instance.setId("1");
+//                    instance.setInstanceId(1);
+//                } else {
+//                    int max = dao.getMaxInstanceId();
+//                    instanceId = IdAutoIncrement.INSTANCE.increment(min, max);
+//                    instance.setId(String.valueOf(instanceId));
+//                    instance.setInstanceId(instanceId);
+//                }
+                int max = dao.getMaxInstanceId();
+                if (max == 0) {
+                    instance.setId("1");
+                    instance.setInstanceId(1);
+                } else {
+                    instance.setId(String.valueOf(max + 1));
+                    instance.setInstanceId(max + 1);
+                }
+
+                dao.save(instance);
+            }
+        }
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<InstanceRegisterSerialWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public InstanceRegisterSerialWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new InstanceRegisterSerialWorker(role(), clusterContext);
+        }
+
+        @Override public int queueSize() {
+            return 256;
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return InstanceRegisterSerialWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new ApplicationDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/dao/IInstanceDAO.java

+package org.skywalking.apm.collector.agentregister.worker.instance.dao;
+
+import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;
+
+/**
+ * @author pengys5
+ */
+public interface IInstanceDAO {
+    int getInstanceId(int applicationId, String agentUUID);
+
+    int getMaxInstanceId();
+
+    int getMinInstanceId();
+
+    void save(InstanceDataDefine.Instance instance);
+
+    void updateHeartbeatTime(int instanceId, long heartbeatTime);
+
+    int getApplicationId(int applicationInstanceId);
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/dao/InstanceEsDAO.java

+package org.skywalking.apm.collector.agentregister.worker.instance.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;
+import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO;
+import org.skywalking.apm.collector.storage.define.register.InstanceTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class InstanceEsDAO extends EsDAO implements IInstanceDAO {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceEsDAO.class);
+
+    @Override public int getInstanceId(int applicationId, String agentUUID) {
+        ElasticSearchClient client = getClient();
+
+        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(InstanceTable.TABLE);
+        searchRequestBuilder.setTypes("type");
+        searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH);
+        BoolQueryBuilder builder = QueryBuilders.boolQuery();
+        builder.must().add(QueryBuilders.termQuery(InstanceTable.COLUMN_APPLICATION_ID, applicationId));
+        builder.must().add(QueryBuilders.termQuery(InstanceTable.COLUMN_AGENT_UUID, agentUUID));
+        searchRequestBuilder.setQuery(builder);
+        searchRequestBuilder.setSize(1);
+
+        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
+        if (searchResponse.getHits().totalHits > 0) {
+            SearchHit searchHit = searchResponse.getHits().iterator().next();
+            return (int)searchHit.getSource().get(InstanceTable.COLUMN_INSTANCE_ID);
+        }
+        return 0;
+    }
+
+    @Override public int getMaxInstanceId() {
+        return getMaxId(InstanceTable.TABLE, InstanceTable.COLUMN_INSTANCE_ID);
+    }
+
+    @Override public int getMinInstanceId() {
+        return getMinId(InstanceTable.TABLE, InstanceTable.COLUMN_INSTANCE_ID);
+    }
+
+    @Override public void save(InstanceDataDefine.Instance instance) {
+        logger.debug("save instance register info, application id: {}, agentUUID: {}", instance.getApplicationId(), instance.getAgentUUID());
+        ElasticSearchClient client = getClient();
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceTable.COLUMN_INSTANCE_ID, instance.getInstanceId());
+        source.put(InstanceTable.COLUMN_APPLICATION_ID, instance.getApplicationId());
+        source.put(InstanceTable.COLUMN_AGENT_UUID, instance.getAgentUUID());
+        source.put(InstanceTable.COLUMN_REGISTER_TIME, instance.getRegisterTime());
+        source.put(InstanceTable.COLUMN_HEARTBEAT_TIME, instance.getHeartBeatTime());
+        source.put(InstanceTable.COLUMN_OS_INFO, instance.getOsInfo());
+
+        IndexResponse response = client.prepareIndex(InstanceTable.TABLE, instance.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+        logger.debug("save instance register info, application id: {}, agentUUID: {}, status: {}", instance.getApplicationId(), instance.getAgentUUID(), response.status().name());
+    }
+
+    @Override public void updateHeartbeatTime(int instanceId, long heartbeatTime) {
+        ElasticSearchClient client = getClient();
+        UpdateRequest updateRequest = new UpdateRequest();
+        updateRequest.index(InstanceTable.TABLE);
+        updateRequest.type("type");
+        updateRequest.id(String.valueOf(instanceId));
+        updateRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
+
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceTable.COLUMN_HEARTBEAT_TIME, heartbeatTime);
+
+        updateRequest.doc(source);
+        client.update(updateRequest);
+    }
+
+    @Override public int getApplicationId(int applicationInstanceId) {
+        GetResponse response = getClient().prepareGet(InstanceTable.TABLE, String.valueOf(applicationInstanceId)).get();
+        if (response.isExists()) {
+            return (int)response.getSource().get(InstanceTable.COLUMN_APPLICATION_ID);
+        } else {
+            return 0;
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/instance/dao/InstanceH2DAO.java

+package org.skywalking.apm.collector.agentregister.worker.instance.dao;
+
+import org.skywalking.apm.collector.storage.define.register.InstanceDataDefine;
+import org.skywalking.apm.collector.storage.h2.dao.H2DAO;
+
+/**
+ * @author pengys5
+ */
+public class InstanceH2DAO extends H2DAO implements IInstanceDAO {
+    @Override public int getInstanceId(int applicationId, String agentUUID) {
+        return 0;
+    }
+
+    @Override public int getMaxInstanceId() {
+        return 0;
+    }
+
+    @Override public int getMinInstanceId() {
+        return 0;
+    }
+
+    @Override public void save(InstanceDataDefine.Instance instance) {
+
+    }
+
+    @Override public void updateHeartbeatTime(int instanceId, long heartbeatTime) {
+
+    }
+
+    @Override public int getApplicationId(int applicationInstanceId) {
+        return 0;
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/ServiceNameEsTableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.servicename;
+
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchColumnDefine;
+import org.skywalking.apm.collector.storage.elasticsearch.define.ElasticSearchTableDefine;
+import org.skywalking.apm.collector.storage.define.register.ServiceNameTable;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameEsTableDefine extends ElasticSearchTableDefine {
+
+    public ServiceNameEsTableDefine() {
+        super(ServiceNameTable.TABLE);
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+
+    @Override public int numberOfShards() {
+        return 2;
+    }
+
+    @Override public int numberOfReplicas() {
+        return 0;
+    }
+
+    @Override public void initialize() {
+        addColumn(new ElasticSearchColumnDefine(ServiceNameTable.COLUMN_APPLICATION_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(ServiceNameTable.COLUMN_SERVICE_NAME, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(ServiceNameTable.COLUMN_SERVICE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/ServiceNameH2TableDefine.java

+package org.skywalking.apm.collector.agentregister.worker.servicename;
+
+import org.skywalking.apm.collector.storage.h2.define.H2ColumnDefine;
+import org.skywalking.apm.collector.storage.h2.define.H2TableDefine;
+import org.skywalking.apm.collector.storage.define.register.ServiceNameTable;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameH2TableDefine extends H2TableDefine {
+
+    public ServiceNameH2TableDefine() {
+        super(ServiceNameTable.TABLE);
+    }
+
+    @Override public void initialize() {
+        addColumn(new H2ColumnDefine(ServiceNameTable.COLUMN_APPLICATION_ID, H2ColumnDefine.Type.Int.name()));
+        addColumn(new H2ColumnDefine(ServiceNameTable.COLUMN_SERVICE_NAME, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(ServiceNameTable.COLUMN_SERVICE_ID, H2ColumnDefine.Type.Int.name()));
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/ServiceNameRegisterRemoteWorker.java

+package org.skywalking.apm.collector.agentregister.worker.servicename;
+
+import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractRemoteWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameRegisterRemoteWorker extends AbstractRemoteWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(ServiceNameRegisterRemoteWorker.class);
+
+    protected ServiceNameRegisterRemoteWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        ServiceNameDataDefine.ServiceName serviceName = (ServiceNameDataDefine.ServiceName)message;
+        logger.debug("service name: {}", serviceName.getServiceName());
+        getClusterContext().lookup(ServiceNameRegisterSerialWorker.WorkerRole.INSTANCE).tell(serviceName);
+    }
+
+    public static class Factory extends AbstractRemoteWorkerProvider<ServiceNameRegisterRemoteWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public ServiceNameRegisterRemoteWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new ServiceNameRegisterRemoteWorker(role(), clusterContext);
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return ServiceNameRegisterRemoteWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new ServiceNameDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/ServiceNameRegisterSerialWorker.java

+package org.skywalking.apm.collector.agentregister.worker.servicename;
+
+import org.skywalking.apm.collector.agentregister.worker.IdAutoIncrement;
+import org.skywalking.apm.collector.agentregister.worker.servicename.dao.IServiceNameDAO;
+import org.skywalking.apm.collector.core.util.Const;
+import org.skywalking.apm.collector.storage.dao.DAOContainer;
+import org.skywalking.apm.collector.storage.define.DataDefine;
+import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorker;
+import org.skywalking.apm.collector.stream.worker.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.ClusterWorkerContext;
+import org.skywalking.apm.collector.stream.worker.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.Role;
+import org.skywalking.apm.collector.stream.worker.WorkerException;
+import org.skywalking.apm.collector.stream.worker.selector.ForeverFirstSelector;
+import org.skywalking.apm.collector.stream.worker.selector.WorkerSelector;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameRegisterSerialWorker extends AbstractLocalAsyncWorker {
+
+    private final Logger logger = LoggerFactory.getLogger(ServiceNameRegisterSerialWorker.class);
+
+    public ServiceNameRegisterSerialWorker(Role role, ClusterWorkerContext clusterContext) {
+        super(role, clusterContext);
+    }
+
+    @Override public void preStart() throws ProviderNotFoundException {
+        super.preStart();
+    }
+
+    @Override protected void onWork(Object message) throws WorkerException {
+        if (message instanceof ServiceNameDataDefine.ServiceName) {
+            ServiceNameDataDefine.ServiceName serviceName = (ServiceNameDataDefine.ServiceName)message;
+            logger.debug("register service name: {}, application id: {}", serviceName.getServiceName(), serviceName.getApplicationId());
+
+            IServiceNameDAO dao = (IServiceNameDAO)DAOContainer.INSTANCE.get(IServiceNameDAO.class.getName());
+            int serviceId = dao.getServiceId(serviceName.getApplicationId(), serviceName.getServiceName());
+
+            if (serviceId == 0) {
+                int min = dao.getMinServiceId();
+                if (min == 0) {
+                    ServiceNameDataDefine.ServiceName noneServiceName = new ServiceNameDataDefine.ServiceName("1", Const.NONE_SERVICE_NAME, 0, Const.NONE_SERVICE_ID);
+                    dao.save(noneServiceName);
+
+                    serviceName.setServiceId(-1);
+                    serviceName.setId("-1");
+                } else {
+                    int max = dao.getMaxServiceId();
+                    serviceId = IdAutoIncrement.INSTANCE.increment(min, max);
+                    serviceName.setId(String.valueOf(serviceId));
+                    serviceName.setServiceId(serviceId);
+                }
+                dao.save(serviceName);
+            }
+        }
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceNameRegisterSerialWorker> {
+        @Override
+        public Role role() {
+            return WorkerRole.INSTANCE;
+        }
+
+        @Override
+        public ServiceNameRegisterSerialWorker workerInstance(ClusterWorkerContext clusterContext) {
+            return new ServiceNameRegisterSerialWorker(role(), clusterContext);
+        }
+
+        @Override public int queueSize() {
+            return 256;
+        }
+    }
+
+    public enum WorkerRole implements Role {
+        INSTANCE;
+
+        @Override
+        public String roleName() {
+            return ServiceNameRegisterSerialWorker.class.getSimpleName();
+        }
+
+        @Override
+        public WorkerSelector workerSelector() {
+            return new ForeverFirstSelector();
+        }
+
+        @Override public DataDefine dataDefine() {
+            return new ServiceNameDataDefine();
+        }
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/dao/IServiceNameDAO.java

+package org.skywalking.apm.collector.agentregister.worker.servicename.dao;
+
+import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
+
+/**
+ * @author pengys5
+ */
+public interface IServiceNameDAO {
+    int getServiceId(int applicationId, String serviceName);
+
+    String getServiceName(int serviceId);
+
+    int getMaxServiceId();
+
+    int getMinServiceId();
+
+    void save(ServiceNameDataDefine.ServiceName serviceName);
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/dao/ServiceNameEsDAO.java

+package org.skywalking.apm.collector.agentregister.worker.servicename.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.search.SearchType;
+import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.search.SearchHit;
+import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.skywalking.apm.collector.core.util.Const;
+import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
+import org.skywalking.apm.collector.storage.define.register.ServiceNameTable;
+import org.skywalking.apm.collector.storage.elasticsearch.dao.EsDAO;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameEsDAO extends EsDAO implements IServiceNameDAO {
+
+    private final Logger logger = LoggerFactory.getLogger(ServiceNameEsDAO.class);
+
+    @Override public int getServiceId(int applicationId, String serviceName) {
+        ElasticSearchClient client = getClient();
+
+        SearchRequestBuilder searchRequestBuilder = client.prepareSearch(ServiceNameTable.TABLE);
+        searchRequestBuilder.setTypes(ServiceNameTable.TABLE_TYPE);
+        searchRequestBuilder.setSearchType(SearchType.QUERY_THEN_FETCH);
+        BoolQueryBuilder builder = QueryBuilders.boolQuery();
+        builder.must().add(QueryBuilders.termQuery(ServiceNameTable.COLUMN_APPLICATION_ID, applicationId));
+        builder.must().add(QueryBuilders.termQuery(ServiceNameTable.COLUMN_SERVICE_NAME, serviceName));
+        searchRequestBuilder.setQuery(builder);
+        searchRequestBuilder.setSize(1);
+
+        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
+        if (searchResponse.getHits().totalHits > 0) {
+            SearchHit searchHit = searchResponse.getHits().iterator().next();
+            int serviceId = (int)searchHit.getSource().get(ServiceNameTable.COLUMN_SERVICE_ID);
+            return serviceId;
+        }
+        return 0;
+    }
+
+    @Override public int getMaxServiceId() {
+        return getMaxId(ServiceNameTable.TABLE, ServiceNameTable.COLUMN_SERVICE_ID);
+    }
+
+    @Override public int getMinServiceId() {
+        return getMinId(ServiceNameTable.TABLE, ServiceNameTable.COLUMN_SERVICE_ID);
+    }
+
+    @Override public String getServiceName(int serviceId) {
+        GetResponse response = getClient().prepareGet(ServiceNameTable.TABLE, String.valueOf(serviceId)).get();
+        if (response.isExists()) {
+            return (String)response.getSource().get(ServiceNameTable.COLUMN_SERVICE_NAME);
+        } else {
+            return Const.EMPTY_STRING;
+        }
+    }
+
+    @Override public void save(ServiceNameDataDefine.ServiceName serviceName) {
+        logger.debug("save service name register info, application id: {}, service name: {}", serviceName.getApplicationId(), serviceName.getServiceName());
+        ElasticSearchClient client = getClient();
+        Map<String, Object> source = new HashMap();
+        source.put(ServiceNameTable.COLUMN_SERVICE_ID, serviceName.getServiceId());
+        source.put(ServiceNameTable.COLUMN_APPLICATION_ID, serviceName.getApplicationId());
+        source.put(ServiceNameTable.COLUMN_SERVICE_NAME, serviceName.getServiceName());
+
+        IndexResponse response = client.prepareIndex(ServiceNameTable.TABLE, serviceName.getId()).setSource(source).setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE).get();
+        logger.debug("save service name register info, application id: {}, service name: {}, status: {}", serviceName.getApplicationId(), serviceName.getServiceName(), response.status().name());
+    }
+}

apm-collector/apm-collector-agentregister/src/main/java/org/skywalking/apm/collector/agentregister/worker/servicename/dao/ServiceNameH2DAO.java

+package org.skywalking.apm.collector.agentregister.worker.servicename.dao;
+
+import org.skywalking.apm.collector.storage.define.register.ServiceNameDataDefine;
+import org.skywalking.apm.collector.storage.h2.dao.H2DAO;
+
+/**
+ * @author pengys5
+ */
+public class ServiceNameH2DAO extends H2DAO implements IServiceNameDAO {
+
+    @Override public int getServiceId(int applicationId, String serviceName) {
+        return 0;
+    }
+
+    @Override public int getMaxServiceId() {
+        return 0;
+    }
+
+    @Override public int getMinServiceId() {
+        return 0;
+    }
+
+    @Override public String getServiceName(int serviceId) {
+        return null;
+    }
+
+    @Override public void save(ServiceNameDataDefine.ServiceName serviceName) {
+
+    }
+}

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/es_dao.define

+org.skywalking.apm.collector.agentregister.worker.application.dao.ApplicationEsDAO
+org.skywalking.apm.collector.agentregister.worker.instance.dao.InstanceEsDAO
+org.skywalking.apm.collector.agentregister.worker.servicename.dao.ServiceNameEsDAO
\ No newline at end of file

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/group.define

-org.skywalking.apm.collector.agentregister.AgentRegisterModuleGroupDefine
\ No newline at end of file

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/h2_dao.define

+org.skywalking.apm.collector.agentregister.worker.application.dao.ApplicationH2DAO
+org.skywalking.apm.collector.agentregister.worker.instance.dao.InstanceH2DAO
+org.skywalking.apm.collector.agentregister.worker.servicename.dao.ServiceNameH2DAO
\ No newline at end of file

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/local_worker_provider.define

+org.skywalking.apm.collector.agentregister.worker.application.ApplicationRegisterSerialWorker$Factory
+org.skywalking.apm.collector.agentregister.worker.instance.InstanceRegisterSerialWorker$Factory
+org.skywalking.apm.collector.agentregister.worker.servicename.ServiceNameRegisterSerialWorker$Factory
\ No newline at end of file

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/module.define

-org.skywalking.apm.collector.agentregister.grpc.AgentRegisterGRPCModuleDefine
-org.skywalking.apm.collector.agentregister.jetty.AgentRegisterJettyModuleDefine
\ No newline at end of file

apm-collector/apm-collector-agentregister/src/main/resources/META-INF/defines/storage.define

+org.skywalking.apm.collector.agentregister.worker.application.ApplicationEsTableDefine
+org.skywalking.apm.collector.agentregister.worker.application.ApplicationH2TableDefine
+
+org.skywalking.apm.collector.agentregister.worker.instance.InstanceEsTableDefine
+org.skywalking.apm.collector.agentregister.worker.instance.InstanceH2TableDefine
+
+org.skywalking.apm.collector.agentregister.worker.servicename.ServiceNameEsTableDefine
+org.skywalking.apm.collector.agentregister.worker.servicename.ServiceNameH2TableDefine
\ No newline at end of file

apm-collector/apm-collector-agentstream/pom.xml

@@ -33,5 +33,15 @@
             <artifactId>apm-collector-storage</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.skywalking</groupId>
+            <artifactId>apm-collector-agentjvm</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.skywalking</groupId>
+            <artifactId>apm-collector-agentregister</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>
\ No newline at end of file

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleGroupDefine.java

@@ -10,10 +10,10 @@ import org.skywalking.apm.collector.core.module.ModuleInstaller;
 public class AgentStreamModuleGroupDefine implements ModuleGroupDefine {
 
     public static final String GROUP_NAME = "agent_stream";
-    private final AgentStreamCommonModuleInstaller installer;
+    private final AgentStreamModuleInstaller installer;
 
     public AgentStreamModuleGroupDefine() {
-        installer = new AgentStreamCommonModuleInstaller();
+        installer = new AgentStreamModuleInstaller();
     }
 
     @Override public String name() {

apm-collector/apm-collector-agentjvm/src/main/java/org/skywalking/apm/collector/agentjvm/AgentJVMCommonModuleInstaller.java → apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/AgentStreamModuleInstaller.java

-package org.skywalking.apm.collector.agentjvm;
+package org.skywalking.apm.collector.agentstream;
 
 import java.util.List;
+import org.skywalking.apm.collector.agentstream.worker.storage.PersistenceTimer;
+import org.skywalking.apm.collector.core.client.ClientException;
+import org.skywalking.apm.collector.core.config.ConfigException;
 import org.skywalking.apm.collector.core.framework.Context;
+import org.skywalking.apm.collector.core.framework.DefineException;
 import org.skywalking.apm.collector.core.module.MultipleCommonModuleInstaller;
+import org.skywalking.apm.collector.core.server.ServerException;
 
 /**
  * @author pengys5
  */
-public class AgentJVMCommonModuleInstaller extends MultipleCommonModuleInstaller {
+public class AgentStreamModuleInstaller extends MultipleCommonModuleInstaller {
 
     @Override public String groupName() {
-        return AgentJVMModuleGroupDefine.GROUP_NAME;
+        return AgentStreamModuleGroupDefine.GROUP_NAME;
     }
 
     @Override public Context moduleContext() {
-        return new AgentJVMModuleContext(groupName());
+        return new AgentStreamModuleContext(groupName());
     }
 
     @Override public List<String> dependenceModules() {
         return null;
     }
+
+    @Override public void install() throws DefineException, ConfigException, ServerException, ClientException {
+        super.install();
+        new PersistenceTimer().start();
+    }
 }

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/grpc/AgentStreamGRPCModuleDefine.java

@@ -2,6 +2,10 @@ package org.skywalking.apm.collector.agentstream.grpc;
 
 import java.util.LinkedList;
 import java.util.List;
+import org.skywalking.apm.collector.agentjvm.grpc.handler.JVMMetricsServiceHandler;
+import org.skywalking.apm.collector.agentregister.grpc.handler.ApplicationRegisterServiceHandler;
+import org.skywalking.apm.collector.agentregister.grpc.handler.InstanceDiscoveryServiceHandler;
+import org.skywalking.apm.collector.agentregister.grpc.handler.ServiceNameDiscoveryServiceHandler;
 import org.skywalking.apm.collector.agentstream.AgentStreamModuleDefine;
 import org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine;
 import org.skywalking.apm.collector.agentstream.grpc.handler.TraceSegmentServiceHandler;
@@ -46,6 +50,10 @@ public class AgentStreamGRPCModuleDefine extends AgentStreamModuleDefine {
     @Override public List<Handler> handlerList() {
         List<Handler> handlers = new LinkedList<>();
         handlers.add(new TraceSegmentServiceHandler());
+        handlers.add(new ApplicationRegisterServiceHandler());
+        handlers.add(new InstanceDiscoveryServiceHandler());
+        handlers.add(new ServiceNameDiscoveryServiceHandler());
+        handlers.add(new JVMMetricsServiceHandler());
         return handlers;
     }
 }

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/jetty/AgentStreamJettyModuleDefine.java

@@ -2,6 +2,9 @@ package org.skywalking.apm.collector.agentstream.jetty;
 
 import java.util.LinkedList;
 import java.util.List;
+import org.skywalking.apm.collector.agentregister.jetty.handler.ApplicationRegisterServletHandler;
+import org.skywalking.apm.collector.agentregister.jetty.handler.InstanceDiscoveryServletHandler;
+import org.skywalking.apm.collector.agentregister.jetty.handler.ServiceNameDiscoveryServiceHandler;
 import org.skywalking.apm.collector.agentstream.AgentStreamModuleDefine;
 import org.skywalking.apm.collector.agentstream.AgentStreamModuleGroupDefine;
 import org.skywalking.apm.collector.agentstream.jetty.handler.TraceSegmentServletHandler;
@@ -46,6 +49,9 @@ public class AgentStreamJettyModuleDefine extends AgentStreamModuleDefine {
     @Override public List<Handler> handlerList() {
         List<Handler> handlers = new LinkedList<>();
         handlers.add(new TraceSegmentServletHandler());
+        handlers.add(new ApplicationRegisterServletHandler());
+        handlers.add(new InstanceDiscoveryServletHandler());
+        handlers.add(new ServiceNameDiscoveryServiceHandler());
         return handlers;
     }
 }

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/cache/InstanceCache.java

@@ -2,7 +2,7 @@ package org.skywalking.apm.collector.agentstream.worker.cache;
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
-import org.skywalking.apm.collector.agentstream.worker.register.instance.dao.IInstanceDAO;
+import org.skywalking.apm.collector.agentregister.worker.instance.dao.IInstanceDAO;
 import org.skywalking.apm.collector.storage.dao.DAOContainer;
 
 /**

apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/cache/ServiceCache.java

@@ -2,7 +2,7 @@ package org.skywalking.apm.collector.agentstream.worker.cache;
 
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
-import org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.IServiceNameDAO;
+import org.skywalking.apm.collector.agentregister.worker.servicename.dao.IServiceNameDAO;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.storage.dao.DAOContainer;
 

apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/es_dao.define

-org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationEsDAO
-org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceEsDAO
-org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.ServiceNameEsDAO
 org.skywalking.apm.collector.agentstream.worker.node.component.dao.NodeComponentEsDAO
 org.skywalking.apm.collector.agentstream.worker.node.mapping.dao.NodeMappingEsDAO
 org.skywalking.apm.collector.agentstream.worker.noderef.dao.NodeReferenceEsDAO

apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/h2_dao.define

-org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationH2DAO
-org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceH2DAO
-org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.ServiceNameH2DAO
 org.skywalking.apm.collector.agentstream.worker.node.component.dao.NodeComponentH2DAO
 org.skywalking.apm.collector.agentstream.worker.node.mapping.dao.NodeMappingH2DAO
 org.skywalking.apm.collector.agentstream.worker.noderef.dao.NodeReferenceH2DAO

apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/local_worker_provider.define

@@ -17,8 +17,4 @@ org.skywalking.apm.collector.agentstream.worker.segment.origin.SegmentPersistenc
 org.skywalking.apm.collector.agentstream.worker.segment.cost.SegmentCostPersistenceWorker$Factory
 org.skywalking.apm.collector.agentstream.worker.global.GlobalTracePersistenceWorker$Factory
 
-org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterSerialWorker$Factory
-org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterSerialWorker$Factory
-org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterSerialWorker$Factory
-
 org.skywalking.apm.collector.agentstream.worker.instance.performance.InstPerformancePersistenceWorker$Factory
\ No newline at end of file

apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/remote_worker_provider.define

-org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationRegisterRemoteWorker$Factory
-org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceRegisterRemoteWorker$Factory
-org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameRegisterRemoteWorker$Factory
+org.skywalking.apm.collector.agentregister.worker.application.ApplicationRegisterRemoteWorker$Factory
+org.skywalking.apm.collector.agentregister.worker.instance.InstanceRegisterRemoteWorker$Factory
+org.skywalking.apm.collector.agentregister.worker.servicename.ServiceNameRegisterRemoteWorker$Factory
 
 org.skywalking.apm.collector.agentstream.worker.node.component.NodeComponentRemoteWorker$Factory
 org.skywalking.apm.collector.agentstream.worker.node.mapping.NodeMappingRemoteWorker$Factory

apm-collector/apm-collector-agentstream/src/main/resources/META-INF/defines/storage.define

@@ -7,15 +7,6 @@ org.skywalking.apm.collector.agentstream.worker.node.mapping.define.NodeMappingH
 org.skywalking.apm.collector.agentstream.worker.noderef.define.NodeReferenceEsTableDefine
 org.skywalking.apm.collector.agentstream.worker.noderef.define.NodeReferenceH2TableDefine
 
-org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationEsTableDefine
-org.skywalking.apm.collector.agentstream.worker.register.application.ApplicationH2TableDefine
-
-org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceEsTableDefine
-org.skywalking.apm.collector.agentstream.worker.register.instance.InstanceH2TableDefine
-
-org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameEsTableDefine
-org.skywalking.apm.collector.agentstream.worker.register.servicename.ServiceNameH2TableDefine
-
 org.skywalking.apm.collector.agentstream.worker.segment.origin.define.SegmentEsTableDefine
 org.skywalking.apm.collector.agentstream.worker.segment.origin.define.SegmentH2TableDefine
 

apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/SegmentPost.java

@@ -5,9 +5,9 @@ import com.google.gson.JsonElement;
 import com.google.gson.JsonObject;
 import java.io.IOException;
 import org.skywalking.apm.collector.agentstream.HttpClientTools;
-import org.skywalking.apm.collector.agentstream.worker.register.application.dao.ApplicationEsDAO;
-import org.skywalking.apm.collector.agentstream.worker.register.instance.dao.InstanceEsDAO;
-import org.skywalking.apm.collector.agentstream.worker.register.servicename.dao.ServiceNameEsDAO;
+import org.skywalking.apm.collector.agentregister.worker.application.dao.ApplicationEsDAO;
+import org.skywalking.apm.collector.agentregister.worker.instance.dao.InstanceEsDAO;
+import org.skywalking.apm.collector.agentregister.worker.servicename.dao.ServiceNameEsDAO;
 import org.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.skywalking.apm.collector.core.CollectorException;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;

apm-collector/pom.xml

@@ -17,6 +17,7 @@
         <module>apm-collector-agentregister</module>
         <module>apm-collector-agentjvm</module>
         <module>apm-collector-remote</module>
+        <module>apm-collector-agent</module>
     </modules>
     <parent>
         <artifactId>apm</artifactId>