Memory metric pyramid aggregate test successful.

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

@@ -68,7 +68,7 @@ public class JVMMetricsServiceHandler extends JVMMetricsServiceGrpc.JVMMetricsSe
             long time = TimeBucketUtils.INSTANCE.getSecondTimeBucket(metric.getTime());
 //            sendToInstanceHeartBeatService(instanceId, metric.getTime());
 //            sendToCpuMetricService(instanceId, time, metric.getCpu());
-//            sendToMemoryMetricService(instanceId, time, metric.getMemoryList());
+            sendToMemoryMetricService(instanceId, time, metric.getMemoryList());
             sendToMemoryPoolMetricService(instanceId, time, metric.getMemoryPoolList());
 //            sendToGCMetricService(instanceId, time, metric.getGcList());
         });

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

@@ -23,6 +23,7 @@ import io.grpc.ManagedChannelBuilder;
 import org.apache.skywalking.apm.network.proto.JVMMetric;
 import org.apache.skywalking.apm.network.proto.JVMMetrics;
 import org.apache.skywalking.apm.network.proto.JVMMetricsServiceGrpc;
+import org.apache.skywalking.apm.network.proto.Memory;
 import org.apache.skywalking.apm.network.proto.MemoryPool;
 import org.apache.skywalking.apm.network.proto.PoolType;
 
@@ -41,7 +42,9 @@ public class JVMMetricServiceHandlerTestCase {
         JVMMetric.Builder metricBuilder = JVMMetric.newBuilder();
         metricBuilder.setTime(System.currentTimeMillis());
 
+        buildMemoryMetric(metricBuilder);
         buildMemoryPoolMetric(metricBuilder);
+
         builder.addMetrics(metricBuilder.build());
 
         blockingStub.collect(builder.build());
@@ -57,4 +60,15 @@ public class JVMMetricServiceHandlerTestCase {
 
         metricBuilder.addMemoryPool(builder);
     }
+
+    private static void buildMemoryMetric(JVMMetric.Builder metricBuilder) {
+        Memory.Builder builder = Memory.newBuilder();
+        builder.setInit(20);
+        builder.setMax(50);
+        builder.setCommitted(20);
+        builder.setUsed(15);
+        builder.setIsHeap(true);
+
+        metricBuilder.addMemory(builder);
+    }
 }

apm-collector/apm-collector-analysis/analysis-jvm/jvm-define/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/define/service/IMemoryMetricService.java

@@ -24,5 +24,5 @@ import org.apache.skywalking.apm.collector.core.module.Service;
  * @author peng-yongsheng
  */
 public interface IMemoryMetricService extends Service {
-    void send(int instanceId, long timeBucket, boolean isHeap, long init, long max, long used, long commited);
+    void send(int instanceId, long timeBucket, boolean isHeap, long init, long max, long used, long committed);
 }

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

@@ -45,15 +45,20 @@ public class MemoryMetricService implements IMemoryMetricService {
     }
 
     @Override
-    public void send(int instanceId, long timeBucket, boolean isHeap, long init, long max, long used, long commited) {
+    public void send(int instanceId, long timeBucket, boolean isHeap, long init, long max, long used, long committed) {
+        String metricId = instanceId + Const.ID_SPLIT + String.valueOf(isHeap);
+        String id = timeBucket + Const.ID_SPLIT + metricId;
+
         MemoryMetric memoryMetric = new MemoryMetric();
-        memoryMetric.setId(timeBucket + Const.ID_SPLIT + instanceId + Const.ID_SPLIT + String.valueOf(isHeap));
+        memoryMetric.setId(id);
+        memoryMetric.setMetricId(metricId);
         memoryMetric.setInstanceId(instanceId);
         memoryMetric.setIsHeap(isHeap);
         memoryMetric.setInit(init);
         memoryMetric.setMax(max);
         memoryMetric.setUsed(used);
-        memoryMetric.setCommitted(commited);
+        memoryMetric.setCommitted(committed);
+        memoryMetric.setTimes(1L);
         memoryMetric.setTimeBucket(timeBucket);
 
         logger.debug("push to memory metric graph, id: {}", memoryMetric.getId());

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

@@ -36,9 +36,10 @@ public class MemoryDayMetricTransformNode implements NodeProcessor<MemoryMetric,
 
     @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);
+        MemoryMetric newMemoryMetric = MemoryMetricCopy.copy(memoryMetric);
+        newMemoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        newMemoryMetric.setTimeBucket(timeBucket);
+        next.execute(newMemoryMetric);
     }
 }

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

