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

Analysis metric module finished.

This commit is contained in:
peng-yongsheng 2017-12-19 18:07:46 +08:00
parent c037eef240
commit f891b1ca49
55 changed files with 1259 additions and 533 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/graph/TraceStreamGraph.java
View File

@ -20,11 +20,11 @@
package org.apache.skywalking.apm.collector.agent.stream.graph;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SegmentStandardizationWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentRemoteWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingAggregationWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingAggregationWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryRemoteWorker;
import org.apache.skywalking.apm.collector.core.graph.Graph;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
import org.apache.skywalking.apm.collector.queue.QueueModule;
@ -33,17 +33,17 @@ import org.apache.skywalking.apm.collector.storage.table.application.Application
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerCreateListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SegmentStandardization;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationComponentAggregationWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.application.ApplicationMappingRemoteWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.global.GlobalTracePersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance.InstanceMetricPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment.SegmentCostPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryAggregationWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceEntryPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricAggregationWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricPersistenceWorker;
import org.apache.skywalking.apm.collector.agent.stream.worker.trace.service.ServiceReferenceMetricRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentAggregationWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTracePersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceMetricPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryAggregationWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceEntryPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricAggregationWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricPersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricRemoteWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.remote.RemoteModule;

138
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricSpanListener.java
View File

@ -1,138 +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.agent.stream.worker.trace.application;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService;
import org.apache.skywalking.apm.collector.configuration.ConfigurationModule;
import org.apache.skywalking.apm.collector.configuration.service.IApdexThresholdService;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.ApdexThresholdUtils;
import org.apache.skywalking.apm.collector.core.util.CollectionUtils;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
*/
public class ApplicationReferenceMetricSpanListener implements EntrySpanListener, ExitSpanListener {
private final Logger logger = LoggerFactory.getLogger(ApplicationReferenceMetricSpanListener.class);
private final IApdexThresholdService apdexThresholdService;
private final InstanceCacheService instanceCacheService;
private final List<ApplicationReferenceMetric> applicationReferenceMetrics;
private final List<ApplicationReferenceMetric> references;
public ApplicationReferenceMetricSpanListener(ModuleManager moduleManager) {
this.applicationReferenceMetrics = new LinkedList<>();
this.references = new LinkedList<>();
this.instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class);
this.apdexThresholdService = moduleManager.find(ConfigurationModule.NAME).getService(IApdexThresholdService.class);
}
@Override
public void parseExit(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
ApplicationReferenceMetric applicationReferenceMetric = new ApplicationReferenceMetric(Const.EMPTY_STRING);
applicationReferenceMetric.setFrontApplicationId(applicationId);
applicationReferenceMetric.setBehindApplicationId(spanDecorator.getPeerId());
applicationReferenceMetric.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime()));
String idBuilder = String.valueOf(applicationReferenceMetric.getTimeBucket()) + Const.ID_SPLIT + applicationId +
Const.ID_SPLIT + spanDecorator.getPeerId();
applicationReferenceMetric.setId(idBuilder);
applicationReferenceMetrics.add(buildApplicationRefSum(applicationReferenceMetric, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError()));
}
@Override
public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId,
String segmentId) {
if (CollectionUtils.isNotEmpty(references)) {
references.forEach(applicationReference -> {
applicationReference.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime()));
String idBuilder = String.valueOf(applicationReference.getTimeBucket()) + Const.ID_SPLIT + applicationReference.getFrontApplicationId() +
Const.ID_SPLIT + applicationReference.getBehindApplicationId();
applicationReference.setId(idBuilder);
applicationReferenceMetrics.add(buildApplicationRefSum(applicationReference, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError()));
});
} else {
ApplicationReferenceMetric applicationReferenceMetric = new ApplicationReferenceMetric(Const.EMPTY_STRING);
applicationReferenceMetric.setFrontApplicationId(Const.USER_ID);
applicationReferenceMetric.setBehindApplicationId(applicationId);
applicationReferenceMetric.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanDecorator.getStartTime()));
String idBuilder = String.valueOf(applicationReferenceMetric.getTimeBucket()) + Const.ID_SPLIT + applicationReferenceMetric.getFrontApplicationId() +
Const.ID_SPLIT + applicationReferenceMetric.getBehindApplicationId();
applicationReferenceMetric.setId(idBuilder);
applicationReferenceMetrics.add(buildApplicationRefSum(applicationReferenceMetric, spanDecorator.getStartTime(), spanDecorator.getEndTime(), spanDecorator.getIsError()));
}
}
// @Override public void parseRef(ReferenceDecorator referenceDecorator, int applicationId, int instanceId,
// String segmentId) {
// int parentApplicationId = instanceCacheService.get(referenceDecorator.getParentApplicationInstanceId());
//
// ApplicationReferenceMetric referenceSum = new ApplicationReferenceMetric(Const.EMPTY_STRING);
// referenceSum.setFrontApplicationId(parentApplicationId);
// referenceSum.setBehindApplicationId(applicationId);
// references.add(referenceSum);
// }
@Override public void build() {
logger.debug("node reference summary listener build");
// Graph<ApplicationReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_REFERENCE_METRIC_GRAPH_ID, ApplicationReferenceMetric.class);
// for (ApplicationReferenceMetric applicationReferenceMetric : applicationReferenceMetrics) {
// graph.start(applicationReferenceMetric);
// }
}
private ApplicationReferenceMetric buildApplicationRefSum(ApplicationReferenceMetric reference,
long startTime, long endTime, boolean isError) {
long duration = endTime - startTime;
reference.setTransactionCalls(1L);
reference.setTransactionDurationSum(duration);
if (isError) {
reference.setTransactionErrorCalls(1L);
reference.setTransactionErrorDurationSum(duration);
}
ApdexThresholdUtils.Apdex apdex = ApdexThresholdUtils.compute(apdexThresholdService.getApplicationApdexThreshold(reference.getBehindApplicationId()), duration);
if (ApdexThresholdUtils.Apdex.Satisfied.equals(apdex)) {
reference.setSatisfiedCount(1L);
} else if (ApdexThresholdUtils.Apdex.Tolerating.equals(apdex)) {
reference.setToleratingCount(1L);
} else {
reference.setFrustratedCount(1L);
}
return reference;
}
}

78
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricSpanListener.java
View File

@ -1,78 +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.agent.stream.worker.trace.instance;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
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.util.Const;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* @author peng-yongsheng
*/
public class InstanceMetricSpanListener implements EntrySpanListener, FirstSpanListener {
private final Logger logger = LoggerFactory.getLogger(InstanceMetricSpanListener.class);
private int applicationId;
private int instanceId;
private boolean isError;
private long duration;
private long timeBucket;
@Override
public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId,
String segmentId) {
}
@Override
public void parseFirst(SpanDecorator spanDecorator, int applicationId, int instanceId,
String segmentId) {
this.applicationId = applicationId;
this.instanceId = instanceId;
this.isError = spanDecorator.getIsError();
this.duration = spanDecorator.getEndTime() - spanDecorator.getStartTime();
timeBucket = TimeBucketUtils.INSTANCE.getSecondTimeBucket(spanDecorator.getStartTime());
}
@Override public void build() {
InstanceMetric instanceMetric = new InstanceMetric(timeBucket + Const.ID_SPLIT + instanceId);
instanceMetric.setApplicationId(applicationId);
instanceMetric.setInstanceId(instanceId);
instanceMetric.setTransactionCalls(1L);
instanceMetric.setTransactionDurationSum(duration);
if (isError) {
instanceMetric.setTransactionErrorCalls(1L);
instanceMetric.setTransactionErrorDurationSum(duration);
}
instanceMetric.setTimeBucket(timeBucket);
Graph<InstanceMetric> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.INSTANCE_METRIC_GRAPH_ID, InstanceMetric.class);
graph.start(instanceMetric);
}
}

