Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

client: support configuring backoff #8679

Merged
merged 8 commits into from
Dec 2, 2024
Merged

client: support configuring backoff #8679

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9c36fff
support configuring backoff for client
rleungx Oct 9, 2024
741f0e8
refactor
rleungx Nov 27, 2024
861c9fc
refactor
rleungx Nov 27, 2024
fc51ff6
fix
rleungx Nov 27, 2024
642c989
address the comment
rleungx Nov 29, 2024
e2d9266
address the comment
rleungx Nov 29, 2024
c0b7765
address the comment
rleungx Dec 2, 2024
6e6ecb3
Merge branch 'master' into introduce-backoffer
ti-chi-bot[bot] Dec 2, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 44 additions & 21 deletions client/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@
"github.com/tikv/pd/client/metrics"
"github.com/tikv/pd/client/opt"
"github.com/tikv/pd/client/pkg/caller"
"github.com/tikv/pd/client/pkg/retry"
"github.com/tikv/pd/client/pkg/utils/grpcutil"
"github.com/tikv/pd/client/pkg/utils/tlsutil"
sd "github.com/tikv/pd/client/servicediscovery"
"go.uber.org/zap"
Expand Down Expand Up @@ -539,7 +541,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := protoClient.GetMembers(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.GetMembers(ctx, req, grpcutil.WithBackoffer(bo))
if err = c.respForErr(metrics.CmdFailedDurationGetAllMembers, start, err, resp.GetHeader()); err != nil {
return nil, err
}
Expand Down Expand Up @@ -611,9 +614,10 @@
return 0, 0, errs.ErrClientGetProtoClient
}

