metric module test success.

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

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import org.apache.skywalking.apm.collector.cluster.ModuleRegistration;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/ApplicationRegisterPost.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/ApplicationRegisterPost.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import com.google.gson.JsonElement;
 import java.io.IOException;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/HttpClientTools.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/HttpClientTools.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import java.io.IOException;
 import java.net.URI;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/InstanceRegisterPost.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/InstanceRegisterPost.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import com.google.gson.JsonElement;
 import java.io.IOException;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/JsonFileReader.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/JsonFileReader.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/SegmentPost.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/SegmentPost.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import com.google.gson.JsonElement;
 import java.io.IOException;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/ServiceNameRegisterPost.java → apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/java/org/apache/skywalking/apm/collector/agent/jetty/provider/ServiceNameRegisterPost.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.agent.jetty.provider.handler;
+package org.apache.skywalking.apm.collector.agent.jetty.provider;
 
 import com.google.gson.JsonElement;
 import java.io.IOException;

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/application-register-consumer.json

+[
+  "dubbox-consumer"
+]
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/application-register-provider.json

+[
+  "dubbox-provider"
+]
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/dubbox-consumer.json

+[
+  {
+    "gt": [
+      [
+        230150,
+        185809,
+        24040000
+      ]
+    ],
+    "sg": {
+      "ts": [
+        230150,
+        185809,
+        24040000
+      ],
+      "ai": -1,
+      "ii": 1,
+      "rs": [],
+      "ss": [
+        {
+          "si": 1,
+          "tv": 1,
+          "lv": 1,
+          "ps": 0,
+          "st": 1501858094526,
+          "et": 1501858097004,
+          "ci": 3,
+          "cn": "",
+          "oi": 0,
+          "on": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()",
+          "pi": 0,
+          "pn": "172.25.0.4:20880",
+          "ie": false,
+          "to": [
+            {
+              "k": "url",
+              "v": "rest://172.25.0.4:20880/org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"
+            }
+          ],
+          "lo": []
+        },
+        {
+          "si": 0,
+          "tv": 0,
+          "lv": 2,
+          "ps": -1,
+          "st": 1501858092409,
+          "et": 1501858097033,
+          "ci": 1,
+          "cn": "",
+          "oi": 0,
+          "on": "/dubbox-case/case/dubbox-rest",
+          "pi": 0,
+          "pn": "",
+          "ie": false,
+          "to": [
+            {
+              "k": "url",
+              "v": "http://localhost:18080/dubbox-case/case/dubbox-rest"
+            },
+            {
+              "k": "http.method",
+              "v": "GET"
+            }
+          ],
+          "lo": []
+        }
+      ]
+    }
+  }
+]
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/dubbox-provider.json

+[
+  {
+    "gt": [
+      [
+        230150,
+        185809,
+        24040000
+      ]
+    ],
+    "sg": {
+      "ts": [
+        137150,
+        185809,
+        48780000
+      ],
+      "ai": 2,
+      "ii": 2,
+      "ss": [
+        {
+          "si": 0,
+          "tv": 0,
+          "lv": 2,
+          "ps": -1,
+          "st": 1501858094726,
+          "et": 1501858096804,
+          "ci": 3,
+          "cn": "",
+          "oi": 0,
+          "on": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()",
+          "pi": 0,
+          "pn": "",
+          "ie": false,
+          "rs": [
+            {
+              "ts": [
+                230150,
+                185809,
+                24040000
+              ],
+              "ai": -1,
+              "si": 1,
+              "vi": 0,
+              "vn": "/dubbox-case/case/dubbox-rest",
+              "ni": 0,
+              "nn": "172.25.0.4:20880",
+              "ea": 2,
+              "ei": 0,
+              "en": "/dubbox-case/case/dubbox-rest",
+              "rn": 0
+            }
+          ],
+          "to": [
+            {
+              "k": "url",
+              "v": "rest://172.25.0.4:20880/org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"
+            },
+            {
+              "k": "http.method",
+              "v": "GET"
+            }
+          ],
+          "lo": []
+        }
+      ]
+    }
+  }
+]
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/instance-register-consumer.json

+{
+  "ai": -1,
+  "au": "dubbox-consumer",
+  "rt": 1501858094526,
+  "oi": {
+    "any_name": "any_value",
+    "any_name1": "any_value1"
+  }
+}
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/instance-register-provider.json

+{
+  "ai": 2,
+  "au": "dubbox-provider",
+  "rt": 1501858094526,
+  "oi": {
+    "any_name": "any_value",
+    "any_name1": "any_value1"
+  }
+}
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/servicename-register-consumer.json

+[
+  {
+    "ai": -1,
+    "sn": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"
+  }
+]
\ No newline at end of file

apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/servicename-register-provider.json

+[
+  {
+    "ai": 2,
+    "sn": "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"
+  }
+]
\ No newline at end of file

apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/AlertingModuleProvider.java

@@ -50,6 +50,6 @@ public class AlertingModuleProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[] {};
+        return new String[0];
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/AnalysisJVMModuleProvider.java

