diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceSpanListener.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceSpanListener.java index 94e408251..cf967c431 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceSpanListener.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/noderef/NodeReferenceSpanListener.java @@ -1,13 +1,13 @@ package org.skywalking.apm.collector.agentstream.worker.noderef; -import java.util.ArrayList; +import java.util.LinkedList; import java.util.List; import org.skywalking.apm.collector.agentstream.worker.cache.InstanceCache; import org.skywalking.apm.collector.agentstream.worker.segment.EntrySpanListener; import org.skywalking.apm.collector.agentstream.worker.segment.ExitSpanListener; -import org.skywalking.apm.collector.agentstream.worker.segment.FirstSpanListener; import org.skywalking.apm.collector.agentstream.worker.segment.RefsListener; import org.skywalking.apm.collector.core.framework.CollectorContextHelper; +import org.skywalking.apm.collector.core.util.CollectionUtils; import org.skywalking.apm.collector.core.util.Const; import org.skywalking.apm.collector.core.util.TimeBucketUtils; import org.skywalking.apm.collector.storage.define.noderef.NodeReferenceDataDefine; @@ -23,73 +23,57 @@ import org.slf4j.LoggerFactory; /** * @author pengys5 */ -public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanListener, FirstSpanListener, RefsListener { +public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanListener, RefsListener { private final Logger logger = LoggerFactory.getLogger(NodeReferenceSpanListener.class); - private List nodeExitReferences = new ArrayList<>(); - private List nodeEntryReferences = new ArrayList<>(); - private List nodeReferences = new ArrayList<>(); - private long timeBucket; - private boolean hasReference = false; - private long startTime; - private long endTime; - private boolean isError; + private List nodeReferences = new LinkedList<>(); + private List references = new LinkedList<>(); @Override public void parseExit(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { - NodeReferenceDataDefine.NodeReference referenceSum = new NodeReferenceDataDefine.NodeReference(); - referenceSum.setFrontApplicationId(applicationId); - referenceSum.setBehindApplicationId(spanObject.getPeerId()); + NodeReferenceDataDefine.NodeReference nodeReference = new NodeReferenceDataDefine.NodeReference(); + nodeReference.setFrontApplicationId(applicationId); + nodeReference.setBehindApplicationId(spanObject.getPeerId()); + nodeReference.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanObject.getStartTime())); - String id = String.valueOf(applicationId); + StringBuilder idBuilder = new StringBuilder(); + idBuilder.append(nodeReference.getTimeBucket()).append(Const.ID_SPLIT).append(applicationId); if (spanObject.getPeerId() != 0) { - referenceSum.setBehindPeer(Const.EMPTY_STRING); - id = id + Const.ID_SPLIT + String.valueOf(spanObject.getPeerId()); + nodeReference.setBehindPeer(Const.EMPTY_STRING); + idBuilder.append(Const.ID_SPLIT).append(spanObject.getPeerId()); } else { - referenceSum.setBehindPeer(spanObject.getPeer()); - id = id + Const.ID_SPLIT + spanObject.getPeer(); + nodeReference.setBehindPeer(spanObject.getPeer()); + idBuilder.append(Const.ID_SPLIT).append(spanObject.getPeer()); } - referenceSum.setId(id); - nodeExitReferences.add(buildNodeRefSum(referenceSum, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError())); + nodeReference.setId(idBuilder.toString()); + nodeReferences.add(buildNodeRefSum(nodeReference, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError())); } @Override public void parseEntry(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { - NodeReferenceDataDefine.NodeReference referenceSum = new NodeReferenceDataDefine.NodeReference(); - referenceSum.setFrontApplicationId(Const.USER_ID); - referenceSum.setBehindApplicationId(applicationId); - referenceSum.setBehindPeer(Const.EMPTY_STRING); + if (CollectionUtils.isNotEmpty(references)) { + references.forEach(nodeReference -> { + nodeReference.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanObject.getStartTime())); + String idBuilder = String.valueOf(nodeReference.getTimeBucket()) + Const.ID_SPLIT + nodeReference.getFrontApplicationId() + + Const.ID_SPLIT + nodeReference.getBehindApplicationId(); - String id = String.valueOf(Const.USER_ID) + Const.ID_SPLIT + String.valueOf(applicationId); - referenceSum.setId(id); - nodeEntryReferences.add(buildNodeRefSum(referenceSum, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError())); - } - - private NodeReferenceDataDefine.NodeReference buildNodeRefSum(NodeReferenceDataDefine.NodeReference referenceSum, - long startTime, long endTime, boolean isError) { - long cost = endTime - startTime; - if (cost <= 1000 && !isError) { - referenceSum.setS1LTE(1); - } else if (1000 < cost && cost <= 3000 && !isError) { - referenceSum.setS3LTE(1); - } else if (3000 < cost && cost <= 5000 && !isError) { - referenceSum.setS5LTE(1); - } else if (5000 < cost && !isError) { - referenceSum.setS5GT(1); + nodeReference.setId(idBuilder); + nodeReferences.add(buildNodeRefSum(nodeReference, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError())); + }); } else { - referenceSum.setError(1); - } - referenceSum.setSummary(1); - return referenceSum; - } + NodeReferenceDataDefine.NodeReference nodeReference = new NodeReferenceDataDefine.NodeReference(); + nodeReference.setFrontApplicationId(Const.USER_ID); + nodeReference.setBehindApplicationId(applicationId); + nodeReference.setBehindPeer(Const.EMPTY_STRING); + nodeReference.setTimeBucket(TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanObject.getStartTime())); - @Override - public void parseFirst(SpanObject spanObject, int applicationId, int applicationInstanceId, String segmentId) { - timeBucket = TimeBucketUtils.INSTANCE.getMinuteTimeBucket(spanObject.getStartTime()); - startTime = spanObject.getStartTime(); - endTime = spanObject.getEndTime(); - isError = spanObject.getIsError(); + String idBuilder = String.valueOf(nodeReference.getTimeBucket()) + Const.ID_SPLIT + nodeReference.getFrontApplicationId() + + Const.ID_SPLIT + nodeReference.getBehindApplicationId(); + + nodeReference.setId(idBuilder); + nodeReferences.add(buildNodeRefSum(nodeReference, spanObject.getStartTime(), spanObject.getEndTime(), spanObject.getIsError())); + } } @Override public void parseRef(TraceSegmentReference reference, int applicationId, int applicationInstanceId, @@ -100,29 +84,14 @@ public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanLis referenceSum.setFrontApplicationId(parentApplicationId); referenceSum.setBehindApplicationId(applicationId); referenceSum.setBehindPeer(Const.EMPTY_STRING); - - String id = String.valueOf(parentApplicationId) + Const.ID_SPLIT + String.valueOf(applicationId); - referenceSum.setId(id); - - hasReference = true; - nodeReferences.add(referenceSum); + references.add(referenceSum); } @Override public void build() { logger.debug("node reference summary listener build"); StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); - if (!hasReference) { - nodeExitReferences.addAll(nodeEntryReferences); - } else { - nodeReferences.forEach(referenceSum -> { - nodeExitReferences.add(buildNodeRefSum(referenceSum, startTime, endTime, isError)); - }); - } - - for (NodeReferenceDataDefine.NodeReference referenceSum : nodeExitReferences) { - referenceSum.setId(timeBucket + Const.ID_SPLIT + referenceSum.getId()); - referenceSum.setTimeBucket(timeBucket); + for (NodeReferenceDataDefine.NodeReference referenceSum : nodeReferences) { try { logger.debug("send to node reference summary aggregation worker, id: {}", referenceSum.getId()); context.getClusterWorkerContext().lookup(NodeReferenceAggregationWorker.WorkerRole.INSTANCE).tell(referenceSum.toData()); @@ -131,4 +100,22 @@ public class NodeReferenceSpanListener implements EntrySpanListener, ExitSpanLis } } } + + private NodeReferenceDataDefine.NodeReference buildNodeRefSum(NodeReferenceDataDefine.NodeReference reference, + long startTime, long endTime, boolean isError) { + long cost = endTime - startTime; + if (cost <= 1000 && !isError) { + reference.setS1LTE(1); + } else if (1000 < cost && cost <= 3000 && !isError) { + reference.setS3LTE(1); + } else if (3000 < cost && cost <= 5000 && !isError) { + reference.setS5LTE(1); + } else if (5000 < cost && !isError) { + reference.setS5GT(1); + } else { + reference.setError(1); + } + reference.setSummary(1); + return reference; + } } diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java index 9f60ca953..38815207d 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/segment/SegmentParse.java @@ -49,9 +49,7 @@ public class SegmentParse { public void parse(List traceIds, TraceSegmentObject segmentObject) { StringBuilder segmentIdBuilder = new StringBuilder(); - segmentObject.getTraceSegmentId().getIdPartsList().forEach(part -> { - segmentIdBuilder.append(part); - }); + segmentObject.getTraceSegmentId().getIdPartsList().forEach(segmentIdBuilder::append); String segmentId = segmentIdBuilder.toString(); for (UniqueId uniqueId : traceIds) { @@ -88,7 +86,7 @@ public class SegmentParse { buildSegment(segmentId, segmentObject.toByteArray()); } - public void buildSegment(String id, byte[] dataBinary) { + private void buildSegment(String id, byte[] dataBinary) { StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); SegmentDataDefine.Segment segment = new SegmentDataDefine.Segment(); segment.setId(id); @@ -103,7 +101,7 @@ public class SegmentParse { } private void notifyListenerToBuild() { - spanListeners.forEach(listener -> listener.build()); + spanListeners.forEach(SpanListener::build); } private void notifyExitListener(SpanObject spanObject, int applicationId, int applicationInstanceId, diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/service/entry/ServiceEntrySpanListener.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/service/entry/ServiceEntrySpanListener.java index d0c144f77..0773b00be 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/service/entry/ServiceEntrySpanListener.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/service/entry/ServiceEntrySpanListener.java @@ -58,7 +58,11 @@ public class ServiceEntrySpanListener implements RefsListener, FirstSpanListener StreamModuleContext context = (StreamModuleContext)CollectorContextHelper.INSTANCE.getContext(StreamModuleGroupDefine.GROUP_NAME); if (!hasReference && hasEntry) { ServiceEntryDataDefine.ServiceEntry serviceEntry = new ServiceEntryDataDefine.ServiceEntry(); - serviceEntry.setId(applicationId + Const.ID_SPLIT + entryServiceName); + if (entryServiceId == 0) { + serviceEntry.setId(applicationId + Const.ID_SPLIT + entryServiceName); + } else { + serviceEntry.setId(applicationId + Const.ID_SPLIT + entryServiceId); + } serviceEntry.setApplicationId(applicationId); serviceEntry.setEntryServiceId(entryServiceId); serviceEntry.setEntryServiceName(entryServiceName); diff --git a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/serviceref/ServiceReferenceSpanListener.java b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/serviceref/ServiceReferenceSpanListener.java index 009870de6..65b1e0271 100644 --- a/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/serviceref/ServiceReferenceSpanListener.java +++ b/apm-collector/apm-collector-agentstream/src/main/java/org/skywalking/apm/collector/agentstream/worker/serviceref/ServiceReferenceSpanListener.java @@ -1,6 +1,6 @@ package org.skywalking.apm.collector.agentstream.worker.serviceref; -import java.util.ArrayList; +import java.util.LinkedList; import java.util.List; import org.skywalking.apm.collector.agentstream.worker.cache.InstanceCache; import org.skywalking.apm.collector.agentstream.worker.segment.EntrySpanListener; @@ -26,7 +26,7 @@ public class ServiceReferenceSpanListener implements FirstSpanListener, EntrySpa private final Logger logger = LoggerFactory.getLogger(ServiceReferenceSpanListener.class); - private List referenceServices = new ArrayList<>(); + private List referenceServices = new LinkedList<>(); private int serviceId = 0; private String serviceName = ""; private long startTime = 0; diff --git a/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryHasExitNoRefSpan.java b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryHasExitNoRefSpan.java new file mode 100644 index 000000000..2eee08472 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryHasExitNoRefSpan.java @@ -0,0 +1,115 @@ +package org.skywalking.apm.collector.agentstream.mock.grpc; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import org.skywalking.apm.network.proto.Downstream; +import org.skywalking.apm.network.proto.SpanLayer; +import org.skywalking.apm.network.proto.SpanObject; +import org.skywalking.apm.network.proto.SpanType; +import org.skywalking.apm.network.proto.TraceSegmentObject; +import org.skywalking.apm.network.proto.TraceSegmentServiceGrpc; +import org.skywalking.apm.network.proto.UniqueId; +import org.skywalking.apm.network.proto.UpstreamSegment; +import org.skywalking.apm.network.trace.component.ComponentsDefine; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class SingleHasEntryHasExitNoRefSpan { + + public static void main(String[] args) { + Post post = new Post(); + post.send(); + + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + } + } + + static class Post { + private final Logger logger = LoggerFactory.getLogger(Post.class); + + public void send() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).maxInboundMessageSize(1024 * 1024 * 50).usePlaintext(true).build(); + + int applicationId = 0; + int instanceId = 0; + int entryServiceId = 0; + while (applicationId == 0) { + applicationId = ApplicationRegister.register(channel, "consumer"); + } + + while (instanceId == 0) { + instanceId = InstanceRegister.register(channel, "ConsumerUUID", applicationId, "consumer_host_name", 1); + } + + while (entryServiceId == 0) { + entryServiceId = ServiceRegister.register(channel, applicationId, "/dubbox-case/case/dubbox-rest"); + } + + TraceSegmentServiceGrpc.TraceSegmentServiceStub stub = TraceSegmentServiceGrpc.newStub(channel); + StreamObserver streamObserver = stub.collect(new StreamObserver() { + @Override public void onNext(Downstream downstream) { + } + + @Override public void onError(Throwable throwable) { + logger.error(throwable.getMessage(), throwable); + } + + @Override public void onCompleted() { + } + }); + + long now = System.currentTimeMillis(); + + int id = 1; + UniqueId.Builder builder = UniqueId.newBuilder(); + builder.addIdParts(id); + builder.addIdParts(id); + builder.addIdParts(id); + UniqueId segmentId = builder.build(); + + UpstreamSegment.Builder upstream = UpstreamSegment.newBuilder(); + upstream.addGlobalTraceIds(segmentId); + + TraceSegmentObject.Builder segmentBuilder = TraceSegmentObject.newBuilder(); + segmentBuilder.setApplicationId(applicationId); + segmentBuilder.setApplicationInstanceId(instanceId); + segmentBuilder.setTraceSegmentId(segmentId); + + SpanObject.Builder entrySpan = SpanObject.newBuilder(); + entrySpan.setSpanId(0); + entrySpan.setSpanType(SpanType.Entry); + entrySpan.setSpanLayer(SpanLayer.Http); + entrySpan.setParentSpanId(-1); + entrySpan.setStartTime(now); + entrySpan.setEndTime(now + 3000); + entrySpan.setComponentId(ComponentsDefine.TOMCAT.getId()); + entrySpan.setOperationNameId(entryServiceId); + entrySpan.setIsError(false); + segmentBuilder.addSpans(entrySpan); + + SpanObject.Builder exitSpan = SpanObject.newBuilder(); + exitSpan.setSpanId(1); + exitSpan.setSpanType(SpanType.Exit); + exitSpan.setSpanLayer(SpanLayer.Database); + exitSpan.setParentSpanId(0); + exitSpan.setStartTime(now); + exitSpan.setEndTime(now + 3000); + exitSpan.setComponentId(ComponentsDefine.MONGODB.getId()); + exitSpan.setOperationNameId(entryServiceId); + exitSpan.setIsError(false); + exitSpan.setPeer("localhost:8888"); + segmentBuilder.addSpans(exitSpan); + + upstream.setSegment(segmentBuilder.build().toByteString()); + + streamObserver.onNext(upstream.build()); + streamObserver.onCompleted(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryNoExitHasRefSpan.java b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryNoExitHasRefSpan.java new file mode 100644 index 000000000..56e256a6d --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleHasEntryNoExitHasRefSpan.java @@ -0,0 +1,132 @@ +package org.skywalking.apm.collector.agentstream.mock.grpc; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import org.skywalking.apm.network.proto.Downstream; +import org.skywalking.apm.network.proto.RefType; +import org.skywalking.apm.network.proto.SpanLayer; +import org.skywalking.apm.network.proto.SpanObject; +import org.skywalking.apm.network.proto.SpanType; +import org.skywalking.apm.network.proto.TraceSegmentObject; +import org.skywalking.apm.network.proto.TraceSegmentReference; +import org.skywalking.apm.network.proto.TraceSegmentServiceGrpc; +import org.skywalking.apm.network.proto.UniqueId; +import org.skywalking.apm.network.proto.UpstreamSegment; +import org.skywalking.apm.network.trace.component.ComponentsDefine; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class SingleHasEntryNoExitHasRefSpan { + + public static void main(String[] args) { + Post post = new Post(); + post.send(); + + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + } + } + + static class Post { + private final Logger logger = LoggerFactory.getLogger(Post.class); + + public void send() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).maxInboundMessageSize(1024 * 1024 * 50).usePlaintext(true).build(); + + int consumerApplicationId = 0; + int providerApplicationId = 0; + int consumerInstanceId = 0; + int providerInstanceId = 0; + int consumerEntryServiceId = 0; + int consumerExitServiceId = 0; + int consumerExitApplicationId = 0; + int providerEntryServiceId = 0; + + while (consumerApplicationId == 0) { + consumerApplicationId = ApplicationRegister.register(channel, "consumer"); + } + while (consumerExitApplicationId == 0) { + consumerExitApplicationId = ApplicationRegister.register(channel, "172.25.0.4:20880"); + } + while (providerApplicationId == 0) { + providerApplicationId = ApplicationRegister.register(channel, "provider"); + } + while (consumerInstanceId == 0) { + consumerInstanceId = InstanceRegister.register(channel, "ConsumerUUID", consumerApplicationId, "consumer_host_name", 1); + } + while (providerInstanceId == 0) { + providerInstanceId = InstanceRegister.register(channel, "ProviderUUID", providerApplicationId, "provider_host_name", 2); + } + while (consumerEntryServiceId == 0) { + consumerEntryServiceId = ServiceRegister.register(channel, consumerApplicationId, "/dubbox-case/case/dubbox-rest"); + } + while (consumerExitServiceId == 0) { + consumerExitServiceId = ServiceRegister.register(channel, consumerApplicationId, "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"); + } + while (providerEntryServiceId == 0) { + providerEntryServiceId = ServiceRegister.register(channel, providerApplicationId, "org.skywaking.apm.testcase.dubbo.services.GreetService.doBusiness()"); + } + + TraceSegmentServiceGrpc.TraceSegmentServiceStub stub = TraceSegmentServiceGrpc.newStub(channel); + StreamObserver streamObserver = stub.collect(new StreamObserver() { + @Override public void onNext(Downstream downstream) { + } + + @Override public void onError(Throwable throwable) { + logger.error(throwable.getMessage(), throwable); + } + + @Override public void onCompleted() { + } + }); + + long now = System.currentTimeMillis(); + + int id = 1; + UniqueId.Builder builder = UniqueId.newBuilder(); + builder.addIdParts(id); + builder.addIdParts(id); + builder.addIdParts(id); + UniqueId segmentId = builder.build(); + + UpstreamSegment.Builder upstream = UpstreamSegment.newBuilder(); + upstream.addGlobalTraceIds(segmentId); + + TraceSegmentObject.Builder segmentBuilder = TraceSegmentObject.newBuilder(); + segmentBuilder.setApplicationId(consumerApplicationId); + segmentBuilder.setApplicationInstanceId(consumerInstanceId); + segmentBuilder.setTraceSegmentId(segmentId); + + TraceSegmentReference.Builder referenceBuilder = TraceSegmentReference.newBuilder(); + referenceBuilder.setEntryApplicationInstanceId(providerInstanceId); + referenceBuilder.setEntryServiceName("/rest/test"); + referenceBuilder.setParentApplicationInstanceId(providerInstanceId); + referenceBuilder.setParentServiceName("/rest/test"); + referenceBuilder.setRefType(RefType.CrossProcess); + referenceBuilder.setNetworkAddress("localhost:8080"); + segmentBuilder.addRefs(referenceBuilder); + + SpanObject.Builder entrySpan = SpanObject.newBuilder(); + entrySpan.setSpanId(0); + entrySpan.setSpanType(SpanType.Entry); + entrySpan.setSpanLayer(SpanLayer.Http); + entrySpan.setParentSpanId(-1); + entrySpan.setStartTime(now); + entrySpan.setEndTime(now + 3000); + entrySpan.setComponentId(ComponentsDefine.TOMCAT.getId()); + entrySpan.setOperationNameId(consumerEntryServiceId); + entrySpan.setIsError(false); + segmentBuilder.addSpans(entrySpan); + + upstream.setSegment(segmentBuilder.build().toByteString()); + + streamObserver.onNext(upstream.build()); + streamObserver.onCompleted(); + } + } +} diff --git a/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleNoEntryHasExitNoRefSpan.java b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleNoEntryHasExitNoRefSpan.java new file mode 100644 index 000000000..2d41d7300 --- /dev/null +++ b/apm-collector/apm-collector-agentstream/src/test/java/org/skywalking/apm/collector/agentstream/mock/grpc/SingleNoEntryHasExitNoRefSpan.java @@ -0,0 +1,103 @@ +package org.skywalking.apm.collector.agentstream.mock.grpc; + +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.StreamObserver; +import org.skywalking.apm.network.proto.Downstream; +import org.skywalking.apm.network.proto.SpanLayer; +import org.skywalking.apm.network.proto.SpanObject; +import org.skywalking.apm.network.proto.SpanType; +import org.skywalking.apm.network.proto.TraceSegmentObject; +import org.skywalking.apm.network.proto.TraceSegmentServiceGrpc; +import org.skywalking.apm.network.proto.UniqueId; +import org.skywalking.apm.network.proto.UpstreamSegment; +import org.skywalking.apm.network.trace.component.ComponentsDefine; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * @author pengys5 + */ +public class SingleNoEntryHasExitNoRefSpan { + + public static void main(String[] args) { + Post post = new Post(); + post.send(); + + try { + Thread.sleep(2000); + } catch (InterruptedException e) { + } + } + + static class Post { + private final Logger logger = LoggerFactory.getLogger(Post.class); + + public void send() { + ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 11800).maxInboundMessageSize(1024 * 1024 * 50).usePlaintext(true).build(); + + int applicationId = 0; + int instanceId = 0; + int entryServiceId = 0; + while (applicationId == 0) { + applicationId = ApplicationRegister.register(channel, "consumer"); + } + + while (instanceId == 0) { + instanceId = InstanceRegister.register(channel, "ConsumerUUID", applicationId, "consumer_host_name", 1); + } + + while (entryServiceId == 0) { + entryServiceId = ServiceRegister.register(channel, applicationId, "/dubbox-case/case/dubbox-rest"); + } + + TraceSegmentServiceGrpc.TraceSegmentServiceStub stub = TraceSegmentServiceGrpc.newStub(channel); + StreamObserver streamObserver = stub.collect(new StreamObserver() { + @Override public void onNext(Downstream downstream) { + } + + @Override public void onError(Throwable throwable) { + logger.error(throwable.getMessage(), throwable); + } + + @Override public void onCompleted() { + } + }); + + long now = System.currentTimeMillis(); + + int id = 1; + UniqueId.Builder builder = UniqueId.newBuilder(); + builder.addIdParts(id); + builder.addIdParts(id); + builder.addIdParts(id); + UniqueId segmentId = builder.build(); + + UpstreamSegment.Builder upstream = UpstreamSegment.newBuilder(); + upstream.addGlobalTraceIds(segmentId); + + TraceSegmentObject.Builder segmentBuilder = TraceSegmentObject.newBuilder(); + segmentBuilder.setApplicationId(applicationId); + segmentBuilder.setApplicationInstanceId(instanceId); + segmentBuilder.setTraceSegmentId(segmentId); + + SpanObject.Builder exitSpan = SpanObject.newBuilder(); + exitSpan.setSpanId(0); + exitSpan.setSpanType(SpanType.Exit); + exitSpan.setSpanLayer(SpanLayer.Database); + exitSpan.setParentSpanId(-1); + exitSpan.setStartTime(now); + exitSpan.setEndTime(now + 3000); + exitSpan.setComponentId(ComponentsDefine.MONGODB.getId()); + exitSpan.setOperationNameId(entryServiceId); + exitSpan.setIsError(false); + exitSpan.setPeer("localhost:8888"); + segmentBuilder.addSpans(exitSpan); + + upstream.setSegment(segmentBuilder.build().toByteString()); + + streamObserver.onNext(upstream.build()); + streamObserver.onCompleted(); + } + } +}