Memory metric pyramid aggregate.

apm-collector/apm-collector-analysis/analysis-jvm/jvm-define/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/define/graph/WorkerIdDefine.java

@@ -44,7 +44,17 @@ public class WorkerIdDefine {
     public static final int GC_MONTH_METRIC_PERSISTENCE_WORKER_ID = 3108;
     public static final int GC_MONTH_METRIC_TRANSFORM_NODE_ID = 3109;
 
+    public static final int MEMORY_METRIC_BRIDGE_NODE_ID = 3200;
+    public static final int MEMORY_SECOND_METRIC_PERSISTENCE_WORKER_ID = 3201;
+    public static final int MEMORY_MINUTE_METRIC_PERSISTENCE_WORKER_ID = 3202;
+    public static final int MEMORY_MINUTE_METRIC_TRANSFORM_NODE_ID = 3203;
+    public static final int MEMORY_HOUR_METRIC_PERSISTENCE_WORKER_ID = 3204;
+    public static final int MEMORY_HOUR_METRIC_TRANSFORM_NODE_ID = 3205;
+    public static final int MEMORY_DAY_METRIC_PERSISTENCE_WORKER_ID = 3206;
+    public static final int MEMORY_DAY_METRIC_TRANSFORM_NODE_ID = 3207;
+    public static final int MEMORY_MONTH_METRIC_PERSISTENCE_WORKER_ID = 3208;
+    public static final int MEMORY_MONTH_METRIC_TRANSFORM_NODE_ID = 3209;
+
     public static final int INST_HEART_BEAT_PERSISTENCE_WORKER_ID = 302;
-    public static final int MEMORY_METRIC_PERSISTENCE_WORKER_ID = 303;
     public static final int MEMORY_POOL_METRIC_PERSISTENCE_WORKER_ID = 303;
 }

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

@@ -33,7 +33,7 @@ import org.apache.skywalking.apm.collector.analysis.jvm.provider.service.MemoryP
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.cpu.CpuMetricPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.gc.GCMetricPersistenceGraph;
 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.memory.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;

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

@@ -0,0 +1,67 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryDayMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
+
+    public MemoryDayMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_DAY_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, MemoryMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IMemoryDayMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemoryDayMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public MemoryDayMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new MemoryDayMetricPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+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.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryDayMetricTransformNode implements NodeProcessor<MemoryMetric, MemoryMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_DAY_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(MemoryMetric memoryMetric, Next<MemoryMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToDay(memoryMetric.getTimeBucket());
+        memoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        memoryMetric.setTimeBucket(timeBucket);
+
+        next.execute(memoryMetric);
+    }
+}

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

@@ -0,0 +1,67 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryHourMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
+
+    public MemoryHourMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_HOUR_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, MemoryMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IMemoryHourMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemoryHourMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public MemoryHourMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new MemoryHourMetricPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+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.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryHourMetricTransformNode implements NodeProcessor<MemoryMetric, MemoryMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_HOUR_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(MemoryMetric memoryMetric, Next<MemoryMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToHour(memoryMetric.getTimeBucket());
+        memoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        memoryMetric.setTimeBucket(timeBucket);
+
+        next.execute(memoryMetric);
+    }
+}

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

@@ -0,0 +1,38 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.core.graph.Next;
+import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMetricBridgeNode implements NodeProcessor<MemoryMetric, MemoryMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_METRIC_BRIDGE_NODE_ID;
+    }
+
+    @Override public void process(MemoryMetric metric, Next<MemoryMetric> next) {
+        next.execute(metric);
+    }
+}

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

@@ -16,11 +16,12 @@
  *
  */
 
