kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Service metric pyramid aggregate.

This commit is contained in:
peng-yongsheng 2018-01-07 20:00:30 +08:00
parent da89ea4d77
commit f3fc91926e
28 changed files with 765 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
apm-collector/apm-collector-analysis/analysis-alarm/alarm-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/alarm/provider/worker/service/ServiceMetricAlarmGraph.java
View File

@ -64,7 +64,7 @@ public class ServiceMetricAlarmGraph {
private void link(Graph<ServiceMetric> 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<ServiceMetric, ServiceMetric>() {
@Override public int id() {
return AlarmWorkerIdDefine.SERVICE_METRIC_ALARM_GRAPH_BRIDGE_WORKER_ID;

12
apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/MetricWorkerIdDefine.java
View File

@ -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;

2
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/AnalysisMetricModuleProvider.java
View File

@ -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;

18
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricPersistenceWorker.java → 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
View File

@ -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<ServiceMetric> {
public class ServiceDayMetricPersistenceWorker extends PersistenceWorker<ServiceMetric> {
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<?, ?, ServiceMetric> 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<ServiceMetric, ServiceMetricPersistenceWorker> {
public static class Factory extends PersistenceWorkerProvider<ServiceMetric, ServiceDayMetricPersistenceWorker> {
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

44
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 Normal file
View File

@ -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<ServiceMetric, ServiceMetric> {
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_DAY_METRIC_TRANSFORM_NODE_ID;
}
@Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
long timeBucket = TimeBucketUtils.INSTANCE.minuteToDay(serviceMetric.getTimeBucket());
serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
serviceMetric.setTimeBucket(timeBucket);
next.execute(serviceMetric);
}
}

67
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 Normal file
View File

@ -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<ServiceMetric> {
public ServiceHourMetricPersistenceWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_HOUR_METRIC_PERSISTENCE_WORKER_ID;
}
@SuppressWarnings("unchecked")
@Override protected IPersistenceDAO<?, ?, ServiceMetric> persistenceDAO() {
return getModuleManager().find(StorageModule.NAME).getService(IServiceHourMetricPersistenceDAO.class);
}
@Override protected boolean needMergeDBData() {
return true;
}
public static class Factory extends PersistenceWorkerProvider<ServiceMetric, ServiceHourMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceHourMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
return new ServiceHourMetricPersistenceWorker(moduleManager);
}
@Override
public int queueSize() {
return 1024;
}
}
}

44
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 Normal file
View File

@ -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<ServiceMetric, ServiceMetric> {
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_HOUR_METRIC_TRANSFORM_NODE_ID;
}
@Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
long timeBucket = TimeBucketUtils.INSTANCE.minuteToHour(serviceMetric.getTimeBucket());
serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
serviceMetric.setTimeBucket(timeBucket);
next.execute(serviceMetric);
}
}

12
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricAggregationWorker.java → 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
View File

@ -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<ServiceRef
}
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_METRIC_AGGREGATION_WORKER_ID;
return MetricWorkerIdDefine.SERVICE_MINUTE_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected ServiceMetric transform(ServiceReferenceMetric serviceReferenceMetric) {
@ -44,9 +44,13 @@ public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceRef
Long timeBucket = serviceReferenceMetric.getTimeBucket();
Integer sourceValue = serviceReferenceMetric.getSourceValue();
String id = String.valueOf(timeBucket) + Const.ID_SPLIT + String.valueOf(serviceId) + Const.ID_SPLIT + String.valueOf(sourceValue);
String metricId = String.valueOf(serviceId) + Const.ID_SPLIT + String.valueOf(sourceValue);
String id = String.valueOf(timeBucket) + Const.ID_SPLIT + metricId;
ServiceMetric serviceMetric = new ServiceMetric();
serviceMetric.setId(id);
serviceMetric.setMetricId(metricId);
ServiceMetric serviceMetric = new ServiceMetric(id);
serviceMetric.setApplicationId(serviceReferenceMetric.getBehindApplicationId());
serviceMetric.setInstanceId(serviceReferenceMetric.getBehindInstanceId());
serviceMetric.setServiceId(serviceId);

20
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java → 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
View File

@ -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.MetricGraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.MetricWorkerIdDefine;
@ -24,10 +24,12 @@ import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCrea
import org.apache.skywalking.apm.collector.core.graph.Graph;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
import org.apache.skywalking.apm.collector.core.graph.Next;
import org.apache.skywalking.apm.collector.core.graph.Node;
import org.apache.skywalking.apm.collector.core.graph.NodeProcessor;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.RemoteModule;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
/**
@ -48,9 +50,19 @@ public class ServiceMetricGraph {
Graph<ServiceReferenceMetric> 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<ServiceMetric, ServiceMetric> 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);
}

4
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricRemoteWorker.java → 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
View File

@ -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<ServiceMetri
}
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_METRIC_REMOTE_WORKER_ID;
return MetricWorkerIdDefine.SERVICE_MINUTE_METRIC_REMOTE_WORKER_ID;
}
@Override protected void onWork(ServiceMetric serviceMetric) throws WorkerException {

67
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 Normal file
View File

@ -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.IServiceMinuteMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
/**
* @author peng-yongsheng
*/
public class ServiceMinuteMetricPersistenceWorker extends PersistenceWorker<ServiceMetric> {
public ServiceMinuteMetricPersistenceWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_MINUTE_METRIC_PERSISTENCE_WORKER_ID;
}
@SuppressWarnings("unchecked")
@Override protected IPersistenceDAO<?, ?, ServiceMetric> persistenceDAO() {
return getModuleManager().find(StorageModule.NAME).getService(IServiceMinuteMetricPersistenceDAO.class);
}
@Override protected boolean needMergeDBData() {
return true;
}
public static class Factory extends PersistenceWorkerProvider<ServiceMetric, ServiceMinuteMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceMinuteMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
return new ServiceMinuteMetricPersistenceWorker(moduleManager);
}
@Override
public int queueSize() {
return 1024;
}
}
}