@@ -35,9 +35,13 @@ import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.GCMetric
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.InstanceHeartBeatPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.MemoryMetricPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.MemoryPoolMetricPersistenceGraph;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
+import org.apache.skywalking.apm.collector.analysis.worker.timer.PersistenceTimer;
 import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
 
 /**
  * @author peng-yongsheng
@@ -63,20 +67,12 @@ public class AnalysisJVMModuleProvider extends ModuleProvider {
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
-        CpuMetricPersistenceGraph cpuMetricPersistenceGraph = new CpuMetricPersistenceGraph(getManager());
-        cpuMetricPersistenceGraph.create();
+        WorkerCreateListener workerCreateListener = new WorkerCreateListener();
 
-        GCMetricPersistenceGraph gcMetricPersistenceGraph = new GCMetricPersistenceGraph(getManager());
-        gcMetricPersistenceGraph.create();
+        graphCreate(workerCreateListener);
 
-        InstanceHeartBeatPersistenceGraph instanceHeartBeatPersistenceGraph = new InstanceHeartBeatPersistenceGraph(getManager());
-        instanceHeartBeatPersistenceGraph.create();
-
-        MemoryMetricPersistenceGraph memoryMetricPersistenceGraph = new MemoryMetricPersistenceGraph(getManager());
-        memoryMetricPersistenceGraph.create();
-
-        MemoryPoolMetricPersistenceGraph memoryPoolMetricPersistenceGraph = new MemoryPoolMetricPersistenceGraph(getManager());
-        memoryPoolMetricPersistenceGraph.create();
+        PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisJVMModule.NAME);
+        persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@@ -84,6 +80,23 @@ public class AnalysisJVMModuleProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[0];
+        return new String[] {StorageModule.NAME, RemoteModule.NAME};
+    }
+
+    private void graphCreate(WorkerCreateListener workerCreateListener) {
+        CpuMetricPersistenceGraph cpuMetricPersistenceGraph = new CpuMetricPersistenceGraph(getManager(), workerCreateListener);
+        cpuMetricPersistenceGraph.create();
+
+        GCMetricPersistenceGraph gcMetricPersistenceGraph = new GCMetricPersistenceGraph(getManager(), workerCreateListener);
+        gcMetricPersistenceGraph.create();
+
+        InstanceHeartBeatPersistenceGraph instanceHeartBeatPersistenceGraph = new InstanceHeartBeatPersistenceGraph(getManager(), workerCreateListener);
+        instanceHeartBeatPersistenceGraph.create();
+
+        MemoryMetricPersistenceGraph memoryMetricPersistenceGraph = new MemoryMetricPersistenceGraph(getManager(), workerCreateListener);
+        memoryMetricPersistenceGraph.create();
+
+        MemoryPoolMetricPersistenceGraph memoryPoolMetricPersistenceGraph = new MemoryPoolMetricPersistenceGraph(getManager(), workerCreateListener);
+        memoryPoolMetricPersistenceGraph.create();
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/CpuMetricPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 public class CpuMetricPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public CpuMetricPersistenceGraph(ModuleManager moduleManager) {
+    public CpuMetricPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.CPU_METRIC_PERSISTENCE_GRAPH_ID, CpuMetric.class)
-            .addNode(new CpuMetricPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new CpuMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/GCMetricPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.jvm.GCMetric;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.GCMetric;
 public class GCMetricPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public GCMetricPersistenceGraph(ModuleManager moduleManager) {
+    public GCMetricPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.GC_METRIC_PERSISTENCE_GRAPH_ID, GCMetric.class)
-            .addNode(new GCMetricPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new GCMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/InstanceHeartBeatPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.register.Instance;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.register.Instance;
 public class InstanceHeartBeatPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public InstanceHeartBeatPersistenceGraph(ModuleManager moduleManager) {
+    public InstanceHeartBeatPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_HEART_BEAT_PERSISTENCE_GRAPH_ID, Instance.class)
-            .addNode(new InstHeartBeatPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new InstHeartBeatPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/MemoryMetricPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 public class MemoryMetricPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public MemoryMetricPersistenceGraph(ModuleManager moduleManager) {
+    public MemoryMetricPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.MEMORY_METRIC_PERSISTENCE_GRAPH_ID, MemoryMetric.class)
-            .addNode(new MemoryMetricPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new MemoryMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/MemoryPoolMetricPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
 public class MemoryPoolMetricPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public MemoryPoolMetricPersistenceGraph(ModuleManager moduleManager) {
+    public MemoryPoolMetricPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.MEMORY_POOL_METRIC_PERSISTENCE_GRAPH_ID, MemoryPoolMetric.class)
-            .addNode(new MemoryPoolMetricPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new MemoryPoolMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java

@@ -39,6 +39,8 @@ import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.servi
 import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricSpanListener;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.AnalysisSegmentParserModule;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParserListenerRegister;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
+import org.apache.skywalking.apm.collector.analysis.worker.timer.PersistenceTimer;
 import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
@@ -64,55 +66,64 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
     @Override public void start(Properties config) throws ServiceNotProvidedException {
         segmentParserListenerRegister();
 
-        ServiceReferenceMetricGraph serviceReferenceMetricGraph = new ServiceReferenceMetricGraph(getManager());
+        WorkerCreateListener workerCreateListener = new WorkerCreateListener();
+
+        graphCreate(workerCreateListener);
+
+        PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisMetricModule.NAME);
+        persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
+    }
+
+    @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
+
+    }
+
+    @Override public String[] requiredModules() {
+        return new String[] {AnalysisSegmentParserModule.NAME};
+    }
+
+    private void segmentParserListenerRegister() {
+        ISegmentParserListenerRegister segmentParserListenerRegister = getManager().find(AnalysisSegmentParserModule.NAME).getService(ISegmentParserListenerRegister.class);
+        segmentParserListenerRegister.register(new ServiceReferenceMetricSpanListener());
+        segmentParserListenerRegister.register(new ApplicationComponentSpanListener());
+        segmentParserListenerRegister.register(new ApplicationMappingSpanListener());
+        segmentParserListenerRegister.register(new ServiceEntrySpanListener(getManager()));
+        segmentParserListenerRegister.register(new GlobalTraceSpanListener());
+        segmentParserListenerRegister.register(new SegmentCostSpanListener(getManager()));
+    }
+
+    private void graphCreate(WorkerCreateListener workerCreateListener) {
+        ServiceReferenceMetricGraph serviceReferenceMetricGraph = new ServiceReferenceMetricGraph(getManager(), workerCreateListener);
         serviceReferenceMetricGraph.create();
 
-        InstanceReferenceMetricGraph instanceReferenceMetricGraph = new InstanceReferenceMetricGraph(getManager());
+        InstanceReferenceMetricGraph instanceReferenceMetricGraph = new InstanceReferenceMetricGraph(getManager(), workerCreateListener);
         instanceReferenceMetricGraph.create();
 
-        ApplicationReferenceMetricGraph applicationReferenceMetricGraph = new ApplicationReferenceMetricGraph(getManager());
+        ApplicationReferenceMetricGraph applicationReferenceMetricGraph = new ApplicationReferenceMetricGraph(getManager(), workerCreateListener);
         applicationReferenceMetricGraph.create();
 
-        ServiceMetricGraph serviceMetricGraph = new ServiceMetricGraph(getManager());
+        ServiceMetricGraph serviceMetricGraph = new ServiceMetricGraph(getManager(), workerCreateListener);
         serviceMetricGraph.create();
 
-        InstanceMetricGraph instanceMetricGraph = new InstanceMetricGraph(getManager());
+        InstanceMetricGraph instanceMetricGraph = new InstanceMetricGraph(getManager(), workerCreateListener);
         instanceMetricGraph.create();
 
-        ApplicationMetricGraph applicationMetricGraph = new ApplicationMetricGraph(getManager());
+        ApplicationMetricGraph applicationMetricGraph = new ApplicationMetricGraph(getManager(), workerCreateListener);
         applicationMetricGraph.create();
 
-        ApplicationComponentGraph applicationComponentGraph = new ApplicationComponentGraph(getManager());
+        ApplicationComponentGraph applicationComponentGraph = new ApplicationComponentGraph(getManager(), workerCreateListener);
         applicationComponentGraph.create();
 
-        ApplicationMappingGraph applicationMappingGraph = new ApplicationMappingGraph(getManager());
+        ApplicationMappingGraph applicationMappingGraph = new ApplicationMappingGraph(getManager(), workerCreateListener);
         applicationMappingGraph.create();
 
-        ServiceEntryGraph serviceEntryGraph = new ServiceEntryGraph(getManager());
+        ServiceEntryGraph serviceEntryGraph = new ServiceEntryGraph(getManager(), workerCreateListener);
         serviceEntryGraph.create();
 
-        GlobalTraceGraph globalTraceGraph = new GlobalTraceGraph(getManager());
+        GlobalTraceGraph globalTraceGraph = new GlobalTraceGraph(getManager(), workerCreateListener);
         globalTraceGraph.create();
 
-        SegmentCostGraph segmentCostGraph = new SegmentCostGraph(getManager());
+        SegmentCostGraph segmentCostGraph = new SegmentCostGraph(getManager(), workerCreateListener);
         segmentCostGraph.create();
     }
-
-    @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
-
-    }
-
-    @Override public String[] requiredModules() {
-        return new String[0];
-    }
-
-    private void segmentParserListenerRegister() {
-        ISegmentParserListenerRegister segmentParserListenerRegister = getManager().find(AnalysisSegmentParserModule.NAME).getService(ISegmentParserListenerRegister.class);
-        segmentParserListenerRegister.register(new ServiceReferenceMetricSpanListener());
-        segmentParserListenerRegister.register(new ApplicationComponentSpanListener());
-        segmentParserListenerRegister.register(new ApplicationMappingSpanListener());
-        segmentParserListenerRegister.register(new ServiceEntrySpanListener(getManager()));
-        segmentParserListenerRegister.register(new GlobalTraceSpanListener());
-        segmentParserListenerRegister.register(new SegmentCostSpanListener(getManager()));
-    }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,17 +32,19 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application
 public class ApplicationComponentGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ApplicationComponentGraph(ModuleManager moduleManager) {
+    public ApplicationComponentGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class)
-            .addNode(new ApplicationComponentAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ApplicationComponentRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID).create(null))
-            .addNext(new ApplicationComponentPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new ApplicationComponentAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ApplicationComponentRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ApplicationComponentPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,17 +32,19 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application
 public class ApplicationMappingGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ApplicationMappingGraph(ModuleManager moduleManager) {
+    public ApplicationMappingGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class)
-            .addNode(new ApplicationMappingAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ApplicationMappingRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID).create(null))
-            .addNext(new ApplicationMappingPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new ApplicationMappingAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ApplicationMappingRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ApplicationMappingPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,17 +32,19 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application
 public class ApplicationMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ApplicationMetricGraph(ModuleManager moduleManager) {
+    public ApplicationMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_METRIC_GRAPH_ID, ApplicationReferenceMetric.class)
-            .addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(null))
-            .addNext(new ApplicationMetricPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ApplicationMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricGraph.java

@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.appl
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.graph.Next;
@@ -35,9 +36,11 @@ import org.apache.skywalking.apm.collector.storage.table.instance.InstanceRefere
 public class ApplicationReferenceMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ApplicationReferenceMetricGraph(ModuleManager moduleManager) {
+    public ApplicationReferenceMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
@@ -45,9 +48,9 @@ public class ApplicationReferenceMetricGraph {
 
         Graph<InstanceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class);
 
-        graph.addNode(new ApplicationReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ApplicationReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID).create(null))
-            .addNext(new ApplicationReferenceMetricPersistenceWorker.Factory(moduleManager).create(null));
+        graph.addNode(new ApplicationReferenceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ApplicationReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ApplicationReferenceMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
         link(graph);
     }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
 public class GlobalTraceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public GlobalTraceGraph(ModuleManager moduleManager) {
+    public GlobalTraceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class)
-            .addNode(new GlobalTracePersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new GlobalTracePersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricGraph.java

@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.inst
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.graph.Next;
@@ -35,9 +36,11 @@ import org.apache.skywalking.apm.collector.storage.table.instance.InstanceRefere
 public class InstanceMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public InstanceMetricGraph(ModuleManager moduleManager) {
+    public InstanceMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
@@ -45,9 +48,9 @@ public class InstanceMetricGraph {
 
         Graph<InstanceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class);
 
-        graph.addNode(new InstanceMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new InstanceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_METRIC_GRAPH_ID).create(null))
-            .addNext(new InstanceMetricPersistenceWorker.Factory(moduleManager).create(null));
+        graph.addNode(new InstanceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new InstanceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new InstanceMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
         link(graph);
     }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricGraph.java

@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.inst
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.graph.Next;
@@ -35,9 +36,11 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenc
 public class InstanceReferenceMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public InstanceReferenceMetricGraph(ModuleManager moduleManager) {
+    public InstanceReferenceMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
@@ -45,9 +48,9 @@ public class InstanceReferenceMetricGraph {
 
         Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
 
-        graph.addNode(new InstanceReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new InstanceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID).create(null))
-            .addNext(new InstanceReferencePersistenceWorker.Factory(moduleManager).create(null));
+        graph.addNode(new InstanceReferenceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new InstanceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new InstanceReferencePersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
         link(graph);
     }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
 public class SegmentCostGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public SegmentCostGraph(ModuleManager moduleManager) {
+    public SegmentCostGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class)
-            .addNode(new SegmentCostPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new SegmentCostPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,17 +32,19 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
 public class ServiceEntryGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ServiceEntryGraph(ModuleManager moduleManager) {
+    public ServiceEntryGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class)
-            .addNode(new ServiceEntryAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ServiceEntryRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_ENTRY_GRAPH_ID).create(null))
-            .addNext(new ServiceEntryPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new ServiceEntryAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ServiceEntryRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_ENTRY_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceEntryPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java

@@ -20,6 +20,7 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.serv
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.Graph;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.graph.Next;
@@ -35,9 +36,11 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenc
 public class ServiceMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ServiceMetricGraph(ModuleManager moduleManager) {
+    public ServiceMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
@@ -45,9 +48,9 @@ public class ServiceMetricGraph {
 
         Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
 
-        graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(null))
-            .addNext(new ServiceMetricPersistenceWorker.Factory(moduleManager).create(null));
+        graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
 
         link(graph);
     }

apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
 
 import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,17 +32,19 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenc
 public class ServiceReferenceMetricGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ServiceReferenceMetricGraph(ModuleManager moduleManager) {
+    public ServiceReferenceMetricGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class)
-            .addNode(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
-            .addNext(new ServiceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID).create(null))
-            .addNext(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null));
+            .addNode(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener))
+            .addNext(new ServiceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceReferenceMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/AnalysisRegisterModuleProvider.java

@@ -29,9 +29,14 @@ import org.apache.skywalking.apm.collector.analysis.register.provider.register.S
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.ApplicationIDService;
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.InstanceIDService;
 import org.apache.skywalking.apm.collector.analysis.register.provider.service.ServiceNameService;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
+import org.apache.skywalking.apm.collector.analysis.worker.timer.PersistenceTimer;
+import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
+import org.apache.skywalking.apm.collector.remote.RemoteModule;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
 
 /**
  * @author peng-yongsheng
@@ -55,14 +60,12 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider {
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
-        ApplicationRegisterGraph applicationRegisterGraph = new ApplicationRegisterGraph(getManager());
-        applicationRegisterGraph.create();
+        WorkerCreateListener workerCreateListener = new WorkerCreateListener();
 
-        InstanceRegisterGraph instanceRegisterGraph = new InstanceRegisterGraph(getManager());
-        instanceRegisterGraph.create();
+        graphCreate(workerCreateListener);
 
-        ServiceNameRegisterGraph serviceNameRegisterGraph = new ServiceNameRegisterGraph(getManager());
-        serviceNameRegisterGraph.create();
+        PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisRegisterModule.NAME);
+        persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
     }
 
     @Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@@ -70,6 +73,17 @@ public class AnalysisRegisterModuleProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[0];
+        return new String[] {StorageModule.NAME, RemoteModule.NAME, CacheModule.NAME};
+    }
+
+    private void graphCreate(WorkerCreateListener workerCreateListener) {
+        ApplicationRegisterGraph applicationRegisterGraph = new ApplicationRegisterGraph(getManager(), workerCreateListener);
+        applicationRegisterGraph.create();
+
+        InstanceRegisterGraph instanceRegisterGraph = new InstanceRegisterGraph(getManager(), workerCreateListener);
+        instanceRegisterGraph.create();
+
+        ServiceNameRegisterGraph serviceNameRegisterGraph = new ServiceNameRegisterGraph(getManager(), workerCreateListener);
+        serviceNameRegisterGraph.create();
     }
 }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ApplicationRegisterGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
 import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,16 +32,18 @@ import org.apache.skywalking.apm.collector.storage.table.register.Application;
 public class ApplicationRegisterGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ApplicationRegisterGraph(ModuleManager moduleManager) {
+    public ApplicationRegisterGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID, Application.class)
-            .addNode(new ApplicationRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID).create(null))
-            .addNext(new ApplicationRegisterSerialWorker.Factory(moduleManager).create(null));
+            .addNode(new ApplicationRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REGISTER_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ApplicationRegisterSerialWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/InstanceRegisterGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
 import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,16 +32,18 @@ import org.apache.skywalking.apm.collector.storage.table.register.Instance;
 public class InstanceRegisterGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public InstanceRegisterGraph(ModuleManager moduleManager) {
+    public InstanceRegisterGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID, Instance.class)
-            .addNode(new InstanceRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID).create(null))
-            .addNext(new InstanceRegisterSerialWorker.Factory(moduleManager).create(null));
+            .addNode(new InstanceRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REGISTER_GRAPH_ID).create(workerCreateListener))
+            .addNext(new InstanceRegisterSerialWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-register/register-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/register/provider/register/ServiceNameRegisterGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.register.provider.register;
 
 import org.apache.skywalking.apm.collector.analysis.register.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.remote.RemoteModule;
@@ -31,16 +32,18 @@ import org.apache.skywalking.apm.collector.storage.table.register.ServiceName;
 public class ServiceNameRegisterGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public ServiceNameRegisterGraph(ModuleManager moduleManager) {
+    public ServiceNameRegisterGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
 
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID, ServiceName.class)
-            .addNode(new ServiceNameRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID).create(null))
-            .addNext(new ServiceNameRegisterSerialWorker.Factory(moduleManager).create(null));
+            .addNode(new ServiceNameRegisterRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_NAME_REGISTER_GRAPH_ID).create(workerCreateListener))
+            .addNext(new ServiceNameRegisterSerialWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/AnalysisSegmentParserModuleProvider.java

@@ -19,18 +19,24 @@
 package org.apache.skywalking.apm.collector.analysis.segment.parser.provider;
 
 import java.util.Properties;
+import org.apache.skywalking.apm.collector.analysis.register.define.AnalysisRegisterModule;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.AnalysisSegmentParserModule;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParseService;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParserListenerRegister;
+import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.buffer.BufferFileConfig;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.buffer.SegmentBufferReader;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.SegmentParserListenerManager;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.SegmentPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.standardization.SegmentStandardizationGraph;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.service.SegmentParseService;
 import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.service.SegmentParserListenerRegister;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
+import org.apache.skywalking.apm.collector.analysis.worker.timer.PersistenceTimer;
+import org.apache.skywalking.apm.collector.cache.CacheModule;
 import org.apache.skywalking.apm.collector.core.module.Module;
 import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
 import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
 
 /**
  * @author peng-yongsheng
@@ -52,13 +58,18 @@ public class AnalysisSegmentParserModuleProvider extends ModuleProvider {
         this.listenerManager = new SegmentParserListenerManager();
         this.registerServiceImplementation(ISegmentParserListenerRegister.class, new SegmentParserListenerRegister(listenerManager));
         this.registerServiceImplementation(ISegmentParseService.class, new SegmentParseService(getManager(), listenerManager));
+
+        BufferFileConfig.Parser parser = new BufferFileConfig.Parser();
+        parser.parse(config);
     }
 
     @Override public void start(Properties config) throws ServiceNotProvidedException {
-        SegmentPersistenceGraph segmentPersistenceGraph = new SegmentPersistenceGraph(getManager());
-        segmentPersistenceGraph.create();
-        SegmentStandardizationGraph segmentStandardizationGraph = new SegmentStandardizationGraph(getManager());
-        segmentStandardizationGraph.create();
+        WorkerCreateListener workerCreateListener = new WorkerCreateListener();
+
+        graphCreate(workerCreateListener);
+
+        PersistenceTimer persistenceTimer = new PersistenceTimer(AnalysisSegmentParserModule.NAME);
+        persistenceTimer.start(getManager(), workerCreateListener.getPersistenceWorkers());
 
         SegmentBufferReader.INSTANCE.setSegmentParserListenerManager(listenerManager);
     }
@@ -68,6 +79,14 @@ public class AnalysisSegmentParserModuleProvider extends ModuleProvider {
     }
 
     @Override public String[] requiredModules() {
-        return new String[0];
+        return new String[] {AnalysisRegisterModule.NAME, CacheModule.NAME, StorageModule.NAME};
+    }
+
+    private void graphCreate(WorkerCreateListener workerCreateListener) {
+        SegmentPersistenceGraph segmentPersistenceGraph = new SegmentPersistenceGraph(getManager(), workerCreateListener);
+        segmentPersistenceGraph.create();
+
+        SegmentStandardizationGraph segmentStandardizationGraph = new SegmentStandardizationGraph(getManager(), workerCreateListener);
+        segmentStandardizationGraph.create();
     }
 }

apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentPersistenceGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser;
 
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.segment.Segment;
@@ -29,13 +30,15 @@ import org.apache.skywalking.apm.collector.storage.table.segment.Segment;
 public class SegmentPersistenceGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public SegmentPersistenceGraph(ModuleManager moduleManager) {
+    public SegmentPersistenceGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SEGMENT_PERSISTENCE_GRAPH_ID, Segment.class)
-            .addNode(new SegmentPersistenceWorker.Factory(moduleManager).create(null));
+            .addNode(new SegmentPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/standardization/SegmentStandardizationGraph.java

@@ -19,6 +19,7 @@
 package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.standardization;
 
 import org.apache.skywalking.apm.collector.analysis.segment.parser.define.graph.GraphIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
 import org.apache.skywalking.apm.collector.core.graph.GraphManager;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 
@@ -28,13 +29,15 @@ import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 public class SegmentStandardizationGraph {
 
     private final ModuleManager moduleManager;
+    private final WorkerCreateListener workerCreateListener;
 
-    public SegmentStandardizationGraph(ModuleManager moduleManager) {
+    public SegmentStandardizationGraph(ModuleManager moduleManager, WorkerCreateListener workerCreateListener) {
         this.moduleManager = moduleManager;
+        this.workerCreateListener = workerCreateListener;
     }
 
     public void create() {
         GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SEGMENT_STANDARDIZATION_GRAPH_ID, SegmentStandardization.class)
-            .addNode(new SegmentStandardizationWorker.Factory(moduleManager).create(null));
+            .addNode(new SegmentStandardizationWorker.Factory(moduleManager).create(workerCreateListener));
     }
 }

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorker.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
+import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 
 /**
@@ -27,7 +28,7 @@ import org.apache.skywalking.apm.collector.core.module.ModuleManager;
  * @author peng-yongsheng
  * @since v3.0-2017
  */
