fix(trace): thread safe in trace span logging

. #1000320592

Signed-off-by: slasher <shenjie1@oppo.com>

blobstore/common/trace/propagation.go

@@ -153,9 +153,11 @@ func (t *TextMapPropagator) Extract(carrier interface{}) (opentracing.SpanContex
 			lowerKey := strings.ToLower(key)
 			if strings.HasPrefix(lowerKey, PrefixBaggage) {
 				k := strings.TrimPrefix(lowerKey, PrefixBaggage)
-				if b := span.context.nextBuffer(k, 8); b != nil {
-					b.WriteString(val)
-				}
+				span.context.appendBaggage(func(nextBuffer func(string, int) *bytes.Buffer) {
+					if b := nextBuffer(k, 8); b != nil {
+						b.WriteString(val)
+					}
+				})
 			}
 		}
 		return nil

blobstore/common/trace/span.go

@@ -71,6 +71,7 @@ type Span interface {
 	// TrackLog returns track log, calls BaggageItem with default key fieldTrackLogKey.
 	TrackLog() []string
 	TrackLogN() int
+	// TrackLogRange b is not thread-safe out of this function.
 	TrackLogRange(func(b *bytes.Buffer) bool)
 
 	// BaseLogger defines interface of application log apis.
@@ -147,15 +148,16 @@ func (s *spanImpl) FinishWithOptions(opts opentracing.FinishOptions) {
 	if s.context.spanFromPool != s {
 		return
 	}
-	if len(s.context.baggage) > 64 { // too many baggages
-		return
-	}
 	ctx := s.context
 	ctx.spanFromPool = nil
 	ctx.clearID()
 	ctx.spanID = 0
 	ctx.parentID = 0
 	ctx.Lock()
+	if len(ctx.baggage) > 64 { // too many baggages
+		ctx.Unlock()
+		return
+	}
 	for key := range ctx.baggage {
 		ctx.baggage[key].setString(nil)
 	}
@@ -333,11 +335,20 @@ func (s *spanImpl) AppendTrackLogWithDuration(module string, duration time.Durat
 	for _, opt := range opts {
 		spanOpt = opt(spanOpt)
 	}
-
-	b := s.context.nextTrack(s.tracer.options.maxInternalTrack)
-	if b == nil {
-		return
+	var appendBytes []byte
+	s.context.appendTrack(func(nextBuffer func(maxTracks int) *bytes.Buffer) {
+		appendBytes = s._append(nextBuffer(s.tracer.options.maxInternalTrack), module, duration, err, spanOpt)
+	})
+	if len(appendBytes) > 0 {
+		s.trackReferences(appendBytes)
 	}
+}
+
+func (s *spanImpl) _append(b *bytes.Buffer, module string, duration time.Duration, err error, spanOpt spanOptions) []byte {
+	if b == nil {
+		return nil
+	}
+	b.Grow(16)
 	b.WriteString(module)
 
 	if spanOpt.duration == durationAny {
@@ -362,7 +373,8 @@ func (s *spanImpl) AppendTrackLogWithDuration(module string, duration time.Durat
 			b.WriteString(msg)
 		}
 	}
-	s.trackReferences(b)
+
+	return b.Bytes()
 }
 
 // AppendTrackLogWithFunc records cost time for the function calling to a module.
@@ -374,13 +386,19 @@ func (s *spanImpl) AppendTrackLogWithFunc(module string, fn func() error, opts .
 
 // AppendRPCTrackLog appends RPC track logs to baggage with default key fieldTrackLogKey.
 func (s *spanImpl) AppendRPCTrackLog(logs []string) {
-	for _, trackLog := range logs {
-		b := s.context.nextTrack(s.tracer.options.maxInternalTrack)
-		if b == nil {
-			return
+	written := 0
+	s.context.appendTrack(func(nextBuffer func(int) *bytes.Buffer) {
+		for _, trackLog := range logs {
+			b := nextBuffer(s.tracer.options.maxInternalTrack)
+			if b == nil {
+				return
+			}
+			b.WriteString(trackLog)
+			written++
 		}
-		b.WriteString(trackLog)
-		s.trackReferences(b)
+	})
+	for idx := 0; idx < written; idx++ {
+		s.trackReferences([]byte(logs[idx]))
 	}
 }
 
@@ -397,16 +415,18 @@ func (s *spanImpl) TrackLogRange(f func(b *bytes.Buffer) bool) {
 	s.context.trackLogsRange(f)
 }
 
-func (s *spanImpl) trackReferences(buf *bytes.Buffer) {
+func (s *spanImpl) trackReferences(p []byte) {
 	maxTracks := s.tracer.options.maxInternalTrack
 	for _, ref := range s.references {
 		spanCtx, ok := ref.ReferencedContext.(*SpanContext)
 		if !ok {
 			continue
 		}
-		if b := spanCtx.nextTrack(maxTracks); b != nil {
-			b.Write(buf.Bytes())
-		}
+		spanCtx.appendTrack(func(nextBuffer func(int) *bytes.Buffer) {
+			if b := nextBuffer(maxTracks); b != nil {
+				b.Write(p)
+			}
+		})
 	}
 }
 

blobstore/common/trace/span_context.go

@@ -77,7 +77,7 @@ type SpanContext struct {
 	// Should be 0 if the current span is a root span.
 	parentID ID
 
-	// Distributed Context baggage.
+	// Distributed Context baggage, write or read logItems's Buffer should with lock.
 	baggage map[string]*logItems
 	sync.RWMutex
 
@@ -106,13 +106,13 @@ func (s *SpanContext) ForeachBaggageItem(handler func(k, v string) bool) {
 
 // ForeachBaggageItems will called the handler function  for each baggage key/values pair.
 func (s *SpanContext) ForeachBaggageItems(handler func(k string, buffers []*bytes.Buffer) bool) {
-	s.Lock()
+	s.RLock()
 	for k, v := range s.baggage {
 		if !handler(k, v.bufs) {
 			break
 		}
 	}
-	s.Unlock()
+	s.RUnlock()
 }
 
 func (s *SpanContext) clearID() {
@@ -143,6 +143,7 @@ func (s *SpanContext) traceID() string {
 	return string(s.id[s.traceIndex:_lenTraceID])
 }
 
+// _getBaggage without lock
 func (s *SpanContext) _getBaggage(key string) (item *logItems) {
 	if s.baggage == nil {
 		s.baggage = map[string]*logItems{key: {}}
@@ -169,47 +170,66 @@ func (s *SpanContext) setBaggageItem(key string, value []string) {
 
 func (s *SpanContext) traceLogs() (logs []string) {
 	s.RLock()
-	s._getBaggage(internalTrackLogKey)
-	logs = s.baggageItem(internalTrackLogKey)
+	track := s.baggage[internalTrackLogKey]
 	s.RUnlock()
+	if track == nil {
+		return nil
+	}
+	logs = s.baggageItem(internalTrackLogKey)
 	return
 }
 
 func (s *SpanContext) trackLogsN() (n int) {
 	s.RLock()
-	n = s._getBaggage(internalTrackLogKey).length()
+	if track := s.baggage[internalTrackLogKey]; track != nil {
+		n = track.length()
+	}
 	s.RUnlock()
 	return
 }
 
 func (s *SpanContext) trackLogsRange(f func(*bytes.Buffer) bool) {
 	s.RLock()
-	track := s._getBaggage(internalTrackLogKey)
-	for idx := range track.bufs {
-		if !f(track.bufs[idx]) {
-			break
+	if track := s.baggage[internalTrackLogKey]; track != nil {
+		for idx := range track.bufs {
+			if !f(track.bufs[idx]) {
+				break
+			}
 		}
 	}
 	s.RUnlock()
 }
 
-func (s *SpanContext) nextTrack(maxTracks int) (b *bytes.Buffer) {
-	return s.nextBuffer(internalTrackLogKey, maxTracks)
+func (s *SpanContext) appendTrack(f func(nextBuffer func(maxTracks int) *bytes.Buffer)) {
+	s.Lock()
+	f(s._nextTrack)
+	s.Unlock()
 }
 
-func (s *SpanContext) nextBuffer(key string, maxTracks int) (b *bytes.Buffer) {
+func (s *SpanContext) appendBaggage(f func(nextBuffer func(key string, maxTracks int) *bytes.Buffer)) {
 	s.Lock()
+	f(s._nextBuffer)
+	s.Unlock()
+}
+
+// _nextTrack without lock
+func (s *SpanContext) _nextTrack(maxTracks int) *bytes.Buffer {
+	return s._nextBuffer(internalTrackLogKey, maxTracks)
+}
+
+// _nextBuffer without lock
+func (s *SpanContext) _nextBuffer(key string, maxTracks int) (b *bytes.Buffer) {
 	track := s._getBaggage(key)
 	if track.length() < maxTracks {
 		track.grow(1)
 		b = track.bufs[track.length()-1]
 		b.Reset()
 	}
-	s.Unlock()
 	return
 }
 
 func (s *SpanContext) baggageItem(key string) (item []string) {
+	s.RLock()
 	track := s.baggage[key]
 	if track.length() > 0 {
 		item = make([]string, 0, track.length())
@@ -217,6 +237,7 @@ func (s *SpanContext) baggageItem(key string) (item []string) {
 			item = append(item, track.bufs[idx].String())
 		}
 	}
+	s.RUnlock()
 	return
 }
 
@@ -227,7 +248,10 @@ func (s *SpanContext) IsValid() bool {
 
 // IsEmpty returns true is span context is empty
 func (s *SpanContext) IsEmpty() bool {
-	return !s.IsValid() && len(s.baggage) == 0
+	s.RLock()
+	l := len(s.baggage)
+	s.RUnlock()
+	return !s.IsValid() && l == 0
 }
 
 type logItems struct {

blobstore/common/trace/span_test.go

@@ -29,6 +29,7 @@ import (
 	ptlog "github.com/opentracing/opentracing-go/log"
 	"github.com/stretchr/testify/require"
 
+	"github.com/cubefs/cubefs/blobstore/util"
 	"github.com/cubefs/cubefs/blobstore/util/log"
 )
 
@@ -186,13 +187,75 @@ func TestSpan_Baggage(t *testing.T) {
 
 	spanChild.SetBaggageItem("k4", "v4")
 	spanCtx := spanChild.(*spanImpl).context
-	b := spanCtx.nextBuffer("k4", 32)
-	b.WriteString("v4")
+	spanCtx.appendBaggage(func(nextBuffer func(string, int) *bytes.Buffer) {
+		b := nextBuffer("k4", 32)
+		b.WriteString("v4")
+	})
 	require.Equal(t, "v4,v4", spanChild.BaggageItem("k4"))
 	require.Equal(t, "v4", span.BaggageItem("k4"))
 	spanChild.Tracer().Inject(spanCtx, HTTPHeaders, HTTPHeadersCarrier(http.Header{}))
 }
 
+func safeLogging(start <-chan struct{}, span Span, name string) {
+	span.SetBaggageItem(name, name)
+	var wg sync.WaitGroup
+	wg.Add(10)
+	for idx := range [10]struct{}{} {
+		go func(idx int) {
+			<-start
+			_ = util.Any2String(span.TrackLog())
+			span.AppendTrackLogWithDuration(name+"-"+util.Any2String(idx), time.Second, nil, ConstOptSpanAny...)
+			wg.Done()
+		}(idx)
+	}
+	wg.Wait()
+}
+
+func TestSpan_BaggageSafe(t *testing.T) {
+	spanRoot, ctx := StartSpanFromContext(context.Background(), "safe baggage")
+	defer spanRoot.Finish()
+
+	start := make(chan struct{})
+	wait := make(chan struct{})
+	go func() {
+		safeLogging(start, spanRoot, "root")
+		wait <- struct{}{}
+	}()
+
+	spanChild1, ctx1 := StartSpanFromContext(ctx, "child of span")
+	go func() {
+		safeLogging(start, spanChild1, "child1")
+		wait <- struct{}{}
+	}()
+	spanChild2, ctx2 := StartSpanFromContext(ctx, "child of span")
+	go func() {
+		safeLogging(start, spanChild2, "child2")
+		wait <- struct{}{}
+	}()
+
+	span1, _ := StartSpanFromContext(ctx1, "child child of span")
+	go func() {
+		safeLogging(start, span1, "span1")
+		wait <- struct{}{}
+	}()
+	span2, _ := StartSpanFromContext(ctx2, "child child of span")
+	go func() {
+		safeLogging(start, span2, "span2")
+		wait <- struct{}{}
+	}()
+
+	close(start)
+	for range [5]struct{}{} {
+		<-wait
+	}
+
+	t.Log(spanRoot.TrackLog())
+	t.Log(spanChild1.TrackLog())
+	t.Log(spanChild2.TrackLog())
+	t.Log(span1.TrackLog())
+	t.Log(span2.TrackLog())
+}
+
 func TestSpan_TrackLog(t *testing.T) {
 	span, ctx := StartSpanFromContext(context.Background(), "test trackLog")
 	defer span.Finish()