don't stop informer delivery on error

Kubernetes-commit: 2fa93da6d5efd97dbcaad262a9e59073de9c5298
This commit is contained in:
David Eads 2018-01-18 13:07:18 -05:00 committed by Kubernetes Publisher
parent 8a8517e82f
commit 1c2a06d4c0
2 changed files with 29 additions and 12 deletions

1
tools/cache/BUILD vendored
View File

@ -83,6 +83,7 @@ go_library(
"//vendor/k8s.io/client-go/rest:go_default_library", "//vendor/k8s.io/client-go/rest:go_default_library",
"//vendor/k8s.io/client-go/tools/pager:go_default_library", "//vendor/k8s.io/client-go/tools/pager:go_default_library",
"//vendor/k8s.io/client-go/util/buffer:go_default_library", "//vendor/k8s.io/client-go/util/buffer:go_default_library",
"//vendor/k8s.io/client-go/util/retry:go_default_library",
], ],
) )

View File

@ -26,6 +26,7 @@ import (
utilruntime "k8s.io/apimachinery/pkg/util/runtime" utilruntime "k8s.io/apimachinery/pkg/util/runtime"
"k8s.io/apimachinery/pkg/util/wait" "k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/util/buffer" "k8s.io/client-go/util/buffer"
"k8s.io/client-go/util/retry"
"github.com/golang/glog" "github.com/golang/glog"
) )
@ -540,20 +541,35 @@ func (p *processorListener) pop() {
} }
func (p *processorListener) run() { func (p *processorListener) run() {
defer utilruntime.HandleCrash() // this call blocks until the channel is closed. When a panic happens during the notification
// we will catch it, **the offending item will be skipped!**, and after a short delay (one second)
// the next notification will be attempted. This is usually better than the alternative of never
// delivering again.
stopCh := make(chan struct{})
wait.Until(func() {
// this gives us a few quick retries before a long pause and then a few more quick retries
err := wait.ExponentialBackoff(retry.DefaultRetry, func() (bool, error) {
for next := range p.nextCh {
switch notification := next.(type) {
case updateNotification:
p.handler.OnUpdate(notification.oldObj, notification.newObj)
case addNotification:
p.handler.OnAdd(notification.newObj)
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next))
}
}
// the only way to get here is if the p.nextCh is empty and closed
return true, nil
})
for next := range p.nextCh { // the only way to get here is if the p.nextCh is empty and closed
switch notification := next.(type) { if err == nil {
case updateNotification: close(stopCh)
p.handler.OnUpdate(notification.oldObj, notification.newObj)
case addNotification:
p.handler.OnAdd(notification.newObj)
case deleteNotification:
p.handler.OnDelete(notification.oldObj)
default:
utilruntime.HandleError(fmt.Errorf("unrecognized notification: %#v", next))
} }
} }, 1*time.Minute, stopCh)
} }
// shouldResync deterimines if the listener needs a resync. If the listener's resyncPeriod is 0, // shouldResync deterimines if the listener needs a resync. If the listener's resyncPeriod is 0,