add delays when handling retries so we don't tight loop

pkg/build/controller/factory/factory.go

@@ -25,10 +25,11 @@ import (
 	imageapi "github.com/openshift/origin/pkg/image/api"
 )
 
-// logAndRetry retries forever - BuildPodController currently has no fatal errors
-func logAndRetry(obj interface{}, err error, _ int) bool {
+// limitedLogAndRetry stops retrying after 60 attempts.  Given the throttler configuration,
+// this means this event has been retried over a period of at least 50 seconds.
+func limitedLogAndRetry(obj interface{}, err error, count int) bool {
 	kutil.HandleError(err)
-	return true
+	return count < 60
 }
 
 // BuildControllerFactory constructs BuildController objects
@@ -62,7 +63,7 @@ func (factory *BuildControllerFactory) Create() controller.RunnableController {
 
 	return &controller.RetryController{
 		Queue:        queue,
-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, logAndRetry),
+		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, limitedLogAndRetry, kutil.NewTokenBucketRateLimiter(1, 10)),
 		Handle: func(obj interface{}) error {
 			build := obj.(*buildapi.Build)
 			return buildController.HandleBuild(build)
@@ -106,7 +107,7 @@ func (factory *BuildPodControllerFactory) Create() controller.RunnableController
 
 	return &controller.RetryController{
 		Queue:        queue,
-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, logAndRetry),
+		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, limitedLogAndRetry, kutil.NewTokenBucketRateLimiter(1, 10)),
 		Handle: func(obj interface{}) error {
 			pod := obj.(*kapi.Pod)
 			return buildPodController.HandlePod(pod)
@@ -145,13 +146,17 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl
 		RetryManager: controller.NewQueueRetryManager(
 			queue,
 			cache.MetaNamespaceKeyFunc,
-			func(obj interface{}, err error, _ int) bool {
+			func(obj interface{}, err error, count int) bool {
 				kutil.HandleError(err)
 				if _, isFatal := err.(buildcontroller.ImageChangeControllerFatalError); isFatal {
 					return false
 				}
+				if count > 60 {
+					return false
+				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			imageRepo := obj.(*imageapi.ImageStream)

pkg/controller/controller.go

@@ -85,6 +85,10 @@ type QueueRetryManager struct {
 
 	// retries maps resources to their current retry count.
 	retries map[string]int
+
+	// limits how fast retries can be enqueued to ensure you can't tight
+	// loop on retries.
+	limiter kutil.RateLimiter
 }
 
 // ReQueue is a queue that allows an object to be requeued
@@ -94,12 +98,13 @@ type ReQueue interface {
 }
 
 // NewQueueRetryManager safely creates a new QueueRetryManager.
-func NewQueueRetryManager(queue ReQueue, keyFn kcache.KeyFunc, retryFn RetryFunc) *QueueRetryManager {
+func NewQueueRetryManager(queue ReQueue, keyFn kcache.KeyFunc, retryFn RetryFunc, limiter kutil.RateLimiter) *QueueRetryManager {
 	return &QueueRetryManager{
 		queue:     queue,
 		keyFunc:   keyFn,
 		retryFunc: retryFn,
 		retries:   make(map[string]int),
+		limiter:   limiter,
 	}
 }
 
@@ -115,6 +120,7 @@ func (r *QueueRetryManager) Retry(resource interface{}, err error) {
 	tries := r.retries[id]
 
 	if r.retryFunc(resource, err, tries) {
+		r.limiter.Accept()
 		// It's important to use AddIfNotPresent to prevent overwriting newer
 		// state in the queue which may have arrived asynchronously.
 		r.queue.AddIfNotPresent(resource)

pkg/controller/controller_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	kcache "github.com/GoogleCloudPlatform/kubernetes/pkg/client/cache"
+	kutil "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 )
 
 func TestRetryController_handleOneRetryableError(t *testing.T) {
@@ -86,6 +87,7 @@ func TestQueueRetryManager_retries(t *testing.T) {
 			return true
 		},
 		retries: make(map[string]int),
+		limiter: kutil.NewTokenBucketRateLimiter(1000, 1000),
 	}
 
 	objects := []testObj{
@@ -129,7 +131,7 @@ func TestRetryController_realFifoEventOrdering(t *testing.T) {
 
 	controller := &RetryController{
 		Queue:        fifo,
-		RetryManager: NewQueueRetryManager(fifo, keyFunc, func(_ interface{}, _ error, _ int) bool { return true }),
+		RetryManager: NewQueueRetryManager(fifo, keyFunc, func(_ interface{}, _ error, _ int) bool { return true }, kutil.NewTokenBucketRateLimiter(1000, 10)),
 		Handle: func(obj interface{}) error {
 			if e, a := 1, obj.(testObj).value; e != a {
 				t.Fatalf("expected to handle test value %d, got %d", e, a)
@@ -157,6 +159,48 @@ func TestRetryController_realFifoEventOrdering(t *testing.T) {
 	}
 }
 
+// This test ensures that when events are retried, the
+// requeue rate does not exceed the configured rate limit,
+// including burst behavior.
+func TestRetryController_ratelimit(t *testing.T) {
+	keyFunc := func(obj interface{}) (string, error) {
+		return "key", nil
+	}
+	fifo := kcache.NewFIFO(keyFunc)
+	limiter := &mockLimiter{}
+	retryManager := NewQueueRetryManager(fifo,
+		keyFunc,
+		func(_ interface{}, _ error, c int) bool {
+			if c < 15 {
+				return true
+			}
+			return false
+		},
+		limiter,
+	)
+	for i := 0; i < 10; i++ {
+		retryManager.Retry("key", nil)
+	}
+
+	if limiter.count != 10 {
+		t.Fatalf("Retries did not invoke rate limiter, expected %d got %d", 10, limiter.count)
+	}
+}
+
+type mockLimiter struct {
+	count int
+}
+
+func (l *mockLimiter) CanAccept() bool {
+	return true
+}
+
+func (l *mockLimiter) Accept() {
+	l.count++
+}
+
+func (l *mockLimiter) Stop() {}
+
 type testObj struct {
 	id    string
 	value int

pkg/deploy/controller/configchange/factory.go

@@ -74,6 +74,7 @@ func (factory *DeploymentConfigChangeControllerFactory) Create() controller.Runn
 				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			config := obj.(*deployapi.DeploymentConfig)

pkg/deploy/controller/deployerpod/factory.go

@@ -11,6 +11,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/fields"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
+	kutil "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 
 	controller "github.com/openshift/origin/pkg/controller"
@@ -73,6 +74,7 @@ func (factory *DeployerPodControllerFactory) Create() controller.RunnableControl
 			podQueue,
 			cache.MetaNamespaceKeyFunc,
 			func(obj interface{}, err error, count int) bool { return count < 1 },
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			pod := obj.(*kapi.Pod)

pkg/deploy/controller/deployment/factory.go

@@ -84,6 +84,7 @@ func (factory *DeploymentControllerFactory) Create() controller.RunnableControll
 				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			deployment := obj.(*kapi.ReplicationController)

pkg/deploy/controller/deploymentconfig/factory.go

@@ -71,6 +71,7 @@ func (factory *DeploymentConfigControllerFactory) Create() controller.RunnableCo
 				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			config := obj.(*deployapi.DeploymentConfig)

pkg/deploy/controller/imagechange/factory.go

@@ -83,6 +83,7 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl
 				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			repo := obj.(*imageapi.ImageStream)

pkg/image/controller/factory.go

@@ -9,6 +9,7 @@ import (
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/runtime"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
+	kutil "github.com/GoogleCloudPlatform/kubernetes/pkg/util"
 	"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
 
 	"github.com/openshift/origin/pkg/client"
@@ -50,6 +51,7 @@ func (f *ImportControllerFactory) Create() controller.RunnableController {
 				util.HandleError(err)
 				return count < 5
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			r := obj.(*api.ImageStream)

pkg/project/controller/factory.go

@@ -56,6 +56,7 @@ func (factory *NamespaceControllerFactory) Create() controller.RunnableControlle
 				}
 				return true
 			},
+			kutil.NewTokenBucketRateLimiter(1, 10),
 		),
 		Handle: func(obj interface{}) error {
 			namespace := obj.(*kapi.Namespace)