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

Use listener factory to create new instance when create SegmentParser instance. Because of the listener is stateful.

This commit is contained in:
peng-yongsheng 2017-12-21 22:47:06 +08:00
parent c1bd835894
commit 4a73a25623
20 changed files with 166 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/main/java/org/apache/skywalking/apm/collector/agent/jetty/provider/handler/reader/ReferenceJsonReader.java
View File

@ -16,7 +16,6 @@
*
*/
package org.apache.skywalking.apm.collector.agent.jetty.provider.handler.reader;
import com.google.gson.stream.JsonReader;
@ -30,16 +29,16 @@ public class ReferenceJsonReader implements StreamJsonReader<TraceSegmentReferen
private UniqueIdJsonReader uniqueIdJsonReader = new UniqueIdJsonReader();
private static final String PARENT_TRACE_SEGMENT_ID = "ts";
private static final String PARENT_APPLICATION_ID = "ai";
private static final String PARENT_SPAN_ID = "si";
private static final String PARENT_SERVICE_ID = "vi";
private static final String PARENT_SERVICE_NAME = "vn";
private static final String PARENT_TRACE_SEGMENT_ID = "pts";
private static final String PARENT_APPLICATION_INSTANCE_ID = "pii";
private static final String PARENT_SPAN_ID = "psp";
private static final String PARENT_SERVICE_ID = "psi";
private static final String PARENT_SERVICE_NAME = "psn";
private static final String NETWORK_ADDRESS_ID = "ni";
private static final String NETWORK_ADDRESS = "nn";
private static final String ENTRY_APPLICATION_INSTANCE_ID = "ea";
private static final String ENTRY_SERVICE_ID = "ei";
private static final String ENTRY_SERVICE_NAME = "en";
private static final String ENTRY_APPLICATION_INSTANCE_ID = "eii";
private static final String ENTRY_SERVICE_ID = "esi";
private static final String ENTRY_SERVICE_NAME = "esn";
private static final String REF_TYPE_VALUE = "rv";
@Override public TraceSegmentReference read(JsonReader reader) throws IOException {
@ -51,7 +50,7 @@ public class ReferenceJsonReader implements StreamJsonReader<TraceSegmentReferen
case PARENT_TRACE_SEGMENT_ID:
builder.setParentTraceSegmentId(uniqueIdJsonReader.read(reader));
break;
case PARENT_APPLICATION_ID:
case PARENT_APPLICATION_INSTANCE_ID:
builder.setParentApplicationInstanceId(reader.nextInt());
break;
case PARENT_SPAN_ID:

16
apm-collector/apm-collector-agent/agent-jetty/agent-jetty-provider/src/test/resources/json/dubbox-provider.json
View File

@ -32,20 +32,20 @@
"ie": false,
"rs": [
{
"ts": [
"pts": [
230150,
185809,
24040000
],
"ai": -1,
"si": 1,
"vi": 0,
"vn": "/dubbox-case/case/dubbox-rest",
"pii": 1,
"psp": 0,
"psi": 0,
"psn": "/dubbox-case/case/dubbox-rest",
"ni": 0,
"nn": "172.25.0.4:20880",
"ea": 2,
"ei": 0,
"en": "/dubbox-case/case/dubbox-rest",
"eii": 1,
"esi": 0,
"esn": "/dubbox-case/case/dubbox-rest",
"rn": 0
}
],

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

@ -84,12 +84,12 @@ public class AnalysisMetricModuleProvider extends ModuleProvider {
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()));
segmentParserListenerRegister.register(new ServiceReferenceMetricSpanListener.Factory());
segmentParserListenerRegister.register(new ApplicationComponentSpanListener.Factory());
segmentParserListenerRegister.register(new ApplicationMappingSpanListener.Factory());
segmentParserListenerRegister.register(new ServiceEntrySpanListener.Factory());
segmentParserListenerRegister.register(new GlobalTraceSpanListener.Factory());
segmentParserListenerRegister.register(new SegmentCostSpanListener.Factory());
}
private void graphCreate(WorkerCreateListener workerCreateListener) {

28
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

@ -25,8 +25,11 @@ import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decora
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.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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.module.ModuleManager;
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.ApplicationComponent;
@ -45,26 +48,21 @@ public class ApplicationComponentSpanListener implements EntrySpanListener, Exit
@Override
public void parseExit(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
ApplicationComponent applicationComponent = new ApplicationComponent(Const.EMPTY_STRING);
String id = spanDecorator.getPeerId() + Const.ID_SPLIT + String.valueOf(spanDecorator.getComponentId());
ApplicationComponent applicationComponent = new ApplicationComponent(id);
applicationComponent.setComponentId(spanDecorator.getComponentId());
applicationComponent.setPeerId(spanDecorator.getPeerId());
String id = String.valueOf(applicationComponent.getComponentId()) + Const.ID_SPLIT + applicationComponent.getPeerId();
applicationComponent.setId(id);
applicationComponents.add(applicationComponent);
}
@Override
public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId,
String segmentId) {
ApplicationComponent applicationComponent = new ApplicationComponent(Const.EMPTY_STRING);
public void parseEntry(SpanDecorator spanDecorator, int applicationId, int instanceId, String segmentId) {
String id = String.valueOf(applicationId) + Const.ID_SPLIT + String.valueOf(spanDecorator.getComponentId());
ApplicationComponent applicationComponent = new ApplicationComponent(id);
applicationComponent.setComponentId(spanDecorator.getComponentId());
applicationComponent.setPeerId(applicationId);
String id = String.valueOf(applicationComponent.getComponentId()) + Const.ID_SPLIT + String.valueOf(applicationId);
applicationComponent.setId(id);
applicationComponents.add(applicationComponent);
}
@ -83,4 +81,10 @@ public class ApplicationComponentSpanListener implements EntrySpanListener, Exit
graph.start(applicationComponent);
});
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new ApplicationComponentSpanListener();
}
}
}

