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

fix(access): force to send message to proxy and client cleans up

Browse Source 
. #22411498

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2024-10-29 10:50:52 +08:00 committed by 梁曟風
parent 6d8981bf9c
commit 3d33e4f992
3 changed files with 10 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -322,6 +322,8 @@ func (h *Handler) Admin() interface{} {
}
func (h *Handler) sendRepairMsgBg(ctx context.Context, blob blobIdent, badIdxes []uint8) {
span := trace.SpanFromContextSafe(ctx)
_, ctx = trace.StartSpanFromContextWithTraceID(context.Background(), "", span.TraceID())
go func() {
h.sendRepairMsg(ctx, blob, badIdxes)
}()

6
blobstore/api/access/client.go
View File

@ -532,16 +532,18 @@ func (c *client) putParts(ctx context.Context, args *PutArgs) (Location, HashSum
return
}
// force to clean up, even canceled context
_, newCtx := trace.StartSpanFromContextWithTraceID(context.Background(), "", span.TraceID())
locations := signArgs.Locations[:]
if len(locations) > 1 {
signArgs.Location = loc.Copy()
signResp := &SignResp{}
if err := rpcClient.PostWith(ctx, "/sign", signResp, signArgs); err == nil {
if err := rpcClient.PostWith(newCtx, "/sign", signResp, signArgs); err == nil {
locations = []Location{signResp.Location.Copy()}
}
}
if len(locations) > 0 {
if _, err := c.Delete(ctx, &DeleteArgs{Locations: locations}); err != nil {
if _, err := c.Delete(newCtx, &DeleteArgs{Locations: locations}); err != nil {
span.Warnf("clean location '%+v' failed %s", locations, err.Error())
}
}

6
blobstore/sdk/sdk_client.go
View File

@ -629,16 +629,18 @@ func (s *sdkHandler) putParts(ctx context.Context, args *acapi.PutArgs) (acapi.L
return
}
// force to clean up, even canceled context
_, newCtx := trace.StartSpanFromContextWithTraceID(context.Background(), "", span.TraceID())
locations := signArgs.Locations[:]
if len(locations) > 1 {
signArgs.Location = loc.Copy()
signResp, err := s.sign(ctx, &signArgs)
signResp, err := s.sign(newCtx, &signArgs)
if err == nil {
locations = []acapi.Location{signResp.Location.Copy()}
}
}
if len(locations) > 0 {
if _, err := s.Delete(ctx, &acapi.DeleteArgs{Locations: locations}); err != nil {
if _, err := s.Delete(newCtx, &acapi.DeleteArgs{Locations: locations}); err != nil {
span.Warnf("clean location '%+v' failed %s", locations, err.Error())
}
}