67
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 Normal file
View File

@ -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<ServiceMetric> {
public ServiceMonthMetricPersistenceWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_MONTH_METRIC_PERSISTENCE_WORKER_ID;
}
@SuppressWarnings("unchecked")
@Override protected IPersistenceDAO<?, ?, ServiceMetric> persistenceDAO() {
return getModuleManager().find(StorageModule.NAME).getService(IServiceMonthMetricPersistenceDAO.class);
}
@Override protected boolean needMergeDBData() {
return true;
}
public static class Factory extends PersistenceWorkerProvider<ServiceMetric, ServiceMonthMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceMonthMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {
return new ServiceMonthMetricPersistenceWorker(moduleManager);
}
@Override
public int queueSize() {
return 1024;
}
}
}

44
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 Normal file
View File

@ -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<ServiceMetric, ServiceMetric> {
@Override public int id() {
return MetricWorkerIdDefine.SERVICE_MONTH_METRIC_TRANSFORM_NODE_ID;
}
@Override public void process(ServiceMetric serviceMetric, Next<ServiceMetric> next) {
long timeBucket = TimeBucketUtils.INSTANCE.minuteToMonth(serviceMetric.getTimeBucket());
serviceMetric.setId(String.valueOf(timeBucket) + Const.ID_SPLIT + serviceMetric.getMetricId());
serviceMetric.setTimeBucket(timeBucket);
next.execute(serviceMetric);
}
}

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

@ -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);

4
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/IServiceMetricPersistenceDAO.java → apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceDayMetricPersistenceDAO.java
View File

@ -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<Insert, Update, DataImpl extends ServiceMetric> extends IPersistenceDAO<Insert, Update, DataImpl> {
public interface IServiceDayMetricPersistenceDAO<INSERT, UPDATE, STREAM_DATA extends ServiceMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
}

28
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceHourMetricPersistenceDAO.java Normal file
View File

@ -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<INSERT, UPDATE, STREAM_DATA extends ServiceMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
}

28
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMinuteMetricPersistenceDAO.java Normal file
View File

@ -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<INSERT, UPDATE, STREAM_DATA extends ServiceMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
}

28
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/dao/smp/IServiceMonthMetricPersistenceDAO.java Normal file
View File

@ -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<INSERT, UPDATE, STREAM_DATA extends ServiceMetric> extends IPersistenceDAO<INSERT, UPDATE, STREAM_DATA> {
}

25
apm-collector/apm-collector-storage/collector-storage-define/src/main/java/org/apache/skywalking/apm/collector/storage/table/service/ServiceMetric.java
View File

@ -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

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

@ -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));

145
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/ServiceMetricEsPersistenceDAO.java
View File

@ -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<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
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<String, Object> 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<String, Object> 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<String, Object> 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);
}
}

99
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/AbstractServiceMetricEsPersistenceDAO.java Normal file
View File

@ -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<ServiceMetric> {
AbstractServiceMetricEsPersistenceDAO(ElasticSearchClient client) {
super(client);
}
@Override protected final String timeBucketColumnNameForDelete() {
return ServiceMetricTable.COLUMN_TIME_BUCKET;
}
@Override protected final ServiceMetric esDataToStreamData(Map<String, Object> 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<String, Object> esStreamDataToEsData(ServiceMetric streamData) {
Map<String, Object> 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;
}
}

42
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceDayMetricEsPersistenceDAO.java Normal file
View File

@ -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<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
public ServiceDayMetricEsPersistenceDAO(ElasticSearchClient client) {
super(client);
}
@Override protected String tableName() {
return ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Day.getName();
}
}

42
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceHourMetricEsPersistenceDAO.java Normal file
View File

@ -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<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
public ServiceHourMetricEsPersistenceDAO(ElasticSearchClient client) {
super(client);
}
@Override protected String tableName() {
return ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Hour.getName();
}
}

42
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMinuteMetricEsPersistenceDAO.java Normal file
View File

@ -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<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
public ServiceMinuteMetricEsPersistenceDAO(ElasticSearchClient client) {
super(client);
}
@Override protected String tableName() {
return ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Minute.getName();
}
}

42
apm-collector/apm-collector-storage/collector-storage-es-provider/src/main/java/org/apache/skywalking/apm/collector/storage/es/dao/smp/ServiceMonthMetricEsPersistenceDAO.java Normal file
View File

@ -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<IndexRequestBuilder, UpdateRequestBuilder, ServiceMetric> {
public ServiceMonthMetricEsPersistenceDAO(ElasticSearchClient client) {
super(client);
}
@Override protected String tableName() {
return ServiceMetricTable.TABLE + Const.ID_SPLIT + TimePyramid.Month.getName();
}
}

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

@ -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));

8
apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMetricH2PersistenceDAO.java → apm-collector/apm-collector-storage/collector-storage-h2-provider/src/main/java/org/apache/skywalking/apm/collector/storage/h2/dao/ServiceMinuteMetricH2PersistenceDAO.java
View File

@ -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<H2SqlEntity, H2SqlEntity, ServiceMetric> {
public class ServiceMinuteMetricH2PersistenceDAO extends H2DAO implements IServiceMinuteMetricPersistenceDAO<H2SqlEntity, H2SqlEntity, ServiceMetric> {
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);
}