mirror of
https://github.com/kubernetes/client-go.git
synced 2025-08-15 05:53:15 +00:00
Merge pull request #118132 from marseel/improve_reflector_retries
Improve backoff policy in reflector. Kubernetes-commit: 7f2a1e8cd3363b04d970c619daf9f9d5ee5d50c0
This commit is contained in:
Kubernetes Publisher
commit
f21df6e02d
4
go.mod
4
go.mod
@ -23,7 +23,7 @@ require (
|
||||
golang.org/x/term v0.7.0
|
||||
golang.org/x/time v0.3.0
|
||||
google.golang.org/protobuf v1.30.0
|
||||
k8s.io/api v0.0.0-20230515170019-2f9553831ec2
|
||||
k8s.io/api v0.0.0-20230527221710-63505c4e796f
|
||||
k8s.io/apimachinery v0.0.0-20230526141509-403fb36f6a53
|
||||
k8s.io/klog/v2 v2.100.1
|
||||
k8s.io/kube-openapi v0.0.0-20230524182850-78281498afbb
|
||||
@ -60,6 +60,6 @@ require (
|
||||
)
|
||||
|
||||
replace (
|
||||
k8s.io/api => k8s.io/api v0.0.0-20230515170019-2f9553831ec2
|
||||
k8s.io/api => k8s.io/api v0.0.0-20230527221710-63505c4e796f
|
||||
k8s.io/apimachinery => k8s.io/apimachinery v0.0.0-20230526141509-403fb36f6a53
|
||||
)
|
||||
|
||||
4
go.sum
4
go.sum
@ -210,8 +210,8 @@ gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
honnef.co/go/tools v0.0.0-20190102054323-c2f93a96b099/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
||||
honnef.co/go/tools v0.0.0-20190523083050-ea95bdfd59fc/go.mod h1:rf3lG4BRIbNafJWhAfAdb/ePZxsR/4RtNHQocxwk9r4=
|
||||
k8s.io/api v0.0.0-20230515170019-2f9553831ec2 h1:ZHyoWIXElhnLI03GkOsebYT1uqYowGjuVCbxBqcjjGw=
|
||||
k8s.io/api v0.0.0-20230515170019-2f9553831ec2/go.mod h1:qZNz3vKcyZb5wFaBlHqUjHk7D5XhIaXu/d/df6PjNKk=
|
||||
k8s.io/api v0.0.0-20230527221710-63505c4e796f h1:5n05Z8OLrrVvkQF6z0OgUK3yGzRm7GY1bUwhUBp3g5g=
|
||||
k8s.io/api v0.0.0-20230527221710-63505c4e796f/go.mod h1:iclI1239YG5Z5YyOH/tKJi2vGhQ0k0dl+1xSvFaR6Dk=
|
||||
k8s.io/apimachinery v0.0.0-20230526141509-403fb36f6a53 h1:8/61R28R74ZzZ1WiUVu4zqCtyQSdVCBa4qCNjPLjOsU=
|
||||
k8s.io/apimachinery v0.0.0-20230526141509-403fb36f6a53/go.mod h1:qCTDst7QeP2n3JDxbpuJSTTaAxalvFKMTN0Lga1+0zU=
|
||||
k8s.io/klog/v2 v2.100.1 h1:7WCHKK6K8fNhTqfBhISHQ97KrnJNFZMcQvKp7gP/tmg=
|
||||
|
||||
19
tools/cache/reflector.go
vendored
19
tools/cache/reflector.go
vendored
@ -70,9 +70,7 @@ type Reflector struct {
|
||||
listerWatcher ListerWatcher
|
||||
// backoff manages backoff of ListWatch
|
||||
backoffManager wait.BackoffManager
|
||||
// initConnBackoffManager manages backoff the initial connection with the Watch call of ListAndWatch.
|
||||
initConnBackoffManager wait.BackoffManager
|
||||
resyncPeriod time.Duration
|
||||
resyncPeriod time.Duration
|
||||
// clock allows tests to manipulate time
|
||||
clock clock.Clock
|
||||
// paginatedResult defines whether pagination should be forced for list calls.
|
||||
@ -221,11 +219,10 @@ func NewReflectorWithOptions(lw ListerWatcher, expectedType interface{}, store S
|
||||
// We used to make the call every 1sec (1 QPS), the goal here is to achieve ~98% traffic reduction when
|
||||
// API server is not healthy. With these parameters, backoff will stop at [30,60) sec interval which is
|
||||
// 0.22 QPS. If we don't backoff for 2min, assume API server is healthy and we reset the backoff.
|
||||
backoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, reflectorClock),
|
||||
initConnBackoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, reflectorClock),
|
||||
clock: reflectorClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
expectedType: reflect.TypeOf(expectedType),
|
||||
backoffManager: wait.NewExponentialBackoffManager(800*time.Millisecond, 30*time.Second, 2*time.Minute, 2.0, 1.0, reflectorClock),
|
||||
clock: reflectorClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
expectedType: reflect.TypeOf(expectedType),
|
||||
}
|
||||
|
||||
if r.name == "" {
|
||||
@ -425,7 +422,7 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
|
||||
select {
|
||||
case <-stopCh:
|
||||
return nil
|
||||
case <-r.initConnBackoffManager.Backoff().C():
|
||||
case <-r.backoffManager.Backoff().C():
|
||||
continue
|
||||
}
|
||||
}
|
||||
@ -451,7 +448,7 @@ func (r *Reflector) watch(w watch.Interface, stopCh <-chan struct{}, resyncerrc
|
||||
select {
|
||||
case <-stopCh:
|
||||
return nil
|
||||
case <-r.initConnBackoffManager.Backoff().C():
|
||||
case <-r.backoffManager.Backoff().C():
|
||||
continue
|
||||
}
|
||||
case apierrors.IsInternalError(err) && retry.ShouldRetry():
|
||||
@ -604,7 +601,7 @@ func (r *Reflector) watchList(stopCh <-chan struct{}) (watch.Interface, error) {
|
||||
isErrorRetriableWithSideEffectsFn := func(err error) bool {
|
||||
if canRetry := isWatchErrorRetriable(err); canRetry {
|
||||
klog.V(2).Infof("%s: watch-list of %v returned %v - backing off", r.name, r.typeDescription, err)
|
||||
<-r.initConnBackoffManager.Backoff().C()
|
||||
<-r.backoffManager.Backoff().C()
|
||||
return true
|
||||
}
|
||||
if isExpiredError(err) || isTooLargeResourceVersionError(err) {
|
||||
|
||||
36
tools/cache/reflector_test.go
vendored
36
tools/cache/reflector_test.go
vendored
@ -411,12 +411,12 @@ func TestReflectorListAndWatchInitConnBackoff(t *testing.T) {
|
||||
},
|
||||
}
|
||||
r := &Reflector{
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
initConnBackoffManager: bm,
|
||||
clock: fakeClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
backoffManager: bm,
|
||||
clock: fakeClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
}
|
||||
start := fakeClock.Now()
|
||||
err := r.ListAndWatch(stopCh)
|
||||
@ -471,12 +471,12 @@ func TestBackoffOnTooManyRequests(t *testing.T) {
|
||||
}
|
||||
|
||||
r := &Reflector{
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
initConnBackoffManager: bm,
|
||||
clock: clock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
backoffManager: bm,
|
||||
clock: clock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
}
|
||||
|
||||
stopCh := make(chan struct{})
|
||||
@ -540,12 +540,12 @@ func TestRetryInternalError(t *testing.T) {
|
||||
}
|
||||
|
||||
r := &Reflector{
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
initConnBackoffManager: bm,
|
||||
clock: fakeClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
name: "test-reflector",
|
||||
listerWatcher: lw,
|
||||
store: NewFIFO(MetaNamespaceKeyFunc),
|
||||
backoffManager: bm,
|
||||
clock: fakeClock,
|
||||
watchErrorHandler: WatchErrorHandler(DefaultWatchErrorHandler),
|
||||
}
|
||||
|
||||
r.MaxInternalErrorRetryDuration = tc.maxInternalDuration
|
||||
|
||||
Loading…
Reference in New Issue
Block a user