9
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

@ -24,8 +24,11 @@ import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdD
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.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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.module.ModuleManager;
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.ApplicationMapping;
@ -70,4 +73,10 @@ public class ApplicationMappingSpanListener implements FirstSpanListener, EntryS
graph.start(applicationMapping);
});
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new ApplicationMappingSpanListener();
}
}
}

1
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

@ -63,6 +63,7 @@ public class ApplicationMetricAggregationWorker extends AggregationWorker<Applic
applicationMetric.setSatisfiedCount(applicationReferenceMetric.getSatisfiedCount());
applicationMetric.setToleratingCount(applicationReferenceMetric.getToleratingCount());
applicationMetric.setFrustratedCount(applicationReferenceMetric.getFrustratedCount());
applicationMetric.setTimeBucket(timeBucket);
return applicationMetric;

3
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

@ -82,6 +82,9 @@ public class ApplicationReferenceMetricAggregationWorker extends AggregationWork
} else {
applicationReferenceMetric.setFrustratedCount(1L);
}
applicationReferenceMetric.setTimeBucket(instanceReferenceMetric.getTimeBucket());
return applicationReferenceMetric;
}

9
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

@ -24,8 +24,11 @@ import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdD
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.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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.module.ModuleManager;
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.global.GlobalTrace;
@ -75,4 +78,10 @@ public class GlobalTraceSpanListener implements FirstSpanListener, GlobalTraceId
graph.start(globalTrace);
}
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new GlobalTraceSpanListener();
}
}
}

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

@ -60,6 +60,7 @@ public class InstanceMetricAggregationWorker extends AggregationWorker<InstanceR
instanceMetric.setMqTransactionDurationSum(instanceReferenceMetric.getMqTransactionDurationSum());
instanceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum());
instanceMetric.setTimeBucket(instanceReferenceMetric.getTimeBucket());
return instanceMetric;
}

1
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

@ -62,6 +62,7 @@ public class InstanceReferenceMetricAggregationWorker extends AggregationWorker<
instanceReferenceMetric.setMqTransactionErrorDurationSum(instanceReferenceMetric.getMqTransactionErrorDurationSum());
instanceReferenceMetric.setSourceValue(serviceReferenceMetric.getSourceValue());
instanceReferenceMetric.setTimeBucket(serviceReferenceMetric.getTimeBucket());
return instanceReferenceMetric;
}