-public abstract class AbstractLocalAsyncWorker<INPUT, OUTPUT> extends AbstractWorker<INPUT, OUTPUT> {
+public abstract class AbstractLocalAsyncWorker<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> extends AbstractWorker<INPUT, OUTPUT> {
 
     public AbstractLocalAsyncWorker(ModuleManager moduleManager) {
         super(moduleManager);

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractLocalAsyncWorkerProvider.java

@@ -18,13 +18,14 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
+import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
 
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractLocalAsyncWorkerProvider<INPUT, OUTPUT, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
+public abstract class AbstractLocalAsyncWorkerProvider<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage, WORKER_TYPE extends AbstractLocalAsyncWorker<INPUT, OUTPUT>> extends AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE> {
 
     public abstract int queueSize();
 
@@ -40,6 +41,7 @@ public abstract class AbstractLocalAsyncWorkerProvider<INPUT, OUTPUT, WORKER_TYP
         LocalAsyncWorkerRef<INPUT, OUTPUT> localAsyncWorkerRef = new LocalAsyncWorkerRef<>(localAsyncWorker);
         DataCarrier<INPUT> dataCarrier = new DataCarrier<>(1, queueSize());
         localAsyncWorkerRef.setQueueEventHandler(dataCarrier);
+        dataCarrier.consume(localAsyncWorkerRef, 1);
         return localAsyncWorkerRef;
     }
 }

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorker.java

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
+import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.graph.Next;
 import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
@@ -27,7 +28,7 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractWorker<INPUT, OUTPUT> implements NodeProcessor<INPUT, OUTPUT> {
+public abstract class AbstractWorker<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> implements NodeProcessor<INPUT, OUTPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(AbstractWorker.class);
 

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/AbstractWorkerProvider.java

@@ -18,12 +18,13 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
+import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 
 /**
  * @author peng-yongsheng
  */
-public abstract class AbstractWorkerProvider<INPUT, OUTPUT, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
+public abstract class AbstractWorkerProvider<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage, WORKER_TYPE extends AbstractWorker<INPUT, OUTPUT>> implements Provider {
 
     private final ModuleManager moduleManager;
 

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/base/LocalAsyncWorkerRef.java

@@ -18,7 +18,9 @@
 
 package org.apache.skywalking.apm.collector.analysis.worker.model.base;
 
+import java.util.Iterator;
 import java.util.List;
+import org.apache.skywalking.apm.collector.core.data.EndOfBatchQueueMessage;
 import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
 import org.apache.skywalking.apm.commons.datacarrier.DataCarrier;
 import org.apache.skywalking.apm.commons.datacarrier.consumer.IConsumer;
@@ -28,7 +30,7 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public class LocalAsyncWorkerRef<INPUT, OUTPUT> extends WorkerRef<INPUT, OUTPUT> implements IConsumer<INPUT> {
+public class LocalAsyncWorkerRef<INPUT extends EndOfBatchQueueMessage, OUTPUT extends EndOfBatchQueueMessage> extends WorkerRef<INPUT, OUTPUT> implements IConsumer<INPUT> {
 
     private final Logger logger = LoggerFactory.getLogger(LocalAsyncWorkerRef.class);
 
@@ -43,7 +45,17 @@ public class LocalAsyncWorkerRef<INPUT, OUTPUT> extends WorkerRef<INPUT, OUTPUT>
     }
 
     @Override public void consume(List<INPUT> data) {
-        data.forEach(this::out);
+        Iterator<INPUT> inputIterator = data.iterator();
+
+        int i = 0;
+        while (inputIterator.hasNext()) {
+            INPUT input = inputIterator.next();
+            i++;
+            if (i == data.size()) {
+                input.setEndOfBatch(true);
+            }
+            out(input);
+        }
     }
 
     @Override public void init() {

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/model/impl/AggregationWorker.java

@@ -47,7 +47,10 @@ public abstract class AggregationWorker<INPUT extends Data, OUTPUT extends Data>
     }
 
     @Override protected final void onWork(INPUT message) throws WorkerException {
+        boolean isEndOfBatch = message.isEndOfBatch();
         OUTPUT output = transform(message);
+        output.setEndOfBatch(isEndOfBatch);
+
         messageNum++;
         aggregate(output);
 
@@ -56,6 +59,7 @@ public abstract class AggregationWorker<INPUT extends Data, OUTPUT extends Data>
             messageNum = 0;
         }
         if (output.isEndOfBatch()) {
+            output.setEndOfBatch(false);
             sendToNext();
         }
     }

apm-collector/apm-collector-analysis/analysis-worker-model/src/main/java/org/apache/skywalking/apm/collector/analysis/worker/timer/PersistenceTimer.java

@@ -35,6 +35,12 @@ import org.slf4j.LoggerFactory;
  */
 public class PersistenceTimer {
 
+    private final String belongsToModuleName;
+
+    public PersistenceTimer(String belongsToModuleName) {
+        this.belongsToModuleName = belongsToModuleName;
+    }
+
     private final Logger logger = LoggerFactory.getLogger(PersistenceTimer.class);
 
     public void start(ModuleManager moduleManager, List<PersistenceWorker> persistenceWorkers) {
@@ -63,7 +69,7 @@ public class PersistenceTimer {
 
             batchDAO.batchPersistence(batchAllCollection);
         } catch (Throwable e) {
-            logger.error(e.getMessage(), e);
+            logger.error("The persistence timer belongs to module name: " + belongsToModuleName + ", error message: " + e.getMessage(), e);
         } finally {
             logger.debug("persistence data save finish");
         }

apm-collector/apm-collector-boot/pom.xml

@@ -98,6 +98,38 @@
             <version>${project.version}</version>
         </dependency>
         <!-- agent jetty provider -->
+        <!-- analysis provider -->
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>segment-parser-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>register-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>jvm-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>metric-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>alarm-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.skywalking</groupId>
+            <artifactId>baseline-computing-provider</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <!-- analysis provider -->
         <!-- ui provider -->
         <dependency>
             <groupId>org.apache.skywalking</groupId>

apm-collector/apm-collector-boot/src/main/resources/application.yml

@@ -20,7 +20,11 @@ agent_jetty:
     host: localhost
     port: 12800
     context_path: /
-agent_stream:
+analysis_register:
+  default:
+analysis_jvm:
+  default:
+analysis_segment_parser:
   default:
     buffer_file_path: ../buffer/
     buffer_offset_max_file_size: 10M

apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/module/BootstrapFlow.java

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.core.module;
 
 import java.util.ArrayList;
@@ -114,11 +113,11 @@ public class BootstrapFlow {
             }
 
             if (numOfToBeSequenced == allProviders.size()) {
-                StringBuilder unsequencedProviders = new StringBuilder();
+                StringBuilder unSequencedProviders = new StringBuilder();
                 allProviders.forEach(provider -> {
-                    unsequencedProviders.append(provider.getModuleName()).append("[provider=").append(provider.getClass().getName()).append("]\n");
+                    unSequencedProviders.append(provider.getModuleName()).append("[provider=").append(provider.getClass().getName()).append("]\n");
                 });
-                throw new CycleDependencyException("Exist cycle module dependencies in \n" + unsequencedProviders.substring(0, unsequencedProviders.length() - 1));
+                throw new CycleDependencyException("Exist cycle module dependencies in \n" + unSequencedProviders.substring(0, unSequencedProviders.length() - 1));
             }
             if (allProviders.size() == 0) {
                 break;

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

@@ -4,8 +4,6 @@ cluster:
     user_name: sa
 cache:
   guava:
-queue:
-  disruptor:
 naming:
   jetty:
     host: localhost
@@ -24,14 +22,16 @@ agent_jetty:
     host: localhost
     port: 12800
     context_path: /
-agent_stream:
+analysis_segment_parser:
   default:
     buffer_file_path: ../buffer/
     buffer_offset_max_file_size: 10M
     buffer_segment_max_file_size: 500M
-configuration:
+analysis_metric:
+  default:
+analysis_alarm:
   default:
-alerting:
+configuration:
   default:
 ui:
   jetty:
@@ -39,9 +39,9 @@ ui:
     port: 12800
     context_path: /
 jetty_manager:
-  jetty:
+  default:
 gRPC_manager:
-  gRPC:
+  default:
 storage:
   h2:
     url: jdbc:h2:~/memorydb

apm-collector/apm-collector-grpc-manager/collector-grpc-manager-provider/src/main/java/org/apache/skywalking/apm/collector/grpc/manager/GRPCManagerProvider.java

@@ -42,7 +42,7 @@ public class GRPCManagerProvider extends ModuleProvider {
     private Map<String, GRPCServer> servers = new HashMap<>();
 
     @Override public String name() {
-        return "gRPC";
+        return "default";
     }
 
     @Override public Class<? extends Module> module() {

apm-collector/apm-collector-jetty-manager/collector-jetty-manager-provider/src/main/java/org/apache/skywalking/apm/collector/jetty/manager/JettyManagerProvider.java

@@ -42,7 +42,7 @@ public class JettyManagerProvider extends ModuleProvider {
     private Map<String, JettyServer> servers = new HashMap<>();
 
     @Override public String name() {
-        return "jetty";
+        return "default";
     }
 
     @Override public Class<? extends Module> module() {

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

@@ -16,18 +16,21 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage;
 
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.skywalking.apm.collector.core.module.Module;
+import org.apache.skywalking.apm.collector.storage.base.dao.IBatchDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IAlertingListPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationCacheDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricUIDAO;
@@ -39,13 +42,16 @@ import org.apache.skywalking.apm.collector.storage.dao.IInstanceCacheDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricUIDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO;
+import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryUIDAO;
@@ -54,12 +60,6 @@ import org.apache.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceUIDAO;
-import org.apache.skywalking.apm.collector.storage.base.dao.IBatchDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IAlertingListPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricUIDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IInstanceUIDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO;
 
 /**
  * @author peng-yongsheng
@@ -104,7 +104,6 @@ public class StorageModule extends Module {
         classes.add(IMemoryPoolMetricPersistenceDAO.class);
 
         classes.add(IGlobalTracePersistenceDAO.class);
-        classes.add(IInstanceMetricPersistenceDAO.class);
         classes.add(IApplicationComponentPersistenceDAO.class);
         classes.add(IApplicationMappingPersistenceDAO.class);
         classes.add(IApplicationMetricPersistenceDAO.class);
@@ -115,6 +114,8 @@ public class StorageModule extends Module {
         classes.add(IServiceMetricPersistenceDAO.class);
         classes.add(IServiceReferenceMetricPersistenceDAO.class);
 
+        classes.add(IInstanceMetricPersistenceDAO.class);
+        classes.add(IInstanceReferenceMetricPersistenceDAO.class);
         classes.add(IInstanceHeartBeatPersistenceDAO.class);
     }
 

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IInstanceReferenceMetricPersistenceDAO.java

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.dao;
 
 import org.apache.skywalking.apm.collector.core.data.Data;

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IServiceReferenceMetricPersistenceDAO.java

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.dao;
 
 import org.apache.skywalking.apm.collector.core.data.Data;

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

@@ -16,17 +16,11 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.es;
 
 import java.util.Properties;
 import java.util.UUID;
 import org.apache.skywalking.apm.collector.client.ClientException;
-import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO;
-import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationEsCacheDAO;
-import org.apache.skywalking.apm.collector.storage.es.dao.SegmentCostEsUIDAO;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
 import org.apache.skywalking.apm.collector.cluster.ClusterModule;
 import org.apache.skywalking.apm.collector.cluster.service.ModuleListenerService;
@@ -55,13 +49,17 @@ import org.apache.skywalking.apm.collector.storage.dao.IGlobalTracePersistenceDA
 import org.apache.skywalking.apm.collector.storage.dao.IGlobalTraceUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceCacheDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricUIDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO;
@@ -76,6 +74,7 @@ import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchS
 import org.apache.skywalking.apm.collector.storage.es.dao.AlertingListEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationComponentEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationComponentEsUIDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationEsCacheDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationEsRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationMappingEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationMappingEsUIDAO;
@@ -94,11 +93,13 @@ import org.apache.skywalking.apm.collector.storage.es.dao.InstanceEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.InstanceHeartBeatEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.InstanceMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.InstanceMetricEsUIDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.InstanceReferenceMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.MemoryMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.MemoryMetricEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.MemoryPoolMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.MemoryPoolMetricEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.SegmentCostEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.SegmentCostEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.SegmentEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.SegmentEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ServiceEntryEsPersistenceDAO;
@@ -202,7 +203,6 @@ public class StorageModuleEsProvider extends ModuleProvider {
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricEsPersistenceDAO(elasticSearchClient));
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceEsPersistenceDAO(elasticSearchClient));
-        this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IApplicationComponentPersistenceDAO.class, new ApplicationComponentEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IApplicationMappingPersistenceDAO.class, new ApplicationMappingEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IApplicationMetricPersistenceDAO.class, new ApplicationMetricEsPersistenceDAO(elasticSearchClient));
@@ -213,6 +213,8 @@ public class StorageModuleEsProvider extends ModuleProvider {
         this.registerServiceImplementation(IServiceMetricPersistenceDAO.class, new ServiceMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IServiceReferenceMetricPersistenceDAO.class, new ServiceReferenceMetricEsPersistenceDAO(elasticSearchClient));
 
+        this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IInstanceReferenceMetricPersistenceDAO.class, new InstanceReferenceMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IInstanceHeartBeatPersistenceDAO.class, new InstanceHeartBeatEsPersistenceDAO(elasticSearchClient));
     }
 

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/InstanceReferenceMetricEsPersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetricTable;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.reindex.BulkByScrollResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng
+ */
+public class InstanceReferenceMetricEsPersistenceDAO extends EsDAO implements IInstanceReferenceMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, InstanceReferenceMetric> {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceReferenceMetricEsPersistenceDAO.class);
+
+    public InstanceReferenceMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override public InstanceReferenceMetric get(String id) {
+        GetResponse getResponse = getClient().prepareGet(InstanceReferenceMetricTable.TABLE, id).get();
+        if (getResponse.isExists()) {
+            logger.debug("getId: {} is exist", id);
+            InstanceReferenceMetric instanceReferenceMetric = new InstanceReferenceMetric(id);
+            Map<String, Object> source = getResponse.getSource();
+            instanceReferenceMetric.setFrontInstanceId((Integer)source.get(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID));
+            instanceReferenceMetric.setBehindInstanceId((Integer)source.get(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID));
+
+            instanceReferenceMetric.setTransactionCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue());
+            instanceReferenceMetric.setTransactionErrorCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue());
+            instanceReferenceMetric.setTransactionDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue());
+            instanceReferenceMetric.setTransactionErrorDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue());
+
+            instanceReferenceMetric.setBusinessTransactionCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue());
+            instanceReferenceMetric.setBusinessTransactionErrorCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue());
+            instanceReferenceMetric.setBusinessTransactionDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue());
+            instanceReferenceMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue());
+
+            instanceReferenceMetric.setMqTransactionCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue());
+            instanceReferenceMetric.setMqTransactionErrorCalls(((Number)source.get(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue());
+            instanceReferenceMetric.setMqTransactionDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue());
+            instanceReferenceMetric.setMqTransactionErrorDurationSum(((Number)source.get(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue());
+
+            instanceReferenceMetric.setTimeBucket(((Number)source.get(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET)).longValue());
+
+            return instanceReferenceMetric;
+        } else {
+            return null;
+        }
+    }
+
+    @Override public IndexRequestBuilder prepareBatchInsert(InstanceReferenceMetric data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, data.getFrontInstanceId());
+        source.put(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, data.getBehindInstanceId());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket());
+
+        return getClient().prepareIndex(InstanceReferenceMetricTable.TABLE, data.getId()).setSource(source);
+    }
+
+    @Override public UpdateRequestBuilder prepareBatchUpdate(InstanceReferenceMetric data) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, data.getFrontInstanceId());
+        source.put(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, data.getBehindInstanceId());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket());
+
+        return getClient().prepareUpdate(InstanceReferenceMetricTable.TABLE, data.getId()).setDoc(source);
+    }
+
+    @Override public void deleteHistory(Long startTimestamp, Long endTimestamp) {
+        long startTimeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(startTimestamp);
+        long endTimeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(endTimestamp);
+        BulkByScrollResponse response = getClient().prepareDelete()
+            .filter(QueryBuilders.rangeQuery(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket))
+            .source(InstanceReferenceMetricTable.TABLE)
+            .get();
+
+        long deleted = response.getDeleted();
+        logger.info("Delete {} rows history from {} index.", deleted, InstanceReferenceMetricTable.TABLE);
+    }
+}

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceReferenceMetricEsPersistenceDAO.java

@@ -16,12 +16,12 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.es.dao;
 
 import java.util.HashMap;
 import java.util.Map;
 import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
@@ -31,7 +31,6 @@ import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.elasticsearch.index.query.QueryBuilders;
 import org.elasticsearch.index.reindex.BulkByScrollResponse;
-import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/InstanceMetricEsTableDefine.java

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.es.define;
 
 import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine;

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/InstanceReferenceMetricEsTableDefine.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define;
+
+import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine;
+import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class InstanceReferenceMetricEsTableDefine extends ElasticSearchTableDefine {
+
+    public InstanceReferenceMetricEsTableDefine() {
+        super(InstanceReferenceMetricTable.TABLE);
+    }
+
+    @Override public int refreshInterval() {
+        return 2;
+    }
+
+    @Override public void initialize() {
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
+
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, ElasticSearchColumnDefine.Type.Long.name()));
+
+        addColumn(new ElasticSearchColumnDefine(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name()));
+    }
+}

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

 org.apache.skywalking.apm.collector.storage.es.define.ApplicationEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.InstanceEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.InstanceMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.InstanceReferenceMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.ServiceNameEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.CpuMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.GCMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.MemoryMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.MemoryPoolMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.GlobalTraceEsTableDefine
-org.apache.skywalking.apm.collector.storage.es.define.InstanceMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.ApplicationComponentEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.ApplicationMappingEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.ApplicationReferenceMetricEsTableDefine

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