38
apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/GraphIdDefine.java Normal file
View File

@ -0,0 +1,38 @@
/*
* 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.define.graph;
/**
* @author peng-yongsheng
*/
public class GraphIdDefine {
public static final int SERVICE_REFERENCE_METRIC_GRAPH_ID = 400;
public static final int INSTANCE_REFERENCE_METRIC_GRAPH_ID = 401;
public static final int APPLICATION_REFERENCE_METRIC_GRAPH_ID = 402;
public static final int SERVICE_METRIC_GRAPH_ID = 403;
public static final int INSTANCE_METRIC_GRAPH_ID = 404;
public static final int APPLICATION_METRIC_GRAPH_ID = 405;
public static final int APPLICATION_COMPONENT_GRAPH_ID = 406;
public static final int APPLICATION_MAPPING_GRAPH_ID = 407;
public static final int SERVICE_ENTRY_GRAPH_ID = 408;
public static final int GLOBAL_TRACE_GRAPH_ID = 409;
public static final int SEGMENT_COST_GRAPH_ID = 410;
}

68
apm-collector/apm-collector-analysis/analysis-metric/metric-define/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/define/graph/WorkerIdDefine.java Normal file
View File

@ -0,0 +1,68 @@
/*
* 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.define.graph;
/**
* @author peng-yongsheng
*/
public class WorkerIdDefine {
public static final int SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 400;
public static final int SERVICE_REFERENCE_METRIC_REMOTE_WORKER_ID = 401;
public static final int SERVICE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 402;
public static final int INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 403;
public static final int INSTANCE_REFERENCE_METRIC_REMOTE_WORKER_ID = 404;
public static final int INSTANCE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 405;
public static final int APPLICATION_REFERENCE_METRIC_AGGREGATION_WORKER_ID = 406;
public static final int APPLICATION_REFERENCE_METRIC_REMOTE_WORKER_ID = 407;
public static final int APPLICATION_REFERENCE_METRIC_PERSISTENCE_WORKER_ID = 408;
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 INSTANCE_METRIC_AGGREGATION_WORKER_ID = 412;
public static final int INSTANCE_METRIC_REMOTE_WORKER_ID = 413;
public static final int INSTANCE_METRIC_PERSISTENCE_WORKER_ID = 414;
public static final int APPLICATION_METRIC_AGGREGATION_WORKER_ID = 415;
public static final int APPLICATION_METRIC_REMOTE_WORKER_ID = 416;
public static final int APPLICATION_METRIC_PERSISTENCE_WORKER_ID = 417;
public static final int APPLICATION_COMPONENT_AGGREGATION_WORKER_ID = 418;
public static final int APPLICATION_COMPONENT_REMOTE_WORKER_ID = 419;
public static final int APPLICATION_COMPONENT_PERSISTENCE_WORKER_ID = 420;
public static final int APPLICATION_MAPPING_AGGREGATION_WORKER_ID = 421;
public static final int APPLICATION_MAPPING_REMOTE_WORKER_ID = 422;
public static final int APPLICATION_MAPPING_PERSISTENCE_WORKER_ID = 423;
public static final int SERVICE_ENTRY_AGGREGATION_WORKER_ID = 424;
public static final int SERVICE_ENTRY_REMOTE_WORKER_ID = 425;
public static final int SERVICE_ENTRY_PERSISTENCE_WORKER_ID = 426;
public static final int GLOBAL_TRACE_PERSISTENCE_WORKER_ID = 427;
public static final int SEGMENT_COST_PERSISTENCE_WORKER_ID = 428;
public static final int INSTANCE_REFERENCE_GRAPH_BRIDGE_WORKER_ID = 429;
public static final int APPLICATION_REFERENCE_GRAPH_BRIDGE_WORKER_ID = 430;
public static final int SERVICE_METRIC_GRAPH_BRIDGE_WORKER_ID = 431;
public static final int INSTANCE_METRIC_GRAPH_BRIDGE_WORKER_ID = 432;
}

20
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/pom.xml
View File

@ -36,6 +36,26 @@
<artifactId>metric-define</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>segment-parser-define</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>collector-configuration-define</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>collector-cache-define</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>collector-storage-define</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.skywalking</groupId>
<artifactId>analysis-worker-model</artifactId>

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

@ -20,6 +20,25 @@ package org.apache.skywalking.apm.collector.analysis.metric.provider;
import java.util.Properties;
import org.apache.skywalking.apm.collector.analysis.metric.define.AnalysisMetricModule;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationComponentSpanListener;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMappingSpanListener;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationMetricGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application.ApplicationReferenceMetricGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global.GlobalTraceSpanListener;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceMetricGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance.InstanceReferenceMetricGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment.SegmentCostGraph;
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.ServiceReferenceMetricGraph;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service.ServiceReferenceMetricSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.AnalysisSegmentParserModule;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParserListenerRegister;
import org.apache.skywalking.apm.collector.core.module.Module;
import org.apache.skywalking.apm.collector.core.module.ModuleProvider;
import org.apache.skywalking.apm.collector.core.module.ServiceNotProvidedException;
@ -43,7 +62,40 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
}
@Override public void start(Properties config) throws ServiceNotProvidedException {
segmentParserListenerRegister();
ServiceReferenceMetricGraph serviceReferenceMetricGraph = new ServiceReferenceMetricGraph(getManager());
serviceReferenceMetricGraph.create();
InstanceReferenceMetricGraph instanceReferenceMetricGraph = new InstanceReferenceMetricGraph(getManager());
instanceReferenceMetricGraph.create();
ApplicationReferenceMetricGraph applicationReferenceMetricGraph = new ApplicationReferenceMetricGraph(getManager());
applicationReferenceMetricGraph.create();
ServiceMetricGraph serviceMetricGraph = new ServiceMetricGraph(getManager());
serviceMetricGraph.create();
InstanceMetricGraph instanceMetricGraph = new InstanceMetricGraph(getManager());
instanceMetricGraph.create();
ApplicationMetricGraph applicationMetricGraph = new ApplicationMetricGraph(getManager());
applicationMetricGraph.create();
ApplicationComponentGraph applicationComponentGraph = new ApplicationComponentGraph(getManager());
applicationComponentGraph.create();
ApplicationMappingGraph applicationMappingGraph = new ApplicationMappingGraph(getManager());
applicationMappingGraph.create();
ServiceEntryGraph serviceEntryGraph = new ServiceEntryGraph(getManager());
serviceEntryGraph.create();
GlobalTraceGraph globalTraceGraph = new GlobalTraceGraph(getManager());
globalTraceGraph.create();
SegmentCostGraph segmentCostGraph = new SegmentCostGraph(getManager());
segmentCostGraph.create();
}
@Override public void notifyAfterCompleted() throws ServiceNotProvidedException {
@ -53,4 +105,14 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
@Override public String[] requiredModules() {
return new String[0];
}
private void segmentParserListenerRegister() {
ISegmentParserListenerRegister segmentParserListenerRegister = getManager().find(AnalysisSegmentParserModule.NAME).getService(ISegmentParserListenerRegister.class);
segmentParserListenerRegister.register(new ServiceReferenceMetricSpanListener());
segmentParserListenerRegister.register(new ApplicationComponentSpanListener());
segmentParserListenerRegister.register(new ApplicationMappingSpanListener());
segmentParserListenerRegister.register(new ServiceEntrySpanListener(getManager()));
segmentParserListenerRegister.register(new GlobalTraceSpanListener());
segmentParserListenerRegister.register(new SegmentCostSpanListener(getManager()));
}
}

