From f3fc91926e6795b41293e9dc0a8f32969dd98256 Mon Sep 17 00:00:00 2001 From: peng-yongsheng <8082209@qq.com> Date: Sun, 7 Jan 2018 20:00:30 +0800 Subject: [PATCH] Service metric pyramid aggregate. --- .../service/ServiceMetricAlarmGraph.java | 2 +- .../define/graph/MetricWorkerIdDefine.java | 12 +- .../AnalysisMetricModuleProvider.java | 2 +- .../ServiceDayMetricPersistenceWorker.java} | 18 +-- .../metric/ServiceDayMetricTransformNode.java | 44 ++++++ .../ServiceHourMetricPersistenceWorker.java | 67 ++++++++ .../ServiceHourMetricTransformNode.java | 44 ++++++ .../ServiceMetricAggregationWorker.java | 12 +- .../{ => metric}/ServiceMetricGraph.java | 20 ++- .../ServiceMetricRemoteWorker.java | 4 +- .../ServiceMinuteMetricPersistenceWorker.java | 67 ++++++++ .../ServiceMonthMetricPersistenceWorker.java | 67 ++++++++ .../ServiceMonthMetricTransformNode.java | 44 ++++++ .../apm/collector/storage/StorageModule.java | 4 +- .../IServiceDayMetricPersistenceDAO.java} | 4 +- .../smp/IServiceHourMetricPersistenceDAO.java | 28 ++++ .../IServiceMinuteMetricPersistenceDAO.java | 28 ++++ .../IServiceMonthMetricPersistenceDAO.java | 28 ++++ .../storage/table/service/ServiceMetric.java | 25 ++- .../storage/es/StorageModuleEsProvider.java | 6 +- .../es/dao/ServiceMetricEsPersistenceDAO.java | 145 ------------------ ...AbstractServiceMetricEsPersistenceDAO.java | 99 ++++++++++++ .../smp/ServiceDayMetricEsPersistenceDAO.java | 42 +++++ .../ServiceHourMetricEsPersistenceDAO.java | 42 +++++ .../ServiceMinuteMetricEsPersistenceDAO.java | 42 +++++ .../ServiceMonthMetricEsPersistenceDAO.java | 42 +++++ .../storage/h2/StorageModuleH2Provider.java | 6 +- ... ServiceMinuteMetricH2PersistenceDAO.java} | 8 +- 28 files changed, 765 insertions(+), 187 deletions(-) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/{ServiceMetricPersistenceWorker.java => metric/ServiceDayMetricPersistenceWorker.java} (76%) create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricTransformNode.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/service/metric/ServiceHourMetricPersistenceWorker.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/service/metric/ServiceHourMetricTransformNode.java rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/{ => metric}/ServiceMetricAggregationWorker.java (92%) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/{ => metric}/ServiceMetricGraph.java (76%) rename apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/{ => metric}/ServiceMetricRemoteWorker.java (96%) create mode 100644 apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMinuteMetricPersistenceWorker.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/service/metric/ServiceMonthMetricPersistenceWorker.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/service/metric/ServiceMonthMetricTransformNode.java rename apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/{IServiceMetricPersistenceDAO.java => smp/IServiceDayMetricPersistenceDAO.java} (82%) create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceHourMetricPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMinuteMetricPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMonthMetricPersistenceDAO.java delete mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/AbstractServiceMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceDayMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceHourMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMinuteMetricEsPersistenceDAO.java create mode 100644 apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMonthMetricEsPersistenceDAO.java rename apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/{ServiceMetricH2PersistenceDAO.java => ServiceMinuteMetricH2PersistenceDAO.java} (95%) diff --git a/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/service/ServiceMetricAlarmGraph.java b/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/service/ServiceMetricAlarmGraph.java index 1cfd716fd..18d87ad6d 100644 --- a/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/service/ServiceMetricAlarmGraph.java +++ b/apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/service/ServiceMetricAlarmGraph.java @@ -64,7 +64,7 @@ public class ServiceMetricAlarmGraph { private void link(Graph graph) { GraphManager.INSTANCE.findGraph(MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceMetric.class) - .toFinder().findNode(MetricWorkerIdDefine.SERVICE_METRIC_PERSISTENCE_WORKER_ID, ServiceMetric.class) + .toFinder().findNode(MetricWorkerIdDefine.SERVICE_MINUTE_METRIC_PERSISTENCE_WORKER_ID, ServiceMetric.class) .addNext(new NodeProcessor() { @Override public int id() { return AlarmWorkerIdDefine.SERVICE_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 fe747eef6..91233e386 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 @@ -52,9 +52,15 @@ public class MetricWorkerIdDefine { public static final int APPLICATION_REFERENCE_MONTH_METRIC_PERSISTENCE_WORKER_ID = 4307; public static final int APPLICATION_REFERENCE_MONTH_METRIC_TRANSFORM_NODE_ID = 4308; - public static final int SERVICE_METRIC_AGGREGATION_WORKER_ID = 409; - public static final int SERVICE_METRIC_REMOTE_WORKER_ID = 410; - public static final int SERVICE_METRIC_PERSISTENCE_WORKER_ID = 411; + public static final int SERVICE_MINUTE_METRIC_AGGREGATION_WORKER_ID = 4400; + public static final int SERVICE_MINUTE_METRIC_REMOTE_WORKER_ID = 4401; + public static final int SERVICE_MINUTE_METRIC_PERSISTENCE_WORKER_ID = 4402; + public static final int SERVICE_HOUR_METRIC_PERSISTENCE_WORKER_ID = 4403; + public static final int SERVICE_HOUR_METRIC_TRANSFORM_NODE_ID = 4404; + public static final int SERVICE_DAY_METRIC_PERSISTENCE_WORKER_ID = 4405; + public static final int SERVICE_DAY_METRIC_TRANSFORM_NODE_ID = 4406; + public static final int SERVICE_MONTH_METRIC_PERSISTENCE_WORKER_ID = 4407; + public static final int SERVICE_MONTH_METRIC_TRANSFORM_NODE_ID = 4408; public static final int INSTANCE_METRIC_AGGREGATION_WORKER_ID = 412; public static final int INSTANCE_METRIC_REMOTE_WORKER_ID = 413; 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 d221afd32..c262eeb61 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 @@ -36,7 +36,7 @@ import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segme import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostSpanListener; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntrySpanListener; -import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceMetricGraph; +import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.metric.ServiceMetricGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.refmetric.ServiceReferenceMetricGraph; import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.refmetric.ServiceReferenceMetricSpanListener; import org.apache.skywalking.apm.collector.analysis.segment.parser.define.AnalysisSegmentParserModule; diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricPersistenceWorker.java similarity index 76% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricPersistenceWorker.java index 4e8de19ad..fb2e112d3 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricPersistenceWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.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.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceDayMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; /** * @author peng-yongsheng */ -public class ServiceMetricPersistenceWorker extends PersistenceWorker { +public class ServiceDayMetricPersistenceWorker extends PersistenceWorker { - public ServiceMetricPersistenceWorker(ModuleManager moduleManager) { + public ServiceDayMetricPersistenceWorker(ModuleManager moduleManager) { super(moduleManager); } @Override public int id() { - return MetricWorkerIdDefine.SERVICE_METRIC_PERSISTENCE_WORKER_ID; + return MetricWorkerIdDefine.SERVICE_DAY_METRIC_PERSISTENCE_WORKER_ID; } @SuppressWarnings("unchecked") @Override protected IPersistenceDAO persistenceDAO() { - return getModuleManager().find(StorageModule.NAME).getService(IServiceMetricPersistenceDAO.class); + return getModuleManager().find(StorageModule.NAME).getService(IServiceDayMetricPersistenceDAO.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 ServiceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { - return new ServiceMetricPersistenceWorker(moduleManager); + @Override public ServiceDayMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ServiceDayMetricPersistenceWorker(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/service/metric/ServiceDayMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricTransformNode.java new file mode 100644 index 000000000..bbd095af4 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceDayMetricTransformNode.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.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.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceDayMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_DAY_METRIC_TRANSFORM_NODE_ID; + } + + @Override public void process(ServiceMetric serviceMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToDay(serviceMetric.getTimeBucket()); + serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId()); + serviceMetric.setTimeBucket(timeBucket); + + next.execute(serviceMetric); + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricPersistenceWorker.java new file mode 100644 index 000000000..712735f49 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricPersistenceWorker.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.service.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.smp.IServiceHourMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceHourMetricPersistenceWorker extends PersistenceWorker { + + public ServiceHourMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_HOUR_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IServiceHourMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ServiceHourMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ServiceHourMetricPersistenceWorker(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/service/metric/ServiceHourMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricTransformNode.java new file mode 100644 index 000000000..656e06541 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceHourMetricTransformNode.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.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.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceHourMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_HOUR_METRIC_TRANSFORM_NODE_ID; + } + + @Override public void process(ServiceMetric serviceMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToHour(serviceMetric.getTimeBucket()); + serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId()); + serviceMetric.setTimeBucket(timeBucket); + + next.execute(serviceMetric); + } +} diff --git a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricAggregationWorker.java similarity index 92% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricAggregationWorker.java index 5930b2786..e76a6c204 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricAggregationWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.metric; import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine; import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider; @@ -36,7 +36,7 @@ public class ServiceMetricAggregationWorker extends AggregationWorker graph = GraphManager.INSTANCE.createIfAbsent(MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceReferenceMetric.class); - graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener)) - .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(workerCreateListener)) - .addNext(new ServiceMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + Node remoteNode = graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(workerCreateListener)) + .addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, MetricGraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(workerCreateListener)); + + remoteNode.addNext(new ServiceMinuteMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ServiceHourMetricTransformNode()) + .addNext(new ServiceHourMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ServiceDayMetricTransformNode()) + .addNext(new ServiceDayMetricPersistenceWorker.Factory(moduleManager).create(workerCreateListener)); + + remoteNode.addNext(new ServiceMonthMetricTransformNode()) + .addNext(new ServiceMonthMetricPersistenceWorker.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/service/ServiceMetricRemoteWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricRemoteWorker.java similarity index 96% rename from apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java rename to apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricRemoteWorker.java index 60b633343..7ec52ac9a 100644 --- a/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMetricRemoteWorker.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service; +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.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 ServiceMetricRemoteWorker extends AbstractRemoteWorker { + + public ServiceMinuteMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_MINUTE_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IServiceMinuteMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ServiceMinuteMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ServiceMinuteMetricPersistenceWorker(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/service/metric/ServiceMonthMetricPersistenceWorker.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMonthMetricPersistenceWorker.java new file mode 100644 index 000000000..cce30ab9c --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMonthMetricPersistenceWorker.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.service.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.smp.IServiceMonthMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceMonthMetricPersistenceWorker extends PersistenceWorker { + + public ServiceMonthMetricPersistenceWorker(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_MONTH_METRIC_PERSISTENCE_WORKER_ID; + } + + @SuppressWarnings("unchecked") + @Override protected IPersistenceDAO persistenceDAO() { + return getModuleManager().find(StorageModule.NAME).getService(IServiceMonthMetricPersistenceDAO.class); + } + + @Override protected boolean needMergeDBData() { + return true; + } + + public static class Factory extends PersistenceWorkerProvider { + + public Factory(ModuleManager moduleManager) { + super(moduleManager); + } + + @Override public ServiceMonthMetricPersistenceWorker workerInstance(ModuleManager moduleManager) { + return new ServiceMonthMetricPersistenceWorker(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/service/metric/ServiceMonthMetricTransformNode.java b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMonthMetricTransformNode.java new file mode 100644 index 000000000..f2c252f76 --- /dev/null +++ b/apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/metric/ServiceMonthMetricTransformNode.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.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.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public class ServiceMonthMetricTransformNode implements NodeProcessor { + + @Override public int id() { + return MetricWorkerIdDefine.SERVICE_MONTH_METRIC_TRANSFORM_NODE_ID; + } + + @Override public void process(ServiceMetric serviceMetric, Next next) { + long timeBucket = TimeBucketUtils.INSTANCE.minuteToMonth(serviceMetric.getTimeBucket()); + serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId()); + serviceMetric.setTimeBucket(timeBucket); + + next.execute(serviceMetric); + } +} 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 382f40c7b..1718e87cb 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 @@ -67,7 +67,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmListPersiste import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceAlarmListPersistenceDAO; @@ -127,7 +127,7 @@ public class StorageModule extends Module { classes.add(ISegmentCostPersistenceDAO.class); classes.add(ISegmentPersistenceDAO.class); classes.add(IServiceEntryPersistenceDAO.class); - classes.add(IServiceMetricPersistenceDAO.class); + classes.add(IServiceMinuteMetricPersistenceDAO.class); classes.add(IServiceReferenceMinuteMetricPersistenceDAO.class); classes.add(IInstanceMetricPersistenceDAO.class); diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IServiceMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceDayMetricPersistenceDAO.java similarity index 82% rename from apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IServiceMetricPersistenceDAO.java rename to apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceDayMetricPersistenceDAO.java index e11e399f8..fa7b0e6d9 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IServiceMetricPersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceDayMetricPersistenceDAO.java @@ -16,7 +16,7 @@ * */ -package org.apache.skywalking.apm.collector.storage.dao; +package org.apache.skywalking.apm.collector.storage.dao.smp; import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; @@ -24,5 +24,5 @@ import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; /** * @author peng-yongsheng */ -public interface IServiceMetricPersistenceDAO extends IPersistenceDAO { +public interface IServiceDayMetricPersistenceDAO extends IPersistenceDAO { } diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceHourMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceHourMetricPersistenceDAO.java new file mode 100644 index 000000000..44e7b692f --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceHourMetricPersistenceDAO.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.smp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public interface IServiceHourMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMinuteMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMinuteMetricPersistenceDAO.java new file mode 100644 index 000000000..dd48b105e --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMinuteMetricPersistenceDAO.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.smp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public interface IServiceMinuteMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMonthMetricPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMonthMetricPersistenceDAO.java new file mode 100644 index 000000000..8c7764b1c --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMonthMetricPersistenceDAO.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.smp; + +import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; + +/** + * @author peng-yongsheng + */ +public interface IServiceMonthMetricPersistenceDAO extends IPersistenceDAO { +} diff --git a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java index bda10fdf3..f4cf9639f 100644 --- a/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java +++ b/apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java @@ -19,7 +19,7 @@ package org.apache.skywalking.apm.collector.storage.table.service; 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 ServiceMetric extends AbstractData implements Metric { +public class ServiceMetric extends StreamData implements Metric { private static final Column[] STRING_COLUMNS = { new Column(ServiceMetricTable.COLUMN_ID, new NonOperation()), + new Column(ServiceMetricTable.COLUMN_METRIC_ID, new NonOperation()), }; private static final Column[] LONG_COLUMNS = { @@ -63,8 +64,24 @@ public class ServiceMetric extends AbstractData implements Metric { private static final Column[] BYTE_COLUMNS = {}; - public ServiceMetric(String id) { - super(id, STRING_COLUMNS, LONG_COLUMNS, DOUBLE_COLUMNS, INTEGER_COLUMNS, BOOLEAN_COLUMNS, BYTE_COLUMNS); + public ServiceMetric() { + 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 c84399684..693f9808b 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 @@ -76,7 +76,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmListPersiste import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceAlarmListPersistenceDAO; @@ -130,7 +130,7 @@ import org.apache.skywalking.apm.collector.storage.es.dao.ServiceAlarmEsPersiste import org.apache.skywalking.apm.collector.storage.es.dao.ServiceAlarmListEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ServiceEntryEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ServiceEntryEsUIDAO; -import org.apache.skywalking.apm.collector.storage.es.dao.ServiceMetricEsPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.es.dao.smp.ServiceMinuteMetricEsPersistenceDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ServiceNameEsCacheDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ServiceNameEsRegisterDAO; import org.apache.skywalking.apm.collector.storage.es.dao.ServiceReferenceAlarmEsPersistenceDAO; @@ -240,7 +240,7 @@ public class StorageModuleEsProvider extends ModuleProvider { this.registerServiceImplementation(ISegmentCostPersistenceDAO.class, new SegmentCostEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IServiceEntryPersistenceDAO.class, new ServiceEntryEsPersistenceDAO(elasticSearchClient)); - this.registerServiceImplementation(IServiceMetricPersistenceDAO.class, new ServiceMetricEsPersistenceDAO(elasticSearchClient)); + this.registerServiceImplementation(IServiceMinuteMetricPersistenceDAO.class, new ServiceMinuteMetricEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IServiceReferenceMinuteMetricPersistenceDAO.class, new ServiceReferenceMinuteMetricEsPersistenceDAO(elasticSearchClient)); this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricEsPersistenceDAO(elasticSearchClient)); diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceMetricEsPersistenceDAO.java deleted file mode 100644 index 2271d7d44..000000000 --- a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceMetricEsPersistenceDAO.java +++ /dev/null @@ -1,145 +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.IServiceMetricPersistenceDAO; -import org.apache.skywalking.apm.collector.storage.es.base.dao.EsDAO; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; -import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; -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 ServiceMetricEsPersistenceDAO extends EsDAO implements IServiceMetricPersistenceDAO { - - private final Logger logger = LoggerFactory.getLogger(ServiceMetricEsPersistenceDAO.class); - - public ServiceMetricEsPersistenceDAO(ElasticSearchClient client) { - super(client); - } - - @Override public ServiceMetric get(String id) { - GetResponse getResponse = getClient().prepareGet(ServiceMetricTable.TABLE, id).get(); - if (getResponse.isExists()) { - ServiceMetric serviceMetric = new ServiceMetric(id); - Map source = getResponse.getSource(); - serviceMetric.setApplicationId(((Number)source.get(ServiceMetricTable.COLUMN_APPLICATION_ID)).intValue()); - serviceMetric.setInstanceId(((Number)source.get(ServiceMetricTable.COLUMN_INSTANCE_ID)).intValue()); - serviceMetric.setServiceId(((Number)source.get(ServiceMetricTable.COLUMN_SERVICE_ID)).intValue()); - serviceMetric.setSourceValue(((Number)source.get(ServiceMetricTable.COLUMN_SOURCE_VALUE)).intValue()); - - serviceMetric.setTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); - serviceMetric.setTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); - serviceMetric.setTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); - serviceMetric.setTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - serviceMetric.setBusinessTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); - serviceMetric.setBusinessTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); - serviceMetric.setBusinessTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); - serviceMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - serviceMetric.setMqTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); - serviceMetric.setMqTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); - serviceMetric.setMqTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); - serviceMetric.setMqTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); - - serviceMetric.setTimeBucket(((Number)source.get(ServiceMetricTable.COLUMN_TIME_BUCKET)).longValue()); - return serviceMetric; - } else { - return null; - } - } - - @Override public IndexRequestBuilder prepareBatchInsert(ServiceMetric data) { - Map source = new HashMap<>(); - source.put(ServiceMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(ServiceMetricTable.COLUMN_INSTANCE_ID, data.getInstanceId()); - source.put(ServiceMetricTable.COLUMN_SERVICE_ID, data.getServiceId()); - source.put(ServiceMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); - - source.put(ServiceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket()); - - return getClient().prepareIndex(ServiceMetricTable.TABLE, data.getId()).setSource(source); - } - - @Override public UpdateRequestBuilder prepareBatchUpdate(ServiceMetric data) { - Map source = new HashMap<>(); - source.put(ServiceMetricTable.COLUMN_APPLICATION_ID, data.getApplicationId()); - source.put(ServiceMetricTable.COLUMN_INSTANCE_ID, data.getInstanceId()); - source.put(ServiceMetricTable.COLUMN_SERVICE_ID, data.getServiceId()); - source.put(ServiceMetricTable.COLUMN_SOURCE_VALUE, data.getSourceValue()); - - source.put(ServiceMetricTable.COLUMN_TRANSACTION_CALLS, data.getTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, data.getTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, data.getTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, data.getTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, data.getBusinessTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, data.getBusinessTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, data.getBusinessTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, data.getBusinessTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, data.getMqTransactionCalls()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, data.getMqTransactionErrorCalls()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, data.getMqTransactionDurationSum()); - source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, data.getMqTransactionErrorDurationSum()); - - source.put(ServiceMetricTable.COLUMN_TIME_BUCKET, data.getTimeBucket()); - - return getClient().prepareUpdate(ServiceMetricTable.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(ServiceMetricTable.COLUMN_TIME_BUCKET).gte(startTimeBucket).lte(endTimeBucket)) - .source(ServiceMetricTable.TABLE) - .get(); - - long deleted = response.getDeleted(); - logger.info("Delete {} rows history from {} index.", deleted, ServiceMetricTable.TABLE); - } -} diff --git a/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/AbstractServiceMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/AbstractServiceMetricEsPersistenceDAO.java new file mode 100644 index 000000000..7ca567681 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/AbstractServiceMetricEsPersistenceDAO.java @@ -0,0 +1,99 @@ +/* + * 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.smp; + +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.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; + +/** + * @author peng-yongsheng + */ +public abstract class AbstractServiceMetricEsPersistenceDAO extends AbstractPersistenceEsDAO { + + AbstractServiceMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected final String timeBucketColumnNameForDelete() { + return ServiceMetricTable.COLUMN_TIME_BUCKET; + } + + @Override protected final ServiceMetric esDataToStreamData(Map source) { + ServiceMetric serviceMetric = new ServiceMetric(); + serviceMetric.setId((String)source.get(ServiceMetricTable.COLUMN_ID)); + serviceMetric.setMetricId((String)source.get(ServiceMetricTable.COLUMN_METRIC_ID)); + + serviceMetric.setApplicationId(((Number)source.get(ServiceMetricTable.COLUMN_APPLICATION_ID)).intValue()); + serviceMetric.setInstanceId(((Number)source.get(ServiceMetricTable.COLUMN_INSTANCE_ID)).intValue()); + serviceMetric.setServiceId(((Number)source.get(ServiceMetricTable.COLUMN_SERVICE_ID)).intValue()); + serviceMetric.setSourceValue(((Number)source.get(ServiceMetricTable.COLUMN_SOURCE_VALUE)).intValue()); + + serviceMetric.setTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_CALLS)).longValue()); + serviceMetric.setTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS)).longValue()); + serviceMetric.setTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_DURATION_SUM)).longValue()); + serviceMetric.setTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + serviceMetric.setBusinessTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS)).longValue()); + serviceMetric.setBusinessTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS)).longValue()); + serviceMetric.setBusinessTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM)).longValue()); + serviceMetric.setBusinessTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + serviceMetric.setMqTransactionCalls(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_CALLS)).longValue()); + serviceMetric.setMqTransactionErrorCalls(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS)).longValue()); + serviceMetric.setMqTransactionDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM)).longValue()); + serviceMetric.setMqTransactionErrorDurationSum(((Number)source.get(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM)).longValue()); + + serviceMetric.setTimeBucket(((Number)source.get(ServiceMetricTable.COLUMN_TIME_BUCKET)).longValue()); + return serviceMetric; + } + + @Override protected final Map esStreamDataToEsData(ServiceMetric streamData) { + Map source = new HashMap<>(); + source.put(ServiceMetricTable.COLUMN_ID, streamData.getId()); + source.put(ServiceMetricTable.COLUMN_METRIC_ID, streamData.getMetricId()); + + source.put(ServiceMetricTable.COLUMN_APPLICATION_ID, streamData.getApplicationId()); + source.put(ServiceMetricTable.COLUMN_INSTANCE_ID, streamData.getInstanceId()); + source.put(ServiceMetricTable.COLUMN_SERVICE_ID, streamData.getServiceId()); + source.put(ServiceMetricTable.COLUMN_SOURCE_VALUE, streamData.getSourceValue()); + + source.put(ServiceMetricTable.COLUMN_TRANSACTION_CALLS, streamData.getTransactionCalls()); + source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_CALLS, streamData.getTransactionErrorCalls()); + source.put(ServiceMetricTable.COLUMN_TRANSACTION_DURATION_SUM, streamData.getTransactionDurationSum()); + source.put(ServiceMetricTable.COLUMN_TRANSACTION_ERROR_DURATION_SUM, streamData.getTransactionErrorDurationSum()); + + source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_CALLS, streamData.getBusinessTransactionCalls()); + source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_CALLS, streamData.getBusinessTransactionErrorCalls()); + source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_DURATION_SUM, streamData.getBusinessTransactionDurationSum()); + source.put(ServiceMetricTable.COLUMN_BUSINESS_TRANSACTION_ERROR_DURATION_SUM, streamData.getBusinessTransactionErrorDurationSum()); + + source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_CALLS, streamData.getMqTransactionCalls()); + source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_CALLS, streamData.getMqTransactionErrorCalls()); + source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_DURATION_SUM, streamData.getMqTransactionDurationSum()); + source.put(ServiceMetricTable.COLUMN_MQ_TRANSACTION_ERROR_DURATION_SUM, streamData.getMqTransactionErrorDurationSum()); + + source.put(ServiceMetricTable.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/smp/ServiceDayMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceDayMetricEsPersistenceDAO.java new file mode 100644 index 000000000..dcefad0c2 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceDayMetricEsPersistenceDAO.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.smp; + +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.smp.IServiceMinuteMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ServiceDayMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO { + + public ServiceDayMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ServiceMetricTable.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/smp/ServiceHourMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceHourMetricEsPersistenceDAO.java new file mode 100644 index 000000000..6db81d609 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceHourMetricEsPersistenceDAO.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.smp; + +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.smp.IServiceMinuteMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ServiceHourMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO { + + public ServiceHourMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ServiceMetricTable.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/smp/ServiceMinuteMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMinuteMetricEsPersistenceDAO.java new file mode 100644 index 000000000..13b3079d0 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMinuteMetricEsPersistenceDAO.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.smp; + +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.smp.IServiceMinuteMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ServiceMinuteMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO { + + public ServiceMinuteMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ServiceMetricTable.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/smp/ServiceMonthMetricEsPersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMonthMetricEsPersistenceDAO.java new file mode 100644 index 000000000..e7c597d12 --- /dev/null +++ b/apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMonthMetricEsPersistenceDAO.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.smp; + +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.smp.IServiceMinuteMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; +import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; +import org.elasticsearch.action.index.IndexRequestBuilder; +import org.elasticsearch.action.update.UpdateRequestBuilder; + +/** + * @author peng-yongsheng + */ +public class ServiceMonthMetricEsPersistenceDAO extends AbstractServiceMetricEsPersistenceDAO implements IServiceMinuteMetricPersistenceDAO { + + public ServiceMonthMetricEsPersistenceDAO(ElasticSearchClient client) { + super(client); + } + + @Override protected String tableName() { + return ServiceMetricTable.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 4529ad5c4..d05590c13 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 @@ -72,7 +72,7 @@ import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmListPersiste import org.apache.skywalking.apm.collector.storage.dao.IServiceAlarmPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryUIDAO; -import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameCacheDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceNameRegisterDAO; import org.apache.skywalking.apm.collector.storage.dao.IServiceReferenceAlarmListPersistenceDAO; @@ -126,7 +126,7 @@ import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceAlarmH2Persiste import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceAlarmListH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceEntryH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceEntryH2UIDAO; -import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceMetricH2PersistenceDAO; +import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceMinuteMetricH2PersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceNameH2CacheDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceNameH2RegisterDAO; import org.apache.skywalking.apm.collector.storage.h2.dao.ServiceReferenceAlarmH2PersistenceDAO; @@ -218,7 +218,7 @@ public class StorageModuleH2Provider extends ModuleProvider { this.registerServiceImplementation(ISegmentCostPersistenceDAO.class, new SegmentCostH2PersistenceDAO(h2Client)); this.registerServiceImplementation(ISegmentPersistenceDAO.class, new SegmentH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IServiceEntryPersistenceDAO.class, new ServiceEntryH2PersistenceDAO(h2Client)); - this.registerServiceImplementation(IServiceMetricPersistenceDAO.class, new ServiceMetricH2PersistenceDAO(h2Client)); + this.registerServiceImplementation(IServiceMinuteMetricPersistenceDAO.class, new ServiceMinuteMetricH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IServiceReferenceMinuteMetricPersistenceDAO.class, new ServiceReferenceMetricH2PersistenceDAO(h2Client)); this.registerServiceImplementation(IInstanceMetricPersistenceDAO.class, new InstanceMetricH2PersistenceDAO(h2Client)); diff --git a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMetricH2PersistenceDAO.java b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMinuteMetricH2PersistenceDAO.java similarity index 95% rename from apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMetricH2PersistenceDAO.java rename to apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMinuteMetricH2PersistenceDAO.java index aae73ed23..ed87fcfb1 100644 --- a/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMetricH2PersistenceDAO.java +++ b/apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMinuteMetricH2PersistenceDAO.java @@ -30,7 +30,7 @@ import org.apache.skywalking.apm.collector.storage.h2.base.define.H2SqlEntity; import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetricTable; import org.apache.skywalking.apm.collector.client.h2.H2Client; import org.apache.skywalking.apm.collector.client.h2.H2ClientException; -import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO; +import org.apache.skywalking.apm.collector.storage.dao.smp.IServiceMinuteMetricPersistenceDAO; import org.apache.skywalking.apm.collector.storage.h2.base.dao.H2DAO; import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric; import org.slf4j.Logger; @@ -39,12 +39,12 @@ import org.slf4j.LoggerFactory; /** * @author peng-yongsheng, clevertension */ -public class ServiceMetricH2PersistenceDAO extends H2DAO implements IServiceMetricPersistenceDAO { +public class ServiceMinuteMetricH2PersistenceDAO extends H2DAO implements IServiceMinuteMetricPersistenceDAO { - private final Logger logger = LoggerFactory.getLogger(ServiceMetricH2PersistenceDAO.class); + private final Logger logger = LoggerFactory.getLogger(ServiceMinuteMetricH2PersistenceDAO.class); private static final String GET_SQL = "select * from {0} where {1} = ?"; - public ServiceMetricH2PersistenceDAO(H2Client client) { + public ServiceMinuteMetricH2PersistenceDAO(H2Client client) { super(client); }