@@ -36,9 +36,10 @@ public class MemoryHourMetricTransformNode implements NodeProcessor<MemoryMetric
 
     @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);
+        MemoryMetric newMemoryMetric = MemoryMetricCopy.copy(memoryMetric);
+        newMemoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        newMemoryMetric.setTimeBucket(timeBucket);
+        next.execute(newMemoryMetric);
     }
 }

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.memory;
+
+import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMetricCopy {
+
+    public static MemoryMetric copy(MemoryMetric memoryMetric) {
+        MemoryMetric newMemoryMetric = new MemoryMetric();
+        newMemoryMetric.setId(memoryMetric.getId());
+        newMemoryMetric.setMetricId(memoryMetric.getMetricId());
+
+        newMemoryMetric.setInstanceId(memoryMetric.getInstanceId());
+        newMemoryMetric.setIsHeap(memoryMetric.getIsHeap());
+
+        newMemoryMetric.setInit(memoryMetric.getInit());
+        newMemoryMetric.setMax(memoryMetric.getMax());
+        newMemoryMetric.setUsed(memoryMetric.getUsed());
+        newMemoryMetric.setCommitted(memoryMetric.getCommitted());
+        newMemoryMetric.setTimes(memoryMetric.getTimes());
+
+        newMemoryMetric.setTimeBucket(memoryMetric.getTimeBucket());
+        return newMemoryMetric;
+    }
+}

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

@@ -36,9 +36,10 @@ public class MemoryMinuteMetricTransformNode implements NodeProcessor<MemoryMetr
 
     @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);
+        MemoryMetric newMemoryMetric = MemoryMetricCopy.copy(memoryMetric);
+        newMemoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        newMemoryMetric.setTimeBucket(timeBucket);
+        next.execute(newMemoryMetric);
     }
 }

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

@@ -36,9 +36,10 @@ public class MemoryMonthMetricTransformNode implements NodeProcessor<MemoryMetri
 
     @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);
+        MemoryMetric newMemoryMetric = MemoryMetricCopy.copy(memoryMetric);
+        newMemoryMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + memoryMetric.getMetricId());
+        newMemoryMetric.setTimeBucket(timeBucket);
+        next.execute(newMemoryMetric);
     }
 }

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