2
apm-collector/apm-collector-agent-stream/collector-agent-stream-define/src/main/java/org/apache/skywalking/apm/collector/agent/stream/service/trace/MetricSource.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/MetricSource.java
View File

@ -17,7 +17,7 @@
*/
package org.apache.skywalking.apm.collector.agent.stream.service.trace;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker;
/**
* @author peng-yongsheng

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentAggregationWorker.java
View File

@ -16,14 +16,13 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
/**
* @author peng-yongsheng
@ -35,13 +34,13 @@ public class ApplicationComponentAggregationWorker extends AggregationWorker<App
}
@Override public int id() {
return 104;
return WorkerIdDefine.APPLICATION_COMPONENT_AGGREGATION_WORKER_ID;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationComponent, ApplicationComponent, ApplicationComponentAggregationWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ApplicationComponent> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationComponentAggregationWorker workerInstance(ModuleManager moduleManager) {

47
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentGraph.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
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.application.ApplicationComponent;
/**
* @author peng-yongsheng
*/
public class ApplicationComponentGraph {
private final ModuleManager moduleManager;
public ApplicationComponentGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class)
.addNode(new ApplicationComponentAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ApplicationComponentRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID).create(null))
.addNext(new ApplicationComponentPersistenceWorker.Factory(moduleManager).create(null));
}
}

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
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.IApplicationComponentPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ApplicationComponentPersistenceWorker extends PersistenceWorker<App
}
@Override public int id() {
return ApplicationComponentPersistenceWorker.class.hashCode();
return WorkerIdDefine.APPLICATION_COMPONENT_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -51,8 +50,8 @@ public class ApplicationComponentPersistenceWorker extends PersistenceWorker<App
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationComponent, ApplicationComponent, ApplicationComponentPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ApplicationComponent> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationComponentPersistenceWorker workerInstance(ModuleManager moduleManager) {

12
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentRemoteWorker.java
View File

@ -16,16 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
/**
* @author peng-yongsheng
@ -37,7 +37,7 @@ public class ApplicationComponentRemoteWorker extends AbstractRemoteWorker<Appli
}
@Override public int id() {
return 10005;
return WorkerIdDefine.APPLICATION_COMPONENT_REMOTE_WORKER_ID;
}
@Override protected void onWork(ApplicationComponent applicationComponent) throws WorkerException {

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationComponentSpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationComponentSpanListener.java
View File

@ -16,19 +16,18 @@
*
*/
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import java.util.ArrayList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.ExitSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
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.util.Const;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationComponent;
import org.slf4j.Logger;
@ -76,7 +75,7 @@ public class ApplicationComponentSpanListener implements EntrySpanListener, Exit
}
@Override public void build() {
Graph<ApplicationComponent> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class);
Graph<ApplicationComponent> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.APPLICATION_COMPONENT_GRAPH_ID, ApplicationComponent.class);
applicationComponents.forEach(applicationComponent -> {
applicationComponent.setId(timeBucket + Const.ID_SPLIT + applicationComponent.getId());

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingAggregationWorker.java
View File

@ -16,14 +16,13 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
/**
* @author peng-yongsheng
@ -35,13 +34,13 @@ public class ApplicationMappingAggregationWorker extends AggregationWorker<Appli
}
@Override public int id() {
return 105;
return WorkerIdDefine.APPLICATION_MAPPING_AGGREGATION_WORKER_ID;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationMapping, ApplicationMapping, ApplicationMappingAggregationWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ApplicationMapping> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationMappingAggregationWorker workerInstance(ModuleManager moduleManager) {

47
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingGraph.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
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.application.ApplicationMapping;
/**
* @author peng-yongsheng
*/
public class ApplicationMappingGraph {
private final ModuleManager moduleManager;
public ApplicationMappingGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class)
.addNode(new ApplicationMappingAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ApplicationMappingRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID).create(null))
.addNext(new ApplicationMappingPersistenceWorker.Factory(moduleManager).create(null));
}
}

