gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(access): new span for background task

Browse Source 
. #1000247577

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2025-07-24 16:17:01 +08:00
parent 5e452b92f9
commit b346da9385
5 changed files with 17 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
blobstore/access/stream/stream.go
View File

@ -379,7 +379,6 @@ func (h *Handler) Admin() interface{} {
}
func (h *Handler) sendRepairMsgBg(ctx context.Context, blob blobIdent, badIdxes []uint8) {
ctx = trace.NewContextFromContext(ctx)
go func() {
h.sendRepairMsg(ctx, blob, badIdxes)
}()

23
blobstore/access/stream/stream_get.go
View File

@ -169,7 +169,9 @@ func (h *Handler) Get(ctx context.Context, w io.Writer, location proto.Location,
}
}
_, ctx = trace.StartSpanFromContextWithTraceID(ctx, "", span.TraceID())
var spanpipe trace.Span
spanpipe, ctx = trace.StartSpanFromContextWithTraceID(context.Background(), "", span.TraceID())
defer spanpipe.Finish()
// data stream flow:
// client <--copy-- pipeline <--swap-- readBlob <--copy-- blobnode
@ -190,7 +192,7 @@ func (h *Handler) Get(ctx context.Context, w io.Writer, location proto.Location,
if blobVolume == nil || blobVolume.Vid != blob.Vid {
blobVolume, err = h.getVolume(ctx, clusterID, blob.Vid, true)
if err != nil {
span.Error("get volume", err)
spanpipe.Error("get volume", err)
ch <- pipeBuffer{err: err}
return
}
@ -198,11 +200,11 @@ func (h *Handler) Get(ctx context.Context, w io.Writer, location proto.Location,
// do not use local shards
ordered := h.CodeModesGetOrdered[blobVolume.CodeMode]
sortedVuids = genSortedVuidByIDC(ctx, serviceController, h.IDC, blobVolume.Units[:tactic.N+tactic.M], ordered)
span.Debugf("to read %s with read-shard-x:%d active-shard-n:%d of data-n:%d party-n:%d",
spanpipe.Debugf("to read %s with read-shard-x:%d active-shard-n:%d of data-n:%d party-n:%d",
blob.ID(), h.MinReadShardsX, len(sortedVuids), tactic.N, tactic.M)
if len(sortedVuids) < tactic.N {
err = fmt.Errorf("broken %s", blob.ID())
span.Error(err)
spanpipe.Error(err)
ch <- pipeBuffer{err: err}
return
}
@ -218,7 +220,7 @@ func (h *Handler) Get(ctx context.Context, w io.Writer, location proto.Location,
err = h.readOneBlob(ctx, getTime, serviceController, blob, sortedVuids, shards)
if err != nil {
span.Error("read one blob", blob.ID(), err)
spanpipe.Error("read one blob", blob.ID(), err)
for _, buf := range shards {
h.memPool.Put(buf)
}
@ -690,16 +692,7 @@ func (h *Handler) getOneShardFromHost(ctx context.Context, serviceController con
}
}
ctxChild := ctx
if cancelChan != nil { // cancelChan == nil means reading data shard only
// new child span to get from blobnode, we should finish it here.
var spanChild trace.Span
spanChild, ctxChild = trace.StartSpanFromContextWithTraceID(
context.Background(), "GetFromBlobnode", span.TraceID())
defer spanChild.Finish()
}
body, crc, err := h.blobnodeClient.RangeGetShard(ctxChild, host, &args)
body, crc, err := h.blobnodeClient.RangeGetShard(ctx, host, &args)
if err == nil {
rbody = body
rcrc = crc

17
blobstore/access/stream/stream_put.go
View File

@ -109,8 +109,6 @@ func (h *Handler) Put(ctx context.Context,
ready <- struct{}{}
}
_, ctx = trace.StartSpanFromContextWithTraceID(ctx, "", span.TraceID())
encoder := h.encoder[selectedCodeMode]
tactic := selectedCodeMode.Tactic()
for _, blob := range location.Spread() {
@ -199,7 +197,6 @@ type shardPutStatus struct {
func (h *Handler) writeToBlobnodes(ctx context.Context,
blob blobIdent, shards [][]byte, empties map[int]struct{}, callback func(),
) (err error) {
span := trace.SpanFromContextSafe(ctx)
clusterID, vid, bid := blob.cid, blob.vid, blob.bid
wg := &sync.WaitGroup{}
@ -227,6 +224,11 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
putQuorum = uint32(num)
}
span := trace.SpanFromContextSafe(ctx)
// new context span to write blobnode in background
span, ctx = trace.StartSpanFromContextWithTraceID(context.Background(), "", span.TraceID())
defer span.Finish()
writeStart := time.Now()
writeTime := int32(0)
@ -266,13 +268,8 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
crcOrigin = crc32.ChecksumIEEE(shards[index])
}
// new child span to write to blobnode, we should finish it here.
spanChild, ctxChild := trace.StartSpanFromContextWithTraceID(
context.Background(), "WriteToBlobnode", span.TraceID())
defer spanChild.Finish()
RETRY:
hostInfo, err := serviceController.GetDiskHost(ctxChild, diskID)
hostInfo, err := serviceController.GetDiskHost(ctx, diskID)
if err != nil {
span.Error("get disk host failed", errors.Detail(err))
return
@ -295,7 +292,7 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
args.Body = bytes.NewReader(shards[index])
}
crc, err = h.blobnodeClient.PutShard(ctxChild, host, args)
crc, err = h.blobnodeClient.PutShard(ctx, host, args)
if err == nil {
if !crcDisable && crc != crcOrigin {
return false, fmt.Errorf("crc mismatch 0x%x != 0x%x", crc, crcOrigin)

2
blobstore/common/trace/propagation.go
View File

@ -162,7 +162,7 @@ func (t *TextMapPropagator) Extract(carrier interface{}) (opentracing.SpanContex
})
span.context.spanID = spanID
span.context.resetID(traceID)
span.context.spanFromPool = span // notify release to pool
// span.context.spanFromPool = span // notify release to pool // TODO: cacheable later
return span.context, nil
}

2
blobstore/common/trace/tracer.go
View File

@ -288,7 +288,7 @@ func StartCacheableSpan(ctx context.Context, operationName, traceID string) (Spa
span := poolSpan.Get().(*spanImpl)
span.context.spanID = RandomID()
span.context.resetID(traceID)
span.context.spanFromPool = span // notify release to pool
// span.context.spanFromPool = span // notify release to pool // TODO: cacheable later
optsCarrier := span.context.optsCarrier
return StartSpanFromContextWithTraceID(ctx, operationName, traceID, optsCarrier...)
}