Merge pull request #482 from freehan/split-neg-sync

add work queue to process endpoint changes

pkg/neg/controller.go

@@ -65,6 +65,8 @@ type Controller struct {
 
 	// serviceQueue takes service key as work item. Service key with format "namespace/name".
 	serviceQueue workqueue.RateLimitingInterface
+	// endpointQueue takes endpoint key as work item. Endpoint key with format "namespace/name".
+	endpointQueue workqueue.RateLimitingInterface
 
 	// syncTracker tracks the latest time that service and endpoint changes are processed
 	syncTracker utils.TimeTracker
@@ -104,6 +106,7 @@ func NewController(
 		ingressLister:  ctx.IngressInformer.GetIndexer(),
 		serviceLister:  ctx.ServiceInformer.GetIndexer(),
 		serviceQueue:   workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
+		endpointQueue:  workqueue.NewRateLimitingQueue(workqueue.DefaultControllerRateLimiter()),
 		syncTracker:    utils.NewTimeTracker(),
 	}
 
@@ -148,10 +151,10 @@ func NewController(
 	})
 
 	ctx.EndpointInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    negController.processEndpoint,
-		DeleteFunc: negController.processEndpoint,
+		AddFunc:    negController.enqueueEndpoint,
+		DeleteFunc: negController.enqueueEndpoint,
 		UpdateFunc: func(old, cur interface{}) {
-			negController.processEndpoint(cur)
+			negController.enqueueEndpoint(cur)
 		},
 	})
 	ctx.AddHealthCheck("neg-controller", negController.IsHealthy)
@@ -171,6 +174,7 @@ func (c *Controller) Run(stopCh <-chan struct{}) {
 	}()
 
 	go wait.Until(c.serviceWorker, time.Second, stopCh)
+	go wait.Until(c.endpointWorker, time.Second, stopCh)
 	go func() {
 		// Wait for gcPeriod to run the first GC
 		// This is to make sure that all services are fully processed before running GC.
@@ -196,23 +200,33 @@ func (c *Controller) IsHealthy() error {
 func (c *Controller) stop() {
 	glog.V(2).Infof("Shutting down network endpoint group controller")
 	c.serviceQueue.ShutDown()
+	c.endpointQueue.ShutDown()
 	c.manager.ShutDown()
 }
 
+func (c *Controller) endpointWorker() {
+	for {
+		func() {
+			key, quit := c.endpointQueue.Get()
+			if quit {
+				return
+			}
+			c.processEndpoint(key.(string))
+			c.endpointQueue.Done(key)
+		}()
+	}
+}
+
 // processEndpoint finds the related syncers and signal it to sync
-func (c *Controller) processEndpoint(obj interface{}) {
+func (c *Controller) processEndpoint(key string) {
 	defer func() {
 		now := c.syncTracker.Track()
 		metrics.LastSyncTimestamp.WithLabelValues().Set(float64(now.UTC().UnixNano()))
 	}()
 
-	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
-	if err != nil {
-		glog.Errorf("Failed to generate endpoint key: %v", err)
-		return
-	}
 	namespace, name, err := cache.SplitMetaNamespaceKey(key)
 	if err != nil {
+		glog.Errorf("Failed to split endpoint namespaced key %q: %v", key, err)
 		return
 	}
 	c.manager.Sync(namespace, name)
@@ -362,6 +376,15 @@ func (c *Controller) handleErr(err error, key interface{}) {
 	c.serviceQueue.AddRateLimited(key)
 }
 
+func (c *Controller) enqueueEndpoint(obj interface{}) {
+	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+	if err != nil {
+		glog.Errorf("Failed to generate endpoint key: %v", err)
+		return
+	}
+	c.endpointQueue.Add(key)
+}
+
 func (c *Controller) enqueueService(obj interface{}) {
 	key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
 	if err != nil {