17
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
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.IApplicationMappingPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ApplicationMappingPersistenceWorker extends PersistenceWorker<Appli
}
@Override public int id() {
return ApplicationMappingPersistenceWorker.class.hashCode();
return WorkerIdDefine.APPLICATION_MAPPING_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -51,8 +50,8 @@ public class ApplicationMappingPersistenceWorker extends PersistenceWorker<Appli
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationMapping, ApplicationMapping, ApplicationMappingPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ApplicationMapping> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationMappingPersistenceWorker workerInstance(ModuleManager moduleManager) {

14
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingRemoteWorker.java
View File

@ -16,16 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMapping;
/**
* @author peng-yongsheng
@ -37,7 +37,7 @@ public class ApplicationMappingRemoteWorker extends AbstractRemoteWorker<Applica
}
@Override public int id() {
return 10007;
return WorkerIdDefine.APPLICATION_MAPPING_REMOTE_WORKER_ID;
}
@Override protected void onWork(ApplicationMapping applicationMapping) throws WorkerException {

13
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMappingSpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMappingSpanListener.java
View File

@ -16,15 +16,14 @@
*
*/
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
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.util.Const;
@ -63,7 +62,7 @@ public class ApplicationMappingSpanListener implements FirstSpanListener, EntryS
@Override public void build() {
logger.debug("node mapping listener build");
Graph<ApplicationMapping> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class);
Graph<ApplicationMapping> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.APPLICATION_MAPPING_GRAPH_ID, ApplicationMapping.class);
applicationMappings.forEach(applicationMapping -> {
applicationMapping.setId(timeBucket + Const.ID_SPLIT + applicationMapping.getId());
applicationMapping.setTimeBucket(timeBucket);

40
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricAggregationWorker.java
View File

@ -16,16 +16,15 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
/**
* @author peng-yongsheng
@ -37,7 +36,7 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker<Applic
}
@Override public int id() {
return ApplicationMetricAggregationWorker.class.hashCode();
return WorkerIdDefine.APPLICATION_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected ApplicationMetric transform(ApplicationReferenceMetric applicationReferenceMetric) {
@ -45,10 +44,22 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker<Applic
Long timeBucket = applicationReferenceMetric.getTimeBucket();
ApplicationMetric applicationMetric = new ApplicationMetric(String.valueOf(timeBucket) + Const.ID_SPLIT + String.valueOf(applicationId));
applicationMetric.setApplicationId(applicationId);
// applicationMetric.setCalls(applicationReferenceMetric.getCalls());
// applicationMetric.setErrorCalls(applicationReferenceMetric.getErrorCalls());
// applicationMetric.setDurationSum(applicationReferenceMetric.getDurationSum());
// applicationMetric.setErrorDurationSum(applicationReferenceMetric.getErrorDurationSum());
applicationMetric.setTransactionCalls(applicationReferenceMetric.getTransactionCalls());
applicationMetric.setTransactionDurationSum(applicationReferenceMetric.getTransactionDurationSum());
applicationMetric.setTransactionErrorCalls(applicationReferenceMetric.getTransactionErrorCalls());
applicationMetric.setTransactionErrorDurationSum(applicationReferenceMetric.getTransactionErrorDurationSum());
applicationMetric.setBusinessTransactionCalls(applicationReferenceMetric.getBusinessTransactionCalls());
applicationMetric.setBusinessTransactionDurationSum(applicationReferenceMetric.getBusinessTransactionDurationSum());
applicationMetric.setBusinessTransactionErrorCalls(applicationReferenceMetric.getBusinessTransactionErrorCalls());
applicationMetric.setBusinessTransactionErrorDurationSum(applicationReferenceMetric.getBusinessTransactionErrorDurationSum());
applicationMetric.setMqTransactionCalls(applicationReferenceMetric.getMqTransactionCalls());
applicationMetric.setMqTransactionDurationSum(applicationReferenceMetric.getMqTransactionDurationSum());
applicationMetric.setMqTransactionErrorCalls(applicationReferenceMetric.getMqTransactionErrorCalls());
applicationMetric.setMqTransactionErrorDurationSum(applicationReferenceMetric.getMqTransactionErrorDurationSum());
applicationMetric.setSatisfiedCount(applicationReferenceMetric.getSatisfiedCount());
applicationMetric.setToleratingCount(applicationReferenceMetric.getToleratingCount());
applicationMetric.setFrustratedCount(applicationReferenceMetric.getFrustratedCount());
@ -59,9 +70,8 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker<Applic
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationReferenceMetric, ApplicationMetric, ApplicationMetricAggregationWorker> {
public Factory(ModuleManager moduleManager,
QueueCreatorService<ApplicationReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationMetricAggregationWorker workerInstance(ModuleManager moduleManager) {

47
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricGraph.java Normal file
View File

@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
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.application.ApplicationReferenceMetric;
/**
* @author peng-yongsheng
*/
public class ApplicationMetricGraph {
private final ModuleManager moduleManager;
public ApplicationMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_METRIC_GRAPH_ID, ApplicationReferenceMetric.class)
.addNode(new ApplicationMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ApplicationMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_METRIC_GRAPH_ID).create(null))
.addNext(new ApplicationMetricPersistenceWorker.Factory(moduleManager).create(null));
}
}