@@ -19,16 +19,6 @@
 package org.apache.skywalking.apm.collector.storage.h2;
 
 import java.util.Properties;
-import org.apache.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO;
-import org.apache.skywalking.apm.collector.storage.h2.base.define.H2StorageInstaller;
-import org.apache.skywalking.apm.collector.storage.h2.dao.AlertingListH2PersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.h2.dao.GCMetricH2PersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceH2CacheDAO;
-import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentH2PersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceEntryH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.client.h2.H2Client;
 import org.apache.skywalking.apm.collector.client.h2.H2ClientException;
 import org.apache.skywalking.apm.collector.core.module.Module;
@@ -54,13 +44,18 @@ import org.apache.skywalking.apm.collector.storage.dao.IGCMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGlobalTracePersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGlobalTraceUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceCacheDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceMetricUIDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IInstanceUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IMemoryPoolMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ISegmentUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO;
@@ -71,6 +66,8 @@ import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceUIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.base.dao.BatchH2DAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2StorageInstaller;
+import org.apache.skywalking.apm.collector.storage.h2.dao.AlertingListH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationComponentH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationComponentH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationH2CacheDAO;
@@ -82,21 +79,26 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceMe
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.CpuMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.CpuMetricH2UIDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.GCMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.GCMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.GlobalTraceH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.GlobalTraceH2UIDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceH2CacheDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceH2RegisterDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceHeartBeatH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceMetricH2UIDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceReferenceMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryPoolMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.MemoryPoolMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentCostH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentCostH2UIDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.SegmentH2UIDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceEntryH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceEntryH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceNameH2CacheDAO;
@@ -179,7 +181,6 @@ public class StorageModuleH2Provider extends ModuleProvider {
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricH2PersistenceDAO(h2Client));
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceH2PersistenceDAO(h2Client));
-        this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IApplicationComponentPersistenceDAO.class, new ApplicationComponentH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IApplicationMappingPersistenceDAO.class, new ApplicationMappingH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IApplicationMetricPersistenceDAO.class, new ApplicationMetricH2PersistenceDAO(h2Client));
@@ -191,6 +192,8 @@ public class StorageModuleH2Provider extends ModuleProvider {
         this.registerServiceImplementation(IServiceReferenceMetricPersistenceDAO.class, new ServiceReferenceMetricH2PersistenceDAO(h2Client));
 
         this.registerServiceImplementation(IInstanceHeartBeatPersistenceDAO.class, new InstanceHeartBeatH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IInstanceReferenceMetricPersistenceDAO.class, new InstanceReferenceMetricH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricH2PersistenceDAO(h2Client));
     }
 
     private void registerUiDAO() throws ServiceNotProvidedException {

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/InstanceReferenceMetricH2PersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.h2.dao;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.skywalking.apm.collector.client.h2.H2Client;
+import org.apache.skywalking.apm.collector.client.h2.H2ClientException;
+import org.apache.skywalking.apm.collector.storage.base.sql.SqlBuilder;
+import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetricTable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author peng-yongsheng, clevertension
+ */
+public class InstanceReferenceMetricH2PersistenceDAO extends H2DAO implements IInstanceReferenceMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, InstanceReferenceMetric> {
+
+    private final Logger logger = LoggerFactory.getLogger(InstanceReferenceMetricH2PersistenceDAO.class);
+    private static final String GET_SQL = "select * from {0} where {1} = ?";
+
+    public InstanceReferenceMetricH2PersistenceDAO(H2Client client) {
+        super(client);
+    }
+
+    @Override public InstanceReferenceMetric get(String id) {
+        H2Client client = getClient();
+        String sql = SqlBuilder.buildSql(GET_SQL, InstanceReferenceMetricTable.TABLE, InstanceReferenceMetricTable.COLUMN_ID);
+        Object[] params = new Object[] {id};
+        try (ResultSet rs = client.executeQuery(sql, params)) {
+            if (rs.next()) {
+                InstanceReferenceMetric instanceReferenceMetric = new InstanceReferenceMetric(id);
+                instanceReferenceMetric.setFrontInstanceId(rs.getInt(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID));
+                instanceReferenceMetric.setBehindInstanceId(rs.getInt(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID));
+
+                instanceReferenceMetric.setTransactionCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS));
+                instanceReferenceMetric.setTransactionErrorCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS));
+                instanceReferenceMetric.setTransactionDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM));
+                instanceReferenceMetric.setTransactionErrorDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM));
+
+                instanceReferenceMetric.setBusinessTransactionCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS));
+                instanceReferenceMetric.setBusinessTransactionErrorCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS));
+                instanceReferenceMetric.setBusinessTransactionDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM));
+                instanceReferenceMetric.setBusinessTransactionErrorDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM));
+
+                instanceReferenceMetric.setMqTransactionCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS));
+                instanceReferenceMetric.setMqTransactionErrorCalls(rs.getLong(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS));
+                instanceReferenceMetric.setMqTransactionDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM));
+                instanceReferenceMetric.setMqTransactionErrorDurationSum(rs.getLong(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM));
+
+                instanceReferenceMetric.setTimeBucket(rs.getLong(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET));
+                return instanceReferenceMetric;
+            }
+        } catch (SQLException | H2ClientException e) {
+            logger.error(e.getMessage(), e);
+        }
+        return null;
+    }
+
+    @Override public H2SqlEntity prepareBatchInsert(InstanceReferenceMetric data) {
+        Map<String, Object> source = new HashMap<>();
+        H2SqlEntity entity = new H2SqlEntity();
+        source.put(InstanceReferenceMetricTable.COLUMN_ID, data.getId());
+        source.put(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, data.getFrontInstanceId());
+        source.put(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, data.getBehindInstanceId());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket());
+        String sql = SqlBuilder.buildBatchInsertSql(InstanceReferenceMetricTable.TABLE, source.keySet());
+        entity.setSql(sql);
+        entity.setParams(source.values().toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public H2SqlEntity prepareBatchUpdate(InstanceReferenceMetric data) {
+        Map<String, Object> source = new HashMap<>();
+        H2SqlEntity entity = new H2SqlEntity();
+        source.put(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, data.getFrontInstanceId());
+        source.put(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, data.getBehindInstanceId());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum());
+        source.put(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum());
+
+        source.put(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket());
+        String sql = SqlBuilder.buildBatchUpdateSql(InstanceReferenceMetricTable.TABLE, source.keySet(), InstanceReferenceMetricTable.COLUMN_ID);
+        entity.setSql(sql);
+        List<Object> values = new ArrayList<>(source.values());
+        values.add(data.getId());
+        entity.setParams(values.toArray(new Object[0]));
+        return entity;
+    }
+
+    @Override public void deleteHistory(Long startTimestamp, Long endTimestamp) {
+    }
+}

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/define/InstanceReferenceMetricH2TableDefine.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.h2.define;
+
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2ColumnDefine;
+import org.apache.skywalking.apm.collector.storage.h2.base.define.H2TableDefine;
+import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class InstanceReferenceMetricH2TableDefine extends H2TableDefine {
+
+    public InstanceReferenceMetricH2TableDefine() {
+        super(InstanceReferenceMetricTable.TABLE);
+    }
+
+    @Override public void initialize() {
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_ID, H2ColumnDefine.Type.Varchar.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_FRONT_INSTANCE_ID, H2ColumnDefine.Type.Int.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_BEHIND_INSTANCE_ID, H2ColumnDefine.Type.Int.name()));
+
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, H2ColumnDefine.Type.Bigint.name()));
+
+        addColumn(new H2ColumnDefine(InstanceReferenceMetricTable.COLUMN_TIME_BUCKET, H2ColumnDefine.Type.Bigint.name()));
+    }
+}

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

@@ -7,6 +7,7 @@ org.apache.skywalking.apm.collector.storage.h2.define.MemoryMetricH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.MemoryPoolMetricH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.GlobalTraceH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.InstanceMetricH2TableDefine
+org.apache.skywalking.apm.collector.storage.h2.define.InstanceReferenceMetricH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.ApplicationComponentH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.ApplicationMappingH2TableDefine
 org.apache.skywalking.apm.collector.storage.h2.define.ApplicationReferenceMetricH2TableDefine