From a09aee872baf37c3c446cd5f52282d4e0aa05707 Mon Sep 17 00:00:00 2001 From: peng-yongsheng <8082209@qq.com> Date: Sun, 7 Jan 2018 20:32:00 +0800 Subject: [PATCH] Application metric pyramid aggregate. --- .../ApplicationMetricAlarmGraph.java | 2 +- .../define/graph/MetricWorkerIdDefine.java | 12 +- .../AnalysisMetricModuleProvider.java | 2 +- ...pplicationDayMetricPersistenceWorker.java} | 18 +-- .../ApplicationDayMetricTransformNode.java | 45 ++++++ ...pplicationHourMetricPersistenceWorker.java | 67 ++++++++ .../ApplicationHourMetricTransformNode.java | 45 ++++++ .../ApplicationMetricAggregationWorker.java | 15 +- .../{ => metric}/ApplicationMetricGraph.java | 20 ++- .../ApplicationMetricRemoteWorker.java | 4 +- ...licationMinuteMetricPersistenceWorker.java | 67 ++++++++ ...plicationMonthMetricPersistenceWorker.java | 67 ++++++++ .../ApplicationMonthMetricTransformNode.java | 45 ++++++ .../apm/collector/storage/StorageModule.java | 4 +- .../IApplicationDayMetricPersistenceDAO.java} | 4 +- .../IApplicationHourMetricPersistenceDAO.java | 28 ++++ ...ApplicationMinuteMetricPersistenceDAO.java | 28 ++++ ...IApplicationMonthMetricPersistenceDAO.java | 28 ++++ .../table/application/ApplicationMetric.java | 25 ++- .../storage/es/StorageModuleEsProvider.java | 6 +- .../ApplicationMetricEsPersistenceDAO.java | 147 ------------------ ...ractApplicationMetricEsPersistenceDAO.java | 102 ++++++++++++ .../ApplicationDayMetricEsPersistenceDAO.java | 42 +++++ ...ApplicationHourMetricEsPersistenceDAO.java | 42 +++++ ...plicationMinuteMetricEsPersistenceDAO.java | 42 +++++ ...pplicationMonthMetricEsPersistenceDAO.java | 42 +++++ .../storage/h2/StorageModuleH2Provider.java | 6 +- ...licationMinuteMetricH2PersistenceDAO.java} | 8 +- 28 files changed, 773 insertions(+), 190 deletions(-) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/{ApplicationMetricPersistenceWorker.java => metric/ApplicationDayMetricPersistenceWorker.java} (75%) create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricTransformNode.java create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricPersistenceWorker.java create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricTransformNode.java rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/{ => metric}/ApplicationMetricAggregationWorker.java (93%) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/{ => metric}/ApplicationMetricGraph.java (76%) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/{ => metric}/ApplicationMetricRemoteWorker.java (95%) create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMinuteMetricPersistenceWorker.java create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricPersistenceWorker.java create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricTransformNode.java rename apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/{IApplicationMetricPersistenceDAO.java => amp/IApplicationDayMetricPersistenceDAO.java} (81%) create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationHourMetricPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMinuteMetricPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMonthMetricPersistenceDAO.java delete mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ApplicationMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/AbstractApplicationMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationDayMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationHourMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMinuteMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMonthMetricEsPersistenceDAO.java rename apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/{ApplicationMetricH2PersistenceDAO.java => ApplicationMinuteMetricH2PersistenceDAO.java} (96%) diff --git a/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/application/ApplicationMetricAlarmGraph.java b/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/application/ApplicationMetricAlarmGraph.java index 01b01af27..0733dbab0 100644 --- a/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/application/ApplicationMetricAlarmGraph.java +++ b/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/application/ApplicationMetricAlarmGraph.java @@ -64,7 +64,7 @@ public class ApplicationMetricAlarmGraph { private void link(Graph graph) { GraphManager.INSTANCE.findGraph(MetricGraphIdDefine.APPLICATION_METRIC_GRAPH_ID, ApplicationMetric.class) - .toFinder().findNode(MetricWorkerIdDefine.APPLICATION_METRIC_PERSISTENCE_WORKER_ID, ApplicationMetric.class) + .toFinder().findNode(MetricWorkerIdDefine.APPLICATION_MINUTE_METRIC_PERSISTENCE_WORKER_ID, ApplicationMetric.class) .addNext(new NodeProcessor() { @Override public int id() { return AlarmWorkerIdDefine.APPLICATION_METRIC_ALARM_GRAPH_BRIDGE_WORKER_ID; diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java index 26466ab69..eeba949ba 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java @@ -72,9 +72,15 @@ public class MetricWorkerIdDefine { public static final int INSTANCE_MONTH_METRIC_PERSISTENCE_WORKER_ID = 4507; public static final int INSTANCE_MONTH_METRIC_TRANSFORM_NODE_ID = 4508; - public static final int APPLICATION_METRIC_AGGREGATION_WORKER_ID = 415; - public static final int APPLICATION_METRIC_REMOTE_WORKER_ID = 416; - public static final int APPLICATION_METRIC_PERSISTENCE_WORKER_ID = 417; + public static final int APPLICATION_MINUTE_METRIC_AGGREGATION_WORKER_ID = 4600; + public static final int APPLICATION_MINUTE_METRIC_REMOTE_WORKER_ID = 4601; + public static final int APPLICATION_MINUTE_METRIC_PERSISTENCE_WORKER_ID = 4602; + public static final int APPLICATION_HOUR_METRIC_PERSISTENCE_WORKER_ID = 4603; + public static final int APPLICATION_HOUR_METRIC_TRANSFORM_NODE_ID = 4604; + public static final int APPLICATION_DAY_METRIC_PERSISTENCE_WORKER_ID = 4605; + public static final int APPLICATION_DAY_METRIC_TRANSFORM_NODE_ID = 4606; + public static final int APPLICATION_MONTH_METRIC_PERSISTENCE_WORKER_ID = 4607; + public static final int APPLICATION_MONTH_METRIC_TRANSFORM_NODE_ID = 4608; public static final int APPLICATION_COMPONENT_AGGREGATION_WORKER_ID = 418; public static final int APPLICATION_COMPONENT_REMOTE_WORKER_ID = 419; diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java index 55a70f157..af0838eb4 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java @@ -24,7 +24,7 @@ import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.appli import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentSpanListener; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingSpanListener; -import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric.ApplicationMetricGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.refmetric.ApplicationReferenceMetricGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceSpanListener; diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricPersistenceWorker.java similarity index 75% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricPersistenceWorker.java index 2044fb6f7..4a64714d9 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricPersistenceWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric; import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker; @@ -24,39 +24,39 @@ 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.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.amp.IApplicationDayMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; /** * @author peng-yongsheng */ -public class ApplicationMetricPersistenceWorker extends PersistenceWorker { +public class ApplicationDayMetricPersistenceWorker extends PersistenceWorker { - public ApplicationMetricPersistenceWorker(ModuleManager moduleManager) { + public ApplicationDayMetricPersistenceWorker(ModuleManager moduleManager) { super(moduleManager); } @Override public int id() { - return MetricWorkerIdDefine.APPLICATION_METRIC_PERSISTENCE_WORKER_ID; + return MetricWorkerIdDefine.APPLICATION_DAY_METRIC_PERSISTENCE_WORKER_ID; } @SuppressWarnings("unchecked") @Override protected IPersistenceDAO persistenceDAO() { - return getModuleManager().find(StorageModule.NAME).getService(IApplicationMetricPersistenceDAO.class); + return getModuleManager().find(StorageModule.NAME).getService(IApplicationDayMetricPersistenceDAO.class); } @Override protected boolean needMergeDBData() { return true; } - public static class Factory extends PersistenceWorkerProvider { + public static class Factory extends PersistenceWorkerProvider { public Factory(ModuleManager moduleManager) { super(moduleManager); } - @Override public ApplicationMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { - return new ApplicationMetricPersistenceWorker(moduleManager); + @Override public ApplicationDayMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ApplicationDayMetricPersistenceWorker(moduleManager); } @Override diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricTransformNode.java new file mode 100644 index 000000000..2b4e43e99 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationDayMetricTransformNode.java @@ -0,0 +1,45 @@ +/* + * 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.metric.provider.worker.application.metric; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; +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.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationDayMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_DAY_METRIC_TRANSFORM_NODE_ID; + } + + @Override + public void process(ApplicationMetric applicationMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToDay(applicationMetric.getTimeBucket()); + applicationMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + applicationMetric.getMetricId()); + applicationMetric.setTimeBucket(timeBucket); + + next.execute(applicationMetric); + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricPersistenceWorker.java new file mode 100644 index 000000000..4f7d4e7b2 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricPersistenceWorker.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; +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.amp.IApplicationHourMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationHourMetricPersistenceWorker extends PersistenceWorker { + + public ApplicationHourMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_HOUR_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IApplicationHourMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ApplicationHourMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ApplicationHourMetricPersistenceWorker(moduleManager); + } + + @Override + public int queueSize() { + return 1024; + } + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricTransformNode.java new file mode 100644 index 000000000..0a3f20dea --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationHourMetricTransformNode.java @@ -0,0 +1,45 @@ +/* + * 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.metric.provider.worker.application.metric; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; +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.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationHourMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_HOUR_METRIC_TRANSFORM_NODE_ID; + } + + @Override + public void process(ApplicationMetric applicationMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToHour(applicationMetric.getTimeBucket()); + applicationMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + applicationMetric.getMetricId()); + applicationMetric.setTimeBucket(timeBucket); + + next.execute(applicationMetric); + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricAggregationWorker.java similarity index 93% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricAggregationWorker.java index 0580dd38f..47a7ed72a 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricAggregationWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric; import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; @@ -36,18 +36,23 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker graph = GraphManager.INSTANCE.createIfAbsent(MetricGraphIdDefine.APPLICATION_METRIC_GRAPH_ID, ApplicationReferenceMetric.class); - graph.addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener)) - .addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(workerCreateListener)) - .addNext(new ApplicationMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + Node remoteNode = graph.addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener)) + .addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(workerCreateListener)); + + remoteNode.addNext(new ApplicationMinuteMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ApplicationHourMetricTransformNode()) + .addNext(new ApplicationHourMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ApplicationDayMetricTransformNode()) + .addNext(new ApplicationDayMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ApplicationMonthMetricTransformNode()) + .addNext(new ApplicationMonthMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); link(graph); } diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricRemoteWorker.java similarity index 95% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricRemoteWorker.java index 543d5fbc0..bc87fe4ad 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMetricRemoteWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric; import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker; @@ -37,7 +37,7 @@ public class ApplicationMetricRemoteWorker extends AbstractRemoteWorker { + + public ApplicationMinuteMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_MINUTE_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IApplicationMinuteMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ApplicationMinuteMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ApplicationMinuteMetricPersistenceWorker(moduleManager); + } + + @Override + public int queueSize() { + return 1024; + } + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricPersistenceWorker.java new file mode 100644 index 000000000..8419cdb9c --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricPersistenceWorker.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.metric; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; +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.amp.IApplicationMonthMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationMonthMetricPersistenceWorker extends PersistenceWorker { + + public ApplicationMonthMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_MONTH_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IApplicationMonthMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ApplicationMonthMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ApplicationMonthMetricPersistenceWorker(moduleManager); + } + + @Override + public int queueSize() { + return 1024; + } + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricTransformNode.java new file mode 100644 index 000000000..7eca20928 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/metric/ApplicationMonthMetricTransformNode.java @@ -0,0 +1,45 @@ +/* + * 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.metric.provider.worker.application.metric; + +import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; +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.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public class ApplicationMonthMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.APPLICATION_MONTH_METRIC_TRANSFORM_NODE_ID; + } + + @Override + public void process(ApplicationMetric applicationMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToMonth(applicationMetric.getTimeBucket()); + applicationMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + applicationMetric.getMetricId()); + applicationMetric.setTimeBucket(timeBucket); + + next.execute(applicationMetric); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java index ac67b3320..141f9a89d 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/StorageModule.java @@ -29,7 +29,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPers import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.amp.IApplicationMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmListPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO; @@ -122,7 +122,7 @@ public class StorageModule extends Module { classes.add(IGlobalTracePersistenceDAO.class); classes.add(IApplicationComponentPersistenceDAO.class); classes.add(IApplicationMappingPersistenceDAO.class); - classes.add(IApplicationMetricPersistenceDAO.class); + classes.add(IApplicationMinuteMetricPersistenceDAO.class); classes.add(IApplicationReferenceMinuteMetricPersistenceDAO.class); classes.add(ISegmentCostPersistenceDAO.class); classes.add(ISegmentPersistenceDAO.class); diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IApplicationMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationDayMetricPersistenceDAO.java similarity index 81% rename from apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IApplicationMetricPersistenceDAO.java rename to apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationDayMetricPersistenceDAO.java index 9ee02ec56..5263b64a7 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IApplicationMetricPersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationDayMetricPersistenceDAO.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.storage.dao; +package org.apache.skywalking.apm.collector.storage.dao.amp; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; @@ -24,5 +24,5 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application /** * @author peng-yongsheng */ -public interface IApplicationMetricPersistenceDAO extends IPersistenceDAO { +public interface IApplicationDayMetricPersistenceDAO extends IPersistenceDAO { } diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationHourMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationHourMetricPersistenceDAO.java new file mode 100644 index 000000000..13adca491 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationHourMetricPersistenceDAO.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.dao.amp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public interface IApplicationHourMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMinuteMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMinuteMetricPersistenceDAO.java new file mode 100644 index 000000000..55459d1f9 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMinuteMetricPersistenceDAO.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.dao.amp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public interface IApplicationMinuteMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMonthMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMonthMetricPersistenceDAO.java new file mode 100644 index 000000000..e6547d84a --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/amp/IApplicationMonthMetricPersistenceDAO.java @@ -0,0 +1,28 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.dao.amp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; + +/** + * @author peng-yongsheng + */ +public interface IApplicationMonthMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMetric.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMetric.java index 80188bf2b..08883b214 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMetric.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/application/ApplicationMetric.java @@ -19,7 +19,7 @@ package org.apache.skywalking.apm.collector.storage.table.application; import org.apache.skywalking.apm.collector.core.data.Column; -import org.apache.skywalking.apm.collector.core.data.AbstractData; +import org.apache.skywalking.apm.collector.core.data.StreamData; import org.apache.skywalking.apm.collector.core.data.operator.AddOperation; import org.apache.skywalking.apm.collector.core.data.operator.NonOperation; import org.apache.skywalking.apm.collector.storage.table.Metric; @@ -27,10 +27,11 @@ import org.apache.skywalking.apm.collector.storage.table.Metric; /** * @author peng-yongsheng */ -public class ApplicationMetric extends AbstractData implements Metric { +public class ApplicationMetric extends StreamData implements Metric { private static final Column[] STRING_COLUMNS = { new Column(ApplicationMetricTable.COLUMN_ID, new NonOperation()), + new Column(ApplicationMetricTable.COLUMN_METRIC_ID, new NonOperation()), }; private static final Column[] LONG_COLUMNS = { @@ -64,8 +65,24 @@ public class ApplicationMetric extends AbstractData implements Metric { private static final Column[] BYTE_COLUMNS = {}; - public ApplicationMetric(String id) { - super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + public ApplicationMetric() { + 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); } @Override diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java index 11d233082..10a89c179 100644 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/StorageModuleEsProvider.java @@ -38,7 +38,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPers import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.amp.IApplicationMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmListPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO; @@ -93,7 +93,7 @@ import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationEsCacheDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationEsRegisterDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationMappingEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationMappingEsUIDAO; -import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationMetricEsPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.es.dao.amp.ApplicationMinuteMetricEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationReferenceAlarmEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ApplicationReferenceAlarmListEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.armp.ApplicationReferenceMinuteMetricEsPersistenceDAO; @@ -235,7 +235,7 @@ public class StorageModuleEsProvider extends ModuleProvider { this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IApplicationComponentPersistenceDAO.class, new ApplicationComponentEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IApplicationMappingPersistenceDAO.class, new ApplicationMappingEsPersistenceDAO(elasticSearchClient)); - this.registerServiceImplementation(IApplicationMetricPersistenceDAO.class, new ApplicationMetricEsPersistenceDAO(elasticSearchClient)); + this.registerServiceImplementation(IApplicationMinuteMetricPersistenceDAO.class, new ApplicationMinuteMetricEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IApplicationReferenceMinuteMetricPersistenceDAO.class, new ApplicationReferenceMinuteMetricEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(ISegmentCostPersistenceDAO.class, new SegmentCostEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentEsPersistenceDAO(elasticSearchClient)); diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ApplicationMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ApplicationMetricEsPersistenceDAO.java deleted file mode 100644 index 3b2d08760..000000000 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ApplicationMetricEsPersistenceDAO.java +++ /dev/null @@ -1,147 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -package org.apache.skywalking.apm.collector.storage.es.dao; - -import java.util.HashMap; -import java.util.Map; -import org.apache.skywalking.apm.collector.client.elasticsearch.ElasticSearchClient; -import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; -import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; -import org.elasticsearch.action.get.GetResponse; -import org.elasticsearch.action.index.IndexRequestBuilder; -import org.elasticsearch.action.update.UpdateRequestBuilder; -import org.elasticsearch.index.query.QueryBuilders; -import org.elasticsearch.index.reindex.BulkByScrollResponse; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -/** - * @author peng-yongsheng - */ -public class ApplicationMetricEsPersistenceDAO extends EsDAO implements IApplicationMetricPersistenceDAO { - - private final Logger logger = LoggerFactory.getLogger(ApplicationMetricEsPersistenceDAO.class); - - public ApplicationMetricEsPersistenceDAO(ElasticSearchClient client) { - super(client); - } - - @Override public ApplicationMetric get(String id) { - GetResponse getResponse = getClient().prepareGet(ApplicationMetricTable.TABLE, id).get(); - if (getResponse.isExists()) { - ApplicationMetric applicationMetric = new ApplicationMetric(id); - Map source = getResponse.getSource(); - applicationMetric.setApplicationId(((Number)source.get(ApplicationMetricTable.COLUMN_APPLICATION_ID)).intValue()); - applicationMetric.setSourceValue(((Number)source.get(ApplicationMetricTable.COLUMN_SOURCE_VALUE)).intValue()); - - applicationMetric.setTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); - applicationMetric.setTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); - applicationMetric.setTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); - applicationMetric.setTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - applicationMetric.setBusinessTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); - applicationMetric.setBusinessTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); - applicationMetric.setBusinessTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); - applicationMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - applicationMetric.setMqTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); - applicationMetric.setMqTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); - applicationMetric.setMqTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); - applicationMetric.setMqTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - applicationMetric.setSatisfiedCount(((Number)source.get(ApplicationMetricTable.COLUMN_SATISFIED_COUNT)).longValue()); - applicationMetric.setToleratingCount(((Number)source.get(ApplicationMetricTable.COLUMN_TOLERATING_COUNT)).longValue()); - applicationMetric.setFrustratedCount(((Number)source.get(ApplicationMetricTable.COLUMN_FRUSTRATED_COUNT)).longValue()); - applicationMetric.setTimeBucket(((Number)source.get(ApplicationMetricTable.COLUMN_TIME_BUCKET)).longValue()); - return applicationMetric; - } else { - return null; - } - } - - @Override public IndexRequestBuilder prepareBatchInsert(ApplicationMetric data) { - Map source = new HashMap<>(); - source.put(ApplicationMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(ApplicationMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); - - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_SATISFIED_COUNT, data.getSatisfiedCount()); - source.put(ApplicationMetricTable.COLUMN_TOLERATING_COUNT, data.getToleratingCount()); - source.put(ApplicationMetricTable.COLUMN_FRUSTRATED_COUNT, data.getFrustratedCount()); - source.put(ApplicationMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket()); - - return getClient().prepareIndex(ApplicationMetricTable.TABLE, data.getId()).setSource(source); - } - - @Override public UpdateRequestBuilder prepareBatchUpdate(ApplicationMetric data) { - Map source = new HashMap<>(); - source.put(ApplicationMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(ApplicationMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); - - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); - - source.put(ApplicationMetricTable.COLUMN_SATISFIED_COUNT, data.getSatisfiedCount()); - source.put(ApplicationMetricTable.COLUMN_TOLERATING_COUNT, data.getToleratingCount()); - source.put(ApplicationMetricTable.COLUMN_FRUSTRATED_COUNT, data.getFrustratedCount()); - - return getClient().prepareUpdate(ApplicationMetricTable.TABLE, data.getId()).setDoc(source); - } - - @Override public void deleteHistory(Long startTimestamp, Long endTimestamp) { - long startTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(startTimestamp); - long endTimeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(endTimestamp); - BulkByScrollResponse response = getClient().prepareDelete() - .filter(QueryBuilders.rangeQuery(ApplicationMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket)) - .source(ApplicationMetricTable.TABLE) - .get(); - - long deleted = response.getDeleted(); - logger.info("Delete {} rows history from {} index.", deleted, ApplicationMetricTable.TABLE); - } -} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/AbstractApplicationMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/AbstractApplicationMetricEsPersistenceDAO.java new file mode 100644 index 000000000..55c66f7bd --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/AbstractApplicationMetricEsPersistenceDAO.java @@ -0,0 +1,102 @@ +/* + * 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.amp; + +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.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; + +/** + * @author peng-yongsheng + */ +public abstract class AbstractApplicationMetricEsPersistenceDAO extends AbstractPersistenceEsDAO { + + AbstractApplicationMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected final String timeBucketColumnNameForDelete() { + return ApplicationMetricTable.COLUMN_TIME_BUCKET; + } + + @Override protected final ApplicationMetric esDataToStreamData(Map source) { + ApplicationMetric applicationMetric = new ApplicationMetric(); + applicationMetric.setId((String)source.get(ApplicationMetricTable.COLUMN_ID)); + applicationMetric.setMetricId((String)source.get(ApplicationMetricTable.COLUMN_METRIC_ID)); + + applicationMetric.setApplicationId(((Number)source.get(ApplicationMetricTable.COLUMN_APPLICATION_ID)).intValue()); + applicationMetric.setSourceValue(((Number)source.get(ApplicationMetricTable.COLUMN_SOURCE_VALUE)).intValue()); + + applicationMetric.setTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); + applicationMetric.setTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); + applicationMetric.setTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); + applicationMetric.setTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + applicationMetric.setBusinessTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); + applicationMetric.setBusinessTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); + applicationMetric.setBusinessTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); + applicationMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + applicationMetric.setMqTransactionCalls(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); + applicationMetric.setMqTransactionErrorCalls(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); + applicationMetric.setMqTransactionDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); + applicationMetric.setMqTransactionErrorDurationSum(((Number)source.get(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + applicationMetric.setSatisfiedCount(((Number)source.get(ApplicationMetricTable.COLUMN_SATISFIED_COUNT)).longValue()); + applicationMetric.setToleratingCount(((Number)source.get(ApplicationMetricTable.COLUMN_TOLERATING_COUNT)).longValue()); + applicationMetric.setFrustratedCount(((Number)source.get(ApplicationMetricTable.COLUMN_FRUSTRATED_COUNT)).longValue()); + applicationMetric.setTimeBucket(((Number)source.get(ApplicationMetricTable.COLUMN_TIME_BUCKET)).longValue()); + + return applicationMetric; + } + + @Override protected final Map esStreamDataToEsData(ApplicationMetric streamData) { + Map source = new HashMap<>(); + source.put(ApplicationMetricTable.COLUMN_ID, streamData.getId()); + source.put(ApplicationMetricTable.COLUMN_METRIC_ID, streamData.getMetricId()); + + source.put(ApplicationMetricTable.COLUMN_APPLICATION_ID, streamData.getApplicationId()); + source.put(ApplicationMetricTable.COLUMN_SOURCE_VALUE, streamData.getSourceValue()); + + source.put(ApplicationMetricTable.COLUMN_TRANSACTION_CALLS, streamData.getTransactionCalls()); + source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, streamData.getTransactionErrorCalls()); + source.put(ApplicationMetricTable.COLUMN_TRANSACTION_DURATION_SUM, streamData.getTransactionDurationSum()); + source.put(ApplicationMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, streamData.getTransactionErrorDurationSum()); + + source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, streamData.getBusinessTransactionCalls()); + source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, streamData.getBusinessTransactionErrorCalls()); + source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, streamData.getBusinessTransactionDurationSum()); + source.put(ApplicationMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, streamData.getBusinessTransactionErrorDurationSum()); + + source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_CALLS, streamData.getMqTransactionCalls()); + source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, streamData.getMqTransactionErrorCalls()); + source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, streamData.getMqTransactionDurationSum()); + source.put(ApplicationMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, streamData.getMqTransactionErrorDurationSum()); + + source.put(ApplicationMetricTable.COLUMN_SATISFIED_COUNT, streamData.getSatisfiedCount()); + source.put(ApplicationMetricTable.COLUMN_TOLERATING_COUNT, streamData.getToleratingCount()); + source.put(ApplicationMetricTable.COLUMN_FRUSTRATED_COUNT, streamData.getFrustratedCount()); + source.put(ApplicationMetricTable.COLUMN_TIME_BUCKET, streamData.getTimeBucket()); + + return source; + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationDayMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationDayMetricEsPersistenceDAO.java new file mode 100644 index 000000000..8d77c13e1 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationDayMetricEsPersistenceDAO.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.es.dao.amp; + +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.amp.IApplicationDayMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ApplicationDayMetricEsPersistenceDAO extends AbstractApplicationMetricEsPersistenceDAO implements IApplicationDayMetricPersistenceDAO { + + public ApplicationDayMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ApplicationMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName(); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationHourMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationHourMetricEsPersistenceDAO.java new file mode 100644 index 000000000..c7a21b617 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationHourMetricEsPersistenceDAO.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.es.dao.amp; + +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.amp.IApplicationHourMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ApplicationHourMetricEsPersistenceDAO extends AbstractApplicationMetricEsPersistenceDAO implements IApplicationHourMetricPersistenceDAO { + + public ApplicationHourMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ApplicationMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName(); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMinuteMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMinuteMetricEsPersistenceDAO.java new file mode 100644 index 000000000..8008e316b --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMinuteMetricEsPersistenceDAO.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.es.dao.amp; + +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.amp.IApplicationMinuteMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ApplicationMinuteMetricEsPersistenceDAO extends AbstractApplicationMetricEsPersistenceDAO implements IApplicationMinuteMetricPersistenceDAO { + + public ApplicationMinuteMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ApplicationMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName(); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMonthMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMonthMetricEsPersistenceDAO.java new file mode 100644 index 000000000..b644d3b40 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/amp/ApplicationMonthMetricEsPersistenceDAO.java @@ -0,0 +1,42 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.storage.es.dao.amp; + +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.amp.IApplicationMonthMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; +import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ApplicationMonthMetricEsPersistenceDAO extends AbstractApplicationMetricEsPersistenceDAO implements IApplicationMonthMetricPersistenceDAO { + + public ApplicationMonthMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ApplicationMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName(); + } +} diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java index 1eb18824e..86633d0e4 100644 --- a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/StorageModuleH2Provider.java @@ -34,7 +34,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentPers import org.apache.skywalking.apm.collector.storage.dao.IApplicationComponentUIDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationMappingUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.amp.IApplicationMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmListPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IApplicationReferenceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.armp.IApplicationReferenceMinuteMetricPersistenceDAO; @@ -89,7 +89,7 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationH2CacheDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationH2RegisterDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationMappingH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationMappingH2UIDAO; -import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationMetricH2PersistenceDAO; +import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationMinuteMetricH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceAlarmH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceAlarmListH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ApplicationReferenceMinuteMetricH2PersistenceDAO; @@ -213,7 +213,7 @@ public class StorageModuleH2Provider extends ModuleProvider { this.registerServiceImplementation(IGlobalTracePersistenceDAO.class, new GlobalTraceH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IApplicationComponentPersistenceDAO.class, new ApplicationComponentH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IApplicationMappingPersistenceDAO.class, new ApplicationMappingH2PersistenceDAO(h2Client)); - this.registerServiceImplementation(IApplicationMetricPersistenceDAO.class, new ApplicationMetricH2PersistenceDAO(h2Client)); + this.registerServiceImplementation(IApplicationMinuteMetricPersistenceDAO.class, new ApplicationMinuteMetricH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IApplicationReferenceMinuteMetricPersistenceDAO.class, new ApplicationReferenceMinuteMetricH2PersistenceDAO(h2Client)); this.registerServiceImplementation(ISegmentCostPersistenceDAO.class, new SegmentCostH2PersistenceDAO(h2Client)); this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentH2PersistenceDAO(h2Client)); diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMetricH2PersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMinuteMetricH2PersistenceDAO.java similarity index 96% rename from apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMetricH2PersistenceDAO.java rename to apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMinuteMetricH2PersistenceDAO.java index 7ac10e014..f8fda212a 100644 --- a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMetricH2PersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ApplicationMinuteMetricH2PersistenceDAO.java @@ -27,7 +27,7 @@ import java.util.List; import java.util.Map; import org.apache.skywalking.apm.collector.client.h2.H2Client; import org.apache.skywalking.apm.collector.storage.base.sql.SqlBuilder; -import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.amp.IApplicationMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO; import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity; import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric; @@ -39,12 +39,12 @@ import org.slf4j.LoggerFactory; /** * @author peng-yongsheng, clevertension */ -public class ApplicationMetricH2PersistenceDAO extends H2DAO implements IApplicationMetricPersistenceDAO { +public class ApplicationMinuteMetricH2PersistenceDAO extends H2DAO implements IApplicationMinuteMetricPersistenceDAO { - private final Logger logger = LoggerFactory.getLogger(ApplicationMetricH2PersistenceDAO.class); + private final Logger logger = LoggerFactory.getLogger(ApplicationMinuteMetricH2PersistenceDAO.class); private static final String GET_SQL = "select * from {0} where {1} = ?"; - public ApplicationMetricH2PersistenceDAO(H2Client client) { + public ApplicationMinuteMetricH2PersistenceDAO(H2Client client) { super(client); }