Back off upon errored kubernetes api requests (#2562)

closes #1009

probe/kubernetes/client.go

@@ -1,11 +1,14 @@
 package kubernetes
 
 import (
+	"fmt"
 	"io"
 	"strconv"
 	"sync"
 	"time"
 
+	"github.com/weaveworks/common/backoff"
+
 	log "github.com/Sirupsen/logrus"
 	"k8s.io/kubernetes/pkg/api"
 	"k8s.io/kubernetes/pkg/apis/extensions"
@@ -16,7 +19,6 @@ import (
 	clientcmdapi "k8s.io/kubernetes/pkg/client/unversioned/clientcmd/api"
 	"k8s.io/kubernetes/pkg/fields"
 	"k8s.io/kubernetes/pkg/labels"
-	"k8s.io/kubernetes/pkg/util/wait"
 )
 
 // Client keeps track of running kubernetes pods and services
@@ -55,15 +57,22 @@ type client struct {
 	podWatches      []func(Event, Pod)
 }
 
-// runReflectorUntil is equivalent to cache.Reflector.RunUntil, but it also logs
-// errors, which cache.Reflector.RunUntil simply ignores
-func runReflectorUntil(r *cache.Reflector, resyncPeriod time.Duration, stopCh <-chan struct{}) {
-	loggingListAndWatch := func() {
-		if err := r.ListAndWatch(stopCh); err != nil {
-			log.Errorf("Kubernetes reflector: %v", err)
+// runReflectorUntil runs cache.Reflector#ListAndWatch in an endless loop.
+// Errors are logged and retried with exponential backoff.
+func runReflectorUntil(r *cache.Reflector, resyncPeriod time.Duration, stopCh <-chan struct{}, msg string) {
+	listAndWatch := func() (bool, error) {
+		select {
+		case <-stopCh:
+			return true, nil
+		default:
+			err := r.ListAndWatch(stopCh)
+			return false, err
 		}
 	}
-	go wait.Until(loggingListAndWatch, resyncPeriod, stopCh)
+	bo := backoff.New(listAndWatch, fmt.Sprintf("Kubernetes reflector (%s)", msg))
+	bo.SetInitialBackoff(resyncPeriod)
+	bo.SetMaxBackoff(5 * time.Minute)
+	go bo.Start()
 }
 
 // ClientConfig establishes the configuration for the kubernetes client
@@ -169,7 +178,7 @@ func (c *client) setupStore(kclient cache.Getter, resource string, itemType inte
 	if store == nil {
 		store = cache.NewStore(cache.MetaNamespaceKeyFunc)
 	}
-	runReflectorUntil(cache.NewReflector(lw, itemType, store, c.resyncPeriod), c.resyncPeriod, c.quit)
+	runReflectorUntil(cache.NewReflector(lw, itemType, store, c.resyncPeriod), c.resyncPeriod, c.quit, resource)
 	return store
 }