@@ -342,14 +342,15 @@ func (c *MasterConfig) RunDeployerPodController() {

 // RunDeploymentConfigController starts the deployment config controller process.

 func (c *MasterConfig) RunDeploymentConfigController() {

+	dcInfomer := c.Informers.DeploymentConfigs().Informer()

+	rcInformer := c.Informers.ReplicationControllers().Informer()

 	osclient, kclient := c.DeploymentConfigControllerClients()

-	factory := deployconfigcontroller.DeploymentConfigControllerFactory{

-		Client:     osclient,

-		KubeClient: kclient,

-		Codec:      c.EtcdHelper.Codec(),

-	}

-	controller := factory.Create()

-	controller.Run()

+

+	controller := deployconfigcontroller.NewDeploymentConfigController(dcInfomer, rcInformer, osclient, kclient, c.EtcdHelper.Codec())

+	// TODO: Make the stop channel actually work.

+	stopCh := make(chan struct{})

+	// TODO: Make the number of workers configurable.

+	go controller.Run(5, stopCh)

 }

 // RunDeploymentTriggerController starts the deployment trigger controller process.

@@ -13,12 +13,16 @@ import (

 	"k8s.io/kubernetes/pkg/watch"

 	oclient "github.com/openshift/origin/pkg/client"

+	oscache "github.com/openshift/origin/pkg/client/cache"

+	deployapi "github.com/openshift/origin/pkg/deploy/api"

 )

 type InformerFactory interface {

 	Start(stopCh <-chan struct{})

 	Pods() PodInformer

+	ReplicationControllers() ReplicationControllerInformer

+	DeploymentConfigs() DeploymentConfigInformer

 }

 type PodInformer interface {

@@ -27,6 +31,18 @@ type PodInformer interface {

 	Lister() *cache.StoreToPodLister

 }

+type ReplicationControllerInformer interface {

+	Informer() framework.SharedIndexInformer

+	Indexer() cache.Indexer

+	Lister() *cache.StoreToReplicationControllerLister

+}

+

+type DeploymentConfigInformer interface {

+	Informer() framework.SharedIndexInformer

+	Indexer() cache.Indexer

+	Lister() *oscache.StoreToDeploymentConfigLister

+}

+

 func NewInformerFactory(kubeClient kclient.Interface, originClient oclient.Interface, defaultResync time.Duration) InformerFactory {

 	return &sharedInformerFactory{

 		kubeClient:    kubeClient,

@@ -57,6 +73,14 @@ func (f *sharedInformerFactory) Pods() PodInformer {

 	return &podInformer{sharedInformerFactory: f}

 }

+func (f *sharedInformerFactory) ReplicationControllers() ReplicationControllerInformer {

+	return &replicationControllerInformer{sharedInformerFactory: f}

+}

+

+func (f *sharedInformerFactory) DeploymentConfigs() DeploymentConfigInformer {

+	return &deploymentConfigInformer{sharedInformerFactory: f}

+}

+

 type podInformer struct {

 	*sharedInformerFactory

 }

@@ -99,3 +123,89 @@ func (f *podInformer) Lister() *cache.StoreToPodLister {

 	informer := f.Informer()

 	return &cache.StoreToPodLister{Indexer: informer.GetIndexer()}

 }

+

+type replicationControllerInformer struct {

+	*sharedInformerFactory

+}

+

+func (f *replicationControllerInformer) Informer() framework.SharedIndexInformer {

+	f.lock.Lock()

+	defer f.lock.Unlock()

+

+	informerObj := &kapi.ReplicationController{}

+	informerType := reflect.TypeOf(informerObj)

+	informer, exists := f.informers[informerType]

+	if exists {

+		return informer

+	}

+

+	informer = framework.NewSharedIndexInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return f.kubeClient.ReplicationControllers(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return f.kubeClient.ReplicationControllers(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		informerObj,

+		f.defaultResync,

+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+	)

+	f.informers[informerType] = informer

+

+	return informer

+}

+

+func (f *replicationControllerInformer) Indexer() cache.Indexer {

+	informer := f.Informer()

+	return informer.GetIndexer()

+}

+

+func (f *replicationControllerInformer) Lister() *cache.StoreToReplicationControllerLister {

+	informer := f.Informer()

+	return &cache.StoreToReplicationControllerLister{Indexer: informer.GetIndexer()}

+}

+

+type deploymentConfigInformer struct {

+	*sharedInformerFactory

+}

+

+func (f *deploymentConfigInformer) Informer() framework.SharedIndexInformer {

+	f.lock.Lock()

+	defer f.lock.Unlock()

+

+	informerObj := &deployapi.DeploymentConfig{}

+	informerType := reflect.TypeOf(informerObj)

+	informer, exists := f.informers[informerType]

+	if exists {

+		return informer

+	}

+

+	informer = framework.NewSharedIndexInformer(

+		&cache.ListWatch{

+			ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+				return f.originClient.DeploymentConfigs(kapi.NamespaceAll).List(options)

+			},

+			WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+				return f.originClient.DeploymentConfigs(kapi.NamespaceAll).Watch(options)

+			},

+		},

+		informerObj,

+		f.defaultResync,

+		cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+	)

+	f.informers[informerType] = informer

+

+	return informer

+}

+

+func (f *deploymentConfigInformer) Indexer() cache.Indexer {

+	informer := f.Informer()

+	return informer.GetIndexer()

+}

+

+func (f *deploymentConfigInformer) Lister() *oscache.StoreToDeploymentConfigLister {

+	informer := f.Informer()

+	return &oscache.StoreToDeploymentConfigLister{Indexer: informer.GetIndexer()}

+}

@@ -8,15 +8,26 @@ import (

 	kapi "k8s.io/kubernetes/pkg/api"

 	"k8s.io/kubernetes/pkg/api/errors"

+	"k8s.io/kubernetes/pkg/client/cache"

 	"k8s.io/kubernetes/pkg/client/record"

 	kclient "k8s.io/kubernetes/pkg/client/unversioned"

 	"k8s.io/kubernetes/pkg/runtime"

+	utilruntime "k8s.io/kubernetes/pkg/util/runtime"

+	"k8s.io/kubernetes/pkg/util/workqueue"

 	osclient "github.com/openshift/origin/pkg/client"

+	oscache "github.com/openshift/origin/pkg/client/cache"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

 	deployutil "github.com/openshift/origin/pkg/deploy/util"

 )

+// fatalError is an error which can't be retried.

+type fatalError string

+

+func (e fatalError) Error() string {

+	return fmt.Sprintf("fatal error handling deployment config: %s", string(e))

+}

+

 // DeploymentConfigController is responsible for creating a new deployment

 // when:

 //

@@ -32,48 +43,41 @@ import (

 // deployments will be cancelled. The controller will not attempt to scale

 // running deployments.

 type DeploymentConfigController struct {

-	// kubeClient provides acceess to Kube resources.

-	kubeClient kclient.Interface

-	// osClient provides access to OpenShift resources.

-	osClient osclient.Interface

+	// dn provides access to deploymentconfigs.

+	dn osclient.DeploymentConfigsNamespacer

+	// rn provides access to replication controllers.

+	rn kclient.ReplicationControllersNamespacer

+

+	// queue contains deployment configs that need to be synced.

+	queue workqueue.RateLimitingInterface

+

+	// dcStore provides a local cache for deployment configs.

+	dcStore oscache.StoreToDeploymentConfigLister

+	// rcStore provides a local cache for replication controllers.

+	rcStore cache.StoreToReplicationControllerLister

+	// dcStoreSynced makes sure the dc store is synced before reconcling any deployment config.

+	dcStoreSynced func() bool

+	// rcStoreSynced makes sure the rc store is synced before reconcling any deployment config.

+	rcStoreSynced func() bool

+

 	// codec is used to build deployments from configs.

 	codec runtime.Codec

 	// recorder is used to record events.

 	recorder record.EventRecorder

 }

-// fatalError is an error which can't be retried.

-type fatalError string

-

-// transientError is an error which should always be retried (indefinitely).

-type transientError string

-

-func (e fatalError) Error() string {

-	return fmt.Sprintf("fatal error handling deployment config: %s", string(e))

-}

-func (e transientError) Error() string {

-	return "transient error handling deployment config: " + string(e)

-}

-

-func NewDeploymentConfigController(kubeClient kclient.Interface, osClient osclient.Interface, codec runtime.Codec, recorder record.EventRecorder) *DeploymentConfigController {

-	return &DeploymentConfigController{

-		kubeClient: kubeClient,

-		osClient:   osClient,

-		codec:      codec,

-		recorder:   recorder,

-	}

-}

-

+// Handle implements the loop that processes deployment configs. Since this controller started

+// using caches, the provided config MUST be deep-copied beforehand (see work() in factory.go).

 func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig) error {

-	// There's nothing to reconcile until the version is nonzero.

-	if config.Status.LatestVersion == 0 {

-		glog.V(5).Infof("Waiting for first version of %q", deployutil.LabelForDeploymentConfig(config))

+	// There's nothing to reconcile until the version is nonzero or when the

+	// deployment config has been marked for deletion.

+	if config.Status.LatestVersion == 0 || config.DeletionTimestamp != nil {

 		return c.updateStatus(config)

 	}

-	// Find all deployments owned by the deploymentConfig.

-	sel := deployutil.ConfigSelector(config.Name)

-	existingDeployments, err := c.kubeClient.ReplicationControllers(config.Namespace).List(kapi.ListOptions{LabelSelector: sel})

+	// Find all deployments owned by the deployment config.

+	selector := deployutil.ConfigSelector(config.Name)

+	existingDeployments, err := c.rcStore.ReplicationControllers(config.Namespace).List(selector)

 	if err != nil {

 		return err

 	}

@@ -83,7 +87,7 @@ func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig)

 	// deployments to allow them to be superceded by the new config version.

 	awaitingCancellations := false

 	if !latestIsDeployed {

-		for _, deployment := range existingDeployments.Items {

+		for _, deployment := range existingDeployments {

 			// Skip deployments with an outcome.

 			if deployutil.IsTerminatedDeployment(&deployment) {

 				continue

@@ -93,7 +97,7 @@ func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig)

 			if !deployutil.IsDeploymentCancelled(&deployment) {

 				deployment.Annotations[deployapi.DeploymentCancelledAnnotation] = deployapi.DeploymentCancelledAnnotationValue

 				deployment.Annotations[deployapi.DeploymentStatusReasonAnnotation] = deployapi.DeploymentCancelledNewerDeploymentExists

-				_, err := c.kubeClient.ReplicationControllers(deployment.Namespace).Update(&deployment)

+				_, err := c.rn.ReplicationControllers(deployment.Namespace).Update(&deployment)

 				if err != nil {

 					c.recorder.Eventf(config, kapi.EventTypeWarning, "DeploymentCancellationFailed", "Failed to cancel deployment %q superceded by version %d: %s", deployment.Name, config.Status.LatestVersion, err)

 				} else {

@@ -106,8 +110,7 @@ func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig)

 	// deployment to avoid competing with existing deployment processes.

 	if awaitingCancellations {

 		c.recorder.Eventf(config, kapi.EventTypeNormal, "DeploymentAwaitingCancellation", "Deployment of version %d awaiting cancellation of older running deployments", config.Status.LatestVersion)

-		// raise a transientError so that the deployment config can be re-queued

-		return transientError(fmt.Sprintf("found previous inflight deployment for %s - requeuing", deployutil.LabelForDeploymentConfig(config)))

+		return fmt.Errorf("found previous inflight deployment for %s - requeuing", deployutil.LabelForDeploymentConfig(config))

 	}

 	// If the latest deployment already exists, reconcile existing deployments

 	// and return early.

@@ -130,7 +133,7 @@ func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig)

 	if err != nil {

 		return fatalError(fmt.Sprintf("couldn't make deployment from (potentially invalid) deployment config %s: %v", deployutil.LabelForDeploymentConfig(config), err))

 	}

-	created, err := c.kubeClient.ReplicationControllers(config.Namespace).Create(deployment)

+	created, err := c.rn.ReplicationControllers(config.Namespace).Create(deployment)

 	if err != nil {

 		// If the deployment was already created, just move on. The cache could be

 		// stale, or another process could have already handled this update.

@@ -157,7 +160,7 @@ func (c *DeploymentConfigController) Handle(config *deployapi.DeploymentConfig)

 // directly. To continue supporting that old behavior we must detect when the

 // deployment has been directly manipulated, and if so, preserve the directly

 // updated value and sync the config with the deployment.

-func (c *DeploymentConfigController) reconcileDeployments(existingDeployments *kapi.ReplicationControllerList, config *deployapi.DeploymentConfig) error {

+func (c *DeploymentConfigController) reconcileDeployments(existingDeployments []kapi.ReplicationController, config *deployapi.DeploymentConfig) error {

 	latestIsDeployed, latestDeployment := deployutil.LatestDeploymentInfo(config, existingDeployments)

 	if !latestIsDeployed {

 		// We shouldn't be reconciling if the latest deployment hasn't been

@@ -226,7 +229,7 @@ func (c *DeploymentConfigController) reconcileDeployments(existingDeployments *k

 		oldReplicas := config.Spec.Replicas

 		config.Spec.Replicas = activeReplicas

 		var err error

-		config, err = c.osClient.DeploymentConfigs(config.Namespace).Update(config)

+		config, err = c.dn.DeploymentConfigs(config.Namespace).Update(config)

 		if err != nil {

 			return err

 		}

@@ -235,7 +238,7 @@ func (c *DeploymentConfigController) reconcileDeployments(existingDeployments *k

 	// Reconcile deployments. The active deployment follows the config, and all

 	// other deployments should be scaled to zero.

-	for _, deployment := range existingDeployments.Items {

+	for _, deployment := range existingDeployments {

 		isActiveDeployment := activeDeployment != nil && deployment.Name == activeDeployment.Name

 		oldReplicaCount := deployment.Spec.Replicas

@@ -259,7 +262,7 @@ func (c *DeploymentConfigController) reconcileDeployments(existingDeployments *k

 			copied.Spec.Replicas = newReplicaCount

 			copied.Annotations[deployapi.DeploymentReplicasAnnotation] = strconv.Itoa(int(newReplicaCount))

-			if _, err := c.kubeClient.ReplicationControllers(copied.Namespace).Update(copied); err != nil {

+			if _, err := c.rn.ReplicationControllers(copied.Namespace).Update(copied); err != nil {

 				c.recorder.Eventf(config, kapi.EventTypeWarning, "DeploymentScaleFailed",

 					"Failed to scale deployment %q from %d to %d: %v", copied.Name, oldReplicaCount, newReplicaCount, err)

 				return err

@@ -278,17 +281,42 @@ func (c *DeploymentConfigController) reconcileDeployments(existingDeployments *k

 }

 func (c *DeploymentConfigController) updateStatus(config *deployapi.DeploymentConfig) error {

-	if config.Generation > config.Status.ObservedGeneration {

-		config.Status.ObservedGeneration = config.Generation

+	// NOTE: We should update the status of the deployment config only if we need to, otherwise

+	// we hotloop between updates.

+	if !needsUpdate(config) {

+		return nil

 	}

-	if _, err := c.osClient.DeploymentConfigs(config.Namespace).UpdateStatus(config); err != nil {

+	config.Status.ObservedGeneration = config.Generation

+	if _, err := c.dn.DeploymentConfigs(config.Namespace).UpdateStatus(config); err != nil {

 		glog.V(2).Infof("Cannot update the status for %q: %v", deployutil.LabelForDeploymentConfig(config), err)

-		return transientError(fmt.Sprintf("cannot update the status for %q - requeuing", deployutil.LabelForDeploymentConfig(config)))

+		return err

 	}

 	glog.V(4).Infof("Updated the status for %q (observed generation: %d)", deployutil.LabelForDeploymentConfig(config), config.Status.ObservedGeneration)

 	return nil

 }

+func (c *DeploymentConfigController) handleErr(err error, key interface{}) {

+	if err == nil {

+		return

+	}

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

+		utilruntime.HandleError(err)

+		c.queue.Forget(key)

+		return

+	}

+

+	if c.queue.NumRequeues(key) < 10 {

+		c.queue.AddRateLimited(key)

+	} else {

+		glog.V(2).Infof(err.Error())

+		c.queue.Forget(key)

+	}

+}

+

+func needsUpdate(config *deployapi.DeploymentConfig) bool {

+	return config.Generation > config.Status.ObservedGeneration

+}

+

 func deploymentCopy(rc *kapi.ReplicationController) (*kapi.ReplicationController, error) {

 	objCopy, err := kapi.Scheme.DeepCopy(rc)

 	if err != nil {

@@ -4,12 +4,15 @@ import (

 	"sort"

 	"strconv"

 	"testing"

+	"time"

 	kapi "k8s.io/kubernetes/pkg/api"

-	"k8s.io/kubernetes/pkg/client/record"

+	"k8s.io/kubernetes/pkg/client/cache"

 	ktestclient "k8s.io/kubernetes/pkg/client/unversioned/testclient"

+	"k8s.io/kubernetes/pkg/controller/framework"

 	"k8s.io/kubernetes/pkg/runtime"

 	"k8s.io/kubernetes/pkg/util/diff"

+	"k8s.io/kubernetes/pkg/watch"

 	"github.com/openshift/origin/pkg/client/testclient"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

@@ -620,19 +623,15 @@ func TestHandleScenarios(t *testing.T) {

 		t.Logf("evaluating test: %s", test.name)

 		deployments := map[string]kapi.ReplicationController{}

+		toStore := []kapi.ReplicationController{}

 		for _, template := range test.before {

 			deployment := mkdeployment(template)

 			deployments[deployment.Name] = deployment

+			toStore = append(toStore, deployment)

 		}

+		oc := &testclient.Fake{}

 		kc := &ktestclient.Fake{}

-		kc.AddReactor("list", "replicationcontrollers", func(action ktestclient.Action) (handled bool, ret runtime.Object, err error) {

-			list := []kapi.ReplicationController{}

-			for _, deployment := range deployments {

-				list = append(list, deployment)

-			}

-			return true, &kapi.ReplicationControllerList{Items: list}, nil

-		})

 		kc.AddReactor("create", "replicationcontrollers", func(action ktestclient.Action) (handled bool, ret runtime.Object, err error) {

 			rc := action.(ktestclient.CreateAction).GetObject().(*kapi.ReplicationController)

 			deployments[rc.Name] = *rc

@@ -643,15 +642,38 @@ func TestHandleScenarios(t *testing.T) {

 			deployments[rc.Name] = *rc

 			return true, rc, nil

 		})

+		codec := kapi.Codecs.LegacyCodec(deployapi.SchemeGroupVersion)

-		oc := &testclient.Fake{}

+		dcInformer := framework.NewSharedIndexInformer(

+			&cache.ListWatch{

+				ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+					return oc.DeploymentConfigs(kapi.NamespaceAll).List(options)

+				},

+				WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+					return oc.DeploymentConfigs(kapi.NamespaceAll).Watch(options)

+				},

+			},

+			&deployapi.DeploymentConfig{},

+			2*time.Minute,

+			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+		)

+		rcInformer := framework.NewSharedIndexInformer(

+			&cache.ListWatch{

+				ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

+					return kc.ReplicationControllers(kapi.NamespaceAll).List(options)

+				},

+				WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

+					return kc.ReplicationControllers(kapi.NamespaceAll).Watch(options)

+				},

+			},

+			&kapi.ReplicationController{},

+			2*time.Minute,

+			cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},

+		)

+		c := NewDeploymentConfigController(dcInformer, rcInformer, oc, kc, codec)

-		recorder := &record.FakeRecorder{}

-		controller := &DeploymentConfigController{

-			kubeClient: kc,

-			osClient:   oc,

-			codec:      kapi.Codecs.LegacyCodec(deployapi.SchemeGroupVersion),

-			recorder:   recorder,

+		for i := range toStore {

+			c.rcStore.Add(&toStore[i])

 		}

 		config := deploytest.OkDeploymentConfig(test.newVersion)

@@ -659,9 +681,10 @@ func TestHandleScenarios(t *testing.T) {

 			config = deploytest.TestDeploymentConfig(config)

 		}

 		config.Spec.Replicas = test.replicas

-		err := controller.Handle(config)

-		if err != nil && !test.errExpected {

-			t.Fatalf("unexpected error: %s", err)

+

+		if err := c.Handle(config); err != nil && !test.errExpected {

+			t.Errorf("unexpected error: %s", err)

+			continue

 		}

 		expectedDeployments := []kapi.ReplicationController{}

@@ -677,22 +700,14 @@ func TestHandleScenarios(t *testing.T) {

 		if e, a := test.expectedReplicas, config.Spec.Replicas; e != a {

 			t.Errorf("expected config replicas to be %d, got %d", e, a)

-			// TODO: Disable as the recorder.Events is now `chan string`

-			//t.Fatalf("events:\n%s", strings.Join(recorder.Events, "\t\n"))

+			continue

 		}

-		anyDeploymentMismatches := false

 		for i := 0; i < len(expectedDeployments); i++ {

 			expected, actual := expectedDeployments[i], actualDeployments[i]

 			if !kapi.Semantic.DeepEqual(expected, actual) {

-				anyDeploymentMismatches = true

 				t.Errorf("actual deployment don't match expected: %v", diff.ObjectDiff(expected, actual))

 			}

 		}

-		if anyDeploymentMismatches {

-			// TODO: Disable as the recorder.Events is now `chan string`

-			//t.Fatalf("events:\n%s", strings.Join(recorder.Events, "\t\n"))

-			t.Fatalf("deployment mismatches detected")

-		}

 	}

 }

@@ -9,82 +9,232 @@ import (

 	"k8s.io/kubernetes/pkg/client/cache"

 	"k8s.io/kubernetes/pkg/client/record"

 	kclient "k8s.io/kubernetes/pkg/client/unversioned"

+	kcontroller "k8s.io/kubernetes/pkg/controller"

+	"k8s.io/kubernetes/pkg/controller/framework"

 	"k8s.io/kubernetes/pkg/runtime"

-	"k8s.io/kubernetes/pkg/util/flowcontrol"

 	utilruntime "k8s.io/kubernetes/pkg/util/runtime"

-	"k8s.io/kubernetes/pkg/watch"

+	"k8s.io/kubernetes/pkg/util/wait"

+	"k8s.io/kubernetes/pkg/util/workqueue"

 	osclient "github.com/openshift/origin/pkg/client"

-	controller "github.com/openshift/origin/pkg/controller"

 	deployapi "github.com/openshift/origin/pkg/deploy/api"

 )

-// DeploymentConfigControllerFactory can create a DeploymentConfigController which obtains

-// DeploymentConfigs from a queue populated from a watch of all DeploymentConfigs.

-type DeploymentConfigControllerFactory struct {

-	// Client is an OpenShift client.

-	Client osclient.Interface

-	// KubeClient is a Kubernetes client.

-	KubeClient kclient.Interface

-	// Codec is used to encode/decode.

-	Codec runtime.Codec

+const (

+	// We must avoid creating new replication controllers until the {replication controller,pods} store

+	// has synced. If it hasn't synced, to avoid a hot loop, we'll wait this long between checks.

+	StoreSyncedPollPeriod = 100 * time.Millisecond

+)

+

+// NewDeploymentConfigController creates a new DeploymentConfigController.

+func NewDeploymentConfigController(dcInformer, rcInformer framework.SharedIndexInformer, oc osclient.Interface, kc kclient.Interface, codec runtime.Codec) *DeploymentConfigController {

+	eventBroadcaster := record.NewBroadcaster()

+	eventBroadcaster.StartRecordingToSink(kc.Events(""))

+	recorder := eventBroadcaster.NewRecorder(kapi.EventSource{Component: "deploymentconfig-controller"})

+

+	c := &DeploymentConfigController{

+		dn: oc,

+		rn: kc,

+

+		queue: workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),

+

+		recorder: recorder,

+		codec:    codec,

+	}

+

+	c.dcStore.Indexer = dcInformer.GetIndexer()

+	dcInformer.AddEventHandler(framework.ResourceEventHandlerFuncs{

+		AddFunc:    c.addDeploymentConfig,

+		UpdateFunc: c.updateDeploymentConfig,

+		DeleteFunc: c.deleteDeploymentConfig,

+	})

+

+	c.rcStore.Indexer = rcInformer.GetIndexer()

+	rcInformer.AddEventHandler(framework.ResourceEventHandlerFuncs{

+		AddFunc:    c.addReplicationController,

+		UpdateFunc: c.updateReplicationController,

+		DeleteFunc: c.deleteReplicationController,

+	})

+

+	c.dcStoreSynced = dcInformer.HasSynced

+	c.rcStoreSynced = rcInformer.HasSynced

+

+	return c

 }

-// Create creates a DeploymentConfigController.

-func (factory *DeploymentConfigControllerFactory) Create() controller.RunnableController {

-	deploymentConfigLW := &cache.ListWatch{

-		ListFunc: func(options kapi.ListOptions) (runtime.Object, error) {

-			return factory.Client.DeploymentConfigs(kapi.NamespaceAll).List(options)

-		},

-		WatchFunc: func(options kapi.ListOptions) (watch.Interface, error) {

-			return factory.Client.DeploymentConfigs(kapi.NamespaceAll).Watch(options)

-		},

+// Run begins watching and syncing.

+func (c *DeploymentConfigController) Run(workers int, stopCh <-chan struct{}) {

+	defer utilruntime.HandleCrash()

+

+	// Wait for the rc and dc stores to sync before starting any work in this controller.

+	ready := make(chan struct{})

+	go c.waitForSyncedStores(ready)

+	select {

+	case <-ready:

+	case <-stopCh:

+		return

 	}

-	queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)

-	cache.NewReflector(deploymentConfigLW, &deployapi.DeploymentConfig{}, queue, 2*time.Minute).Run()

-	eventBroadcaster := record.NewBroadcaster()

-	eventBroadcaster.StartRecordingToSink(factory.KubeClient.Events(""))

-	recorder := eventBroadcaster.NewRecorder(kapi.EventSource{Component: "deploymentconfig-controller"})

+	for i := 0; i < workers; i++ {

+		go wait.Until(c.worker, time.Second, stopCh)

+	}

+

+	<-stopCh

+	glog.Infof("Shutting down deploymentconfig controller")

+	c.queue.ShutDown()

+}

+

+func (c *DeploymentConfigController) waitForSyncedStores(ready chan struct{}) {

+	defer utilruntime.HandleCrash()

+

+	for !c.dcStoreSynced() || !c.rcStoreSynced() {

+		glog.V(4).Infof("Waiting for the dc and rc controllers to sync before starting the deployment config controller workers")

+		time.Sleep(StoreSyncedPollPeriod)

+	}

+	close(ready)

+}

+

+func (c *DeploymentConfigController) addDeploymentConfig(obj interface{}) {

+	dc := obj.(*deployapi.DeploymentConfig)

+	glog.V(4).Infof("Adding deployment config %q", dc.Name)

+	c.enqueueDeploymentConfig(dc)

+}

+

+func (c *DeploymentConfigController) updateDeploymentConfig(old, cur interface{}) {

+	oldDc := old.(*deployapi.DeploymentConfig)

+	glog.V(4).Infof("Updating deployment config %q", oldDc.Name)

+	c.enqueueDeploymentConfig(cur.(*deployapi.DeploymentConfig))

+}

+

+func (c *DeploymentConfigController) deleteDeploymentConfig(obj interface{}) {

+	dc, ok := obj.(*deployapi.DeploymentConfig)

+	if !ok {

+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)

+		if !ok {

+			glog.Errorf("Couldn't get object from tombstone %+v", obj)

+			return

+		}

+		dc, ok = tombstone.Obj.(*deployapi.DeploymentConfig)

+		if !ok {

+			glog.Errorf("Tombstone contained object that is not a deployment config: %+v", obj)

+			return

+		}

+	}

+	glog.V(4).Infof("Deleting deployment config %q", dc.Name)

+	c.enqueueDeploymentConfig(dc)

+}

+

+// addReplicationController enqueues the deployment that manages a replicationcontroller when the replicationcontroller is created.

+func (c *DeploymentConfigController) addReplicationController(obj interface{}) {

+	rc := obj.(*kapi.ReplicationController)

+	glog.V(4).Infof("Replication controller %q added.", rc.Name)

+	// We are waiting for the deployment config store to sync but still there are pathological

+	// cases of highly latent watches.

+	if dc, err := c.dcStore.GetConfigForController(rc); err == nil && dc != nil {

+		c.enqueueDeploymentConfig(dc)

+	}

+}

+

+// updateReplicationController figures out which deploymentconfig is managing this replication

+// controller and requeues the deployment config.

+func (c *DeploymentConfigController) updateReplicationController(old, cur interface{}) {

+	// A periodic relist will send update events for all known controllers.

+	if kapi.Semantic.DeepEqual(old, cur) {

+		return

+	}

+

+	curRC := cur.(*kapi.ReplicationController)

+	glog.V(4).Infof("Replication controller %q updated.", curRC.Name)

+	if dc, err := c.dcStore.GetConfigForController(curRC); err == nil && dc != nil {

+		c.enqueueDeploymentConfig(dc)

+	}

+}

+

+// deleteReplicationController enqueues the deployment that manages a replicationcontroller when

+// the replicationcontroller is deleted. obj could be an *kapi.ReplicationController, or

+// a DeletionFinalStateUnknown marker item.

+func (c *DeploymentConfigController) deleteReplicationController(obj interface{}) {

+	rc, ok := obj.(*kapi.ReplicationController)

+

+	// When a delete is dropped, the relist will notice a pod in the store not

+	// in the list, leading to the insertion of a tombstone object which contains

+	// the deleted key/value.

+	if !ok {

+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)

+		if !ok {

+			glog.Errorf("Couldn't get object from tombstone %#v", obj)

+			return

+		}

+		rc, ok = tombstone.Obj.(*kapi.ReplicationController)

+		if !ok {

+			glog.Errorf("Tombstone contained object that is not a replication controller %#v", obj)

+			return

+		}

+	}

+	glog.V(4).Infof("Replication controller %q deleted.", rc.Name)

+	if dc, err := c.dcStore.GetConfigForController(rc); err == nil && dc != nil {

+		c.enqueueDeploymentConfig(dc)

+	}

+}

+

+func (c *DeploymentConfigController) enqueueDeploymentConfig(dc *deployapi.DeploymentConfig) {

+	key, err := kcontroller.KeyFunc(dc)

+	if err != nil {

+		glog.Errorf("Couldn't get key for object %#v: %v", dc, err)

+		return

+	}

+	c.queue.Add(key)

+}

+

+func (c *DeploymentConfigController) worker() {

+	for {

+		if quit := c.work(); quit {

+			return

+		}

+	}

+}

-	configController := NewDeploymentConfigController(factory.KubeClient, factory.Client, factory.Codec, recorder)

-

-	return &controller.RetryController{

-		Queue: queue,

-		RetryManager: controller.NewQueueRetryManager(

-			queue,

-			cache.MetaNamespaceKeyFunc,

-			func(obj interface{}, err error, retries controller.Retry) bool {

-				config := obj.(*deployapi.DeploymentConfig)

-				// no retries for a fatal error

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

-					glog.V(4).Infof("Will not retry fatal error for deploymentConfig %s/%s: %v", config.Namespace, config.Name, err)

-					utilruntime.HandleError(err)

-					return false

-				}

-				// infinite retries for a transient error

-				if _, isTransient := err.(transientError); isTransient {

-					glog.V(4).Infof("Retrying deploymentConfig %s/%s with error: %v", config.Namespace, config.Name, err)

-					return true

-				}

-				utilruntime.HandleError(err)

-				// no retries for anything else

-				if retries.Count > 0 {

-					return false

-				}

-				return true

-			},

-			flowcontrol.NewTokenBucketRateLimiter(1, 10),

-		),

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

-			config := obj.(*deployapi.DeploymentConfig)

-			copied, err := dcCopy(config)

-			if err != nil {

-				return err

-			}

-			return configController.Handle(copied)

-		},

+func (c *DeploymentConfigController) work() bool {

+	key, quit := c.queue.Get()

+	if quit {

+		return true

 	}

+	defer c.queue.Done(key)

+

+	dc, err := c.getByKey(key.(string))

+	if err != nil {

+		glog.Error(err.Error())

+	}

+

+	if dc == nil {

+		return false

+	}

+

+	copied, err := dcCopy(dc)

+	if err != nil {

+		glog.Error(err.Error())

+		return false

+	}

+

+	err = c.Handle(copied)

+	c.handleErr(err, key)

+

+	return false

+}

+

+func (c *DeploymentConfigController) getByKey(key string) (*deployapi.DeploymentConfig, error) {

+	obj, exists, err := c.dcStore.Indexer.GetByKey(key)

+	if err != nil {

+		glog.V(2).Infof("Unable to retrieve deployment config %q from store: %v", key, err)

+		c.queue.AddRateLimited(key)

+		return nil, err

+	}

+	if !exists {

+		glog.V(4).Infof("Deployment config %q has been deleted", key)

+		return nil, nil

+	}

+

+	return obj.(*deployapi.DeploymentConfig), nil

 }

 func dcCopy(dc *deployapi.DeploymentConfig) (*deployapi.DeploymentConfig, error) {

@@ -22,22 +22,22 @@ func LatestDeploymentNameForConfig(config *deployapi.DeploymentConfig) string {

 // LatestDeploymentInfo returns info about the latest deployment for a config,

 // or nil if there is no latest deployment. The latest deployment is not

 // always the same as the active deployment.

-func LatestDeploymentInfo(config *deployapi.DeploymentConfig, deployments *api.ReplicationControllerList) (bool, *api.ReplicationController) {

-	if config.Status.LatestVersion == 0 || len(deployments.Items) == 0 {

+func LatestDeploymentInfo(config *deployapi.DeploymentConfig, deployments []api.ReplicationController) (bool, *api.ReplicationController) {

+	if config.Status.LatestVersion == 0 || len(deployments) == 0 {

 		return false, nil

 	}

-	sort.Sort(ByLatestVersionDesc(deployments.Items))

-	candidate := &deployments.Items[0]

+	sort.Sort(ByLatestVersionDesc(deployments))

+	candidate := &deployments[0]

 	return DeploymentVersionFor(candidate) == config.Status.LatestVersion, candidate

 }

 // ActiveDeployment returns the latest complete deployment, or nil if there is

 // no such deployment. The active deployment is not always the same as the

 // latest deployment.

-func ActiveDeployment(config *deployapi.DeploymentConfig, deployments *api.ReplicationControllerList) *api.ReplicationController {

-	sort.Sort(ByLatestVersionDesc(deployments.Items))

+func ActiveDeployment(config *deployapi.DeploymentConfig, deployments []api.ReplicationController) *api.ReplicationController {

+	sort.Sort(ByLatestVersionDesc(deployments))

 	var activeDeployment *api.ReplicationController

-	for _, deployment := range deployments.Items {

+	for _, deployment := range deployments {

 		if DeploymentStatusFor(&deployment) == deployapi.DeploymentStatusComplete {

 			activeDeployment = &deployment

 			break