bo := retry.FromContext(ctx)
resp, err := protoClient.GetMinTS(ctx, &pdpb.GetMinTSRequest{
Header: c.requestHeader(),
})
}, grpcutil.WithBackoffer(bo))
if err != nil {
if strings.Contains(err.Error(), "Unimplemented") {
// If the method is not supported, we fallback to GetTS.
Expand Down Expand Up @@ -698,6 +702,7 @@
}
start := time.Now()
defer func() { metrics.CmdDurationGetRegion.Observe(time.Since(start).Seconds()) }()

ctx, cancel := context.WithTimeout(ctx, c.inner.option.Timeout)
defer cancel()

Expand All @@ -715,15 +720,16 @@
if serviceClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetRegion(cctx, req)

bo := retry.FromContext(ctx)
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetRegion(cctx, req, grpcutil.WithBackoffer(bo))
if serviceClient.NeedRetry(resp.GetHeader().GetError(), err) {
protoClient, cctx := c.getClientAndContext(ctx)
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err = protoClient.GetRegion(cctx, req)
}

if err = c.respForErr(metrics.CmdFailedDurationGetRegion, start, err, resp.GetHeader()); err != nil {
return nil, err
}
Expand Down Expand Up @@ -755,7 +761,8 @@
if serviceClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetPrevRegion(cctx, req)
bo := retry.FromContext(ctx)
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetPrevRegion(cctx, req, grpcutil.WithBackoffer(bo))
if serviceClient.NeedRetry(resp.GetHeader().GetError(), err) {
protoClient, cctx := c.getClientAndContext(ctx)
if protoClient == nil {
Expand Down Expand Up @@ -795,7 +802,8 @@
if serviceClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetRegionByID(cctx, req)
bo := retry.FromContext(ctx)
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).GetRegionByID(cctx, req, grpcutil.WithBackoffer(bo))
if serviceClient.NeedRetry(resp.GetHeader().GetError(), err) {
protoClient, cctx := c.getClientAndContext(ctx)
if protoClient == nil {
Expand Down Expand Up @@ -840,8 +848,9 @@
if serviceClient == nil {
return nil, errs.ErrClientGetProtoClient
}
bo := retry.FromContext(ctx)

Check warning on line 851 in client/client.go

View check run for this annotation

Codecov / codecov/patch

client/client.go#L851 

Added line #L851 was not covered by tests
//nolint:staticcheck
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).ScanRegions(cctx, req)
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).ScanRegions(cctx, req, grpcutil.WithBackoffer(bo))

Check warning on line 853 in client/client.go

View check run for this annotation

Codecov / codecov/patch

client/client.go#L853 

Added line #L853 was not covered by tests
failpoint.Inject("responseNil", func() {
resp = nil
})
Expand Down Expand Up @@ -896,7 +905,8 @@
if serviceClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).BatchScanRegions(cctx, req)
bo := retry.FromContext(ctx)
resp, err := pdpb.NewPDClient(serviceClient.GetClientConn()).BatchScanRegions(cctx, req, grpcutil.WithBackoffer(bo))
failpoint.Inject("responseNil", func() {
resp = nil
})
Expand Down Expand Up @@ -980,7 +990,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := protoClient.GetStore(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.GetStore(ctx, req, grpcutil.WithBackoffer(bo))

if err = c.respForErr(metrics.CmdFailedDurationGetStore, start, err, resp.GetHeader()); err != nil {
return nil, err
Expand Down Expand Up @@ -1024,7 +1035,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := protoClient.GetAllStores(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.GetAllStores(ctx, req, grpcutil.WithBackoffer(bo))

if err = c.respForErr(metrics.CmdFailedDurationGetAllStores, start, err, resp.GetHeader()); err != nil {
return nil, err
Expand All @@ -1051,7 +1063,8 @@
if protoClient == nil {
return 0, errs.ErrClientGetProtoClient
}
resp, err := protoClient.UpdateGCSafePoint(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.UpdateGCSafePoint(ctx, req, grpcutil.WithBackoffer(bo))

if err = c.respForErr(metrics.CmdFailedDurationUpdateGCSafePoint, start, err, resp.GetHeader()); err != nil {
return 0, err
Expand Down Expand Up @@ -1084,7 +1097,8 @@
if protoClient == nil {
return 0, errs.ErrClientGetProtoClient
}
resp, err := protoClient.UpdateServiceGCSafePoint(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.UpdateServiceGCSafePoint(ctx, req, grpcutil.WithBackoffer(bo))

if err = c.respForErr(metrics.CmdFailedDurationUpdateServiceGCSafePoint, start, err, resp.GetHeader()); err != nil {
return 0, err
Expand Down Expand Up @@ -1116,7 +1130,8 @@
if protoClient == nil {
return errs.ErrClientGetProtoClient
}
resp, err := protoClient.ScatterRegion(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.ScatterRegion(ctx, req, grpcutil.WithBackoffer(bo))
if err != nil {
return err
}
Expand Down Expand Up @@ -1160,7 +1175,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
return protoClient.SplitAndScatterRegions(ctx, req)
bo := retry.FromContext(ctx)
return protoClient.SplitAndScatterRegions(ctx, req, grpcutil.WithBackoffer(bo))

Check warning on line 1179 in client/client.go

View check run for this annotation

Codecov / codecov/patch

client/client.go#L1178-L1179 

Added lines #L1178 - L1179 were not covered by tests
}

// GetOperator implements the RPCClient interface.
Expand All @@ -1182,7 +1198,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
return protoClient.GetOperator(ctx, req)
bo := retry.FromContext(ctx)
return protoClient.GetOperator(ctx, req, grpcutil.WithBackoffer(bo))
}

// SplitRegions split regions by given split keys
Expand All @@ -1208,7 +1225,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
return protoClient.SplitRegions(ctx, req)
bo := retry.FromContext(ctx)
return protoClient.SplitRegions(ctx, req, grpcutil.WithBackoffer(bo))

Check warning on line 1229 in client/client.go

View check run for this annotation

Codecov / codecov/patch

client/client.go#L1228-L1229 

Added lines #L1228 - L1229 were not covered by tests
}

func (c *client) requestHeader() *pdpb.RequestHeader {
Expand Down Expand Up @@ -1240,7 +1258,8 @@
if protoClient == nil {
return nil, errs.ErrClientGetProtoClient
}
resp, err := protoClient.ScatterRegion(ctx, req)
bo := retry.FromContext(ctx)
resp, err := protoClient.ScatterRegion(ctx, req, grpcutil.WithBackoffer(bo))

if err != nil {
return nil, err
Expand All @@ -1259,7 +1278,8 @@
if protoClient == nil {
return nil, 0, errs.ErrClientGetProtoClient
}
resp, err := protoClient.LoadGlobalConfig(ctx, &pdpb.LoadGlobalConfigRequest{Names: names, ConfigPath: configPath})
bo := retry.FromContext(ctx)
resp, err := protoClient.LoadGlobalConfig(ctx, &pdpb.LoadGlobalConfigRequest{Names: names, ConfigPath: configPath}, grpcutil.WithBackoffer(bo))
if err != nil {
return nil, 0, err
}
Expand Down Expand Up @@ -1291,7 +1311,8 @@
if protoClient == nil {
return errs.ErrClientGetProtoClient
}
_, err := protoClient.StoreGlobalConfig(ctx, &pdpb.StoreGlobalConfigRequest{Changes: resArr, ConfigPath: configPath})
bo := retry.FromContext(ctx)
_, err := protoClient.StoreGlobalConfig(ctx, &pdpb.StoreGlobalConfigRequest{Changes: resArr, ConfigPath: configPath}, grpcutil.WithBackoffer(bo))
if err != nil {
return err
}
Expand Down Expand Up @@ -1358,9 +1379,10 @@
if protoClient == nil {
return 0, errs.ErrClientGetProtoClient
}
bo := retry.FromContext(ctx)
resp, err := protoClient.GetExternalTimestamp(ctx, &pdpb.GetExternalTimestampRequest{
Header: c.requestHeader(),
})
}, grpcutil.WithBackoffer(bo))
if err != nil {
return 0, err
}
Expand All @@ -1379,10 +1401,11 @@
if protoClient == nil {
return errs.ErrClientGetProtoClient
}
bo := retry.FromContext(ctx)
resp, err := protoClient.SetExternalTimestamp(ctx, &pdpb.SetExternalTimestampRequest{
Header: c.requestHeader(),
Timestamp: timestamp,
})
}, grpcutil.WithBackoffer(bo))
if err != nil {
return err
}
Expand Down
7 changes: 5 additions & 2 deletions client/meta_storage_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"github.com/tikv/pd/client/errs"
"github.com/tikv/pd/client/metrics"
"github.com/tikv/pd/client/opt"
"github.com/tikv/pd/client/pkg/retry"
"github.com/tikv/pd/client/pkg/utils/grpcutil"
)

Expand Down Expand Up @@ -78,7 +79,8 @@ func (c *innerClient) Put(ctx context.Context, key, value []byte, opts ...opt.Me
cancel()
return nil, errs.ErrClientGetMetaStorageClient
}
resp, err := cli.Put(ctx, req)
bo := retry.FromContext(ctx)
resp, err := cli.Put(ctx, req, grpcutil.WithBackoffer(bo))
cancel()

if err = c.respForMetaStorageErr(metrics.CmdFailedDurationPut, start, err, resp.GetHeader()); err != nil {
Expand Down Expand Up @@ -117,7 +119,8 @@ func (c *innerClient) Get(ctx context.Context, key []byte, opts ...opt.MetaStora
cancel()
return nil, errs.ErrClientGetMetaStorageClient
}
resp, err := cli.Get(ctx, req)
bo := retry.FromContext(ctx)
resp, err := cli.Get(ctx, req, grpcutil.WithBackoffer(bo))
cancel()

if err = c.respForMetaStorageErr(metrics.CmdFailedDurationGet, start, err, resp.GetHeader()); err != nil {
Expand Down
9 changes: 9 additions & 0 deletions client/opt/option.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@

"github.com/pingcap/errors"
"github.com/prometheus/client_golang/prometheus"
"github.com/tikv/pd/client/pkg/retry"
"google.golang.org/grpc"
)

Expand Down Expand Up @@ -61,6 +62,7 @@
UseTSOServerProxy bool
MetricsLabels prometheus.Labels
InitMetrics bool
Backoffer *retry.Backoffer

// Dynamic options.
dynamicOptions [dynamicOptionCount]atomic.Value
Expand Down Expand Up @@ -199,6 +201,13 @@
}
}

// WithBackoffer configures the client with backoffer.
func WithBackoffer(bo *retry.Backoffer) ClientOption {
return func(op *Option) {
op.Backoffer = bo
}

Check warning on line 208 in client/opt/option.go

View check run for this annotation

Codecov / codecov/patch

client/opt/option.go#L205-L208 

Added lines #L205 - L208 were not covered by tests
}

// GetStoreOp represents available options when getting stores.
type GetStoreOp struct {
ExcludeTombstone bool
Expand Down
22 changes: 22 additions & 0 deletions client/pkg/retry/backoff.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -191,3 +191,25 @@
strs := strings.Split(runtime.FuncForPC(reflect.ValueOf(f).Pointer()).Name(), ".")
return strings.Split(strs[len(strs)-1], "-")[0]
}

// Define context key type
type boCtxKey struct{}

// Key used to store backoffer
var backofferKey = boCtxKey{}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why using struct{}, not string?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I prefer to use struct{}


// FromContext retrieves the backoffer from the context
func FromContext(ctx context.Context) *Backoffer {
if ctx == nil {
return nil
}

Check warning on line 205 in client/pkg/retry/backoff.go

View check run for this annotation

Codecov / codecov/patch

client/pkg/retry/backoff.go#L204-L205 

Added lines #L204 - L205 were not covered by tests
if bo, ok := ctx.Value(backofferKey).(*Backoffer); ok {
return bo
}
return nil
}

// WithBackoffer stores the backoffer into a new context
func WithBackoffer(ctx context.Context, bo *Backoffer) context.Context {
return context.WithValue(ctx, backofferKey, bo)
}
Loading