21
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.dao.IApplicationMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ApplicationMetricPersistenceWorker extends PersistenceWorker<Applic
}
@Override public int id() {
return 111;
return WorkerIdDefine.APPLICATION_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected IPersistenceDAO persistenceDAO() {
@ -51,8 +50,8 @@ public class ApplicationMetricPersistenceWorker extends PersistenceWorker<Applic
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationMetric, ApplicationMetric, ApplicationMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ApplicationMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {

12
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationMetricRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationMetricRemoteWorker.java
View File

@ -16,16 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationMetric;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
/**
* @author peng-yongsheng
@ -37,7 +37,7 @@ public class ApplicationMetricRemoteWorker extends AbstractRemoteWorker<Applicat
}
@Override public int id() {
return 10002;
return WorkerIdDefine.APPLICATION_METRIC_REMOTE_WORKER_ID;
}
@Override protected void onWork(ApplicationMetric applicationMetric) throws WorkerException {

38
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricAggregationWorker.java
View File

@ -16,19 +16,20 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ApplicationGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.service.InstanceCacheService;
import org.apache.skywalking.apm.collector.configuration.ConfigurationModule;
import org.apache.skywalking.apm.collector.configuration.service.IApdexThresholdService;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.ApdexThresholdUtils;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
/**
* @author peng-yongsheng
@ -36,14 +37,16 @@ import org.apache.skywalking.apm.collector.analysis.worker.model.impl.Aggregatio
public class ApplicationReferenceMetricAggregationWorker extends AggregationWorker<InstanceReferenceMetric, ApplicationReferenceMetric> {
private final InstanceCacheService instanceCacheService;
private final IApdexThresholdService apdexThresholdService;
public ApplicationReferenceMetricAggregationWorker(ModuleManager moduleManager) {
super(moduleManager);
this.instanceCacheService = moduleManager.find(CacheModule.NAME).getService(InstanceCacheService.class);
this.apdexThresholdService = moduleManager.find(ConfigurationModule.NAME).getService(IApdexThresholdService.class);
}
@Override public int id() {
return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_AGGREGATION_NODE_ID;
return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected ApplicationReferenceMetric transform(InstanceReferenceMetric instanceReferenceMetric) {
@ -71,14 +74,21 @@ public class ApplicationReferenceMetricAggregationWorker extends AggregationWork
applicationReferenceMetric.setMqTransactionDurationSum(instanceReferenceMetric.getMqTransactionDurationSum());
applicationReferenceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum());
ApdexThresholdUtils.Apdex apdex = ApdexThresholdUtils.compute(apdexThresholdService.getApplicationApdexThreshold(behindApplicationId), instanceReferenceMetric.getTransactionDurationSum());
if (ApdexThresholdUtils.Apdex.Satisfied.equals(apdex)) {
applicationReferenceMetric.setSatisfiedCount(1L);
} else if (ApdexThresholdUtils.Apdex.Tolerating.equals(apdex)) {
applicationReferenceMetric.setToleratingCount(1L);
} else {
applicationReferenceMetric.setFrustratedCount(1L);
}
return applicationReferenceMetric;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<InstanceReferenceMetric, ApplicationReferenceMetric, ApplicationReferenceMetricAggregationWorker> {
public Factory(ModuleManager moduleManager,
QueueCreatorService<InstanceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {

69
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricGraph.java Normal file
View File

@ -0,0 +1,69 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
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.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.instance.InstanceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class ApplicationReferenceMetricGraph {
private final ModuleManager moduleManager;
public ApplicationReferenceMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
Graph<InstanceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class);
graph.addNode(new ApplicationReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ApplicationReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.APPLICATION_REFERENCE_METRIC_GRAPH_ID).create(null))
.addNext(new ApplicationReferenceMetricPersistenceWorker.Factory(moduleManager).create(null));
link(graph);
}
private void link(Graph<InstanceReferenceMetric> graph) {
GraphManager.INSTANCE.findGraph(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class)
.toFinder().findNode(WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, InstanceReferenceMetric.class)
.addNext(new NodeProcessor<InstanceReferenceMetric, InstanceReferenceMetric>() {
@Override public int id() {
return WorkerIdDefine.APPLICATION_REFERENCE_GRAPH_BRIDGE_WORKER_ID;
}
@Override public void process(InstanceReferenceMetric instanceReferenceMetric,
Next<InstanceReferenceMetric> next) {
graph.start(instanceReferenceMetric);
}
});
}
}

17
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricPersistenceWorker.java
View File

@ -16,18 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ApplicationGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
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.IApplicationReferenceMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
/**
* @author peng-yongsheng
@ -39,7 +37,7 @@ public class ApplicationReferenceMetricPersistenceWorker extends PersistenceWork
}
@Override public int id() {
return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_PERSISTENCE_NODE_ID;
return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -52,9 +50,8 @@ public class ApplicationReferenceMetricPersistenceWorker extends PersistenceWork
public static class Factory extends AbstractLocalAsyncWorkerProvider<ApplicationReferenceMetric, ApplicationReferenceMetric, ApplicationReferenceMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager,
QueueCreatorService<ApplicationReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ApplicationReferenceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {

11
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/application/ApplicationReferenceMetricRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/application/ApplicationReferenceMetricRemoteWorker.java
View File

@ -16,15 +16,14 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.application;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.application;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ApplicationGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.application.ApplicationReferenceMetric;
@ -38,7 +37,7 @@ public class ApplicationReferenceMetricRemoteWorker extends AbstractRemoteWorker
}
@Override public int id() {
return ApplicationGraphNodeIdDefine.APPLICATION_REFERENCE_METRIC_REMOTE_NODE_ID;
return WorkerIdDefine.APPLICATION_REFERENCE_METRIC_REMOTE_WORKER_ID;
}
@Override protected void onWork(ApplicationReferenceMetric applicationReferenceMetric) throws WorkerException {

41
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceGraph.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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.global;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
/**
* @author peng-yongsheng
*/
public class GlobalTraceGraph {
private final ModuleManager moduleManager;
public GlobalTraceGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class)
.addNode(new GlobalTracePersistenceWorker.Factory(moduleManager).create(null));
}
}

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTracePersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTracePersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.global;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
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.IGlobalTracePersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class GlobalTracePersistenceWorker extends PersistenceWorker<GlobalTrace,
}
@Override public int id() {
return 120;
return WorkerIdDefine.GLOBAL_TRACE_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -51,8 +50,8 @@ public class GlobalTracePersistenceWorker extends PersistenceWorker<GlobalTrace,
public static class Factory extends AbstractLocalAsyncWorkerProvider<GlobalTrace, GlobalTrace, GlobalTracePersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<GlobalTrace> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public GlobalTracePersistenceWorker workerInstance(ModuleManager moduleManager) {

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/global/GlobalTraceSpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/global/GlobalTraceSpanListener.java
View File

@ -16,20 +16,19 @@
*
*/
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.global;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.global;
import java.util.ArrayList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.GlobalTraceIdsListener;
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.util.Const;
import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
import org.apache.skywalking.apm.collector.agent.stream.parser.GlobalTraceIdsListener;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
import org.apache.skywalking.apm.collector.storage.table.global.GlobalTrace;
import org.apache.skywalking.apm.network.proto.UniqueId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -67,7 +66,7 @@ public class GlobalTraceSpanListener implements FirstSpanListener, GlobalTraceId
@Override public void build() {
logger.debug("global trace listener build");
Graph<GlobalTrace> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class);
Graph<GlobalTrace> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.GLOBAL_TRACE_GRAPH_ID, GlobalTrace.class);
for (String globalTraceId : globalTraceIds) {
GlobalTrace globalTrace = new GlobalTrace(segmentId + Const.ID_SPLIT + globalTraceId);
globalTrace.setGlobalTraceId(globalTraceId);

81
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricAggregationWorker.java Normal file
View File

@ -0,0 +1,81 @@
/*
* 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.instance;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class InstanceMetricAggregationWorker extends AggregationWorker<InstanceReferenceMetric, InstanceMetric> {
public InstanceMetricAggregationWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public int id() {
return WorkerIdDefine.INSTANCE_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected InstanceMetric transform(InstanceReferenceMetric instanceReferenceMetric) {
String id = instanceReferenceMetric.getTimeBucket() + Const.ID_SPLIT + instanceReferenceMetric.getBehindInstanceId();
InstanceMetric instanceMetric = new InstanceMetric(id);
instanceMetric.setInstanceId(instanceReferenceMetric.getBehindInstanceId());
instanceMetric.setTransactionCalls(instanceReferenceMetric.getTransactionCalls());
instanceMetric.setTransactionErrorCalls(instanceReferenceMetric.getTransactionErrorCalls());
instanceMetric.setTransactionDurationSum(instanceReferenceMetric.getTransactionDurationSum());
instanceMetric.setTransactionErrorDurationSum(instanceReferenceMetric.getTransactionErrorDurationSum());
instanceMetric.setBusinessTransactionCalls(instanceReferenceMetric.getBusinessTransactionCalls());
instanceMetric.setBusinessTransactionErrorCalls(instanceReferenceMetric.getBusinessTransactionErrorCalls());
instanceMetric.setBusinessTransactionDurationSum(instanceReferenceMetric.getBusinessTransactionDurationSum());
instanceMetric.setBusinessTransactionErrorDurationSum(instanceReferenceMetric.getBusinessTransactionErrorDurationSum());
instanceMetric.setMqTransactionCalls(instanceReferenceMetric.getMqTransactionCalls());
instanceMetric.setMqTransactionErrorCalls(instanceReferenceMetric.getMqTransactionErrorCalls());
instanceMetric.setMqTransactionDurationSum(instanceReferenceMetric.getMqTransactionDurationSum());
instanceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum());
return instanceMetric;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<InstanceReferenceMetric, InstanceMetric, InstanceMetricAggregationWorker> {
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public InstanceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {
return new InstanceMetricAggregationWorker(moduleManager);
}
@Override
public int queueSize() {
return 1024;
}
}
}

71
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricGraph.java Normal file
View File

@ -0,0 +1,71 @@
/*
* 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.instance;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
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.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.instance.InstanceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class InstanceMetricGraph {
private final ModuleManager moduleManager;
public InstanceMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
Graph<InstanceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class);
graph.addNode(new InstanceMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new InstanceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_METRIC_GRAPH_ID).create(null))
.addNext(new InstanceMetricPersistenceWorker.Factory(moduleManager).create(null));
link(graph);
}
private void link(Graph<InstanceReferenceMetric> graph) {
GraphManager.INSTANCE.findGraph(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, InstanceReferenceMetric.class)
.toFinder().findNode(WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, InstanceReferenceMetric.class)
.addNext(new NodeProcessor<InstanceReferenceMetric, InstanceReferenceMetric>() {
@Override public int id() {
return WorkerIdDefine.INSTANCE_METRIC_GRAPH_BRIDGE_WORKER_ID;
}
@Override
public void process(InstanceReferenceMetric instanceReferenceMetric,
Next<InstanceReferenceMetric> next) {
graph.start(instanceReferenceMetric);
}
});
}
}

17
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceMetricPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.IInstanceMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class InstanceMetricPersistenceWorker extends PersistenceWorker<InstanceM
}
@Override public int id() {
return 118;
return WorkerIdDefine.INSTANCE_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -51,8 +50,8 @@ public class InstanceMetricPersistenceWorker extends PersistenceWorker<InstanceM
public static class Factory extends AbstractLocalAsyncWorkerProvider<InstanceMetric, InstanceMetric, InstanceMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<InstanceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public InstanceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {

61
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceMetricRemoteWorker.java Normal file
View File

@ -0,0 +1,61 @@
/*
* 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.instance;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceMetric;
/**
* @author peng-yongsheng
*/
public class InstanceMetricRemoteWorker extends AbstractRemoteWorker<InstanceMetric, InstanceMetric> {
public InstanceMetricRemoteWorker(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public int id() {
return WorkerIdDefine.INSTANCE_METRIC_REMOTE_WORKER_ID;
}
@Override public Selector selector() {
return Selector.HashCode;
}
@Override protected void onWork(InstanceMetric instanceMetric) throws WorkerException {
onNext(instanceMetric);
}
public static class Factory extends AbstractRemoteWorkerProvider<InstanceMetric, InstanceMetric, InstanceMetricRemoteWorker> {
public Factory(ModuleManager moduleManager, RemoteSenderService remoteSenderService, int graphId) {
super(moduleManager, remoteSenderService, graphId);
}
@Override public InstanceMetricRemoteWorker workerInstance(ModuleManager moduleManager) {
return new InstanceMetricRemoteWorker(moduleManager);
}
}
}

20
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricAggregationWorker.java
View File

@ -16,17 +16,15 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.InstanceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
/**
* @author peng-yongsheng
@ -38,7 +36,7 @@ public class InstanceReferenceMetricAggregationWorker extends AggregationWorker<
}
@Override public int id() {
return InstanceGraphNodeIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_NODE_ID;
return WorkerIdDefine.INSTANCE_REFERENCE_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected InstanceReferenceMetric transform(ServiceReferenceMetric serviceReferenceMetric) {
@ -69,8 +67,8 @@ public class InstanceReferenceMetricAggregationWorker extends AggregationWorker<
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, InstanceReferenceMetric, InstanceReferenceMetricAggregationWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public InstanceReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {

69
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricGraph.java Normal file
View File

@ -0,0 +1,69 @@
/*
* 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.instance;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
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.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.ServiceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class InstanceReferenceMetricGraph {
private final ModuleManager moduleManager;
public InstanceReferenceMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
graph.addNode(new InstanceReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new InstanceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.INSTANCE_REFERENCE_METRIC_GRAPH_ID).create(null))
.addNext(new InstanceReferencePersistenceWorker.Factory(moduleManager).create(null));
link(graph);
}
private void link(Graph<ServiceReferenceMetric> graph) {
GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class)
.toFinder().findNode(WorkerIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, ServiceReferenceMetric.class)
.addNext(new NodeProcessor<ServiceReferenceMetric, ServiceReferenceMetric>() {
@Override public int id() {
return WorkerIdDefine.INSTANCE_REFERENCE_GRAPH_BRIDGE_WORKER_ID;
}
@Override
public void process(ServiceReferenceMetric serviceReferenceMetric, Next<ServiceReferenceMetric> next) {
graph.start(serviceReferenceMetric);
}
});
}
}

17
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferenceMetricRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferenceMetricRemoteWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.InstanceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class InstanceReferenceMetricRemoteWorker extends AbstractRemoteWorker<In
}
@Override public int id() {
return InstanceGraphNodeIdDefine.INSTANCE_REFERENCE_METRIC_REMOTE_NODE_ID;
return WorkerIdDefine.INSTANCE_REFERENCE_METRIC_REMOTE_WORKER_ID;
}
@Override public Selector selector() {

20
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/instance/InstanceReferencePersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/instance/InstanceReferencePersistenceWorker.java
View File

@ -16,18 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.instance;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.instance;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.InstanceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.IInstanceReferenceMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.instance.InstanceReferenceMetric;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
/**
* @author peng-yongsheng
@ -39,7 +37,7 @@ public class InstanceReferencePersistenceWorker extends PersistenceWorker<Instan
}
@Override public int id() {
return InstanceGraphNodeIdDefine.INSTANCE_REFERENCE_METRIC_PERSISTENCE_NODE_ID;
return WorkerIdDefine.INSTANCE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected IPersistenceDAO persistenceDAO() {
@ -52,8 +50,8 @@ public class InstanceReferencePersistenceWorker extends PersistenceWorker<Instan
public static class Factory extends AbstractLocalAsyncWorkerProvider<InstanceReferenceMetric, InstanceReferenceMetric, InstanceReferencePersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<InstanceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public InstanceReferencePersistenceWorker workerInstance(ModuleManager moduleManager) {

41
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostGraph.java Normal file
View File

@ -0,0 +1,41 @@
/*
* 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.segment;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
/**
* @author peng-yongsheng
*/
public class SegmentCostGraph {
private final ModuleManager moduleManager;
public SegmentCostGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class)
.addNode(new SegmentCostPersistenceWorker.Factory(moduleManager).create(null));
}
}

21
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.dao.ISegmentCostPersistenceDAO;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.ISegmentCostPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.segment.SegmentCost;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class SegmentCostPersistenceWorker extends PersistenceWorker<SegmentCost,
}
@Override public int id() {
return 110;
return WorkerIdDefine.SEGMENT_COST_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -51,8 +50,8 @@ public class SegmentCostPersistenceWorker extends PersistenceWorker<SegmentCost,
public static class Factory extends AbstractLocalAsyncWorkerProvider<SegmentCost, SegmentCost, SegmentCostPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<SegmentCost> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public SegmentCostPersistenceWorker workerInstance(ModuleManager moduleManager) {

17
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/segment/SegmentCostSpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/segment/SegmentCostSpanListener.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.segment;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.segment;
import java.util.ArrayList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.LocalSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.ExitSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.LocalSpanListener;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService;
import org.apache.skywalking.apm.collector.core.graph.Graph;
@ -95,7 +94,7 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe
}
@Override public void build() {
Graph<SegmentCost> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SEGMENT_COST_GRAPH_ID, SegmentCost.class);
Graph<SegmentCost> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.SEGMENT_COST_GRAPH_ID, SegmentCost.class);
logger.debug("segment cost listener build");
for (SegmentCost segmentCost : segmentCosts) {
segmentCost.setIsError(isError);

15
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryAggregationWorker.java
View File

@ -16,14 +16,13 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
/**
* @author peng-yongsheng
@ -35,13 +34,13 @@ public class ServiceEntryAggregationWorker extends AggregationWorker<ServiceEntr
}
@Override public int id() {
return 103;
return WorkerIdDefine.SERVICE_ENTRY_AGGREGATION_WORKER_ID;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceEntry, ServiceEntry, ServiceEntryAggregationWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceEntry> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceEntryAggregationWorker workerInstance(ModuleManager moduleManager) {

47
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryGraph.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
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.ServiceEntry;
/**
* @author peng-yongsheng
*/
public class ServiceEntryGraph {
private final ModuleManager moduleManager;
public ServiceEntryGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class)
.addNode(new ServiceEntryAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ServiceEntryRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_ENTRY_GRAPH_ID).create(null))
.addNext(new ServiceEntryPersistenceWorker.Factory(moduleManager).create(null));
}
}

