Add async span API (#47)

Author: mrproliu

docs/en/development-and-contribution/development-guide.md

@@ -347,6 +347,8 @@ After creating a Span, you can perform additional operations on it.
 ```go
 // Span for plugin API
 type Span interface {
+	// AsyncSpan for the async API
+	AsyncSpan
 	// Tag set the Tag of the Span
 	Tag(Tag, string)
 	// SetSpanLayer set the SpanLayer of the Span
@@ -364,6 +366,28 @@ type Span interface {
 }
 ```
 
+#### Async Span
+
+There is a set of advanced APIs in Span which is specifically designed for async use cases.
+When setting name, tags, logs, and other operations (including end span) of the span in another goroutine, you should use these APIs.
+
+```go
+type AsyncSpan interface {
+    // PrepareAsync the span finished at current tracing context, but current span is still alive until AsyncFinish called
+    PrepareAsync()
+    // AsyncFinish to finished current async span
+    AsyncFinish()
+}
+```
+
+Following the previous API define, you should following these steps to use the async API:
+1. Call `span.PrepareAsync()` to prepare the span to do any operation in another goroutine.
+2. Use `Span.End()` in the original goroutine when your job in the current goroutine is complete.
+3. Propagate the span to any other goroutine in your plugin.
+4. Once the above steps are all set, call `span.AsyncFinish()` in any goroutine.
+5. When the `span.AsyncFinish()` is complete for all spans, the all spans would be finished and report to the backend.
+
+
 ## Import Plugin
 
 Once you have finished developing the plugin, you need to import the completed module into the Agent program and [define it in the corresponding file](../../../tools/go-agent/instrument/plugins/register.go). 

plugins/core/span_default.go

@@ -19,6 +19,7 @@ package core
 
 import (
 	"math"
+	"sync"
 	"time"
 
 	"github.com/apache/skywalking-go/plugins/core/reporter"
@@ -41,6 +42,10 @@ type DefaultSpan struct {
 	IsError       bool
 	SpanType      SpanType
 	Parent        TracingSpan
+
+	InAsyncMode       bool
+	AsyncModeFinished bool
+	AsyncOpLocker     *sync.Mutex
 }
 
 func NewDefaultSpan(tracer *Tracer, parent TracingSpan) *DefaultSpan {
@@ -54,6 +59,10 @@ func NewDefaultSpan(tracer *Tracer, parent TracingSpan) *DefaultSpan {
 
 // For TracingSpan
 func (ds *DefaultSpan) SetOperationName(name string) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	ds.OperationName = name
 }
 
@@ -62,6 +71,10 @@ func (ds *DefaultSpan) GetOperationName() string {
 }
 
 func (ds *DefaultSpan) SetPeer(peer string) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	ds.Peer = peer
 }
 
@@ -70,6 +83,10 @@ func (ds *DefaultSpan) GetPeer() string {
 }
 
 func (ds *DefaultSpan) SetSpanLayer(layer int32) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	ds.Layer = agentv3.SpanLayer(layer)
 }
 
@@ -78,6 +95,10 @@ func (ds *DefaultSpan) GetSpanLayer() agentv3.SpanLayer {
 }
 
 func (ds *DefaultSpan) SetComponent(componentID int32) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	ds.ComponentID = componentID
 }
 
