JVM worker

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

@@ -48,18 +48,18 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe
 
     private final Logger logger = LoggerFactory.getLogger(JVMMetricsServiceHandler.class);
 
-    private final Graph memoryMetricGraph;
-    private final Graph memoryPoolMetricGraph;
-    private final Graph gcMetricGraph;
-    private final Graph cpuMetricGraph;
-    private final Graph heartBeatGraph;
+    private final Graph<MemoryMetric> memoryMetricGraph;
+    private final Graph<MemoryPoolMetric> memoryPoolMetricGraph;
+    private final Graph<GCMetric> gcMetricGraph;
+    private final Graph<CpuMetric> cpuMetricGraph;
+    private final Graph<Instance> heartBeatGraph;
 
     public JVMMetricsServiceHandler() {
-        memoryPoolMetricGraph = GraphManager.INSTANCE.findGraph(JvmMetricStreamGraph.MEMORY_POOL_METRIC_GRAPH_ID);
-        memoryMetricGraph = GraphManager.INSTANCE.findGraph(JvmMetricStreamGraph.MEMORY_METRIC_GRAPH_ID);
-        gcMetricGraph = GraphManager.INSTANCE.findGraph(JvmMetricStreamGraph.GC_METRIC_GRAPH_ID);
-        cpuMetricGraph = GraphManager.INSTANCE.findGraph(JvmMetricStreamGraph.CPU_METRIC_GRAPH_ID);
-        heartBeatGraph = GraphManager.INSTANCE.findGraph(JvmMetricStreamGraph.INST_HEART_BEAT_GRAPH_ID);
+        memoryMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_METRIC_GRAPH_ID, MemoryMetric.class);
+        memoryPoolMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class);
+        gcMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.GC_METRIC_GRAPH_ID, GCMetric.class);
+        cpuMetricGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.CPU_METRIC_GRAPH_ID, CpuMetric.class);
+        heartBeatGraph = GraphManager.INSTANCE.createIfAbsent(JvmMetricStreamGraph.INST_HEART_BEAT_GRAPH_ID, Instance.class);
     }
 
     @Override public void collect(JVMMetrics request, StreamObserver<Downstream> responseObserver) {

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

@@ -18,10 +18,6 @@
 
 package org.skywalking.apm.collector.agent.stream.graph;
 
-import org.skywalking.apm.collector.agent.stream.worker.jvm.GCMetricPersistenceWorker;
-import org.skywalking.apm.collector.agent.stream.worker.jvm.InstHeartBeatPersistenceWorker;
-import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryMetricPersistenceWorker;
-import org.skywalking.apm.collector.agent.stream.worker.jvm.MemoryPoolMetricPersistenceWorker;
 import org.skywalking.apm.collector.core.graph.Graph;
 import org.skywalking.apm.collector.core.graph.GraphManager;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;
@@ -42,32 +38,28 @@ public class JvmMetricStreamGraph {
     public static final int CPU_METRIC_GRAPH_ID = 103;
     public static final int INST_HEART_BEAT_GRAPH_ID = 104;
 
-    public Graph<GCMetric> createGCMetricGraph() {
+    public Graph<GCMetric> createGcMetricGraph() {
         Graph<GCMetric> graph = GraphManager.INSTANCE.createIfAbsent(GC_METRIC_GRAPH_ID, GCMetric.class);
-        graph.addNode(new GCMetricPersistenceWorker());
         return graph;
     }
 
-    public Graph<CpuMetric> createCPUMetricGraph() throws ProviderNotFoundException {
+    public Graph<CpuMetric> createCpuMetricGraph() throws ProviderNotFoundException {
         Graph<CpuMetric> graph = GraphManager.INSTANCE.createIfAbsent(CPU_METRIC_GRAPH_ID, CpuMetric.class);
         return graph;
     }
 
     public Graph<MemoryMetric> createMemoryMetricGraph() {
         Graph<MemoryMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_METRIC_GRAPH_ID, MemoryMetric.class);
-        graph.addNode(new MemoryMetricPersistenceWorker());
         return graph;
     }
 
     public Graph<MemoryPoolMetric> createMemoryPoolMetricGraph() {
         Graph<MemoryPoolMetric> graph = GraphManager.INSTANCE.createIfAbsent(MEMORY_POOL_METRIC_GRAPH_ID, MemoryPoolMetric.class);
-        graph.addNode(new MemoryPoolMetricPersistenceWorker());
         return graph;
     }
 
     public Graph<Instance> createHeartBeatGraph() {
         Graph<Instance> graph = GraphManager.INSTANCE.createIfAbsent(INST_HEART_BEAT_GRAPH_ID, Instance.class);
-        graph.addNode(new InstHeartBeatPersistenceWorker());
         return graph;
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/CpuMetricPersistenceWorker.java

@@ -20,7 +20,7 @@ package org.skywalking.apm.collector.agent.stream.worker.jvm;
 
 import org.skywalking.apm.collector.queue.service.QueueCreatorService;
 import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.skywalking.apm.collector.storage.dao.ICpuMetricStreamDAO;
+import org.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 import org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider;
@@ -47,7 +47,7 @@ public class CpuMetricPersistenceWorker extends PersistenceWorker<CpuMetric, Cpu
     }
 
     @Override protected IPersistenceDAO persistenceDAO() {
-        return daoService.getPersistenceDAO(ICpuMetricStreamDAO.class);
+        return daoService.getPersistenceDAO(ICpuMetricPersistenceDAO.class);
     }
 
     public static class Factory extends AbstractLocalAsyncWorkerProvider<CpuMetric, CpuMetric, CpuMetricPersistenceWorker> {

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/GCMetricPersistenceWorker.java

@@ -18,20 +18,52 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.jvm;
 
-import org.skywalking.apm.collector.core.graph.Next;
-import org.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
+import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.jvm.GCMetric;
+import org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
 
 /**
  * @author peng-yongsheng
  */
-public class GCMetricPersistenceWorker implements NodeProcessor<GCMetric, GCMetric> {
+public class GCMetricPersistenceWorker extends PersistenceWorker<GCMetric, GCMetric> {
+
+    private final DAOService daoService;
 
     @Override public int id() {
-        return 1;
+        return 0;
+    }
+
+    public GCMetricPersistenceWorker(DAOService daoService) {
+        super(daoService);
+        this.daoService = daoService;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return false;
     }
 
-    @Override public void process(GCMetric INPUT, Next<GCMetric> next) {
+    @Override protected IPersistenceDAO persistenceDAO() {
+        return daoService.getPersistenceDAO(IGCMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<GCMetric, GCMetric, GCMetricPersistenceWorker> {
+
+        public Factory(DAOService daoService, QueueCreatorService<GCMetric> queueCreatorService) {
+            super(daoService, queueCreatorService);
+        }
+
+        @Override
+        public GCMetricPersistenceWorker workerInstance(DAOService daoService) {
+            return new GCMetricPersistenceWorker(daoService);
+        }
 
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/InstHeartBeatPersistenceWorker.java

@@ -18,20 +18,52 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.jvm;
 
-import org.skywalking.apm.collector.core.graph.Next;
-import org.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
+import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.register.Instance;
+import org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
 
 /**
  * @author peng-yongsheng
  */
-public class InstHeartBeatPersistenceWorker implements NodeProcessor<Instance, Instance> {
+public class InstHeartBeatPersistenceWorker extends PersistenceWorker<Instance, Instance> {
+
+    private final DAOService daoService;
 
     @Override public int id() {
         return 0;
     }
 
-    @Override public void process(Instance INPUT, Next<Instance> next) {
+    public InstHeartBeatPersistenceWorker(DAOService daoService) {
+        super(daoService);
+        this.daoService = daoService;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @Override protected IPersistenceDAO persistenceDAO() {
+        return daoService.getPersistenceDAO(IInstanceHeartBeatPersistenceDAO.class);
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<Instance, Instance, InstHeartBeatPersistenceWorker> {
+
+        public Factory(DAOService daoService, QueueCreatorService<Instance> queueCreatorService) {
+            super(daoService, queueCreatorService);
+        }
+
+        @Override
+        public InstHeartBeatPersistenceWorker workerInstance(DAOService daoService) {
+            return new InstHeartBeatPersistenceWorker(daoService);
+        }
 
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryMetricPersistenceWorker.java

@@ -18,20 +18,52 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.jvm;
 
-import org.skywalking.apm.collector.core.graph.Next;
-import org.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
+import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
 
 /**
  * @author peng-yongsheng
  */
-public class MemoryMetricPersistenceWorker implements NodeProcessor<MemoryMetric, MemoryMetric> {
+public class MemoryMetricPersistenceWorker extends PersistenceWorker<MemoryMetric, MemoryMetric> {
+
+    private final DAOService daoService;
 
     @Override public int id() {
         return 0;
     }
 
-    @Override public void process(MemoryMetric INPUT, Next<MemoryMetric> next) {
+    public MemoryMetricPersistenceWorker(DAOService daoService) {
+        super(daoService);
+        this.daoService = daoService;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return false;
+    }
+
+    @Override protected IPersistenceDAO persistenceDAO() {
+        return daoService.getPersistenceDAO(IMemoryMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<MemoryMetric, MemoryMetric, MemoryMetricPersistenceWorker> {
+
+        public Factory(DAOService daoService, QueueCreatorService<MemoryMetric> queueCreatorService) {
+            super(daoService, queueCreatorService);
+        }
+
+        @Override
+        public MemoryMetricPersistenceWorker workerInstance(DAOService daoService) {
+            return new MemoryMetricPersistenceWorker(daoService);
+        }
 
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
     }
 }

apm-collector/apm-collector-agent/collector-agent-stream/src/main/java/org/skywalking/apm/collector/agent/stream/worker/jvm/MemoryPoolMetricPersistenceWorker.java

@@ -18,20 +18,52 @@
 
 package org.skywalking.apm.collector.agent.stream.worker.jvm;
 
-import org.skywalking.apm.collector.core.graph.Next;
-import org.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.skywalking.apm.collector.queue.service.QueueCreatorService;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
+import org.skywalking.apm.collector.storage.service.DAOService;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
+import org.skywalking.apm.collector.stream.worker.base.AbstractLocalAsyncWorkerProvider;
+import org.skywalking.apm.collector.stream.worker.impl.PersistenceWorker;
 
 /**
  * @author peng-yongsheng
  */
-public class MemoryPoolMetricPersistenceWorker implements NodeProcessor<MemoryPoolMetric, MemoryPoolMetric> {
+public class MemoryPoolMetricPersistenceWorker extends PersistenceWorker<MemoryPoolMetric, MemoryPoolMetric> {
+
+    private final DAOService daoService;
 
     @Override public int id() {
         return 0;
     }
 
-    @Override public void process(MemoryPoolMetric INPUT, Next<MemoryPoolMetric> next) {
+    public MemoryPoolMetricPersistenceWorker(DAOService daoService) {
+        super(daoService);
+        this.daoService = daoService;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return false;
+    }
+
+    @Override protected IPersistenceDAO persistenceDAO() {
+        return daoService.getPersistenceDAO(IMemoryPoolMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends AbstractLocalAsyncWorkerProvider<MemoryPoolMetric, MemoryPoolMetric, MemoryPoolMetricPersistenceWorker> {
+
+        public Factory(DAOService daoService, QueueCreatorService<MemoryPoolMetric> queueCreatorService) {
+            super(daoService, queueCreatorService);
+        }
+
+        @Override
+        public MemoryPoolMetricPersistenceWorker workerInstance(DAOService daoService) {
+            return new MemoryPoolMetricPersistenceWorker(daoService);
+        }
 
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
     }
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IGCMetricStreamDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/ICpuMetricPersistenceDAO.java

@@ -18,8 +18,11 @@
 
 package org.skywalking.apm.collector.storage.dao;
 
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+
 /**
  * @author peng-yongsheng
  */
-public interface IGCMetricStreamDAO {
+public interface ICpuMetricPersistenceDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/ICpuMetricStreamDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IGCMetricPersistenceDAO.java

@@ -24,5 +24,5 @@ import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 /**
  * @author peng-yongsheng
  */
-public interface ICpuMetricStreamDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
+public interface IGCMetricPersistenceDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IInstanceHeartBeatDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IInstanceHeartBeatPersistenceDAO.java

@@ -18,8 +18,11 @@
 
 package org.skywalking.apm.collector.storage.dao;
 
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+
 /**
  * @author peng-yongsheng
  */
-public interface IInstanceHeartBeatDAO {
+public interface IInstanceHeartBeatPersistenceDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
 }
\ No newline at end of file

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryPoolMetricDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryMetricPersistenceDAO.java

@@ -18,8 +18,11 @@
 
 package org.skywalking.apm.collector.storage.dao;
 
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+
 /**
  * @author peng-yongsheng
  */
-public interface IMemoryPoolMetricDAO {
+public interface IMemoryMetricPersistenceDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
 }

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryMetricDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/skywalking/apm/collector/storage/dao/IMemoryPoolMetricPersistenceDAO.java

@@ -18,8 +18,11 @@
 
 package org.skywalking.apm.collector.storage.dao;
 
+import org.skywalking.apm.collector.core.data.Data;
+import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+
 /**
  * @author peng-yongsheng
  */
-public interface IMemoryMetricDAO {
+public interface IMemoryPoolMetricPersistenceDAO<Insert, Update, DataImpl extends Data> extends IPersistenceDAO<Insert, Update, DataImpl> {
 }

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/CpuMetricEsStreamDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/CpuMetricEsPersistenceDAO.java

@@ -22,7 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
-import org.skywalking.apm.collector.storage.dao.ICpuMetricStreamDAO;
+import org.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
@@ -32,9 +32,9 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public class CpuMetricEsStreamDAO extends EsDAO implements ICpuMetricStreamDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+public class CpuMetricEsPersistenceDAO extends EsDAO implements ICpuMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
 
-    private final Logger logger = LoggerFactory.getLogger(CpuMetricEsStreamDAO.class);
+    private final Logger logger = LoggerFactory.getLogger(CpuMetricEsPersistenceDAO.class);
 
     @Override public CpuMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/GCMetricEsStreamDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/GCMetricEsPersistenceDAO.java

@@ -22,8 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.skywalking.apm.collector.storage.dao.IGCMetricStreamDAO;
+import org.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.table.jvm.GCMetric;
 import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable;
 /**
  * @author peng-yongsheng
  */
-public class GCMetricEsStreamDAO extends EsDAO implements IGCMetricStreamDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, GCMetric> {
+public class GCMetricEsPersistenceDAO extends EsDAO implements IGCMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, GCMetric> {
 
     @Override public GCMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/InstanceHeartBeatEsDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/InstanceHeartBeatEsPersistenceDAO.java

@@ -24,8 +24,7 @@ import org.elasticsearch.action.get.GetResponse;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
 import org.skywalking.apm.collector.core.UnexpectedException;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatDAO;
+import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.table.register.Instance;
 import org.skywalking.apm.collector.storage.table.register.InstanceTable;
@@ -35,9 +34,9 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng
  */
-public class InstanceHeartBeatEsDAO extends EsDAO implements IInstanceHeartBeatDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, Instance> {
+public class InstanceHeartBeatEsPersistenceDAO extends EsDAO implements IInstanceHeartBeatPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, Instance> {
 
-    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatEsDAO.class);
+    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatEsPersistenceDAO.class);
 
     @Override public Instance get(String id) {
         GetResponse getResponse = getClient().prepareGet(InstanceTable.TABLE, id).get();

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryMetricEsDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryMetricEsPersistenceDAO.java

@@ -22,8 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.skywalking.apm.collector.storage.dao.IMemoryMetricDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
 /**
  * @author peng-yongsheng
  */
-public class MemoryMetricEsDAO extends EsDAO implements IMemoryMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+public class MemoryMetricEsPersistenceDAO extends EsDAO implements IMemoryMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
 
     @Override public MemoryMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryPoolMetricEsDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/skywalking/apm/collector/storage/es/dao/MemoryPoolMetricEsPersistenceDAO.java

@@ -22,8 +22,7 @@ import java.util.HashMap;
 import java.util.Map;
 import org.elasticsearch.action.index.IndexRequestBuilder;
 import org.elasticsearch.action.update.UpdateRequestBuilder;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.es.base.dao.EsDAO;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable;
 /**
  * @author peng-yongsheng
  */
-public class MemoryPoolMetricEsDAO extends EsDAO implements IMemoryPoolMetricDAO, IPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryPoolMetric> {
+public class MemoryPoolMetricEsPersistenceDAO extends EsDAO implements IMemoryPoolMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryPoolMetric> {
 
     @Override public MemoryPoolMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/CpuMetricH2StreamDAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/CpuMetricH2PersistenceDAO.java

@@ -21,7 +21,7 @@ package org.skywalking.apm.collector.storage.h2.dao;
 import java.util.HashMap;
 import java.util.Map;
 import org.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.skywalking.apm.collector.storage.dao.ICpuMetricStreamDAO;
+import org.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.skywalking.apm.collector.storage.table.jvm.CpuMetric;
@@ -32,9 +32,9 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class CpuMetricH2StreamDAO extends H2DAO implements ICpuMetricStreamDAO<H2SqlEntity, H2SqlEntity, CpuMetric> {
+public class CpuMetricH2PersistenceDAO extends H2DAO implements ICpuMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, CpuMetric> {
 
-    private final Logger logger = LoggerFactory.getLogger(CpuMetricH2StreamDAO.class);
+    private final Logger logger = LoggerFactory.getLogger(CpuMetricH2PersistenceDAO.class);
 
     @Override public CpuMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/GCMetricH2StreamDAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/GCMetricH2PersistenceDAO.java

@@ -20,9 +20,8 @@ package org.skywalking.apm.collector.storage.h2.dao;
 
 import java.util.HashMap;
 import java.util.Map;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 import org.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.skywalking.apm.collector.storage.dao.IGCMetricStreamDAO;
+import org.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.skywalking.apm.collector.storage.table.jvm.GCMetric;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.GCMetricTable;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class GCMetricH2StreamDAO extends H2DAO implements IGCMetricStreamDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity, GCMetric> {
+public class GCMetricH2PersistenceDAO extends H2DAO implements IGCMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, GCMetric> {
 
     @Override public GCMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/InstanceHeartBeatH2DAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/InstanceHeartBeatH2PersistenceDAO.java

@@ -27,9 +27,8 @@ import java.util.Map;
 import org.skywalking.apm.collector.client.h2.H2Client;
 import org.skywalking.apm.collector.client.h2.H2ClientException;
 import org.skywalking.apm.collector.core.UnexpectedException;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 import org.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatDAO;
+import org.skywalking.apm.collector.storage.dao.IInstanceHeartBeatPersistenceDAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.skywalking.apm.collector.storage.table.register.Instance;
@@ -40,9 +39,9 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class InstanceHeartBeatH2DAO extends H2DAO implements IInstanceHeartBeatDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity, Instance> {
+public class InstanceHeartBeatH2PersistenceDAO extends H2DAO implements IInstanceHeartBeatPersistenceDAO<H2SqlEntity, H2SqlEntity, Instance> {
 
-    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatH2DAO.class);
+    private final Logger logger = LoggerFactory.getLogger(InstanceHeartBeatH2PersistenceDAO.class);
 
     private static final String GET_INSTANCE_HEARTBEAT_SQL = "select * from {0} where {1} = ?";
 

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryMetricH2DAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryMetricH2PersistenceDAO.java

@@ -20,9 +20,8 @@ package org.skywalking.apm.collector.storage.h2.dao;
 
 import java.util.HashMap;
 import java.util.Map;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 import org.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.skywalking.apm.collector.storage.dao.IMemoryMetricDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class MemoryMetricH2DAO extends H2DAO implements IMemoryMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryMetric> {
+public class MemoryMetricH2PersistenceDAO extends H2DAO implements IMemoryMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryMetric> {
 
     @Override public MemoryMetric get(String id) {
         return null;

apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryPoolMetricH2DAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/skywalking/apm/collector/storage/h2/dao/MemoryPoolMetricH2PersistenceDAO.java

@@ -20,9 +20,8 @@ package org.skywalking.apm.collector.storage.h2.dao;
 
 import java.util.HashMap;
 import java.util.Map;
-import org.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 import org.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricDAO;
+import org.skywalking.apm.collector.storage.dao.IMemoryPoolMetricPersistenceDAO;
 import org.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetric;
@@ -31,7 +30,7 @@ import org.skywalking.apm.collector.storage.table.jvm.MemoryPoolMetricTable;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class MemoryPoolMetricH2DAO extends H2DAO implements IMemoryPoolMetricDAO, IPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryPoolMetric> {
+public class MemoryPoolMetricH2PersistenceDAO extends H2DAO implements IMemoryPoolMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryPoolMetric> {
 
     @Override public MemoryPoolMetric get(String id) {
         return null;