diff --git a/blobstore/common/trace/propagation.go b/blobstore/common/trace/propagation.go index 2d9377c50..f7f38df2e 100644 --- a/blobstore/common/trace/propagation.go +++ b/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 diff --git a/blobstore/common/trace/span.go b/blobstore/common/trace/span.go index 49bdd4b08..8d3068bce 100644 --- a/blobstore/common/trace/span.go +++ b/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) + } + }) } } diff --git a/blobstore/common/trace/span_context.go b/blobstore/common/trace/span_context.go index 32db287cb..63b8c632d 100644 --- a/blobstore/common/trace/span_context.go +++ b/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 { diff --git a/blobstore/common/trace/span_test.go b/blobstore/common/trace/span_test.go index e28ce58b2..5603e98e8 100644 --- a/blobstore/common/trace/span_test.go +++ b/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()