-package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
+package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.memory;
 
 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.graph.Node;
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 
@@ -38,7 +39,21 @@ public class MemoryMetricPersistenceGraph {
     }
 
     public void create() {
-        GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.MEMORY_METRIC_PERSISTENCE_GRAPH_ID, MemoryMetric.class)
-            .addNode(new MemoryMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+        Node<MemoryMetric, MemoryMetric> bridgeNode = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.MEMORY_METRIC_PERSISTENCE_GRAPH_ID, MemoryMetric.class)
+            .addNode(new MemoryMetricBridgeNode());
+
+        bridgeNode.addNext(new MemorySecondMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new MemoryMinuteMetricTransformNode())
+            .addNext(new MemoryMinuteMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new MemoryHourMetricTransformNode())
+            .addNext(new MemoryHourMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new MemoryDayMetricTransformNode())
+            .addNext(new MemoryDayMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new MemoryMonthMetricTransformNode())
+            .addNext(new MemoryMonthMetricPersistenceWorker.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/memory/MemoryMinuteMetricPersistenceWorker.java

@@ -0,0 +1,67 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMinuteMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
+
+    public MemoryMinuteMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_MINUTE_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, MemoryMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IMemoryMinuteMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemoryMinuteMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public MemoryMinuteMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new MemoryMinuteMetricPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+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.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMinuteMetricTransformNode implements NodeProcessor<MemoryMetric, MemoryMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_MINUTE_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(MemoryMetric memoryMetric, Next<MemoryMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToMinute(memoryMetric.getTimeBucket());
+        memoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        memoryMetric.setTimeBucket(timeBucket);
+
+        next.execute(memoryMetric);
+    }
+}

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

@@ -0,0 +1,67 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
+import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorkerProvider;
+import org.apache.skywalking.apm.collector.core.module.ModuleManager;
+import org.apache.skywalking.apm.collector.storage.StorageModule;
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMonthMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
+
+    public MemoryMonthMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_MONTH_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, MemoryMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(IMemoryMonthMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemoryMonthMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public MemoryMonthMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new MemoryMonthMetricPersistenceWorker(moduleManager);
+        }
+
+        @Override
+        public int queueSize() {
+            return 1024;
+        }
+    }
+}

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

@@ -0,0 +1,44 @@
+/*
+ * 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.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
+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.util.Const;
+import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMonthMetricTransformNode implements NodeProcessor<MemoryMetric, MemoryMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.MEMORY_MONTH_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(MemoryMetric memoryMetric, Next<MemoryMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToMonth(memoryMetric.getTimeBucket());
+        memoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        memoryMetric.setTimeBucket(timeBucket);
+
+        next.execute(memoryMetric);
+    }
+}

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

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker;
+package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.memory;
 
 import org.apache.skywalking.apm.collector.analysis.jvm.define.graph.WorkerIdDefine;
 import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
@@ -24,20 +24,20 @@ import org.apache.skywalking.apm.collector.analysis.worker.model.impl.Persistenc
 import org.apache.skywalking.apm.collector.core.module.ModuleManager;
 import org.apache.skywalking.apm.collector.storage.StorageModule;
 import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 
 /**
  * @author peng-yongsheng
  */
