Merge pull request apache#631 from OpenSkywalking/feature/new-network-protocol

peng-yongsheng · web-flow · commit 47af075a72f5 · 2017-12-01T16:30:44.000+08:00
Change network protocol and refactor agent core.
diff --git a/apm-network/src/main/proto/TraceSegmentService.proto b/apm-network/src/main/proto/TraceSegmentService.proto
@@ -22,11 +22,10 @@ message UniqueId {
 
 message TraceSegmentObject {
     UniqueId traceSegmentId = 1;
-    repeated TraceSegmentReference refs = 2;
-    repeated SpanObject spans = 3;
-    int32 applicationId = 4;
-    int32 applicationInstanceId = 5;
-    bool isSizeLimited = 6;
+    repeated SpanObject spans = 2;
+    int32 applicationId = 3;
+    int32 applicationInstanceId = 4;
+    bool isSizeLimited = 5;
 }
 
 message TraceSegmentReference {
@@ -48,17 +47,18 @@ message SpanObject {
     int32 parentSpanId = 2;
     int64 startTime = 3;
     int64 endTime = 4;
-    int32 operationNameId = 5;
-    string operationName = 6;
-    int32 peerId = 7;
-    string peer = 8;
-    SpanType spanType = 9;
-    SpanLayer spanLayer = 10;
-    int32 componentId = 11;
-    string component = 12;
-    bool isError = 13;
-    repeated KeyWithStringValue tags = 14;
-    repeated LogMessage logs = 15;
+    repeated TraceSegmentReference refs = 5;
+    int32 operationNameId = 6;
+    string operationName = 7;
+    int32 peerId = 8;
+    string peer = 9;
+    SpanType spanType = 10;
+    SpanLayer spanLayer = 11;
+    int32 componentId = 12;
+    string component = 13;
+    bool isError = 14;
+    repeated KeyWithStringValue tags = 15;
+    repeated LogMessage logs = 16;
 }
 
 enum RefType {
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java
@@ -25,17 +25,15 @@
 import org.skywalking.apm.agent.core.context.trace.AbstractSpan;
 import org.skywalking.apm.agent.core.context.trace.TraceSegment;
 import org.skywalking.apm.agent.core.dictionary.DictionaryUtil;
-import org.skywalking.apm.agent.core.sampling.SamplingService;
 import org.skywalking.apm.agent.core.logging.api.ILog;
 import org.skywalking.apm.agent.core.logging.api.LogManager;
+import org.skywalking.apm.agent.core.sampling.SamplingService;
 import org.skywalking.apm.util.StringUtil;
 
 /**
  * {@link ContextManager} controls the whole context of {@link TraceSegment}. Any {@link TraceSegment} relates to
  * single-thread, so this context use {@link ThreadLocal} to maintain the context, and make sure, since a {@link
- * TraceSegment} starts, all ChildOf spans are in the same context.
- * <p>
- * What is 'ChildOf'? {@see
+ * TraceSegment} starts, all ChildOf spans are in the same context. <p> What is 'ChildOf'? {@see
  * https://github.com/opentracing/specification/blob/master/specification.md#references-between-spans}
  *
  * <p> Also, {@link ContextManager} delegates to all {@link AbstractTracerContext}'s major methods.
@@ -100,15 +98,18 @@ public static String getGlobalTraceId() {
 
     public static AbstractSpan createEntrySpan(String operationName, ContextCarrier carrier) {
         SamplingService samplingService = ServiceManager.INSTANCE.findService(SamplingService.class);
+        AbstractSpan span;
         AbstractTracerContext context;
         if (carrier != null && carrier.isValid()) {
             samplingService.forceSampled();
             context = getOrCreate(operationName, true);
+            span = context.createEntrySpan(operationName);
             context.extract(carrier);
         } else {
             context = getOrCreate(operationName, false);
+            span = context.createEntrySpan(operationName);
         }
-        return context.createEntrySpan(operationName);
+        return span;
     }
 
     public static AbstractSpan createLocalSpan(String operationName) {
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java
@@ -153,8 +153,13 @@ public void inject(ContextCarrier carrier) {
      */
     @Override
     public void extract(ContextCarrier carrier) {
-        this.segment.ref(new TraceSegmentRef(carrier));
+        TraceSegmentRef ref = new TraceSegmentRef(carrier);
+        this.segment.ref(ref);
         this.segment.relatedGlobalTraces(carrier.getDistributedTraceId());
+        AbstractSpan span = this.activeSpan();
+        if (span instanceof EntrySpan) {
+            ((EntrySpan)span).ref(ref);
+        }
     }
 
     /**
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/EntrySpan.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/EntrySpan.java
@@ -18,7 +18,10 @@
 
 package org.skywalking.apm.agent.core.context.trace;
 
+import java.util.LinkedList;
+import java.util.List;
 import org.skywalking.apm.agent.core.dictionary.DictionaryUtil;
+import org.skywalking.apm.network.proto.SpanObject;
 import org.skywalking.apm.network.trace.component.Component;
 
 /**
@@ -29,12 +32,18 @@
  *
  * But with the last <code>EntrySpan</code>'s tags and logs, which have more details about a service provider.
  *
- * Such as: Tomcat Embed -> Dubbox
- * The <code>EntrySpan</code> represents the Dubbox span.
+ * Such as: Tomcat Embed -> Dubbox The <code>EntrySpan</code> represents the Dubbox span.
  *
  * @author wusheng
  */
 public class EntrySpan extends StackBasedTracingSpan {
+    /**
+     * The refs of parent trace segments, except the primary one. For most RPC call, {@link #refs} contains only one
+     * element, but if this segment is a start span of batch process, the segment faces multi parents, at this moment,
+     * we use this {@link #refs} to link them.
+     */
+    private List<TraceSegmentRef> refs;
+
     private int currentMaxDepth;
 
     public EntrySpan(int spanId, int parentSpanId, String operationName) {
@@ -126,6 +135,25 @@ public EntrySpan log(Throwable t) {
         return false;
     }
 
+    @Override public SpanObject.Builder transform() {
+        SpanObject.Builder builder = super.transform();
+        if (refs != null) {
+            for (TraceSegmentRef ref : refs) {
+                builder.addRefs(ref.transform());
+            }
+        }
+        return builder;
+    }
+
+    public void ref(TraceSegmentRef ref) {
+        if (refs == null) {
+            refs = new LinkedList<TraceSegmentRef>();
+        }
+        if (!refs.contains(ref)) {
+            refs.add(ref);
+        }
+    }
+
     private void clearWhenRestart() {
         this.componentId = DictionaryUtil.nullValue();
         this.componentName = null;
diff --git a/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java b/apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegment.java
@@ -46,6 +46,8 @@ public class TraceSegment {
      * The refs of parent trace segments, except the primary one. For most RPC call, {@link #refs} contains only one
      * element, but if this segment is a start span of batch process, the segment faces multi parents, at this moment,
      * we use this {@link #refs} to link them.
+     *
+     * This field will not be serialized. Keeping this field is only for quick accessing.
      */
     private List<TraceSegmentRef> refs;
 
@@ -165,12 +167,8 @@ public UpstreamSegment transform() {
          * Trace Segment
          */
         traceSegmentBuilder.setTraceSegmentId(this.traceSegmentId.transform());
-        // TraceSegmentReference
-        if (this.refs != null) {
-            for (TraceSegmentRef ref : this.refs) {
-                traceSegmentBuilder.addRefs(ref.transform());
-            }
-        }
+        // Don't serialize TraceSegmentReference
+
         // SpanObject
         for (AbstractTracingSpan span : this.spans) {
             traceSegmentBuilder.addSpans(span.transform());
diff --git a/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/context/ContextManagerTest.java b/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/context/ContextManagerTest.java
@@ -255,14 +255,14 @@ public void testTransform() throws InvalidProtocolBufferException {
         UpstreamSegment upstreamSegment = actualSegment.transform();
         assertThat(upstreamSegment.getGlobalTraceIdsCount(), is(1));
         TraceSegmentObject traceSegmentObject = TraceSegmentObject.parseFrom(upstreamSegment.getSegment());
-        TraceSegmentReference reference = traceSegmentObject.getRefs(0);
+        TraceSegmentReference reference = traceSegmentObject.getSpans(1).getRefs(0);
 
         assertThat(reference.getEntryServiceName(), is("/portal/"));
         assertThat(reference.getNetworkAddress(), is("127.0.0.1:8080"));
         assertThat(reference.getParentSpanId(), is(3));
 
         assertThat(traceSegmentObject.getApplicationId(), is(1));
-        assertThat(traceSegmentObject.getRefsCount(), is(1));
+        assertThat(traceSegmentObject.getSpans(1).getRefsCount(), is(1));
 
         assertThat(traceSegmentObject.getSpansCount(), is(2));
 
diff --git a/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClientTest.java b/apm-sniffer/apm-agent-core/src/test/java/org/skywalking/apm/agent/core/remote/TraceSegmentServiceClientTest.java
@@ -127,7 +127,7 @@ public void testSendTraceSegmentWithoutException() throws InvalidProtocolBufferE
         UpstreamSegment upstreamSegment = upstreamSegments.get(0);
         assertThat(upstreamSegment.getGlobalTraceIdsCount(), is(1));
         TraceSegmentObject traceSegmentObject = TraceSegmentObject.parseFrom(upstreamSegment.getSegment());
-        assertThat(traceSegmentObject.getRefsCount(), is(0));
+        assertThat(traceSegmentObject.getSpans(0).getRefsCount(), is(0));
         assertThat(traceSegmentObject.getSpansCount(), is(1));
 
         SpanObject spanObject = traceSegmentObject.getSpans(0);