8
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

@ -26,6 +26,8 @@ import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listen
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.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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;
@ -102,4 +104,10 @@ public class SegmentCostSpanListener implements EntrySpanListener, ExitSpanListe
graph.start(segmentCost);
}
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new SegmentCostSpanListener(moduleManager);
}
}
}

8
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

@ -22,6 +22,8 @@ import org.apache.skywalking.apm.collector.analysis.metric.define.graph.GraphIdD
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.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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;
@ -85,4 +87,10 @@ public class ServiceEntrySpanListener implements FirstSpanListener, EntrySpanLis
graph.start(serviceEntry);
}
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new ServiceEntrySpanListener(moduleManager);
}
}
}

13
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

@ -27,8 +27,11 @@ import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decora
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.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
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.module.ModuleManager;
import org.apache.skywalking.apm.collector.core.util.Const;
import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.collector.core.util.TimeBucketUtils;
@ -79,7 +82,7 @@ public class ServiceReferenceMetricSpanListener implements FirstSpanListener, En
serviceReferenceMetric.setFrontServiceId(Const.NONE_SERVICE_ID);
serviceReferenceMetric.setFrontInstanceId(instanceId);
serviceReferenceMetric.setBehindServiceId(spanDecorator.getOperationNameId());
serviceReferenceMetric.setBehindServiceId(instanceId);
serviceReferenceMetric.setBehindInstanceId(instanceId);
serviceReferenceMetric.setSourceValue(MetricSource.Entry.ordinal());
calculateCost(serviceReferenceMetric, spanDecorator, false);
@ -100,7 +103,7 @@ public class ServiceReferenceMetricSpanListener implements FirstSpanListener, En
private void calculateCost(ServiceReferenceMetric serviceReferenceMetric, SpanDecorator spanDecorator,
boolean hasReference) {
long duration = spanDecorator.getStartTime() - spanDecorator.getEndTime();
long duration = spanDecorator.getEndTime() - spanDecorator.getStartTime();
if (spanDecorator.getIsError()) {
serviceReferenceMetric.setTransactionErrorCalls(1L);
@ -161,4 +164,10 @@ public class ServiceReferenceMetricSpanListener implements FirstSpanListener, En
graph.start(serviceReferenceMetric);
});
}
public static class Factory implements SpanListenerFactory {
@Override public SpanListener create(ModuleManager moduleManager) {
return new ServiceReferenceMetricSpanListener();
}
}
}

15
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-define/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/define/decorator/SegmentDecorator.java
View File

