Cpu 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

@@ -22,7 +22,17 @@ package org.apache.skywalking.apm.collector.analysis.jvm.define.graph;
  * @author peng-yongsheng
  */
 public class WorkerIdDefine {
-    public static final int CPU_METRIC_PERSISTENCE_WORKER_ID = 300;
+    public static final int CPU_METRIC_BRIDGE_NODE_ID = 3000;
+    public static final int CPU_SECOND_METRIC_PERSISTENCE_WORKER_ID = 3001;
+    public static final int CPU_MINUTE_METRIC_PERSISTENCE_WORKER_ID = 3002;
+    public static final int CPU_MINUTE_METRIC_TRANSFORM_NODE_ID = 3003;
+    public static final int CPU_HOUR_METRIC_PERSISTENCE_WORKER_ID = 3004;
+    public static final int CPU_HOUR_METRIC_TRANSFORM_NODE_ID = 3005;
+    public static final int CPU_DAY_METRIC_PERSISTENCE_WORKER_ID = 3006;
+    public static final int CPU_DAY_METRIC_TRANSFORM_NODE_ID = 3007;
+    public static final int CPU_MONTH_METRIC_PERSISTENCE_WORKER_ID = 3008;
+    public static final int CPU_MONTH_METRIC_TRANSFORM_NODE_ID = 3009;
+
     public static final int GC_METRIC_PERSISTENCE_WORKER_ID = 301;
     public static final int INST_HEART_BEAT_PERSISTENCE_WORKER_ID = 302;
     public static final int MEMORY_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

@@ -30,7 +30,7 @@ import org.apache.skywalking.apm.collector.analysis.jvm.provider.service.GCMetri
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.service.InstanceHeartBeatService;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.service.MemoryMetricService;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.service.MemoryPoolMetricService;
-import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.CpuMetricPersistenceGraph;
+import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.cpu.CpuMetricPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.GCMetricPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.InstanceHeartBeatPersistenceGraph;
 import org.apache.skywalking.apm.collector.analysis.jvm.provider.worker.MemoryMetricPersistenceGraph;

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuDayMetricPersistenceWorker.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.cpu;
+
+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.cpump.ICpuDayMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuDayMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
+
+    public CpuDayMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_DAY_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, CpuMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(ICpuDayMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuDayMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public CpuDayMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new CpuDayMetricPersistenceWorker(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/cpu/CpuDayMetricTransformNode.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.cpu;
+
+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.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuDayMetricTransformNode implements NodeProcessor<CpuMetric, CpuMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_DAY_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(CpuMetric cpuMetric, Next<CpuMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToDay(cpuMetric.getTimeBucket());
+        cpuMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + cpuMetric.getMetricId());
+        cpuMetric.setTimeBucket(timeBucket);
+
+        next.execute(cpuMetric);
+    }
+}

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuHourMetricPersistenceWorker.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.cpu;
+
+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.cpump.ICpuHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuHourMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
+
+    public CpuHourMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_HOUR_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, CpuMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(ICpuHourMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuHourMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public CpuHourMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new CpuHourMetricPersistenceWorker(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/cpu/CpuHourMetricTransformNode.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.cpu;
+
+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.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuHourMetricTransformNode implements NodeProcessor<CpuMetric, CpuMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_HOUR_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(CpuMetric cpuMetric, Next<CpuMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToHour(cpuMetric.getTimeBucket());
+        cpuMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + cpuMetric.getMetricId());
+        cpuMetric.setTimeBucket(timeBucket);
+
+        next.execute(cpuMetric);
+    }
+}

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuMetricBridgeNode.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.cpu;
+
+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.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMetricBridgeNode implements NodeProcessor<CpuMetric, CpuMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_METRIC_BRIDGE_NODE_ID;
+    }
+
+    @Override public void process(CpuMetric metric, Next<CpuMetric> 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/CpuMetricPersistenceGraph.java → apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuMetricPersistenceGraph.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.cpu;
 
 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.CpuMetric;
 
@@ -38,7 +39,21 @@ public class CpuMetricPersistenceGraph {
     }
 
     public void create() {
-        GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.CPU_METRIC_PERSISTENCE_GRAPH_ID, CpuMetric.class)
-            .addNode(new CpuMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+        Node<CpuMetric, CpuMetric> bridgeNode = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.CPU_METRIC_PERSISTENCE_GRAPH_ID, CpuMetric.class)
+            .addNode(new CpuMetricBridgeNode());
+
+        bridgeNode.addNext(new CpuSecondMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new CpuMinuteMetricTransformNode())
+            .addNext(new CpuMinuteMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new CpuHourMetricTransformNode())
+            .addNext(new CpuHourMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new CpuDayMetricTransformNode())
+            .addNext(new CpuDayMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener));
+
+        bridgeNode.addNext(new CpuMonthMetricTransformNode())
+            .addNext(new CpuMonthMetricPersistenceWorker.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/cpu/CpuMinuteMetricPersistenceWorker.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.cpu;
+
+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.cpump.ICpuMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMinuteMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
+
+    public CpuMinuteMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_MINUTE_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, CpuMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(ICpuMinuteMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuMinuteMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public CpuMinuteMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new CpuMinuteMetricPersistenceWorker(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/cpu/CpuMinuteMetricTransformNode.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.cpu;
+
+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.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMinuteMetricTransformNode implements NodeProcessor<CpuMetric, CpuMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_MINUTE_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(CpuMetric cpuMetric, Next<CpuMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToMinute(cpuMetric.getTimeBucket());
+        cpuMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + cpuMetric.getMetricId());
+        cpuMetric.setTimeBucket(timeBucket);
+
+        next.execute(cpuMetric);
+    }
+}

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuMonthMetricPersistenceWorker.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.cpu;
+
+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.cpump.ICpuMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMonthMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
+
+    public CpuMonthMetricPersistenceWorker(ModuleManager moduleManager) {
+        super(moduleManager);
+    }
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_MONTH_METRIC_PERSISTENCE_WORKER_ID;
+    }
+
+    @Override protected boolean needMergeDBData() {
+        return true;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override protected IPersistenceDAO<?, ?, CpuMetric> persistenceDAO() {
+        return getModuleManager().find(StorageModule.NAME).getService(ICpuMonthMetricPersistenceDAO.class);
+    }
+
+    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuMonthMetricPersistenceWorker> {
+
+        public Factory(ModuleManager moduleManager) {
+            super(moduleManager);
+        }
+
+        @Override public CpuMonthMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new CpuMonthMetricPersistenceWorker(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/cpu/CpuMonthMetricTransformNode.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.cpu;
+
+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.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMonthMetricTransformNode implements NodeProcessor<CpuMetric, CpuMetric> {
+
+    @Override public int id() {
+        return WorkerIdDefine.CPU_MONTH_METRIC_TRANSFORM_NODE_ID;
+    }
+
+    @Override public void process(CpuMetric cpuMetric, Next<CpuMetric> next) {
+        long timeBucket = TimeBucketUtils.INSTANCE.secondToMonth(cpuMetric.getTimeBucket());
+        cpuMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + cpuMetric.getMetricId());
+        cpuMetric.setTimeBucket(timeBucket);
+
+        next.execute(cpuMetric);
+    }
+}

apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/CpuMetricPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-jvm/jvm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/jvm/provider/worker/cpu/CpuSecondMetricPersistenceWorker.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.cpu;
 
 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.ICpuMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 
 /**
  * @author peng-yongsheng
  */
-public class CpuMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
+public class CpuSecondMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
 
-    public CpuMetricPersistenceWorker(ModuleManager moduleManager) {
+    public CpuSecondMetricPersistenceWorker(ModuleManager moduleManager) {
         super(moduleManager);
     }
 
     @Override public int id() {
-        return WorkerIdDefine.CPU_METRIC_PERSISTENCE_WORKER_ID;
+        return WorkerIdDefine.CPU_SECOND_METRIC_PERSISTENCE_WORKER_ID;
     }
 
     @Override protected boolean needMergeDBData() {
@@ -46,17 +46,17 @@ public class CpuMetricPersistenceWorker extends PersistenceWorker<CpuMetric> {
 
     @SuppressWarnings("unchecked")
     @Override protected IPersistenceDAO<?, ?, CpuMetric> persistenceDAO() {
-        return getModuleManager().find(StorageModule.NAME).getService(ICpuMetricPersistenceDAO.class);
+        return getModuleManager().find(StorageModule.NAME).getService(ICpuSecondMetricPersistenceDAO.class);
     }
 
-    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuMetricPersistenceWorker> {
+    public static class Factory extends PersistenceWorkerProvider<CpuMetric, CpuSecondMetricPersistenceWorker> {
 
         public Factory(ModuleManager moduleManager) {
             super(moduleManager);
         }
 
-        @Override public CpuMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
-            return new CpuMetricPersistenceWorker(moduleManager);
+        @Override public CpuSecondMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
+            return new CpuSecondMetricPersistenceWorker(moduleManager);
         }
 
         @Override

apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/storage/TimePyramid.java

@@ -22,7 +22,7 @@ package org.apache.skywalking.apm.collector.core.storage;
  * @author peng-yongsheng
  */
 public enum TimePyramid {
-    Minute(0, "minute"), Hour(1, "hour"), Day(2, "day"), Month(3, "month");
+    Second(0, "second"), Minute(1, "minute"), Hour(2, "hour"), Day(3, "day"), Month(4, "month");
 
     private final int value;
     private final String name;

apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/util/TimeBucketUtils.java

@@ -74,6 +74,22 @@ public enum TimeBucketUtils {
         return minuteBucket / 100 / 100 / 100;
     }
 
+    public long secondToMinute(long secondBucket) {
+        return secondBucket / 100;
+    }
+
+    public long secondToHour(long secondBucket) {
+        return secondBucket / 100 / 100;
+    }
+
+    public long secondToDay(long secondBucket) {
+        return secondBucket / 100 / 100 / 100;
+    }
+
+    public long secondToMonth(long secondBucket) {
+        return secondBucket / 100 / 100 / 100;
+    }
+
     public long changeTimeBucket2TimeStamp(String timeBucketType, long timeBucket) {
         if (TimeBucketType.SECOND.name().toLowerCase().equals(timeBucketType.toLowerCase())) {
             Calendar calendar = Calendar.getInstance();

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

@@ -35,7 +35,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlar
 import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationRegisterDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricUIDAO;
@@ -114,7 +114,7 @@ public class StorageModule extends Module {
     }
 
     private void addPersistenceDAO(List<Class> classes) {
-        classes.add(ICpuMetricPersistenceDAO.class);
+        classes.add(ICpuSecondMetricPersistenceDAO.class);
         classes.add(IGCMetricPersistenceDAO.class);
         classes.add(IMemoryMetricPersistenceDAO.class);
         classes.add(IMemoryPoolMetricPersistenceDAO.class);

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/ICpuMetricPersistenceDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cpump/ICpuDayMetricPersistenceDAO.java

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

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cpump/ICpuHourMetricPersistenceDAO.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.dao.cpump;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ICpuHourMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends CpuMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cpump/ICpuMinuteMetricPersistenceDAO.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.dao.cpump;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ICpuMinuteMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends CpuMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cpump/ICpuMonthMetricPersistenceDAO.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.dao.cpump;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ICpuMonthMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends CpuMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
+}

apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/cpump/ICpuSecondMetricPersistenceDAO.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.dao.cpump;
+
+import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+
+/**
+ * @author peng-yongsheng
+ */
+public interface ICpuSecondMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends CpuMetric> 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/CpuMetric.java

@@ -16,11 +16,10 @@
  *
  */
 
-
 package org.apache.skywalking.apm.collector.storage.table.jvm;
 
-import org.apache.skywalking.apm.collector.core.data.AbstractData;
 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.NonOperation;
@@ -28,10 +27,11 @@ import org.apache.skywalking.apm.collector.core.data.operator.NonOperation;
 /**
  * @author peng-yongsheng
  */
-public class CpuMetric extends AbstractData {
+public class CpuMetric extends StreamData {
 
     private static final Column[] STRING_COLUMNS = {
         new Column(CpuMetricTable.COLUMN_ID, new NonOperation()),
+        new Column(CpuMetricTable.COLUMN_METRIC_ID, new NonOperation()),
     };
 
     private static final Column[] LONG_COLUMNS = {
@@ -49,8 +49,25 @@ public class CpuMetric extends AbstractData {
     private static final Column[] BOOLEAN_COLUMNS = {};
     private static final Column[] BYTE_COLUMNS = {};
 
-    public CpuMetric(String id) {
-        super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS);
+    public CpuMetric() {
+        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 Integer getInstanceId() {

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

@@ -29,6 +29,7 @@ import java.util.concurrent.TimeUnit;
 import org.apache.skywalking.apm.collector.storage.dao.acp.IApplicationComponentMinutePersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ampp.IApplicationMappingMinutePersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.imp.IInstanceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.srmp.IServiceReferenceMinuteMetricPersistenceDAO;
 
@@ -74,7 +75,7 @@ public class DataTTLKeeperTimer {
     }
 
     private void deleteJVMRelatedData(long startTimestamp, long endTimestamp) {
-        ICpuMetricPersistenceDAO cpuMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(ICpuMetricPersistenceDAO.class);
+        ICpuSecondMetricPersistenceDAO cpuMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(ICpuSecondMetricPersistenceDAO.class);
         cpuMetricPersistenceDAO.deleteHistory(startTimestamp, endTimestamp);
 
         IGCMetricPersistenceDAO gcMetricPersistenceDAO = moduleManager.find(StorageModule.NAME).getService(IGCMetricPersistenceDAO.class);

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

@@ -44,7 +44,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlar
 import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationRegisterDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricUIDAO;
@@ -98,7 +98,7 @@ import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationReferenceAl
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationReferenceAlarmListEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.armp.ApplicationReferenceMinuteMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationReferenceMetricEsUIDAO;
-import org.apache.skywalking.apm.collector.storage.es.dao.CpuMetricEsPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.es.dao.cpump.CpuSecondMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.CpuMetricEsUIDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.GCMetricEsPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.es.dao.GCMetricEsUIDAO;
@@ -227,7 +227,7 @@ public class StorageModuleEsProvider extends ModuleProvider {
     }
 
     private void registerPersistenceDAO() throws ServiceNotProvidedException {
-        this.registerServiceImplementation(ICpuMetricPersistenceDAO.class, new CpuMetricEsPersistenceDAO(elasticSearchClient));
+        this.registerServiceImplementation(ICpuSecondMetricPersistenceDAO.class, new CpuSecondMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IGCMetricPersistenceDAO.class, new GCMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IMemoryMetricPersistenceDAO.class, new MemoryMetricEsPersistenceDAO(elasticSearchClient));
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricEsPersistenceDAO(elasticSearchClient));

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

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
+
+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.CpuMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+
+/**
+ * @author peng-yongsheng
+ */
+public abstract class AbstractCpuMetricEsPersistenceDAO extends AbstractPersistenceEsDAO<CpuMetric> {
+
+    AbstractCpuMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected final String timeBucketColumnNameForDelete() {
+        return CpuMetricTable.COLUMN_TIME_BUCKET;
+    }
+
+    @Override protected final CpuMetric esDataToStreamData(Map<String, Object> source) {
+        return null;
+    }
+
+    @Override protected final Map<String, Object> esStreamDataToEsData(CpuMetric streamData) {
+        Map<String, Object> source = new HashMap<>();
+        source.put(CpuMetricTable.COLUMN_ID, streamData.getId());
+        source.put(CpuMetricTable.COLUMN_METRIC_ID, streamData.getMetricId());
+
+        source.put(CpuMetricTable.COLUMN_INSTANCE_ID, streamData.getInstanceId());
+        source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, streamData.getUsagePercent());
+        source.put(CpuMetricTable.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/CpuMetricEsPersistenceDAO.java → apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/cpump/CpuDayMetricEsPersistenceDAO.java

@@ -16,61 +16,27 @@
  *
  */
 
-package org.apache.skywalking.apm.collector.storage.es.dao;
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
 
-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.ICpuMetricPersistenceDAO;
-import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO;
+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.cpump.ICpuDayMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
 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 CpuMetricEsPersistenceDAO extends EsDAO implements ICpuMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+public class CpuDayMetricEsPersistenceDAO extends AbstractCpuMetricEsPersistenceDAO implements ICpuDayMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
 
-    private final Logger logger = LoggerFactory.getLogger(CpuMetricEsPersistenceDAO.class);
-
-    public CpuMetricEsPersistenceDAO(ElasticSearchClient client) {
+    public CpuDayMetricEsPersistenceDAO(ElasticSearchClient client) {
         super(client);
     }
 
-    @Override public CpuMetric get(String id) {
-        return null;
-    }
-
-    @Override public IndexRequestBuilder prepareBatchInsert(CpuMetric cpuMetric) {
-        Map<String, Object> source = new HashMap<>();
-        source.put(CpuMetricTable.COLUMN_INSTANCE_ID, cpuMetric.getInstanceId());
-        source.put(CpuMetricTable.COLUMN_USAGE_PERCENT, cpuMetric.getUsagePercent());
-        source.put(CpuMetricTable.COLUMN_TIME_BUCKET, cpuMetric.getTimeBucket());
-
-        logger.debug("prepare cpu metric batch insert, getId: {}", cpuMetric.getId());
-        return getClient().prepareIndex(CpuMetricTable.TABLE, cpuMetric.getId()).setSource(source);
-    }
-
-    @Override public UpdateRequestBuilder prepareBatchUpdate(CpuMetric cpuMetric) {
-        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(CpuMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket))
-            .source(CpuMetricTable.TABLE)
-            .get();
-
-        long deleted = response.getDeleted();
-        logger.info("Delete {} rows history from {} index.", deleted, CpuMetricTable.TABLE);
+    @Override protected String tableName() {
+        return CpuMetricTable.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/cpump/CpuHourMetricEsPersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
+
+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.cpump.ICpuHourMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuHourMetricEsPersistenceDAO extends AbstractCpuMetricEsPersistenceDAO implements ICpuHourMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+
+    public CpuHourMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return CpuMetricTable.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/cpump/CpuMinuteMetricEsPersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
+
+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.cpump.ICpuMinuteMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMinuteMetricEsPersistenceDAO extends AbstractCpuMetricEsPersistenceDAO implements ICpuMinuteMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+
+    public CpuMinuteMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return CpuMetricTable.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/cpump/CpuMonthMetricEsPersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
+
+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.cpump.ICpuMonthMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuMonthMetricEsPersistenceDAO extends AbstractCpuMetricEsPersistenceDAO implements ICpuMonthMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+
+    public CpuMonthMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return CpuMetricTable.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/cpump/CpuSecondMetricEsPersistenceDAO.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.apm.collector.storage.es.dao.cpump;
+
+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.cpump.ICpuSecondMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
+import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.action.update.UpdateRequestBuilder;
+
+/**
+ * @author peng-yongsheng
+ */
+public class CpuSecondMetricEsPersistenceDAO extends AbstractCpuMetricEsPersistenceDAO implements ICpuSecondMetricPersistenceDAO<IndexRequestBuilder, UpdateRequestBuilder, CpuMetric> {
+
+    public CpuSecondMetricEsPersistenceDAO(ElasticSearchClient client) {
+        super(client);
+    }
+
+    @Override protected String tableName() {
+        return CpuMetricTable.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

@@ -40,7 +40,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlar
 import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IApplicationRegisterDAO;
-import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricUIDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.dao.IGCMetricUIDAO;
@@ -94,7 +94,7 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceAl
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceAlarmListH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceMinuteMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceMetricH2UIDAO;
-import org.apache.skywalking.apm.collector.storage.h2.dao.CpuMetricH2PersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.h2.dao.CpuSecondMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.CpuMetricH2UIDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.GCMetricH2PersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.dao.GCMetricH2UIDAO;
@@ -205,7 +205,7 @@ public class StorageModuleH2Provider extends ModuleProvider {
     }
 
     private void registerPersistenceDAO() throws ServiceNotProvidedException {
-        this.registerServiceImplementation(ICpuMetricPersistenceDAO.class, new CpuMetricH2PersistenceDAO(h2Client));
+        this.registerServiceImplementation(ICpuSecondMetricPersistenceDAO.class, new CpuSecondMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IGCMetricPersistenceDAO.class, new GCMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IMemoryMetricPersistenceDAO.class, new MemoryMetricH2PersistenceDAO(h2Client));
         this.registerServiceImplementation(IMemoryPoolMetricPersistenceDAO.class, new MemoryPoolMetricH2PersistenceDAO(h2Client));

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

@@ -26,7 +26,7 @@ import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetric;
 import org.apache.skywalking.apm.collector.storage.table.jvm.CpuMetricTable;
 import org.apache.skywalking.apm.collector.client.h2.H2Client;
-import org.apache.skywalking.apm.collector.storage.dao.ICpuMetricPersistenceDAO;
+import org.apache.skywalking.apm.collector.storage.dao.cpump.ICpuSecondMetricPersistenceDAO;
 import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -34,11 +34,11 @@ import org.slf4j.LoggerFactory;
 /**
  * @author peng-yongsheng, clevertension
  */
-public class CpuMetricH2PersistenceDAO extends H2DAO implements ICpuMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, CpuMetric> {
+public class CpuSecondMetricH2PersistenceDAO extends H2DAO implements ICpuSecondMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, CpuMetric> {
 
-    private final Logger logger = LoggerFactory.getLogger(CpuMetricH2PersistenceDAO.class);
+    private final Logger logger = LoggerFactory.getLogger(CpuSecondMetricH2PersistenceDAO.class);
 
-    public CpuMetricH2PersistenceDAO(H2Client client) {
+    public CpuSecondMetricH2PersistenceDAO(H2Client client) {
         super(client);
     }