@@ -23,6 +23,12 @@ 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 {

+	kutil.HandleError(err)

+	return true

+}

+

 // BuildControllerFactory constructs BuildController objects

 type BuildControllerFactory struct {

 	OSClient            osclient.Interface

@@ -54,12 +60,7 @@ func (factory *BuildControllerFactory) Create() controller.RunnableController {

 	return &controller.RetryController{

 		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, -1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			kutil.HandleError(err)

-			// BuildController currently has no fatal errors.

-			return true

-		},

+		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, logAndRetry),

 		Handle: func(obj interface{}) error {

 			build := obj.(*buildapi.Build)

 			return buildController.HandleBuild(build)

@@ -103,12 +104,7 @@ func (factory *BuildPodControllerFactory) Create() controller.RunnableController

 	return &controller.RetryController{

 		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, -1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			kutil.HandleError(err)

-			// BuildPodController currently has no fatal errors.

-			return true

-		},

+		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, logAndRetry),

 		Handle: func(obj interface{}) error {

 			pod := obj.(*kapi.Pod)

 			return buildPodController.HandlePod(pod)

@@ -143,15 +139,18 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl

 	}

 	return &controller.RetryController{

-		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, -1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			kutil.HandleError(err)

-			if _, isFatal := err.(buildcontroller.ImageChangeControllerFatalError); isFatal {

-				return false

-			}

-			return true

-		},

+		Queue: queue,

+		RetryManager: controller.NewQueueRetryManager(

+			queue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, _ int) bool {

+				kutil.HandleError(err)

+				if _, isFatal := err.(buildcontroller.ImageChangeControllerFatalError); isFatal {

+					return false

+				}

+				return true

+			},

+		),

 		Handle: func(obj interface{}) error {

 			imageRepo := obj.(*imageapi.ImageRepository)

 			return imageChangeController.HandleImageRepo(imageRepo)

@@ -16,25 +16,20 @@ type RunnableController interface {

 // which failed to be successfully handled.

 type RetryController struct {

 	// Queue is where work is retrieved for Handle.

-	Queue Queue

+	Queue

 	// Handle is expected to process the next resource from the queue.

 	Handle func(interface{}) error

-	// ShouldRetry returns true if the resource and error returned from

-	// HandleNext should trigger a retry via the RetryManager.

-	ShouldRetry func(interface{}, error) bool

-

 	// RetryManager is fed the handled resource if Handle returns a Retryable

 	// error. If Handle returns no error, the RetryManager is asked to forget

 	// the resource.

-	RetryManager RetryManager

+	RetryManager

 }

 // Queue is a narrow abstraction of a cache.FIFO.

 type Queue interface {

 	Pop() interface{}

-	AddIfNotPresent(interface{}) error

 }

 // Run begins processing resources from Queue asynchronously.

@@ -42,17 +37,19 @@ func (c *RetryController) Run() {

 	go kutil.Forever(func() { c.handleOne(c.Queue.Pop()) }, 0)

 }

+// RunUntil begins processing resources from Queue asynchronously until stopCh is closed.

+func (c *RetryController) RunUntil(stopCh <-chan struct{}) {

+	go kutil.Until(func() { c.handleOne(c.Queue.Pop()) }, 0, stopCh)

+}

+

 // handleOne processes resource with Handle. If Handle returns a retryable

 // error, the handled resource is passed to the RetryManager. If no error is

 // returned from Handle, the RetryManager is asked to forget the processed

 // resource.

 func (c *RetryController) handleOne(resource interface{}) {

-	err := c.Handle(resource)

-	if err != nil {

-		if c.ShouldRetry(resource, err) {

-			c.RetryManager.Retry(resource)

-			return

-		}

+	if err := c.Handle(resource); err != nil {

+		c.Retry(resource, err)

+		return

 	}

 	c.RetryManager.Forget(resource)

 }

@@ -62,7 +59,7 @@ func (c *RetryController) handleOne(resource interface{}) {

 type RetryManager interface {

 	// Retry will cause resource processing to be retried (for example, by

 	// requeueing resource)

-	Retry(resource interface{})

+	Retry(resource interface{}, err error)

 	// Forget will cause the manager to erase all prior knowledge of resource

 	// and reclaim internal resources associated with state tracking of

@@ -70,38 +67,46 @@ type RetryManager interface {

 	Forget(resource interface{})

 }

-// QueueRetryManager retries a resource by re-queueing it into a Queue up to

-// MaxRetries number of times.

+// RetryFunc should return true if the given object and error should be retried after

+// the provided number of times.

+type RetryFunc func(obj interface{}, err error, retries int) bool

+

+// QueueRetryManager retries a resource by re-queueing it into a ReQueue as long as

+// retryFunc returns true.

 type QueueRetryManager struct {

 	// queue is where resources are re-queued.

-	queue Queue

+	queue ReQueue

 	// keyFunc is used to index resources.

 	keyFunc kcache.KeyFunc

-	// maxRetries is the total number of attempts to requeue an individual

-	// resource before giving up. A value of -1 is interpreted as retry forever.

-	maxRetries int

+	// retryFunc returns true if the resource and error returned should be retried.

+	retryFunc RetryFunc

 	// retries maps resources to their current retry count.

 	retries map[string]int

 }

+// ReQueue is a queue that allows an object to be requeued

+type ReQueue interface {

+	Queue

+	AddIfNotPresent(interface{}) error

+}

+

 // NewQueueRetryManager safely creates a new QueueRetryManager.

-func NewQueueRetryManager(queue Queue, keyFunc kcache.KeyFunc, maxRetries int) *QueueRetryManager {

+func NewQueueRetryManager(queue ReQueue, keyFn kcache.KeyFunc, retryFn RetryFunc) *QueueRetryManager {

 	return &QueueRetryManager{

-		queue:      queue,

-		keyFunc:    keyFunc,

-		maxRetries: maxRetries,

-		retries:    make(map[string]int),

+		queue:     queue,

+		keyFunc:   keyFn,

+		retryFunc: retryFn,

+		retries:   make(map[string]int),

 	}

 }

-// Retry will enqueue resource until maxRetries for that resource has been

-// exceeded, at which point resource will be forgotten and no longer retried.

-//

-// A maxRetries value of -1 is interpreted as retry forever.

-func (r *QueueRetryManager) Retry(resource interface{}) {

+// Retry will enqueue resource until retryFunc returns false for that resource has been

+// exceeded, at which point resource will be forgotten and no longer retried. The current

+// retry count will be passed to each invocation of retryFunc.

+func (r *QueueRetryManager) Retry(resource interface{}, err error) {

 	id, _ := r.keyFunc(resource)

 	if _, exists := r.retries[id]; !exists {

@@ -109,7 +114,7 @@ func (r *QueueRetryManager) Retry(resource interface{}) {

 	}

 	tries := r.retries[id]

-	if tries < r.maxRetries || r.maxRetries == -1 {

+	if r.retryFunc(resource, err, tries) {

 		// It's important to use AddIfNotPresent to prevent overwriting newer

 		// state in the queue which may have arrived asynchronously.

 		r.queue.AddIfNotPresent(resource)

@@ -10,16 +10,17 @@ import (

 func TestRetryController_handleOneRetryableError(t *testing.T) {

 	retried := false

+	retryErr := fmt.Errorf("retryable error")

 	controller := &RetryController{

 		Handle: func(obj interface{}) error {

-			return fmt.Errorf("retryable error")

-		},

-		ShouldRetry: func(interface{}, error) bool {

-			return true

+			return retryErr

 		},

 		RetryManager: &testRetryManager{

-			RetryFunc: func(resource interface{}) {

+			RetryFunc: func(resource interface{}, err error) {

+				if err != retryErr {

+					t.Fatalf("unexpected error: %v", err)

+				}

 				retried = true

 			},

 			ForgetFunc: func(resource interface{}) {

@@ -35,33 +36,6 @@ func TestRetryController_handleOneRetryableError(t *testing.T) {

 	}

 }

-func TestRetryController_handleOneFatalError(t *testing.T) {

-	forgotten := false

-

-	controller := &RetryController{

-		Handle: func(obj interface{}) error {

-			return fmt.Errorf("fatal error")

-		},

-		ShouldRetry: func(interface{}, error) bool {

-			return false

-		},

-		RetryManager: &testRetryManager{

-			RetryFunc: func(resource interface{}) {

-				t.Fatalf("unexpected call to retry %v", resource)

-			},

-			ForgetFunc: func(resource interface{}) {

-				forgotten = true

-			},

-		},

-	}

-

-	controller.handleOne(struct{}{})

-

-	if !forgotten {

-		t.Fatalf("expected to forget")

-	}

-}

-

 func TestRetryController_handleOneNoError(t *testing.T) {

 	forgotten := false

@@ -69,12 +43,8 @@ func TestRetryController_handleOneNoError(t *testing.T) {

 		Handle: func(obj interface{}) error {

 			return nil

 		},

-		ShouldRetry: func(interface{}, error) bool {

-			t.Fatalf("unexpected retry check")

-			return true

-		},

 		RetryManager: &testRetryManager{

-			RetryFunc: func(resource interface{}) {

+			RetryFunc: func(resource interface{}, err error) {

 				t.Fatalf("unexpected call to retry %v", resource)

 			},

 			ForgetFunc: func(resource interface{}) {

@@ -109,8 +79,13 @@ func TestQueueRetryManager_retries(t *testing.T) {

 		keyFunc: func(obj interface{}) (string, error) {

 			return obj.(testObj).id, nil

 		},

-		maxRetries: retries,

-		retries:    make(map[string]int),

+		retryFunc: func(obj interface{}, err error, count int) bool {

+			if count > 4 {

+				return false

+			}

+			return true

+		},

+		retries: make(map[string]int),

 	}

 	objects := []testObj{

@@ -122,7 +97,7 @@ func TestQueueRetryManager_retries(t *testing.T) {

 	// Retry one more than the max

 	for _, obj := range objects {

 		for i := 0; i < retries+1; i++ {

-			manager.Retry(obj)

+			manager.Retry(obj, nil)

 		}

 	}

@@ -154,10 +129,7 @@ func TestRetryController_realFifoEventOrdering(t *testing.T) {

 	controller := &RetryController{

 		Queue:        fifo,

-		RetryManager: NewQueueRetryManager(fifo, keyFunc, 1),

-		ShouldRetry: func(interface{}, error) bool {

-			return true

-		},

+		RetryManager: NewQueueRetryManager(fifo, keyFunc, func(_ interface{}, _ error, _ int) bool { return true }),

 		Handle: func(obj interface{}) error {

 			if e, a := 1, obj.(testObj).value; e != a {

 				t.Fatalf("expected to handle test value %d, got %d")

@@ -204,12 +176,12 @@ func (t *testFifo) Pop() interface{} {

 }

 type testRetryManager struct {

-	RetryFunc  func(resource interface{})

+	RetryFunc  func(resource interface{}, err error)

 	ForgetFunc func(resource interface{})

 }

-func (m *testRetryManager) Retry(resource interface{}) {

-	m.RetryFunc(resource)

+func (m *testRetryManager) Retry(resource interface{}, err error) {

+	m.RetryFunc(resource, err)

 }

 func (m *testRetryManager) Forget(resource interface{}) {

@@ -57,15 +57,21 @@ func (factory *DeploymentConfigChangeControllerFactory) Create() controller.Runn

 	}

 	return &controller.RetryController{

-		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, 1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			if _, isFatal := err.(fatalError); isFatal {

+		Queue: queue,

+		RetryManager: controller.NewQueueRetryManager(

+			queue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, count int) bool {

 				kutil.HandleError(err)

-				return false

-			}

-			return true

-		},

+				if _, isFatal := err.(fatalError); isFatal {

+					return false

+				}

+				if count > 0 {

+					return false

+				}

+				return true

+			},

+		),

 		Handle: func(obj interface{}) error {

 			config := obj.(*deployapi.DeploymentConfig)

 			return changeController.Handle(config)

@@ -67,11 +67,12 @@ func (factory *DeployerPodControllerFactory) Create() controller.RunnableControl

 	}

 	return &controller.RetryController{

-		Queue:        podQueue,

-		RetryManager: controller.NewQueueRetryManager(podQueue, cache.MetaNamespaceKeyFunc, 1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			return true

-		},

+		Queue: podQueue,

+		RetryManager: controller.NewQueueRetryManager(

+			podQueue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, count int) bool { return count < 1 },

+		),

 		Handle: func(obj interface{}) error {

 			pod := obj.(*kapi.Pod)

 			return podController.Handle(pod)

@@ -68,15 +68,21 @@ func (factory *DeploymentControllerFactory) Create() controller.RunnableControll

 	}

 	return &controller.RetryController{

-		Queue:        deploymentQueue,

-		RetryManager: controller.NewQueueRetryManager(deploymentQueue, cache.MetaNamespaceKeyFunc, 1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			if _, isFatal := err.(fatalError); isFatal {

-				kutil.HandleError(err)

-				return false

-			}

-			return true

-		},

+		Queue: deploymentQueue,

+		RetryManager: controller.NewQueueRetryManager(

+			deploymentQueue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, count int) bool {

+				if _, isFatal := err.(fatalError); isFatal {

+					kutil.HandleError(err)

+					return false

+				}

+				if count > 1 {

+					return false

+				}

+				return true

+			},

+		),

 		Handle: func(obj interface{}) error {

 			deployment := obj.(*kapi.ReplicationController)

 			return deployController.Handle(deployment)

@@ -54,15 +54,21 @@ func (factory *DeploymentConfigControllerFactory) Create() controller.RunnableCo

 	}

 	return &controller.RetryController{

-		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, 1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			if _, isFatal := err.(fatalError); isFatal {

+		Queue: queue,

+		RetryManager: controller.NewQueueRetryManager(

+			queue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, count int) bool {

 				kutil.HandleError(err)

-				return false

-			}

-			return true

-		},

+				if _, isFatal := err.(fatalError); isFatal {

+					return false

+				}

+				if count > 0 {

+					return false

+				}

+				return true

+			},

+		),

 		Handle: func(obj interface{}) error {

 			config := obj.(*deployapi.DeploymentConfig)

 			return configController.Handle(config)

@@ -66,15 +66,21 @@ func (factory *ImageChangeControllerFactory) Create() controller.RunnableControl

 	}

 	return &controller.RetryController{

-		Queue:        queue,

-		RetryManager: controller.NewQueueRetryManager(queue, cache.MetaNamespaceKeyFunc, 1),

-		ShouldRetry: func(obj interface{}, err error) bool {

-			if _, isFatal := err.(fatalError); isFatal {

+		Queue: queue,

+		RetryManager: controller.NewQueueRetryManager(

+			queue,

+			cache.MetaNamespaceKeyFunc,

+			func(obj interface{}, err error, count int) bool {

 				kutil.HandleError(err)

-				return false

-			}

-			return true

-		},

+				if _, isFatal := err.(fatalError); isFatal {

+					return false

+				}

+				if count > 0 {

+					return false

+				}

+				return true

+			},

+		),

 		Handle: func(obj interface{}) error {

 			repo := obj.(*imageapi.ImageRepository)

 			return changeController.Handle(repo)