21
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.dao.IServiceEntryPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.IServiceEntryPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ServiceEntryPersistenceWorker extends PersistenceWorker<ServiceEntr
}
@Override public int id() {
return 121;
return WorkerIdDefine.SERVICE_ENTRY_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -50,8 +49,8 @@ public class ServiceEntryPersistenceWorker extends PersistenceWorker<ServiceEntr
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceEntry, ServiceEntry, ServiceEntryPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceEntry> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceEntryPersistenceWorker workerInstance(ModuleManager moduleManager) {

12
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntryRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntryRemoteWorker.java
View File

@ -16,16 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceEntry;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
/**
* @author peng-yongsheng
@ -37,7 +37,7 @@ public class ServiceEntryRemoteWorker extends AbstractRemoteWorker<ServiceEntry,
}
@Override public int id() {
return 10003;
return WorkerIdDefine.SERVICE_ENTRY_REMOTE_WORKER_ID;
}
@Override protected void onWork(ServiceEntry serviceEntry) throws WorkerException {

13
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceEntrySpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceEntrySpanListener.java
View File

@ -16,13 +16,12 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
import org.apache.skywalking.apm.collector.cache.CacheModule;
import org.apache.skywalking.apm.collector.cache.service.ServiceNameCacheService;
import org.apache.skywalking.apm.collector.core.graph.Graph;
@ -82,7 +81,7 @@ public class ServiceEntrySpanListener implements FirstSpanListener, EntrySpanLis
serviceEntry.setNewestTime(timeBucket);
logger.debug("push to service entry aggregation worker, id: {}", serviceEntry.getId());
Graph<ServiceEntry> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class);
Graph<ServiceEntry> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_ENTRY_GRAPH_ID, ServiceEntry.class);
graph.start(serviceEntry);
}
}

40
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/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/ServiceMetricAggregationWorker.java
View File

@ -16,16 +16,15 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
/**
* @author peng-yongsheng
@ -37,7 +36,7 @@ public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceRef
}
@Override public int id() {
return ServiceMetricAggregationWorker.class.hashCode();
return WorkerIdDefine.SERVICE_METRIC_AGGREGATION_WORKER_ID;
}
@Override protected ServiceMetric transform(ServiceReferenceMetric serviceReferenceMetric) {
@ -45,10 +44,22 @@ public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceRef
Long timeBucket = serviceReferenceMetric.getTimeBucket();
ServiceMetric serviceMetric = new ServiceMetric(String.valueOf(timeBucket) + Const.ID_SPLIT + String.valueOf(serviceId));
serviceMetric.setServiceId(serviceId);
// serviceMetric.setCalls(serviceReferenceMetric.getCalls());
// serviceMetric.setErrorCalls(serviceReferenceMetric.getErrorCalls());
// serviceMetric.setDurationSum(serviceReferenceMetric.getDurationSum());
// serviceMetric.setErrorDurationSum(serviceReferenceMetric.getErrorDurationSum());
serviceMetric.setTransactionCalls(serviceReferenceMetric.getTransactionCalls());
serviceMetric.setTransactionDurationSum(serviceReferenceMetric.getTransactionDurationSum());
serviceMetric.setTransactionErrorCalls(serviceReferenceMetric.getTransactionErrorCalls());
serviceMetric.setTransactionErrorDurationSum(serviceReferenceMetric.getTransactionErrorDurationSum());
serviceMetric.setBusinessTransactionCalls(serviceReferenceMetric.getBusinessTransactionCalls());
serviceMetric.setBusinessTransactionDurationSum(serviceReferenceMetric.getBusinessTransactionDurationSum());
serviceMetric.setBusinessTransactionErrorCalls(serviceReferenceMetric.getBusinessTransactionErrorCalls());
serviceMetric.setBusinessTransactionErrorDurationSum(serviceReferenceMetric.getBusinessTransactionErrorDurationSum());
serviceMetric.setMqTransactionCalls(serviceReferenceMetric.getMqTransactionCalls());
serviceMetric.setMqTransactionDurationSum(serviceReferenceMetric.getMqTransactionDurationSum());
serviceMetric.setMqTransactionErrorCalls(serviceReferenceMetric.getMqTransactionErrorCalls());
serviceMetric.setMqTransactionErrorDurationSum(serviceReferenceMetric.getMqTransactionErrorDurationSum());
serviceMetric.setTimeBucket(timeBucket);
return serviceMetric;
@ -56,9 +67,8 @@ public class ServiceMetricAggregationWorker extends AggregationWorker<ServiceRef
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, ServiceMetric, ServiceMetricAggregationWorker> {
public Factory(ModuleManager moduleManager,
QueueCreatorService<ServiceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {

69
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceMetricGraph.java Normal file
View File

@ -0,0 +1,69 @@
/*
* 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;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
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.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.ServiceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class ServiceMetricGraph {
private final ModuleManager moduleManager;
public ServiceMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
graph.addNode(new ServiceMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ServiceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_METRIC_GRAPH_ID).create(null))
.addNext(new ServiceMetricPersistenceWorker.Factory(moduleManager).create(null));
link(graph);
}
private void link(Graph<ServiceReferenceMetric> graph) {
GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class)
.toFinder().findNode(WorkerIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID, ServiceReferenceMetric.class)
.addNext(new NodeProcessor<ServiceReferenceMetric, ServiceReferenceMetric>() {
@Override public int id() {
return WorkerIdDefine.SERVICE_METRIC_GRAPH_BRIDGE_WORKER_ID;
}
@Override
public void process(ServiceReferenceMetric serviceReferenceMetric, Next<ServiceReferenceMetric> next) {
graph.start(serviceReferenceMetric);
}
});
}
}

21
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/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/ServiceMetricPersistenceWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.base.dao.IPersistenceDAO;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.storage.dao.IServiceMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.table.service.ServiceMetric;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ServiceMetricPersistenceWorker extends PersistenceWorker<ServiceMet
}
@Override public int id() {
return 116;
return WorkerIdDefine.SERVICE_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected IPersistenceDAO persistenceDAO() {
@ -51,8 +50,8 @@ public class ServiceMetricPersistenceWorker extends PersistenceWorker<ServiceMet
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceMetric, ServiceMetric, ServiceMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {

12
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/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/ServiceMetricRemoteWorker.java
View File

@ -16,16 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
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.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceMetric;
/**
* @author peng-yongsheng
@ -37,7 +37,7 @@ public class ServiceMetricRemoteWorker extends AbstractRemoteWorker<ServiceMetri
}
@Override public int id() {
return 10004;
return WorkerIdDefine.SERVICE_METRIC_REMOTE_WORKER_ID;
}
@Override protected void onWork(ServiceMetric serviceMetric) throws WorkerException {

16
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricAggregationWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricAggregationWorker.java
View File

@ -16,15 +16,13 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ServiceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.AggregationWorker;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
/**
* @author peng-yongsheng
@ -36,13 +34,13 @@ public class ServiceReferenceMetricAggregationWorker extends AggregationWorker<S
}
@Override public int id() {
return ServiceGraphNodeIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_NODE_ID;
return WorkerIdDefine.SERVICE_REFERENCE_METRIC_AGGREGATION_WORKER_ID;
}
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, ServiceReferenceMetric, ServiceReferenceMetricAggregationWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceReferenceMetricAggregationWorker workerInstance(ModuleManager moduleManager) {

47
apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricGraph.java Normal file
View File

@ -0,0 +1,47 @@
/*
* 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;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.core.graph.GraphManager;
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.ServiceReferenceMetric;
/**
* @author peng-yongsheng
*/
public class ServiceReferenceMetricGraph {
private final ModuleManager moduleManager;
public ServiceReferenceMetricGraph(ModuleManager moduleManager) {
this.moduleManager = moduleManager;
}
public void create() {
RemoteSenderService remoteSenderService = moduleManager.find(RemoteModule.NAME).getService(RemoteSenderService.class);
GraphManager.INSTANCE.createIfAbsent(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class)
.addNode(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null))
.addNext(new ServiceReferenceMetricRemoteWorker.Factory(moduleManager, remoteSenderService, GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID).create(null))
.addNext(new ServiceReferenceMetricAggregationWorker.Factory(moduleManager).create(null));
}
}

18
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricPersistenceWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricPersistenceWorker.java
View File

@ -16,18 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ServiceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
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.IServiceReferenceMetricPersistenceDAO;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.collector.queue.service.QueueCreatorService;
import org.apache.skywalking.apm.collector.storage.StorageModule;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractLocalAsyncWorkerProvider;
import org.apache.skywalking.apm.collector.analysis.worker.model.impl.PersistenceWorker;
/**
* @author peng-yongsheng
@ -39,7 +37,7 @@ public class ServiceReferenceMetricPersistenceWorker extends PersistenceWorker<S
}
@Override public int id() {
return ServiceGraphNodeIdDefine.SERVICE_REFERENCE_METRIC_PERSISTENCE_NODE_ID;
return WorkerIdDefine.SERVICE_REFERENCE_METRIC_PERSISTENCE_WORKER_ID;
}
@Override protected boolean needMergeDBData() {
@ -52,8 +50,8 @@ public class ServiceReferenceMetricPersistenceWorker extends PersistenceWorker<S
public static class Factory extends AbstractLocalAsyncWorkerProvider<ServiceReferenceMetric, ServiceReferenceMetric, ServiceReferenceMetricPersistenceWorker> {
public Factory(ModuleManager moduleManager, QueueCreatorService<ServiceReferenceMetric> queueCreatorService) {
super(moduleManager, queueCreatorService);
public Factory(ModuleManager moduleManager) {
super(moduleManager);
}
@Override public ServiceReferenceMetricPersistenceWorker workerInstance(ModuleManager moduleManager) {

13
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricRemoteWorker.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricRemoteWorker.java
View File

@ -16,17 +16,16 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.WorkerIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorkerProvider;
import org.apache.skywalking.apm.collector.agent.stream.service.graph.ServiceGraphNodeIdDefine;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
import org.apache.skywalking.apm.collector.remote.service.RemoteSenderService;
import org.apache.skywalking.apm.collector.remote.service.Selector;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.AbstractRemoteWorker;
import org.apache.skywalking.apm.collector.analysis.worker.model.base.WorkerException;
/**
* @author peng-yongsheng
@ -38,7 +37,7 @@ public class ServiceReferenceMetricRemoteWorker extends AbstractRemoteWorker<Ser
}
@Override public int id() {
return ServiceGraphNodeIdDefine.SERVICE_REFERENCE_METRIC_REMOTE_NODE_ID;
return WorkerIdDefine.SERVICE_REFERENCE_METRIC_REMOTE_WORKER_ID;
}
@Override protected void onWork(ServiceReferenceMetric serviceReferenceMetric) throws WorkerException {

19
apm-collector/apm-collector-agent-stream/collector-agent-stream-provider/src/main/java/org/apache/skywalking/apm/collector/agent/stream/worker/trace/service/ServiceReferenceMetricSpanListener.java → apm-collector/apm-collector-analysis/analysis-metric/metric-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/metric/provider/worker/service/ServiceReferenceMetricSpanListener.java
View File

@ -16,22 +16,21 @@
*
*/
package org.apache.skywalking.apm.collector.agent.stream.worker.trace.service;
package org.apache.skywalking.apm.collector.analysis.metric.provider.worker.service;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.apm.collector.agent.stream.graph.TraceStreamGraph;
import org.apache.skywalking.apm.collector.agent.stream.parser.EntrySpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.ExitSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.FirstSpanListener;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.SpanDecorator;
import org.apache.skywalking.apm.collector.agent.stream.service.trace.MetricSource;
import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdDefine;
import org.apache.skywalking.apm.collector.analysis.metric.provider.worker.MetricSource;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.ReferenceDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SpanDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.EntrySpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.ExitSpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.FirstSpanListener;
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.util.Const;
import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.collector.agent.stream.parser.standardization.ReferenceDecorator;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
import org.apache.skywalking.apm.collector.storage.table.service.ServiceReferenceMetric;
import org.apache.skywalking.apm.network.proto.SpanLayer;
@ -134,7 +133,7 @@ public class ServiceReferenceMetricSpanListener implements FirstSpanListener, En
@Override public void build() {
logger.debug("service reference listener build");
Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.createIfAbsent(TraceStreamGraph.SERVICE_REFERENCE_GRAPH_ID, ServiceReferenceMetric.class);
Graph<ServiceReferenceMetric> graph = GraphManager.INSTANCE.findGraph(GraphIdDefine.SERVICE_REFERENCE_METRIC_GRAPH_ID, ServiceReferenceMetric.class);
entryReferenceMetric.forEach(serviceReferenceMetric -> {
String id = timeBucket + Const.ID_SPLIT + serviceReferenceMetric.getEntryServiceId() + Const.ID_SPLIT + serviceReferenceMetric.getFrontServiceId() + Const.ID_SPLIT + serviceReferenceMetric.getBehindServiceId();

20
apm-collector/apm-collector-core/src/main/java/org/apache/skywalking/apm/collector/core/graph/GraphNodeFinder.java
View File

@ -16,7 +16,6 @@
*
*/
package org.apache.skywalking.apm.collector.core.graph;
import java.util.concurrent.ConcurrentHashMap;
@ -41,4 +40,23 @@ public class GraphNodeFinder {
}
return node.getNext();
}
/**
* Find an exist node to build the graph.
*
* @param handlerId of specific node in graph.
* @param outputClass of the found node
* @param <NODEOUTPUT> type of given output class
* @return Node instance.
*/
public <NODEOUTPUT> Node<?, NODEOUTPUT> findNode(int handlerId, Class<NODEOUTPUT> outputClass) {
ConcurrentHashMap<Integer, Node> graphNodeIndex = graph.getNodeIndex();
Node node = graphNodeIndex.get(handlerId);
if (node == null) {
throw new NodeNotFoundException("Can't find node with handlerId="
+ handlerId
+ " in graph[" + graph.getId() + "]");
}
return node;
}
}