-public class MemoryMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
+public class MemorySecondMetricPersistenceWorker extends PersistenceWorker<MemoryMetric> {
 
-    public MemoryMetricPersistenceWorker(ModuleManager moduleManager) {
+    public MemorySecondMetricPersistenceWorker(ModuleManager moduleManager) {
         super(moduleManager);
     }
 
     @Override public int id() {
-        return WorkerIdDefine.MEMORY_METRIC_PERSISTENCE_WORKER_ID;
+        return WorkerIdDefine.MEMORY_SECOND_METRIC_PERSISTENCE_WORKER_ID;
     }
 
     @Override protected boolean needMergeDBData() {
@@ -46,17 +46,17 @@ public class MemoryMetricPersistenceWorker extends PersistenceWorker<MemoryMetri
 
     @SuppressWarnings("unchecked")
     @Override protected IPersistenceDAO<?, ?, MemoryMetric> persistenceDAO() {
-        return getModuleManager().find(StorageModule.NAME).getService(IMemoryMetricPersistenceDAO.class);
+        return getModuleManager().find(StorageModule.NAME).getService(IMemorySecondMetricPersistenceDAO.class);
     }
 
-    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemoryMetricPersistenceWorker> {
+    public static class Factory extends PersistenceWorkerProvider<MemoryMetric, MemorySecondMetricPersistenceWorker> {
 
         public Factory(ModuleManager moduleManager) {
             super(moduleManager);
         }
 
-        @Override public MemoryMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
-            return new MemoryMetricPersistenceWorker(moduleManager);
+        @Override public MemorySecondMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new MemorySecondMetricPersistenceWorker(moduleManager);
         }
 
         @Override

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

@@ -53,7 +53,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceAlarmPe
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMinuteMetricPersistenceDAO;
 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.memorymp.IMemorySecondMetricPersistenceDAO;
 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;
@@ -116,7 +116,7 @@ public class StorageModule extends Module {
     private void addPersistenceDAO(List<Class> classes) {
         classes.add(ICpuSecondMetricPersistenceDAO.class);
         classes.add(IGCSecondMetricPersistenceDAO.class);
-        classes.add(IMemoryMetricPersistenceDAO.class);
+        classes.add(IMemorySecondMetricPersistenceDAO.class);
         classes.add(IMemoryPoolMetricPersistenceDAO.class);
 
         classes.add(IGlobalTracePersistenceDAO.class);

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

@@ -16,7 +16,7 @@
  *
  */
 
-package org.apache.skywalking.apm.collector.storage.dao;
+package org.apache.skywalking.apm.collector.storage.dao.memorymp;
 
 import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
@@ -24,5 +24,5 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
 /**
  * @author peng-yongsheng
  */
-public interface IMemoryMetricPersistenceDAO<Insert, Update, DataImpl extends MemoryMetric> extends IPersistenceDAO<Insert, Update, DataImpl> {
+public interface IMemoryDayMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends MemoryMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
 }

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

@@ -0,0 +1,28 @@
+/*
+ * 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.dao.memorymp;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemoryHourMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends MemoryMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

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

@@ -0,0 +1,28 @@
+/*
+ * 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.dao.memorymp;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemoryMinuteMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends MemoryMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

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

@@ -0,0 +1,28 @@
+/*
+ * 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.dao.memorymp;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemoryMonthMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends MemoryMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

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

@@ -0,0 +1,28 @@
+/*
+ * 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.dao.memorymp;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface IMemorySecondMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends MemoryMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/jvm/GCMetric.java

@@ -68,7 +68,6 @@ public class GCMetric extends StreamData {
 
     @Override public void setMetricId(String metricId) {
         setDataString(1, metricId);
-
     }
 
     public Long getCount() {

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/jvm/MemoryMetric.java

@@ -16,21 +16,21 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.table.jvm;
 
 import org.apache.skywalking.apm.collector.core.data.Column;
-import org.apache.skywalking.apm.collector.core.data.AbstractData;
+import org.apache.skywalking.apm.collector.core.data.StreamData;
 import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
 
 /**
  * @author peng-yongsheng
  */
-public class MemoryMetric extends AbstractData {
+public class MemoryMetric extends StreamData {
 
     private static final Column[] STRING_COLUMNS = {
         new Column(MemoryMetricTable.COLUMN_ID, new NonOperation()),
+        new Column(MemoryMetricTable.COLUMN_METRIC_ID, new NonOperation()),
     };
 
     private static final Column[] LONG_COLUMNS = {
@@ -53,8 +53,24 @@ public class MemoryMetric extends AbstractData {
     };
     private static final Column[] BYTE_COLUMNS = {};
 
-    public MemoryMetric(String id) {
-        super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS);
+    public MemoryMetric() {
+        super(STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS);
+    }
+
+    @Override public String getId() {
+        return getDataString(0);
+    }
+
+    @Override public void setId(String id) {
+        setDataString(0, id);
+    }
+
+    @Override public String getMetricId() {
+        return getDataString(1);
+    }
+
+    @Override public void setMetricId(String metricId) {
+        setDataString(1, metricId);
     }
 
     public Long getInit() {

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

@@ -32,6 +32,7 @@ import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenc
 import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.gcmp.IGCSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.imp.IInstanceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceMinuteMetricPersistenceDAO;
 
 /**
@@ -82,7 +83,7 @@ public class DataTTLKeeperTimer {
         IGCSecondMetricPersistenceDAO gcMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(IGCSecondMetricPersistenceDAO.class);
         gcMetricPersistenceDAO.deleteHistory(startTimestamp, endTimestamp);
 
-        IMemoryMetricPersistenceDAO memoryMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(IMemoryMetricPersistenceDAO.class);
+        IMemorySecondMetricPersistenceDAO memoryMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(IMemorySecondMetricPersistenceDAO.class);
         memoryMetricPersistenceDAO.deleteHistory(startTimestamp, endTimestamp);
 
         IMemoryPoolMetricPersistenceDAO memoryPoolMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(IMemoryPoolMetricPersistenceDAO.class);

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

@@ -62,7 +62,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceAlarmPe
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMinuteMetricPersistenceDAO;
 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.memorymp.IMemorySecondMetricPersistenceDAO;
 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;
@@ -116,7 +116,7 @@ import org.apache.skywalking.apm.collector.storage.es.dao.InstanceMetricEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.InstanceReferenceAlarmEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.InstanceReferenceAlarmListEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.irmp.InstanceReferenceMinuteMetricEsPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.es.dao.MemoryMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemorySecondMetricEsPersistenceDAO;
 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;
@@ -229,7 +229,7 @@ public class StorageModuleEsProvider extends ModuleProvider {
     private void registerPersistenceDAO() throws ServiceNotProvidedException {
         this.registerServiceImplementation(ICpuSecondMetricPersistenceDAO.class, new CpuSecondMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IGCSecondMetricPersistenceDAO.class, new GCSecondMetricEsPersistenceDAO(elasticSearchClient));
-        this.registerServiceImplementation(IMemoryMetricPersistenceDAO.class, new MemoryMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IMemorySecondMetricPersistenceDAO.class, new MemorySecondMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricEsPersistenceDAO(elasticSearchClient));
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceEsPersistenceDAO(elasticSearchClient));

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

@@ -1,79 +0,0 @@
-/*
- * 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.IMemoryMetricPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO;
-import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
-import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
-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 MemoryMetricEsPersistenceDAO extends EsDAO implements IMemoryMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
-
-    private final Logger logger = LoggerFactory.getLogger(MemoryMetricEsPersistenceDAO.class);
-
-    public MemoryMetricEsPersistenceDAO(ElasticSearchClient client) {
-        super(client);
-    }
-
-    @Override public MemoryMetric get(String id) {
-        return null;
-    }
-
-    @Override public IndexRequestBuilder prepareBatchInsert(MemoryMetric data) {
-        Map<String, Object> source = new HashMap<>();
-        source.put(MemoryMetricTable.COLUMN_INSTANCE_ID, data.getInstanceId());
-        source.put(MemoryMetricTable.COLUMN_IS_HEAP, data.getIsHeap());
-        source.put(MemoryMetricTable.COLUMN_INIT, data.getInit());
-        source.put(MemoryMetricTable.COLUMN_MAX, data.getMax());
-        source.put(MemoryMetricTable.COLUMN_USED, data.getUsed());
-        source.put(MemoryMetricTable.COLUMN_COMMITTED, data.getCommitted());
-        source.put(MemoryMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket());
-
-        return getClient().prepareIndex(MemoryMetricTable.TABLE, data.getId()).setSource(source);
-    }
-
-    @Override public UpdateRequestBuilder prepareBatchUpdate(MemoryMetric data) {
-        return null;
-    }
-
-    @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(MemoryMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket))
-            .source(MemoryMetricTable.TABLE)
-            .get();
-
-        long deleted = response.getDeleted();
-        logger.info("Delete {} rows history from {} index.", deleted, MemoryMetricTable.TABLE);
-    }
-}

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

@@ -0,0 +1,60 @@
+/*
+ * 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.memorymp;
+
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.storage.es.base.dao.AbstractPersistenceEsDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractMemoryMetricEsPersistenceDAO extends AbstractPersistenceEsDAO<MemoryMetric> {
+
+    AbstractMemoryMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected final String timeBucketColumnNameForDelete() {
+        return MemoryMetricTable.COLUMN_TIME_BUCKET;
+    }
+
+    @Override protected final MemoryMetric esDataToStreamData(Map<String, Object> source) {
+        return null;
+    }
+
+    @Override protected final Map<String, Object> esStreamDataToEsData(MemoryMetric streamData) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(MemoryMetricTable.COLUMN_ID, streamData.getId());
+        source.put(MemoryMetricTable.COLUMN_METRIC_ID, streamData.getMetricId());
+
+        source.put(MemoryMetricTable.COLUMN_INSTANCE_ID, streamData.getInstanceId());
+        source.put(MemoryMetricTable.COLUMN_IS_HEAP, streamData.getIsHeap());
+        source.put(MemoryMetricTable.COLUMN_INIT, streamData.getInit());
+        source.put(MemoryMetricTable.COLUMN_MAX, streamData.getMax());
+        source.put(MemoryMetricTable.COLUMN_USED, streamData.getUsed());
+        source.put(MemoryMetricTable.COLUMN_COMMITTED, streamData.getCommitted());
+        source.put(MemoryMetricTable.COLUMN_TIME_BUCKET, streamData.getTimeBucket());
+
+        return source;
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.memorymp;
+
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryDayMetricEsPersistenceDAO extends AbstractMemoryMetricEsPersistenceDAO implements IMemoryDayMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+
+    public MemoryDayMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName();
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.memorymp;
+
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryHourMetricEsPersistenceDAO extends AbstractMemoryMetricEsPersistenceDAO implements IMemoryHourMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+
+    public MemoryHourMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName();
+    }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * 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.memorymp;
+
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMinuteMetricEsPersistenceDAO extends AbstractMemoryMetricEsPersistenceDAO implements IMemoryMinuteMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+
+    public MemoryMinuteMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName();
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.memorymp;
+
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMonthMetricEsPersistenceDAO extends AbstractMemoryMetricEsPersistenceDAO implements IMemoryMonthMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+
+    public MemoryMonthMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName();
+    }
+}

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

@@ -0,0 +1,42 @@
+/*
+ * 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.memorymp;
+
+import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient;
+import org.apache.skywalking.apm.collector.core.storage.TimePyramid;
+import org.apache.skywalking.apm.collector.core.util.Const;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemorySecondMetricEsPersistenceDAO extends AbstractMemoryMetricEsPersistenceDAO implements IMemorySecondMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, MemoryMetric> {
+
+    public MemorySecondMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Second.getName();
+    }
+}

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

@@ -58,7 +58,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IInstanceReferenceAlarmPe
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMinuteMetricPersistenceDAO;
 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.memorymp.IMemorySecondMetricPersistenceDAO;
 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;
@@ -112,7 +112,7 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceReferenceAlarmH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.InstanceReferenceAlarmListH2PersistenceDAO;
 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.MemorySecondMetricH2PersistenceDAO;
 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;
@@ -207,7 +207,7 @@ public class StorageModuleH2Provider extends ModuleProvider {
     private void registerPersistenceDAO() throws ServiceNotProvidedException {
         this.registerServiceImplementation(ICpuSecondMetricPersistenceDAO.class, new CpuSecondMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IGCSecondMetricPersistenceDAO.class, new GCSecondMetricH2PersistenceDAO(h2Client));
-        this.registerServiceImplementation(IMemoryMetricPersistenceDAO.class, new MemoryMetricH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(IMemorySecondMetricPersistenceDAO.class, new MemorySecondMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricH2PersistenceDAO(h2Client));
 
         this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceH2PersistenceDAO(h2Client));

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

@@ -23,7 +23,7 @@ import java.util.HashMap;
 import java.util.Map;
 import org.apache.skywalking.apm.collector.client.h2.H2Client;
 import org.apache.skywalking.apm.collector.storage.base.sql.SqlBuilder;
-import org.apache.skywalking.apm.collector.storage.dao.IMemoryMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
 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.jvm.MemoryMetric;
@@ -32,9 +32,9 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class MemoryMetricH2PersistenceDAO extends H2DAO implements IMemoryMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryMetric> {
+public class MemorySecondMetricH2PersistenceDAO extends H2DAO implements IMemorySecondMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, MemoryMetric> {
 
-    public MemoryMetricH2PersistenceDAO(H2Client client) {
+    public MemorySecondMetricH2PersistenceDAO(H2Client client) {
         super(client);
     }