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

feat(access): retry uniqe proxy service in access

Browse Source 
. #1000403229

Signed-off-by: slasher <shenjie1@oppo.com>
This commit is contained in:
slasher 2025-10-28 15:37:43 +08:00 committed by 梁曟風
parent b6ee58a15f
commit 5a588bb032
2 changed files with 39 additions and 35 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -104,7 +104,7 @@ type StreamHandler interface {
Delete(ctx context.Context, location *proto.Location) error
// Admin returns internal admin interface.
Admin() interface{}
Admin() any
// GetBlob returns location
GetBlob(ctx context.Context, args *access.GetBlobArgs) (*proto.Location, error)
// DeleteBlob returns error
@ -389,7 +389,7 @@ func (h *Handler) Delete(ctx context.Context, location *proto.Location) error {
}
// Admin returns internal admin interface.
func (h *Handler) Admin() interface{} {
func (h *Handler) Admin() any {
return &StreamAdmin{
Config: h.StreamConfig,
MemPool: h.memPool,
@ -423,13 +423,17 @@ func (h *Handler) sendRepairMsg(ctx context.Context, blob blobIdent, badIdxes []
Reason: "access-repair",
}
hosts, err := serviceController.GetServiceHosts(ctx, serviceProxy)
if err != nil {
span.Error(errors.Detail(err))
return
}
for len(hosts) < 3 {
hosts = append(hosts, hosts...)
}
if err := retry.Timed(3, 200).On(func() error {
host, err := serviceController.GetServiceHost(ctx, serviceProxy)
if err != nil {
reportUnhealth(clusterID, "repair.msg", serviceProxy, "-", "failed")
span.Warn(err)
return err
}
host := hosts[0]
hosts = hosts[1:]
err = h.proxyClient.SendShardRepairMsg(ctx, host, repairArgs)
if err != nil {
if errorTimeout(err) || errorConnectionRefused(err) {
@ -471,17 +475,21 @@ func (h *Handler) clearGarbage(ctx context.Context, location *proto.Location) er
})
}
var logMsg interface{} = location
var logMsg any = location
if len(deleteArgs.Blobs) <= 20 {
logMsg = deleteArgs
}
hosts, err := serviceController.GetServiceHosts(ctx, serviceProxy)
if err != nil {
span.Error(err)
return err
}
for len(hosts) < 3 {
hosts = append(hosts, hosts...)
}
if err := retry.Timed(3, 200).On(func() error {
host, err := serviceController.GetServiceHost(ctx, serviceProxy)
if err != nil {
reportUnhealth(location.ClusterID, "delete.msg", serviceProxy, "-", "failed")
span.Warn(err)
return err
}
host := hosts[0]
hosts = hosts[1:]
err = h.proxyClient.SendDeleteMsg(ctx, host, deleteArgs)
if err != nil {
if errorTimeout(err) || errorConnectionRefused(err) {

36
blobstore/access/stream/stream_alloc.go
View File

@ -116,29 +116,25 @@ func (h *Handler) allocFromAllocator(ctx context.Context,
BidCount: util.AlignedBlocks(size, uint64(blobSize)),
}
serviceController, err := h.clusterController.GetServiceController(clusterID)
if err != nil {
span.Error(err)
return 0, nil, err
}
hosts, err := serviceController.GetServiceHosts(ctx, serviceProxy)
if err != nil {
span.Error(err)
return 0, nil, err
}
for len(hosts) < h.AllocRetryTimes {
hosts = append(hosts, hosts...)
}
var allocRets []proxy.AllocRet
var allocHost string
hostsSet := make(map[string]struct{}, 1)
if err := retry.ExponentialBackoff(h.AllocRetryTimes, uint32(h.AllocRetryIntervalMS)).On(func() error {
serviceController, err := h.clusterController.GetServiceController(clusterID)
if err != nil {
span.Warn(err)
return errors.Info(err, "get service controller", clusterID)
}
var host string
for range [10]struct{}{} {
host, err = serviceController.GetServiceHost(ctx, serviceProxy)
if err != nil {
span.Warn(err)
return errors.Info(err, "get proxy host", clusterID)
}
if _, ok := hostsSet[host]; ok {
continue
}
hostsSet[host] = struct{}{}
break
}
host := hosts[0]
hosts = hosts[1:]
allocHost = host
allocRets, err = h.proxyClient.VolumeAlloc(ctx, host, &args)