Support cross-thread propagation.

apm-network/src/main/proto/TraceSegmentService.proto

@@ -20,7 +20,7 @@ message TraceSegmentObject {
     string traceSegmentId = 1;
     repeated TraceSegmentReference refs = 2;
     repeated SpanObject spans = 3;
-    int32 applicationInstanceId = 4;
+    int32 applicationId = 4;
     int32 applicationInstanceId = 5;
 }
 
@@ -28,7 +28,7 @@ message TraceSegmentReference {
     RefType refType = 1;
     string parentTraceSegmentId = 2;
     int32 parentSpanId = 3;
-    int32 parentApplicationId = 4;
+    int32 parentApplicationInstanceId = 4;
     string networkAddress = 5;
     int32 networkAddressId = 6;
     string entryServiceName = 7;

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/AbstractTracerContext.java

@@ -12,6 +12,10 @@ public interface AbstractTracerContext {
 
     void extract(ContextCarrier carrier);
 
+    ContextSnapshot capture();
+
+    void continued(ContextSnapshot snapshot);
+
     String getGlobalTraceId();
 
     AbstractSpan createEntrySpan(String operationName);
@@ -23,6 +27,4 @@ public interface AbstractTracerContext {
     AbstractSpan activeSpan();
 
     void stopSpan(AbstractSpan span);
-
-    void dispose();
 }

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextCarrier.java

@@ -23,7 +23,7 @@ public class ContextCarrier implements Serializable {
 
     private int spanId = -1;
 
-    private int applicationId = DictionaryUtil.nullValue();
+    private int applicationInstanceId = DictionaryUtil.nullValue();
 
     private String peerHost;
 
@@ -45,7 +45,7 @@ public class ContextCarrier implements Serializable {
             return StringUtil.join('|',
                 this.getTraceSegmentId(),
                 this.getSpanId() + "",
-                this.getApplicationId() + "",
+                this.getApplicationInstanceId() + "",
                 this.getPeerHost(),
                 this.getEntryOperationName(),
                 this.serializeDistributedTraceIds());
@@ -66,7 +66,7 @@ public class ContextCarrier implements Serializable {
                 try {
                     this.traceSegmentId = parts[0];
                     this.spanId = Integer.parseInt(parts[1]);
-                    this.applicationId = Integer.parseInt(parts[2]);
+                    this.applicationInstanceId = Integer.parseInt(parts[2]);
                     this.peerHost = parts[3];
                     this.entryOperationName = parts[4];
                     this.distributedTraceIds = deserializeDistributedTraceIds(parts[5]);
@@ -86,7 +86,7 @@ public class ContextCarrier implements Serializable {
     public boolean isValid() {
         return !StringUtil.isEmpty(traceSegmentId)
             && getSpanId() > -1
-            && applicationId != DictionaryUtil.nullValue()
+            && applicationInstanceId != DictionaryUtil.nullValue()
             && !StringUtil.isEmpty(peerHost)
             && !StringUtil.isEmpty(entryOperationName)
             && distributedTraceIds != null;
@@ -120,12 +120,12 @@ public class ContextCarrier implements Serializable {
         this.spanId = spanId;
     }
 
-    public int getApplicationId() {
-        return applicationId;
+    public int getApplicationInstanceId() {
+        return applicationInstanceId;
     }
 
-    void setApplicationId(int applicationId) {
-        this.applicationId = applicationId;
+    void setApplicationInstanceId(int applicationInstanceId) {
+        this.applicationInstanceId = applicationInstanceId;
     }
 
     public String getPeerHost() {

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextManager.java

@@ -106,6 +106,17 @@ public class ContextManager implements TracingContextListener, BootService, Igno
         return span;
     }
 
+    public ContextSnapshot capture() {
+        return get().capture();
+    }
+
+    public void continued(ContextSnapshot snapshot) {
+        if (snapshot == null) {
+            throw new IllegalArgumentException("ContextSnapshot can't be null.");
+        }
+        get().continued(snapshot);
+    }
+
     public static AbstractSpan activeSpan() {
         return get().activeSpan();
     }

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/ContextSnapshot.java

+package org.skywalking.apm.agent.core.context;
+
+import java.util.List;
+import org.skywalking.apm.agent.core.context.ids.DistributedTraceId;
+
+/**
+ * The <code>ContextSnapshot</code> is a snapshot for current context.
+ *
+ * @author wusheng
+ */
+public class ContextSnapshot {
+    private String traceSegmentId;
+
+    private int spanId = -1;
+
+    /**
+     * {@link DistributedTraceId}
+     */
+    private List<DistributedTraceId> distributedTraceIds;
+
+    ContextSnapshot(String traceSegmentId, int spanId,
+        List<DistributedTraceId> distributedTraceIds) {
+        this.traceSegmentId = traceSegmentId;
+        this.spanId = spanId;
+        this.distributedTraceIds = distributedTraceIds;
+    }
+
+    public List<DistributedTraceId> getDistributedTraceIds() {
+        return distributedTraceIds;
+    }
+
+    public String getTraceSegmentId() {
+        return traceSegmentId;
+    }
+
+    public int getSpanId() {
+        return spanId;
+    }
+}

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/IgnoredTracerContext.java

@@ -31,6 +31,14 @@ public class IgnoredTracerContext implements AbstractTracerContext {
 
     }
 
+    @Override public ContextSnapshot capture() {
+        return new ContextSnapshot(null, -1, null);
+    }
+
+    @Override public void continued(ContextSnapshot snapshot) {
+
+    }
+
     @Override
     public String getGlobalTraceId() {
         return "[Ignored Trace]";
@@ -67,11 +75,6 @@ public class IgnoredTracerContext implements AbstractTracerContext {
         }
     }
 
-    @Override
-    public void dispose() {
-
-    }
-
     public static class ListenerManager {
         private static List<IgnoreTracerContextListener> LISTENERS = new LinkedList<IgnoreTracerContextListener>();
 

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/TracingContext.java

@@ -53,7 +53,7 @@ public class TracingContext implements AbstractTracerContext {
         carrier.setTraceSegmentId(this.segment.getTraceSegmentId());
         carrier.setSpanId(span.getSpanId());
 
-        carrier.setApplicationId(segment.getApplicationId());
+        carrier.setApplicationInstanceId(segment.getApplicationId());
 
         if (DictionaryUtil.isNull(exitSpan.getPeerId())) {
             carrier.setPeerHost(exitSpan.getPeer());
@@ -87,6 +87,20 @@ public class TracingContext implements AbstractTracerContext {
         this.segment.relatedGlobalTraces(carrier.getDistributedTraceIds());
     }
 
+    @Override
+    public ContextSnapshot capture() {
+        return new ContextSnapshot(segment.getTraceSegmentId(),
+            activeSpan().getSpanId(),
+            segment.getRelatedGlobalTraces()
+        );
+    }
+
+    @Override
+    public void continued(ContextSnapshot snapshot) {
+        this.segment.ref(new TraceSegmentRef(snapshot));
+        this.segment.relatedGlobalTraces(snapshot.getDistributedTraceIds());
+    }
+
     @Override
     public String getGlobalTraceId() {
         return segment.getRelatedGlobalTraces().get(0).get();
@@ -236,12 +250,6 @@ public class TracingContext implements AbstractTracerContext {
         TracingContext.ListenerManager.notifyFinish(finishedSegment);
     }
 
-    @Override
-    public void dispose() {
-        this.segment = null;
-        this.activeSpanStack = null;
-    }
-
     public static class ListenerManager {
         private static List<TracingContextListener> LISTENERS = new LinkedList<TracingContextListener>();
 

apm-sniffer/apm-agent-core/src/main/java/org/skywalking/apm/agent/core/context/trace/TraceSegmentRef.java

 package org.skywalking.apm.agent.core.context.trace;
 
 import org.skywalking.apm.agent.core.context.ContextCarrier;
+import org.skywalking.apm.agent.core.context.ContextSnapshot;
 import org.skywalking.apm.agent.core.dictionary.DictionaryUtil;
+import org.skywalking.apm.network.proto.RefType;
 import org.skywalking.apm.network.proto.TraceSegmentReference;
 
 /**
@@ -11,11 +13,13 @@ import org.skywalking.apm.network.proto.TraceSegmentReference;
  * Created by wusheng on 2017/2/17.
  */
 public class TraceSegmentRef {
+    private SegmentRefType type;
+
     private String traceSegmentId;
 
     private int spanId = -1;
 
-    private int applicationId;
+    private int applicationInstanceId;
 
     private String peerHost;
 
@@ -31,9 +35,10 @@ public class TraceSegmentRef {
      * @param carrier the valid cross-process propagation format.
      */
     public TraceSegmentRef(ContextCarrier carrier) {
+        this.type = SegmentRefType.CROSS_PROCESS;
         this.traceSegmentId = carrier.getTraceSegmentId();
         this.spanId = carrier.getSpanId();
-        this.applicationId = carrier.getApplicationId();
+        this.applicationInstanceId = carrier.getApplicationInstanceId();
         String host = carrier.getPeerHost();
         if (host.charAt(0) == '#') {
             this.peerHost = host.substring(1);
@@ -48,6 +53,12 @@ public class TraceSegmentRef {
         }
     }
 
+    public TraceSegmentRef(ContextSnapshot snapshot) {
+        this.type = SegmentRefType.CROSS_THREAD;
+        this.traceSegmentId = snapshot.getTraceSegmentId();
+        this.spanId = snapshot.getSpanId();
+    }
+
     public String getOperationName() {
         return operationName;
     }
@@ -58,19 +69,25 @@ public class TraceSegmentRef {
 
     public TraceSegmentReference transform() {
         TraceSegmentReference.Builder refBuilder = TraceSegmentReference.newBuilder();
+        if(SegmentRefType.CROSS_PROCESS.equals(type)) {
+            refBuilder.setRefType(RefType.CrossProcess);
+            refBuilder.setParentApplicationInstanceId(applicationInstanceId);
+            if (peerId == DictionaryUtil.nullValue()) {
+                refBuilder.setNetworkAddress(peerHost);
+            } else {
+                refBuilder.setNetworkAddressId(peerId);
+            }
+            if (operationId == DictionaryUtil.nullValue()) {
+                refBuilder.setEntryServiceName(operationName);
+            } else {
+                refBuilder.setEntryServiceId(operationId);
+            }
+        }else{
+            refBuilder.setRefType(RefType.CrossThread);
+        }
         refBuilder.setParentTraceSegmentId(traceSegmentId);
         refBuilder.setParentSpanId(spanId);
-        refBuilder.setParentApplicationId(applicationId);
-        if (peerId == DictionaryUtil.nullValue()) {
-            refBuilder.setNetworkAddress(peerHost);
-        } else {
-            refBuilder.setNetworkAddressId(peerId);
-        }
-        if (operationId == DictionaryUtil.nullValue()) {
-            refBuilder.setEntryServiceName(operationName);
-        } else {
-            refBuilder.setEntryServiceId(operationId);
-        }
+
         return refBuilder.build();
     }
 
@@ -94,4 +111,9 @@ public class TraceSegmentRef {
         result = 31 * result + spanId;
         return result;
     }
+
+    public enum SegmentRefType {
+        CROSS_PROCESS,
+        CROSS_THREAD
+    }
 }