Merge branch 'feature/collector-modelization' into feature/agent-memory-control

README.md

@@ -21,8 +21,8 @@ Sky Walking | [中文](README_ZH.md)
 * The UI released on [skywalking-ui](https://github.com/OpenSkywalking/sky-walking-ui)
 
 # Architecture
-* Architecture graph for 3.2+
-<img src="https://skywalkingtest.github.io/page-resources/3.x-architecture.jpg"/>
+* Architecture graph for 3.2.5+
+<img src="https://skywalkingtest.github.io/page-resources/3.2.5%2b_architecture.jpg"/>
 
 # Document
 [![EN doc](https://img.shields.io/badge/document-English-blue.svg)](docs/README.md) [![cn doc](https://img.shields.io/badge/document-中文-blue.svg)](docs/README_ZH.md)

README_ZH.md

@@ -25,7 +25,7 @@ Sky Walking | [English](README.md)
 
 # Architecture
 * 3.2+版本架构图
-<img src="https://skywalkingtest.github.io/page-resources/3.x-architecture.jpg"/>
+<img src="https://skywalkingtest.github.io/page-resources/3.2.5%2b_architecture.jpg"/>
 
 # Document
 [![EN doc](https://img.shields.io/badge/document-English-blue.svg)](docs/README.md) [![cn doc](https://img.shields.io/badge/document-中文-blue.svg)](docs/README_ZH.md)

apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/AgentModuleGRPCProvider.java

@@ -42,7 +42,7 @@ import org.skywalking.apm.collector.naming.service.NamingHandlerRegisterService;
 import org.skywalking.apm.collector.remote.RemoteModule;
 import org.skywalking.apm.collector.server.Server;
 import org.skywalking.apm.collector.storage.StorageModule;
-import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;
+import org.skywalking.apm.collector.stream.StreamModule;
 
 /**
  * @author peng-yongsheng
@@ -82,7 +82,7 @@ public class AgentModuleGRPCProvider extends ModuleProvider {
         GRPCManagerService managerService = getManager().find(GRPCManagerModule.NAME).getService(GRPCManagerService.class);
         Server gRPCServer = managerService.createIfAbsent(host, port);
 
-        AgentStreamSingleton.getInstance(getManager(), new WorkerCreateListener());
+        AgentStreamSingleton.createInstanceIfAbsent(getManager());
         addHandlers(gRPCServer);
     }
 
@@ -91,7 +91,7 @@ public class AgentModuleGRPCProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, GRPCManagerModule.NAME, CacheModule.NAME, RemoteModule.NAME};
+        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, GRPCManagerModule.NAME, CacheModule.NAME, RemoteModule.NAME, StreamModule.NAME};
     }
 
     private void addHandlers(Server gRPCServer) {

apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/main/java/org/skywalking/apm/collector/agent/grpc/handler/JVMMetricsServiceHandler.java

@@ -68,7 +68,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe
 
         request.getMetricsList().forEach(metric -> {
             long time = TimeBucketUtils.INSTANCE.getSecondTimeBucket(metric.getTime());
-            senToInstanceHeartBeatPersistenceWorker(instanceId, metric.getTime());
+            sendToInstanceHeartBeatPersistenceWorker(instanceId, metric.getTime());
             sendToCpuMetricPersistenceWorker(instanceId, time, metric.getCpu());
             sendToMemoryMetricPersistenceWorker(instanceId, time, metric.getMemoryList());
             sendToMemoryPoolMetricPersistenceWorker(instanceId, time, metric.getMemoryPoolList());
@@ -79,7 +79,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe
         responseObserver.onCompleted();
     }
 
-    private void senToInstanceHeartBeatPersistenceWorker(int instanceId, long heartBeatTime) {
+    private void sendToInstanceHeartBeatPersistenceWorker(int instanceId, long heartBeatTime) {
         Instance instance = new Instance(String.valueOf(instanceId));
         instance.setHeartBeatTime(TimeBucketUtils.INSTANCE.getSecondTimeBucket(heartBeatTime));
         instance.setInstanceId(instanceId);

apm-collector/apm-collector-agent/collector-agent-grpc-provider/src/test/java/org/skywalking/apm/collector/agent/grpc/handler/ApplicationRegisterServiceHandlerTestCase.java

@@ -35,7 +35,6 @@ public class ApplicationRegisterServiceHandlerTestCase {
 
     private ApplicationRegisterServiceGrpc.ApplicationRegisterServiceBlockingStub stub;
 
-    //@Test
     public void testRegister() {
         ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).usePlaintext(true).build();
         stub = ApplicationRegisterServiceGrpc.newBlockingStub(channel);

apm-collector/apm-collector-agent/collector-agent-jetty-provider/src/main/java/org/skywalking/apm/collector/agent/jetty/AgentModuleJettyProvider.java

@@ -36,7 +36,7 @@ import org.skywalking.apm.collector.naming.NamingModule;
 import org.skywalking.apm.collector.naming.service.NamingHandlerRegisterService;
 import org.skywalking.apm.collector.server.Server;
 import org.skywalking.apm.collector.storage.StorageModule;
-import org.skywalking.apm.collector.storage.service.DAOService;
+import org.skywalking.apm.collector.stream.StreamModule;
 
 /**
  * @author peng-yongsheng
@@ -75,11 +75,9 @@ public class AgentModuleJettyProvider extends ModuleProvider {
         NamingHandlerRegisterService namingHandlerRegisterService = getManager().find(NamingModule.NAME).getService(NamingHandlerRegisterService.class);
         namingHandlerRegisterService.register(new AgentJettyNamingHandler(namingListener));
 
-        DAOService daoService = getManager().find(StorageModule.NAME).getService(DAOService.class);
-
         JettyManagerService managerService = getManager().find(JettyManagerModule.NAME).getService(JettyManagerService.class);
         Server jettyServer = managerService.createIfAbsent(host, port, contextPath);
-        addHandlers(daoService, jettyServer);
+        addHandlers(jettyServer);
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@@ -87,10 +85,10 @@ public class AgentModuleJettyProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, JettyManagerModule.NAME, CacheModule.NAME};
+        return new String[] {ClusterModule.NAME, NamingModule.NAME, StorageModule.NAME, JettyManagerModule.NAME, CacheModule.NAME, StreamModule.NAME};
     }
 
-    private void addHandlers(DAOService daoService, Server jettyServer) {
+    private void addHandlers(Server jettyServer) {
         jettyServer.addHandler(new TraceSegmentServletHandler());
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/AgentStreamSingleton.java

@@ -18,9 +18,12 @@
 
 package org.skywalking.apm.collector.agent.stream;
 
+import org.skywalking.apm.collector.agent.stream.graph.JvmMetricStreamGraph;
 import org.skywalking.apm.collector.agent.stream.graph.RegisterStreamGraph;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.ObjectUtils;
+import org.skywalking.apm.collector.stream.timer.PersistenceTimer;
 import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;
 
 /**
@@ -33,24 +36,35 @@ public class AgentStreamSingleton {
     private final ModuleManager moduleManager;
     private final WorkerCreateListener workerCreateListener;
 
-    public AgentStreamSingleton(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
+    private AgentStreamSingleton(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
-        this.workerCreateListener = workerCreateListener;
-        createJVMGraph();
-        createRegisterGraph();
-        createTraceGraph();
+        this.workerCreateListener = new WorkerCreateListener();
+        this.create();
     }
 
-    public static synchronized AgentStreamSingleton getInstance(ModuleManager moduleManager,
-        WorkerCreateListener workerCreateListener) {
+    public static synchronized AgentStreamSingleton createInstanceIfAbsent(ModuleManager moduleManager) {
         if (ObjectUtils.isEmpty(INSTANCE)) {
-            INSTANCE = new AgentStreamSingleton(moduleManager, workerCreateListener);
+            INSTANCE = new AgentStreamSingleton(moduleManager);
         }
         return INSTANCE;
     }
 
-    private void createJVMGraph() {
+    private void create() {
+        createJVMGraph();
+        createRegisterGraph();
+        createTraceGraph();
 
+        PersistenceTimer timer = new PersistenceTimer();
+        timer.start(moduleManager, workerCreateListener.getPersistenceWorkers());
+    }
+
+    private void createJVMGraph() {
+        JvmMetricStreamGraph jvmMetricStreamGraph = new JvmMetricStreamGraph(moduleManager, workerCreateListener);
+        jvmMetricStreamGraph.createCpuMetricGraph();
+        jvmMetricStreamGraph.createGcMetricGraph();
+        jvmMetricStreamGraph.createMemoryMetricGraph();
+        jvmMetricStreamGraph.createMemoryPoolMetricGraph();
+        jvmMetricStreamGraph.createHeartBeatGraph();
     }
 
     private void createRegisterGraph() {
@@ -61,6 +75,16 @@ public class AgentStreamSingleton {
     }
 
     private void createTraceGraph() {
-
+        TraceStreamGraph traceStreamGraph = new TraceStreamGraph(moduleManager, workerCreateListener);
+        traceStreamGraph.createSegmentStandardizationGraph();
+        traceStreamGraph.createGlobalTraceGraph();
+        traceStreamGraph.createInstPerformanceGraph();
+        traceStreamGraph.createNodeComponentGraph();
+        traceStreamGraph.createNodeMappingGraph();
+        traceStreamGraph.createNodeReferenceGraph();
+        traceStreamGraph.createServiceEntryGraph();
+        traceStreamGraph.createServiceReferenceGraph();
+        traceStreamGraph.createSegmentGraph();
+        traceStreamGraph.createSegmentCostGraph();
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/graph/JvmMetricStreamGraph.java

@@ -18,14 +18,22 @@
 
 package org.skywalking.apm.collector.agent.stream.graph;
 
+import org.skywalking.apm.collector.agent.stream.worker.jvm.CpuMetricPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.jvm.GCMetricPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.jvm.InstHeartBeatPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryMetricPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryPoolMetricPersistenceWorker;
 import org.skywalking.apm.collector.core.graph.Graph;
 import org.skywalking.apm.collector.core.graph.GraphManager;
+import org.skywalking.apm.collector.core.module.ModuleManager;
+import org.skywalking.apm.collector.queue.QueueModule;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 import org.skywalking.apm.collector.storage.table.jvm.GCMetric;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
 import org.skywalking.apm.collector.storage.table.register.Instance;
-import org.skywalking.apm.collector.stream.worker.base.ProviderNotFoundException;
+import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;
 
 /**
  * @author peng-yongsheng
@@ -38,28 +46,56 @@ public class JvmMetricStreamGraph {
     public static final int CPU_METRIC_GRAPH_ID = 103;
     public static final int INST_HEART_BEAT_GRAPH_ID = 104;
 
+    private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
+
+    public JvmMetricStreamGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
+        this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
+    }
+
+    @SuppressWarnings("unchecked")
     public Graph<GCMetric> createGcMetricGraph() {
+        QueueCreatorService<GCMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
         Graph<GCMetric> graph = GraphManager.INSTANCE.createIfAbsent(GC_METRIC_GRAPH_ID, GCMetric.class);
+        graph.addNode(new GCMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
         return graph;
     }
 
-    public Graph<CpuMetric> createCpuMetricGraph() throws ProviderNotFoundException {
+    @SuppressWarnings("unchecked")
+    public Graph<CpuMetric> createCpuMetricGraph() {
+        QueueCreatorService<CpuMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
         Graph<CpuMetric> graph = GraphManager.INSTANCE.createIfAbsent(CPU_METRIC_GRAPH_ID, CpuMetric.class);
+        graph.addNode(new CpuMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
         return graph;
     }
 
+    @SuppressWarnings("unchecked")
     public Graph<MemoryMetric> createMemoryMetricGraph() {
+        QueueCreatorService<MemoryMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
         Graph<MemoryMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_METRIC_GRAPH_ID, MemoryMetric.class);
+        graph.addNode(new MemoryMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
         return graph;
     }
 
+    @SuppressWarnings("unchecked")
     public Graph<MemoryPoolMetric> createMemoryPoolMetricGraph() {
+        QueueCreatorService<MemoryPoolMetric> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
         Graph<MemoryPoolMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class);
+        graph.addNode(new MemoryPoolMetricPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
         return graph;
     }
 
+    @SuppressWarnings("unchecked")
     public Graph<Instance> createHeartBeatGraph() {
+        QueueCreatorService<Instance> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
         Graph<Instance> graph = GraphManager.INSTANCE.createIfAbsent(INST_HEART_BEAT_GRAPH_ID, Instance.class);
+        graph.addNode(new InstHeartBeatPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
         return graph;
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/graph/RegisterStreamGraph.java

@@ -31,8 +31,6 @@ import org.skywalking.apm.collector.queue.QueueModule;
 import org.skywalking.apm.collector.queue.service.QueueCreatorService;
 import org.skywalking.apm.collector.remote.RemoteModule;
 import org.skywalking.apm.collector.remote.service.RemoteSenderService;
-import org.skywalking.apm.collector.storage.StorageModule;
-import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.register.Application;
 import org.skywalking.apm.collector.storage.table.register.Instance;
 import org.skywalking.apm.collector.storage.table.register.ServiceName;
@@ -57,7 +55,6 @@ public class RegisterStreamGraph {
 
     @SuppressWarnings("unchecked")
     public Graph<Application> createApplicationRegisterGraph() {
-        DAOService daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         QueueCreatorService<Application> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
@@ -70,7 +67,6 @@ public class RegisterStreamGraph {
 
     @SuppressWarnings("unchecked")
     public Graph<Instance> createInstanceRegisterGraph() {
-        DAOService daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         QueueCreatorService<Instance> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
@@ -83,7 +79,6 @@ public class RegisterStreamGraph {
 
     @SuppressWarnings("unchecked")
     public Graph<ServiceName> createServiceNameRegisterGraph() {
-        DAOService daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         QueueCreatorService<ServiceName> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java

+/*
+ * Copyright 2017, OpenSkywalking Organization All rights reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * Project repository: https://github.com/OpenSkywalking/skywalking
+ */
+
+package org.skywalking.apm.collector.agent.stream.graph;
+
+import org.skywalking.apm.collector.agent.stream.parser.standardization.SegmentStandardizationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.global.GlobalTracePersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.instance.InstPerformancePersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeComponentAggregationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeComponentPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeComponentRemoteWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeMappingAggregationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeMappingPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.node.NodeMappingRemoteWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.noderef.NodeReferenceAggregationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.noderef.NodeReferencePersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.noderef.NodeReferenceRemoteWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentCostPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryAggregationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryPersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryRemoteWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.serviceref.ServiceReferenceAggregationWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.serviceref.ServiceReferencePersistenceWorker;
+import org.skywalking.apm.collector.agent.stream.worker.trace.serviceref.ServiceReferenceRemoteWorker;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
+import org.skywalking.apm.collector.core.module.ModuleManager;
+import org.skywalking.apm.collector.queue.QueueModule;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
+import org.skywalking.apm.collector.remote.RemoteModule;
+import org.skywalking.apm.collector.remote.service.RemoteSenderService;
+import org.skywalking.apm.collector.storage.table.global.GlobalTrace;
+import org.skywalking.apm.collector.storage.table.instance.InstPerformance;
+import org.skywalking.apm.collector.storage.table.node.NodeComponent;
+import org.skywalking.apm.collector.storage.table.node.NodeMapping;
+import org.skywalking.apm.collector.storage.table.noderef.NodeReference;
+import org.skywalking.apm.collector.storage.table.segment.Segment;
+import org.skywalking.apm.collector.storage.table.segment.SegmentCost;
+import org.skywalking.apm.collector.storage.table.service.ServiceEntry;
+import org.skywalking.apm.collector.storage.table.serviceref.ServiceReference;
+import org.skywalking.apm.collector.stream.worker.base.WorkerCreateListener;
+import org.skywalking.apm.network.proto.UpstreamSegment;
+
+/**
+ * @author peng-yongsheng
+ */
+public class TraceStreamGraph {
+
+    public static final int GLOBAL_TRACE_GRAPH_ID = 300;
+    public static final int INST_PERFORMANCE_GRAPH_ID = 301;
+    public static final int NODE_COMPONENT_GRAPH_ID = 302;
+    public static final int NODE_MAPPING_GRAPH_ID = 303;
+    public static final int NODE_REFERENCE_GRAPH_ID = 304;
+    public static final int SERVICE_ENTRY_GRAPH_ID = 305;
+    public static final int SERVICE_REFERENCE_GRAPH_ID = 306;
+    public static final int SEGMENT_GRAPH_ID = 307;
+    public static final int SEGMENT_COST_GRAPH_ID = 308;
+    public static final int SEGMENT_STANDARDIZATION_GRAPH_ID = 309;
+
+    private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
+
+    public TraceStreamGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
+        this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<UpstreamSegment> createSegmentStandardizationGraph() {
+        QueueCreatorService<UpstreamSegment> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
+        Graph<UpstreamSegment> graph = GraphManager.INSTANCE.createIfAbsent(SEGMENT_STANDARDIZATION_GRAPH_ID, UpstreamSegment.class);
+        graph.addNode(new SegmentStandardizationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<GlobalTrace> createGlobalTraceGraph() {
+        QueueCreatorService<GlobalTrace> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
+        Graph<GlobalTrace> graph = GraphManager.INSTANCE.createIfAbsent(GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class);
+        graph.addNode(new GlobalTracePersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<InstPerformance> createInstPerformanceGraph() {
+        QueueCreatorService<InstPerformance> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
+        Graph<InstPerformance> graph = GraphManager.INSTANCE.createIfAbsent(INST_PERFORMANCE_GRAPH_ID, InstPerformance.class);
+        graph.addNode(new InstPerformancePersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<NodeComponent> createNodeComponentGraph() {
+        QueueCreatorService<NodeComponent> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Graph<NodeComponent> graph = GraphManager.INSTANCE.createIfAbsent(NODE_COMPONENT_GRAPH_ID, NodeComponent.class);
+        graph.addNode(new NodeComponentAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
+            .addNext(new NodeComponentRemoteWorker.Factory(moduleManager, remoteSenderService, NODE_COMPONENT_GRAPH_ID).create(workerCreateListener))
+            .addNext(new NodeComponentPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<NodeMapping> createNodeMappingGraph() {
+        QueueCreatorService<NodeMapping> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Graph<NodeMapping> graph = GraphManager.INSTANCE.createIfAbsent(NODE_MAPPING_GRAPH_ID, NodeMapping.class);
+        graph.addNode(new NodeMappingAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
+            .addNext(new NodeMappingRemoteWorker.Factory(moduleManager, remoteSenderService, NODE_MAPPING_GRAPH_ID).create(workerCreateListener))
+            .addNext(new NodeMappingPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<NodeReference> createNodeReferenceGraph() {
+        QueueCreatorService<NodeReference> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Graph<NodeReference> graph = GraphManager.INSTANCE.createIfAbsent(NODE_REFERENCE_GRAPH_ID, NodeReference.class);
+        graph.addNode(new NodeReferenceAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
+            .addNext(new NodeReferenceRemoteWorker.Factory(moduleManager, remoteSenderService, NODE_REFERENCE_GRAPH_ID).create(workerCreateListener))
+            .addNext(new NodeReferencePersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<ServiceEntry> createServiceEntryGraph() {
+        QueueCreatorService<ServiceEntry> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Graph<ServiceEntry> graph = GraphManager.INSTANCE.createIfAbsent(SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class);
+        graph.addNode(new ServiceEntryAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
+            .addNext(new ServiceEntryRemoteWorker.Factory(moduleManager, remoteSenderService, SERVICE_ENTRY_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceEntryPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<ServiceReference> createServiceReferenceGraph() {
+        QueueCreatorService<ServiceReference> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+        RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
+
+        Graph<ServiceReference> graph = GraphManager.INSTANCE.createIfAbsent(SERVICE_REFERENCE_GRAPH_ID, ServiceReference.class);
+        graph.addNode(new ServiceReferenceAggregationWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener))
+            .addNext(new ServiceReferenceRemoteWorker.Factory(moduleManager, remoteSenderService, SERVICE_REFERENCE_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceReferencePersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<Segment> createSegmentGraph() {
+        QueueCreatorService<Segment> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
+        Graph<Segment> graph = GraphManager.INSTANCE.createIfAbsent(SEGMENT_GRAPH_ID, Segment.class);
+        graph.addNode(new SegmentPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+
+    @SuppressWarnings("unchecked")
+    public Graph<SegmentCost> createSegmentCostGraph() {
+        QueueCreatorService<SegmentCost> queueCreatorService = moduleManager.find(QueueModule.NAME).getService(QueueCreatorService.class);
+
+        Graph<SegmentCost> graph = GraphManager.INSTANCE.createIfAbsent(SEGMENT_COST_GRAPH_ID, SegmentCost.class);
+        graph.addNode(new SegmentCostPersistenceWorker.Factory(moduleManager, queueCreatorService).create(workerCreateListener));
+        return graph;
+    }
+}

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/SegmentParse.java

@@ -21,6 +21,7 @@ package org.skywalking.apm.collector.agent.stream.parser;
 import com.google.protobuf.InvalidProtocolBufferException;
 import java.util.ArrayList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.ReferenceDecorator;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.ReferenceIdExchanger;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SegmentDecorator;
@@ -34,6 +35,8 @@ import org.skywalking.apm.collector.agent.stream.worker.trace.noderef.NodeRefere
 import org.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentCostSpanListener;
 import org.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.worker.trace.serviceref.ServiceReferenceSpanListener;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.storage.table.segment.Segment;
 import org.skywalking.apm.network.proto.SpanType;
@@ -150,11 +153,14 @@ public class SegmentParse {
     private void buildSegment(String id, byte[] dataBinary) {
         Segment segment = new Segment(id);
         segment.setDataBinary(dataBinary);
+        Graph<Segment> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_GRAPH_ID, Segment.class);
+        graph.start(segment);
     }
 
     private void writeToBufferFile(String id, UpstreamSegment upstreamSegment) {
         logger.debug("send to segment buffer write worker, id: {}", id);
-//        context.getClusterWorkerContext().lookup(SegmentStandardizationWorker.WorkerRole.INSTANCE).tell(upstreamSegment);
+        Graph<UpstreamSegment> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_STANDARDIZATION_GRAPH_ID, UpstreamSegment.class);
+        graph.start(upstreamSegment);
     }
 
     private void notifyListenerToBuild() {

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/parser/standardization/SegmentStandardizationWorker.java

@@ -41,7 +41,7 @@ public class SegmentStandardizationWorker extends AbstractLocalAsyncWorker<Upstr
     }
 
     @Override public int id() {
-        return 0;
+        return SegmentStandardizationWorker.class.hashCode();
     }
 
     @Override protected void onWork(UpstreamSegment upstreamSegment) throws WorkerException {

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ApplicationIDService.java

@@ -21,6 +21,7 @@ package org.skywalking.apm.collector.agent.stream.worker.register;
 import org.skywalking.apm.collector.agent.stream.graph.RegisterStreamGraph;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.ApplicationCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
 import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.module.ModuleNotFoundException;
@@ -37,12 +38,13 @@ public class ApplicationIDService {
     private final Logger logger = LoggerFactory.getLogger(ApplicationIDService.class);
 
     private final ModuleManager moduleManager;
+    private final Graph<Application> applicationRegisterGraph;
 
     public ApplicationIDService(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
+        this.applicationRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID, Application.class);
     }
 
-    @SuppressWarnings("unchecked")
     public int getOrCreate(String applicationCode) throws ModuleNotFoundException, ServiceNotProvidedException {
         ApplicationCacheService service = moduleManager.find(CacheModule.NAME).getService(ApplicationCacheService.class);
         int applicationId = service.get(applicationCode);
@@ -52,7 +54,7 @@ public class ApplicationIDService {
             application.setApplicationCode(applicationCode);
             application.setApplicationId(0);
 
-            GraphManager.INSTANCE.findGraph(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID).start(application);
+            applicationRegisterGraph.start(application);
         }
         return applicationId;
     }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/InstanceIDService.java

@@ -21,6 +21,7 @@ package org.skywalking.apm.collector.agent.stream.worker.register;
 import org.skywalking.apm.collector.agent.stream.graph.RegisterStreamGraph;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.InstanceCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
 import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.module.ModuleNotFoundException;
@@ -40,12 +41,13 @@ public class InstanceIDService {
     private final Logger logger = LoggerFactory.getLogger(InstanceIDService.class);
 
     private final ModuleManager moduleManager;
+    private final Graph<Instance> instanceRegisterGraph;
 
     public InstanceIDService(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
+        this.instanceRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.INSTANCE_REGISTER_GRAPH_ID, Instance.class);
     }
 
-    @SuppressWarnings("unchecked")
     public int getOrCreate(int applicationId, String agentUUID, long registerTime,
         String osInfo) throws ModuleNotFoundException, ServiceNotProvidedException {
         logger.debug("get or create instance id, application id: {}, agentUUID: {}, registerTime: {}, osInfo: {}", applicationId, agentUUID, registerTime, osInfo);
@@ -61,7 +63,7 @@ public class InstanceIDService {
             instance.setInstanceId(0);
             instance.setOsInfo(osInfo);
 
-            GraphManager.INSTANCE.findGraph(RegisterStreamGraph.INSTANCE_REGISTER_GRAPH_ID).start(instance);
+            instanceRegisterGraph.start(instance);
         }
         return instanceId;
     }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/register/ServiceNameService.java

@@ -21,6 +21,7 @@ package org.skywalking.apm.collector.agent.stream.worker.register;
 import org.skywalking.apm.collector.agent.stream.graph.RegisterStreamGraph;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.ServiceIdCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
 import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.storage.table.register.ServiceName;
@@ -35,12 +36,13 @@ public class ServiceNameService {
     private final Logger logger = LoggerFactory.getLogger(ServiceNameService.class);
 
     private final ModuleManager moduleManager;
+    private final Graph<ServiceName> serviceNameRegisterGraph;
 
     public ServiceNameService(ModuleManager moduleManager) {
         this.moduleManager = moduleManager;
+        this.serviceNameRegisterGraph = GraphManager.INSTANCE.createIfAbsent(RegisterStreamGraph.APPLICATION_REGISTER_GRAPH_ID, ServiceName.class);
     }
 
-    @SuppressWarnings("unchecked")
     public int getOrCreate(int applicationId, String serviceName) {
         ServiceIdCacheService idCacheService = moduleManager.find(CacheModule.NAME).getService(ServiceIdCacheService.class);
         int serviceId = idCacheService.get(applicationId, serviceName);
@@ -51,7 +53,7 @@ public class ServiceNameService {
             service.setServiceName(serviceName);
             service.setServiceId(0);
 
-            GraphManager.INSTANCE.findGraph(RegisterStreamGraph.SERVICE_NAME_REGISTER_GRAPH_ID).start(service);
+            serviceNameRegisterGraph.start(service);
         }
         return serviceId;
     }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTraceSpanListener.java

@@ -20,9 +20,12 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.global;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.GlobalTraceIdsListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.skywalking.apm.collector.storage.table.global.GlobalTrace;
@@ -63,11 +66,13 @@ public class GlobalTraceSpanListener implements FirstSpanListener, GlobalTraceId
     @Override public void build() {
         logger.debug("global trace listener build");
 
+        Graph<GlobalTrace> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class);
         for (String globalTraceId : globalTraceIds) {
             GlobalTrace globalTrace = new GlobalTrace(segmentId + Const.ID_SPLIT + globalTraceId);
             globalTrace.setGlobalTraceId(globalTraceId);
             globalTrace.setSegmentId(segmentId);
             globalTrace.setTimeBucket(timeBucket);
+            graph.start(globalTrace);
         }
     }
 }
\ No newline at end of file

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/instance/InstPerformanceSpanListener.java

@@ -18,9 +18,12 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.trace.instance;
 
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.skywalking.apm.collector.storage.table.instance.InstPerformance;
@@ -60,5 +63,8 @@ public class InstPerformanceSpanListener implements EntrySpanListener, FirstSpan
         instPerformance.setCalls(1);
         instPerformance.setCostTotal(cost);
         instPerformance.setTimeBucket(timeBucket);
+
+        Graph<InstPerformance> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.INST_PERFORMANCE_GRAPH_ID, InstPerformance.class);
+        graph.start(instPerformance);
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeComponentSpanListener.java

@@ -20,10 +20,13 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.node;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.skywalking.apm.collector.storage.table.node.NodeComponent;
@@ -89,9 +92,12 @@ public class NodeComponentSpanListener implements EntrySpanListener, ExitSpanLis
     }
 
     @Override public void build() {
+        Graph<NodeComponent> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.NODE_COMPONENT_GRAPH_ID, NodeComponent.class);
+
         nodeComponents.forEach(nodeComponent -> {
             nodeComponent.setId(timeBucket + Const.ID_SPLIT + nodeComponent.getId());
             nodeComponent.setTimeBucket(timeBucket);
+            graph.start(nodeComponent);
         });
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/node/NodeMappingSpanListener.java

@@ -20,10 +20,13 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.node;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.RefsListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.ReferenceDecorator;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.skywalking.apm.collector.storage.table.node.NodeMapping;
@@ -59,10 +62,13 @@ public class NodeMappingSpanListener implements RefsListener, FirstSpanListener
 
     @Override public void build() {
         logger.debug("node mapping listener build");
+        Graph<NodeMapping> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.NODE_MAPPING_GRAPH_ID, NodeMapping.class);
+
         for (NodeMapping nodeMapping : nodeMappings) {
             nodeMapping.setId(timeBucket + Const.ID_SPLIT + nodeMapping.getId());
             nodeMapping.setTimeBucket(timeBucket);
             logger.debug("send to node mapping aggregation worker, id: {}", nodeMapping.getId());
+            graph.start(nodeMapping);
         }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/noderef/NodeReferenceSpanListener.java

@@ -20,6 +20,7 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.noderef;
 
 import java.util.LinkedList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.RefsListener;
@@ -27,6 +28,8 @@ import org.skywalking.apm.collector.agent.stream.parser.standardization.Referenc
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.InstanceCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.CollectionUtils;
 import org.skywalking.apm.collector.core.util.Const;
@@ -105,7 +108,9 @@ public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanLis
 
     @Override public void build() {
         logger.debug("node reference summary listener build");
+        Graph<NodeReference> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.NODE_REFERENCE_GRAPH_ID, NodeReference.class);
         for (NodeReference nodeReference : nodeReferences) {
+            graph.start(nodeReference);
         }
     }
 

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostSpanListener.java

@@ -20,6 +20,7 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.segment;
 
 import java.util.ArrayList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
@@ -27,6 +28,8 @@ import org.skywalking.apm.collector.agent.stream.parser.LocalSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.ServiceNameCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
@@ -91,10 +94,12 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe
     }
 
     @Override public void build() {
+        Graph<SegmentCost> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_COST_GRAPH_ID, SegmentCost.class);
         logger.debug("segment cost listener build");
         for (SegmentCost segmentCost : segmentCosts) {
             segmentCost.setIsError(isError);
             segmentCost.setTimeBucket(timeBucket);
+            graph.start(segmentCost);
         }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java

@@ -18,6 +18,7 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.trace.service;
 
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.RefsListener;
@@ -25,6 +26,8 @@ import org.skywalking.apm.collector.agent.stream.parser.standardization.Referenc
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
 import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.cache.service.ServiceNameCacheService;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
@@ -82,6 +85,8 @@ public class ServiceEntrySpanListener implements RefsListener, FirstSpanListener
             serviceEntry.setNewestTime(timeBucket);
 
             logger.debug("send to service entry aggregation worker, id: {}", serviceEntry.getId());
+            Graph<ServiceEntry> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class);
+            graph.start(serviceEntry);
         }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/trace/serviceref/ServiceReferenceSpanListener.java

@@ -20,11 +20,14 @@ package org.skywalking.apm.collector.agent.stream.worker.trace.serviceref;
 
 import java.util.LinkedList;
 import java.util.List;
+import org.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
 import org.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
 import org.skywalking.apm.collector.agent.stream.parser.RefsListener;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.ReferenceDecorator;
 import org.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
+import org.skywalking.apm.collector.core.graph.Graph;
+import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.skywalking.apm.collector.storage.table.serviceref.ServiceReference;
@@ -128,5 +131,8 @@ public class ServiceReferenceSpanListener implements FirstSpanListener, EntrySpa
         serviceReference.setId(idBuilder.toString());
         serviceReference.setTimeBucket(timeBucket);
         logger.debug("send to service reference aggregation worker, id: {}", serviceReference.getId());
+
+        Graph<ServiceReference> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_REFERENCE_GRAPH_ID, ServiceReference.class);
+        graph.start(serviceReference);
     }
 }

apm-collector/apm-collector-boot/src/main/resources/log4j2.xml

@@ -17,7 +17,7 @@
   ~ Project repository: https://github.com/OpenSkywalking/skywalking
   -->
 
-<Configuration status="info">
+<Configuration status="debug">
     <Appenders>
         <Console name="Console" target="SYSTEM_OUT">
             <PatternLayout charset="UTF-8" pattern="%d - %c -%-4r [%t] %-5p %x - %m%n"/>
@@ -26,9 +26,9 @@
     <Loggers>
         <logger name="org.eclipse.jetty" level="INFO"/>
         <logger name="org.apache.zookeeper" level="INFO"/>
-        <logger name="org.skywalking.apm.collector.agentstream.worker.storage.PersistenceTimer" level="INFO"/>
-        <logger name="io.grpc.netty.NettyServerHandler" level="INFO"/>
-        <Root level="info">
+        <logger name="org.skywalking.apm.collector.agentstream.worker.storage.PersistenceTimer" level="debug"/>
+        <logger name="io.grpc.netty" level="INFO"/>
+        <Root level="debug">
             <AppenderRef ref="Console"/>
         </Root>
     </Loggers>

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/CacheModuleGuavaProvider.java

@@ -32,7 +32,6 @@ import org.skywalking.apm.collector.core.module.Module;
 import org.skywalking.apm.collector.core.module.ModuleProvider;
 import org.skywalking.apm.collector.core.module.ServiceNotProvidedException;
 import org.skywalking.apm.collector.storage.StorageModule;
-import org.skywalking.apm.collector.storage.service.DAOService;
 
 /**
  * @author peng-yongsheng
@@ -48,15 +47,13 @@ public class CacheModuleGuavaProvider extends ModuleProvider {
     }
 
     @Override public void prepare(Properties config) throws ServiceNotProvidedException {
+        this.registerServiceImplementation(ApplicationCacheService.class, new ApplicationCacheGuavaService(getManager()));
+        this.registerServiceImplementation(InstanceCacheService.class, new InstanceCacheGuavaService(getManager()));
+        this.registerServiceImplementation(ServiceIdCacheService.class, new ServiceIdCacheGuavaService(getManager()));
+        this.registerServiceImplementation(ServiceNameCacheService.class, new ServiceNameCacheGuavaService(getManager()));
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
-        DAOService daoService = getManager().find(StorageModule.NAME).getService(DAOService.class);
-
-        this.registerServiceImplementation(ApplicationCacheService.class, new ApplicationCacheGuavaService(daoService));
-        this.registerServiceImplementation(InstanceCacheService.class, new InstanceCacheGuavaService(daoService));
-        this.registerServiceImplementation(ServiceIdCacheService.class, new ServiceIdCacheGuavaService(daoService));
-        this.registerServiceImplementation(ServiceNameCacheService.class, new ServiceNameCacheGuavaService(daoService));
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ApplicationCacheGuavaService.java

@@ -21,8 +21,10 @@ package org.skywalking.apm.collector.cache.guava.service;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import org.skywalking.apm.collector.cache.service.ApplicationCacheService;
+import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.StringUtils;
+import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.dao.IApplicationCacheDAO;
 import org.skywalking.apm.collector.storage.service.DAOService;
 import org.slf4j.Logger;
@@ -39,8 +41,8 @@ public class ApplicationCacheGuavaService implements ApplicationCacheService {
 
     private final DAOService daoService;
 
-    public ApplicationCacheGuavaService(DAOService daoService) {
-        this.daoService = daoService;
+    public ApplicationCacheGuavaService(ModuleManager moduleManager) {
+        this.daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
     }
 
     public int get(String applicationCode) {

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/InstanceCacheGuavaService.java

@@ -21,7 +21,9 @@ package org.skywalking.apm.collector.cache.guava.service;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import org.skywalking.apm.collector.cache.service.InstanceCacheService;
+import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
+import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.dao.IInstanceCacheDAO;
 import org.skywalking.apm.collector.storage.service.DAOService;
 import org.slf4j.Logger;
@@ -40,8 +42,8 @@ public class InstanceCacheGuavaService implements InstanceCacheService {
 
     private final DAOService daoService;
 
-    public InstanceCacheGuavaService(DAOService daoService) {
-        this.daoService = daoService;
+    public InstanceCacheGuavaService(ModuleManager moduleManager) {
+        this.daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
     }
 
     public int get(int applicationInstanceId) {

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceIdCacheGuavaService.java

@@ -21,7 +21,9 @@ package org.skywalking.apm.collector.cache.guava.service;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import org.skywalking.apm.collector.cache.service.ServiceIdCacheService;
+import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
+import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO;
 import org.skywalking.apm.collector.storage.service.DAOService;
 import org.slf4j.Logger;
@@ -38,8 +40,8 @@ public class ServiceIdCacheGuavaService implements ServiceIdCacheService {
 
     private final DAOService daoService;
 
-    public ServiceIdCacheGuavaService(DAOService daoService) {
-        this.daoService = daoService;
+    public ServiceIdCacheGuavaService(ModuleManager moduleManager) {
+        this.daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
     }
 
     public int get(int applicationId, String serviceName) {

apm-collector/apm-collector-cache/collector-cache-guava-provider/src/main/java/org/skywalking/apm/collector/cache/guava/service/ServiceNameCacheGuavaService.java

@@ -21,8 +21,10 @@ package org.skywalking.apm.collector.cache.guava.service;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import org.skywalking.apm.collector.cache.service.ServiceNameCacheService;
+import org.skywalking.apm.collector.core.module.ModuleManager;
 import org.skywalking.apm.collector.core.util.Const;
 import org.skywalking.apm.collector.core.util.StringUtils;
+import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO;
 import org.skywalking.apm.collector.storage.service.DAOService;
 import org.slf4j.Logger;
@@ -39,8 +41,8 @@ public class ServiceNameCacheGuavaService implements ServiceNameCacheService {
 
     private final DAOService daoService;
 
-    public ServiceNameCacheGuavaService(DAOService daoService) {
-        this.daoService = daoService;
+    public ServiceNameCacheGuavaService(ModuleManager moduleManager) {
+        this.daoService = moduleManager.find(StorageModule.NAME).getService(DAOService.class);
     }
 
     public String get(int serviceId) {

apm-collector/apm-collector-core/src/main/resources/application-default.yml

@@ -24,6 +24,8 @@ jetty_manager:
   jetty:
 gRPC_manager:
   gRPC:
+stream:
+  worker:
 storage:
   h2:
     url: jdbc:h2:~/memorydb

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/StorageModule.java

@@ -20,6 +20,7 @@ package org.skywalking.apm.collector.storage;
 
 import org.skywalking.apm.collector.core.module.Module;
 import org.skywalking.apm.collector.storage.service.DAOService;
+import org.skywalking.apm.collector.storage.base.dao.IBatchDAO;
 
 /**
  * @author peng-yongsheng
@@ -33,6 +34,6 @@ public class StorageModule extends Module {
     }
 
     @Override public Class[] services() {
-        return new Class[] {DAOService.class};
+        return new Class[] {DAOService.class, IBatchDAO.class};
     }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/base/dao/DAO.java

@@ -18,8 +18,10 @@
 
 package org.skywalking.apm.collector.storage.base.dao;
 
+import org.skywalking.apm.collector.core.module.Service;
+
 /**
  * @author peng-yongsheng
  */
-public interface DAO {
+public interface DAO extends Service {
 }

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java

@@ -29,6 +29,8 @@ import org.skywalking.apm.collector.core.module.ServiceNotProvidedException;
 import org.skywalking.apm.collector.storage.StorageException;
 import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.base.dao.DAOContainer;
+import org.skywalking.apm.collector.storage.base.dao.IBatchDAO;
+import org.skywalking.apm.collector.storage.es.base.dao.BatchEsDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAODefineLoader;
 import org.skywalking.apm.collector.storage.es.base.define.ElasticSearchStorageInstaller;
@@ -72,6 +74,10 @@ public class StorageModuleEsProvider extends ModuleProvider {
         elasticSearchClient = new ElasticSearchClient(clusterName, clusterTransportSniffer, clusterNodes);
 
         this.registerServiceImplementation(DAOService.class, new ElasticSearchDAOService(daoContainer));
+
+        BatchEsDAO batchEsDAO = new BatchEsDAO();
+        batchEsDAO.setClient(elasticSearchClient);
+        this.registerServiceImplementation(IBatchDAO.class, batchEsDAO);
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java

@@ -29,6 +29,8 @@ import org.skywalking.apm.collector.core.module.ServiceNotProvidedException;
 import org.skywalking.apm.collector.storage.StorageException;
 import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.base.dao.DAOContainer;
+import org.skywalking.apm.collector.storage.base.dao.IBatchDAO;
+import org.skywalking.apm.collector.storage.h2.base.dao.BatchH2DAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAODefineLoader;
 import org.skywalking.apm.collector.storage.h2.base.define.H2StorageInstaller;
@@ -70,6 +72,7 @@ public class StorageModuleH2Provider extends ModuleProvider {
         client = new H2Client(url, userName, password);
 
         this.registerServiceImplementation(DAOService.class, new H2DAOService(daoContainer));
+        this.registerServiceImplementation(IBatchDAO.class, new BatchH2DAO());
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/StreamModuleProvider.java

@@ -19,17 +19,9 @@
 package org.skywalking.apm.collector.stream;
 
 import java.util.Properties;
-import org.skywalking.apm.collector.cache.CacheModule;
 import org.skywalking.apm.collector.core.module.Module;
 import org.skywalking.apm.collector.core.module.ModuleProvider;
 import org.skywalking.apm.collector.core.module.ServiceNotProvidedException;
-import org.skywalking.apm.collector.queue.QueueModule;
-import org.skywalking.apm.collector.queue.service.QueueCreatorService;
-import org.skywalking.apm.collector.remote.RemoteModule;
-import org.skywalking.apm.collector.remote.service.RemoteSenderService;
-import org.skywalking.apm.collector.storage.StorageModule;
-import org.skywalking.apm.collector.storage.service.DAOService;
-import org.skywalking.apm.collector.stream.timer.PersistenceTimer;
 
 /**
  * @author peng-yongsheng
@@ -48,11 +40,6 @@ public class StreamModuleProvider extends ModuleProvider {
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
-        PersistenceTimer persistenceTimer = new PersistenceTimer();
-        QueueCreatorService queueCreatorService = getManager().find(QueueModule.NAME).getService(QueueCreatorService.class);
-        RemoteSenderService remoteSenderService = getManager().find(RemoteModule.NAME).getService(RemoteSenderService.class);
-        DAOService daoService = getManager().find(StorageModule.NAME).getService(DAOService.class);
-        persistenceTimer.start(daoService);
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@@ -60,6 +47,6 @@ public class StreamModuleProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[] {RemoteModule.NAME, QueueModule.NAME, StorageModule.NAME, CacheModule.NAME};
+        return new String[] {};
     }
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/timer/PersistenceTimer.java

@@ -22,11 +22,11 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
+import org.skywalking.apm.collector.core.module.ModuleManager;
+import org.skywalking.apm.collector.storage.StorageModule;
 import org.skywalking.apm.collector.storage.base.dao.IBatchDAO;
-import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.stream.worker.base.WorkerException;
 import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
-import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorkerContainer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -37,19 +37,19 @@ public class PersistenceTimer {
 
     private final Logger logger = LoggerFactory.getLogger(PersistenceTimer.class);
 
-    public void start(DAOService daoService) {
+    public void start(ModuleManager moduleManager, List<PersistenceWorker> persistenceWorkers) {
         logger.info("persistence timer start");
         //TODO timer value config
 //        final long timeInterval = EsConfig.Es.Persistence.Timer.VALUE * 1000;
         final long timeInterval = 3;
-        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> extractDataAndSave(daoService), 1, timeInterval, TimeUnit.SECONDS);
+        IBatchDAO batchDAO = moduleManager.find(StorageModule.NAME).getService(IBatchDAO.class);
+        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> extractDataAndSave(batchDAO, persistenceWorkers), 1, timeInterval, TimeUnit.SECONDS);
     }
 
-    private void extractDataAndSave(DAOService daoService) {
+    private void extractDataAndSave(IBatchDAO batchDAO, List<PersistenceWorker> persistenceWorkers) {
         try {
-            List<PersistenceWorker> workers = PersistenceWorkerContainer.INSTANCE.getPersistenceWorkers();
             List batchAllCollection = new ArrayList<>();
-            workers.forEach((PersistenceWorker worker) -> {
+            persistenceWorkers.forEach((PersistenceWorker worker) -> {
                 logger.debug("extract {} worker data and save", worker.getClass().getName());
                 try {
                     worker.flushAndSwitch();
@@ -61,8 +61,7 @@ public class PersistenceTimer {
                 }
             });
 
-            IBatchDAO dao = (IBatchDAO)daoService.get(IBatchDAO.class);
-            dao.batchPersistence(batchAllCollection);
+            batchDAO.batchPersistence(batchAllCollection);
         } catch (Throwable e) {
             logger.error(e.getMessage(), e);
         } finally {

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/base/WorkerCreateListener.java

@@ -18,12 +18,28 @@
 
 package org.skywalking.apm.collector.stream.worker.base;
 
+import java.util.ArrayList;
+import java.util.List;
+import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
+
 /**
  * @author peng-yongsheng
  */
 public class WorkerCreateListener {
 
+    private final List<PersistenceWorker> persistenceWorkers;
+
+    public WorkerCreateListener() {
+        this.persistenceWorkers = new ArrayList<>();
+    }
+
     public void addWorker(AbstractWorker worker) {
+        if (worker instanceof PersistenceWorker) {
+            persistenceWorkers.add((PersistenceWorker)worker);
+        }
+    }
 
+    public List<PersistenceWorker> getPersistenceWorkers() {
+        return persistenceWorkers;
     }
 }

apm-collector/apm-collector-stream/collector-stream-provider/src/main/java/org/skywalking/apm/collector/stream/worker/impl/PersistenceWorkerContainer.java

-/*
- * Copyright 2017, OpenSkywalking Organization All rights reserved.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- * Project repository: https://github.com/OpenSkywalking/skywalking
- */
-
-package org.skywalking.apm.collector.stream.worker.impl;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * @author peng-yongsheng
- */
-public enum PersistenceWorkerContainer {
-    INSTANCE;
-
-    private List<PersistenceWorker> persistenceWorkers = new ArrayList<>();
-
-    public void addWorker(PersistenceWorker worker) {
-        persistenceWorkers.add(worker);
-    }
-
-    public List<PersistenceWorker> getPersistenceWorkers() {
-        return persistenceWorkers;
-    }
-}

pom.xml

@@ -25,38 +25,29 @@
     <artifactId>apm</artifactId>
     <version>3.2.4-2017</version>
 
-    <licenses>
-        <license>
-            <name>GNU GENERAL PUBLIC LICENSE V3</name>
-            <url>https://github.com/wu-sheng/sky-walking/blob/master/LICENSE</url>
-        </license>
-    </licenses>
-
     <developers>
         <developer>
             <name>Wu Sheng</name>
             <email>wu.sheng@foxmail.com</email>
-            <url>https://wu-sheng.github.io/me/</url>
-        </developer>
-        <developer>
-            <name>Zhang Xin</name>
-            <url>https://github.com/ascrutae</url>
-        </developer>
-        <developer>
-            <name>Tan Zhen</name>
-            <url>https://github.com/mircoteam</url>
-        </developer>
-        <developer>
-            <name>Xu Yan</name>
-            <url>https://github.com/TastySummer</url>
+            <url>https://github.com/wu-sheng</url>
+            <roles>
+                <role>Founder</role>
+                <role>PMC member</role>
+            </roles>
         </developer>
         <developer>
             <name>Peng Yongsheng</name>
-            <email>8082209@qq.com</email>
             <url>https://github.com/peng-yongsheng</url>
+            <roles>
+                <role>PMC member</role>
+            </roles>
         </developer>
         <developer>
-            <name>Dai Wen</name>
+            <name>Zhang Xin</name>
+            <url>https://github.com/ascrutae</url>
+            <roles>
+                <role>PMC member</role>
+            </roles>
         </developer>
     </developers>