@@ -86,6 +107,10 @@ func (ds *DefaultSpan) GetComponent() int32 {
 }
 
 func (ds *DefaultSpan) Tag(key, value string) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	for _, tag := range ds.Tags {
 		if tag.Key == key {
 			tag.Value = value
@@ -96,6 +121,10 @@ func (ds *DefaultSpan) Tag(key, value string) {
 }
 
 func (ds *DefaultSpan) Log(ll ...string) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	data := make([]*commonv3.KeyStringValuePair, 0, int32(math.Ceil(float64(len(ll))/2.0)))
 	var kvp *commonv3.KeyStringValuePair
 	for i, l := range ll {
@@ -111,14 +140,20 @@ func (ds *DefaultSpan) Log(ll ...string) {
 }
 
 func (ds *DefaultSpan) Error(ll ...string) {
+	if ds.InAsyncMode {
+		ds.AsyncOpLocker.Lock()
+		defer ds.AsyncOpLocker.Unlock()
+	}
 	ds.IsError = true
 	ds.Log(ll...)
 }
 
-func (ds *DefaultSpan) End() {
+func (ds *DefaultSpan) End(changeParent bool) {
 	ds.EndTime = time.Now()
-	if ctx := getTracingContext(); ctx != nil {
-		ctx.SaveActiveSpan(ds.Parent)
+	if changeParent {
+		if ctx := getTracingContext(); ctx != nil {
+			ctx.SaveActiveSpan(ds.Parent)
+		}
 	}
 }
 
@@ -137,3 +172,23 @@ func (ds *DefaultSpan) IsValid() bool {
 func (ds *DefaultSpan) ParentSpan() TracingSpan {
 	return ds.Parent
 }
+
+func (ds *DefaultSpan) PrepareAsync() {
+	if ds.InAsyncMode {
+		panic("already in async mode")
+	}
+	ds.InAsyncMode = true
+	ds.AsyncModeFinished = false
+	ds.AsyncOpLocker = &sync.Mutex{}
+}
+
+func (ds *DefaultSpan) AsyncFinish() {
+	if !ds.InAsyncMode {
+		panic("not in async mode")
+	}
+	if ds.AsyncModeFinished {
+		panic("already finished async")
+	}
+	ds.AsyncModeFinished = true
+	ds.AsyncOpLocker = nil
+}

plugins/core/span_noop.go

@@ -93,3 +93,9 @@ func (*NoopSpan) IsValid() bool {
 func (n *NoopSpan) ParentSpan() TracingSpan {
 	return nil
 }
+
+func (n *NoopSpan) PrepareAsync() {
+}
+
+func (n *NoopSpan) AsyncFinish() {
+}

plugins/core/span_tracing.go

@@ -100,12 +100,23 @@ func (s *SegmentSpanImpl) End() {
 	if !s.IsValid() {
 		return
 	}
-	s.DefaultSpan.End()
-	go func() {
-		s.SegmentContext.collect <- s
-	}()
+	s.DefaultSpan.End(true)
+	s.end0()
 }
 
+func (s *SegmentSpanImpl) AsyncFinish() {
+	s.DefaultSpan.AsyncFinish()
+	s.DefaultSpan.End(false)
+	s.end0()
+}
+
+func (s *SegmentSpanImpl) end0() {
+	if !s.DefaultSpan.InAsyncMode || s.DefaultSpan.AsyncModeFinished {
+		go func() {
+			s.SegmentContext.collect <- s
+		}()
+	}
+}
 func (s *SegmentSpanImpl) GetDefaultSpan() *DefaultSpan {
 	return &s.DefaultSpan
 }
@@ -220,10 +231,22 @@ func (rs *RootSegmentSpan) End() {
 	if !rs.IsValid() {
 		return
 	}
-	rs.DefaultSpan.End()
-	go func() {
-		rs.doneCh <- atomic.SwapInt32(rs.SegmentContext.refNum, -1)
-	}()
+	rs.DefaultSpan.End(true)
+	rs.end0()
+}
+
+func (rs *RootSegmentSpan) AsyncFinish() {
+	rs.DefaultSpan.AsyncFinish()
+	rs.DefaultSpan.End(false)
+	rs.end0()
+}
+
+func (rs *RootSegmentSpan) end0() {
+	if !rs.InAsyncMode || rs.AsyncModeFinished {
+		go func() {
+			rs.doneCh <- atomic.SwapInt32(rs.SegmentContext.refNum, -1)
+		}()
+	}
 }
 
 func (rs *RootSegmentSpan) createRootSegmentContext(ctx *TracingContext, _ SegmentSpan) (err error) {
@@ -295,6 +318,14 @@ func (s *SnapshotSpan) segmentRegister() bool {
 	}
 }
 
+func (s *SnapshotSpan) PrepareAsync() {
+	panic("please use the PrepareAsync on right goroutine")
+}
+
+func (s *SnapshotSpan) AsyncFinish() {
+	panic("please use the AsyncFinish on right goroutine")
+}
+
 func newSegmentRoot(segmentSpan *SegmentSpanImpl) *RootSegmentSpan {
 	s := &RootSegmentSpan{
 		SegmentSpanImpl: segmentSpan,

plugins/core/tracing.go

@@ -27,6 +27,8 @@ import (
 	"github.com/apache/skywalking-go/plugins/core/tracing"
 )
 
+var snapshotType = reflect.TypeOf(&SnapshotSpan{})
+
 func (t *Tracer) Tracing() interface{} {
 	return t
 }
@@ -48,7 +50,7 @@ func (t *Tracer) CreateEntrySpan(operationName string, extractor interface{}, op
 		saveSpanToActiveIfNotError(ctx, s, err)
 	}()
 	// if parent span is entry span, then use parent span as result
-	if tracingSpan != nil && tracingSpan.IsEntry() {
+	if tracingSpan != nil && tracingSpan.IsEntry() && reflect.ValueOf(tracingSpan).Type() != snapshotType {
 		tracingSpan.SetOperationName(operationName)
 		return tracingSpan, nil
 	}
@@ -85,7 +87,7 @@ func (t *Tracer) CreateExitSpan(operationName, peer string, injector interface{}
 	}()
 
 	// if parent span is exit span, then use parent span as result
-	if tracingSpan != nil && tracingSpan.IsExit() {
+	if tracingSpan != nil && tracingSpan.IsExit() && reflect.ValueOf(tracingSpan).Type() != snapshotType {
 		return tracingSpan, nil
 	}
 	span, err := t.createSpan0(ctx, tracingSpan, opts, withSpanType(SpanTypeExit), withOperationName(operationName), withPeer(peer))

plugins/core/tracing/api.go

@@ -188,3 +188,7 @@ func (n *NoopSpan) Error(...string) {
 }
 func (n *NoopSpan) End() {
 }
+func (n *NoopSpan) PrepareAsync() {
+}
+func (n *NoopSpan) AsyncFinish() {
+}

plugins/core/tracing/bridge.go

@@ -17,8 +17,17 @@
 
 package tracing
 
+type AsyncSpan interface {
+	// PrepareAsync the span finished at current tracing context, but current span is still alive until AsyncFinish called
+	PrepareAsync()
+	// AsyncFinish to finished current async span
+	AsyncFinish()
+}
+
 // AdaptSpan for adapt with agent core
 type AdaptSpan interface {
+	AsyncSpan
+
 	GetTraceID() string
 	GetSegmentID() string
 	GetSpanID() int32
@@ -79,3 +88,11 @@ func (s *SpanWrapper) Error(v ...string) {
 func (s *SpanWrapper) End() {
 	s.Span.End()
 }
+
+func (s *SpanWrapper) PrepareAsync() {
+	s.Span.PrepareAsync()
+}
+
+func (s *SpanWrapper) AsyncFinish() {
+	s.Span.AsyncFinish()
+}

plugins/core/tracing/span.go

@@ -104,6 +104,9 @@ type InjectorWrapper interface {
 
 // Span for plugin API
 type Span interface {
+	// AsyncSpan Async API
+	AsyncSpan
+
 	// TraceID of span
 	TraceID() string
 	// TraceSegmentID current segment ID of span