@@ -85,6 +85,10 @@ import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceDa
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMonthMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.mpoolmp.IMemoryPoolDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.mpoolmp.IMemoryPoolHourMetricPersistenceDAO;
@@ -145,7 +149,12 @@ public class StorageModule extends Module {
     private void addPersistenceDAO(List<Class> classes) {
         classes.add(ICpuSecondMetricPersistenceDAO.class);
         classes.add(IGCSecondMetricPersistenceDAO.class);
+
         classes.add(IMemorySecondMetricPersistenceDAO.class);
+        classes.add(IMemoryMinuteMetricPersistenceDAO.class);
+        classes.add(IMemoryHourMetricPersistenceDAO.class);
+        classes.add(IMemoryDayMetricPersistenceDAO.class);
+        classes.add(IMemoryMonthMetricPersistenceDAO.class);
 
         classes.add(IMemoryPoolSecondMetricPersistenceDAO.class);
         classes.add(IMemoryPoolMinuteMetricPersistenceDAO.class);

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

@@ -20,7 +20,10 @@ 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.StreamData;
+import org.apache.skywalking.apm.collector.core.data.operator.AddOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.CoverOperation;
+import org.apache.skywalking.apm.collector.core.data.operator.MaxOperation;
+import org.apache.skywalking.apm.collector.core.data.operator.MinOperation;
 import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
 
 /**
@@ -34,11 +37,12 @@ public class MemoryMetric extends StreamData {
     };
 
     private static final Column[] LONG_COLUMNS = {
-        new Column(MemoryMetricTable.COLUMN_INIT, new CoverOperation()),
-        new Column(MemoryMetricTable.COLUMN_MAX, new CoverOperation()),
-        new Column(MemoryMetricTable.COLUMN_USED, new CoverOperation()),
-        new Column(MemoryMetricTable.COLUMN_COMMITTED, new CoverOperation()),
-        new Column(MemoryMetricTable.COLUMN_TIME_BUCKET, new CoverOperation()),
+        new Column(MemoryMetricTable.COLUMN_INIT, new MinOperation()),
+        new Column(MemoryMetricTable.COLUMN_MAX, new MaxOperation()),
+        new Column(MemoryMetricTable.COLUMN_USED, new AddOperation()),
+        new Column(MemoryMetricTable.COLUMN_COMMITTED, new AddOperation()),
+        new Column(MemoryMetricTable.COLUMN_TIMES, new AddOperation()),
+        new Column(MemoryMetricTable.COLUMN_TIME_BUCKET, new NonOperation()),
     };
 
     private static final Column[] DOUBLE_COLUMNS = {
@@ -105,12 +109,20 @@ public class MemoryMetric extends StreamData {
         setDataLong(3, committed);
     }
 
-    public Long getTimeBucket() {
+    public Long getTimes() {
         return getDataLong(4);
     }
 
+    public void setTimes(Long times) {
+        setDataLong(4, times);
+    }
+
+    public Long getTimeBucket() {
+        return getDataLong(5);
+    }
+
     public void setTimeBucket(Long timeBucket) {
-        setDataLong(4, timeBucket);
+        setDataLong(5, timeBucket);
     }
 
     public Boolean getIsHeap() {

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

@@ -16,7 +16,6 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.table.jvm;
 
 import org.apache.skywalking.apm.collector.core.data.CommonTable;
@@ -32,4 +31,5 @@ public class MemoryMetricTable extends CommonTable {
     public static final String COLUMN_MAX = "max";
     public static final String COLUMN_USED = "used";
     public static final String COLUMN_COMMITTED = "committed";
+    public static final String COLUMN_TIMES = "times";
 }

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

@@ -94,6 +94,10 @@ import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceDa
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceHourMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.irmp.IInstanceReferenceMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemoryMonthMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.memorymp.IMemorySecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.mpoolmp.IMemoryPoolDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.mpoolmp.IMemoryPoolHourMetricPersistenceDAO;
@@ -177,6 +181,10 @@ import org.apache.skywalking.apm.collector.storage.es.dao.irmp.InstanceReference
 import org.apache.skywalking.apm.collector.storage.es.dao.irmp.InstanceReferenceHourMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.irmp.InstanceReferenceMinuteMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.irmp.InstanceReferenceMonthMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemoryDayMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemoryHourMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemoryMinuteMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemoryMonthMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.memorymp.MemorySecondMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.mpoolmp.MemoryPoolDayMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.mpoolmp.MemoryPoolHourMetricEsPersistenceDAO;
@@ -287,7 +295,12 @@ 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(IMemorySecondMetricPersistenceDAO.class, new MemorySecondMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IMemoryMinuteMetricPersistenceDAO.class, new MemoryMinuteMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IMemoryHourMetricPersistenceDAO.class, new MemoryHourMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IMemoryDayMetricPersistenceDAO.class, new MemoryDayMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(IMemoryMonthMetricPersistenceDAO.class, new MemoryMonthMetricEsPersistenceDAO(elasticSearchClient));
 
         this.registerServiceImplementation(IMemoryPoolSecondMetricPersistenceDAO.class, new MemoryPoolSecondMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IMemoryPoolMinuteMetricPersistenceDAO.class, new MemoryPoolMinuteMetricEsPersistenceDAO(elasticSearchClient));

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

@@ -16,11 +16,10 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.es;
 
-import org.apache.skywalking.apm.collector.core.util.Const;
 import org.apache.skywalking.apm.collector.cluster.ModuleRegistration;
+import org.apache.skywalking.apm.collector.core.util.Const;
 
 /**
  * @author peng-yongsheng

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

@@ -39,7 +39,21 @@ public abstract class AbstractMemoryMetricEsPersistenceDAO extends AbstractPersi
     }
 
     @Override protected final MemoryMetric esDataToStreamData(Map<String, Object> source) {
-        return null;
+        MemoryMetric memoryMetric = new MemoryMetric();
+        memoryMetric.setId((String)source.get(MemoryMetricTable.COLUMN_ID));
+        memoryMetric.setMetricId((String)source.get(MemoryMetricTable.COLUMN_METRIC_ID));
+
+        memoryMetric.setInstanceId(((Number)source.get(MemoryMetricTable.COLUMN_INSTANCE_ID)).intValue());
+        memoryMetric.setIsHeap((Boolean)source.get(MemoryMetricTable.COLUMN_IS_HEAP));
+
+        memoryMetric.setInit(((Number)source.get(MemoryMetricTable.COLUMN_INIT)).longValue());
+        memoryMetric.setMax(((Number)source.get(MemoryMetricTable.COLUMN_MAX)).longValue());
+        memoryMetric.setUsed(((Number)source.get(MemoryMetricTable.COLUMN_USED)).longValue());
+        memoryMetric.setCommitted(((Number)source.get(MemoryMetricTable.COLUMN_COMMITTED)).longValue());
+        memoryMetric.setTimes(((Number)source.get(MemoryMetricTable.COLUMN_TIMES)).longValue());
+
+        memoryMetric.setTimeBucket(((Number)source.get(MemoryMetricTable.COLUMN_TIME_BUCKET)).longValue());
+        return memoryMetric;
     }
 
     @Override protected final Map<String, Object> esStreamDataToEsData(MemoryMetric streamData) {
@@ -53,6 +67,7 @@ public abstract class AbstractMemoryMetricEsPersistenceDAO extends AbstractPersi
         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_TIMES, streamData.getTimes());
         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/define/MemoryMetricEsTableDefine.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/define/memory/AbstractMemoryMetricEsTableDefine.java

@@ -16,8 +16,7 @@
  *
  */
 
-
-package org.apache.skywalking.apm.collector.storage.es.define;
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
 
 import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchColumnDefine;
 import org.apache.skywalking.apm.collector.storage.es.base.define.ElasticSearchTableDefine;
@@ -26,23 +25,23 @@ import org.apache.skywalking.apm.collector.storage.table.jvm.MemoryMetricTable;
 /**
  * @author peng-yongsheng
  */
-public class MemoryMetricEsTableDefine extends ElasticSearchTableDefine {
-
-    public MemoryMetricEsTableDefine() {
-        super(MemoryMetricTable.TABLE);
-    }
+public abstract class AbstractMemoryMetricEsTableDefine extends ElasticSearchTableDefine {
 
-    @Override public int refreshInterval() {
-        return 1;
+    public AbstractMemoryMetricEsTableDefine(String name) {
+        super(name);
     }
 
-    @Override public void initialize() {
+    @Override public final void initialize() {
+        addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_ID, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_METRIC_ID, ElasticSearchColumnDefine.Type.Keyword.name()));
+        addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_INSTANCE_ID, ElasticSearchColumnDefine.Type.Integer.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_IS_HEAP, ElasticSearchColumnDefine.Type.Boolean.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_INIT, ElasticSearchColumnDefine.Type.Long.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_MAX, ElasticSearchColumnDefine.Type.Long.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_USED, ElasticSearchColumnDefine.Type.Long.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_COMMITTED, ElasticSearchColumnDefine.Type.Long.name()));
+        addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_TIMES, ElasticSearchColumnDefine.Type.Long.name()));
         addColumn(new ElasticSearchColumnDefine(MemoryMetricTable.COLUMN_TIME_BUCKET, ElasticSearchColumnDefine.Type.Long.name()));
     }
 }

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
+
+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.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryDayMetricEsTableDefine extends AbstractMemoryMetricEsTableDefine {
+
+    public MemoryDayMetricEsTableDefine() {
+        super(MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 1;
+    }
+}

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
+
+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.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryHourMetricEsTableDefine extends AbstractMemoryMetricEsTableDefine {
+
+    public MemoryHourMetricEsTableDefine() {
+        super(MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 1;
+    }
+}

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
+
+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.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMinuteMetricEsTableDefine extends AbstractMemoryMetricEsTableDefine {
+
+    public MemoryMinuteMetricEsTableDefine() {
+        super(MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 1;
+    }
+}

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
+
+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.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemoryMonthMetricEsTableDefine extends AbstractMemoryMetricEsTableDefine {
+
+    public MemoryMonthMetricEsTableDefine() {
+        super(MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 1;
+    }
+}

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.define.memory;
+
+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.table.jvm.MemoryMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public class MemorySecondMetricEsTableDefine extends AbstractMemoryMetricEsTableDefine {
+
+    public MemorySecondMetricEsTableDefine() {
+        super(MemoryMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Second.getName());
+    }
+
+    @Override public int refreshInterval() {
+        return 1;
+    }
+}

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

@@ -69,4 +69,10 @@ org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolSecondMetr
 org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolMinuteMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolHourMetricEsTableDefine
 org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolDayMetricEsTableDefine
-org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolMonthMetricEsTableDefine
\ No newline at end of file
+org.apache.skywalking.apm.collector.storage.es.define.mpool.MemoryPoolMonthMetricEsTableDefine
+
+org.apache.skywalking.apm.collector.storage.es.define.memory.MemorySecondMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.memory.MemoryMinuteMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.memory.MemoryHourMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.memory.MemoryDayMetricEsTableDefine
+org.apache.skywalking.apm.collector.storage.es.define.memory.MemoryMonthMetricEsTableDefine
\ No newline at end of file