@ -16,9 +16,9 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator;
import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.network.proto.TraceSegmentObject;
import org.apache.skywalking.apm.network.proto.UniqueId;
@ -29,9 +29,11 @@ public class SegmentDecorator implements StandardBuilder {
private boolean isOrigin = true;
private final TraceSegmentObject segmentObject;
private TraceSegmentObject.Builder segmentBuilder;
private final SpanDecorator[] spanDecorators;
public SegmentDecorator(TraceSegmentObject segmentObject) {
this.segmentObject = segmentObject;
this.spanDecorators = new SpanDecorator[segmentObject.getSpansCount()];
}
public int getApplicationId() {
@ -51,11 +53,14 @@ public class SegmentDecorator implements StandardBuilder {
}
public SpanDecorator getSpans(int index) {
if (isOrigin) {
return new SpanDecorator(segmentObject.getSpans(index), this);
} else {
return new SpanDecorator(segmentBuilder.getSpansBuilder(index), this);
if (ObjectUtils.isEmpty(spanDecorators[index])) {
if (isOrigin) {
spanDecorators[index] = new SpanDecorator(segmentObject.getSpans(index), this);
} else {
spanDecorators[index] = new SpanDecorator(segmentBuilder.getSpansBuilder(index), this);
}
}
return spanDecorators[index];
}
public byte[] toByteArray() {

16
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-define/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/define/decorator/SpanDecorator.java
View File

@ -16,9 +16,9 @@
*
*/
package org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator;
import org.apache.skywalking.apm.collector.core.util.ObjectUtils;
import org.apache.skywalking.apm.network.proto.SpanLayer;
import org.apache.skywalking.apm.network.proto.SpanObject;
import org.apache.skywalking.apm.network.proto.SpanType;
@ -31,16 +31,19 @@ public class SpanDecorator implements StandardBuilder {
private StandardBuilder standardBuilder;
private SpanObject spanObject;
private SpanObject.Builder spanBuilder;
private final ReferenceDecorator[] referenceDecorators;
public SpanDecorator(SpanObject spanObject, StandardBuilder standardBuilder) {
this.spanObject = spanObject;
this.standardBuilder = standardBuilder;
this.referenceDecorators = new ReferenceDecorator[spanObject.getRefsCount()];
}
public SpanDecorator(SpanObject.Builder spanBuilder, StandardBuilder standardBuilder) {
this.spanBuilder = spanBuilder;
this.standardBuilder = standardBuilder;
this.isOrigin = false;
this.referenceDecorators = new ReferenceDecorator[spanBuilder.getRefsCount()];
}
public int getSpanId() {
@ -200,11 +203,14 @@ public class SpanDecorator implements StandardBuilder {
}
public ReferenceDecorator getRefs(int index) {
if (isOrigin) {
return new ReferenceDecorator(spanObject.getRefs(index), this);
} else {
return new ReferenceDecorator(spanBuilder.getRefsBuilder(index), this);
if (ObjectUtils.isEmpty(referenceDecorators[index])) {
if (isOrigin) {
referenceDecorators[index] = new ReferenceDecorator(spanObject.getRefs(index), this);
} else {
referenceDecorators[index] = new ReferenceDecorator(spanBuilder.getRefsBuilder(index), this);
}
}
return referenceDecorators[index];
}
@Override public void toBuilder() {

28
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-define/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/define/listener/SpanListenerFactory.java Normal file
View File

@ -0,0 +1,28 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
package org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener;
import org.apache.skywalking.apm.collector.core.module.ModuleManager;
/**
* @author peng-yongsheng
*/
public interface SpanListenerFactory {
SpanListener create(ModuleManager moduleManager);
}

4
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-define/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/define/service/ISegmentParserListenerRegister.java
View File

@ -18,12 +18,12 @@
package org.apache.skywalking.apm.collector.analysis.segment.parser.define.service;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
import org.apache.skywalking.apm.collector.core.module.Service;
/**
* @author peng-yongsheng
*/
public interface ISegmentParserListenerRegister extends Service {
void register(SpanListener spanListener);
void register(SpanListenerFactory spanListenerFactory);
}

21
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParse.java
View File

@ -19,6 +19,7 @@
package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser;
import com.google.protobuf.InvalidProtocolBufferException;
import java.util.LinkedList;
import java.util.List;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.ReferenceDecorator;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.decorator.SegmentDecorator;
@ -55,6 +56,7 @@ public class SegmentParse {
private final Logger logger = LoggerFactory.getLogger(SegmentParse.class);
private final ModuleManager moduleManager;
private List<SpanListener> spanListeners;
private final SegmentParserListenerManager listenerManager;
private String segmentId;
private long timeBucket = 0;
@ -62,9 +64,12 @@ public class SegmentParse {
public SegmentParse(ModuleManager moduleManager, SegmentParserListenerManager listenerManager) {
this.moduleManager = moduleManager;
this.listenerManager = listenerManager;
this.spanListeners = new LinkedList<>();
}
public boolean parse(UpstreamSegment segment, ISegmentParseService.Source source) {
createSpanListeners();
try {
List<UniqueId> traceIds = segment.getGlobalTraceIdsList();
TraceSegmentObject segmentObject = TraceSegmentObject.parseFrom(segment.getSegment());
@ -173,12 +178,12 @@ public class SegmentParse {
}
private void notifyListenerToBuild() {
listenerManager.getSpanListeners().forEach(SpanListener::build);
spanListeners.forEach(SpanListener::build);
}
private void notifyExitListener(SpanDecorator spanDecorator, int applicationId, int applicationInstanceId,
String segmentId) {
for (SpanListener listener : listenerManager.getSpanListeners()) {
for (SpanListener listener : spanListeners) {
if (listener instanceof ExitSpanListener) {
((ExitSpanListener)listener).parseExit(spanDecorator, applicationId, applicationInstanceId, segmentId);
}
@ -187,7 +192,7 @@ public class SegmentParse {
private void notifyEntryListener(SpanDecorator spanDecorator, int applicationId, int applicationInstanceId,
String segmentId) {
for (SpanListener listener : listenerManager.getSpanListeners()) {
for (SpanListener listener : spanListeners) {
if (listener instanceof EntrySpanListener) {
((EntrySpanListener)listener).parseEntry(spanDecorator, applicationId, applicationInstanceId, segmentId);
}
@ -196,7 +201,7 @@ public class SegmentParse {
private void notifyLocalListener(SpanDecorator spanDecorator, int applicationId, int applicationInstanceId,
String segmentId) {
for (SpanListener listener : listenerManager.getSpanListeners()) {
for (SpanListener listener : spanListeners) {
if (listener instanceof LocalSpanListener) {
((LocalSpanListener)listener).parseLocal(spanDecorator, applicationId, applicationInstanceId, segmentId);
}
@ -205,7 +210,7 @@ public class SegmentParse {
private void notifyFirstListener(SpanDecorator spanDecorator, int applicationId, int applicationInstanceId,
String segmentId) {
for (SpanListener listener : listenerManager.getSpanListeners()) {
for (SpanListener listener : spanListeners) {
if (listener instanceof FirstSpanListener) {
((FirstSpanListener)listener).parseFirst(spanDecorator, applicationId, applicationInstanceId, segmentId);
}
@ -213,10 +218,14 @@ public class SegmentParse {
}
private void notifyGlobalsListener(UniqueId uniqueId) {
for (SpanListener listener : listenerManager.getSpanListeners()) {
for (SpanListener listener : spanListeners) {
if (listener instanceof GlobalTraceIdsListener) {
((GlobalTraceIdsListener)listener).parseGlobalTraceId(uniqueId);
}
}
}
private void createSpanListeners() {
listenerManager.getSpanListenerFactories().forEach(spanListenerFactory -> spanListeners.add(spanListenerFactory.create(moduleManager)));
}
}

14
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/parser/SegmentParserListenerManager.java
View File

@ -20,24 +20,24 @@ package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.par
import java.util.ArrayList;
import java.util.List;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
/**
* @author peng-yongsheng
*/
public class SegmentParserListenerManager {
private List<SpanListener> spanListeners;
private List<SpanListenerFactory> spanListenerFactories;
public SegmentParserListenerManager() {
this.spanListeners = new ArrayList<>();
this.spanListenerFactories = new ArrayList<>();
}
public void add(SpanListener spanListener) {
spanListeners.add(spanListener);
public void add(SpanListenerFactory spanListenerFactory) {
spanListenerFactories.add(spanListenerFactory);
}
public List<SpanListener> getSpanListeners() {
return spanListeners;
public List<SpanListenerFactory> getSpanListenerFactories() {
return spanListenerFactories;
}
}

6
apm-collector/apm-collector-analysis/analysis-segment-parser/segment-parser-provider/src/main/java/org/apache/skywalking/apm/collector/analysis/segment/parser/provider/service/SegmentParserListenerRegister.java
View File

@ -18,7 +18,7 @@
package org.apache.skywalking.apm.collector.analysis.segment.parser.provider.service;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListener;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.listener.SpanListenerFactory;
import org.apache.skywalking.apm.collector.analysis.segment.parser.define.service.ISegmentParserListenerRegister;
import org.apache.skywalking.apm.collector.analysis.segment.parser.provider.parser.SegmentParserListenerManager;
@ -33,7 +33,7 @@ public class SegmentParserListenerRegister implements ISegmentParserListenerRegi
this.listenerManager = listenerManager;
}
@Override public void register(SpanListener spanListener) {
this.listenerManager.add(spanListener);
@Override public void register(SpanListenerFactory spanListenerFactory) {
this.listenerManager.add(spanListenerFactory);
}
}